fix(event_handler): fix prefix stripping, gzip negotiation, JSON detection and CORS Vary

aws_lambda_powertools/event_handler/api_gateway.py

@@ -242,6 +242,9 @@
             headers["Access-Control-Max-Age"] = str(self.max_age)
         if origin != "*" and self.allow_credentials is True:
             headers["Access-Control-Allow-Credentials"] = "true"
+        if origin != "*":
+            # The response echoes the request Origin, so shared caches must not reuse it for other origins
+            headers["Vary"] = "Origin"
         return headers
 
     def allowed_origin(self, extracted_origin: str) -> str | None:
@@ -362,7 +365,8 @@
         """
         Returns True if the response is JSON, based on the Content-Type.
         """
-        content_type = self.headers.get("Content-Type", "")
+        # Header names are case-insensitive, so match "content-type" in any casing
+        content_type = next((value for key, value in self.headers.items() if key.lower() == "content-type"), "")
         if isinstance(content_type, list):
             content_type = content_type[0]
         return content_type.startswith(DEFAULT_CONTENT_TYPE)
@@ -746,6 +750,20 @@
         return operation_id
 
 
+def _accepts_gzip(accept_encoding: str) -> bool:
+    """Returns True if an Accept-Encoding header accepts gzip, honouring casing and a zero q-value"""
+    for coding in accept_encoding.split(","):
+        name, _, params = coding.partition(";")
+        if name.strip().lower() != "gzip":
+            continue
+        _, _, weight = params.replace(" ", "").lower().partition("q=")
+        try:
+            return float(weight) > 0 if weight else True
+        except ValueError:
+            return False
+    return False
+
+
 class ResponseBuilder(Generic[ResponseEventT]):
     """Internally used Response builder"""
 
@@ -765,7 +783,14 @@
 
         origin = cors.allowed_origin(extracted_origin_header)
         if origin is not None:
-            self.response.headers.update(cors.to_dict(origin))
+            cors_headers = cors.to_dict(origin)
+            # Keep any Vary value the route already set, and add Origin to it
+            existing_vary = self.response.headers.get("Vary")
+            if isinstance(existing_vary, list):
+                existing_vary = ", ".join(existing_vary)
+            if existing_vary and "Vary" in cors_headers:
+                cors_headers["Vary"] = f"{existing_vary}, Origin"
+            self.response.headers.update(cors_headers)
 
     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."""
@@ -798,11 +823,9 @@
             True if compression is enabled and the "gzip" encoding is accepted, False otherwise.
         """
         encoding = event.resolved_headers_field.get("accept-encoding", "")
-        gzip_accepted = False
-        if isinstance(encoding, str):
-            gzip_accepted = "gzip" in encoding
-        elif isinstance(encoding, list):
-            gzip_accepted = "gzip" in ",".join(encoding)
+        if isinstance(encoding, list):
+            encoding = ",".join(encoding)
+        gzip_accepted = isinstance(encoding, str) and _accepts_gzip(encoding)
 
         if gzip_accepted:
             if response_compression is not None:
@@ -845,8 +868,8 @@
         """Build the full response dict to be returned by the lambda"""
 
         # We only apply the serializer when the content type is JSON and the
-        # body is not a str, to avoid double encoding
-        if self.response.is_json() and not isinstance(self.response.body, str):
+        # body is not a str or bytes, to avoid double encoding (bytes are base64 encoded below)
+        if self.response.is_json() and not isinstance(self.response.body, (str, bytes)):
             self.response.body = self.serializer(self.response.body)
 
         self._route(event, cors)
@@ -2869,7 +2892,7 @@
 
         return self._handle_not_found(method=method, path=path)
 
     def _remove_prefix(self, path: str) -> str:
         """Remove the configured prefix from the path"""
         if not isinstance(self._strip_prefixes, list):
             return path
@@ -2883,7 +2906,10 @@
                     return path[len(prefix) :]
 
             if isinstance(prefix, Pattern):
-                path = re.sub(prefix, "", path)
+                # Only strip a match at the start of the path, and only once
+                match = prefix.match(path)
+                if match:
+                    path = path[match.end() :]
 
                 # When using regexes, we might get into a point where everything is removed
                 # from the string, so we check if it's empty and return /, since there's nothing

