feat: New DisableFunctionDefaultPermissions property to block the cre…

…ation of permissions resource from SAM API Auth (#2885)

samtranslator/model/api/api_generator.py

@@ -404,7 +404,7 @@ def _construct_stage(
             stage_logical_id = generator.gen()
         stage = ApiGatewayStage(stage_logical_id, attributes=self.passthrough_resource_attributes)
         stage.RestApiId = ref(self.logical_id)
-        stage.update_deployment_ref(deployment.logical_id)  # type: ignore[no-untyped-call]
+        stage.update_deployment_ref(deployment.logical_id)
         stage.StageName = self.stage_name
         stage.CacheClusterEnabled = self.cache_cluster_enabled
         stage.CacheClusterSize = self.cache_cluster_size
@@ -415,7 +415,7 @@ def _construct_stage(
         stage.TracingEnabled = self.tracing_enabled
 
         if swagger is not None:
-            deployment.make_auto_deployable(  # type: ignore[no-untyped-call]
+            deployment.make_auto_deployable(
                 stage, self.remove_extra_stage, swagger, self.domain, redeploy_restapi_parameters
             )
 
@@ -1125,6 +1125,7 @@ def _get_authorizers(self, authorizers_config, default_authorizer=None):  # type
                 function_payload_type=authorizer.get("FunctionPayloadType"),
                 function_invoke_role=authorizer.get("FunctionInvokeRole"),
                 authorization_scopes=authorizer.get("AuthorizationScopes"),
+                disable_function_default_permissions=authorizer.get("DisableFunctionDefaultPermissions"),
             )
         return authorizers
 
@@ -1140,7 +1141,7 @@ def _get_permission(self, authorizer_name, authorizer_lambda_function_arn):  # t
         partition = ArnGenerator.get_partition_name()
         resource = "${__ApiId__}/authorizers/*"
         source_arn = fnSub(
-            ArnGenerator.generate_arn(partition=partition, service="execute-api", resource=resource),  # type: ignore[no-untyped-call]
+            ArnGenerator.generate_arn(partition=partition, service="execute-api", resource=resource),
             {"__ApiId__": api_id},
         )
 
@@ -1168,7 +1169,7 @@ def _construct_authorizer_lambda_permission(self) -> List[LambdaPermission]:
 
         for authorizer_name, authorizer in authorizers.items():
             # Construct permissions for Lambda Authorizers only
-            if not authorizer.function_arn:
+            if not authorizer.function_arn or authorizer.disable_function_default_permissions:
                 continue
 
             permission = self._get_permission(authorizer_name, authorizer.function_arn)  # type: ignore[no-untyped-call]

samtranslator/model/apigateway.py

@@ -1,6 +1,6 @@
 import json
 from re import match
-from typing import Any, Dict, List, Optional
+from typing import Any, Dict, List, Optional, Union
 
 from samtranslator.model import PropertyType, Resource
 from samtranslator.model.exceptions import InvalidResourceException
@@ -65,7 +65,7 @@ class ApiGatewayStage(Resource):
 
     runtime_attrs = {"stage_name": lambda self: ref(self.logical_id)}
 
-    def update_deployment_ref(self, deployment_logical_id):  # type: ignore[no-untyped-def]
+    def update_deployment_ref(self, deployment_logical_id: str) -> None:
         self.DeploymentId = ref(deployment_logical_id)
 
 
@@ -87,13 +87,19 @@ class ApiGatewayDeployment(Resource):
 
     runtime_attrs = {"deployment_id": lambda self: ref(self.logical_id)}
 
-    def make_auto_deployable(  # type: ignore[no-untyped-def]
-        self, stage, openapi_version=None, swagger=None, domain=None, redeploy_restapi_parameters=None
-    ):
+    def make_auto_deployable(
+        self,
+        stage: ApiGatewayStage,
+        openapi_version: Optional[Union[Dict[str, Any], str]] = None,
+        swagger: Optional[Dict[str, Any]] = None,
+        domain: Optional[Dict[str, Any]] = None,
+        redeploy_restapi_parameters: Optional[Any] = None,
+    ) -> None:
         """
         Sets up the resource such that it will trigger a re-deployment when Swagger changes
         or the openapi version changes or a domain resource changes.
 
+        :param stage: The ApiGatewayStage object which will be re-deployed
         :param swagger: Dictionary containing the Swagger definition of the API
         :param openapi_version: string containing value of OpenApiVersion flag in the template
         :param domain: Dictionary containing the custom domain configuration for the API
@@ -158,7 +164,7 @@ def __init__(
     def generate_swagger(self) -> Py27Dict:
         # Applying Py27Dict here as this goes into swagger
         swagger = Py27Dict()
-        swagger["responseParameters"] = self._add_prefixes(self.response_parameters)  # type: ignore[no-untyped-call]
+        swagger["responseParameters"] = self._add_prefixes(self.response_parameters)
         swagger["responseTemplates"] = self.response_templates
 
         # Prevent "null" being written.
@@ -167,7 +173,7 @@ def generate_swagger(self) -> Py27Dict:
 
         return swagger
 
-    def _add_prefixes(self, response_parameters):  # type: ignore[no-untyped-def]
+    def _add_prefixes(self, response_parameters: Dict[str, Any]) -> Dict[str, str]:
         GATEWAY_RESPONSE_PREFIX = "gatewayresponse."
         # applying Py27Dict as this is part of swagger
         prefixed_parameters = Py27Dict()
@@ -273,6 +279,7 @@ def __init__(  # type: ignore[no-untyped-def]# noqa: too-many-arguments
         function_invoke_role=None,
         is_aws_iam_authorizer=False,
         authorization_scopes=None,
+        disable_function_default_permissions=False,
     ):
         if authorization_scopes is None:
             authorization_scopes = []
@@ -286,6 +293,7 @@ def __init__(  # type: ignore[no-untyped-def]# noqa: too-many-arguments
         self.function_invoke_role = function_invoke_role
         self.is_aws_iam_authorizer = is_aws_iam_authorizer
         self.authorization_scopes = authorization_scopes
+        self.disable_function_default_permissions = disable_function_default_permissions
 
         if function_payload_type not in ApiGatewayAuthorizer._VALID_FUNCTION_PAYLOAD_TYPES:
             raise InvalidResourceException(
@@ -300,8 +308,15 @@ def __init__(  # type: ignore[no-untyped-def]# noqa: too-many-arguments
                 "of Headers, QueryStrings, StageVariables, or Context.",
             )
 
-        if authorization_scopes is not None and not isinstance(authorization_scopes, list):
-            raise InvalidResourceException(api_logical_id, "AuthorizationScopes must be a list.")
+        if authorization_scopes is not None:
+            sam_expect(authorization_scopes, api_logical_id, f"Authorizers.{name}.AuthorizationScopes").to_be_a_list()
+
+        if disable_function_default_permissions is not None:
+            sam_expect(
+                disable_function_default_permissions,
+                api_logical_id,
+                f"Authorizers.{name}.DisableFunctionDefaultPermissions",
+            ).to_be_a_bool()
 
     def _is_missing_identity_source(self, identity: Dict[str, Any]) -> bool:
         if not identity:
@@ -349,7 +364,7 @@ def generate_swagger(self) -> Py27Dict:
             partition = ArnGenerator.get_partition_name()
             resource = "lambda:path/2015-03-31/functions/${__FunctionArn__}/invocations"
             authorizer_uri = fnSub(
-                ArnGenerator.generate_arn(  # type: ignore[no-untyped-call]
+                ArnGenerator.generate_arn(
                     partition=partition, service="apigateway", resource=resource, include_account_id=False
                 ),
                 {"__FunctionArn__": self.function_arn},

samtranslator/model/apigatewayv2.py

@@ -223,7 +223,7 @@ def generate_openapi(self) -> Dict[str, Any]:
             partition = ArnGenerator.get_partition_name()
             resource = "lambda:path/2015-03-31/functions/${__FunctionArn__}/invocations"
             authorizer_uri = fnSub(
-                ArnGenerator.generate_arn(  # type: ignore[no-untyped-call]
+                ArnGenerator.generate_arn(
                     partition=partition, service="apigateway", resource=resource, include_account_id=False
                 ),
                 {"__FunctionArn__": self.function_arn},

samtranslator/model/eventsources/cloudwatchlogs.py

@@ -45,7 +45,7 @@ def get_source_arn(self) -> Dict[str, Any]:
         partition = ArnGenerator.get_partition_name()
 
         return fnSub(
-            ArnGenerator.generate_arn(partition=partition, service="logs", resource=resource),  # type: ignore[no-untyped-call]
+            ArnGenerator.generate_arn(partition=partition, service="logs", resource=resource),
             {"__LogGroupName__": self.LogGroupName},
         )
 

samtranslator/model/eventsources/push.py

@@ -741,7 +741,7 @@ def _get_permission(self, resources_to_link, stage, suffix):  # type: ignore[no-
         resource = f"${{__ApiId__}}/${{__Stage__}}/{method}{path}"
         partition = ArnGenerator.get_partition_name()
         source_arn = fnSub(
-            ArnGenerator.generate_arn(partition=partition, service="execute-api", resource=resource),  # type: ignore[no-untyped-call]
+            ArnGenerator.generate_arn(partition=partition, service="execute-api", resource=resource),
             {"__ApiId__": api_id, "__Stage__": stage},
         )
 
@@ -1055,7 +1055,7 @@ def to_cloudformation(self, **kwargs):  # type: ignore[no-untyped-def]
 
         partition = ArnGenerator.get_partition_name()
         source_arn = fnSub(
-            ArnGenerator.generate_arn(partition=partition, service="iot", resource=resource),  # type: ignore[no-untyped-call]
+            ArnGenerator.generate_arn(partition=partition, service="iot", resource=resource),
             {"RuleName": ref(self.logical_id)},
         )
         source_account = fnSub("${AWS::AccountId}")
@@ -1304,7 +1304,7 @@ def _get_permission(self, resources_to_link, stage):  # type: ignore[no-untyped-
 
         # ApiId can be a simple string or intrinsic function like !Ref. Using Fn::Sub will handle both cases
         source_arn = fnSub(
-            ArnGenerator.generate_arn(partition="${AWS::Partition}", service="execute-api", resource=resource),  # type: ignore[no-untyped-call]
+            ArnGenerator.generate_arn(partition="${AWS::Partition}", service="execute-api", resource=resource),
             {"__ApiId__": api_id, "__Stage__": stage},
         )
 

samtranslator/model/exceptions.py

@@ -9,6 +9,7 @@ class ExpectedType(Enum):
     LIST = ("list", list)
     STRING = ("string", str)
     INTEGER = ("integer", int)
+    BOOLEAN = ("boolean", bool)
 
 
 class ExceptionWithMessage(ABC, Exception):

samtranslator/schema/schema.json

@@ -188488,6 +188488,10 @@
           "title": "AuthorizationScopes",
           "type": "array"
         },
+        "DisableFunctionDefaultPermissions": {
+          "title": "Disablefunctiondefaultpermissions",
+          "type": "boolean"
+        },
         "FunctionArn": {
           "anyOf": [
             {
@@ -188601,6 +188605,10 @@
           "title": "AuthorizationScopes",
           "type": "array"
         },
+        "DisableFunctionDefaultPermissions": {
+          "title": "Disablefunctiondefaultpermissions",
+          "type": "boolean"
+        },
         "FunctionArn": {
           "anyOf": [
             {

samtranslator/translator/arn_generator.py

@@ -33,7 +33,9 @@ class ArnGenerator:
     BOTO_SESSION_REGION_NAME = None
 
     @classmethod
-    def generate_arn(cls, partition, service, resource, include_account_id=True):  # type: ignore[no-untyped-def]
+    def generate_arn(
+        cls, partition: str, service: str, resource: str, include_account_id: Optional[bool] = True
+    ) -> str:
         if not service or not resource:
             raise RuntimeError("Could not construct ARN for resource.")
 

samtranslator/validator/value_validator.py

@@ -120,5 +120,12 @@ def to_be_an_integer(self, message: Optional[str] = "") -> int:
         """
         return cast(int, self.to_be_a(ExpectedType.INTEGER, message))
 
+    def to_be_a_bool(self, message: Optional[str] = "") -> bool:
+        """
+        Return the value with type hint "bool".
+        Raise InvalidResourceException/InvalidEventException if the value is not.
+        """
+        return cast(bool, self.to_be_a(ExpectedType.BOOLEAN, message))
+
 
 sam_expect = _ResourcePropertyValueValidator

schema_source/aws_serverless_api.py

@@ -77,6 +77,7 @@ class LambdaTokenAuthorizer(BaseModel):
     FunctionInvokeRole: Optional[str] = lambdatokenauthorizer("FunctionInvokeRole")
     FunctionPayloadType: Optional[Literal["TOKEN"]] = lambdatokenauthorizer("FunctionPayloadType")
     Identity: Optional[LambdaTokenAuthorizerIdentity] = lambdatokenauthorizer("Identity")
+    DisableFunctionDefaultPermissions: Optional[bool]  # TODO Add docs
 
 
 class LambdaRequestAuthorizer(BaseModel):
@@ -85,6 +86,7 @@ class LambdaRequestAuthorizer(BaseModel):
     FunctionInvokeRole: Optional[str] = lambdarequestauthorizer("FunctionInvokeRole")
     FunctionPayloadType: Optional[Literal["REQUEST"]] = lambdarequestauthorizer("FunctionPayloadType")
     Identity: Optional[LambdaRequestAuthorizerIdentity] = lambdarequestauthorizer("Identity")
+    DisableFunctionDefaultPermissions: Optional[bool]  # TODO Add docs
 
 
 class UsagePlan(BaseModel):

schema_source/sam.schema.json

@@ -1593,6 +1593,10 @@
           "title": "AuthorizationScopes",
           "type": "array"
         },
+        "DisableFunctionDefaultPermissions": {
+          "title": "Disablefunctiondefaultpermissions",
+          "type": "boolean"
+        },
         "FunctionArn": {
           "anyOf": [
             {
@@ -1706,6 +1710,10 @@
           "title": "AuthorizationScopes",
           "type": "array"
         },
+        "DisableFunctionDefaultPermissions": {
+          "title": "Disablefunctiondefaultpermissions",
+          "type": "boolean"
+        },
         "FunctionArn": {
           "anyOf": [
             {

tests/translator/input/api_rest_auth_disable_default_function_permissions.yaml

@@ -0,0 +1,82 @@
+Resources:
+  MyApiWithNoDisableFunctionDefaultPermissions:
+    Type: AWS::Serverless::Api
+    Properties:
+      StageName: Prod
+      Auth:
+        DefaultAuthorizer: Auth
+        Authorizers:
+          Auth:
+            FunctionPayloadType: REQUEST
+            FunctionArn: !GetAtt MyAuthFn.Arn
+            FunctionInvokeRole: arn:{AWS::Partition}:iam::123456789012:role/test-role
+            Identity:
+              Headers:
+              - Authorization1
+
+  MyApiWithLambdaTokenAuth:
+    Type: AWS::Serverless::Api
+    Properties:
+      StageName: Prod
+      Auth:
+        DefaultAuthorizer: LambdaTokenAuth
+        Authorizers:
+          LambdaTokenAuth:
+            FunctionPayloadType: REQUEST
+            FunctionArn: !GetAtt MyAuthFn.Arn
+            FunctionInvokeRole: arn:{AWS::Partition}:iam::123456789012:role/test-role
+            DisableFunctionDefaultPermissions: true
+            Identity:
+              Headers:
+              - Authorization1
+
+  MyApiWithLambdaRequestAuth:
+    Type: AWS::Serverless::Api
+    Properties:
+      StageName: Prod
+      Auth:
+        DefaultAuthorizer: LambdaRequestAuth
+        Authorizers:
+          LambdaRequestAuth:
+            FunctionPayloadType: REQUEST
+            FunctionArn: !GetAtt MyAuthFn.Arn
+            FunctionInvokeRole: arn:{AWS::Partition}:iam::123456789012:role/test-role
+            DisableFunctionDefaultPermissions: true
+            Identity:
+              Headers:
+              - Authorization1
+
+  MyApiFunctionEvent:
+    Type: AWS::Serverless::Api
+    Properties:
+      StageName: Prod
+      Auth:
+        DefaultAuthorizer: Auth
+        Authorizers:
+          Auth:
+            FunctionPayloadType: REQUEST
+            FunctionArn: !GetAtt MyAuthFn.Arn
+            DisableFunctionDefaultPermissions: true
+            Identity:
+              Headers:
+              - Authorization
+
+  MyAuthFn:
+    Type: AWS::Serverless::Function
+    Properties:
+      CodeUri: s3://bucket/key
+      Handler: index.handler
+      Runtime: nodejs12.x
+  MyFn:
+    Type: AWS::Serverless::Function
+    Properties:
+      CodeUri: s3://bucket/key
+      Handler: index.handler
+      Runtime: nodejs12.x
+      Events:
+        LambdaEvent:
+          Type: Api
+          Properties:
+            RestApiId: !Ref MyApiFunctionEvent
+            Method: get
+            Path: /foo

tests/translator/input/error_api_invalid_auth.yaml

@@ -250,3 +250,13 @@ Resources:
       Auth:
         Authorizers:
           MyAuth: AWS_IAM # It should be a dict
+
+  AuthorizerWithBadDisableFunctionDefaultPermissionsType:
+    Type: AWS::Serverless::Api
+    Properties:
+      StageName: Prod
+      Auth:
+        Authorizers:
+          MyAuth:
+            FunctionArn: !GetAtt MyAuthFn.Arn
+            DisableFunctionDefaultPermissions: foo