fix: Raise correct exception when apigateway-endpoint-configuration is not a map (#2753)

Co-authored-by: Chris Rehn <[email protected]>

Author: aahung

samtranslator/model/api/api_generator.py

@@ -324,7 +324,7 @@ def _add_endpoint_extension(self) -> None:
                 self.logical_id, "DisableExecuteApiEndpoint works only within 'DefinitionBody' property."
             )
         editor = SwaggerEditor(self.definition_body)
-        editor.add_disable_execute_api_endpoint_extension(self.disable_execute_api_endpoint)  # type: ignore[no-untyped-call]
+        editor.add_disable_execute_api_endpoint_extension(self.disable_execute_api_endpoint)
         self.definition_body = editor.swagger
 
     def _construct_body_s3_dict(self) -> Dict[str, Any]:

samtranslator/model/apigateway.py

@@ -4,8 +4,8 @@
 
 from samtranslator.model import PropertyType, Resource
 from samtranslator.model.exceptions import InvalidResourceException
-from samtranslator.model.types import IS_DICT, is_type, one_of, IS_STR, list_of
-from samtranslator.model.intrinsics import ref, fnSub
+from samtranslator.model.intrinsics import fnSub, ref
+from samtranslator.model.types import IS_DICT, IS_STR, is_type, list_of, one_of
 from samtranslator.schema.common import PassThrough
 from samtranslator.translator import logical_id_generator
 from samtranslator.translator.arn_generator import ArnGenerator
@@ -32,7 +32,16 @@ class ApiGatewayRestApi(Resource):
 
     Body: Optional[Dict[str, Any]]
     BodyS3Location: Optional[Dict[str, Any]]
+    CloneFrom: Optional[PassThrough]
+    Description: Optional[PassThrough]
+    FailOnWarnings: Optional[PassThrough]
+    Name: Optional[PassThrough]
+    Parameters: Optional[Dict[str, Any]]
     EndpointConfiguration: Optional[Dict[str, Any]]
+    BinaryMediaTypes: Optional[List[Any]]
+    MinimumCompressionSize: Optional[PassThrough]
+    Mode: Optional[PassThrough]
+    ApiKeySourceType: Optional[PassThrough]
 
     runtime_attrs = {"rest_api_id": lambda self: ref(self.logical_id)}
 

samtranslator/swagger/swagger.py

@@ -6,8 +6,10 @@
 from samtranslator.model.intrinsics import ref, make_conditional, fnSub
 from samtranslator.model.exceptions import InvalidDocumentException, InvalidTemplateException
 from samtranslator.open_api.base_editor import BaseEditor
+from samtranslator.schema.common import PassThrough
 from samtranslator.translator.arn_generator import ArnGenerator
 from samtranslator.utils.py27hash_fix import Py27Dict, Py27UniStr
+from samtranslator.utils.utils import InvalidValueType, dict_deep_set
 
 
 class SwaggerEditor(BaseEditor):
@@ -37,6 +39,7 @@ class SwaggerEditor(BaseEditor):
     _POLICY_TYPE_IAM = "Iam"
     _POLICY_TYPE_IP = "Ip"
     _POLICY_TYPE_VPC = "Vpc"
