feat(cors): support allow credentials cors configuration (#464)

Author: cast

samtranslator/model/api/api_generator.py

@@ -13,9 +13,9 @@
 from samtranslator.translator.arn_generator import ArnGenerator
 
 _CORS_WILDCARD = "'*'"
-CorsProperties = namedtuple("_CorsProperties", ["AllowMethods", "AllowHeaders", "AllowOrigin", "MaxAge"])
-# Default the Cors Properties to '*' wildcard. Other properties are actually Optional
-CorsProperties.__new__.__defaults__ = (None, None, _CORS_WILDCARD, None)
+CorsProperties = namedtuple("_CorsProperties", ["AllowMethods", "AllowHeaders", "AllowOrigin", "MaxAge", "AllowCredentials"])
+# Default the Cors Properties to '*' wildcard and False AllowCredentials. Other properties are actually Optional
+CorsProperties.__new__.__defaults__ = (None, None, _CORS_WILDCARD, None, False)
 
 AuthProperties = namedtuple("_AuthProperties", ["Authorizers", "DefaultAuthorizer"])
 AuthProperties.__new__.__defaults__ = (None, None)
@@ -214,10 +214,15 @@ def _add_cors(self):
             raise InvalidResourceException(self.logical_id, "Unable to add Cors configuration because "
                                                             "'DefinitionBody' does not contain a valid Swagger")
 
+        if properties.AllowCredentials is True and properties.AllowOrigin == _CORS_WILDCARD:
+            raise InvalidResourceException(self.logical_id, "Unable to add Cors configuration because "
+                                                            "'AllowCredentials' can not be true when "
+                                                            "'AllowOrigin' is \"'*'\" or not set")
+
         editor = SwaggerEditor(self.definition_body)
         for path in editor.iter_on_path():
             editor.add_cors(path,  properties.AllowOrigin, properties.AllowHeaders, properties.AllowMethods,
-                            max_age=properties.MaxAge)
+                            max_age=properties.MaxAge, allow_credentials=properties.AllowCredentials)
 
         # Assign the Swagger back to template
         self.definition_body = editor.swagger

samtranslator/swagger/swagger.py

@@ -16,6 +16,7 @@ class SwaggerEditor(object):
     _X_APIGW_INTEGRATION = 'x-amazon-apigateway-integration'
     _X_ANY_METHOD = 'x-amazon-apigateway-any-method'
 
+
     def __init__(self, doc):
         """
         Initialize the class with a swagger dictionary. This class creates a copy of the Swagger and performs all
@@ -118,7 +119,8 @@ def iter_on_path(self):
         for path, value in self.paths.items():
             yield path
 
-    def add_cors(self, path, allowed_origins, allowed_headers=None, allowed_methods=None, max_age=None):
+    def add_cors(self, path, allowed_origins, allowed_headers=None, allowed_methods=None, max_age=None,
+                 allow_credentials=None):
         """
         Add CORS configuration to this path. Specifically, we will add a OPTIONS response config to the Swagger that
         will return headers required for CORS. Since SAM uses aws_proxy integration, we cannot inject the headers
