Always clear env in tool sandboxes, pass vars via env capability

Signed-off-by: Cong Wang <cwang@multikernel.io>

Author: congwang-mk

examples/mcp_agent.py

@@ -106,19 +106,17 @@ def web_fetch(url: str) -> str:
 async def run_agent(user_prompt: str, workspace: str):
     """Run the agent loop: OpenAI reasoning + sandboxed local tool execution."""
 
-    # Set workspace as env var so tool functions can find it inside the sandbox
-    os.environ["SANDLOCK_WORKSPACE"] = workspace
-
     # -- Set up McpSandbox with local tools --
     mcp = McpSandbox(workspace=workspace)
 
-    # Deny by default — each tool explicitly declares what it needs.
-    # No capabilities = read-only system paths + workspace, no network.
+    # Deny by default: clean env, no writes, no network.
+    # Each tool gets only the env vars and permissions it needs.
+    ws_env = {"SANDLOCK_WORKSPACE": workspace}
 
     mcp.add_tool(
         "read_file", read_file,
         description="Read a file from the workspace. Path is relative to workspace root.",
-        # No capabilities needed — default read-only is sufficient
+        capabilities={"env": ws_env},
         input_schema={
             "type": "object",
             "properties": {"path": {"type": "string", "description": "Relative file path"}},
@@ -128,7 +126,7 @@ async def run_agent(user_prompt: str, workspace: str):
     mcp.add_tool(
         "write_file", write_file,
         description="Write content to a file in the workspace. Creates parent directories.",
-        capabilities={"fs_writable": [workspace]},  # only grant: write to workspace
+        capabilities={"fs_writable": [workspace], "env": ws_env},
         input_schema={
             "type": "object",
             "properties": {
@@ -141,7 +139,7 @@ async def run_agent(user_prompt: str, workspace: str):
     mcp.add_tool(
         "run_python", run_python,
         description="Run Python code and return stdout. No filesystem or network access.",
-        capabilities={"max_memory": "128M"},  # only grant: memory limit
+        capabilities={"max_memory": "128M"},
         input_schema={
             "type": "object",
             "properties": {"code": {"type": "string", "description": "Python code to execute"}},
@@ -151,7 +149,7 @@ async def run_agent(user_prompt: str, workspace: str):
     mcp.add_tool(
         "list_files", list_files,
         description="List files in the workspace directory.",
-        # No capabilities needed — default read-only is sufficient
+        capabilities={"env": ws_env},
         input_schema={"type": "object", "properties": {}},
     )
     mcp.add_tool(

src/sandlock/mcp/_policy.py

@@ -36,30 +36,39 @@ def policy_for_tool(
     **Deny by default**: no capabilities = read-only access to system
     paths and the workspace.  Every permission must be granted.
 
+    Environment is always cleared.  Use ``env`` capability to pass
+    specific variables::
+
+        capabilities={"env": {"API_KEY": "..."}}
+
     Args:
         workspace: Filesystem path the sandbox can read.
         capabilities: Grants keyed by Policy field name.  Common keys:
 
             - ``fs_writable: ["/tmp/workspace"]``
-            - ``net_connect: [443]``
             - ``net_allow_hosts: ["api.example.com"]``
+            - ``env: {"KEY": "value"}``
             - ``max_memory: "256M"``
 
     Returns:
         A frozen :class:`Policy` instance.
     """
+    # Fields that users cannot override — always enforced.
+    _ENFORCED = {"clean_env"}
+
     kwargs: dict[str, Any] = {
         "fs_writable": [],
         "fs_readable": [workspace, "/usr", "/lib", "/etc", "/bin", "/sbin"],
         "net_connect": [],
         "isolate_pids": True,
         "isolate_ipc": True,
         "no_raw_sockets": True,
+        "clean_env": True,
     }
 
     if capabilities:
         for key, value in capabilities.items():
-            if key in _POLICY_FIELDS:
+            if key in _POLICY_FIELDS and key not in _ENFORCED:
                 kwargs[key] = value
 
         # net_allow_hosts implies net_connect: [80, 443] unless explicit

tests/test_mcp_integration.py

@@ -133,20 +133,47 @@ def _run(self, coro):
 
     def test_read_only_by_default(self, tmp_path):
         workspace = str(tmp_path)
-        os.environ["SANDLOCK_WORKSPACE"] = workspace
 
         mcp = McpSandbox(workspace=workspace)
         mcp.add_tool("run_python", _run_python_tool)
 
         result = self._run(mcp.call_tool("run_python", {"code": "print(42)"}))
         assert "42" in result
 
+    def test_clean_env_by_default(self, tmp_path):
+        """Tools get clean env — can't see agent's env vars."""
+        workspace = str(tmp_path)
+        os.environ["SECRET_API_KEY"] = "should-not-leak"
+
+        mcp = McpSandbox(workspace=workspace)
+        mcp.add_tool("run_python", _run_python_tool)
+
+        result = self._run(mcp.call_tool("run_python", {
+            "code": "import os; print(os.environ.get('SECRET_API_KEY', 'HIDDEN'))",
+        }))
+        assert "HIDDEN" in result
+        assert "should-not-leak" not in result
+
+    def test_env_capability_passes_vars(self, tmp_path):
+        """Only explicitly granted env vars are visible."""
+        workspace = str(tmp_path)
+        ws_env = {"SANDLOCK_WORKSPACE": workspace}
+
+        mcp = McpSandbox(workspace=workspace)
+        mcp.add_tool("read_file", _read_file_tool,
+                      capabilities={"env": ws_env})
+
+        (tmp_path / "test.txt").write_text("hello")
+        result = self._run(mcp.call_tool("read_file", {"path": "test.txt"}))
+        assert "hello" in result
+
     def test_write_requires_capability(self, tmp_path):
         workspace = str(tmp_path)
-        os.environ["SANDLOCK_WORKSPACE"] = workspace
+        ws_env = {"SANDLOCK_WORKSPACE": workspace}
 
         mcp = McpSandbox(workspace=workspace)
-        mcp.add_tool("write_file", _write_file_tool)  # no capabilities
+        mcp.add_tool("write_file", _write_file_tool,
+                      capabilities={"env": ws_env})  # env but no fs_writable
 
         with pytest.raises(RuntimeError, match="failed"):
             self._run(mcp.call_tool(
@@ -155,12 +182,13 @@ def test_write_requires_capability(self, tmp_path):
 
     def test_write_with_capability(self, tmp_path):
         workspace = str(tmp_path)
-        os.environ["SANDLOCK_WORKSPACE"] = workspace
+        ws_env = {"SANDLOCK_WORKSPACE": workspace}
 
         mcp = McpSandbox(workspace=workspace)
         mcp.add_tool("write_file", _write_file_tool,
-                      capabilities={"fs_writable": [workspace]})
-        mcp.add_tool("read_file", _read_file_tool)
+                      capabilities={"fs_writable": [workspace], "env": ws_env})
+        mcp.add_tool("read_file", _read_file_tool,
+                      capabilities={"env": ws_env})
 
         self._run(mcp.call_tool(
             "write_file", {"path": "test.txt", "content": "hello"},
@@ -176,6 +204,7 @@ def test_get_policy(self, tmp_path):
                       capabilities={"fs_writable": [workspace]})
 
         assert mcp.get_policy("reader").fs_writable == []
+        assert mcp.get_policy("reader").clean_env is True
         assert workspace in mcp.get_policy("writer").fs_writable
 
     def test_unknown_tool_raises(self, tmp_path):
@@ -194,14 +223,16 @@ def test_openai_format(self, tmp_path):
 
     def test_full_workflow(self, tmp_path):
         workspace = str(tmp_path)
-        os.environ["SANDLOCK_WORKSPACE"] = workspace
+        ws_env = {"SANDLOCK_WORKSPACE": workspace}
 
         mcp = McpSandbox(workspace=workspace)
         mcp.add_tool("write_file", _write_file_tool,
-                      capabilities={"fs_writable": [workspace]})
-        mcp.add_tool("read_file", _read_file_tool)
+                      capabilities={"fs_writable": [workspace], "env": ws_env})
+        mcp.add_tool("read_file", _read_file_tool,
+                      capabilities={"env": ws_env})
         mcp.add_tool("run_python", _run_python_tool)
-        mcp.add_tool("list_files", _list_files_tool)
+        mcp.add_tool("list_files", _list_files_tool,
+                      capabilities={"env": ws_env})
 
         async def workflow():
             await mcp.call_tool("write_file",