Confine metered runs to an ephemeral owner Messages route

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a07ff8-19d0-7820-b4d0-1353833cb7fc

src/sandboxer/core/manager.py

@@ -6,6 +6,7 @@ import re
 from pathlib import Path
 from threading import Lock
 
+from sandboxer.extensions.messages_route import OwnerMessagesRoute
 from sandboxer.extensions.registry import load_extension, resolve_backend
 from sandboxer.lifecycle.expire import (
     ExpireCandidate,
@@ -52,10 +53,14 @@ class SandboxManager:
         store: SandboxStore | None = None,
         credits: CreditsStore | None = None,
         snapshots: SnapshotStore | None = None,
+        *,
+        messages_route: OwnerMessagesRoute | None = None,
     ) -> None:
         self.store = store or SandboxStore()
         self.credits = credits or CreditsStore()
         self.snapshots = snapshots or SnapshotStore()
+        self._messages_route = messages_route
+        self._messages_created = False
         self._exec_locks: dict[str, Lock] = {}
 
     @staticmethod
@@ -101,6 +106,12 @@ class SandboxManager:
         profile = load_profile(request.profile)
         extension = resolve_extension(profile, request.inputs, host_override=host)
         backend = resolve_backend(extension)
+        if self._messages_route is not None:
+            self._messages_route.validate(profile, request.consumer, backend, request.inputs)
+            if self._messages_created:
+                raise ValueError("metered owner manager is single-run and single-sandbox")
+            self._messages_created = True
+            backend.messages_route = self._messages_route
         resolved_host = self._resolved_host(profile, extension, host)
         wants_telemetry = profile_wants_telemetry(profile)
         base_dir = extension.config.get("base_dir", "/tmp/sandboxer")
@@ -276,6 +287,11 @@ class SandboxManager:
         profile = load_profile(status.profile_id)
         extension = load_extension(status.extension_id)
         backend = resolve_backend(extension)
+        if self._messages_route is not None:
+            self._messages_route.validate(profile, status.consumer, backend, status.inputs)
+            if request.credential_route_refs:
+                raise ValueError("metered execution refuses provider credential delivery")
+            backend.messages_route = self._messages_route
         if not backend.supports_execution():
             raise RuntimeError(
                 f"Extension {status.extension_id} has no owner-mediated execution boundary"

src/sandboxer/extensions/bwrap.py

@@ -17,6 +17,7 @@ from typing import Any
 
 from sandboxer.extensions.base import SandboxExtension
 from sandboxer.extensions.egress import destinations
+from sandboxer.extensions.messages_route import OwnerMessagesRoute
 from sandboxer.extensions.runtime import RUNTIME_MOUNT, verified_runtime
 from sandboxer.models import Profile
 
@@ -35,6 +36,7 @@ class BwrapExtension(SandboxExtension):
 
     def __init__(self, config: dict[str, Any] | None = None) -> None:
         super().__init__(config)
+        self.messages_route: OwnerMessagesRoute | None = None
         cfg = self.config
         self.base_dir: str = cfg.get("base_dir", "/tmp/sandboxer-bwrap")
         self.bwrap_bin: str = cfg.get("bwrap_bin", "bwrap")
@@ -75,9 +77,16 @@ class BwrapExtension(SandboxExtension):
         runner = Path(__file__).with_name("bwrap_runner.py")
         argv += ["--dir", "/run", "--dir", "/run/sandboxer"]
         argv += ["--ro-bind", str(runner), "/run/sandboxer/bwrap_runner.py"]
-        if egress_socket:
+        if self.messages_route is not None:
+            if egress_socket:
+                raise ValueError("metered route cannot coexist with CONNECT egress")
+            argv += [
+                "--ro-bind", str(self.messages_route.socket_path), "/run/sandboxer/messages.sock"
+            ]
+        if egress_socket or self.messages_route is not None:
             egress_module = Path(__file__).with_name("egress.py")
             argv += ["--ro-bind", str(egress_module), "/run/sandboxer/egress.py"]
+        if egress_socket:
             argv += ["--ro-bind", egress_socket, "/run/sandboxer/egress.sock"]
         runtime = verified_runtime(self.config)
         if runtime is not None:
@@ -98,6 +107,8 @@ class BwrapExtension(SandboxExtension):
             argv.append("--runtime")
         if egress_socket:
             argv.append("--egress")
+        if self.messages_route is not None:
+            argv.append("--messages")
         return argv
 
     @staticmethod
@@ -211,7 +222,7 @@ class BwrapExtension(SandboxExtension):
         try:
             return self._wait_ready(handle)
         except Exception:
-            if handle.get("egress_pid"):
+            if handle.get("egress_pid") or self.messages_route is not None:
                 self.teardown(handle)
             raise
 
@@ -276,6 +287,10 @@ class BwrapExtension(SandboxExtension):
             "max_output_bytes": max_output_bytes,
             "stdin_text": stdin_text,
         }
+        if self.messages_route is not None:
+            if credential_route_refs:
+                raise ValueError("metered route refuses provider credentials")
+            request["messages_token"] = self.messages_route.token
         if credential_route_refs:
             from sandboxer.extensions.credential_delivery import execute
 

src/sandboxer/extensions/bwrap_runner.py

@@ -4,6 +4,7 @@ from __future__ import annotations
 
 import json
 import os
+import re
 import signal
 import socket
 import subprocess
