Coverage for lambda / proxy-shared / proxy_utils.py: 100.00%
180 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-09-14 22:07 +0000
1"""Shared, fail-closed utilities for API Gateway backend proxy Lambdas."""
3import hashlib
4import hmac
5import json
6import logging
7import os
8import re
9import secrets
10import threading
11import time
12from typing import Any
13from urllib.parse import quote, urlencode, urlsplit, urlunsplit
15import boto3
16import urllib3
17from backend_tls import get_backend_http_pool
19# <pyflowchart-code-diagram> BEGIN - auto-inserted, do not edit
20# Generated at (UTC): 2026-09-01T14:42:56Z
21# Generated from Git commit: 89b000378ed5a912a38c06f4feab2b029936ebcc
22# Flowchart(s) generated from this file:
23# * ``build_signed_headers`` -> ``diagrams/code_diagrams/lambda/proxy-shared/proxy_utils.build_signed_headers.html``
24# (PNG: ``diagrams/code_diagrams/lambda/proxy-shared/proxy_utils.build_signed_headers.png``)
25# Regenerate with ``SOURCE_DATE_EPOCH=<unix-seconds> GCO_DIAGRAM_SOURCE_COMMIT=<40-char-sha> python diagrams/generate.py --code-only``.
26# <pyflowchart-code-diagram> END
29logger = logging.getLogger(__name__)
31_HOP_BY_HOP_HEADERS = frozenset(
32 {
33 "connection",
34 "keep-alive",
35 "proxy-authenticate",
36 "proxy-authorization",
37 "te",
38 "trailer",
39 "transfer-encoding",
40 "upgrade",
41 }
42)
43_ALLOWED_REQUEST_HEADERS = frozenset(
44 {
45 "accept",
46 "accept-encoding",
47 "cache-control",
48 "content-encoding",
49 "content-type",
50 "idempotency-key",
51 "if-match",
52 "if-none-match",
53 "prefer",
54 "range",
55 "user-agent",
56 "x-request-id",
57 }
58)
59_INTERNAL_SIGNATURE_HEADERS = frozenset(
60 {
61 "x-gco-signature-version",
62 "x-gco-signature",
63 "x-gco-timestamp",
64 "x-gco-nonce",
65 "x-gco-content-sha256",
66 }
67)
68_RETRYABLE_STATUS_CODES = frozenset({429, 502, 503, 504})
69_RETRYABLE_METHODS = frozenset({"GET", "HEAD", "OPTIONS"})
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
80def _bounded_env_int(name: str, default: int, minimum: int, maximum: int) -> int:
81 try:
82 value = int(os.getenv(name, str(default)))
83 except ValueError:
84 return default
85 return value if minimum <= value <= maximum else default
88_secret_lock = threading.Lock()
89_cached_secret: str | None = None
90_cache_timestamp = 0.0
91_last_successful_refresh = 0.0
92_last_refresh_attempt = 0.0
93_CACHE_TTL_SECONDS = _bounded_env_float("SECRET_CACHE_TTL_SECONDS", 300.0, 1.0, 3600.0)
94_CACHE_MAX_STALE_SECONDS = max(
95 _CACHE_TTL_SECONDS,
96 _bounded_env_float("SECRET_CACHE_MAX_STALE_SECONDS", 900.0, 1.0, 7200.0),
97)
98_CACHE_RETRY_SECONDS = _bounded_env_float("SECRET_CACHE_RETRY_SECONDS", 5.0, 0.1, 60.0)
101def _secret_region(secret_arn: str) -> str | None:
102 """Return the owning region for a Secrets Manager ARN, if present."""
103 parts = secret_arn.split(":", 5)
104 if len(parts) == 6 and parts[0] == "arn" and parts[2] == "secretsmanager":
105 return parts[3] or None
106 return None
109# The regional VPC proxy runs outside the API Gateway region where the shared
110# HMAC key lives. Secrets Manager clients do not route cross-region ARNs to the
111# owning endpoint automatically, so bind the client to the ARN's region.
112_secrets_client = boto3.client(
113 "secretsmanager",
114 region_name=_secret_region(os.getenv("SECRET_ARN", "")),
115)
118def get_secret_token() -> str:
119 """Return a cached signing key, with a strictly bounded stale grace period."""
120 global _cached_secret, _cache_timestamp, _last_successful_refresh, _last_refresh_attempt
122 now = time.monotonic()
123 age = now - _last_successful_refresh
124 if _cached_secret is not None and age < _CACHE_TTL_SECONDS:
125 return _cached_secret
126 if (
127 _cached_secret is not None
128 and age <= _CACHE_MAX_STALE_SECONDS
129 and now - _last_refresh_attempt < _CACHE_RETRY_SECONDS
130 ):
131 return _cached_secret
133 with _secret_lock:
134 now = time.monotonic()
135 age = now - _last_successful_refresh
136 if _cached_secret is not None and age < _CACHE_TTL_SECONDS:
137 return _cached_secret
138 if (
139 _cached_secret is not None
140 and age <= _CACHE_MAX_STALE_SECONDS
141 and now - _last_refresh_attempt < _CACHE_RETRY_SECONDS
142 ):
143 return _cached_secret
145 _last_refresh_attempt = now
146 try:
147 response = _secrets_client.get_secret_value(SecretId=os.environ["SECRET_ARN"])
148 secret_data = json.loads(response["SecretString"])
149 token = secret_data.get("token")
150 if not isinstance(token, str) or not token:
151 raise ValueError("secret token is missing")
152 except Exception as error:
153 if _cached_secret is not None and age <= _CACHE_MAX_STALE_SECONDS:
154 logger.warning("Secrets Manager refresh failed; using bounded stale signing key")
155 return _cached_secret
156 raise RuntimeError("Authentication signing key is unavailable") from error
158 _cached_secret = token
159 _last_successful_refresh = now
160 _cache_timestamp = now # Backward-compatible observability/test alias.
161 return token
164def sanitize_request_headers(headers: dict[str, Any]) -> dict[str, str]:
165 """Apply a case-insensitive end-to-end allowlist at the IAM trust boundary."""
166 sanitized: dict[str, str] = {}
167 for name, value in headers.items():
168 normalized = str(name).strip().lower()
169 if normalized not in _ALLOWED_REQUEST_HEADERS or value is None:
170 continue
171 sanitized[normalized] = str(value)
172 return sanitized
175def _request_target(target_url: str) -> str:
176 parsed = urlsplit(target_url)
177 return parsed.path + (f"?{parsed.query}" if parsed.query else "")
180def build_signed_headers(
181 signing_key: str,
182 http_method: str,
183 target_url: str,
184 body: str | None,
185) -> dict[str, str]:
186 """Build short-lived request authentication without transmitting the signing key."""
187 timestamp = str(int(time.time()))
188 nonce = secrets.token_hex(16)
189 content_hash = hashlib.sha256((body or "").encode("utf-8")).hexdigest()
190 canonical = "\n".join(
191 ["v1", timestamp, nonce, http_method.upper(), _request_target(target_url), content_hash]
192 )
193 signature = hmac.new(
194 signing_key.encode("utf-8"),
195 canonical.encode("utf-8"),
196 hashlib.sha256,
197 ).hexdigest()
198 return {
199 "x-gco-signature-version": "v1",
200 "x-gco-signature": signature,
201 "x-gco-timestamp": timestamp,
202 "x-gco-nonce": nonce,
203 "x-gco-content-sha256": content_hash,
204 }
207def _outbound_headers(headers: dict[str, Any]) -> dict[str, str]:
208 """Re-validate caller headers while retaining only generated auth fields."""
209 allowed = _ALLOWED_REQUEST_HEADERS | _INTERNAL_SIGNATURE_HEADERS
210 return {
211 str(name).lower(): str(value)
212 for name, value in headers.items()
213 if str(name).lower() in allowed and value is not None
214 }
217_MAX_RETRIES = _bounded_env_int("PROXY_MAX_RETRIES", 3, 1, 5)
218_RETRY_BACKOFF_BASE = _bounded_env_float("PROXY_RETRY_BACKOFF_BASE", 0.3, 0.0, 5.0)
219_http: urllib3.PoolManager | None = None
222def _tls_failure_response() -> dict[str, Any]:
223 """Return a bounded error without exposing certificate or trust details."""
224 return {
225 "statusCode": 502,
226 "headers": {"Content-Type": "application/json"},
227 "body": json.dumps({"error": "Backend TLS verification failed"}),
228 }
231def forward_request(
232 target_url: str,
233 http_method: str,
234 headers: dict[str, str],
235 body: str | None,
236 timeout: float = 29.0,
237) -> dict[str, Any]:
238 """Forward over authenticated TLS within one deadline.
240 Retries are limited to safe, read-only methods. Plaintext, non-443, and
241 credential-bearing targets are rejected before any network request.
242 """
243 parsed_target = urlsplit(target_url)
244 try:
245 target_port = parsed_target.port
246 except ValueError as exc:
247 raise ValueError("Backend proxy target has an invalid port") from exc
248 if (
249 parsed_target.scheme.lower() != "https"
250 or not parsed_target.hostname
251 or parsed_target.username is not None
252 or parsed_target.password is not None
253 or target_port not in {None, 443}
254 ):
255 raise ValueError("Backend proxy targets must use HTTPS on port 443")
257 # Anchor the deadline before acquiring transport: the caller computed the
258 # budget from the Lambda's remaining time, so a cold-start trust-bundle
259 # refresh must consume this budget. Anchoring after it extended the wall
260 # clock past the Lambda timeout, killing the function mid-flight instead
261 # of returning its bounded 504 when the backend black-holed.
262 deadline = time.monotonic() + max(timeout, 0.0)
264 try:
265 transport = _http or get_backend_http_pool()
266 except RuntimeError:
267 logger.exception("Backend TLS trust is unavailable")
268 return {
269 "statusCode": 503,
270 "headers": {"Content-Type": "application/json"},
271 "body": json.dumps({"error": "Backend trust is temporarily unavailable"}),
272 }
274 method = http_method.upper()
275 encoded_body = body.encode("utf-8") if body else None
276 max_attempts = _MAX_RETRIES if method in _RETRYABLE_METHODS else 1
277 last_exception: Exception | None = None
278 last_response: urllib3.BaseHTTPResponse | None = None
279 attempts_made = 0
281 for attempt in range(max_attempts):
282 remaining = deadline - time.monotonic()
283 if remaining <= 0:
284 break
285 attempts_made = attempt + 1
286 try:
287 response = transport.request(
288 method,
289 target_url,
290 headers=_outbound_headers(headers),
291 body=encoded_body,
292 timeout=urllib3.Timeout(total=remaining),
293 )
294 if response.status not in _RETRYABLE_STATUS_CODES:
295 return _build_success_response(response)
296 last_response = response
297 logger.warning(
298 "Retryable upstream status %d on attempt %d/%d for %s",
299 response.status,
300 attempt + 1,
301 max_attempts,
302 method,
303 )
304 if attempt == max_attempts - 1:
305 return _build_success_response(response)
306 response.release_conn()
307 except urllib3.exceptions.SSLError:
308 logger.exception("Backend TLS verification failed")
309 return _tls_failure_response()
310 except urllib3.exceptions.MaxRetryError as error:
311 if isinstance(getattr(error, "reason", None), urllib3.exceptions.SSLError):
312 logger.exception("Backend TLS verification failed")
313 return _tls_failure_response()
314 last_exception = error
315 logger.warning(
316 "Upstream %s failed on attempt %d/%d",
317 method,
318 attempt + 1,
319 max_attempts,
320 )
321 except urllib3.exceptions.TimeoutError as error:
322 last_exception = error
323 logger.warning(
324 "Upstream %s failed on attempt %d/%d",
325 method,
326 attempt + 1,
327 max_attempts,
328 )
329 except Exception:
330 logger.exception("Unexpected proxy forwarding failure")
331 return {
332 "statusCode": 500,
333 "headers": {"Content-Type": "application/json"},
334 "body": json.dumps({"error": "Internal server error"}),
335 }
337 if attempt < max_attempts - 1:
338 remaining = deadline - time.monotonic()
339 backoff = _RETRY_BACKOFF_BASE * (2**attempt)
340 if remaining <= backoff:
341 break
342 time.sleep(backoff)
344 if last_response is not None:
345 return _build_success_response(last_response)
346 if last_exception is not None:
347 status_code = 503 if isinstance(last_exception, urllib3.exceptions.MaxRetryError) else 504
348 return {
349 "statusCode": status_code,
350 "headers": {"Content-Type": "application/json"},
351 "body": json.dumps(
352 {
353 "error": "Service unavailable" if status_code == 503 else "Gateway timeout",
354 "message": f"Upstream failed after {attempts_made} attempt(s)",
355 }
356 ),
357 }
358 return {
359 "statusCode": 504,
360 "headers": {"Content-Type": "application/json"},
361 "body": json.dumps({"error": "Gateway timeout"}),
362 }
365def _build_success_response(response: urllib3.BaseHTTPResponse) -> dict[str, Any]:
366 """Build an API Gateway response without hop-by-hop framing headers."""
367 response_headers = {
368 str(name): str(value)
369 for name, value in response.headers.items()
370 if str(name).lower() not in _HOP_BY_HOP_HEADERS
371 }
372 return {
373 "statusCode": response.status,
374 "headers": response_headers,
375 "body": response.data.decode("utf-8"),
376 }
379def build_target_url(
380 endpoint: str,
381 path: str,
382 query_params: dict[str, str | list[str]] | None,
383) -> str:
384 """Build one HTTPS/443 upstream URL without losing repeated query keys."""
385 base_url = endpoint if "://" in endpoint else f"https://{endpoint}"
386 parsed_endpoint = urlsplit(base_url)
387 try:
388 endpoint_port = parsed_endpoint.port
389 except ValueError as exc:
390 raise ValueError(f"Invalid proxy endpoint: {endpoint!r}") from exc
391 if (
392 parsed_endpoint.scheme.lower() != "https"
393 or not parsed_endpoint.hostname
394 or parsed_endpoint.username is not None
395 or parsed_endpoint.password is not None
396 or endpoint_port not in {None, 443}
397 or parsed_endpoint.query
398 or parsed_endpoint.fragment
399 ):
400 raise ValueError(f"Proxy endpoint must use HTTPS on port 443: {endpoint!r}")
402 request_path = path if path.startswith("/") else f"/{path}"
403 request_path = re.sub(r"%(?![0-9A-Fa-f]{2})", "%25", request_path)
404 encoded_path = quote(request_path, safe="/:@-._~!$&'()*+,;=%")
405 endpoint_path = parsed_endpoint.path.rstrip("/")
406 return urlunsplit(
407 (
408 parsed_endpoint.scheme,
409 parsed_endpoint.netloc,
410 f"{endpoint_path}{encoded_path}",
411 urlencode(query_params or {}, doseq=True),
412 "",
413 )
414 )