fix(wpt): parse fixture response pipes like wptserve

Tokenize pipe commands with wptserve escape, argument, and repeated-query rules. Return HTTP errors for malformed pipes and preserve ordered header operations without response header injection.

Source: 232701464d
(cherry picked from commit 90258a7da8)

moli-benchmark/moli_benchmark/wpt_cross/server.py

@@ -820,9 +820,7 @@ BENCH_TESTDRIVER_VENDOR_BRIDGE = b"""\
 })();
 """
 
-_TRICKLE_PIPE_RE = re.compile(r"(?:^|[|,])trickle\(d([0-9]+(?:\.[0-9]+)?)\)(?:$|[|,])")
-_HEADER_PIPE_RE = re.compile(r"^header\(([^,()]+),([^()]*)\)$")
-_STATUS_PIPE_RE = re.compile(r"^status\(([0-9]{3})\)$")
+_TRICKLE_DELAY_RE = re.compile(r"d([0-9]+(?:\.[0-9]+)?)")
 _GET_TEMPLATE_RE = re.compile(rb"\{\{GET\[([^\]\r\n]+)\]\}\}")
 _REQUEST_TEMPLATE_RE = re.compile(
     rb"\{\{(?:GET\[(?P<query>[^\]\r\n]+)\]|"
@@ -965,10 +963,11 @@ def _pipe_trickle_delay_seconds(query: str) -> float:
     """
 
     delay = 0.0
-    for name, value in parse_qsl(query, keep_blank_values=True):
-        if name != "pipe":
+    for name, args in parse_pipe_commands(query):
+        if name != "trickle":
             continue
-        for match in _TRICKLE_PIPE_RE.finditer(value):
+        match = _TRICKLE_DELAY_RE.fullmatch(args[0])
+        if match is not None:
             delay = max(delay, float(match.group(1)))
     return min(delay, _MAX_TRICKLE_DELAY_SECONDS)
 
@@ -977,37 +976,18 @@ def _pipe_response_header_operations(query: str) -> list[tuple[str, str, bool]]:
     """Parse WPT ``pipe=header(Name,Value[,Append])`` operations."""
 
     operations: list[tuple[str, str, bool]] = []
-    for name, value in parse_qsl(query, keep_blank_values=True):
-        if name != "pipe":
+    for name, args in parse_pipe_commands(query):
+        if name != "header":
             continue
-        for command in value.split("|"):
-            match = _HEADER_PIPE_RE.match(command.strip())
-            if match is None:
-                continue
-            header_name = match.group(1).strip()
-            raw_header_value = match.group(2)
-            header_value_without_append, separator, raw_append = (
-                raw_header_value.rpartition(",")
-            )
-            normalized_append = raw_append.strip().lower()
-            has_append_argument = separator != "" and normalized_append in {
-                "true",
-                "false",
-                "1",
-                "0",
-            }
-            append = has_append_argument and normalized_append in {"true", "1"}
-            if has_append_argument:
-                raw_header_value = header_value_without_append
-            # wptserve writes pipe values byte-for-byte, including encoded
-            # CR/LF used by parser tests. Python's static HTTP server cannot
-            # safely do that, so preserve their whitespace semantics without
-            # allowing a query string to inject another response header.
-            header_value = (
-                raw_header_value.strip().replace("\r", " ").replace("\n", " ")
-            )
-            if _valid_static_response_header(header_name, header_value):
-                operations.append((header_name, header_value, append))
+        header_name, raw_header_value = args[:2]
+        append = len(args) == 3 and args[2].lower() in {"true", "1"}
+        # wptserve writes pipe values byte-for-byte, including encoded
+        # CR/LF used by parser tests. Python's static HTTP server cannot
+        # safely do that, so preserve their whitespace semantics without
+        # allowing a query string to inject another response header.
+        header_value = raw_header_value.replace("\r", " ").replace("\n", " ")
+        if _valid_static_response_header(header_name, header_value):
+            operations.append((header_name, header_value, append))
     return operations
 
 
@@ -1031,14 +1011,9 @@ def _pipe_response_status(query: str) -> int | None:
     """Return a valid WPT ``pipe=status(NNN)`` response status, if present."""
 
     status: int | None = None
-    for name, value in parse_qsl(query, keep_blank_values=True):
-        if name != "pipe":
-            continue
-        for command in value.split("|"):
-            match = _STATUS_PIPE_RE.match(command.strip())
-            if match is None:
-                continue
-            code = int(match.group(1))
+    for name, args in parse_pipe_commands(query):
+        if name == "status":
+            code = int(args[0])
             if 100 <= code <= 599:
                 status = code
     return status

moli-benchmark/tests/test_wpt_cross_fixture_responses.py

@@ -9,7 +9,7 @@ class WptCrossFixtureResponsesTests(WptCrossTestCase):
         self.assertEqual(_pipe_trickle_delay_seconds("pipe=header(X,Y)|trickle(d2.5)"), 2.5)
         self.assertEqual(
             _pipe_trickle_delay_seconds("pipe=trickle(d3)&pipe=trickle(d1)"),
-            3.0,
+            1.0,
         )
         self.assertEqual(_pipe_trickle_delay_seconds("pipe=trickle(d999)"), 10.0)
         self.assertEqual(_pipe_trickle_delay_seconds("notpipe=trickle(d1)"), 0.0)

moli-benchmark/tests/test_wpt_pipes_server.py

@@ -0,0 +1,162 @@
+from __future__ import annotations
+
+import tempfile
+import unittest
+from contextlib import ExitStack
+from http.client import HTTPConnection
+from pathlib import Path
+from unittest.mock import patch
+from urllib.parse import urlencode
+
+from moli_benchmark.wpt_cross.server import WptFixtureServer
+
+
+class WptPipeFixtureTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.stack = ExitStack()
+        self.addCleanup(self.stack.close)
+        self.root = Path(self.stack.enter_context(tempfile.TemporaryDirectory()))
+        (self.root / "resources").mkdir()
+        (self.root / "resources/testharness.js").write_text("// testharness")
+        (self.root / "module.json").write_bytes(b'{"value": 1}')
+        (self.root / "template.txt").write_bytes(b"{{host}}")
+        self.stack.enter_context(patch(
+            "moli_benchmark.wpt_cross.server._global_ipv6_address", return_value=None
+        ))
+        self.server = self.stack.enter_context(WptFixtureServer(self.root))
+
+    def request(self, query: str, *, method: str = "GET", path: str = "/module.json"):
+        connection = HTTPConnection("127.0.0.1", self.server.port, timeout=2)
+        try:
+            connection.request(method, path + "?" + query)
+            response = connection.getresponse()
+            return response.status, response.headers, response.read()
+        finally:
+            connection.close()
+
+    def test_header_values_follow_wptserve_argument_escaping(self) -> None:
+        cases = (
+            ("applic(ation/vnd.api+json", "applic(ation/vnd.api+json"),
+            (r"application/vnd\,api+json", "application/vnd,api+json"),
+            (r"application/vnd\)api+json", "application/vnd)api+json"),
+            (r"a\(b\)c", "a(b)c"),
+            ("text/plain|status(201", "text/plain|status(201"),
+            (r"text/plain\;a=b", "text/plain;a=b"),
+            (r"text/plain; x=a\\b", "text/plain; x=a\\b"),
+            ("text/plain; x=a=b", "text/plain; x=a=b"),
+            ("text/plain; x=%2C", "text/plain; x=%2C"),
+            ("application/café+json", "application/café+json"),
+            ("\u00a0application/json\u00a0", "\u00a0application/json\u00a0"),
+            (r"text/plain; x=a\tb", "text/plain; x=a\tb"),
+        )
+        for value, expected in cases:
+            for method in ("GET", "HEAD"):
+                with self.subTest(value=value, method=method):
+                    status, headers, body = self.request(
+                        urlencode({"pipe": f"header(Content-Type,{value})"}),
+                        method=method,
+                    )
+                    self.assertEqual(status, 200)
+                    self.assertEqual(headers.get_all("Content-Type"), [expected])
+                    self.assertEqual(headers["Content-Length"], "12")
+                    self.assertEqual(body, b"" if method == "HEAD" else b'{"value": 1}')
+
+    def test_malformed_pipes_return_http_errors_instead_of_default_json(self) -> None:
+        commands = (
+            "header(Content-Type,applic)ation/vnd.api+json)",
+            "header(Content-Type,application/vnd)api+json)",
+            "header(Content-Type,app(lic)ation/vnd(api)+json)",
+            "header(Content-Type,applic,ation/vnd.api+json)",
+            "header(Content-Type,application/vnd,api+json)",
+            "header(Content-Type,application/json,true,extra)",
+            "header(Content-Type,application/json, true)",
+            "header(Content-Type)",
+            "header",
+            "header(Content-Type,application/json)|unknown",
+            "header(Content-Type,application/json)| status(201)",
+            "header(Content-Type,application/json)extra",
+            "header(Content-Type,application/json\\",
+            "status(invalid)",
+            "status(201,202)",
+        )
+        for command in commands:
+            for method in ("GET", "HEAD"):
+                with self.subTest(command=command, method=method):
+                    status, _, body = self.request(
+                        urlencode({"pipe": command}), method=method
+                    )
+                    self.assertEqual(status, 500)
+                    self.assertNotEqual(body, b'{"value": 1}')
+                    if method == "HEAD":
+                        self.assertEqual(body, b"")
+
+    def test_non_latin1_headers_return_http_errors_without_aborting(self) -> None:
+        for value in ("申请/vnd.api+json", "application/vnd.api🚀+json", "app™/json"):
+            for method in ("GET", "HEAD"):
+                with self.subTest(value=value, method=method):
+                    status, _, body = self.request(
+                        urlencode({"pipe": f"header(Content-Type,{value})"}),
+                        method=method,
+                    )
+                    self.assertEqual(status, 500)
+                    if method == "HEAD":
+                        self.assertEqual(body, b"")
+
+    def test_header_append_and_replacement_keep_command_order(self) -> None:
+        (self.root / "module.json.headers").write_text("X-Test: sidecar\n")
+        commands = (
+            ("header(X-Test,one,TrUe)|header(x-test,two,1)", ["sidecar", "one", "two"]),
+            ("header(X-Test,one,true)|header(x-test,two,0)", ["two"]),
+            ("header(X-Test,one,false)|header(x-test,two,true)", ["one", "two"]),
+            ("header(X-Test,one)header(x-test,two,true)", ["one", "two"]),
+            ("header(X-Test,one", ["one"]),
+            (r"header(X-Test,one\,true)", ["one,true"]),
+            ("header(X-Test,)", [""]),
+        )
+        for command, expected in commands:
+            with self.subTest(command=command):
+                status, headers, _ = self.request(urlencode({"pipe": command}))
+                self.assertEqual(status, 200)
+                self.assertEqual(headers.get_all("X-Test"), expected)
+
+    def test_only_last_nonempty_pipe_parameter_is_applied(self) -> None:
+        cases = (
+            (["unknown", "header(X-Test,last)"], 200, "last"),
+            (["header(X-Test,first)", "status(201)"], 201, None),
+            (["header(X-Test,first)", ""], 200, "first"),
+            (["header(X-Test,first)", "unknown"], 500, None),
+        )
+        for commands, expected_status, expected_header in cases:
+            with self.subTest(commands=commands):
+                status, headers, _ = self.request(urlencode([
+                    ("pipe", command) for command in commands
+                ]))
+                self.assertEqual(status, expected_status)
+                self.assertEqual(headers.get("X-Test"), expected_header)
+
+    def test_pipes_embedded_in_values_do_not_change_the_response(self) -> None:
+        command = r"header(X-Test,first|sub|trickle(d3\)|last)|status(201)"
+        with patch("moli_benchmark.wpt_cross.server.time.sleep") as sleep:
+            status, headers, body = self.request(
+                urlencode({"pipe": command}), path="/template.txt"
+            )
+        self.assertEqual(status, 201)
+        self.assertEqual(headers["X-Test"], "first|sub|trickle(d3)|last")
+        self.assertEqual(body, b"{{host}}")
+        sleep.assert_not_called()
+        status, _, body = self.request("pipe=sub(none)", path="/template.txt")
+        self.assertEqual(status, 200)
+        self.assertEqual(body, b"localhost")
+
+    def test_header_newlines_cannot_inject_response_headers(self) -> None:
+        for value in ("before\r\nInjected: value", r"before\r\nInjected: value"):
+            with self.subTest(value=value):
+                status, headers, _ = self.request(urlencode({"pipe": f"header(X-Test,{value})"}))
+                self.assertEqual(status, 200)
+                self.assertEqual(headers["X-Test"], "before  Injected: value")
+                self.assertIsNone(headers["Injected"])
+
+
+
+if __name__ == "__main__":
+    unittest.main()