From 5d079fee17cd945fdc40295588361255876889aa Mon Sep 17 00:00:00 2001 From: hanzckernel <43314240+hanzckernel@users.noreply.github.com> Date: Mon, 18 May 2026 20:25:03 -0700 Subject: [PATCH] fix: harden Kanban worker Hermes command resolution --- hermes_cli/kanban_db.py | 107 +++++++++++++++++++++++++--- tests/hermes_cli/test_kanban_db.py | 109 ++++++++++++++++++++++++++++- 2 files changed, 204 insertions(+), 12 deletions(-) diff --git a/hermes_cli/kanban_db.py b/hermes_cli/kanban_db.py index 49552a4fcfb..28ea93aa4e8 100644 --- a/hermes_cli/kanban_db.py +++ b/hermes_cli/kanban_db.py @@ -4285,15 +4285,96 @@ def _rotate_worker_log( pass +def _module_hermes_argv() -> list[str]: + """Return the interpreter-bound Hermes CLI invocation.""" + # ``hermes_cli.main`` is the console-script target declared in + # pyproject.toml, NOT a top-level ``hermes`` package — there is no + # ``hermes`` package to import. + return [sys.executable, "-m", "hermes_cli.main"] + + +def _absolute_hermes_path(path: str) -> str: + """Return an absolute filesystem path for a resolved Hermes shim.""" + expanded = os.path.expanduser(path) + return expanded if os.path.isabs(expanded) else os.path.abspath(expanded) + + +def _looks_like_path(value: str) -> bool: + """Return true when a command override is an explicit path, not a name.""" + expanded = os.path.expanduser(value) + return ( + expanded.startswith("~") + or os.path.isabs(expanded) + or bool(os.path.dirname(expanded)) + or "\\" in expanded + or bool(re.match(r"^[A-Za-z]:", expanded)) + ) + + +def _is_windows_batch_shim(path: str) -> bool: + """Return true for Windows shell/batch shims that should not be argv[0].""" + return path.lower().endswith((".cmd", ".bat")) + + +def _path_search_names(command: str) -> list[str]: + """Return executable names to try for an unqualified command.""" + if not _IS_WINDOWS or os.path.splitext(command)[1]: + return [command] + raw = os.environ.get("PATHEXT") or ".COM;.EXE;.BAT;.CMD" + exts = [ext for ext in raw.split(";") if ext] + return [command + ext for ext in exts] + + +def _safe_which_no_cwd(command: str) -> Optional[str]: + """Resolve a bare command from PATH without implicit current-dir search. + + ``shutil.which`` follows platform search behavior. On Windows that can + include the current directory before PATH for bare names, which is not a + safe dispatcher primitive. This resolver only considers explicit PATH + entries and skips empty / ``.`` entries. + """ + path_env = os.environ.get("PATH", "") + for raw_dir in path_env.split(os.pathsep): + if not raw_dir or raw_dir == ".": + continue + directory = os.path.expanduser(raw_dir) + for name in _path_search_names(command): + candidate = os.path.join(directory, name) + if not os.path.isfile(candidate): + continue + if _IS_WINDOWS or os.access(candidate, os.X_OK): + return candidate + return None + + +def _hermes_path_argv(path: str) -> list[str]: + """Return argv for a resolved Hermes executable path. + + Windows batch shims (`.cmd` / `.bat`) are not safe as argv[0] for + worker launches because the argument vector includes task-derived + values. Prefer the interpreter-bound module form whenever the resolved + executable is only a shell shim. + """ + if _IS_WINDOWS and _is_windows_batch_shim(path): + return _module_hermes_argv() + return [_absolute_hermes_path(path)] + + def _resolve_hermes_argv() -> list[str]: """Resolve the ``hermes`` invocation as argv parts for ``Popen``. Tries in order: - 1. ``shutil.which("hermes")`` — the console-script shim, the same form - that shows up in ``ps`` output and existing logs. Preferred so live - systems' diagnostics stay familiar. - 2. ``sys.executable -m hermes_cli.main`` — fallback for setups where + 1. ``$HERMES_BIN`` — explicit operator override. Path-like values are + normalized to absolute paths; bare command names keep normal PATH + semantics and never prefer a same-directory file before ``PATH``. + 2. ``shutil.which("hermes")`` — the console-script shim, normalized to + an absolute path. On Windows, ``which`` can return a relative + ``.\\hermes.CMD`` when the current directory is on ``PATH``; directly + launching batch shims is also unsafe with task-derived argv. The + dispatcher therefore falls back to the interpreter-bound module form + for implicit ``.cmd`` / ``.bat`` shims. + 3. ``sys.executable -m hermes_cli.main`` — fallback for setups where Hermes is launched from a venv and the ``hermes`` shim is not on the dispatcher's ``$PATH`` (cron, systemd ``User=`` services, launchd jobs, detached processes, etc.). Goes through the running @@ -4305,13 +4386,19 @@ def _resolve_hermes_argv() -> list[str]: """ import shutil - hermes_bin = shutil.which("hermes") + env_bin = os.environ.get("HERMES_BIN", "").strip() + if env_bin: + if _looks_like_path(env_bin): + return _hermes_path_argv(env_bin) + resolved_env_bin = _safe_which_no_cwd(env_bin) + if resolved_env_bin: + return _hermes_path_argv(resolved_env_bin) + return _module_hermes_argv() + + hermes_bin = _safe_which_no_cwd("hermes") if _IS_WINDOWS else shutil.which("hermes") if hermes_bin: - return [hermes_bin] - # Fallback to the module form. ``hermes_cli.main`` is the actual - # console-script target declared in pyproject.toml, NOT a top-level - # ``hermes`` package — there is no ``hermes`` package to import. - return [sys.executable, "-m", "hermes_cli.main"] + return _hermes_path_argv(hermes_bin) + return _module_hermes_argv() def _worker_terminal_timeout_env( diff --git a/tests/hermes_cli/test_kanban_db.py b/tests/hermes_cli/test_kanban_db.py index bd7797ce45a..384f026a14a 100644 --- a/tests/hermes_cli/test_kanban_db.py +++ b/tests/hermes_cli/test_kanban_db.py @@ -1564,11 +1564,113 @@ def test_resolve_hermes_argv_prefers_path_shim(monkeypatch): import shutil import hermes_cli.kanban_db as kb + monkeypatch.delenv("HERMES_BIN", raising=False) monkeypatch.setattr(shutil, "which", lambda name: "/usr/local/bin/hermes") argv = kb._resolve_hermes_argv() assert argv == ["/usr/local/bin/hermes"] +def test_resolve_hermes_argv_absolutizes_relative_exe_shim(monkeypatch, tmp_path): + """A relative executable override must not remain workspace-cwd-dependent.""" + import hermes_cli.kanban_db as kb + + monkeypatch.chdir(tmp_path) + monkeypatch.setenv("HERMES_BIN", ".\\hermes.exe") + monkeypatch.setattr(kb, "_IS_WINDOWS", True) + + assert kb._resolve_hermes_argv() == [os.path.abspath(".\\hermes.exe")] + + +def test_resolve_hermes_argv_avoids_implicit_windows_batch_shim(monkeypatch, tmp_path): + """Implicit .cmd/.bat shims use the module fallback, not batch argv[0].""" + import sys + import hermes_cli.kanban_db as kb + + bin_dir = tmp_path / "bin" + bin_dir.mkdir() + (bin_dir / "hermes.CMD").write_text("@echo off\n", encoding="utf-8") + monkeypatch.delenv("HERMES_BIN", raising=False) + monkeypatch.setenv("PATH", str(bin_dir)) + monkeypatch.setenv("PATHEXT", ".CMD") + monkeypatch.setattr(kb, "_IS_WINDOWS", True) + + assert kb._resolve_hermes_argv() == [sys.executable, "-m", "hermes_cli.main"] + + +def test_resolve_hermes_argv_honors_hermes_bin_path_override(monkeypatch, tmp_path): + """An explicit path-like HERMES_BIN lets service managers pin the executable.""" + import shutil + import hermes_cli.kanban_db as kb + + shim = tmp_path / "bin" / "hermes" + shim.parent.mkdir() + shim.write_text("#!/bin/sh\n", encoding="utf-8") + monkeypatch.setenv("HERMES_BIN", str(shim)) + monkeypatch.setattr(shutil, "which", lambda name: None) + + assert kb._resolve_hermes_argv() == [str(shim)] + + +def test_resolve_hermes_argv_hermes_bin_bare_name_uses_path(monkeypatch, tmp_path): + """Bare HERMES_BIN values keep PATH semantics instead of cwd shadowing.""" + import stat + import hermes_cli.kanban_db as kb + + cwd_hermes = tmp_path / "hermes" + cwd_hermes.write_text("wrong\n", encoding="utf-8") + cwd_hermes.chmod(cwd_hermes.stat().st_mode | stat.S_IXUSR) + path_hermes = tmp_path / "bin" / "hermes" + path_hermes.parent.mkdir() + path_hermes.write_text("right\n", encoding="utf-8") + path_hermes.chmod(path_hermes.stat().st_mode | stat.S_IXUSR) + monkeypatch.chdir(tmp_path) + monkeypatch.setenv("PATH", str(path_hermes.parent)) + monkeypatch.setenv("HERMES_BIN", "hermes") + + assert kb._resolve_hermes_argv() == [str(path_hermes)] + + +def test_resolve_hermes_argv_hermes_bin_bare_name_ignores_cwd(monkeypatch, tmp_path): + """Bare HERMES_BIN does not accept current-directory shadow executables.""" + import sys + import hermes_cli.kanban_db as kb + + (tmp_path / "hermes.exe").write_text("wrong\n", encoding="utf-8") + monkeypatch.chdir(tmp_path) + monkeypatch.setenv("PATH", "") + monkeypatch.setenv("HERMES_BIN", "hermes") + monkeypatch.setattr(kb, "_IS_WINDOWS", True) + + assert kb._resolve_hermes_argv() == [sys.executable, "-m", "hermes_cli.main"] + + +def test_resolve_hermes_argv_hermes_bin_bare_cmd_uses_module_fallback(monkeypatch, tmp_path): + """A PATH-resolved HERMES_BIN batch shim is not used as worker argv[0].""" + import sys + import hermes_cli.kanban_db as kb + + bin_dir = tmp_path / "bin" + bin_dir.mkdir() + (bin_dir / "hermes.CMD").write_text("@echo off\n", encoding="utf-8") + monkeypatch.setenv("PATH", str(bin_dir)) + monkeypatch.setenv("PATHEXT", ".CMD") + monkeypatch.setenv("HERMES_BIN", "hermes") + monkeypatch.setattr(kb, "_IS_WINDOWS", True) + + assert kb._resolve_hermes_argv() == [sys.executable, "-m", "hermes_cli.main"] + + +def test_resolve_hermes_argv_hermes_bin_unresolved_bare_name_falls_back(monkeypatch): + """Unresolved HERMES_BIN command names do not delegate cwd search to Popen.""" + import sys + import hermes_cli.kanban_db as kb + + monkeypatch.setenv("PATH", "") + monkeypatch.setenv("HERMES_BIN", "hermes") + + assert kb._resolve_hermes_argv() == [sys.executable, "-m", "hermes_cli.main"] + + def test_resolve_hermes_argv_falls_back_to_module_form_when_no_path_shim(monkeypatch): """When the shim is not on PATH, fall back to `python -m hermes_cli.main`. @@ -1581,6 +1683,7 @@ def test_resolve_hermes_argv_falls_back_to_module_form_when_no_path_shim(monkeyp import sys import hermes_cli.kanban_db as kb + monkeypatch.delenv("HERMES_BIN", raising=False) monkeypatch.setattr(shutil, "which", lambda name: None) argv = kb._resolve_hermes_argv() assert argv == [sys.executable, "-m", "hermes_cli.main"] @@ -1601,8 +1704,10 @@ def test_resolve_hermes_argv_module_actually_runs(): import shutil import unittest.mock as mock - with mock.patch.object(shutil, "which", return_value=None): - argv = kb._resolve_hermes_argv() + with mock.patch.dict(os.environ, {}, clear=False): + os.environ.pop("HERMES_BIN", None) + with mock.patch.object(shutil, "which", return_value=None): + argv = kb._resolve_hermes_argv() r = subprocess.run(argv + ["--version"], capture_output=True, text=True, timeout=30) assert r.returncode == 0, ( f"`{' '.join(argv)} --version` failed (rc={r.returncode}); "