diff --git a/packages/flow/openadapt_flow/__main__.py b/packages/flow/openadapt_flow/__main__.py index e6ec9a83bf..7647dd4040 100644 --- a/packages/flow/openadapt_flow/__main__.py +++ b/packages/flow/openadapt_flow/__main__.py @@ -2547,6 +2547,10 @@ def new_page(url: str): if browser is None: raise opened = browser.new_page(viewport=page.viewport_size) + from openadapt_flow.backends.playwright_backend import prepare_ws_guard + + # Guard the page's WebSockets from its first document (wire check). + prepare_ws_guard(opened) opened.goto(url) try: opened.wait_for_load_state("load") @@ -2805,6 +2809,10 @@ def run_on_page(page, *, entry_url: Optional[str]): page = context.new_page() else: page = browser.new_page(viewport=_VIEWPORT, service_workers="block") + from openadapt_flow.backends.playwright_backend import prepare_ws_guard + + # Guard the page's WebSockets from its first document (wire check). + prepare_ws_guard(page) page.goto(url) try: report = run_on_page(page, entry_url=url) @@ -4526,6 +4534,12 @@ def _resume_with(backend: "Backend") -> "RunReport": with sync_playwright() as p: browser = launch_chromium(p.chromium, headless=not headed) page = browser.new_page(viewport=_VIEWPORT) + from openadapt_flow.backends.playwright_backend import ( + prepare_ws_guard, + ) + + # Guard the page's WebSockets from its first document (wire check). + prepare_ws_guard(page) page.goto(url) try: report = _resume_with(build_backend(backend_cfg, page=page)) @@ -5042,6 +5056,10 @@ def backend_factory(): with sync_playwright() as p: browser = launch_chromium(p.chromium, headless=not args.headed) page = browser.new_page(viewport=_VIEWPORT) + from openadapt_flow.backends.playwright_backend import prepare_ws_guard + + # Guard the page's WebSockets from its first document (wire check). + prepare_ws_guard(page) page.goto(target_url) try: yield PlaywrightBackend(page) diff --git a/packages/flow/openadapt_flow/agent/browser_teach.py b/packages/flow/openadapt_flow/agent/browser_teach.py index 13d6822557..806d44d20e 100644 --- a/packages/flow/openadapt_flow/agent/browser_teach.py +++ b/packages/flow/openadapt_flow/agent/browser_teach.py @@ -830,6 +830,10 @@ def _run_steps( ) page = browser.page try: + from openadapt_flow.backends.playwright_backend import prepare_ws_guard + + # Guard the page's WebSockets from its first document (wire check). + prepare_ws_guard(page) page.goto(program["url"], wait_until="domcontentloaded") for i, step in enumerate(program["steps"], 1): if browser.blocked: @@ -1383,6 +1387,10 @@ def open_page() -> str: allowed_writes=contract.writes if contract is not None else (), secrets=list(secret_values.values()), ) + from openadapt_flow.backends.playwright_backend import prepare_ws_guard + + # Guard the page's WebSockets from its first document (wire check). + prepare_ws_guard(run.browser.page) run.browser.page.goto(url, wait_until="domcontentloaded") return self._listing(run) diff --git a/packages/flow/openadapt_flow/backends/playwright_backend.py b/packages/flow/openadapt_flow/backends/playwright_backend.py index 8cab1549df..8ef452be74 100644 --- a/packages/flow/openadapt_flow/backends/playwright_backend.py +++ b/packages/flow/openadapt_flow/backends/playwright_backend.py @@ -1714,15 +1714,16 @@ def __init__( closed_page_frame: bool = False, bring_to_front_before_capture: bool = False, native_fill: Optional[bool] = None, - ws_guard: Optional[tuple[str, dict[str, Any]]] = None, + ws_guard: Optional[str] = None, ) -> None: """Wrap an existing Playwright page. Args: page: A Playwright page. ws_guard: What :func:`prepare_ws_guard` returned for the page's - context before its first document loaded (as :meth:`launch` - does), so every document starts guarded and routed. + context. Optional: the backend looks it up (or prepares it) + itself; documents opened before it was prepared can't hold a + create-form save. native_fill: Insert typed text in one call when the focused element is an ordinary text field. Off by default; the ``fast`` speed profile (``runtime/speed.py``) turns it on @@ -1797,28 +1798,21 @@ def __init__( self._fill_store_key = f"__oaflow_filled_{uuid.uuid4().hex}" self._fill_value_slot = f"__oaflow_filled_values_{uuid.uuid4().hex}" self._form_wire: Any = None - self._ws_routed = False - # Page-side state of the WebSocket guard (see _arm_wire_check), kept - # under a private key in each document; the generation counts holds. - self._ws_guard_key = f"__oaflow_ws_{uuid.uuid4().hex}" - self._ws_guard_generation = 0 + # The page-side guard (see _arm_wire_check): one per browser context, + # kept in each document under a private key, installed by an init + # script so documents loaded from now on are guarded from their + # start. A page opened before it (``launch`` and the CLI prepare + # theirs first) can't hold a save. + self._ws_guard_key = ws_guard or prepare_ws_guard(self.page) + self._ws_hold: Optional[str] = None self._ws_hold_script: Any = None self._held_pages: list[tuple[Any, Callable[[Any], None]]] = [] - if ws_guard is not None: - self._ws_guard_key, routes = ws_guard - routes["backend"] = self - self._ws_routed = bool(routes.get("routed")) try: - # A frame sent on this page's WebSockets while a save is held is - # reported (the wire check routes only sockets opened then). + # A message that still reaches the wire from this page's + # WebSockets while a save is held is reported. self.page.on("websocket", self._watch_websocket) except Exception: # noqa: BLE001 - a page stand-in without events pass - # Every document this context loads from now on wraps its WebSocket - # constructor before any page script runs, so a socket constructed - # while a save is held is refused in the page, at construction. - if ws_guard is None: - self._add_ws_guard_script() self._structural_tokens: dict[str, _StructuralGuard] = {} self._guarded_coordinate: Optional[_CoordinateGuard] = None # One-shot expectation that the next click closes this page surface @@ -2591,23 +2585,22 @@ def _arm_wire_check(self, submission: ExpectedSubmission) -> Optional[str]: browser context (a popup or new tab included), and a GET navigation to the form's address, goes through a :class:`~openadapt_flow.wire_check.WireCheck`: the checked submission - goes once; anything else is aborted before it is sent. A WebSocket - constructed meanwhile is refused in the page at construction (a - guard on every document of the context, installed in each open - frame now and at the start of every document loaded later), so it is - never created, and one that still reaches the route is never - connected; both are reported. The hold on the page side is set - before the click and released only when the outcome has been read - (:meth:`_disarm_wire_check`). A message sent on a socket already - open is dropped and reported. Workers' WebSockets skip both the - guard and the route, so a page running a dedicated worker is refused - before the click, and a worker (or shared worker) constructed - meanwhile is refused in the page; one that still starts is - reported. Limits: a shared worker started before the hold can't be - seen, and a page script that kept a reference to the browser's own - constructor before the guard was installed is outside it; pages - from :meth:`launch` are guarded before their first document loads, - which closes that gap for them. + goes once; anything else is aborted before it is sent. Connections + that don't go through those routes are held in the page by a guard + installed on the context (:func:`prepare_ws_guard`) that runs + before any page script in every document: a WebSocket, + WebTransport, RTCPeerConnection, Worker or SharedWorker constructed + meanwhile is refused at construction, so it is never created, and a + message sent on a WebSocket or RTCDataChannel already open is + dropped; each is reported. The page-side hold is set before the + click and released only when the outcome has been read + (:meth:`_disarm_wire_check`). A save is refused before the click + when a document on the page loaded before the guard (its scripts + may hold the browser's own constructors), when the guard can't be + kept in a frame, and when a dedicated worker runs (its WebSockets + skip the guard); a worker that still starts is reported. Not held: + a WebTransport session's streams opened before the hold, and a + shared worker started before it. A service worker's requests skip routes, so a page a service worker controls is refused before the click (the runner's contexts block service workers), and one that starts meanwhile is reported. @@ -2668,65 +2661,52 @@ def on_page_worker(_worker: Any) -> None: continue self._held_pages.append((page, on_page_worker)) self._form_wire = (handler, check, on_worker, notes) - # Guard the open documents before Playwright's routing replaces their - # WebSocket (1.64 does that in open documents too): the browser's - # own WebSocket prototype is then guarded as well, so a message sent - # on a socket from a reference a page script kept is still dropped. - self._ws_guard_frames({}) - if not self._ws_routed: - # Routed once for the context's life; it only acts while armed. - # Playwright installs its routing in the page by replacing - # WebSocket, and (1.61) only in documents loaded afterwards; the - # guard script goes in again after it so it wraps the routed one. - try: - context.route_web_socket(lambda _url: True, self._route_websocket) - self._ws_routed = True - except Exception: # noqa: BLE001 - self._disarm_wire_check() - return "could not hold the page's WebSockets" - self._add_ws_guard_script() # Hold the page side: documents loaded from now on start held, and # every open document is held now. The page decides at construction # time, so no socket's fate depends on when Python services it. - self._ws_guard_generation += 1 + hold = uuid.uuid4().hex + self._ws_hold = hold try: self._ws_hold_script = context.add_init_script( f"({_WS_GUARD_JS})({json.dumps(self._ws_guard_key)}, " - f"{{armed: {self._ws_guard_generation}}})" + f"{{armed: {json.dumps(hold)}}})" ) except Exception: # noqa: BLE001 self._disarm_wire_check() return "could not hold the page's WebSockets" - _counts, unheld = self._ws_guard_frames({"hold": True}) + (_counts, late), unheld = self._ws_guard_frames({"hold": hold}) if unheld: self._disarm_wire_check() return ( "could not hold the WebSockets of a frame on the page, so " "the save can't be held" ) - return None - - def _add_ws_guard_script(self) -> None: - try: - self.page.context.add_init_script( - f"({_WS_GUARD_JS})({json.dumps(self._ws_guard_key)}, {{}})" + if late: + # Its scripts ran before the guard and may hold the browser's own + # WebSocket, Worker or send: nothing here can stop those. + self._disarm_wire_check() + return ( + "the page loaded before the WebSocket guard was installed, so " + "the save can't be held; open the page after the backend " + "exists (reload it)" ) - except Exception: # noqa: BLE001 - a page stand-in; arming re-checks - pass + return None def _ws_guard_frames( self, options: dict[str, Any] - ) -> tuple[tuple[int, int, int], int]: + ) -> tuple[tuple[tuple[int, int, int], int], int]: """Run the page-side guard in every frame of every page of this context (installing it where the document predates it). Returns - what it refused since the hold began (sockets, workers, messages) - and how many frames it could not reach.""" + what it refused for the current hold (sockets, workers, messages), + how many documents were not guarded from their start, and how many + frames it could not reach.""" - sockets = workers = sends = unreached = 0 + sockets = workers = sends = late = unreached = 0 + options = {"id": self._ws_hold, **options} try: pages = list(self.page.context.pages) except Exception: # noqa: BLE001 - the context is gone - return (0, 0, 0), 1 + return ((0, 0, 0), 0), 1 for page in pages: try: frames = list(page.frames) @@ -2744,23 +2724,26 @@ def _ws_guard_frames( continue if ( isinstance(count, list) - and len(count) == 3 - and all(isinstance(n, int) and n >= 0 for n in count) + and len(count) == 4 + and all(isinstance(n, int) and n >= 0 for n in count[:3]) + and isinstance(count[3], bool) ): sockets += count[0] workers += count[1] sends += count[2] + late += not count[3] else: unreached += 1 - return (sockets, workers, sends), unreached + return ((sockets, workers, sends), late), unreached def _ws_guard_notes(self) -> list[str]: - (sockets, workers, sends), _unreached = self._ws_guard_frames({}) + ((sockets, workers, sends), _late), _unreached = self._ws_guard_frames({}) notes = [] if sockets: notes.append( - "the page opened a WebSocket while the save was held; it was " - "refused before it connected" + "the page opened a connection (WebSocket, WebTransport or " + "WebRTC) while the save was held; it was refused before it " + "connected" ) if workers: notes.append( @@ -2769,13 +2752,16 @@ def _ws_guard_notes(self) -> list[str]: ) if sends: notes.append( - "the page sent a WebSocket message while the save was held; " - "it was not sent" + "the page sent a WebSocket or data channel message while the " + "save was held; it was not sent" ) return notes def _release_ws_guard(self) -> None: + hold, self._ws_hold = self._ws_hold, None handle, self._ws_hold_script = self._ws_hold_script, None + if hold is None: + return context = self.page.context dispose = getattr(handle, "dispose", None) try: @@ -2786,22 +2772,11 @@ def _release_ws_guard(self) -> None: # later one marks this hold released for new documents. context.add_init_script( f"({_WS_GUARD_JS})({json.dumps(self._ws_guard_key)}, " - f"{{released: {self._ws_guard_generation}}})" + f"{{released: {json.dumps(hold)}}})" ) except Exception: # noqa: BLE001 - the context is gone pass - self._ws_guard_frames({"release": self._ws_guard_generation}) - - def _route_websocket(self, ws: Any) -> None: - wire = self._form_wire - if wire is None: - ws.connect_to_server() - return - # Never connected to the server: what the page sends on it goes - # nowhere. (Closing it from a sync route handler deadlocks.) - wire[3].append( - "the page opened a WebSocket while the save was held; it was not connected" - ) + self._ws_guard_frames({"release": hold, "id": hold}) def _watch_websocket(self, ws: Any) -> None: def on_frame(_payload: Any) -> None: @@ -2819,12 +2794,9 @@ def _disarm_wire_check(self) -> list[str]: return [] handler, check, on_worker, notes = wire reasons = _wire_reasons(check, notes + self._ws_guard_notes()) - # Release the page side first, while the Python side still holds: - # a socket that reaches the route meanwhile is never connected. One - # more round trip to every frame lets the route callbacks of sockets - # opened before the release arrive while still held. + # The page decides at construction, so releasing it is the last + # step that can matter: nothing waits on Python to refuse a socket. self._release_ws_guard() - self._ws_guard_frames({}) held_pages, self._held_pages = self._held_pages, [] for page, listener in held_pages: try: @@ -4781,9 +4753,9 @@ def close() -> None: except BaseException: close() raise - # Guard and route WebSockets before the first document loads, so no - # page script can keep a reference to the unguarded constructor. - ws_guard = prepare_ws_guard(page.context) + # Guard WebSockets before the first document loads, so no page + # script can keep a reference to an unguarded constructor. + ws_guard = prepare_ws_guard(page) page.goto(url) backend = cls(page, ws_guard=ws_guard) # Read by the generic web environment observer: the environment @@ -4799,32 +4771,55 @@ def close() -> None: ) # The page-side guard, run as ``(key, options)`` at the start of every -# document (an init script) and in every open frame when a save is held. It -# wraps the page's WebSocket, Worker and SharedWorker constructors -# (Playwright's routed WebSocket when that is installed) so one constructed -# while the document is held is refused there and never created, and it -# drops what is sent on a socket while held. The wrapped constructor is -# reachable from no prototype: each ``prototype.constructor`` on the chain -# points at the wrapper, whose own prototype is EventTarget. A document -# loaded during a hold starts held (``armed`` above ``released``). Options: -# ``hold`` holds this document and resets its counts, ``release`` (a hold's -# generation) releases it, ``armed``/``released`` record a hold's generation. -# Returns ``[sockets, workers, sends]`` refused since the hold began, or -1 -# when the guard can't be kept here. +# document (an init script, ``start``) and in every open frame when a save +# is held. It wraps the page's WebSocket, WebTransport, RTCPeerConnection, +# Worker and SharedWorker constructors so one constructed while the document +# is held is refused there and never created, and it drops what is sent on a +# WebSocket or RTCDataChannel while held. No ``prototype.constructor`` on a +# wrapped chain leads back to the unwrapped constructor, and the wrapper's +# own prototype is EventTarget. A document is held while any hold id is in +# ``holds``: ``hold``/``release`` add and remove one in an open document, +# and ``armed``/``released`` init scripts do so for documents loaded during +# a hold. Returns ``[sockets, workers, sends, early]`` for hold ``id`` +# (``early``: the guard ran before any page script), or -1 when the guard +# can't be kept here. _WS_GUARD_JS = r"""(key, o) => { const g = globalThis; let s = Object.prototype.hasOwnProperty.call(g, key) ? g[key] : null; if (!s || typeof s !== 'object') { - s = {held: false, armed: 0, released: 0, sockets: 0, workers: 0, - sends: 0, guarded: {}, sendsPatched: []}; + s = {early: !!o.start, holds: new Set(), counts: new Map(), + guarded: {}, patched: []}; try { Object.defineProperty(g, key, {value: s, enumerable: false}); } catch (e) { return -1; } } - const held = () => s.held || s.armed > s.released; + const held = () => s.holds.size > 0; + const bump = (i) => { + for (const h of s.holds) { + const c = s.counts.get(h) || [0, 0, 0]; + c[i] += 1; + s.counts.set(h, c); + } + }; const base = typeof g.EventTarget === 'function' ? g.EventTarget : null; const stop = base ? base.prototype : Object.prototype; - const wrap = (name, count) => { + const patchSend = (q) => { + if (!q || !Object.prototype.hasOwnProperty.call(q, 'send') || + s.patched.includes(q)) return true; + const send = q.send; + if (typeof send !== 'function') return true; + try { + Object.defineProperty(q, 'send', { + value: function send_(...args) { + if (held()) { bump(2); return undefined; } + return Reflect.apply(send, this, args); + }, writable: true, configurable: true, enumerable: true, + }); + } catch (e) { return false; } + s.patched.push(q); + return true; + }; + const wrap = (name, slot) => { const inner = g[name]; if (typeof inner !== 'function' || inner === s.guarded[name]) return true; const Guarded = {[name]: function (...args) { @@ -4833,7 +4828,7 @@ def close() -> None: "Please use the 'new' operator"); } if (held()) { - s[count] += 1; + bump(slot); throw new DOMException( 'A ' + name + ' cannot start while a checked save is held', 'SecurityError'); @@ -4856,20 +4851,7 @@ def close() -> None: value: Guarded, writable: true, configurable: true, enumerable: false, }); } catch (e) { return false; } - if (name !== 'WebSocket' || - !Object.prototype.hasOwnProperty.call(q, 'send') || - s.sendsPatched.includes(q)) continue; - const send = q.send; - if (typeof send !== 'function') continue; - try { - Object.defineProperty(q, 'send', { - value: function send_(...args) { - if (held()) { s.sends += 1; return undefined; } - return Reflect.apply(send, this, args); - }, writable: true, configurable: true, enumerable: true, - }); - } catch (e) { return false; } - s.sendsPatched.push(q); + if (name === 'WebSocket' && !patchSend(q)) return false; } try { Object.defineProperty(g, name, { @@ -4879,51 +4861,47 @@ def close() -> None: s.guarded[name] = Guarded; return true; }; - if (!wrap('WebSocket', 'sockets') || !wrap('Worker', 'workers') || - !wrap('SharedWorker', 'workers')) { + const kept = [ + ['WebSocket', 0], ['WebTransport', 0], ['RTCPeerConnection', 0], + ['webkitRTCPeerConnection', 0], ['Worker', 1], ['SharedWorker', 1], + ].every(([name, slot]) => wrap(name, slot)); + if (!kept || (g.RTCDataChannel && !patchSend(g.RTCDataChannel.prototype))) { return -1; } - if (typeof o.armed === 'number') s.armed = Math.max(s.armed, o.armed); - if (typeof o.released === 'number') { - s.released = Math.max(s.released, o.released); + if (typeof o.armed === 'string') s.holds.add(o.armed); + if (typeof o.released === 'string') s.holds.delete(o.released); + if (typeof o.hold === 'string') { + s.holds.add(o.hold); + s.counts.set(o.hold, [0, 0, 0]); } - if (o.hold) { s.held = true; s.sockets = 0; s.workers = 0; s.sends = 0; } - if (typeof o.release === 'number') { - s.held = false; - s.released = Math.max(s.released, o.release); - } - return [s.sockets, s.workers, s.sends]; + if (typeof o.release === 'string') s.holds.delete(o.release); + const c = (typeof o.id === 'string' && s.counts.get(o.id)) || [0, 0, 0]; + if (typeof o.release === 'string') s.counts.delete(o.release); + return [c[0], c[1], c[2], s.early]; }""" -def prepare_ws_guard(context: Any) -> tuple[str, dict[str, Any]]: - """Install the page-side WebSocket guard and the WebSocket route on a - browser context before its pages load anything, for a - :class:`PlaywrightBackend` created later with ``ws_guard=``. Until that - backend exists (and while no save is held) sockets connect as usual. - Never raises: what could not be installed is installed again, in the - open frames, when a save is held.""" - +def prepare_ws_guard(target: Any) -> str: + """Install the page-side WebSocket guard on a browser context (once) and + return its key. ``target`` is the context or one of its pages. Call it + before the context's pages load anything, as + :meth:`PlaywrightBackend.launch` and the CLI do: a document loaded + before it can't hold a create-form save. It changes nothing while no + save is held. Never raises: if it can't install, holding refuses.""" + + context = target + if not hasattr(target, "new_page"): # a page: guard its context + context = getattr(target, "context", None) + key = getattr(context, "_oaflow_ws_guard_key", None) + if isinstance(key, str): + return key key = f"__oaflow_ws_{uuid.uuid4().hex}" - routes: dict[str, Any] = {"backend": None, "routed": False} - - def route(ws: Any) -> None: - backend = routes["backend"] - if backend is None: - ws.connect_to_server() - else: - backend._route_websocket(ws) - - script = f"({_WS_GUARD_JS})({json.dumps(key)}, {{}})" try: - context.add_init_script(script) - context.route_web_socket(lambda _url: True, route) - routes["routed"] = True - # Again after Playwright's routing, which replaces WebSocket. - context.add_init_script(script) - except Exception: # noqa: BLE001 - arming installs what is missing + context.add_init_script(f"({_WS_GUARD_JS})({json.dumps(key)}, {{start: true}})") + context._oaflow_ws_guard_key = key + except Exception: # noqa: BLE001 - a stand-in context; holding refuses pass - return key, routes + return key def _wire_reasons(check: WireCheck, notes: list[str]) -> list[str]: diff --git a/packages/flow/openadapt_flow/benchmark/reliability.py b/packages/flow/openadapt_flow/benchmark/reliability.py index 8deddb4e90..f749942403 100644 --- a/packages/flow/openadapt_flow/benchmark/reliability.py +++ b/packages/flow/openadapt_flow/benchmark/reliability.py @@ -178,6 +178,10 @@ def record_app(app: AppSpec, out_dir: Path, browser: Any) -> Path: device_scale_factor=1, ) try: + from openadapt_flow.backends.playwright_backend import prepare_ws_guard + + # Guard the page's WebSockets from its first document (wire check). + prepare_ws_guard(page) page.goto(app.url, wait_until="domcontentloaded", timeout=GOTO_TIMEOUT_MS) page.wait_for_timeout(int(POST_GOTO_SETTLE_S * 1000)) backend = PlaywrightBackend(page) @@ -409,6 +413,10 @@ def run_target(app: AppSpec, workdir: Path, browser: Any) -> dict: device_scale_factor=1, ) try: + from openadapt_flow.backends.playwright_backend import prepare_ws_guard + + # Guard the page's WebSockets from its first document (wire check). + prepare_ws_guard(page) page.goto(app.url, wait_until="domcontentloaded", timeout=GOTO_TIMEOUT_MS) page.wait_for_timeout(int(POST_GOTO_SETTLE_S * 1000)) backend = PlaywrightBackend(page) diff --git a/packages/flow/openadapt_flow/cli_worklist_pool.py b/packages/flow/openadapt_flow/cli_worklist_pool.py index 459345a937..9b88f6ab96 100644 --- a/packages/flow/openadapt_flow/cli_worklist_pool.py +++ b/packages/flow/openadapt_flow/cli_worklist_pool.py @@ -230,6 +230,10 @@ def open(self) -> None: self._context = contexts[0] self._page = self._context.new_page() self._page.set_viewport_size(dict(self.viewport)) + from openadapt_flow.backends.playwright_backend import prepare_ws_guard + + # Guard the page's WebSockets from its first document (wire check). + prepare_ws_guard(self._page) self._page.goto(self.entry_url) else: from openadapt_flow._browser_setup import launch_chromium @@ -253,6 +257,10 @@ def run_item(self, idx: int, row: dict[str, Any], item_dir: Path) -> ItemRun: if self.open_entry is not None: self.open_entry(self._page) else: + from openadapt_flow.backends.playwright_backend import prepare_ws_guard + + # Guard the page's WebSockets from its first document (wire check). + prepare_ws_guard(self._page) self._page.goto(self.entry_url) report = self.run_item_on_page(self._page, idx, row, item_dir, self._pipeline) return _item_run(report, self._pipeline) @@ -337,9 +345,12 @@ def sign_in_with_bundle( ) -> None: """Replay the recorded sign-in bundle with this account's credentials.""" + from openadapt_flow.backends.playwright_backend import prepare_ws_guard from openadapt_flow.ir import Workflow from openadapt_flow.runtime import Replayer + # Guard the page's WebSockets from its first document (wire check). + prepare_ws_guard(page) page.goto(url) workflow = Workflow.load(bundle) params = {k: os.environ[v] for k, v in account.params_env.items()} @@ -364,8 +375,11 @@ def entry_with_bundle( from uuid import uuid4 + from openadapt_flow.backends.playwright_backend import prepare_ws_guard from openadapt_flow.runtime import Replayer + # Guard the page's WebSockets from its first document (wire check). + prepare_ws_guard(page) page.goto(url) report = Replayer(backend_factory(page)).run( workflow, params={}, bundle_dir=bundle, run_dir=run_dir / uuid4().hex[:12] diff --git a/packages/flow/openadapt_flow/compiler/probing.py b/packages/flow/openadapt_flow/compiler/probing.py index bfe68b08f4..c262aa86fc 100644 --- a/packages/flow/openadapt_flow/compiler/probing.py +++ b/packages/flow/openadapt_flow/compiler/probing.py @@ -687,6 +687,10 @@ def run( try: context = browser.new_context(viewport={"width": 1280, "height": 800}) page = context.new_page() + from openadapt_flow.backends.playwright_backend import prepare_ws_guard + + # Guard the page's WebSockets from its first document (wire check). + prepare_ws_guard(page) page.goto(backend_cfg.url) backend = build_backend(backend_cfg, page=page) return self._replay(backend, workflow, bundle_dir, run_dir, params) diff --git a/packages/flow/openadapt_flow/emit/mcp_tool.py b/packages/flow/openadapt_flow/emit/mcp_tool.py index 9d09c134ab..0689c6922d 100644 --- a/packages/flow/openadapt_flow/emit/mcp_tool.py +++ b/packages/flow/openadapt_flow/emit/mcp_tool.py @@ -77,7 +77,10 @@ def {func_name}(url: str{param_sig}) -> dict: from openadapt_flow._browser_setup import launch_chromium - from openadapt_flow.backends.playwright_backend import PlaywrightBackend + from openadapt_flow.backends.playwright_backend import ( + PlaywrightBackend, + prepare_ws_guard, + ) from openadapt_flow.ir import Workflow from openadapt_flow.runtime import Replayer @@ -86,6 +89,7 @@ def {func_name}(url: str{param_sig}) -> dict: with sync_playwright() as p: browser = launch_chromium(p.chromium, headless=True) page = browser.new_page(viewport={{"width": 1280, "height": 800}}) + prepare_ws_guard(page) page.goto(url) try: backend = PlaywrightBackend(page) diff --git a/packages/flow/openadapt_flow/integrations/harness2/rollout.py b/packages/flow/openadapt_flow/integrations/harness2/rollout.py index 24b46dbe44..c404a95717 100644 --- a/packages/flow/openadapt_flow/integrations/harness2/rollout.py +++ b/packages/flow/openadapt_flow/integrations/harness2/rollout.py @@ -166,6 +166,10 @@ def _drive( browser = pw.chromium.launch(headless=headless) try: page = browser.new_page(viewport={"width": 1100, "height": 800}) + from openadapt_flow.backends.playwright_backend import prepare_ws_guard + + # Guard the page's WebSockets from its first document (wire check). + prepare_ws_guard(page) page.goto(url) page.wait_for_selector("#app *", timeout=10_000) driver = driver_factory(page) diff --git a/packages/flow/openadapt_flow/interactive_recorder.py b/packages/flow/openadapt_flow/interactive_recorder.py index 2529a329ed..5ad9f90767 100644 --- a/packages/flow/openadapt_flow/interactive_recorder.py +++ b/packages/flow/openadapt_flow/interactive_recorder.py @@ -2465,6 +2465,10 @@ def start(self) -> None: binding_target.add_init_script(init_js) self._init_js = init_js if self._owns_browser: + from openadapt_flow.backends.playwright_backend import prepare_ws_guard + + # Guard the page's WebSockets from its first document (wire check). + prepare_ws_guard(self.page) self.page.goto(self._url) try: self.page.wait_for_load_state("load") diff --git a/packages/flow/openadapt_flow/qualify/session.py b/packages/flow/openadapt_flow/qualify/session.py index 268a8d9990..1d66b53c59 100644 --- a/packages/flow/openadapt_flow/qualify/session.py +++ b/packages/flow/openadapt_flow/qualify/session.py @@ -61,6 +61,10 @@ def open_session() -> tuple[Any, Callable[[], None]]: service_workers="block", ) page = context.new_page() + from openadapt_flow.backends.playwright_backend import prepare_ws_guard + + # Guard the page's WebSockets from its first document (wire check). + prepare_ws_guard(page) page.goto(app.entry_url or app.url) except Exception: pw.stop() diff --git a/packages/flow/openadapt_flow/runtime/durable/attended_service.py b/packages/flow/openadapt_flow/runtime/durable/attended_service.py index e20be7c7e8..558a426e95 100644 --- a/packages/flow/openadapt_flow/runtime/durable/attended_service.py +++ b/packages/flow/openadapt_flow/runtime/durable/attended_service.py @@ -337,6 +337,10 @@ def replayer_for_manifest(manifest: Any) -> Any: browser = launch_chromium(playwright.chromium, headless=False) try: page = browser.new_page(viewport=cast(Any, _VIEWPORT)) + from openadapt_flow.backends.playwright_backend import prepare_ws_guard + + # Guard the page's WebSockets from its first document (wire check). + prepare_ws_guard(page) page.goto(backend_cfg.url) yield bound_executor(build_backend(backend_cfg, page=page)) finally: diff --git a/packages/flow/openadapt_flow/undo_proposal.py b/packages/flow/openadapt_flow/undo_proposal.py index ec10b25774..0c33d77687 100644 --- a/packages/flow/openadapt_flow/undo_proposal.py +++ b/packages/flow/openadapt_flow/undo_proposal.py @@ -1431,6 +1431,10 @@ def run( try: context = browser.new_context(viewport={"width": 1280, "height": 800}) page = context.new_page() + from openadapt_flow.backends.playwright_backend import prepare_ws_guard + + # Guard the page's WebSockets from its first document (wire check). + prepare_ws_guard(page) page.goto(backend_cfg.url) backend = build_backend(backend_cfg, page=page) return self._replay(backend, workflow, bundle_dir, run_dir, params) diff --git a/packages/flow/tests/test_create_form_identity.py b/packages/flow/tests/test_create_form_identity.py index ae74e49087..fbe698ecd9 100644 --- a/packages/flow/tests/test_create_form_identity.py +++ b/packages/flow/tests/test_create_form_identity.py @@ -339,7 +339,10 @@ def formattr_bundle(url: str, tmp_path_factory: pytest.TempPathFactory) -> Path: def _replay(url: str, bundle: Path, run_dir: Path) -> tuple[Any, list]: - from openadapt_flow.backends.playwright_backend import PlaywrightBackend + from openadapt_flow.backends.playwright_backend import ( + PlaywrightBackend, + prepare_ws_guard, + ) from openadapt_flow.runtime import Replayer since = len(SUBMISSIONS) @@ -347,6 +350,8 @@ def _replay(url: str, bundle: Path, run_dir: Path) -> tuple[Any, list]: browser = playwright.chromium.launch(headless=True) try: page = browser.new_page(viewport={"width": 1280, "height": 800}) + # As the runner does: guarded before the first document. + prepare_ws_guard(page.context) page.goto(url) page.wait_for_selector("[name=field_summary]", state="visible") report = Replayer(PlaywrightBackend(page)).run( diff --git a/packages/flow/tests/test_create_form_wire.py b/packages/flow/tests/test_create_form_wire.py index 5d9afaf672..df95b967fe 100644 --- a/packages/flow/tests/test_create_form_wire.py +++ b/packages/flow/tests/test_create_form_wire.py @@ -18,6 +18,7 @@ import http.server import json import threading +import time from pathlib import Path from typing import Any, Callable, Iterator from urllib.parse import parse_qs, parse_qsl, urlsplit @@ -250,7 +251,10 @@ def _replay( *, service_workers: str = "allow", ) -> tuple[Any, list]: - from openadapt_flow.backends.playwright_backend import PlaywrightBackend + from openadapt_flow.backends.playwright_backend import ( + PlaywrightBackend, + prepare_ws_guard, + ) from openadapt_flow.runtime import Replayer before = len(server.received) @@ -261,6 +265,8 @@ def _replay( viewport={"width": 1280, "height": 800}, service_workers=service_workers, # type: ignore[arg-type] ) + # As the runner does: guarded before the first document. + prepare_ws_guard(page.context) page.goto(f"{server.url}?v={variant}") page.wait_for_selector("[name=field_summary]", state="visible") if variant == "sw" and service_workers == "allow": @@ -373,134 +379,98 @@ def test_a_websocket_opened_by_the_save_halts_the_run( assert "WebSocket" in errors, errors -@pytest.mark.parametrize("decide", ["at_once", "after_settle", "late_route"]) -def test_a_websocket_opened_while_held_never_connects( - server: _Server, monkeypatch: pytest.MonkeyPatch, decide: str -) -> None: - """A socket the page opens while the save is held never reaches the - server, however late the backend services it. The page loaded before - the backend existed (as the runner hands pages over), which Playwright - 1.61 does not route; ``late_route`` runs the socket's route callback - only after the run decided its outcome and disarmed, as a loaded host - can.""" +def _browser_page(playwright: Any, *, guarded: bool = True) -> Any: + """A page in a fresh context, guarded before its first document (as the + runner's pages are) unless ``guarded`` is False.""" - from openadapt_flow.backends.playwright_backend import PlaywrightBackend + from openadapt_flow.backends.playwright_backend import prepare_ws_guard + + browser = playwright.chromium.launch(headless=True) + page = browser.new_context(service_workers="block").new_page() + if guarded: + prepare_ws_guard(page.context) + return browser, page + + +def _expected(server: _Server) -> Any: from openadapt_flow.wire_check import ExpectedSubmission - deferred: list[tuple[Any, Any]] = [] - route = PlaywrightBackend._route_websocket - if decide == "late_route": - monkeypatch.setattr( - PlaywrightBackend, - "_route_websocket", - lambda self, ws: deferred.append((self, ws)), - ) - before = len(server.received) origin = server.url.rsplit("/", 1)[0] - with sync_api.sync_playwright() as playwright: - browser = playwright.chromium.launch(headless=True) - try: - page = browser.new_page() - page.goto(f"{server.url}?v=fetch") - backend = PlaywrightBackend(page) - held = backend._arm_wire_check( - ExpectedSubmission(method="POST", address=f"{origin}/ticket") - ) - assert held is None, held - opened = page.evaluate( - """(u) => { - try { - const ws = new WebSocket(u); - ws.onopen = () => ws.send('close-account'); - return 'constructed'; - } catch (e) { return 'refused'; } - }""", - f"ws://{origin.split('://', 1)[1]}/ws", - ) - if decide == "after_settle": - backend.wait_form_submit_guard(500) - refusal = backend.form_submit_guard_refusal(disarm=True) - assert not backend.form_submit_guard_armed - for owner, ws in deferred: - route(owner, ws) - page.wait_for_timeout(1000) - finally: - browser.close() - received = server.received[before:] - assert "/ws" not in [r["path"] for r in received], (opened, received) - assert refusal is not None and "WebSocket" in refusal, (opened, refusal) + return ExpectedSubmission(method="POST", address=f"{origin}/ticket") -def test_websockets_connect_again_once_the_hold_is_released(server: _Server) -> None: - """The guard refuses sockets only while a save is held: a document - loaded during the hold starts held, and once the outcome is read every - document (old or new) connects its sockets as before.""" +def _hold(backend: Any, server: _Server) -> None: + held = backend._arm_wire_check(_expected(server)) + assert held is None, held - from openadapt_flow.backends.playwright_backend import PlaywrightBackend - from openadapt_flow.wire_check import ExpectedSubmission - origin = server.url.rsplit("/", 1)[0] - ws_url = f"ws://{origin.split('://', 1)[1]}/ws" - open_socket = """([u, m]) => { - try { - const ws = new WebSocket(u); - ws.onopen = () => ws.send(m); - return 'constructed'; - } catch (e) { return 'refused'; } - }""" - before = len(server.received) - with sync_api.sync_playwright() as playwright: - browser = playwright.chromium.launch(headless=True) - try: - page = browser.new_context().new_page() - page.goto(f"{server.url}?v=fetch") - backend = PlaywrightBackend(page) - assert ( - backend._arm_wire_check( - ExpectedSubmission(method="POST", address=f"{origin}/ticket") - ) - is None - ) - popup = page.context.new_page() - popup.goto(f"{server.url}?v=plain") - assert popup.evaluate(open_socket, [ws_url, "held"]) == "refused" - refusal = backend.form_submit_guard_refusal(disarm=True) - assert refusal is not None and "WebSocket" in refusal - assert page.evaluate(open_socket, [ws_url, "old"]) == "constructed" - assert popup.evaluate(open_socket, [ws_url, "popup"]) == "constructed" - fresh = page.context.new_page() - fresh.goto(f"{server.url}?v=plain") - assert fresh.evaluate(open_socket, [ws_url, "new"]) == "constructed" - page.wait_for_timeout(1500) - finally: - browser.close() - sent = sorted( +def _ws_url(server: _Server) -> str: + return "ws://" + server.url.split("://", 1)[1].rsplit("/", 1)[0] + "/ws" + + +def _leaked(server: _Server, before: int, marker: str) -> list[dict[str, Any]]: + return [r for r in server.received[before:] if marker in json.dumps(r)] + + +def _messages(server: _Server, before: int) -> list[str]: + return [ r["fields"]["message"] for r in server.received[before:] if r["path"] == "/ws" - ) - assert sent == ["new", "old", "popup"], sent + ] -# Ways a page reaches a WebSocket constructor other than the global, each -# run while a save is held (the #1581 review's probes). ``M`` is sent on -# whatever socket results; it must never reach the server. +# The page keeps the browser's own constructors and an open socket, then +# reaches for a connection every other way while a save is held (the #1581 +# reviews' probes). ``M`` marks what it sends; it must never arrive. +_KEEP_JS = """(U) => new Promise((done) => { + window.__kept = WebSocket; window.__keptWorker = Worker; + window.__keptSend = WebSocket.prototype.send; + const s = new WebSocket(U); window.__open = s; + s.onopen = () => done(true); setTimeout(() => done(false), 3000); +})""" +_WORKER_SRC = ( + "`const s = new WebSocket(${JSON.stringify(U)});" + " s.onopen = () => s.send(${JSON.stringify(M)});`" +) _REACHES = { "prototype_constructor": "new WebSocket.prototype.constructor(U)", "constructor_prototype": "new (Object.getPrototypeOf(WebSocket))(U)", "instance_constructor": ( - "(() => { const C = Object.getPrototypeOf(WebSocket.prototype)" - ".constructor; return new C(U); })()" + "new (Object.getPrototypeOf(WebSocket.prototype).constructor)(U)" ), "new_iframe": ( "(() => { const f = document.createElement('iframe');" " document.body.appendChild(f); return new f.contentWindow.WebSocket(U); })()" ), + "new_iframe_prototype": ( + "(() => { const f = document.createElement('iframe');" + " document.body.appendChild(f);" + " return new f.contentWindow.WebSocket.prototype.constructor(U); })()" + ), + "blob_iframe": ( + "(() => { const f = document.createElement('iframe');" + " f.src = URL.createObjectURL(new Blob([`