Skip to content
Open
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
20 changes: 17 additions & 3 deletions src/agents/sandbox/sandboxes/unix_local.py
Original file line number Diff line number Diff line change
Expand Up @@ -949,17 +949,31 @@ 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,
*,
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)
Comment on lines 975 to +976

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Enforce sticky-directory ownership for user-scoped rm

When user= names a non-owner removing a symlink from a sticky world-writable directory, _RM_ACCESS_CHECK_SCRIPT succeeds because it checks only parent write/execute permission, even though an actual rm as that user is rejected by the sticky bit. The subsequent normalized.unlink() runs as the SDK process and therefore deletes an entry the requested user cannot remove; execute the removal under the requested identity or extend the check to enforce sticky-directory ownership.

Useful? React with 👍 / 👎.

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is pre-existing behavior of the user-scoped path rather than something this change introduces: before this PR, rm("file", user=...) ran the same _RM_ACCESS_CHECK_SCRIPT against the same parent directory and then unlinked as the SDK process, so a non-owner in a sticky world-writable directory was already able to remove another user's regular file. The only difference now is that the check and the unlink target the symlink entry instead of its resolved target, which is the intended rm semantics.

I'd like to keep this PR scoped to the symlink-target bug. Enforcing sticky-directory ownership in the access check (or performing the removal under the requested identity) affects every user-scoped rm, not just symlinks, and is hard to exercise in the test suite because CI does not run as a second uid. Happy to send it as a follow-up PR if maintainers want it.

try:
if normalized.is_dir() and not normalized.is_symlink():
if recursive:
Expand Down
12 changes: 11 additions & 1 deletion src/agents/sandbox/session/base_sandbox_session.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand All @@ -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:
Expand Down
24 changes: 21 additions & 3 deletions src/agents/sandbox/workspace_paths.py
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand All @@ -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)
Expand Down Expand Up @@ -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 == self._root
):
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
Expand Down
196 changes: 195 additions & 1 deletion tests/sandbox/test_unix_local.py
Original file line number Diff line number Diff line change
Expand Up @@ -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 (
Expand Down Expand Up @@ -470,6 +474,196 @@ 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)
session = _RecordingUnixLocalSession(root_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,
Expand Down