use remote access validator for sandboxes

Author: qiyaoq-oai

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(
@@ -409,7 +416,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, for_write=True)
+        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/daytona/sandbox.py

@@ -52,6 +52,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, BaseSandboxClientOptions
 from ....sandbox.snapshot import SnapshotBase, SnapshotSpec, resolve_snapshot
 from ....sandbox.types import ExecResult, ExposedPortEndpoint, User
@@ -354,6 +355,12 @@ async def _shutdown_backend(self) -> None:
         except Exception:
             pass
 
+    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,)
+
     async def mkdir(
         self,
         path: Path | str,
@@ -811,7 +818,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, for_write=True)
+        workspace_path = await self._validate_path_access(path, for_write=True)
         try:
             await self._sandbox.fs.upload_file(
                 bytes(payload),

src/agents/extensions/sandbox/vercel/sandbox.py

@@ -50,6 +50,7 @@
 from ....sandbox.session.base_sandbox_session import BaseSandboxSession
 from ....sandbox.session.dependencies import Dependencies
 from ....sandbox.session.manager import Instrumentation
+from ....sandbox.session.runtime_helpers import RESOLVE_WORKSPACE_PATH_HELPER, RuntimeHelperScript
 from ....sandbox.session.sandbox_client import BaseSandboxClient, BaseSandboxClientOptions
 from ....sandbox.snapshot import SnapshotBase, SnapshotSpec, resolve_snapshot
 from ....sandbox.types import ExecResult, ExposedPortEndpoint, User
@@ -277,6 +278,12 @@ def _prepare_exec_command(
             self._reject_user_arg(op="exec", user=user)
         return super()._prepare_exec_command(*command, shell=shell, user=user)
 
+    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,)
+
     def _validate_tar_bytes(self, raw: bytes) -> None:
         try:
             with tarfile.open(fileobj=io.BytesIO(raw), mode="r:*") as tar:
@@ -576,7 +583,7 @@ async def write(
         if user is not None:
             self._reject_user_arg(op="write", user=user)
 
-        normalized_path = self.normalize_path(path, for_write=True)
+        normalized_path = await self._validate_path_access(path, for_write=True)
         payload = data.read()
         if isinstance(payload, str):
             payload = payload.encode("utf-8")

tests/_fake_workspace_paths.py

@@ -0,0 +1,108 @@
+from __future__ import annotations
+
+import shlex
+from collections.abc import Sequence
+from dataclasses import dataclass
+from pathlib import PurePosixPath
+
+
+@dataclass(frozen=True)
+class FakeResolveWorkspaceResult:
+    exit_code: int
+    stdout: str = ""
+    stderr: str = ""
+
+
+def resolve_fake_workspace_path(
+    command: str | Sequence[str],
+    *,
+    symlinks: dict[str, str],
+    home_dir: str,
+) -> FakeResolveWorkspaceResult | None:
+    tokens = shlex.split(command) if isinstance(command, str) else list(command)
+    helper_index = next(
+        (
+            index
+            for index, token in enumerate(tokens)
+            if token.startswith("/tmp/openai-agents/bin/resolve-workspace-path-")
+        ),
+        None,
+    )
+    if helper_index is None or len(tokens) < helper_index + 4:
+        return None
+
+    root = _resolve_fake_path(tokens[helper_index + 1], symlinks=symlinks, home_dir=home_dir)
+    candidate = _resolve_fake_path(tokens[helper_index + 2], symlinks=symlinks, home_dir=home_dir)
+    for_write = tokens[helper_index + 3]
+    grant_tokens = tokens[helper_index + 4 :]
+
+    if _fake_path_is_under(candidate, root):
+        return FakeResolveWorkspaceResult(exit_code=0, stdout=candidate.as_posix())
+
+    best_grant: tuple[PurePosixPath, str, str] | None = None
+    for index in range(0, len(grant_tokens), 2):
+        grant_original = grant_tokens[index]
+        read_only = grant_tokens[index + 1]
+        grant_root = _resolve_fake_path(grant_original, symlinks=symlinks, home_dir=home_dir)
+        if not _fake_path_is_under(candidate, grant_root):
+            continue
+        if best_grant is None or len(grant_root.parts) > len(best_grant[0].parts):
+            best_grant = (grant_root, grant_original, read_only)
+
+    if best_grant is not None:
+        _grant_root, grant_original, read_only = best_grant
+        if for_write == "1" and read_only == "1":
+            return FakeResolveWorkspaceResult(
+                exit_code=114,
+                stderr=(
+                    f"read-only extra path grant: {grant_original}\n"
+                    f"resolved path: {candidate.as_posix()}\n"
+                ),
+            )
+        return FakeResolveWorkspaceResult(exit_code=0, stdout=candidate.as_posix())
+
+    return FakeResolveWorkspaceResult(
+        exit_code=111,
+        stderr=f"workspace escape: {candidate.as_posix()}\n",
+    )
+
+
+def _resolve_fake_path(
+    raw_path: str,
+    *,
+    symlinks: dict[str, str],
+    home_dir: str,
+    depth: int = 0,
+) -> PurePosixPath:
+    if depth > 64:
+        raise RuntimeError(f"symlink resolution depth exceeded: {raw_path}")
+
+    path = PurePosixPath(raw_path)
+    if not path.is_absolute():
+        path = PurePosixPath(home_dir) / path
+
+    parts = path.parts
+    current = PurePosixPath("/")
+    for index, part in enumerate(parts[1:], start=1):
+        current = current / part
+        target = symlinks.get(current.as_posix())
+        if target is None:
+            continue
+
+        target_path = PurePosixPath(target)
+        if not target_path.is_absolute():
+            target_path = current.parent / target_path
+        for remaining in parts[index + 1 :]:
+            target_path /= remaining
+        return _resolve_fake_path(
+            target_path.as_posix(),
+            symlinks=symlinks,
+            home_dir=home_dir,
+            depth=depth + 1,
+        )
+
+    return path
+
+
+def _fake_path_is_under(path: PurePosixPath, root: PurePosixPath) -> bool:
+    return path == root or root in path.parents

tests/extensions/test_sandbox_blaxel.py

@@ -14,7 +14,7 @@
 import pytest
 from pydantic import ValidationError
 
-from agents.sandbox import Manifest
+from agents.sandbox import Manifest, SandboxPathGrant
 from agents.sandbox.config import DEFAULT_PYTHON_SANDBOX_IMAGE
 from agents.sandbox.errors import (
     ExecTimeoutError,
@@ -29,6 +29,7 @@
 from agents.sandbox.snapshot import NoopSnapshot
 from agents.sandbox.types import ExposedPortEndpoint
 from agents.sandbox.util.tar_utils import validate_tar_bytes
+from tests._fake_workspace_paths import resolve_fake_workspace_path
 
 # ---------------------------------------------------------------------------
 # Package re-export test
@@ -71,9 +72,21 @@ def __init__(self) -> None:
         self.next_result = _FakeExecResult()
         self._results_queue: list[_FakeExecResult] = []
         self.delay: float = 0.0
+        self.symlinks: dict[str, str] = {}
 
     async def exec(self, config: dict[str, Any], **kwargs: object) -> _FakeExecResult:
         self.exec_calls.append((config, dict(kwargs)))
+        resolved = resolve_fake_workspace_path(
+            str(config.get("command", "")),
+            symlinks=self.symlinks,
+            home_dir="/workspace",
+        )
+        if resolved is not None:
+            return _FakeExecResult(
+                exit_code=resolved.exit_code,
+                output=resolved.stdout,
+                stderr=resolved.stderr,
+            )
         if self.delay > 0:
             await asyncio.sleep(self.delay)
         if self._results_queue:
@@ -92,6 +105,7 @@ def __init__(self) -> None:
         self.write_error: Exception | None = None
         self.mkdir_error: Exception | None = None
         self.return_str: bool = False
+        self.write_binary_calls: list[tuple[str, bytes]] = []
 
     async def mkdir(self, path: str, permissions: str = "0755") -> None:
         self.mkdir_calls.append(path)
@@ -110,6 +124,7 @@ async def read_binary(self, path: str) -> bytes | str:
         return data
 
     async def write_binary(self, path: str, data: bytes) -> None:
+        self.write_binary_calls.append((path, data))
         if self.write_error is not None:
             raise self.write_error
         self.files[path] = data
@@ -243,6 +258,7 @@ def _make_state(
     root: str = "/workspace",
     pause_on_exit: bool = False,
     sandbox_url: str | None = "https://test.bl.run",
+    extra_path_grants: tuple[SandboxPathGrant, ...] = (),
 ) -> Any:
     from agents.extensions.sandbox.blaxel.sandbox import (
         BlaxelSandboxSessionState,
@@ -251,7 +267,7 @@ def _make_state(
 
     return BlaxelSandboxSessionState(
         session_id=uuid.uuid4(),
-        manifest=Manifest(root=root),
+        manifest=Manifest(root=root, extra_path_grants=extra_path_grants),
         snapshot=NoopSnapshot(id="test-snapshot"),
         sandbox_name=sandbox_name,
         pause_on_exit=pause_on_exit,
@@ -349,6 +365,29 @@ async def test_write(self, fake_sandbox: _FakeSandboxInstance) -> None:
         await session.write("output.txt", io.BytesIO(b"written data"))
         assert fake_sandbox.fs.files["/workspace/output.txt"] == b"written data"
 
+    @pytest.mark.asyncio
+    async def test_write_rejects_workspace_symlink_to_read_only_extra_path_grant(
+        self,
+        fake_sandbox: _FakeSandboxInstance,
+    ) -> None:
+        state = _make_state(
+            extra_path_grants=(SandboxPathGrant(path="/tmp/protected", read_only=True),)
+        )
+        session = _make_session(fake_sandbox, state=state)
+        fake_sandbox.process.symlinks["/workspace/link"] = "/tmp/protected"
+
+        with pytest.raises(WorkspaceArchiveWriteError) as exc_info:
+            await session.write("link/out.txt", io.BytesIO(b"blocked"))
+
+        assert fake_sandbox.fs.write_binary_calls == []
+        assert str(exc_info.value) == "failed to write archive for path: /workspace/link/out.txt"
+        assert exc_info.value.context == {
+            "path": "/workspace/link/out.txt",
+            "reason": "read_only_extra_path_grant",
+            "grant_path": "/tmp/protected",
+            "resolved_path": "/tmp/protected/out.txt",
+        }
+
     @pytest.mark.asyncio
     async def test_running(self, fake_sandbox: _FakeSandboxInstance) -> None:
         session = _make_session(fake_sandbox)