feat: support sandbox extra path grants (#2920)

- Add SandboxPathGrant manifest support for explicit access to absolute
paths outside sandbox workspace.
- Centralize path handling in WorkspacePathPolicy.normalize_path(...),
including extra grant matching, symlink-aware host validation, and
most-specific grant selection.
- Harden access boundaries by rejecting filesystem-root grants, // root
aliases, and grants that resolve to /.
- Preserve nested grant semantics, including writable parent + read-only
child cases through remote symlink targets and macOS exec confinement.
- Update sandbox provider integrations to use shared path policy across
Docker, Unix-local, Runloop, Vercel, Cloudflare, E2B, Modal, Daytona,
and Blaxel.

Author: qiyaoq-oai

docs/sandbox/guide.md

@@ -216,7 +216,7 @@ Prefer built-in capabilities when they fit. Write a custom capability only when
 
 ### Manifest
 
-A [`Manifest`][agents.sandbox.manifest.Manifest] describes the workspace for a fresh sandbox session. It can set the workspace `root`, declare files and directories, copy in local files, clone Git repos, attach remote storage mounts, set environment variables, and define users or groups.
+A [`Manifest`][agents.sandbox.manifest.Manifest] describes the workspace for a fresh sandbox session. It can set the workspace `root`, declare files and directories, copy in local files, clone Git repos, attach remote storage mounts, set environment variables, define users or groups, and grant access to specific absolute paths outside the workspace.
 
 Manifest entry paths are workspace-relative. They cannot be absolute paths or escape the workspace with `..`, which keeps the workspace contract portable across local, Docker, and hosted clients.
 
@@ -237,6 +237,21 @@ Mount entries describe what storage to expose; mount strategies describe how a s
 
 Good manifest design usually means keeping the workspace contract narrow, putting long task recipes in workspace files such as `repo/task.md`, and using relative workspace paths in instructions, for example `repo/task.md` or `output/report.md`. If the agent edits files with the `Filesystem` capability's `apply_patch` tool, remember that patch paths are relative to the sandbox workspace root, not the shell `workdir`.
 
+Use `extra_path_grants` only when the agent needs a concrete absolute path outside the workspace, such as `/tmp` for temporary tool output or `/opt/toolchain` for a read-only runtime. A grant applies to both SDK file APIs and shell execution where the backend can enforce filesystem policy:
+
+```python
+from agents.sandbox import Manifest, SandboxPathGrant
+
+manifest = Manifest(
+    extra_path_grants=(
+        SandboxPathGrant(path="/tmp"),
+        SandboxPathGrant(path="/opt/toolchain", read_only=True),
+    ),
+)
+```
+
+Snapshots and `persist_workspace()` still include only the workspace root. Extra granted paths are runtime access, not durable workspace state.
+
 ### Permissions
 
 `Permissions` controls filesystem permissions for manifest entries. It is about the files the sandbox materializes, not model permissions, approval policy, or API credentials.

examples/sandbox/unix_local_runner.py

@@ -7,14 +7,17 @@
 
 import argparse
 import asyncio
+import io
 import sys
+import tempfile
 from pathlib import Path
 
 from openai.types.responses import ResponseTextDeltaEvent
 
 from agents import Runner
 from agents.run import RunConfig
-from agents.sandbox import SandboxAgent, SandboxRunConfig
+from agents.sandbox import Manifest, SandboxAgent, SandboxPathGrant, SandboxRunConfig
+from agents.sandbox.errors import WorkspaceArchiveWriteError
 from agents.sandbox.sandboxes.unix_local import UnixLocalSandboxClient
 
 if __package__ is None or __package__ == "":
@@ -29,9 +32,9 @@
 )
 
 
-async def main(model: str, question: str, stream: bool) -> None:
+def _build_manifest(external_dir: Path, scratch_dir: Path) -> Manifest:
     # The manifest is the file tree that will be materialized into the sandbox workspace.
