fix: handle non-dict DefinitionBody path item in _openapi_postprocess

samtranslator/model/api/api_generator.py

@@ -969,41 +969,30 @@ def _openapi_postprocess(self, definition_body):
                 del definition_body["definitions"]
             # removes `consumes` and `produces` options for CORS in openapi3 and
             # adds `schema` for the headers in responses for openapi3
-            if definition_body.get("paths"):
-                for path in definition_body.get("paths"):
-                    if definition_body.get("paths").get(path).get("options"):
-                        definition_body_options = definition_body.get("paths").get(path).get("options").copy()
-                        for field in definition_body_options.keys():
+            paths = definition_body.get("paths")
+            if paths:
+                for path, path_item in paths.items():
+                    SwaggerEditor.validate_path_item_is_dict(path_item, path)
+                    if path_item.get("options"):
+                        options = path_item.get("options").copy()
+                        for field, field_val in options.items():
                             # remove unsupported produces and consumes in options for openapi3
                             if field in ["produces", "consumes"]:
                                 del definition_body["paths"][path]["options"][field]
                             # add schema for the headers in options section for openapi3
                             if field in ["responses"]:
-                                options_path = definition_body["paths"][path]["options"]
-                                if options_path and not isinstance(options_path.get(field), dict):
-                                    raise InvalidDocumentException(
-                                        [
-                                            InvalidTemplateException(
-                                                "Value of responses in options method for path {} must be a "
-                                                "dictionary according to Swagger spec.".format(path)
-                                            )
-                                        ]
-                                    )
-                                if (
-                                    options_path
-                                    and options_path.get(field).get("200")
-                                    and options_path.get(field).get("200").get("headers")
-                                ):
-                                    headers = definition_body["paths"][path]["options"][field]["200"]["headers"]
-                                    for header in headers.keys():
-                                        header_value = {
-                                            "schema": definition_body["paths"][path]["options"][field]["200"][
-                                                "headers"
-                                            ][header]
-                                        }
+                                SwaggerEditor.validate_is_dict(
+                                    field_val,
+                                    "Value of responses in options method for path {} must be a "
+                                    "dictionary according to Swagger spec.".format(path),
+                                )
+                                if field_val.get("200") and field_val.get("200").get("headers"):
+                                    headers = field_val["200"]["headers"]
+                                    for header, header_val in headers.items():
+                                        new_header_val_with_schema = {"schema": header_val}
                                         definition_body["paths"][path]["options"][field]["200"]["headers"][
                                             header
-                                        ] = header_value
+                                        ] = new_header_val_with_schema
 
         return definition_body
 

samtranslator/swagger/swagger.py

@@ -139,16 +139,7 @@ def add_path(self, path, method=None):
         method = self._normalize_method_name(method)
 
         path_dict = self.paths.setdefault(path, {})
-
-        if not isinstance(path_dict, dict):
-            # Either customers has provided us an invalid Swagger, or this class has messed it somehow
-            raise InvalidDocumentException(
-                [
-                    InvalidTemplateException(
-                        "Value of '{}' path must be a dictionary according to Swagger spec.".format(path)
-                    )
-                ]
-            )
+        SwaggerEditor.validate_path_item_is_dict(path_dict, path)
 
         if self._CONDITIONAL_IF in path_dict:
             path_dict = path_dict[self._CONDITIONAL_IF][1]
