fix(cron): harden lifecycle guard against multi-line payloads, binaries, and killall

cron/lifecycle_guard.py

@@ -74,8 +74,8 @@ class GatewayLifecycleBlocked(ValueError):
     r"|(?:systemctl\s+(?:-\S+\s+)*(?:restart|stop|start)\b[^\n]*\bhermes[.\-]?gateway)"
     # Branch D: pkill / kill targeting the hermes gateway process. Both
     # token orders because real reproductions show both.
-    r"|(?:p?kill\b[^\n]*\bhermes\b[^\n]*\bgateway)"
-    r"|(?:p?kill\b[^\n]*\bgateway\b[^\n]*\bhermes)"
+    r"|(?:p?kill[a-z]*\b[^\n]*\bhermes\b[^\n]*\bgateway)"
+    r"|(?:p?kill[a-z]*\b[^\n]*\bgateway\b[^\n]*\bhermes)"
 )
 
 
@@ -105,17 +105,92 @@ def contains_gateway_lifecycle_command(text: str) -> bool:
 _MAX_REFERENCED_SCRIPT_BYTES = 1024 * 1024
 _MAX_REFERENCED_SCRIPT_DEPTH = 8
 _CONTROL_CHARS = frozenset(";&|()")
+# Header bytes inspected to tell a script from a binary.
+_SCRIPT_HEADER_BYTES = 4096
+
+
+def _looks_like_script(path: Path) -> bool:
+    """Return True when *path* is a local file plausibly holding a shell
+    script: its header contains no NUL bytes (real scripts are NUL-free
+    text; shebang or not). Binaries — ELF/Mach-O/PE, UTF-16, compressed
+    data — have NUL bytes and are not shell scripts, so the reference walk
+    skips them instead of decoding machine code as text (#76762, #77173).
+    """
+    try:
+        with open(path, "rb") as fh:
+            head = fh.read(_SCRIPT_HEADER_BYTES)
+    except (OSError, ValueError):
+        return False
+    return bool(head) and b"\x00" not in head
 
 
 
 
 _ReadRemoteScriptFn = Callable[[str], Optional[str]]
 
 
+def _split_logical_lines(text: str) -> Iterator[str]:
+    """Split *text* into logical lines at newlines that are OUTSIDE quotes.
+
+    Plain ``splitlines()`` destroys quote context: a multi-line payload (a
+    ``python -c "...`` string spanning newlines, a heredoc body,
+    ``$'...'``) has its quotes terminated at each physical newline, so
+    unquoted fragments of the payload — notably parenthesized paths like
+    ``open('/x/y.json')`` — were promoted to segment executables and
+    treated as referenced shell scripts. The referenced file was then read
+    and scanned: an innocent command touching any text file that merely
+    *mentions* a lifecycle command got blocked, and a binary decoded as
+    text crashed the scan (#76762 sibling). Splitting only at *unquoted*
+    newlines keeps quoted strings intact across physical lines, which is
+    what the shell does, while still giving every real command line its
+    own segment (so a line-starting ``./script.sh`` stays an executable).
+    """
+    out: list[str] = []
+    quote: Optional[str] = None
+    i, n = 0, len(text)
+    while i < n:
+        ch = text[i]
+        if quote == "'":
+            out.append(ch)
+            if ch == "'":
+                quote = None
+            i += 1
+            continue
+        if quote == '"':
+            out.append(ch)
+            if ch == "\\" and i + 1 < n:
+                out.append(text[i + 1])
+                i += 2
+                continue
+            if ch == '"':
+                quote = None
+            i += 1
+            continue
+        if ch in ("'", '"'):
+            quote = ch
+            out.append(ch)
+            i += 1
+            continue
+        if ch == "\\" and i + 1 < n:
+            out.append(ch)
+            out.append(text[i + 1])
+            i += 2
+            continue
+        if ch == "\n":
+            yield "".join(out)
+            out = []
+            i += 1
+            continue
+        out.append(ch)
+        i += 1
+    if out or not text:
+        yield "".join(out)
+
+
 def _iter_command_segments(command: str) -> Iterator[list[str]]:
     """Yield shell-tokenized command segments, honoring quotes and comments."""
     normalized = command.replace("\\\n", "")
