Merge branch 'develop' into fix-pydantic1-usage

Author: seshubaws

.cfnlintrc.yaml

@@ -1,6 +1,9 @@
 templates:
   - tests/translator/output/**/*.json
 ignore_templates:
+  - tests/translator/output/**/function_with_function_url_config.json
+  - tests/translator/output/**/function_with_function_url_config_and_autopublishalias.json
+  - tests/translator/output/**/function_with_function_url_config_without_cors_config.json
   - tests/translator/output/**/error_*.json # Fail by design
   - tests/translator/output/**/api_http_paths_with_if_condition.json
   - tests/translator/output/**/api_http_paths_with_if_condition_no_value_else_case.json
@@ -139,7 +142,6 @@ ignore_templates:
   - tests/translator/output/**/managed_policies_everything.json # intentionally contains wrong arns
   - tests/translator/output/**/function_with_provisioned_poller_config.json
   - tests/translator/output/**/function_with_metrics_config.json
-  - tests/translator/output/**/api_with_custom_domains_private.json
 
 ignore_checks:
   - E2531 # Deprecated runtime; not relevant for transform tests
@@ -148,4 +150,4 @@ ignore_checks:
   - E3001 # Invalid or unsupported Type; common in transform tests since they focus on SAM resources
   - W2001 # Parameter not used
   - E3006 # Resource type check; we have some Foo Bar resources
-  - W3037 # Ignore cfn-lint check for non existing IAM permissions
+  - W3037 # Ignore cfn-lint check for non existing IAM permissions

integration/resources/expected/single/basic_function_with_function_url_dual_auth.json

@@ -0,0 +1,22 @@
+[
+  {
+    "LogicalResourceId": "MyLambdaFunction",
+    "ResourceType": "AWS::Lambda::Function"
+  },
+  {
+    "LogicalResourceId": "MyLambdaFunctionUrl",
+    "ResourceType": "AWS::Lambda::Url"
+  },
+  {
+    "LogicalResourceId": "MyLambdaFunctionUrlPublicPermissions",
+    "ResourceType": "AWS::Lambda::Permission"
+  },
+  {
+    "LogicalResourceId": "MyLambdaFunctionURLInvokeAllowPublicAccess",
+    "ResourceType": "AWS::Lambda::Permission"
+  },
+  {
+    "LogicalResourceId": "MyLambdaFunctionRole",
+    "ResourceType": "AWS::IAM::Role"
+  }
+]

integration/resources/expected/single/basic_function_with_function_url_with_autopuplishalias_dual_auth.json

@@ -0,0 +1,30 @@
+[
+  {
+    "LogicalResourceId": "MyLambdaFunction",
+    "ResourceType": "AWS::Lambda::Function"
+  },
+  {
+    "LogicalResourceId": "MyLambdaFunctionRole",
+    "ResourceType": "AWS::IAM::Role"
+  },
+  {
+    "LogicalResourceId": "MyLambdaFunctionVersion",
+    "ResourceType": "AWS::Lambda::Version"
+  },
+  {
+    "LogicalResourceId": "MyLambdaFunctionAliaslive",
+    "ResourceType": "AWS::Lambda::Alias"
+  },
+  {
+    "LogicalResourceId": "MyLambdaFunctionUrlPublicPermissions",
+    "ResourceType": "AWS::Lambda::Permission"
+  },
+  {
+    "LogicalResourceId": "MyLambdaFunctionURLInvokeAllowPublicAccess",
+    "ResourceType": "AWS::Lambda::Permission"
+  },
+  {
+    "LogicalResourceId": "MyLambdaFunctionUrl",
+    "ResourceType": "AWS::Lambda::Url"
+  }
+]

integration/resources/templates/single/basic_function_with_function_url_dual_auth.yaml

@@ -0,0 +1,27 @@
+Resources:
+  MyLambdaFunction:
+    Type: AWS::Serverless::Function
+    Properties:
+      Handler: index.handler
+      Runtime: nodejs18.x
+      CodeUri: ${codeuri}
+      MemorySize: 128
+      FunctionUrlConfig:
+        AuthType: NONE
+        Cors:
+          AllowOrigins:
+          - https://foo.com
+          AllowMethods:
+          - POST
+          AllowCredentials: true
+          AllowHeaders:
+          - x-Custom-Header
+          ExposeHeaders:
+          - x-amzn-header
+          MaxAge: 10
+Outputs:
+  FunctionUrl:
+    Description: URL of the Lambda function
+    Value: !GetAtt MyLambdaFunctionUrl.FunctionUrl
+Metadata:
+  SamTransformTest: true

