diff --git a/agent/agent_runtime_helpers.py b/agent/agent_runtime_helpers.py index ade4831d1b8..da9c20c0510 100644 --- a/agent/agent_runtime_helpers.py +++ b/agent/agent_runtime_helpers.py @@ -2094,12 +2094,12 @@ def invoke_tool(agent, function_name: str, function_args: dict, effective_task_i except Exception as _mw_err: logger.debug("tool_request middleware error: %s", _mw_err) - # Check plugin hooks for a block directive before executing anything. + # Check plugin hooks for a block or approval directive before executing. block_message: Optional[str] = None if not pre_tool_block_checked: try: - from hermes_cli.plugins import get_pre_tool_call_block_message - block_message = get_pre_tool_call_block_message( + from hermes_cli.plugins import resolve_pre_tool_block + block_message = resolve_pre_tool_block( function_name, function_args, task_id=effective_task_id or "", @@ -2110,7 +2110,7 @@ def invoke_tool(agent, function_name: str, function_args: dict, effective_task_i middleware_trace=list(_tool_middleware_trace), ) except Exception: - pass + block_message = None if block_message is not None: result = json.dumps({"error": block_message}, ensure_ascii=False) try: diff --git a/agent/tool_executor.py b/agent/tool_executor.py index 44b9a367c90..9688f5d3dc6 100644 --- a/agent/tool_executor.py +++ b/agent/tool_executor.py @@ -415,8 +415,8 @@ def execute_tool_calls_concurrent(agent, assistant_message, messages: list, effe ) else: try: - from hermes_cli.plugins import get_pre_tool_call_block_message - block_message = get_pre_tool_call_block_message( + from hermes_cli.plugins import resolve_pre_tool_block + block_message = resolve_pre_tool_block( function_name, function_args, task_id=effective_task_id or "", @@ -1034,8 +1034,8 @@ def execute_tool_calls_sequential(agent, assistant_message, messages: list, effe _block_error_type = "tool_scope_block" else: try: - from hermes_cli.plugins import get_pre_tool_call_block_message - _block_msg = get_pre_tool_call_block_message( + from hermes_cli.plugins import resolve_pre_tool_block + _block_msg = resolve_pre_tool_block( function_name, function_args, task_id=effective_task_id or "", diff --git a/hermes_cli/plugins.py b/hermes_cli/plugins.py index d5e4b3ff8c1..e8b1b59b798 100644 --- a/hermes_cli/plugins.py +++ b/hermes_cli/plugins.py @@ -2046,7 +2046,7 @@ def clear_thread_tool_whitelist() -> None: _thread_tool_whitelist.allowed = None -def get_pre_tool_call_block_message( +def get_pre_tool_call_directive( tool_name: str, args: Optional[Dict[str, Any]], task_id: str = "", @@ -2055,22 +2055,38 @@ def get_pre_tool_call_block_message( turn_id: str = "", api_request_id: str = "", middleware_trace: Optional[List[Dict[str, Any]]] = None, -) -> Optional[str]: - """Check ``pre_tool_call`` hooks for a blocking directive. +) -> tuple[Optional[str], Optional[str]]: + """Check ``pre_tool_call`` hooks for a blocking or approval directive. Plugins that need to enforce policy (rate limiting, security - restrictions, approval workflows) can return:: + restrictions, approval workflows) can return one of:: - {"action": "block", "message": "Reason the tool was blocked"} + {"action": "block", "message": "Reason the tool was blocked"} + {"action": "approve", "message": "Why this needs human confirmation"} - from their ``pre_tool_call`` callback. The first valid block - directive wins. Invalid or irrelevant hook return values are - silently ignored so existing observer-only hooks are unaffected. + from their ``pre_tool_call`` callback. + + - ``block`` vetoes the tool call outright (the message becomes the tool + result the model sees). + - ``approve`` ESCALATES to the existing human-approval gate + (``prompt_dangerous_approval`` on CLI, the approval callback on the + gateway) — the same mechanism Tier-2 dangerous shell patterns use. + This lets a plugin require a human ``[o]nce/[s]ession/[a]lways/[d]eny`` + decision on ANY tool, not just terminal command strings. The caller is + responsible for invoking the gate (see + :func:`tools.approval.request_tool_approval`). + + The first valid directive wins. Invalid or irrelevant hook return values + are silently ignored so existing observer-only hooks are unaffected. + + Returns: + ``(directive, message)`` where ``directive`` is ``"block"``, + ``"approve"``, or ``None``. """ allowed = getattr(_thread_tool_whitelist, "allowed", None) if allowed is not None and tool_name not in allowed: fmt = getattr(_thread_tool_whitelist, "fmt", "Tool '{tool_name}' denied") - return fmt.format(tool_name=tool_name) + return ("block", fmt.format(tool_name=tool_name)) hook_results = invoke_hook( "pre_tool_call", @@ -2087,12 +2103,91 @@ def get_pre_tool_call_block_message( for result in hook_results: if not isinstance(result, dict): continue - if result.get("action") != "block": + action = result.get("action") + if action not in ("block", "approve"): continue message = result.get("message") - if isinstance(message, str) and message: - return message + message = message if isinstance(message, str) and message else None + # A block directive requires a message (it becomes the tool result); + # an approve directive can carry an optional reason. + if action == "block" and not message: + continue + return (action, message) + return (None, None) + + +def get_pre_tool_call_block_message( + tool_name: str, + args: Optional[Dict[str, Any]], + task_id: str = "", + session_id: str = "", + tool_call_id: str = "", + turn_id: str = "", + api_request_id: str = "", + middleware_trace: Optional[List[Dict[str, Any]]] = None, +) -> Optional[str]: + """Back-compat shim: return only a ``block`` message (or ``None``). + + Deprecated in favor of :func:`get_pre_tool_call_directive`, which also + surfaces the ``approve`` escalation directive. Kept so any external caller + importing the old name keeps working; ``approve`` directives are invisible + to this shim (it only reports blocks). + """ + directive, message = get_pre_tool_call_directive( + tool_name, args, task_id=task_id, session_id=session_id, + tool_call_id=tool_call_id, turn_id=turn_id, + api_request_id=api_request_id, middleware_trace=middleware_trace, + ) + return message if directive == "block" else None + + +def resolve_pre_tool_block( + tool_name: str, + args: Optional[Dict[str, Any]], + task_id: str = "", + session_id: str = "", + tool_call_id: str = "", + turn_id: str = "", + api_request_id: str = "", + middleware_trace: Optional[List[Dict[str, Any]]] = None, +) -> Optional[str]: + """Resolve the pre_tool_call directive to a final block message (or None). + + Single entry point for every tool-dispatch site: fetches the plugin + directive and, for an ``approve`` escalation, invokes the human-approval + gate (:func:`tools.approval.request_tool_approval`). Returns the message + the tool result should carry when the call is blocked, or ``None`` when + the call may proceed. + + Centralizing this keeps the security-critical fail-closed logic in ONE + place instead of copy-pasted across the concurrent/sequential/helper + dispatch paths: an ``approve`` directive whose gate errors, denies, or + times out is fail-closed to a block; ``block`` blocks with its message; + anything else proceeds. + """ + directive, message = get_pre_tool_call_directive( + tool_name, args, task_id=task_id, session_id=session_id, + tool_call_id=tool_call_id, turn_id=turn_id, + api_request_id=api_request_id, middleware_trace=middleware_trace, + ) + if directive == "block": + return message + if directive == "approve": + try: + from tools.approval import request_tool_approval + result = request_tool_approval( + tool_name, message or "", rule_key=tool_name, + ) + except Exception: + # Fail-closed: if the gate itself errors, block rather than + # silently execute an action a plugin flagged for approval. + return f"BLOCKED: plugin approval gate failed for {tool_name}" + if not result.get("approved"): + return str( + result.get("message") + or f"BLOCKED: plugin approval required for {tool_name}" + ) return None diff --git a/model_tools.py b/model_tools.py index e07f6da68fe..a6fad7e31c1 100644 --- a/model_tools.py +++ b/model_tools.py @@ -1046,21 +1046,22 @@ def handle_function_call( if function_name in _AGENT_LOOP_TOOLS: return json.dumps({"error": f"{function_name} must be handled by the agent loop"}) - # Check plugin hooks for a block directive (unless caller already - # checked — e.g. run_agent._invoke_tool passes skip=True to + # Check plugin hooks for a block/approve directive (unless caller + # already checked — e.g. run_agent._invoke_tool passes skip=True to # avoid double-firing the hook). # # Single-fire contract: pre_tool_call fires exactly once per tool - # execution. get_pre_tool_call_block_message() internally calls - # invoke_hook("pre_tool_call", ...) and returns the first block - # directive (if any), so observer plugins see the hook on that same - # pass. When skip=True, the caller already fired it — do nothing - # here. + # execution. resolve_pre_tool_block() internally calls + # invoke_hook("pre_tool_call", ...) once and returns the block message + # for a `block` directive OR for an `approve` directive whose human + # gate denied/timed-out/errored (fail-closed). Observer plugins see + # the hook on that same pass. When skip=True, the caller already + # fired it — do nothing here. if not skip_pre_tool_call_hook: block_message: Optional[str] = None try: - from hermes_cli.plugins import get_pre_tool_call_block_message - block_message = get_pre_tool_call_block_message( + from hermes_cli.plugins import resolve_pre_tool_block + block_message = resolve_pre_tool_block( function_name, function_args, task_id=task_id or "", diff --git a/tests/hermes_cli/test_plugins.py b/tests/hermes_cli/test_plugins.py index 0df9790c31a..10d68f81ea8 100644 --- a/tests/hermes_cli/test_plugins.py +++ b/tests/hermes_cli/test_plugins.py @@ -859,6 +859,115 @@ class TestPreToolCallBlocking: assert get_pre_tool_call_block_message("terminal", {}) == "first blocker" +class TestPreToolCallDirective: + """Tests for the extended (block | approve) directive helper.""" + + def test_approve_directive_returned(self, monkeypatch): + from hermes_cli.plugins import get_pre_tool_call_directive + monkeypatch.setattr( + "hermes_cli.plugins.invoke_hook", + lambda hook_name, **kwargs: [ + {"action": "approve", "message": "needs human ok"} + ], + ) + assert get_pre_tool_call_directive("write_file", {}) == ( + "approve", "needs human ok") + + def test_approve_without_message_is_valid(self, monkeypatch): + """approve may omit a message (block may not).""" + from hermes_cli.plugins import get_pre_tool_call_directive + monkeypatch.setattr( + "hermes_cli.plugins.invoke_hook", + lambda hook_name, **kwargs: [{"action": "approve"}], + ) + assert get_pre_tool_call_directive("write_file", {}) == ("approve", None) + + def test_block_still_requires_message(self, monkeypatch): + from hermes_cli.plugins import get_pre_tool_call_directive + monkeypatch.setattr( + "hermes_cli.plugins.invoke_hook", + lambda hook_name, **kwargs: [{"action": "block"}], + ) + assert get_pre_tool_call_directive("terminal", {}) == (None, None) + + def test_first_directive_wins_across_actions(self, monkeypatch): + from hermes_cli.plugins import get_pre_tool_call_directive + monkeypatch.setattr( + "hermes_cli.plugins.invoke_hook", + lambda hook_name, **kwargs: [ + {"action": "approve", "message": "gate first"}, + {"action": "block", "message": "block second"}, + ], + ) + assert get_pre_tool_call_directive("terminal", {}) == ( + "approve", "gate first") + + def test_shim_ignores_approve(self, monkeypatch): + """Back-compat shim only reports block, never approve.""" + monkeypatch.setattr( + "hermes_cli.plugins.invoke_hook", + lambda hook_name, **kwargs: [ + {"action": "approve", "message": "gate"} + ], + ) + assert get_pre_tool_call_block_message("write_file", {}) is None + + +class TestResolvePreToolBlock: + """Tests for the single dispatch-site chokepoint that resolves a + directive (incl. the approve→gate escalation) to a block message.""" + + def test_block_returns_message(self, monkeypatch): + from hermes_cli.plugins import resolve_pre_tool_block + monkeypatch.setattr( + "hermes_cli.plugins.invoke_hook", + lambda hook_name, **kwargs: [{"action": "block", "message": "no"}], + ) + assert resolve_pre_tool_block("terminal", {}) == "no" + + def test_no_directive_returns_none(self, monkeypatch): + from hermes_cli.plugins import resolve_pre_tool_block + monkeypatch.setattr( + "hermes_cli.plugins.invoke_hook", lambda hook_name, **kwargs: []) + assert resolve_pre_tool_block("terminal", {}) is None + + def test_approve_denied_blocks(self, monkeypatch): + from hermes_cli.plugins import resolve_pre_tool_block + monkeypatch.setattr( + "hermes_cli.plugins.invoke_hook", + lambda hook_name, **kwargs: [{"action": "approve", "message": "why"}], + ) + monkeypatch.setattr( + "tools.approval.request_tool_approval", + lambda *a, **k: {"approved": False, "message": "user denied it"}, + ) + assert resolve_pre_tool_block("write_file", {}) == "user denied it" + + def test_approve_granted_allows(self, monkeypatch): + from hermes_cli.plugins import resolve_pre_tool_block + monkeypatch.setattr( + "hermes_cli.plugins.invoke_hook", + lambda hook_name, **kwargs: [{"action": "approve", "message": "why"}], + ) + monkeypatch.setattr( + "tools.approval.request_tool_approval", + lambda *a, **k: {"approved": True, "message": None}, + ) + assert resolve_pre_tool_block("write_file", {}) is None + + def test_approve_gate_exception_fails_closed(self, monkeypatch): + from hermes_cli.plugins import resolve_pre_tool_block + monkeypatch.setattr( + "hermes_cli.plugins.invoke_hook", + lambda hook_name, **kwargs: [{"action": "approve", "message": "why"}], + ) + def _boom(*a, **k): + raise RuntimeError("gate crashed") + monkeypatch.setattr("tools.approval.request_tool_approval", _boom) + msg = resolve_pre_tool_block("terminal", {}) + assert msg is not None and "gate failed" in msg # fail-closed + + class TestGetPreVerifyContinueMessage: """`pre_verify` directive aggregation — mirrors the pre_tool_call block path.""" diff --git a/tests/run_agent/test_run_agent.py b/tests/run_agent/test_run_agent.py index 575a9700ac3..18dced788b4 100644 --- a/tests/run_agent/test_run_agent.py +++ b/tests/run_agent/test_run_agent.py @@ -2966,7 +2966,7 @@ class TestConcurrentToolExecution: """Agent-owned tool paths should close observer tool spans.""" hook_calls = [] monkeypatch.setattr( - "hermes_cli.plugins.get_pre_tool_call_block_message", + "hermes_cli.plugins.resolve_pre_tool_block", lambda *args, **kwargs: None, ) monkeypatch.setattr( @@ -2990,7 +2990,7 @@ class TestConcurrentToolExecution: def test_invoke_tool_blocked_returns_error_and_skips_execution(self, agent, monkeypatch): """_invoke_tool should return error JSON when a plugin blocks the tool.""" monkeypatch.setattr( - "hermes_cli.plugins.get_pre_tool_call_block_message", + "hermes_cli.plugins.resolve_pre_tool_block", lambda *args, **kwargs: "Blocked by test policy", ) with patch("tools.todo_tool.todo_tool", side_effect=AssertionError("should not run")) as mock_todo: @@ -3002,7 +3002,7 @@ class TestConcurrentToolExecution: def test_invoke_tool_blocked_skips_handle_function_call(self, agent, monkeypatch): """Blocked registry tools should not reach handle_function_call.""" monkeypatch.setattr( - "hermes_cli.plugins.get_pre_tool_call_block_message", + "hermes_cli.plugins.resolve_pre_tool_block", lambda *args, **kwargs: "Blocked", ) with patch("run_agent.handle_function_call", side_effect=AssertionError("should not run")): @@ -3019,7 +3019,7 @@ class TestConcurrentToolExecution: messages = [] monkeypatch.setattr( - "hermes_cli.plugins.get_pre_tool_call_block_message", + "hermes_cli.plugins.resolve_pre_tool_block", lambda *args, **kwargs: "Blocked by policy", ) agent._checkpoint_mgr.enabled = True @@ -3049,7 +3049,7 @@ class TestConcurrentToolExecution: hook_calls = [] monkeypatch.setattr( - "hermes_cli.plugins.get_pre_tool_call_block_message", + "hermes_cli.plugins.resolve_pre_tool_block", lambda *args, **kwargs: "Blocked by policy", ) monkeypatch.setattr( @@ -3076,7 +3076,7 @@ class TestConcurrentToolExecution: hook_calls = [] monkeypatch.setattr( - "hermes_cli.plugins.get_pre_tool_call_block_message", + "hermes_cli.plugins.resolve_pre_tool_block", lambda *args, **kwargs: None, ) monkeypatch.setattr( @@ -3123,7 +3123,7 @@ class TestConcurrentToolExecution: lambda kind, **kwargs: [request_middleware(**kwargs)] if kind == "tool_request" else [], ) monkeypatch.setattr( - "hermes_cli.plugins.get_pre_tool_call_block_message", + "hermes_cli.plugins.resolve_pre_tool_block", lambda *args, **kwargs: None, ) monkeypatch.setattr( @@ -3161,7 +3161,7 @@ class TestConcurrentToolExecution: lambda kind, **kwargs: [request_middleware(**kwargs)] if kind == "tool_request" else [], ) monkeypatch.setattr( - "hermes_cli.plugins.get_pre_tool_call_block_message", + "hermes_cli.plugins.resolve_pre_tool_block", lambda *args, **kwargs: None, ) monkeypatch.setattr( @@ -3196,7 +3196,7 @@ class TestConcurrentToolExecution: """Blocked memory tool should not reset the nudge counter.""" agent._turns_since_memory = 5 monkeypatch.setattr( - "hermes_cli.plugins.get_pre_tool_call_block_message", + "hermes_cli.plugins.resolve_pre_tool_block", lambda *args, **kwargs: "Blocked", ) with patch("tools.memory_tool.memory_tool", side_effect=AssertionError("should not run")): @@ -3209,7 +3209,7 @@ class TestConcurrentToolExecution: def test_invoke_tool_memory_remove_notifies_provider_with_old_text(self, agent, monkeypatch): monkeypatch.setattr( - "hermes_cli.plugins.get_pre_tool_call_block_message", + "hermes_cli.plugins.resolve_pre_tool_block", lambda *args, **kwargs: None, ) calls = [] @@ -3241,7 +3241,7 @@ class TestConcurrentToolExecution: def test_invoke_tool_memory_failed_remove_skips_provider_notification(self, agent, monkeypatch): monkeypatch.setattr( - "hermes_cli.plugins.get_pre_tool_call_block_message", + "hermes_cli.plugins.resolve_pre_tool_block", lambda *args, **kwargs: None, ) notify = MagicMock(side_effect=AssertionError("should not notify")) @@ -3281,7 +3281,7 @@ class TestConcurrentToolExecution: messages = [] monkeypatch.setattr( - "hermes_cli.plugins.get_pre_tool_call_block_message", + "hermes_cli.plugins.resolve_pre_tool_block", lambda *args, **kwargs: "Blocked" if args[0] == "write_file" else None, ) @@ -3308,7 +3308,7 @@ class TestConcurrentToolExecution: messages = [] monkeypatch.setattr( - "hermes_cli.plugins.get_pre_tool_call_block_message", + "hermes_cli.plugins.resolve_pre_tool_block", lambda *args, **kwargs: "Blocked" if args[0] == "patch" else None, ) @@ -3335,7 +3335,7 @@ class TestConcurrentToolExecution: messages = [] monkeypatch.setattr( - "hermes_cli.plugins.get_pre_tool_call_block_message", + "hermes_cli.plugins.resolve_pre_tool_block", lambda *args, **kwargs: "Blocked" if args[0] == "terminal" else None, ) @@ -3369,7 +3369,7 @@ class TestConcurrentToolExecution: return "Blocked" if call_count["n"] == 1 else None monkeypatch.setattr( - "hermes_cli.plugins.get_pre_tool_call_block_message", + "hermes_cli.plugins.resolve_pre_tool_block", block_first_only, ) diff --git a/tests/run_agent/test_tool_call_guardrail_runtime.py b/tests/run_agent/test_tool_call_guardrail_runtime.py index e7ab376281a..36ced43c795 100644 --- a/tests/run_agent/test_tool_call_guardrail_runtime.py +++ b/tests/run_agent/test_tool_call_guardrail_runtime.py @@ -228,7 +228,7 @@ def test_plugin_pre_tool_block_wins_without_counting_as_toolguard_block(): messages = [] with ( - patch("hermes_cli.plugins.get_pre_tool_call_block_message", return_value="plugin policy"), + patch("hermes_cli.plugins.resolve_pre_tool_block", return_value="plugin policy"), patch("run_agent.handle_function_call", return_value="SHOULD_NOT_RUN") as mock_hfc, ): agent._execute_tool_calls_sequential(msg, messages, "task-1") diff --git a/tests/tools/test_request_tool_approval.py b/tests/tools/test_request_tool_approval.py new file mode 100644 index 00000000000..16414fc054c --- /dev/null +++ b/tests/tools/test_request_tool_approval.py @@ -0,0 +1,167 @@ +"""Tests for tools.approval.request_tool_approval — the plugin pre_tool_call +``{"action": "approve"}`` escalation into the human-approval gate. + +These verify that a plugin-driven approval reuses the SAME machinery as a +Tier-2 dangerous-command match: session/permanent allowlist, the CLI prompt, +the gateway submit_pending path, cron_mode, and fail-closed timeouts. +""" + +import pytest + +import tools.approval as approval +from tools.approval import request_tool_approval + + +@pytest.fixture(autouse=True) +def _isolate_approval_state(monkeypatch): + """Give each test a clean session key and empty allowlists.""" + monkeypatch.setattr( + approval, "get_current_session_key", + lambda default="default": "test-session", + ) + # Empty session + permanent approval stores so nothing pre-approves. + monkeypatch.setattr(approval, "is_approved", lambda sk, pk: False) + # Not a yolo session (the shared gate checks this first). + monkeypatch.setattr(approval, "is_current_session_yolo_enabled", lambda: False) + monkeypatch.setattr(approval, "_YOLO_MODE_FROZEN", False, raising=False) + # No thread-registered CLI callback by default. + monkeypatch.setattr( + "tools.terminal_tool._get_approval_callback", lambda: None, raising=False + ) + yield + + +class TestRequestToolApproval: + def test_session_cached_approval_short_circuits(self, monkeypatch): + monkeypatch.setattr(approval, "is_approved", lambda sk, pk: True) + # Should NOT prompt at all. + monkeypatch.setattr( + approval, "prompt_dangerous_approval", + lambda *a, **k: pytest.fail("should not prompt when already approved"), + ) + res = request_tool_approval("write_file", "sensitive path", rule_key="ssh") + assert res == {"approved": True, "message": None} + + def test_cli_approve_once(self, monkeypatch): + monkeypatch.setattr(approval, "_is_interactive_cli", lambda: True) + monkeypatch.setattr(approval, "_is_gateway_approval_context", lambda: False) + monkeypatch.setattr(approval, "prompt_dangerous_approval", lambda *a, **k: "once") + res = request_tool_approval("write_file", "writing ~/.ssh/authorized_keys") + assert res["approved"] is True + + def test_cli_deny_blocks(self, monkeypatch): + monkeypatch.setattr(approval, "_is_interactive_cli", lambda: True) + monkeypatch.setattr(approval, "_is_gateway_approval_context", lambda: False) + monkeypatch.setattr(approval, "prompt_dangerous_approval", lambda *a, **k: "deny") + res = request_tool_approval("terminal", "curl PUT to external API") + assert res["approved"] is False + assert "denied" in res["message"].lower() + assert res["pattern_key"].startswith("plugin_rule:") + + def test_cli_session_persists_session_only(self, monkeypatch): + monkeypatch.setattr(approval, "_is_interactive_cli", lambda: True) + monkeypatch.setattr(approval, "_is_gateway_approval_context", lambda: False) + monkeypatch.setattr(approval, "prompt_dangerous_approval", lambda *a, **k: "session") + calls = {"session": [], "permanent": []} + monkeypatch.setattr(approval, "approve_session", + lambda sk, pk: calls["session"].append(pk)) + monkeypatch.setattr(approval, "approve_permanent", + lambda pk: calls["permanent"].append(pk)) + monkeypatch.setattr(approval, "save_permanent_allowlist", lambda x: None) + res = request_tool_approval("write_file", "reason", rule_key="ssh-writes") + assert res["approved"] is True + assert calls["session"] == ["plugin_rule:ssh-writes"] + assert calls["permanent"] == [] # session != always + + def test_cli_always_persists_permanent(self, monkeypatch): + monkeypatch.setattr(approval, "_is_interactive_cli", lambda: True) + monkeypatch.setattr(approval, "_is_gateway_approval_context", lambda: False) + monkeypatch.setattr(approval, "prompt_dangerous_approval", lambda *a, **k: "always") + persisted = {} + monkeypatch.setattr(approval, "approve_session", lambda sk, pk: None) + monkeypatch.setattr(approval, "approve_permanent", + lambda pk: persisted.setdefault("key", pk)) + monkeypatch.setattr(approval, "save_permanent_allowlist", + lambda x: persisted.setdefault("saved", True)) + res = request_tool_approval("write_file", "reason", rule_key="ssh-writes") + assert res["approved"] is True + assert persisted["key"] == "plugin_rule:ssh-writes" + assert persisted["saved"] is True + + def test_gateway_path_submits_pending_and_defers(self, monkeypatch): + monkeypatch.setattr(approval, "_is_interactive_cli", lambda: False) + monkeypatch.setattr(approval, "_is_gateway_approval_context", lambda: True) + submitted = {} + monkeypatch.setattr(approval, "submit_pending", + lambda sk, data: submitted.update(data)) + res = request_tool_approval("browser_navigate", "external URL", + rule_key="ext-nav") + assert res["approved"] is False + assert res["status"] == "approval_required" + assert submitted["pattern_key"] == "plugin_rule:ext-nav" + + def test_cron_deny_mode_blocks(self, monkeypatch): + monkeypatch.setattr(approval, "_is_interactive_cli", lambda: False) + monkeypatch.setattr(approval, "_is_gateway_approval_context", lambda: False) + monkeypatch.setattr(approval, "env_var_enabled", + lambda v: v == "HERMES_CRON_SESSION") + monkeypatch.setattr(approval, "_get_cron_approval_mode", lambda: "deny") + res = request_tool_approval("terminal", "smtp send") + assert res["approved"] is False + assert "cron" in res["message"].lower() + + def test_cron_approve_mode_allows(self, monkeypatch): + monkeypatch.setattr(approval, "_is_interactive_cli", lambda: False) + monkeypatch.setattr(approval, "_is_gateway_approval_context", lambda: False) + monkeypatch.setattr(approval, "env_var_enabled", + lambda v: v == "HERMES_CRON_SESSION") + monkeypatch.setattr(approval, "_get_cron_approval_mode", lambda: "approve") + res = request_tool_approval("terminal", "smtp send") + assert res["approved"] is True + + def test_rule_key_derived_from_tool_and_reason(self, monkeypatch): + """With no explicit rule_key, the pattern key is derived from + tool + a hash of the reason (so distinct reasons persist apart).""" + monkeypatch.setattr(approval, "_is_interactive_cli", lambda: True) + monkeypatch.setattr(approval, "_is_gateway_approval_context", lambda: False) + monkeypatch.setattr(approval, "prompt_dangerous_approval", lambda *a, **k: "deny") + res = request_tool_approval("patch", "reason") # no rule_key + assert res["pattern_key"].startswith("plugin_rule:patch:") + + def test_distinct_reasons_get_distinct_keys(self, monkeypatch): + """Two different reasons on the SAME tool must not share an [a]lways + allowlist entry (Finding 3: tool_name alone was too coarse).""" + monkeypatch.setattr(approval, "_is_interactive_cli", lambda: True) + monkeypatch.setattr(approval, "_is_gateway_approval_context", lambda: False) + monkeypatch.setattr(approval, "prompt_dangerous_approval", lambda *a, **k: "deny") + k1 = request_tool_approval("write_file", "write to ~/.ssh")["pattern_key"] + k2 = request_tool_approval("write_file", "send an email")["pattern_key"] + assert k1 != k2 + + def test_explicit_rule_key_overrides_derivation(self, monkeypatch): + monkeypatch.setattr(approval, "_is_interactive_cli", lambda: True) + monkeypatch.setattr(approval, "_is_gateway_approval_context", lambda: False) + monkeypatch.setattr(approval, "prompt_dangerous_approval", lambda *a, **k: "deny") + res = request_tool_approval("terminal", "any", rule_key="my-rule") + assert res["pattern_key"] == "plugin_rule:my-rule" + + def test_no_human_non_cron_fails_closed(self, monkeypatch): + """Non-interactive, non-gateway, NON-cron context blocks (fail-closed) + — a plugin-flagged action never runs ungated without a human.""" + monkeypatch.setattr(approval, "_is_interactive_cli", lambda: False) + monkeypatch.setattr(approval, "_is_gateway_approval_context", lambda: False) + monkeypatch.setattr(approval, "env_var_enabled", lambda v: False) # not cron + res = request_tool_approval("terminal", "smtp send") + assert res["approved"] is False + assert "no interactive user or gateway" in res["message"].lower() + + def test_yolo_session_bypasses_gate(self, monkeypatch): + """A --yolo session skips the plugin approval gate (parity with the + dangerous-command path, via the shared _run_approval_gate).""" + monkeypatch.setattr(approval, "is_current_session_yolo_enabled", lambda: True) + monkeypatch.setattr( + approval, "prompt_dangerous_approval", + lambda *a, **k: pytest.fail("yolo must not prompt"), + ) + res = request_tool_approval("terminal", "curl PUT", rule_key="ext") + assert res == {"approved": True, "message": None} diff --git a/tools/approval.py b/tools/approval.py index 53519b5c048..cf9f8b09d62 100644 --- a/tools/approval.py +++ b/tools/approval.py @@ -11,6 +11,7 @@ This module is the single source of truth for the dangerous command system: import contextvars import fnmatch import functools +import hashlib import logging import os import re @@ -1936,6 +1937,158 @@ def _smart_approve(command: str, description: str) -> str: return "escalate" +def _run_approval_gate( + *, + pattern_key: str, + description: str, + display_target: str, + approval_callback=None, + cron_deny_message: str, + autoapprove_log_prefix: str, + fail_closed_when_no_human: bool = False, + no_human_block_message: str = "", +) -> dict: + """Shared human-approval gate for a flagged action (command or tool). + + This is the single decision core reused by both + :func:`check_dangerous_command` (dangerous shell patterns) and + :func:`request_tool_approval` (plugin ``pre_tool_call`` ``approve`` + escalations). Extracting it keeps the fail-closed / cron / gateway / + persist policy in ONE place so the two entry points can never drift. + + Ordering mirrors the historical ``check_dangerous_command`` tail: + yolo bypass → session-cache short-circuit → interactive/gateway/cron + branch → prompt → ``deny/session/always`` persistence. The caller is + responsible for the checks that are specific to its input shape + (hardline detection, command-string permanent allowlist, dangerous- + pattern detection) BEFORE calling this gate. + + Args: + pattern_key: Allowlist/session key this decision is stored under. + description: Human-facing reason shown in the prompt. + display_target: The command string or synthetic tool label shown + to the user (redacted by ``prompt_dangerous_approval``). + approval_callback: Optional CLI prompt callback. When ``None`` the + per-thread callback registered via + ``tools.terminal_tool.set_approval_callback`` is used. + cron_deny_message: Message returned when a cron job hits this gate + under ``cron_mode: deny``. + autoapprove_log_prefix: Log line prefix for the non-interactive + auto-approve warning (identifies command vs plugin origin). + fail_closed_when_no_human: When True, a non-interactive non-gateway + context that is NOT a cron session (e.g. a bare script with + HERMES_INTERACTIVE unset) BLOCKS instead of auto-approving. The + dangerous-command path keeps its historical fail-open default + (False); the plugin-escalation path opts in to fail-closed so a + plugin-flagged action never runs ungated without a human. + no_human_block_message: Message returned when + ``fail_closed_when_no_human`` blocks. + + Returns: + ``{"approved": bool, "message": str|None, ...}`` — shape shared with + ``check_dangerous_command`` so all callers handle it uniformly. + """ + # --yolo bypasses all approval prompts (session- or process-scoped). + # Hardline blocks are handled by the caller BEFORE this gate, so yolo + # here only skips the recoverable approval layer. + if _YOLO_MODE_FROZEN or is_current_session_yolo_enabled(): + return {"approved": True, "message": None} + + session_key = get_current_session_key() + if is_approved(session_key, pattern_key): + return {"approved": True, "message": None} + + if approval_callback is None: + try: + from tools.terminal_tool import _get_approval_callback + approval_callback = _get_approval_callback() + except Exception: + approval_callback = None + + is_cli = _is_interactive_cli() + is_gateway = _is_gateway_approval_context() + + if not is_cli and not is_gateway: + # Cron sessions: respect cron_mode config + if env_var_enabled("HERMES_CRON_SESSION"): + if _get_cron_approval_mode() == "deny": + return { + "approved": False, + "message": cron_deny_message, + "pattern_key": pattern_key, + "description": description, + } + # cron_mode: approve — fall through to auto-approve below. + elif fail_closed_when_no_human: + # Non-cron, non-interactive, no gateway: no human can answer. + # The plugin-escalation path opts in to fail-closed here so a + # plugin-flagged action never runs ungated. (The dangerous- + # command path keeps the historical fail-open default.) + logger.warning( + "%s (pattern: %s): %s — no interactive user/gateway present; " + "BLOCKED (fail-closed). Set HERMES_INTERACTIVE or " + "HERMES_GATEWAY_SESSION to answer the prompt.", + autoapprove_log_prefix, pattern_key, description, + ) + return { + "approved": False, + "message": no_human_block_message or ( + f"BLOCKED: approval required ({description}) but no " + "interactive user or gateway is present to approve it." + ), + "pattern_key": pattern_key, + "description": description, + } + logger.warning( + "%s (pattern: %s): %s — set HERMES_INTERACTIVE or " + "HERMES_GATEWAY_SESSION to require approval.", + autoapprove_log_prefix, pattern_key, description, + ) + return {"approved": True, "message": None} + + if is_gateway or env_var_enabled("HERMES_EXEC_ASK"): + submit_pending(session_key, { + "command": display_target, + "pattern_key": pattern_key, + "description": description, + }) + return { + "approved": False, + "pattern_key": pattern_key, + "status": "approval_required", + "command": display_target, + "description": description, + "message": ( + f"⚠️ This action is potentially dangerous ({description}). " + f"Asking the user for approval.\n\n**Target:**\n```\n{display_target}\n```" + ), + } + + choice = prompt_dangerous_approval(display_target, description, + approval_callback=approval_callback) + + if choice == "deny": + return { + "approved": False, + "message": ( + f"BLOCKED: User denied this potentially dangerous action " + f"(matched '{description}'). Do NOT retry — the user has " + "explicitly rejected it." + ), + "pattern_key": pattern_key, + "description": description, + } + + if choice == "session": + approve_session(session_key, pattern_key) + elif choice == "always": + approve_session(session_key, pattern_key) + approve_permanent(pattern_key) + save_permanent_allowlist(_permanent_approved) + + return {"approved": True, "message": None} + + def _should_skip_container_guards(env_type: str, has_host_access: bool = False) -> bool: """Return True when the backend is isolated enough to skip dangerous-command prompts. @@ -1993,71 +2146,109 @@ def check_dangerous_command(command: str, env_type: str, if not is_dangerous: return {"approved": True, "message": None} - session_key = get_current_session_key() - if is_approved(session_key, pattern_key): - return {"approved": True, "message": None} + return _run_approval_gate( + pattern_key=pattern_key, + description=description, + display_target=command, + approval_callback=approval_callback, + cron_deny_message=( + f"BLOCKED: Command flagged as dangerous ({description}) " + "but cron jobs run without a user present to approve it. " + "Find an alternative approach that avoids this command. " + "To allow dangerous commands in cron jobs, set " + "approvals.cron_mode: approve in config.yaml." + ), + autoapprove_log_prefix=( + "AUTO-APPROVED dangerous command in non-interactive non-gateway context" + ), + ) - is_cli = _is_interactive_cli() - is_gateway = _is_gateway_approval_context() - if not is_cli and not is_gateway: - # Cron sessions: respect cron_mode config - if env_var_enabled("HERMES_CRON_SESSION"): - if _get_cron_approval_mode() == "deny": - return { - "approved": False, - "message": ( - f"BLOCKED: Command flagged as dangerous ({description}) " - "but cron jobs run without a user present to approve it. " - "Find an alternative approach that avoids this command. " - "To allow dangerous commands in cron jobs, set " - "approvals.cron_mode: approve in config.yaml." - ), - } - logger.warning( - "AUTO-APPROVED dangerous command in non-interactive non-gateway context " - "(pattern: %s): %s — set HERMES_INTERACTIVE or HERMES_GATEWAY_SESSION to require approval.", - description, command[:200], - ) - return {"approved": True, "message": None} +def request_tool_approval( + tool_name: str, + reason: str, + *, + rule_key: str = "", + approval_callback=None, +) -> dict: + """Escalate an arbitrary tool call to the human-approval gate. - if is_gateway or env_var_enabled("HERMES_EXEC_ASK"): - submit_pending(session_key, { - "command": command, - "pattern_key": pattern_key, - "description": description, - }) - return { - "approved": False, - "pattern_key": pattern_key, - "status": "approval_required", - "command": command, - "description": description, - "message": ( - f"⚠️ This command is potentially dangerous ({description}). " - f"Asking the user for approval.\n\n**Command:**\n```\n{command}\n```" - ), - } + This is the entry point for a plugin ``pre_tool_call`` hook that returns + ``{"action": "approve", "message": ...}``: instead of the plugin vetoing + the call (``action: block``) or silently allowing it, it asks the SAME + human gate that Tier-2 dangerous shell patterns use. The LLM cannot skip + or bypass this — the tool call is intercepted before execution. - choice = prompt_dangerous_approval(command, description, - approval_callback=approval_callback) + It reuses the existing approval primitives (session/permanent allowlist, + ``prompt_dangerous_approval`` for CLI, ``submit_pending`` for the gateway + callback, ``[o]nce/[s]ession/[a]lways/[d]eny``, timeout fail-closed) so + behavior is identical to a dangerous-command match — only the trigger + (a plugin rule on any tool) differs. - if choice == "deny": - return { - "approved": False, - "message": f"BLOCKED: User denied this potentially dangerous command (matched '{description}' pattern). Do NOT retry this command - the user has explicitly rejected it.", - "pattern_key": pattern_key, - "description": description, - } + Args: + tool_name: The tool being gated (e.g. ``"write_file"``, ``"terminal"``). + reason: Human-facing message from the plugin explaining why approval + is needed (rendered in the prompt). + rule_key: Optional stable identifier the plugin can supply to control + the ``[a]lways`` allowlist grain. When empty, the key is derived + from ``tool_name`` + a hash of ``reason`` so that DISTINCT reasons + on the same tool persist independently (answering ``[a]lways`` to + "write to ~/.ssh" does NOT auto-approve a later "send email" rule + on the same tool). + approval_callback: Optional CLI callback for interactive prompts + (same contract as ``check_dangerous_command``). - if choice == "session": - approve_session(session_key, pattern_key) - elif choice == "always": - approve_session(session_key, pattern_key) - approve_permanent(pattern_key) - save_permanent_allowlist(_permanent_approved) + Returns: + ``{"approved": True, "message": None}`` when allowed, or + ``{"approved": False, "message": , ...}`` when denied / + blocked. Shape matches ``check_dangerous_command`` so callers handle + both paths identically. - return {"approved": True, "message": None} + Non-interactive contexts: cron jobs honor ``approvals.cron_mode`` (parity + with dangerous commands); any OTHER non-interactive non-gateway context + (a bare script with no ``HERMES_INTERACTIVE``) fails CLOSED — a plugin- + flagged action never runs ungated without a human. + """ + description = reason or f"Plugin requires approval for {tool_name}" + # Allowlist grain: an explicit plugin rule_key wins; otherwise derive from + # tool + a short hash of the reason so distinct reasons on the same tool + # get independent [a]lways entries (Finding: rule_key=tool_name alone was + # too coarse — one "always" would blanket every rule on that tool). + if rule_key: + key_suffix = rule_key + else: + _reason_hash = hashlib.sha256(description.encode("utf-8")).hexdigest()[:12] + key_suffix = f"{tool_name}:{_reason_hash}" + # Synthetic pattern key so plugin-rule approvals live in the same + # session/permanent allowlist machinery as command patterns, namespaced + # to avoid ever colliding with a real command pattern key. + pattern_key = f"plugin_rule:{key_suffix}" + # A synthetic "command" string for the display/allowlist layer. It never + # executes; it only labels the gate. Namespaced identically. + display_target = f"<{tool_name}> (plugin approval rule)" + + return _run_approval_gate( + pattern_key=pattern_key, + description=description, + display_target=display_target, + approval_callback=approval_callback, + cron_deny_message=( + f"BLOCKED: Tool '{tool_name}' requires approval ({description}) " + "but cron jobs run without a user present to approve it. Find an " + "alternative approach. To allow flagged actions in cron jobs, set " + "approvals.cron_mode: approve in config.yaml." + ), + autoapprove_log_prefix=( + f"plugin-escalated tool call '{tool_name}' in " + "non-interactive non-gateway context" + ), + fail_closed_when_no_human=True, + no_human_block_message=( + f"BLOCKED: Tool '{tool_name}' requires approval ({description}) " + "but no interactive user or gateway is present to approve it. " + "A plugin flagged this action for human confirmation." + ), + ) # =========================================================================