From be51600b607bbbfa07ef5274e08219b59088b842 Mon Sep 17 00:00:00 2001 From: MTirnova <88968436+MTirnova@users.noreply.github.com> Date: Sun, 18 Oct 2026 22:33:18 +0300 Subject: [PATCH 1/2] fix(event_handler): harden multipart and JSON body validation - Return 415 when multipart/form-data has no boundary, and 422 for other multipart parsing errors (invalid base64, undecodable text) instead of letting them escape as 500/502. - Decode non-ASCII text fields in non-base64 multipart bodies correctly. - Only unwrap single-element lists for form and multipart bodies, so a JSON array sent to an object or scalar body parameter returns 422. - Parse Content-Disposition parameters so filename= before name= works. --- .../middlewares/openapi_validation.py | 83 ++++++--- .../test_openapi_comma_parameters.py | 17 +- .../test_openapi_validation_middleware.py | 166 +++++++++++++++++- 3 files changed, 236 insertions(+), 30 deletions(-) diff --git a/aws_lambda_powertools/event_handler/middlewares/openapi_validation.py b/aws_lambda_powertools/event_handler/middlewares/openapi_validation.py index 31bcfd70cbc..07a4dad1d17 100644 --- a/aws_lambda_powertools/event_handler/middlewares/openapi_validation.py +++ b/aws_lambda_powertools/event_handler/middlewares/openapi_validation.py @@ -4,6 +4,7 @@ import dataclasses import json import logging +import re import warnings from typing import TYPE_CHECKING, Any, Union, cast, get_args, get_origin from urllib.parse import parse_qs, unquote @@ -123,6 +124,7 @@ def handler(self, app: EventHandlerInstance, next_middleware: NextMiddleware) -> (body_values, body_errors) = _request_body_to_args( required_params=route.dependant.body_params, received_body=self._get_body(app), + is_form_body=_is_form_content_type(app.current_event.headers.get("content-type", "")), ) values.update(body_values) errors.extend(body_errors) @@ -211,12 +213,23 @@ def _parse_form_data(self, app: EventHandlerInstance) -> dict[str, Any]: def _parse_multipart_data(self, app: EventHandlerInstance, content_type: str) -> dict[str, Any]: """Parse multipart/form-data from the request body (file uploads).""" - try: - # Extract the boundary from the content-type header - boundary = _extract_multipart_boundary(content_type) - if not boundary: - raise ValueError("Missing boundary in multipart/form-data content-type header") + # Extract the boundary from the content-type header + boundary = _extract_multipart_boundary(content_type) + if not boundary: + raise RequestUnsupportedContentType( + "Missing boundary in multipart/form-data content-type header", + errors=[ + { + "type": "missing_boundary", + "loc": ("body",), + "msg": "Missing boundary in multipart/form-data content-type header", + "input": {}, + "ctx": {}, + }, + ], + ) + try: # Get raw body bytes raw_body = app.current_event.body or "" if app.current_event.is_base64_encoded: @@ -231,14 +244,10 @@ def _parse_multipart_data(self, app: EventHandlerInstance, content_type: str) -> "api-gateway-payload-encodings.html", stacklevel=2, ) - # Use latin-1 to preserve all byte values (0-255) since the body - # may contain raw binary data that isn't valid UTF-8 - body_bytes = raw_body.encode("latin-1") + body_bytes = _encode_non_base64_body(raw_body) return _parse_multipart_body(body_bytes, boundary) - except ValueError: - raise except Exception as e: raise RequestValidationError( [ @@ -436,9 +445,13 @@ def _request_params_to_args( def _request_body_to_args( required_params: list[ModelField], received_body: dict[str, Any] | None, + is_form_body: bool = False, ) -> tuple[dict[str, Any], list[dict[str, Any]]]: """ Convert the request body to a dictionary of values using validation, and returns a list of errors. + + Form and multipart parsers collect values into lists, so single values are only unwrapped + from lists when ``is_form_body`` is set. JSON lists are passed through to validation as-is. """ values: dict[str, Any] = {} errors: list[dict[str, Any]] = [] @@ -458,7 +471,7 @@ def _request_body_to_args( _handle_missing_field_value(field, values, errors, loc) continue - value = _normalize_field_value(value=value, field_info=field.field_info) + value = _normalize_field_value(value=value, field_info=field.field_info, unwrap_lists=is_form_body) # UploadFile objects bypass Pydantic validation — they're already constructed if isinstance(value, UploadFile): @@ -538,7 +551,7 @@ def _is_or_contains_sequence(annotation: Any) -> bool: return False -def _normalize_field_value(value: Any, field_info: FieldInfo) -> Any: +def _normalize_field_value(value: Any, field_info: FieldInfo, unwrap_lists: bool = True) -> Any: """Normalize field value, converting lists to single values for non-sequence fields.""" # When annotation is bytes but value is UploadFile, extract raw content if isinstance(value, UploadFile) and field_info.annotation is bytes: @@ -546,7 +559,7 @@ def _normalize_field_value(value: Any, field_info: FieldInfo) -> Any: if _is_or_contains_sequence(field_info.annotation): return value - elif isinstance(value, list) and value: + elif unwrap_lists and isinstance(value, list) and value: return value[0] return value @@ -758,7 +771,7 @@ def _parse_multipart_body(body: bytes, boundary: str) -> dict[str, Any]: if header_end == -1: continue - header_section = chunk[:header_end].decode("utf-8") + header_section = _decode_text(chunk[:header_end]) body_section = chunk[header_end + 4 :] # Parse Content-Disposition to get the field name and optional filename @@ -782,7 +795,7 @@ def _parse_multipart_body(body: bytes, boundary: str) -> dict[str, Any]: if filename is not None: value: Any = UploadFile(content=body_section, filename=filename, content_type=content_type_header) else: - value = body_section.decode("utf-8") + value = _decode_text(body_section) # Collect multiple values for same field name into a list if field_name in result: @@ -799,12 +812,36 @@ def _parse_multipart_body(body: bytes, boundary: str) -> dict[str, Any]: def _extract_header_param(header_line: str, param_name: str) -> str | None: """Extract a parameter value from a header line (e.g., name="file" from Content-Disposition).""" - search = f'{param_name}="' - idx = header_line.find(search) - if idx == -1: - return None - start = idx + len(search) - end = header_line.find('"', start) - if end == -1: + # Match the parameter only at the start of a "; "-separated segment, so that + # looking up "name" does not match inside "filename" + match = re.search(rf'(?:^|[;:])\s*{re.escape(param_name)}="([^"]*)"', header_line, re.IGNORECASE) + if match is None: return None - return header_line[start:end] + return match.group(1) + + +def _is_form_content_type(content_type: str) -> bool: + """Whether the content type is parsed into list values (URL-encoded form or multipart).""" + content_type = content_type.strip() + return content_type.startswith((APPLICATION_FORM_CONTENT_TYPE, MULTIPART_FORM_DATA_CONTENT_TYPE)) + + +def _encode_non_base64_body(raw_body: str) -> bytes: + """Convert a multipart body that was not base64-encoded back to bytes.""" + try: + # Use latin-1 to preserve all byte values (0-255) since the body + # may contain raw binary data that isn't valid UTF-8 + return raw_body.encode("latin-1") + except UnicodeEncodeError: + # Characters outside latin-1 mean the body was decoded as UTF-8 text + return raw_body.encode("utf-8") + + +def _decode_text(data: bytes) -> str: + """Decode multipart headers and text fields, falling back to latin-1 for non UTF-8 bytes.""" + try: + return data.decode("utf-8") + except UnicodeDecodeError: + # A non-base64 body encoded back with latin-1 turns UTF-8 text such as "café" + # into single bytes that are not valid UTF-8; latin-1 restores the original text + return data.decode("latin-1") diff --git a/tests/functional/event_handler/_pydantic/test_openapi_comma_parameters.py b/tests/functional/event_handler/_pydantic/test_openapi_comma_parameters.py index cccca599794..8800b887295 100644 --- a/tests/functional/event_handler/_pydantic/test_openapi_comma_parameters.py +++ b/tests/functional/event_handler/_pydantic/test_openapi_comma_parameters.py @@ -341,8 +341,16 @@ def handler(search: str): assert result["statusCode"] == 422 -@pytest.mark.parametrize("content_type", ["application/json", "application/x-www-form-urlencoded"]) -def test_body_scalar_lists_keep_existing_normalization(content_type): +@pytest.mark.parametrize( + ("content_type", "expected_status", "expected_body"), + [ + # JSON arrays are not unwrapped for scalar body fields (#8578) + ("application/json", 422, None), + # Form values are parsed into lists, so the first value is used + ("application/x-www-form-urlencoded", 200, {"search": "first"}), + ], +) +def test_body_scalar_lists_normalization(content_type, expected_status, expected_body): app = APIGatewayRestResolver(enable_validation=True) event = load_event("apiGatewayProxyEvent.json") event.update(path="/search", httpMethod="POST", isBase64Encoded=False) @@ -363,5 +371,6 @@ def handler(search: Annotated[str, param]): result = app.resolve(event, {}) - assert result["statusCode"] == 200 - assert json.loads(result["body"]) == {"search": "first"} + assert result["statusCode"] == expected_status + if expected_body is not None: + assert json.loads(result["body"]) == expected_body diff --git a/tests/functional/event_handler/_pydantic/test_openapi_validation_middleware.py b/tests/functional/event_handler/_pydantic/test_openapi_validation_middleware.py index 0af4cdcfb0e..bfa850d37fe 100644 --- a/tests/functional/event_handler/_pydantic/test_openapi_validation_middleware.py +++ b/tests/functional/event_handler/_pydantic/test_openapi_validation_middleware.py @@ -3812,7 +3812,7 @@ def upload(file_data: Annotated[UploadFile, File(description="A file")]): def test_multipart_missing_boundary(gw_event): - """Test that missing boundary in content-type raises ValueError.""" + """Test that missing boundary in content-type returns 415.""" from aws_lambda_powertools.event_handler.openapi.params import File app = APIGatewayRestResolver(enable_validation=True) @@ -3827,8 +3827,168 @@ def upload(file_data: Annotated[bytes, File()]): gw_event["body"] = base64.b64encode(b"some data").decode() gw_event["isBase64Encoded"] = True - with pytest.raises(ValueError, match="Missing boundary"): - app(gw_event, {}) + result = app(gw_event, {}) + assert result["statusCode"] == 415 + body = json.loads(result["body"]) + assert body["detail"][0]["type"] == "missing_boundary" + + +def test_multipart_invalid_base64_body(gw_event): + """Test that a multipart body that is not valid base64 returns 422 instead of crashing.""" + from aws_lambda_powertools.event_handler.openapi.params import File + + app = APIGatewayRestResolver(enable_validation=True) + + @app.post("/upload") + def upload(file_data: Annotated[bytes, File()]): + return {"size": len(file_data)} + + gw_event["httpMethod"] = "POST" + gw_event["path"] = "/upload" + gw_event["headers"]["content-type"] = "multipart/form-data; boundary=----TestBoundary" + gw_event["body"] = "not-base64!" + gw_event["isBase64Encoded"] = True + + result = app(gw_event, {}) + assert result["statusCode"] == 422 + body = json.loads(result["body"]) + assert body["detail"][0]["type"] == "multipart_invalid" + + +@pytest.mark.parametrize("note", ["café", "İstanbul şehri €"]) +def test_multipart_non_ascii_text_field_non_base64(gw_event, note): + """Test non-ASCII form fields in a non-base64 multipart body (latin-1 and non latin-1 text).""" + from aws_lambda_powertools.event_handler.openapi.params import File + + app = APIGatewayRestResolver(enable_validation=True) + + @app.post("/upload") + def upload(file_data: Annotated[bytes, File()], note: Annotated[str, Form()]): + return {"note": note, "size": len(file_data)} + + boundary = "----TestBoundary" + raw_body = ( + f"--{boundary}\r\n" + f'Content-Disposition: form-data; name="file_data"; filename="a.txt"\r\n' + f"\r\n" + f"hello\r\n" + f"--{boundary}\r\n" + f'Content-Disposition: form-data; name="note"\r\n' + f"\r\n" + f"{note}\r\n" + f"--{boundary}--\r\n" + ) + + gw_event["httpMethod"] = "POST" + gw_event["path"] = "/upload" + gw_event["headers"]["content-type"] = f"multipart/form-data; boundary={boundary}" + gw_event["body"] = raw_body + gw_event["isBase64Encoded"] = False + + with warnings.catch_warnings(record=True): + warnings.simplefilter("always") + result = app(gw_event, {}) + + assert result["statusCode"] == 200 + assert json.loads(result["body"]) == {"note": note, "size": 5} + + +def test_multipart_non_ascii_text_field_base64(gw_event): + """Test a UTF-8 form field in a base64-encoded multipart body.""" + app = APIGatewayRestResolver(enable_validation=True) + + @app.post("/form") + def form(note: Annotated[str, Form()]): + return {"note": note} + + body, content_type = _build_multipart_body([{"name": "note", "value": "café ş"}]) + + gw_event["httpMethod"] = "POST" + gw_event["path"] = "/form" + gw_event["headers"]["content-type"] = content_type + gw_event["body"] = body + gw_event["isBase64Encoded"] = True + + result = app(gw_event, {}) + assert result["statusCode"] == 200 + assert json.loads(result["body"]) == {"note": "café ş"} + + +def test_multipart_filename_before_name(gw_event): + """Test that Content-Disposition is parsed when filename= comes before name=.""" + from aws_lambda_powertools.event_handler.openapi.params import File, UploadFile + + app = APIGatewayRestResolver(enable_validation=True) + + @app.post("/upload") + def upload(file_data: Annotated[UploadFile, File()]): + return {"filename": file_data.filename, "size": len(file_data)} + + boundary = "----TestBoundary" + raw = ( + f"--{boundary}\r\n" + f'Content-Disposition: form-data; filename="a.txt"; name="file_data"\r\n' + f"\r\n" + f"hello\r\n" + f"--{boundary}--\r\n" + ).encode() + + gw_event["httpMethod"] = "POST" + gw_event["path"] = "/upload" + gw_event["headers"]["content-type"] = f"multipart/form-data; boundary={boundary}" + gw_event["body"] = base64.b64encode(raw).decode() + gw_event["isBase64Encoded"] = True + + result = app(gw_event, {}) + assert result["statusCode"] == 200 + assert json.loads(result["body"]) == {"filename": "a.txt", "size": 5} + + +def test_multipart_extract_header_param_does_not_match_inside_filename(): + """Test that looking up name= does not match inside filename=.""" + from aws_lambda_powertools.event_handler.middlewares.openapi_validation import _extract_header_param + + header = 'Content-Disposition: form-data; filename="a.txt"' + assert _extract_header_param(header, "name") is None + assert _extract_header_param(header, "filename") == "a.txt" + + +def test_json_array_for_model_body_returns_422(gw_event): + """Test that a JSON array sent to a parameter declared as an object is rejected.""" + app = APIGatewayRestResolver(enable_validation=True) + + class Item(BaseModel): + name: str + price: float + + @app.post("/items") + def create(item: Item): + return item.model_dump() + + gw_event["httpMethod"] = "POST" + gw_event["path"] = "/items" + gw_event["headers"]["content-type"] = "application/json" + gw_event["body"] = json.dumps([{"name": "a", "price": 1}, {"name": "b", "price": 2}]) + + result = app(gw_event, {}) + assert result["statusCode"] == 422 + + +def test_json_array_for_scalar_body_returns_422(gw_event): + """Test that a JSON array sent to a scalar Body() parameter is rejected.""" + app = APIGatewayRestResolver(enable_validation=True) + + @app.post("/note") + def note(value: Annotated[str, Body()]): + return {"value": value} + + gw_event["httpMethod"] = "POST" + gw_event["path"] = "/note" + gw_event["headers"]["content-type"] = "application/json" + gw_event["body"] = json.dumps(["a", "b"]) + + result = app(gw_event, {}) + assert result["statusCode"] == 422 def test_multipart_quoted_boundary(gw_event): From c222e131986f8182563fee772689fb0a8098b83a Mon Sep 17 00:00:00 2001 From: MTirnova <88968436+MTirnova@users.noreply.github.com> Date: Mon, 19 Oct 2026 23:04:09 +0300 Subject: [PATCH 2/2] fix(event_handler): reject multipart bodies mangled by UTF-8 decoding Without binary media types, API Gateway decodes the body as UTF-8 and replaces invalid bytes with U+FFFD. The UTF-8 fallback then passed a corrupted file on with a 200. Return 422 multipart_invalid instead, with a message pointing at Binary Media Types. --- .../middlewares/openapi_validation.py | 10 ++++- .../test_openapi_validation_middleware.py | 39 +++++++++++++++++++ 2 files changed, 48 insertions(+), 1 deletion(-) diff --git a/aws_lambda_powertools/event_handler/middlewares/openapi_validation.py b/aws_lambda_powertools/event_handler/middlewares/openapi_validation.py index 07a4dad1d17..f0c7ec7a103 100644 --- a/aws_lambda_powertools/event_handler/middlewares/openapi_validation.py +++ b/aws_lambda_powertools/event_handler/middlewares/openapi_validation.py @@ -833,7 +833,15 @@ def _encode_non_base64_body(raw_body: str) -> bytes: # may contain raw binary data that isn't valid UTF-8 return raw_body.encode("latin-1") except UnicodeEncodeError: - # Characters outside latin-1 mean the body was decoded as UTF-8 text + # Characters outside latin-1 mean the body was decoded as UTF-8 text. Without binary media + # types, API Gateway replaces invalid bytes with U+FFFD in that decoding, so a binary upload + # can't be restored: fail instead of passing on a corrupted file. + if "�" in raw_body: + raise ValueError( + "Multipart body contains bytes that were replaced during UTF-8 decoding. " + "If using API Gateway REST API (v1), configure Binary Media Types " + "to include 'multipart/form-data'.", + ) from None return raw_body.encode("utf-8") diff --git a/tests/functional/event_handler/_pydantic/test_openapi_validation_middleware.py b/tests/functional/event_handler/_pydantic/test_openapi_validation_middleware.py index bfa850d37fe..c9257beb075 100644 --- a/tests/functional/event_handler/_pydantic/test_openapi_validation_middleware.py +++ b/tests/functional/event_handler/_pydantic/test_openapi_validation_middleware.py @@ -3893,6 +3893,45 @@ def upload(file_data: Annotated[bytes, File()], note: Annotated[str, Form()]): assert json.loads(result["body"]) == {"note": note, "size": 5} +def test_multipart_non_base64_binary_mangled_by_utf8_decoding(gw_event): + """Test that a binary upload API Gateway mangled while decoding it as UTF-8 returns 422.""" + from aws_lambda_powertools.event_handler.openapi.params import File + + app = APIGatewayRestResolver(enable_validation=True) + + @app.post("/upload") + def upload(file_data: Annotated[bytes, File()]): + return {"size": len(file_data)} + + # Without binary media types, API Gateway decodes the body as UTF-8 and replaces invalid bytes + binary_content = b"\xff\xd8\xff\xe0\x00\x10JFIF\x00\x80\x81" + boundary = "----TestBoundary" + raw_bytes = ( + ( + f"--{boundary}\r\n" + f'Content-Disposition: form-data; name="file_data"; filename="photo.jpg"\r\n' + f"Content-Type: image/jpeg\r\n" + f"\r\n" + ).encode() + + binary_content + + f"\r\n--{boundary}--\r\n".encode() + ) + + gw_event["httpMethod"] = "POST" + gw_event["path"] = "/upload" + gw_event["headers"]["content-type"] = f"multipart/form-data; boundary={boundary}" + gw_event["body"] = raw_bytes.decode("utf-8", errors="replace") + gw_event["isBase64Encoded"] = False + + with warnings.catch_warnings(record=True): + warnings.simplefilter("always") + result = app(gw_event, {}) + + assert result["statusCode"] == 422 + body = json.loads(result["body"]) + assert body["detail"][0]["type"] == "multipart_invalid" + + def test_multipart_non_ascii_text_field_base64(gw_event): """Test a UTF-8 form field in a base64-encoded multipart body.""" app = APIGatewayRestResolver(enable_validation=True)