From b3941fddbf82b5c73e7071293978246018e1e506 Mon Sep 17 00:00:00 2001 From: Eugeniusz Gilewski Date: Sat, 23 May 2026 12:32:08 +0200 Subject: [PATCH 1/2] Guard execute_code in gateway approvals --- tests/tools/test_code_execution.py | 111 ++++++++++++ tools/approval.py | 262 +++++++++++++++++++++++++---- tools/code_execution_tool.py | 47 ++++++ 3 files changed, 383 insertions(+), 37 deletions(-) diff --git a/tests/tools/test_code_execution.py b/tests/tools/test_code_execution.py index 2d08265fb7b65..adfd7a41b35ba 100644 --- a/tests/tools/test_code_execution.py +++ b/tests/tools/test_code_execution.py @@ -17,6 +17,7 @@ import json import os +import tempfile os.environ["TERMINAL_ENV"] = "local" @@ -836,6 +837,116 @@ def test_nonoverlapping_tools_fallback(self): self.assertEqual(result["status"], "success") self.assertIn("fallback ok", result["output"]) + @unittest.skipIf(sys.platform == "win32", "UDS not available on Windows") + def test_gateway_execute_code_denial_blocks_child_process(self): + """Gateway approval denial must stop execute_code before spawn.""" + from tools.approval import ( + clear_session, + register_gateway_notify, + reset_current_session_key, + resolve_gateway_approval, + set_current_session_key, + unregister_gateway_notify, + ) + + session_key = "execute-code-deny" + notified = [] + result_holder = [] + + with tempfile.TemporaryDirectory() as tmp: + marker = os.path.join(tmp, "marker.txt") + code = ( + "from pathlib import Path\n" + f"Path({marker!r}).write_text('ran')\n" + "print('should-not-run')\n" + ) + + register_gateway_notify(session_key, lambda data: notified.append(data)) + token = set_current_session_key(session_key) + os.environ["HERMES_GATEWAY_SESSION"] = "1" + os.environ["HERMES_EXEC_ASK"] = "1" + os.environ["HERMES_SESSION_KEY"] = session_key + try: + thread = threading.Thread( + target=lambda: result_holder.append(json.loads(execute_code( + code, + task_id="test-exec-deny", + enabled_tools=[], + ))) + ) + thread.start() + + deadline = time.monotonic() + 5 + while not notified and time.monotonic() < deadline: + time.sleep(0.05) + + self.assertEqual(len(notified), 1) + self.assertIn("execute_code <<'PY'", notified[0]["command"]) + self.assertFalse(os.path.exists(marker)) + + resolve_gateway_approval(session_key, "deny") + thread.join(timeout=10) + + self.assertFalse(thread.is_alive()) + self.assertEqual(result_holder[0]["status"], "blocked") + self.assertFalse(os.path.exists(marker)) + finally: + os.environ.pop("HERMES_GATEWAY_SESSION", None) + os.environ.pop("HERMES_EXEC_ASK", None) + os.environ.pop("HERMES_SESSION_KEY", None) + unregister_gateway_notify(session_key) + clear_session(session_key) + reset_current_session_key(token) + + @unittest.skipIf(sys.platform == "win32", "UDS not available on Windows") + def test_gateway_execute_code_runs_after_one_shot_approval(self): + """Approving the execute_code preflight allows the script to run.""" + from tools.approval import ( + clear_session, + register_gateway_notify, + reset_current_session_key, + resolve_gateway_approval, + set_current_session_key, + unregister_gateway_notify, + ) + + session_key = "execute-code-approve" + notified = [] + result_holder = [] + register_gateway_notify(session_key, lambda data: notified.append(data)) + token = set_current_session_key(session_key) + os.environ["HERMES_GATEWAY_SESSION"] = "1" + os.environ["HERMES_EXEC_ASK"] = "1" + os.environ["HERMES_SESSION_KEY"] = session_key + try: + thread = threading.Thread( + target=lambda: result_holder.append(json.loads(execute_code( + "print('approved-run')", + task_id="test-exec-approve", + enabled_tools=[], + ))) + ) + thread.start() + + deadline = time.monotonic() + 5 + while not notified and time.monotonic() < deadline: + time.sleep(0.05) + + self.assertEqual(len(notified), 1) + resolve_gateway_approval(session_key, "once") + thread.join(timeout=10) + + self.assertFalse(thread.is_alive()) + self.assertEqual(result_holder[0]["status"], "success") + self.assertIn("approved-run", result_holder[0]["output"]) + finally: + os.environ.pop("HERMES_GATEWAY_SESSION", None) + os.environ.pop("HERMES_EXEC_ASK", None) + os.environ.pop("HERMES_SESSION_KEY", None) + unregister_gateway_notify(session_key) + clear_session(session_key) + reset_current_session_key(token) + # --------------------------------------------------------------------------- # _load_config diff --git a/tools/approval.py b/tools/approval.py index 399b9d6c2d21e..be8cad592af79 100644 --- a/tools/approval.py +++ b/tools/approval.py @@ -910,6 +910,228 @@ def _smart_approve(command: str, description: str) -> str: return "escalate" +def _gateway_approval_wait_timeout() -> int: + """Return the gateway approval wait timeout in seconds.""" + timeout = _get_approval_config().get("gateway_timeout", 300) + try: + return int(timeout) + except (ValueError, TypeError): + return 300 + + +def _wait_for_gateway_entry(entry: _ApprovalEntry, label: str) -> bool: + """Wait for a gateway approval entry while keeping activity alive.""" + timeout = _gateway_approval_wait_timeout() + try: + from tools.environments.base import touch_activity_if_due + except Exception: # pragma: no cover + touch_activity_if_due = None + + now = time.monotonic() + deadline = now + max(timeout, 0) + activity_state = {"last_touch": now, "start": now} + + while True: + remaining = deadline - time.monotonic() + if remaining <= 0: + return False + if entry.event.wait(timeout=min(1.0, remaining)): + return True + if touch_activity_if_due is not None: + touch_activity_if_due(activity_state, label) + + +def check_execute_code_guard(code: str, env_type: str) -> dict: + """Require approval before local/SSH execute_code can run in async contexts. + + execute_code runs arbitrary Python, and that Python can call subprocess, + os.system, ctypes, or other process/file APIs directly. Those calls do not + pass through terminal() and therefore cannot be reliably inspected by + DANGEROUS_PATTERNS at the shell-string layer. In gateway/ask contexts we + fail closed by approving the script execution itself before the child + process is spawned. + """ + if env_type in {"docker", "singularity", "modal", "daytona", "vercel_sandbox"}: + return {"approved": True, "message": None} + + approval_mode = _get_approval_mode() + if is_truthy_value(os.getenv("HERMES_YOLO_MODE")) or is_current_session_yolo_enabled() or approval_mode == "off": + return {"approved": True, "message": None} + + is_gateway = _is_gateway_approval_context() + is_ask = env_var_enabled("HERMES_EXEC_ASK") + is_cron = env_var_enabled("HERMES_CRON_SESSION") + + if is_cron and _get_cron_approval_mode() == "deny": + return { + "approved": False, + "message": ( + "BLOCKED: execute_code can run arbitrary local Python, " + "including subprocess calls that bypass shell-string approval " + "checks. Cron jobs run without a user present to approve it. " + "Use normal tools instead, or set approvals.cron_mode: approve " + "only if this cron profile is intentionally trusted." + ), + "pattern_key": "execute_code", + "description": "execute_code script execution", + "outcome": "blocked", + "user_consent": False, + } + + # Preserve ordinary non-interactive script behavior outside gateway/ask. + if not is_gateway and not is_ask: + return {"approved": True, "message": None} + + session_key = get_current_session_key() + pattern_key = "execute_code" + description = ( + "execute_code script execution. The script can spawn subprocesses " + "or mutate files without passing through terminal command approval; " + "approval is one-shot for this run." + ) + command = f"execute_code <<'PY'\n{code}\nPY" + + if approval_mode == "smart": + verdict = _smart_approve(command, description) + if verdict == "approve": + logger.debug("Smart approval: auto-approved execute_code for session %s", session_key) + return { + "approved": True, + "message": None, + "smart_approved": True, + "description": description, + } + if verdict == "deny": + return { + "approved": False, + "message": ( + "BLOCKED by smart approval: execute_code script execution " + "was assessed as genuinely dangerous. Do NOT retry." + ), + "smart_denied": True, + "pattern_key": pattern_key, + "description": description, + "outcome": "denied", + "user_consent": False, + } + + with _lock: + notify_cb = _gateway_notify_cbs.get(session_key) + + if notify_cb is None: + submit_pending(session_key, { + "command": command, + "pattern_key": pattern_key, + "pattern_keys": [pattern_key], + "description": description, + }) + return { + "approved": False, + "pattern_key": pattern_key, + "status": "pending_approval", + "approval_pending": True, + "command": command, + "description": description, + "message": ( + f"⚠️ {description}. Asking the user for approval.\n\n" + f"**Code:**\n```python\n{code}\n```" + ), + } + + approval_data = { + "command": command, + "pattern_key": pattern_key, + "pattern_keys": [pattern_key], + "description": description, + "allow_permanent": False, + } + entry = _ApprovalEntry(approval_data) + with _lock: + _gateway_queues.setdefault(session_key, []).append(entry) + + _fire_approval_hook( + "pre_approval_request", + command=command, + description=description, + pattern_key=pattern_key, + pattern_keys=[pattern_key], + session_key=session_key, + surface="gateway", + ) + + try: + notify_cb(approval_data) + except Exception as exc: + logger.warning("Gateway execute_code approval notify failed: %s", exc) + with _lock: + queue = _gateway_queues.get(session_key, []) + if entry in queue: + queue.remove(entry) + if not queue: + _gateway_queues.pop(session_key, None) + return { + "approved": False, + "message": "BLOCKED: Failed to send execute_code approval request to user. Do NOT retry.", + "pattern_key": pattern_key, + "description": description, + "outcome": "notify_failed", + "user_consent": False, + } + + resolved = _wait_for_gateway_entry(entry, "waiting for execute_code approval") + + with _lock: + queue = _gateway_queues.get(session_key, []) + if entry in queue: + queue.remove(entry) + if not queue: + _gateway_queues.pop(session_key, None) + + choice = entry.result + outcome = "timeout" if not resolved else (choice if choice else "timeout") + _fire_approval_hook( + "post_approval_response", + command=command, + description=description, + pattern_key=pattern_key, + pattern_keys=[pattern_key], + session_key=session_key, + surface="gateway", + choice=outcome, + ) + + if not resolved or choice is None or choice == "deny": + if not resolved: + reason = "timed out without user response" + timeout_addendum = " Silence is not consent." + result_outcome = "timeout" + else: + reason = "denied by user" + timeout_addendum = "" + result_outcome = "denied" + return { + "approved": False, + "message": ( + f"BLOCKED: execute_code {reason}. The user has NOT consented " + f"to running this script. Do NOT retry this command, do NOT " + f"rephrase it, and do NOT attempt the same outcome through " + f"subprocess, terminal, or another tool." + f"{timeout_addendum}" + ), + "pattern_key": pattern_key, + "description": description, + "outcome": result_outcome, + "user_consent": False, + } + + return { + "approved": True, + "message": None, + "user_approved": True, + "description": description, + } + + def check_dangerous_command(command: str, env_type: str, approval_callback=None) -> dict: """Check if a command is dangerous and handle approval. @@ -1233,43 +1455,9 @@ def check_all_command_guards(command: str, env_type: str, "description": combined_desc, } - # Block until the user responds or timeout (default 5 min). - # Poll in short slices so we can fire activity heartbeats every - # ~10s to the agent's inactivity tracker. Without this, the - # blocking event.wait() never touches activity, and the - # gateway's inactivity watchdog (agent.gateway_timeout, default - # 1800s) kills the agent while the user is still responding to - # the approval prompt. Mirrors the _wait_for_process() cadence - # in tools/environments/base.py. - timeout = _get_approval_config().get("gateway_timeout", 300) - try: - timeout = int(timeout) - except (ValueError, TypeError): - timeout = 300 - - try: - from tools.environments.base import touch_activity_if_due - except Exception: # pragma: no cover - touch_activity_if_due = None - - _now = time.monotonic() - _deadline = _now + max(timeout, 0) - _activity_state = {"last_touch": _now, "start": _now} - resolved = False - while True: - _remaining = _deadline - time.monotonic() - if _remaining <= 0: - break - # 1s poll slice — the event is set immediately when the - # user responds, so slice length only controls heartbeat - # cadence, not user-visible responsiveness. - if entry.event.wait(timeout=min(1.0, _remaining)): - resolved = True - break - if touch_activity_if_due is not None: - touch_activity_if_due( - _activity_state, "waiting for user approval" - ) + # Block until the user responds or timeout while keeping the + # gateway inactivity watchdog fed during the approval wait. + resolved = _wait_for_gateway_entry(entry, "waiting for user approval") # Clean up this entry from the queue with _lock: diff --git a/tools/code_execution_tool.py b/tools/code_execution_tool.py index bdbc4bfbe1bfb..684fafc324a29 100644 --- a/tools/code_execution_tool.py +++ b/tools/code_execution_tool.py @@ -1066,6 +1066,53 @@ def execute_code( # Dispatch: remote backends use file-based RPC, local uses UDS from tools.terminal_tool import _get_env_config env_type = _get_env_config()["env_type"] + + try: + from tools.approval import check_execute_code_guard + + approval = check_execute_code_guard(code, env_type) + except Exception as exc: + logger.error("execute_code approval guard failed: %s", exc, exc_info=True) + return json.dumps({ + "status": "blocked", + "output": "", + "error": ( + "execute_code approval guard failed before the script could " + "run. This path is fail-closed because execute_code can spawn " + "subprocesses outside terminal command approval." + ), + "tool_calls_made": 0, + "duration_seconds": 0, + }, ensure_ascii=False) + + if not approval.get("approved"): + if approval.get("status") == "pending_approval": + return json.dumps({ + "status": "pending_approval", + "approval_pending": True, + "output": "", + "error": "", + "command": approval.get("command", "execute_code"), + "description": approval.get("description", "execute_code approval"), + "pattern_key": approval.get("pattern_key", "execute_code"), + "tool_calls_made": 0, + "duration_seconds": 0, + }, ensure_ascii=False) + return json.dumps({ + "status": "blocked", + "output": "", + "error": approval.get( + "message", + "BLOCKED: execute_code was not approved by the user.", + ), + "description": approval.get("description", "execute_code approval"), + "pattern_key": approval.get("pattern_key", "execute_code"), + "outcome": approval.get("outcome", "blocked"), + "user_consent": approval.get("user_consent", False), + "tool_calls_made": 0, + "duration_seconds": 0, + }, ensure_ascii=False) + if env_type != "local": return _execute_remote(code, task_id, enabled_tools) From 72ff96b16a0da3fa905d8411f2aa97cb6a377112 Mon Sep 17 00:00:00 2001 From: Eugeniusz Gilewski Date: Sat, 23 May 2026 14:19:48 +0200 Subject: [PATCH 2/2] Stabilize unrelated CI tests --- tests/acp/test_server.py | 34 +++++++++----------------- tests/tools/test_browser_supervisor.py | 18 ++++++++------ 2 files changed, 23 insertions(+), 29 deletions(-) diff --git a/tests/acp/test_server.py b/tests/acp/test_server.py index 6dce8d8702b8d..8feb6700c2b81 100644 --- a/tests/acp/test_server.py +++ b/tests/acp/test_server.py @@ -971,17 +971,13 @@ def fake_agent(**kwargs): "hermes_cli.runtime_provider.resolve_runtime_provider", fake_resolve_runtime_provider, ) - # Pin the parser so this test doesn't depend on live - # ``_KNOWN_PROVIDER_NAMES`` / ``_PROVIDER_ALIASES`` module state - # (sibling of the same hardening on - # ``test_model_switch_uses_requested_provider``). + # This test covers the ACP model-switch handoff, not model-string + # parsing. Patch the ACP resolver directly so live provider registry + # state from unrelated tests cannot shadow the provider under test. monkeypatch.setattr( - "hermes_cli.models.parse_model_input", - lambda raw, current: ("anthropic", "claude-sonnet-4-6"), - ) - monkeypatch.setattr( - "hermes_cli.models.detect_provider_for_model", - lambda model, current: None, + HermesACPAgent, + "_resolve_model_selection", + staticmethod(lambda raw, current: ("anthropic", "claude-sonnet-4-6")), ) manager = SessionManager(db=SessionDB(tmp_path / "state.db")) @@ -1555,19 +1551,13 @@ def fake_agent(**kwargs): "hermes_cli.runtime_provider.resolve_runtime_provider", fake_resolve_runtime_provider, ) - # Pin the model-string parser independently of the live - # ``_KNOWN_PROVIDER_NAMES`` / ``_PROVIDER_ALIASES`` module state. - # Otherwise any test in the same xdist worker that mutates those - # globals (e.g. registers a custom provider that shadows - # ``anthropic``) flakes this one — observed once in CI as - # ``'custom' == 'anthropic'``. - monkeypatch.setattr( - "hermes_cli.models.parse_model_input", - lambda raw, current: ("anthropic", "claude-sonnet-4-6"), - ) + # This test covers the ACP model-switch handoff, not model-string + # parsing. Patch the ACP resolver directly so live provider registry + # state from unrelated tests cannot shadow the provider under test. monkeypatch.setattr( - "hermes_cli.models.detect_provider_for_model", - lambda model, current: None, + HermesACPAgent, + "_resolve_model_selection", + staticmethod(lambda raw, current: ("anthropic", "claude-sonnet-4-6")), ) manager = SessionManager(db=SessionDB(tmp_path / "state.db")) diff --git a/tests/tools/test_browser_supervisor.py b/tests/tools/test_browser_supervisor.py index 179a94506ed48..2a1965d7f9dd6 100644 --- a/tests/tools/test_browser_supervisor.py +++ b/tests/tools/test_browser_supervisor.py @@ -40,6 +40,15 @@ def _find_chrome() -> str: pytest.skip("no Chrome binary found") +def _terminate_chrome(proc: subprocess.Popen) -> None: + proc.terminate() + try: + proc.wait(timeout=5) + except subprocess.TimeoutExpired: + proc.kill() + proc.wait(timeout=5) + + @pytest.fixture def chrome_cdp(request): """Start a headless Chrome with --remote-debugging-port, yield its WS URL. @@ -89,18 +98,13 @@ def chrome_cdp(request): except Exception: time.sleep(0.25) if ws_url is None: - proc.terminate() - proc.wait(timeout=5) + _terminate_chrome(proc) shutil.rmtree(profile, ignore_errors=True) pytest.skip("Chrome didn't expose CDP in time") yield ws_url, port - proc.terminate() - try: - proc.wait(timeout=3) - except Exception: - proc.kill() + _terminate_chrome(proc) shutil.rmtree(profile, ignore_errors=True)