fix(benchmark): route WPT permission changes through CDP

Forward permission descriptors and states through the native automation
connection using the selected frame's origin pair and case browser context.
Verify iframe owner identity, reject stale or unavailable frames, and remove
the previous storage-access no-op.

Reset permission overrides during case cleanup and still dispose targets
when reset fails. Accept close races only after confirming the target is gone.

Validation: 613 Python tests; 63 WPT pages with no regressions (Moli 227/238,
Chromium 236/238); real permission, iframe and default-context cleanup probes.

moli-benchmark/moli_benchmark/wpt_cross/native_input.py

@@ -1,8 +1,10 @@
-"""Trusted WPT input, served on a separate CDP connection from harness probes.
+"""Native WPT automation on a separate CDP connection from harness probes.
 
 The input connection must remain runnable while the harness connection awaits a
 JavaScript promise. Each instance belongs to one case target and is disposed
 with it; no pressed keys, pending bindings, or pointer state cross case borders.
+Permission requests use frames exposed by the attached page session; out-of-
+process iframe targets require a separate attachment and report an error here.
 """
 
 from __future__ import annotations
@@ -14,6 +16,7 @@ import math
 import time
 from dataclasses import dataclass, field
 from typing import Any
+from urllib.parse import urlsplit
 
 from ..raw_cdp import (
     RawCdpConnectionClosed,
@@ -174,22 +177,42 @@ def validate_actions(sources: Any) -> None:
                 raise ValueError("native pointer geometry properties are not supported")
 
 
+def permission_origin(frame: dict[str, Any]) -> str:
+    # Frame.securityOrigin preserves inherited about:blank origins and opaque
+    # sandbox origins, which cannot be reconstructed from location.href.
+    origin = frame.get("securityOrigin", frame.get("url"))
+    if not isinstance(origin, str):
+        raise ValueError("permission frame has no origin")
+    parsed = urlsplit(origin)
+    if parsed.scheme not in {"http", "https"} or not parsed.hostname or parsed.username or parsed.password:
+        raise ValueError("cannot set permission for an opaque or unsupported origin")
+    return f"{parsed.scheme}://{parsed.netloc}"
+
+
 class NativeInput:
-    def __init__(self, client: RoutedRawCdpClient, session_id: str) -> None:
+    def __init__(
+        self, client: RoutedRawCdpClient, session_id: str,
+        browser_context_id: str | None = None,
+    ) -> None:
         self.client = client
         self.session_id = session_id
+        self.browser_context_id = browser_context_id
         self.deadline: float | None = None
         self.task: asyncio.Task[None] | None = None
         self.keyboards: dict[str, dict[str, Key]] = {}
         self.pointers: dict[str, Pointer] = {}
         self._closed = False
+        self._permissions_changed = False
+        self._permission_cleanup_error: Exception | None = None
 
     @classmethod
-    async def attach(cls, endpoint: str, target_id: str) -> NativeInput:
+    async def attach(
+        cls, endpoint: str, target_id: str, browser_context_id: str | None = None,
+    ) -> NativeInput:
         client = await connect_routed_raw_cdp(endpoint)
         try:
             result = await client.command("Target.attachToTarget", {"targetId": target_id, "flatten": True})
-            instance = cls(client, result.response["result"]["sessionId"])
+            instance = cls(client, result.response["result"]["sessionId"], browser_context_id)
             await instance.command("Runtime.enable")
             await instance.command("Runtime.addBinding", {"name": BINDING_NAME})
             instance.task = asyncio.create_task(instance.run(), name="wpt-native-input")
@@ -200,12 +223,20 @@ class NativeInput:
 
     async def close(self) -> None:
         if self._closed:
+            if self._permission_cleanup_error is not None:
+                raise RawCdpError("failed to reset WPT permissions") from self._permission_cleanup_error
             return
         self._closed = True
         if self.task is not None:
             self.task.cancel()
             with contextlib.suppress(asyncio.CancelledError, Exception):
                 await self.task
+        if self._permissions_changed:
+            params = {"browserContextId": self.browser_context_id} if self.browser_context_id else {}
+            try:
+                await self.client.command("Browser.resetPermissions", params, timeout=5)
+            except Exception as error:
+                self._permission_cleanup_error = error
         # The engine can exit before cleanup. Still detach when possible and
         # always close our own receiver and socket, even after a send fails.
         with contextlib.suppress(Exception):
@@ -218,6 +249,8 @@ class NativeInput:
                 "Target.detachFromTarget", {"sessionId": self.session_id}, timeout=5,
             )
         await self.client.close()
+        if self._permission_cleanup_error is not None:
+            raise RawCdpError("failed to reset WPT permissions") from self._permission_cleanup_error
 
     async def command(self, method: str, params: dict[str, Any] | None = None) -> dict[str, Any]:
         timeout = 10.0 if self.deadline is None else max(0.01, self.deadline - time.perf_counter())
@@ -392,7 +425,9 @@ class NativeInput:
         request_id = request["id"]
         token = request["token"]
         kind = request["kind"]
-        if kind == "send_keys":
+        if kind == "set_permission":
+            await self.set_permission(context_id, request)
+        elif kind == "send_keys":
             await self.evaluate(context_id, f"{STATE_NAME}.focus({request_id}, {json.dumps(token)})")
             await self.send_keys(request["keys"])
         elif kind == "click":
@@ -435,6 +470,73 @@ class NativeInput:
         else:
             raise ValueError(f"unsupported native input request: {kind}")
 
+    async def set_permission(self, context_id: int, request: dict[str, Any]) -> None:
+        descriptor = request.get("descriptor")
+        state = request.get("state")
+        if not isinstance(descriptor, dict) or not isinstance(descriptor.get("name"), str) or not descriptor["name"]:
+            raise ValueError("permission descriptor requires a name")
+        if not isinstance(state, str) or state not in {"granted", "denied", "prompt"}:
+            raise ValueError("invalid permission state")
+        path = await self.evaluate(
+            context_id,
+            f"{STATE_NAME}.permissionFramePath({request['id']}, {json.dumps(request['token'])})",
+        )
+        if not isinstance(path, list) or any(type(index) is not int or index < 0 for index in path):
+            raise ValueError("invalid permission frame path")
+        tree = (await self.command("Page.getFrameTree")).get("frameTree")
+        if not isinstance(tree, dict) or not isinstance(tree.get("frame"), dict):
+            raise ValueError("permission target has no frame tree")
+        embedding_origin = permission_origin(tree["frame"])
+        for index in path:
+            children = tree.get("childFrames", [])
+            if not isinstance(children, list) or index >= len(children):
+                raise ValueError("permission frame is unavailable in this CDP session (detached or out of process)")
+            tree = children[index]
+        embedded_origin = permission_origin(tree["frame"])
+        if path:
+            # A session's frame tree can omit out-of-process frames. Verify the
+            # owner instead of letting an omitted sibling shift the indices.
+            owner = await self.command("DOM.getFrameOwner", {"frameId": tree["frame"]["id"]})
+            resolved = await self.command("DOM.resolveNode", {
+                "backendNodeId": owner["backendNodeId"], "executionContextId": context_id,
+            })
+            object_id = resolved["object"]["objectId"]
+            try:
+                matches = await self.command("Runtime.callFunctionOn", {
+                    "executionContextId": context_id,
+                    "functionDeclaration": (
+                        "function(owner, id, token) { return "
+                        f"{STATE_NAME}.permissionFrameMatches(id, token, owner); }}"
+                    ),
+                    "arguments": [
+                        {"objectId": object_id}, {"value": request["id"]}, {"value": request["token"]},
+                    ],
+                    "returnByValue": True,
+                })
+                if matches.get("exceptionDetails") or matches.get("result", {}).get("value") is not True:
+                    raise ValueError("permission frame does not match the requested Window")
+            finally:
+                await self.command("Runtime.releaseObject", {"objectId": object_id})
+        # Resolve the retained Window again after the protocol read, so a
+        # detached or moved frame cannot grant a sibling's origin by index.
+        current_path = await self.evaluate(
+            context_id,
+            f"{STATE_NAME}.permissionFramePath({request['id']}, {json.dumps(request['token'])})",
+        )
+        if current_path != path:
+            raise ValueError("permission frame changed during the request")
+        params = {
+            "permission": descriptor, "setting": state,
+            "origin": embedding_origin, "embeddedOrigin": embedded_origin,
+        }
+        if self.browser_context_id is not None:
+            params["browserContextId"] = self.browser_context_id
+        # A command can be applied even if its response is lost. Always reset
+        # after attempting a mutation, and never attach a page sessionId to it.
+        self._permissions_changed = True
+        timeout = 10.0 if self.deadline is None else max(0.01, self.deadline - time.perf_counter())
+        await self.client.command("Browser.setPermission", params, timeout=timeout)
+
     async def action(
         self, context_id: int, request_id: int, token: str,
         source: dict[str, Any], action: dict[str, Any],

moli-benchmark/moli_benchmark/wpt_cross/runner.py

@@ -261,7 +261,14 @@ def _target_ids(infos: list[dict[str, Any]]) -> frozenset[str]:
 
 async def _close_target(client: RawCdpClient, target_id: str) -> None:
     command_id = await client.send("Target.closeTarget", {"targetId": target_id})
-    response, _ = await client.recv_until_id(command_id, timeout=5)
+    try:
+        response, _ = await client.recv_until_id(command_id, timeout=5)
+    except RawCdpError:
+        # Closing a parent or disposing a context can remove a target between
+        # getTargets and closeTarget. Only ignore the error if it is gone.
+        if target_id not in _target_ids(await _target_infos(client)):
+            return
+        raise
     success = (response.get("result") or {}).get("success")
     if success is False:
         raise RawCdpError(f"Target.closeTarget rejected target {target_id}")
@@ -270,8 +277,12 @@ async def _close_target(client: RawCdpClient, target_id: str) -> None:
 async def _close_page(client: RawCdpClient, page: _AttachedPage) -> None:
     """Dispose one case's storage context and every target created inside it."""
 
