fix(gateway): refuse to uninstall a systemd unit pinned to another HERMES_HOME

Defence at the exact boundary the incident crossed: systemd_uninstall() and
uninstall._remove_systemd_gateway() unlinked whatever get_systemd_unit_path()
returned. Before stop/disable/unlink, read the unit's own
Environment="HERMES_HOME=..." line (the parser status/refresh already use)
and, when it names a different home than this process, warn with both paths
and leave the unit alone. A unit without the line (hand-written) is still
removed as before.
This commit is contained in:
Teknium
2026-09-09 06:02:53 -07:00
parent 1fc033a8ef
commit 19f2f19987
3 changed files with 39 additions and 2 deletions
+16
View File
@@ -3218,8 +3218,24 @@ def _systemd_scope_preamble(
return system
def _systemd_unit_belongs_to_current_home(system: bool = False) -> bool:
"""False (with a warning) when the installed unit pins a HERMES_HOME other than this process's: the
service name then resolved to ANOTHER install's gateway, and stop/disable/unlink would take it down."""
_sync_hermes_home_from_systemd_unit(system=system) # sudo strips HERMES_HOME; adopt the unit's first
unit_home = _hermes_home_from_systemd_unit_file(system=system)
if unit_home is None or Path(unit_home).expanduser().resolve() == get_hermes_home().resolve():
return True
print_warning(
f"Refusing to remove {get_systemd_unit_path(system=system)}: it runs HERMES_HOME={unit_home}, "
f"but this process has HERMES_HOME={get_hermes_home()}"
)
return False
def systemd_uninstall(system: bool = False):
system = _systemd_scope_preamble("uninstall", system, require_installed=False)
if not _systemd_unit_belongs_to_current_home(system):
return
_run_systemctl(["stop", get_service_name()], system=system, check=False, timeout=90)
_run_systemctl(["disable", get_service_name()], system=system, check=False, timeout=30)
+4 -2
View File
@@ -192,12 +192,14 @@ def uninstall_gateway_service():
def _remove_systemd_gateway() -> bool:
"""Linux: uninstall systemd services (both user and system scopes)."""
from hermes_cli.gateway import _systemctl_cmd, get_service_name, get_systemd_unit_path
from hermes_cli.gateway import (
_systemctl_cmd, _systemd_unit_belongs_to_current_home, get_service_name, get_systemd_unit_path,
)
svc_name = get_service_name()
removed_any = False
for is_system, scope in ((False, "user"), (True, "system")):
unit_path = get_systemd_unit_path(system=is_system)
if not unit_path.exists():
if not unit_path.exists() or not _systemd_unit_belongs_to_current_home(is_system):
continue
try:
if is_system and os.geteuid() != 0: # windows-footgun: ok — Linux-only systemd path
+19
View File
@@ -218,6 +218,25 @@ class TestServiceIdentityForForeignHome:
assert gateway_cli.get_service_name() == "hermes-gateway-alpha"
class TestUninstallRefusesForeignUnit:
"""systemd_uninstall must not stop/disable/unlink a unit pinned to another HERMES_HOME."""
def test_unit_for_other_home_is_left_alone(self, tmp_path, monkeypatch, capsys):
unit_path = tmp_path / "hermes-gateway.service"
unit_path.write_text('[Service]\nEnvironment="HERMES_HOME=/somewhere/else"\n', encoding="utf-8")
monkeypatch.setenv("HERMES_HOME", str(tmp_path / "mine"))
monkeypatch.setattr(gateway_cli, "get_systemd_unit_path", lambda system=False: unit_path)
monkeypatch.setattr(gateway_cli, "_systemd_scope_preamble", lambda *a, **k: False)
calls = []
monkeypatch.setattr(gateway_cli, "_run_systemctl", lambda args, **k: calls.append(args))
gateway_cli.systemd_uninstall(system=False)
assert unit_path.exists()
assert calls == []
assert "/somewhere/else" in capsys.readouterr().out
class TestGetCronDrainTimeout:
def test_missing_config_falls_back_to_default(self, monkeypatch):
monkeypatch.delenv("HERMES_CRON_DRAIN_TIMEOUT", raising=False)