diff --git a/lxc-frontend/app/api/auth.py b/lxc-frontend/app/api/auth.py index 3935972..fe016d0 100644 --- a/lxc-frontend/app/api/auth.py +++ b/lxc-frontend/app/api/auth.py @@ -49,11 +49,19 @@ def needs_setup(session: Session = Depends(get_session)): @router.post("/login") def login(payload: LoginIn, request: Request, session: Session = Depends(get_session)): - user = get_user_by_username(session, payload.username.strip()) - if not user or not verify_password(payload.password, user.password_hash): + username = payload.username.strip() + user = get_user_by_username(session, username) + if not user: + log.warning("Login fehlgeschlagen: User '%s' nicht gefunden", username) + raise HTTPException(status.HTTP_401_UNAUTHORIZED, "Benutzername oder Passwort falsch") + if not verify_password(payload.password, user.password_hash): + log.warning( + "Login fehlgeschlagen: Passwort falsch für user=%s (hash_len=%d, prefix=%s)", + user.username, len(user.password_hash), user.password_hash[:7], + ) raise HTTPException(status.HTTP_401_UNAUTHORIZED, "Benutzername oder Passwort falsch") request.session["user_id"] = user.id - log.info("Login: user=%s id=%d admin=%s", user.username, user.id, user.is_admin) + log.info("Login OK: user=%s id=%d admin=%s", user.username, user.id, user.is_admin) return _user_dict(user) diff --git a/lxc-frontend/app/cli.py b/lxc-frontend/app/cli.py new file mode 100644 index 0000000..74abdad --- /dev/null +++ b/lxc-frontend/app/cli.py @@ -0,0 +1,156 @@ +"""CLI-Tool für Notfall-User-Verwaltung. + +Aufruf am LXC: + sudo -u deploy /var/www/voice-agent/lxc-frontend/.venv/bin/python -m app.cli [...] + +Befehle: + list — alle User auflisten + create [--admin] — User anlegen + reset — Passwort setzen + promote — zum Admin machen + delete — User löschen (nicht den letzten Admin) + verify — Passwort testen (sagt nur OK / NEIN) +""" +from __future__ import annotations + +import sys + +from sqlmodel import Session, select + +from app.core.auth import hash_password, verify_password +from app.core.db import engine, init_db +from app.core.migrate import run_migrations +from app.models.user import User + + +def _setup() -> None: + init_db() + run_migrations() + + +def cmd_list() -> int: + _setup() + with Session(engine) as s: + users = s.exec(select(User).order_by(User.id)).all() + if not users: + print("(keine User in der DB)") + return 0 + print(f"{'ID':>4} {'USERNAME':<30} {'ADMIN':<6} {'HASH':<10} CREATED") + for u in users: + print( + f"{u.id:>4} {u.username:<30} {'ja' if u.is_admin else 'nein':<6} " + f"len={len(u.password_hash):<5} {u.created_at.isoformat(timespec='seconds')}" + ) + return 0 + + +def cmd_create(username: str, password: str, is_admin: bool) -> int: + _setup() + if len(password) < 8: + print("FEHLER: Passwort muss min. 8 Zeichen haben.", file=sys.stderr) + return 2 + with Session(engine) as s: + if s.exec(select(User).where(User.username == username)).first(): + print(f"FEHLER: User '{username}' existiert bereits.", file=sys.stderr) + return 2 + u = User(username=username, password_hash=hash_password(password), is_admin=is_admin) + s.add(u) + s.commit() + s.refresh(u) + print(f"OK — User #{u.id} '{u.username}' angelegt (admin={u.is_admin}).") + return 0 + + +def cmd_reset(username: str, password: str) -> int: + _setup() + if len(password) < 8: + print("FEHLER: Passwort muss min. 8 Zeichen haben.", file=sys.stderr) + return 2 + with Session(engine) as s: + u = s.exec(select(User).where(User.username == username)).first() + if not u: + print(f"FEHLER: User '{username}' nicht gefunden.", file=sys.stderr) + return 2 + u.password_hash = hash_password(password) + s.add(u) + s.commit() + print(f"OK — Passwort für '{username}' gesetzt.") + return 0 + + +def cmd_promote(username: str) -> int: + _setup() + with Session(engine) as s: + u = s.exec(select(User).where(User.username == username)).first() + if not u: + print(f"FEHLER: User '{username}' nicht gefunden.", file=sys.stderr) + return 2 + if u.is_admin: + print(f"OK — '{username}' war bereits Admin.") + return 0 + u.is_admin = True + s.add(u) + s.commit() + print(f"OK — '{username}' ist jetzt Admin.") + return 0 + + +def cmd_delete(username: str) -> int: + _setup() + with Session(engine) as s: + u = s.exec(select(User).where(User.username == username)).first() + if not u: + print(f"FEHLER: User '{username}' nicht gefunden.", file=sys.stderr) + return 2 + if u.is_admin: + admins = s.exec(select(User).where(User.is_admin == True)).all() # noqa: E712 + if len(admins) <= 1: + print("FEHLER: letzten Admin kann man nicht löschen.", file=sys.stderr) + return 2 + s.delete(u) + s.commit() + print(f"OK — '{username}' gelöscht.") + return 0 + + +def cmd_verify(username: str, password: str) -> int: + _setup() + with Session(engine) as s: + u = s.exec(select(User).where(User.username == username)).first() + if not u: + print(f"User '{username}' existiert nicht.") + return 1 + if verify_password(password, u.password_hash): + print(f"OK — Passwort für '{username}' stimmt.") + return 0 + print(f"NEIN — Passwort stimmt nicht. Hash-Länge in DB: {len(u.password_hash)}, Präfix: {u.password_hash[:7]}") + return 1 + + +def main(argv: list[str]) -> int: + if len(argv) < 2: + print(__doc__) + return 1 + cmd = argv[1] + try: + if cmd == "list": + return cmd_list() + if cmd == "create" and len(argv) >= 4: + return cmd_create(argv[2], argv[3], "--admin" in argv[4:]) + if cmd == "reset" and len(argv) == 4: + return cmd_reset(argv[2], argv[3]) + if cmd == "promote" and len(argv) == 3: + return cmd_promote(argv[2]) + if cmd == "delete" and len(argv) == 3: + return cmd_delete(argv[2]) + if cmd == "verify" and len(argv) == 4: + return cmd_verify(argv[2], argv[3]) + except Exception as e: # noqa: BLE001 + print(f"FEHLER: {e}", file=sys.stderr) + return 3 + print(__doc__) + return 1 + + +if __name__ == "__main__": + sys.exit(main(sys.argv))