exp: add sandbox isolation lifecycle prototype

experiments/__init__.py

@@ -0,0 +1 @@
+"""Experiments that are intentionally outside the production application path."""

experiments/sandbox_isolation/README.md

@@ -0,0 +1,102 @@
+# Sandbox isolation lifecycle prototype
+
+## Scope
+
+This is an experiment-only prototype derived from PR #15. It is not imported by
+the Flask application, `utils.sandbox_runner`, the online evaluation queue, or
+any deployment entry point. It does not change production configuration and
+contains no credentials.
+
+The current target platform is Windows 11 with Python 3.10+ (validated with the
+workspace Python 3.12 runtime). The prototype intentionally uses an in-memory
+backend instead of Job Object/cgroup APIs. It verifies the lifecycle contract
+that a real adapter must satisfy before untrusted code can run:
+
+1. prepare the isolation boundary;
+2. create the process in a non-running state;
+3. enroll and verify it;
+4. launch only after enrollment succeeds;
+5. collect stdout/stderr with a hard byte bound and bounded EOF waiting;
+6. clean the whole isolation unit on every terminal path;
+7. refuse an unsafe rollback target.
+
+This is a contract and failure-mode prototype, not proof of OS containment.
+The real Windows Job Object and Linux cgroup v2 adapters require a separate
+implementation PR, platform-specific security review, and responsible-owner
+approval before merge or deployment.
+
+## Reproducible commands
+
+From the repository root:
+
+```powershell
+python -m unittest discover -s experiments -t . -p "test_*.py" -v
+python -m experiments.sandbox_isolation.test_prototype
+python -m compileall -q experiments/sandbox_isolation
+```
+
+No `pytest`, compiler, database, network service, or application startup is
+required. The tests use only Python's standard library and write no persistent
+state.
+
+Observed on the Windows workspace with the bundled Python 3.12 runtime:
+
+```text
+python -m unittest discover -s experiments -t . -p "test_*.py" -v
+Ran 9 tests in 0.000s
+OK
+
+python -m experiments.sandbox_isolation.test_prototype
+Ran 9 tests in 0.000s
+OK
+
+python -m compileall -q experiments/sandbox_isolation
+exit code 0
+```
+
+The first exploratory command without `-t .` was intentionally corrected after
+Python reported `ImportError: attempted relative import with no known parent
+package`; the package-aware commands above are the reproducible commands.
+
+## Boundary record
+
+| Boundary | Prototype behavior | Evidence | Not proven |
+| --- | --- | --- | --- |
+| Target platform | Windows 11 + Python 3.12 | standard-library unittest run | Linux runtime behavior |
+| Process/PID | in-memory parent/descendant records; descendants inherit the modeled boundary | `inherit-boundary`, `terminate-isolation-unit`, `verify-empty` events | real PID/job membership |
+| CPU/memory/disk | policy fields only; no quota enforcement | explicit policy object | OS quota enforcement |
+| Output | stdout/stderr capped at 4096 bytes by default; retained handles produce bounded non-pass | output-limit and retained-handle tests | OS pipe behavior under real descendants |
+| Network | policy records `deny-all`; no socket is opened | no network dependency in tests | firewall/namespace/egress enforcement |
+| Filesystem | policy records private workdir only; no files are created | tests leave no persistent state | ACL/mount/namespace enforcement |
+| Permissions | policy records low privilege/no secrets | no credential access | token/capability/ACL enforcement |
+| Rollback | only a verified isolated worker may be selected; otherwise queue is paused | safe/unsafe rollback tests | real queue routing |
+
+## Regression evidence
+
+The tests cover:
+
+- enrollment before launch;
+- enrollment failure with no launch and cleanup attempted;
+- a normal parent result with a surviving descendant;
+- descendants retaining stdout or stderr handles;
+- bounded output overflow as an explicit failure;
+- safe rollback only to a verified worker, and fail-closed pause otherwise.
+
+The `RecordingIsolationBackend` is deliberately observable so a future native
+adapter can reuse the same lifecycle assertions without importing it into the
+production path.
+
+## Risks and unverified items
+
+- The prototype does not create or kill real processes and cannot prove that a
+  real child cannot escape a Job Object/cgroup.
+- It does not enforce CPU, memory, PID, disk, filesystem, network, or Windows
+  permission boundaries.
+- It does not test Windows `CREATE_SUSPENDED`/`AssignProcessToJobObject` or
+  Linux cgroup v2 membership and migration permissions.
+- It does not validate descendants that keep native pipe handles open across a
+  real process exit; the handle behavior is modeled deterministically.
+- It does not change or exercise the online evaluation chain.
+
+Any native OS isolation implementation, production integration, or deployment
+must be proposed in a separate PR and approved by the responsible owner first.

experiments/sandbox_isolation/__init__.py

