my_voice_assistant_v3_jamulix/app/auth.py

import base64
import hashlib
import hmac
import json

from fastapi import Header, HTTPException, Request

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

# Cookie mit dem Capability-Token (Ein-Klick-Login der Senioren-App).
CAPABILITY_COOKIE = "va_token"


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


def _b64url_decode(data: str) -> bytes:
    return base64.urlsafe_b64decode(data + "=" * (-len(data) % 4))


def _cookie_value(cookie_header: str | None, name: str) -> str | None:
    """Liest einen Cookie-Wert robust aus dem Cookie-Header (ohne SimpleCookie)."""
    if not cookie_header or not name:
        return None
    for part in cookie_header.split(";"):
        part = part.strip()
        if part.startswith(name + "="):
            return part[len(name) + 1:]
    return None


def _username_from_cookie(headers, cfg: Settings) -> str | None:
    """Extrahiert den Usernamen aus einem JWT-Cookie (z. B. YunoHost 'yunohost.portal').

    Mit gesetztem `trusted_auth_jwt_secret` wird die HS256-Signatur geprueft. Ohne
    Secret wird die Payload ungeprueft gelesen - das ist nur sicher, weil (a) nur die
    Proxy-Quell-IP akzeptiert wird und (b) das SSO unauthentifizierte Anfragen gar nicht
    erst durchlaesst (also nur vom SSO validierte Cookies hier ankommen).
    """
    token = _cookie_value(headers.get("cookie"), cfg.trusted_auth_cookie)
    if not token or token.count(".") != 2:
        return None
    header_b64, payload_b64, sig_b64 = token.split(".")
    secret = cfg.trusted_auth_jwt_secret.strip()
    if secret:
        expected = hmac.new(
            secret.encode(), f"{header_b64}.{payload_b64}".encode(), hashlib.sha256
        ).digest()
        try:
            if not hmac.compare_digest(expected, _b64url_decode(sig_b64)):
                return None
        except (ValueError, TypeError):
            return None
    try:
        payload = json.loads(_b64url_decode(payload_b64))
    except (ValueError, TypeError):
        return None
    value = payload.get(cfg.trusted_auth_cookie_claim)
    return value.strip() if isinstance(value, str) and value.strip() else None


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:
        return False
    if getattr(user, "is_admin", False):
        return True  # persistentes DB-Flag (vom Admin gesetzt)
    if 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-Auth: Identitaet aus Header ODER (signiertem) Cookie - nur von der
    #    Proxy-Quell-IP akzeptiert.
    if cfg.trusted_auth_header or cfg.trusted_auth_cookie:
        ips = _csv_set(cfg.trusted_proxy_ips)
        if ips and client_host in ips:
            external = None
            if cfg.trusted_auth_header:
                external = (headers.get(cfg.trusted_auth_header) or "").strip() or None
            if not external and cfg.trusted_auth_cookie:
                external = _username_from_cookie(headers, cfg)
            if external:
                user = store.get_or_create_user_by_external_id(external, display_name=external)
                user.is_admin = is_admin_user(user, cfg)
                return user
            # Kein SSO-Identitaetsheader (z. B. App-Route ohne Authelia) -> unten
            # auf Capability-/Bearer-Token-Auth durchfallen.
        # Nicht von der Proxy-IP -> 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"))
        or request.query_params.get("k")
        or request.cookies.get(CAPABILITY_COOKIE)
    )
    user = authenticate(request.headers, client_host, token)
    if user is None:
        raise HTTPException(status_code=401, detail="Authentication required")
    return user


def require_admin(
    request: Request, x_admin_key: str | None = Header(default=None)
) -> None:
    """Schuetzt Admin-Endpunkte: ADMIN_API_KEY-Header ODER SSO-Admin-Nutzer."""
    expected = settings.admin_api_key.strip()
    if expected and x_admin_key and x_admin_key.strip() == expected:
        return
    client_host = request.client.host if request.client else ""
    token = _bearer_token(request.headers.get("authorization"))
    user = authenticate(request.headers, client_host, token)
    if is_admin_user(user):
        return
    if not expected:
        raise HTTPException(status_code=503, detail="Admin API not configured (ADMIN_API_KEY unset)")
    raise HTTPException(status_code=401, detail="Admin privileges required")


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