#!/usr/bin/env python3 """Opt-in real-model/MicroVM gate on a disposable Linux API host. Credentials stay in Pipe AI. Only a connection and VM named for this run may be used. The connection is disconnected; the borrowed VM/network are preserved. Fixtures, transport-only tests and a successful assistant message do not pass. """ import argparse import datetime import json import os from pathlib import Path import platform import re import shutil import subprocess import sys import time from urllib.error import HTTPError from urllib.parse import urlsplit from urllib.request import HTTPRedirectHandler, ProxyHandler, Request, build_opener import uuid class QaFailure(Exception): """A failed gate; never include credentials or a raw HTTP error body.""" def require(condition, message): if not condition: raise QaFailure(message) class NoRedirect(HTTPRedirectHandler): def redirect_request(self, req, fp, code, msg, headers, newurl): return None class Api: def __init__(self, base): parsed = urlsplit(base) require(parsed.scheme == "http" and parsed.hostname == "127.0.0.1" and parsed.port and not parsed.username and not parsed.password and parsed.path in ("", "/") and not parsed.query and not parsed.fragment, "Run on the Linux API host with an explicit http://127.0.0.1:PORT URL") self.base = base.rstrip("/") self.http = build_opener(ProxyHandler({}), NoRedirect()) def request(self, method, path, body=None, allowed=(200,), timeout=360): request = Request(self.base + path, method=method, data=None if body is None else json.dumps(body).encode(), headers={"Content-Type": "application/json"}) try: response = self.http.open(request, timeout=timeout) except HTTPError as error: response = error with response: raw = response.read(2 * 1024 * 1024 + 1) require(len(raw) <= 2 * 1024 * 1024, "API response exceeds the QA limit") document = json.loads(raw) if response.code not in allowed: error = document.get("error") if isinstance(document, dict) else None code = error.get("code", "unknown") if isinstance(error, dict) else "unknown" raise QaFailure(f"{method} {path}: HTTP {response.code}, code={code}") return response.code, document def guest_command(run_id): return f"""set -eu test "$(id -u)" -ne 0 test "$(awk '/^NoNewPrivs:/ {{print $2}}' /proc/self/status)" = 1 test -z "${{OPENAI_API_KEY-}}${{ANTHROPIC_API_KEY-}}${{SSH_AUTH_SOCK-}}" nonce=$(od -An -N16 -tx1 /dev/urandom | tr -d ' \\n') printf '%s\\n' "$nonce" > guest-result.txt printf 'FIRECRAB414:{run_id}:stdout:%s\\n' "$nonce" printf 'FIRECRAB414:{run_id}:stderr\\n' >&2 exit 7""" def validate_result(task, project, vm, selection, run_id): require(task.get("projectId") == project and task.get("selection") == selection, "Task lost its project or effective model selection") require(task.get("state") == "result", f"Task did not finish: {task.get('state')}, {task.get('errorCode')}") runtime = task.get("runtime") or {} require(runtime.get("vmId") == vm and runtime.get("runtimeId"), "Missing admitted VM/runtime reference") require(uuid.UUID(task["attemptId"]).int and uuid.UUID(runtime["runtimeId"]).int, "Attempt/runtime identity cannot be nil") events = task.get("events", []) require(len(events) == 1 and events[0].get("name") == "run_command", "Expected exactly one real guest command, without other tool calls") event = events[0] require(json.loads(event["input"]).get("command") == guest_command(run_id), "Model changed the authorized QA command") match = re.fullmatch(rf"FIRECRAB414:{re.escape(run_id)}:stdout:([0-9a-f]{{32}})\n", event.get("output", "")) require(match is not None, "Missing unpredictable guest-generated stdout nonce") nonce = match.group(1) require(event.get("stderr") == f"FIRECRAB414:{run_id}:stderr\n" and event.get("exitCode") == 7, "Separate stderr or actual nonzero guest exit status was lost") require(nonce in task.get("result", ""), "The model did not receive and return the guest result") changes = task.get("changes", []) require(len(changes) == 1 and changes[0].get("path") == "guest-result.txt" and changes[0].get("before") == "not executed\n" and changes[0].get("after") == nonce + "\n" and not changes[0].get("applied"), "The guest result was not retained as an unapplied source change") return nonce def wait_task(api, key, report, output, timeout=600): deadline = time.monotonic() + timeout while time.monotonic() < deadline: _, task = api.request("GET", f"/api/ai-sandbox/tasks/{key}") report["tasks"][key] = task save_report(output, report) if task["state"] not in ("waiting", "accepted", "in_progress"): return task time.sleep(1) raise QaFailure(f"Task deadline exceeded; inspect recorded task {key} before retrying") def save_report(output, report): temporary = output.with_name(f".{output.name}.{uuid.uuid4()}.tmp") fd = os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) try: with os.fdopen(fd, "w") as stream: json.dump(report, stream, indent=2) stream.write("\n") stream.flush() os.fsync(stream.fileno()) os.replace(temporary, output) finally: temporary.unlink(missing_ok=True) def owned_inputs(api, run_id, connection, vm): label = f"sandbox-foundation-{run_id}" _, connections = api.request("GET", "/api/pipe/ai/connections") account = next((c for c in connections if c["id"] == connection), None) require(account is not None and account.get("name") == label, "Connection must be created for this run; borrowed credentials cannot be disconnected") _, machine = api.request("GET", f"/api/vms/{vm}") require(machine.get("id") == vm and machine.get("name") == label and machine.get("purpose", "instance") == "instance" and machine.get("state") == "running", "Select this run's running disposable user MicroVM") _, projects = api.request("GET", "/api/ai-sandbox/projects") require(not any(p["spec"].get("vmId") == vm or (p["spec"].get("selection") or {}).get("connectionId") == connection for p in projects), "VM or connection is already referenced by another project") return account, machine, {c["id"] for c in connections} def source_git(source, *args): command = ["/usr/bin/git", "-c", "core.hooksPath=/dev/null", "-c", "core.fsmonitor=false", "-c", "user.name=Firecrab QA", "-c", "user.email=qa@localhost", "-C", str(source), *args] return subprocess.run(command, check=True, capture_output=True, text=True, env={"PATH": "/usr/bin:/bin", "GIT_CONFIG_GLOBAL": "/dev/null", "GIT_CONFIG_NOSYSTEM": "1"}).stdout.strip() def run_gate(args, api): run_id, connection, vm = map(str, (args.run_id, args.connection, args.vm)) require(args.default_model != args.override_model, "Use two different account-visible models") account, machine, original_connections = owned_inputs(api, run_id, connection, vm) results = args.results_dir.resolve() results.mkdir(mode=0o700, parents=True, exist_ok=False) output = results / "summary.json" source = results / "source" source.mkdir(mode=0o700) (source / "guest-result.txt").write_text("not executed\n") source_git(source, "init", "-q") source_git(source, "add", "guest-result.txt") source_git(source, "commit", "-q", "--no-gpg-sign", "-m", "Sandbox foundation QA baseline") project = str(uuid.uuid4()) report = {"runId": run_id, "status": "RUNNING", "startedAt": datetime.datetime.now(datetime.timezone.utc).isoformat(), "platform": platform.platform(), "api": args.api, "projectId": project, "sourcePath": str(source), "sourceCommit": source_git(source, "rev-parse", "HEAD"), "connection": {k: account[k] for k in ("id", "provider", "authentication")}, "vm": {k: machine[k] for k in ("id", "name", "state", "microNetworkId", "template", "templateVersion")}, "checks": {}, "tasks": {}, "retainedOnFailure": True} save_report(output, report) try: # Validate the API's view of the source before any paid model probe. # A systemd service can have a different /tmp from this process. spec = {"name": f"sandbox-foundation-{run_id}", "sourcePath": str(source), "vmId": vm, "selection": None, "instructions": "Use only the exact command authorized by this QA task."} api.request("PUT", f"/api/ai-sandbox/projects/{project}", spec) report["checks"]["sourceVisibleToApi"] = "PASS" save_report(output, report) _, catalogue = api.request("GET", f"/api/ai-sandbox/connections/{connection}/models?discover=true") available = {m["id"] for m in catalogue["models"]} require({args.default_model, args.override_model} <= available, "Requested models are not account-visible") for model in (args.default_model, args.override_model): _, proof = api.request("POST", f"/api/ai-sandbox/connections/{connection}/verify", {"model": model, "tools": True}) require(proof.get("verified") and proof.get("toolsVerified"), "Real model native-tool verification failed") report["checks"]["nativeModelPermissions"] = "PASS" default = {"connectionId": connection, "model": args.default_model} spec["selection"] = default api.request("PUT", f"/api/ai-sandbox/projects/{project}", spec) nonces = [] attempts = set() runtime = None for label, model in (("projectDefault", args.default_model), ("taskOverride", args.override_model)): key = str(uuid.uuid4()) selected = {"connectionId": connection, "model": model} prompt = ("Call run_command exactly once with timeout_seconds=30 and this exact command. " "Do not use any other tool or retry its deliberate exit 7. After receiving the result, " "reply with the actual stdout nonce, stderr, and exit status. The nonce must come from the guest.\n```sh\n" + guest_command(run_id) + "\n```") request = {"kind": "code", "input": prompt} if label == "taskOverride": request["selection"] = selected report["tasks"][key] = {"id": key, "request": request} save_report(output, report) _, admitted = api.request("PUT", f"/api/ai-sandbox/projects/{project}/tasks/{key}", request, (200, 202)) task = wait_task(api, key, report, output) require(task["attemptId"] == admitted["attemptId"] and task["runtime"] == admitted["runtime"], "Admission references changed while executing") nonces.append(validate_result(task, project, vm, selected, run_id)) require(task["attemptId"] not in attempts, "Separate tasks reused one attempt identity") attempts.add(task["attemptId"]) require(runtime is None or runtime == task["runtime"], "MicroVM rebooted during the gate") runtime = task["runtime"] _, reloaded = api.request("GET", f"/api/ai-sandbox/tasks/{key}") require(reloaded == task, "Reload lost task, attempt, runtime or tool results") _, current_project = api.request("GET", f"/api/ai-sandbox/projects/{project}") require(current_project["spec"]["selection"] == default, "Task override mutated the project default") require((source / "guest-result.txt").read_text() == "not executed\n", "Execution changed the host source before apply") require(not source_git(source, "status", "--porcelain"), "Execution changed the source Git worktree") report["checks"][label] = "PASS" save_report(output, report) require(len(set(nonces)) == 2, "Separate attempts did not produce separate guest entropy") denied_model = f"unverified-{run_id}" status, _ = api.request("POST", f"/api/ai-sandbox/connections/{connection}/verify", {"model": denied_model, "tools": True}, (400, 403, 404, 409, 422, 502)) _, proofs = api.request("GET", f"/api/ai-sandbox/connections/{connection}/models") require(not any(m["id"] == denied_model and m.get("verified") for m in proofs["models"]), "Denied model gained an inference permission proof") report["checks"]["deniedModelVerification"] = {"status": "PASS", "httpStatus": status} for label, selection in (("deniedModel", {"connectionId": connection, "model": denied_model}), ("afterDisconnect", default)): if label == "afterDisconnect": # Recheck ownership immediately before the only credential mutation. _, connections = api.request("GET", "/api/pipe/ai/connections") require(any(c["id"] == connection and c["name"] == spec["name"] for c in connections), "Connection ownership changed") _, receipt = api.request("DELETE", f"/api/pipe/ai/connections/{connection}") require(receipt.get("deleted") is True, "QA connection deletion was not confirmed") report["connectionDisconnect"] = receipt api.request("GET", f"/api/ai-sandbox/connections/{connection}/models?discover=true", allowed=(404, 409)) api.request("POST", f"/api/ai-sandbox/connections/{connection}/verify", {"model": args.default_model, "tools": True}, (404, 409)) key = str(uuid.uuid4()) status, _ = api.request("PUT", f"/api/ai-sandbox/projects/{project}/tasks/{key}", {"kind": "code", "input": "This request must never reach a provider or guest.", "selection": selection}, (400, 403, 409)) api.request("GET", f"/api/ai-sandbox/tasks/{key}", allowed=(404,)) report["checks"][label] = {"status": "PASS", "httpStatus": status, "taskId": key, "admitted": False} _, remaining = api.request("GET", "/api/pipe/ai/connections") require({c["id"] for c in remaining} == original_connections - {connection}, "Disconnect affected another account") # Export results before deleting only the project created above. report["checks"]["resultExport"] = "PASS" save_report(output, report) _, owned = api.request("GET", f"/api/ai-sandbox/projects/{project}") require(all(owned["spec"].get(k) == spec[k] for k in ("name", "sourcePath", "vmId")), "Project ownership changed; preserve it") api.request("DELETE", f"/api/ai-sandbox/projects/{project}") _, preserved = api.request("GET", f"/api/vms/{vm}") require(all(preserved.get(k) == machine[k] for k in ("id", "name", "state", "microNetworkId")), "Borrowed VM/network changed") report["checks"]["ownedCleanup"] = "PASS" report.update(status="PASS", retainedOnFailure=False) shutil.rmtree(source) except (Exception, KeyboardInterrupt) as error: report.update(status="FAILED", failure=f"{type(error).__name__}: {str(error)[:300]}") # Do not delete an uncertain attempt or its source. The report contains # stable IDs for inspect/cancel/reconcile, without a blind retry. raise finally: report["finishedAt"] = datetime.datetime.now(datetime.timezone.utc).isoformat() save_report(output, report) print(f"{report['status']}: {output}", flush=True) def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--api", required=True) parser.add_argument("--run-id", required=True, type=uuid.UUID) parser.add_argument("--connection", required=True, type=uuid.UUID) parser.add_argument("--vm", required=True, type=uuid.UUID) parser.add_argument("--default-model", required=True) parser.add_argument("--override-model", required=True) parser.add_argument("--results-dir", required=True, type=Path) parser.add_argument("--live", action="store_true", help="Authorize real model requests and plan usage") parser.add_argument("--disconnect-owned-connection", action="store_true", help="Authorize disconnecting only this run's named QA connection") args = parser.parse_args() if sys.platform != "linux" or not args.live or not args.disconnect_owned_connection: parser.error("Requires Linux, --live, and --disconnect-owned-connection; use a separate QA account connection") try: run_gate(args, Api(args.api)) except (Exception, KeyboardInterrupt) as error: print(f"FAILED: {type(error).__name__}: {str(error)[:300]}", file=sys.stderr) return 1 return 0 if __name__ == "__main__": sys.exit(main())