File size: 7,645 Bytes
b5b9c2e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 | """Tests for tools.env_passthrough — skill and config env var passthrough."""
import os
import pytest
import yaml
import tools.env_passthrough as _ep_mod
from tools.env_passthrough import (
clear_env_passthrough,
get_all_passthrough,
is_env_passthrough,
register_env_passthrough,
)
@pytest.fixture(autouse=True)
def _clean_passthrough():
"""Ensure a clean passthrough state for every test."""
clear_env_passthrough()
_ep_mod._config_passthrough = None
yield
clear_env_passthrough()
_ep_mod._config_passthrough = None
class TestSkillScopedPassthrough:
def test_register_and_check(self):
assert not is_env_passthrough("TENOR_API_KEY")
register_env_passthrough(["TENOR_API_KEY"])
assert is_env_passthrough("TENOR_API_KEY")
def test_register_multiple(self):
register_env_passthrough(["FOO_TOKEN", "BAR_SECRET"])
assert is_env_passthrough("FOO_TOKEN")
assert is_env_passthrough("BAR_SECRET")
assert not is_env_passthrough("OTHER_KEY")
def test_clear(self):
register_env_passthrough(["TENOR_API_KEY"])
assert is_env_passthrough("TENOR_API_KEY")
clear_env_passthrough()
assert not is_env_passthrough("TENOR_API_KEY")
def test_get_all(self):
register_env_passthrough(["A_KEY", "B_TOKEN"])
result = get_all_passthrough()
assert "A_KEY" in result
assert "B_TOKEN" in result
def test_strips_whitespace(self):
register_env_passthrough([" SPACED_KEY "])
assert is_env_passthrough("SPACED_KEY")
def test_skips_empty(self):
register_env_passthrough(["", " ", "VALID_KEY"])
assert is_env_passthrough("VALID_KEY")
assert not is_env_passthrough("")
class TestConfigPassthrough:
def test_reads_from_config(self, tmp_path, monkeypatch):
config = {"terminal": {"env_passthrough": ["MY_CUSTOM_KEY", "ANOTHER_TOKEN"]}}
config_path = tmp_path / "config.yaml"
config_path.write_text(yaml.dump(config))
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
_ep_mod._config_passthrough = None
assert is_env_passthrough("MY_CUSTOM_KEY")
assert is_env_passthrough("ANOTHER_TOKEN")
assert not is_env_passthrough("UNRELATED_VAR")
def test_empty_config(self, tmp_path, monkeypatch):
config = {"terminal": {"env_passthrough": []}}
config_path = tmp_path / "config.yaml"
config_path.write_text(yaml.dump(config))
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
_ep_mod._config_passthrough = None
assert not is_env_passthrough("ANYTHING")
def test_missing_config_key(self, tmp_path, monkeypatch):
config = {"terminal": {"backend": "local"}}
config_path = tmp_path / "config.yaml"
config_path.write_text(yaml.dump(config))
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
_ep_mod._config_passthrough = None
assert not is_env_passthrough("ANYTHING")
def test_no_config_file(self, tmp_path, monkeypatch):
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
_ep_mod._config_passthrough = None
assert not is_env_passthrough("ANYTHING")
def test_union_of_skill_and_config(self, tmp_path, monkeypatch):
config = {"terminal": {"env_passthrough": ["CONFIG_KEY"]}}
config_path = tmp_path / "config.yaml"
config_path.write_text(yaml.dump(config))
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
_ep_mod._config_passthrough = None
register_env_passthrough(["SKILL_KEY"])
all_pt = get_all_passthrough()
assert "CONFIG_KEY" in all_pt
assert "SKILL_KEY" in all_pt
class TestExecuteCodeIntegration:
"""Verify that the passthrough is checked in execute_code's env filtering."""
def test_secret_substring_blocked_by_default(self):
"""TENOR_API_KEY should be blocked without passthrough."""
_SAFE_ENV_PREFIXES = ("PATH", "HOME", "USER", "LANG", "LC_", "TERM",
"TMPDIR", "TMP", "TEMP", "SHELL", "LOGNAME",
"XDG_", "PYTHONPATH", "VIRTUAL_ENV", "CONDA")
_SECRET_SUBSTRINGS = ("KEY", "TOKEN", "SECRET", "PASSWORD", "CREDENTIAL",
"PASSWD", "AUTH")
test_env = {"PATH": "/usr/bin", "TENOR_API_KEY": "test123", "HOME": "/home/user"}
child_env = {}
for k, v in test_env.items():
if is_env_passthrough(k):
child_env[k] = v
continue
if any(s in k.upper() for s in _SECRET_SUBSTRINGS):
continue
if any(k.startswith(p) for p in _SAFE_ENV_PREFIXES):
child_env[k] = v
assert "PATH" in child_env
assert "HOME" in child_env
assert "TENOR_API_KEY" not in child_env
def test_passthrough_allows_secret_through(self):
"""TENOR_API_KEY should pass through when registered."""
_SAFE_ENV_PREFIXES = ("PATH", "HOME", "USER", "LANG", "LC_", "TERM",
"TMPDIR", "TMP", "TEMP", "SHELL", "LOGNAME",
"XDG_", "PYTHONPATH", "VIRTUAL_ENV", "CONDA")
_SECRET_SUBSTRINGS = ("KEY", "TOKEN", "SECRET", "PASSWORD", "CREDENTIAL",
"PASSWD", "AUTH")
register_env_passthrough(["TENOR_API_KEY"])
test_env = {"PATH": "/usr/bin", "TENOR_API_KEY": "test123", "HOME": "/home/user"}
child_env = {}
for k, v in test_env.items():
if is_env_passthrough(k):
child_env[k] = v
continue
if any(s in k.upper() for s in _SECRET_SUBSTRINGS):
continue
if any(k.startswith(p) for p in _SAFE_ENV_PREFIXES):
child_env[k] = v
assert "PATH" in child_env
assert "HOME" in child_env
assert "TENOR_API_KEY" in child_env
assert child_env["TENOR_API_KEY"] == "test123"
class TestTerminalIntegration:
"""Verify that the passthrough is checked in terminal's env sanitizers."""
def test_blocklisted_var_blocked_by_default(self):
from tools.environments.local import _sanitize_subprocess_env, _HERMES_PROVIDER_ENV_BLOCKLIST
# Pick a var we know is in the blocklist
blocked_var = next(iter(_HERMES_PROVIDER_ENV_BLOCKLIST))
env = {blocked_var: "secret_value", "PATH": "/usr/bin"}
result = _sanitize_subprocess_env(env)
assert blocked_var not in result
assert "PATH" in result
def test_passthrough_allows_blocklisted_var(self):
from tools.environments.local import _sanitize_subprocess_env, _HERMES_PROVIDER_ENV_BLOCKLIST
blocked_var = next(iter(_HERMES_PROVIDER_ENV_BLOCKLIST))
register_env_passthrough([blocked_var])
env = {blocked_var: "secret_value", "PATH": "/usr/bin"}
result = _sanitize_subprocess_env(env)
assert blocked_var in result
assert result[blocked_var] == "secret_value"
def test_make_run_env_passthrough(self, monkeypatch):
from tools.environments.local import _make_run_env, _HERMES_PROVIDER_ENV_BLOCKLIST
blocked_var = next(iter(_HERMES_PROVIDER_ENV_BLOCKLIST))
monkeypatch.setenv(blocked_var, "secret_value")
# Without passthrough — blocked
result_before = _make_run_env({})
assert blocked_var not in result_before
# With passthrough — allowed
register_env_passthrough([blocked_var])
result_after = _make_run_env({})
assert blocked_var in result_after
|