diff --git a/benchmark/rdp_ladder/render_presentation.py b/benchmark/rdp_ladder/render_presentation.py index 7aff4232..c08df608 100644 --- a/benchmark/rdp_ladder/render_presentation.py +++ b/benchmark/rdp_ladder/render_presentation.py @@ -4,7 +4,8 @@ The renderer consumes: * exact RDP frames and input events from the isolated presentation capture; -* the exact ``ProgramGraphSpec`` emitted by ``build_program_graph``; +* the exact ``ProgramGraphSpec`` emitted by ``build_program_graph`` and shown + beside the video; * the exact run parameters used by the first governed replay; * the independent SQL rows retained in the run summary. @@ -36,7 +37,6 @@ ) BG = "#071113" PANEL = "#0d1c1f" -PANEL_2 = "#13272a" TEXT = "#f2f8f7" MUTED = "#a8bcba" TEAL = "#45d6c3" @@ -466,7 +466,7 @@ def _demonstration_frames( ( _overlay( last_frame, - phase="1 · Demonstrate", + phase="Demonstrate", detail="Human-paced input over RDP", cursor=cursor, ), @@ -493,7 +493,7 @@ def _demonstration_frames( ( _overlay( last_frame, - phase="1 · Demonstrate", + phase="Demonstrate", detail="Mouse and keyboard are retained", cursor=point, ), @@ -506,7 +506,7 @@ def _demonstration_frames( ( _overlay( last_frame, - phase="1 · Demonstrate", + phase="Demonstrate", detail="Action recorded", cursor=cursor, click=True, @@ -521,83 +521,6 @@ def _demonstration_frames( return images -def _graph_view(spec: dict, *, active_index: int | None = None) -> Image.Image: - image = _base() - draw = ImageDraw.Draw(image) - _brand(draw, section="Real compiled artifact") - bundle = spec["bundle"] - draw.text( - (44, 104), - bundle["name"], - font=_font(34, bold=True), - fill=TEXT, - ) - params = [f"${item['name']}" for item in bundle.get("params", [])] - draw.text( - (45, 150), - "Parameters: " + (" · ".join(params) if params else "none"), - font=_font(16), - fill=MUTED, - ) - - nodes = spec["nodes"] - positions: list[tuple[int, int, int, int]] = [] - for index, node in enumerate(nodes): - col, row = index % 3, index // 3 - x = 38 + col * 414 - y = 194 + row * 176 - box = (x, y, x + 378, y + 138) - positions.append(box) - highlighted = active_index is None or index <= active_index - fill = PANEL_2 if highlighted else PANEL - outline = TEAL if index == active_index else "#284248" - draw.rounded_rectangle(box, 18, fill=fill, outline=outline, width=3) - draw.text( - (x + 18, y + 14), - f"{index + 1:02d}", - font=_font(15, bold=True), - fill=TEAL if highlighted else MUTED, - ) - title_lines = _wrap(node["title"], 35)[:2] - title_y = y + 12 - for line in title_lines: - draw.text((x + 58, title_y), line, font=_font(16, bold=True), fill=TEXT) - title_y += 23 - details: list[str] = [] - resolution = node.get("resolution") - if resolution and resolution.get("top_rung"): - details.append(f"resolve: {resolution['top_rung']}") - if node.get("param"): - details.append(f"input: ${node['param']}") - if node.get("badges"): - details.extend(node["badges"][:2]) - detail = " · ".join(details) or node.get("kind", "") - for line in _wrap(detail, 47)[:2]: - draw.text((x + 18, title_y + 8), line, font=_font(13), fill=MUTED) - title_y += 19 - - for index in range(len(positions) - 1): - left = positions[index] - right = positions[index + 1] - if index % 3 != 2: - start = (left[2] + 4, (left[1] + left[3]) // 2) - end = (right[0] - 6, (right[1] + right[3]) // 2) - else: - start = ((left[0] + left[2]) // 2, left[3] + 3) - end = ((right[0] + right[2]) // 2, right[1] - 5) - draw.line((start, end), fill="#527277", width=3) - - provenance = bundle.get("provenance") or {} - digest = provenance.get("content_digest") or "not available" - draw.text( - (45, 742), - f"Graph spec v{spec['spec_version']} · bundle {str(digest)[:20]}", - font=_font(14), - fill=MUTED, - ) - return image - - def _selected_frames( frames: list[tuple[dict, Image.Image]], limit: int = 18, @@ -872,26 +795,13 @@ def render(presentation_dir: Path, output: Path) -> dict: ) presented_counts["01-demonstration"] = len(demo_images) - nodes = graph["nodes"] - for index in range(len(nodes)): - count = _write_repeated( - process, _graph_view(graph, active_index=index), 1.1 - ) - timeline.append( - count, - phase="compiled_workflow", - compiled_graph={"node_id": nodes[index]["id"], "node_index": index}, - ) - count = _write_repeated(process, _graph_view(graph), 3.6) - timeline.append(count, phase="compiled_workflow") - replay_images = _selected_frames(phase_frames["02-verified-replay"]) for source_event, image in replay_images: count = _write_repeated( process, _overlay( image, - phase="3 · Execute", + phase="Execute", detail="The correct record is checked before input", ), 0.38, diff --git a/docs/REMOTE_FRAME_CONTRACT.md b/docs/REMOTE_FRAME_CONTRACT.md new file mode 100644 index 00000000..1f1c55e1 --- /dev/null +++ b/docs/REMOTE_FRAME_CONTRACT.md @@ -0,0 +1,11 @@ +# Remote frame contract + +`remote_frame_contract` is a versioned deployment field for remote RDP and +remote-display settling. It binds exact frame dimensions and reviewed volatile +rectangles. The runtime rejects a geometry change or an overlap with a declared +protected region. It retains raw frame bytes and raw lease hashes. It masks only +transient derived inputs used for pointer-settle and final pre-input content +comparisons. + +The current deployment schema can declare protected regions. Desktop editing +of these reviewed regions is a separate follow-up; no runtime learning occurs. diff --git a/openadapt_flow/__init__.py b/openadapt_flow/__init__.py index 1cfee0a6..54938cf1 100644 --- a/openadapt_flow/__init__.py +++ b/openadapt_flow/__init__.py @@ -1,6 +1,6 @@ """openadapt-flow: record once, compile, replay deterministically, heal on drift.""" -__version__ = "1.27.0" +__version__ = "1.27.1" from openadapt_flow.ir import ( # noqa: F401 ActionKind, diff --git a/openadapt_flow/backend.py b/openadapt_flow/backend.py index e38fff05..aa22e74c 100644 --- a/openadapt_flow/backend.py +++ b/openadapt_flow/backend.py @@ -528,7 +528,9 @@ class RemoteActuationBackend(Protocol): The backend owns a one-shot content lease for the returned frame. Its next input method captures once more under the backend input lock and refuses before the first input edge if the window/session, dimensions, readiness, - or exact frame content changed. The lease is consumed once so a + or exact frame content changed. A sealed remote frame contract can exclude + reviewed volatile regions from a derived comparison. Raw frame evidence + and the exact lease stay unmodified. The lease is consumed once so a multi-character type or double-click gesture cannot invalidate itself. """ @@ -537,6 +539,15 @@ def acquire_actuation_frame(self) -> bytes: ... +@runtime_checkable +class RemoteFrameContractBackend(Protocol): + """Optional pre-input protected-region binding for remote comparison masks.""" + + def arm_remote_frame_contract( + self, *, protected_regions: tuple[tuple[int, int, int, int], ...] + ) -> None: ... + + @runtime_checkable class FreshActuationReacquisitionBackend(Protocol): """Reset one proved zero-input invalidation for bounded reacquisition. diff --git a/openadapt_flow/backends/factory.py b/openadapt_flow/backends/factory.py index 991b38f9..4fe0a322 100644 --- a/openadapt_flow/backends/factory.py +++ b/openadapt_flow/backends/factory.py @@ -196,6 +196,7 @@ def _build_rdp_backend( application_version_marker=cfg.rdp_application_version_marker, environment_marker=cfg.rdp_environment_marker, session_marker=cfg.rdp_session_marker, + remote_frame_contract=cfg.remote_frame_contract, ) if rdp_transport is not None or has_host: @@ -220,6 +221,7 @@ def _build_rdp_backend( application_version_marker=cfg.rdp_application_version_marker, environment_marker=cfg.rdp_environment_marker, session_marker=cfg.rdp_session_marker, + remote_frame_contract=cfg.remote_frame_contract, ) if window_client is not None or has_window: @@ -236,6 +238,7 @@ def _build_rdp_backend( kwargs["application_version_marker"] = cfg.rdp_application_version_marker kwargs["environment_marker"] = cfg.rdp_environment_marker kwargs["session_marker"] = cfg.rdp_session_marker + kwargs["remote_frame_contract"] = cfg.remote_frame_contract return RemoteDisplayBackend(window_client, **kwargs) raise ValueError( diff --git a/openadapt_flow/backends/rdp_backend.py b/openadapt_flow/backends/rdp_backend.py index 05b6e7fb..6b725c39 100644 --- a/openadapt_flow/backends/rdp_backend.py +++ b/openadapt_flow/backends/rdp_backend.py @@ -69,6 +69,7 @@ StructuralResolutionRefused, ) from openadapt_flow.ir import ActionDeliveryReceipt, Point +from openadapt_flow.remote_frame_contract import RemoteFrameContract from openadapt_flow.runtime.resolver import visual_resolution_point_fingerprint # What a transport may hand back as the current frame: a PIL image, or raw @@ -307,8 +308,10 @@ def __init__( session_marker: Optional[str] = None, session_marker_probe: Optional[Callable[[bytes], bool]] = None, session_identity_observer: Optional[Callable[[], Optional[str]]] = None, + remote_frame_contract: Optional["RemoteFrameContract"] = None, ) -> None: self._transport = transport + self._remote_frame_contract = remote_frame_contract self._viewport = viewport self._max_frame_age_s = float(max_frame_age_s) if self._max_frame_age_s <= 0: @@ -357,6 +360,7 @@ def __init__( self._session_identity_observer = session_identity_observer self._last_frame_monotonic: Optional[float] = None self._last_frame_digest: Optional[bytes] = None + self._last_comparison_digest: Optional[bytes] = None self._last_session_identity: Optional[str] = None self._qualification_environment: Optional[tuple[str, str, str, str]] = None self._qualification_input_guard: Optional[Callable[[], None]] = None @@ -398,8 +402,15 @@ def screenshot(self) -> bytes: # and screenshot can never disagree. self._viewport = img.size png = self._png_bytes(img) + if self._remote_frame_contract is not None: + self._remote_frame_contract.require_geometry(img.size) self._last_frame_monotonic = time.monotonic() self._last_frame_digest = self._canonical_frame_digest(img) + self._last_comparison_digest = ( + self._remote_frame_contract.comparison_digest(png) + if self._remote_frame_contract is not None + else self._last_frame_digest + ) self._last_session_identity = self._session_identity_from_frame(png) if self._actuation_lease_state == _LEASE_ARMED: self._invalidate_actuation_lease() @@ -444,6 +455,12 @@ def acquire_actuation_frame(self) -> bytes: self._actuation_lease_state = _LEASE_ARMED return png + def arm_remote_frame_contract( + self, *, protected_regions: tuple[tuple[int, int, int, int], ...] + ) -> None: + if self._remote_frame_contract is not None: + self._remote_frame_contract.arm(protected_regions) + def reset_fresh_actuation_state(self) -> None: """Reset only a typed zero-input content invalidation. @@ -1102,8 +1119,16 @@ def _ensure_input_ready( "target resolution; refusing input" ) if self._actuation_lease_state == _LEASE_ARMED: - digest = self._canonical_frame_digest(current_img) - if self._last_frame_digest is None or digest != self._last_frame_digest: + raw_digest = self._canonical_frame_digest(current_img) + digest = ( + self._remote_frame_contract.comparison_digest(current_png) + if self._remote_frame_contract is not None + else raw_digest + ) + if ( + self._last_comparison_digest is None + or digest != self._last_comparison_digest + ): changed_pixel_count, changed_bbox = self._frame_difference( self._actuation_frame_png, current_img, diff --git a/openadapt_flow/backends/remote_display.py b/openadapt_flow/backends/remote_display.py index aeb79d2d..68f16023 100644 --- a/openadapt_flow/backends/remote_display.py +++ b/openadapt_flow/backends/remote_display.py @@ -73,6 +73,7 @@ StructuralResolutionRefused, ) from openadapt_flow.ir import ActionDeliveryReceipt +from openadapt_flow.remote_frame_contract import RemoteFrameContract from openadapt_flow.runtime.resolver import visual_resolution_point_fingerprint _PNG_SIGNATURE = b"\x89PNG\r\n\x1a\n" @@ -526,8 +527,10 @@ def __init__( session_marker: Optional[str] = None, session_marker_probe: Optional[Callable[[bytes], bool]] = None, session_identity_observer: Optional[Callable[[], Optional[str]]] = None, + remote_frame_contract: Optional["RemoteFrameContract"] = None, ) -> None: self._client = client if client is not None else _default_window_client() + self._remote_frame_contract = remote_frame_contract self._owner_substr = owner_substr self._title_substr = title_substr self._require_input_trust = require_input_trust @@ -589,6 +592,7 @@ def __init__( self._frame_window: Optional[WindowInfo] = None self._last_frame_monotonic: Optional[float] = None self._last_frame_digest: Optional[bytes] = None + self._last_comparison_digest: Optional[bytes] = None self._actuation_frame_png: Optional[bytes] = None self._last_session_identity: Optional[str] = None self._qualification_environment: Optional[tuple[str, str, str, str]] = None @@ -714,11 +718,18 @@ def screenshot(self) -> bytes: f"({scale_x:.4f}x vs {scale_y:.4f}y); refusing uncalibrated input" ) self._viewport = (w, h) + if self._remote_frame_contract is not None: + self._remote_frame_contract.require_geometry(self._viewport) self._scale_x, self._scale_y = scale_x, scale_y self._scale = scale_x # compatibility for existing diagnostics self._frame_window = win self._last_frame_monotonic = time.monotonic() self._last_frame_digest = _canonical_rgb_digest(png) + self._last_comparison_digest = ( + self._remote_frame_contract.comparison_digest(png) + if self._remote_frame_contract is not None + else self._last_frame_digest + ) self._last_session_identity = self._session_identity_from_frame(png) # An ordinary observation is not permission to perform a # consequential remote action. Only acquire_actuation_frame arms @@ -798,6 +809,12 @@ def acquire_actuation_frame(self) -> bytes: self._actuation_frame_png = png return png + def arm_remote_frame_contract( + self, *, protected_regions: tuple[tuple[int, int, int, int], ...] + ) -> None: + if self._remote_frame_contract is not None: + self._remote_frame_contract.arm(protected_regions) + def reset_fresh_actuation_state(self) -> None: """Reset only a typed zero-input content invalidation. @@ -1281,8 +1298,9 @@ def _wait_for_pointer_settle(self) -> None: hover/cursor update reaches the remote framebuffer. Sampling one frame after a fixed sleep can therefore arm a lease on the old pixels and invalidate it moments later. This gate waits for consecutive, - byte-decoded RGB-identical frames; it never masks cursor regions or - weakens the exact post-resolution digest used at the delivery edge. + byte-decoded RGB-identical frames outside any reviewed volatile regions. + The contract applies only to derived comparisons. It never changes the + exact raw frame evidence or learns regions from ordinary runs. """ deadline = time.monotonic() + self._pointer_settle_timeout_s @@ -1291,7 +1309,7 @@ def _wait_for_pointer_settle(self) -> None: poll_s = max(0.01, self._settle_s) while time.monotonic() < deadline: self.screenshot() - digest = self._last_frame_digest + digest = self._last_comparison_digest if digest is not None and digest == previous_digest: stable_frames += 1 else: @@ -1422,8 +1440,16 @@ def _ensure_input_ready( # Consume once before the first input edge. A double click or # multi-character type is one gesture and must not invalidate # itself after its first state-changing edge. - digest = _canonical_rgb_digest(png) - if self._last_frame_digest is None or digest != self._last_frame_digest: + raw_digest = _canonical_rgb_digest(png) + digest = ( + self._remote_frame_contract.comparison_digest(png) + if self._remote_frame_contract is not None + else raw_digest + ) + if ( + self._last_comparison_digest is None + or digest != self._last_comparison_digest + ): changed_pixel_count, changed_bbox = self._frame_difference( self._actuation_frame_png, png, diff --git a/openadapt_flow/connector/client.py b/openadapt_flow/connector/client.py index 0b2a0d4f..93e5fb1f 100644 --- a/openadapt_flow/connector/client.py +++ b/openadapt_flow/connector/client.py @@ -28,6 +28,8 @@ from openadapt_flow.hosted import HostedError +MANAGED_DELIVERY_AUTHORITY_CAPABILITY = "managed_delivery_authority_v1" + class ConnectorClientError(HostedError): """An outbound control-plane call failed.""" @@ -88,7 +90,12 @@ def poll(self, wait_s: int) -> Optional[dict[str, Any]]: """Long-poll for the next leased job. Returns the poll envelope ``{"job": {...}}`` or None on a 204 (no work in the wait window).""" resp = self._client.post( - "/api/connector/poll", json={"wait": wait_s}, headers=self._bearer() + "/api/connector/poll", + json={ + "wait": wait_s, + "capabilities": [MANAGED_DELIVERY_AUTHORITY_CAPABILITY], + }, + headers=self._bearer(), ) if resp.status_code == 204: return None diff --git a/openadapt_flow/connector/executor.py b/openadapt_flow/connector/executor.py index 37fc40b3..1791372d 100644 --- a/openadapt_flow/connector/executor.py +++ b/openadapt_flow/connector/executor.py @@ -41,13 +41,21 @@ import tempfile from dataclasses import dataclass, field from pathlib import Path -from typing import Any, Callable, Optional +from typing import Any, Callable, Mapping, Optional +from urllib.parse import urlsplit from openadapt_flow.connector.config import ConnectorSettings from openadapt_flow.connector.protocol import ByocGovernanceError, ByocJob from openadapt_flow.connector.storage import CustomerStorage, extract_bundle_archive from openadapt_flow.failure_signals import automation_failure_signal from openadapt_flow.ir import RunReport +from openadapt_flow.runner.dispatch_envelope import ( + write_managed_dispatch_envelope_value, +) +from openadapt_flow.runtime.durable.authority import ( + REMOTE_AUTHORITY_TOKEN_ENV, + REMOTE_AUTHORITY_URL_ENV, +) #: A run-gate refusal (fail-closed admission denied) exits 2 before the replay #: creates report.json. @@ -64,7 +72,9 @@ class RunOutcome: #: A runner maps argv -> RunOutcome. The default shells the governed ``run`` CLI #: in a child process; tests inject a fake to avoid launching a GUI. -Runner = Callable[[list[str], Path], RunOutcome] +Runner = Callable[[list[str], Path, Mapping[str, str]], RunOutcome] + +_MANAGED_DELIVERY_AUTHORITY_PATH = "/api/internal/managed-delivery-permit" @dataclass @@ -104,6 +114,7 @@ def build_run_argv( bundle_dir: Path, run_dir: Path, params_file: Optional[Path], + managed_dispatch_file: Optional[Path] = None, ) -> list[str]: """The exact governed CLI invocation for a verified BYOC dispatch. @@ -132,14 +143,23 @@ def build_run_argv( argv += ["--params-file", str(params_file)] if job.target_url: argv += ["--url", job.target_url] + if managed_dispatch_file is not None: + argv += ["--managed-dispatch-file", str(managed_dispatch_file)] if settings.allow_unencrypted: # Local escape hatch, mirroring the governed run CLI; default OFF. argv.append("--allow-unencrypted") return argv -def _subprocess_runner(argv: list[str], run_dir: Path) -> RunOutcome: - proc = subprocess.run(argv, capture_output=True, text=True) # nosec - fixed argv +def _subprocess_runner( + argv: list[str], run_dir: Path, child_env: Mapping[str, str] +) -> RunOutcome: + proc = subprocess.run( # nosec - fixed argv and exact process-local env + argv, + capture_output=True, + text=True, + env=dict(child_env), + ) report_path = run_dir / "report.json" report: dict[str, Any] = {} if report_path.is_file(): @@ -267,6 +287,44 @@ def _write_policy_audit(job: ByocJob, run_dir: Path) -> None: pass +def _assert_managed_authority_is_pinned( + job: ByocJob, settings: ConnectorSettings +) -> None: + """Keep the run capability on the enrolled control-plane origin.""" + + if job.managed_dispatch is None: + return + try: + configured = urlsplit(settings.control_plane_url) + delivered = urlsplit(job.managed_delivery_authority_url or "") + configured_port = configured.port or ( + 443 if configured.scheme == "https" else None + ) + delivered_port = delivered.port or ( + 443 if delivered.scheme == "https" else None + ) + except ValueError as exc: + raise ByocGovernanceError( + "byoc managed delivery authority URL is invalid" + ) from exc + if ( + configured.scheme != "https" + or not configured.hostname + or configured.username is not None + or configured.password is not None + or bool(configured.query) + or bool(configured.fragment) + or delivered.scheme != "https" + or delivered.hostname != configured.hostname + or delivered_port != configured_port + or delivered.path != _MANAGED_DELIVERY_AUTHORITY_PATH + ): + raise ByocGovernanceError( + "byoc managed delivery authority is not pinned to the enrolled " + "control plane (fail closed)" + ) + + def execute_job( job: ByocJob, settings: ConnectorSettings, @@ -284,6 +342,7 @@ def execute_job( # 1. Governance gates (fail closed BEFORE any bundle is fetched or run). try: job.ensure_governed(require_run_token=require_run_token) + _assert_managed_authority_is_pinned(job, settings) except ByocGovernanceError as exc: return ExecutionResult("failed", {}, None, job.report_ref(), str(exc)) if not _grounding_env_available(job): @@ -332,10 +391,32 @@ def execute_job( try: params_file = _write_params_file(job.params, run_dir) + managed_dispatch_file = None + child_env = os.environ.copy() + # Never let a long-running Connector inherit stale authority from + # its service environment. Add the exact run capability only for + # the child that also receives the matching dispatch envelope. + child_env.pop(REMOTE_AUTHORITY_URL_ENV, None) + child_env.pop(REMOTE_AUTHORITY_TOKEN_ENV, None) + if job.managed_dispatch is not None: + managed_dispatch_file = write_managed_dispatch_envelope_value( + run_dir / "managed-dispatch.json", job.managed_dispatch + ) + child_env[REMOTE_AUTHORITY_URL_ENV] = str( + job.managed_delivery_authority_url + ) + child_env[REMOTE_AUTHORITY_TOKEN_ENV] = str(job.run_token) # 3. The governed, fail-closed child invocation. - argv = build_run_argv(job, settings, Path(bundle_dir), run_dir, params_file) - outcome = runner(argv, run_dir) + argv = build_run_argv( + job, + settings, + Path(bundle_dir), + run_dir, + params_file, + managed_dispatch_file, + ) + outcome = runner(argv, run_dir, child_env) except Exception as exc: return ExecutionResult( "failed", diff --git a/openadapt_flow/connector/protocol.py b/openadapt_flow/connector/protocol.py index be6734fe..5126d9ec 100644 --- a/openadapt_flow/connector/protocol.py +++ b/openadapt_flow/connector/protocol.py @@ -21,10 +21,12 @@ from __future__ import annotations from typing import Any, Optional +from urllib.parse import urlsplit from pydantic import BaseModel, ConfigDict, Field, ValidationError from openadapt_flow.ir import ExecutionTargetKind +from openadapt_flow.runner.dispatch_envelope import ManagedDispatchEnvelope class ByocJobParseError(ValueError): @@ -100,9 +102,15 @@ class ByocJob(BaseModel): bundle_download_url: Optional[str] = None # --- Governed callback binding (fail-closed) ------------------------------- - #: Run-scoped HMAC capability presented as ``x-run-token`` on the PHI-free - #: callback. Not a secret value — proves this run, forbids forging another. + #: Run-scoped bearer capability presented as ``x-run-token`` on the PHI-free + #: callback. It proves this run and must remain private. run_token: Optional[str] = None + #: Exact Cloud-issued, one-run authority. It stays PHI-free and is written + #: to a private local file before the governed child starts. + managed_dispatch: Optional[ManagedDispatchEnvelope] = None + #: HTTPS endpoint that issues monotonic per-action delivery permits for the + #: exact managed dispatch. The run-scoped ``run_token`` authenticates it. + managed_delivery_authority_url: Optional[str] = None bundle_version_id: Optional[str] = None runtime_validation_id: Optional[str] = None #: SHA-256 of the exact approved sanitized derivative ZIP bytes staged at @@ -162,6 +170,46 @@ def ensure_governed(self, *, require_run_token: bool = True) -> None: "byoc dispatch is missing a run-scoped callback token; refusing " "to run a job whose outcome we could not report (fail closed)" ) + if (self.managed_dispatch is None) != ( + self.managed_delivery_authority_url is None + ): + raise ByocGovernanceError( + "byoc managed dispatch and delivery authority must be supplied " + "together (fail closed)" + ) + if self.managed_dispatch is not None: + try: + authorization = self.managed_dispatch.exact_authorization() + except ValueError as exc: + raise ByocGovernanceError( + "byoc managed dispatch has an invalid authorization binding" + ) from exc + if self.managed_dispatch.run_id != self.run_id: + raise ByocGovernanceError( + "byoc managed dispatch belongs to a different run (fail closed)" + ) + if not authorization.approval_source.startswith("hosted:"): + raise ByocGovernanceError( + "byoc managed dispatch lacks hosted authorization provenance" + ) + try: + parsed = urlsplit(self.managed_delivery_authority_url or "") + except ValueError as exc: + raise ByocGovernanceError( + "byoc managed delivery authority URL is invalid" + ) from exc + if ( + parsed.scheme != "https" + or not parsed.hostname + or parsed.username is not None + or parsed.password is not None + or bool(parsed.query) + or bool(parsed.fragment) + ): + raise ByocGovernanceError( + "byoc managed delivery authority must be a credential-free " + "HTTPS endpoint without query or fragment" + ) if not self.bundle_version_id or not self.runtime_validation_id: raise ByocGovernanceError( "byoc dispatch is missing its immutable bundle-version or " diff --git a/openadapt_flow/deployment.py b/openadapt_flow/deployment.py index 91c4ee2e..34dffa80 100644 --- a/openadapt_flow/deployment.py +++ b/openadapt_flow/deployment.py @@ -26,6 +26,8 @@ from pydantic import BaseModel, Field, field_validator, model_validator +from openadapt_flow.remote_frame_contract import RemoteFrameContract + # Import-light (pydantic only): the effect CONTRACT types double as the # declarative config vocabulary, so a deployment YAML binds run parameters # with the exact ``{param: ...}`` / ``{literal: ...}`` form bundles use. @@ -141,6 +143,9 @@ class BackendConfig(BaseModel): #: Optional exact case-insensitive title. Zero or multiple exact matches are #: refused; the backend never selects a largest partial match. rdp_window_title: Optional[str] = None + # Exact reviewed geometry for derived remote settle comparisons. The + # enclosing DeploymentConfig serialization binds this value at admission. + remote_frame_contract: Optional[RemoteFrameContract] = None class EffectsConfig(BaseModel): diff --git a/openadapt_flow/remote_frame_contract.py b/openadapt_flow/remote_frame_contract.py new file mode 100644 index 00000000..7cc21040 --- /dev/null +++ b/openadapt_flow/remote_frame_contract.py @@ -0,0 +1,76 @@ +"""Immutable, deployment-bound comparison masks for remote frame settling.""" + +from __future__ import annotations + +import hashlib +import io +from typing import Literal + +from PIL import Image, ImageDraw +from pydantic import BaseModel, ConfigDict, Field, model_validator + +Region = tuple[int, int, int, int] + + +class RemoteFrameContract(BaseModel): + """Reviewed exact-geometry exclusions for derived settle inputs only.""" + + model_config = ConfigDict(extra="forbid", frozen=True) + schema_version: Literal["openadapt.remote-frame-contract/v1"] = ( + "openadapt.remote-frame-contract/v1" + ) + frame_width: int = Field(gt=0, le=32768) + frame_height: int = Field(gt=0, le=32768) + volatile_regions: tuple[Region, ...] = Field(min_length=1, max_length=32) + protected_regions: tuple[Region, ...] = Field(default_factory=tuple, max_length=128) + + @model_validator(mode="after") + def validate_regions(self) -> "RemoteFrameContract": + for volatile in self.volatile_regions: + self._validate(volatile) + for protected in self.protected_regions: + self._validate(protected) + if _overlap(volatile, protected): + raise ValueError("volatile region overlaps a protected region") + return self + + def _validate(self, region: Region) -> None: + x, y, width, height = region + if ( + width <= 0 + or height <= 0 + or x < 0 + or y < 0 + or x + width > self.frame_width + or y + height > self.frame_height + ): + raise ValueError("remote frame region is outside the exact qualified frame") + + def require_geometry(self, size: tuple[int, int]) -> None: + if size != (self.frame_width, self.frame_height): + raise ValueError("remote frame contract geometry does not match live frame") + + def arm(self, protected_regions: tuple[Region, ...]) -> None: + """Refuse a newly observed target/identity/effect overlap before input.""" + for region in protected_regions: + self._validate(region) + if any(_overlap(region, volatile) for volatile in self.volatile_regions): + raise ValueError("volatile region overlaps a runtime protected region") + + def comparison_digest(self, png: bytes) -> bytes: + """Hash a derived masked copy. Raw evidence and leases stay unmasked.""" + image = Image.open(io.BytesIO(png)).convert("RGB") + self.require_geometry(image.size) + derived = image.copy() + draw = ImageDraw.Draw(derived) + for x, y, width, height in self.volatile_regions: + draw.rectangle((x, y, x + width - 1, y + height - 1), fill=(0, 0, 0)) + out = io.BytesIO() + derived.save(out, format="PNG") + return hashlib.sha256(out.getvalue()).digest() + + +def _overlap(left: Region, right: Region) -> bool: + x, y, width, height = left + a, b, c, d = right + return x < a + c and a < x + width and y < b + d and b < y + height diff --git a/openadapt_flow/runner/dispatch_envelope.py b/openadapt_flow/runner/dispatch_envelope.py index 5fdff6b1..693e2652 100644 --- a/openadapt_flow/runner/dispatch_envelope.py +++ b/openadapt_flow/runner/dispatch_envelope.py @@ -93,18 +93,18 @@ def exact_authorization(self) -> GovernedRunAuthorization: return self.authorization -def write_managed_dispatch_envelope(path: Path, verified: "VerifiedDispatch") -> Path: - """Write one already verified dispatch without following a path.""" - - run_id = verified.payload.run_id - authorization = verified.payload.authorization - envelope = ManagedDispatchEnvelope( - run_id=run_id, - bundle_content_digest=authorization.bundle_content_digest, - runtime_inputs_digest=authorization.runtime_inputs_digest, - authorization=authorization, - dispatch_binding_sha256=verified.payload.dispatch_binding_sha256, - ) +def write_managed_dispatch_envelope_value( + path: Path, envelope: ManagedDispatchEnvelope +) -> Path: + """Write one strictly parsed dispatch envelope without following a path. + + This is the shared process-boundary writer for both push runners and the + customer-controlled outbound-pull Connector. The caller must already hold + the strict :class:`ManagedDispatchEnvelope` model; this function rechecks + its internal authorization binding before any bytes reach disk. + """ + + envelope.exact_authorization() path = Path(path) path.parent.mkdir(parents=True, exist_ok=True) flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL @@ -136,6 +136,21 @@ def write_managed_dispatch_envelope(path: Path, verified: "VerifiedDispatch") -> return path +def write_managed_dispatch_envelope(path: Path, verified: "VerifiedDispatch") -> Path: + """Write one already verified runner dispatch without following a path.""" + + run_id = verified.payload.run_id + authorization = verified.payload.authorization + envelope = ManagedDispatchEnvelope( + run_id=run_id, + bundle_content_digest=authorization.bundle_content_digest, + runtime_inputs_digest=authorization.runtime_inputs_digest, + authorization=authorization, + dispatch_binding_sha256=verified.payload.dispatch_binding_sha256, + ) + return write_managed_dispatch_envelope_value(path, envelope) + + def _read_managed_dispatch_envelope(path: Path) -> ManagedDispatchEnvelope: """Read only a regular, private file owned by this effective user.""" diff --git a/openadapt_flow/runtime/durable/authority.py b/openadapt_flow/runtime/durable/authority.py index 130f4d59..ca42a0ce 100644 --- a/openadapt_flow/runtime/durable/authority.py +++ b/openadapt_flow/runtime/durable/authority.py @@ -31,7 +31,7 @@ from typing import Any, Iterator, Literal, Optional from urllib.error import HTTPError, URLError from urllib.parse import urlparse -from urllib.request import Request, urlopen +from urllib.request import HTTPRedirectHandler, Request, build_opener from pydantic import BaseModel, ConfigDict @@ -70,6 +70,27 @@ } +class _RefuseRemoteAuthorityRedirects(HTTPRedirectHandler): + """Keep the runner credential on the configured authority origin.""" + + def redirect_request( + self, + req: Request, + fp: Any, + code: int, + msg: str, + headers: Any, + newurl: str, + ) -> None: + raise HTTPError( + req.full_url, + code, + "remote delivery authority redirects are refused", + headers, + fp, + ) + + def _is_windows() -> bool: return os.name == "nt" @@ -1645,7 +1666,8 @@ def _require_remote_delivery_permit( response_bytes = self._remote_transport(url, headers, body) else: request = Request(url, data=body, headers=headers, method="POST") - with urlopen(request, timeout=10) as response: # nosec B310 - HTTPS above + opener = build_opener(_RefuseRemoteAuthorityRedirects()) + with opener.open(request, timeout=10) as response: # nosec B310 - HTTPS above if not 200 <= response.status < 300: raise DurableAuthorityBusy("remote delivery authority refused") response_bytes = response.read( diff --git a/openadapt_flow/runtime/replayer.py b/openadapt_flow/runtime/replayer.py index 513e64dd..90d6490e 100644 --- a/openadapt_flow/runtime/replayer.py +++ b/openadapt_flow/runtime/replayer.py @@ -61,6 +61,7 @@ GuardedSelectOptionBackend, PreparedPointerActuationBackend, RemoteActuationBackend, + RemoteFrameContractBackend, RichPointerActionBackend, SelectOptionBackend, StructuralResolutionRefused, @@ -7543,6 +7544,33 @@ def _requires_atomic_identity_keyboard( and self._step_has_identity_contract(step, workflow) ) + def _typed_remote_receipt_required(self) -> bool: + """Whether a remote pointer edge must carry a typed delivery receipt. + + :class:`RemoteActuationBackend` already owns the exact-frame contract: + after ``acquire_actuation_frame`` the backend captures once more under + its own input lock and refuses *before the first input edge* when the + window/session, dimensions, readiness, or exact frame content changed. + :class:`GuardedRemotePointerActionBackend` adds an explicit + expected-frame hash plus a typed, target-bound receipt on top of that + contract; its absence is therefore missing *evidence*, not a missing + safety property. + + A governed (Standard/Regulated) run and an exact qualification-fault + proof both consume that receipt as retained evidence, so they must + refuse before delivery rather than actuate without it. Ordinary replay + keeps the documented lease path — refusing there over-halts every + conforming opaque remote backend, including a pixel-only no-DOM canvas + that can only expose the lease. Such a result stays unlabeled, which + the outcome classifier already maps to ``COMPLETED_UNVERIFIED`` instead + of a production-eligible success. + """ + + return ( + self.governed_authorization is not None + or self.qualification_fault_driver is not None + ) + def _cancel_guarded_coordinate(self) -> None: """Clean an unconsumed local coordinate lease, if one was armed.""" @@ -7921,6 +7949,36 @@ def _revalidate_consequential_actuation( self._cancel_guarded_keyboard() if error is not None and focused_element_backend: self._cancel_guarded_keyboard() + if error is None and isinstance(self.backend, RemoteFrameContractBackend): + protected: list[Region] = [] + if fresh_region is not None: + protected.append(fresh_region) + elif fresh_resolution is not None: + x, y = fresh_resolution.point + protected.append((x - 1, y - 1, 3, 3)) + protected.extend( + pc.region for pc in step.expect if pc.region is not None + ) + if workflow.qualification is not None: + policy = workflow.qualification.identity_policies.get(step.id) + if policy is not None: + protected.extend( + s.region for s in policy.signals if s.region is not None + ) + try: + self.backend.arm_remote_frame_contract( + protected_regions=tuple(protected) + ) + except Exception as exc: + return ( + fresh_resolution, + fresh_region, + fresh_png, + ( + "Actuation preflight HALTED because the remote frame mask " + f"overlaps protected evidence: {type(exc).__name__}" + ), + ) # Retain the exact observation that authorizes the next input edge. # Composite TYPE/SELECT_OPTION and retry paths can re-resolve inside # ``_act`` after the outer scope captured its initial geometry. A typed @@ -8577,39 +8635,58 @@ def _act( self._step_is_consequential(step, workflow) or self.qualification_fault_driver is not None ) + typed_remote = isinstance( + self.backend, + GuardedRemotePointerActionBackend, + ) + if ( + remote_consequential + and not typed_remote + and self._typed_remote_receipt_required() + ): + result.safety_halt = True + result.failure_category = "safety_halt" + return ( + f"Step '{step.id}' ({step.intent}) is a consequential " + "remote click, but this backend cannot bind its exact " + "fresh frame and target to delivery; run aborted" + ) if remote_consequential: - if not isinstance( - self.backend, - GuardedRemotePointerActionBackend, - ): - result.safety_halt = True - result.failure_category = "safety_halt" - return ( - f"Step '{step.id}' ({step.intent}) is a consequential " - "remote click, but this backend cannot bind its exact " - "fresh frame and target to delivery; run aborted" - ) refusal = self._delivery_authorization_refusal( workflow, params, step, result ) if refusal is not None: return refusal self._require_qualification_environment_current() - remote_pointer = cast( - GuardedRemotePointerActionBackend, self.backend - ) - result.delivery_receipt = self._deliver_backend_call( - result, - lambda: remote_pointer.click_guarded( - x, - y, - expected_frame_sha256=hashlib.sha256( - before_png - ).hexdigest(), - double=step.action is ActionKind.DOUBLE_CLICK, - ), - ) - result.actuation = "remote_guarded" + if typed_remote: + remote_pointer = cast( + GuardedRemotePointerActionBackend, self.backend + ) + result.delivery_receipt = self._deliver_backend_call( + result, + lambda: remote_pointer.click_guarded( + x, + y, + expected_frame_sha256=hashlib.sha256( + before_png + ).hexdigest(), + double=step.action is ActionKind.DOUBLE_CLICK, + ), + ) + result.actuation = "remote_guarded" + else: + # See ``_typed_remote_receipt_required``: the exact + # fresh frame is already bound by the backend's own + # one-shot lease, armed immediately above by + # ``_revalidate_consequential_actuation``. + self._deliver_backend_call( + result, + lambda: self.backend.click( + x, + y, + double=step.action is ActionKind.DOUBLE_CLICK, + ), + ) elif requires_atomic_identity: if isinstance(self.backend, GuardedCoordinateActionBackend): refusal = self._delivery_authorization_refusal( @@ -8683,16 +8760,39 @@ def _act( self._step_is_consequential(step, workflow) or self.qualification_fault_driver is not None ) - if remote_consequential: - if not isinstance( - self.backend, - GuardedRemotePointerActionBackend, - ): + if remote_consequential and not isinstance( + self.backend, + GuardedRemotePointerActionBackend, + ): + if self._typed_remote_receipt_required(): return ( f"Step '{step.id}' ({step.intent}) is a consequential " "remote right click, but this backend cannot bind its " "exact fresh frame and target to delivery; run aborted" ) + # See ``_typed_remote_receipt_required``: the exact fresh frame + # is already bound by the backend's own one-shot lease, and its + # first input edge consumes it. + if not isinstance(self.backend, RichPointerActionBackend): + return ( + f"Step '{step.id}' ({step.intent}) requires a right " + "click, but this backend has no bounded right-click " + "operation" + ) + refusal = self._delivery_authorization_refusal( + workflow, params, step, result + ) + if refusal is not None: + return refusal + self._require_qualification_environment_current() + self._deliver_backend_call( + result, + lambda: cast(RichPointerActionBackend, self.backend).right_click( + x, y + ), + ) + return None + if remote_consequential: refusal = self._delivery_authorization_refusal( workflow, params, step, result ) @@ -8825,9 +8925,16 @@ def _act( ), ) result.actuation = "dom" - elif isinstance(self.backend, RemoteActuationBackend) and ( - self._step_is_consequential(step, workflow) - or self.qualification_fault_driver is not None + elif ( + isinstance(self.backend, RemoteActuationBackend) + and ( + self._step_is_consequential(step, workflow) + or self.qualification_fault_driver is not None + ) + and ( + isinstance(self.backend, GuardedRemotePointerActionBackend) + or self._typed_remote_receipt_required() + ) ): if not isinstance( self.backend, diff --git a/pyproject.toml b/pyproject.toml index 3b20a037..d70b09c1 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "openadapt-flow" -version = "1.27.0" +version = "1.27.1" description = "Compile demonstrated GUI workflows into deterministic local replay with governed repair and refusal" readme = "README.md" license = { text = "MIT" } diff --git a/tests/test_connector.py b/tests/test_connector.py index 6d23edf0..79a39a3e 100644 --- a/tests/test_connector.py +++ b/tests/test_connector.py @@ -20,6 +20,8 @@ import hashlib import io import json +import os +import stat import zipfile from pathlib import Path @@ -44,6 +46,13 @@ from openadapt_flow.connector.protocol import ByocGovernanceError from openadapt_flow.connector.storage import LocalCustomerStorage from openadapt_flow.failure_signals import automation_failure_signal +from openadapt_flow.runner.dispatch_envelope import ManagedDispatchEnvelope +from openadapt_flow.runner.protocol import dispatch_binding_sha256 +from openadapt_flow.runtime.authorization import GovernedRunAuthorization +from openadapt_flow.runtime.durable.authority import ( + REMOTE_AUTHORITY_TOKEN_ENV, + REMOTE_AUTHORITY_URL_ENV, +) def _bundle_archive_bytes() -> bytes: @@ -90,8 +99,27 @@ def _payload(**overrides): return payload +def _managed_dispatch(*, run_id: str | None = None) -> dict: + exact_run_id = run_id or _payload()["run_id"] + authorization = GovernedRunAuthorization( + bundle_content_digest="b" * 64, + runtime_inputs_digest="c" * 64, + admitted_policy_name="standard", + execution_profile="standard", + approval_source="hosted:execute-api", + ) + envelope = ManagedDispatchEnvelope( + run_id=exact_run_id, + bundle_content_digest=authorization.bundle_content_digest, + runtime_inputs_digest=authorization.runtime_inputs_digest, + authorization=authorization, + dispatch_binding_sha256=dispatch_binding_sha256(exact_run_id, authorization), + ) + return envelope.model_dump(mode="json") + + def _fake_success_runner(report): - def runner(argv, run_dir: Path) -> RunOutcome: + def runner(argv, run_dir: Path, _env) -> RunOutcome: (run_dir / "report.json").write_text(json.dumps(report)) return RunOutcome(returncode=0, report=report) @@ -134,6 +162,68 @@ def test_execute_job_success_writes_report_to_customer_storage_only(): assert storage.written["org_demo/run_5/report.json"]["success"] is True +def test_execute_job_consumes_exact_managed_dispatch_through_private_child_boundary( + monkeypatch, +): + authority_url = "https://app.openadapt.ai/api/internal/managed-delivery-permit" + token = "a" * 64 + managed_dispatch = _managed_dispatch() + monkeypatch.setenv(REMOTE_AUTHORITY_URL_ENV, "https://stale.invalid/permit") + monkeypatch.setenv(REMOTE_AUTHORITY_TOKEN_ENV, "stale-parent-token") + job = parse_job( + _payload( + managed_dispatch=managed_dispatch, + managed_delivery_authority_url=authority_url, + ), + lease_job_id="bjob_1", + ) + storage = InMemoryCustomerStorage(bundle_bytes=_BUNDLE_BYTES) + observed: dict[str, object] = {} + + def runner(argv, run_dir: Path, child_env) -> RunOutcome: + envelope_path = Path(argv[argv.index("--managed-dispatch-file") + 1]) + observed["argv"] = argv + observed["mode"] = stat.S_IMODE(envelope_path.stat().st_mode) + observed["envelope"] = json.loads(envelope_path.read_text()) + observed["authority_url"] = child_env[REMOTE_AUTHORITY_URL_ENV] + observed["authority_token"] = child_env[REMOTE_AUTHORITY_TOKEN_ENV] + observed["parent_token"] = os.environ.get(REMOTE_AUTHORITY_TOKEN_ENV) + (run_dir / "report.json").write_text(json.dumps(SUCCESS_REPORT)) + return RunOutcome(returncode=0, report=SUCCESS_REPORT) + + result = execute_job(job, ConnectorSettings(), storage, runner=runner) + + assert result.status == "success" + assert observed["mode"] == 0o600 + assert observed["envelope"] == managed_dispatch + assert observed["authority_url"] == authority_url + assert observed["authority_token"] == token + assert observed["parent_token"] == "stale-parent-token" + assert token not in json.dumps(observed["argv"]) + assert token not in json.dumps(observed["envelope"]) + + +def test_connector_drops_inherited_authority_from_a_job_without_an_envelope( + monkeypatch, +): + monkeypatch.setenv(REMOTE_AUTHORITY_URL_ENV, "https://stale.invalid/permit") + monkeypatch.setenv(REMOTE_AUTHORITY_TOKEN_ENV, "stale-parent-token") + job = parse_job(_payload(), lease_job_id="bjob_1") + storage = InMemoryCustomerStorage(bundle_bytes=_BUNDLE_BYTES) + + def runner(argv, run_dir: Path, child_env) -> RunOutcome: + assert REMOTE_AUTHORITY_URL_ENV not in child_env + assert REMOTE_AUTHORITY_TOKEN_ENV not in child_env + assert "--managed-dispatch-file" not in argv + (run_dir / "report.json").write_text(json.dumps(SUCCESS_REPORT)) + return RunOutcome(returncode=0, report=SUCCESS_REPORT) + + assert ( + execute_job(job, ConnectorSettings(), storage, runner=runner).status + == "success" + ) + + def test_execute_job_refuses_child_report_for_different_substrate(): job = parse_job(_payload(), lease_job_id="bjob_1") settings = ConnectorSettings(profile=None) @@ -281,7 +371,7 @@ def test_halt_maps_to_halt_status_and_present_flag(): settings = ConnectorSettings() storage = InMemoryCustomerStorage(bundle_bytes=_BUNDLE_BYTES) - def runner(argv, run_dir: Path) -> RunOutcome: + def runner(argv, run_dir: Path, _env) -> RunOutcome: (run_dir / "report.json").write_text(json.dumps(halt_report)) return RunOutcome(returncode=2, report=halt_report) @@ -365,6 +455,64 @@ def test_dispatch_without_run_token_is_refused(): job.ensure_governed() +def test_managed_dispatch_for_different_run_is_refused(): + job = parse_job( + _payload( + managed_dispatch=_managed_dispatch( + run_id="00000000-0000-4000-8000-000000000006" + ), + managed_delivery_authority_url=( + "https://app.openadapt.ai/api/internal/managed-delivery-permit" + ), + ), + lease_job_id="bjob_1", + ) + with pytest.raises(ByocGovernanceError, match="different run"): + job.ensure_governed() + + +def test_execute_refuses_managed_authority_on_a_different_https_origin(): + job = parse_job( + _payload( + managed_dispatch=_managed_dispatch(), + managed_delivery_authority_url=( + "https://attacker.invalid/api/internal/managed-delivery-permit" + ), + ), + lease_job_id="bjob_1", + ) + result = execute_job( + job, + ConnectorSettings(control_plane_url="https://app.openadapt.ai"), + InMemoryCustomerStorage(bundle_bytes=_BUNDLE_BYTES), + runner=_fake_success_runner(SUCCESS_REPORT), + ) + assert result.status == "failed" + assert "pinned to the enrolled control plane" in (result.error or "") + + +@pytest.mark.parametrize( + "overrides, match", + [ + ( + {"managed_dispatch": _managed_dispatch()}, + "must be supplied together", + ), + ( + { + "managed_dispatch": _managed_dispatch(), + "managed_delivery_authority_url": "http://localhost/permit", + }, + "credential-free HTTPS endpoint", + ), + ], +) +def test_incomplete_or_unsafe_managed_authority_is_refused(overrides, match): + job = parse_job(_payload(**overrides), lease_job_id="bjob_1") + with pytest.raises(ByocGovernanceError, match=match): + job.ensure_governed() + + @pytest.mark.parametrize("field", ["bundle_version_id", "runtime_validation_id"]) def test_dispatch_without_immutable_validation_binding_is_refused(field): job = parse_job(_payload(**{field: None}), lease_job_id="bjob_1") @@ -398,7 +546,7 @@ def test_execute_refuses_when_required_grounding_key_is_absent(monkeypatch): # A runner that would "succeed" — the governance gate must refuse BEFORE it. called = {"ran": False} - def runner(argv, run_dir: Path) -> RunOutcome: + def runner(argv, run_dir: Path, _env) -> RunOutcome: called["ran"] = True return RunOutcome(returncode=0, report=SUCCESS_REPORT) @@ -413,7 +561,7 @@ def test_execute_refuses_archive_digest_mismatch_before_runner(): storage = InMemoryCustomerStorage(bundle_bytes=_BUNDLE_BYTES) called = {"ran": False} - def runner(argv, run_dir: Path) -> RunOutcome: + def runner(argv, run_dir: Path, _env) -> RunOutcome: called["ran"] = True return RunOutcome(returncode=0, report=SUCCESS_REPORT) @@ -429,7 +577,7 @@ def test_child_failure_after_archive_verification_carries_verified_digest(): job = parse_job(_payload(), lease_job_id="bjob_1") storage = InMemoryCustomerStorage(bundle_bytes=_BUNDLE_BYTES) - def runner(argv, run_dir: Path) -> RunOutcome: + def runner(argv, run_dir: Path, _env) -> RunOutcome: raise RuntimeError("child unavailable") result = execute_job(job, ConnectorSettings(), storage, runner=runner) @@ -507,6 +655,7 @@ def __init__(self): self.tokens = {} # token -> org_id self.jobs = [] # queued jobs (each {id, org_id, payload, status, leased_by}) self.callbacks = [] + self.poll_bodies = [] self.callback_attempts = 0 self.callback_status = 200 self.acks = [] @@ -542,6 +691,7 @@ def handler(self, request: httpx.Request) -> httpx.Response: if path == "/api/connector/poll": if org is None: return httpx.Response(401, json={"error": "invalid connector token"}) + self.poll_bodies.append(body) for j in self.jobs: # ISOLATION: a connector only ever leases ITS OWN org jobs. if j["status"] == "queued" and j["org_id"] == org: @@ -611,6 +761,9 @@ def test_full_loop_dispatch_execute_callback_ack(): storage_factory=lambda job: storage, ) assert result["status"] == "success" + assert cp.poll_bodies == [ + {"wait": 0, "capabilities": ["managed_delivery_authority_v1"]} + ] # A PHI-free callback carrying the run-scoped token was posted. assert len(cp.callbacks) == 1 diff --git a/tests/test_durable_authority_v13.py b/tests/test_durable_authority_v13.py index f9f26953..bbc65f6e 100644 --- a/tests/test_durable_authority_v13.py +++ b/tests/test_durable_authority_v13.py @@ -17,6 +17,7 @@ import pytest +import openadapt_flow.runtime.durable.authority as durable_authority_module from openadapt_flow.ir import ActionKind, RunReport, Step, StepResult, Workflow from openadapt_flow.runtime.authorization import GovernedRunAuthorization from openadapt_flow.runtime.durable.approval import ( @@ -296,6 +297,60 @@ def transport(_url: str, _headers: dict[str, str], _body: bytes) -> bytes: assert authority.validate(manifest).delivery_sequence == 0 +def test_remote_authority_refuses_redirect_before_forwarding_bearer_token( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + manifest, authority, owner = _remote_ready_authority( + tmp_path, monkeypatch, lambda *_args: b"{}" + ) + authority._remote_transport = None + monkeypatch.setenv( + REMOTE_AUTHORITY_URL_ENV, + "https://control.example/api/internal/managed-delivery-permit", + ) + attempted_urls: list[str] = [] + + class RedirectingOpener: + def __init__(self, handler: object) -> None: + self.handler = handler + + def open(self, request: object, *, timeout: int) -> object: + assert timeout == 10 + attempted_urls.append(request.full_url) # type: ignore[attr-defined] + assert request.get_header("Authorization") == "Bearer secret-token" # type: ignore[attr-defined] + self.handler.redirect_request( # type: ignore[attr-defined] + request, + None, + 307, + "Temporary Redirect", + {}, + "https://attacker.invalid/permit", + ) + raise AssertionError("redirect refusal must stop the request") + + def build_redirecting_opener(handler: object) -> RedirectingOpener: + assert isinstance( + handler, durable_authority_module._RefuseRemoteAuthorityRedirects + ) + return RedirectingOpener(handler) + + monkeypatch.setattr( + durable_authority_module, + "build_opener", + build_redirecting_opener, + ) + with pytest.raises(DurableAuthorityBusy, match="unavailable or refused"): + authority.before_delivery( + manifest, + attempt_id="aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + owner_nonce_sha256=owner, + ) + assert attempted_urls == [ + "https://control.example/api/internal/managed-delivery-permit" + ] + assert authority.validate(manifest).delivery_sequence == 0 + + def test_initial_customer_local_delivery_needs_no_cloud_credentials( tmp_path: Path, ) -> None: diff --git a/tests/test_rdp_backend.py b/tests/test_rdp_backend.py index 21c7234a..36aa1f3d 100644 --- a/tests/test_rdp_backend.py +++ b/tests/test_rdp_backend.py @@ -37,6 +37,7 @@ RDPTransport, normalize_chord, ) +from openadapt_flow.remote_frame_contract import RemoteFrameContract from openadapt_flow.runtime.resolver import visual_resolution_point_fingerprint VIEWPORT = (1280, 800) @@ -625,6 +626,67 @@ def test_bound_actuation_refuses_same_session_content_change_before_input() -> N assert raised.value.frame_size == VIEWPORT +def test_bound_actuation_allows_only_qualified_rdp_volatile_change() -> None: + transport = FakeRDPTransport(app_screens()) + contract = RemoteFrameContract( + frame_width=VIEWPORT[0], + frame_height=VIEWPORT[1], + volatile_regions=((0, 0, 32, 32),), + protected_regions=(BUTTON,), + ) + backend = FreeRDPBackend( + transport, + readiness_probe=lambda _png: True, + remote_frame_contract=contract, + ) + backend.acquire_actuation_frame() + backend.arm_remote_frame_contract(protected_regions=(BUTTON,)) + leased_raw_digest = backend._last_frame_digest + changed = transport.screens[0].copy() + changed.putpixel((1, 1), (0, 0, 0)) + transport.screens[0] = changed + + assert leased_raw_digest is not None + assert backend._canonical_frame_digest(changed) != leased_raw_digest + + backend.click(*BUTTON_CENTER) + + assert transport.pointer_events == [ + (*BUTTON_CENTER, "left", True), + (*BUTTON_CENTER, "left", False), + ] + assert backend._last_frame_digest == leased_raw_digest + + +def test_bound_actuation_refuses_rdp_change_outside_qualified_volatile_region() -> None: + transport = FakeRDPTransport(app_screens()) + contract = RemoteFrameContract( + frame_width=VIEWPORT[0], + frame_height=VIEWPORT[1], + volatile_regions=((0, 0, 32, 32),), + protected_regions=(BUTTON,), + ) + backend = FreeRDPBackend( + transport, + readiness_probe=lambda _png: True, + remote_frame_contract=contract, + ) + backend.acquire_actuation_frame() + backend.arm_remote_frame_contract(protected_regions=(BUTTON,)) + changed = transport.screens[0].copy() + changed.putpixel((50, 50), (0, 0, 0)) + transport.screens[0] = changed + + with pytest.raises(FreshActuationRequired) as raised: + backend.click(*BUTTON_CENTER) + + assert transport.pointer_events == [] + assert raised.value.operation == "rdp_click" + assert raised.value.changed_pixel_count == 1 + assert raised.value.changed_bbox == (50, 50, 1, 1) + assert raised.value.frame_size == VIEWPORT + + def test_bound_actuation_can_reset_only_a_typed_zero_edge_invalidation() -> None: transport = FakeRDPTransport(app_screens()) backend = FreeRDPBackend(transport, readiness_probe=lambda _png: True) diff --git a/tests/test_rdp_ladder_qualification.py b/tests/test_rdp_ladder_qualification.py index 704f4edd..a969d52c 100644 --- a/tests/test_rdp_ladder_qualification.py +++ b/tests/test_rdp_ladder_qualification.py @@ -251,11 +251,8 @@ def test_presentation_is_hash_bound_and_renders_without_staged_video_frames( ) assert all("sha256" in entry["source_frame"] for entry in source_entries) assert all("target_geometry" not in entry for entry in timeline["timeline"]) - graph_entries = [ - entry for entry in timeline["timeline"] if "compiled_graph" in entry - ] - assert graph_entries - assert all("node_id" in entry["compiled_graph"] for entry in graph_entries) + assert "compiled_workflow" not in {entry["phase"] for entry in timeline["timeline"]} + assert not any("compiled_graph" in entry for entry in timeline["timeline"]) source_manifests = { name: json.loads( (tmp_path / name / "manifest.json").read_text(encoding="utf-8") @@ -276,9 +273,10 @@ def test_presentation_is_hash_bound_and_renders_without_staged_video_frames( ]["sha256"] = "0" * 64 invalid_timelines.append((invalid, "does not match its manifest")) invalid = copy.deepcopy(timeline) - next(entry for entry in invalid["timeline"] if "compiled_graph" in entry)[ - "compiled_graph" - ]["node_id"] = "different-step" + invalid["timeline"][0]["compiled_graph"] = { + "node_index": 0, + "node_id": "different-step", + } invalid_timelines.append((invalid, "does not match the graph")) invalid = copy.deepcopy(timeline) invalid["timeline"][0]["facts"]["record_value"] = "not allowed" diff --git a/tests/test_remote_display_backend.py b/tests/test_remote_display_backend.py index 88f15352..ea9b8c17 100644 --- a/tests/test_remote_display_backend.py +++ b/tests/test_remote_display_backend.py @@ -34,9 +34,11 @@ RemoteDisplayBackend, RemoteDisplayError, WindowInfo, + _canonical_rgb_digest, _split_chord, resolve_mac_key, ) +from openadapt_flow.remote_frame_contract import RemoteFrameContract from openadapt_flow.runtime.resolver import visual_resolution_point_fingerprint @@ -72,6 +74,7 @@ def __init__( self._key_window_id = key_window_id self._hit_window_id = hit_window_id self.frame_color = (11, 22, 33) + self.frame_overrides: dict[tuple[int, int], tuple[int, int, int]] = {} self.png_kwargs = {} self.calls: list[tuple] = [] @@ -99,6 +102,8 @@ def window_at_point(self, x, y): def capture(self, window_id): img = Image.new("RGB", self.px, self.frame_color) + for point, color in self.frame_overrides.items(): + img.putpixel(point, color) buf = io.BytesIO() img.save(buf, format="PNG", **self.png_kwargs) return buf.getvalue(), self.px[0], self.px[1] @@ -674,6 +679,69 @@ def test_bound_actuation_refuses_same_window_content_change_before_input() -> No assert raised.value.frame_size == client.px +def _remote_frame_contract_backend() -> tuple[RemoteDisplayBackend, FakeClient]: + size = (300, 200) + client = FakeClient( + window=WindowInfo( + window_id=1, + owner="Parallels Desktop", + title="Windows 11", + pid=99, + bounds=(0.0, 0.0, float(size[0]), float(size[1])), + on_screen=True, + ), + px=size, + ) + contract = RemoteFrameContract( + frame_width=size[0], + frame_height=size[1], + volatile_regions=((0, 0, 32, 32),), + protected_regions=((90, 90, 20, 20),), + ) + backend = RemoteDisplayBackend( + client=client, + settle_s=0.0, + readiness_probe=lambda _png: True, + remote_frame_contract=contract, + ) + backend.screenshot() + backend.prepare_pointer_actuation(100, 100) + backend.acquire_actuation_frame() + backend.arm_remote_frame_contract(protected_regions=((90, 90, 20, 20),)) + return backend, client + + +def test_bound_actuation_allows_only_qualified_remote_display_volatile_change() -> None: + backend, client = _remote_frame_contract_backend() + leased_raw_digest = backend._last_frame_digest + client.frame_overrides[(1, 1)] = (0, 0, 0) + current_png, _, _ = client.capture(client.window.window_id) + + assert leased_raw_digest is not None + assert _canonical_rgb_digest(current_png) != leased_raw_digest + + backend.click(100, 100) + + assert len([call for call in client.calls if call[0] == "mouse"]) == 2 + assert backend._last_frame_digest == leased_raw_digest + + +def test_bound_actuation_refuses_remote_display_change_outside_volatile_region() -> ( + None +): + backend, client = _remote_frame_contract_backend() + client.frame_overrides[(50, 50)] = (0, 0, 0) + + with pytest.raises(FreshActuationRequired) as raised: + backend.click(100, 100) + + assert not any(call[0] == "mouse" for call in client.calls) + assert raised.value.operation == "remote_click" + assert raised.value.changed_pixel_count == 1 + assert raised.value.changed_bbox == (50, 50, 1, 1) + assert raised.value.frame_size == client.px + + def test_replayer_reacquires_real_remote_display_lease_after_zero_edge_change( tmp_path, ) -> None: diff --git a/tests/test_remote_frame_contract.py b/tests/test_remote_frame_contract.py new file mode 100644 index 00000000..3564d6c6 --- /dev/null +++ b/tests/test_remote_frame_contract.py @@ -0,0 +1,67 @@ +from __future__ import annotations + +import hashlib +import io + +import pytest +from PIL import Image, ImageDraw + +from openadapt_flow.remote_frame_contract import RemoteFrameContract + + +def _png(clock: int, *, target: int = 0, identity: int = 0) -> bytes: + image = Image.new("RGB", (100, 80), "white") + draw = ImageDraw.Draw(image) + draw.rectangle((80, 0, 99, 15), fill=(clock, 0, 0)) + draw.text((80, 0), str(clock), fill="white") + draw.rectangle((0, 20, 40, 60), fill=(target, 0, 0)) + draw.rectangle((45, 20, 75, 60), fill=(0, identity, 0)) + output = io.BytesIO() + image.save(output, format="PNG") + return output.getvalue() + + +def _contract() -> RemoteFrameContract: + return RemoteFrameContract( + frame_width=100, + frame_height=80, + volatile_regions=((80, 0, 20, 16),), + protected_regions=((0, 20, 40, 40), (45, 20, 30, 40)), + ) + + +def test_clock_only_change_matches_only_in_derived_contract_input() -> None: + contract = _contract() + first, second = _png(1), _png(2) + assert hashlib.sha256(first).digest() != hashlib.sha256(second).digest() + assert contract.comparison_digest(first) == contract.comparison_digest(second) + + +@pytest.mark.parametrize("field", ["target", "identity"]) +def test_protected_changes_do_not_match(field: str) -> None: + kwargs = {field: 200} + assert _contract().comparison_digest(_png(1)) != _contract().comparison_digest( + _png(2, **kwargs) + ) + + +def test_overlap_and_geometry_mismatch_fail_closed() -> None: + with pytest.raises(ValueError, match="overlaps"): + RemoteFrameContract( + frame_width=100, + frame_height=80, + volatile_regions=((0, 0, 10, 10),), + protected_regions=((0, 0, 10, 10),), + ) + with pytest.raises(ValueError, match="geometry"): + _contract().require_geometry((99, 80)) + with pytest.raises(ValueError, match="schema_version"): + RemoteFrameContract.model_validate( + {**_contract().model_dump(), "schema_version": "unsupported/v1"} + ) + + +def test_runtime_target_or_identity_overlap_refuses_after_static_review() -> None: + contract = _contract() + with pytest.raises(ValueError, match="runtime protected"): + contract.arm(((80, 0, 10, 10),)) diff --git a/tests/test_replayer.py b/tests/test_replayer.py index 0b6771be..609e3c50 100644 --- a/tests/test_replayer.py +++ b/tests/test_replayer.py @@ -462,6 +462,69 @@ def click(self, x, y, *, double=False): ) +class PixelOnlyRemoteBackend: + """Opaque remote surface exposing ONLY the two-phase actuation lease. + + The exact protocol surface of the no-DOM HTML5-canvas backend in + ``benchmark/canvas_ladder``: pixels in, coordinates out, no structural + tree, no identity seam, and no typed delivery receipt. It implements + :class:`RemoteActuationBackend` and + :class:`PreparedPointerActuationBackend` and nothing else, so the exact + frame is bound by the backend's own one-shot lease, which its next input + method consumes and validates before the first input edge. + """ + + def __init__(self, *, frame=None, viewport=VIEWPORT): + self._frame = frame if frame is not None else make_png(viewport) + self._viewport = viewport + self.actions: list = [] + self.prepared_pointer_points: list = [] + self.acquire_count = 0 + self._leased_frame_sha256 = None + self.frame_after_lease = None + + @property + def viewport(self): + return self._viewport + + def screenshot(self): + return self._frame + + def prepare_pointer_actuation(self, x, y): + self._leased_frame_sha256 = None + self.prepared_pointer_points.append((int(x), int(y))) + + def acquire_actuation_frame(self) -> bytes: + self.acquire_count += 1 + self._leased_frame_sha256 = hashlib.sha256(self._frame).hexdigest() + if self.frame_after_lease is not None: + self._frame = self.frame_after_lease + return self._frame + + def _consume_lease(self): + leased = self._leased_frame_sha256 + self._leased_frame_sha256 = None + if leased is None: + return + if hashlib.sha256(self._frame).hexdigest() != leased: + raise RuntimeError("remote frame content changed before the input edge") + + def click(self, x, y, *, double=False): + self._consume_lease() + self.actions.append(("click", x, y, double)) + + def type_text(self, text): + self._consume_lease() + self.actions.append(("type", text)) + + def press(self, key): + self._consume_lease() + self.actions.append(("press", key)) + + def scroll(self, dx, dy): + self.actions.append(("scroll", dx, dy)) + + def click_step( step_id="s1", *, @@ -639,6 +702,94 @@ def test_consequential_remote_click_re_resolves_on_fresh_frame(bundle, run_dir): assert report.results[0].resolution.point == (110, 105) +def test_consequential_click_uses_lease_when_backend_has_no_typed_receipt( + bundle, run_dir +): + """A pixel-only opaque remote surface must actuate, not over-halt. + + ``GuardedRemotePointerActionBackend`` adds an explicit expected-frame hash + and a typed receipt; a plain ``RemoteActuationBackend`` already refuses + before the first input edge when the leased frame changed. Refusing the + latter halts every no-DOM canvas/VDI workflow on its write step. + """ + backend = PixelOnlyRemoteBackend() + vision = FakeVision() + vision.template_results = [ + Match(point=(110, 105), region=(100, 100, 50, 20), confidence=0.95), + Match(point=(110, 105), region=(100, 100, 50, 20), confidence=0.95), + ] + + report = Replayer(backend, vision=vision).run( + Workflow(name="wf", steps=[click_step(risk="irreversible")]), + bundle_dir=bundle, + run_dir=run_dir, + ) + + assert report.success is True + assert backend.prepared_pointer_points == [(110, 105)] + assert backend.acquire_count == 1 + assert backend.actions == [("click", 110, 105, False)] + assert report.rung_counts == {"template": 1} + assert report.model_calls == 0 + # No typed receipt exists, so the result must not claim a closed + # production actuation path. + assert report.results[0].actuation is None + assert report.results[0].delivery_receipt is None + + +def test_consequential_lease_click_still_refuses_a_changed_frame(bundle, run_dir): + """The lease is the safety property: a changed frame must stop delivery.""" + backend = PixelOnlyRemoteBackend() + backend.frame_after_lease = make_png(color=(10, 20, 30)) + vision = FakeVision() + vision.template_results = [ + Match(point=(110, 105), region=(100, 100, 50, 20), confidence=0.95), + Match(point=(110, 105), region=(100, 100, 50, 20), confidence=0.95), + ] + + report = Replayer(backend, vision=vision).run( + Workflow(name="wf", steps=[click_step(risk="irreversible")]), + bundle_dir=bundle, + run_dir=run_dir, + ) + + assert report.success is False + assert backend.actions == [] + + +def test_governed_consequential_click_requires_a_typed_remote_receipt(bundle, run_dir): + """A governed run still refuses a remote click it cannot evidence.""" + backend = PixelOnlyRemoteBackend() + vision = FakeVision() + vision.template_results = [ + Match(point=(110, 105), region=(100, 100, 50, 20), confidence=0.95), + Match(point=(110, 105), region=(100, 100, 50, 20), confidence=0.95), + ] + workflow = Workflow(name="wf", steps=[click_step(risk="irreversible")]) + workflow.save(bundle) + workflow = Workflow.load(bundle) + assert workflow.manifest is not None + authorization = GovernedRunAuthorization( + bundle_content_digest=workflow.manifest.content_digest, + runtime_inputs_digest=runtime_inputs_digest(workflow, None, None), + admitted_policy_name="test", + ) + + report = Replayer( + backend, + vision=vision, + governed_authorization=authorization, + ).run( + workflow, + bundle_dir=bundle, + run_dir=run_dir, + ) + + assert report.success is False + assert backend.actions == [] + assert "cannot bind its exact" in (report.results[0].error or "") + + def test_consequential_remote_hover_target_movement_halts_before_input(bundle, run_dir): backend = RemoteLeaseBackend( initial_frame=make_png(color=(240, 240, 240)), diff --git a/uv.lock b/uv.lock index 0f473ab2..a927ac78 100644 --- a/uv.lock +++ b/uv.lock @@ -2135,7 +2135,7 @@ wheels = [ [[package]] name = "openadapt-flow" -version = "1.27.0" +version = "1.27.1" source = { editable = "." } dependencies = [ { name = "cryptography" },