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