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 for safe read-only methods (default: 3). 

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

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

23""" 

24 

25import json 

26import os 

27from typing import Any 

28 

29from proxy_utils import ( 

30 build_signed_headers, 

31 build_target_url, 

32 forward_request, 

33 get_secret_token, 

34 sanitize_request_headers, 

35) 

36 

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

38# Generated at (UTC): 2026-09-01T14:42:56Z 

39# Generated from Git commit: 89b000378ed5a912a38c06f4feab2b029936ebcc 

40# Flowchart(s) generated from this file: 

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

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

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

44# <pyflowchart-code-diagram> END 

45 

46 

47_MAX_FORWARD_REQUEST_SECONDS = 28.0 

48_LAMBDA_RESPONSE_HEADROOM_SECONDS = 1.0 

49 

50 

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

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

53 for key in list(headers): 

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

55 value = headers.pop(key) 

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

57 return None 

58 

59 

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

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

62 return { 

63 "statusCode": status_code, 

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

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

66 } 

67 

68 

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

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

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

72 if not callable(get_remaining_time): 

73 return _MAX_FORWARD_REQUEST_SECONDS 

74 

75 remaining_seconds = float(get_remaining_time()) / 1000.0 

76 available_seconds = remaining_seconds - _LAMBDA_RESPONSE_HEADROOM_SECONDS 

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

78 

79 

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

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

82 try: 

83 signing_key = get_secret_token() 

84 except KeyError, RuntimeError: 

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

86 

87 http_method = event["httpMethod"] 

88 path = event["path"] 

89 query_string = ( 

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

91 ) 

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

93 

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

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

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

97 return _error_response( 

98 400, 

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

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

101 ) 

102 

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

104 if event.get("isBase64Encoded"): 

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

106 

107 try: 

108 target_url = build_target_url( 

109 os.environ["GLOBAL_ACCELERATOR_ENDPOINT"], 

110 path, 

111 query_string, 

112 ) 

113 except KeyError, ValueError: 

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

115 

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

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

118 headers = sanitize_request_headers(headers) 

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

120 

121 return forward_request( 

122 target_url, 

123 http_method, 

124 headers, 

125 body, 

126 timeout=_get_request_timeout_seconds(context), 

127 )