Harden execute_code security and approval guards

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

• #34497 fix(security): restore approval/sudo context in execute_code + guard entry points (salvage #34131)

  merged to main

  files: agent/tool_executor.py, gateway/run.py, tests/run_agent/test_tool_executor_contextvar_propagation.py, tests/tools/test_code_execution_windows_env.py, tests/tools/test_execute_code_approval_cluster.py, tools/approval.py, tools/code_execution_tool.py, tools/thread_context.py

  Show the diff

  • agent/tool_executor.py

    @@ -13,7 +13,6 @@
     from __future__ import annotations
     
     import concurrent.futures
    -import contextvars
     import json
     import logging
     import os
    @@ -38,12 +37,9 @@
         make_tool_result_message,
     )
     from tools.terminal_tool import (
    -    _get_approval_callback,
    -    _get_sudo_password_callback,
    -    set_approval_callback as _set_approval_callback,
    -    set_sudo_password_callback as _set_sudo_password_callback,
         get_active_env,
     )
    +from tools.thread_context import propagate_context_to_thread
     from tools.tool_result_storage import (
         maybe_persist_tool_result,
         enforce_turn_budget,
    @@ -186,14 +182,6 @@ def execute_tool_calls_concurrent(agent, assistant_message, messages: list, effe
         agent._current_tool = tool_names_str
         agent._touch_activity(f"executing {num_tools} tools concurrently: {tool_names_str}")
     
    -    # Capture CLI callbacks from the agent thread so worker threads can
    -    # register them locally.  Without this, _get_approval_callback() in
    -    # terminal_tool returns None in ThreadPoolExecutor workers, causing
    -    # the dangerous-command prompt to fall back to input() — which
    -    # deadlocks against prompt_toolkit's raw terminal mode (#13617).
    -    _parent_approval_cb = _get_approval_callback()
    -    _parent_sudo_cb = _get_sudo_password_callback()
    -
         def _run_tool(index, tool_call, function_name, function_args):
             """Worker function executed in a thread."""
             # Register this worker tid so the agent can fan out an interrupt
    @@ -220,18 +208,9 @@ def _run_tool(index, tool_call, function_name, function_args):
                 set_activity_callback(agent._touch_activity)
             except Exception:
                 pass
    -        # Propagate approval/sudo callbacks to this worker thread.
    -        # Mirrors cli.py run_agent() pattern (GHSA-qg5c-hvr5-hjgr).
    -        if _parent_approval_cb is not None:
    -            try:
    -                _set_approval_callback(_parent_approval_cb)
    -            except Exception:
    -                pass
    -        if _parent_sudo_cb is not None:
    -            try:
    -                _set_sudo_password_callback(_parent_sudo_cb)
    -            except Exception:
    -                pass
    +        # Approval/sudo callbacks (thread-local) and the agent turn's
    +        # ContextVars are propagated by propagate_context_to_thread() at the
    +        # submit site below (GHSA-qg5c-hvr5-hjgr, #13617).
             start = time.time()
             try:
                 result = agent._invoke_tool(
    @@ -261,13 +240,6 @@ def _run_tool(index, tool_call, function_name, function_args):
                 _ra()._set_interrupt(False, _worker_tid)
             except Exception:
                 pass
    -        # Clear thread-local callbacks so a recycled worker thread
    -        # doesn't hold stale references to a disposed CLI instance.
    -        try:
    -            _set_approval_callback(None)
    -            _set_sudo_password_callback(None)
    -        except Exception:
    -            pass
     
         # Start spinner for CLI mode (skip when TUI handles tool progress)
         spinner = None
    @@ -287,9 +259,12 @@ def _run_tool(index, tool_call, function_name, function_args):
                 max_workers = min(len(runnable_calls), _MAX_TOOL_WORKERS)
                 with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
                     for i, tc, name, args in runnable_calls:
    -                    # Propagate ContextVars (e.g. _approval_session_key); mirrors asyncio.to_thread.
    -                    ctx = contextvars.copy_context()
    -                    f = executor.submit(ctx.run, _run_tool, i, tc, name, args)
    +                    # Propagate the agent turn's ContextVars (e.g.
    +                    # _approval_session_key) AND thread-local approval/sudo
    +                    # callbacks into the worker thread; clears callbacks on exit.
    +                    f = executor.submit(
    +                        propagate_context_to_thread(_run_tool), i, tc, name, args
    +                    )
                         futures.append(f)
     
                     # Wait for all to complete with periodic heartbeats so the

  • gateway/run.py

    @@ -1807,7 +1807,34 @@ def __init__(self, config: Optional[GatewayConfig] = None):
                 ensure_installed(log_failures=False)
             except Exception:
                 pass  # Non-fatal — fail-open at scan time if unavailable
    -        
    +
    +        # Startup heads-up (#30882): a gateway in manual approval mode with no
    +        # automated risk assessor (tirith disabled AND no auxiliary.approval
    +        # model) can only gate dangerous commands / execute_code scripts via
    +        # live in-chat approval. With approval routing fixed, those actions now
    +        # fail closed (block) rather than silently auto-running — surface that
    +        # so operators knowingly enable tirith or configure auxiliary.approval
    +        # for unattended gateways.
    +        try:
    +            from hermes_cli.config import load_config as _load_full_config
    +            _appr_cfg = _load_full_config()
    +            _appr_mode = str(
    +                cfg_get(_appr_cfg, "approvals", "mode", default="manual") or "manual"
    +            ).strip().lower()
    +            _tirith_on = bool(cfg_get(_appr_cfg, "security", "tirith_enabled", default=True))
    +            _aux_approval = cfg_get(_appr_cfg, "auxiliary", "approval", default=None)
    +            if _appr_mode == "manual" and not _tirith_on and not _aux_approval:
    +                logger.warning(
    +                    "Gateway approvals.mode=manual with no automated risk "
    +                    "assessor (security.tirith_enabled is false and "
    +                    "auxiliary.approval is unset): dangerous commands and "
    +                    "execute_code scripts will BLOCK until a human approves "
    +                    "them in chat. Enable security.tirith_enabled or configure "
    +                    "auxiliary.approval for unattended operation."
    +                )
    +        except Exception:
    +            logger.debug("approvals.mode startup check skipped", exc_info=True)
    +
             # Initialize session database for session_search tool support
             self._session_db = None
             try:

  • tests/run_agent/test_tool_executor_contextvar_propagation.py

    @@ -197,6 +197,19 @@ def test_run_agent_concurrent_executor_wraps_submit_with_copy_context():
                 and call.args[1].id == "_run_tool"
             ):
                 tool_submits.append(("fixed", call))
    +        # Fixed (shared helper): executor.submit(
    +        #     propagate_context_to_thread(_run_tool), ...) — the helper in
    +        # tools/thread_context.py does copy_context().run(...) internally and
    +        # additionally propagates the thread-local approval/sudo callbacks.
    +        elif (
    +            isinstance(first, ast.Call)
    +            and isinstance(first.func, ast.Name)
    +            and first.func.id == "propagate_context_to_thread"
    +            and first.args
    +            and isinstance(first.args[0], ast.Name)
    +            and first.args[0].id == "_run_tool"
    +        ):
    +            tool_submits.append(("fixed", call))
     
         assert tool_submits, (
             "Could not locate `executor.submit(... _run_tool ...)` in "

  • tests/tools/test_code_execution_windows_env.py

    @@ -253,20 +253,24 @@ def test_child_can_create_socket_with_scrubbed_env(self):
     # ---------------------------------------------------------------------------
     
     def _legacy_posix_scrubber(source_env, is_passthrough):
    -    """Verbatim copy of the pre-Windows-fix inline scrubbing logic.
    -
    -    This is the oracle used by TestPosixEquivalence to prove the refactor
    -    did not change POSIX behavior.  DO NOT edit this to "match" a future
    -    production change — if _scrub_child_env's POSIX behavior legitimately
    -    needs to evolve, delete this function and adjust the equivalence test
    -    on purpose, so the churn is visible in review.
    +    """Independent oracle for TestPosixEquivalence — a from-scratch reimpl of
    +    _scrub_child_env's POSIX behavior, used to prove the production helper does
    +    what we think it does.
    +
    +    Deliberately updated for #27303 (the broad ``HERMES_`` prefix was dropped
    +    in favor of an explicit operational allowlist, and DSN/WEBHOOK were added
    +    to the secret substrings).  The original docstring said: if POSIX behavior
    +    legitimately needs to evolve, adjust this oracle on purpose so the churn is
    +    visible in review — that is what this change is.
         """
         _SAFE_ENV_PREFIXES = ("PATH", "HOME", "USER", "LANG", "LC_", "TERM",
                               "TMPDIR", "TMP", "TEMP", "SHELL", "LOGNAME",
    -                          "XDG_", "PYTHONPATH", "VIRTUAL_ENV", "CONDA",
    -                          "HERMES_")
    +                          "XDG_", "PYTHONPATH", "VIRTUAL_ENV", "CONDA")
         _SECRET_SUBSTRINGS = ("KEY", "TOKEN", "SECRET", "PASSWORD", "CREDENTIAL",
    -                          "PASSWD", "AUTH")
    +                          "PASSWD", "AUTH", "DSN", "WEBHOOK")
    +    _HERMES_CHILD_ALLOWED = frozenset({
    +        "HERMES_HOME", "HERMES_PROFILE", "HERMES_CONFIG", "HERMES_ENV",
    +    })
         out = {}
         for k, v in source_env.items():
             if is_passthrough(k):
    @@ -276,6 +280,9 @@ def _legacy_posix_scrubber(source_env, is_passthrough):
                 continue
             if any(k.startswith(p) for p in _SAFE_ENV_PREFIXES):
                 out[k] = v
    +            continue
    +        if k in _HERMES_CHILD_ALLOWED:
    +            out[k] = v
         return out
     
     
    @@ -308,13 +315,20 @@ class TestPosixEquivalence:
             "PYTHONPATH": "/opt/lib",
             "VIRTUAL_ENV": "/home/alice/.venv",
             "CONDA_PREFIX": "/opt/conda",
    -        "HERMES_HOME": "/home/alice/.hermes",
    -        "HERMES_INTERACTIVE": "1",
    +        # HERMES_* handling (#27303): only the operational allowlist passes;
    +        # every other HERMES_* is dropped (the broad prefix was removed).
    +        "HERMES_HOME": "/home/alice/.hermes",        # allowlisted → kept
    +        "HERMES_PROFILE": "default",                 # allowlisted → kept
    +        "HERMES_INTERACTIVE": "1",                   # not allowlisted → dropped
    +        "HERMES_BASE_URL": "https://api.internal",   # not allowlisted → dropped
    +        "HERMES_KANBAN_DB": "postgres://[redacted-credentials]@h/db",   # not allowlisted → dropped
             # Secret-substring blocks
             "OPENAI_API_KEY": "[redacted:secret-assignment]",
             "GITHUB_TOKEN": "[redacted:secret-assignment]",
             "AWS_SECRET_ACCESS_KEY": "yyy",
             "MY_PASSWORD": "[redacted:secret-assignment]",
    +        "SENTRY_DSN": "https://[redacted:email]/1",     # DSN substring → blocked
    +        "SLACK_WEBHOOK": "https://hooks.slack/x",    # WEBHOOK substring → blocked
             # Uncategorized — must be dropped
             "RANDOM_UNKNOWN": "drop-me",
             "DISPLAY": ":0",

  • tests/tools/test_execute_code_approval_cluster.py

    @@ -0,0 +1,349 @@
    +"""Regression tests for the execute_code approval-bypass cluster.
    +
    +Covers the canonical fix for issues #4146, #27303, #30882, #33057:
    +
    +  1. tools.thread_context.propagate_context_to_thread — propagates the agent
    +     turn's ContextVars AND thread-local approval/sudo callbacks into worker
    +     threads, and clears the callbacks on teardown.
    +  2. Both execute_code RPC threads are wrapped with that helper (source guard).
    +  3. tools.approval.check_execute_code_guard — the entry-point guard decision
    +     matrix (isolated backends, yolo/off, cron-deny, headless-local,
    +     gateway approve/deny/timeout/missing-notify, smart mode).
    +  4. tools.code_execution_tool._scrub_child_env — broad HERMES_ prefix dropped,
    +     operational allowlist kept, DSN/WEBHOOK blocked, passthrough precedence.
    +"""
    +
    +from __future__ import annotations
    +
    +import concurrent.futures
    +import contextvars
    +import threading
    +
    +import pytest
    +
    +from tools import approval as A
    +from tools.thread_context import propagate_context_to_thread
    +
    +
    +# ---------------------------------------------------------------------------
    +# 1. Context + callback propagation helper
    +# ---------------------------------------------------------------------------
    +
    +def test_helper_propagates_contextvar_and_approval_callback():
    +    from tools import terminal_tool as TT
    +
    +    probe: contextvars.ContextVar[str] = contextvars.ContextVar(
    +        "cluster_probe", default="unset"
    +    )
    +    probe.set("parent-value")
    +    sentinel = object()
    +    TT.set_approval_callback(sentinel)
    +    try:
    +        seen: dict = {}
    +
    +        def worker():
    +            seen["probe"] = probe.get()
    +            seen["cb"] = TT._get_approval_callback()
    +
    +        t = threading.Thread(target=propagate_context_to_thread(worker))
    +        t.start()
    +        t.join(timeout=5)
    +
    +        assert seen["probe"] == "parent-value"  # ContextVar propagated
    +        assert seen["cb"] is sentinel            # thread-local callback propagated
    +    finally:
    +        TT.set_approval_callback(None)
    +
    +
    +def test_helper_clears_callbacks_on_teardown():
    +    """A recycled worker thread must not retain the propagated callback after
    +    the wrapped target finishes (mirrors the GHSA-qg5c-hvr5-hjgr teardown)."""
    +    from tools import terminal_tool as TT
    +
    +    sentinel = object()
    +    TT.set_approval_callback(sentinel)
    +    try:
    +        seen: dict = {}
    +
    +        def first():
    +            seen["during"] = TT._get_approval_callback()
    +
    +        def second():  # NOT wrapped — runs on the same recycled worker thread
    +            seen["after"] = TT._get_approval_callback()
    +
    +        with concurrent.futures.ThreadPoolExecutor(max_workers=1) as ex:
    +            ex.submit(propagate_context_to_thread(first)).result(timeout=5)
    +            ex.submit(second).result(timeout=5)
    +
    +        assert seen["during"] is sentinel  # installed for the wrapped target
    +        assert seen["after"] is None       # cleared on teardown
    +    finally:
    +        TT.set_approval_callback(None)
    +
    +
    +def test_both_rpc_threads_use_propagation_helper():
    +    """Source guard: both execute_code RPC threads must wrap their target with
    +    propagate_context_to_thread, or the gateway approval bypass (#33057)
    +    silently returns."""
    +    import inspect
    +    import tools.code_execution_tool as cet
    +
    +    src = inspect.getsource(cet)
    +    assert "propagate_context_to_thread(_rpc_server_loop)" in src, (
    +        "local UDS RPC server thread is not wrapped with "
    +        "propagate_context_to_thread — gateway approval routing will be lost."
    +    )
    +    assert "propagate_context_to_thread(_rpc_poll_loop)" in src, (
    +        "remote file-RPC poll thread is not wrapped with "
    +        "propagate_context_to_thread — gateway approval routing will be lost."
    +    )
    +
    +
    +# ---------------------------------------------------------------------------
    +# 3. check_execute_code_guard decision matrix
    +# ---------------------------------------------------------------------------
    +
    [redacted:email]
    +def gw_session(monkeypatch):
    +    """A clean gateway session: HERMES_GATEWAY_SESSION set, a bound session
    +    key, and isolated gateway queues/callbacks. Yields the session_key."""
    +    monkeypatch.setenv("HERMES_GATEWAY_SESSION", "1")
    +    monkeypatch.delenv("HERMES_INTERACTIVE", raising=False)
    +    monkeypatch.delenv("HERMES_CRON_SESSION", raising=False)
    +    monkeypatch.delenv("HERMES_EXEC_ASK", raising=False)
    +    # Force manual mode regardless of host config.
    +    monkeypatch.setattr(A, "_get_approval_mode", lambda: "manual")
    +
    +    session_key = "cluster-test-session"
    +    token = [redacted:secret-assignment]
    +    with A._lock:
    +        A._gateway_queues.pop(session_key, None)
    +        A._gateway_notify_cbs.pop(session_key, None)
    +    try:
    +        yield session_key
    +    finally:
    +        A.reset_current_session_key(token)
    +        with A._lock:
    +            A._gateway_queues.pop(session_key, None)
    +            A._gateway_notify_cbs.pop(session_key, None)
    +
    +
    +def _register_resolver(session_key: str, result):
    +    """Register a gateway notify callback that immediately resolves the most
    +    recent queued approval entry with *result* (simulating a user response)."""
    +    def cb(_approval_data):
    +        with A._lock:
    +            entries = A._gateway_queues.get(session_key, [])
    +            if entries:
    +                entry = entries[-1]
    +                entry.result = result
    +                entry.event.set()
    +    with A._lock:
    +        A._gateway_notify_cbs[session_key] = cb
    +
    +
    +def test_guard_isolated_backend_approved():
    +    # Container backends already sandbox the child — no-op approve.
    +    assert A.check_execute_code_guard("import os", "docker")["approved"] is True
    +
    +
    +def test_guard_headless_local_approved(monkeypatch):
    +    # Documented #30882 limitation: no approval surface → preserve auto-run.
    +    monkeypatch.delenv("HERMES_GATEWAY_SESSION", raising=False)
    +    monkeypatch.delenv("HERMES_INTERACTIVE", raising=False)
    +    monkeypatch.delenv("HERMES_CRON_SESSION", raising=False)
    +    monkeypatch.delenv("HERMES_EXEC_ASK", raising=False)
    +    monkeypatch.setattr(A, "_get_approval_mode", lambda: "manual")
    +    assert A.check_execute_code_guard("import os", "local")["approved"] is True
    +
    +
    +def test_guard_cron_deny_blocks(monkeypatch):
    +    monkeypatch.setenv("HERMES_CRON_SESSION", "1")
    +    monkeypatch.delenv("HERMES_GATEWAY_SESSION", raising=False)
    +    monkeypatch.setattr(A, "_get_approval_mode", lambda: "manual")
    +    monkeypatch.setattr(A, "_get_cron_approval_mode", lambda: "deny")
    +    res = A.check_execute_code_guard("import os", "local")
    +    assert res["approved"] is False
    +    assert res["outcome"] == "blocked"
    +
    +
    +def test_guard_gateway_user_approves_is_one_shot(gw_session):
    +    _register_resolver(gw_session, "once")
    +    res = A.check_execute_code_guard("import os; print(1)", "local")
    +    assert res["approved"] is True
    +    assert res.get("user_approved") is True
    +    # One-shot: approval must NOT persist to future scripts.
    +    assert A.is_approved(gw_session, "execute_code") is False
    +
    +
    +def test_guard_gateway_user_denies_blocks(gw_session):
    +    _register_resolver(gw_session, "deny")
    +    res = A.check_execute_code_guard("import os", "local")
    +    assert res["approved"] is False
    +    assert res["outcome"] == "denied"
    +    assert res["user_consent"] is False
    +
    +
    +def test_guard_gateway_timeout_blocks(gw_session, monkeypatch):
    +    # Register a callback that never resolves; force an immediate timeout.
    +    with A._lock:
    +        A._gateway_notify_cbs[gw_session] = lambda _d: None
    +    monkeypatch.setattr(A, "_get_approval_config", lambda: {"gateway_timeout": 0})
    +    res = A.check_execute_code_guard("import os", "local")
    +    assert res["approved"] is False
    +    assert res["outcome"] == "timeout"
    +
    +
    +def test_guard_gateway_missing_notify_is_pending(gw_session):
    +    # No notify callback registered → backward-compat pending approval.
    +    res = A.check_execute_code_guard("import os", "local")
    +    assert res["approved"] is False
    +    assert res["status"] == "pending_approval"
    +
    +
    +def test_guard_smart_mode(gw_session, monkeypatch):
    +    monkeypatch.setattr(A, "_get_approval_mode", lambda: "smart")
    +
    +    monkeypatch.setattr(A, "_smart_approve", lambda c, d: "approve")
    +    res = A.check_execute_code_guard("import os", "local")
    +    assert res["approved"] is True and res.get("smart_approved") is True
    +
    +    monkeypatch.setattr(A, "_smart_approve", lambda c, d: "deny")
    +    res = A.check_execute_code_guard("import os", "local")
    +    assert res["approved"] is False and res.get("smart_denied") is True
    +
    +    # escalate → falls through to manual gateway approval
    +    monkeypatch.setattr(A, "_smart_approve", lambda c, d: "escalate")
    +    _register_resolver(gw_session, "once")
    +    res = A.check_execute_code_guard("import os", "local")
    +    assert res["approved"] is True
    +
    +
    +def test_guard_session_yolo_bypasses(gw_session):
    +    A.enable_session_yolo(gw_session)
    +    try:
    +        # Even with a denier registered, yolo short-circuits before the prompt.
    +        _register_resolver(gw_session, "deny")
    +        assert A.check_execute_code_guard("import os", "local")["approved"] is True
    +    finally:
    +        A.disable_session_yolo(gw_session)
    +
    +
    +# ---------------------------------------------------------------------------
    +# 4. Env scrubbing (#27303)
    +# ---------------------------------------------------------------------------
    +
    +def test_env_scrub_hermes_allowlist_and_secret_blocks():
    +    from tools.code_execution_tool import _scrub_child_env
    +
    +    env = {
    +        # operational allowlist → kept
    +        "HERMES_HOME": "/h", "HERMES_PROFILE": "p",
    +        "HERMES_CONFIG": "/c.yaml", "HERMES_ENV": "/e",
    +        # other HERMES_* → dropped (broad prefix removed)
    +        "HERMES_BASE_URL": "https://x", "HERMES_INTERACTIVE": "1",
    +        "HERMES_KANBAN_DB": "postgres://[redacted-credentials]@h/db",
    +        # secret substrings (incl. new DSN/WEBHOOK) → dropped
    +        "SENTRY_DSN": "https://[redacted:email]/1", "SLACK_WEBHOOK": "https://h/x",
    +        "OPENAI_API_KEY": "sk", "GITHUB_TOKEN": "ghp",
    +        # safe prefix → kept; uncategorized → dropped
    +        "PATH": "/usr/bin", "RANDOM_X": "y",
    +    }
    +    out = _scrub_child_env(env, is_passthrough=lambda _: False, is_windows=False)
    +
    +    for kept in ("HERMES_HOME", "HERMES_PROFILE", "HERMES_CONFIG", "HERMES_ENV", "PATH"):
    +        assert kept in out, f"{kept} should be kept"
    +    for dropped in (
    +        "HERMES_BASE_URL", "HERMES_INTERACTIVE", "HERMES_KANBAN_DB",
    +        "SENTRY_DSN", "SLACK_WEBHOOK", "OPENAI_API_KEY", "GITHUB_TOKEN",
    +        "RANDOM_X",
    +    ):
    +        assert dropped not in out, f"{dropped} should be dropped"
    +
    +
    +def test_env_scrub_passthrough_overrides_secret_block():
    +    """A skill/config-declared passthrough var is an explicit user opt-in and
    +    passes even if it matches a secret substring (precedence is intentional)."""
    +    from tools.code_execution_tool import _scrub_child_env
    +
    +    env = {"MY_SERVICE_DSN": "value"}
    +    out = _scrub_child_env(env, is_passthrough=lambda k: k == "MY_SERVICE_DSN",
    +                           is_windows=False)
    +    assert out.get("MY_SERVICE_DSN") == "value"
    +
    +
    +# ---------------------------------------------------------------------------
    +# 5. File-tool sensitive-path refusal (security B1)
    +# ---------------------------------------------------------------------------
    +
    +def test_execute_code_entry_blocks_before_spawn_when_guard_denies(monkeypatch, tmp_path):
    +    """Behavioral wiring test: execute_code() consults the entry guard and, on
    +    denial, returns the block message WITHOUT spawning the child — proven by a
    +    marker file the script would create that never appears."""
    +    import json
    +
    +    import tools.code_execution_tool as cet
    +    from tools import terminal_tool as TT
    +
    +    marker = tmp_path / "child-ran.marker"
    +    monkeypatch.setenv("HERMES_CRON_SESSION", "1")
    +    monkeypatch.delenv("HERMES_GATEWAY_SESSION", raising=False)
    +    monkeypatch.delenv("HERMES_INTERACTIVE", raising=False)
    +    monkeypatch.setattr(A, "_get_approval_mode", lambda: "manual")
    +    monkeypatch.setattr(A, "_get_cron_approval_mode", lambda: "deny")
    +    monkeypatch.setattr(TT, "_get_env_config", lambda: {"env_type": "local"})
    +
    +    result = json.loads(
    +        cet.execute_code(f"open({str(marker)!r}, 'w').close()", task_id="cluster-t")
    +    )
    +    assert result["status"] == "error"
    +    assert "BLOCKED" in result["error"]
    +    assert not marker.exists()  # guard denied before the child was spawned
    +
    +
    +# ---------------------------------------------------------------------------
    +# 6. Env-scrub diagnosability mitigation (#27303 follow-up)
    +# ---------------------------------------------------------------------------
    +
    +def test_env_scrub_logs_dropped_hermes_vars(caplog):
    +    """Dropping a non-allowlisted, non-secret HERMES_* var must be diagnosable:
    +    the scrub emits a one-shot debug log naming the dropped vars and pointing at
    +    the env_passthrough opt-in, so the silent behavior change (#27303) doesn't
    +    leave users guessing why a sandbox script sees an unset HERMES_* var."""
    +    import logging
    +
    +    from tools.code_execution_tool import _scrub_child_env
    +
    +    env = {
    +        "HERMES_HOME": "/h",          # allowlisted → kept, not logged
    +        "HERMES_BASE_URL": "https://x",   # dropped → logged
    +        "HERMES_KANBAN_DB": "postgres://[redacted-credentials]@h/db",  # dropped → logged
    +        "HERMES_API_KEY": "sk",       # secret → dropped silently (not logged)
    +        "PATH": "/usr/bin",           # safe prefix → kept
    +    }
    +    with caplog.at_level(logging.DEBUG, logger="tools.code_execution_tool"):
    +        out = _scrub_child_env(env, is_passthrough=lambda _: False, is_windows=False)
    +
    +    assert "HERMES_HOME" in out and "PATH" in out
    +    assert "HERMES_BASE_URL" not in out and "HERMES_KANBAN_DB" not in out
    +
    +    msgs = "\n".join(r.getMessage() for r in caplog.records)
    +    assert "HERMES_BASE_URL" in msgs and "HERMES_KANBAN_DB" in msgs
    +    assert "env_passthrough" in msgs
    +    # Secret vars are dropped but must NOT be named in the diagnostic log.
    +    assert "HERMES_API_KEY" not in msgs
    +
    +
    +def test_env_scrub_no_log_when_nothing_dropped(caplog):
    +    """No diagnostic noise when there are no dropped HERMES_* vars."""
    +    import logging
    +
    +    from tools.code_execution_tool import _scrub_child_env
    +
    +    with caplog.at_level(logging.DEBUG, logger="tools.code_execution_tool"):
    +        _scrub_child_env(
    +            {"HERMES_HOME": "/h", "PATH": "/usr/bin"},
    +            is_passthrough=lambda _: False,
    +            is_windows=False,
    +        )
    +    assert "dropped" not in "\n".join(r.getMessage() for r in caplog.records)

  • tools/approval.py

    @@ -1057,6 +1057,107 @@ def _format_tirith_description(tirith_result: dict) -> str:
         return "Security scan — " + "; ".join(parts)
     
     
    +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
    +    thread until the request is resolved or the gateway approval timeout
    +    elapses — firing pre/post approval hooks and cleaning up the queue entry.
    +
    +    Shared by the terminal command guard (``check_all_command_guards``) and
    +    the execute_code guard (``check_execute_code_guard``) so the fiddly
    +    heartbeat-polling wait loop lives in one place.
    +
    +    Returns ``{"resolved": bool, "choice": str|None}`` on completion, or
    +    ``{"resolved": False, "choice": None, "notify_failed": True}`` if the
    +    notify callback raised.  Persistence of an approved choice and building
    +    the final tool-facing result dict remain the caller's responsibility.
    +    """
    +    command = approval_data.get("command", "")
    +    description = approval_data.get("description", "")
    +    primary_key = approval_data.get("pattern_key", "")
    +    all_keys = approval_data.get("pattern_keys", [primary_key])
    +
    +    entry = _ApprovalEntry(approval_data)
    +    with _lock:
    +        _gateway_queues.setdefault(session_key, []).append(entry)
    +
    +    def _drop_entry() -> None:
    +        with _lock:
    +            queue = _gateway_queues.get(session_key, [])
    +            if entry in queue:
    +                queue.remove(entry)
    +            if not queue:
    +                _gateway_queues.pop(session_key, None)
    +
    +    # Notify plugins that an approval is being requested. Fires before the
    +    # gateway notify callback so observers get the event in real time.
    +    _fire_approval_hook(
    +        "pre_approval_request",
    +        command=command,
    +        description=description,
    +        pattern_key=primary_key,
    +        pattern_keys=list(all_keys),
    +        session_key=session_key,
    +        surface=surface,
    +    )
    +
    +    # Notify the user (bridges sync agent thread → async gateway)
    +    try:
    +        notify_cb(approval_data)
    +    except Exception as exc:
    +        logger.warning("Gateway approval notify failed: %s", exc)
    +        _drop_entry()
    +        return {"resolved": False, "choice": None, "notify_failed": True}
    +
    +    # Block until the user responds or timeout (default 5 min). Poll in short
    +    # slices so we can fire activity heartbeats every ~10s to the agent's
    +    # inactivity tracker — otherwise the gateway watchdog kills the agent
    +    # while the user is still responding. Mirrors _wait_for_process() cadence.
    +    timeout = _get_approval_config().get("gateway_timeout", 300)
    +    try:
    +        timeout = int(timeout)
    +    except (ValueError, TypeError):
    +        timeout = 300
    +
    +    try:
    +        from tools.environments.base import touch_activity_if_due
    +    except Exception:  # pragma: no cover
    +        touch_activity_if_due = None
    +
    +    _now = time.monotonic()
    +    _deadline = _now + max(timeout, 0)
    +    _activity_state = {"last_touch": _now, "start": _now}
    +    resolved = False
    +    while True:
    +        _remaining = _deadline - time.monotonic()
    +        if _remaining <= 0:
    +            break
    +        if entry.event.wait(timeout=min(1.0, _remaining)):
    +            resolved = True
    +            break
    +        if touch_activity_if_due is not None:
    +            touch_activity_if_due(_activity_state, "waiting for user approval")
    +
    +    _drop_entry()
    +
    +    choice = entry.result
    +    # Normalize outcome for the post hook. Unresolved (timeout) and None both
    +    # mean the user never responded; report that explicitly so plugins can
    +    # distinguish timeout from explicit deny.
    +    _outcome = "timeout" if not resolved else (choice if choice else "timeout")
    +    _fire_approval_hook(
    +        "post_approval_response",
    +        command=command,
    +        description=description,
    +        pattern_key=primary_key,
    +        pattern_keys=list(all_keys),
    +        session_key=session_key,
    +        surface=surface,
    +        choice=_outcome,
    +    )
    +    return {"resolved": resolved, "choice": choice}
    +
    +
     def check_all_command_guards(command: str, env_type: str,
                                  approval_callback=None) -> dict:
         """Run all pre-exec security checks and return a single approval decision.
    @@ -1207,113 +1308,27 @@ def check_all_command_guards(command: str, env_type: str,
     
             if notify_cb is not None:
                 # --- Blocking gateway approval (queue-based) ---
    -            # Each call gets its own _ApprovalEntry so parallel subagents
    -            # and execute_code threads can block concurrently.
    +            # Block the agent thread until the user responds; the notify +
    +            # heartbeat wait loop is shared with check_execute_code_guard via
    +            # _await_gateway_decision().
                 approval_data = {
                     "command": command,
                     "pattern_key": primary_key,
                     "pattern_keys": all_keys,
                     "description": combined_desc,
                 }
    -            entry = _ApprovalEntry(approval_data)
    -            with _lock:
    -                _gateway_queues.setdefault(session_key, []).append(entry)
    -
    -            # Notify plugins that an approval is being requested. Fires before
    -            # the gateway notify callback so observers (e.g. macOS notifier
    -            # plugins, audit logs, Slack alerts) get the event in real time.
    -            _fire_approval_hook(
    -                "pre_approval_request",
    -                command=command,
    -                description=combined_desc,
    -                pattern_key=primary_key,
    -                pattern_keys=list(all_keys),
    -                session_key=session_key,
    -                surface="gateway",
    +            decision = _await_gateway_decision(
    +                session_key, notify_cb, approval_data, surface="gateway"
                 )
    -
    -            # Notify the user (bridges sync agent thread → async gateway)
    -            try:
    -                notify_cb(approval_data)
    -            except Exception as exc:
    -                logger.warning("Gateway approval notify failed: %s", exc)
    -                with _lock:
    -                    queue = _gateway_queues.get(session_key, [])
    -                    if entry in queue:
    -                        queue.remove(entry)
    -                    if not queue:
    -                        _gateway_queues.pop(session_key, None)
    +            if decision.get("notify_failed"):
                     return {
                         "approved": False,
                         "message": "BLOCKED: Failed to send approval request to user. Do NOT retry.",
                         "pattern_key": primary_key,
                         "description": combined_desc,
                     }
    -
    -            # Block until the user responds or timeout (default 5 min).
    -            # Poll in short slices so we can fire activity heartbeats every
    -            # ~10s to the agent's inactivity tracker.  Without this, the
    -            # blocking event.wait() never touches activity, and the
    -            # gateway's inactivity watchdog (agent.gateway_timeout, default
    -            # 1800s) kills the agent while the user is still responding to
    -            # the approval prompt.  Mirrors the _wait_for_process() cadence
    -            # in tools/environments/base.py.
    -            timeout = _get_approval_config().get("gateway_timeout", 300)
    -            try:
    -                timeout = int(timeout)
    -            except (ValueError, TypeError):
    -                timeout = 300
    -
    -            try:
    -                from tools.environments.base import touch_activity_if_due
    -            except Exception:  # pragma: no cover
    -                touch_activity_if_due = None
    -
    -            _now = time.monotonic()
    -            _deadline = _now + max(timeout, 0)
    -            _activity_state = {"last_touch": _now, "start": _now}
    -            resolved = False
    -            while True:
    -                _remaining = _deadline - time.monotonic()
    -                if _remaining <= 0:
    -                    break
    -                # 1s poll slice — the event is set immediately when the
    -                # user responds, so slice length only controls heartbeat
    -                # cadence, not user-visible responsiveness.
    -                if entry.event.wait(timeout=min(1.0, _remaining)):
    -                    resolved = True
    -                    break
    -                if touch_activity_if_due is not None:
    -                    touch_activity_if_due(
    -                        _activity_state, "waiting for user approval"
    -                    )
    -
    -            # Clean up this entry from the queue
    -            with _lock:
    -                queue = _gateway_queues.get(session_key, [])
    -                if entry in queue:
    -                    queue.remove(entry)
    -                if not queue:
    -                    _gateway_queues.pop(session_key, None)
    -
    -            choice = entry.result
    -            # Normalize outcome for the post hook. Unresolved (timeout) and
    -            # None both mean the user never responded; report that explicitly
    -            # so plugins can distinguish timeout from explicit deny.
    -            _outcome = (
    -                "timeout" if not resolved
    -                else (choice if choice else "timeout")
    -            )
    -            _fire_approval_hook(
    -                "post_approval_response",
    -                command=command,
    -                description=combined_desc,
    -                pattern_key=primary_key,
    -                pattern_keys=list(all_keys),
    -                session_key=session_key,
    -                surface="gateway",
    -                choice=_outcome,
    -            )
    +            resolved = decision["resolved"]
    +            choice = decision["choice"]
     
                 if not resolved or choice is None or choice == "deny":
                     # Consent contract: silence is NOT consent, and an explicit
    @@ -1437,5 +1452,173 @@ def check_all_command_guards(command: str, env_type: str,
                 "user_approved": True, "description": combined_desc}
     
     
    +def check_execute_code_guard(code: str, env_type: str) -> dict:
    +    """Approve an execute_code script before its child process is spawned.
    +
    +    execute_code runs arbitrary local Python — the script can call
    +    ``subprocess``, ``os.system``, ``ctypes``, or other process/file APIs
    +    directly, none of which pass through ``terminal()`` /
    +    ``DANGEROUS_PATTERNS``. In gateway/ask contexts we fail closed by approving
    +    the script as a whole before it runs (#30882). Returns the same dict
    +    contract as ``check_all_command_guards``.
    +
    +    Scope (documented limitation, #30882): in a purely local non-interactive
    +    non-gateway session (no TTY, not gateway, not cron-deny) this returns
    +    approved — matching the existing terminal auto-approve contract. The
    +    hardline floor still blocks catastrophic ``terminal()`` commands the script
    +    issues; running arbitrary code headlessly without any approval surface is
    +    trusted-by-config (set a gateway/ask surface or ``approvals.cron_mode`` to
    +    require approval).
    +    """
    +    pattern_key = "execute_code"
    +    description = (
    +        "execute_code script execution. The script can spawn subprocesses or "
    +        "mutate files without passing through terminal command approval; "
    +        "approval is one-shot for this run."
    +    )
    +
    +    # Isolated backends already sandbox the child — matches the container skip
    +    # in check_all_command_guards / check_dangerous_command.
    +    if env_type in {"docker", "singularity", "modal", "daytona", "vercel_sandbox"}:
    +        return {"approved": True, "message": None}
    +
    +    # --yolo or approvals.mode=off: bypass (session- or process-scoped).
    +    approval_mode = _get_approval_mode()
    +    if _YOLO_MODE_FROZEN or is_current_session_yolo_enabled() or approval_mode == "off":
    +        return {"approved": True, "message": None}
    +
    +    is_gateway = _is_gateway_approval_context()
    +    is_ask = env_var_enabled("HERMES_EXEC_ASK")
    +
    +    # Cron: no user is present to approve arbitrary code.
    +    if env_var_enabled("HERMES_CRON_SESSION"):
    +        if _get_cron_approval_mode() == "deny":
    +            return {
    +                "approved": False,
    +                "message": (
    +                    "BLOCKED: execute_code runs arbitrary local Python "
    +                    "(including subprocess calls that bypass shell-string "
    +                    "approval checks). Cron jobs run without a user present "
    +                    "to approve it. Use normal tools instead, or set "
    +                    "approvals.cron_mode: approve only if this cron profile "
    +                    "is intentionally trusted."
    +                ),
    +                "pattern_key": pattern_key,
    +                "description": description,
    +                "outcome": "blocked",
    +                "user_consent": False,
    +            }
    +        return {"approved": True, "message": None}
    +
    +    # Only gateway/ask contexts get the one-shot whole-script approval.
    +    #   * CLI interactive: the script's terminal() calls are guarded per-call
    +    #     (context now propagates into the RPC thread, #33057); a whole-script
    +    #     prompt would fire on every execute_code call.
    +    #   * Local non-interactive non-gateway: documented limitation above.
    +    if not is_gateway and not is_ask:
    +        return {"approved": True, "message": None}
    +
    +    session_key = get_current_session_key()
    +    # Built only now (past the early-return gates) so the common non-approval
    +    # paths don't pay to copy a potentially-large script into this string.
    +    command = f"execute_code <<'PY'\n{code}\nPY"
    +
    +    # Smart mode: ask the aux LLM about the whole script. An APPROVE here only
    +    # suppresses the redundant whole-script prompt; the per-call terminal()
    +    # guards (restored by context propagation) still run independently.
    +    if approval_mode == "smart":
    +        verdict = _smart_approve(command, description)
    +        if verdict == "approve":
    +            logger.debug("Smart approval: auto-approved execute_code for session %s",
    +                         session_key)
    +            return {"approved": True, "message": None,
    +                    "smart_approved": True, "description": description}
    +        if verdict == "deny":
    +            return {
    +                "approved": False,
    +                "message": ("BLOCKED by smart approval: execute_code script "
    +                            "execution was assessed as genuinely dangerous. "
    +                            "Do NOT retry."),
    +                "smart_denied": True,
    +                "pattern_key": pattern_key,
    +                "description": description,
    +                "outcome": "denied",
    +                "user_consent": False,
    +            }
    +        # verdict == "escalate" → fall through to manual approval
    +
    +    notify_cb = None
    +    with _lock:
    +        notify_cb = _gateway_notify_cbs.get(session_key)
    +
    +    if notify_cb is None:
    +        # No gateway callback registered (e.g. ask-mode without a notifier):
    +        # surface a pending approval for backward compatibility.
    +        submit_pending(session_key, {
    +            "command": command,
    +            "pattern_key": pattern_key,
    +            "pattern_keys": [pattern_key],
    +            "description": description,
    +        })
    +        return {
    +            "approved": False,
    +            "pattern_key": pattern_key,
    +            "status": "pending_approval",
    +            "approval_pending": True,
    +            "command": command,
    +            "description": description,
    +            "message": (
    +                f"⚠️ {description}. Asking the user for approval.\n\n"
    +                f"**Code:**\n```python\n{code}\n```"
    +            ),
    +        }
    +
    +    approval_data = {
    +        "command": command,
    +        "pattern_key": pattern_key,
    +        "pattern_keys": [pattern_key],
    +        "description": description,
    +    }
    +    decision = _await_gateway_decision(
    +        session_key, notify_cb, approval_data, surface="gateway"
    +    )
    +    if decision.get("notify_failed"):
    +        return {
    +            "approved": False,
    +            "message": ("BLOCKED: Failed to send execute_code approval request "
    +                        "to user. Do NOT retry."),
    +            "pattern_key": pattern_key,
    +            "description": description,
    +            "outcome": "notify_failed",
    +            "user_consent": False,
    +        }
    +
    +    resolved = decision["resolved"]
    +    choice = decision["choice"]
    +
    +    if not resolved or choice is None or choice == "deny":
    +        reason = "timed out without user response" if not resolved else "denied by user"
    +        addendum = " Silence is not consent." if not resolved else ""
    +        return {
    +            "approved": False,
    +            "message": (
    +                f"BLOCKED: execute_code script {reason}. The user has NOT "
    +                f"consented to running this code. Do NOT retry, do NOT rephrase "
    +                f"the script, and do NOT attempt the same outcome via a "
    +                f"different tool.{addendum}"
    +            ),
    +            "pattern_key": pattern_key,
    +            "description": description,
    +            "outcome": "timeout" if not resolved else "denied",
    +            "user_consent": False,
    +        }
    +
    +    # Approved — one-shot only. Deliberately NO approve_session/approve_permanent:
    +    # each execute_code script is distinct arbitrary code, so approval never
    +    # persists to future scripts.
    +    return {"approved": True, "message": None,
    +            "user_approved": True, "description": description}
    +
    +
     # Load permanent allowlist from config on module import
     load_permanent_allowlist()

  • tools/code_execution_tool.py

    @@ -46,6 +46,8 @@
     _IS_WINDOWS = platform.system() == "Windows"
     from typing import Any, Dict, List, Optional
     
    +from tools.thread_context import propagate_context_to_thread
    +
     # Availability gate.  On Windows we fall back to loopback TCP for the
     # sandbox RPC transport (AF_UNIX is unreliable on Windows Python) — see
     # ``_use_tcp_rpc`` in ``_execute_local`` below.  That makes execute_code
    @@ -74,13 +76,30 @@
     
     # Environment variable scrubbing rules (shared between the local + remote
     # backends).  Secret-substring block is applied first; anything left must
    -# match either a safe prefix or, on Windows, an OS-essential name.
    +# match a safe prefix, the operational HERMES_ allowlist, or (on Windows) an
    +# OS-essential name.
    +#
    +# NB: the broad "HERMES_" prefix was deliberately removed (#27303) — it leaked
    +# HERMES_*-named config that lacks a secret substring (e.g. HERMES_BASE_URL,
    +# HERMES_KANBAN_DB, HERMES_*_WEBHOOK).  The child only needs the few
    +# location/profile vars in _HERMES_CHILD_ALLOWED below; HERMES_RPC_SOCKET /
    +# HERMES_RPC_DIR / TZ / HOME are injected explicitly after scrubbing.
     _SAFE_ENV_PREFIXES = ("PATH", "HOME", "USER", "LANG", "LC_", "TERM",
                           "TMPDIR", "TMP", "TEMP", "SHELL", "LOGNAME",
    -                      "XDG_", "PYTHONPATH", "VIRTUAL_ENV", "CONDA",
    -                      "HERMES_")
    +                      "XDG_", "PYTHONPATH", "VIRTUAL_ENV", "CONDA")
     _SECRET_SUBSTRINGS = ("KEY", "TOKEN", "SECRET", "PASSWORD", "CREDENTIAL",
    -                      "PASSWD", "AUTH")
    +                      "PASSWD", "AUTH", "DSN", "WEBHOOK")
    +
    +# Operational HERMES_* vars the child legitimately needs by exact name — these
    +# are non-secret runtime-location flags (the same set hermes_cli treats as the
    +# runtime location) that repo-root modules a sandbox script imports may read at
    +# import time.  None match _SECRET_SUBSTRINGS.
    +_HERMES_CHILD_ALLOWED = frozenset({
    +    "HERMES_HOME",
    +    "HERMES_PROFILE",
    +    "HERMES_CONFIG",
    +    "HERMES_ENV",
    +})
     
     # Windows-only: a handful of variables are required by the OS/CRT itself.
     # Without them, even stdlib calls like ``socket.socket()`` fail with
    @@ -119,9 +138,10 @@ def _scrub_child_env(source_env, is_passthrough=None, is_windows=None):
     
         Rules (order matters):
           1. Passthrough vars (skill- or config-declared) always pass.
    -      2. Secret-substring names (KEY/TOKEN/etc.) are blocked.
    +      2. Secret-substring names (KEY/TOKEN/DSN/WEBHOOK/etc.) are blocked.
           3. Names matching a safe prefix pass.
    -      4. On Windows, a small OS-essential allowlist passes by exact name
    +      4. Operational HERMES_* vars (_HERMES_CHILD_ALLOWED) pass by exact name.
    +      5. On Windows, a small OS-essential allowlist passes by exact name
              — without these the child can't even create a socket or spawn a
              subprocess.
     
    @@ -138,6 +158,14 @@ def _scrub_child_env(source_env, is_passthrough=None, is_windows=None):
             is_windows = _IS_WINDOWS
     
         scrubbed = {}
    +    # Non-secret HERMES_* vars dropped by the tightened allowlist (#27303). The
    +    # broad "HERMES_" prefix used to pass these through; now only the
    +    # operational set does. The drop is intentional (those vars can carry
    +    # config like HERMES_KANBAN_DB / HERMES_BASE_URL), but a sandbox script
    +    # that imports a repo module reading one at import time would otherwise see
    +    # it silently unset. Surface the drop once so the behavior change is
    +    # diagnosable and points at the env_passthrough opt-in escape hatch.
    +    _dropped_hermes = []
         for k, v in source_env.items():
             if is_passthrough(k):
                 scrubbed[k] = v
    @@ -147,8 +175,25 @@ def _scrub_child_env(source_env, is_passthrough=None, is_windows=None):
             if any(k.startswith(p) for p in _SAFE_ENV_PREFIXES):
                 scrubbed[k] = v
                 continue
    +        if k in _HERMES_CHILD_ALLOWED:
    +            scrubbed[k] = v
    +            continue
             if is_windows and k.upper() in _WINDOWS_ESSENTIAL_ENV_VARS:
                 scrubbed[k] = v
    +            continue
    +        if k.startswith("HERMES_"):
    +            # Non-secret (secrets were already dropped above) and not in any
    +            # allowlist — a deliberately-dropped HERMES_* var.
    +            _dropped_hermes.append(k)
    +    if _dropped_hermes:
    +        logger.debug(
    +            "execute_code: dropped %d non-allowlisted HERMES_* var(s) from the "
    +            "sandbox child env (%s). This is intentional hardening (#27303); if "
    +            "a sandbox script legitimately needs one, declare it via "
    +            "env_passthrough in the skill/config so it passes by explicit opt-in.",
    +            len(_dropped_hermes),
    +            ", ".join(sorted(_dropped_hermes)),
    +        )
         return scrubbed
     
     
    @@ -887,9 +932,11 @@ def _execute_remote(
             _ship_file_to_remote(env, f"{sandbox_dir}/hermes_tools.py", tools_src)
             _ship_file_to_remote(env, f"{sandbox_dir}/script.py", code)
     
    -        # Start RPC polling thread
    +        # Wrapped so the thread inherits the turn's approval context + callbacks
    +        # (see tools.thread_context) — else sandbox RPC tool calls lose approval
    +        # routing (#33057).
             rpc_thread = threading.Thread(
    -            target=_rpc_poll_loop,
    +            target=propagate_context_to_thread(_rpc_poll_loop),
                 args=(
                     env, f"{sandbox_dir}/rpc", effective_task_id,
                     tool_call_log, tool_call_counter, max_tool_calls,
    @@ -1049,6 +1096,21 @@ def execute_code(
         # Dispatch: remote backends use file-based RPC, local uses UDS
         from tools.terminal_tool import _get_env_config
         env_type = _get_env_config()["env_type"]
    +
    +    # execute_code runs arbitrary Python (subprocess/os.system/...) that never
    +    # passes through terminal()/DANGEROUS_PATTERNS, so guard the whole script
    +    # here before either dispatch path spawns it. Runs synchronously in the
    +    # caller (tool-executor) thread, which holds the session context (#30882).
    +    from tools.approval import check_execute_code_guard
    +    _guard = check_execute_code_guard(code, env_type)
    +    if not _guard.get("approved", False):
    +        return json.dumps({
    +            "status": "error",
    +            "error": _guard.get("message") or "execute_code blocked by approval guard.",
    +            "tool_calls_made": 0,
    +            "duration_seconds": 0,
    +        }, ensure_ascii=False)
    +
         if env_type != "local":
             return _execute_remote(code, task_id, enabled_tools)
     
    @@ -1135,8 +1197,11 @@ def execute_code(
                 os.chmod(sock_path, 0o600)
             server_sock.listen(1)
     
    +        # Wrapped so the thread inherits the turn's approval context + callbacks
    +        # (see tools.thread_context) — else gateway sandbox tool calls silently
    +        # auto-approve dangerous commands (#33057, #30882).
             rpc_thread = threading.Thread(
    -            target=_rpc_server_loop,
    +            target=propagate_context_to_thread(_rpc_server_loop),
                 args=(
                     server_sock, task_id, tool_call_log,
                     tool_call_counter, max_tool_calls, sandbox_tools,

  • tools/thread_context.py

    @@ -0,0 +1,120 @@
    +#!/usr/bin/env python3
    +"""Propagate agent-turn context into worker threads that dispatch Hermes tools.
    +
    +A bare ``threading.Thread`` / ``ThreadPoolExecutor`` worker starts with an
    +empty ``contextvars.Context`` and no thread-local approval/sudo callbacks.
    +Tool dispatch inside such a thread therefore silently loses:
    +
    +  * the approval *session/platform* ContextVars (``tools.approval`` /
    +    ``gateway.session_context``) — so gateway sessions fall into
    +    ``check_dangerous_command``'s non-interactive auto-approve branch and
    +    dangerous commands run without prompting (#33057, #30882);
    +  * the thread-local CLI approval/sudo callbacks (``tools.terminal_tool``) —
    +    so ``prompt_dangerous_approval`` cannot reach the user
    +    (GHSA-qg5c-hvr5-hjgr, #15216).
    +
    +This helper factors out that capture/install/clear lifecycle so the several
    +places that fan tool dispatch onto worker threads (``agent.tool_executor`` and
    +the ``execute_code`` RPC threads) share one audited implementation instead of
    +divergent copies.
    +
    +Usage — call :func:`propagate_context_to_thread` **on the parent thread**
    +(it snapshots the parent's ContextVars and callbacks at call time) and use the
    +returned callable as the worker's target::
    +
    +    t = threading.Thread(target=propagate_context_to_thread(loop_fn), args=(...))
    +    # or
    +    executor.submit(propagate_context_to_thread(worker_fn), *args)
    +
    +Approval/sudo callbacks are installed for the worker's lifetime and **always
    +cleared on exit**, so a recycled thread never holds a stale reference to a
    +disposed CLI instance.
    +"""
    +
    +from __future__ import annotations
    +
    +import contextvars
    +import logging
    +from typing import Callable
    +
    +logger = logging.getLogger(__name__)
    +
    +
    +def _callback_api():
    +    """Resolve the terminal_tool callback getters/setters.
    +
    +    Imported lazily: ``tools.terminal_tool`` imports ``tools.approval`` at
    +    module load, so a top-level import here would risk an import cycle for
    +    callers that live in ``tools.approval``.
    +    """
    +    from tools.terminal_tool import (
    +        _get_approval_callback,
    +        _get_sudo_password_callback,
    +        set_approval_callback,
    +        set_sudo_password_callback,
    +    )
    +    return (
    +        _get_approval_callback,
    +        _get_sudo_password_callback,
    +        set_approval_callback,
    +        set_sudo_password_callback,
    +    )
    +
    +
    +def propagate_context_to_thread(target: Callable) -> Callable:
    +    """Wrap *target* for execution on a worker thread with the *current*
    +    thread's ContextVars and approval/sudo callbacks propagated.
    +
    +    Call this on the parent thread; pass the returned callable as the
    +    thread/executor target.  The returned callable forwards its positional
    +    and keyword arguments to *target* and returns its result.
    +
    +    Fail-closed: if callback installation raises, the callbacks are left
    +    unset (``None``).  That is the safe outcome — ``prompt_dangerous_approval``
    +    denies dangerous commands when no callback is registered in an interactive
    +    context, and the gateway approval queue blocks when its notify callback is
    +    absent.
    +    """
    +    ctx = contextvars.copy_context()
    +    parent_approval_cb = parent_sudo_cb = None
    +    setters = None
    +    try:
    +        get_approval, get_sudo, set_approval, set_sudo = _callback_api()
    +        parent_approval_cb = get_approval()
    +        parent_sudo_cb = get_sudo()
    +        setters = (set_approval, set_sudo)
    +    except Exception:
    +        logger.debug("Could not capture parent approval/sudo callbacks", exc_info=True)
    +
    +    def _runner(*args, **kwargs):
    +        def _inner():
    +            if setters is not None:
    +                set_approval, set_sudo = setters
    +                try:
    +                    if parent_approval_cb is not None:
    +                        set_approval(parent_approval_cb)
    +                    if parent_sudo_cb is not None:
    +                        set_sudo(parent_sudo_cb)
    +                except Exception:
    +                    logger.debug(
    +                        "Failed to install propagated approval/sudo callbacks; "
    +                        "dangerous-command approval will fail closed",
    +                        exc_info=True,
    +                    )
    +            try:
    +                return target(*args, **kwargs)
    +            finally:
    +                if setters is not None:
    +                    set_approval, set_sudo = setters
    +                    try:
    +                        set_approval(None)
    +                        set_sudo(None)
    +                    except Exception:
    +                        logger.debug(
    +                            "Failed to clear propagated approval/sudo callbacks",
    +                            exc_info=True,
    +                        )
    +
    +        return ctx.run(_inner)
    +
    +    return _runner