feat(auth): CLI-Tool für Notfall-User-Verwaltung + besseres Login-Logging
- app/cli.py: list / create / reset / promote / delete / verify Aufruf: .venv/bin/python -m app.cli <befehl> - Login-Endpoint loggt jetzt getrennt "User nicht gefunden" vs. "Passwort falsch" inkl. Hash-Länge — erlaubt Diagnose ohne Code-Änderung
This commit is contained in:
root
parent
a1167df4f1
commit
44528ef655
@ -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)
|
||||
|
||||
|
||||
|
||||
156
lxc-frontend/app/cli.py
Normal file
156
lxc-frontend/app/cli.py
Normal file
@ -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 <befehl> [...]
|
||||
|
||||
Befehle:
|
||||
list — alle User auflisten
|
||||
create <user> <pass> [--admin] — User anlegen
|
||||
reset <user> <pass> — Passwort setzen
|
||||
promote <user> — zum Admin machen
|
||||
delete <user> — User löschen (nicht den letzten Admin)
|
||||
verify <user> <pass> — 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))
|
||||
Loading…
x
Reference in New Issue
Block a user