fix(gateway): harden Docker/container gateway pathway

Centralize container detection in hermes_constants.is_container() with
process-lifetime caching, matching existing is_wsl()/is_termux() patterns.
Dedup _is_inside_container() in config.py to delegate to the new function.

Add _run_systemctl() wrapper that converts FileNotFoundError to RuntimeError
for defense-in-depth — all 10 bare subprocess.run(_systemctl_cmd(...)) call
sites now route through it.

Make supports_systemd_services() return False in containers and when
systemctl binary is absent (shutil.which check).

Add Docker-specific guidance in gateway_command() for install/uninstall/start
subcommands — exit 0 with helpful instructions instead of crashing.

Make 'hermes status' show 'Manager: docker (foreground)' and 'hermes dump'
show 'running (docker, pid N)' inside containers.

Fix setup_gateway() to use supports_systemd instead of _is_linux for all
systemd-related branches, and show Docker restart policy instructions in
containers.

Replace inline /.dockerenv check in voice_mode.py with is_container().

Fixes #7420

Co-authored-by: teknium1 <teknium1@users.noreply.github.com>
This commit is contained in:
alt-glitch 2026-04-12 14:42:46 -07:00 committed by Teknium
parent 18ab5c99d1
commit 5e1197a42e
11 changed files with 428 additions and 125 deletions

View file

@ -148,25 +148,6 @@ def managed_error(action: str = "modify configuration"):
# Container-aware CLI (NixOS container mode) # Container-aware CLI (NixOS container mode)
# ============================================================================= # =============================================================================
def _is_inside_container() -> bool:
"""Detect if we're already running inside a Docker/Podman container."""
# Standard Docker/Podman indicators
if os.path.exists("/.dockerenv"):
return True
# Podman uses /run/.containerenv
if os.path.exists("/run/.containerenv"):
return True
# Check cgroup for container runtime evidence (works for both Docker & Podman)
try:
with open("/proc/1/cgroup", "r") as f:
cgroup = f.read()
if "docker" in cgroup or "podman" in cgroup or "/lxc/" in cgroup:
return True
except OSError:
pass
return False
def get_container_exec_info() -> Optional[dict]: def get_container_exec_info() -> Optional[dict]:
"""Read container mode metadata from HERMES_HOME/.container-mode. """Read container mode metadata from HERMES_HOME/.container-mode.
@ -181,7 +162,8 @@ def get_container_exec_info() -> Optional[dict]:
if os.environ.get("HERMES_DEV") == "1": if os.environ.get("HERMES_DEV") == "1":
return None return None
if _is_inside_container(): from hermes_constants import is_container
if is_container():
return None return None
container_mode_file = get_hermes_home() / ".container-mode" container_mode_file = get_hermes_home() / ".container-mode"

View file

@ -44,6 +44,16 @@ def _redact(value: str) -> str:
def _gateway_status() -> str: def _gateway_status() -> str:
"""Return a short gateway status string.""" """Return a short gateway status string."""
if sys.platform.startswith("linux"): if sys.platform.startswith("linux"):
from hermes_constants import is_container
if is_container():
try:
from hermes_cli.gateway import find_gateway_pids
pids = find_gateway_pids()
if pids:
return f"running (docker, pid {pids[0]})"
return "stopped (docker)"
except Exception:
return "stopped (docker)"
try: try:
from hermes_cli.gateway import get_service_name from hermes_cli.gateway import get_service_name
svc = get_service_name() svc = get_service_name()

View file

