Coverage for lambda / ga-registration / handler.py: 100.00%

1"""Gateway ALB registration and endpoint-publication Lambda handler. 

2 

3The handler converges the regional Gateway API ALB after the Gateway manifests 

4have been applied: 

5 

61. Read the exact ``gco-system/gco-gateway`` Gateway status address. 

72. Resolve that address to an active internal ALB, with an exact-tag fallback 

8 while the Gateway status is still being populated. 

93. When a Global Accelerator endpoint group is configured, register only that 

10 ALB, remove stale endpoints, and enforce the HTTPS health-check contract. 

114. Always publish the selected ALB hostname to the SSM endpoint registry. 

12 

13``EndpointGroupArn`` is optional. Deployments without Global Accelerator still 

14publish the Gateway hostname. 

15 

16Entrypoints: 

17 - ``handle_task`` is the final Step Functions convergence task. 

18 - ``lambda_handler`` preserves the legacy raw CloudFormation custom-resource 

19 protocol and dispatches Step Functions events carrying ``Action``. 

20 - ``on_delete_event`` is the CDK provider delete guard. 

21 

22The endpoint registry parameter is always: 

23``/{ProjectName}/alb-hostname-{Region}`` in ``RegistryRegion``. The legacy 

24``GlobalRegion`` property remains accepted for existing deployments. 

25""" 

26 

27import base64 

28import json 

29import logging 

30import os 

31import tempfile 

32import time 

33from contextlib import suppress 

34from typing import Any 

35 

36import boto3 

37import urllib3 

38from botocore.exceptions import ClientError 

39 

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

41# Generated at (UTC): 2026-10-22T03:05:59Z 

42# Generated from Git commit: 503e80ca9adfd5d217fe28b2f27f74fdcdb775a0 

43# Flowchart(s) generated from this file: 

44# * ``lambda_handler`` -> ``diagrams/code_diagrams/lambda/ga-registration/handler.lambda_handler.html`` 

45# (PNG: ``diagrams/code_diagrams/lambda/ga-registration/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 

50logger = logging.getLogger() 

51logger.setLevel(logging.INFO) 

52 

53MAX_WAIT_SECONDS = 840 

54ALB_POLL_INTERVAL = 5 

55GA_DEPLOYED_WAIT_SECONDS = 720 

56GA_DEPLOYED_POLL_INTERVAL = 15 

57DEFAULT_REGISTRY_REGION = "us-east-2" 

58 

59GATEWAY_NAMESPACE = "gco-system" 

60GATEWAY_NAME = "gco-gateway" 

61GATEWAY_REFERENCE = f"{GATEWAY_NAMESPACE}/{GATEWAY_NAME}" 

62GATEWAY_TAG = "gco.aws/gateway" 

63CLUSTER_TAG = "elbv2.k8s.aws/cluster" 

64GATEWAY_API_PATH = ( 

65 f"/apis/gateway.networking.k8s.io/v1/namespaces/{GATEWAY_NAMESPACE}/gateways/{GATEWAY_NAME}" 

66) 

67 

68 

69def send_response( 

70 event: dict[str, Any], 

71 context: Any, 

72 status: str, 

73 data: dict[str, Any], 

74 physical_id: str, 

75 reason: str | None = None, 

76) -> None: 

77 """Send a response to a raw CloudFormation custom resource.""" 

78 response_body = { 

79 "Status": status, 

80 "Reason": reason or f"See CloudWatch Log Stream: {context.log_stream_name}", 

81 "PhysicalResourceId": physical_id, 

82 "StackId": event["StackId"], 

83 "RequestId": event["RequestId"], 

84 "LogicalResourceId": event["LogicalResourceId"], 

85 "Data": data, 

86 } 

87 logger.info("Sending CFN response: Status=%s, PhysicalResourceId=%s", status, physical_id) 

88 http = urllib3.PoolManager() 

89 try: 

90 http.request( 

91 "PUT", 

92 event["ResponseURL"], 

93 body=json.dumps(response_body).encode("utf-8"), 

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

95 timeout=10.0, 

96 ) 

97 except Exception as exc: # callback failure can only be logged 

98 logger.error("Failed to send CloudFormation response: %s", exc) 

99 

100 

101def _remove_temporary_ca_file(ca_path: str | None) -> None: 

102 """Unlink a temporary Kubernetes CA file without masking the real result.""" 

103 if not ca_path: 

104 return 

105 try: 

106 os.unlink(ca_path) 

107 except FileNotFoundError: 

