#!/usr/bin/env python3 # SPDX-License-Identifier: CC0-1.0 """Bounded synthetic bzip2 salvage experiment. Never accepts an external archive.""" import argparse import hashlib import json import os from pathlib import Path import platform import re import shutil import subprocess import tempfile from datetime import datetime, timezone BLOCK_MAGIC = "001100010100000101011001001001100101001101011001" END_MAGIC = "000101110111001001000101001110000101000010010000" PAYLOAD_BYTES = 320000 def require(condition, message): if not condition: raise RuntimeError(message) def sha(data): return hashlib.sha256(data).hexdigest() def payload(): return b"".join(hashlib.sha256(b"hesela-bzip2-salvage-v1:" + n.to_bytes(4, "big")).digest() for n in range(PAYLOAD_BYTES // 32)) def bit_positions(data, magic): # Small synthetic fixtures only; this is not a general-purpose format parser. bits = "".join(format(byte, "08b") for byte in data) return [match.start() for match in re.finditer("(?=" + magic + ")", bits)] def flip_bit(data, offset): if not 0 <= offset < len(data) * 8: raise ValueError("bit offset outside fixture") out = bytearray(data) out[offset // 8] ^= 1 << (7 - offset % 8) return bytes(out) def invoke(argv, cwd, data=None): env = {key: value for key, value in os.environ.items() if key not in ("BZIP", "BZIP2")} env["LC_ALL"] = "C" return subprocess.run(argv, cwd=cwd, input=data, capture_output=True, env=env, timeout=15, check=False) def run(): bzip = shutil.which("bzip2") recover = shutil.which("bzip2recover") require(bzip and recover, "Install bzip2 and bzip2recover before running") original = payload() with tempfile.TemporaryDirectory(prefix="hesela-bzip2-") as temp: root = Path(temp) compressed = invoke([bzip, "-1c"], root, original) require(compressed.returncode == 0, "compression failed") clean = compressed.stdout markers = bit_positions(clean, BLOCK_MAGIC) ends = bit_positions(clean, END_MAGIC) require(len(markers) == 4 and len(ends) == 1, "fixture no longer has four blocks and one end marker") require(clean[:4] == b"BZh1" and markers[0] == 32, "unexpected compression header") require(markers[-1] < ends[0], "end marker precedes data") second_mid = (markers[1] + markers[2]) // 2 truncation_byte = ((markers[2] + markers[3]) // 2) // 8 fixtures = [ ("clean", clean, {"kind": "none"}), ("block-crc-bit", flip_bit(clean, markers[1] + 48), {"kind": "one-bit-flip", "bitOffset": markers[1] + 48}), ("block-data-bit", flip_bit(clean, second_mid), {"kind": "one-bit-flip", "bitOffset": second_mid}), ("block-marker-bit", flip_bit(clean, markers[1]), {"kind": "one-bit-flip", "bitOffset": markers[1]}), ("stream-crc-bit", flip_bit(clean, ends[0] + 48), {"kind": "one-bit-flip", "bitOffset": ends[0] + 48}), ("truncated-third-block", clean[:truncation_byte], {"kind": "truncate", "firstMissingByteOffset": truncation_byte}), ] oracle = [] observations = [] for fixture_id, archive, change in fixtures: folder = root / fixture_id folder.mkdir() (folder / "fixture.bz2").write_bytes(archive) test = invoke([bzip, "-t", "fixture.bz2"], folder) decode = invoke([bzip, "-dc", "fixture.bz2"], folder) recovered = invoke([recover, "fixture.bz2"], folder) candidates = sorted(folder.glob("rec*fixture.bz2")) require(recovered.returncode == 0 and candidates, "recovery tool failed to create bounded candidates") # Source scanner reports inclusive input-bit ranges after each marker. ranges = [(int(a), int(b)) for a, b in re.findall(rb"block \d+ runs from (\d+) to (\d+)\n", recovered.stderr)] require(len(ranges) == len(candidates), "candidate range parsing drift") records, accepted_outputs, matched = [], [], [] start = 0 for candidate, (first_bit, last_bit) in zip(candidates, ranges): check = invoke([bzip, "-t", candidate.name], folder) unpack = invoke([bzip, "-dc", candidate.name], folder) valid = check.returncode == 0 and unpack.returncode == 0 match = None if fixture_id == "clean": require(valid, "pristine recovery candidate failed validation") expected = original[start:start + len(unpack.stdout)] require(unpack.stdout == expected, "pristine output differs from source") match = len(oracle) + 1 oracle.append({"block": match, "startByte": start, "endByteExclusive": start + len(expected), "bytes": len(expected), "sha256": sha(expected)}) start += len(expected) elif valid: hits = [block for block in oracle if unpack.stdout == original[block["startByte"]:block["endByteExclusive"]]] require(len(hits) == 1, "valid candidate lacks a unique reference match") match = hits[0]["block"] if valid: accepted_outputs.append(unpack.stdout) matched.append(match) records.append({ "file": candidate.name, "compressedSha256": sha(candidate.read_bytes()), "inputBitRangeInclusive": [first_bit, last_bit], "integrityExit": check.returncode, "decodeExit": unpack.returncode, "returnedBytes": len(unpack.stdout), "returnedSha256": sha(unpack.stdout), "validationPassed": valid, "referenceBlock": match, }) joined = b"".join(accepted_outputs) if fixture_id == "clean": require(joined == original, "pristine split did not reconstruct whole payload") missing = [b for b in oracle if b["block"] not in matched] observations.append({ "id": fixture_id, "mutation": change, "compressedBytes": len(archive), "compressedSha256": sha(archive), "wholeIntegrityExit": test.returncode, "wholeDecodeExit": decode.returncode, "wholeReturnedBytes": len(decode.stdout), "wholeReturnedSha256": sha(decode.stdout), "wholeOutputMatches": decode.stdout == original, "wholeAccepted": decode.returncode == 0 and decode.stdout == original, "recoveryExit": recovered.returncode, "candidateCount": len(candidates), "candidates": records, "validatedReferenceBlocks": matched, "validatedBytes": len(joined), "missingByteRanges": [[b["startByte"], b["endByteExclusive"]] for b in missing], "concatenatedSha256": sha(joined), "concatenationMatchesOriginal": joined == original, }) expected_blocks = { "clean": [1, 2, 3, 4], "block-crc-bit": [1, 3, 4], "block-data-bit": [1, 3, 4], "block-marker-bit": [3, 4], "stream-crc-bit": [1, 2, 3, 4], "truncated-third-block": [1, 2], } for row in observations: require(row["validatedReferenceBlocks"] == expected_blocks[row["id"]], "observed salvage changed: " + row["id"]) require(row["wholeIntegrityExit"] == (0 if row["id"] == "clean" else 2), "whole-file integrity control changed") require(row["validatedBytes"] + sum(b - a for a, b in row["missingByteRanges"]) == len(original), "coverage accounting mismatch") version = invoke([bzip, "--help"], root) recovery_version = invoke([recover], root) return { "schemaVersion": 1, "experiment": "bzip2-salvage", "evidenceType": "controlled synthetic corruption experiment", "license": "CC0-1.0", "generatedAt": datetime.now(timezone.utc).isoformat(), "harnessSha256": sha(Path(__file__).read_bytes()), "environment": { "python": platform.python_version(), "system": platform.system(), "architecture": platform.machine(), "bzip2": (version.stderr + version.stdout).decode().splitlines()[0], "bzip2recover": recovery_version.stderr.decode().splitlines()[0], "bzip2BinarySha256": sha(Path(bzip).read_bytes()), "bzip2recoverBinarySha256": sha(Path(recover).read_bytes()), }, "input": {"bytes": len(original), "sha256": sha(original), "generator": "SHA-256 of ASCII hesela-bzip2-salvage-v1: plus 4-byte big-endian counter 0..9999, concatenated", "compression": "bzip2 -1c", "blockMarkerBitOffsets": markers, "endMarkerBitOffset": ends[0]}, "referenceBlocks": oracle, "observations": observations, "limits": [ "Six chosen fixtures on recorded executables; not random fault sampling or a recovery probability.", "Reference byte intervals are known because the pristine payload is available in this experiment.", "Candidate numbers and compressed bit offsets are not original uncompressed byte offsets.", "CRC validation is not authentication, application validity or a completeness guarantee.", "Only validated candidates are joined; missing intervals remain explicit in the receipt.", "This is not a repair tool and accepts no real archive or device path." ], } def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--output", required=True, type=Path, help="new JSON file; existing files are never overwritten") args = parser.parse_args() # Reserve before running; exclusive creation also rejects existing symlinks. with args.output.open("x", encoding="utf-8") as target: result = run() json.dump(result, target, indent=2) target.write("\n") print("Six fixtures passed: " + str(args.output)) if __name__ == "__main__": main()