"""Installed owner + CLI + bwrap proof with a fake provider and an empty fake queue. Run using the candidate's python -I -B. Never acquires a real credential. """ from __future__ import annotations import argparse from datetime import UTC, datetime, timedelta import hashlib import http.server import importlib import json import os from pathlib import Path import subprocess import sys import tempfile import threading import yaml from glas_harness.contract import ExecutionLimits, OperationalReadiness from glas_harness.profiles import ProfileCatalog from llm_connect.messages_gate import MessagesPolicy from llm_connect.models import RunConfig from rein_aharness.adapter import AgenticClaudeCodeAdapter from rein_aharness.execution_cancel import ExecutionCancel from rein_aharness.messages_owner import MessagesOwner from rein_aharness.ops_run_client import OpsRun, OpsRunConfig from rein_aharness.repository_grant import RepositoryGrant from rein_aharness.request_admission import RequestLedger from rein_aharness.spend_admission import SpendLedger, SpendPolicy, digest from sandboxer.extensions.runtime import runtime_digest from sandboxer.models import Consumer, SandboxCreateRequest, SandboxExecRequest from sandboxer.profiles.loader import profiles_dir from sandboxer.extensions.registry import extensions_dir BETAS = ( "claude-code-20250219", "interleaved-thinking-2025-05-14", "thinking-token-count-2026-05-13", "context-management-2025-06-27", "prompt-caching-scope-2026-01-05", "effort-2025-11-24", ) DUMMY_KEY = "synthetic-provider-key-no-live-authority" def response_bytes(model, tool=False): events = [ {"type": "message_start", "message": {"id": "msg_fixture", "type": "message", "role": "assistant", "model": model, "content": [], "stop_reason": None, "stop_sequence": None, "usage": {"input_tokens": 100, "output_tokens": 0, "cache_creation_input_tokens": 0, "cache_read_input_tokens": 0}}}, {"type": "content_block_start", "index": 0, "content_block": {"type": "text", "text": ""}}, {"type": "content_block_delta", "index": 0, "delta": {"type": "text_delta", "text": "fixture complete"}}, {"type": "content_block_stop", "index": 0}, {"type": "message_delta", "delta": {"stop_reason": "end_turn", "stop_sequence": None}, "usage": {"input_tokens": 100, "output_tokens": 10, "cache_creation_input_tokens": 0, "cache_read_input_tokens": 0}}, {"type": "message_stop"}, ] if tool: events[1]["content_block"] = {"type": "tool_use", "id": "fixture_git_status", "name": "Bash", "input": {}} events[2]["delta"] = {"type": "input_json_delta", "partial_json": json.dumps({"command": "git status --short", "description": "Synthetic permitted tool proof"})} events[4]["delta"]["stop_reason"] = "tool_use" return "".join("event: " + x["type"] + "\ndata: " + json.dumps(x) + "\n\n" for x in events).encode() class FixtureServer(http.server.ThreadingHTTPServer): provider_calls = 0 queue_claims = 0 provider_key_correct = False expected_model = "" request_shapes = None tool_cycle = False class Handler(http.server.BaseHTTPRequestHandler): def log_message(self, *args): pass def reply(self, body, content_type="application/json"): self.send_response(200) self.send_header("Content-Type", content_type) self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def do_GET(self): if self.path.startswith("/ops-runs"): self.reply(b'{"items":[]}') else: self.send_error(404) def do_POST(self): raw = self.rfile.read(int(self.headers.get("Content-Length", "0"))) if self.path == "/ops-runs/claim": self.server.queue_claims += 1 self.reply(b'{"items":[]}') elif self.path == "/v1/messages": self.server.provider_calls += 1 self.server.provider_key_correct = self.headers.get("x-api-key") == DUMMY_KEY request = json.loads(raw) thinking = request.get("thinking", {}) shape = {"model": request.get("model"), "max_tokens": request.get("max_tokens"), "thinking_type": thinking.get("type"), "manual_thinking_budget": "budget_tokens" in thinking, "sampling_parameters": {k: request[k] for k in ("temperature", "top_p", "top_k") if k in request}, "betas": self.headers.get("anthropic-beta", "").split(","), "tool_results": [{"id": b.get("tool_use_id"), "is_error": b.get("is_error", False)} for m in request.get("messages", []) if isinstance(m.get("content"), list) for b in m["content"] if b.get("type") == "tool_result"]} self.server.request_shapes.append(shape) if shape["model"] != self.server.expected_model: self.send_error(400) return # Sonnet 5 rejects manual thinking and non-default sampling settings. # Match that documented refusal rather than hiding it behind a 200. if shape["model"] == "claude-sonnet-5" and ( shape["thinking_type"] == "enabled" or shape["manual_thinking_budget"] or any(value != {"temperature": 1, "top_p": 1, "top_k": 0}[key] for key, value in shape["sampling_parameters"].items()) ): self.send_error(400) return self.reply(response_bytes(self.server.expected_model, self.server.tool_cycle and len(self.server.request_shapes) == 1), "text/event-stream") else: self.send_error(404) def case(root, runtime, checksum, server, cap, args, bootstrap=False): root.mkdir() private = root / "private"; private.mkdir(mode=0o700) source = root / "source"; source.mkdir() subprocess.run(["git", "init", "-q", str(source)], check=True) (source / "README.md").write_text("Disposable fixture. Do not edit files.\n") profiles = root / "profiles"; profiles.mkdir() catalog = ProfileCatalog() profile, descriptor = catalog.resolve(args.profile_ref) assert str(profile.ref) == args.profile_ref, "resolved profile mismatch" assert profile.model.model == args.expected_model, "resolved model mismatch" profile = profile.model_copy(update={ "limits": ExecutionLimits(max_budget_usd=cap, max_turns=4), "operational_readiness": OperationalReadiness(status="ready", reason="disposable fixture only", owner="tests", evidence_ref="test:installed-owner"), }) (profiles / "fixture.yaml").write_text(yaml.safe_dump(profile.model_dump(mode="json"))) grant = RepositoryGrant("1", ("result.txt",), 1, 1, False) policy = SpendPolicy( version="1", envelope_id="fixture", authority_ref="fixture:no-live-authority", valid_from="2026-09-01T00:00:00Z", expires_at="2099-01-01T00:00:00Z", timezone="Europe/Berlin", worker_id="fixture-worker", activity_definition_id="fixture-definition", target_repo=str(source), project="fixture-factory", profile_ref=str(profile.ref), profile_sha256=digest(profile.model_dump(mode="json")), descriptor_sha256=digest(descriptor.model_dump(mode="json")), repository_grant_id=grant.grant_id, max_budget_usd=str(cap), max_liability_usd=("10" if args.tool_cycle else "5") if cap > 0.01 else "0.01", max_turns=4, eur_per_usd="1", per_run_eur="10" if args.tool_cycle else "5", daily_eur="20", total_eur="30", ) parent = SpendLedger(private / "spend.sqlite3", policy); parent.initialize() meter = RequestLedger(parent); meter.initialize() policy_path = private / "spend-policy.json" policy_path.write_text(json.dumps(policy.__dict__)); policy_path.chmod(0o600) admission_requests = [] class RecordingPolicy(MessagesPolicy): def validate(self, data, betas): shape = {"model": data.get("model"), "max_tokens": data.get("max_tokens"), "thinking_type": data.get("thinking", {}).get("type"), "top_level_fields": sorted(data), "betas": betas.split(","), "message_roles": [m.get("role") for m in data.get("messages", [])], "output_config_keys": sorted(data.get("output_config", {})), "output_effort": data.get("output_config", {}).get("effort")} admission_requests.append(shape) try: return super().validate(data, betas) except Exception as exc: shape["refusal"] = str(exc) raise messages = RecordingPolicy("fixture:not-live-prices", profile.model.model, args.context_tokens, args.max_output_tokens, args.input_rate, args.output_rate, allowed_betas=BETAS + tuple(args.extra_beta)) server.expected_model = profile.model.model server.request_shapes = [] server.tool_cycle = args.tool_cycle owner_path = private / "owner.json" owner_path.write_text(json.dumps({"version": "1", "authority_ref": policy.authority_ref, "spend_policy_sha256": policy.sha256, "messages_policy": messages.__dict__, "runtime": {"path": str(runtime), "sha256": checksum}})); owner_path.chmod(0o600) bootstrap_result = None if bootstrap: env = {"PATH": os.defpath, "LANG": "C.UTF-8", "PYTHONDONTWRITEBYTECODE": "1", "ACTIVITY_CORE_URL": f"http://127.0.0.1:{server.server_port}", "AGENT_HARNESS_WORKER_ID": policy.worker_id, "AGENT_HARNESS_EXECUTION_PROJECT": policy.project, "AGENT_HARNESS_SPEND_POLICY": str(policy_path), "AGENT_HARNESS_SPEND_LEDGER": str(parent.path), "REIN_AHARNESS_STATE_DIR": str(private / "worker-state"), "GLAS_PROFILE_DIR": str(profiles), "XDG_DATA_HOME": str(root / "data"), "ANTHROPIC_API_KEY": DUMMY_KEY} child = subprocess.run([str(runtime / "bin/python3"), "-I", "-B", "-m", "rein_aharness.cli", "metered-once", "--owner-config", str(owner_path), "--no-hub"], env=env, capture_output=True, text=True, timeout=30) assert child.returncode == 0, "installed bootstrap refused disposable configuration" bootstrap_result = json.loads(child.stdout) assert bootstrap_result["empty"] and server.queue_claims == 1 assert server.provider_calls == 0 and not parent.status()["reservations"] config = OpsRunConfig(worker_id=policy.worker_id, execution_project=policy.project) item = OpsRun(id="fixture-run", activity_definition_id=policy.activity_definition_id, idempotency_key="fixture-key", target_repo=str(source), title="fixture", description="", state="claimed", claim_owner=policy.worker_id, attempt=1, repository_grant=grant, harness_profile_ref=str(profile.ref), lease_until=(datetime.now(UTC) + timedelta(seconds=90)).isoformat()) parent.reserve(item) owner = MessagesOwner(messages, DUMMY_KEY, f"http://127.0.0.1:{server.server_port}", True, runtime_path=runtime, runtime_sha256=checksum) before = server.provider_calls consumer = Consumer(actor="agt", project=policy.project, run_id=item.id) with owner.activate(item, config, parent, profile, ExecutionCancel()) as manager: status = manager.create(SandboxCreateRequest(profile="profile.bwrap-local", consumer=consumer, inputs={"repo": str(source)})) try: assert status.state.value == "ready", "sandbox not ready" probe = manager.execute(status.sandbox_id, SandboxExecRequest( consumer=consumer, timeout_seconds=30, command=["python3", "-c", """import os,json,subprocess,sys from pathlib import Path runtime=Path('/opt/sandboxer/runtime') try: (runtime/'must-not-write').write_text('no') except OSError: readonly=True else: readonly=False print(json.dumps({'runtime_readonly':readonly,'python_prefix':sys.prefix, 'cli_version':subprocess.check_output(['claude','--version'],text=True).strip(), 'private_absent':not Path(sys.argv[1]).exists(),'source_absent':not Path(sys.argv[2]).exists(), 'proxy_absent':not any('proxy' in k.lower() for k in os.environ), 'interfaces':[x.split(':')[0].strip() for x in Path('/proc/net/dev').read_text().splitlines()[2:]]})) """, str(private), str(source)])) assert probe.exit_code == 0, "installed runtime probe failed" facts = json.loads(probe.stdout) assert facts["runtime_readonly"] and facts["private_absent"] and facts["source_absent"] assert facts["proxy_absent"] and facts["interfaces"] == ["lo"] assert facts["cli_version"] == "2.1.266 (Claude Code)" adapter = AgenticClaudeCodeAdapter(workdir=Path(status.inputs["workspace_dir"]), cli_path="/opt/sandboxer/runtime/bin/claude", model=profile.model.model) argv = adapter._build_command(RunConfig(model_params={"max_budget_usd": cap, "max_turns": 4})) executed = manager.execute(status.sandbox_id, SandboxExecRequest( consumer=consumer, command=argv, stdin_text="Reply fixture complete. Do not edit files.", timeout_seconds=45)) terminal = json.loads(executed.stdout) if cap <= 0.01: assert executed.exit_code != 0 and terminal["is_error"] assert server.provider_calls == before and not meter.status() else: assert executed.exit_code == 0 and not terminal.get("is_error"), json.dumps({"exit_code": executed.exit_code, "subtype": terminal.get("subtype"), "request_shapes": server.request_shapes, "admission_requests": admission_requests}) assert server.provider_calls == before + (2 if args.tool_cycle else 1) and server.provider_key_correct assert len(meter.status()) == (2 if args.tool_cycle else 1) assert all(row["state"] == "charged" for row in meter.status()) if args.tool_cycle: assert {"id": "fixture_git_status", "is_error": False} in server.request_shapes[-1]["tool_results"] token = manager._messages_route.token finally: manager.destroy(status.sandbox_id) assert not Path(status.inputs["workspace_dir"]).exists() with parent._db() as db: assert db.execute("SELECT revoked FROM request_routes").fetchone()[0] == 1 output = {"native_threshold_usd": cap, "probe": facts, "provider_requests": server.provider_calls-before, "cli_exit_code": executed.exit_code, "cli_is_error": terminal.get("is_error"), "cli_estimated_usd": terminal.get("total_cost_usd"), "request_reservations": meter.status(), "route_revoked": True, "workspace_removed": True, "bootstrap": bootstrap_result, "profile_ref": args.profile_ref, "model": profile.model.model, "request_shapes": list(server.request_shapes), "policy_context_tokens": messages.context_tokens, "admission_requests": admission_requests} assert token not in json.dumps(output) and DUMMY_KEY not in json.dumps(output) return output def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--runtime", required=True, type=Path) parser.add_argument("--sha256", required=True) parser.add_argument("--profile-ref", required=True, help="Exact bundled profile under test; never infer from the artifact name") parser.add_argument("--expected-model", required=True) parser.add_argument("--tool-cycle", action="store_true", help="Two fake-provider calls and an actual permitted git-status tool, synthetic EUR 10 envelope") parser.add_argument("--context-tokens", type=int, default=200000) parser.add_argument("--max-output-tokens", type=int, default=32000) parser.add_argument("--extra-beta", action="append", default=[]) parser.add_argument("--input-rate", type=int, default=3) parser.add_argument("--output-rate", type=int, default=15) args = parser.parse_args() runtime = args.runtime.resolve() assert sys.flags.isolated and sys.dont_write_bytecode, "run candidate python with -I -B" assert runtime_digest(runtime) == args.sha256 imports = {} for package in ("rein_aharness", "llm_connect", "glas_harness", "sandboxer"): path = Path(importlib.import_module(package).__file__).resolve() assert path.is_relative_to(runtime), "editable source fallback" imports[package] = str(path.relative_to(runtime)) assert profiles_dir().is_relative_to(runtime) and extensions_dir().is_relative_to(runtime) os.environ["SANDBOXER_NO_STATE_HUB"] = "1" with tempfile.TemporaryDirectory(prefix="installed-metered-proof-") as directory: root = Path(directory) os.environ["XDG_DATA_HOME"] = str(root / "data") server = FixtureServer(("127.0.0.1", 0), Handler) threading.Thread(target=server.serve_forever, daemon=True).start() try: positive = case(root / "positive", runtime, args.sha256, server, 1.0, args, bootstrap=True) refused = case(root / "refused", runtime, args.sha256, server, 0.01, args) finally: server.shutdown(); server.server_close() assert runtime_digest(runtime) == args.sha256, "proof mutated the candidate artifact" print(json.dumps({"scope": "standalone installed packages and real CLI/bwrap; synthetic key/provider/empty queue", "ok": True, "runtime_sha256": args.sha256, "proof_script_sha256": hashlib.sha256(Path(__file__).read_bytes()).hexdigest(), "claude_sha256": hashlib.sha256((runtime/'bin/claude').read_bytes()).hexdigest(), "imports": imports, "packaged_definitions": True, "positive": positive, "refused": refused, "artifact_unchanged": True, "live_factory_attempts": 0}, indent=2)) if __name__ == "__main__": main()