Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 12 additions & 0 deletions src/agents/sandbox/sandboxes/unix_local.py
Original file line number Diff line number Diff line change
Expand Up @@ -949,6 +949,17 @@ async def mkdir(
except OSError as e:
raise WorkspaceArchiveWriteError(path=normalized, cause=e) from e

def _raise_if_workspace_root_removal(self, path: Path) -> None:
# The validated path is a host realpath here, so also compare against the resolved
# root (Manifest.root may be a symlink, and /tmp is one on macOS).
root = Path(self.state.manifest.root)
if path == root.resolve(strict=False):
raise WorkspaceArchiveWriteError(
path=path,
context={"reason": "workspace_root_removal_refused"},
)
super()._raise_if_workspace_root_removal(path)

async def rm(
self,
path: Path | str,
Expand All @@ -960,6 +971,7 @@ async def rm(
normalized = await self._check_rm_with_exec(path, recursive=recursive, user=user)
else:
normalized = self.normalize_path(path, for_write=True)
self._raise_if_workspace_root_removal(normalized)
try:
if normalized.is_dir() and not normalized.is_symlink():
if recursive:
Expand Down
22 changes: 22 additions & 0 deletions src/agents/sandbox/session/base_sandbox_session.py
Original file line number Diff line number Diff line change
Expand Up @@ -830,6 +830,27 @@ def _workspace_root_path(self) -> Path:
async def _validate_path_access(self, path: Path | str, *, for_write: bool = False) -> Path:
return self.normalize_path(path, for_write=for_write)

def _raise_if_workspace_root_removal(self, path: Path) -> None:
"""Refuse ``rm`` of the workspace root itself.

`rm(".")`, `rm("")` or `rm("<root>")` passed path validation and then deleted the
whole workspace directory, after which every exec, read and write in the session
failed with WorkspaceRootNotFoundError. Removing the root is never what a caller
wants from a file operation; clearing the workspace is `rm` of its entries.
"""

# Compare POSIX spellings only: the manifest root names a path inside the sandbox,
# and resolving it on the SDK host would compare against the wrong filesystem.
candidates = {
sandbox_path_str(self.state.manifest.root),
sandbox_path_str(self._workspace_root_path()),
}
if sandbox_path_str(path) in candidates:
raise WorkspaceArchiveWriteError(
path=path,
context={"reason": "workspace_root_removal_refused"},
)

async def _validate_remote_path_access(
self,
path: Path | str,
Expand Down Expand Up @@ -1136,6 +1157,7 @@ async def rm(
:param user: Optional sandbox user to remove as.
"""
path = await self._validate_path_access(path, for_write=True)
self._raise_if_workspace_root_removal(path)

cmd: list[str] = ["rm"]
if recursive:
Expand Down
22 changes: 22 additions & 0 deletions tests/sandbox/test_session_utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,7 @@
from agents.sandbox.errors import (
MountConfigError,
WorkspaceArchiveReadError,
WorkspaceArchiveWriteError,
WorkspaceReadNotFoundError,
)
from agents.sandbox.files import EntryKind, FileEntry
Expand Down Expand Up @@ -237,6 +238,27 @@ async def test_check_mkdir_with_exec_runs_non_destructive_probe_as_user() -> Non
assert session.last_command[-2:] == ("/workspace/nested/dir", "1")


@pytest.mark.asyncio
@pytest.mark.parametrize("root_spelling", [".", "", "/workspace", "/workspace/", "sub/.."])
async def test_rm_refuses_to_remove_the_workspace_root(root_spelling: str) -> None:
session = _CaptureExecSession()

with pytest.raises(WorkspaceArchiveWriteError) as excinfo:
await session.rm(root_spelling, recursive=True)

assert excinfo.value.context.get("reason") == "workspace_root_removal_refused"
assert session.last_command is None


@pytest.mark.asyncio
async def test_rm_of_a_workspace_entry_still_runs() -> None:
session = _CaptureExecSession()

await session.rm("sub", recursive=True)

assert session.last_command == ("rm", "-rf", "--", "/workspace/sub")


@pytest.mark.asyncio
async def test_check_rm_with_exec_runs_parent_write_probe_as_user() -> None:
session = _CaptureExecSession()
Expand Down
34 changes: 33 additions & 1 deletion tests/sandbox/test_unix_local.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,7 @@
import pytest

from agents.sandbox import SandboxPathGrant
from agents.sandbox.errors import PtySessionNotFoundError
from agents.sandbox.errors import PtySessionNotFoundError, WorkspaceArchiveWriteError
from agents.sandbox.manifest import Environment, Manifest
from agents.sandbox.sandboxes import unix_local as unix_local_module
from agents.sandbox.sandboxes.unix_local import (
Expand Down Expand Up @@ -470,6 +470,38 @@ async def test_rm_as_user_checks_permissions_then_uses_local_fs(
assert not any(part.startswith("rm ") for part in session.exec_commands[0])


class TestUnixLocalRmWorkspaceRoot:
@pytest.mark.asyncio
@pytest.mark.parametrize("root_spelling", [".", "", "{root}", "{root}/", "sub/.."])
async def test_rm_refuses_to_remove_the_workspace_root(
self,
tmp_path: Path,
root_spelling: str,
) -> None:
workspace = tmp_path / "workspace"
(workspace / "sub").mkdir(parents=True)
(workspace / "sub" / "keep.txt").write_text("keep", encoding="utf-8")
session = _RecordingUnixLocalSession(workspace)

with pytest.raises(WorkspaceArchiveWriteError) as excinfo:
await session.rm(root_spelling.format(root=workspace), recursive=True)

assert excinfo.value.context.get("reason") == "workspace_root_removal_refused"
assert (workspace / "sub" / "keep.txt").read_text(encoding="utf-8") == "keep"

@pytest.mark.asyncio
async def test_rm_of_a_workspace_entry_still_removes_it(self, tmp_path: Path) -> None:
workspace = tmp_path / "workspace"
(workspace / "sub").mkdir(parents=True)
(workspace / "sub" / "old.txt").write_text("old", encoding="utf-8")
session = _RecordingUnixLocalSession(workspace)

await session.rm("sub", recursive=True)

assert workspace.is_dir()
assert not (workspace / "sub").exists()


@pytest.mark.asyncio
async def test_hydrate_workspace_cancellation_waits_for_the_extracting_worker(
tmp_path: Path,
Expand Down