"""Offline consistency checks only; never execute or extract archived inputs.""" from pathlib import Path, PurePosixPath from datetime import datetime import argparse import hashlib import io import json import math import re import stat import zipfile RAW_FILES = {"started.json", "process.json", "result.json", "stdout.log", "stderr.log", "inputs.json", "inputs.zip"} OUTER_FILES = {"prepare.py", "verify.py", "README.md", "manifest.json", "evidence.zip"} def need(condition, message): if not condition: raise ValueError(message) def pin(data): return {"bytes": len(data), "sha256": hashlib.sha256(data).hexdigest()} def valid_pin(value): return (isinstance(value, dict) and set(value) == {"bytes", "sha256"} and type(value["bytes"]) is int and value["bytes"] >= 0 and isinstance(value["sha256"], str) and re.fullmatch(r"[0-9a-f]{64}", value["sha256"]) is not None) def safe(name): return (isinstance(name, str) and bool(name) and "\\" not in name and ":" not in name and all(ord(char) >= 32 for char in name) and not PurePosixPath(name).is_absolute() and all(part not in ("", ".", "..") for part in name.split("/"))) def run_name(name): return isinstance(name, str) and re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._-]*", name) is not None def parsed(data): def pairs(items): result = {} for key, value in items: need(key not in result, "duplicate JSON key: " + key) result[key] = value return result def constant(value): raise ValueError("nonfinite JSON constant: " + value) return json.loads(data, object_pairs_hook=pairs, parse_constant=constant) def inventory(archive, expected): names = archive.namelist() need(len(names) == len(set(names)) and set(names) == set(expected), "exact ZIP membership") for entry in archive.infolist(): need(safe(entry.filename) and not entry.is_dir(), "unsafe ZIP member: " + entry.filename) need(not stat.S_ISLNK(entry.external_attr >> 16), "ZIP symlink: " + entry.filename) need(valid_pin(expected[entry.filename]), "invalid member pin: " + entry.filename) need(entry.file_size == expected[entry.filename]["bytes"], "ZIP recorded size: " + entry.filename) def ordered_times(*values): times = [datetime.fromisoformat(value) for value in values] return all(value.utcoffset() is not None for value in times) and times == sorted(times) def verify_data(manifest, read): """Shared read-only verifier for packaged bytes or prepare's in-memory payload.""" need(manifest["schema"] == 1 and manifest["kind"] == "allocation-multiselect-cargo-gates", "manifest schema") runs, logical, stored = manifest["runs"], manifest["logicalFiles"], manifest["members"] helpers = manifest["captureHelpers"] need(isinstance(helpers, dict) and set(helpers) == {"gate.py", "source-proof.py"}, "capture helper names") need(all(isinstance(name, str) and re.fullmatch(r"helpers/[0-9a-f]{64}\.py", name) for name in helpers.values()), "capture helper mapping") need(isinstance(runs, dict) and bool(runs) and all(run_name(run) for run in runs), "explicit run inventory") need(all(safe(name) for name in logical) and all(safe(name) for name in stored), "unsafe manifest path") need(set(logical.values()) == set(stored), "unreferenced or absent stored members") for name, expected in stored.items(): need(valid_pin(expected) and pin(read(name)) == expected, "stored member hash: " + name) expected_logical = {"cargo/" + run + "/" + name for run in runs for name in RAW_FILES} expected_logical.update(spec["helper"] for spec in runs.values()) expected_logical.update(helpers.values()) for run, spec in runs.items(): if "sourceProof" in spec: proof_spec = spec["sourceProof"] need(set(proof_spec) == {"commit", "pin", "comparator"}, run + ": exact source proof spec") need(isinstance(proof_spec["commit"], str) and re.fullmatch(r"[0-9a-f]{40}", proof_spec["commit"]), run + ": named full source commit") need(valid_pin(proof_spec["pin"]), run + ": source proof pin") need(isinstance(proof_spec["comparator"], str) and re.fullmatch(r"comparators/[0-9a-f]{64}\.py", proof_spec["comparator"]), run + ": comparator path") expected_logical.update({"cargo/" + run + "/source-proof.json", proof_spec["comparator"]}) need(set(logical) == expected_logical, "exact logical raw/helper inventory") for name, destination in logical.items(): if name.startswith("cargo/") and name.rsplit("/", 1)[1] in {"inputs.json", "inputs.zip"}: expected = "capsules/" + stored[destination]["sha256"] + PurePosixPath(name).suffix need(destination == expected, "input capsule mapping: " + name) else: need(destination == name, "only input maps/archives may be deduplicated: " + name) def raw(name): return read(logical[name]) for name, helper in helpers.items(): need(helper == "helpers/" + pin(raw(helper))["sha256"] + ".py", "named helper hash: " + name) seen_capsules = set() for run, spec in runs.items(): prefix = "cargo/" + run + "/" start = parsed(raw(prefix + "started.json")) process = parsed(raw(prefix + "process.json")) result = parsed(raw(prefix + "result.json")) required_start = {"start", "command", "cwd", "head", "scope", "inputs", "archive", "inputCount", "environment", "helper", "cargo", "rustc"} need(required_start <= start.keys(), run + ": incomplete start receipt") need(all(result.get(key) == value for key, value in start.items()), run + ": start/result relationship") command = result["command"] need(isinstance(command, list) and command and command[0] == "cargo" and all(isinstance(arg, str) for arg in command), run + ": Cargo command") need(command == process["command"] == spec["command"], run + ": command relationship") need(result["head"] == spec["head"] and re.fullmatch(r"(?:[0-9a-f]{40}|[0-9a-f]{64})", result["head"]), run + ": HEAD") need(type(result["exitCode"]) is int and result["exitCode"] == spec["exitCode"], run + ": recorded exit") need(result["inputsUnchanged"] is True, run + ": changed inputs refused") need(type(result["seconds"]) in (int, float) and math.isfinite(result["seconds"]) and result["seconds"] >= 0, run + ": duration") need(type(process["pid"]) is int and process["pid"] > 0, run + ": process identity") need(ordered_times(start["start"], process["observed"], result["finish"]), run + ": completed timestamp order") for field, filename in [("inputs", "inputs.json"), ("archive", "inputs.zip"), ("stdout", "stdout.log"), ("stderr", "stderr.log")]: need(result[field] == pin(raw(prefix + filename)), run + ": " + field + " pin") need(valid_pin(result["helper"]) and result["helper"] == pin(raw(spec["helper"])), run + ": helper pin") need(spec["helper"] == "helpers/" + result["helper"]["sha256"] + ".py", run + ": helper mapping") inputs = parsed(raw(prefix + "inputs.json")) need(isinstance(inputs, dict) and all(safe(name) for name in inputs), run + ": safe input inventory") need(type(result["inputCount"]) is int and len(inputs) == result["inputCount"] == spec["inputCount"], run + ": input counts") if "sourceProof" in spec: proof_spec = spec["sourceProof"] proof_bytes = raw(prefix + "source-proof.json") proof = parsed(proof_bytes) need(pin(proof_bytes) == proof_spec["pin"], run + ": source proof pinned bytes") need(proof["run"] == run and proof["commit"] == proof_spec["commit"], run + ": proof run/commit identity") need(proof["status"] == "PASS exact captured input inventory and Git blob bytes", run + ": source proof status") need(proof["inputCount"] == len(inputs) and proof["inputMap"] == result["inputs"], run + ": proof input relationship") need(proof["result"] == pin(raw(prefix + "result.json")), run + ": proof result relationship") need(ordered_times(result["finish"], proof["verified"]), run + ": proof time after closed gate") comparator = proof_spec["comparator"] need(valid_pin(proof["comparator"]) and pin(raw(comparator)) == proof["comparator"], run + ": comparator bytes") need(comparator == "comparators/" + proof["comparator"]["sha256"] + ".py", run + ": comparator identity") need(sum(name.startswith("crates/vendor/iced_widget/") for name in inputs) == 71, run + ": vendor count") capsule = (logical[prefix + "inputs.json"], logical[prefix + "inputs.zip"]) if capsule not in seen_capsules: with zipfile.ZipFile(io.BytesIO(raw(prefix + "inputs.zip"))) as source: inventory(source, inputs) for name, expected in inputs.items(): need(pin(source.read(name)) == expected, run + ": source hash " + name) seen_capsules.add(capsule) artifacts = [] for line in raw(prefix + "stdout.log").decode("utf-8", errors="replace").splitlines(): try: value = json.loads(line) except json.JSONDecodeError: continue if isinstance(value, dict) and value.get("reason") == "compiler-artifact" and value.get("executable"): artifacts.append(value) need(isinstance(result["artifacts"], list), run + ": artifact records") need(artifacts == [record["cargo"] for record in result["artifacts"]], run + ": Cargo JSON artifact relationship") for record in result["artifacts"]: need(set(record) == {"cargo", "readback"} and (record["readback"] is None or valid_pin(record["readback"])), run + ": artifact readback pin") print(run, "recorded exit", result["exitCode"], "inputs", len(inputs), "executable records", len(artifacts)) return {"runs": len(runs), "recordedFailures": sum(spec["exitCode"] != 0 for spec in runs.values()), "logicalFiles": len(logical), "storedMembers": len(stored), "sourceCapsulePairs": len(seen_capsules), "captureHelpers": helpers, "sourceProofs": sum("sourceProof" in spec for spec in runs.values())} def verify_package(root): files = parsed((root / "files.json").read_bytes()) need(set(files) == OUTER_FILES, "outer file inventory") for name, expected in files.items(): need(valid_pin(expected) and pin((root / name).read_bytes()) == expected, "outer hash: " + name) manifest = parsed((root / "manifest.json").read_bytes()) with zipfile.ZipFile(root / "evidence.zip") as archive: inventory(archive, manifest["members"]) return verify_data(manifest, archive.read) def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("package", nargs="?", type=Path, default=Path(__file__).resolve().parent) args = parser.parse_args() print("PASS offline consistency:", verify_package(args.package)) print("Historical receipts only; no source/binary execution or native acceptance claim.") if __name__ == "__main__": main()