refactor: one mismatch rule for builder and verifier

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.

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]:

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