fix: KeyError: u'AWS_IAM' and add typing

samtranslator/model/api/api_generator.py

@@ -1,6 +1,6 @@
 import logging
 from collections import namedtuple
-from typing import List, Optional, Set
+from typing import List, Optional, Set, Dict
 
 from samtranslator.metrics.method_decorator import cw_timer
 from samtranslator.model.intrinsics import ref, fnGetAtt, make_or_condition
@@ -661,7 +661,7 @@ def _add_cors(self):  # type: ignore[no-untyped-def]
             )
 
         editor = SwaggerEditor(self.definition_body)  # type: ignore[no-untyped-call]
-        for path in editor.iter_on_path():  # type: ignore[no-untyped-call]
+        for path in editor.iter_on_path():
             try:
                 editor.add_cors(  # type: ignore[no-untyped-call]
                     path,
@@ -724,12 +724,11 @@ def _add_auth(self):  # type: ignore[no-untyped-def]
 
         if authorizers:
             swagger_editor.add_authorizers_security_definitions(authorizers)  # type: ignore[no-untyped-call]
-            self._set_default_authorizer(  # type: ignore[no-untyped-call]
+            self._set_default_authorizer(
                 swagger_editor,
                 authorizers,
                 auth_properties.DefaultAuthorizer,
                 auth_properties.AddDefaultAuthorizerToCorsPreflight,
-                auth_properties.Authorizers,
             )
 
         if auth_properties.ApiKeyRequired:
@@ -740,7 +739,7 @@ def _add_auth(self):  # type: ignore[no-untyped-def]
             SwaggerEditor.validate_is_dict(
                 auth_properties.ResourcePolicy, "ResourcePolicy must be a map (ResourcePolicyStatement)."
             )
-            for path in swagger_editor.iter_on_path():  # type: ignore[no-untyped-call]
+            for path in swagger_editor.iter_on_path():
                 swagger_editor.add_resource_policy(auth_properties.ResourcePolicy, path, self.stage_name)  # type: ignore[no-untyped-call]
             if auth_properties.ResourcePolicy.get("CustomStatements"):
                 swagger_editor.add_custom_statements(auth_properties.ResourcePolicy.get("CustomStatements"))  # type: ignore[no-untyped-call]
@@ -1151,9 +1150,13 @@ def _construct_authorizer_lambda_permission(self):  # type: ignore[no-untyped-de
 
         return permissions
 
-    def _set_default_authorizer(  # type: ignore[no-untyped-def]
-        self, swagger_editor, authorizers, default_authorizer, add_default_auth_to_preflight=True, api_authorizers=None
-    ):
+    def _set_default_authorizer(
+        self,
+        swagger_editor: SwaggerEditor,
+        authorizers: Dict[str, ApiGatewayAuthorizer],
+        default_authorizer: str,
+        add_default_auth_to_preflight: bool = True,
+    ) -> None:
         if not default_authorizer:
             return
 
@@ -1177,7 +1180,6 @@ def _set_default_authorizer(  # type: ignore[no-untyped-def]
                 default_authorizer,
                 authorizers=authorizers,
                 add_default_auth_to_preflight=add_default_auth_to_preflight,
-                api_authorizers=api_authorizers,
             )
 
     def _set_default_apikey_required(self, swagger_editor):  # type: ignore[no-untyped-def]

samtranslator/model/api/http_api_generator.py

@@ -470,9 +470,7 @@ def _add_auth(self) -> None:
 
         # authorizers is guaranteed to return a value or raise an exception
         open_api_editor.add_authorizers_security_definitions(authorizers)
-        self._set_default_authorizer(
-            open_api_editor, authorizers, auth_properties.DefaultAuthorizer, auth_properties.Authorizers
-        )
+        self._set_default_authorizer(open_api_editor, authorizers, auth_properties.DefaultAuthorizer)
         self.definition_body = open_api_editor.openapi
 
     def _add_tags(self) -> None:
@@ -510,7 +508,6 @@ def _set_default_authorizer(
         open_api_editor: OpenApiEditor,
         authorizers: Dict[str, ApiGatewayV2Authorizer],
         default_authorizer: str,
-        api_authorizers: Dict[str, Any],
     ) -> None:
         """
         Sets the default authorizer if one is given in the template
@@ -540,7 +537,7 @@ def _set_default_authorizer(
             )
 
         for path in open_api_editor.iter_on_path():
-            open_api_editor.set_path_default_authorizer(path, default_authorizer, authorizers, api_authorizers)
+            open_api_editor.set_path_default_authorizer(path, default_authorizer, authorizers)
 
     def _get_authorizers(
         self, authorizers_config: Any, enable_iam_authorizer: bool = False

samtranslator/open_api/open_api.py

@@ -1,6 +1,6 @@
 import copy
 import re
-from typing import Any, Dict, Iterator, List, Optional
+from typing import Any, Dict, Iterator, Optional
 
 from samtranslator.model.apigatewayv2 import ApiGatewayV2Authorizer
 from samtranslator.model.intrinsics import ref, make_conditional, is_intrinsic, is_intrinsic_no_value
@@ -371,7 +371,6 @@ def set_path_default_authorizer(
         path: str,
         default_authorizer: str,
         authorizers: Dict[str, ApiGatewayV2Authorizer],
-        api_authorizers: Dict[str, Any],
     ) -> None:
         """
         Adds the default_authorizer to the security block for each method on this path unless an Authorizer
@@ -417,12 +416,10 @@ def set_path_default_authorizer(
                         existing_security = method_definition.get("security", [])
                         if existing_security:
                             continue
-                        authorizer_list: List[str] = []
-                        if authorizers:
-                            authorizer_list.extend(authorizers.keys())
+
                         security_dict = {}
-                        security_dict[default_authorizer] = self._get_authorization_scopes(  # type: ignore[no-untyped-call]
-                            api_authorizers, default_authorizer
+                        security_dict[default_authorizer] = self._get_authorization_scopes(
+                            authorizers, default_authorizer
                         )
                         authorizer_security = [security_dict]
 
@@ -675,18 +672,16 @@ def gen_skeleton() -> Py27Dict:
         return skeleton
 
     @staticmethod
-    def _get_authorization_scopes(authorizers, default_authorizer):  # type: ignore[no-untyped-def]
+    def _get_authorization_scopes(authorizers: Dict[str, ApiGatewayV2Authorizer], default_authorizer: str) -> Any:
         """
         Returns auth scopes for an authorizer if present
         :param authorizers: authorizer definitions
         :param default_authorizer: name of the default authorizer
         """
-        if authorizers is not None:
-            if (
-                authorizers[default_authorizer]
-                and authorizers[default_authorizer].get("AuthorizationScopes") is not None
-            ):
-                return authorizers[default_authorizer].get("AuthorizationScopes")
+        authorizer = authorizers[default_authorizer]
+        if authorizer and authorizer.authorization_scopes is not None:
+            return authorizer.authorization_scopes
+
         return []
 
     @staticmethod

samtranslator/swagger/swagger.py

@@ -1,7 +1,8 @@
 import copy
 import re
-from typing import Dict, Any
+from typing import Dict, Any, Iterator
 
+from samtranslator.model.apigateway import ApiGatewayAuthorizer
 from samtranslator.model.intrinsics import ref, make_conditional, fnSub, is_intrinsic_no_value
 from samtranslator.model.exceptions import InvalidDocumentException, InvalidTemplateException
 from samtranslator.utils.py27hash_fix import Py27Dict, Py27UniStr
@@ -64,7 +65,7 @@ def __init__(self, doc):  # type: ignore[no-untyped-def]
         # each path item object must be a dict (even it is empty).
         # We can do an early path validation on path item objects,
         # so we don't need to validate wherever we use them.
-        for path in self.iter_on_path():  # type: ignore[no-untyped-call]
+        for path in self.iter_on_path():
             for path_item in self.get_conditional_contents(self.paths.get(path)):  # type: ignore[no-untyped-call]
                 SwaggerEditor.validate_path_item_is_dict(path_item, path)  # type: ignore[no-untyped-call]
 
@@ -326,7 +327,7 @@ def _get_invoke_role(self, invoke_role):  # type: ignore[no-untyped-def]
         CALLER_CREDENTIALS_ARN = "arn:aws:iam::*:user/*"
         return invoke_role if invoke_role and invoke_role != "CALLER_CREDENTIALS" else CALLER_CREDENTIALS_ARN
 
-    def iter_on_path(self):  # type: ignore[no-untyped-def]
+    def iter_on_path(self) -> Iterator[str]:
         """
         Yields all the paths available in the Swagger. As a caller, if you add new paths to Swagger while iterating,
         they will not show up in this iterator
@@ -632,9 +633,13 @@ def add_apikey_security_definition(self):  # type: ignore[no-untyped-def]
         if "api_key" not in self.security_definitions:
             self.security_definitions.update(api_key_security_definition)
 
-    def set_path_default_authorizer(  # type: ignore[no-untyped-def]
-        self, path, default_authorizer, authorizers, add_default_auth_to_preflight=True, api_authorizers=None
-    ):
+    def set_path_default_authorizer(
+        self,
+        path: str,
+        default_authorizer: str,
+        authorizers: Dict[str, ApiGatewayAuthorizer],
+        add_default_auth_to_preflight: bool = True,
+    ) -> None:
         """
         Adds the default_authorizer to the security block for each method on this path unless an Authorizer
         was defined at the Function/Path/Method level. This is intended to be used to set the
@@ -701,7 +706,7 @@ def set_path_default_authorizer(  # type: ignore[no-untyped-def]
             # No existing Authorizer found; use default
             else:
                 security_dict = Py27Dict()
-                security_dict[default_authorizer] = self._get_authorization_scopes(api_authorizers, default_authorizer)  # type: ignore[no-untyped-call]
+                security_dict[default_authorizer] = self._get_authorization_scopes(authorizers, default_authorizer)
                 authorizer_security = [security_dict]
 
             security = existing_non_authorizer_security + authorizer_security
@@ -1374,18 +1379,15 @@ def gen_skeleton() -> Py27Dict:
         return skeleton
 
     @staticmethod
-    def _get_authorization_scopes(authorizers, default_authorizer):  # type: ignore[no-untyped-def]
+    def _get_authorization_scopes(authorizers: Dict[str, ApiGatewayAuthorizer], default_authorizer: str) -> Any:
         """
         Returns auth scopes for an authorizer if present
         :param authorizers: authorizer definitions
         :param default_authorizer: name of the default authorizer
         """
-        if authorizers is not None:
-            if (
-                authorizers.get(default_authorizer)
-                and authorizers[default_authorizer].get("AuthorizationScopes") is not None
-            ):
-                return authorizers[default_authorizer].get("AuthorizationScopes")
+        authorizer = authorizers.get(default_authorizer)
+        if authorizer and authorizer.authorization_scopes is not None:
+            return authorizer.authorization_scopes
         return []
 
     @staticmethod

tests/translator/input/api_http_with_default_iam_authorizer.yaml

@@ -0,0 +1,101 @@
+Resources:
+  HelloWorldFunction:
+    Type: AWS::Serverless::Function
+    Properties:
+      CodeUri: s3://bucket/key
+      Handler: app.lambda_handler
+      Runtime: python3.8
+      Role:
+        Fn::GetAtt:
+        - HelloWorldFunctionRole
+        - Arn
+      Architectures:
+      - x86_64
+      Events:
+        HelloWorld:
+          Type: HttpApi
+          Properties:
+            ApiId:
+              Ref: HttpApi
+            Path: /{proxy+}
+            Method: ANY
+        Preflight:
+          Type: HttpApi
+          Properties:
+            ApiId:
+              Ref: HttpApi
+            Path: /{proxy+}
+            Method: OPTIONS
+  HelloWorldFunctionRole:
+    Type: AWS::IAM::Role
+    Properties:
+      AssumeRolePolicyDocument:
+        Version: '2012-10-17'
+        Statement:
+        - Effect: Allow
+          Principal:
+            Service: lambda.amazonaws.com
+          Action: sts:AssumeRole
+      Policies:
+      - PolicyName: HelloWorldFunctionPolicy
+        PolicyDocument:
+          Version: '2012-10-17'
+          Statement:
+          - Effect: Allow
+            Action:
+            - logs:CreateLogGroup
+            - logs:CreateLogStream
+            - logs:PutLogEvents
+            - cognito-idp:List*
+            - cognito-idp:AdminListGroupsForUser
+            - sts:AssumeRole
+            Resource: '*'
+    Metadata:
+      SamResourceId: HelloWorldFunctionRole
+
+  HttpApi:
+    Type: AWS::Serverless::HttpApi
+    Properties:
+      StageName:
+        Fn::Sub: ${StageName}
+      DefinitionBody:
+        openapi: 3.0.1
+        info:
+          title:
+            Ref: AWS::StackName
+        paths: {}
+      CorsConfiguration:
+        AllowOrigins:
+        - '*'
+        AllowCredentials: false
+        AllowHeaders:
+        - Content-Type
+        - X-CSRF-TOKEN
+        - X-Amz-Date
+        - Authorization
+        - X-Requested-With
+        - X-Requested-By
+        - X-Api-Key
+        - X-Forwarded-For
+        - X-Amz-Security-Token
+        AllowMethods:
+        - '*'
+      Auth:
+        EnableIamAuthorizer: true
+        DefaultAuthorizer: AWS_IAM
+        Authorizers:
+          MyAuthorizer:
+            IdentitySource: $request.header.Authorization
+            JwtConfiguration:
+              audience:
+              - Ref: UserPoolClient
+              - Ref: UserPoolClientApp
+              issuer:
+                Fn::Join:
+                - ''
+                - - https://cognito-idp.
+                  - Fn::Sub: ${AWS::Region}
+                  - .amazonaws.com/
+                  - Ref: UserPool
+    Metadata:
+      SamResourceId: HttpApi