diff --git a/cron/lifecycle_guard.py b/cron/lifecycle_guard.py index b09635c998bb..00c1824fb43c 100644 --- a/cron/lifecycle_guard.py +++ b/cron/lifecycle_guard.py @@ -190,7 +190,7 @@ def _iter_referenced_shell_scripts( executable = segment[index] executable_name = Path(executable).name - if executable_name in {".", "source"}: + if executable in {".", "source"} or executable_name == "source": if len(segment) > index + 1: yield _resolve_terminal_script_path(segment[index + 1], cwd) continue @@ -219,8 +219,12 @@ def _iter_referenced_shell_scripts( yield _resolve_terminal_script_path(arguments[arg_index], cwd) continue - if "/" in executable or executable.endswith((".sh", ".bash", ".zsh")): + if executable.endswith((".sh", ".bash", ".zsh")): yield _resolve_terminal_script_path(executable, cwd) + elif "/" in executable: + candidate = _resolve_terminal_script_path(executable, cwd) + if candidate.suffix in (".sh", ".bash", ".zsh") or _is_shell_script_file(candidate): + yield candidate def _iter_shell_command_payloads(command: str) -> Iterator[str]: @@ -252,7 +256,7 @@ def _read_referenced_script(path: Path) -> tuple[Optional[str], bool]: flags = os.O_RDONLY | getattr(os, "O_NONBLOCK", 0) try: descriptor = os.open(path, flags) - except OSError: + except (OSError, ValueError): return None, False try: metadata = os.fstat(descriptor) @@ -270,6 +274,35 @@ def _read_referenced_script(path: Path) -> tuple[Optional[str], bool]: return data.decode("utf-8", errors="replace"), False +def _is_shell_script_file(path: Path) -> bool: + """Return whether *path* is a bounded-read script candidate. + + Text candidates are scanned (defence-in-depth: POSIX shells execute + extensionless text scripts without a shebang via the ENOEXEC fallback), + while clearly binary files are skipped using a bounded NUL-byte header + check — a binary decoded as UTF-8 embeds NUL bytes that previously + crashed the resolver (ValueError: embedded null byte) or produced + false-positive gateway-lifecycle verdicts. + """ + flags = os.O_RDONLY | getattr(os, "O_NONBLOCK", 0) + try: + descriptor = os.open(path, flags) + except (OSError, ValueError): + return False + try: + metadata = os.fstat(descriptor) + if not stat.S_ISREG(metadata.st_mode): + return False + data = os.read(descriptor, 512) + except OSError: + return False + finally: + os.close(descriptor) + if not data: + return False + return b"\x00" not in data + + def _contains_unsafe_gateway_action( command: str, *, @@ -298,7 +331,7 @@ def _contains_unsafe_gateway_action( for script_path in _iter_referenced_shell_scripts(command, cwd=cwd): try: resolved = script_path.resolve(strict=False) - except OSError: + except (OSError, ValueError): resolved = script_path if resolved in visited: continue diff --git a/cron/scheduler.py b/cron/scheduler.py index 77c277276223..37be0c9a3411 100644 --- a/cron/scheduler.py +++ b/cron/scheduler.py @@ -3008,11 +3008,6 @@ 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 @@ -3110,12 +3105,17 @@ def run_job( else: _terminal_cwd_lock.acquire_read() + _cron_flag_token = None + # Everything after the acquire MUST live inside this try, so the finally # below always releases the lock even if the env override or any later # statement raises. A leaked writer would deadlock the whole scheduler # (every future job blocks on acquire_*); a leaked reader blocks all # future writers. Acquire itself can't leak (it either blocks or returns). try: + from gateway.session_context import HERMES_CRON_SESSION_CONTEXTVAR + + _cron_flag_token = HERMES_CRON_SESSION_CONTEXTVAR.set(True) if _job_workdir: os.environ["TERMINAL_CWD"] = _job_workdir logger.info("Job '%s': using workdir %s", job_id, _job_workdir) @@ -3745,6 +3745,8 @@ def _heartbeat_run_claim_if_due(): return False, output, "", error_msg finally: + if _cron_flag_token is not None: + HERMES_CRON_SESSION_CONTEXTVAR.reset(_cron_flag_token) # Restore TERMINAL_CWD to whatever it was before this job ran. We # only ever mutate it when the job has a workdir; see the setup block # at the top of run_job for the serialization guarantee. diff --git a/gateway/session_context.py b/gateway/session_context.py index c742a024e698..e66460538a7b 100644 --- a/gateway/session_context.py +++ b/gateway/session_context.py @@ -95,6 +95,10 @@ def session_context_engaged() -> bool: _SESSION_PROFILE: ContextVar = ContextVar("HERMES_SESSION_PROFILE", default=_UNSET) +HERMES_CRON_SESSION_CONTEXTVAR: ContextVar[bool] = ContextVar( + "hermes_cron_session_contextvar", default=False +) + # Whether the current session's delivery channel can route an ASYNC completion # back to the agent AFTER the current turn ends (i.e. wake a fresh turn). # diff --git a/tests/cron/test_lifecycle_guard_regressions.py b/tests/cron/test_lifecycle_guard_regressions.py new file mode 100644 index 000000000000..b4bce3ecdc46 --- /dev/null +++ b/tests/cron/test_lifecycle_guard_regressions.py @@ -0,0 +1,226 @@ +from pathlib import Path +import sys + +from cron.lifecycle_guard import ( + _contains_unsafe_gateway_action, + contains_gateway_lifecycle_command_or_referenced_script, +) +from gateway.session_context import HERMES_CRON_SESSION_CONTEXTVAR +from tools import approval + + +def test_full_path_non_shell_binary_is_not_scanned(tmp_path: Path) -> None: + binary = tmp_path / "python3" + binary.write_bytes(b"\x00" * 4096) + + command = f'{binary} -c "print(1)"' + + assert not contains_gateway_lifecycle_command_or_referenced_script(command, cwd=str(tmp_path)) + assert not _contains_unsafe_gateway_action( + command, cwd=str(tmp_path), depth=0, visited=set() + ) + + +def test_extensionless_text_script_without_shebang_is_scanned(tmp_path: Path) -> None: + script = tmp_path / "restart-helper" + script.write_text("hermes gateway restart\n") + + assert _contains_unsafe_gateway_action( + str(script), cwd=".", depth=0, visited=set() + ) + + +def test_shell_script_reference_is_still_scanned(tmp_path: Path) -> None: + script = tmp_path / "script.sh" + script.write_text("#!/bin/sh\nhermes gateway restart\n") + + assert contains_gateway_lifecycle_command_or_referenced_script( + f"./{script.name}", cwd=str(tmp_path) + ) + + +def test_actual_python_binary_is_not_scanned() -> None: + command = f'{sys.executable} -c "print(1)"' + + assert not _contains_unsafe_gateway_action( + command, cwd=".", depth=0, visited=set() + ) + + +def test_dot_source_and_source_script_references_are_scanned(tmp_path: Path) -> None: + script = tmp_path / "evil.sh" + script.write_text("#!/bin/sh\nhermes gateway restart\n") + + for prefix in (".", "source"): + assert _contains_unsafe_gateway_action( + f"{prefix} {script}", cwd=".", depth=0, visited=set() + ) + + +def test_bash_script_reference_is_scanned(tmp_path: Path) -> None: + script = tmp_path / "evil.sh" + script.write_text("#!/bin/sh\nhermes gateway restart\n") + + assert _contains_unsafe_gateway_action( + f"bash {script}", cwd=".", depth=0, visited=set() + ) + + +def test_large_extensionless_shell_script_is_scanned(tmp_path: Path) -> None: + script = tmp_path / "large-script" + script.write_text("#!/bin/sh\n" + ("# " + "x" * 5000 + "\n") + "hermes gateway restart\n") + + assert _contains_unsafe_gateway_action( + str(script), cwd=".", depth=0, visited=set() + ) + + +def test_embedded_null_path_does_not_raise() -> None: + command = "./script\x00.sh" + + assert not _contains_unsafe_gateway_action( + command, cwd="/tmp", depth=0, visited=set() + ) + + +def test_cron_session_contextvar_takes_precedence(monkeypatch) -> None: + monkeypatch.delenv("HERMES_CRON_SESSION", raising=False) + token = HERMES_CRON_SESSION_CONTEXTVAR.set(True) + try: + assert approval._is_cron_session() + finally: + HERMES_CRON_SESSION_CONTEXTVAR.reset(token) + + +def test_cron_session_contextvar_reset_restores_false(monkeypatch) -> None: + monkeypatch.delenv("HERMES_CRON_SESSION", raising=False) + token = HERMES_CRON_SESSION_CONTEXTVAR.set(True) + try: + assert approval._is_cron_session() is True + finally: + HERMES_CRON_SESSION_CONTEXTVAR.reset(token) + + assert approval._is_cron_session() is False + + +def test_cron_session_is_false_without_context_or_env(monkeypatch) -> None: + monkeypatch.delenv("HERMES_CRON_SESSION", raising=False) + + assert not approval._is_cron_session() + + +def test_cron_session_env_fallback(monkeypatch) -> None: + monkeypatch.setenv("HERMES_CRON_SESSION", "1") + + assert approval._is_cron_session() + + +def test_run_job_exception_releases_lock_and_resets_cron_flag(tmp_path, monkeypatch) -> None: + """The real scheduler cleanup path releases its lock and ContextVar.""" + import threading + from unittest.mock import MagicMock, patch + + import cron.scheduler as scheduler + + workdir = tmp_path / "cron-workdir" + workdir.mkdir() + job = {"id": "r3-test", "name": "cleanup", "prompt": "hi", "workdir": str(workdir)} + + monkeypatch.delenv("HERMES_CRON_SESSION", raising=False) + assert HERMES_CRON_SESSION_CONTEXTVAR.get() is False + + # Check the writer lock is available before invoking run_job, then leave it + # free for run_job to acquire. + scheduler._terminal_cwd_lock.acquire_write() + scheduler._terminal_cwd_lock.release_write() + + real_info = scheduler.logger.info + + def raise_on_workdir_log(message, *args, **kwargs): + if isinstance(message, str) and "using workdir" in message: + raise RuntimeError("r3 cleanup probe") + return real_info(message, *args, **kwargs) + + with patch("cron.scheduler._hermes_home", tmp_path), \ + patch("cron.scheduler._resolve_origin", return_value=None), \ + patch("hermes_cli.env_loader.load_hermes_dotenv"), \ + patch("hermes_cli.env_loader.reset_secret_source_cache"), \ + patch.object(scheduler.logger, "info", side_effect=raise_on_workdir_log), \ + patch("hermes_state.SessionDB", return_value=MagicMock()): + result = scheduler.run_job(job) + + assert result[0] is False + assert HERMES_CRON_SESSION_CONTEXTVAR.get() is False + + # A leaked writer would block this acquisition indefinitely; the lock test + # uses the same bounded thread pattern for this synchronization primitive. + acquired = threading.Event() + + def acquire_and_release() -> None: + scheduler._terminal_cwd_lock.acquire_write() + try: + acquired.set() + finally: + scheduler._terminal_cwd_lock.release_write() + + thread = threading.Thread(target=acquire_and_release, daemon=True) + thread.start() + assert acquired.wait(timeout=1), "writer lock was leaked by run_job" + thread.join(timeout=1) + assert not thread.is_alive() + + +def test_run_job_handoff_propagates_cron_context_and_isolates_concurrent_thread( + tmp_path, monkeypatch +) -> None: + import threading + from unittest.mock import MagicMock, patch + + import cron.scheduler as scheduler + + workdir = tmp_path / "cron-workdir" + workdir.mkdir() + job = { + "id": "r2-handoff", + "name": "handoff", + "prompt": "probe", + "workdir": str(workdir), + } + worker_probe = [] + concurrent_probe = [] + + def probe_run_conversation(*args, **kwargs): + worker_probe.append(HERMES_CRON_SESSION_CONTEXTVAR.get()) + + def probe_concurrent_thread() -> None: + concurrent_probe.append(HERMES_CRON_SESSION_CONTEXTVAR.get()) + + thread = threading.Thread(target=probe_concurrent_thread) + thread.start() + thread.join(timeout=1) + assert not thread.is_alive() + raise RuntimeError("r2 handoff probe") + + monkeypatch.delenv("HERMES_CRON_SESSION", raising=False) + assert HERMES_CRON_SESSION_CONTEXTVAR.get() is False + scheduler._terminal_cwd_lock.acquire_write() + scheduler._terminal_cwd_lock.release_write() + + agent = MagicMock() + agent.run_conversation.side_effect = probe_run_conversation + with patch("cron.scheduler._hermes_home", tmp_path), \ + patch("cron.scheduler._resolve_origin", return_value=None), \ + patch("hermes_cli.env_loader.load_hermes_dotenv"), \ + patch("hermes_cli.env_loader.reset_secret_source_cache"), \ + patch( + "hermes_cli.runtime_provider.resolve_runtime_provider", + return_value={"provider": "test", "model": "test", "api_key": "key"}, + ), \ + patch("hermes_state.SessionDB", return_value=MagicMock()), \ + patch("run_agent.AIAgent", return_value=agent): + result = scheduler.run_job(job) + + assert result[0] is False + assert worker_probe == [True] + assert concurrent_probe == [False] + assert HERMES_CRON_SESSION_CONTEXTVAR.get() is False diff --git a/tests/tools/test_execute_code_approval_cluster.py b/tests/tools/test_execute_code_approval_cluster.py index 100ed7e02f58..c8819eaaeab6 100644 --- a/tests/tools/test_execute_code_approval_cluster.py +++ b/tests/tools/test_execute_code_approval_cluster.py @@ -119,6 +119,10 @@ def gw_session(monkeypatch): session_key = "cluster-test-session" token = A.set_current_session_key(session_key) + execute_code_aliases = A._approval_key_aliases("execute_code") + with A._lock: + permanent_snapshot = set(A._permanent_approved) + A._permanent_approved.difference_update(execute_code_aliases) with A._lock: A._gateway_queues.pop(session_key, None) A._gateway_notify_cbs.pop(session_key, None) @@ -127,6 +131,8 @@ def gw_session(monkeypatch): finally: A.reset_current_session_key(token) with A._lock: + A._permanent_approved.clear() + A._permanent_approved.update(permanent_snapshot) A._gateway_queues.pop(session_key, None) A._gateway_notify_cbs.pop(session_key, None) @@ -177,6 +183,21 @@ def test_guard_headless_local_approved(monkeypatch): assert A.check_execute_code_guard("import os", "local")["approved"] is True +def test_guard_permanent_allowlist_is_isolated(monkeypatch): + monkeypatch.delenv("HERMES_GATEWAY_SESSION", raising=False) + snapshot = set(A._permanent_approved) + try: + A.approve_permanent("execute_code") + assert A.check_execute_code_guard("import os", "local") == { + "approved": True, + "message": None, + } + finally: + with A._lock: + A._permanent_approved.clear() + A._permanent_approved.update(snapshot) + + def test_guard_cron_deny_blocks(monkeypatch): monkeypatch.setattr(A, "_YOLO_MODE_FROZEN", False) monkeypatch.setenv("HERMES_CRON_SESSION", "1") diff --git a/tools/approval.py b/tools/approval.py index 5db5065f9063..6f6744c33990 100644 --- a/tools/approval.py +++ b/tools/approval.py @@ -93,6 +93,18 @@ def _is_interactive_cli() -> bool: return env_var_enabled("HERMES_INTERACTIVE") +def _is_cron_session() -> bool: + """Return whether the current task is a cron job.""" + try: + from gateway.session_context import HERMES_CRON_SESSION_CONTEXTVAR + + if HERMES_CRON_SESSION_CONTEXTVAR.get(): + return True + except Exception: + pass + return env_var_enabled("HERMES_CRON_SESSION") + + def _fire_approval_hook(hook_name: str, **kwargs) -> None: """Invoke a plugin lifecycle hook for the approval system. @@ -238,7 +250,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 +2911,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, @@ -3427,7 +3439,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) @@ -3878,7 +3890,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,