glas-harness/scripts/prove-owner-boundary.py

"""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())