integration/resources/templates/single/basic_function_with_function_url_with_autopuplishalias_dual_auth.yaml

@@ -0,0 +1,28 @@
+Resources:
+  MyLambdaFunction:
+    Type: AWS::Serverless::Function
+    Properties:
+      Handler: index.handler
+      Runtime: nodejs18.x
+      CodeUri: ${codeuri}
+      MemorySize: 128
+      AutoPublishAlias: live
+      FunctionUrlConfig:
+        AuthType: NONE
+        Cors:
+          AllowOrigins:
+          - https://foo.com
+          AllowMethods:
+          - POST
+          AllowCredentials: true
+          AllowHeaders:
+          - x-Custom-Header
+          ExposeHeaders:
+          - x-amzn-header
+          MaxAge: 10
+Outputs:
+  FunctionUrl:
+    Description: URL of the Lambda function alias
+    Value: !GetAtt MyLambdaFunctionUrl.FunctionUrl
+Metadata:
+  SamTransformTest: true

integration/single/test_basic_function.py

@@ -130,6 +130,73 @@ def test_basic_function_with_url_config(self, file_name, qualifier):
         self.assertEqual(function_url_config["Cors"], cors_config)
         self._assert_invoke(lambda_client, function_name, qualifier, 200)
 
+    @parameterized.expand(
+        [
+            ("single/basic_function_with_function_url_dual_auth", None),
+            ("single/basic_function_with_function_url_with_autopuplishalias_dual_auth", "live"),
+        ]
+    )
+    @skipIf(current_region_does_not_support([LAMBDA_URL]), "Lambda Url is not supported in this testing region")
+    def test_basic_function_with_url_dual_auth(self, file_name, qualifier):
+        """
+        Creates a basic lambda function with Function Url with authtype: None
+        Verifies that 2 AWS::Lambda::Permission resources are created:
+        - lambda:InvokeFunctionUrl
+        - lambda:InvokeFunction with InvokedViaFunctionUrl: True
+        """
+        self.create_and_verify_stack(file_name)
+
+        # Get Lambda permissions
+        lambda_permissions = self.get_stack_resources("AWS::Lambda::Permission")
+
+        # Verify we have exactly 2 permissions
+        self.assertEqual(len(lambda_permissions), 2, "Expected exactly 2 Lambda permissions")
+
+        # Check for the expected permission logical IDs
+        invoke_function_url_permission = None
+        invoke_permission = None
+
+        for permission in lambda_permissions:
+            logical_id = permission["LogicalResourceId"]
+            if "MyLambdaFunctionUrlPublicPermissions" in logical_id:
+                invoke_function_url_permission = permission
+            elif "MyLambdaFunctionURLInvokeAllowPublicAccess" in logical_id:
+                invoke_permission = permission
+
+        # Verify both permissions exist
+        self.assertIsNotNone(invoke_function_url_permission, "Expected MyLambdaFunctionUrlPublicPermissions to exist")
+        self.assertIsNotNone(invoke_permission, "Expected MyLambdaFunctionURLInvokeAllowPublicAccess to exist")
+
+        # Get the function name and URL
+        function_name = self.get_physical_id_by_type("AWS::Lambda::Function")
+        lambda_client = self.client_provider.lambda_client
+
+        # Get the function URL configuration to verify auth type
+        function_url_config = (
+            lambda_client.get_function_url_config(FunctionName=function_name, Qualifier=qualifier)
+            if qualifier
+            else lambda_client.get_function_url_config(FunctionName=function_name)
+        )
+
+        # Verify the auth type is NONE
+        self.assertEqual(function_url_config["AuthType"], "NONE", "Expected AuthType to be NONE")
+
+        # Get the template to check for InvokedViaFunctionUrl property
+        cfn_client = self.client_provider.cfn_client
+        template = cfn_client.get_template(StackName=self.stack_name, TemplateStage="Processed")
+        template_body = template["TemplateBody"]
+
+        # Check if the InvokePermission has InvokedViaFunctionUrl: True
+        # This is a bit hacky but we don't have direct access to the resource properties
+        # We're checking if the string representation of the template contains this property
+        template_str = str(template_body)
+        self.assertIn("InvokedViaFunctionUrl", template_str, "Expected InvokedViaFunctionUrl property in the template")
+
+        # Get the function URL from stack outputs
+        function_url = self.get_stack_output("FunctionUrl")["OutputValue"]
+        # Invoke the function URL and verify the response
+        self._verify_get_request(function_url, self.FUNCTION_OUTPUT)
+
     @skipIf(current_region_does_not_support([CODE_DEPLOY]), "CodeDeploy is not supported in this testing region")
     def test_function_with_deployment_preference_alarms_intrinsic_if(self):
         self.create_and_verify_stack("single/function_with_deployment_preference_alarms_intrinsic_if")

