#!/usr/bin/env python3 """Offline plan/ownership/syntax checks. Never execute the guest probe or contact an API.""" import ast import copy import hashlib import importlib.util import json from pathlib import Path import stat import tempfile import unittest import uuid PATH = Path(__file__).with_name("qa-sandbox-openrc-v1.py") spec = importlib.util.spec_from_file_location("openrc_plan", PATH) qa = importlib.util.module_from_spec(spec) spec.loader.exec_module(qa) RUN = uuid.UUID("659958b5-b7a8-4e81-a5c0-b0f70585ca5e") VM = "e5f36c4d-eb59-4c5d-8a0e-631335234f75" RUNTIME = "4fe36daa-0f63-4b13-8b32-53bc5c9e6d90" GENERATION = "399122a6-939a-4c2d-8751-d6df845b15af" def exported_baseline(): return ({"id": qa.PROJECT, "owner": "local", "spec": {"environmentId": None, "vmId": None, "sourcePath": None, "selection": None, "instructions": "NoModel: retain strict seed baseline"}}, {"id": qa.ENVIRONMENT, "projectId": qa.PROJECT, "owner": "local", "state": "ready", "generation": GENERATION, "dnsMode": "private_only", "roles": [{"role": "debug", "vmId": VM, "runtime": {"vmId": VM, "runtimeId": RUNTIME}, "agent": "ready", "network": "ready"}]}) class OfflineTests(unittest.TestCase): def test_preparer_has_no_network_process_or_ssh_execution_surface(self): tree = ast.parse(PATH.read_text()) imported = {node.names[0].name.split(".")[0] if isinstance(node, ast.Import) else node.module.split(".")[0] for node in ast.walk(tree) if isinstance(node, (ast.Import, ast.ImportFrom))} self.assertTrue(imported.isdisjoint({"urllib", "http", "socket", "subprocess", "requests", "asyncio"})) probe = PATH.parent / "fixtures/sandbox-openrc-v1/probe.py" ast.parse(probe.read_text()) # Parse literal contracts without importing or running the guest payload. literals = {node.targets[0].id: ast.literal_eval(node.value) for node in ast.parse(probe.read_text()).body if isinstance(node, ast.Assign) and isinstance(node.targets[0], ast.Name) and node.targets[0].id in ("LITERAL_ARGS", "LITERAL_ENV", "STDIN")} self.assertEqual(literals["LITERAL_ARGS"], qa.LITERAL_ARGS) self.assertEqual(literals["LITERAL_ENV"], qa.LITERAL_ENV) self.assertEqual(literals["STDIN"], qa.STDIN.encode()) def test_fixed_target_and_manual_gates_do_not_prepare_a_project_mutation(self): planned = qa.plan(RUN) self.assertTrue(planned["preparedOnly"]) self.assertFalse(planned["liveQaAccepted"]) self.assertFalse(planned["projectMutationPrepared"]) self.assertTrue(all(value is False for value in planned["rootGates"].values())) self.assertEqual(planned["projectId"], qa.PROJECT) self.assertEqual(planned["environmentId"], qa.ENVIRONMENT) self.assertEqual(planned["role"], "debug") self.assertEqual(planned["selection"], qa.SELECTION) self.assertTrue(planned["remoteSourcePath"].startswith("/srv/firecrab-qa/openrc-v1-414-")) self.assertIsNone(planned["frozenBaseline"]) self.assertIs(planned["rootGates"]["explicitEnvironmentBindingToAReviewed"], False) def test_no_model_baseline_and_owned_strict_runtime_are_frozen_without_changes(self): project, environment = exported_baseline() original = copy.deepcopy((project, environment)) frozen = qa.baseline(project, environment) self.assertEqual((project, environment), original) self.assertEqual(frozen["projectSpecSha256"], qa.digest(project["spec"])) self.assertIsNone(frozen["projectBefore"]["spec"]["environmentId"]) self.assertIsNone(frozen["originalEnvironmentId"]) self.assertEqual(frozen["requiredExecutionEnvironmentId"], qa.ENVIRONMENT) bound = copy.deepcopy(project) bound["spec"]["environmentId"] = qa.ENVIRONMENT bound_before = copy.deepcopy(bound) bound_frozen = qa.baseline(bound, environment) self.assertEqual(bound, bound_before) self.assertEqual(bound_frozen["projectSpecSha256"], qa.digest(bound_before["spec"])) self.assertEqual(bound_frozen["originalEnvironmentId"], qa.ENVIRONMENT) self.assertEqual((project, environment), original) for changed in ("project", "environment", "owner", "noModel", "binding", "dns", "state", "runtime", "agent"): p, e = copy.deepcopy(original) if changed == "project": p["id"] = str(uuid.uuid4()) if changed == "environment": e["id"] = str(uuid.uuid4()) if changed == "owner": e["owner"] = "foreign" if changed == "noModel": p["spec"]["instructions"] = "Run arbitrary tasks" if changed == "binding": p["spec"]["environmentId"] = str(uuid.uuid4()) if changed == "dns": e["dnsMode"] = "legacy_forward" if changed == "state": e["state"] = "stopped" if changed == "runtime": e["roles"][0]["runtime"]["vmId"] = str(uuid.uuid4()) if changed == "agent": e["roles"][0]["agent"] = "unsupported" with self.subTest(changed=changed), self.assertRaises(ValueError): qa.baseline(p, e) def test_native_tool_literal_nul_stdin_and_hard_limits_survive_json(self): planned = json.loads(json.dumps(qa.plan(RUN))) self.assertEqual(len({item["id"] for item in planned["tasks"]}), 5) for item in planned["tasks"]: request = item["request"] self.assertEqual(request["selection"], qa.SELECTION) self.assertEqual(request["environmentRole"], "debug") self.assertEqual(request["coordination"]["retryLimit"], 0) self.assertIsNone(request["retryOf"]) for command in item["invocations"]: self.assertEqual(command["version"], 1) self.assertIsInstance(command["env"], list) self.assertEqual(command["cwd"], "qa cwd") self.assertFalse(any("\0" in value for value in command["argv"])) self.assertTrue(1 <= command["limits"]["wallSeconds"] <= 120) self.assertEqual(command["limits"]["cpuSeconds"], 3) literal = planned["tasks"][0]["invocations"][0] self.assertEqual(literal["argv"][4:], qa.LITERAL_ARGS) self.assertIn("\0", literal["stdin"]) self.assertEqual(literal["stdin"].encode(), b"public-input\x00bytes\n") self.assertEqual(planned["tasks"][1]["invocations"][0]["limits"]["memoryMib"], 64) self.assertEqual(planned["tasks"][2]["invocations"][0]["limits"]["wallSeconds"], 2) self.assertEqual(planned["tasks"][3]["invocations"][0]["limits"]["outputBytes"], 256) cancel = planned["tasks"][4] self.assertEqual([c["argv"][2] for c in cancel["invocations"]], ["cancel-ready", "cancel-running"]) self.assertIn("retain_artifact", cancel["request"]["input"]) def test_admission_guards_keep_attempt_runtime_and_exact_request_identity(self): project, environment = exported_baseline() planned = qa.plan(RUN, qa.baseline(project, environment)) item = planned["tasks"][0] saved = {"id": item["id"], "projectId": qa.PROJECT, "selection": qa.SELECTION, "request": item["request"], "environment": planned["frozenBaseline"]["environment"], "runtime": planned["frozenBaseline"]["runtime"], "attemptId": str(uuid.uuid4())} refs = qa.admission(saved, planned, item) self.assertEqual(refs, qa.admission(copy.deepcopy(saved), planned, item, refs)) for field in ("id", "projectId", "selection", "request", "environment", "runtime", "attemptId"): changed = copy.deepcopy(saved) changed[field] = str(uuid.uuid4()) if field in ("id", "projectId", "attemptId") else {} with self.subTest(field=field), self.assertRaises(ValueError): qa.admission(changed, planned, item, refs) def test_prepare_writes_private_public_payload_files_once_and_never_runs_them(self): with tempfile.TemporaryDirectory() as directory: target = Path(directory) / "prepared" qa.prepare(target, RUN) self.assertEqual(stat.S_IMODE(target.stat().st_mode), 0o700) for p in target.rglob("*"): if p.is_file(): self.assertEqual(stat.S_IMODE(p.stat().st_mode), 0o600) self.assertEqual((target / "source/probe.txt").read_text(), "not executed\n") self.assertEqual(len(list((target / "source/qa cwd").iterdir())), 1) self.assertFalse(list((target / "source").rglob("*-literal.json"))) with self.assertRaises(FileExistsError): qa.prepare(target, RUN) def test_artifact_checks_bind_nonce_bytes_literal_contract_and_actual_exit(self): project, environment = exported_baseline() planned = qa.plan(RUN, qa.baseline(project, environment));item = planned["tasks"][0] nonce = "d" * 32 value = {"runId": RUN.hex, "case": "literal", "nonce": nonce, "uid": 1000, "euid": 1000, "gid": 2000, "egid": 2000, "groups": [], "noNewPrivs": 1, "capEff": "0000", "pid": 123, "startTicks": 456, "bootId": str(uuid.uuid4()), "cwdName": "qa cwd", "addressSpaceLimit": [96 * 1024 * 1024] * 2, "cpuLimit": [3, 3], "nofileLimit": [1024, 1024], "argv": qa.LITERAL_ARGS, "env": {"QA_V1_LITERAL": qa.LITERAL_ENV, "QA_V1_RUN": RUN.hex}, "stdinHex": qa.STDIN.encode().hex(), "stdinSha256": hashlib.sha256(qa.STDIN.encode()).hexdigest()} payload = (json.dumps(value) + "\n").encode() saved = {"id": item["id"], "projectId": qa.PROJECT, "attemptId": str(uuid.uuid4()), "selection": qa.SELECTION, "request": item["request"], "state": "result", "runtime": planned["frozenBaseline"]["runtime"], "environment": planned["frozenBaseline"]["environment"], "events": [{"name": "run_argv", "sequence": 3, "input": json.dumps(item["invocations"][0]), "output": f"OPENRCV1:{RUN.hex}:literal:stdout:{nonce}\n", "stderr": f"OPENRCV1:{RUN.hex}:literal:stderr\n", "exitCode": 7}], "tests": [{"eventSequence": 3, "passed": False}], "result": f"Actual guest nonce {nonce}"} metadata = {"id": str(uuid.uuid4()), "taskId": saved["id"], "attemptId": saved["attemptId"], "path": item["artifactPath"], "size": len(payload), "sha256": hashlib.sha256(payload).hexdigest()} saved["artifacts"] = [metadata] report = qa.artifact(saved, planned, item, metadata, payload) self.assertEqual(report["rawExitCodes"], [7]);self.assertFalse(report["liveQaAccepted"]) for changed in ("bytes", "owner", "nonce", "exit", "stderr", "passed", "replay", "edit", "source", "result", "rootGid", "effectiveGid", "missingGid", "effectiveUid"): task = copy.deepcopy(saved);meta = copy.deepcopy(metadata);body = payload if changed == "bytes": body += b" " if changed == "owner": meta["attemptId"] = str(uuid.uuid4()) if changed == "nonce": task["events"][0]["output"] = task["events"][0]["output"].replace(nonce, "e" * 32) if changed == "exit": task["events"][0]["exitCode"] = 0 if changed == "stderr": task["events"][0]["stderr"] = "" if changed == "passed": task["tests"][0]["passed"] = True if changed == "replay": task["events"].append(task["events"][0]) if changed == "edit": task["events"].append({"name": "write_file"}) if changed == "source": task["changes"] = [{"path": "qa cwd/probe.py", "after": "fake probe"}] if changed == "result": task["result"] = "No actual nonce" if changed in ("rootGid", "effectiveGid", "missingGid", "effectiveUid"): value_changed = copy.deepcopy(value) if changed == "rootGid": value_changed.update(gid=0, egid=0) if changed == "effectiveGid": value_changed["egid"] = 1001 if changed == "missingGid": del value_changed["gid"] del value_changed["egid"] if changed == "effectiveUid": value_changed["euid"] = 1001 body = (json.dumps(value_changed) + "\n").encode() meta.update(size=len(body), sha256=hashlib.sha256(body).hexdigest()) task["artifacts"] = [meta] with self.subTest(changed=changed), self.assertRaises(ValueError): qa.artifact(task, planned, item, meta, body) if __name__ == "__main__": unittest.main(verbosity=2)