@@ -536,15 +527,10 @@ def set_path_default_authorizer(
                 # It is possible that the method could have two definitions in a Fn::If block.
                 for method_definition in self.get_method_contents(method):
 
+                    SwaggerEditor.validate_is_dict(
+                        method_definition, "{} for path {} is not a valid dictionary.".format(method_definition, path)
+                    )
                     # If no integration given, then we don't need to process this definition (could be AWS::NoValue)
-                    if not isinstance(method_definition, dict):
-                        raise InvalidDocumentException(
-                            [
-                                InvalidTemplateException(
-                                    "{} for path {} is not a valid dictionary.".format(method_definition, path)
-                                )
-                            ]
-                        )
                     if not self.method_definition_has_integration(method_definition):
                         continue
                     existing_security = method_definition.get("security", [])
@@ -559,14 +545,9 @@ def set_path_default_authorizer(
                     # (e.g. sigv4 (AWS_IAM), api_key (API Key/Usage Plans), NONE (marker for ignoring default))
                     # We want to ensure only a single Authorizer security entry exists while keeping everything else
                     for security in existing_security:
-                        if not isinstance(security, dict):
-                            raise InvalidDocumentException(
-                                [
-                                    InvalidTemplateException(
-                                        "{} in Security for path {} is not a valid dictionary.".format(security, path)
-                                    )
-                                ]
-                            )
+                        SwaggerEditor.validate_is_dict(
+                            security, "{} in Security for path {} is not a valid dictionary.".format(security, path)
+                        )
                         if authorizer_names.isdisjoint(security.keys()):
                             existing_non_authorizer_security.append(security)
                         else:
@@ -912,8 +893,7 @@ def add_resource_policy(self, resource_policy, path, stage):
         """
         if resource_policy is None:
             return
-        if not isinstance(resource_policy, dict):
-            raise InvalidDocumentException([InvalidTemplateException("Resource Policy is not a valid dictionary.")])
+        SwaggerEditor.validate_is_dict(resource_policy, "Resource Policy is not a valid dictionary.")
 
         aws_account_whitelist = resource_policy.get("AwsAccountWhitelist")
         aws_account_blacklist = resource_policy.get("AwsAccountBlacklist")
@@ -1244,6 +1224,31 @@ def is_valid(data):
                 )
         return False
 
+    @staticmethod
+    def validate_is_dict(obj, exception_message):
+        """
+        Throws exception if obj is not a dict
+
+        :param obj: object being validated
+        :param exception_message: message to include in exception if obj is not a dict
+        """
+
+        if not isinstance(obj, dict):
+            raise InvalidDocumentException([InvalidTemplateException(exception_message)])
+
+    @staticmethod
+    def validate_path_item_is_dict(path_item, path):
+        """
+        Throws exception if path_item is not a dict
+
+        :param path_item: path_item (value at the path) being validated
+        :param path: path name
+        """
+
+        SwaggerEditor.validate_is_dict(
+            path_item, "Value of '{}' path must be a dictionary according to Swagger spec.".format(path)
+        )
+
     @staticmethod
     def gen_skeleton():
         """

tests/translator/input/error_api_no_cors_invalid_openapi_null_path.yaml

@@ -0,0 +1,12 @@
+Resources:
+  ApiWithInvalidPath:
+    Type: AWS::Serverless::Api
+    Properties:
+      StageName: Prod
+      OpenApiVersion: 3.0.1
+      DefinitionBody:
+        openapi: 3.0.1
+        info:
+          title: test invalid paths Api
+        paths:
+          /foo: null

tests/translator/input/error_api_no_cors_invalid_openapi_string_path.yaml

@@ -0,0 +1,12 @@
+Resources:
+  ApiWithInvalidPath:
+    Type: AWS::Serverless::Api
+    Properties:
+      StageName: Prod
+      OpenApiVersion: 3.0.1
+      DefinitionBody:
+        openapi: 3.0.1
+        info:
+          title: test invalid paths Api
+        paths:
+          /foo: invalid

tests/translator/output/error_api_no_cors_invalid_openapi_null_path.json

@@ -0,0 +1,3 @@
+{
+  "errorMessage": "Invalid Serverless Application Specification document. Number of errors found: 1. Structure of the SAM template is invalid. Value of '/foo' path must be a dictionary according to Swagger spec."
+}

tests/translator/output/error_api_no_cors_invalid_openapi_string_path.json

@@ -0,0 +1,3 @@
+{
+  "errorMessage": "Invalid Serverless Application Specification document. Number of errors found: 1. Structure of the SAM template is invalid. Value of '/foo' path must be a dictionary according to Swagger spec."
+}