From 00ba600fc00650896700e9ed2fddaf718be6eada Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Mon, 5 Oct 2026 00:18:44 -0700 Subject: [PATCH] refactor(cli): container exec routing moves out of main.py into main_container.py main.py may only shrink vs main (code-health FILE_LINES); this branch grew it by 9 lines. _probe_container and _exec_in_container (the NixOS container-mode re-exec, 96 lines) move verbatim to the topical sibling hermes_cli/main_container.py; the entry point is now public (exec_in_container) and main() late-imports it at its only call site. The test import and the managed-runtime resolution allowlist entry follow the move. No re-export shim. Behaviour is unchanged. --- hermes_cli/main.py | 100 +----------------- hermes_cli/main_container.py | 104 +++++++++++++++++++ tests/fixtures/resolution_allowlist.json | 4 +- tests/hermes_cli/test_container_aware_cli.py | 6 +- 4 files changed, 112 insertions(+), 102 deletions(-) create mode 100644 hermes_cli/main_container.py diff --git a/hermes_cli/main.py b/hermes_cli/main.py index 1d6b68f9f4..f100a2a215 100644 --- a/hermes_cli/main.py +++ b/hermes_cli/main.py @@ -1352,102 +1352,6 @@ def _resolve_last_session(source: str = "cli") -> Optional[str]: return None -def _probe_container(cmd: list, backend: str, via_sudo: bool = False): - """Run a container inspect probe, returning the CompletedProcess. - - Catches TimeoutExpired specifically for a human-readable message; - all other exceptions propagate naturally. - """ - try: - return subprocess.run(cmd, capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=15) - except subprocess.TimeoutExpired: - label = f"sudo {backend}" if via_sudo else backend - print( - f"Error: timed out waiting for {label} to respond.\n" - f"The {backend} daemon may be unresponsive or starting up.", - file=sys.stderr, - ) - sys.exit(1) - - -def _exec_in_container(container_info: dict, cli_args: list): - """Replace the current process with a command inside the managed container. - - Probes whether sudo is needed (rootful containers), then os.execvp - into the container. On success the Python process is replaced entirely - and the container's exit code becomes the process exit code (OS semantics). - On failure, OSError propagates naturally. - - Args: - container_info: dict with backend, container_name, exec_user, hermes_bin - cli_args: the original CLI arguments (everything after 'hermes') - """ - - backend = container_info["backend"] - container_name = container_info["container_name"] - exec_user = container_info["exec_user"] - hermes_bin = container_info["hermes_bin"] - - runtime = shutil.which(backend) - if not runtime: - print( - f"Error: {backend} not found on PATH. Cannot route to container.", - file=sys.stderr, - ) - sys.exit(1) - - # Rootful containers (NixOS systemd service) are invisible to unprivileged - # users — Podman uses per-user namespaces, Docker needs group access. - # Probe whether the runtime can see the container; if not, try via sudo. - inspect_cmd = [runtime, "inspect", "--format", "ok", container_name] - cmd_prefix = [runtime] - if _probe_container(inspect_cmd, backend).returncode != 0: - sudo_path = shutil.which("sudo") - if not sudo_path: - print( - f"Error: container '{container_name}' not found via {backend}.\n" - f"The container may be running under root. Try: sudo hermes {' '.join(cli_args)}", - file=sys.stderr, - ) - sys.exit(1) - cmd_prefix = [sudo_path, "-n", runtime] - if _probe_container(cmd_prefix[:2] + inspect_cmd, backend, via_sudo=True).returncode != 0: - print( - f"Error: container '{container_name}' not found via {backend}.\n" - f"\n" - f"The container is likely running as root. Your user cannot see it\n" - f"because {backend} uses per-user namespaces. Grant passwordless\n" - f"sudo for {backend} — the -n (non-interactive) flag is required\n" - f"because a password prompt would hang or break piped commands.\n" - f"\n" - f"On NixOS:\n" - f"\n" - f" security.sudo.extraRules = [{{\n" - f' users = [ "{os.getenv("USER", "your-user")}" ];\n' - f' commands = [{{ command = "{runtime}"; options = [ "NOPASSWD" ]; }}];\n' - f" }}];\n" - f"\n" - f"Or run: sudo hermes {' '.join(cli_args)}", - file=sys.stderr, - ) - sys.exit(1) - - env_flags = [] - for var in ("TERM", "COLORTERM", "LANG", "LC_ALL"): - val = os.environ.get(var) - if val: - env_flags.extend(["-e", f"{var}={val}"]) - - exec_cmd = ( - cmd_prefix - + ["exec", "-it" if sys.stdin.isatty() else "-i", "-u", exec_user] - + env_flags - + [container_name, hermes_bin] - + cli_args - ) - os.execvp(exec_cmd[0], exec_cmd) - - def _resolve_session_by_name_or_id(name_or_id: str) -> Optional[str]: """Resolve a session title or ID to a session ID (None if neither matches). @@ -3695,7 +3599,9 @@ def main(): container_info = get_container_exec_info() if container_info: - _exec_in_container(container_info, sys.argv[1:]) + from hermes_cli.main_container import exec_in_container + + exec_in_container(container_info, sys.argv[1:]) sys.exit(1) # unreachable: execvp replaces the process or raises args = _parse_cli_args(parser, subparsers, sys.argv[1:]) diff --git a/hermes_cli/main_container.py b/hermes_cli/main_container.py new file mode 100644 index 0000000000..805f481036 --- /dev/null +++ b/hermes_cli/main_container.py @@ -0,0 +1,104 @@ +"""NixOS container mode: route a ``hermes`` invocation into the managed container (``main()``).""" + +from __future__ import annotations + +import os +import shutil +import subprocess +import sys + + +def _probe_container(cmd: list, backend: str, via_sudo: bool = False): + """Run a container inspect probe, returning the CompletedProcess. + + Catches TimeoutExpired specifically for a human-readable message; + all other exceptions propagate naturally. + """ + try: + return subprocess.run(cmd, capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=15) + except subprocess.TimeoutExpired: + label = f"sudo {backend}" if via_sudo else backend + print( + f"Error: timed out waiting for {label} to respond.\n" + f"The {backend} daemon may be unresponsive or starting up.", + file=sys.stderr, + ) + sys.exit(1) + + +def exec_in_container(container_info: dict, cli_args: list): + """Replace the current process with a command inside the managed container. + + Probes whether sudo is needed (rootful containers), then os.execvp + into the container. On success the Python process is replaced entirely + and the container's exit code becomes the process exit code (OS semantics). + On failure, OSError propagates naturally. + + Args: + container_info: dict with backend, container_name, exec_user, hermes_bin + cli_args: the original CLI arguments (everything after 'hermes') + """ + + backend = container_info["backend"] + container_name = container_info["container_name"] + exec_user = container_info["exec_user"] + hermes_bin = container_info["hermes_bin"] + + runtime = shutil.which(backend) + if not runtime: + print( + f"Error: {backend} not found on PATH. Cannot route to container.", + file=sys.stderr, + ) + sys.exit(1) + + # Rootful containers (NixOS systemd service) are invisible to unprivileged + # users — Podman uses per-user namespaces, Docker needs group access. + # Probe whether the runtime can see the container; if not, try via sudo. + inspect_cmd = [runtime, "inspect", "--format", "ok", container_name] + cmd_prefix = [runtime] + if _probe_container(inspect_cmd, backend).returncode != 0: + sudo_path = shutil.which("sudo") + if not sudo_path: + print( + f"Error: container '{container_name}' not found via {backend}.\n" + f"The container may be running under root. Try: sudo hermes {' '.join(cli_args)}", + file=sys.stderr, + ) + sys.exit(1) + cmd_prefix = [sudo_path, "-n", runtime] + if _probe_container(cmd_prefix[:2] + inspect_cmd, backend, via_sudo=True).returncode != 0: + print( + f"Error: container '{container_name}' not found via {backend}.\n" + f"\n" + f"The container is likely running as root. Your user cannot see it\n" + f"because {backend} uses per-user namespaces. Grant passwordless\n" + f"sudo for {backend} — the -n (non-interactive) flag is required\n" + f"because a password prompt would hang or break piped commands.\n" + f"\n" + f"On NixOS:\n" + f"\n" + f" security.sudo.extraRules = [{{\n" + f' users = [ "{os.getenv("USER", "your-user")}" ];\n' + f' commands = [{{ command = "{runtime}"; options = [ "NOPASSWD" ]; }}];\n' + f" }}];\n" + f"\n" + f"Or run: sudo hermes {' '.join(cli_args)}", + file=sys.stderr, + ) + sys.exit(1) + + env_flags = [] + for var in ("TERM", "COLORTERM", "LANG", "LC_ALL"): + val = os.environ.get(var) + if val: + env_flags.extend(["-e", f"{var}={val}"]) + + exec_cmd = ( + cmd_prefix + + ["exec", "-it" if sys.stdin.isatty() else "-i", "-u", exec_user] + + env_flags + + [container_name, hermes_bin] + + cli_args + ) + os.execvp(exec_cmd[0], exec_cmd) diff --git a/tests/fixtures/resolution_allowlist.json b/tests/fixtures/resolution_allowlist.json index 30a0d3d461..ff94cfcb30 100644 --- a/tests/fixtures/resolution_allowlist.json +++ b/tests/fixtures/resolution_allowlist.json @@ -250,8 +250,8 @@ "kind": "bare_which" }, { - "path": "hermes_cli/main.py", - "symbol": "_exec_in_container", + "path": "hermes_cli/main_container.py", + "symbol": "exec_in_container", "kind": "bare_which" }, { diff --git a/tests/hermes_cli/test_container_aware_cli.py b/tests/hermes_cli/test_container_aware_cli.py index 033218755c..c3604d3afa 100644 --- a/tests/hermes_cli/test_container_aware_cli.py +++ b/tests/hermes_cli/test_container_aware_cli.py @@ -51,7 +51,7 @@ def test_get_container_exec_info_returns_metadata(container_env): # ============================================================================= -# _exec_in_container +# exec_in_container (hermes_cli/main_container.py) # ============================================================================= @@ -68,7 +68,7 @@ def docker_container_info(): def test_exec_in_container_calls_execvp(docker_container_info): """Verifies os.execvp is called with correct args: runtime, tty flags, user, env vars, container name, binary, and CLI args.""" - from hermes_cli.main import _exec_in_container + from hermes_cli.main_container import exec_in_container with patch("shutil.which", return_value="/usr/bin/docker"), \ patch("subprocess.run") as mock_run, \ @@ -79,7 +79,7 @@ def test_exec_in_container_calls_execvp(docker_container_info): mock_stdin.isatty.return_value = True mock_run.return_value = MagicMock(returncode=0) - _exec_in_container(docker_container_info, ["chat", "-m", "opus"]) + exec_in_container(docker_container_info, ["chat", "-m", "opus"]) mock_execvp.assert_called_once() cmd = mock_execvp.call_args[0][1]