diff --git a/aws_lambda_powertools/event_handler/api_gateway.py b/aws_lambda_powertools/event_handler/api_gateway.py index 7a8aaa84a95..72c760f126e 100644 --- a/aws_lambda_powertools/event_handler/api_gateway.py +++ b/aws_lambda_powertools/event_handler/api_gateway.py @@ -242,6 +242,9 @@ def to_dict(self, origin: str | None) -> dict[str, str]: 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 @@ def is_json(self) -> bool: """ 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 @@ def _generate_operation_id(self) -> str: 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 @@ def _add_cors(self, event: ResponseEventT, cors: CORSConfig): 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 @@ def _has_compression_enabled( 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 @@ def build(self, event: ResponseEventT, cors: CORSConfig | None = None) -> dict[s """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) @@ -2883,7 +2906,10 @@ def _remove_prefix(self, path: str) -> str: 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 diff --git a/tests/functional/event_handler/required_dependencies/test_api_gateway.py b/tests/functional/event_handler/required_dependencies/test_api_gateway.py index fc7ee54767d..2f21174bd97 100644 --- a/tests/functional/event_handler/required_dependencies/test_api_gateway.py +++ b/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//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/") + 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)")])