fix: execute local reins through the sandbox owner

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a0726e-5232-73f2-aaca-2c05ceb62efb

scripts/prove-owner-boundary.py

@@ -0,0 +1,141 @@
+"""Bounded non-secret gateway proof; does not certify a production rein runtime.
+
+Run with .venv/bin/python scripts/prove-owner-boundary.py.
+"""
+
+import json
+import subprocess
+import tempfile
+from pathlib import Path
+
+from sandboxer.core.manager import SandboxManager
+from sandboxer.lifecycle.store import SandboxStore
+from sandboxer.payments.credits import CreditsStore
+from sandboxer.snapshots.store import SnapshotStore
+
+from glas_harness.contract import ExecutionRequest, OperationalReadiness, ToolResult
+from glas_harness.gateway import run_execution
+from glas_harness.profiles import ProfileCatalog
+from glas_harness.reins.rein_aharness import ReinAharness
+
+
+class BoundaryProof(ReinAharness):
+    """Use real adapter setup/cleanup, with a deterministic non-agent dispatch."""
+
+    def __init__(self, source):
+        super().__init__()
+        self.source = source
+        self.proof = {}
+        self.workspace = None
+        self.task_removed = False
+
+    def start_session(self, profile, inputs, sandbox):
+        self.workspace = sandbox.reachability.get("workspace_dir")
+        return super().start_session(profile, inputs, sandbox)
+
+    def dispatch_tool(self, session, tool_call):
+        transport = session["transport"]
+        code = r"""
+import json, os, stat, sys
+from pathlib import Path
+source, task = map(Path, sys.argv[1:])
+assert not source.exists(), 'source checkout visible'
+try:
+    (source / 'sentinel').write_text('escaped')
+except OSError:
+    pass
+else:
+    raise AssertionError('source mutation succeeded')
+assert Path('sentinel').read_text() == 'unchanged'
+assert stat.S_IMODE(task.stat().st_mode) == 0o600
+assert json.loads(task.read_text())['description'] == 'non-secret boundary probe'
+Path('proof.txt').write_text('sandbox copy only\n')
+interfaces = [line.split(':', 1)[0].strip()
+              for line in Path('/proc/net/dev').read_text().splitlines()[2:]]
+assert interfaces == ['lo']
+assert os.environ['SANDBOXER_ACTOR'] == 'agt'
+assert os.environ['SANDBOXER_PROJECT'] == 'glas-boundary-proof'
+assert os.environ['SANDBOXER_RUN_ID'] == 'glas-wp-0005-boundary-proof'
+print(json.dumps({'source_absent': True, 'source_mutation_refused': True,
+                  'task_mode': '0600', 'interfaces': interfaces,
+                  'cwd': str(Path.cwd()), 'identity_exact': True}))
+"""
+        proc = transport.run(
+            ["/usr/bin/python3", "-c", code, str(self.source), session["task_file"]],
+            timeout=30,
+        )
+        if proc.returncode:
+            return ToolResult(ok=False, error=proc.stderr)
+        self.proof = json.loads(proc.stdout)
+        for command in (["git", "add", "proof.txt"],
+                        ["git", "commit", "-q", "-m", "non-secret boundary proof"]):
+            result = transport.run(command, timeout=30)
+            if result.returncode:
+                return ToolResult(ok=False, error=result.stderr)
+        return ToolResult(ok=True, output=proc.stdout)
+
+    def cleanup_session(self, session):
+        super().cleanup_session(session)
+        proc = session["transport"].run(
+            ["test", "!", "-e", session["task_file"]], timeout=30
+        )
+        self.task_removed = proc.returncode == 0
+        assert self.task_removed
+
+
+def main():
+    with tempfile.TemporaryDirectory(prefix="glas-owner-proof-") as root:
+        root = Path(root)
+        source = root / "source"
+        source.mkdir()
+        for command in (["git", "init", "-q"],
+                        ["git", "config", "user.name", "Glas Boundary Proof"],
+                        ["git", "config", "user.email", "proof@example.invalid"]):
+            subprocess.run(command, cwd=source, check=True)
+        (source / "sentinel").write_text("unchanged")
+        subprocess.run(["git", "add", "sentinel"], cwd=source, check=True)
+        subprocess.run(["git", "commit", "-qm", "fixture"], cwd=source, check=True)
+        source_head = subprocess.check_output(["git", "rev-parse", "HEAD"], cwd=source)
+        manager = SandboxManager(
+            store=SandboxStore(path=root / "sandboxes.json"),
+            credits=CreditsStore(path=root / "credits.json"),
+            snapshots=SnapshotStore(path=root / "snapshots.json"),
+        )
+        catalog = ProfileCatalog()
+        profile, _ = catalog.resolve("harness.agent-dev-local@1.0.0")
+        # Only this in-memory injected non-agent fixture is eligible for proof.
+        catalog.profiles()[(profile.id, profile.version)] = profile.model_copy(update={
+            "operational_readiness": OperationalReadiness(
+                status="unverified", reason="non-secret boundary fixture only",
+                owner="glas-harness", evidence_ref="GLAS-WP-0005",
+            ),
+            "credential_route_refs": [],
+            "limits": profile.limits.model_copy(update={"timeout_seconds": 30}),
+        })
+        rein = BoundaryProof(source)
+        result = run_execution(ExecutionRequest(
+            harness_profile_ref=str(profile.ref), repo=str(source), title="boundary proof",
+            description="non-secret boundary probe", actor="agt",
+            project="glas-boundary-proof", request_id="glas-wp-0005-boundary-proof",
+            report_to_hub=False,
+        ), catalog=catalog, rein=rein, manager=manager)
+        unchanged = (
+            (source / "sentinel").read_text() == "unchanged"
+            and not (source / "proof.txt").exists()
+            and subprocess.check_output(["git", "rev-parse", "HEAD"], cwd=source) == source_head
+        )
+        state = manager.store.get(result.evidence.sandbox_id) if result.evidence.sandbox_id else None
+        removed = bool(rein.workspace) and not Path(rein.workspace).exists()
+        payload = {
+            "ok": result.ok, "sandbox_id": result.evidence.sandbox_id,
+            "proof": rein.proof, "source_unchanged": unchanged,
+            "task_removed_before_teardown": rein.task_removed,
+            "workspace_removed": removed, "state": state.state.value if state else None,
+            "error": result.tool_error, "production_rein_proven": False,
+        }
+        print(json.dumps(payload, indent=2))
+        return 0 if result.ok and unchanged and removed and rein.task_removed else 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())