Hardened local scratchpad env access

anton/core/backends/base.py

@@ -258,4 +258,5 @@ def __call__(
         coding_api_key: str,
         coding_base_url: str,
         workspace_path: Path | None,
-    ) -> ScratchpadRuntime: ...
+        allowed_env_keys: set[str] | None = None,
+    ) -> ScratchpadRuntime: ...

anton/core/backends/local.py

@@ -41,6 +41,7 @@ def __init__(
         cells: list[Cell] | None = None,
         workspace_path: Path | None = None,
         _venvs_base: Path | None = None,
+        allowed_env_keys: set[str] | None = None,
     ) -> None:
         super().__init__(
             name,
@@ -59,6 +60,9 @@ def __init__(
         # is a "where to put scratchpad venvs" hint; the explicit
         # arg is "the agent's project, when known".
         self._explicit_workspace_path: Path | None = workspace_path
+        # When set, only these env-var names are passed to the scratchpad
+        # subprocess. None means copy everything (legacy / CLI behaviour).
+        self._allowed_env_keys: set[str] | None = allowed_env_keys
         self._proc: asyncio.subprocess.Process | None = None
         self._boot_path: str | None = None
         self._venv_dir: str | None = None
@@ -330,11 +334,38 @@ async def start(self) -> None:
         os.close(fd)
         self._boot_path = path
 
-        env = os.environ.copy()
+        # Build the subprocess env: either a full copy (legacy/CLI default when
+        # allowed_env_keys is None) or a filtered subset (cowork-server passes
+        # only the DS_* vars for connections enabled in this conversation, plus
+        # system essentials). This prevents unrelated credentials from leaking
+        # into the scratchpad subprocess.
+        if self._allowed_env_keys is None:
+            env = os.environ.copy()
+        else:
+            # Always include system essentials so the subprocess can run at all.
+            _SYSTEM_KEYS = {
+                "PATH", "HOME", "USER", "LOGNAME", "SHELL",
+                "TMPDIR", "TEMP", "TMP",
+                "LANG", "LC_ALL", "LC_CTYPE",
+                "PYTHONPATH", "PYTHONHASHSEED",
+                "SystemRoot", "COMSPEC",  # Windows
+            }
+            env = {
+                k: v for k, v in os.environ.items()
+                if k in _SYSTEM_KEYS or k in self._allowed_env_keys
+            }
         if self._coding_model:
             env["ANTON_SCRATCHPAD_MODEL"] = self._coding_model
+        else:
+            # Explicitly remove any inherited ANTON_SCRATCHPAD_MODEL from the
+            # parent env — otherwise a model configured in the parent process
+            # (e.g. Anton's own _code_ model) leaks into test scratchpads that
+            # intentionally have no model, injecting get_llm() unexpectedly.
+            env.pop("ANTON_SCRATCHPAD_MODEL", None)
         if self._coding_provider:
             env["ANTON_SCRATCHPAD_PROVIDER"] = self._coding_provider
+        else:
+            env.pop("ANTON_SCRATCHPAD_PROVIDER", None)
         if "ANTHROPIC_API_KEY" not in env and "ANTON_ANTHROPIC_API_KEY" in env:
             env["ANTHROPIC_API_KEY"] = env["ANTON_ANTHROPIC_API_KEY"]
         if "OPENAI_API_KEY" not in env and "ANTON_OPENAI_API_KEY" in env:
@@ -742,6 +773,7 @@ def local_scratchpad_runtime_factory(
     coding_base_url: str,
     cells: list[Cell] | None,
     workspace_path: Path | None,
+    allowed_env_keys: set[str] | None = None,
 ) -> ScratchpadRuntime:
     return LocalScratchpadRuntime(
         name=name,
@@ -751,4 +783,5 @@ def local_scratchpad_runtime_factory(
         coding_base_url=coding_base_url,
         cells=cells,
         workspace_path=workspace_path,
+        allowed_env_keys=allowed_env_keys,
     )

anton/core/backends/manager.py

@@ -19,6 +19,7 @@ def __init__(
         coding_base_url: str,
         cells: list[Cell] | None = None,
         workspace_path: Path | None = None,
+        allowed_env_keys: set[str] | None = None,
     ) -> None:
         self._pads: dict[str, ScratchpadRuntime] = {}
         self._runtime_factory = runtime_factory
@@ -28,6 +29,7 @@ def __init__(
         self._coding_base_url = coding_base_url
         self._cells = cells
         self._workspace_path = workspace_path
+        self._allowed_env_keys = allowed_env_keys
         self._available_packages: list[str] = self.probe_packages()
 
     @property
@@ -58,6 +60,7 @@ async def get_or_create(self, name: str) -> ScratchpadRuntime:
                 coding_api_key=self._coding_api_key,
                 coding_base_url=self._coding_base_url,
                 workspace_path=self._workspace_path,
+                allowed_env_keys=self._allowed_env_keys,
             )
             await pad.start()
             self._pads[name] = pad

anton/core/session.py

@@ -168,6 +168,14 @@ class ChatSessionConfig:
     # so resuming a conversation days later still reports the real "now".
     # None → fall back to today.
     started_at: datetime | None = None
+    # Restrict which environment variables the scratchpad subprocess may see.
+    # When set, the subprocess env is filtered to this set plus system
+    # essentials (PATH, HOME, …). None (default) preserves legacy behaviour:
+    # the full os.environ is copied — safe for the CLI, where all DS_* vars
+    # are the user's own. Cowork-server sets this to the DS_* keys for the
+    # connections enabled in the current conversation, preventing unrelated
+    # credentials from leaking into the scratchpad.
+    scratchpad_env_keys: set[str] | None = None
     selection_elicitor: SelectionElicitor | None = None
 
 
@@ -235,6 +243,7 @@ def __init__(self, config: ChatSessionConfig) -> None:
             coding_base_url=coding_conn.base_url or "",
             cells=config.cells,
             workspace_path=config.workspace.base if config.workspace else None,
+            allowed_env_keys=config.scratchpad_env_keys,
         )
 
         self.tool_registry = ToolRegistry()

tests/test_scratchpad_env_isolation.py

@@ -0,0 +1,262 @@
+"""Tests for scratchpad env-var isolation (Phase 1 security).
+
+Verifies that LocalScratchpadRuntime correctly filters os.environ when
+allowed_env_keys is set, ensuring unrelated DS_* credentials are not
+visible inside the scratchpad subprocess.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import os
+
+import pytest
+
+from anton.core.backends.local import LocalScratchpadRuntime, local_scratchpad_runtime_factory
+from anton.core.backends.manager import ScratchpadManager
+
+
+_SCRATCHPAD_DEFAULTS = dict(
+    coding_provider="anthropic",
+    coding_model="",
+    coding_api_key="",
+    coding_base_url="",
+)
+
+
+def make_scratchpad(name: str, **kwargs) -> LocalScratchpadRuntime:
+    return LocalScratchpadRuntime(name=name, **{**_SCRATCHPAD_DEFAULTS, **kwargs})
+
+
+# ---------------------------------------------------------------------------
+# Unit tests — no subprocess needed, just inspect the env dict built in start()
+# ---------------------------------------------------------------------------
+
+class TestEnvFiltering:
+    """Verify the env dict passed to the subprocess is correctly filtered."""
+
+    def _build_env(self, pad: LocalScratchpadRuntime, environ: dict) -> dict:
+        """Simulate the env-building logic from start() against a given environ."""
+        if pad._allowed_env_keys is None:
+            return environ.copy()
+        _SYSTEM_KEYS = {
+            "PATH", "HOME", "USER", "LOGNAME", "SHELL",
+            "TMPDIR", "TEMP", "TMP",
+            "LANG", "LC_ALL", "LC_CTYPE",
+            "PYTHONPATH", "PYTHONHASHSEED",
+            "SystemRoot", "COMSPEC",
+        }
+        return {
+            k: v for k, v in environ.items()
+            if k in _SYSTEM_KEYS or k in pad._allowed_env_keys
+        }
+
+    def test_none_allowed_env_keys_copies_everything(self):
+        """allowed_env_keys=None → full env copy (legacy / CLI behaviour)."""
+        pad = make_scratchpad("test-none")
+        assert pad._allowed_env_keys is None
+
+        fake_env = {
+            "PATH": "/usr/bin",
+            "DS_POSTGRES_PROD__PASSWORD": "secret",
+            "DS_SLACK_MAIN__BOT_TOKEN": "xoxb-123",
+            "SOME_OTHER_VAR": "value",
+        }
+        result = self._build_env(pad, fake_env)
+        assert result == fake_env  # everything passes through
+
+    def test_empty_allowed_env_keys_strips_all_ds_vars(self):
+        """allowed_env_keys={} → DS_* vars stripped, system keys kept."""
+        pad = make_scratchpad("test-empty", allowed_env_keys=set())
+
+        fake_env = {
+            "PATH": "/usr/bin",
+            "HOME": "/Users/test",
+            "DS_POSTGRES_PROD__PASSWORD": "secret",
+            "DS_SLACK_MAIN__BOT_TOKEN": "xoxb-123",
+        }
+        result = self._build_env(pad, fake_env)
+        assert "DS_POSTGRES_PROD__PASSWORD" not in result
+        assert "DS_SLACK_MAIN__BOT_TOKEN" not in result
+        assert result["PATH"] == "/usr/bin"
+        assert result["HOME"] == "/Users/test"
+
+    def test_specific_allowed_keys_only_those_ds_vars_pass(self):
+        """Only the explicitly allowed DS_* var should be visible."""
+        allowed = {"DS_POSTGRES_PROD__HOST", "DS_POSTGRES_PROD__PASSWORD"}
+        pad = make_scratchpad("test-specific", allowed_env_keys=allowed)
+
+        fake_env = {
+            "PATH": "/usr/bin",
+            "DS_POSTGRES_PROD__HOST": "db.example.com",
+            "DS_POSTGRES_PROD__PASSWORD": "s3cr3t",
+            "DS_SLACK_MAIN__BOT_TOKEN": "xoxb-should-not-appear",
+            "DS_WHATSAPP_DEFAULT__ACCESS_TOKEN": "wh-should-not-appear",
+        }
+        result = self._build_env(pad, fake_env)
+
+        # Allowed DS vars — present
+        assert result["DS_POSTGRES_PROD__HOST"] == "db.example.com"
+        assert result["DS_POSTGRES_PROD__PASSWORD"] == "s3cr3t"
+        # Unrelated DS vars — blocked
+        assert "DS_SLACK_MAIN__BOT_TOKEN" not in result
+        assert "DS_WHATSAPP_DEFAULT__ACCESS_TOKEN" not in result
+        # System key — always present
+        assert result["PATH"] == "/usr/bin"
+
+    def test_system_keys_always_pass_regardless_of_allowed_set(self):
+        """PATH, HOME etc. must always be present even with a restrictive allowlist."""
+        pad = make_scratchpad("test-sys", allowed_env_keys=set())  # empty — block all DS_*
+
+        fake_env = {
+            "PATH": "/usr/bin:/usr/local/bin",
+            "HOME": "/Users/test",
+            "USER": "test",
+            "LANG": "en_US.UTF-8",
+            "DS_SOME_SECRET__KEY": "should-be-blocked",
+        }
+        result = self._build_env(pad, fake_env)
+        assert result["PATH"] == "/usr/bin:/usr/local/bin"
+        assert result["HOME"] == "/Users/test"
+        assert result["USER"] == "test"
+        assert result["LANG"] == "en_US.UTF-8"
+        assert "DS_SOME_SECRET__KEY" not in result
+
+
+# ---------------------------------------------------------------------------
+# Integration tests — actual subprocess execution
+# ---------------------------------------------------------------------------
+
+class TestEnvIsolationSubprocess:
+    """Run real scratchpad cells and verify env var visibility."""
+
+    async def test_blocked_ds_var_not_visible_in_subprocess(self):
+        """A DS_* var NOT in allowed_env_keys must not appear in the subprocess."""
+        sentinel = "SUPER_SECRET_TOKEN_XYZ_12345"
+        env_key = "DS_BLOCKED_SERVICE__TOKEN"
+
+        # Plant the secret in the parent process env
+        os.environ[env_key] = sentinel
+        try:
+            pad = make_scratchpad(
+                "test-blocked",
+                allowed_env_keys=set(),  # block all DS_* vars
+            )
+            await pad.start()
+            try:
+                cell = await pad.execute_streaming(
+                    f"import os; print(os.environ.get({env_key!r}, 'NOT_FOUND'))",
+                    description="check blocked var",
+                    estimated_seconds=5,
+                ).__anext__()
+                # Drain the generator to get the final Cell
+                async for item in pad.execute_streaming(
+                    f"import os; print(os.environ.get({env_key!r}, 'NOT_FOUND'))",
+                    description="check blocked var",
+                    estimated_seconds=5,
+                ):
+                    last = item
+                assert "NOT_FOUND" in last.stdout
+                assert sentinel not in last.stdout
+            finally:
+                await pad.close()
+        finally:
+            del os.environ[env_key]
+
+    async def test_allowed_ds_var_visible_in_subprocess(self):
+        """A DS_* var IN allowed_env_keys must be visible in the subprocess."""
+        sentinel = "ALLOWED_TOKEN_ABC_67890"
+        env_key = "DS_ALLOWED_SERVICE__TOKEN"
+
+        os.environ[env_key] = sentinel
+        try:
+            pad = make_scratchpad(
+                "test-allowed",
+                allowed_env_keys={env_key},
+            )
+            await pad.start()
+            try:
+                async for item in pad.execute_streaming(
+                    f"import os; print(os.environ.get({env_key!r}, 'NOT_FOUND'))",
+                    description="check allowed var",
+                    estimated_seconds=5,
+                ):
+                    last = item
+                assert last.stdout.strip() == sentinel
+                assert last.error is None
+            finally:
+                await pad.close()
+        finally:
+            del os.environ[env_key]
+
+    async def test_legacy_none_mode_passes_all_ds_vars(self):
+        """allowed_env_keys=None (default) → DS_* vars still visible (CLI compat)."""
+        sentinel = "CLI_COMPAT_SECRET_99999"
+        env_key = "DS_LEGACY_SERVICE__TOKEN"
+
+        os.environ[env_key] = sentinel
+        try:
+            pad = make_scratchpad(
+                "test-legacy",
+                allowed_env_keys=None,  # explicit None = legacy behaviour
+            )
+            await pad.start()
+            try:
+                async for item in pad.execute_streaming(
+                    f"import os; print(os.environ.get({env_key!r}, 'NOT_FOUND'))",
+                    description="check legacy var",
+                    estimated_seconds=5,
+                ):
+                    last = item
+                assert last.stdout.strip() == sentinel
+                assert last.error is None
+            finally:
+                await pad.close()
+        finally:
+            del os.environ[env_key]
+
+
+# ---------------------------------------------------------------------------
+# ScratchpadManager threading test
+# ---------------------------------------------------------------------------
+
+class TestManagerThreadsAllowedEnvKeys:
+    """Verify ScratchpadManager correctly threads allowed_env_keys to new runtimes."""
+
+    def test_manager_stores_allowed_env_keys(self):
+        mgr = ScratchpadManager(
+            runtime_factory=local_scratchpad_runtime_factory,
+            coding_provider="anthropic",
+            coding_model="",
+            coding_api_key="",
+            coding_base_url="",
+            allowed_env_keys={"DS_POSTGRES_PROD__HOST"},
+        )
+        assert mgr._allowed_env_keys == {"DS_POSTGRES_PROD__HOST"}
+
+    def test_manager_none_allowed_env_keys_is_default(self):
+        mgr = ScratchpadManager(
+            runtime_factory=local_scratchpad_runtime_factory,
+            coding_provider="anthropic",
+            coding_model="",
+            coding_api_key="",
+            coding_base_url="",
+        )
+        assert mgr._allowed_env_keys is None
+
+    async def test_manager_passes_allowed_env_keys_to_runtime(self):
+        """Runtime created by manager should have the same allowed_env_keys."""
+        allowed = {"DS_POSTGRES_PROD__HOST", "DS_POSTGRES_PROD__PASSWORD"}
+        mgr = ScratchpadManager(
+            runtime_factory=local_scratchpad_runtime_factory,
+            coding_provider="anthropic",
+            coding_model="",
+            coding_api_key="",
+            coding_base_url="",
+            allowed_env_keys=allowed,
+        )
+        pad = await mgr.get_or_create("test-mgr")
+        try:
+            assert pad._allowed_env_keys == allowed
+        finally:
+            await mgr.close_all()