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 83c8955175)

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

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""
+                            )