-    for line in normalized.splitlines() or [normalized]:
+    for line in _split_logical_lines(normalized):
         try:
             lexer = shlex.shlex(
                 line,
@@ -226,7 +301,19 @@ def _iter_referenced_shell_scripts(
         # (#77131). Skip pure-separator tokens.
         if executable.strip("/"):
             if "/" in executable or executable.endswith((".sh", ".bash", ".zsh")):
-                yield _resolve_terminal_script_path(executable, cwd)
+                candidate = _resolve_terminal_script_path(executable, cwd)
+                # Only follow candidates that are plausibly scripts. Local
+                # binaries (anything with NUL bytes in its header) are not
+                # shell scripts: skipping them avoids the wasted 1MB read
+                # AND the decoded-binary recursion that crashed the guard
+                # (#76762, #77173). Paths that don't exist locally are
+                # still yielded so a remote backend can fetch them via
+                # read_remote_script.
+                try:
+                    if not candidate.exists() or _looks_like_script(candidate):
+                        yield candidate
+                except (OSError, ValueError):
+                    continue
 
 
 def _iter_shell_command_payloads(command: str) -> Iterator[str]:
@@ -258,17 +345,28 @@ 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):
+        # OSError: unreadable/long paths. ValueError: embedded NUL byte —
+        # a guarded path must never crash the guard (#76762).
         return None, False
     try:
         metadata = os.fstat(descriptor)
+        if stat.S_ISDIR(metadata.st_mode):
+            # A directory can never be a shell script (the shell rejects
+            # executing it), and bare "/" separators / pathlib division
+            # operators resolve here. "Nothing to scan", not unsafe —
+            # otherwise innocent commands touching directories get blocked
+            # (#77173).
+            return None, False
         if not stat.S_ISREG(metadata.st_mode):
+            # FIFOs/devices/sockets can stream script content we cannot
+            # inspect — fail closed.
             return None, True
         # Read a bounded chunk first — even for oversized files, the first
         # chunk tells us if this is a binary (NUL bytes) that should be
         # skipped as "nothing to scan" rather than failing closed (#76762).
         data = os.read(descriptor, _MAX_REFERENCED_SCRIPT_BYTES + 1)
-    except OSError:
+    except (OSError, ValueError):
         return None, False
     finally:
         os.close(descriptor)
@@ -332,6 +430,12 @@ def _contains_unsafe_gateway_action(
         # Relative references inside a script resolve against that script's
         # directory, not the original command's cwd.
         script_dir = _resolve_script_directory(str(resolved)) or cwd
+        if script_text and "\x00" in script_text:
+            # A remote backend may have returned a binary's bytes decoded
+            # as text (NUL is valid UTF-8). Tokenizing that would feed
+            # NUL-bearing paths into os.open and crash the guard (#76762
+            # sibling); there is nothing meaningful to scan.
+            continue
         if script_text and _contains_unsafe_gateway_action(
             script_text,
             cwd=script_dir,
@@ -383,9 +487,10 @@ def _resolve_script_path(script_path: str) -> Path:
 def _read_script_for_scanning(script_path: str) -> str:
     """Read a cron script with the bounded terminal-script scanner.
 
-    Non-regular or oversized inputs fail closed by returning a lifecycle-shaped
-    sentinel, while missing/unreadable paths remain empty so ordinary scheduler
-    path validation can report them.
+    Non-regular (FIFO/device/socket) or oversized inputs fail closed by
+    returning a lifecycle-shaped sentinel; directories are "nothing to scan"
+    (a directory can never be a script) and missing/unreadable paths remain
+    empty so ordinary scheduler path validation can report them.
     """
     script_text, unsafe = _read_referenced_script(_resolve_script_path(script_path))
     if unsafe:

tests/hermes_cli/test_gateway_restart_loop.py

@@ -857,3 +857,122 @@ def test_cron_nested_wrapper_script_is_scanned(self, tmp_path, capsys, monkeypat
         assert rc == 1
         out = capsys.readouterr().out
         assert "Blocked" in out
+
+
+class TestLifecycleGuardTokenizerAndBinaryHandling:
+    """Regression tests for the guard's command-tokenization and
+    referenced-script classification:
+
+    - B2: multi-line payloads (python -c / heredocs) must not promote
+      parenthesized path arguments to "referenced scripts" and scan the
+      files they name — an innocent command touching a text file that
+      merely mentions a lifecycle command was blocked.
+    - B1: NUL-bearing paths (from a direct payload or from a remote
+      backend returning a binary decoded as text) must never crash the
+      guard with ValueError: embedded null byte.
+    - B4: directory candidates are "nothing to scan", not unsafe.
+    - B3: killall/pkill variants targeting hermes-gateway are detected.
+    """
+
+    def test_multiline_python_payload_path_not_scanned_as_script(self, tmp_path):
+        from cron.lifecycle_guard import (
+            contains_gateway_lifecycle_command_or_referenced_script,
+        )
+        datafile = tmp_path / "session.json"
+        datafile.write_text("note: hermes gateway restart was blocked here\n")
+        cmd = (
+            "python3 -c \"\n"
+            f"import json\nwith open('{datafile}') as f:\n    d = json.load(f)\n"
+            "\""
+        )
+        assert contains_gateway_lifecycle_command_or_referenced_script(
+            cmd, cwd=str(tmp_path)
+        ) is False
+
+    def test_heredoc_payload_with_paths_not_scanned_as_scripts(self, tmp_path):
+        from cron.lifecycle_guard import (
+            contains_gateway_lifecycle_command_or_referenced_script,
+        )
+        cfg = tmp_path / "config.yaml"
+        cfg.write_text("model: hermes gateway restart\n")
+        cmd = (
+            "/usr/bin/python3 - <<'EOF'\n"
+            f"p = '{cfg}'\n"
+            "with open(p) as f:\n"
+            "    data = f.read()\n"
+            "EOF\n"
+        )
+        assert contains_gateway_lifecycle_command_or_referenced_script(
+            cmd, cwd=str(tmp_path)
+        ) is False
+
+    def test_nul_path_in_payload_does_not_crash(self):
+        from cron.lifecycle_guard import (
+            contains_gateway_lifecycle_command_or_referenced_script,
+        )
+        cmd = "bash -c '/tmp/a\x00b/script.sh x'"
+        assert contains_gateway_lifecycle_command_or_referenced_script(cmd) is False
+
+    def test_remote_read_returning_binary_text_does_not_crash(self, tmp_path):
+        from cron.lifecycle_guard import (
+            contains_gateway_lifecycle_command_or_referenced_script,
+        )
+        # A remote backend returning a binary decoded as text: NUL bytes
+        # survive as valid UTF-8, so the recursion must skip the text
+        # rather than tokenize NUL-bearing paths into os.open.
+        binary_text = b"\x7fELF\x02\x01\x01\x00" + b"\x00" * 64 + b"/tmp/x\x00y.sh\n"
+
+        def read_remote(script_path):
+            return binary_text.decode("utf-8", errors="replace")
+
+        assert contains_gateway_lifecycle_command_or_referenced_script(
+            "/nonexistent/runner --version",
+            cwd=str(tmp_path),
+            read_remote_script=read_remote,
+        ) is False
+
+    def test_directory_candidate_not_blocked(self, tmp_path):
+        from cron.lifecycle_guard import (
+            contains_gateway_lifecycle_command_or_referenced_script,
+        )
+        assert contains_gateway_lifecycle_command_or_referenced_script(
+            f"/bin/bash {tmp_path}", cwd=str(tmp_path)
+        ) is False
+
+    def test_killall_and_pkill_gateway_detected(self):
+        from cron.lifecycle_guard import contains_gateway_lifecycle_command
+        for text in [
+            "killall hermes-gateway",
+            "killall -9 hermes-gateway",
+            "pkill -f hermes-gateway",
+            "pkill hermes gateway",
+        ]:
+            assert contains_gateway_lifecycle_command(text), f"Should match: {text!r}"
+
+    def test_killall_without_gateway_not_blocked(self):
+        from cron.lifecycle_guard import contains_gateway_lifecycle_command
+        assert contains_gateway_lifecycle_command("killall chrome") is False
+
+    def test_lifecycle_commands_still_detected_after_tokenizer_change(self):
+        from cron.lifecycle_guard import (
+            contains_gateway_lifecycle_command_or_referenced_script,
+        )
+        for cmd in [
+            "hermes gateway restart",
+            "systemctl restart hermes-gateway",
+            "launchctl kickstart gui/501/ai.hermes.gateway",
+            "bash -c 'hermes gateway restart'",
+        ]:
+            assert contains_gateway_lifecycle_command_or_referenced_script(cmd), cmd
+
+    def test_shell_c_payload_with_paths_still_scanned(self, tmp_path):
+        """The single-pass tokenizer must keep the -c payload walk intact."""
+        from cron.lifecycle_guard import (
+            contains_gateway_lifecycle_command_or_referenced_script,
+        )
+        evil = tmp_path / "evil.sh"
+        evil.write_text("#!/bin/bash\nhermes gateway stop\n")
+        cmd = f"/bin/bash -c '/bin/bash {evil}'"
+        assert contains_gateway_lifecycle_command_or_referenced_script(
+            cmd, cwd=str(tmp_path)
+        ) is True

tools/terminal_tool.py

@@ -2534,6 +2534,12 @@ def _read_script_in_env(script_path: str) -> Optional[str]:
                         if stat.S_ISREG(metadata.st_mode) and metadata.st_size <= 1024 * 1024:
                             data = local_path.read_bytes()
                             if len(data) <= 1024 * 1024:
+                                # Binaries decode to NUL-laden text and the
+                                # recursive scan would crash on NUL-bearing
+                                # paths — mirror _read_referenced_script's
+                                # "nothing to scan" semantics (#76762).
+                                if b"\x00" in data[:4096]:
+                                    return None
                                 return data.decode("utf-8", errors="replace")
                 except Exception:
                     pass