From 690d80dbeddb96819cd066dbcf9e1eb7089a5a00 Mon Sep 17 00:00:00 2001 From: ldm0 Date: Thu, 17 Sep 2026 15:34:58 +0800 Subject: [PATCH] fix(wpt): serve XHR PUT and wildcard preflight fixtures Model the XHR PUT CORS grant and wildcard request-header preflight resources. Preserve method-specific grants, raw PUT bytes, missing-Origin failures and responses without Content-Type; cover exact paths and discovery. --- .../moli_benchmark/wpt_cross/case_set.py | 4 +- .../moli_benchmark/wpt_cross/server.py | 53 ++++++++ .../tests/test_wpt_xhr_cors_fixture.py | 123 ++++++++++++++++++ 3 files changed, 179 insertions(+), 1 deletion(-) diff --git a/moli-benchmark/moli_benchmark/wpt_cross/case_set.py b/moli-benchmark/moli_benchmark/wpt_cross/case_set.py index 6214837c6b..cc508cd94c 100644 --- a/moli-benchmark/moli_benchmark/wpt_cross/case_set.py +++ b/moli-benchmark/moli_benchmark/wpt_cross/case_set.py @@ -853,7 +853,9 @@ def _xhr_response_handler_reference_patterns(directory: str) -> tuple[re.Pattern "inspect-headers.py", "echo-headers.py", "content.py", "echo-content-type.py", "win-1252-xml.py", "win-1252-html.py", "invalid-utf8-html.py", "shift-jis-html.py", "img-utf8-html.py", - "empty-div-utf8-html.py"): + "empty-div-utf8-html.py", + "access-control-basic-put-allow.py", + "access-control-preflight-request-allow-headers-returns-star.py"): resource = f"xhr/resources/{name}" relative = posixpath.relpath(resource, directory) references.extend(("/" + resource, relative, "./" + relative)) diff --git a/moli-benchmark/moli_benchmark/wpt_cross/server.py b/moli-benchmark/moli_benchmark/wpt_cross/server.py index 5b48ccecbf..f73d750549 100644 --- a/moli-benchmark/moli_benchmark/wpt_cross/server.py +++ b/moli-benchmark/moli_benchmark/wpt_cross/server.py @@ -94,6 +94,8 @@ XHR_RESPONSE_RESOURCE_PATHS = { *XHR_BODY_RESOURCE_PATHS, "/xhr/resources/inspect-headers.py", "/xhr/resources/echo-headers.py", + "/xhr/resources/access-control-basic-put-allow.py", + "/xhr/resources/access-control-preflight-request-allow-headers-returns-star.py", "/xhr/resources/corsenabled.py", "/xhr/resources/status.py", "/xhr/resources/last-modified.py", @@ -1844,6 +1846,8 @@ def _make_handler( return if unquote(urlparse(self.path).path) in { "/xhr/resources/inspect-headers.py", "/xhr/resources/echo-headers.py", + "/xhr/resources/access-control-basic-put-allow.py", + "/xhr/resources/access-control-preflight-request-allow-headers-returns-star.py", } | XHR_BODY_RESOURCE_PATHS: self._serve_xhr_response_resource() return @@ -2727,6 +2731,49 @@ def _make_handler( ), ) + def _serve_xhr_preflight_fixture(self, parsed, *, emit_body: bool) -> None: + status, headers, body = 200, [], b"" + try: + if unquote(parsed.path) == "/xhr/resources/access-control-basic-put-allow.py": + headers = [("Content-Type", "text/plain")] + if self.command in {"OPTIONS", "PUT"}: + origin = self.headers.get("Origin") + if origin is None: + raise ValueError("upstream handler requires Origin") + headers.extend([ + ("Access-Control-Allow-Credentials", "true"), + ("Access-Control-Allow-Origin", origin), + ]) + if self.command == "OPTIONS": + headers.append(("Access-Control-Allow-Methods", "PUT")) + else: + request_body = self._read_content_length_request_body() + if request_body is None: + return + body = b"PASS: Cross-domain access allowed.\n" + request_body + else: + body = b"Wrong method: " + self.command.encode("latin-1") + elif self.command == "OPTIONS": + headers = [ + ("Access-Control-Allow-Origin", "*"), + ("Access-Control-Allow-Headers", "*"), + ] + elif self.command == "GET": + headers = [("Access-Control-Allow-Origin", "*")] + if self.headers.get("X-Test"): + headers.append(("Content-Type", "text/plain")) + body = b"PASS" + else: + status = 400 + except ValueError: + self.send_error(500) + return + self.close_connection = True + headers.append(("Connection", "close")) + self._send_bytes(None, body, emit_body=emit_body, extra_headers=headers, + status_code=status, cache_control=None) + + def _serve_xhr_response_resource(self, *, emit_body: bool = True) -> bool: parsed = urlparse(self.path) path = unquote(parsed.path) @@ -2735,6 +2782,12 @@ def _make_handler( if path in XHR_BODY_RESOURCE_PATHS: self._serve_xhr_body_resource(parsed, emit_body=emit_body) return True + if path in { + "/xhr/resources/access-control-basic-put-allow.py", + "/xhr/resources/access-control-preflight-request-allow-headers-returns-star.py", + }: + self._serve_xhr_preflight_fixture(parsed, emit_body=emit_body) + return True if path == "/xhr/resources/corsenabled.py": self._serve_xhr_cors_echo(parsed, emit_body=emit_body) return True diff --git a/moli-benchmark/tests/test_wpt_xhr_cors_fixture.py b/moli-benchmark/tests/test_wpt_xhr_cors_fixture.py index 347b34bdf9..d7caca4927 100644 --- a/moli-benchmark/tests/test_wpt_xhr_cors_fixture.py +++ b/moli-benchmark/tests/test_wpt_xhr_cors_fixture.py @@ -114,3 +114,126 @@ class XhrCorsFixtureTests(unittest.TestCase): self.assertEqual([case.case_path for case in selected], [ "xhr/absolute.html", "xhr/nested/relative.html", "xhr/relative.html", ]) + + def test_put_preflight_and_upload_preserve_origin_and_bytes(self) -> None: + server = self.stack.enter_context(WptFixtureServer(self.root)) + for origin in ("https://caller.test", "null", ""): + with self.subTest(origin=origin): + status, headers, body = self.request( + server, "OPTIONS", headers={"Origin": origin}, resource=PUT_RESOURCE + ) + self.assertEqual((status, body), (200, b"")) + self.assertEqual(headers["Content-Type"], "text/plain") + self.assertEqual(headers["Access-Control-Allow-Origin"], origin) + self.assertEqual(headers["Access-Control-Allow-Credentials"], "true") + self.assertEqual(headers["Access-Control-Allow-Methods"], "PUT") + self.assertIsNone(headers["Access-Control-Allow-Headers"]) + payload = b"PUT \x00\xff body" + status, headers, body = self.request( + server, "PUT", body=payload, headers={"Origin": origin}, + resource=PUT_RESOURCE, + ) + self.assertEqual((status, body), + (200, b"PASS: Cross-domain access allowed.\n" + payload)) + self.assertEqual(headers["Content-Type"], "text/plain") + self.assertEqual(headers["Access-Control-Allow-Origin"], origin) + self.assertEqual(headers["Access-Control-Allow-Credentials"], "true") + self.assertIsNone(headers["Access-Control-Allow-Methods"]) + + + def test_put_fixture_missing_origin_matches_upstream_failure(self) -> None: + server = self.stack.enter_context(WptFixtureServer(self.root)) + for method in ("OPTIONS", "PUT"): + with self.subTest(method=method): + status, headers, _ = self.request(server, method, resource=PUT_RESOURCE) + self.assertEqual(status, 500) + self.assertIsNone(headers["Access-Control-Allow-Origin"]) + + + def test_put_fixture_reports_other_methods_without_cors_headers(self) -> None: + server = self.stack.enter_context(WptFixtureServer(self.root)) + for method in ("GET", "HEAD", "POST", "DELETE", "FOO"): + with self.subTest(method=method): + status, headers, body = self.request(server, method, resource=PUT_RESOURCE) + expected = b"Wrong method: " + method.encode("ascii") + self.assertEqual((status, body), + (200, b"" if method == "HEAD" else expected)) + self.assertEqual(headers["Content-Length"], str(len(expected))) + self.assertIsNone(headers["Access-Control-Allow-Origin"]) + + + def test_wildcard_preflight_and_actual_request_have_distinct_headers(self) -> None: + server = self.stack.enter_context(WptFixtureServer(self.root)) + status, headers, body = self.request(server, "OPTIONS", resource=STAR_RESOURCE) + self.assertEqual((status, body), (200, b"")) + self.assertEqual(headers["Access-Control-Allow-Origin"], "*") + self.assertEqual(headers["Access-Control-Allow-Headers"], "*") + self.assertIsNone(headers["Access-Control-Allow-Methods"]) + self.assertIsNone(headers["Access-Control-Allow-Credentials"]) + self.assertIsNone(headers["Content-Type"]) + for value in ("foobar", "0"): + with self.subTest(value=value): + status, headers, body = self.request( + server, headers={"x-test": value}, resource=STAR_RESOURCE + ) + self.assertEqual((status, body), (200, b"PASS")) + self.assertEqual(headers["Access-Control-Allow-Origin"], "*") + self.assertEqual(headers["Content-Type"], "text/plain") + self.assertIsNone(headers["Access-Control-Allow-Headers"]) + + + def test_wildcard_fixture_rejects_missing_or_empty_test_header(self) -> None: + server = self.stack.enter_context(WptFixtureServer(self.root)) + for request_headers in ({}, {"X-Test": ""}): + with self.subTest(headers=request_headers): + status, headers, body = self.request( + server, headers=request_headers, resource=STAR_RESOURCE + ) + self.assertEqual((status, body), (400, b"")) + self.assertEqual(headers["Access-Control-Allow-Origin"], "*") + self.assertIsNone(headers["Content-Type"]) + + + def test_wildcard_fixture_other_methods_have_no_cors_grants(self) -> None: + server = self.stack.enter_context(WptFixtureServer(self.root)) + for method in ("HEAD", "POST", "PUT", "DELETE", "FOO"): + with self.subTest(method=method): + status, headers, body = self.request( + server, method, headers={"X-Test": "foobar"}, resource=STAR_RESOURCE + ) + self.assertEqual((status, body), (200, b"")) + self.assertIsNone(headers["Access-Control-Allow-Origin"]) + self.assertIsNone(headers["Content-Type"]) + + + def test_preflight_fixtures_require_exact_resource_paths(self) -> None: + server = self.stack.enter_context(WptFixtureServer(self.root)) + for resource in (PUT_RESOURCE, STAR_RESOURCE): + for wrong in (resource + "2", resource + ".js", "/wrong" + resource): + with self.subTest(path=wrong): + status, _, _ = self.request(server, resource=wrong) + self.assertEqual(status, 404) + + + def test_preflight_case_selection_rejects_unsupported_dependencies(self) -> None: + for resource in (PUT_RESOURCE, STAR_RESOURCE): + name = Path(resource).name + sources = { + f"xhr/absolute-{name}.html": f"fetch('{resource}');", + f"xhr/relative-{name}.html": f"fetch('resources/{name}');", + f"xhr/nested/relative-{name}.html": f"fetch('../resources/{name}?x=1');", + f"xhr/unknown-{name}.html": f"fetch('{resource}'); fetch('unknown.py');", + f"xhr/wrong-{name}.html": f"fetch('{name}');", + } + for path, script in sources.items(): + target = self.root / path + target.parent.mkdir(parents=True, exist_ok=True) + target.write_text('') + selected = enumerate_cases(self.root, dir_prefixes=("xhr",)) + expected = sorted( + f"xhr/{prefix}{Path(resource).name}.html" + for resource in (PUT_RESOURCE, STAR_RESOURCE) + for prefix in ("absolute-", "relative-", "nested/relative-") + ) + self.assertEqual([case.case_path for case in selected], expected)