Coverage for gco / stacks / regional_api_gateway_stack.py: 100%
48 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-04-30 21:47 +0000
1"""
2Regional API Gateway stack for private EKS cluster access.
4This stack creates a regional API Gateway with a VPC Lambda that can access
5internal ALBs directly. This enables API access when public access is disabled
6(internal ALB only).
8Use Case:
9 When eks_cluster.endpoint_access is PRIVATE and you want the ALB to also
10 be internal-only, this stack provides authenticated API access via a
11 Lambda function deployed inside the VPC.
13Architecture:
14 API Gateway (Regional) → VPC Lambda → Internal ALB → EKS pods
16Security:
17 - API Gateway uses IAM authentication (SigV4)
18 - Lambda runs inside the VPC with access to internal ALB
19 - Same auth token validation as the global path
20 - No public exposure of ALB or EKS API
22Configuration:
23 Enable in cdk.json:
24 {
25 "api_gateway": {
26 "regional_api_enabled": true
27 }
28 }
29"""
31from typing import Any
33from aws_cdk import (
34 CfnOutput,
35 Duration,
36 RemovalPolicy,
37 Stack,
38)
39from aws_cdk import aws_apigateway as apigateway
40from aws_cdk import aws_ec2 as ec2
41from aws_cdk import aws_iam as iam
42from aws_cdk import aws_lambda as lambda_
43from aws_cdk import aws_logs as logs
44from constructs import Construct
46from gco.config.config_loader import ConfigLoader
47from gco.stacks.constants import LAMBDA_PYTHON_RUNTIME
50class GCORegionalApiGatewayStack(Stack):
51 """
52 Regional API Gateway with VPC Lambda for private cluster access.
54 This stack enables API access when the ALB is internal-only by deploying
55 a Lambda function inside the VPC that can reach the internal ALB directly.
57 Attributes:
58 api: Regional REST API with IAM authentication
59 proxy_lambda: VPC Lambda that forwards requests to internal ALB
60 """
62 def __init__(
63 self,
64 scope: Construct,
65 construct_id: str,
66 config: ConfigLoader,
67 region: str,
68 vpc: ec2.IVpc,
69 alb_dns_name: str,
70 auth_secret_arn: str,
71 **kwargs: Any,
72 ) -> None:
73 super().__init__(scope, construct_id, **kwargs)
75 self.config = config
76 self.deployment_region = region
77 self.vpc = vpc
78 self.alb_dns_name = alb_dns_name
79 self.auth_secret_arn = auth_secret_arn
81 # Create VPC Lambda for proxying requests
82 self.proxy_lambda = self._create_vpc_proxy_lambda()
84 # Create regional API Gateway
85 self.api = self._create_api_gateway()
87 # Export outputs
88 self._create_outputs()
90 # Apply cdk-nag suppressions
91 self._apply_nag_suppressions()
93 def _apply_nag_suppressions(self) -> None:
94 """Apply cdk-nag suppressions for this stack."""
95 from gco.stacks.nag_suppressions import apply_all_suppressions
97 apply_all_suppressions(self, stack_type="regional_api_gateway")
99 def _create_vpc_proxy_lambda(self) -> lambda_.Function:
100 """Create VPC Lambda that proxies requests to internal ALB."""
101 project_name = self.config.get_project_name()
103 # Create security group for Lambda
104 lambda_sg = ec2.SecurityGroup(
105 self,
106 "ProxyLambdaSg",
107 vpc=self.vpc,
108 description="Security group for regional API proxy Lambda",
109 allow_all_outbound=True,
110 )
112 # Create IAM role for Lambda
113 # role_name intentionally omitted - let CDK generate unique name
114 lambda_role = iam.Role(
115 self,
116 "ProxyLambdaRole",
117 assumed_by=iam.ServicePrincipal("lambda.amazonaws.com"),
118 managed_policies=[
119 iam.ManagedPolicy.from_aws_managed_policy_name(
120 "service-role/AWSLambdaVPCAccessExecutionRole"
121 )
122 ],
123 )
125 # Grant read access to auth secret
126 lambda_role.add_to_policy(
127 iam.PolicyStatement(
128 effect=iam.Effect.ALLOW,
129 actions=[
130 "secretsmanager:GetSecretValue",
131 "secretsmanager:DescribeSecret",
132 ],
133 resources=[f"{self.auth_secret_arn}*"],
134 )
135 )
137 # Create log group
138 # log_group_name intentionally omitted - let CDK generate unique name
139 log_group = logs.LogGroup(
140 self,
141 "ProxyLambdaLogGroup",
142 retention=logs.RetentionDays.ONE_WEEK,
143 removal_policy=RemovalPolicy.DESTROY,
144 )
146 # Create Lambda function in VPC
147 proxy_lambda = lambda_.Function(
148 self,
149 "RegionalProxyFunction",
150 function_name=f"{project_name}-regional-proxy-{self.deployment_region}",
151 runtime=getattr(lambda_.Runtime, LAMBDA_PYTHON_RUNTIME),
152 handler="handler.lambda_handler",
153 code=lambda_.Code.from_asset("lambda/regional-api-proxy"),
154 timeout=Duration.seconds(29),
155 memory_size=256,
156 role=lambda_role,
157 vpc=self.vpc,
158 vpc_subnets=ec2.SubnetSelection(subnet_type=ec2.SubnetType.PRIVATE_WITH_EGRESS),
159 security_groups=[lambda_sg],
160 environment={
161 "ALB_ENDPOINT": self.alb_dns_name,
162 "SECRET_ARN": self.auth_secret_arn,
163 },
164 log_group=log_group,
165 description=f"Regional API proxy for {self.deployment_region} (VPC Lambda)",
166 tracing=lambda_.Tracing.ACTIVE,
167 )
169 return proxy_lambda
171 def _create_api_gateway(self) -> apigateway.RestApi:
172 """Create regional API Gateway with IAM authentication."""
173 project_name = self.config.get_project_name()
175 # Create CloudWatch log group
176 # log_group_name intentionally omitted - let CDK generate unique name
177 api_log_group = logs.LogGroup(
178 self,
179 "ApiGatewayLogs",
180 retention=logs.RetentionDays.ONE_MONTH,
181 removal_policy=RemovalPolicy.DESTROY,
182 )
184 # Create regional REST API
185 api = apigateway.RestApi(
186 self,
187 "RegionalApi",
188 rest_api_name=f"{project_name}-regional-api-{self.deployment_region}",
189 description=f"Regional API for {project_name} in {self.deployment_region} (private access)",
190 endpoint_types=[apigateway.EndpointType.REGIONAL],
191 deploy=True,
192 deploy_options=apigateway.StageOptions(
193 stage_name="prod",
194 throttling_rate_limit=1000,
195 throttling_burst_limit=2000,
196 logging_level=apigateway.MethodLoggingLevel.INFO,
197 data_trace_enabled=True,
198 metrics_enabled=True,
199 tracing_enabled=True,
200 access_log_destination=apigateway.LogGroupLogDestination(api_log_group),
201 access_log_format=apigateway.AccessLogFormat.json_with_standard_fields(
202 caller=True,
203 http_method=True,
204 ip=True,
205 protocol=True,
206 request_time=True,
207 resource_path=True,
208 response_length=True,
209 status=True,
210 user=True,
211 ),
212 ),
213 cloud_watch_role=True,
214 )
216 # Add resource policy to restrict to account
217 api.add_to_resource_policy(
218 iam.PolicyStatement(
219 effect=iam.Effect.ALLOW,
220 principals=[iam.AnyPrincipal()],
221 actions=["execute-api:Invoke"],
222 resources=["execute-api:/*"],
223 conditions={"StringEquals": {"aws:PrincipalAccount": self.account}},
224 )
225 )
227 # Create Lambda integration
228 lambda_integration = apigateway.LambdaIntegration(
229 self.proxy_lambda, proxy=True, timeout=Duration.seconds(29)
230 )
232 # Create /api/v1 resource structure
233 api_resource = api.root.add_resource("api")
234 v1_resource = api_resource.add_resource("v1")
236 # Add proxy resource to catch all paths
237 proxy_resource = v1_resource.add_resource("{proxy+}")
239 # Add methods with IAM authentication
240 for method in ["GET", "POST", "PUT", "DELETE", "PATCH"]:
241 proxy_resource.add_method(
242 method,
243 lambda_integration,
244 authorization_type=apigateway.AuthorizationType.IAM,
245 method_responses=[
246 apigateway.MethodResponse(status_code="200"),
247 apigateway.MethodResponse(status_code="400"),
248 apigateway.MethodResponse(status_code="403"),
249 apigateway.MethodResponse(status_code="500"),
250 ],
251 )
253 return api
255 def _create_outputs(self) -> None:
256 """Export regional API Gateway endpoint."""
257 project_name = self.config.get_project_name()
259 CfnOutput(
260 self,
261 "RegionalApiEndpoint",
262 value=self.api.url,
263 description=f"Regional API Gateway endpoint for {self.deployment_region}",
264 export_name=f"{project_name}-regional-api-endpoint-{self.deployment_region}",
265 )