diff --git a/src/agents/sandbox/apply_patch.py b/src/agents/sandbox/apply_patch.py index 30623fdf82..5b6d09bc95 100644 --- a/src/agents/sandbox/apply_patch.py +++ b/src/agents/sandbox/apply_patch.py @@ -11,6 +11,7 @@ ApplyPatchDiffError, ApplyPatchFileNotFoundError, ApplyPatchPathError, + ExecNonZeroError, InvalidManifestPathError, WorkspaceReadNotFoundError, ) @@ -71,13 +72,17 @@ async def apply_operation( ) -> ApplyPatchResult: format_impl = _resolve_patch_format(patch_format) relative_path, display_path = self._resolve_path(operation.path) - destination = self._session.normalize_path(relative_path) if operation.type == "delete_file": - await self._ensure_exists(destination, display_path=display_path) - await self._session.rm(destination, user=self._user) + # Remove the workspace entry the model named, not the file it resolves to: a + # symlink is checked and removed as the link itself, so its target survives and + # a dangling or outward-pointing link can still be deleted. + await self._ensure_entry_exists(relative_path, display_path=display_path) + await self._session.rm(relative_path, user=self._user) return ApplyPatchResult(output=f"Deleted {display_path}") + destination = self._session.normalize_path(relative_path) + if operation.diff is None: raise ApplyPatchDiffError( message=( @@ -113,7 +118,7 @@ async def apply_operation( moved_destination = self._session.normalize_path(moved_relative_path) await self._write_text(moved_destination, updated_text) if moved_destination != destination: - await self._session.rm(destination, user=self._user) + await self._session.rm(relative_path, user=self._user) return ApplyPatchResult( output=f"Updated {display_path}\nMoved {display_path} to {moved_display_path}" ) @@ -178,13 +183,31 @@ def _validate_path(self, path: str | Path) -> Path: cause=exc, ) from exc - async def _ensure_exists(self, destination: Path, *, display_path: str) -> None: + async def _ensure_entry_exists(self, relative_path: Path, *, display_path: str) -> None: + not_found: BaseException | None = None try: - handle = await self._session.read(destination, user=self._user) - except (FileNotFoundError, WorkspaceReadNotFoundError) as exc: - raise ApplyPatchFileNotFoundError(path=Path(display_path), cause=exc) from exc + destination = self._session.normalize_path(relative_path) + except InvalidManifestPathError as exc: + # The leaf resolves outside every allowed root; the entry itself may still exist. + not_found = exc else: - handle.close() + try: + handle = await self._session.read(destination, user=self._user) + except (FileNotFoundError, WorkspaceReadNotFoundError) as exc: + not_found = exc + else: + handle.close() + return + + # A dangling or outward-pointing symlink cannot be read, but it is still an entry + # of its parent directory, which is what delete_file removes. + try: + entries = await self._session.ls(relative_path.parent, user=self._user) + except ExecNonZeroError: + entries = [] + if any(Path(entry.path).name == relative_path.name for entry in entries): + return + raise ApplyPatchFileNotFoundError(path=Path(display_path), cause=not_found) from not_found async def _read_text(self, destination: Path, *, op_path: str, decode_path: Path) -> str: try: 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..9f9b6dcdad 100644 --- a/src/agents/sandbox/session/base_sandbox_session.py +++ b/src/agents/sandbox/session/base_sandbox_session.py @@ -192,7 +192,15 @@ " exit $?\n" "fi\n" 'parent=$(dirname "$target")\n' - '[ -d "$parent" ] && [ -w "$parent" ] && [ -x "$parent" ]\n' + '[ -d "$parent" ] && [ -w "$parent" ] && [ -x "$parent" ] || exit 1\n' + # A sticky directory (e.g. /tmp) lets only the entry's owner or the directory's owner + # unlink an entry, even with write access to the directory. `stat` without -L reports + # the entry itself (GNU `-c`, BSD/macOS `-f`), so a symlink is judged by the link's + # owner, not its target's. root (CAP_FOWNER) may unlink anything it can reach. + 'if [ "$(id -u)" != 0 ] && [ -k "$parent" ] && [ ! -O "$parent" ]; then\n' + ' owner=$(stat -c %u "$target" 2>/dev/null || stat -f %u "$target" 2>/dev/null)\n' + ' [ -n "$owner" ] && [ "$owner" = "$(id -u)" ]\n' + "fi\n" ) @@ -1056,6 +1064,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 +1094,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..e3a93baf9c 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,26 @@ 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 and configured grant roots are always addressed through their + # resolved form, so a symlinked root alias stays authorized; only entries below them + # keep their leaf. + if ( + follow_leaf_symlink + or absolute_path.name in ("", ".", "..") + or self._is_configured_root_alias(absolute_path) + ): + 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 @@ -442,6 +461,22 @@ def _resolved_host_path_and_grant( raise self._invalid_path_error(original) return resolved, grant + def _is_configured_root_alias(self, absolute_path: Path) -> bool: + normalized = PurePosixPath(posixpath.normpath(absolute_path.as_posix())) + if normalized == self._normalized_root(): + return True + # Compare against the configured spelling: a grant root that is itself a symlink is + # addressed as configured, while its resolved form is what `_matching_grant` checks. + return any( + normalized + == PurePosixPath( + posixpath.normpath( + Path(grant.host_path if grant.host_path is not None else grant.path).as_posix() + ) + ) + for grant in self._extra_path_grants + ) + def _sandbox_path_and_grant( self, original: PurePosixPath, diff --git a/tests/sandbox/_apply_patch_test_session.py b/tests/sandbox/_apply_patch_test_session.py index 24ce567011..d13804ab3a 100644 --- a/tests/sandbox/_apply_patch_test_session.py +++ b/tests/sandbox/_apply_patch_test_session.py @@ -6,6 +6,7 @@ from agents.sandbox import Manifest from agents.sandbox.errors import WorkspaceReadNotFoundError +from agents.sandbox.files import EntryKind, FileEntry, Permissions from agents.sandbox.session.base_sandbox_session import BaseSandboxSession from agents.sandbox.snapshot import NoopSnapshot from agents.sandbox.types import ExecResult, User @@ -93,6 +94,27 @@ async def rm( self.rm_calls.append((normalized, recursive)) self.files.pop(normalized, None) + async def ls( + self, + path: Path | str, + *, + user: str | User | None = None, + ) -> list[FileEntry]: + _ = user + normalized = self.normalize_path(path) + return [ + FileEntry( + path=str(file_path), + permissions=Permissions.from_mode(0o644), + owner="0", + group="0", + size=len(payload), + kind=EntryKind.FILE, + ) + for file_path, payload in self.files.items() + if file_path.parent == normalized + ] + class ProviderNotFoundApplyPatchSession(ApplyPatchSession): async def read(self, path: Path, *, user: str | User | None = None) -> io.BytesIO: diff --git a/tests/sandbox/test_session_utils.py b/tests/sandbox/test_session_utils.py index d1ddc828ef..ca68a41d1c 100644 --- a/tests/sandbox/test_session_utils.py +++ b/tests/sandbox/test_session_utils.py @@ -3,8 +3,10 @@ import io import os import shlex +import shutil import subprocess import sys +import tempfile import uuid from concurrent.futures import ThreadPoolExecutor from pathlib import Path @@ -23,6 +25,7 @@ from agents.sandbox.session.base_sandbox_session import ( _READ_PATH_PROBE_SCRIPT, _READ_PATH_PROBE_TIMEOUT_S, + _RM_ACCESS_CHECK_SCRIPT, BaseSandboxSession, ) from agents.sandbox.session.events import SandboxSessionFinishEvent, validate_sandbox_session_event @@ -237,6 +240,78 @@ async def test_check_mkdir_with_exec_runs_non_destructive_probe_as_user() -> Non assert session.last_command[-2:] == ("/workspace/nested/dir", "1") +def _run_rm_access_check(target: Path, *, recursive: bool = False) -> int: + return subprocess.run( + ["sh", "-c", _RM_ACCESS_CHECK_SCRIPT, "sh", str(target), "1" if recursive else "0"], + check=False, + ).returncode + + +@pytest.mark.skipif(sys.platform == "win32", reason="POSIX shell access probe") +def test_rm_access_check_allows_own_entries_in_a_writable_directory(tmp_path: Path) -> None: + (tmp_path / "own.txt").write_text("x", encoding="utf-8") + (tmp_path / "link").symlink_to("/nonexistent") + + assert _run_rm_access_check(tmp_path / "own.txt") == 0 + assert _run_rm_access_check(tmp_path / "link") == 0 # dangling symlink is still an entry + assert _run_rm_access_check(tmp_path / "missing") == 1 + assert _run_rm_access_check(tmp_path / "missing", recursive=True) == 0 + + +@pytest.mark.skipif( + sys.platform == "win32" or os.geteuid() != 0, + reason="needs root to run the probe as another user against a sticky directory", +) +def test_rm_access_check_enforces_sticky_directory_ownership_for_unprivileged_users() -> None: + """In a sticky directory owned by root, an unprivileged user may remove only entries it + owns (files and symlinks alike, judged by the link itself); root may remove any of them.""" + other_uid = 65534 # nobody + base = Path(tempfile.mkdtemp(prefix="openclaw-sticky-")) + try: + base.chmod(0o711) + sticky = base / "sticky" + sticky.mkdir() + sticky.chmod(0o1777) + root_file = sticky / "root.txt" + root_file.write_text("x", encoding="utf-8") + root_link = sticky / "root-link" + root_link.symlink_to("/etc/hostname") + their_file = sticky / "theirs.txt" + their_file.write_text("x", encoding="utf-8") + os.chown(their_file, other_uid, other_uid) + their_link = sticky / "their-link" + their_link.symlink_to("/etc/hostname") + os.lchown(their_link, other_uid, other_uid) + + def as_nobody(target: Path) -> int: + return subprocess.run( + ["sh", "-c", _RM_ACCESS_CHECK_SCRIPT, "sh", str(target), "0"], + check=False, + user=other_uid, + group=other_uid, + extra_groups=[], + ).returncode + + assert as_nobody(root_file) == 1 + assert as_nobody(root_link) == 1 + assert as_nobody(their_file) == 0 + assert as_nobody(their_link) == 0 + # The privileged user is not bound by the sticky ownership rule, even in a sticky + # directory it does not own that holds another user's entries. + foreign_sticky = base / "foreign-sticky" + foreign_sticky.mkdir() + foreign_file = foreign_sticky / "theirs.txt" + foreign_file.write_text("x", encoding="utf-8") + os.chown(foreign_file, other_uid, other_uid) + foreign_sticky.chmod(0o1777) + os.chown(foreign_sticky, other_uid, other_uid) + assert _run_rm_access_check(foreign_file) == 0 + assert _run_rm_access_check(their_file) == 0 + assert _run_rm_access_check(their_link) == 0 + finally: + shutil.rmtree(base, ignore_errors=True) + + @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..b954196d70 100644 --- a/tests/sandbox/test_unix_local.py +++ b/tests/sandbox/test_unix_local.py @@ -12,8 +12,15 @@ import pytest +from agents.editor import ApplyPatchOperation from agents.sandbox import SandboxPathGrant -from agents.sandbox.errors import PtySessionNotFoundError +from agents.sandbox.apply_patch import WorkspaceEditor +from agents.sandbox.errors import ( + ApplyPatchFileNotFoundError, + 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 +477,300 @@ 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_apply_patch_delete_file_removes_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 / "alias" + link.symlink_to("plain.txt") + session = _RecordingUnixLocalSession(workspace) + + await WorkspaceEditor(session).apply_patch( + ApplyPatchOperation(type="delete_file", path="alias") + ) + + assert not link.is_symlink() + assert target.read_text(encoding="utf-8") == "keep" + + @pytest.mark.asyncio + async def test_apply_patch_move_from_symlink_removes_link_not_its_target( + self, tmp_path: Path + ) -> None: + workspace = tmp_path / "workspace" + workspace.mkdir() + target = workspace / "plain.txt" + target.write_text("old\n", encoding="utf-8") + link = workspace / "alias" + link.symlink_to("plain.txt") + session = _RecordingUnixLocalSession(workspace) + + await WorkspaceEditor(session).apply_patch( + ApplyPatchOperation( + type="update_file", + path="alias", + diff="@@\n-old\n+new\n", + move_to="moved.txt", + ) + ) + + assert not link.is_symlink() + assert target.read_text(encoding="utf-8") == "old\n" + assert (workspace / "moved.txt").read_text(encoding="utf-8") == "new\n" + + @pytest.mark.asyncio + @pytest.mark.parametrize("target", ["missing.txt", "/etc/hostname"]) + async def test_apply_patch_delete_file_removes_dangling_and_outward_symlinks( + self, tmp_path: Path, target: str + ) -> None: + workspace = tmp_path / "workspace" + workspace.mkdir() + link = workspace / "alias" + link.symlink_to(target) + session = _RecordingUnixLocalSession(workspace) + + result = await WorkspaceEditor(session).apply_patch( + ApplyPatchOperation(type="delete_file", path="alias") + ) + + assert result == "Done!" + assert not link.is_symlink() + assert not link.exists() + + @pytest.mark.asyncio + async def test_apply_patch_delete_file_reports_a_missing_entry(self, tmp_path: Path) -> None: + workspace = tmp_path / "workspace" + workspace.mkdir() + session = _RecordingUnixLocalSession(workspace) + + with pytest.raises(ApplyPatchFileNotFoundError): + await WorkspaceEditor(session).apply_patch( + ApplyPatchOperation(type="delete_file", path="nothing-here") + ) + + @pytest.mark.asyncio + async def test_rm_accepts_a_symlinked_grant_root_alias(self, tmp_path: Path) -> None: + """A grant configured through a symlink is addressed by its resolved form, like root.""" + workspace = tmp_path / "workspace" + workspace.mkdir() + shared = tmp_path / "shared" + shared.mkdir() + (shared / "scratch.txt").write_text("scratch", encoding="utf-8") + shared_link = tmp_path / "shared-link" + shared_link.symlink_to(shared) + session = UnixLocalSandboxSession( + state=UnixLocalSandboxSessionState( + manifest=Manifest( + root=str(workspace), + extra_path_grants=(SandboxPathGrant(path=str(shared_link)),), + ), + snapshot=NoopSnapshot(id="noop"), + ) + ) + + await session.rm(str(shared_link / "scratch.txt")) + assert not (shared / "scratch.txt").exists() + + await session.rm(str(shared_link), recursive=True) + assert not shared.exists() + + @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 + # The configured (noncanonical) spelling itself must also name the root. + assert session._rm_target_path(str(tmp_path / "dummy" / ".." / "ws-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,