scripts/test-ci-qa-ai-sandbox-recovery.py
Rama main · réplica de lectura del repositorio original.
#!/usr/bin/env python3
"""Offline recovery-gate guard tests. These do not establish live acceptance."""
import importlib.util
import contextlib
import io
import json
from pathlib import Path
import stat
import tempfile
from types import SimpleNamespace
import unittest
from unittest.mock import Mock, patch
import uuid
spec = importlib.util.spec_from_file_location("recovery", Path(__file__).with_name("ci-qa-ai-sandbox-recovery.py"))
qa = importlib.util.module_from_spec(spec)
spec.loader.exec_module(qa)
RUN, CONNECTION, VM, NETWORK, PROJECT = [str(uuid.uuid4()) for _ in range(5)]
NONCE = "a" * 32
LABEL = f"sandbox-recovery-{RUN}"
class Api:
base = "http://127.0.0.1:5523"
def __init__(self):
self.calls = []
self.accounts = [{"id": CONNECTION, "name": LABEL}]
self.vms = [{"id": VM, "name": LABEL, "state": "running", "purpose": "instance", "microNetworkId": NETWORK, "egressPolicy": "isolated"}]
self.network = {"id": NETWORK, "name": LABEL, "nat": {"enabled": False}, "subnet": {"ipv6Cidr": None}}
self.projects = []
def request(self, method, path, body=None, allowed=(200,), timeout=360):
self.calls.append((method, path, body))
if method == "GET":
if path.startswith("/api/ai-sandbox/tasks/"):
return 200, {"id": path.rsplit("/", 1)[1]}
return 200, {"/api/pipe/ai/connections": self.accounts, "/api/vms": self.vms,
f"/api/micro-networks/{NETWORK}": self.network, "/api/ai-sandbox/projects": self.projects}[path]
if method == "PUT":
return 200, body
raise qa.qa.QaFailure("Deliberate validation stop")
def result():
script = qa.command(RUN, "cancel", 90)
return {"projectId": PROJECT, "selection": {"connectionId": CONNECTION, "model": "verified"},
"runtime": {"vmId": VM}, "state": "cancelled",
"events": [{"name": "run_command", "input": json.dumps({"command": script, "timeout_seconds": 120}),
"output": f"RECOVERY414:{RUN}:cancel:stdout:{NONCE}\n",
"stderr": f"RECOVERY414:{RUN}:cancel:stderr\n", "exitCode": None}],
"changes": [{"path": "executions.txt", "after": "once\n", "applied": False},
{"path": "nonce", "after": NONCE, "applied": False}]}
class RecoveryTests(unittest.TestCase):
def test_ownership_marker_requires_private_root_file_and_controlled_directory(self):
path = Mock()
path.parent.lstat.return_value = SimpleNamespace(st_mode=stat.S_IFDIR | 0o755, st_uid=0)
path.lstat.return_value = SimpleNamespace(st_mode=stat.S_IFREG | 0o600, st_uid=0)
path.read_text.return_value = RUN + "\n"
qa.owner_marker(path, RUN)
for mode, uid in [(stat.S_IFLNK | 0o600, 0), (stat.S_IFREG | 0o644, 0), (stat.S_IFREG | 0o600, 501)]:
path.lstat.return_value = SimpleNamespace(st_mode=mode, st_uid=uid)
with self.assertRaises(qa.qa.QaFailure):
qa.owner_marker(path, RUN)
path.lstat.return_value = SimpleNamespace(st_mode=stat.S_IFREG | 0o600, st_uid=0)
path.parent.lstat.return_value = SimpleNamespace(st_mode=stat.S_IFDIR | 0o777, st_uid=0)
with self.assertRaises(qa.qa.QaFailure):
qa.owner_marker(path, RUN)
path.parent.lstat.return_value = SimpleNamespace(st_mode=stat.S_IFDIR | 0o755, st_uid=0)
path.read_text.return_value = str(uuid.uuid4())
with self.assertRaises(qa.qa.QaFailure):
qa.owner_marker(path, RUN)
def test_shared_or_foreign_resources_are_rejected_before_any_write_or_controller_access(self):
for changed in ("account", "extraAccount", "vm", "extraVm", "stoppedVm", "internalVm", "network", "project", "egress", "internet", "ipv6"):
api = Api()
if changed == "account": api.accounts[0]["name"] = "borrowed"
if changed == "extraAccount": api.accounts.append({"id": str(uuid.uuid4())})
if changed == "vm": api.vms[0]["name"] = "borrowed"
if changed == "extraVm": api.vms.append({"id": str(uuid.uuid4())})
if changed == "stoppedVm": api.vms[0]["state"] = "stopped"
if changed == "internalVm": api.vms[0]["purpose"] = "image"
if changed == "network": api.network["name"] = "borrowed"
if changed == "project": api.projects.append({"id": PROJECT})
if changed == "egress": api.vms[0]["egressPolicy"] = "internet"
if changed == "internet": api.network["nat"]["enabled"] = True
if changed == "ipv6": api.network["subnet"]["ipv6Cidr"] = "fd00::/64"
with self.subTest(changed=changed), patch.object(qa, "control") as control:
with self.assertRaises(qa.qa.QaFailure):
qa.owned(api, RUN, CONNECTION, VM)
control.assert_not_called()
self.assertTrue(all(method == "GET" for method, _, _ in api.calls))
def test_owns_exactly_one_api_vm_and_network(self):
label, vm, network = qa.owned(Api(), RUN, CONNECTION, VM)
self.assertEqual((label, vm["id"], network["id"]), (LABEL, VM, NETWORK))
def test_controller_pid_must_own_the_selected_loopback_listener(self):
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
process = root / "42"
(process / "fd").mkdir(parents=True)
executable = root / "firecrab-api"
executable.touch()
(process / "exe").symlink_to(executable)
(process / "fd/3").symlink_to("socket:[123]")
(root / "net").mkdir()
tcp = root / "net/tcp"
tcp.write_text("header\n0: 0100007F:1593 00000000:0000 0A 0 0 0 0 0 123\n")
def proc(path): return root / str(path).removeprefix("/proc/")
with patch.object(qa, "control", return_value="42"), patch.object(qa, "Path", side_effect=proc):
self.assertEqual(qa.controller_pid(Api(), "firecrab-api.service"), "42")
tcp.write_text("header\n0: 0100007F:1593 00000000:0000 0A 0 0 0 0 0 456\n")
with self.assertRaises(qa.qa.QaFailure):
qa.controller_pid(Api(), "firecrab-api.service")
tcp.write_text("header\n0: 00000000:1593 00000000:0000 0A 0 0 0 0 0 123\n")
with self.assertRaises(qa.qa.QaFailure):
qa.controller_pid(Api(), "firecrab-api.service")
def test_zero_or_non_numeric_controller_pid_cannot_trigger_a_fault(self):
for value in ("0", "1", "invalid", ""):
with patch.object(qa, "control", return_value=value), self.assertRaises(qa.qa.QaFailure):
qa.controller_pid(Api(), "firecrab-api.service")
def test_partial_reboot_output_keeps_unknown_exit_without_inventing_success(self):
task = result()
task["events"][0]["stderr"] += "Killed\n"
self.assertEqual(qa.validate(task, PROJECT, VM, task["selection"], qa.command(RUN, "cancel", 90),
RUN, "cancel", "cancelled", NONCE), NONCE)
self.assertIsNone(task["events"][0]["exitCode"])
def test_replay_forged_output_lost_stderr_and_wrong_references_fail_validation(self):
for changed in ("replay", "stdout", "stderr", "project", "vm", "selection", "command", "applied"):
task = result()
selection = dict(task["selection"])
if changed == "replay": task["changes"][0]["after"] = "once\nonce\n"
if changed == "stdout": task["events"][0]["output"] = "hallucinated"
if changed == "stderr": task["events"][0]["stderr"] = ""
if changed == "project": task["projectId"] = str(uuid.uuid4())
if changed == "vm": task["runtime"]["vmId"] = str(uuid.uuid4())
if changed == "selection": task["selection"]["model"] = "other"
if changed == "command": task["events"][0]["input"] = json.dumps({"command": "other"})
if changed == "applied": task["changes"][0]["applied"] = True
with self.subTest(changed=changed), self.assertRaises(qa.qa.QaFailure):
qa.validate(task, PROJECT, VM, selection, qa.command(RUN, "cancel", 90), RUN, "cancel", "cancelled", NONCE)
def test_disconnected_admission_sends_once_and_does_not_read_or_resubmit(self):
with patch.object(qa.socket, "create_connection") as connect:
client = connect.return_value.__enter__.return_value
qa.disconnected_put(Api(), "/api/owned-task", {"kind": "code"}, PROJECT)
client.sendall.assert_called_once()
self.assertIn(b"PUT /api/owned-task HTTP/1.1\r\n", client.sendall.call_args.args[0])
client.recv.assert_not_called()
def test_batch_retention_preserves_only_the_same_owned_account(self):
api = Api()
self.assertEqual(qa.finish_connection(api, CONNECTION, LABEL, True), {"retainedForFurtherValidation": True})
self.assertTrue(all(method == "GET" for method, _, _ in api.calls))
api.accounts[0]["name"] = "another account"
with self.assertRaises(qa.qa.QaFailure):
qa.finish_connection(api, CONNECTION, LABEL, True)
def test_default_cleanup_requires_confirmed_remote_revocation(self):
api = Api()
original = api.request
def request(method, path, *args, **kwargs):
if method == "DELETE":
self.assertEqual(path, f"/api/pipe/ai/connections/{CONNECTION}")
return 200, {"deleted": True, "remoteRevocationConfirmed": False}
return original(method, path, *args, **kwargs)
api.request = request
with self.assertRaises(qa.qa.QaFailure):
qa.finish_connection(api, CONNECTION, LABEL, False)
def test_provider_failure_retains_owned_source_and_report_without_cleanup_or_faults(self):
with tempfile.TemporaryDirectory() as temporary:
args = SimpleNamespace(run_id=RUN, connection=CONNECTION, vm=VM, model="verified",
owner_marker=Path(temporary) / "owner", vms_dir=Path(temporary),
results_dir=Path(temporary) / "evidence", controller_unit="firecrab-api.service")
api = Api()
with patch.object(qa, "owner_marker"), patch.object(qa, "controller_pid", return_value="42"), \
patch.object(qa, "binary_digest", return_value="e" * 64), \
patch.object(qa, "Guest") as guest, patch.object(qa, "control", return_value="root") as control, \
patch.object(qa.os, "chown"):
guest.return_value.run.return_value = (0, NONCE)
with self.assertRaises(qa.qa.QaFailure), contextlib.redirect_stdout(io.StringIO()):
qa.run_gate(args, api)
self.assertTrue((args.results_dir / "source/executions.txt").exists())
report = json.loads((args.results_dir / "summary.json").read_text())
self.assertEqual(report["status"], "FAILED")
self.assertTrue(report["retainedOnFailure"])
self.assertEqual(report["vmId"], VM)
self.assertEqual(stat.S_IMODE((args.results_dir / "summary.json").stat().st_mode), 0o600)
self.assertFalse(any(method == "DELETE" or "/stop" in path for method, path, _ in api.calls))
self.assertTrue(all(call.args[0] == "show" for call in control.call_args_list))
if __name__ == "__main__":
unittest.main()