"""API key authentication middleware.""" import secrets from fastapi import Request from starlette.middleware.base import BaseHTTPMiddleware from starlette.responses import JSONResponse from app.core.config import settings EXEMPT_PATHS = frozenset([ "/", "/health", "/docs", "/redoc", "/openapi.json", "/api/v1/subtitles/health", "/api/v1/embeddings/health" ]) class APIKeyMiddleware(BaseHTTPMiddleware): """Middleware to enforce API key authentication.""" async def dispatch(self, request: Request, call_next): """Process request and validate API key for protected endpoints.""" if request.url.path in EXEMPT_PATHS: return await call_next(request) api_key = request.headers.get("x-api-key") if not api_key: return JSONResponse( status_code=401, content={"status": "error", "message": "Missing API key. Include 'x-api-key' header."} ) if not settings.api_keys_set: return JSONResponse( status_code=500, content={"status": "error", "message": "No API keys configured on server"} ) is_valid = any( secrets.compare_digest(api_key, valid_key) for valid_key in settings.api_keys_set ) if not is_valid: return JSONResponse( status_code=401, content={"status": "error", "message": "Invalid API key"} ) request.state.api_key = api_key return await call_next(request)