File size: 1,729 Bytes
9db586c 20548ac 6851411 20548ac 6851411 20548ac ee07ed2 20548ac ee07ed2 20548ac ee07ed2 6851411 ee07ed2 20548ac 9db586c ee07ed2 20548ac 6851411 |
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 |
import hmac
from fastapi import Request
from fastapi.responses import JSONResponse, Response
from typing import Callable, Awaitable, Union
from app.config import settings
# Public endpoints that don't require authentication
PUBLIC_PATHS = frozenset(["/", "/health", "/docs", "/redoc", "/openapi.json"])
async def api_key_guard(request: Request, call_next: Callable[[Request], Awaitable[Response]]) -> Union[Response, JSONResponse]:
"""
Middleware to protect API endpoints with optional API key authentication.
Args:
request: FastAPI request object
call_next: Next middleware/handler in the chain
Returns:
Response from next handler or 401 if unauthorized
"""
# Skip auth for public endpoints
if request.url.path in PUBLIC_PATHS:
return await call_next(request)
# Skip auth if no API key is configured
if not settings.service_api_key:
return await call_next(request)
# Check API key from headers
api_key = request.headers.get("x-api-key")
if not api_key:
# Also check Authorization header with Bearer token
auth_header = request.headers.get("authorization", "")
if auth_header.startswith("Bearer "):
api_key = auth_header.replace("Bearer ", "").strip()
if api_key:
# Use constant-time comparison to prevent timing attacks
expected_key = str(settings.service_api_key) if settings.service_api_key else ""
if hmac.compare_digest(str(api_key), expected_key):
return await call_next(request)
return JSONResponse(
content={"error": {"message": "unauthorized", "type": "authentication_error"}},
status_code=401
)
|