82 lines
2.8 KiB
Python
82 lines
2.8 KiB
Python
"""Optional API-key auth.
|
|
|
|
Behaviour:
|
|
|
|
- If ``API_KEY`` is empty (default) every request is allowed - matches the
|
|
original dev configuration.
|
|
- If ``API_KEY`` is set, every request to a route under ``app_api_prefix``
|
|
must carry either ``X-API-Key: <value>`` or ``Authorization: Bearer <value>``.
|
|
- ``/health`` is intentionally exempt so external probes (compose healthcheck,
|
|
reverse proxy, monitoring) keep working without leaking the key.
|
|
- The root ``/`` page stays open so the OpenAPI banner and docs links remain
|
|
reachable.
|
|
|
|
This is a defence-in-depth layer behind whatever reverse proxy / OAuth gateway
|
|
runs in production - not a replacement.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hmac
|
|
from collections.abc import Awaitable, Callable
|
|
|
|
from fastapi import FastAPI, Request, Response
|
|
from fastapi.responses import JSONResponse
|
|
from starlette.types import ASGIApp
|
|
|
|
EXEMPT_PATHS: tuple[str, ...] = ("/", "/docs", "/redoc", "/openapi.json")
|
|
EXEMPT_SUFFIXES: tuple[str, ...] = ("/health",)
|
|
|
|
|
|
def _extract_token(request: Request) -> str | None:
|
|
header = request.headers.get("x-api-key")
|
|
if header:
|
|
return header.strip()
|
|
auth = request.headers.get("authorization") or ""
|
|
if auth.lower().startswith("bearer "):
|
|
return auth[7:].strip()
|
|
return None
|
|
|
|
|
|
def install_api_key_auth(app: FastAPI) -> None:
|
|
"""Attach the middleware. Always safe to call; becomes a no-op when no key
|
|
is configured.
|
|
|
|
Reads ``app.config.settings`` lazily so test fixtures can reload the config
|
|
module and have the new ``API_KEY`` value take effect on the next install.
|
|
"""
|
|
from app.config import settings as fresh_settings # re-resolve after reloads
|
|
|
|
settings = fresh_settings
|
|
expected = settings.api_key.strip() if settings.api_key else ""
|
|
if not expected:
|
|
return
|
|
|
|
@app.middleware("http")
|
|
async def _api_key_middleware( # type: ignore[no-redef]
|
|
request: Request,
|
|
call_next: Callable[[Request], Awaitable[Response]],
|
|
) -> Response:
|
|
path = request.url.path
|
|
if request.method == "OPTIONS":
|
|
return await call_next(request)
|
|
if path in EXEMPT_PATHS:
|
|
return await call_next(request)
|
|
if any(path.endswith(s) for s in EXEMPT_SUFFIXES):
|
|
return await call_next(request)
|
|
if not path.startswith(settings.app_api_prefix):
|
|
return await call_next(request)
|
|
|
|
token = _extract_token(request)
|
|
if not token or not hmac.compare_digest(token, expected):
|
|
return JSONResponse(
|
|
status_code=401,
|
|
content={"detail": "invalid or missing api key"},
|
|
headers={"WWW-Authenticate": "Bearer"},
|
|
)
|
|
return await call_next(request)
|
|
|
|
|
|
__all__ = ["install_api_key_auth"]
|
|
_ = ASGIApp # re-export hint to keep mypy happy on older Starlette versions
|