feat: custom domain name support for private endpoints

samtranslator/internal/schema_source/aws_serverless_api.py

@@ -184,6 +184,24 @@ class Domain(BaseModel):
     )
 
 
+class DomainV2(BaseModel):
+    BasePath: Optional[PassThroughProp] = domain("BasePath")
+    NormalizeBasePath: Optional[bool] = domain("NormalizeBasePath")
+    CertificateArn: PassThroughProp = domain("CertificateArn")
+    DomainName: PassThroughProp = passthrough_prop(
+        DOMAIN_STEM,
+        "DomainName",
+        ["AWS::ApiGateway::DomainNameV2", "Properties", "DomainName"],
+    )
+    EndpointConfiguration: Optional[SamIntrinsicable[Literal["PRIVATE"]]] = domain("EndpointConfiguration")
+    Route53: Optional[Route53] = domain("Route53")
+    SecurityPolicy: Optional[PassThroughProp] = passthrough_prop(
+        DOMAIN_STEM,
+        "SecurityPolicy",
+        ["AWS::ApiGateway::DomainNameV2", "Properties", "SecurityPolicy"],
+    )
+
+
 class DefinitionUri(BaseModel):
     Bucket: PassThroughProp = passthrough_prop(
         DEFINITION_URI_STEM,

samtranslator/model/api/api_generator.py

@@ -10,8 +10,10 @@
     ApiGatewayApiKey,
     ApiGatewayAuthorizer,
     ApiGatewayBasePathMapping,
+    ApiGatewayBasePathMappingV2,
     ApiGatewayDeployment,
     ApiGatewayDomainName,
+    ApiGatewayDomainNameV2,
     ApiGatewayResponse,
     ApiGatewayRestApi,
     ApiGatewayStage,
@@ -79,6 +81,13 @@ class ApiDomainResponse:
     recordset_group: Any
 
 
+@dataclass
+class ApiDomainResponseV2:
+    domain: Optional[ApiGatewayDomainNameV2]
+    apigw_basepath_mapping_list: Optional[List[ApiGatewayBasePathMappingV2]]
+    recordset_group: Any
+
+
 class SharedApiUsagePlan:
     """
     Collects API information from different API resources in the same template,
@@ -603,6 +612,129 @@ def _construct_api_domain(  # noqa: PLR0912, PLR0915
 
         return ApiDomainResponse(domain, basepath_resource_list, record_set_group)
 
+    def _construct_api_domain_v2(
+        self, rest_api: ApiGatewayRestApi, route53_record_set_groups: Any
+    ) -> ApiDomainResponseV2:
+        """
+        Constructs and returns the ApiGateway Domain and BasepathMapping
+        """
+        if self.domain is None:
+            return ApiDomainResponseV2(None, None, None)
+
+        sam_expect(self.domain, self.logical_id, "Domain").to_be_a_map()
+        domain_name: PassThrough = sam_expect(
+            self.domain.get("DomainName"), self.logical_id, "Domain.DomainName"
+        ).to_not_be_none()
+        domain_name_arn: PassThrough = sam_expect(
+            self.domain.get("DomainNameArn"), self.logical_id, "Domain.DomainNameArn"
+        )
+        certificate_arn: PassThrough = sam_expect(
+            self.domain.get("CertificateArn"), self.logical_id, "Domain.CertificateArn"
+        ).to_not_be_none()
+
+        api_domain_name = "{}{}".format("ApiGatewayDomainNameV2", LogicalIdGenerator("", domain_name).gen())
+        self.domain["ApiDomainName"] = api_domain_name
+        domain = ApiGatewayDomainNameV2(api_domain_name, attributes=self.passthrough_resource_attributes)
+
+        domain.DomainName = domain_name
+        endpoint = self.domain.get("EndpointConfiguration")
+
+        if endpoint not in ["EDGE", "REGIONAL", "PRIVATE"]:
+            raise InvalidResourceException(
+                self.logical_id,
+                "EndpointConfiguration for Custom Domains must be"
+                " one of {}.".format(["EDGE", "REGIONAL", "PRIVATE"]),
+            )
+
+        domain.CertificateArn = certificate_arn
+
+        domain.EndpointConfiguration = {"Types": [endpoint]}
+
+        if self.domain.get("SecurityPolicy", None):
+            domain.SecurityPolicy = self.domain["SecurityPolicy"]
+
+        if self.domain.get("Policy", None):
+            domain.Policy = self.domain["Policy"]
+
+        basepaths: Optional[List[str]]
+        basepath_value = self.domain.get("BasePath")
+        # Create BasepathMappings
+        if self.domain.get("BasePath") and isinstance(basepath_value, str):
+            basepaths = [basepath_value]
+        elif self.domain.get("BasePath") and isinstance(basepath_value, list):
+            basepaths = cast(Optional[List[Any]], basepath_value)
+        else:
+            basepaths = None
+
+        # Boolean to allow/disallow symbols in BasePath property
+        normalize_basepath = self.domain.get("NormalizeBasePath", True)
+
+        basepath_resource_list: List[ApiGatewayBasePathMappingV2] = []
+        if basepaths is None:
+            basepath_mapping = ApiGatewayBasePathMappingV2(
+                self.logical_id + "BasePathMapping", attributes=self.passthrough_resource_attributes
+            )
+            basepath_mapping.DomainNameArn = ref(domain_name_arn)
+            basepath_mapping.RestApiId = ref(rest_api.logical_id)
+            basepath_mapping.Stage = ref(rest_api.logical_id + ".Stage")
+            basepath_resource_list.extend([basepath_mapping])
+        else:
+            sam_expect(basepaths, self.logical_id, "Domain.BasePath").to_be_a_list_of(ExpectedType.STRING)
+            for basepath in basepaths:
+                # Remove possible leading and trailing '/' because a base path may only
+                # contain letters, numbers, and one of "$-_.+!*'()"
+                path = "".join(e for e in basepath if e.isalnum())
+                logical_id = "{}{}{}".format(self.logical_id, path, "BasePathMapping")
+                basepath_mapping = ApiGatewayBasePathMappingV2(
+                    logical_id, attributes=self.passthrough_resource_attributes
+                )
+                basepath_mapping.DomainNameArn = ref(domain_name_arn)
+                basepath_mapping.RestApiId = ref(rest_api.logical_id)
+                basepath_mapping.Stage = ref(rest_api.logical_id + ".Stage")
+                basepath_mapping.BasePath = path if normalize_basepath else basepath
+                basepath_resource_list.extend([basepath_mapping])
+
+        # Create the Route53 RecordSetGroup resource
+        record_set_group = None
+        route53 = self.domain.get("Route53")
+        if route53 is not None:
+            sam_expect(route53, self.logical_id, "Domain.Route53").to_be_a_map()
+            if route53.get("HostedZoneId") is None and route53.get("HostedZoneName") is None:
+                raise InvalidResourceException(
+                    self.logical_id,
+                    "HostedZoneId or HostedZoneName is required to enable Route53 support on Custom Domains.",
+                )
+
+            logical_id_suffix = LogicalIdGenerator(
+                "", route53.get("HostedZoneId") or route53.get("HostedZoneName")
+            ).gen()
+            logical_id = "RecordSetGroup" + logical_id_suffix
+
+            record_set_group = route53_record_set_groups.get(logical_id)
+
+            if route53.get("SeparateRecordSetGroup"):
+                sam_expect(
+                    route53.get("SeparateRecordSetGroup"), self.logical_id, "Domain.Route53.SeparateRecordSetGroup"
+                ).to_be_a_bool()
+                return ApiDomainResponseV2(
+                    domain,
+                    basepath_resource_list,
+                    self._construct_single_record_set_group(self.domain, domain_name, route53),
+                )
+
+            if not record_set_group:
+                record_set_group = Route53RecordSetGroup(logical_id, attributes=self.passthrough_resource_attributes)
+                if "HostedZoneId" in route53:
+                    record_set_group.HostedZoneId = route53.get("HostedZoneId")
+                if "HostedZoneName" in route53:
+                    record_set_group.HostedZoneName = route53.get("HostedZoneName")
+                record_set_group.RecordSets = []
+                route53_record_set_groups[logical_id] = record_set_group
+
+            record_set_group.RecordSets += self._construct_record_sets_for_domain(self.domain, domain_name, route53)
+
+        return ApiDomainResponseV2(domain, basepath_resource_list, record_set_group)
+
     def _construct_single_record_set_group(
         self, domain: Dict[str, Any], api_domain_name: str, route53: Any
     ) -> Route53RecordSetGroup:
@@ -677,9 +809,15 @@ def to_cloudformation(
         :rtype: tuple
         """
         rest_api = self._construct_rest_api()
-        api_domain_response = self._construct_api_domain(rest_api, route53_record_set_groups)
-        domain = api_domain_response.domain
-        basepath_mapping = api_domain_response.apigw_basepath_mapping_list
+        if self.endpoint_configuration == "PRIVATE":
+            api_domain_response = self._construct_api_domain_v2(rest_api, route53_record_set_groups)
+            domain = api_domain_response.domain
+            basepath_mapping = api_domain_response.apigw_basepath_mapping_list
+        else:
+            api_domain_response = self._construct_api_domain(rest_api, route53_record_set_groups)
+            domain = api_domain_response.domain
+            basepath_mapping = api_domain_response.apigw_basepath_mapping_list
+
         route53_recordsetGroup = api_domain_response.recordset_group
 
         deployment = self._construct_deployment(rest_api)

samtranslator/model/apigateway.py

@@ -230,6 +230,23 @@ class ApiGatewayDomainName(Resource):
     OwnershipVerificationCertificateArn: Optional[PassThrough]
 
 
+class ApiGatewayDomainNameV2(Resource):
+    resource_type = "AWS::ApiGateway::DomainNameV2"
+    property_types = {
+        "DomainName": GeneratedProperty(),
+        "EndpointConfiguration": GeneratedProperty(),
+        "SecurityPolicy": GeneratedProperty(),
+        "CertificateArn": GeneratedProperty(),
+        "Tags": GeneratedProperty(),
+    }
+
+    DomainName: PassThrough
+    EndpointConfiguration: Optional[PassThrough]
+    SecurityPolicy: Optional[PassThrough]
+    CertificateArn: Optional[PassThrough]
+    Tags: Optional[PassThrough]
+
+
 class ApiGatewayBasePathMapping(Resource):
     resource_type = "AWS::ApiGateway::BasePathMapping"
     property_types = {
@@ -240,6 +257,16 @@ class ApiGatewayBasePathMapping(Resource):
     }
 
 
+class ApiGatewayBasePathMappingV2(Resource):
+    resource_type = "AWS::ApiGateway::BasePathMappingV2"
+    property_types = {
+        "BasePath": GeneratedProperty(),
+        "DomainNameArn": GeneratedProperty(),
+        "RestApiId": GeneratedProperty(),
+        "Stage": GeneratedProperty(),
+    }
+
+
 class ApiGatewayUsagePlan(Resource):
     resource_type = "AWS::ApiGateway::UsagePlan"
     property_types = {

samtranslator/model/sam_resources.py

@@ -53,6 +53,7 @@
     ApiGatewayApiKey,
     ApiGatewayDeployment,
     ApiGatewayDomainName,
+    ApiGatewayDomainNameV2,
     ApiGatewayStage,
     ApiGatewayUsagePlan,
     ApiGatewayUsagePlanKey,
@@ -1310,6 +1311,7 @@ class SamApi(SamResourceMacro):
         "Stage": ApiGatewayStage.resource_type,
         "Deployment": ApiGatewayDeployment.resource_type,
         "DomainName": ApiGatewayDomainName.resource_type,
+        "DomainNameV2": ApiGatewayDomainNameV2.resource_type,
         "UsagePlan": ApiGatewayUsagePlan.resource_type,
         "UsagePlanKey": ApiGatewayUsagePlanKey.resource_type,
         "ApiKey": ApiGatewayApiKey.resource_type,

samtranslator/translator/verify_logical_id.py

@@ -15,6 +15,8 @@
     "AWS::Cognito::UserPool": "AWS::Cognito::UserPool",
     "AWS::ApiGateway::DomainName": "AWS::ApiGateway::DomainName",
     "AWS::ApiGateway::BasePathMapping": "AWS::ApiGateway::BasePathMapping",
+    "AWS::ApiGateway::DomainNameV2": "AWS::ApiGateway::DomainNameV2",
+    "AWS::ApiGateway::BasePathMappingV2": "AWS::ApiGateway::BasePathMappingV2",
     "AWS::StepFunctions::StateMachine": "AWS::Serverless::StateMachine",
     "AWS::AppSync::GraphQLApi": "AWS::Serverless::GraphQLApi",
 }

tests/translator/input/api_with_custom_domains_private.yaml

@@ -0,0 +1,69 @@
+Parameters:
+  DomainName:
+    Type: String
+    Default: private.example.com
+    Description: Custom domain name for the API
+
+  CertificateArn:
+    Type: String
+    Default: arn:aws:acm:us-west-2:123456789:certificate/abcd-000-1234-0000-000000abcd
+    Description: ARN of the ACM certificate for the domain
+
+  VpcEndpointId:
+    Type: String
+    Default: vpce-abcd1234efg
+    Description: VPC Endpoint ID for private API access
+
+Resources:
+  MyApi:
+    Type: AWS::Serverless::Api
+    Properties:
+      StageName: prod
+      EndpointConfiguration:
+        Type: PRIVATE
+      Auth:
+        ResourcePolicy:
+          CustomStatements:
+          - Effect: Allow
+            Principal: '*'
+            Action: execute-api:Invoke
+            Resource: execute-api:/*
+          - Effect: Deny
+            Principal: '*'
+            Action: execute-api:Invoke
+            Resource: execute-api:/*
+            Condition:
+              StringNotEquals:
+                aws:SourceVpce: !Ref VpcEndpointId
+
+  ApiDomainName:
+    Type: AWS::ApiGateway::DomainNameV2
+    Properties:
+      DomainName: !Ref DomainName
+      EndpointConfiguration:
+        Types:
+        - PRIVATE
+      SecurityPolicy: TLS_1_2
+      RegionalCertificateArn: !Ref CertificateArn
+
+  ApiMapping:
+    Type: AWS::ApiGateway::BasePathMappingV2
+    Properties:
+      DomainName: !Ref ApiDomainName
+      RestApiId: !Ref MyApi
+      Stage: prod
+
+  DomainNameVpcEndpointAssociation:
+    Type: AWS::ApiGateway::VpcEndpointAssociation
+    Properties:
+      DomainName: !Ref DomainName
+      VpcEndpointId: !Ref VpcEndpointId
+
+Outputs:
+  ApiDomainName:
+    Description: Custom Domain Name for the API
+    Value: !Ref DomainName
+
+  ApiEndpoint:
+    Description: API Gateway endpoint URL
+    Value: !Sub https://${MyApi}.execute-api.${AWS::Region}.amazonaws.com/prod/