Add bounded stdin to owner execution

Assistant: codex
Assistant-Model: gpt-5.6-sol
Assistant-Session: 01a06def-6490-7033-8448-2eab2d12ed44

src/sandboxer/extensions/base.py

@@ -60,6 +60,7 @@ class SandboxExtension(ABC):
         execution_context: dict[str, str],
         timeout_seconds: int,
         max_output_bytes: int,
+        stdin_text: str | None = None,
     ) -> dict[str, object]:
         """Run a bounded command through an owner-mediated sandbox boundary."""
         raise NotImplementedError(f"{type(self).__name__} does not support execution")

src/sandboxer/extensions/bwrap.py

@@ -184,6 +184,7 @@ class BwrapExtension(SandboxExtension):
         execution_context: dict[str, str],
         timeout_seconds: int,
         max_output_bytes: int,
+        stdin_text: str | None = None,
     ) -> dict[str, object]:
         """Ask the broker already inside bwrap to run an argument-vector command."""
         pid = int(handle.get("pid", "0"))
@@ -197,6 +198,7 @@ class BwrapExtension(SandboxExtension):
             "execution_context": execution_context,
             "timeout_seconds": timeout_seconds,
             "max_output_bytes": max_output_bytes,
+            "stdin_text": stdin_text,
         }
         response_limit = max_output_bytes * 2 + 65_536
         chunks: list[bytes] = []

src/sandboxer/extensions/bwrap_runner.py

@@ -10,7 +10,9 @@ import subprocess
 import sys
 from pathlib import Path
 
-_MAX_REQUEST_BYTES = 1_048_576
+# A validated 1 MB stdin can expand sixfold when JSON escapes control bytes;
+# leave bounded headroom for the argv, route references, and execution context.
+_MAX_REQUEST_BYTES = 7_000_000
 
 
 def _bounded_output(value: bytes | None, limit: int) -> tuple[str, bool]:
@@ -27,6 +29,7 @@ def _run(payload: dict, workspace: Path) -> dict[str, object]:
     max_output_bytes = int(payload["max_output_bytes"])
     credential_refs = payload.get("credential_route_refs", [])
     context = payload.get("execution_context", {})
+    stdin_text = payload.get("stdin_text")
     child_env = {
         "HOME": str(workspace),
         "LANG": "C.UTF-8",
@@ -39,12 +42,16 @@ def _run(payload: dict, workspace: Path) -> dict[str, object]:
         command,
         cwd=workspace,
         env=child_env,
+        stdin=subprocess.PIPE,
         stdout=subprocess.PIPE,
         stderr=subprocess.PIPE,
         start_new_session=True,
     )
     try:
-        stdout_raw, stderr_raw = process.communicate(timeout=timeout_seconds)
+        stdin_bytes = stdin_text.encode("utf-8") if stdin_text is not None else None
+        stdout_raw, stderr_raw = process.communicate(
+            input=stdin_bytes, timeout=timeout_seconds
+        )
         exit_code = process.returncode
     except subprocess.TimeoutExpired:
         timed_out = True

src/sandboxer/extensions/registry.py

@@ -36,6 +36,7 @@ class ExtensionBackend(Protocol):
         execution_context: dict[str, str],
         timeout_seconds: int,
         max_output_bytes: int,
+        stdin_text: str | None = None,
     ) -> dict[str, object]: ...