108 return 

109 except OSError as exc: 

110 logger.warning("Failed to remove temporary Kubernetes CA file: %s", exc) 

111 

112 

113def get_k8s_client(cluster_name: str, region: str) -> tuple[str, str, str]: 

114 """Return ``(endpoint, bearer_token, temporary_ca_path)`` for an EKS cluster. 

115 

116 The caller owns the returned CA path and must remove it in a ``finally`` 

117 block. ``mkstemp`` prevents name races; the explicit mode keeps the 

118 certificate readable only by the Lambda process. 

119 """ 

120 eks = boto3.client("eks", region_name=region) 

121 cluster_info = eks.describe_cluster(name=cluster_name)["cluster"] 

122 

123 session = boto3.Session() 

124 sts_client = session.client("sts", region_name=region) 

125 sts_endpoint = str(sts_client.meta.endpoint_url).rstrip("/") 

126 sts_url = f"{sts_endpoint}/?Action=GetCallerIdentity&Version=2011-06-15" 

127 signed_url = sts_client._request_signer.generate_presigned_url( 

128 request_dict={ 

129 "method": "GET", 

130 "url": sts_url, 

131 "body": {}, 

132 "headers": {"x-k8s-aws-id": cluster_name}, 

133 "context": {}, 

134 }, 

135 operation_name="GetCallerIdentity", 

136 expires_in=60, 

137 ) 

138 token = "k8s-aws-v1." + base64.urlsafe_b64encode(signed_url.encode()).decode().rstrip("=") 

139 

140 ca_cert = base64.b64decode(cluster_info["certificateAuthority"]["data"]) 

141 fd, ca_path = tempfile.mkstemp(suffix=".crt") 

142 try: 

143 os.fchmod(fd, 0o600) 

144 with os.fdopen(fd, "wb") as ca_file: 

145 ca_file.write(ca_cert) 

146 except Exception: 

147 with suppress(OSError): 

148 os.close(fd) 

149 _remove_temporary_ca_file(ca_path) 

150 raise 

151 

152 return str(cluster_info["endpoint"]), token, ca_path 

153 

154 

155def _response_json(response: Any) -> dict[str, Any]: 

156 """Decode a Kubernetes JSON response, returning an object mapping.""" 

157 raw_data = response.data 

158 if isinstance(raw_data, bytes): 

159 raw_data = raw_data.decode("utf-8") 

160 document = json.loads(raw_data) 

161 return document if isinstance(document, dict) else {} 

162 

163 

164def find_gateway_address( 

165 http: urllib3.PoolManager, 

166 endpoint: str, 

167 headers: dict[str, str], 

168) -> str | None: 

169 """Read the nonempty hostname from the exact GCO Gateway status.""" 

170 try: 

171 response = http.request( 

172 "GET", 

173 f"{endpoint}{GATEWAY_API_PATH}", 

174 headers=headers, 

175 timeout=10.0, 

176 ) 

177 if response.status == 404: 

178 logger.debug("Gateway %s not found yet", GATEWAY_REFERENCE) 

179 return None 

180 if response.status != 200: 

181 logger.warning("Gateway status request returned HTTP %s", response.status) 

182 return None 

183 

184 addresses = _response_json(response).get("status", {}).get("addresses", []) 

185 for address in addresses: 

186 if not isinstance(address, dict): 

187 continue 

188 address_type = address.get("type", "Hostname") 

189 value = address.get("value") 

190 if address_type == "Hostname" and isinstance(value, str) and value.strip(): 

191 return value.strip() 

192 except Exception as exc: # polling falls back to exact tags 

193 logger.warning("Error checking Gateway status: %s", exc) 

194 return None 

195 

196 

197def _list_load_balancers(elb_client: Any) -> list[dict[str, Any]]: 

198 """List every load balancer, following ELBv2 marker pagination.""" 

199 load_balancers: list[dict[str, Any]] = [] 

200 marker: str | None = None 

201 seen_markers: set[str] = set() 

202 while True: 

203 kwargs = {"Marker": marker} if marker else {} 

204 response = elb_client.describe_load_balancers(**kwargs) 

205 load_balancers.extend(response.get("LoadBalancers", [])) 

206 next_marker = response.get("NextMarker") 

207 if not isinstance(next_marker, str) or not next_marker: 

208 return load_balancers 

209 if next_marker in seen_markers: 

210 raise RuntimeError(f"ELB pagination repeated marker {next_marker!r}") 

211 seen_markers.add(next_marker) 

