fix cors

Author: sthulb

aws_lambda_powertools/event_handler/api_gateway.py

@@ -188,9 +188,12 @@ def __init__(
         allow_credentials: bool
             A boolean value that sets the value of `Access-Control-Allow-Credentials`
         """
-        self._allowed_origins = [allow_origin]
+
+        self.allowed_origins = [allow_origin]
+
         if extra_origins:
-            self._allowed_origins.extend(extra_origins)
+            self.allowed_origins.extend(extra_origins)
+
         self.allow_headers = set(self._REQUIRED_HEADERS + (allow_headers or []))
         self.expose_headers = expose_headers or []
         self.max_age = max_age
@@ -205,7 +208,7 @@ def to_dict(self, origin: Optional[str]) -> Dict[str, str]:
 
         # If the origin doesn't match any of the allowed origins, and we don't allow all origins ("*"),
         # don't add any CORS headers
-        if origin not in self._allowed_origins and "*" not in self._allowed_origins:
+        if origin not in self.allowed_origins and "*" not in self.allowed_origins:
             return {}
 
         # The origin matched an allowed origin, so return the CORS headers
@@ -218,7 +221,7 @@ def to_dict(self, origin: Optional[str]) -> Dict[str, str]:
             headers["Access-Control-Expose-Headers"] = ",".join(self.expose_headers)
         if self.max_age is not None:
             headers["Access-Control-Max-Age"] = str(self.max_age)
-        if self.allow_credentials is True:
+        if origin != "*" and self.allow_credentials is True:
             headers["Access-Control-Allow-Credentials"] = "true"
         return headers
 
@@ -806,10 +809,11 @@ def __init__(
     def _add_cors(self, event: ResponseEventT, cors: CORSConfig):
         """Update headers to include the configured Access-Control headers"""
         extracted_origin_header = extract_origin_header(event.resolved_headers_field)
-        if extracted_origin_header is None:
-            self.response.headers.update(cors.to_dict("*"))
-        else:
+
+        if extracted_origin_header in cors.allowed_origins:
             self.response.headers.update(cors.to_dict(extracted_origin_header))
+        if extracted_origin_header is not None and "*" in cors.allowed_origins:
+            self.response.headers.update(cors.to_dict("*"))
 
     def _add_cache_control(self, cache_control: str):
         """Set the specified cache control headers for 200 http responses. For non-200 `no-cache` is used."""

tests/functional/event_handler/required_dependencies/test_api_gateway.py

@@ -324,7 +324,7 @@ def handler(event, context):
 def test_cors():
     # GIVEN a function with cors=True
     # AND http method set to GET
-    app = ApiGatewayResolver()
+    app = ApiGatewayResolver(cors=CORSConfig("https://aws.amazon.com", allow_credentials=True))
 
     @app.get("/my/path", cors=True)
     def with_cors() -> Response:
@@ -345,7 +345,7 @@ def handler(event, context):
     headers = result["multiValueHeaders"]
     assert headers["Content-Type"] == [content_types.TEXT_HTML]
     assert headers["Access-Control-Allow-Origin"] == ["https://aws.amazon.com"]
-    assert "Access-Control-Allow-Credentials" not in headers
+    assert "Access-Control-Allow-Credentials" in headers
     assert headers["Access-Control-Allow-Headers"] == [",".join(sorted(CORSConfig._REQUIRED_HEADERS))]
 
     # THEN for routes without cors flag return no cors headers
@@ -354,7 +354,7 @@ def handler(event, context):
     assert "Access-Control-Allow-Origin" not in result["multiValueHeaders"]
 
 
-def test_cors_no_origin():
+def test_cors_no_request_origin():
     # GIVEN a function with cors=True
     # AND http method set to GET
     app = ApiGatewayResolver()
@@ -366,8 +366,41 @@ def with_cors() -> Response:
     def handler(event, context):
         return app.resolve(event, context)
 
-    # remove origin header from request
-    del LOAD_GW_EVENT["multiValueHeaders"]["Origin"]
+    event = LOAD_GW_EVENT.copy()
+    del event["headers"]["Origin"]
+    del event["multiValueHeaders"]["Origin"]
+
+    # WHEN calling the event handler
+    result = handler(LOAD_GW_EVENT, None)
+
+    # THEN the headers should include cors headers
+    assert "multiValueHeaders" in result
+    headers = result["multiValueHeaders"]
+    assert headers["Content-Type"] == [content_types.TEXT_HTML]
+    assert "Access-Control-Allow-Credentials" not in headers
+    assert "Access-Control-Allow-Origin" not in result["multiValueHeaders"]
+
+
+def test_cors_allow_all_request_origins():
+    # GIVEN a function with cors=True
+    # AND http method set to GET
+    app = ApiGatewayResolver(
+        cors=CORSConfig(
+            allow_origin="*",
+            allow_credentials=True,
+        ),
+    )
+
+    @app.get("/my/path", cors=True)
+    def with_cors() -> Response:
+        return Response(200, content_types.TEXT_HTML, "test")
+
+    @app.get("/without-cors")
+    def without_cors() -> Response:
+        return Response(200, content_types.TEXT_HTML, "test")
+
+    def handler(event, context):
+        return app.resolve(event, context)
 
     # WHEN calling the event handler
     result = handler(LOAD_GW_EVENT, None)
@@ -380,6 +413,11 @@ def handler(event, context):
     assert "Access-Control-Allow-Credentials" not in headers
     assert headers["Access-Control-Allow-Headers"] == [",".join(sorted(CORSConfig._REQUIRED_HEADERS))]
 
+    # THEN for routes without cors flag return no cors headers
+    mock_event = {"path": "/my/request", "httpMethod": "GET"}
+    result = handler(mock_event, None)
+    assert "Access-Control-Allow-Origin" not in result["multiValueHeaders"]
+
 
 def test_cors_preflight_body_is_empty_not_null():
     # GIVEN CORS is configured