Coverage for gco/stacks/regional_api_gateway_stack.py: 97.14%
95 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-30 21:22 +0000
1"""
2Regional API Gateway bridge for authenticated access to private regional ALBs.
4Every deployment creates this regional bridge so the centralized aggregator has
5a reachable, IAM-authenticated path into each regional VPC. In the commercial
6``aws`` partition, ``api_gateway.regional_api_enabled`` optionally admits other
7same-account principals. In every other AWS partition, Global Accelerator is
8omitted and this regional IAM path is enabled for same-account callers
9regardless of that setting.
11Architecture:
12 Aggregator → Regional API Gateway → buffered VPC Lambda → Internal ALB → EKS pods
13 User (optional in ``aws``; required elsewhere) ────────┤
14 └→ streaming VPC Lambda → inference proxy
16Security:
17 - API Gateway uses AWS-managed TLS and IAM authentication (SigV4)
18 - The resource policy always admits only the aggregator role by default
19 - Optional direct mode additionally admits IAM-authorized account principals
20 - Lambda runs inside the VPC with access to the internal ALB
21 - Lambda verifies the deployment-local ALB certificate with explicit SNI
22 - Lambda adds a short-lived per-request HMAC envelope to the ALB request
23 - No public exposure of the ALB or EKS API
25Configuration:
26 In the commercial ``aws`` partition, set
27 ``api_gateway.regional_api_enabled`` to ``true`` when callers need direct
28 region-pinned access. Outside that partition, the regional API is the
29 supported workload ingress and same-account access is forced on. Global
30 aggregation always uses its dedicated role in every partition.
31"""
33from typing import Any
35from aws_cdk import (
36 CfnOutput,
37 Duration,
38 RemovalPolicy,
39 Stack,
40)
41from aws_cdk import aws_apigateway as apigateway
42from aws_cdk import aws_ec2 as ec2
43from aws_cdk import aws_iam as iam
44from aws_cdk import aws_lambda as lambda_
45from aws_cdk import aws_logs as logs
46from constructs import Construct
48from gco.config.config_loader import ConfigLoader
49from gco.stacks.constants import (
50 AGGREGATOR_REGIONAL_API_ROUTES,
51 DEFAULT_MAX_REQUEST_BODY_BYTES,
52 LAMBDA_NODEJS_RUNTIME,
53 LAMBDA_PYTHON_RUNTIME,
54 backend_tls_root_ca_parameter_name,
55 backend_tls_server_name,
56 validated_request_body_limit,
57)
59# <pyflowchart-code-diagram> BEGIN - auto-inserted, do not edit
60# Generated at (UTC): 2026-07-18T01:03:40Z
61# Flowchart(s) generated from this file:
62# * ``GCORegionalApiGatewayStack.__init__`` -> ``diagrams/code_diagrams/gco/stacks/regional_api_gateway_stack.GCORegionalApiGatewayStack___init__.html``
63# (PNG: ``diagrams/code_diagrams/gco/stacks/regional_api_gateway_stack.GCORegionalApiGatewayStack___init__.png``)
64# Regenerate with ``python diagrams/code_diagrams/generate.py``.
65# <pyflowchart-code-diagram> END
68class GCORegionalApiGatewayStack(Stack):
69 """Regional aggregation bridge with optional direct caller access.
71 The VPC Lambda gives the global aggregator a reachable path to one internal
72 regional ALB. Direct region-pinned access for other IAM-authorized account
73 principals is optional in the commercial ``aws`` partition and mandatory
74 in partitions where Global Accelerator is unavailable.
76 Attributes:
77 api: Regional REST API with IAM authentication.
78 proxy_lambda: Buffered VPC Lambda for ``/api/v1/*`` requests.
79 inference_proxy_lambda: Response-streaming VPC Lambda for ``/inference/*``.
80 """
82 def __init__(
83 self,
84 scope: Construct,
85 construct_id: str,
86 config: ConfigLoader,
87 region: str,
88 vpc: ec2.IVpc,
89 auth_secret_arn: str,
90 aggregator_role_arn: str,
91 alb_dns_name: str | None = None,
92 **kwargs: Any,
93 ) -> None:
94 super().__init__(scope, construct_id, **kwargs)
96 self.config = config
97 self.deployment_region = region
98 self.vpc = vpc
99 self.alb_dns_name = alb_dns_name
100 supports_global_accelerator = getattr(config, "supports_global_accelerator", None)
101 self.global_accelerator_enabled = (
102 bool(supports_global_accelerator()) if callable(supports_global_accelerator) else True
103 )
104 self.auth_secret_arn = auth_secret_arn
105 self.aggregator_role_arn = aggregator_role_arn
107 # Keep control-plane calls on the established buffered Python proxy and
108 # give inference a separate Node.js response-streaming runtime.
109 self.proxy_lambda = self._create_vpc_proxy_lambda()
110 self.inference_proxy_lambda = self._create_inference_proxy_lambda()
112 # Create regional API Gateway
113 self.api = self._create_api_gateway()
115 # Export outputs
116 self._create_outputs()
118 # Apply cdk-nag suppressions
119 self._apply_nag_suppressions()
121 def _apply_nag_suppressions(self) -> None:
122 """Apply cdk-nag suppressions for this stack."""
123 from gco.stacks.nag_suppressions import apply_all_suppressions
125 apply_all_suppressions(
126 self,
127 stack_type="regional_api_gateway",
128 global_region=self.config.get_global_region(),
129 project_name=self.config.get_project_name(),
130 )
132 def _create_vpc_proxy_lambda(self) -> lambda_.Function:
133 """Create VPC Lambda that proxies requests to internal ALB."""
134 project_name = self.config.get_project_name()
135 backend_tls_config = self.config.get_backend_tls_config()
136 root_ca_parameter_name = backend_tls_root_ca_parameter_name(project_name)
138 # Create security group for Lambda
139 lambda_sg = ec2.SecurityGroup(
140 self,
141 "ProxyLambdaSg",
142 vpc=self.vpc,
143 description="Security group for regional API proxy Lambdas",
144 allow_all_outbound=True,
145 )
146 self._proxy_lambda_security_group = lambda_sg
148 # Create IAM role for Lambda
149 # role_name intentionally omitted - let CDK generate unique name
150 lambda_role = iam.Role(
151 self,
152 "ProxyLambdaRole",
153 assumed_by=iam.ServicePrincipal("lambda.amazonaws.com"),
154 managed_policies=[
155 iam.ManagedPolicy.from_aws_managed_policy_name(
156 "service-role/AWSLambdaVPCAccessExecutionRole"
157 )
158 ],
159 )
161 # Grant read access to auth secret.
162 lambda_role.add_to_policy(
163 iam.PolicyStatement(
164 effect=iam.Effect.ALLOW,
165 actions=[
166 "secretsmanager:GetSecretValue",
167 "secretsmanager:DescribeSecret",
168 ],
169 resources=[f"{self.auth_secret_arn}*"],
170 )
171 )
173 # The Ingress-created ALB does not exist during CDK synthesis. Resolve
174 # its current hostname from the project-scoped SSM registry at request
175 # time, then verify that the hostname belongs to this account, region,
176 # EKS cluster, and platform Ingress before forwarding any request.
177 registry_region = self.config.get_global_region()
178 registry_parameter_arn = (
179 f"arn:{self.partition}:ssm:{registry_region}:{self.account}:"
180 f"parameter/{project_name}/alb-hostname-{self.deployment_region}"
181 )
182 root_ca_parameter_arn = (
183 f"arn:{self.partition}:ssm:{registry_region}:{self.account}:"
184 f"parameter/{root_ca_parameter_name.lstrip('/')}"
185 )
186 lambda_role.add_to_policy(
187 iam.PolicyStatement(
188 effect=iam.Effect.ALLOW,
189 actions=["ssm:GetParameter"],
190 resources=[registry_parameter_arn, root_ca_parameter_arn],
191 )
192 )
193 lambda_role.add_to_policy(
194 iam.PolicyStatement(
195 effect=iam.Effect.ALLOW,
196 actions=[
197 "elasticloadbalancing:DescribeLoadBalancers",
198 "elasticloadbalancing:DescribeTags",
199 ],
200 resources=["*"],
201 )
202 )
204 from gco.stacks.nag_suppressions import acknowledge_nag_findings
206 acknowledge_nag_findings(
207 lambda_role,
208 [
209 {
210 "id": "AwsSolutions-IAM5",
211 "reason": (
212 "ELB DescribeLoadBalancers and DescribeTags do not support "
213 "resource-level scoping. They are read-only and are used only "
214 "to verify that the SSM-registered hostname belongs to this "
215 "account's exact regional GCO cluster and platform Ingress."
216 ),
217 "appliesTo": ["Resource::*"],
218 }
219 ],
220 )
222 # Create log group
223 # log_group_name intentionally omitted - let CDK generate unique name
224 log_group = logs.LogGroup(
225 self,
226 "ProxyLambdaLogGroup",
227 retention=logs.RetentionDays.ONE_WEEK,
228 removal_policy=RemovalPolicy.DESTROY,
229 )
231 # A literal endpoint remains available for isolated stack synthesis and
232 # compatibility callers. Production app wiring omits it so replacements
233 # are discovered from SSM without requiring an ALB at deploy time.
234 environment = {
235 "SECRET_ARN": self.auth_secret_arn,
236 "REGISTRY_REGION": registry_region,
237 "TARGET_REGION": self.deployment_region,
238 "PROJECT_NAME": project_name,
239 "AWS_ACCOUNT_ID": self.account,
240 "AWS_URL_SUFFIX": self.url_suffix,
241 "BACKEND_TLS_SERVER_NAME": backend_tls_server_name(project_name),
242 "BACKEND_TLS_ROOT_CA_PARAMETER": root_ca_parameter_name,
243 "BACKEND_TLS_ROOT_CA_REGION": registry_region,
244 "BACKEND_TLS_CA_CACHE_TTL_SECONDS": str(backend_tls_config["trust_cache_ttl_seconds"]),
245 "BACKEND_TLS_CA_MAX_STALE_SECONDS": str(
246 backend_tls_config["trust_cache_max_stale_seconds"]
247 ),
248 }
249 if self.alb_dns_name:
250 environment["ALB_ENDPOINT"] = self.alb_dns_name
252 # Create Lambda function in VPC
253 proxy_lambda = lambda_.Function(
254 self,
255 "RegionalProxyFunction",
256 function_name=f"{project_name}-regional-proxy-{self.deployment_region}",
257 runtime=getattr(lambda_.Runtime, LAMBDA_PYTHON_RUNTIME),
258 handler="handler.lambda_handler",
259 code=lambda_.Code.from_asset("lambda/regional-api-proxy"),
260 timeout=Duration.seconds(29),
261 memory_size=256,
262 role=lambda_role,
263 vpc=self.vpc,
264 vpc_subnets=ec2.SubnetSelection(subnet_type=ec2.SubnetType.PRIVATE_WITH_EGRESS),
265 security_groups=[lambda_sg],
266 environment=environment,
267 log_group=log_group,
268 description=f"Regional API proxy for {self.deployment_region} (VPC Lambda)",
269 tracing=lambda_.Tracing.ACTIVE,
270 )
272 return proxy_lambda
274 def _create_inference_proxy_lambda(self) -> lambda_.Function:
275 """Create the VPC Lambda that streams inference responses from the ALB."""
276 project_name = self.config.get_project_name()
277 backend_tls_config = self.config.get_backend_tls_config()
278 max_request_body_bytes = validated_request_body_limit(
279 self.config.get_manifest_processor_config().get(
280 "max_request_body_bytes", DEFAULT_MAX_REQUEST_BODY_BYTES
281 )
282 )
283 registry_region = self.config.get_global_region()
284 root_ca_parameter_name = backend_tls_root_ca_parameter_name(project_name)
285 registry_parameter_arn = (
286 f"arn:{self.partition}:ssm:{registry_region}:{self.account}:"
287 f"parameter/{project_name}/alb-hostname-{self.deployment_region}"
288 )
289 root_ca_parameter_arn = (
290 f"arn:{self.partition}:ssm:{registry_region}:{self.account}:"
291 f"parameter/{root_ca_parameter_name.lstrip('/')}"
292 )
294 role = iam.Role(
295 self,
296 "InferenceStreamingProxyRole",
297 assumed_by=iam.ServicePrincipal("lambda.amazonaws.com"),
298 managed_policies=[
299 iam.ManagedPolicy.from_aws_managed_policy_name(
300 "service-role/AWSLambdaVPCAccessExecutionRole"
301 )
302 ],
303 )
304 role.add_to_policy(
305 iam.PolicyStatement(
306 effect=iam.Effect.ALLOW,
307 actions=[
308 "secretsmanager:GetSecretValue",
309 "secretsmanager:DescribeSecret",
310 ],
311 resources=[f"{self.auth_secret_arn}*"],
312 )
313 )
314 role.add_to_policy(
315 iam.PolicyStatement(
316 effect=iam.Effect.ALLOW,
317 actions=["ssm:GetParameter"],
318 resources=[registry_parameter_arn, root_ca_parameter_arn],
319 )
320 )
321 role.add_to_policy(
322 iam.PolicyStatement(
323 effect=iam.Effect.ALLOW,
324 actions=[
325 "elasticloadbalancing:DescribeLoadBalancers",
326 "elasticloadbalancing:DescribeTags",
327 ],
328 resources=["*"],
329 )
330 )
332 from gco.stacks.nag_suppressions import acknowledge_nag_findings
334 acknowledge_nag_findings(
335 role,
336 [
337 {
338 "id": "AwsSolutions-IAM5",
339 "reason": (
340 "ELB ownership verification and the Lambda VPC/X-Ray APIs do not "
341 "support resource-level scoping. Secret and SSM reads remain "
342 "scoped to this deployment's exact resources."
343 ),
344 "appliesTo": ["Resource::*"],
345 }
346 ],
347 )
349 log_group = logs.LogGroup(
350 self,
351 "InferenceStreamingProxyLogGroup",
352 retention=logs.RetentionDays.ONE_WEEK,
353 removal_policy=RemovalPolicy.DESTROY,
354 )
355 return lambda_.Function(
356 self,
357 "InferenceStreamingProxyFunction",
358 function_name=(f"{project_name}-regional-inference-proxy-{self.deployment_region}"),
359 runtime=getattr(lambda_.Runtime, LAMBDA_NODEJS_RUNTIME),
360 handler="index.handler",
361 code=lambda_.Code.from_asset("lambda/inference-streaming-proxy-build"),
362 timeout=Duration.minutes(15),
363 memory_size=256,
364 role=role,
365 vpc=self.vpc,
366 vpc_subnets=ec2.SubnetSelection(subnet_type=ec2.SubnetType.PRIVATE_WITH_EGRESS),
367 security_groups=[self._proxy_lambda_security_group],
368 environment={
369 "ROUTING_MODE": "regional",
370 "MAX_REQUEST_BODY_BYTES": str(max_request_body_bytes),
371 "SECRET_ARN": self.auth_secret_arn,
372 "REGISTRY_REGION": registry_region,
373 "TARGET_REGION": self.deployment_region,
374 "PROJECT_NAME": project_name,
375 "AWS_ACCOUNT_ID": self.account,
376 "AWS_URL_SUFFIX": self.url_suffix,
377 "BACKEND_TLS_SERVER_NAME": backend_tls_server_name(project_name),
378 "BACKEND_TLS_ROOT_CA_PARAMETER": root_ca_parameter_name,
379 "BACKEND_TLS_ROOT_CA_REGION": registry_region,
380 "BACKEND_TLS_CA_CACHE_TTL_SECONDS": str(
381 backend_tls_config["trust_cache_ttl_seconds"]
382 ),
383 "BACKEND_TLS_CA_MAX_STALE_SECONDS": str(
384 backend_tls_config["trust_cache_max_stale_seconds"]
385 ),
386 },
387 log_group=log_group,
388 description=(
389 f"Regional inference response-streaming proxy for {self.deployment_region}"
390 ),
391 tracing=lambda_.Tracing.ACTIVE,
392 )
394 def _create_api_gateway(self) -> apigateway.RestApi:
395 """Create regional API Gateway with IAM authentication."""
396 project_name = self.config.get_project_name()
398 # Create CloudWatch log group
399 # log_group_name intentionally omitted - let CDK generate unique name
400 api_log_group = logs.LogGroup(
401 self,
402 "ApiGatewayLogs",
403 retention=logs.RetentionDays.ONE_MONTH,
404 removal_policy=RemovalPolicy.DESTROY,
405 )
407 api_config = self.config.get_api_gateway_config()
408 configured_log_level = str(api_config["log_level"]).upper()
409 logging_levels = {
410 "OFF": apigateway.MethodLoggingLevel.OFF,
411 "ERROR": apigateway.MethodLoggingLevel.ERROR,
412 "INFO": apigateway.MethodLoggingLevel.INFO,
413 }
414 if configured_log_level not in logging_levels: 414 ↛ 415line 414 didn't jump to line 415 because the condition on line 414 was never true
415 raise ValueError(
416 "api_gateway.log_level must be one of OFF, ERROR, or INFO; "
417 f"got {configured_log_level!r}"
418 )
420 # Create regional REST API
421 api = apigateway.RestApi(
422 self,
423 "RegionalApi",
424 rest_api_name=f"{project_name}-regional-api-{self.deployment_region}",
425 description=f"Direct regional API for {project_name} in {self.deployment_region}",
426 endpoint_types=[apigateway.EndpointType.REGIONAL],
427 deploy=True,
428 deploy_options=apigateway.StageOptions(
429 stage_name="prod",
430 throttling_rate_limit=api_config["throttle_rate_limit"],
431 throttling_burst_limit=api_config["throttle_burst_limit"],
432 logging_level=logging_levels[configured_log_level],
433 # Never put inference prompts/responses (or other API bodies)
434 # into execution logs. Standard access logs and metrics remain.
435 data_trace_enabled=False,
436 metrics_enabled=api_config["metrics_enabled"],
437 tracing_enabled=api_config["tracing_enabled"],
438 access_log_destination=apigateway.LogGroupLogDestination(api_log_group),
439 access_log_format=apigateway.AccessLogFormat.json_with_standard_fields(
440 caller=True,
441 http_method=True,
442 ip=True,
443 protocol=True,
444 request_time=True,
445 resource_path=True,
446 response_length=True,
447 status=True,
448 user=True,
449 ),
450 ),
451 cloud_watch_role=True,
452 # CDK otherwise retains the generated API Gateway account role.
453 cloud_watch_role_removal_policy=RemovalPolicy.DESTROY,
454 )
456 # The bridge is private at the authorization layer by default: only
457 # the aggregator execution role is named in the API resource policy.
458 api.add_to_resource_policy(
459 iam.PolicyStatement(
460 effect=iam.Effect.ALLOW,
461 principals=[iam.ArnPrincipal(self.aggregator_role_arn)],
462 actions=["execute-api:Invoke"],
463 resources=[
464 f"execute-api:/*/{method}/{path}"
465 for method, path in AGGREGATOR_REGIONAL_API_ROUTES
466 ],
467 )
468 )
470 # Direct regional mode is an explicit opt-in in the commercial
471 # partition. It becomes the required supported ingress in partitions
472 # where Global Accelerator does not exist. Methods still require SigV4
473 # and callers still need identity-policy permission to invoke.
474 if ( 474 ↛ 494line 474 didn't jump to line 494 because the condition on line 474 was always true
475 self.config.get_api_gateway_config()["regional_api_enabled"]
476 or not self.global_accelerator_enabled
477 ):
478 api.add_to_resource_policy(
479 iam.PolicyStatement(
480 effect=iam.Effect.ALLOW,
481 principals=[iam.AnyPrincipal()],
482 actions=["execute-api:Invoke"],
483 resources=["execute-api:/*"],
484 conditions={
485 "StringEquals": {"aws:PrincipalAccount": self.account},
486 "ArnNotEquals": {"aws:PrincipalArn": self.aggregator_role_arn},
487 },
488 )
489 )
491 # Keep control-plane integration semantics unchanged. Inference uses
492 # InvokeWithResponseStream and may remain open for API Gateway's full
493 # 15-minute streaming integration window; request bodies are buffered.
494 control_plane_integration = apigateway.LambdaIntegration(
495 self.proxy_lambda, proxy=True, timeout=Duration.seconds(29)
496 )
497 inference_integration = apigateway.LambdaIntegration(
498 self.inference_proxy_lambda,
499 proxy=True,
500 timeout=Duration.minutes(15),
501 response_transfer_mode=apigateway.ResponseTransferMode.STREAM,
502 )
504 # API Gateway greedy resources do not cross a root segment, so
505 # /api/v1/{proxy+} cannot match /inference/{endpoint}/....
506 api_resource = api.root.add_resource("api")
507 v1_resource = api_resource.add_resource("v1")
508 api_proxy_resource = v1_resource.add_resource("{proxy+}")
509 inference_resource = api.root.add_resource("inference")
510 inference_proxy_resource = inference_resource.add_resource("{proxy+}")
512 for method in ["GET", "HEAD", "POST", "PUT", "DELETE", "PATCH", "OPTIONS"]:
513 api_proxy_resource.add_method(
514 method,
515 control_plane_integration,
516 authorization_type=apigateway.AuthorizationType.IAM,
517 method_responses=[
518 apigateway.MethodResponse(status_code="200"),
519 apigateway.MethodResponse(status_code="400"),
520 apigateway.MethodResponse(status_code="403"),
521 apigateway.MethodResponse(status_code="500"),
522 ],
523 )
525 for method in ["GET", "HEAD", "POST"]:
526 inference_proxy_resource.add_method(
527 method,
528 inference_integration,
529 authorization_type=apigateway.AuthorizationType.IAM,
530 method_responses=[
531 apigateway.MethodResponse(status_code="200"),
532 apigateway.MethodResponse(status_code="400"),
533 apigateway.MethodResponse(status_code="404"),
534 apigateway.MethodResponse(status_code="500"),
535 apigateway.MethodResponse(status_code="502"),
536 ],
537 )
539 from gco.stacks.nag_suppressions import acknowledge_nag_findings
541 acknowledge_nag_findings(
542 api.deployment_stage,
543 [
544 {
545 "id": "AwsSolutions-APIG3",
546 "reason": (
547 "This regional bridge is not a general public API: every method "
548 "requires SigV4 and its resource policy admits only the exact "
549 "aggregator role unless account-local direct access is explicitly "
550 "enabled. A separate WAF would duplicate those identity controls."
551 ),
552 },
553 {
554 "id": "NIST.800.53.R5-APIGWAssociatedWithWAF",
555 "reason": (
556 "The IAM-authenticated regional bridge has an aggregator-only resource "
557 "policy by default; unauthorized traffic is rejected before integration."
558 ),
559 },
560 {
561 "id": "PCI.DSS.321-APIGWAssociatedWithWAF",
562 "reason": (
563 "The IAM-authenticated regional bridge has an aggregator-only resource "
564 "policy by default and carries no payment-card-specific public surface."
565 ),
566 },
567 ],
568 )
570 return api
572 def _create_outputs(self) -> None:
573 """Export regional API Gateway endpoint."""
574 project_name = self.config.get_project_name()
576 CfnOutput(
577 self,
578 "RegionalApiEndpoint",
579 value=self.api.url,
580 description=f"Regional API Gateway endpoint for {self.deployment_region}",
581 export_name=f"{project_name}-regional-api-endpoint-{self.deployment_region}",
582 )