Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
173 changes: 173 additions & 0 deletions tests/tools/test_local_env_blocklist.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,173 @@
"""Tests for provider env var blocklist in LocalEnvironment.

Verifies that Hermes-internal provider env vars (OPENAI_BASE_URL, etc.)
are stripped from subprocess environments so external CLIs are not
silently misrouted.

See: https://github.com/NousResearch/hermes-agent/issues/1002
"""

import os
import threading
from unittest.mock import MagicMock, patch

from tools.environments.local import (
LocalEnvironment,
_HERMES_PROVIDER_ENV_BLOCKLIST,
_HERMES_PROVIDER_ENV_FORCE_PREFIX,
)


def _make_fake_popen(captured: dict):
"""Return a fake Popen constructor that records the env kwarg."""
def fake_popen(cmd, **kwargs):
captured["env"] = kwargs.get("env", {})
proc = MagicMock()
proc.poll.return_value = 0
proc.returncode = 0
proc.stdout = iter([])
proc.stdout.close = lambda: None
proc.stdin = MagicMock()
return proc
return fake_popen


def _run_with_env(extra_os_env=None, self_env=None):
"""Execute a command via LocalEnvironment with mocked Popen
and return the env dict passed to the subprocess."""
captured = {}
fake_interrupt = threading.Event()
test_environ = {
"PATH": "/usr/bin:/bin",
"HOME": "/home/user",
"USER": "testuser",
}
if extra_os_env:
test_environ.update(extra_os_env)

env = LocalEnvironment(cwd="/tmp", timeout=10, env=self_env)

with patch("tools.environments.local._find_bash", return_value="/bin/bash"), \
patch("subprocess.Popen", side_effect=_make_fake_popen(captured)), \
patch("tools.terminal_tool._interrupt_event", fake_interrupt), \
patch.dict(os.environ, test_environ, clear=True):
env.execute("echo hello")

return captured.get("env", {})


class TestProviderEnvBlocklist:
"""Provider env vars loaded from ~/.hermes/.env must not leak."""

def test_blocked_vars_are_stripped(self):
"""OPENAI_BASE_URL and other provider vars must not appear in subprocess env."""
leaked_vars = {
"OPENAI_BASE_URL": "http://localhost:8000/v1",
"OPENAI_API_KEY": "sk-fake-key",
"OPENROUTER_API_KEY": "or-fake-key",
"ANTHROPIC_API_KEY": "ant-fake-key",
"LLM_MODEL": "anthropic/claude-opus-4-6",
}
result_env = _run_with_env(extra_os_env=leaked_vars)

for var in leaked_vars:
assert var not in result_env, f"{var} leaked into subprocess env"

def test_registry_derived_vars_are_stripped(self):
"""Vars from the provider registry (ANTHROPIC_TOKEN, ZAI_API_KEY, etc.)
must also be blocked — not just the hand-written extras."""
registry_vars = {
"ANTHROPIC_TOKEN": "ant-tok",
"CLAUDE_CODE_OAUTH_TOKEN": "cc-tok",
"ZAI_API_KEY": "zai-key",
"Z_AI_API_KEY": "z-ai-key",
"GLM_API_KEY": "glm-key",
"KIMI_API_KEY": "kimi-key",
"MINIMAX_API_KEY": "mm-key",
"MINIMAX_CN_API_KEY": "mmcn-key",
}
result_env = _run_with_env(extra_os_env=registry_vars)

for var in registry_vars:
assert var not in result_env, f"{var} leaked into subprocess env"

def test_safe_vars_are_preserved(self):
"""Standard env vars (PATH, HOME, USER) must still be passed through."""
result_env = _run_with_env()

assert "HOME" in result_env
assert result_env["HOME"] == "/home/user"
assert "USER" in result_env
assert "PATH" in result_env

def test_self_env_blocked_vars_also_stripped(self):
"""Blocked vars in self.env are stripped; non-blocked vars pass through."""
result_env = _run_with_env(self_env={
"OPENAI_BASE_URL": "http://custom:9999/v1",
"MY_CUSTOM_VAR": "keep-this",
})

assert "OPENAI_BASE_URL" not in result_env
assert "MY_CUSTOM_VAR" in result_env
assert result_env["MY_CUSTOM_VAR"] == "keep-this"


class TestForceEnvOptIn:
"""Callers can opt in to passing a blocked var via _HERMES_FORCE_ prefix."""

def test_force_prefix_passes_blocked_var(self):
"""_HERMES_FORCE_OPENAI_API_KEY in self.env should inject OPENAI_API_KEY."""
result_env = _run_with_env(self_env={
f"{_HERMES_PROVIDER_ENV_FORCE_PREFIX}OPENAI_API_KEY": "sk-explicit",
})

assert "OPENAI_API_KEY" in result_env
assert result_env["OPENAI_API_KEY"] == "sk-explicit"
# The force-prefixed key itself must not appear
assert f"{_HERMES_PROVIDER_ENV_FORCE_PREFIX}OPENAI_API_KEY" not in result_env

