import asyncio import time from contextlib import asynccontextmanager from pathlib import Path from fastapi import FastAPI, Request from fastapi.staticfiles import StaticFiles from starlette.responses import RedirectResponse, JSONResponse, HTMLResponse from app.config import settings from urllib.parse import urlencode from app.auth import authenticate, _bearer_token, CAPABILITY_COOKIE from app.dependencies import get_store from app.core.warmup import warmup_local_models from app.metrics import metrics from app.api.health import router as health_router from app.api.chat import router as chat_router from app.api.transcribe import router as transcribe_router from app.api.speak import router as speak_router from app.api.devices import router as devices_router from app.api.sessions import router as sessions_router from app.api.config import router as config_router from app.api.admin import router as admin_router from app.api.me import router as me_router from app.api.metrics import router as metrics_router from app.api.ws import router as ws_router # --- Logging: App-eigene Logger sichtbar machen --------------------------- # uvicorn konfiguriert nur seine eigenen Logger; "app.*"-INFO-Logs (z. B. die # Notruf-Diagnose) würden sonst durch den Default-Level (WARNING) verschluckt. import logging import os _app_logger = logging.getLogger("app") _app_logger.setLevel(getattr(logging, os.getenv("LOG_LEVEL", "INFO").upper(), logging.INFO)) if not _app_logger.handlers: _log_handler = logging.StreamHandler() _log_handler.setFormatter(logging.Formatter("%(levelname)s:%(name)s:%(message)s")) _app_logger.addHandler(_log_handler) _app_logger.propagate = False @asynccontextmanager async def _lifespan(app: FastAPI): # Lokale Modelle im Hintergrund vorladen -> Server ist sofort verfuegbar, # der erste Nutzer zahlt nicht den Kaltstart. task = asyncio.create_task(warmup_local_models()) yield if not task.done(): task.cancel() app = FastAPI(title="Voice Assistant Gateway", lifespan=_lifespan) @app.middleware("http") async def record_metrics(request: Request, call_next): start = time.perf_counter() response = await call_next(request) duration = time.perf_counter() - start # Route-Template (z. B. /api/sessions/{session_id}/route) statt konkreter URL, # um die Label-Kardinalitaet niedrig zu halten. route = request.scope.get("route") path = getattr(route, "path", request.url.path) labels = {"method": request.method, "path": path} metrics.inc("http_requests_total", {**labels, "status": response.status_code}) metrics.observe("http_request_duration_seconds", duration, labels) return response # Pfade mit eigener Auth, Health-Checks und Favicons: nicht gaten. _NO_ACCESS_PAGE = """ Zugang
🔒

Persönlicher Zugang nötig

Senioren: bitte die App über den persönlichen Zugangslink öffnen.

Betreuer und Admins können sich mit Benutzername und Passwort anmelden:

Mit Benutzername & Passwort anmelden
""" _PUBLIC_PREFIXES = ("/api", "/ws", "/health", "/favicon", "/apple-touch-icon") @app.middleware("http") async def gate_web_ui(request: Request, call_next): """Schuetzt die statische Web-UI: unauthentifizierte Seitenaufrufe -> SSO-Login (sonst 401). Defense-in-Depth zusaetzlich zu SSOwat: Selbst wenn der Reverse-Proxy einen unauthentifizierten Request durchliesse (oder jemand den Port direkt trifft), bekommt er die Seite nicht ausgeliefert. API/WS haben ihre eigene Auth (require_user -> 401/JSON), Health-Checks bleiben offen. Bei abgeschalteter Auth (LAN-Dev) liefert authenticate() einen anonymen Nutzer -> die Seite wird wie bisher ausgeliefert. """ path = request.url.path if not path.startswith(_PUBLIC_PREFIXES): # Ein-Klick-Link: ?k= -> Token im Cookie speichern, aus der URL entfernen. k = request.query_params.get("k") if k and get_store().get_user_by_token(k): params = {kk: vv for kk, vv in request.query_params.items() if kk != "k"} clean = path + ("?" + urlencode(params) if params else "") resp = RedirectResponse(clean, status_code=303) resp.set_cookie(CAPABILITY_COOKIE, k, httponly=True, secure=True, samesite="lax", max_age=31536000) return resp client_host = request.client.host if request.client else "" token = (_bearer_token(request.headers.get("authorization")) or k or request.cookies.get(CAPABILITY_COOKIE)) if authenticate(request.headers, client_host, token) is None: # Kein gültiger Zugang -> freundliche Hinweisseite (KEIN SSO-Redirect, der # nach dem nginx-Cutover in eine Schleife liefe). Senioren nutzen ihren Link. return HTMLResponse(_NO_ACCESS_PAGE, status_code=401) return await call_next(request) app.include_router(health_router) app.include_router(chat_router, prefix="/api") app.include_router(transcribe_router, prefix="/api") app.include_router(speak_router, prefix="/api") app.include_router(devices_router, prefix="/api") app.include_router(sessions_router, prefix="/api") app.include_router(config_router, prefix="/api") app.include_router(admin_router, prefix="/api") app.include_router(me_router, prefix="/api") app.include_router(metrics_router, prefix="/api") app.include_router(ws_router) # Statische Web-UI (same-origin -> kein CORS). Muss NACH allen API-/WS-Routen # gemountet werden, damit "/" nur die uebrigen Pfade abfaengt. class _NoCacheStaticFiles(StaticFiles): """Liefert die UI ohne Caching aus -> Aenderungen sind sofort sichtbar (kein Safari-Cache).""" def file_response(self, *args, **kwargs): response = super().file_response(*args, **kwargs) response.headers["Cache-Control"] = "no-cache, no-store, must-revalidate" return response _WEB_DIR = Path(__file__).resolve().parent / "web" if _WEB_DIR.is_dir(): app.mount("/", _NoCacheStaticFiles(directory=str(_WEB_DIR), html=True), name="web")