212 marker = next_marker 

213 

214 

215def find_alb_by_gateway_hostname( 

216 elb_client: Any, hostname: str, cluster_name: str 

217) -> tuple[str | None, str | None, str | None]: 

218 """Resolve a Gateway hostname only to its exactly owned internal ALB.""" 

219 try: 

220 candidates = [ 

221 load_balancer 

222 for load_balancer in _list_load_balancers(elb_client) 

223 if load_balancer.get("Type") == "application" 

224 and load_balancer.get("Scheme") == "internal" 

225 and load_balancer.get("DNSName") == hostname 

226 ] 

227 if not candidates: 

228 return None, None, None 

229 

230 arns = [str(load_balancer["LoadBalancerArn"]) for load_balancer in candidates] 

231 tags_by_arn = _describe_tags(elb_client, arns) 

232 for load_balancer in candidates: 

233 arn = str(load_balancer["LoadBalancerArn"]) 

234 tags = tags_by_arn.get(arn, {}) 

235 if not ( 

236 tags.get(GATEWAY_TAG) == GATEWAY_REFERENCE and tags.get(CLUSTER_TAG) == cluster_name 

237 ): 

238 logger.warning("Rejecting hostname-matched ALB without exact ownership: %s", arn) 

239 continue 

240 state = str(load_balancer.get("State", {}).get("Code", "unknown")) 

241 logger.info( 

242 "Found exactly owned Gateway ALB by hostname: %s (state: %s)", 

243 load_balancer.get("LoadBalancerName", "<unknown>"), 

244 state, 

245 ) 

246 return str(load_balancer["DNSName"]), arn, state 

247 except Exception as exc: # discovery polling retries 

248 logger.warning("Error finding Gateway ALB by hostname: %s", exc) 

249 return None, None, None 

250 

251 

252def _describe_tags(elb_client: Any, load_balancer_arns: list[str]) -> dict[str, dict[str, str]]: 

253 """Return ELB tags by ARN, respecting the API's 20-resource limit.""" 

254 tags_by_arn: dict[str, dict[str, str]] = {} 

255 for index in range(0, len(load_balancer_arns), 20): 

256 response = elb_client.describe_tags(ResourceArns=load_balancer_arns[index : index + 20]) 

257 for description in response.get("TagDescriptions", []): 

258 arn = description.get("ResourceArn") 

259 if not isinstance(arn, str): 

260 continue 

261 tags_by_arn[arn] = { 

262 str(tag["Key"]): str(tag["Value"]) 

263 for tag in description.get("Tags", []) 

264 if "Key" in tag and "Value" in tag 

265 } 

266 return tags_by_arn 

267 

268 

269def find_platform_alb_by_tags( 

270 elb_client: Any, cluster_name: str 

271) -> tuple[str | None, str | None, str | None]: 

272 """Find the Gateway ALB only when both exact ownership tags match. 

273 

274 This fallback is used only while the exact Gateway status has no address. 

275 Cluster-only matches, alternative cluster tags, NLBs, and internet-facing 

276 ALBs are deliberately rejected. 

277 """ 

278 try: 

279 load_balancers = [ 

280 load_balancer 

281 for load_balancer in _list_load_balancers(elb_client) 

282 if load_balancer.get("Type") == "application" 

283 and load_balancer.get("Scheme") == "internal" 

284 ] 

285 if not load_balancers: 

286 return None, None, None 

287 

288 arns = [str(load_balancer["LoadBalancerArn"]) for load_balancer in load_balancers] 

289 tags_by_arn = _describe_tags(elb_client, arns) 

290 for load_balancer in load_balancers: 

291 arn = str(load_balancer["LoadBalancerArn"]) 

292 tags = tags_by_arn.get(arn, {}) 

293 if not ( 

294 tags.get(GATEWAY_TAG) == GATEWAY_REFERENCE and tags.get(CLUSTER_TAG) == cluster_name 

295 ): 

296 continue 

297 state = str(load_balancer.get("State", {}).get("Code", "unknown")) 

298 logger.info( 

299 "Found Gateway ALB by exact tags: %s (state: %s)", 

300 load_balancer.get("LoadBalancerName", "<unknown>"), 

301 state, 

302 ) 

303 return str(load_balancer["DNSName"]), arn, state 

304 except Exception as exc: # discovery polling retries 

305 logger.warning("Error finding Gateway ALB by tags: %s", exc) 

306 return None, None, None 

307 

308 

