fix: wait for a quiet frame tree before keeping a secret-masked screenshot

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,15 +2332,25 @@ 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
             ):
                 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"

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 = '<input type="password" value="tmp">';
+                          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(