diff --git a/packages/capture/openadapt_capture/recorder.py b/packages/capture/openadapt_capture/recorder.py index 38af558e59..2b50f1a655 100644 --- a/packages/capture/openadapt_capture/recorder.py +++ b/packages/capture/openadapt_capture/recorder.py @@ -87,6 +87,7 @@ observe_structural_action, ) from openadapt_capture.window_capture import ( + FOCUS_EVIDENCE_WINDOW, WindowCaptureAmbiguousError, WindowCaptureError, WindowCapturePermissionError, @@ -361,20 +362,26 @@ def __init__( self._keyboard_focus_at_receipt = keyboard_focus_at_receipt self._admitted = False - def admit(self, point: tuple[float, float] | None) -> None: + def admit(self, point: tuple[float, float] | None) -> str: """Accept input only if the target window received it. + Returns: + The focus evidence level (see ``assert_input_on_target``). + Raises: WindowInputNotOnTarget: The input went to another window. """ if point is None: - self._window_scope.assert_input_on_target( + evidence = self._window_scope.assert_input_on_target( self._geometry[0], keyboard_focus_at_receipt=self._keyboard_focus_at_receipt, ) else: - self._window_scope.assert_input_on_target(self._geometry[0], point=point) + evidence = self._window_scope.assert_input_on_target( + self._geometry[0], point=point + ) self._admitted = True + return evidence def admit_continuation(self) -> None: """Accept the release or drag of a press the target already received.""" @@ -469,8 +476,14 @@ class WindowInputGapLog: input kind, a timestamp and, for pointer input, a point rounded to a coarse grid. It never holds key identity or text. + A ``keyboard_focus_evidence`` marker is written when the evidence for + accepted keyboard input changes level: ``process`` means the OS named + only the target's process, not the target window, so the input could + have gone to another window of the same app. + Markers are appended to ``window-input-gaps.jsonl`` in the capture - directory, created only when input is dropped, and kept in ``records``. + directory, created only when input is dropped or its evidence weakens, + and kept in ``records``. """ def __init__(self, capture_dir: str | os.PathLike[str] | None = None) -> None: @@ -479,6 +492,7 @@ def __init__(self, capture_dir: str | os.PathLike[str] | None = None) -> None: ) self._lock = threading.Lock() self._active_reason: str | None = None + self._keyboard_evidence = FOCUS_EVIDENCE_WINDOW self.records: list[dict[str, Any]] = [] def _append_locked(self, record: dict[str, Any]) -> None: @@ -528,6 +542,21 @@ def left_target( reason, ) + def keyboard_evidence(self, evidence: str, timestamp: float | None) -> None: + """Mark a change in how accepted keyboard input reached the target.""" + with self._lock: + if evidence == self._keyboard_evidence: + return + self._keyboard_evidence = evidence + self._append_locked( + { + "schema_version": WINDOW_INPUT_GAP_SCHEMA_VERSION, + "event": "keyboard_focus_evidence", + "evidence": evidence, + "timestamp": timestamp, + } + ) + def returned(self, timestamp: float | None) -> None: """Close the current run of dropped input, if any.""" with self._lock: @@ -2648,18 +2677,20 @@ def admit(self, event: ObservedInput, reservation: object | None) -> bool: point = None if isinstance(event, ObservedKey) else (event.x, event.y) try: if isinstance(reservation, WindowActionReservation): - reservation.admit(point) + evidence = reservation.admit(point) else: window = self._scope.reserve_action_geometry()[0] if point is None: - self._scope.assert_input_on_target(window) + evidence = self._scope.assert_input_on_target(window) else: - self._scope.assert_input_on_target(window, point=point) + evidence = self._scope.assert_input_on_target(window, point=point) except WindowInputNotOnTarget as exc: self._discard(reservation) self.dropped(event, exc.reason) return False + if isinstance(event, ObservedKey): + self._gap_log.keyboard_evidence(evidence, event.timestamp) if isinstance(event, ObservedKey): self._held_keys.add(_key_identity(event)) elif isinstance(event, ObservedMouseButton): diff --git a/packages/capture/openadapt_capture/window_capture.py b/packages/capture/openadapt_capture/window_capture.py index 6931d3d02d..9d9a10fa31 100644 --- a/packages/capture/openadapt_capture/window_capture.py +++ b/packages/capture/openadapt_capture/window_capture.py @@ -147,11 +147,17 @@ class WindowCaptureUnavailableError(WindowCaptureError): INPUT_KEYBOARD_FOCUS_ELSEWHERE = "keyboard_focus_elsewhere" +INPUT_KEYBOARD_FOCUS_IN_DIALOG = "keyboard_focus_in_dialog" INPUT_KEYBOARD_FOCUS_UNKNOWN = "keyboard_focus_unknown" INPUT_POINTER_OVER_OTHER_WINDOW = "pointer_over_other_window" INPUT_POINTER_TARGET_UNKNOWN = "pointer_target_unknown" INPUT_TARGET_NOT_ON_SCREEN = "target_not_on_screen" _NOT_PROBED = object() +# How an accepted input was shown to reach the target. ``window``: the OS +# named the target window as the receiver. ``process``: it named only the +# target's process, which cannot rule out another window of the same app. +FOCUS_EVIDENCE_WINDOW = "window" +FOCUS_EVIDENCE_PROCESS = "process" class WindowInputNotOnTarget(WindowCaptureError): @@ -643,13 +649,21 @@ def assert_input_on_target( *, point: tuple[float, float] | None = None, keyboard_focus_at_receipt: object = _NOT_PROBED, - ) -> None: + ) -> str: """Refuse input that ``window`` did not receive. A pointer event (``point`` set) needs ``window`` topmost at the point. - A keyboard event needs keyboard focus on ``window`` or its process, - both at native receipt (when a receipt reading is given) and now. An - unknown answer or a probe failure is a refusal. + A keyboard event needs keyboard focus on ``window``, both at native + receipt (when a receipt reading is given) and now. A reading that + names a window must name ``window`` itself: another window of the + same process (a second browser window, an app's password dialog) is + a refusal. Only a reading that names no window can fall back to the + process, and a dialog or sheet never counts. An unknown answer or a + probe failure is a refusal. + + Returns: + ``FOCUS_EVIDENCE_WINDOW`` when some reading named the target + window, else ``FOCUS_EVIDENCE_PROCESS`` (weaker evidence). Raises: WindowInputNotOnTarget: With a stable reason code. @@ -667,7 +681,7 @@ def assert_input_on_target( raise WindowInputNotOnTarget(INPUT_POINTER_TARGET_UNKNOWN) if owner.window_id != window.window_id: raise WindowInputNotOnTarget(INPUT_POINTER_OVER_OTHER_WINDOW) - return + return FOCUS_EVIDENCE_WINDOW try: live = self._focus_probe.keyboard_focus() except Exception: @@ -675,16 +689,21 @@ def assert_input_on_target( readings = [live] if keyboard_focus_at_receipt is not _NOT_PROBED: readings.append(keyboard_focus_at_receipt) + evidence = FOCUS_EVIDENCE_PROCESS for owner in readings: if not isinstance(owner, FocusOwner): raise WindowInputNotOnTarget(INPUT_KEYBOARD_FOCUS_UNKNOWN) if owner.window_id is None and owner.pid is None: raise WindowInputNotOnTarget(INPUT_KEYBOARD_FOCUS_UNKNOWN) - on_target = (owner.window_id is not None and owner.window_id == window.window_id) or ( - owner.pid is not None and owner.pid == window.pid - ) - if not on_target: + if owner.dialog: + raise WindowInputNotOnTarget(INPUT_KEYBOARD_FOCUS_IN_DIALOG) + if owner.window_id is not None: + if owner.window_id != window.window_id: + raise WindowInputNotOnTarget(INPUT_KEYBOARD_FOCUS_ELSEWHERE) + evidence = FOCUS_EVIDENCE_WINDOW + elif owner.pid != window.pid: raise WindowInputNotOnTarget(INPUT_KEYBOARD_FOCUS_ELSEWHERE) + return evidence def _geometry_for_action( self, diff --git a/packages/capture/openadapt_capture/window_input_focus.py b/packages/capture/openadapt_capture/window_input_focus.py index a07b37940a..511bd7038a 100644 --- a/packages/capture/openadapt_capture/window_input_focus.py +++ b/packages/capture/openadapt_capture/window_input_focus.py @@ -28,10 +28,15 @@ class FocusOwner: ``window_id`` uses the same identifier space as :attr:`openadapt_capture.window_capture.TargetWindow.window_id` on the same platform (CGWindowID, top-level HWND, or X11 client window). + A probe that can name the focused window must set it: the scope then + requires that exact window, and a matching ``pid`` alone is not enough. + ``dialog`` is set when focus is in a dialog or sheet, which never counts + as the target even when the target's own process owns it. """ window_id: int | None = None pid: int | None = None + dialog: bool = False class InputFocusProbe(Protocol): @@ -55,11 +60,57 @@ def _point_in_bounds(x: float, y: float, bounds: tuple[float, float, float, floa return left <= x < left + width and top <= y < top + height +# Quartz window levels: kCGNormalWindowLevel and kCGStatusWindowLevel. +# A window between them is raised above document windows: a floating panel +# (3), a modal panel such as an NSAlert or password prompt (8), a torn-off +# menu (3). The status level and above are the menu bar, status items and +# system overlays, which never take an app's keyboard focus. +_MACOS_NORMAL_LAYER = 0 +_MACOS_STATUS_LAYER = 25 +# Accessibility subroles of a focused window that is a dialog, not a +# document window. Typing there never counts as typing into the target. +_MACOS_DIALOG_SUBROLES = frozenset({"AXDialog", "AXSystemDialog"}) +_MACOS_AX_FOCUS_TIMEOUT_SECONDS = 0.25 +_MACOS_AX_MAX_ANCESTORS = 32 + + +def _macos_ax_window_number(element: object) -> int | None: + """Return the CGWindowID behind an AX window element, or ``None``. + + Uses the long-standing HIServices ``_AXUIElementGetWindow`` call. There + is no public API for this mapping; any failure is reported as unknown. + """ + try: + import ctypes + + import objc + + library = ctypes.CDLL( + "/System/Library/Frameworks/ApplicationServices.framework/ApplicationServices" + ) + get_window = library._AXUIElementGetWindow + get_window.restype = ctypes.c_int32 + get_window.argtypes = [ctypes.c_void_p, ctypes.POINTER(ctypes.c_uint32)] + number = ctypes.c_uint32(0) + if get_window(objc.pyobjc_id(element), ctypes.byref(number)) != 0: + return None + return int(number.value) or None + except Exception: + return None + + class MacOSInputFocusProbe: """macOS: frontmost application and front-to-back Quartz window list. Bounds come from ``kCGWindowBounds``: global screen points with a top-left origin, the same space the macOS resolver binds. + + The focused window is the front-most on-screen layer-0 window of the + frontmost application. A raised window of that application (a modal or + floating panel) in front of it takes keyboard focus, so it is reported + as a dialog. When Accessibility can name the application's + ``AXFocusedWindow``, it must agree with that window, and a focused dialog + or sheet is reported as such. """ @staticmethod @@ -74,7 +125,12 @@ def _frontmost_pid() -> int | None: @staticmethod def _onscreen_app_windows() -> Iterator[dict]: - """Yield on-screen layer-0 windows front to back, lazily.""" + """Yield on-screen app windows front to back, lazily. + + These are document windows (layer 0) and the raised windows above + them up to, not including, the status level: floating panels and + modal panels. Each item carries its ``layer``. + """ import Quartz options = ( @@ -82,7 +138,8 @@ def _onscreen_app_windows() -> Iterator[dict]: ) windows = Quartz.CGWindowListCopyWindowInfo(options, Quartz.kCGNullWindowID) for info in windows or []: - if int(info.get("kCGWindowLayer", 0) or 0) != 0: + layer = int(info.get("kCGWindowLayer", 0) or 0) + if not _MACOS_NORMAL_LAYER <= layer < _MACOS_STATUS_LAYER: continue # menu bar, status items, overlays if float(info.get("kCGWindowAlpha", 1.0) or 0.0) <= 0.0: continue # invisible windows do not receive clicks @@ -99,27 +156,97 @@ def _onscreen_app_windows() -> Iterator[dict]: "window_id": int(info.get("kCGWindowNumber", 0) or 0), "pid": int(info.get("kCGWindowOwnerPID", 0) or 0), "bounds": bounds, + "layer": layer, } - def keyboard_focus(self, *, at_receipt: bool = False) -> FocusOwner | None: - """Return the frontmost application's process. + @staticmethod + def _ax_focused_window(pid: int) -> tuple[int | None, bool] | None: + """Return ``(window number, is dialog)`` for the app's focused window. + + ``None`` when Accessibility cannot name a focused window (not + trusted, timed out, or the app exposes none). + """ + try: + import ApplicationServices as ax + + success = ax.kAXErrorSuccess + + def attribute(element: object, name: str) -> object | None: + error, value = ax.AXUIElementCopyAttributeValue(element, name, None) + return value if error == success else None + + application = ax.AXUIElementCreateApplication(pid) + if ( + ax.AXUIElementSetMessagingTimeout( + application, _MACOS_AX_FOCUS_TIMEOUT_SECONDS + ) + != success + ): + return None + window = attribute(application, "AXFocusedWindow") + if window is None or attribute(window, "AXRole") != "AXWindow": + return None + dialog = attribute(window, "AXSubrole") in _MACOS_DIALOG_SUBROLES + # A sheet hangs off its parent window. Focus inside it still + # names the parent as the focused window. + element = attribute(application, "AXFocusedUIElement") + for _ in range(_MACOS_AX_MAX_ANCESTORS): + if element is None or dialog: + break + if attribute(element, "AXRole") == "AXSheet": + dialog = True + break + element = attribute(element, "AXParent") + return _macos_ax_window_number(window), dialog + except Exception: + return None - Inside the event tap only the cheap ``NSWorkspace`` read runs. At - delivery the answer must also agree with the owner of the front-most - on-screen application window. ``NSWorkspace`` is updated through the - main run loop and can be stale in a process that does not run one, so - a disagreement is reported as unknown. + def keyboard_focus(self, *, at_receipt: bool = False) -> FocusOwner | None: + """Return the focused window and its process. + + Inside the event tap only the cheap ``NSWorkspace`` read runs and the + answer names the process alone. At delivery the front-most on-screen + application window must belong to that process. ``NSWorkspace`` is + updated through the main run loop and can be stale in a process that + does not run one, so a disagreement is reported as unknown. That + window is the focused window, unless one of the application's raised + windows is in front of it (a dialog), or Accessibility names a + different focused window (unknown) or a dialog or sheet. """ pid = self._frontmost_pid() if pid is None or at_receipt: return None if pid is None else FocusOwner(pid=pid) - front = next(self._onscreen_app_windows(), None) + raised: dict | None = None + front: dict | None = None + for info in self._onscreen_app_windows(): + if info["layer"] != _MACOS_NORMAL_LAYER: + if info["pid"] == pid and raised is None: + raised = info + continue + front = info + break if front is None or front["pid"] != pid: return None - return FocusOwner(pid=pid) + if raised is not None: + # A modal or floating panel of the app sits over its document + # windows. Keys typed while it is up go to the panel. + return FocusOwner(window_id=raised["window_id"] or None, pid=pid, dialog=True) + window_id = front["window_id"] or None + focused = self._ax_focused_window(pid) + if focused is not None: + ax_window_id, dialog = focused + if dialog: + return FocusOwner(window_id=ax_window_id or window_id, pid=pid, dialog=True) + if ax_window_id is not None and ax_window_id != window_id: + return None + return FocusOwner(window_id=window_id, pid=pid) def window_at(self, x: float, y: float) -> FocusOwner | None: - """Return the first front-to-back layer-0 window containing the point.""" + """Return the first front-to-back app window containing the point. + + A raised panel over the target (a modal password prompt) is returned + as itself, so a click on it never counts as a click on the target. + """ for info in self._onscreen_app_windows(): if _point_in_bounds(x, y, info["bounds"]): return FocusOwner(window_id=info["window_id"], pid=info["pid"]) diff --git a/packages/capture/scripts/linux_xvfb_focus_check.py b/packages/capture/scripts/linux_xvfb_focus_check.py new file mode 100644 index 0000000000..2bed1801ec --- /dev/null +++ b/packages/capture/scripts/linux_xvfb_focus_check.py @@ -0,0 +1,173 @@ +"""Check window-scoped input focus against real X11 windows. + +Run by ``linux_xvfb_quickstart.sh`` inside its container, under Xvfb with +openbox and xdotool. This process opens three windows of its own: the +target, a second window, and a password dialog marked transient for the +target. All three share one process, as a browser's windows do. The check +pins the target with the real resolver, focuses each window with +``xdotool windowactivate`` (an EWMH request the window manager carries +out), and asks the real X11 probe and the scope whether input reached the +target. + +It prints one JSON line of observations and exits non-zero on a mismatch. +""" + +from __future__ import annotations + +import json +import os +import struct +import subprocess +import sys +import time + +import psutil +import xcffib +import xcffib.xproto as xproto + +from openadapt_capture.window_capture import ( + FOCUS_EVIDENCE_WINDOW, + INPUT_KEYBOARD_FOCUS_ELSEWHERE, + INPUT_POINTER_OVER_OTHER_WINDOW, + WindowCaptureScope, + WindowInputNotOnTarget, + WindowTarget, +) +from openadapt_capture.window_capture_linux import X11CompositeClient +from openadapt_capture.window_input_focus import X11InputFocusProbe + + +def main() -> int: + conn = xcffib.connect(display=os.environ["DISPLAY"]) + screen = conn.get_setup().roots[conn.pref_screen] + + def atom(name: str) -> int: + return int(conn.core.InternAtom(False, len(name), name).reply().atom) + + def set_property(window: int, name: int, kind: int, fmt: int, values: bytes) -> None: + count = len(values) // (fmt // 8) + conn.core.ChangeProperty(xproto.PropMode.Replace, window, name, kind, fmt, count, values) + + def open_window(title: str, x: int, y: int, *, transient_for: int | None = None) -> int: + window = conn.generate_id() + conn.core.CreateWindow( + screen.root_depth, + window, + screen.root, + x, + y, + 400, + 300, + 0, + xproto.WindowClass.InputOutput, + screen.root_visual, + xproto.CW.BackPixel, + [screen.white_pixel], + ) + encoded = title.encode() + set_property(window, atom("_NET_WM_NAME"), atom("UTF8_STRING"), 8, encoded) + set_property(window, xproto.Atom.WM_NAME, xproto.Atom.STRING, 8, encoded) + set_property( + window, atom("_NET_WM_PID"), xproto.Atom.CARDINAL, 32, struct.pack("=I", os.getpid()) + ) + if transient_for is not None: + set_property( + window, + xproto.Atom.WM_TRANSIENT_FOR, + xproto.Atom.WINDOW, + 32, + struct.pack("=I", transient_for), + ) + set_property( + window, + atom("_NET_WM_WINDOW_TYPE"), + xproto.Atom.ATOM, + 32, + struct.pack("=I", atom("_NET_WM_WINDOW_TYPE_DIALOG")), + ) + conn.core.MapWindow(window) + conn.flush() + return window + + target_id = open_window("OA focus target", 40, 40) + second_id = open_window("OA focus second", 140, 120) + dialog_id = open_window("OA focus password", 200, 200, transient_for=target_id) + time.sleep(1.0) + # The window manager may place new windows where it likes. Overlap the + # second window with the target so a click can land on it over the target. + for window, x, y in ((target_id, 40, 40), (second_id, 140, 120)): + subprocess.run(["xdotool", "windowmove", "--sync", str(window), str(x), str(y)], check=True) + time.sleep(0.5) + + def activate(window: int) -> None: + subprocess.run(["xdotool", "windowactivate", "--sync", str(window)], check=True) + time.sleep(0.5) + + probe = X11InputFocusProbe() + scope = WindowCaptureScope( + WindowTarget(owner=psutil.Process().name(), title="OA focus target"), + capturer=lambda _window: None, + focus_probe=probe, + ) + target = scope.resolve() + observed: dict[str, object] = { + "ids": {"target": target_id, "second": second_id, "dialog": dialog_id}, + "pinned": target.window_id, + } + failures: list[str] = [] + if target.window_id != target_id: + failures.append("the resolver did not pin the target window") + + def refusal(**kwargs) -> str | None: + try: + scope.assert_input_on_target(target, **kwargs) + except WindowInputNotOnTarget as exc: + return exc.reason + return None + + activate(target_id) + focus = probe.keyboard_focus() + observed["focus on target"] = vars(focus) if focus else None + evidence = scope.assert_input_on_target(target) + if focus is None or focus.window_id != target_id or evidence != FOCUS_EVIDENCE_WINDOW: + failures.append("keys in the focused target were not accepted with window evidence") + + activate(second_id) + focus = probe.keyboard_focus() + observed["focus on second"] = vars(focus) if focus else None + if focus is None or focus.window_id != second_id: + failures.append("the probe did not name the second window") + if refusal() != INPUT_KEYBOARD_FOCUS_ELSEWHERE: + failures.append("keys in a second window of the same process were kept") + # The second window is now on top. Click inside it where it covers the target. + with X11CompositeClient() as client: + record = client.window_record(second_id) + if record is None: + failures.append("the second window has no record") + point = (0.0, 0.0) + else: + point = (float(record.bounds[0] + 20), float(record.bounds[1] + 20)) + tx, ty, tw, th = target.bounds + observed["overlap point inside target"] = tx <= point[0] < tx + tw and ty <= point[1] < ty + th + if not observed["overlap point inside target"]: + failures.append("the second window does not overlap the target") + owner = probe.window_at(*point) + observed["window at overlap"] = vars(owner) if owner else None + if refusal(point=point) != INPUT_POINTER_OVER_OTHER_WINDOW: + failures.append("a click on the second window over the target was kept") + + activate(dialog_id) + focus = probe.keyboard_focus() + observed["focus on dialog"] = vars(focus) if focus else None + if refusal() != INPUT_KEYBOARD_FOCUS_ELSEWHERE: + failures.append("keys in the target's own password dialog were kept") + + print(json.dumps(observed, sort_keys=True)) + for failure in failures: + print(f"focus check failed: {failure}", file=sys.stderr) + conn.disconnect() + return 1 if failures else 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/packages/capture/scripts/linux_xvfb_quickstart.sh b/packages/capture/scripts/linux_xvfb_quickstart.sh index 1252d5fa8c..43ded48163 100755 --- a/packages/capture/scripts/linux_xvfb_quickstart.sh +++ b/packages/capture/scripts/linux_xvfb_quickstart.sh @@ -8,7 +8,8 @@ # # It clicks into a terminal window, types "hello", stops the recorder the way # Ctrl+C does, and exits non-zero unless the saved recording holds that click -# and that text. Nothing leaves the container; the recording is discarded with +# and that text. It then runs scripts/linux_xvfb_focus_check.py, which checks +# window-scoped focus against three real windows of one process. Nothing leaves the container; the recording is discarded with # it. set -euo pipefail @@ -113,3 +114,7 @@ print(f"click at (200, 200): {len(clicks)}; typed text: {text!r}") if len(clicks) != 1 or text != "hello": raise SystemExit("the recording does not hold the click and the typed text") EOF + +# Window-scoped input: keys and clicks count only when the target window +# itself has them, not another window of the same process. +python /tmp/src/scripts/linux_xvfb_focus_check.py diff --git a/packages/capture/tests/macos_focus_helper.py b/packages/capture/tests/macos_focus_helper.py new file mode 100644 index 0000000000..d647191a8e --- /dev/null +++ b/packages/capture/tests/macos_focus_helper.py @@ -0,0 +1,92 @@ +"""A small Cocoa app for the live macOS focus check (test_window_input_focus.py). + +It opens two document windows and, on request, a panel at the modal-panel +level, the level an NSAlert password prompt uses. Commands arrive one per +line on stdin and each reply is one JSON line on stdout: + + front A | front B make that window key and the app frontmost + panel show the panel in front of window A + close-panel close the panel + quit exit +""" + +from __future__ import annotations + +import json +import os +import sys +import threading + +import AppKit +from Foundation import NSObject +from PyObjCTools import AppHelper + + +def _window(title: str, x: float) -> AppKit.NSWindow: + window = AppKit.NSWindow.alloc().initWithContentRect_styleMask_backing_defer_( + AppKit.NSMakeRect(x, 200, 420, 300), + AppKit.NSWindowStyleMaskTitled | AppKit.NSWindowStyleMaskClosable, + AppKit.NSBackingStoreBuffered, + False, + ) + window.setTitle_(title) + window.setReleasedWhenClosed_(False) + return window + + +class Controller(NSObject): + def setUp(self) -> None: + self.windows = {"A": _window("OA focus A", 100), "B": _window("OA focus B", 300)} + for window in self.windows.values(): + window.makeKeyAndOrderFront_(None) + self.panel = AppKit.NSPanel.alloc().initWithContentRect_styleMask_backing_defer_( + AppKit.NSMakeRect(160, 260, 300, 120), + AppKit.NSWindowStyleMaskTitled, + AppKit.NSBackingStoreBuffered, + False, + ) + self.panel.setTitle_("OA focus password") + self.panel.setLevel_(AppKit.NSModalPanelWindowLevel) + self.panel.setReleasedWhenClosed_(False) + + def handle_(self, line: str) -> None: + command = line.split() + app = AppKit.NSApplication.sharedApplication() + if command[:1] == ["front"]: + app.activateIgnoringOtherApps_(True) + self.windows[command[1]].makeKeyAndOrderFront_(None) + elif command == ["panel"]: + app.activateIgnoringOtherApps_(True) + self.panel.makeKeyAndOrderFront_(None) + elif command == ["close-panel"]: + self.panel.orderOut_(None) + elif command == ["quit"]: + app.terminate_(None) + return + reply = { + "pid": os.getpid(), + "A": int(self.windows["A"].windowNumber()), + "B": int(self.windows["B"].windowNumber()), + "panel": int(self.panel.windowNumber()), + } + print(json.dumps(reply), flush=True) + + +def _read_commands(controller: Controller) -> None: + for line in sys.stdin: + AppHelper.callAfter(controller.handle_, line.strip()) + AppHelper.callAfter(controller.handle_, "quit") + + +def main() -> None: + app = AppKit.NSApplication.sharedApplication() + app.setActivationPolicy_(AppKit.NSApplicationActivationPolicyRegular) + controller = Controller.alloc().init() + controller.setUp() + threading.Thread(target=_read_commands, args=(controller,), daemon=True).start() + AppHelper.callAfter(controller.handle_, "front A") + AppHelper.runEventLoop() + + +if __name__ == "__main__": + main() diff --git a/packages/capture/tests/test_window_input_focus.py b/packages/capture/tests/test_window_input_focus.py index 03f6f645d2..29aedefcc7 100644 --- a/packages/capture/tests/test_window_input_focus.py +++ b/packages/capture/tests/test_window_input_focus.py @@ -13,6 +13,7 @@ import multiprocessing import os import sqlite3 +import sys import threading import time import uuid @@ -35,7 +36,10 @@ WindowInputGapLog, ) from openadapt_capture.window_capture import ( + FOCUS_EVIDENCE_PROCESS, + FOCUS_EVIDENCE_WINDOW, INPUT_KEYBOARD_FOCUS_ELSEWHERE, + INPUT_KEYBOARD_FOCUS_IN_DIALOG, INPUT_KEYBOARD_FOCUS_UNKNOWN, INPUT_POINTER_OVER_OTHER_WINDOW, INPUT_POINTER_TARGET_UNKNOWN, @@ -49,6 +53,10 @@ SLACK = FocusOwner(window_id=7, pid=555) PASSWORD_DIALOG = FocusOwner(window_id=8, pid=666) +# The fake target is window 42 of process 1234. +SECOND_TARGET_APP_WINDOW = FocusOwner(window_id=43, pid=1234) +TARGET_APP_PASSWORD_WINDOW = FocusOwner(window_id=44, pid=1234) +TARGET_APP_PASSWORD_SHEET = FocusOwner(window_id=42, pid=1234, dialog=True) # Inside the fake target's rectangle (300, 150, 800, 600). INSIDE = (400.0, 300.0) @@ -418,6 +426,75 @@ def test_release_of_an_accepted_press_survives_a_focus_change(tmp_path, monkeypa assert recording.gaps() == [] +def test_second_window_of_the_target_app_drops_keys_and_never_stores_them( + tmp_path, monkeypatch, fake +): + """A second browser window shares the target's process but is not it.""" + canary = f"pw-canary-{uuid.uuid4().hex}" + recording = WindowRecording(tmp_path, monkeypatch, fake) + with recording as observer: + type_text(observer, "ok") + fake.keyboard_owner = SECOND_TARGET_APP_WINDOW + type_text(observer, "hunter2", canary=canary) + fake.keyboard_owner = TARGET + type_text(observer, "go") + + keys = _key_rows(recording.actions()) + assert "".join(row["key_char"] for row in keys if row["name"] == "press") == "okgo" + reasons = [gap["reason"] for gap in recording.gaps() if gap["event"] == "focus_left_target"] + assert reasons == [INPUT_KEYBOARD_FOCUS_ELSEWHERE] + recording.assert_nowhere(canary, "hunter2") + + +@pytest.mark.parametrize( + ("dialog", "reason"), + [ + (TARGET_APP_PASSWORD_WINDOW, INPUT_KEYBOARD_FOCUS_ELSEWHERE), + (TARGET_APP_PASSWORD_SHEET, INPUT_KEYBOARD_FOCUS_IN_DIALOG), + ], + ids=["own-window", "sheet-on-target"], +) +def test_password_dialog_of_the_target_app_drops_keys(tmp_path, monkeypatch, fake, dialog, reason): + """The target app's own password prompt never counts as the target. + + A prompt in its own window has another window id. A sheet attached to the + target reports the target's window id and is flagged as a dialog. + """ + canary = f"pw-canary-{uuid.uuid4().hex}" + recording = WindowRecording(tmp_path, monkeypatch, fake) + with recording as observer: + fake.keyboard_owner = dialog + type_text(observer, "s3cr3t!", canary=canary) + fake.keyboard_owner = TARGET + type_text(observer, "x") + + keys = _key_rows(recording.actions()) + assert [row["key_char"] for row in keys if row["name"] == "press"] == ["x"] + reasons = [gap["reason"] for gap in recording.gaps() if gap["event"] == "focus_left_target"] + assert reasons == [reason] + recording.assert_nowhere(canary, "s3cr3t!") + + +def test_process_only_focus_is_kept_and_marked_as_weaker_evidence(tmp_path, monkeypatch, fake): + """Where the OS names no window, a process match is kept and marked.""" + recording = WindowRecording(tmp_path, monkeypatch, fake) + with recording as observer: + type_text(observer, "a") + fake.keyboard_owner = FocusOwner(window_id=None, pid=fake.pid) + type_text(observer, "bc") + fake.keyboard_owner = TARGET + type_text(observer, "d") + + keys = _key_rows(recording.actions()) + assert "".join(row["key_char"] for row in keys if row["name"] == "press") == "abcd" + evidence = [gap for gap in recording.gaps() if gap["event"] == "keyboard_focus_evidence"] + assert [gap["evidence"] for gap in evidence] == [FOCUS_EVIDENCE_PROCESS, FOCUS_EVIDENCE_WINDOW] + assert all( + set(gap) == {"schema_version", "event", "evidence", "timestamp"} for gap in evidence + ) + recording.assert_nowhere("bc") + + # --------------------------------------------------------------------------- # Scope-level decisions (no pipeline) # --------------------------------------------------------------------------- @@ -462,6 +539,36 @@ def test_keyboard_focus_on_target_process_is_accepted(scope, fake): assert isinstance(scope.generation_for_action(), int) +def test_keyboard_focus_evidence_names_window_or_process(scope, fake): + window = scope.reserve_action_geometry()[0] + assert scope.assert_input_on_target(window) == FOCUS_EVIDENCE_WINDOW + fake.keyboard_owner = FocusOwner(window_id=None, pid=fake.pid) + assert scope.assert_input_on_target(window) == FOCUS_EVIDENCE_PROCESS + # A pid-only receipt reading plus a window reading at delivery is + # window evidence (the macOS event tap reads only the process). + fake.keyboard_owner = TARGET + assert ( + scope.assert_input_on_target( + window, keyboard_focus_at_receipt=FocusOwner(pid=fake.pid) + ) + == FOCUS_EVIDENCE_WINDOW + ) + + +def test_a_named_window_never_falls_back_to_the_process(scope, fake): + window = scope.reserve_action_geometry()[0] + fake.keyboard_owner = SECOND_TARGET_APP_WINDOW + with pytest.raises(WindowInputNotOnTarget) as excinfo: + scope.assert_input_on_target(window) + assert excinfo.value.reason == INPUT_KEYBOARD_FOCUS_ELSEWHERE + fake.keyboard_owner = TARGET + with pytest.raises(WindowInputNotOnTarget) as excinfo: + scope.assert_input_on_target( + window, keyboard_focus_at_receipt=SECOND_TARGET_APP_WINDOW + ) + assert excinfo.value.reason == INPUT_KEYBOARD_FOCUS_ELSEWHERE + + def test_legacy_translation_refuses_other_window(scope, fake): fake.pointer_owner = SLACK with pytest.raises(WindowInputNotOnTarget): @@ -522,8 +629,41 @@ def fake_macos(monkeypatch): appkit = SimpleNamespace( NSWorkspace=SimpleNamespace(sharedWorkspace=lambda: workspace), ) + # Accessibility is unavailable unless a test names a focused window. + state.ax_window = None # (window number, subrole, focused element role) + success = 0 + + def copy_attribute(element, name, _out): + if state.ax_window is None: + return (-25211, None) # kAXErrorAPIDisabled + number, subrole, focused_role = state.ax_window + values = { + ("app", "AXFocusedWindow"): "window", + ("app", "AXFocusedUIElement"): "focused", + ("window", "AXRole"): "AXWindow", + ("window", "AXSubrole"): subrole, + ("focused", "AXRole"): focused_role, + ("focused", "AXParent"): "window", + } + value = values.get((element, name)) + return (success, value) if value is not None else (-25212, None) + + applicationservices = SimpleNamespace( + kAXErrorSuccess=success, + AXUIElementCreateApplication=lambda _pid: "app", + AXUIElementSetMessagingTimeout=lambda _element, _seconds: success, + AXUIElementCopyAttributeValue=copy_attribute, + ) + import openadapt_capture.window_input_focus as focus_module + + monkeypatch.setattr( + focus_module, + "_macos_ax_window_number", + lambda element: state.ax_window[0] if element == "window" else None, + ) monkeypatch.setitem(sys.modules, "Quartz", quartz) monkeypatch.setitem(sys.modules, "AppKit", appkit) + monkeypatch.setitem(sys.modules, "ApplicationServices", applicationservices) return state @@ -535,7 +675,7 @@ def test_macos_probe_requires_window_list_to_agree_with_frontmost_app(fake_macos _quartz_window(1, 99, (0, 0, 2000, 25), layer=25), # menu bar extra _quartz_window(42, 1234, (300, 150, 800, 600)), ] - assert probe.keyboard_focus() == FocusOwner(pid=1234) + assert probe.keyboard_focus() == FocusOwner(window_id=42, pid=1234) # NSWorkspace can be stale; the front-most app window belongs to Slack. fake_macos.windows.insert(1, _quartz_window(7, 555, (0, 0, 500, 500))) assert probe.keyboard_focus() is None @@ -543,6 +683,51 @@ def test_macos_probe_requires_window_list_to_agree_with_frontmost_app(fake_macos assert probe.keyboard_focus(at_receipt=True) == FocusOwner(pid=1234) +def test_macos_probe_names_the_front_window_and_checks_accessibility(fake_macos): + from openadapt_capture.window_input_focus import MacOSInputFocusProbe + + probe = MacOSInputFocusProbe() + fake_macos.windows = [ + _quartz_window(43, 1234, (0, 0, 500, 500)), # second window, in front + _quartz_window(42, 1234, (300, 150, 800, 600)), + ] + # Without Accessibility the front-most window of the app is focused. + assert probe.keyboard_focus() == FocusOwner(window_id=43, pid=1234) + # Accessibility agrees. + fake_macos.ax_window = (43, "AXStandardWindow", "AXTextField") + assert probe.keyboard_focus() == FocusOwner(window_id=43, pid=1234) + # Accessibility names another window: unknown, so input is refused. + fake_macos.ax_window = (42, "AXStandardWindow", "AXTextField") + assert probe.keyboard_focus() is None + # A dialog window, or a sheet under the focused element, is a dialog. + fake_macos.ax_window = (43, "AXDialog", "AXSecureTextField") + assert probe.keyboard_focus() == FocusOwner(window_id=43, pid=1234, dialog=True) + fake_macos.ax_window = (43, "AXStandardWindow", "AXSheet") + assert probe.keyboard_focus() == FocusOwner(window_id=43, pid=1234, dialog=True) + + +def test_macos_probe_treats_the_apps_modal_panel_as_a_dialog(fake_macos): + """An NSAlert-style password panel sits above layer 0 and takes the keys. + + Accessibility is unavailable here, so only Quartz can see the panel. + """ + from openadapt_capture.window_input_focus import MacOSInputFocusProbe + + probe = MacOSInputFocusProbe() + fake_macos.windows = [ + _quartz_window(1, 99, (0, 0, 2000, 25), layer=25), # menu bar extra + _quartz_window(9, 1234, (500, 300, 400, 200), layer=8), # modal panel + _quartz_window(42, 1234, (300, 150, 800, 600)), + ] + assert probe.keyboard_focus() == FocusOwner(window_id=9, pid=1234, dialog=True) + # A click on the panel is a click on the panel, not on the target below. + assert probe.window_at(600.0, 400.0) == FocusOwner(window_id=9, pid=1234) + assert probe.window_at(310.0, 170.0) == FocusOwner(window_id=42, pid=1234) + # Another app's floating panel does not take the target app's keys. + fake_macos.windows[1] = _quartz_window(9, 777, (1500, 300, 200, 200), layer=3) + assert probe.keyboard_focus() == FocusOwner(window_id=42, pid=1234) + + def test_macos_probe_window_at_is_front_to_back_layer_zero(fake_macos): from openadapt_capture.window_input_focus import MacOSInputFocusProbe @@ -675,3 +860,90 @@ def test_x11_probe_reports_active_window_and_point_owner(): assert probe.keyboard_focus() == FocusOwner(window_id=42, pid=1234) assert probe.window_at(400.4, 300.9) == FocusOwner(window_id=7, pid=555) assert probe.window_at(0, 0) is None + + +# --------------------------------------------------------------------------- +# Live macOS check: two real windows of one app and a modal-level panel +# --------------------------------------------------------------------------- + + +_LIVE_MACOS = sys.platform == "darwin" and os.environ.get("OPENADAPT_CAPTURE_MACOS_FOCUS_CHECK") == "1" + + +@pytest.mark.slow +@pytest.mark.skipif( + not _LIVE_MACOS, + reason="set OPENADAPT_CAPTURE_MACOS_FOCUS_CHECK=1 on a logged-in macOS desktop", +) +def test_live_macos_focus_names_the_window_and_the_modal_panel(): + """Real Cocoa windows: the probe names the key window and flags the panel. + + Run by hand on a logged-in desktop; it activates a small helper app. + """ + import subprocess + + from openadapt_capture.window_capture import TargetWindow + from openadapt_capture.window_input_focus import MacOSInputFocusProbe + + helper = subprocess.Popen( + [sys.executable, str(Path(__file__).with_name("macos_focus_helper.py"))], + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + text=True, + ) + probe = MacOSInputFocusProbe() + + def send(command: str) -> dict: + assert helper.stdin is not None and helper.stdout is not None + helper.stdin.write(command + "\n") + helper.stdin.flush() + reply = json.loads(helper.stdout.readline()) + time.sleep(0.5) # let the window server apply the activation + return reply + + try: + assert helper.stdout is not None + ids = json.loads(helper.stdout.readline()) + target = TargetWindow( + window_id=ids["A"], + owner="Python", + title="OA focus A", + pid=ids["pid"], + bounds=(100.0, 200.0, 420.0, 300.0), + process_start_time=1.0, + ) + scope = WindowCaptureScope( + WindowTarget(owner="Python"), + resolver=lambda _target: target, + capturer=lambda _window: None, + focus_probe=probe, + ) + observed = {} + + send("front A") + observed["front A"] = probe.keyboard_focus() + assert observed["front A"] == FocusOwner(window_id=ids["A"], pid=ids["pid"]) + assert scope.assert_input_on_target(target) == FOCUS_EVIDENCE_WINDOW + + send("front B") + observed["front B"] = probe.keyboard_focus() + assert observed["front B"] == FocusOwner(window_id=ids["B"], pid=ids["pid"]) + with pytest.raises(WindowInputNotOnTarget) as excinfo: + scope.assert_input_on_target(target) + assert excinfo.value.reason == INPUT_KEYBOARD_FOCUS_ELSEWHERE + + send("front A") + send("panel") + observed["panel"] = probe.keyboard_focus() + assert observed["panel"] is not None and observed["panel"].dialog + with pytest.raises(WindowInputNotOnTarget) as excinfo: + scope.assert_input_on_target(target) + assert excinfo.value.reason == INPUT_KEYBOARD_FOCUS_IN_DIALOG + print(json.dumps({key: vars(value) for key, value in observed.items()})) + finally: + try: + send("quit") + except Exception: + pass + helper.kill() + helper.wait(timeout=10)