ouyangbo/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 12 Packages Projects Releases Wiki Activity

fix(plugins): harden approval transport boundaries

Browse Source
This commit is contained in:
Teknium
2026-08-08 13:22:46 -07:00
parent de56e49a7c
commit cd7c674d74
4 changed files with 77 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
hermes_cli/approval_transport.py
+8 -5
View File
@@ -150,7 +150,8 @@ def invoke_approval_transport(
logger.warning("Approval transport worker capacity exhausted")
return ApprovalTransportResult("deny", "busy")
results: queue.Queue[tuple[str, object]] = queue.Queue(maxsize=1)
results: queue.Queue[tuple[str, object, float]] = queue.Queue(maxsize=1)
deadline = time.monotonic() + max(float(timeout_seconds), 0.0)
async def _await_value(value):
return await value
@@ -160,10 +161,10 @@ def invoke_approval_transport(
value = present(request)
if inspect.isawaitable(value):
value = asyncio.run(_await_value(value))
results.put_nowait(("result", value))
results.put_nowait(("result", value, time.monotonic()))
except BaseException as exc: # fail closed even for unusual callback exits
try:
results.put_nowait(("error", exc))
results.put_nowait(("error", exc, time.monotonic()))
except queue.Full:
pass
finally:
@@ -180,7 +181,6 @@ def invoke_approval_transport(
_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)
@@ -190,7 +190,7 @@ def invoke_approval_transport(
logger.warning("Approval transport timed out for request %s", request.request_id)
return ApprovalTransportResult("deny", "timeout")
try:
kind, value = results.get(
kind, value, completed_at = results.get(
timeout=min(max(float(poll_interval), 0.001), remaining)
)
break
@@ -201,6 +201,9 @@ def invoke_approval_transport(
except Exception:
logger.debug("Approval transport poll callback failed", exc_info=True)
if completed_at > deadline:
logger.warning("Approval transport timed out for request %s", request.request_id)
return ApprovalTransportResult("deny", "timeout")
if kind == "error":
logger.warning("Approval transport failed for request %s", request.request_id)
return ApprovalTransportResult("deny", "error")
tests/hermes_cli/test_approval_transport.py
+56 -3
View File
@@ -202,6 +202,29 @@ def test_host_transport_wait_is_interruptible_and_pollable():
assert polls
def test_host_rejects_decision_completed_after_deadline(monkeypatch):
import hermes_cli.approval_transport as transport_module
main_thread = threading.get_ident()
def clock():
# The host starts at t=0 with a one-second budget. The worker reports
# completion at t=2 before the host dequeues the result. Acceptance is
# bound to completion time, not to a queue/scheduler race.
return 0.0 if threading.get_ident() == main_thread else 2.0
monkeypatch.setattr(transport_module.time, "monotonic", clock)
result = transport_module.invoke_approval_transport(
lambda request: request.respond("once"),
_request(),
timeout_seconds=1,
)
assert result.choice == "deny"
assert result.failure == "timeout"
def test_host_caps_hung_transport_workers():
from hermes_cli.approval_transport import (
_MAX_ACTIVE_TRANSPORT_WORKERS,
@@ -342,6 +365,34 @@ def test_transport_failure_denies_without_builtin_fallback(monkeypatch):
assert builtin_calls == []
def test_transport_resolution_error_does_not_log_plugin_exception(monkeypatch, caplog):
from tools import approval
monkeypatch.setattr(
approval, "_get_approval_transport_config", lambda: ("phone", None)
)
def broken_manager():
raise RuntimeError("plugin-owned-secret-value")
monkeypatch.setattr(approval, "get_plugin_manager", broken_manager)
result = approval._present_with_selected_transport(
command="rm -rf /tmp/example",
description="dangerous",
pattern_key="rm_recursive",
pattern_keys=["rm_recursive"],
session_key="session-a",
surface="cli",
allow_session=True,
allow_permanent=True,
)
assert result["choice"] == "deny"
assert result["failure"] == "unavailable"
assert "plugin-owned-secret-value" not in caplog.text
def test_redaction_failure_denies_before_transport_callback(monkeypatch):
import agent.redact
from tools import approval
@@ -353,7 +404,8 @@ def test_redaction_failure_denies_before_transport_callback(monkeypatch):
)
_configure_manual_guard(monkeypatch, approval, manager)
def redaction_failed(text):
def redaction_failed(text, *, force=False):
assert force is True
if text in {"rm -rf /tmp/example", "dangerous"}:
raise RuntimeError("redactor unavailable")
return text
@@ -448,7 +500,6 @@ def register(ctx):
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()
@@ -474,7 +525,9 @@ def register(ctx):
records = [
json.loads(line)
for line in (home / "transport-invocations.jsonl").read_text().splitlines()
for line in (home / "transport-invocations.jsonl")
.read_text(encoding="utf-8")
.splitlines()
]
assert routed["approved"] is True
assert reloaded["approved"] is True
tools/approval.py
+10 -4
View File
@@ -3668,8 +3668,13 @@ def _format_tirith_description(tirith_result: dict) -> str:
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
from hermes_cli.plugins import discover_plugins, get_plugin_manager as _get_manager
# Approval can be imported before model_tools, whose import normally
# triggers general plugin discovery. Ensure an explicitly selected
# transport is available on that first approval rather than treating the
# still-undiscovered registry as an unavailable transport.
discover_plugins()
return _get_manager()
@@ -3708,7 +3713,8 @@ def _present_with_selected_transport(
try:
registered = get_plugin_manager().get_approval_transport(name)
except Exception:
logger.warning("Could not resolve selected approval transport %r", name, exc_info=True)
# Plugin/discovery exception text may contain plugin-owned secrets.
logger.warning("Could not resolve selected approval transport %r", name)
registered = None
if registered is None:
logger.warning("Selected approval transport %r is unavailable", name)
@@ -3726,8 +3732,8 @@ def _present_with_selected_transport(
timeout_seconds = _get_approval_timeout()
request = ApprovalRequest.create(
command=redact_sensitive_text(command),
description=redact_sensitive_text(description),
command=redact_sensitive_text(command, force=True),
description=redact_sensitive_text(description, force=True),
pattern_key=pattern_key,
pattern_keys=tuple(pattern_keys),
session_key=session_key,
website/docs/user-guide/features/plugins.md
+3 -2
View File
@@ -203,8 +203,9 @@ def register(ctx):
`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.
request is immutable and contains redacted display text, its host presentation
class (`cli` or `gateway`), 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