more normalize_path replace

Author: qiyaoq-oai

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

@@ -364,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, for_write=True)
+            path = await self._validate_path_access(path, for_write=True)
         if path == Path("/"):
             return
         try:

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

@@ -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)}

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

@@ -371,7 +371,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, for_write=True)
+            path = await self._validate_path_access(path, for_write=True)
         if path == Path("/"):
             return
         try:

tests/extensions/test_sandbox_blaxel.py

@@ -66,6 +66,27 @@ def __init__(
         self.pid = pid
 
 
+def _fake_helper_exec_result(command: str, *, symlinks: dict[str, str]) -> _FakeExecResult | None:
+    resolved = resolve_fake_workspace_path(
+        command,
+        symlinks=symlinks,
+        home_dir="/workspace",
+    )
+    if resolved is not None:
+        return _FakeExecResult(
+            exit_code=resolved.exit_code,
+            output=resolved.stdout,
+            stderr=resolved.stderr,
+        )
+
+    if "INSTALL_RUNTIME_HELPER_V1" in command or command.startswith(
+        "test -x /tmp/openai-agents/bin/resolve-workspace-path-"
+    ):
+        return _FakeExecResult()
+
+    return None
+
+
 class _FakeProcess:
     def __init__(self) -> None:
         self.exec_calls: list[tuple[dict[str, Any], dict[str, object]]] = []
@@ -76,17 +97,12 @@ def __init__(self) -> None:
 
     async def exec(self, config: dict[str, Any], **kwargs: object) -> _FakeExecResult:
         self.exec_calls.append((config, dict(kwargs)))
-        resolved = resolve_fake_workspace_path(
+        helper_result = _fake_helper_exec_result(
             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 helper_result is not None:
+            return helper_result
         if self.delay > 0:
             await asyncio.sleep(self.delay)
         if self._results_queue:
@@ -409,6 +425,29 @@ async def test_write_rejects_workspace_symlink_to_read_only_extra_path_grant(
             "resolved_path": "/tmp/protected/out.txt",
         }
 
+    @pytest.mark.asyncio
+    async def test_mkdir_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.mkdir("link/newdir")
+
+        assert fake_sandbox.fs.mkdir_calls == []
+        assert str(exc_info.value) == "failed to write archive for path: /workspace/link/newdir"
+        assert exc_info.value.context == {
+            "path": "/workspace/link/newdir",
+            "reason": "read_only_extra_path_grant",
+            "grant_path": "/tmp/protected",
+            "resolved_path": "/tmp/protected/newdir",
+        }
+
     @pytest.mark.asyncio
     async def test_running(self, fake_sandbox: _FakeSandboxInstance) -> None:
         session = _make_session(fake_sandbox)
@@ -2270,11 +2309,17 @@ async def test_hydrate_cleanup_rm_failure_suppressed(
         async def _counting_exec(config: dict[str, Any], **kw: object) -> _FakeExecResult:
             nonlocal call_count
             call_count += 1
-            if "tar" in config.get("command", ""):
-                if "xf" in config["command"]:
+            command = str(config.get("command", ""))
+            helper_result = _fake_helper_exec_result(
+                command, symlinks=fake_sandbox.process.symlinks
+            )
+            if helper_result is not None:
+                return helper_result
+            if "tar" in command:
+                if "xf" in command:
                     # tar extract succeeds.
                     return _FakeExecResult(exit_code=0, output="")
-            if "rm" in config.get("command", ""):
+            if "rm" in command:
                 raise ConnectionError("rm failed")
             return _FakeExecResult(exit_code=0, output="")
 
@@ -2369,8 +2414,13 @@ async def test_persist_rm_exception_suppressed(
         tar_data = _make_tar({"file.txt": b"hello"})
 
         async def _exec_with_rm_fail(config: dict[str, Any], **kw: object) -> _FakeExecResult:
-            cmd = config.get("command", "")
-            if "rm" in cmd:
+            command = str(config.get("command", ""))
+            helper_result = _fake_helper_exec_result(
+                command, symlinks=fake_sandbox.process.symlinks
+            )
+            if helper_result is not None:
+                return helper_result
+            if "rm" in command:
                 raise OSError("rm failed")
             return _FakeExecResult(exit_code=0, output="")
 
@@ -2390,8 +2440,13 @@ async def test_hydrate_rm_exception_suppressed(
         tar_data = _make_tar({"file.txt": b"hello"})
 
         async def _exec_with_rm_fail(config: dict[str, Any], **kw: object) -> _FakeExecResult:
-            cmd = config.get("command", "")
-            if "rm" in cmd:
+            command = str(config.get("command", ""))
+            helper_result = _fake_helper_exec_result(
+                command, symlinks=fake_sandbox.process.symlinks
+            )
+            if helper_result is not None:
+                return helper_result
+            if "rm" in command:
                 raise OSError("rm failed")
             return _FakeExecResult(exit_code=0, output="")
 

tests/extensions/test_sandbox_cloudflare.py

@@ -30,6 +30,7 @@
     MountConfigError,
     PtySessionNotFoundError,
     WorkspaceArchiveReadError,
+    WorkspaceArchiveWriteError,
     WorkspaceReadNotFoundError,
     WorkspaceWriteTypeError,
 )
@@ -38,6 +39,7 @@
 from agents.sandbox.session.pty_types import PTY_PROCESSES_MAX, allocate_pty_process_id
 from agents.sandbox.snapshot import NoopSnapshot, SnapshotBase
 from agents.sandbox.types import ExecResult
+from agents.sandbox.workspace_paths import SandboxPathGrant
 
 _WORKER_URL = "https://sandbox-cf.example.workers.dev"
 
@@ -693,6 +695,70 @@ async def test_cloudflare_mount_and_unmount_bucket_use_http_endpoints() -> None:
     assert unmount_call["json"] == {"mountPath": "/workspace/data"}
 
 
+@pytest.mark.asyncio
+async def test_cloudflare_mount_and_unmount_validate_path_access_for_write() -> None:
+    fake_http = _FakeHttp(
+        {
+            "POST /mount": _FakeResponse(status=200, json_body={"ok": True}),
+            "POST /unmount": _FakeResponse(status=200, json_body={"ok": True}),
+        }
+    )
+    sess = _make_session(fake_http=fake_http)
+    calls: list[tuple[str, bool]] = []
+
+    async def _tracking_normalize(path: Path | str, *, for_write: bool = False) -> Path:
+        calls.append((str(path), for_write))
+        return sess.normalize_path(path, for_write=for_write)
+
+    sess._validate_path_access = _tracking_normalize  # type: ignore[method-assign]
+
+    await sess.mount_bucket(
+        bucket="my-bucket",
+        mount_path=Path("/workspace/data"),
+        options={
+            "endpoint": "https://s3.amazonaws.com",
+            "readOnly": True,
+        },
+    )
+    await sess.unmount_bucket(Path("/workspace/data"))
+
+    assert calls == [
+        ("/workspace/data", True),
+        ("/workspace/data", True),
+    ]
+
+
+@pytest.mark.asyncio
+async def test_cloudflare_mount_rejects_read_only_extra_path_grant() -> None:
+    fake_http = _FakeHttp({"POST /mount": _FakeResponse(status=200, json_body={"ok": True})})
+    sess = _make_session(
+        state=_make_state(
+            manifest=Manifest(
+                extra_path_grants=(SandboxPathGrant(path="/tmp/protected", read_only=True),)
+            )
+        ),
+        fake_http=fake_http,
+    )
+
+    with pytest.raises(WorkspaceArchiveWriteError) as exc_info:
+        await sess.mount_bucket(
+            bucket="my-bucket",
+            mount_path=Path("/tmp/protected/data"),
+            options={
+                "endpoint": "https://s3.amazonaws.com",
+                "readOnly": True,
+            },
+        )
+
+    assert fake_http.calls == []
+    assert str(exc_info.value) == "failed to write archive for path: /tmp/protected/data"
+    assert exc_info.value.context == {
+        "path": "/tmp/protected/data",
+        "reason": "read_only_extra_path_grant",
+        "grant_path": "/tmp/protected",
+    }
+
+
 async def test_cloudflare_read_decodes_streamed_file_payload() -> None:
     sess = _make_session(
         fake_http=_FakeHttp(

tests/extensions/test_sandbox_daytona.py

@@ -933,6 +933,42 @@ async def test_write_rejects_workspace_symlink_to_read_only_extra_path_grant(
             "resolved_path": "/tmp/protected/out.txt",
         }
 
+    @pytest.mark.asyncio
+    async def test_mkdir_rejects_workspace_symlink_to_read_only_extra_path_grant(
+        self,
+        monkeypatch: pytest.MonkeyPatch,
+    ) -> None:
+        daytona_module = _load_daytona_module(monkeypatch)
+
+        async with daytona_module.DaytonaSandboxClient() as client:
+            session = await client.create(
+                manifest=Manifest(
+                    root=daytona_module.DEFAULT_DAYTONA_WORKSPACE_ROOT,
+                    extra_path_grants=(SandboxPathGrant(path="/tmp/protected", read_only=True),),
+                ),
+                options=daytona_module.DaytonaSandboxClientOptions(),
+            )
+            sandbox = _FakeAsyncDaytona.current_sandbox
+            assert sandbox is not None
+            sandbox.process.symlinks[f"{daytona_module.DEFAULT_DAYTONA_WORKSPACE_ROOT}/link"] = (
+                "/tmp/protected"
+            )
+
+            with pytest.raises(daytona_module.WorkspaceArchiveWriteError) as exc_info:
+                await session.mkdir("link/newdir")
+
+        assert sandbox.fs.create_folder_calls == []
+        assert str(exc_info.value) == (
+            "failed to write archive for path: "
+            f"{daytona_module.DEFAULT_DAYTONA_WORKSPACE_ROOT}/link/newdir"
+        )
+        assert exc_info.value.context == {
+            "path": f"{daytona_module.DEFAULT_DAYTONA_WORKSPACE_ROOT}/link/newdir",
+            "reason": "read_only_extra_path_grant",
+            "grant_path": "/tmp/protected",
+            "resolved_path": "/tmp/protected/newdir",
+        }
+
     @pytest.mark.asyncio
     async def test_mkdir_as_user_checks_permissions_then_uses_files_api(
         self,