#!/usr/bin/env python3 """CC0-1.0. Validate observations, then derive public QCOW2 evidence artifacts.""" import argparse import csv import hashlib import json from pathlib import Path import shutil CASES = { "intact": (0, 0, True, True, 0), "base-replaced": (0, 1, True, False, 131072), "base-missing": (1, 2, True, None, None), "overlay-only-relocated": (1, 2, True, None, None), "chain-relocated": (0, 0, True, True, 0), "payload-byte-flip": (0, 1, False, False, 1), "safe-rebase": (0, 0, False, True, 0), "unsafe-rebase": (0, 1, False, False, 131072), "base-truncated": (0, 1, True, False, 131072), } COLUMNS = [ ("repetition", "Repeat number on one host, not an independent reliability trial."), ("case", "Controlled fixture manipulation."), ("checkExit", "qemu-img check status: 0 completed consistently; 1 could not open the missing dependency. Null for raw size control."), ("compareExit", "qemu-img compare: 0 equal under default rules; 1 differs; 2 opening error."), ("exportExit", "Full raw export: 0 success; 1 opening failure; null for size control."), ("topHashUnchanged", "Overlay SHA-256 equals its own pre-manipulation hash; null for size control."), ("guestMatchesBaseline", "Every exported byte equals the independent baseline; null when unreadable or not applicable."), ("changedGuestBytes", "Differing byte positions against the equal-length baseline; null when unavailable."), ("strictCompareExit", "Strict compare status in the zero-tail control; null elsewhere."), ] COLUMN_TYPES = { "repetition": "integer", "case": "string", "compareExit": "integer", "checkExit": "integer|null", "exportExit": "integer|null", "topHashUnchanged": "boolean|null", "guestMatchesBaseline": "boolean|null", "changedGuestBytes": "integer|null", "strictCompareExit": "integer|null", } CITATIONS = [{"id":"qemu-tool","title":"QEMU 8.2.2: qemu-img check, compare and rebase","url":"https://github.com/qemu/qemu/blob/v8.2.2/docs/tools/qemu-img.rst","accessed":"2026-10-18"},{"id":"qcow2-format","title":"QEMU 8.2.2: QCOW2 image file format, cluster mapping and zero clusters","url":"https://github.com/qemu/qemu/blob/v8.2.2/docs/interop/qcow2.txt","accessed":"2026-10-18"},{"id":"qemu-refcounts","title":"QEMU 8.2.2: qcow2_check_refcounts implementation","url":"https://github.com/qemu/qemu/blob/v8.2.2/block/qcow2-refcount.c","accessed":"2026-10-18"},{"id":"libvirt-chain","title":"libvirt: disk image chains, backing formats and relocation","url":"https://libvirt.org/kbase/backing_chains.html","accessed":"2026-10-18"}] def require(value, message): if not value: raise ValueError(message) def sha(data): return hashlib.sha256(data).hexdigest() def validate_receipts(row): require(type(row["repetition"]) is int, "repetition type") commands = row["commands"] require(isinstance(commands, list), "command receipts") tail = row["case"] == "zero-tail-size" require(len(commands) >= (2 if tail else 3), "missing command receipts") for receipt in commands: require(type(receipt["exitCode"]) is int, "receipt exit type") require(isinstance(receipt["command"], list) and all(isinstance(arg, str) for arg in receipt["command"]), "command arguments") require(isinstance(receipt["stdout"], str) and isinstance(receipt["stderr"], str), "command output") if tail: expected = [ (["qemu-img", "compare", "-f", "raw", "-F", "raw", "small.raw", "large.raw"], "compareExit"), (["qemu-img", "compare", "-s", "-f", "raw", "-F", "raw", "small.raw", "large.raw"], "strictCompareExit"), ] else: image = "restore/overlay.qcow2" if "relocated" in row["case"] else "overlay.qcow2" expected = [ (["qemu-img", "check", "--output=json", "-f", "qcow2", image], "checkExit"), (["qemu-img", "compare", "-f", "qcow2", "-F", "raw", image, "expected.raw"], "compareExit"), (["qemu-img", "convert", "-f", "qcow2", "-O", "raw", image, "after.raw"], "exportExit"), ] check_stdout = commands[-3]["stdout"] parsed = json.loads(check_stdout) if check_stdout.strip() else None require(row["checkResult"] == parsed, "check JSON receipt") for receipt, (command, field) in zip(commands[-len(expected):], expected): require(receipt["command"] == command, "final command sequence") require(type(row[field]) is int and row[field] == receipt["exitCode"], "receipt mismatch: " + field) def validate(data, harness): require(data["schemaVersion"] == 1, "schema") require(data["harnessSha256"] == sha(harness), "harness hash") repetitions = data["repetitions"] require(type(repetitions) is int and 1 <= repetitions <= 10, "repetitions") require(data["clusterBytes"] == 65536 and data["virtualSizeBytes"] == 262144, "geometry") baseline = bytes([34]) * 65536 + bytes(65536) + bytes([17]) * 131072 changed = bytes([34]) * 65536 + bytes(65536) + bytes([51]) * 131072 expected_ids = {(r, c) for r in range(1, repetitions + 1) for c in [*CASES, "zero-tail-size"]} rows = data["observations"] require(len(rows) == len(expected_ids), "row count") require({(r["repetition"], r["case"]) for r in rows} == expected_ids, "coverage") for row in rows: validate_receipts(row) if row["case"] == "zero-tail-size": require((row["compareExit"], row["strictCompareExit"]) == (0, 1), "size comparison") require((row["smallSizeBytes"], row["largeSizeBytes"]) == (65536, 131072), "sizes") small = bytes([17]) * 65536 require(row["smallSha256"] == sha(small), "small payload") require(row["largeSha256"] == sha(small + bytes(65536)), "large payload") continue expected = CASES[row["case"]] fields = ("checkExit", "compareExit", "topHashUnchanged", "guestMatchesBaseline", "changedGuestBytes") for key, value in zip(fields, expected): require(type(row[key]) is type(value) and row[key] == value, key) require(row["topHashUnchanged"] == (row["topBeforeSha256"] == row["topAfterSha256"]), "top hashes") require(row["baselineGuestSha256"] == sha(baseline), "baseline hash") missing = expected[3] is None require(row["exportExit"] == (1 if missing else 0), "export") payload = changed if row["case"] in ("base-replaced", "unsafe-rebase") else baseline if row["case"] == "payload-byte-flip": payload = bytes([35]) + baseline[1:] if row["case"] == "base-truncated": payload = bytes([34]) * 65536 + bytes(3 * 65536) target = None if missing else sha(payload) require(row["guestSha256"] == target and row["expectedGuestSha256"] == target, "payload hash") return rows def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--input", type=Path, required=True) parser.add_argument("--harness", type=Path, required=True) parser.add_argument("--site", type=Path, required=True) parser.add_argument("--surface", choices=("com", "dev"), required=True) args = parser.parse_args() data = json.loads(args.input.read_text()) harness = args.harness.read_bytes() source_rows = validate(data, harness) public = args.site / "public" research = public / "research" research.mkdir(parents=True, exist_ok=True) (research / "qcow2-dependencies.py").write_bytes(harness) raw_bytes = (json.dumps(data, indent=2) + "\n").encode("utf-8") (research / "qcow2-dependencies.json").write_bytes(raw_bytes) shutil.copyfile(__file__, research / "build-qcow2.py") rows = [{key: r.get(key) for key, _ in COLUMNS} for r in source_rows] metadata = { "schemaVersion": 1, "slug": "qcow2-dependency-observations", "title": "QCOW2 dependency and verification observations", "summary": "Ten controlled cases separate overlay identity, backing dependencies, image consistency and guest-byte comparison.", "published": "2026-10-18", "updated": "2026-10-18", "generatedAt": data["generatedAt"], "evidenceType": "measured-tool-behavior-with-synthetic-inputs", "license": "CC0-1.0 for original Hesela code and observations; QEMU and upstream documentation retain their own licenses.", "licenseUrl": "https://creativecommons.org/publicdomain/zero/1.0/", "humanPage": "https://hesela.com/analyses/qcow2-backing-chain-verification/", "sourceUrl": "https://hesela.com/analyses/qcow2-backing-chain-verification/", "downloadPath": "/data/datasets/qcow2-dependency-observations.csv", "rawPath": "/research/qcow2-dependencies.json", "codePath": "/research/qcow2-dependencies.py", "generatorPath": "/research/build-qcow2.py", "rowCount": len(rows), "columns": [{"name": n, "type": COLUMN_TYPES[n], "description": d} for n, d in COLUMNS], "rows": rows, "environment": data["environment"], "citations": CITATIONS, "method": f"Nine offline QCOW2 scenarios use four 65536-byte clusters and a known full-byte oracle. A tenth raw-image control compares unequal virtual sizes with a zero tail. {data['repetitions']} repetitions on one host; explicit formats, closed writers, no live VM.", "limits": ["Synthetic bytes and controlled file mutations, not a VM restore, power-loss test, performance benchmark or failure-rate sample.", "One Ubuntu QEMU 8.2.2 package; other versions, options and formats can differ.", "Uncompressed, unencrypted images with standard clusters and one raw base; no internal snapshots, external data files or dirty bitmaps.", "Successful guest-byte export does not establish guest filesystem, application or multi-disk consistency."], "provenance": {"harnessSha256": data["harnessSha256"], "rawSha256": sha(raw_bytes)} } if args.surface == "com": dest = args.site / "src/data/research/qcow2-dependencies.json" dest.parent.mkdir(parents=True, exist_ok=True) dest.write_text(json.dumps(metadata, indent=2) + "\n") else: dest = public / "data/datasets" dest.mkdir(parents=True, exist_ok=True) (dest / "qcow2-dependency-observations.json").write_text(json.dumps(metadata, indent=2) + "\n") with (dest / "qcow2-dependency-observations.csv").open("w", newline="") as output: writer = csv.DictWriter(output, fieldnames=[n for n, _ in COLUMNS], lineterminator="\n") writer.writeheader() writer.writerows(rows) print(f"Validated {len(rows)} observations and generated {args.surface} artifacts.") if __name__ == "__main__": main()