starred/moli
1
0
Fork 0
mirror of https://github.com/lexmount/moli.git synced 2026-10-06 16:01:00 +00:00
Code Issues Packages Projects Releases Wiki Activity

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

Browse Source
This commit is contained in:
ldm0
2026-09-13 17:13:54 +08:00
parent d1034b0f62
commit cc32d97e57
2 changed files with 90 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
moli-benchmark/moli_benchmark/wpt_cross/server.py
+25 -8
View File
@@ -105,6 +105,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",
}
@@ -2311,10 +2312,7 @@ def _make_handler(
status_code=status_code,
)
return
if path in {
"/common/redirect.py",
"/common/redirect-opt-in.py",
}:
if path == "/common/redirect-opt-in.py":
redirect = _redirect_fixture_response(parsed.query)
if redirect is None:
self.send_error(400)
@@ -2324,11 +2322,8 @@ def _make_handler(
("Cache-Control", "no-cache"),
("Pragma", "no-cache"),
("Location", location),
("Timing-Allow-Origin", "*"),
]
if path == "/common/redirect-opt-in.py":
redirect_headers.append(("Timing-Allow-Origin", "*"))
else:
redirect_headers.append(("Access-Control-Allow-Origin", "*"))
self._send_bytes(
"text/plain; charset=utf-8",
b"",
@@ -3013,6 +3008,28 @@ def _make_handler(
)
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
+65
View File
@@ -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")):