85 lines
3.9 KiB
Python
85 lines
3.9 KiB
Python
|
#!/usr/bin/env python3
|
||
|
|
"""CUST-WP-0073-T02: exercise existing sand-boxer isolation without a model call.
|
||
|
|
|
||
|
|
Run with ~/glas-harness/.venv/bin/python. This does not switch the current
|
||
|
|
interactive agent into the sandbox or certify a complete agent runtime.
|
||
|
|
"""
|
||
|
|
import json
|
||
|
|
import tempfile
|
||
|
|
from datetime import datetime, timezone
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
from sandboxer.core.manager import SandboxManager
|
||
|
|
from sandboxer.lifecycle.store import SandboxStore
|
||
|
|
from sandboxer.models import Consumer, SandboxCreateRequest, SandboxExecRequest
|
||
|
|
from sandboxer.payments.credits import CreditsStore
|
||
|
|
from sandboxer.snapshots.store import SnapshotStore
|
||
|
|
|
||
|
|
|
||
|
|
PROBE = r'''
|
||
|
|
import json, os, socket
|
||
|
|
from pathlib import Path
|
||
|
|
paths = ['/home/worsch', '/home/tegwick', '/root', '/etc/rancher/k3s',
|
||
|
|
'/run/docker.sock', '/var/run/docker.sock', '/run/containerd',
|
||
|
|
'/run/user/1000', '/mnt/c']
|
||
|
|
checks = {'admin_paths_absent': all(not Path(p).exists() for p in paths),
|
||
|
|
'admin_environment_absent': not any(os.environ.get(k) for k in
|
||
|
|
['SSH_AUTH_SOCK', 'KUBECONFIG', 'BAO_TOKEN', 'VAULT_TOKEN']),
|
||
|
|
'only_loopback_interface': socket.if_nameindex() == [(1, 'lo')],
|
||
|
|
'approved_observation_readable': Path('observation.txt').read_text() == 'synthetic observation\n'}
|
||
|
|
Path('proposal.txt').write_text('synthetic privileged action proposal; not executed\n')
|
||
|
|
checks['proposal_preparation_works'] = Path('proposal.txt').is_file()
|
||
|
|
print(json.dumps(checks))
|
||
|
|
'''
|
||
|
|
|
||
|
|
|
||
|
|
def main():
|
||
|
|
report = {"captured_at": datetime.now(timezone.utc).isoformat(),
|
||
|
|
"workplan_task": "CUST-WP-0073-T02", "profile": "profile.bwrap-local",
|
||
|
|
"model_called": False, "interactive_agent_migrated": False}
|
||
|
|
with tempfile.TemporaryDirectory(prefix="cust-supervised-proof-") as temp:
|
||
|
|
root = Path(temp)
|
||
|
|
source = root / "source"
|
||
|
|
source.mkdir()
|
||
|
|
(source / "observation.txt").write_text("synthetic observation\n")
|
||
|
|
manager = SandboxManager(
|
||
|
|
store=SandboxStore(path=root / "sandboxes.json"),
|
||
|
|
credits=CreditsStore(path=root / "credits.json"),
|
||
|
|
snapshots=SnapshotStore(path=root / "snapshots.json"),
|
||
|
|
)
|
||
|
|
consumer = Consumer(actor="agt", project="the-custodian",
|
||
|
|
run_id="cust-wp-0073-supervised-proof")
|
||
|
|
status = manager.create(SandboxCreateRequest(
|
||
|
|
profile="profile.bwrap-local", inputs={"repo": str(source)},
|
||
|
|
consumer=consumer, ttl="5m",
|
||
|
|
))
|
||
|
|
report["sandbox_id"] = status.sandbox_id
|
||
|
|
try:
|
||
|
|
result = manager.execute(status.sandbox_id, SandboxExecRequest(
|
||
|
|
command=["/usr/bin/python3", "-c", PROBE], consumer=consumer,
|
||
|
|
timeout_seconds=15,
|
||
|
|
))
|
||
|
|
if result.exit_code or result.timed_out or result.output_truncated:
|
||
|
|
raise RuntimeError("sandbox probe failed; child output suppressed")
|
||
|
|
report["checks"] = json.loads(result.stdout)
|
||
|
|
wrong = Consumer(actor="agt", project="the-custodian", run_id="wrong-run")
|
||
|
|
try:
|
||
|
|
manager.execute(status.sandbox_id, SandboxExecRequest(
|
||
|
|
command=["/bin/true"], consumer=wrong, timeout_seconds=5))
|
||
|
|
except (ValueError, PermissionError):
|
||
|
|
report["checks"]["wrong_consumer_denied"] = True
|
||
|
|
else:
|
||
|
|
report["checks"]["wrong_consumer_denied"] = False
|
||
|
|
finally:
|
||
|
|
destroyed = manager.destroy(status.sandbox_id)
|
||
|
|
report["checks"]["workspace_removed"] = not Path(status.reachability.workspace_dir).exists()
|
||
|
|
report["checks"]["destroyed"] = destroyed.state.value == "destroyed"
|
||
|
|
report["checks"]["host_source_unchanged"] = not (source / "proposal.txt").exists()
|
||
|
|
report["passed"] = all(report["checks"].values())
|
||
|
|
print(json.dumps(report, indent=2))
|
||
|
|
return 0 if report["passed"] else 1
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
raise SystemExit(main())
|