class UnixLocalSandboxSession(BaseSandboxSession):
"""
Unix-only session implementation that runs commands on the host and uses the host filesystem
as the workspace (rooted at `self.state.manifest.root`).
On Linux, commands run without OS-level confinement added by this backend. On macOS,
commands use sandbox-exec filesystem restrictions, which do not provide network isolation.
Workspace paths and SDK file API guards do not confine arbitrary Linux shell commands.
Use this backend for trusted local execution or within externally provided isolation.
User-scoped listing and writing require sudo access to a system python3 and its standard
library. These operations run a trusted file worker in Python isolated mode, independently
of the application's interpreter or virtual environment.
"""
state: UnixLocalSandboxSessionState
_running: bool
_pty_lock: asyncio.Lock
_pty_processes: dict[int, _UnixPtyProcessEntry]
_reserved_pty_process_ids: set[int]
_fd_close_tasks: set[asyncio.Task[None]]
_host_environment_allowlist: frozenset[str] | None
def __init__(self, *, state: UnixLocalSandboxSessionState) -> None:
self.state = state
self._running = False
self._pty_lock = asyncio.Lock()
self._pty_processes = {}
self._reserved_pty_process_ids = set()
self._fd_close_tasks = set()
self._host_environment_allowlist = None
self._files = _UnixLocalFiles(state.manifest)
@classmethod
def from_state(cls, state: UnixLocalSandboxSessionState) -> "UnixLocalSandboxSession":
return cls(state=state)
async def _validate_manifest_application(
self,
*,
only_ephemeral: bool = False,
manifest: Manifest | None = None,
session_running: bool | None = None,
) -> None:
_ = (only_ephemeral, session_running)
from .._mount_security import validate_manifest_mount_credential_boundaries
validate_manifest_mount_credential_boundaries(
manifest or self.state.manifest,
provider_backend_id="unix_local",
)
async def _prepare_backend_workspace(self) -> None:
workspace = Path(self.state.manifest.root)
try:
await run_blocking_workspace_io(partial(workspace.mkdir, parents=True, exist_ok=True))
except OSError as e:
raise WorkspaceStartError(path=workspace, cause=e) from e
async def _after_start(self) -> None:
# Mark the session live only after restore/apply completes. A resumed UnixLocal session may
# recreate an empty workspace after cleanup deleted the previous root, so reporting
# "running" too early can incorrectly skip snapshot restoration based on a stale
# fingerprint cache file.
self._running = True
async def _after_start_failed(self) -> None:
self._running = False
def _wrap_stop_error(self, error: Exception) -> Exception:
return WorkspaceStopError(path=Path(self.state.manifest.root), cause=error)
async def _apply_manifest(
self,
*,
only_ephemeral: bool = False,
provision_accounts: bool = True,
) -> MaterializationResult:
_assert_unix_local_host_path_grants_unsupported(self.state.manifest)
if self.state.manifest.users or self.state.manifest.groups:
raise ValueError(
"UnixLocalSandboxSession does not support manifest users or groups because "
"provisioning would run on the host machine"
)
return await super()._apply_manifest(
only_ephemeral=only_ephemeral,
provision_accounts=provision_accounts,
)
async def provision_manifest_accounts(self) -> None:
if self.state.manifest.users or self.state.manifest.groups:
raise ValueError(
"UnixLocalSandboxSession does not support manifest users or groups because "
"provisioning would run on the host machine"
)
async def _after_shutdown(self) -> None:
await self._wait_for_fd_close_tasks()
# Best-effort: mark session not running. We intentionally do not delete the workspace
# directory here; cleanup is handled by the Client.delete().
self._running = False
async def _after_stop(self) -> None:
await self._wait_for_fd_close_tasks()
async def _resolve_exposed_port(self, port: int) -> ExposedPortEndpoint:
return ExposedPortEndpoint(host="127.0.0.1", port=port, tls=False)
def supports_pty(self) -> bool:
return True
def _prepare_exec_command(
self,
*command: str | Path,
shell: bool | list[str],
user: str | User | None,
) -> list[str]:
if shell is True:
shell = ["sh", "-c"]
return super()._prepare_exec_command(*command, shell=shell, user=user)
async def _exec_internal(
self, *command: str | Path, timeout: float | None = None
) -> ExecResult:
env, cwd = await self._resolved_exec_context()
workspace_root = await asyncio.to_thread(Path(cwd).resolve)
command_parts = self._workspace_relative_command_parts(command, workspace_root)
process_cwd, command_parts = self._shell_workspace_process_context(
command_parts=command_parts,
workspace_root=workspace_root,
cwd=cwd,
)
exec_command = self._confined_exec_command(
command_parts=command_parts,
workspace_root=workspace_root,
env=env,
)
try:
proc = await asyncio.create_subprocess_exec(
*exec_command,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
cwd=process_cwd,
env=env,
start_new_session=True,
)
try:
stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=timeout)
except asyncio.TimeoutError as e:
try:
# process tree cleanup
os.killpg(proc.pid, signal.SIGKILL)
except Exception:
pass
raise ExecTimeoutError(command=command, timeout_s=timeout, cause=e) from e
except ExecTimeoutError:
raise
except Exception as e:
raise ExecTransportError(command=command, cause=e) from e
return ExecResult(
stdout=stdout or b"", stderr=stderr or b"", exit_code=proc.returncode or 0
)
async def pty_exec_start(
self,
*command: str | Path,
timeout: float | None = None,
shell: bool | list[str] = True,
user: str | User | None = None,
tty: bool = False,
yield_time_s: float | None = None,
max_output_tokens: int | None = None,
) -> PtyExecUpdate:
_ = timeout
env, cwd = await self._resolved_exec_context()
workspace_root = await asyncio.to_thread(Path(cwd).resolve)
sanitized_command = self._prepare_exec_command(*command, shell=shell, user=user)
command_parts = self._workspace_relative_command_parts(sanitized_command, workspace_root)
process_cwd, command_parts = self._shell_workspace_process_context(
command_parts=command_parts,
workspace_root=workspace_root,
cwd=cwd,
)
exec_command = self._confined_exec_command(
command_parts=command_parts,
workspace_root=workspace_root,
env=env,
)
if tty:
primary_fd, secondary_fd = os.openpty()
def _preexec() -> None:
os.setsid()
fcntl.ioctl(secondary_fd, termios.TIOCSCTTY, 0)
# PTY children should use default terminal signal behavior even if the parent
# process temporarily ignores signals under the test runner.
_restore_pty_child_signal_defaults()
try:
process = await asyncio.create_subprocess_exec(
*exec_command,
stdin=secondary_fd,
stdout=secondary_fd,
stderr=secondary_fd,
cwd=process_cwd,
env=env,
preexec_fn=_preexec,
)
except BaseException:
with suppress(OSError):
os.close(primary_fd)
with suppress(OSError):
os.close(secondary_fd)
raise
else:
with suppress(OSError):
os.close(secondary_fd)
entry = _UnixPtyProcessEntry(process=process, tty=True, primary_fd=primary_fd)
entry.pump_tasks = [asyncio.create_task(self._pump_pty_primary_fd(entry))]
else:
process = await asyncio.create_subprocess_exec(
*exec_command,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
cwd=process_cwd,
env=env,
start_new_session=True,
)
entry = _UnixPtyProcessEntry(process=process, tty=False)
entry.pump_tasks = [
asyncio.create_task(self._pump_process_stream(entry, process.stdout)),
asyncio.create_task(self._pump_process_stream(entry, process.stderr)),
]
entry.wait_task = asyncio.create_task(self._watch_process_exit(entry))
pruned_entry: _UnixPtyProcessEntry | None = None
async with self._pty_lock:
process_id = allocate_pty_process_id(self._reserved_pty_process_ids)
self._reserved_pty_process_ids.add(process_id)
pruned_entry = self._prune_pty_processes_if_needed()
self._pty_processes[process_id] = entry
process_count = len(self._pty_processes)
if pruned_entry is not None:
await self._terminate_pty_entry(pruned_entry)
if process_count >= PTY_PROCESSES_WARNING:
logger.warning(
"PTY process count reached warning threshold: %s active sessions",
process_count,
)
yield_time_ms = 10_000 if yield_time_s is None else int(yield_time_s * 1000)
output, original_token_count, output_closed = await self._collect_pty_output(
entry=entry,
yield_time_ms=clamp_pty_yield_time_ms(yield_time_ms),
max_output_tokens=max_output_tokens,
)
return await self._finalize_pty_update(
process_id=process_id,
entry=entry,
output=output,
original_token_count=original_token_count,
output_closed=output_closed,
)
async def pty_write_stdin(
self,
*,
session_id: int,
chars: str,
yield_time_s: float | None = None,
max_output_tokens: int | None = None,
) -> PtyExecUpdate:
async with self._pty_lock:
entry = self._resolve_pty_session_entry(
pty_processes=self._pty_processes,
session_id=session_id,
)
if chars:
if not entry.tty or entry.primary_fd is None:
raise RuntimeError("stdin is not available for this process")
try:
os.write(entry.primary_fd, chars.encode("utf-8"))
except OSError as e:
if e.errno not in {
errno.EIO,
errno.EBADF,
errno.EPIPE,
errno.ECONNRESET,
}:
raise
await asyncio.sleep(0.1)
yield_time_ms = 250 if yield_time_s is None else int(yield_time_s * 1000)
output, original_token_count, output_closed = await self._collect_pty_output(
entry=entry,
yield_time_ms=resolve_pty_write_yield_time_ms(
yield_time_ms=yield_time_ms, input_empty=chars == ""
),
max_output_tokens=max_output_tokens,
)
entry.last_used = time.monotonic()
return await self._finalize_pty_update(
process_id=session_id,
entry=entry,
output=output,
original_token_count=original_token_count,
output_closed=output_closed,
)
async def pty_terminate_all(self) -> None:
async with self._pty_lock:
entries = list(self._pty_processes.values())
self._pty_processes.clear()
self._reserved_pty_process_ids.clear()
for entry in entries:
await self._terminate_pty_entry(entry)
async def _resolved_exec_context(self) -> tuple[dict[str, str], str]:
if self._host_environment_allowlist is None:
env = dict(os.environ)
else:
env = {
name: value
for name, value in os.environ.items()
if name in self._host_environment_allowlist
}
env.update(await self.state.manifest.environment.resolve())
workspace = Path(self.state.manifest.root)
if not await asyncio.to_thread(workspace.exists):
raise WorkspaceRootNotFoundError(path=workspace)
env["HOME"] = str(workspace)
return env, str(workspace)
async def _pump_process_stream(
self,
entry: _UnixPtyProcessEntry,
stream: asyncio.StreamReader | None,
) -> None:
if stream is None:
return
while True:
chunk = await stream.read(_PTY_READ_CHUNK_BYTES)
if chunk == b"":
break
async with entry.output_lock:
entry.output_chunks.append(chunk)
entry.output_notify.set()
async def _watch_process_exit(self, entry: _UnixPtyProcessEntry) -> None:
await entry.process.wait()
if entry.pump_tasks:
await asyncio.gather(*entry.pump_tasks, return_exceptions=True)
entry.output_closed.set()
entry.output_notify.set()
async def _pump_pty_primary_fd(self, entry: _UnixPtyProcessEntry) -> None:
primary_fd = entry.primary_fd
if primary_fd is None:
return
loop = asyncio.get_running_loop()
while True:
try:
chunk = await loop.run_in_executor(None, os.read, primary_fd, _PTY_READ_CHUNK_BYTES)
except OSError as e:
if e.errno in {errno.EIO, errno.EBADF}:
break
raise
if chunk == b"":
break
async with entry.output_lock:
entry.output_chunks.append(chunk)
entry.output_notify.set()
async def _collect_pty_output(
self,
*,
entry: _UnixPtyProcessEntry,
yield_time_ms: int,
max_output_tokens: int | None,
) -> tuple[bytes, int | None, bool]:
return await collect_pty_output(
output_chunks=entry.output_chunks,
output_lock=entry.output_lock,
output_notify=entry.output_notify,
is_done=entry.output_closed.is_set,
yield_time_ms=yield_time_ms,
max_output_tokens=max_output_tokens,
)
async def _finalize_pty_update(
self,
*,
process_id: int,
entry: _UnixPtyProcessEntry,
output: bytes,
original_token_count: int | None,
output_closed: bool,
) -> PtyExecUpdate:
exit_code: int | None = entry.process.returncode if output_closed else None
live_process_id: int | None = process_id
if exit_code is not None:
async with self._pty_lock:
removed = self._pty_processes.pop(process_id, None)
self._reserved_pty_process_ids.discard(process_id)
if removed is not None:
await self._terminate_pty_entry(removed)
live_process_id = None
return PtyExecUpdate(
process_id=live_process_id,
output=output,
exit_code=exit_code,
original_token_count=original_token_count,
)
def _prune_pty_processes_if_needed(self) -> _UnixPtyProcessEntry | None:
if len(self._pty_processes) < PTY_PROCESSES_MAX:
return None
meta = [
(process_id, entry.last_used, entry.output_closed.is_set())
for process_id, entry in self._pty_processes.items()
]
process_id = process_id_to_prune_from_meta(meta)
if process_id is None:
return None
self._reserved_pty_process_ids.discard(process_id)
return self._pty_processes.pop(process_id, None)
async def _terminate_pty_entry(self, entry: _UnixPtyProcessEntry) -> None:
process = entry.process
primary_fd = entry.primary_fd
entry.primary_fd = None
if process.returncode is None and process.pid is not None:
with suppress(ProcessLookupError):
os.killpg(process.pid, signal.SIGKILL)
for task in entry.pump_tasks:
task.cancel()
if entry.wait_task is not None:
entry.wait_task.cancel()
if entry.tty:
if primary_fd is not None:
# On macOS we have observed os.close() on the PTY master fd block while a
# background reader thread is still inside os.read(). Keep the close task owned
# by the session without making PTY termination wait indefinitely for it.
self._schedule_fd_close(primary_fd)
entry.output_closed.set()
entry.output_notify.set()
return
if primary_fd is not None:
_close_fd_quietly(primary_fd)
await asyncio.gather(*entry.pump_tasks, return_exceptions=True)
if entry.wait_task is not None:
await asyncio.gather(entry.wait_task, return_exceptions=True)
def _schedule_fd_close(self, fd: int) -> None:
task = asyncio.create_task(asyncio.to_thread(_close_fd_quietly, fd))
self._fd_close_tasks.add(task)
task.add_done_callback(self._fd_close_tasks.discard)
async def _wait_for_fd_close_tasks(self) -> None:
tasks = tuple(self._fd_close_tasks)
if tasks:
await asyncio.wait(tasks, timeout=_PTY_FD_CLOSE_GRACE_SECONDS)
def _confined_exec_command(
self,
*,
command_parts: list[str],
workspace_root: Path,
env: Mapping[str, str],
) -> list[str]:
if sys.platform != "darwin":
return command_parts
sandbox_exec = shutil.which("sandbox-exec")
if not sandbox_exec:
raise ExecTransportError(
command=command_parts,
context={
"reason": "unix_local_confinement_unavailable",
"platform": sys.platform,
"workspace_root": str(workspace_root),
},
)
profile = self._darwin_exec_profile(
workspace_root,
extra_read_paths=self._darwin_additional_read_paths(
command_parts=command_parts,
env=env,
),
extra_path_grants=self._darwin_extra_path_grant_roots(),
)
return [sandbox_exec, "-p", profile, *command_parts]
@staticmethod
def _workspace_relative_command_parts(
command: Sequence[str | Path],
workspace_root: Path,
) -> list[str]:
command_parts = [str(part) for part in command]
rewritten = [command_parts[0]]
for part in command_parts[1:]:
path_part = Path(part)
if not path_part.is_absolute():
rewritten.append(part)
continue
try:
relative = path_part.relative_to(workspace_root)
except ValueError:
rewritten.append(part)
continue
rewritten.append("." if not relative.parts else relative.as_posix())
return rewritten
@staticmethod
def _darwin_allowable_read_roots(
path: Path,
*,
host_home: Path,
allow_virtual_environment_root: bool = False,
) -> list[Path]:
candidates: set[Path] = set()
normalized = path.expanduser()
try:
resolved = normalized.resolve(strict=False)
except OSError:
resolved = normalized
if normalized.is_dir():
candidates.add(normalized)
else:
candidates.add(normalized.parent)
if resolved.is_dir():
candidates.add(resolved)
else:
candidates.add(resolved.parent)
if allow_virtual_environment_root:
for candidate in (normalized, resolved):
if candidate.name != "bin":
continue
virtual_env_root = candidate.parent
if (virtual_env_root / "pyvenv.cfg").is_file():
candidates.add(virtual_env_root)
resolved_text = resolved.as_posix()
if resolved_text == "/opt/homebrew" or resolved_text.startswith("/opt/homebrew/"):
candidates.add(Path("/opt/homebrew"))
if resolved_text == "/usr/local" or resolved_text.startswith("/usr/local/"):
candidates.add(Path("/usr/local"))
if resolved_text == "/Library/Frameworks" or resolved_text.startswith(
"/Library/Frameworks/"
):
candidates.add(Path("/Library/Frameworks"))
try:
relative_to_home = resolved.relative_to(host_home)
except ValueError:
relative_to_home = None
if relative_to_home is not None and relative_to_home.parts:
first_segment = relative_to_home.parts[0]
if first_segment.startswith("."):
candidates.add(host_home / first_segment)
elif len(relative_to_home.parts) >= 2 and relative_to_home.parts[:2] == (
"Library",
"Python",
):
candidates.add(host_home / "Library" / "Python")
return sorted(
candidates, key=lambda candidate: (len(candidate.parts), candidate.as_posix())
)
def _darwin_additional_read_paths(
self,
*,
command_parts: list[str],
env: Mapping[str, str],
) -> list[Path]:
host_home = Path.home().resolve()
allowed: list[Path] = []
seen: set[str] = set()
def _append(
path: str | Path | None,
*,
allow_virtual_environment_root: bool = False,
) -> None:
if path is None:
return
candidate = Path(path).expanduser()
if not candidate.is_absolute():
return
for root in self._darwin_allowable_read_roots(
candidate,
host_home=host_home,
allow_virtual_environment_root=allow_virtual_environment_root,
):
key = root.as_posix()
if key in seen:
continue
seen.add(key)
allowed.append(root)
child_path_entries = env.get("PATH", "").split(os.pathsep)
for path_entry in child_path_entries:
if path_entry:
_append(path_entry)
executable = shutil.which(command_parts[0], path=env.get("PATH"))
_append(executable)
# The client must permit PATH inheritance before retained host entries can widen a bin
# grant to its virtual environment root. Matching manifest values cannot grant authority.
if self._host_environment_allowlist is None or "PATH" in self._host_environment_allowlist:
for path_entry in os.environ.get("PATH", "").split(os.pathsep):
if path_entry and path_entry in child_path_entries:
_append(path_entry, allow_virtual_environment_root=True)
return allowed
def _darwin_extra_path_grant_roots(self) -> list[tuple[Path, bool]]:
roots: list[tuple[Path, bool]] = []
seen: set[tuple[str, bool]] = set()
def _append(path: Path, *, read_only: bool) -> None:
_raise_if_filesystem_root(path, resolved=True)
key = (path.as_posix(), read_only)
if key in seen:
return
seen.add(key)
roots.append((path, read_only))
for grant in self.state.manifest.extra_path_grants:
grant_path = Path(grant.path).expanduser()
try:
resolved = grant_path.resolve(strict=False)
except OSError:
_append(grant_path, read_only=grant.read_only)
continue
_raise_if_filesystem_root(resolved, resolved=True)
_append(grant_path, read_only=grant.read_only)
if resolved != grant_path:
_append(resolved, read_only=grant.read_only)
return roots
def _darwin_exec_profile(
self,
workspace_root: Path,
*,
extra_read_paths: Sequence[Path] = (),
extra_path_grants: Sequence[tuple[Path, bool]] = (),
) -> str:
def _literal(path: Path | str) -> str:
escaped = str(path).replace("\\", "\\\\").replace('"', '\\"')
return f'"{escaped}"'
denied_paths = [
Path("/Users"),
Path("/Volumes"),
Path("/Applications"),
Path("/Library"),
Path("/opt"),
Path("/etc"),
Path("/private/etc"),
Path("/tmp"),
Path("/private/tmp"),
Path("/private"),
Path("/var"),
Path("/usr"),
]
allow_rules = [
f"(allow file-read-data file-read-metadata (subpath {_literal(workspace_root)}))",
f"(allow file-write* (subpath {_literal(workspace_root)}))",
*[
f"(allow file-read-data file-read-metadata (subpath {_literal(path)}))"
for path in extra_read_paths
],
*[
f"(allow file-read-data file-read-metadata (subpath {_literal(path)}))"
for path, _read_only in extra_path_grants
],
*[
f"(allow file-write* (subpath {_literal(path)}))"
for path, read_only in extra_path_grants
if not read_only
],
*[
f"(deny file-write* (subpath {_literal(path)}))"
for path, read_only in extra_path_grants
if read_only
],
'(allow file-read-data file-read-metadata (subpath "/usr/bin"))',
'(allow file-read-data file-read-metadata (subpath "/usr/lib"))',
'(allow file-read-data file-read-metadata (subpath "/bin"))',
'(allow file-read-data file-read-metadata (subpath "/System"))',
'(allow file-read-data file-read-metadata (literal "/private/var/select/sh"))',
'(allow file-write* (literal "/dev/null"))',
]
deny_rules = "\n".join(
f"(deny file-read-data (subpath {_literal(path)}))\n"
f"(deny file-write* (subpath {_literal(path)}))"
for path in denied_paths
)
return "\n".join(
[
"(version 1)",
"(allow default)",
deny_rules,
*allow_rules,
]
)
@staticmethod
def _shell_workspace_process_context(
*,
command_parts: list[str],
workspace_root: Path,
cwd: str,
) -> tuple[str, list[str]]:
if len(command_parts) < 3 or command_parts[0] != "sh" or command_parts[1] != "-c":
return cwd, command_parts
workspace_cd = f"cd {shlex.quote(str(workspace_root))} && {command_parts[2]}"
rewritten = [*command_parts]
rewritten[2] = workspace_cd
return "/", rewritten
def normalize_path(self, path: Path | str, *, for_write: bool = False) -> Path:
self._files.configure(self.state.manifest)
policy = self._workspace_path_policy()
return policy.normalize_path(path, for_write=for_write, resolve_symlinks=True)
async def ls(
self,
path: Path | str,
*,
user: str | User | None = None,
) -> list[FileEntry]:
normalized = self.normalize_path(path)
if user is not None:
command = ("ls", "-la", "--", str(normalized))
try:
result = await self._run_file_operation_as_user("ls", normalized, user=user)
except OSError as e:
raise ExecNonZeroError(
ExecResult(stdout=b"", stderr=str(e).encode("utf-8"), exit_code=1),
command=command,
cause=e,
) from e
if result.returncode:
raise ExecNonZeroError(
ExecResult(
stdout=result.stdout, stderr=result.stderr, exit_code=result.returncode
),
command=command,
)
return [
FileEntry(
path=entry["path"],
permissions=Permissions.from_mode(entry["mode"]).model_copy(
update={"directory": entry["kind"] == "directory"}
),
owner=entry["owner"],
group=entry["group"],
size=entry["size"],
kind=EntryKind(entry["kind"]),
)
for entry in json.loads(result.stdout)
]
command = ("ls", "-la", "--", str(normalized))
try:
with (
self._files.directory(normalized) as directory_fd,
os.scandir(directory_fd) as entries,
):
listed: list[FileEntry] = []
for entry in entries:
stat_result = entry.stat(follow_symlinks=False)
if entry.is_symlink():
kind = EntryKind.SYMLINK
elif entry.is_dir(follow_symlinks=False):
kind = EntryKind.DIRECTORY
elif entry.is_file(follow_symlinks=False):
kind = EntryKind.FILE
else:
kind = EntryKind.OTHER
listed.append(
FileEntry(
path=str(normalized / entry.name),
permissions=Permissions.from_mode(stat_result.st_mode),
owner=str(stat_result.st_uid),
group=str(stat_result.st_gid),
size=stat_result.st_size,
kind=kind,
)
)
return listed
except OSError as e:
raise ExecNonZeroError(
ExecResult(stdout=b"", stderr=str(e).encode("utf-8"), exit_code=1),
command=command,
cause=e,
) from e
async def mkdir(
self,
path: Path | str,
*,
parents: bool = False,
user: str | User | None = None,
) -> None:
if user is not None:
normalized = await self._check_mkdir_with_exec(path, parents=parents, user=user)
else:
normalized = self.normalize_path(path, for_write=True)
try:
self._files.mkdir(normalized, parents=parents)
except OSError as e:
raise WorkspaceArchiveWriteError(path=normalized, cause=e) from e
async def rm(
self,
path: Path | str,
*,
recursive: bool = False,
user: str | User | None = None,
) -> None:
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)
try:
if recursive:
await run_blocking_workspace_io(partial(self._files.rm, normalized, recursive=True))
else:
self._files.rm(normalized, recursive=False)
except FileNotFoundError as e:
if recursive:
return
raise ExecNonZeroError(
ExecResult(stdout=b"", stderr=str(e).encode("utf-8"), exit_code=1),
command=("rm", "-rf" if recursive else "--", str(normalized)),
cause=e,
) from e
except OSError as e:
raise WorkspaceArchiveWriteError(path=normalized, cause=e) from e
async def _remove_workspace_entry_on_resume(self, path: Path) -> None:
# Snapshot cleanup owns the entry, not the target of a stale symlink.
target = self.normalize_path(path.parent, for_write=True) / path.name
try:
await run_blocking_workspace_io(partial(self._files.rm, target, recursive=True))
except FileNotFoundError:
return
except OSError as e:
raise WorkspaceArchiveWriteError(path=target, cause=e) from e
async def _read_bounded(self, path: Path, *, max_bytes: int) -> bytes:
return self._files.read_bounded(self.normalize_path(path), max_bytes)
async def read(self, path: Path, *, user: str | User | None = None) -> io.IOBase:
if user is not None:
await self._check_read_with_exec(path, user=user)
workspace_path = self.normalize_path(path)
try:
return self._files.read(workspace_path)
except FileNotFoundError as e:
raise WorkspaceReadNotFoundError(path=path, cause=e) from e
except OSError as e:
raise WorkspaceArchiveReadError(path=path, cause=e) from e
async def write(
self,
path: Path,
data: io.IOBase,
*,
user: str | User | None = None,
) -> None:
payload = coerce_write_payload(path=path, data=data)
workspace_path = self.normalize_path(path, for_write=True)
if user is not None:
await self._write_stream_with_exec(workspace_path, payload.stream, user=user)
return
try:
self._files.write(workspace_path, payload.stream)
except OSError as e:
raise WorkspaceArchiveWriteError(path=workspace_path, cause=e) from e
async def mv(
self,
source: Path | str,
destination: Path | str,
*,
user: str | User | None = None,
) -> None:
# A rename of the entry, descriptor-relative like every other file operation here,
# so the paths validated above are the ones acted on. `os.rename` never puts the
# source inside an existing directory the way `mv` does; it fails instead.
normalized_source = self._normalize_entry_path(source, for_write=True)
normalized_destination = self._normalize_entry_path(destination, for_write=True)
command = ("mv", "-f", "--", str(normalized_source), str(normalized_destination))
if user is not None:
try:
result = await self._run_file_operation_as_user(
"rename", normalized_source, normalized_destination, user=user
)
except OSError as e:
raise ExecNonZeroError(
ExecResult(stdout=b"", stderr=str(e).encode("utf-8"), exit_code=1),
command=command,
cause=e,
) from e
if result.returncode:
raise ExecNonZeroError(
ExecResult(
stdout=result.stdout, stderr=result.stderr, exit_code=result.returncode
),
command=command,
)
return
try:
self._files.rename(normalized_source, normalized_destination)
except OSError as e:
raise ExecNonZeroError(
ExecResult(stdout=b"", stderr=str(e).encode("utf-8"), exit_code=1),
command=command,
cause=e,
) from e
async def same_file(
self,
left: Path | str,
right: Path | str,
*,
follow_symlinks: bool = True,
user: str | User | None = None,
) -> bool:
normalize = self.normalize_path if follow_symlinks else self._normalize_entry_path
normalized_left = normalize(left)
normalized_right = normalize(right)
command = ("test", str(normalized_left), "-ef", str(normalized_right))
if user is not None:
try:
result = await self._run_file_operation_as_user(
"same_file",
normalized_left,
normalized_right,
user=user,
payload=b"1" if follow_symlinks else b"0",
)
except OSError as e:
raise ExecNonZeroError(
ExecResult(stdout=b"", stderr=str(e).encode("utf-8"), exit_code=1),
command=command,
cause=e,
) from e
if result.returncode:
raise ExecNonZeroError(
ExecResult(
stdout=result.stdout, stderr=result.stderr, exit_code=result.returncode
),
command=command,
)
return bool(json.loads(result.stdout))
try:
return self._files.same_file(
normalized_left, normalized_right, follow_symlinks=follow_symlinks
)
except OSError as e:
raise ExecNonZeroError(
ExecResult(stdout=b"", stderr=str(e).encode("utf-8"), exit_code=1),
command=command,
cause=e,
) from e
def _normalize_entry_path(self, path: Path | str, *, for_write: bool = False) -> Path:
# Resolve and authorize the parent without following the leaf that rename/stat owns.
self._files.configure(self.state.manifest)
lexical = Path(path)
if not lexical.is_absolute():
lexical = self._workspace_path_policy().absolute_workspace_path(path)
resolved = lexical.parent.resolve(strict=False) / lexical.name
return self._files.authorize(resolved, for_write=for_write)
async def _write_new_file(
self,
path: Path,
data: io.IOBase,
*,
user: str | User | None = None,
) -> None:
payload = coerce_write_payload(path=path, data=data)
# Resolve the parent the way the ordinary write path does, so a supported internal
# symlink such as "internal -> real" still works, then keep the leaf name
# unresolved so the file ops open it with O_NOFOLLOW and a symlink at the target
# name is rejected rather than followed.
requested = Path(path)
target = self.normalize_path(requested.parent, for_write=True) / requested.name
if user is not None:
await self._write_new_stream_with_exec(target, payload.stream, user=user)
return
try:
self._files.write_new(target, payload.stream)
except _unix_local_file_ops._IncompleteCreateError as e:
raise WorkspaceArchiveWriteError(
path=target,
cause=e,
retryable=False,
message=_INCOMPLETE_CREATE_MESSAGE,
) from e
except FileExistsError:
raise
except OSError as e:
raise WorkspaceArchiveWriteError(path=target, cause=e) from e
async def _write_new_stream_with_exec(
self,
path: Path,
stream: io.IOBase,
*,
user: str | User,
) -> None:
payload = stream.read()
if isinstance(payload, str):
payload = payload.encode("utf-8")
elif not isinstance(payload, bytes):
payload = bytes(payload)
try:
result = await self._run_file_operation_as_user(
"write_new", path, user=user, payload=payload
)
except OSError as e:
raise WorkspaceArchiveWriteError(path=path, cause=e) from e
if result.returncode == _unix_local_file_ops._EXISTING_TARGET_EXIT_CODE:
raise FileExistsError(str(path))
if result.returncode == _unix_local_file_ops._INCOMPLETE_CREATE_EXIT_CODE:
raise WorkspaceArchiveWriteError(
path=path,
retryable=False,
message=_INCOMPLETE_CREATE_MESSAGE,
)
if result.returncode:
raise WorkspaceArchiveWriteError(
path=path,
context={
"stderr": result.stderr.decode("utf-8", errors="replace"),
"operation": "write_new",
},
)
async def _write_stream_with_exec(
self,
path: Path,
stream: io.IOBase,
*,
user: str | User,
) -> None:
payload = stream.read()
if isinstance(payload, str):
payload = payload.encode("utf-8")
elif not isinstance(payload, bytes):
payload = bytes(payload)
try:
result = await self._run_file_operation_as_user(
"write", path, user=user, payload=payload
)
except OSError as e:
raise WorkspaceArchiveWriteError(path=path, cause=e) from e
if result.returncode:
raise WorkspaceArchiveWriteError(
path=path,
context={
"stderr": result.stderr.decode("utf-8", errors="replace"),
"operation": "write",
},
)
async def _run_file_operation_as_user(
self,
operation: Literal["ls", "write", "write_new", "rename", "same_file"],
path: Path,
*more_paths: Path,
user: str | User,
payload: bytes = b"",
) -> subprocess.CompletedProcess[bytes]:
# Authorization is synchronous and captured for this operation before dispatch.
for_write = operation in ("write", "write_new", "rename")
paths = [self._files.authorize(each, for_write=for_write) for each in (path, *more_paths)]
command = self._prepare_exec_command(
"python3",
"-I",
"-S",
"-c",
_USER_FILE_WORKER_SOURCE,
operation,
*(str(each) for each in paths),
shell=False,
user=user,
)
# Resolve sudo from trusted host configuration, never the workspace's command environment.
executable = shutil.which(command[0])
if executable is None:
raise FileNotFoundError(f"UnixLocal user dispatcher is unavailable: {command[0]}")
command[0] = executable
# This trusted worker enforces the file boundary itself, like direct host file I/O.
# It does not execute workspace commands or load workspace Python modules.
def run_worker() -> subprocess.CompletedProcess[bytes]:
return subprocess.run(
command,
input=payload,
capture_output=True,
cwd="/",
env={"PATH": os.defpath},
start_new_session=True,
check=False,
)
return await run_blocking_workspace_io(run_worker)
async def running(self) -> bool:
return self._running
async def persist_workspace(self) -> io.IOBase:
root = Path(self.state.manifest.root)
if not await asyncio.to_thread(root.exists):
raise WorkspaceArchiveReadError(
path=root, context={"reason": "workspace_root_not_found"}
)
skip = self._persist_workspace_skip_relpaths()
buf = io.BytesIO()
def _archive_workspace() -> None:
roots = (root, root.resolve(strict=False))
symlinks: list[tarfile.TarInfo] = []
with tarfile.open(fileobj=buf, mode="w") as tar:
def filter_member(member: tarfile.TarInfo) -> tarfile.TarInfo | None:
# tarfile records inodes before filtering. Clear even excluded entries so
# every retained hardlink has its own payload. Unlike dereference=True,
# this preserves symlinks instead of reading their targets on the host.
getattr(tar, "inodes").clear() # noqa: B009 - Not exposed by typeshed.
if should_skip_tar_member(member.name, skip_rel_paths=skip, root_name=None):
return None
if member.isfifo() or member.ischr() or member.isblk():
return None
if member.issym():
symlinks.append(member)
return None
return member
tar.add(root, arcname=".", filter=filter_member)
# Defer symlink headers until capture is complete. The live tree can change
# during tar.add, so only the captured topology can prove containment.
members = {
PurePosixPath(member.name).as_posix(): member
for member in [*tar.getmembers(), *symlinks]
}
for member in symlinks:
# Keep the proof graph unchanged: a hop through an originally absolute
# target must stay unprovable regardless of symlink emission order.
archived_member = copy.copy(member)
if member.linkname.startswith("/"):
rebased = _rebase_symlink_target(
member.linkname, link_name=member.name, roots=roots
)
if rebased != member.linkname and _symlink_target_stays_under(
members, link_name=member.name, target=rebased
):
archived_member.linkname = rebased
tar.addfile(archived_member)
try:
await run_blocking_workspace_io(_archive_workspace)
except (tarfile.TarError, OSError) as e:
raise WorkspaceArchiveReadError(path=root, cause=e) from e
buf.seek(0)
return buf
async def _restore_snapshot_into_workspace_on_resume(self) -> None:
root = Path(self.state.manifest.root)
archive = await self.state.snapshot.restore(dependencies=self.dependencies)
def validate_archive() -> None:
try:
with tarfile.open(fileobj=archive, mode="r:*") as tar:
validate_tarfile(tar, allow_external_symlink_targets=False)
archive.seek(0)
except UnsafeTarMemberError as e:
raise WorkspaceArchiveWriteError(
path=root, context={"reason": e.reason, "member": e.member}, cause=e
) from e
except (tarfile.TarError, OSError) as e:
raise WorkspaceArchiveWriteError(path=root, cause=e) from e
try:
# Older snapshots may contain unsupported members. Reject them before discarding
# the live files; keep hydrate_workspace's own validation for direct callers too.
await run_blocking_workspace_io(validate_archive)
await self._clear_workspace_root_on_resume()
await self.hydrate_workspace(archive)
finally:
with suppress(Exception):
archive.close()
async def hydrate_workspace(self, data: io.IOBase) -> None:
root = Path(self.state.manifest.root)
def _extract_workspace() -> None:
root.mkdir(parents=True, exist_ok=True)
with tarfile.open(fileobj=data, mode="r:*") as tar:
safe_extract_tarfile(
tar,
root=root,
allow_external_symlink_targets=False,
)
try:
await run_blocking_workspace_io(_extract_workspace)
except UnsafeTarMemberError as e:
raise WorkspaceArchiveWriteError(
path=root, context={"reason": e.reason, "member": e.member}, cause=e
) from e
except (tarfile.TarError, OSError) as e:
raise WorkspaceArchiveWriteError(path=root, cause=e) from e