309def find_active_alb( 

310 elb_client: Any, 

311 http: urllib3.PoolManager, 

312 k8s_endpoint: str, 

313 k8s_headers: dict[str, str], 

314 cluster_name: str, 

315) -> tuple[str | None, str | None]: 

316 """Find the active ALB owned by the exact GCO Gateway. 

317 

318 A nonempty Gateway status address is authoritative. Tag fallback is allowed 

319 only when that address is absent; it never overrides a provisioning or 

320 otherwise unresolved ALB named by Gateway status. 

321 """ 

322 hostname = find_gateway_address(http, k8s_endpoint, k8s_headers) 

323 if hostname: 

324 dns_name, arn, state = find_alb_by_gateway_hostname(elb_client, hostname, cluster_name) 

325 if arn and state == "active": 

326 logger.info("Found active ALB from Gateway status: %s", hostname) 

327 return dns_name, arn 

328 if arn: 

329 logger.info("Gateway ALB state is %r; waiting for 'active'", state) 

330 return None, None 

331 

332 dns_name, arn, state = find_platform_alb_by_tags(elb_client, cluster_name) 

333 if arn and state == "active": 

334 return dns_name, arn 

335 if arn: 

336 logger.info("Gateway ALB found by tags but state is %r; waiting for 'active'", state) 

337 return None, None 

338 

339 

340def check_existing_ga_endpoint(ga_client: Any, endpoint_group_arn: str, alb_arn: str) -> bool: 

341 """Return whether the exact ALB is already registered with GA.""" 

342 try: 

343 endpoint_group = ga_client.describe_endpoint_group(EndpointGroupArn=endpoint_group_arn) 

344 endpoints = endpoint_group.get("EndpointGroup", {}).get("EndpointDescriptions", []) 

345 if any(endpoint.get("EndpointId") == alb_arn for endpoint in endpoints): 

346 logger.info("ALB %s is already registered with GA", alb_arn) 

347 return True 

348 except Exception as exc: # add_endpoints remains authoritative 

349 logger.warning("Error checking existing GA endpoints: %s", exc) 

350 return False 

351 

352 

353def scrub_stale_ga_endpoints(ga_client: Any, endpoint_group_arn: str, correct_alb_arn: str) -> None: 

354 """Remove every GA endpoint other than the exact Gateway ALB.""" 

355 try: 

356 endpoint_group = ga_client.describe_endpoint_group(EndpointGroupArn=endpoint_group_arn) 

357 endpoints = endpoint_group.get("EndpointGroup", {}).get("EndpointDescriptions", []) 

358 for endpoint in endpoints: 

359 endpoint_id = endpoint.get("EndpointId", "") 

360 if not endpoint_id or endpoint_id == correct_alb_arn: 

361 continue 

362 logger.warning( 

363 "Removing stale GA endpoint %s; exact Gateway ALB is %s", 

364 endpoint_id, 

365 correct_alb_arn, 

366 ) 

367 try: 

368 ga_client.remove_endpoints( 

369 EndpointGroupArn=endpoint_group_arn, 

370 EndpointIdentifiers=[{"EndpointId": endpoint_id}], 

371 ) 

372 except ClientError as exc: 

373 error_code = exc.response.get("Error", {}).get("Code", "") 

374 if error_code == "EndpointNotFoundException": 

375 logger.info("GA endpoint %s was already absent", endpoint_id) 

376 else: 

377 logger.error("Failed to remove stale GA endpoint %s: %s", endpoint_id, exc) 

378 raise 

379 except Exception as exc: 

380 logger.error("Error scrubbing stale GA endpoints: %s", exc) 

381 raise 

382 

383 

384def register_alb_with_ga(ga_client: Any, endpoint_group_arn: str, alb_arn: str) -> None: 

385 """Register the exact Gateway ALB with GA, idempotently.""" 

386 if check_existing_ga_endpoint(ga_client, endpoint_group_arn, alb_arn): 

387 return 

388 try: 

389 ga_client.add_endpoints( 

390 EndpointGroupArn=endpoint_group_arn, 

391 EndpointConfigurations=[ 

392 { 

393 "EndpointId": alb_arn, 

394 "Weight": 100, 

395 "ClientIPPreservationEnabled": True, 

396 } 

397 ], 

398 ) 

399 logger.info("Registered Gateway ALB %s with Global Accelerator", alb_arn) 

400 except ClientError as exc: 

401 if exc.response.get("Error", {}).get("Code") == "EndpointAlreadyExists": 

402 logger.info("Gateway ALB was already registered with Global Accelerator") 

