From e391d0924fbbabc47b23d31a7cbe799284dc07fc Mon Sep 17 00:00:00 2001 From: ldm0 Date: Thu, 17 Sep 2026 00:31:24 +0800 Subject: [PATCH] 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: 232701464d25c5bd2e4728672177a3c5f23b6721 (cherry picked from commit 90258a7da8d985d2968371f02745b80142b21902) --- .../moli_benchmark/wpt_cross/server.py | 63 ++----- .../tests/test_wpt_cross_fixture_responses.py | 2 +- moli-benchmark/tests/test_wpt_pipes_server.py | 162 ++++++++++++++++++ 3 files changed, 182 insertions(+), 45 deletions(-) create mode 100644 moli-benchmark/tests/test_wpt_pipes_server.py diff --git a/moli-benchmark/moli_benchmark/wpt_cross/server.py b/moli-benchmark/moli_benchmark/wpt_cross/server.py index 4b9dddaace..fbb8abd4ea 100644 --- a/moli-benchmark/moli_benchmark/wpt_cross/server.py +++ b/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[^\]\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 diff --git a/moli-benchmark/tests/test_wpt_cross_fixture_responses.py b/moli-benchmark/tests/test_wpt_cross_fixture_responses.py index abf2a0a1cf..8c7a711885 100644 --- a/moli-benchmark/tests/test_wpt_cross_fixture_responses.py +++ b/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) diff --git a/moli-benchmark/tests/test_wpt_pipes_server.py b/moli-benchmark/tests/test_wpt_pipes_server.py new file mode 100644 index 0000000000..5ce8f2508b --- /dev/null +++ b/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()