"""Anmeldung über server-seitige Sitzung im Cookie, inklusive Zwei-Faktor. Ablauf mit aktivem TOTP: ``/login`` prüft das Passwort und legt eine Sitzung an, die als ``pending_2fa`` markiert ist – die gilt nirgends als angemeldet. Erst ``/2fa/verify`` tauscht sie gegen eine echte Sitzung (neues Cookie, damit kein vorab untergeschobenes Token weiterlebt). """ from datetime import datetime, timezone from fastapi import APIRouter, Depends, HTTPException, Request, Response, status from fastapi.security import OAuth2PasswordRequestForm from sqlalchemy.orm import Session from ..config import get_settings from ..database import get_db from ..deps import get_current_user, user_out from ..models import User from ..schemas import ( LoginOut, PasswordChange, RecoveryCodesOut, TwoFactorActivate, TwoFactorPassword, TwoFactorSetupOut, TwoFactorStatusOut, TwoFactorVerify, UserOut, ) from ..security import hash_password, needs_rehash, verify_password from ..services import policy as policy_service from ..services import sessions, twofactor router = APIRouter(prefix="/auth", tags=["auth"]) def _set_session_cookie(response: Response, raw_token: str, max_age: int | None = None) -> None: settings = get_settings() response.set_cookie( key=settings.cookie_name, value=raw_token, max_age=max_age if max_age is not None else settings.session_ttl_days * 86400, httponly=True, secure=settings.cookie_secure, samesite=settings.cookie_samesite, path="/", ) @router.post("/login", response_model=LoginOut) def login( response: Response, request: Request, form: OAuth2PasswordRequestForm = Depends(), db: Session = Depends(get_db), ) -> LoginOut: user = db.query(User).filter(User.username == form.username).first() # Ein reines SSO-Konto hat kein Passwort. Der Vergleich wird gar nicht erst # versucht – argon2 bekaeme sonst None und wuerfe einen Fehler, aus dem # herauszulesen waere, dass es diesen Benutzernamen gibt. ohne_passwort = user is not None and user.password_hash is None if user is None or ohne_passwort or not verify_password(form.password, user.password_hash): raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Benutzername oder Passwort falsch", ) if not user.is_active: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Dieses Konto ist deaktiviert", ) # Einziger Moment, in dem das Passwort im Klartext vorliegt: hier lässt sich # prüfen, ob es die (womöglich verschärfte) Richtlinie noch erfüllt. Über den # gespeicherten Hash ginge das nicht. if policy_service.violations(form.password, policy_service.load(db)): user.must_change_password = True user.must_change_password_reason = "policy" if needs_rehash(user.password_hash): user.password_hash = hash_password(form.password) db.commit() if user.totp_enabled: raw = sessions.create_session( db, user, request.headers.get("user-agent"), pending_2fa=True ) _set_session_cookie(response, raw, max_age=sessions.PENDING_TTL_MINUTES * 60) return LoginOut(status="2fa_required", user=None) raw = sessions.create_session(db, user, request.headers.get("user-agent")) _set_session_cookie(response, raw) return LoginOut(status="ok", user=user_out(db, user)) def _bestaetige_mit_passwort(user: User, passwort: str) -> None: """Die eigene Identitaet mit dem Passwort bestaetigen. Ein reines SSO-Konto hat keines – dort ist diese Bestaetigung nicht "falsch", sondern gar nicht moeglich. Ohne diese Unterscheidung bekaeme argon2 None und der Nutzer eine Fehlermeldung, die nichts erklaert. """ if user.password_hash is None: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=("Dieses Konto meldet sich ueber einen Anmeldedienst an und " "hat kein Passwort."), ) if not verify_password(passwort, user.password_hash): raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Passwort ist falsch" ) @router.post("/logout", status_code=status.HTTP_204_NO_CONTENT) def logout(request: Request, response: Response, db: Session = Depends(get_db)) -> None: token = request.cookies.get(get_settings().cookie_name, "") if token: sessions.revoke(db, token) response.delete_cookie(get_settings().cookie_name, path="/") @router.get("/me", response_model=UserOut) def me( current_user: User = Depends(get_current_user), db: Session = Depends(get_db) ) -> UserOut: return user_out(db, current_user) @router.post("/change-password", response_model=UserOut) def change_password( request: Request, response: Response, payload: PasswordChange, current_user: User = Depends(get_current_user), db: Session = Depends(get_db), ) -> UserOut: if current_user.password_hash is None: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=("Dieses Konto meldet sich ueber einen Anmeldedienst an und " "hat kein Passwort."), ) if not verify_password(payload.current_password, current_user.password_hash): raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Aktuelles Passwort ist falsch", ) if payload.new_password == payload.current_password: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Das neue Passwort muss sich vom bisherigen unterscheiden", ) policy_service.assert_ok(db, payload.new_password) current_user.password_hash = hash_password(payload.new_password) current_user.must_change_password = False current_user.must_change_password_reason = None db.commit() # Ein Passwortwechsel soll alle übrigen Sitzungen beenden – wer das Passwort # kannte, soll danach draussen sein. Der aufrufende Client bekommt sofort # eine frische Sitzung, damit er nicht selbst herausfliegt. sessions.revoke_all(db, current_user) raw = sessions.create_session(db, current_user, request.headers.get("user-agent")) _set_session_cookie(response, raw) db.refresh(current_user) return user_out(db, current_user) # --- Zwei-Faktor ------------------------------------------------------------ @router.post("/2fa/verify", response_model=LoginOut) def verify_two_factor( request: Request, response: Response, payload: TwoFactorVerify, db: Session = Depends(get_db), ) -> LoginOut: """Zweiter Schritt der Anmeldung: TOTP-Code oder Wiederherstellungscode.""" token = request.cookies.get(get_settings().cookie_name, "") offen = sessions.validate_pending(db, token) if offen is None: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Anmeldung abgelaufen – bitte neu anmelden", ) entry, user = offen ok = ( twofactor.consume_recovery_code(db, user, payload.code) if payload.recovery else twofactor.consume_code(db, user, payload.code) ) if not ok: verbraucht = sessions.note_failed_2fa(db, entry) if verbraucht: response.delete_cookie(get_settings().cookie_name, path="/") raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Zu viele Fehlversuche – bitte neu anmelden", ) raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Code stimmt nicht" ) # Cookie tauschen statt weiterverwenden: die Pending-Sitzung verschwindet, # der Client bekommt ein frisches Token. db.delete(entry) db.commit() raw = sessions.create_session(db, user, request.headers.get("user-agent")) _set_session_cookie(response, raw) return LoginOut(status="ok", user=user_out(db, user)) @router.get("/2fa/status", response_model=TwoFactorStatusOut) def two_factor_status( current_user: User = Depends(get_current_user), db: Session = Depends(get_db) ) -> TwoFactorStatusOut: return TwoFactorStatusOut( enabled=current_user.totp_enabled, recovery_codes_left=twofactor.codes_left(db, current_user), ) @router.post("/2fa/setup", response_model=TwoFactorSetupOut) def setup_two_factor( current_user: User = Depends(get_current_user), db: Session = Depends(get_db) ) -> TwoFactorSetupOut: """Legt ein neues, noch nicht aktives Secret an und liefert den QR-Code.""" if current_user.totp_enabled: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Zwei-Faktor ist bereits aktiv", ) secret = twofactor.new_secret() current_user.totp_secret = secret current_user.totp_last_step = None db.commit() uri = twofactor.otpauth_uri(current_user, secret) return TwoFactorSetupOut(secret=secret, otpauth_uri=uri, qr_png=twofactor.qr_data_uri(uri)) @router.post("/2fa/activate", response_model=RecoveryCodesOut) def activate_two_factor( payload: TwoFactorActivate, current_user: User = Depends(get_current_user), db: Session = Depends(get_db), ) -> RecoveryCodesOut: if current_user.totp_enabled: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Zwei-Faktor ist bereits aktiv", ) if not current_user.totp_secret: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Zuerst die Einrichtung starten", ) if not twofactor.consume_code(db, current_user, payload.code): raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Code stimmt nicht") current_user.totp_enabled = True current_user.totp_confirmed_at = datetime.now(timezone.utc) db.commit() return RecoveryCodesOut(codes=twofactor.generate_recovery_codes(db, current_user)) @router.post("/2fa/recovery-codes", response_model=RecoveryCodesOut) def new_recovery_codes( payload: TwoFactorPassword, current_user: User = Depends(get_current_user), db: Session = Depends(get_db), ) -> RecoveryCodesOut: if not current_user.totp_enabled: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Zwei-Faktor ist nicht aktiv" ) _bestaetige_mit_passwort(current_user, payload.password) return RecoveryCodesOut(codes=twofactor.generate_recovery_codes(db, current_user)) @router.post("/2fa/disable", response_model=UserOut) def disable_two_factor( payload: TwoFactorPassword, current_user: User = Depends(get_current_user), db: Session = Depends(get_db), ) -> UserOut: _bestaetige_mit_passwort(current_user, payload.password) if policy_service.load(db).mfa_required: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Die Richtlinie verlangt Zwei-Faktor – Abschalten ist nicht möglich", ) twofactor.disable(db, current_user) db.refresh(current_user) return user_out(db, current_user)