fix: Extend validation of apiauth Identity sub values from Headers only

samtranslator/model/apigatewayv2.py

@@ -3,7 +3,7 @@
 from samtranslator.model import PropertyType, Resource
 from samtranslator.model.types import is_type, one_of, is_str, list_of
 from samtranslator.model.intrinsics import ref, fnSub
-from samtranslator.model.exceptions import InvalidResourceException
+from samtranslator.model.exceptions import ExpectedType, InvalidResourceException
 from samtranslator.translator.arn_generator import ArnGenerator
 from samtranslator.utils.types import Intrinsicable
 from samtranslator.validator.value_validator import sam_expect
@@ -186,21 +186,12 @@ def _validate_lambda_authorizer(self):  # type: ignore[no-untyped-def]
                 self.api_logical_id, f"{self.name} Lambda Authorizer must define 'AuthorizerPayloadFormatVersion'."
             )
 
-        if self.identity:
-            sam_expect(self.identity, self.api_logical_id, f"Authorizer.{self.name}.Identity").to_be_a_map()
-            headers = self.identity.get("Headers")
-            if headers:
-                sam_expect(headers, self.api_logical_id, f"Authorizer.{self.name}.Identity.Headers").to_be_a_list()
-                for index, header in enumerate(headers):
-                    sam_expect(
-                        header, self.api_logical_id, f"Authorizer.{self.name}.Identity.Headers[{index}]"
-                    ).to_be_a_string()
-
     def generate_openapi(self) -> Dict[str, Any]:
         """
         Generates OAS for the securitySchemes section
         """
         authorizer_type = self._get_auth_type()  # type: ignore[no-untyped-call]
+        openapi: Dict[str, Any]
 
         if authorizer_type == "AWS_IAM":
             openapi = {
@@ -210,21 +201,23 @@ def generate_openapi(self) -> Dict[str, Any]:
                 "x-amazon-apigateway-authtype": "awsSigv4",
             }
 
-        if authorizer_type == "JWT":
-            openapi = {"type": "oauth2"}
-            openapi[APIGATEWAY_AUTHORIZER_KEY] = {  # type: ignore[assignment]
-                "jwtConfiguration": self.jwt_configuration,
-                "identitySource": self.id_source,
-                "type": "jwt",
+        elif authorizer_type == "JWT":
+            openapi = {
+                "type": "oauth2",
+                APIGATEWAY_AUTHORIZER_KEY: {
+                    "jwtConfiguration": self.jwt_configuration,
+                    "identitySource": self.id_source,
+                    "type": "jwt",
+                },
             }
 
-        if authorizer_type == "REQUEST":
+        elif authorizer_type == "REQUEST":
             openapi = {
                 "type": "apiKey",
                 "name": "Unused",
                 "in": "header",
+                APIGATEWAY_AUTHORIZER_KEY: {"type": "request"},
             }
-            openapi[APIGATEWAY_AUTHORIZER_KEY] = {"type": "request"}  # type: ignore[assignment]
 
             # Generate the lambda arn
             partition = ArnGenerator.get_partition_name()
@@ -235,31 +228,35 @@ def generate_openapi(self) -> Dict[str, Any]:
                 ),
                 {"__FunctionArn__": self.function_arn},
             )
-            openapi[APIGATEWAY_AUTHORIZER_KEY]["authorizerUri"] = authorizer_uri  # type: ignore[index]
+            openapi[APIGATEWAY_AUTHORIZER_KEY]["authorizerUri"] = authorizer_uri
 
             # Set authorizerCredentials if present
             function_invoke_role = self._get_function_invoke_role()  # type: ignore[no-untyped-call]
             if function_invoke_role:
-                openapi[APIGATEWAY_AUTHORIZER_KEY]["authorizerCredentials"] = function_invoke_role  # type: ignore[index]
-
-            # Set authorizerResultTtlInSeconds if present
-            reauthorize_every = self._get_reauthorize_every()  # type: ignore[no-untyped-call]
-            if reauthorize_every is not None:
-                openapi[APIGATEWAY_AUTHORIZER_KEY]["authorizerResultTtlInSeconds"] = reauthorize_every  # type: ignore[index]
+                openapi[APIGATEWAY_AUTHORIZER_KEY]["authorizerCredentials"] = function_invoke_role
 
             # Set identitySource if present
             if self.identity:
-                openapi[APIGATEWAY_AUTHORIZER_KEY]["identitySource"] = self._get_identity_source()  # type: ignore[no-untyped-call, index]
+                sam_expect(self.identity, self.api_logical_id, f"Auth.Authorizers.{self.name}.Identity").to_be_a_map()
+                # Set authorizerResultTtlInSeconds if present
+                reauthorize_every = self.identity.get("ReauthorizeEvery")
+                if reauthorize_every is not None:
+                    openapi[APIGATEWAY_AUTHORIZER_KEY]["authorizerResultTtlInSeconds"] = reauthorize_every
+
+                # Set identitySource if present
+                openapi[APIGATEWAY_AUTHORIZER_KEY]["identitySource"] = self._get_identity_source(self.identity)
 
             # Set authorizerPayloadFormatVersion. It's a required parameter
