"""One disposable sandbox per request, not one per worker. Why this exists --------------- The worker used to hold a single isolation posture for its whole lifetime: a Job Object around the model process, and Deno permission flags around generated code. Two requests from two strangers ran with the same ambient state, in the same directories, under the same identity. That is wrong for a reason that has nothing to do with whether Deno's flags hold. Pydantic archived their own Deno-plus-Pyodide runner in January 2026 with this as the first listed failure mode: code running in a shared runtime can **taint it to control later invocations**. A worker that reuses a sandbox is offering user B whatever user A left behind, and it is doing so on a volunteer's desktop where nobody is watching. So the unit of isolation is now the request. What a fresh sandbox means here ------------------------------- Every request gets, and only gets: * a **new OS process**, never a reused interpreter; * a **new empty writable directory**, created for it and deleted after it, which is the only path it may write; * a **read-only view** of the shared runtime, never a per-request copy; * **no network**, expressed as an absence of capability rather than as a filter; * a **cleared environment**, with only what was explicitly injected; * a **hard teardown** that cannot leave survivors. Per platform ------------ **Windows: AppContainer, plus the existing Job Object.** An AppContainer (a LowBox token) is default-deny for the filesystem, the registry, other apps and the network; a profile created with *zero* capabilities has no network at all, with no firewall rule, no local account and no elevation. That last part is what makes it right for this project: OpenAI's Codex sandbox and Anthropic's `srt` both take the restricted-token route on Windows, and both need a one-time UAC-elevated install that creates local user accounts. Asking a volunteer to do that to lend a spare desktop is asking too much. AppContainer needs neither. The Job Object stays, because it does the other half of the job: process count, memory, CPU time, UI restrictions, and kill-on-close teardown. A Job Object is not a security boundary and is not described as one anywhere here. **Linux: bubblewrap.** Unprivileged, no daemon, and the only widely available option that gives a genuinely fresh filesystem *view* rather than a filter over the real one. ``--unshare-all`` without ``--share-net`` leaves an empty network namespace, which is stronger than blocking syscalls: there is no interface to reach. Landlock is the fallback where unprivileged user namespaces are restricted (Ubuntu 24.04 and later, unless an AppArmor profile permits them), and it is a weaker fallback, which the report says. Honesty about what this is -------------------------- AppContainer and bubblewrap are the class of boundary Chrome and Flatpak rely on: well attacked, good, occasionally broken. The genuine step up is a virtual machine, which needs Hyper-V or KVM and therefore administrator rights, which this project has ruled out. Assume containment can fail: the worker holds no secret the sandbox would want, and the server is told what was actually applied rather than what was requested. Execution status ---------------- **The bubblewrap path has been executed** on a Linux kernel with unprivileged user namespaces available, and each property was observed rather than assumed: an empty root containing only what was bound, no view of the operator's home directory, a read-only ``/usr`` that refuses a write, a cleared environment, an outbound connection that fails because there is no interface, and a scratch directory that is gone after the context manager exits. ``tests/test_request_sandbox.py`` performs exactly those checks and skips where ``bwrap`` is absent, rather than passing vacuously. Two things remain unverified and are reported as such rather than claimed. **The AppContainer path has never been run**, because creating a profile needs a Windows kernel; it is written against the documented API and every entry point returns a reason instead of raising. **The distributions that restrict unprivileged user namespaces** (Ubuntu 24.04 and later) have not been tested; the restriction is detected and explained, but the degraded path it leads to has not been exercised on such a host. ``RestrictionReport.applied`` and ``SandboxReport.confinement_applied`` are set only after the platform layer confirms each step, so an unverified path reports itself as not applied rather than being assumed to work. """ from __future__ import annotations import os import shutil import subprocess import sys import tempfile import uuid from collections.abc import Mapping, Sequence from dataclasses import dataclass, field from pathlib import Path from typing import Any from .base import ProcessRestriction, RestrictionReport, spawn_restricted #: Environment variables a sandboxed child is allowed to inherit. #: #: Everything else is dropped. The list is short on purpose: each entry is #: something a runtime genuinely cannot start without, and adding to it is a #: decision to widen what a request can see about the machine it landed on. INHERITABLE_ENVIRONMENT = ("PATH", "SYSTEMROOT", "COMSPEC", "TMP", "TEMP", "LANG", "LC_ALL") @dataclass(frozen=True) class SandboxReport: """What one request's sandbox actually got, layer by layer. Kept separate from :class:`~distinct_agent.isolation.base.RestrictionReport` because a request has two layers and they can succeed independently. A single boolean would hide the case that matters most: resource limits applied, privilege reduction not. """ request_id: str platform: str #: The outer, privilege-reducing layer: AppContainer, bubblewrap, or none. confinement: str confinement_applied: bool #: The resource-control layer: Job Object, cgroup scope, rlimits, or none. limits: str limits_applied: bool scratch: str network: str reason: str | None = None @property def summary(self) -> str: parts = [ f"{self.confinement} {'applied' if self.confinement_applied else 'NOT applied'}", f"{self.limits} {'applied' if self.limits_applied else 'NOT applied'}", f"network {self.network}", ] if self.reason: parts.append(self.reason) return "; ".join(parts) def to_dict(self) -> dict[str, Any]: return { "request_id": self.request_id, "platform": self.platform, "confinement": self.confinement, "confinement_applied": self.confinement_applied, "limits": self.limits, "limits_applied": self.limits_applied, "scratch": self.scratch, "network": self.network, "reason": self.reason, } @dataclass class RequestSandbox: """A disposable execution context owned by exactly one request. Use it as a context manager. On exit the scratch directory is removed and any platform handle is released, in that order, so nothing survives that could be handed to the next request. """ request_id: str scratch: Path restriction: ProcessRestriction read_only_paths: tuple[Path, ...] = () _closed: bool = field(default=False, repr=False) _handles: list[Any] = field(default_factory=list, repr=False) _confinement: Any = field(default=None, repr=False) # Construction ------------------------------------------------------ @classmethod def create( cls, *, request_id: str | None = None, root: Path | str | None = None, read_only_paths: Sequence[Path | str] = (), restriction: ProcessRestriction | None = None, ) -> RequestSandbox: identifier = request_id or uuid.uuid4().hex base = Path(root) if root is not None else Path(tempfile.gettempdir()) / "distinct" # One shallow directory, created fresh. Codex measured ACL stamping # across nested trees as the dominant per-request cost on Windows, so # the writable area is deliberately flat and small. scratch = base / f"request-{identifier}" scratch.mkdir(parents=True, exist_ok=False) return cls( request_id=identifier, scratch=scratch, restriction=restriction or ProcessRestriction(mode="auto"), read_only_paths=tuple(Path(path) for path in read_only_paths), ) # Environment ------------------------------------------------------- def environment(self, extra: Mapping[str, str] | None = None) -> dict[str, str]: """A cleared environment with the scratch directory as its temp space. Inheriting the operator's environment hands a request their proxy credentials, their API keys and the shape of their machine. It is cleared rather than filtered for hostile values, because a deny list of variable names is a list somebody has to keep complete. """ environment = { name: os.environ[name] for name in INHERITABLE_ENVIRONMENT if name in os.environ } environment["TMPDIR"] = str(self.scratch) environment["TMP"] = str(self.scratch) environment["TEMP"] = str(self.scratch) environment.update(dict(extra or {})) return environment # Execution --------------------------------------------------------- def spawn( self, command: Sequence[str], *, extra_environment: Mapping[str, str] | None = None, **popen_kwargs: Any, ) -> tuple[subprocess.Popen, SandboxReport]: """Start ``command`` inside this request's sandbox. Returns the process and the report of what was actually applied. The report is not a prediction: each layer sets its own flag only after the platform confirmed it, and a layer that could not be applied says so rather than being assumed. """ if self._closed: raise RuntimeError("this request's sandbox has already been torn down") environment = self.environment(extra_environment) popen_kwargs.setdefault("cwd", str(self.scratch)) popen_kwargs["env"] = environment if sys.platform.startswith("linux"): wrapped, confinement, applied, reason = self._linux_wrap(command) elif sys.platform == "win32": wrapped, confinement, applied, reason = self._windows_wrap(command) else: wrapped, confinement, applied, reason = ( list(command), "none", False, f"no request-level confinement backend exists for {sys.platform!r}", ) process, limits_report, guard = spawn_restricted( wrapped, self.restriction, **popen_kwargs ) self._handles.append(guard) return process, SandboxReport( request_id=self.request_id, platform=sys.platform, confinement=confinement, confinement_applied=applied, limits=limits_report.mechanism, limits_applied=limits_report.applied, scratch=str(self.scratch), network="denied" if applied else "not confined at the OS layer", reason=reason or limits_report.reason, ) # Platform wrapping ------------------------------------------------- def _linux_wrap(self, command: Sequence[str]) -> tuple[list[str], str, bool, str | None]: """Wrap in bubblewrap when it is usable, and say so when it is not.""" bwrap = shutil.which("bwrap") if not bwrap: return ( list(command), "bubblewrap", False, "bwrap is not installed, so this request ran without filesystem " "or network confinement", ) blocked = _unprivileged_userns_blocked() if blocked: return list(command), "bubblewrap", False, blocked wrapped = [ bwrap, # Every namespace, including the network one. No --share-net: an # empty network namespace has no interface to reach, which is a # stronger statement than a syscall filter that must enumerate. "--unshare-all", "--die-with-parent", # setsid, which also closes the TIOCSTI terminal-injection route. "--new-session", "--clearenv", "--tmpfs", "/", "--proc", "/proc", "--dev", "/dev", ] for path in self.read_only_paths: resolved = Path(path).resolve() if resolved.exists(): wrapped += ["--ro-bind", str(resolved), str(resolved)] # Merged-usr layouts put the loader at /lib64 and the tools at /bin, # both symlinks into /usr that --tmpfs / has just removed. Recreating # them costs nothing and is the difference between a sandbox that runs # and one that reports "No such file or directory" for a binary that is # plainly there: the missing file is the ELF interpreter, not the # program, which is a confusing error to debug from the outside. wrapped += _merged_usr_symlinks() wrapped += [ "--bind", str(self.scratch.resolve()), "/work", "--chdir", "/work", "--", ] return wrapped + _resolved_command(command), "bubblewrap", True, None def _windows_wrap(self, command: Sequence[str]) -> tuple[list[str], str, bool, str | None]: """Prepare an AppContainer for this request, if the kernel allows one. The command itself is not rewritten: on Windows the confinement is a property of the token the process is created with, not a wrapper program. The profile is created here and consumed by the spawn layer. """ from .windows_appcontainer import create_profile profile, reason = create_profile(self.request_id, grant_paths=(self.scratch,)) if profile is None: return list(command), "windows-appcontainer", False, reason self._confinement = profile return list(command), "windows-appcontainer", True, None # Teardown ---------------------------------------------------------- def close(self) -> None: """Release every handle and delete the scratch directory. Ordering matters. The job handle is closed first, which kills the process tree, so nothing is still writing when the directory is removed. Deletion is best-effort by necessity: a file held open by a process that has already been killed can linger briefly on Windows, and raising here would turn a tidy-up problem into a failed run. """ if self._closed: return self._closed = True for handle in reversed(self._handles): try: handle.close() except Exception: continue self._handles.clear() if self._confinement is not None: try: self._confinement.close() except Exception: pass self._confinement = None shutil.rmtree(self.scratch, ignore_errors=True) def __enter__(self) -> RequestSandbox: return self def __exit__(self, *exc: object) -> None: self.close() def _merged_usr_symlinks() -> list[str]: """Recreate the merged-usr top-level symlinks inside the empty root.""" arguments: list[str] = [] for name in ("bin", "sbin", "lib", "lib32", "lib64"): top = Path("/") / name target = Path("/usr") / name if top.is_symlink() and target.exists(): arguments += ["--symlink", f"usr/{name}", f"/{name}"] return arguments def _resolved_command(command: Sequence[str]) -> list[str]: """Resolve the executable so the sandbox runs the file, not a symlink chain. ``/usr/bin/python3`` is commonly a symlink into ``/etc/alternatives``, which a sandbox that binds only ``/usr`` cannot follow. Resolving here means the caller binds what it means to bind, and it also means the sandbox executes exactly the file it decided on rather than whatever the link points at by the time the child starts. """ parts = list(command) if not parts: return parts located = shutil.which(parts[0]) if located: try: parts[0] = str(Path(located).resolve()) except OSError: parts[0] = located return parts def _unprivileged_userns_blocked() -> str | None: """Ubuntu 24.04 and later restrict unprivileged user namespaces. Returned as a sentence rather than a boolean because the operator can fix it, and a report that says "not applied" without saying why leaves them nothing to act on. """ knob = Path("/proc/sys/kernel/apparmor_restrict_unprivileged_userns") try: if knob.is_file() and knob.read_text(encoding="utf-8").strip() == "1": return ( "unprivileged user namespaces are restricted by AppArmor on this " "kernel, so bubblewrap cannot build a sandbox. Install an " "/etc/apparmor.d/bwrap profile granting 'userns,' to enable it." ) except OSError: return None return None def describe_request_isolation() -> RestrictionReport: """What this host can enforce per request, without spawning anything.""" if sys.platform == "win32": from .windows_appcontainer import describe_appcontainer_support return describe_appcontainer_support() if sys.platform.startswith("linux"): blocked = _unprivileged_userns_blocked() if not shutil.which("bwrap"): return RestrictionReport( platform=sys.platform, applied=False, mechanism="bubblewrap", reason="bwrap is not installed", ) if blocked: return RestrictionReport( platform=sys.platform, applied=False, mechanism="bubblewrap", reason=blocked ) return RestrictionReport( platform=sys.platform, applied=False, mechanism="bubblewrap", reason=( "bwrap is present and unprivileged namespaces are permitted, but " "no sandbox has been built on this host yet, so nothing is claimed" ), ) return RestrictionReport( platform=sys.platform, applied=False, mechanism="none", reason=f"no request-level confinement backend exists for {sys.platform!r}", )