distinct_agent/isolation/request.py

"""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}",
    )