address comments

Author: qiyaoq-oai

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

@@ -379,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):

src/agents/extensions/sandbox/daytona/sandbox.py

@@ -784,9 +784,10 @@ async def _terminate_pty_entry(self, entry: _DaytonaPtySessionEntry) -> None:
     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)
         daytona_exc = _import_daytona_exceptions()
         not_found_exc = daytona_exc.get("not_found")
 

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

@@ -563,8 +563,8 @@ async def read(self, path: Path, *, user: str | User | None = None) -> io.IOBase
         if user is not None:
             self._reject_user_arg(op="read", user=user)
 
+        normalized_path = await self._validate_path_access(path)
         sandbox = await self._ensure_sandbox()
-        normalized_path = self.normalize_path(path)
         try:
             payload = await sandbox.read_file(str(normalized_path))
         except Exception as exc:

tests/extensions/test_sandbox_blaxel.py

@@ -105,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.read_binary_calls: list[str] = []
         self.write_binary_calls: list[tuple[str, bytes]] = []
 
     async def mkdir(self, path: str, permissions: str = "0755") -> None:
@@ -114,6 +115,7 @@ async def mkdir(self, path: str, permissions: str = "0755") -> None:
         self.dirs.append(path)
 
     async def read_binary(self, path: str) -> bytes | str:
+        self.read_binary_calls.append(path)
         if self.read_error is not None:
             raise self.read_error
         if path not in self.files:
@@ -359,6 +361,25 @@ async def test_read_not_found(self, fake_sandbox: _FakeSandboxInstance) -> None:
         with pytest.raises(WorkspaceReadNotFoundError):
             await session.read("nonexistent.txt")
 
+    @pytest.mark.asyncio
+    async def test_read_rejects_workspace_symlink_to_ungranted_path(
+        self,
+        fake_sandbox: _FakeSandboxInstance,
+    ) -> None:
+        session = _make_session(fake_sandbox)
+        fake_sandbox.process.symlinks["/workspace/link"] = "/private"
+
+        with pytest.raises(InvalidManifestPathError) as exc_info:
+            await session.read("link/secret.txt")
+
+        assert fake_sandbox.fs.read_binary_calls == []
+        assert str(exc_info.value) == "manifest path must not escape root: link/secret.txt"
+        assert exc_info.value.context == {
+            "rel": "link/secret.txt",
+            "reason": "escape_root",
+            "resolved_path": "workspace escape: /private/secret.txt",
+        }
+
     @pytest.mark.asyncio
     async def test_write(self, fake_sandbox: _FakeSandboxInstance) -> None:
         session = _make_session(fake_sandbox)

tests/extensions/test_sandbox_daytona.py

@@ -199,14 +199,15 @@ async def delete_session(self, session_id: str) -> None:
 class _FakeFs:
     def __init__(self) -> None:
         self.create_folder_calls: list[tuple[str, str]] = []
+        self.download_file_calls: list[tuple[str, float | None]] = []
         self.upload_file_calls: list[tuple[bytes, str, float | None]] = []
         self.download_value: bytes = b""
 
     async def create_folder(self, path: str, mode: str) -> None:
         self.create_folder_calls.append((path, mode))
 
     async def download_file(self, path: str, timeout: float | None = None) -> bytes:
-        _ = (path, timeout)
+        self.download_file_calls.append((path, timeout))
         return self.download_value
 
     async def upload_file(self, data: bytes, path: str, *, timeout: float | None = None) -> None:
@@ -870,6 +871,32 @@ async def test_read_and_write_reject_paths_outside_workspace_root(
             with pytest.raises(daytona_module.InvalidManifestPathError):
                 await session.write("/etc/passwd", io.BytesIO(b"nope"))
 
+    @pytest.mark.asyncio
+    async def test_read_rejects_workspace_symlink_to_ungranted_path(
+        self,
+        monkeypatch: pytest.MonkeyPatch,
+    ) -> None:
+        daytona_module = _load_daytona_module(monkeypatch)
+
+        async with daytona_module.DaytonaSandboxClient() as client:
+            session = await client.create(options=daytona_module.DaytonaSandboxClientOptions())
+            sandbox = _FakeAsyncDaytona.current_sandbox
+            assert sandbox is not None
+            sandbox.process.symlinks[f"{daytona_module.DEFAULT_DAYTONA_WORKSPACE_ROOT}/link"] = (
+                "/private"
+            )
+
+            with pytest.raises(daytona_module.InvalidManifestPathError) as exc_info:
+                await session.read("link/secret.txt")
+
+        assert sandbox.fs.download_file_calls == []
+        assert str(exc_info.value) == "manifest path must not escape root: link/secret.txt"
+        assert exc_info.value.context == {
+            "rel": "link/secret.txt",
+            "reason": "escape_root",
+            "resolved_path": "workspace escape: /private/secret.txt",
+        }
+
     @pytest.mark.asyncio
     async def test_write_rejects_workspace_symlink_to_read_only_extra_path_grant(
         self,

tests/extensions/test_sandbox_vercel.py

@@ -129,6 +129,7 @@ def __init__(
         self.next_command_result = _FakeCommandFinished()
         self.run_command_calls: list[tuple[str, list[str], str | None]] = []
         self.refresh_calls = 0
+        self.read_file_calls: list[tuple[str, str | None]] = []
         self.stop_calls = 0
         self.wait_for_status_calls: list[tuple[object, float | None]] = []
         self.wait_for_status_error: BaseException | None = None
@@ -266,6 +267,7 @@ async def run_command(
         return self.next_command_result
 
     async def read_file(self, path: str, *, cwd: str | None = None) -> bytes | None:
+        self.read_file_calls.append((path, cwd))
         resolved = path if path.startswith("/") or cwd is None else f"{cwd.rstrip('/')}/{path}"
         return self.files.get(resolved)
 
@@ -682,6 +684,34 @@ async def test_vercel_read_and_write_reject_paths_outside_workspace_root(
         await session.write("/etc/passwd", io.BytesIO(b"nope"))
 
 
+@pytest.mark.asyncio
+async def test_vercel_read_rejects_workspace_symlink_to_ungranted_path(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    vercel_module = _load_vercel_module(monkeypatch)
+
+    state = vercel_module.VercelSandboxSessionState(
+        session_id="00000000-0000-0000-0000-000000000016",
+        manifest=Manifest(root="/workspace"),
+        snapshot=NoopSnapshot(id="snapshot"),
+        sandbox_id="sandbox-read-escape-link",
+    )
+    sandbox = _FakeAsyncSandbox(sandbox_id="sandbox-read-escape-link")
+    sandbox.symlinks["/workspace/link"] = "/private"
+    session = vercel_module.VercelSandboxSession.from_state(state, sandbox=sandbox)
+
+    with pytest.raises(InvalidManifestPathError) as exc_info:
+        await session.read("link/secret.txt")
+
+    assert sandbox.read_file_calls == []
+    assert str(exc_info.value) == "manifest path must not escape root: link/secret.txt"
+    assert exc_info.value.context == {
+        "rel": "link/secret.txt",
+        "reason": "escape_root",
+        "resolved_path": "workspace escape: /private/secret.txt",
+    }
+
+
 @pytest.mark.asyncio
 async def test_vercel_write_rejects_workspace_symlink_to_read_only_extra_path_grant(
     monkeypatch: pytest.MonkeyPatch,