6f4a16e1a8
Optional-skill scripts can run inside the backend process, where a child that inherits fd 0 can end the TUI/desktop JSON-RPC session with stdin EOF (#14036). Add optional-skills/ to the guard's context dirs. SKIP_DIRS used to match any part of the absolute path, so a context dir named optional-skills (and any subdir named scripts, skills, gateway, cron or hermes_cli) was never scanned. Match only "tests", relative to each context dir. The wider scan flags three calls; they now pass stdin=subprocess.DEVNULL: comfyui auto_fix_deps.run_cmd, unbroker crypto.ensure_identity and ast-grep cmd_install (its installers do not read stdin). Touching those lines makes the code-health ratchet treat their missing timeout as new: age-keygen gets timeout=30; the comfyui node/model installs and the ast-grep installer download for as long as the network needs, so they carry a reviewed HX006 allow comment instead.
275 lines
10 KiB
Python
275 lines
10 KiB
Python
#!/usr/bin/env python3
|
|
"""Check that subprocess calls in TUI-context code specify stdin=.
|
|
|
|
When Hermes runs in TUI mode, the gateway child process communicates with
|
|
the Node.js parent over a JSON-RPC protocol on stdin. Subprocess calls that
|
|
inherit this fd can cause the gateway to exit with stdin EOF during tool
|
|
execution (issue #14036, PR #39257).
|
|
|
|
This script checks that all subprocess.run() and subprocess.Popen() calls
|
|
in TUI-context files (agent/, tools/, plugins/, tui_gateway/,
|
|
optional-skills/) explicitly set stdin= to prevent fd inheritance.
|
|
|
|
Exit codes:
|
|
0 — all calls are safe
|
|
1 — violations found
|
|
2 — script error
|
|
|
|
Usage:
|
|
python scripts/check_subprocess_stdin.py [--fix]
|
|
|
|
With --fix, prints the commands to add stdin=subprocess.DEVNULL to each
|
|
violation (does not modify files).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import ast
|
|
import os
|
|
import re
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
# Directories that run inside the TUI gateway child process.
|
|
TUI_CONTEXT_DIRS = [
|
|
"agent/",
|
|
"tools/",
|
|
"plugins/",
|
|
"tui_gateway/",
|
|
# Optional-skill scripts can run inside the backend process (a command
|
|
# builder calling a skill helper on a backend thread); their children would
|
|
# inherit the JSON-RPC stdin too.
|
|
"optional-skills/",
|
|
]
|
|
|
|
# User plugin roots — scanned at runtime if they exist. Plugins load from
|
|
# ``get_hermes_home() / "plugins"`` (user) and ``./.hermes/plugins/`` (project,
|
|
# gated behind ``HERMES_ENABLE_PROJECT_PLUGINS``) — see
|
|
# ``hermes_cli/plugins.py:10-12``. The guard only checked the bundled
|
|
# ``plugins/`` dir, missing user-installed code that spawns subprocesses
|
|
# (gap reported in #67639).
|
|
#
|
|
# Import is deferred to ``main()`` (after ``os.chdir(repo_root)``) because
|
|
# this script runs as a standalone subprocess — ``hermes_constants`` isn't
|
|
# on ``sys.path`` until the repo root is added.
|
|
|
|
# subprocess and os APIs that inherit stdin by default when called without
|
|
# an explicit stdin= argument. The original regex only covered run/Popen
|
|
# (gap #1 in #67639); call, check_output, check_call, os.system, and
|
|
# asyncio.create_subprocess_* all inherit fd 0 equally.
|
|
_SUBPROCESS_PATTERNS = [
|
|
r"subprocess\.(run|Popen|call|check_output|check_call)\s*\([\"'a-zA-Z_\[\(]",
|
|
r"os\.system\s*\([\"'a-zA-Z_\[\(]",
|
|
r"asyncio\.create_subprocess_(exec|shell)\s*\([\"'a-zA-Z_\[\(]",
|
|
]
|
|
|
|
# Files with intentional stdin= override (e.g. input= creates a pipe).
|
|
# Format: "filepath:line" or just "filepath" to skip the whole file.
|
|
KNOWN_SAFE = {
|
|
"agent/shell_hooks.py", # uses input=stdin_json, creates a pipe
|
|
"plugins/security-guidance/patterns.py", # subprocess mentions are in reminder strings, not calls
|
|
}
|
|
|
|
# Inline marker that exempts a single subprocess call from this check.
|
|
# Put it in a comment on (or within) the call when the process MUST inherit
|
|
# stdin — e.g. an interactive login the user explicitly invokes. Travels with
|
|
# the line, so it survives edits that shift line numbers (unlike a pinned
|
|
# file:line entry).
|
|
EXEMPT_MARKER = "noqa: subprocess-stdin"
|
|
|
|
# Directory names skipped at any depth below a context dir. Matching against the
|
|
# path relative to that dir keeps a skill's own ``scripts/`` folder in scope.
|
|
SKIP_DIRS = {"tests"}
|
|
|
|
|
|
_SPLAT_RE = re.compile(r"\*\*\s*([A-Za-z_][A-Za-z0-9_]*)")
|
|
|
|
|
|
def _splat_carries_stdin(call_text: str, content: str) -> bool:
|
|
"""True when the call splats ``**name`` / ``**name(...)`` and ``name`` is defined in
|
|
the same file (assignment or ``def``) whose OWN expression/body sets ``stdin=``.
|
|
|
|
Shared kwargs helpers (``_RUN_KW = dict(..., stdin=DEVNULL)``, ``def _run_kwargs(): return
|
|
dict(..., stdin=DEVNULL)``) legitimately carry the guard; we only accept them when the
|
|
definition provably sets stdin= — never on the helper's name alone, and never because an
|
|
unrelated later call in the file happens to pass ``stdin=``.
|
|
"""
|
|
names = set(_SPLAT_RE.findall(call_text))
|
|
if not names:
|
|
return False
|
|
try:
|
|
tree = ast.parse(content)
|
|
except SyntaxError:
|
|
return False
|
|
for name in names:
|
|
node = None
|
|
for n in ast.walk(tree):
|
|
if isinstance(n, (ast.FunctionDef, ast.AsyncFunctionDef)) and n.name == name:
|
|
node = n
|
|
break
|
|
if isinstance(n, (ast.Assign, ast.AnnAssign)):
|
|
targets = n.targets if isinstance(n, ast.Assign) else [n.target]
|
|
if any(isinstance(t, ast.Name) and t.id == name for t in targets):
|
|
node = n.value if n.value is not None else n
|
|
break
|
|
if node is None:
|
|
return False
|
|
# stdin appears as a keyword (dict(stdin=...)) or as a dict-literal key ({"stdin": ...})
|
|
# somewhere INSIDE this definition — not merely nearby in the file.
|
|
has = any(
|
|
(isinstance(sub, ast.keyword) and sub.arg == "stdin")
|
|
or (isinstance(sub, ast.Constant) and sub.value == "stdin")
|
|
for sub in ast.walk(node)
|
|
)
|
|
if not has:
|
|
return False
|
|
return True
|
|
|
|
|
|
def find_subprocess_calls(content: str, filepath: str) -> list[dict]:
|
|
"""Find all subprocess/os/asyncio calls missing stdin= in content."""
|
|
violations = []
|
|
lines = content.split("\n")
|
|
|
|
# Match only actual function calls — not comments, docstrings, or prose.
|
|
# Multiple patterns cover subprocess.run/Popen/call/check_output/check_call,
|
|
# os.system, and asyncio.create_subprocess_exec/shell.
|
|
patterns = [re.compile(p) for p in _SUBPROCESS_PATTERNS]
|
|
|
|
for i, line in enumerate(lines):
|
|
# Skip comments.
|
|
stripped = line.lstrip()
|
|
if stripped.startswith("#"):
|
|
continue
|
|
|
|
# Skip lines where the match is inside backticks (docstring references).
|
|
if "``subprocess" in line:
|
|
continue
|
|
|
|
if not any(p.search(line) for p in patterns):
|
|
continue
|
|
|
|
# Collect the full call (may span multiple lines).
|
|
call_start = i
|
|
paren_depth = 0
|
|
found_open = False
|
|
call_lines = []
|
|
for j in range(i, min(i + 30, len(lines))):
|
|
call_lines.append(lines[j])
|
|
for ch in lines[j]:
|
|
if ch == "(":
|
|
paren_depth += 1
|
|
found_open = True
|
|
elif ch == ")":
|
|
paren_depth -= 1
|
|
if found_open and paren_depth == 0:
|
|
call_text = "\n".join(call_lines)
|
|
|
|
# Already has stdin= → safe.
|
|
if "stdin=" in call_text:
|
|
break
|
|
|
|
# Has input= → creates a pipe, safe.
|
|
if "input=" in call_text:
|
|
break
|
|
|
|
# Splats a same-file kwargs helper whose definition
|
|
# sets stdin= → the guard travels with the helper.
|
|
if _splat_carries_stdin(call_text, content):
|
|
break
|
|
|
|
# Inline exemption marker on the call itself or within
|
|
# the few comment lines immediately above it → the call
|
|
# intentionally inherits stdin.
|
|
window_start = max(0, i - 4)
|
|
preceding = "\n".join(lines[window_start:i])
|
|
if EXEMPT_MARKER in call_text or EXEMPT_MARKER in preceding:
|
|
break
|
|
|
|
violations.append({
|
|
"file": filepath,
|
|
"line": i + 1,
|
|
"snippet": line.strip()[:120],
|
|
})
|
|
break
|
|
else:
|
|
continue
|
|
break
|
|
|
|
return violations
|
|
|
|
|
|
def main() -> int:
|
|
fix_mode = "--fix" in sys.argv
|
|
repo_root = Path(__file__).resolve().parent.parent
|
|
os.chdir(repo_root)
|
|
|
|
# Add repo root to sys.path so we can import hermes_constants (this script
|
|
# runs as a standalone subprocess, not as a module).
|
|
sys.path.insert(0, str(repo_root))
|
|
from hermes_constants import get_hermes_home
|
|
|
|
all_violations = []
|
|
|
|
for tui_dir in TUI_CONTEXT_DIRS:
|
|
dirpath = repo_root / tui_dir
|
|
if not dirpath.exists():
|
|
continue
|
|
|
|
for py_file in dirpath.rglob("*.py"):
|
|
rel = str(py_file.relative_to(repo_root))
|
|
|
|
# Skip known-safe files. ``relative_to`` returns a host-separated
|
|
# path (backslashes on Windows) while KNOWN_SAFE is forward-slash.
|
|
if py_file.relative_to(repo_root).as_posix() in KNOWN_SAFE:
|
|
continue
|
|
|
|
# Skip test files inside tools/ etc.
|
|
if SKIP_DIRS & set(py_file.relative_to(dirpath).parts):
|
|
continue
|
|
|
|
content = py_file.read_text(encoding="utf-8-sig")
|
|
violations = find_subprocess_calls(content, rel)
|
|
all_violations.extend(violations)
|
|
|
|
# Scan user plugin directories (Gap 1: guard missed user-installed
|
|
# plugins in get_hermes_home()/plugins/ and project plugins in
|
|
# ./.hermes/plugins/, where code like ori/hooks.py can spawn
|
|
# subprocesses with inherited stdin — #67639).
|
|
plugin_roots: list[Path] = [get_hermes_home() / "plugins"]
|
|
if os.environ.get("HERMES_ENABLE_PROJECT_PLUGINS"):
|
|
plugin_roots.append(Path.cwd() / ".hermes" / "plugins")
|
|
seen_roots: set[Path] = set()
|
|
for plugin_root in plugin_roots:
|
|
resolved = plugin_root.resolve()
|
|
if resolved in seen_roots or not resolved.is_dir():
|
|
continue
|
|
seen_roots.add(resolved)
|
|
|
|
for py_file in resolved.rglob("*.py"):
|
|
rel = str(py_file)
|
|
if py_file.name in ("conftest.py",) or "/tests/" in rel:
|
|
continue
|
|
|
|
try:
|
|
content = py_file.read_text(encoding="utf-8-sig")
|
|
except Exception:
|
|
continue
|
|
violations = find_subprocess_calls(content, rel)
|
|
all_violations.extend(violations)
|
|
|
|
if all_violations:
|
|
print(f"❌ {len(all_violations)} subprocess calls missing stdin=:")
|
|
for v in all_violations:
|
|
print(f" {v['file']}:{v['line']}: {v['snippet']}")
|
|
if fix_mode:
|
|
print("\nAdd stdin=subprocess.DEVNULL to each call above.")
|
|
return 1
|
|
else:
|
|
print("✅ All TUI-context subprocess calls have explicit stdin=")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|