sand-boxer/scripts/smoke-bwrap-owner-api.py
tegwick b6655d8859
All checks were successful
CI Smoke / host-smoke (push) Successful in 0s
CI Smoke / container-smoke (push) Successful in 2s
Add bounded stdin to owner execution
Assistant: codex
Assistant-Model: gpt-5.6-sol
Assistant-Session: 01a06def-6490-7033-8448-2eab2d12ed44
2026-09-04 22:30:11 +02:00

111 lines
4.3 KiB
Python

#!/usr/bin/env python3
"""Live authenticated HTTP proof for the ext.bwrap owner execution route."""
from __future__ import annotations
import json
import os
import tempfile
from pathlib import Path
from fastapi.testclient import TestClient
from sandboxer.api import app as api_module
from sandboxer.core.manager import SandboxManager
from sandboxer.lifecycle.store import SandboxStore
def main() -> int:
with tempfile.TemporaryDirectory(prefix="sandboxer-owner-api-proof-") as temp_value:
temp_dir = Path(temp_value)
source = temp_dir / "source"
source.mkdir()
(source / "copied.txt").write_text("copied\n")
outside = temp_dir / "host-source-sentinel"
outside.write_text("must-not-be-visible\n")
api_module._manager = SandboxManager(
store=SandboxStore(path=temp_dir / "sandboxes.json")
)
os.environ["SANDBOXER_EXEC_TOKEN"] = "non-secret-smoke-capability"
os.environ["SANDBOXER_NO_STATE_HUB"] = "1"
client = TestClient(api_module.app)
consumer = {
"actor": "agt",
"project": "sand-boxer-api-smoke",
"session_id": "api-proof",
"run_id": "api-proof-1",
}
sandbox_id = None
try:
created = client.post(
"/v1/sandboxes",
json={
"profile": "profile.bwrap-local",
"inputs": {"repo": str(source)},
"consumer": consumer,
"ttl": "5m",
},
)
created.raise_for_status()
sandbox_id = created.json()["sandbox_id"]
code = (
"import json, os, sys; from pathlib import Path; task=sys.stdin.read(); "
f"outside=Path({str(outside)!r}).exists(); "
"interfaces=[line.split(':',1)[0].strip() for line in "
"Path('/proc/net/dev').read_text().splitlines()[2:]]; "
"print(json.dumps({'cwd':str(Path.cwd()), 'copied':Path('copied.txt').is_file(), "
"'host_source_visible':outside, 'network_interfaces':interfaces, "
"'run_id':os.environ.get('SANDBOXER_RUN_ID'), "
"'stdin_bytes':len(task.encode())}))"
)
executed = client.post(
f"/v1/sandboxes/{sandbox_id}/exec",
headers={"Authorization": "Bearer non-secret-smoke-capability"},
json={
"command": ["/usr/bin/python3", "-c", code],
"consumer": consumer,
"stdin_text": '{"title":"non-secret API smoke"}',
"timeout_seconds": 30,
"max_output_bytes": 65_536,
},
)
if not executed.is_success:
raise RuntimeError(
f"owner exec HTTP {executed.status_code}: {executed.text}"
)
result = executed.json()
proof = json.loads(result["stdout"])
destroyed = client.delete(f"/v1/sandboxes/{sandbox_id}")
destroyed.raise_for_status()
payload = {
"sandbox_id": sandbox_id,
"http_exec_status": executed.status_code,
"exit_code": result["exit_code"],
"consumer": result["consumer"],
"network_default": result["network_default"],
"network_egress": result["network_egress"],
"proof": proof,
"teardown": {
"state": destroyed.json()["state"],
"workspace_removed": not Path(result["workspace_dir"]).exists(),
},
}
print(json.dumps(payload, indent=2))
passed = (
result["exit_code"] == 0
and result["consumer"] == consumer
and proof["copied"]
and not proof["host_source_visible"]
and proof["network_interfaces"] == ["lo"]
and proof["stdin_bytes"] == 32
and "stdin_text" not in result
and payload["teardown"]["workspace_removed"]
)
return 0 if passed else 1
finally:
if sandbox_id:
api_module._manager.destroy(sandbox_id)
if __name__ == "__main__":
raise SystemExit(main())