mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-31 19:16:29 +00:00
feat(approvals): add per-tool policy rules
This commit is contained in:
teknium1
parent
a7f65e3bcd
commit
da9e9bc876
4 changed files with 508 additions and 12 deletions
|
|
@ -2534,6 +2534,13 @@ DEFAULT_CONFIG = {
|
|||
"mode": "manual",
|
||||
"timeout": 60,
|
||||
"cron_mode": "deny",
|
||||
# Declarative per-tool approval policy. Keys are case-insensitive
|
||||
# fnmatch globs matched against both the tool name and its registered
|
||||
# toolset name; values are deny, ask, or allow. When several patterns
|
||||
# match, the most restrictive decision wins (deny > ask > allow).
|
||||
# Explicit allow never bypasses command hardlines/user denies, session
|
||||
# tool scope, cron policy, plugin escalation, or path protections.
|
||||
"tool_policies": {},
|
||||
# User-defined deny rules: fnmatch globs matched against terminal
|
||||
# commands. A match blocks the command unconditionally — BEFORE the
|
||||
# --yolo / /yolo / mode=off bypass — making this the user-editable
|
||||
|
|
|
|||
|
|
@ -7,6 +7,9 @@ contract helpers here so agent-loop call sites and plugins share one vocabulary.
|
|||
|
||||
from __future__ import annotations
|
||||
|
||||
import contextvars
|
||||
import fnmatch
|
||||
import json
|
||||
import logging
|
||||
from copy import deepcopy
|
||||
from dataclasses import dataclass, field
|
||||
|
|
@ -14,6 +17,11 @@ from typing import Any, Callable, Dict, List, Optional
|
|||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
_TOOL_POLICY_PRIORITY = {"allow": 0, "ask": 1, "deny": 2}
|
||||
_active_tool_policy_checks: contextvars.ContextVar[tuple[str, ...]] = (
|
||||
contextvars.ContextVar("active_tool_policy_checks", default=())
|
||||
)
|
||||
|
||||
OBSERVER_SCHEMA_VERSION = "hermes.observer.v1"
|
||||
MIDDLEWARE_SCHEMA_VERSION = "hermes.middleware.v1"
|
||||
|
||||
|
|
@ -195,19 +203,129 @@ def run_tool_execution_middleware(
|
|||
next_call: Callable[[Dict[str, Any]], Any],
|
||||
**context: Any,
|
||||
) -> Any:
|
||||
"""Run tool execution through registered tool execution middleware."""
|
||||
callbacks = _get_middleware_callbacks(TOOL_EXECUTION_MIDDLEWARE)
|
||||
if not callbacks:
|
||||
return next_call(args)
|
||||
return _run_execution_chain(
|
||||
TOOL_EXECUTION_MIDDLEWARE,
|
||||
callbacks,
|
||||
next_call,
|
||||
tool_name=tool_name,
|
||||
args=args,
|
||||
original_args=context.pop("original_args", args),
|
||||
**context,
|
||||
"""Run a tool through the shared policy gate and plugin middleware chain."""
|
||||
active_checks = _active_tool_policy_checks.get()
|
||||
policy_already_checked = tool_name in active_checks
|
||||
token = None
|
||||
terminal_allow_token = None
|
||||
if not policy_already_checked:
|
||||
policy = resolve_tool_approval_policy(tool_name)
|
||||
blocked = _apply_tool_approval_policy(tool_name, policy)
|
||||
if blocked is not None:
|
||||
return blocked
|
||||
token = _active_tool_policy_checks.set((*active_checks, tool_name))
|
||||
if tool_name == "terminal" and policy == "allow":
|
||||
from tools.approval import set_tool_policy_terminal_allow
|
||||
|
||||
terminal_allow_token = set_tool_policy_terminal_allow()
|
||||
|
||||
try:
|
||||
callbacks = _get_middleware_callbacks(TOOL_EXECUTION_MIDDLEWARE)
|
||||
if not callbacks:
|
||||
return next_call(args)
|
||||
return _run_execution_chain(
|
||||
TOOL_EXECUTION_MIDDLEWARE,
|
||||
callbacks,
|
||||
next_call,
|
||||
tool_name=tool_name,
|
||||
args=args,
|
||||
original_args=context.pop("original_args", args),
|
||||
**context,
|
||||
)
|
||||
finally:
|
||||
if terminal_allow_token is not None:
|
||||
from tools.approval import reset_tool_policy_terminal_allow
|
||||
|
||||
reset_tool_policy_terminal_allow(terminal_allow_token)
|
||||
if token is not None:
|
||||
_active_tool_policy_checks.reset(token)
|
||||
|
||||
|
||||
def resolve_tool_approval_policy(tool_name: str) -> Optional[str]:
|
||||
"""Resolve glob policies for a tool name and its registered toolset."""
|
||||
try:
|
||||
from hermes_cli.config import load_config
|
||||
|
||||
config = load_config() or {}
|
||||
approvals = config.get("approvals", {}) or {}
|
||||
patterns = approvals.get("tool_policies", {}) or {}
|
||||
except Exception as exc:
|
||||
logger.warning("Failed to load tool approval policies: %s", exc)
|
||||
return None
|
||||
if not isinstance(patterns, dict):
|
||||
return None
|
||||
|
||||
identifiers = [str(tool_name or "").strip().lower()]
|
||||
try:
|
||||
from tools.registry import registry
|
||||
|
||||
entry = registry.get_entry(tool_name)
|
||||
toolset = str(getattr(entry, "toolset", "") or "").strip().lower()
|
||||
if toolset:
|
||||
identifiers.append(toolset)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
matches: list[str] = []
|
||||
for pattern, raw_policy in patterns.items():
|
||||
if not isinstance(pattern, str) or not isinstance(raw_policy, str):
|
||||
continue
|
||||
normalized_pattern = pattern.strip().lower()
|
||||
policy = raw_policy.strip().lower()
|
||||
if not normalized_pattern or policy not in _TOOL_POLICY_PRIORITY:
|
||||
continue
|
||||
if any(
|
||||
fnmatch.fnmatchcase(identifier, normalized_pattern)
|
||||
for identifier in identifiers
|
||||
):
|
||||
matches.append(policy)
|
||||
if not matches:
|
||||
return None
|
||||
return max(matches, key=_TOOL_POLICY_PRIORITY.__getitem__)
|
||||
|
||||
|
||||
def _apply_tool_approval_policy(
|
||||
tool_name: str, policy: Optional[str]
|
||||
) -> Optional[str]:
|
||||
"""Return a synthetic blocked result, or None to continue execution."""
|
||||
if policy == "deny":
|
||||
return json.dumps(
|
||||
{
|
||||
"error": (
|
||||
f"BLOCKED: Tool '{tool_name}' is denied by "
|
||||
"approvals.tool_policies in config.yaml."
|
||||
),
|
||||
"policy": "deny",
|
||||
"tool": tool_name,
|
||||
},
|
||||
ensure_ascii=False,
|
||||
)
|
||||
if policy != "ask":
|
||||
return None
|
||||
|
||||
try:
|
||||
from tools.approval import request_tool_approval
|
||||
|
||||
decision = request_tool_approval(
|
||||
tool_name,
|
||||
f"config.yaml requires approval for tool '{tool_name}'",
|
||||
rule_key=f"tool_policy:{tool_name}",
|
||||
)
|
||||
except Exception as exc:
|
||||
logger.warning("Tool approval policy check failed for %s: %s", tool_name, exc)
|
||||
decision = {
|
||||
"approved": False,
|
||||
"message": "BLOCKED: tool approval policy could not be evaluated.",
|
||||
}
|
||||
if decision.get("approved"):
|
||||
return None
|
||||
payload = dict(decision)
|
||||
payload["error"] = payload.pop("message", None) or (
|
||||
f"BLOCKED: Tool '{tool_name}' was not approved."
|
||||
)
|
||||
payload.setdefault("policy", "ask")
|
||||
payload.setdefault("tool", tool_name)
|
||||
return json.dumps(payload, ensure_ascii=False)
|
||||
|
||||
|
||||
def run_api_execution_middleware(
|
||||
|
|
|
|||
346
tests/test_tool_approval_policies.py
Normal file
346
tests/test_tool_approval_policies.py
Normal file
|
|
@ -0,0 +1,346 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from types import SimpleNamespace
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _no_plugin_execution_middleware(monkeypatch):
|
||||
monkeypatch.setattr(
|
||||
"hermes_cli.middleware._get_middleware_callbacks", lambda _kind: []
|
||||
)
|
||||
|
||||
|
||||
def test_policy_resolves_tool_name_before_toolset_and_uses_most_restrictive(monkeypatch):
|
||||
from hermes_cli import middleware
|
||||
|
||||
monkeypatch.setattr(
|
||||
"hermes_cli.config.load_config",
|
||||
lambda: {
|
||||
"approvals": {
|
||||
"tool_policies": {
|
||||
"file*": "allow",
|
||||
"write_*": "ask",
|
||||
"write_file": "deny",
|
||||
}
|
||||
}
|
||||
},
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
"tools.registry.registry.get_entry",
|
||||
lambda _name: SimpleNamespace(toolset="files"),
|
||||
)
|
||||
|
||||
assert middleware.resolve_tool_approval_policy("write_file") == "deny"
|
||||
|
||||
|
||||
def test_policy_can_match_registered_toolset(monkeypatch):
|
||||
from hermes_cli import middleware
|
||||
|
||||
monkeypatch.setattr(
|
||||
"hermes_cli.config.load_config",
|
||||
lambda: {"approvals": {"tool_policies": {"browser*": "ask"}}},
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
"tools.registry.registry.get_entry",
|
||||
lambda _name: SimpleNamespace(toolset="browser"),
|
||||
)
|
||||
|
||||
assert middleware.resolve_tool_approval_policy("browser_click") == "ask"
|
||||
|
||||
|
||||
def test_malformed_policy_entries_are_ignored(monkeypatch):
|
||||
from hermes_cli import middleware
|
||||
|
||||
monkeypatch.setattr(
|
||||
"hermes_cli.config.load_config",
|
||||
lambda: {
|
||||
"approvals": {
|
||||
"tool_policies": {
|
||||
"terminal": "maybe",
|
||||
42: "deny",
|
||||
"*": None,
|
||||
}
|
||||
}
|
||||
},
|
||||
)
|
||||
|
||||
assert middleware.resolve_tool_approval_policy("terminal") is None
|
||||
|
||||
|
||||
def test_deny_policy_blocks_without_calling_tool(monkeypatch):
|
||||
from hermes_cli import middleware
|
||||
|
||||
monkeypatch.setattr(
|
||||
middleware, "resolve_tool_approval_policy", lambda _name: "deny"
|
||||
)
|
||||
called = []
|
||||
|
||||
result = middleware.run_tool_execution_middleware(
|
||||
"write_file", {"path": "notes.txt"}, lambda args: called.append(args)
|
||||
)
|
||||
|
||||
assert called == []
|
||||
assert json.loads(result)["error"].startswith("BLOCKED: Tool 'write_file'")
|
||||
|
||||
|
||||
def test_ask_policy_uses_shared_fail_closed_approval_gate(monkeypatch):
|
||||
from hermes_cli import middleware
|
||||
|
||||
monkeypatch.setattr(
|
||||
middleware, "resolve_tool_approval_policy", lambda _name: "ask"
|
||||
)
|
||||
requested = []
|
||||
monkeypatch.setattr(
|
||||
"tools.approval.request_tool_approval",
|
||||
lambda tool_name, reason, **kwargs: requested.append(
|
||||
(tool_name, reason, kwargs)
|
||||
)
|
||||
or {"approved": False, "message": "BLOCKED by cron policy"},
|
||||
)
|
||||
called = []
|
||||
|
||||
result = middleware.run_tool_execution_middleware(
|
||||
"terminal", {"command": "printf ok"}, lambda args: called.append(args)
|
||||
)
|
||||
|
||||
assert json.loads(result)["error"] == "BLOCKED by cron policy"
|
||||
assert called == []
|
||||
assert requested[0][0] == "terminal"
|
||||
assert requested[0][2]["rule_key"] == "tool_policy:terminal"
|
||||
|
||||
|
||||
def test_allow_policy_does_not_skip_downstream_execution_middleware(monkeypatch):
|
||||
from hermes_cli import middleware
|
||||
|
||||
monkeypatch.setattr(
|
||||
middleware, "resolve_tool_approval_policy", lambda _name: "allow"
|
||||
)
|
||||
events = []
|
||||
|
||||
def plugin_middleware(**kwargs):
|
||||
events.append("plugin")
|
||||
return kwargs["next_call"](kwargs["args"])
|
||||
|
||||
monkeypatch.setattr(
|
||||
middleware, "_get_middleware_callbacks", lambda _kind: [plugin_middleware]
|
||||
)
|
||||
|
||||
result = middleware.run_tool_execution_middleware(
|
||||
"read_file", {"path": "README.md"}, lambda _args: events.append("tool") or "ok"
|
||||
)
|
||||
|
||||
assert result == "ok"
|
||||
assert events == ["plugin", "tool"]
|
||||
|
||||
|
||||
def test_direct_registry_dispatch_is_policy_gated(monkeypatch):
|
||||
import model_tools
|
||||
from hermes_cli import middleware
|
||||
|
||||
monkeypatch.setattr(
|
||||
middleware, "resolve_tool_approval_policy", lambda _name: "deny"
|
||||
)
|
||||
dispatched = []
|
||||
monkeypatch.setattr(
|
||||
model_tools.registry,
|
||||
"dispatch",
|
||||
lambda name, args, **kwargs: dispatched.append((name, args)) or "ran",
|
||||
)
|
||||
|
||||
result = model_tools.handle_function_call("read_file", {"path": "README.md"})
|
||||
|
||||
assert dispatched == []
|
||||
assert "Tool 'read_file' is denied" in json.loads(result)["error"]
|
||||
|
||||
|
||||
def test_sequential_agent_level_dispatch_is_policy_gated(monkeypatch):
|
||||
from agent import tool_executor
|
||||
from hermes_cli import middleware
|
||||
|
||||
monkeypatch.setattr(
|
||||
middleware, "resolve_tool_approval_policy", lambda _name: "deny"
|
||||
)
|
||||
called = []
|
||||
agent = SimpleNamespace(
|
||||
session_id="s",
|
||||
_current_turn_id="t",
|
||||
_current_api_request_id="r",
|
||||
)
|
||||
|
||||
result, _ = tool_executor._run_agent_tool_execution_middleware(
|
||||
agent,
|
||||
function_name="todo",
|
||||
function_args={"todos": []},
|
||||
effective_task_id="task",
|
||||
tool_call_id="call",
|
||||
execute=lambda args: called.append(args) or "ran",
|
||||
)
|
||||
|
||||
assert called == []
|
||||
assert "Tool 'todo' is denied" in json.loads(result)["error"]
|
||||
|
||||
|
||||
def test_concurrent_agent_level_dispatch_is_policy_gated(monkeypatch):
|
||||
from agent import agent_runtime_helpers
|
||||
from hermes_cli import middleware
|
||||
|
||||
monkeypatch.setattr(
|
||||
middleware, "resolve_tool_approval_policy", lambda _name: "deny"
|
||||
)
|
||||
called = []
|
||||
agent = SimpleNamespace(
|
||||
session_id="s",
|
||||
_current_turn_id="t",
|
||||
_current_api_request_id="r",
|
||||
_todo_store=object(),
|
||||
_memory_manager=None,
|
||||
valid_tool_names=set(),
|
||||
enabled_toolsets=None,
|
||||
disabled_toolsets=None,
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
"tools.todo_tool.todo_tool", lambda **kwargs: called.append(kwargs) or "ran"
|
||||
)
|
||||
|
||||
result = agent_runtime_helpers.invoke_tool(
|
||||
agent, "todo", {"todos": []}, "task", tool_call_id="call",
|
||||
pre_tool_block_checked=True,
|
||||
)
|
||||
|
||||
assert called == []
|
||||
assert "Tool 'todo' is denied" in json.loads(result)["error"]
|
||||
|
||||
|
||||
def test_explicit_allow_does_not_bypass_plugin_escalation(monkeypatch):
|
||||
import model_tools
|
||||
from hermes_cli import middleware
|
||||
|
||||
monkeypatch.setattr(
|
||||
middleware, "resolve_tool_approval_policy", lambda _name: "allow"
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
"hermes_cli.plugins.resolve_pre_tool_block",
|
||||
lambda *_args, **_kwargs: "plugin requires approval",
|
||||
)
|
||||
dispatched = []
|
||||
monkeypatch.setattr(
|
||||
model_tools.registry,
|
||||
"dispatch",
|
||||
lambda name, args, **kwargs: dispatched.append((name, args)) or "ran",
|
||||
)
|
||||
|
||||
result = model_tools.handle_function_call("read_file", {"path": "README.md"})
|
||||
|
||||
assert dispatched == []
|
||||
assert json.loads(result)["error"] == "plugin requires approval"
|
||||
|
||||
|
||||
def test_explicit_allow_does_not_bypass_terminal_hardline(monkeypatch):
|
||||
from hermes_cli import middleware
|
||||
from tools.approval import check_all_command_guards
|
||||
|
||||
monkeypatch.setattr(
|
||||
middleware, "resolve_tool_approval_policy", lambda _name: "allow"
|
||||
)
|
||||
called = []
|
||||
|
||||
result = middleware.run_tool_execution_middleware(
|
||||
"terminal",
|
||||
{"command": "rm -rf /"},
|
||||
lambda args: called.append(args)
|
||||
or check_all_command_guards(args["command"], "local"),
|
||||
)
|
||||
|
||||
assert called == [{"command": "rm -rf /"}]
|
||||
assert result["approved"] is False
|
||||
assert "hardline" in result["message"].lower()
|
||||
|
||||
|
||||
def test_explicit_allow_bypasses_only_ordinary_terminal_prompt(monkeypatch):
|
||||
from hermes_cli import middleware
|
||||
from tools import approval
|
||||
|
||||
monkeypatch.setattr(
|
||||
middleware, "resolve_tool_approval_policy", lambda _name: "allow"
|
||||
)
|
||||
prompted = []
|
||||
|
||||
result = middleware.run_tool_execution_middleware(
|
||||
"terminal",
|
||||
{"command": "rm -rf /tmp/hermes-policy-test"},
|
||||
lambda args: approval.check_all_command_guards(
|
||||
args["command"],
|
||||
"local",
|
||||
approval_callback=lambda *_args, **_kwargs: prompted.append(True) or "deny",
|
||||
),
|
||||
)
|
||||
|
||||
assert result["approved"] is True
|
||||
assert prompted == []
|
||||
|
||||
|
||||
def test_explicit_allow_does_not_bypass_terminal_user_deny(monkeypatch):
|
||||
from hermes_cli import middleware
|
||||
from tools import approval
|
||||
|
||||
monkeypatch.setattr(
|
||||
middleware, "resolve_tool_approval_policy", lambda _name: "allow"
|
||||
)
|
||||
monkeypatch.setattr(approval, "_match_user_deny_rule", lambda _command: "git push *")
|
||||
|
||||
result = middleware.run_tool_execution_middleware(
|
||||
"terminal",
|
||||
{"command": "git push origin main"},
|
||||
lambda args: approval.check_all_command_guards(args["command"], "local"),
|
||||
)
|
||||
|
||||
assert result["approved"] is False
|
||||
assert "user-defined deny rule" in result["message"].lower()
|
||||
|
||||
|
||||
def test_explicit_allow_does_not_bypass_credential_path_guard(monkeypatch, tmp_path):
|
||||
from hermes_cli import middleware
|
||||
from tools.file_tools import write_file_tool
|
||||
|
||||
monkeypatch.setattr(
|
||||
middleware, "resolve_tool_approval_policy", lambda _name: "allow"
|
||||
)
|
||||
profile = tmp_path / ".hermes"
|
||||
profile.mkdir()
|
||||
monkeypatch.setenv("HERMES_HOME", str(profile))
|
||||
|
||||
result = middleware.run_tool_execution_middleware(
|
||||
"write_file",
|
||||
{"path": str(profile / ".env"), "content": "SECRET=value"},
|
||||
lambda args: write_file_tool(**args),
|
||||
)
|
||||
|
||||
assert not (profile / ".env").exists()
|
||||
assert "write denied" in json.loads(result)["error"].lower()
|
||||
|
||||
|
||||
def test_ask_policy_honors_cron_deny(monkeypatch):
|
||||
from hermes_cli import middleware
|
||||
|
||||
monkeypatch.setattr(
|
||||
middleware, "resolve_tool_approval_policy", lambda _name: "ask"
|
||||
)
|
||||
monkeypatch.setenv("HERMES_CRON_SESSION", "1")
|
||||
monkeypatch.setattr("tools.approval._get_cron_approval_mode", lambda: "deny")
|
||||
called = []
|
||||
|
||||
result = middleware.run_tool_execution_middleware(
|
||||
"read_file", {"path": "README.md"}, lambda args: called.append(args)
|
||||
)
|
||||
|
||||
assert called == []
|
||||
assert "cron jobs run without a user present" in json.loads(result)["error"]
|
||||
|
||||
|
||||
def test_default_config_exposes_empty_profile_local_policy_mapping():
|
||||
from hermes_cli.config import DEFAULT_CONFIG
|
||||
|
||||
assert DEFAULT_CONFIG["approvals"]["tool_policies"] == {}
|
||||
|
|
@ -49,6 +49,19 @@ _approval_tool_call_id: contextvars.ContextVar[str] = contextvars.ContextVar(
|
|||
"approval_tool_call_id",
|
||||
default="",
|
||||
)
|
||||
_tool_policy_terminal_allow: contextvars.ContextVar[bool] = contextvars.ContextVar(
|
||||
"tool_policy_terminal_allow",
|
||||
default=False,
|
||||
)
|
||||
|
||||
|
||||
def set_tool_policy_terminal_allow() -> contextvars.Token[bool]:
|
||||
"""Allow ordinary terminal prompts for the current tool execution only."""
|
||||
return _tool_policy_terminal_allow.set(True)
|
||||
|
||||
|
||||
def reset_tool_policy_terminal_allow(token: contextvars.Token[bool]) -> None:
|
||||
_tool_policy_terminal_allow.reset(token)
|
||||
|
||||
# Interactive-CLI flag. Concurrent ACP sessions run on a shared
|
||||
# ThreadPoolExecutor (acp_adapter/server.py), so mutating the process-global
|
||||
|
|
@ -2273,6 +2286,12 @@ def check_dangerous_command(command: str, env_type: str,
|
|||
if _YOLO_MODE_FROZEN or is_current_session_yolo_enabled():
|
||||
return {"approved": True, "message": None}
|
||||
|
||||
# A declarative terminal allow skips only the ordinary dangerous-command
|
||||
# approval layer. Hardline and user-deny checks above still win, and cron
|
||||
# never honors this shortcut because its unattended policy outranks allow.
|
||||
if _tool_policy_terminal_allow.get() and not env_var_enabled("HERMES_CRON_SESSION"):
|
||||
return {"approved": True, "message": None}
|
||||
|
||||
if _command_matches_permanent_allowlist(command):
|
||||
return {"approved": True, "message": None}
|
||||
|
||||
|
|
@ -2588,6 +2607,12 @@ def check_all_command_guards(command: str, env_type: str,
|
|||
if _YOLO_MODE_FROZEN or is_current_session_yolo_enabled() or approval_mode == "off":
|
||||
return {"approved": True, "message": None}
|
||||
|
||||
# Per-tool allow is intentionally narrower than yolo/mode=off: it skips
|
||||
# routine terminal prompts only after hardline/user-deny checks, and never
|
||||
# overrides cron's unattended policy.
|
||||
if _tool_policy_terminal_allow.get() and not env_var_enabled("HERMES_CRON_SESSION"):
|
||||
return {"approved": True, "message": None}
|
||||
|
||||
if _command_matches_permanent_allowlist(command):
|
||||
return {"approved": True, "message": None}
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue