76df695111
- Minimale Web-UI (app/web/, vanilla, same-origin -> kein CORS): Text-Prompt + Mikrofon-Button (Aufnahme im Browser -> /ws/voice -> Antwort wird vorgelesen), Token-Streaming, PCM-Wiedergabe, Identitaet/Logout/Admin im Menue - Forward-/Trusted-Header-Auth (app/auth.py): Identitaet aus SSO-Header, nur von TRUSTED_PROXY_IPS akzeptiert; sonst Token/Anonymous-Fallback. Auto-Provisioning via store.get_or_create_user_by_external_id (+ external_id-Spalte/Migration) - /api/me um is_admin + sso_logout_url erweitert; GET /api/admin/users (Liste) und GET /api/admin/request-headers (SSO-Header-Discovery), Admin-gated - StaticFiles-Mount; Config: TRUSTED_AUTH_HEADER/_PROXY_IPS, ADMIN_USERS, SSO_LOGOUT_URL - WS-Auth liest Identitaet aus dem Handshake-Header - Deploy: nginx-Vorlage (WS-Upgrade!) + deploy/README.md (HTTPS/SSO/Firewall/Discovery) - Tests: Forward-Auth (Provisioning, Admin-Flag, Proxy-IP-Trust, 401/403, Static) Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
93 lines
3.6 KiB
Python
93 lines
3.6 KiB
Python
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")
|