feat(shell): control subprocess environment to stop secret inheritance (#284)

* feat(shell): control subprocess environment to stop secret inheritance

Shell spawned every command with the inherited parent environment, so an
agent running model-written commands in a sandbox could read the host's
LLM API keys and tokens. This blocked replacing Pydanty's hand-rolled
runner with Shell.

Add two composable controls: `env` replaces inheritance entirely (a hard
boundary -- the subprocess sees only what you pass), and
`denied_env_patterns` strips glob-matched names from the base environment
(mirrors `denied_commands`). When both are set, patterns also filter the
explicit `env`. The default stays inherit-everything. Export
`LLM_API_KEY_ENV_PATTERNS` for the common provider-credential denylist;
opt-in, since stripping env silently would break agents that rely on
inherited credentials.

Closes #281

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* docs(shell): tighten env-control wording after adversarial review

Three-lens self-review (API breadth, correctness, docs/tests) surfaced
doc-precision gaps and one missing end-to-end test; no correctness defects.

- denied_env_patterns matches by glob, not exact match like denied_commands:
  correct the "mirrors denied_commands" claim to name the real parallel
  (the denied_* naming convention).
- LLM_API_KEY_ENV_PATTERNS targets LLM credentials only and uses coarse
  prefixes (GOOGLE_* also strips GOOGLE_APPLICATION_CREDENTIALS); say so, and
  note it does not cover other host secrets like LOGFIRE/GitHub tokens.
- env is enforced at spawn; clarify that with denied_env_patterns it is the
  resolved (filtered) env, and warn that stripping PATH/HOME breaks command
  resolution.
- Add an end-to-end test proving env and denied_env_patterns compose at spawn.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: David SF <david.sanchez@pydantic.dev>
Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

pydantic_ai_harness/__init__.py

@@ -6,11 +6,12 @@ if TYPE_CHECKING:
     from .code_mode import CodeMode
     from .filesystem import FileSystem
     from .logfire import ManagedPrompt
-    from .shell import Shell
+    from .shell import LLM_API_KEY_ENV_PATTERNS, Shell
 
 __all__ = [
     'CodeMode',
     'FileSystem',
+    'LLM_API_KEY_ENV_PATTERNS',
     'ManagedPrompt',
     'Shell',
 ]
@@ -33,4 +34,8 @@ def __getattr__(name: str) -> object:
         from .shell import Shell
 
         return Shell
+    if name == 'LLM_API_KEY_ENV_PATTERNS':
+        from .shell import LLM_API_KEY_ENV_PATTERNS
+
+        return LLM_API_KEY_ENV_PATTERNS
     raise AttributeError(f'module {__name__!r} has no attribute {name!r}')

pydantic_ai_harness/shell/README.md

@@ -63,6 +63,55 @@ check.
 > For hard guarantees, run the agent inside OS-level isolation -- a container or
 > sandbox.
 
+## Environment control
+
+By default a spawned command inherits the agent process's full environment. In a
+sandbox that holds LLM API keys, tokens, or other secrets, a command the model
+writes can read them. Two fields control what the subprocess sees:
+
+| Field | Effect |
+|---|---|
+| `env` | Explicit environment that replaces inheritance entirely. The subprocess sees exactly these variables and nothing else. |
+| `denied_env_patterns` | Glob patterns (`fnmatch`) for variable names stripped from the base environment. Mirrors `denied_commands`. |
+
+`env` is a hard boundary: set it and inherited secrets cannot reach the
+subprocess at all (you supply `PATH` and anything else the command needs).
+`denied_env_patterns` is a denylist over the inherited environment -- lighter to
+configure when you only need to drop a few known-sensitive names. The two
+compose: when both are set, patterns also filter the explicit `env`. Leaving
+both unset preserves the inherit-everything default.
+
+```python
+from pydantic_ai_harness import Shell
+from pydantic_ai_harness.shell import LLM_API_KEY_ENV_PATTERNS
+
+# Strip provider credentials from the inherited environment.
+Shell(cwd='./repo', denied_env_patterns=LLM_API_KEY_ENV_PATTERNS)
+
+# Or hand the subprocess a fixed environment, inheriting nothing.
+import os
+Shell(cwd='./repo', env={'PATH': os.environ['PATH'], 'HOME': os.environ['HOME']})
+```
+
+`LLM_API_KEY_ENV_PATTERNS` covers common provider prefixes (`ANTHROPIC_*`,
+`OPENAI_*`, `OPENROUTER_*`, `GOOGLE_*`, `GEMINI_*`, `GATEWAY_*`) plus
+`PYDANTIC_AI_GATEWAY_API_KEY`. It targets LLM credentials only -- it does not
+cover other host secrets (a `LOGFIRE_TOKEN`, a GitHub token, cloud
+credentials), and its prefixes are coarse, so `GOOGLE_*` also strips
+non-credential vars like `GOOGLE_APPLICATION_CREDENTIALS`. Treat it as a
+starting point and add your own patterns. It is not the default: stripping
+environment variables silently would break agents that rely on inherited
+credentials, so it is opt-in.
+
+`env` is enforced at spawn, not applied as a post-hoc filter on a running
+process: the subprocess starts with exactly the resolved environment (your
+`env`, minus anything `denied_env_patterns` removes from it). That makes it a
+real boundary, unlike the best-effort command denylist. The flip side is that a
+pattern broad enough to strip `PATH` or `HOME`, or an `env` that omits them, can
+break command resolution. External commands may still run via the shell's
+built-in default `PATH` on some systems, but don't rely on it -- set `PATH`
+explicitly when you replace the environment.
+
 ## Background processes
 
 `start_command` writes stdout/stderr to temp files and returns a short ID. Use
@@ -96,6 +145,8 @@ Shell(
     max_output_chars=50_000,       # output cap returned to the model
     persist_cwd=False,             # make cd sticky across calls
     allow_interactive=False,       # allow TTY-style commands
+    env=None,                      # explicit env, replacing inheritance (None = inherit)
+    denied_env_patterns=[],        # glob patterns stripped from the inherited env
 )
 ```
 

pydantic_ai_harness/shell/__init__.py

@@ -1,6 +1,6 @@
 """Shell capability: gives agents configurable command execution."""
 
-from pydantic_ai_harness.shell._capability import Shell
+from pydantic_ai_harness.shell._capability import LLM_API_KEY_ENV_PATTERNS, Shell
 from pydantic_ai_harness.shell._toolset import ShellToolset
 
-__all__ = ['Shell', 'ShellToolset']
+__all__ = ['LLM_API_KEY_ENV_PATTERNS', 'Shell', 'ShellToolset']

pydantic_ai_harness/shell/_capability.py

@@ -2,7 +2,7 @@
 
 from __future__ import annotations
 
-from collections.abc import Sequence
+from collections.abc import Mapping, Sequence
 from dataclasses import dataclass, field
 from pathlib import Path
 
@@ -25,6 +25,24 @@ _DEFAULT_DENIED_COMMANDS: list[str] = [
     'init',
 ]
 
+LLM_API_KEY_ENV_PATTERNS: tuple[str, ...] = (
+    'ANTHROPIC_*',
+    'GATEWAY_*',
+    'GEMINI_*',
+    'GOOGLE_*',
+    'OPENAI_*',
+    'OPENROUTER_*',
+    'PYDANTIC_AI_GATEWAY_API_KEY',
+)
+"""Glob patterns for common LLM provider credentials, for `denied_env_patterns`.
+
+Pass these when an agent runs untrusted commands that must not read the host's
+LLM API keys. Covers provider prefixes only -- not other host secrets, and the
+prefixes are coarse (`GOOGLE_*` also strips `GOOGLE_APPLICATION_CREDENTIALS`),
+so treat it as a starting point. Not a default: stripping env silently would
+break agents that rely on inherited credentials, so opt in explicitly.
+"""
+
 
 @dataclass
 class Shell(AbstractCapability[AgentDepsT]):
@@ -62,6 +80,26 @@ class Shell(AbstractCapability[AgentDepsT]):
     allow_interactive: bool = False
     """If True, allow interactive commands (vi, nano, ssh, etc.). Blocked by default."""
 
+    env: Mapping[str, str] | None = None
+    """Explicit environment for spawned subprocesses, replacing inheritance.
+
+    When `None` (default) the subprocess inherits the parent environment. Set
+    this to a fixed mapping to start subprocesses with exactly these variables
+    and nothing else -- a hard boundary that keeps host secrets (LLM API keys,
+    tokens) out of commands the agent runs.
+    """
+
+    denied_env_patterns: Sequence[str] = field(default_factory=list[str])
+    """Glob patterns for environment variable names to strip before spawning.
+
+    Follows the `denied_*` naming convention but matches by glob (`fnmatch`,
+    e.g. `OPENAI_*`), since env secrets cluster by prefix -- unlike
+    `denied_commands`, which matches executable names exactly. Names matching
+    any pattern are removed from the base environment; applied on top of `env`
+    when both are set, so patterns filter an explicit `env` too. See
+    `LLM_API_KEY_ENV_PATTERNS` for a ready-made provider-credential denylist.
+    """
+
     def get_toolset(self) -> AgentToolset[AgentDepsT]:
         """Build and return the shell toolset."""
         return ShellToolset[AgentDepsT](
@@ -73,4 +111,6 @@ class Shell(AbstractCapability[AgentDepsT]):
             max_output_chars=self.max_output_chars,
             persist_cwd=self.persist_cwd,
             allow_interactive=self.allow_interactive,
+            env=self.env,
+            denied_env_patterns=self.denied_env_patterns,
         )

pydantic_ai_harness/shell/_toolset.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import fnmatch
 import functools
 import os
 import re
@@ -10,7 +11,7 @@ import signal
 import subprocess
 import tempfile
 import uuid
-from collections.abc import Awaitable, Callable, Sequence
+from collections.abc import Awaitable, Callable, Mapping, Sequence
 from pathlib import Path
 from typing import Any, Concatenate, ParamSpec
 
@@ -102,6 +103,8 @@ class ShellToolset(FunctionToolset[AgentDepsT]):
         max_output_chars: int,
         persist_cwd: bool,
         allow_interactive: bool,
+        env: Mapping[str, str] | None = None,
+        denied_env_patterns: Sequence[str] = (),
     ) -> None:
         super().__init__()
         self._cwd = cwd.resolve()
@@ -115,6 +118,8 @@ class ShellToolset(FunctionToolset[AgentDepsT]):
         self._max_output_chars = max_output_chars
         self._persist_cwd = persist_cwd
         self._allow_interactive = allow_interactive
+        self._env = dict(env) if env is not None else None
+        self._denied_env_patterns = list(denied_env_patterns)
         self._background: dict[str, _BackgroundProcess] = {}
 
         if self._allowed_commands and self._denied_commands:
@@ -143,8 +148,31 @@ class ShellToolset(FunctionToolset[AgentDepsT]):
             max_output_chars=self._max_output_chars,
             persist_cwd=self._persist_cwd,
             allow_interactive=self._allow_interactive,
+            env=self._env,
+            denied_env_patterns=self._denied_env_patterns,
         )
 
+    def _resolve_env(self) -> dict[str, str] | None:
+        """Compute the environment passed to spawned subprocesses.
+
+        Returns `None` -- meaning the subprocess inherits the parent env -- only
+        when neither `env` nor `denied_env_patterns` is configured, so the
+        default behavior is unchanged. An explicit `env` replaces inheritance
+        entirely; `denied_env_patterns` then strips matching names (glob, via
+        `fnmatch`) from whichever base applies, so the two compose: patterns
+        filter an explicit `env` just as they filter the inherited environment.
+        """
+        if self._env is None and not self._denied_env_patterns:
+            return None
+        base = dict(self._env) if self._env is not None else dict(os.environ)
+        if not self._denied_env_patterns:
+            return base
+        return {
+            name: value
+            for name, value in base.items()
+            if not any(fnmatch.fnmatchcase(name, pattern) for pattern in self._denied_env_patterns)
+        }
+
     async def __aexit__(self, *args: Any) -> None:
         """Terminate all remaining background processes and clean up temp files."""
         for bg in self._background.values():
@@ -299,6 +327,7 @@ class ShellToolset(FunctionToolset[AgentDepsT]):
                 stdout=subprocess.PIPE,
                 stderr=subprocess.PIPE,
                 start_new_session=True,
+                env=self._resolve_env(),
             )
             stdout_chunks: list[bytes] = []
             stderr_chunks: list[bytes] = []
@@ -379,6 +408,7 @@ class ShellToolset(FunctionToolset[AgentDepsT]):
                 stdout=stdout_file,
                 stderr=stderr_file,
                 start_new_session=True,
+                env=self._resolve_env(),
             )
         except BaseException:
             stdout_file.close()

tests/shell/test_shell.py

@@ -5,6 +5,7 @@ from __future__ import annotations
 import os
 import shlex
 import sys
+from collections.abc import Mapping, Sequence
 from pathlib import Path
 from unittest.mock import MagicMock, patch
 
@@ -15,13 +16,39 @@ from pydantic_ai.exceptions import ModelRetry
 from pydantic_ai.models.test import TestModel
 from pydantic_ai.usage import RunUsage
 
-from pydantic_ai_harness.shell import Shell
+from pydantic_ai_harness.shell import LLM_API_KEY_ENV_PATTERNS, Shell
 from pydantic_ai_harness.shell._toolset import (
     ShellToolset,
     _is_interactive_command,
 )
 
 
+def _env_toolset(
+    shell_dir: Path,
+    *,
+    env: Mapping[str, str] | None = None,
+    denied_env_patterns: Sequence[str] = (),
+) -> ShellToolset[None]:
+    """Build a ShellToolset wired for env-control tests, with safe defaults."""
+    return ShellToolset(
+        cwd=shell_dir,
+        allowed_commands=[],
+        denied_commands=[],
+        denied_operators=[],
+        default_timeout=10.0,
+        max_output_chars=50_000,
+        persist_cwd=False,
+        allow_interactive=False,
+        env=env,
+        denied_env_patterns=denied_env_patterns,
+    )
+
+
+def _read_env_var(name: str) -> str:
+    """Shell command that prints an env var's value, or ABSENT if unset."""
+    return f'{sys.executable} -c "import os; print(os.environ.get({name!r}, \'ABSENT\'))"'
+
+
 def _run_context() -> RunContext[None]:
     """Minimal `RunContext` for invoking `for_run` directly in tests."""
     return RunContext[None](
@@ -1414,3 +1441,154 @@ class TestStopCommandAlreadyFinished:
         result = await ts.stop_command(command_id)
         assert '[stopped:' in result
         assert '[exit code:' not in result
+
+
+class TestResolveEnv:
+    """Unit coverage for the env-resolution branches."""
+
+    def test_inherits_when_unconfigured(self, shell_dir: Path) -> None:
+        # Neither env nor patterns set -> None, so the subprocess inherits.
+        assert _env_toolset(shell_dir)._resolve_env() is None
+
+    def test_explicit_env_replaces(self, shell_dir: Path) -> None:
+        resolved = _env_toolset(shell_dir, env={'FOO': 'bar'})._resolve_env()
+        assert resolved == {'FOO': 'bar'}
+
+    def test_explicit_empty_env_is_not_inheritance(self, shell_dir: Path) -> None:
+        # {} is a hard boundary (no vars), distinct from None (inherit all).
+        assert _env_toolset(shell_dir, env={})._resolve_env() == {}
+
+    def test_patterns_strip_from_inherited(self, shell_dir: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.setenv('OPENAI_API_KEY', 'secret')
+        monkeypatch.setenv('SAFE_VAR', 'keep')
+        resolved = _env_toolset(shell_dir, denied_env_patterns=['OPENAI_*'])._resolve_env()
+        assert resolved is not None
+        assert 'OPENAI_API_KEY' not in resolved
+        assert resolved.get('SAFE_VAR') == 'keep'
+
+    def test_patterns_strip_from_explicit_env(self, shell_dir: Path) -> None:
+        resolved = _env_toolset(
+            shell_dir,
+            env={'OPENAI_API_KEY': 'secret', 'PATH': '/usr/bin'},
+            denied_env_patterns=['OPENAI_*'],
+        )._resolve_env()
+        assert resolved == {'PATH': '/usr/bin'}
+
+    def test_patterns_no_match_keeps_base(self, shell_dir: Path) -> None:
+        resolved = _env_toolset(
+            shell_dir,
+            env={'FOO': 'bar'},
+            denied_env_patterns=['OPENAI_*'],
+        )._resolve_env()
+        assert resolved == {'FOO': 'bar'}
+
+    def test_pattern_match_is_case_sensitive(self, shell_dir: Path) -> None:
+        # Env var names are case-sensitive on POSIX; lowercase must not match.
+        resolved = _env_toolset(
+            shell_dir,
+            env={'openai_api_key': 'secret'},
+            denied_env_patterns=['OPENAI_*'],
+        )._resolve_env()
+        assert resolved == {'openai_api_key': 'secret'}
+
+
+class TestEnvControlExecution:
+    """End-to-end: the resolved env actually reaches spawned subprocesses."""
+
+    async def test_explicit_env_seen_by_command(self, shell_dir: Path) -> None:
+        ts = _env_toolset(shell_dir, env={'MY_TOKEN': 'present', 'PATH': os.environ['PATH']})
+        result = await ts.run_command(_read_env_var('MY_TOKEN'))
+        assert 'present' in result
+
+    async def test_explicit_env_hides_inherited_secret(self, shell_dir: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.setenv('OPENROUTER_API_KEY', 'leak-me')
+        ts = _env_toolset(shell_dir, env={'PATH': os.environ['PATH']})
+        result = await ts.run_command(_read_env_var('OPENROUTER_API_KEY'))
+        assert 'ABSENT' in result
+        assert 'leak-me' not in result
+
+    async def test_denied_pattern_strips_inherited_secret(
+        self, shell_dir: Path, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        monkeypatch.setenv('ANTHROPIC_API_KEY', 'leak-me')
+        ts = _env_toolset(shell_dir, denied_env_patterns=['ANTHROPIC_*'])
+        result = await ts.run_command(_read_env_var('ANTHROPIC_API_KEY'))
+        assert 'ABSENT' in result
+        assert 'leak-me' not in result
+
+    async def test_unstripped_inherited_var_still_visible(
+        self, shell_dir: Path, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        # A var not matched by any pattern is still inherited as before.
+        monkeypatch.setenv('HARNESS_KEEP', 'visible')
+        ts = _env_toolset(shell_dir, denied_env_patterns=['ANTHROPIC_*'])
+        result = await ts.run_command(_read_env_var('HARNESS_KEEP'))
+        assert 'visible' in result
+
+    async def test_default_inherits_parent_env(self, shell_dir: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+        # Backward compatible: with no env control, inherited vars pass through.
+        monkeypatch.setenv('HARNESS_INHERITED', 'yes')
+        ts = _env_toolset(shell_dir)
+        result = await ts.run_command(_read_env_var('HARNESS_INHERITED'))
+        assert 'yes' in result
+
+    async def test_env_and_patterns_compose_at_spawn(self, shell_dir: Path) -> None:
+        # Both set: a pattern strips a key from the explicit env, the rest survives.
+        ts = _env_toolset(
+            shell_dir,
+            env={'SECRET_KEY': 'leak-me', 'KEEP_VAR': 'kept', 'PATH': os.environ['PATH']},
+            denied_env_patterns=['SECRET_*'],
+        )
+        stripped = await ts.run_command(_read_env_var('SECRET_KEY'))
+        assert 'ABSENT' in stripped
+        assert 'leak-me' not in stripped
+        survived = await ts.run_command(_read_env_var('KEEP_VAR'))
+        assert 'kept' in survived
+
+    async def test_background_command_honors_env(self, shell_dir: Path) -> None:
+        ts = _env_toolset(shell_dir, env={'BG_TOKEN': 'bg-present', 'PATH': os.environ['PATH']})
+        start_result = await ts.start_command(_read_env_var('BG_TOKEN'))
+        command_id = _parse_command_id(start_result)
+        await anyio.sleep(0.5)
+        stop_result = await ts.stop_command(command_id)
+        assert 'bg-present' in stop_result
+
+
+class TestEnvControlPropagation:
+    """The capability and `for_run` carry env control through unchanged."""
+
+    async def test_for_run_propagates_env(self, shell_dir: Path) -> None:
+        ts = _env_toolset(shell_dir, env={'FOO': 'bar'}, denied_env_patterns=['OPENAI_*'])
+        run_ts = await ts.for_run(_run_context())
+        assert isinstance(run_ts, ShellToolset)
+        assert run_ts._resolve_env() == {'FOO': 'bar'}
+
+    def test_capability_defaults_inherit(self) -> None:
+        shell = Shell()
+        assert shell.env is None
+        assert list(shell.denied_env_patterns) == []
+
+    def test_capability_passes_env_to_toolset(self, tmp_path: Path) -> None:
+        shell = Shell(
+            cwd=tmp_path,
+            env={'FOO': 'bar'},
+            denied_env_patterns=['OPENAI_*'],
+        )
+        toolset = shell.get_toolset()
+        assert isinstance(toolset, ShellToolset)
+        assert toolset._resolve_env() == {'FOO': 'bar'}
+
+    def test_llm_pattern_constant_strips_provider_keys(self, tmp_path: Path) -> None:
+        shell = Shell(cwd=tmp_path, denied_env_patterns=list(LLM_API_KEY_ENV_PATTERNS))
+        toolset = shell.get_toolset()
+        assert isinstance(toolset, ShellToolset)
+        resolved = toolset._resolve_env()
+        assert resolved is not None
+        # None of the provider-credential prefixes survive.
+        leaked = {
+            name
+            for name in resolved
+            if name.startswith(('ANTHROPIC_', 'OPENAI_', 'OPENROUTER_', 'GEMINI_', 'GOOGLE_', 'GATEWAY_'))
+            or name == 'PYDANTIC_AI_GATEWAY_API_KEY'
+        }
+        assert leaked == set()

tests/test_placeholder.py

@@ -27,6 +27,13 @@ def test_lazy_import_shell():
     assert hasattr(Shell, 'get_toolset')
 
 
+def test_lazy_import_llm_api_key_env_patterns():
+    from pydantic_ai_harness import LLM_API_KEY_ENV_PATTERNS
+
+    assert isinstance(LLM_API_KEY_ENV_PATTERNS, tuple)
+    assert 'OPENAI_*' in LLM_API_KEY_ENV_PATTERNS
+
+
 def test_lazy_import_unknown():
     with pytest.raises(AttributeError, match='has no attribute'):
         pydantic_ai_harness.__getattr__('Nonexistent')