punimtag/scripts/rotate-dev-admin-password.py

#!/usr/bin/env python3
"""Rotate DEV admin login: DB password_hash + .env ADMIN_PASSWORD."""
from __future__ import annotations

import base64
import secrets
import string
import subprocess
import sys
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
TMP = ROOT / ".dev-admin-password.tmp"
HOST = "root@10.0.10.201"
LXC = "9101"


def main() -> int:
    alphabet = string.ascii_letters + string.digits
    pw = "".join(secrets.choice(alphabet) for _ in range(28))
    TMP.write_text(pw + "\n")
    TMP.chmod(0o600)
    gi = ROOT / ".gitignore"
    if gi.exists() and ".dev-admin-password.tmp" not in gi.read_text():
        gi.write_text(gi.read_text().rstrip() + "\n.dev-admin-password.tmp\n")

    remote_py = f"""
from pathlib import Path
import os, re, sys
sys.path.insert(0, '/opt/punimtag')
os.chdir('/opt/punimtag')
# load .env into os.environ for DATABASE_URL
env_path = Path('/opt/punimtag/.env')
text = env_path.read_text()
for line in text.splitlines():
    if not line.strip() or line.strip().startswith('#') or '=' not in line:
        continue
    k, v = line.split('=', 1)
    os.environ.setdefault(k.strip(), v.strip().strip('"').strip("'"))

pw = {pw!r}
# update .env ADMIN_PASSWORD
if re.search(r'^ADMIN_PASSWORD=.*$', text, re.M):
    text = re.sub(r'^ADMIN_PASSWORD=.*$', 'ADMIN_PASSWORD=' + pw, text, count=1, flags=re.M)
else:
    text = text.rstrip() + '\\nADMIN_PASSWORD=' + pw + '\\n'
env_path.write_text(text)

from backend.utils.password import hash_password, verify_password
from backend.db.session import SessionLocal
from backend.db.models import User

db = SessionLocal()
try:
    users = db.query(User).filter(User.username == 'admin').all()
    if not users:
        # also try is_admin
        users = db.query(User).filter(User.is_admin == True).all()
    print('users_found', [(u.id, u.username, bool(u.password_hash)) for u in users])
    if not users:
        raise SystemExit('no admin user in DB')
    h = hash_password(pw)
    for u in users:
        if u.username == 'admin' or u.is_admin:
            u.password_hash = h
            u.password_change_required = False
            db.add(u)
    db.commit()
    # verify hash
    u = db.query(User).filter(User.username == 'admin').first()
    assert u and verify_password(pw, u.password_hash)
    assert not verify_password('admin', u.password_hash)
    print('db_rotated_ok')
finally:
    db.close()
"""
    b64 = base64.b64encode(remote_py.encode()).decode()
    update = f"""set -euo pipefail
printf '%s' '{b64}' | base64 -d > /tmp/rotate_admin_db.py
sudo -u appuser bash -lc 'cd /opt/punimtag && ./venv/bin/python /tmp/rotate_admin_db.py'
sudo -u appuser bash -lc 'cd /opt/punimtag && pm2 restart punimtag-api --update-env'
sleep 5
"""
    subprocess.run(
        ["ssh", "-o", "BatchMode=yes", HOST, f"pct exec {LXC} -- bash -s"],
        input=update.encode(),
        check=True,
    )

    verify = f"""set -euo pipefail
export PW=$(printf '%s' '{base64.b64encode(pw.encode()).decode()}' | base64 -d)
python3 - <<'PY'
import json, os, urllib.request, urllib.error
def login(pw):
    req = urllib.request.Request(
        'http://127.0.0.1:8000/api/v1/auth/login',
        data=json.dumps({{'username':'admin','password':pw}}).encode(),
        headers={{'Content-Type':'application/json'}},
    )
    try:
        with urllib.request.urlopen(req) as r:
            return r.status
    except urllib.error.HTTPError as e:
        return e.code
print('old', login('admin'))
print('new', login(os.environ['PW']))
PY
"""
    out = subprocess.run(
        ["ssh", "-o", "BatchMode=yes", HOST, f"pct exec {LXC} -- bash -s"],
        input=verify.encode(),
        check=True,
        capture_output=True,
    )
    print(out.stdout.decode())
    if b"old 401" not in out.stdout or b"new 200" not in out.stdout:
        print("VERIFY FAILED", out.stderr.decode(), file=sys.stderr)
        return 1
    print(f"OK — password in {TMP} (gitignored). Update Vaultwarden PunimTag admin entry.")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())