From 1e4c2b42b22cb7bf5823ccf2649799d845ac449a Mon Sep 17 00:00:00 2001 From: qbit-mirror-bot Date: Thu, 2 Jul 2026 06:01:13 +0000 Subject: [PATCH] fix(approval): use per-job ContextVar for cron-session flag instead of leaking env var (#56771) --- cron/scheduler.py | 203 +++++++++++++++++++++++++- gateway/session_context.py | 63 ++++++++ tests/tools/test_cron_session_leak.py | 198 +++++++++++++++++++++++++ tools/approval.py | 24 ++- 4 files changed, 476 insertions(+), 12 deletions(-) create mode 100644 tests/tools/test_cron_session_leak.py diff --git a/cron/scheduler.py b/cron/scheduler.py index 60c7ef3d6ee7..5262fc4b978f 100644 --- a/cron/scheduler.py +++ b/cron/scheduler.py @@ -701,6 +701,102 @@ def _seed_cron_thread_session( ) +def _seed_cron_channel_session( + job: dict, + adapter, + platform_name: str, + chat_id: str, + mirror_text: str, + *, + is_dm: bool, + user_id: Optional[str], + chat_name: Optional[str] = None, +) -> bool: + """Seed the FLAT (thread_id=None) session for an ``in_channel`` cron delivery. + + The ``in_channel`` surface (D1/D2) delivers the brief flat into the channel + with no thread, so the continuation surface is the whole-channel / + whole-DM session keyed ``thread_id=None`` — the same bucket + ``reply_in_thread: false`` routes an inbound plain reply to. + + Unlike the thread path, the shipped delivery-mirror alone is NOT sufficient + here: ``mirror_to_session`` only APPENDS to a session that already EXISTS + (``_find_session_id`` → no-op when none matches), and a flat channel + ``(…, None)`` row is only created when a human posts a top-level message the + bot processes — a ``chat_postMessage`` cron delivery never goes through the + inbound handler, so the row is usually absent and the mirror silently drops + the brief (verified live: the brief never landed, the reply had no context). + So we CREATE the flat session row first, exactly like + ``_seed_cron_thread_session`` does for threads, then mirror into it. + + The session KEY must match what the user's later inbound reply resolves to + (``build_session_key``): + - **Channel** (``chat_type="group"``): key is + ``…:group::`` — user-isolated — so the seed MUST carry + the **origin's real ``user_id``** (the member who scheduled the job), NOT + a synthetic ``system:cron`` id, or the reply keys to a different session. + - **1:1 DM** (``chat_type="dm"``): the key is ``…:dm:`` and does + NOT embed ``user_id``, so any ``user_id`` resolves to the same session. + ``chat_type`` mirrors the inbound handler's own choice + (``"dm" if is_dm else "group"``, ``adapter.py``), so the seeded key is + byte-identical to the reply's key. + + Returns True if a seed row was created and the brief mirrored, else False + (caller falls back to the plain mirror). Best-effort — a delivery that + already succeeded is never failed by a seeding problem. + """ + text = (mirror_text or "").strip() + if not text: + return False + try: + from gateway.config import Platform + from gateway.session import SessionSource + + chat_type = "dm" if is_dm else "group" + session_store = getattr(adapter, "_session_store", None) + if session_store is not None: + try: + platform_enum = Platform(platform_name.lower()) + except (ValueError, KeyError): + platform_enum = None + if platform_enum is not None: + dest_source = SessionSource( + platform=platform_enum, + chat_id=str(chat_id), + chat_name=chat_name, + chat_type=chat_type, + user_id=str(user_id) if user_id else None, + thread_id=None, # flat — the whole-channel/DM session + ) + # Create the flat session row so the mirror has a target and the + # user's later plain reply joins the SAME session. + session_store.get_or_create_session(dest_source) + + from gateway.mirror import mirror_to_session + + ok = mirror_to_session( + platform_name, + str(chat_id), + f"[Cron delivery: {job.get('name') or job.get('id', 'cron')}]\n{text}", + source_label="cron", + thread_id=None, + user_id=str(user_id) if user_id else None, + role="user", + ) + if ok: + logger.info( + "Job '%s': seeded flat in_channel session on %s:%s (chat_type=%s)", + job.get("id", "?"), platform_name, chat_id, chat_type, + ) + return bool(ok) + except Exception as e: + logger.debug( + "Job '%s': seeding in_channel session failed for %s:%s: %s", + job.get("id", "?"), platform_name, chat_id, e, + ) + return False + + def _cron_job_origin_log_suffix(job: dict) -> str: """Return safe provenance details for security warnings about a cron job. @@ -1261,6 +1357,50 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Option delivered = False target_errors = [] + # Continuable cron surface (D1/D2/D6): resolve the delivery surface for + # this platform generically from its config ``extra``. Default "thread" + # (today's behaviour, byte-identical). "in_channel" delivers the brief + # FLAT into the channel (no dedicated thread) so a plain channel reply + # continues the job in-context via the shared-channel session + # ``(platform, chat_id, None)`` — the same bucket ``reply_in_thread: + # false`` routes inbound channel messages to. The key is read + # generically here (any platform); the ``in_channel`` branch is gated on + # the adapter capability flag ``supports_inchannel_continuable`` so an + # unsupported platform fails SAFE to "thread" (Slack is the first + # consumer; "first consumer ≠ definition"). + surface_mode = "thread" + try: + surface_raw = (pconfig.extra or {}).get("cron_continuable_surface") + if surface_raw is not None and str(surface_raw).strip().lower() == "in_channel": + surface_mode = "in_channel" + except Exception: + surface_mode = "thread" + in_channel_surface = surface_mode == "in_channel" + if in_channel_surface and runtime_adapter is not None and not getattr( + runtime_adapter, "supports_inchannel_continuable", False + ): + # Fail safe (D6): platform has no in_channel continuation primitive. + logger.debug( + "Job '%s': cron_continuable_surface=in_channel not supported on " + "%s, using thread", + job.get("id", "?"), platform_name, + ) + in_channel_surface = False + + # For an in_channel delivery the flat continuation session is created + # explicitly below (the shipped mirror only APPENDS to an existing + # session, and the flat channel row is otherwise absent for a + # chat_postMessage delivery). ``is_dm`` selects the session chat_type so + # the seeded key matches the inbound reply's key: a 1:1 DM keys as + # ``dm`` (Slack DM channel ids start with "D"; or the origin says so), + # everything else as ``group`` (shared channel). ``inchannel_seeded`` + # suppresses the generic mirror below so the brief is not double-written. + origin_chat_type = str(origin.get("chat_type") or "").lower() + is_dm_target = origin_chat_type == "dm" or ( + not origin_chat_type and str(chat_id).startswith("D") + ) + inchannel_seeded = False + # Continuable cron (thread-preferred): when mirroring is enabled for the # origin target and the gateway is live, try to open a DEDICATED thread # for this job and deliver the brief into it. On thread-capable @@ -1269,10 +1409,20 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Option # continues with full context. On DM-only platforms (WhatsApp/Signal) # create_handoff_thread returns None and we fall back to mirroring into # the origin DM session (handled after delivery). Cf. _process_handoff. + # + # in_channel surface (D2): SKIP thread creation entirely — leave + # thread_id=None so the delivery posts flat, then + # ``_seed_cron_channel_session`` (below) CREATES the shared-channel + # session and mirrors the brief into it. The shipped mirror alone is + # NOT enough here: ``mirror_to_session`` only APPENDS to an existing + # session and a flat ``(platform, chat_id, None)`` row is otherwise + # absent for a ``chat_postMessage`` delivery, so the seed must create + # the row first (F5). thread_seeded = False opened_thread_id: Optional[str] = None if ( mirror_this_target + and not in_channel_surface and runtime_adapter is not None and loop is not None and not thread_id # never override an explicit origin thread/topic @@ -1510,10 +1660,21 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Option chat_name=origin.get("chat_name"), ) thread_seeded = True + # in_channel surface: CREATE + seed the flat channel/DM + # session (the shipped mirror only appends to an existing + # session — the flat row is otherwise absent for a + # chat_postMessage delivery, so the brief would be lost). + if in_channel_surface and mirror_this_target and not thread_seeded: + inchannel_seeded = _seed_cron_channel_session( + job, runtime_adapter, platform_name, chat_id, + mirror_text, is_dm=is_dm_target, + user_id=origin_user_id, + chat_name=origin.get("chat_name"), + ) _maybe_mirror_cron_delivery( job, platform_name, chat_id, mirror_text, thread_id=thread_id, user_id=origin_user_id, - enabled=mirror_this_target and not thread_seeded, + enabled=mirror_this_target and not thread_seeded and not inchannel_seeded, ) except Exception as e: err_msg = f"live adapter delivery to {platform_name}:{chat_id} failed: {e}" @@ -1535,12 +1696,30 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Option # prevent "coroutine was never awaited" RuntimeWarning, then retry in a # fresh thread that has no running loop. coro.close() - with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool: - future = pool.submit(asyncio.run, _send_to_platform(platform, pconfig, chat_id, cleaned_delivery_content, thread_id=thread_id, media_files=media_files)) - result = future.result(timeout=30) + # The thread-pool fallback can itself raise (SMTP ConnectionError, + # future.result timeout, etc.). An exception raised inside this + # `except RuntimeError` block is NOT caught by the sibling + # `except Exception` below — it would escape _deliver_result() + # and crash the whole delivery loop, silently skipping every + # remaining target (#47163). Wrap the fallback in its own + # try/except so a per-target failure is logged and the loop + # continues to the next target. + try: + pool = concurrent.futures.ThreadPoolExecutor(max_workers=1) + try: + future = pool.submit(asyncio.run, _send_to_platform(platform, pconfig, chat_id, cleaned_delivery_content, thread_id=thread_id, media_files=media_files)) + result = future.result(timeout=30) + finally: + pool.shutdown(wait=False) + except Exception as e: + msg = f"delivery to {platform_name}:{chat_id} failed: {e}" + logger.error("Job '%s': %s", job["id"], msg, exc_info=True) + target_errors.extend([msg]) + delivery_errors.extend(target_errors) + continue except Exception as e: msg = f"delivery to {platform_name}:{chat_id} failed: {e}" - logger.error("Job '%s': %s", job["id"], msg) + logger.error("Job '%s': %s", job["id"], msg, exc_info=True) target_errors.extend([msg]) delivery_errors.extend(target_errors) continue @@ -2260,9 +2439,11 @@ def run_job(job: dict) -> tuple[bool, str, str, Optional[str]]: agent = None # Mark this as a cron session so the approval system can apply cron_mode. - # This env var is process-wide and persists for the lifetime of the - # scheduler process — every job this process runs is a cron job. - os.environ["HERMES_CRON_SESSION"] = "1" + # Use a per-job ContextVar (NOT a process-global env var) so the flag is + # task-local: concurrent interactive gateway sessions in the same process + # never inherit it (#56771). + from gateway.session_context import set_cron_session as _set_cron_session + _set_cron_session(True) # Use ContextVars for per-job session/delivery state so parallel jobs # don't clobber each other's targets (os.environ is process-global). @@ -2862,6 +3043,12 @@ def run_job(job: dict) -> tuple[bool, str, str, Optional[str]]: clear_session_vars(_ctx_tokens) for _var_name in _cron_delivery_vars: _VAR_MAP[_var_name].set("") + # Reset the per-job cron-session flag (#56771). + try: + from gateway.session_context import clear_cron_session + clear_cron_session() + except Exception: + pass if _session_db: # Title the cron session from the job (name → short prompt → id) so # sidebars/history show a meaningful label instead of the injected diff --git a/gateway/session_context.py b/gateway/session_context.py index cdd1a8bfafef..0a74758ae20a 100644 --- a/gateway/session_context.py +++ b/gateway/session_context.py @@ -113,6 +113,25 @@ def session_context_engaged() -> bool: _CRON_AUTO_DELIVER_CHAT_ID: ContextVar = ContextVar("HERMES_CRON_AUTO_DELIVER_CHAT_ID", default=_UNSET) _CRON_AUTO_DELIVER_THREAD_ID: ContextVar = ContextVar("HERMES_CRON_AUTO_DELIVER_THREAD_ID", default=_UNSET) +# Per-job cron-session flag (issue #56771). +# +# The scheduler used to set ``os.environ["HERMES_CRON_SESSION"] = "1"`` +# process-wide at job start and never clear it. When the gateway and scheduler +# share a process (the normal architecture) the env var leaked into every +# concurrent interactive session, causing the approval system to treat user +# chats as cron and block ``execute_code`` / dangerous commands. +# +# This ContextVar replaces that process-global env var. It is task-local: +# only the asyncio task / worker thread that runs the cron job sees it. +# Concurrent interactive sessions never inherit it. +# +# Tri-state semantics (mirrors the _UNSET pattern of the session vars): +# _UNSET — never set in this context → fall back to os.environ for backward +# compat (tests, CLI cron that sets the env var directly). +# True — this task is running a cron job → approval applies cron_mode. +# False — explicitly not a cron job (overrides a stale env var leak). +_CRON_SESSION: ContextVar = ContextVar("HERMES_CRON_SESSION", default=_UNSET) + _VAR_MAP = { "HERMES_SESSION_PLATFORM": _SESSION_PLATFORM, "HERMES_SESSION_SOURCE": _SESSION_SOURCE, @@ -333,3 +352,47 @@ def async_delivery_supported() -> bool: if value is _UNSET: return True return bool(value) + + +# --------------------------------------------------------------------------- +# Cron-session flag (issue #56771) +# --------------------------------------------------------------------------- + +def set_cron_session(value: bool = True) -> None: + """Mark the current task as running (or not running) a cron job. + + Called by ``cron/scheduler.py::run_job()`` before the agent starts so the + approval system applies ``approvals.cron_mode``. Because this is a + ContextVar, the flag is visible only within the scheduler's task/thread — + concurrent interactive sessions in the same process never inherit it. + """ + _CRON_SESSION.set(bool(value)) + + +def clear_cron_session() -> None: + """Reset the cron-session flag to the "never set" sentinel. + + Called in the ``run_job()`` finally block so a re-entrant or reused context + does not retain the flag after the job completes. + """ + _CRON_SESSION.set(_UNSET) + + +def is_cron_session() -> bool: + """Whether the current task is running a cron job. + + Resolution order: + 1. ContextVar — if explicitly set (True/False), that value is authoritative. + 2. ``HERMES_CRON_SESSION`` env var — fallback when the ContextVar was never + bound in this context (tests, CLI cron that sets the env var directly). + In production the scheduler no longer sets this env var, so interactive + sessions fall through to ``False``. + """ + import os + + value = _CRON_SESSION.get() + if value is not _UNSET: + return bool(value) + return os.getenv("HERMES_CRON_SESSION", "").strip().lower() in ( + "1", "true", "yes", "on", + ) diff --git a/tests/tools/test_cron_session_leak.py b/tests/tools/test_cron_session_leak.py new file mode 100644 index 000000000000..209e4ebf7451 --- /dev/null +++ b/tests/tools/test_cron_session_leak.py @@ -0,0 +1,198 @@ +"""Regression tests for #56771: HERMES_CRON_SESSION env var leaks from the +scheduler process into interactive gateway sessions, blocking execute_code +and dangerous commands for users who never ran a cron job in their chat. + +The fix replaces the process-global ``os.environ["HERMES_CRON_SESSION"]`` +set with a per-job ``ContextVar`` so the cron flag is task-local and cannot +leak into concurrent interactive sessions. +""" + +from __future__ import annotations + +import os +import threading + +import pytest + +from tools import approval as A +from tools.approval import check_execute_code_guard + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +@pytest.fixture(autouse=True) +def _reset_cron_contextvar(): + """Ensure the cron ContextVar is cleared between tests.""" + try: + from gateway.session_context import clear_cron_session + clear_cron_session() + except ImportError: + pass + yield + try: + from gateway.session_context import clear_cron_session + clear_cron_session() + except ImportError: + pass + + +# --------------------------------------------------------------------------- +# 1. is_cron_session() — contextvar-based cron detection +# --------------------------------------------------------------------------- + +class TestCronSessionContextVar: + def test_false_by_default(self, monkeypatch): + """Without a cron contextvar or env var, is_cron_session() is False.""" + from gateway.session_context import is_cron_session + monkeypatch.delenv("HERMES_CRON_SESSION", raising=False) + assert is_cron_session() is False + + def test_true_when_contextvar_set(self, monkeypatch): + """Setting the cron contextvar makes is_cron_session() True.""" + from gateway.session_context import set_cron_session, is_cron_session + monkeypatch.delenv("HERMES_CRON_SESSION", raising=False) + set_cron_session(True) + assert is_cron_session() is True + + def test_contextvar_overrides_leaked_env(self, monkeypatch): + """Even if HERMES_CRON_SESSION leaked into env, an explicitly-cleared + contextvar means NOT cron (interactive session in same process).""" + from gateway.session_context import set_cron_session, is_cron_session + monkeypatch.setenv("HERMES_CRON_SESSION", "1") + set_cron_session(False) + assert is_cron_session() is False + + def test_env_fallback_when_contextvar_unset(self, monkeypatch): + """Backward compat: env var still works when contextvar was never set + (tests, CLI cron that sets the env var directly).""" + from gateway.session_context import is_cron_session + monkeypatch.setenv("HERMES_CRON_SESSION", "1") + assert is_cron_session() is True + + +# --------------------------------------------------------------------------- +# 2. Thread isolation — the core fix +# --------------------------------------------------------------------------- + +class TestCronContextVarThreadIsolation: + def test_scheduler_thread_does_not_leak_to_gateway_thread(self, monkeypatch): + """The cron ContextVar set in the scheduler thread must NOT be visible + in a concurrent gateway/interactive thread. This is the core mechanism + that prevents #56771 — os.environ leaks across threads, ContextVars do + not.""" + from gateway.session_context import set_cron_session, is_cron_session + + monkeypatch.delenv("HERMES_CRON_SESSION", raising=False) + set_cron_session(True) + assert is_cron_session() is True # scheduler thread sees it + + seen: dict = {} + + def gateway_handler(): + # Fresh thread → ContextVar at default (_UNSET) → env fallback + # → env not set (scheduler no longer sets it) → False + seen["is_cron"] = is_cron_session() + + t = threading.Thread(target=gateway_handler) + t.start() + t.join(timeout=5) + + assert seen["is_cron"] is False + + def test_env_var_does_leak_across_threads(self, monkeypatch): + """Documents the pre-fix bug: os.environ IS visible across threads.""" + monkeypatch.setenv("HERMES_CRON_SESSION", "1") + seen: dict = {} + + def handler(): + seen["env"] = os.environ.get("HERMES_CRON_SESSION") + + t = threading.Thread(target=handler) + t.start() + t.join(timeout=5) + + assert seen["env"] == "1" # process-global env leaks — this is the bug + + +# --------------------------------------------------------------------------- +# 3. check_execute_code_guard — cron blocks, interactive doesn't +# --------------------------------------------------------------------------- + +class TestExecuteCodeGuardCronContextVar: + def _setup(self, monkeypatch): + monkeypatch.delenv("HERMES_GATEWAY_SESSION", raising=False) + monkeypatch.delenv("HERMES_INTERACTIVE", raising=False) + monkeypatch.delenv("HERMES_EXEC_ASK", raising=False) + monkeypatch.setattr(A, "_get_approval_mode", lambda: "manual") + monkeypatch.setattr(A, "_get_cron_approval_mode", lambda: "deny") + + def test_cron_contextvar_blocks_execute_code(self, monkeypatch): + """When cron ContextVar is set, execute_code is blocked (real cron job).""" + from gateway.session_context import set_cron_session + self._setup(monkeypatch) + monkeypatch.delenv("HERMES_CRON_SESSION", raising=False) + set_cron_session(True) + + result = check_execute_code_guard("print('hi')", "local") + assert result["approved"] is False + assert "BLOCKED" in result["message"] + + def test_interactive_not_blocked_without_cron_contextvar(self, monkeypatch): + """Interactive session (no cron ContextVar, no leaked env) can run + execute_code.""" + self._setup(monkeypatch) + monkeypatch.delenv("HERMES_CRON_SESSION", raising=False) + + result = check_execute_code_guard("print('hi')", "local") + # Headless local non-gateway non-cron → approved (existing contract) + assert result["approved"] is True + + def test_interactive_thread_not_blocked_after_scheduler_ran(self, monkeypatch): + """Regression for #56771: after the scheduler thread sets the cron + ContextVar, a concurrent interactive thread should still be able to + run execute_code.""" + from gateway.session_context import set_cron_session + self._setup(monkeypatch) + monkeypatch.delenv("HERMES_CRON_SESSION", raising=False) + + # Scheduler thread sets the cron flag + set_cron_session(True) + + seen: dict = {} + + def interactive_session(): + # Fresh thread: ContextVar _UNSET, env not set → not cron + result = check_execute_code_guard("print('hi')", "local") + seen["approved"] = result["approved"] + + t = threading.Thread(target=interactive_session) + t.start() + t.join(timeout=5) + + assert seen["approved"] is True + + +# --------------------------------------------------------------------------- +# 4. _is_gateway_approval_context — cron contextvar short-circuits gateway +# --------------------------------------------------------------------------- + +class TestGatewayContextCronShortCircuit: + def test_cron_contextvar_returns_false_for_gateway(self, monkeypatch): + """When cron ContextVar is set, _is_gateway_approval_context() is False + even if HERMES_GATEWAY_SESSION is also set (cron takes precedence).""" + from gateway.session_context import set_cron_session + monkeypatch.delenv("HERMES_CRON_SESSION", raising=False) + monkeypatch.setenv("HERMES_GATEWAY_SESSION", "1") + set_cron_session(True) + + assert A._is_gateway_approval_context() is False + + def test_interactive_still_gateway_without_cron(self, monkeypatch): + """Interactive gateway session (no cron ContextVar) is recognized as + gateway context.""" + monkeypatch.delenv("HERMES_CRON_SESSION", raising=False) + monkeypatch.setenv("HERMES_GATEWAY_SESSION", "1") + + assert A._is_gateway_approval_context() is True diff --git a/tools/approval.py b/tools/approval.py index 53519b5c0480..8a93c6fdf47a 100644 --- a/tools/approval.py +++ b/tools/approval.py @@ -175,6 +175,22 @@ def _get_session_platform() -> str: return os.getenv("HERMES_SESSION_PLATFORM", "") or "" +def _is_cron_session() -> bool: + """Whether the current task is running a cron job. + + Uses a per-task ContextVar so the flag does not leak from the scheduler + process into concurrent interactive gateway sessions (#56771). Falls back + to the legacy ``HERMES_CRON_SESSION`` env var for backward compat (tests, + CLI cron that sets the env var directly). + """ + try: + from gateway.session_context import is_cron_session + + return is_cron_session() + except Exception: + return env_var_enabled("HERMES_CRON_SESSION") + + def _is_gateway_approval_context() -> bool: """True when this call is inside a gateway/API session. @@ -189,7 +205,7 @@ def _is_gateway_approval_context() -> bool: fall through to the gateway branch would submit a pending approval with no listener and block the job indefinitely. """ - if env_var_enabled("HERMES_CRON_SESSION"): + if _is_cron_session(): return False if env_var_enabled("HERMES_GATEWAY_SESSION"): return True @@ -2002,7 +2018,7 @@ def check_dangerous_command(command: str, env_type: str, if not is_cli and not is_gateway: # Cron sessions: respect cron_mode config - if env_var_enabled("HERMES_CRON_SESSION"): + if _is_cron_session(): if _get_cron_approval_mode() == "deny": return { "approved": False, @@ -2265,7 +2281,7 @@ def check_all_command_guards(command: str, env_type: str, # flows, we do not block on approvals and we skip external guard work. if not is_cli and not is_gateway and not is_ask: # Cron sessions: respect cron_mode config - if env_var_enabled("HERMES_CRON_SESSION"): + if _is_cron_session(): if _get_cron_approval_mode() == "deny": # Run detection to get a description for the block message is_dangerous, _pk, description = detect_dangerous_command(command) @@ -2652,7 +2668,7 @@ def check_execute_code_guard(code: str, env_type: str, 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 _is_cron_session(): if _get_cron_approval_mode() == "deny": return { "approved": False,