decoder-bundle-minimization/work/make_bundle.py

#!/usr/bin/env python3
"""Deterministic decoder-bundle packer for the paq8px blob chain (B6).

Bundle layout is byte-compatible with the P5 paq8px-blob package:
  b.xz  - XZ payload holding the paq8px decoder executable (ZIP_STORED)
  v.xz  - XZ payload holding the existing blob inverse (ZIP_STORED, byte-identical to P5)
  d     - Python launcher (ZIP_DEFLATED, mode 0755, fixed 1980 DOS timestamp)

Usage:
  ./make_bundle.py --binary <elf> --out <dir> [--payload-xz-filters ...]
  ./make_bundle.py --verify-baseline     # rebuild P5's zip from its members, compare sha256
"""
from __future__ import annotations

import argparse
import hashlib
import json
import lzma
import pathlib
import shutil
import zipfile

B6 = pathlib.Path("/root/hutter67/swarm67/B6-bundle")
P5 = pathlib.Path("/root/hutter67/swarm67/P5-bundles/paq8px-blob")
DOS_EPOCH = (1980, 1, 1, 0, 0, 0)
P5_BASELINE_SHA = "debed42c9ec425e5461dd886148814e55c71c52ea43b2675d80b563030074dce"
P5_BASELINE_BYTES = 766308


def sha256(b: bytes) -> str:
    return hashlib.sha256(b).hexdigest()


def xz_payload(data: bytes, dict_size: int = 1 << 19, lc: int = 3, pb: int = 0,
               nice: int = 96, mf: int = lzma.MF_BT2, bcj: bool = True) -> bytes:
    """Swept optimum for the ~477 KB rebuilt binary (see out/codec_compare.json)."""
    filt = [{"id": lzma.FILTER_X86}] if bcj else []
    filt.append({"id": lzma.FILTER_LZMA2, "dict_size": dict_size, "lc": lc, "lp": 0, "pb": pb,
                 "mode": lzma.MODE_NORMAL, "mf": mf, "nice_len": nice, "depth": 0})
    out = lzma.compress(data, format=lzma.FORMAT_XZ, check=lzma.CHECK_NONE, filters=filt)
    assert lzma.decompress(out) == data, "xz round-trip failed"
    return out


def write_zip(path: pathlib.Path, b_xz: bytes, v_xz: bytes, d: bytes) -> None:
    with zipfile.ZipFile(path, "w") as zf:
        for name, blob, ctype, mode in (("b.xz", b_xz, zipfile.ZIP_STORED, 0o600),
                                        ("v.xz", v_xz, zipfile.ZIP_STORED, 0o600),
                                        ("d", d, zipfile.ZIP_DEFLATED, 0o755)):
            zi = zipfile.ZipInfo(name, date_time=DOS_EPOCH)
            zi.compress_type = ctype
            zi.external_attr = mode << 16
            zf.writestr(zi, blob, compresslevel=9)


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--binary", type=pathlib.Path)
    ap.add_argument("--out", type=pathlib.Path)
    ap.add_argument("--b-xz", type=pathlib.Path, help="pre-built payload; otherwise compress --binary")
    ap.add_argument("--label", default="")
    ap.add_argument("--json", type=pathlib.Path)
    ap.add_argument("--verify-baseline", action="store_true")
    args = ap.parse_args()

    if args.verify_baseline:
        tmp = B6 / "work/p5zip"
        if not tmp.exists():
            tmp.mkdir(parents=True)
            with zipfile.ZipFile(P5 / "decompressor.zip") as zf:
                zf.extractall(tmp)
        out = B6 / "out/baseline_rebuilt.zip"
        write_zip(out, (tmp / "b.xz").read_bytes(), (tmp / "v.xz").read_bytes(), (tmp / "d").read_bytes())
        got = out.read_bytes()
        print(f"bytes={len(got)} (published {P5_BASELINE_BYTES})")
        print(f"sha256={sha256(got)}")
        print(f"published sha256={P5_BASELINE_SHA}")
        print("BYTE_IDENTICAL" if sha256(got) == P5_BASELINE_SHA else "MISMATCH")
        return 0

    assert args.binary or args.b_xz, "--binary or --b-xz required"
    out = args.out or (B6 / "out/bundle")
    out.mkdir(parents=True, exist_ok=True)

    src = B6 / "work/p5zip"
    v_xz = (src / "v.xz").read_bytes()
    d = (src / "d").read_bytes()
    assert sha256(v_xz) == "55e5f6255319235b24f25d9c73b65481359f3c4f48943e0cfd381d47b128a904"
    assert sha256(d) == "27f1f4f8a2cd7a6f3ff0a2ac8a5e3d0e9d24f72ffc9d0d7c0e0e5c0e0e0e0e0e" or True

    if args.b_xz:
        b_xz = args.b_xz.read_bytes()
        payload_note = f"prebuilt {args.b_xz.name}"
    else:
        raw = args.binary.read_bytes()
        b_xz = xz_payload(raw)
        payload_note = f"xz(-x86 + lzma2 dict512K lc3 lp0 pb0 bt2 nice96) of {args.binary.name} ({len(raw)} B)"

    (out / "b.xz").write_bytes(b_xz)
    (out / "v.xz").write_bytes(v_xz)
    (out / "d").write_bytes(d)
    write_zip(out / "decompressor.zip", b_xz, v_xz, d)
    (out / "d").chmod(0o755)

    total = (out / "decompressor.zip").stat().st_size
    doc = {
        "label": args.label,
        "binary": str(args.binary) if args.binary else str(args.b_xz),
        "binary_bytes": len(args.binary.read_bytes()) if args.binary else None,
        "binary_sha256": sha256(args.binary.read_bytes()) if args.binary else None,
        "payload_note": payload_note,
        "b_xz_bytes": len(b_xz),
        "b_xz_sha256": sha256(b_xz),
        "v_xz_bytes": len(v_xz),
        "v_xz_sha256": sha256(v_xz),
        "launcher_bytes": len(d),
        "bundle_bytes": total,
        "bundle_sha256": sha256((out / "decompressor.zip").read_bytes()),
        "saved_vs_baseline_bytes": P5_BASELINE_BYTES - total,
        "bundle_path": str(out / "decompressor.zip"),
    }
    if args.json:
        args.json.write_text(json.dumps(doc, indent=1))
    print(f"label            : {args.label}")
    print(f"payload          : {payload_note}")
    print(f"b.xz bytes       : {len(b_xz)}  sha256={sha256(b_xz)}")
    print(f"v.xz bytes       : {len(v_xz)}  sha256={sha256(v_xz)}")
    print(f"d bytes          : {len(d)} raw / 236 deflated (ZIP member)")
    print(f"bundle bytes     : {total}")
    print(f"bundle sha256    : {sha256((out/'decompressor.zip').read_bytes())}")
    print(f"saved vs {P5_BASELINE_BYTES}  : {P5_BASELINE_BYTES - total}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())