Spaces:
Running
Running
File size: 1,591 Bytes
9fe5a30 781f7b0 9fe5a30 781f7b0 9fe5a30 781f7b0 9fe5a30 781f7b0 9fe5a30 781f7b0 9fe5a30 781f7b0 9fe5a30 781f7b0 9fe5a30 781f7b0 9fe5a30 781f7b0 9fe5a30 781f7b0 9fe5a30 781f7b0 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 |
"""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) |