import pulumi import pulumi_aws as aws import json # import time def create_api_gatewayv2(api_config, l_api, fn, aws_region, project_name): # Create VPC endpoint for PRIVATE API Gateway # if api_config["type"] == "PRIVATE": # vpce = aws.ec2.VpcEndpoint(f"vpce-{l_api['name']}", # vpc_id=l_api["network_config"]["vpc_id"], # service_name=f"com.amazonaws.{aws_region}.execute-api", # subnet_ids=l_api["network_config"]["private_subnet_ids"], # private_dns_enabled=True, # vpc_endpoint_type="Interface" # ) # HTTP API apigwv2 # Create API Gateway V2 HTTP API api = aws.apigatewayv2.Api(f"api-{l_api['name']}", name=l_api['name'], protocol_type="HTTP", ) sg_vpc_link = aws.ec2.SecurityGroup(f"secgroup-{l_api['name']}", vpc_id=l_api["network_config"]["vpc_id"], ingress=[ aws.ec2.SecurityGroupIngressArgs( protocol="tcp", from_port=0, to_port=0, cidr_blocks=["3.14.44.224/32"] ) ], egress=[ aws.ec2.SecurityGroupEgressArgs( protocol="-1", from_port=0, to_port=0, cidr_blocks=["0.0.0.0/0"] ) ] ) # Create a VPC Link vpc_link = aws.apigatewayv2.VpcLink( f"VpcLink-{project_name}", subnet_ids=l_api["network_config"]["private_subnet_ids"], security_group_ids=sg_vpc_link, ) # Add IAM resource policy to restrict access by VPC (for PRIVATE type) # if api_config["type"] == "PRIVATE": # api_policy = aws.apigatewayv2.ApiPolicy(f"policy-{l_api['name']}", # api_id=api.id, # policy=pulumi.Output.all(api.arn, l_api["network_config"]["vpc_id"]).apply( # lambda args: json.dumps({ # "Version": "2012-10-17", # "Statement": [{ # "Effect": "Allow", # "Principal": "*", # "Action": "execute-api:Invoke", # "Resource": f"{args[0]}/*", # "Condition": { # "StringEquals": { # "aws:sourceVpc": args[1] # } # } # }] # }) # ) # ) integration_get = None integration_post = None # Create Lambda integrations for route in api_config["routes"]: if route["method"] == "GET" and not integration_get: integration_get = aws.apigatewayv2.Integration(f"integration-{l_api['name']}-{route['path'].replace('/', '-')}", api_id=api.id, integration_type="AWS_PROXY", integration_method="GET", integration_uri=fn.invoke_arn, # connection_type="INTERNET", payload_format_version="2.0", # connection_id=vpc_link.id ) elif route["method"] == "POST" and not integration_post: integration_post = aws.apigatewayv2.Integration(f"integration-{l_api['name']}-{route['path'].replace('/', '-')}", api_id=api.id, integration_type="AWS_PROXY", integration_uri=fn.invoke_arn, integration_method="POST", payload_format_version="2.0" ) # Create routes dynamically from config routes = [] for route in api_config["routes"]: if route['method'] == "GET": integration = integration_get elif route['method'] == "POST": integration = integration_post r = aws.apigatewayv2.Route( f"route-{l_api['name']}-{route['method']}-{route['path'].replace('/', '-')}", api_id=api.id, route_key=f"{route['method']} {route['path']}", target=integration.id.apply(lambda id: f"integrations/{id}") ) routes.append(r) # Lambda permission for API Gateway permission = aws.lambda_.Permission( f"permission-{l_api['name']}", action="lambda:InvokeFunction", function=fn.name, principal="apigateway.amazonaws.com", source_arn=api.execution_arn.apply(lambda arn: f"{arn}/*/*") ) # Create stage stage = aws.apigatewayv2.Stage(f"stage-{l_api['name']}", api_id=api.id, name="$default", auto_deploy=True ) # Export the API URL pulumi.export(f"{l_api['name']}-url", api.api_endpoint) def create_api_gateway(api_config, l_api, fn, account_id, aws_region, project_name): vpce_id = None # Create API Gateway if api_config["type"] == "PRIVATE": if api_config["create_and_allow_vpce"]: # Cria uma nova Security Group para o VPC Endpoint vpc_endpoint_sg = aws.ec2.SecurityGroup(f"api-gateway-vpce-sg-{project_name}", vpc_id=l_api["network_config"]["vpc_id"], description=f"Security Group for API Gateway VPC Endpoint - {project_name}", ingress=[ aws.ec2.SecurityGroupIngressArgs( protocol="tcp", from_port=0, to_port=0, cidr_blocks=["0.0.0.0/0"], description="Allow HTTPS traffic to API Gateway Endpoint" ), ], egress=[ # Permite todo o tráfego de saída. Pode ser restringido se necessário. aws.ec2.SecurityGroupEgressArgs( protocol="-1", # "-1" significa todos os protocolos from_port=0, to_port=0, cidr_blocks=["0.0.0.0/0"], ), ] ) # Create VPC endpoint for PRIVATE API Gateway vpce = aws.ec2.VpcEndpoint(f"vpce-{l_api['name']}", vpc_id=l_api["network_config"]["vpc_id"], service_name=f"com.amazonaws.{aws_region}.execute-api", subnet_ids=l_api["network_config"]["private_subnet_ids"], private_dns_enabled=False, vpc_endpoint_type="Interface", security_group_ids=[vpc_endpoint_sg] ) vpce_id = vpce.id # api = aws.apigateway.RestApi(f"api-{l_api["name"]}", # description=l_api["name"], # fail_on_warnings=False, # put_rest_api_mode='merge', # endpoint_configuration={ # "types": api_config["type"], # "vpc_endpoint_ids": [vpce.id] if api_config["type"] == "PRIVATE" and api_config["create_and_allow_vpce"] else None # } # ) api = aws.apigateway.RestApi(f"api-{l_api["name"]}", description=l_api["name"], put_rest_api_mode='merge' if api_config["type"] == "PRIVATE" else 'overwrite', fail_on_warnings=False, endpoint_configuration={ "types": api_config["type"], "vpc_endpoint_ids": [vpce_id] if api_config["type"] == "PRIVATE" and api_config["create_and_allow_vpce"] else None } ) # Build policy statements policy_outputs = [] if api_config["allow_inbound_any"]: policy_outputs.append( pulumi.Output.all(aws_region, account_id, api.id, vpce_id).apply( lambda args: { "Effect": "Allow", "Principal": "*", "Action": "execute-api:Invoke", "Resource": f"arn:aws:execute-api:{args[0]}:{args[1]}:{args[2]}/*" } ) ) else: if api_config["type"] == "PRIVATE" and api_config["create_and_allow_vpce"]: policy_outputs.append( pulumi.Output.all(aws_region, account_id, api.id, vpce_id).apply( lambda args: { "Effect": "Allow", "Principal": "*", "Action": "execute-api:Invoke", "Resource": f"arn:aws:execute-api:{args[0]}:{args[1]}:{args[2]}/*", "Condition": { "StringEquals": { "aws:sourceVpce": args[3] } } } ) ) if api_config.get("allow_inbound_cidrs"): policy_outputs.append( pulumi.Output.all(aws_region, account_id, api.id).apply( lambda args: { "Effect": "Allow", "Principal": "*", "Action": "execute-api:Invoke", "Resource": f"arn:aws:execute-api:{args[0]}:{args[1]}:{args[2]}/*", "Condition": { "IpAddress": { "aws:SourceIp": api_config.get("allow_inbound_cidrs", []) } } } ) ) if len(policy_outputs) > 0: # Resource policy for private API resource_policy = aws.apigateway.RestApiPolicy(f"policy-{l_api['name']}", rest_api_id=api.id, policy=pulumi.Output.all(*policy_outputs).apply( lambda statements: json.dumps({ "Version": "2012-10-17", "Statement": statements }) ) ) # Create resources and methods dynamically resources = {} api_dependencies = [] # routes = [] for route in api_config["routes"]: path = route["path"].strip("/") path_parts = path.split("/") # Build nested resources parent_id = api.root_resource_id resource_path = "" for part in path_parts: resource_path += f"/{part}" resource_key = resource_path if resource_key not in resources: resources[resource_key] = aws.apigateway.Resource( f"resource-{l_api['name']}-{part}", rest_api=api.id, parent_id=parent_id, path_part=part ) parent_id = resources[resource_key].id # Create method for this route method = aws.apigateway.Method( f"method-{l_api['name']}-{route['method']}-{path.replace('/', '-')}", rest_api=api.id, resource_id=parent_id, http_method=route["method"], authorization=api_config["authorization"] ) # Create integration integration = aws.apigateway.Integration( f"integration-{l_api['name']}-{route['method']}-{path.replace('/', '-')}", rest_api=api.id, resource_id=parent_id, http_method=method.http_method, integration_http_method="POST", # for Lambda integration, it's always POST type="AWS_PROXY", uri=fn.invoke_arn ) method_response = aws.apigateway.MethodResponse(f"methodResponse-{path.replace('/', '-')}", rest_api=api.id, resource_id=parent_id, http_method=method.http_method, status_code="200", response_models={"application/json": "Empty"} ) integration_response = aws.apigateway.IntegrationResponse(f"integrationResponse-{path.replace('/', '-')}", rest_api=api.id, resource_id=parent_id, http_method=method.http_method, status_code="200", selection_pattern="", response_templates={"application/json": ""}, opts=pulumi.ResourceOptions(depends_on=[integration]) ) # # Lambda permission for API Gateway # permission = aws.lambda_.Permission( # f"permission-{l_api['name']}-{route['method']}-{path.replace('/', '-')}", # action="lambda:InvokeFunction", # function=fn.name, # principal="apigateway.amazonaws.com", # source_arn=api.execution_arn.apply( # lambda arn, m=route['method'], r=resource_path: f"{arn}/*/{m}{r}" # ) # ) api_dependencies.append(method) api_dependencies.append(integration) api_dependencies.append(method_response) api_dependencies.append(integration_response) # Lambda permission for API Gateway permission = aws.lambda_.Permission( f"permission-{l_api['name']}-general", action="lambda:InvokeFunction", function=fn.name, principal="apigateway.amazonaws.com", source_arn=api.execution_arn.apply( lambda arn: f"{arn}/*/*" ) ) api_dependencies.append(permission) # method = getattr(apigateway.Method, route["method"]) # routes.append(apigateway.RouteArgs( # path=route["path"], # method=method, # event_handler=fn # )) # api = apigateway.RestAPI("api", # routes=routes, # # type=api_config["type"], # # put_rest_api_mode="merge" if api_config["type"] == "PRIVATE" else "overwrite" # ) # # Create a VPC link to integrate API Gateway with the VPC # vpc_link = aws.apigateway.VpcLink(f"VpcLink-{l_api['name']}", # name=f"VpcLink-{l_api['name']}", # target_arn=l_api["network_config"]["vpc_id"], # tags={ # "Name": f'VpcLink-{l_api["name"]}', # }) # Create a deployment for the API Gateway deployment = aws.apigateway.Deployment(f"deployment-{l_api['name']}", rest_api=api.id, # triggers={"redeployment": str(int(time.time()))}, opts=pulumi.ResourceOptions(depends_on=list(resources.values())+api_dependencies) ) # Create a stage stage = aws.apigateway.Stage(f"stage-{l_api['name']}", deployment=deployment.id, rest_api=api.id, stage_name=api_config["stage_name"], opts=pulumi.ResourceOptions(depends_on=[deployment]) ) if False: #use_api_key: #TODO api_key = aws.apigateway.ApiKey(api_gateway_config["name"], name=api_gateway_config["name"], description=api_gateway_config["description"], enabled=True ) # API Key to Stage usage_plan = aws.apigateway.UsagePlan("api-usage-plan", name=api_gateway_config["usage_plan_name"], description="Usage plan for API Gateway associated with API Key", api_stages=[aws.apigateway.UsagePlanApiStageArgs( api_id=api.id, stage=deployment.stage_name )] ) # API Key to Usage Plan aws.apigateway.UsagePlanKey("api-key-usage-plan-association", key_id=api_key.id, key_type="API_KEY", usage_plan_id=usage_plan.id ) # Export the stage URL pulumi.export(f"{l_api['name']}-url", stage.invoke_url)