diff --git a/openadapt_flow/backends/playwright_backend.py b/openadapt_flow/backends/playwright_backend.py index 478cc1fd..534fed86 100644 --- a/openadapt_flow/backends/playwright_backend.py +++ b/openadapt_flow/backends/playwright_backend.py @@ -12,6 +12,7 @@ import hmac import math import re +import time import uuid from dataclasses import dataclass, field from datetime import datetime, timezone @@ -33,7 +34,11 @@ ) VIEWPORT: tuple[int, int] = (1280, 800) -_MASKED_SCREENSHOT_ATTEMPTS = 3 +# Secret-masked screenshots must not keep bytes from a changing frame tree. +# Login pages often attach and detach iframes for a few seconds after a click. +# Retry until the tree is quiet. Still refuse if it never settles. +_MASKED_SCREENSHOT_TIMEOUT_S = 8.0 +_MASKED_SCREENSHOT_RETRY_SLEEP_S = 0.05 _MODIFIER_ALIASES = { "meta": "Meta", @@ -2292,18 +2297,26 @@ def _same_frames(left: tuple[Any, ...], right: tuple[Any, ...]) -> bool: def screenshot(self) -> bytes: """Return a stable current full-viewport frame as PNG bytes.""" - if self._screenshot_guard is not None: - self._screenshot_guard() base_options: dict[str, Any] = {} if self._screenshot_scale == "css": base_options["scale"] = "css" if not self._screenshot_mask_selectors: + if self._screenshot_guard is not None: + self._screenshot_guard() return self.page.screenshot(type="png", full_page=False, **base_options) - for _attempt in range(_MASKED_SCREENSHOT_ATTEMPTS): + deadline = time.monotonic() + _MASKED_SCREENSHOT_TIMEOUT_S + while True: + # A retry can observe a different secret boundary even after the + # frame tree settles. Rebind or refuse before every capture. + if self._screenshot_guard is not None: + self._screenshot_guard() generation = self._screenshot_frame_generation frames = tuple(self.page.frames) if generation != self._screenshot_frame_generation: + if time.monotonic() >= deadline: + break + time.sleep(_MASKED_SCREENSHOT_RETRY_SLEEP_S) continue options = dict(base_options) options["mask"] = [ @@ -2319,8 +2332,15 @@ def screenshot(self) -> bytes: self.page.evaluate("() => null") except Exception: if generation != self._screenshot_frame_generation: + if time.monotonic() >= deadline: + break + time.sleep(_MASKED_SCREENSHOT_RETRY_SLEEP_S) continue raise + # A closed root can appear during capture without changing the + # frame tree. Its privacy refusal must escape the retry handler. + if self._screenshot_guard is not None: + self._screenshot_guard() current_frames = tuple(self.page.frames) if generation == self._screenshot_frame_generation and self._same_frames( frames, current_frames @@ -2328,6 +2348,9 @@ def screenshot(self) -> bytes: return png # ``png`` is intentionally discarded here. It never reaches the # recorder, disk, or a compiled bundle. + if time.monotonic() >= deadline: + break + time.sleep(_MASKED_SCREENSHOT_RETRY_SLEEP_S) raise ScreenshotMaskStabilityError( "the browser frame tree changed during every secret-masked " "screenshot attempt; recording was refused" diff --git a/tests/test_browser_attach.py b/tests/test_browser_attach.py index d23199b5..6c5b1adb 100644 --- a/tests/test_browser_attach.py +++ b/tests/test_browser_attach.py @@ -268,6 +268,122 @@ def screenshot(self, **kwargs): assert not any(page.listeners.values()) +def test_masked_screenshot_survives_three_iframe_storms_then_settles( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Reproduce the ServiceNow Sign-in refuse with a fake page. + + After Sign in, a login SPA attaches and detaches iframes. The old loop + tried three captures with no wait and then raised + ScreenshotMaskStabilityError. This page storms for those three captures + and then goes quiet. The record must keep a masked frame, not stop. + """ + + class Frame: + def __init__(self, name: str) -> None: + self.name = name + + def locator(self, selector): + return f"locator:{self.name}:{selector}" + + class Page: + def __init__(self) -> None: + self.frames = [Frame("main")] + self.listeners: dict[str, list] = {} + self.capture_count = 0 + + def on(self, event, listener): + self.listeners.setdefault(event, []).append(listener) + + def remove_listener(self, event, listener): + self.listeners[event].remove(listener) + + def evaluate(self, _script): + return None + + def screenshot(self, **kwargs): + self.capture_count += 1 + if self.capture_count <= 3: + frame = Frame(f"storm-{self.capture_count}") + self.frames.append(frame) + for listener in self.listeners.get("frameattached", []): + listener(frame) + self.frames.pop() + for listener in self.listeners.get("framedetached", []): + listener(frame) + return b"png" + + monkeypatch.setattr( + "openadapt_flow.backends.playwright_backend._MASKED_SCREENSHOT_RETRY_SLEEP_S", + 0.0, + ) + page = Page() + backend = PlaywrightBackend( # type: ignore[arg-type] + page, + screenshot_mask_selectors=("input[type='password']",), + ) + assert backend.screenshot() == b"png" + assert page.capture_count >= 4 + backend.stop_screenshot_mask_tracking() + + +def test_masked_screenshot_still_refuses_a_frame_tree_that_never_settles( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A tree that never goes quiet must still refuse. Do not keep those bytes.""" + + class Frame: + def __init__(self, name: str) -> None: + self.name = name + + def locator(self, selector): + return f"locator:{self.name}:{selector}" + + class Page: + def __init__(self) -> None: + self.frames = [Frame("main")] + self.listeners: dict[str, list] = {} + self.capture_count = 0 + + def on(self, event, listener): + self.listeners.setdefault(event, []).append(listener) + + def remove_listener(self, event, listener): + self.listeners[event].remove(listener) + + def evaluate(self, _script): + return None + + def screenshot(self, **kwargs): + self.capture_count += 1 + frame = Frame(f"churn-{self.capture_count}") + self.frames.append(frame) + for listener in self.listeners.get("frameattached", []): + listener(frame) + self.frames.pop() + for listener in self.listeners.get("framedetached", []): + listener(frame) + return b"png" + + monkeypatch.setattr( + "openadapt_flow.backends.playwright_backend._MASKED_SCREENSHOT_TIMEOUT_S", + 0.0, + ) + monkeypatch.setattr( + "openadapt_flow.backends.playwright_backend._MASKED_SCREENSHOT_RETRY_SLEEP_S", + 0.0, + ) + page = Page() + backend = PlaywrightBackend( # type: ignore[arg-type] + page, + screenshot_mask_selectors=("input[type='password']",), + ) + with pytest.raises(ScreenshotMaskStabilityError, match="frame tree"): + backend.screenshot() + assert page.capture_count >= 1 + backend.stop_screenshot_mask_tracking() + + def test_declared_secret_selectors_use_css_string_escaping() -> None: selectors = _secret_screenshot_selectors({"päss", 'quote"\\line\nend'}) @@ -1024,6 +1140,86 @@ def test_launched_browser_refuses_static_unbound_closed_shadow_password( assert not output.exists() +@pytest.mark.timeout(60) +@pytest.mark.parametrize("injection_point", ("before_retry", "during_capture")) +@pytest.mark.parametrize("trial", range(3)) +def test_masked_screenshot_rechecks_new_closed_shadow_secret_boundaries( + attach_app_url: str, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + injection_point: str, + trial: int, +) -> None: + """No retry may retain a secret that appears after the first privacy scan.""" + + if _chromium_executable() is None: + pytest.skip("no Chromium executable is installed") + output = tmp_path / f"new-closed-boundary-{injection_point}-{trial}" + session = InteractiveRecorder( + attach_app_url, + output, + headless=True, + secret_fields=("retry-secret",), + ) + session.start() + try: + page, backend = session.page, session.backend + assert page is not None and backend is not None + original_screenshot = page.screenshot + original_sleep = time.sleep + captures = 0 + inserted = False + + def insert_unbound_secret() -> None: + nonlocal inserted + inserted = True + page.evaluate( + """() => { + const host = document.createElement('x-unbound-secret'); + document.body.appendChild(host); + const root = host.attachShadow({mode: 'closed'}); + const input = document.createElement('input'); + input.name = 'retry-secret'; + input.value = 'SYNTHETIC-RETRY-SECRET'; + root.appendChild(input); + }""" + ) + + def screenshot(**kwargs): + nonlocal captures + captures += 1 + if injection_point == "before_retry" and captures <= 3: + page.evaluate( + """() => { + const frame = document.createElement('iframe'); + document.body.appendChild(frame); + frame.remove(); + }""" + ) + elif injection_point == "during_capture" and captures == 1: + insert_unbound_secret() + return original_screenshot(**kwargs) + + def retry_sleep(seconds: float) -> None: + if injection_point == "before_retry" and captures == 3 and not inserted: + insert_unbound_secret() + original_sleep(seconds) + + with monkeypatch.context() as patch: + patch.setattr(page, "screenshot", screenshot) + patch.setattr( + "openadapt_flow.backends.playwright_backend.time.sleep", + retry_sleep, + ) + with pytest.raises(BrowserAttachError, match="closed shadow root"): + backend.screenshot() + assert inserted + assert captures == (3 if injection_point == "before_retry" else 1) + finally: + session.abort() + assert not output.exists() + + @pytest.mark.timeout(30) def test_page_closure_scrubs_replaced_prefilled_and_reflected_secrets() -> None: """Real Chromium proves the page-local guard before screenshot handling.""" @@ -2284,9 +2480,45 @@ def attach_and_detach_during_every_capture(**kwargs): "screenshot", attach_and_detach_during_every_capture, ) + patch_context.setattr( + "openadapt_flow.backends.playwright_backend._MASKED_SCREENSHOT_TIMEOUT_S", + 0.0, + ) + patch_context.setattr( + "openadapt_flow.backends.playwright_backend._MASKED_SCREENSHOT_RETRY_SLEEP_S", + 0.0, + ) with pytest.raises(ScreenshotMaskStabilityError, match="frame tree"): race_backend.screenshot() - assert churn_attempts == 3 + assert churn_attempts >= 1 + + settle_attempts = 0 + + def churn_twice_then_keep(**kwargs): + nonlocal settle_attempts + settle_attempts += 1 + if settle_attempts <= 2: + race_page.evaluate( + """attempt => { + const frame = document.createElement('iframe'); + frame.id = `settle-${attempt}`; + frame.srcdoc = ''; + document.body.appendChild(frame); + frame.remove(); + }""", + settle_attempts, + ) + return original_screenshot(**kwargs) + + with monkeypatch.context() as patch_context: + patch_context.setattr( + race_page, + "screenshot", + churn_twice_then_keep, + ) + png = race_backend.screenshot() + assert png + assert settle_attempts >= 3 finally: if frame_race_session.page is not None: frame_race_session.page.evaluate(