diff --git a/src/agents/sandbox/sandboxes/unix_local.py b/src/agents/sandbox/sandboxes/unix_local.py index ea7f83e9d8..942f2b528f 100644 --- a/src/agents/sandbox/sandboxes/unix_local.py +++ b/src/agents/sandbox/sandboxes/unix_local.py @@ -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, @@ -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: diff --git a/src/agents/sandbox/session/base_sandbox_session.py b/src/agents/sandbox/session/base_sandbox_session.py index d377bea9ef..ec0df004a3 100644 --- a/src/agents/sandbox/session/base_sandbox_session.py +++ b/src/agents/sandbox/session/base_sandbox_session.py @@ -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("")` 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, @@ -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: diff --git a/tests/sandbox/test_session_utils.py b/tests/sandbox/test_session_utils.py index d1ddc828ef..04238a78ae 100644 --- a/tests/sandbox/test_session_utils.py +++ b/tests/sandbox/test_session_utils.py @@ -15,6 +15,7 @@ from agents.sandbox.errors import ( MountConfigError, WorkspaceArchiveReadError, + WorkspaceArchiveWriteError, WorkspaceReadNotFoundError, ) from agents.sandbox.files import EntryKind, FileEntry @@ -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() diff --git a/tests/sandbox/test_unix_local.py b/tests/sandbox/test_unix_local.py index fb13be3c51..06d87e713a 100644 --- a/tests/sandbox/test_unix_local.py +++ b/tests/sandbox/test_unix_local.py @@ -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 ( @@ -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,