diff --git a/tests/tools/test_windows_native_support.py b/tests/tools/test_windows_native_support.py index b1adc95bf6..254ad1daf3 100644 --- a/tests/tools/test_windows_native_support.py +++ b/tests/tools/test_windows_native_support.py @@ -719,7 +719,8 @@ class TestLocalEnvironmentWindowsTempDir: source = (root / "tools" / "environments" / "local.py").read_text(encoding="utf-8") assert "if _IS_WINDOWS:" in source assert "get_hermes_home" in source - assert 'cache_dir = get_hermes_home() / "cache" / "terminal"' in source + assert 'get_hermes_home() / "cache" / "terminal"' in source + assert "_default_terminal_temp_dir()" in source class TestLocalEnvironmentPathInjectionGated: diff --git a/tools/env_passthrough.py b/tools/env_passthrough.py index 0c921c84cf..1ee587968e 100644 --- a/tools/env_passthrough.py +++ b/tools/env_passthrough.py @@ -1,24 +1,15 @@ -"""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. -""" +"""Environment variable passthrough registry: the session-scoped allowlist of vars a +skill's ``required_environment_variables`` (registered by ``skill_view``) or +``terminal.env_passthrough`` in config.yaml may forward into sandboxed children +(execute_code, terminal), which strip secrets by default. Under profile multiplexing, +forwarded values resolve through the profile's secret scope, not the process env.""" from __future__ import annotations import logging from contextvars import ContextVar from typing import Iterable -from hermes_cli.config import cfg_get +from hermes_cli.config import cfg_get, read_raw_config logger = logging.getLogger(__name__) @@ -43,110 +34,78 @@ _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). - - 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. - - 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. - """ + ``_HERMES_PROVIDER_ENV_BLOCKLIST`` or a dynamic Hermes-internal secret + (AUXILIARY_*_API_KEY / _BASE_URL, GATEWAY_RELAY_*). Skill-declared + ``required_environment_variables`` must not override this — that was the + GHSA-rhgp-j443-p4rf bypass (a skill registered ``OPENAI_API_KEY`` and received it + in the ``execute_code`` child); non-Hermes keys (TENOR_API_KEY, …) stay + registerable. Fails closed when the blocklist cannot be imported.""" try: from tools.environments.local import ( - _HERMES_PROVIDER_ENV_BLOCKLIST, - _is_hermes_internal_secret, - ) + _HERMES_PROVIDER_ENV_BLOCKLIST, _is_hermes_internal_secret) except Exception as e: logger.warning( "env passthrough: provider credential blocklist import failed; " - "failing closed and refusing passthrough registration for %r: %s", - name, - e, - ) + "failing closed and refusing passthrough registration for %r: %s", name, 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: - """Register env var names as allowed in sandboxed environments (typically - 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. - """ - for name in var_names: - name = name.strip() - if not name: - continue - if _is_hermes_provider_credential(name): - logger.warning( - "env passthrough: refusing to register Hermes provider " - "credential %r (blocked by _HERMES_PROVIDER_ENV_BLOCKLIST). " - "Skills must not override the execute_code sandbox's " - "credential scrubbing; see GHSA-rhgp-j443-p4rf.", - name, - ) - continue + """Register env var names as allowed in sandboxed environments (typically a + skill's ``required_environment_variables``). Hermes-managed provider credentials + are rejected (GHSA-rhgp-j443-p4rf) — such skills should use the main-process tools + (web_search, web_extract, …); third-party keys pass normally.""" + for name in _accepted((n.strip() for n in var_names), ( + "env passthrough: refusing to register Hermes provider " + "credential %r (blocked by _HERMES_PROVIDER_ENV_BLOCKLIST). " + "Skills must not override the execute_code sandbox's " + "credential scrubbing; see GHSA-rhgp-j443-p4rf." + )): _get_allowed().add(name) logger.debug("env passthrough: registered %s", name) +def _accepted(names, refusal_msg: str): + """Yield non-empty *names* that are not Hermes provider credentials; refused + names are logged with *refusal_msg* (``%r`` = name).""" + for name in names: + if not name: + continue + if _is_hermes_provider_credential(name): + logger.warning(refusal_msg, name) + continue + yield name + + 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 - 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") + items = passthrough if isinstance(passthrough, list) else () + result.update(_accepted((i.strip() for i in items if isinstance(i, str)), ( + "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." + ))) except Exception as e: logger.debug("Could not read tools.env_passthrough from config: %s", e) - _config_passthrough = frozenset(result) return _config_passthrough 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,41 +113,22 @@ 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: - """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. - """ +def resolve_passthrough_value(name: str, fallback: str | None = None) -> str | None: + """Resolve an allowlisted variable without crossing profile boundaries. ``fallback`` + is what the caller would have forwarded before secret scopes existed (a snapshot of + ``os.environ`` / the profile ``.env``). An active multiplex scope is authoritative: + a missing key returns ``None``, never the process-global env, and an unscoped read + raises the fail-closed ``UnscopedSecretError``. Outside multiplexing an installed + scope keeps overlay semantics and an unscoped caller keeps its fallback.""" from agent.secret_scope import ( - _is_global_env, - current_secret_scope, - get_secret, - 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. + _is_global_env, current_secret_scope, get_secret, is_multiplex_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..49405141a6 100644 --- a/tools/env_probe.py +++ b/tools/env_probe.py @@ -1,17 +1,8 @@ -"""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``). - -Toggle via ``agent.environment_probe`` in config.yaml (default True). -""" +"""Local-environment toolchain probe for the system prompt: when the terminal backend +is local, one deterministic line about Python tooling (python3/python versions, missing +pip, pip bound to another Python, PEP 668) so models don't discover it by hitting +walls. Cached per process; "" when clean. Remote backends are skipped (the sandbox has +its own probe in agent/prompt_builder). Toggle: ``agent.environment_probe`` in config.yaml.""" from __future__ import annotations @@ -26,26 +17,17 @@ 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. +# Concurrency model: exactly ONE background worker runs the probe; ``_PROBE_DONE`` +# signals completion. Callers block at most ``_PROBE_WAIT_TIMEOUT`` s then fail open +# with "" — a stuck probe (e.g. a Windows pipe wedged 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. -_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. -_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. -_WAIT_ALREADY_TIMED_OUT = False +_PROBE_GEN = 0 # bumped on reset so a stale worker can't publish into the fresh generation +_PROBE_WAIT_TIMEOUT = 10.0 # healthy runtime ~0.5s +_WAIT_ALREADY_TIMED_OUT = False # after one full wait, later callers only peek # Keep in sync with agent/prompt_builder.py:_REMOTE_TERMINAL_BACKENDS. # Duplicated rather than imported to avoid a circular import. @@ -68,90 +50,63 @@ 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). - - 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. - """ + """Run a short subprocess -> (returncode, stdout, stderr); failures (binary + missing, timeout, OSError) return (-1, "", ""). 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 make ``wait()`` cover only the 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, - stdin=subprocess.DEVNULL, - # CREATE_NO_WINDOW (0 on POSIX): windowless hosts (pythonw - # gateway / kanban workers) would otherwise flash a console - # window per probe subprocess. - creationflags=windows_hide_flags(), - ) + cmd, stdout=out_f, stderr=err_f, timeout=timeout, check=False, + # CREATE_NO_WINDOW (0 on POSIX): pythonw hosts would flash a console + stdin=subprocess.DEVNULL, creationflags=windows_hide_flags()) except subprocess.TimeoutExpired: return -1, "", "timeout" out_f.seek(0) err_f.seek(0) - out = out_f.read().decode("utf-8", "replace").strip() - err = err_f.read().decode("utf-8", "replace").strip() - return result.returncode, out, err + return (result.returncode, out_f.read().decode("utf-8", "replace").strip(), + err_f.read().decode("utf-8", "replace").strip()) except FileNotFoundError: return -1, "", "not found" except OSError as exc: return -1, "", f"oserror: {exc}" -def _python_version_of(binary: str) -> Optional[str]: - """Return a short version string like ``3.12.4`` for ``binary``, or None.""" +def _py_out(binary: str, *args: str) -> Optional[str]: + """stdout of `` *args`` when the binary is on PATH and exits 0, else None.""" if not shutil.which(binary): return None - rc, out, err = _run([binary, "-c", "import sys; print(f'{sys.version_info.major}.{sys.version_info.minor}.{sys.version_info.micro}')"]) - return out if rc == 0 and out else None + rc, out, _err = _run([binary, *args]) + return out if rc == 0 else None + + +def _python_version_of(binary: str) -> Optional[str]: + """Return a short version string like ``3.12.4`` for ``binary``, or None.""" + code = "import sys; print('.'.join(map(str, sys.version_info[:3])))" + return _py_out(binary, "-c", code) or None 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 _py_out(binary, "-m", "pip", "--version") is not None def _detect_pep668(binary: str) -> bool: """True when ```` is PEP-668 externally-managed (``EXTERNALLY-MANAGED`` marker next to the stdlib, as Debian/Ubuntu ship).""" - if not shutil.which(binary): - return False - code = ( - "import sys, os;" - "stdlib = os.path.dirname(os.__file__);" - "marker = os.path.join(stdlib, 'EXTERNALLY-MANAGED');" - "print('yes' if os.path.exists(marker) else 'no')" - ) - rc, out, _err = _run([binary, "-c", code]) - return rc == 0 and out.strip() == "yes" + code = ("import os; print('yes' if os.path.exists(os.path.join(" + "os.path.dirname(os.__file__), 'EXTERNALLY-MANAGED')) else 'no')") + return (_py_out(binary, "-c", code) or "").strip() == "yes" 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 not shutil.which("pip"): - return None - rc, out, _err = _run(["pip", "--version"]) - if rc != 0 or not out: - return None + """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.""" + out = _py_out("pip", "--version") or "" if "(python " in out and out.endswith(")"): return out.rsplit("(python ", 1)[1][:-1].strip() return None @@ -169,104 +124,74 @@ 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 includes + # the Hermes-managed $HERMES_HOME/bin via local.py. 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") - if py_ver and py_ver != py3_ver: bits.append(f"python={py_ver}") elif not py_ver and py3_ver: # Common on Debian/Ubuntu — stop the model typing `python`. bits.append("python=missing (use python3)") - if pip_bound_to: 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 worker + # thread does not inherit. Remote backends answer "" 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 "" - if _PROBE_DONE.is_set(): return _CACHED_LINE or "" - _ensure_probe_started() 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( "env_probe did not finish within %.0fs; building the system " - "prompt without the Python toolchain line", - _PROBE_WAIT_TIMEOUT, - ) + "prompt without the Python toolchain line", _PROBE_WAIT_TIMEOUT) return "" return _CACHED_LINE or "" @@ -290,26 +215,17 @@ 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..4b0ef14a2b 100644 --- a/tools/environments/local.py +++ b/tools/environments/local.py @@ -1,5 +1,6 @@ """Local execution environment — spawn-per-call with session snapshot.""" +import contextlib import logging import ntpath import os @@ -26,37 +27,30 @@ 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. +# --- Terminal temp-cache pruning --- +# 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)$") @@ -69,12 +63,9 @@ def _default_terminal_temp_dir() -> "Path | None": return None -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 @@ -91,8 +82,7 @@ def cleanup_terminal_temp_cache( mtimes[f] = mt = f.stat().st_mtime except OSError: continue - m = _BG_GROUP_RE.match(f.name) - if m: + if m := _BG_GROUP_RE.match(f.name): group_newest[m.group(1)] = max(group_newest.get(m.group(1), 0.0), mt) removed = 0 @@ -101,10 +91,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 @@ -125,28 +112,21 @@ def _prune_terminal_temp_once() -> None: # --- Windows / MSYS path translation --- - - 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.""" - if not _IS_WINDOWS or not cwd: - return cwd - m = re.match(r'^/(?:(?:cygdrive|mnt)/)?([a-zA-Z])(/.*)?$', cwd) + """``/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.""" + m = _IS_WINDOWS and cwd and 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) @@ -156,10 +136,8 @@ def _resolve_local_initial_cwd(cwd: str) -> str: return expanded if os.path.isabs(expanded): return expanded - 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,50 +147,40 @@ 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.""" - if not _IS_WINDOWS or not cwd: - return cwd - m = re.match(r'^([a-zA-Z]):[\\/]*(.*)$', cwd) + """Native ``C:\\Users\\x`` -> Git Bash ``/c/Users/x`` so ``builtin cd`` resolves + it. No-op off Windows / for non-drive paths.""" + m = _IS_WINDOWS and cwd and 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.""" - if not _IS_WINDOWS or not path: - return path - return _windows_to_msys_path(path).replace("\\", "/") + """*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.""" + return _windows_to_msys_path(path).replace("\\", "/") if _IS_WINDOWS and path else path 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 = _msys_to_windows_path(cwd) if _IS_WINDOWS else cwd + """``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 cwd and _cwd_usable(cwd): return cwd if cwd and os.path.isdir(cwd): @@ -222,55 +190,38 @@ 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): - return parent + while parent and not _cwd_usable(parent): next_parent = os.path.dirname(parent) if next_parent == parent: - break # filesystem root itself is unusable + return tempfile.gettempdir() # filesystem root itself is unusable parent = next_parent - return tempfile.gettempdir() + return parent or tempfile.gettempdir() # --- 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.""" + from hermes_constants import apply_subprocess_home_env, get_hermes_home_override try: - from hermes_constants import get_hermes_home_override - - value = get_hermes_home_override() - if value: + if value := get_hermes_home_override(): 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) 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 on a concurrent multi-session host, so once the session + context is engaged ContextVars are authoritative — a bound value (incl. "") wins + and an _UNSET var is STRIPPED, not inherited. An unengaged CLI keeps the mirror.""" try: from gateway.session_context import _UNSET, _VAR_MAP, session_context_engaged except Exception: return - _engaged = session_context_engaged() for var_name, var in _VAR_MAP.items(): value = var.get() @@ -280,11 +231,48 @@ def _inject_session_context_env(env: dict) -> None: env.pop(var_name, None) -def _scrub_delegated_child_kanban_env(env: dict[str, str]) -> dict[str, str]: - """Strip dispatcher-owned Kanban env from delegate_task child subprocesses.""" +def _filter_secret_env( + 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 extras / terminal env), else is + dropped. Blocklisted names survive only via env_passthrough registration or as + context-entitled first-party ``BUZZ_*`` vars; the latter are used directly, never + scope-resolved (UnscopedSecretError under multiplex).""" try: - from agent.delegation_context import is_delegated_child_process_context, scrub_kanban_env + 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 + 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_env_passthrough(key) + if key in _HERMES_PROVIDER_ENV_BLOCKLIST and not (passthrough or first_party): + continue + if passthrough and not first_party: + 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.""" + _apply_profile_home(env) + _inject_session_context_env(env) + _strip_hermes_owned_pythonpath_and_runtime_markers(env) + _apply_windows_msys_bash_env_defaults(env) + try: # strip dispatcher-owned Kanban env from delegate_task child subprocesses + 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,140 +280,56 @@ 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: - """Copy *items* into *out*, dropping Hermes-managed secrets. - - ``_HERMES_FORCE_`` unwraps to ``NAME`` when ``unwrap_force`` (caller - extras / terminal env), else is dropped. Blocklisted names survive only via - env_passthrough registration or as context-entitled first-party ``BUZZ_*`` - 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() - 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 - 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) - 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 - - -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.""" - _apply_profile_home(env) - _inject_session_context_env(env) - _strip_hermes_owned_pythonpath_and_runtime_markers(env) - _apply_windows_msys_bash_env_defaults(env) - return _scrub_delegated_child_kanban_env(env) +def _scrubbed_env(parts, plugin_strip: frozenset, fix_path) -> dict: + """Filter each ``(items, unwrap_force)`` in *parts* into one env, rewrite PATH via + *fix_path* (always prepending the hermes install dir so bare ``hermes`` resolves + for children of a systemd/cron-launched gateway), then apply the shared guards.""" + out: dict[str, str] = {} + for items, unwrap_force in parts: + _filter_secret_env(items, out, unwrap_force=unwrap_force, plugin_strip=plugin_strip) + path_key = _path_env_key(out) + if path_key is not None: + out[path_key] = _prepend_hermes_bin_dir(fix_path(out.get(path_key, ""))) + return _finalize_child_env(out) 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. - """ - 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). - 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) + """Filter Hermes-managed secrets from a subprocess environment (background/PTY + spawn path, search workers, computer-use driver, user-script runners).""" + return _scrubbed_env([(base_env or {}, False), (extra_env or {}, True)], + _plugin_terminal_env_strip_keys(), lambda p: p) 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``. - """ + """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). Terminal/execute_code use ``_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. ``base=None`` snapshots ``os.environ``. + ``scrub_secrets=True`` -> :func:`_sanitize_subprocess_env` (profile home inherent, + ``inherit_profile_home`` ignored). ``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: @@ -434,32 +338,22 @@ def build_subprocess_env( # --- Shell discovery --- - - 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 "") - - 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")) - + """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.""" + getenv = os.environ.get + lad = getenv("LOCALAPPDATA", "") + roots = [ + lad and os.path.join(lad, "hermes", "git", "bin"), + lad and os.path.join(lad, "hermes", "git", "usr", "bin"), + os.path.join(getenv("ProgramFiles", r"C:\Program Files"), "Git", "bin"), + os.path.join(getenv("ProgramFiles(x86)", r"C:\Program Files (x86)"), "Git", "bin"), + lad and os.path.join(lad, "Programs", "Git", "bin"), + ] + raw = [custom or "", *(os.path.join(r, "bash.exe") for r in roots if r)] + candidates = list(dict.fromkeys(c for c in raw if c and os.path.isfile(c))) found = shutil.which("bash") if found and found not in candidates: candidates.append(found) @@ -469,53 +363,41 @@ def _windows_bash_candidates(custom: "str | None") -> list[str]: def _find_bash() -> str: """Find bash for command execution.""" if not _IS_WINDOWS: - return ( - shutil.which("bash") - 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" - ) - + return (shutil.which("bash") + or next((p for p in ("/usr/bin/bash", "/bin/bash") if os.path.isfile(p)), None) + or os.environ.get("SHELL") 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 # real bash error instead of a less useful "not found". return candidates[0] - 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 [] @@ -527,69 +409,50 @@ def _compute_git_bash_bin_dirs() -> list[str]: bash = _find_bash() except Exception: return [] - bin_dir = os.path.dirname(bash) # \bin or \usr\bin (MinGit) - parent = os.path.dirname(bin_dir) + parent = os.path.dirname(os.path.dirname(bash)) # bash in \bin or \usr\bin (MinGit) root = os.path.dirname(parent) if os.path.basename(parent).lower() == "usr" else parent - dirs: list[str] = [] - for sub in (("mingw64", "bin"), ("mingw32", "bin"), ("usr", "local", "bin"), ("usr", "bin"), ("bin",)): - candidate = os.path.join(root, *sub) - if os.path.isdir(candidate) and candidate not in dirs: - dirs.append(candidate) - return dirs + subs = ("mingw64/bin", "mingw32/bin", "usr/local/bin", "usr/bin", "bin") + dirs = (os.path.join(root, *sub.split("/")) for sub in subs) + return list(dict.fromkeys(d for d in dirs if os.path.isdir(d))) 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.""" - if not dirs: - return existing_path - sep = os.pathsep - entries = [e for e in existing_path.split(sep) if e] if existing_path else [] + """Prepend *dirs* missing from *existing_path* (``os.pathsep``); an already-listed + dir keeps its position; unchanged input when nothing is missing.""" + entries = [e for e in existing_path.split(os.pathsep) if e] 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]) if missing else existing_path 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,60 +461,42 @@ _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). A gateway + launched by systemd/cron/a desktop launcher lacks the install dir on PATH 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 - - if candidate and not os.path.isdir(candidate): - candidate = None - - _HERMES_BIN_DIR = candidate - return candidate + elif (os.path.isabs(argv0) and (base == "hermes" or base.startswith("hermes.")) + and os.path.isfile(argv0)): + candidate = os.path.dirname(argv0) + else: + candidate = exe_dir if exe_dir and os.path.isfile(os.path.join(exe_dir, shim)) else None + _HERMES_BIN_DIR = candidate if candidate and os.path.isdir(candidate) else None + return _HERMES_BIN_DIR 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()] + return [str(d) for d in (*iter_hermes_node_dirs(), get_hermes_home() / "bin") if d.is_dir()] except Exception: return [] @@ -663,189 +508,118 @@ 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 - ) - if not existing_path: - return ":".join(sane_entries) # dict preserves first-occurrence order; empty entries dropped. ordered = dict.fromkeys(entry for entry in existing_path.split(":") if entry) - ordered.update(dict.fromkeys(sane_entries)) + ordered.update(dict.fromkeys([*_SANE_PATH.split(":"), *_managed_runtime_path_entries()])) return ":".join(ordered) 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.""" - if not _IS_WINDOWS: - return - env.setdefault("MSYS_NO_PATHCONV", "1") - env.setdefault("MSYS2_ARG_CONV_EXCL", "*") + """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 _IS_WINDOWS: + env.setdefault("MSYS_NO_PATHCONV", "1") + env.setdefault("MSYS2_ARG_CONV_EXCL", "*") 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.""" - if not _IS_WINDOWS: - return "PATH" - for key in run_env: - if key.upper() == "PATH": - return key - return None + """PATH env key to update without altering Windows casing (``Path`` vs ``PATH``); + None when a Windows env has no PATH key at all.""" + return next((k for k in run_env if k.upper() == "PATH"), None) if _IS_WINDOWS else "PATH" def _make_run_env(env: dict) -> dict: """Build a run environment with a sane PATH and provider-var stripping.""" - run_env: dict = {} - _filter_secret_env(dict(os.environ | env), run_env, unwrap_force=True) - 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) - return _finalize_child_env(run_env) + return _scrubbed_env([(dict(os.environ | env), True)], frozenset(), + lambda p: _prepend_git_bash_dirs(_append_missing_sane_path_entries(p))) # --- 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. +# Owned here; read lazily by tools.environments.local_pythonpath (tests patch here). +# The Electron app prepends the repo root to PYTHONPATH so the backend can ``import +# tools``; other subprocesses must not inherit it. Aliases: launchers may emit other +# spellings — the Windows gateway launcher renders Hermes-owned paths under the +# configured HERMES_HOME spelling (possibly a junction to another drive). _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. _hermes_repo_root_aliases: tuple[Path, ...] = _build_hermes_repo_root_aliases( - _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). -_in_venv: bool = ( - 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 + _hermes_repo_root, Path(__file__).absolute().parents[2], get_process_hermes_home()) +_in_venv: bool = (getattr(sys, "base_prefix", sys.prefix) != sys.prefix + or hasattr(sys, "real_prefix")) # real_prefix: virtualenv<20 +_hermes_site_packages: list[Path] | None = None # lazily cached by local_pythonpath # --- Login-shell init files --- - - 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: path = os.path.expandvars(os.path.expanduser(raw)) + if path and os.path.isfile(path): + resolved.append(path) except Exception: continue - if path and os.path.isfile(path): - resolved.append(path) return resolved 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"] - 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 + safe = [p.replace("'", "'\\''") for p in files] + prelude = ["set +e", *(f"[ -r '{p}' ] && . '{p}' 2>/dev/null || true" for p in safe)] + return "\n".join(prelude) + "\n" + cmd_string # --- Process-group teardown (POSIX) --- - - -def _group_alive(pgid: int) -> bool: - """POSIX-only probe; callers are behind the _IS_WINDOWS gate.""" - try: - os.killpg(pgid, 0) # windows-footgun: ok — POSIX process-group alive probe - return True - except ProcessLookupError: - return False - except PermissionError: - return True # exists, even if we cannot signal it - - 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). + POSIX-only; callers are behind the _IS_WINDOWS gate.""" deadline = time.monotonic() + timeout - while time.monotonic() < deadline: + while True: try: proc.poll() except Exception: pass - if not _group_alive(pgid): + try: + os.killpg(pgid, 0) # windows-footgun: ok — POSIX process-group alive probe + except ProcessLookupError: return True + except PermissionError: + pass # exists, even if we cannot signal it + 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(): @@ -853,7 +627,7 @@ def _sweep_escaped_descendants(descendants: list, pgid: int) -> None: try: if os.getpgid(child.pid) == pgid: continue # group-kill already covers it - except (ProcessLookupError, PermissionError, OSError): + except OSError: # ProcessLookupError / PermissionError included pass child.kill() except Exception: @@ -861,129 +635,91 @@ 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: + if (pgid := getattr(proc, "_hermes_pgid", None)) is None: raise - - descendants = _snapshot_descendants(proc) - + try: # psutil children snapshot; empty on any failure (must never break the kill) + import psutil + descendants = psutil.Process(proc.pid).children(recursive=True) + except Exception: + descendants = [] 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) + with contextlib.suppress(subprocess.TimeoutExpired, OSError): + proc.wait(timeout=0.2) 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) def _kill_process_windows(proc) -> None: + """Identity-checked terminate (start time guards against PID reuse), else kill.""" 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: + with contextlib.suppress(subprocess.TimeoutExpired, OSError): proc.wait(timeout=2.0) - except (subprocess.TimeoutExpired, OSError): - pass 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. - """ + """Run commands directly on the host: every execute() spawns a fresh bash; + the session snapshot preserves env vars across calls; CWD persists via the + stdout marker.""" _profile_scoped_passthrough = True - - # Commands run on the Hermes host itself — controller-side platform - # behavior (macOS TCC pruning, etc.) legitimately applies here. + # Commands run on the Hermes host itself — controller-side platform behavior + # (macOS TCC pruning, etc.) legitimately applies here. 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``, TMPDIR/TMP/TEMP + (Termux has no /tmp), ``HERMES_HOME/cache/terminal`` (real storage: tmpfs /tmp + fills under Hermes load; pruned by ``cleanup_terminal_temp_cache``), /tmp, + ``tempfile.gettempdir()``; backend env before process env so terminal.env + overrides work. Windows: ``%TEMP%`` often has spaces that break unquoted bash, + so always the HERMES_HOME cache dir with forward slashes (bash- and Python-valid).""" 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("\\", "/") - def _posix(p: str) -> str: return p.rstrip("/") or "/" - - 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"): + for env_var in ("TERMINAL_TEMP_DIR", "TMPDIR", "TMP", "TEMP"): candidate = self.env.get(env_var) or os.environ.get(env_var) - if candidate and candidate.startswith("/"): + if candidate and candidate.startswith("/") and ( + env_var != "TERMINAL_TEMP_DIR" or os.path.isdir(candidate)): 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,12 +727,10 @@ 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" + fallback = tempfile.gettempdir() + return _posix(fallback) if fallback.startswith("/") else "/tmp" @staticmethod def _quote_cwd_for_cd(cwd: str) -> str: @@ -1008,91 +742,59 @@ 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 - normalized = _msys_to_windows_path(self.cwd) if _IS_WINDOWS else self.cwd - if safe_cwd != normalized: + if safe_cwd != _msys_to_windows_path(self.cwd): 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) - args = [bash, "-l", "-c", cmd_string] if login else [bash, "-c", cmd_string] - run_env = _make_run_env(self.env) - + cmd_string = _prepend_shell_init(cmd_string, _resolve_shell_init_files()) + args = [bash, *(["-l"] if login else []), "-c", cmd_string] 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=_make_run_env(self.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, + **({"creationflags": windows_hide_flags()} if _IS_WINDOWS else {})) if not _IS_WINDOWS: - try: + with contextlib.suppress(ProcessLookupError): 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): """Kill the entire process group (all children).""" try: - if _IS_WINDOWS: - _kill_process_windows(proc) - else: - _kill_process_group_posix(proc) - except (ProcessLookupError, PermissionError, OSError): - try: + (_kill_process_windows if _IS_WINDOWS else _kill_process_group_posix)(proc) + except OSError: # ProcessLookupError / PermissionError included + with contextlib.suppress(Exception): proc.kill() - 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: - normalized = _msys_to_windows_path(self.cwd) if _IS_WINDOWS else self.cwd + normalized = _msys_to_windows_path(self.cwd) if normalized and os.path.isdir(normalized): self.cwd = normalized result["cwd"] = normalized @@ -1105,13 +807,10 @@ 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: stale = [] for f in (self._snapshot_path, self._cwd_file, *stale): - try: + with contextlib.suppress(OSError): os.unlink(f) - except OSError: - pass diff --git a/tools/environments/local_env_policy.py b/tools/environments/local_env_policy.py index 2c91639557..d6ead6fbda 100644 --- a/tools/environments/local_env_policy.py +++ b/tools/environments/local_env_policy.py @@ -1,23 +1,18 @@ -"""Secret-scrub policy for Hermes child processes: pure data + predicates for -which env names are Hermes-managed credentials. The env *builders* applying it -(``_sanitize_subprocess_env``, ``_make_run_env``, ``hermes_subprocess_env``, -``build_subprocess_env``) live in ``tools.environments.local``.""" +"""Secret-scrub policy for Hermes child processes: pure data + predicates for which env +names are Hermes-managed credentials. The env *builders* applying it (``_make_run_env``, +``_sanitize_subprocess_env``, ``hermes_subprocess_env``) live in ``tools.environments.local``.""" 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. -_AWS_SDK_CREDENTIAL_ENV_VARS = frozenset({ - "AWS_BEARER_TOKEN_BEDROCK", -}) +# Hermes-managed AWS *inference* credentials for ``auth_type="aws_sdk"`` (Bedrock): +# only the Bedrock bearer token, which no aws/terraform/boto3 toolchain uses. The +# general AWS 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. +_AWS_SDK_CREDENTIAL_ENV_VARS = frozenset({"AWS_BEARER_TOKEN_BEDROCK"}) _STATIC_PROVIDER_ENV_BLOCKLIST = frozenset({ "OPENAI_BASE_URL", "OPENAI_API_KEY", "OPENAI_API_BASE", "OPENAI_ORG_ID", @@ -49,7 +44,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,61 +54,50 @@ 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(): category = metadata.get("category") if category in {"tool", "messaging"} or ( - category == "setting" and metadata.get("password") - ): + category == "setting" and metadata.get("password")): 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 (via the anthropic registry entry) belongs to the user's + # Claude Code install, not Hermes: 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. BUZZ_* is deliberately NOT discarded: this list feeds + # every scrub surface, so an import-time discard would leak BUZZ_PRIVATE_KEY into + # non-terminal children; the Buzz carve-out is terminal-only and context-gated + # (``_is_terminal_first_party_env``). 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``. 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); 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 +109,24 @@ 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. +# Active-venv markers that must NOT leak: VIRTUAL_ENV/CONDA_PREFIX make uv/poetry sync +# ANOTHER project's deps into the Hermes venv (still reachable via PATH, so stripping +# is safe); PYTHONHOME redirects a child interpreter's stdlib to the Hermes venv +# (version-mismatch crashes). PYTHONPATH is handled separately (Hermes-owned entries only). _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 +134,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 @@ -166,10 +145,9 @@ def _plugin_terminal_env_strip_keys() -> frozenset: return frozenset() -# Tier-1 secrets: stripped from EVERY spawned subprocess even under -# inherit_credentials (claude/codex/gemini). Not provider credentials — no child -# needs them and they are the highest-value secrets to keep from a compromised -# dependency. Provider keys are the conditional Tier-2 strip. +# Tier-1 secrets: stripped from EVERY spawned subprocess even under inherit_credentials +# (claude/codex/gemini). Not provider credentials — no child needs them and they are the +# highest-value secrets to keep from a compromised dependency. Provider keys = Tier 2. _ALWAYS_STRIP_KEYS: frozenset[str] = frozenset({ # GitHub auth "GH_TOKEN", "GITHUB_TOKEN", "GITHUB_APP_ID", "GITHUB_APP_PRIVATE_KEY_PATH", diff --git a/tools/environments/local_gitbash_probe.py b/tools/environments/local_gitbash_probe.py index a57fcb0937..eed5eb083f 100644 --- a/tools/environments/local_gitbash_probe.py +++ b/tools/environments/local_gitbash_probe.py @@ -37,34 +37,19 @@ def _mandatory_aslr_enabled() -> "bool | None": global _mandatory_aslr_enabled_cache if _mandatory_aslr_enabled_cache is not None: return _mandatory_aslr_enabled_cache - + cmd = [shutil.which("powershell.exe") or "powershell.exe", "-NoProfile", "-NonInteractive", + "-Command", "(Get-ProcessMitigation -System).Aslr.ForceRelocateImages.ToString()"] try: - powershell = shutil.which("powershell.exe") or "powershell.exe" - result = subprocess.run( - [ - powershell, - "-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(), - ) - 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"}: - _mandatory_aslr_enabled_cache = False - return False + result = subprocess.run(cmd, capture_output=True, text=True, encoding="utf-8", + errors="replace", timeout=10, creationflags=windows_hide_flags()) except Exception as exc: logger.debug("Could not query Windows Mandatory ASLR state: %s", exc) - return None + return None + if result.returncode != 0: + return None + value = (result.stdout or "").strip().upper() + _mandatory_aslr_enabled_cache = {"ON": True, "OFF": False, "NOTSET": False}.get(value) + return _mandatory_aslr_enabled_cache def _git_root_from_bash(bash: str) -> str: @@ -73,15 +58,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 " @@ -102,27 +84,21 @@ def _bash_starts(bash: str) -> bool: """True if *bash* can launch external MSYS programs (cached per path). ``--noprofile --norc`` so a broken login post-install (``Directory \\drivers\\etc``) does not falsely condemn an otherwise usable bash.""" - cached = _bash_starts_cache.get(bash) - if cached is not None: - return cached - + if bash in _bash_starts_cache: + return _bash_starts_cache[bash] 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..54a46b7bd2 100644 --- a/tools/environments/local_pythonpath.py +++ b/tools/environments/local_pythonpath.py @@ -1,14 +1,10 @@ -"""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. -""" +"""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. 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``) lives in ``tools.environments.local`` (via :func:`_state`) +so tests monkeypatching it there keep working.""" import logging import os @@ -32,162 +28,113 @@ 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).""" - aliases: list[Path] = [] - - def add(candidate: Path) -> None: - if not any(_same_path(candidate, existing) for existing in aliases): - aliases.append(candidate) - - add(resolved_root) - add(lexical_root) - + ``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).""" + candidates = [resolved_root, 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) - for home in home_candidates: 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: + candidates.append(home / os.path.relpath(str(resolved_root), str(resolved_home))) except (OSError, ValueError): pass - # Repo-level junction recovery (commonpath raises across drives, so the # home-relative mapping above cannot express a cross-drive link). for home in home_candidates: repo_candidate = home / resolved_root.name try: if repo_candidate.resolve(strict=True) == resolved_root.resolve(strict=True): - add(repo_candidate) + candidates.append(repo_candidate) except OSError: pass - + aliases: list[Path] = [] + for candidate in candidates: + if not any(_same_path(candidate, existing) for existing in aliases): + aliases.append(candidate) return tuple(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``.""" - value = env.get("VIRTUAL_ENV") - if not value: + """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``.""" + candidate = Path(env.get("VIRTUAL_ENV") or "") + if not env.get("VIRTUAL_ENV") or not any( + _same_path(candidate, root / "venv") for root in _state()._hermes_repo_root_aliases): 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): - 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") - + pyver = f"python{sys.version_info[0]}.{sys.version_info[1]}" + result.append(Path(sys.prefix) / "Lib" / "site-packages" if _IS_WINDOWS + else 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. Order is + load-bearing: PYTHONPATH filtering runs BEFORE the markers go 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 - - 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 - ) - (stripped if owned else kept).append(entry) - + owned_paths = [*_get_hermes_site_packages(env), *_state()._hermes_repo_root_aliases] + entries = pp.split(os.pathsep) + stripped = [e for e in entries if e and any(_same_path(Path(e), p) for p in owned_paths)] + kept = [e for e in entries if e not in stripped] 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)