Coverage for gco/stacks/regional_api_gateway_stack.py: 100%

1""" 

2Regional API Gateway stack for private EKS cluster access. 

3 

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). 

7 

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. 

12 

13Architecture: 

14 API Gateway (Regional) → VPC Lambda → Internal ALB → EKS pods 

15 

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 

21 

22Configuration: 

23 Enable in cdk.json: 

24 { 

25 "api_gateway": { 

26 "regional_api_enabled": true 

27 } 

28 } 

29""" 

30 

31from typing import Any 

32 

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 

45 

46from gco.config.config_loader import ConfigLoader 

47from gco.stacks.constants import LAMBDA_PYTHON_RUNTIME 

48 

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

50# Flowchart(s) generated from this file: 

51# * ``GCORegionalApiGatewayStack.__init__`` -> ``diagrams/code_diagrams/gco/stacks/regional_api_gateway_stack.GCORegionalApiGatewayStack___init__.html`` 

52# (PNG: ``diagrams/code_diagrams/gco/stacks/regional_api_gateway_stack.GCORegionalApiGatewayStack___init__.png``) 

53# Regenerate with ``python diagrams/code_diagrams/generate.py``. 

54# <pyflowchart-code-diagram> END 

55 

56 

57class GCORegionalApiGatewayStack(Stack): 

58 """ 

59 Regional API Gateway with VPC Lambda for private cluster access. 

60 

61 This stack enables API access when the ALB is internal-only by deploying 

62 a Lambda function inside the VPC that can reach the internal ALB directly. 

63 

64 Attributes: 

65 api: Regional REST API with IAM authentication 

66 proxy_lambda: VPC Lambda that forwards requests to internal ALB 

67 """ 

68 

69 def __init__( 

70 self, 

71 scope: Construct, 

72 construct_id: str, 

73 config: ConfigLoader, 

74 region: str, 

75 vpc: ec2.IVpc, 

76 alb_dns_name: str, 

77 auth_secret_arn: str, 

78 **kwargs: Any, 

79 ) -> None: 

80 super().__init__(scope, construct_id, **kwargs) 

81 

82 self.config = config 

83 self.deployment_region = region 

84 self.vpc = vpc 

85 self.alb_dns_name = alb_dns_name 

86 self.auth_secret_arn = auth_secret_arn 

87 

88 # Create VPC Lambda for proxying requests 

89 self.proxy_lambda = self._create_vpc_proxy_lambda() 

90 

91 # Create regional API Gateway 

92 self.api = self._create_api_gateway() 

93 

94 # Export outputs 

95 self._create_outputs() 

96 

97 # Apply cdk-nag suppressions 

98 self._apply_nag_suppressions() 

99 

100 def _apply_nag_suppressions(self) -> None: 

101 """Apply cdk-nag suppressions for this stack.""" 

102 from gco.stacks.nag_suppressions import apply_all_suppressions 

103 

104 apply_all_suppressions(self, stack_type="regional_api_gateway") 

105 

106 def _create_vpc_proxy_lambda(self) -> lambda_.Function: 

107 """Create VPC Lambda that proxies requests to internal ALB.""" 

108 project_name = self.config.get_project_name() 

109 

110 # Create security group for Lambda 

111 lambda_sg = ec2.SecurityGroup( 

112 self, 

113 "ProxyLambdaSg", 

114 vpc=self.vpc, 

115 description="Security group for regional API proxy Lambda", 

116 allow_all_outbound=True, 

117 ) 

118 

119 # Create IAM role for Lambda 

120 # role_name intentionally omitted - let CDK generate unique name 

121 lambda_role = iam.Role( 

122 self, 

123 "ProxyLambdaRole", 

124 assumed_by=iam.ServicePrincipal("lambda.amazonaws.com"), 

125 managed_policies=[ 

126 iam.ManagedPolicy.from_aws_managed_policy_name( 

127 "service-role/AWSLambdaVPCAccessExecutionRole" 

128 ) 

129 ], 

130 ) 

131 

132 # Grant read access to auth secret 

133 lambda_role.add_to_policy( 

134 iam.PolicyStatement( 

135 effect=iam.Effect.ALLOW, 

136 actions=[ 

137 "secretsmanager:GetSecretValue", 

138 "secretsmanager:DescribeSecret", 

139 ], 

140 resources=[f"{self.auth_secret_arn}*"], 

141 ) 

142 ) 

143 

144 # Create log group 

145 # log_group_name intentionally omitted - let CDK generate unique name 

146 log_group = logs.LogGroup( 

147 self, 

148 "ProxyLambdaLogGroup", 

149 retention=logs.RetentionDays.ONE_WEEK, 

150 removal_policy=RemovalPolicy.DESTROY, 

151 ) 

152 

153 # Create Lambda function in VPC 

