hermes-agent/tests/tools/test_request_tool_approval.py
kshitijk4poor f512d6f020 feat(plugins): pre_tool_call approve action escalates to human gate
Extend the pre_tool_call plugin hook return contract with a new directive:

    {"action": "approve", "message": "why this needs human confirmation"}

Previously a pre_tool_call hook could only veto a tool call (action: block)
or allow it silently. It could not escalate to the existing human-approval
flow. This unlocks user-defined runtime approval rules on ANY tool (HTTP
writes, file writes to sensitive paths, email sends), enforced at runtime —
resolving #51221 as a pure plugin, with no core approval.py rule schema.

Mechanism:
- get_pre_tool_call_directive() returns (action, message) for block|approve;
  get_pre_tool_call_block_message() kept as a block-only back-compat shim.
- resolve_pre_tool_block() is the single dispatch-site chokepoint: fetches
  the directive and, for approve, invokes the human gate; fail-closed to a
  block on denial, timeout, or gate exception. ALL FOUR tool-dispatch sites
  now call it: tool_executor (concurrent + sequential), agent_runtime_helpers,
  and model_tools.handle_function_call.
- request_tool_approval() escalates via the SAME machinery as Tier-2
  dangerous commands: session/permanent allowlist, prompt_dangerous_approval
  (CLI) / submit_pending (gateway), [o]nce/[s]ession/[a]lways/[d]eny,
  timeout fail-closed, approvals.cron_mode for cron contexts.

Architecture: extracted the shared decision core into _run_approval_gate(),
called by BOTH check_dangerous_command() and request_tool_approval() so the
fail-closed / cron / gateway / yolo / persist policy lives in ONE place and
cannot drift. Fixed a latent divergence — the plugin path now honors --yolo.

Approval grain: [a]lways is keyed on tool_name + a hash of the reason (an
explicit plugin rule_key overrides), so distinct reasons on the same tool
persist independently instead of one 'always' blanketing the whole tool.

Non-interactive: cron honors approvals.cron_mode (parity with commands); any
other non-interactive non-gateway context fails CLOSED for the plugin path
(the command path keeps its historical fail-open default, unchanged).

No new config schema, no new env vars, no new hook events.
2026-07-05 12:48:11 +05:30

167 lines
9.1 KiB
Python

"""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}