diff --git a/scripts/packverify.py b/scripts/packverify.py index ddff77e8..656b25bf 100644 --- a/scripts/packverify.py +++ b/scripts/packverify.py @@ -23,6 +23,7 @@ import functools import hashlib from common import filter_systems_by_target from hashing import parse_md5_list +import io from common import load_emulator_profiles from common import load_platform_config import os @@ -34,7 +35,6 @@ from packresolve import resolve_file from common import resolve_platform_cores from common import sanitize_pack_path import zipfile -import io from collections.abc import Callable @@ -100,6 +100,25 @@ def _settle_untracked( return "untracked", file_name +def _member_crcs(archive: zipfile.ZipFile) -> dict[str, int]: + return {i.filename: i.CRC for i in archive.infolist() if not i.is_dir()} + + +def _same_members(pack: zipfile.ZipFile, name: str, source: str) -> bool: + """Whether a ZIP inside the pack holds the members of a ZIP on disk. + + A rebuilt archive (TorrentZip) differs byte for byte and is the same + set when every member and its CRC agree. The two checks that ask it + each carried their own copy. + """ + try: + with zipfile.ZipFile(io.BytesIO(pack.read(name))) as packed, \ + zipfile.ZipFile(source) as held: + return _member_crcs(packed) == _member_crcs(held) + except (zipfile.BadZipFile, OSError): + return False + + def verify_pack( zip_path: str, db: dict, data_registry: dict | None = None ) -> tuple[bool, dict]: @@ -193,27 +212,10 @@ def verify_pack( _src_path = files_db[_src_sha1]["path"] if not os.path.exists(_src_path): continue - try: - import io as _io - - with zipfile.ZipFile(_src_path) as _sz: - _sc = { - i.filename: i.CRC - for i in _sz.infolist() - if not i.is_dir() - } - with zipfile.ZipFile(_io.BytesIO(zf.read(name))) as _pz: - _pc = { - i.filename: i.CRC - for i in _pz.infolist() - if not i.is_dir() - } - if _sc == _pc: - status = "verified_rebuild" - file_name = _bn - break - except (zipfile.BadZipFile, OSError): - continue + if _same_members(zf, name, _src_path): + status = "verified_rebuild" + file_name = _bn + break # Data directory: check against cached files if status == "untracked" and _data_index: @@ -239,28 +241,13 @@ def verify_pack( status = "verified_data" file_name = _bn break - if name.endswith(".zip") and _dp.endswith(".zip"): - try: - import io as _io2 - - with zipfile.ZipFile(_io2.BytesIO(zf.read(name))) as _pz2: - _pc2 = { - i.filename: i.CRC - for i in _pz2.infolist() - if not i.is_dir() - } - with zipfile.ZipFile(_dp) as _dz: - _dc = { - i.filename: i.CRC - for i in _dz.infolist() - if not i.is_dir() - } - if _pc2 == _dc: - status = "verified_data" - file_name = _bn - break - except (zipfile.BadZipFile, OSError): - continue + if ( + name.endswith(".zip") and _dp.endswith(".zip") + and _same_members(zf, name, _dp) + ): + status = "verified_data" + file_name = _bn + break status, file_name = _settle_untracked( status, name, zf, by_md5, held_inside, errors, file_name diff --git a/tests/test_pack_member_identity.py b/tests/test_pack_member_identity.py index 4302442e..63d15f47 100644 --- a/tests/test_pack_member_identity.py +++ b/tests/test_pack_member_identity.py @@ -82,5 +82,40 @@ class SchemaAcceptsEveryStatus(unittest.TestCase): self.assertEqual(sorted(written - allowed), []) +class OneComparisonOfArchiveMembers(unittest.TestCase): + """verify_pack compared member CRCs twice, each copy with its own local + import and names (_sc/_pc, _pc2/_dc).""" + + def test_the_helper_answers_both_ways(self): + import io + import tempfile + import zipfile + + sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts")) + from packverify import _same_members # noqa: PLC0415 + + def archive(members: dict) -> bytes: + buffer = io.BytesIO() + with zipfile.ZipFile(buffer, "w") as zf: + for member, data in members.items(): + zf.writestr(member, data) + return buffer.getvalue() + + with tempfile.TemporaryDirectory() as directory: + held = Path(directory) / "held.zip" + held.write_bytes(archive({"a.rom": b"A", "b.rom": b"B"})) + pack = io.BytesIO() + with zipfile.ZipFile(pack, "w") as zf: + zf.writestr("same.zip", archive({"b.rom": b"B", "a.rom": b"A"})) + zf.writestr("other.zip", archive({"a.rom": b"X"})) + zf.writestr("broken.zip", b"not a zip") + with zipfile.ZipFile(pack) as zf: + self.assertTrue(_same_members(zf, "same.zip", str(held))) + self.assertFalse(_same_members(zf, "other.zip", str(held))) + self.assertFalse(_same_members(zf, "broken.zip", str(held))) + source = (Path(__file__).resolve().parents[1] / "scripts" / "packverify.py").read_text() + self.assertNotIn("import io as _io", source) + + if __name__ == "__main__": unittest.main()