mirror of
https://github.com/Abdess/retroarch_system.git
synced 2026-10-18 09:23:26 -05:00
Abdessamad Derraz
ae3dc63bc3
RomM's verify_file_hashes compares the size, then md5, sha1 or crc. We compared md5 alone, so a fixture entry with one mistyped hash was left out of the pack RomM would accept it from. The rule is declared in romm.yml and read by verify, the builder and the manifest.
784 lines
30 KiB
Python
784 lines
30 KiB
Python
"""Checking a built pack the way the frontend would.
|
|
|
|
The builder and the verifier have to reach the same verdict on the
|
|
same file, so both read the native mode from one place: a pack and a
|
|
coverage report that disagree describe different collections."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections import Counter
|
|
|
|
from auto_fetch import DEFAULT_BIOS_DIR
|
|
from artifacts import _build_timestamp
|
|
from packextras import _collect_emulator_extras
|
|
from packextras import _detect_extras_prefix
|
|
from packpaths import _has_path_conflict
|
|
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 baseline_verdict
|
|
from nativemode import hash_mismatch_excludes_file
|
|
import functools
|
|
import hashlib
|
|
from common import filter_systems_by_target
|
|
from common import load_emulator_profiles
|
|
from common import load_platform_config
|
|
import os
|
|
from packextras import platform_region_groups
|
|
from packpaths import _flat
|
|
from nativemode import reads_file_contents
|
|
import region as region_mod
|
|
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
|
|
|
|
|
|
@functools.lru_cache(maxsize=None)
|
|
def _file_sha1(path: str) -> str:
|
|
"""SHA-1 of a cached data-directory file, read once per run."""
|
|
digest = hashlib.sha1()
|
|
with open(path, "rb") as handle:
|
|
for chunk in iter(lambda: handle.read(1 << 20), b""):
|
|
digest.update(chunk)
|
|
return digest.hexdigest()
|
|
|
|
|
|
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()]
|
|
if not members:
|
|
return False
|
|
for info in members:
|
|
digest = hashlib.md5(archive.read(info)).hexdigest()
|
|
if digest not in by_md5 and digest not in held_inside:
|
|
return False
|
|
except (zipfile.BadZipFile, OSError):
|
|
return False
|
|
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]:
|
|
"""Verify a generated pack ZIP by re-hashing every file inside.
|
|
|
|
Checks against database.json, data directory caches, and verifies
|
|
rebuilt ZIP content by comparing inner CRC32s against source.
|
|
Returns (all_ok, manifest_dict).
|
|
"""
|
|
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
|
|
_data_index: dict[str, list[str]] = {}
|
|
_data_path_index: dict[str, str] = {}
|
|
if data_registry:
|
|
for _dk, _de in data_registry.items():
|
|
cache = _de.get("local_cache", "")
|
|
if not cache or not os.path.isdir(cache):
|
|
continue
|
|
for _r, _d, _fns in os.walk(cache):
|
|
for _fn in _fns:
|
|
_fp = os.path.join(_r, _fn)
|
|
_rel = os.path.relpath(_fp, cache)
|
|
_data_path_index[_rel] = _fp
|
|
_data_index.setdefault(_fn, []).append(_fp)
|
|
|
|
manifest = {
|
|
"schema_version": 1,
|
|
"version": 1,
|
|
"generator": "retrobios generate_pack.py",
|
|
"generated": _build_timestamp(db),
|
|
"files": [],
|
|
}
|
|
errors = []
|
|
|
|
with zipfile.ZipFile(zip_path, "r") as zf:
|
|
for info in zf.infolist():
|
|
if info.is_dir():
|
|
continue
|
|
name = info.filename
|
|
# The builder writes its notes beside the slot they explain
|
|
# (bios/INSTRUCTIONS_x.txt, games/keen6/INSTRUCTIONS_y.txt), so
|
|
# they are told by their own name, not by a root prefix.
|
|
base = name.rsplit("/", 1)[-1]
|
|
if base.startswith(("INSTRUCTIONS_", "RENAMED_")) or name in (
|
|
"manifest.json",
|
|
"README.txt",
|
|
):
|
|
continue
|
|
with zf.open(info) as f:
|
|
sha1_h = hashlib.sha1()
|
|
md5_h = hashlib.md5()
|
|
size = 0
|
|
for chunk in iter(lambda: f.read(65536), b""):
|
|
sha1_h.update(chunk)
|
|
md5_h.update(chunk)
|
|
size += len(chunk)
|
|
sha1 = sha1_h.hexdigest()
|
|
md5 = md5_h.hexdigest()
|
|
|
|
# Look up in database: files_db keyed by SHA1
|
|
db_entry = files_db.get(sha1)
|
|
status = "verified"
|
|
file_name = ""
|
|
if db_entry:
|
|
file_name = db_entry.get("name", "")
|
|
else:
|
|
# Try MD5 -> SHA1 lookup
|
|
ref_sha1 = by_md5.get(md5)
|
|
if ref_sha1:
|
|
db_entry = files_db.get(ref_sha1)
|
|
if db_entry:
|
|
file_name = db_entry.get("name", "")
|
|
status = "verified_md5"
|
|
else:
|
|
status = "untracked"
|
|
else:
|
|
status = "untracked"
|
|
|
|
# Rebuilt ZIP: verify inner ROM CRC32s match source
|
|
if status == "untracked" and name.endswith(".zip"):
|
|
_bn = os.path.basename(name)
|
|
for _src_sha1 in by_name.get(_bn, []):
|
|
if _src_sha1 not in files_db:
|
|
continue
|
|
_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
|
|
|
|
# Data directory: check against cached files
|
|
if status == "untracked" and _data_index:
|
|
_bn = os.path.basename(name)
|
|
_pr = name
|
|
for _known_prefix in ("system/", "bios/", "BIOS/", "Firmware/"):
|
|
if name.startswith(_known_prefix):
|
|
_pr = name[len(_known_prefix):]
|
|
break
|
|
_cands = []
|
|
if _pr in _data_path_index:
|
|
_cands.append(_data_path_index[_pr])
|
|
for _dp in _data_index.get(_bn, []):
|
|
if _dp not in _cands:
|
|
_cands.append(_dp)
|
|
for _dp in _cands:
|
|
if not os.path.exists(_dp):
|
|
continue
|
|
# The content, not the size: a member written corrupt
|
|
# keeps its length, and a same-sized homonym from another
|
|
# cache is another file.
|
|
if os.path.getsize(_dp) == size and _file_sha1(_dp) == sha1:
|
|
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
|
|
|
|
status, file_name = _settle_untracked(
|
|
status, name, zf, by_md5, held_inside, errors, file_name
|
|
)
|
|
|
|
manifest["files"].append(
|
|
{
|
|
"path": name,
|
|
"sha1": sha1,
|
|
"md5": md5,
|
|
"size": size,
|
|
"status": status,
|
|
"name": file_name,
|
|
}
|
|
)
|
|
|
|
# Corruption check: SHA1 in DB but doesn't match what we computed
|
|
# This should never happen (we looked up by SHA1), but catches
|
|
# edge cases where by_md5 resolved to a different SHA1
|
|
if db_entry and status == "verified_md5":
|
|
expected_sha1 = db_entry.get("sha1", "")
|
|
if expected_sha1 and expected_sha1.lower() != sha1.lower():
|
|
errors.append(
|
|
f"{name}: SHA1 mismatch (expected {expected_sha1}, got {sha1})"
|
|
)
|
|
|
|
verified = sum(1 for f in manifest["files"] if f["status"].startswith("verified"))
|
|
untracked = sum(1 for f in manifest["files"] if f["status"] == "untracked")
|
|
total = len(manifest["files"])
|
|
manifest["summary"] = {
|
|
"total_files": total,
|
|
"verified": verified,
|
|
"untracked": untracked,
|
|
"errors": len(errors),
|
|
}
|
|
manifest["errors"] = errors
|
|
|
|
all_ok = len(errors) == 0
|
|
return all_ok, manifest
|
|
|
|
def generate_sha256sums(output_dir: str) -> str | None:
|
|
"""Generate SHA256SUMS.txt for all ZIP files in output_dir."""
|
|
sums_path = os.path.join(output_dir, "SHA256SUMS.txt")
|
|
entries = []
|
|
for name in sorted(os.listdir(output_dir)):
|
|
if not name.endswith(".zip"):
|
|
continue
|
|
path = os.path.join(output_dir, name)
|
|
sha256 = hashlib.sha256()
|
|
with open(path, "rb") as f:
|
|
for chunk in iter(lambda: f.read(65536), b""):
|
|
sha256.update(chunk)
|
|
entries.append(f"{sha256.hexdigest()} {name}")
|
|
if not entries:
|
|
return None
|
|
with open(sums_path, "w") as f:
|
|
f.write("\n".join(entries) + "\n")
|
|
print(f"\n{sums_path}: {len(entries)} pack checksums")
|
|
return sums_path
|
|
|
|
def _hash_matches(declared: str, actual: str) -> bool:
|
|
"""Compare a declared hash value against an actual hex digest.
|
|
|
|
Handles comma-separated multi-hash lists and uppercase (Recalbox)
|
|
and truncated MD5s (Batocera 29-char): a declared value shorter
|
|
than the digest matches by prefix.
|
|
"""
|
|
actual = actual.lower()
|
|
for cand in declared.split(","):
|
|
cand = cand.strip().lower()
|
|
if not cand:
|
|
continue
|
|
if len(cand) < len(actual):
|
|
if actual.startswith(cand):
|
|
return True
|
|
elif actual == cand:
|
|
return True
|
|
return False
|
|
|
|
def _check_member_hash(
|
|
zf: zipfile.ZipFile, member: str, file_entry: dict, mode: str
|
|
) -> str | None:
|
|
"""Verify a pack member against its platform-declared hash.
|
|
|
|
Reproduces the platform's native check inside the pack: md5/sha1 of
|
|
the member bytes, or checkInsideZip when zipped_file is set
|
|
(Batocera hashes a ROM inside the ZIP, matched case-insensitively).
|
|
Returns an error string, or None when the member passes.
|
|
"""
|
|
declared = str(file_entry.get(mode) or "").strip()
|
|
if not declared:
|
|
return None
|
|
|
|
zipped_file = file_entry.get("zipped_file")
|
|
if zipped_file:
|
|
import io
|
|
|
|
try:
|
|
with zipfile.ZipFile(io.BytesIO(zf.read(member))) as inner:
|
|
want = zipped_file.casefold()
|
|
target = next(
|
|
(n for n in inner.namelist() if n.casefold() == want), None
|
|
)
|
|
if target is None:
|
|
return f"{member}: {zipped_file} not found inside ZIP"
|
|
actual = hashlib.md5(inner.read(target)).hexdigest()
|
|
except zipfile.BadZipFile:
|
|
return f"{member}: not a valid ZIP"
|
|
if not _hash_matches(declared, actual):
|
|
return (
|
|
f"{member}: {zipped_file} inside-zip md5 {actual} "
|
|
f"!= declared {declared}"
|
|
)
|
|
return None
|
|
|
|
h = hashlib.md5() if mode == "md5" else hashlib.sha1()
|
|
with zf.open(member) as f:
|
|
for chunk in iter(lambda: f.read(65536), b""):
|
|
h.update(chunk)
|
|
actual = h.hexdigest()
|
|
if _hash_matches(declared, actual):
|
|
return None
|
|
|
|
# Recalbox Md5Composite: MD5 over sorted inner contents of a ZIP,
|
|
# independent of compression and metadata (Zip::Md5Composite()).
|
|
if mode == "md5" and member.endswith(".zip"):
|
|
import io
|
|
|
|
try:
|
|
with zipfile.ZipFile(io.BytesIO(zf.read(member))) as inner:
|
|
names = sorted(n for n in inner.namelist() if not n.endswith("/"))
|
|
ch = hashlib.md5()
|
|
for n in names:
|
|
ch.update(inner.read(n))
|
|
if _hash_matches(declared, ch.hexdigest()):
|
|
return None
|
|
except (zipfile.BadZipFile, OSError):
|
|
pass
|
|
|
|
return f"{member}: {mode} {actual} != declared {declared}"
|
|
|
|
def _repo_satisfies_declaration(
|
|
entries: list[dict], db: dict, mode: str
|
|
) -> bool:
|
|
"""Check whether any repo file matches one of the declared hashes.
|
|
|
|
Used to separate pack divergence (a matching file exists but was not
|
|
packed) from data coverage gaps (no repo file matches the upstream
|
|
declaration): only the former is a pack generation error.
|
|
"""
|
|
from common import md5_composite
|
|
|
|
files_db = db.get("files", {})
|
|
by_md5 = db.get("indexes", {}).get("by_md5", {})
|
|
by_name = db.get("indexes", {}).get("by_name", {})
|
|
|
|
for fe in entries:
|
|
declared = str(fe.get(mode) or "").strip()
|
|
if not declared:
|
|
continue
|
|
hashes = [h.strip().lower() for h in declared.split(",") if h.strip()]
|
|
for h in hashes:
|
|
if mode == "sha1":
|
|
entry = files_db.get(h)
|
|
else:
|
|
entry = files_db.get(by_md5.get(h, ""))
|
|
if entry and os.path.exists(entry.get("path", "")):
|
|
return True
|
|
if mode == "md5":
|
|
hash_set = set(hashes)
|
|
for cand_sha in by_name.get(fe.get("name", ""), []):
|
|
entry = files_db.get(cand_sha)
|
|
if not entry:
|
|
continue
|
|
path = entry.get("path", "")
|
|
if not path.endswith(".zip") or not os.path.exists(path):
|
|
continue
|
|
try:
|
|
if md5_composite(path).lower() in hash_set:
|
|
return True
|
|
except (zipfile.BadZipFile, OSError):
|
|
continue
|
|
return False
|
|
|
|
def _repo_holds_member(file_entry: dict, db: dict) -> bool:
|
|
"""Whether a held archive of that name carries the declared member."""
|
|
files_db = db.get("files", {})
|
|
declared = [m.strip() for m in str(file_entry.get("md5") or "").split(",") if m.strip()]
|
|
for sha1 in db.get("indexes", {}).get("by_name", {}).get(file_entry.get("name", ""), []):
|
|
path = (files_db.get(sha1) or {}).get("path", "")
|
|
if not path.endswith(".zip") or not os.path.exists(path):
|
|
continue
|
|
if any(check_inside_zip(path, file_entry["zipped_file"], m) == "ok" for m in declared):
|
|
return True
|
|
return False
|
|
|
|
|
|
def _intentional_hash_exclusion(
|
|
entries: list[dict],
|
|
db: dict,
|
|
bios_dir: str = DEFAULT_BIOS_DIR,
|
|
zip_contents: dict | None = None,
|
|
data_dir_registry: dict | None = None,
|
|
verification_mode: str = "md5",
|
|
hash_rule: str = "",
|
|
) -> bool:
|
|
"""Return whether the builder must omit every declaration as unsafe.
|
|
|
|
A hash platform never substitutes a same-named payload for an explicitly
|
|
hash-identified one, so a missing pack member is accounted for when every
|
|
declaration for that destination resolves to a local hash mismatch. An
|
|
existence platform reads no bytes and the builder ships the file anyway,
|
|
so nothing is ever excluded on its behalf. Genuine absence,
|
|
external-download failure and a packable alternative stay conformance
|
|
errors in both modes.
|
|
"""
|
|
if not entries or not hash_mismatch_excludes_file(verification_mode):
|
|
return False
|
|
archive_index = zip_contents if zip_contents is not None else {}
|
|
for entry in entries:
|
|
local_path, status = resolve_file(
|
|
entry,
|
|
db,
|
|
bios_dir,
|
|
archive_index,
|
|
data_dir_registry=data_dir_registry,
|
|
offline=True,
|
|
)
|
|
# The builder's own verdict: a zipped_file archive is refused when the
|
|
# member is missing whatever the outer resolution said, a hash
|
|
# mismatch when the frontend's digest disagrees too.
|
|
_status, refusal = baseline_verdict(
|
|
entry, local_path, status, verification_mode, hash_rule
|
|
)
|
|
if refusal is None:
|
|
return False
|
|
return True
|
|
|
|
def _structural_errors(zf: zipfile.ZipFile, zip_set: set) -> list[str]:
|
|
"""Faults in the archive itself, before any file is looked up.
|
|
|
|
A duplicate entry, an absolute path or a traversal makes the pack
|
|
unsafe to extract whatever it contains, and a zero-byte member is a
|
|
file the user will believe they have. The markers Dolphin ships for
|
|
its graphics mods are legitimately empty.
|
|
"""
|
|
errors: list[str] = []
|
|
# Structural checks
|
|
dupes = sum(1 for c in Counter(zf.namelist()).values() if c > 1)
|
|
if dupes:
|
|
errors.append(f"{dupes} duplicate entries")
|
|
for n in zip_set:
|
|
if "//" in n:
|
|
errors.append(f"double slash: {n}")
|
|
if n.startswith("/"):
|
|
errors.append(f"absolute path: {n}")
|
|
if ".." in n:
|
|
errors.append(f"path traversal: {n}")
|
|
|
|
# Zero-byte check (exclude Dolphin GraphicMods markers)
|
|
for info in zf.infolist():
|
|
if info.file_size == 0 and not info.is_dir():
|
|
if "GraphicMods" not in info.filename and info.filename not in (
|
|
"manifest.json",
|
|
"README.txt",
|
|
):
|
|
errors.append(f"zero-byte: {info.filename}")
|
|
|
|
return errors
|
|
|
|
|
|
def verify_pack_against_platform(
|
|
zip_path: str,
|
|
platform_name: str,
|
|
platforms_dir: str,
|
|
db: dict | None = None,
|
|
emulators_dir: str = "emulators",
|
|
emu_profiles: dict | None = None,
|
|
regions: list[str] | None = None,
|
|
data_registry: dict | None = None,
|
|
target_cores: set[str] | None = None,
|
|
) -> tuple[bool, int, int, list[str], int, int, int, int, int]:
|
|
"""Verify a pack ZIP against its platform config and core requirements.
|
|
|
|
A region priority list narrows the expectation to what the builder would
|
|
have packed, using the same selection function. A hardware target narrows
|
|
it the same way and for the same reason: the systems it removes are not
|
|
missing from the pack, they were never asked for.
|
|
|
|
Checks:
|
|
1. Every baseline file declared by the platform exists in the ZIP
|
|
at the correct destination path
|
|
2. Every in-repo core extra file (from emulator profiles) is present
|
|
3. No duplicate entries
|
|
4. No path anomalies (double slash, absolute, traversal)
|
|
5. No unexpected zero-byte BIOS files
|
|
|
|
Returns ``(all_ok, checked, present, errors, baseline_checked,
|
|
baseline_present, core_checked, core_present, baseline_excluded)``.
|
|
"""
|
|
from collections import Counter
|
|
|
|
config = load_platform_config(platform_name, platforms_dir)
|
|
base_dest = config.get("base_destination", "")
|
|
errors: list[str] = []
|
|
checked = 0
|
|
present = 0
|
|
|
|
if emu_profiles is None:
|
|
emu_profiles = load_emulator_profiles(emulators_dir)
|
|
|
|
expected_systems = config.get("systems", {})
|
|
if target_cores is not None:
|
|
expected_systems = filter_systems_by_target(
|
|
expected_systems,
|
|
emu_profiles,
|
|
target_cores,
|
|
resolve_platform_cores(config, emu_profiles),
|
|
)
|
|
|
|
region_drops: set[str] = set()
|
|
if regions:
|
|
region_index = region_mod.platform_region_index(
|
|
config, emu_profiles, target_cores
|
|
)
|
|
held: set[str] = set()
|
|
region_groups, _extra_dests = platform_region_groups(
|
|
config,
|
|
expected_systems,
|
|
emulators_dir,
|
|
db,
|
|
base_dest,
|
|
emu_profiles,
|
|
target_cores=target_cores,
|
|
held_out=held,
|
|
)
|
|
region_drops = region_mod.resolve_region_drops(
|
|
region_groups, region_index, regions, held
|
|
)
|
|
|
|
with zipfile.ZipFile(zip_path, "r") as zf:
|
|
zip_set = set(zf.namelist())
|
|
zip_lower = {n.lower(): n for n in zip_set}
|
|
|
|
# Auto-detect flat vs nested ZIP
|
|
is_flat = bool(base_dest) and not any(
|
|
n.startswith(base_dest + "/")
|
|
for n in zip_set
|
|
if n not in ("README.txt", "manifest.json") and not n.endswith("/")
|
|
)
|
|
|
|
errors.extend(_structural_errors(zf, zip_set))
|
|
|
|
# 1. Baseline file presence + native hash check
|
|
verification_mode = config.get("verification_mode", "existence")
|
|
baseline_checked = 0
|
|
baseline_present = 0
|
|
baseline_excluded = 0
|
|
decl_by_member: dict[str, list[dict]] = {}
|
|
# Mirror the builder's path-conflict logic: a declaration whose path
|
|
# collides file-vs-directory with a packed member was skipped by the
|
|
# builder (upstream declares e.g. both SGB1.sfc and SGB1.sfc/program.rom)
|
|
zip_parents: set[str] = set()
|
|
for n in zip_set:
|
|
parts = n.split("/")
|
|
for i in range(1, len(parts)):
|
|
zip_parents.add("/".join(parts[:i]))
|
|
baseline_groups: dict[str, list[dict]] = {}
|
|
for _sys_id, system in expected_systems.items():
|
|
for fe in system.get("files", []):
|
|
dest = sanitize_pack_path(fe.get("destination", fe.get("name", "")))
|
|
if not dest:
|
|
continue
|
|
if region_drops and dest in region_drops:
|
|
continue
|
|
expected = f"{base_dest}/{dest}" if base_dest and not is_flat else dest
|
|
baseline_groups.setdefault(expected, []).append(fe)
|
|
|
|
baseline_checked = len(baseline_groups)
|
|
exclusion_index = build_zip_contents_index(db) if db is not None else {}
|
|
for expected, declarations in baseline_groups.items():
|
|
if expected in zip_set:
|
|
member = expected
|
|
elif expected.lower() in zip_lower:
|
|
member = zip_lower[expected.lower()]
|
|
elif _has_path_conflict(expected, zip_set, zip_parents):
|
|
# Skipped by the builder for the same reason: not an error
|
|
baseline_present += 1
|
|
continue
|
|
elif db is not None and _intentional_hash_exclusion(
|
|
declarations,
|
|
db,
|
|
zip_contents=exclusion_index,
|
|
data_dir_registry=data_registry,
|
|
verification_mode=verification_mode,
|
|
hash_rule=config.get("hash_rule", ""),
|
|
):
|
|
baseline_excluded += 1
|
|
continue
|
|
else:
|
|
errors.append(f"baseline missing: {expected}")
|
|
continue
|
|
baseline_present += 1
|
|
decl_by_member.setdefault(member, []).extend(declarations)
|
|
|
|
# Reproduce the platform's native hash check on pack bytes.
|
|
# A destination declared by several entries passes when the packed
|
|
# member satisfies any of them. A failure only counts as a pack
|
|
# error when the repo holds a file matching a declaration: without
|
|
# one, the pack ships its best effort and the gap is a data issue
|
|
# reported by verify.py, not a generation bug.
|
|
digest = digest_algorithm(verification_mode)
|
|
if reads_file_contents(verification_mode) and digest:
|
|
for member, decl_entries in decl_by_member.items():
|
|
checkable = [
|
|
fe
|
|
for fe in decl_entries
|
|
if str(fe.get(digest) or "").strip()
|
|
]
|
|
if not checkable:
|
|
continue
|
|
# An archive declared once per ROM it must hold is checked
|
|
# once per ROM by the frontend: each such declaration has to
|
|
# pass on its own, where the repository holds an archive
|
|
# that would.
|
|
for fe in checkable:
|
|
if not fe.get("zipped_file"):
|
|
continue
|
|
err = _check_member_hash(zf, member, fe, digest)
|
|
if err is not None and (db is None or _repo_holds_member(fe, db)):
|
|
errors.append(err)
|
|
member_errors = []
|
|
satisfied = False
|
|
for fe in checkable:
|
|
err = _check_member_hash(zf, member, fe, digest)
|
|
if err is None:
|
|
satisfied = True
|
|
break
|
|
member_errors.append(err)
|
|
if satisfied:
|
|
continue
|
|
if db is None or _repo_satisfies_declaration(
|
|
checkable, db, verification_mode
|
|
):
|
|
errors.append(member_errors[0])
|
|
|
|
# 2. Core extras presence (files from emulator profiles, in repo)
|
|
# Mirror the pack builder's skip logic: only count files that
|
|
# can actually be resolved and don't have path conflicts.
|
|
core_checked = 0
|
|
core_present = 0
|
|
if db is not None:
|
|
core_files = _collect_emulator_extras(
|
|
config,
|
|
emulators_dir,
|
|
db,
|
|
set(),
|
|
base_dest,
|
|
emu_profiles,
|
|
target_cores=target_cores,
|
|
)
|
|
seen_conformance: set[str] = set(zip_set)
|
|
seen_parents: set[str] = set()
|
|
for n in zip_set:
|
|
parts = n.split("/")
|
|
for i in range(1, len(parts)):
|
|
seen_parents.add("/".join(parts[:i]))
|
|
extras_pfx = _detect_extras_prefix(config, base_dest)
|
|
for fe in core_files:
|
|
raw_dest = fe.get("destination", fe.get("name", ""))
|
|
dest = sanitize_pack_path(raw_dest)
|
|
if not dest:
|
|
continue
|
|
if region_drops and sanitize_pack_path(dest) in region_drops:
|
|
continue
|
|
# The builder's own placement: the prefix, then the flattening
|
|
# of a pack written at the BIOS root. A destination that
|
|
# already starts with the prefix (aethersx2's bios/<image>)
|
|
# lands at the root of a flat pack, not under bios/.
|
|
if extras_pfx and not dest.startswith(f"{extras_pfx}/"):
|
|
full = f"{extras_pfx}/{dest}"
|
|
else:
|
|
full = dest
|
|
full = _flat(full, base_dest, is_flat)
|
|
# Skip path conflicts (same logic as pack builder)
|
|
if _has_path_conflict(full, seen_conformance, seen_parents):
|
|
continue
|
|
# Skip unresolvable files (game_data dirs, etc.)
|
|
local_path, status = resolve_file(
|
|
fe,
|
|
db,
|
|
"bios",
|
|
{},
|
|
dest_hint=raw_dest,
|
|
data_dir_registry=data_registry,
|
|
offline=True,
|
|
)
|
|
if status in ("not_found", "external", "user_provided"):
|
|
continue
|
|
core_checked += 1
|
|
if full in zip_set or full.lower() in zip_lower:
|
|
core_present += 1
|
|
seen_conformance.add(full)
|
|
_register_path(full, seen_conformance, seen_parents)
|
|
else:
|
|
errors.append(f"core missing: {full}")
|
|
|
|
checked = baseline_checked + core_checked
|
|
present = baseline_present + core_present
|
|
|
|
return (
|
|
len(errors) == 0,
|
|
checked,
|
|
present,
|
|
errors,
|
|
baseline_checked,
|
|
baseline_present,
|
|
core_checked,
|
|
core_present,
|
|
baseline_excluded,
|
|
)
|