#!/usr/bin/env python3 """CC0-1.0. Validate receipts and build the additive external-data supplement.""" import argparse import csv import io import json from pathlib import Path from observe import BASELINE, CASES, CLUSTER, SIZE, require, sha FAILED = {"data-missing", "metadata-only-moved", "pair-moved"} COLUMNS = ("case", "rawMode", "repetition", "checkExit", "infoExit", "compareExit", "exportExit", "metadataUnchanged", "changedGuestBytes", "initialDirectRawChangedBytes") def expected_payload(case): if case in FAILED: return None if case == "data-byte-flip": return bytes([16]) + BASELINE[1:] if case in ("data-replaced", "runtime-override"): return bytes([85]) * CLUSTER + bytes(CLUSTER) + bytes([85]) * (2 * CLUSTER) if case == "data-truncated": return bytes([17]) * CLUSTER + bytes(3 * CLUSTER) return BASELINE def validate(result, harness): require(result["schemaVersion"] == 1, "Unknown schema") require(result["harnessSha256"] == sha(harness.read_bytes()), "Harness mismatch") require(result["fixture"]["baselineSha256"] == sha(BASELINE), "Baseline mismatch") require(result["fixture"]["virtualBytes"] == SIZE, "Capacity mismatch") require(result["fixture"]["clusterBytes"] == CLUSTER, "Cluster mismatch") repeats = result["fixture"]["repetitions"] require(type(repeats) is int and 1 <= repeats <= 5, "Bad repetitions") expected_keys = {(case, raw, rep) for case in CASES for raw in (False, True) for rep in range(1, repeats + 1)} rows = result["rows"] keys = [(r["case"], r["rawMode"], r["repetition"]) for r in rows] require(len(keys) == len(expected_keys) and set(keys) == expected_keys, "Coverage mismatch") for row in rows: case, raw = row["case"], row["rawMode"] require(type(raw) is bool, "Invalid raw-mode type") payload = expected_payload(case) require(row["metadataUnchanged"] is True and row["before"]["metadata"] == row["after"]["metadata"], "Metadata changed") require(row["initialDirectRawChangedBytes"] == (0 if raw else CLUSTER), "Raw baseline mismatch") baseline_raw = BASELINE if raw else bytes([17]) * CLUSTER + bytes([34]) * CLUSTER + BASELINE[2 * CLUSTER:] require(row["before"]["data"] == {"size": SIZE, "sha256": sha(baseline_raw)}, "Initial data mismatch") after = row["after"]["data"] if case in ("data-missing", "metadata-only-moved"): require(after is None, "Missing file unexpectedly present") else: data = baseline_raw if case == "data-byte-flip": data = bytes([16]) + data[1:] elif case == "data-replaced": data = bytes([85]) * SIZE elif case == "data-truncated": data = data[:2 * CLUSTER] require(after == {"size": len(data), "sha256": sha(data)}, "Mutated data mismatch") if case == "runtime-override": require(row["after"]["override"] == {"size": SIZE, "sha256": sha(bytes([85]) * SIZE)}, "Override mismatch") else: require(row["after"]["override"] is None, "Unexpected override") changed = None if payload is None else sum(a != b for a, b in zip(payload, BASELINE)) codes = (1, 0, 2, 1) if payload is None else (0, 0, int(changed != 0), 0) fields = ("checkExit", "infoExit", "compareExit", "exportExit") require(tuple(row[k] for k in fields) == codes, "Outcome mismatch") require(row["changedGuestBytes"] == changed, "Changed-byte count mismatch") require(row["guestSha256"] == (None if payload is None else sha(payload)), "Guest hash mismatch") hashes = None if payload is None else [sha(payload[n:n + CLUSTER]) for n in range(0, SIZE, CLUSTER)] require(row["guestClusterSha256"] == hashes, "Guest cluster hash mismatch") commands = row["commands"] require(len(commands) == 13, "Command coverage mismatch") require(all(type(row[field]) is int for field in fields) and all(type(command["exitCode"]) is int for command in commands), "Exit code type") require(type(row["repetition"]) is int, "Invalid repetition type") image = "restore/disk.qcow2" if case in ( "metadata-only-moved", "pair-moved", "pair-moved-explicit") else "disk.qcow2" inputs = ["-f", "qcow2", image] if case == "pair-moved-explicit": inputs = ["--image-opts", "driver=qcow2,file.filename=restore/disk.qcow2,data-file.driver=file,data-file.filename=restore/payload.bin"] elif case == "runtime-override": inputs = ["--image-opts", "driver=qcow2,file.filename=disk.qcow2,data-file.driver=file,data-file.filename=replacement.bin"] comparison = ([*inputs, "driver=raw,file.filename=expected.raw"] if inputs[0] == "--image-opts" else ["-f", "qcow2", "-F", "raw", image, "expected.raw"]) expected_observers = [ ["qemu-img", "check", "--output=json", *inputs], ["qemu-img", "info", "--backing-chain", "--output=json", *inputs], ["qemu-img", "compare", *comparison], ["qemu-img", "convert", "-O", "raw", *inputs, "after.raw"], ] require([command["command"] for command in commands[-4:]] == expected_observers, "Observer arguments mismatch") if row["checkExit"] == 0: report = json.loads(commands[-4]["stdout"]) require(report.get("format") == "qcow2" and type(report.get("check-errors")) is int and report["check-errors"] == 0 and all(type(report.get(key, 0)) is int and report.get(key, 0) == 0 for key in ("corruptions", "leaks")), "Check report contradicts success") require(row["initialInfo"] == json.loads(commands[8]["stdout"]), "Initial info receipt mismatch") for command in commands[:-4]: require(command["exitCode"] == 0, "Setup failed") for command, name, field in zip(commands[-4:], ("check", "info", "compare", "convert"), fields): require(command["command"][:2] == ["qemu-img", name], "Wrong observer") require(command["exitCode"] == row[field], "Receipt/summary mismatch") for command in commands[-4:]: require(command["cwd"] == ("restore" if case == "pair-moved-cwd" else "."), "Wrong working directory") info = json.loads(commands[-3]["stdout"]) require(len(info) == 1 and info[0]["virtual-size"] == SIZE, "Info shape mismatch") reported = info[0]["format-specific"]["data"] require(reported["data-file"] == "payload.bin" and reported["data-file-raw"] == raw, "Info declaration mismatch") require("backing-filename" not in info[0], "Unexpected backing file") return rows CITATIONS = [ { "id": "qcow2-format", "title": "QEMU v8.2.2: external-data feature bits and standard cluster descriptors", "url": "https://github.com/qemu/qemu/blob/v8.2.2/docs/interop/qcow2.txt", "accessed": "2026-10-21" }, { "id": "qcow2-external-test", "title": "QEMU v8.2.2 iotest 244: external data, zero writes and raw-mode controls", "url": "https://github.com/qemu/qemu/blob/v8.2.2/tests/qemu-iotests/244", "accessed": "2026-10-21" }, { "id": "qcow2-external-open", "title": "QEMU v8.2.2: external data-file opening and creation implementation", "url": "https://github.com/qemu/qemu/blob/v8.2.2/block/qcow2.c", "accessed": "2026-10-21" }, { "id": "qemu-refcounts", "title": "QEMU v8.2.2: external-data treatment in reference-count checks", "url": "https://github.com/qemu/qemu/blob/v8.2.2/block/qcow2-refcount.c", "accessed": "2026-10-21" }, { "id": "qemu-info-implementation", "title": "QEMU v8.2.2: image information collection and NO_IO opening", "url": "https://github.com/qemu/qemu/blob/v8.2.2/qemu-img.c", "accessed": "2026-10-21" } ] COLUMNS_META = [ { "name": "case", "type": "string", "description": "Controlled manipulation; every case begins from a fresh private fixture." }, { "name": "rawMode", "type": "boolean", "description": "data_file_raw creation flag, not a guarantee after out-of-band modification." }, { "name": "repetition", "type": "integer", "description": "Repeat on one host, not an independent reliability sample." }, { "name": "checkExit", "type": "integer", "description": "qemu-img check exit: 0 consistent; 1 opening failed in these fixtures." }, { "name": "infoExit", "type": "integer", "description": "qemu-img info --backing-chain exit; zero does not prove payload readability." }, { "name": "compareExit", "type": "integer", "description": "qemu-img compare exit: 0 equal; 1 completed difference; 2 opening failed." }, { "name": "exportExit", "type": "integer", "description": "Raw export exit: 0 completed; 1 opening failed in these fixtures." }, { "name": "metadataUnchanged", "type": "boolean", "description": "Metadata file size and SHA-256 unchanged from its pre-manipulation snapshot." }, { "name": "changedGuestBytes", "type": "integer|null", "description": "Differing positions across all 262144 guest bytes; null when unreadable." }, { "name": "initialDirectRawChangedBytes", "type": "integer", "description": "Difference count when reading the intact external file directly, before manipulation." } ] def summary(result, raw_bytes): return {"schemaVersion": 1, "generatedAt": result["generatedAt"], "license": "CC0-1.0", "slug": "qcow2-external-data-observations", "title": "QCOW2 external-data observations", "summary": f"{'Sixty' if len(result['rows']) == 60 else len(result['rows'])} offline observations separate external payload identity, working-directory resolution, zero-write semantics and successful metadata inspection.", "published": "2026-10-21", "updated": "2026-10-21", "licenseUrl": "https://creativecommons.org/publicdomain/zero/1.0/", "humanPage": "https://hesela.com/analyses/qcow2-backing-chain-verification/#external-data", "sourceUrl": "https://hesela.com/analyses/qcow2-backing-chain-verification/#external-data", "downloadPath": "/research/qcow2-external/observations.csv", "rawPath": "/research/qcow2-external/results.json", "codePath": "/research/qcow2-external/observe.py", "generatorPath": "/research/qcow2-external/build.py", "citations": CITATIONS, "columns": COLUMNS_META, "method": f"Ten isolated cases x two data_file_raw modes x {'three' if result['fixture']['repetitions'] == 3 else result['fixture']['repetitions']} repetitions on one host. A four-cluster, 256 KiB disk is written with known bytes and one cluster is zeroed. Every successful raw export is compared at every byte position with an independent oracle.", "limits": ["Synthetic offline files, not a VM restore, crash experiment, field failure sample or benchmark.", "One Ubuntu QEMU 8.2.2 package and regular-file backend; no assurance about other versions or runtimes.", "No backing file, guest filesystem, compression, encryption, internal snapshot, dirty bitmap or live writer.", "Raw-mode agreement is checked before manipulation; out-of-band payload edits can break that agreement.", "File hashes identify bytes, not authenticity, intended recovery state or application consistency."], "evidenceType": result["evidenceType"], "environment": result["environment"], "fixture": result["fixture"], "rowCount": len(result["rows"]), "rawSha256": sha(raw_bytes), "harnessSha256": result["harnessSha256"], "rows": [{key: row[key] for key in COLUMNS} for row in result["rows"]]} def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--input", type=Path, required=True) parser.add_argument("--output-dir", type=Path, required=True) args = parser.parse_args() raw = args.input.read_bytes() result = json.loads(raw) validate(result, Path(__file__).with_name("observe.py")) args.output_dir.mkdir(parents=True, exist_ok=False) (args.output_dir / "summary.json").write_text(json.dumps(summary(result, raw), indent=2) + "\n") csv_text = io.StringIO() writer = csv.DictWriter(csv_text, fieldnames=COLUMNS, lineterminator="\n") writer.writeheader() for row in result["rows"]: writer.writerow({key: row[key] for key in COLUMNS}) (args.output_dir / "observations.csv").write_text(csv_text.getvalue()) print(f"Validated {len(result['rows'])} native observations") if __name__ == "__main__": main()