mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-31 19:16:29 +00:00
feat(approvals): add profile-local per-tool policies
This commit is contained in:
Teknium
parent
e589b739ca
commit
5bff2c28ee
13 changed files with 1307 additions and 39 deletions
|
|
@ -1906,6 +1906,8 @@ def init_agent(
|
|||
# would still leak lcm_* tools into the tool surface and incur the
|
||||
# same local-model latency penalty.
|
||||
agent._context_engine_tool_names: set = set()
|
||||
if not hasattr(agent, "_dynamic_tool_entries"):
|
||||
agent._dynamic_tool_entries = {}
|
||||
if (
|
||||
hasattr(agent, "context_compressor")
|
||||
and agent.context_compressor
|
||||
|
|
@ -1940,6 +1942,10 @@ def init_agent(
|
|||
agent.tools.append(_wrapped)
|
||||
agent.valid_tool_names.add(_tname)
|
||||
agent._context_engine_tool_names.add(_tname)
|
||||
from tools.registry import DynamicToolEntry as _DynamicToolEntry
|
||||
agent._dynamic_tool_entries[_tname] = _DynamicToolEntry(
|
||||
_tname, "context_engine"
|
||||
)
|
||||
_existing_tool_names.add(_tname)
|
||||
|
||||
# Notify context engine of session start
|
||||
|
|
|
|||
|
|
@ -2304,6 +2304,15 @@ def invoke_tool(agent, function_name: str, function_args: dict, effective_task_i
|
|||
),
|
||||
)
|
||||
return _finish_agent_tool(result, next_args)
|
||||
elif (
|
||||
getattr(agent, "_context_engine_tool_names", None)
|
||||
and function_name in agent._context_engine_tool_names
|
||||
):
|
||||
def _execute(next_args: dict) -> Any:
|
||||
return _finish_agent_tool(
|
||||
agent.context_compressor.handle_tool_call(function_name, next_args),
|
||||
next_args,
|
||||
)
|
||||
elif agent._memory_manager and agent._memory_manager.has_tool(function_name):
|
||||
def _execute(next_args: dict) -> Any:
|
||||
return _finish_agent_tool(agent._memory_manager.handle_tool_call(function_name, next_args), next_args)
|
||||
|
|
@ -2333,6 +2342,8 @@ def invoke_tool(agent, function_name: str, function_args: dict, effective_task_i
|
|||
def _execute(next_args: dict) -> Any:
|
||||
return _finish_agent_tool(agent._dispatch_delegate_task(next_args), next_args)
|
||||
else:
|
||||
from model_tools import registry
|
||||
|
||||
def _execute(next_args: dict) -> Any:
|
||||
return _ra().handle_function_call(
|
||||
function_name, next_args, effective_task_id,
|
||||
|
|
@ -2346,9 +2357,15 @@ def invoke_tool(agent, function_name: str, function_args: dict, effective_task_i
|
|||
enabled_toolsets=getattr(agent, "enabled_toolsets", None),
|
||||
disabled_toolsets=getattr(agent, "disabled_toolsets", None),
|
||||
tool_request_middleware_trace=list(_tool_middleware_trace),
|
||||
dispatch_registry=registry,
|
||||
)
|
||||
|
||||
from hermes_cli.middleware import run_tool_execution_middleware
|
||||
from model_tools import registry
|
||||
|
||||
registry_entry = registry.get_entry(function_name)
|
||||
if registry_entry is None:
|
||||
registry_entry = getattr(agent, "_dynamic_tool_entries", {}).get(function_name)
|
||||
|
||||
return run_tool_execution_middleware(
|
||||
function_name,
|
||||
|
|
@ -2360,6 +2377,8 @@ def invoke_tool(agent, function_name: str, function_args: dict, effective_task_i
|
|||
tool_call_id=tool_call_id or "",
|
||||
turn_id=getattr(agent, "_current_turn_id", "") or "",
|
||||
api_request_id=getattr(agent, "_current_api_request_id", "") or "",
|
||||
registry_entry=registry_entry,
|
||||
dispatch_registry=registry if registry_entry is registry.get_entry(function_name) else None,
|
||||
)
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -35,7 +35,7 @@ from typing import Any, Callable, Dict, List, Optional
|
|||
|
||||
from agent.memory_provider import MemoryProvider
|
||||
from agent.skill_commands import extract_user_instruction_from_skill_message
|
||||
from tools.registry import tool_error
|
||||
from tools.registry import DynamicToolEntry, tool_error
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
|
@ -123,6 +123,10 @@ def inject_memory_provider_tools(agent: Any) -> int:
|
|||
if valid_tool_names is None:
|
||||
valid_tool_names = set()
|
||||
agent.valid_tool_names = valid_tool_names
|
||||
dynamic_entries = getattr(agent, "_dynamic_tool_entries", None)
|
||||
if dynamic_entries is None:
|
||||
dynamic_entries = {}
|
||||
agent._dynamic_tool_entries = dynamic_entries
|
||||
|
||||
added = 0
|
||||
for raw_schema in get_schemas():
|
||||
|
|
@ -139,6 +143,7 @@ def inject_memory_provider_tools(agent: Any) -> int:
|
|||
continue
|
||||
tools.append({"type": "function", "function": schema})
|
||||
valid_tool_names.add(tool_name)
|
||||
dynamic_entries[tool_name] = DynamicToolEntry(tool_name, "memory")
|
||||
existing_tool_names.add(tool_name)
|
||||
added += 1
|
||||
|
||||
|
|
|
|||
|
|
@ -307,6 +307,11 @@ def _run_agent_tool_execution_middleware(
|
|||
return execute(observed_args)
|
||||
|
||||
from hermes_cli.middleware import run_tool_execution_middleware
|
||||
from model_tools import registry
|
||||
|
||||
registry_entry = registry.get_entry(function_name)
|
||||
if registry_entry is None:
|
||||
registry_entry = getattr(agent, "_dynamic_tool_entries", {}).get(function_name)
|
||||
|
||||
result = run_tool_execution_middleware(
|
||||
function_name,
|
||||
|
|
@ -318,6 +323,7 @@ def _run_agent_tool_execution_middleware(
|
|||
tool_call_id=tool_call_id or "",
|
||||
turn_id=getattr(agent, "_current_turn_id", "") or "",
|
||||
api_request_id=getattr(agent, "_current_api_request_id", "") or "",
|
||||
registry_entry=registry_entry,
|
||||
)
|
||||
return result, observed_args
|
||||
|
||||
|
|
|
|||
|
|
@ -2534,6 +2534,14 @@ DEFAULT_CONFIG = {
|
|||
"mode": "smart",
|
||||
"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. Across all matching
|
||||
# tool-name and toolset rules, the safest 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,13 +7,50 @@ 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
|
||||
import threading
|
||||
from copy import deepcopy
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Any, Callable, Dict, List, Optional
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
_TOOL_POLICY_PRIORITY = {"allow": 0, "ask": 1, "deny": 2}
|
||||
|
||||
|
||||
class _RegistryDispatchCapability:
|
||||
"""Opaque, single-use permission for one wrapper→registry handoff.
|
||||
|
||||
Context copies share this object's consumed state, so only one dispatch can
|
||||
redeem it even if a caller copies the Context or reuses the same args.
|
||||
"""
|
||||
|
||||
__slots__ = ("_registry", "_tool_name", "_args", "_consumed", "_lock")
|
||||
|
||||
def __init__(self, registry: Any, tool_name: str, args: Dict[str, Any]) -> None:
|
||||
self._registry = registry
|
||||
self._tool_name = tool_name
|
||||
self._args = args
|
||||
self._consumed = False
|
||||
self._lock = threading.Lock()
|
||||
|
||||
def consume(self, registry: Any, tool_name: str, args: Dict[str, Any]) -> bool:
|
||||
if registry is not self._registry or tool_name != self._tool_name or args is not self._args:
|
||||
return False
|
||||
with self._lock:
|
||||
if self._consumed:
|
||||
return False
|
||||
self._consumed = True
|
||||
return True
|
||||
|
||||
|
||||
_pending_registry_dispatch: contextvars.ContextVar[Optional[_RegistryDispatchCapability]] = (
|
||||
contextvars.ContextVar("pending_registry_dispatch", default=None)
|
||||
)
|
||||
|
||||
OBSERVER_SCHEMA_VERSION = "hermes.observer.v1"
|
||||
MIDDLEWARE_SCHEMA_VERSION = "hermes.middleware.v1"
|
||||
|
||||
|
|
@ -195,19 +232,153 @@ 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.
|
||||
|
||||
``dispatch_registry`` marks a wrapper whose immediate ``next_call`` enters
|
||||
that registry. Only that exact wrapper→registry handoff is deduplicated;
|
||||
nested dispatches from inside the handler remain genuine executions.
|
||||
"""
|
||||
dispatch_registry = context.pop("dispatch_registry", None)
|
||||
registry_entry = context.pop("registry_entry", None)
|
||||
registry_dispatch = context.pop("registry_dispatch", None)
|
||||
expected_dispatch = _pending_registry_dispatch.get()
|
||||
is_wrapped_registry_dispatch = bool(
|
||||
registry_dispatch
|
||||
and expected_dispatch is not None
|
||||
and expected_dispatch.consume(registry_dispatch, tool_name, args)
|
||||
)
|
||||
if is_wrapped_registry_dispatch:
|
||||
_pending_registry_dispatch.set(None)
|
||||
return next_call(args)
|
||||
|
||||
policy = (
|
||||
resolve_tool_approval_policy(tool_name, entry=registry_entry)
|
||||
if registry_entry is not None
|
||||
else resolve_tool_approval_policy(tool_name)
|
||||
)
|
||||
blocked = _apply_tool_approval_policy(tool_name, policy)
|
||||
if blocked is not None:
|
||||
return blocked
|
||||
|
||||
terminal_allow_token = None
|
||||
if tool_name == "terminal" and policy == "allow":
|
||||
from tools.approval import set_tool_policy_terminal_allow
|
||||
|
||||
terminal_allow_token = set_tool_policy_terminal_allow()
|
||||
|
||||
def call_next(next_args: Dict[str, Any]) -> Any:
|
||||
if dispatch_registry is None:
|
||||
return next_call(next_args)
|
||||
token = _pending_registry_dispatch.set(
|
||||
_RegistryDispatchCapability(dispatch_registry, tool_name, next_args)
|
||||
)
|
||||
try:
|
||||
return next_call(next_args)
|
||||
finally:
|
||||
_pending_registry_dispatch.reset(token)
|
||||
|
||||
try:
|
||||
callbacks = _get_middleware_callbacks(TOOL_EXECUTION_MIDDLEWARE)
|
||||
if not callbacks:
|
||||
return call_next(args)
|
||||
return _run_execution_chain(
|
||||
TOOL_EXECUTION_MIDDLEWARE,
|
||||
callbacks,
|
||||
call_next,
|
||||
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)
|
||||
|
||||
|
||||
def resolve_tool_approval_policy(
|
||||
tool_name: str, *, entry: Any = None
|
||||
) -> Optional[str]:
|
||||
"""Resolve glob policies for a tool name and its actual dispatch entry."""
|
||||
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()]
|
||||
toolset = str(getattr(entry, "toolset", "") or "").strip().lower()
|
||||
if toolset:
|
||||
identifiers.append(toolset)
|
||||
|
||||
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}",
|
||||
enforce_under_yolo=True,
|
||||
)
|
||||
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(
|
||||
|
|
|
|||
|
|
@ -1037,6 +1037,7 @@ def handle_function_call(
|
|||
tool_request_middleware_trace: Optional[List[Dict[str, Any]]] = None,
|
||||
enabled_toolsets: Optional[List[str]] = None,
|
||||
disabled_toolsets: Optional[List[str]] = None,
|
||||
dispatch_registry: Any = None,
|
||||
) -> str:
|
||||
"""
|
||||
Main function call dispatcher that routes calls to the tool registry.
|
||||
|
|
@ -1276,19 +1277,26 @@ def handle_function_call(
|
|||
session_id=session_id,
|
||||
user_task=user_task,
|
||||
)
|
||||
from hermes_cli.middleware import run_tool_execution_middleware
|
||||
if dispatch_registry is not None:
|
||||
# An outer runtime wrapper already ran policy/middleware and
|
||||
# armed the exact wrapper→registry handoff capability.
|
||||
result = _dispatch(function_args)
|
||||
else:
|
||||
from hermes_cli.middleware import run_tool_execution_middleware
|
||||
|
||||
result = run_tool_execution_middleware(
|
||||
function_name,
|
||||
function_args,
|
||||
_dispatch,
|
||||
original_args=_tool_original_args,
|
||||
task_id=task_id or "",
|
||||
session_id=session_id or "",
|
||||
tool_call_id=tool_call_id or "",
|
||||
turn_id=turn_id or "",
|
||||
api_request_id=api_request_id or "",
|
||||
)
|
||||
result = run_tool_execution_middleware(
|
||||
function_name,
|
||||
function_args,
|
||||
_dispatch,
|
||||
original_args=_tool_original_args,
|
||||
task_id=task_id or "",
|
||||
session_id=session_id or "",
|
||||
tool_call_id=tool_call_id or "",
|
||||
turn_id=turn_id or "",
|
||||
api_request_id=api_request_id or "",
|
||||
registry_entry=registry.get_entry(function_name),
|
||||
dispatch_registry=registry,
|
||||
)
|
||||
finally:
|
||||
if _approval_tokens is not None and reset_current_observability_context is not None:
|
||||
try:
|
||||
|
|
|
|||
|
|
@ -2854,6 +2854,8 @@ class TestConcurrentToolExecution:
|
|||
|
||||
def test_invoke_tool_dispatches_to_handle_function_call(self, agent):
|
||||
"""_invoke_tool should route regular tools through handle_function_call."""
|
||||
import model_tools
|
||||
|
||||
with patch("run_agent.handle_function_call", return_value="result") as mock_hfc:
|
||||
result = agent._invoke_tool("web_search", {"q": "test"}, "task-1")
|
||||
mock_hfc.assert_called_once_with(
|
||||
|
|
@ -2868,6 +2870,7 @@ class TestConcurrentToolExecution:
|
|||
enabled_toolsets=agent.enabled_toolsets,
|
||||
disabled_toolsets=agent.disabled_toolsets,
|
||||
tool_request_middleware_trace=[],
|
||||
dispatch_registry=model_tools.registry,
|
||||
)
|
||||
assert result == "result"
|
||||
|
||||
|
|
|
|||
942
tests/test_tool_approval_policies.py
Normal file
942
tests/test_tool_approval_policies.py
Normal file
|
|
@ -0,0 +1,942 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import contextvars
|
||||
import json
|
||||
from concurrent.futures import ThreadPoolExecutor
|
||||
from threading import Barrier, Lock
|
||||
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",
|
||||
}
|
||||
}
|
||||
},
|
||||
)
|
||||
entry = SimpleNamespace(toolset="files")
|
||||
|
||||
assert middleware.resolve_tool_approval_policy("write_file", entry=entry) == "deny"
|
||||
|
||||
|
||||
def test_broad_deny_outweighs_exact_allow(monkeypatch):
|
||||
from hermes_cli import middleware
|
||||
|
||||
monkeypatch.setattr(
|
||||
"hermes_cli.config.load_config",
|
||||
lambda: {
|
||||
"approvals": {
|
||||
"tool_policies": {
|
||||
"*": "deny",
|
||||
"read_file": "allow",
|
||||
}
|
||||
}
|
||||
},
|
||||
)
|
||||
|
||||
assert middleware.resolve_tool_approval_policy("read_file") == "deny"
|
||||
|
||||
|
||||
def test_toolset_deny_outweighs_tool_name_allow(monkeypatch):
|
||||
from hermes_cli import middleware
|
||||
|
||||
monkeypatch.setattr(
|
||||
"hermes_cli.config.load_config",
|
||||
lambda: {
|
||||
"approvals": {
|
||||
"tool_policies": {
|
||||
"read_file": "allow",
|
||||
"file": "deny",
|
||||
}
|
||||
}
|
||||
},
|
||||
)
|
||||
entry = SimpleNamespace(toolset="file")
|
||||
|
||||
assert middleware.resolve_tool_approval_policy("read_file", entry=entry) == "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"}}},
|
||||
)
|
||||
entry = SimpleNamespace(toolset="browser")
|
||||
|
||||
assert middleware.resolve_tool_approval_policy("browser_click", entry=entry) == "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, **_kwargs: "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, **_kwargs: "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, **_kwargs: "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, **_kwargs: "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_registry_dispatch_itself_is_policy_gated(monkeypatch):
|
||||
from hermes_cli import middleware
|
||||
from tools.registry import ToolRegistry
|
||||
|
||||
monkeypatch.setattr(
|
||||
middleware, "resolve_tool_approval_policy", lambda _name, **_kwargs: "deny"
|
||||
)
|
||||
called = []
|
||||
registry = ToolRegistry()
|
||||
registry.register(
|
||||
name="plugin_tool",
|
||||
toolset="plugin",
|
||||
schema={},
|
||||
handler=lambda args, **kwargs: called.append(args) or '{"ran": true}',
|
||||
)
|
||||
|
||||
result = registry.dispatch("plugin_tool", {"value": 1})
|
||||
|
||||
assert called == []
|
||||
assert "Tool 'plugin_tool' is denied" in json.loads(result)["error"]
|
||||
|
||||
|
||||
def test_custom_registry_dispatch_resolves_policy_from_its_own_toolset(monkeypatch):
|
||||
from tools.registry import ToolRegistry
|
||||
|
||||
monkeypatch.setattr(
|
||||
"hermes_cli.config.load_config",
|
||||
lambda: {"approvals": {"tool_policies": {"custom-toolset": "deny"}}},
|
||||
)
|
||||
called = []
|
||||
custom_registry = ToolRegistry()
|
||||
custom_registry.register(
|
||||
name="shared_name",
|
||||
toolset="custom-toolset",
|
||||
schema={},
|
||||
handler=lambda args, **kwargs: called.append(args) or "ran",
|
||||
)
|
||||
|
||||
result = custom_registry.dispatch("shared_name", {})
|
||||
|
||||
assert called == []
|
||||
assert json.loads(result)["policy"] == "deny"
|
||||
|
||||
|
||||
def test_genuine_recursive_same_name_dispatch_rechecks_policy_and_middleware(monkeypatch):
|
||||
from hermes_cli import middleware
|
||||
from tools.registry import ToolRegistry
|
||||
|
||||
monkeypatch.setattr(
|
||||
middleware, "resolve_tool_approval_policy", lambda _name, **_kwargs: "ask"
|
||||
)
|
||||
approvals = []
|
||||
monkeypatch.setattr(
|
||||
"tools.approval.request_tool_approval",
|
||||
lambda *_args, **_kwargs: approvals.append(True)
|
||||
or {"approved": True, "message": None},
|
||||
)
|
||||
executions = []
|
||||
|
||||
def execution_middleware(**kwargs):
|
||||
executions.append(kwargs["args"]["depth"])
|
||||
return kwargs["next_call"](kwargs["args"])
|
||||
|
||||
monkeypatch.setattr(
|
||||
middleware, "_get_middleware_callbacks", lambda _kind: [execution_middleware]
|
||||
)
|
||||
custom_registry = ToolRegistry()
|
||||
|
||||
def recursive_handler(args, **_kwargs):
|
||||
if args["depth"]:
|
||||
return custom_registry.dispatch("recursive", {"depth": args["depth"] - 1})
|
||||
return '{"done": true}'
|
||||
|
||||
custom_registry.register("recursive", "custom", {}, recursive_handler)
|
||||
|
||||
assert json.loads(custom_registry.dispatch("recursive", {"depth": 1})) == {"done": True}
|
||||
assert approvals == [True, True]
|
||||
assert executions == [1, 0]
|
||||
|
||||
|
||||
def test_wrapper_to_registry_dispatch_is_deduped_exactly_once(monkeypatch):
|
||||
from hermes_cli import middleware
|
||||
from tools.registry import ToolRegistry
|
||||
|
||||
monkeypatch.setattr(
|
||||
middleware, "resolve_tool_approval_policy", lambda _name, **_kwargs: "ask"
|
||||
)
|
||||
approvals = []
|
||||
monkeypatch.setattr(
|
||||
"tools.approval.request_tool_approval",
|
||||
lambda *_args, **_kwargs: approvals.append(True)
|
||||
or {"approved": True, "message": None},
|
||||
)
|
||||
executions = []
|
||||
|
||||
def execution_middleware(**kwargs):
|
||||
executions.append(kwargs["tool_name"])
|
||||
return kwargs["next_call"](kwargs["args"])
|
||||
|
||||
monkeypatch.setattr(
|
||||
middleware, "_get_middleware_callbacks", lambda _kind: [execution_middleware]
|
||||
)
|
||||
custom_registry = ToolRegistry()
|
||||
custom_registry.register("wrapped", "custom", {}, lambda _args, **_kwargs: "ok")
|
||||
args = {"value": 1}
|
||||
|
||||
result = middleware.run_tool_execution_middleware(
|
||||
"wrapped",
|
||||
args,
|
||||
lambda next_args: custom_registry.dispatch("wrapped", next_args),
|
||||
dispatch_registry=custom_registry,
|
||||
)
|
||||
|
||||
assert result == "ok"
|
||||
assert approvals == [True]
|
||||
assert executions == ["wrapped"]
|
||||
|
||||
|
||||
def test_copied_wrapper_context_cannot_reuse_consumed_registry_handoff(monkeypatch):
|
||||
from hermes_cli import middleware
|
||||
from tools.registry import ToolRegistry
|
||||
|
||||
monkeypatch.setattr(middleware, "resolve_tool_approval_policy", lambda _name, **_kwargs: "ask")
|
||||
approvals = []
|
||||
monkeypatch.setattr(
|
||||
"tools.approval.request_tool_approval",
|
||||
lambda *_args, **_kwargs: approvals.append(True) or {"approved": True, "message": None},
|
||||
)
|
||||
custom_registry = ToolRegistry()
|
||||
custom_registry.register("wrapped", "custom", {}, lambda _args, **_kwargs: "ok")
|
||||
args = {"value": 1}
|
||||
copied = None
|
||||
|
||||
def dispatch_and_copy(next_args):
|
||||
nonlocal copied
|
||||
copied = contextvars.copy_context()
|
||||
return custom_registry.dispatch("wrapped", next_args)
|
||||
|
||||
assert middleware.run_tool_execution_middleware(
|
||||
"wrapped", args, dispatch_and_copy, dispatch_registry=custom_registry
|
||||
) == "ok"
|
||||
assert copied is not None
|
||||
assert copied.run(custom_registry.dispatch, "wrapped", args) == "ok"
|
||||
assert approvals == [True, True]
|
||||
|
||||
|
||||
def test_one_wrapper_handoff_cannot_suppress_two_same_argument_dispatches(monkeypatch):
|
||||
from hermes_cli import middleware
|
||||
from tools.registry import ToolRegistry
|
||||
|
||||
monkeypatch.setattr(middleware, "resolve_tool_approval_policy", lambda _name, **_kwargs: "ask")
|
||||
approvals = []
|
||||
monkeypatch.setattr(
|
||||
"tools.approval.request_tool_approval",
|
||||
lambda *_args, **_kwargs: approvals.append(True) or {"approved": True, "message": None},
|
||||
)
|
||||
custom_registry = ToolRegistry()
|
||||
custom_registry.register("wrapped", "custom", {}, lambda _args, **_kwargs: "ok")
|
||||
args = {"value": 1}
|
||||
|
||||
def dispatch_twice(next_args):
|
||||
custom_registry.dispatch("wrapped", next_args)
|
||||
return custom_registry.dispatch("wrapped", next_args)
|
||||
|
||||
assert middleware.run_tool_execution_middleware(
|
||||
"wrapped", args, dispatch_twice, dispatch_registry=custom_registry
|
||||
) == "ok"
|
||||
assert approvals == [True, True]
|
||||
|
||||
|
||||
def test_wrapper_dedupe_does_not_cover_a_different_tool_name(monkeypatch):
|
||||
from hermes_cli import middleware
|
||||
from tools.registry import ToolRegistry
|
||||
|
||||
monkeypatch.setattr(
|
||||
middleware, "resolve_tool_approval_policy", lambda _name, **_kwargs: "ask"
|
||||
)
|
||||
approvals = []
|
||||
monkeypatch.setattr(
|
||||
"tools.approval.request_tool_approval",
|
||||
lambda *_args, **_kwargs: approvals.append(True)
|
||||
or {"approved": True, "message": None},
|
||||
)
|
||||
custom_registry = ToolRegistry()
|
||||
custom_registry.register("outer", "custom", {}, lambda _args, **_kwargs: "unused")
|
||||
custom_registry.register("inner", "custom", {}, lambda _args, **_kwargs: "ok")
|
||||
args = {"value": 1}
|
||||
|
||||
result = middleware.run_tool_execution_middleware(
|
||||
"outer",
|
||||
args,
|
||||
lambda next_args: custom_registry.dispatch("inner", next_args),
|
||||
dispatch_registry=custom_registry,
|
||||
)
|
||||
|
||||
assert result == "ok"
|
||||
assert approvals == [True, True]
|
||||
|
||||
|
||||
def test_wrapper_registry_dedupe_tokens_are_isolated_across_concurrent_contexts(monkeypatch):
|
||||
from hermes_cli import middleware
|
||||
from tools.registry import ToolRegistry
|
||||
|
||||
monkeypatch.setattr(
|
||||
middleware, "resolve_tool_approval_policy", lambda _name, **_kwargs: "ask"
|
||||
)
|
||||
approvals = []
|
||||
lock = Lock()
|
||||
|
||||
def approve(*_args, **_kwargs):
|
||||
with lock:
|
||||
approvals.append(True)
|
||||
return {"approved": True, "message": None}
|
||||
|
||||
monkeypatch.setattr("tools.approval.request_tool_approval", approve)
|
||||
executions = []
|
||||
|
||||
def execution_middleware(**kwargs):
|
||||
with lock:
|
||||
executions.append(kwargs["args"]["call"])
|
||||
return kwargs["next_call"](kwargs["args"])
|
||||
|
||||
monkeypatch.setattr(
|
||||
middleware, "_get_middleware_callbacks", lambda _kind: [execution_middleware]
|
||||
)
|
||||
barrier = Barrier(2)
|
||||
custom_registry = ToolRegistry()
|
||||
|
||||
def handler(args, **_kwargs):
|
||||
barrier.wait(timeout=5)
|
||||
return str(args["call"])
|
||||
|
||||
custom_registry.register("concurrent", "custom", {}, handler)
|
||||
|
||||
def invoke(call):
|
||||
args = {"call": call}
|
||||
return middleware.run_tool_execution_middleware(
|
||||
"concurrent",
|
||||
args,
|
||||
lambda next_args: custom_registry.dispatch("concurrent", next_args),
|
||||
dispatch_registry=custom_registry,
|
||||
)
|
||||
|
||||
with ThreadPoolExecutor(max_workers=2) as pool:
|
||||
results = list(pool.map(invoke, (1, 2)))
|
||||
|
||||
assert results == ["1", "2"]
|
||||
assert len(approvals) == 2
|
||||
assert sorted(executions) == [1, 2]
|
||||
|
||||
|
||||
def test_plugin_context_dispatch_cannot_bypass_policy(monkeypatch):
|
||||
from hermes_cli import middleware
|
||||
from hermes_cli.plugins import PluginContext, PluginManager, PluginManifest
|
||||
from tools.registry import registry
|
||||
|
||||
monkeypatch.setattr(
|
||||
middleware, "resolve_tool_approval_policy", lambda _name, **_kwargs: "deny"
|
||||
)
|
||||
called = []
|
||||
registry.register(
|
||||
name="plugin_context_policy_test",
|
||||
toolset="plugin-test",
|
||||
schema={},
|
||||
handler=lambda args, **kwargs: called.append(args) or '{"ran": true}',
|
||||
)
|
||||
try:
|
||||
context = PluginContext(
|
||||
PluginManifest(name="policy-test", source="user"), PluginManager()
|
||||
)
|
||||
result = context.dispatch_tool("plugin_context_policy_test", {})
|
||||
finally:
|
||||
registry.deregister("plugin_context_policy_test")
|
||||
|
||||
assert called == []
|
||||
assert "is denied" in json.loads(result)["error"]
|
||||
|
||||
|
||||
def test_nested_model_and_registry_dispatch_ask_only_once(monkeypatch):
|
||||
import model_tools
|
||||
from hermes_cli import middleware
|
||||
|
||||
monkeypatch.setattr(
|
||||
middleware, "resolve_tool_approval_policy", lambda _name, **_kwargs: "ask"
|
||||
)
|
||||
approvals = []
|
||||
monkeypatch.setattr(
|
||||
"tools.approval.request_tool_approval",
|
||||
lambda *_args, **_kwargs: approvals.append(True)
|
||||
or {"approved": True, "message": None},
|
||||
)
|
||||
|
||||
result = model_tools.handle_function_call(
|
||||
"read_file", {"path": "README.md"}, skip_pre_tool_call_hook=True
|
||||
)
|
||||
|
||||
assert json.loads(result).get("error") is None
|
||||
assert approvals == [True]
|
||||
|
||||
|
||||
def test_nested_model_and_registry_dispatch_runs_execution_middleware_once(monkeypatch):
|
||||
import model_tools
|
||||
from hermes_cli import middleware
|
||||
|
||||
monkeypatch.setattr(
|
||||
middleware, "resolve_tool_approval_policy", lambda _name, **_kwargs: None
|
||||
)
|
||||
executions = []
|
||||
|
||||
def execution_middleware(**kwargs):
|
||||
executions.append(kwargs["tool_name"])
|
||||
return kwargs["next_call"](kwargs["args"])
|
||||
|
||||
monkeypatch.setattr(
|
||||
middleware,
|
||||
"_get_middleware_callbacks",
|
||||
lambda _kind: [execution_middleware],
|
||||
)
|
||||
|
||||
result = model_tools.handle_function_call(
|
||||
"read_file", {"path": "README.md"}, skip_pre_tool_call_hook=True
|
||||
)
|
||||
|
||||
assert json.loads(result).get("error") is None
|
||||
assert executions == ["read_file"]
|
||||
|
||||
|
||||
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, **_kwargs: "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, **_kwargs: "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_runtime_fallback_registry_runs_policy_and_execution_middleware_once(monkeypatch):
|
||||
"""The real invoke_tool fallback must hand off directly to the registry once."""
|
||||
import model_tools
|
||||
from agent import agent_runtime_helpers
|
||||
from hermes_cli import middleware
|
||||
|
||||
approvals = []
|
||||
executions = []
|
||||
monkeypatch.setattr(
|
||||
middleware, "resolve_tool_approval_policy", lambda _name, **_kwargs: "ask"
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
"tools.approval.request_tool_approval",
|
||||
lambda *_args, **_kwargs: approvals.append(True)
|
||||
or {"approved": True, "message": None},
|
||||
)
|
||||
|
||||
def execution_middleware(**kwargs):
|
||||
executions.append(kwargs["tool_name"])
|
||||
return kwargs["next_call"](kwargs["args"])
|
||||
|
||||
monkeypatch.setattr(
|
||||
middleware, "_get_middleware_callbacks", lambda _kind: [execution_middleware]
|
||||
)
|
||||
tool_name = "runtime_fallback_policy_probe"
|
||||
model_tools.registry.register(
|
||||
tool_name, "probe", {}, lambda _args, **_kwargs: '{"success": true}'
|
||||
)
|
||||
agent = SimpleNamespace(
|
||||
session_id="s",
|
||||
_current_turn_id="t",
|
||||
_current_api_request_id="r",
|
||||
_memory_manager=None,
|
||||
_context_engine_tool_names=set(),
|
||||
valid_tool_names={tool_name},
|
||||
enabled_toolsets=None,
|
||||
disabled_toolsets=None,
|
||||
)
|
||||
try:
|
||||
result = agent_runtime_helpers.invoke_tool(
|
||||
agent, tool_name, {}, "task", tool_call_id="call", pre_tool_block_checked=True
|
||||
)
|
||||
finally:
|
||||
model_tools.registry.deregister(tool_name)
|
||||
|
||||
assert json.loads(result) == {"success": True}
|
||||
assert approvals == [True]
|
||||
assert executions == [tool_name]
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("family", "tool_name", "toolset"),
|
||||
[
|
||||
("context", "context_policy_probe", "context_engine"),
|
||||
("memory", "memory_policy_probe", "memory"),
|
||||
],
|
||||
)
|
||||
def test_dynamic_tool_family_deny_overrides_exact_allow_on_runtime_path(
|
||||
monkeypatch, family, tool_name, toolset
|
||||
):
|
||||
"""Dynamic schemas carry logical toolset identity without global registration."""
|
||||
from agent import agent_runtime_helpers
|
||||
|
||||
monkeypatch.setattr(
|
||||
"hermes_cli.config.load_config",
|
||||
lambda: {
|
||||
"approvals": {
|
||||
"tool_policies": {tool_name: "allow", toolset: "deny"}
|
||||
}
|
||||
},
|
||||
)
|
||||
from tools.mcp_tool import _reinject_post_build_tools
|
||||
|
||||
calls = []
|
||||
schema = {"name": tool_name, "description": "probe", "parameters": {}}
|
||||
memory_manager = SimpleNamespace(
|
||||
has_tool=lambda name: family == "memory" and name == tool_name,
|
||||
handle_tool_call=lambda name, args: calls.append((name, args)) or "ran",
|
||||
get_all_tool_schemas=lambda: [schema] if family == "memory" else [],
|
||||
)
|
||||
context_engine = SimpleNamespace(
|
||||
handle_tool_call=lambda name, args, **kwargs: calls.append((name, args)) or "ran",
|
||||
get_tool_schemas=lambda: [schema] if family == "context" else [],
|
||||
)
|
||||
dynamic_entries = {}
|
||||
agent = SimpleNamespace(
|
||||
session_id="s",
|
||||
_current_turn_id="t",
|
||||
_current_api_request_id="r",
|
||||
_memory_manager=memory_manager,
|
||||
context_compressor=context_engine,
|
||||
enabled_toolsets=None,
|
||||
disabled_toolsets=None,
|
||||
)
|
||||
tools = []
|
||||
names = set()
|
||||
engine_names = _reinject_post_build_tools(
|
||||
agent, tools, names, dynamic_entries=dynamic_entries
|
||||
)
|
||||
agent._context_engine_tool_names = engine_names
|
||||
agent._dynamic_tool_entries = dynamic_entries
|
||||
agent.valid_tool_names = names
|
||||
|
||||
assert dynamic_entries[tool_name].toolset == toolset
|
||||
result = agent_runtime_helpers.invoke_tool(
|
||||
agent, tool_name, {}, "task", tool_call_id="call", pre_tool_block_checked=True
|
||||
)
|
||||
|
||||
assert calls == []
|
||||
assert json.loads(result)["policy"] == "deny"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"tool_name,toolset", [("read_terminal", "terminal"), ("delegate_task", "delegation")]
|
||||
)
|
||||
def test_sequential_fast_paths_apply_registered_toolset_policy(monkeypatch, tool_name, toolset):
|
||||
import model_tools
|
||||
from agent import tool_executor
|
||||
|
||||
entry = model_tools.registry.get_entry(tool_name)
|
||||
assert entry is not None and entry.toolset == toolset
|
||||
monkeypatch.setattr(
|
||||
"hermes_cli.config.load_config",
|
||||
lambda: {"approvals": {"tool_policies": {tool_name: "allow", toolset: "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=tool_name,
|
||||
function_args={},
|
||||
effective_task_id="task",
|
||||
tool_call_id="call",
|
||||
execute=lambda args: called.append(args) or "ran",
|
||||
)
|
||||
|
||||
assert called == []
|
||||
assert json.loads(result)["policy"] == "deny"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"tool_name,toolset", [("read_terminal", "terminal"), ("delegate_task", "delegation")]
|
||||
)
|
||||
def test_concurrent_fast_paths_apply_registered_toolset_policy(monkeypatch, tool_name, toolset):
|
||||
import model_tools
|
||||
from agent import agent_runtime_helpers
|
||||
|
||||
entry = model_tools.registry.get_entry(tool_name)
|
||||
assert entry is not None and entry.toolset == toolset
|
||||
monkeypatch.setattr(
|
||||
"hermes_cli.config.load_config",
|
||||
lambda: {"approvals": {"tool_policies": {tool_name: "allow", toolset: "deny"}}},
|
||||
)
|
||||
called = []
|
||||
agent = SimpleNamespace(
|
||||
session_id="s",
|
||||
_current_turn_id="t",
|
||||
_current_api_request_id="r",
|
||||
_memory_manager=None,
|
||||
valid_tool_names=set(),
|
||||
enabled_toolsets=None,
|
||||
disabled_toolsets=None,
|
||||
read_terminal_callback=lambda *_a: called.append("read"),
|
||||
_dispatch_delegate_task=lambda args: called.append(args) or "ran",
|
||||
)
|
||||
|
||||
result = agent_runtime_helpers.invoke_tool(
|
||||
agent, tool_name, {}, "task", tool_call_id="call", pre_tool_block_checked=True
|
||||
)
|
||||
|
||||
assert called == []
|
||||
assert json.loads(result)["policy"] == "deny"
|
||||
|
||||
|
||||
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, **_kwargs: "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, **_kwargs: "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, **_kwargs: "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, **_kwargs: "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, **_kwargs: "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, **_kwargs: "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_ask_policy_is_not_bypassed_by_session_yolo(monkeypatch):
|
||||
from hermes_cli import middleware
|
||||
from tools import approval
|
||||
|
||||
monkeypatch.setattr(
|
||||
middleware, "resolve_tool_approval_policy", lambda _name, **_kwargs: "ask"
|
||||
)
|
||||
monkeypatch.setattr(approval, "is_current_session_yolo_enabled", lambda: True)
|
||||
monkeypatch.setattr(approval, "_is_interactive_cli", lambda: True)
|
||||
monkeypatch.setattr(approval, "prompt_dangerous_approval", lambda *_a, **_kw: "deny")
|
||||
called = []
|
||||
|
||||
result = middleware.run_tool_execution_middleware(
|
||||
"write_file", {"path": "notes.txt"}, lambda args: called.append(args)
|
||||
)
|
||||
|
||||
assert called == []
|
||||
assert "User denied" 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"] == {}
|
||||
|
||||
|
||||
def test_reloading_allowlist_replaces_previous_profiles_state(monkeypatch):
|
||||
from tools import approval
|
||||
|
||||
configs = iter(
|
||||
[
|
||||
{"command_allowlist": ["default-only"]},
|
||||
{"command_allowlist": ["work-only"]},
|
||||
]
|
||||
)
|
||||
monkeypatch.setattr("hermes_cli.config.load_config", lambda: next(configs))
|
||||
monkeypatch.setattr(approval, "_permanent_approved", set())
|
||||
|
||||
approval.load_permanent_allowlist()
|
||||
assert approval.is_approved("session", "default-only") is True
|
||||
|
||||
approval.load_permanent_allowlist()
|
||||
assert approval.is_approved("session", "default-only") is False
|
||||
assert approval.is_approved("session", "work-only") is True
|
||||
|
||||
|
||||
def test_failed_profile_allowlist_load_clears_previous_profile_state(monkeypatch):
|
||||
from tools import approval
|
||||
|
||||
configs = iter([{"command_allowlist": ["default-only"]}, ValueError("malformed profile")])
|
||||
|
||||
def load_config():
|
||||
value = next(configs)
|
||||
if isinstance(value, Exception):
|
||||
raise value
|
||||
return value
|
||||
|
||||
monkeypatch.setattr("hermes_cli.config.load_config", load_config)
|
||||
monkeypatch.setattr(approval, "_permanent_approved", set())
|
||||
approval.load_permanent_allowlist()
|
||||
assert approval.is_approved("session", "default-only") is True
|
||||
|
||||
assert approval.load_permanent_allowlist() == set()
|
||||
assert approval.is_approved("session", "default-only") is False
|
||||
|
|
@ -50,6 +50,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
|
||||
|
|
@ -1605,8 +1618,9 @@ def approve_permanent(pattern_key: str):
|
|||
|
||||
|
||||
def load_permanent(patterns: set):
|
||||
"""Bulk-load permanent allowlist entries from config."""
|
||||
"""Replace permanent entries with the active profile's allowlist."""
|
||||
with _lock:
|
||||
_permanent_approved.clear()
|
||||
_permanent_approved.update(patterns)
|
||||
|
||||
|
||||
|
|
@ -1662,10 +1676,14 @@ def load_permanent_allowlist() -> set:
|
|||
from hermes_cli.config import load_config
|
||||
config = load_config()
|
||||
patterns = set(config.get("command_allowlist", []) or [])
|
||||
if patterns:
|
||||
load_permanent(patterns)
|
||||
# Replace rather than merge: long-lived processes can reload under a
|
||||
# different profile, whose permanent approvals must remain isolated.
|
||||
load_permanent(patterns)
|
||||
return patterns
|
||||
except Exception as e:
|
||||
# Fail closed across malformed profile transitions: stale grants from
|
||||
# the previously active profile must not remain live in this process.
|
||||
load_permanent(set())
|
||||
logger.warning("Failed to load permanent allowlist: %s", e)
|
||||
return set()
|
||||
|
||||
|
|
@ -2031,6 +2049,7 @@ def _run_approval_gate(
|
|||
autoapprove_log_prefix: str,
|
||||
fail_closed_when_no_human: bool = False,
|
||||
no_human_block_message: str = "",
|
||||
enforce_under_yolo: bool = False,
|
||||
) -> dict:
|
||||
"""Shared human-approval gate for a flagged action (command or tool).
|
||||
|
||||
|
|
@ -2075,7 +2094,9 @@ def _run_approval_gate(
|
|||
# --yolo bypasses all approval prompts (session- or process-scoped).
|
||||
# Hardline blocks are handled by the caller BEFORE this gate, so yolo
|
||||
# here only skips the recoverable approval layer.
|
||||
if _YOLO_MODE_FROZEN or is_current_session_yolo_enabled():
|
||||
if not enforce_under_yolo and (
|
||||
_YOLO_MODE_FROZEN or is_current_session_yolo_enabled()
|
||||
):
|
||||
return {"approved": True, "message": None}
|
||||
|
||||
session_key = get_current_session_key()
|
||||
|
|
@ -2298,6 +2319,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}
|
||||
|
||||
|
|
@ -2329,6 +2356,7 @@ def request_tool_approval(
|
|||
*,
|
||||
rule_key: str = "",
|
||||
approval_callback=None,
|
||||
enforce_under_yolo: bool = False,
|
||||
) -> dict:
|
||||
"""Escalate an arbitrary tool call to the human-approval gate.
|
||||
|
||||
|
|
@ -2407,6 +2435,7 @@ def request_tool_approval(
|
|||
"but no interactive user or gateway is present to approve it. "
|
||||
"A plugin flagged this action for human confirmation."
|
||||
),
|
||||
enforce_under_yolo=enforce_under_yolo,
|
||||
)
|
||||
|
||||
|
||||
|
|
@ -2613,6 +2642,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}
|
||||
|
||||
|
|
|
|||
|
|
@ -5315,7 +5315,10 @@ def refresh_agent_mcp_tools(
|
|||
# (``build_api_kwargs``) can't see a partial rebuild or a cross-attribute
|
||||
# half-swap. ``staged_engine_names`` are the context-engine routing names
|
||||
# this rebuild actually appended (matching agent_init's dedup-aware add).
|
||||
staged_engine_names = _reinject_post_build_tools(agent, new_defs, new_names)
|
||||
staged_dynamic_entries: dict = {}
|
||||
staged_engine_names = _reinject_post_build_tools(
|
||||
agent, new_defs, new_names, dynamic_entries=staged_dynamic_entries
|
||||
)
|
||||
|
||||
# Single atomic read-diff-publish so the returned ``added`` is consistent
|
||||
# with what was actually published, even under concurrent callers, and a
|
||||
|
|
@ -5335,12 +5338,14 @@ def refresh_agent_mcp_tools(
|
|||
for t in (getattr(agent, "tools", None) or [])
|
||||
}
|
||||
if new_names == current:
|
||||
# No change → leave the live snapshot untouched (no churn), but
|
||||
# record the generation so an in-flight older caller can't clobber.
|
||||
# No schema change, but refresh dynamic policy identities in case
|
||||
# this agent predates metadata propagation.
|
||||
agent._dynamic_tool_entries = staged_dynamic_entries
|
||||
agent._tool_snapshot_generation = max(published_gen, snapshot_generation)
|
||||
return set()
|
||||
agent.tools = new_defs
|
||||
agent.valid_tool_names = new_names
|
||||
agent._dynamic_tool_entries = staged_dynamic_entries
|
||||
# Publish context-engine routing names atomically with the snapshot.
|
||||
engine_names = getattr(agent, "_context_engine_tool_names", None)
|
||||
if isinstance(engine_names, set):
|
||||
|
|
@ -5350,7 +5355,9 @@ def refresh_agent_mcp_tools(
|
|||
return new_names - current
|
||||
|
||||
|
||||
def _reinject_post_build_tools(agent, tools_list: list, name_set: set) -> set:
|
||||
def _reinject_post_build_tools(
|
||||
agent, tools_list: list, name_set: set, *, dynamic_entries: Optional[dict] = None
|
||||
) -> set:
|
||||
"""Append memory-provider and context-engine tools onto staged locals.
|
||||
|
||||
Mirrors the post-``get_tool_definitions`` injection in ``agent_init`` so a
|
||||
|
|
@ -5365,12 +5372,16 @@ def _reinject_post_build_tools(agent, tools_list: list, name_set: set) -> set:
|
|||
caller publishes this into ``agent._context_engine_tool_names`` atomically
|
||||
with the snapshot.
|
||||
"""
|
||||
def _add(schema: dict) -> bool:
|
||||
def _add(schema: dict, toolset: str) -> bool:
|
||||
name = schema.get("name", "")
|
||||
if not name or name in name_set:
|
||||
return False
|
||||
tools_list.append({"type": "function", "function": schema})
|
||||
name_set.add(name)
|
||||
if dynamic_entries is not None:
|
||||
from tools.registry import DynamicToolEntry
|
||||
|
||||
dynamic_entries[name] = DynamicToolEntry(name, toolset)
|
||||
return True
|
||||
|
||||
# Memory-provider tools (mem0/honcho/byterover/supermemory/…).
|
||||
|
|
@ -5383,7 +5394,7 @@ def _reinject_post_build_tools(agent, tools_list: list, name_set: set) -> set:
|
|||
if "memory" in name_set or memory_provider_tools_enabled(getattr(agent, "enabled_toolsets", None)):
|
||||
for schema in get_mem_schemas():
|
||||
if isinstance(schema, dict):
|
||||
_add(schema)
|
||||
_add(schema, "memory")
|
||||
except Exception:
|
||||
logger.debug("Memory-provider tool re-injection skipped", exc_info=True)
|
||||
|
||||
|
|
@ -5407,7 +5418,7 @@ def _reinject_post_build_tools(agent, tools_list: list, name_set: set) -> set:
|
|||
# Only claim the routing name when WE appended the schema, so a
|
||||
# name already owned by a registry/plugin tool keeps its own
|
||||
# dispatch (matches agent_init.py's `continue`-before-claim).
|
||||
if _add(schema) and name:
|
||||
if _add(schema, "context_engine") and name:
|
||||
staged_engine_names.add(name)
|
||||
except Exception:
|
||||
logger.debug("Context-engine tool re-injection skipped", exc_info=True)
|
||||
|
|
|
|||
|
|
@ -21,6 +21,7 @@ import logging
|
|||
import sys
|
||||
import threading
|
||||
import time
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import Callable, Dict, List, Optional, Set
|
||||
|
||||
|
|
@ -107,6 +108,14 @@ class ToolEntry:
|
|||
self.dynamic_schema_overrides = dynamic_schema_overrides
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class DynamicToolEntry:
|
||||
"""Policy identity for a tool advertised outside the global registry."""
|
||||
|
||||
name: str
|
||||
toolset: str
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# check_fn TTL cache
|
||||
#
|
||||
|
|
@ -614,13 +623,31 @@ class ToolRegistry:
|
|||
entry = self.get_entry(name)
|
||||
if not entry:
|
||||
return json.dumps({"error": f"Unknown tool: {name}"})
|
||||
try:
|
||||
|
||||
def execute(next_args: dict) -> str | dict:
|
||||
if entry.is_async:
|
||||
from model_tools import _run_async
|
||||
result = _run_async(entry.handler(args, **kwargs))
|
||||
result = _run_async(entry.handler(next_args, **kwargs))
|
||||
else:
|
||||
result = entry.handler(args, **kwargs)
|
||||
result = entry.handler(next_args, **kwargs)
|
||||
return self._normalize_handler_result(name, result)
|
||||
|
||||
try:
|
||||
# This is the narrowest shared execution boundary: model calls,
|
||||
# PluginContext, MCP, and direct framework callers converge here.
|
||||
# The actual entry is passed through so custom registries resolve
|
||||
# their own toolset. Only an explicitly marked, immediate wrapper
|
||||
# handoff is deduplicated; recursive dispatch remains a new call.
|
||||
from hermes_cli.middleware import run_tool_execution_middleware
|
||||
|
||||
return run_tool_execution_middleware(
|
||||
name,
|
||||
args,
|
||||
execute,
|
||||
registry_entry=entry,
|
||||
registry_dispatch=self,
|
||||
**kwargs,
|
||||
)
|
||||
except Exception as e:
|
||||
logger.exception("Tool %s dispatch error: %s", name, e)
|
||||
# Route through the sanitizer so framing tokens / CDATA / fences
|
||||
|
|
|
|||
|
|
@ -33,6 +33,10 @@ approvals:
|
|||
mode: smart # smart | manual | off
|
||||
timeout: 60 # seconds to wait for user response (default: 60)
|
||||
cron_mode: deny # deny | approve — what cron jobs do when they hit a dangerous command
|
||||
tool_policies: # optional per-tool / per-toolset glob rules
|
||||
"*": ask # supervise every tool by default
|
||||
read_file: allow # still asks because the broader rule also matches
|
||||
terminal: deny # block this tool without prompting
|
||||
mcp_reload_confirm: true # /reload-mcp asks before invalidating the MCP tool cache
|
||||
destructive_slash_confirm: true # /clear, /new, /reset, /undo prompt before discarding state
|
||||
```
|
||||
|
|
@ -44,6 +48,7 @@ The full set of keys:
|
|||
| `mode` | `smart` | Approval policy for dangerous shell commands — see the table below. |
|
||||
| `timeout` | `60` | Seconds Hermes waits for an approval reply before timing out. |
|
||||
| `cron_mode` | `deny` | How [cron jobs](./features/cron.md) behave headlessly when they trigger a dangerous-command prompt. `deny` blocks the command (the agent must find another path); `approve` auto-approves everything in cron context. |
|
||||
| `tool_policies` | `{}` | Profile-local, case-insensitive glob mapping for tools and registered toolsets. Values are `allow`, `ask`, or `deny`. |
|
||||
| `mcp_reload_confirm` | `true` | When true, `/reload-mcp` asks before rebuilding the MCP tool set. Rebuilding invalidates the provider prompt cache (tool schemas live in the system prompt), so the next message re-sends full input tokens. Users who click **Always Approve** flip this key to `false`. |
|
||||
| `destructive_slash_confirm` | `true` | When true, destructive session slash commands (`/clear`, `/new`, `/reset`, `/undo`) prompt before discarding conversation state. Three-option dialog (Approve Once / Always Approve / Cancel) routed through native yes/no buttons on Telegram, Discord, and Slack; text fallback elsewhere. Users who click **Always Approve** flip this key to `false`. TUI uses its own modal overlay (set `HERMES_TUI_NO_CONFIRM=1` to opt out there). |
|
||||
|
||||
|
|
@ -53,6 +58,28 @@ The full set of keys:
|
|||
| **manual** | Always prompt the user for approval on dangerous commands. |
|
||||
| **off** | Disable all approval checks — equivalent to running with `--yolo`. All commands execute without prompts. |
|
||||
|
||||
### Per-tool policies
|
||||
|
||||
`approvals.tool_policies` applies at the shared execution boundary used by the
|
||||
CLI, gateway platforms, MCP tools, plugins, and both sequential and concurrent
|
||||
agent dispatch. It does not alter the advertised tool schemas or system prompt,
|
||||
so changing policy does not invalidate the conversation prompt cache.
|
||||
|
||||
- `deny` returns a blocked tool result without executing the tool.
|
||||
- `ask` uses the normal CLI or gateway approval UI and fails closed when no
|
||||
human approval surface is available. Explicit `ask` rules remain active in
|
||||
YOLO / `approvals.mode: off`; cron still follows `approvals.cron_mode`.
|
||||
- `allow` skips this policy prompt. For `terminal`, it also skips the ordinary
|
||||
dangerous-command prompt, but never bypasses hardline blocks, user-defined
|
||||
`approvals.deny` rules, cron policy, plugin escalation, tool scope, or
|
||||
sensitive-path protections.
|
||||
|
||||
Patterns are case-insensitive Python `fnmatch` globs matched against both the
|
||||
tool name and its registered toolset. Across every overlapping match, the most
|
||||
restrictive result always wins (`deny` > `ask` > `allow`), regardless of pattern
|
||||
specificity or whether the match came from the tool name or toolset. Always
|
||||
quote wildcard keys in YAML.
|
||||
|
||||
:::warning
|
||||
Setting `approvals.mode: off` disables all safety prompts. Use only in trusted environments (CI/CD, containers, etc.).
|
||||
:::
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue