diff --git a/tests/tools/test_terminal_env_sanitization.py b/tests/tools/test_terminal_env_sanitization.py new file mode 100644 index 000000000..e15245378 --- /dev/null +++ b/tests/tools/test_terminal_env_sanitization.py @@ -0,0 +1,150 @@ +"""Regression tests for terminal subprocess environment scoping.""" + +import os +import shlex +import sys +import tempfile +import time +import types +from pathlib import Path + +_IMPORT_HERMES_HOME = Path(tempfile.mkdtemp(prefix="hermes_test_import_")) +for _subdir in ("sessions", "cron", "memories", "skills"): + (_IMPORT_HERMES_HOME / _subdir).mkdir() +os.environ["HERMES_HOME"] = str(_IMPORT_HERMES_HOME) + +for _key in ( + "OPENAI_API_KEY", + "OPENAI_BASE_URL", + "OPENROUTER_API_KEY", + "OPENROUTER_BASE_URL", + "ANTHROPIC_API_KEY", + "ALL_PROXY", + "all_proxy", + "HTTP_PROXY", + "http_proxy", + "HTTPS_PROXY", + "https_proxy", +): + os.environ.pop(_key, None) + +from tools.environments.local import LocalEnvironment +from tools.process_registry import ProcessRegistry + + +_BLOCKED_VAR = "OPENAI_BASE_URL" +_PARENT_VALUE = "http://parent.invalid/v1" +_EXPLICIT_VALUE = "http://explicit.invalid/v1" + + +def _print_env_command(var_name: str) -> str: + code = f'import os; print(os.getenv("{var_name}", ""))' + return f"{shlex.quote(sys.executable)} -c {shlex.quote(code)}" + + +def _wait_for_exit(session, timeout: float = 5.0): + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + if session.exited: + return + time.sleep(0.05) + raise AssertionError(f"Timed out waiting for session {session.id} to exit") + + +class TestLocalEnvironmentEnvScoping: + def test_execute_scrubs_provider_env_by_default(self, monkeypatch, tmp_path): + monkeypatch.setenv(_BLOCKED_VAR, _PARENT_VALUE) + + env = LocalEnvironment(cwd=str(tmp_path), timeout=5) + result = env.execute(_print_env_command(_BLOCKED_VAR)) + + assert result["returncode"] == 0 + assert result["output"].strip() == "" + + def test_execute_allows_explicit_provider_env_override(self, monkeypatch, tmp_path): + monkeypatch.setenv(_BLOCKED_VAR, _PARENT_VALUE) + + env = LocalEnvironment( + cwd=str(tmp_path), + timeout=5, + env={_BLOCKED_VAR: _EXPLICIT_VALUE}, + ) + result = env.execute(_print_env_command(_BLOCKED_VAR)) + + assert result["returncode"] == 0 + assert result["output"].strip() == _EXPLICIT_VALUE + + +class TestProcessRegistryEnvScoping: + def test_spawn_local_scrubs_provider_env_by_default(self, monkeypatch, tmp_path): + monkeypatch.setenv(_BLOCKED_VAR, _PARENT_VALUE) + + registry = ProcessRegistry() + session = registry.spawn_local( + command=_print_env_command(_BLOCKED_VAR), + cwd=str(tmp_path), + ) + + _wait_for_exit(session) + assert session.exit_code == 0 + assert session.output_buffer.strip() == "" + + def test_spawn_local_allows_explicit_provider_env_override(self, monkeypatch, tmp_path): + monkeypatch.setenv(_BLOCKED_VAR, _PARENT_VALUE) + + registry = ProcessRegistry() + session = registry.spawn_local( + command=_print_env_command(_BLOCKED_VAR), + cwd=str(tmp_path), + env_vars={_BLOCKED_VAR: _EXPLICIT_VALUE}, + ) + + _wait_for_exit(session) + assert session.exit_code == 0 + assert session.output_buffer.strip() == _EXPLICIT_VALUE + + def test_spawn_local_pty_scrubs_provider_env_by_default(self, monkeypatch, tmp_path): + monkeypatch.setenv(_BLOCKED_VAR, _PARENT_VALUE) + + captured = {} + + class _FakePty: + pid = 123 + exitstatus = 0 + + def isalive(self): + return False + + def read(self, _size): + return b"" + + def wait(self): + return 0 + + class _FakePtyProcess: + @staticmethod + def spawn(argv, cwd, env, dimensions): + captured["argv"] = list(argv) + captured["cwd"] = cwd + captured["env"] = dict(env) + captured["dimensions"] = dimensions + return _FakePty() + + monkeypatch.setitem( + sys.modules, + "ptyprocess", + types.SimpleNamespace(PtyProcess=_FakePtyProcess), + ) + + registry = ProcessRegistry() + session = registry.spawn_local( + command="true", + cwd=str(tmp_path), + use_pty=True, + ) + + _wait_for_exit(session) + assert session.exit_code == 0 + assert captured["cwd"] == str(tmp_path) + assert captured["env"]["PYTHONUNBUFFERED"] == "1" + assert _BLOCKED_VAR not in captured["env"] diff --git a/tools/environments/base.py b/tools/environments/base.py index 295c84daa..cf713792e 100644 --- a/tools/environments/base.py +++ b/tools/environments/base.py @@ -6,6 +6,27 @@ from pathlib import Path +_TERMINAL_BLOCKED_PROVIDER_ENV_KEYS = frozenset({ + "ANTHROPIC_API_KEY", + "GLM_API_KEY", + "GLM_BASE_URL", + "KIMI_API_KEY", + "KIMI_BASE_URL", + "MINIMAX_API_KEY", + "MINIMAX_BASE_URL", + "MINIMAX_CN_API_KEY", + "MINIMAX_CN_BASE_URL", + "NOUS_API_KEY", + "NOUS_BASE_URL", + "OPENAI_API_KEY", + "OPENAI_BASE_URL", + "OPENROUTER_API_KEY", + "OPENROUTER_BASE_URL", + "ZAI_API_KEY", + "Z_AI_API_KEY", +}) + + def get_sandbox_dir() -> Path: """Return the host-side root for all sandbox storage (Docker workspaces, Singularity overlays/SIF cache, etc.). @@ -21,6 +42,16 @@ def get_sandbox_dir() -> Path: return p +def build_terminal_subprocess_env(extra_env: dict | None = None) -> dict: + """Build a subprocess env without leaking Hermes runtime provider state.""" + env = dict(os.environ) + for key in _TERMINAL_BLOCKED_PROVIDER_ENV_KEYS: + env.pop(key, None) + if extra_env: + env.update(extra_env) + return env + + class BaseEnvironment(ABC): """Common interface for all Hermes execution backends. diff --git a/tools/environments/local.py b/tools/environments/local.py index 828de8181..54d242629 100644 --- a/tools/environments/local.py +++ b/tools/environments/local.py @@ -10,7 +10,7 @@ _IS_WINDOWS = platform.system() == "Windows" -from tools.environments.base import BaseEnvironment +from tools.environments.base import BaseEnvironment, build_terminal_subprocess_env # Unique marker to isolate real command output from shell init/exit noise. # printf (no trailing newline) keeps the boundaries clean for splitting. @@ -192,7 +192,7 @@ def execute(self, command: str, cwd: str = "", *, # Ensure PATH always includes standard dirs — systemd services # and some terminal multiplexers inherit a minimal PATH. _SANE_PATH = "/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin" - run_env = dict(os.environ | self.env) + run_env = build_terminal_subprocess_env(self.env) existing_path = run_env.get("PATH", "") if "/usr/bin" not in existing_path.split(":"): run_env["PATH"] = f"{existing_path}:{_SANE_PATH}" if existing_path else _SANE_PATH diff --git a/tools/process_registry.py b/tools/process_registry.py index 10d8c291a..3e1b5b768 100644 --- a/tools/process_registry.py +++ b/tools/process_registry.py @@ -42,6 +42,7 @@ import uuid _IS_WINDOWS = platform.system() == "Windows" +from tools.environments.base import build_terminal_subprocess_env from tools.environments.local import _find_shell from dataclasses import dataclass, field from pathlib import Path @@ -153,7 +154,7 @@ def spawn_local( else: from ptyprocess import PtyProcess as _PtyProcessCls user_shell = _find_shell() - pty_env = os.environ | (env_vars or {}) + pty_env = build_terminal_subprocess_env(env_vars) pty_env["PYTHONUNBUFFERED"] = "1" pty_proc = _PtyProcessCls.spawn( [user_shell, "-lic", command], @@ -194,7 +195,7 @@ def spawn_local( # Force unbuffered output for Python scripts so progress is visible # during background execution (libraries like tqdm/datasets buffer when # stdout is a pipe, hiding output from process(action="poll")). - bg_env = os.environ | (env_vars or {}) + bg_env = build_terminal_subprocess_env(env_vars) bg_env["PYTHONUNBUFFERED"] = "1" proc = subprocess.Popen( [user_shell, "-lic", command],