Coverage for scripts / live_release_validation / checks / alb_tls.py: 100.00%
151 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-09-14 22:07 +0000
1"""Live ELBv2 evidence for the TLS-only GCO Gateway data path."""
3from __future__ import annotations
5import json
6import time
7from typing import Any
9from gco.stacks.constants import backend_tls_certificate_arn_parameter_name
11from ..models import RunContext, utc_now
13_GATEWAY_TAG = "gco.aws/gateway"
14_GATEWAY_TAG_VALUE = "gco-system/gco-gateway"
15_CLUSTER_TAG = "elbv2.k8s.aws/cluster"
16_TARGET_GROUP_BACKEND_TAG = "gco.aws/backend"
17_EXPECTED_TARGET_GROUP_BACKENDS = frozenset(
18 {"health-monitor", "manifest-processor", "inference-proxy"}
19)
20_TARGET_CONVERGENCE_TIMEOUT_SECONDS = 5 * 60
21_EXPECTED_REGISTERED_TARGET_PORT = 8443
22_ALLOWED_TARGET_GROUP_DEFAULT_PORTS = frozenset({1, _EXPECTED_REGISTERED_TARGET_PORT})
25def _ssm_string_parameter(client: Any, name: str) -> str:
26 response = client.get_parameter(Name=name)
27 parameter = response.get("Parameter") if isinstance(response, dict) else None
28 if not isinstance(parameter, dict):
29 raise RuntimeError(f"SSM parameter response is malformed for {name}")
30 if parameter.get("Name") != name or parameter.get("Type") != "String":
31 raise RuntimeError(f"SSM parameter identity/type is invalid for {name}")
32 value = parameter.get("Value")
33 if not isinstance(value, str) or not value:
34 raise RuntimeError(f"SSM parameter has no String value: {name}")
35 return value
38def _alb_https_target_evidence(
39 ctx: RunContext,
40 *,
41 region: str,
42 cluster_name: str,
43) -> dict[str, Any]:
44 """Require the exact HTTPS listener, certificate, target groups, and health."""
45 client = ctx.session.client("elbv2", region_name=region)
46 load_balancers: list[dict[str, Any]] = []
47 for page in client.get_paginator("describe_load_balancers").paginate():
48 load_balancers.extend(page.get("LoadBalancers", []))
50 tags_by_arn: dict[str, dict[str, str]] = {}
51 for start in range(0, len(load_balancers), 20):
52 arns = [
53 str(item.get("LoadBalancerArn") or "") for item in load_balancers[start : start + 20]
54 ]
55 if not all(arns):
56 raise RuntimeError(f"ELBv2 returned a load balancer without an ARN in {region}")
57 for description in client.describe_tags(ResourceArns=arns).get("TagDescriptions", []):
58 arn = str(description.get("ResourceArn") or "")
59 tags_by_arn[arn] = {
60 str(tag.get("Key") or ""): str(tag.get("Value") or "")
61 for tag in description.get("Tags", [])
62 }
64 matches = []
65 for load_balancer in load_balancers:
66 arn = str(load_balancer.get("LoadBalancerArn") or "")
67 tags = tags_by_arn.get(arn, {})
68 if tags.get(_GATEWAY_TAG) == _GATEWAY_TAG_VALUE and tags.get(_CLUSTER_TAG) == cluster_name:
69 matches.append(load_balancer)
70 if len(matches) != 1:
71 raise RuntimeError(
72 f"Expected exactly one owned GCO Gateway ALB in {region}; found {len(matches)}"
73 )
75 load_balancer = matches[0]
76 load_balancer_arn = str(load_balancer.get("LoadBalancerArn") or "")
77 if load_balancer.get("Scheme") != "internal" or load_balancer.get("Type") != "application":
78 raise RuntimeError(f"GCO Gateway load balancer has an invalid type or scheme in {region}")
79 if (load_balancer.get("State") or {}).get("Code") != "active":
80 raise RuntimeError(f"GCO Gateway load balancer is not active in {region}")
82 certificate_parameter = backend_tls_certificate_arn_parameter_name(
83 ctx.config.project_name,
84 region,
85 )
86 expected_certificate_arn = _ssm_string_parameter(
87 ctx.session.client("ssm", region_name=ctx.config.global_region),
88 certificate_parameter,
89 )
90 listeners: list[dict[str, Any]] = []
91 for page in client.get_paginator("describe_listeners").paginate(
92 LoadBalancerArn=load_balancer_arn
93 ):
94 listeners.extend(page.get("Listeners", []))
95 if len(listeners) != 1:
96 raise RuntimeError(
97 f"GCO Gateway ALB in {region} has {len(listeners)} listeners; expected exactly 1"
98 )
99 listener = listeners[0]
100 listener_arn = str(listener.get("ListenerArn") or "")
101 if not listener_arn:
102 raise RuntimeError(f"GCO Gateway listener in {region} has no ARN")
103 default_certificates = {
104 str(item.get("CertificateArn") or "") for item in listener.get("Certificates", [])
105 }
106 certificate_descriptions: list[dict[str, Any]] = []
107 for page in client.get_paginator("describe_listener_certificates").paginate(
108 ListenerArn=listener_arn
109 ):
110 certificate_descriptions.extend(page.get("Certificates", []))
111 listener_certificates = {
112 str(item.get("CertificateArn") or "") for item in certificate_descriptions
113 }
114 default_listener_certificates = {
115 str(item.get("CertificateArn") or "")
116 for item in certificate_descriptions
117 if item.get("IsDefault") is True
118 }
119 expected_listener = {
120 "Protocol": "HTTPS",
121 "Port": 443,
122 "SslPolicy": "ELBSecurityPolicy-TLS13-1-2-2021-06",
123 }
124 listener_mismatches = {
125 field: {"expected": value, "actual": listener.get(field)}
126 for field, value in expected_listener.items()
127 if listener.get(field) != value
128 }
129 if default_certificates != {expected_certificate_arn}:
130 listener_mismatches["DefaultCertificate"] = {
131 "expected": [expected_certificate_arn],
132 "actual": sorted(default_certificates),
133 }
134 if listener_certificates != {expected_certificate_arn} or default_listener_certificates != {
135 expected_certificate_arn
136 }:
137 listener_mismatches["ListenerCertificates"] = {
138 "expected": [{"arn": expected_certificate_arn, "is_default": True}],
139 "actual": sorted(
140 (
141 {
142 "arn": str(item.get("CertificateArn") or ""),
143 "is_default": item.get("IsDefault") is True,
144 }
145 for item in certificate_descriptions
146 ),
147 key=lambda item: item["arn"],
148 ),
149 }
150 if listener_mismatches:
151 raise RuntimeError(
152 f"GCO Gateway listener in {region} is not the exact HTTPS-only contract: "
153 f"{json.dumps(listener_mismatches, sort_keys=True)}"
154 )
156 target_groups: list[dict[str, Any]] = []
157 for page in client.get_paginator("describe_target_groups").paginate(
158 LoadBalancerArn=load_balancer_arn
159 ):
160 target_groups.extend(page.get("TargetGroups", []))
161 target_group_arns = [
162 str(target_group.get("TargetGroupArn") or "") for target_group in target_groups
163 ]
164 if not all(target_group_arns):
165 raise RuntimeError(f"ELBv2 returned a target group without an ARN in {region}")
167 target_group_tags_by_arn: dict[str, dict[str, str]] = {}
168 for start in range(0, len(target_group_arns), 20):
169 batch = target_group_arns[start : start + 20]
170 for description in client.describe_tags(ResourceArns=batch).get("TagDescriptions", []):
171 arn = str(description.get("ResourceArn") or "")
172 target_group_tags_by_arn[arn] = {
173 str(tag.get("Key") or ""): str(tag.get("Value") or "")
174 for tag in description.get("Tags", [])
175 }
177 backend_to_arn: dict[str, str] = {}
178 for arn in target_group_arns:
179 tags = target_group_tags_by_arn.get(arn, {})
180 backend = tags.get(_TARGET_GROUP_BACKEND_TAG, "")
181 if backend not in _EXPECTED_TARGET_GROUP_BACKENDS:
182 raise RuntimeError(
183 f"GCO Gateway target group {arn} has invalid {_TARGET_GROUP_BACKEND_TAG} "
184 f"identity {backend!r}; expected one of {sorted(_EXPECTED_TARGET_GROUP_BACKENDS)}"
185 )
186 if backend in backend_to_arn:
187 raise RuntimeError(
188 f"GCO Gateway has duplicate target groups for backend {backend!r}: "
189 f"{backend_to_arn[backend]}, {arn}"
190 )
191 backend_to_arn[backend] = arn
192 missing_backends = sorted(_EXPECTED_TARGET_GROUP_BACKENDS - set(backend_to_arn))
193 if missing_backends:
194 raise RuntimeError(f"GCO Gateway is missing target groups for backends: {missing_backends}")
196 evidence: dict[str, Any] = {
197 "region": region,
198 "cluster_name": cluster_name,
199 "load_balancer_arn": load_balancer_arn,
200 "scheme": load_balancer.get("Scheme"),
201 "state": (load_balancer.get("State") or {}).get("Code"),
202 "listener": {
203 "listener_arn": listener.get("ListenerArn"),
204 "protocol": listener.get("Protocol"),
205 "port": listener.get("Port"),
206 "ssl_policy": listener.get("SslPolicy"),
207 "certificates": sorted(listener_certificates),
208 "default_certificates": sorted(default_listener_certificates),
209 },
210 "target_groups": [],
211 }
212 state = ctx.checkpoint.state.setdefault("topology_alb_https_targets", {})
213 state[region] = evidence
214 ctx.persist()
216 poll_seconds = max(1.0, float(ctx.settings.poll_interval_seconds))
217 for target_group in sorted(
218 target_groups,
219 key=lambda item: str(item.get("TargetGroupArn") or ""),
220 ):
221 arn = str(target_group.get("TargetGroupArn") or "")
222 if not arn:
223 raise RuntimeError(f"ELBv2 returned a target group without an ARN in {region}")
224 expected = {
225 "Protocol": "HTTPS",
226 "HealthCheckProtocol": "HTTPS",
227 "TargetType": "ip",
228 "HealthCheckPath": "/healthz",
229 }
230 mismatches = {
231 field: {"expected": value, "actual": target_group.get(field)}
232 for field, value in expected.items()
233 if target_group.get(field) != value
234 }
235 default_port = target_group.get("Port")
236 if default_port not in _ALLOWED_TARGET_GROUP_DEFAULT_PORTS:
237 mismatches["Port"] = {
238 "expected_any_of": sorted(_ALLOWED_TARGET_GROUP_DEFAULT_PORTS),
239 "actual": default_port,
240 }
241 if mismatches:
242 raise RuntimeError(
243 f"GCO Gateway target group {arn} is not HTTPS-hardened: "
244 f"{json.dumps(mismatches, sort_keys=True)}"
245 )
247 group_tags = target_group_tags_by_arn[arn]
248 group_evidence: dict[str, Any] = {
249 "target_group_arn": arn,
250 "backend": group_tags[_TARGET_GROUP_BACKEND_TAG],
251 "tags": group_tags,
252 # The controller uses 1 as the group-wide sentinel when a Service
253 # has a named targetPort. The effective data-plane port is carried
254 # by every TargetHealthDescription.Target registration below.
255 "default_port": default_port,
256 "protocol": target_group.get("Protocol"),
257 "health_check_protocol": target_group.get("HealthCheckProtocol"),
258 "health_check_path": target_group.get("HealthCheckPath"),
259 "target_type": target_group.get("TargetType"),
260 "health_observations": [],
261 }
262 evidence["target_groups"].append(group_evidence)
263 deadline = time.monotonic() + _TARGET_CONVERGENCE_TIMEOUT_SECONDS
264 while True:
265 health_response = client.describe_target_health(TargetGroupArn=arn)
266 descriptions = health_response.get("TargetHealthDescriptions", [])
267 registered_targets: list[dict[str, Any]] = []
268 for item in descriptions:
269 target = item.get("Target") or {}
270 target_health = item.get("TargetHealth") or {}
271 registered_targets.append(
272 {
273 "id": str(target.get("Id") or ""),
274 "port": target.get("Port"),
275 "availability_zone": target.get("AvailabilityZone"),
276 "health_check_port": item.get("HealthCheckPort"),
277 "state": str(target_health.get("State") or ""),
278 "reason": str(target_health.get("Reason") or ""),
279 }
280 )
281 states = [target["state"] for target in registered_targets]
282 observation = {
283 "observed_at": utc_now(),
284 "states": states,
285 "reasons": [target["reason"] for target in registered_targets],
286 "registered_targets": registered_targets,
287 }
288 group_evidence["health_observations"].append(observation)
289 group_evidence["target_states"] = states
290 group_evidence["registered_target_ports"] = sorted(
291 {target["port"] for target in registered_targets if isinstance(target["port"], int)}
292 )
293 ctx.persist()
295 incorrect_effective_ports = [
296 {
297 "id": target["id"],
298 "port": target["port"],
299 "health_check_port": target["health_check_port"],
300 "state": target["state"],
301 }
302 for target in registered_targets
303 if target["port"] != _EXPECTED_REGISTERED_TARGET_PORT
304 or str(target["health_check_port"]) != str(_EXPECTED_REGISTERED_TARGET_PORT)
305 ]
306 group_evidence["incorrect_effective_ports"] = incorrect_effective_ports
307 active_incorrect_ports = [
308 target for target in incorrect_effective_ports if target["state"] != "draining"
309 ]
310 if active_incorrect_ports:
311 raise RuntimeError(
312 f"GCO Gateway target group {arn} has non-draining targets with a traffic "
313 f"or health-check port other than {_EXPECTED_REGISTERED_TARGET_PORT}: "
314 f"{json.dumps(active_incorrect_ports, sort_keys=True)}"
315 )
317 invalid_states = sorted(
318 {
319 state_value
320 for state_value in states
321 if state_value not in {"healthy", "draining"}
322 }
323 )
324 if (
325 descriptions
326 and "healthy" in states
327 and not invalid_states
328 and not incorrect_effective_ports
329 ):
330 break
331 terminal_states = sorted(
332 {
333 state_value
334 for state_value in invalid_states
335 if state_value not in {"initial", "unused", "unavailable"}
336 }
337 )
338 if terminal_states:
339 raise RuntimeError(
340 f"GCO Gateway target group {arn} has nonhealthy targets: {terminal_states}"
341 )
342 remaining = deadline - time.monotonic()
343 if remaining <= 0:
344 raise RuntimeError(
345 f"GCO Gateway target group {arn} did not acquire healthy HTTPS targets "
346 f"within {_TARGET_CONVERGENCE_TIMEOUT_SECONDS} seconds; states={states}; "
347 f"incorrect_effective_ports={incorrect_effective_ports}"
348 )
349 time.sleep(min(poll_seconds, remaining))
351 return evidence