#!/usr/bin/env python3 """CC0-1.0. Bounded Parquet fault injection; never edits user-supplied files.""" import argparse import hashlib import inspect import json import platform import struct import tempfile from datetime import datetime, timezone from pathlib import Path import pyarrow as pa import pyarrow.parquet as pq ROWS = 128 GROUP = 64 TARGET = 17 def require(condition, message): if not condition: raise RuntimeError(message) def sha(data): return hashlib.sha256(data).hexdigest() def fixture(path, checksum): schema = pa.schema([pa.field("id", pa.int64(), nullable=False), pa.field("measurement", pa.int64(), nullable=False)]) table = pa.Table.from_arrays( [pa.array(range(ROWS), type=pa.int64()), pa.array(range(100000, 100000 + ROWS), type=pa.int64())], schema=schema) pq.write_table(table, path, row_group_size=GROUP, compression="NONE", use_dictionary=False, write_statistics=True, version="2.6", data_page_version="1.0", write_page_checksum=checksum) require(pq.read_table(path, page_checksum_verification=True).equals(table), "Fresh fixture roundtrip failed") return table def corrupt(path, dest): raw = path.read_bytes() meta = pq.read_metadata(path) require(meta.num_rows == ROWS and meta.num_row_groups == 2, "Wrong fixture geometry") column = meta.row_group(0).column(1) require(column.compression == "UNCOMPRESSED" and not column.has_dictionary_page, "Fault locator requires uncompressed non-dictionary fixture") start = column.data_page_offset end = start + column.total_compressed_size # Locate the full known PLAIN payload only within the API-reported column chunk. pattern = struct.pack("<" + "q" * GROUP, *range(100000, 100000 + GROUP)) segment = raw[start:end] require(segment.count(pattern) == 1, "Fixture payload is not uniquely identifiable") offset = start + segment.index(pattern) + TARGET * 8 changed = bytearray(raw) changed[offset] ^= 1 require(sum((a ^ b).bit_count() for a, b in zip(raw, changed)) == 1, "Fault must change exactly one bit") dest.write_bytes(changed) return {"byteOffset": offset, "column": "measurement", "rowGroup": 0, "rowIndex": TARGET, "xorMask": 1, "beforeValue": 100000 + TARGET, "afterValue": (100000 + TARGET) ^ 1, "fileBytes": len(raw), "originalSha256": sha(raw), "modifiedSha256": sha(changed)} def observe(case, path, expected, *, verify=None, columns=None, group=None, metadata=False): record = {"case": case, "verification": "not-applicable" if metadata else "default" if verify is None else verify, "scope": "metadata" if metadata else ("row-group-1" if group is not None else "id-only" if columns else "all-data")} try: if metadata: value = pq.read_metadata(path) record.update(outcome="metadata-readable", rows=value.num_rows, rowGroups=value.num_row_groups, mismatchCount=None) return record options = {} if verify is None else {"page_checksum_verification": verify} if group is None: actual = pq.read_table(path, columns=columns, **options) else: with pq.ParquetFile(path, **options) as reader: actual = reader.read_row_group(group, columns=columns) wanted = expected.slice(GROUP, GROUP) if group is not None else expected if columns is not None: wanted = wanted.select(columns) require(actual.schema == wanted.schema and actual.num_rows == wanted.num_rows, "Unexpected result shape") diffs = [] for name in wanted.column_names: for index, (a, b) in enumerate(zip(actual[name].to_pylist(), wanted[name].to_pylist())): if a != b: diffs.append({"column": name, "rowWithinResult": index, "expected": b, "actual": a}) record.update(outcome="wrong-values" if diffs else "matches-selected-data", rows=actual.num_rows, mismatchCount=len(diffs), mismatches=diffs) except OSError as error: require("checksum" in str(error).lower(), "Unexpected read failure: " + str(error)) record.update(outcome="checksum-error", rows=None, mismatchCount=None, errorType=type(error).__name__, error=str(error)) return record def run_once(root): checked, unchecked = root / "checked.parquet", root / "unchecked.parquet" bad, bad_unchecked = root / "bad.parquet", root / "bad-unchecked.parquet" table = fixture(checked, True) fixture(unchecked, False) faults = {"checksumPresent": corrupt(checked, bad), "checksumAbsent": corrupt(unchecked, bad_unchecked)} rows = [ observe("clean-checked", checked, table, verify=True), observe("corrupt-default", bad, table), observe("corrupt-check-off", bad, table, verify=False), observe("corrupt-check-on", bad, table, verify=True), observe("projection-check-on", bad, table, verify=True, columns=["id"]), observe("other-group-check-on", bad, table, verify=True, group=1), observe("metadata-only", bad, table, metadata=True), observe("absent-crc-check-on", bad_unchecked, table, verify=True), observe("clean-no-crc-check-on", unchecked, table, verify=True), ] expected = ["matches-selected-data", "wrong-values", "wrong-values", "checksum-error", "matches-selected-data", "matches-selected-data", "metadata-readable", "wrong-values", "matches-selected-data"] require([r["outcome"] for r in rows] == expected, "Behavior changed; review raw results") for row in rows: if row["outcome"] == "wrong-values": require(row["mismatches"] == [{"column": "measurement", "rowWithinResult": TARGET, "expected": 100017, "actual": 100016}], "Unexpected corruption propagation") return {"faults": faults, "observations": rows} def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--repetitions", type=int, default=3) parser.add_argument("--output", type=Path) args = parser.parse_args() require(args.output is None or not (args.output.exists() or args.output.is_symlink()), "Output already exists; choose a new path") require(1 <= args.repetitions <= 20, "Repetitions must be between 1 and 20") require(pa.__version__ == "21.0.0", "Use pinned PyArrow 21.0.0; review other versions separately") require(inspect.signature(pq.read_table).parameters["page_checksum_verification"].default is False, "Reader checksum default changed") require(inspect.signature(pq.write_table).parameters["write_page_checksum"].default is False, "Writer checksum default changed") runs = [] for _ in range(args.repetitions): with tempfile.TemporaryDirectory(prefix="hesela-parquet-") as directory: runs.append(run_once(Path(directory))) require(all(run == runs[0] for run in runs), "Repetitions disagree") source = Path(__file__).read_bytes() requirements = Path(__file__).with_name("parquet-integrity-requirements.txt").read_bytes() result = { "schemaVersion": 1, "experiment": "parquet-integrity-boundaries", "evidenceType": "measured-api-behavior", "generatedAt": datetime.now(timezone.utc).isoformat(), "sourceSha256": sha(source), "requirementsSha256": sha(requirements), "environment": {"python": platform.python_version(), "pyarrow": pa.__version__, "arrowCpp": pa.cpp_version, "kernel": platform.release(), "system": platform.system(), "machine": platform.machine()}, "repetitions": args.repetitions, "casesPerRepetition": 9, "inputs": {"rows": ROWS, "rowGroupRows": GROUP, "columns": ["id", "measurement"], "type": "required INT64", "compression": "NONE", "dictionary": False, "dataPageVersion": "1.0", "fileVersion": "2.6", "synthetic": True}, "method": "Write two deterministic fixtures with/without page CRC, flip one payload bit in copies, compare nine read scopes/settings against complete expected selected values; repeat on fresh temporary files.", "limits": "Controlled synthetic one-bit fault on one host and pinned library; not field corruption frequency, power-loss testing, a storage benchmark, hostile-tampering assessment or exhaustive file validation. No compressed, encrypted, nested or dictionary pages tested. Projection and row-group cases compare only selected data.", "runs": runs} text = json.dumps(result, indent=2) + "\n" if args.output: with args.output.open("x", encoding="utf-8") as output: output.write(text) else: print(text, end="") if __name__ == "__main__": main()