diff --git a/src/agents/sandbox/sandboxes/unix_local.py b/src/agents/sandbox/sandboxes/unix_local.py index ea7f83e9d8..d9f852ef2f 100644 --- a/src/agents/sandbox/sandboxes/unix_local.py +++ b/src/agents/sandbox/sandboxes/unix_local.py @@ -949,6 +949,21 @@ async def mkdir( except OSError as e: raise WorkspaceArchiveWriteError(path=normalized, cause=e) from e + def _rm_target_path(self, path: Path | str) -> Path: + """Return the workspace entry ``rm`` removes without following a leaf symlink. + + ``normalize_path`` resolves every symlink, so for a symlink it names the link target. + ``rm`` on the target deleted the real file or directory tree and left the link dangling, + and a link that pointed outside the workspace (or nowhere) could not be removed at all. + POSIX ``rm`` and the exec-backed sessions remove the link itself, so validate the entry + at its own location: parents are resolved, the leaf is kept, and the workspace root, + extra grants (longest match) and read-only grants apply to that location. + """ + + return self._workspace_path_policy().normalize_path( + path, for_write=True, resolve_symlinks=True, follow_leaf_symlink=False + ) + async def rm( self, path: Path | str, @@ -956,10 +971,9 @@ async def rm( recursive: bool = False, user: str | User | None = None, ) -> None: + normalized = self._rm_target_path(path) if user is not None: - normalized = await self._check_rm_with_exec(path, recursive=recursive, user=user) - else: - normalized = self.normalize_path(path, for_write=True) + await self._check_rm_access_with_exec(normalized, recursive=recursive, user=user) 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..bb410bf9c7 100644 --- a/src/agents/sandbox/session/base_sandbox_session.py +++ b/src/agents/sandbox/session/base_sandbox_session.py @@ -1056,6 +1056,17 @@ async def _check_rm_with_exec( user: str | User | None = None, ) -> Path: workspace_path = await self._validate_path_access(path, for_write=True) + await self._check_rm_access_with_exec(workspace_path, recursive=recursive, user=user) + return workspace_path + + async def _check_rm_access_with_exec( + self, + workspace_path: Path, + *, + recursive: bool = False, + user: str | User | None = None, + ) -> None: + """Run the sandbox-side ``rm`` access check for an already validated workspace path.""" recursive_flag = "1" if recursive else "0" path_arg = sandbox_path_str(workspace_path) cmd = ("sh", "-lc", _RM_ACCESS_CHECK_SCRIPT, "sh", path_arg, recursive_flag) @@ -1075,7 +1086,6 @@ async def _check_rm_with_exec( "stderr": result.stderr.decode("utf-8", errors="replace"), }, ) - return workspace_path @abc.abstractmethod async def running(self) -> bool: diff --git a/src/agents/sandbox/workspace_paths.py b/src/agents/sandbox/workspace_paths.py index 2a5b28a606..e092d64cf9 100644 --- a/src/agents/sandbox/workspace_paths.py +++ b/src/agents/sandbox/workspace_paths.py @@ -368,11 +368,15 @@ def normalize_path( *, for_write: bool = False, resolve_symlinks: bool = False, + follow_leaf_symlink: bool = True, ) -> Path: """Return a validated absolute path under the workspace or an extra grant. `resolve_symlinks` follows symlinks on the host filesystem. Use it only when the sandbox workspace is a real local host directory, such as UnixLocalSandboxSession. + With `follow_leaf_symlink=False`, only the parent directories are resolved and the + final path component is kept as the entry itself, so an operation on a symlink (such as + removing it) is validated at the link's own location rather than at its target. """ if resolve_symlinks: @@ -382,7 +386,9 @@ def normalize_path( raise self._invalid_path_error(windows_path) else: original = Path(path) - result, grant = self._resolved_host_path_and_grant(original) + result, grant = self._resolved_host_path_and_grant( + original, follow_leaf_symlink=follow_leaf_symlink + ) else: if (windows_path := windows_absolute_path(path)) is not None: native_path = _native_path_from_windows_absolute(windows_path) @@ -427,13 +433,25 @@ def root_is_existing_host_path(self) -> bool: def _resolved_host_path_and_grant( self, original: Path, + *, + follow_leaf_symlink: bool = True, ) -> tuple[Path, SandboxPathGrant | None]: workspace_root = self._root.resolve(strict=False) if original.is_absolute(): - resolved = original.resolve(strict=False) + absolute_path = original else: absolute = self._absolute_workspace_posix_path(coerce_posix_path(original)) - resolved = Path(str(absolute)).resolve(strict=False) + absolute_path = Path(str(absolute)) + # The workspace root itself is always addressed through its resolved form, so a + # symlinked root alias stays authorized; only entries below it keep their leaf. + if ( + follow_leaf_symlink + or absolute_path.name in ("", ".", "..") + or absolute_path == Path(self._normalized_root().as_posix()) + ): + resolved = absolute_path.resolve(strict=False) + else: + resolved = absolute_path.parent.resolve(strict=False) / absolute_path.name if self._is_under(resolved, workspace_root): return resolved, None diff --git a/tests/sandbox/test_unix_local.py b/tests/sandbox/test_unix_local.py index fb13be3c51..40dc76a8aa 100644 --- a/tests/sandbox/test_unix_local.py +++ b/tests/sandbox/test_unix_local.py @@ -13,7 +13,11 @@ import pytest from agents.sandbox import SandboxPathGrant -from agents.sandbox.errors import PtySessionNotFoundError +from agents.sandbox.errors import ( + InvalidManifestPathError, + 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 +474,198 @@ 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 TestUnixLocalRmSymlinks: + @pytest.mark.asyncio + async def test_rm_removes_file_symlink_not_its_target(self, tmp_path: Path) -> None: + workspace = tmp_path / "workspace" + workspace.mkdir() + target = workspace / "plain.txt" + target.write_text("keep", encoding="utf-8") + link = workspace / "linkfile" + link.symlink_to("plain.txt") + session = _RecordingUnixLocalSession(workspace) + + await session.rm("linkfile") + + assert not link.is_symlink() + assert target.read_text(encoding="utf-8") == "keep" + + @pytest.mark.asyncio + @pytest.mark.parametrize("recursive", [False, True]) + async def test_rm_removes_directory_symlink_not_its_target( + self, + tmp_path: Path, + recursive: bool, + ) -> None: + workspace = tmp_path / "workspace" + workspace.mkdir() + target_dir = workspace / "realdir" / "inner" + target_dir.mkdir(parents=True) + data = target_dir / "data.txt" + data.write_text("keep", encoding="utf-8") + link = workspace / "linkdir" + link.symlink_to("realdir") + session = _RecordingUnixLocalSession(workspace) + + await session.rm("linkdir", recursive=recursive) + + assert not link.is_symlink() + assert data.read_text(encoding="utf-8") == "keep" + + @pytest.mark.asyncio + async def test_rm_removes_symlink_pointing_outside_the_workspace( + self, + tmp_path: Path, + ) -> None: + workspace = tmp_path / "workspace" + workspace.mkdir() + outside = tmp_path / "outside.txt" + outside.write_text("secret", encoding="utf-8") + link = workspace / "escape" + link.symlink_to(outside) + session = _RecordingUnixLocalSession(workspace) + + await session.rm("escape") + + assert not link.is_symlink() + assert outside.read_text(encoding="utf-8") == "secret" + + @pytest.mark.asyncio + async def test_rm_removes_dangling_symlink(self, tmp_path: Path) -> None: + workspace = tmp_path / "workspace" + workspace.mkdir() + link = workspace / "dangling" + link.symlink_to(tmp_path / "missing") + session = _RecordingUnixLocalSession(workspace) + + await session.rm("dangling") + + assert not link.is_symlink() + + @pytest.mark.asyncio + async def test_rm_still_rejects_entries_reached_through_an_escaping_symlink( + self, + tmp_path: Path, + ) -> None: + workspace = tmp_path / "workspace" + workspace.mkdir() + outside_dir = tmp_path / "outside" + outside_dir.mkdir() + victim = outside_dir / "victim.txt" + victim.write_text("secret", encoding="utf-8") + (workspace / "escape_dir").symlink_to(outside_dir) + session = _RecordingUnixLocalSession(workspace) + + with pytest.raises(InvalidManifestPathError): + await session.rm("escape_dir/victim.txt") + + assert victim.read_text(encoding="utf-8") == "secret" + + @pytest.mark.asyncio + @pytest.mark.parametrize("raw_path", ["C:\\outside\\file", "/outside/file", "../file"]) + async def test_rm_rejects_absolute_and_escaping_raw_paths_before_splitting_the_leaf( + self, + tmp_path: Path, + raw_path: str, + ) -> None: + workspace = tmp_path / "workspace" + workspace.mkdir() + # A workspace entry literally named like the raw path must not be what gets removed. + decoy = workspace / raw_path.split("/")[-1] + decoy.write_text("keep", encoding="utf-8") + session = _RecordingUnixLocalSession(workspace) + + with pytest.raises(InvalidManifestPathError): + await session.rm(raw_path) + + assert decoy.read_text(encoding="utf-8") == "keep" + + @pytest.mark.asyncio + async def test_rm_applies_the_most_specific_grant_to_the_leaf(self, tmp_path: Path) -> None: + """A link into a writable grant must not let rm delete a nested read-only grant root.""" + workspace = tmp_path / "workspace" + workspace.mkdir() + shared = tmp_path / "shared" + protected = shared / "protected" + protected.mkdir(parents=True) + (protected / "keep.txt").write_text("keep", encoding="utf-8") + (workspace / "tmp-link").symlink_to(shared) + session = UnixLocalSandboxSession( + state=UnixLocalSandboxSessionState( + manifest=Manifest( + root=str(workspace), + extra_path_grants=( + SandboxPathGrant(path=str(shared)), + SandboxPathGrant(path=str(protected), read_only=True), + ), + ), + snapshot=NoopSnapshot(id="noop"), + ) + ) + + with pytest.raises(WorkspaceArchiveWriteError): + await session.rm("tmp-link/protected", recursive=True) + + assert (protected / "keep.txt").read_text(encoding="utf-8") == "keep" + (shared / "scratch.txt").write_text("scratch", encoding="utf-8") + await session.rm("tmp-link/scratch.txt") + assert not (shared / "scratch.txt").exists() + + @pytest.mark.asyncio + async def test_rm_accepts_paths_resolved_through_a_symlinked_root( + self, + tmp_path: Path, + ) -> None: + """Manifest.root may be a symlink; ls() reports resolved paths that rm() must accept.""" + real_root = tmp_path / "ws" + real_root.mkdir() + root_link = tmp_path / "ws-link" + root_link.symlink_to(real_root) + (real_root / "via-real.txt").write_text("x", encoding="utf-8") + (real_root / "via-link.txt").write_text("x", encoding="utf-8") + session = _RecordingUnixLocalSession(root_link) + + await session.rm(str(real_root / "via-real.txt")) + await session.rm(str(root_link / "via-link.txt")) + + assert not (real_root / "via-real.txt").exists() + assert not (real_root / "via-link.txt").exists() + + @pytest.mark.asyncio + async def test_rm_validates_the_symlinked_root_alias_itself(self, tmp_path: Path) -> None: + """Naming the root through its alias must not be misread as removing the alias link.""" + real_root = tmp_path / "ws" + real_root.mkdir() + root_link = tmp_path / "ws-link" + root_link.symlink_to(real_root) + (tmp_path / "dummy").mkdir() + # A noncanonical spelling of the root alias must be recognized as the root too. + session = _RecordingUnixLocalSession(tmp_path / "dummy" / ".." / "ws-link") + + assert session._rm_target_path(".") == real_root + assert session._rm_target_path(str(root_link)) == real_root + + @pytest.mark.asyncio + async def test_rm_as_user_checks_the_symlink_entry_and_keeps_its_target( + self, + tmp_path: Path, + ) -> None: + workspace = tmp_path / "workspace" + workspace.mkdir() + target = workspace / "plain.txt" + target.write_text("keep", encoding="utf-8") + link = workspace / "linkfile" + link.symlink_to("plain.txt") + session = _RecordingUnixLocalSession(workspace) + + await session.rm("linkfile", user=User(name="sandbox-user")) + + assert not link.is_symlink() + assert target.read_text(encoding="utf-8") == "keep" + assert len(session.exec_commands) == 1 + assert session.exec_commands[0][-2:] == (str(link), "0") + + @pytest.mark.asyncio async def test_hydrate_workspace_cancellation_waits_for_the_extracting_worker( tmp_path: Path,