From 8b249e4b3d44c3dec656df24bbf81ae771997981 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 22:24:19 -0700 Subject: [PATCH] refactor(tools): compact local env builders, env_probe, env_passthrough, pythonpath/gitbash helpers; fold force-unwrap and kill-group flows --- tests/tools/test_windows_native_support.py | 1 + tools/env_passthrough.py | 130 ++--- tools/env_probe.py | 150 ++--- tools/environments/local.py | 647 ++++++++------------- tools/environments/local_env_policy.py | 98 ++-- tools/environments/local_gitbash_probe.py | 40 +- tools/environments/local_pythonpath.py | 106 ++-- 7 files changed, 420 insertions(+), 752 deletions(-) diff --git a/tests/tools/test_windows_native_support.py b/tests/tools/test_windows_native_support.py index b1adc95bf6..96df4cb424 100644 --- a/tests/tools/test_windows_native_support.py +++ b/tests/tools/test_windows_native_support.py @@ -720,6 +720,7 @@ class TestLocalEnvironmentWindowsTempDir: assert "if _IS_WINDOWS:" in source assert "get_hermes_home" in source assert 'cache_dir = get_hermes_home() / "cache" / "terminal"' in source + assert "_default_terminal_temp_dir() or Path(tempfile.gettempdir())" in source class TestLocalEnvironmentPathInjectionGated: diff --git a/tools/env_passthrough.py b/tools/env_passthrough.py index 0c921c84cf..64fb99f784 100644 --- a/tools/env_passthrough.py +++ b/tools/env_passthrough.py @@ -1,16 +1,11 @@ """Environment variable passthrough registry. -Skills that declare ``required_environment_variables`` need those vars in -sandboxed execution environments (execute_code, terminal), which strip secrets -from the child process environment by default. This module is the -session-scoped allowlist, fed by two sources: skill declarations (registered -automatically by ``skill_view``) and ``terminal.env_passthrough`` in -config.yaml. - -``code_execution_tool.py`` and ``tools/environments/local.py`` consult -:func:`is_env_passthrough` before stripping a variable. When profile -multiplexing is active, forwarded values are resolved through the current -profile's secret scope rather than the process environment. +Skills that declare ``required_environment_variables`` need those vars in sandboxed +execution environments (execute_code, terminal), which strip secrets from the child +env by default. This module is the session-scoped allowlist, fed by skill +declarations (registered by ``skill_view``) and ``terminal.env_passthrough`` in +config.yaml. When profile multiplexing is active, forwarded values are resolved +through the current profile's secret scope rather than the process environment. """ from __future__ import annotations @@ -43,18 +38,17 @@ _config_passthrough: frozenset[str] | None = None def _is_hermes_provider_credential(name: str) -> bool: """True if ``name`` is a Hermes-managed provider credential per - ``_HERMES_PROVIDER_ENV_BLOCKLIST`` (or a dynamic Hermes-internal secret). + ``_HERMES_PROVIDER_ENV_BLOCKLIST`` or a dynamic Hermes-internal secret + (AUXILIARY_*_API_KEY / _BASE_URL, GATEWAY_RELAY_*, injected per task/relay so + the static list can't enumerate them). - Skill-declared ``required_environment_variables`` must not override this - list — that was the GHSA-rhgp-j443-p4rf bypass, where a malicious skill - registered ``OPENAI_API_KEY`` as passthrough and received it in the - ``execute_code`` child, defeating the sandbox's scrubbing guarantee. - Non-Hermes API keys (TENOR_API_KEY, NOTION_TOKEN, …) are not in the - blocklist and remain registerable. + Skill-declared ``required_environment_variables`` must not override this list — + that was the GHSA-rhgp-j443-p4rf bypass (a malicious skill registered + ``OPENAI_API_KEY`` as passthrough and received it in the ``execute_code`` child). + Non-Hermes API keys (TENOR_API_KEY, NOTION_TOKEN, …) remain registerable. - Fail closed: if the authoritative blocklist cannot be imported (partial - install, import-time error), treat the name as protected and refuse - passthrough rather than fall open. + Fail closed: if the authoritative blocklist cannot be imported (partial install, + import-time error), treat the name as protected. """ try: from tools.environments.local import ( @@ -69,12 +63,7 @@ def _is_hermes_provider_credential(name: str) -> bool: e, ) return True - # Dynamically-generated Hermes-internal secrets (AUXILIARY_*_API_KEY / - # _BASE_URL, GATEWAY_RELAY_*) are injected per task/relay at gateway - # startup, so the static blocklist can't enumerate them. - if _is_hermes_internal_secret(name): - return True - return name in _HERMES_PROVIDER_ENV_BLOCKLIST + return _is_hermes_internal_secret(name) or name in _HERMES_PROVIDER_ENV_BLOCKLIST def register_env_passthrough(var_names: Iterable[str]) -> None: @@ -82,10 +71,9 @@ def register_env_passthrough(var_names: Iterable[str]) -> None: from a skill's ``required_environment_variables``). Hermes-managed provider credentials are rejected to preserve the - ``execute_code`` sandbox's credential-scrubbing guarantee - (GHSA-rhgp-j443-p4rf); a skill needing a Hermes-managed provider should - use the main-process tools (web_search, web_extract, …) where the - credential stays in the main process. Third-party keys pass normally. + ``execute_code`` sandbox's credential-scrubbing guarantee (GHSA-rhgp-j443-p4rf); + a skill needing a Hermes-managed provider should use the main-process tools + (web_search, web_extract, …). Third-party keys pass normally. """ for name in var_names: name = name.strip() @@ -105,7 +93,9 @@ def register_env_passthrough(var_names: Iterable[str]) -> None: def _load_config_passthrough() -> frozenset[str]: - """Load ``tools.env_passthrough`` from config.yaml (cached).""" + """Load ``tools.env_passthrough`` from config.yaml (cached). Same credential + filter as register_env_passthrough: operator config must not tunnel provider + credentials into sandbox children either (GHSA-rhgp-j443-p4rf).""" global _config_passthrough if _config_passthrough is not None: return _config_passthrough @@ -113,28 +103,23 @@ def _load_config_passthrough() -> frozenset[str]: result: set[str] = set() try: from hermes_cli.config import read_raw_config - cfg = read_raw_config() - passthrough = cfg_get(cfg, "terminal", "env_passthrough") - if isinstance(passthrough, list): - for item in passthrough: - if not isinstance(item, str) or not item.strip(): - continue - name = item.strip() - # Same filter as register_env_passthrough: provider credentials - # must not reach sandbox children whether the request came from - # a skill or from config.yaml (GHSA-rhgp-j443-p4rf). - if _is_hermes_provider_credential(name): - logger.warning( - "env passthrough: refusing to register Hermes " - "provider credential %r from config.yaml (blocked " - "by _HERMES_PROVIDER_ENV_BLOCKLIST). Operator " - "configuration must not override the execute_code " - "sandbox's credential scrubbing; see " - "GHSA-rhgp-j443-p4rf.", - name, - ) - continue - result.add(name) + passthrough = cfg_get(read_raw_config(), "terminal", "env_passthrough") + for item in passthrough if isinstance(passthrough, list) else (): + name = item.strip() if isinstance(item, str) else "" + if not name: + continue + if _is_hermes_provider_credential(name): + logger.warning( + "env passthrough: refusing to register Hermes " + "provider credential %r from config.yaml (blocked " + "by _HERMES_PROVIDER_ENV_BLOCKLIST). Operator " + "configuration must not override the execute_code " + "sandbox's credential scrubbing; see " + "GHSA-rhgp-j443-p4rf.", + name, + ) + continue + result.add(name) except Exception as e: logger.debug("Could not read tools.env_passthrough from config: %s", e) @@ -144,9 +129,7 @@ def _load_config_passthrough() -> frozenset[str]: def is_env_passthrough(var_name: str) -> bool: """True if *var_name* was registered by a skill or listed in config.""" - if var_name in _get_allowed(): - return True - return var_name in _load_config_passthrough() + return var_name in _get_allowed() or var_name in _load_config_passthrough() def get_all_passthrough() -> frozenset[str]: @@ -154,20 +137,16 @@ def get_all_passthrough() -> frozenset[str]: return frozenset(_get_allowed()) | _load_config_passthrough() -def resolve_passthrough_value( - name: str, - fallback: str | None = None, -) -> str | None: +def resolve_passthrough_value(name: str, fallback: str | None = None) -> str | None: """Resolve an allowlisted variable without crossing profile boundaries. - ``fallback`` is the value the caller would have forwarded before profile - secret scopes existed (typically a snapshot of ``os.environ`` or the - current profile's ``.env``). An active multiplex scope is authoritative: - a missing key returns ``None`` and never falls back to the process-global - environment; an unscoped read while multiplexing is active raises the - fail-closed ``UnscopedSecretError`` from :mod:`agent.secret_scope`. - Outside multiplexing, an installed scope keeps the overlay semantics and - an unscoped caller keeps its already-resolved fallback. + ``fallback`` is the value the caller would have forwarded before profile secret + scopes existed (a snapshot of ``os.environ`` or the profile's ``.env``). An + active multiplex scope is authoritative: a missing key returns ``None`` and + never falls back to the process-global environment; an unscoped read while + multiplexing is active raises the fail-closed ``UnscopedSecretError`` from + :mod:`agent.secret_scope`. Outside multiplexing, an installed scope keeps the + overlay semantics and an unscoped caller keeps its already-resolved fallback. """ from agent.secret_scope import ( _is_global_env, @@ -176,19 +155,14 @@ def resolve_passthrough_value( is_multiplex_active, ) - # Global terminal/runtime settings are not profile secrets. ``fallback`` - # is already the caller's effective value (including an explicit per-call - # override), so preserve it rather than replacing it with the process-wide - # value while a multiplex scope is active. + # Global terminal/runtime settings are not profile secrets; ``fallback`` is + # already the caller's effective value (incl. an explicit per-call override). if _is_global_env(name) and fallback is not None: return fallback - scope = current_secret_scope() multiplex_active = is_multiplex_active() - if scope is None: - if multiplex_active: - return get_secret(name) - return fallback + if current_secret_scope() is None: + return get_secret(name) if multiplex_active else fallback return get_secret(name, None if multiplex_active else fallback) diff --git a/tools/env_probe.py b/tools/env_probe.py index fe1065c32d..fb625fbf8c 100644 --- a/tools/env_probe.py +++ b/tools/env_probe.py @@ -1,15 +1,10 @@ """Local-environment toolchain probe for the system prompt. -When the terminal backend is local, surface one deterministic line about -Python tooling state (python3/python versions, missing pip module, pip bound -to a different Python than ``python3``, PEP 668 externally-managed) so models -don't discover it by hitting walls. The probe is cheap (~50ms), cached for -the process lifetime, and emits nothing when the environment is clean. - -Remote terminal backends (docker, modal, ssh, …) are skipped: the host's -Python state is irrelevant when tools run inside a sandbox, which has its own -probe (``_probe_remote_backend`` in ``agent/prompt_builder.py``). - +When the terminal backend is local, surface one deterministic line about Python +tooling state (python3/python versions, missing pip module, pip bound to a +different Python, PEP 668) so models don't discover it by hitting walls. Cheap +(~50ms), cached for the process lifetime, "" when the environment is clean. +Remote backends are skipped (the sandbox has its own probe in agent/prompt_builder). Toggle via ``agent.environment_probe`` in config.yaml (default True). """ @@ -27,24 +22,19 @@ from hermes_cli._subprocess_compat import windows_hide_flags logger = logging.getLogger(__name__) # Concurrency model: the probe runs in exactly ONE background worker thread; -# ``_PROBE_DONE`` signals completion. Callers never execute the probe -# themselves and block at most ``_PROBE_WAIT_TIMEOUT`` seconds on the event -# before failing open with "" — a stuck probe (e.g. a Windows pipe wedged open -# by an orphaned pip descendant) can degrade only the probe line, never -# system-prompt construction. +# ``_PROBE_DONE`` signals completion. Callers never execute the probe themselves +# and block at most ``_PROBE_WAIT_TIMEOUT`` seconds before failing open with "" — +# a stuck probe (e.g. a Windows pipe wedged open by an orphaned pip descendant) +# can degrade only the probe line, never system-prompt construction. _CACHE_LOCK = threading.Lock() _CACHED_LINE: Optional[str] = None # None = not probed yet; "" = probed, nothing to say. _PROBE_DONE = threading.Event() _PROBE_THREAD: Optional[threading.Thread] = None -# Generation counter — bumped on every reset so a stale worker (started -# before a test reset) can't publish its result into the fresh generation. +# Bumped on every reset so a stale worker can't publish into the fresh generation. _PROBE_GEN = 0 - -# Upper bound a prompt build will wait for the probe. Generous vs the ~0.5s -# healthy runtime, but finite: prompt construction must always proceed. +# Upper bound a prompt build will wait for the probe (healthy runtime ~0.5s). _PROBE_WAIT_TIMEOUT = 10.0 -# Once one caller has burned the full wait, later callers only peek at the -# event. If the stuck worker ever finishes, the line resumes appearing. +# Once one caller has burned the full wait, later callers only peek at the event. _WAIT_ALREADY_TIMED_OUT = False # Keep in sync with agent/prompt_builder.py:_REMOTE_TERMINAL_BACKENDS. @@ -68,33 +58,24 @@ def _plugin_backend_is_remote(backend: str) -> bool: def _run(cmd: list[str], timeout: float = 3.0) -> tuple[int, str, str]: - """Run a short subprocess. Returns (returncode, stdout, stderr). + """Run a short subprocess -> (returncode, stdout, stderr); failures (binary + missing, timeout, OSError) return (-1, "", ""). - Failures (binary missing, timeout, OSError) return (-1, "", ""). - - Output is captured through temp files rather than pipes so ``timeout`` - bounds the *whole* call, even on native Windows: a console-script launcher - (e.g. ``pip.exe``) can spawn a descendant that inherits the captured - handles and outlives its parent. With OS pipes, ``communicate()``'s reader - threads block until that grandchild closes the write end — which the - timeout does not cover, since killing the direct child leaves the - grandchild holding the pipe (a warm probe could hang ~28 min holding - ``_CACHE_LOCK``). Temp files have no reader threads, so ``wait()`` only - waits on the direct child and the probe genuinely fails open on timeout. + Output goes through temp files, not pipes, so ``timeout`` bounds the *whole* + call even on native Windows: a console-script launcher (``pip.exe``) can spawn a + descendant that inherits the captured handles and outlives its parent; with OS + pipes ``communicate()``'s reader threads block until that grandchild closes the + write end (a warm probe could hang ~28 min holding ``_CACHE_LOCK``). Temp files + have no reader threads, so ``wait()`` only waits on the direct child. """ try: with tempfile.TemporaryFile() as out_f, tempfile.TemporaryFile() as err_f: try: result = subprocess.run( - cmd, - stdout=out_f, - stderr=err_f, - timeout=timeout, - check=False, + cmd, stdout=out_f, stderr=err_f, timeout=timeout, check=False, stdin=subprocess.DEVNULL, # CREATE_NO_WINDOW (0 on POSIX): windowless hosts (pythonw - # gateway / kanban workers) would otherwise flash a console - # window per probe subprocess. + # gateway / kanban workers) would otherwise flash a console. creationflags=windows_hide_flags(), ) except subprocess.TimeoutExpired: @@ -122,8 +103,7 @@ def _has_pip_module(binary: str) -> bool: """True if `` -m pip --version`` succeeds.""" if not shutil.which(binary): return False - rc, _out, _err = _run([binary, "-m", "pip", "--version"]) - return rc == 0 + return _run([binary, "-m", "pip", "--version"])[0] == 0 def _detect_pep668(binary: str) -> bool: @@ -142,17 +122,12 @@ def _detect_pep668(binary: str) -> bool: def _pip_python_version() -> Optional[str]: - """If ``pip`` is on PATH, return the Python version it's bound to. - - Parses the trailing ``(python X.Y)`` of ``pip --version`` output, e.g. - ``pip 24.0 from /usr/lib/python3/dist-packages/pip (python 3.12)`` → ``"3.12"``. - """ + """If ``pip`` is on PATH, the Python version it's bound to — the trailing + ``(python X.Y)`` of ``pip --version`` (e.g. ``"3.12"``), else None.""" if not shutil.which("pip"): return None rc, out, _err = _run(["pip", "--version"]) - if rc != 0 or not out: - return None - if "(python " in out and out.endswith(")"): + if rc == 0 and out and "(python " in out and out.endswith(")"): return out.rsplit("(python ", 1)[1][:-1].strip() return None @@ -169,37 +144,26 @@ def _resolve_terminal_backend() -> str: def _build_probe_line() -> str: - """Build the one-liner. Returns "" when nothing notable is detected — - the goal is to save the model from an avoidable wall, not narrate a - healthy environment.""" + """Build the one-liner; "" when nothing notable is detected — the goal is to + save the model from an avoidable wall, not narrate a healthy environment.""" py3_ver = _python_version_of("python3") py_ver = _python_version_of("python") # for systems with a `python` alias py3_has_pip = _has_pip_module("python3") if py3_ver else False pip_bound_to = _pip_python_version() py3_pep668 = _detect_pep668("python3") if py3_ver else False - # Bare which() is correct here, unlike Hermes's own uv call sites: this - # reports the environment *the model will see* in the terminal tool, whose - # PATH (via local.py) includes the Hermes-managed $HERMES_HOME/bin. - # Claiming uv the model cannot invoke would be worse than claiming none. + # Bare which() is correct here, unlike Hermes's own uv call sites: this reports + # the environment *the model will see* in the terminal tool, whose PATH (via + # local.py) includes the Hermes-managed $HERMES_HOME/bin. has_uv = shutil.which("uv") is not None mismatch = bool(pip_bound_to and py3_ver and not py3_ver.startswith(pip_bound_to)) - silent_conditions = ( - py3_ver is not None - and py3_has_pip - and not mismatch - and (not py3_pep668 or has_uv) - ) - if silent_conditions: + if py3_ver is not None and py3_has_pip and not mismatch and (not py3_pep668 or has_uv): return "" # Compact factual summary; ONE line so it doesn't dominate the prompt. bits: list[str] = [] if py3_ver: - py3_bit = f"python3={py3_ver}" - if not py3_has_pip: - py3_bit += " (no pip module)" - bits.append(py3_bit) + bits.append(f"python3={py3_ver}" + ("" if py3_has_pip else " (no pip module)")) else: bits.append("python3=missing") @@ -213,41 +177,31 @@ def _build_probe_line() -> str: if mismatch: bits.append(f"pip→python{pip_bound_to} (mismatch)") elif not py3_has_pip: - # pip script works but `python3 -m pip` doesn't. - bits.append(f"pip→python{pip_bound_to}") + bits.append(f"pip→python{pip_bound_to}") # pip script works, `-m pip` doesn't elif not py3_has_pip: # (when `pip` is off PATH but `python3 -m pip` works, say nothing) bits.append("pip=missing") if py3_pep668: bits.append("PEP 668=yes (use venv or uv)") - if has_uv: bits.append("uv=installed") - return "Python toolchain: " + ", ".join(bits) + "." def get_environment_probe_line(*, force_refresh: bool = False) -> str: - """Return the cached probe line (building it on first call). - - Returns "" when the environment is clean — the system prompt assembler - should drop the section rather than emit an empty heading. The probe runs - in a single background worker; this waits at most ``_PROBE_WAIT_TIMEOUT`` - seconds on its completion event and then fails open with "", so a wedged - probe subprocess can never block system-prompt construction. - - ``force_refresh`` is for tests; real callers should never need it. - """ + """Return the cached probe line (building it on first call); "" when the + environment is clean, so the prompt assembler drops the section. Waits at most + ``_PROBE_WAIT_TIMEOUT`` on the single worker, then fails open with "". + ``force_refresh`` is for tests.""" global _WAIT_ALREADY_TIMED_OUT if force_refresh: _reset_cache_for_tests() - # Resolve the backend HERE, in the caller's context: under gateway - # multiplexing the routed profile's backend lives in the per-turn terminal - # scope, which the bare probe worker thread does not inherit. A remote - # backend answers "" without consulting the cache — the cached line - # describes the HOST toolchain, not where that profile's tools run. + # Resolve the backend HERE, in the caller's context: under gateway multiplexing + # the routed profile's backend lives in the per-turn terminal scope, which the + # bare worker thread does not inherit. A remote backend answers "" without + # consulting the cache — the cached line describes the HOST toolchain. backend = _resolve_terminal_backend() if backend in _REMOTE_BACKENDS or _plugin_backend_is_remote(backend): return "" @@ -259,7 +213,7 @@ def get_environment_probe_line(*, force_refresh: bool = False) -> str: wait_timeout = 0.05 if _WAIT_ALREADY_TIMED_OUT else _PROBE_WAIT_TIMEOUT if not _PROBE_DONE.wait(timeout=wait_timeout): # Probe stuck or pathologically slow: the line is a nice-to-have, - # blocking prompt construction is an outage. Fail open. + # blocking prompt construction is an outage. Fail open. if not _WAIT_ALREADY_TIMED_OUT: _WAIT_ALREADY_TIMED_OUT = True logger.warning( @@ -290,26 +244,18 @@ def _ensure_probe_started() -> None: """Start the probe worker if it isn't running and hasn't finished.""" global _PROBE_THREAD with _CACHE_LOCK: - if _PROBE_DONE.is_set(): - return - if _PROBE_THREAD is not None and _PROBE_THREAD.is_alive(): + if _PROBE_DONE.is_set() or (_PROBE_THREAD is not None and _PROBE_THREAD.is_alive()): return _PROBE_THREAD = threading.Thread( - target=_probe_worker, - args=(_PROBE_GEN,), - name="env-probe", - daemon=True, + target=_probe_worker, args=(_PROBE_GEN,), name="env-probe", daemon=True, ) _PROBE_THREAD.start() def warm_environment_probe_async() -> None: - """Start the probe in the background so the first system-prompt build - doesn't pay the ~0.5s of subprocess calls on the time-to-first-token path. - - Idempotent and fail-safe; ``get_environment_probe_line`` waits (bounded) - on the same worker instead of recomputing. Called from agent init. - """ + """Start the probe in the background so the first system-prompt build doesn't + pay the ~0.5s of subprocess calls on the time-to-first-token path. Idempotent; + ``get_environment_probe_line`` waits (bounded) on the same worker.""" _ensure_probe_started() diff --git a/tools/environments/local.py b/tools/environments/local.py index 2978e3d23c..39a9972dcb 100644 --- a/tools/environments/local.py +++ b/tools/environments/local.py @@ -26,37 +26,32 @@ from tools.environments.local_env_policy import ( # noqa: F401 _TERMINAL_FIRST_PARTY_ENV_PREFIXES, _build_provider_env_blocklist, _buzz_terminal_context_active, _is_hermes_internal_secret, _is_terminal_first_party_env, _matches_terminal_first_party_prefix, - _plugin_terminal_env_strip_keys, -) + _plugin_terminal_env_strip_keys) from tools.environments.local_gitbash_probe import ( # noqa: F401 _BASH_EXTERNAL_PROGRAM_PROBE, _bash_probe_details_cache, _bash_starts, _bash_starts_cache, _git_bash_aslr_help, _git_root_from_bash, - _looks_like_msys_spawn_failure, _mandatory_aslr_enabled, -) + _looks_like_msys_spawn_failure, _mandatory_aslr_enabled) from tools.environments.local_pythonpath import ( # noqa: F401 _build_hermes_repo_root_aliases, _get_hermes_site_packages, _same_path, _strip_hermes_owned_pythonpath, _strip_hermes_owned_pythonpath_and_runtime_markers, - _validated_runtime_venv, -) + _validated_runtime_venv) _IS_WINDOWS = platform.system() == "Windows" logger = logging.getLogger(__name__) # --- Terminal temp-cache pruning --------------------------------------------- -# get_temp_dir() defaults to HERMES_HOME/cache/terminal (real storage, not tmpfs -# /tmp), so stale artifacts no longer vanish on reboot for free: the gateway -# housekeeping loop prunes hourly and a once-per-process sweep covers CLI-only -# installs. +# get_temp_dir() defaults to HERMES_HOME/cache/terminal (real storage, not tmpfs), +# so stale artifacts don't vanish on reboot: the gateway housekeeping loop prunes +# hourly and a once-per-process sweep covers CLI-only installs. TERMINAL_TEMP_MAX_AGE_HOURS = 72 _terminal_temp_prune_lock = threading.Lock() _terminal_temp_pruned_once = False -# Background-process artifacts come in triplets (hermes_bg_.log/.pid/.exit). -# A live server's .pid never changes mtime while its .log does, so age is judged -# per GROUP (newest mtime sharing a stem) to avoid yanking pid/exit files from -# under a still-running background session. +# Background artifacts come in triplets (hermes_bg_.log/.pid/.exit). A live +# server's .pid never changes mtime while its .log does, so age is judged per +# GROUP (newest mtime sharing a stem) to keep pid/exit files of live sessions. _BG_GROUP_RE = re.compile(r"^(hermes_bg_[A-Za-z0-9_-]+)\.(log|pid|exit)$") @@ -64,17 +59,15 @@ def _default_terminal_temp_dir() -> "Path | None": """Return HERMES_HOME/cache/terminal, or None if unresolvable.""" try: from hermes_constants import get_hermes_home - return get_hermes_home() / "cache" / "terminal" + cache_dir = get_hermes_home() / "cache" / "terminal" except Exception: return None + return cache_dir -def cleanup_terminal_temp_cache( - max_age_hours: int = TERMINAL_TEMP_MAX_AGE_HOURS, -) -> int: - """Delete session temp artifacts older than *max_age_hours*; return count - (``cleanup_*_cache`` contract). Only the managed default dir is pruned — - never a user-pointed ``terminal.temp_dir`` we don't own.""" +def cleanup_terminal_temp_cache(max_age_hours: int = TERMINAL_TEMP_MAX_AGE_HOURS) -> int: + """Delete session temp artifacts older than *max_age_hours*; return count. + Only the managed default dir is pruned — never a user-pointed ``terminal.temp_dir``.""" root = _default_terminal_temp_dir() if root is None: return 0 @@ -101,10 +94,7 @@ def cleanup_terminal_temp_cache( if (group_newest[m.group(1)] if m else mt) >= cutoff: continue try: - if f.is_dir(): - shutil.rmtree(f, ignore_errors=True) - else: - f.unlink() + shutil.rmtree(f, ignore_errors=True) if f.is_dir() else f.unlink() removed += 1 except OSError: continue @@ -128,25 +118,22 @@ def _prune_terminal_temp_once() -> None: def _msys_to_windows_path(cwd: str) -> str: - """Translate a Git Bash / MSYS path (``/c/Users/x``, ``/cygdrive/c/..``, - ``/mnt/c/..``) to native ``C:\\Users\\x`` so ``isdir``/``Popen(cwd=)`` find - it. No-op off Windows, for empty input and for multi-segment POSIX paths - like ``/home/x``; idempotent on native paths.""" + """``/c/Users/x`` / ``/cygdrive/c/..`` / ``/mnt/c/..`` -> native ``C:\\Users\\x`` so + ``isdir``/``Popen(cwd=)`` find it. No-op off Windows, for empty input and for + multi-segment POSIX paths like ``/home/x``; idempotent on native paths.""" if not _IS_WINDOWS or not cwd: return cwd m = re.match(r'^/(?:(?:cygdrive|mnt)/)?([a-zA-Z])(/.*)?$', cwd) if not m: return cwd - drive = m.group(1).upper() tail = (m.group(2) or "").replace('/', '\\') - return f"{drive}:{tail or chr(92)}" # chr(92) = backslash, avoid raw-string escape + return f"{m.group(1).upper()}:{tail or chr(92)}" # chr(92) = backslash def _resolve_local_initial_cwd(cwd: str) -> str: - """Resolve the initial cwd to an absolute host path. A relative - ``TERMINAL_CWD`` like ``hermes-agent`` naming the launch directory would - otherwise make the wrapper ``cd hermes-agent`` *inside* the project; anchor - it once so ``Popen(cwd=)`` and the in-shell ``cd`` agree.""" + """Resolve the initial cwd to an absolute host path. A relative ``TERMINAL_CWD`` + naming the launch directory would otherwise make the wrapper ``cd`` *inside* + the project; anchor it once so ``Popen(cwd=)`` and the in-shell ``cd`` agree.""" expanded = os.path.expanduser(cwd) if cwd else os.getcwd() if _IS_WINDOWS: expanded = _msys_to_windows_path(expanded) @@ -159,7 +146,6 @@ def _resolve_local_initial_cwd(cwd: str) -> str: candidate = os.path.abspath(expanded) current = os.getcwd() - # Relative name matching the tail of the current dir: use the current dir. if not os.path.isdir(candidate): wanted, have = Path(expanded).parts, Path(current).parts @@ -169,23 +155,21 @@ def _resolve_local_initial_cwd(cwd: str) -> str: def _windows_to_msys_path(cwd: str) -> str: - """Translate native ``C:\\Users\\x`` to Git Bash form ``/c/Users/x`` so - ``builtin cd`` resolves it. No-op off Windows / for non-drive paths.""" + """Native ``C:\\Users\\x`` -> Git Bash ``/c/Users/x`` so ``builtin cd`` resolves + it. No-op off Windows / for non-drive paths.""" if not _IS_WINDOWS or not cwd: return cwd m = re.match(r'^([a-zA-Z]):[\\/]*(.*)$', cwd) if not m: return cwd - drive = m.group(1).lower() tail = (m.group(2) or "").replace('\\', '/').lstrip('/') - return f"/{drive}/{tail}" if tail else f"/{drive}/" + return f"/{m.group(1).lower()}/{tail}" def _bash_safe_path(path: str) -> str: - """Return *path* in a form safe to embed in a Git Bash script: native - ``C:\\Users\\x`` / ``C:/Users/x`` become ``/c/Users/x`` (MSYS argument - conversion mangles ``C:/`` forms), and leftover backslashes are normalized - so bash does not eat ``\\U``. No-op off Windows and for empty input.""" + """*path* safe to embed in a Git Bash script: ``C:\\Users\\x`` / ``C:/Users/x`` + become ``/c/Users/x`` (MSYS argument conversion mangles ``C:/`` forms) and + leftover backslashes are normalized so bash does not eat ``\\U``. No-op off Windows.""" if not _IS_WINDOWS or not path: return path return _windows_to_msys_path(path).replace("\\", "/") @@ -194,24 +178,20 @@ def _bash_safe_path(path: str) -> str: def _quote_bash_path(path: str) -> str: """Quote *path* for safe interpolation into a Git Bash script on Windows.""" import shlex - return shlex.quote(_bash_safe_path(path)) def _cwd_usable(path: str) -> bool: - """True when *path* is a directory this process can actually chdir into. - ``isdir`` alone is not enough: stat() on ``/root`` succeeds for a non-root - user but ``Popen(cwd='/root')`` dies with PermissionError (a root-launched - CLI leaking ``/root`` into a non-root gateway's shared state).""" + """True when *path* is a directory this process can actually chdir into + (``isdir`` alone passes ``/root`` for a non-root user; ``Popen(cwd=)`` then dies).""" return os.path.isdir(path) and os.access(path, os.X_OK) def _resolve_safe_cwd(cwd: str) -> str: - """Return ``cwd`` if enterable, else the nearest usable ancestor, else - ``tempfile.gettempdir()``. MSYS paths are normalized first on Windows so a - valid ``pwd -P`` result is not rejected as missing. Lets ``_run_bash`` - recover from a deleted/inaccessible cwd instead of ``Popen`` raising before - bash starts and wedging every subsequent terminal call.""" + """``cwd`` if enterable, else the nearest usable ancestor, else + ``tempfile.gettempdir()``. MSYS paths are normalized first on Windows so a valid + ``pwd -P`` result is not rejected. Lets ``_run_bash`` recover from a deleted or + inaccessible cwd instead of ``Popen`` raising and wedging every later call.""" cwd = _msys_to_windows_path(cwd) if _IS_WINDOWS else cwd if cwd and _cwd_usable(cwd): return cwd @@ -222,8 +202,7 @@ def _resolve_safe_cwd(cwd: str) -> str: "directory. If this is a gateway/cron process, check for " "root-owned paths leaking into terminal.cwd / TERMINAL_CWD " "(#65583).", - cwd, getattr(os, "getuid", lambda: "?")(), - ) + cwd, getattr(os, "getuid", lambda: "?")()) parent = os.path.dirname(cwd) if cwd else "" while parent: if _cwd_usable(parent): @@ -238,21 +217,15 @@ def _resolve_safe_cwd(cwd: str) -> str: # --- Child-process environment construction --- -def _inject_context_hermes_home(env: dict) -> None: - """Bridge the context-local Hermes home override into subprocess env.""" +def _apply_profile_home(env: dict) -> None: + """Bridge the context-local HERMES_HOME override, then the subprocess HOME contract.""" try: from hermes_constants import get_hermes_home_override - value = get_hermes_home_override() if value: env["HERMES_HOME"] = value except Exception: pass - - -def _apply_profile_home(env: dict) -> None: - """HERMES_HOME override bridge + the subprocess HOME contract.""" - _inject_context_hermes_home(env) from hermes_constants import apply_subprocess_home_env apply_subprocess_home_env(env) @@ -260,11 +233,11 @@ def _apply_profile_home(env: dict) -> None: def _inject_session_context_env(env: dict) -> None: """Bridge gateway session ContextVars (HERMES_SESSION_*) into a child env. - Cross-session leak guard: the vars also have a last-writer-wins ``os.environ`` - mirror that, under a concurrent multi-session host, may belong to another - turn. Once the session-context system is engaged, ContextVars are - authoritative: a bound value (incl. "") wins and an _UNSET var is STRIPPED - rather than inherited. A CLI that never engaged it keeps the inherited value. + Cross-session leak guard: the vars' last-writer-wins ``os.environ`` mirror may + belong to another turn under a concurrent multi-session host. Once the + session-context system is engaged, ContextVars are authoritative: a bound value + (incl. "") wins and an _UNSET var is STRIPPED rather than inherited. A CLI that + never engaged it keeps the inherited value. """ try: from gateway.session_context import _UNSET, _VAR_MAP, session_context_engaged @@ -284,7 +257,6 @@ def _scrub_delegated_child_kanban_env(env: dict[str, str]) -> dict[str, str]: """Strip dispatcher-owned Kanban env from delegate_task child subprocesses.""" try: from agent.delegation_context import is_delegated_child_process_context, scrub_kanban_env - if is_delegated_child_process_context(): return scrub_kanban_env(env) except Exception: @@ -292,24 +264,9 @@ def _scrub_delegated_child_kanban_env(env: dict[str, str]) -> dict[str, str]: return env -def _passthrough_hooks(): - """Return ``(is_passthrough, resolve_passthrough_value)`` from the - env_passthrough skill registry, or inert fallbacks.""" - try: - from tools.env_passthrough import is_env_passthrough, resolve_passthrough_value - - return is_env_passthrough, resolve_passthrough_value - except Exception: - return (lambda _: False), (lambda _name, fallback: fallback) - - def _filter_secret_env( - items: Mapping[str, str], - out: dict, - *, - unwrap_force: bool, - plugin_strip: frozenset = frozenset(), -) -> None: + items: Mapping[str, str], out: dict, *, unwrap_force: bool, + plugin_strip: frozenset = frozenset()) -> None: """Copy *items* into *out*, dropping Hermes-managed secrets. ``_HERMES_FORCE_`` unwraps to ``NAME`` when ``unwrap_force`` (caller @@ -318,33 +275,34 @@ def _filter_secret_env( vars; the latter are used directly, never scope-resolved (UnscopedSecretError under multiplex) — only passthrough names resolve through the secret scope. """ - is_passthrough, resolve_passthrough_value = _passthrough_hooks() + try: + from tools.env_passthrough import is_env_passthrough, resolve_passthrough_value + except Exception: + is_env_passthrough, resolve_passthrough_value = (lambda _: False), (lambda _n, fb: fb) for key, value in items.items(): if key.startswith(_HERMES_PROVIDER_ENV_FORCE_PREFIX): if not unwrap_force: continue - real_key = key[len(_HERMES_PROVIDER_ENV_FORCE_PREFIX):] - if _is_hermes_internal_secret(real_key): - continue - out[real_key] = value + key = key[len(_HERMES_PROVIDER_ENV_FORCE_PREFIX):] + if not _is_hermes_internal_secret(key): + out[key] = value continue if _is_hermes_internal_secret(key) or key in plugin_strip: continue first_party = _is_terminal_first_party_env(key) - passthrough = is_passthrough(key) + passthrough = is_env_passthrough(key) if key in _HERMES_PROVIDER_ENV_BLOCKLIST and not (passthrough or first_party): continue - resolved = value if passthrough and not first_party: - resolved = resolve_passthrough_value(key, value) - if resolved is not None: - out[key] = resolved + value = resolve_passthrough_value(key, value) + if value is not None: + out[key] = value def _finalize_child_env(env: dict) -> dict: - """Guards shared by every spawn surface: profile-home propagation, - session-context bridging, Hermes-owned PYTHONPATH + venv-marker strip, MSYS - defaults, delegate_task Kanban scrub. Returns the (possibly new) dict.""" + """Guards shared by every spawn surface: profile-home propagation, session-context + bridging, Hermes-owned PYTHONPATH + venv-marker strip, MSYS defaults, delegate_task + Kanban scrub. Returns the (possibly new) dict.""" _apply_profile_home(env) _inject_session_context_env(env) _strip_hermes_owned_pythonpath_and_runtime_markers(env) @@ -353,79 +311,53 @@ def _finalize_child_env(env: dict) -> dict: def _sanitize_subprocess_env(base_env: dict | None, extra_env: dict | None = None) -> dict: - """Filter Hermes-managed secrets from a subprocess environment. - - Background/PTY spawn path (``process_registry.spawn_local``), search workers, - the computer-use driver, and user-script runners build their env here. - """ + """Filter Hermes-managed secrets from a subprocess environment (background/PTY + spawn path, search workers, computer-use driver, user-script runners).""" plugin_strip = _plugin_terminal_env_strip_keys() sanitized: dict[str, str] = {} _filter_secret_env(base_env or {}, sanitized, unwrap_force=False, plugin_strip=plugin_strip) _filter_secret_env(extra_env or {}, sanitized, unwrap_force=True, plugin_strip=plugin_strip) - # Keep bare ``hermes`` resolvable for children even when the gateway was - # launched by a service manager or cron without the console-script dir on - # PATH (cron scripts use this sanitizer directly). + # launched by a service manager or cron without the console-script dir on PATH. path_key = _path_env_key(sanitized) if path_key is not None: sanitized[path_key] = _prepend_hermes_bin_dir(sanitized.get(path_key, "")) - return _finalize_child_env(sanitized) def hermes_subprocess_env(*, inherit_credentials: bool = False) -> dict[str, str]: """Sanitized env for the **non-terminal** spawn surface (browser, ACP/CLI executors, computer-use driver, TUI Node host). Tier 1 (``_ALWAYS_STRIP_KEYS``, - plugin keys, force-prefixed hints, dynamic internal secrets) is always - removed; Tier 2 (the provider/tool blocklist) unless ``inherit_credentials`` - — pass that **only** for children that legitimately need LLM credentials - (user-blessed claude/codex/gemini CLI, TUI Node host); it is grep-able for - audit. Terminal/execute_code use the skill-aware ``_sanitize_subprocess_env``. - """ + plugin keys, force-prefixed hints, dynamic internal secrets) is always removed; + Tier 2 (the provider/tool blocklist) unless ``inherit_credentials`` — pass that + **only** for children that legitimately need LLM credentials (user-blessed + claude/codex/gemini CLI, TUI Node host); it is grep-able for audit. + Terminal/execute_code use the skill-aware ``_sanitize_subprocess_env``.""" env = os.environ.copy() - - for key in _ALWAYS_STRIP_KEYS: - env.pop(key, None) - for key in _plugin_terminal_env_strip_keys(): - env.pop(key, None) - for key in list(env): - if key.startswith(_HERMES_PROVIDER_ENV_FORCE_PREFIX) or _is_hermes_internal_secret(key): - env.pop(key, None) - + strip = _ALWAYS_STRIP_KEYS | _plugin_terminal_env_strip_keys() if not inherit_credentials: - for key in _HERMES_PROVIDER_ENV_BLOCKLIST: - env.pop(key, None) - - # Windows UTF-8 safety for spawned processes. - env.setdefault("PYTHONUTF8", "1") - + strip |= _HERMES_PROVIDER_ENV_BLOCKLIST + for key in list(env): + if (key in strip or key.startswith(_HERMES_PROVIDER_ENV_FORCE_PREFIX) + or _is_hermes_internal_secret(key)): + del env[key] + env.setdefault("PYTHONUTF8", "1") # Windows UTF-8 safety for spawned processes return _finalize_child_env(env) def build_subprocess_env( - base: "Mapping[str, str] | None" = None, - *, - inherit_profile_home: bool = True, - scrub_secrets: bool = True, - extra: "Mapping[str, str] | None" = None, -) -> dict[str, str]: - """Single factory for child-process environments, so profile-home - propagation and the secret-scrub policy have one owner. - - ``base=None`` snapshots ``os.environ``. ``scrub_secrets=True`` delegates to - :func:`_sanitize_subprocess_env` (``extra`` -> ``extra_env``; profile home is - inherent, ``inherit_profile_home`` ignored). ``scrub_secrets=False`` keeps the - base byte-for-byte (git credential flows, ``bws``/``op``); then - ``inherit_profile_home`` bridges HERMES_HOME + HOME and ``extra`` is applied - last so caller overrides win. - """ - if scrub_secrets: - return _sanitize_subprocess_env( - dict(base) if base is not None else os.environ.copy(), - dict(extra) if extra else None, - ) - + base: "Mapping[str, str] | None" = None, *, inherit_profile_home: bool = True, + scrub_secrets: bool = True, extra: "Mapping[str, str] | None" = None) -> dict[str, str]: + """Single factory for child-process envs (one owner for profile-home propagation + and the secret-scrub policy). ``base=None`` snapshots ``os.environ``. + ``scrub_secrets=True`` delegates to :func:`_sanitize_subprocess_env` (``extra`` + -> ``extra_env``; ``inherit_profile_home`` ignored, home is inherent). + ``scrub_secrets=False`` keeps the base byte-for-byte (git credential flows, + ``bws``/``op``); ``inherit_profile_home`` bridges HERMES_HOME + HOME and + ``extra`` is applied last so caller overrides win.""" env: dict[str, str] = dict(base) if base is not None else os.environ.copy() + if scrub_secrets: + return _sanitize_subprocess_env(env, dict(extra) if extra else None) if inherit_profile_home: _apply_profile_home(env) if extra: @@ -437,29 +369,23 @@ def build_subprocess_env( def _windows_bash_candidates(custom: "str | None") -> list[str]: - """Ordered bash.exe candidates on Windows: HERMES_GIT_BASH_PATH, our - portable Git under %LOCALAPPDATA%\\hermes\\git (PortableGit ``bin`` and - MinGit ``usr\\bin`` layouts), known Git-for-Windows dirs, then PATH last — - ``shutil.which`` may return WSL's bash, which fails silently on Windows paths.""" - candidates: list[str] = [] - - def add(candidate: str) -> None: - if candidate and os.path.isfile(candidate) and candidate not in candidates: - candidates.append(candidate) - - add(custom or "") - + """Ordered bash.exe candidates on Windows: HERMES_GIT_BASH_PATH, our portable Git + under %LOCALAPPDATA%\\hermes\\git (PortableGit ``bin`` and MinGit ``usr\\bin``), + known Git-for-Windows dirs, then PATH last — ``shutil.which`` may return WSL's + bash, which fails silently on Windows paths.""" local_appdata = os.environ.get("LOCALAPPDATA", "") - if local_appdata: - portable = os.path.join(local_appdata, "hermes", "git") - add(os.path.join(portable, "bin", "bash.exe")) - add(os.path.join(portable, "usr", "bin", "bash.exe")) - - add(os.path.join(os.environ.get("ProgramFiles", r"C:\Program Files"), "Git", "bin", "bash.exe")) - add(os.path.join(os.environ.get("ProgramFiles(x86)", r"C:\Program Files (x86)"), "Git", "bin", "bash.exe")) - if local_appdata: - add(os.path.join(local_appdata, "Programs", "Git", "bin", "bash.exe")) - + portable = os.path.join(local_appdata, "hermes", "git") + raw = [ + custom or "", + os.path.join(portable, "bin", "bash.exe") if local_appdata else "", + os.path.join(portable, "usr", "bin", "bash.exe") if local_appdata else "", + os.path.join(os.environ.get("ProgramFiles", r"C:\Program Files"), "Git", "bin", "bash.exe"), + os.path.join(os.environ.get("ProgramFiles(x86)", r"C:\Program Files (x86)"), "Git", "bin", "bash.exe"), + os.path.join(local_appdata, "Programs", "Git", "bin", "bash.exe") if local_appdata else ""] + candidates: list[str] = [] + for c in raw: + if c and os.path.isfile(c) and c not in candidates: + candidates.append(c) found = shutil.which("bash") if found and found not in candidates: candidates.append(found) @@ -474,26 +400,22 @@ def _find_bash() -> str: or ("/usr/bin/bash" if os.path.isfile("/usr/bin/bash") else None) or ("/bin/bash" if os.path.isfile("/bin/bash") else None) or os.environ.get("SHELL") - or "/bin/sh" - ) + or "/bin/sh") custom = os.environ.get("HERMES_GIT_BASH_PATH") candidates = _windows_bash_candidates(custom) - # First candidate that can actually start wins: a stale HERMES_GIT_BASH_PATH # pointing at a broken install must not beat a healthy portable Git. for candidate in candidates: if _bash_starts(candidate): if candidate != custom and custom and os.path.isfile(custom): logger.warning( - "HERMES_GIT_BASH_PATH=%s fails to start; using %s instead", custom, candidate, - ) + "HERMES_GIT_BASH_PATH=%s fails to start; using %s instead", custom, candidate) return candidate if candidates: probe_details = "\n".join( - detail for c in candidates if (detail := _bash_probe_details_cache.get(c)) - ) + detail for c in candidates if (detail := _bash_probe_details_cache.get(c))) if _mandatory_aslr_enabled() is True or _looks_like_msys_spawn_failure(probe_details): raise RuntimeError(_git_bash_aslr_help(candidates[0], probe_details)) # Unknown failure class: return the first path so the caller sees the @@ -503,19 +425,17 @@ def _find_bash() -> str: raise RuntimeError( "Git Bash not found. Hermes Agent requires Git for Windows on Windows.\n" "Install it from: https://git-scm.com/download/win\n" - "Or set HERMES_GIT_BASH_PATH to your bash.exe location." - ) + "Or set HERMES_GIT_BASH_PATH to your bash.exe location.") _git_bash_bin_dirs_cache: "list[str] | None" = None def _git_bash_bin_dirs() -> list[str]: - """Git Bash's coreutils dirs in ``/etc/profile`` order (mingw first so - coreutils beat System32 lookalikes); ``[]`` off Windows. A non-login - ``bash -c`` (fallback when ``bash -l`` is broken) never sources - ``/etc/profile``, so without these ``cat``/``mktemp``/``mv`` are missing: - ``write_file`` fails with an empty error and commands exit 127.""" + """Git Bash's coreutils dirs in ``/etc/profile`` order (mingw first so coreutils + beat System32 lookalikes); ``[]`` off Windows. A non-login ``bash -c`` (fallback + when ``bash -l`` is broken) never sources ``/etc/profile``, so without these + ``cat``/``mktemp``/``mv`` are missing and commands exit 127.""" global _git_bash_bin_dirs_cache if _git_bash_bin_dirs_cache is None: _git_bash_bin_dirs_cache = _compute_git_bash_bin_dirs() if _IS_WINDOWS else [] @@ -539,57 +459,45 @@ def _compute_git_bash_bin_dirs() -> list[str]: def _prepend_missing_path_entries(existing_path: str, dirs: list[str]) -> str: - """Prepend *dirs* missing from *existing_path* (``os.pathsep``); an - already-listed dir keeps its position and the input is returned unchanged - when nothing is missing.""" + """Prepend *dirs* missing from *existing_path* (``os.pathsep``); an already-listed + dir keeps its position; unchanged input when nothing is missing.""" if not dirs: return existing_path - sep = os.pathsep - entries = [e for e in existing_path.split(sep) if e] if existing_path else [] + entries = [e for e in existing_path.split(os.pathsep) if e] if existing_path else [] missing = [d for d in dirs if d not in entries] if not missing: return existing_path - return sep.join([*missing, *entries]) + return os.pathsep.join([*missing, *entries]) def _prepend_git_bash_dirs(existing_path: str) -> str: """Prepend Git Bash's binary dirs if missing (no-op off Windows), so the - non-login ``bash -c`` fallback can find coreutils when no login snapshot - re-exports the full PATH inside the shell.""" + non-login ``bash -c`` fallback can find coreutils.""" return _prepend_missing_path_entries(existing_path, _git_bash_bin_dirs()) -# POSIX-sh-family shells that understand spawn_local's ``[shell, "-lic", -# "set +m; …"]`` invocation. fish, csh/tcsh, nushell, elvish, xonsh would error -# on that syntax, so _find_shell falls back to bash for them. +# POSIX-sh-family shells that understand spawn_local's ``[shell, "-lic", "set +m; …"]`` +# invocation; fish, csh/tcsh, nushell, elvish, xonsh would error, so _find_shell +# falls back to bash for them. _SPAWN_COMPATIBLE_SHELLS = frozenset({"bash", "zsh", "sh", "dash", "ksh", "mksh"}) def _find_shell() -> str: - """User's login shell for background spawning: ``$SHELL`` on POSIX when it - is an executable sh-family shell, else ``_find_bash``. macOS's system bash - 3.2 under ``-l`` with stdin ``/dev/null`` sources ``~/.bash_profile``, which - often ``exec /bin/zsh -l`` and drops ``-c`` — the command silently never - runs. Non-allowlisted shells would trade that for a parse error.""" - if not _IS_WINDOWS: - user_shell = os.environ.get("SHELL") - if ( - user_shell - and os.path.isfile(user_shell) - and os.access(user_shell, os.X_OK) - and Path(user_shell).name in _SPAWN_COMPATIBLE_SHELLS - ): - return user_shell + """User's login shell for background spawning: ``$SHELL`` on POSIX when it is an + executable sh-family shell, else ``_find_bash``. macOS's system bash 3.2 under + ``-l`` with stdin ``/dev/null`` sources ``~/.bash_profile``, which often + ``exec /bin/zsh -l`` and drops ``-c`` — the command silently never runs.""" + user_shell = "" if _IS_WINDOWS else os.environ.get("SHELL") + if (user_shell and os.path.isfile(user_shell) and os.access(user_shell, os.X_OK) + and Path(user_shell).name in _SPAWN_COMPATIBLE_SHELLS): + return user_shell return _find_bash() # --- PATH completion for the terminal subshell --- # Standard PATH entries for environments with minimal PATH. -_SANE_PATH = ( - "/opt/homebrew/bin:/opt/homebrew/sbin:" - "/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin" -) +_SANE_PATH = "/opt/homebrew/bin:/opt/homebrew/sbin:/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin" # Cached directory containing the ``hermes`` console-script. # ``_SENTINEL`` distinguishes "not resolved yet" from a resolved ``None``. @@ -598,58 +506,46 @@ _HERMES_BIN_DIR: "str | None | object" = _SENTINEL def _resolve_hermes_bin_dir() -> str | None: - """Directory holding the ``hermes`` console-script, or None (cached). - Launched by systemd/cron/a desktop launcher, the gateway's PATH lacks the - install dir (``~/.local/bin``, venv ``bin``, pipx, nix) and bare ``hermes`` - exits 127. Order: ``which``; absolute ``sys.argv[0]`` naming a real hermes - executable; ``sys.executable``'s dir if it holds the shim.""" + """Directory holding the ``hermes`` console-script, or None (cached). Launched + by systemd/cron/a desktop launcher, the gateway's PATH lacks the install dir + (``~/.local/bin``, venv ``bin``, pipx, nix) and bare ``hermes`` exits 127. + Order: ``which``; absolute ``sys.argv[0]`` naming a real hermes executable; + ``sys.executable``'s dir if it holds the shim.""" global _HERMES_BIN_DIR if _HERMES_BIN_DIR is not _SENTINEL: return _HERMES_BIN_DIR # type: ignore[return-value] candidate: str | None = None - which = shutil.which("hermes") + argv0 = sys.argv[0] if sys.argv else "" + base = os.path.basename(argv0).lower() + exe_dir = os.path.dirname(sys.executable) if sys.executable else "" + shim = "hermes.exe" if _IS_WINDOWS else "hermes" if which: candidate = os.path.dirname(which) - - if candidate is None: - argv0 = sys.argv[0] if sys.argv else "" - base = os.path.basename(argv0).lower() - if ( - os.path.isabs(argv0) - and (base == "hermes" or base.startswith("hermes.")) - and os.path.isfile(argv0) - ): - candidate = os.path.dirname(argv0) - - if candidate is None: - exe_dir = os.path.dirname(sys.executable) if sys.executable else "" - shim = "hermes.exe" if _IS_WINDOWS else "hermes" - if exe_dir and os.path.isfile(os.path.join(exe_dir, shim)): - candidate = exe_dir - + elif (os.path.isabs(argv0) and (base == "hermes" or base.startswith("hermes.")) + and os.path.isfile(argv0)): + candidate = os.path.dirname(argv0) + elif exe_dir and os.path.isfile(os.path.join(exe_dir, shim)): + candidate = exe_dir if candidate and not os.path.isdir(candidate): candidate = None - _HERMES_BIN_DIR = candidate return candidate def _prepend_hermes_bin_dir(existing_path: str) -> str: - """Prepend the hermes install dir to ``existing_path`` if it's missing - (unchanged when already present or unresolvable).""" + """Prepend the hermes install dir to ``existing_path`` if missing.""" bin_dir = _resolve_hermes_bin_dir() return _prepend_missing_path_entries(existing_path, [bin_dir] if bin_dir else []) def _managed_runtime_path_entries() -> list[str]: - """Existing Hermes-managed runtime dirs: ``$HERMES_HOME/node`` (+``/bin``) - and ``$HERMES_HOME/bin`` (managed ``uv``). Per call, not cached: home is + """Existing Hermes-managed runtime dirs: ``$HERMES_HOME/node`` (+``/bin``) and + ``$HERMES_HOME/bin`` (managed ``uv``). Per call, not cached: home is profile-scoped and a managed tree can appear mid-process.""" try: from hermes_constants import get_hermes_home, iter_hermes_node_dirs - candidates = [*iter_hermes_node_dirs(), get_hermes_home() / "bin"] return [str(d) for d in candidates if d.is_dir()] except Exception: @@ -663,11 +559,9 @@ def _append_missing_sane_path_entries(existing_path: str) -> str: precedence. Windows is a no-op passthrough (native ``;`` PATH untouched).""" if _IS_WINDOWS: return existing_path - sane_entries = [entry for entry in _SANE_PATH.split(":") if entry] sane_entries.extend( - entry for entry in _managed_runtime_path_entries() if entry not in sane_entries - ) + entry for entry in _managed_runtime_path_entries() if entry not in sane_entries) if not existing_path: return ":".join(sane_entries) # dict preserves first-occurrence order; empty entries dropped. @@ -677,10 +571,9 @@ def _append_missing_sane_path_entries(existing_path: str) -> str: def _apply_windows_msys_bash_env_defaults(env: dict) -> None: - """Disable MSYS argument path conversion (``/FO`` -> ``C:/.../git/FO`` - breaks tasklist/schtasks/wmic/``cmd /c``). Git for Windows honors - ``MSYS_NO_PATHCONV``; MSYS2/Cygwin bash honor ``MSYS2_ARG_CONV_EXCL`` — set - both. Users can override in their env.""" + """Disable MSYS argument path conversion (``/FO`` -> ``C:/.../git/FO`` breaks + tasklist/schtasks/wmic/``cmd /c``). Git for Windows honors ``MSYS_NO_PATHCONV``; + MSYS2/Cygwin bash honor ``MSYS2_ARG_CONV_EXCL`` — set both; users can override.""" if not _IS_WINDOWS: return env.setdefault("MSYS_NO_PATHCONV", "1") @@ -688,14 +581,11 @@ def _apply_windows_msys_bash_env_defaults(env: dict) -> None: def _path_env_key(run_env: dict) -> str | None: - """Return the PATH env key to update without altering Windows casing - (``Path`` vs ``PATH``); None when a Windows env has no PATH key at all.""" + """PATH env key to update without altering Windows casing (``Path`` vs ``PATH``); + None when a Windows env has no PATH key at all.""" if not _IS_WINDOWS: return "PATH" - for key in run_env: - if key.upper() == "PATH": - return key - return None + return next((key for key in run_env if key.upper() == "PATH"), None) def _make_run_env(env: dict) -> dict: @@ -705,35 +595,26 @@ def _make_run_env(env: dict) -> dict: path_key = _path_env_key(run_env) if path_key is not None: new_path = _append_missing_sane_path_entries(run_env.get(path_key, "")) - new_path = _prepend_git_bash_dirs(new_path) - run_env[path_key] = _prepend_hermes_bin_dir(new_path) + run_env[path_key] = _prepend_hermes_bin_dir(_prepend_git_bash_dirs(new_path)) return _finalize_child_env(run_env) # --- Hermes venv / repo-root detection (module-level, computed once) --- # Owned here; read lazily by tools.environments.local_pythonpath. -#: The Hermes repository root (three levels up from this file). The Electron -#: app prepends it to PYTHONPATH so the backend can ``import tools``; other -#: subprocesses don't need it and it can shadow local packages. +#: Repo root (three levels up). The Electron app prepends it to PYTHONPATH so the +#: backend can ``import tools``; other subprocesses must not inherit it. _hermes_repo_root: Path = Path(__file__).resolve().parents[2] -#: Alternate repo-root spellings Hermes launchers may emit. ``resolve()`` -#: canonicalizes junctions, but the Windows gateway launcher renders -#: Hermes-owned paths under the configured HERMES_HOME spelling (possibly a -#: junction to another drive); the unresolved ``Path(__file__)`` keeps it. +#: Alternate repo-root spellings launchers may emit: ``resolve()`` canonicalizes +#: junctions, but the Windows gateway launcher renders Hermes-owned paths under the +#: configured HERMES_HOME spelling (possibly a junction to another drive). _hermes_repo_root_aliases: tuple[Path, ...] = _build_hermes_repo_root_aliases( - _hermes_repo_root, - Path(__file__).absolute().parents[2], - get_process_hermes_home(), -) + _hermes_repo_root, Path(__file__).absolute().parents[2], get_process_hermes_home()) -#: Whether the interpreter runs inside a venv (``sys.real_prefix`` is the old -#: virtualenv<20 marker). +#: Whether the interpreter runs inside a venv (``sys.real_prefix``: virtualenv<20). _in_venv: bool = ( - getattr(sys, "base_prefix", sys.prefix) != sys.prefix - or hasattr(sys, "real_prefix") -) + getattr(sys, "base_prefix", sys.prefix) != sys.prefix or hasattr(sys, "real_prefix")) #: Lazily-cached site-packages dirs of the running interpreter's own venv. _hermes_site_packages: list[Path] | None = None @@ -743,36 +624,28 @@ _hermes_site_packages: list[Path] | None = None def _read_terminal_shell_init_config() -> tuple[list[str], bool]: - """Return (shell_init_files, auto_source_bashrc) from config.yaml. - Best-effort: defaults on any failure so terminal execution never breaks.""" + """(shell_init_files, auto_source_bashrc) from config.yaml; defaults on any + failure so terminal execution never breaks.""" try: from hermes_cli.config import load_config - - cfg = load_config() or {} - terminal_cfg = cfg.get("terminal") or {} + terminal_cfg = (load_config() or {}).get("terminal") or {} files = terminal_cfg.get("shell_init_files") or [] if not isinstance(files, list): files = [] - auto_bashrc = bool(terminal_cfg.get("auto_source_bashrc", True)) - return [str(f) for f in files if f], auto_bashrc + return [str(f) for f in files if f], bool(terminal_cfg.get("auto_source_bashrc", True)) except Exception: return [], True def _resolve_shell_init_files() -> list[str]: - """Files to source before the login-shell snapshot (``~``/``${VAR}`` - expanded, missing dropped). ``auto_source_bashrc`` applies only without an - explicit list: ~/.profile and ~/.bash_profile first (no interactivity guard; - where n/nvm/asdf/pyenv add PATH), ~/.bashrc last (Debian's default returns - early when non-interactive, but guard-less bashrcs keep working).""" + """Files to source before the login-shell snapshot (``~``/``${VAR}`` expanded, + missing dropped). ``auto_source_bashrc`` applies only without an explicit list: + ~/.profile and ~/.bash_profile first (no interactivity guard; where + n/nvm/asdf/pyenv add PATH), ~/.bashrc last (Debian's returns early when + non-interactive, but guard-less bashrcs keep working).""" explicit, auto_bashrc = _read_terminal_shell_init_config() - - candidates: list[str] = [] - if explicit: - candidates.extend(explicit) - elif auto_bashrc and not _IS_WINDOWS: - candidates.extend(["~/.profile", "~/.bash_profile", "~/.bashrc"]) - + candidates = explicit or (["~/.profile", "~/.bash_profile", "~/.bashrc"] + if auto_bashrc and not _IS_WINDOWS else []) resolved: list[str] = [] for raw in candidates: try: @@ -785,18 +658,15 @@ def _resolve_shell_init_files() -> list[str]: def _prepend_shell_init(cmd_string: str, files: list[str]) -> str: - """Prepend guarded, silent ``source `` lines to a bash script: - ``set +e`` keeps going on errors, ``2>/dev/null`` hides noisy prompts, - ``|| true`` neutralises the exit status.""" + """Prepend guarded, silent ``source `` lines: ``set +e`` keeps going on + errors, ``2>/dev/null`` hides noisy prompts, ``|| true`` neutralises the status.""" if not files: return cmd_string - - prelude_parts = ["set +e"] + prelude = ["set +e"] for path in files: safe = path.replace("'", "'\\''") - prelude_parts.append(f"[ -r '{safe}' ] && . '{safe}' 2>/dev/null || true") - prelude = "\n".join(prelude_parts) + "\n" - return prelude + cmd_string + prelude.append(f"[ -r '{safe}' ] && . '{safe}' 2>/dev/null || true") + return "\n".join(prelude) + "\n" + cmd_string # --- Process-group teardown (POSIX) --- @@ -814,38 +684,34 @@ def _group_alive(pgid: int) -> bool: def _wait_for_group_exit(proc, pgid: int, timeout: float) -> bool: - """Wait until the process group is gone, reaping the wrapper as we go - (a dead but unreaped group leader still makes ``killpg(pgid, 0)`` succeed).""" + """Wait until the process group is gone, reaping the wrapper as we go (a dead + but unreaped group leader still makes ``killpg(pgid, 0)`` succeed).""" deadline = time.monotonic() + timeout - while time.monotonic() < deadline: + while True: try: proc.poll() except Exception: pass if not _group_alive(pgid): return True + if time.monotonic() >= deadline: + return False time.sleep(0.05) - try: - proc.poll() - except Exception: - pass - return not _group_alive(pgid) def _snapshot_descendants(proc) -> list: """psutil children snapshot; empty on any failure (must never break the kill).""" try: import psutil - return psutil.Process(proc.pid).children(recursive=True) except Exception: return [] def _sweep_escaped_descendants(descendants: list, pgid: int) -> None: - """SIGKILL snapshotted survivors that escaped the process group via - ``setsid`` — after TERM→KILL so in-group members keep their grace; psutil's - identity-aware Process skips recycled PIDs. POSIX-only (see _IS_WINDOWS gate).""" + """SIGKILL snapshotted survivors that escaped the process group via ``setsid`` + — after TERM→KILL so in-group members keep their grace; psutil's identity-aware + Process skips recycled PIDs. POSIX-only (see _IS_WINDOWS gate in caller).""" for child in descendants: try: if not child.is_running(): @@ -861,38 +727,28 @@ def _sweep_escaped_descendants(descendants: list, pgid: int) -> None: def _kill_process_group_posix(proc) -> None: - """TERM the group, wait, KILL, then sweep setsid escapees. POSIX-only - (_IS_WINDOWS handled by the caller). Descendants are snapshotted BEFORE the - first signal — once the wrapper dies they reparent to init — and we wait on - the group, not the wrapper, which can exit before grandchildren under load.""" + """TERM the group, wait, KILL, then sweep setsid escapees. Descendants are + snapshotted BEFORE the first signal — once the wrapper dies they reparent to + init — and we wait on the group, not the wrapper, which can exit before + grandchildren under load. POSIX-only (_IS_WINDOWS handled by the caller).""" try: pgid = os.getpgid(proc.pid) except ProcessLookupError: pgid = getattr(proc, "_hermes_pgid", None) if pgid is None: raise - descendants = _snapshot_descendants(proc) - try: - os.killpg(pgid, signal.SIGTERM) # windows-footgun: ok — POSIX only (see _IS_WINDOWS gate in caller) + # windows-footgun: ok — POSIX only (see _IS_WINDOWS gate in caller) + os.killpg(pgid, signal.SIGTERM) + if not _wait_for_group_exit(proc, pgid, 1.0): + os.killpg(pgid, signal.SIGKILL) + _wait_for_group_exit(proc, pgid, 2.0) + try: + proc.wait(timeout=0.2) + except (subprocess.TimeoutExpired, OSError): + pass except ProcessLookupError: - _sweep_escaped_descendants(descendants, pgid) - return - - if _wait_for_group_exit(proc, pgid, 1.0): - _sweep_escaped_descendants(descendants, pgid) - return - - try: - os.killpg(pgid, signal.SIGKILL) # windows-footgun: ok — POSIX only (see _IS_WINDOWS gate in caller) - except ProcessLookupError: - _sweep_escaped_descendants(descendants, pgid) - return - _wait_for_group_exit(proc, pgid, 2.0) - try: - proc.wait(timeout=0.2) - except (subprocess.TimeoutExpired, OSError): pass _sweep_escaped_descendants(descendants, pgid) @@ -900,12 +756,7 @@ def _kill_process_group_posix(proc) -> None: def _kill_process_windows(proc) -> None: try: from gateway.status import get_process_start_time, terminate_pid - - terminate_pid( - proc.pid, - force=True, - expected_start_time=get_process_start_time(proc.pid), - ) + terminate_pid(proc.pid, force=True, expected_start_time=get_process_start_time(proc.pid)) except Exception: proc.kill() try: @@ -917,9 +768,8 @@ def _kill_process_windows(proc) -> None: class LocalEnvironment(BaseEnvironment): """Run commands directly on the host machine. - Spawn-per-call: every execute() spawns a fresh bash process. - Session snapshot preserves env vars across calls. - CWD persists via file-based read after each command. + Spawn-per-call: every execute() spawns a fresh bash process. Session snapshot + preserves env vars across calls; CWD persists via the stdout marker. """ _profile_scoped_passthrough = True @@ -929,42 +779,30 @@ class LocalEnvironment(BaseEnvironment): is_local = True def _additional_profile_scoped_passthrough_names(self) -> tuple[str, ...]: - """First-party ``BUZZ_*`` names present in the env, excluded from the - shared session snapshot. env_passthrough can never list them (it refuses - blocklisted names), so under a multiplexed gateway profile A's - BUZZ_PRIVATE_KEY would land in the snapshot and be sourced by profile B. - Prefix-only and monotonic on purpose: conservative even when the - context-gated carve-out is inactive.""" + """First-party ``BUZZ_*`` names present in the env, excluded from the shared + session snapshot. env_passthrough can never list them (it refuses blocklisted + names), so under a multiplexed gateway profile A's BUZZ_PRIVATE_KEY would land + in the snapshot and be sourced by profile B. Prefix-only and monotonic on + purpose: conservative even when the context-gated carve-out is inactive.""" merged = dict(os.environ | self.env) - return tuple( - sorted( - name - for name in merged - if isinstance(name, str) and _matches_terminal_first_party_prefix(name) - ) - ) + return tuple(sorted( + name for name in merged + if isinstance(name, str) and _matches_terminal_first_party_prefix(name))) def __init__(self, cwd: str = "", timeout: int = 60, env: dict = None): - cwd = _resolve_local_initial_cwd(cwd) - super().__init__(cwd=cwd, timeout=timeout, env=env) + super().__init__(cwd=_resolve_local_initial_cwd(cwd), timeout=timeout, env=env) self.init_session() def get_temp_dir(self) -> str: - """Shell-safe writable temp dir. Precedence: ``TERMINAL_TEMP_DIR``, - POSIX TMPDIR/TMP/TEMP (Termux has no /tmp), ``HERMES_HOME/cache/terminal``, - /tmp, POSIX ``tempfile.gettempdir()``; backend env before process env so - terminal.env overrides work. The default is real storage because tmpfs - /tmp fills under Hermes load (pruned by ``cleanup_terminal_temp_cache``). - Windows: ``%TEMP%`` often has spaces that break unquoted bash, so always - the HERMES_HOME cache dir with forward slashes (valid in bash and Python). - """ + """Shell-safe writable temp dir. Precedence: ``TERMINAL_TEMP_DIR``, POSIX + TMPDIR/TMP/TEMP (Termux has no /tmp), ``HERMES_HOME/cache/terminal``, /tmp, + POSIX ``tempfile.gettempdir()``; backend env before process env so + terminal.env overrides work. The default is real storage because tmpfs /tmp + fills under Hermes load (pruned by ``cleanup_terminal_temp_cache``). + Windows: ``%TEMP%`` often has spaces that break unquoted bash, so always the + HERMES_HOME cache dir with forward slashes (valid in bash and Python).""" if _IS_WINDOWS: - # Forward slashes: one string valid in bash interpolation AND Python open(). - try: - from hermes_constants import get_hermes_home - cache_dir = get_hermes_home() / "cache" / "terminal" - except Exception: - cache_dir = Path(tempfile.gettempdir()) / "hermes_terminal" + cache_dir = _default_terminal_temp_dir() or Path(tempfile.gettempdir()) / "hermes_terminal" cache_dir.mkdir(parents=True, exist_ok=True) _prune_terminal_temp_once() return str(cache_dir).replace("\\", "/") @@ -975,15 +813,12 @@ class LocalEnvironment(BaseEnvironment): configured = self.env.get("TERMINAL_TEMP_DIR") or os.environ.get("TERMINAL_TEMP_DIR") if configured and configured.startswith("/") and os.path.isdir(configured): return _posix(configured) - for env_var in ("TMPDIR", "TMP", "TEMP"): candidate = self.env.get(env_var) or os.environ.get(env_var) if candidate and candidate.startswith("/"): return _posix(candidate) - try: - from hermes_constants import get_hermes_home - cache_dir = get_hermes_home() / "cache" / "terminal" + cache_dir = _default_terminal_temp_dir() cache_dir.mkdir(parents=True, exist_ok=True) resolved = str(cache_dir) if resolved.startswith("/") and os.access(resolved, os.W_OK | os.X_OK): @@ -991,10 +826,8 @@ class LocalEnvironment(BaseEnvironment): return _posix(resolved) except Exception: pass - if os.path.isdir("/tmp") and os.access("/tmp", os.W_OK | os.X_OK): return "/tmp" - candidate = tempfile.gettempdir() return _posix(candidate) if candidate.startswith("/") else "/tmp" @@ -1008,10 +841,10 @@ class LocalEnvironment(BaseEnvironment): return _quote_bash_path(path) def _recover_cwd(self) -> None: - """Swap ``self.cwd`` for a usable directory if it vanished or is - inaccessible (e.g. a previous command ``rm -rf``'d its own cwd) — - otherwise Popen raises before bash starts and every subsequent call - fails. A benign MSYS→Windows normalization is not warned about.""" + """Swap ``self.cwd`` for a usable directory if it vanished or is inaccessible + (e.g. a command ``rm -rf``'d its own cwd) — otherwise Popen raises before bash + starts and every subsequent call fails. A benign MSYS→Windows normalization + is not warned about.""" safe_cwd = _resolve_safe_cwd(self.cwd) if safe_cwd == self.cwd: return @@ -1020,50 +853,32 @@ class LocalEnvironment(BaseEnvironment): logger.warning( "LocalEnvironment cwd %r is missing on disk; " "falling back to %r so terminal commands keep working.", - self.cwd, - safe_cwd, - ) + self.cwd, safe_cwd) self.cwd = safe_cwd - def _run_bash(self, cmd_string: str, *, login: bool = False, - timeout: int = 120, + def _run_bash(self, cmd_string: str, *, login: bool = False, timeout: int = 120, stdin_data: str | None = None) -> subprocess.Popen: bash = _find_bash() # Login invocations (init_session's env snapshot) source the user's rc / # custom init files so nvm/asdf/pyenv land on PATH in the snapshot. if login: - init_files = _resolve_shell_init_files() - if init_files: - cmd_string = _prepend_shell_init(cmd_string, init_files) + cmd_string = _prepend_shell_init(cmd_string, _resolve_shell_init_files()) args = [bash, "-l", "-c", cmd_string] if login else [bash, "-c", cmd_string] run_env = _make_run_env(self.env) - self._recover_cwd() - _popen_kwargs = {"creationflags": windows_hide_flags()} if _IS_WINDOWS else {} - proc = subprocess.Popen( - args, - text=True, - env=run_env, - encoding="utf-8", - errors="replace", - stdout=subprocess.PIPE, - stderr=subprocess.STDOUT, + args, text=True, env=run_env, encoding="utf-8", errors="replace", + stdout=subprocess.PIPE, stderr=subprocess.STDOUT, stdin=subprocess.PIPE if stdin_data is not None else subprocess.DEVNULL, - start_new_session=True, - cwd=self.cwd, - **_popen_kwargs, - ) + start_new_session=True, cwd=self.cwd, **_popen_kwargs) if not _IS_WINDOWS: try: proc._hermes_pgid = os.getpgid(proc.pid) except ProcessLookupError: pass - if stdin_data is not None: _pipe_stdin(proc, stdin_data) - return proc def _kill_process(self, proc): @@ -1079,16 +894,11 @@ class LocalEnvironment(BaseEnvironment): except Exception: pass - def _update_cwd(self, result: dict): - """Update cwd from the stdout marker the base wrapper emits (``pwd -P``), - sharing the remote backends' parser instead of re-reading a temp file.""" - self._extract_cwd_from_output(result) - def _extract_cwd_from_output(self, result: dict): """Base semantics plus: Git Bash ``pwd -P`` emits MSYS form on Windows — normalize to native and require the dir to exist, else ``_run_bash`` would - warn every command. A stale path rolls back to the previous cwd, which - this command did not observe, so ``cwd_observed`` is dropped.""" + warn every command. A stale path rolls back to the previous cwd, which this + command did not observe, so ``cwd_observed`` is dropped.""" prev_cwd = self.cwd super()._extract_cwd_from_output(result) if self.cwd != prev_cwd: @@ -1105,7 +915,6 @@ class LocalEnvironment(BaseEnvironment): """Clean up temp files, including orphaned atomic-write snapshots (``snap.tmp.``) a failed/interrupted mv could leave behind.""" import glob - try: stale = glob.glob(f"{self._snapshot_path}.tmp.*") except Exception: diff --git a/tools/environments/local_env_policy.py b/tools/environments/local_env_policy.py index 2c91639557..5b2bea5dd7 100644 --- a/tools/environments/local_env_policy.py +++ b/tools/environments/local_env_policy.py @@ -8,13 +8,12 @@ import os # Prefix a caller uses in ``extra_env`` to force a blocklisted var through. _HERMES_PROVIDER_ENV_FORCE_PREFIX = "_HERMES_FORCE_" -# Hermes-managed AWS *inference* credentials for ``auth_type="aws_sdk"`` (Bedrock). -# Deliberately only the Bedrock bearer token — an inference secret like -# OPENAI_API_KEY that no aws/terraform/boto3 toolchain uses. The general AWS -# credential chain stays inheritable on purpose: the local terminal is the user's -# trusted operator shell (SECURITY.md §3.2), and env_passthrough can never -# re-allow a blocklisted name (GHSA-rhgp-j443-p4rf), so blocking would be -# unrecoverable for every aws/terraform user. +# Hermes-managed AWS *inference* credentials for ``auth_type="aws_sdk"`` (Bedrock): +# deliberately only the Bedrock bearer token, which no aws/terraform/boto3 toolchain +# uses. The general AWS credential chain stays inheritable on purpose — the local +# terminal is the user's trusted operator shell (SECURITY.md §3.2) and env_passthrough +# can never re-allow a blocklisted name (GHSA-rhgp-j443-p4rf), so blocking it would +# be unrecoverable for every aws/terraform user. _AWS_SDK_CREDENTIAL_ENV_VARS = frozenset({ "AWS_BEARER_TOKEN_BEDROCK", }) @@ -49,7 +48,6 @@ _STATIC_PROVIDER_ENV_BLOCKLIST = frozenset({ def _build_provider_env_blocklist() -> frozenset: """Derive the blocklist from provider, tool, and gateway config.""" blocked: set[str] = set(_STATIC_PROVIDER_ENV_BLOCKLIST) - try: from hermes_cli.auth import PROVIDER_REGISTRY for pconfig in PROVIDER_REGISTRY.values(): @@ -60,7 +58,6 @@ def _build_provider_env_blocklist() -> frozenset: blocked.add(pconfig.base_url_env_var) except ImportError: pass - try: from hermes_cli.config import OPTIONAL_ENV_VARS for name, metadata in OPTIONAL_ENV_VARS.items(): @@ -71,50 +68,44 @@ def _build_provider_env_blocklist() -> frozenset: blocked.add(name) except ImportError: pass - - # CLAUDE_CODE_OAUTH_TOKEN is owned by the user's Claude Code install, not a - # Hermes credential (subscription auth is not a Hermes provider path). - # Stripping it made agent-spawned ``claude`` CLIs fall through to the shared - # Keychain / ~/.claude credentials store and, on auth failure, wipe it — - # logging the user out. It arrives via the anthropic registry entry above. + # CLAUDE_CODE_OAUTH_TOKEN (arrives via the anthropic registry entry) is owned by + # the user's Claude Code install, not a Hermes credential. Stripping it made + # agent-spawned ``claude`` CLIs fall through to the shared Keychain / ~/.claude + # store and, on auth failure, wipe it — logging the user out. blocked.discard("CLAUDE_CODE_OAUTH_TOKEN") - # BUZZ_* is deliberately NOT discarded, even for Buzz-managed agents: this - # blocklist feeds every scrub surface (terminal, execute_code, the - # hermes_subprocess_env Tier-2 strip), so an import-time discard would leak - # BUZZ_PRIVATE_KEY into non-terminal children. The Buzz carve-out is a - # terminal-only, context-gated scrub-path exemption — see - # ``_is_terminal_first_party_env``. + # BUZZ_* is deliberately NOT discarded: this blocklist feeds every scrub surface + # (terminal, execute_code, hermes_subprocess_env Tier-2), so an import-time + # discard would leak BUZZ_PRIVATE_KEY into non-terminal children. The Buzz + # carve-out is a terminal-only, context-gated exemption + # (``_is_terminal_first_party_env``). return frozenset(blocked) _HERMES_PROVIDER_ENV_BLOCKLIST = _build_provider_env_blocklist() -# First-party platform credentials (``BUZZ_*``, driving the platform-mandated -# ``buzz`` CLI) carved out of the TERMINAL scrub only (``_make_run_env``, +# First-party platform credentials (``BUZZ_*``, driving the platform-mandated ``buzz`` +# CLI) carved out of the TERMINAL scrub only (``_make_run_env``, # ``_sanitize_subprocess_env``); execute_code, hermes_subprocess_env, docker and -# env_passthrough registration stay sealed, so GHSA-rhgp-j443-p4rf holds. -# CONTEXT-GATED (``_buzz_terminal_context_active``): a Telegram/CLI/cron session -# on a host that also runs a Buzz gateway must not get the signing key. Values -# are used directly, never scope-resolved (UnscopedSecretError under multiplex), -# and the snapshot treats them as profile-scoped so they never persist across -# profiles. Prefix-based so future BUZZ_* names need no code change. +# env_passthrough registration stay sealed (GHSA-rhgp-j443-p4rf). CONTEXT-GATED via +# ``_buzz_terminal_context_active``: a Telegram/CLI/cron session on a host that also +# runs a Buzz gateway must not get the signing key. Values are used directly, never +# scope-resolved (UnscopedSecretError under multiplex), and the snapshot treats them +# as profile-scoped. Prefix-based so future BUZZ_* names need no code change. _TERMINAL_FIRST_PARTY_ENV_PREFIXES = ("BUZZ_",) def _matches_terminal_first_party_prefix(name: str) -> bool: - """Pure name check (``BUZZ_*``), regardless of session context — the - snapshot exclusion must stay conservative even when the carve-out is inactive.""" + """Pure name check (``BUZZ_*``), regardless of session context — the snapshot + exclusion must stay conservative even when the carve-out is inactive.""" return name.startswith(_TERMINAL_FIRST_PARTY_ENV_PREFIXES) def _buzz_terminal_context_active() -> bool: - """True when this process/session operates as a Buzz agent. - - Either signal suffices: ``BUZZ_MANAGED_AGENT`` in the process env (set only - by Buzz Desktop's buzz-acp harness), or the live session's platform is - ``buzz`` via the gateway ContextVar — authoritative under a concurrent - multi-session host, so a sibling Telegram session resolves its OWN platform. - """ + """True when this process/session operates as a Buzz agent: ``BUZZ_MANAGED_AGENT`` + in the process env (set only by Buzz Desktop's buzz-acp harness), or the live + session's platform is ``buzz`` via the gateway ContextVar — authoritative under + a concurrent multi-session host, so a sibling Telegram session resolves its OWN + platform.""" if os.environ.get("BUZZ_MANAGED_AGENT"): return True try: @@ -126,26 +117,25 @@ def _buzz_terminal_context_active() -> bool: def _is_terminal_first_party_env(name: str) -> bool: - """``name`` is a first-party platform credential (``BUZZ_*``) AND the - current process/session context entitles it to reach terminal children.""" + """``name`` is a first-party platform credential (``BUZZ_*``) AND the current + process/session context entitles it to reach terminal children.""" return _matches_terminal_first_party_prefix(name) and _buzz_terminal_context_active() # Active-venv markers that must NOT leak: a leaked VIRTUAL_ENV/CONDA_PREFIX makes -# uv/poetry sync ANOTHER project's deps into the Hermes venv (clobbering it; the -# venv stays reachable via PATH so stripping is safe), and PYTHONHOME redirects -# any child interpreter's stdlib to the Hermes venv (version-mismatch crashes). -# PYTHONPATH is handled separately — only Hermes-owned entries are removed. +# uv/poetry sync ANOTHER project's deps into the Hermes venv (the venv stays +# reachable via PATH so stripping is safe), and PYTHONHOME redirects any child +# interpreter's stdlib to the Hermes venv (version-mismatch crashes). PYTHONPATH is +# handled separately — only Hermes-owned entries are removed. _ACTIVE_VENV_MARKER_VARS = ("VIRTUAL_ENV", "CONDA_PREFIX", "PYTHONHOME") def _is_hermes_internal_secret(key: str) -> bool: - """True for Hermes-internal secrets injected under *dynamic* names the - static blocklist cannot enumerate: ``AUXILIARY__API_KEY``/``_BASE_URL`` - (per-task side-LLM credentials) and ``GATEWAY_RELAY_*_SECRET``/``_KEY``/ - ``_TOKEN`` (relay auth; non-secret routing hints stay visible). Single source - of truth for every spawn path, stripped regardless of env_passthrough - registration or ``inherit_credentials``.""" + """True for Hermes-internal secrets injected under *dynamic* names the static + blocklist cannot enumerate: ``AUXILIARY__API_KEY``/``_BASE_URL`` (per-task + side-LLM credentials) and ``GATEWAY_RELAY_*_SECRET``/``_KEY``/``_TOKEN`` (relay + auth; non-secret routing hints stay visible). Stripped on every spawn path + regardless of env_passthrough registration or ``inherit_credentials``.""" upper = key.upper() if upper.startswith("AUXILIARY_") and upper.endswith(("_API_KEY", "_BASE_URL")): return True @@ -153,11 +143,9 @@ def _is_hermes_internal_secret(key: str) -> bool: def _plugin_terminal_env_strip_keys() -> frozenset: - """Credential env keys owned by plugin-registered terminal backends. - - Computed at call time because plugins register after import. Tier-1: - stripped from every spawned subprocess unconditionally. Fail-soft to empty. - """ + """Credential env keys owned by plugin-registered terminal backends (Tier-1: + stripped from every spawned subprocess). Computed at call time because plugins + register after import; fail-soft to empty.""" try: from agent.terminal_env_registry import plugin_strip_env_keys diff --git a/tools/environments/local_gitbash_probe.py b/tools/environments/local_gitbash_probe.py index a57fcb0937..5007cb1d4b 100644 --- a/tools/environments/local_gitbash_probe.py +++ b/tools/environments/local_gitbash_probe.py @@ -37,31 +37,24 @@ def _mandatory_aslr_enabled() -> "bool | None": global _mandatory_aslr_enabled_cache if _mandatory_aslr_enabled_cache is not None: return _mandatory_aslr_enabled_cache - try: - powershell = shutil.which("powershell.exe") or "powershell.exe" result = subprocess.run( [ - powershell, - "-NoProfile", - "-NonInteractive", - "-Command", + shutil.which("powershell.exe") or "powershell.exe", + "-NoProfile", "-NonInteractive", "-Command", "(Get-ProcessMitigation -System).Aslr.ForceRelocateImages.ToString()", ], - capture_output=True, - text=True, encoding="utf-8", errors="replace", - timeout=10, - creationflags=windows_hide_flags(), + capture_output=True, text=True, encoding="utf-8", errors="replace", + timeout=10, creationflags=windows_hide_flags(), ) if result.returncode != 0: return None value = (result.stdout or "").strip().upper() if value == "ON": _mandatory_aslr_enabled_cache = True - return True - if value in {"OFF", "NOTSET"}: + elif value in {"OFF", "NOTSET"}: _mandatory_aslr_enabled_cache = False - return False + return _mandatory_aslr_enabled_cache except Exception as exc: logger.debug("Could not query Windows Mandatory ASLR state: %s", exc) return None @@ -73,15 +66,12 @@ def _git_root_from_bash(bash: str) -> str: if ntpath.basename(bin_dir).lower() != "bin": return ntpath.dirname(bin_dir) parent = ntpath.dirname(bin_dir) - if ntpath.basename(parent).lower() == "usr": - return ntpath.dirname(parent) - return parent + return ntpath.dirname(parent) if ntpath.basename(parent).lower() == "usr" else parent def _git_bash_aslr_help(bash: str, details: str = "") -> str: """Build the targeted per-program Mandatory-ASLR remediation.""" - git_root = _git_root_from_bash(bash) - escaped_root = git_root.replace("'", "''") + escaped_root = _git_root_from_bash(bash).replace("'", "''") detail_line = f"\nGit Bash probe output: {details[:500]}" if details else "" return ( f"Git Bash at {bash} cannot launch required MSYS child processes while " @@ -105,24 +95,20 @@ def _bash_starts(bash: str) -> bool: cached = _bash_starts_cache.get(bash) if cached is not None: return cached - try: result = subprocess.run( [bash, "--noprofile", "--norc", "-c", _BASH_EXTERNAL_PROGRAM_PROBE], - capture_output=True, - text=True, encoding="utf-8", errors="replace", - timeout=15, - creationflags=windows_hide_flags() if _IS_WINDOWS else 0, + capture_output=True, text=True, encoding="utf-8", errors="replace", + timeout=15, creationflags=windows_hide_flags() if _IS_WINDOWS else 0, ) ok = result.returncode == 0 if not ok: - combined = f"{result.stdout or ''}{result.stderr or ''}" - _bash_probe_details_cache[bash] = combined.strip()[:2000] - logger.debug("bash probe failed for %s: %s", bash, combined.strip()[:200]) + combined = f"{result.stdout or ''}{result.stderr or ''}".strip() + _bash_probe_details_cache[bash] = combined[:2000] + logger.debug("bash probe failed for %s: %s", bash, combined[:200]) except Exception as exc: _bash_probe_details_cache[bash] = str(exc)[:2000] logger.debug("bash probe error for %s: %s", bash, exc) ok = False - _bash_starts_cache[bash] = ok return ok diff --git a/tools/environments/local_pythonpath.py b/tools/environments/local_pythonpath.py index daef83fc91..901fbb48dd 100644 --- a/tools/environments/local_pythonpath.py +++ b/tools/environments/local_pythonpath.py @@ -1,13 +1,12 @@ """Hermes-owned PYTHONPATH stripping for child processes. -Launchers prepend the repo root and the Hermes venv's site-packages so the -backend can ``import tools``; leaked into a child Python of a DIFFERENT version -they load the backend's C extensions and crash (numpy, PIL, cryptography). Only -entries proven Hermes-owned by *path provenance* are removed — never by a -cross-version heuristic — so user entries survive. Module state -(``_hermes_repo_root_aliases``, ``_in_venv``, ``_hermes_site_packages``) stays -in ``tools.environments.local`` and is read via :func:`_state` so tests that -monkeypatch it there keep working. +Launchers prepend the repo root and the Hermes venv's site-packages so the backend +can ``import tools``; leaked into a child Python of a DIFFERENT version they load +the backend's C extensions and crash. Only entries proven Hermes-owned by *path +provenance* are removed — never by a cross-version heuristic. Module state +(``_hermes_repo_root_aliases``, ``_in_venv``, ``_hermes_site_packages``) stays in +``tools.environments.local`` (read via :func:`_state`) so tests monkeypatching it +there keep working. """ import logging @@ -32,22 +31,18 @@ def _state(): def _same_path(left: Path, right: Path) -> bool: """Compare path spellings with host filesystem case semantics.""" - left_parts = [os.path.normcase(part) for part in left.parts] - right_parts = [os.path.normcase(part) for part in right.parts] - return left_parts == right_parts + return [os.path.normcase(p) for p in left.parts] == [os.path.normcase(p) for p in right.parts] def _build_hermes_repo_root_aliases( - resolved_root: Path, - lexical_root: Path, - configured_home: Path, + resolved_root: Path, lexical_root: Path, configured_home: Path, ) -> tuple[Path, ...]: """Exact repo-root spellings emitted by Hermes launchers. Mirrors - ``gateway_windows._preserve_hermes_home_path`` (physical path under the - resolved HERMES_HOME -> configured spelling) so a junction-backed install - matches without treating arbitrary HERMES_HOME descendants as Hermes-owned. - A repo-level junction (possibly cross-drive) is accepted only when a strict - resolve proves / is the physical root (fail-closed).""" + ``gateway_windows._preserve_hermes_home_path`` (physical path under the resolved + HERMES_HOME -> configured spelling) so a junction-backed install matches without + treating arbitrary HERMES_HOME descendants as Hermes-owned. A repo-level junction + (possibly cross-drive) is accepted only when a strict resolve proves + / is the physical root (fail-closed).""" aliases: list[Path] = [] def add(candidate: Path) -> None: @@ -58,8 +53,7 @@ def _build_hermes_repo_root_aliases( add(lexical_root) # Profile re-home: with --profile the configured home is /profiles/ - # and the repo lives beside the profiles dir, so derive the root lexically the - # same way get_default_hermes_root() does and map against it too. + # and the repo lives beside the profiles dir (as get_default_hermes_root() does). home_candidates = [configured_home] if configured_home.parent.name == "profiles": home_candidates.append(configured_home.parent.parent) @@ -68,10 +62,8 @@ def _build_hermes_repo_root_aliases( try: resolved_home = home.resolve() home_key = os.path.normcase(str(resolved_home)) - root_key = os.path.normcase(str(resolved_root)) - if os.path.commonpath([home_key, root_key]) == home_key: - relative_root = os.path.relpath(str(resolved_root), str(resolved_home)) - add(home / relative_root) + if os.path.commonpath([home_key, os.path.normcase(str(resolved_root))]) == home_key: + add(home / os.path.relpath(str(resolved_root), str(resolved_home))) except (OSError, ValueError): pass @@ -89,105 +81,77 @@ def _build_hermes_repo_root_aliases( def _validated_runtime_venv(env: dict) -> Path | None: - """Producer-owned runtime venv identified by VIRTUAL_ENV, or None. The - variable alone is not provenance (users carry unrelated venvs): require the - legacy Windows base-Python producer's exact ``/venv`` layout AND a - real ``pyvenv.cfg``.""" + """Producer-owned runtime venv identified by VIRTUAL_ENV, or None. The variable + alone is not provenance (users carry unrelated venvs): require the legacy Windows + base-Python producer's exact ``/venv`` layout AND a real ``pyvenv.cfg``.""" value = env.get("VIRTUAL_ENV") if not value: return None - candidate = Path(value) - aliases = _state()._hermes_repo_root_aliases - if not any(_same_path(candidate, repo_root / "venv") for repo_root in aliases): + if not any(_same_path(candidate, root / "venv") for root in _state()._hermes_repo_root_aliases): return None - try: - if not (candidate / "pyvenv.cfg").is_file(): - return None + return candidate if (candidate / "pyvenv.cfg").is_file() else None except OSError: return None - return candidate - def _get_hermes_site_packages(env: dict) -> list[Path]: """Exact site-packages dirs owned by the Hermes runtime (cached): ``site.getsitepackages()`` with a ``sys.prefix`` fallback, plus a validated Windows base-interpreter launch's ``VIRTUAL_ENV/Lib/site-packages``.""" local = _state() - if local._hermes_site_packages is not None: - result = list(local._hermes_site_packages) - else: - result = [] + if local._hermes_site_packages is None: + result: list[Path] = [] if local._in_venv: try: import site - for sp in site.getsitepackages(): - result.append(Path(sp)) + result.extend(Path(sp) for sp in site.getsitepackages()) except Exception: pass - if not result: if _IS_WINDOWS: result.append(Path(sys.prefix) / "Lib" / "site-packages") else: pyver = f"python{sys.version_info[0]}.{sys.version_info[1]}" result.append(Path(sys.prefix) / "lib" / pyver / "site-packages") - local._hermes_site_packages = list(result) + result = list(local._hermes_site_packages) runtime_venv = _validated_runtime_venv(env) if runtime_venv is not None: runtime_site_packages = runtime_venv / "Lib" / "site-packages" if not any(_same_path(runtime_site_packages, existing) for existing in result): result.append(runtime_site_packages) - return result def _strip_hermes_owned_pythonpath_and_runtime_markers(env: dict) -> None: - """Strip Hermes-owned PYTHONPATH entries, then the runtime marker vars. - - Ordering is load-bearing: PYTHONPATH filtering must run BEFORE the markers - are removed so a validated Windows base-interpreter launch - (VIRTUAL_ENV -> /venv) can still prove ownership. - """ + """Strip Hermes-owned PYTHONPATH entries, then the runtime marker vars. Ordering + is load-bearing: PYTHONPATH filtering runs BEFORE the markers are removed so a + validated Windows base-interpreter launch (VIRTUAL_ENV -> /venv) can + still prove ownership.""" _strip_hermes_owned_pythonpath(env) for _marker in _ACTIVE_VENV_MARKER_VARS: env.pop(_marker, None) def _strip_hermes_owned_pythonpath(env: dict) -> None: - """Remove Hermes-owned PYTHONPATH entries. Only exact matches of the repo - root (any launcher spelling) and runtime site-packages are stripped — never - children/descendants, which are user paths. Empty components (= cwd) and - everything else are preserved byte-for-byte.""" + """Remove Hermes-owned PYTHONPATH entries: only exact matches of the repo root + (any launcher spelling) and runtime site-packages — never descendants, which are + user paths. Empty components (= cwd) and everything else are preserved.""" pp = env.get("PYTHONPATH") if not pp: return - - hermes_site_packages = _get_hermes_site_packages(env) - repo_roots = _state()._hermes_repo_root_aliases - + owned_paths = [*_get_hermes_site_packages(env), *_state()._hermes_repo_root_aliases] kept: list[str] = [] stripped: list[str] = [] - for entry in pp.split(os.pathsep): - if entry == "": - kept.append(entry) - continue - - entry_path = Path(entry) - owned = any(_same_path(entry_path, sp) for sp in hermes_site_packages) or any( - _same_path(entry_path, repo_root) for repo_root in repo_roots - ) + owned = entry != "" and any(_same_path(Path(entry), p) for p in owned_paths) (stripped if owned else kept).append(entry) - if kept: env["PYTHONPATH"] = os.pathsep.join(kept) else: env.pop("PYTHONPATH", None) - if stripped: logger.debug("Stripped Hermes-owned entries from PYTHONPATH: %s", stripped)