Coverage for lambda / tls-shared / backend_tls.py: 100.00%

1"""Strict private-root TLS transport shared by GCO backend callers. 

2 

3Callers connect to dynamic Global Accelerator or internal-ALB DNS names while 

4presenting and verifying the stable deployment-local identity configured in 

5``BACKEND_TLS_SERVER_NAME``. The trust bundle is public material retrieved from 

6the project-scoped SSM parameter; no root private key is available to callers. 

7""" 

8 

9from __future__ import annotations 

10 

11import logging 

12import os 

13import re 

14import ssl 

15import threading 

16import time 

17 

18import boto3 

19import urllib3 

20 

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

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

23# Generated from Git commit: 89b000378ed5a912a38c06f4feab2b029936ebcc 

24# Flowchart(s) generated from this file: 

25# * ``get_backend_http_pool`` -> ``diagrams/code_diagrams/lambda/tls-shared/backend_tls.get_backend_http_pool.html`` 

26# (PNG: ``diagrams/code_diagrams/lambda/tls-shared/backend_tls.get_backend_http_pool.png``) 

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

28# <pyflowchart-code-diagram> END 

29 

30 

31LOGGER = logging.getLogger(__name__) 

32 

33_DNS_RE = re.compile( 

34 r"(?=.{1,253}\Z)(?:[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?\.)+" 

35 r"[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?", 

36 re.IGNORECASE, 

37) 

38_pool_lock = threading.Lock() 

39_cached_pool: urllib3.PoolManager | None = None 

40_last_successful_refresh = 0.0 

41_last_refresh_attempt = 0.0 

42 

43 

44def _bounded_env_float(name: str, default: float, minimum: float, maximum: float) -> float: 

45 try: 

46 value = float(os.getenv(name, str(default))) 

47 except ValueError: 

48 return default 

49 return value if minimum <= value <= maximum else default 

50 

51 

52def _tls_settings() -> tuple[str, str, str, float, float, float]: 

53 server_name = os.getenv("BACKEND_TLS_SERVER_NAME", "").strip().rstrip(".") 

54 parameter_name = os.getenv("BACKEND_TLS_ROOT_CA_PARAMETER", "").strip() 

55 parameter_region = os.getenv("BACKEND_TLS_ROOT_CA_REGION", "").strip() 

56 if _DNS_RE.fullmatch(server_name) is None: 

57 raise RuntimeError("Backend TLS server identity is not configured") 

58 if not parameter_name.startswith("/") or not parameter_region: 

59 raise RuntimeError("Backend TLS trust parameter is not configured") 

60 ttl = _bounded_env_float("BACKEND_TLS_CA_CACHE_TTL_SECONDS", 300.0, 1.0, 3600.0) 

61 max_stale = max( 

62 ttl, 

63 _bounded_env_float( 

64 "BACKEND_TLS_CA_MAX_STALE_SECONDS", 

65 3600.0, 

66 1.0, 

67 86400.0, 

68 ), 

69 ) 

70 retry = _bounded_env_float("BACKEND_TLS_CA_RETRY_SECONDS", 5.0, 0.1, 60.0) 

71 return server_name, parameter_name, parameter_region, ttl, max_stale, retry 

72 

73 

74def _new_pool(server_name: str, trust_bundle: str) -> urllib3.PoolManager: 

75 if "PRIVATE KEY" in trust_bundle or "-----BEGIN CERTIFICATE-----" not in trust_bundle: 

76 raise RuntimeError("Backend TLS trust parameter contains invalid public material") 

77 context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) 

78 context.minimum_version = ssl.TLSVersion.TLSv1_2 

79 context.verify_mode = ssl.CERT_REQUIRED 

80 context.check_hostname = True 

81 try: 

82 context.load_verify_locations(cadata=trust_bundle) 

83 except ssl.SSLError as exc: 

84 raise RuntimeError("Backend TLS trust parameter contains malformed certificates") from exc 

85 return urllib3.PoolManager( 

86 num_pools=4, 

87 maxsize=10, 

88 retries=False, 

89 ssl_context=context, 

90 server_hostname=server_name, 

91 assert_hostname=server_name, 

92 ) 

93 

94 

95def get_backend_http_pool() -> urllib3.PoolManager: 

96 """Return a verified HTTPS pool, refreshing its public trust bundle safely.""" 

97 global _cached_pool, _last_successful_refresh, _last_refresh_attempt 

98 

99 server_name, parameter_name, parameter_region, ttl, max_stale, retry = _tls_settings() 

100 now = time.monotonic() 

101 age = now - _last_successful_refresh 

102 if _cached_pool is not None and age < ttl: 

103 return _cached_pool 

104 if _cached_pool is not None and age <= max_stale and now - _last_refresh_attempt < retry: 

105 return _cached_pool 

106 

107 with _pool_lock: 

108 now = time.monotonic() 

109 age = now - _last_successful_refresh 

110 if _cached_pool is not None and age < ttl: 

111 return _cached_pool 

112 if _cached_pool is not None and age <= max_stale and now - _last_refresh_attempt < retry: 

113 return _cached_pool 

114 

115 _last_refresh_attempt = now 

116 try: 

117 response = boto3.client("ssm", region_name=parameter_region).get_parameter( 

118 Name=parameter_name 

119 ) 

120 trust_bundle = str(response.get("Parameter", {}).get("Value", "")) 

121 refreshed_pool = _new_pool(server_name, trust_bundle) 

122 except Exception as exc: 

123 if _cached_pool is not None and age <= max_stale: 

124 LOGGER.warning("Backend TLS trust refresh failed; using bounded stale trust bundle") 

125 return _cached_pool 

126 raise RuntimeError("Backend TLS trust bundle is unavailable") from exc 

127 

128 _cached_pool = refreshed_pool 

129 _last_successful_refresh = now 

130 return refreshed_pool 

131 

132 

133def reset_backend_tls_cache() -> None: 

134 """Clear process-local trust state for deterministic tests and cold-start simulation.""" 

135 global _cached_pool, _last_successful_refresh, _last_refresh_attempt 

136 with _pool_lock: 

137 _cached_pool = None 

138 _last_successful_refresh = 0.0 

139 _last_refresh_attempt = 0.0