#!/usr/bin/env python3 """Synthetic Zstandard formatted-dictionary identity experiment; no user files.""" import argparse from datetime import datetime, timezone import hashlib import json import os from pathlib import Path import platform import re import shutil import subprocess import tempfile def require(condition, message): if not condition: raise ValueError(message) def digest(data): return hashlib.sha256(data).hexdigest() def run(args, data=b""): env = {k: v for k, v in os.environ.items() if not k.startswith("ZSTD")} env["LC_ALL"] = "C" return subprocess.run(["zstd", *args], input=data, capture_output=True, timeout=30, check=False, env=env) def frame_id(frame): # Parse only the dictionary-ID field, not the complete frame. require(len(frame) >= 5 and frame[:4] == bytes.fromhex("28b52ffd"), "Missing Zstandard frame header") descriptor = frame[4] require(descriptor & 0x18 == 0, "Unsupported reserved/unused descriptor bits") size = (0, 1, 2, 4)[descriptor & 3] offset = 5 + (0 if descriptor & 0x20 else 1) require(len(frame) >= offset + size, "Truncated dictionary-ID field") return int.from_bytes(frame[offset:offset + size], "little") EXPECTED = { "correct-checked": (True, True), "missing-dictionary": (False, False), "different-id-same-content": (False, False), "same-id-changed-content-checked": (False, None), "same-id-changed-content-unchecked": (True, False), "same-id-changed-content-check-ignored": (True, False), "correct-unchecked": (True, True), "omitted-id-same-content": (True, True), } def validate(rows): require(len(rows) == len(EXPECTED) and {r["id"] for r in rows} == set(EXPECTED), "Missing, duplicate or unexpected case") for row in rows: for key in ("exitCode", "outputBytes", "expectedBytes", "frameDictionaryId"): require(type(row[key]) is int and row[key] >= 0, "Invalid " + key) for key in ("bytesMatch", "accepted", "checksumPresent", "verifyChecksumIfPresent"): require(type(row[key]) is bool, "Invalid " + key) for key in ("outputSha256", "expectedSha256", "compressedSha256"): require(isinstance(row[key], str) and re.fullmatch("[0-9a-f]{64}", row[key]), "Invalid " + key) success, match = EXPECTED[row["id"]] require((row["exitCode"] == 0) == success, "Unexpected process status") if match is not None: require(row["bytesMatch"] == match, "Unexpected byte comparison") require(row["bytesMatch"] == (row["outputBytes"] == row["expectedBytes"] and row["outputSha256"] == row["expectedSha256"]), "Inconsistent oracle") require(row["accepted"] == (success and row["bytesMatch"]), "Invalid acceptance") require(row["frameDictionaryId"] == (0 if row["id"] == "omitted-id-same-content" else 123456), "Unexpected frame ID") expected_id = (None if row["id"] == "missing-dictionary" else 123457 if row["id"] in {"different-id-same-content", "omitted-id-same-content"} else 123456) require(row["suppliedDictionaryId"] == expected_id, "Unexpected supplied ID") require(row["expectedBytes"] == 256, "Unexpected payload length") if expected_id is None: require(row["dictionarySha256"] is None, "Unexpected dictionary digest") else: require(isinstance(row["dictionarySha256"], str) and re.fullmatch("[0-9a-f]{64}", row["dictionarySha256"]), "Invalid dictionary digest") by_id = {row["id"]: row for row in rows} for key in ("compressedSha256", "expectedSha256", "expectedBytes"): require(by_id["correct-checked"][key] == by_id["same-id-changed-content-checked"][key] == by_id["same-id-changed-content-check-ignored"][key], "Checked comparison inputs differ") require(len({r["expectedSha256"] for r in rows}) == 1, "Expected payload changed") for row in rows: no_checksum = row["id"] in {"different-id-same-content", "same-id-changed-content-unchecked", "correct-unchecked"} require(row["checksumPresent"] == (not no_checksum), "Changed stored checksum") require(row["verifyChecksumIfPresent"] == ( row["id"] != "same-id-changed-content-check-ignored"), "Changed reader policy") def experiment(): require(shutil.which("zstd") is not None, "zstd CLI is required") with tempfile.TemporaryDirectory(prefix="hesela-zstd-id-") as directory: root = Path(directory) samples = [] sample_hash = hashlib.sha256() for i in range(128): data = b"".join( ("record=%04d key=%02d value=" % (i, j)).encode() + hashlib.sha256(("group-%d-%d" % (i % 8, j)).encode()).hexdigest().encode() + b"\n" for j in range(8)) path = root / ("sample-%03d" % i) path.write_bytes(data) samples.append(str(path)) sample_hash.update(len(data).to_bytes(8, "little")) sample_hash.update(data) good = root / "correct.dict" training = ["-q", "--train-cover=k=50,d=8", "-T1", "--maxdict=2048", "--dictID=123456", "-o", str(good), *samples] trained = run(training) require(trained.returncode == 0, trained.stderr.decode("utf-8", "replace")) dictionary = good.read_bytes() require(dictionary[:4] == bytes.fromhex("37a430ec") and int.from_bytes(dictionary[4:8], "little") == 123456 and len(dictionary) == 2048, "Unexpected trained dictionary") changed = dictionary[:-1] + bytes([dictionary[-1] ^ 1]) renamed = dictionary[:4] + (123457).to_bytes(4, "little") + dictionary[8:] wrong, alias = root / "changed.dict", root / "renamed.dict" wrong.write_bytes(changed) alias.write_bytes(renamed) payload = dictionary[-256:] frames = {} for name, check, omit in [("checked", True, False), ("unchecked", False, False), ("omitted", True, True)]: args = ["-q", "-c", "-3", "--single-thread", "-D", str(good), "--check" if check else "--no-check"] if omit: args.append("--no-dictID") encoded = run(args, payload) require(encoded.returncode == 0, "Compression failed") frames[name] = encoded.stdout cases = [ ("correct-checked", "checked", good, True), ("missing-dictionary", "checked", None, True), ("different-id-same-content", "unchecked", alias, True), ("same-id-changed-content-checked", "checked", wrong, True), ("same-id-changed-content-unchecked", "unchecked", wrong, True), ("same-id-changed-content-check-ignored", "checked", wrong, False), ("correct-unchecked", "unchecked", good, True), ("omitted-id-same-content", "omitted", alias, True), ] rows = [] for case_id, frame_key, selected, verify in cases: frame = frames[frame_key] args = ["-q", "-d", "-c", "--check" if verify else "--no-check"] if selected is not None: args += ["-D", str(selected)] decoded = run(args, frame) exact = decoded.stdout == payload rows.append({ "id": case_id, "frameDictionaryId": frame_id(frame), "suppliedDictionaryId": (int.from_bytes(selected.read_bytes()[4:8], "little") if selected is not None else None), "dictionarySha256": digest(selected.read_bytes()) if selected else None, "checksumPresent": bool(frame[4] & 4), "verifyChecksumIfPresent": verify, "compressedSha256": digest(frame), "exitCode": decoded.returncode, "outputBytes": len(decoded.stdout), "outputSha256": digest(decoded.stdout), "expectedBytes": len(payload), "expectedSha256": digest(payload), "bytesMatch": exact, "accepted": decoded.returncode == 0 and exact, "stderr": decoded.stderr.decode("utf-8", "replace").strip(), }) validate(rows) return { "schemaVersion": 1, "generatedAt": datetime.now(timezone.utc).isoformat(), "sourceSha256": digest(Path(__file__).read_bytes()), "environment": {"zstd": run(["--version"]).stdout.decode().strip(), "python": platform.python_version(), "system": platform.system(), "machine": platform.machine()}, "method": { "type": "synthetic-formatted-dictionary-identity-demonstration", "training": "128 deterministic 8-record samples; --train-cover=k=50,d=8 -T1 --maxdict=2048 --dictID=123456", "trainingCorpusSha256": sample_hash.hexdigest(), "trainingCorpusHashEncoding": "ordered concatenation of 8-byte little-endian sample length then sample bytes", "dictionaryBytes": len(dictionary), "dictionarySha256": digest(dictionary), "changedDictionarySha256": digest(changed), "renamedDictionarySha256": digest(renamed), "contentMutation": "XOR final dictionary byte with 1, retaining ID and all other bytes", "idMutation": "replace 4-byte little-endian ID 123456 with 123457 only", "payload": "final 256 bytes of original trained dictionary", "scope": "Deliberate dependency misuse, not a spontaneous collision, codec defect, benchmark or failure-rate estimate", }, "observations": rows, } if __name__ == "__main__": parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--output", type=Path) options = parser.parse_args() output = json.dumps(experiment(), indent=2) + "\n" if options.output: options.output.write_text(output, encoding="utf-8") else: print(output, end="")