diff --git a/cron/scheduler.py b/cron/scheduler.py index 77c277276223..b96790809fa5 100644 --- a/cron/scheduler.py +++ b/cron/scheduler.py @@ -3008,15 +3008,21 @@ def run_job( 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 ContextVars for per-job session/delivery state so parallel jobs # don't clobber each other's targets (os.environ is process-global). from gateway.session_context import set_session_vars, clear_session_vars, _VAR_MAP + # Mark this as a cron session so the approval gate applies cron_mode. + # This MUST be a per-job ContextVar, not os.environ: the default + # deployment runs this ticker in-process inside the gateway, so a + # process-global marker persists after the first job and misroutes every + # concurrent interactive user's approval into the cron branch (a + # process-wide env var never cleared here would either hard-block their + # dangerous commands under cron_mode=deny or, worse, auto-approve them + # under cron_mode=approve). The set + clear both live inside the + # try/finally below so a raise before dispatch can't leave it set on a + # reused loop-thread context. + # Cron execution is an internal scheduler context, not a live inbound # gateway message. Do not seed HERMES_SESSION_* contextvars from the # stored ``origin`` (which is delivery routing metadata, not a sender @@ -3105,6 +3111,9 @@ def run_job( _prior_terminal_cwd = os.environ.get("TERMINAL_CWD", "_UNSET_") _holds_cwd_write = _job_workdir is not None + # Predeclare the cron-marker token so the finally below can guard on it + # even if the try raises before the marker is set. + _cron_marker_token = None if _holds_cwd_write: _terminal_cwd_lock.acquire_write() else: @@ -3116,6 +3125,17 @@ def run_job( # (every future job blocks on acquire_*); a leaked reader blocks all # future writers. Acquire itself can't leak (it either blocks or returns). try: + # Set the cron-session marker (see the ContextVar note above) as the + # first statement in the try so the finally below always restores it. + # It still precedes the copy_context() dispatch further down, so the + # pool thread running the conversation inherits it. Keep the token: the + # finally must reset() to the pre-job state rather than set("") because + # get_session_env treats any explicitly-set value, including "", as + # authoritative and never falls back to os.environ, so a leftover "" + # would misclassify a later standalone/env-marked cron read in this + # context as non-cron. + _cron_marker_token = _VAR_MAP["HERMES_CRON_SESSION"].set("1") + if _job_workdir: os.environ["TERMINAL_CWD"] = _job_workdir logger.info("Job '%s': using workdir %s", job_id, _job_workdir) @@ -3759,10 +3779,16 @@ def _heartbeat_run_claim_if_due(): _terminal_cwd_lock.release_write() else: _terminal_cwd_lock.release_read() - # Clean up ContextVar session/delivery state for this job. - # clear_session_vars also clears _SESSION_CWD internally, so no - # separate clear_session_cwd() call is needed. + # Clean up ContextVar session/delivery state for this job. Reset the + # cron-session marker to its pre-job state (normally the _UNSET + # default) so a reused context is not treated as a cron session and a + # later env-marked read in this context can still fall back to + # os.environ. set("") would break that fallback. clear_session_vars + # also clears _SESSION_CWD internally, so no separate + # clear_session_cwd() call is needed. clear_session_vars(_ctx_tokens) + if _cron_marker_token is not None: + _VAR_MAP["HERMES_CRON_SESSION"].reset(_cron_marker_token) for _var_name in _cron_delivery_vars: _VAR_MAP[_var_name].set("") if _session_db: diff --git a/gateway/session_context.py b/gateway/session_context.py index cf227fac1211..244f137b34a3 100644 --- a/gateway/session_context.py +++ b/gateway/session_context.py @@ -114,6 +114,13 @@ def session_context_engaged() -> bool: # propagates that into this contextvar at session-bind time. _SESSION_ASYNC_DELIVERY: ContextVar = ContextVar("HERMES_SESSION_ASYNC_DELIVERY", default=_UNSET) +# Cron-session marker, set per-job in run_job() so the in-process gateway +# ticker does not leak "this is a cron context" into concurrent interactive +# gateway sessions on the same process. Task-local, unlike a process-global +# env var; the approval gate reads it via get_session_env (contextvar first, +# os.environ fallback for the standalone `hermes cron` process and tests). +_CRON_SESSION: ContextVar = ContextVar("HERMES_CRON_SESSION", default=_UNSET) + # Cron auto-delivery vars — set per-job in run_job() so concurrent jobs # don't clobber each other's delivery targets. _CRON_AUTO_DELIVER_PLATFORM: ContextVar = ContextVar("HERMES_CRON_AUTO_DELIVER_PLATFORM", default=_UNSET) @@ -121,6 +128,7 @@ def session_context_engaged() -> bool: _CRON_AUTO_DELIVER_THREAD_ID: ContextVar = ContextVar("HERMES_CRON_AUTO_DELIVER_THREAD_ID", default=_UNSET) _VAR_MAP = { + "HERMES_CRON_SESSION": _CRON_SESSION, "HERMES_SESSION_PLATFORM": _SESSION_PLATFORM, "HERMES_SESSION_SOURCE": _SESSION_SOURCE, "HERMES_SESSION_CHAT_ID": _SESSION_CHAT_ID, diff --git a/tests/cron/conftest.py b/tests/cron/conftest.py index caaec4559487..5542d70106d2 100644 --- a/tests/cron/conftest.py +++ b/tests/cron/conftest.py @@ -19,3 +19,29 @@ def _default_cron_test_model(monkeypatch): """Pin a default HERMES_MODEL so cron run_job tests have a resolvable model.""" monkeypatch.setenv("HERMES_MODEL", "test-cron-default-model") yield + + +@pytest.fixture(autouse=True) +def _reset_session_context_vars(): + """Reset every session-context ContextVar to its _UNSET default per test. + + Cron tests drive the real ``run_job`` directly in the pytest context, and + its ``clear_session_vars`` finally intentionally pins every session var to + an explicit ``""`` (the gateway relies on that to suppress the + ``os.environ`` fallback). In production the ticker confines that to a + per-job ``copy_context()``, but in a single-process test run it leaks into + later tests that rely on the env fallback: the approval timeout tests + resolve their session key through ``get_session_env`` and stop finding + their registered gateway callback after any cron test has run ``run_job``. + Restoring the defaults on both sides of each test keeps the cron suite + order-independent. + """ + from gateway.session_context import _VAR_MAP, _UNSET + + def _reset_all(): + for var in _VAR_MAP.values(): + var.set(_UNSET) + + _reset_all() + yield + _reset_all() diff --git a/tests/cron/test_cron_session_marker_isolation.py b/tests/cron/test_cron_session_marker_isolation.py new file mode 100644 index 000000000000..f6fd5adfb774 --- /dev/null +++ b/tests/cron/test_cron_session_marker_isolation.py @@ -0,0 +1,431 @@ +"""Regression tests for the cron-session approval marker. + +The default deployment runs the cron ticker in-process inside the gateway, in +the same process that serves interactive users. The "this is a cron context" +marker must therefore be per-job context state, not a process-global env var. +A leaked marker makes the approval gate treat every later interactive user as a +cron session: under cron_mode=deny their dangerous commands are hard-blocked +with a misleading message, and under cron_mode=approve they are auto-approved +with no human prompt. + +The pre-fix code set os.environ["HERMES_CRON_SESSION"] in run_job and never +cleared it. The fix carries the marker on a per-job ContextVar instead. +""" + +import contextvars +import os +import sys +import types + + +sys.modules.setdefault("fire", types.SimpleNamespace(Fire=lambda *a, **k: None)) +sys.modules.setdefault("firecrawl", types.SimpleNamespace(Firecrawl=object)) +sys.modules.setdefault("fal_client", types.SimpleNamespace()) + +import cron.scheduler as cron_scheduler # noqa: E402 +import run_agent # noqa: E402 + + +_JOB = {"id": "j1", "name": "Marker Test", "prompt": "ping", "model": "test-cron-default-model"} + + +class _FakeOpenAI: + def __init__(self, **kwargs): + self.kwargs = kwargs + + def close(self): + return None + + +# Per-test ContextVar hygiene (resetting every session var, including the cron +# marker, to its _UNSET default) lives in the shared autouse fixture in +# tests/cron/conftest.py, so it also covers the other cron tests that drive +# the real run_job directly in the pytest context. + + +def _patch_agent_bootstrap(monkeypatch): + monkeypatch.setattr( + run_agent, + "get_tool_definitions", + lambda **kwargs: [ + { + "type": "function", + "function": { + "name": "terminal", + "description": "Run shell commands.", + "parameters": {"type": "object", "properties": {}}, + }, + } + ], + ) + monkeypatch.setattr(run_agent, "check_toolset_requirements", lambda: {}) + monkeypatch.setattr(run_agent, "OpenAI", _FakeOpenAI) + # Accept the full current resolve_runtime_provider keyword surface + # (requested, explicit_api_key, explicit_base_url, target_model, …). + # run_job now always passes target_model=…; a narrow lambda raises + # TypeError before any cron-marker assertion is reached. + monkeypatch.setattr( + "hermes_cli.runtime_provider.resolve_runtime_provider", + lambda requested=None, **kwargs: { + "provider": "openai", + "api_mode": "chat_completions", + "base_url": "https://api.openai.com/v1", + "api_key": "test-key", + }, + ) + monkeypatch.setattr( + "hermes_cli.runtime_provider.format_runtime_provider_error", lambda exc: str(exc) + ) + + +def _make_stub_agent(record=None): + """An AIAgent whose turn returns a canned dict without any network call. + + When *record* is given, it captures whether the approval gate sees a cron + session while the job's turn is executing (proving cron_mode still applies). + """ + + class _StubAgent(run_agent.AIAgent): + def __init__(self, *args, **kwargs): + kwargs.setdefault("skip_context_files", True) + kwargs.setdefault("skip_memory", True) + kwargs.setdefault("max_iterations", 2) + super().__init__(*args, **kwargs) + self._cleanup_task_resources = lambda task_id: None + self._persist_session = lambda messages, history=None: None + self._save_trajectory = lambda messages, user_message, completed: None + + def run_conversation(self, user_message, conversation_history=None, task_id=None): + if record is not None: + from tools.approval import _is_cron_session + + record["cron_in_job"] = _is_cron_session() + return {"final_response": "done", "turn_exit_reason": ""} + + return _StubAgent + + +def test_run_job_does_not_leak_cron_marker_into_process_env(monkeypatch): + """run_job must not leave HERMES_CRON_SESSION set in os.environ. + + Fails on the pre-fix code, which set the process-global env var and never + cleared it, so the in-process gateway ticker leaked it to every later user. + """ + _patch_agent_bootstrap(monkeypatch) + monkeypatch.setattr(run_agent, "AIAgent", _make_stub_agent()) + monkeypatch.delenv("HERMES_CRON_SESSION", raising=False) + + success, _out, final_response, error = cron_scheduler.run_job(dict(_JOB)) + + assert success is True and error is None + assert final_response == "done" + assert os.environ.get("HERMES_CRON_SESSION") is None, "cron marker leaked into process env" + + +def test_bound_gateway_session_not_shadowed_by_in_process_cron(monkeypatch): + """After a cron job runs in-process, a bound interactive gateway session is + still recognized as a gateway approval context, not routed to cron_mode. + + Fails on the pre-fix code: the leaked process-global marker made + _is_gateway_approval_context return False for the real user. + """ + from gateway.session_context import set_session_vars, clear_session_vars + from tools.approval import _is_gateway_approval_context + + _patch_agent_bootstrap(monkeypatch) + monkeypatch.setattr(run_agent, "AIAgent", _make_stub_agent()) + monkeypatch.delenv("HERMES_CRON_SESSION", raising=False) + + cron_scheduler.run_job(dict(_JOB)) + + tokens = set_session_vars(platform="telegram", chat_id="c1", chat_name="Chat") + try: + assert _is_gateway_approval_context() is True + finally: + clear_session_vars(tokens) + + +def test_marker_is_active_inside_the_cron_job(monkeypatch): + """The approval gate sees the cron marker while the job's turn executes, so + cron_mode is still applied to a cron agent's commands.""" + record = {} + _patch_agent_bootstrap(monkeypatch) + monkeypatch.setattr(run_agent, "AIAgent", _make_stub_agent(record)) + monkeypatch.delenv("HERMES_CRON_SESSION", raising=False) + + cron_scheduler.run_job(dict(_JOB)) + + assert record.get("cron_in_job") is True + + +def test_is_cron_session_prefers_contextvar_then_env(monkeypatch): + from gateway.session_context import _VAR_MAP, _UNSET + from tools.approval import _is_cron_session + + monkeypatch.delenv("HERMES_CRON_SESSION", raising=False) + _VAR_MAP["HERMES_CRON_SESSION"].set(_UNSET) + assert _is_cron_session() is False + + _VAR_MAP["HERMES_CRON_SESSION"].set("1") + assert _is_cron_session() is True + + # An explicitly-set empty value is authoritative (no env fallback) and + # reads as non-cron. This is exactly why run_job must reset() the marker + # to its pre-job state instead of set(""), see the test below. + _VAR_MAP["HERMES_CRON_SESSION"].set("") + assert _is_cron_session() is False + + # os.environ fallback keeps the standalone `hermes cron` process and tests working + _VAR_MAP["HERMES_CRON_SESSION"].set(_UNSET) + monkeypatch.setenv("HERMES_CRON_SESSION", "1") + assert _is_cron_session() is True + + +def test_env_fallback_survives_a_completed_run_job(monkeypatch): + """After a real run_job completes in this context, the os.environ fallback + still works: an env-marked read (the standalone `hermes cron` process, or + a test that sets the flag directly) is still classified as cron. + + Fails when run_job resets the marker with set("") instead of + reset(token): get_session_env treats the explicit "" as authoritative and + never falls back to os.environ, so every env-marked cron read after the + first job in this context is misclassified as non-cron, cron_mode is + skipped, and a dangerous command is auto-approved instead of blocked + under cron_mode deny. + """ + from tools.approval import _is_cron_session + + _patch_agent_bootstrap(monkeypatch) + monkeypatch.setattr(run_agent, "AIAgent", _make_stub_agent()) + monkeypatch.delenv("HERMES_CRON_SESSION", raising=False) + + ok = cron_scheduler.run_job(dict(_JOB))[0] + assert ok is True + # The marker is back to its pre-job state, not pinned to an explicit "". + assert _is_cron_session() is False + + monkeypatch.setenv("HERMES_CRON_SESSION", "1") + assert _is_cron_session() is True, ( + "env-marked cron read misclassified as non-cron after a completed run_job" + ) + + +def test_cron_marker_isolated_between_contexts(monkeypatch): + """A marker set inside one job's context is invisible to a sibling context + (a concurrent gateway request). This per-context isolation is the fix.""" + from gateway.session_context import _VAR_MAP + from tools.approval import _is_cron_session + + monkeypatch.delenv("HERMES_CRON_SESSION", raising=False) + + def _job_ctx(): + _VAR_MAP["HERMES_CRON_SESSION"].set("1") + return _is_cron_session() + + ctx = contextvars.copy_context() + assert ctx.run(_job_ctx) is True + # Outside that copied context the marker was never set. + assert _is_cron_session() is False + + +def test_two_real_run_jobs_isolate_marker_across_contexts(monkeypatch): + """Two real run_job calls, each dispatched in its own context (as the + in-process ticker does), keep the marker to their own job: each turn sees + cron_mode, and neither leaves it set in a sibling context or the base + thread. Drives the raw ContextVar isolation through the real run_job path, + not a synthetic set().""" + from gateway.session_context import set_session_vars, clear_session_vars + from tools.approval import _is_cron_session, _is_gateway_approval_context + + _patch_agent_bootstrap(monkeypatch) + monkeypatch.delenv("HERMES_CRON_SESSION", raising=False) + + # Job A in its own context sees cron while its turn runs. + rec_a = {} + monkeypatch.setattr(run_agent, "AIAgent", _make_stub_agent(rec_a)) + ctx_a = contextvars.copy_context() + ok_a = ctx_a.run(lambda: cron_scheduler.run_job(dict(_JOB))[0]) + assert ok_a is True and rec_a.get("cron_in_job") is True + # A's marker never escaped into the base context. + assert _is_cron_session() is False + + # A concurrent interactive gateway session (base context) stays gateway. + tokens = set_session_vars(platform="telegram", chat_id="c1", chat_name="Chat") + try: + assert _is_gateway_approval_context() is True + finally: + clear_session_vars(tokens) + + # Job B in a second context: same story, no bleed from A's run. + rec_b = {} + monkeypatch.setattr(run_agent, "AIAgent", _make_stub_agent(rec_b)) + ctx_b = contextvars.copy_context() + ok_b = ctx_b.run(lambda: cron_scheduler.run_job({**_JOB, "id": "j2"})[0]) + assert ok_b is True and rec_b.get("cron_in_job") is True + assert _is_cron_session() is False + + +def test_marker_cleared_even_when_agent_raises(monkeypatch): + """An exception inside AIAgent.run_conversation still unwinds the marker + via run_job's finally. A raise mid-tick must not leave the contextvar set + on a reused loop-thread context (or in os.environ). + """ + from tools.approval import _is_cron_session + + class _RaisingAgent(run_agent.AIAgent): + def __init__(self, *args, **kwargs): + kwargs.setdefault("skip_context_files", True) + kwargs.setdefault("skip_memory", True) + kwargs.setdefault("max_iterations", 2) + super().__init__(*args, **kwargs) + self._cleanup_task_resources = lambda task_id: None + self._persist_session = lambda messages, history=None: None + self._save_trajectory = lambda messages, user_message, completed: None + + def run_conversation(self, user_message, conversation_history=None, task_id=None): + raise RuntimeError("simulated agent crash mid-tick") + + _patch_agent_bootstrap(monkeypatch) + monkeypatch.setattr(run_agent, "AIAgent", _RaisingAgent) + monkeypatch.delenv("HERMES_CRON_SESSION", raising=False) + + success, _out, _final, error = cron_scheduler.run_job(dict(_JOB)) + + assert success is False + assert error and "simulated agent crash" in error + assert _is_cron_session() is False + assert os.environ.get("HERMES_CRON_SESSION") is None + + +def test_stale_process_env_does_not_reclassify_bound_gateway_session(monkeypatch): + """A leftover HERMES_CRON_SESSION=1 in os.environ must not push a bound + interactive gateway session into the cron approval branch. + + The ContextVar is never bound for that interactive turn, so a naive env + fallback would fire. Live gateway identity wins: the user stays on the + interactive approval path instead of the cron hard-block. + """ + from gateway.session_context import set_session_vars, clear_session_vars + from tools.approval import ( + _is_cron_session, + _is_gateway_approval_context, + check_dangerous_command, + ) + + monkeypatch.setenv("HERMES_CRON_SESSION", "1") + monkeypatch.setattr("tools.approval._get_cron_approval_mode", lambda: "deny") + + tokens = set_session_vars(platform="telegram", chat_id="c1", chat_name="Chat") + try: + assert _is_cron_session() is False + assert _is_gateway_approval_context() is True + result = check_dangerous_command("rm -rf /tmp/stuff", "local") + msg = (result.get("message") or "").lower() + # Interactive gateway path asks the user; cron deny would hard-block + # with "without a user present" and never set approval_required. + assert "without a user present" not in msg + assert result.get("status") == "approval_required" + finally: + clear_session_vars(tokens) + + +def test_cron_contextvar_drives_deny_for_dangerous_and_execute_code(monkeypatch): + """When the per-job ContextVar is bound, cron_mode=deny hard-blocks both + the terminal dangerous-command path and execute_code. The marker is the + real policy input, not just a helper used by unit tests. + """ + from gateway.session_context import _VAR_MAP, _UNSET + from tools import approval as approval_mod + + monkeypatch.delenv("HERMES_CRON_SESSION", raising=False) + monkeypatch.setattr(approval_mod, "_get_cron_approval_mode", lambda: "deny") + _VAR_MAP["HERMES_CRON_SESSION"].set("1") + try: + term = approval_mod.check_dangerous_command("rm -rf /tmp/stuff", "local") + assert term.get("approved") is False + assert "without a user present" in (term.get("message") or "").lower() + + code = approval_mod.check_execute_code_guard( + "import os; os.system('id')", "local" + ) + assert code.get("approved") is False + assert "without a user present" in (code.get("message") or "").lower() + finally: + _VAR_MAP["HERMES_CRON_SESSION"].set(_UNSET) + + +def test_cron_deny_not_bypassed_by_exec_ask_flag(monkeypatch): + """HERMES_EXEC_ASK=1 must not skip cron-deny in check_all_command_guards. + + Cron often shares a process with the gateway, which can leave interactive + approval flags set. Cron has no user to answer an ask prompt, so + cron_mode=deny has to win over the ask gate. + """ + from gateway.session_context import _VAR_MAP, _UNSET + from tools import approval as approval_mod + + monkeypatch.delenv("HERMES_CRON_SESSION", raising=False) + monkeypatch.setenv("HERMES_EXEC_ASK", "1") + monkeypatch.delenv("HERMES_GATEWAY_SESSION", raising=False) + monkeypatch.delenv("HERMES_INTERACTIVE", raising=False) + monkeypatch.delenv("HERMES_YOLO_MODE", raising=False) + monkeypatch.setattr(approval_mod, "_get_cron_approval_mode", lambda: "deny") + _VAR_MAP["HERMES_CRON_SESSION"].set("1") + try: + result = approval_mod.check_all_command_guards("rm -rf /tmp/stuff", "local") + assert result.get("approved") is False + assert "without a user present" in (result.get("message") or "").lower() + assert result.get("status") != "approval_required" + finally: + _VAR_MAP["HERMES_CRON_SESSION"].set(_UNSET) + + +def test_stale_env_does_not_block_execute_code_in_gateway_session(monkeypatch): + """A stale HERMES_CRON_SESSION=1 in os.environ must not push a bound + interactive gateway session's execute_code into the cron deny path. + + Companion to test_stale_process_env_does_not_reclassify_bound_gateway_session + which covers the same scenario for check_dangerous_command. The + execute_code guard has its own cron branch, so it needs its own + assertion that the stale-env fallback is suppressed when a gateway + session is live. + + Production trigger (#73195): a Feishu user replies to a cron-delivered + message card. The gateway process still has HERMES_CRON_SESSION=1 in + os.environ from the prior cron tick, but the interactive turn should + reach the normal approval path, not the cron hard-block. + """ + from gateway.session_context import set_session_vars, clear_session_vars + from tools import approval as approval_mod + + # Simulate a leaked process env from a prior cron tick and pin manual mode + # so the expected interactive path is an approval request, not a yolo/off + # or smart-mode approval. + monkeypatch.setenv("HERMES_CRON_SESSION", "1") + monkeypatch.setattr(approval_mod, "_get_cron_approval_mode", lambda: "deny") + monkeypatch.setattr(approval_mod, "_get_approval_mode", lambda: "manual") + monkeypatch.setattr(approval_mod, "_YOLO_MODE_FROZEN", False) + # Keep the assertion independent of a developer's persistent/session + # approval allowlist when this file is run outside the canonical sandbox. + monkeypatch.setattr(approval_mod, "is_approved", lambda *_args: False) + monkeypatch.setattr( + approval_mod, "is_current_session_yolo_enabled", lambda: False + ) + + # Gateway binds an interactive session (e.g. Feishu reply) + tokens = set_session_vars(platform="feishu", chat_id="c1", chat_name="Chat") + try: + result = approval_mod.check_execute_code_guard( + "print('hello world')", "local", has_host_access=False + ) + # Must NOT be hard-blocked by cron deny. The gateway path returns + # approval_pending (interactive prompt queued) which is correct. + msg = (result.get("message") or "").lower() + assert "without a user present" not in msg, ( + f"Cron deny message reached an interactive gateway session: {result}" + ) + assert result.get("approved") is False + assert result.get("approval_pending") is True + assert result.get("status") == "pending_approval" + finally: + clear_session_vars(tokens) diff --git a/tests/tools/test_cron_approval_mode.py b/tests/tools/test_cron_approval_mode.py index a2a5a839a1ab..cd7952f26770 100644 --- a/tests/tools/test_cron_approval_mode.py +++ b/tests/tools/test_cron_approval_mode.py @@ -361,26 +361,34 @@ def test_non_cron_non_interactive_still_auto_approves(self, monkeypatch): class TestCronWithGatewayOrigin: - """Cron jobs originating from a gateway platform must NOT be treated as gateway. - - cron/scheduler.py binds HERMES_SESSION_PLATFORM via contextvars for - delivery routing (so cron output lands back in the origin chat). The - API-server approvals work (PR #20311) made check_dangerous_command treat - any contextvar-bound platform as a gateway session. That would route - cron-from-telegram/discord/etc. through submit_pending with no listener, - hanging the job instead of respecting approvals.cron_mode. + """Cron jobs that also bind a gateway platform must NOT be treated as gateway. + + A job may carry platform/chat routing state alongside the per-job cron + marker (delivery metadata, origin chat, etc.). The API-server approvals + work (PR #20311) made check_dangerous_command treat any contextvar-bound + platform as a gateway session. That would route cron-from-telegram through + submit_pending with no listener, hanging the job instead of respecting + approvals.cron_mode. The per-job HERMES_CRON_SESSION ContextVar is + authoritative over a co-bound platform. """ def test_cron_with_telegram_origin_uses_cron_mode_not_gateway(self, monkeypatch): - """Cron + contextvar platform=telegram + cron_mode=deny → BLOCKED, not pending.""" - monkeypatch.setenv("HERMES_CRON_SESSION", "1") + """Cron marker + platform=telegram + cron_mode=deny → BLOCKED, not pending.""" + monkeypatch.delenv("HERMES_CRON_SESSION", raising=False) monkeypatch.delenv("HERMES_INTERACTIVE", raising=False) monkeypatch.delenv("HERMES_GATEWAY_SESSION", raising=False) monkeypatch.delenv("HERMES_YOLO_MODE", raising=False) monkeypatch.delenv("HERMES_EXEC_ASK", raising=False) - from gateway.session_context import set_session_vars, clear_session_vars + from gateway.session_context import ( + _UNSET, + _VAR_MAP, + clear_session_vars, + set_session_vars, + ) + tokens = set_session_vars(platform="telegram", chat_id="123") + _VAR_MAP["HERMES_CRON_SESSION"].set("1") try: from unittest.mock import patch as mock_patch with mock_patch("tools.approval._get_cron_approval_mode", return_value="deny"): @@ -391,18 +399,26 @@ def test_cron_with_telegram_origin_uses_cron_mode_not_gateway(self, monkeypatch) assert "cron_mode" in result["message"] assert result.get("status") != "approval_required" finally: + _VAR_MAP["HERMES_CRON_SESSION"].set(_UNSET) clear_session_vars(tokens) def test_cron_with_telegram_origin_approve_mode_allows(self, monkeypatch): - """Cron + contextvar platform=telegram + cron_mode=approve → allowed via cron path.""" - monkeypatch.setenv("HERMES_CRON_SESSION", "1") + """Cron marker + platform=discord + cron_mode=approve → allowed via cron path.""" + monkeypatch.delenv("HERMES_CRON_SESSION", raising=False) monkeypatch.delenv("HERMES_INTERACTIVE", raising=False) monkeypatch.delenv("HERMES_GATEWAY_SESSION", raising=False) monkeypatch.delenv("HERMES_YOLO_MODE", raising=False) monkeypatch.delenv("HERMES_EXEC_ASK", raising=False) - from gateway.session_context import set_session_vars, clear_session_vars + from gateway.session_context import ( + _UNSET, + _VAR_MAP, + clear_session_vars, + set_session_vars, + ) + tokens = set_session_vars(platform="discord", chat_id="456") + _VAR_MAP["HERMES_CRON_SESSION"].set("1") try: from unittest.mock import patch as mock_patch with mock_patch("tools.approval._get_cron_approval_mode", return_value="approve"): @@ -411,5 +427,33 @@ def test_cron_with_telegram_origin_approve_mode_allows(self, monkeypatch): # Should NOT be a gateway-approval response. assert result.get("status") != "approval_required" finally: + _VAR_MAP["HERMES_CRON_SESSION"].set(_UNSET) clear_session_vars(tokens) + def test_cron_with_telegram_origin_combined_guard_uses_cron_mode(self, monkeypatch): + """check_all_command_guards must also honor cron_mode over gateway classification.""" + monkeypatch.delenv("HERMES_CRON_SESSION", raising=False) + monkeypatch.delenv("HERMES_INTERACTIVE", raising=False) + monkeypatch.delenv("HERMES_GATEWAY_SESSION", raising=False) + monkeypatch.delenv("HERMES_YOLO_MODE", raising=False) + monkeypatch.delenv("HERMES_EXEC_ASK", raising=False) + + from gateway.session_context import ( + _UNSET, + _VAR_MAP, + clear_session_vars, + set_session_vars, + ) + + tokens = set_session_vars(platform="telegram", chat_id="789") + _VAR_MAP["HERMES_CRON_SESSION"].set("1") + try: + from unittest.mock import patch as mock_patch + with mock_patch("tools.approval._get_cron_approval_mode", return_value="deny"): + result = check_all_command_guards("rm -rf /tmp/stuff", "local") + assert not result["approved"] + assert "BLOCKED" in result["message"] + assert result.get("status") != "approval_required" + finally: + _VAR_MAP["HERMES_CRON_SESSION"].set(_UNSET) + clear_session_vars(tokens) diff --git a/tests/tools/test_request_tool_approval.py b/tests/tools/test_request_tool_approval.py index 54ca18fcd523..7ffffb7dcbf7 100644 --- a/tests/tools/test_request_tool_approval.py +++ b/tests/tools/test_request_tool_approval.py @@ -77,8 +77,8 @@ def test_cli_session_persists_session_only(self, monkeypatch): def test_cron_deny_mode_blocks(self, monkeypatch): monkeypatch.setattr(approval, "_is_interactive_cli", lambda: False) monkeypatch.setattr(approval, "_is_gateway_approval_context", lambda: False) - monkeypatch.setattr(approval, "env_var_enabled", - lambda v: v == "HERMES_CRON_SESSION") + # Gate uses _is_cron_session(), not env_var_enabled("HERMES_CRON_SESSION"). + monkeypatch.setattr(approval, "_is_cron_session", lambda: True) monkeypatch.setattr(approval, "_get_cron_approval_mode", lambda: "deny") res = request_tool_approval("terminal", "smtp send") assert res["approved"] is False @@ -87,8 +87,7 @@ def test_cron_deny_mode_blocks(self, monkeypatch): def test_cron_approve_mode_allows(self, monkeypatch): monkeypatch.setattr(approval, "_is_interactive_cli", lambda: False) monkeypatch.setattr(approval, "_is_gateway_approval_context", lambda: False) - monkeypatch.setattr(approval, "env_var_enabled", - lambda v: v == "HERMES_CRON_SESSION") + monkeypatch.setattr(approval, "_is_cron_session", lambda: True) monkeypatch.setattr(approval, "_get_cron_approval_mode", lambda: "approve") res = request_tool_approval("terminal", "smtp send") assert res["approved"] is True @@ -113,10 +112,10 @@ def test_explicit_rule_key_overrides_derivation(self, monkeypatch): def test_no_human_non_cron_fails_closed(self, monkeypatch): """Non-interactive, non-gateway, NON-cron context blocks (fail-closed) - — a plugin-flagged action never runs ungated without a human.""" + so a plugin-flagged action never runs ungated without a human.""" monkeypatch.setattr(approval, "_is_interactive_cli", lambda: False) monkeypatch.setattr(approval, "_is_gateway_approval_context", lambda: False) - monkeypatch.setattr(approval, "env_var_enabled", lambda v: False) # not cron + monkeypatch.setattr(approval, "_is_cron_session", lambda: False) res = request_tool_approval("terminal", "smtp send") assert res["approved"] is False assert "no interactive user or gateway" in res["message"].lower() diff --git a/tools/approval.py b/tools/approval.py index 5db5065f9063..b6055da5823c 100644 --- a/tools/approval.py +++ b/tools/approval.py @@ -224,6 +224,41 @@ def _get_session_platform() -> str: return os.getenv("HERMES_SESSION_PLATFORM", "") or "" +def _is_cron_session() -> bool: + """True when the current execution context is a cron job. + + Resolution order: + 1. The per-job ``HERMES_CRON_SESSION`` ContextVar set by ``run_job`` — + authoritative whenever it has been bound in this context (including + an explicit empty value, which means "not cron"). + 2. ``os.environ["HERMES_CRON_SESSION"]`` — only when the ContextVar was + never bound here. That keeps the standalone ``hermes cron`` process + and tests that set the flag directly working. + + The env fallback is suppressed when a live gateway session is already + bound (``HERMES_SESSION_PLATFORM`` or ``HERMES_GATEWAY_SESSION``). A + leftover process-global env value must not reclassify an interactive + user as a cron session; the in-process ticker used to leave exactly + that residue after the first tick. + """ + try: + from gateway.session_context import _UNSET, _VAR_MAP + + value = _VAR_MAP["HERMES_CRON_SESSION"].get() + if value is not _UNSET: + return is_truthy_value(value, default=False) + except Exception: + # session_context unavailable — fall through to env only. + return is_truthy_value(os.getenv("HERMES_CRON_SESSION", ""), default=False) + + # ContextVar never bound in this context. Skip the process-env fallback + # for a live gateway session so a residual HERMES_CRON_SESSION=1 cannot + # shadow interactive approvals. + if _get_session_platform() or env_var_enabled("HERMES_GATEWAY_SESSION"): + return False + return is_truthy_value(os.getenv("HERMES_CRON_SESSION", ""), default=False) + + def _is_gateway_approval_context() -> bool: """True when this call is inside a gateway/API session. @@ -238,7 +273,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 @@ -2899,7 +2934,7 @@ def _run_approval_gate( 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, @@ -3423,72 +3458,78 @@ def check_all_command_guards(command: str, env_type: str, is_gateway = _is_gateway_approval_context() is_ask = env_var_enabled("HERMES_EXEC_ASK") - # Preserve the existing non-interactive behavior: outside CLI/gateway/ask - # 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 _get_cron_approval_mode() == "deny": - # Run detection to get a description for the block message - is_dangerous, _pk, description = detect_dangerous_command(command) - if is_dangerous: + # Cron has no interactive approval surface. Apply cron_mode BEFORE the + # CLI/gateway/ask gates so an inherited HERMES_EXEC_ASK=1 (common in a + # shared gateway process) cannot skip cron-deny and fall into the ask + # approval flow with no user present. Mirrors check_execute_code_guard, + # which already evaluates cron first. + 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) + if is_dangerous: + return { + "approved": False, + "message": ( + f"BLOCKED: Command flagged as dangerous ({description}) " + "but cron jobs run without a user present to approve it. " + "Find an alternative approach that avoids this command. " + "To allow dangerous commands in cron jobs, set " + "approvals.cron_mode: approve in config.yaml." + ), + } + # Also run tirith check in cron-deny mode so content-level + # threats (homograph URLs, pipe-to-interpreter, terminal + # injection, etc.) are caught even when they do not match + # the pattern-based detection above. + try: + from tools.tirith_security import check_command_security + _cron_tirith = check_command_security(command) + if _cron_tirith.get("action") in ("block", "warn"): + _cron_desc = _format_tirith_description(_cron_tirith) return { "approved": False, "message": ( - f"BLOCKED: Command flagged as dangerous ({description}) " + f"BLOCKED: {_cron_desc} " "but cron jobs run without a user present to approve it. " "Find an alternative approach that avoids this command. " "To allow dangerous commands in cron jobs, set " "approvals.cron_mode: approve in config.yaml." ), } - # Also run tirith check in cron-deny mode so content-level - # threats (homograph URLs, pipe-to-interpreter, terminal - # injection, etc.) are caught even when they do not match - # the pattern-based detection above. + except ImportError: + # Tirith not installed. Honour security.tirith_fail_open: + # the default (True) allows as before, but when an operator + # has explicitly opted into fail-closed the command cannot + # be silently allowed — and a cron session has no user to + # approve it, so fail-closed means block (mirrors the + # fail-closed synthesis in the main flow below; see #20733). + _cron_fail_open = True # safe default if config is unreadable try: - from tools.tirith_security import check_command_security - _cron_tirith = check_command_security(command) - if _cron_tirith.get("action") in ("block", "warn"): - _cron_desc = _format_tirith_description(_cron_tirith) - return { - "approved": False, - "message": ( - f"BLOCKED: {_cron_desc} " - "but cron jobs run without a user present to approve it. " - "Find an alternative approach that avoids this command. " - "To allow dangerous commands in cron jobs, set " - "approvals.cron_mode: approve in config.yaml." - ), - } - except ImportError: - # Tirith not installed. Honour security.tirith_fail_open: - # the default (True) allows as before, but when an operator - # has explicitly opted into fail-closed the command cannot - # be silently allowed — and a cron session has no user to - # approve it, so fail-closed means block (mirrors the - # fail-closed synthesis in the main flow below; see #20733). - _cron_fail_open = True # safe default if config is unreadable - try: - from hermes_cli.config import load_config as _load_cfg - _sec = (_load_cfg() or {}).get("security", {}) or {} - if _sec.get("tirith_enabled", True): - _cron_fail_open = _sec.get("tirith_fail_open", True) - except Exception: - pass - if not _cron_fail_open: - return { - "approved": False, - "message": ( - "BLOCKED: the Tirith security scanner could not be " - "imported and security.tirith_fail_open is false, " - "so this command cannot be silently allowed — and " - "cron jobs run without a user present to approve it. " - "Find an alternative approach, install tirith, or set " - "approvals.cron_mode: approve in config.yaml." - ), - } - # else: tirith_fail_open is True — allow as before + from hermes_cli.config import load_config as _load_cfg + _sec = (_load_cfg() or {}).get("security", {}) or {} + if _sec.get("tirith_enabled", True): + _cron_fail_open = _sec.get("tirith_fail_open", True) + except Exception: + pass + if not _cron_fail_open: + return { + "approved": False, + "message": ( + "BLOCKED: the Tirith security scanner could not be " + "imported and security.tirith_fail_open is false, " + "so this command cannot be silently allowed — and " + "cron jobs run without a user present to approve it. " + "Find an alternative approach, install tirith, or set " + "approvals.cron_mode: approve in config.yaml." + ), + } + # else: tirith_fail_open is True — allow as before + return {"approved": True, "message": None} + + # Preserve the existing non-interactive behavior: outside CLI/gateway/ask + # 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: return {"approved": True, "message": None} # --- Phase 1: Gather findings from both checks --- @@ -3878,7 +3919,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,