diff --git a/hermes_cli/config.py b/hermes_cli/config.py index 981533f15e7..dc4ae04f74f 100644 --- a/hermes_cli/config.py +++ b/hermes_cli/config.py @@ -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 diff --git a/hermes_cli/middleware.py b/hermes_cli/middleware.py index 8795952a2b7..1cc506267ee 100644 --- a/hermes_cli/middleware.py +++ b/hermes_cli/middleware.py @@ -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( diff --git a/tests/test_tool_approval_policies.py b/tests/test_tool_approval_policies.py new file mode 100644 index 00000000000..e42de040c8d --- /dev/null +++ b/tests/test_tool_approval_policies.py @@ -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"] == {} diff --git a/tools/approval.py b/tools/approval.py index 05f2eb523cc..37c9bab9ffa 100644 --- a/tools/approval.py +++ b/tools/approval.py @@ -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}