diff --git a/scripts/auto_fetch.py b/scripts/auto_fetch.py index cb04c680..c39587b7 100644 --- a/scripts/auto_fetch.py +++ b/scripts/auto_fetch.py @@ -25,6 +25,7 @@ import sys import urllib.error import urllib.parse import urllib.request +import zlib from pathlib import Path sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) @@ -101,6 +102,8 @@ def find_missing(config: dict, db: dict) -> list[dict]: "system": sys_id, "sha1": file_entry.get("sha1"), "md5": file_entry.get("md5"), + "sha256": file_entry.get("sha256"), + "crc32": file_entry.get("crc32"), "size": file_entry.get("size"), "destination": file_entry.get("destination", name), }) @@ -112,7 +115,9 @@ def find_missing(config: dict, db: dict) -> list[dict]: "system": u.get("system") or "", "sha1": u.get("sha1"), "md5": u.get("md5"), - "size": None, + "sha256": u.get("sha256"), + "crc32": u.get("crc32"), + "size": u.get("size"), "destination": u.get("path") or u["name"], "emulator": u.get("emulator", ""), }) @@ -131,14 +136,45 @@ def _accepted(entry: dict, field: str) -> set[str]: return {str(v).strip().lower() for v in values if str(v).strip()} +def _crc32s(entry: dict) -> set[str]: + return {value.removeprefix("0x").zfill(8) for value in _accepted(entry, "crc32")} + + +def _size(entry: dict) -> int | None: + try: + return int(entry["size"]) if entry.get("size") is not None else None + except (TypeError, ValueError): + return None + + +def verifiable(entry: dict) -> bool: + """Whether a download can be proved to be this file. + + A sha1, sha256 or md5 proves it, a crc32 only with the size beside it, + the way the resolver accepts one. Without any of them a search can find + a file of that name and never accept it. + """ + if any(_accepted(entry, field) for field in ("sha1", "sha256", "md5")): + return True + return bool(_crc32s(entry)) and _size(entry) is not None + + def verify_content(data: bytes, expected: dict) -> bool: - """Verify downloaded content matches one of the expected hashes.""" - sha1s = _accepted(expected, "sha1") - if sha1s: - return hashlib.sha1(data).hexdigest() in sha1s - md5s = _accepted(expected, "md5") - if md5s: - return hashlib.md5(data).hexdigest() in md5s + """Verify downloaded content matches one of the expected hashes. + + The strongest hash the entry declares decides. + """ + for field, digest in ( + ("sha1", hashlib.sha1), + ("sha256", hashlib.sha256), + ("md5", hashlib.md5), + ): + accepted = _accepted(expected, field) + if accepted: + return digest(data).hexdigest() in accepted + crcs, size = _crc32s(expected), _size(expected) + if crcs and size is not None: + return len(data) == size and f"{zlib.crc32(data) & 0xFFFFFFFF:08x}" in crcs return False @@ -158,11 +194,20 @@ def step1_crossref_db(entry: dict, db: dict) -> str | None: for sha1 in sorted(_accepted(entry, "sha1")): if sha1 in files: return files[sha1]["path"] - by_md5 = db.get("indexes", {}).get("by_md5", {}) - for md5 in sorted(_accepted(entry, "md5")): - sha1_match = by_md5.get(md5) - if sha1_match and sha1_match in files: - return files[sha1_match]["path"] + indexes = db.get("indexes", {}) + for field, index in (("sha256", "by_sha256"), ("md5", "by_md5")): + by_hash = indexes.get(index, {}) + for value in sorted(_accepted(entry, field)): + sha1_match = by_hash.get(value) + if sha1_match and sha1_match in files: + return files[sha1_match]["path"] + size = _size(entry) + if size is not None: + by_crc32 = indexes.get("by_crc32", {}) + for crc in sorted(_crc32s(entry)): + sha1_match = by_crc32.get(crc) + if sha1_match in files and files[sha1_match].get("size") == size: + return files[sha1_match]["path"] return None @@ -352,7 +397,7 @@ def fetch_missing( dry_run: bool = False, ) -> dict: """Run the 5-step auto-fetch pipeline for missing files.""" - stats = {"found": 0, "not_found": 0, "errors": 0} + stats = {"found": 0, "not_found": 0, "errors": 0, "unverifiable": 0} still_missing = [] for entry in missing: @@ -366,6 +411,14 @@ def fetch_missing( stats["found"] += 1 continue + if not verifiable(entry): + # Searched, a file of that name could only ever be refused. + print(" [-] No sha1, sha256, md5 or crc32 with size to check a " + "download against: not searched") + still_missing.append(entry) + stats["unverifiable"] += 1 + continue + if dry_run: print(" [DRY RUN] Would search branches, repos, archive.org") still_missing.append(entry) @@ -511,7 +564,8 @@ def main() -> None: stats = result["stats"] print( f"\n Results: {stats['found']} found, {stats['not_found']} not found, " - f"{stats['errors']} not fully searched" + f"{stats['errors']} not fully searched, " + f"{stats['unverifiable']} without a hash to search by" ) if args.create_issues and all_still_missing: diff --git a/tests/test_audit_regressions.py b/tests/test_audit_regressions.py index f5afb746..a263fb3e 100644 --- a/tests/test_audit_regressions.py +++ b/tests/test_audit_regressions.py @@ -5598,6 +5598,52 @@ class AnUnansweredSearchIsNotANotFound(unittest.TestCase): self.assertEqual(result["stats"]["errors"], 0) +class AutoFetchProvesEveryHashItIsGiven(unittest.TestCase): + """auto_fetch checked a download by sha1 or md5 only: RetroDECK's + dsifirmware.bin, pinned by sha256 alone, was announced as searched and + could never be accepted.""" + + DATA = b"firmware bytes" + + def test_sha256_and_crc32_with_size_prove_a_download(self): + import hashlib + import zlib + + from scripts import auto_fetch + + sha256 = {"sha256": hashlib.sha256(self.DATA).hexdigest()} + self.assertTrue(auto_fetch.verify_content(self.DATA, sha256)) + self.assertFalse(auto_fetch.verify_content(b"other", sha256)) + crc = {"crc32": f"{zlib.crc32(self.DATA):08X}", "size": len(self.DATA)} + self.assertTrue(auto_fetch.verify_content(self.DATA, crc)) + self.assertFalse(auto_fetch.verify_content(self.DATA, dict(crc, size=1))) + self.assertFalse(auto_fetch.verifiable({"crc32": crc["crc32"]})) + + def test_a_held_sha256_is_found_in_the_database(self): + from scripts import auto_fetch + + db = {"files": {"s": {"path": "bios/x/dsi.bin", "size": 3}}, + "indexes": {"by_sha256": {"f" * 64: "s"}}} + self.assertEqual( + auto_fetch.step1_crossref_db({"sha256": "F" * 64}, db), "bios/x/dsi.bin" + ) + + def test_an_entry_without_a_hash_is_not_searched(self): + from unittest import mock + + from scripts import auto_fetch + + entry = {"name": "DICROM.ROM", "system": "fujitsu-fm7"} + out = io.StringIO() + with mock.patch.object(auto_fetch, "step2_scan_branches") as branches, \ + contextlib.redirect_stdout(out): + result = auto_fetch.fetch_missing([entry], {"files": {}}, "bios", dry_run=True) + branches.assert_not_called() + self.assertEqual(result["stats"]["unverifiable"], 1) + self.assertNotIn("Would search", out.getvalue()) + self.assertIn("not searched", out.getvalue()) + + class AForcedRehashIsCheckedAgainstTheLastOne(unittest.TestCase): """This host has computed a wrong SHA-1 under load. The pipeline's forced rebuild threw the previous digests away and wrote whatever it read.""" diff --git a/wiki/tools.md b/wiki/tools.md index 173caa6f..51be8bab 100644 --- a/wiki/tools.md +++ b/wiki/tools.md @@ -653,7 +653,7 @@ python scripts/refresh_stale.py --only platforms,targets --jobs 6 | `region_audit.py` | Confront every profile `region:` with the regions No-Intro names for the same dump (`--json`, `--db`, `--emulators-dir`). An opinion, never an authority: the code decides, and a disagreement is a line to read in the emulator's source | | `validate_pr.py` | Validate BIOS files in pull requests, post markdown report | | `validate_schemas.py` | Validate the data contracts: schemas and semantic invariants. `--source-only` checks `emulators/` and `platforms/` alone, which is what PR validation runs | -| `auto_fetch.py` | Fetch missing BIOS files from known sources (4-step pipeline) | +| `auto_fetch.py` | Fetch missing BIOS files from known sources (4-step pipeline). A download is accepted by the sha1, sha256 or md5 the entry declares, or its crc32 with the size. An entry with none of them is reported and not searched | | `list_platforms.py` | List active platforms (`--include-archived` adds the archived ones, used by CI) | | `download.py` | Download a pack from GitHub releases, every part checked (Python, stdlib only) | | `download.sh` | Same, as a shell one-liner (`curl` + `unzip`) |