154 proxy_lambda = lambda_.Function( 

155 self, 

156 "RegionalProxyFunction", 

157 function_name=f"{project_name}-regional-proxy-{self.deployment_region}", 

158 runtime=getattr(lambda_.Runtime, LAMBDA_PYTHON_RUNTIME), 

159 handler="handler.lambda_handler", 

160 code=lambda_.Code.from_asset("lambda/regional-api-proxy"), 

161 timeout=Duration.seconds(29), 

162 memory_size=256, 

163 role=lambda_role, 

164 vpc=self.vpc, 

165 vpc_subnets=ec2.SubnetSelection(subnet_type=ec2.SubnetType.PRIVATE_WITH_EGRESS), 

166 security_groups=[lambda_sg], 

167 environment={ 

168 "ALB_ENDPOINT": self.alb_dns_name, 

169 "SECRET_ARN": self.auth_secret_arn, 

170 }, 

171 log_group=log_group, 

172 description=f"Regional API proxy for {self.deployment_region} (VPC Lambda)", 

173 tracing=lambda_.Tracing.ACTIVE, 

174 ) 

175 

176 return proxy_lambda 

177 

178 def _create_api_gateway(self) -> apigateway.RestApi: 

179 """Create regional API Gateway with IAM authentication.""" 

180 project_name = self.config.get_project_name() 

181 

182 # Create CloudWatch log group 

183 # log_group_name intentionally omitted - let CDK generate unique name 

184 api_log_group = logs.LogGroup( 

185 self, 

186 "ApiGatewayLogs", 

187 retention=logs.RetentionDays.ONE_MONTH, 

188 removal_policy=RemovalPolicy.DESTROY, 

189 ) 

190 

191 # Create regional REST API 

192 api = apigateway.RestApi( 

193 self, 

194 "RegionalApi", 

195 rest_api_name=f"{project_name}-regional-api-{self.deployment_region}", 

196 description=f"Regional API for {project_name} in {self.deployment_region} (private access)", 

197 endpoint_types=[apigateway.EndpointType.REGIONAL], 

198 deploy=True, 

199 deploy_options=apigateway.StageOptions( 

200 stage_name="prod", 

201 throttling_rate_limit=1000, 

202 throttling_burst_limit=2000, 

203 logging_level=apigateway.MethodLoggingLevel.INFO, 

204 data_trace_enabled=True, 

205 metrics_enabled=True, 

206 tracing_enabled=True, 

207 access_log_destination=apigateway.LogGroupLogDestination(api_log_group), 

208 access_log_format=apigateway.AccessLogFormat.json_with_standard_fields( 

209 caller=True, 

210 http_method=True, 

211 ip=True, 

212 protocol=True, 

213 request_time=True, 

214 resource_path=True, 

215 response_length=True, 

216 status=True, 

217 user=True, 

218 ), 

219 ), 

220 cloud_watch_role=True, 

221 ) 

222 

223 # Add resource policy to restrict to account 

224 api.add_to_resource_policy( 

225 iam.PolicyStatement( 

226 effect=iam.Effect.ALLOW, 

227 principals=[iam.AnyPrincipal()], 

228 actions=["execute-api:Invoke"], 

229 resources=["execute-api:/*"], 

230 conditions={"StringEquals": {"aws:PrincipalAccount": self.account}}, 

231 ) 

232 ) 

233 

234 # Create Lambda integration 

235 lambda_integration = apigateway.LambdaIntegration( 

236 self.proxy_lambda, proxy=True, timeout=Duration.seconds(29) 

237 ) 

238 

239 # Create /api/v1 resource structure 

240 api_resource = api.root.add_resource("api") 

241 v1_resource = api_resource.add_resource("v1") 

242 

243 # Add proxy resource to catch all paths 

244 proxy_resource = v1_resource.add_resource("{proxy+}") 

245 

246 # Add methods with IAM authentication 

247 for method in ["GET", "POST", "PUT", "DELETE", "PATCH"]: 

248 proxy_resource.add_method( 

249 method, 

250 lambda_integration, 

251 authorization_type=apigateway.AuthorizationType.IAM, 

252 method_responses=[ 

253 apigateway.MethodResponse(status_code="200"), 

254 apigateway.MethodResponse(status_code="400"), 

255 apigateway.MethodResponse(status_code="403"), 

256 apigateway.MethodResponse(status_code="500"), 

257 ], 

258 ) 

259 

260 return api 

261 

262 def _create_outputs(self) -> None: 

263 """Export regional API Gateway endpoint.""" 

264 project_name = self.config.get_project_name() 

265 

266 CfnOutput( 

267 self, 

268 "RegionalApiEndpoint", 

269 value=self.api.url, 

270 description=f"Regional API Gateway endpoint for {self.deployment_region}", 

271 export_name=f"{project_name}-regional-api-endpoint-{self.deployment_region}", 

272 )