From 49a4b181bf6a4999493d1e99be1b3df07a53d39e Mon Sep 17 00:00:00 2001 From: Abdessamad Derraz <3028866+Abdess@users.noreply.github.com> Date: Tue, 6 Oct 2026 06:57:01 +0200 Subject: [PATCH] refactor: one mismatch rule for builder and verifier --- scripts/generate_pack.py | 48 +++------------------ scripts/packverify.py | 91 ++++++++++++++++++++++------------------ scripts/validation.py | 40 ++++++++++++++++++ 3 files changed, 96 insertions(+), 83 deletions(-) diff --git a/scripts/generate_pack.py b/scripts/generate_pack.py index 6f9f60fa..a9c7adcc 100644 --- a/scripts/generate_pack.py +++ b/scripts/generate_pack.py @@ -35,7 +35,6 @@ from common import ( ArtifactLockBusy, build_target_cores_cache, build_zip_contents_index, - check_inside_zip, compute_hashes, expand_directory_entries, expand_platform_declared_names, @@ -58,7 +57,6 @@ from common import ( resolve_local_file, sanitize_pack_path, select_emulator_profiles, - size_fits, yaml_load, ) import packresolve @@ -75,6 +73,8 @@ from nativemode import ( reads_file_contents, ) from validation import ( + inner_rom_check, + settle_mismatch, agnostic_substitute, frontend_digest_matches, _build_validation_index, @@ -287,24 +287,6 @@ def download_external(file_entry: dict, dest_path: str) -> bool: PACK_DOCUMENTS = ("README.txt", "manifest.json") -def _inner_rom_check(file_entry: dict, local_path: str) -> str: - """How an archive answers an entry that pins a ROM inside it. - - Batocera, RetroBat and ROCKNIX hash a member of the ZIP, never the ZIP: - the resolver hands back the archive as a mismatch and this decides it. - Returns check_inside_zip's answer for the first accepted MD5 that - matches, else for the last one tried. The pack and the install manifest - both read it, so a file one of them ships is a file the other lists. - """ - declared = [m.strip() for m in file_entry.get("md5", "").split(",") if m.strip()] - result = "not_in_zip" - for candidate in declared or [""]: - result = check_inside_zip(local_path, file_entry["zipped_file"], candidate) - if result == "ok": - break - return result - - def _data_directory_members( pack_systems: dict, data_registry: dict | None, @@ -494,7 +476,7 @@ def _preferred_entries( def accepted(path: str | None) -> int: if not path or not path.endswith(".zip"): return -1 - return sum(_inner_rom_check(fe, path) == "ok" for fe in members) + return sum(inner_rom_check(fe, path) == "ok" for fe in members) fe, path, _st = max(resolved, key=lambda item: accepted(item[1])) if accepted(path) > 0: @@ -874,7 +856,7 @@ def generate_pack( ): zf_name = file_entry.get("zipped_file") if zf_name and local_path: - last_result = _inner_rom_check(file_entry, local_path) + last_result = inner_rom_check(file_entry, local_path) zip_ok = last_result == "ok" if zip_ok: status = "zip_exact" @@ -2973,26 +2955,6 @@ def _manifest_core_entries( return total_size -def _settle_mismatch( - file_entry: dict, local_path: str | None, status: str, verification_mode: str -) -> str: - """Upgrade a hash mismatch the frontend itself would accept. - - Batocera pins the md5 of a ROM inside the archive (zipped_file), and a - frontend that hashes one digest accepts a file whose other declared - hashes disagree. - """ - if status != "hash_mismatch" or not local_path: - return status - if file_entry.get("zipped_file"): - return "zip_exact" if _inner_rom_check(file_entry, local_path) == "ok" else status - if hash_mismatch_excludes_file(verification_mode) and frontend_digest_matches( - file_entry, local_path, digest_algorithm(verification_mode) - ): - return "frontend_digest_exact" - return status - - def _manifest_region_drops( config: dict, pack_systems: dict, @@ -3191,7 +3153,7 @@ def generate_manifest( if os.path.basename(local_path) != file_entry.get("name", ""): # The pack explains the rename in a note of its own. rename_notes.add(file_entry.get("name", "")) - status = _settle_mismatch(file_entry, local_path, status, verification_mode) + status = settle_mismatch(file_entry, local_path, status, verification_mode) # An existence platform never reads the bytes, so a declared # hash the local dump contradicts is not a reason to withhold # the file. Hash platforms would reject it, so they omit it. diff --git a/scripts/packverify.py b/scripts/packverify.py index f00eefe6..4a18bef7 100644 --- a/scripts/packverify.py +++ b/scripts/packverify.py @@ -17,7 +17,7 @@ from packpaths import _register_path from ziptools import build_zip_contents_index from ziptools import check_inside_zip from nativemode import digest_algorithm -from validation import frontend_digest_matches +from validation import settle_mismatch from nativemode import hash_mismatch_excludes_file import hashlib from common import filter_systems_by_target @@ -31,11 +31,12 @@ from packresolve import resolve_file from common import resolve_platform_cores from common import sanitize_pack_path import zipfile -def _members_are_held(data: bytes, by_md5: dict, db: dict) -> bool: - """Whether every member of an archive is a dump the collection holds.""" - import io +import io +from collections.abc import Callable - held_inside = build_zip_contents_index(db) + +def _members_are_held(data: bytes, by_md5: dict, held_inside: dict) -> bool: + """Whether every member of an archive is a dump the collection holds.""" try: with zipfile.ZipFile(io.BytesIO(data)) as archive: members = [i for i in archive.infolist() if not i.is_dir()] @@ -50,6 +51,42 @@ def _members_are_held(data: bytes, by_md5: dict, db: dict) -> bool: return True +def _once(build: Callable[[], dict]) -> Callable[[], dict]: + """A value built on first use and kept.""" + built: list[dict] = [] + + def get() -> dict: + if not built: + built.append(build()) + return built[0] + + return get + + +def _settle_untracked( + status: str, + name: str, + zf: zipfile.ZipFile, + by_md5: dict, + held_inside: Callable[[], dict], + errors: list[str], + file_name: str, +) -> tuple[str, str]: + """Status of a member no hash recognised, and the name to record. + + An archive the builder assembled (a MAME clone set) passes when every + member is a dump the collection holds, loose or inside a romset. Bytes + nothing recognises were written wrong or come from a source that no + longer matches: counted and passed, they went unseen. + """ + if status != "untracked": + return status, file_name + if name.endswith(".zip") and _members_are_held(zf.read(name), by_md5, held_inside()): + return "verified_members", os.path.basename(name) + errors.append(f"{name}: content matches no collected file") + return "untracked", file_name + + def verify_pack( zip_path: str, db: dict, data_registry: dict | None = None ) -> tuple[bool, dict]: @@ -61,6 +98,9 @@ def verify_pack( """ files_db = db.get("files", {}) # SHA1 -> file_info by_md5 = db.get("indexes", {}).get("by_md5", {}) # MD5 -> SHA1 + # Built on the first archive no hash recognises, once per pack: the + # index opens every archive of the collection. + held_inside = _once(lambda: build_zip_contents_index(db)) by_name = db.get("indexes", {}).get("by_name", {}) # name -> [SHA1] # Data directory file index @@ -202,21 +242,9 @@ def verify_pack( except (zipfile.BadZipFile, OSError): continue - # An archive the builder assembled (a MAME clone set): every - # member must be a dump the collection holds, loose or inside a - # romset. - if ( - status == "untracked" - and name.endswith(".zip") - and _members_are_held(zf.read(name), by_md5, db) - ): - status = "verified_members" - file_name = os.path.basename(name) - - if status == "untracked": - # Bytes nothing recognises: written wrong, or a source that - # no longer matches. Counted and passed, it went unseen. - errors.append(f"{name}: content matches no collected file") + status, file_name = _settle_untracked( + status, name, zf, by_md5, held_inside, errors, file_name + ) manifest["files"].append( { @@ -440,27 +468,10 @@ def _intentional_hash_exclusion( data_dir_registry=data_dir_registry, offline=True, ) - if status != "hash_mismatch": + # The builder ships what the frontend would accept: the exact inner + # ROM of a zipped_file entry, or a file its own digest matches. + if settle_mismatch(entry, local_path, status, verification_mode) != "hash_mismatch": return False - if not entry.get("zipped_file") and frontend_digest_matches( - entry, local_path, digest_algorithm(verification_mode) - ): - # The frontend's own digest accepts it: the builder ships it. - return False - - # A container can mismatch the outer declaration while still carrying - # the exact inner ROM requested by Batocera-style zipped_file entries. - zipped_file = entry.get("zipped_file") - if zipped_file and local_path: - declared = str(entry.get("md5") or "") - candidates = [value.strip() for value in declared.split(",") if value.strip()] - if not candidates: - candidates = [""] - if any( - check_inside_zip(local_path, zipped_file, candidate) == "ok" - for candidate in candidates - ): - return False return True def _structural_errors(zf, zip_set: set) -> list[str]: diff --git a/scripts/validation.py b/scripts/validation.py index 0a710cbc..a088a4c4 100644 --- a/scripts/validation.py +++ b/scripts/validation.py @@ -11,6 +11,8 @@ import os from common import compute_hashes, size_fits from hashing import parse_md5_list +from nativemode import digest_algorithm, hash_mismatch_excludes_file +from ziptools import check_inside_zip # Validation types that require console-specific cryptographic keys. # verify.py cannot reproduce these -size checks still apply if combined. @@ -450,3 +452,41 @@ def frontend_digest_matches(file_entry: dict, local_path: str, algorithm: str) - if not declared or not local_path: return False return compute_hashes(local_path)[algorithm].lower() in declared + + +def inner_rom_check(file_entry: dict, local_path: str) -> str: + """How an archive answers an entry that pins a ROM inside it. + + Batocera, RetroBat and ROCKNIX hash a member of the ZIP, never the ZIP: + the resolver hands back the archive as a mismatch and this decides it. + Returns check_inside_zip's answer for the first accepted MD5 that + matches, else for the last one tried. The pack and the install manifest + both read it, so a file one of them ships is a file the other lists. + """ + declared = [m.strip() for m in file_entry.get("md5", "").split(",") if m.strip()] + result = "not_in_zip" + for candidate in declared or [""]: + result = check_inside_zip(local_path, file_entry["zipped_file"], candidate) + if result == "ok": + break + return result + + +def settle_mismatch( + file_entry: dict, local_path: str | None, status: str, verification_mode: str +) -> str: + """Upgrade a hash mismatch the frontend itself would accept. + + Batocera pins the md5 of a ROM inside the archive (zipped_file), and a + frontend that hashes one digest accepts a file whose other declared + hashes disagree. + """ + if status != "hash_mismatch" or not local_path: + return status + if file_entry.get("zipped_file"): + return "zip_exact" if inner_rom_check(file_entry, local_path) == "ok" else status + if hash_mismatch_excludes_file(verification_mode) and frontend_digest_matches( + file_entry, local_path, digest_algorithm(verification_mode) + ): + return "frontend_digest_exact" + return status