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