fix(cron): scope lifecycle guard to real shell scripts; per-job cron flag via ContextVar

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

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.

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).
 #

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

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")