From de56e49a7c776f52ffcff8e0d768d64f0faa48cf Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Sat, 8 Aug 2026 12:50:15 -0700 Subject: [PATCH] feat(plugins): add approval transport interface --- cli-config.yaml.example | 3 + hermes_cli/approval_transport.py | 216 +++++++++ hermes_cli/config_defaults.py | 10 + hermes_cli/plugins.py | 63 +++ tests/hermes_cli/test_approval_transport.py | 509 ++++++++++++++++++++ tools/approval.py | 272 +++++++++++ website/docs/user-guide/features/plugins.md | 55 +++ 7 files changed, 1128 insertions(+) create mode 100644 hermes_cli/approval_transport.py create mode 100644 tests/hermes_cli/test_approval_transport.py diff --git a/cli-config.yaml.example b/cli-config.yaml.example index 29d3c3d38a..dd83df0157 100644 --- a/cli-config.yaml.example +++ b/cli-config.yaml.example @@ -396,6 +396,9 @@ terminal: # tirith_path: "tirith" # Path to tirith binary (supports ~ expansion) # tirith_timeout: 5 # Scan timeout in seconds # tirith_fail_open: true # Allow commands if tirith unavailable +# approval: +# transport: builtin # Or an explicitly enabled plugin transport name +# transport_fallback: deny # Set builtin to opt into fallback on transport failure # ============================================================================= # Browser Tool Configuration diff --git a/hermes_cli/approval_transport.py b/hermes_cli/approval_transport.py new file mode 100644 index 0000000000..2596dfa382 --- /dev/null +++ b/hermes_cli/approval_transport.py @@ -0,0 +1,216 @@ +"""Host-owned contract for plugin-provided human approval transports. + +Transports only present an immutable, redacted request and return a correlated +human decision. They do not participate in command detection or authorization +policy. The host validates scope, request binding, and timeout fail-closed. +""" + +from __future__ import annotations + +import asyncio +import hashlib +import inspect +import json +import logging +import queue +import threading +import time +import uuid +from dataclasses import dataclass +from typing import Awaitable, Callable, Literal + +logger = logging.getLogger(__name__) + +_MAX_ACTIVE_TRANSPORT_WORKERS = 8 +_transport_worker_slots = threading.BoundedSemaphore(_MAX_ACTIVE_TRANSPORT_WORKERS) + +ApprovalChoice = Literal["once", "session", "always", "deny"] +ApprovalPresentFn = Callable[ + ["ApprovalRequest"], "ApprovalDecision | Awaitable[ApprovalDecision]" +] + + +@dataclass(frozen=True) +class ApprovalDecision: + """A transport response bound to one exact host-created request.""" + + request_id: str + request_digest: str + choice: str + + +@dataclass(frozen=True) +class ApprovalRequest: + """Immutable, display-only approval request passed to a transport plugin.""" + + schema_version: int + request_id: str + digest: str + command: str + description: str + pattern_key: str + pattern_keys: tuple[str, ...] + surface: str + timeout_seconds: float + allowed_choices: tuple[ApprovalChoice, ...] + + @classmethod + def create( + cls, + *, + command: str, + description: str, + pattern_key: str, + pattern_keys: tuple[str, ...], + session_key: str, + surface: str, + allow_session: bool, + allow_permanent: bool, + timeout_seconds: float = 300, + ) -> "ApprovalRequest": + request_id = uuid.uuid4().hex + choices: list[ApprovalChoice] = ["once"] + if allow_session: + choices.append("session") + if allow_permanent: + choices.append("always") + choices.append("deny") + canonical = { + "schema_version": 1, + "request_id": request_id, + "command": command, + "description": description, + "pattern_key": pattern_key, + "pattern_keys": list(pattern_keys), + "session_key": session_key, + "surface": surface, + "timeout_seconds": timeout_seconds, + "allowed_choices": choices, + } + digest = hashlib.sha256( + json.dumps(canonical, sort_keys=True, separators=(",", ":")).encode("utf-8") + ).hexdigest() + return cls( + schema_version=1, + request_id=request_id, + digest=digest, + command=command, + description=description, + pattern_key=pattern_key, + pattern_keys=pattern_keys, + surface=surface, + timeout_seconds=timeout_seconds, + allowed_choices=tuple(choices), + ) + + def respond(self, choice: ApprovalChoice | str) -> ApprovalDecision: + """Build the correlated response a transport should return.""" + return ApprovalDecision( + request_id=self.request_id, + request_digest=self.digest, + choice=choice, + ) + + +@dataclass(frozen=True) +class ApprovalTransportResult: + """Normalized host result. Any failure is represented as a denial.""" + + choice: ApprovalChoice + failure: str | None = None + + +@dataclass(frozen=True) +class RegisteredApprovalTransport: + """Plugin-owned registration retained by one profile's PluginManager.""" + + name: str + present: ApprovalPresentFn + plugin_id: str + profile_home: str + + +def invoke_approval_transport( + present: ApprovalPresentFn, + request: ApprovalRequest, + *, + timeout_seconds: float, + poll_interval: float = 1.0, + on_poll: Callable[[], None] | None = None, + is_interrupted: Callable[[], bool] | None = None, +) -> ApprovalTransportResult: + """Run a sync or async transport on a bounded daemon worker. + + Async callbacks are awaited with ``asyncio.run`` on that worker, never on a + gateway or TUI event loop. A callback must return before the host timeout; + late results are discarded and cannot authorize another request. + """ + + if not _transport_worker_slots.acquire(blocking=False): + logger.warning("Approval transport worker capacity exhausted") + return ApprovalTransportResult("deny", "busy") + + results: queue.Queue[tuple[str, object]] = queue.Queue(maxsize=1) + + async def _await_value(value): + return await value + + def _run() -> None: + try: + value = present(request) + if inspect.isawaitable(value): + value = asyncio.run(_await_value(value)) + results.put_nowait(("result", value)) + except BaseException as exc: # fail closed even for unusual callback exits + try: + results.put_nowait(("error", exc)) + except queue.Full: + pass + finally: + _transport_worker_slots.release() + + worker = threading.Thread( + target=_run, + name=f"approval-transport-{request.request_id[:8]}", + daemon=True, + ) + try: + worker.start() + except BaseException: + _transport_worker_slots.release() + logger.warning("Could not start approval transport worker") + return ApprovalTransportResult("deny", "error") + deadline = time.monotonic() + max(float(timeout_seconds), 0.0) + while True: + if is_interrupted is not None and is_interrupted(): + logger.info("Approval transport wait interrupted for %s", request.request_id) + return ApprovalTransportResult("deny", "interrupted") + remaining = deadline - time.monotonic() + if remaining <= 0: + logger.warning("Approval transport timed out for request %s", request.request_id) + return ApprovalTransportResult("deny", "timeout") + try: + kind, value = results.get( + timeout=min(max(float(poll_interval), 0.001), remaining) + ) + break + except queue.Empty: + if on_poll is not None: + try: + on_poll() + except Exception: + logger.debug("Approval transport poll callback failed", exc_info=True) + + if kind == "error": + logger.warning("Approval transport failed for request %s", request.request_id) + return ApprovalTransportResult("deny", "error") + if not isinstance(value, ApprovalDecision): + logger.warning("Approval transport returned an invalid decision type") + return ApprovalTransportResult("deny", "invalid") + if value.request_id != request.request_id or value.request_digest != request.digest: + logger.warning("Approval transport returned a stale or mismatched decision") + return ApprovalTransportResult("deny", "stale") + if value.choice not in request.allowed_choices: + logger.warning("Approval transport returned a disallowed choice") + return ApprovalTransportResult("deny", "invalid") + return ApprovalTransportResult(value.choice) diff --git a/hermes_cli/config_defaults.py b/hermes_cli/config_defaults.py index 63e3f1c6e2..8a083744d1 100644 --- a/hermes_cli/config_defaults.py +++ b/hermes_cli/config_defaults.py @@ -2205,6 +2205,16 @@ DEFAULT_CONFIG = { "security": { "allow_private_urls": False, # Allow requests to private/internal IPs (for OpenWrt, proxies, VPNs) "redact_secrets": True, + # Human approval presentation transport. "builtin" preserves the + # current CLI/TUI/gateway/ACP surfaces. A plugin transport is used only + # when named explicitly here. Transport timeout/error/invalid response + # denies unless transport_fallback is explicitly set to "builtin". + # This is presentation only: plugins cannot detect, suppress, or + # auto-approve commands outside a correlated human response. + "approval": { + "transport": "builtin", + "transport_fallback": "deny", + }, # Writes to agent-instruction files (AGENTS.md/CLAUDE.md/SOUL.md/ # .cursorrules, project-local .hermes config) always require human # approval — even under auto-approve/yolo. Extra patterns are diff --git a/hermes_cli/plugins.py b/hermes_cli/plugins.py index 3270e2fa4e..1318726a16 100644 --- a/hermes_cli/plugins.py +++ b/hermes_cli/plugins.py @@ -718,6 +718,23 @@ class PluginContext: except Exception: return "default" + # -- approval transport registration ------------------------------------ + + def register_approval_transport(self, name: str, present_fn: Callable) -> None: + """Register a human approval presentation transport. + + The transport is inactive until the operator explicitly selects + ``security.approval.transport: ``. It receives a host-created, + redacted ``ApprovalRequest`` and may only return a correlated human + decision; command policy and approval persistence remain host-owned. + ``present_fn`` may be synchronous or async. + """ + self._manager.register_approval_transport( + name, + present_fn, + plugin_id=self.manifest.key or self.manifest.name, + ) + # -- tool registration -------------------------------------------------- def register_tool( @@ -1613,6 +1630,8 @@ class PluginManager: # Plugin-registered auxiliary tasks: key → {key, display_name, # description, defaults, plugin}. See PluginContext.register_auxiliary_task. self._aux_tasks: Dict[str, Dict[str, Any]] = {} + # Explicitly-selected, profile-scoped human approval transports. + self._approval_transports: Dict[str, Any] = {} # Slack Block Kit action handlers registered by plugins. Each entry # is (matcher, callback, plugin_name); the Slack adapter wires them # into its slack_bolt App at connect() time. ``matcher`` is whatever @@ -1649,6 +1668,7 @@ class PluginManager: self._plugin_skills.clear() self._portable_mcp_servers.clear() self._aux_tasks.clear() + self._approval_transports.clear() self._slack_action_handlers.clear() self._context_engine = None # Set the flag up front as a re-entrancy guard (a plugin's register() @@ -1839,6 +1859,49 @@ class PluginManager: sum(1 for p in self._plugins.values() if p.enabled), ) + def register_approval_transport( + self, + name: str, + present_fn: Callable, + *, + plugin_id: str, + ) -> None: + """Register one plugin-owned approval transport for this profile.""" + import re + + from hermes_cli.approval_transport import RegisteredApprovalTransport + + clean = str(name).strip().lower() + if clean == "builtin": + raise ValueError("approval transport name 'builtin' is reserved") + if not re.fullmatch(r"[a-z0-9][a-z0-9_-]{0,63}", clean): + raise ValueError( + "approval transport name must match [a-z0-9][a-z0-9_-]{0,63}" + ) + if not callable(present_fn): + raise TypeError("approval transport present_fn must be callable") + if clean in self._approval_transports: + owner = self._approval_transports[clean].plugin_id + raise ValueError( + f"approval transport {clean!r} is already registered by {owner!r}" + ) + self._approval_transports[clean] = RegisteredApprovalTransport( + name=clean, + present=present_fn, + plugin_id=plugin_id, + profile_home=str(get_hermes_home().resolve()), + ) + logger.info("Plugin %s registered approval transport: %s", plugin_id, clean) + + def get_approval_transport(self, name: str): + """Return a transport only inside the profile that registered it.""" + registered = self._approval_transports.get(str(name).strip().lower()) + if registered is None: + return None + if registered.profile_home != str(get_hermes_home().resolve()): + return None + return registered + def _collect_directory_manifests(self) -> List[PluginManifest]: """Collect directory manifests in the same order as full discovery. diff --git a/tests/hermes_cli/test_approval_transport.py b/tests/hermes_cli/test_approval_transport.py new file mode 100644 index 0000000000..421f37239d --- /dev/null +++ b/tests/hermes_cli/test_approval_transport.py @@ -0,0 +1,509 @@ +"""Approval transport plugin contract and fail-closed host routing.""" + +from __future__ import annotations + +import asyncio +import json +import threading +import time + +import pytest +import yaml + +from hermes_cli.plugins import PluginContext, PluginManager, PluginManifest + + +def _manifest(name: str = "fixture-approval") -> PluginManifest: + return PluginManifest( + name=name, + version="1.0.0", + description="fixture", + source="user", + key=name, + ) + + +def _context(manager: PluginManager, name: str = "fixture-approval") -> PluginContext: + return PluginContext(_manifest(name), manager) + + +def _request(): + from hermes_cli.approval_transport import ApprovalRequest + + return ApprovalRequest.create( + command="rm -rf /tmp/example", + description="recursive delete", + pattern_key="rm_recursive", + pattern_keys=("rm_recursive",), + session_key="session-a", + surface="cli", + allow_session=True, + allow_permanent=True, + ) + + +def test_context_registers_owned_approval_transport(): + manager = PluginManager() + callback = lambda request: request.respond("deny") + + _context(manager).register_approval_transport("phone", callback) + + registered = manager.get_approval_transport("phone") + assert registered is not None + assert registered.name == "phone" + assert registered.plugin_id == "fixture-approval" + assert registered.present is callback + + +def test_transport_names_are_unique_and_builtin_is_reserved(): + manager = PluginManager() + _context(manager, "first").register_approval_transport("phone", lambda request: None) + + with pytest.raises(ValueError, match="already registered"): + _context(manager, "second").register_approval_transport("phone", lambda request: None) + with pytest.raises(ValueError, match="reserved"): + _context(manager).register_approval_transport("builtin", lambda request: None) + + +def test_force_reload_clears_transport_registry(monkeypatch): + manager = PluginManager() + _context(manager).register_approval_transport("phone", lambda request: None) + manager._discovered = True + monkeypatch.setattr(manager, "_discover_and_load_inner", lambda: None) + + manager.discover_and_load(force=True) + + assert manager.get_approval_transport("phone") is None + + +def test_transport_registry_is_manager_and_profile_isolated(monkeypatch, tmp_path): + first = PluginManager() + second = PluginManager() + monkeypatch.setenv("HERMES_HOME", str(tmp_path / "profiles" / "work")) + _context(first).register_approval_transport("phone", lambda request: None) + + assert first.get_approval_transport("phone") is not None + assert second.get_approval_transport("phone") is None + monkeypatch.setenv("HERMES_HOME", str(tmp_path / "profiles" / "personal")) + assert first.get_approval_transport("phone") is None + + +def test_host_accepts_bound_sync_and_async_decisions(): + from hermes_cli.approval_transport import invoke_approval_transport + + request = _request() + sync_result = invoke_approval_transport( + lambda received: received.respond("session"), request, timeout_seconds=1 + ) + + async def present(received): + await asyncio.sleep(0) + return received.respond("once") + + async_result = invoke_approval_transport(present, request, timeout_seconds=1) + + assert sync_result.choice == "session" + assert sync_result.failure is None + assert async_result.choice == "once" + assert async_result.failure is None + + +def test_host_rejects_scope_not_offered_by_request(): + from hermes_cli.approval_transport import ApprovalRequest, invoke_approval_transport + + request = ApprovalRequest.create( + command="dangerous", + description="dangerous", + pattern_key="danger", + pattern_keys=("danger",), + session_key="session-a", + surface="cli", + allow_session=False, + allow_permanent=False, + ) + + result = invoke_approval_transport( + lambda received: received.respond("always"), request, timeout_seconds=1 + ) + + assert result.choice == "deny" + assert result.failure == "invalid" + + +@pytest.mark.parametrize( + ("present", "failure"), + [ + (lambda request: {"choice": "once"}, "invalid"), + (lambda request: request.respond("bogus"), "invalid"), + ( + lambda request: type(request.respond("once"))( + request_id="stale", + request_digest=request.digest, + choice="once", + ), + "stale", + ), + ( + lambda request: type(request.respond("once"))( + request_id=request.request_id, + request_digest="changed", + choice="once", + ), + "stale", + ), + ], +) +def test_host_rejects_invalid_or_stale_decisions(present, failure): + from hermes_cli.approval_transport import invoke_approval_transport + + result = invoke_approval_transport(present, _request(), timeout_seconds=1) + + assert result.choice == "deny" + assert result.failure == failure + + +def test_host_timeout_and_exception_deny_without_waiting_forever(): + from hermes_cli.approval_transport import invoke_approval_transport + + def hangs(_request): + time.sleep(1) + + def crashes(_request): + raise RuntimeError("transport offline") + + started = time.monotonic() + timeout = invoke_approval_transport(hangs, _request(), timeout_seconds=0.02) + exception = invoke_approval_transport(crashes, _request(), timeout_seconds=1) + + assert time.monotonic() - started < 0.5 + assert (timeout.choice, timeout.failure) == ("deny", "timeout") + assert (exception.choice, exception.failure) == ("deny", "error") + + +def test_host_transport_wait_is_interruptible_and_pollable(): + from hermes_cli.approval_transport import invoke_approval_transport + + polls = [] + + def hangs(_request): + time.sleep(1) + + result = invoke_approval_transport( + hangs, + _request(), + timeout_seconds=1, + poll_interval=0.01, + on_poll=lambda: polls.append(1), + is_interrupted=lambda: len(polls) >= 1, + ) + + assert result.choice == "deny" + assert result.failure == "interrupted" + assert polls + + +def test_host_caps_hung_transport_workers(): + from hermes_cli.approval_transport import ( + _MAX_ACTIVE_TRANSPORT_WORKERS, + invoke_approval_transport, + ) + + release = threading.Event() + + def hangs(_request): + release.wait() + + try: + results = [ + invoke_approval_transport(hangs, _request(), timeout_seconds=0.001) + for _ in range(_MAX_ACTIVE_TRANSPORT_WORKERS + 1) + ] + finally: + release.set() + + assert results[-1].choice == "deny" + assert results[-1].failure == "busy" + + +def _configure_manual_guard(monkeypatch, approval_module, manager, *, fallback=None): + monkeypatch.setattr(approval_module, "_get_approval_mode", lambda: "manual") + monkeypatch.setattr(approval_module, "_is_interactive_cli", lambda: True) + monkeypatch.setattr(approval_module, "_is_gateway_approval_context", lambda: False) + monkeypatch.setattr(approval_module, "detect_hardline_command", lambda command: (False, "")) + monkeypatch.setattr(approval_module, "_check_sudo_stdin_guard", lambda command: (False, "")) + monkeypatch.setattr(approval_module, "_match_user_deny_rule", lambda command: None) + monkeypatch.setattr(approval_module, "_command_matches_permanent_allowlist", lambda command: False) + monkeypatch.setattr(approval_module, "detect_dangerous_command", lambda command: (True, "danger", "dangerous")) + monkeypatch.setattr( + approval_module, + "get_current_session_key", + lambda *args, **kwargs: "session-a", + ) + monkeypatch.setattr(approval_module, "is_approved", lambda *args: False) + monkeypatch.setattr(approval_module, "get_plugin_manager", lambda: manager, raising=False) + monkeypatch.setattr( + approval_module, + "_get_approval_transport_config", + lambda: ("phone", fallback), + raising=False, + ) + monkeypatch.setattr("tools.tirith_security.check_command_security", lambda command: {"action": "allow"}) + + +def test_cli_selected_transport_replaces_builtin_prompt(monkeypatch): + from tools import approval + + manager = PluginManager() + seen = [] + _context(manager).register_approval_transport( + "phone", lambda request: seen.append(request) or request.respond("once") + ) + _configure_manual_guard(monkeypatch, approval, manager) + + def builtin(*args, **kwargs): + raise AssertionError("builtin prompt must not materialize") + + result = approval.check_all_command_guards( + "rm -rf /tmp/example", "local", approval_callback=builtin + ) + + assert result["approved"] is True + assert result["user_approved"] is True + assert len(seen) == 1 + assert seen[0].command == "rm -rf /tmp/example" + assert seen[0].allowed_choices == ("once", "session", "always", "deny") + + +def test_gateway_selected_transport_does_not_require_gateway_notifier(monkeypatch): + from tools import approval + + manager = PluginManager() + seen = [] + _context(manager).register_approval_transport( + "phone", lambda request: seen.append(request) or request.respond("once") + ) + _configure_manual_guard(monkeypatch, approval, manager) + monkeypatch.setattr(approval, "_is_interactive_cli", lambda: False) + monkeypatch.setattr(approval, "_is_gateway_approval_context", lambda: True) + monkeypatch.setattr(approval, "_gateway_notify_cbs", {}) + + result = approval.check_all_command_guards("rm -rf /tmp/example", "local") + + assert result["approved"] is True + assert len(seen) == 1 + assert seen[0].surface == "gateway" + + +def test_execute_code_gateway_uses_selected_transport(monkeypatch): + from tools import approval + + manager = PluginManager() + seen = [] + _context(manager).register_approval_transport( + "phone", lambda request: seen.append(request) or request.respond("once") + ) + monkeypatch.setattr(approval, "_get_approval_mode", lambda: "manual") + monkeypatch.setattr(approval, "_is_gateway_approval_context", lambda: True) + monkeypatch.setattr(approval, "_is_cron_approval_context", lambda: False) + monkeypatch.setattr( + approval, "get_current_session_key", lambda *args, **kwargs: "session-a" + ) + monkeypatch.setattr(approval, "is_approved", lambda *args: False) + monkeypatch.setattr(approval, "get_plugin_manager", lambda: manager) + monkeypatch.setattr( + approval, "_get_approval_transport_config", lambda: ("phone", None) + ) + monkeypatch.setattr(approval, "_gateway_notify_cbs", {}) + + result = approval.check_execute_code_guard("print('ok')", "local") + + assert result["approved"] is True + assert len(seen) == 1 + assert seen[0].pattern_key == "execute_code" + assert seen[0].surface == "gateway" + + +def test_transport_failure_denies_without_builtin_fallback(monkeypatch): + from tools import approval + + manager = PluginManager() + _context(manager).register_approval_transport("phone", lambda request: {"choice": "once"}) + _configure_manual_guard(monkeypatch, approval, manager) + builtin_calls = [] + + result = approval.check_all_command_guards( + "rm -rf /tmp/example", + "local", + approval_callback=lambda *args, **kwargs: builtin_calls.append(1) or "once", + ) + + assert result["approved"] is False + assert result["outcome"] == "transport_invalid" + assert builtin_calls == [] + + +def test_redaction_failure_denies_before_transport_callback(monkeypatch): + import agent.redact + from tools import approval + + manager = PluginManager() + calls = [] + _context(manager).register_approval_transport( + "phone", lambda request: calls.append(request) or request.respond("once") + ) + _configure_manual_guard(monkeypatch, approval, manager) + + def redaction_failed(text): + if text in {"rm -rf /tmp/example", "dangerous"}: + raise RuntimeError("redactor unavailable") + return text + + monkeypatch.setattr(agent.redact, "redact_sensitive_text", redaction_failed) + + result = approval.check_all_command_guards("rm -rf /tmp/example", "local") + + assert result["approved"] is False + assert result["outcome"] == "transport_error" + assert calls == [] + + +def test_explicit_builtin_fallback_uses_existing_surface(monkeypatch): + from tools import approval + + manager = PluginManager() + _context(manager).register_approval_transport("phone", lambda request: {"choice": "once"}) + _configure_manual_guard(monkeypatch, approval, manager, fallback="builtin") + builtin_calls = [] + + result = approval.check_all_command_guards( + "rm -rf /tmp/example", + "local", + approval_callback=lambda *args, **kwargs: builtin_calls.append(1) or "once", + ) + + assert result["approved"] is True + assert builtin_calls == [1] + + +def test_live_temp_home_fixture_plugin_routes_and_hardline_stays_core_owned( + tmp_path, monkeypatch +): + """Real discovery + config + guard path under an isolated HERMES_HOME.""" + import hermes_cli.plugins as plugins_module + from tools import approval + + home = tmp_path / "hermes-home" + plugin_dir = home / "plugins" / "fixture-approval" + bundled = tmp_path / "empty-bundled" + plugin_dir.mkdir(parents=True) + bundled.mkdir() + (plugin_dir / "plugin.yaml").write_text( + yaml.safe_dump( + { + "name": "fixture-approval", + "version": "1.0.0", + "description": "approval transport fixture", + } + ) + ) + (plugin_dir / "__init__.py").write_text( + """import json +import os +from pathlib import Path + + +def present(request): + output = Path(os.environ["HERMES_HOME"]) / "transport-invocations.jsonl" + with output.open("a", encoding="utf-8") as handle: + handle.write(json.dumps({ + "request_id": request.request_id, + "digest": request.digest, + "command": request.command, + "surface": request.surface, + "timeout_seconds": request.timeout_seconds, + }) + "\\n") + return request.respond("once") + + +def register(ctx): + ctx.register_approval_transport("fixture", present) +""" + ) + home.mkdir(exist_ok=True) + (home / "config.yaml").write_text( + yaml.safe_dump( + { + "plugins": {"enabled": ["fixture-approval"]}, + "approvals": {"mode": "manual", "timeout": 2}, + "security": { + "tirith_enabled": False, + "approval": {"transport": "fixture"}, + }, + } + ) + ) + monkeypatch.setenv("HERMES_HOME", str(home)) + monkeypatch.setenv("HERMES_BUNDLED_PLUGINS", str(bundled)) + monkeypatch.delenv("HERMES_GATEWAY_SESSION", raising=False) + monkeypatch.setattr(approval, "_YOLO_MODE_FROZEN", False) + manager = PluginManager() + monkeypatch.setattr(plugins_module, "_plugin_manager", manager) + manager.discover_and_load() + token = approval.set_hermes_interactive_context(True) + approval.clear_session("local") + approval._permanent_approved.clear() + try: + routed = approval.check_all_command_guards( + "rm -rf /tmp/hermes-approval-transport-fixture", "local" + ) + manager.discover_and_load(force=True) + reloaded = approval.check_all_command_guards( + "rm -rf /tmp/hermes-approval-transport-fixture-reloaded", "local" + ) + gateway_token = approval.set_hermes_interactive_context(False) + monkeypatch.setenv("HERMES_GATEWAY_SESSION", "1") + try: + gateway_routed = approval.check_all_command_guards( + "rm -rf /tmp/hermes-approval-transport-fixture-gateway", "local" + ) + finally: + approval.reset_hermes_interactive_context(gateway_token) + hardline = approval.check_all_command_guards("rm -rf /", "local") + finally: + approval.reset_hermes_interactive_context(token) + + records = [ + json.loads(line) + for line in (home / "transport-invocations.jsonl").read_text().splitlines() + ] + assert routed["approved"] is True + assert reloaded["approved"] is True + assert gateway_routed["approved"] is True + assert records[0]["request_id"] + assert records[0]["digest"] + assert records[0]["surface"] == "cli" + assert records[0]["timeout_seconds"] == 2 + assert records[2]["surface"] == "gateway" + assert hardline["approved"] is False + assert len(records) == 3 + + +def test_hardline_blocks_before_selected_transport(monkeypatch): + from tools import approval + + manager = PluginManager() + calls = [] + _context(manager).register_approval_transport( + "phone", lambda request: calls.append(request) or request.respond("once") + ) + _configure_manual_guard(monkeypatch, approval, manager) + monkeypatch.setattr( + approval, + "detect_hardline_command", + lambda command: (True, "recursive delete of root filesystem"), + ) + + result = approval.check_all_command_guards("rm -rf /", "local") + + assert result["approved"] is False + assert calls == [] diff --git a/tools/approval.py b/tools/approval.py index ce3f586f89..60b07c5bd6 100644 --- a/tools/approval.py +++ b/tools/approval.py @@ -3666,6 +3666,160 @@ def _format_tirith_description(tirith_result: dict) -> str: return "Security scan — " + "; ".join(parts) +def get_plugin_manager(): + """Lazy plugin-manager seam used by tests and early tool-only imports.""" + from hermes_cli.plugins import get_plugin_manager as _get_manager + + return _get_manager() + + +def _get_approval_transport_config() -> tuple[str, str | None]: + """Return explicitly selected transport and fail-closed fallback mode.""" + try: + from hermes_cli.config import load_config_readonly + + config = load_config_readonly() or {} + approval_config = ((config.get("security") or {}).get("approval") or {}) + selected = str(approval_config.get("transport") or "builtin").strip().lower() + fallback = str(approval_config.get("transport_fallback") or "").strip().lower() + except Exception: + # An unreadable/malformed selection must not silently materialize a + # prompt on a built-in surface the operator may not be watching. + return "config-error", None + return selected or "builtin", "builtin" if fallback == "builtin" else None + + +def _present_with_selected_transport( + *, + command: str, + description: str, + pattern_key: str, + pattern_keys: list[str], + session_key: str, + surface: str, + allow_session: bool, + allow_permanent: bool, +) -> dict: + """Present through an explicitly selected plugin transport, if any.""" + name, fallback = _get_approval_transport_config() + if name == "builtin": + return {"selected": False} + + try: + registered = get_plugin_manager().get_approval_transport(name) + except Exception: + logger.warning("Could not resolve selected approval transport %r", name, exc_info=True) + registered = None + if registered is None: + logger.warning("Selected approval transport %r is unavailable", name) + return { + "selected": True, + "choice": "deny", + "failure": "unavailable", + "fallback": fallback, + "name": name, + } + + try: + from agent.redact import redact_sensitive_text + from hermes_cli.approval_transport import ApprovalRequest, invoke_approval_transport + + timeout_seconds = _get_approval_timeout() + request = ApprovalRequest.create( + command=redact_sensitive_text(command), + description=redact_sensitive_text(description), + pattern_key=pattern_key, + pattern_keys=tuple(pattern_keys), + session_key=session_key, + surface=surface, + allow_session=allow_session, + allow_permanent=allow_permanent, + timeout_seconds=timeout_seconds, + ) + except Exception: + # Never fall back to raw text if redaction or request construction + # fails. The selected boundary exists, so fail closed without calling + # the plugin or leaking the unredacted payload to logs/hooks. + logger.warning("Could not build redacted plugin approval request") + return { + "selected": True, + "choice": "deny", + "failure": "error", + "fallback": None, + "name": name, + } + hook_surface = f"transport:{name}" + _fire_approval_hook( + "pre_approval_request", + command=request.command, + description=request.description, + pattern_key=pattern_key, + pattern_keys=list(pattern_keys), + session_key=session_key, + surface=hook_surface, + request_id=request.request_id, + request_digest=request.digest, + ) + try: + from tools.environments.base import touch_activity_if_due + except Exception: # pragma: no cover - minimal tool-only environments + touch_activity_if_due = None + now = time.monotonic() + activity_state = {"last_touch": now, "start": now} + + def _poll() -> None: + if touch_activity_if_due is not None: + touch_activity_if_due(activity_state, "waiting for plugin approval transport") + + with human_wait_window(session_key): + result = invoke_approval_transport( + registered.present, + request, + timeout_seconds=timeout_seconds, + on_poll=_poll, + is_interrupted=is_interrupted, + ) + hook_choice = result.choice if result.failure is None else f"transport_{result.failure}" + _fire_approval_hook( + "post_approval_response", + command=request.command, + description=request.description, + pattern_key=pattern_key, + pattern_keys=list(pattern_keys), + session_key=session_key, + surface=hook_surface, + choice=hook_choice, + request_id=request.request_id, + request_digest=request.digest, + ) + return { + "selected": True, + "choice": result.choice, + "failure": result.failure, + "fallback": fallback, + "name": name, + } + + +def _transport_denied_result( + *, pattern_key: str, description: str, failure: str +) -> dict: + breaker_addendum = _denial_breaker_addendum(get_current_session_key()) + return { + "approved": False, + "message": ( + f"BLOCKED: Selected approval transport failed ({failure}); the user " + "has NOT consented to this action. Do NOT retry this command or " + "attempt the same outcome through another route." + f"{breaker_addendum}" + ), + "pattern_key": pattern_key, + "description": description, + "outcome": f"transport_{failure}", + "user_consent": False, + } + + def _await_gateway_decision(session_key: str, notify_cb, approval_data: dict, *, surface: str = "gateway") -> dict: """Enqueue *approval_data*, notify the user, and block the calling agent @@ -4060,6 +4214,70 @@ def check_all_command_guards(command: str, env_type: str, # session — the UI was stricter than the persistence layer. has_permanent_capable = any(not is_t for _, _, is_t in warnings) + # An explicitly selected plugin transport replaces every built-in prompt + # surface (CLI/TUI/gateway/ACP). Detection, allowed scopes, persistence, + # timeout, and final authorization remain host-owned. A failed transport + # reaches a built-in surface only under the explicit fallback opt-in. + transport_attempt = _present_with_selected_transport( + command=command, + description=combined_desc, + pattern_key=primary_key, + pattern_keys=all_keys, + session_key=session_key, + surface="gateway" if (is_gateway or is_ask) else "cli", + allow_session=not smart_denied_for_owner, + allow_permanent=has_permanent_capable and not smart_denied_for_owner, + ) + if transport_attempt.get("selected"): + transport_failure = transport_attempt.get("failure") + if transport_failure and transport_attempt.get("fallback") == "builtin": + logger.warning( + "Approval transport %r failed (%s); using explicit builtin fallback", + transport_attempt.get("name"), + transport_failure, + ) + elif transport_failure: + return _transport_denied_result( + pattern_key=primary_key, + description=combined_desc, + failure=transport_failure, + ) + else: + transport_choice = transport_attempt.get("choice") + if transport_choice == "deny": + _record_denial(session_key) + breaker_addendum = _denial_breaker_addendum(session_key) + return { + "approved": False, + "message": ( + "BLOCKED: User denied this command through the selected " + "approval transport. The user has NOT consented to this " + "action. Do NOT retry or attempt the same outcome through " + f"another route.{breaker_addendum}" + ), + "pattern_key": primary_key, + "description": combined_desc, + "outcome": "denied", + "user_consent": False, + } + if not smart_denied_for_owner: + for key, _, is_tirith in warnings: + if transport_choice == "session" or ( + transport_choice == "always" and is_tirith + ): + approve_session(session_key, key) + elif transport_choice == "always": + approve_session(session_key, key) + approve_permanent(key) + save_permanent_allowlist(_permanent_approved) + _reset_denials(session_key) + return { + "approved": True, + "message": None, + "user_approved": True, + "description": combined_desc, + } + # Gateway/async approval — block the agent thread until the user # responds with /approve or /deny, mirroring the CLI's synchronous # input() flow. The agent never sees "approval_required"; it either @@ -4423,6 +4641,60 @@ def check_execute_code_guard(code: str, env_type: str, display_code = redact_sensitive_text(code) display_description = redact_sensitive_text(description) + transport_attempt = _present_with_selected_transport( + command=command, + description=description, + pattern_key=pattern_key, + pattern_keys=[pattern_key], + session_key=session_key, + surface="gateway", + allow_session=not smart_denied_for_owner, + allow_permanent=not smart_denied_for_owner, + ) + if transport_attempt.get("selected"): + transport_failure = transport_attempt.get("failure") + if transport_failure and transport_attempt.get("fallback") == "builtin": + logger.warning( + "Approval transport %r failed (%s); using explicit builtin fallback", + transport_attempt.get("name"), + transport_failure, + ) + elif transport_failure: + return _transport_denied_result( + pattern_key=pattern_key, + description=description, + failure=transport_failure, + ) + else: + choice = transport_attempt.get("choice") + if choice == "deny": + _record_denial(session_key) + return { + "approved": False, + "message": ( + "BLOCKED: User denied execute_code through the selected " + "approval transport. The user has NOT consented." + ), + "pattern_key": pattern_key, + "description": description, + "outcome": "denied", + "user_consent": False, + } + if not smart_denied_for_owner: + if choice == "session": + approve_session(session_key, pattern_key) + elif choice == "always": + approve_session(session_key, pattern_key) + approve_permanent(pattern_key) + save_permanent_allowlist(_permanent_approved) + _reset_denials(session_key) + return { + "approved": True, + "message": None, + "user_approved": True, + "description": description, + } + notify_cb = None with _lock: notify_cb = _gateway_notify_cbs.get(session_key) diff --git a/website/docs/user-guide/features/plugins.md b/website/docs/user-guide/features/plugins.md index 3896124965..e4e170e850 100644 --- a/website/docs/user-guide/features/plugins.md +++ b/website/docs/user-guide/features/plugins.md @@ -112,6 +112,7 @@ Every `ctx.*` API below is available inside a plugin's `register(ctx)` function. | Register an image-generation backend | `ctx.register_image_gen_provider(provider)` — see [Image Generation Provider Plugins](/developer-guide/image-gen-provider-plugin) | | Register a video-generation backend | `ctx.register_video_gen_provider(provider)` — see [Video Generation Provider Plugins](/developer-guide/video-gen-provider-plugin) | | Register a context-compression engine | `ctx.register_context_engine(engine)` — see [Context Engine Plugins](/developer-guide/context-engine-plugin) | +| Route human approval prompts | `ctx.register_approval_transport(name, present_fn)` — see [Approval transports](#approval-transports) | | Register a memory backend | Subclass `MemoryProvider` in `plugins/memory//__init__.py` — see [Memory Provider Plugins](/developer-guide/memory-provider-plugin) (uses a separate discovery system) | | Run a host-owned LLM call | `ctx.llm.complete(...)` / `ctx.llm.complete_structured(...)` — borrow the user's active model + auth for a one-shot completion with optional JSON schema validation. See [Plugin LLM Access](/developer-guide/plugin-llm-access) | | Register an inference backend (LLM provider) | `register_provider(ProviderProfile(...))` in `plugins/model-providers//__init__.py` — see [Model Provider Plugins](/developer-guide/model-provider-plugin) (uses a separate discovery system) | @@ -182,6 +183,60 @@ Several categories of plugin bypass `plugins.enabled` — they're part of Hermes In short: **bundled "always-works" infrastructure loads automatically; third-party general plugins are opt-in.** The `plugins.enabled` allow-list is the gate specifically for arbitrary code a user drops into `~/.hermes/plugins/`. +### Approval transports + +An approval transport changes **where a human sees and answers** an existing +Hermes tool-approval request. It does not decide whether a command needs +approval and it is not an authorization-policy API. + +```python +def present(request): + # Deliver request.command and request.description to your UI, wait for + # its authenticated human response, then return a request-bound decision. + choice = send_to_my_ui_and_wait(request) # once/session/always/deny + return request.respond(choice) + + +def register(ctx): + ctx.register_approval_transport("my-ui", present) +``` + +`present` may be synchronous or async. Hermes runs it on a bounded worker and +enforces the canonical `approvals.timeout` even if the plugin does not. The +request is immutable and contains redacted display text, its originating +surface, the host timeout, allowed choices, and an opaque request ID/digest. +Return the result of +`request.respond(choice)`; unbound dictionaries and stale or changed request +IDs/digests are rejected. A plugin cannot return a scope that the host did not +offer (for example, `always` on a once-only request). + +Registration alone does nothing. Enabling the plugin and explicitly selecting +its transport are separate consent steps: + +```yaml +plugins: + enabled: [my-approval-plugin] + +security: + approval: + transport: my-ui + transport_fallback: deny # default +``` + +Transport exceptions, timeouts, unavailable registrations, invalid choices, +and stale responses deny by default. To deliberately show the prompt on the +ordinary CLI/TUI/gateway/ACP surface when the selected transport fails, set +`transport_fallback: builtin`. Without that exact opt-in, Hermes never +materializes the prompt on another surface. + +Hermes still owns hardline blocks, sudo-stdin protection, user deny rules, +request binding, allowed scopes, persistence, hooks, and final authorization. +Hardline commands are blocked before any transport callback. There is +intentionally **no plugin approval policy, auto-allow callback, or required +`pre_tool_call` policy** in this interface. A future approval-policy capability +may use the plugin capability-consent model, but transport selection does not +grant it. + ### Migration for existing users When you upgrade to a version of Hermes that has opt-in plugins (config schema v21+), any user plugins already installed under `~/.hermes/plugins/` that weren't already in `plugins.disabled` are **automatically grandfathered** into `plugins.enabled`. Your existing setup keeps working. Bundled standalone plugins are NOT grandfathered — even existing users have to opt in explicitly. (Bundled platform/backend plugins never needed grandfathering because they were never gated.)