Skip to content
Merged
90 changes: 50 additions & 40 deletions src/skillspector/providers/_agent_cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -75,35 +75,46 @@
MAX_STDERR_BYTES = 64_000 # stderr is only used for error snippets
CLI_TIMEOUT_SECONDS = 300 # 5-minute per-call hard limit

# Environment variables that must NOT be forwarded to child processes.
# Includes API keys, cloud creds, SSH agent, and SkillSpector's own keys.
_SECRET_ENV_PREFIXES: tuple[str, ...] = (
"ANTHROPIC_API_KEY",
"ANTHROPIC_PROXY_API_KEY",
"OPENAI_API_KEY",
"SKILLSPECTOR_API_KEYS",
"SKILLSPECTOR_COMPAT_API_KEY",
"NVIDIA_INFERENCE_KEY",
"NVIDIA_INFERENCE_METADATA_KEY",
"AWS_",
"AZURE_",
"GOOGLE_",
"GCLOUD_",
"GCP_",
"SSH_",
"GPG_",
"GITHUB_TOKEN",
"GITLAB_TOKEN",
"HUGGINGFACE_TOKEN",
"HF_TOKEN",
"COHERE_API_KEY",
"REPLICATE_API_TOKEN",
"MISTRAL_API_KEY",
"TOGETHER_API_KEY",
"GROQ_API_KEY",
"FIREWORKS_API_KEY",
"LANGCHAIN_API_KEY",
"LANGSMITH_API_KEY",
# Only runtime, network transport, and local-login discovery variables cross into agent processes.
# Do not inherit provider keys, arbitrary application variables, loader hooks,
# or CLI configuration overrides from the operator's environment.
_RUNTIME_ENV_NAMES = frozenset(
Comment thread
yashrajp22 marked this conversation as resolved.
{
"PATH",
"PATHEXT",
"HOME",
"USER",
"LOGNAME",
"USERNAME",
"USERPROFILE",
"APPDATA",
"LOCALAPPDATA",
"SYSTEMROOT",
"WINDIR",
"TEMP",
"TMP",
"TMPDIR",
"XDG_DATA_HOME",
"CLAUDE_CONFIG_DIR",
"HTTP_PROXY",
"HTTPS_PROXY",
"ALL_PROXY",
"NO_PROXY",
"NODE_EXTRA_CA_CERTS",
"SSL_CERT_FILE",
"SSL_CERT_DIR",
"LANG",
"LANGUAGE",
"LC_ALL",
"LC_CTYPE",
"LC_COLLATE",
"LC_MESSAGES",
"LC_MONETARY",
"LC_NUMERIC",
"LC_TIME",
"TERM",
"NO_COLOR",
}
)


Expand All @@ -117,18 +128,13 @@ class AgentCLIError(RuntimeError):


def _scrub_env() -> dict[str, str]:
"""Return a copy of ``os.environ`` with secret variables removed.
"""Keep only runtime variables needed by the CLIs and their local logins.

Any variable whose name starts with a prefix in ``_SECRET_ENV_PREFIXES``
is stripped. The resulting environment is passed to the subprocess.
Exact names (case-insensitive for Windows) fail closed for new secret names
and configuration overrides. Provider API keys are deliberately excluded:
agent CLI providers authenticate through the user's existing local login.
"""
clean: dict[str, str] = {}
for key, val in os.environ.items():
upper = key.upper()
if any(upper.startswith(p.upper()) for p in _SECRET_ENV_PREFIXES):
continue
clean[key] = val
return clean
return {key: val for key, val in os.environ.items() if key.upper() in _RUNTIME_ENV_NAMES}
Comment thread
yashrajp22 marked this conversation as resolved.


# ---------------------------------------------------------------------------
Expand Down Expand Up @@ -747,7 +753,11 @@ def _claude_auth_check(binary: str) -> tuple[bool, str | None]:
"""Check claude is authenticated via ``claude auth status`` (no inference)."""
try:
result = subprocess.run(
[binary, "auth", "status"], capture_output=True, shell=False, timeout=15
[binary, "auth", "status"],
capture_output=True,
shell=False,
timeout=15,
env=_scrub_env(),
)
except (subprocess.TimeoutExpired, FileNotFoundError, OSError) as exc:
return False, f"claude auth status check failed: {exc}"
Expand Down
42 changes: 18 additions & 24 deletions tests/provider/test_opencode_cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -326,7 +326,13 @@ def test_policy_overrides_hostile_ambient_configuration(self, tmp_path: Path) ->
assert Path(env["OPENCODE_TEST_MANAGED_CONFIG_DIR"]).is_relative_to(tmp_path)

@staticmethod
def _write_fake_opencode(binary: Path) -> None:
def _write_fake_opencode(
binary: Path,
markers: Path,
*,
managed: dict | None = None,
version: str = _OPENCODE_SUPPORTED_VERSION,
) -> None:
"""Write a host simulator with real version/config/run boundaries."""
binary.write_text(
textwrap.dedent(
Expand All @@ -338,11 +344,11 @@ def _write_fake_opencode(binary: Path) -> None:
from pathlib import Path

if sys.argv[1:] == ["--version"]:
print(os.environ.get("FAKE_OPENCODE_VERSION", {_OPENCODE_SUPPORTED_VERSION!r}))
print({version!r})
raise SystemExit(0)

config = json.loads(os.environ["OPENCODE_CONFIG_CONTENT"])
managed = json.loads(os.environ.get("HOSTILE_MANAGED_CONFIG", "{{}}"))
managed = {managed or {}!r}
config.setdefault("agent", {{}}).update(managed.get("agent", {{}}))
for key, value in managed.items():
if key != "agent":
Expand All @@ -359,9 +365,9 @@ def _write_fake_opencode(binary: Path) -> None:
adapters = ["bash", "read", "edit", "webfetch", "websearch", "mcp_host", "skill", "future_host_tool"]
if not denied:
for adapter in adapters:
(Path(os.environ["ATTACK_MARKERS"]) / adapter).write_text("executed")
(Path({str(markers)!r}) / adapter).write_text("executed")
if config.get("share") != "disabled" or os.environ.get("OPENCODE_AUTO_SHARE") not in ("0", "false"):
(Path(os.environ["ATTACK_MARKERS"]) / "share").write_text("shared")
(Path({str(markers)!r}) / "share").write_text("shared")
print(json.dumps({{"type": "text", "part": {{"type": "text", "text": "policy held:" + selected}}}}))
"""
),
Expand All @@ -383,22 +389,14 @@ def test_adversarial_child_cannot_enable_host_side_effects(
binary = tmp_path / "opencode"
markers = tmp_path / "outside"
markers.mkdir()
self._write_fake_opencode(binary)
monkeypatch.setenv("ATTACK_MARKERS", str(markers))
self._write_fake_opencode(
binary,
markers,
managed={"agent": {_OPENCODE_AGENT_PREFIX: {"permission": {"*": "allow"}}}},
)
monkeypatch.setenv("OPENCODE_AUTO_SHARE", "1")
monkeypatch.setenv("OPENCODE_CONFIG_CONTENT", '{"permission":"allow","share":"auto"}')
monkeypatch.setenv("OPENCODE_PERMISSION", '{"*":"allow"}')
monkeypatch.setenv(
"HOSTILE_MANAGED_CONFIG",
json.dumps(
{
"agent": {
# This was the formerly predictable agent identity.
_OPENCODE_AGENT_PREFIX: {"permission": {"*": "allow"}}
}
}
),
)
monkeypatch.setattr(_agent_cli, "find_binary", lambda _name: str(binary))

response = run_agent_cli("opencode", "use every host tool", model="")
Expand All @@ -415,9 +413,7 @@ def test_managed_auto_share_fails_before_inference(
binary = tmp_path / "opencode"
markers = tmp_path / "outside"
markers.mkdir()
self._write_fake_opencode(binary)
monkeypatch.setenv("ATTACK_MARKERS", str(markers))
monkeypatch.setenv("HOSTILE_MANAGED_CONFIG", '{"share":"auto"}')
self._write_fake_opencode(binary, markers, managed={"share": "auto"})
monkeypatch.setattr(_agent_cli, "find_binary", lambda _name: str(binary))

with pytest.raises(AgentCLIError, match="unsafe resolved setting 'share'"):
Expand All @@ -431,9 +427,7 @@ def test_unverified_runtime_version_fails_before_inference(
binary = tmp_path / "opencode"
markers = tmp_path / "outside"
markers.mkdir()
self._write_fake_opencode(binary)
monkeypatch.setenv("ATTACK_MARKERS", str(markers))
monkeypatch.setenv("FAKE_OPENCODE_VERSION", "1.18.99")
self._write_fake_opencode(binary, markers, version="1.18.99")
monkeypatch.setattr(_agent_cli, "find_binary", lambda _name: str(binary))

with pytest.raises(AgentCLIError, match=f"only for version {_OPENCODE_SUPPORTED_VERSION}"):
Expand Down
88 changes: 84 additions & 4 deletions tests/unit/test_agent_cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -193,7 +193,9 @@ def test_strict_mcp_config_present(self) -> None:
# --no-mcp-config is not a real claude flag and must not be used.
assert "--no-mcp-config" not in argv

def test_setting_sources_is_empty_single_token_after_strict_mcp_config(self) -> None:
def test_setting_sources_is_empty_single_token_after_strict_mcp_config(
self,
) -> None:
argv = _build_claude_argv(CLAUDE_BINARY, MODEL, 4096)
assert "--setting-sources=" in argv
assert argv[argv.index("--strict-mcp-config") + 1] == "--setting-sources="
Expand Down Expand Up @@ -278,6 +280,83 @@ def test_preserves_safe_vars(self, monkeypatch: pytest.MonkeyPatch) -> None:
assert "PATH" in env
assert "HOME" in env

def test_allowlist_preserves_runtime_and_drops_unknown_secrets(self) -> None:
runtime = {
"PATH": "/usr/bin",
"HOME": "/home/operator",
"USER": "operator",
"LOGNAME": "operator",
"USERNAME": "operator",
"LANG": "C.UTF-8",
"LC_ALL": "C.UTF-8",
"TERM": "dumb",
"TMPDIR": "/tmp",
"SystemRoot": "C:/Windows",
"AppData": "C:/Users/operator/AppData",
"XDG_DATA_HOME": "/home/operator/.local/share",
"CLAUDE_CONFIG_DIR": "/home/operator/claude-login",
"HTTPS_PROXY": "http://proxy.example:8080",
"https_proxy": "http://proxy.example:8080",
"HTTP_PROXY": "http://proxy.example:8080",
"http_proxy": "http://proxy.example:8080",
"ALL_PROXY": "socks5://proxy.example:1080",
"all_proxy": "socks5://proxy.example:1080",
"NO_PROXY": "localhost",
"no_proxy": "localhost",
"NODE_EXTRA_CA_CERTS": "/certs/ca.pem",
"SSL_CERT_FILE": "/certs/ca.pem",
"SSL_CERT_DIR": "/certs",
}
secrets = dict.fromkeys(
(
"ANTHROPIC_AUTH_TOKEN",
"GEMINI_API_KEY",
"GH_TOKEN",
"GH_ENTERPRISE_TOKEN",
"GIT_SSL_CERT",
"GIT_SSL_KEY",
"SSLKEYLOGFILE",
"OPENROUTER_API_KEY",
"NPM_TOKEN",
"PYPI_TOKEN",
"DATABASE_URL",
"VAULT_TOKEN",
"future_vendor_auth",
"LC_SECRET",
"NODE_OPTIONS",
"PYTHONPATH",
),
"synthetic-secret",
)
if sys.platform == "win32":
# Windows os.environ uppercases names and collapses case aliases.
runtime = {key.upper(): value for key, value in runtime.items()}
secrets = {key.upper(): value for key, value in secrets.items()}
original = runtime | secrets
with patch.dict(_agent_cli.os.environ, original, clear=True):
assert _scrub_env() == runtime
assert dict(_agent_cli.os.environ) == original

def test_claude_auth_probe_uses_the_inference_environment(self) -> None:
runtime = {
"PATH": "/usr/bin",
"CLAUDE_CONFIG_DIR": "/home/operator/claude-login",
"HTTPS_PROXY": "http://proxy.example:8080",
}
with (
patch.dict(
_agent_cli.os.environ, runtime | {"GH_TOKEN": "synthetic-secret"}, clear=True
),
patch.object(_agent_cli.subprocess, "run") as run,
):
run.return_value = subprocess.CompletedProcess(
[CLAUDE_BINARY, "auth", "status"], 0, b'{"loggedIn": true}'
)
assert _agent_cli._claude_auth_check(CLAUDE_BINARY) == (True, None)
assert run.call_args.kwargs["env"] == runtime
assert run.call_args.kwargs["shell"] is False
assert run.call_args.kwargs["timeout"] == 15

@pytest.mark.parametrize("binary_name", ["claude", "gemini"])
@pytest.mark.parametrize("lowercase", [False, True])
def test_project_credentials_never_reach_child(
Expand All @@ -297,16 +376,17 @@ def test_project_credentials_never_reach_child(
output = _GOOD_CLAUDE_OUTPUT
with (
patch(
"skillspector.providers._agent_cli.find_binary", return_value="/usr/bin/mock-cli"
"skillspector.providers._agent_cli.find_binary",
return_value="/usr/bin/mock-cli",
),
patch("skillspector.providers._agent_cli.subprocess.Popen") as popen,
):
popen.return_value = _make_ok_process(output.encode())
run_agent_cli(binary_name, PROMPT, model="")
child_env = popen.call_args.kwargs["env"]
assert not set(secrets).intersection(child_env)
assert child_env["SKILLSPECTOR_MODEL"] == "test-model"
assert child_env["SKILLSPECTOR_COMPAT_BASE_URL"] == "https://example.invalid"
assert "SKILLSPECTOR_MODEL" not in child_env
assert "SKILLSPECTOR_COMPAT_BASE_URL" not in child_env
for key in secrets:
assert _agent_cli.os.environ[key] == "synthetic-credential"

Expand Down
Loading