Coverage for src/lumen/core/security_middleware.py: 90%
72 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-08-31 18:46 +0000
1"""
2Phase 16 hardening: in-memory rate limiting, request-size limiting, and
3baseline browser security headers (HSTS + CSP for the Swagger/ReDoc docs
4routes). No new infrastructure dependency (no Redis) — per-process only,
5consistent with Lumen's single-process design. A limit is per server
6instance, not global across a horizontally-scaled deployment; that's a known
7trade-off for staying dependency-free, not an oversight.
8"""
10from __future__ import annotations
12import time
13from collections import deque
15from starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpoint
16from starlette.requests import Request
17from starlette.responses import JSONResponse, Response
19from ..config import RateLimitSettings, SecurityHeadersSettings
21_WINDOW_SECONDS = 60.0
23# Bound worst-case memory: once this many distinct client buckets exist,
24# sweep out any that have gone quiet (no request within the window). Without
25# this, a flood of distinct/spoofed source IPs would grow the dict forever.
26_SWEEP_THRESHOLD = 10_000
28# Health checks (container orchestrator liveness/readiness probes) must never
29# be throttled — that would turn a rate limit into a self-inflicted outage.
30_RATE_LIMIT_EXEMPT_PATHS = frozenset({"/health"})
33def _client_key(request: Request) -> str:
34 return request.client.host if request.client else "unknown"
37class RateLimitMiddleware(BaseHTTPMiddleware):
38 """Sliding-window rate limit, keyed by client IP."""
40 def __init__(self, app: object, settings: RateLimitSettings) -> None:
41 super().__init__(app) # type: ignore[arg-type]
42 self._settings = settings
43 self._buckets: dict[str, deque[float]] = {}
45 def _sweep_if_needed(self, now: float) -> None:
46 if len(self._buckets) < _SWEEP_THRESHOLD:
47 return
48 cutoff = now - _WINDOW_SECONDS
49 stale = [key for key, bucket in self._buckets.items() if not bucket or bucket[-1] < cutoff]
50 for key in stale:
51 del self._buckets[key]
53 async def dispatch(self, request: Request, call_next: RequestResponseEndpoint) -> Response:
54 if not self._settings.enabled or request.url.path in _RATE_LIMIT_EXEMPT_PATHS:
55 return await call_next(request)
57 now = time.monotonic()
58 key = _client_key(request)
59 bucket = self._buckets.setdefault(key, deque())
60 cutoff = now - _WINDOW_SECONDS
61 while bucket and bucket[0] < cutoff:
62 bucket.popleft()
64 if len(bucket) >= self._settings.requests_per_minute:
65 retry_after = max(1, int(_WINDOW_SECONDS - (now - bucket[0])))
66 return JSONResponse(
67 status_code=429,
68 content={
69 "resourceType": "OperationOutcome",
70 "issue": [
71 {
72 "severity": "error",
73 "code": "throttled",
74 "diagnostics": "Rate limit exceeded. Try again later.",
75 }
76 ],
77 },
78 headers={"Retry-After": str(retry_after)},
79 )
81 bucket.append(now)
82 self._sweep_if_needed(now)
83 return await call_next(request)
86class MaxBodySizeMiddleware(BaseHTTPMiddleware):
87 """Reject requests whose declared `Content-Length` exceeds the configured
88 max. A client that omits `Content-Length` (e.g. real chunked transfer
89 encoding) is not caught here — a known limitation of this first cut;
90 catching that case needs active byte-counting via the ASGI receive
91 channel, which is more invasive and left for a follow-up if it proves
92 necessary in practice.
93 """
95 def __init__(self, app: object, settings: RateLimitSettings) -> None:
96 super().__init__(app) # type: ignore[arg-type]
97 self._settings = settings
99 async def dispatch(self, request: Request, call_next: RequestResponseEndpoint) -> Response:
100 if not self._settings.enabled:
101 return await call_next(request)
103 content_length = request.headers.get("content-length")
104 if content_length is not None:
105 try:
106 length = int(content_length)
107 except ValueError:
108 length = 0
109 if length > self._settings.max_body_bytes:
110 return JSONResponse(
111 status_code=413,
112 content={
113 "resourceType": "OperationOutcome",
114 "issue": [
115 {
116 "severity": "error",
117 "code": "too-costly",
118 "diagnostics": (
119 "Request body exceeds the maximum allowed size of "
120 f"{self._settings.max_body_bytes} bytes."
121 ),
122 }
123 ],
124 },
125 )
126 return await call_next(request)
129# The FHIR JSON API itself never executes a script — the only browser-facing,
130# HTML/JS-serving surface is the Swagger UI / ReDoc docs pages, so that's the
131# only place a CSP is meaningful. FastAPI's default Swagger UI loads its
132# assets from jsdelivr's CDN, so the CSP allows that origin explicitly rather
133# than blocking the docs page it's supposed to protect.
134_DOCS_PATHS = frozenset({"/docs", "/redoc", "/openapi.json"})
135_DOCS_CSP = (
136 "default-src 'self'; "
137 "script-src 'self' https://cdn.jsdelivr.net; "
138 "style-src 'self' https://cdn.jsdelivr.net 'unsafe-inline'; "
139 "img-src 'self' data: https://fastapi.tiangolo.com; "
140 "font-src 'self' https://cdn.jsdelivr.net"
141)
144class SecurityHeadersMiddleware(BaseHTTPMiddleware):
145 """Baseline browser security headers on every response, plus a CSP scoped
146 to the docs routes."""
148 def __init__(self, app: object, settings: SecurityHeadersSettings) -> None:
149 super().__init__(app) # type: ignore[arg-type]
150 self._settings = settings
152 async def dispatch(self, request: Request, call_next: RequestResponseEndpoint) -> Response:
153 response = await call_next(request)
154 if not self._settings.enabled:
155 return response
157 response.headers["X-Content-Type-Options"] = "nosniff"
158 response.headers["X-Frame-Options"] = "DENY"
159 response.headers["Referrer-Policy"] = "no-referrer"
160 response.headers["Strict-Transport-Security"] = (
161 f"max-age={self._settings.hsts_max_age_seconds}; includeSubDomains"
162 )
163 if request.url.path in _DOCS_PATHS:
164 response.headers["Content-Security-Policy"] = _DOCS_CSP
165 return response