diff --git a/benchmark/rdp_ladder/render_presentation.py b/benchmark/rdp_ladder/render_presentation.py index c08df608..411c3d6d 100644 --- a/benchmark/rdp_ladder/render_presentation.py +++ b/benchmark/rdp_ladder/render_presentation.py @@ -17,9 +17,11 @@ from __future__ import annotations import argparse +import base64 import hashlib import json import math +import os import shutil import subprocess from pathlib import Path @@ -54,6 +56,13 @@ "outcome", } ) +PUBLICATION_APPROVAL_SCHEMA = "openadapt.rdp-publication-approval.v2" +PUBLICATION_APPROVAL_SCOPE = "openadapt.rdp-publication.finalize.v1" +CANDIDATE_VIDEO_NAME = "openadapt-rdp-demo.mp4" + + +def _sha256(path: Path) -> str: + return hashlib.sha256(path.read_bytes()).hexdigest() class HybridTimeline: @@ -233,11 +242,11 @@ def validate_hybrid_timeline( if source_frame is not None: if not isinstance(source_frame, dict): raise RuntimeError("hybrid timeline source frame is invalid") - presentation_phase = source_frame.get("presentation_phase") + source_phase = source_frame.get("presentation_phase") file = source_frame.get("file") - if not isinstance(presentation_phase, str) or not isinstance(file, str): + if not isinstance(source_phase, str) or not isinstance(file, str): raise RuntimeError("hybrid timeline source frame is incomplete") - source = source_frames.get((presentation_phase, file)) + source = source_frames.get((source_phase, file)) if source is None: raise RuntimeError("hybrid timeline source frame is not retained") for key, source_key in ( @@ -940,6 +949,140 @@ def render(presentation_dir: Path, output: Path) -> dict: return render_manifest +def _candidate_paths(candidate_dir: Path) -> tuple[Path, Path, Path]: + """Return the only files that a public RDP presentation may contain.""" + video = candidate_dir / CANDIDATE_VIDEO_NAME + return ( + video, + video.with_suffix(".timeline.json"), + video.with_suffix(".manifest.json"), + ) + + +def _candidate_inventory(candidate_dir: Path) -> tuple[Path, Path, Path]: + """Refuse a candidate directory with any unsigned extra file or directory.""" + if candidate_dir.is_symlink() or not candidate_dir.is_dir(): + raise RuntimeError( + "public artifact candidate directory is not a real directory" + ) + video, timeline, manifest = _candidate_paths(candidate_dir) + expected = {video.name, timeline.name, manifest.name} + actual = {path.name for path in candidate_dir.iterdir()} + if actual != expected or not all( + path.is_file() and not path.is_symlink() for path in (video, timeline, manifest) + ): + raise RuntimeError("public artifact candidate inventory is not exact") + return video, timeline, manifest + + +def render_candidate(presentation_dir: Path, candidate_dir: Path) -> dict: + """Render an isolated review candidate. This function never publishes it.""" + if candidate_dir.exists(): + raise RuntimeError("public artifact candidate directory already exists") + candidate_dir.mkdir(parents=True) + video, _timeline, _manifest = _candidate_paths(candidate_dir) + return render(presentation_dir, video) + + +def approve_public_artifact_set( + candidate_dir: Path, + *, + approval_path: Path, + key_id: str, + private_key: bytes, +) -> Path: + """Create a detached approval for a reviewed exact candidate manifest.""" + from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey + + _video, _timeline, manifest = _candidate_inventory(candidate_dir) + unsigned = { + "schema_version": PUBLICATION_APPROVAL_SCHEMA, + "candidate_sha256": _sha256(manifest), + "signer_key_id": key_id, + "approval_scope": PUBLICATION_APPROVAL_SCOPE, + } + signature = Ed25519PrivateKey.from_private_bytes(private_key).sign( + json.dumps(unsigned, sort_keys=True, separators=(",", ":")).encode() + ) + approval_path.write_text( + json.dumps( + {**unsigned, "signature": base64.b64encode(signature).decode()}, + indent=2, + sort_keys=True, + ) + + "\n", + encoding="utf-8", + ) + return approval_path + + +def finalize_public_artifact_set( + presentation_dir: Path, + candidate_dir: Path, + output_dir: Path, + *, + approval_path: Path, + trusted_public_keys: dict[str, bytes], +) -> None: + """Verify one closed candidate directory, then publish it by one rename.""" + if candidate_dir.resolve() == output_dir.resolve(): + raise RuntimeError("candidate and public artifact directories must differ") + if approval_path.resolve().is_relative_to(candidate_dir.resolve()): + raise RuntimeError("publication approval must be detached from the candidate") + if output_dir.exists(): + raise RuntimeError("public artifact output directory already exists") + video, timeline_path, manifest_path = _candidate_inventory(candidate_dir) + approval = _load_json(approval_path) + required = { + "schema_version", + "candidate_sha256", + "signer_key_id", + "approval_scope", + "signature", + } + if ( + not isinstance(approval, dict) + or set(approval) != required + or approval.get("schema_version") != PUBLICATION_APPROVAL_SCHEMA + or approval.get("approval_scope") != PUBLICATION_APPROVAL_SCOPE + or approval.get("candidate_sha256") != _sha256(manifest_path) + or not isinstance(approval.get("signer_key_id"), str) + ): + raise RuntimeError("publication approval does not bind the exact candidate") + key = trusted_public_keys.get(approval["signer_key_id"]) + if key is None: + raise RuntimeError("publication approval signer is not trusted") + from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey + + unsigned = {key: value for key, value in approval.items() if key != "signature"} + try: + Ed25519PublicKey.from_public_bytes(key).verify( + base64.b64decode(approval["signature"], validate=True), + json.dumps(unsigned, sort_keys=True, separators=(",", ":")).encode(), + ) + except Exception as exc: + raise RuntimeError("publication approval signature is invalid") from exc + + manifest = _load_json(manifest_path) + timeline = _load_json(timeline_path) + if ( + not isinstance(manifest, dict) + or manifest.get("video") != video.name + or manifest.get("video_sha256") != _sha256(video) + or manifest.get("hybrid_timeline") != timeline_path.name + or manifest.get("hybrid_timeline_sha256") != _sha256(timeline_path) + ): + raise RuntimeError("public artifact candidate hashes do not match") + source_manifests = { + name: _load_json(presentation_dir / name / "manifest.json") + for name in PHASE_DIRS + } + graph = _load_json(presentation_dir / "02-compiled-workflow" / "program-graph.json") + validate_hybrid_timeline(timeline, manifests=source_manifests, graph=graph) + output_dir.parent.mkdir(parents=True, exist_ok=True) + os.replace(candidate_dir, output_dir) + + def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--presentation-dir", type=Path, required=True) diff --git a/openadapt_flow/runtime/replayer.py b/openadapt_flow/runtime/replayer.py index 513e64dd..0bc8d141 100644 --- a/openadapt_flow/runtime/replayer.py +++ b/openadapt_flow/runtime/replayer.py @@ -7543,6 +7543,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.""" @@ -8577,39 +8604,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 +8729,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 +8894,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/tests/test_rdp_ladder_qualification.py b/tests/test_rdp_ladder_qualification.py index a969d52c..73c08ad5 100644 --- a/tests/test_rdp_ladder_qualification.py +++ b/tests/test_rdp_ladder_qualification.py @@ -17,6 +17,8 @@ from pathlib import Path import pytest +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey from PIL import Image from openadapt_flow.ir import ( @@ -295,6 +297,96 @@ def test_presentation_is_hash_bound_and_renders_without_staged_video_frames( ] +def _signed_publication_candidate( + tmp_path: Path, +) -> tuple[Path, Path, dict[str, bytes]]: + candidate = tmp_path / "candidate" + candidate.mkdir() + video, timeline, manifest = renderer._candidate_paths(candidate) + video.write_bytes(b"video") + timeline.write_text("{}", encoding="utf-8") + manifest.write_text("{}", encoding="utf-8") + private_key = Ed25519PrivateKey.generate() + approval = renderer.approve_public_artifact_set( + candidate, + approval_path=tmp_path / "approval.json", + key_id="test-key", + private_key=private_key.private_bytes( + serialization.Encoding.Raw, + serialization.PrivateFormat.Raw, + serialization.NoEncryption(), + ), + ) + trusted_keys = { + "test-key": private_key.public_key().public_bytes( + serialization.Encoding.Raw, + serialization.PublicFormat.Raw, + ) + } + return candidate, approval, trusted_keys + + +def _finalize_signed_candidate( + tmp_path: Path, + candidate: Path, + approval: Path, + trusted_keys: dict[str, bytes], +) -> None: + renderer.finalize_public_artifact_set( + tmp_path / "presentation", + candidate, + tmp_path / "published", + approval_path=approval, + trusted_public_keys=trusted_keys, + ) + + +@pytest.mark.parametrize( + "artifact_name", + [ + renderer.CANDIDATE_VIDEO_NAME, + "openadapt-rdp-demo.timeline.json", + "openadapt-rdp-demo.manifest.json", + ], +) +def test_public_presentation_finalizer_refuses_symlinked_candidate_artifacts( + tmp_path: Path, artifact_name: str +) -> None: + candidate, approval, trusted_keys = _signed_publication_candidate(tmp_path) + external = tmp_path / "outside" + external.write_bytes(b"outside") + artifact = candidate / artifact_name + artifact.unlink() + artifact.symlink_to(external) + + with pytest.raises(RuntimeError, match="inventory is not exact"): + _finalize_signed_candidate(tmp_path, candidate, approval, trusted_keys) + assert not (tmp_path / "published").exists() + + +def test_public_presentation_finalizer_refuses_a_symlinked_candidate_directory( + tmp_path: Path, +) -> None: + candidate, approval, trusted_keys = _signed_publication_candidate(tmp_path) + candidate_link = tmp_path / "candidate-link" + candidate_link.symlink_to(candidate, target_is_directory=True) + + with pytest.raises(RuntimeError, match="not a real directory"): + _finalize_signed_candidate(tmp_path, candidate_link, approval, trusted_keys) + assert not (tmp_path / "published").exists() + + +def test_public_presentation_finalizer_refuses_an_extra_candidate_directory( + tmp_path: Path, +) -> None: + candidate, approval, trusted_keys = _signed_publication_candidate(tmp_path) + (candidate / "unreviewed").mkdir() + + with pytest.raises(RuntimeError, match="inventory is not exact"): + _finalize_signed_candidate(tmp_path, candidate, approval, trusted_keys) + assert not (tmp_path / "published").exists() + + @pytest.mark.parametrize( ("character", "keysym"), [("-", "minus"), ("/", "slash"), (":", "colon")], 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)),