fix(acp): wire approval callback + make it thread-local (#13525)

Two related ACP approval issues:

GHSA-96vc-wcxf-jjff — ACP's _run_agent never set HERMES_INTERACTIVE
(or any other flag recognized by tools.approval), so check_all_command_guards
took the non-interactive auto-approve path and never consulted the
ACP-supplied approval callback (conn.request_permission). Dangerous
commands executed in ACP sessions without operator approval despite
the callback being installed. Fix: set HERMES_INTERACTIVE=1 around
the agent run so check_all_command_guards routes through
prompt_dangerous_approval(approval_callback=...) — the correct shape
for ACP's per-session request_permission call. HERMES_EXEC_ASK would
have routed through the gateway-queue path instead, which requires a
notify_cb registered in _gateway_notify_cbs (not applicable to ACP).

GHSA-qg5c-hvr5-hjgr — _approval_callback and _sudo_password_callback
were module-level globals in terminal_tool. Concurrent ACP sessions
running in ThreadPoolExecutor threads each installed their own callback
into the same slot, racing. Fix: store both callbacks in threading.local()
so each thread has its own slot. CLI mode (single thread) is unaffected;
gateway mode uses a separate queue-based approval path and was never
touched.

set_approval_callback is now called INSIDE _run_agent (the executor
thread) rather than before dispatching — so the TLS write lands on the
correct thread.

Tests: 5 new in tests/acp/test_approval_isolation.py covering
thread-local isolation of both callbacks and the HERMES_INTERACTIVE
callback routing. Existing tests/acp/ (159 tests) and tests/tools/
approval-related tests continue to pass.

Fixes GHSA-96vc-wcxf-jjff
Fixes GHSA-qg5c-hvr5-hjgr

acp_adapter/server.py

@@ -4,6 +4,7 @@ from __future__ import annotations
 
 import asyncio
 import logging
+import os
 from collections import defaultdict, deque
 from concurrent.futures import ThreadPoolExecutor
 from typing import Any, Deque, Optional
@@ -554,15 +555,32 @@ class HermesACPAgent(acp.Agent):
         agent.step_callback = step_cb
         agent.message_callback = message_cb
 
-        if approval_cb:
+        # Approval callback is per-thread (thread-local, GHSA-qg5c-hvr5-hjgr).
-            try:
+        # Set it INSIDE _run_agent so the TLS write happens in the executor
-                from tools import terminal_tool as _terminal_tool
+        # thread — setting it here would write to the event-loop thread's TLS,
-                previous_approval_cb = getattr(_terminal_tool, "_approval_callback", None)
+        # not the executor's. Also set HERMES_INTERACTIVE so approval.py
-                _terminal_tool.set_approval_callback(approval_cb)
+        # takes the CLI-interactive path (which calls the registered
-            except Exception:
+        # callback via prompt_dangerous_approval) instead of the
-                logger.debug("Could not set ACP approval callback", exc_info=True)
+        # non-interactive auto-approve branch (GHSA-96vc-wcxf-jjff).
+        # ACP's conn.request_permission maps cleanly to the interactive
+        # callback shape — not the gateway-queue HERMES_EXEC_ASK path,
+        # which requires a notify_cb registered in _gateway_notify_cbs.
+        previous_approval_cb = None
+        previous_interactive = None
 
         def _run_agent() -> dict:
+            nonlocal previous_approval_cb, previous_interactive
+            if approval_cb:
+                try:
+                    from tools import terminal_tool as _terminal_tool
+                    previous_approval_cb = _terminal_tool._get_approval_callback()
+                    _terminal_tool.set_approval_callback(approval_cb)
+                except Exception:
+                    logger.debug("Could not set ACP approval callback", exc_info=True)
+            # Signal to tools.approval that we have an interactive callback
+            # and the non-interactive auto-approve path must not fire.
+            previous_interactive = os.environ.get("HERMES_INTERACTIVE")
+            os.environ["HERMES_INTERACTIVE"] = "1"
             try:
                 result = agent.run_conversation(
                     user_message=user_text,
@@ -574,6 +592,11 @@ class HermesACPAgent(acp.Agent):
                 logger.exception("Agent error in session %s", session_id)
                 return {"final_response": f"Error: {e}", "messages": state.history}
             finally:
+                # Restore HERMES_INTERACTIVE.
+                if previous_interactive is None:
+                    os.environ.pop("HERMES_INTERACTIVE", None)
+                else:
+                    os.environ["HERMES_INTERACTIVE"] = previous_interactive
                 if approval_cb:
                     try:
                         from tools import terminal_tool as _terminal_tool

tests/acp/test_approval_isolation.py

@@ -0,0 +1,170 @@
+"""Tests for GHSA-96vc-wcxf-jjff and GHSA-qg5c-hvr5-hjgr.
+
+Two related ACP approval-flow issues:
+- 96vc: ACP didn't set HERMES_EXEC_ASK, so `check_all_command_guards`
+  took the non-interactive auto-approve path and never consulted the
+  ACP-supplied callback.
+- qg5c: `_approval_callback` was a module-global in terminal_tool;
+  overlapping ACP sessions overwrote each other's callback slot.
+
+Both fixed together by:
+1. Setting HERMES_EXEC_ASK inside _run_agent (wraps the agent call).
+2. Storing the callback in thread-local state so concurrent executor
+   threads don't collide.
+"""
+
+import os
+import threading
+from unittest.mock import MagicMock
+
+import pytest
+
+
+class TestThreadLocalApprovalCallback:
+    """GHSA-qg5c-hvr5-hjgr: set_approval_callback must be per-thread so
+    concurrent ACP sessions don't stomp on each other's handlers."""
+
+    def test_set_and_get_in_same_thread(self):
+        from tools.terminal_tool import (
+            set_approval_callback,
+            _get_approval_callback,
+        )
+
+        cb1 = lambda cmd, desc: "once"  # noqa: E731
+        set_approval_callback(cb1)
+        assert _get_approval_callback() is cb1
+
+    def test_callback_not_visible_in_different_thread(self):
+        """Thread A's callback is NOT visible to Thread B."""
+        from tools.terminal_tool import (
+            set_approval_callback,
+            _get_approval_callback,
+        )
+
+        cb_a = lambda cmd, desc: "thread_a"  # noqa: E731
+        cb_b = lambda cmd, desc: "thread_b"  # noqa: E731
+
+        seen_in_a = []
+        seen_in_b = []
+
+        def thread_a():
+            set_approval_callback(cb_a)
+            # Pause so thread B has time to set its own callback
+            import time
+            time.sleep(0.05)
+            seen_in_a.append(_get_approval_callback())
+
+        def thread_b():
+            set_approval_callback(cb_b)
+            import time
+            time.sleep(0.05)
+            seen_in_b.append(_get_approval_callback())
+
+        ta = threading.Thread(target=thread_a)
+        tb = threading.Thread(target=thread_b)
+        ta.start()
+        tb.start()
+        ta.join()
+        tb.join()
+
+        # Each thread must see ONLY its own callback — not the other's
+        assert seen_in_a == [cb_a]
+        assert seen_in_b == [cb_b]
+
+    def test_main_thread_callback_not_leaked_to_worker(self):
+        """A callback set in the main thread does NOT leak into a
+        freshly-spawned worker thread."""
+        from tools.terminal_tool import (
+            set_approval_callback,
+            _get_approval_callback,
+        )
+
+        cb_main = lambda cmd, desc: "main"  # noqa: E731
+        set_approval_callback(cb_main)
+
+        worker_saw = []
+
+        def worker():
+            worker_saw.append(_get_approval_callback())
+
+        t = threading.Thread(target=worker)
+        t.start()
+        t.join()
+
+        # Worker thread has no callback set — TLS is empty for it
+        assert worker_saw == [None]
+        # Main thread still has its callback
+        assert _get_approval_callback() is cb_main
+
+    def test_sudo_password_callback_also_thread_local(self):
+        """Same protection applies to the sudo password callback."""
+        from tools.terminal_tool import (
+            set_sudo_password_callback,
+            _get_sudo_password_callback,
+        )
+
+        cb_main = lambda: "main-password"  # noqa: E731
+        set_sudo_password_callback(cb_main)
+
+        worker_saw = []
+
+        def worker():
+            worker_saw.append(_get_sudo_password_callback())
+
+        t = threading.Thread(target=worker)
+        t.start()
+        t.join()
+
+        assert worker_saw == [None]
+        assert _get_sudo_password_callback() is cb_main
+
+
+class TestAcpExecAskGate:
+    """GHSA-96vc-wcxf-jjff: ACP's _run_agent must set HERMES_INTERACTIVE so
+    that tools.approval.check_all_command_guards takes the CLI-interactive
+    path (consults the registered callback via prompt_dangerous_approval)
+    instead of the non-interactive auto-approve shortcut.
+
+    (HERMES_EXEC_ASK takes the gateway-queue path which requires a
+    notify_cb registered in _gateway_notify_cbs — not applicable to ACP,
+    which uses a direct callback shape.)"""
+
+    def test_interactive_env_var_routes_to_callback(self, monkeypatch):
+        """When HERMES_INTERACTIVE is set and an approval callback is
+        registered, a dangerous command must route through the callback."""
+        # Clean env
+        monkeypatch.delenv("HERMES_INTERACTIVE", raising=False)
+        monkeypatch.delenv("HERMES_GATEWAY_SESSION", raising=False)
+        monkeypatch.delenv("HERMES_EXEC_ASK", raising=False)
+        monkeypatch.delenv("HERMES_YOLO_MODE", raising=False)
+
+        from tools.approval import check_all_command_guards
+
+        called_with = []
+
+        def fake_cb(command, description, *, allow_permanent=True):
+            called_with.append((command, description))
+            return "once"
+
+        # Without HERMES_INTERACTIVE: takes auto-approve path, callback NOT called
+        result = check_all_command_guards(
+            "rm -rf /tmp/test-exec-ask", "local", approval_callback=fake_cb,
+        )
+        assert result["approved"] is True
+        assert called_with == [], (
+            "without HERMES_INTERACTIVE the non-interactive auto-approve "
+            "path should fire without consulting the callback"
+        )
+
+        # With HERMES_INTERACTIVE: callback IS called, approval flows through it
+        monkeypatch.setenv("HERMES_INTERACTIVE", "1")
+        called_with.clear()
+        result = check_all_command_guards(
+            "rm -rf /tmp/test-exec-ask", "local", approval_callback=fake_cb,
+        )
+        assert called_with, (
+            "with HERMES_INTERACTIVE the approval path should consult the "
+            "registered callback — this was the ACP bypass in "
+            "GHSA-96vc-wcxf-jjff"
+        )
+        assert result["approved"] is True

tools/terminal_tool.py

@@ -114,22 +114,44 @@ _cached_sudo_password: str = ""
 # Optional UI callbacks for interactive prompts. When set, these are called
 # instead of the default /dev/tty or input() readers. The CLI registers these
 # so prompts route through prompt_toolkit's event loop.
-#   _sudo_password_callback() -> str  (return password or "" to skip)
+# Callback slots used by the approval prompt and sudo password prompt
-#   _approval_callback(command, description) -> str  ("once"/"session"/"always"/"deny")
+# routines. Stored in thread-local state so overlapping ACP sessions —
-_sudo_password_callback = None
+# each running in its own ThreadPoolExecutor thread — don't stomp on
-_approval_callback = None
+# each other's callbacks. See GHSA-qg5c-hvr5-hjgr.
+#
+# CLI mode is single-threaded, so each thread (the only one) holds its
+# own callback exactly like before. Gateway mode resolves approvals via
+# the per-session queue in tools.approval, not through these callbacks,
+# so it's unaffected.
+import threading
+_callback_tls = threading.local()
+
+
+def _get_sudo_password_callback():
+    return getattr(_callback_tls, "sudo_password", None)
+
+
+def _get_approval_callback():
+    return getattr(_callback_tls, "approval", None)
 
 
 def set_sudo_password_callback(cb):
-    """Register a callback for sudo password prompts (used by CLI)."""
+    """Register a callback for sudo password prompts (used by CLI).
-    global _sudo_password_callback
+
-    _sudo_password_callback = cb
+    Per-thread scope — ACP sessions that run concurrently in a
+    ThreadPoolExecutor each have their own callback slot.
+    """
+    _callback_tls.sudo_password = cb
 
 
 def set_approval_callback(cb):
-    """Register a callback for dangerous command approval prompts (used by CLI)."""
+    """Register a callback for dangerous command approval prompts.
-    global _approval_callback
+
-    _approval_callback = cb
+    Per-thread scope — ACP sessions that run concurrently in a
+    ThreadPoolExecutor each have their own callback slot. See
+    GHSA-qg5c-hvr5-hjgr.
+    """
+    _callback_tls.approval = cb
 
 # =============================================================================
 # Dangerous Command Approval System
@@ -144,7 +166,7 @@ from tools.approval import (
 def _check_all_guards(command: str, env_type: str) -> dict:
     """Delegate to consolidated guard (tirith + dangerous cmd) with CLI callback."""
     return _check_all_guards_impl(command, env_type,
-                                  approval_callback=_approval_callback)
+                                  approval_callback=_get_approval_callback())
 
 
 # Allowlist: characters that can legitimately appear in directory paths.
@@ -219,9 +241,10 @@ def _prompt_for_sudo_password(timeout_seconds: int = 45) -> str:
     import sys
     
     # Use the registered callback when available (prompt_toolkit-compatible)
-    if _sudo_password_callback is not None:
+    _sudo_cb = _get_sudo_password_callback()
+    if _sudo_cb is not None:
         try:
-            return _sudo_password_callback() or ""
+            return _sudo_cb() or ""
         except Exception:
             return ""