diff --git a/cron/lifecycle_guard.py b/cron/lifecycle_guard.py index 21aa0bf3abcf8..eeec888da39cd 100644 --- a/cron/lifecycle_guard.py +++ b/cron/lifecycle_guard.py @@ -45,6 +45,7 @@ import re import shlex import stat +import threading from pathlib import Path from typing import Callable, Iterator, Optional @@ -55,6 +56,16 @@ class GatewayLifecycleBlocked(ValueError): """Raised when a cron job spec contains a gateway-lifecycle command.""" +# Stable marker attached to every terminal-tool result that was refused by +# THIS guard. It exists so a consumer can distinguish "the guard refused" +# from "the command ran and failed" WITHOUT string-matching prose that is +# free to change. The execute_code sandbox stub keys on it to raise instead +# of handing the caller a result that looks like an ordinary non-zero exit +# (a script that does not inspect the returned dict otherwise completes +# silently with rc 0, which reads as success — measured 2026-09-20). +GATEWAY_LIFECYCLE_BLOCK_MARKER = "gateway_lifecycle_guard" + + # Shell-level command shapes that target the gateway lifecycle. Each branch # is anchored on a concrete command identifier so a match can only fire on # actual shell-command-shaped strings, not on prose. @@ -1100,15 +1111,20 @@ def contains_launchctl_submit_command(command: str) -> bool: if _executable_name(segment[index]) == "launchctl": arguments = segment[index + 1 :] if arguments and arguments[0].lower() in {"submit", "bootstrap"}: - if arguments[0].lower() == "bootstrap" and _bootstrap_targets_sibling_plist( - arguments[1:] + if arguments[0].lower() == "bootstrap" and ( + _bootstrap_targets_sibling_plist(arguments[1:]) + or _bootstrap_targets_readable_non_gateway_plist(arguments[1:]) ): # `launchctl bootstrap gui/ .plist` loads a # DIFFERENT gateway's job. It cannot restart-loop this # process, and it is exactly how a break-glass profile - # brings a wedged sibling back up. `submit` stays blocked - # unconditionally (its label is attacker-chosen, so the - # text proves nothing about the job it creates). + # brings a wedged sibling back up. The second predicate + # covers the much more common case: bootstrapping a plist + # that is ALREADY on disk and is not a gateway job at all + # (reloading a fleet daemon after editing it). `submit` + # stays blocked unconditionally (its label is + # attacker-chosen, so the text proves nothing about the + # job it creates). continue return True return False @@ -1139,6 +1155,200 @@ def _bootstrap_targets_sibling_plist(arguments: list[str]) -> bool: return True +# A launchd plist is a small configuration file; the real ones on this fleet +# are well under 8 KiB. Bound the read so a hostile/oversized file can never +# be streamed into the guard, mirroring `_MAX_REFERENCED_SCRIPT_BYTES`. +_MAX_PLIST_BYTES = 256 * 1024 + +# Tokens that mean "this job IS a hermes gateway" when they appear in a +# plist's Program / ProgramArguments. The label check alone is not enough: a +# plist can carry a neutral Label (`ai.hermes.svc-reload-tmp`) while its argv +# runs `python -m hermes_cli.main gateway run`, which is the #62891 laundering +# shape with a file instead of a `submit` line. +_PLIST_GATEWAY_ARGV_MARKERS = ("hermes_cli.main", "hermes-gateway", "hermes_gateway") + +# `_plist_argv_runs_gateway_lifecycle` scans a plist's argv with the same +# scanner that resolves `launchctl bootstrap ` — so a plist whose argv +# bootstraps another plist re-enters it. Bound the nesting per thread (the +# guard runs on the gateway's request threads) and fail CLOSED at the bound. +_MAX_PLIST_ARGV_SCAN_DEPTH = 2 +_PLIST_ARGV_SCAN_STATE = threading.local() + + +def _plist_program_tokens(payload: dict) -> list[str]: + """Flatten a plist's Program / ProgramArguments into scannable strings.""" + tokens: list[str] = [] + program = payload.get("Program") + if isinstance(program, str): + tokens.append(program) + arguments = payload.get("ProgramArguments") + if isinstance(arguments, (list, tuple)): + tokens.extend(item for item in arguments if isinstance(item, str)) + return tokens + + +def _plist_argv_runs_gateway_lifecycle(payload: dict) -> bool: + """True when loading this plist would EXECUTE a gateway-lifecycle command. + + ``_plist_declares_gateway_job`` asks whether the plist *is* a gateway job. + That is not the question the guard exists to answer: a plist with a wholly + neutral ``Label`` and a non-entrypoint argv can still run + + /bin/sh -c 'launchctl kickstart -k system/ai.hermes.gateway-daedalus' + + — the #62891 laundering shape with a file instead of a ``submit`` line, and + it needs no root (``launchctl bootstrap gui/ $TMPDIR/x.plist``). Run + the flattened argv through the lifecycle scanner that already handles both + inline commands and commands hidden in a referenced script. + + Each token is scanned on its own (a ``sh -c `` string is its own + command text, and a bare script path is a referenced script) AND joined (a + lifecycle command spelled across separate argv words). ``or`` of the two: + every extra match is a REFUSAL, which is the safe direction. + """ + tokens = _plist_program_tokens(payload) + if not tokens: + return False + depth = getattr(_PLIST_ARGV_SCAN_STATE, "depth", 0) + if depth >= _MAX_PLIST_ARGV_SCAN_DEPTH: + # A plist whose argv bootstraps another plist re-enters this scan. + # Two plists can reference each other, so bound it and fail CLOSED: + # a plist that bootstraps a plist is not the routine maintenance + # shape this allow-path exists for. + return True + _PLIST_ARGV_SCAN_STATE.depth = depth + 1 + try: + if any( + contains_gateway_lifecycle_command_or_referenced_script(token) + for token in tokens + ): + return True + return contains_gateway_lifecycle_command_or_referenced_script(" ".join(tokens)) + finally: + _PLIST_ARGV_SCAN_STATE.depth = depth + + +def _plist_declares_gateway_job(payload: dict) -> bool: + """True when *payload* describes, or drives, a hermes GATEWAY job. + + Three independent tells, any one is enough: + + * ``Label`` is a full hermes gateway label; or + * ``Program``/``ProgramArguments`` reference a hermes gateway entrypoint + (``hermes_cli.main ... gateway``, a ``hermes-gateway`` launcher, ...); or + * that same argv EXECUTES a gateway-lifecycle command, inline or via a + referenced script (see ``_plist_argv_runs_gateway_lifecycle``). + """ + label = payload.get("Label") + if not isinstance(label, str) or not label.strip(): + # No Label at all: launchd would reject it, and we cannot reason + # about what it registers. Treat as gateway-ish (fail closed). + return True + if _LAUNCHD_GATEWAY_LABEL_RE.fullmatch(_normalize_service_name(label)): + return True + tokens = [token.lower() for token in _plist_program_tokens(payload)] + if any(marker in token for token in tokens for marker in _PLIST_GATEWAY_ARGV_MARKERS): + return True + # `... -m hermes_cli.main gateway run` spelled across separate argv words. + if "gateway" in tokens and any("hermes" in token for token in tokens): + return True + return _plist_argv_runs_gateway_lifecycle(payload) + + +def _read_plist_label_payload(path: Path) -> Optional[dict]: + """Bounded, regular-file-only plist read. ``None`` means "cannot read". + + Fails closed on every ambiguity the caller must not paper over: cloud + placeholders (an evicted FileProvider read can hang the guard, #88052), + non-regular files, directories, oversized files, and anything ``plistlib`` + cannot parse into a mapping. + """ + import plistlib + + if _is_cloud_placeholder_path(path): + return None + try: + resolved = path.resolve(strict=False) + except (OSError, ValueError): + resolved = path + if _is_cloud_placeholder_path(resolved): + return None + flags = os.O_RDONLY | getattr(os, "O_NONBLOCK", 0) + try: + descriptor = os.open(path, flags) + except (OSError, ValueError): + return None + try: + metadata = os.fstat(descriptor) + if not stat.S_ISREG(metadata.st_mode): + # Directories (`launchctl bootstrap ` loads EVERY + # plist inside), FIFOs, devices: nothing we can read once and + # reason about. + return None + if metadata.st_size > _MAX_PLIST_BYTES: + return None + data = b"" + while len(data) <= _MAX_PLIST_BYTES: + chunk = os.read(descriptor, _MAX_PLIST_BYTES + 1 - len(data)) + if not chunk: + break + data += chunk + except OSError: + return None + finally: + os.close(descriptor) + if len(data) > _MAX_PLIST_BYTES: + return None + try: + payload = plistlib.loads(data) + except Exception: + # Malformed / not a plist at all. plistlib raises several unrelated + # exception types depending on format; none of them mean "safe". + return None + return payload if isinstance(payload, dict) else None + + +def _bootstrap_targets_readable_non_gateway_plist(arguments: list[str]) -> bool: + """True when every plist argument is an EXISTING, non-gateway plist. + + ``submit`` and ``bootstrap`` are otherwise handled label-independently + because a NEW job's label is chosen by whoever writes the command + (#62891). That reasoning holds for ``submit`` (pure text) and for + bootstrapping a path that does not exist yet — but NOT for a plist already + on disk: launchd reads ``Label`` out of that file, so the label is a + readable fact, not an attacker's claim. Refusing those blocked routine + fleet maintenance (reloading ``ai.hermes.fleetreview-router`` after a + ``plutil -replace`` edit) and pushed operators onto the deprecated, + label-gated ``launchctl load -w``. + + Deliberately strict — every condition must hold: + + 1. at least one argument looks like a plist path; + 2. no argument carries an unexpanded shell value (the path we read would + not be the path launchd loads); + 3. every plist argument resolves to a readable regular file within the + size bound; and + 4. none of those files declares a hermes gateway job (by ``Label`` or by + entrypoint). + """ + plists = [argument for argument in arguments if argument.lower().endswith(".plist")] + if not plists: + return False + for argument in arguments: + if _UNEXPANDED_SHELL_VALUE_RE.search(argument): + return False + for plist in plists: + candidate = _expand_candidate_path(plist) + if candidate is None: + return False + payload = _read_plist_label_payload(candidate) + if payload is None: + return False + if _plist_declares_gateway_job(payload): + return False + return True + + def _mask_data_sink_arguments(text: str) -> str: """Replace data-sink executables' arguments with a neutral placeholder. diff --git a/tests/cron/test_lifecycle_guard_bootstrap_existing_plist.py b/tests/cron/test_lifecycle_guard_bootstrap_existing_plist.py new file mode 100644 index 0000000000000..f079cb190711e --- /dev/null +++ b/tests/cron/test_lifecycle_guard_bootstrap_existing_plist.py @@ -0,0 +1,515 @@ +"""``launchctl bootstrap`` of an EXISTING non-gateway plist must be ALLOWED. + +``contains_launchctl_submit_command`` treats ``submit`` and ``bootstrap`` +label-INDEPENDENTLY because a NEW job's label is chosen by whoever writes the +command (#62891). That is correct for ``submit`` (the label is pure text) and +for ``bootstrap`` of a path that does not exist yet, but it is wrong for +``bootstrap`` of a plist that is ALREADY on disk: launchd reads the ``Label`` +key out of that file, so the label is not attacker-chosen at all — it is a +readable fact about the job that will be registered. + +Measured 2026-09-20 inside the supervised default gateway on the Studio: +reloading the fleetreview router after a ``plutil -replace`` edit — + + sudo launchctl bootout system/ai.hermes.fleetreview-router + sudo launchctl bootstrap system /Library/LaunchDaemons/ai.hermes.fleetreview-router.plist + +— was refused with the gateway-lifecycle block even though +``ai.hermes.fleetreview-router`` is not a gateway label. The workaround used +was the legacy, label-gated ``launchctl load -w``; that is a gap, not a design. + +Everything that cannot be READ stays blocked: a non-existent path, an +unreadable/oversized/malformed file, a path carrying an unexpanded shell +value, a plist whose ``Label`` IS a gateway label, a plist whose +``ProgramArguments`` invoke a hermes gateway entrypoint, and ``submit`` in +every shape. +""" + +from __future__ import annotations + +import os +import pathlib +import plistlib + +import pytest + +from cron import lifecycle_guard +from cron.lifecycle_guard import ( + contains_gateway_lifecycle_command_or_referenced_script, + contains_launchctl_submit_command, +) + + +SELF_LAUNCHD = "ai.hermes.gateway-aegis" + + +@pytest.fixture +def launchd_identity(monkeypatch): + """Pin self-identity to the Aegis launchd job, as launchd would.""" + monkeypatch.setenv("XPC_SERVICE_NAME", SELF_LAUNCHD) + monkeypatch.delenv("INVOCATION_ID", raising=False) + monkeypatch.setattr(lifecycle_guard, "_profile_derived_self_names", lambda: set()) + return SELF_LAUNCHD + + +@pytest.fixture +def no_identity(monkeypatch): + """No determinable identity at all — the guard must fail closed.""" + monkeypatch.delenv("XPC_SERVICE_NAME", raising=False) + monkeypatch.delenv("INVOCATION_ID", raising=False) + monkeypatch.setattr(lifecycle_guard, "_profile_derived_self_names", lambda: set()) + + +def _write_plist(path, payload) -> str: + """Write *payload* as a binary plist and return its path as a string.""" + with open(path, "wb") as handle: + plistlib.dump(payload, handle, fmt=plistlib.FMT_BINARY) + return str(path) + + +@pytest.fixture +def router_plist(tmp_path): + """The real fleetreview-router plist shape, minus the irrelevant keys.""" + return _write_plist( + tmp_path / "ai.hermes.fleetreview-router.plist", + { + "Label": "ai.hermes.fleetreview-router", + "ProgramArguments": [ + "/usr/local/libexec/fleetreview-router/venv/bin/" + "fleetreview-router-service", + "--live", + ], + "RunAtLoad": True, + }, + ) + + +# --------------------------------------------------------------------------- +# POSITIVE CONTROL — today's exact command shape must be ALLOWED +# --------------------------------------------------------------------------- + + +class TestExistingNonGatewayPlistAllowed: + def test_incident_command_shape_allowed(self, router_plist, launchd_identity): + """The 2026-09-20 reload that was refused.""" + text = f"sudo launchctl bootstrap system {router_plist}" + assert not contains_launchctl_submit_command(text) + assert not contains_gateway_lifecycle_command_or_referenced_script(text) + + def test_bootout_then_bootstrap_sequence_allowed( + self, router_plist, launchd_identity + ): + """Both halves of the reload, as actually typed.""" + text = ( + "sudo launchctl bootout system/ai.hermes.fleetreview-router && " + f"sudo launchctl bootstrap system {router_plist}" + ) + assert not contains_launchctl_submit_command(text) + assert not contains_gateway_lifecycle_command_or_referenced_script(text) + + def test_gui_domain_form_allowed(self, router_plist, launchd_identity): + text = f"launchctl bootstrap gui/501 {router_plist}" + assert not contains_launchctl_submit_command(text) + + def test_quoted_path_allowed(self, tmp_path, launchd_identity): + path = _write_plist( + tmp_path / "ai.hermes.cert watch.plist", + {"Label": "ai.hermes.cert-watch", "ProgramArguments": ["/bin/true"]}, + ) + text = f'launchctl bootstrap system "{path}"' + assert not contains_launchctl_submit_command(text) + + def test_allowed_without_self_identity(self, router_plist, no_identity): + """A non-gateway Label is safe even when we cannot name ourselves. + + The sibling-plist path needs self-identity (it compares labels); this + one does not — a Label that is not a gateway label at all cannot be + OUR gateway regardless of which gateway we are. + """ + text = f"launchctl bootstrap system {router_plist}" + assert not contains_launchctl_submit_command(text) + + +# --------------------------------------------------------------------------- +# NEGATIVE CONTROLS — everything unreadable or gateway-ish stays BLOCKED +# --------------------------------------------------------------------------- + + +class TestBootstrapStillBlocked: + def test_gateway_label_plist_blocked(self, tmp_path, launchd_identity): + """A plist whose Label IS our own gateway label.""" + path = _write_plist( + tmp_path / "some-name.plist", + {"Label": SELF_LAUNCHD, "ProgramArguments": ["/bin/true"]}, + ) + text = f"launchctl bootstrap gui/501 {path}" + assert contains_launchctl_submit_command(text) + assert contains_gateway_lifecycle_command_or_referenced_script(text) + + def test_new_file_gateway_suffix_label_blocked(self, tmp_path, launchd_identity): + """Filename is neutral, Label is `ai.hermes.gateway-foo` — blocked. + + The file basename is deliberately NOT a gateway label, so only reading + the Label key can catch this. + """ + path = _write_plist( + tmp_path / "reload-helper.plist", + {"Label": "ai.hermes.gateway-foo", "ProgramArguments": ["/bin/true"]}, + ) + text = f"launchctl bootstrap gui/501 {path}" + assert contains_launchctl_submit_command(text) + + def test_gateway_entrypoint_in_program_arguments_blocked( + self, tmp_path, launchd_identity + ): + """Neutral Label, but the job runs a hermes gateway.""" + path = _write_plist( + tmp_path / "ai.hermes.svc-reload-tmp.plist", + { + "Label": "ai.hermes.svc-reload-tmp", + "ProgramArguments": [ + "/Users/ace/.hermes/runtime/hermes-agent/venv/bin/python", + "-m", + "hermes_cli.main", + "gateway", + "run", + "--replace", + ], + }, + ) + text = f"launchctl bootstrap gui/501 {path}" + assert contains_launchctl_submit_command(text) + + def test_gateway_entrypoint_in_program_blocked(self, tmp_path, launchd_identity): + """Same, via the scalar ``Program`` key rather than the argv list.""" + path = _write_plist( + tmp_path / "ai.hermes.svc-other.plist", + { + "Label": "ai.hermes.svc-other", + "Program": "/Users/ace/.hermes/bin/hermes-gateway-launcher", + }, + ) + text = f"launchctl bootstrap gui/501 {path}" + assert contains_launchctl_submit_command(text) + + def test_gateway_entrypoint_split_across_argv_blocked( + self, tmp_path, launchd_identity + ): + """Neutral Label, no marker token — `hermes` and `gateway` as + SEPARATE argv words. + + A bare launcher (`/usr/local/bin/hermes gateway run`) contains none + of ``_PLIST_GATEWAY_ARGV_MARKERS``; only the split-token rule sees it. + """ + path = _write_plist( + tmp_path / "ai.hermes.svc-bare.plist", + { + "Label": "ai.hermes.svc-bare", + "ProgramArguments": ["/usr/local/bin/hermes", "gateway", "run"], + }, + ) + text = f"launchctl bootstrap gui/501 {path}" + assert contains_launchctl_submit_command(text) + + def test_argv_runs_gateway_lifecycle_inline_blocked( + self, tmp_path, launchd_identity + ): + """Neutral Label, non-entrypoint argv — but it KICKSTARTS us. + + Review round 1's finding: "is this plist a gateway job?" is not the + question the guard exists to answer. This plist is not a gateway job + by any tell (Label is neutral, argv is ``/bin/sh``), yet loading it + restarts THIS gateway — the #62891 laundering shape with a file + instead of a ``submit`` line. Needs no root: a user-writable + ``$TMPDIR`` path bootstrapped into ``gui/`` is enough. + """ + path = _write_plist( + tmp_path / "ai.hermes.helper.plist", + { + "Label": "ai.hermes.helper", + "ProgramArguments": [ + "/bin/sh", + "-c", + f"launchctl kickstart -k system/{SELF_LAUNCHD}", + ], + }, + ) + text = f"launchctl bootstrap gui/501 {path}" + assert contains_launchctl_submit_command(text) + assert contains_gateway_lifecycle_command_or_referenced_script(text) + + def test_argv_runs_gateway_lifecycle_via_script_blocked( + self, tmp_path, launchd_identity + ): + """Same laundering, one indirection deeper: argv names a SCRIPT. + + The inline-string witness alone would pass a fix that only scanned + ``sh -c`` payloads; the lifecycle scanner's referenced-script walk is + what closes this one. + """ + script = tmp_path / "boot.sh" + script.write_text( + f"#!/bin/sh\nlaunchctl bootout system/{SELF_LAUNCHD}\n", encoding="utf-8" + ) + script.chmod(0o755) + path = _write_plist( + tmp_path / "ai.hermes.helper2.plist", + { + "Label": "ai.hermes.helper2", + "ProgramArguments": ["/bin/sh", str(script)], + }, + ) + text = f"launchctl bootstrap gui/501 {path}" + assert contains_launchctl_submit_command(text) + assert contains_gateway_lifecycle_command_or_referenced_script(text) + + def test_argv_runs_gateway_lifecycle_direct_argv_blocked( + self, tmp_path, launchd_identity + ): + """The lifecycle command IS the argv — no ``sh -c``, no script. + + ``ProgramArguments`` is already a split command line, so no single + token contains a lifecycle command; only scanning the JOINED argv + sees it. Without this witness a fix that scans tokens individually + passes every other negative here. + """ + path = _write_plist( + tmp_path / "ai.hermes.direct.plist", + { + "Label": "ai.hermes.direct", + "ProgramArguments": [ + "/bin/launchctl", + "kickstart", + "-k", + f"system/{SELF_LAUNCHD}", + ], + }, + ) + text = f"launchctl bootstrap gui/501 {path}" + assert contains_launchctl_submit_command(text) + assert contains_gateway_lifecycle_command_or_referenced_script(text) + + def test_argv_bootstraps_another_plist_blocked(self, tmp_path, monkeypatch, launchd_identity): + """A plist whose argv bootstraps a plist re-enters the argv scan. + + Two plists can reference each other; the scan is depth-bounded and + fails CLOSED at the bound, so the cycle terminates as a refusal. + + The verdict alone does NOT gate the bound — measured: with the bound + removed the cycle blows the Python stack, and the guard's own + ``except Exception`` fallback still returns the same ``True``. So + assert the WORK done: a bounded scan reads a handful of plists, an + unbounded one read 116 before the interpreter gave out. + """ + first = tmp_path / "ai.hermes.chain-a.plist" + second = tmp_path / "ai.hermes.chain-b.plist" + _write_plist( + first, + { + "Label": "ai.hermes.chain-a", + "ProgramArguments": [ + "/bin/sh", + "-c", + f"launchctl bootstrap gui/501 {second}", + ], + }, + ) + _write_plist( + second, + { + "Label": "ai.hermes.chain-b", + "ProgramArguments": [ + "/bin/sh", + "-c", + f"launchctl bootstrap gui/501 {first}", + ], + }, + ) + real_read = lifecycle_guard._read_plist_label_payload + reads = [] + + def counting_read(path): + reads.append(path) + return real_read(path) + + monkeypatch.setattr( + lifecycle_guard, "_read_plist_label_payload", counting_read + ) + assert contains_launchctl_submit_command( + f"launchctl bootstrap gui/501 {first}" + ) + assert len(reads) <= 8, f"unbounded plist recursion: {len(reads)} reads" + # The per-thread depth counter must unwind to 0, or the NEXT scan in + # this process starts pre-charged and fails closed on a benign plist. + assert getattr(lifecycle_guard._PLIST_ARGV_SCAN_STATE, "depth", 0) == 0 + + def test_nonexistent_path_blocked(self, tmp_path, launchd_identity): + """Nothing to read → the label is still attacker-chosen (#62891).""" + text = f"launchctl bootstrap gui/501 {tmp_path / 'not-written-yet.plist'}" + assert contains_launchctl_submit_command(text) + + def test_variable_in_path_blocked(self, launchd_identity): + """Unexpanded shell value: the path we read is not the path that runs.""" + text = "launchctl bootstrap gui/501 $PLIST" + assert contains_launchctl_submit_command(text) + + def test_variable_in_domain_argument_blocked(self, router_plist, launchd_identity): + """The plist is real and benign, but ANOTHER argument is unexpanded. + + `launchctl bootstrap gui/$UID ` is the common spelling. The + shell-value check covers every argument, not just the plist paths — + a variable elsewhere in the segment still means we are reasoning + about a command whose expanded form we have not seen. + """ + text = f"launchctl bootstrap gui/$UID {router_plist}" + assert contains_launchctl_submit_command(text) + + def test_variable_inside_existing_path_blocked(self, router_plist, launchd_identity): + """A real file plus a variable segment is still unresolvable.""" + text = f"launchctl bootstrap gui/501 ${{DIR}}/{router_plist}" + assert contains_launchctl_submit_command(text) + + def test_command_substitution_in_path_blocked(self, launchd_identity): + text = "launchctl bootstrap gui/501 $(mktemp).plist" + assert contains_launchctl_submit_command(text) + + def test_directory_argument_blocked(self, tmp_path, launchd_identity): + """`bootstrap ` loads every plist in the directory.""" + directory = tmp_path / "agents" + directory.mkdir() + _write_plist( + directory / "ai.hermes.gateway-aegis.plist", + {"Label": SELF_LAUNCHD, "ProgramArguments": ["/bin/true"]}, + ) + text = f"launchctl bootstrap gui/501 {directory}" + assert contains_launchctl_submit_command(text) + + def test_fifo_argument_blocked(self, tmp_path, launchd_identity): + """A FIFO named `.plist` must be refused WITHOUT being read. + + Unlike a directory (whose ``os.read`` raises ``IsADirectoryError`` + and is caught downstream), an unopened-for-write FIFO would make the + read block. ``O_NONBLOCK`` keeps the open from hanging, but only the + ``S_ISREG`` check makes the refusal explicit rather than incidental. + """ + path = tmp_path / "ai.hermes.fifo.plist" + os.mkfifo(path) + assert lifecycle_guard._read_plist_label_payload(path) is None + text = f"launchctl bootstrap gui/501 {path}" + assert contains_launchctl_submit_command(text) + + def test_directory_named_like_a_plist_blocked(self, tmp_path, launchd_identity): + """A DIRECTORY whose name ends in `.plist` is not a readable job.""" + path = tmp_path / "ai.hermes.dir.plist" + path.mkdir() + assert lifecycle_guard._read_plist_label_payload(path) is None + text = f"launchctl bootstrap gui/501 {path}" + assert contains_launchctl_submit_command(text) + + def test_unparseable_file_blocked(self, tmp_path, launchd_identity): + """A file that is not a plist at all cannot clear the guard.""" + path = tmp_path / "ai.hermes.broken.plist" + path.write_text("this is not a plist\n") + text = f"launchctl bootstrap gui/501 {path}" + assert contains_launchctl_submit_command(text) + + def test_plist_without_label_blocked(self, tmp_path, launchd_identity): + path = _write_plist( + tmp_path / "ai.hermes.nolabel.plist", + {"ProgramArguments": ["/bin/true"]}, + ) + text = f"launchctl bootstrap gui/501 {path}" + assert contains_launchctl_submit_command(text) + + def test_oversized_file_blocked(self, tmp_path, launchd_identity): + """Bounded read: anything over the cap is refused, not streamed.""" + path = tmp_path / "ai.hermes.huge.plist" + payload = { + "Label": "ai.hermes.huge", + "ProgramArguments": ["/bin/true"], + "Padding": "x" * (lifecycle_guard._MAX_PLIST_BYTES + 1024), + } + _write_plist(path, payload) + assert path.stat().st_size > lifecycle_guard._MAX_PLIST_BYTES + text = f"launchctl bootstrap gui/501 {path}" + assert contains_launchctl_submit_command(text) + + def test_oversized_file_reader_returns_none(self, tmp_path): + """Direct unit gate on the reader, independent of the verb path. + + ``_read_plist_label_payload`` carries THREE overlapping bounds (an + ``st_size`` pre-check, a bounded read loop, and a post-read length + check). Any one of them alone produces this ``None``, so no single + mutation of them can be killed — this asserts the property they + jointly guarantee rather than any one implementation of it. + """ + path = tmp_path / "ai.hermes.huge.plist" + _write_plist( + path, + { + "Label": "ai.hermes.huge", + "Padding": "x" * (lifecycle_guard._MAX_PLIST_BYTES + 1024), + }, + ) + assert path.stat().st_size > lifecycle_guard._MAX_PLIST_BYTES + assert lifecycle_guard._read_plist_label_payload(path) is None + + def test_reader_accepts_a_normal_plist(self, router_plist): + """Positive control for the reader: the bound is not a blanket refusal.""" + payload = lifecycle_guard._read_plist_label_payload( + pathlib.Path(router_plist) + ) + assert payload is not None + assert payload["Label"] == "ai.hermes.fleetreview-router" + + def test_mixed_readable_and_unreadable_blocked(self, router_plist, launchd_identity): + """Every plist argument must clear; one unreadable one blocks all.""" + text = ( + f"launchctl bootstrap gui/501 {router_plist} " + "/tmp/definitely-not-here-9f2c.plist" + ) + assert contains_launchctl_submit_command(text) + + def test_no_plist_argument_blocked(self, launchd_identity): + text = "launchctl bootstrap gui/501" + assert contains_launchctl_submit_command(text) + + +class TestSubmitUnchanged: + @pytest.mark.parametrize( + "text", + [ + "launchctl submit -l ai.hermes.gateway -- /bin/sh helper.sh", + "launchctl submit -l ai.hermes.fleetreview-router -- /bin/sh helper.sh", + "launchctl submit -l neutral-name -- /bin/sh helper.sh", + ], + ) + def test_submit_blocked_regardless_of_label(self, text, launchd_identity): + """`submit` never reads a file — its label proves nothing.""" + assert contains_launchctl_submit_command(text), f"Should match: {text!r}" + assert contains_gateway_lifecycle_command_or_referenced_script(text) + + def test_submit_with_a_readable_plist_argument_still_blocked( + self, router_plist, launchd_identity + ): + """Naming a benign plist must not launder a `submit`.""" + text = f"launchctl submit -l tmp -- /bin/cat {router_plist}" + assert contains_launchctl_submit_command(text) + + +class TestSiblingGatewayPlistStillAllowed: + """The pre-existing sibling-gateway exemption must survive unchanged.""" + + def test_sibling_gateway_plist_by_name_allowed(self, launchd_identity): + text = ( + "launchctl bootstrap gui/501 " + "/Users/ace/Library/LaunchAgents/ai.hermes.gateway.plist" + ) + assert not contains_launchctl_submit_command(text) + + def test_self_gateway_plist_by_name_blocked(self, launchd_identity): + text = ( + "launchctl bootstrap gui/501 " + f"/Users/ace/Library/LaunchAgents/{SELF_LAUNCHD}.plist" + ) + assert contains_launchctl_submit_command(text) diff --git a/tests/tools/test_execute_code_surfaces_blocks.py b/tests/tools/test_execute_code_surfaces_blocks.py new file mode 100644 index 0000000000000..4675c7ffaebcc --- /dev/null +++ b/tests/tools/test_execute_code_surfaces_blocks.py @@ -0,0 +1,191 @@ +"""``execute_code`` must SURFACE a guard block, never return empty success. + +The gateway-lifecycle guard refuses a ``terminal()`` call by RETURNING a +result dict. Inside the ``execute_code`` sandbox that is a value, not an +error — so the overwhelmingly common script shape :: + + r = terminal("sudo launchctl bootstrap system /path/to.plist") + +runs to completion without inspecting ``r``, and ``execute_code`` reports +``status=success``, ``exit_code=0``, ``output=""``. Measured 2026-09-20: a +silent block, strictly worse than the direct terminal tool, which at least +prints why it refused. + +The generated ``hermes_tools`` stub therefore raises ``ToolCallBlocked`` on +any result carrying the guard's marker key, so the refusal lands in the +cell's traceback and ``execute_code`` reports ``status=error`` with a +non-zero exit code. +""" + +from __future__ import annotations + +import json + +from cron.lifecycle_guard import GATEWAY_LIFECYCLE_BLOCK_MARKER +from tools.code_execution_tool import ( + SANDBOX_ALLOWED_TOOLS, + _TOOL_STUBS, + generate_hermes_tools_module, +) + + +def _stub_namespace(tool_names, call_result, transport="uds"): + """Exec the generated module with ``_call`` replaced by a canned result.""" + source = generate_hermes_tools_module(list(tool_names), transport=transport) + namespace: dict = {} + exec(compile(source, "", "exec"), namespace) + namespace["_call"] = lambda tool, args: call_result + return namespace + + +def _blocked_result(): + """The exact shape tools/terminal_tool.py returns on a lifecycle block.""" + return { + "output": "", + "exit_code": 1, + "error": ( + "Blocked: launchctl submit registers a persistent KeepAlive job " + "and is unsafe from inside the gateway process." + ), + "status": "error", + "blocked_by": GATEWAY_LIFECYCLE_BLOCK_MARKER, + } + + +class TestStubRaisesOnBlock: + def test_terminal_stub_raises_on_a_blocked_result(self): + namespace = _stub_namespace(["terminal"], _blocked_result()) + blocked = namespace["ToolCallBlocked"] + try: + namespace["terminal"]("sudo launchctl bootstrap system /tmp/x.plist") + except blocked as exc: + assert "was blocked and did not run" in str(exc) + assert "launchctl submit" in str(exc) + else: + raise AssertionError("a blocked terminal() call returned instead of raising") + + def test_blocked_marker_key_matches_the_producer(self): + """Contract test: the stub's key is the guard's key. + + The stub is generated source and cannot import the guard, so the key + is duplicated. Derive BOTH sides here rather than retyping either — + a rename on the producer side must fail this test, not silently + disarm the stub. + """ + source = generate_hermes_tools_module(["terminal"]) + assert '_BLOCKED_MARKER_KEY = "blocked_by"' in source + result = _blocked_result() + assert result["blocked_by"] == GATEWAY_LIFECYCLE_BLOCK_MARKER + namespace = _stub_namespace(["terminal"], result) + assert namespace["_BLOCKED_MARKER_KEY"] in result + + def test_every_generated_stub_routes_through_the_guard(self): + """Not just terminal(): every tool the sandbox exposes. + + Calls each stub with placeholder arguments derived from its own + signature, so a newly-added tool is covered without editing this + test. + """ + import inspect + + names = sorted(SANDBOX_ALLOWED_TOOLS & set(_TOOL_STUBS)) + assert names, "no sandbox tool stubs to check" + namespace = _stub_namespace(names, _blocked_result()) + blocked = namespace["ToolCallBlocked"] + checked = [] + for name in names: + func = namespace[name] + signature = inspect.signature(func) + args = [ + ["x"] if parameter.annotation is list else "x" + for parameter in signature.parameters.values() + if parameter.default is inspect.Parameter.empty + ] + try: + func(*args) + except blocked: + checked.append(name) + continue + raise AssertionError(f"{name}() did not raise on a blocked result") + assert checked == names + + def test_file_transport_stub_raises_too(self): + """Remote backends generate a different header, same contract.""" + namespace = _stub_namespace( + ["terminal"], _blocked_result(), transport="file" + ) + blocked = namespace["ToolCallBlocked"] + try: + namespace["terminal"]("launchctl submit -l x -- /bin/true") + except blocked: + return + raise AssertionError("file-transport stub returned a blocked result") + + +class TestNormalResultsUnaffected: + def test_successful_result_passes_through_unchanged(self): + ok = {"output": "hi\n", "exit_code": 0, "error": None} + namespace = _stub_namespace(["terminal"], ok) + assert namespace["terminal"]("echo hi") == ok + + def test_ordinary_nonzero_exit_is_not_a_block(self): + """A command that RAN and failed must still return, not raise. + + This is the discriminating half: if the stub raised on any non-zero + exit code, every legitimate `grep` miss or `exit 7` would explode. + """ + failed = {"output": "", "exit_code": 7, "error": None} + namespace = _stub_namespace(["terminal"], failed) + assert namespace["terminal"]("exit 7") == failed + + def test_error_status_without_the_marker_is_not_a_block(self): + """`status: error` alone is not a refusal — only the marker is.""" + errored = { + "output": "", + "exit_code": -1, + "error": "Failed to execute command: boom", + "status": "error", + } + namespace = _stub_namespace(["terminal"], errored) + assert namespace["terminal"]("boom") == errored + + def test_non_dict_result_passes_through(self): + namespace = _stub_namespace(["terminal"], "raw string result") + assert namespace["terminal"]("echo hi") == "raw string result" + + def test_falsy_marker_value_is_not_a_block(self): + """A key present but empty must not trip the guard path.""" + result = {"output": "", "exit_code": 0, "blocked_by": ""} + namespace = _stub_namespace(["terminal"], result) + assert namespace["terminal"]("echo hi") == result + + +class TestTerminalToolStampsTheMarker: + """The producer side: both refusal returns must carry the marker.""" + + def test_both_block_returns_carry_the_marker(self): + import re + from pathlib import Path + + import tools.terminal_tool as terminal_tool + + source = Path(terminal_tool.__file__).read_text(encoding="utf-8") + # Every `"status": "error"` return in the supervised-gateway guard + # block must be stamped. Locate the guard region lexically. + start = source.index("if _is_supervised_gateway_process():") + end = source.index("# Validate before the source guard resolves", start) + region = source[start:end] + blocks = re.findall(r'"status": "error",', region) + markers = re.findall( + r'"blocked_by": _GATEWAY_LIFECYCLE_BLOCK_MARKER,', region + ) + assert len(blocks) == 2, f"expected 2 guard refusals, found {len(blocks)}" + assert len(markers) == len(blocks), ( + f"{len(blocks)} guard refusals but only {len(markers)} stamped " + "with the block marker — an unstamped refusal is a silent block " + "inside execute_code" + ) + + def test_marker_survives_json_round_trip(self): + payload = json.dumps(_blocked_result(), ensure_ascii=False) + assert json.loads(payload)["blocked_by"] == GATEWAY_LIFECYCLE_BLOCK_MARKER diff --git a/tools/code_execution_tool.py b/tools/code_execution_tool.py index 5a29e194f1407..8bde7416c79cb 100644 --- a/tools/code_execution_tool.py +++ b/tools/code_execution_tool.py @@ -573,7 +573,9 @@ def generate_hermes_tools_module(enabled_tools: List[str], stub_functions.append( f"def {func_name}({sig}):\n" f" {doc}\n" - f" return _call({func_name!r}, {args_expr})\n" + f" return _raise_if_blocked(\n" + f" {func_name!r}, _call({func_name!r}, {args_expr})\n" + f" )\n" ) export_names.append(func_name) @@ -626,6 +628,36 @@ def retry(fn, max_attempts=3, delay=2): time.sleep(delay * (2 ** attempt)) raise last_err + +class ToolCallBlocked(RuntimeError): + """A tool call was REFUSED by a Hermes guard; it never ran.""" + + +# Marker key set by tools/terminal_tool.py on a gateway-lifecycle refusal. +# Kept in sync with cron.lifecycle_guard.GATEWAY_LIFECYCLE_BLOCK_MARKER by +# tests/tools/test_execute_code_surfaces_blocks.py. +_BLOCKED_MARKER_KEY = "blocked_by" + + +def _raise_if_blocked(tool_name, result): + """Turn a guard REFUSAL into an exception instead of a return value. + + A refused call comes back as an ordinary dict. A script that does not + inspect it (the common shape: ``r = terminal(cmd)`` with no check) then + runs to completion, and execute_code reports status=success / + exit_code=0 with empty output — a silent block, strictly worse than the + direct terminal tool, which at least prints why. Raising puts the + refusal in the cell's traceback so the caller always sees it. + """ + if not isinstance(result, dict): + return result + if not result.get(_BLOCKED_MARKER_KEY): + return result + message = result.get("error") or "blocked by a Hermes guard" + raise ToolCallBlocked( + tool_name + "() was blocked and did not run: " + str(message) + ) + ''' # ---- UDS transport (local backend) --------------------------------------- diff --git a/tools/terminal_tool.py b/tools/terminal_tool.py index 943ab6c8c5f42..d7b803c8c40c6 100644 --- a/tools/terminal_tool.py +++ b/tools/terminal_tool.py @@ -3169,6 +3169,7 @@ def terminal_tool( if _is_supervised_gateway_process(): from cron.lifecycle_guard import ( _MAX_REFERENCED_SCRIPT_BYTES, + GATEWAY_LIFECYCLE_BLOCK_MARKER as _GATEWAY_LIFECYCLE_BLOCK_MARKER, contains_gateway_lifecycle_command_or_referenced_script, contains_launchctl_submit_command, describe_self_gateway_identity, @@ -3192,9 +3193,11 @@ def terminal_tool( "KeepAlive job and is unsafe from inside the gateway process. " "Use Hermes cron for one-shot delayed work, or install an " "explicit LaunchAgent from a separate shell. " - "(launchctl bootstrap of a SIBLING gateway's plist is allowed.)" + "(launchctl bootstrap of a SIBLING gateway's plist, or of an " + "EXISTING non-gateway plist, is allowed.)" ), "status": "error", + "blocked_by": _GATEWAY_LIFECYCLE_BLOCK_MARKER, }, ensure_ascii=False) guard_cwd_base = get_session_cwd(session_key) if guard_cwd_base is None: @@ -3278,6 +3281,7 @@ def _read_script_in_env(script_path: str) -> Optional[str]: "restart` from a separate shell outside the running gateway." ), "status": "error", + "blocked_by": _GATEWAY_LIFECYCLE_BLOCK_MARKER, }, ensure_ascii=False) # Validate before the source guard resolves an explicit workdir.