samtranslator/__init__.py

@@ -1 +1 @@
-__version__ = "1.95.0"
+__version__ = "1.97.0"

samtranslator/internal/schema_source/aws_serverless_api.py

@@ -154,6 +154,12 @@ class Route53(BaseModel):
     SetIdentifier: Optional[PassThroughProp]  # TODO: add docs
     Region: Optional[PassThroughProp]  # TODO: add docs
     SeparateRecordSetGroup: Optional[bool]  # TODO: add docs
+    VpcEndpointDomainName: Optional[PassThroughProp]  # TODO: add docs
+    VpcEndpointHostedZoneId: Optional[PassThroughProp]  # TODO: add docs
+
+
+class AccessAssociation(BaseModel):
+    VpcEndpointId: PassThroughProp  # TODO: add docs
 
 
 class Domain(BaseModel):
@@ -185,6 +191,7 @@ class Domain(BaseModel):
         "SecurityPolicy",
         ["AWS::ApiGateway::DomainName", "Properties", "SecurityPolicy"],
     )
+    AccessAssociation: Optional[AccessAssociation]
 
 
 class DefinitionUri(BaseModel):
@@ -307,6 +314,7 @@ class Properties(BaseModel):
     OpenApiVersion: Optional[OpenApiVersion] = properties("OpenApiVersion")
     StageName: SamIntrinsicable[str] = properties("StageName")
     Tags: Optional[DictStrAny] = properties("Tags")