-            openapi[APIGATEWAY_AUTHORIZER_KEY][  # type: ignore[index]
+            openapi[APIGATEWAY_AUTHORIZER_KEY][
                 "authorizerPayloadFormatVersion"
             ] = self.authorizer_payload_format_version
 
-            # Set authorizerPayloadFormatVersion. It's a required parameter
+            # Set enableSimpleResponses if present
             if self.enable_simple_responses:
-                openapi[APIGATEWAY_AUTHORIZER_KEY]["enableSimpleResponses"] = self.enable_simple_responses  # type: ignore[index]
+                openapi[APIGATEWAY_AUTHORIZER_KEY]["enableSimpleResponses"] = self.enable_simple_responses
 
+        else:
+            raise ValueError(f"Unexpected authorizer_type: {authorizer_type}")
         return openapi
 
     def _get_function_invoke_role(self):  # type: ignore[no-untyped-def]
@@ -268,43 +265,37 @@ def _get_function_invoke_role(self):  # type: ignore[no-untyped-def]
 
         return self.function_invoke_role
 
-    def _get_identity_source(self):  # type: ignore[no-untyped-def]
-        identity_source_headers = []
-        identity_source_query_strings = []
-        identity_source_stage_variables = []
-        identity_source_context = []
-
-        if self.identity.get("Headers"):
-            identity_source_headers = list(map(lambda h: "$request.header." + h, self.identity.get("Headers")))  # type: ignore[no-any-return]
-
-        if self.identity.get("QueryStrings"):
-            identity_source_query_strings = list(
-                map(lambda qs: "$request.querystring." + qs, self.identity.get("QueryStrings"))  # type: ignore[no-any-return]
-            )
-
-        if self.identity.get("StageVariables"):
-            identity_source_stage_variables = list(
-                map(lambda sv: "$stageVariables." + sv, self.identity.get("StageVariables"))  # type: ignore[no-any-return]
-            )
-
-        if self.identity.get("Context"):
-            identity_source_context = list(map(lambda c: "$context." + c, self.identity.get("Context")))  # type: ignore[no-any-return]
-
-        identity_source = (
-            identity_source_headers
-            + identity_source_query_strings
-            + identity_source_stage_variables
-            + identity_source_context
-        )
+    def _get_identity_source(self, auth_identity: Dict[str, Any]) -> List[str]:
+        """
+        Generate the list of identitySource using authorizer's Identity config by flatting them.
+        For the format of identitySource, see:
+        https://docs.aws.amazon.com/apigateway/latest/developerguide/api-gateway-swagger-extensions-authorizer.html
+
+        It will add API GW prefix to each item:
+        - prefix "$request.header." to all values in "Headers"
+        - prefix "$request.querystring." to all values in "QueryStrings"
+        - prefix "$stageVariables." to all values in "StageVariables"
+        - prefix "$context." to all values in "Context"
+        """
+        identity_source: List[str] = []
+
+        identity_property_path = f"Authorizers.{self.name}.Identity"
+
+        for prefix, property_name in [
+            ("$request.header.", "Headers"),
+            ("$request.querystring.", "QueryStrings"),
+            ("$stageVariables.", "StageVariables"),
+            ("$context.", "Context"),
+        ]:
+            property_values = auth_identity.get(property_name)
+            if property_values:
+                sam_expect(
+                    property_values, self.api_logical_id, f"{identity_property_path}.{property_name}"
+                ).to_be_a_list_of(ExpectedType.STRING)
+                identity_source += [prefix + value for value in property_values]
 
         return identity_source
 
-    def _get_reauthorize_every(self):  # type: ignore[no-untyped-def]
-        if not self.identity:
-            return None
-
-        return self.identity.get("ReauthorizeEvery")
-
     @staticmethod
     def _get_jwt_configuration(props: Optional[Dict[str, Union[str, List[str]]]]) -> Optional[JwtConfiguration]:
         """Make sure that JWT configuration dict keys are lower case.

samtranslator/validator/value_validator.py

@@ -13,20 +13,25 @@
 
 class _ResourcePropertyValueValidator(Generic[T]):
     value: Optional[T]
-    resource_logical_id: Optional[str]
-    event_id: Optional[str]
+    resource_id: str
     property_identifier: str
+    is_sam_event: bool
 
     def __init__(
         self, value: Optional[T], resource_id: str, property_identifier: str, is_sam_event: bool = False
     ) -> None:
         self.value = value
+        self.resource_id = resource_id
         self.property_identifier = property_identifier
-        self.resource_logical_id, self.event_id = (None, None)
-        if is_sam_event:
-            self.event_id = resource_id
-        else:
-            self.resource_logical_id = resource_id
+        self.is_sam_event = is_sam_event
+
+    @property
+    def resource_logical_id(self) -> Optional[str]:
+        return None if self.is_sam_event else self.resource_id
+
+    @property
+    def event_id(self) -> Optional[str]:
+        return self.resource_id if self.is_sam_event else None
 
     def to_be_a(self, expected_type: ExpectedType, message: Optional[str] = "") -> T:
         """
@@ -73,6 +78,14 @@ def to_be_a_map(self, message: Optional[str] = "") -> T:
     def to_be_a_list(self, message: Optional[str] = "") -> T:
         return self.to_be_a(ExpectedType.LIST, message)
 
+    def to_be_a_list_of(self, expected_type: ExpectedType, message: Optional[str] = "") -> T:
+        value = self.to_be_a(ExpectedType.LIST, message)
+        for index, item in enumerate(value):  # type: ignore
+            sam_expect(
+                item, self.resource_id, f"{self.property_identifier}[{index}]", is_sam_event=self.is_sam_event
+            ).to_be_a(expected_type, message)
+        return value
+
     def to_be_a_string(self, message: Optional[str] = "") -> T:
         return self.to_be_a(ExpectedType.STRING, message)
 

tests/translator/input/error_api_authorizer_property_indentity_header_with_invalid_type.yaml

@@ -54,3 +54,23 @@ Resources:
               - Ref: AuthKeyName
             AuthorizerPayloadFormatVersion: 1.0
         DefaultAuthorizer: MyLambdaAuthUpdated
+
+  MyApi2:
+    Type: AWS::Serverless::HttpApi
+    Properties:
+      Auth:
+        Authorizers:
+          MyLambdaAuthUpdated:
+            FunctionArn:
+              Fn::GetAtt:
+              - MyAuthFn
+              - Arn
+            FunctionInvokeRole:
+              Fn::GetAtt:
+              - MyAuthFnRole
+              - Arn
+            Identity:
+              QueryStrings:
+                This: should be a list
+            AuthorizerPayloadFormatVersion: 1.0
+        DefaultAuthorizer: MyLambdaAuthUpdated

tests/translator/output/error_api_authorizer_property_indentity_header_with_invalid_type.json

@@ -1,3 +1,3 @@
 {
-  "errorMessage": "Invalid Serverless Application Specification document. Number of errors found: 1. Resource with id [MyApi] is invalid. Property 'Authorizer.MyLambdaAuthUpdated.Identity.Headers[0]' should be a string."
+  "errorMessage": "Invalid Serverless Application Specification document. Number of errors found: 2. Resource with id [MyApi] is invalid. Property 'Authorizers.MyLambdaAuthUpdated.Identity.Headers[0]' should be a string. Resource with id [MyApi2] is invalid. Property 'Authorizers.MyLambdaAuthUpdated.Identity.QueryStrings' should be a list."
 }

tests/translator/output/error_api_authorizer_property_indentity_with_invalid_type.json

@@ -1,3 +1,3 @@
 {
-  "errorMessage": "Invalid Serverless Application Specification document. Number of errors found: 4. Resource with id [MyApi] is invalid. Property 'Authorizer.MyLambdaAuthUpdated.Identity' should be a map. Resource with id [MyRestApi] is invalid. Property 'Authorizer.LambdaRequestIdentityNotObject.Identity' should be a map. Resource with id [MyRestApiInvalidHeadersItemType] is invalid. Property 'Auth.Authorizers.LambdaRequestIdentityNotObject.Identity.Headers[1]' should be a string. Resource with id [MyRestApiInvalidHeadersType] is invalid. Property 'Auth.Authorizers.LambdaRequestIdentityNotObject.Identity.Headers' should be a list."
+  "errorMessage": "Invalid Serverless Application Specification document. Number of errors found: 4. Resource with id [MyApi] is invalid. Property 'Auth.Authorizers.MyLambdaAuthUpdated.Identity' should be a map. Resource with id [MyRestApi] is invalid. Property 'Authorizer.LambdaRequestIdentityNotObject.Identity' should be a map. Resource with id [MyRestApiInvalidHeadersItemType] is invalid. Property 'Auth.Authorizers.LambdaRequestIdentityNotObject.Identity.Headers[1]' should be a string. Resource with id [MyRestApiInvalidHeadersType] is invalid. Property 'Auth.Authorizers.LambdaRequestIdentityNotObject.Identity.Headers' should be a list."
 }