From b6ab666b6deef400a8f73c733672cae4466193a8 Mon Sep 17 00:00:00 2001 From: "ang-fleet-workers[bot]" <333956806+ang-fleet-workers[bot]@users.noreply.github.com> Date: Mon, 5 Oct 2026 12:17:56 -0700 Subject: [PATCH] fix(gateway): a service definition is written only by the home it pins Two homes can resolve to one gateway service name: `/profiles/` and `~/.hermes/profiles/` both take `hermes-gateway-` (`ai.hermes.gateway-` on launchd). Every gateway boot, start and restart calls `refresh_systemd_unit_if_needed` / `refresh_launchd_plist_if_needed`, which rewrote whatever definition sat at that name with no check of the HERMES_HOME it pins. A scratch/E2E gateway started from a custom root therefore repointed the real install's unit at the scratch directory; the real gateway failed on its next restart (200/CHDIR once the scratch dir was cleaned up). `systemd_install --force` and `launchd_install --force` did the same. Uninstall already refused a unit pinning another home (`_systemd_unit_belongs_to_current_home`); the writers did not. - hermes_cli/gateway_service_owner.py (new sibling): `pinned_home()` reads the HERMES_HOME a unit or plist pins; `definition_belongs_to_home()` refuses a write from any other home; `refuse_foreign_home_install()` gates `gateway install` for a home outside the platform-default tree. - Writers refuse a definition pinned to another home: the systemd and launchd refreshes (boot/start/restart) and both installers, `--force` included. The system-scope check runs on the caller's own home, before `systemd_unit_is_current` adopts the unit's home into os.environ (else the check compares the unit with itself); a sudo-stripped caller with no HERMES_HOME still acts for the unit's pinned home, as before. - `gateway install` refuses a HERMES_HOME outside the account's Hermes tree that owns no bare service name and has no service installed yet: a service registered for a scratch home outlives the run. - New `gateway install --force-unit-path` overrides both refusals (an explicit repoint of the definition at this home). - Refused installs exit 1, so `gateway install` no longer goes on to start the other home's service; the setup wizard reports the refusal and continues, as ensure_gateway_service already did. - Existing tests whose sandbox HERMES_HOME is a tmp dir opt in to the admission check (5 files, one monkeypatch each). Tests: tests/hermes_cli/test_gateway_service_owner.py, 18 tests. On e36a818033 (sources reverted, tests kept): 11 failed, 7 passed (foreign unit/plist rewritten on refresh and --force, refused install starts the service, scratch install proceeds, no --force-unit-path). With this change: 18 passed. test_colliding_profile_names_share_a_unit pins the premise on the real name resolution. --- hermes_cli/gateway.py | 42 +++- hermes_cli/gateway_launchd.py | 12 +- hermes_cli/gateway_service_owner.py | 122 +++++++++ hermes_cli/gateway_setup_wizard.py | 5 + hermes_cli/subcommands/gateway.py | 3 + tests/hermes_cli/test_gateway.py | 5 +- .../test_gateway_multiplex_served_record.py | 2 + .../test_gateway_no_new_standalone_profile.py | 2 + tests/hermes_cli/test_gateway_service.py | 2 + .../hermes_cli/test_gateway_service_owner.py | 231 ++++++++++++++++++ .../test_launchd_install_start_now.py | 2 + 11 files changed, 418 insertions(+), 10 deletions(-) create mode 100644 hermes_cli/gateway_service_owner.py create mode 100644 tests/hermes_cli/test_gateway_service_owner.py diff --git a/hermes_cli/gateway.py b/hermes_cli/gateway.py index 9bce60513d..42797a3d9f 100644 --- a/hermes_cli/gateway.py +++ b/hermes_cli/gateway.py @@ -3452,6 +3452,12 @@ def refresh_systemd_unit_if_needed(system: bool = False) -> bool: if not unit_path.exists(): return False + # Runs on every gateway boot/start/restart. A home whose service name collides with another install's + # (``/profiles/x`` vs ``~/.hermes/profiles/x``) must not rewrite that install's unit. Checked on + # the caller's own home, before systemd_unit_is_current adopts the unit's home into os.environ. + from hermes_cli.gateway_service_owner import definition_belongs_to_home, service_home_for_unit + if not definition_belongs_to_home(unit_path, service_home_for_unit(unit_path, system), "rewrite"): + return False # systemd_unit_is_current is the HERMES_HOME-sync chokepoint; its env mutation persists for the regenerate below. current = systemd_unit_is_current(system=system) if _retire_hermes_replace_dropin(system=system): @@ -3640,10 +3646,25 @@ def systemd_install( run_as_user: str | None = None, enable_on_startup: bool = True, non_interactive: bool = False, + force_unit_path: bool = False, ): if system: _require_root_for_system_service("install") + # A unit pinning another home is another install's gateway: refuse before any removal, sync or write, + # --force included. --force-unit-path is the explicit repoint. + unit_path = get_systemd_unit_path(system=system) + if not force_unit_path: + from hermes_cli.gateway_service_owner import definition_belongs_to_home, service_home_for_unit + target_home = service_home_for_unit(unit_path, system) + if system and run_as_user: + try: + target_home = Path(_hermes_home_for_target_user(_system_service_identity(run_as_user)[2])) + except ValueError: + pass # unknown account: generate_systemd_unit refuses it below with the real error + if not definition_belongs_to_home(unit_path, target_home, "overwrite"): + sys.exit(1) + # Offer to remove legacy units first: alongside the new unit they flap-fight for the bot token. if has_legacy_hermes_units(): print() @@ -3653,15 +3674,15 @@ def systemd_install( remove_legacy_hermes_units(interactive=False) print() - unit_path = get_systemd_unit_path(system=system) scope_label = _service_scope_label(system) sudo, scope_flag, user_flag = _systemd_cli_bits(system) - # Existing system units already pin HERMES_HOME; adopt it before any regenerate. - if unit_path.exists(): + # Existing system units already pin HERMES_HOME; adopt it before any regenerate. A repoint keeps the + # caller's home (adopting the old one would regenerate the unit unchanged). + if unit_path.exists() and not force_unit_path: _sync_hermes_home_from_systemd_unit(system=system) - if unit_path.exists() and not force: + if unit_path.exists() and not (force or force_unit_path): if not systemd_unit_is_current(system=system): print(f"↻ Repairing outdated {scope_label} systemd service at: {unit_path}") refresh_systemd_unit_if_needed(system=system) @@ -5246,6 +5267,7 @@ def _install_systemd_from_cli(args, *, force: bool, system: bool, run_as_user) - systemd_install( force=force, system=system, run_as_user=run_as_user, enable_on_startup=start_on_login, non_interactive=non_interactive, + force_unit_path=getattr(args, "force_unit_path", False), ) if start_now: systemd_start(system=system) @@ -5266,12 +5288,18 @@ def _cmd_install(args): if _service_mgmt_blocked(): _no_backend_exit("install", "termux") backend = _service_backend() - if backend == "systemd": - if refuses_container_user_scope_install(system): + force_unit_path = getattr(args, "force_unit_path", False) + if backend == "systemd" and refuses_container_user_scope_install(system): + sys.exit(1) + if backend in ("systemd", "launchd"): + from hermes_cli.gateway_service_owner import refuse_foreign_home_install + if refuse_foreign_home_install(get_hermes_home(), force_unit_path): sys.exit(1) + if backend == "systemd": _install_systemd_from_cli(args, force=force, system=system, run_as_user=run_as_user) elif backend == "launchd": - launchd_install(force, start_now=getattr(args, "start_now", None) is not False) + launchd_install(force, start_now=getattr(args, "start_now", None) is not False, + force_unit_path=force_unit_path) elif backend == "windows": _gw_windows().install( force=force, diff --git a/hermes_cli/gateway_launchd.py b/hermes_cli/gateway_launchd.py index 5b6ab34465..608ad9b48f 100644 --- a/hermes_cli/gateway_launchd.py +++ b/hermes_cli/gateway_launchd.py @@ -540,6 +540,10 @@ def refresh_launchd_plist_if_needed() -> bool: plist_path = _gw().get_launchd_plist_path() if not plist_path.exists() or _gw().launchd_plist_is_current(): return False + # Runs on every gateway start/restart: a plist pinning another home is another install's gateway. + from hermes_cli.gateway_service_owner import definition_belongs_to_home + if not definition_belongs_to_home(plist_path, _gw().get_hermes_home(), "rewrite"): + return False new_plist = _gw().generate_launchd_plist() if _gw()._refuse_temp_home_service_write(new_plist, "launchd plist"): @@ -599,14 +603,18 @@ def refresh_launchd_plist_if_needed() -> bool: return True -def launchd_install(force: bool = False, *, start_now: bool = True): +def launchd_install(force: bool = False, *, start_now: bool = True, force_unit_path: bool = False): plist_path = _gw().get_launchd_plist_path() label = _gw().get_launchd_label() + # A plist pinning another home is another install's gateway, --force included; --force-unit-path repoints. + from hermes_cli.gateway_service_owner import definition_belongs_to_home + if not force_unit_path and not definition_belongs_to_home(plist_path, _gw().get_hermes_home(), "overwrite"): + sys.exit(1) # Loading the plist starts the gateway (RunAtLoad), so a no-start install writes it without # loading it. A gateway that launchd already runs is still reloaded; this install did not start it. load = start_now or _gw()._launchctl_label_supervising_process(label) - if plist_path.exists() and not force: + if plist_path.exists() and not (force or force_unit_path): if _gw().launchd_plist_is_current(): print(f"Service already installed at: {plist_path}") print("Use --force to reinstall") diff --git a/hermes_cli/gateway_service_owner.py b/hermes_cli/gateway_service_owner.py new file mode 100644 index 0000000000..14ce14d4e7 --- /dev/null +++ b/hermes_cli/gateway_service_owner.py @@ -0,0 +1,122 @@ +"""Who may write a gateway service definition (systemd unit, launchd plist). + +A service definition pins the ``HERMES_HOME`` its gateway runs with. Service names are derived from the +home (bare / profile name / path hash), and two homes can still resolve to the same name: a custom root's +``/profiles/`` takes the same ``hermes-gateway-`` unit as ``~/.hermes/profiles/``. +Every gateway boot refreshes "its" unit, so a scratch or E2E gateway started from such a home rewrote the +real install's unit to point at the scratch directory, and the real gateway failed on its next restart. + +The rule: a definition that pins a home is written only by a process acting for that home. Uninstall +already refused a unit pinning another home (``_systemd_unit_belongs_to_current_home``); the writers now +refuse the same way. ``hermes gateway install --force-unit-path`` is the explicit repoint. +""" + +from __future__ import annotations + +import os +import plistlib +from pathlib import Path + + +def pinned_home(definition_path: Path) -> str | None: + """``HERMES_HOME`` pinned by the systemd unit or launchd plist at *definition_path*, or None.""" + if definition_path.suffix == ".plist": + try: + with definition_path.open("rb") as fh: + data = plistlib.load(fh) + except Exception: # unreadable / not a plist (hand-edited, truncated): pins nobody + return None + env = data.get("EnvironmentVariables") if isinstance(data, dict) else None + value = env.get("HERMES_HOME") if isinstance(env, dict) else None + return (str(value).strip() or None) if value is not None else None + from hermes_cli import gateway as gw + return gw._hermes_home_pinned_by_unit(definition_path) + + +def _resolve(raw: str | Path) -> Path | None: + try: + return Path(raw).expanduser().resolve() + except (OSError, RuntimeError, ValueError): + return None + + +def service_home_for_unit(unit_path: Path, system: bool) -> Path: + """The ``HERMES_HOME`` this process acts for when it writes the systemd unit at *unit_path*. + + User scope: its own home. System scope: an explicit ``HERMES_HOME`` remapped to the unit's ``User=`` + account the way ``generate_systemd_unit`` pins it; with none (``sudo`` strips it, HOME=/root), the + unit's own pinned home, which is the only thing naming the install being operated on. Pure: it never + adopts the unit's home into ``os.environ``, so a check on this value compares caller and unit. + """ + from hermes_cli import gateway as gw + + if not system: + return gw.get_hermes_home() + if not os.environ.get("HERMES_HOME", "").strip(): + pinned = pinned_home(unit_path) if unit_path.exists() else None + return Path(pinned).expanduser() if pinned else gw.get_hermes_home() + user = gw._read_systemd_user_from_unit(unit_path) + if not user: + return gw.get_hermes_home() + try: + home_dir = gw._system_service_identity(run_as_user=user)[2] + except ValueError: # unknown User=: generate_systemd_unit refuses the same way + return gw.get_hermes_home() + return Path(gw._hermes_home_for_target_user(home_dir)) + + +def definition_belongs_to_home(definition_path: Path, home: Path, action: str) -> bool: + """False (with guidance printed) when the existing definition pins a ``HERMES_HOME`` other than *home*. + + A missing definition, or one that pins no home (hand-written, pre-pinning), belongs to nobody else. + """ + if not definition_path.exists(): + return True + raw = pinned_home(definition_path) + if raw is None or _resolve(raw) == _resolve(home): + return True + print(f"✗ Refusing to {action} {definition_path}: it runs HERMES_HOME={raw}, " + f"but this process has HERMES_HOME={home}.") + print(" That file is another install's gateway. Use that install, or pass --force-unit-path") + print(" to `hermes gateway install` if you really mean to repoint it at this home.") + return False + + +def _installed_definition_pins(home: Path) -> bool: + """True when a definition at this home's service path already pins *home* (a re-install).""" + from hermes_cli import gateway as gw + + paths = [gw.get_systemd_unit_path(system=False), gw.get_systemd_unit_path(system=True)] + if gw.is_macos(): + paths.append(gw.get_launchd_plist_path()) + for path in paths: + raw = pinned_home(path) if path.exists() else None + if raw is not None and _resolve(raw) == home: + return True + return False + + +def home_may_install_service(home: Path) -> bool: + """True for a home that may own a host gateway service: inside the platform-default Hermes tree + (``~/.hermes`` and its profiles, the sudo invoker's included), the bare-name owner, or a home whose + installed definition already pins it.""" + from hermes_cli import gateway as gw + + resolved = _resolve(home) + if resolved is None: + return False + if any(resolved == root or root in resolved.parents for root in gw._native_service_homes()): + return True + return gw._home_owns_bare_service_name(resolved) or _installed_definition_pins(resolved) + + +def refuse_foreign_home_install(home: Path, force_unit_path: bool) -> bool: + """True (with guidance printed) when ``hermes gateway install`` must not register a service for *home*.""" + if force_unit_path or home_may_install_service(home): + return False + print(f"✗ Refusing to install a gateway service for HERMES_HOME={home}.") + print(" It is outside this account's Hermes tree and no service is installed for it yet, so it looks") + print(" like a scratch / test / E2E home. A service installed for it would outlive that run and restart") + print(" a gateway on this host forever. Run that gateway in the foreground (`hermes gateway run`),") + print(" or pass --force-unit-path if this custom home really is meant to own a host service.") + return True diff --git a/hermes_cli/gateway_setup_wizard.py b/hermes_cli/gateway_setup_wizard.py index 9f702a1cf9..dd296a24a7 100644 --- a/hermes_cli/gateway_setup_wizard.py +++ b/hermes_cli/gateway_setup_wizard.py @@ -872,6 +872,11 @@ def _wizard_install_service(backend: str) -> None: except subprocess.CalledProcessError as e: _gw().print_error(f" Install failed: {e}") _gw().print_info(" You can try manually: hermes gateway install") + except SystemExit: + # The backend installers exit non-zero on a refused write (a unit pinning another home); + # setup reports it and goes on rather than ending the wizard. + _gw().print_warning(" Gateway service install did not complete.") + _gw().print_info(" You can try manually: hermes gateway install") def _wizard_post_setup() -> None: diff --git a/hermes_cli/subcommands/gateway.py b/hermes_cli/subcommands/gateway.py index 569e6fbe43..d67b26fb16 100644 --- a/hermes_cli/subcommands/gateway.py +++ b/hermes_cli/subcommands/gateway.py @@ -133,6 +133,9 @@ def build_gateway_parser( help="Do not enable the service to start on login/boot") _flag(gateway_install, "--if-missing", dest="if_missing", help="Do nothing when a gateway service is already installed") + _flag(gateway_install, "--force-unit-path", dest="force_unit_path", + help="Install for a HERMES_HOME outside this account's Hermes tree, or repoint a service " + "definition that pins another home") _flag(gateway_install, "--elevated-handoff", dest="elevated_handoff", help=argparse.SUPPRESS) gateway_uninstall = gateway_subparsers.add_parser("uninstall", help="Uninstall gateway service") diff --git a/tests/hermes_cli/test_gateway.py b/tests/hermes_cli/test_gateway.py index 081917600c..9a80be91c8 100644 --- a/tests/hermes_cli/test_gateway.py +++ b/tests/hermes_cli/test_gateway.py @@ -1136,7 +1136,10 @@ def test_install_if_missing_only_installs_when_no_service_exists(monkeypatch, in monkeypatch.setattr(gateway, "_guard_named_profile_under_multiplexer", lambda force: None) monkeypatch.setattr(gateway, "_service_mgmt_blocked", lambda: False) monkeypatch.setattr(gateway, "_service_backend", lambda: "launchd") - monkeypatch.setattr(gateway, "launchd_install", lambda force, start_now: installs.append(force)) + # The sandbox HERMES_HOME is a scratch home, which `gateway install` refuses by design. + monkeypatch.setattr("hermes_cli.gateway_service_owner.home_may_install_service", lambda home: True) + monkeypatch.setattr(gateway, "launchd_install", + lambda force, start_now, force_unit_path=False: installs.append(force)) gateway._cmd_install(SimpleNamespace(if_missing=True, force=False, system=False, run_as_user=None)) diff --git a/tests/hermes_cli/test_gateway_multiplex_served_record.py b/tests/hermes_cli/test_gateway_multiplex_served_record.py index ecc309eb87..ff4f362c96 100644 --- a/tests/hermes_cli/test_gateway_multiplex_served_record.py +++ b/tests/hermes_cli/test_gateway_multiplex_served_record.py @@ -161,6 +161,8 @@ def test_service_verbs_do_not_start_a_second_gateway(served_root, monkeypatch, v monkeypatch.setattr(gw, "_installed_service_kind_for", lambda *a, **k: "systemd") monkeypatch.setattr(gw, "is_managed", lambda: False) monkeypatch.setattr(gw, "is_termux", lambda: False) + # The served root is a tmp dir, which `gateway install` refuses as a scratch home by design. + monkeypatch.setattr("hermes_cli.gateway_service_owner.home_may_install_service", lambda home: True) fn = getattr(gw, f"_cmd_{verb}") ns = argparse.Namespace(system=False, all=False, force=False, run_as_user=None) if verb == "restart": diff --git a/tests/hermes_cli/test_gateway_no_new_standalone_profile.py b/tests/hermes_cli/test_gateway_no_new_standalone_profile.py index 356f32a3da..5881f52c67 100644 --- a/tests/hermes_cli/test_gateway_no_new_standalone_profile.py +++ b/tests/hermes_cli/test_gateway_no_new_standalone_profile.py @@ -39,6 +39,8 @@ def quiet_host(tmp_path, monkeypatch): monkeypatch.setattr(gw, "_dispatch_via_service_manager_if_s6", lambda v: False) monkeypatch.setattr(gw, "is_managed", lambda: False) monkeypatch.setattr(gw, "is_termux", lambda: False) + # tmp root is outside the platform-default tree, which `gateway install` refuses by design. + monkeypatch.setattr("hermes_cli.gateway_service_owner.home_may_install_service", lambda home: True) calls: list[str] = [] monkeypatch.setattr(gw, "_install_systemd_from_cli", lambda *a, **k: calls.append("install")) monkeypatch.setattr(gw, "_service_call", lambda backend, v, system: calls.append(v)) diff --git a/tests/hermes_cli/test_gateway_service.py b/tests/hermes_cli/test_gateway_service.py index 2850557cc7..eacd7e58d6 100644 --- a/tests/hermes_cli/test_gateway_service.py +++ b/tests/hermes_cli/test_gateway_service.py @@ -2071,6 +2071,8 @@ class TestDockerAwareGateway: monkeypatch.setattr(gateway_cli, "is_termux", lambda: False) monkeypatch.setattr(gateway_cli, "supports_systemd_services", lambda: True) monkeypatch.setattr(gateway_cli, "is_container", lambda: True) + # The sandbox HERMES_HOME is a scratch home, which `gateway install` refuses by design. + monkeypatch.setattr("hermes_cli.gateway_service_owner.home_may_install_service", lambda home: True) calls = [] monkeypatch.setattr( gateway_cli, diff --git a/tests/hermes_cli/test_gateway_service_owner.py b/tests/hermes_cli/test_gateway_service_owner.py new file mode 100644 index 0000000000..a9a039f951 --- /dev/null +++ b/tests/hermes_cli/test_gateway_service_owner.py @@ -0,0 +1,231 @@ +"""Gateway service-definition writes belong to the home the definition pins. + +Two homes can resolve to one service name: ``/profiles/`` and ``~/.hermes/profiles/`` +both take ``hermes-gateway-``. Every gateway boot refreshes "its" unit, so a scratch or E2E gateway +started from such a home rewrote the real install's unit (WorkingDirectory and HERMES_HOME pointed at the +scratch dir) and the real gateway failed on its next restart. The collision is simulated here by pointing +the unit/plist path at a file that pins another home; ``test_colliding_profile_names_share_a_unit`` proves +the collision itself on the real name resolution. +""" + +import argparse +import plistlib +from pathlib import Path +from types import SimpleNamespace + +import pytest + +import hermes_cli.gateway as gw +from hermes_cli import gateway_launchd, gateway_service_owner +from hermes_cli.subcommands.gateway import build_gateway_parser + +REAL_UNIT = ( + "[Service]\n" + "ExecStart=/home/ace/.hermes/hermes-agent/venv/bin/python -m hermes_cli.main gateway run\n" + "WorkingDirectory={home}\n" + 'Environment="HERMES_HOME={home}"\n' +) + + +@pytest.fixture +def homes(tmp_path, monkeypatch): + real = tmp_path / "account" / ".hermes" / "profiles" / "coder" + scratch = tmp_path / "ci-scratch" / "run-1" / "profiles" / "coder" + real.mkdir(parents=True) + scratch.mkdir(parents=True) + # The process is the scratch gateway. A scratch dir is not always under a temp root (CI scratch + # volumes), so the temp-home guard is out of the way: the ownership check is what is under test. + monkeypatch.setenv("HERMES_HOME", str(scratch)) + monkeypatch.setattr(gw, "_refuse_temp_home_service_write", lambda definition, kind: False) + return SimpleNamespace(real=real, scratch=scratch) + + +def test_colliding_profile_names_share_a_unit(homes, tmp_path, monkeypatch): + """The premise: a custom root's profile resolves to the same unit as the account's profile.""" + monkeypatch.setattr(gw, "_bare_unit_pinned_home", lambda: None) + monkeypatch.setattr(gw, "user_systemd_unit_dir", lambda: tmp_path / "systemd-user") + monkeypatch.setattr(gw, "_native_service_homes", lambda: {(tmp_path / "account" / ".hermes").resolve()}) + scratch_unit = gw.get_systemd_unit_path(system=False) + monkeypatch.setenv("HERMES_HOME", str(homes.real)) + assert gw.get_systemd_unit_path(system=False) == scratch_unit + + +@pytest.fixture +def systemd_unit(tmp_path, homes, monkeypatch): + unit = tmp_path / "systemd" / "hermes-gateway-coder.service" + unit.parent.mkdir() + unit.write_text(REAL_UNIT.format(home=homes.real), encoding="utf-8") + calls = [] + monkeypatch.setattr(gw, "get_systemd_unit_path", lambda system=False: unit) + monkeypatch.setattr(gw, "systemd_unit_is_current", lambda system=False: False) + monkeypatch.setattr(gw, "_retire_hermes_replace_dropin", lambda system=False: False) + monkeypatch.setattr(gw, "_prepare_service_launcher", lambda system=False, run_as_user=None: None) + monkeypatch.setattr(gw, "_run_systemctl", lambda args, **kw: calls.append(tuple(args)) or + SimpleNamespace(returncode=0, stdout="", stderr="")) + monkeypatch.setattr(gw, "has_legacy_hermes_units", lambda: False) + monkeypatch.setattr(gw, "_ensure_linger_enabled", lambda *a, **k: True) + monkeypatch.setattr(gw, "_ensure_system_service_linger", lambda *a, **k: None) + monkeypatch.setattr(gw, "print_systemd_scope_conflict_warning", lambda: None) + monkeypatch.setattr(gw, "print_legacy_unit_warning", lambda: None) + monkeypatch.setattr(gw, "generate_systemd_unit", lambda system=False, run_as_user=None: + REAL_UNIT.format(home=Path(gw.get_hermes_home()))) + return SimpleNamespace(path=unit, calls=calls, original=unit.read_text(encoding="utf-8")) + + +class TestSystemdWriters: + def test_boot_refresh_leaves_a_unit_pinned_to_another_home_untouched(self, systemd_unit, capsys): + assert gw.refresh_systemd_unit_if_needed(system=False) is False + assert systemd_unit.path.read_text(encoding="utf-8") == systemd_unit.original + assert ("daemon-reload",) not in systemd_unit.calls + assert "Refusing to rewrite" in capsys.readouterr().out + + @pytest.fixture + def marker_free_unit(self, monkeypatch): + # refresh's own test belt refuses a generated unit naming a pytest tmpdir; any marker-free body works. + monkeypatch.setattr(gw, "generate_systemd_unit", lambda system=False, run_as_user=None: "ExecStart=new\n") + + def test_boot_refresh_still_rewrites_its_own_stale_unit(self, systemd_unit, homes, marker_free_unit, + monkeypatch): + monkeypatch.setenv("HERMES_HOME", str(homes.real)) + assert gw.refresh_systemd_unit_if_needed(system=False) is True + assert systemd_unit.path.read_text(encoding="utf-8") == "ExecStart=new\n" + assert ("daemon-reload",) in systemd_unit.calls + + def test_unpinned_unit_is_nobody_elses(self, systemd_unit, marker_free_unit): + systemd_unit.path.write_text("[Service]\nExecStart=/usr/bin/hermes gateway run\n", encoding="utf-8") + assert gw.refresh_systemd_unit_if_needed(system=False) is True + + def test_force_install_does_not_overwrite_another_homes_unit(self, systemd_unit): + with pytest.raises(SystemExit) as exc: + gw.systemd_install(force=True, non_interactive=True) + assert exc.value.code == 1 + assert systemd_unit.path.read_text(encoding="utf-8") == systemd_unit.original + assert systemd_unit.calls == [] + + def test_refused_install_never_starts_the_other_homes_service(self, systemd_unit, monkeypatch): + started = [] + monkeypatch.setattr(gw, "systemd_start", lambda system=False: started.append(system)) + args = SimpleNamespace(start_now=True, start_on_login=True, force_unit_path=False) + with pytest.raises(SystemExit): + gw._install_systemd_from_cli(args, force=False, system=False, run_as_user=None) + assert started == [] + + def test_force_unit_path_repoints_the_unit(self, systemd_unit, homes): + gw.systemd_install(force_unit_path=True, non_interactive=True) + assert f"HERMES_HOME={homes.scratch}" in systemd_unit.path.read_text(encoding="utf-8") + + def test_system_refresh_checks_the_callers_home_before_adopting_the_units( + self, systemd_unit, homes, monkeypatch): + # systemd_unit_is_current adopts a system unit's HERMES_HOME into os.environ (sudo strips it); + # checking after that adoption compared the unit with itself. + monkeypatch.setattr(gw, "_read_systemd_user_from_unit", lambda path: None) + + def is_current(system=False): + gw._sync_hermes_home_from_systemd_unit(system=system) + return False + monkeypatch.setattr(gw, "systemd_unit_is_current", is_current) + assert gw.refresh_systemd_unit_if_needed(system=True) is False + assert systemd_unit.path.read_text(encoding="utf-8") == systemd_unit.original + assert Path(gw.get_hermes_home()) == homes.scratch + + def test_sudo_stripped_home_still_refreshes_the_system_unit(self, systemd_unit, homes, marker_free_unit, + monkeypatch): + # Under sudo there is no explicit HERMES_HOME: the unit's own pinned home names the install. + monkeypatch.delenv("HERMES_HOME") + monkeypatch.setattr(gw, "_read_systemd_user_from_unit", lambda path: None) + assert gw.refresh_systemd_unit_if_needed(system=True) is True + + +class TestLaunchdWriters: + @pytest.fixture + def plist(self, tmp_path, homes, monkeypatch): + path = tmp_path / "LaunchAgents" / "ai.hermes.gateway-coder.plist" + path.parent.mkdir() + data = {"Label": "ai.hermes.gateway-coder", "EnvironmentVariables": {"HERMES_HOME": str(homes.real)}} + path.write_bytes(plistlib.dumps(data)) + monkeypatch.setattr(gw, "get_launchd_plist_path", lambda: path) + monkeypatch.setattr(gw, "get_launchd_label", lambda: "ai.hermes.gateway-coder") + monkeypatch.setattr(gw, "launchd_plist_is_current", lambda: False) + monkeypatch.setattr(gw, "generate_launchd_plist", lambda: "new") + monkeypatch.setattr(gw, "_prepare_service_launcher", lambda *a, **k: None) + monkeypatch.setattr(gw, "_launchctl_label_supervising_process", lambda label: None) + monkeypatch.setattr(gateway_launchd.subprocess, "run", lambda *a, **k: SimpleNamespace(returncode=0)) + return SimpleNamespace(path=path, original=path.read_bytes()) + + def test_refresh_leaves_a_plist_pinned_to_another_home_untouched(self, plist): + assert gateway_launchd.refresh_launchd_plist_if_needed() is False + assert plist.path.read_bytes() == plist.original + + def test_force_install_does_not_overwrite_another_homes_plist(self, plist): + with pytest.raises(SystemExit): + gateway_launchd.launchd_install(force=True, start_now=False) + assert plist.path.read_bytes() == plist.original + + def test_force_unit_path_repoints_the_plist(self, plist): + gateway_launchd.launchd_install(start_now=False, force_unit_path=True) + assert plist.path.read_text(encoding="utf-8") == "new" + + +class TestInstallAdmission: + """`gateway install` registers a persistent host service; a scratch/E2E home must not get one.""" + + @pytest.fixture + def install_cli(self, homes, tmp_path, monkeypatch): + installs = [] + monkeypatch.setattr(gw, "is_managed", lambda: False) + monkeypatch.setattr(gw, "_service_mgmt_blocked", lambda: False) + monkeypatch.setattr(gw, "_guard_named_profile_under_multiplexer", lambda force: None) + monkeypatch.setattr(gw, "_service_backend", lambda: "launchd") + monkeypatch.setattr(gw, "is_macos", lambda: True) + monkeypatch.setattr(gw, "_native_service_homes", lambda: {(tmp_path / "account" / ".hermes").resolve()}) + monkeypatch.setattr(gw, "_bare_unit_pinned_home", lambda: None) + monkeypatch.setattr(gw, "get_systemd_unit_path", lambda system=False: tmp_path / "none.service") + monkeypatch.setattr(gw, "get_launchd_plist_path", lambda: tmp_path / "none.plist") + monkeypatch.setattr(gw, "launchd_install", lambda force, start_now, force_unit_path=False: + installs.append(force_unit_path)) + return installs + + @staticmethod + def _args(**kw): + return SimpleNamespace(if_missing=False, force=False, system=False, run_as_user=None, **kw) + + def test_scratch_home_is_refused(self, install_cli): + with pytest.raises(SystemExit) as exc: + gw._cmd_install(self._args(force_unit_path=False)) + assert exc.value.code == 1 + assert install_cli == [] + + def test_force_unit_path_admits_a_custom_home(self, install_cli): + gw._cmd_install(self._args(force_unit_path=True)) + assert install_cli == [True] + + def test_account_profile_is_admitted(self, install_cli, homes, monkeypatch): + monkeypatch.setenv("HERMES_HOME", str(homes.real)) + gw._cmd_install(self._args(force_unit_path=False)) + assert install_cli == [False] + + def test_custom_home_with_its_own_installed_service_is_admitted(self, install_cli, homes, tmp_path, + monkeypatch): + own = tmp_path / "own.plist" + own.write_bytes(plistlib.dumps({"EnvironmentVariables": {"HERMES_HOME": str(homes.scratch)}})) + monkeypatch.setattr(gw, "get_launchd_plist_path", lambda: own) + gw._cmd_install(self._args(force_unit_path=False)) + assert install_cli == [False] + + def test_parser_accepts_force_unit_path(self): + parser = argparse.ArgumentParser() + build_gateway_parser(parser.add_subparsers(dest="command"), cmd_gateway=lambda a: None, + cmd_proxy=lambda a: None, cmd_gateway_enroll=lambda a: None) + assert parser.parse_args(["gateway", "install", "--force-unit-path"]).force_unit_path is True + + +def test_pinned_home_reads_both_definition_kinds(tmp_path): + unit = tmp_path / "g.service" + unit.write_text('[Service]\nEnvironment="HERMES_HOME=/srv/a"\n', encoding="utf-8") + plist = tmp_path / "g.plist" + plist.write_bytes(plistlib.dumps({"EnvironmentVariables": {"HERMES_HOME": "/srv/b"}})) + broken = tmp_path / "broken.plist" + broken.write_text("not a plist", encoding="utf-8") + assert gateway_service_owner.pinned_home(unit) == "/srv/a" + assert gateway_service_owner.pinned_home(plist) == "/srv/b" + assert gateway_service_owner.pinned_home(broken) is None diff --git a/tests/hermes_cli/test_launchd_install_start_now.py b/tests/hermes_cli/test_launchd_install_start_now.py index 7ead1f1b13..2e208f1c4e 100644 --- a/tests/hermes_cli/test_launchd_install_start_now.py +++ b/tests/hermes_cli/test_launchd_install_start_now.py @@ -41,6 +41,8 @@ def launchd(tmp_path, monkeypatch): monkeypatch.setattr(gateway_cli, "get_launchd_label", lambda: LABEL) monkeypatch.setattr(gateway_cli, "_launchd_domain", lambda: DOMAIN) monkeypatch.setattr(gateway_cli, "_refuse_temp_home_service_write", lambda *a: False) + # The sandbox HERMES_HOME is a scratch home, which `gateway install` refuses by design. + monkeypatch.setattr("hermes_cli.gateway_service_owner.home_may_install_service", lambda home: True) monkeypatch.setattr(gateway_cli, "_clear_launchd_unsupported_marker", lambda: None) monkeypatch.setattr(gateway_cli, "_launchctl_supervised_pid", lambda label: state.supervised_pid) monkeypatch.setattr(gateway_cli, "_launchctl_bootstrap", lambda *a, **k: state.bootstraps.append(a))