-    manifest = text_manifest(
+    return text_manifest(
         {
             "account_brief.md": (
                 "# Northwind Health\n\n"
@@ -57,54 +60,155 @@ async def main(model: str, question: str, stream: bool) -> None:
                 "- Customer procurement requires final legal language by April 1.\n"
             ),
         }
+    ).model_copy(
+        update={
+            "extra_path_grants": (
+                SandboxPathGrant(
+                    path=str(external_dir),
+                    read_only=True,
+                    description="read-only external renewal packet notes",
+                ),
+                SandboxPathGrant(
+                    path=str(scratch_dir),
+                    description="temporary renewal packet scratch files",
+                ),
+            )
+        },
+        deep=True,
     )
 
-    # The sandbox agent sees the manifest as its workspace and uses one shared shell tool
-    # to inspect the files before answering.
-    agent = SandboxAgent(
-        name="Renewal Packet Analyst",
-        model=model,
-        instructions=(
-            "You review renewal packets for an account team. Inspect the packet before answering. "
-            "Keep the response concise, business-focused, and cite the file names that support "
-            "each conclusion. "
-            "If a conclusion depends on a file, mention that file by name. Do not invent numbers "
-            "or statuses that are not present in the workspace."
-        ),
-        default_manifest=manifest,
-        capabilities=[WorkspaceShellCapability()],
-    )
-
-    # With Unix-local sandboxes, the runner creates and cleans up the temporary workspace for us.
-    run_config = RunConfig(
-        sandbox=SandboxRunConfig(client=UnixLocalSandboxClient()),
-        workflow_name="Unix local sandbox review",
-    )
 
-    if not stream:
-        result = await Runner.run(agent, question, run_config=run_config)
-        print(result.final_output)
-        return
+async def _verify_extra_path_grants() -> None:
+    with tempfile.TemporaryDirectory(prefix="agents-unix-local-extra-") as extra_root_text:
+        extra_root = Path(extra_root_text)
+        external_dir = extra_root / "external"
+        scratch_dir = extra_root / "scratch"
+        external_dir.mkdir()
+        scratch_dir.mkdir()
+        external_input = external_dir / "external_input.txt"
+        read_only_output = external_dir / "blocked.txt"
+        sdk_output = scratch_dir / "sdk_output.txt"
+        exec_output = scratch_dir / "exec_output.txt"
+        external_input.write_text("external grant input\n", encoding="utf-8")
+
+        client = UnixLocalSandboxClient()
+        sandbox = await client.create(manifest=_build_manifest(external_dir, scratch_dir))
+        try:
+            async with sandbox:
+                payload = await sandbox.read(external_input)
+                try:
+                    await sandbox.write(read_only_output, io.BytesIO(b"should fail\n"))
+                except WorkspaceArchiveWriteError:
+                    pass
+                else:
+                    raise RuntimeError(
+                        "SDK write to read-only extra path grant unexpectedly worked."
+                    )
+                await sandbox.write(sdk_output, io.BytesIO(b"sdk grant output\n"))
+                exec_result = await sandbox.exec(
+                    "sh",
+                    "-c",
+                    'cat "$1"; printf "%s\\n" "exec grant output" > "$2"',
+                    "sh",
+                    external_input,
+                    exec_output,
+                    shell=False,
+                )
+
+                if payload.read() != b"external grant input\n":
+                    raise RuntimeError(
+                        "SDK read from extra path grant returned unexpected content."
+                    )
+                if sdk_output.read_text(encoding="utf-8") != "sdk grant output\n":
+                    raise RuntimeError("SDK write to extra path grant failed.")
+                if exec_result.stdout != b"external grant input\n" or exec_result.exit_code != 0:
+                    raise RuntimeError("Shell read from extra path grant failed.")
+                if exec_output.read_text(encoding="utf-8") != "exec grant output\n":
+                    raise RuntimeError("Shell write to extra path grant failed.")
+        finally:
+            await client.delete(sandbox)
+
+    print("extra_path_grants verification passed")
 
-    # The streaming path prints text deltas as they arrive so the example behaves like a demo.
-    stream_result = Runner.run_streamed(agent, question, run_config=run_config)
-    saw_text_delta = False
-    async for event in stream_result.stream_events():
-        if event.type == "raw_response_event" and isinstance(event.data, ResponseTextDeltaEvent):
-            if not saw_text_delta:
-                print("assistant> ", end="", flush=True)
-                saw_text_delta = True
-            print(event.data.delta, end="", flush=True)
 
-    if saw_text_delta:
-        print()
+async def main(model: str, question: str, stream: bool) -> None:
+    with tempfile.TemporaryDirectory(prefix="agents-unix-local-extra-") as extra_root_text:
+        extra_root = Path(extra_root_text)
+        external_dir = extra_root / "external"
+        scratch_dir = extra_root / "scratch"
+        external_dir.mkdir()
+        scratch_dir.mkdir()
+        external_note = external_dir / "external_renewal_note.md"
+        scratch_note = scratch_dir / "scratch_summary.md"
+        external_note.write_text(
+            "# External renewal note\n\n"
+            "Finance approved discount authority up to 10 percent, but anything higher needs "
+            "CFO approval before legal can finalize terms.\n",
+            encoding="utf-8",
+        )
+        manifest = _build_manifest(external_dir, scratch_dir)
+
+        # The sandbox agent sees the manifest as its workspace and uses one shared shell tool
+        # to inspect the files before answering.
+        agent = SandboxAgent(
+            name="Renewal Packet Analyst",
+            model=model,
+            instructions=(
+                "You review renewal packets for an account team. Inspect the packet before "
+                "answering. Keep the response concise, business-focused, and cite the file names "
+                "that support each conclusion. If a conclusion depends on a file, mention that "
+                "file by name. Do not invent numbers or statuses that are not present in the "
+                "workspace. The manifest also grants read-only access to an external note at "
+                f"`{external_note}` and read-write access to a scratch directory at "
+                f"`{scratch_dir}`. Read the external note before answering, and write a brief "
+                f"scratch note to `{scratch_note}`."
+            ),
+            default_manifest=manifest,
+            capabilities=[WorkspaceShellCapability()],
+        )
+
+        # With Unix-local sandboxes, the runner creates and cleans up the temporary workspace for us.
+        run_config = RunConfig(
+            sandbox=SandboxRunConfig(client=UnixLocalSandboxClient()),
+            workflow_name="Unix local sandbox review",
+            tracing_disabled=True,
+        )
+
+        if not stream:
+            result = await Runner.run(agent, question, run_config=run_config)
+            print(result.final_output)
+            return
+
+        # The streaming path prints text deltas as they arrive so the example behaves like a demo.
+        stream_result = Runner.run_streamed(agent, question, run_config=run_config)
+        saw_text_delta = False
+        async for event in stream_result.stream_events():
+            if event.type == "raw_response_event" and isinstance(
+                event.data, ResponseTextDeltaEvent
+            ):
+                if not saw_text_delta:
+                    print("assistant> ", end="", flush=True)
+                    saw_text_delta = True
+                print(event.data.delta, end="", flush=True)
+
+        if saw_text_delta:
+            print()
 
 
 if __name__ == "__main__":
     parser = argparse.ArgumentParser()
     parser.add_argument("--model", default="gpt-5.4", help="Model name to use.")
     parser.add_argument("--question", default=DEFAULT_QUESTION, help="Prompt to send to the agent.")
     parser.add_argument("--stream", action="store_true", default=False, help="Stream the response.")
+    parser.add_argument(
+        "--verify-extra-path-grants",
+        action="store_true",
+        default=False,
+        help="Run a local extra_path_grants smoke test without calling a model.",
+    )
     args = parser.parse_args()
 
-    asyncio.run(main(args.model, args.question, args.stream))
+    if args.verify_extra_path_grants:
+        asyncio.run(_verify_extra_path_grants())
+    else:
+        asyncio.run(main(args.model, args.question, args.stream))

src/agents/extensions/sandbox/blaxel/sandbox.py

@@ -54,6 +54,7 @@
     resolve_pty_write_yield_time_ms,
     truncate_text_by_tokens,
 )
