diff --git a/CHANGELOG.md b/CHANGELOG.md index 8ee909f..e9d32cd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,13 @@ ### Features +- Add `window_tree` structural observations and a fail-closed authoring + projector. The raw tree may persist on disk for compile, except password + and secure-field values. Native AX, UIA, AT-SPI, and ARIA roles map onto + the types ElementRole enum. The vendor-wire payload is + `openadapt.authoring.observe/v1` without values, titles, or screenshots. + The observe fixture is pinned to openadapt-types PR 35. OS-injected input + still does not persist. - Add attended authentication handoffs. Capture suppresses sensitive source data, seals a bounded method marker, and retains a fresh exact frame before native input resumes. The same retry-safe operations are available through diff --git a/docs/DESIGN.md b/docs/DESIGN.md index ad7b8db..85facb1 100644 --- a/docs/DESIGN.md +++ b/docs/DESIGN.md @@ -212,6 +212,18 @@ typelib/runtime, and an interactive desktop accessibility bus. The native provider describes the local accessibility tree. It does not describe controls inside an RDP or Citrix pixel stream. +Authoring adds `query_kind: "window_tree"`. The raw tree may persist on disk +for compile. Password and secure-field values are omitted from that tree. +`openadapt_capture.authoring_project` projects the tree into +`openadapt.authoring.observe/v1` for a vendor wire: no values, titles, +screenshots, or extra keys. Native AX, UIA, AT-SPI, and ARIA roles map onto +the types `ElementRole` enum; unmapped roles never reach the wire. The +observe fixture is pinned to `openadapt-types` PR 35. Names and automation +ids that fail the coach-hint bar (length, `://`, `@`, six or more digits, +phone, SSN, email) or the types projected-label grammar are dropped. RDP and +Citrix observe payloads are coach-only with an empty tree. Capture still +drops OS-injected input; there is no `record_injected` API. + ## Video and frame timing Capture does not import, link, or bundle FFmpeg, and it downloads nothing on diff --git a/openadapt_capture/__init__.py b/openadapt_capture/__init__.py index 83d5983..009afa3 100644 --- a/openadapt_capture/__init__.py +++ b/openadapt_capture/__init__.py @@ -25,6 +25,12 @@ FreshFrameProof, load_authentication_handoffs, ) +from openadapt_capture.authoring_project import ( + AUTHORING_OBSERVE_SCHEMA_VERSION, + AuthoringObserve, + AuthoringProjection, + project_authoring_observe, +) from openadapt_capture.browser_events import ( BoundingBox, BrowserClickEvent, @@ -135,9 +141,11 @@ StructuralObservationRequest, StructuralObserver, StructuralProcessIdentity, + StructuralTreeNode, StructuralWindowIdentity, create_structural_observer, observe_structural_action, + observe_window_tree, ) # Visualization @@ -188,9 +196,16 @@ "StructuralObservationRequest", "StructuralObserver", "StructuralProcessIdentity", + "StructuralTreeNode", "StructuralWindowIdentity", "create_structural_observer", "observe_structural_action", + "observe_window_tree", + # Authoring observe projector (vendor wire; no values/titles/screenshots) + "AUTHORING_OBSERVE_SCHEMA_VERSION", + "AuthoringObserve", + "AuthoringProjection", + "project_authoring_observe", # Window-scoped capture "WindowTarget", "TargetWindow", diff --git a/openadapt_capture/authoring_project.py b/openadapt_capture/authoring_project.py new file mode 100644 index 0000000..941af95 --- /dev/null +++ b/openadapt_capture/authoring_project.py @@ -0,0 +1,759 @@ +"""Fail-closed projector from a raw window tree to authoring observe JSON. + +The raw accessibility tree may persist on disk for compile. This module is the +only path that shapes that tree for a vendor wire. It never emits field values, +window titles, screenshots, or extra keys. +""" + +from __future__ import annotations + +import hashlib +import hmac +import json +import logging +import math +import re +from dataclasses import dataclass +from typing import Any, Literal + +from pydantic import BaseModel, ConfigDict, Field, model_validator + +from openadapt_capture.structural import ( + StructuralBounds, + StructuralObservation, + StructuralTreeNode, +) + +AUTHORING_OBSERVE_SCHEMA_VERSION = "openadapt.authoring.observe/v1" +MAX_AUTHORING_NODES = 200 +MAX_AUTHORING_WIRE_BYTES = 32 * 1024 +MAX_AUTHORING_LABEL_LENGTH = 80 +NODE_ID_PATTERN = r"^n_[0-9a-f]{8}$" +PROCESS_NAME_PATTERN = r"^[A-Za-z0-9 ._-]{1,64}$" +PROJECTED_LABEL_PATTERN = r"^[A-Za-z0-9][A-Za-z0-9 ._-]{0,79}$" + +AuthoringBackend = Literal["macos", "linux", "windows", "web", "rdp", "citrix"] +AuthoringProvider = Literal[ + "playwright_ax", + "macos_ax", + "windows_uia", + "linux_atspi", + "none", +] +AuthoringRole = Literal[ + "button", + "text_input", + "text_static", + "label", + "link", + "checkbox", + "radio", + "combobox", + "list_item", + "menu", + "menu_item", + "tab", + "tree_item", + "image", + "icon", + "toolbar", + "scrollbar", + "slider", + "window", + "dialog", + "group", + "table", + "table_cell", + "table_row", + "heading", + "paragraph", + "unknown", +] + +_AGENT_DRIVE_BACKENDS = frozenset({"macos", "linux", "web"}) +_COACH_ONLY_BACKENDS = frozenset({"windows", "rdp", "citrix"}) +_EMPTY_TREE_BACKENDS = frozenset({"rdp", "citrix"}) +_KNOWN_BACKENDS = frozenset({"macos", "linux", "windows", "web", "rdp", "citrix"}) +_KNOWN_PROVIDERS = frozenset( + {"playwright_ax", "macos_ax", "windows_uia", "linux_atspi", "none"} +) +_AUTHORING_ROLES = frozenset( + { + "button", + "text_input", + "text_static", + "label", + "link", + "checkbox", + "radio", + "combobox", + "list_item", + "menu", + "menu_item", + "tab", + "tree_item", + "image", + "icon", + "toolbar", + "scrollbar", + "slider", + "window", + "dialog", + "group", + "table", + "table_cell", + "table_row", + "heading", + "paragraph", + "unknown", + } +) + +_PROCESS_NAME = re.compile(PROCESS_NAME_PATTERN) +_PROJECTED_LABEL = re.compile(PROJECTED_LABEL_PATTERN) +_SIX_DIGITS = re.compile(r"\d{6,}") +_SSN = re.compile(r"(? "AuthoringNormalizedBounds": + """Reject non-finite or out-of-range overlay coordinates.""" + values = (self.x, self.y, self.w, self.h) + if not all(math.isfinite(value) for value in values): + raise ValueError("normalized bounds must be finite") + if any(value < 0 or value > 1 for value in values): + raise ValueError("normalized bounds must lie in [0, 1]") + if self.x + self.w > 1 + 1e-9 or self.y + self.h > 1 + 1e-9: + raise ValueError("normalized bounds exceed the viewport") + return self + + +class AuthoringPixelBounds(BaseModel): + """Screen-space rectangle for backend clicks. Laptop-only.""" + + model_config = ConfigDict(extra="forbid") + + x: float + y: float + w: float + h: float + + @model_validator(mode="after") + def validate_pixels(self) -> "AuthoringPixelBounds": + """Reject non-finite or inverted pixel rectangles.""" + values = (self.x, self.y, self.w, self.h) + if not all(math.isfinite(value) for value in values): + raise ValueError("pixel bounds must be finite") + if self.w < 0 or self.h < 0: + raise ValueError("pixel bounds must not be inverted") + return self + + +class AuthoringWireNode(BaseModel): + """One projected accessibility node on the vendor wire.""" + + model_config = ConfigDict(extra="forbid") + + node_id: str = Field(pattern=NODE_ID_PATTERN) + role: AuthoringRole + control_type: str | None = Field(default=None, pattern=PROJECTED_LABEL_PATTERN) + automation_id: str | None = Field(default=None, pattern=PROJECTED_LABEL_PATTERN) + name: str | None = Field(default=None, pattern=PROJECTED_LABEL_PATTERN) + class_name: str | None = Field(default=None, pattern=PROJECTED_LABEL_PATTERN) + enabled: bool + focused: bool + bounds: AuthoringNormalizedBounds + + +class AuthoringWindow(BaseModel): + """Projected top-level window identity. Titles never appear here.""" + + model_config = ConfigDict(extra="forbid") + + process_name: str = Field(pattern=PROCESS_NAME_PATTERN) + role: Literal["window"] = "window" + bounds: AuthoringNormalizedBounds + + +class AuthoringObserve(BaseModel): + """PHI-safe ``openadapt.authoring.observe/v1`` payload.""" + + model_config = ConfigDict(extra="forbid") + + schema_version: Literal["openadapt.authoring.observe/v1"] = ( + AUTHORING_OBSERVE_SCHEMA_VERSION + ) + backend: AuthoringBackend + provider: AuthoringProvider + mode: Literal["authoring"] = "authoring" + agent_drive: bool + coach_only: bool + recording: bool = False + window: AuthoringWindow | None = None + tree: list[AuthoringWireNode] = Field(default_factory=list, max_length=MAX_AUTHORING_NODES) + truncated: bool = False + node_count: int = Field(default=0, ge=0, le=MAX_AUTHORING_NODES) + reason: Literal["empty_projection"] | None = None + + +class AuthoringNodeTableEntry(BaseModel): + """Laptop-only click table. Never serialized onto the vendor wire.""" + + model_config = ConfigDict(extra="forbid") + + node_id: str = Field(pattern=NODE_ID_PATTERN) + backend_pixels: AuthoringPixelBounds | None = None + normalized: AuthoringNormalizedBounds | None = None + provider_runtime_id: str | None = None + observed_at: int + + +class AuthoringRawNode(BaseModel): + """Unprojected accessibility node. May contain PHI. Never send to MCP.""" + + model_config = ConfigDict(extra="forbid") + + provider_runtime_id: str | None = None + automation_id: str | None = None + role: str | None = None + control_type: str | None = None + name: str | None = None + class_name: str | None = None + value: str | None = None + title: str | None = None + enabled: bool | None = None + focused: bool | None = None + bounds: StructuralBounds | None = None + children: list[AuthoringRawNode] = Field(default_factory=list) + + +class AuthoringRawWindow(BaseModel): + """Unprojected window identity. Title is dropped by the projector.""" + + model_config = ConfigDict(extra="forbid") + + process_name: str | None = None + role: str | None = None + title: str | None = None + bounds: StructuralBounds | None = None + + +@dataclass(frozen=True) +class AuthoringProjection: + """Wire payload plus the laptop-only node table.""" + + observe: AuthoringObserve + node_table: tuple[AuthoringNodeTableEntry, ...] + + def wire_dict(self) -> dict[str, Any]: + """Return the vendor-wire object with omitted empty optional fields.""" + return self.observe.model_dump(mode="json", exclude_none=True) + + +def mint_node_id( + *, + hmac_key: bytes, + lease_nonce: bytes, + provider_runtime_id: str, +) -> str: + """Mint ``n_`` + 8 hex of HMAC-SHA256(lease_nonce || provider_runtime_id).""" + + if not hmac_key: + raise ValueError("hmac_key is required") + if not provider_runtime_id: + raise ValueError("provider_runtime_id is required") + digest = hmac.new( + hmac_key, + lease_nonce + provider_runtime_id.encode("utf-8"), + hashlib.sha256, + ).hexdigest() + return f"n_{digest[:8]}" + + +def project_text(value: str | None) -> str | None: + """Collapse whitespace and drop labels that fail the coach-hint bar.""" + + if not isinstance(value, str): + return None + collapsed = " ".join(value.split()) + if not collapsed or len(collapsed) > MAX_AUTHORING_LABEL_LENGTH: + return None + if "://" in collapsed or "@" in collapsed: + return None + if _SIX_DIGITS.search(collapsed): + return None + if _SSN.search(collapsed) or _PHONE.search(collapsed) or _EMAIL.search(collapsed): + return None + if not _PROJECTED_LABEL.fullmatch(collapsed): + return None + return collapsed + + +def project_process_name(value: str | None) -> str | None: + """Keep a process name only when it matches the closed identifier grammar.""" + + if not isinstance(value, str): + return None + collapsed = " ".join(value.split()) + if not collapsed or not _PROCESS_NAME.fullmatch(collapsed): + return None + return collapsed + + +def raw_nodes_from_observation( + observation: StructuralObservation, +) -> list[AuthoringRawNode]: + """Copy a persisted window tree into projector input.""" + + return [_raw_from_structural(node) for node in observation.tree or []] + + +def raw_window_from_observation( + observation: StructuralObservation, +) -> AuthoringRawWindow: + """Copy persisted window identity, including a title that must not go to MCP.""" + + window = observation.window + process_name = None + if observation.process is not None: + process_name = observation.process.process_name + title = None + bounds = None + if window is not None: + title = window.title + bounds = window.bounds + return AuthoringRawWindow( + process_name=process_name, + role=observation.element.role, + title=title, + bounds=bounds, + ) + + +def project_authoring_observe( + observation: StructuralObservation | None = None, + *, + backend: str, + hmac_key: bytes, + lease_nonce: bytes | str, + provider: str | None = None, + recording: bool = False, + raw_tree: list[AuthoringRawNode] | None = None, + raw_window: AuthoringRawWindow | None = None, + observed_at_ms: int | None = None, +) -> AuthoringProjection: + """Project a raw tree to ``openadapt.authoring.observe/v1``. + + Never returns a raw fallback. RDP and Citrix yield an empty coach-only + tree. Windows native is coach-only but may still carry a projected tree. + """ + + if backend not in _KNOWN_BACKENDS: + raise ValueError(f"unknown authoring backend: {backend!r}") + resolved_provider = provider + if observation is not None: + resolved_provider = resolved_provider or observation.provider + if raw_tree is None: + raw_tree = raw_nodes_from_observation(observation) + if raw_window is None: + raw_window = raw_window_from_observation(observation) + if observed_at_ms is None: + observed_at_ms = int(observation.observed_at * 1000) + if not resolved_provider: + raise ValueError("provider is required") + if resolved_provider not in _KNOWN_PROVIDERS: + raise ValueError(f"unknown authoring provider: {resolved_provider!r}") + nonce = lease_nonce.encode("utf-8") if isinstance(lease_nonce, str) else lease_nonce + if not nonce: + raise ValueError("lease_nonce is required") + if observed_at_ms is None: + observed_at_ms = 0 + + coach_only = backend in _COACH_ONLY_BACKENDS + agent_drive = backend in _AGENT_DRIVE_BACKENDS + window = _project_window(raw_window) + viewport = None if raw_window is None else raw_window.bounds + raw_truncated = bool(observation.tree_truncated) if observation is not None else False + + tree: list[AuthoringWireNode] = [] + table: list[AuthoringNodeTableEntry] = [] + truncated = raw_truncated + if backend not in _EMPTY_TREE_BACKENDS: + tree, table, truncated = _project_tree( + raw_tree or [], + hmac_key=hmac_key, + lease_nonce=nonce, + viewport=viewport, + observed_at_ms=observed_at_ms, + window=window, + backend=backend, + provider=resolved_provider, + agent_drive=agent_drive, + coach_only=coach_only, + recording=recording, + raw_truncated=raw_truncated, + ) + + reason: Literal["empty_projection"] | None = None + if not tree and backend not in _EMPTY_TREE_BACKENDS: + reason = "empty_projection" + + observe = AuthoringObserve( + backend=backend, # type: ignore[arg-type] + provider=resolved_provider, # type: ignore[arg-type] + agent_drive=agent_drive, + coach_only=coach_only, + recording=recording, + window=window, + tree=tree, + truncated=truncated, + node_count=len(tree), + reason=reason, + ) + if truncated: + _logger.info("authoring observe truncated node_count=%s", observe.node_count) + return AuthoringProjection(observe=observe, node_table=tuple(table)) + + +def _raw_from_structural(node: StructuralTreeNode) -> AuthoringRawNode: + return AuthoringRawNode( + provider_runtime_id=node.provider_runtime_id, + automation_id=node.automation_id, + role=node.role, + control_type=node.control_type, + name=node.name, + class_name=node.class_name, + value=node.value, + enabled=node.enabled, + focused=node.focused, + bounds=node.bounds, + children=[_raw_from_structural(child) for child in node.children or []], + ) + + +def _role_lookup_keys(value: str) -> list[str]: + collapsed = " ".join(value.split()) + folded = collapsed.casefold() + dotted = folded.rsplit(".", 1)[-1] + compact = dotted.replace("_", "").replace(" ", "").replace("-", "") + keys = [collapsed, folded, dotted, compact] + if compact.startswith("ax") and len(compact) > 2: + keys.append(compact[2:]) + return list(dict.fromkeys(keys)) + + +def _map_one_role(value: str | None, *, provider: str) -> AuthoringRole | None: + if not isinstance(value, str): + return None + for key in _role_lookup_keys(value): + if key in _AUTHORING_ROLES: + return key # type: ignore[return-value] + compact = key.replace("_", "").replace(" ", "").replace("-", "") + if compact in _COMPACT_AUTHORING_ROLES: + return _COMPACT_AUTHORING_ROLES[compact] # type: ignore[return-value] + if compact == "text": + return _PROVIDER_TEXT_ROLE.get(provider) + aliased = _ROLE_ALIASES.get(compact) + if aliased is not None: + return aliased # type: ignore[return-value] + return None + + +def _project_role( + value: str | None, + *, + provider: str, + control_type: str | None = None, +) -> AuthoringRole | None: + mapped = _map_one_role(value, provider=provider) + if mapped is not None: + return mapped + return _map_one_role(control_type, provider=provider) + + +def _project_window(raw_window: AuthoringRawWindow | None) -> AuthoringWindow | None: + if raw_window is None or raw_window.bounds is None: + return None + process_name = project_process_name(raw_window.process_name) + if process_name is None: + return None + return AuthoringWindow( + process_name=process_name, + role="window", + bounds=AuthoringNormalizedBounds(x=0.0, y=0.0, w=1.0, h=1.0), + ) + + +def _flatten(nodes: list[AuthoringRawNode]) -> list[AuthoringRawNode]: + flat: list[AuthoringRawNode] = [] + stack = list(reversed(nodes)) + while stack: + node = stack.pop() + flat.append(node) + stack.extend(reversed(node.children)) + return flat + + +def _normalize_bounds( + bounds: StructuralBounds | None, + viewport: StructuralBounds | None, +) -> AuthoringNormalizedBounds | None: + if bounds is None or viewport is None: + return None + width = viewport.right - viewport.left + height = viewport.bottom - viewport.top + if width <= 0 or height <= 0: + return None + x = _clamp01((bounds.left - viewport.left) / width) + y = _clamp01((bounds.top - viewport.top) / height) + w = _clamp01((bounds.right - bounds.left) / width) + h = _clamp01((bounds.bottom - bounds.top) / height) + if x + w > 1: + w = 1.0 - x + if y + h > 1: + h = 1.0 - y + try: + return AuthoringNormalizedBounds(x=x, y=y, w=w, h=h) + except (TypeError, ValueError): + return None + + +def _pixel_bounds(bounds: StructuralBounds | None) -> AuthoringPixelBounds | None: + if bounds is None: + return None + try: + return AuthoringPixelBounds( + x=bounds.left, + y=bounds.top, + w=bounds.right - bounds.left, + h=bounds.bottom - bounds.top, + ) + except (TypeError, ValueError): + return None + + +def _clamp01(value: float) -> float: + if value < 0: + return 0.0 + if value > 1: + return 1.0 + return float(value) + + +def _wire_size( + *, + tree: list[AuthoringWireNode], + window: AuthoringWindow | None, + backend: str, + provider: str, + agent_drive: bool, + coach_only: bool, + recording: bool, + truncated: bool, +) -> int: + payload: dict[str, Any] = { + "schema_version": AUTHORING_OBSERVE_SCHEMA_VERSION, + "backend": backend, + "provider": provider, + "mode": "authoring", + "agent_drive": agent_drive, + "coach_only": coach_only, + "recording": recording, + "tree": [node.model_dump(mode="json", exclude_none=True) for node in tree], + "truncated": truncated, + "node_count": len(tree), + } + if window is not None: + payload["window"] = window.model_dump(mode="json", exclude_none=True) + return len(json.dumps(payload, separators=(",", ":"), ensure_ascii=False).encode("utf-8")) + + +def _project_tree( + raw_tree: list[AuthoringRawNode], + *, + hmac_key: bytes, + lease_nonce: bytes, + viewport: StructuralBounds | None, + observed_at_ms: int, + window: AuthoringWindow | None, + backend: str, + provider: str, + agent_drive: bool, + coach_only: bool, + recording: bool, + raw_truncated: bool, +) -> tuple[list[AuthoringWireNode], list[AuthoringNodeTableEntry], bool]: + tree: list[AuthoringWireNode] = [] + table: list[AuthoringNodeTableEntry] = [] + truncated = raw_truncated + for index, raw in enumerate(_flatten(raw_tree)): + role = _project_role( + raw.role, + provider=provider, + control_type=raw.control_type, + ) + bounds = _normalize_bounds(raw.bounds, viewport) + if ( + role is None + or raw.enabled is None + or raw.focused is None + or bounds is None + ): + continue + runtime_id = raw.provider_runtime_id or f"anon:{index}" + node = AuthoringWireNode( + node_id=mint_node_id( + hmac_key=hmac_key, + lease_nonce=lease_nonce, + provider_runtime_id=runtime_id, + ), + role=role, + control_type=project_text(raw.control_type), + automation_id=project_text(raw.automation_id), + name=project_text(raw.name), + class_name=project_text(raw.class_name), + enabled=raw.enabled, + focused=raw.focused, + bounds=bounds, + ) + candidate_tree = [*tree, node] + if len(candidate_tree) > MAX_AUTHORING_NODES: + truncated = True + break + size = _wire_size( + tree=candidate_tree, + window=window, + backend=backend, + provider=provider, + agent_drive=agent_drive, + coach_only=coach_only, + recording=recording, + truncated=True, + ) + if size > MAX_AUTHORING_WIRE_BYTES: + truncated = True + break + tree = candidate_tree + table.append( + AuthoringNodeTableEntry( + node_id=node.node_id, + backend_pixels=_pixel_bounds(raw.bounds), + normalized=node.bounds, + provider_runtime_id=raw.provider_runtime_id, + observed_at=observed_at_ms, + ) + ) + return tree, table, truncated + + +__all__ = [ + "AUTHORING_OBSERVE_SCHEMA_VERSION", + "AuthoringBackend", + "AuthoringNodeTableEntry", + "AuthoringProvider", + "AuthoringRole", + "AuthoringNormalizedBounds", + "AuthoringObserve", + "AuthoringPixelBounds", + "AuthoringProjection", + "AuthoringRawNode", + "AuthoringRawWindow", + "AuthoringWindow", + "AuthoringWireNode", + "MAX_AUTHORING_LABEL_LENGTH", + "MAX_AUTHORING_NODES", + "MAX_AUTHORING_WIRE_BYTES", + "mint_node_id", + "project_authoring_observe", + "project_process_name", + "project_text", + "raw_nodes_from_observation", + "raw_window_from_observation", +] diff --git a/openadapt_capture/structural.py b/openadapt_capture/structural.py index abd360f..0058363 100644 --- a/openadapt_capture/structural.py +++ b/openadapt_capture/structural.py @@ -10,7 +10,7 @@ import logging import math import sys -from dataclasses import dataclass +from dataclasses import dataclass, replace from typing import Annotated, Literal, Protocol, runtime_checkable from pydantic import BaseModel, ConfigDict, Field, model_validator @@ -18,6 +18,8 @@ STRUCTURAL_OBSERVATION_SCHEMA_VERSION = "openadapt.capture.structural-observation/v1" MAX_STRUCTURAL_TEXT_LENGTH = 512 MAX_STRUCTURAL_ANCESTRY_DEPTH = 32 +MAX_STRUCTURAL_TREE_NODES = 1024 +StructuralQueryKind = Literal["point", "focused", "window_tree"] _PROVIDER_PATTERN = r"^[a-z][a-z0-9_.-]*$" _StructuralText = Annotated[str, Field(max_length=MAX_STRUCTURAL_TEXT_LENGTH)] @@ -25,10 +27,36 @@ str, Field(min_length=1, max_length=64, pattern=_PROVIDER_PATTERN), ] +_SECRET_VALUE_ROLE_KEYS = frozenset( + { + "password", + "passwordbox", + "passwordtext", + "password text", + "securetextfield", + "axsecuretextfield", + "secureedit", + } +) _logger = logging.getLogger(__name__) +def omit_tree_value(*roles: str | None, is_password: bool = False) -> bool: + """Return True when a password or secure-field value must not be retained.""" + + if is_password: + return True + for role in roles: + if not isinstance(role, str): + continue + collapsed = " ".join(role.replace("_", " ").casefold().split()) + compact = collapsed.replace(" ", "") + if collapsed in _SECRET_VALUE_ROLE_KEYS or compact in _SECRET_VALUE_ROLE_KEYS: + return True + return False + + class StructuralBounds(BaseModel): """Screen-space bounds reported by the accessibility provider.""" @@ -70,6 +98,33 @@ class StructuralElement(BaseModel): ) +class StructuralTreeNode(BaseModel): + """One node of a raw window accessibility tree retained on disk. + + Non-secret provider values may persist for compile. Password and + secure-field values are omitted. This is not the vendor-wire payload; + :mod:`openadapt_capture.authoring_project` projects a PHI-safe subset for + ``openadapt.authoring.observe/v1``. + """ + + model_config = ConfigDict(extra="forbid") + + provider_runtime_id: _StructuralText | None = None + automation_id: _StructuralText | None = None + role: _StructuralText | None = None + control_type: _StructuralText | None = None + name: _StructuralText | None = None + class_name: _StructuralText | None = None + value: _StructuralText | None = None + enabled: bool | None = None + focused: bool | None = None + bounds: StructuralBounds | None = None + children: list[StructuralTreeNode] | None = Field( + default=None, + max_length=MAX_STRUCTURAL_TREE_NODES, + ) + + class StructuralAncestor(BaseModel): """One parent in the target's accessibility ancestry.""" @@ -130,7 +185,7 @@ class StructuralObservation(BaseModel): ) event_timestamp: float observed_at: float - query_kind: Literal["point", "focused"] + query_kind: StructuralQueryKind element: StructuralElement process: StructuralProcessIdentity | None = None window: StructuralWindowIdentity | None = None @@ -140,6 +195,23 @@ class StructuralObservation(BaseModel): ) candidate_count: int | None = Field(default=None, ge=0) candidate_context: StructuralCandidateContext | None = None + tree: list[StructuralTreeNode] | None = Field( + default=None, + max_length=MAX_STRUCTURAL_TREE_NODES, + description="Raw window accessibility tree. Only valid for window_tree.", + ) + tree_truncated: bool | None = None + + @model_validator(mode="after") + def validate_window_tree_fields(self) -> "StructuralObservation": + """Keep the raw tree off point and focused action evidence.""" + if self.query_kind == "window_tree": + return self + if self.tree is not None: + raise ValueError("tree is only valid for window_tree observations") + if self.tree_truncated: + raise ValueError("tree_truncated is only valid for window_tree observations") + return self @dataclass(frozen=True) @@ -150,6 +222,7 @@ class StructuralObservationRequest: action_name: str x: float | None = None y: float | None = None + query_kind: StructuralQueryKind | None = None @runtime_checkable @@ -224,9 +297,21 @@ def observe_structural_action( return None +def observe_window_tree( + observer: StructuralObserver | None, + request: StructuralObservationRequest, +) -> StructuralObservation | None: + """Observe the top-level window tree for authoring or compile.""" + + if request.query_kind != "window_tree": + request = replace(request, query_kind="window_tree") + return observe_structural_action(observer, request) + + __all__ = [ "MAX_STRUCTURAL_ANCESTRY_DEPTH", "MAX_STRUCTURAL_TEXT_LENGTH", + "MAX_STRUCTURAL_TREE_NODES", "STRUCTURAL_OBSERVATION_SCHEMA_VERSION", "StructuralAncestor", "StructuralBounds", @@ -236,7 +321,11 @@ def observe_structural_action( "StructuralObservationRequest", "StructuralObserver", "StructuralProcessIdentity", + "StructuralQueryKind", + "StructuralTreeNode", "StructuralWindowIdentity", "create_structural_observer", "observe_structural_action", + "observe_window_tree", + "omit_tree_value", ] diff --git a/openadapt_capture/structural_observer/linux.py b/openadapt_capture/structural_observer/linux.py index cabd7f7..51d4958 100644 --- a/openadapt_capture/structural_observer/linux.py +++ b/openadapt_capture/structural_observer/linux.py @@ -16,13 +16,16 @@ from openadapt_capture.structural import ( MAX_STRUCTURAL_ANCESTRY_DEPTH, MAX_STRUCTURAL_TEXT_LENGTH, + MAX_STRUCTURAL_TREE_NODES, StructuralAncestor, StructuralBounds, StructuralElement, StructuralObservation, StructuralObservationRequest, StructuralProcessIdentity, + StructuralTreeNode, StructuralWindowIdentity, + omit_tree_value, ) _logger = logging.getLogger(__name__) @@ -285,6 +288,24 @@ def process_id(self, element: Any) -> int | None: current = parent return None + def runtime_id(self, element: Any) -> str | None: + return _text(_call(element, "get_path", "getPath")) + + def text_value(self, element: Any) -> str | None: + text = _call(element, "get_text_iface", "queryText", "get_text") + if text is None: + return _text(_call(element, "get_description", "getDescription")) + count = _integer(_call(text, "get_character_count", "getCharacterCount")) + if count is None or count <= 0: + return None + try: + return _text(text.get_text(0, min(count, MAX_STRUCTURAL_TEXT_LENGTH))) + except Exception: + try: + return _text(text.getText(0, min(count, MAX_STRUCTURAL_TEXT_LENGTH))) + except Exception: + return None + def action_names(self, element: Any) -> list[str] | None: action = _call(element, "get_action_iface", "queryAction", "get_action") if action is None: @@ -378,6 +399,96 @@ def _window(runtime: Any, element: Any) -> StructuralWindowIdentity | None: return identity if identity.model_dump(exclude_none=True) else None +def _window_root(runtime: Any, element: Any) -> Any: + current = element + candidate = element + for _ in range(MAX_STRUCTURAL_ANCESTRY_DEPTH): + if (runtime.role_name(current) or "").casefold() in { + "alert", + "dialog", + "frame", + "window", + }: + candidate = current + break + parent = runtime.parent(current) + if parent is None: + break + current = parent + return candidate + + +def _bool_state(runtime: Any, element: Any, name: str) -> bool | None: + method = getattr(runtime, "_state_contains", None) + if not callable(method): + return None + try: + return bool(method(element, name)) + except Exception: + return None + + +def _tree_value(runtime: Any, element: Any) -> str | None: + reader = getattr(runtime, "text_value", None) + if not callable(reader): + return None + try: + return _text(reader(element)) + except Exception: + return None + + +def _tree_runtime_id(runtime: Any, element: Any) -> str | None: + reader = getattr(runtime, "runtime_id", None) + if not callable(reader): + return None + try: + return _text(reader(element)) + except Exception: + return None + + +def _as_tree_node( + runtime: Any, + element: Any, + state: dict[str, int | bool], +) -> StructuralTreeNode | None: + remaining = int(state["remaining"]) + if remaining <= 0: + state["truncated"] = True + return None + state["remaining"] = remaining - 1 + fields = _fields(runtime, element) + children: list[StructuralTreeNode] = [] + try: + raw_children = runtime.children(element) + except Exception: + raw_children = [] + for child in raw_children or []: + node = _as_tree_node(runtime, child, state) + if node is not None: + children.append(node) + if state["truncated"]: + break + return StructuralTreeNode( + provider_runtime_id=_tree_runtime_id(runtime, element), + automation_id=fields.get("automation_id"), + role=fields.get("role"), + control_type=fields.get("control_type"), + name=fields.get("name"), + class_name=fields.get("class_name"), + value=( + None + if omit_tree_value(fields.get("role"), fields.get("control_type")) + else _tree_value(runtime, element) + ), + enabled=_bool_state(runtime, element, "ENABLED"), + focused=_bool_state(runtime, element, "FOCUSED"), + bounds=fields.get("bounds"), + children=children or None, + ) + + class LinuxATSpiStructuralObserver: """Read exact AT-SPI evidence for a pointer or focused action.""" @@ -432,18 +543,15 @@ def observe(self, request: StructuralObservationRequest) -> StructuralObservatio runtime = getattr(self._thread_state, "runtime", None) if runtime is None: return None + window_tree = request.query_kind == "window_tree" if request.x is not None and request.y is not None: element = runtime.element_at_point(request.x, request.y) - query_kind = "point" + query_kind = "window_tree" if window_tree else "point" else: element = runtime.focused_element() - query_kind = "focused" + query_kind = "window_tree" if window_tree else "focused" if element is None: return None - fields = _fields(runtime, element) - observed_element = StructuralElement(**fields) - if not observed_element.model_dump(exclude_none=True): - return None pid = _integer(runtime.process_id(element)) process = None if pid is not None and pid > 0: @@ -451,6 +559,32 @@ def observe(self, request: StructuralObservationRequest) -> StructuralObservatio process_id=pid, process_name=self.process_name_resolver(pid), ) + if window_tree: + root = _window_root(runtime, element) + fields = _fields(runtime, root) + observed_element = StructuralElement(**fields) + if not observed_element.model_dump(exclude_none=True): + return None + tree_state: dict[str, int | bool] = { + "remaining": MAX_STRUCTURAL_TREE_NODES, + "truncated": False, + } + tree_root = _as_tree_node(runtime, root, tree_state) + return StructuralObservation( + provider="linux_atspi", + event_timestamp=request.event_timestamp, + observed_at=self.clock(), + query_kind="window_tree", + element=observed_element, + process=process, + window=_window(runtime, element), + tree=[tree_root] if tree_root is not None else None, + tree_truncated=True if tree_state["truncated"] else None, + ) + fields = _fields(runtime, element) + observed_element = StructuralElement(**fields) + if not observed_element.model_dump(exclude_none=True): + return None return StructuralObservation( provider="linux_atspi", event_timestamp=request.event_timestamp, diff --git a/openadapt_capture/structural_observer/macos.py b/openadapt_capture/structural_observer/macos.py index a381d35..5c2334d 100644 --- a/openadapt_capture/structural_observer/macos.py +++ b/openadapt_capture/structural_observer/macos.py @@ -16,13 +16,16 @@ from openadapt_capture.structural import ( MAX_STRUCTURAL_ANCESTRY_DEPTH, MAX_STRUCTURAL_TEXT_LENGTH, + MAX_STRUCTURAL_TREE_NODES, StructuralAncestor, StructuralBounds, StructuralElement, StructuralObservation, StructuralObservationRequest, StructuralProcessIdentity, + StructuralTreeNode, StructuralWindowIdentity, + omit_tree_value, ) _logger = logging.getLogger(__name__) @@ -99,6 +102,12 @@ def actions(self, element: Any) -> list[Any] | None: return None return list(values) + def children(self, element: Any) -> list[Any]: + value = self.attribute(element, "AXChildren") + if isinstance(value, (list, tuple)): + return list(value) + return [] + def process_id(self, element: Any) -> int | None: result = self.ax.AXUIElementGetPid(element, None) if isinstance(result, tuple): @@ -209,6 +218,75 @@ def _window(runtime: Any, element: Any) -> StructuralWindowIdentity | None: return identity if identity.model_dump(exclude_none=True) else None +def _ax_children(runtime: Any, element: Any) -> list[Any]: + reader = getattr(runtime, "children", None) + if callable(reader): + try: + children = reader(element) + except Exception: + return [] + if isinstance(children, (list, tuple)): + return list(children) + return [] + value = runtime.attribute(element, "AXChildren") + if isinstance(value, (list, tuple)): + return list(value) + return [] + + +def _ax_bool(runtime: Any, element: Any, name: str) -> bool | None: + value = runtime.attribute(element, name) + return value if isinstance(value, bool) else None + + +def _ax_value(runtime: Any, element: Any) -> str | None: + value = runtime.attribute(element, "AXValue") + if value is None or isinstance(value, bool): + return None + if isinstance(value, str): + return _text(value) + try: + return _text(str(value)) + except Exception: + return None + + +def _as_tree_node( + runtime: Any, + element: Any, + state: dict[str, int | bool], +) -> StructuralTreeNode | None: + remaining = int(state["remaining"]) + if remaining <= 0: + state["truncated"] = True + return None + state["remaining"] = remaining - 1 + fields = _element_fields(runtime, element) + children: list[StructuralTreeNode] = [] + for child in _ax_children(runtime, element): + node = _as_tree_node(runtime, child, state) + if node is not None: + children.append(node) + if state["truncated"]: + break + value = None + if not omit_tree_value(fields.get("role"), fields.get("control_type")): + value = _ax_value(runtime, element) + return StructuralTreeNode( + provider_runtime_id=fields.get("automation_id"), + automation_id=fields.get("automation_id"), + role=fields.get("role"), + control_type=fields.get("control_type"), + name=fields.get("name"), + class_name=fields.get("class_name"), + value=value, + enabled=_ax_bool(runtime, element, "AXEnabled"), + focused=_ax_bool(runtime, element, "AXFocused"), + bounds=fields.get("bounds"), + children=children or None, + ) + + class MacOSAXStructuralObserver: """Read exact AX evidence for a pointer or focused action.""" @@ -263,18 +341,15 @@ def observe(self, request: StructuralObservationRequest) -> StructuralObservatio runtime = getattr(self._thread_state, "runtime", None) if runtime is None: return None + window_tree = request.query_kind == "window_tree" if request.x is not None and request.y is not None: element = runtime.element_at_point(request.x, request.y) - query_kind = "point" + query_kind = "window_tree" if window_tree else "point" else: element = runtime.focused_element() - query_kind = "focused" + query_kind = "window_tree" if window_tree else "focused" if element is None: return None - fields = _element_fields(runtime, element) - observed_element = StructuralElement(**fields) - if not observed_element.model_dump(exclude_none=True): - return None pid = _integer(runtime.process_id(element)) process = None if pid is not None and pid > 0: @@ -282,6 +357,32 @@ def observe(self, request: StructuralObservationRequest) -> StructuralObservatio process_id=pid, process_name=self.process_name_resolver(pid), ) + if window_tree: + root = runtime.attribute(element, "AXWindow") or element + fields = _element_fields(runtime, root) + observed_element = StructuralElement(**fields) + if not observed_element.model_dump(exclude_none=True): + return None + tree_state: dict[str, int | bool] = { + "remaining": MAX_STRUCTURAL_TREE_NODES, + "truncated": False, + } + tree_root = _as_tree_node(runtime, root, tree_state) + return StructuralObservation( + provider="macos_ax", + event_timestamp=request.event_timestamp, + observed_at=self.clock(), + query_kind="window_tree", + element=observed_element, + process=process, + window=_window(runtime, element), + tree=[tree_root] if tree_root is not None else None, + tree_truncated=True if tree_state["truncated"] else None, + ) + fields = _element_fields(runtime, element) + observed_element = StructuralElement(**fields) + if not observed_element.model_dump(exclude_none=True): + return None return StructuralObservation( provider="macos_ax", event_timestamp=request.event_timestamp, diff --git a/openadapt_capture/structural_observer/windows.py b/openadapt_capture/structural_observer/windows.py index 2464258..90631e9 100644 --- a/openadapt_capture/structural_observer/windows.py +++ b/openadapt_capture/structural_observer/windows.py @@ -19,6 +19,7 @@ from openadapt_capture.structural import ( MAX_STRUCTURAL_ANCESTRY_DEPTH, MAX_STRUCTURAL_TEXT_LENGTH, + MAX_STRUCTURAL_TREE_NODES, StructuralAncestor, StructuralBounds, StructuralCandidateContext, @@ -26,7 +27,9 @@ StructuralObservation, StructuralObservationRequest, StructuralProcessIdentity, + StructuralTreeNode, StructuralWindowIdentity, + omit_tree_value, ) _logger = logging.getLogger(__name__) @@ -379,6 +382,101 @@ def _resolve_process_name(process_id: int) -> str | None: return None +def _runtime_id(info: Any) -> str | None: + if info is None: + return None + rid = _safe_value(info, "runtime_id") + if isinstance(rid, (list, tuple)): + parts: list[str] = [] + for item in rid: + try: + parts.append(str(int(item))) + except (TypeError, ValueError): + return None + return ".".join(parts) if parts else None + return _present_string(rid) + + +def _enabled(wrapper: Any) -> bool | None: + value = _safe_call(wrapper, "is_enabled") + return value if isinstance(value, bool) else None + + +def _focused(wrapper: Any, info: Any) -> bool | None: + value = _safe_value(info, "has_keyboard_focus") if info is not None else None + if isinstance(value, bool): + return value + value = _safe_call(wrapper, "has_keyboard_focus") + return value if isinstance(value, bool) else None + + +def _is_password(wrapper: Any, fields: dict[str, Any] | None = None) -> bool: + info = _element_info(wrapper) + if info is not None: + for attr in ("is_password", "IsPassword"): + if _safe_value(info, attr) is True: + return True + element = _safe_value(info, "element") + if _safe_value(element, "CurrentIsPassword") is True: + return True + resolved = fields if fields is not None else _element_fields(wrapper) + return omit_tree_value( + resolved.get("role"), + resolved.get("control_type"), + resolved.get("class_name"), + ) + + +def _value(wrapper: Any, fields: dict[str, Any] | None = None) -> str | None: + if _is_password(wrapper, fields): + return None + value = _present_string(_safe_call(wrapper, "get_value")) + if value is not None: + return value + iface = _safe_value(wrapper, "iface_value") + return _present_string(_safe_value(iface, "CurrentValue")) + + +def _children(wrapper: Any) -> list[Any]: + children = _safe_call(wrapper, "children") + if children is None: + return [] + try: + return list(children) + except TypeError: + return [] + + +def _as_tree_node(wrapper: Any, state: dict[str, int | bool]) -> StructuralTreeNode | None: + remaining = int(state["remaining"]) + if remaining <= 0: + state["truncated"] = True + return None + state["remaining"] = remaining - 1 + fields = _element_fields(wrapper) + info = _element_info(wrapper) + children: list[StructuralTreeNode] = [] + for child in _children(wrapper): + node = _as_tree_node(child, state) + if node is not None: + children.append(node) + if state["truncated"]: + break + return StructuralTreeNode( + provider_runtime_id=_runtime_id(info), + automation_id=fields.get("automation_id"), + role=fields.get("role"), + control_type=fields.get("control_type"), + name=fields.get("name"), + class_name=fields.get("class_name"), + value=_value(wrapper, fields), + enabled=_enabled(wrapper), + focused=_focused(wrapper, info), + bounds=fields.get("bounds"), + children=children or None, + ) + + def _observe_with_runtime( runtime: Any, request: StructuralObservationRequest, @@ -389,12 +487,13 @@ def _observe_with_runtime( ) -> StructuralObservation | None: """Build one observation while every UIA wrapper stays on its owner thread.""" + window_tree = request.query_kind == "window_tree" if request.x is not None and request.y is not None: target = runtime.from_point(request.x, request.y) - query_kind = "point" + query_kind = "window_tree" if window_tree else "point" else: target = runtime.focused_element() - query_kind = "focused" + query_kind = "window_tree" if window_tree else "focused" if target is None: return None @@ -406,7 +505,6 @@ def _observe_with_runtime( # element that legitimately exposes nothing. Report no observation. return None - element = _as_element(target) process_id = _present_int(_safe_value(info, "process_id")) process = None if process_id is not None: @@ -414,6 +512,27 @@ def _observe_with_runtime( process_id=process_id, process_name=_present_string(process_name_resolver(process_id)), ) + + if window_tree: + root = _safe_call(target, "top_level_parent") or target + tree_state: dict[str, int | bool] = { + "remaining": MAX_STRUCTURAL_TREE_NODES, + "truncated": False, + } + tree_root = _as_tree_node(root, tree_state) + return StructuralObservation( + provider="windows_uia", + event_timestamp=request.event_timestamp, + observed_at=clock(), + query_kind="window_tree", + element=_as_element(root), + process=process, + window=_window_identity(target), + tree=[tree_root] if tree_root is not None else None, + tree_truncated=True if tree_state["truncated"] else None, + ) + + element = _as_element(target) candidate_count, candidate_context = _candidate_cardinality(target) return StructuralObservation( provider="windows_uia", diff --git a/tests/fixtures/authoring-observe-v1.json b/tests/fixtures/authoring-observe-v1.json new file mode 100644 index 0000000..84f4a1d --- /dev/null +++ b/tests/fixtures/authoring-observe-v1.json @@ -0,0 +1,302 @@ +{ + "$defs": { + "AuthoringBackendV1": { + "enum": [ + "macos", + "linux", + "windows", + "web", + "rdp", + "citrix" + ], + "title": "AuthoringBackendV1", + "type": "string" + }, + "AuthoringEmptyProjectionReason": { + "enum": [ + "empty_projection" + ], + "title": "AuthoringEmptyProjectionReason", + "type": "string" + }, + "AuthoringNodeV1": { + "additionalProperties": false, + "properties": { + "automation_id": { + "anyOf": [ + { + "pattern": "^[A-Za-z0-9][A-Za-z0-9 ._-]{0,79}$", + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Automation Id" + }, + "bounds": { + "$ref": "#/$defs/AuthoringNormalizedBoundsV1" + }, + "class_name": { + "anyOf": [ + { + "pattern": "^[A-Za-z0-9][A-Za-z0-9 ._-]{0,79}$", + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Class Name" + }, + "control_type": { + "anyOf": [ + { + "pattern": "^[A-Za-z0-9][A-Za-z0-9 ._-]{0,79}$", + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Control Type" + }, + "enabled": { + "title": "Enabled", + "type": "boolean" + }, + "focused": { + "title": "Focused", + "type": "boolean" + }, + "name": { + "anyOf": [ + { + "pattern": "^[A-Za-z0-9][A-Za-z0-9 ._-]{0,79}$", + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Name" + }, + "node_id": { + "pattern": "^n_[0-9a-f]{8}$", + "title": "Node Id", + "type": "string" + }, + "role": { + "$ref": "#/$defs/ElementRole" + } + }, + "required": [ + "node_id", + "role", + "enabled", + "focused", + "bounds" + ], + "title": "AuthoringNodeV1", + "type": "object" + }, + "AuthoringNormalizedBoundsV1": { + "additionalProperties": false, + "description": "Viewport-normalized overlay coordinates. Not backend pixels.", + "properties": { + "h": { + "maximum": 1, + "minimum": 0, + "title": "H", + "type": "number" + }, + "w": { + "maximum": 1, + "minimum": 0, + "title": "W", + "type": "number" + }, + "x": { + "maximum": 1, + "minimum": 0, + "title": "X", + "type": "number" + }, + "y": { + "maximum": 1, + "minimum": 0, + "title": "Y", + "type": "number" + } + }, + "required": [ + "x", + "y", + "w", + "h" + ], + "title": "AuthoringNormalizedBoundsV1", + "type": "object" + }, + "AuthoringProviderV1": { + "enum": [ + "playwright_ax", + "macos_ax", + "windows_uia", + "linux_atspi", + "none" + ], + "title": "AuthoringProviderV1", + "type": "string" + }, + "AuthoringWindowV1": { + "additionalProperties": false, + "properties": { + "bounds": { + "$ref": "#/$defs/AuthoringNormalizedBoundsV1" + }, + "process_name": { + "pattern": "^[A-Za-z0-9 ._-]{1,64}$", + "title": "Process Name", + "type": "string" + }, + "role": { + "const": "window", + "default": "window", + "title": "Role", + "type": "string" + } + }, + "required": [ + "process_name", + "bounds" + ], + "title": "AuthoringWindowV1", + "type": "object" + }, + "ElementRole": { + "description": "Normalized UI element roles across platforms.\n\nCovers Windows UIA, macOS AX, web ARIA, and OCR-detected elements.", + "enum": [ + "button", + "text_input", + "text_static", + "label", + "link", + "checkbox", + "radio", + "combobox", + "list_item", + "menu", + "menu_item", + "tab", + "tree_item", + "image", + "icon", + "toolbar", + "scrollbar", + "slider", + "window", + "dialog", + "group", + "table", + "table_cell", + "table_row", + "heading", + "paragraph", + "unknown" + ], + "title": "ElementRole", + "type": "string" + } + }, + "additionalProperties": false, + "description": "PHI-safe projected tree. No screenshots, titles, or field values.", + "properties": { + "agent_drive": { + "title": "Agent Drive", + "type": "boolean" + }, + "backend": { + "$ref": "#/$defs/AuthoringBackendV1" + }, + "coach_only": { + "title": "Coach Only", + "type": "boolean" + }, + "mode": { + "const": "authoring", + "default": "authoring", + "title": "Mode", + "type": "string" + }, + "node_count": { + "maximum": 200, + "minimum": 0, + "title": "Node Count", + "type": "integer" + }, + "provider": { + "$ref": "#/$defs/AuthoringProviderV1" + }, + "reason": { + "anyOf": [ + { + "$ref": "#/$defs/AuthoringEmptyProjectionReason" + }, + { + "type": "null" + } + ], + "default": null + }, + "recording": { + "title": "Recording", + "type": "boolean" + }, + "schema_version": { + "const": "openadapt.authoring.observe/v1", + "default": "openadapt.authoring.observe/v1", + "title": "Schema Version", + "type": "string" + }, + "tree": { + "default": [], + "items": { + "$ref": "#/$defs/AuthoringNodeV1" + }, + "maxItems": 200, + "title": "Tree", + "type": "array" + }, + "truncated": { + "title": "Truncated", + "type": "boolean" + }, + "window": { + "anyOf": [ + { + "$ref": "#/$defs/AuthoringWindowV1" + }, + { + "type": "null" + } + ], + "default": null + } + }, + "required": [ + "backend", + "provider", + "agent_drive", + "coach_only", + "recording", + "truncated", + "node_count" + ], + "title": "AuthoringObserveV1", + "type": "object" +} diff --git a/tests/test_authoring_project.py b/tests/test_authoring_project.py new file mode 100644 index 0000000..7048d8c --- /dev/null +++ b/tests/test_authoring_project.py @@ -0,0 +1,544 @@ +"""Fail-closed authoring observe projector and window_tree contracts.""" + +from __future__ import annotations + +import ast +import json +import queue +import re +from pathlib import Path + +import pytest +from pydantic import ValidationError + +from openadapt_capture.authoring_project import ( + AUTHORING_OBSERVE_SCHEMA_VERSION, + MAX_AUTHORING_NODES, + MAX_AUTHORING_WIRE_BYTES, + AuthoringObserve, + AuthoringRawNode, + AuthoringRawWindow, + AuthoringWindow, + AuthoringWireNode, + mint_node_id, + project_authoring_observe, + project_text, +) +from openadapt_capture.input_observer.windows import ( + LLMHF_INJECTED, + MSLLHOOKSTRUCT, + POINT, + WM_LBUTTONDOWN, + _mouse_event, +) +from openadapt_capture.recorder import on_click, on_move, on_scroll +from openadapt_capture.structural import ( + StructuralBounds, + StructuralElement, + StructuralObservation, + StructuralProcessIdentity, + StructuralTreeNode, + StructuralWindowIdentity, +) + +ROOT = Path(__file__).resolve().parents[1] +PACKAGE = ROOT / "openadapt_capture" +# Pinned from OpenAdaptAI/openadapt-types#35 @ 3abf298b. +SCHEMA_PATH = ROOT / "tests" / "fixtures" / "authoring-observe-v1.json" +TEST_HMAC_KEY = bytes.fromhex("11" * 32) +TEST_LEASE_NONCE = b"test-lease-nonce" +FORBIDDEN_WIRE_KEYS = {"value", "title", "screenshot", "text", "url"} + + +def _all_keys(obj: object) -> set[str]: + keys: set[str] = set() + if isinstance(obj, dict): + keys.update(obj) + for value in obj.values(): + keys.update(_all_keys(value)) + elif isinstance(obj, list): + for item in obj: + keys.update(_all_keys(item)) + return keys + + +def _viewport() -> StructuralBounds: + return StructuralBounds(left=0, top=0, right=1000, bottom=1000) + + +def _raw_node(**kwargs) -> AuthoringRawNode: + bounds = kwargs.pop("bounds", StructuralBounds(left=720, top=880, right=860, bottom=930)) + kwargs.setdefault("enabled", True) + kwargs.setdefault("focused", False) + return AuthoringRawNode(bounds=bounds, **kwargs) + + +def _project(raw_tree: list[AuthoringRawNode], *, backend: str = "web", **kwargs): + return project_authoring_observe( + backend=backend, + provider=kwargs.pop("provider", "playwright_ax"), + hmac_key=TEST_HMAC_KEY, + lease_nonce=TEST_LEASE_NONCE, + raw_tree=raw_tree, + raw_window=kwargs.pop( + "raw_window", + AuthoringRawWindow( + process_name="Chromium", + role="window", + title="Patient chart — do not leak", + bounds=_viewport(), + ), + ), + observed_at_ms=1_785_500_000_123, + **kwargs, + ) + + +def test_schema_fixture_forbids_value_title_screenshot_and_extra_keys() -> None: + schema = json.loads(SCHEMA_PATH.read_text(encoding="utf-8")) + assert schema["title"] == "AuthoringObserveV1" + assert schema["additionalProperties"] is False + node = schema["$defs"]["AuthoringNodeV1"] + window = schema["$defs"]["AuthoringWindowV1"] + bounds = schema["$defs"]["AuthoringNormalizedBoundsV1"] + assert node["additionalProperties"] is False + assert window["additionalProperties"] is False + assert bounds["additionalProperties"] is False + for forbidden in ("value", "title", "screenshot"): + assert forbidden not in schema["properties"] + assert forbidden not in window["properties"] + assert forbidden not in node["properties"] + assert set(node["required"]) == {"node_id", "role", "enabled", "focused", "bounds"} + assert schema["$defs"]["AuthoringProviderV1"]["enum"] == [ + "playwright_ax", + "macos_ax", + "windows_uia", + "linux_atspi", + "none", + ] + + +def test_wire_models_reject_value_title_screenshot_and_extra_keys() -> None: + with pytest.raises(ValidationError): + AuthoringWireNode.model_validate( + {"node_id": "n_abcdef01", "role": "button", "value": "secret"} + ) + with pytest.raises(ValidationError): + AuthoringWindow.model_validate({"title": "Patient chart"}) + with pytest.raises(ValidationError): + AuthoringObserve.model_validate( + { + "schema_version": AUTHORING_OBSERVE_SCHEMA_VERSION, + "backend": "web", + "provider": "playwright_ax", + "mode": "authoring", + "agent_drive": True, + "coach_only": False, + "recording": False, + "window": {}, + "tree": [], + "truncated": False, + "node_count": 0, + "screenshot": "iVBOR", + } + ) + + +def test_six_digit_phone_ssn_email_at_and_url_names_are_dropped() -> None: + assert project_text("Invoice 123456") is None + assert project_text("Call 555-123-4567") is None + assert project_text("123-45-6789") is None + assert project_text("user@example.com") is None + assert project_text("ping @operator") is None + assert project_text("https://example.invalid/path") is None + assert project_text(" Save now ") == "Save now" + assert project_text("x" * 81) is None + assert project_text("btnContinue") == "btnContinue" + + projection = _project( + [ + _raw_node( + provider_runtime_id="invoice", + role="button", + name="Invoice 123456", + automation_id="https://example.invalid/id", + ), + _raw_node( + provider_runtime_id="ssn", + role="text_input", + name="123-45-6789", + automation_id="member@clinic.invalid", + ), + _raw_node( + provider_runtime_id="ok", + role="button", + name="Continue", + automation_id="btnContinue", + ), + ] + ) + wire = projection.wire_dict() + names = [node.get("name") for node in wire["tree"]] + automation_ids = [node.get("automation_id") for node in wire["tree"]] + assert "Invoice 123456" not in names + assert "123-45-6789" not in names + assert "Continue" in names + assert "btnContinue" in automation_ids + assert all(item is None or "://" not in item for item in automation_ids) + assert all(item is None or "@" not in item for item in automation_ids) + + +def test_value_title_and_screenshot_never_appear_on_the_wire() -> None: + observation = StructuralObservation( + provider="macos_ax", + event_timestamp=101.0, + observed_at=101.25, + query_kind="window_tree", + element=StructuralElement(role="AXWindow", name="Chart"), + process=StructuralProcessIdentity(process_id=7, process_name="Chromium"), + window=StructuralWindowIdentity( + title="Patient SSN 123-45-6789", + bounds=_viewport(), + ), + tree=[ + StructuralTreeNode( + provider_runtime_id="ax-1", + role="text_input", + name="Note", + value="typed secret", + enabled=True, + focused=False, + bounds=StructuralBounds(left=10, top=10, right=110, bottom=40), + ) + ], + ) + raw = observation.model_dump(mode="json", exclude_none=True) + assert raw["window"]["title"] == "Patient SSN 123-45-6789" + assert raw["tree"][0]["value"] == "typed secret" + + projection = project_authoring_observe( + observation, + backend="macos", + hmac_key=TEST_HMAC_KEY, + lease_nonce=TEST_LEASE_NONCE, + ) + wire = projection.wire_dict() + keys = _all_keys(wire) + assert FORBIDDEN_WIRE_KEYS.isdisjoint(keys) + assert "title" not in wire["window"] + assert all("value" not in node for node in wire["tree"]) + dumped = json.dumps(wire) + assert "typed secret" not in dumped + assert "Patient SSN" not in dumped + + +def test_rdp_and_citrix_return_empty_coach_only_trees() -> None: + secret = [ + _raw_node( + provider_runtime_id="remote", + role="button", + name="Continue", + value="should not leak", + ) + ] + for backend in ("rdp", "citrix"): + projection = _project(secret, backend=backend, provider="windows_uia") + wire = projection.wire_dict() + assert wire["backend"] == backend + assert wire["coach_only"] is True + assert wire["agent_drive"] is False + assert wire["tree"] == [] + assert wire["node_count"] == 0 + assert projection.node_table == () + assert "value" not in _all_keys(wire) + + +def test_windows_native_is_coach_only_but_may_keep_a_projected_tree() -> None: + projection = _project( + [_raw_node(provider_runtime_id="ok", role="button", automation_id="btnContinue")], + backend="windows", + provider="windows_uia", + ) + wire = projection.wire_dict() + assert wire["coach_only"] is True + assert wire["agent_drive"] is False + assert wire["tree"] + assert wire["tree"][0]["automation_id"] == "btnContinue" + + +def test_node_id_is_hmac_prefix_and_missing_runtime_id_mints_anon() -> None: + expected = mint_node_id( + hmac_key=TEST_HMAC_KEY, + lease_nonce=TEST_LEASE_NONCE, + provider_runtime_id="ax-elem-1", + ) + assert re.fullmatch(r"n_[0-9a-f]{8}", expected) + projection = _project( + [ + _raw_node(provider_runtime_id="ax-elem-1", role="button"), + _raw_node(role="button", name="Save"), + ] + ) + assert projection.observe.tree[0].node_id == expected + anon = mint_node_id( + hmac_key=TEST_HMAC_KEY, + lease_nonce=TEST_LEASE_NONCE, + provider_runtime_id="anon:1", + ) + assert projection.observe.tree[1].node_id == anon + assert projection.node_table[0].provider_runtime_id == "ax-elem-1" + assert projection.node_table[1].provider_runtime_id is None + + +def test_caps_at_200_nodes_and_32kib(monkeypatch: pytest.MonkeyPatch) -> None: + many = [ + _raw_node(provider_runtime_id=f"n{i}", role="button", name=f"Action{i}") + for i in range(MAX_AUTHORING_NODES + 25) + ] + projection = _project(many) + assert projection.observe.truncated is True + assert projection.observe.node_count == MAX_AUTHORING_NODES + assert len(projection.observe.tree) == MAX_AUTHORING_NODES + + monkeypatch.setattr( + "openadapt_capture.authoring_project.MAX_AUTHORING_WIRE_BYTES", + 900, + ) + bulky = [ + _raw_node( + provider_runtime_id=f"b{i}", + role="button", + name="ContinueActionLabel", + automation_id="btnContinueAction", + class_name="ChromeButtonClass", + ) + for i in range(40) + ] + bulky_projection = _project(bulky) + wire = bulky_projection.wire_dict() + assert bulky_projection.observe.truncated is True + encoded = json.dumps(wire, separators=(",", ":"), ensure_ascii=False).encode("utf-8") + assert len(encoded) <= 900 + + +def test_empty_projection_never_falls_back_to_raw() -> None: + projection = _project( + [ + _raw_node( + provider_runtime_id="secret", + name="123-45-6789", + value="typed secret", + title="Patient", + bounds=None, + ) + ] + ) + wire = projection.wire_dict() + assert wire["tree"] == [] + assert wire["reason"] == "empty_projection" + assert "typed secret" not in json.dumps(wire) + assert FORBIDDEN_WIRE_KEYS.isdisjoint(_all_keys(wire)) + + +def test_invalid_process_name_is_dropped() -> None: + projection = _project( + [_raw_node(provider_runtime_id="ok", role="button")], + raw_window=AuthoringRawWindow( + process_name="C:\\secret.exe", + role="window", + title="do not leak", + bounds=_viewport(), + ), + ) + assert projection.observe.window is None + assert "window" not in projection.wire_dict() + + +def test_normalized_bounds_use_the_top_level_viewport() -> None: + projection = _project( + [ + _raw_node( + provider_runtime_id="ok", + role="button", + bounds=StructuralBounds(left=720, top=880, right=860, bottom=930), + ) + ] + ) + bounds = projection.observe.tree[0].bounds + assert bounds is not None + assert bounds.x == pytest.approx(0.72) + assert bounds.y == pytest.approx(0.88) + assert bounds.w == pytest.approx(0.14) + assert bounds.h == pytest.approx(0.05) + pixels = projection.node_table[0].backend_pixels + assert pixels is not None + assert (pixels.x, pixels.y, pixels.w, pixels.h) == (720, 880, 140, 50) + + +def test_package_has_no_record_injected_api() -> None: + for path in PACKAGE.rglob("*.py"): + tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path)) + for node in ast.walk(tree): + if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): + assert node.name != "record_injected" + if isinstance(node, ast.arg): + assert node.arg != "record_injected" + + +def test_injected_clicks_still_do_not_persist() -> None: + events: queue.Queue = queue.Queue() + on_click( + events, + None, + 25, + 35, + "left", + True, + injected=True, + timestamp=101.0, + ) + on_move(events, None, 26, 36, injected=True, timestamp=101.1) + on_scroll( + events, + None, + 26, + 36, + 0, + 1, + injected=True, + timestamp=101.2, + ) + assert events.empty() + + +def test_windows_llmhf_injected_still_returns_none() -> None: + payload = MSLLHOOKSTRUCT(pt=POINT(12, 34), flags=LLMHF_INJECTED) + assert ( + _mouse_event( + WM_LBUTTONDOWN, + payload, + capture_mouse_moves=True, + ) + is None + ) + + +def test_playwright_shaped_tree_projects_without_native_capture() -> None: + projection = _project( + [ + AuthoringRawNode( + provider_runtime_id="btnContinue", + role="button", + control_type="button", + automation_id="btnContinue", + enabled=True, + focused=False, + bounds=StructuralBounds(left=720, top=880, right=860, bottom=930), + ) + ] + ) + wire = projection.wire_dict() + assert wire["schema_version"] == AUTHORING_OBSERVE_SCHEMA_VERSION + assert wire["backend"] == "web" + assert wire["provider"] == "playwright_ax" + assert wire["agent_drive"] is True + assert wire["coach_only"] is False + assert wire["tree"][0]["automation_id"] == "btnContinue" + assert wire["window"]["process_name"] == "Chromium" + assert MAX_AUTHORING_WIRE_BYTES == 32 * 1024 + + +def test_native_roles_map_onto_element_role() -> None: + macos = _project( + [_raw_node(provider_runtime_id="ax", role="AXButton", name="Save")], + backend="macos", + provider="macos_ax", + ) + assert macos.observe.tree[0].role == "button" + assert macos.observe.tree[0].name == "Save" + + windows = _project( + [ + _raw_node( + provider_runtime_id="edit", + role="Edit", + control_type="Edit", + name="Note", + ) + ], + backend="windows", + provider="windows_uia", + ) + assert windows.observe.tree[0].role == "text_input" + + linux = _project( + [_raw_node(provider_runtime_id="btn", role="push button", name="Save")], + backend="linux", + provider="linux_atspi", + ) + assert linux.observe.tree[0].role == "button" + + web = _project( + [_raw_node(provider_runtime_id="box", role="textbox", name="Note")], + backend="web", + provider="playwright_ax", + ) + assert web.observe.tree[0].role == "text_input" + + +def test_unmapped_native_role_fail_closes_to_empty_projection() -> None: + projection = _project( + [_raw_node(provider_runtime_id="ax", role="AXPrivateMysteryRole", name="Save")] + ) + wire = projection.wire_dict() + assert wire["tree"] == [] + assert wire["reason"] == "empty_projection" + assert FORBIDDEN_WIRE_KEYS.isdisjoint(_all_keys(wire)) + + +def test_credentials_never_appear_as_values_on_the_wire() -> None: + projection = _project( + [ + _raw_node( + provider_runtime_id="password", + role="AXSecureTextField", + name="Password", + value="typed-secret", + ) + ], + backend="macos", + provider="macos_ax", + ) + wire = projection.wire_dict() + assert wire["tree"][0]["role"] == "text_input" + assert "value" not in wire["tree"][0] + dumped = json.dumps(wire) + assert "typed-secret" not in dumped + assert FORBIDDEN_WIRE_KEYS.isdisjoint(_all_keys(wire)) + + +def test_projected_wire_keys_match_pinned_types_schema() -> None: + schema = json.loads(SCHEMA_PATH.read_text(encoding="utf-8")) + projection = _project( + [ + _raw_node( + provider_runtime_id="btnContinue", + role="button", + control_type="button", + automation_id="btnContinue", + ) + ] + ) + wire = projection.wire_dict() + assert set(wire) <= set(schema["properties"]) + node_schema = schema["$defs"]["AuthoringNodeV1"] + window_schema = schema["$defs"]["AuthoringWindowV1"] + assert set(wire["window"]) <= set(window_schema["properties"]) + for required in window_schema["required"]: + assert required in wire["window"] + for node in wire["tree"]: + assert set(node) <= set(node_schema["properties"]) + for required in node_schema["required"]: + assert required in node + AuthoringObserve.model_validate(wire) + diff --git a/tests/test_structural_observation.py b/tests/test_structural_observation.py index 18ce195..74de087 100644 --- a/tests/test_structural_observation.py +++ b/tests/test_structural_observation.py @@ -22,8 +22,11 @@ StructuralElement, StructuralObservation, StructuralObservationRequest, + StructuralTreeNode, create_structural_observer, observe_structural_action, + observe_window_tree, + omit_tree_value, ) from openadapt_capture.structural_observer.linux import ( LinuxATSpiStructuralObserver, @@ -54,24 +57,37 @@ def __init__( parent: "_Wrapper | None" = None, top: "_Wrapper | None" = None, descendants: list["_Wrapper"] | None = None, + children: list["_Wrapper"] | None = None, title: str | None = None, patterns: tuple[str, ...] = (), + runtime_id: object | None = None, + enabled: bool | None = True, + focused: bool = False, + value: str | None = None, + is_password: bool = False, + class_name: str | None = None, ) -> None: self.element_info = SimpleNamespace( automation_id=automation_id, control_type=control_type, name=name, - class_name=None, + class_name=class_name, framework_id=None, handle=None, process_id=process_id, rectangle=SimpleNamespace(left=10, top=20, right=110, bottom=60), + runtime_id=runtime_id, + has_keyboard_focus=focused, + is_password=is_password, ) self._role = role self._parent = parent self._top = top self._descendants = descendants or [] + self._children = children or [] self._title = title + self._enabled = enabled + self._value = value self.descendant_queries: list[dict[str, object]] = [] for pattern in patterns: setattr(self, f"iface_{pattern}", object()) @@ -89,6 +105,15 @@ def descendants(self, **kwargs) -> list["_Wrapper"]: self.descendant_queries.append(kwargs) return self._descendants + def children(self) -> list["_Wrapper"]: + return self._children + + def is_enabled(self) -> bool | None: + return self._enabled + + def get_value(self) -> str | None: + return self._value + def window_text(self) -> str | None: return self._title @@ -290,6 +315,10 @@ def __init__(self) -> None: (self.window, "AXRole"): "AXWindow", (self.window, "AXTitle"): "Orders", (self.window, "AXWindowNumber"): 44, + (self.target, "AXEnabled"): True, + (self.target, "AXValue"): "on", + (self.parent, "AXEnabled"): True, + (self.window, "AXEnabled"): True, } def attribute(self, element, name): @@ -309,6 +338,13 @@ def process_id(self, element): def actions(self, element): return ["AXPress"] if element is self.target else None + def children(self, element): + if element is self.window: + return [self.parent] + if element is self.parent: + return [self.target] + return [] + def bounds(self, element): if element is self.target: from openadapt_capture.structural import StructuralBounds @@ -422,9 +458,27 @@ def action_names(self, element): return ["click"] if element is self.target else None def process_id(self, element): - assert element is self.target return 42 + def children(self, element): + if element is self.window: + return [self.parent_element] + if element is self.parent_element: + return [self.target] + return [] + + def text_value(self, element): + if element is self.target: + return "Submit" + return None + + def runtime_id(self, element): + if element is self.target: + return "atspi:submit" + if element is self.window: + return "atspi:window" + return None + def test_linux_atspi_observer_returns_action_time_evidence() -> None: observer = LinuxATSpiStructuralObserver( @@ -930,3 +984,233 @@ def test_live_native_structural_provider_returns_exact_focused_evidence() -> Non assert observation.process is not None assert observation.process.process_id is not None assert observation.window is not None + + +def test_windows_uia_window_tree_omits_password_values() -> None: + field = _Wrapper( + automation_id="member-id", + control_type="Edit", + name="Member", + role="Edit", + runtime_id=(1, 4), + value="queued", + ) + secret = _Wrapper( + automation_id="password", + control_type="Edit", + name="Password", + role="Edit", + runtime_id=(1, 5), + value="typed-secret", + is_password=True, + ) + button = _Wrapper( + automation_id="btnContinue", + control_type="Button", + name="Continue", + role="Button", + runtime_id=(1, 3), + ) + window = _Wrapper( + automation_id="main-window", + control_type="Window", + name="Orders", + role="Window", + title="Orders - Example", + process_id=42, + runtime_id=(1, 1), + children=[field, secret, button], + ) + field._top = window + secret._top = window + button._top = window + window._top = window + observer = WindowsUIAStructuralObserver( + runtime=SimpleNamespace( + from_point=lambda _x, _y: button, + focused_element=lambda: button, + ), + process_name_resolver=lambda process_id: "example.exe" if process_id == 42 else None, + clock=lambda: 101.25, + ) + + observed = observe_window_tree( + observer, + StructuralObservationRequest( + event_timestamp=101.0, + action_name="observe", + x=25, + y=35, + ), + ) + + assert observed is not None + assert observed.query_kind == "window_tree" + assert observed.window is not None + assert observed.window.title == "Orders - Example" + assert observed.tree is not None + root = observed.tree[0] + children = {child.automation_id: child for child in root.children or []} + assert children["member-id"].value == "queued" + assert children["member-id"].name == "Member" + assert children["password"].value is None + assert children["password"].name == "Password" + assert children["btnContinue"].provider_runtime_id == "1.3" + dumped = root.model_dump_json() + assert "typed-secret" not in dumped + + +def test_macos_ax_window_tree_walks_from_the_ax_window() -> None: + observer = MacOSAXStructuralObserver( + runtime=_AXFakeRuntime(), + process_name_resolver=lambda pid: "Example" if pid == 42 else None, + clock=lambda: 101.25, + ) + observed = observe_window_tree( + observer, + StructuralObservationRequest( + event_timestamp=101.0, + action_name="observe", + x=25, + y=35, + ), + ) + assert observed is not None + assert observed.query_kind == "window_tree" + assert observed.element.role == "AXWindow" + assert observed.tree is not None + assert observed.tree[0].role == "AXWindow" + assert observed.tree[0].children is not None + group = observed.tree[0].children[0] + target = group.children[0] + assert target.automation_id == "submit-order" + assert target.provider_runtime_id == "submit-order" + assert target.value == "on" + + +def test_linux_atspi_window_tree_walks_from_the_frame() -> None: + observer = LinuxATSpiStructuralObserver( + runtime=_ATSpiFakeRuntime(), + process_name_resolver=lambda pid: "example" if pid == 42 else None, + clock=lambda: 101.5, + ) + observed = observe_window_tree( + observer, + StructuralObservationRequest(event_timestamp=101.0, action_name="observe"), + ) + assert observed is not None + assert observed.query_kind == "window_tree" + assert observed.tree is not None + assert observed.tree[0].role == "frame" + assert observed.tree[0].provider_runtime_id == "atspi:window" + target = observed.tree[0].children[0].children[0] + assert target.automation_id == "submit-order" + assert target.value == "Submit" + + +def test_window_tree_is_rejected_on_point_observations() -> None: + with pytest.raises(ValueError, match="window_tree"): + StructuralObservation( + provider="windows_uia", + event_timestamp=101.0, + observed_at=101.1, + query_kind="point", + element=StructuralElement(role="Button"), + tree=[StructuralTreeNode(role="Button")], + ) + + +def test_omit_tree_value_detects_password_and_secure_roles() -> None: + assert omit_tree_value("AXSecureTextField") + assert omit_tree_value("password text") + assert omit_tree_value("PasswordBox") + assert omit_tree_value("Edit", is_password=True) + assert not omit_tree_value("AXButton") + assert not omit_tree_value("Edit") + + +def test_macos_ax_secure_text_field_omits_value() -> None: + runtime = _AXFakeRuntime() + secure = object() + runtime.attributes[runtime.window, "AXChildren"] = [runtime.parent, secure] + runtime.attributes[secure, "AXRole"] = "AXSecureTextField" + runtime.attributes[secure, "AXIdentifier"] = "password" + runtime.attributes[secure, "AXValue"] = "typed-secret" + runtime.attributes[secure, "AXEnabled"] = True + + def children(element): + if element is runtime.window: + return [runtime.parent, secure] + if element is runtime.parent: + return [runtime.target] + return [] + + runtime.children = children + observer = MacOSAXStructuralObserver( + runtime=runtime, + process_name_resolver=lambda pid: "Example" if pid == 42 else None, + clock=lambda: 101.25, + ) + observed = observe_window_tree( + observer, + StructuralObservationRequest( + event_timestamp=101.0, + action_name="observe", + x=25, + y=35, + ), + ) + assert observed is not None + assert observed.tree is not None + secure_node = observed.tree[0].children[1] + assert secure_node.role == "AXSecureTextField" + assert secure_node.value is None + assert "typed-secret" not in observed.model_dump_json() + + +def test_linux_atspi_password_text_omits_value() -> None: + runtime = _ATSpiFakeRuntime() + secret = _ATSpiElement("Password", runtime.window) + + def role_name(element): + if element is secret: + return "password text" + return { + runtime.target: "push button", + runtime.parent_element: "panel", + runtime.window: "frame", + runtime.application: "application", + }[element] + + def children(element): + if element is runtime.window: + return [runtime.parent_element, secret] + if element is runtime.parent_element: + return [runtime.target] + return [] + + def text_value(element): + if element is secret: + return "typed-secret" + if element is runtime.target: + return "Submit" + return None + + runtime.role_name = role_name + runtime.children = children + runtime.text_value = text_value + observer = LinuxATSpiStructuralObserver( + runtime=runtime, + process_name_resolver=lambda pid: "example" if pid == 42 else None, + clock=lambda: 101.5, + ) + observed = observe_window_tree( + observer, + StructuralObservationRequest(event_timestamp=101.0, action_name="observe"), + ) + assert observed is not None + assert observed.tree is not None + secret_node = observed.tree[0].children[1] + assert secret_node.role == "password text" + assert secret_node.value is None + assert "typed-secret" not in observed.model_dump_json()