From e032e9fb4f50cec96828212caa88feeee18201a2 Mon Sep 17 00:00:00 2001 From: ldm0 Date: Wed, 23 Sep 2026 14:19:42 +0800 Subject: [PATCH] fix(wpt): honor common redirect CORS opt-in Require enable-cors and a nonempty Origin before exposing CORS response headers from common/redirect.py. Preserve the first status and Location values, handle extension methods, and respond without waiting for unused uploads or adding a Content-Length header. Validation: 580 Python tests passed with 32 workers; 1296 HTTP responses matched official wptserve for status, body, and 15 response fields. New regression tests fail on the unmodified main baseline. Source-commit: 99b528aece8383f352774c159aeaeef4819258a6 --- .../moli_benchmark/wpt_cross/server.py | 43 +++++++- .../tests/test_wpt_common_redirect_server.py | 100 ++++++++++++++++++ 2 files changed, 142 insertions(+), 1 deletion(-) create mode 100644 moli-benchmark/tests/test_wpt_common_redirect_server.py diff --git a/moli-benchmark/moli_benchmark/wpt_cross/server.py b/moli-benchmark/moli_benchmark/wpt_cross/server.py index 5b48ccecbf..815d32d8cb 100644 --- a/moli-benchmark/moli_benchmark/wpt_cross/server.py +++ b/moli-benchmark/moli_benchmark/wpt_cross/server.py @@ -99,6 +99,7 @@ XHR_RESPONSE_RESOURCE_PATHS = { "/xhr/resources/last-modified.py", } COMMON_ECHO_PATH = "/common/echo.py" +COMMON_REDIRECT_PATH = "/common/redirect.py" FETCH_EMPTY_LOCATION_PATH = "/fetch/api/resources/redirect-empty-location.py" FETCH_ABORT_RESOURCE_PATHS = { "/fetch/api/resources/stash-put.py", @@ -1718,6 +1719,8 @@ def _make_handler( def do_OPTIONS(self) -> None: # noqa: N802 if self._serve_common_echo_resource(): return + if self._serve_common_redirect_resource(): + return if self._serve_empty_location_resource(emit_body=self.command != "HEAD"): return if self._serve_xhr_response_resource(): @@ -1752,6 +1755,8 @@ def _make_handler( def do_POST(self) -> None: # noqa: N802 if self._serve_common_echo_resource(): return + if self._serve_common_redirect_resource(): + return if self._serve_empty_location_resource(emit_body=self.command != "HEAD"): return if self._serve_xhr_response_resource(): @@ -1807,6 +1812,8 @@ def _make_handler( def _serve_fetch_resource_method(self) -> None: if self._serve_common_echo_resource(): return + if self._serve_common_redirect_resource(): + return if self._serve_empty_location_resource(emit_body=self.command != "HEAD"): return if self._serve_xhr_response_resource(): @@ -1840,6 +1847,8 @@ def _make_handler( def do_YO(self) -> None: # noqa: N802 (WPT custom method) if self._serve_common_echo_resource(): return + if self._serve_common_redirect_resource(): + return if self._serve_empty_location_resource(emit_body=self.command != "HEAD"): return if unquote(urlparse(self.path).path) in { @@ -2059,6 +2068,8 @@ def _make_handler( def _serve_response(self, *, emit_body: bool) -> None: if self._serve_common_echo_resource(): return + if self._serve_common_redirect_resource(): + return if self._serve_empty_location_resource(emit_body=emit_body): return if self._serve_xhr_response_resource(emit_body=emit_body): @@ -2144,7 +2155,6 @@ def _make_handler( if path in { "/fetch/api/resources/redirect.py", - "/common/redirect.py", "/common/redirect-opt-in.py", }: redirect = _redirect_fixture_response(parsed.query) @@ -2596,6 +2606,8 @@ def _make_handler( def __getattr__(self, name: str): if name.startswith("do_") and unquote(urlsplit(self.path).path) == COMMON_ECHO_PATH: return self._serve_common_echo_resource + if name.startswith("do_") and unquote(urlsplit(self.path).path) == COMMON_REDIRECT_PATH: + return self._serve_common_redirect_resource if name.startswith("do_") and unquote(urlparse(self.path).path) == NAVIGATION_SECOND_VISIT_PATH: return self._serve_navigation_second_visit if name.startswith("do_") and unquote(urlparse(self.path).path) == FETCH_EMPTY_LOCATION_PATH: @@ -2667,6 +2679,35 @@ def _make_handler( self._send_bytes(None, body, emit_body=emit_body, extra_headers=headers, status_code=status, status_text=reason) + def _serve_common_redirect_resource(self) -> bool: + parsed = urlsplit(self.path) + if unquote(parsed.path) != COMMON_REDIRECT_PATH: + return False + # This handler responds without reading uploads, including OPTIONS. + self.close_connection = True + params = parse_qs(parsed.query, keep_blank_values=True, encoding="latin-1") + status = 302 + try: + status = int(params.get("status", ["302"])[0].encode("latin-1")) + except ValueError: + pass + if "location" not in params: + self.send_error(500) + return True + headers = [("Connection", "close"), ("Location", params["location"][0])] + origin = self.headers.get("Origin") + if "enable-cors" in params and origin: + headers.extend([ + ("Content-Type", "text/plain"), + ("Access-Control-Allow-Origin", origin), + ("Access-Control-Allow-Credentials", "true"), + ]) + self.send_response(status) + for name, value in headers: + self.send_header(name, value) + self.end_headers() + return True + def _serve_empty_location_resource(self, *, emit_body: bool = True) -> bool: if unquote(urlparse(self.path).path) != FETCH_EMPTY_LOCATION_PATH: return False diff --git a/moli-benchmark/tests/test_wpt_common_redirect_server.py b/moli-benchmark/tests/test_wpt_common_redirect_server.py new file mode 100644 index 0000000000..770fc07d6b --- /dev/null +++ b/moli-benchmark/tests/test_wpt_common_redirect_server.py @@ -0,0 +1,100 @@ +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 moli_benchmark.wpt_cross.server import WptFixtureServer + + +COMMON_REDIRECT = "/common/redirect.py" + + +class CommonRedirectFixtureTests(unittest.TestCase): + def setUp(self): + 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.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="", *, method="GET", headers=None, body=None, port=None, path=COMMON_REDIRECT): + connection = HTTPConnection("127.0.0.1", port or self.server.port, timeout=2) + try: + connection.request(method, path + "?" + query, body, headers or {}) + response = connection.getresponse() + return response.status, response.headers, response.read() + finally: + connection.close() + + def test_common_redirect_cors_requires_opt_in_and_nonempty_origin(self): + for flag in ("", "&enable-cors", "&enable-cors=false"): + for origin in (None, "", "null", "https://caller.test"): + with self.subTest(flag=flag, origin=origin): + status, headers, body = self.request( + "location=/target" + flag, path=COMMON_REDIRECT, + headers={} if origin is None else {"Origin": origin}, + ) + self.assertEqual((status, body), (302, b"")) + self.assertEqual(headers["Location"], "/target") + allowed = bool(flag and origin) + self.assertEqual(headers["Access-Control-Allow-Origin"], origin if allowed else None) + self.assertEqual(headers["Access-Control-Allow-Credentials"], "true" if allowed else None) + self.assertEqual(headers["Content-Type"], "text/plain" if allowed else None) + self.assertIsNone(headers["Access-Control-Allow-Methods"]) + self.assertIsNone(headers["Cache-Control"]) + self.assertIsNone(headers["Content-Length"]) + + def test_common_redirect_handles_standard_and_extension_methods(self): + for method in ("GET", "HEAD", "OPTIONS", "POST", "PUT", "PATCH", "DELETE", "YO", "chicken"): + with self.subTest(method=method): + status, headers, body = self.request( + "location=/target&status=307", method=method, path=COMMON_REDIRECT, + headers={"Origin": "https://caller.test"}, body=b"unused upload", + ) + self.assertEqual((status, body), (307, b"")) + self.assertEqual(headers["Location"], "/target") + self.assertIsNone(headers["Access-Control-Allow-Origin"]) + self.assertIsNone(headers["Timing-Allow-Origin"]) + + def test_common_redirect_uses_first_parameters_and_preserves_raw_bytes(self): + for query, status, location in [ + ("location=", 302, ""), + ("location=/first&location=/second", 302, "/first"), + ("location=/target&status=200", 200, "/target"), + ("location=/target&status=404", 404, "/target"), + ("location=/target&status=301&status=307", 301, "/target"), + ("location=/target&status=wrong", 302, "/target"), + ("location=/target&status=", 302, "/target"), + ("location=/target&status=%A0307%A0", 302, "/target"), + ("location=/target&redirect_status=307", 302, "/target"), + ("location=/caf%E9", 302, "/café"), + ]: + with self.subTest(query=query): + actual_status, headers, body = self.request(query, path=COMMON_REDIRECT) + self.assertEqual((actual_status, body), (status, b"")) + self.assertEqual(headers.get_all("Location"), [location]) + self.assertEqual(self.request(path=COMMON_REDIRECT)[0], 500) + + def test_common_redirect_responds_before_unused_uploads_finish(self): + for method in ("POST", "OPTIONS"): + connection = HTTPConnection("127.0.0.1", self.server.port, timeout=2) + try: + connection.putrequest(method, COMMON_REDIRECT + "?location=/target&enable-cors") + connection.putheader("Content-Length", "1000000") + connection.putheader("Origin", "https://caller.test") + connection.endheaders() + response = connection.getresponse() + self.assertEqual((response.status, response.read()), (302, b"")) + self.assertEqual(response.headers["Location"], "/target") + self.assertEqual(response.headers["Access-Control-Allow-Origin"], "https://caller.test") + self.assertEqual(response.headers["Connection"], "close") + finally: + connection.close()