fix(wpt): honor CORS opt-in for common redirects

moli-benchmark/moli_benchmark/wpt_cross/server.py

@@ -137,6 +137,7 @@ FETCH_PREFLIGHT_RESOURCE_PATHS = {
     "/fetch/api/resources/clean-stash.py",
 }
 FETCH_REDIRECT_RESOURCE_PATHS = {
+    "/common/redirect.py",
     "/fetch/api/resources/redirect.py",
     "/fetch/api/resources/redirect-empty-location.py",
 }
@@ -3442,6 +3443,28 @@ requestExecutor("{executor_uuid}", {start_on_js});
                 )
                 return
             params = parse_qs(query, keep_blank_values=True, encoding="latin-1")
+            if unquote(urlsplit(self.path).path) == "/common/redirect.py":
+                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
+                headers = [*connection_headers, ("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_bytes(
+                    None, b"", emit_body=emit_body, status_code=status,
+                    extra_headers=headers, cache_control=None,
+                )
+                return
             stash_path = urlsplit(self.path).path
             headers = [*connection_headers, ("Content-Type", "text/plain"), ("Pragma", "no-cache")]
             if "Origin" in self.headers:

moli-benchmark/tests/test_wpt_fetch_redirect_server.py

@@ -17,6 +17,7 @@ from moli_benchmark.wpt_cross.server import WptFixtureServer
 
 REDIRECT = "/fetch/api/resources/redirect.py"
 EMPTY_LOCATION = "/fetch/api/resources/redirect-empty-location.py"
+COMMON_REDIRECT = "/common/redirect.py"
 
 
 class FetchRedirectFixtureTests(unittest.TestCase):
@@ -225,6 +226,70 @@ class FetchRedirectFixtureTests(unittest.TestCase):
             with self.subTest(path=path):
                 self.assertEqual(self.request(path=path)[0], 404)
 
+    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"])
+
+    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()
+
     def test_empty_location_responds_before_unused_uploads_finish(self):
         for method in ("POST", "PUT", "OPTIONS"):
             for framing in (("Content-Length", "1000000"), ("Transfer-Encoding", "chunked")):