Coverage for gco/services/auth_middleware.py: 92.74%
196 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-30 21:22 +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-07-18T01:03:40Z
50# Flowchart(s) generated from this file:
51# * ``AuthenticationMiddleware.dispatch`` -> ``diagrams/code_diagrams/gco/services/auth_middleware.AuthenticationMiddleware_dispatch.html``
52# (PNG: ``diagrams/code_diagrams/gco/services/auth_middleware.AuthenticationMiddleware_dispatch.png``)
53# Regenerate with ``python diagrams/code_diagrams/generate.py``.
54# <pyflowchart-code-diagram> END
57logger = logging.getLogger(__name__)
59# Module-level cache for secret signing keys and replay nonces.
60_cached_tokens: set[str] = set()
61_token_expirations: dict[str, float] = {}
62_cache_timestamp = 0.0
63_last_successful_refresh = 0.0
64_last_refresh_attempt = 0.0
65_secrets_client = None
66_nonce_lock = threading.Lock()
67_seen_nonces: dict[str, float] = {}
68_NONCE_PATTERN = re.compile(r"^[0-9a-f]{32}$")
71def _bounded_env_float(name: str, default: float, minimum: float, maximum: float) -> float:
72 try:
73 value = float(os.getenv(name, str(default)))
74 except ValueError:
75 return default
76 return value if minimum <= value <= maximum else default
79CACHE_TTL_SECONDS = _bounded_env_float("AUTH_CACHE_TTL_SECONDS", 300.0, 1.0, 3600.0)
80CACHE_MAX_STALE_SECONDS = max(
81 CACHE_TTL_SECONDS,
82 _bounded_env_float("AUTH_CACHE_MAX_STALE_SECONDS", 900.0, 1.0, 7200.0),
83)
84CACHE_RETRY_SECONDS = _bounded_env_float("AUTH_CACHE_RETRY_SECONDS", 5.0, 0.1, 60.0)
85SIGNATURE_MAX_AGE_SECONDS = _bounded_env_float("AUTH_SIGNATURE_MAX_AGE_SECONDS", 30.0, 5.0, 300.0)
86_MAX_TRACKED_NONCES = 10_000
88# Endpoints that bypass authentication (health checks for load balancers and
89# Global Accelerator). /api/v1/health is included so GA can perform HTTP
90# health checks for intelligent routing without an HMAC envelope.
91UNAUTHENTICATED_PATHS = frozenset(["/healthz", "/readyz", "/metrics", "/api/v1/health"])
94def get_secrets_client() -> Any:
95 """
96 Get Secrets Manager client with lazy initialization.
98 The client is configured to use the region from the AUTH_SECRET_ARN
99 environment variable, which may be different from the default region.
101 Returns:
102 boto3 Secrets Manager client instance
103 """
104 global _secrets_client
105 if _secrets_client is None:
106 # Extract region from the secret ARN
107 # Format: arn:aws:secretsmanager:REGION:ACCOUNT:secret:NAME
108 secret_arn = os.environ.get("AUTH_SECRET_ARN", "")
109 region = None
110 if secret_arn:
111 parts = secret_arn.split(":")
112 if len(parts) >= 4: 112 ↛ 114line 112 didn't jump to line 114 because the condition on line 112 was always true
113 region = parts[3]
114 _secrets_client = boto3.client("secretsmanager", region_name=region)
115 return _secrets_client
118def _is_cache_valid() -> bool:
119 """Return whether keys are still inside the normal refresh TTL."""
120 return bool(_cached_tokens) and (time.monotonic() - _last_successful_refresh) < (
121 CACHE_TTL_SECONDS
122 )
125def _previous_token_valid_until(secrets_client: Any, secret_arn: str) -> float | None:
126 """Return the fixed AWSPREVIOUS deadline from rotation completion metadata.
128 ``LastRotatedDate`` is set when Secrets Manager finishes rotation. Deriving
129 the deadline from it prevents successful cache refreshes from renewing a
130 displaced key indefinitely. Missing or malformed metadata fails closed for
131 AWSPREVIOUS without making AWSCURRENT or AWSPENDING unavailable.
132 """
133 try:
134 metadata = secrets_client.describe_secret(SecretId=secret_arn)
135 last_rotated = metadata.get("LastRotatedDate")
136 timestamp = getattr(last_rotated, "timestamp", None)
137 if not callable(timestamp):
138 raise ValueError("LastRotatedDate is missing")
139 return float(timestamp()) + CACHE_MAX_STALE_SECONDS
140 except Exception:
141 logger.debug("AWSPREVIOUS rotation metadata is unavailable")
142 return None
145def _refresh_cache() -> bool:
146 """Refresh current and overlap keys without extending stale lifetime on failure."""
147 global _cached_tokens, _token_expirations
148 global _cache_timestamp, _last_successful_refresh, _last_refresh_attempt
150 secret_arn = os.environ.get("AUTH_SECRET_ARN")
151 if not secret_arn:
152 return False
154 now = time.monotonic()
155 _last_refresh_attempt = now
156 try:
157 secrets_client = get_secrets_client()
158 response = secrets_client.get_secret_value(
159 SecretId=secret_arn,
160 VersionStage="AWSCURRENT",
161 )
162 secret_data = json.loads(response["SecretString"])
163 current = secret_data.get("token")
164 if not isinstance(current, str) or not current: 164 ↛ 165line 164 didn't jump to line 165 because the condition on line 164 was never true
165 raise ValueError("AWSCURRENT token is missing")
166 new_tokens = {current}
167 new_token_expirations: dict[str, float] = {}
169 # Signers cache AWSCURRENT independently for up to
170 # CACHE_MAX_STALE_SECONDS. During rotation, Secrets Manager moves the
171 # displaced current version to AWSPREVIOUS, so validators must retain
172 # both optional overlap stages while a warm signer can still use them.
173 # AWSPREVIOUS gets a fixed wall-clock deadline from LastRotatedDate;
174 # unlike the aggregate cache TTL, successful refreshes cannot renew it.
175 for version_stage in ("AWSPENDING", "AWSPREVIOUS"):
176 try:
177 response = secrets_client.get_secret_value(
178 SecretId=secret_arn,
179 VersionStage=version_stage,
180 )
181 overlap_data = json.loads(response["SecretString"])
182 overlap = overlap_data.get("token")
183 if not isinstance(overlap, str) or not overlap: 183 ↛ 184line 183 didn't jump to line 184 because the condition on line 183 was never true
184 continue
185 if version_stage == "AWSPREVIOUS" and overlap not in new_tokens:
186 valid_until = _previous_token_valid_until(secrets_client, secret_arn)
187 if valid_until is None or time.time() > valid_until:
188 logger.info("AWSPREVIOUS signing key is outside its overlap window")
189 continue
190 new_token_expirations[overlap] = valid_until
191 new_tokens.add(overlap)
192 except secrets_client.exceptions.ResourceNotFoundException:
193 pass
194 except Exception:
195 logger.debug("%s signing key is unavailable", version_stage)
196 except Exception:
197 logger.exception("Failed to refresh authentication signing keys")
198 return False
200 _cached_tokens = new_tokens
201 _token_expirations = new_token_expirations
202 _last_successful_refresh = now
203 _cache_timestamp = now
204 logger.info("Authentication signing-key cache refreshed")
205 return True
208def get_valid_tokens() -> set[str]:
209 """Return current signing keys with a strictly bounded stale grace period."""
210 now = time.monotonic()
211 if not _is_cache_valid() and now - _last_refresh_attempt >= CACHE_RETRY_SECONDS:
212 _refresh_cache()
213 age = time.monotonic() - _last_successful_refresh
214 if _cached_tokens and age <= CACHE_MAX_STALE_SECONDS:
215 wall_clock = time.time()
216 return {
217 token
218 for token in _cached_tokens
219 if token not in _token_expirations or wall_clock <= _token_expirations[token]
220 }
221 return set()
224def get_secret_token() -> str | None:
225 """Return one primary signing key for compatibility callers.
227 HMAC validation should use :func:`get_valid_tokens` so both current and
228 pending rotation keys are considered.
229 """
230 tokens = get_valid_tokens()
231 return next(iter(tokens), None) if tokens else None
234def clear_token_cache() -> None:
235 """Clear signing-key and replay caches, forcing a refresh."""
236 global _cached_tokens, _token_expirations
237 global _cache_timestamp, _last_successful_refresh, _last_refresh_attempt
238 _cached_tokens = set()
239 _token_expirations = {}
240 _cache_timestamp = 0.0
241 _last_successful_refresh = 0.0
242 _last_refresh_attempt = 0.0
243 with _nonce_lock:
244 _seen_nonces.clear()
245 logger.info("Authentication signing-key cache cleared")
248def _request_target(request: Request) -> str:
249 raw_path = request.scope.get("raw_path")
250 path = raw_path.decode("latin-1") if isinstance(raw_path, bytes) else request.url.path
251 raw_query = request.scope.get("query_string", b"")
252 query = raw_query.decode("latin-1") if isinstance(raw_query, bytes) else str(raw_query)
253 return path + (f"?{query}" if query else "")
256def _accept_nonce(nonce: str, now: float) -> bool:
257 """Reject process-local replays and keep the nonce cache strictly bounded."""
258 expires_at = now + SIGNATURE_MAX_AGE_SECONDS
259 with _nonce_lock:
260 expired = [key for key, expiry in _seen_nonces.items() if expiry < now]
261 for key in expired: 261 ↛ 262line 261 didn't jump to line 262 because the loop on line 261 never started
262 _seen_nonces.pop(key, None)
263 if nonce in _seen_nonces:
264 return False
265 if len(_seen_nonces) >= _MAX_TRACKED_NONCES: 265 ↛ 266line 265 didn't jump to line 266 because the condition on line 265 was never true
266 oldest = min(_seen_nonces, key=_seen_nonces.__getitem__)
267 _seen_nonces.pop(oldest, None)
268 _seen_nonces[nonce] = expires_at
269 return True
272async def _has_valid_signature(request: Request, signing_keys: set[str]) -> bool:
273 """Validate the short-lived HMAC envelope added by the trusted Lambda."""
274 headers = request.headers
275 if headers.get("x-gco-signature-version") != "v1":
276 return False
277 signature = headers.get("x-gco-signature", "")
278 timestamp_value = headers.get("x-gco-timestamp", "")
279 nonce = headers.get("x-gco-nonce", "")
280 claimed_content_hash = headers.get("x-gco-content-sha256", "")
281 if (
282 len(signature) != 64
283 or len(claimed_content_hash) != 64
284 or _NONCE_PATTERN.fullmatch(nonce) is None
285 ):
286 return False
287 try:
288 timestamp = int(timestamp_value)
289 except ValueError:
290 return False
291 now = time.time()
292 if abs(now - timestamp) > SIGNATURE_MAX_AGE_SECONDS: 292 ↛ 293line 292 didn't jump to line 293 because the condition on line 292 was never true
293 return False
295 body = await request.body()
296 actual_content_hash = hashlib.sha256(body).hexdigest()
297 if not hmac.compare_digest(actual_content_hash, claimed_content_hash): 297 ↛ 298line 297 didn't jump to line 298 because the condition on line 297 was never true
298 return False
299 canonical = "\n".join(
300 [
301 "v1",
302 timestamp_value,
303 nonce,
304 request.method.upper(),
305 _request_target(request),
306 actual_content_hash,
307 ]
308 )
309 valid = any(
310 hmac.compare_digest(
311 signature,
312 hmac.new(
313 key.encode("utf-8"),
314 canonical.encode("utf-8"),
315 hashlib.sha256,
316 ).hexdigest(),
317 )
318 for key in signing_keys
319 )
320 return valid and _accept_nonce(nonce, now)
323class AuthenticationMiddleware(BaseHTTPMiddleware):
324 """Validate the API Gateway proxy's short-lived HMAC request envelope.
326 Health-check endpoints are excluded for load balancer probes. During key
327 rotation, signatures from both AWSCURRENT and AWSPENDING are accepted.
328 """
330 def __init__(self, app: ASGIApp) -> None:
331 super().__init__(app)
332 # Startup-time configuration check — surface misconfigurations early
333 secret_arn = os.environ.get("AUTH_SECRET_ARN")
334 if not secret_arn:
335 dev_mode = os.environ.get("GCO_DEV_MODE", "").lower() == "true"
336 if dev_mode:
337 logger.warning(
338 "GCO_DEV_MODE=true with no AUTH_SECRET_ARN — "
339 "authentication is bypassed. Do NOT use in production."
340 )
341 else:
342 logger.error(
343 "AUTH_SECRET_ARN is not configured and GCO_DEV_MODE is not enabled. "
344 "All non-health-check requests will be denied with 503."
345 )
347 async def dispatch(
348 self,
349 request: Request,
350 call_next: Callable[[Request], Awaitable[Response]],
351 ) -> Response:
352 """
353 Process incoming request and validate authentication.
355 Args:
356 request: The incoming FastAPI request
357 call_next: The next middleware/handler in the chain
359 Returns:
360 Response from the next handler, or a bounded JSON authentication error.
361 """
362 # Skip authentication for health check endpoints
363 if request.url.path in UNAUTHENTICATED_PATHS:
364 return await call_next(request)
366 valid_tokens = get_valid_tokens()
368 # No tokens available — determine whether to fail open or closed
369 if not valid_tokens:
370 secret_arn = os.environ.get("AUTH_SECRET_ARN")
371 if not secret_arn:
372 # No secret configured. Only allow requests if the operator
373 # explicitly opted into dev mode. This prevents accidental
374 # unauthenticated deployments due to misconfiguration.
375 dev_mode = os.environ.get("GCO_DEV_MODE", "").lower() == "true"
376 if dev_mode:
377 logger.warning(
378 "Authentication bypassed - GCO_DEV_MODE=true, no secret configured"
379 )
380 return await call_next(request)
381 # Fail closed: no secret + no dev mode = deny
382 logger.error(
383 "No AUTH_SECRET_ARN configured and GCO_DEV_MODE is not enabled. "
384 "Set AUTH_SECRET_ARN for production or GCO_DEV_MODE=true for local development."
385 )
386 return JSONResponse(
387 status_code=503,
388 content={"detail": "Service unavailable - authentication not configured"},
389 )
390 # Secret configured but couldn't load - deny access
391 logger.error("Failed to load authentication tokens")
392 return JSONResponse(
393 status_code=503,
394 content={"detail": "Service temporarily unavailable - authentication error"},
395 )
397 if not await _has_valid_signature(request, valid_tokens):
398 client_ip = request.client.host if request.client else "unknown"
399 logger.warning(
400 "Invalid backend signature from %s for %s",
401 client_ip,
402 request.url.path,
403 )
404 return JSONResponse(
405 status_code=403,
406 content={
407 "detail": ("Forbidden - requests must come through authenticated API Gateway")
408 },
409 )
411 return await call_next(request)