403 return 

404 raise 

405 

406 

407def ensure_https_health_check( 

408 ga_client: Any, 

409 endpoint_group_arn: str, 

410 health_check_path: str = "/api/v1/health", 

411 expected_alb_arn: str | None = None, 

412 health_check_interval: int = 30, 

413 health_check_threshold: int = 3, 

414) -> None: 

415 """Enforce the endpoint group's configured HTTPS/443 health-check contract. 

416 

417 Interval and threshold come from the cdk.json ``global_accelerator`` block 

418 (baked into the convergence task payload at synth time); the defaults 

419 match the values this handler historically hardcoded, so legacy payloads 

420 without the keys keep converging identically. Comparing them here also 

421 means a non-default configured interval is no longer silently reset to 30 

422 whenever the path or protocol drifts. 

423 

424 When ``expected_alb_arn`` is supplied, only that endpoint is preserved in an 

425 update, preventing an eventually consistent stale endpoint description from 

426 being reintroduced after the scrub. 

427 """ 

428 try: 

429 endpoint_group = ga_client.describe_endpoint_group(EndpointGroupArn=endpoint_group_arn) 

430 group = endpoint_group.get("EndpointGroup", {}) 

431 current_protocol = group.get("HealthCheckProtocol", "TCP") 

432 current_port = int(group.get("HealthCheckPort", 0)) 

433 current_path = group.get("HealthCheckPath", "") 

434 current_interval = int(group.get("HealthCheckIntervalSeconds", 0)) 

435 current_threshold = int(group.get("ThresholdCount", 0)) 

436 if ( 

437 current_protocol == "HTTPS" 

438 and current_port == 443 

439 and current_path == health_check_path 

440 and current_interval == health_check_interval 

441 and current_threshold == health_check_threshold 

442 ): 

443 return 

444 

445 existing_endpoints = [ 

446 { 

447 "EndpointId": endpoint["EndpointId"], 

448 "Weight": endpoint.get("Weight", 100), 

449 "ClientIPPreservationEnabled": endpoint.get("ClientIPPreservationEnabled", True), 

450 } 

451 for endpoint in group.get("EndpointDescriptions", []) 

452 if endpoint.get("EndpointId") 

453 and (expected_alb_arn is None or endpoint.get("EndpointId") == expected_alb_arn) 

454 ] 

455 # EndpointConfigurations replaces the group's endpoint list. Right after 

456 # AddEndpoints the description can still omit the ALB this run just 

457 # registered, and sending the filtered list as-is would then remove 

458 # every endpoint and leave the Region out of Global Accelerator. 

459 if expected_alb_arn is not None and not existing_endpoints: 

460 existing_endpoints = [ 

461 { 

462 "EndpointId": expected_alb_arn, 

463 "Weight": 100, 

464 "ClientIPPreservationEnabled": True, 

465 } 

466 ] 

467 ga_client.update_endpoint_group( 

468 EndpointGroupArn=endpoint_group_arn, 

469 HealthCheckPort=443, 

470 HealthCheckProtocol="HTTPS", 

471 HealthCheckPath=health_check_path, 

472 HealthCheckIntervalSeconds=health_check_interval, 

473 ThresholdCount=health_check_threshold, 

474 EndpointConfigurations=existing_endpoints, 

475 ) 

476 logger.info( 

477 "Global Accelerator health check set to HTTPS/443 %s (interval %ss, threshold %s)", 

478 health_check_path, 

479 health_check_interval, 

480 health_check_threshold, 

481 ) 

482 except ClientError as exc: 

483 logger.error("Failed to enforce GA health-check configuration: %s", exc) 

484 raise 

485 

486 

487def _get_registry_region(properties: dict[str, Any], default: str | None = None) -> str | None: 

488 """Return ``RegistryRegion``, accepting ``GlobalRegion`` for compatibility.""" 

489 value = properties.get("RegistryRegion") or properties.get("GlobalRegion") or default 

490 return str(value) if value else None 

491 

492 

493def store_alb_hostname_in_ssm( 

494 region: str, alb_hostname: str, registry_region: str, project_name: str 

495) -> None: 

496 """Publish the Gateway ALB hostname to the regional endpoint registry.""" 

497 ssm_client = boto3.client("ssm", region_name=registry_region) 

498 parameter_name = f"/{project_name}/alb-hostname-{region}" 

499 ssm_client.put_parameter( 

500 Name=parameter_name, 

501 Value=alb_hostname, 

502 Type="String", 

503 Overwrite=True, 

504 Description=f"ALB hostname for {region} regional cluster", 

505 ) 

