Coverage for lambda / api-gateway-proxy / handler.py: 100.00%

1"""Global API Gateway proxy for the authenticated TLS backend path. 

2 

3API Gateway authenticates callers with IAM, this Lambda signs each exact 

4request with the deployment HMAC key, and a strict private-root TLS transport 

5forwards it through Global Accelerator to a healthy regional ALB. The reusable 

6HMAC key and the root CA private key are never transmitted. 

7 

8``X-GCO-Target-Region`` is intentionally rejected here. This Lambda is not 

9attached to regional VPCs and therefore cannot route directly to private ALBs; 

10authorized callers that need region pinning must invoke that region's always- 

11deployed API bridge directly. Direct caller access is an explicit opt-in even 

12though the bridge itself is required for aggregation. 

13 

14Environment variables: 

15 GLOBAL_ACCELERATOR_ENDPOINT: Global Accelerator DNS name. 

16 SECRET_ARN: Secrets Manager ARN containing the HMAC signing key. 

17 BACKEND_TLS_SERVER_NAME: Private certificate identity asserted via SNI. 

18 BACKEND_TLS_ROOT_CA_PARAMETER: SSM parameter containing public CA roots. 

19 BACKEND_TLS_ROOT_CA_REGION: Region containing the public trust parameter. 

20 PROXY_MAX_RETRIES: Max attempts per request (default: 3). Read-only 

21 methods spend them on any retryable answer; mutating methods only on 

22 the ALB's own 503 page, served before a target is selected. 

23 PROXY_RETRY_BACKOFF_BASE: Base retry backoff in seconds (default: 0.3). 

24 SECRET_CACHE_TTL_SECONDS: Signing-key cache TTL in seconds (default: 300). 

25""" 

26 

27import json 

28import os 

29from typing import Any 

30 

31from proxy_utils import ( 

32 build_signed_headers, 

33 build_target_url, 

34 caller_identity_headers, 

35 forward_request, 

36 get_secret_token, 

37 sanitize_request_headers, 

38) 

39 

40# <pyflowchart-code-diagram> BEGIN - auto-inserted, do not edit 

41# Generated at (UTC): 2026-08-30T12:00:00Z 

42# Generated from Git commit: 8cb8e766461f5b7f35205e12db980988598cf6d5 

43# Flowchart(s) generated from this file: 

44# * ``lambda_handler`` -> ``diagrams/code_diagrams/lambda/api-gateway-proxy/handler.lambda_handler.html`` 

45# (PNG: ``diagrams/code_diagrams/lambda/api-gateway-proxy/handler.lambda_handler.png``) 

46# Regenerate with ``SOURCE_DATE_EPOCH=<unix-seconds> GCO_DIAGRAM_SOURCE_COMMIT=<40-char-sha> python diagrams/generate.py --code-only``. 

47# <pyflowchart-code-diagram> END 

48 

49 

50_MAX_FORWARD_REQUEST_SECONDS = 28.0 

51_LAMBDA_RESPONSE_HEADROOM_SECONDS = 1.0 

52 

53 

54def _pop_header(headers: dict[str, str], header_name: str) -> str | None: 

55 """Remove and return one case-insensitive header value.""" 

56 for key in list(headers): 

57 if key.lower() == header_name.lower(): 

58 value = headers.pop(key) 

59 return str(value).strip() if value is not None else "" 

60 return None 

61 

62 

63def _error_response(status_code: int, message: str) -> dict[str, Any]: 

64 """Build a bounded API Gateway error response.""" 

65 return { 

66 "statusCode": status_code, 

67 "headers": {"Content-Type": "application/json"}, 

68 "body": json.dumps({"error": message}), 

69 } 

70 

71 

72def _get_request_timeout_seconds(context: Any) -> float: 

73 """Bound upstream work while retaining time to return a Lambda response.""" 

74 get_remaining_time = getattr(context, "get_remaining_time_in_millis", None) 

75 if not callable(get_remaining_time): 

76 return _MAX_FORWARD_REQUEST_SECONDS 

77 

78 remaining_seconds = float(get_remaining_time()) / 1000.0 

79 available_seconds = remaining_seconds - _LAMBDA_RESPONSE_HEADROOM_SECONDS 

80 return max(0.0, min(_MAX_FORWARD_REQUEST_SECONDS, available_seconds)) 

81 

82 

83def lambda_handler(event: dict[str, Any], context: Any) -> dict[str, Any]: 

84 """Proxy one IAM-authenticated request through Global Accelerator.""" 

85 try: 

86 signing_key = get_secret_token() 

87 except KeyError, RuntimeError: 

88 return _error_response(503, "Backend authentication is temporarily unavailable") 

89 

90 http_method = event["httpMethod"] 

91 path = event["path"] 

92 query_string = ( 

93 event.get("multiValueQueryStringParameters") or event.get("queryStringParameters") or {} 

94 ) 

95 headers = dict(event.get("headers") or {}) 

96 

97 # A non-VPC Lambda cannot reach any region's internal ALB directly. Never 

98 # pretend region pinning succeeded or silently route the request elsewhere. 

99 if _pop_header(headers, "X-GCO-Target-Region") is not None: 

100 return _error_response( 

101 400, 

102 "X-GCO-Target-Region is not supported by the global endpoint; " 

103 "use the target region's regional API endpoint if authorized for direct access", 

104 ) 

105 

106 body = event.get("body") or "" 

107 if event.get("isBase64Encoded"): 

108 return _error_response(415, "Base64-encoded request bodies are not supported") 

109 

110 try: 

111 target_url = build_target_url( 

112 os.environ["GLOBAL_ACCELERATOR_ENDPOINT"], 

113 path, 

114 query_string, 

115 ) 

116 except KeyError, ValueError: 

117 return _error_response(503, "Global backend routing is temporarily unavailable") 

118 

119 # Only a short-lived HMAC envelope is transmitted. TLS independently 

120 # authenticates the ALB certificate and encrypts the complete request. 

121 headers = sanitize_request_headers(headers) 

122 headers.update(caller_identity_headers(event)) 

123 headers.update(build_signed_headers(signing_key, http_method, target_url, body)) 

124 

125 return forward_request( 

126 target_url, 

127 http_method, 

128 headers, 

129 body, 

130 timeout=_get_request_timeout_seconds(context), 

131 )