#!/usr/bin/env python3 """Universal BIOS installer for retrogaming platforms. Self-contained script using only Python stdlib. Downloads missing BIOS files from the retrobios repository and places them in the correct location for the detected emulator platform. Usage: python install.py python install.py --platform retroarch --dest ~/custom/bios python install.py --check python install.py --list-platforms """ from __future__ import annotations import argparse import concurrent.futures import hashlib import json import os import platform import re import shutil import sys import tempfile import urllib.error import urllib.parse import urllib.request import xml.etree.ElementTree as ET from pathlib import Path, PurePosixPath # Manifests are read from the same ref the bootstrap verified this installer # against, so a file list and the code reading it always come from one commit. # RETROBIOS_REF pins an installation to a tag when reproducibility matters. DEFAULT_RELEASE_REF = "main" RELEASE_REF = os.environ.get("RETROBIOS_REF", DEFAULT_RELEASE_REF) DEFAULT_BASE_URL = ( "https://raw.githubusercontent.com/Abdess/retrobios/" + urllib.parse.quote(RELEASE_REF, safe="") ) _LOOPBACK_HOSTS = ("localhost", "127.0.0.1", "::1") def _checked_base_url(value: str) -> str: """Refuse a base URL that is neither HTTPS nor loopback. This base serves the manifest and the files it declares, so whoever controls it controls the expected hashes too and verification stops proving anything. install.sh and install.ps1 make the same check on the URL they fetch the installer from. Plain HTTP to loopback stays allowed: nothing sits between the two ends to intercept it, and it is how the installer is exercised end to end. """ parsed = urllib.parse.urlparse(value) if parsed.scheme == "https": return value host = (parsed.hostname or "").lower() if parsed.scheme == "http" and host in _LOOPBACK_HOSTS: return value print( f"Error: RETROBIOS_BASE_URL must use HTTPS, got {value!r}", file=sys.stderr, ) sys.exit(1) BASE_URL = _checked_base_url(os.environ.get("RETROBIOS_BASE_URL", DEFAULT_BASE_URL)) MANIFEST_URL = f"{BASE_URL}/install/{{platform}}.json" TARGETS_URL = f"{BASE_URL}/install/targets/{{platform}}.json" RAW_FILE_URL = f"{BASE_URL}/{{path}}" RELEASE_URL = ( "https://github.com/Abdess/retrobios/releases/download/large-files/{asset}" ) MAX_RETRIES = 3 MAX_MANIFEST_BYTES = 16 * 1024 * 1024 MAX_TARGETS_BYTES = 4 * 1024 * 1024 MAX_MANIFEST_FILES = 100_000 MAX_DOWNLOAD_SIZE = 1024 * 1024 * 1024 MAX_TOTAL_DOWNLOAD_SIZE = 64 * 1024 * 1024 * 1024 _SHA1_RE = re.compile(r"^[0-9a-fA-F]{40}$") _SHA256_RE = re.compile(r"^[0-9a-fA-F]{64}$") # Platforms with a manifest in install/. Manifest URLs are case sensitive, # so user input is normalized against this list before any fetch. AVAILABLE_PLATFORMS = ( "retroarch", "batocera", "recalbox", "retrobat", "emudeck", "lakka", "retrodeck", "rocknix", "romm", "bizhawk", "misterfpga", "retropie", ) # Fallback BIOS destination per platform when --platform is forced # but auto-detection finds nothing on the machine. DEFAULT_DESTS = { "batocera": Path("/userdata/bios"), "recalbox": Path("/recalbox/share/bios"), "lakka": Path("/storage/system"), "retrodeck": Path.home() / "retrodeck", "emudeck": Path.home() / "Emulation" / "bios", "rocknix": Path("/storage/roms/bios"), "misterfpga": Path("/media/fat/games"), "retropie": Path.home() / "RetroPie" / "BIOS", } def detect_os() -> str: """Return normalized OS identifier.""" system = platform.system().lower() if system == "linux": proc_version = Path("/proc/version") if proc_version.exists(): try: content = proc_version.read_text(encoding="utf-8", errors="replace") if "microsoft" in content.lower(): return "wsl" except OSError: pass return "linux" if system == "darwin": return "darwin" if system == "windows": return "windows" return system def _parse_os_release() -> dict[str, str]: """Parse /etc/os-release KEY=value format.""" result: dict[str, str] = {} path = Path("/etc/os-release") if not path.exists(): return result try: for line in path.read_text(encoding="utf-8", errors="replace").splitlines(): line = line.strip() if "=" not in line or line.startswith("#"): continue key, _, value = line.partition("=") value = value.strip('"').strip("'") result[key] = value except OSError: pass return result def _expand_retroarch_path(value: str, app_dir: Path) -> Path: """Expand the notations RetroArch writes into its config values. fill_pathname_expand_special (libretro-common/file/file_path.c) maps a leading '~' to the home directory and a leading ':' to the application directory, dropping the two leading characters in both cases. """ if value[:1] == "~": return Path(str(Path.home())) / value[2:] if value[:1] == ":": return app_dir / value[2:] return Path(os.path.expandvars(os.path.expanduser(value))) def _parse_retroarch_system_dir( cfg_path: Path, app_dir: Path | None = None ) -> Path | None: """Parse system_directory from retroarch.cfg. app_dir is RetroArch's application directory, used to expand ':' values. It defaults to the directory holding the config, which is where a portable install keeps both. """ if not cfg_path.exists(): return None app_dir = app_dir or cfg_path.parent try: for line in cfg_path.read_text(encoding="utf-8", errors="replace").splitlines(): line = line.strip() if line.startswith("system_directory"): _, _, value = line.partition("=") value = value.strip().strip('"').strip("'") if not value or value == "default": return app_dir / "system" return _expand_retroarch_path(value, app_dir) except OSError: pass return None def _shell_unquote(value: str) -> str: """Resolve a shell right-hand side into its effective string value. Handles concatenated quoted/unquoted segments the way bash does (e.g. "/run/media/deck/EmuSD"/Emulation), expands variables in double-quoted and unquoted segments, and stops at unquoted whitespace or a comment. """ parts: list[str] = [] i = 0 n = len(value) while i < n: c = value[i] if c == '"': end = value.find('"', i + 1) if end == -1: parts.append(os.path.expandvars(value[i + 1:])) break parts.append(os.path.expandvars(value[i + 1:end])) i = end + 1 elif c == "'": end = value.find("'", i + 1) if end == -1: parts.append(value[i + 1:]) break parts.append(value[i + 1:end]) i = end + 1 elif c.isspace() or c == "#": break else: end = i while end < n and value[end] not in "\"'#" and not value[end].isspace(): end += 1 parts.append(os.path.expandvars(value[i:end])) i = end return os.path.expanduser("".join(parts)) def _parse_bash_var(path: Path, key: str) -> str | None: """Extract the effective value of key= from a bash/shell file.""" if not path.exists(): return None try: for line in path.read_text(encoding="utf-8", errors="replace").splitlines(): line = line.strip() if line.startswith(f"{key}="): _, _, value = line.partition("=") return _shell_unquote(value) except OSError: pass return None def _parse_json_path(path: Path, *keys: str) -> str | None: """Extract a nested string value from a JSON file.""" if not path.exists(): return None try: data = json.loads(path.read_text(encoding="utf-8", errors="replace")) for key in keys: data = data[key] return data if isinstance(data, str) and data else None except (OSError, json.JSONDecodeError, KeyError, TypeError): return None def _parse_ps1_var(path: Path, key: str) -> str | None: """Extract value of $key= or key= from a PowerShell file.""" if not path.exists(): return None normalized = key.lstrip("$") try: for line in path.read_text(encoding="utf-8", errors="replace").splitlines(): line = line.strip() check = line.lstrip("$") if check.startswith(f"{normalized}="): _, _, value = check.partition("=") return value.strip('"').strip("'") except OSError: pass return None def _detect_embedded() -> list[tuple[str, Path]]: """Check for embedded Linux retrogaming OSes.""" found: list[tuple[str, Path]] = [] os_release = _parse_os_release() os_id = os_release.get("ID", "").lower() if os_id == "rocknix": found.append(("rocknix", Path("/storage/roms/bios"))) return found # MiSTer keeps the SD card at /media/fat with the main binary at its root # (Main_MiSTer file_io.cpp:1150-1156), and cores read from games/ # (file_io.h:168) if Path("/media/fat/MiSTer").exists(): found.append(("misterfpga", Path("/media/fat/games"))) return found if Path("/etc/knulli-release").exists(): found.append(("batocera", Path("/userdata/bios"))) return found if os_id == "lakka": found.append(("lakka", Path("/storage/system"))) return found if Path("/etc/batocera-version").exists(): found.append(("batocera", Path("/userdata/bios"))) return found if ( Path("/recalbox/recalbox.version").exists() or Path("/usr/bin/recalbox-settings").exists() ): found.append(("recalbox", Path("/recalbox/share/bios"))) return found if Path("/opt/muos").exists() or Path("/mnt/mmc/MUOS/").exists(): found.append(("retroarch", Path("/mnt/mmc/MUOS/bios"))) return found if Path("/home/ark").exists() and Path("/opt/system").exists(): found.append(("retroarch", Path("/roms/bios"))) return found if Path("/mnt/vendor/bin/dmenu.bin").exists(): found.append(("retroarch", Path("/mnt/mmc/bios"))) return found return found def _lnk_target(lnk_path: Path, exe_name: str) -> Path | None: """Read the target path of a Windows shortcut. Shortcuts embed the target as a plain local path; scanning for it avoids parsing the whole .lnk structure for the one field that matters here. """ try: if lnk_path.stat().st_size > 64 * 1024: return None blob = lnk_path.read_bytes() except OSError: return None pattern = re.compile( rb"[A-Za-z]:\\[ -~]{0,260}?" + re.escape(exe_name.encode()), re.IGNORECASE ) match = pattern.search(blob) if not match: return None return Path(match.group().decode("ascii", "replace").replace("\\", "/")) def launchbox_root(os_type: str) -> Path | None: """Locate the LaunchBox installation directory. LaunchBox installs wherever the user points its installer and writes no uninstall registry key, so the Start menu shortcut is the only record of that choice. Everything else falls back to the default location. """ if os_type not in ("windows", "wsl"): return None candidates: list[Path] = [] appdata = os.environ.get("APPDATA", "") if appdata: lnk = ( Path(appdata) / "Microsoft" / "Windows" / "Start Menu" / "Programs" / "LaunchBox" / "LaunchBox.lnk" ) if lnk.exists(): target = _lnk_target(lnk, "LaunchBox.exe") if target: # The shortcut points at Core/LaunchBox.exe candidates.append(target.parent.parent) candidates.append(target.parent) userprofile = os.environ.get("USERPROFILE", "") if userprofile: candidates.append(Path(userprofile) / "LaunchBox") for root in candidates: if (root / "Data" / "Emulators.xml").exists(): return root return None def launchbox_emulators(emulators_xml: Path) -> dict[str, Path]: """Map each emulator LaunchBox knows about to its installation directory. LaunchBox stores its emulator list in Data/Emulators.xml; ApplicationPath is either absolute or relative to the LaunchBox root directory. Keys are the lowercase executable names. """ found: dict[str, Path] = {} if not emulators_xml.exists(): return found try: if emulators_xml.stat().st_size > 10 * 1024 * 1024: return found raw = emulators_xml.read_text(encoding="utf-8", errors="replace") # ElementTree expands internal entities; LaunchBox writes no doctype if " Path | None: """Resolve the system dir of a RetroArch referenced by LaunchBox. Mirrors the plugin's UpdateSystemPath: read system_directory from the retroarch.cfg sitting next to the executable, else assume system/. """ emu_dir = launchbox_emulators(emulators_xml).get("retroarch.exe") if emu_dir is None: return None system_dir = _parse_retroarch_system_dir(emu_dir / "retroarch.cfg", emu_dir) return system_dir or emu_dir / "system" def _pcsx2_bios_dir(emu_dir: Path) -> Path: """Resolve the BIOS folder of a PCSX2 installed under LaunchBox. EmuFolders (pcsx2/Pcsx2Config.cpp) enters portable mode when portable.ini or portable.txt sits next to the executable, and then takes the data root from the contents of portable.txt, relative to the executable. Otherwise the data root is Documents/PCSX2. Bios defaults to "bios" below it and the Bios key of PCSX2.ini overrides that. """ documents = Path(os.environ.get("USERPROFILE", str(Path.home()))) / "Documents" portable_txt = emu_dir / "portable.txt" portable = (emu_dir / "portable.ini").exists() or portable_txt.exists() if portable: subpath = "" try: subpath = portable_txt.read_text( encoding="utf-8", errors="replace" ).strip() except OSError: pass data_root = emu_dir / subpath if subpath else emu_dir else: data_root = documents / "PCSX2" ini = data_root / "inis" / "PCSX2.ini" try: for line in ini.read_text(encoding="utf-8", errors="replace").splitlines(): if line.startswith("Bios = "): value = line[len("Bios = "):].strip() candidate = Path(value) return candidate if candidate.is_absolute() else data_root / value except OSError: pass return data_root / "bios" def _xemu_bios_dir(emu_dir: Path) -> Path: """Resolve the BIOS folder of a xemu managed by LaunchBox. Follows the plugin: xemu.toml next to the executable else the roaming copy, with bootrom_path and flashrom_path naming a file whose directory holds the images. Both default to bios/ under the executable. """ appdata = os.environ.get("APPDATA", "") toml = emu_dir / "xemu.toml" if not toml.exists() and appdata: toml = Path(appdata) / "xemu" / "xemu" / "xemu.toml" if toml.exists(): try: for line in toml.read_text(encoding="utf-8", errors="replace").splitlines(): if line.startswith(("bootrom_path", "flashrom_path")): parts = line.split("'") if len(parts) > 1 and Path(parts[1]).exists(): return Path(parts[1]).parent except OSError: pass return emu_dir / "bios" def launchbox_bios_dirs(root: Path) -> dict[str, Path]: """Map emulator ids to the BIOS directory LaunchBox expects for them. Only emulators whose destination LaunchBox itself computes are listed; the rest keep the default locations declared in the manifest. """ emulators = launchbox_emulators(root / "Data" / "Emulators.xml") dirs: dict[str, Path] = {} emu_dir = emulators.get("pcsx2.exe") if emu_dir: dirs["pcsx2"] = _pcsx2_bios_dir(emu_dir) emu_dir = emulators.get("xemu.exe") if emu_dir: dirs["xemu"] = _xemu_bios_dir(emu_dir) emu_dir = emulators.get("dolphin.exe") if emu_dir and (emu_dir / "portable.txt").exists(): # portable.txt beside the executable moves the user directory to # User (SetUserDirectory in Source/Core/UICommon/UICommon.cpp, # PORTABLE_USER_DIR in Source/Core/Common/CommonPaths.h) dirs["dolphin"] = emu_dir / "User" return dirs def detect_frontends(os_type: str, home: Path | None = None) -> list[str]: """Detect frontends that reference emulators without owning a BIOS dir. ES-DE resolves its application data directory to $ESDE_APPDATA_DIR or /ES-DE (es-core FileSystemUtil.cpp getAppDataDirectory). """ home = home or Path.home() frontends: list[str] = [] esde_env = os.environ.get("ESDE_APPDATA_DIR", "") if (esde_env and Path(esde_env).is_dir()) or (home / "ES-DE").is_dir(): frontends.append("esde") if launchbox_root(os_type) is not None: frontends.append("launchbox") return frontends def detect_platforms(os_type: str) -> list[tuple[str, Path]]: """Detect installed emulator platforms and their BIOS directories.""" found: list[tuple[str, Path]] = [] if os_type in ("linux", "wsl"): found.extend(_detect_embedded()) # EmuDeck (Linux/SteamOS) home = Path.home() emudeck_settings = home / ".config" / "EmuDeck" / "settings.sh" if emudeck_settings.exists(): emu_path = _parse_bash_var(emudeck_settings, "emulationPath") if emu_path: bios_dir = Path(emu_path) / "bios" found.append(("emudeck", bios_dir)) # RetroDECK: retrodeck.json since the cfg-to-json migration, which # renames the old retrodeck.cfg to retrodeck.bak (global.sh:149-153) retrodeck_conf_dir = home / ".var" / "app" / "net.retrodeck.retrodeck" / "config" / "retrodeck" retrodeck_json = retrodeck_conf_dir / "retrodeck.json" retrodeck_cfg = retrodeck_conf_dir / "retrodeck.cfg" if retrodeck_json.exists(): rd_home = _parse_json_path(retrodeck_json, "paths", "rd_home_path") found.append(("retrodeck", Path(rd_home) if rd_home else home / "retrodeck")) elif retrodeck_cfg.exists(): rd_home = _parse_bash_var(retrodeck_cfg, "rdhome") found.append(("retrodeck", Path(rd_home) if rd_home else home / "retrodeck")) # RetroArch Flatpak flatpak_cfg = home / ".var" / "app" / "org.libretro.RetroArch" / "config" / "retroarch" / "retroarch.cfg" ra_dir = _parse_retroarch_system_dir(flatpak_cfg) if ra_dir: found.append(("retroarch", ra_dir)) # RetroArch Snap snap_cfg = home / "snap" / "retroarch" / "current" / ".config" / "retroarch" / "retroarch.cfg" ra_dir = _parse_retroarch_system_dir(snap_cfg) if ra_dir: found.append(("retroarch", ra_dir)) # RetroArch native native_cfg = home / ".config" / "retroarch" / "retroarch.cfg" ra_dir = _parse_retroarch_system_dir(native_cfg) if ra_dir: found.append(("retroarch", ra_dir)) if os_type == "darwin": home = Path.home() mac_cfg = home / "Library" / "Application Support" / "RetroArch" / "retroarch.cfg" ra_dir = _parse_retroarch_system_dir(mac_cfg) if ra_dir: found.append(("retroarch", ra_dir)) if os_type in ("windows", "wsl"): # EmuDeck Windows home = Path.home() emudeck_ps1 = Path(os.environ.get("APPDATA", "")) / "EmuDeck" / "settings.ps1" if emudeck_ps1.exists(): emu_path = _parse_ps1_var(emudeck_ps1, "$emulationPath") if emu_path: found.append(("emudeck", Path(emu_path) / "bios")) # RetroArch Windows appdata = os.environ.get("APPDATA", "") if appdata: win_cfg = Path(appdata) / "RetroArch" / "retroarch.cfg" ra_dir = _parse_retroarch_system_dir(win_cfg) if ra_dir: found.append(("retroarch", ra_dir)) # Portable RetroArch referenced by LaunchBox lb_root = launchbox_root(os_type) if lb_root and not any(name == "retroarch" for name, _ in found): system_dir = _launchbox_retroarch_system_dir( lb_root / "Data" / "Emulators.xml" ) if system_dir: found.append(("retroarch", system_dir)) return found def normalize_platform(name: str) -> str: """Return the canonical platform id for user-supplied input.""" plat = name.strip().lower() if plat not in AVAILABLE_PLATFORMS: print(f"Unknown platform '{name}'. Available platforms:", file=sys.stderr) for p in AVAILABLE_PLATFORMS: print(f" {p}", file=sys.stderr) sys.exit(1) return plat def _read_limited_json(response, limit: int, label: str) -> object: """Read a bounded UTF-8 JSON response.""" raw_length = response.headers.get("Content-Length") if response.headers else None if raw_length: try: if int(raw_length) > limit: raise ValueError(f"{label} exceeds {limit} bytes") except ValueError as exc: if "exceeds" in str(exc): raise payload = response.read(limit + 1) if len(payload) > limit: raise ValueError(f"{label} exceeds {limit} bytes") try: return json.loads(payload.decode("utf-8")) except (UnicodeDecodeError, json.JSONDecodeError) as exc: raise ValueError(f"invalid {label}: {exc}") from exc def _safe_relative_path(value: object, field: str) -> PurePosixPath: """Validate a manifest-controlled relative POSIX path.""" if not isinstance(value, str) or not value or len(value) > 1024: raise ValueError(f"invalid {field}") if ( "\\" in value or "\x00" in value or "//" in value or value.endswith("/") or re.match(r"^[A-Za-z]:", value) ): raise ValueError(f"unsafe {field}: {value!r}") path = PurePosixPath(value) if path.is_absolute() or any(part in ("", ".", "..") for part in path.parts): raise ValueError(f"unsafe {field}: {value!r}") return path def _within(candidate: Path, root: Path) -> bool: """Whether *candidate* stays under *root* once both are resolved.""" try: (root / candidate.name).resolve().relative_to(root) except (OSError, ValueError): return False return True def _safe_target_dir(value: object, field: str) -> str: """Validate a manifest-controlled standalone-copy directory. These name emulator install directories, so unlike a BIOS destination they are legitimately absolute and outside the BIOS tree. What they must never do is climb: a '..' anywhere turns "copy next to the emulator" into "write wherever the manifest likes". """ if not isinstance(value, str) or not value or len(value) > 2048: raise ValueError(f"invalid {field}") if "\x00" in value: raise ValueError(f"unsafe {field}: {value!r}") parts = re.split(r"[\\/]", value) if any(part == ".." for part in parts): raise ValueError(f"unsafe {field}: {value!r}") return value def _destination_path(root: Path, value: object) -> Path: """Resolve a manifest destination and prove it remains below *root*.""" relative = _safe_relative_path(value, "dest") resolved_root = root.resolve() candidate = (resolved_root / Path(*relative.parts)).resolve() try: candidate.relative_to(resolved_root) except ValueError as exc: raise ValueError(f"destination escapes BIOS directory: {value!r}") from exc return candidate def _validate_manifest(data: object, plat: str) -> dict: """Validate the untrusted install-manifest boundary using stdlib only.""" if not isinstance(data, dict): raise ValueError("manifest root must be an object") if data.get("manifest_version") not in (1, 2): raise ValueError("unsupported manifest_version") if data.get("platform") != plat: raise ValueError("manifest platform does not match request") files = data.get("files") if not isinstance(files, list) or len(files) > MAX_MANIFEST_FILES: raise ValueError("invalid manifest files list") seen_destinations: set[str] = set() total_size = 0 for index, entry in enumerate(files): if not isinstance(entry, dict): raise ValueError(f"files[{index}] must be an object") dest = str(_safe_relative_path(entry.get("dest"), f"files[{index}].dest")) if dest in seen_destinations: raise ValueError(f"duplicate manifest destination: {dest}") seen_destinations.add(dest) size = entry.get("size") if isinstance(size, bool) or not isinstance(size, int) or not (0 <= size <= MAX_DOWNLOAD_SIZE): raise ValueError(f"invalid size for {dest}") total_size += size if total_size > MAX_TOTAL_DOWNLOAD_SIZE: raise ValueError("manifest total size exceeds safety limit") sha1 = entry.get("sha1", "") sha256 = entry.get("sha256", "") if sha1 and (not isinstance(sha1, str) or not _SHA1_RE.fullmatch(sha1)): raise ValueError(f"invalid SHA1 for {dest}") if sha256 and ( not isinstance(sha256, str) or not _SHA256_RE.fullmatch(sha256) ): raise ValueError(f"invalid SHA256 for {dest}") if not sha1 and not sha256: raise ValueError(f"missing content hash for {dest}") release_asset = entry.get("release_asset") repo_path = entry.get("repo_path") if release_asset: asset = _safe_relative_path(release_asset, f"files[{index}].release_asset") if len(asset.parts) != 1: raise ValueError(f"release asset must be a basename: {release_asset}") elif repo_path: source = _safe_relative_path(repo_path, f"files[{index}].repo_path") if source.parts[0] != "bios": raise ValueError(f"repo_path outside bios/: {repo_path}") else: raise ValueError(f"no download source for {dest}") cores = entry.get("cores") if cores is not None and ( not isinstance(cores, list) or not all(isinstance(core, str) for core in cores) ): raise ValueError(f"invalid cores list for {dest}") declared_total_files = data.get("total_files") if declared_total_files is not None and declared_total_files != len(files): raise ValueError("manifest total_files does not match files list") declared_total_size = data.get("total_size") if declared_total_size is not None and declared_total_size != total_size: raise ValueError("manifest total_size does not match file sizes") omitted = data.get("omitted_files", []) if not isinstance(omitted, list) or len(omitted) > MAX_MANIFEST_FILES: raise ValueError("invalid omitted_files list") seen_omitted: set[str] = set() allowed_omission_reasons = { "hash_mismatch", "not_found", "external", "user_provided" } for index, entry in enumerate(omitted): if not isinstance(entry, dict): raise ValueError(f"omitted_files[{index}] must be an object") dest = str( _safe_relative_path( entry.get("dest"), f"omitted_files[{index}].dest" ) ) if dest in seen_destinations or dest in seen_omitted: raise ValueError(f"duplicate or conflicting omitted destination: {dest}") seen_omitted.add(dest) if not isinstance(entry.get("name"), str) or not entry["name"]: raise ValueError(f"invalid omitted file name for {dest}") if not isinstance(entry.get("system", ""), str): raise ValueError(f"invalid omitted system for {dest}") if not isinstance(entry.get("required"), bool): raise ValueError(f"invalid omitted required flag for {dest}") if entry.get("reason") not in allowed_omission_reasons: raise ValueError(f"invalid omission reason for {dest}") cores = entry.get("cores") if cores is not None and ( not isinstance(cores, list) or not all(isinstance(core, str) for core in cores) ): raise ValueError(f"invalid omitted cores list for {dest}") declared_total_omitted = data.get("total_omitted") if ( declared_total_omitted is not None and declared_total_omitted != len(omitted) ): raise ValueError("manifest total_omitted does not match omitted_files") copies = data.get("standalone_copies", []) if not isinstance(copies, list) or len(copies) > 10_000: raise ValueError("invalid standalone_copies") for index, entry in enumerate(copies): if not isinstance(entry, dict): raise ValueError(f"standalone_copies[{index}] must be an object") if "file" in entry: _safe_relative_path( entry["file"], f"standalone_copies[{index}].file" ) if "pattern" in entry: pattern = entry["pattern"] if ( not isinstance(pattern, str) or not pattern or len(pattern) > 256 or "/" in pattern or "\\" in pattern or ".." in pattern ): raise ValueError(f"invalid standalone copy pattern: {pattern!r}") targets = entry.get("targets", {}) if targets and not isinstance(targets, dict): raise ValueError(f"invalid standalone copy targets at index {index}") for values in (targets or {}).values(): if not isinstance(values, list) or len(values) > 100: raise ValueError(f"invalid standalone copy targets at index {index}") for value in values: _safe_target_dir(value, f"standalone_copies[{index}].targets") return data def _validate_targets(data: object) -> dict[str, dict]: """Validate and normalize legacy list-valued target manifests. A null core list means the target publishes no core inventory. That is a known target with no filter, not a broken manifest: rejecting it would discard every other target on the platform. """ if not isinstance(data, dict) or len(data) > 10_000: raise ValueError("invalid targets manifest") normalized: dict[str, dict] = {} for target, value in data.items(): if not isinstance(target, str) or not target or len(target) > 128: raise ValueError("invalid target name") if isinstance(value, dict): cores = value.get("cores") else: cores = value if cores is None: normalized[target] = {"cores": None} continue if not isinstance(cores, list) or len(cores) > 10_000 or not all( isinstance(core, str) and 0 < len(core) <= 256 for core in cores ): raise ValueError(f"invalid core list for target {target}") normalized[target] = {"cores": cores} return normalized def fetch_manifest(plat: str) -> dict: """Download platform manifest JSON.""" url = MANIFEST_URL.format(platform=plat) try: with urllib.request.urlopen(url, timeout=30) as resp: return _validate_manifest( _read_limited_json(resp, MAX_MANIFEST_BYTES, "manifest"), plat ) except (urllib.error.URLError, urllib.error.HTTPError, OSError, ValueError) as exc: print(f" Failed to fetch manifest for {plat}: {exc}", file=sys.stderr) sys.exit(1) def fetch_targets(plat: str) -> dict: """Download target core list. Returns empty dict on 404.""" url = TARGETS_URL.format(platform=plat) try: with urllib.request.urlopen(url, timeout=30) as resp: return _validate_targets( _read_limited_json(resp, MAX_TARGETS_BYTES, "targets manifest") ) except urllib.error.HTTPError as exc: if exc.code == 404: return {} print(f" Warning: failed to fetch targets for {plat}: {exc}", file=sys.stderr) return {} except (urllib.error.URLError, OSError, ValueError): return {} def _filter_by_target( files: list[dict], target_cores: list[str] ) -> list[dict]: """Keep files where cores is None or overlaps with target_cores.""" result: list[dict] = [] target_set = set(target_cores) for f in files: cores = f.get("cores") if cores is None or any(c in target_set for c in cores): result.append(f) return result def _digest_file(path: Path, algorithms: tuple[str, ...]) -> dict[str, str]: """Compute several digests of a file in a single read. Checking a 3 GB collection against both a SHA-1 and a SHA-256 used to walk every file twice, single threaded. Hashing is cheap next to the I/O, so the read is what has to happen once. """ hashers = {name: hashlib.new(name) for name in algorithms} with open(path, "rb") as fh: for chunk in iter(lambda: fh.read(1024 * 1024), b""): for hasher in hashers.values(): hasher.update(chunk) return {name: hasher.hexdigest() for name, hasher in hashers.items()} def _sha1_file(path: Path) -> str: """Compute SHA1 of a file.""" return _digest_file(path, ("sha1",))["sha1"] def _sha256_file(path: Path) -> str: """Compute SHA256 of a file.""" return _digest_file(path, ("sha256",))["sha256"] def _classify_local(entry: dict, dest: Path) -> str: """Return 'up_to_date' or 'mismatched' for a file already on disk.""" expected = { name: entry.get(name, "").lower() for name in ("sha256", "sha1") if entry.get(name) } if not expected: return "up_to_date" actual = _digest_file(dest, tuple(expected)) if all(actual[name] == value for name, value in expected.items()): return "up_to_date" return "mismatched" def check_local( files: list[dict], bios_path: Path, jobs: int = 8 ) -> tuple[list[dict], list[dict], list[dict]]: """Check which files exist locally and have correct hashes. Returns (to_download, up_to_date, mismatched). """ to_download: list[dict] = [] present: list[tuple[dict, Path]] = [] for f in files: dest = _destination_path(bios_path, f["dest"]) if dest.exists(): present.append((f, dest)) else: to_download.append(f) up_to_date: list[dict] = [] mismatched: list[dict] = [] if present: with concurrent.futures.ThreadPoolExecutor(max_workers=jobs) as pool: verdicts = pool.map( lambda item: (item[0], _classify_local(item[0], item[1])), present ) for entry, verdict in verdicts: if verdict == "up_to_date": up_to_date.append(entry) else: mismatched.append(entry) return to_download, up_to_date, mismatched def _download_one( f: dict, bios_path: Path, verbose: bool = False ) -> tuple[str, bool]: """Download a single file. Returns (dest, success).""" try: dest = _destination_path(bios_path, f["dest"]) except ValueError: return str(f.get("dest", "?")), False dest.parent.mkdir(parents=True, exist_ok=True) if f.get("release_asset"): url = RELEASE_URL.format(asset=urllib.parse.quote(f["release_asset"], safe="")) else: url = RAW_FILE_URL.format(path=urllib.parse.quote(f["repo_path"], safe="/")) for attempt in range(1, MAX_RETRIES + 1): tmp_path: Path | None = None try: with urllib.request.urlopen(url, timeout=60) as resp: expected_size = f["size"] raw_length = resp.headers.get("Content-Length") if resp.headers else None if raw_length and int(raw_length) != expected_size: raise ValueError( f"Content-Length {raw_length} != expected {expected_size}" ) with tempfile.NamedTemporaryFile( mode="wb", dir=dest.parent, prefix=f".{dest.name}.", suffix=".part", delete=False, ) as out: tmp_path = Path(out.name) downloaded = 0 while True: chunk = resp.read(1024 * 1024) if not chunk: break downloaded += len(chunk) if downloaded > expected_size or downloaded > MAX_DOWNLOAD_SIZE: raise ValueError("download exceeded declared size") out.write(chunk) if downloaded != expected_size: raise ValueError( f"downloaded {downloaded} bytes; expected {expected_size}" ) expected_sha256 = f.get("sha256", "") expected_sha1 = f.get("sha1", "") if expected_sha256 and _sha256_file(tmp_path) != expected_sha256.lower(): if verbose: print(f" SHA256 mismatch on attempt {attempt}", file=sys.stderr) tmp_path.unlink(missing_ok=True) continue if expected_sha1: actual = _sha1_file(tmp_path) if actual != expected_sha1.lower(): if verbose: print(f" SHA1 mismatch on attempt {attempt}", file=sys.stderr) tmp_path.unlink(missing_ok=True) continue os.replace(tmp_path, dest) return f["dest"], True except ( urllib.error.URLError, urllib.error.HTTPError, OSError, ValueError, ) as exc: if verbose: print(f" Attempt {attempt} failed: {exc}", file=sys.stderr) if tmp_path is not None: tmp_path.unlink(missing_ok=True) return f["dest"], False def download_files( files: list[dict], bios_path: Path, jobs: int = 8, verbose: bool = False ) -> list[str]: """Download files in parallel. Returns list of failed file names.""" failed: list[str] = [] total = len(files) with concurrent.futures.ThreadPoolExecutor(max_workers=jobs) as pool: future_map = { pool.submit(_download_one, f, bios_path, verbose): f for f in files } done_count = 0 for future in concurrent.futures.as_completed(future_map): done_count += 1 dest, success = future.result() status = "ok" if success else "FAILED" print(f" [{done_count}/{total}] {dest} {status}") if not success: failed.append(dest) return failed def do_standalone_copies( manifest: dict, bios_path: Path, os_type: str, extra_dirs: dict[str, Path] | None = None, ) -> tuple[int, int]: """Copy BIOS files to standalone emulator directories. Supports: - file: single file copy - pattern: glob match (e.g. "scph*.bin") - note: informational message when detect path exists - WSL fallback to linux targets extra_dirs maps the emulator id of an entry to a root that replaces the default per-OS locations, for setups such as LaunchBox that keep their emulators outside them. The layout below the root is the same. Returns (copied_count, skipped_count). """ from fnmatch import fnmatch copies = manifest.get("standalone_copies", []) if not copies: return 0, 0 copied = 0 skipped = 0 for entry in copies: # Note entries: print message if emulator detected if "note" in entry: detect_paths = entry.get("detect", {}).get(os_type, []) if not detect_paths and os_type == "wsl": detect_paths = entry.get("detect", {}).get("linux", []) for dp in detect_paths: expanded = Path(os.path.expandvars(os.path.expanduser(dp))) if expanded.is_dir(): print(f" {entry['note']}") break continue # Resolve source files if "pattern" in entry: sources = [ f for f in bios_path.rglob("*") if fnmatch(f.name, entry["pattern"]) and f.is_file() ] else: src = bios_path / entry["file"] sources = [src] if src.exists() else [] if not sources: continue # Resolve target directories with WSL fallback targets = entry.get("targets", {}).get(os_type, []) if not targets and os_type == "wsl": targets = entry.get("targets", {}).get("linux", []) target_dirs = [ Path(os.path.expandvars(os.path.expanduser(t))) for t in targets ] extra_root = (extra_dirs or {}).get(entry.get("emulator", "")) if extra_root is not None: subdir = PurePosixPath(entry.get("file", "")).parent extra = extra_root if str(subdir) == "." else extra_root / subdir if extra not in target_dirs: target_dirs.append(extra) for target_dir in target_dirs: if not target_dir.is_dir(): skipped += len(sources) continue resolved_dir = target_dir.resolve() for src in sources: dest = target_dir / src.name # A symlink already sitting at the destination would redirect # the write outside the directory the user opted into, and a # crafted source name would climb out of it. if dest.is_symlink() or not _within(dest, resolved_dir): skipped += 1 continue try: shutil.copy2(src, dest) copied += 1 except OSError: skipped += 1 return copied, skipped def format_size(n: int) -> str: """Human-readable file size.""" if n < 1024: return f"{n} B" if n < 1024 * 1024: return f"{n / 1024:.1f} KB" if n < 1024 * 1024 * 1024: return f"{n / (1024 * 1024):.1f} MB" return f"{n / (1024 * 1024 * 1024):.1f} GB" def _prompt_platform_choice( platforms: list[tuple[str, Path]], ) -> list[tuple[str, Path]]: """Prompt user to choose among detected platforms.""" print("\nInstall for:") for i, (name, path) in enumerate(platforms, 1): print(f" {i}) {name.capitalize()} ({path})") if len(platforms) > 1: print(f" {len(platforms) + 1}) All") print() while True: try: choice = input("> ").strip() except (EOFError, KeyboardInterrupt): print() sys.exit(0) if not choice: continue try: idx = int(choice) except ValueError: continue if 1 <= idx <= len(platforms): return [platforms[idx - 1]] if idx == len(platforms) + 1 and len(platforms) > 1: return platforms def main() -> None: """Entry point.""" parser = argparse.ArgumentParser( description="Download missing BIOS files for retrogaming emulators.", ) parser.add_argument( "--platform", help="target platform (retroarch, batocera, emudeck, ...)", ) parser.add_argument( "--dest", type=Path, help="override BIOS destination directory", ) parser.add_argument( "--target", help="hardware target for core filtering (switch, rpi4, ...)", ) parser.add_argument( "--check", action="store_true", help="check existing files without downloading", ) parser.add_argument( "--list-platforms", action="store_true", help="list detected platforms and exit", ) parser.add_argument( "--list-targets", action="store_true", help="list available targets for a platform and exit", ) parser.add_argument( "--jobs", "-j", type=int, default=8, help="parallel download threads (default: 8)", ) parser.add_argument( "--verbose", "-v", action="store_true", help="verbose output", ) parser.add_argument( "--standalone-copies", action="store_true", help="opt in to copies into detected standalone-emulator directories", ) args = parser.parse_args() if not 1 <= args.jobs <= 32: parser.error("--jobs must be between 1 and 32") print("RetroBIOS\n") os_type = detect_os() # Early exit for listing if args.list_platforms: print("Available platforms:") for p in AVAILABLE_PLATFORMS: print(f" {p}") detected = detect_platforms(os_type) if detected: print("\nDetected on this system:") for name, path in detected: print(f" {name}: {path}") return # Platform detection or override if args.platform: args.platform = normalize_platform(args.platform) if args.platform and args.dest: platforms = [(args.platform, args.dest)] elif args.platform: print("Detecting platform...") detected = detect_platforms(os_type) matched = [(n, p) for n, p in detected if n == args.platform] if matched: platforms = matched else: default_dest = DEFAULT_DESTS.get(args.platform, Path.home() / "bios") print( f" Platform '{args.platform}' not detected, " f"using default path: {default_dest}" ) platforms = [(args.platform, default_dest)] elif args.dest: print(f" Using destination: {args.dest}") platforms = [("retroarch", args.dest)] else: print("Detecting platform...") frontends = detect_frontends(os_type) if "esde" in frontends: print(" Found ES-DE (BIOS files go to each emulator, not to ES-DE itself)") if "launchbox" in frontends: print(" Found LaunchBox") platforms = detect_platforms(os_type) if not platforms: print(" No supported platform detected.") print(" Use --platform and --dest to specify manually.") sys.exit(1) for name, path in platforms: print(f" Found {name.capitalize()} at {path}") if len(platforms) > 1 and not args.list_targets and sys.stdin.isatty(): platforms = _prompt_platform_choice(platforms) total_downloaded = 0 total_up_to_date = 0 total_errors = 0 total_omitted = 0 for plat_name, bios_path in platforms: print(f"\nFetching file index for {plat_name}...") manifest = fetch_manifest(plat_name) files = manifest.get("files", []) omitted_files = manifest.get("omitted_files", []) if args.list_targets: targets = fetch_targets(plat_name) if not targets: print(f" No targets available for {plat_name}") else: for t in sorted(targets.keys()): cores = targets[t].get("cores") label = "no core list" if cores is None else f"{len(cores)} cores" print(f" {t} ({label})") continue # Target filtering if args.target: targets = fetch_targets(plat_name) target_info = targets.get(args.target) if target_info is None: # Carrying on would install every file, which is the opposite # of what a filter was asked for: a 4 GB download instead of # the subset the target needs. Refuse rather than ignore. print( f"Error: unknown target '{args.target}' for {plat_name}.", file=sys.stderr, ) if targets: print( f" available: {', '.join(sorted(targets))}", file=sys.stderr ) else: print( f" {plat_name} publishes no target list", file=sys.stderr ) sys.exit(1) elif target_info.get("cores") is None: print( f" Target '{args.target}' publishes no core list; " "installing every file" ) else: target_cores = target_info["cores"] before = len(files) files = _filter_by_target(files, target_cores) omitted_files = _filter_by_target(omitted_files, target_cores) print(f" Filtered {before} -> {len(files)} files for target {args.target}") total_size = sum(f.get("size", 0) for f in files) print(f" {len(files)} files ({format_size(total_size)})") if omitted_files: required_omitted = sum( 1 for entry in omitted_files if entry.get("required", True) ) reasons: dict[str, int] = {} for entry in omitted_files: reason = entry.get("reason", "unknown") reasons[reason] = reasons.get(reason, 0) + 1 reason_summary = ", ".join( f"{reason.replace('_', ' ')}: {count}" for reason, count in sorted(reasons.items()) ) print( f" Safety notice: {len(omitted_files)} unavailable or unsafe " f"entries omitted ({required_omitted} required; {reason_summary})." ) total_omitted += len(omitted_files) print("\nChecking existing files...") to_download, up_to_date, mismatched = check_local( files, bios_path, jobs=args.jobs ) present = len(up_to_date) + len(mismatched) print( f" {present}/{len(files)} present " f"({len(up_to_date)} verified, {len(mismatched)} wrong hash)" ) # Mismatched files need re-download to_download.extend(mismatched) if args.check: if to_download: print(f"\n {len(to_download)} files need downloading.") else: print("\n All files up to date.") continue if to_download: dl_size = sum(f.get("size", 0) for f in to_download) print(f"\nDownloading {len(to_download)} files ({format_size(dl_size)})...") bios_path.mkdir(parents=True, exist_ok=True) failed = download_files( to_download, bios_path, jobs=args.jobs, verbose=args.verbose ) total_downloaded += len(to_download) - len(failed) total_errors += len(failed) else: print("\n All files up to date.") total_up_to_date += len(up_to_date) # Standalone copies if ( manifest.get("standalone_copies") and not args.check and args.standalone_copies ): print("\nStandalone emulators:") lb_root = launchbox_root(os_type) extra_dirs = launchbox_bios_dirs(lb_root) if lb_root else None copied, skipped = do_standalone_copies( manifest, bios_path, os_type, extra_dirs ) if copied or skipped: print(f" {copied} copied, {skipped} skipped (dir not found)") elif manifest.get("standalone_copies") and not args.check: print( "\nStandalone copies skipped " "(use --standalone-copies to opt in)." ) if not args.check and not args.list_targets: print( f"\nDone. {total_downloaded} downloaded, " f"{total_up_to_date} up to date, {total_errors} errors, " f"{total_omitted} safely omitted." ) if __name__ == "__main__": main()