Coverage for gco / services / auth_middleware.py: 100.00%
197 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-10-28 16:19 +0000
1"""
2Authentication middleware for validating requests from API Gateway.
4Except for explicit health and metrics probes, every request must carry the
5short-lived HMAC envelope generated by a trusted API Gateway proxy Lambda. The
6envelope binds the signature version, timestamp, random nonce, HTTP method,
7exact path and query string, and SHA-256 body digest. The middleware validates
8freshness and integrity and rejects process-local nonce replays.
10Security Flow:
11 1. API Gateway validates client IAM credentials (SigV4)
12 2. Lambda signs the exact backend request with the shared signing key
13 3. This middleware validates the HMAC envelope and consumes its nonce
14 4. Invalid, stale, tampered, or replayed envelopes result in 403 Forbidden
16Secret Rotation Support:
17 During rotation, signatures are validated against both AWSCURRENT and
18 AWSPENDING signing keys for zero-downtime rotation. Successful refreshes
19 are cached with a bounded stale grace period; expired caches fail closed.
21Environment Variables:
22 AUTH_SECRET_ARN: ARN of the Secrets Manager secret containing the signing key
23 GCO_DEV_MODE: Set to "true" to allow unauthenticated requests when no
24 secret is configured. Without this flag, missing AUTH_SECRET_ARN
25 causes 503 errors (fail-closed). This prevents accidental
26 unauthenticated deployments due to misconfiguration.
27"""
29from __future__ import annotations
31import hashlib
32import hmac
33import json
34import logging
35import os
36import re
37import threading
38import time
39from collections.abc import Awaitable, Callable
40from typing import Any
42import boto3
43from fastapi import Request
44from starlette.middleware.base import BaseHTTPMiddleware
45from starlette.responses import JSONResponse, Response
46from starlette.types import ASGIApp
48# <pyflowchart-code-diagram> BEGIN - auto-inserted, do not edit
49# Generated at (UTC): 2026-09-28T07:34:51Z
50# Generated from Git commit: 95213a3dfe214f41ea8e3977b79711b1be061ac0
51# Flowchart(s) generated from this file:
52# * ``AuthenticationMiddleware.dispatch`` -> ``diagrams/code_diagrams/gco/services/auth_middleware.AuthenticationMiddleware_dispatch.html``
53# (PNG: ``diagrams/code_diagrams/gco/services/auth_middleware.AuthenticationMiddleware_dispatch.png``)
54# Regenerate with ``SOURCE_DATE_EPOCH=<unix-seconds> GCO_DIAGRAM_SOURCE_COMMIT=<40-char-sha> python diagrams/generate.py --code-only``.
55# <pyflowchart-code-diagram> END
58logger = logging.getLogger(__name__)
60# Module-level cache for secret signing keys and replay nonces.
61_cached_tokens: set[str] = set()
62_token_expirations: dict[str, float] = {}
63_cache_timestamp = 0.0
64_last_successful_refresh = 0.0
65_last_refresh_attempt = 0.0
66_secrets_client = None
67_nonce_lock = threading.Lock()
68_seen_nonces: dict[str, float] = {}
69_NONCE_PATTERN = re.compile(r"^[0-9a-f]{32}$")
70#: A lowercase SHA-256 hex digest, as both signers emit. Checked before any
71#: ``hmac.compare_digest``, which raises ``TypeError`` for a ``str`` holding a
72#: non-ASCII character (Starlette decodes header bytes as latin-1), so a
73#: malformed header is a 403 rather than a 500.
74_HEX_DIGEST_PATTERN = re.compile(r"^[0-9a-f]{64}$")
77def _bounded_env_float(name: str, default: float, minimum: float, maximum: float) -> float:
78 try:
79 value = float(os.getenv(name, str(default)))
80 except ValueError:
81 return default
82 return value if minimum <= value <= maximum else default
85CACHE_TTL_SECONDS = _bounded_env_float("AUTH_CACHE_TTL_SECONDS", 300.0, 1.0, 3600.0)
86CACHE_MAX_STALE_SECONDS = max(
87 CACHE_TTL_SECONDS,
88 _bounded_env_float("AUTH_CACHE_MAX_STALE_SECONDS", 900.0, 1.0, 7200.0),
89)
90CACHE_RETRY_SECONDS = _bounded_env_float("AUTH_CACHE_RETRY_SECONDS", 5.0, 0.1, 60.0)
91SIGNATURE_MAX_AGE_SECONDS = _bounded_env_float("AUTH_SIGNATURE_MAX_AGE_SECONDS", 30.0, 5.0, 300.0)
92_MAX_TRACKED_NONCES = 10_000
94# Endpoints that bypass authentication (health checks for load balancers and
95# Global Accelerator). /api/v1/health is included so GA can perform HTTP
96# health checks for intelligent routing without an HMAC envelope.
97UNAUTHENTICATED_PATHS = frozenset(["/healthz", "/readyz", "/metrics", "/api/v1/health"])
100def get_secrets_client() -> Any:
101 """
102 Get Secrets Manager client with lazy initialization.
104 The client is configured to use the region from the AUTH_SECRET_ARN
105 environment variable, which may be different from the default region.
107 Returns:
108 boto3 Secrets Manager client instance
109 """
110 global _secrets_client
111 if _secrets_client is None:
112 # Extract region from the secret ARN
113 # Format: arn:aws:secretsmanager:REGION:ACCOUNT:secret:NAME
114 secret_arn = os.environ.get("AUTH_SECRET_ARN", "")
115 region = None
116 if secret_arn:
117 parts = secret_arn.split(":")
118 if len(parts) >= 4:
119 region = parts[3]
120 _secrets_client = boto3.client("secretsmanager", region_name=region)
121 return _secrets_client
124def _is_cache_valid() -> bool:
125 """Return whether keys are still inside the normal refresh TTL."""
126 return bool(_cached_tokens) and (time.monotonic() - _last_successful_refresh) < (
127 CACHE_TTL_SECONDS
128 )
131def _previous_token_valid_until(secrets_client: Any, secret_arn: str) -> float | None:
132 """Return the fixed AWSPREVIOUS deadline from rotation completion metadata.
134 ``LastRotatedDate`` is set when Secrets Manager finishes rotation. Deriving
135 the deadline from it prevents successful cache refreshes from renewing a
136 displaced key indefinitely. Missing or malformed metadata fails closed for
137 AWSPREVIOUS without making AWSCURRENT or AWSPENDING unavailable.
138 """
139 try:
140 metadata = secrets_client.describe_secret(SecretId=secret_arn)
141 last_rotated = metadata.get("LastRotatedDate")
142 timestamp = getattr(last_rotated, "timestamp", None)
143 if not callable(timestamp):
144 raise ValueError("LastRotatedDate is missing")
145 return float(timestamp()) + CACHE_MAX_STALE_SECONDS
146 except Exception:
147 logger.debug("AWSPREVIOUS rotation metadata is unavailable")
148 return None
151def _refresh_cache() -> bool:
152 """Refresh current and overlap keys without extending stale lifetime on failure."""
153 global _cached_tokens, _token_expirations
154 global _cache_timestamp, _last_successful_refresh, _last_refresh_attempt
156 secret_arn = os.environ.get("AUTH_SECRET_ARN")
157 if not secret_arn:
158 return False
160 now = time.monotonic()
161 _last_refresh_attempt = now
162 try:
163 secrets_client = get_secrets_client()
164 response = secrets_client.get_secret_value(
165 SecretId=secret_arn,
166 VersionStage="AWSCURRENT",
167 )
168 secret_data = json.loads(response["SecretString"])
169 current = secret_data.get("token")
170 if not isinstance(current, str) or not current:
171 raise ValueError("AWSCURRENT token is missing")
172 new_tokens = {current}
173 new_token_expirations: dict[str, float] = {}
175 # Signers cache AWSCURRENT independently for up to
176 # CACHE_MAX_STALE_SECONDS. During rotation, Secrets Manager moves the
177 # displaced current version to AWSPREVIOUS, so validators must retain
178 # both optional overlap stages while a warm signer can still use them.
179 # AWSPREVIOUS gets a fixed wall-clock deadline from LastRotatedDate;
180 # unlike the aggregate cache TTL, successful refreshes cannot renew it.
181 for version_stage in ("AWSPENDING", "AWSPREVIOUS"):
182 try:
183 response = secrets_client.get_secret_value(
184 SecretId=secret_arn,
185 VersionStage=version_stage,
186 )
187 overlap_data = json.loads(response["SecretString"])
188 overlap = overlap_data.get("token")
189 if not isinstance(overlap, str) or not overlap:
190 continue
191 if version_stage == "AWSPREVIOUS" and overlap not in new_tokens:
192 valid_until = _previous_token_valid_until(secrets_client, secret_arn)
193 if valid_until is None or time.time() > valid_until:
194 logger.info("AWSPREVIOUS signing key is outside its overlap window")
195 continue
196 new_token_expirations[overlap] = valid_until
197 new_tokens.add(overlap)
198 except secrets_client.exceptions.ResourceNotFoundException:
199 pass
200 except Exception:
201 logger.debug("%s signing key is unavailable", version_stage)
202 except Exception:
203 logger.exception("Failed to refresh authentication signing keys")
204 return False
206 _cached_tokens = new_tokens
207 _token_expirations = new_token_expirations
208 _last_successful_refresh = now
209 _cache_timestamp = now
210 logger.info("Authentication signing-key cache refreshed")
211 return True
214def get_valid_tokens() -> set[str]:
215 """Return current signing keys with a strictly bounded stale grace period."""
216 now = time.monotonic()
217 if not _is_cache_valid() and now - _last_refresh_attempt >= CACHE_RETRY_SECONDS:
218 _refresh_cache()
219 age = time.monotonic() - _last_successful_refresh
220 if _cached_tokens and age <= CACHE_MAX_STALE_SECONDS:
221 wall_clock = time.time()
222 return {
223 token
224 for token in _cached_tokens
225 if token not in _token_expirations or wall_clock <= _token_expirations[token]
226 }
227 return set()
230def get_secret_token() -> str | None:
231 """Return one primary signing key for compatibility callers.
233 HMAC validation should use :func:`get_valid_tokens` so both current and
234 pending rotation keys are considered.
235 """
236 tokens = get_valid_tokens()
237 return next(iter(tokens), None) if tokens else None
240def clear_token_cache() -> None:
241 """Clear signing-key and replay caches, forcing a refresh."""
242 global _cached_tokens, _token_expirations
243 global _cache_timestamp, _last_successful_refresh, _last_refresh_attempt
244 _cached_tokens = set()
245 _token_expirations = {}
246 _cache_timestamp = 0.0
247 _last_successful_refresh = 0.0
248 _last_refresh_attempt = 0.0
249 with _nonce_lock:
250 _seen_nonces.clear()
251 logger.info("Authentication signing-key cache cleared")
254def _request_target(request: Request) -> str:
255 raw_path = request.scope.get("raw_path")
256 path = raw_path.decode("latin-1") if isinstance(raw_path, bytes) else request.url.path
257 raw_query = request.scope.get("query_string", b"")
258 query = raw_query.decode("latin-1") if isinstance(raw_query, bytes) else str(raw_query)
259 return path + (f"?{query}" if query else "")
262def _accept_nonce(nonce: str, now: float) -> bool:
263 """Reject process-local replays and keep the nonce cache strictly bounded."""
264 expires_at = now + SIGNATURE_MAX_AGE_SECONDS
265 with _nonce_lock:
266 expired = [key for key, expiry in _seen_nonces.items() if expiry < now]
267 for key in expired:
268 _seen_nonces.pop(key, None)
269 if nonce in _seen_nonces:
270 return False
271 if len(_seen_nonces) >= _MAX_TRACKED_NONCES:
272 oldest = min(_seen_nonces, key=_seen_nonces.__getitem__)
273 _seen_nonces.pop(oldest, None)
274 _seen_nonces[nonce] = expires_at
275 return True
278async def _has_valid_signature(request: Request, signing_keys: set[str]) -> bool:
279 """Validate the short-lived HMAC envelope added by the trusted Lambda."""
280 headers = request.headers
281 if headers.get("x-gco-signature-version") != "v1":
282 return False
283 signature = headers.get("x-gco-signature", "")
284 timestamp_value = headers.get("x-gco-timestamp", "")
285 nonce = headers.get("x-gco-nonce", "")
286 claimed_content_hash = headers.get("x-gco-content-sha256", "")
287 if (
288 _HEX_DIGEST_PATTERN.fullmatch(signature) is None
289 or _HEX_DIGEST_PATTERN.fullmatch(claimed_content_hash) is None
290 or _NONCE_PATTERN.fullmatch(nonce) is None
291 ):
292 return False
293 try:
294 timestamp = int(timestamp_value)
295 except ValueError:
296 return False
297 now = time.time()
298 if abs(now - timestamp) > SIGNATURE_MAX_AGE_SECONDS:
299 return False
301 body = await request.body()
302 actual_content_hash = hashlib.sha256(body).hexdigest()
303 if not hmac.compare_digest(actual_content_hash, claimed_content_hash):
304 return False
305 canonical = "\n".join(
306 [
307 "v1",
308 timestamp_value,
309 nonce,
310 request.method.upper(),
311 _request_target(request),
312 actual_content_hash,
313 ]
314 )
315 valid = any(
316 hmac.compare_digest(
317 signature,
318 hmac.new(
319 key.encode("utf-8"),
320 canonical.encode("utf-8"),
321 hashlib.sha256,
322 ).hexdigest(),
323 )
324 for key in signing_keys
325 )
326 return valid and _accept_nonce(nonce, now)
329class AuthenticationMiddleware(BaseHTTPMiddleware):
330 """Validate the API Gateway proxy's short-lived HMAC request envelope.
332 Health-check endpoints are excluded for load balancer probes. During key
333 rotation, signatures from both AWSCURRENT and AWSPENDING are accepted.
334 """
336 def __init__(self, app: ASGIApp) -> None:
337 super().__init__(app)
338 # Startup-time configuration check — surface misconfigurations early
339 secret_arn = os.environ.get("AUTH_SECRET_ARN")
340 if not secret_arn:
341 dev_mode = os.environ.get("GCO_DEV_MODE", "").lower() == "true"
342 if dev_mode:
343 logger.warning(
344 "GCO_DEV_MODE=true with no AUTH_SECRET_ARN — "
345 "authentication is bypassed. Do NOT use in production."
346 )
347 else:
348 logger.error(
349 "AUTH_SECRET_ARN is not configured and GCO_DEV_MODE is not enabled. "
350 "All non-health-check requests will be denied with 503."
351 )
353 async def dispatch(
354 self,
355 request: Request,
356 call_next: Callable[[Request], Awaitable[Response]],
357 ) -> Response:
358 """
359 Process incoming request and validate authentication.
361 Args:
362 request: The incoming FastAPI request
363 call_next: The next middleware/handler in the chain
365 Returns:
366 Response from the next handler, or a bounded JSON authentication error.
367 """
368 # Skip authentication for health check endpoints
369 if request.url.path in UNAUTHENTICATED_PATHS:
370 return await call_next(request)
372 valid_tokens = get_valid_tokens()
374 # No tokens available — determine whether to fail open or closed
375 if not valid_tokens:
376 secret_arn = os.environ.get("AUTH_SECRET_ARN")
377 if not secret_arn:
378 # No secret configured. Only allow requests if the operator
379 # explicitly opted into dev mode. This prevents accidental
380 # unauthenticated deployments due to misconfiguration.
381 dev_mode = os.environ.get("GCO_DEV_MODE", "").lower() == "true"
382 if dev_mode:
383 logger.warning(
384 "Authentication bypassed - GCO_DEV_MODE=true, no secret configured"
385 )
386 return await call_next(request)
387 # Fail closed: no secret + no dev mode = deny
388 logger.error(
389 "No AUTH_SECRET_ARN configured and GCO_DEV_MODE is not enabled. "
390 "Set AUTH_SECRET_ARN for production or GCO_DEV_MODE=true for local development."
391 )
392 return JSONResponse(
393 status_code=503,
394 content={"detail": "Service unavailable - authentication not configured"},
395 )
396 # Secret configured but couldn't load - deny access
397 logger.error("Failed to load authentication tokens")
398 return JSONResponse(
399 status_code=503,
400 content={"detail": "Service temporarily unavailable - authentication error"},
401 )
403 if not await _has_valid_signature(request, valid_tokens):
404 client_ip = request.client.host if request.client else "unknown"
405 logger.warning(
406 "Invalid backend signature from %s for %s",
407 client_ip,
408 request.url.path,
409 )
410 return JSONResponse(
411 status_code=403,
412 content={
413 "detail": ("Forbidden - requests must come through authenticated API Gateway")
414 },
415 )
417 return await call_next(request)