506 logger.info("Stored Gateway ALB hostname in SSM: %s = %s", parameter_name, alb_hostname) 

507 

508 

509def delete_alb_hostname_from_ssm( 

510 region: str, 

511 registry_region: str, 

512 project_name: str, 

513 *, 

514 strict: bool = False, 

515) -> None: 

516 """Remove this region's endpoint-registry parameter during cleanup.""" 

517 ssm_client = boto3.client("ssm", region_name=registry_region) 

518 parameter_name = f"/{project_name}/alb-hostname-{region}" 

519 try: 

520 ssm_client.delete_parameter(Name=parameter_name) 

521 logger.info("Deleted Gateway ALB hostname from SSM: %s", parameter_name) 

522 except ClientError as exc: 

523 if exc.response.get("Error", {}).get("Code") == "ParameterNotFound": 

524 logger.info("SSM parameter %s was already absent", parameter_name) 

525 elif strict: 

526 raise 

527 else: 

528 logger.warning("Failed to delete Gateway ALB hostname from SSM: %s", exc) 

529 

530 

531def remove_ga_endpoints( 

532 ga_client: Any, 

533 endpoint_group_arn: str, 

534 *, 

535 strict: bool = False, 

536) -> None: 

537 """Remove every endpoint from one regional GA group.""" 

538 try: 

539 endpoint_group = ga_client.describe_endpoint_group(EndpointGroupArn=endpoint_group_arn) 

540 endpoints = endpoint_group.get("EndpointGroup", {}).get("EndpointDescriptions", []) 

541 for endpoint in endpoints: 

542 endpoint_id = endpoint.get("EndpointId") 

543 if not endpoint_id: 

544 continue 

545 try: 

546 ga_client.remove_endpoints( 

547 EndpointGroupArn=endpoint_group_arn, 

548 EndpointIdentifiers=[{"EndpointId": endpoint_id}], 

549 ) 

550 except ClientError as exc: 

551 if exc.response.get("Error", {}).get("Code") == "EndpointNotFoundException": 

552 continue 

553 if strict: 

554 raise 

555 logger.warning("Failed to remove GA endpoint %s: %s", endpoint_id, exc) 

556 except ClientError as exc: 

557 code = exc.response.get("Error", {}).get("Code") 

558 if code in {"EndpointGroupNotFoundException", "AcceleratorNotFoundException"}: 

559 logger.info("Global Accelerator endpoint group was already absent") 

560 return 

561 if strict: 

562 raise 

563 logger.warning("Failed to clean up GA endpoints: %s", exc) 

564 except Exception as exc: # delete guards remain best effort 

565 if strict: 

566 raise 

567 logger.warning("Failed to clean up GA endpoints: %s", exc) 

568 

569 

570def _accelerator_arn_from_endpoint_group(endpoint_group_arn: str) -> str: 

571 """Derive an accelerator ARN from one of its endpoint-group ARNs.""" 

572 return endpoint_group_arn.split("/listener/")[0] 

573 

574 

575def wait_for_accelerator_deployed( 

576 ga_client: Any, 

577 endpoint_group_arn: str, 

578 timeout_seconds: int = GA_DEPLOYED_WAIT_SECONDS, 

579 *, 

580 strict: bool = False, 

581) -> bool: 

582 """Wait until GA finishes redeploying and releases its managed ENIs. 

583 

584 In strict mode a describe failure raises immediately instead of being 

585 reported as a timeout; a permissions gap must surface as itself. 

586 """ 

587 accelerator_arn = _accelerator_arn_from_endpoint_group(endpoint_group_arn) 

588 start_time = time.time() 

589 while time.time() - start_time < timeout_seconds: 

590 try: 

591 accelerator = ga_client.describe_accelerator(AcceleratorArn=accelerator_arn) 

592 status = accelerator.get("Accelerator", {}).get("Status", "") 

593 except ClientError as exc: 

594 if exc.response.get("Error", {}).get("Code") == "AcceleratorNotFoundException": 

595 return True 

596 if strict: 

597 raise 

598 logger.warning("Failed to describe accelerator status: %s", exc) 

599 return False 

600 if status == "DEPLOYED": 

601 return True 

602 logger.info("Accelerator status=%r; waiting for DEPLOYED", status) 

603 # nosemgrep: arbitrary-sleep - intentional GA redeployment polling 

604 time.sleep(GA_DEPLOYED_POLL_INTERVAL) 