tests/functional/event_handler/required_dependencies/test_api_gateway.py

@@ -523,6 +523,39 @@ def test_response_is_json_with_multiple_json_content_types():
     assert response.is_json() is True
 
 
+def test_response_with_bytes_body_and_json_content_type_is_base64_encoded():
+    # GIVEN a route that returns a bytes body with a JSON content type
+    app = ApiGatewayResolver()
+    body = b'{"a":1}'
+
+    @app.get("/my/path")
+    def get_bytes() -> Response:
+        return Response(200, content_types.APPLICATION_JSON, body)
+
+    # WHEN calling the event handler
+    result = app({"path": "/my/path", "httpMethod": "GET"}, None)
+
+    # THEN the bytes are base64 encoded instead of being passed to the JSON serializer
+    assert result["isBase64Encoded"] is True
+    assert base64.b64decode(result["body"]) == body
+
+
+def test_response_with_lowercase_content_type_header_serializes_json_body():
+    # GIVEN a route that sets the JSON content type with a lowercase header name
+    app = ApiGatewayResolver()
+
+    @app.get("/my/path")
+    def get_lowercase_header() -> Response:
+        return Response(200, headers={"content-type": content_types.APPLICATION_JSON}, body={"a": 1})
+
+    # WHEN calling the event handler
+    result = app({"path": "/my/path", "httpMethod": "GET"}, None)
+
+    # THEN the response is recognised as JSON and the body is serialized
+    assert isinstance(result["body"], str)
+    assert json.loads(result["body"]) == {"a": 1}
+
+
 def test_compress():
     # GIVEN a function that has compress=True
     # AND an event with a "Accept-Encoding" that include gzip
@@ -588,6 +621,32 @@ def return_text() -> Response:
     assert result["body"] == expected_value
 
 
+@pytest.mark.parametrize(
+    "accept_encoding, expected_compressed",
+    [
+        pytest.param("gzip;q=0", False, id="gzip refused with a zero weight"),
+        pytest.param("deflate, gzip; q=0.0", False, id="gzip refused with a zero weight and spaces"),
+        pytest.param("GZIP", True, id="gzip accepted in uppercase"),
+        pytest.param("deflate, gzip;q=0.5", True, id="gzip accepted with a non-zero weight"),
+    ],
+)
+def test_compress_respects_accept_encoding_weights_and_casing(accept_encoding: str, expected_compressed: bool):
+    # GIVEN a function with compress=True
+    app = ApiGatewayResolver()
+
+    @app.get("/my/path", compress=True)
+    def return_text() -> Response:
+        return Response(200, content_types.TEXT_PLAIN, "Foo")
+
+    # WHEN the client sends an Accept-Encoding header with a weight or different casing
+    event = {"path": "/my/path", "httpMethod": "GET", "headers": {"Accept-Encoding": accept_encoding}}
+    result = app(event, None)
+
+    # THEN the response is only gzipped when the client accepts gzip
+    assert result["isBase64Encoded"] is expected_compressed
+    assert ("Content-Encoding" in result["multiValueHeaders"]) is expected_compressed
+
+
 def test_compress_no_accept_encoding_null_headers():
     # GIVEN a function with compress=True
     # AND the request has no headers
@@ -784,6 +843,59 @@ def get_with_cors():
     assert "Access-Control-Allow-Origin" not in headers
 
 