@ -331,7 +331,7 @@ def is_linux() -> bool:
return sys.platform.startswith('linux') return sys.platform.startswith('linux')
from hermes_constants import is_termux, is_wsl from hermes_constants import is_container, is_termux, is_wsl
def _wsl_systemd_operational() -> bool: def _wsl_systemd_operational() -> bool:
@ -353,7 +353,9 @@ def _wsl_systemd_operational() -> bool:
def supports_systemd_services() -> bool: def supports_systemd_services() -> bool:
if not is_linux() or is_termux(): if not is_linux() or is_termux() or is_container():
return False
if shutil.which("systemctl") is None:
return False return False
if is_wsl(): if is_wsl():
return _wsl_systemd_operational() return _wsl_systemd_operational()
@ -483,6 +485,21 @@ def _journalctl_cmd(system: bool = False) -> list[str]:
return ["journalctl"] if system else ["journalctl", "--user"] return ["journalctl"] if system else ["journalctl", "--user"]
def _run_systemctl(args: list[str], *, system: bool = False, **kwargs) -> subprocess.CompletedProcess:
"""Run a systemctl command, raising RuntimeError if systemctl is missing.
Defense-in-depth: callers are gated by ``supports_systemd_services()``,
but this ensures any future caller that bypasses the gate still gets a
clear error instead of a raw ``FileNotFoundError`` traceback.
"""
try:
return subprocess.run(_systemctl_cmd(system) + args, **kwargs)
except FileNotFoundError:
raise RuntimeError(
"systemctl is not available on this system"
) from None
def _service_scope_label(system: bool = False) -> str: def _service_scope_label(system: bool = False) -> str:
return "system" if system else "user" return "system" if system else "user"
@ -929,7 +946,7 @@ def refresh_systemd_unit_if_needed(system: bool = False) -> bool:
expected_user = _read_systemd_user_from_unit(unit_path) if system else None expected_user = _read_systemd_user_from_unit(unit_path) if system else None
unit_path.write_text(generate_systemd_unit(system=system, run_as_user=expected_user), encoding="utf-8") unit_path.write_text(generate_systemd_unit(system=system, run_as_user=expected_user), encoding="utf-8")
subprocess.run(_systemctl_cmd(system) + ["daemon-reload"], check=True, timeout=30) _run_systemctl(["daemon-reload"], system=system, check=True, timeout=30)
print(f"↻ Updated gateway {_service_scope_label(system)} service definition to match the current Hermes install") print(f"↻ Updated gateway {_service_scope_label(system)} service definition to match the current Hermes install")
return True return True
@ -1025,7 +1042,7 @@ def systemd_install(force: bool = False, system: bool = False, run_as_user: str
if not systemd_unit_is_current(system=system): if not systemd_unit_is_current(system=system):
print(f"↻ Repairing outdated {_service_scope_label(system)} systemd service at: {unit_path}") print(f"↻ Repairing outdated {_service_scope_label(system)} systemd service at: {unit_path}")
refresh_systemd_unit_if_needed(system=system) refresh_systemd_unit_if_needed(system=system)
subprocess.run(_systemctl_cmd(system) + ["enable", get_service_name()], check=True, timeout=30) _run_systemctl(["enable", get_service_name()], system=system, check=True, timeout=30)
print(f"{_service_scope_label(system).capitalize()} service definition updated") print(f"{_service_scope_label(system).capitalize()} service definition updated")
return return
print(f"Service already installed at: {unit_path}") print(f"Service already installed at: {unit_path}")
@ -1036,8 +1053,8 @@ def systemd_install(force: bool = False, system: bool = False, run_as_user: str
print(f"Installing {_service_scope_label(system)} systemd service to: {unit_path}") print(f"Installing {_service_scope_label(system)} systemd service to: {unit_path}")
unit_path.write_text(generate_systemd_unit(system=system, run_as_user=run_as_user), encoding="utf-8") unit_path.write_text(generate_systemd_unit(system=system, run_as_user=run_as_user), encoding="utf-8")
subprocess.run(_systemctl_cmd(system) + ["daemon-reload"], check=True, timeout=30) _run_systemctl(["daemon-reload"], system=system, check=True, timeout=30)
subprocess.run(_systemctl_cmd(system) + ["enable", get_service_name()], check=True, timeout=30) _run_systemctl(["enable", get_service_name()], system=system, check=True, timeout=30)
print() print()
print(f"{_service_scope_label(system).capitalize()} service installed and enabled!") print(f"{_service_scope_label(system).capitalize()} service installed and enabled!")
@ -1063,15 +1080,15 @@ def systemd_uninstall(system: bool = False):
if system: if system:
_require_root_for_system_service("uninstall") _require_root_for_system_service("uninstall")
subprocess.run(_systemctl_cmd(system) + ["stop", get_service_name()], check=False, timeout=90) _run_systemctl(["stop", get_service_name()], system=system, check=False, timeout=90)
subprocess.run(_systemctl_cmd(system) + ["disable", get_service_name()], check=False, timeout=30) _run_systemctl(["disable", get_service_name()], system=system, check=False, timeout=30)
unit_path = get_systemd_unit_path(system=system) unit_path = get_systemd_unit_path(system=system)
if unit_path.exists(): if unit_path.exists():
unit_path.unlink() unit_path.unlink()
print(f"✓ Removed {unit_path}") print(f"✓ Removed {unit_path}")
subprocess.run(_systemctl_cmd(system) + ["daemon-reload"], check=True, timeout=30) _run_systemctl(["daemon-reload"], system=system, check=True, timeout=30)
print(f"{_service_scope_label(system).capitalize()} service uninstalled") print(f"{_service_scope_label(system).capitalize()} service uninstalled")
@ -1080,7 +1097,7 @@ def systemd_start(system: bool = False):
if system: if system:
_require_root_for_system_service("start") _require_root_for_system_service("start")
refresh_systemd_unit_if_needed(system=system) refresh_systemd_unit_if_needed(system=system)
subprocess.run(_systemctl_cmd(system) + ["start", get_service_name()], check=True, timeout=30) _run_systemctl(["start", get_service_name()], system=system, check=True, timeout=30)
print(f"{_service_scope_label(system).capitalize()} service started") print(f"{_service_scope_label(system).capitalize()} service started")
@ -1089,7 +1106,7 @@ def systemd_stop(system: bool = False):
system = _select_systemd_scope(system) system = _select_systemd_scope(system)
if system: if system:
_require_root_for_system_service("stop") _require_root_for_system_service("stop")
subprocess.run(_systemctl_cmd(system) + ["stop", get_service_name()], check=True, timeout=90) _run_systemctl(["stop", get_service_name()], system=system, check=True, timeout=90)
print(f"{_service_scope_label(system).capitalize()} service stopped") print(f"{_service_scope_label(system).capitalize()} service stopped")
@ -1105,7 +1122,7 @@ def systemd_restart(system: bool = False):
if pid is not None and _request_gateway_self_restart(pid): if pid is not None and _request_gateway_self_restart(pid):
print(f"{_service_scope_label(system).capitalize()} service restart requested") print(f"{_service_scope_label(system).capitalize()} service restart requested")
return return
subprocess.run(_systemctl_cmd(system) + ["reload-or-restart", get_service_name()], check=True, timeout=90) _run_systemctl(["reload-or-restart", get_service_name()], system=system, check=True, timeout=90)
print(f"{_service_scope_label(system).capitalize()} service restarted") print(f"{_service_scope_label(system).capitalize()} service restarted")
@ -1129,14 +1146,16 @@ def systemd_status(deep: bool = False, system: bool = False):
print(f" Run: {'sudo ' if system else ''}hermes gateway restart{scope_flag} # auto-refreshes the unit") print(f" Run: {'sudo ' if system else ''}hermes gateway restart{scope_flag} # auto-refreshes the unit")
print() print()
subprocess.run( _run_systemctl(
_systemctl_cmd(system) + ["status", get_service_name(), "--no-pager"], ["status", get_service_name(), "--no-pager"],
system=system,
capture_output=False, capture_output=False,
timeout=10, timeout=10,
) )
result = subprocess.run( result = _run_systemctl(
_systemctl_cmd(system) + ["is-active", get_service_name()], ["is-active", get_service_name()],
system=system,
capture_output=True, capture_output=True,
text=True, text=True,
timeout=10, timeout=10,
@ -2123,24 +2142,24 @@ def _is_service_running() -> bool:
if user_unit_exists: if user_unit_exists:
try: try:
result = subprocess.run( result = _run_systemctl(
_systemctl_cmd(False) + ["is-active", get_service_name()], ["is-active", get_service_name()],
capture_output=True, text=True, timeout=10, system=False, capture_output=True, text=True, timeout=10,
) )
if result.stdout.strip() == "active": if result.stdout.strip() == "active":
return True return True
except subprocess.TimeoutExpired: except (RuntimeError, subprocess.TimeoutExpired):
pass pass
if system_unit_exists: if system_unit_exists:
try: try:
result = subprocess.run( result = _run_systemctl(
_systemctl_cmd(True) + ["is-active", get_service_name()], ["is-active", get_service_name()],
capture_output=True, text=True, timeout=10, system=True, capture_output=True, text=True, timeout=10,
) )
if result.stdout.strip() == "active": if result.stdout.strip() == "active":
return True return True
except subprocess.TimeoutExpired: except (RuntimeError, subprocess.TimeoutExpired):
pass pass
return False return False
@ -2774,6 +2793,15 @@ def gateway_command(args):
print(" tmux new -s hermes 'hermes gateway run' # persistent via tmux") print(" tmux new -s hermes 'hermes gateway run' # persistent via tmux")
print(" nohup hermes gateway run > ~/.hermes/logs/gateway.log 2>&1 & # background") print(" nohup hermes gateway run > ~/.hermes/logs/gateway.log 2>&1 & # background")
sys.exit(1) sys.exit(1)
elif is_container():
print("Service installation is not needed inside a Docker container.")
print("The container runtime is your service manager — use Docker restart policies instead:")
print()
print(" docker run --restart unless-stopped ... # auto-restart on crash/reboot")
print(" docker restart <container> # manual restart")
print()
print("To run the gateway: hermes gateway run")
sys.exit(0)
else: else:
print("Service installation not supported on this platform.") print("Service installation not supported on this platform.")
print("Run manually: hermes gateway run") print("Run manually: hermes gateway run")
@ -2792,6 +2820,13 @@ def gateway_command(args):
systemd_uninstall(system=system) systemd_uninstall(system=system)
elif is_macos(): elif is_macos():
launchd_uninstall() launchd_uninstall()
elif is_container():
print("Service uninstall is not applicable inside a Docker container.")
print("To stop the gateway, stop or remove the container:")
print()
print(" docker stop <container>")
print(" docker rm <container>")
sys.exit(0)
else: else:
print("Not supported on this platform.") print("Not supported on this platform.")
sys.exit(1) sys.exit(1)
@ -2816,6 +2851,15 @@ def gateway_command(args):
print() print()
print("To enable systemd: add systemd=true to /etc/wsl.conf and run 'wsl --shutdown' from PowerShell.") print("To enable systemd: add systemd=true to /etc/wsl.conf and run 'wsl --shutdown' from PowerShell.")
sys.exit(1) sys.exit(1)
elif is_container():
print("Service start is not applicable inside a Docker container.")
print("The gateway runs as the container's main process.")
print()
print(" docker start <container> # start a stopped container")
print(" docker restart <container> # restart a running container")
print()
print("Or run the gateway directly: hermes gateway run")
sys.exit(0)
else: else:
print("Not supported on this platform.") print("Not supported on this platform.")
sys.exit(1) sys.exit(1)

View file

@ -2232,6 +2232,7 @@ def setup_gateway(config: dict):
from hermes_cli.gateway import ( from hermes_cli.gateway import (
_is_service_installed, _is_service_installed,
_is_service_running, _is_service_running,
supports_systemd_services,
has_conflicting_systemd_units, has_conflicting_systemd_units,
install_linux_gateway_from_setup, install_linux_gateway_from_setup,
print_systemd_scope_conflict_warning, print_systemd_scope_conflict_warning,
@ -2244,16 +2245,18 @@ def setup_gateway(config: dict):
service_installed = _is_service_installed() service_installed = _is_service_installed()
service_running = _is_service_running() service_running = _is_service_running()
supports_systemd = supports_systemd_services()
supports_service_manager = supports_systemd or _is_macos
print() print()
if _is_linux and has_conflicting_systemd_units(): if supports_systemd and has_conflicting_systemd_units():
print_systemd_scope_conflict_warning() print_systemd_scope_conflict_warning()
print() print()
if service_running: if service_running:
if prompt_yes_no(" Restart the gateway to pick up changes?", True): if prompt_yes_no(" Restart the gateway to pick up changes?", True):
try: try:
if _is_linux: if supports_systemd:
systemd_restart() systemd_restart()
elif _is_macos: elif _is_macos:
launchd_restart() launchd_restart()
@ -2262,14 +2265,14 @@ def setup_gateway(config: dict):
elif service_installed: elif service_installed:
if prompt_yes_no(" Start the gateway service?", True): if prompt_yes_no(" Start the gateway service?", True):
try: try:
if _is_linux: if supports_systemd:
systemd_start() systemd_start()
elif _is_macos: elif _is_macos:
launchd_start() launchd_start()
except Exception as e: except Exception as e:
print_error(f" Start failed: {e}") print_error(f" Start failed: {e}")
elif _is_linux or _is_macos: elif supports_service_manager:
svc_name = "systemd" if _is_linux else "launchd" svc_name = "systemd" if supports_systemd else "launchd"
if prompt_yes_no( if prompt_yes_no(
f" Install the gateway as a {svc_name} service? (runs in background, starts on boot)", f" Install the gateway as a {svc_name} service? (runs in background, starts on boot)",
True, True,
@ -2277,7 +2280,7 @@ def setup_gateway(config: dict):
try: try:
installed_scope = None installed_scope = None
did_install = False did_install = False
if _is_linux: if supports_systemd:
installed_scope, did_install = install_linux_gateway_from_setup(force=False) installed_scope, did_install = install_linux_gateway_from_setup(force=False)
else: else:
launchd_install(force=False) launchd_install(force=False)
@ -2285,7 +2288,7 @@ def setup_gateway(config: dict):
print() print()
if did_install and prompt_yes_no(" Start the service now?", True): if did_install and prompt_yes_no(" Start the service now?", True):
try: try:
if _is_linux: if supports_systemd:
systemd_start(system=installed_scope == "system") systemd_start(system=installed_scope == "system")
elif _is_macos: elif _is_macos:
launchd_start() launchd_start()
@ -2296,9 +2299,18 @@ def setup_gateway(config: dict):
print_info(" You can try manually: hermes gateway install") print_info(" You can try manually: hermes gateway install")
else: else:
print_info(" You can install later: hermes gateway install") print_info(" You can install later: hermes gateway install")
if _is_linux: if supports_systemd:
print_info(" Or as a boot-time service: sudo hermes gateway install --system") print_info(" Or as a boot-time service: sudo hermes gateway install --system")
print_info(" Or run in foreground: hermes gateway") print_info(" Or run in foreground: hermes gateway")
else:
from hermes_constants import is_container
if is_container():
print_info("Start the gateway to bring your bots online:")
print_info(" hermes gateway run # Run as container main process")
print_info("")
print_info("For automatic restarts, use a Docker restart policy:")
print_info(" docker run --restart unless-stopped ...")
print_info(" docker restart <container> # Manual restart")
else: else:
print_info("Start the gateway to bring your bots online:") print_info("Start the gateway to bring your bots online:")
print_info(" hermes gateway # Run in foreground") print_info(" hermes gateway # Run in foreground")

View file

@ -346,6 +346,18 @@ def show_status(args):
print(" Note: Android may stop background jobs when Termux is suspended") print(" Note: Android may stop background jobs when Termux is suspended")
elif sys.platform.startswith('linux'): elif sys.platform.startswith('linux'):
from hermes_constants import is_container
if is_container():
# Docker/Podman: no systemd — check for running gateway processes
try:
from hermes_cli.gateway import find_gateway_pids
gateway_pids = find_gateway_pids()
is_active = len(gateway_pids) > 0
except Exception:
is_active = False
print(f" Status: {check_mark(is_active)} {'running' if is_active else 'stopped'}")
print(" Manager: docker (foreground)")
else:
try: try:
from hermes_cli.gateway import get_service_name from hermes_cli.gateway import get_service_name
_gw_svc = get_service_name() _gw_svc = get_service_name()

View file

@ -189,6 +189,37 @@ def is_wsl() -> bool:
return _wsl_detected return _wsl_detected
_container_detected: bool | None = None
def is_container() -> bool:
"""Return True when running inside a Docker/Podman container.
Checks ``/.dockerenv`` (Docker), ``/run/.containerenv`` (Podman),
and ``/proc/1/cgroup`` for container runtime markers. Result is
cached for the process lifetime. Import-safe no heavy deps.
"""
global _container_detected
if _container_detected is not None:
return _container_detected
if os.path.exists("/.dockerenv"):
_container_detected = True
return True
if os.path.exists("/run/.containerenv"):
_container_detected = True
return True
try:
with open("/proc/1/cgroup", "r") as f:
cgroup = f.read()
if "docker" in cgroup or "podman" in cgroup or "/lxc/" in cgroup:
_container_detected = True
return True
except OSError:
pass
_container_detected = False
return False
# ─── Well-Known Paths ───────────────────────────────────────────────────────── # ─── Well-Known Paths ─────────────────────────────────────────────────────────

View file

@ -12,49 +12,10 @@ from unittest.mock import MagicMock, patch
import pytest import pytest
from hermes_cli.config import ( from hermes_cli.config import (
_is_inside_container,
get_container_exec_info, get_container_exec_info,
) )
# =============================================================================
# _is_inside_container
# =============================================================================
def test_is_inside_container_dockerenv():
"""Detects /.dockerenv marker file."""
with patch("os.path.exists") as mock_exists:
mock_exists.side_effect = lambda p: p == "/.dockerenv"
assert _is_inside_container() is True
def test_is_inside_container_containerenv():
"""Detects Podman's /run/.containerenv marker."""
with patch("os.path.exists") as mock_exists:
mock_exists.side_effect = lambda p: p == "/run/.containerenv"
assert _is_inside_container() is True
def test_is_inside_container_cgroup_docker():
"""Detects 'docker' in /proc/1/cgroup."""
with patch("os.path.exists", return_value=False), \
patch("builtins.open", create=True) as mock_open:
mock_open.return_value.__enter__ = lambda s: s
mock_open.return_value.__exit__ = MagicMock(return_value=False)
mock_open.return_value.read = MagicMock(
return_value="12:memory:/docker/abc123\n"
)
assert _is_inside_container() is True
def test_is_inside_container_false_on_host():
"""Returns False when none of the container indicators are present."""
with patch("os.path.exists", return_value=False), \
patch("builtins.open", side_effect=OSError("no such file")):
assert _is_inside_container() is False
# ============================================================================= # =============================================================================
# get_container_exec_info # get_container_exec_info
# ============================================================================= # =============================================================================
@ -81,7 +42,7 @@ def container_env(tmp_path, monkeypatch):
def test_get_container_exec_info_returns_metadata(container_env): def test_get_container_exec_info_returns_metadata(container_env):
"""Reads .container-mode and returns all fields including exec_user.""" """Reads .container-mode and returns all fields including exec_user."""
with patch("hermes_cli.config._is_inside_container", return_value=False): with patch("hermes_constants.is_container", return_value=False):
info = get_container_exec_info() info = get_container_exec_info()
assert info is not None assert info is not None
@ -93,7 +54,7 @@ def test_get_container_exec_info_returns_metadata(container_env):
def test_get_container_exec_info_none_inside_container(container_env): def test_get_container_exec_info_none_inside_container(container_env):
"""Returns None when we're already inside a container.""" """Returns None when we're already inside a container."""
with patch("hermes_cli.config._is_inside_container", return_value=True): with patch("hermes_constants.is_container", return_value=True):
info = get_container_exec_info() info = get_container_exec_info()
assert info is None assert info is None
@ -106,7 +67,7 @@ def test_get_container_exec_info_none_without_file(tmp_path, monkeypatch):
monkeypatch.setenv("HERMES_HOME", str(hermes_home)) monkeypatch.setenv("HERMES_HOME", str(hermes_home))
monkeypatch.delenv("HERMES_DEV", raising=False) monkeypatch.delenv("HERMES_DEV", raising=False)
with patch("hermes_cli.config._is_inside_container", return_value=False): with patch("hermes_constants.is_container", return_value=False):
info = get_container_exec_info() info = get_container_exec_info()
assert info is None assert info is None
@ -116,7 +77,7 @@ def test_get_container_exec_info_skipped_when_hermes_dev(container_env, monkeypa
"""Returns None when HERMES_DEV=1 is set (dev mode bypass).""" """Returns None when HERMES_DEV=1 is set (dev mode bypass)."""
monkeypatch.setenv("HERMES_DEV", "1") monkeypatch.setenv("HERMES_DEV", "1")
with patch("hermes_cli.config._is_inside_container", return_value=False): with patch("hermes_constants.is_container", return_value=False):
info = get_container_exec_info() info = get_container_exec_info()
assert info is None assert info is None
@ -126,7 +87,7 @@ def test_get_container_exec_info_not_skipped_when_hermes_dev_zero(container_env,
"""HERMES_DEV=0 does NOT trigger bypass — only '1' does.""" """HERMES_DEV=0 does NOT trigger bypass — only '1' does."""
monkeypatch.setenv("HERMES_DEV", "0") monkeypatch.setenv("HERMES_DEV", "0")
with patch("hermes_cli.config._is_inside_container", return_value=False): with patch("hermes_constants.is_container", return_value=False):
info = get_container_exec_info() info = get_container_exec_info()
assert info is not None assert info is not None
@ -143,7 +104,7 @@ def test_get_container_exec_info_defaults():
"# minimal file with no keys\n" "# minimal file with no keys\n"
) )
with patch("hermes_cli.config._is_inside_container", return_value=False), \ with patch("hermes_constants.is_container", return_value=False), \
patch("hermes_cli.config.get_hermes_home", return_value=hermes_home), \ patch("hermes_cli.config.get_hermes_home", return_value=hermes_home), \
patch.dict(os.environ, {}, clear=False): patch.dict(os.environ, {}, clear=False):
os.environ.pop("HERMES_DEV", None) os.environ.pop("HERMES_DEV", None)
@ -165,7 +126,7 @@ def test_get_container_exec_info_docker_backend(container_env):
"hermes_bin=/opt/hermes/bin/hermes\n" "hermes_bin=/opt/hermes/bin/hermes\n"
) )
with patch("hermes_cli.config._is_inside_container", return_value=False): with patch("hermes_constants.is_container", return_value=False):
info = get_container_exec_info() info = get_container_exec_info()
assert info["backend"] == "docker" assert info["backend"] == "docker"
@ -176,7 +137,7 @@ def test_get_container_exec_info_docker_backend(container_env):
def test_get_container_exec_info_crashes_on_permission_error(container_env): def test_get_container_exec_info_crashes_on_permission_error(container_env):
"""PermissionError propagates instead of being silently swallowed.""" """PermissionError propagates instead of being silently swallowed."""
with patch("hermes_cli.config._is_inside_container", return_value=False), \ with patch("hermes_constants.is_container", return_value=False), \
patch("builtins.open", side_effect=PermissionError("permission denied")): patch("builtins.open", side_effect=PermissionError("permission denied")):
with pytest.raises(PermissionError): with pytest.raises(PermissionError):
get_container_exec_info() get_container_exec_info()

View file

@ -394,6 +394,21 @@ class TestLaunchdServiceRecovery:
class TestGatewayServiceDetection: class TestGatewayServiceDetection:
def test_supports_systemd_services_requires_systemctl_binary(self, monkeypatch):
monkeypatch.setattr(gateway_cli, "is_linux", lambda: True)
monkeypatch.setattr(gateway_cli, "is_termux", lambda: False)
monkeypatch.setattr(gateway_cli.shutil, "which", lambda name: None)
assert gateway_cli.supports_systemd_services() is False
def test_supports_systemd_services_returns_true_when_systemctl_present(self, monkeypatch):
monkeypatch.setattr(gateway_cli, "is_linux", lambda: True)
monkeypatch.setattr(gateway_cli, "is_termux", lambda: False)
monkeypatch.setattr(gateway_cli, "is_wsl", lambda: False)
monkeypatch.setattr(gateway_cli.shutil, "which", lambda name: "/usr/bin/systemctl")
assert gateway_cli.supports_systemd_services() is True
def test_is_service_running_checks_system_scope_when_user_scope_is_inactive(self, monkeypatch): def test_is_service_running_checks_system_scope_when_user_scope_is_inactive(self, monkeypatch):
user_unit = SimpleNamespace(exists=lambda: True) user_unit = SimpleNamespace(exists=lambda: True)
system_unit = SimpleNamespace(exists=lambda: True) system_unit = SimpleNamespace(exists=lambda: True)
@ -418,6 +433,23 @@ class TestGatewayServiceDetection:
assert gateway_cli._is_service_running() is True assert gateway_cli._is_service_running() is True
def test_is_service_running_returns_false_when_systemctl_missing(self, monkeypatch):
unit = SimpleNamespace(exists=lambda: True)
monkeypatch.setattr(gateway_cli, "supports_systemd_services", lambda: True)
monkeypatch.setattr(
gateway_cli,
"get_systemd_unit_path",
lambda system=False: unit,
)
def fake_run(*args, **kwargs):
raise FileNotFoundError("systemctl")
monkeypatch.setattr(gateway_cli.subprocess, "run", fake_run)
assert gateway_cli._is_service_running() is False
class TestGatewaySystemServiceRouting: class TestGatewaySystemServiceRouting:
def test_systemd_restart_self_requests_graceful_restart_without_reload_or_restart(self, monkeypatch, capsys): def test_systemd_restart_self_requests_graceful_restart_without_reload_or_restart(self, monkeypatch, capsys):
@ -1001,3 +1033,91 @@ class TestSystemUnitPathRemapping:
# Target user paths should be present # Target user paths should be present
assert "/home/alice" in unit assert "/home/alice" in unit
assert "WorkingDirectory=/home/alice/.hermes/hermes-agent" in unit assert "WorkingDirectory=/home/alice/.hermes/hermes-agent" in unit
class TestDockerAwareGateway:
"""Tests for Docker container awareness in gateway commands."""
def test_run_systemctl_raises_runtimeerror_when_missing(self, monkeypatch):
"""_run_systemctl raises RuntimeError with container guidance when systemctl is absent."""
import pytest
def fake_run(cmd, **kwargs):
raise FileNotFoundError("systemctl")
monkeypatch.setattr(gateway_cli.subprocess, "run", fake_run)
with pytest.raises(RuntimeError, match="systemctl is not available"):
gateway_cli._run_systemctl(["start", "hermes-gateway"])
def test_run_systemctl_passes_through_on_success(self, monkeypatch):
"""_run_systemctl delegates to subprocess.run when systemctl exists."""
calls = []
def fake_run(cmd, **kwargs):
calls.append(cmd)
return SimpleNamespace(returncode=0, stdout="", stderr="")
monkeypatch.setattr(gateway_cli.subprocess, "run", fake_run)
result = gateway_cli._run_systemctl(["status", "hermes-gateway"])
assert result.returncode == 0
assert len(calls) == 1
assert "status" in calls[0]
def test_install_in_container_prints_docker_guidance(self, monkeypatch, capsys):
"""'hermes gateway install' inside Docker exits 0 with container guidance."""
import pytest
monkeypatch.setattr(gateway_cli, "is_managed", lambda: False)
monkeypatch.setattr(gateway_cli, "is_termux", lambda: False)
monkeypatch.setattr(gateway_cli, "supports_systemd_services", lambda: False)
monkeypatch.setattr(gateway_cli, "is_macos", lambda: False)
monkeypatch.setattr(gateway_cli, "is_wsl", lambda: False)
monkeypatch.setattr(gateway_cli, "is_container", lambda: True)
args = SimpleNamespace(gateway_command="install", force=False, system=False, run_as_user=None)
with pytest.raises(SystemExit) as exc_info:
gateway_cli.gateway_command(args)
assert exc_info.value.code == 0
out = capsys.readouterr().out
assert "Docker" in out or "docker" in out
assert "restart" in out.lower()
def test_uninstall_in_container_prints_docker_guidance(self, monkeypatch, capsys):
"""'hermes gateway uninstall' inside Docker exits 0 with container guidance."""
import pytest
monkeypatch.setattr(gateway_cli, "is_managed", lambda: False)
monkeypatch.setattr(gateway_cli, "is_termux", lambda: False)
monkeypatch.setattr(gateway_cli, "supports_systemd_services", lambda: False)
monkeypatch.setattr(gateway_cli, "is_macos", lambda: False)
monkeypatch.setattr(gateway_cli, "is_container", lambda: True)
args = SimpleNamespace(gateway_command="uninstall", system=False)
with pytest.raises(SystemExit) as exc_info:
gateway_cli.gateway_command(args)
assert exc_info.value.code == 0
out = capsys.readouterr().out
assert "docker" in out.lower()
def test_start_in_container_prints_docker_guidance(self, monkeypatch, capsys):
"""'hermes gateway start' inside Docker exits 0 with container guidance."""
import pytest
monkeypatch.setattr(gateway_cli, "is_termux", lambda: False)
monkeypatch.setattr(gateway_cli, "supports_systemd_services", lambda: False)
monkeypatch.setattr(gateway_cli, "is_macos", lambda: False)
monkeypatch.setattr(gateway_cli, "is_wsl", lambda: False)
monkeypatch.setattr(gateway_cli, "is_container", lambda: True)
args = SimpleNamespace(gateway_command="start", system=False)
with pytest.raises(SystemExit) as exc_info:
gateway_cli.gateway_command(args)
assert exc_info.value.code == 0
out = capsys.readouterr().out
assert "docker" in out.lower()
assert "hermes gateway run" in out

View file

@ -1,5 +1,4 @@
"""Tests for setup_model_provider — verifies the delegation to """Tests for setup.py configuration flows."""
select_provider_and_model() and config dict sync."""
import json import json
import sys import sys
import types import types
@ -8,6 +7,7 @@ import pytest
from hermes_cli.auth import get_active_provider from hermes_cli.auth import get_active_provider
from hermes_cli.config import load_config, save_config from hermes_cli.config import load_config, save_config
from hermes_cli import setup as setup_mod
from hermes_cli.setup import setup_model_provider from hermes_cli.setup import setup_model_provider
@ -144,6 +144,85 @@ def test_setup_custom_providers_synced(tmp_path, monkeypatch):
assert reloaded.get("custom_providers") == [{"name": "Local", "base_url": "http://localhost:8080/v1"}] assert reloaded.get("custom_providers") == [{"name": "Local", "base_url": "http://localhost:8080/v1"}]
def test_setup_gateway_skips_service_install_when_systemctl_missing(monkeypatch, capsys):
env = {
"TELEGRAM_BOT_TOKEN": "",
"TELEGRAM_HOME_CHANNEL": "",
"DISCORD_BOT_TOKEN": "",
"DISCORD_HOME_CHANNEL": "",
"SLACK_BOT_TOKEN": "",
"SLACK_HOME_CHANNEL": "",
"MATRIX_HOMESERVER": "https://matrix.example.com",
"MATRIX_USER_ID": "@alice:example.com",
"MATRIX_PASSWORD": "",
"MATRIX_ACCESS_TOKEN": "token",
"BLUEBUBBLES_SERVER_URL": "",
"BLUEBUBBLES_HOME_CHANNEL": "",
"WHATSAPP_ENABLED": "",
"WEBHOOK_ENABLED": "",
}
monkeypatch.setattr(setup_mod, "get_env_value", lambda key: env.get(key, ""))
monkeypatch.setattr(setup_mod, "prompt_yes_no", lambda *args, **kwargs: False)
monkeypatch.setattr("platform.system", lambda: "Linux")
import hermes_cli.gateway as gateway_mod
monkeypatch.setattr(gateway_mod, "supports_systemd_services", lambda: False)
monkeypatch.setattr(gateway_mod, "is_macos", lambda: False)
monkeypatch.setattr(gateway_mod, "_is_service_installed", lambda: False)
monkeypatch.setattr(gateway_mod, "_is_service_running", lambda: False)
setup_mod.setup_gateway({})
out = capsys.readouterr().out
assert "Messaging platforms configured!" in out
assert "Start the gateway to bring your bots online:" in out
assert "hermes gateway" in out
def test_setup_gateway_in_container_shows_docker_guidance(monkeypatch, capsys):
"""setup_gateway() in a Docker container shows Docker-specific restart instructions."""
env = {
"TELEGRAM_BOT_TOKEN": "",
"TELEGRAM_HOME_CHANNEL": "",
"DISCORD_BOT_TOKEN": "",
"DISCORD_HOME_CHANNEL": "",
"SLACK_BOT_TOKEN": "",
"SLACK_HOME_CHANNEL": "",
"MATRIX_HOMESERVER": "https://matrix.example.com",
"MATRIX_USER_ID": "@alice:example.com",
"MATRIX_PASSWORD": "",
"MATRIX_ACCESS_TOKEN": "token",
"BLUEBUBBLES_SERVER_URL": "",
"BLUEBUBBLES_HOME_CHANNEL": "",
"WHATSAPP_ENABLED": "",
"WEBHOOK_ENABLED": "",
}
monkeypatch.setattr(setup_mod, "get_env_value", lambda key: env.get(key, ""))
monkeypatch.setattr(setup_mod, "prompt_yes_no", lambda *args, **kwargs: False)
monkeypatch.setattr("platform.system", lambda: "Linux")
import hermes_cli.gateway as gateway_mod
monkeypatch.setattr(gateway_mod, "supports_systemd_services", lambda: False)
monkeypatch.setattr(gateway_mod, "is_macos", lambda: False)
monkeypatch.setattr(gateway_mod, "_is_service_installed", lambda: False)
monkeypatch.setattr(gateway_mod, "_is_service_running", lambda: False)
# Patch is_container at the import location in setup.py
import hermes_constants
monkeypatch.setattr(hermes_constants, "is_container", lambda: True)
setup_mod.setup_gateway({})
out = capsys.readouterr().out
assert "Messaging platforms configured!" in out
assert "docker" in out.lower() or "Docker" in out
assert "restart" in out.lower()
def test_setup_syncs_custom_provider_removal_from_disk(tmp_path, monkeypatch): def test_setup_syncs_custom_provider_removal_from_disk(tmp_path, monkeypatch):
"""Removing the last custom provider in model setup should persist.""" """Removing the last custom provider in model setup should persist."""
monkeypatch.setenv("HERMES_HOME", str(tmp_path)) monkeypatch.setenv("HERMES_HOME", str(tmp_path))

View file

@ -6,7 +6,8 @@ from unittest.mock import patch
import pytest import pytest
from hermes_constants import get_default_hermes_root import hermes_constants
from hermes_constants import get_default_hermes_root, is_container
class TestGetDefaultHermesRoot: class TestGetDefaultHermesRoot:
@ -60,3 +61,53 @@ class TestGetDefaultHermesRoot:
monkeypatch.setattr(Path, "home", lambda: tmp_path) monkeypatch.setattr(Path, "home", lambda: tmp_path)
monkeypatch.setenv("HERMES_HOME", str(profile)) monkeypatch.setenv("HERMES_HOME", str(profile))
assert get_default_hermes_root() == docker_root assert get_default_hermes_root() == docker_root
class TestIsContainer:
"""Tests for is_container() — Docker/Podman detection."""
def _reset_cache(self, monkeypatch):
"""Reset the cached detection result before each test."""
monkeypatch.setattr(hermes_constants, "_container_detected", None)
def test_detects_dockerenv(self, monkeypatch, tmp_path):
"""/.dockerenv triggers container detection."""
self._reset_cache(monkeypatch)
monkeypatch.setattr(os.path, "exists", lambda p: p == "/.dockerenv")
assert is_container() is True
def test_detects_containerenv(self, monkeypatch, tmp_path):
"""/run/.containerenv triggers container detection (Podman)."""
self._reset_cache(monkeypatch)
monkeypatch.setattr(os.path, "exists", lambda p: p == "/run/.containerenv")
assert is_container() is True
def test_detects_cgroup_docker(self, monkeypatch, tmp_path):
"""/proc/1/cgroup containing 'docker' triggers detection."""
import builtins
self._reset_cache(monkeypatch)
monkeypatch.setattr(os.path, "exists", lambda p: False)
cgroup_file = tmp_path / "cgroup"
cgroup_file.write_text("12:memory:/docker/abc123\n")
_real_open = builtins.open
monkeypatch.setattr("builtins.open", lambda p, *a, **kw: _real_open(str(cgroup_file), *a, **kw) if p == "/proc/1/cgroup" else _real_open(p, *a, **kw))
assert is_container() is True
def test_negative_case(self, monkeypatch, tmp_path):
"""Returns False on a regular Linux host."""
import builtins
self._reset_cache(monkeypatch)
monkeypatch.setattr(os.path, "exists", lambda p: False)
cgroup_file = tmp_path / "cgroup"
cgroup_file.write_text("12:memory:/\n")
_real_open = builtins.open
monkeypatch.setattr("builtins.open", lambda p, *a, **kw: _real_open(str(cgroup_file), *a, **kw) if p == "/proc/1/cgroup" else _real_open(p, *a, **kw))
assert is_container() is False
def test_caches_result(self, monkeypatch):
"""Second call uses cached value without re-probing."""
monkeypatch.setattr(hermes_constants, "_container_detected", True)
assert is_container() is True
# Even if we make os.path.exists return False, cached value wins
monkeypatch.setattr(os.path, "exists", lambda p: False)
assert is_container() is True

View file

@ -106,8 +106,9 @@ def detect_audio_environment() -> dict:
if any(os.environ.get(v) for v in ('SSH_CLIENT', 'SSH_TTY', 'SSH_CONNECTION')): if any(os.environ.get(v) for v in ('SSH_CLIENT', 'SSH_TTY', 'SSH_CONNECTION')):
warnings.append("Running over SSH -- no audio devices available") warnings.append("Running over SSH -- no audio devices available")
# Docker detection # Docker/Podman container detection
if os.path.exists('/.dockerenv'): from hermes_constants import is_container
if is_container():
warnings.append("Running inside Docker container -- no audio devices") warnings.append("Running inside Docker container -- no audio devices")
# WSL detection — PulseAudio bridge makes audio work in WSL. # WSL detection — PulseAudio bridge makes audio work in WSL.