+    Policy: Optional[PassThroughProp]  # TODO: add docs
     PropagateTags: Optional[bool]  # TODO: add docs
     TracingEnabled: Optional[TracingEnabled] = passthrough_prop(
         PROPERTIES_STEM,

samtranslator/model/api/api_generator.py

@@ -13,6 +13,7 @@
     ApiGatewayBasePathMappingV2,
     ApiGatewayDeployment,
     ApiGatewayDomainName,
+    ApiGatewayDomainNameAccessAssociation,
     ApiGatewayDomainNameV2,
     ApiGatewayResponse,
     ApiGatewayRestApi,
@@ -86,6 +87,7 @@ class ApiDomainResponseV2:
     domain: Optional[ApiGatewayDomainNameV2]
     apigw_basepath_mapping_list: Optional[List[ApiGatewayBasePathMappingV2]]
     recordset_group: Any
+    domain_access_association: Any
 
 
 class SharedApiUsagePlan:
@@ -218,6 +220,7 @@ def __init__(  # noqa: PLR0913
         api_key_source_type: Optional[Intrinsicable[str]] = None,
         always_deploy: Optional[bool] = False,
         feature_toggle: Optional[FeatureToggle] = None,
+        policy: Optional[Union[Dict[str, Any], Intrinsicable[str]]] = None,
     ):
         """Constructs an API Generator class that generates API Gateway resources
 
@@ -275,6 +278,7 @@ def __init__(  # noqa: PLR0913
         self.api_key_source_type = api_key_source_type
         self.always_deploy = always_deploy
         self.feature_toggle = feature_toggle
+        self.policy = policy
 
     def _construct_rest_api(self) -> ApiGatewayRestApi:
         """Constructs and returns the ApiGateway RestApi.
@@ -328,6 +332,9 @@ def _construct_rest_api(self) -> ApiGatewayRestApi:
         if self.api_key_source_type:
             rest_api.ApiKeySourceType = self.api_key_source_type
 
+        if self.policy:
+            rest_api.Policy = self.policy
+
         return rest_api
 
     def _validate_properties(self) -> None:
@@ -602,7 +609,7 @@ def _construct_api_domain_v2(
         Constructs and returns the ApiGateway Domain V2 and BasepathMapping V2
         """
         if self.domain is None:
-            return ApiDomainResponseV2(None, None, None)
+            return ApiDomainResponseV2(None, None, None, None)
 
         sam_expect(self.domain, self.logical_id, "Domain").to_be_a_map()
         domain_name: PassThrough = sam_expect(
@@ -657,6 +664,14 @@ def _construct_api_domain_v2(
                 basepath_mapping.BasePath = path if normalize_basepath else basepath
                 basepath_resource_list.extend([basepath_mapping])
 
+        # Create the DomainNameAccessAssociation
+        domain_access_association = self.domain.get("AccessAssociation")
+        domain_access_association_resource = None
+        if domain_access_association is not None:
+            domain_access_association_resource = self._generate_domain_access_association(
+                domain_access_association, domain_name_arn, api_domain_name
+            )
+
         # Create the Route53 RecordSetGroup resource
         record_set_group = None
         route53 = self.domain.get("Route53")
@@ -683,6 +698,7 @@ def _construct_api_domain_v2(
                     domain,
                     basepath_resource_list,
                     self._construct_single_record_set_group(self.domain, domain_name, route53),
+                    domain_access_association_resource,
                 )
 
             if not record_set_group:
@@ -691,7 +707,7 @@ def _construct_api_domain_v2(
 
             record_set_group.RecordSets += self._construct_record_sets_for_domain(self.domain, domain_name, route53)
 
-        return ApiDomainResponseV2(domain, basepath_resource_list, record_set_group)
+        return ApiDomainResponseV2(domain, basepath_resource_list, record_set_group, domain_access_association_resource)
 
     def _get_basepaths(self) -> Optional[List[str]]:
         if self.domain is None:
@@ -779,11 +795,14 @@ def _construct_alias_target(self, domain: Dict[str, Any], api_domain_name: str,
         if domain.get("EndpointConfiguration") == "REGIONAL":
             alias_target["HostedZoneId"] = fnGetAtt(api_domain_name, "RegionalHostedZoneId")
             alias_target["DNSName"] = fnGetAtt(api_domain_name, "RegionalDomainName")
-        else:
+        elif domain.get("EndpointConfiguration") == "EDGE":
             if route53.get("DistributionDomainName") is None:
                 route53["DistributionDomainName"] = fnGetAtt(api_domain_name, "DistributionDomainName")
             alias_target["HostedZoneId"] = "Z2FDTNDATAQYW2"
             alias_target["DNSName"] = route53.get("DistributionDomainName")
+        else:
+            alias_target["HostedZoneId"] = route53.get("VpcEndpointHostedZoneId")
+            alias_target["DNSName"] = route53.get("VpcEndpointDomainName")
         return alias_target
 
     def _create_basepath_mapping(
@@ -833,12 +852,17 @@ def to_cloudformation(
         domain: Union[Resource, None]
         basepath_mapping: Union[List[ApiGatewayBasePathMapping], List[ApiGatewayBasePathMappingV2], None]
         rest_api = self._construct_rest_api()
+        is_private_domain = isinstance(self.domain, dict) and self.domain.get("EndpointConfiguration") == "PRIVATE"
         api_domain_response = (
             self._construct_api_domain_v2(rest_api, route53_record_set_groups)
-            if isinstance(self.domain, dict) and self.domain.get("EndpointConfiguration") == "PRIVATE"
+            if is_private_domain
             else self._construct_api_domain(rest_api, route53_record_set_groups)
         )
 
+        domain_access_association = None
+        if is_private_domain:
+            domain_access_association = cast(ApiDomainResponseV2, api_domain_response).domain_access_association
+
         domain = api_domain_response.domain
         basepath_mapping = api_domain_response.apigw_basepath_mapping_list
 
@@ -882,6 +906,9 @@ def to_cloudformation(
             ]
         )
 
+        if domain_access_association is not None:
+            generated_resources.append(domain_access_association)
+
         # Make a list of single resources
         generated_resources_list: List[Resource] = []
         for resource in generated_resources:
@@ -1513,3 +1540,24 @@ def _set_endpoint_configuration(self, rest_api: ApiGatewayRestApi, value: Union[
         else:
             rest_api.EndpointConfiguration = {"Types": [value]}
             rest_api.Parameters = {"endpointConfigurationTypes": value}
+
+    def _generate_domain_access_association(
+        self,
+        domain_access_association: Dict[str, Any],
+        domain_name_arn: Dict[str, str],
+        domain_logical_id: str,
+    ) -> ApiGatewayDomainNameAccessAssociation:
+        """
+        Generate domain access association resource
+        """
+        vpcEndpointId = domain_access_association.get("VpcEndpointId")
+        logical_id = LogicalIdGenerator("DomainNameAccessAssociation", [vpcEndpointId, domain_logical_id]).gen()
+
+        domain_access_association_resource = ApiGatewayDomainNameAccessAssociation(
+            logical_id, attributes=self.passthrough_resource_attributes
+        )
+        domain_access_association_resource.DomainNameArn = domain_name_arn
+        domain_access_association_resource.AccessAssociationSourceType = "VPCE"
+        domain_access_association_resource.AccessAssociationSource = vpcEndpointId
+
+        return domain_access_association_resource