mirror of
https://github.com/Abdess/retroarch_system.git
synced 2026-11-11 20:43:50 -06:00
feat: run content checks profiles declare
validation: [romdir] walks a PS2 ROM directory the way the profile's fork reads it, masked_sha1 hashes a file with a byte range cleared, and size_multiple bounds a size to a granule. The resolver name steps, the pack and manifest scans, the emulator expansion and verify call the same content_fits. A PCSX2 companion is taken only beside an image that passes, and a companion placeholder no longer seeds a scan of its own.
This commit is contained in:
1 parent
51e5fd0dbd
commit
ec9fe03673
12 files changed
+777
-29
No files matched your search
+124
-11
@@ -316,11 +316,110 @@ def adler32_fits(file_entry: dict, path: str, db: dict | None = None) -> bool:
|
||||
return chosen is None or actual == chosen
|
||||
|
||||
|
||||
# The content tests a profile can declare in ``validation:``, each run by
|
||||
# reading the candidate the way the emulator does.
|
||||
CONTENT_CHECKS = frozenset({"romdir", "masked_sha1"})
|
||||
|
||||
|
||||
def _core_checks(file_entry: dict) -> list:
|
||||
validation = file_entry.get("validation")
|
||||
if isinstance(validation, dict):
|
||||
validation = validation.get("core", [])
|
||||
return list(validation or [])
|
||||
|
||||
|
||||
def declared_content_checks(file_entry: dict) -> set[str]:
|
||||
"""The content tests the entry's code runs on a candidate."""
|
||||
return set(CONTENT_CHECKS.intersection(_core_checks(file_entry)))
|
||||
|
||||
|
||||
@functools.lru_cache(maxsize=4096)
|
||||
def _romdir_verdict(path: str, size: int, mtime_ns: int, spec: str) -> bool:
|
||||
from romdir import romdir_accepts # noqa: PLC0415
|
||||
|
||||
with open(path, "rb") as f:
|
||||
data = f.read()
|
||||
return romdir_accepts(data, json.loads(spec))
|
||||
|
||||
|
||||
@functools.lru_cache(maxsize=4096)
|
||||
def _masked_sha1_verdict(path: str, size: int, mtime_ns: int, spec: str) -> bool:
|
||||
"""Amiberry's flash image identification (getromdatabydata, rommgr.cpp).
|
||||
|
||||
A CyberStorm or Blizzard PPC image carries its board configuration in
|
||||
bytes 0x10-0x2f, so the table SHA-1 is taken with those bytes cleared,
|
||||
over the first 0x1c000 bytes, and for a 512 KB file also over the
|
||||
first 0x50000. ``reads`` lists the file sizes the code tries the
|
||||
transform on and the lengths it hashes for each.
|
||||
"""
|
||||
rules = json.loads(spec)
|
||||
reads = {int(r["size"]): [int(n) for n in r["lengths"]] for r in rules.get("reads", [])}
|
||||
if size not in reads:
|
||||
return False
|
||||
with open(path, "rb") as f:
|
||||
data = bytearray(f.read())
|
||||
start, end = (int(v) for v in rules.get("zeroed", [0, 0]))
|
||||
data[start:end] = bytes(max(0, min(end, len(data)) - start))
|
||||
wanted = str(rules.get("sha1", "")).lower()
|
||||
return any(hashlib.sha1(data[:length]).hexdigest() == wanted for length in reads[size])
|
||||
|
||||
|
||||
def content_fits(file_entry: dict, path: str) -> bool:
|
||||
"""Whether a candidate passes the content tests its entry declares.
|
||||
|
||||
The PCSX2 family lists any file of its BIOS folder whose ROM directory
|
||||
holds RESET and ROMVER (LoadBiosVersion, pcsx2/ps2/BiosTools.cpp), with
|
||||
no name and, in most forks, no size: ``validation: [romdir]`` and the
|
||||
``romdir:`` spec of how that fork reads the table (romdir.py). Amiberry
|
||||
names a flash image by the SHA-1 of its bytes with the configuration
|
||||
cleared (``masked_sha1``). A file that cannot be read fails. An entry
|
||||
declaring no content test accepts every candidate.
|
||||
"""
|
||||
checks = declared_content_checks(file_entry)
|
||||
if not checks:
|
||||
return True
|
||||
try:
|
||||
stat = os.stat(path)
|
||||
except OSError:
|
||||
return False
|
||||
verdicts = {"romdir": _romdir_verdict, "masked_sha1": _masked_sha1_verdict}
|
||||
for check in sorted(checks):
|
||||
spec = json.dumps(file_entry.get(check) or {}, sort_keys=True)
|
||||
try:
|
||||
if not verdicts[check](path, stat.st_size, stat.st_mtime_ns, spec):
|
||||
return False
|
||||
except OSError:
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
def shape_fits(file_entry: dict, path: str, size: int | None = None) -> bool:
|
||||
"""Size and content together: what a scan by shape checks per candidate."""
|
||||
if size is None:
|
||||
try:
|
||||
size = os.path.getsize(path)
|
||||
except OSError:
|
||||
return False
|
||||
return size_fits(file_entry, size) and content_fits(file_entry, path)
|
||||
|
||||
|
||||
def declares_shape(file_entry: dict) -> bool:
|
||||
"""Whether the entry says anything that tells one candidate from another."""
|
||||
return bool(
|
||||
file_entry.get("size") or file_entry.get("min_size")
|
||||
or file_entry.get("max_size") or file_entry.get("size_multiple")
|
||||
or declared_content_checks(file_entry)
|
||||
)
|
||||
|
||||
|
||||
def size_fits(file_entry: dict, candidate_size: int) -> bool:
|
||||
"""Whether a size is one the entry declares.
|
||||
|
||||
``size`` is an exact size or the list of sizes the code accepts, and it
|
||||
outranks a ``min_size``/``max_size`` range. Nothing declared fits all.
|
||||
outranks a ``min_size``/``max_size`` range. ``size_multiple`` is a
|
||||
granule the size must be a whole number of, as xemu maps any flash
|
||||
image whose size is a positive multiple of 64 KiB (hw/xbox/xbox.c).
|
||||
Nothing declared fits all.
|
||||
"""
|
||||
declared = file_entry.get("size")
|
||||
if declared is not None:
|
||||
@@ -330,7 +429,10 @@ def size_fits(file_entry: dict, candidate_size: int) -> bool:
|
||||
high = file_entry.get("max_size")
|
||||
if low is not None and candidate_size < low:
|
||||
return False
|
||||
return not (high is not None and candidate_size > high)
|
||||
if high is not None and candidate_size > high:
|
||||
return False
|
||||
multiple = file_entry.get("size_multiple")
|
||||
return not (multiple and candidate_size % int(multiple))
|
||||
|
||||
|
||||
|
||||
@@ -993,6 +1095,7 @@ def declared_hash_verdict(
|
||||
if file_entry is not None and not (
|
||||
name_match_size_ok(file_entry, os.path.getsize(candidate))
|
||||
and adler32_fits(file_entry, candidate, db)
|
||||
and content_fits(file_entry, candidate)
|
||||
):
|
||||
return "hash_mismatch"
|
||||
return "data_dir"
|
||||
@@ -1100,7 +1203,9 @@ def _data_dir_tail(
|
||||
for candidate in hits:
|
||||
if name_match_size_ok(
|
||||
file_entry, os.path.getsize(candidate)
|
||||
) and adler32_fits(file_entry, candidate, db):
|
||||
) and adler32_fits(file_entry, candidate, db) and content_fits(
|
||||
file_entry, candidate
|
||||
):
|
||||
return candidate
|
||||
return None
|
||||
|
||||
@@ -1156,7 +1261,7 @@ def _resolve_agnostic(
|
||||
path = entry.get("path", "")
|
||||
if not path.startswith(prefix):
|
||||
continue
|
||||
if size_fits(file_entry, entry.get("size", 0)) and os.path.exists(path):
|
||||
if os.path.exists(path) and shape_fits(file_entry, path, entry.get("size", 0)):
|
||||
return path, "agnostic_fallback"
|
||||
return None
|
||||
|
||||
@@ -1465,12 +1570,12 @@ def _by_affinity(paths: list[str], file_entry: dict, dest_hint: str) -> list[str
|
||||
hint = dest_hint or file_entry.get("path") or file_entry.get("destination") or ""
|
||||
wanted = _affinity_tokens(hint.rsplit("/", 1)[0] if "/" in hint else "")
|
||||
owners = _owner_tokens(file_entry)
|
||||
sized = any(file_entry.get(k) for k in ("size", "min_size", "max_size"))
|
||||
sized = declares_shape(file_entry)
|
||||
|
||||
def score(path: str) -> tuple[int, bool, bool, bool, int]:
|
||||
fits = 0
|
||||
if sized and os.path.exists(path):
|
||||
fits = int(size_fits(file_entry, os.path.getsize(path)))
|
||||
fits = int(shape_fits(file_entry, path))
|
||||
tokens = _affinity_tokens(path.rsplit("/", 1)[0])
|
||||
# The owner's own copy ranks above any shared tree segment: NetherSX2's
|
||||
# shaders/common/ beat armsx2's own fxaa.fx on two directory tokens.
|
||||
@@ -1797,7 +1902,9 @@ def resolve_local_file(
|
||||
# cpc6128.rom is not cap32's 32 KB one at the same tail.
|
||||
if not name_match_size_ok(
|
||||
file_entry, files_db[match_sha1].get("size")
|
||||
) or not adler32_fits(file_entry, path, db):
|
||||
) or not adler32_fits(file_entry, path, db) or not content_fits(
|
||||
file_entry, path
|
||||
):
|
||||
continue
|
||||
return path, "path_exact"
|
||||
status = _record_match_status(match_sha1)
|
||||
@@ -1830,7 +1937,7 @@ def resolve_local_file(
|
||||
record = files_db.get(match_sha1, {})
|
||||
return name_match_size_ok(file_entry, record.get("size")) and adler32_fits(
|
||||
file_entry, record.get("path", ""), db
|
||||
)
|
||||
) and content_fits(file_entry, record.get("path", ""))
|
||||
|
||||
# An entry the profile documents as unsourceable is one the collection
|
||||
# cannot hold: a font that only exists inside a paid package, cheat codes
|
||||
@@ -2030,7 +2137,10 @@ def resolve_local_file(
|
||||
# both an Apple IIgs image and a Macintosh one. It ranks rather than
|
||||
# excludes, because a present file of the wrong size is reported as
|
||||
# present and untested, and dropping it would report it as absent.
|
||||
sized = [c for c in named if name_match_size_ok(file_entry, c[2])]
|
||||
sized = [
|
||||
c for c in named
|
||||
if name_match_size_ok(file_entry, c[2]) and content_fits(file_entry, c[0])
|
||||
]
|
||||
if sized:
|
||||
named = sized
|
||||
|
||||
@@ -2107,7 +2217,10 @@ def resolve_local_file(
|
||||
if canonical and canonical != name:
|
||||
canonical_entry = {
|
||||
field: file_entry[field]
|
||||
for field in ("size", "min_size", "max_size", "validation")
|
||||
for field in (
|
||||
"size", "min_size", "max_size", "size_multiple", "validation",
|
||||
*sorted(CONTENT_CHECKS),
|
||||
)
|
||||
if field in file_entry
|
||||
}
|
||||
canonical_entry["name"] = canonical
|
||||
@@ -3142,7 +3255,7 @@ def platform_layouts(
|
||||
# validation on the way and resolved to an IBM PCjr ROM of the same name.
|
||||
PROFILE_IDENTITY_FIELDS = (
|
||||
"sha1", "md5", "sha256", "crc32", "size", "min_size", "max_size",
|
||||
"validation", "aliases", "unsourceable",
|
||||
"size_multiple", "validation", *sorted(CONTENT_CHECKS), "aliases", "unsourceable",
|
||||
)
|
||||
|
||||
|
||||
|
||||
@@ -42,6 +42,7 @@ from common import (
|
||||
load_database,
|
||||
load_emulator_profiles,
|
||||
load_platform_config,
|
||||
content_fits,
|
||||
name_match_size_ok,
|
||||
parse_md5_list,
|
||||
require_platform,
|
||||
@@ -247,6 +248,7 @@ def _resolve_source(
|
||||
sha1s = [sha1s]
|
||||
return any(
|
||||
name_match_size_ok(file_entry, db_files.get(sha1, {}).get("size"))
|
||||
and content_fits(file_entry, db_files.get(sha1, {}).get("path", ""))
|
||||
and _record_carries(sha1, db_files.get(sha1, {}), declared)
|
||||
for sha1 in sha1s
|
||||
)
|
||||
|
||||
+68
-12
@@ -19,7 +19,7 @@ from common import resolve_platform_cores
|
||||
from common import core_system_dirs, platform_layouts, system_dir_path
|
||||
from common import standalone_set as platform_standalone_set
|
||||
from common import OUTER_ROOTS, outer_layout, sanitize_pack_path
|
||||
from common import size_fits
|
||||
from common import declares_shape, shape_fits, size_fits
|
||||
from validation import deliverable, read_from_save_dir, read_from_system_dir
|
||||
from packpaths import libretro_systems, save_destination, save_layout
|
||||
from region import FILE_GROUP
|
||||
@@ -602,7 +602,7 @@ def _agnostic_scan_extras(
|
||||
# the flag icons of an Android package walked into a PS2 pack.
|
||||
if f.get("category", "bios") != "bios":
|
||||
continue
|
||||
if not (f.get("size") or f.get("min_size") or f.get("max_size")):
|
||||
if not declares_shape(f):
|
||||
continue
|
||||
# The free name is the BIOS directory's: a shader, a sound or a
|
||||
# font the emulator reads from its resources tree has a fixed
|
||||
@@ -663,7 +663,7 @@ def _agnostic_scan_extras(
|
||||
continue
|
||||
elif not path.startswith(path_prefix):
|
||||
continue
|
||||
if not companion_ext and not size_fits(f, entry.get("size", 0)):
|
||||
if not companion_ext and not shape_fits(f, path, entry.get("size", 0)):
|
||||
continue
|
||||
scan_name = entry.get("name", "")
|
||||
if not scan_name:
|
||||
@@ -672,6 +672,12 @@ def _agnostic_scan_extras(
|
||||
stem, _, ext = scan_name.rpartition(".")
|
||||
if not stem or ext.lower() != companion_ext.lower():
|
||||
continue
|
||||
# Read only beside the image the core selected, so a
|
||||
# companion counts only next to an image the image's
|
||||
# own test accepts: Namco System 246 .nvm and .mec
|
||||
# dumps sit beside no PS2 BIOS.
|
||||
if not _image_beside(path, stem, image, files_db):
|
||||
continue
|
||||
scan_name = f"{stem}.{companion_ext}"
|
||||
# Into the directory the seed entry names, not the root:
|
||||
# yaps2 reads its BIOS from pcsx2/bios/.
|
||||
@@ -745,7 +751,8 @@ def placeholder_directories(
|
||||
if f.get("size") and not size_fits(seed, f["size"]):
|
||||
continue
|
||||
local, status = resolve_local_file({**f, "source_profile": other_name}, db)
|
||||
if local and "/" in local and resolution_is_hash_exact(status):
|
||||
if (local and "/" in local and resolution_is_hash_exact(status)
|
||||
and shape_fits(seed, local)):
|
||||
directories.add(local.rsplit("/", 1)[0])
|
||||
return frozenset(directories)
|
||||
|
||||
@@ -768,7 +775,11 @@ def _placeholder_scan(
|
||||
name = str(seed.get("name", ""))
|
||||
if "<" not in name or seed.get("category", "bios") != "bios":
|
||||
continue
|
||||
if not (seed.get("size") or seed.get("min_size") or seed.get("max_size")):
|
||||
if not declares_shape(seed):
|
||||
continue
|
||||
# <bios>.nvm is read beside the image the core selected, never on
|
||||
# its own: its files come with each scanned image (_companions).
|
||||
if _companion_of(profile, name)[1] is not None:
|
||||
continue
|
||||
directories = placeholder_directories(
|
||||
seed, emu_name, profile, db,
|
||||
@@ -780,7 +791,7 @@ def _placeholder_scan(
|
||||
path = entry.get("path", "")
|
||||
if "/" not in path or path.rsplit("/", 1)[0] not in directories:
|
||||
continue
|
||||
if "/.variants/" in path or not size_fits(seed, entry.get("size", 0)):
|
||||
if "/.variants/" in path or not shape_fits(seed, path, entry.get("size", 0)):
|
||||
continue
|
||||
scan_name = entry.get("name", "")
|
||||
dest = f"{folder}/{scan_name}" if folder else scan_name
|
||||
@@ -788,6 +799,7 @@ def _placeholder_scan(
|
||||
"name": scan_name,
|
||||
"destination": dest,
|
||||
"sha1": sha1,
|
||||
"source_name": name,
|
||||
"source_system": seed.get("system"),
|
||||
"region": seed.get("region"),
|
||||
})
|
||||
@@ -821,16 +833,34 @@ def expand_agnostic_entries(
|
||||
db.get("indexes", {}).get("by_name", {}), set(), "",
|
||||
{emu_name} if standalone else set(),
|
||||
)
|
||||
# A placeholder is answered by files found in its own folder: PCEE2's
|
||||
# eeprom.dat, scanned at the root, said nothing about pcsx2/bios/.
|
||||
covered = {e["destination"].rsplit("/", 1)[0] if "/" in e["destination"] else "" for e in scanned}
|
||||
|
||||
# A placeholder is answered by the files its own scan found in its own
|
||||
# folder: PCEE2's eeprom.dat, scanned at the root, said nothing about
|
||||
# pcsx2/bios/, and its COH-H images, scanned into pcsx2/bios/, say
|
||||
# nothing about the console <bios>.bin beside them.
|
||||
def answered(extras: list[dict]) -> set[tuple[str, str]]:
|
||||
return {
|
||||
(str(e.get("source_name", "")),
|
||||
e["destination"].rsplit("/", 1)[0] if "/" in e["destination"] else "")
|
||||
for e in extras
|
||||
}
|
||||
|
||||
def is_answered(f: dict, covered: set[tuple[str, str]]) -> bool:
|
||||
# A companion placeholder (<bios>.rom1) is read beside its image, so
|
||||
# it is answered wherever the image placeholder is.
|
||||
_ext, image = _companion_of(profile, str(f.get("name", "")))
|
||||
names = {f.get("name", "")} | ({image.get("name", "")} if image else set())
|
||||
return any((n, folder(f)) in covered for n in names)
|
||||
|
||||
covered = answered(scanned)
|
||||
open_seeds = [
|
||||
f for f in files if "<" in str(f.get("name", "")) and folder(f) not in covered
|
||||
f for f in files
|
||||
if "<" in str(f.get("name", "")) and not is_answered(f, covered)
|
||||
]
|
||||
scanned += _placeholder_scan(
|
||||
open_seeds, emu_name, profile, db, standalone, all_profiles
|
||||
)
|
||||
covered = {e["destination"].rsplit("/", 1)[0] if "/" in e["destination"] else "" for e in scanned}
|
||||
covered = answered(scanned)
|
||||
if not scanned:
|
||||
return files
|
||||
concrete = [
|
||||
@@ -848,7 +878,7 @@ def expand_agnostic_entries(
|
||||
]
|
||||
return [
|
||||
f for f in files
|
||||
if "<" not in str(f.get("name", "")) or folder(f) not in covered
|
||||
if "<" not in str(f.get("name", "")) or not is_answered(f, covered)
|
||||
] + concrete
|
||||
|
||||
|
||||
@@ -913,6 +943,32 @@ def _companion_of(profile: dict, name: str) -> tuple[str, dict | None]:
|
||||
return "", None
|
||||
|
||||
|
||||
def _image_beside(
|
||||
companion_path: str, stem: str, image: dict | None, files_db: dict
|
||||
) -> bool:
|
||||
"""Whether the folder of a companion holds an image it is read beside.
|
||||
|
||||
PCSX2 opens `<bios>.rom1`, then the BIOS path with its extension
|
||||
replaced (BiosTools.cpp LoadExtraRom, CDVD.cpp cdvdLoadNVRAM): the
|
||||
companion's stem is the image's name or the image's own stem. The
|
||||
image has to pass the image entry's test, its size and its ROM
|
||||
directory, as the scan that selected it does.
|
||||
"""
|
||||
if image is None:
|
||||
return False
|
||||
directory = companion_path.rsplit("/", 1)[0] + "/"
|
||||
wanted = stem.lower()
|
||||
for _sha1, entry in _files_by_directory(files_db).get(directory, ()):
|
||||
name = str(entry.get("name", ""))
|
||||
if name.lower() != wanted and name.rpartition(".")[0].lower() != wanted:
|
||||
continue
|
||||
if name.lower() == companion_path.rsplit("/", 1)[-1].lower():
|
||||
continue
|
||||
if shape_fits(image, entry.get("path", ""), entry.get("size", 0)):
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def _companion_extensions(profile: dict, seed_name: str) -> set[str]:
|
||||
"""Extensions the profile declares beside a free-named BIOS entry.
|
||||
|
||||
|
||||
@@ -0,0 +1,162 @@
|
||||
"""The PS2 ROM directory test the PCSX2 family runs on every BIOS candidate.
|
||||
|
||||
A BIOS image starts with a table of 16-byte entries (a 10-byte name, a
|
||||
16-bit extended info size, a 32-bit file size) whose first entry is RESET.
|
||||
``LoadBiosVersion`` (pcsx2/ps2/BiosTools.cpp) reads the file 16 bytes at a
|
||||
time from its start until an entry named RESET, then walks the table from
|
||||
there, adding up the entry sizes rounded to 16 bytes to find where each
|
||||
file of the ROM lies, and reads the 14 bytes of ROMVER at its offset. The
|
||||
image is a BIOS when the walk met ROMVER.
|
||||
|
||||
The forks differ in how they compare the names and in what an unreadable
|
||||
read does, so a profile declares how its code reads the table (``romdir:``
|
||||
on the entry) and ``romdir_accepts`` runs that same walk. Stdlib only:
|
||||
install.py carries a copy of the walk.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import struct
|
||||
from dataclasses import dataclass
|
||||
|
||||
ENTRY_SIZE = 16
|
||||
NAME_SIZE = 10
|
||||
# for (i = 0; i < 512 * 1024; i++) in every LoadBiosVersion.
|
||||
SEARCH_ENTRIES = 512 * 1024
|
||||
ROMVER_SIZE = 14
|
||||
EXTINFO_SIZE = 15
|
||||
|
||||
# What a read the code cannot complete does, by fork.
|
||||
REJECT = "rejects" # wxWidgets streams throw (pxStreams.cpp, EndOfStream)
|
||||
ENDS_WALK = "ends_walk" # the walk breaks, the result stands
|
||||
IGNORED = "ignored" # AetherSX2 zero-fills romver and counts it found
|
||||
|
||||
SPEC_KEYS = frozenset({
|
||||
"reset_compare", "missing_reset", "terminated_names", "extinfo", "unreadable_romver",
|
||||
"truncated_walk", "offset_width", "refused_serials", "zones", "refused_zones",
|
||||
})
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Romdir:
|
||||
"""What a walk that met ROMVER read: its 14 bytes and the EXTINFO serial."""
|
||||
|
||||
romver: bytes
|
||||
serial: str
|
||||
|
||||
|
||||
def _name_is(name: bytes, wanted: bytes, compared: int) -> bool:
|
||||
"""strncmp(name, wanted, compared) == 0 over a 10-byte field."""
|
||||
a = name[:compared]
|
||||
b = (wanted + b"\x00")[:compared]
|
||||
for x, y in zip(a, b.ljust(len(a), b"\x00")):
|
||||
if x != y:
|
||||
return False
|
||||
if x == 0:
|
||||
return True
|
||||
return True
|
||||
|
||||
|
||||
def walk(data: bytes, spec: dict) -> Romdir | None:
|
||||
"""The ROM directory walk one fork's LoadBiosVersion runs, or None.
|
||||
|
||||
``reset_compare`` is the length strncmp compares RESET over: 5 in the
|
||||
wxWidgets and AetherSX2 code, which takes any name starting with RESET,
|
||||
sizeof(fileName) (10) after, which wants RESET and a NUL.
|
||||
``terminated_names`` stops the walk at a name with no NUL in its 10
|
||||
bytes (strnlen check of the later code). ``missing_reset`` says what
|
||||
512 Ki entries read without RESET do: ``rejects`` where the code throws
|
||||
(wxWidgets), ``ends_walk`` where the loop just ends and the walk starts
|
||||
at the last entry read. ``extinfo`` reads 15 bytes at
|
||||
the EXTINFO offset + 0x10 and ends the walk when they are not there.
|
||||
``unreadable_romver`` and ``truncated_walk`` say what a ROMVER read and
|
||||
a next-entry read past the end of the file do: ``rejects``,
|
||||
``ends_walk`` or, for ROMVER, ``ignored``. ``offset_width`` is the
|
||||
width of the offset sum: 32 where it is a uint (wxWidgets code), 64
|
||||
where it is an s64.
|
||||
"""
|
||||
size = len(data)
|
||||
reset_compare = int(spec.get("reset_compare", NAME_SIZE))
|
||||
terminated = bool(spec.get("terminated_names", True))
|
||||
extinfo = bool(spec.get("extinfo", True))
|
||||
unreadable = spec.get("unreadable_romver", ENDS_WALK)
|
||||
truncated = spec.get("truncated_walk", ENDS_WALK)
|
||||
missing_reset = spec.get("missing_reset", ENDS_WALK)
|
||||
offset_mask = (1 << int(spec.get("offset_width", 64))) - 1
|
||||
|
||||
pos = 0
|
||||
for _ in range(SEARCH_ENTRIES):
|
||||
if pos + ENTRY_SIZE > size:
|
||||
return None
|
||||
name = data[pos:pos + NAME_SIZE]
|
||||
pos += ENTRY_SIZE
|
||||
if _name_is(name, b"RESET", reset_compare):
|
||||
break
|
||||
else:
|
||||
if missing_reset == REJECT:
|
||||
return None
|
||||
|
||||
entry = data[pos - ENTRY_SIZE:pos]
|
||||
offset = 0
|
||||
romver: bytes | None = None
|
||||
serial = ""
|
||||
while True:
|
||||
name = entry[:NAME_SIZE]
|
||||
if name[0] == 0 or (terminated and b"\x00" not in name):
|
||||
break
|
||||
if extinfo and _name_is(name, b"EXTINFO", NAME_SIZE):
|
||||
start = offset + 0x10
|
||||
if start + EXTINFO_SIZE > size:
|
||||
break
|
||||
serial = data[start:start + EXTINFO_SIZE].split(b"\x00", 1)[0].decode("latin-1")
|
||||
if _name_is(name, b"ROMVER", NAME_SIZE):
|
||||
if offset + ROMVER_SIZE > size:
|
||||
if unreadable == REJECT:
|
||||
return None
|
||||
if unreadable == ENDS_WALK:
|
||||
break
|
||||
romver = bytes(ROMVER_SIZE)
|
||||
else:
|
||||
romver = data[offset:offset + ROMVER_SIZE]
|
||||
(file_size,) = struct.unpack_from("<I", entry, 12)
|
||||
offset += file_size if file_size % 0x10 == 0 else (file_size + 0x10) & 0xFFFFFFF0
|
||||
offset &= offset_mask
|
||||
if pos + ENTRY_SIZE > size:
|
||||
if truncated == REJECT:
|
||||
return None
|
||||
break
|
||||
entry = data[pos:pos + ENTRY_SIZE]
|
||||
pos += ENTRY_SIZE
|
||||
if romver is None:
|
||||
return None
|
||||
return Romdir(romver, serial)
|
||||
|
||||
|
||||
def zone_code(romver: bytes) -> str:
|
||||
"""The zone characters LoadBiosVersion switches on: ROMVER[4], and [5]
|
||||
where the code reads it to tell COH-H (T then Z) from T10K."""
|
||||
return romver[4:6].decode("latin-1")
|
||||
|
||||
|
||||
def romdir_accepts(data: bytes, spec: dict) -> bool:
|
||||
"""Whether a candidate passes the profile's ROM directory test.
|
||||
|
||||
Past the walk, ``refused_serials`` turns away an EXTINFO serial
|
||||
(ARMSX2's NOT_PS2_BIOS_STAMPS), and ``zones``/``refused_zones`` keep or
|
||||
drop an image by its ROMVER zone, as a scan that wants a COH-H board
|
||||
or passes one over does. A zone is written as the code tests it: ``J``
|
||||
is ROMVER[4], ``TZ`` ROMVER[4] and [5].
|
||||
"""
|
||||
found = walk(data, spec)
|
||||
if found is None:
|
||||
return False
|
||||
if found.serial in set(spec.get("refused_serials") or ()):
|
||||
return False
|
||||
code = zone_code(found.romver)
|
||||
|
||||
def matches(zones: list | None) -> bool:
|
||||
return any(code.startswith(str(z)) for z in zones or ())
|
||||
|
||||
if spec.get("zones") and not matches(spec["zones"]):
|
||||
return False
|
||||
return not matches(spec.get("refused_zones"))
|
||||
@@ -95,7 +95,9 @@ _CHECK_FIELDS = {
|
||||
"md5": ("md5",),
|
||||
"sha1": ("sha1",),
|
||||
"sha256": ("sha256",),
|
||||
"size": ("size", "min_size", "max_size"),
|
||||
"size": ("size", "min_size", "max_size", "size_multiple"),
|
||||
"romdir": ("romdir",),
|
||||
"masked_sha1": ("masked_sha1",),
|
||||
}
|
||||
|
||||
|
||||
|
||||
+44
-3
@@ -15,10 +15,14 @@ from collections.abc import Callable, Iterator, Mapping, Sequence
|
||||
from common import (
|
||||
adler32_of,
|
||||
archive_members,
|
||||
CONTENT_CHECKS,
|
||||
compute_hashes,
|
||||
content_fits,
|
||||
declared_content_checks,
|
||||
entry_system_ids,
|
||||
name_match_size_ok,
|
||||
sanitize_pack_path,
|
||||
shape_fits,
|
||||
size_fits,
|
||||
)
|
||||
from hashing import md5_composite, parse_hash_list, parse_md5_list
|
||||
@@ -35,6 +39,11 @@ def _adler32_byteswapped(path: str) -> str:
|
||||
return adler32_of(path, byteswap=True)
|
||||
|
||||
|
||||
_CONTENT_REASONS = {
|
||||
"romdir": "romdir mismatch: no RESET and ROMVER the code accepts",
|
||||
"masked_sha1": "masked_sha1 mismatch: the cleared image is not the one the code names",
|
||||
}
|
||||
|
||||
# All reproducible validation types.
|
||||
_HASH_CHECKS = frozenset({"crc32", "md5", "sha1", "adler32"})
|
||||
|
||||
@@ -75,6 +84,7 @@ def _entry_rule(emu_name: str, f: dict, checks: set[str]) -> dict:
|
||||
- {None},
|
||||
"min_size": f.get("min_size") if "size" in checks else None,
|
||||
"max_size": f.get("max_size") if "size" in checks else None,
|
||||
"size_multiple": f.get("size_multiple") if "size" in checks else None,
|
||||
"crc32": values("crc32"),
|
||||
"md5": values("md5"),
|
||||
"sha1": values("sha1"),
|
||||
@@ -84,6 +94,9 @@ def _entry_rule(emu_name: str, f: dict, checks: set[str]) -> dict:
|
||||
},
|
||||
"adler32_byteswap": bool(f.get("adler32_byteswap")),
|
||||
"crypto": {c for c in checks if c in _CRYPTO_CHECKS},
|
||||
"content": {
|
||||
check: f.get(check) or {} for check in sorted(CONTENT_CHECKS & set(checks))
|
||||
},
|
||||
# The checks whose failure the code only warns about: the file runs.
|
||||
"warn": warned_checks(f, checks),
|
||||
}
|
||||
@@ -122,6 +135,8 @@ def _rule_failure(
|
||||
return f"size too small: min {rule['min_size']}, got {actual_size}"
|
||||
if rule["max_size"] is not None and actual_size > rule["max_size"]:
|
||||
return f"size too large: max {rule['max_size']}, got {actual_size}"
|
||||
if rule.get("size_multiple") and actual_size % int(rule["size_multiple"]):
|
||||
return f"size mismatch: got {actual_size}, accepted multiples of {rule['size_multiple']}"
|
||||
for hash_type in ("crc32", "md5", "sha1", "sha256"):
|
||||
if rule[hash_type] and hash_type not in skip:
|
||||
if not hashes:
|
||||
@@ -139,6 +154,9 @@ def _rule_failure(
|
||||
if actual not in rule["adler32"]:
|
||||
accepted = ",".join(sorted(rule["adler32"]))
|
||||
return f"adler32 mismatch: got 0x{actual}, accepted [{accepted}]"
|
||||
for check, spec in rule.get("content", {}).items():
|
||||
if check not in skip and not content_fits({"validation": [check], check: spec}, local_path):
|
||||
return _CONTENT_REASONS[check]
|
||||
if rule["crypto"] - skip:
|
||||
from crypto_verify import check_crypto_validation
|
||||
|
||||
@@ -253,6 +271,7 @@ def _build_validation_index(profiles: dict) -> dict[str, dict]:
|
||||
"sizes": set(),
|
||||
"min_size": None,
|
||||
"max_size": None,
|
||||
"size_multiples": set(),
|
||||
"crc32": set(),
|
||||
"md5": set(),
|
||||
"sha1": set(),
|
||||
@@ -288,6 +307,8 @@ def _build_validation_index(profiles: dict) -> dict[str, dict]:
|
||||
index[fname]["min_size"] = (
|
||||
min(cur, f["min_size"]) if cur is not None else f["min_size"]
|
||||
)
|
||||
if f.get("size_multiple"):
|
||||
index[fname]["size_multiples"].add(int(f["size_multiple"]))
|
||||
if f.get("max_size") is not None:
|
||||
cur = index[fname]["max_size"]
|
||||
index[fname]["max_size"] = (
|
||||
@@ -325,7 +346,7 @@ def _build_validation_index(profiles: dict) -> dict[str, dict]:
|
||||
# Per-emulator ground truth detail
|
||||
expected: dict = {}
|
||||
if "size" in checks:
|
||||
for key in ("size", "min_size", "max_size"):
|
||||
for key in ("size", "min_size", "max_size", "size_multiple"):
|
||||
if f.get(key) is not None:
|
||||
expected[key] = f[key]
|
||||
for hash_type in ("crc32", "md5", "sha1", "sha256"):
|
||||
@@ -334,6 +355,8 @@ def _build_validation_index(profiles: dict) -> dict[str, dict]:
|
||||
adler_val_pe = f.get("known_hash_adler32") or f.get("adler32")
|
||||
if adler_val_pe:
|
||||
expected["adler32"] = adler_val_pe
|
||||
for check in sorted(CONTENT_CHECKS & set(checks)):
|
||||
expected[check] = f.get(check) or {}
|
||||
pe_entry = {
|
||||
"checks": sorted(checks),
|
||||
"source_ref": f.get("source_ref"),
|
||||
@@ -479,6 +502,20 @@ def check_file_validation(
|
||||
if entry["max_size"] is not None and actual_size > entry["max_size"]:
|
||||
emus = _emulators_for_check("size", pe)
|
||||
return f"size too large: max {entry['max_size']}, got {actual_size}", emus
|
||||
multiples = entry.get("size_multiples") or set()
|
||||
if multiples and all(actual_size % m for m in multiples):
|
||||
emus = _emulators_for_check("size", pe)
|
||||
accepted = ",".join(str(m) for m in sorted(multiples))
|
||||
return f"size mismatch: got {actual_size}, accepted multiples of [{accepted}]", emus
|
||||
|
||||
# A content test passes when one of the entries declaring it accepts
|
||||
# the file, each read as its own code reads it.
|
||||
for check in sorted(CONTENT_CHECKS & set(checks)):
|
||||
specs = [r["content"][check] for r in entry.get("rules", []) if check in r.get("content", {})]
|
||||
if specs and not any(
|
||||
content_fits({"validation": [check], check: spec}, local_path) for spec in specs
|
||||
):
|
||||
return _CONTENT_REASONS[check], _emulators_for_check(check, pe)
|
||||
|
||||
# Hash checks: compute once, reuse for all hash types.
|
||||
# Each hash field is a set of accepted values (multiple valid ROM versions).
|
||||
@@ -855,8 +892,8 @@ def agnostic_substitute(
|
||||
held = candidate.get("path", "")
|
||||
if (
|
||||
held.startswith(prefix)
|
||||
and size_fits(entry, candidate.get("size", 0))
|
||||
and os.path.exists(held)
|
||||
and shape_fits(entry, held, candidate.get("size", 0))
|
||||
):
|
||||
return held, prefix
|
||||
break
|
||||
@@ -1148,8 +1185,12 @@ def core_size_refusal(
|
||||
if not local_path or not os.path.isfile(local_path):
|
||||
return None
|
||||
size = os.path.getsize(local_path)
|
||||
if any(name_match_size_ok(entry, size) for entry in (file_entry, *alternatives)):
|
||||
sized = [entry for entry in (file_entry, *alternatives) if name_match_size_ok(entry, size)]
|
||||
if any(content_fits(entry, local_path) for entry in sized):
|
||||
return None
|
||||
if sized:
|
||||
tests = sorted({check for entry in sized for check in declared_content_checks(entry)})
|
||||
return f"the core's {', '.join(tests)} test turns it away"
|
||||
bounds = (
|
||||
f"{file_entry['size']}" if file_entry.get("size") is not None
|
||||
else f"{file_entry.get('min_size', 0)} to {file_entry.get('max_size', 'any')}"
|
||||
|
||||
+1
-1
@@ -400,7 +400,7 @@ def _build_expected(file_entry: dict, checks: list[str]) -> dict:
|
||||
if not checks:
|
||||
return expected
|
||||
if "size" in checks:
|
||||
for key in ("size", "min_size", "max_size"):
|
||||
for key in ("size", "min_size", "max_size", "size_multiple"):
|
||||
if file_entry.get(key) is not None:
|
||||
expected[key] = file_entry[key]
|
||||
for hash_type in ("crc32", "md5", "sha1", "sha256"):
|
||||
|
||||
Reference in new issue
Block a user