fix(guard): block launchctl submit/bootstrap and wrapped-script gateway restarts

cron/lifecycle_guard.py

@@ -23,19 +23,24 @@
 rate without preventing the actual foot-gun, which requires a real
 command shape.
 
-This is a defence-in-depth layer.  ``tools/terminal_tool.py`` already
-blocks these commands at *execution* time when ``_HERMES_GATEWAY=1``, and
-``hermes gateway stop|restart`` refuse to self-target from inside the
-gateway.  Blocking at *creation* time as well means the agent gets an
-immediate, informative rejection instead of scheduling a job that will
-only fail (silently) when it fires.
+This is a defence-in-depth layer.  ``tools/terminal_tool.py`` blocks direct
+commands and shell scripts they reference when ``_HERMES_GATEWAY=1``. It also
+rejects ``launchctl submit`` in gateway sessions because launchd treats that
+primitive as a persistent KeepAlive job, not a one-shot task. ``hermes gateway
+stop|restart`` separately refuse to self-target from inside the gateway.
+Blocking cron specs at creation time as well means the agent gets an immediate,
+informative rejection instead of scheduling a job that will only fail
+(silently) when it fires.
 """
 
 from __future__ import annotations
 
+import os
 import re
+import shlex
+import stat
 from pathlib import Path
-from typing import Optional
+from typing import Callable, Iterator, Optional
 
 
 class GatewayLifecycleBlocked(ValueError):
@@ -56,7 +61,15 @@ class GatewayLifecycleBlocked(ValueError):
     # labels look like `ai.hermes.gateway` / `hermes-gateway`. Requiring the
     # gateway identifier prevents blocking unrelated hermes services (e.g.
     # `launchctl unload ai.hermes.update-checker.plist`).
-    r"|(?:launchctl\s+(?:kickstart|unload|load|stop|restart)\b[^\n]*\bhermes[.\-]?gateway)"
+    # `submit` and `bootstrap` are included alongside the direct verbs
+    # (kickstart/etc.): `launchctl submit -l ai.hermes.gateway-<suffix> --
+    # <helper-script>` (or `launchctl bootstrap gui/<uid> <plist>`) creates
+    # a NEW keepalive job wrapping an arbitrary helper, which is how a
+    # blocked direct restart/kill gets laundered into a persistent restart
+    # loop instead (#62891) — same foot-gun, indirect shape. Neutral-label
+    # submissions that dodge this text anchor are caught separately by
+    # `contains_launchctl_submit_command` (execution-aware, label-independent).
+    r"|(?:launchctl\s+(?:kickstart|unload|load|stop|restart|submit|bootstrap)\b[^\n]*\bhermes[.\-]?gateway)"
     # Branch C: systemctl ops on a hermes-gateway unit.
     r"|(?:systemctl\s+(?:-\S+\s+)*(?:restart|stop|start)\b[^\n]*\bhermes[.\-]?gateway)"
     # Branch D: pkill / kill targeting the hermes gateway process. Both
@@ -66,11 +79,268 @@ class GatewayLifecycleBlocked(ValueError):
 )
 
 
+# A backslash immediately followed by a newline is a POSIX shell line
+# continuation — the shell joins the two lines before parsing. Every branch
+# above uses `[^\n]*` between its verb and the gateway identifier so the
+# match can't span unrelated lines of a longer cron prompt/script, but that
+# also means a real multi-line shell invocation split across continuation
+# lines (e.g. `launchctl submit \` / `  -l ai.hermes.gateway-... \` / `  -- ...`,
+# the exact reported shape in #62891) would otherwise slip past. Collapse
+# continuations to a single space before matching, mirroring what the shell
+# itself does, rather than loosening `[^\n]*` and risking false positives
+# across genuinely separate lines.
+_SHELL_LINE_CONTINUATION = re.compile(r"\\\r?\n[ \t]*")
+
+
 def contains_gateway_lifecycle_command(text: str) -> bool:
     """Return True if *text* contains a gateway lifecycle command pattern."""
     if not text:
         return False
-    return bool(_GATEWAY_LIFECYCLE_PATTERN.search(text))
+    normalized = _SHELL_LINE_CONTINUATION.sub(" ", text)
+    return bool(_GATEWAY_LIFECYCLE_PATTERN.search(normalized))
+
+
+_SHELL_EXECUTABLES = frozenset({"sh", "bash", "dash", "ksh", "zsh"})
+_SHELL_OPTIONS_WITH_VALUES = frozenset({"-O", "+O", "-o", "+o"})
+_MAX_REFERENCED_SCRIPT_BYTES = 1024 * 1024
+_MAX_REFERENCED_SCRIPT_DEPTH = 8
+_CONTROL_CHARS = frozenset(";&|()")
+
+
+
+
+_ReadRemoteScriptFn = Callable[[str], Optional[str]]
+
+
+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]:
+        try:
+            lexer = shlex.shlex(
+                line,
+                posix=True,
+                punctuation_chars=";&|()",
+            )
+            lexer.whitespace_split = True
+            lexer.commenters = "#"
+            tokens = list(lexer)
+        except ValueError:
+            continue
+
+        segment: list[str] = []
+        for token in tokens:
+            if token and set(token) <= _CONTROL_CHARS:
+                if segment:
+                    yield segment
+                    segment = []
+                continue
+            segment.append(token)
+        if segment:
+            yield segment
+
+
+def _command_token_index(segment: list[str]) -> Optional[int]:
+    """Return the executable token index after simple env assignments."""
+    for index, token in enumerate(segment):
+        if re.match(r"^[A-Za-z_][A-Za-z0-9_]*=", token):
+            continue
+        return index
+    return None
+
+
+def contains_launchctl_submit_command(command: str) -> bool:
+    """Detect an executed ``launchctl submit``/``bootstrap``, not quoted text.
+
+    Label-independent by design: the label of a submitted/bootstrapped job is
+    chosen by whoever writes it, so a neutral name (``ai.hermes.svc-reload-tmp``)
+    defeats any label-anchored regex (#62891, second reproduction). Both verbs
+    register a NEW persistent launchd job (``submit`` jobs get KeepAlive
+    semantics; ``bootstrap`` loads an arbitrary plist), which is never safe to
+    do from inside the gateway process.
+    """
+    for segment in _iter_command_segments(command):
+        index = _command_token_index(segment)
+        if index is None:
+            continue
+        if Path(segment[index]).name == "launchctl":
+            arguments = segment[index + 1 :]
+            if arguments and arguments[0].lower() in {"submit", "bootstrap"}:
+                return True
+    return False
+
+
+def _resolve_terminal_script_path(candidate: str, cwd: Optional[str]) -> Path:
+    path = Path(candidate).expanduser()
+    if not path.is_absolute():
+        path = Path(cwd or Path.cwd()) / path
+    return path
+
+
+def _iter_referenced_shell_scripts(
+    command: str,
+    *,
+    cwd: Optional[str] = None,
+) -> Iterator[Path]:
+    """Yield scripts executed directly or through a POSIX shell."""
+    for segment in _iter_command_segments(command):
+        index = _command_token_index(segment)
+        if index is None:
+            continue
+        executable = segment[index]
+        executable_name = Path(executable).name
+
+        if executable_name in {".", "source"}:
+            if len(segment) > index + 1:
+                yield _resolve_terminal_script_path(segment[index + 1], cwd)
+            continue
+
+        if executable_name in _SHELL_EXECUTABLES:
+            arguments = segment[index + 1 :]
+            arg_index = 0
+            while arg_index < len(arguments):
+                argument = arguments[arg_index]
+                if argument == "--":
+                    arg_index += 1
+                    break
+                if argument in {"-c", "--command"}:
+                    break
+                if argument in _SHELL_OPTIONS_WITH_VALUES:
+                    arg_index += 2
+                    continue
+                if argument.startswith("-"):
+                    arg_index += 1
+                    continue
+                break
+            if arg_index < len(arguments) and arguments[arg_index] not in {
+                "-c",
+                "--command",
+            }:
+                yield _resolve_terminal_script_path(arguments[arg_index], cwd)
+            continue
+
+        if "/" in executable or executable.endswith((".sh", ".bash", ".zsh")):
+            yield _resolve_terminal_script_path(executable, cwd)
+
+
+def _iter_shell_command_payloads(command: str) -> Iterator[str]:
+    """Yield code passed through ``sh|bash|... -c`` for recursive scanning."""
+    for segment in _iter_command_segments(command):
+        index = _command_token_index(segment)
+        if index is None or Path(segment[index]).name not in _SHELL_EXECUTABLES:
+            continue
+        arguments = segment[index + 1 :]
+        for arg_index, argument in enumerate(arguments[:-1]):
+            if argument in {"-c", "--command"}:
+                yield arguments[arg_index + 1]
+                break
+
+
+def _resolve_script_directory(script_path: str) -> Optional[str]:
+    """Return the directory *script_path* resolves to, handling relative names."""
+    try:
+        path = _resolve_script_path(script_path)
+        if path.is_absolute():
+            return str(path.parent)
+    except Exception:
+        pass
+    return None
+
+
+def _read_referenced_script(path: Path) -> tuple[Optional[str], bool]:
+    """Return ``(text, unsafe)`` using bounded, regular-file-only reads."""
+    flags = os.O_RDONLY | getattr(os, "O_NONBLOCK", 0)
+    try:
+        descriptor = os.open(path, flags)
+    except OSError:
+        return None, False
+    try:
+        metadata = os.fstat(descriptor)
+        if not stat.S_ISREG(metadata.st_mode):
+            return None, True
+        if metadata.st_size > _MAX_REFERENCED_SCRIPT_BYTES:
+            return None, True
+        data = os.read(descriptor, _MAX_REFERENCED_SCRIPT_BYTES + 1)
+    except OSError:
+        return None, False
+    finally:
+        os.close(descriptor)
+    if len(data) > _MAX_REFERENCED_SCRIPT_BYTES:
+        return None, True
+    return data.decode("utf-8", errors="replace"), False
+
+
+def _contains_unsafe_gateway_action(
+    command: str,
+    *,
+    cwd: Optional[str],
+    depth: int,
+    visited: set[Path],
+    read_remote_script: Optional[_ReadRemoteScriptFn] = None,
+) -> bool:
+    if contains_gateway_lifecycle_command(command) or contains_launchctl_submit_command(
+        command
+    ):
+        return True
+    if depth >= _MAX_REFERENCED_SCRIPT_DEPTH:
+        return True
+
+    for payload in _iter_shell_command_payloads(command):
+        if _contains_unsafe_gateway_action(
+            payload,
+            cwd=cwd,
+            depth=depth + 1,
+            visited=visited,
+            read_remote_script=read_remote_script,
+        ):
+            return True
+
+    for script_path in _iter_referenced_shell_scripts(command, cwd=cwd):
+        try:
+            resolved = script_path.resolve(strict=False)
+        except OSError:
+            resolved = script_path
+        if resolved in visited:
+            continue
+        visited.add(resolved)
+        script_text, unsafe = _read_referenced_script(script_path)
+        if unsafe:
+            return True
+        if script_text is None and read_remote_script is not None:
+            # Local path missing; try the remote backend if one is available.
+            script_text = read_remote_script(str(script_path))
+        if not script_text:
+            continue
+        # 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 _contains_unsafe_gateway_action(
+            script_text,
+            cwd=script_dir,
+            depth=depth + 1,
+            visited=visited,
+            read_remote_script=read_remote_script,
+        ):
+            return True
+    return False
+
+
+def contains_gateway_lifecycle_command_or_referenced_script(
+    command: str,
+    *,
+    cwd: Optional[str] = None,
+    read_remote_script: Optional[_ReadRemoteScriptFn] = None,
+) -> bool:
+    """Detect lifecycle/submit commands, including bounded nested scripts."""
+    return _contains_unsafe_gateway_action(
+        command,
+        cwd=cwd,
+        depth=0,
+        visited=set(),
+        read_remote_script=read_remote_script,
+    )
+
+
 
 
 def _resolve_script_path(script_path: str) -> Path:
@@ -93,20 +363,16 @@ def _resolve_script_path(script_path: str) -> Path:
 
 
 def _read_script_for_scanning(script_path: str) -> str:
-    """Read a script file for lifecycle-pattern scanning.
+    """Read a cron script with the bounded terminal-script scanner.
 
-    Decodes with ``errors="replace"`` so binary or non-UTF-8 content does not
-    silently bypass the check — a plain text-mode read raises
-    ``UnicodeDecodeError`` on such files, and swallowing that error would let
-    an attacker hide the command in binary noise.  Returns an empty string
-    only when the file cannot be read at all.
+    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.
     """
-    try:
-        return _resolve_script_path(script_path).read_bytes().decode(
-            "utf-8", errors="replace"
-        )
-    except OSError:
-        return ""
+    script_text, unsafe = _read_referenced_script(_resolve_script_path(script_path))
+    if unsafe:
+        return "hermes gateway restart"
+    return script_text or ""
 
 
 def check_gateway_lifecycle(
@@ -131,10 +397,14 @@ def check_gateway_lifecycle(
         if script_text:
             combined = f"{combined}\n{script_text}"
 
-    if contains_gateway_lifecycle_command(combined):
+    script_dir = _resolve_script_directory(script) if script else None
+    if contains_gateway_lifecycle_command_or_referenced_script(
+        combined,
+        cwd=script_dir,
+    ):
         raise GatewayLifecycleBlocked(
-            "Blocked: cron job contains a gateway lifecycle command "
-            "(restart/stop/kill). This is blocked to prevent agent-driven "
+            "Blocked: cron job contains a gateway lifecycle command or persistent "
+            "launchctl submit operation. This is blocked to prevent agent-driven "
             "SIGTERM-respawn loops under launchd/systemd supervision "
             "(#30719). Run `hermes gateway restart` from a shell outside "
             "the running gateway instead."