diff --git a/cli.py b/cli.py index 86980375fc..dc45f5b837 100644 --- a/cli.py +++ b/cli.py @@ -3015,12 +3015,6 @@ class HermesCLI(CLIProcessNotificationsMixin, CLIAgentSetupMixin, CLICommandsMix set_unlock_prompt_callback(self._vault_unlock_callback) set_save_login_prompt_callback(self._vault_save_login_callback) set_code_prompt_callback(self._vault_code_callback) - try: - from tools.computer_use_tool import set_approval_callback as _set_cu_cb - - _set_cu_cb(self._computer_use_approval_callback) - except ImportError: - pass self._tool_callbacks_installed = True def _ensure_tirith_security(self) -> None: diff --git a/hermes_cli/cli_modal_mixin.py b/hermes_cli/cli_modal_mixin.py index f1eb50ba93..c2c940da0d 100644 --- a/hermes_cli/cli_modal_mixin.py +++ b/hermes_cli/cli_modal_mixin.py @@ -815,20 +815,6 @@ class CLIModalMixin: choices.append("view") return choices - def _computer_use_approval_callback(self, action: str, args: dict, summary: str) -> str: - """Adapt the generic approval UI (once/session/always/deny) to the computer_use verdicts - (approve_once/approve_session/always_approve/deny).""" - verdict = self._approval_callback( - command=f"computer_use: {summary}", - description=f"Allow computer_use to perform `{action}`?") - return { - "once": "approve_once", - "session": "approve_session", - "always": "always_approve", - "deny": "deny", - "timeout": "timeout", - }.get(verdict, "deny") - def _handle_approval_selection(self) -> None: """Process the currently selected dangerous-command approval choice.""" state = self._approval_state diff --git a/tests/conftest.py b/tests/conftest.py index 62201be60a..f69c015404 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1766,25 +1766,15 @@ def _audio_playback_guard(request, monkeypatch): @pytest.fixture(autouse=True) def _isolate_computer_use_approval_state(): - """Reset computer-use approval globals after every test. + """Reset the computer-use explicit approval callback after every test. - ``tools.computer_use.tool`` keeps three module-globals for the CLI - approval flow: ``_approval_callback`` (set by the CLI console on init) - plus the per-session unlock stores ``_always_allow`` / - ``_session_auto_approve``. A test that installs a callback — or drives - CLI init far enough that the real one is registered — and does not reset - it poisons every later computer-use test in the same process: - - * a leaked callback that raises (dead UI/queue infra, or a stale - two-argument signature — the real contract is ``(action, args, - summary)``) turns into ``verdict = "deny"`` in ``_request_approval``, - so dispatch tests fail with an empty backend call list; - * a leaked callback that blocks (the real CLI one waits on an answer - queue) hangs the whole single-process run forever — pytest-timeout is - the only thing that can cut it. - - Both symptoms are order-dependent: the affected files pass in isolation - and only fail in full-suite runs. Teardown-only, so tests that install + ``tools.computer_use.tool._approval_callback`` is a module-global handed to + the shared approval gate as its explicit callback, where it takes precedence + over the per-thread terminal one. A test that installs it and does not + reset it poisons every later computer-use test in the same process: a + leaked callback that raises becomes a deny, a leaked one that blocks (the + real CLI one waits on an answer queue) hangs the whole single-process run. + Both symptoms are order-dependent. Teardown-only, so tests that install their own callback keep it for their own duration. """ yield @@ -1792,9 +1782,6 @@ def _isolate_computer_use_approval_state(): from tools.computer_use import tool as _cu_tool _cu_tool.set_approval_callback(None) - with _cu_tool._approval_lock: - _cu_tool._always_allow.clear() - _cu_tool._session_auto_approve.clear() except Exception: pass diff --git a/tests/tools/conftest.py b/tests/tools/conftest.py index f5c2c431fd..a12be220a0 100644 --- a/tests/tools/conftest.py +++ b/tests/tools/conftest.py @@ -104,6 +104,23 @@ def register_all_web_providers(): register_provider(cls()) +@pytest.fixture +def grant_computer_use_approvals(monkeypatch): + """Answer every computer_use approval prompt with "once" through the shared gate. + + computer_use fails CLOSED when nobody can answer (no interactive user, no + gateway), so dispatch tests that only care about routing must present an + interactive CLI with a granting callback. "once" persists nothing, so no + grant leaks into ``tools.approval``'s session/permanent stores. + """ + from tools.computer_use import tool as cu_tool + + monkeypatch.setenv("HERMES_INTERACTIVE", "1") + cu_tool.set_approval_callback(lambda command, description, **kw: "once") + yield + cu_tool.set_approval_callback(None) + + @pytest.fixture def web_registry_populated(): """Populate the web-search-provider registry for one test, then reset.""" diff --git a/tests/tools/test_computer_use.py b/tests/tools/test_computer_use.py index 9e39e502e7..65dce0aead 100644 --- a/tests/tools/test_computer_use.py +++ b/tests/tools/test_computer_use.py @@ -18,8 +18,9 @@ import pytest # --------------------------------------------------------------------------- @pytest.fixture(autouse=True) -def _reset_backend(): - """Tear down the cached backend between tests.""" +def _reset_backend(grant_computer_use_approvals): + """Tear down the cached backend between tests; destructive actions get an interactive "once" + through the shared approval gate (the tool fails closed with nobody to ask).""" from tools.computer_use.tool import reset_backend_for_tests reset_backend_for_tests() # Force the noop backend. diff --git a/tests/tools/test_computer_use_approval_isolation.py b/tests/tools/test_computer_use_approval_isolation.py index 06ce41509b..1c1e3f9aae 100644 --- a/tests/tools/test_computer_use_approval_isolation.py +++ b/tests/tools/test_computer_use_approval_isolation.py @@ -1,17 +1,22 @@ -"""Regression: leaked approval callbacks must not poison later tests. +"""computer_use approval is the shared ``tools.approval`` gate — no private grant store, no default-allow. -``tools.computer_use.tool._approval_callback`` and the per-session unlock -stores are module-globals. Without the autouse reset fixture in -``tests/conftest.py``, a test that installs a callback and "forgets" it -changes the behavior of every later computer-use test in the process: -a raising callback becomes ``verdict = "deny"`` (dispatch tests see an -empty backend call list), a blocking callback hangs the run. The pair -below simulates the forgetful test and asserts the next test still sees -default-allow behavior. +Two contracts: + +* With nobody able to answer (no interactive CLI, no gateway, yolo off) a destructive action is REFUSED and + never reaches the backend; under yolo it runs. Historically the tool default-allowed whenever no CLI callback + was wired, which made every headless host (cron, api_server, tui_gateway, gateway turns) run desktop input + ungated. +* A grant answered through computer_use lives in ``tools.approval``'s store under computer_use's own scope key, + so ``is_approved`` sees it and ``clear_session`` retires it like any terminal pattern. + +A leaked callback still poisons later tests (a raising one becomes deny, a blocking one hangs), so the autouse +reset in ``tests/conftest.py`` stays and the polluter/observer pair below keeps proving it. """ import json +import pytest + def _install_backend(cu_tool): class _RecordingBackend: @@ -47,29 +52,79 @@ def _install_backend(cu_tool): return backend -def test_a_forgets_a_poisoned_approval_callback(): - """Simulates the polluter: installs a callback with the LEGACY - two-argument signature and deliberately does not reset it.""" - from tools.computer_use import tool as cu_tool +@pytest.fixture +def _nobody_to_ask(monkeypatch): + """No interactive CLI, no gateway, no per-thread terminal callback, yolo off.""" + from tools import approval - def stale_two_arg_callback(action, args): # wrong arity on purpose - return "approve_once" - - cu_tool.set_approval_callback(stale_two_arg_callback) - # no reset — the autouse fixture must clean this up + for name in ("HERMES_INTERACTIVE", "HERMES_GATEWAY_SESSION", "HERMES_EXEC_ASK", "HERMES_YOLO_MODE"): + monkeypatch.delenv(name, raising=False) + monkeypatch.setattr(approval, "_YOLO_MODE_FROZEN", False) + monkeypatch.setattr("tools.terminal_tool._get_approval_callback", lambda: None) + yield -def test_b_still_dispatches_with_default_allow(): - """Without the isolation fixture this fails: the stale callback raises - (arity), ``_request_approval`` converts that into a deny, and the - backend never sees the click.""" +def test_no_callback_refuses_unless_yolo(_nobody_to_ask, monkeypatch): + """Fail closed: with no human reachable the click is blocked and the backend sees nothing; yolo lets it run.""" + from tools import approval from tools.computer_use import tool as cu_tool backend = _install_backend(cu_tool) + result = json.loads(cu_tool.handle_computer_use({"action": "click", "element": 3})) + assert result["error"].startswith("BLOCKED"), result + assert result["action"] == "click" + assert backend.calls == [] + + monkeypatch.setattr(approval, "_YOLO_MODE_FROZEN", True) result = cu_tool.handle_computer_use({"action": "click", "element": 3}) - call_names = [c[0] for c in backend.calls] - assert "click" in call_names, ( - f"leaked approval callback poisoned this test: {result!r}" - ) - payload = json.loads(result) if isinstance(result, str) else result - assert not (isinstance(payload, dict) and payload.get("error")) + assert [name for name, _ in backend.calls] == ["click"], result + + +def test_always_grant_lands_in_the_shared_store(monkeypatch): + """One grant store: an "always" answered through computer_use is what ``tools.approval.is_approved`` reports + for the same session and ``cua::`` key, and the next call is served from that store.""" + from tools import approval + from tools.approval_context import reset_current_session_key, set_current_session_key + from tools.computer_use import tool as cu_tool + + monkeypatch.setenv("HERMES_INTERACTIVE", "1") + monkeypatch.setattr(approval, "_YOLO_MODE_FROZEN", False) + monkeypatch.setattr(approval, "save_permanent_allowlist", lambda patterns: None) + prompts = [] + cu_tool.set_approval_callback(lambda command, description, **kw: prompts.append(command) or "always") + token = set_current_session_key("cua-grant-session") + try: + assert not approval.is_approved("cua-grant-session", "cua:click:background") + assert cu_tool._request_approval("click", {"element": 3}) is None + assert approval.is_approved("cua-grant-session", "cua:click:background") + assert cu_tool._request_approval("click", {"element": 3}) is None + assert len(prompts) == 1 + finally: + cu_tool.set_approval_callback(None) + reset_current_session_key(token) + approval.clear_session("cua-grant-session") + with approval._lock: + approval._permanent_set().discard("cua:click:background") + + +def test_a_forgets_a_poisoned_approval_callback(): + """Simulates the polluter: installs a raising callback and deliberately does not reset it.""" + from tools.computer_use import tool as cu_tool + + def poisoned(command, description, **kw): + raise RuntimeError("dead UI") + + cu_tool.set_approval_callback(poisoned) + # no reset — the autouse fixture must clean this up + + +def test_b_still_dispatches_after_the_polluter(monkeypatch): + """Answers through the per-thread terminal callback only. The explicit computer_use callback takes precedence + in the shared gate, so if the polluter's raising one had leaked, this click would be denied.""" + from tools.computer_use import tool as cu_tool + + monkeypatch.setenv("HERMES_INTERACTIVE", "1") + monkeypatch.setattr("tools.terminal_tool._get_approval_callback", lambda: lambda command, description, **kw: "once") + backend = _install_backend(cu_tool) + result = cu_tool.handle_computer_use({"action": "click", "element": 3}) + assert [name for name, _ in backend.calls] == ["click"], f"leaked approval callback poisoned this test: {result!r}" diff --git a/tests/tools/test_computer_use_cua_0_10_permissions.py b/tests/tools/test_computer_use_cua_0_10_permissions.py index 7b968d1474..5d35b9f7eb 100644 --- a/tests/tools/test_computer_use_cua_0_10_permissions.py +++ b/tests/tools/test_computer_use_cua_0_10_permissions.py @@ -143,15 +143,11 @@ def test_release_seam_stops_backend_and_clears_session_state(): computer_use._backends["session-a"] = backend computer_use._backend_call_locks["session-a"] = computer_use.threading.RLock() computer_use._backend_permission_modes["session-a"] = "unrestricted" - computer_use._session_auto_approve["session-a"] = True - computer_use._always_allow["session-a"] = {("click", "background")} assert computer_use.release_computer_use_session("session-a") is True assert computer_use.release_computer_use_session("session-a") is False backend.stop.assert_called_once_with() assert "session-a" not in computer_use._backend_permission_modes - assert "session-a" not in computer_use._session_auto_approve - assert "session-a" not in computer_use._always_allow def test_yolo_toggle_immediately_releases_mode_dependent_backend(): diff --git a/tests/tools/test_computer_use_cua_0_9.py b/tests/tools/test_computer_use_cua_0_9.py index 76fb3cdfbc..f64f60d270 100644 --- a/tests/tools/test_computer_use_cua_0_9.py +++ b/tests/tools/test_computer_use_cua_0_9.py @@ -259,10 +259,6 @@ def test_release_seam_stops_exact_backend_and_clears_session_state(): "conversation-a": computer_use.threading.RLock(), "conversation-b": computer_use.threading.RLock(), }) - computer_use._session_auto_approve["conversation-a"] = True - computer_use._always_allow["conversation-a"] = { - ("click", "background"), - } assert computer_use.release_computer_use_session("conversation-a") is True assert computer_use.release_computer_use_session("conversation-a") is False @@ -271,8 +267,6 @@ def test_release_seam_stops_exact_backend_and_clears_session_state(): second.stop.assert_not_called() assert "conversation-a" not in computer_use._backends assert "conversation-a" not in computer_use._backend_call_locks - assert "conversation-a" not in computer_use._session_auto_approve - assert "conversation-a" not in computer_use._always_allow assert computer_use._backends["conversation-b"] is second @@ -283,12 +277,10 @@ def test_release_seam_evicts_state_even_when_backend_stop_fails(): backend.stop.side_effect = RuntimeError("driver teardown failed") computer_use._backends["failed-run"] = backend computer_use._backend_call_locks["failed-run"] = computer_use.threading.RLock() - computer_use._session_auto_approve["failed-run"] = True assert computer_use.release_computer_use_session("failed-run") is True assert "failed-run" not in computer_use._backends assert "failed-run" not in computer_use._backend_call_locks - assert "failed-run" not in computer_use._session_auto_approve def test_release_seam_waits_for_in_flight_action_before_stopping_backend(): @@ -359,15 +351,18 @@ def test_concurrent_hermes_sessions_do_not_share_backend_state(): assert len(created) == 2 -def test_persistent_focus_has_a_separate_approval_scope(): +def test_persistent_focus_has_a_separate_approval_scope(monkeypatch): from tools.computer_use import tool as computer_use seen = [] - def approve(action, args, summary): + def approve(command, description, **kw): + # The shared gate prompts once per scope key: the click itself, then the separate bring_to_front scope. + action = description.split("`")[1] seen.append(action) - return "approve_once" if action == "click" else "deny" + return "once" if action == "click" else "deny" + monkeypatch.setenv("HERMES_INTERACTIVE", "1") computer_use.set_approval_callback(approve) try: result = json.loads( @@ -385,6 +380,6 @@ def test_persistent_focus_has_a_separate_approval_scope(): computer_use.set_approval_callback(None) assert seen == ["click", "bring_to_front"] - assert result["error"] == "denied by user" + assert result["error"].startswith("BLOCKED: User denied") assert result["action"] == "bring_to_front" diff --git a/tests/tools/test_computer_use_delivery_ladder.py b/tests/tools/test_computer_use_delivery_ladder.py index 1d20acdff2..cc1c415a87 100644 --- a/tests/tools/test_computer_use_delivery_ladder.py +++ b/tests/tools/test_computer_use_delivery_ladder.py @@ -8,8 +8,8 @@ Covers NousResearch/hermes-agent#67052: with foreground_unsupported on an old driver rather than silently downgrading to background. - Phase C: foreground approval is scoped by (action, delivery_mode) and by - session_id, so a background approval never silently authorizes foreground - and one run's unlock never leaks into another. + the shared gate's session key, so a background approval never silently + authorizes foreground and one run's unlock never leaks into another. Stdlib + pytest + unittest.mock only. No live cua-driver, no network. """ @@ -220,7 +220,7 @@ def test_bad_delivery_mode_rejected(): assert res.code == "bad_delivery_mode" -def test_dispatcher_threads_delivery_mode_to_backend(): +def test_dispatcher_threads_delivery_mode_to_backend(grant_computer_use_approvals): """End-to-end through the tool dispatcher with the noop backend.""" from tools.computer_use import tool as cu with patch.dict(os.environ, {"HERMES_COMPUTER_USE_BACKEND": "noop"}, clear=False): @@ -237,70 +237,100 @@ def test_dispatcher_threads_delivery_mode_to_backend(): # Phase C — foreground approval scoping (action + delivery_mode + session) # --------------------------------------------------------------------------- -def test_background_approval_does_not_authorize_foreground(): +@pytest.fixture +def _interactive_session(monkeypatch): + """Interactive CLI presence for the shared gate plus a fresh approval session key; grants made here are + wiped from ``tools.approval``'s store afterwards so nothing leaks between tests.""" + from tools import approval + from tools.approval_context import reset_current_session_key, set_current_session_key + + monkeypatch.setenv("HERMES_INTERACTIVE", "1") + monkeypatch.setattr(approval, "save_permanent_allowlist", lambda patterns: None) + keys: list[str] = [] + + def use(session_key: str): + keys.append(session_key) + return set_current_session_key(session_key) + + yield use + for key in keys: + approval.clear_session(key) + reset_current_session_key(set_current_session_key("")) + + +def test_background_approval_does_not_authorize_foreground(_interactive_session): from tools.computer_use import tool as cu seen = [] - def cb(action, args, summary): - seen.append((action, args.get("delivery_mode"))) - return "approve_session" + def cb(command, description, **kw): + seen.append((command, description)) + return "session" cu.set_approval_callback(cb) + _interactive_session("sess-A") try: # Background click, approve for session. - assert cu._request_approval("click", {}, "sess-A") is None - # A second background click needs no prompt (cached). - assert cu._request_approval("click", {}, "sess-A") is None + assert cu._request_approval("click", {}) is None + # A second background click needs no prompt (cached in the shared session store). + assert cu._request_approval("click", {}) is None assert len(seen) == 1 - # Foreground click on the SAME action must prompt again — the - # background approval does not cover it. - assert cu._request_approval("click", {"delivery_mode": "foreground"}, "sess-A") is None + # Foreground click on the SAME action must prompt again — the background grant does not cover it. + assert cu._request_approval("click", {"delivery_mode": "foreground"}) is None assert len(seen) == 2 - assert seen[-1] == ("click", "foreground") + assert "FOREGROUND" in seen[-1][0] finally: cu.set_approval_callback(None) -def test_approval_state_is_session_scoped(): +def test_approval_state_is_session_scoped(_interactive_session): from tools.computer_use import tool as cu calls = [] - def cb(action, args, summary): - calls.append((action, args.get("delivery_mode"))) - return "approve_session" + def cb(command, description, **kw): + calls.append(command) + return "session" cu.set_approval_callback(cb) try: # Run A approves foreground click. - cu._request_approval("click", {"delivery_mode": "foreground"}, "run-A") + _interactive_session("run-A") + cu._request_approval("click", {"delivery_mode": "foreground"}) # Run B has NOT — it must prompt independently. n_before = len(calls) - cu._request_approval("click", {"delivery_mode": "foreground"}, "run-B") + _interactive_session("run-B") + cu._request_approval("click", {"delivery_mode": "foreground"}) assert len(calls) == n_before + 1 finally: cu.set_approval_callback(None) -def test_always_approve_covers_foreground(): +def test_always_grant_is_per_scope_key_and_visible_to_shared_store(_interactive_session): + """One grant store: an "always" answered through computer_use lands in ``tools.approval`` under the same + ``cua::`` key, and — unlike the old blanket unlock — covers only that scope, so the visible + foreground variant still prompts.""" + from tools import approval from tools.computer_use import tool as cu calls = [] - def cb(action, args, summary): - calls.append(action) - return "always_approve" + def cb(command, description, **kw): + calls.append(command) + return "always" cu.set_approval_callback(cb) + _interactive_session("run-C") try: - # First call unlocks everything for this session. - cu._request_approval("click", {}, "run-C") - # Foreground now sails through without another prompt. - cu._request_approval("click", {"delivery_mode": "foreground"}, "run-C") - assert len(calls) == 1 + assert cu._request_approval("click", {}) is None + assert approval.is_approved("run-C", "cua:click:background") + assert not approval.is_approved("run-C", "cua:click:foreground") + assert cu._request_approval("click", {"delivery_mode": "foreground"}) is None + assert len(calls) == 2 finally: cu.set_approval_callback(None) + with approval._lock: + approval._permanent_set().difference_update({"cua:click:background", "cua:click:foreground"}) def test_foreground_summary_warns_about_focus_change(): diff --git a/tools/computer_use/tool.py b/tools/computer_use/tool.py index afb000913f..d55de0f15b 100644 --- a/tools/computer_use/tool.py +++ b/tools/computer_use/tool.py @@ -25,11 +25,13 @@ from tools.computer_use.backend import ActionResult, CaptureResult, ComputerUseB logger = logging.getLogger(__name__) # ── Approval & safety ─────────────────────────────────────────────────────── +# Optional computer_use-specific prompt handed to the shared gate as its explicit ``approval_callback``; when None the +# gate resolves the per-thread CLI callback (``tools.terminal_tool.set_approval_callback``) like every other tool, so +# in-tree hosts never call this. Same contract as that callback: ``cb(command, description, **kw)`` -> +# "once" | "session" | "always" | "deny" | "timeout". _approval_callback = None def set_approval_callback(cb) -> None: - """Register the CLI approval prompt (terminal_tool pattern); ``cb(action, args, summary)`` -> - "approve_once" | "approve_session" | "always_approve" | "deny".""" global _approval_callback _approval_callback = cb @@ -82,14 +84,10 @@ _backend_permission_modes: Dict[str, str] = {} # (home key, provider, model) → bool. The decision reads the active profile's config (auxiliary.vision # override, declared supports_vision), so a multiplexed process must not serve profile A's verdict to B. _AUX_VISION_ROUTE_CACHE: Dict[Tuple[str, str, str], bool] = {} -# Approval state keyed by session_id so a gateway serving concurrent sessions can't leak one run's -# "always approve" into another; callers without a session_id share "". -# Falls back to a shared "" bucket for callers that don't pass a session_id (e.g. the classic single-run -# CLI). Values: _session_auto_approve[sid] -> bool ("always_approve everything") _always_allow[sid] -# -> set of (action, delivery_mode) scope keys See NousResearch/hermes-agent#67052 gap 4. +# Approval grants live in the shared store (``tools.approval``: session set + permanent allowlist), keyed by the +# gate's session key, so a computer_use "always" is one allowlist entry like any terminal pattern. Only the +# once-per-session escalation warning is tracked here. _approval_lock = threading.Lock() -_session_auto_approve: Dict[str, bool] = {} # sid -> "always_approve everything" -_always_allow: Dict[str, set] = {} # sid -> set of (action, delivery_mode) scope keys _escalation_warned: set = set() # sids already warned that a bypass widened the driver mode def _cua_permission_mode(session_id: str) -> str: @@ -178,13 +176,11 @@ def _get_backend(session_id: str = "") -> ComputerUseBackend: def release_computer_use_session(session_id: str) -> bool: """Release one session-owned backend (lifecycle seam for hosts/plugins); idempotent, True iff one was released. - Cache entries are removed BEFORE stopping so new lookups cannot retain the stale target/ref namespace; approval - state is cleared even without a backend.""" + Cache entries are removed BEFORE stopping so new lookups cannot retain the stale target/ref namespace. Approval + grants are not touched here: they live in the shared store and die with ``tools.approval.clear_session``.""" sid = str(session_id or "") with _backend_lock: backend, call_lock = _detach_locked(sid) - with _approval_lock: - _session_auto_approve.pop(sid, None), _always_allow.pop(sid, None) if backend is None: return False _stop_backend(backend, call_lock, @@ -209,7 +205,7 @@ def _shutdown_backend_atexit() -> None: _backend = None _backends.clear(), _backend_call_locks.clear(), _backend_permission_modes.clear() with _approval_lock: - _session_auto_approve.clear(), _always_allow.clear(), _escalation_warned.clear() + _escalation_warned.clear() for backend, call_lock in unique.values(): _stop_backend(backend, call_lock, lambda e: logger.debug("cua-driver atexit teardown failed: %s", e)) @@ -253,7 +249,7 @@ def handle_computer_use(args: Dict[str, Any], **kwargs) -> Any: scopes = ([action] if action in _ACTIONS and _ACTIONS[action].destructive else []) + ( ["bring_to_front"] if args.get("bring_to_front") or (action == "focus_app" and args.get("raise_window")) else []) for scope in scopes: - if (err := _request_approval(scope, args, session_id)) is not None: + if (err := _request_approval(scope, args)) is not None: return err try: backend = _get_backend(session_id=session_id) @@ -270,36 +266,29 @@ def handle_computer_use(args: Dict[str, Any], **kwargs) -> Any: logger.exception("computer_use %s failed", action) return json.dumps({"error": f"{action} failed: {e}"}) -def _request_approval(action: str, args: Dict[str, Any], session_id: str = "") -> Optional[str]: - """None if approved, else a JSON error string. Scoped by (action, delivery_mode) AND session_id: foreground - delivery is a visible focus change, so a background ``approve_session`` must NOT cover it; the blanket - ``always_approve`` does. No CLI approval wired -> default allow (gateway approval runs one layer out). - - ``always_approve`` (the blanket "auto-approve everything" unlock) still covers foreground, since the - user explicitly opted into unattended operation. State is keyed on session_id so concurrent runs don't - leak unlocks into one another. See #67052. +def _request_approval(action: str, args: Dict[str, Any]) -> Optional[str]: + """None if approved, else a JSON error string. The decision (yolo bypass, session/permanent grants, CLI prompt, + gateway pending, cron/unattended policy, fail-closed with nobody to ask) is ``tools.approval``'s shared gate, + so a computer_use grant is one store entry like any terminal pattern. Scope key ``cua::``: + foreground delivery is a visible focus change, so a background ``session`` grant must NOT cover it (#67052). """ - scope_key = (action, "foreground" if args.get("delivery_mode") == "foreground" else "background") - with _approval_lock: - if _session_auto_approve.get(session_id) or scope_key in _always_allow.get(session_id, set()): - return None - if (cb := _approval_callback) is None: + from tools.approval import _run_approval_gate + + mode = "foreground" if args.get("delivery_mode") == "foreground" else "background" + description = f"Allow computer_use to perform `{action}`?" + result = _run_approval_gate( + pattern_key=f"cua:{action}:{mode}", description=description, + display_target=f"computer_use: {_summarize_action(action, args)}", approval_callback=_approval_callback, + subject=f"computer_use `{action}` requires approval", noun="desktop actions", + advice="Find an alternative approach that avoids driving the desktop.", + autoapprove_log_prefix="computer_use action in non-interactive non-gateway context", + fail_closed_when_no_human=True, + no_human_block_message=(f"BLOCKED: computer_use `{action}` requires approval but no interactive user or " + "gateway is present to approve it."), + ) + if result.get("approved"): return None - try: - verdict = cb(action, args, _summarize_action(action, args)) - except Exception as e: - logger.warning("approval callback failed: %s", e) - verdict = "deny" - if verdict in ("approve_session", "always_approve"): - with _approval_lock: - _always_allow.setdefault(session_id, set()).add(scope_key) - if verdict == "always_approve": - _session_auto_approve[session_id] = True - if verdict in ("approve_once", "approve_session", "always_approve"): - return None - return json.dumps({"error": ("approval prompt timed out — the user did not respond. Silence is not consent; " - "do not retry without the user.") if verdict == "timeout" else "denied by user", - "action": action}) + return json.dumps({"error": result.get("message") or "denied by user", "action": action}) def _summarize_action(action: str, args: Dict[str, Any]) -> str: fg = " [FOREGROUND — briefly raises the window / changes focus]" if args.get("delivery_mode") == "foreground" else "" diff --git a/website/docs/user-guide/features/computer-use.md b/website/docs/user-guide/features/computer-use.md index 1148e2949a..514f5b3aee 100644 --- a/website/docs/user-guide/features/computer-use.md +++ b/website/docs/user-guide/features/computer-use.md @@ -340,8 +340,15 @@ magic-byte sniffing. Hermes applies multi-layer guardrails: - Destructive actions (click, type, drag, scroll, key, focus_app) - require approval — either interactively via the CLI dialog or via the - messaging-platform approval buttons. + require approval through the same gate as dangerous shell commands — + interactively via the CLI dialog or the messaging-platform approval + buttons. Once/session/always grants are keyed + `cua::` and live in the shared + session/`command_allowlist` store (a background grant never covers the + visible foreground variant). Where nobody can answer — cron + (`approvals.cron_mode`), single-query, unattended platforms, or any + headless run — the action is refused rather than auto-approved; + `--yolo` / `/yolo` still bypass. - Hard-blocked key combos at the tool level: empty trash, force delete, lock screen, log out, force log out. - Hard-blocked type patterns: `curl | bash`, `sudo rm -rf /`, fork