From 8ed1a737a870bf3c2c6d7b2818cb07293a906f38 Mon Sep 17 00:00:00 2001 From: Abdessamad Derraz <3028866+Abdess@users.noreply.github.com> Date: Wed, 21 Oct 2026 09:58:21 +0200 Subject: [PATCH] fix: keep external downloads under the run policy download_external stayed in generate_pack, outside the offline switch: it fetched under --offline, wrote to the system /tmp and held the body in memory. It now lives in packresolve, honours the switch, streams into the repository tmp/ and hashes on the way. --- scripts/generate_pack.py | 69 ++------------------------------- scripts/packresolve.py | 63 +++++++++++++++++++++++++++++- tests/test_audit_regressions.py | 62 +++++++++++++++++++++++++++++ 3 files changed, 126 insertions(+), 68 deletions(-) diff --git a/scripts/generate_pack.py b/scripts/generate_pack.py index 67e79315..417d9437 100644 --- a/scripts/generate_pack.py +++ b/scripts/generate_pack.py @@ -25,8 +25,6 @@ import re import shutil import sys import tempfile -import urllib.error -import urllib.request import zipfile from pathlib import Path @@ -345,68 +343,6 @@ def _platform_pack_stem( return "_".join(_name_part(n) for n in names) + version_tag - - - - - -def download_external(file_entry: dict, dest_path: str) -> bool: - """Download an external BIOS file, verify hash, save to dest_path.""" - url = file_entry.get("source_url") - if not url: - return False - - sha256 = file_entry.get("sha256") - sha1 = file_entry.get("sha1") - md5 = file_entry.get("md5") - - if not (sha256 or sha1 or md5): - print( - f" WARNING: no hash for {file_entry['name']}, skipping unverifiable download" - ) - return False - - try: - req = urllib.request.Request( - url, headers={"User-Agent": "retrobios-pack-gen/1.0"} - ) - with urllib.request.urlopen(req, timeout=120) as resp: - data = resp.read() - except urllib.error.URLError as e: - print(f" WARNING: Failed to download {url}: {e}") - return False - - if sha256: - actual = hashlib.sha256(data).hexdigest() - if actual != sha256: - print(f" WARNING: SHA256 mismatch for {file_entry['name']}") - return False - elif sha1: - actual = hashlib.sha1(data).hexdigest() - if actual != sha1: - print(f" WARNING: SHA1 mismatch for {file_entry['name']}") - return False - elif md5: - actual = hashlib.md5(data).hexdigest() - if actual != md5: - print(f" WARNING: MD5 mismatch for {file_entry['name']}") - return False - - os.makedirs(os.path.dirname(dest_path), exist_ok=True) - with open(dest_path, "wb") as f: - f.write(data) - return True - - - - - - - - - - - # What the builder writes into a platform pack beside the files it carries. PACK_DOCUMENTS = ("README.txt", "manifest.json") @@ -952,7 +888,7 @@ def generate_pack( if status == "external": file_ext = os.path.splitext(file_entry["name"])[1] or "" with tempfile.NamedTemporaryFile( - delete=False, suffix=file_ext + delete=False, suffix=file_ext, dir=packresolve.scratch_dir() ) as tmp: tmp_path = tmp.name @@ -1709,7 +1645,7 @@ def generate_emulator_pack( if how == "external": file_ext = os.path.splitext(fe["name"])[1] or "" with tempfile.NamedTemporaryFile( - delete=False, suffix=file_ext + delete=False, suffix=file_ext, dir=packresolve.scratch_dir() ) as tmp: tmp_path = tmp.name try: @@ -4926,6 +4862,7 @@ from packresolve import ( # noqa: E402,F401 parse_hash_file, lookup_hashes, resolve_file, + download_external, ) if __name__ == "__main__": diff --git a/scripts/packresolve.py b/scripts/packresolve.py index 1d54413b..e53d5f24 100644 --- a/scripts/packresolve.py +++ b/scripts/packresolve.py @@ -6,9 +6,14 @@ and the offline switch that forbids reaching for the network.""" from __future__ import annotations -import re -from largefiles import fetch_large_file +import hashlib import os +import re +import urllib.error +import urllib.request +from pathlib import Path + +from largefiles import fetch_large_file from common import ( declared_hash_verdict, resolution_is_hash_exact, @@ -26,6 +31,60 @@ def set_offline(value: bool) -> None: global _OFFLINE _OFFLINE = value + +def scratch_dir() -> str: + """The repository's tmp/, never the system /tmp (a 4 GB tmpfs on WSL).""" + path = Path(__file__).resolve().parent.parent / "tmp" + path.mkdir(exist_ok=True) + return str(path) + + +def download_external( + file_entry: dict, dest_path: str, offline: bool | None = None +) -> bool: + """Download a `storage: external` file, verify its hash, write dest_path. + + The run's network policy applies: under --offline nothing is fetched and + the entry stays missing, as verify reports it. The body is streamed to + dest_path and hashed on the way, and removed when the hash disagrees. + """ + url = file_entry.get("source_url") + if not url: + return False + name = file_entry.get("name", url) + declared = next( + ( + (algo, str(file_entry[algo]).lower()) + for algo in ("sha256", "sha1", "md5") + if file_entry.get(algo) + ), + None, + ) + if declared is None: + print(f" WARNING: no hash for {name}, skipping unverifiable download") + return False + if _OFFLINE if offline is None else offline: + print(f" WARNING: {name} is external and the run is offline, not fetched") + return False + algo, expected = declared + digest = hashlib.new(algo) + Path(dest_path).parent.mkdir(parents=True, exist_ok=True) + try: + req = urllib.request.Request(url, headers={"User-Agent": "retrobios-pack-gen/1.0"}) + with urllib.request.urlopen(req, timeout=120) as resp, open(dest_path, "wb") as out: + for chunk in iter(lambda: resp.read(1 << 20), b""): + digest.update(chunk) + out.write(chunk) + except (urllib.error.URLError, OSError) as exc: + print(f" WARNING: Failed to download {url}: {exc}") + Path(dest_path).unlink(missing_ok=True) + return False + if digest.hexdigest() != expected: + print(f" WARNING: {algo.upper()} mismatch for {name}") + Path(dest_path).unlink(missing_ok=True) + return False + return True + def _values(raw: object) -> list[str]: """A hash field's values: one, a list, or a comma-separated string.""" items = raw if isinstance(raw, list) else [raw] if raw else [] diff --git a/tests/test_audit_regressions.py b/tests/test_audit_regressions.py index 85e7343b..a50cd08c 100644 --- a/tests/test_audit_regressions.py +++ b/tests/test_audit_regressions.py @@ -6215,5 +6215,67 @@ class CommentsSayWhy(unittest.TestCase): self.assertEqual(found, []) +class ExternalDownloadsFollowTheRunPolicy(unittest.TestCase): + """download_external lived outside packresolve and fetched under + --offline, into the system /tmp, holding the whole body in memory.""" + + ENTRY = {"name": "ext.bin", "source_url": "https://example.invalid/ext.bin", + "sha1": hashlib.sha1(b"payload").hexdigest()} + + def _response(self, body: bytes): + stream = io.BytesIO(body) + response = mock.MagicMock() + response.__enter__.return_value = stream + response.__exit__.return_value = False + return response + + def test_offline_never_reaches_the_network(self): + import packresolve + + with tempfile.TemporaryDirectory() as tmp, \ + mock.patch.object(packresolve.urllib.request, "urlopen") as urlopen, \ + contextlib.redirect_stdout(io.StringIO()): + ok = packresolve.download_external(self.ENTRY, f"{tmp}/x", offline=True) + self.assertFalse(ok) + urlopen.assert_not_called() + + def test_the_run_switch_is_inherited(self): + import packresolve + + saved = packresolve._OFFLINE + packresolve.set_offline(True) + try: + with tempfile.TemporaryDirectory() as tmp, \ + mock.patch.object(packresolve.urllib.request, "urlopen") as urlopen, \ + contextlib.redirect_stdout(io.StringIO()): + self.assertFalse(packresolve.download_external(self.ENTRY, f"{tmp}/x")) + urlopen.assert_not_called() + finally: + packresolve.set_offline(saved) + + def test_a_verified_body_lands_and_a_wrong_one_is_removed(self): + import packresolve + + with tempfile.TemporaryDirectory() as tmp: + target = Path(tmp) / "x" + with mock.patch.object(packresolve.urllib.request, "urlopen", + return_value=self._response(b"payload")): + self.assertTrue(packresolve.download_external(self.ENTRY, str(target), offline=False)) + self.assertEqual(target.read_bytes(), b"payload") + target.unlink() + with mock.patch.object(packresolve.urllib.request, "urlopen", + return_value=self._response(b"other")), \ + contextlib.redirect_stdout(io.StringIO()): + self.assertFalse(packresolve.download_external(self.ENTRY, str(target), offline=False)) + self.assertFalse(target.exists()) + + def test_scratch_files_stay_in_the_repository(self): + source = (ROOT / "scripts" / "generate_pack.py").read_text(encoding="utf-8") + self.assertNotIn("def download_external", source) + self.assertNotIn("urlopen", source) + for call in re.findall(r"NamedTemporaryFile\(([^)]*)\)", source): + self.assertIn("dir=", call) + + if __name__ == "__main__": unittest.main()