From f6cd42fdd6761929decd2530a479f23c4e3e59ad Mon Sep 17 00:00:00 2001 From: abrichr Date: Mon, 31 Aug 2026 18:43:51 -0400 Subject: [PATCH 1/5] feat(desktop): openadapt://runner URI grammar, distinct from connect Parse-only runner deep links (pack, bind, origin) beside the existing connect action. Prefix parsers reject oar_/oap_ mix-ups, hex-bodied oab_ tokens, and base64url-bodied oals_ secrets. No claim, keychain, or mailbox poll. --- engine/auth/runner_bind.py | 123 +++++++++++++ src-tauri/src/pairing.rs | 196 ++++++++++++++++++--- tests/test_engine/test_auth_runner_bind.py | 107 +++++++++++ tests/test_pairing_protocol_boundary.py | 20 +++ 4 files changed, 425 insertions(+), 21 deletions(-) create mode 100644 engine/auth/runner_bind.py create mode 100644 tests/test_engine/test_auth_runner_bind.py diff --git a/engine/auth/runner_bind.py b/engine/auth/runner_bind.py new file mode 100644 index 0000000..8cfb0fc --- /dev/null +++ b/engine/auth/runner_bind.py @@ -0,0 +1,123 @@ +"""Parse-only grammar for ``openadapt://runner`` authoring bind URIs. + +Tauri validates the same fields first. Python parses again so neither IPC nor +an operating-system protocol invocation can become a general command. This +module does not claim, store, or poll. +""" + +from __future__ import annotations + +import re +from urllib.parse import parse_qs, urlparse, urlsplit + +AUTHORING_ORIGIN = "https://openadapt.ai" +MAX_URI_BYTES = 2048 +ALLOWED_FIELDS = frozenset({"pack", "bind", "origin"}) + +BIND_TOKEN_RE = re.compile(r"^oab_[A-Za-z0-9_-]{43}$") +LEASE_SECRET_RE = re.compile(r"^oals_[a-f0-9]{64}$") +PACK_ALIAS_RE = re.compile(r"^p\.[A-Za-z0-9_-]{12}$") +PACK_CIPHER_RE = re.compile(r"^v1\.[A-Za-z0-9_-]{32,2000}$") +CLOUD_RUNNER_TOKEN_RE = re.compile(r"^oar_[a-f0-9]{64}$") +PAIRING_SECRET_RE = re.compile(r"^oap_[A-Za-z0-9_-]{43}$") +HEX_BODY_RE = re.compile(r"^[a-f0-9]+$") +UNRESERVED_BODY_RE = re.compile(r"^[A-Za-z0-9_-]+$") + + +class RunnerBindError(RuntimeError): + """A safe, user-facing runner-link failure with no secret-bearing text.""" + + +def valid_bind_token(value: object) -> bool: + """Return whether ``value`` is exactly one ``oab_`` bind token.""" + + if not isinstance(value, str) or BIND_TOKEN_RE.fullmatch(value) is None: + return False + body = value[4:] + # 32-byte hex (the ``oar_`` body) is not a bind token even with this prefix. + if HEX_BODY_RE.fullmatch(body) is not None and len(body) == 64: + return False + return True + + +def valid_lease_secret(value: object) -> bool: + """Return whether ``value`` is exactly one ``oals_`` mailbox lease secret.""" + + if not isinstance(value, str) or LEASE_SECRET_RE.fullmatch(value) is None: + return False + body = value[5:] + # 32-byte base64url (the ``oab_`` body) is not a lease secret. + if len(body) == 43 and UNRESERVED_BODY_RE.fullmatch(body) is not None: + return False + return True + + +def valid_pack_id(value: object) -> bool: + """Return whether ``value`` is a ``p.`` alias or ``v1.`` ciphertext id.""" + + if not isinstance(value, str): + return False + return PACK_ALIAS_RE.fullmatch(value) is not None or PACK_CIPHER_RE.fullmatch(value) is not None + + +def canonical_authoring_origin(value: object) -> str: + """Return the pinned production authoring origin, or raise.""" + + if not isinstance(value, str): + raise RunnerBindError("Runner link does not name the OpenAdapt authoring origin") + try: + parsed = urlsplit(value) + port = parsed.port + except ValueError as exc: + raise RunnerBindError("Runner link does not name the OpenAdapt authoring origin") from exc + if ( + parsed.scheme != "https" + or parsed.hostname != "openadapt.ai" + or parsed.netloc != "openadapt.ai" + or parsed.username + or parsed.password + or parsed.path not in ("",) + or parsed.query + or parsed.fragment + or port is not None + or value != AUTHORING_ORIGIN + ): + raise RunnerBindError("Runner link does not name the OpenAdapt authoring origin") + return AUTHORING_ORIGIN + + +def parse_runner_uri(uri: object) -> dict[str, str]: + """Parse the fixed runner action and reject ambiguity or extra fields.""" + + if not isinstance(uri, str) or not uri or len(uri) > MAX_URI_BYTES: + raise RunnerBindError("Invalid OpenAdapt runner link") + parsed = urlparse(uri) + if ( + parsed.scheme != "openadapt" + or parsed.netloc != "runner" + or parsed.path not in ("", "/") + or parsed.params + or parsed.fragment + or parsed.username + or parsed.password + ): + raise RunnerBindError("Invalid OpenAdapt runner link") + try: + query = parse_qs(parsed.query, keep_blank_values=True, strict_parsing=True) + except ValueError as exc: + raise RunnerBindError("Invalid OpenAdapt runner link") from exc + if set(query) - ALLOWED_FIELDS or any(len(values) != 1 for values in query.values()): + raise RunnerBindError("Runner link contains unknown or duplicate fields") + if set(query) != ALLOWED_FIELDS: + raise RunnerBindError("Runner link is missing pack, bind, or origin") + + pack = query["pack"][0] + bind = query["bind"][0] + origin = canonical_authoring_origin(query["origin"][0]) + if not valid_pack_id(pack): + raise RunnerBindError("Pack id is malformed") + if CLOUD_RUNNER_TOKEN_RE.fullmatch(bind) or PAIRING_SECRET_RE.fullmatch(bind): + raise RunnerBindError("Bind token is malformed") + if not valid_bind_token(bind): + raise RunnerBindError("Bind token is malformed") + return {"pack": pack, "bind": bind, "origin": origin} diff --git a/src-tauri/src/pairing.rs b/src-tauri/src/pairing.rs index 12d22fe..d9c9b41 100644 --- a/src-tauri/src/pairing.rs +++ b/src-tauri/src/pairing.rs @@ -1,9 +1,14 @@ -//! Strict operating-system deep-link boundary for one-click Cloud pairing. +//! Strict operating-system deep-link boundary for Cloud pairing and authoring. //! //! The protocol handler never opens a URL or constructs a process command. It -//! accepts one fixed `openadapt://connect` URI, validates every field, and -//! forwards the original URI as one JSON string to the fixed Python -//! `connect_uri` sidecar action. +//! accepts two fixed schemes, validates every field, and forwards the original +//! URI as one JSON string to one sidecar action: +//! +//! - `openadapt://connect` → `connect_uri` +//! - `openadapt://runner` → `claim_runner_uri` +//! +//! Connect is not widened to accept runner fields, and runner is not widened +//! to accept connect fields. use std::collections::{HashMap, HashSet}; use std::error::Error; @@ -18,6 +23,7 @@ use url::Url; use crate::sidecar::SidecarInner; const MANAGED_HOST: &str = "app.openadapt.ai"; +const AUTHORING_ORIGIN: &str = "https://openadapt.ai"; const MAX_URI_BYTES: usize = 2048; const MAX_RECENT_LINKS: usize = 64; @@ -72,7 +78,12 @@ fn route_urls( let action = match single_action(urls) { Ok(action) => action, Err(error) => { - emit_state(&app, "error", Some(error)); + let event = if urls.iter().any(|url| url.host_str() == Some("runner")) { + "engine://authoring_state" + } else { + "engine://pairing_state" + }; + emit_state(&app, event, "error", Some(error)); return; } }; @@ -92,32 +103,48 @@ fn route_urls( handled.insert(fingerprint); } - emit_state(&app, "connecting", None); + let event = status_event(action.command); + let connecting = if action.command == "claim_runner_uri" { + "claiming" + } else { + "connecting" + }; + emit_state(&app, event, connecting, None); tauri::async_runtime::spawn(async move { let result = engine .send_command(action.command, json!({ "uri": action.uri })) .await; match result { Ok(data) => { - let _ = app.emit( - "engine://pairing_state", - json!({ "status": "connected", "data": data }), - ); + let connected = if action.command == "claim_runner_uri" { + "bound" + } else { + "connected" + }; + let _ = app.emit(event, json!({ "status": connected, "data": data })); } Err(error) => { eprintln!("[pairing] connection failed: {error}"); - emit_state(&app, "error", Some(&error)); + emit_state(&app, event, "error", Some(&error)); } } }); } -fn emit_state(app: &AppHandle, status: &str, error: Option<&str>) { +fn status_event(command: &str) -> &'static str { + if command == "claim_runner_uri" { + "engine://authoring_state" + } else { + "engine://pairing_state" + } +} + +fn emit_state(app: &AppHandle, event: &str, status: &str, error: Option<&str>) { let payload = match error { Some(error) => json!({ "status": status, "error": error }), None => json!({ "status": status }), }; - let _ = app.emit("engine://pairing_state", payload); + let _ = app.emit(event, payload); } fn fingerprint(uri: &str) -> u64 { @@ -135,18 +162,31 @@ fn single_action(urls: &[Url]) -> Result { fn action_for_url(url: &Url) -> Result { let uri = url.as_str(); + let runner = url.host_str() == Some("runner"); + let invalid = if runner { + "Invalid OpenAdapt runner link" + } else { + "Invalid OpenAdapt connect link" + }; if uri.len() > MAX_URI_BYTES || url.scheme() != "openadapt" - || url.host_str() != Some("connect") || !url.username().is_empty() || url.password().is_some() || url.port().is_some() || !matches!(url.path(), "" | "/") || url.fragment().is_some() { - return Err("Invalid OpenAdapt connect link"); + return Err(invalid); + } + + match url.host_str() { + Some("connect") => connect_action(url, uri), + Some("runner") => runner_action(url, uri), + _ => Err("Invalid OpenAdapt connect link"), } +} +fn connect_action(url: &Url, uri: &str) -> Result { let mut fields: HashMap = HashMap::new(); for (key, value) in url.query_pairs() { if !matches!(key.as_ref(), "pairing" | "host" | "destination_kind") @@ -175,12 +215,65 @@ fn action_for_url(url: &Url) -> Result { }) } +fn runner_action(url: &Url, uri: &str) -> Result { + let mut fields: HashMap = HashMap::new(); + for (key, value) in url.query_pairs() { + if !matches!(key.as_ref(), "pack" | "bind" | "origin") + || fields + .insert(key.into_owned(), value.into_owned()) + .is_some() + { + return Err("Runner link contains unknown or duplicate fields"); + } + } + + let pack = fields + .get("pack") + .ok_or("Runner link is missing pack, bind, or origin")?; + let bind = fields + .get("bind") + .ok_or("Runner link is missing pack, bind, or origin")?; + let origin = fields + .get("origin") + .ok_or("Runner link is missing pack, bind, or origin")?; + if !valid_pack_id(pack) { + return Err("Pack id is malformed"); + } + if !valid_bind_token(bind) { + return Err("Bind token is malformed"); + } + if origin != AUTHORING_ORIGIN { + return Err("Runner link does not name the OpenAdapt authoring origin"); + } + + Ok(PairingAction { + command: "claim_runner_uri", + uri: uri.to_owned(), + }) +} + fn valid_pairing_secret(value: &str) -> bool { - value.len() == 47 - && value.starts_with("oap_") - && value[4..] - .bytes() - .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'_' | b'-')) + value.len() == 47 && value.starts_with("oap_") && unreserved_body(&value[4..]) +} + +fn valid_bind_token(value: &str) -> bool { + value.len() == 47 && value.starts_with("oab_") && unreserved_body(&value[4..]) +} + +fn valid_pack_id(value: &str) -> bool { + if let Some(body) = value.strip_prefix("p.") { + return body.len() == 12 && unreserved_body(body); + } + if let Some(body) = value.strip_prefix("v1.") { + return (32..=2000).contains(&body.len()) && unreserved_body(body); + } + false +} + +fn unreserved_body(value: &str) -> bool { + value + .bytes() + .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'_' | b'-')) } fn validate_destination(host: &str, destination_kind: Option<&str>) -> Result<(), &'static str> { @@ -230,6 +323,13 @@ mod tests { Url::parse(raw).unwrap() } + const BIND: &str = "oab_AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"; + const PACK: &str = "p.abcdefghijkl"; + + fn runner_uri() -> String { + format!("openadapt://runner?pack={PACK}&bind={BIND}&origin=https%3A%2F%2Fopenadapt.ai") + } + #[test] fn accepts_only_fixed_connect_action() { let url = parse(&format!( @@ -244,8 +344,30 @@ mod tests { format!("https://connect?pairing={SECRET}&host=https://app.openadapt.ai"), format!("openadapt://connect/run?pairing={SECRET}&host=https://app.openadapt.ai"), format!("openadapt://connect?pairing={SECRET}&host=https://app.openadapt.ai#x"), + runner_uri(), + format!("openadapt://connect?pack={PACK}&bind={BIND}&origin=https://openadapt.ai"), + ] { + assert!(action_for_url(&parse(&raw)).is_err(), "{raw}"); + } + } + + #[test] + fn accepts_only_fixed_runner_action() { + let url = parse(&runner_uri()); + let action = action_for_url(&url).unwrap(); + assert_eq!(action.command, "claim_runner_uri"); + assert_eq!(action.uri, url.as_str()); + + for raw in [ + format!("openadapt://run?pack={PACK}&bind={BIND}&origin=https://openadapt.ai"), + format!( + "openadapt://connect/runner?pack={PACK}&bind={BIND}&origin=https://openadapt.ai" + ), + format!("{}#x", runner_uri()), + format!("openadapt://connect?pairing={SECRET}&host=https://app.openadapt.ai"), + format!("openadapt://runner?pairing={SECRET}&host=https://app.openadapt.ai"), ] { - assert!(action_for_url(&parse(&raw)).is_err()); + assert!(action_for_url(&parse(&raw)).is_err(), "{raw}"); } } @@ -265,6 +387,38 @@ mod tests { } } + #[test] + fn runner_rejects_malformed_duplicate_unknown_and_foreign_tokens() { + for raw in [ + format!("openadapt://runner?pack=short&bind={BIND}&origin=https://openadapt.ai"), + format!("openadapt://runner?pack={PACK}&bind={BIND}"), + format!( + "openadapt://runner?pack={PACK}&bind={BIND}&bind={BIND}&origin=https://openadapt.ai" + ), + format!( + "openadapt://runner?pack={PACK}&bind={BIND}&origin=https://openadapt.ai&command=run" + ), + format!( + "openadapt://runner?pack={PACK}&bind=oar_{}&origin=https://openadapt.ai", + "a".repeat(64) + ), + format!("openadapt://runner?pack={PACK}&bind={SECRET}&origin=https://openadapt.ai"), + format!( + "openadapt://runner?pack={PACK}&bind=oab_{}&origin=https://openadapt.ai", + "a".repeat(64) + ), + format!( + "openadapt://runner?pack={PACK}&bind=oals_{}&origin=https://openadapt.ai", + "A".repeat(43) + ), + format!( + "openadapt://runner?pack={PACK}&bind={BIND}&origin=https://preview.openadapt.ai" + ), + ] { + assert!(action_for_url(&parse(&raw)).is_err(), "{raw}"); + } + } + #[test] fn argument_shaped_data_never_changes_the_fixed_action() { let encoded_argument = "%2D%2Dhost%3Dhttps%3A%2F%2Fevil.example"; diff --git a/tests/test_engine/test_auth_runner_bind.py b/tests/test_engine/test_auth_runner_bind.py new file mode 100644 index 0000000..0e0080d --- /dev/null +++ b/tests/test_engine/test_auth_runner_bind.py @@ -0,0 +1,107 @@ +"""Parse-only tests for ``openadapt://runner`` bind grammar.""" + +from __future__ import annotations + +import pytest + +from engine.auth import pairing +from engine.auth.runner_bind import ( + RunnerBindError, + parse_runner_uri, + valid_bind_token, + valid_lease_secret, + valid_pack_id, +) + +BIND = "oab_" + "A" * 43 +PACK = "p.abcdefghijkl" +ORIGIN = "https://openadapt.ai" +VALID_URI = f"openadapt://runner?pack={PACK}&bind={BIND}&origin=https%3A%2F%2Fopenadapt.ai" +CONNECT_SECRET = "oap_" + "A" * 43 +CONNECT_URI = ( + f"openadapt://connect?pairing={CONNECT_SECRET}&host=https%3A%2F%2Fapp.openadapt.ai" +) + + +def test_parser_accepts_only_the_fixed_runner_action() -> None: + assert parse_runner_uri(VALID_URI) == { + "pack": PACK, + "bind": BIND, + "origin": ORIGIN, + } + for uri in ( + VALID_URI.replace("://runner?", "://run?"), + VALID_URI.replace("://runner?", "://connect?"), + VALID_URI.replace("openadapt:", "https:"), + VALID_URI.replace("runner?", "runner/claim?"), + VALID_URI + "#fragment", + f"openadapt://user@runner?pack={PACK}&bind={BIND}&origin={ORIGIN}", + CONNECT_URI, + ): + with pytest.raises(RunnerBindError, match="Invalid OpenAdapt runner link"): + parse_runner_uri(uri) + + +def test_connect_parser_rejects_runner_shapes() -> None: + with pytest.raises(pairing.PairingError, match="Invalid OpenAdapt connect link"): + pairing.parse_connect_uri(VALID_URI) + with pytest.raises(pairing.PairingError): + pairing.parse_connect_uri( + f"openadapt://connect?pack={PACK}&bind={BIND}&origin={ORIGIN}" + ) + + +def test_parser_rejects_malformed_missing_duplicate_and_unknown_fields() -> None: + bad = ( + "", + "openadapt://runner?pack", + f"openadapt://runner?pack=short&bind={BIND}&origin={ORIGIN}", + f"openadapt://runner?pack={PACK}&bind={BIND}", + f"openadapt://runner?pack={PACK}&bind={BIND}&bind={BIND}&origin={ORIGIN}", + f"openadapt://runner?pack={PACK}&bind={BIND}&origin={ORIGIN}&command=whoami", + f"openadapt://runner?pack={PACK}&bind={BIND}&origin=https://preview.openadapt.ai", + f"openadapt://runner?pack={PACK}&bind={BIND}&origin=https://openadapt.ai/", + f"openadapt://runner?pack={PACK}&bind={BIND}&origin=https://openadapt.ai:443", + f"openadapt://runner?pack={PACK}&bind={BIND}&origin=http://openadapt.ai", + ) + for uri in bad: + with pytest.raises(RunnerBindError): + parse_runner_uri(uri) + + +def test_prefix_parsers_reject_foreign_and_swapped_encodings() -> None: + oar = "oar_" + "a" * 64 + oap = "oap_" + "A" * 43 + oab_hex = "oab_" + "a" * 64 + oals_b64 = "oals_" + "A" * 43 + oa_prefix = "oa" + "A" * 43 + oab = BIND + oals = "oals_" + "a" * 64 + + assert valid_bind_token(oab) is True + assert valid_lease_secret(oals) is True + assert valid_pack_id(PACK) is True + assert valid_pack_id("v1." + "A" * 48) is True + + for value in (oar, oap, oab_hex, oals_b64, oals, oa_prefix, PACK): + assert valid_bind_token(value) is False, value + for value in (oar, oap, oab, oab_hex, oals_b64, oa_prefix, PACK): + assert valid_lease_secret(value) is False, value + for value in (oar, oap, oab, oals, oa_prefix, "p.short", "v1.short"): + assert valid_pack_id(value) is False, value + + for bind in (oar, oap, oab_hex, oals_b64, oa_prefix): + uri = f"openadapt://runner?pack={PACK}&bind={bind}&origin={ORIGIN}" + with pytest.raises(RunnerBindError, match="Bind token is malformed"): + parse_runner_uri(uri) + + +def test_argument_shaped_values_remain_data_and_cannot_select_an_action() -> None: + for payload in ( + "--origin=https://evil.example", + "%2D%2Dorigin%3Dhttps%3A%2F%2Fevil.example", + "oab_" + "A" * 42 + ";", + ): + uri = f"openadapt://runner?pack={PACK}&bind={payload}&origin={ORIGIN}" + with pytest.raises(RunnerBindError, match="Bind token is malformed"): + parse_runner_uri(uri) diff --git a/tests/test_pairing_protocol_boundary.py b/tests/test_pairing_protocol_boundary.py index c30736a..0feee65 100644 --- a/tests/test_pairing_protocol_boundary.py +++ b/tests/test_pairing_protocol_boundary.py @@ -21,10 +21,14 @@ def test_single_instance_precedes_deep_link_and_handoff_is_fixed() -> None: deep_link = main.index(".plugin(tauri_plugin_deep_link::init())") assert single < deep_link assert 'command: "connect_uri"' in pairing + assert 'command: "claim_runner_uri"' in pairing assert 'json!({ "uri": action.uri })' in pairing assert "std::process::Command" not in pairing assert "open_external" not in pairing assert "ShellExt" not in pairing + assert 'Some("connect") => connect_action' in pairing + assert 'Some("runner") => runner_action' in pairing + assert "pack" not in pairing.split("fn connect_action", 1)[1].split("fn runner_action", 1)[0] def test_python_pairing_action_has_no_shell_or_navigation_escape_hatch() -> None: @@ -43,3 +47,19 @@ def test_python_pairing_action_has_no_shell_or_navigation_escape_hatch() -> None assert "write_text" not in pairing + store assert "open(" not in pairing + store assert "logger." not in pairing + + +def test_python_runner_bind_is_parse_only() -> None: + runner_bind = (ROOT / "engine/auth/runner_bind.py").read_text() + assert "def parse_runner_uri" in runner_bind + assert "httpx" not in runner_bind + assert "keyring" not in runner_bind + assert "subprocess" not in runner_bind + assert "webbrowser" not in runner_bind + assert "os.system" not in runner_bind + assert "wait_seconds" not in runner_bind + assert "DEFAULT_WAIT_S" not in runner_bind + assert "oar_" in runner_bind + assert "oap_" in runner_bind + assert "oab_" in runner_bind + assert "oals_" in runner_bind From 48675aac418a8b20cbd1bbe1e6f77aa5000862a0 Mon Sep 17 00:00:00 2001 From: abrichr Date: Mon, 31 Aug 2026 18:54:08 -0400 Subject: [PATCH 2/5] feat(desktop): authoring claim, Allow-per-sub, mailbox poll, record_observed Outbound HTTPS claim of oab_ binds, keychain oals_ leases, wait=0 poll with a local 1s sleep, and Allow of a specific connector sub. Overlay Continue persists pause-target input via record_observed, never type_text. Windows native is COACH_ONLY; do not spawn win_agent. --- engine/auth/store.py | 83 ++ engine/authoring_runner.py | 1050 ++++++++++++++++++++ engine/dispatch.py | 56 +- src/App.tsx | 75 ++ src/lib/engine.ts | 6 + src/styles/app.css | 13 + tests/test_engine/test_authoring_runner.py | 457 +++++++++ tests/test_pairing_protocol_boundary.py | 1 + 8 files changed, 1740 insertions(+), 1 deletion(-) create mode 100644 engine/authoring_runner.py create mode 100644 tests/test_engine/test_authoring_runner.py diff --git a/engine/auth/store.py b/engine/auth/store.py index 1f9d2ce..ebfa9ee 100644 --- a/engine/auth/store.py +++ b/engine/auth/store.py @@ -815,6 +815,89 @@ def clear_runner_credential(host: str) -> None: _kr_delete(_keyring(), host + _RUNNER_SUFFIX) +_AUTHORING_LEASE_PREFIX = "openadapt-authoring-lease|" +_AUTHORING_LEASE_KEYS = frozenset( + { + "pack", + "origin", + "lease_secret", + "lease_s", + "claimed_at", + "allowed_sub", + "allowed_client_id", + "allowed_at", + } +) +_SHA256_HEX_VALUE = re.compile(r"^[a-f0-9]{64}$") + + +def _authoring_lease_account(pack_id: str) -> str: + from engine.auth.runner_bind import valid_pack_id + + if not valid_pack_id(pack_id): + raise ValueError("pack id is malformed") + return _AUTHORING_LEASE_PREFIX + pack_id + + +def store_authoring_lease(pack_id: str, payload: dict) -> bool: + """Persist one authoring mailbox lease in the OS keychain.""" + + from engine.auth.runner_bind import ( + AUTHORING_ORIGIN, + valid_lease_secret, + valid_pack_id, + ) + + if ( + not isinstance(payload, dict) + or set(payload) != _AUTHORING_LEASE_KEYS + or not valid_pack_id(payload.get("pack")) + or payload.get("pack") != pack_id + or payload.get("origin") != AUTHORING_ORIGIN + or not valid_lease_secret(payload.get("lease_secret")) + or not isinstance(payload.get("lease_s"), int) + or isinstance(payload.get("lease_s"), bool) + or payload.get("lease_s") <= 0 + or not isinstance(payload.get("claimed_at"), str) + or payload.get("claimed_at") == "" + ): + return False + for key in ("allowed_sub", "allowed_client_id", "allowed_at"): + value = payload.get(key) + if value is None: + continue + if key == "allowed_at" and isinstance(value, str) and value: + continue + if key != "allowed_at" and isinstance(value, str) and _SHA256_HEX_VALUE.fullmatch(value): + continue + return False + account = _authoring_lease_account(pack_id) + encoded = json.dumps(payload, sort_keys=True, separators=(",", ":")) + return _apply_exact(_keyring(), account, encoded) + + +def load_authoring_lease(pack_id: str) -> dict | None: + """Load the authoring mailbox lease for ``pack_id``, or None.""" + + account = _authoring_lease_account(pack_id) + readable, raw = _strict_get(_keyring(), account) + if not readable or raw is None: + return None + try: + payload = json.loads(raw) + except (json.JSONDecodeError, TypeError): + return None + if not isinstance(payload, dict) or set(payload) != _AUTHORING_LEASE_KEYS: + return None + return payload + + +def clear_authoring_lease(pack_id: str) -> None: + """Delete the authoring mailbox lease for ``pack_id``.""" + + _kr_delete(_keyring(), _authoring_lease_account(pack_id)) + + def canonical_host_origin(host: str) -> str: """Return a safe web origin for credential binding, or ``""``. diff --git a/engine/authoring_runner.py b/engine/authoring_runner.py new file mode 100644 index 0000000..a5d6690 --- /dev/null +++ b/engine/authoring_runner.py @@ -0,0 +1,1050 @@ +"""Outbound authoring mailbox client for ChatGPT.com / Claude.ai drive-once. + +Copy poll / lease / TTL / kill-as-command / metadata-callback *shape* from +:mod:`engine.hosted_runner`. Do not copy org, Stripe, ``oar_``, a 25s poll wait, +trust-manifest, or journal dispatch. Windows native is COACH_ONLY: this module +must not spawn ``win_agent`` or call ``parallels_vm.launch_agent``. +""" + +from __future__ import annotations + +import hashlib +import hmac +import json +import os +import re +import stat +import threading +import time +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Callable +from urllib.parse import quote + +import httpx +from loguru import logger + +from engine.auth.runner_bind import ( + AUTHORING_ORIGIN, + parse_runner_uri, + valid_lease_secret, + valid_pack_id, +) +from engine.auth.store import load_authoring_lease, store_authoring_lease +from engine.config import EngineConfig + +API_TIMEOUT_S = 10.0 +DEFAULT_LEASE_S = 900 +POLL_WAIT_S = 0 +LOCAL_POLL_SLEEP_S = 1.0 +NODE_TABLE_LIFETIME_S = 15 * 60 +COMMAND_ENVELOPE_SCHEMA = "openadapt.authoring.command/v1" +OBSERVE_SCHEMA = "openadapt.authoring.observe/v1" +CLIENT_DISPLAYS = frozenset({"ChatGPT", "Claude"}) +ENQUEUE_REQUIRING_ALLOW = frozenset( + { + "observe", + "click", + "start_record", + "pause_for_input", + "stop_record", + "compile", + "set_coach", + "get_coach", + "halt", + } +) +COACH_ONLY_BACKENDS = frozenset({"windows", "rdp", "citrix"}) +PROCESS_NAME_RE = re.compile(r"^[A-Za-z0-9 ._-]{1,64}$") +SIX_DIGITS_RE = re.compile(r"\d{6,}") +_SHA256_HEX = re.compile(r"^[a-f0-9]{64}$") +_SAFE_PARAM = re.compile(r"^[A-Za-z0-9_]{1,40}$") +_NODE_ID = re.compile(r"^n_[a-f0-9]{8}$") +_COMMAND_ID = re.compile(r"^[A-Za-z0-9_.:-]{1,200}$") + + +class AuthoringError(RuntimeError): + """A safe, user-facing authoring failure with no secret-bearing text.""" + + +class AuthoringCoachOnly(AuthoringError): + """This substrate cannot agent-drive in v1.""" + + +class AuthoringTransportError(AuthoringError): + """The mailbox HTTPS transport did not confirm an operation.""" + + +def _utc_now() -> str: + return datetime.now(timezone.utc).isoformat() + + +def _sha256_hex(value: str) -> str: + return hashlib.sha256(value.encode("utf-8")).hexdigest() + + +def _pack_dir(data_dir: Path, pack_id: str) -> Path: + return Path(data_dir) / "authoring" / _sha256_hex(pack_id)[:16] + + +def filter_coach_hint(text: object) -> str | None: + """Apply the 80-character / no-URL / no-``@`` / no-6-digits coach filter.""" + + if not isinstance(text, str): + return None + collapsed = " ".join(text.split()) + if not collapsed or len(collapsed) > 80: + return None + if "://" in collapsed or "@" in collapsed or SIX_DIGITS_RE.search(collapsed): + return None + return collapsed + + +def _client_display(value: object) -> str: + if value in CLIENT_DISPLAYS: + return str(value) + return "ChatGPT" + + +class NodeTable: + """Laptop-only node table. Mode 0600, 15-minute lifetime.""" + + def __init__(self, path: Path, hmac_key: bytes) -> None: + self._path = Path(path) + self._hmac_key = hmac_key + self._lock = threading.Lock() + + def clear(self) -> None: + with self._lock: + try: + self._path.unlink() + except FileNotFoundError: + return + + def mint_node_id(self, provider_runtime_id: str) -> str: + digest = hmac.new( + self._hmac_key, + provider_runtime_id.encode("utf-8"), + hashlib.sha256, + ).hexdigest() + return f"n_{digest[:8]}" + + def replace(self, rows: list[dict[str, Any]]) -> None: + payload = { + "updated_at": time.time(), + "rows": rows, + } + encoded = json.dumps(payload, sort_keys=True, separators=(",", ":")) + self._path.parent.mkdir(parents=True, mode=0o700, exist_ok=True) + if not os.name == "nt": + os.chmod(self._path.parent, 0o700) + flags = os.O_WRONLY | os.O_CREAT | os.O_TRUNC | getattr(os, "O_NOFOLLOW", 0) + descriptor = os.open(self._path, flags, 0o600) + try: + if os.name != "nt": + os.fchmod(descriptor, 0o600) + os.write(descriptor, encoded.encode("utf-8")) + finally: + os.close(descriptor) + + def get(self, node_id: str) -> dict[str, Any] | None: + with self._lock: + try: + raw = self._path.read_text(encoding="utf-8") + details = self._path.stat() + except OSError: + return None + if os.name != "nt" and stat.S_IMODE(details.st_mode) != 0o600: + return None + try: + payload = json.loads(raw) + except json.JSONDecodeError: + return None + if not isinstance(payload, dict) or not isinstance(payload.get("rows"), list): + return None + updated = payload.get("updated_at") + stale = not isinstance(updated, (int, float)) or ( + time.time() - updated > NODE_TABLE_LIFETIME_S + ) + if stale: + return None + for row in payload["rows"]: + if isinstance(row, dict) and row.get("node_id") == node_id: + observed = row.get("observed_at") + if ( + isinstance(observed, (int, float)) + and time.time() - (observed / 1000.0) > NODE_TABLE_LIFETIME_S + ): + return None + return row + return None + + +def project_observe( + *, + backend: str, + provider: str, + recording: bool, + agent_drive: bool, + coach_only: bool, + process_name: str | None, + raw_nodes: list[dict[str, Any]], + node_table: NodeTable, +) -> dict[str, Any]: + """PHI-safe observe projection. Fail closed; never a raw fallback.""" + + window = { + "process_name": ( + process_name if process_name and PROCESS_NAME_RE.fullmatch(process_name) else None + ), + "role": "window", + "bounds": {"x": 0.0, "y": 0.0, "w": 1.0, "h": 1.0}, + } + if window["process_name"] is None: + window.pop("process_name") + tree: list[dict[str, Any]] = [] + rows: list[dict[str, Any]] = [] + if agent_drive and not coach_only: + for raw in raw_nodes: + projected, row = _project_node(raw, node_table) + if projected is None or row is None: + continue + tree.append(projected) + rows.append(row) + if len(tree) >= 200: + break + node_table.replace(rows) + payload: dict[str, Any] = { + "schema_version": OBSERVE_SCHEMA, + "backend": backend, + "provider": provider, + "mode": "authoring", + "agent_drive": agent_drive and not coach_only, + "coach_only": coach_only or not agent_drive, + "recording": recording, + "window": window, + "tree": tree, + "truncated": len(raw_nodes) > len(tree), + "node_count": len(tree), + } + encoded = json.dumps(payload, separators=(",", ":")).encode("utf-8") + if len(encoded) > 32 * 1024: + payload["tree"] = [] + payload["truncated"] = True + payload["node_count"] = 0 + payload["reason"] = "empty_projection" + node_table.replace([]) + elif not tree: + payload["reason"] = "empty_projection" + return payload + + +def _project_node( + raw: dict[str, Any], + node_table: NodeTable, +) -> tuple[dict[str, Any] | None, dict[str, Any] | None]: + if not isinstance(raw, dict): + return None, None + runtime_id = raw.get("provider_runtime_id") + pixels = raw.get("backend_pixels") + bounds = raw.get("bounds") + if not isinstance(runtime_id, str) or not runtime_id: + return None, None + if not isinstance(pixels, dict) or not isinstance(bounds, dict): + return None, None + try: + pixel_box = {key: int(pixels[key]) for key in ("x", "y", "w", "h")} + normalized = {key: float(bounds[key]) for key in ("x", "y", "w", "h")} + except (KeyError, TypeError, ValueError): + return None, None + node_id = node_table.mint_node_id(runtime_id) + projected: dict[str, Any] = { + "node_id": node_id, + "role": str(raw.get("role") or "unknown")[:40], + "control_type": str(raw.get("control_type") or "")[:40], + "enabled": bool(raw.get("enabled", True)), + "focused": bool(raw.get("focused", False)), + "bounds": normalized, + } + automation_id = _project_label(raw.get("automation_id")) + if automation_id: + projected["automation_id"] = automation_id + name = _project_label(raw.get("name")) + if name: + projected["name"] = name + row = { + "node_id": node_id, + "backend_pixels": pixel_box, + "normalized": normalized, + "provider_runtime_id": runtime_id, + "observed_at": int(time.time() * 1000), + } + return projected, row + + +def _project_label(value: object) -> str | None: + if not isinstance(value, str): + return None + collapsed = " ".join(value.split()) + if not collapsed or len(collapsed) > 80: + return None + if "://" in collapsed or "@" in collapsed or SIX_DIGITS_RE.search(collapsed): + return None + return collapsed + + +class AuthoringMailboxTransport: + """Outbound HTTPS bind/poll/callback. Wait is always 0.""" + + def __init__( + self, + *, + origin: str, + audit: Any, + client: httpx.Client | None = None, + ) -> None: + if origin != AUTHORING_ORIGIN: + raise AuthoringTransportError("The authoring origin is not pinned.") + self.origin = origin + self._audit = audit + if client is not None: + base = str(client.base_url).removesuffix("/") + if base != origin: + raise AuthoringTransportError("The authoring HTTP client differs from its origin.") + self._client = client or httpx.Client( + base_url=origin, + timeout=API_TIMEOUT_S, + follow_redirects=False, + ) + self._owns_client = client is None + + def close(self) -> None: + if self._owns_client: + self._client.close() + + def _path(self, pack_id: str, action: str) -> str: + if not valid_pack_id(pack_id) or action not in {"claim", "poll", "callback"}: + raise AuthoringTransportError("The authoring mailbox path is invalid.") + return f"/j/{quote(pack_id, safe='._-')}/runner/{action}" + + def _post( + self, + path: str, + body: dict[str, Any], + *, + headers: dict[str, str], + expected: tuple[int, ...], + allow_empty: bool = False, + ) -> tuple[int, dict[str, Any] | None]: + operation = path.rsplit("/", 1)[-1] + self._audit.log( + "authoring_request", + operation=operation, + destination=self.origin, + path=path.rsplit("/", 3)[0] + "/runner/" + operation, + ) + try: + response = self._client.post( + path, + json=body, + headers=headers, + follow_redirects=False, + ) + except (httpx.HTTPError, OSError) as exc: + self._audit.log( + "authoring_transport_failed", + operation=operation, + destination=self.origin, + error_type=type(exc).__name__, + ) + raise AuthoringTransportError( + f"The authoring {operation} request did not complete." + ) from exc + self._audit.log( + "authoring_response", + operation=operation, + destination=self.origin, + status_code=response.status_code, + ) + if response.status_code == 401: + raise AuthoringTransportError("The authoring mailbox credential was rejected.") + if allow_empty and response.status_code == 204: + return 204, None + if response.status_code not in expected: + raise AuthoringTransportError( + f"The authoring {operation} request returned HTTP {response.status_code}." + ) + cache = (response.headers.get("cache-control") or "").strip().lower() + if cache != "no-store": + raise AuthoringTransportError( + f"The authoring {operation} response was not marked no-store." + ) + try: + parsed = response.json() + except (ValueError, json.JSONDecodeError) as exc: + raise AuthoringTransportError( + f"The authoring {operation} response was not valid JSON." + ) from exc + if not isinstance(parsed, dict): + raise AuthoringTransportError(f"The authoring {operation} response was not an object.") + return response.status_code, parsed + + def claim(self, pack_id: str, bind: str) -> dict[str, Any]: + path = self._path(pack_id, "claim") + status, body = self._post( + path, + {"bind": bind}, + headers={"Content-Type": "application/json"}, + expected=(201,), + ) + assert status == 201 + assert body is not None + secret = body.get("leaseSecret") + lease_s = body.get("lease_s", DEFAULT_LEASE_S) + if not valid_lease_secret(secret) or not isinstance(lease_s, int) or lease_s <= 0: + raise AuthoringTransportError("The authoring claim response was not a mailbox lease.") + return {"leaseSecret": secret, "lease_s": lease_s} + + def poll(self, pack_id: str, lease_secret: str) -> dict[str, Any] | None: + if not valid_lease_secret(lease_secret): + raise AuthoringTransportError("The authoring mailbox credential is malformed.") + path = self._path(pack_id, "poll") + _, body = self._post( + path, + {"wait_seconds": POLL_WAIT_S, "lease_seconds": DEFAULT_LEASE_S}, + headers={ + "Authorization": f"Bearer {lease_secret}", + "Content-Type": "application/json", + }, + expected=(200,), + allow_empty=True, + ) + return body + + def callback( + self, + pack_id: str, + lease_secret: str, + payload: dict[str, Any], + ) -> None: + if not valid_lease_secret(lease_secret): + raise AuthoringTransportError("The authoring mailbox credential is malformed.") + path = self._path(pack_id, "callback") + self._post( + path, + payload, + headers={ + "Authorization": f"Bearer {lease_secret}", + "Content-Type": "application/json", + }, + expected=(200, 202), + ) + + +class AuthoringRunner: + """Claim, Allow-per-sub, wait=0 poll, and Flow record_observed session.""" + + def __init__( + self, + config: EngineConfig, + *, + emit: Callable[[str, dict[str, Any]], None] | None = None, + audit: Any | None = None, + client: httpx.Client | None = None, + sleep: Callable[[float], None] = time.sleep, + observe_nodes: Callable[[], list[dict[str, Any]]] | None = None, + recorder_factory: Callable[..., Any] | None = None, + compile_recording: Callable[..., Any] | None = None, + playwright_launcher: Callable[[str], Any] | None = None, + text_value_at: Callable[[dict[str, int]], str | None] | None = None, + ) -> None: + self.config = config + self.emit = emit or (lambda _event, _data: None) + self.audit = audit + self._client = client + self._sleep = sleep + self._observe_nodes = observe_nodes or (lambda: []) + self._recorder_factory = recorder_factory + self._compile_recording = compile_recording + self._playwright_launcher = playwright_launcher + self._text_value_at = text_value_at + self._transport: AuthoringMailboxTransport | None = None + self._lock = threading.RLock() + self._stop = threading.Event() + self._thread: threading.Thread | None = None + self._pack: str | None = None + self._lease_secret: str | None = None + self._allowed_sub: str | None = None + self._allowed_client_id: str | None = None + self._pending_allow: dict[str, Any] | None = None + self._pin: dict[str, Any] = {"backend": "macos"} + self._node_table: NodeTable | None = None + self._recorder: Any | None = None + self._recording = False + self._paused = False + self._pause_target: dict[str, Any] | None = None + self._secret_pause = False + self._secret_type_recorded = False + self._actuation_started = False + self._uncertain = False + self._coach_hint: str | None = None + self._out_dir: Path | None = None + self._playwright: Any | None = None + + def is_bound(self) -> bool: + return self._pack is not None and self._lease_secret is not None + + def has_pause(self) -> bool: + return self._paused and self._recorder is not None + + def status_dict(self) -> dict[str, Any] | None: + if not self.is_bound(): + return None + if self._paused: + return { + "recording": True, + "paused": True, + "capture_id": None, + "controls": {"pause": False, "resume": True, "stop": True}, + } + if self._recording: + return { + "recording": True, + "paused": False, + "capture_id": None, + "controls": {"pause": False, "resume": False, "stop": True}, + } + return { + "recording": False, + "paused": False, + "capture_id": None, + "controls": {"pause": False, "resume": False, "stop": self.is_bound()}, + } + + def status(self) -> dict[str, Any]: + pending = self._pending_allow + if pending and self._allowed_sub and pending.get("oauth_sub_sha256") != self._allowed_sub: + state = "replace_allow" + elif pending: + state = "pending_allow" + elif self.is_bound(): + state = "bound" + else: + state = "idle" + return { + "status": state, + "pack_bound": bool(self._pack), + "allowed": bool(self._allowed_sub), + "client_display": pending.get("client_display") if pending else None, + "coach_only": self._pin.get("backend") in COACH_ONLY_BACKENDS, + } + + def pin_target(self, **fields: Any) -> dict[str, Any]: + backend = str(fields.get("backend") or self._pin.get("backend") or "macos") + pin = { + "backend": backend, + "url": fields.get("url"), + "macos_app": fields.get("macos_app"), + "macos_window_title": fields.get("macos_window_title"), + "linux_app": fields.get("linux_app"), + "linux_window_title": fields.get("linux_window_title"), + "window_title_unique": fields.get("window_title_unique", True), + } + self._pin = pin + return {"ok": True, "backend": backend} + + def claim_uri(self, uri: str, *, start_loop: bool = True) -> dict[str, Any]: + parsed = parse_runner_uri(uri) + pack = parsed["pack"] + bind = parsed["bind"] + origin = parsed["origin"] + transport = AuthoringMailboxTransport( + origin=origin, + audit=self.audit or _NullAudit(), + client=self._client, + ) + try: + claimed = transport.claim(pack, bind) + except AuthoringTransportError: + close = getattr(transport, "close", None) + if self._client is None and callable(close): + close() + raise + lease_secret = claimed["leaseSecret"] + payload = { + "pack": pack, + "origin": origin, + "lease_secret": lease_secret, + "lease_s": int(claimed["lease_s"]), + "claimed_at": _utc_now(), + "allowed_sub": None, + "allowed_client_id": None, + "allowed_at": None, + } + if not store_authoring_lease(pack, payload): + close = getattr(transport, "close", None) + if self._client is None and callable(close): + close() + raise AuthoringError( + "Desktop could not store the authoring lease in the OS keychain." + ) + with self._lock: + self._transport = transport + self._pack = pack + self._lease_secret = lease_secret + hmac_key = hashlib.sha256(lease_secret.encode("utf-8")).digest() + self._node_table = NodeTable( + _pack_dir(self.config.data_dir, pack) / "nodes.json", + hmac_key, + ) + self._out_dir = _pack_dir(self.config.data_dir, pack) / "recording" + if self.audit: + self.audit.log("authoring_bind_claimed", pack_hash=_sha256_hex(pack)[:16]) + if start_loop: + self.start() + self.emit("authoring_state", {"status": "bound"}) + return {"bound": True, "origin": origin, "pack_prefix": pack[:2]} + + def start(self) -> None: + with self._lock: + if self._thread and self._thread.is_alive(): + return + self._stop.clear() + self._thread = threading.Thread(target=self._loop, name="authoring-poll", daemon=True) + self._thread.start() + + def stop_loop(self) -> None: + self._stop.set() + thread = self._thread + if thread is not None: + thread.join(timeout=2.0) + + def _loop(self) -> None: + while not self._stop.is_set(): + try: + self.poll_once() + except AuthoringError: + logger.warning("authoring poll failed") + self._sleep(LOCAL_POLL_SLEEP_S) + + def poll_once(self) -> None: + transport = self._transport + pack = self._pack + secret = self._lease_secret + if transport is None or pack is None or secret is None: + return + body = transport.poll(pack, secret) + if body is None: + return + if body.get("halted") is True or ( + isinstance(body.get("head"), dict) and body["head"].get("halted") is True + ): + self._halt(unsigned=True, command_id=None) + return + envelope = body.get("command") if isinstance(body.get("command"), dict) else body + if isinstance(envelope, dict) and envelope.get("tool"): + self.handle_envelope(envelope) + + def handle_envelope(self, envelope: dict[str, Any]) -> None: + tool = envelope.get("tool") + command_id = envelope.get("command_id") + pack_id = envelope.get("pack_id") + if tool not in ENQUEUE_REQUIRING_ALLOW | {"bind_pack"}: + self._callback_error(command_id, "unknown_tool") + return + if not isinstance(command_id, str) or _COMMAND_ID.fullmatch(command_id) is None: + return + if pack_id != self._pack: + self._callback_error(command_id, "pack_mismatch") + return + sub = envelope.get("oauth_sub_sha256") + if tool == "bind_pack": + self._queue_allow(envelope) + return + if not self._allowed_sub or sub != self._allowed_sub: + self._callback_error(command_id, "not_allowed") + return + args = envelope.get("args") if isinstance(envelope.get("args"), dict) else {} + try: + result = self._dispatch_tool(str(tool), args) + except AuthoringCoachOnly: + self._callback( + { + "command_id": command_id, + "status": "done", + "result": {"error": "COACH_ONLY", "agent_drive": False, "coach_only": True}, + } + ) + return + except AuthoringError as exc: + self._callback_error(command_id, str(exc)) + return + self._callback({"command_id": command_id, "status": "done", "result": result}) + + def _queue_allow(self, envelope: dict[str, Any]) -> None: + sub = envelope.get("oauth_sub_sha256") + client = envelope.get("client_id_sha256") + if not isinstance(sub, str) or _SHA256_HEX.fullmatch(sub) is None: + self._callback_error(envelope.get("command_id"), "invalid_allow") + return + if client is not None and ( + not isinstance(client, str) or _SHA256_HEX.fullmatch(client) is None + ): + self._callback_error(envelope.get("command_id"), "invalid_allow") + return + display = _client_display(envelope.get("client_display")) + self._pending_allow = { + "command_id": envelope.get("command_id"), + "oauth_sub_sha256": sub, + "client_id_sha256": client, + "client_display": display, + } + status = ( + "replace_allow" + if self._allowed_sub and self._allowed_sub != sub + else "pending_allow" + ) + copy = ( + f"A different {display} account is asking. Allow it to replace the current one?" + if status == "replace_allow" + else f"Allow {display} to drive this job" + ) + self.emit( + "authoring_state", + {"status": status, "client_display": display, "prompt": copy}, + ) + + def allow(self, *, replace: bool = False) -> dict[str, Any]: + pending = self._pending_allow + if pending is None: + raise AuthoringError("There is no pending Allow request.") + if ( + self._allowed_sub + and self._allowed_sub != pending["oauth_sub_sha256"] + and not replace + ): + return self.status() + granted_at = _utc_now() + self._allowed_sub = pending["oauth_sub_sha256"] + self._allowed_client_id = pending.get("client_id_sha256") + stored = load_authoring_lease(self._pack or "") if self._pack else None + if stored is not None: + stored["allowed_sub"] = self._allowed_sub + stored["allowed_client_id"] = self._allowed_client_id + stored["allowed_at"] = granted_at + store_authoring_lease(self._pack or "", stored) + command_id = pending.get("command_id") + display = pending.get("client_display") + self._pending_allow = None + if isinstance(command_id, str): + self._callback( + { + "command_id": command_id, + "status": "done", + "result": {"allowed": True}, + } + ) + if self.audit: + self.audit.log( + "authoring_allowed", + pack_hash=_sha256_hex(self._pack or "")[:16], + allowed_sub_prefix=(self._allowed_sub or "")[:8], + client_display=display, + ) + self.emit("authoring_state", {"status": "bound", "allowed": True}) + return {"allowed": True, "client_display": display} + + def deny(self) -> dict[str, Any]: + pending = self._pending_allow + self._pending_allow = None + if pending and isinstance(pending.get("command_id"), str): + self._callback_error(pending["command_id"], "denied") + self.emit("authoring_state", {"status": "bound"}) + return {"allowed": False} + + def continue_pause(self) -> dict[str, Any]: + if not self.has_pause() or self._pause_target is None or self._recorder is None: + return self.status_dict() or {"recording": False, "paused": False} + recorder = self._recorder + target = self._pause_target + if hasattr(recorder, "type_text"): + original = recorder.type_text + + def _forbidden(*_args: Any, **_kwargs: Any) -> None: + raise AuthoringError("Continue must not type") + + recorder.type_text = _forbidden + else: + original = None + try: + if target.get("secret"): + recorder.record_observed( + event={"kind": "type"}, + param=target.get("param"), + secret=True, + redact_region=target.get("backend_pixels"), + ) + self._secret_type_recorded = True + else: + text = None + if self._text_value_at is not None: + text = self._text_value_at(target["backend_pixels"]) + recorder.record_observed( + event={"kind": "type"}, + param=target.get("param"), + text=text, + ) + finally: + if original is not None: + recorder.type_text = original + self._paused = False + self._pause_target = None + self.emit("status_update", self.status_dict() or {}) + if self.audit: + self.audit.log( + "authoring_pause_typed", + param=target.get("param"), + secret=bool(target.get("secret")), + ) + return self.status_dict() or {} + + def operator_stop(self) -> dict[str, Any]: + self._halt(unsigned=True, command_id=None) + return {"recording": False, "paused": False, "halted": True} + + def _dispatch_tool(self, tool: str, args: dict[str, Any]) -> dict[str, Any]: + if tool == "observe": + return self._observe() + if tool == "start_record": + return self._start_record() + if tool == "click": + return self._click(args) + if tool == "pause_for_input": + return self._pause_for_input(args) + if tool == "stop_record": + return self._stop_record() + if tool == "compile": + return self._compile() + if tool == "halt": + self._halt(unsigned=False, command_id=None) + return {"halted": True} + if tool == "set_coach": + hint = filter_coach_hint(args.get("hint") or args.get("text")) + self._coach_hint = hint + return {"ok": hint is not None} + if tool == "get_coach": + return {"hint": self._coach_hint} + raise AuthoringError("unknown_tool") + + def _coach_only(self) -> bool: + backend = str(self._pin.get("backend") or "") + if backend in COACH_ONLY_BACKENDS: + return True + if backend == "linux" and not self._pin.get("window_title_unique"): + return True + return False + + def _observe(self) -> dict[str, Any]: + backend = str(self._pin.get("backend") or "macos") + coach_only = self._coach_only() + agent_drive = not coach_only + if self._node_table is None: + raise AuthoringError("not_bound") + raw = [] if coach_only else list(self._observe_nodes()) + provider = { + "web": "playwright_ax", + "macos": "ax", + "linux": "atspi", + }.get(backend, "none") + return project_observe( + backend=backend, + provider=provider, + recording=self._recording, + agent_drive=agent_drive, + coach_only=coach_only, + process_name="Chromium" if backend == "web" else None, + raw_nodes=raw, + node_table=self._node_table, + ) + + def _start_record(self) -> dict[str, Any]: + if self._coach_only(): + raise AuthoringCoachOnly("COACH_ONLY") + backend = str(self._pin.get("backend") or "macos") + if backend == "windows": + raise AuthoringCoachOnly("COACH_ONLY") + if backend == "web": + url = self._pin.get("url") + if not isinstance(url, str) or not url.startswith("https://"): + raise AuthoringError("A Playwright job needs a URL typed into Desktop.") + if self._playwright_launcher is not None: + self._playwright = self._playwright_launcher(url) + factory = self._recorder_factory + if factory is None: + raise AuthoringError("Authoring recorder is unavailable.") + out_dir = self._out_dir or _pack_dir(self.config.data_dir, self._pack or "p.invalidpackid") + out_dir.mkdir(parents=True, exist_ok=True) + self._recorder = factory(out_dir) + self._recording = True + self._secret_pause = False + self._secret_type_recorded = False + self.emit("status_update", self.status_dict() or {}) + return {"recording": True} + + def _click(self, args: dict[str, Any]) -> dict[str, Any]: + if self._coach_only(): + raise AuthoringCoachOnly("COACH_ONLY") + if self._uncertain: + raise AuthoringError("RECONCILIATION_REQUIRED") + node_id = args.get("node_id") + if not isinstance(node_id, str) or _NODE_ID.fullmatch(node_id) is None: + raise AuthoringError("stale_node") + if self._node_table is None or self._recorder is None: + raise AuthoringError("stale_node") + row = self._node_table.get(node_id) + if row is None: + raise AuthoringError("stale_node") + pixels = row["backend_pixels"] + x = int(pixels["x"] + pixels["w"] / 2) + y = int(pixels["y"] + pixels["h"] / 2) + self._actuation_started = True + try: + self._recorder.click(x, y) + except Exception: + self._uncertain = True + raise AuthoringError("RECONCILIATION_REQUIRED") from None + finally: + self._actuation_started = False + return {"clicked": True} + + def _pause_for_input(self, args: dict[str, Any]) -> dict[str, Any]: + if self._recorder is None: + raise AuthoringError("not_recording") + node_id = args.get("node_id") + param = args.get("param") or "note" + if not isinstance(param, str) or _SAFE_PARAM.fullmatch(param) is None: + raise AuthoringError("invalid_param") + secret = bool(args.get("secret")) + row = None + if isinstance(node_id, str) and self._node_table is not None: + row = self._node_table.get(node_id) + if row is None: + raise AuthoringError("stale_node") + self._pause_target = { + "node_id": row["node_id"], + "backend_pixels": row["backend_pixels"], + "param": param, + "secret": secret, + } + if secret: + self._secret_pause = True + self._paused = True + self.emit("status_update", self.status_dict() or {}) + return {"paused": True, "param": param} + + def _stop_record(self) -> dict[str, Any]: + recorder = self._recorder + if recorder is None: + return {"recording": False} + finish = getattr(recorder, "finish", None) + if callable(finish): + finish() + self._recording = False + self._paused = False + self.emit("status_update", self.status_dict() or {}) + return {"recording": False} + + def _compile(self) -> dict[str, Any]: + if self._secret_pause and not self._secret_type_recorded: + if self.audit: + self.audit.log("authoring_compile_refused_missing_type") + raise AuthoringError("secret_type_missing") + compile_recording = self._compile_recording + workflow_id = "wf_local" + if callable(compile_recording) and self._out_dir is not None: + workflow = compile_recording(self._out_dir) + workflow_id = str( + getattr(workflow, "id", None) or getattr(workflow, "workflow_id", workflow_id) + ) + if self._node_table is not None: + self._node_table.clear() + return { + "status": "needs_human_admit", + "workflow_id": workflow_id, + "recording_retained": True, + } + + def _halt(self, *, unsigned: bool, command_id: str | None) -> None: + if self._actuation_started: + self._uncertain = True + self._callback( + { + "command_id": command_id, + "status": "error", + "result": {"error": "RECONCILIATION_REQUIRED"}, + } + ) + return + self._recording = False + self._paused = False + self._recorder = None + if self._node_table is not None: + self._node_table.clear() + if unsigned and self._pack and self._lease_secret and self._transport: + self._transport.callback( + self._pack, + self._lease_secret, + {"halted": True, "status": "halted"}, + ) + self.emit("status_update", {"recording": False, "paused": False, "halted": True}) + + def _callback(self, payload: dict[str, Any]) -> None: + if self._transport is None or self._pack is None or self._lease_secret is None: + return + closed = { + key: payload[key] + for key in ("command_id", "status", "result", "halted") + if key in payload + } + self._transport.callback(self._pack, self._lease_secret, closed) + + def _callback_error(self, command_id: object, error: str) -> None: + if not isinstance(command_id, str): + return + self._callback( + { + "command_id": command_id, + "status": "error", + "result": {"error": error}, + } + ) + + +class _NullAudit: + def log(self, *_args: Any, **_kwargs: Any) -> None: + return + + +def restore_authoring_runner( + config: EngineConfig, pack_id: str, **kwargs: Any +) -> AuthoringRunner | None: + """Rebuild an authoring runner from a stored lease without re-claiming.""" + + if not valid_pack_id(pack_id): + return None + stored = load_authoring_lease(pack_id) + if stored is None: + return None + runner = AuthoringRunner(config, **kwargs) + runner._pack = stored["pack"] + runner._lease_secret = stored["lease_secret"] + runner._allowed_sub = stored.get("allowed_sub") + runner._allowed_client_id = stored.get("allowed_client_id") + hmac_key = hashlib.sha256(stored["lease_secret"].encode("utf-8")).digest() + runner._node_table = NodeTable(_pack_dir(config.data_dir, pack_id) / "nodes.json", hmac_key) + runner._transport = AuthoringMailboxTransport( + origin=stored["origin"], + audit=kwargs.get("audit") or _NullAudit(), + client=kwargs.get("client"), + ) + return runner diff --git a/engine/dispatch.py b/engine/dispatch.py index cbec8c3..ff2e8fd 100644 --- a/engine/dispatch.py +++ b/engine/dispatch.py @@ -121,6 +121,7 @@ def __init__( flow_bridge: Any = None, runner: Any = None, portal: Any = None, + authoring: Any = None, ) -> None: self.config = config self._db = db @@ -135,6 +136,7 @@ def __init__( # The mobile decision portal is likewise built on first use so the # engine never binds a socket or spawns a console it was not asked for. self.portal = portal + self.authoring = authoring @property def db(self) -> Any: @@ -284,6 +286,11 @@ def _register(self) -> None: "login_browser": self.login_browser, "login_paste": self.login_paste, "connect_uri": self.connect_uri, + "claim_runner_uri": self.claim_runner_uri, + "authoring_allow": self.authoring_allow, + "authoring_deny": self.authoring_deny, + "authoring_status": self.authoring_status, + "authoring_pin_target": self.authoring_pin_target, "logout": self.logout, "get_auth_status": self.get_auth_status, # config / settings @@ -567,6 +574,9 @@ def _remember_first_workflow_recording( def stop_recording(self, **params: Any) -> dict: """Stop the active recording, retain it, and compile it automatically.""" + authoring = self.services.authoring + if authoring is not None and authoring.is_bound(): + return authoring.operator_stop() controller = self.services.controller active = self._flow_recording if active is not None: @@ -654,11 +664,19 @@ def pause_recording(self, **params: Any) -> dict: return self.get_status() def resume_recording(self, **params: Any) -> dict: - """Resume is not supported (stop/start instead); report current status.""" + """Overlay Resume during an authoring pause records the typed field.""" + authoring = self.services.authoring + if authoring is not None and authoring.has_pause(): + return authoring.continue_pause() return self.get_status() def get_status(self, **params: Any) -> dict: """Return the current :class:`EngineStatus`-shaped recording status.""" + authoring = self.services.authoring + if authoring is not None: + status = authoring.status_dict() + if status is not None and (status.get("recording") or status.get("halted")): + return status return self._status_dict(self.services.controller) def _status_dict(self, controller: Any) -> dict: @@ -3261,6 +3279,42 @@ def connect_uri(self, **params: Any) -> dict: ) return result + def _authoring_service(self) -> Any: + if self.services.authoring is None: + from engine.authoring_runner import AuthoringRunner + + self.services.authoring = AuthoringRunner( + self.config, + emit=self.emit, + audit=self.services.audit, + ) + return self.services.authoring + + def claim_runner_uri(self, **params: Any) -> dict: + """Claim one validated ``openadapt://runner`` bind URI.""" + uri = params.get("uri") + if not isinstance(uri, str): + raise ValueError("uri is required") + result = self._authoring_service().claim_uri(uri) + self.emit("authoring_state", {"status": "bound"}) + return result + + def authoring_allow(self, **params: Any) -> dict: + """Allow the pending connector ``sub`` to drive this job.""" + return self._authoring_service().allow(replace=params.get("replace") is True) + + def authoring_deny(self, **params: Any) -> dict: + """Refuse the pending Allow request.""" + return self._authoring_service().deny() + + def authoring_status(self, **params: Any) -> dict: + """Return the local authoring bind / Allow state.""" + return self._authoring_service().status() + + def authoring_pin_target(self, **params: Any) -> dict: + """Pin the local authoring backend. Titles never go to MCP.""" + return self._authoring_service().pin_target(**params) + def logout(self, **params: Any) -> dict: """Clear only the credential for the selected safe hosted origin.""" from engine.auth.store import ( diff --git a/src/App.tsx b/src/App.tsx index 398bf8d..2dad928 100644 --- a/src/App.tsx +++ b/src/App.tsx @@ -170,6 +170,12 @@ export default function App() { const [sync, setSync] = useState({ state: "synced", queued: 0 }); const [breaks, setBreaks] = useState(0); const [pairing, setPairing] = useState(null); + const [authoring, setAuthoring] = useState<{ + status: string; + client_display?: string; + prompt?: string; + error?: string; + } | null>(null); const [firstRunPersistencePending, setFirstRunPersistencePending] = useState(false); const [firstWorkflowRunning, setFirstWorkflowRunning] = useState(false); @@ -312,6 +318,17 @@ export default function App() { }); } }), + onEngineEvent( + EVT.AUTHORING_STATE, + (state: { + status: string; + client_display?: string; + prompt?: string; + error?: string; + }) => { + setAuthoring(state); + }, + ), ]; return () => unsubs.forEach((p) => p.then((u) => u()).catch(() => {})); // eslint-disable-next-line react-hooks/exhaustive-deps @@ -341,6 +358,60 @@ export default function App() { )} ) : null; + const authoringNotice = + authoring && + (authoring.status === "pending_allow" || + authoring.status === "replace_allow" || + authoring.status === "error") ? ( +
+ + {authoring.status === "error" + ? authoring.error || "The authoring bind could not be completed." + : authoring.prompt || + (authoring.status === "replace_allow" + ? `A different ${authoring.client_display || "ChatGPT"} account is asking. Allow it to replace the current one?` + : `Allow ${authoring.client_display || "ChatGPT"} to drive this job`)} + + {authoring.status !== "error" && ( + + + + + )} + {authoring.status === "error" && ( + + )} +
+ ) : null; const firstWorkflowStageNotice = firstWorkflowStageError ? (
{firstWorkflowStageError} @@ -358,6 +429,7 @@ export default function App() { return ( <> {pairingNotice} + {authoringNotice} {firstWorkflowStageNotice}
Loading…
@@ -370,6 +442,7 @@ export default function App() { return ( <> {pairingNotice} + {authoringNotice} {firstWorkflowStageNotice} {pairingNotice} + {authoringNotice} {firstWorkflowStageNotice} {pairingNotice} + {authoringNotice} {firstWorkflowStageNotice}
diff --git a/src/lib/engine.ts b/src/lib/engine.ts index 438bcef..6859ac9 100644 --- a/src/lib/engine.ts +++ b/src/lib/engine.ts @@ -70,6 +70,11 @@ export const CMD = { LOGIN_PASTE: "login_paste", LOGOUT: "logout", GET_AUTH_STATUS: "get_auth_status", + CLAIM_RUNNER_URI: "claim_runner_uri", + AUTHORING_ALLOW: "authoring_allow", + AUTHORING_DENY: "authoring_deny", + AUTHORING_STATUS: "authoring_status", + AUTHORING_PIN_TARGET: "authoring_pin_target", // config / settings (lane, phi_mode, hosted host) GET_CONFIG: "get_config", SET_CONFIG: "set_config", @@ -119,6 +124,7 @@ export const EVT = { BREAK_COUNT: "break_count", SIDECAR_STATE: "sidecar_state", PAIRING_STATE: "pairing_state", + AUTHORING_STATE: "authoring_state", RUNNER_STATE: "runner_state", PORTAL_STATE: "portal_state", // Carries only {title, body, open_count, route}; see attentionNotification.ts. diff --git a/src/styles/app.css b/src/styles/app.css index facdde4..8780e76 100644 --- a/src/styles/app.css +++ b/src/styles/app.css @@ -1515,6 +1515,19 @@ select.input:focus { font-size: 20px; } +.pairing-notice .allow-actions { + display: flex; + gap: 8px; + flex-shrink: 0; +} + +.pairing-notice .allow-actions button { + border: 1px solid var(--border); + border-radius: 8px; + padding: 6px 12px; + font-size: 13px; +} + /* decision portal — pairing panel and paired-device list */ .pairing-panel { display: flex; diff --git a/tests/test_engine/test_authoring_runner.py b/tests/test_engine/test_authoring_runner.py new file mode 100644 index 0000000..edd295a --- /dev/null +++ b/tests/test_engine/test_authoring_runner.py @@ -0,0 +1,457 @@ +"""Authoring mailbox claim, Allow-per-sub, wait=0 poll, and record_observed.""" + +from __future__ import annotations + +import json +import stat +from pathlib import Path +from types import SimpleNamespace +from typing import Any + +import httpx +import pytest + +from engine.auth.store import load_authoring_lease +from engine.authoring_runner import ( + POLL_WAIT_S, + AuthoringCoachOnly, + AuthoringError, + AuthoringRunner, + project_observe, +) +from engine.config import EngineConfig + +BIND = "oab_" + "A" * 43 +PACK = "p.abcdefghijkl" +LEASE = "oals_" + "a" * 64 +ORIGIN = "https://openadapt.ai" +URI = f"openadapt://runner?pack={PACK}&bind={BIND}&origin=https%3A%2F%2Fopenadapt.ai" +SUB = "b" * 64 +OTHER_SUB = "d" * 64 +CLIENT = "c" * 64 + + +class FakeAudit: + def __init__(self) -> None: + self.events: list[tuple[str, dict[str, Any]]] = [] + + def log(self, event: str, **data: Any) -> None: + self.events.append((event, data)) + + +class FakeRecorder: + def __init__(self) -> None: + self.clicks: list[tuple[int, int]] = [] + self.observed: list[dict[str, Any]] = [] + self.typed: list[Any] = [] + self.finished = False + + def click(self, x: int, y: int) -> None: + self.clicks.append((x, y)) + + def type_text(self, *args: Any, **kwargs: Any) -> None: + self.typed.append((args, kwargs)) + raise AssertionError("Continue must not call type_text") + + def record_observed(self, **kwargs: Any) -> None: + self.observed.append(kwargs) + + def finish(self) -> SimpleNamespace: + self.finished = True + return SimpleNamespace(ok=True) + + +def _nodes() -> list[dict[str, Any]]: + return [ + { + "provider_runtime_id": "ax-elem-1", + "role": "button", + "control_type": "button", + "automation_id": "btnContinue", + "enabled": True, + "focused": False, + "bounds": {"x": 0.72, "y": 0.88, "w": 0.14, "h": 0.05}, + "backend_pixels": {"x": 920, "y": 640, "w": 180, "h": 36}, + "value": "SSN-SECRET", + "title": "Patient chart", + "name": "Continue", + }, + { + "provider_runtime_id": "ax-note", + "role": "text_input", + "control_type": "edit", + "automation_id": "note", + "enabled": True, + "focused": True, + "bounds": {"x": 0.2, "y": 0.4, "w": 0.5, "h": 0.1}, + "backend_pixels": {"x": 200, "y": 400, "w": 500, "h": 40}, + "name": "note", + }, + ] + + +def _mailbox( + tmp_path: Path, + *, + claim_status: int = 201, + claim_body: dict[str, Any] | None = None, + poll_bodies: list[dict[str, Any] | None] | None = None, +) -> tuple[AuthoringRunner, list[httpx.Request], FakeRecorder, FakeAudit]: + requests: list[httpx.Request] = [] + polls = list(poll_bodies or []) + recorder = FakeRecorder() + audit = FakeAudit() + + def handler(request: httpx.Request) -> httpx.Response: + requests.append(request) + path = request.url.path + if path.endswith("/runner/claim"): + if claim_status == 201: + return httpx.Response( + 201, + headers={"Cache-Control": "no-store"}, + json=claim_body + or {"leaseSecret": LEASE, "lease_s": 900}, + ) + return httpx.Response( + claim_status, + headers={"Cache-Control": "no-store"}, + json={"error": "rejected"}, + ) + if path.endswith("/runner/poll"): + body = json.loads(request.content) + assert body["wait_seconds"] == 0 + assert body["lease_seconds"] == 900 + assert request.headers["Authorization"] == f"Bearer {LEASE}" + if not polls: + return httpx.Response(204) + next_body = polls.pop(0) + if next_body is None: + return httpx.Response(204) + return httpx.Response( + 200, + headers={"Cache-Control": "no-store"}, + json=next_body, + ) + if path.endswith("/runner/callback"): + assert request.headers["Authorization"] == f"Bearer {LEASE}" + return httpx.Response( + 202, + headers={"Cache-Control": "no-store"}, + json={"accepted": True}, + ) + return httpx.Response(404) + + client = httpx.Client( + base_url=ORIGIN, + transport=httpx.MockTransport(handler), + follow_redirects=False, + ) + config = EngineConfig(data_dir=tmp_path / ".openadapt", log_level="WARNING") + config.data_dir.mkdir(parents=True, exist_ok=True) + runner = AuthoringRunner( + config, + audit=audit, + client=client, + sleep=lambda _seconds: None, + observe_nodes=_nodes, + recorder_factory=lambda _out_dir: recorder, + compile_recording=lambda _out_dir: SimpleNamespace(id="wf_mockmed"), + playwright_launcher=lambda url: SimpleNamespace(url=url, cookies=lambda: []), + text_value_at=lambda _pixels: "follow up in two weeks", + ) + return runner, requests, recorder, audit + + +def _envelope(tool: str, *, sub: str = SUB, args: dict[str, Any] | None = None) -> dict[str, Any]: + return { + "schema_version": "openadapt.authoring.command/v1", + "command_id": f"cmd_{tool}", + "pack_id": PACK, + "tool": tool, + "args": args or {}, + "oauth_sub_sha256": sub, + "client_id_sha256": CLIENT, + "client_display": "ChatGPT", + } + + +def test_claim_stores_lease_without_returning_the_secret(tmp_path: Path) -> None: + runner, requests, _recorder, _audit = _mailbox(tmp_path) + result = runner.claim_uri(URI, start_loop=False) + assert result["bound"] is True + assert "leaseSecret" not in result + assert "lease_secret" not in result + stored = load_authoring_lease(PACK) + assert stored is not None + assert stored["lease_secret"] == LEASE + assert stored["origin"] == ORIGIN + assert requests[0].url.path == f"/j/{PACK}/runner/claim" + assert json.loads(requests[0].content) == {"bind": BIND} + + +@pytest.mark.parametrize("status", [409, 410, 404, 401]) +def test_claim_maps_mailbox_failures(tmp_path: Path, status: int) -> None: + runner, _requests, _recorder, _audit = _mailbox(tmp_path, claim_status=status) + with pytest.raises(AuthoringError): + runner.claim_uri(URI, start_loop=False) + assert load_authoring_lease(PACK) is None + + +def test_poll_wait_is_zero_not_twenty_five(tmp_path: Path) -> None: + assert POLL_WAIT_S == 0 + source = Path("engine/authoring_runner.py").read_text(encoding="utf-8") + assert "DEFAULT_WAIT_S" not in source + assert '"wait_seconds": POLL_WAIT_S' in source + assert "from openadapt_flow.backends.win_agent" not in source + assert "launch_agent(" not in source + runner, requests, _recorder, _audit = _mailbox(tmp_path) + runner.claim_uri(URI, start_loop=False) + runner.poll_once() + poll = next(item for item in requests if item.url.path.endswith("/poll")) + assert json.loads(poll.content)["wait_seconds"] == 0 + + +def test_bind_pack_allow_is_per_sub_and_required_for_halt(tmp_path: Path) -> None: + runner, requests, recorder, _audit = _mailbox(tmp_path) + events: list[tuple[str, dict[str, Any]]] = [] + runner.emit = lambda event, data: events.append((event, data)) + runner.claim_uri(URI, start_loop=False) + runner.handle_envelope(_envelope("bind_pack")) + assert any( + event == "authoring_state" and data["status"] == "pending_allow" + for event, data in events + ) + runner.handle_envelope(_envelope("observe")) + runner.handle_envelope(_envelope("halt")) + denied = [ + json.loads(item.content) + for item in requests + if item.url.path.endswith("/callback") + ] + assert {item["result"]["error"] for item in denied} == {"not_allowed"} + assert runner.allow()["allowed"] is True + runner.handle_envelope(_envelope("observe")) + observe = json.loads(requests[-1].content)["result"] + assert observe["schema_version"] == "openadapt.authoring.observe/v1" + assert "value" not in json.dumps(observe) + assert "title" not in json.dumps(observe) + assert "SSN-SECRET" not in json.dumps(observe) + runner.handle_envelope(_envelope("halt", sub=OTHER_SUB)) + assert json.loads(requests[-1].content)["result"]["error"] == "not_allowed" + runner.handle_envelope(_envelope("halt")) + assert json.loads(requests[-1].content)["result"] == {"halted": True} + assert recorder.clicks == [] + + +def test_unsigned_stop_uses_lease_bearer_not_mcp_jwt(tmp_path: Path) -> None: + runner, requests, _recorder, _audit = _mailbox(tmp_path) + runner.claim_uri(URI, start_loop=False) + runner.handle_envelope(_envelope("bind_pack")) + runner.allow() + runner.operator_stop() + callback = next(item for item in reversed(requests) if item.url.path.endswith("/callback")) + body = json.loads(callback.content) + assert body["halted"] is True + assert callback.headers["Authorization"] == f"Bearer {LEASE}" + + +def test_continue_records_observed_on_pause_target_never_type_text(tmp_path: Path) -> None: + runner, _requests, recorder, _audit = _mailbox(tmp_path) + runner.claim_uri(URI, start_loop=False) + runner.handle_envelope(_envelope("bind_pack")) + runner.allow() + runner.pin_target(backend="web", url="https://openadapt.ai/mockmed") + runner.handle_envelope(_envelope("start_record")) + observe = runner._observe() + note = next(node for node in observe["tree"] if node.get("automation_id") == "note") + runner.handle_envelope( + _envelope("pause_for_input", args={"node_id": note["node_id"], "param": "note"}) + ) + runner.continue_pause() + assert recorder.typed == [] + assert recorder.observed[0]["event"] == {"kind": "type"} + assert recorder.observed[0]["text"] == "follow up in two weeks" + assert "secret" not in recorder.observed[0] or recorder.observed[0].get("secret") is not True + + +def test_secret_continue_has_no_text_and_compile_refuses_if_missing(tmp_path: Path) -> None: + runner, _requests, recorder, _audit = _mailbox(tmp_path) + runner.claim_uri(URI, start_loop=False) + runner.handle_envelope(_envelope("bind_pack")) + runner.allow() + runner.pin_target(backend="macos") + runner.handle_envelope(_envelope("start_record")) + observe = runner._observe() + note = next(node for node in observe["tree"] if node.get("automation_id") == "note") + runner.handle_envelope( + _envelope( + "pause_for_input", + args={"node_id": note["node_id"], "param": "ssn", "secret": True}, + ) + ) + with pytest.raises(AuthoringError, match="secret_type_missing"): + runner._compile() + runner.continue_pause() + assert "text" not in recorder.observed[0] + assert recorder.observed[0]["secret"] is True + compiled = runner._compile() + assert compiled == { + "status": "needs_human_admit", + "workflow_id": "wf_mockmed", + "recording_retained": True, + } + assert compiled.get("success") is not True + + +def test_click_uses_backend_pixels_and_stale_node_does_not_retry(tmp_path: Path) -> None: + runner, _requests, recorder, _audit = _mailbox(tmp_path) + runner.claim_uri(URI, start_loop=False) + runner.handle_envelope(_envelope("bind_pack")) + runner.allow() + runner.pin_target(backend="macos") + runner.handle_envelope(_envelope("start_record")) + observe = runner._observe() + button = next(node for node in observe["tree"] if node.get("automation_id") == "btnContinue") + runner.handle_envelope(_envelope("click", args={"node_id": button["node_id"]})) + assert recorder.clicks == [(1010, 658)] + with pytest.raises(AuthoringError, match="stale_node"): + runner._click({"node_id": "n_deadbeef"}) + runner._uncertain = True + with pytest.raises(AuthoringError, match="RECONCILIATION_REQUIRED"): + runner._click({"node_id": button["node_id"]}) + assert recorder.clicks == [(1010, 658)] + + +def test_linux_without_a_unique_title_is_coach_only(tmp_path: Path) -> None: + runner, _requests, recorder, _audit = _mailbox(tmp_path) + runner.claim_uri(URI, start_loop=False) + runner.handle_envelope(_envelope("bind_pack")) + runner.allow() + runner.pin_target(backend="linux", window_title_unique=False) + observe = runner._observe() + assert observe["coach_only"] is True + with pytest.raises(AuthoringCoachOnly): + runner._start_record() + assert recorder.clicks == [] + + +def test_windows_native_is_coach_only_and_does_not_start_recorder(tmp_path: Path) -> None: + runner, _requests, recorder, _audit = _mailbox(tmp_path) + runner.claim_uri(URI, start_loop=False) + runner.handle_envelope(_envelope("bind_pack")) + runner.allow() + runner.pin_target(backend="windows") + observe = runner._observe() + assert observe["coach_only"] is True + assert observe["agent_drive"] is False + assert observe["tree"] == [] + with pytest.raises(AuthoringCoachOnly): + runner._start_record() + with pytest.raises(AuthoringCoachOnly): + runner._click({"node_id": "n_00000000"}) + assert recorder.clicks == [] + assert recorder.finished is False + + +def test_playwright_launcher_gets_the_desktop_url_not_an_mcp_string(tmp_path: Path) -> None: + launched: list[str] = [] + runner, _requests, _recorder, _audit = _mailbox(tmp_path) + runner._playwright_launcher = lambda url: launched.append(url) or SimpleNamespace( + cookies=lambda: [] + ) + runner.claim_uri(URI, start_loop=False) + runner.handle_envelope(_envelope("bind_pack")) + runner.allow() + runner.pin_target(backend="web", url="https://openadapt.ai/mockmed") + runner._start_record() + assert launched == ["https://openadapt.ai/mockmed"] + + +def test_node_table_is_mode_0600(tmp_path: Path) -> None: + runner, _requests, _recorder, _audit = _mailbox(tmp_path) + runner.claim_uri(URI, start_loop=False) + runner.handle_envelope(_envelope("bind_pack")) + runner.allow() + runner._observe() + assert runner._node_table is not None + mode = stat.S_IMODE(runner._node_table._path.stat().st_mode) + assert mode == 0o600 + + +def test_projector_drops_value_title_and_six_digit_names(tmp_path: Path) -> None: + from engine.authoring_runner import NodeTable + + table = NodeTable(tmp_path / "nodes.json", b"key") + payload = project_observe( + backend="web", + provider="playwright_ax", + recording=False, + agent_drive=True, + coach_only=False, + process_name="Chromium", + raw_nodes=[ + { + "provider_runtime_id": "x", + "role": "field", + "control_type": "edit", + "automation_id": "ssn-123456", + "name": "acct 1234567", + "value": "000-00-0000", + "title": "Secret", + "bounds": {"x": 0.1, "y": 0.1, "w": 0.1, "h": 0.1}, + "backend_pixels": {"x": 1, "y": 1, "w": 2, "h": 2}, + } + ], + node_table=table, + ) + dumped = json.dumps(payload) + assert "000-00-0000" not in dumped + assert "Secret" not in dumped + assert "ssn-123456" not in dumped + assert "acct 1234567" not in dumped + assert "value" not in dumped + assert payload["tree"][0]["node_id"].startswith("n_") + + +def test_dispatch_resume_and_stop_use_authoring_when_bound(tmp_path: Path) -> None: + from engine.db import IndexDB + from engine.dispatch import EngineDispatcher, EngineServices + + runner, _requests, recorder, _audit = _mailbox(tmp_path) + runner.claim_uri(URI, start_loop=False) + runner.handle_envelope(_envelope("bind_pack")) + runner.allow() + runner.pin_target(backend="macos") + runner.handle_envelope(_envelope("start_record")) + observe = runner._observe() + note = next(node for node in observe["tree"] if node.get("automation_id") == "note") + runner.handle_envelope( + _envelope("pause_for_input", args={"node_id": note["node_id"], "param": "note"}) + ) + db = IndexDB(tmp_path / "index.db") + db.initialize() + disp = EngineDispatcher( + runner.config, + services=EngineServices(runner.config, db=db, authoring=runner, audit=FakeAudit()), + ) + resumed = disp.dispatch("resume_recording", {}) + assert resumed["paused"] is False + assert recorder.typed == [] + assert recorder.observed + stopped = disp.dispatch("stop_recording", {}) + assert stopped["halted"] is True + db.close() + + +def test_replace_allow_required_for_a_second_sub(tmp_path: Path) -> None: + runner, _requests, _recorder, _audit = _mailbox(tmp_path) + runner.claim_uri(URI, start_loop=False) + runner.handle_envelope(_envelope("bind_pack")) + runner.allow() + runner.handle_envelope(_envelope("bind_pack", sub=OTHER_SUB)) + assert runner.status()["status"] == "replace_allow" + assert runner.allow()["allowed"] is True + assert runner._allowed_sub == SUB + runner.allow(replace=True) + assert runner._allowed_sub == OTHER_SUB diff --git a/tests/test_pairing_protocol_boundary.py b/tests/test_pairing_protocol_boundary.py index 0feee65..9f245c1 100644 --- a/tests/test_pairing_protocol_boundary.py +++ b/tests/test_pairing_protocol_boundary.py @@ -36,6 +36,7 @@ def test_python_pairing_action_has_no_shell_or_navigation_escape_hatch() -> None store = (ROOT / "engine/auth/store.py").read_text() dispatch = (ROOT / "engine/dispatch.py").read_text() assert '"connect_uri": self.connect_uri' in dispatch + assert '"claim_runner_uri": self.claim_runner_uri' in dispatch assert "subprocess" not in pairing assert "shell=" not in pairing assert "webbrowser" not in pairing From e8d6b45b2fb7b8e5f8bac9a22b0738a04d3e5850 Mon Sep 17 00:00:00 2001 From: abrichr Date: Tue, 1 Sep 2026 14:40:57 -0400 Subject: [PATCH 3/5] fix(desktop): close remaining authoring D1/D2 Allow, pause, and substrate holes Store mailbox allowedSub via POST /runner/allow. Hold pause_for_input until overlay Continue, then callback {recorded, param} with no field value. macOS/Linux default to COACH_ONLY until the window is unique; Playwright launches empty-cookie Chromium from a Desktop URL. Titles never go to MCP. --- engine/auth/runner_bind.py | 28 ++- engine/authoring_runner.py | 250 +++++++++++++++++---- engine/dispatch.py | 2 +- src-tauri/src/pairing.rs | 13 ++ src/App.tsx | 117 +++++++++- src/lib/types.ts | 1 + src/overlay/ControlOverlay.tsx | 14 +- src/overlay/state.test.ts | 18 ++ src/overlay/state.ts | 6 + src/styles/app.css | 11 + tests/test_engine/test_authoring_runner.py | 107 ++++++++- tests/test_pairing_protocol_boundary.py | 7 + 12 files changed, 506 insertions(+), 68 deletions(-) diff --git a/engine/auth/runner_bind.py b/engine/auth/runner_bind.py index 8cfb0fc..567d39f 100644 --- a/engine/auth/runner_bind.py +++ b/engine/auth/runner_bind.py @@ -20,8 +20,8 @@ PACK_CIPHER_RE = re.compile(r"^v1\.[A-Za-z0-9_-]{32,2000}$") CLOUD_RUNNER_TOKEN_RE = re.compile(r"^oar_[a-f0-9]{64}$") PAIRING_SECRET_RE = re.compile(r"^oap_[A-Za-z0-9_-]{43}$") -HEX_BODY_RE = re.compile(r"^[a-f0-9]+$") -UNRESERVED_BODY_RE = re.compile(r"^[A-Za-z0-9_-]+$") +BIND_HEX_BODY_RE = re.compile(r"^oab_[a-f0-9]{64}$") +LEASE_BASE64URL_BODY_RE = re.compile(r"^oals_[A-Za-z0-9_-]{43}$") class RunnerBindError(RuntimeError): @@ -31,25 +31,29 @@ class RunnerBindError(RuntimeError): def valid_bind_token(value: object) -> bool: """Return whether ``value`` is exactly one ``oab_`` bind token.""" - if not isinstance(value, str) or BIND_TOKEN_RE.fullmatch(value) is None: + if not isinstance(value, str): return False - body = value[4:] - # 32-byte hex (the ``oar_`` body) is not a bind token even with this prefix. - if HEX_BODY_RE.fullmatch(body) is not None and len(body) == 64: + if ( + CLOUD_RUNNER_TOKEN_RE.fullmatch(value) is not None + or PAIRING_SECRET_RE.fullmatch(value) is not None + or BIND_HEX_BODY_RE.fullmatch(value) is not None + ): return False - return True + return BIND_TOKEN_RE.fullmatch(value) is not None def valid_lease_secret(value: object) -> bool: """Return whether ``value`` is exactly one ``oals_`` mailbox lease secret.""" - if not isinstance(value, str) or LEASE_SECRET_RE.fullmatch(value) is None: + if not isinstance(value, str): return False - body = value[5:] - # 32-byte base64url (the ``oab_`` body) is not a lease secret. - if len(body) == 43 and UNRESERVED_BODY_RE.fullmatch(body) is not None: + if ( + CLOUD_RUNNER_TOKEN_RE.fullmatch(value) is not None + or PAIRING_SECRET_RE.fullmatch(value) is not None + or LEASE_BASE64URL_BODY_RE.fullmatch(value) is not None + ): return False - return True + return LEASE_SECRET_RE.fullmatch(value) is not None def valid_pack_id(value: object) -> bool: diff --git a/engine/authoring_runner.py b/engine/authoring_runner.py index a5d6690..256b87e 100644 --- a/engine/authoring_runner.py +++ b/engine/authoring_runner.py @@ -41,6 +41,7 @@ COMMAND_ENVELOPE_SCHEMA = "openadapt.authoring.command/v1" OBSERVE_SCHEMA = "openadapt.authoring.observe/v1" CLIENT_DISPLAYS = frozenset({"ChatGPT", "Claude"}) +PAUSE_PROMPT = "Type in the application. Continue here when done." ENQUEUE_REQUIRING_ALLOW = frozenset( { "observe", @@ -55,12 +56,33 @@ } ) COACH_ONLY_BACKENDS = frozenset({"windows", "rdp", "citrix"}) +UNIQUE_WINDOW_BACKENDS = frozenset({"macos", "linux"}) PROCESS_NAME_RE = re.compile(r"^[A-Za-z0-9 ._-]{1,64}$") SIX_DIGITS_RE = re.compile(r"\d{6,}") _SHA256_HEX = re.compile(r"^[a-f0-9]{64}$") _SAFE_PARAM = re.compile(r"^[A-Za-z0-9_]{1,40}$") _NODE_ID = re.compile(r"^n_[a-f0-9]{8}$") _COMMAND_ID = re.compile(r"^[A-Za-z0-9_.:-]{1,200}$") +FORBIDDEN_RESULT_KEYS = frozenset( + { + "value", + "text", + "title", + "screenshot", + "png", + "ocr", + "backend_pixels", + "pixels", + "events", + "window_title", + "image", + "raw", + "leaseSecret", + "lease_secret", + "bind", + } +) +MAILBOX_ACTIONS = frozenset({"claim", "poll", "callback", "allow"}) class AuthoringError(RuntimeError): @@ -83,6 +105,62 @@ def _sha256_hex(value: str) -> str: return hashlib.sha256(value.encode("utf-8")).hexdigest() +def _lease_hmac_key(lease_secret: str) -> bytes: + """Use the lease secret body as HMAC key material. Do not hash it as a password.""" + + if not valid_lease_secret(lease_secret): + raise AuthoringError("The authoring mailbox credential is malformed.") + return bytes.fromhex(lease_secret[5:]) + + +def _sanitize_result(value: Any) -> Any: + """Drop titles, values, pixels, and other vendor-forbidden keys.""" + + if isinstance(value, dict): + return { + key: _sanitize_result(child) + for key, child in value.items() + if key not in FORBIDDEN_RESULT_KEYS + } + if isinstance(value, list): + return [_sanitize_result(item) for item in value] + return value + + +def _require_empty_cookies(browser: Any) -> None: + cookies_fn = getattr(browser, "cookies", None) + if cookies_fn is None: + context = getattr(browser, "context", None) + cookies_fn = getattr(context, "cookies", None) + if not callable(cookies_fn): + raise AuthoringError("Playwright Chromium did not start with empty cookies.") + cookies = cookies_fn() + if cookies: + raise AuthoringError("Playwright Chromium did not start with empty cookies.") + + +def launch_empty_playwright_chromium(url: str) -> Any: + """Launch a fresh Chromium with empty cookies. Never attach to logged-in Chrome.""" + + if not isinstance(url, str) or not url.startswith("https://"): + raise AuthoringError("A Playwright job needs a URL typed into Desktop.") + try: + from playwright.sync_api import sync_playwright + except ImportError as exc: + raise AuthoringError("Playwright Chromium is unavailable.") from exc + playwright = sync_playwright().start() + browser = playwright.chromium.launch(headless=False) + context = browser.new_context() + cookies = context.cookies() + if cookies: + browser.close() + playwright.stop() + raise AuthoringError("Playwright Chromium did not start with empty cookies.") + page = context.new_page() + page.goto(url) + return page + + def _pack_dir(data_dir: Path, pack_id: str) -> Path: return Path(data_dir) / "authoring" / _sha256_hex(pack_id)[:16] @@ -323,7 +401,7 @@ def close(self) -> None: self._client.close() def _path(self, pack_id: str, action: str) -> str: - if not valid_pack_id(pack_id) or action not in {"claim", "poll", "callback"}: + if not valid_pack_id(pack_id) or action not in MAILBOX_ACTIONS: raise AuthoringTransportError("The authoring mailbox path is invalid.") return f"/j/{quote(pack_id, safe='._-')}/runner/{action}" @@ -440,6 +518,22 @@ def callback( expected=(200, 202), ) + def allow(self, pack_id: str, lease_secret: str, command_id: str) -> None: + if not valid_lease_secret(lease_secret): + raise AuthoringTransportError("The authoring mailbox credential is malformed.") + if not isinstance(command_id, str) or _COMMAND_ID.fullmatch(command_id) is None: + raise AuthoringTransportError("The authoring Allow request is malformed.") + path = self._path(pack_id, "allow") + self._post( + path, + {"command_id": command_id}, + headers={ + "Authorization": f"Bearer {lease_secret}", + "Content-Type": "application/json", + }, + expected=(200, 202), + ) + class AuthoringRunner: """Claim, Allow-per-sub, wait=0 poll, and Flow record_observed session.""" @@ -457,6 +551,7 @@ def __init__( compile_recording: Callable[..., Any] | None = None, playwright_launcher: Callable[[str], Any] | None = None, text_value_at: Callable[[dict[str, int]], str | None] | None = None, + unique_window: Callable[[], dict[str, Any] | None] | None = None, ) -> None: self.config = config self.emit = emit or (lambda _event, _data: None) @@ -468,6 +563,7 @@ def __init__( self._compile_recording = compile_recording self._playwright_launcher = playwright_launcher self._text_value_at = text_value_at + self._unique_window = unique_window self._transport: AuthoringMailboxTransport | None = None self._lock = threading.RLock() self._stop = threading.Event() @@ -477,12 +573,14 @@ def __init__( self._allowed_sub: str | None = None self._allowed_client_id: str | None = None self._pending_allow: dict[str, Any] | None = None - self._pin: dict[str, Any] = {"backend": "macos"} + self._pin: dict[str, Any] = {"backend": "macos", "window_title_unique": False} self._node_table: NodeTable | None = None self._recorder: Any | None = None self._recording = False self._paused = False self._pause_target: dict[str, Any] | None = None + self._pause_command_id: str | None = None + self._active_command_id: str | None = None self._secret_pause = False self._secret_type_recorded = False self._actuation_started = False @@ -505,6 +603,7 @@ def status_dict(self) -> dict[str, Any] | None: "recording": True, "paused": True, "capture_id": None, + "pause_prompt": PAUSE_PROMPT, "controls": {"pause": False, "resume": True, "stop": True}, } if self._recording: @@ -541,17 +640,39 @@ def status(self) -> dict[str, Any]: def pin_target(self, **fields: Any) -> dict[str, Any]: backend = str(fields.get("backend") or self._pin.get("backend") or "macos") + unique = fields.get("window_title_unique") + macos_app = fields.get("macos_app") + macos_title = fields.get("macos_window_title") + linux_app = fields.get("linux_app") + linux_title = fields.get("linux_window_title") + if fields.get("use_frontmost") is True: + probe = self._unique_window() if self._unique_window is not None else None + if not isinstance(probe, dict) or probe.get("unique") is not True: + unique = False + else: + unique = True + backend = str(probe.get("backend") or backend) + macos_app = probe.get("process_name") or macos_app + macos_title = probe.get("window_title") or macos_title + linux_app = probe.get("process_name") or linux_app + linux_title = probe.get("window_title") or linux_title + if unique is None: + unique = False if backend in UNIQUE_WINDOW_BACKENDS else True pin = { "backend": backend, - "url": fields.get("url"), - "macos_app": fields.get("macos_app"), - "macos_window_title": fields.get("macos_window_title"), - "linux_app": fields.get("linux_app"), - "linux_window_title": fields.get("linux_window_title"), - "window_title_unique": fields.get("window_title_unique", True), + "url": fields.get("url") if backend == "web" else None, + "macos_app": macos_app, + "macos_window_title": macos_title, + "linux_app": linux_app, + "linux_window_title": linux_title, + "window_title_unique": bool(unique), } self._pin = pin - return {"ok": True, "backend": backend} + return { + "ok": True, + "backend": backend, + "coach_only": self._coach_only(), + } def claim_uri(self, uri: str, *, start_loop: bool = True) -> dict[str, Any]: parsed = parse_runner_uri(uri) @@ -592,7 +713,7 @@ def claim_uri(self, uri: str, *, start_loop: bool = True) -> dict[str, Any]: self._transport = transport self._pack = pack self._lease_secret = lease_secret - hmac_key = hashlib.sha256(lease_secret.encode("utf-8")).digest() + hmac_key = _lease_hmac_key(lease_secret) self._node_table = NodeTable( _pack_dir(self.config.data_dir, pack) / "nodes.json", hmac_key, @@ -602,7 +723,7 @@ def claim_uri(self, uri: str, *, start_loop: bool = True) -> dict[str, Any]: self.audit.log("authoring_bind_claimed", pack_hash=_sha256_hex(pack)[:16]) if start_loop: self.start() - self.emit("authoring_state", {"status": "bound"}) + self.emit("authoring_state", self.status()) return {"bound": True, "origin": origin, "pack_prefix": pack[:2]} def start(self) -> None: @@ -657,6 +778,14 @@ def handle_envelope(self, envelope: dict[str, Any]) -> None: if pack_id != self._pack: self._callback_error(command_id, "pack_mismatch") return + if command_id in {self._active_command_id, self._pause_command_id}: + return + if ( + tool == "bind_pack" + and self._pending_allow + and self._pending_allow.get("command_id") == command_id + ): + return sub = envelope.get("oauth_sub_sha256") if tool == "bind_pack": self._queue_allow(envelope) @@ -665,9 +794,11 @@ def handle_envelope(self, envelope: dict[str, Any]) -> None: self._callback_error(command_id, "not_allowed") return args = envelope.get("args") if isinstance(envelope.get("args"), dict) else {} + self._active_command_id = command_id try: result = self._dispatch_tool(str(tool), args) except AuthoringCoachOnly: + self._active_command_id = None self._callback( { "command_id": command_id, @@ -677,8 +808,13 @@ def handle_envelope(self, envelope: dict[str, Any]) -> None: ) return except AuthoringError as exc: + self._active_command_id = None self._callback_error(command_id, str(exc)) return + if tool == "pause_for_input": + self._pause_command_id = command_id + return + self._active_command_id = None self._callback({"command_id": command_id, "status": "done", "result": result}) def _queue_allow(self, envelope: dict[str, Any]) -> None: @@ -724,6 +860,15 @@ def allow(self, *, replace: bool = False) -> dict[str, Any]: and not replace ): return self.status() + command_id = pending.get("command_id") + display = pending.get("client_display") + if ( + isinstance(command_id, str) + and self._transport is not None + and self._pack + and self._lease_secret + ): + self._transport.allow(self._pack, self._lease_secret, command_id) granted_at = _utc_now() self._allowed_sub = pending["oauth_sub_sha256"] self._allowed_client_id = pending.get("client_id_sha256") @@ -733,17 +878,7 @@ def allow(self, *, replace: bool = False) -> dict[str, Any]: stored["allowed_client_id"] = self._allowed_client_id stored["allowed_at"] = granted_at store_authoring_lease(self._pack or "", stored) - command_id = pending.get("command_id") - display = pending.get("client_display") self._pending_allow = None - if isinstance(command_id, str): - self._callback( - { - "command_id": command_id, - "status": "done", - "result": {"allowed": True}, - } - ) if self.audit: self.audit.log( "authoring_allowed", @@ -751,7 +886,7 @@ def allow(self, *, replace: bool = False) -> dict[str, Any]: allowed_sub_prefix=(self._allowed_sub or "")[:8], client_display=display, ) - self.emit("authoring_state", {"status": "bound", "allowed": True}) + self.emit("authoring_state", self.status()) return {"allowed": True, "client_display": display} def deny(self) -> dict[str, Any]: @@ -777,7 +912,12 @@ def _forbidden(*_args: Any, **_kwargs: Any) -> None: else: original = None try: + text = None + if self._text_value_at is not None: + text = self._text_value_at(target["backend_pixels"]) if target.get("secret"): + if text is not None and not text: + return self.status_dict() or {"recording": True, "paused": True} recorder.record_observed( event={"kind": "type"}, param=target.get("param"), @@ -786,9 +926,6 @@ def _forbidden(*_args: Any, **_kwargs: Any) -> None: ) self._secret_type_recorded = True else: - text = None - if self._text_value_at is not None: - text = self._text_value_at(target["backend_pixels"]) recorder.record_observed( event={"kind": "type"}, param=target.get("param"), @@ -797,8 +934,11 @@ def _forbidden(*_args: Any, **_kwargs: Any) -> None: finally: if original is not None: recorder.type_text = original + command_id = self._pause_command_id self._paused = False self._pause_target = None + self._pause_command_id = None + self._active_command_id = None self.emit("status_update", self.status_dict() or {}) if self.audit: self.audit.log( @@ -806,6 +946,14 @@ def _forbidden(*_args: Any, **_kwargs: Any) -> None: param=target.get("param"), secret=bool(target.get("secret")), ) + if isinstance(command_id, str): + self._callback( + { + "command_id": command_id, + "status": "done", + "result": {"recorded": True, "param": target.get("param")}, + } + ) return self.status_dict() or {} def operator_stop(self) -> dict[str, Any]: @@ -840,7 +988,7 @@ def _coach_only(self) -> bool: backend = str(self._pin.get("backend") or "") if backend in COACH_ONLY_BACKENDS: return True - if backend == "linux" and not self._pin.get("window_title_unique"): + if backend in UNIQUE_WINDOW_BACKENDS and not self._pin.get("window_title_unique"): return True return False @@ -856,13 +1004,20 @@ def _observe(self) -> dict[str, Any]: "macos": "ax", "linux": "atspi", }.get(backend, "none") + process_name = None + if backend == "web": + process_name = "Chromium" + elif backend == "macos": + process_name = self._pin.get("macos_app") + elif backend == "linux": + process_name = self._pin.get("linux_app") return project_observe( backend=backend, provider=provider, recording=self._recording, agent_drive=agent_drive, coach_only=coach_only, - process_name="Chromium" if backend == "web" else None, + process_name=process_name if isinstance(process_name, str) else None, raw_nodes=raw, node_table=self._node_table, ) @@ -877,8 +1032,9 @@ def _start_record(self) -> dict[str, Any]: url = self._pin.get("url") if not isinstance(url, str) or not url.startswith("https://"): raise AuthoringError("A Playwright job needs a URL typed into Desktop.") - if self._playwright_launcher is not None: - self._playwright = self._playwright_launcher(url) + launcher = self._playwright_launcher or launch_empty_playwright_chromium + self._playwright = launcher(url) + _require_empty_cookies(self._playwright) factory = self._recorder_factory if factory is None: raise AuthoringError("Authoring recorder is unavailable.") @@ -894,8 +1050,9 @@ def _start_record(self) -> dict[str, Any]: def _click(self, args: dict[str, Any]) -> dict[str, Any]: if self._coach_only(): raise AuthoringCoachOnly("COACH_ONLY") - if self._uncertain: - raise AuthoringError("RECONCILIATION_REQUIRED") + with self._lock: + if self._uncertain: + raise AuthoringError("RECONCILIATION_REQUIRED") node_id = args.get("node_id") if not isinstance(node_id, str) or _NODE_ID.fullmatch(node_id) is None: raise AuthoringError("stale_node") @@ -907,14 +1064,17 @@ def _click(self, args: dict[str, Any]) -> dict[str, Any]: pixels = row["backend_pixels"] x = int(pixels["x"] + pixels["w"] / 2) y = int(pixels["y"] + pixels["h"] / 2) - self._actuation_started = True + with self._lock: + self._actuation_started = True try: self._recorder.click(x, y) except Exception: - self._uncertain = True + with self._lock: + self._uncertain = True raise AuthoringError("RECONCILIATION_REQUIRED") from None finally: - self._actuation_started = False + with self._lock: + self._actuation_started = False return {"clicked": True} def _pause_for_input(self, args: dict[str, Any]) -> dict[str, Any]: @@ -975,8 +1135,19 @@ def _compile(self) -> dict[str, Any]: } def _halt(self, *, unsigned: bool, command_id: str | None) -> None: - if self._actuation_started: - self._uncertain = True + with self._lock: + if self._actuation_started: + self._uncertain = True + uncertain = True + else: + uncertain = False + self._recording = False + self._paused = False + self._recorder = None + self._pause_target = None + self._pause_command_id = None + self._active_command_id = None + if uncertain: self._callback( { "command_id": command_id, @@ -985,9 +1156,6 @@ def _halt(self, *, unsigned: bool, command_id: str | None) -> None: } ) return - self._recording = False - self._paused = False - self._recorder = None if self._node_table is not None: self._node_table.clear() if unsigned and self._pack and self._lease_secret and self._transport: @@ -1006,6 +1174,8 @@ def _callback(self, payload: dict[str, Any]) -> None: for key in ("command_id", "status", "result", "halted") if key in payload } + if "result" in closed: + closed["result"] = _sanitize_result(closed["result"]) self._transport.callback(self._pack, self._lease_secret, closed) def _callback_error(self, command_id: object, error: str) -> None: @@ -1040,7 +1210,7 @@ def restore_authoring_runner( runner._lease_secret = stored["lease_secret"] runner._allowed_sub = stored.get("allowed_sub") runner._allowed_client_id = stored.get("allowed_client_id") - hmac_key = hashlib.sha256(stored["lease_secret"].encode("utf-8")).digest() + hmac_key = _lease_hmac_key(stored["lease_secret"]) runner._node_table = NodeTable(_pack_dir(config.data_dir, pack_id) / "nodes.json", hmac_key) runner._transport = AuthoringMailboxTransport( origin=stored["origin"], diff --git a/engine/dispatch.py b/engine/dispatch.py index ff2e8fd..fc91fa9 100644 --- a/engine/dispatch.py +++ b/engine/dispatch.py @@ -3296,7 +3296,7 @@ def claim_runner_uri(self, **params: Any) -> dict: if not isinstance(uri, str): raise ValueError("uri is required") result = self._authoring_service().claim_uri(uri) - self.emit("authoring_state", {"status": "bound"}) + self.emit("authoring_state", self._authoring_service().status()) return result def authoring_allow(self, **params: Any) -> dict: diff --git a/src-tauri/src/pairing.rs b/src-tauri/src/pairing.rs index d9c9b41..c83557c 100644 --- a/src-tauri/src/pairing.rs +++ b/src-tauri/src/pairing.rs @@ -257,9 +257,22 @@ fn valid_pairing_secret(value: &str) -> bool { } fn valid_bind_token(value: &str) -> bool { + if value.starts_with("oar_") || value.starts_with("oap_") { + return false; + } + // Cloud runner bodies are 64 hex. That encoding is never a bind token. + if value.len() == 68 && value.starts_with("oab_") && hex_body(&value[4..]) { + return false; + } value.len() == 47 && value.starts_with("oab_") && unreserved_body(&value[4..]) } +fn hex_body(value: &str) -> bool { + value + .bytes() + .all(|byte| matches!(byte, b'0'..=b'9' | b'a'..=b'f')) +} + fn valid_pack_id(value: &str) -> bool { if let Some(body) = value.strip_prefix("p.") { return body.len() == 12 && unreserved_body(body); diff --git a/src/App.tsx b/src/App.tsx index 2dad928..9386235 100644 --- a/src/App.tsx +++ b/src/App.tsx @@ -175,7 +175,10 @@ export default function App() { client_display?: string; prompt?: string; error?: string; + allowed?: boolean; + coach_only?: boolean; } | null>(null); + const [authoringUrl, setAuthoringUrl] = useState(""); const [firstRunPersistencePending, setFirstRunPersistencePending] = useState(false); const [firstWorkflowRunning, setFirstWorkflowRunning] = useState(false); @@ -257,6 +260,17 @@ export default function App() { setBreaks(na.count); const ss = await engineTry(CMD.GET_SYNC_STATE, {}, sync); setSync(ss); + const authoringStatus = await engineTry<{ + status: string; + client_display?: string; + prompt?: string; + error?: string; + allowed?: boolean; + coach_only?: boolean; + }>(CMD.AUTHORING_STATUS, {}, { status: "idle" }); + if (authoringStatus.status && authoringStatus.status !== "idle") { + setAuthoring(authoringStatus); + } setCheckedAuth(true); })(); @@ -325,6 +339,8 @@ export default function App() { client_display?: string; prompt?: string; error?: string; + allowed?: boolean; + coach_only?: boolean; }) => { setAuthoring(state); }, @@ -362,7 +378,8 @@ export default function App() { authoring && (authoring.status === "pending_allow" || authoring.status === "replace_allow" || - authoring.status === "error") ? ( + authoring.status === "error" || + authoring.status === "bound") ? (
{authoring.status === "error" ? authoring.error || "The authoring bind could not be completed." - : authoring.prompt || - (authoring.status === "replace_allow" - ? `A different ${authoring.client_display || "ChatGPT"} account is asking. Allow it to replace the current one?` - : `Allow ${authoring.client_display || "ChatGPT"} to drive this job`)} + : authoring.status === "bound" + ? authoring.coach_only + ? "This job is coach-only on this window. ChatGPT can suggest; you click." + : authoring.allowed + ? "Pin the browser URL or use this window. Titles stay on this computer." + : "This computer is bound. Pin the window, then Allow ChatGPT to drive this job." + : authoring.prompt || + (authoring.status === "replace_allow" + ? `A different ${authoring.client_display || "ChatGPT"} account is asking. Allow it to replace the current one?` + : `Allow ${authoring.client_display || "ChatGPT"} to drive this job`)} - {authoring.status !== "error" && ( + {(authoring.status === "pending_allow" || + authoring.status === "replace_allow") && ( )} + {authoring.status === "bound" && ( + + setAuthoringUrl(event.target.value)} + placeholder="https://" + type="url" + value={authoringUrl} + /> + + + + )} {authoring.status === "error" && (