mirror of
https://github.com/Abdess/retroarch_system.git
synced 2026-10-11 05:53:23 -05:00
Abdessamad Derraz
24e9b820ae
Two builds of the same pack from the same inputs produced different archives. Of 67 members, 65 were already identical: only README.txt and manifest.json differed, both stamped with the wall clock by writestr and the second carrying a generated timestamp. Generated members now use the epoch the archive rebuilder already applies, and the timestamp comes from the database snapshot the pack was built from, so the same data yields the same bytes. Install manifests skipped archived platforms, which is why RetroPie had none; archived means upstream is no longer scraped, not that the packs stopped shipping. A target-filtered manifest also had no record of its filter beyond the filename, so it carries one the way a region-filtered manifest already does.
254 lines
8.9 KiB
Python
254 lines
8.9 KiB
Python
"""Tests for deterministic ZIP rebuilding."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import tempfile
|
|
import time
|
|
import unittest
|
|
import zipfile
|
|
from pathlib import Path
|
|
|
|
from scripts.deterministic_zip import (
|
|
_FIXED_DATE_TIME,
|
|
rebuild_zip_deterministic,
|
|
verify_zip_determinism,
|
|
)
|
|
|
|
CONTENT = {
|
|
"03.wav": b"\x00\x01\x02" * 500,
|
|
"01.wav": b"sample one",
|
|
"sub/02.wav": b"sample two",
|
|
}
|
|
|
|
|
|
def _make_source(path: Path, date_time=(2021, 6, 5, 4, 3, 2), compression=None) -> None:
|
|
"""Write a ZIP with deliberately non-deterministic metadata."""
|
|
comp = zipfile.ZIP_STORED if compression is None else compression
|
|
with zipfile.ZipFile(path, "w", comp) as zf:
|
|
for name in ("sub/02.wav", "03.wav", "01.wav"): # unsorted on purpose
|
|
info = zipfile.ZipInfo(filename=name, date_time=date_time)
|
|
info.compress_type = comp
|
|
info.external_attr = 0o100777 << 16
|
|
info.create_system = 3 # Unix
|
|
zf.writestr(info, CONTENT[name])
|
|
|
|
|
|
class TestRebuild(unittest.TestCase):
|
|
def setUp(self):
|
|
self._tmp = tempfile.TemporaryDirectory()
|
|
self.tmp = Path(self._tmp.name)
|
|
self.src = self.tmp / "src.zip"
|
|
_make_source(self.src)
|
|
|
|
def tearDown(self):
|
|
self._tmp.cleanup()
|
|
|
|
def _rebuild(self, out_name="out.zip", src=None):
|
|
out = self.tmp / out_name
|
|
sha1 = rebuild_zip_deterministic(src or self.src, out)
|
|
return out, sha1
|
|
|
|
def test_content_is_preserved(self):
|
|
out, _ = self._rebuild()
|
|
with zipfile.ZipFile(out) as zf:
|
|
for name, data in CONTENT.items():
|
|
self.assertEqual(zf.read(name), data)
|
|
|
|
def test_entries_sorted_by_name(self):
|
|
out, _ = self._rebuild()
|
|
with zipfile.ZipFile(out) as zf:
|
|
self.assertEqual(zf.namelist(), ["01.wav", "03.wav", "sub/02.wav"])
|
|
|
|
def test_metadata_is_normalized(self):
|
|
out, _ = self._rebuild()
|
|
with zipfile.ZipFile(out) as zf:
|
|
for info in zf.infolist():
|
|
self.assertEqual(info.date_time, _FIXED_DATE_TIME)
|
|
self.assertEqual(info.create_system, 0)
|
|
self.assertEqual(info.external_attr, 0o100644 << 16)
|
|
self.assertEqual(info.compress_type, zipfile.ZIP_DEFLATED)
|
|
|
|
def test_same_input_gives_same_hash(self):
|
|
_, first = self._rebuild("a.zip")
|
|
_, second = self._rebuild("b.zip")
|
|
self.assertEqual(first, second)
|
|
|
|
def test_source_metadata_does_not_affect_hash(self):
|
|
"""Different timestamps and compression, identical rebuild."""
|
|
other = self.tmp / "other.zip"
|
|
_make_source(other, date_time=(1999, 1, 1, 1, 1, 2), compression=zipfile.ZIP_DEFLATED)
|
|
_, from_first = self._rebuild("a.zip")
|
|
_, from_other = self._rebuild("b.zip", src=other)
|
|
self.assertEqual(from_first, from_other)
|
|
|
|
def test_returns_sha1_of_output(self):
|
|
import hashlib
|
|
|
|
out, sha1 = self._rebuild()
|
|
self.assertEqual(sha1, hashlib.sha1(out.read_bytes()).hexdigest())
|
|
|
|
def test_directory_entries_are_dropped(self):
|
|
src = self.tmp / "withdir.zip"
|
|
with zipfile.ZipFile(src, "w") as zf:
|
|
zf.writestr(zipfile.ZipInfo("adir/"), b"")
|
|
zf.writestr("adir/file.bin", b"x")
|
|
out, _ = self._rebuild("nodir.zip", src=src)
|
|
with zipfile.ZipFile(out) as zf:
|
|
self.assertEqual(zf.namelist(), ["adir/file.bin"])
|
|
|
|
def test_corrupt_entry_raises(self):
|
|
"""A tampered payload must fail the source CRC, not copy through."""
|
|
raw = bytearray(self.src.read_bytes())
|
|
marker = CONTENT["01.wav"]
|
|
idx = raw.find(marker)
|
|
self.assertGreater(idx, 0)
|
|
raw[idx : idx + len(marker)] = b"tampered!!"
|
|
corrupt = self.tmp / "corrupt.zip"
|
|
corrupt.write_bytes(bytes(raw))
|
|
with self.assertRaises(zipfile.BadZipFile):
|
|
rebuild_zip_deterministic(corrupt, self.tmp / "never.zip")
|
|
|
|
def test_empty_zip(self):
|
|
src = self.tmp / "empty.zip"
|
|
with zipfile.ZipFile(src, "w"):
|
|
pass
|
|
out, sha1 = self._rebuild("emptyout.zip", src=src)
|
|
with zipfile.ZipFile(out) as zf:
|
|
self.assertEqual(zf.namelist(), [])
|
|
self.assertTrue(sha1)
|
|
|
|
def test_larger_than_copy_chunk(self):
|
|
"""Streaming path must handle payloads bigger than one chunk."""
|
|
src = self.tmp / "big.zip"
|
|
payload = bytes(range(256)) * 8192 # 2 MiB, two chunks
|
|
with zipfile.ZipFile(src, "w") as zf:
|
|
zf.writestr("big.bin", payload)
|
|
out, _ = self._rebuild("bigout.zip", src=src)
|
|
with zipfile.ZipFile(out) as zf:
|
|
self.assertEqual(zf.read("big.bin"), payload)
|
|
|
|
|
|
class TestVerifyDeterminism(unittest.TestCase):
|
|
def setUp(self):
|
|
self._tmp = tempfile.TemporaryDirectory()
|
|
self.tmp = Path(self._tmp.name)
|
|
|
|
def tearDown(self):
|
|
self._tmp.cleanup()
|
|
|
|
def test_non_deterministic_source_reported(self):
|
|
src = self.tmp / "src.zip"
|
|
_make_source(src)
|
|
ok, original, rebuilt = verify_zip_determinism(src)
|
|
self.assertFalse(ok)
|
|
self.assertNotEqual(original, rebuilt)
|
|
|
|
def test_already_deterministic_source_reported(self):
|
|
src = self.tmp / "src.zip"
|
|
_make_source(src)
|
|
normalized = self.tmp / "norm.zip"
|
|
rebuild_zip_deterministic(src, normalized)
|
|
ok, original, rebuilt = verify_zip_determinism(normalized)
|
|
self.assertTrue(ok)
|
|
self.assertEqual(original, rebuilt)
|
|
|
|
|
|
class TestPackDeterminism(unittest.TestCase):
|
|
"""A pack is a function of its inputs, not of the clock it was built on."""
|
|
|
|
def setUp(self):
|
|
self._tmp = tempfile.TemporaryDirectory()
|
|
self.tmp = Path(self._tmp.name)
|
|
self.bios = self.tmp / "bios"
|
|
self.platforms = self.tmp / "platforms"
|
|
self.bios.mkdir()
|
|
self.platforms.mkdir()
|
|
|
|
payload = b"DETERMINISM_PAYLOAD"
|
|
(self.bios / "boot.rom").write_bytes(payload)
|
|
sha1 = hashlib.sha1(payload).hexdigest()
|
|
self.db = {
|
|
"generated_at": "2026-01-02T03:04:05Z",
|
|
"files": {
|
|
sha1: {
|
|
"path": str(self.bios / "boot.rom"),
|
|
"name": "boot.rom",
|
|
"size": len(payload),
|
|
"sha1": sha1,
|
|
"md5": hashlib.md5(payload).hexdigest(),
|
|
},
|
|
},
|
|
"indexes": {
|
|
"by_md5": {hashlib.md5(payload).hexdigest(): sha1},
|
|
"by_name": {"boot.rom": [sha1]},
|
|
"by_crc32": {},
|
|
"by_sha256": {},
|
|
"by_path_suffix": {},
|
|
},
|
|
}
|
|
(self.platforms / "detplat.yml").write_text(
|
|
"platform: DetPlat\n"
|
|
"verification_mode: existence\n"
|
|
"base_destination: system\n"
|
|
"systems:\n"
|
|
" test-sys:\n"
|
|
" files:\n"
|
|
" - name: boot.rom\n"
|
|
" destination: boot.rom\n"
|
|
" required: true\n"
|
|
)
|
|
|
|
def tearDown(self):
|
|
self._tmp.cleanup()
|
|
|
|
def _build(self, out_name: str) -> Path:
|
|
from scripts.generate_pack import generate_pack
|
|
|
|
out = self.tmp / out_name
|
|
out.mkdir()
|
|
path = generate_pack(
|
|
"detplat", str(self.platforms), self.db, str(self.bios), str(out),
|
|
offline=True,
|
|
)
|
|
self.assertIsNotNone(path)
|
|
return Path(path)
|
|
|
|
def test_two_builds_are_byte_identical(self):
|
|
first = self._build("out1")
|
|
time.sleep(2.1) # ZIP mtimes are 2s-granular: cross a whole tick
|
|
second = self._build("out2")
|
|
self.assertEqual(
|
|
hashlib.sha256(first.read_bytes()).hexdigest(),
|
|
hashlib.sha256(second.read_bytes()).hexdigest(),
|
|
)
|
|
|
|
def test_generated_members_carry_the_fixed_epoch(self):
|
|
pack = self._build("out1")
|
|
with zipfile.ZipFile(pack) as zf:
|
|
generated = [
|
|
i for i in zf.infolist()
|
|
if i.filename.rsplit("/", 1)[-1].startswith(
|
|
("README.txt", "INSTRUCTIONS_", "RENAMED_")
|
|
)
|
|
]
|
|
self.assertTrue(generated, "pack should carry a generated README")
|
|
for info in generated:
|
|
self.assertEqual(info.date_time, _FIXED_DATE_TIME)
|
|
|
|
def test_injected_manifest_is_pinned_to_the_database(self):
|
|
from scripts.generate_pack import inject_manifest, verify_pack
|
|
|
|
pack = self._build("out1")
|
|
ok, manifest = verify_pack(str(pack), self.db)
|
|
self.assertTrue(ok, manifest.get("errors"))
|
|
self.assertEqual(manifest["generated"], self.db["generated_at"])
|
|
inject_manifest(str(pack), manifest)
|
|
with zipfile.ZipFile(pack) as zf:
|
|
info = zf.getinfo("manifest.json")
|
|
self.assertEqual(info.date_time, _FIXED_DATE_TIME)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|