+from ....sandbox.session.runtime_helpers import RESOLVE_WORKSPACE_PATH_HELPER, RuntimeHelperScript
 from ....sandbox.session.sandbox_client import BaseSandboxClient
 from ....sandbox.snapshot import SnapshotBase, SnapshotSpec, resolve_snapshot
 from ....sandbox.types import ExecResult, ExposedPortEndpoint, User
@@ -345,6 +346,12 @@ async def shutdown(self) -> None:
         except Exception as e:
             logger.warning("sandbox delete failed during shutdown: %s", e)
 
+    async def _validate_path_access(self, path: Path | str, *, for_write: bool = False) -> Path:
+        return await self._validate_remote_path_access(path, for_write=for_write)
+
+    def _runtime_helpers(self) -> tuple[RuntimeHelperScript, ...]:
+        return (RESOLVE_WORKSPACE_PATH_HELPER,)
+
     # -- file operations -----------------------------------------------------
 
     async def mkdir(
@@ -357,7 +364,7 @@ async def mkdir(
         if user is not None:
             path = await self._check_mkdir_with_exec(path, parents=parents, user=user)
         else:
-            path = self.normalize_path(path)
+            path = await self._validate_path_access(path, for_write=True)
         if path == Path("/"):
             return
         try:
@@ -372,9 +379,10 @@ async def mkdir(
     async def read(self, path: Path | str, *, user: str | User | None = None) -> io.IOBase:
         path = Path(path)
         if user is not None:
-            await self._check_read_with_exec(path, user=user)
+            workspace_path = await self._check_read_with_exec(path, user=user)
+        else:
+            workspace_path = await self._validate_path_access(path)
 
-        workspace_path = self.normalize_path(path)
         try:
             data: Any = await self._sandbox.fs.read_binary(str(workspace_path))
             if isinstance(data, str):
@@ -409,7 +417,7 @@ async def write(
         if not isinstance(payload, bytes | bytearray):
             raise WorkspaceWriteTypeError(path=path, actual_type=type(payload).__name__)
 
-        workspace_path = self.normalize_path(path)
+        workspace_path = await self._validate_path_access(path, for_write=True)
         try:
             await self._sandbox.fs.write_binary(str(workspace_path), bytes(payload))
         except Exception as e:

src/agents/extensions/sandbox/cloudflare/sandbox.py

@@ -319,8 +319,8 @@ def _runtime_helpers(self) -> tuple[RuntimeHelperScript, ...]:
     def _current_runtime_helper_cache_key(self) -> object | None:
         return self.state.sandbox_id
 
-    async def _normalize_path_for_io(self, path: Path | str) -> Path:
-        return await self._normalize_path_for_remote_io(path)
+    async def _validate_path_access(self, path: Path | str, *, for_write: bool = False) -> Path:
+        return await self._validate_remote_path_access(path, for_write=for_write)
 
     async def _resolve_exposed_port(self, port: int) -> ExposedPortEndpoint:
         """Cloudflare sandboxes do not yet support exposed port resolution."""
@@ -345,7 +345,7 @@ async def mount_bucket(
         mount_path: Path | str,
         options: dict[str, object],
     ) -> None:
-        workspace_path = self.normalize_path(mount_path)
+        workspace_path = await self._validate_path_access(mount_path, for_write=True)
         http = self._session()
         url = self._url("mount")
         payload = {
@@ -389,7 +389,7 @@ async def mount_bucket(
             ) from e
 
     async def unmount_bucket(self, mount_path: Path | str) -> None:
-        workspace_path = self.normalize_path(mount_path)
+        workspace_path = await self._validate_path_access(mount_path, for_write=True)
         http = self._session()
         url = self._url("unmount")
         payload = {"mountPath": str(workspace_path)}
@@ -969,7 +969,7 @@ async def read(self, path: Path | str, *, user: str | User | None = None) -> io.
         if user is not None:
             await self._check_read_with_exec(path, user=user)
 
-        workspace_path = await self._normalize_path_for_io(path)
+        workspace_path = await self._validate_path_access(path)
         http = self._session()
         url_path = quote(str(workspace_path).lstrip("/"), safe="/")
         url = self._url(f"file/{url_path}")
@@ -1040,7 +1040,7 @@ async def write(
             raise WorkspaceWriteTypeError(path=path, actual_type=type(payload).__name__)
 
         payload_bytes = bytes(payload)
-        workspace_path = await self._normalize_path_for_io(path)
+        workspace_path = await self._validate_path_access(path, for_write=True)
 
         http = self._session()
         url_path = quote(str(workspace_path).lstrip("/"), safe="/")