605 logger.warning("Timed out waiting for Global Accelerator to reach DEPLOYED") 

606 return False 

607 

608 

609def deregister_alb_from_ga( 

610 ga_client: Any, 

611 endpoint_group_arn: str, 

612 *, 

613 strict: bool = False, 

614) -> None: 

615 """Remove regional GA endpoints, then wait for managed ENI release.""" 

616 if strict: 

617 remove_ga_endpoints(ga_client, endpoint_group_arn, strict=True) 

618 else: 

619 remove_ga_endpoints(ga_client, endpoint_group_arn) 

620 deployed = wait_for_accelerator_deployed(ga_client, endpoint_group_arn, strict=strict) 

621 if strict and not deployed: 

622 raise TimeoutError("Global Accelerator did not reach DEPLOYED after endpoint removal") 

623 

624 

625def _optional_endpoint_group_arn(properties: dict[str, Any]) -> str | None: 

626 """Normalize an optional endpoint-group property.""" 

627 value = properties.get("EndpointGroupArn") 

628 if value is None: 

629 return None 

630 normalized = str(value).strip() 

631 return normalized or None 

632 

633 

634def _health_check_contract(properties: dict[str, Any]) -> dict[str, Any]: 

635 """Extract the optional ``GaHealthCheck*`` payload keys with legacy defaults. 

636 

637 The regional stack bakes the configured contract into the convergence 

638 payload at synth time. Payloads persisted before these keys existed (and 

639 raw CloudFormation properties from older deployments) fall back to the 

640 values this handler historically hardcoded. 

641 """ 

642 return { 

643 "health_check_path": str(properties.get("GaHealthCheckPath", "/api/v1/health")), 

644 "health_check_interval": int(properties.get("GaHealthCheckInterval", 30)), 

645 "health_check_threshold": int(properties.get("GaHealthCheckThreshold", 3)), 

646 } 

647 

648 

649def register_ga_endpoint( 

650 cluster_name: str, 

651 region: str, 

652 endpoint_group_arn: str | None = None, 

653 registry_region: str = DEFAULT_REGISTRY_REGION, 

654 project_name: str = "gco", 

655 *, 

656 health_check_path: str = "/api/v1/health", 

657 health_check_interval: int = 30, 

658 health_check_threshold: int = 3, 

659) -> dict[str, str]: 

660 """Converge the Gateway ALB, optional GA endpoint, registry, and migration.""" 

661 ca_path: str | None = None 

662 try: 

663 k8s_endpoint, token, ca_path = get_k8s_client(cluster_name, region) 

664 http = urllib3.PoolManager(cert_reqs="CERT_REQUIRED", ca_certs=ca_path) 

665 k8s_headers = {"Authorization": f"Bearer {token}"} 

666 elb_client = boto3.client("elbv2", region_name=region) 

667 

668 logger.info("Waiting for active %s ALB", GATEWAY_REFERENCE) 

669 start_time = time.time() 

670 last_log_time = start_time 

671 alb_hostname: str | None = None 

672 alb_arn: str | None = None 

673 while time.time() - start_time < MAX_WAIT_SECONDS: 

674 alb_hostname, alb_arn = find_active_alb( 

675 elb_client, 

676 http, 

677 k8s_endpoint, 

678 k8s_headers, 

679 cluster_name, 

680 ) 

681 if alb_arn: 

682 break 

683 if time.time() - last_log_time >= 30: 

684 elapsed = int(time.time() - start_time) 

685 logger.info("Still waiting for Gateway ALB (%ss elapsed)", elapsed) 

686 last_log_time = time.time() 

687 # nosemgrep: arbitrary-sleep - intentional Gateway/ALB polling 

688 time.sleep(ALB_POLL_INTERVAL) 

689 

690 if not alb_arn or not alb_hostname: 

691 elapsed = int(time.time() - start_time) 

692 raise TimeoutError( 

693 f"Timed out waiting for active {GATEWAY_REFERENCE} ALB after {elapsed} seconds" 

694 ) 

695 

696 normalized_endpoint_group = ( 

697 str(endpoint_group_arn).strip() if endpoint_group_arn is not None else "" 

698 ) 

699 if normalized_endpoint_group: 

700 ga_client = boto3.client("globalaccelerator", region_name="us-west-2") 

701 register_alb_with_ga(ga_client, normalized_endpoint_group, alb_arn) 

702 scrub_stale_ga_endpoints(ga_client, normalized_endpoint_group, alb_arn) 

