Docker/container gateway hardening and systemctl error

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

• #8614 fix(gateway): harden Docker/container gateway pathway

  merged to main

  files: hermes_cli/config.py, hermes_cli/dump.py, hermes_cli/gateway.py, hermes_cli/setup.py, hermes_cli/status.py, hermes_constants.py, tests/hermes_cli/test_container_aware_cli.py, tests/hermes_cli/test_gateway_service.py, tests/hermes_cli/test_setup.py, tests/test_hermes_constants.py, tools/voice_mode.py

  Show the diff

  • hermes_cli/config.py

    @@ -148,25 +148,6 @@ def managed_error(action: str = "modify configuration"):
     # 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]:
         """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":
             return None
     
    -    if _is_inside_container():
    +    from hermes_constants import is_container
    +    if is_container():
             return None
     
         container_mode_file = get_hermes_home() / ".container-mode"

  • hermes_cli/dump.py

    @@ -44,6 +44,16 @@ def _redact(value: str) -> str:
     def _gateway_status() -> str:
         """Return a short gateway status string."""
         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:
                 from hermes_cli.gateway import get_service_name
                 svc = get_service_name()

  • hermes_cli/gateway.py

    @@ -331,7 +331,7 @@ def is_linux() -> bool:
         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:
    @@ -353,7 +353,9 @@ def _wsl_systemd_operational() -> 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
         if is_wsl():
             return _wsl_systemd_operational()
    @@ -483,6 +485,21 @@ def _journalctl_cmd(system: bool = False) -> list[str]:
         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:
         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
         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")
         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):
                 print(f"↻ Repairing outdated {_service_scope_label(system)} systemd service at: {unit_path}")
                 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")
                 return
             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}")
         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)
    -    subprocess.run(_systemctl_cmd(system) + ["enable", get_service_name()], check=True, timeout=30)
    +    _run_systemctl(["daemon-reload"], system=system, check=True, timeout=30)
    +    _run_systemctl(["enable", get_service_name()], system=system, check=True, timeout=30)
     
         print()
         print(f"✓ {_service_scope_label(system).capitalize()} service installed and enabled!")
    @@ -1063,15 +1080,15 @@ def systemd_uninstall(system: bool = False):
         if system:
             _require_root_for_system_service("uninstall")
     
    -    subprocess.run(_systemctl_cmd(system) + ["stop", get_service_name()], check=False, timeout=90)
    -    subprocess.run(_systemctl_cmd(system) + ["disable", get_service_name()], check=False, timeout=30)
    +    _run_systemctl(["stop", get_service_name()], system=system, check=False, timeout=90)
    +    _run_systemctl(["disable", get_service_name()], system=system, check=False, timeout=30)
     
         unit_path = get_systemd_unit_path(system=system)
         if unit_path.exists():
             unit_path.unlink()
             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")
     
     
    @@ -1080,7 +1097,7 @@ def systemd_start(system: bool = False):
         if system:
             _require_root_for_system_service("start")
         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")
     
     
    @@ -1089,7 +1106,7 @@ def systemd_stop(system: bool = False):
         system = _select_systemd_scope(system)
         if system:
             _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")
     
     
    @@ -1105,7 +1122,7 @@ def systemd_restart(system: bool = False):
         if pid is not None and _request_gateway_self_restart(pid):
             print(f"✓ {_service_scope_label(system).capitalize()} service restart requested")
             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")
     
     
    @@ -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()
     
    -    subprocess.run(
    -        _systemctl_cmd(system) + ["status", get_service_name(), "--no-pager"],
    +    _run_systemctl(
    +        ["status", get_service_name(), "--no-pager"],
    +        system=system,
             capture_output=False,
             timeout=10,
         )
     
    -    result = subprocess.run(
    -        _systemctl_cmd(system) + ["is-active", get_service_name()],
    +    result = _run_systemctl(
    +        ["is-active", get_service_name()],
    +        system=system,
             capture_output=True,
             text=True,
             timeout=10,
    @@ -2123,24 +2142,24 @@ def _is_service_running() -> bool:
     
             if user_unit_exists:
                 try:
    -                result = subprocess.run(
    -                    _systemctl_cmd(False) + ["is-active", get_service_name()],
    -                    capture_output=True, text=True, timeout=10,
    +                result = _run_systemctl(
    +                    ["is-active", get_service_name()],
    +                    system=False, capture_output=True, text=True, timeout=10,
                     )
                     if result.stdout.strip() == "active":
                         return True
    -            except subprocess.TimeoutExpired:
    +            except (RuntimeError, subprocess.TimeoutExpired):
                     pass
     
             if system_unit_exists:
                 try:
    -                result = subprocess.run(
    -                    _systemctl_cmd(True) + ["is-active", get_service_name()],
    -                    capture_output=True, text=True, timeout=10,
    +                result = _run_systemctl(
    +                    ["is-active", get_service_name()],
    +                    system=True, capture_output=True, text=True, timeout=10,
                     )
                     if result.stdout.strip() == "active":
                         return True
    -            except subprocess.TimeoutExpired:
    +            except (RuntimeError, subprocess.TimeoutExpired):
                     pass
     
             return False
    @@ -2774,6 +2793,15 @@ def gateway_command(args):
                 print("  tmux new -s hermes 'hermes gateway run'         # persistent via tmux")
                 print("  nohup hermes gateway run > ~/.hermes/logs/gateway.log 2>&1 &  # background")
                 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:
                 print("Service installation not supported on this platform.")
                 print("Run manually: hermes gateway run")
    @@ -2792,10 +2820,17 @@ def gateway_command(args):
                 systemd_uninstall(system=system)
             elif is_macos():
                 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:
                 print("Not supported on this platform.")
                 sys.exit(1)
    -    
    +
         elif subcmd == "start":
             system = getattr(args, 'system', False)
             if is_termux():
    @@ -2816,10 +2851,19 @@ def gateway_command(args):
                 print()
                 print("To enable systemd: add systemd=true to /etc/wsl.conf and run 'wsl --shutdown' from PowerShell.")
                 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:
                 print("Not supported on this platform.")
                 sys.exit(1)
    -    
    +
         elif subcmd == "stop":
             stop_all = getattr(args, 'all', False)
             system = getattr(args, 'system', False)

  • hermes_cli/setup.py

    @@ -2232,6 +2232,7 @@ def setup_gateway(config: dict):
             from hermes_cli.gateway import (
                 _is_service_installed,
                 _is_service_running,
    +            supports_systemd_services,
                 has_conflicting_systemd_units,
                 install_linux_gateway_from_setup,
                 print_systemd_scope_conflict_warning,
    @@ -2244,16 +2245,18 @@ def setup_gateway(config: dict):
     
             service_installed = _is_service_installed()
             service_running = _is_service_running()
    +        supports_systemd = supports_systemd_services()
    +        supports_service_manager = supports_systemd or _is_macos
     
             print()
    -        if _is_linux and has_conflicting_systemd_units():
    +        if supports_systemd and has_conflicting_systemd_units():
                 print_systemd_scope_conflict_warning()
                 print()
     
             if service_running:
                 if prompt_yes_no("  Restart the gateway to pick up changes?", True):
                     try:
    -                    if _is_linux:
    +                    if supports_systemd:
                             systemd_restart()
                         elif _is_macos:
                             launchd_restart()
    @@ -2262,14 +2265,14 @@ def setup_gateway(config: dict):
             elif service_installed:
                 if prompt_yes_no("  Start the gateway service?", True):
                     try:
    -                    if _is_linux:
    +                    if supports_systemd:
                             systemd_start()
                         elif _is_macos:
                             launchd_start()
                     except Exception as e:
                         print_error(f"  Start failed: {e}")
    -        elif _is_linux or _is_macos:
    -            svc_name = "systemd" if _is_linux else "launchd"
    +        elif supports_service_manager:
    +            svc_name = "systemd" if supports_systemd else "launchd"
                 if prompt_yes_no(
                     f"  Install the gateway as a {svc_name} service? (runs in background, starts on boot)",
                     True,
    @@ -2277,7 +2280,7 @@ def setup_gateway(config: dict):
                     try:
                         installed_scope = None
                         did_install = False
    -                    if _is_linux:
    +                    if supports_systemd:
                             installed_scope, did_install = install_linux_gateway_from_setup(force=False)
                         else:
                             launchd_install(force=False)
    @@ -2285,7 +2288,7 @@ def setup_gateway(config: dict):
                         print()
                         if did_install and prompt_yes_no("  Start the service now?", True):
                             try:
    -                            if _is_linux:
    +                            if supports_systemd:
                                     systemd_start(system=installed_scope == "system")
                                 elif _is_macos:
                                     launchd_start()
    @@ -2296,12 +2299,21 @@ def setup_gateway(config: dict):
                         print_info("  You can try manually: hermes gateway install")
                 else:
                     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 run in foreground:  hermes gateway")
             else:
    -            print_info("Start the gateway to bring your bots online:")
    -            print_info("   hermes gateway              # Run in foreground")
    +            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:
    +                print_info("Start the gateway to bring your bots online:")
    +                print_info("   hermes gateway              # Run in foreground")
     
             print_info("━" * 50)
     

  • hermes_cli/status.py

    @@ -346,23 +346,35 @@ def show_status(args):
                 print("  Note:         Android may stop background jobs when Termux is suspended")
     
         elif sys.platform.startswith('linux'):
    -        try:
    -            from hermes_cli.gateway import get_service_name
    -            _gw_svc = get_service_name()
    -        except Exception:
    -            _gw_svc = "hermes-gateway"
    -        try:
    -            result = subprocess.run(
    -                ["systemctl", "--user", "is-active", _gw_svc],
    -                capture_output=True,
    -                text=True,
    -                timeout=5
    -            )
    -            is_active = result.stdout.strip() == "active"
    -        except (FileNotFoundError, subprocess.TimeoutExpired):
    -            is_active = False
    -        print(f"  Status:       {check_mark(is_active)} {'running' if is_active else 'stopped'}")
    -        print("  Manager:      systemd (user)")
    +        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:
    +                from hermes_cli.gateway import get_service_name
    +                _gw_svc = get_service_name()
    +            except Exception:
    +                _gw_svc = "hermes-gateway"
    +            try:
    +                result = subprocess.run(
    +                    ["systemctl", "--user", "is-active", _gw_svc],
    +                    capture_output=True,
    +                    text=True,
    +                    timeout=5
    +                )
    +                is_active = result.stdout.strip() == "active"
    +            except (FileNotFoundError, subprocess.TimeoutExpired):
    +                is_active = False
    +            print(f"  Status:       {check_mark(is_active)} {'running' if is_active else 'stopped'}")
    +            print("  Manager:      systemd (user)")
             
         elif sys.platform == 'darwin':
             from hermes_cli.gateway import get_launchd_label

  • hermes_constants.py

    @@ -189,6 +189,37 @@ def is_wsl() -> bool:
         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 ─────────────────────────────────────────────────────────
     
     

  • tests/hermes_cli/test_container_aware_cli.py

    @@ -12,49 +12,10 @@
     import pytest
     
     from hermes_cli.config import (
    -    _is_inside_container,
         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
     # =============================================================================
    @@ -81,7 +42,7 @@ def container_env(tmp_path, monkeypatch):
     
     def test_get_container_exec_info_returns_metadata(container_env):
         """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()
     
         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):
         """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()
     
         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.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()
     
         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)."""
         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()
     
         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."""
         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()
     
         assert info is not None
    @@ -143,7 +104,7 @@ def test_get_container_exec_info_defaults():
                 "# 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.dict(os.environ, {}, clear=False):
                 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"
         )
     
    -    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()
     
         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):
         """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")):
             with pytest.raises(PermissionError):
                 get_container_exec_info()

  • tests/hermes_cli/test_gateway_service.py

    @@ -394,6 +394,21 @@ def test_launchd_status_reports_local_stale_plist_when_unloaded(self, tmp_path,
     
     
     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):
             user_unit = SimpleNamespace(exists=lambda: True)
             system_unit = SimpleNamespace(exists=lambda: True)
    @@ -418,6 +433,23 @@ def fake_run(cmd, capture_output=True, text=True, **kwargs):
     
             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:
         def test_systemd_restart_self_requests_graceful_restart_without_reload_or_restart(self, monkeypatch, capsys):
    @@ -1001,3 +1033,91 @@ def test_system_unit_has_no_root_paths(self, monkeypatch, tmp_path):
             # Target user paths should be present
             assert "/home/alice" 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

  • tests/hermes_cli/test_setup.py

    @@ -1,5 +1,4 @@
    -"""Tests for setup_model_provider — verifies the delegation to
    -select_provider_and_model() and config dict sync."""
    +"""Tests for setup.py configuration flows."""
     import json
     import sys
     import types
    @@ -8,6 +7,7 @@
     
     from hermes_cli.auth import get_active_provider
     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
     
     
    @@ -144,6 +144,85 @@ def fake_select():
         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):
         """Removing the last custom provider in model setup should persist."""
         monkeypatch.setenv("HERMES_HOME", str(tmp_path))

  • tests/test_hermes_constants.py

    @@ -6,7 +6,8 @@
     
     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:
    @@ -60,3 +61,53 @@ def test_docker_profile_active(self, tmp_path, monkeypatch):
             monkeypatch.setattr(Path, "home", lambda: tmp_path)
             monkeypatch.setenv("HERMES_HOME", str(profile))
             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

  • tools/voice_mode.py

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