@@ -24,7 +25,8 @@ def _bounded_output(value: bytes | None, limit: int) -> tuple[str, bool]:
 
 
 def _run(
-    payload: dict, workspace: Path, *, runtime_enabled: bool = False, proxy_port: int | None = None
+    payload: dict, workspace: Path, *, runtime_enabled: bool = False, proxy_port: int | None = None,
+    messages_port: int | None = None,
 ) -> dict[str, object]:
     command = payload["command"]
     timeout_seconds = int(payload["timeout_seconds"])
@@ -63,6 +65,16 @@ def _run(
         )
     ):
         raise ValueError("invalid owner credential envelope")
+    messages_token = payload.get("messages_token")
+    if messages_port is not None:
+        if (proxy_port is not None or credential_env or credential_refs
+                or not isinstance(messages_token, str)
+                or not re.fullmatch(r"[A-Za-z0-9_-]{43,100}", messages_token)):
+            raise ValueError("invalid metered owner envelope")
+        child_env["ANTHROPIC_BASE_URL"] = f"http://127.0.0.1:{messages_port}"
+        credential_env = {"ANTHROPIC_API_KEY": messages_token}
+    elif messages_token is not None:
+        raise ValueError("metered owner route unavailable")
     child_env.update(credential_env)
     timed_out = False
     process = subprocess.Popen(
@@ -103,13 +115,20 @@ def main() -> int:
     workspace = Path(sys.argv[1]).resolve(strict=True)
     socket_path = Path(sys.argv[2])
     runtime_enabled = "--runtime" in sys.argv[3:]
-    if any(flag not in {"--runtime", "--egress"} for flag in sys.argv[3:]):
+    if any(flag not in {"--runtime", "--egress", "--messages"} for flag in sys.argv[3:]):
         raise ValueError("invalid owner runtime mode")
     proxy_port = None
     if "--egress" in sys.argv[3:]:
         from egress import bridge
 
         proxy_port = bridge("/run/sandboxer/egress.sock")
+    messages_port = None
+    if "--messages" in sys.argv[3:]:
+        if proxy_port is not None:
+            raise ValueError("metered route cannot coexist with CONNECT egress")
+        from egress import bridge
+
+        messages_port = bridge("/run/sandboxer/messages.sock")
     state = Path("/run/sandboxer/state")
     state.mkdir(mode=0o700)
     for name in ("home", "config", "cache", "data", "tmp"):
@@ -141,6 +160,7 @@ def main() -> int:
                         workspace,
                         runtime_enabled=runtime_enabled,
                         proxy_port=proxy_port,
+                        messages_port=messages_port,
                     )
                 except Exception as exc:
                     response = {"boundary_error": str(exc)}

src/sandboxer/extensions/messages_route.py

@@ -0,0 +1,65 @@
+"""Ephemeral trusted-owner binding; never a profile, API or stored secret."""
+
+from __future__ import annotations
+
+import os
+import re
+import stat
+from dataclasses import dataclass, field
+from pathlib import Path
+
+
+@dataclass(frozen=True)
+class OwnerMessagesRoute:
+    socket_path: Path = field(repr=False)
+    token: str = field(repr=False)
+    profile_id: str
+    actor: str
+    project: str
+    run_id: str
+    private_paths: tuple[Path, ...] = field(default=(), repr=False)
+
+    def validate(self, profile, consumer, backend, inputs) -> None:
+        from sandboxer.extensions.bwrap import BwrapExtension
+
+        if type(backend) is not BwrapExtension:
+            raise ValueError("metered route requires the bwrap owner")
+        if (
+            profile.id != self.profile_id
+            or consumer.actor.value != self.actor
+            or consumer.project != self.project
+            or consumer.run_id != self.run_id
+            or not self.run_id
+        ):
+            raise ValueError("metered route consumer mismatch")
+        if profile.setup.secret_refs:
+            raise ValueError("metered route refuses setup credential acquisition")
+        if profile.network.default != "deny" or profile.network.egress:
+            raise ValueError("metered route requires empty default-deny egress")
+        if backend.config.get("allowed_egress") or backend.config.get("credential_routes"):
+            raise ValueError("metered route cannot coexist with provider egress or credentials")
+        if not re.fullmatch(r"[A-Za-z0-9_-]{43,100}", self.token):
+            raise ValueError("invalid metered route token")
+        path = self.socket_path
+        metadata = path.lstat()
+        parent = path.parent.lstat()
+        if (
+            not path.is_absolute() or path.resolve() != path
+            or not stat.S_ISSOCK(metadata.st_mode) or metadata.st_uid != os.getuid()
+            or stat.S_IMODE(metadata.st_mode) != 0o600
+            or parent.st_uid != os.getuid() or stat.S_IMODE(parent.st_mode) != 0o700
+        ):
+            raise ValueError("metered route requires a private owner socket")
+        safe_binds = {"/usr", "/bin", "/lib", "/lib64", "/etc/resolv.conf", "/etc/ssl/certs"}
+        if not set(backend.ro_binds).issubset(safe_binds):
+            raise ValueError("metered route refuses additional host mounts")
+        exposed = [Path(backend.base_dir), *map(Path, backend.ro_binds)]
+        if inputs.get("repo"):
+            exposed.append(Path(inputs["repo"]))
+        if backend.config.get("runtime"):
+            exposed.append(Path(backend.config["runtime"]["path"]))
+        for private in (path.parent, *self.private_paths):
+            for visible in exposed:
+                a, b = private.resolve(), visible.resolve()
+                if a.is_relative_to(b) or b.is_relative_to(a):
+                    raise ValueError("metered owner state overlaps a workload mount")