fix: harden Kanban worker Hermes command resolution

This commit is contained in:
hanzckernel 2026-05-18 20:25:03 -07:00 committed by Teknium
parent 0b547aea03
commit 5d079fee17
2 changed files with 204 additions and 12 deletions

View file

@ -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(

View file

@ -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}); "