@@ -139,6 +141,7 @@ def add_cors(self, path, allowed_origins, allowed_headers=None, allowed_methods=
             Value can also be an intrinsic function dict.
         :param integer/dict max_age: Maximum duration to cache the CORS Preflight request. Value is set on
             Access-Control-Max-Age header. Value can also be an intrinsic function dict.
+        :param bool/None allow_credentials: Flags whether request is allowed to contain credentials.
         :raises ValueError: When values for one of the allowed_* variables is empty
         """
 
@@ -156,15 +159,19 @@ def add_cors(self, path, allowed_origins, allowed_headers=None, allowed_methods=
             # APIGW expects the value to be a "string expression". Hence wrap in another quote. Ex: "'GET,POST,DELETE'"
             allowed_methods = "'{}'".format(allowed_methods)
 
+        if allow_credentials is not True:
+            allow_credentials = False
+
         # Add the Options method and the CORS response
         self.add_path(path, self._OPTIONS_METHOD)
         self.paths[path][self._OPTIONS_METHOD] = self._options_method_response_for_cors(allowed_origins,
                                                                                         allowed_headers,
                                                                                         allowed_methods,
-                                                                                        max_age)
+                                                                                        max_age,
+                                                                                        allow_credentials)
 
     def _options_method_response_for_cors(self, allowed_origins, allowed_headers=None, allowed_methods=None,
-                                          max_age=None):
+                                          max_age=None, allow_credentials=None):
         """
         Returns a Swagger snippet containing configuration for OPTIONS HTTP Method to configure CORS.
 
@@ -179,6 +186,7 @@ def _options_method_response_for_cors(self, allowed_origins, allowed_headers=Non
             Value can also be an intrinsic function dict.
         :param integer/dict max_age: Maximum duration to cache the CORS Preflight request. Value is set on
             Access-Control-Max-Age header. Value can also be an intrinsic function dict.
+        :param bool allow_credentials: Flags whether request is allowed to contain credentials.
 
         :return dict: Dictionary containing Options method configuration for CORS
         """
@@ -187,6 +195,7 @@ def _options_method_response_for_cors(self, allowed_origins, allowed_headers=Non
         ALLOW_HEADERS = "Access-Control-Allow-Headers"
         ALLOW_METHODS = "Access-Control-Allow-Methods"
         MAX_AGE = "Access-Control-Max-Age"
+        ALLOW_CREDENTIALS = "Access-Control-Allow-Credentials"
         HEADER_RESPONSE = lambda x: "method.response.header."+x
 
         response_parameters = {
@@ -217,6 +226,11 @@ def _options_method_response_for_cors(self, allowed_origins, allowed_headers=Non
             # MaxAge can be set to 0, which is a valid value. So explicitly check against None
             response_parameters[HEADER_RESPONSE(MAX_AGE)] = max_age
             response_headers[MAX_AGE] = {"type": "integer"}
+        if allow_credentials is True:
+            # Allow-Credentials only has a valid value of true, it should be omitted otherwise.
+            # https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Access-Control-Allow-Credentials
+            response_parameters[HEADER_RESPONSE(ALLOW_CREDENTIALS)] = "'true'"
+            response_headers[ALLOW_CREDENTIALS] = {"type": "string"}
 
         return {
             "summary": "CORS support",

tests/swagger/test_swagger.py

@@ -8,6 +8,7 @@
 
 _X_INTEGRATION = "x-amazon-apigateway-integration"
 _X_ANY_METHOD = 'x-amazon-apigateway-any-method'
+_ALLOW_CREDENTALS_TRUE = "'true'"
 
 class TestSwaggerEditor_init(TestCase):
 
@@ -308,18 +309,21 @@ def test_must_add_options_to_new_path(self):
         allowed_headers = ["headers", "2"]
         allowed_methods = {"key": "methods"}
         max_age = 60
+        allow_credentials = True
+        options_method_response_allow_credentials = True
         path = "/foo"
         expected = {"some cors": "return value"}
 
         self.editor._options_method_response_for_cors = Mock()
         self.editor._options_method_response_for_cors.return_value = expected
 
-        self.editor.add_cors(path, allowed_origins, allowed_headers, allowed_methods, max_age)
+        self.editor.add_cors(path, allowed_origins, allowed_headers, allowed_methods, max_age, allow_credentials)
         self.assertEquals(expected, self.editor.swagger["paths"][path]["options"])
         self.editor._options_method_response_for_cors.assert_called_with(allowed_origins,
                                                                          allowed_headers,
                                                                          allowed_methods,
-                                                                         max_age)
+                                                                         max_age,
+                                                                         options_method_response_allow_credentials)
 
     def test_must_skip_existing_path(self):
         path = "/withoptions"
@@ -346,28 +350,33 @@ def test_must_work_for_optional_allowed_headers(self):
         allowed_headers = None # No Value
         allowed_methods = "methods"
         max_age = 60
+        allow_credentials = True
+        options_method_response_allow_credentials = True
 
         expected = {"some cors": "return value"}
         path = "/foo"
 
         self.editor._options_method_response_for_cors = Mock()
         self.editor._options_method_response_for_cors.return_value = expected
 
-        self.editor.add_cors(path, allowed_origins, allowed_headers, allowed_methods, max_age)
+        self.editor.add_cors(path, allowed_origins, allowed_headers, allowed_methods, max_age, allow_credentials)
 
         self.assertEquals(expected, self.editor.swagger["paths"][path]["options"])
 
         self.editor._options_method_response_for_cors.assert_called_with(allowed_origins,
                                                                          allowed_headers,
                                                                          allowed_methods,
-                                                                         max_age)
+                                                                         max_age,
+                                                                         options_method_response_allow_credentials)
 
     def test_must_make_default_value_with_optional_allowed_methods(self):
 
         allowed_origins = "origins"
         allowed_headers = "headers"
         allowed_methods = None  # No Value
         max_age = 60
+        allow_credentials = True
+        options_method_response_allow_credentials = True
 
         default_allow_methods_value = "some default value"
         default_allow_methods_value_with_quotes = "'{}'".format(default_allow_methods_value)
@@ -380,7 +389,7 @@ def test_must_make_default_value_with_optional_allowed_methods(self):
         self.editor._options_method_response_for_cors = Mock()
         self.editor._options_method_response_for_cors.return_value = expected
 
-        self.editor.add_cors(path, allowed_origins, allowed_headers, allowed_methods, max_age)
+        self.editor.add_cors(path, allowed_origins, allowed_headers, allowed_methods, max_age, allow_credentials)
 
         self.assertEquals(expected, self.editor.swagger["paths"][path]["options"])
 
@@ -389,7 +398,29 @@ def test_must_make_default_value_with_optional_allowed_methods(self):
                                                                          # Must be called with default value.
                                                                          # And value must be quoted
                                                                          default_allow_methods_value_with_quotes,
-                                                                         max_age)
+                                                                         max_age,
+                                                                         options_method_response_allow_credentials)
+
+    def test_must_accept_none_allow_credentials(self):
+        allowed_origins = "origins"
+        allowed_headers = ["headers", "2"]
+        allowed_methods = {"key": "methods"}
+        max_age = 60
+        allow_credentials = None
+        options_method_response_allow_credentials = False
+        path = "/foo"
+        expected = {"some cors": "return value"}
+
+        self.editor._options_method_response_for_cors = Mock()
+        self.editor._options_method_response_for_cors.return_value = expected
+
+        self.editor.add_cors(path, allowed_origins, allowed_headers, allowed_methods, max_age, allow_credentials)
+        self.assertEquals(expected, self.editor.swagger["paths"][path]["options"])
+        self.editor._options_method_response_for_cors.assert_called_with(allowed_origins,
+                                                                         allowed_headers,
+                                                                         allowed_methods,
+                                                                         max_age,
+                                                                         options_method_response_allow_credentials)
 
 
 class TestSwaggerEditor_options_method_response_for_cors(TestCase):
@@ -400,6 +431,7 @@ def test_correct_value_is_returned(self):
         methods = {"a": "b"}
         origins = [1,2,3]
         max_age = 60
+        allow_credentials = True
 
         expected = {
             "summary": "CORS support",
@@ -414,10 +446,11 @@ def test_correct_value_is_returned(self):
                     "default": {
                         "statusCode": "200",
                         "responseParameters": {
+                            "method.response.header.Access-Control-Allow-Credentials": _ALLOW_CREDENTALS_TRUE,
                             "method.response.header.Access-Control-Allow-Headers": headers,
                             "method.response.header.Access-Control-Allow-Methods": methods,
                             "method.response.header.Access-Control-Allow-Origin": origins,
-                            "method.response.header.Access-Control-Max-Age": max_age
+                            "method.response.header.Access-Control-Max-Age": max_age,
                         },
                         "responseTemplates": {
                             "application/json": "{}\n"
@@ -429,6 +462,9 @@ def test_correct_value_is_returned(self):
                 "200": {
                     "description": "Default response for CORS method",
                     "headers": {
+                        "Access-Control-Allow-Credentials": {
+                            "type": "string"
+                        },
                         "Access-Control-Allow-Headers": {
                             "type": "string"
                         },
@@ -446,20 +482,27 @@ def test_correct_value_is_returned(self):
             }
         }
 
-        actual = SwaggerEditor(SwaggerEditor.gen_skeleton())._options_method_response_for_cors(origins, headers, methods, max_age)
+        actual = SwaggerEditor(SwaggerEditor.gen_skeleton())._options_method_response_for_cors(origins, headers,
+                                                                                               methods, max_age,
+                                                                                               allow_credentials)
         self.assertEquals(expected, actual)
 
     def test_allow_headers_is_skipped_with_no_value(self):
         headers = None # No value
         methods = "methods"
         origins = "origins"
+        allow_credentials = True
 
         expected = {
+            "method.response.header.Access-Control-Allow-Credentials": _ALLOW_CREDENTALS_TRUE,
             "method.response.header.Access-Control-Allow-Methods": methods,
-            "method.response.header.Access-Control-Allow-Origin": origins
+            "method.response.header.Access-Control-Allow-Origin": origins,
         }
 
         expected_headers = {
+            "Access-Control-Allow-Credentials": {
+                "type": "string"
+            },
             "Access-Control-Allow-Methods": {
                 "type": "string"
             },
@@ -469,7 +512,7 @@ def test_allow_headers_is_skipped_with_no_value(self):
         }
 
         options_config = SwaggerEditor(SwaggerEditor.gen_skeleton())._options_method_response_for_cors(
-            origins, headers, methods)
+            origins, headers, methods, allow_credentials=allow_credentials)
 
         actual = options_config[_X_INTEGRATION]["responses"]["default"]["responseParameters"]
         self.assertEquals(expected, actual)
@@ -479,14 +522,16 @@ def test_allow_methods_is_skipped_with_no_value(self):
         headers = "headers"
         methods = None # No value
         origins = "origins"
+        allow_credentials = True
 
         expected = {
+            "method.response.header.Access-Control-Allow-Credentials": _ALLOW_CREDENTALS_TRUE,
             "method.response.header.Access-Control-Allow-Headers": headers,
-            "method.response.header.Access-Control-Allow-Origin": origins
+            "method.response.header.Access-Control-Allow-Origin": origins,
         }
 
         options_config = SwaggerEditor(SwaggerEditor.gen_skeleton())._options_method_response_for_cors(
-            origins, headers, methods)
+            origins, headers, methods, allow_credentials=allow_credentials)
 
         actual = options_config[_X_INTEGRATION]["responses"]["default"]["responseParameters"]
         self.assertEquals(expected, actual)
@@ -495,14 +540,15 @@ def test_allow_origins_is_not_skipped_with_no_value(self):
         headers = None
         methods = None
         origins = None
+        allow_credentials = False
 
         expected = {
             # We will ALWAYS set AllowOrigin. This is a minimum requirement for CORS
             "method.response.header.Access-Control-Allow-Origin": origins
         }
 
         options_config = SwaggerEditor(SwaggerEditor.gen_skeleton())._options_method_response_for_cors(
-            origins, headers, methods)
+            origins, headers, methods, allow_credentials=allow_credentials)
 
         actual = options_config[_X_INTEGRATION]["responses"]["default"]["responseParameters"]
         self.assertEquals(expected, actual)
@@ -512,19 +558,38 @@ def test_max_age_can_be_set_to_zero(self):
         methods = "methods"
         origins = "origins"
         max_age = 0
+        allow_credentials = True
 
         expected = {
+            "method.response.header.Access-Control-Allow-Credentials": _ALLOW_CREDENTALS_TRUE,
             "method.response.header.Access-Control-Allow-Methods": methods,
             "method.response.header.Access-Control-Allow-Origin": origins,
-            "method.response.header.Access-Control-Max-Age": max_age
+            "method.response.header.Access-Control-Max-Age": max_age,
         }
 
         options_config = SwaggerEditor(SwaggerEditor.gen_skeleton())._options_method_response_for_cors(
-            origins, headers, methods, max_age)
+            origins, headers, methods, max_age, allow_credentials)
 
         actual = options_config[_X_INTEGRATION]["responses"]["default"]["responseParameters"]
         self.assertEquals(expected, actual)
 
+    def test_allow_credentials_is_skipped_with_false_value(self):
+        headers = "headers"
+        methods = "methods"
+        origins = "origins"
+        allow_credentials = False
+
+        expected = {
+            "method.response.header.Access-Control-Allow-Headers": headers,
+            "method.response.header.Access-Control-Allow-Methods": methods,
+            "method.response.header.Access-Control-Allow-Origin": origins,
+        }
+
+        options_config = SwaggerEditor(SwaggerEditor.gen_skeleton())._options_method_response_for_cors(
+            origins, headers, methods, allow_credentials=allow_credentials)
+
+        actual = options_config[_X_INTEGRATION]["responses"]["default"]["responseParameters"]
+        self.assertEquals(expected, actual)
 
 class TestSwaggerEditor_make_cors_allowed_methods_for_path(TestCase):
 

tests/translator/input/api_with_cors.yaml

@@ -66,3 +66,4 @@ Resources:
         AllowMethods: "methods"
         AllowHeaders: "headers"
         AllowOrigin: "origins"
+        AllowCredentials: true