my_voice_assistant_v3/app/auth.py

from fastapi import Header, HTTPException, Request

from app.config import settings, Settings
from app.dependencies import get_store
from app.store import User


def _csv_set(value: str) -> set[str]:
    return {item.strip() for item in (value or "").split(",") if item.strip()}


def is_admin_user(user: User | None, cfg: Settings = settings) -> bool:
    """True, wenn der Nutzer (per SSO-Identitaet) in ADMIN_USERS steht."""
    if user is None or not user.external_id:
        return False
    return user.external_id in _csv_set(cfg.admin_users)


def authenticate(headers, client_host: str, token: str | None,
                 cfg: Settings = settings) -> User | None:
    """Gemeinsame Auth-Logik fuer HTTP und WebSocket.

    Praezedenz:
    1. Forward-Auth: trusted_auth_header gesetzt UND Request von einer Proxy-Quell-IP
       -> Identitaet aus dem Header, interner Nutzer wird ggf. angelegt.
    2. AUTH_ENABLED=false -> anonymer Standardnutzer (dev/Test).
    3. Bearer-Token.
    Liefert den Nutzer oder None (nicht authentifiziert).
    """
    store = get_store()

    # 1. Forward-/Trusted-Header-Auth (nur von der Proxy-Quell-IP akzeptiert).
    if cfg.trusted_auth_header:
        ips = _csv_set(cfg.trusted_proxy_ips)
        if ips and client_host in ips:
            external = (headers.get(cfg.trusted_auth_header) or "").strip()
            if not external:
                return None  # SSO sollte den Header immer setzen -> 401
            user = store.get_or_create_user_by_external_id(external, display_name=external)
            user.is_admin = is_admin_user(user, cfg)
            return user
        # Nicht von der Proxy-IP -> Header ignorieren, normale Auth unten.

    # 2. Auth abgeschaltet (dev/Test).
    if not cfg.auth_enabled:
        return store.ensure_anonymous_user()

    # 3. Bearer-Token.
    if not token:
        return None
    return store.get_user_by_token(token)


def _bearer_token(authorization: str | None) -> str | None:
    if authorization and authorization.lower().startswith("bearer "):
        return authorization.split(" ", 1)[1].strip()
    return None


def require_user(request: Request) -> User:
    """FastAPI-Dependency: liefert den authentifizierten Nutzer (sonst 401)."""
    client_host = request.client.host if request.client else ""
    token = _bearer_token(request.headers.get("authorization"))
    user = authenticate(request.headers, client_host, token)
    if user is None:
        raise HTTPException(status_code=401, detail="Authentication required")
    return user


def require_admin(x_admin_key: str | None = Header(default=None)) -> None:
    """Schuetzt die Nutzerverwaltung ueber ADMIN_API_KEY (aus der Umgebung)."""
    expected = settings.admin_api_key.strip()
    if not expected:
        raise HTTPException(
            status_code=503, detail="Admin API not configured (ADMIN_API_KEY unset)"
        )
    if not x_admin_key or x_admin_key.strip() != expected:
        raise HTTPException(status_code=401, detail="Invalid admin key")


def require_admin_or_user(
    request: Request, x_admin_key: str | None = Header(default=None)
) -> User | None:
    """Erlaubt Zugriff fuer Admin-Nutzer (SSO/ADMIN_USERS) ODER gueltigen ADMIN_API_KEY."""
    expected = settings.admin_api_key.strip()
    if expected and x_admin_key and x_admin_key.strip() == expected:
        return None
    client_host = request.client.host if request.client else ""
    token = _bearer_token(request.headers.get("authorization"))
    user = authenticate(request.headers, client_host, token)
    if user is not None and is_admin_user(user):
        return user
    raise HTTPException(status_code=403, detail="Admin privileges required")