def test_force_prefix_overrides_os_environ_block(self):
"""Force-prefix in self.env wins even when os.environ has the blocked var."""
result_env = _run_with_env(
extra_os_env={"OPENAI_BASE_URL": "http://leaked/v1"},
self_env={f"{_HERMES_PROVIDER_ENV_FORCE_PREFIX}OPENAI_BASE_URL": "http://intended/v1"},
)

assert result_env["OPENAI_BASE_URL"] == "http://intended/v1"


class TestBlocklistCoverage:
"""Sanity checks that the blocklist covers all known providers."""

def test_issue_1002_offenders(self):
"""Blocklist includes the main offenders from issue #1002."""
must_block = {
"OPENAI_BASE_URL",
"OPENAI_API_KEY",
"OPENROUTER_API_KEY",
"ANTHROPIC_API_KEY",
"LLM_MODEL",
}
assert must_block.issubset(_HERMES_PROVIDER_ENV_BLOCKLIST)

def test_registry_vars_are_in_blocklist(self):
"""Every api_key_env_var and base_url_env_var from PROVIDER_REGISTRY
must appear in the blocklist — ensures no drift."""
from hermes_cli.auth import PROVIDER_REGISTRY

for pconfig in PROVIDER_REGISTRY.values():
for var in pconfig.api_key_env_vars:
assert var in _HERMES_PROVIDER_ENV_BLOCKLIST, (
f"Registry var {var} (provider={pconfig.id}) missing from blocklist"
)
if pconfig.base_url_env_var:
assert pconfig.base_url_env_var in _HERMES_PROVIDER_ENV_BLOCKLIST, (
f"Registry base_url_env_var {pconfig.base_url_env_var} "
f"(provider={pconfig.id}) missing from blocklist"
)

def test_extra_auth_vars_covered(self):
"""Non-registry auth vars (ANTHROPIC_TOKEN, CLAUDE_CODE_OAUTH_TOKEN)
must also be in the blocklist."""
extras = {"ANTHROPIC_TOKEN", "CLAUDE_CODE_OAUTH_TOKEN"}
assert extras.issubset(_HERMES_PROVIDER_ENV_BLOCKLIST)
59 changes: 58 additions & 1 deletion tools/environments/local.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,52 @@
# printf (no trailing newline) keeps the boundaries clean for splitting.
_OUTPUT_FENCE = "__HERMES_FENCE_a9f7b3__"

# Hermes-internal env vars that should NOT leak into terminal subprocesses.
# These are loaded from ~/.hermes/.env for Hermes' own LLM/provider calls
# but can break external CLIs (e.g. codex) that also honor them.
# See: https://github.com/NousResearch/hermes-agent/issues/1002
#
# Built dynamically from the provider registry so new providers are
# automatically covered without manual blocklist maintenance.
_HERMES_PROVIDER_ENV_FORCE_PREFIX = "_HERMES_FORCE_"


def _build_provider_env_blocklist() -> frozenset:
"""Derive the blocklist from the provider registry + known extras.

Automatically picks up api_key_env_vars and base_url_env_var from
every registered provider, so adding a new provider to auth.py is
enough — no manual list to keep in sync.
"""
blocked: set[str] = set()

try:
from hermes_cli.auth import PROVIDER_REGISTRY
for pconfig in PROVIDER_REGISTRY.values():
blocked.update(pconfig.api_key_env_vars)
if pconfig.base_url_env_var:
blocked.add(pconfig.base_url_env_var)
except ImportError:
pass

# Vars not in the registry but still Hermes-internal / conflict-prone
blocked.update({
"OPENAI_BASE_URL",
"OPENAI_API_KEY",
"OPENAI_API_BASE", # legacy alias
"OPENAI_ORG_ID",
"OPENAI_ORGANIZATION",
"OPENROUTER_API_KEY",
"ANTHROPIC_BASE_URL",
"ANTHROPIC_TOKEN", # OAuth token (not in registry as env var)
"CLAUDE_CODE_OAUTH_TOKEN",
"LLM_MODEL",
})
return frozenset(blocked)


_HERMES_PROVIDER_ENV_BLOCKLIST = _build_provider_env_blocklist()


def _find_bash() -> str:
"""Find bash for command execution.
Expand Down Expand Up @@ -192,7 +238,18 @@ 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)
# Strip Hermes-internal provider vars so external CLIs
# (e.g. codex) are not silently misrouted. Callers that
# truly need a blocked var can opt in by prefixing the key
# with _HERMES_FORCE_ in self.env (e.g. _HERMES_FORCE_OPENAI_API_KEY).
merged = dict(os.environ | self.env)
run_env = {}
for k, v in merged.items():
if k.startswith(_HERMES_PROVIDER_ENV_FORCE_PREFIX):
real_key = k[len(_HERMES_PROVIDER_ENV_FORCE_PREFIX):]
run_env[real_key] = v
elif k not in _HERMES_PROVIDER_ENV_BLOCKLIST:
run_env[k] = v
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
Expand Down