diff --git a/scripts/artifacts.py b/scripts/artifacts.py new file mode 100644 index 00000000..0bc28eff --- /dev/null +++ b/scripts/artifacts.py @@ -0,0 +1,88 @@ +"""Writing generated files, and not writing them. + +A generated artefact carries a timestamp but must not be rewritten +when only the clock moved: the CI freshness guard is a git diff, and +it is only meaningful if the comparison ignores the hour.""" + +from __future__ import annotations + +import contextlib +import os +import re + + +_TIMESTAMP_PATTERNS = [ + re.compile(r'"generated_at":\s*"[^"]*"'), # database.json + re.compile(r'"imported_at":\s*"[^"]*"'), # provenance snapshots + re.compile(r"\*Auto-generated on [^*]*\*"), # README.md + re.compile(r"\*Generated on [^*]*\*"), # docs site pages + # The decorated pages carry the same stamp again as a rendered element, + # and missing it rewrote every page on every run for the clock alone. + re.compile(r'
[^<]*
'), +] + +def write_if_changed(path: str, content: str, normalize=None) -> bool: + """Write content to path only if the non-timestamp content differs. + + Compares new and existing content after stripping timestamp lines. + Returns True if the file was written, False if skipped (unchanged). + + A caller that writes a file in two passes -a body, then the same body + wrapped in front matter -passes ``normalize`` to reduce both sides to the + part it owns. Without it the second pass always sees a difference, the + file is rewritten, and the fresh timestamp defeats the comparison. + """ + if os.path.exists(path): + with open(path) as f: + existing = f.read() + before, after = ( + (normalize(existing), normalize(content)) + if normalize + else (existing, content) + ) + if _strip_timestamps(before) == _strip_timestamps(after): + return False + with open(path, "w") as f: + f.write(content) + return True + +def _strip_timestamps(text: str) -> str: + """Remove known timestamp patterns for content comparison.""" + result = text + for pattern in _TIMESTAMP_PATTERNS: + result = pattern.sub("", result) + return result + +class ArtifactLockBusy(RuntimeError): + """Raised when another process already holds the artifact directory.""" + +@contextlib.contextmanager +def artifact_lock(directory: str, exclusive: bool = True): + """Serialize access to a shared artifact directory across processes. + + Two pipeline runs building the same dist/ leave readers looking at + half-written ZIPs, which surfaces as BadZipFile far from its cause. + Writers take the lock exclusively, readers share it. On platforms + without flock the lock is a no-op. + """ + try: + import fcntl + except ImportError: + yield + return + + os.makedirs(directory, exist_ok=True) + lock_path = os.path.join(directory, ".lock") + mode = fcntl.LOCK_EX if exclusive else fcntl.LOCK_SH + with open(lock_path, "w") as handle: + try: + fcntl.flock(handle, mode | fcntl.LOCK_NB) + except OSError as exc: + raise ArtifactLockBusy( + f"{directory} is in use by another run " + f"(lock: {lock_path}). Wait for it to finish." + ) from exc + try: + yield + finally: + fcntl.flock(handle, fcntl.LOCK_UN) diff --git a/scripts/common.py b/scripts/common.py index 11fbdeb0..eb8cb8be 100644 --- a/scripts/common.py +++ b/scripts/common.py @@ -6,123 +6,24 @@ and file resolution - eliminates DRY violations across scripts. from __future__ import annotations -import contextlib import hashlib import json import os import re -import stat -import tempfile -import urllib.error -import urllib.parse -import urllib.request -import xml.etree.ElementTree as ET import zipfile -import zlib from pathlib import Path -def parse_untrusted_xml(content: str | bytes, label: str = "XML") -> ET.Element: - """Parse XML fetched from a third party. - - ElementTree expands internal entities, so a document that declares them - can make the parser build a payload far larger than the bytes downloaded. - Nothing this project reads (DAT packs, es_bios.xml, Emulators.xml) ever - declares one, so a declaration is grounds to refuse the document rather - than something to expand carefully. - - The check targets dict[str, str]: - """Compute file hashes. Pass *algorithms* to limit which are computed.""" - algos = algorithms or _ALL_ALGORITHMS - sha1 = hashlib.sha1() if "sha1" in algos else None - md5 = hashlib.md5() if "md5" in algos else None - sha256 = hashlib.sha256() if "sha256" in algos else None - do_crc = "crc32" in algos - do_adler = "adler32" in algos - crc = 0 - adler = 1 # zlib.adler32 initial value - with open(filepath, "rb") as f: - for chunk in iter(lambda: f.read(65536), b""): - if sha1: - sha1.update(chunk) - if md5: - md5.update(chunk) - if sha256: - sha256.update(chunk) - if do_crc: - crc = zlib.crc32(chunk, crc) - if do_adler: - adler = zlib.adler32(chunk, adler) - result: dict[str, str] = {} - if sha1: - result["sha1"] = sha1.hexdigest() - if md5: - result["md5"] = md5.hexdigest() - if sha256: - result["sha256"] = sha256.hexdigest() - if do_crc: - result["crc32"] = format(crc & 0xFFFFFFFF, "08x") - if do_adler: - result["adler32"] = format(adler & 0xFFFFFFFF, "08x") - return result def load_database(db_path: str) -> dict: @@ -131,20 +32,6 @@ def load_database(db_path: str) -> dict: return json.load(f) -def md5sum(source: str | Path | object) -> str: - """Compute MD5 of a file path or file-like object - matches Batocera's md5sum().""" - h = hashlib.md5() - if hasattr(source, "read"): - for chunk in iter(lambda: source.read(65536), b""): - h.update(chunk) - else: - with open(source, "rb") as f: - for chunk in iter(lambda: f.read(65536), b""): - h.update(chunk) - return h.hexdigest() - - -_md5_composite_cache: dict[str, str] = {} _casefold_index_cache: dict[int, dict[str, list[str]]] = {} @@ -186,40 +73,6 @@ def name_match_size_ok(file_entry: dict, candidate_size: int | None) -> bool: return not (high is not None and candidate_size > high) -def md5_composite(filepath: str | Path) -> str: - """Compute composite MD5 of a ZIP - matches Recalbox's Zip::Md5Composite(). - - Sorts filenames alphabetically, reads each file's contents in order, - feeds everything into a single MD5 hasher. The result is independent - of ZIP compression level or metadata. Results are cached per path. - """ - key = str(filepath) - cached = _md5_composite_cache.get(key) - if cached is not None: - return cached - with zipfile.ZipFile(filepath) as zf: - names = sorted(n for n in zf.namelist() if not n.endswith("/")) - h = hashlib.md5() - for name in names: - info = zf.getinfo(name) - if info.file_size > 512 * 1024 * 1024: - continue # skip oversized entries - h.update(zf.read(name)) - result = h.hexdigest() - _md5_composite_cache[key] = result - return result - - -def parse_md5_list(raw: str | list | None) -> list[str]: - """Normalize an md5 field into a lowercase list. - - Platform YAMLs carry Recalbox multi-hash as one comma-separated string, - emulator profiles carry a YAML list. Both reach here. - """ - if not raw: - return [] - values = raw if isinstance(raw, list) else str(raw).split(",") - return [str(m).strip().lower() for m in values if str(m).strip()] _shared_yml_cache: dict[str, dict] = {} @@ -923,67 +776,7 @@ def get_mame_clone_map() -> dict[str, str]: return _mame_clone_map_cache -def check_inside_zip(container: str, file_name: str, expected_md5: str) -> str: - """Check a ROM inside a ZIP -replicates Batocera checkInsideZip(). - Returns "ok", "untested", "not_in_zip", or "error". - """ - try: - with zipfile.ZipFile(container) as archive: - for fname in archive.namelist(): - if fname.casefold() == file_name.casefold(): - info = archive.getinfo(fname) - if info.file_size > 512 * 1024 * 1024: - return "error" - if expected_md5 == "": - return "ok" - with archive.open(fname) as entry: - actual = md5sum(entry) - return "ok" if actual == expected_md5 else "untested" - return "not_in_zip" - except (zipfile.BadZipFile, OSError, KeyError): - return "error" - - -_zip_contents_cache: tuple[frozenset[tuple[str, float]], dict] | None = None - - -def build_zip_contents_index(db: dict, max_entry_size: int = 512 * 1024 * 1024) -> dict: - """Build {inner_rom_md5: zip_file_sha1} for ROMs inside ZIP files. - - Results are cached in-process; repeated calls with unchanged ZIPs return - the cached index. - """ - global _zip_contents_cache - - # Build fingerprint from ZIP paths + mtimes for cache invalidation - zip_entries: list[tuple[str, str]] = [] - for sha1, entry in db.get("files", {}).items(): - path = entry["path"] - if path.endswith(".zip") and os.path.exists(path): - zip_entries.append((path, sha1)) - - fingerprint = frozenset((path, os.path.getmtime(path)) for path, _ in zip_entries) - if _zip_contents_cache is not None and _zip_contents_cache[0] == fingerprint: - return _zip_contents_cache[1] - - index: dict[str, str] = {} - for path, sha1 in zip_entries: - try: - with zipfile.ZipFile(path, "r") as zf: - for info in zf.infolist(): - if info.is_dir() or info.file_size > max_entry_size: - continue - h = hashlib.md5() - with zf.open(info.filename) as inner: - for chunk in iter(lambda: inner.read(65536), b""): - h.update(chunk) - index[h.hexdigest()] = sha1 - except (zipfile.BadZipFile, OSError): - continue - - _zip_contents_cache = (fingerprint, index) - return index _emulator_profiles_cache: dict[tuple[str, bool], dict[str, dict]] = {} @@ -1384,136 +1177,14 @@ def expand_platform_declared_names(config: dict, db: dict) -> set[str]: import re -_TIMESTAMP_PATTERNS = [ - re.compile(r'"generated_at":\s*"[^"]*"'), # database.json - re.compile(r'"imported_at":\s*"[^"]*"'), # provenance snapshots - re.compile(r"\*Auto-generated on [^*]*\*"), # README.md - re.compile(r"\*Generated on [^*]*\*"), # docs site pages - # The decorated pages carry the same stamp again as a rendered element, - # and missing it rewrote every page on every run for the clock alone. - re.compile(r'
[^<]*
'), -] -def write_if_changed(path: str, content: str) -> bool: - """Write content to path only if the non-timestamp content differs. - - Compares new and existing content after stripping timestamp lines. - Returns True if the file was written, False if skipped (unchanged). - """ - if os.path.exists(path): - with open(path) as f: - existing = f.read() - if _strip_timestamps(existing) == _strip_timestamps(content): - return False - with open(path, "w") as f: - f.write(content) - return True - - -def _strip_timestamps(text: str) -> str: - """Remove known timestamp patterns for content comparison.""" - result = text - for pattern in _TIMESTAMP_PATTERNS: - result = pattern.sub("", result) - return result - # Validation and mode filtering -extracted to validation.py for SoC. # Re-exported below for backward compatibility. -LARGE_FILES_RELEASE = "large-files" -LARGE_FILES_REPO = "Abdess/retrobios" -LARGE_FILES_CACHE = ".cache/large" - -def fetch_large_file( - name: str, - dest_dir: str = LARGE_FILES_CACHE, - expected_sha1: str = "", - expected_md5: str = "", - *, - offline: bool = False, -) -> str | None: - """Return a verified cached large file, downloading it only when allowed.""" - cached = os.path.join(dest_dir, name) - if os.path.exists(cached): - if expected_sha1 or expected_md5: - hashes = compute_hashes(cached) - if expected_sha1 and hashes["sha1"].lower() != expected_sha1.lower(): - os.unlink(cached) - elif expected_md5: - md5_list = [ - m.strip().lower() for m in expected_md5.split(",") if m.strip() - ] - if hashes["md5"].lower() not in md5_list: - os.unlink(cached) - else: - return cached - else: - return cached - else: - return cached - - if offline: - return None - - os.makedirs(dest_dir, exist_ok=True) - # A per-process scratch name: two runs fetching the same asset into one - # shared path interleave their writes into a full-size, corrupt file. - tmp_fd, tmp_path = tempfile.mkstemp( - dir=dest_dir, prefix=os.path.basename(cached) + ".", suffix=".tmp" - ) - os.close(tmp_fd) - # GitHub rewrites spaces to dots in release asset names, so a file whose - # name contains spaces is published under a dotted name. - candidates = [name] - if " " in name: - candidates.append(name.replace(" ", ".")) - - downloaded = False - for candidate in candidates: - encoded_name = urllib.parse.quote(candidate) - url = ( - f"https://github.com/{LARGE_FILES_REPO}/releases/download/" - f"{LARGE_FILES_RELEASE}/{encoded_name}" - ) - try: - req = urllib.request.Request(url, headers={"User-Agent": "retrobios/1.0"}) - with urllib.request.urlopen(req, timeout=300) as resp: - with open(tmp_path, "wb") as f: - while True: - chunk = resp.read(65536) - if not chunk: - break - f.write(chunk) - downloaded = True - break - except (urllib.error.URLError, urllib.error.HTTPError): - if os.path.exists(tmp_path): - os.unlink(tmp_path) - - if not downloaded: - if os.path.exists(tmp_path): - os.unlink(tmp_path) - return None - - if expected_sha1 or expected_md5: - hashes = compute_hashes(tmp_path) - if expected_sha1 and hashes["sha1"].lower() != expected_sha1.lower(): - os.unlink(tmp_path) - return None - if expected_md5: - md5_list = [m.strip().lower() for m in expected_md5.split(",") if m.strip()] - if hashes["md5"].lower() not in md5_list: - os.unlink(tmp_path) - return None - os.replace(tmp_path, cached) - return cached - - -MAX_ZIP_MEMBERS = 100_000 def sanitize_pack_path(raw: str) -> str: """Strip traversal components from a relative destination. @@ -1525,117 +1196,6 @@ def sanitize_pack_path(raw: str) -> str: return "/".join(p for p in raw.split("/") if p and p not in ("..", ".")) -MAX_ZIP_MEMBER_SIZE = 8 * 1024 * 1024 * 1024 -# The largest generated pack is already ~5 GB uncompressed and the collection -# only grows; this bounds a malicious archive without capping a real one. -MAX_ZIP_TOTAL_SIZE = 64 * 1024 * 1024 * 1024 -# DEFLATE cannot exceed roughly 1,032:1, so this rejects a declared ratio no -# real DEFLATE member can reach. Methods with a higher ceiling (bzip2, LZMA) -# are exempt and bounded by the per-member and per-archive size limits alone. -MAX_ZIP_COMPRESSION_RATIO = 1_100 -_BOUNDED_RATIO_METHODS = (zipfile.ZIP_STORED, zipfile.ZIP_DEFLATED) - - -def safe_extract_zip( - zip_path: str, - dest_dir: str, - *, - max_members: int = MAX_ZIP_MEMBERS, - max_member_size: int = MAX_ZIP_MEMBER_SIZE, - max_total_size: int = MAX_ZIP_TOTAL_SIZE, - max_compression_ratio: int = MAX_ZIP_COMPRESSION_RATIO, -) -> None: - """Extract a ZIP with traversal, link and resource-limit protection. - - Files are streamed to a temporary sibling and atomically installed only - after their declared length and CRC have been checked by ``zipfile``. - """ - dest = os.path.realpath(dest_dir) - os.makedirs(dest, exist_ok=True) - with zipfile.ZipFile(zip_path, "r") as zf: - members = zf.infolist() - if len(members) > max_members: - raise ValueError( - f"ZIP has {len(members)} members; limit is {max_members}" - ) - - declared_total = 0 - seen: set[str] = set() - for member in members: - # Archives written on Windows store a backslash separator. It is a - # separator, not a filename character, so it is normalized before - # the component checks rather than rejected. - name = member.filename.replace("\\", "/") - if not name or "\x00" in name: - raise ValueError(f"Unsafe ZIP member name: {member.filename!r}") - if name.startswith("/") or re.match(r"^[A-Za-z]:", name): - raise ValueError(f"Absolute ZIP member path: {name}") - parts = [part for part in name.split("/") if part] - if any(part in (".", "..") for part in parts): - raise ValueError(f"ZIP traversal detected: {name}") - normalized = "/".join(parts) - if normalized in seen: - raise ValueError(f"Duplicate ZIP member path: {name}") - seen.add(normalized) - - mode = (member.external_attr >> 16) & 0xFFFF - file_type = stat.S_IFMT(mode) - if file_type not in (0, stat.S_IFREG, stat.S_IFDIR): - raise ValueError(f"ZIP link or special file rejected: {name}") - if member.flag_bits & 0x1: - raise ValueError(f"Encrypted ZIP member rejected: {name}") - if member.file_size > max_member_size: - raise ValueError( - f"ZIP member {name} is {member.file_size} bytes; " - f"limit is {max_member_size}" - ) - declared_total += member.file_size - if declared_total > max_total_size: - raise ValueError( - f"ZIP expands to {declared_total} bytes; limit is {max_total_size}" - ) - if member.file_size and member.compress_type in _BOUNDED_RATIO_METHODS: - if member.compress_size == 0: - raise ValueError(f"Invalid compression size for ZIP member: {name}") - if member.file_size / member.compress_size > max_compression_ratio: - raise ValueError(f"Suspicious compression ratio for ZIP member: {name}") - - target = os.path.realpath(os.path.join(dest, *parts)) - if not target.startswith(dest + os.sep) and target != dest: - raise ValueError(f"ZIP traversal detected: {name}") - if member.is_dir() or name.endswith("/"): - os.makedirs(target, exist_ok=True) - continue - - os.makedirs(os.path.dirname(target), exist_ok=True) - tmp_path = "" - try: - with tempfile.NamedTemporaryFile( - mode="wb", dir=os.path.dirname(target), delete=False - ) as tmp_file: - tmp_path = tmp_file.name - actual_size = 0 - with zf.open(member, "r") as source: - while True: - chunk = source.read(1024 * 1024) - if not chunk: - break - actual_size += len(chunk) - if actual_size > member.file_size or actual_size > max_member_size: - raise ValueError( - f"ZIP member exceeded declared or configured size: {name}" - ) - tmp_file.write(chunk) - if actual_size != member.file_size: - raise ValueError( - f"ZIP member size mismatch for {name}: " - f"{actual_size} != {member.file_size}" - ) - os.replace(tmp_path, target) - tmp_path = "" - finally: - if tmp_path and os.path.exists(tmp_path): - os.unlink(tmp_path) def list_emulator_profiles(emulators_dir: str, skip_aliases: bool = True) -> None: @@ -1709,125 +1269,51 @@ def build_target_cores_cache( return cache, kept -DEFAULT_PROVENANCE_DIR = "provenance" - - -def load_provenance_snapshots(provenance_dir: str = DEFAULT_PROVENANCE_DIR) -> dict: - """Load dump-catalog snapshots from provenance/*.json. - - Returns {source_name: snapshot} where snapshot holds the normalized - entries written by the redump scraper or the pack importer. Missing - directory means no snapshots: returns an empty dict. - """ - snapshots = {} - prov_path = Path(provenance_dir) - if not prov_path.is_dir(): - return snapshots - for path in sorted(prov_path.glob("*.json")): - with open(path) as f: - snapshot = json.load(f) - source = snapshot.get("source") - if source and snapshot.get("entries"): - snapshots[source] = snapshot - return snapshots - - -def build_provenance_index(snapshots: dict) -> dict: - """Index snapshot entries by sha1 and by (md5, size) per source. - - First entry wins on hash collisions within a source; entries are - pre-sorted at snapshot write time so the outcome is deterministic. - """ - index = {} - for source, snapshot in snapshots.items(): - by_sha1 = {} - by_md5_size = {} - for entry in snapshot["entries"]: - sha1 = entry.get("sha1", "") - md5 = entry.get("md5", "") - if sha1 and sha1 not in by_sha1: - by_sha1[sha1] = entry - if md5 and entry.get("size"): - key = (md5, entry["size"]) - if key not in by_md5_size: - by_md5_size[key] = entry - index[source] = {"by_sha1": by_sha1, "by_md5_size": by_md5_size} - return index - - -def annotate_provenance(files: dict, snapshots: dict) -> dict[str, int]: - """Attach a provenance field to database file entries. - - Matches by SHA1 first, then MD5 + size. Returns per-source match - counts. Files without any catalog match keep no provenance field. - """ - index = build_provenance_index(snapshots) - counts = dict.fromkeys(index, 0) - for sha1, entry in files.items(): - matches = {} - for source in sorted(index): - src_index = index[source] - hit = src_index["by_sha1"].get(sha1) or src_index["by_md5_size"].get( - (entry.get("md5", ""), entry.get("size", 0)) - ) - if hit: - matches[source] = { - "dat": hit.get("dat", ""), - "name": hit.get("name", ""), - "description": hit.get("description", ""), - } - counts[source] += 1 - if matches: - entry["provenance"] = matches - else: - entry.pop("provenance", None) - return counts - - -def write_provenance_snapshot( - path: str, source: str, imported_at: str, dats: dict, entries: list[dict] -) -> bool: - """Write a normalized provenance snapshot, sorted for determinism.""" - snapshot = { - "source": source, - "imported_at": imported_at, - "dats": dict(sorted(dats.items())), - "entries": sorted(entries, key=lambda e: (e["dat"], e["name"])), - } - return write_if_changed(path, json.dumps(snapshot, indent=2) + "\n") - - -class ArtifactLockBusy(RuntimeError): - """Raised when another process already holds the artifact directory.""" - - -@contextlib.contextmanager -def artifact_lock(directory: str, exclusive: bool = True): - """Serialize access to a shared artifact directory across processes. - - Two pipeline runs building the same dist/ leave readers looking at - half-written ZIPs, which surfaces as BadZipFile far from its cause. - Writers take the lock exclusively, readers share it. On platforms - without flock the lock is a no-op. - """ - try: - import fcntl - except ImportError: - yield - return - - os.makedirs(directory, exist_ok=True) - lock_path = os.path.join(directory, ".lock") - mode = fcntl.LOCK_EX if exclusive else fcntl.LOCK_SH - with open(lock_path, "w") as handle: - try: - fcntl.flock(handle, mode | fcntl.LOCK_NB) - except OSError as exc: - raise ArtifactLockBusy( - f"{directory} is in use by another run " - f"(lock: {lock_path}). Wait for it to finish." - ) from exc - try: - yield - finally: - fcntl.flock(handle, fcntl.LOCK_UN) +# Re-exported so the existing call sites keep working while the +# modules above become the place to import from. +from safeparse import ( # noqa: E402,F401 + parse_untrusted_xml, + require_yaml, + yaml_load, + _pick_yaml_loader, + _YAML_LOADER, +) +from hashing import ( # noqa: E402,F401 + compute_hashes, + md5sum, + md5_composite, + parse_md5_list, + _ALL_ALGORITHMS, + _md5_composite_cache, +) +from ziptools import ( # noqa: E402,F401 + check_inside_zip, + build_zip_contents_index, + MAX_ZIP_MEMBERS, + MAX_ZIP_MEMBER_SIZE, + MAX_ZIP_TOTAL_SIZE, + MAX_ZIP_COMPRESSION_RATIO, + safe_extract_zip, + _zip_contents_cache, + _BOUNDED_RATIO_METHODS, +) +from artifacts import ( # noqa: E402,F401 + write_if_changed, + ArtifactLockBusy, + artifact_lock, + _TIMESTAMP_PATTERNS, + _strip_timestamps, +) +from largefiles import ( # noqa: E402,F401 + LARGE_FILES_RELEASE, + LARGE_FILES_REPO, + LARGE_FILES_CACHE, + fetch_large_file, +) +from dumpcatalog import ( # noqa: E402,F401 + DEFAULT_PROVENANCE_DIR, + load_provenance_snapshots, + build_provenance_index, + annotate_provenance, + write_provenance_snapshot, +) diff --git a/scripts/dumpcatalog.py b/scripts/dumpcatalog.py new file mode 100644 index 00000000..1211722e --- /dev/null +++ b/scripts/dumpcatalog.py @@ -0,0 +1,96 @@ +"""Dump catalogues joined onto the collection. + +Redump, No-Intro and TOSEC annotate what the repo holds. They are an +opinion on provenance, never an authority: the emulator source is.""" + +from __future__ import annotations + +import json +import os +from pathlib import Path + +from artifacts import write_if_changed + + +DEFAULT_PROVENANCE_DIR = "provenance" + +def load_provenance_snapshots(provenance_dir: str = DEFAULT_PROVENANCE_DIR) -> dict: + """Load dump-catalog snapshots from provenance/*.json. + + Returns {source_name: snapshot} where snapshot holds the normalized + entries written by the redump scraper or the pack importer. Missing + directory means no snapshots: returns an empty dict. + """ + snapshots = {} + prov_path = Path(provenance_dir) + if not prov_path.is_dir(): + return snapshots + for path in sorted(prov_path.glob("*.json")): + with open(path) as f: + snapshot = json.load(f) + source = snapshot.get("source") + if source and snapshot.get("entries"): + snapshots[source] = snapshot + return snapshots + +def build_provenance_index(snapshots: dict) -> dict: + """Index snapshot entries by sha1 and by (md5, size) per source. + + First entry wins on hash collisions within a source; entries are + pre-sorted at snapshot write time so the outcome is deterministic. + """ + index = {} + for source, snapshot in snapshots.items(): + by_sha1 = {} + by_md5_size = {} + for entry in snapshot["entries"]: + sha1 = entry.get("sha1", "") + md5 = entry.get("md5", "") + if sha1 and sha1 not in by_sha1: + by_sha1[sha1] = entry + if md5 and entry.get("size"): + key = (md5, entry["size"]) + if key not in by_md5_size: + by_md5_size[key] = entry + index[source] = {"by_sha1": by_sha1, "by_md5_size": by_md5_size} + return index + +def annotate_provenance(files: dict, snapshots: dict) -> dict[str, int]: + """Attach a provenance field to database file entries. + + Matches by SHA1 first, then MD5 + size. Returns per-source match + counts. Files without any catalog match keep no provenance field. + """ + index = build_provenance_index(snapshots) + counts = dict.fromkeys(index, 0) + for sha1, entry in files.items(): + matches = {} + for source in sorted(index): + src_index = index[source] + hit = src_index["by_sha1"].get(sha1) or src_index["by_md5_size"].get( + (entry.get("md5", ""), entry.get("size", 0)) + ) + if hit: + matches[source] = { + "dat": hit.get("dat", ""), + "name": hit.get("name", ""), + "description": hit.get("description", ""), + } + counts[source] += 1 + if matches: + entry["provenance"] = matches + else: + entry.pop("provenance", None) + return counts + +def write_provenance_snapshot( + path: str, source: str, imported_at: str, dats: dict, entries: list[dict] +) -> bool: + """Write a normalized provenance snapshot, sorted for determinism.""" + snapshot = { + "source": source, + "imported_at": imported_at, + "dats": dict(sorted(dats.items())), + "entries": sorted(entries, key=lambda e: (e["dat"], e["name"])), + } + return write_if_changed(path, json.dumps(snapshot, indent=2) + "\n") diff --git a/scripts/generate_site.py b/scripts/generate_site.py index e992349d..f0576227 100644 --- a/scripts/generate_site.py +++ b/scripts/generate_site.py @@ -39,11 +39,70 @@ from common import ( parse_md5_list, require_yaml, unique_emulator_profiles, - write_if_changed, + write_if_changed as _write_artifact, yaml_load, ) from nativemode import reads_file_contents +# Every path this run produced. The generated directories used to be deleted +# before the build, which made every page new: write_if_changed had no earlier +# version to compare against, so the whole site was rewritten for the clock +# alone and a deploy republished 600 unchanged pages. They are swept instead, +# and a page is only removed once nothing produced it. +_produced: set[str] = set() + + +def _record(path) -> None: + _produced.add(os.path.realpath(str(path))) + + +def _undecorated(markdown: str) -> str: + """Return a page body without the front matter the decoration pass adds.""" + if markdown.startswith("---\n") and "generated_by: retrobios-site" in markdown[:300]: + end = markdown.find("\n---\n", 4) + if end != -1: + markdown = markdown[end + 5:].lstrip("\n") + script_end = markdown.find("\n\n") + if markdown.startswith('\n\n"):] + return markdown + + +def write_if_changed(path: str, content: str) -> bool: + """Write a page body only when it moved, and remember it either way. + + A page is written twice: the body here, then the same body wrapped in + front matter by the decoration pass. This comparison therefore reduces + the file on disk to its body. The decoration pass compares in full - + normalizing there too would make its own write look like a no-op and + every page would lose its front matter. + """ + _record(path) + return _write_artifact(path, content, normalize=_undecorated) + + +def write_decorated(path, content: str) -> bool: + """Write the wrapped page, comparing front matter and body together.""" + _record(path) + return _write_artifact(str(path), content) + + +def _sweep_generated(docs: Path, directories: list[str]) -> int: + """Delete files in the generated tree that this run did not produce.""" + removed = 0 + for name in directories: + root = docs / name + if not root.is_dir(): + continue + for path in sorted(root.rglob("*")): + if path.is_file() and os.path.realpath(str(path)) not in _produced: + path.unlink() + removed += 1 + for path in sorted(root.rglob("*"), reverse=True): + if path.is_dir() and not any(path.iterdir()): + path.rmdir() + return removed + yaml = require_yaml() from generate_readme import compute_coverage, manifest_totals from profile_sync import source_ref_values, split_source_ref @@ -1134,6 +1193,7 @@ def generate_data_exports( ) for schema_name in schema_names: shutil.copy2(Path("schemas") / schema_name, schemas_dest / schema_name) + _record(schemas_dest / schema_name) file_rows = [] for sha1, entry in sorted(db.get("files", {}).items()): @@ -1197,6 +1257,7 @@ def generate_data_exports( downloads / "retrobios.sqlite", db, platform_items, platform_files, emulator_items, gap_rows, ) + _record(downloads / "retrobios.sqlite") assets = [ (api / "database.json", "Content database", "application/json", "schemas/database.schema.json"), @@ -1386,13 +1447,9 @@ def decorate_markdown_pages(docs: Path) -> None: if relative.parts and relative.parts[0] == "superpowers": continue markdown = path.read_text(encoding="utf-8") - if markdown.startswith("---\n") and "generated_by: retrobios-site" in markdown[:300]: - end = markdown.find("\n---\n", 4) - if end != -1: - markdown = markdown[end + 5:].lstrip("\n") - script_end = markdown.find("\n\n") - if markdown.startswith('\n\n"):] + stripped = _undecorated(markdown) + if stripped is not markdown: + markdown = stripped elif markdown.startswith("---\n"): continue @@ -1432,7 +1489,7 @@ def decorate_markdown_pages(docs: Path) -> None: f"{structured_json}\n" "\n\n" ) - write_if_changed(str(path), front_matter + markdown) + write_decorated(path, front_matter + markdown) # Platform pages @@ -3881,12 +3938,6 @@ def main(): db = load_database(args.db) docs = Path(args.docs_dir) - # Clean generated dirs (preserve docs/superpowers/) - for d in GENERATED_DIRS: - target = docs / d - if target.exists(): - shutil.rmtree(target) - # Ensure output dirs for d in GENERATED_DIRS: (docs / d).mkdir(parents=True, exist_ok=True) @@ -4059,6 +4110,7 @@ def main(): if wiki_src.is_dir(): for src_file in wiki_src.glob("*.md"): shutil.copy2(src_file, wiki_dest / src_file.name) + _record(wiki_dest / src_file.name) # data-model.md is generated (contains live DB stats) write_if_changed( str(wiki_dest / "data-model.md"), generate_wiki_data_model(db, profiles) @@ -4208,7 +4260,10 @@ validation: + 1 # contributing + 1 # data and API ) + stale = _sweep_generated(docs, GENERATED_DIRS) print(f"\nGenerated {total_pages} pages in {args.docs_dir}/") + if stale: + print(f"Removed {stale} page(s) nothing produced this run") if __name__ == "__main__": diff --git a/scripts/hashing.py b/scripts/hashing.py new file mode 100644 index 00000000..f1b95758 --- /dev/null +++ b/scripts/hashing.py @@ -0,0 +1,101 @@ +"""Content digests. + +SHA-1 is the collection's primary key; the rest are what the various +frontends check against.""" + +from __future__ import annotations + +import hashlib +import os +import zipfile +import zlib +from pathlib import Path + + +_ALL_ALGORITHMS = frozenset({"sha1", "md5", "sha256", "crc32", "adler32"}) + +def compute_hashes( + filepath: str | Path, + algorithms: frozenset[str] | None = None, +) -> dict[str, str]: + """Compute file hashes. Pass *algorithms* to limit which are computed.""" + algos = algorithms or _ALL_ALGORITHMS + sha1 = hashlib.sha1() if "sha1" in algos else None + md5 = hashlib.md5() if "md5" in algos else None + sha256 = hashlib.sha256() if "sha256" in algos else None + do_crc = "crc32" in algos + do_adler = "adler32" in algos + crc = 0 + adler = 1 # zlib.adler32 initial value + with open(filepath, "rb") as f: + for chunk in iter(lambda: f.read(65536), b""): + if sha1: + sha1.update(chunk) + if md5: + md5.update(chunk) + if sha256: + sha256.update(chunk) + if do_crc: + crc = zlib.crc32(chunk, crc) + if do_adler: + adler = zlib.adler32(chunk, adler) + result: dict[str, str] = {} + if sha1: + result["sha1"] = sha1.hexdigest() + if md5: + result["md5"] = md5.hexdigest() + if sha256: + result["sha256"] = sha256.hexdigest() + if do_crc: + result["crc32"] = format(crc & 0xFFFFFFFF, "08x") + if do_adler: + result["adler32"] = format(adler & 0xFFFFFFFF, "08x") + return result + +def md5sum(source: str | Path | object) -> str: + """Compute MD5 of a file path or file-like object - matches Batocera's md5sum().""" + h = hashlib.md5() + if hasattr(source, "read"): + for chunk in iter(lambda: source.read(65536), b""): + h.update(chunk) + else: + with open(source, "rb") as f: + for chunk in iter(lambda: f.read(65536), b""): + h.update(chunk) + return h.hexdigest() + +_md5_composite_cache: dict[str, str] = {} + +def md5_composite(filepath: str | Path) -> str: + """Compute composite MD5 of a ZIP - matches Recalbox's Zip::Md5Composite(). + + Sorts filenames alphabetically, reads each file's contents in order, + feeds everything into a single MD5 hasher. The result is independent + of ZIP compression level or metadata. Results are cached per path. + """ + key = str(filepath) + cached = _md5_composite_cache.get(key) + if cached is not None: + return cached + with zipfile.ZipFile(filepath) as zf: + names = sorted(n for n in zf.namelist() if not n.endswith("/")) + h = hashlib.md5() + for name in names: + info = zf.getinfo(name) + if info.file_size > 512 * 1024 * 1024: + continue # skip oversized entries + h.update(zf.read(name)) + result = h.hexdigest() + _md5_composite_cache[key] = result + return result + +def parse_md5_list(raw: str | list | None) -> list[str]: + """Normalize an md5 field into a lowercase list. + + Platform YAMLs carry Recalbox multi-hash as one comma-separated string, + emulator profiles carry a YAML list. Both reach here. + """ + if not raw: + return [] + values = raw if isinstance(raw, list) else str(raw).split(",") + return [str(m).strip().lower() for m in values if str(m).strip()] diff --git a/scripts/largefiles.py b/scripts/largefiles.py new file mode 100644 index 00000000..348d65a4 --- /dev/null +++ b/scripts/largefiles.py @@ -0,0 +1,106 @@ +"""Files too large for the repository. + +They live as release assets and are fetched at build time, verified +against the hash the caller declares.""" + +from __future__ import annotations + +import os +import tempfile +import urllib.error +import urllib.request + +from hashing import compute_hashes + + +LARGE_FILES_RELEASE = "large-files" + +LARGE_FILES_RELEASE = "large-files" +LARGE_FILES_REPO = "Abdess/retrobios" + +LARGE_FILES_REPO = "Abdess/retrobios" +LARGE_FILES_CACHE = ".cache/large" + +def fetch_large_file( + name: str, + dest_dir: str = LARGE_FILES_CACHE, + expected_sha1: str = "", + expected_md5: str = "", + *, + offline: bool = False, +) -> str | None: + """Return a verified cached large file, downloading it only when allowed.""" + cached = os.path.join(dest_dir, name) + if os.path.exists(cached): + if expected_sha1 or expected_md5: + hashes = compute_hashes(cached) + if expected_sha1 and hashes["sha1"].lower() != expected_sha1.lower(): + os.unlink(cached) + elif expected_md5: + md5_list = [ + m.strip().lower() for m in expected_md5.split(",") if m.strip() + ] + if hashes["md5"].lower() not in md5_list: + os.unlink(cached) + else: + return cached + else: + return cached + else: + return cached + + if offline: + return None + + os.makedirs(dest_dir, exist_ok=True) + # A per-process scratch name: two runs fetching the same asset into one + # shared path interleave their writes into a full-size, corrupt file. + tmp_fd, tmp_path = tempfile.mkstemp( + dir=dest_dir, prefix=os.path.basename(cached) + ".", suffix=".tmp" + ) + os.close(tmp_fd) + # GitHub rewrites spaces to dots in release asset names, so a file whose + # name contains spaces is published under a dotted name. + candidates = [name] + if " " in name: + candidates.append(name.replace(" ", ".")) + + downloaded = False + for candidate in candidates: + encoded_name = urllib.parse.quote(candidate) + url = ( + f"https://github.com/{LARGE_FILES_REPO}/releases/download/" + f"{LARGE_FILES_RELEASE}/{encoded_name}" + ) + try: + req = urllib.request.Request(url, headers={"User-Agent": "retrobios/1.0"}) + with urllib.request.urlopen(req, timeout=300) as resp: + with open(tmp_path, "wb") as f: + while True: + chunk = resp.read(65536) + if not chunk: + break + f.write(chunk) + downloaded = True + break + except (urllib.error.URLError, urllib.error.HTTPError): + if os.path.exists(tmp_path): + os.unlink(tmp_path) + + if not downloaded: + if os.path.exists(tmp_path): + os.unlink(tmp_path) + return None + + if expected_sha1 or expected_md5: + hashes = compute_hashes(tmp_path) + if expected_sha1 and hashes["sha1"].lower() != expected_sha1.lower(): + os.unlink(tmp_path) + return None + if expected_md5: + md5_list = [m.strip().lower() for m in expected_md5.split(",") if m.strip()] + if hashes["md5"].lower() not in md5_list: + os.unlink(tmp_path) + return None + os.replace(tmp_path, cached) + return cached diff --git a/scripts/safeparse.py b/scripts/safeparse.py new file mode 100644 index 00000000..c4aa9810 --- /dev/null +++ b/scripts/safeparse.py @@ -0,0 +1,65 @@ +"""Parsers for input the repo does not control. + +Upstream DATs and scraped documents are data, never instructions, and +the parsers here are the boundary that keeps it that way.""" + +from __future__ import annotations + +import xml.etree.ElementTree as ET + +try: + import yaml +except ImportError: # optional at import time + yaml = None + + +def parse_untrusted_xml(content: str | bytes, label: str = "XML") -> ET.Element: + """Parse XML fetched from a third party. + + ElementTree expands internal entities, so a document that declares them + can make the parser build a payload far larger than the bytes downloaded. + Nothing this project reads (DAT packs, es_bios.xml, Emulators.xml) ever + declares one, so a declaration is grounds to refuse the document rather + than something to expand carefully. + + The check targets str: + """Check a ROM inside a ZIP -replicates Batocera checkInsideZip(). + + Returns "ok", "untested", "not_in_zip", or "error". + """ + try: + with zipfile.ZipFile(container) as archive: + for fname in archive.namelist(): + if fname.casefold() == file_name.casefold(): + info = archive.getinfo(fname) + if info.file_size > 512 * 1024 * 1024: + return "error" + if expected_md5 == "": + return "ok" + with archive.open(fname) as entry: + actual = md5sum(entry) + return "ok" if actual == expected_md5 else "untested" + return "not_in_zip" + except (zipfile.BadZipFile, OSError, KeyError): + return "error" + +_zip_contents_cache: tuple[frozenset[tuple[str, float]], dict] | None = None + +def build_zip_contents_index(db: dict, max_entry_size: int = 512 * 1024 * 1024) -> dict: + """Build {inner_rom_md5: zip_file_sha1} for ROMs inside ZIP files. + + Results are cached in-process; repeated calls with unchanged ZIPs return + the cached index. + """ + global _zip_contents_cache + + # Build fingerprint from ZIP paths + mtimes for cache invalidation + zip_entries: list[tuple[str, str]] = [] + for sha1, entry in db.get("files", {}).items(): + path = entry["path"] + if path.endswith(".zip") and os.path.exists(path): + zip_entries.append((path, sha1)) + + fingerprint = frozenset((path, os.path.getmtime(path)) for path, _ in zip_entries) + if _zip_contents_cache is not None and _zip_contents_cache[0] == fingerprint: + return _zip_contents_cache[1] + + index: dict[str, str] = {} + for path, sha1 in zip_entries: + try: + with zipfile.ZipFile(path, "r") as zf: + for info in zf.infolist(): + if info.is_dir() or info.file_size > max_entry_size: + continue + h = hashlib.md5() + with zf.open(info.filename) as inner: + for chunk in iter(lambda: inner.read(65536), b""): + h.update(chunk) + index[h.hexdigest()] = sha1 + except (zipfile.BadZipFile, OSError): + continue + + _zip_contents_cache = (fingerprint, index) + return index + +MAX_ZIP_MEMBERS = 100_000 + +MAX_ZIP_MEMBER_SIZE = 8 * 1024 * 1024 * 1024 + +MAX_ZIP_MEMBER_SIZE = 8 * 1024 * 1024 * 1024 +# The largest generated pack is already ~5 GB uncompressed and the collection +# only grows; this bounds a malicious archive without capping a real one. +MAX_ZIP_TOTAL_SIZE = 64 * 1024 * 1024 * 1024 + +MAX_ZIP_TOTAL_SIZE = 64 * 1024 * 1024 * 1024 +# DEFLATE cannot exceed roughly 1,032:1, so this rejects a declared ratio no +# real DEFLATE member can reach. Methods with a higher ceiling (bzip2, LZMA) +# are exempt and bounded by the per-member and per-archive size limits alone. +MAX_ZIP_COMPRESSION_RATIO = 1_100 + +MAX_ZIP_COMPRESSION_RATIO = 1_100 +_BOUNDED_RATIO_METHODS = (zipfile.ZIP_STORED, zipfile.ZIP_DEFLATED) + +def safe_extract_zip( + zip_path: str, + dest_dir: str, + *, + max_members: int = MAX_ZIP_MEMBERS, + max_member_size: int = MAX_ZIP_MEMBER_SIZE, + max_total_size: int = MAX_ZIP_TOTAL_SIZE, + max_compression_ratio: int = MAX_ZIP_COMPRESSION_RATIO, +) -> None: + """Extract a ZIP with traversal, link and resource-limit protection. + + Files are streamed to a temporary sibling and atomically installed only + after their declared length and CRC have been checked by ``zipfile``. + """ + dest = os.path.realpath(dest_dir) + os.makedirs(dest, exist_ok=True) + with zipfile.ZipFile(zip_path, "r") as zf: + members = zf.infolist() + if len(members) > max_members: + raise ValueError( + f"ZIP has {len(members)} members; limit is {max_members}" + ) + + declared_total = 0 + seen: set[str] = set() + for member in members: + # Archives written on Windows store a backslash separator. It is a + # separator, not a filename character, so it is normalized before + # the component checks rather than rejected. + name = member.filename.replace("\\", "/") + if not name or "\x00" in name: + raise ValueError(f"Unsafe ZIP member name: {member.filename!r}") + if name.startswith("/") or re.match(r"^[A-Za-z]:", name): + raise ValueError(f"Absolute ZIP member path: {name}") + parts = [part for part in name.split("/") if part] + if any(part in (".", "..") for part in parts): + raise ValueError(f"ZIP traversal detected: {name}") + normalized = "/".join(parts) + if normalized in seen: + raise ValueError(f"Duplicate ZIP member path: {name}") + seen.add(normalized) + + mode = (member.external_attr >> 16) & 0xFFFF + file_type = stat.S_IFMT(mode) + if file_type not in (0, stat.S_IFREG, stat.S_IFDIR): + raise ValueError(f"ZIP link or special file rejected: {name}") + if member.flag_bits & 0x1: + raise ValueError(f"Encrypted ZIP member rejected: {name}") + if member.file_size > max_member_size: + raise ValueError( + f"ZIP member {name} is {member.file_size} bytes; " + f"limit is {max_member_size}" + ) + declared_total += member.file_size + if declared_total > max_total_size: + raise ValueError( + f"ZIP expands to {declared_total} bytes; limit is {max_total_size}" + ) + if member.file_size and member.compress_type in _BOUNDED_RATIO_METHODS: + if member.compress_size == 0: + raise ValueError(f"Invalid compression size for ZIP member: {name}") + if member.file_size / member.compress_size > max_compression_ratio: + raise ValueError(f"Suspicious compression ratio for ZIP member: {name}") + + target = os.path.realpath(os.path.join(dest, *parts)) + if not target.startswith(dest + os.sep) and target != dest: + raise ValueError(f"ZIP traversal detected: {name}") + if member.is_dir() or name.endswith("/"): + os.makedirs(target, exist_ok=True) + continue + + os.makedirs(os.path.dirname(target), exist_ok=True) + tmp_path = "" + try: + with tempfile.NamedTemporaryFile( + mode="wb", dir=os.path.dirname(target), delete=False + ) as tmp_file: + tmp_path = tmp_file.name + actual_size = 0 + with zf.open(member, "r") as source: + while True: + chunk = source.read(1024 * 1024) + if not chunk: + break + actual_size += len(chunk) + if actual_size > member.file_size or actual_size > max_member_size: + raise ValueError( + f"ZIP member exceeded declared or configured size: {name}" + ) + tmp_file.write(chunk) + if actual_size != member.file_size: + raise ValueError( + f"ZIP member size mismatch for {name}: " + f"{actual_size} != {member.file_size}" + ) + os.replace(tmp_path, target) + tmp_path = "" + finally: + if tmp_path and os.path.exists(tmp_path): + os.unlink(tmp_path) diff --git a/tests/test_e2e.py b/tests/test_e2e.py index 6fdf22ad..b70bfb56 100644 --- a/tests/test_e2e.py +++ b/tests/test_e2e.py @@ -28,6 +28,7 @@ from pathlib import Path sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "scripts")) +import largefiles import yaml from common import ( build_zip_contents_index, @@ -5120,8 +5121,8 @@ struct BurnDriver BurnDrvneogeo = { raise urllib.error.HTTPError(url, 404, "Not Found", {}, None) return FakeResponse(b"payload") - original = common.urllib.request.urlopen - common.urllib.request.urlopen = fake_urlopen + original = largefiles.urllib.request.urlopen + largefiles.urllib.request.urlopen = fake_urlopen try: with tempfile.TemporaryDirectory() as tmpdir: path = common.fetch_large_file( @@ -5131,7 +5132,7 @@ struct BurnDriver BurnDrvneogeo = { with open(path, "rb") as fh: self.assertEqual(fh.read(), b"payload") finally: - common.urllib.request.urlopen = original + largefiles.urllib.request.urlopen = original self.assertEqual(len(attempts), 2, attempts) self.assertIn("MAME%200.174", attempts[0]) diff --git a/tests/test_large_file_cache.py b/tests/test_large_file_cache.py index c0b1998c..e0f3b84e 100644 --- a/tests/test_large_file_cache.py +++ b/tests/test_large_file_cache.py @@ -23,6 +23,7 @@ REPO_ROOT = Path(__file__).resolve().parent.parent sys.path.insert(0, str(REPO_ROOT / "scripts")) import common # noqa: E402 +import largefiles PAYLOAD_A = b"A" * (256 * 1024) PAYLOAD_B = b"B" * (256 * 1024) @@ -54,10 +55,10 @@ class _SlowResponse(io.BytesIO): class LargeFileCacheTest(unittest.TestCase): def setUp(self): self.dir = tempfile.mkdtemp() - self._urlopen = common.urllib.request.urlopen + self._urlopen = largefiles.urllib.request.urlopen def tearDown(self): - common.urllib.request.urlopen = self._urlopen + largefiles.urllib.request.urlopen = self._urlopen def test_concurrent_fetches_do_not_mix(self): barrier = threading.Barrier(2, timeout=10) @@ -70,7 +71,7 @@ class LargeFileCacheTest(unittest.TestCase): i = next(index) return _SlowResponse(payloads[i], barrier) - common.urllib.request.urlopen = fake_urlopen + largefiles.urllib.request.urlopen = fake_urlopen results: list[str | None] = [None, None] def worker(slot: int): @@ -90,7 +91,7 @@ class LargeFileCacheTest(unittest.TestCase): self.assertIn(digest, accepted) def test_no_scratch_file_survives_a_successful_fetch(self): - common.urllib.request.urlopen = lambda req, timeout=None: _SlowResponse( + largefiles.urllib.request.urlopen = lambda req, timeout=None: _SlowResponse( PAYLOAD_A ) common.fetch_large_file("asset.bin", dest_dir=self.dir) @@ -101,12 +102,12 @@ class LargeFileCacheTest(unittest.TestCase): def fail(req, timeout=None): raise urllib.error.URLError("offline") - common.urllib.request.urlopen = fail + largefiles.urllib.request.urlopen = fail self.assertIsNone(common.fetch_large_file("asset.bin", dest_dir=self.dir)) self.assertEqual(os.listdir(self.dir), []) def test_hash_mismatch_leaves_no_scratch_file(self): - common.urllib.request.urlopen = lambda req, timeout=None: _SlowResponse( + largefiles.urllib.request.urlopen = lambda req, timeout=None: _SlowResponse( PAYLOAD_A ) result = common.fetch_large_file( @@ -122,7 +123,7 @@ class LargeFileCacheTest(unittest.TestCase): def fail(req, timeout=None): raise AssertionError("must not download when the cache is valid") - common.urllib.request.urlopen = fail + largefiles.urllib.request.urlopen = fail self.assertEqual( common.fetch_large_file("asset.bin", dest_dir=self.dir), str(cached) ) @@ -131,7 +132,7 @@ class LargeFileCacheTest(unittest.TestCase): def fail(req, timeout=None): raise AssertionError("offline mode must not open the network") - common.urllib.request.urlopen = fail + largefiles.urllib.request.urlopen = fail self.assertIsNone( common.fetch_large_file( "asset.bin", dest_dir=self.dir, offline=True @@ -146,7 +147,7 @@ class LargeFileCacheTest(unittest.TestCase): def fail(req, timeout=None): raise AssertionError("offline cache hit must not open the network") - common.urllib.request.urlopen = fail + largefiles.urllib.request.urlopen = fail self.assertEqual( common.fetch_large_file( "asset.bin",