From 938fda3ddace3164b3b82bacdbd45f84753fd6de Mon Sep 17 00:00:00 2001 From: ldm0 Date: Thu, 17 Sep 2026 15:29:04 +0800 Subject: [PATCH] fix(wpt): validate form submission fixture bodies Serve the form-submission body checker for GET, HEAD and POST. Preserve its exact URL-encoded and plain-text checks and MIME-aware multipart field parsing, including duplicate fields, file parts and transfer-encoding boundaries. (cherry picked from commit 83c89551755cb18918133d6a9957196e85cb8960) --- .../moli_benchmark/wpt_cross/server.py | 72 +++++++++++++++ .../tests/test_wpt_cross_fixture_responses.py | 92 +++++++++++++++++++ 2 files changed, 164 insertions(+) diff --git a/moli-benchmark/moli_benchmark/wpt_cross/server.py b/moli-benchmark/moli_benchmark/wpt_cross/server.py index ae2c227656..452ece9d7a 100644 --- a/moli-benchmark/moli_benchmark/wpt_cross/server.py +++ b/moli-benchmark/moli_benchmark/wpt_cross/server.py @@ -43,6 +43,9 @@ import time import uuid from collections.abc import Callable, Mapping from html import escape as html_escape +from email import policy +from email.parser import BytesParser + from email.utils import formatdate from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from itertools import product @@ -175,6 +178,9 @@ FETCH_PREFLIGHT_RESOURCE_PATHS = { "/fetch/api/resources/clean-stash.py", } +FORM_SUBMISSION_PATH = ( + "/html/semantics/forms/form-submission-0/resources/form-submission.py" +) BENCH_TIMEOUT_MULTIPLIER_QUERY = "__moli_bench_timeout_multiplier" BENCH_REPORT_BRIDGE_SRC_RE = re.compile( @@ -1202,6 +1208,52 @@ def _xhr_inspect_headers_fixture_response( return headers, b"".join(parts) +def _form_submission_response( + query: str, content_type: str | None, body: bytes +) -> bytes: + """Validate entity bodies like the upstream form-submission.py fixture.""" + params = dict(parse_qsl(query)) + if params.get("query") == "1": + if content_type == "application/x-www-form-urlencoded": + valid = body == b"foo=bara" + elif content_type == "text/plain": + valid = body == b"qux=baz\r\n" + else: + # The upstream fallback compares the first parsed foo field, not + # the raw body. MIME parsing must respect boundaries and headers; + # a matching value elsewhere in the payload is not sufficient. + form_content_type = content_type or "application/x-www-form-urlencoded" + message = BytesParser(policy=policy.HTTP).parsebytes( + f"Content-Type: {form_content_type}\r\n\r\n".encode("latin-1") + body + ) + # WPT's FieldStorage dispatches on the case-sensitive media token. + media_type = form_content_type.partition(";")[0].strip() + valid = False + if media_type == "application/x-www-form-urlencoded": + fields = parse_qsl(body.decode("latin-1"), keep_blank_values=True) + valid = ( + next((value for name, value in fields if name == "foo"), None) + == "bar" + ) + elif ( + media_type == "multipart/form-data" + and message.is_multipart() + ): + for part in message.iter_parts(): + if part.get_param("name", header="content-disposition") == "foo": + # FieldStorage does not decode Content-Transfer-Encoding + # for form fields, and uploaded files are not byte values. + valid = ( + not part.get_filename() and part.get_payload() == "bar" + ) + break + elif "expected_body" in params: + valid = body == params["expected_body"].encode("utf-8") + else: + valid = False + return b"OK" if valid else b"FAIL" + + def _redirect_fixture_response(query: str) -> tuple[int, str] | None: """Return the shared redirect response used by static WPT fixture handlers.""" @@ -2008,6 +2060,17 @@ requestExecutor("{executor_uuid}", {start_on_js}); }: self._serve_fetch_resource_method() return + if path == FORM_SUBMISSION_PATH: + raw = self._read_content_length_request_body() + if raw is not None: + self._send_bytes( + "text/plain", + _form_submission_response( + parsed.query, self.headers.get("Content-Type"), raw + ), + emit_body=True, + ) + return if path == "/xhr/resources/delay.py": if self._consume_request_body(): self._serve_xhr_delay(parsed.query, emit_body=True) @@ -2397,6 +2460,15 @@ requestExecutor("{executor_uuid}", {start_on_js}); if path == JSON_THEN_JS_PATH: self._serve_json_then_js(parsed.query, emit_body=emit_body) return + if path == FORM_SUBMISSION_PATH: + self._send_bytes( + "text/plain", + _form_submission_response( + parsed.query, self.headers.get("Content-Type"), b"" + ), + emit_body=emit_body, + ) + return if path == "/xhr/resources/delay.py": self._serve_xhr_delay(parsed.query, emit_body=emit_body) return diff --git a/moli-benchmark/tests/test_wpt_cross_fixture_responses.py b/moli-benchmark/tests/test_wpt_cross_fixture_responses.py index 8c7a711885..ed207f461e 100644 --- a/moli-benchmark/tests/test_wpt_cross_fixture_responses.py +++ b/moli-benchmark/tests/test_wpt_cross_fixture_responses.py @@ -413,3 +413,95 @@ class WptCrossFixtureResponsesTests(WptCrossTestCase): f"frame-src 'none'; report-uri /report.py?reportID={cookie_value}", ), ) + + def test_fixture_server_validates_form_submission_entity_bodies(self) -> None: + def multipart(*parts: bytes) -> bytes: + return ( + b"--form-boundary\r\n" + + b"\r\n--form-boundary\r\n".join(parts) + + b"\r\n--form-boundary--\r\n" + ) + + foo = b'Content-Disposition: form-data; name="foo"\r\n\r\nbar' + wrong_foo = b'Content-Disposition: form-data; name="foo"\r\n\r\nwrong' + multipart_type = 'multipart/form-data; boundary="form-boundary"' + cases = [ + ("query=1", "application/x-www-form-urlencoded", b"foo=bara", b"OK"), + ("query=1", "application/x-www-form-urlencoded", b"foo=bar", b"FAIL"), + ("query=1", "application/x-www-form-urlencoded", b"foo=ba%72a", b"FAIL"), + ("query=1", "application/x-www-form-urlencoded", b"foo=bara&extra=1", b"FAIL"), + ("query=1", "application/x-www-form-urlencoded; charset=UTF-8", b"foo=bar", b"OK"), + ("query=1", "text/plain", b"qux=baz\r\n", b"OK"), + ("query=1", "text/plain", b"qux=baz\n", b"FAIL"), + ("query=1", multipart_type, multipart(foo), b"OK"), + ("query=1", multipart_type, multipart(foo, wrong_foo), b"OK"), + ("query=1", multipart_type, multipart(wrong_foo, foo), b"FAIL"), + ("query=1", multipart_type, multipart(foo.replace(b'"foo"', b'"other"')), b"FAIL"), + ( + "query=1", + multipart_type, + multipart(foo.replace(b'name="foo"', b'name="foo"; filename="field.txt"')), + b"FAIL", + ), + ( + "query=1", + multipart_type, + multipart(b'Content-Disposition: form-data; name="foo"\r\nContent-Transfer-Encoding: base64\r\n\r\nYmFy'), + b"FAIL", + ), + ("query=1", "multipart/form-data; boundary=wrong", multipart(foo), b"FAIL"), + ("query=1", "multipart/form-data", multipart(foo), b"FAIL"), + ("query=1", "Multipart/Form-Data; boundary=form-boundary", multipart(foo), b"FAIL"), + ( + "expected_body=foo.x%3D0%26foo.y%3D0", + "application/x-www-form-urlencoded", + b"foo.x=0&foo.y=0", + b"OK", + ), + ( + "expected_body=foo.x%3D0%26foo.y%3D0", + "application/x-www-form-urlencoded", + b"foo.x=1&foo.y=0", + b"FAIL", + ), + ("expected_body=%E9%9B%AA", "text/plain", "雪".encode("utf-8"), b"OK"), + ("expected_body=wrong&expected_body=right", "text/plain", b"right", b"OK"), + ( + "query=1&expected_body=wrong", + "application/x-www-form-urlencoded", + b"foo=bara", + b"OK", + ), + ("expected_body=", "text/plain", b"", b"FAIL"), + ("", "application/x-www-form-urlencoded", b"foo=bara", b"FAIL"), + ] + with tempfile.TemporaryDirectory() as root: + root_path = Path(root) + (root_path / "resources").mkdir() + (root_path / "resources" / "testharness.js").write_text("// testharness") + with WptFixtureServer(root_path) as server: + url = ( + f"{server.base_url}/html/semantics/forms/" + "form-submission-0/resources/form-submission.py" + ) + for query, content_type, body, expected in cases: + with self.subTest(query=query, content_type=content_type, body=body): + request = Request( + f"{url}?{query}", + data=body, + headers={"Content-Type": content_type}, + ) + with urlopen(request, timeout=2) as response: + self.assertEqual(response.status, 200) + self.assertEqual(response.headers["Content-Type"], "text/plain") + self.assertEqual(response.read(), expected) + + for method in ("GET", "HEAD"): + with self.subTest(method=method): + with urlopen(Request(url, method=method), timeout=2) as response: + self.assertEqual(response.status, 200) + self.assertEqual(response.headers["Content-Type"], "text/plain") + self.assertEqual(response.headers["Content-Length"], "4") + self.assertEqual( + response.read(), b"FAIL" if method == "GET" else b"" + )