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: 99b528aece

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

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()