certain/backend/app/deps.py

"""Gemeinsame FastAPI-Abhängigkeiten: aktueller Benutzer (Cookie), Admin-Pflicht.

Die Anmeldung läuft über eine server-seitige Sitzung im Cookie (siehe
``services/sessions``). ``get_current_user`` liest das Cookie, prüft die Sitzung
und liefert den Benutzer – oder 401.

``enforce_account_state`` hängt am zentralen ``/api``-Router und sperrt Benutzer
aus, die erst noch etwas erledigen müssen (Passwort wechseln, zweiten Faktor
einrichten). Bisher wurde das nur im Frontend umgeleitet – wer die API direkt
ansprach, kam daran vorbei.
"""

from datetime import datetime, timezone

from fastapi import Depends, HTTPException, Request, status
from sqlalchemy.orm import Session

from .config import get_settings
from .database import get_db
from .models import ApiToken, Role, TokenScope, User
from .schemas import UserOut
from .services import policy as policy_service
from .services import sessions
from .security import hash_session_token

_credentials_exc = HTTPException(
    status_code=status.HTTP_401_UNAUTHORIZED,
    detail="Nicht angemeldet oder Sitzung abgelaufen",
)

# Pfade, die trotz offener Pflicht erreichbar bleiben müssen – sonst könnte
# niemand die Pflicht je erfüllen. Präfix-Vergleich, ohne führendes /api.
_ALWAYS_ALLOWED = (
    "/health",
    "/version",
    "/auth/me",
    "/auth/logout",
    "/auth/login",
    "/auth/change-password",
    "/auth/2fa",
)
# Die Richtlinie darf gelesen werden (die Checkliste braucht sie), aber ein
# Admin mit offener Pflicht soll sie nicht erst einmal entschärfen können.
_READ_ONLY_ALLOWED = ("/policy",)


def _session_user(request: Request, db: Session) -> User | None:
    """Validiert die Sitzung genau einmal pro Anfrage und merkt sie sich.

    ``enforce_account_state`` läuft vor den Endpunkt-Abhängigkeiten; ohne diesen
    Zwischenspeicher würde ``sessions.validate`` (und damit der Schreibzugriff
    auf ``last_seen_at``) pro Anfrage zweimal laufen."""
    if hasattr(request.state, "session_user"):
        return request.state.session_user
    token = request.cookies.get(get_settings().cookie_name, "")
    user = sessions.validate(db, token)
    request.state.session_user = user
    return user


def get_current_user(request: Request, db: Session = Depends(get_db)) -> User:
    user = _session_user(request, db)
    if user is None:
        raise _credentials_exc
    return user


def pending_reason(db: Session, user: User) -> str | None:
    """Was der Benutzer erledigen muss, bevor er die App benutzen darf."""
    if user.must_change_password:
        return "password_change_required"
    if policy_service.load(db).mfa_required and not user.totp_enabled:
        return "mfa_setup_required"
    return None


def enforce_account_state(request: Request, db: Session = Depends(get_db)) -> None:
    """Router-weite Sperre für Konten mit offener Pflicht (403, sonst Durchlass).

    Nicht angemeldete Anfragen laufen hier ungehindert durch – um die kümmert
    sich ``get_current_user`` am jeweiligen Endpunkt."""
    pfad = request.url.path
    rest = pfad[4:] if pfad.startswith("/api") else pfad
    if rest.startswith(_ALWAYS_ALLOWED):
        return
    if request.method == "GET" and rest.startswith(_READ_ONLY_ALLOWED):
        return
    user = _session_user(request, db)
    if user is None:
        return
    grund = pending_reason(db, user)
    if grund is not None:
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=grund)


_token_exc = HTTPException(
    status_code=status.HTTP_401_UNAUTHORIZED,
    detail="Ungültige oder fehlende Gerätemarke",
)


def require_quick_token(request: Request, db: Session = Depends(get_db)) -> ApiToken:
    """Prüft die Gerätemarke im Header ``X-Certain-Token``.

    Bewusst eine EIGENE Abhängigkeit statt eines Sonderfalls in
    ``get_current_user``: So gilt eine Marke ausschliesslich an den Endpunkten,
    die sie ausdrücklich verlangen. Auf allen übrigen Wegen ist sie schlicht
    keine Anmeldung – wer sie erbeutet, kommt an keine Daten.
    """
    roh = (request.headers.get("X-Certain-Token") or "").strip()
    if not roh:
        raise _token_exc
    marke = (
        db.query(ApiToken)
        .filter(ApiToken.token_hash == hash_session_token(roh),
                ApiToken.scope == TokenScope.quick_add)
        .first()
    )
    if marke is None:
        raise _token_exc
    besitzer = db.get(User, marke.owner_id)
    if besitzer is None or not besitzer.is_active:
        raise _token_exc
    # Wer sein Passwort wechseln muss, soll auch nicht per Kurzbefehl buchen.
    grund = pending_reason(db, besitzer)
    if grund is not None:
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=grund)
    marke.last_used_at = datetime.now(timezone.utc)
    db.commit()
    request.state.token_user = besitzer
    return marke


def require_admin(current_user: User = Depends(get_current_user)) -> User:
    if current_user.role != Role.admin:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Nur Administratoren dürfen diese Aktion ausführen",
        )
    return current_user


def user_out(db: Session, user: User) -> UserOut:
    """``UserOut`` inklusive des berechneten ``mfa_setup_required``."""
    out = UserOut.model_validate(user)
    out.mfa_setup_required = policy_service.load(db).mfa_required and not user.totp_enabled
    return out


def get_owned(db: Session, model, obj_id: int, user: User):
    """Holt eine Zeile und stellt sicher, dass sie dem Nutzer gehört – sonst 404.

    Bewusst 404 (nicht 403): fremde IDs sollen nicht einmal als existent gelten.
    Zentrale Stelle für die Mandanten-Isolation."""
    obj = db.get(model, obj_id)
    if obj is None or getattr(obj, "owner_id", None) != user.id:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Nicht gefunden")
    return obj