certain/backend/app/routers/auth.py

"""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)