diff --git a/hermes_cli/plugins.py b/hermes_cli/plugins.py index ea0b8ea2ffe1..b3387854aad3 100644 --- a/hermes_cli/plugins.py +++ b/hermes_cli/plugins.py @@ -172,17 +172,20 @@ def _install_plugin_debug_handler(force: bool = False) -> None: # Kwargs: event: MessageEvent, gateway: GatewayRunner, session_store. "pre_gateway_dispatch", # Approval lifecycle hooks. Fired by tools/approval.py when a dangerous - # command needs user approval -- fires BOTH for CLI-interactive prompts - # and for gateway/ACP approvals (Telegram, Discord, Slack, TUI, etc.). - # Observers only: return values are ignored. Plugins cannot veto or - # pre-answer an approval from these hooks (use pre_tool_call to block - # a tool before it reaches approval). + # command needs an approval decision -- fires for CLI-interactive prompts, + # gateway/ACP approvals (Telegram, Discord, Slack, TUI, etc.), AND + # approvals.mode=smart auto approve/deny decisions made by the auxiliary + # LLM (surface="smart"). Observers only: return values are ignored. + # Plugins cannot veto or pre-answer an approval from these hooks (use + # pre_tool_call to block a tool before it reaches approval). # # Kwargs for pre_approval_request: # command: str, description: str, pattern_key: str, pattern_keys: list[str], - # session_key: str, surface: "cli" | "gateway" + # session_key: str, surface: "cli" | "gateway" | "smart" # Kwargs for post_approval_response: same as above plus # choice: "once" | "session" | "always" | "deny" | "timeout" + # | "smart_approve" | "smart_deny" + # decided_by: "aux_llm" -- present only on surface="smart" responses "pre_approval_request", "post_approval_response", # Kanban task lifecycle hooks. Fired by hermes_cli.kanban_db when a task diff --git a/tests/tools/test_approval_plugin_hooks.py b/tests/tools/test_approval_plugin_hooks.py index 58ccb2f8a76f..bb4eaf5c4d3f 100644 --- a/tests/tools/test_approval_plugin_hooks.py +++ b/tests/tools/test_approval_plugin_hooks.py @@ -13,6 +13,7 @@ import tools.approval as approval_module from tools.approval import ( check_all_command_guards, + check_execute_code_guard, set_current_session_key, clear_session, ) @@ -150,3 +151,283 @@ class TestGatewayPathFiresHooks: thread.""" +class TestSmartModeFiresHooks: + """approvals.mode=smart auto-approve/auto-deny decisions are real approval + outcomes and must fire the same pre/post hooks as the manual and gateway + surfaces, so observers (nemo_relay, notifiers, audit) don't miss the + majority of decisions made on smart-mode surfaces. + + Regression for: smart auto-approve/deny returned BEFORE ever calling + _fire_approval_hook, so every approval observer silently missed them. + + Covers BOTH _smart_approve call sites: + * check_all_command_guards (terminal command guard) + * check_execute_code_guard (execute_code whole-script guard) + """ + + def _capture(self): + captured = [] + + def fake_invoke_hook(hook_name, **kwargs): + captured.append((hook_name, kwargs)) + return [] + + return captured, fake_invoke_hook + + # -- check_all_command_guards (main terminal guard) ------------------- + + def test_command_guard_auto_approve_fires_hooks( + self, isolated_session, monkeypatch + ): + monkeypatch.setenv("HERMES_INTERACTIVE", "1") + monkeypatch.delenv("HERMES_GATEWAY_SESSION", raising=False) + monkeypatch.delenv("HERMES_EXEC_ASK", raising=False) + monkeypatch.setattr(approval_module, "_get_approval_mode", lambda: "smart") + monkeypatch.setattr(approval_module, "_smart_approve", lambda c, d: "approve") + + captured, fake_invoke_hook = self._capture() + with patch("hermes_cli.plugins.invoke_hook", side_effect=fake_invoke_hook): + result = check_all_command_guards("rm -rf /tmp/test-smart-ok", "local") + + # Behavior unchanged: the aux-LLM's approve still auto-approves. + assert result["approved"] is True + assert result.get("smart_approved") is True + + names = [n for n, _ in captured] + assert "pre_approval_request" in names + assert "post_approval_response" in names + + pre = next(kw for n, kw in captured if n == "pre_approval_request") + assert pre["surface"] == "smart" + assert pre["command"] == "rm -rf /tmp/test-smart-ok" + assert pre["session_key"] == isolated_session + assert isinstance(pre["pattern_keys"], list) and pre["pattern_keys"] + assert pre["pattern_key"] + assert pre["description"] + + post = next(kw for n, kw in captured if n == "post_approval_response") + assert post["surface"] == "smart" + assert post["choice"] == "smart_approve" + assert post["decided_by"] == "aux_llm" + assert post["command"] == "rm -rf /tmp/test-smart-ok" + + def test_command_guard_auto_deny_fires_hooks( + self, isolated_session, monkeypatch + ): + monkeypatch.setenv("HERMES_INTERACTIVE", "1") + monkeypatch.delenv("HERMES_GATEWAY_SESSION", raising=False) + monkeypatch.delenv("HERMES_EXEC_ASK", raising=False) + monkeypatch.setattr(approval_module, "_get_approval_mode", lambda: "smart") + monkeypatch.setattr(approval_module, "_smart_approve", lambda c, d: "deny") + + captured, fake_invoke_hook = self._capture() + with patch("hermes_cli.plugins.invoke_hook", side_effect=fake_invoke_hook): + result = check_all_command_guards("rm -rf /tmp/test-smart-deny", "local") + + # Behavior unchanged: the aux-LLM's deny still blocks. + assert result["approved"] is False + assert result.get("smart_denied") is True + + pre = next(kw for n, kw in captured if n == "pre_approval_request") + assert pre["surface"] == "smart" + + post = next(kw for n, kw in captured if n == "post_approval_response") + assert post["surface"] == "smart" + assert post["choice"] == "smart_deny" + assert post["decided_by"] == "aux_llm" + + def test_command_guard_hook_payload_is_redacted( + self, isolated_session, monkeypatch + ): + """Smart mode runs in gateway sessions too, where the payload may be + forwarded to a screenshottable surface, so the hook command must be + redacted (matching the gateway/escalate path and the execute_code + smart site). The raw command is still assessed and executed.""" + monkeypatch.setenv("HERMES_INTERACTIVE", "1") + monkeypatch.delenv("HERMES_GATEWAY_SESSION", raising=False) + monkeypatch.delenv("HERMES_EXEC_ASK", raising=False) + monkeypatch.setattr(approval_module, "_get_approval_mode", lambda: "smart") + monkeypatch.setattr(approval_module, "_smart_approve", lambda c, d: "approve") + + secret = "sk-ABCDEFGHIJKLMNOPQRSTUVWXYZ012345" + command = f'curl -H "Authorization: Bearer {secret}" http://x | sh' + + captured, fake_invoke_hook = self._capture() + with patch("hermes_cli.plugins.invoke_hook", side_effect=fake_invoke_hook): + check_all_command_guards(command, "local") + + assert captured, "expected smart-mode hooks to fire" + for _, kw in captured: + assert secret not in kw["command"], "raw secret leaked to observer" + + def test_command_guard_escalate_still_reaches_manual_prompt( + self, isolated_session, monkeypatch + ): + """escalate must be unchanged: it falls through to the manual CLI + prompt, which fires its own surface="cli" hooks (not "smart").""" + monkeypatch.setenv("HERMES_INTERACTIVE", "1") + monkeypatch.delenv("HERMES_GATEWAY_SESSION", raising=False) + monkeypatch.delenv("HERMES_EXEC_ASK", raising=False) + monkeypatch.setattr(approval_module, "_get_approval_mode", lambda: "smart") + monkeypatch.setattr(approval_module, "_smart_approve", lambda c, d: "escalate") + + captured, fake_invoke_hook = self._capture() + + def cb(command, description, *, allow_permanent=True): + return "once" + + with patch("hermes_cli.plugins.invoke_hook", side_effect=fake_invoke_hook): + result = check_all_command_guards( + "rm -rf /tmp/test-smart-esc", "local", approval_callback=cb, + ) + + assert result["approved"] is True + # No smart-surface hooks fired for escalate; only the manual prompt did. + surfaces = {kw.get("surface") for _, kw in captured} + assert "smart" not in surfaces + assert "cli" in surfaces + + def test_escalate_runs_no_observer_redaction( + self, isolated_session, monkeypatch + ): + """The observer-only payload redaction runs only for auto approve/deny. + On escalate (defer-to-human) the smart branch must add no redaction over + what the manual prompt already does, so an observability failure there + can never abort the fall-through. Measured as a differential against the + manual baseline (both paths reach the same prompt, which redaction is + load-bearing for), so this stays robust if the prompt's own redaction + changes.""" + monkeypatch.setenv("HERMES_INTERACTIVE", "1") + monkeypatch.delenv("HERMES_GATEWAY_SESSION", raising=False) + monkeypatch.delenv("HERMES_EXEC_ASK", raising=False) + + import agent.redact as _redact + real_redact = _redact.redact_sensitive_text + calls = {"n": 0} + + def counting_redact(text, *a, **k): + calls["n"] += 1 + return real_redact(text, *a, **k) + + monkeypatch.setattr("agent.redact.redact_sensitive_text", counting_redact) + + def cb(command, description, *, allow_permanent=True): + return "once" + + cmd = "rm -rf /tmp/test-smart-esc-count" + noop_hook = patch("hermes_cli.plugins.invoke_hook", side_effect=lambda *a, **k: []) + + # Baseline: manual mode reaches the prompt and does its own redaction. + monkeypatch.setattr(approval_module, "_get_approval_mode", lambda: "manual") + with noop_hook: + r_manual = check_all_command_guards(cmd, "local", approval_callback=cb) + baseline = calls["n"] + + # Smart mode that escalates hits the same prompt and must add no + # redaction over that baseline (pre-fix it redacted twice more, up front). + calls["n"] = 0 + monkeypatch.setattr(approval_module, "_get_approval_mode", lambda: "smart") + monkeypatch.setattr(approval_module, "_smart_approve", lambda c, d: "escalate") + with noop_hook: + r_smart = check_all_command_guards(cmd, "local", approval_callback=cb) + + assert r_manual["approved"] is True + assert r_smart["approved"] is True + assert baseline > 0 # sanity: the prompt really does redact + assert calls["n"] == baseline # escalate added no observer redaction + + # -- check_execute_code_guard (whole-script guard) ------------------- + + def test_execute_code_auto_approve_fires_hooks( + self, isolated_session, monkeypatch + ): + # HERMES_EXEC_ASK gives execute_code an approval surface without a + # gateway notify callback; smart approve/deny return before Phase 3. + monkeypatch.setenv("HERMES_EXEC_ASK", "1") + monkeypatch.delenv("HERMES_GATEWAY_SESSION", raising=False) + monkeypatch.delenv("HERMES_CRON_SESSION", raising=False) + monkeypatch.setattr(approval_module, "_get_approval_mode", lambda: "smart") + monkeypatch.setattr(approval_module, "_smart_approve", lambda c, d: "approve") + + captured, fake_invoke_hook = self._capture() + with patch("hermes_cli.plugins.invoke_hook", side_effect=fake_invoke_hook): + result = check_execute_code_guard("import os", "local") + + assert result["approved"] is True + assert result.get("smart_approved") is True + + pre = next(kw for n, kw in captured if n == "pre_approval_request") + assert pre["surface"] == "smart" + assert pre["pattern_key"] == "execute_code" + assert pre["pattern_keys"] == ["execute_code"] + assert pre["session_key"] == isolated_session + + post = next(kw for n, kw in captured if n == "post_approval_response") + assert post["surface"] == "smart" + assert post["choice"] == "smart_approve" + assert post["decided_by"] == "aux_llm" + + def test_execute_code_auto_deny_fires_hooks( + self, isolated_session, monkeypatch + ): + monkeypatch.setenv("HERMES_EXEC_ASK", "1") + monkeypatch.delenv("HERMES_GATEWAY_SESSION", raising=False) + monkeypatch.delenv("HERMES_CRON_SESSION", raising=False) + monkeypatch.setattr(approval_module, "_get_approval_mode", lambda: "smart") + monkeypatch.setattr(approval_module, "_smart_approve", lambda c, d: "deny") + + captured, fake_invoke_hook = self._capture() + with patch("hermes_cli.plugins.invoke_hook", side_effect=fake_invoke_hook): + result = check_execute_code_guard("import os", "local") + + assert result["approved"] is False + assert result.get("smart_denied") is True + + post = next(kw for n, kw in captured if n == "post_approval_response") + assert post["surface"] == "smart" + assert post["choice"] == "smart_deny" + assert post["decided_by"] == "aux_llm" + + def test_execute_code_hook_payload_is_redacted( + self, isolated_session, monkeypatch + ): + """execute_code scripts can embed secrets and the payload is forwarded + to observers, so the hook command must use the redacted display copy + (matching the execute_code gateway path).""" + monkeypatch.setenv("HERMES_EXEC_ASK", "1") + monkeypatch.delenv("HERMES_GATEWAY_SESSION", raising=False) + monkeypatch.delenv("HERMES_CRON_SESSION", raising=False) + monkeypatch.setattr(approval_module, "_get_approval_mode", lambda: "smart") + monkeypatch.setattr(approval_module, "_smart_approve", lambda c, d: "approve") + + secret = "sk-ABCDEFGHIJKLMNOPQRSTUVWXYZ012345" + code = f'api_key = "{secret}"\nprint(api_key)' + + captured, fake_invoke_hook = self._capture() + with patch("hermes_cli.plugins.invoke_hook", side_effect=fake_invoke_hook): + check_execute_code_guard(code, "local") + + for _, kw in captured: + assert secret not in kw["command"], "raw secret leaked to observer" + + def test_hook_crash_does_not_change_smart_verdict( + self, isolated_session, monkeypatch + ): + """Fail-open: a crashing observer must never flip the smart verdict or + the return value. Hooks are pure observers.""" + monkeypatch.setenv("HERMES_INTERACTIVE", "1") + monkeypatch.delenv("HERMES_GATEWAY_SESSION", raising=False) + monkeypatch.delenv("HERMES_EXEC_ASK", raising=False) + monkeypatch.setattr(approval_module, "_get_approval_mode", lambda: "smart") + monkeypatch.setattr(approval_module, "_smart_approve", lambda c, d: "approve") + + def boom(hook_name, **kwargs): + raise RuntimeError("observer crashed") + + with patch("hermes_cli.plugins.invoke_hook", side_effect=boom): + result = check_all_command_guards("rm -rf /tmp/test-smart-boom", "local") + + assert result["approved"] is True + assert result.get("smart_approved") is True + + diff --git a/tools/approval.py b/tools/approval.py index 05f2eb523cc3..94c24ce2370d 100644 --- a/tools/approval.py +++ b/tools/approval.py @@ -2744,17 +2744,63 @@ def check_all_command_guards(command: str, env_type: str, if approval_mode == "smart": combined_desc_for_llm = "; ".join(desc for _, desc, _ in warnings) verdict = _smart_approve(command, combined_desc_for_llm) + # Fire the same approval hooks as the CLI/gateway surfaces so observers + # (nemo_relay, notifiers, audit) see smart-mode auto decisions too, not + # just escalations. Observers only: _fire_approval_hook swallows errors, + # and the verdict, return value, and approve_session ordering below are + # unchanged. surface="smart" and the smart_* choices are additive. The + # payload prep lives inside this gate so the escalate path never runs + # observer-only code: it falls straight through to the manual prompt and + # an observability failure can't abort a defer-to-human. Redacted like + # the gateway path because smart mode also runs in gateway sessions that + # forward it off-box; the raw command is still what runs. + if verdict in ("approve", "deny"): + from agent.redact import redact_sensitive_text + _hook_command = redact_sensitive_text(command) + _hook_desc = redact_sensitive_text(combined_desc_for_llm) + _primary_key = warnings[0][0] + _all_keys = [key for key, _, _ in warnings] + _fire_approval_hook( + "pre_approval_request", + command=_hook_command, + description=_hook_desc, + pattern_key=_primary_key, + pattern_keys=_all_keys, + session_key=session_key, + surface="smart", + ) if verdict == "approve": # Auto-approve and grant session-level approval for these patterns for key, _, _ in warnings: approve_session(session_key, key) logger.debug("Smart approval: auto-approved '%s' (%s)", command[:60], combined_desc_for_llm) + _fire_approval_hook( + "post_approval_response", + command=_hook_command, + description=_hook_desc, + pattern_key=_primary_key, + pattern_keys=_all_keys, + session_key=session_key, + surface="smart", + choice="smart_approve", + decided_by="aux_llm", + ) return {"approved": True, "message": None, "smart_approved": True, "description": combined_desc_for_llm} elif verdict == "deny": - combined_desc_for_llm = "; ".join(desc for _, desc, _ in warnings) + _fire_approval_hook( + "post_approval_response", + command=_hook_command, + description=_hook_desc, + pattern_key=_primary_key, + pattern_keys=_all_keys, + session_key=session_key, + surface="smart", + choice="smart_deny", + decided_by="aux_llm", + ) return { "approved": False, "message": f"BLOCKED by smart approval: {combined_desc_for_llm}. " @@ -3048,12 +3094,49 @@ def check_execute_code_guard(code: str, env_type: str, # guards (restored by context propagation) still run independently. if approval_mode == "smart": verdict = _smart_approve(command, description) + # Fire the same approval hooks the gateway surface fires so observers + # see smart-mode execute_code decisions too. Observers only: + # _fire_approval_hook swallows errors and the verdict/return value are + # unchanged. Reuse the redacted display copies (a script can embed + # secrets), matching the gateway path. surface="smart". + if verdict in ("approve", "deny"): + _fire_approval_hook( + "pre_approval_request", + command=display_command, + description=display_description, + pattern_key=pattern_key, + pattern_keys=[pattern_key], + session_key=session_key, + surface="smart", + ) if verdict == "approve": logger.debug("Smart approval: auto-approved execute_code for session %s", session_key) + _fire_approval_hook( + "post_approval_response", + command=display_command, + description=display_description, + pattern_key=pattern_key, + pattern_keys=[pattern_key], + session_key=session_key, + surface="smart", + choice="smart_approve", + decided_by="aux_llm", + ) return {"approved": True, "message": None, "smart_approved": True, "description": description} if verdict == "deny": + _fire_approval_hook( + "post_approval_response", + command=display_command, + description=display_description, + pattern_key=pattern_key, + pattern_keys=[pattern_key], + session_key=session_key, + surface="smart", + choice="smart_deny", + decided_by="aux_llm", + ) return { "approved": False, "message": ("BLOCKED by smart approval: execute_code script " diff --git a/website/docs/user-guide/features/hooks.md b/website/docs/user-guide/features/hooks.md index 224d20198b30..ca1d89d5c3bb 100644 --- a/website/docs/user-guide/features/hooks.md +++ b/website/docs/user-guide/features/hooks.md @@ -1060,7 +1060,7 @@ def register(ctx): ### `pre_approval_request` -Fires **immediately before** an approval request is shown to the user — covers every surface: interactive CLI, the Ink TUI, gateway platforms (Telegram, Discord, Slack, WhatsApp, Matrix, etc.), and ACP clients (VS Code, Zed, JetBrains). +Fires **immediately before** an approval request is shown to the user — covers every surface: interactive CLI, the Ink TUI, gateway platforms (Telegram, Discord, Slack, WhatsApp, Matrix, etc.), and ACP clients (VS Code, Zed, JetBrains). In `approvals.mode=smart` it also fires when the auxiliary LLM auto-approves or auto-denies (no human is prompted), with `surface="smart"`. This is the right place to wire a custom notifier — for example, a macOS menu-bar app that pops an allow/deny notification, or an audit log that records every approval request with context. @@ -1085,7 +1085,7 @@ def my_callback( | `pattern_key` | `str` | Primary pattern key that triggered the approval (e.g. `"rm_rf"`, `"sudo"`) | | `pattern_keys` | `list[str]` | All pattern keys that matched | | `session_key` | `str` | Session identifier, useful for scoping notifications per-chat | -| `surface` | `str` | `"cli"` for interactive CLI/TUI prompts, `"gateway"` for async platform approvals | +| `surface` | `str` | `"cli"` for interactive CLI/TUI prompts, `"gateway"` for async platform approvals, `"smart"` for aux-LLM auto approve/deny decisions (`approvals.mode=smart`) | **Return value:** ignored. Hooks here are observer-only; they cannot veto or pre-answer the approval. Use [`pre_tool_call`](#pre_tool_call) to block a tool before it reaches the approval system. @@ -1133,7 +1133,8 @@ Same kwargs as `pre_approval_request`, plus: | Parameter | Type | Description | |-----------|------|-------------| -| `choice` | `str` | One of `"once"`, `"session"`, `"always"`, `"deny"`, or `"timeout"` | +| `choice` | `str` | One of `"once"`, `"session"`, `"always"`, `"deny"`, or `"timeout"` for prompted surfaces, or `"smart_approve"` / `"smart_deny"` for `surface="smart"` auto decisions | +| `decided_by` | `str` | Present only on `surface="smart"` responses: `"aux_llm"` when the auxiliary LLM decided. Absent on CLI/gateway responses. | **Return value:** ignored.