File size: 14,092 Bytes
7acbd80 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 | """Tests for pre_approval_request / post_approval_response plugin hooks.
These hooks fire in tools/approval.py::check_all_command_guards whenever a
dangerous command needs user approval. They are observer-only (return values
ignored) and must fire on BOTH the CLI-interactive path and the async gateway
path, so external tools like macOS notifiers can be alerted regardless of
which surface the user is on.
"""
from unittest.mock import patch
import pytest
import tools.approval as approval_module
from tools import approval_context
from tools import approval_smart
from tools.approval import check_all_command_guards, check_execute_code_guard, clear_session
from tools.approval_context import set_current_session_key
@pytest.fixture
def isolated_session(monkeypatch, tmp_path):
"""Give each test a fresh session_key, clean approval-state, and isolated
HERMES_HOME so the real user's command_allowlist doesn't leak in."""
import tools.approval as _am
from tools import approval_context
session_key = "test:session:approval_hooks"
token = set_current_session_key(session_key)
monkeypatch.setenv("HERMES_SESSION_KEY", session_key)
# Make sure we don't skip guards via yolo / approvals.mode=off
monkeypatch.delenv("HERMES_YOLO_MODE", raising=False)
# Isolate from the real user's permanent allowlist + session state
_saved_permanent = _am._permanent_approved.copy()
_saved_session = {k: v.copy() for k, v in _am._session_approved.items()}
_am._permanent_approved.clear()
_am._session_approved.clear()
try:
yield session_key
finally:
_am._permanent_approved.update(_saved_permanent)
_am._session_approved.update(_saved_session)
try:
approval_context._approval_session_key.reset(token)
except Exception:
pass
clear_session(session_key)
class TestCliPathFiresHooks:
"""CLI-interactive approval path: HERMES_INTERACTIVE is set, the
prompt_dangerous_approval() result decides the outcome."""
def test_pre_and_post_fire_with_expected_kwargs(
self, isolated_session, monkeypatch
):
monkeypatch.setenv("HERMES_INTERACTIVE", "1")
monkeypatch.delenv("HERMES_GATEWAY_SESSION", raising=False)
monkeypatch.delenv("HERMES_EXEC_ASK", raising=False)
# approvals.mode=manual so we actually reach the prompt site
monkeypatch.setattr(approval_context, "_get_approval_mode", lambda: "manual")
captured = []
def fake_invoke_hook(hook_name, **kwargs):
captured.append((hook_name, kwargs))
return []
# Force the user to "approve once" via the approval_callback contract
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-hook", "local", approval_callback=cb,
)
assert result["approved"] is True
hook_names = [c[0] for c in captured]
assert "pre_approval_request" in hook_names
assert "post_approval_response" in hook_names
pre_kwargs = next(kw for name, kw in captured if name == "pre_approval_request")
assert pre_kwargs["command"] == "rm -rf /tmp/test-hook"
assert pre_kwargs["surface"] == "cli"
assert pre_kwargs["session_key"] == isolated_session
assert isinstance(pre_kwargs["pattern_keys"], list)
assert pre_kwargs["pattern_key"] # non-empty primary pattern
assert pre_kwargs["description"]
post_kwargs = next(kw for name, kw in captured if name == "post_approval_response")
assert post_kwargs["choice"] == "once"
assert post_kwargs["surface"] == "cli"
assert post_kwargs["command"] == "rm -rf /tmp/test-hook"
def test_deny_reported_to_post_hook(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_context, "_get_approval_mode", lambda: "manual")
captured = []
def fake_invoke_hook(hook_name, **kwargs):
captured.append((hook_name, kwargs))
return []
def cb(command, description, *, allow_permanent=True):
return "deny"
with patch("hermes_cli.plugins.invoke_hook", side_effect=fake_invoke_hook):
result = check_all_command_guards(
"rm -rf /tmp/test-deny", "local", approval_callback=cb,
)
assert result["approved"] is False
post_kwargs = next(kw for name, kw in captured if name == "post_approval_response")
assert post_kwargs["choice"] == "deny"
def test_plugin_hook_crash_does_not_break_approval(
self, isolated_session, monkeypatch
):
"""A crashing plugin must never prevent the approval flow from
reaching the user. Hooks are observer-only and safety-critical
behavior must be preserved."""
monkeypatch.setenv("HERMES_INTERACTIVE", "1")
monkeypatch.delenv("HERMES_GATEWAY_SESSION", raising=False)
monkeypatch.delenv("HERMES_EXEC_ASK", raising=False)
monkeypatch.setattr(approval_context, "_get_approval_mode", lambda: "manual")
def boom(hook_name, **kwargs):
raise RuntimeError("plugin crashed")
def cb(command, description, *, allow_permanent=True):
return "once"
with patch("hermes_cli.plugins.invoke_hook", side_effect=boom):
result = check_all_command_guards(
"rm -rf /tmp/test-crash", "local", approval_callback=cb,
)
# User's approval was still honored despite the plugin crashing
assert result["approved"] is True
class TestGatewayPathFiresHooks:
"""Async gateway approval path: HERMES_GATEWAY_SESSION is set and a
gateway notify callback is registered. The agent thread blocks on the
approval event until resolve_gateway_approval() is called from another
thread."""
class TestSmartModeFiresHooks:
def _configure(self, monkeypatch, verdict):
monkeypatch.setenv("HERMES_INTERACTIVE", "1")
monkeypatch.setenv("HERMES_EXEC_ASK", "1")
monkeypatch.delenv("HERMES_GATEWAY_SESSION", raising=False)
monkeypatch.delenv("HERMES_CRON_SESSION", raising=False)
monkeypatch.setattr(approval_module, "_YOLO_MODE_FROZEN", False)
monkeypatch.setattr(approval_context, "_get_approval_mode", lambda: "smart")
monkeypatch.setattr(approval_smart, "_smart_approve", lambda *_: verdict)
monkeypatch.setattr(
"tools.tirith_security.check_command_security",
lambda _: {"action": "allow", "findings": [], "summary": ""},
)
@pytest.mark.parametrize(
("guard", "value", "verdict", "approved", "choice", "pattern_key"),
[
(check_all_command_guards, "rm -rf /tmp/smart-hook", "approve", True, "smart_approve", None),
(check_all_command_guards, "rm -rf /tmp/smart-hook", "deny", False, "smart_deny", None),
(check_execute_code_guard, "print('smart hook')", "approve", True, "smart_approve", "execute_code"),
(check_execute_code_guard, "print('smart hook')", "deny", False, "smart_deny", "execute_code"),
],
)
def test_smart_verdict_fires_redacted_pre_and_post_hooks(
self, isolated_session, monkeypatch, guard, value, verdict, approved, choice, pattern_key
):
self._configure(monkeypatch, verdict)
secret = "sk-ABCDEFGHIJKLMNOPQRSTUVWXYZ012345"
value = f'{value} # Authorization: Bearer {secret}'
captured = []
with patch(
"hermes_cli.plugins.invoke_hook",
side_effect=lambda name, **kwargs: captured.append((name, kwargs)),
):
result = guard(value, "local")
assert result["approved"] is approved
assert result[f"smart_{'approved' if approved else 'denied'}"] is True
assert [name for name, _ in captured] == [
"pre_approval_request",
"post_approval_response",
]
pre, post = (kwargs for _, kwargs in captured)
assert pre["surface"] == post["surface"] == "smart"
assert post["choice"] == choice
assert post["decided_by"] == "aux_llm"
assert pre["session_key"] == post["session_key"] == isolated_session
assert secret not in pre["command"]
assert secret not in post["command"]
assert pre["pattern_keys"]
assert pre["pattern_key"] == post["pattern_key"]
if pattern_key is not None:
assert pre["pattern_key"] == pattern_key
assert pre["pattern_keys"] == [pattern_key]
@pytest.mark.parametrize("guard,value", [
(check_all_command_guards, "rm -rf /tmp/smart-order"),
(check_execute_code_guard, "print('smart order')"),
])
def test_pre_hook_fires_before_aux_llm_decision(
self, isolated_session, monkeypatch, guard, value
):
self._configure(monkeypatch, "approve")
events = []
def decide(*_):
events.append("smart_approve")
return "approve"
monkeypatch.setattr(approval_smart, "_smart_approve", decide)
with patch(
"hermes_cli.plugins.invoke_hook",
side_effect=lambda name, **kwargs: events.append(name),
):
result = guard(value, "local")
assert result["approved"] is True
assert events == [
"pre_approval_request",
"smart_approve",
"post_approval_response",
]
@pytest.mark.parametrize("guard,value", [
(check_all_command_guards, "rm -rf /tmp/smart-force-redaction"),
(check_execute_code_guard, "print('smart force redaction')"),
])
def test_smart_observer_redaction_is_forced_when_config_disables_redaction(
self, isolated_session, monkeypatch, guard, value
):
self._configure(monkeypatch, "approve")
force_values = []
def redact(text, *, force=False):
force_values.append(force)
return f"redacted:{text}"
with (
patch("agent.redact.redact_sensitive_text", side_effect=redact),
patch("hermes_cli.plugins.invoke_hook"),
):
result = guard(value, "local")
assert result["approved"] is True
assert force_values == [True, True]
@pytest.mark.parametrize("guard,value", [
(check_all_command_guards, "rm -rf /tmp/smart-hook-crash"),
(check_execute_code_guard, "print('smart hook crash')"),
])
@pytest.mark.parametrize("verdict,approved", [("approve", True), ("deny", False)])
def test_observer_exception_never_changes_smart_verdict(
self, isolated_session, monkeypatch, guard, value, verdict, approved
):
self._configure(monkeypatch, verdict)
with patch(
"hermes_cli.plugins.invoke_hook",
side_effect=RuntimeError("observer failed"),
):
result = guard(value, "local")
assert result["approved"] is approved
@pytest.mark.parametrize("guard,value", [
(check_all_command_guards, "rm -rf /tmp/smart-redactor-crash"),
(check_execute_code_guard, "print('smart redactor crash')"),
])
@pytest.mark.parametrize("verdict,approved", [("approve", True), ("deny", False)])
def test_redactor_exception_never_changes_smart_verdict_or_leaks_payload(
self, isolated_session, monkeypatch, guard, value, verdict, approved
):
self._configure(monkeypatch, verdict)
captured = []
def fail_observer_redaction(text, *, force=False):
if force:
raise RuntimeError("observer redactor failed")
return text
with (
patch("agent.redact.redact_sensitive_text", side_effect=fail_observer_redaction),
patch(
"hermes_cli.plugins.invoke_hook",
side_effect=lambda name, **kwargs: captured.append((name, kwargs)),
),
):
result = guard(value, "local")
assert result["approved"] is approved
assert captured == []
@pytest.mark.parametrize("guard,first_value,second_value", [
(
check_all_command_guards,
"rm -rf /tmp/first-smart-command",
"rm -rf /tmp/second-smart-command",
),
(
check_execute_code_guard,
"print('first smart script')",
"print('second smart script')",
),
])
def test_smart_approval_is_per_command(
self, isolated_session, monkeypatch, guard, first_value, second_value
):
verdicts = iter(("approve", "deny"))
monkeypatch.setenv("HERMES_INTERACTIVE", "1")
monkeypatch.setenv("HERMES_EXEC_ASK", "1")
monkeypatch.delenv("HERMES_GATEWAY_SESSION", raising=False)
monkeypatch.setattr(approval_module, "_YOLO_MODE_FROZEN", False)
monkeypatch.setattr(approval_context, "_get_approval_mode", lambda: "smart")
monkeypatch.setattr(approval_smart, "_smart_approve", lambda *_: next(verdicts))
monkeypatch.setattr(
"tools.tirith_security.check_command_security",
lambda _: {"action": "allow", "findings": [], "summary": ""},
)
captured = []
with patch(
"hermes_cli.plugins.invoke_hook",
side_effect=lambda name, **kwargs: captured.append((name, kwargs)),
):
first = guard(first_value, "local")
second = guard(second_value, "local")
assert first["approved"] is True
assert second["approved"] is False
assert [kwargs["choice"] for name, kwargs in captured if name == "post_approval_response"] == [
"smart_approve",
"smart_deny",
]
|