itssKarthiii's picture
Upload 70 files
6b408d7 verified
"""
API Key authentication middleware.
Provides middleware for API key validation.
"""
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import JSONResponse
from app.config import get_settings
from app.utils.logger import get_logger
logger = get_logger(__name__)
class APIKeyMiddleware(BaseHTTPMiddleware):
"""
Middleware for API key authentication.
This middleware checks for valid API keys on protected endpoints.
Public endpoints (health, docs) are excluded from authentication.
"""
# Endpoints that don't require authentication
PUBLIC_PATHS: set[str] = {
"/api/health",
"/api/ready",
"/api/languages",
"/api/",
"/docs",
"/redoc",
"/openapi.json",
"/",
}
async def dispatch(self, request: Request, call_next):
"""
Process request and validate API key for protected endpoints.
Args:
request: Incoming request
call_next: Next middleware/handler
Returns:
Response from next handler or 401 error
"""
# Skip authentication for public paths
path = request.url.path.rstrip("/") or "/"
# Check if path is public
is_public = path in self.PUBLIC_PATHS or any(
path.startswith(public.rstrip("/")) for public in self.PUBLIC_PATHS if public != "/"
)
if is_public:
return await call_next(request)
# Get API key from header
api_key = request.headers.get("x-api-key")
if not api_key:
logger.warning(
"Request without API key",
path=path,
method=request.method,
)
return JSONResponse(
status_code=401,
content={
"status": "error",
"message": "API key is required",
},
headers={"WWW-Authenticate": "ApiKey"},
)
# Validate API key
settings = get_settings()
valid_keys = settings.api_keys_list
if not valid_keys:
logger.error("No API keys configured")
return JSONResponse(
status_code=500,
content={
"status": "error",
"message": "Server configuration error",
},
)
if api_key not in valid_keys:
logger.warning(
"Invalid API key",
path=path,
key_prefix=api_key[:8] + "..." if len(api_key) > 8 else "***",
)
return JSONResponse(
status_code=401,
content={
"status": "error",
"message": "Invalid API key",
},
headers={"WWW-Authenticate": "ApiKey"},
)
# Continue to next handler
return await call_next(request)