Files
libretro/scripts/validation.py
T

719 lines
29 KiB
Python

"""Emulator-level file validation logic.
Builds validation indexes from emulator profiles, checks files against
emulator-declared constraints (size, hash, crypto), and formats ground
truth data for reporting.
"""
from __future__ import annotations
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.
_CRYPTO_CHECKS = frozenset({"signature", "crypto"})
def _adler32_byteswapped(path: str) -> str:
"""Compute adler32 on 16-bit byte-swapped data.
Dolphin's DSP loader swaps every 16-bit word before hashing
(Common::swap16 in DSPLLE.cpp:LoadDSPRom). This reproduces that
transform so verify.py can match the expected adler32 values.
"""
import struct
import zlib
with open(path, "rb") as f:
data = f.read()
# Pad to even length if necessary
if len(data) % 2:
data += b"\x00"
swapped = struct.pack(f">{len(data) // 2}H", *struct.unpack(f"<{len(data) // 2}H", data))
return format(zlib.adler32(swapped) & 0xFFFFFFFF, "08x")
# All reproducible validation types.
_HASH_CHECKS = frozenset({"crc32", "md5", "sha1", "adler32"})
def _parse_validation(validation: list | dict | None) -> list[str]:
"""Extract the validation check list from a file's validation field.
Handles both simple list and divergent (core/upstream) dict forms.
For dicts, uses the ``core`` key since RetroArch users run the core.
"""
if validation is None:
return []
if isinstance(validation, list):
return validation
if isinstance(validation, dict):
return validation.get("core", [])
return []
def _entry_rule(emu_name: str, f: dict, checks: set[str]) -> dict:
"""One profile entry's own checks, kept apart from the merged view.
Merged by name, Azahar's crypto check on the 3DS otp.bin judged Cemu's
1 KiB Wii U otp.bin, and the variant search handed Cemu the 3DS file.
"""
def values(field: str) -> set[str]:
raw = f.get(field)
if not raw or field not in checks:
return set()
items = raw if isinstance(raw, list) else str(raw).split(",")
out = set()
for item in items:
norm = str(item).strip().lower()
if field == "crc32" and norm.startswith("0x"):
norm = norm[2:]
if norm:
out.add(norm)
return out
raw_size = f.get("size") if "size" in checks else None
adler = f.get("known_hash_adler32") or f.get("adler32")
adler_values = adler if isinstance(adler, list) else [adler] if adler else []
return {
"emulator": emu_name,
"checks": set(checks),
"sizes": set(raw_size if isinstance(raw_size, list) else [raw_size])
- {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,
"crc32": values("crc32"),
"md5": values("md5"),
"sha1": values("sha1"),
"sha256": values("sha256"),
"adler32": {
str(v).lower().removeprefix("0x") for v in adler_values if v
},
"adler32_byteswap": bool(f.get("adler32_byteswap")),
"crypto": {c for c in checks if c in _CRYPTO_CHECKS},
}
def _rule_failure(
rule: dict, local_path: str, filename: str, bios_dir: str, hashes: dict,
) -> str | None:
"""Why one entry's checks reject a file, or None when they accept it."""
actual_size = os.path.getsize(local_path)
if rule["sizes"] and actual_size not in rule["sizes"]:
accepted = ",".join(str(s) for s in sorted(rule["sizes"]))
return f"size mismatch: got {actual_size}, accepted [{accepted}]"
if rule["min_size"] is not None and actual_size < rule["min_size"]:
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}"
for hash_type in ("crc32", "md5", "sha1", "sha256"):
if rule[hash_type]:
if not hashes:
hashes.update(compute_hashes(local_path))
if hashes[hash_type].lower() not in rule[hash_type]:
accepted = ",".join(sorted(rule[hash_type]))
return f"{hash_type} mismatch: got {hashes[hash_type]}, accepted [{accepted}]"
if rule["adler32"]:
if rule["adler32_byteswap"]:
actual = _adler32_byteswapped(local_path)
else:
if not hashes:
hashes.update(compute_hashes(local_path))
actual = hashes["adler32"].lower()
if actual not in rule["adler32"]:
accepted = ",".join(sorted(rule["adler32"]))
return f"adler32 mismatch: got 0x{actual}, accepted [{accepted}]"
if rule["crypto"]:
from crypto_verify import check_crypto_validation
return check_crypto_validation(local_path, filename, bios_dir)
return None
def _build_validation_index(profiles: dict) -> dict[str, dict]:
"""Build per-filename validation rules from emulator profiles.
Returns {filename: {"checks": [str], "size": int|None, "min_size": int|None,
"max_size": int|None, "crc32": str|None, "md5": str|None, "sha1": str|None,
"adler32": str|None, "crypto_only": [str], "per_emulator": {emu: detail}}}.
``crypto_only`` lists validation types we cannot reproduce (signature, crypto)
so callers can report them as non-verifiable rather than silently skipping.
``per_emulator`` preserves each core's individual checks, source_ref, and
expected values before merging, for ground truth reporting.
When multiple emulators reference the same file, merges checks (union).
Raises ValueError if two profiles declare conflicting values.
"""
index: dict[str, dict] = {}
for emu_name, profile in profiles.items():
if profile.get("type") in ("launcher", "alias"):
continue
for f in profile.get("files", []):
fname = f.get("name", "")
if not fname:
continue
checks = _parse_validation(f.get("validation"))
if not checks:
continue
if fname not in index:
index[fname] = {
"checks": set(),
"sizes": set(),
"min_size": None,
"max_size": None,
"crc32": set(),
"md5": set(),
"sha1": set(),
"sha256": set(),
"adler32": set(),
"adler32_byteswap": False,
"crypto_only": set(),
"emulators": set(),
"per_emulator": {},
"rules": [],
}
index[fname]["rules"].append(_entry_rule(emu_name, f, checks))
index[fname]["emulators"].add(emu_name)
index[fname]["checks"].update(checks)
# Track non-reproducible crypto checks
index[fname]["crypto_only"].update(c for c in checks if c in _CRYPTO_CHECKS)
# Size checks
if "size" in checks:
raw_size = f.get("size")
if raw_size is not None:
if isinstance(raw_size, list):
index[fname]["sizes"].update(raw_size)
else:
index[fname]["sizes"].add(raw_size)
if f.get("min_size") is not None:
cur = index[fname]["min_size"]
index[fname]["min_size"] = (
min(cur, f["min_size"]) if cur is not None else f["min_size"]
)
if f.get("max_size") is not None:
cur = index[fname]["max_size"]
index[fname]["max_size"] = (
max(cur, f["max_size"]) if cur is not None else f["max_size"]
)
# Hash checks -collect all accepted hashes as sets (multiple valid
# versions of the same file, e.g. MT-32 ROM versions)
if "crc32" in checks and f.get("crc32"):
crc_val = f["crc32"]
crc_list = crc_val if isinstance(crc_val, list) else [crc_val]
for cv in crc_list:
norm = str(cv).lower()
if norm.startswith("0x"):
norm = norm[2:]
index[fname]["crc32"].add(norm)
# A hash field may carry several accepted values, as a list or as
# the comma-separated string the schema spells out
for hash_type in ("md5", "sha1", "sha256"):
if hash_type in checks and f.get(hash_type):
val = f[hash_type]
values = val if isinstance(val, list) else str(val).split(",")
for h in values:
if str(h).strip():
index[fname][hash_type].add(str(h).strip().lower())
# Adler32 -stored as known_hash_adler32 field (not in validation: list
# for Dolphin, but support it in both forms for future profiles)
adler_val = f.get("known_hash_adler32") or f.get("adler32")
for value in adler_val if isinstance(adler_val, list) else [adler_val] if adler_val else []:
norm = str(value).lower()
if norm.startswith("0x"):
norm = norm[2:]
index[fname]["adler32"].add(norm)
if f.get("adler32_byteswap"):
index[fname]["adler32_byteswap"] = True
# Per-emulator ground truth detail
expected: dict = {}
if "size" in checks:
for key in ("size", "min_size", "max_size"):
if f.get(key) is not None:
expected[key] = f[key]
for hash_type in ("crc32", "md5", "sha1", "sha256"):
if hash_type in checks and f.get(hash_type):
expected[hash_type] = f[hash_type]
adler_val_pe = f.get("known_hash_adler32") or f.get("adler32")
if adler_val_pe:
expected["adler32"] = adler_val_pe
pe_entry = {
"checks": sorted(checks),
"source_ref": f.get("source_ref"),
"expected": expected,
}
pe = index[fname]["per_emulator"]
if emu_name in pe:
# Merge checks from multiple file entries for same emulator
existing = pe[emu_name]
merged_checks = sorted(
set(existing["checks"]) | set(pe_entry["checks"])
)
existing["checks"] = merged_checks
existing["expected"].update(pe_entry["expected"])
if pe_entry["source_ref"] and not existing["source_ref"]:
existing["source_ref"] = pe_entry["source_ref"]
else:
pe[emu_name] = pe_entry
# Convert sets to sorted tuples/lists for determinism
for v in index.values():
v["checks"] = sorted(v["checks"])
v["crypto_only"] = sorted(v["crypto_only"])
v["emulators"] = sorted(v["emulators"])
# Keep hash sets as frozensets for O(1) lookup in check_file_validation
return index
def build_ground_truth(filename: str, validation_index: dict[str, dict]) -> list[dict]:
"""Format per-emulator ground truth for a file from the validation index.
Returns a sorted list of {emulator, checks, source_ref, expected} dicts.
Returns [] if the file has no emulator validation data.
"""
entry = validation_index.get(filename)
if not entry or not entry.get("per_emulator"):
return []
result = []
for emu_name in sorted(entry["per_emulator"]):
detail = entry["per_emulator"][emu_name]
result.append(
{
"emulator": emu_name,
"checks": detail["checks"],
"source_ref": detail.get("source_ref"),
"expected": detail.get("expected", {}),
}
)
return result
def _emulators_for_check(
check_type: str, per_emulator: dict[str, dict],
) -> list[str]:
"""Return emulator names that validate a specific check type."""
result = []
for emu, detail in per_emulator.items():
emu_checks = detail.get("checks", [])
if check_type in emu_checks:
result.append(emu)
# adler32 is stored as known_hash, not always in validation list
if check_type == "adler32" and detail.get("expected", {}).get("adler32"):
if emu not in result:
result.append(emu)
return sorted(result)
def check_file_validation(
local_path: str,
filename: str,
validation_index: dict[str, dict],
bios_dir: str = "bios",
emulators: set[str] | None = None,
) -> tuple[str, list[str]] | None:
"""Check emulator-level validation on a resolved file.
Supports: size (exact/min/max), crc32, md5, sha1, adler32,
signature (RSA-2048 PKCS1v15 SHA256), crypto (AES-128-CBC + SHA256).
With ``emulators``, the cores that read the file's destination, each of
them must accept it through one of its own entries for the name, the
way each runs its own check. Without, the checks of every core naming
the file are merged.
Returns None if all checks pass or no validation applies.
Returns (reason, emulators) tuple on failure, where *emulators*
lists only those cores whose check actually failed.
"""
entry = validation_index.get(filename)
if not entry:
return None
if emulators is not None:
hashes: dict = {}
failures: dict[str, str] = {}
for emu in sorted(emulators):
rules = [r for r in entry.get("rules", []) if r["emulator"] == emu]
if not rules:
continue
reasons = [
_rule_failure(rule, local_path, filename, bios_dir, hashes)
for rule in rules
]
if all(reasons):
failures[emu] = reasons[0]
if not failures:
return None
first = sorted(failures)[0]
return failures[first], sorted(failures)
checks = entry["checks"]
pe = entry.get("per_emulator", {})
# Size checks -sizes is a set of accepted values
if "size" in checks:
actual_size = os.path.getsize(local_path)
if entry["sizes"] and actual_size not in entry["sizes"]:
expected = ",".join(str(s) for s in sorted(entry["sizes"]))
emus = _emulators_for_check("size", pe)
return f"size mismatch: got {actual_size}, accepted [{expected}]", emus
if entry["min_size"] is not None and actual_size < entry["min_size"]:
emus = _emulators_for_check("size", pe)
return f"size too small: min {entry['min_size']}, got {actual_size}", emus
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
# Hash checks -compute once, reuse for all hash types.
# Each hash field is a set of accepted values (multiple valid ROM versions).
need_hashes = any(
h in checks and entry.get(h) for h in ("crc32", "md5", "sha1", "sha256")
) or entry.get("adler32")
if need_hashes:
hashes = compute_hashes(local_path)
for hash_type in ("crc32", "md5", "sha1", "sha256"):
if hash_type in checks and entry[hash_type]:
if hashes[hash_type].lower() not in entry[hash_type]:
expected = ",".join(sorted(entry[hash_type]))
emus = _emulators_for_check(hash_type, pe)
return (
f"{hash_type} mismatch: got {hashes[hash_type]}, "
f"accepted [{expected}]",
emus,
)
if entry["adler32"]:
actual_adler = hashes["adler32"].lower()
if entry.get("adler32_byteswap"):
actual_adler = _adler32_byteswapped(local_path)
if actual_adler not in entry["adler32"]:
expected = ",".join(sorted(entry["adler32"]))
emus = _emulators_for_check("adler32", pe)
return (
f"adler32 mismatch: got 0x{actual_adler}, accepted [{expected}]",
emus,
)
# Signature/crypto checks (3DS RSA, AES)
if entry["crypto_only"]:
from crypto_verify import check_crypto_validation
crypto_reason = check_crypto_validation(local_path, filename, bios_dir)
if crypto_reason:
emus = sorted(entry.get("emulators", []))
return crypto_reason, emus
return None
def validate_cli_modes(args, mode_attrs: list[str]) -> None:
"""Validate mutual exclusion of CLI mode arguments."""
modes = sum(1 for attr in mode_attrs if getattr(args, attr, None))
if modes == 0:
raise SystemExit(f"Specify one of: --{' --'.join(mode_attrs)}")
if modes > 1:
raise SystemExit(f"Options are mutually exclusive: --{' --'.join(mode_attrs)}")
def read_from_system_dir(file_entry: dict) -> bool:
"""Whether the emulator reads this entry from the system directory.
A file loaded from the save, content or working directory is not a BIOS:
no platform lists it, no pack carries it, and the truth model must not
ask a platform to declare it.
"""
load_from = file_entry.get("load_from", "")
return not load_from or load_from == "system_dir"
def outside_gap_scope(file_entry: dict, is_standalone: bool) -> bool:
"""Whether an entry is a file the platform's pack would carry for this build.
Read by the gap report and by verify's undeclared-file pass, so both set
aside the same entries: a placeholder standing for a family of files, a
file the user imports through the UI (``path: null``), an entry of the
build the platform does not run, a file read outside the system
directory, and a filename-agnostic entry the builder scans for itself.
"""
name = file_entry.get("name", "")
if "<" in name or ">" in name or "*" in name:
return True
if "path" in file_entry and file_entry["path"] is None:
return True
mode = file_entry.get("mode")
if (mode == "standalone" and not is_standalone) or (
mode == "libretro" and is_standalone
):
return True
return not read_from_system_dir(file_entry) or bool(file_entry.get("agnostic"))
def filter_files_by_mode(files: list[dict], standalone: bool) -> list[dict]:
"""Filter file entries by libretro/standalone mode."""
result = []
for f in files:
fmode = f.get("mode", "")
if standalone and fmode == "libretro":
continue
if not standalone and fmode == "standalone":
continue
result.append(f)
return result
def find_validated_variant(
file_entry: dict,
db: dict,
current_path: str,
validation_index: dict,
bios_dir: str = "bios",
platform_digest: str | None = None,
emulators: set[str] | None = None,
) -> str | None:
"""A held file the emulator's own checks accept, in place of current_path.
Candidates come first from the hashes the emulator declares, which finds
a dump stored under another name, then from the files sharing the name.
platform_digest is the hash the frontend compares ("md5", "sha1") or None
when it reads no bytes: a candidate must then also carry a value the
entry declares, since the frontend would reject anything else. The
report and the packs read this one function, so the file a report counts
as satisfying the emulator is the file a pack ships.
"""
fname = file_entry.get("name", "")
if not fname or fname not in validation_index:
return None
accepted: set[str] = set()
if platform_digest:
if file_entry.get("zipped_file"):
return None
declared = file_entry.get(platform_digest) or ""
if platform_digest == "md5":
accepted = set(parse_md5_list(declared))
elif isinstance(declared, str) and declared:
accepted = {declared.lower()}
elif isinstance(declared, list):
accepted = {str(value).lower() for value in declared}
files_db = db.get("files", {})
indexes = db.get("indexes", {})
current_real = os.path.realpath(current_path)
seen: set[str] = set()
def candidates():
expected = validation_index[fname]
for hash_type, index_key in (
("sha1", None), ("md5", "by_md5"), ("crc32", "by_crc32"), ("sha256", "by_sha256"),
):
for value in expected.get(hash_type) or []:
if index_key is None:
yield value
continue
found = indexes.get(index_key, {}).get(value)
if isinstance(found, str):
yield found
elif isinstance(found, list):
yield from found
yield from indexes.get("by_name", {}).get(fname, [])
for sha1 in candidates():
entry = files_db.get(sha1) or {}
path = entry.get("path", "")
if not path or not os.path.exists(path):
continue
real = os.path.realpath(path)
if real == current_real or real in seen:
continue
seen.add(real)
if accepted and str(entry.get(platform_digest, "")).lower() not in accepted:
continue
if check_file_validation(
path, fname, validation_index, bios_dir, emulators
) is None:
return path
return None
def agnostic_substitute(
file_entry: dict, sys_id: str, db: dict, platform_profiles: dict[str, dict]
) -> tuple[str, str] | None:
"""(path, directory) of a held file a filename-agnostic core can boot.
A core with `bios_mode: agnostic` (PCSX2 picks any image in its BIOS
folder) is served by any image of the right size, renamed to what the
platform declares. That only satisfies a frontend that checks existence:
a digest frontend compares the bytes. Only the platform's own cores are
asked, and the pack and verify read this one answer. An entry that
declares a content hash names one file: no substitute stands for it.
"""
if any(file_entry.get(h) for h in ("sha1", "md5", "sha256", "crc32")):
return None
by_name = db.get("indexes", {}).get("by_name", {})
files_db = db.get("files", {})
for profile in platform_profiles.values():
if profile.get("bios_mode") != "agnostic":
continue
if sys_id not in set(profile.get("systems", [])):
continue
for entry in profile.get("files", []):
for sha1 in by_name.get(entry.get("name", ""), []):
path = files_db.get(sha1, {}).get("path", "")
if not path:
continue
prefix = path.rsplit("/", 1)[0] + "/"
for candidate in files_db.values():
held = candidate.get("path", "")
if (
held.startswith(prefix)
and size_fits(entry, candidate.get("size", 0))
and os.path.exists(held)
):
return held, prefix
break
return None
def frontend_digest_matches(file_entry: dict, local_path: str, algorithm: str) -> bool:
"""Whether the frontend's own digest accepts the file.
resolve_local_file calls a file `hash_mismatch` when any declared hash
disagrees, but a frontend compares one digest. RomM's list carries md5,
sha1 and crc32: a mistyped crc32 next to the right md5 must not withhold
a file RomM accepts. Same comparison as verify_entry_md5/_sha1.
"""
if algorithm == "md5":
declared = parse_md5_list(file_entry.get("md5"))
else:
value = file_entry.get(algorithm)
values = value if isinstance(value, list) else [value]
declared = [str(v).strip().lower() for v in values if v]
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
def destination_owners(profiles: dict) -> dict[str, list[tuple[str, str]]]:
"""For each file name, the path tail each emulator reads it under."""
owners: dict[str, list[tuple[str, str]]] = {}
for emu_name, profile in profiles.items():
for f in profile.get("files", []):
name = str(f.get("name") or "")
if not name:
continue
for declared in {f.get("path"), f.get("standalone_path")} - {None, ""} or {name}:
owners.setdefault(name.lower(), []).append(
(str(declared).replace("\\", "/").lower(), emu_name)
)
return owners
def _owners_of(destination: str, name: str, owners: dict) -> set[str]:
"""Emulators whose declared path is the longest tail of the destination."""
destination = destination.replace("\\", "/").lower()
matching = [
(tail, emu) for tail, emu in owners.get(name.lower(), [])
if destination == tail or destination.endswith("/" + tail)
]
if not matching:
return set()
longest = max(len(tail) for tail, _emu in matching)
return {emu for tail, emu in matching if len(tail) == longest}
def validated_choice(
file_entry: dict,
local_path: str | None,
db: dict,
validation_index: dict,
bios_dir: str,
platform_digest: str | None,
destination: str = "",
owners: dict | None = None,
) -> tuple[str | None, str | None]:
"""The file a platform destination ships, and the disagreement if any.
A frontend's own check can pass while an emulator it runs rejects the
file; a held variant both accept replaces it. The pack and the install
manifest both read this, so they name the same file. A destination
another emulator declares more precisely is that emulator's: ZEsarUX's
48 KB cpc6128.rom check has no say over ep128emu/roms/cpc6128.rom.
"""
if not local_path or not validation_index:
return local_path, None
name = file_entry.get("name", "")
rules = validation_index.get(name)
judges: set[str] | None = None
if rules and owners and destination:
owning = _owners_of(destination, name, owners)
if not owning & set(rules["emulators"]):
# No core that checks this name reads this destination: PC-98's
# bios.rom rules have no say over roms/ibmpcjr/bios.rom.
return local_path, None
# The cores that read this destination judge it, each by its own
# entries.
judges = owning & set(rules["emulators"])
check = check_file_validation(
local_path, file_entry.get("name", ""), validation_index, bios_dir, judges
)
if not check:
return local_path, None
better = find_validated_variant(
file_entry, db, local_path, validation_index, bios_dir,
platform_digest=platform_digest, emulators=judges,
)
if better:
return better, None
reason, emulators = check
return local_path, f"{', '.join(emulators)} says {reason}"
def existence_discrepancy(file_entry: dict, local_path: str, platform_display: str) -> str:
"""What an existence platform is told when its declared hash contradicts
the file it loads: the file is kept, the gap is said out loud. The pack
and verify both print this one line."""
declared = file_entry.get("md5", "") or file_entry.get("sha1", "")
actual = compute_hashes(local_path)
return (
f"packed per {platform_display} existence check; declared hash {declared}, "
f"file md5 {actual['md5']} sha1 {actual['sha1']}"
)