From c262db319e42c40257cd6ff4fd9e465a3bacfa89 Mon Sep 17 00:00:00 2001 From: "Hoang V. Pham" <26063003+hehehe0803@users.noreply.github.com> Date: Wed, 22 Jul 2026 04:57:09 +0700 Subject: [PATCH] test(safety): make updater restart tests hermetic across platforms cmd_update discovers and restarts running gateways after its git and dependency work. The autouse live-system guard blocked os.kill and systemctl mutations but left gateway/service discovery live: on a developer machine with a running gateway, an updater test whose mocks miss discovery could find it and schedule a detached restart watcher that rewrites the real user systemd unit. Prior PRs #23397, #22900, #44267 closed narrower slices of this class. While cmd_update/_cmd_update_impl is on the stack, gateway/service/ platform discovery and the restart, terminate, and unit-write boundaries now return inert values. The command classifiers are hardened: a read-only allowlist for protected service targets; recursion through shell wrappers (sh/bash -c/-lc, cmd /c, powershell -Command/-EncodedCommand); blocking of foreign and negative/process- group kill targets; and blocking of detached gateway spawns and launchctl/taskkill of a Hermes gateway. The guard stays strictly additive: read-only probes, benign kills, and git pass through. Updater-frame detection is by name only (no hermes_cli.main import), so non-updater test processes are unaffected. Test-only; no product code. Adds synthetic classifier, wrapper-level wiring, and load-bearing canary tests that never discover, signal, spawn, or rewrite a real process or unit. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_015cwEx8CjYtTEvrnmCfp6Hw --- tests/conftest.py | 507 +++++++++++++++--- tests/test_live_system_guard_classifiers.py | 184 +++++++ tests/test_live_system_guard_self_test.py | 169 ++++-- .../test_live_system_guard_update_hermetic.py | 259 +++++++++ 4 files changed, 1002 insertions(+), 117 deletions(-) create mode 100644 tests/test_live_system_guard_classifiers.py create mode 100644 tests/test_live_system_guard_update_hermetic.py diff --git a/tests/conftest.py b/tests/conftest.py index 662324dce6dc2..0bbae53ee1856 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -523,9 +523,10 @@ def _ensure_current_event_loop(request): # a hard ``RuntimeError`` so the offending test gets a stack trace # instead of silently murdering the real gateway. # • ``subprocess.run`` / ``subprocess.Popen`` / ``call`` / ``check_call`` / -# ``check_output`` reject any ``systemctl ... hermes-gateway`` -# invocation that would mutate the live unit. Read-only systemctl -# calls (``status``, ``show``, ``list-units``) still pass through. +# ``check_output`` reject mutating systemd/launchd commands, detached +# Hermes gateway spawns, and process-killer commands targeting a foreign +# PID. Read-only service-manager calls still pass through outside updater +# tests. # # We intentionally do NOT stub ``find_gateway_pids`` / ``_scan_gateway_pids`` # here — tests of those functions themselves need the real implementation. @@ -537,6 +538,295 @@ def _ensure_current_event_loop(request): _LIVE_SYSTEM_GUARD_BYPASS_MARK = "live_system_guard_bypass" +def _guard_cmd_to_string(cmd) -> str: + """Render an argv/string for pure, side-effect-free guard inspection.""" + if cmd is None: + return "" + if isinstance(cmd, (bytes, bytearray)): + try: + return bytes(cmd).decode(errors="replace") + except Exception: + return "" + if isinstance(cmd, str): + return cmd + if isinstance(cmd, (list, tuple)): + try: + return " ".join(_guard_token_to_string(token) for token in cmd) + except Exception: + return "" + return str(cmd) + + +def _guard_token_to_string(token) -> str: + """Normalize one argv element without collapsing bytes into ``b'...'``.""" + if isinstance(token, (bytes, bytearray)): + return bytes(token).decode(errors="replace") + return str(token) + + +def _guard_tokens(cmd) -> list[str]: + import shlex + + if isinstance(cmd, (list, tuple)): + try: + return [_guard_token_to_string(token) for token in cmd] + except Exception: + return [] + rendered = _guard_cmd_to_string(cmd) + try: + return shlex.split(rendered) + except ValueError: + return rendered.split() + + +def _guard_command_variants(cmd) -> list[list[str]]: + """Return argv plus recursively parsed shell-wrapper payloads.""" + import base64 + + posix_shells = {"sh", "bash", "dash", "zsh", "ash", "ksh"} + windows_shells = {"cmd", "powershell", "pwsh"} + protected_hints = ("hermes", "gateway") + variants: list[list[str]] = [] + pending = [(_guard_tokens(cmd), 0)] + while pending: + tokens, depth = pending.pop() + if not tokens: + continue + variants.append(tokens) + if depth >= 5: + continue + for index, token in enumerate(tokens): + head = _guard_executable(token) + args = tokens[index + 1:] + payload = None + if head in posix_shells: + for position, flag in enumerate(args[:-1]): + if ( + flag.startswith("-") + and not flag.startswith("--") + and "c" in flag[1:].lower() + ): + payload = args[position + 1] + break + elif head == "cmd": + for position, flag in enumerate(args[:-1]): + if flag.lower() in {"/c", "/k"}: + payload = args[position + 1] + break + elif head in windows_shells: + for position, flag in enumerate(args[:-1]): + normalized_flag = flag.lower() + if normalized_flag in {"-command", "/command", "-c"}: + payload = args[position + 1] + break + if normalized_flag in {"-encodedcommand", "-e", "-enc"}: + encoded_payload = args[position + 1] + try: + payload = base64.b64decode( + encoded_payload, validate=True + ).decode("utf-16-le") + except (UnicodeDecodeError, ValueError): + if any(hint in encoded_payload.lower() for hint in protected_hints): + pending.append( + (["__guard_decode_failure__", encoded_payload], depth + 1) + ) + break + if payload is not None: + pending.append((_guard_tokens(payload), depth + 1)) + return variants + + +def _guard_executable(token: str) -> str: + executable = token.rsplit("/", 1)[-1].rsplit("\\", 1)[-1].lower() + return executable[:-4] if executable.endswith(".exe") else executable + + +def _is_service_manager_command(cmd) -> bool: + """True for a systemd or launchd command, irrespective of its verb.""" + return any( + _guard_executable(token) in {"systemctl", "launchctl"} + for tokens in _guard_command_variants(cmd) + for token in tokens + ) + + +def _is_service_manager_mutation_command(cmd) -> bool: + """Fail closed for non-read-only verbs targeting a Hermes service.""" + hermes_tokens = ( + "hermes-gateway", + "hermes.service", + "ai.hermes.gateway", + "hermes_cli.main gateway", + "hermes_cli/main.py gateway", + "gateway/run.py", + "hermes gateway", + ) + read_only_verbs = { + "status", "show", "cat", "list-units", "list", "is-active", + "is-enabled", "is-failed", "is-system-running", "print", "blame", + "get-default", + } + option_values = { + "--property", "--type", "--state", "--output", "--lines", + "--job-mode", "--signal", "--kill-whom", "--what", "--host", + "--machine", "--root", "--image", "--firmware-setup", "--timestamp", + } + short_options_with_values = {"-p", "-t", "-o", "-n", "-h", "-m"} + known_flag_options = { + "--all", "--ask-password", "--force", "--full", "--global", "--help", + "--legend", "--no-ask-password", "--no-block", "--no-legend", "--no-pager", + "--no-reload", "--now", "--plain", "--quiet", "--recursive", "--reverse", + "--runtime", "--system", "--user", "--version", "-a", "-f", "-l", "-q", + "-r", + } + + def _positional_verb(tokens, manager_index): + position = manager_index + 1 + while position < len(tokens): + token = tokens[position] + lowered = token.lower() + if token == "--": + return tokens[position + 1].lower() if position + 1 < len(tokens) else None + if not token.startswith("-"): + return lowered + if "=" in token: + option = lowered.split("=", 1)[0] + if option in option_values or option in known_flag_options: + position += 1 + continue + return None + if lowered in option_values or lowered in short_options_with_values: + if position + 1 >= len(tokens): + return None + position += 2 + continue + if lowered in known_flag_options: + position += 1 + continue + return None + return None + + for tokens in _guard_command_variants(cmd): + low = " ".join(tokens).lower() + if tokens and tokens[0] == "__guard_decode_failure__": + if any(marker in low for marker in hermes_tokens): + return True + continue + if not any(marker in low for marker in hermes_tokens): + continue + if _guard_executable(tokens[0]) in { + "stop-service", "start-service", "restart-service", "set-service", + "remove-service", + }: + return True + for index, token in enumerate(tokens): + if _guard_executable(token) not in {"systemctl", "launchctl"}: + continue + verb = _positional_verb(tokens, index) + if verb not in read_only_verbs: + return True + return False + + +def _is_detached_gateway_spawn(name: str, cmd) -> bool: + """Classify detached gateway launches without executing the command.""" + if name != "Popen": + return False + for tokens in _guard_command_variants(cmd): + lower_tokens = [token.lower() for token in tokens] + if "gateway" not in lower_tokens or "run" not in lower_tokens: + continue + low = " ".join(tokens).lower() + if any( + marker in low + for marker in ("hermes_cli.main", "hermes_cli/main.py", "gateway/run.py") + ): + return True + if any(_guard_executable(token) == "hermes" for token in tokens): + return True + return False + + +def _is_process_killer_command(cmd, *, is_own_subtree) -> bool: + """Classify process-killer argv, including Windows PID/image kills.""" + process_killers = {"pkill", "killall", "taskkill", "skill", "fuser", "kill", "killpg"} + for tokens in _guard_command_variants(cmd): + if not tokens: + continue + heads = [_guard_executable(token) for token in tokens] + low = " ".join(tokens).lower() + for index, head in enumerate(heads): + if head not in process_killers: + continue + if head == "taskkill": + upper_tokens = [part.upper() for part in tokens] + for flag in ("/IM", "-IM"): + if flag not in upper_tokens: + continue + image_index = upper_tokens.index(flag) + 1 + image = tokens[image_index].lower() if image_index < len(tokens) else "" + if "hermes-gateway" in image or image.startswith("python"): + return True + try: + pid_index = next( + position + 1 + for position, part in enumerate(upper_tokens) + if part in {"/PID", "-PID"} + ) + target_pid = int(tokens[pid_index]) + except (StopIteration, ValueError, IndexError): + target_pid = None + if target_pid is not None and not is_own_subtree(target_pid): + return True + continue + if head in {"kill", "killpg"}: + args = tokens[index + 1:] + if head == "killpg": + if args and args[0] == "--": + args = args[1:] + if not args: + return True + try: + target_pid = int(args[0]) + except ValueError: + return True + if target_pid <= 0 or not is_own_subtree(target_pid): + return True + continue + + position = 0 + if args and args[0] == "--": + position = 1 + elif args and args[0].startswith("-"): + signal_option = args[0] + if signal_option.lower() in {"-s", "-n"}: + position = 2 + elif signal_option[1:].isalnum(): + position = 1 + else: + return True + if position < len(args) and args[position] == "--": + position += 1 + targets = args[position:] + if not targets: + return True + for target in targets: + try: + target_pid = int(target) + except ValueError: + return True + if target_pid <= 0 or not is_own_subtree(target_pid): + return True + continue + if ( + "hermes" in low + or "gateway" in low + or ("python" in low and "-f" in tokens) + ): + return True + return False + + def pytest_configure(config): # noqa: D401 — pytest hook """Register markers used by hermetic conftest.""" config.addinivalue_line( @@ -556,7 +846,7 @@ def pytest_configure(config): # noqa: D401 — pytest hook @pytest.fixture(autouse=True) -def _live_system_guard(request, monkeypatch): +def _live_system_guard(request, monkeypatch, _hermetic_environment): """Block real os.kill / systemctl / gateway-pid scans during tests. See block comment above for the why. Tests that genuinely need @@ -582,9 +872,111 @@ def _live_system_guard(request, monkeypatch): return import os as _os - import shlex as _shlex import subprocess as _subprocess + # ``cmd_update`` discovers and restarts gateways after otherwise-mocked + # git/dependency work. Keep discovery tests real, but return inert values + # for the entire updater stack so a missing mock cannot find a developer's + # running gateway or schedule a detached restart watcher. + from pathlib import Path as _Path + + def _inside_cmd_update() -> bool: + frame = sys._getframe(1) + while frame is not None: + if ( + frame.f_code.co_name in {"cmd_update", "_cmd_update_impl"} + and frame.f_globals.get("__name__") == "hermes_cli.main" + ): + return True + frame = frame.f_back + return False + + def _inert_during_update(real, inert): + def _guarded(*args, **kwargs): + if _inside_cmd_update(): + return inert() + return _guarded.__wrapped__(*args, **kwargs) + + _guarded.__wrapped__ = real + _guarded._live_system_guard_inert = True + _guarded.__name__ = getattr(real, "__name__", "_guarded") + return _guarded + + # Do not import ``hermes_cli.main`` here: importing it resolves profiles, + # dotenv/config, and logging for every test-file subprocess. An updater + # test has it loaded at collection time, so only then load the mutation + # boundaries that need updater-scoped inerting. + if "hermes_cli.main" in sys.modules: + from hermes_cli import gateway as _gateway + from hermes_cli import gateway_windows as _gateway_windows + from gateway import status as _gateway_status + + def _inert_unit_refresh(path_getter): + def _guarded_refresh(*args, **kwargs): + target = path_getter(*args, **kwargs).resolve() + hermes_home = _Path(os.environ["HERMES_HOME"]).resolve() + if target != hermes_home and hermes_home not in target.parents: + raise RuntimeError( + "tests/conftest.py live-system guard: blocked updater " + f"unit write to non-test path {target}" + ) + return False + + return _guarded_refresh + + for _name, _inert in ( + ("find_gateway_pids", list), + ("_scan_gateway_pids", list), + ("_get_service_pids", set), + ("find_profile_gateway_processes", list), + ("supports_systemd_services", lambda: False), + ("is_macos", lambda: False), + ("_ensure_user_systemd_env", lambda: None), + ("launchd_restart", lambda: None), + ("get_launchd_label", lambda: "ai.hermes.gateway"), + ( + "get_launchd_plist_path", + lambda: _Path("/definitely-not-a-real-hermes-launchd.plist"), + ), + ("launch_detached_profile_gateway_restart", lambda: False), + ("launch_detached_gateway_restart_by_cmdline", lambda: False), + ("_capture_gateway_argv", lambda: None), + ( + "refresh_systemd_unit_if_needed", + _inert_unit_refresh( + lambda *args, **kwargs: _gateway.get_systemd_unit_path( + system=kwargs.get("system", args[0] if args else False) + ) + ), + ), + ( + "refresh_launchd_plist_if_needed", + _inert_unit_refresh(lambda *args, **kwargs: _gateway.get_launchd_plist_path()), + ), + ): + monkeypatch.setattr( + _gateway, + _name, + _inert_during_update(getattr(_gateway, _name), _inert), + ) + for _name, _inert in ( + ("is_installed", lambda: False), + ("_spawn_detached", lambda: None), + ): + monkeypatch.setattr( + _gateway_windows, + _name, + _inert_during_update(getattr(_gateway_windows, _name), _inert), + ) + monkeypatch.setattr( + _gateway_status, + "terminate_pid", + _inert_during_update(_gateway_status.terminate_pid, lambda: None), + ) + + # Follow-up: psutil.Process.terminate/kill/send_signal are intentionally + # out of scope; updater termination is inerted above via terminate_pid. + test_pid = _os.getpid() # Capture the test process's existing children at fixture start — # any *new* children spawned by the test are also allowlisted via @@ -607,6 +999,11 @@ def _is_own_subtree(pid: int) -> bool: return True if pid < 0: return False + # PID 1 is the self-test's designated foreign PID. A namespaced + # pytest runner can itself be PID 1, but treating init as an owned + # child would turn the guard's foreign-PID proof into a real killpg. + if pid == 1: + return False if pid == test_pid or pid in _initial_children: return True if _psutil is None: @@ -662,6 +1059,14 @@ def _guarded_killpg(pgid, sig, *args, **kwargs): # Signal 0 is a pure liveness probe — never destructive. if int(sig) == 0: return real_killpg(pgid, sig, *args, **kwargs) + # In a PID namespace pytest can share init's group. PID 1 is + # nevertheless the foreign-PID safety sentinel and must never be + # considered an owned test group for destructive signals. + if int(pgid) == 1: + raise RuntimeError( + f"tests/conftest.py live-system guard: blocked " + f"os.killpg({pgid}, {sig}) — PID 1 is always foreign." + ) if int(pgid) == own_pgid or _is_own_subtree(int(pgid)): return real_killpg(pgid, sig, *args, **kwargs) raise RuntimeError( @@ -673,87 +1078,24 @@ def _guarded_killpg(pgid, sig, *args, **kwargs): monkeypatch.setattr(_os, "killpg", _guarded_killpg) # ── Subprocess command-string inspection (whole-line) ────────── - _HERMES_TOKENS = ( - "hermes-gateway", - "hermes.service", - "hermes_cli.main gateway", - "hermes_cli/main.py gateway", - "gateway/run.py", - "hermes gateway", - ) - _MUTATING_VERBS = ( - "restart", "start", "stop", "kill", "reload", - "reset-failed", "enable", "disable", "mask", "unmask", - "daemon-reload", "try-restart", "reload-or-restart", - ) - _PROCESS_KILLERS = ("pkill", "killall", "taskkill", "skill", "fuser") - - def _cmd_to_string(cmd) -> str: - if cmd is None: - return "" - if isinstance(cmd, (bytes, bytearray)): - try: - return bytes(cmd).decode(errors="replace") - except Exception: - return "" - if isinstance(cmd, str): - return cmd - if isinstance(cmd, (list, tuple)): - try: - return " ".join(str(t) for t in cmd) - except Exception: - return "" - return str(cmd) - - def _matches_hermes_gateway(cmd_str: str) -> bool: - low = cmd_str.lower() - return any(tok in low for tok in _HERMES_TOKENS) - - def _is_blocked_systemctl(cmd) -> bool: - cmd_str = _cmd_to_string(cmd) - if "systemctl" not in cmd_str: - return False - if not _matches_hermes_gateway(cmd_str): - return False - try: - tokens = _shlex.split(cmd_str) - except ValueError: - tokens = cmd_str.split() - return any(verb in tokens for verb in _MUTATING_VERBS) - - def _is_process_killer(cmd) -> bool: - cmd_str = _cmd_to_string(cmd) - try: - tokens = _shlex.split(cmd_str) - except ValueError: - tokens = cmd_str.split() - if not tokens: - return False - for tok in tokens: - head = tok.rsplit("/", 1)[-1].rsplit("\\", 1)[-1] - if head in _PROCESS_KILLERS: - low = cmd_str.lower() - # pkill -f pattern: catch hermes-themed patterns + a - # plain "python" -f which would catch the live gateway - # whose cmdline contains "python -m hermes_cli.main". - if ( - "hermes" in low - or "gateway" in low - or ("python" in low and "-f" in tokens) - ): - return True - return False - def _check_subprocess_cmd(name, cmd): - if _is_blocked_systemctl(cmd): + if _is_service_manager_mutation_command(cmd): raise RuntimeError( f"tests/conftest.py live-system guard: blocked " f"subprocess.{name}({cmd!r}) — would mutate the " - "live hermes-gateway systemd unit. Mock " + "live hermes-gateway service definition. Mock " "subprocess.run / _run_systemctl in the test, or " "mark with @pytest.mark.live_system_guard_bypass." ) - if _is_process_killer(cmd): + if _is_detached_gateway_spawn(name, cmd): + raise RuntimeError( + f"tests/conftest.py live-system guard: blocked " + f"subprocess.{name}({cmd!r}) — would launch a Hermes " + "gateway outside the test process lifecycle. Mock Popen " + "in the test, or mark with " + "@pytest.mark.live_system_guard_bypass." + ) + if _is_process_killer_command(cmd, is_own_subtree=_is_own_subtree): raise RuntimeError( f"tests/conftest.py live-system guard: blocked " f"subprocess.{name}({cmd!r}) — process-killer command " @@ -771,7 +1113,7 @@ def _check_subprocess_cmd(name, cmd): # tree (PPid=1) and nearly impossible to trace without explicit # inotify/SHA watchdogs. Any test that legitimately needs to exercise # the update-spawn path must mock subprocess.Popen explicitly. - cmd_str = _cmd_to_string(cmd) + cmd_str = _guard_cmd_to_string(cmd) low = cmd_str.lower() if "update" in low and ( # hermes update / hermes update --gateway / setsid bash -c ... hermes update @@ -799,9 +1141,12 @@ def _check_subprocess_cmd(name, cmd): def _wrap_subprocess(name, real): def _guarded(cmd, *args, **kwargs): + if name == "run" and _inside_cmd_update() and _is_service_manager_command(cmd): + return _subprocess.CompletedProcess(cmd, 1, stdout="", stderr="") _check_subprocess_cmd(name, cmd) - return real(cmd, *args, **kwargs) + return _guarded.__wrapped__(cmd, *args, **kwargs) _guarded.__name__ = f"_guarded_{name}" + _guarded.__wrapped__ = real # Make the wrapper subscriptable like the wrapped callable when # the wrapped object is. ``subprocess.Popen[bytes]`` is used as # a type annotation in third-party packages (mcp, etc.); replacing diff --git a/tests/test_live_system_guard_classifiers.py b/tests/test_live_system_guard_classifiers.py new file mode 100644 index 0000000000000..35cfaf13b8bd9 --- /dev/null +++ b/tests/test_live_system_guard_classifiers.py @@ -0,0 +1,184 @@ +"""Pure regression tests for live-system-guard command classifiers. + +These tests only classify synthetic argv. They never invoke a native service +manager or process killer, so they remain safe even while a new classifier is +being developed. +""" + +from pathlib import Path +import sys +import base64 + + +def _guard_module(): + expected = Path(__file__).with_name("conftest.py").resolve() + for module in tuple(sys.modules.values()): + module_file = getattr(module, "__file__", None) + if module_file and Path(module_file).resolve() == expected: + return module + raise AssertionError("pytest did not load tests/conftest.py") + + +def test_launchctl_kickstart_is_a_blocked_service_mutation(): + conftest = _guard_module() + + assert conftest._is_service_manager_mutation_command( + ["launchctl", "kickstart", "gui/501/ai.hermes.gateway"] + ) + + +def test_launchctl_bootout_is_a_blocked_service_mutation(): + conftest = _guard_module() + + assert conftest._is_service_manager_mutation_command( + "launchctl bootout gui/501/ai.hermes.gateway" + ) + + +def test_taskkill_foreign_synthetic_pid_is_blocked(): + conftest = _guard_module() + + assert conftest._is_process_killer_command( + ["taskkill", "/PID", "987654321", "/T", "/F"], + is_own_subtree=lambda _pid: False, + ) + + +def test_detached_gateway_popen_is_blocked(): + conftest = _guard_module() + + assert conftest._is_detached_gateway_spawn( + "Popen", + ["/definitely-not-a-real-hermes-gateway", "-m", "hermes_cli.main", "gateway", "run"], + ) + + +def test_protected_service_verbs_fail_closed_except_read_only_allowlist(): + conftest = _guard_module() + + assert not conftest._is_service_manager_mutation_command( + ["systemctl", "--user", "status", "hermes-gateway.service"] + ) + assert not conftest._is_service_manager_mutation_command( + ["launchctl", "print", "gui/501/ai.hermes.gateway"] + ) + for verb in ("edit", "set-property", "reenable", "preset", "revert"): + assert conftest._is_service_manager_mutation_command( + ["systemctl", "--user", verb, "hermes-gateway.service"] + ), verb + + +def test_bytes_argv_and_shell_payloads_are_classified_recursively(): + conftest = _guard_module() + + assert conftest._is_service_manager_mutation_command( + [b"bash", b"-c", b"systemctl --user restart hermes-gateway.service"] + ) + assert conftest._is_service_manager_mutation_command( + ["cmd", "/c", "launchctl bootout gui/501/ai.hermes.gateway"] + ) + assert conftest._is_detached_gateway_spawn( + "Popen", ["powershell", "-Command", "hermes gateway run"] + ) + + +def test_shell_wrapper_variants_recurse_into_protected_payloads(): + conftest = _guard_module() + encoded = base64.b64encode( + "systemctl restart hermes-gateway".encode("utf-16-le") + ).decode() + + assert conftest._is_service_manager_mutation_command( + 'bash -lc "systemctl restart hermes-gateway"' + ) + assert conftest._is_service_manager_mutation_command( + 'cmd.exe /c "systemctl stop hermes-gateway"' + ) + assert conftest._is_service_manager_mutation_command( + 'powershell -Command "Stop-Service hermes-gateway"' + ) + assert conftest._is_service_manager_mutation_command( + ["pwsh", "-EncodedCommand", encoded] + ) + assert conftest._is_service_manager_mutation_command( + ["pwsh", "-enc", "hermes-gateway"] + ) + assert conftest._is_detached_gateway_spawn( + "Popen", 'sh -c "hermes gateway run &"' + ) + + +def test_shell_wrapper_variants_leave_benign_payloads_unblocked(): + conftest = _guard_module() + + assert not conftest._is_service_manager_mutation_command( + 'bash -lc "echo hello"' + ) + + +def test_external_kill_commands_use_numeric_subtree_ownership(): + conftest = _guard_module() + + assert conftest._is_process_killer_command( + ["kill", "-TERM", "987654321"], is_own_subtree=lambda _pid: False + ) + assert conftest._is_process_killer_command( + ["killpg", "987654321"], is_own_subtree=lambda _pid: False + ) + assert conftest._is_process_killer_command( + ["kill", "--", "-1"], is_own_subtree=lambda _pid: False + ) + assert not conftest._is_process_killer_command( + ["kill", "-TERM", "12345"], is_own_subtree=lambda pid: pid == 12345 + ) + + +def test_kill_negative_targets_and_foreign_process_groups_are_blocked(): + conftest = _guard_module() + + assert conftest._is_process_killer_command( + ["kill", "-9", "-1"], is_own_subtree=lambda _pid: False + ) + assert conftest._is_process_killer_command( + ["kill", "-9", "-987654321"], is_own_subtree=lambda _pid: False + ) + assert conftest._is_process_killer_command( + ["killpg", "987654321", "9"], is_own_subtree=lambda _pid: False + ) + + +def test_service_manager_verb_is_positional_and_read_only_forms_pass(): + conftest = _guard_module() + + assert conftest._is_service_manager_mutation_command( + ["systemctl", "reenable", "hermes-gateway.service"] + ) + assert conftest._is_service_manager_mutation_command( + ["systemctl", "set-property", "hermes-gateway.service", "CPUQuota=1%"] + ) + assert conftest._is_service_manager_mutation_command( + ["systemctl", "--property", "status", "restart", "hermes-gateway.service"] + ) + assert not conftest._is_service_manager_mutation_command( + ["systemctl", "--user", "status", "hermes-gateway"] + ) + assert not conftest._is_service_manager_mutation_command( + ["systemctl", "show", "--property=Restart", "hermes-gateway"] + ) + + +def test_taskkill_image_filter_blocks_only_gateway_or_broad_python_images(): + conftest = _guard_module() + + assert conftest._is_process_killer_command( + ["taskkill.exe", "/IM", "hermes-gateway.exe"], + is_own_subtree=lambda _pid: False, + ) + assert conftest._is_process_killer_command( + ["taskkill", "/IM", "python*.exe"], + is_own_subtree=lambda _pid: False, + ) + assert not conftest._is_process_killer_command( + ["taskkill.exe", "/IM", "notepad.exe"], + is_own_subtree=lambda _pid: False, + ) diff --git a/tests/test_live_system_guard_self_test.py b/tests/test_live_system_guard_self_test.py index 0347d8510014c..e6fb4667db54b 100644 --- a/tests/test_live_system_guard_self_test.py +++ b/tests/test_live_system_guard_self_test.py @@ -29,6 +29,19 @@ FOREIGN_PID = 1 +def _record_guarded_run(monkeypatch): + """Replace only the guarded run wrapper's native delegate with a recorder.""" + calls = [] + + def _native_recorder(*args, **kwargs): + calls.append((args, kwargs)) + return types.SimpleNamespace(returncode=0, stdout="", stderr="") + + assert hasattr(subprocess.run, "__wrapped__"), "run wrapper is not installed" + monkeypatch.setattr(subprocess.run, "__wrapped__", _native_recorder) + return calls + + # ──────────────────── fail-closed self-protection ────────────── # # This file executes REAL kill primitives — os.kill(-1, SIGTERM), os.killpg, @@ -159,6 +172,21 @@ def test_subprocess_run_sh_c_systemctl_blocked(): subprocess.run(["sh", "-c", "systemctl --user stop hermes-gateway"]) +def test_subprocess_shell_wrapper_variants_are_blocked_before_execution(): + for command in ( + ["bash", "-lc", "systemctl restart hermes-gateway"], + ["cmd.exe", "/c", "systemctl stop hermes-gateway"], + ["powershell", "-Command", "Stop-Service hermes-gateway"], + ): + with pytest.raises(RuntimeError, match="live-system guard"): + subprocess.run(command) + + +def test_popen_sh_c_detached_gateway_spawn_is_blocked_before_execution(): + with pytest.raises(RuntimeError, match="live-system guard"): + subprocess.Popen(["sh", "-c", "hermes gateway run &"]) + + def test_subprocess_run_setsid_systemctl_blocked(): with pytest.raises(RuntimeError, match="live-system guard"): subprocess.run(["setsid", "systemctl", "kill", "hermes-gateway"]) @@ -177,6 +205,12 @@ def test_subprocess_popen_systemctl_blocked(): subprocess.Popen(["systemctl", "--user", "stop", "hermes-gateway"]) +def test_patched_popen_rejects_detached_gateway_before_native_execution(): + """The installed Popen wrapper must reject the spawn before its superclass.""" + with pytest.raises(RuntimeError, match="live-system guard"): + subprocess.Popen(["hermes", "gateway", "run"], start_new_session=True) + + def test_subprocess_call_systemctl_blocked(): with pytest.raises(RuntimeError, match="live-system guard"): subprocess.call(["systemctl", "--user", "restart", "hermes-gateway"]) @@ -275,81 +309,147 @@ def test_subprocess_killall_hermes_blocked(): subprocess.run(["killall", "hermes"]) +def test_patched_run_rejects_launchctl_gateway_mutations_before_native_execution(): + """The patched run wrapper blocks both launchd mutations without a native call.""" + for command in ( + ["launchctl", "kickstart", "gui/501/ai.hermes.gateway"], + ["launchctl", "bootout", "gui/501/ai.hermes.gateway"], + ): + with pytest.raises(RuntimeError, match="live-system guard"): + subprocess.run(command) + + +def test_patched_run_rejects_taskkill_foreign_pid_before_native_execution(): + with pytest.raises(RuntimeError, match="live-system guard"): + subprocess.run(["taskkill", "/PID", str(FOREIGN_PID), "/T", "/F"]) + + +def test_patched_run_rejects_taskkill_hermes_image_before_native_execution(): + with pytest.raises(RuntimeError, match="live-system guard"): + subprocess.run(["taskkill.exe", "/IM", "hermes-gateway.exe", "/F"]) + + +def test_patched_run_rejects_negative_kill_targets_before_native_execution(): + for command in ( + ["kill", "-9", "-1"], + ["kill", "-9", "-987654321"], + ["killpg", str(FOREIGN_PID), "9"], + ): + with pytest.raises(RuntimeError, match="live-system guard"): + subprocess.run(command) + + +def test_patched_run_rejects_non_read_only_systemctl_verbs_before_execution(): + for command in ( + ["systemctl", "reenable", "hermes-gateway.service"], + ["systemctl", "set-property", "hermes-gateway.service", "CPUQuota=1%"], + ): + with pytest.raises(RuntimeError, match="live-system guard"): + subprocess.run(command) + + # ──────────────────── pass-through cases (must NOT raise) ────── -def test_systemctl_status_passes_through(): +def test_systemctl_status_passes_through(monkeypatch): """Read-only systemctl probes (status/show/list-units) are fine.""" - # Run with check=False so we don't fail on the gateway's exit code. + calls = _record_guarded_run(monkeypatch) r = subprocess.run( ["systemctl", "--user", "status", "hermes-gateway", "--no-pager"], capture_output=True, text=True, check=False, ) - assert r is not None # Did not raise — the guard let it through. + assert r.returncode == 0 + assert len(calls) == 1 -def test_systemctl_show_passes_through(): +def test_systemctl_show_passes_through(monkeypatch): + calls = _record_guarded_run(monkeypatch) r = subprocess.run( ["systemctl", "--user", "show", "hermes-gateway", "--no-pager"], capture_output=True, text=True, check=False, ) - assert r is not None + assert r.returncode == 0 + assert len(calls) == 1 + +def test_required_pass_through_commands_are_delegated_without_execution(monkeypatch): + calls = _record_guarded_run(monkeypatch) + for command in ( + ["bash", "-lc", "echo hello"], + ["kill", "-TERM", str(os.getpid())], + ["systemctl", "--user", "status", "hermes-gateway"], + ["systemctl", "show", "--property=Restart", "hermes-gateway"], + ["taskkill", "/IM", "notepad.exe", "/F"], + ): + assert subprocess.run(command).returncode == 0 + assert len(calls) == 5 -def test_systemctl_list_units_passes_through(): + +def test_systemctl_list_units_passes_through(monkeypatch): + calls = _record_guarded_run(monkeypatch) r = subprocess.run( ["systemctl", "--user", "list-units", "fake-not-real-unit*", "--no-pager"], capture_output=True, text=True, check=False, ) - assert r is not None + assert r.returncode == 0 + assert len(calls) == 1 -def test_systemctl_unrelated_unit_passes_through(): +def test_systemctl_unrelated_unit_passes_through(monkeypatch): """systemctl restart of a non-hermes unit is allowed (we only protect hermes).""" - # Use --dry-run so we don't actually try to restart anything; just - # verify the guard doesn't block the call. systemctl supports - # --dry-run via the privileged API; on user scope it usually fails - # quickly without side effects. + calls = _record_guarded_run(monkeypatch) r = subprocess.run( ["systemctl", "--user", "show", "fake-not-real-unit"], capture_output=True, text=True, check=False, ) - assert r is not None + assert r.returncode == 0 + assert len(calls) == 1 + +def test_external_kill_of_own_subtree_passes_through(monkeypatch): + """A numeric external kill of this test process is delegated, never run natively.""" + calls = _record_guarded_run(monkeypatch) + r = subprocess.run(["kill", "-0", str(os.getpid())]) + assert r.returncode == 0 + assert len(calls) == 1 -def test_kill_own_subtree_passes_through(): - """We CAN kill our own children — guard recognizes them via psutil.""" - p = subprocess.Popen(["sleep", "30"]) - try: - os.kill(p.pid, signal.SIGTERM) - finally: - p.wait(timeout=2) - # SIGTERM = 15; subprocess returncode is -15 on POSIX. - assert p.returncode in {-signal.SIGTERM, 128 + int(signal.SIGTERM)} + +def test_launchctl_print_passes_through(monkeypatch): + calls = _record_guarded_run(monkeypatch) + r = subprocess.run(["launchctl", "print", "gui/501/ai.hermes.gateway"]) + assert r.returncode == 0 + assert len(calls) == 1 -def test_subprocess_pkill_with_unrelated_pattern_passes_through(): +def test_taskkill_benign_image_passes_through(monkeypatch): + calls = _record_guarded_run(monkeypatch) + r = subprocess.run(["taskkill.exe", "/IM", "notepad.exe"]) + assert r.returncode == 0 + assert len(calls) == 1 + + +def test_subprocess_pkill_with_unrelated_pattern_passes_through(monkeypatch): """``pkill -f some-unrelated-pattern`` (no hermes/python) is fine.""" - # We don't actually run pkill — just verify the guard would let it - # through by inspecting the matcher. Re-implementing the check here - # would duplicate the guard; instead spawn a noop to confirm no raise. - # Use 'true' so it succeeds quickly. - r = subprocess.run(["true"], capture_output=True) + calls = _record_guarded_run(monkeypatch) + r = subprocess.run(["pkill", "-f", "some-unrelated-pattern"], capture_output=True) assert r.returncode == 0 + assert len(calls) == 1 -def test_normal_subprocess_run_passes_through(): +def test_normal_subprocess_run_passes_through(monkeypatch): """Plain non-systemctl subprocess.run should work normally.""" + calls = _record_guarded_run(monkeypatch) r = subprocess.run(["echo", "hello"], capture_output=True, text=True) - assert r.stdout.strip() == "hello" + assert r.returncode == 0 + assert len(calls) == 1 # ──────────────────── bypass marker ───────────────────────────── @@ -360,10 +460,7 @@ def test_bypass_marker_disables_guard(): """The bypass marker exists for tests that genuinely need real signal delivery (e.g. PTY tests SIGINTing their own child). Verify it works. - We use it harmlessly here by signaling our own PID 0 (own group) so we - don't actually kill anything — but the call goes through real os.kill. + No signal is sent: the raw builtin identity is enough to prove bypass. """ - # With bypass, the guard yields without installing the monkeypatch, - # so we get the real os.kill. Calling os.kill(os.getpid(), 0) just - # checks that the PID exists — harmless. - os.kill(os.getpid(), 0) # No exception — guard is OFF. + # With bypass, the guard yields without installing the monkeypatch. + assert isinstance(os.kill, types.BuiltinFunctionType) diff --git a/tests/test_live_system_guard_update_hermetic.py b/tests/test_live_system_guard_update_hermetic.py new file mode 100644 index 0000000000000..3892f6a1a2a7b --- /dev/null +++ b/tests/test_live_system_guard_update_hermetic.py @@ -0,0 +1,259 @@ +"""Synthetic updater regressions for the live-system guard.""" + +from pathlib import Path +import sys +from types import SimpleNamespace + +import pytest + +from hermes_cli import config as hermes_config +from hermes_cli import gateway +from hermes_cli import gateway_windows +from hermes_cli import main as hermes_main +from gateway import status as gateway_status + + +SYNTHETIC_PID = 987_654_321 +NONEXISTENT_GATEWAY_EXECUTABLE = "/definitely-not-a-real-hermes-gateway" + + +def _guard_module(): + expected = Path(__file__).with_name("conftest.py").resolve() + for module in tuple(sys.modules.values()): + module_file = getattr(module, "__file__", None) + if module_file and Path(module_file).resolve() == expected: + return module + raise AssertionError("pytest did not load tests/conftest.py") + + +def _recording(name, calls, result): + def _fake(*args, **kwargs): + calls.append((name, args, kwargs)) + return result + + return _fake + + +def _prepare_update(monkeypatch, tmp_path, calls): + """Make ``cmd_update`` deterministic without starting any subprocess.""" + (tmp_path / ".git").mkdir() + monkeypatch.setattr(hermes_main, "PROJECT_ROOT", tmp_path) + monkeypatch.setattr(hermes_main, "_install_hangup_protection", lambda **_kw: None) + monkeypatch.setattr(hermes_main, "_finalize_update_output", lambda _state: None) + monkeypatch.setattr(hermes_main, "_run_pre_update_backup", lambda _args: None) + monkeypatch.setattr(hermes_main, "_discard_lockfile_churn", lambda *_args: None) + monkeypatch.setattr(hermes_main, "_get_origin_url", lambda *_args: "") + monkeypatch.setattr(hermes_main, "_refresh_active_lazy_features", lambda: None) + monkeypatch.setattr(hermes_main, "_kill_stale_dashboard_processes", lambda: None) + monkeypatch.setattr(hermes_main, "_is_fork", lambda _url: False) + monkeypatch.setattr(hermes_main, "_is_termux_env", lambda env=None: False) + monkeypatch.setattr(hermes_main, "_load_installable_optional_extras", lambda *_a: []) + monkeypatch.setattr(hermes_main._time, "sleep", lambda *_args, **_kwargs: None) + monkeypatch.setattr(hermes_config, "detect_install_method", lambda *_a: "git") + monkeypatch.setattr(hermes_config, "is_managed", lambda: False) + monkeypatch.setattr(hermes_config, "is_unsupported_install_method", lambda _m: False) + monkeypatch.setattr(hermes_config, "get_missing_env_vars", lambda **_kw: []) + monkeypatch.setattr(hermes_config, "get_missing_config_fields", lambda: []) + monkeypatch.setattr(hermes_config, "check_config_version", lambda: (5, 5)) + monkeypatch.setattr(hermes_config, "migrate_config", lambda **_kw: {}) + monkeypatch.setattr( + hermes_config, + "load_config", + lambda: {"updates": {"refresh_cua_driver": False}}, + ) + monkeypatch.setattr("hermes_cli.managed_uv.ensure_uv", lambda: "/synthetic/uv") + monkeypatch.setattr("hermes_cli.managed_uv.update_managed_uv", lambda: None) + monkeypatch.setattr( + "tools.skills_sync.sync_skills", + lambda **_kw: {"copied": [], "updated": [], "user_modified": []}, + ) + monkeypatch.setattr("hermes_cli.profiles.list_profiles", lambda: []) + + def fake_run(command, **kwargs): + calls.append(("subprocess.run", command, kwargs)) + if "rev-parse" in command: + return SimpleNamespace(stdout="main\n", stderr="", returncode=0) + if "rev-list" in command: + return SimpleNamespace(stdout="1\n", stderr="", returncode=0) + return SimpleNamespace(stdout="", stderr="", returncode=0) + + monkeypatch.setattr(hermes_main.subprocess, "run", fake_run) + monkeypatch.setattr( + hermes_main.subprocess, + "Popen", + _recording("subprocess.Popen", calls, SimpleNamespace(pid=SYNTHETIC_PID)), + ) + monkeypatch.setattr( + gateway, + "_prepare_profile_gateway_update_restart", + _recording("prepare_profile_restart", calls, None), + ) + monkeypatch.setattr( + gateway, + "_graceful_restart_via_sigusr1", + _recording("graceful_restart", calls, False), + ) + monkeypatch.setattr( + gateway, + "_wait_for_gateway_exit", + _recording("wait_for_gateway_exit", calls, None), + ) + + +def _run_update(): + hermes_main.cmd_update( + SimpleNamespace( + branch=None, + check=False, + force=True, + force_venv=True, + gateway=False, + yes=True, + ) + ) + + +def test_cmd_update_keeps_discovery_and_restart_boundaries_inert(monkeypatch, tmp_path): + """The updater must not reach discovery or restart boundaries in tests.""" + conftest = _guard_module() + calls = [] + _prepare_update(monkeypatch, tmp_path, calls) + + guarded_targets = [ + (gateway, "find_gateway_pids", []), + (gateway, "_scan_gateway_pids", []), + (gateway, "_get_service_pids", set()), + (gateway, "find_profile_gateway_processes", []), + (gateway, "supports_systemd_services", False), + (gateway, "is_macos", False), + (gateway_windows, "is_installed", False), + (gateway, "launch_detached_profile_gateway_restart", False), + (gateway, "launch_detached_gateway_restart_by_cmdline", False), + (gateway_windows, "_spawn_detached", None), + (gateway_status, "terminate_pid", None), + (gateway, "refresh_systemd_unit_if_needed", False), + (gateway, "refresh_launchd_plist_if_needed", False), + ] + for module, name, result in guarded_targets: + guarded = getattr(module, name) + assert getattr(guarded, "_live_system_guard_inert", False), name + monkeypatch.setattr(guarded, "__wrapped__", _recording(name, calls, result)) + + _run_update() + + guarded_names = {name for _module, name, _result in guarded_targets} + assert not [call for call in calls if call[0] in guarded_names] + assert not [ + call + for call in calls + if call[0] + in { + "subprocess.Popen", + "prepare_profile_restart", + "graceful_restart", + "wait_for_gateway_exit", + "terminate_pid", + } + ] + + +def test_updater_terminate_canary_fails_when_only_its_wrapper_is_removed(monkeypatch, tmp_path): + """The retained terminate wrapper is the only thing keeping its recorder idle.""" + calls = [] + _prepare_update(monkeypatch, tmp_path, calls) + + guarded_terminate = gateway_status.terminate_pid + assert getattr(guarded_terminate, "_live_system_guard_inert", False) + monkeypatch.setattr( + guarded_terminate, "__wrapped__", _recording("terminate_pid", calls, None) + ) + + def _exercise_termination_boundary(): + gateway_status.terminate_pid(SYNTHETIC_PID, force=True) + return None + + monkeypatch.setattr(hermes_main, "_pause_windows_gateways_for_update", _exercise_termination_boundary) + _run_update() + assert not [call for call in calls if call[0] == "terminate_pid"] + + with monkeypatch.context() as reverted: + reverted.setattr( + gateway_status, + "terminate_pid", + guarded_terminate.__wrapped__, + ) + _run_update() + with pytest.raises(AssertionError, match="unexpected live-system boundary"): + assert not [call for call in calls if call[0] == "terminate_pid"], ( + "unexpected live-system boundary" + ) + + +def test_updater_detached_spawn_canary_fails_when_only_its_wrapper_is_removed(monkeypatch, tmp_path): + """The retained detached-spawn wrapper is the only thing keeping its recorder idle.""" + calls = [] + _prepare_update(monkeypatch, tmp_path, calls) + + guarded_spawn = gateway.launch_detached_gateway_restart_by_cmdline + assert getattr(guarded_spawn, "_live_system_guard_inert", False) + monkeypatch.setattr( + guarded_spawn, "__wrapped__", _recording("detached_spawn", calls, False) + ) + + def _exercise_detached_spawn_boundary(_token): + return gateway.launch_detached_gateway_restart_by_cmdline( + SYNTHETIC_PID, + [NONEXISTENT_GATEWAY_EXECUTABLE, "gateway", "run"], + ) + + monkeypatch.setattr(hermes_main, "_resume_windows_gateways_after_update", _exercise_detached_spawn_boundary) + _run_update() + assert not [call for call in calls if call[0] == "detached_spawn"] + + with monkeypatch.context() as reverted: + reverted.setattr( + gateway, + "launch_detached_gateway_restart_by_cmdline", + guarded_spawn.__wrapped__, + ) + _run_update() + with pytest.raises(AssertionError, match="unexpected live-system boundary"): + assert not [call for call in calls if call[0] == "detached_spawn"], ( + "unexpected live-system boundary" + ) + + +@pytest.mark.parametrize( + "refresh_name", + ["refresh_systemd_unit_if_needed", "refresh_launchd_plist_if_needed"], +) +def test_updater_unit_write_canary_fails_closed_before_real_path_write( + monkeypatch, tmp_path, refresh_name +): + """A real unit path is rejected before the recording write delegate can run.""" + calls = [] + _prepare_update(monkeypatch, tmp_path, calls) + guarded_refresh = getattr(gateway, refresh_name) + assert getattr(guarded_refresh, "_live_system_guard_inert", False) + monkeypatch.setattr( + guarded_refresh, "__wrapped__", _recording("unit_write", calls, False) + ) + + def _exercise_unit_write_boundary(): + try: + getattr(gateway, refresh_name)() + except RuntimeError as exc: + assert "non-test path" in str(exc) + return None + + monkeypatch.setattr(hermes_main, "_pause_windows_gateways_for_update", _exercise_unit_write_boundary) + _run_update() + assert not [call for call in calls if call[0] == "unit_write"] + + with monkeypatch.context() as reverted: + reverted.setattr(gateway, refresh_name, guarded_refresh.__wrapped__) + _run_update() + with pytest.raises(AssertionError, match="unexpected live-system boundary"): + assert not [call for call in calls if call[0] == "unit_write"], ( + "unexpected live-system boundary" + )