+    _DISABLE_EXECUTE_API_ENDPOINT = "disableExecuteApiEndpoint"
 
     def __init__(self, doc: Optional[Dict[str, Any]]) -> None:
         """
@@ -66,32 +69,40 @@ def __init__(self, doc: Optional[Dict[str, Any]]) -> None:
             for path_item in self.get_conditional_contents(self.paths.get(path)):
                 SwaggerEditor.validate_path_item_is_dict(path_item, path)
 
-    @staticmethod
-    def _update_dict(obj: Dict[Any, Any], k: str, v: Dict[Any, Any]) -> None:
-        if not obj.get(k):
-            obj[k] = {}
-        obj[k].update(v)
-
-    def add_disable_execute_api_endpoint_extension(self, disable_execute_api_endpoint):  # type: ignore[no-untyped-def]
+    def add_disable_execute_api_endpoint_extension(self, disable_execute_api_endpoint: PassThrough) -> None:
         """Add endpoint configuration to _X_APIGW_ENDPOINT_CONFIG in open api definition as extension
         Following this guide:
         https://docs.aws.amazon.com/apigateway/latest/developerguide/api-gateway-swagger-extensions-endpoint-configuration.html
         :param boolean disable_execute_api_endpoint: Specifies whether clients can invoke your API by using the default execute-api endpoint.
         """
-        DISABLE_EXECUTE_API_ENDPOINT = "disableExecuteApiEndpoint"
-        set_disable_api_endpoint = {DISABLE_EXECUTE_API_ENDPOINT: disable_execute_api_endpoint}
+
+        disable_execute_api_endpoint_path = f"{self._X_ENDPOINT_CONFIG}.{self._DISABLE_EXECUTE_API_ENDPOINT}"
 
         # Check if the OpenAPI version is 3.0, if it is then the extension needs to added to the Servers field,
         # if not then it gets added to the top level (same level as "paths" and "info")
         if self._doc.get("openapi") and self.validate_open_api_version_3(self._doc["openapi"]):
             # Add the x-amazon-apigateway-endpoint-configuration extension to the Servers objects
             servers_configurations = self._doc.get(self._SERVERS, [Py27Dict()])
-            for config in servers_configurations:
-                SwaggerEditor._update_dict(config, self._X_ENDPOINT_CONFIG, set_disable_api_endpoint)
+            for index, config in enumerate(servers_configurations):
+                try:
+                    dict_deep_set(config, disable_execute_api_endpoint_path, disable_execute_api_endpoint)
+                except InvalidValueType as ex:
+                    raise InvalidDocumentException(
+                        [
+                            InvalidTemplateException(
+                                f"Invalid OpenAPI definition of '{self._SERVERS}[{index}]': {str(ex)}."
+                            )
+                        ]
+                    ) from ex
 
             self._doc[self._SERVERS] = servers_configurations
         else:
-            SwaggerEditor._update_dict(self._doc, self._X_ENDPOINT_CONFIG, set_disable_api_endpoint)
+            try:
+                dict_deep_set(self._doc, disable_execute_api_endpoint_path, disable_execute_api_endpoint)
+            except InvalidValueType as ex:
+                raise InvalidDocumentException(
+                    [InvalidTemplateException(f"Invalid OpenAPI definition: {str(ex)}.")]
+                ) from ex
 
     def add_lambda_integration(
         self,

samtranslator/utils/utils.py

@@ -49,3 +49,25 @@ def dict_deep_get(d: Any, path: str) -> Optional[Any]:
         relative_path = (relative_path + f".{_path_nodes[0]}").lstrip(".")
         _path_nodes = _path_nodes[1:]
     return d
+
+
+def dict_deep_set(d: Any, path: str, value: Any) -> None:
+    """
+    Set the value deep in the dict.
+
+    If any value along the path doesn't exist, set to {}.
+    If any parent node exists but is not a dict, raise InvalidValueType.
+    """
+    relative_path = ""
+    if not path:
+        raise ValueError("path cannot be empty")
+    _path_nodes = path.split(".")
+    while len(_path_nodes) > 1:
+        if not isinstance(d, dict):
+            raise InvalidValueType(relative_path)
+        d = d.setdefault(_path_nodes[0], {})
+        relative_path = (relative_path + f".{_path_nodes[0]}").lstrip(".")
+        _path_nodes = _path_nodes[1:]
+    if not isinstance(d, dict):
+        raise InvalidValueType(relative_path)
+    d[_path_nodes[0]] = value

tests/translator/input/error_api_invalid_endpoint_configuration.yaml

@@ -0,0 +1,10 @@
+Resources:
+  ExplicitApi:
+    Type: AWS::Serverless::Api
+    Properties:
+      StageName: SomeStage
+      DefinitionBody:
+        swagger: 2.0
+        paths: {}
+        x-amazon-apigateway-endpoint-configuration: this should be a dict
+      DisableExecuteApiEndpoint: true

tests/translator/input/error_api_invalid_endpoint_configuration_openapi_3.yaml

@@ -0,0 +1,12 @@
+Resources:
+  ExplicitApi:
+    Type: AWS::Serverless::Api
+    Properties:
+      StageName: SomeStage
+      DefinitionBody:
+        openapi: 3.0.1
+        paths: {}
+        servers:
+        - x-amazon-apigateway-endpoint-configuration:
+          - this should be a dict
+      DisableExecuteApiEndpoint: true

tests/translator/output/error_api_invalid_endpoint_configuration.json

@@ -0,0 +1,3 @@
+{
+  "errorMessage": "Invalid Serverless Application Specification document. Number of errors found: 1. Structure of the SAM template is invalid. Invalid OpenAPI definition: The value of 'x-amazon-apigateway-endpoint-configuration' should be a map."
+}

tests/translator/output/error_api_invalid_endpoint_configuration_openapi_3.json

@@ -0,0 +1,3 @@
+{
+  "errorMessage": "Invalid Serverless Application Specification document. Number of errors found: 1. Structure of the SAM template is invalid. Invalid OpenAPI definition of 'servers[0]': The value of 'x-amazon-apigateway-endpoint-configuration' should be a map."
+}

tests/utils/test_utils.py

@@ -1,6 +1,6 @@
 from unittest import TestCase
 
-from samtranslator.utils.utils import as_array, insert_unique
+from samtranslator.utils.utils import InvalidValueType, as_array, dict_deep_set, insert_unique
 
 
 class TestUtils(TestCase):
@@ -26,3 +26,24 @@ def test_insert_unique(self):
         ret = insert_unique(xs, vs)
         self.assertFalse(ret is xs)
         self.assertFalse(ret is vs)
+
+    def test_dict_deep_set(self):
+        d = {"a": {"b": {"c": "hi"}}}
+        dict_deep_set(d, "a.b.d.hello", "world")
+        self.assertEqual(d, {"a": {"b": {"c": "hi", "d": {"hello": "world"}}}})
+        dict_deep_set(d, "a.b.hello", "world")
+        dict_deep_set(d, "a.hello", "world1")
+        self.assertEqual(d, {"a": {"hello": "world1", "b": {"hello": "world", "c": "hi", "d": {"hello": "world"}}}})
+
+    def test_dict_deep_set_invalid_type(self):
+        d = {"a": {"b": {"c": "hi"}}}
+        with self.assertRaisesRegex(InvalidValueType, r"The value of 'a\.b\.c' should be a map"):
+            dict_deep_set(d, "a.b.c.hello", "world")
+
+        with self.assertRaisesRegex(InvalidValueType, r"It should be a map"):
+            dict_deep_set("a str", "a.b.c.hello", "world")
+
+    def test_dict_deep_set_invalid_path(self):
+        d = {"a": {"b": {"c": "hi"}}}
+        with self.assertRaisesRegex(ValueError, r"path cannot be empty"):
+            dict_deep_set(d, "", "world")