From 966c0e5b5960d5427fb608adef282f3514e1e496 Mon Sep 17 00:00:00 2001 From: Abdessamad Derraz <3028866+Abdess@users.noreply.github.com> Date: Thu, 6 Aug 2026 16:41:47 +0200 Subject: [PATCH] feat: add torrentzip builder for romset reconstruction --- scripts/torrentzip.py | 146 +++++++++++++++++++++++++++++++++++++++ tests/test_torrentzip.py | 104 ++++++++++++++++++++++++++++ 2 files changed, 250 insertions(+) create mode 100644 scripts/torrentzip.py create mode 100644 tests/test_torrentzip.py diff --git a/scripts/torrentzip.py b/scripts/torrentzip.py new file mode 100644 index 00000000..a04ee621 --- /dev/null +++ b/scripts/torrentzip.py @@ -0,0 +1,146 @@ +#!/usr/bin/env python3 +"""TorrentZip archive builder for MAME/FBNeo ROM sets. + +TorrentZip is the archive format MAME romset distributions use. It fixes +every non-content variable (member order, timestamps, compression level, +extra fields, attributes), so the archive bytes are a pure function of the +member names and their contents. + +That property makes it possible to rebuild a romset ZIP that matches an +upstream-published hash exactly, and to identify which romset revision a +published hash refers to. + +Usage: + from torrentzip import build_torrentzip, identify_romset + + data = build_torrentzip([("sp-s2.sp1", rom_bytes), ...]) + md5 = hashlib.md5(data).hexdigest() + +CLI: + python scripts/torrentzip.py --check bios/Arcade/Arcade/naomi2.zip + python scripts/torrentzip.py --rebuild in.zip --output out.zip +""" +from __future__ import annotations + +import argparse +import hashlib +import struct +import sys +import zipfile +import zlib +from pathlib import Path + +# TorrentZip constants: 1996-12-24 23:32:00, deflate level 9 +_DOS_DATE = ((1996 - 1980) << 9) | (12 << 5) | 24 +_DOS_TIME = (23 << 11) | (32 << 5) | 0 +_COMMENT_PREFIX = "TORRENTZIPPED-" + + +def build_torrentzip(members: list[tuple[str, bytes]]) -> bytes: + """Build a TorrentZip archive from (name, data) members. + + Members are sorted case-insensitively by name, as the format requires. + Returns the complete archive bytes. + """ + ordered = sorted(members, key=lambda m: m[0].lower()) + body = bytearray() + central = bytearray() + + for name, data in ordered: + raw = name.encode("ascii") + compressor = zlib.compressobj(9, zlib.DEFLATED, -15) + blob = compressor.compress(data) + compressor.flush() + crc = zlib.crc32(data) & 0xFFFFFFFF + offset = len(body) + body += struct.pack( + " bytes: + """Read an archive and return its TorrentZip normalization.""" + with zipfile.ZipFile(source) as zf: + members = [ + (info.filename, zf.read(info.filename)) + for info in zf.infolist() + if not info.is_dir() + ] + return build_torrentzip(members) + + +def is_torrentzip(source: str | Path) -> bool: + """Report whether an archive already carries the TorrentZip signature.""" + with zipfile.ZipFile(source) as zf: + return zf.comment.decode("ascii", "replace").startswith(_COMMENT_PREFIX) + + +def identify_romset( + recipes: dict[str, list[tuple[str, str]]], + atoms: dict[str, bytes], + target_md5: str, +) -> str | None: + """Find which recipe rebuilds to a target archive MD5. + + recipes maps a label (e.g. a MAME version) to a list of + (member name, CRC32 hex) pairs; atoms maps CRC32 hex to ROM bytes. + Returns the matching label, or None when no recipe reproduces the hash. + """ + target = target_md5.lower() + for label, recipe in recipes.items(): + if any(crc.lower() not in atoms for _, crc in recipe): + continue + members = [(name, atoms[crc.lower()]) for name, crc in recipe] + if hashlib.md5(build_torrentzip(members)).hexdigest() == target: + return label + return None + + +def main() -> None: + """Entry point.""" + parser = argparse.ArgumentParser(description="TorrentZip builder and checker") + parser.add_argument("--check", metavar="ZIP", help="report TorrentZip conformance") + parser.add_argument("--rebuild", metavar="ZIP", help="normalize an archive") + parser.add_argument("--output", "-o", help="output path for --rebuild") + args = parser.parse_args() + + if args.check: + data = Path(args.check).read_bytes() + rebuilt = rebuild_torrentzip(args.check) + print(f"{args.check}") + print(f" signature: {is_torrentzip(args.check)}") + print(f" md5: {hashlib.md5(data).hexdigest()}") + print(f" rebuilt: {hashlib.md5(rebuilt).hexdigest()}") + print(f" conform: {data == rebuilt}") + return + + if args.rebuild: + data = rebuild_torrentzip(args.rebuild) + dest = args.output or args.rebuild + Path(dest).write_bytes(data) + print(f"{dest}: {len(data)} bytes, md5 {hashlib.md5(data).hexdigest()}") + return + + parser.error("nothing to do: pass --check or --rebuild") + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/test_torrentzip.py b/tests/test_torrentzip.py new file mode 100644 index 00000000..b8682a1a --- /dev/null +++ b/tests/test_torrentzip.py @@ -0,0 +1,104 @@ +"""Tests for the TorrentZip builder against real MAME romsets.""" +from __future__ import annotations + +import hashlib +import os +import sys +import tempfile +import unittest +import zipfile + +REPO_ROOT = os.path.join(os.path.dirname(__file__), "..") +sys.path.insert(0, os.path.join(REPO_ROOT, "scripts")) + +from torrentzip import ( # noqa: E402 + build_torrentzip, + identify_romset, + is_torrentzip, + rebuild_torrentzip, +) + + +class TorrentZipTest(unittest.TestCase): + """The format is deterministic: same members always give the same bytes.""" + + def test_deterministic_output(self): + members = [("b.rom", b"second"), ("a.rom", b"first")] + self.assertEqual(build_torrentzip(members), build_torrentzip(members)) + + def test_member_order_is_irrelevant(self): + a = build_torrentzip([("a.rom", b"one"), ("b.rom", b"two")]) + b = build_torrentzip([("b.rom", b"two"), ("a.rom", b"one")]) + self.assertEqual(a, b) + + def test_case_insensitive_sorting(self): + data = build_torrentzip([("B.rom", b"x"), ("a.rom", b"y")]) + with zipfile.ZipFile(_as_file(self, data)) as zf: + self.assertEqual([i.filename for i in zf.infolist()], ["a.rom", "B.rom"]) + + def test_fixed_timestamp_and_signature(self): + data = build_torrentzip([("a.rom", b"payload")]) + with zipfile.ZipFile(_as_file(self, data)) as zf: + self.assertEqual(zf.infolist()[0].date_time, (1996, 12, 24, 23, 32, 0)) + self.assertTrue(zf.comment.decode().startswith("TORRENTZIPPED-")) + + def test_roundtrip_readable_content(self): + data = build_torrentzip([("a.rom", b"hello"), ("b.rom", b"world")]) + with zipfile.ZipFile(_as_file(self, data)) as zf: + self.assertEqual(zf.read("a.rom"), b"hello") + self.assertEqual(zf.read("b.rom"), b"world") + + def test_rebuild_matches_real_mame_set(self): + """Rebuilding a shipped MAME romset reproduces it byte for byte.""" + candidates = [ + os.path.join(REPO_ROOT, "bios", "Arcade", "Arcade", "naomi2.zip"), + os.path.join(REPO_ROOT, "bios", "Arcade", "MAME", "naomi2.zip"), + ] + checked = 0 + for path in candidates: + if not os.path.exists(path) or not is_torrentzip(path): + continue + with open(path, "rb") as fh: + original = fh.read() + self.assertEqual(rebuild_torrentzip(path), original, path) + checked += 1 + if not checked: + self.skipTest("no TorrentZip romset available") + + def test_identify_romset_selects_matching_recipe(self): + atoms = {"3610a686": b"alpha", "9d0d1b46": b"beta"} + import zlib + + atoms = { + f"{zlib.crc32(b'alpha') & 0xffffffff:08x}": b"alpha", + f"{zlib.crc32(b'beta') & 0xffffffff:08x}": b"beta", + } + crc_a, crc_b = list(atoms) + wanted = hashlib.md5( + build_torrentzip([("a.rom", b"alpha"), ("b.rom", b"beta")]) + ).hexdigest() + recipes = { + "v1": [("a.rom", crc_a)], + "v2": [("a.rom", crc_a), ("b.rom", crc_b)], + } + self.assertEqual(identify_romset(recipes, atoms, wanted), "v2") + self.assertIsNone(identify_romset(recipes, atoms, "0" * 32)) + + def test_identify_romset_skips_recipes_with_absent_atoms(self): + atoms = {"aaaaaaaa": b"x"} + recipes = {"v1": [("a.rom", "ffffffff")]} + self.assertIsNone(identify_romset(recipes, atoms, "0" * 32)) + + +def _as_file(test: unittest.TestCase, data: bytes) -> str: + """Write bytes to a temp file cleaned up with the test.""" + fd, path = tempfile.mkstemp(suffix=".zip") + os.close(fd) + with open(path, "wb") as fh: + fh.write(data) + test.addCleanup(os.unlink, path) + return path + + +if __name__ == "__main__": + unittest.main()