@@ -0,0 +1,4 @@
+"""Lifecycle-only sandbox isolation prototype.
+
+This package is not imported by the application or the online evaluation path.
+"""

experiments/sandbox_isolation/prototype.py

@@ -0,0 +1,289 @@
+"""A safe, OS-API-free isolation lifecycle prototype.
+
+The prototype models the sequencing and fail-closed contract required before a
+real Job Object/cgroup adapter is approved. It deliberately does not create OS
+process groups, call Job Object/cgroup APIs, change permissions, or connect to
+the application sandbox.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Callable, Iterable, Optional
+
+
+class BoundaryError(RuntimeError):
+    """Raised when the isolation lifecycle cannot proceed safely."""
+
+
+@dataclass(frozen=True)
+class IsolationPolicy:
+    """The boundary contract that a future platform adapter must implement."""
+
+    network: str = "deny-all"
+    filesystem: str = "private-workdir-only"
+    permissions: str = "low-privilege-no-secrets"
+    pid_limit: int = 8
+    output_limit: int = 4096
+
+
+@dataclass(frozen=True)
+class Scenario:
+    """A deterministic scenario used by the lifecycle regression tests."""
+
+    stdout_chunks: tuple[bytes, ...] = (b"42\n",)
+    stderr_chunks: tuple[bytes, ...] = ()
+    exit_code: int = 0
+    descendant: bool = True
+    descendant_holds_stdout: bool = False
+    descendant_holds_stderr: bool = False
+
+
+@dataclass(frozen=True)
+class Capture:
+    data: bytes
+    truncated: bool
+    complete: bool
+    bounded: bool = True
+
+
+@dataclass(frozen=True)
+class RunResult:
+    status: str
+    started: bool
+    launched: bool
+    stdout: Capture
+    stderr: Capture
+    cleanup_verified: bool
+    events: tuple[str, ...]
+    error: str = ""
+
+
+def bounded_capture(
+    chunks: Iterable[bytes], limit: int, *, eof: bool
+) -> Capture:
+    """Capture at most ``limit`` bytes without waiting for an inherited handle.
+
+    ``complete`` is false when a descendant retains the pipe handle. A real
+    runner must then return a non-pass result after its bounded observation
+    window; it must not wait forever for EOF.
+    """
+
+    if limit <= 0:
+        raise ValueError("limit must be positive")
+
+    captured = bytearray()
+    truncated = False
+    for chunk in chunks:
+        remaining = limit - len(captured)
+        if remaining <= 0:
+            truncated = True
+            break
+        if len(chunk) > remaining:
+            captured.extend(chunk[:remaining])
+            truncated = True
+            break
+        captured.extend(chunk)
+
+    return Capture(
+        data=bytes(captured),
+        truncated=truncated,
+        complete=eof,
+    )
+
+
+@dataclass
+class _Process:
+    process_id: str
+    parent_id: Optional[str] = None
+    enrolled: bool = False
+    launched: bool = False
+    alive: bool = True
+
+
+@dataclass
+class RecordingIsolationBackend:
+    """In-memory stand-in for a future Windows/Linux isolation adapter."""
+
+    policy: IsolationPolicy = field(default_factory=IsolationPolicy)
+    fail_stage: Optional[str] = None
+    events: list[str] = field(default_factory=list)
+    boundary_ready: bool = False
+    processes: dict[str, _Process] = field(default_factory=dict)
+    open_pipes: set[str] = field(default_factory=set)
+    _next_id: int = 0
+
+    def _fail_if_requested(self, stage: str) -> None:
+        if self.fail_stage == stage:
+            raise BoundaryError(f"forced failure at {stage}")
+
+    def prepare(self) -> None:
+        self._fail_if_requested("prepare")
+        self.boundary_ready = True
+        self.events.append("prepare-boundary")
+
+    def create_suspended(self) -> str:
+        if not self.boundary_ready:
+            raise BoundaryError("process creation attempted before boundary setup")
+        self._next_id += 1
+        process_id = f"p{self._next_id}"
+        self.processes[process_id] = _Process(process_id=process_id)
+        self.events.append(f"create-suspended:{process_id}")
+        return process_id
+
+    def enroll(self, process_id: str) -> None:
+        self._fail_if_requested("enroll")
+        process = self.processes[process_id]
+        if not self.boundary_ready:
+            raise BoundaryError("enrollment attempted before boundary setup")
+        process.enrolled = True
+        self.events.append(f"enroll:{process_id}")
+
+    def launch(self, process_id: str) -> None:
+        self._fail_if_requested("launch")
+        process = self.processes[process_id]
+        if not process.enrolled:
+            raise BoundaryError("launch attempted before enrollment")
+        process.launched = True
+        self.events.append(f"resume:{process_id}")
+
+    def spawn_descendant(self, parent_id: str) -> str:
+        parent = self.processes[parent_id]
+        if not parent.launched:
+            raise BoundaryError("descendant created before parent launch")
+        self._next_id += 1
+        child_id = f"p{self._next_id}"
+        self.processes[child_id] = _Process(
+            process_id=child_id,
+            parent_id=parent_id,
+            enrolled=True,
+            launched=True,
+        )
+        self.events.append(f"inherit-boundary:{child_id}")
+        return child_id
+
+    def hold_pipe(self, process_id: str, stream: str) -> None:
+        if process_id not in self.processes:
+            raise BoundaryError(f"unknown process {process_id}")
+        self.open_pipes.add(f"{process_id}:{stream}")
+        self.events.append(f"hold-pipe:{process_id}:{stream}")
+
+    def cleanup(self) -> None:
+        self._fail_if_requested("cleanup")
+        self.events.append("terminate-isolation-unit")
+        for process in self.processes.values():
+            process.alive = False
+        self.events.append("close-output-handles")
+        self.open_pipes.clear()
+        self.processes.clear()
+        self.boundary_ready = False
+        self.events.append("verify-empty")
+
+    def is_empty(self) -> bool:
+        return not self.processes and not self.open_pipes and not self.boundary_ready
+
+
+ScenarioHook = Callable[[RecordingIsolationBackend, str], None]
+
+
+class LifecycleRunner:
+    """Runs the lifecycle contract and always attempts unit cleanup."""
+
+    def __init__(self, backend: RecordingIsolationBackend):
+        self.backend = backend
+
+    def execute(self, scenario: Scenario, hook: Optional[ScenarioHook] = None) -> RunResult:
+        stdout = bounded_capture((), self.backend.policy.output_limit, eof=True)
+        stderr = bounded_capture((), self.backend.policy.output_limit, eof=True)
+        started = False
+        launched = False
+        process_id: Optional[str] = None
+        status = "failed"
+        error = ""
+
+        try:
+            self.backend.prepare()
+            process_id = self.backend.create_suspended()
+            self.backend.enroll(process_id)
+            self.backend.launch(process_id)
+            started = True
+            launched = True
+
+            if scenario.descendant:
+                descendant_id = self.backend.spawn_descendant(process_id)
+                if scenario.descendant_holds_stdout:
+                    self.backend.hold_pipe(descendant_id, "stdout")
+                if scenario.descendant_holds_stderr:
+                    self.backend.hold_pipe(descendant_id, "stderr")
+
+            if hook is not None:
+                hook(self.backend, process_id)
+
+            stdout = bounded_capture(
+                scenario.stdout_chunks,
+                self.backend.policy.output_limit,
+                eof=not scenario.descendant_holds_stdout,
+            )
+            stderr = bounded_capture(
+                scenario.stderr_chunks,
+                self.backend.policy.output_limit,
+                eof=not scenario.descendant_holds_stderr,
+            )
+            expected = stdout.data.rstrip().decode("utf-8", errors="replace") == "42"
+            if scenario.exit_code != 0:
+                status = "runtime_error"
+            elif stdout.truncated or stderr.truncated:
+                status = "output_limit_exceeded"
+            elif not stdout.complete or not stderr.complete:
+                status = "output_collection_incomplete"
+            elif expected:
+                status = "passed"
+            else:
+                status = "wrong_output"
+        except BoundaryError as exc:
+            status = "isolation_setup_failed"
+            error = str(exc)
+        finally:
+            try:
+                self.backend.cleanup()
+            except BoundaryError as exc:
+                status = "cleanup_failed"
+                error = str(exc)
+
+        cleanup_verified = self.backend.is_empty()
+        if not cleanup_verified and status != "cleanup_failed":
+            status = "cleanup_failed"
+            error = "isolation unit is not empty after cleanup"
+
+        return RunResult(
+            status=status,
+            started=started,
+            launched=launched,
+            stdout=stdout,
+            stderr=stderr,
+            cleanup_verified=cleanup_verified,
+            events=tuple(self.backend.events),
+            error=error,
+        )
+
+
+@dataclass(frozen=True)
+class Worker:
+    name: str
+    isolation_verified: bool
+    available: bool = True
+
+
+@dataclass(frozen=True)
+class RouteDecision:
+    status: str
+    worker: Optional[str]
+
+
+def choose_rollback_worker(preferred: Worker, fallback: Optional[Worker]) -> RouteDecision:
+    """Fail closed when no verified isolated rollback target exists."""
+
+    for candidate in (preferred, fallback):
+        if candidate is not None and candidate.available and candidate.isolation_verified:
+            return RouteDecision(status="routed", worker=candidate.name)
+    return RouteDecision(status="paused_no_safe_rollback", worker=None)