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:
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 = 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 = 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 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 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 _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"],
path: Path,
*,
user: str | User,
payload: bytes = b"",
) -> subprocess.CompletedProcess[bytes]:
# Authorization is synchronous and captured for this operation before dispatch.
path = self._files.authorize(path, for_write=operation == "write")
command = self._prepare_exec_command(
"python3",
"-I",
"-S",
"-c",
_USER_FILE_WORKER_SOURCE,
operation,
str(path),
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 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:
with tarfile.open(fileobj=buf, mode="w") as tar:
tar.add(
root,
arcname=".",
filter=lambda ti: (
None
if should_skip_tar_member(
ti.name,
skip_rel_paths=skip,
root_name=None,
)
else ti
),
)
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 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