703 ensure_https_health_check( 

704 ga_client, 

705 normalized_endpoint_group, 

706 health_check_path=health_check_path, 

707 expected_alb_arn=alb_arn, 

708 health_check_interval=health_check_interval, 

709 health_check_threshold=health_check_threshold, 

710 ) 

711 # AddEndpoints/RemoveEndpoints/UpdateEndpointGroup only submit a 

712 # configuration change; the accelerator serves it from its edge 

713 # locations only after returning to DEPLOYED. Returning success 

714 # earlier reports the deployment as complete while brand-new 

715 # connections to the global endpoint still black-hole for several 

716 # minutes, which a live release run proved by timing out on the 

717 # first health probe after deploy. Wait strictly, within whatever 

718 # remains of this handler's wall-clock budget. 

719 remaining_budget = min( 

720 GA_DEPLOYED_WAIT_SECONDS, 

721 int(MAX_WAIT_SECONDS - (time.time() - start_time)), 

722 ) 

723 if remaining_budget <= 0 or not wait_for_accelerator_deployed( 

724 ga_client, 

725 normalized_endpoint_group, 

726 timeout_seconds=remaining_budget, 

727 strict=True, 

728 ): 

729 raise TimeoutError( 

730 "Global Accelerator did not reach DEPLOYED after endpoint registration" 

731 ) 

732 else: 

733 logger.info("EndpointGroupArn is not configured; skipping Global Accelerator") 

734 

735 # Publication is mandatory even when Global Accelerator is disabled. 

736 store_alb_hostname_in_ssm( 

737 region, 

738 alb_hostname, 

739 registry_region, 

740 project_name, 

741 ) 

742 return {"AlbArn": alb_arn, "AlbHostname": alb_hostname} 

743 finally: 

744 _remove_temporary_ca_file(ca_path) 

745 

746 

747def cleanup_gateway_endpoint( 

748 *, 

749 region: str, 

750 endpoint_group_arn: str | None, 

751 registry_region: str, 

752 project_name: str, 

753) -> dict[str, bool]: 

754 """Strictly fence endpoint publication before Gateway deletion.""" 

755 delete_alb_hostname_from_ssm( 

756 region, 

757 registry_region, 

758 project_name, 

759 strict=True, 

760 ) 

761 if endpoint_group_arn: 

762 ga_client = boto3.client("globalaccelerator", region_name="us-west-2") 

763 deregister_alb_from_ga(ga_client, endpoint_group_arn, strict=True) 

764 return { 

765 "RegistryParameterDeleted": True, 

766 "GlobalAcceleratorDeregistered": bool(endpoint_group_arn), 

767 } 

768 

769 

770def handle_task(event: dict[str, Any]) -> dict[str, Any]: 

771 """Handle Step Functions convergence and teardown task invocations.""" 

772 if event.get("Action") == "cleanup_gateway_endpoint": 

773 return cleanup_gateway_endpoint( 

774 region=str(event["Region"]), 

775 endpoint_group_arn=_optional_endpoint_group_arn(event), 

776 registry_region=_get_registry_region(event, DEFAULT_REGISTRY_REGION) 

777 or DEFAULT_REGISTRY_REGION, 

778 project_name=str(event.get("ProjectName", "gco")), 

779 ) 

780 

781 return register_ga_endpoint( 

782 cluster_name=str(event["ClusterName"]), 

783 region=str(event["Region"]), 

784 endpoint_group_arn=_optional_endpoint_group_arn(event), 

785 registry_region=_get_registry_region(event, DEFAULT_REGISTRY_REGION) 

786 or DEFAULT_REGISTRY_REGION, 

787 project_name=str(event.get("ProjectName", "gco")), 

788 **_health_check_contract(event), 

789 ) 

790 

791 

792def handle_create_update( 

793 event: dict[str, Any], context: Any, props: dict[str, Any], physical_id: str 

794) -> None: 

795 """Handle Create/Update through the raw CloudFormation protocol.""" 

796 data = register_ga_endpoint( 

797 cluster_name=str(props["ClusterName"]), 

798 region=str(props["Region"]), 

799 endpoint_group_arn=_optional_endpoint_group_arn(props), 

800 registry_region=_get_registry_region(props, DEFAULT_REGISTRY_REGION) 

801 or DEFAULT_REGISTRY_REGION, 

802 project_name=str(props.get("ProjectName", "gco")), 

803 **_health_check_contract(props), 

804 ) 

805 send_response(event, context, "SUCCESS", data, physical_id)