+def test_cors_adds_vary_origin_when_echoing_a_specific_origin():
+    # GIVEN a cors configuration with more than one allowed origin
+    cors_config = CORSConfig(allow_origin="https://a.com", extra_origins=["https://b.com"])
+    app = ApiGatewayResolver(cors=cors_config)
+
+    @app.get("/cors")
+    def get_with_cors():
+        return {}
+
+    # WHEN calling the event handler with one of the allowed origins
+    event = {"path": "/cors", "httpMethod": "GET", "headers": {"Origin": "https://a.com"}}
+    result = app(event, None)
+
+    # THEN the echoed origin comes with Vary: Origin so shared caches don't reuse it for other origins
+    headers = result["multiValueHeaders"]
+    assert headers["Access-Control-Allow-Origin"] == ["https://a.com"]
+    assert headers["Vary"] == ["Origin"]
+
+
+def test_cors_keeps_existing_vary_header_when_adding_origin():
+    # GIVEN a route that already sets a Vary header
+    app = ApiGatewayResolver(cors=CORSConfig(allow_origin="https://a.com"))
+
+    @app.get("/cors")
+    def get_with_vary() -> Response:
+        return Response(200, content_types.APPLICATION_JSON, "{}", headers={"Vary": "Accept-Encoding"})
+
+    # WHEN calling the event handler with the allowed origin
+    event = {"path": "/cors", "httpMethod": "GET", "headers": {"Origin": "https://a.com"}}
+    result = app(event, None)
+
+    # THEN Origin is added to the existing Vary value instead of replacing it
+    assert result["multiValueHeaders"]["Vary"] == ["Accept-Encoding, Origin"]
+
+
+def test_cors_does_not_add_vary_origin_for_wildcard_origin():
+    # GIVEN a cors configuration that allows every origin
+    app = ApiGatewayResolver(cors=CORSConfig(allow_origin="*"))
+
+    @app.get("/cors")
+    def get_with_cors():
+        return {}
+
+    # WHEN calling the event handler with any origin
+    event = {"path": "/cors", "httpMethod": "GET", "headers": {"Origin": "https://a.com"}}
+    result = app(event, None)
+
+    # THEN the wildcard response is the same for every origin, so no Vary header is needed
+    headers = result["multiValueHeaders"]
+    assert headers["Access-Control-Allow-Origin"] == ["*"]
+    assert "Vary" not in headers
+
+
 def test_custom_cors_config():
     # GIVEN a custom cors configuration
     allow_header = ["foo2"]
@@ -1146,6 +1258,35 @@ def test_debug_print_event(capsys):
     assert json.loads(output) == event
 
 
+@pytest.mark.parametrize(
+    "rule, path, expected_route_keys",
+    [
+        pytest.param(".+", "/any/deep/path", {}, id="catch-all matches any path"),
+        pytest.param("/files/.+", "/files/reports/2026/summary.pdf", {}, id="catch-all after a static prefix"),
+        pytest.param(
+            "/users/<user_id>/files/.+",
+            "/users/123/files/docs/report.txt",
+            {"user_id": "123"},
+            id="catch-all after a dynamic parameter",
+        ),
+    ],
+)
+def test_route_rule_supports_documented_catch_all_regex(rule: str, path: str, expected_route_keys: dict):
+    # GIVEN a route that uses a regex catch-all, as shown in the "Catch-all routes" docs
+    app = APIGatewayRestResolver()
+
+    @app.get(rule)
+    def handler(**route_keys):
+        return route_keys
+
+    # WHEN calling the event handler with a nested path
+    result = app({"httpMethod": "GET", "path": path}, None)
+
+    # THEN the regex in the rule matches the whole path and dynamic parameters are still captured
+    assert result["statusCode"] == 200
+    assert json.loads(result["body"]) == expected_route_keys
+
+
 def test_similar_dynamic_routes():
     # GIVEN
     app = ApiGatewayResolver()
@@ -1323,6 +1464,29 @@ def foo(): ...
     assert response["statusCode"] == 200
 
 
+@pytest.mark.parametrize(
+    "path",
+    [
+        pytest.param("/v1/users/v2", id="prefix pattern also appears later in the path"),
+        pytest.param("/users/v2", id="prefix pattern only appears later in the path"),
+    ],
+)
+def test_remove_prefix_by_regex_only_strips_leading_match(path: str):
+    # GIVEN a regex strip_prefix that could also match a later path segment
+    app = ApiGatewayResolver(strip_prefixes=[re.compile(r"/v[1-9]+")])
+
+    @app.get("/users/<name>")
+    def get_user(name: str):
+        return {"name": name}
+
+    # WHEN calling the handler with the pattern repeated later in the path
+    response = app({"httpMethod": "GET", "path": path}, None)
+
+    # THEN only a leading match is stripped, so the route matches and keeps the later segment
+    assert response["statusCode"] == 200
+    assert json.loads(response["body"]) == {"name": "v2"}
+
+
 def test_empty_path_when_using_regexes():
     app = ApiGatewayResolver(strip_prefixes=[re.compile(r"/(dev|stg)")])