+    input_cleanup_error = None
     if page.native_input is not None:
-        await page.native_input.close()
+        try:
+            await page.native_input.close()
+        except Exception as error:
+            input_cleanup_error = error
 
     try:
         before = await _target_infos(client)
@@ -305,6 +316,8 @@ async def _close_page(client: RawCdpClient, page: _AttachedPage) -> None:
     try:
         after = await _target_infos(client)
     except (RawCdpError, asyncio.TimeoutError):
+        if input_cleanup_error is not None and page.browser_context_id is None:
+            raise input_cleanup_error
         return
 
     residual_ids = {
@@ -339,6 +352,8 @@ async def _close_page(client: RawCdpClient, page: _AttachedPage) -> None:
             raise RawCdpError(
                 f"case cleanup left auxiliary targets alive: {sorted(leaked)}"
             )
+    if input_cleanup_error is not None and page.browser_context_id is None:
+        raise input_cleanup_error
 
 
 async def _attach_page(
@@ -415,7 +430,7 @@ async def _attach_page(
             target_id=target,
             session_id=session_id,
             baseline_target_ids=baseline_target_ids,
-            native_input=await NativeInput.attach(input_endpoint, target) if input_endpoint else None,
+            native_input=await NativeInput.attach(input_endpoint, target, browser_context_id) if input_endpoint else None,
         )
     except BaseException:
         if target is not None:

moli-benchmark/moli_benchmark/wpt_cross/server.py

@@ -594,12 +594,6 @@ BENCH_TESTDRIVER_VENDOR_BRIDGE = (
   }
   window.test_driver_internal.in_automation = true;
   window.test_driver_internal.get_computed_label = getComputedLabel;
-  window.test_driver_internal.set_permission = async function(params) {
-    if (params && params.descriptor && params.descriptor.name === 'storage-access') {
-      return;
-    }
-    throw new Error("set_permission() is not implemented by the Moli WPT bridge");
-  };
 })();
 """
 )

moli-benchmark/moli_benchmark/wpt_cross/testdriver_input.js

@@ -62,6 +62,25 @@
       if (error !== null) item.reject(new Error(error));
       else item.resolve();
     },
+    permissionFramePath(id, expectedToken) {
+      let context = entry(id, expectedToken).context;
+      if (!context || context.closed || context.top !== window.top) {
+        throw new Error('Permission context is not in the current test target');
+      }
+      const path = [];
+      while (context !== window.top) {
+        const parent = context.parent;
+        let index = 0;
+        while (index < parent.length && parent[index] !== context) index++;
+        if (index === parent.length) throw new Error('Permission frame is no longer attached');
+        path.unshift(index);
+        context = parent;
+      }
+      return path;
+    },
+    permissionFrameMatches(id, expectedToken, owner) {
+      return owner.isConnected && owner.contentWindow === entry(id, expectedToken).context;
+    },
     focus(id, expectedToken) {
       const element = entry(id, expectedToken).element;
       if (!element || !element.isConnected) throw new Error('stale element reference');
@@ -110,6 +129,9 @@
   driver.in_automation = true;
   driver.click = (element, coords) => request({kind: 'click', x: coords.x, y: coords.y}, element);
   driver.send_keys = (element, keys) => request({kind: 'send_keys', keys}, element);
+  driver.set_permission = async (params, context = null) => request({
+    kind: 'set_permission', descriptor: params.descriptor, state: params.state,
+  }, null, context || window);
   driver.action_sequence = (actions, context = null) => {
     const elements = [];
     const serialized = actions.map(source => ({...source, actions: source.actions.map(action => {

moli-benchmark/tests/test_wpt_cross.py

@@ -4112,16 +4112,6 @@ test(() => {}, "ok");
         self.assertIsNone(store.wait_for_final("example.html", timeout=0))
         self.assertEqual(store.get("example.html"), {"source": "incremental"})
 
-    def test_testdriver_vendor_bridge_accepts_storage_access_permission_setup(self) -> None:
-        self.assertIn(
-            b"params.descriptor.name === 'storage-access'",
-            BENCH_TESTDRIVER_VENDOR_BRIDGE,
-        )
-        self.assertIn(
-            b"set_permission() is not implemented by the Moli WPT bridge",
-            BENCH_TESTDRIVER_VENDOR_BRIDGE,
-        )
-
     def test_testdriver_vendor_bridge_provides_computed_label(self) -> None:
         self.assertIn(b"get_computed_label", BENCH_TESTDRIVER_VENDOR_BRIDGE)
         self.assertIn(b"resolveReferenceTarget", BENCH_TESTDRIVER_VENDOR_BRIDGE)

moli-benchmark/tests/test_wpt_native_input.py

@@ -8,10 +8,11 @@ from pathlib import Path
 from types import SimpleNamespace
 from unittest.mock import AsyncMock
 
-from moli_benchmark.raw_cdp import RawCdpConnectionClosed
+from moli_benchmark.raw_cdp import RawCdpConnectionClosed, RawCdpError
 from moli_benchmark.wpt_cross.__main__ import _case_references_testdriver
 from moli_benchmark.wpt_cross.case_set import WptCase
-from moli_benchmark.wpt_cross.native_input import NativeInput, Pointer, key_description
+from moli_benchmark.wpt_cross.native_input import NativeInput, Pointer, key_description, permission_origin
+from moli_benchmark.wpt_cross.runner import _AttachedPage, _close_page, _close_target
 
 
 class RecordingClient:
@@ -172,8 +173,151 @@ class NativeInputTests(unittest.IsolatedAsyncioTestCase):
         self.assertEqual(self.client.command.await_count, 2)
         self.client.close.assert_awaited_once()
 
+    async def test_permissions_use_the_selected_frame_origin_and_case_browser_context(self):
+        self.driver.browser_context_id = "case-context"
+        self.driver.evaluate = AsyncMock(return_value=[1, 0])
+        self.driver.command = AsyncMock(side_effect=[{"frameTree": {
+            "frame": {"securityOrigin": "https://top.test"},
+            "childFrames": [
+                {"frame": {"securityOrigin": "https://sibling.test"}},
+                {"frame": {"securityOrigin": "https://outer.test"}, "childFrames": [
+                    {"frame": {"id": "child-frame", "securityOrigin": "https://child.test", "url": "about:blank"}},
+                ]},
+            ],
+        }}, {"backendNodeId": 42}, {"object": {"objectId": "owner-object"}},
+            {"result": {"value": True}}, {}])
+        descriptor = {"name": "clipboard-write", "allowWithoutSanitization": True}
+        await self.driver.perform(9, {
+            "id": 4, "token": "realm", "kind": "set_permission", "descriptor": descriptor, "state": "denied",
+        })
+        method, params, kwargs = self.client.commands[-1]
+        self.assertEqual(method, "Browser.setPermission")
+        self.assertEqual(params, {
+            "permission": descriptor, "setting": "denied", "browserContextId": "case-context",
+            "origin": "https://top.test", "embeddedOrigin": "https://child.test",
+        })
+        self.assertNotIn("session_id", kwargs)
+        self.driver.command.assert_any_await("DOM.getFrameOwner", {"frameId": "child-frame"})
+        self.driver.command.assert_any_await("Runtime.releaseObject", {"objectId": "owner-object"})
+        self.driver.command.side_effect = None
+        self.driver.command.return_value = {}
+        await self.driver.close()
+        reset = next(command for command in self.client.commands if command[0] == "Browser.resetPermissions")
+        self.assertEqual(reset[1], {"browserContextId": "case-context"})
+        self.assertNotIn("session_id", reset[2])
+
+    async def test_permissions_reject_malformed_requests_before_any_protocol_mutation(self):
+        for descriptor, state in [({}, "granted"), ({"name": ""}, "granted"), ({"name": 1}, "granted"),
+                                  ({"name": "geolocation"}, "invalid"), ({"name": "geolocation"}, {})]:
+            with self.subTest(descriptor=descriptor, state=state), self.assertRaises(ValueError):
+                await self.driver.perform(1, {"id": 1, "token": "realm", "kind": "set_permission",
+                                              "descriptor": descriptor, "state": state})
+        self.assertEqual(self.client.commands, [])
+
+    async def test_permissions_reject_detached_moved_and_opaque_frames(self):
+        request = {"id": 1, "token": "realm", "kind": "set_permission", "descriptor": {"name": "geolocation"}, "state": "granted"}
+        for paths, frame in [
+            ([[0]], {"securityOrigin": "https://top.test"}),
+            ([[], [0]], {"securityOrigin": "https://top.test"}),
+            ([[]], {"securityOrigin": "null", "url": "https://opaque.test/"}),
+        ]:
+            with self.subTest(paths=paths, frame=frame):
+                self.driver.evaluate = AsyncMock(side_effect=paths)
+                self.driver.command = AsyncMock(return_value={"frameTree": {"frame": frame}})
+                with self.assertRaises(ValueError):
+                    await self.driver.perform(1, request)
+        self.assertEqual(self.client.commands, [])
+
+    async def test_failed_permission_command_propagates_and_still_schedules_reset(self):
+        self.driver.evaluate = AsyncMock(return_value=[])
+        self.driver.command = AsyncMock(return_value={"frameTree": {"frame": {"securityOrigin": "https://top.test"}}})
+        self.client.command = AsyncMock(side_effect=RawCdpError("unsupported permission"))
+        with self.assertRaisesRegex(RawCdpError, "unsupported permission"):
+            await self.driver.perform(1, {"id": 1, "token": "realm", "kind": "set_permission",
+                                          "descriptor": {"name": "unsupported"}, "state": "granted"})
+        with self.assertRaisesRegex(RawCdpError, "reset WPT permissions"):
+            await self.driver.close()
+        self.assertEqual([call.args[0] for call in self.client.command.await_args_list], [
+            "Browser.setPermission", "Browser.resetPermissions", "Runtime.removeBinding", "Target.detachFromTarget",
+        ])
+        self.client.close.assert_awaited_once()
+        with self.assertRaisesRegex(RawCdpError, "reset WPT permissions"):
+            await self.driver.close()
+        self.client.close.assert_awaited_once()
+
+    async def test_omitted_frame_cannot_grant_permission_to_a_sibling(self):
+        self.driver.evaluate = AsyncMock(return_value=[0])
+        self.driver.command = AsyncMock(side_effect=[
+            {"frameTree": {"frame": {"securityOrigin": "https://top.test"}, "childFrames": [
+                {"frame": {"id": "sibling", "securityOrigin": "https://sibling.test"}},
+            ]}},
+            {"backendNodeId": 42}, {"object": {"objectId": "sibling-owner"}},
+            {"result": {"value": False}}, {},
+        ])
+        with self.assertRaisesRegex(ValueError, "does not match"):
+            await self.driver.perform(1, {"id": 1, "token": "realm", "kind": "set_permission",
+                                          "descriptor": {"name": "geolocation"}, "state": "granted"})
+        self.assertEqual(self.client.commands, [])
+        self.driver.command.assert_awaited_with("Runtime.releaseObject", {"objectId": "sibling-owner"})
+
+    async def test_failed_permission_cleanup_still_disposes_case_targets(self):
+        class PageClient:
+            def __init__(self):
+                self.commands = []
+
+            async def send(self, method, params=None, **kwargs):
+                self.commands.append((method, params))
+                return len(self.commands)
+
+            async def recv_until_id(self, command_id, **kwargs):
+                method, _ = self.commands[command_id - 1]
+                result = {"targetInfos": []} if method == "Target.getTargets" else {"success": True}
+                return {"result": result}, []
+
+        for browser_context_id in (None, "isolated-context"):
+            with self.subTest(browser_context_id=browser_context_id):
+                client = PageClient()
+                native = SimpleNamespace(close=AsyncMock(side_effect=RawCdpError("reset failed")))
+                page = _AttachedPage(browser_context_id, "test-target", "test-session", frozenset(), native)
+                if browser_context_id is None:
+                    with self.assertRaisesRegex(RawCdpError, "reset failed"):
+                        await _close_page(client, page)
+                    self.assertIn(("Target.closeTarget", {"targetId": "test-target"}), client.commands)
+                else:
+                    await _close_page(client, page)
+                    self.assertIn(("Target.disposeBrowserContext", {"browserContextId": browser_context_id}), client.commands)
+
+
+    async def test_target_cleanup_accepts_a_concurrently_closed_target(self):
+        client = SimpleNamespace(
+            send=AsyncMock(side_effect=[1, 2]),
+            recv_until_id=AsyncMock(side_effect=[
+                RawCdpError("No target with given id found"),
+                ({"result": {"targetInfos": [{"targetId": "unrelated"}]}}, []),
+            ]),
+        )
+        await _close_target(client, "case-target")
+
+    async def test_target_cleanup_preserves_errors_for_a_live_target(self):
+        client = SimpleNamespace(
+            send=AsyncMock(side_effect=[1, 2]),
+            recv_until_id=AsyncMock(side_effect=[
+                RawCdpError("close rejected"),
+                ({"result": {"targetInfos": [{"targetId": "case-target"}]}}, []),
+            ]),
+        )
+        with self.assertRaisesRegex(RawCdpError, "close rejected"):
+            await _close_target(client, "case-target")
+
 
 class NativeInputSelectionTests(unittest.TestCase):
+    def test_permission_origins_keep_inherited_and_opaque_origin_semantics(self):
+        self.assertEqual(permission_origin({"url": "about:blank", "securityOrigin": "https://parent.test:8443"}), "https://parent.test:8443")
+        self.assertEqual(permission_origin({"url": "https://top.test/path?q=1"}), "https://top.test")
+        for origin in ("null", "", "file://", "https://user:password@top.test"):
+            with self.subTest(origin=origin), self.assertRaises(ValueError):
+                permission_origin({"url": "https://top.test", "securityOrigin": origin})
+
     def test_testdriver_detection_in_html_and_generated_variants(self):
         with tempfile.TemporaryDirectory() as directory:
             root = Path(directory)