Coverage for src/ph/retrobios.py: 97.3%

1"""Offline RetroBIOS requirements, verification, and explicit firmware retrieval.""" 

2 

3import hashlib 

4import json 

5import logging 

6import os 

7import shutil 

8import urllib.error 

9import urllib.request 

10from collections.abc import Callable, Iterable, Mapping, Sequence 

11from concurrent.futures import ThreadPoolExecutor, as_completed 

12from dataclasses import dataclass 

13from enum import StrEnum 

14from pathlib import Path, PurePosixPath 

15from time import time 

16from typing import Any 

17from urllib.parse import quote 

18 

19import yaml 

20 

21from ph.cache_policy import DEFAULT_CATALOGUE_TTL_DAYS, catalogue_ttl_seconds 

22from ph.models import Platform 

23from ph.organizer import ROM_AND_SHARED_BIOS, ROM_LOCAL_BIOS 

24 

25LOGGER = logging.getLogger(__name__) 

26 

27RETROBIOS_REPOSITORY = "https://github.com/Abdess/retrobios" 

28RETROBIOS_RAW_BASE = "https://raw.githubusercontent.com/Abdess/retrobios" 

29MAX_FIRMWARE_BYTES = 64 * 1024 * 1024 

30MAX_CATALOGUE_BYTES = 16 * 1024 * 1024 

31RETROBIOS_CACHE_FILENAME = "catalogue.json" 

32RETROBIOS_API_URL = "https://api.github.com/repos/Abdess/retrobios/commits/main" 

33 

34type BiosProgress = Callable[[str, int, int | None], None] 

35type CancellationCheck = Callable[[], bool] 

36 

37 

38class BiosError(RuntimeError): 

39 """A BIOS catalogue, verification, or installation operation failed.""" 

40 

41 

42class BiosDownloadCancelled(BiosError): 

43 """The user cancelled a RetroBIOS download.""" 

44 

45 

46class BiosState(StrEnum): 

47 """Verification state of a BIOS requirement on one ROM card.""" 

48 

49 VALID = "valid" 

50 MISSING = "missing" 

51 INVALID = "invalid" 

52 

53 

54@dataclass(frozen=True, slots=True) 

55class BiosRequirement: 

56 """One firmware file described by the pinned RetroBIOS catalogue.""" 

57 

58 name: str 

59 destination: str 

60 required: bool 

61 size: int | None 

62 sha256: str | None 

63 sha1: str | None 

64 md5: str | None 

65 source_path: str | None 

66 description: str | None = None 

67 note: str | None = None 

68 region: str | None = None 

69 hle_fallback: bool = False 

70 

71 

72@dataclass(frozen=True, slots=True) 

73class BiosSystem: 

74 """RetroBIOS metadata relevant to one emulated system.""" 

75 

76 system_id: str 

77 name: str 

78 core: str | None 

79 docs: str | None 

80 requirements: tuple[BiosRequirement, ...] 

81 

82 

83@dataclass(frozen=True, slots=True) 

84class BiosCheck: 

85 """Verification result for one BIOS requirement.""" 

86 

87 requirement: BiosRequirement 

88 state: BiosState 

89 paths: tuple[Path, ...] 

90 

91 

92# ROM-folder/platform identifiers differ from RetroBIOS system identifiers. 

93# Keep the mapping explicit so a similarly named but incompatible system is never 

94# selected by fuzzy matching. 

95RETROBIOS_SYSTEM_BY_PLATFORM: Mapping[str, str] = { 

96 "3do": "3do", 

97 "amiga": "commodore-amiga", 

98 "amiga-cd32": "commodore-amiga", 

99 "amstrad-cpc": "amstrad-cpc", 

100 "arcade": "arcade", 

101 "atari-800": "atari-400-800", 

102 "atari-5200": "atari-5200", 

103 "atari-7800": "atari-7800", 

104 "atari-lynx": "atari-lynx", 

105 "atari-st": "atari-st", 

106 "atomiswave": "sega-dreamcast-arcade", 

107 "colecovision": "coleco-colecovision", 

108 "commodore-128": "commodore-c128", 

109 "cps-1": "arcade", 

110 "cps-2": "arcade", 

111 "cps-3": "arcade", 

112 "dos": "dos", 

113 "doom": "doom", 

114 "dreamcast": "sega-dreamcast", 

115 "enterprise": "enterprise-64-128", 

116 "fairchild-channel-f": "fairchild-channel-f", 

117 "famicom-disk-system": "nintendo-fds", 

118 "game-boy": "nintendo-gb", 

119 "game-boy-advance": "nintendo-gba", 

120 "game-boy-color": "nintendo-gbc", 

121 "game-gear": "sega-game-gear", 

122 "genesis": "sega-mega-drive", 

123 "intellivision": "mattel-intellivision", 

124 "mame-2003": "arcade", 

125 "mame-2010": "arcade", 

126 "master-system": "sega-master-system", 

127 "msx": "microsoft-msx", 

128 "msx2": "microsoft-msx", 

129 "naomi": "sega-dreamcast-arcade", 

130 "neo-geo": "arcade", 

131 "neo-geo-cd": "snk-neogeo-cd", 

132 "nintendo-64dd": "nintendo-64dd", 

133 "nintendo-ds": "nintendo-ds", 

134 "nintendo": "nintendo-nes", 

135 "odyssey-2": "magnavox-odyssey2", 

136 "pc-98": "nec-pc-98", 

137 "pc-engine": "nec-pc-engine", 

138 "pc-engine-cd": "nec-pc-engine", 

139 "pc-fx": "nec-pc-fx", 

140 "playstation": "sony-playstation", 

141 "pokemon-mini": "nintendo-pokemon-mini", 

142 "ps-portable": "sony-psp", 

143 "psp-minis": "sony-psp", 

144 "satellaview": "nintendo-satellaview", 

145 "scummvm": "scummvm", 

146 "sega-cd": "sega-mega-cd", 

147 "saturn": "sega-saturn", 

148 "sharp-x1": "sharp-x1", 

149 "sharp-x68000": "sharp-x68000", 

150 "sufami-turbo": "nintendo-sufami-turbo", 

151 "super-cassette-vision": "epoch-scv", 

152 "super-game-boy": "nintendo-sgb", 

153 "super-nintendo": "nintendo-snes", 

154 "ti-99": "ti-83", 

155 "videopac": "philips-videopac", 

156 "videoton-tvc": "videoton-tvc", 

157 "vircon32": "vircon32", 

158 "wolfenstein": "wolfenstein-3d", 

159 "zx-spectrum": "sinclair-zx-spectrum", 

160} 

161 

162 

163class RetroBiosCatalog: 

164 """Read the release-pinned RetroBIOS manifest shipped with the application.""" 

165 

166 def __init__( 

167 self, 

168 revision: str, 

169 systems: Mapping[str, BiosSystem], 

170 generated_at: str | None = None, 

171 retroarch_version: str | None = None, 

172 fetched_at: float | None = None, 

173 ttl_seconds: int = catalogue_ttl_seconds(DEFAULT_CATALOGUE_TTL_DAYS), 

174 ) -> None: 

175 self.revision = revision 

176 self.systems = systems 

177 self.generated_at = generated_at 

178 self.retroarch_version = retroarch_version 

179 self.fetched_at = fetched_at 

180 self.ttl_seconds = ttl_seconds 

181 

182 @classmethod 

183 def from_json( 

184 cls, 

185 content: str, 

186 ttl_seconds: int = catalogue_ttl_seconds(DEFAULT_CATALOGUE_TTL_DAYS), 

187 ) -> RetroBiosCatalog: 

188 """Parse a generated manifest while rejecting unsafe or malformed entries.""" 

189 

190 try: 

191 payload = json.loads(content) 

192 except json.JSONDecodeError as error: 

193 raise BiosError(f"The bundled RetroBIOS catalogue is invalid: {error}") from error 

194 if not isinstance(payload, dict) or payload.get("schema_version") != 1: 

195 raise BiosError("The bundled RetroBIOS catalogue has an unsupported format.") 

196 revision = payload.get("source_commit") 

197 raw_systems = payload.get("systems") 

198 if ( 

199 not isinstance(revision, str) 

200 or len(revision) != 40 

201 or not isinstance(raw_systems, dict) 

202 ): 

203 raise BiosError("The bundled RetroBIOS catalogue is incomplete.") 

204 systems: dict[str, BiosSystem] = {} 

205 for system_id, raw_system in raw_systems.items(): 

206 if not isinstance(system_id, str) or not isinstance(raw_system, dict): 

207 continue 

208 raw_requirements = raw_system.get("files", ()) 

209 if not isinstance(raw_requirements, list): 

210 continue 

211 requirements = tuple( 

212 requirement 

213 for item in raw_requirements 

214 if (requirement := _parse_requirement(item)) is not None 

215 ) 

216 systems[system_id] = BiosSystem( 

217 system_id=system_id, 

218 name=_optional_string(raw_system.get("name")) or system_id, 

219 core=_optional_string(raw_system.get("core")), 

220 docs=_optional_string(raw_system.get("docs")), 

221 requirements=requirements, 

222 ) 

223 return cls( 

224 revision, 

225 systems, 

226 _optional_string(payload.get("source_generated_at")), 

227 _optional_string(payload.get("retroarch_version")), 

228 _optional_float(payload.get("fetched_at")), 

229 ttl_seconds, 

230 ) 

231 

232 def cache_age_seconds(self) -> float | None: 

233 """Return the local catalogue age when its download time is known.""" 

234 

235 return None if self.fetched_at is None else max(0.0, time() - self.fetched_at) 

236 

237 def cache_is_stale(self) -> bool: 

238 """Return whether this catalogue is missing a timestamp or exceeds its lifetime.""" 

239 

240 age = self.cache_age_seconds() 

241 return age is None or age > self.ttl_seconds 

242 

243 def system_for(self, platform: Platform) -> BiosSystem | None: 

244 """Return the exact RetroBIOS system mapped to a local platform.""" 

245 

246 system_id = RETROBIOS_SYSTEM_BY_PLATFORM.get(platform.slug) 

247 return self.systems.get(system_id) if system_id is not None else None 

248 

249 def requirements_for( 

250 self, 

251 platform: Platform, 

252 region: str | None = None, 

253 *, 

254 required_only: bool = False, 

255 ) -> tuple[BiosRequirement, ...]: 

256 """Return requirements, narrowing region-specific mandatory alternatives.""" 

257 

258 system = self.system_for(platform) 

259 if system is None: 

260 return () 

261 requirements = system.requirements 

262 if required_only: 

263 requirements = tuple(item for item in requirements if item.required) 

264 normalized_region = _normalize_region(region) 

265 if normalized_region is None: 

266 return requirements 

267 regional = tuple(item for item in requirements if item.required and item.region) 

268 matching = tuple( 

269 item for item in regional if _normalize_region(item.region) == normalized_region 

270 ) 

271 if not matching: 

272 return requirements 

273 nonregional = tuple(item for item in requirements if not item.required or not item.region) 

274 return nonregional + matching 

275 

276 def source_url(self, requirement: BiosRequirement) -> str | None: 

277 """Build the immutable raw GitHub URL for a downloadable requirement.""" 

278 

279 if requirement.source_path is None: 

280 return None 

281 return f"{RETROBIOS_RAW_BASE}/{self.revision}/{quote(requirement.source_path, safe='/')}" 

282 

283 

284class RetroBiosRepository: 

285 """Persist an atomic local catalogue downloaded only on first use or request.""" 

286 

287 def __init__( 

288 self, 

289 cache_directory: Path, 

290 timeout_seconds: float, 

291 ttl_seconds: int = catalogue_ttl_seconds(DEFAULT_CATALOGUE_TTL_DAYS), 

292 ) -> None: 

293 self.cache_path = cache_directory / "retrobios" / RETROBIOS_CACHE_FILENAME 

294 self.timeout_seconds = timeout_seconds 

295 self.ttl_seconds = ttl_seconds 

296 

297 def load(self) -> RetroBiosCatalog | None: 

298 """Load the existing cache without contacting GitHub.""" 

299 

300 try: 

301 content = self.cache_path.read_text(encoding="utf-8") 

302 except FileNotFoundError: 

303 return None 

304 except OSError as error: 

305 raise BiosError(f"Could not read the RetroBIOS catalogue: {error}") from error 

306 return RetroBiosCatalog.from_json(content, self.ttl_seconds) 

307 

308 def ensure( 

309 self, 

310 progress: BiosProgress | None = None, 

311 cancelled: CancellationCheck | None = None, 

312 ) -> RetroBiosCatalog: 

313 """Use the cache when present, downloading it once when missing.""" 

314 

315 cached = self.load() 

316 return cached if cached is not None else self.update(progress, cancelled) 

317 

318 def update( 

319 self, 

320 progress: BiosProgress | None = None, 

321 cancelled: CancellationCheck | None = None, 

322 ) -> RetroBiosCatalog: 

323 """Download, validate, and atomically replace the cached catalogue.""" 

324 

325 if progress is not None: 

326 progress("Finding the latest RetroBIOS revision", 0, 1) 

327 api_payload = _download_json( 

328 RETROBIOS_API_URL, 

329 self.timeout_seconds, 

330 cancelled, 

331 max_bytes=1024 * 1024, 

332 ) 

333 revision = api_payload.get("sha") if isinstance(api_payload, dict) else None 

334 if ( 

335 not isinstance(revision, str) 

336 or len(revision) != 40 

337 or any(character not in "0123456789abcdef" for character in revision.casefold()) 

338 ): 

339 raise BiosError("GitHub returned an invalid RetroBIOS revision.") 

340 

341 platform_url = f"{RETROBIOS_RAW_BASE}/{revision}/platforms/retroarch.yml" 

342 platform = _download_yaml(platform_url, self.timeout_seconds, cancelled) 

343 systems = platform.get("systems") 

344 if not isinstance(systems, dict): 

345 raise BiosError("RetroBIOS returned an invalid RetroArch platform catalogue.") 

346 cores = sorted( 

347 { 

348 core 

349 for system in systems.values() 

350 if isinstance(system, dict) 

351 and isinstance((core := system.get("core")), str) 

352 and core 

353 } 

354 ) 

355 total_steps = len(cores) + 2 

356 if progress is not None: 

357 progress("Downloading RetroBIOS core profiles", 1, total_steps) 

358 profiles: dict[str, dict[str, Any]] = {} 

359 

360 def download_profile(core: str) -> tuple[str, dict[str, Any]]: 

361 profile_url = f"{RETROBIOS_RAW_BASE}/{revision}/emulators/{quote(core)}.yml" 

362 return core, _download_yaml(profile_url, self.timeout_seconds, cancelled) 

363 

364 with ThreadPoolExecutor(max_workers=8, thread_name_prefix="retrobios-metadata") as executor: 

365 futures = {executor.submit(download_profile, core): core for core in cores} 

366 for completed, future in enumerate(as_completed(futures), start=2): 

367 if cancelled is not None and cancelled(): 367 ↛ 368line 367 didn't jump to line 368 because the condition on line 367 was never true

368 for pending in futures: 

369 pending.cancel() 

370 raise BiosDownloadCancelled("RetroBIOS catalogue update cancelled.") 

371 core, profile = future.result() 

372 profiles[core] = profile 

373 if progress is not None: 373 ↛ 366line 373 didn't jump to line 366 because the condition on line 373 was always true

374 progress("Downloading RetroBIOS core profiles", completed, total_steps) 

375 

376 database_url = f"{RETROBIOS_RAW_BASE}/{revision}/database.json" 

377 database = _download_json( 

378 database_url, 

379 self.timeout_seconds, 

380 cancelled, 

381 max_bytes=MAX_CATALOGUE_BYTES, 

382 ) 

383 manifest = _build_manifest(revision, platform, profiles, database) 

384 serialized = json.dumps(manifest, ensure_ascii=False, indent=2) + "\n" 

385 catalogue = RetroBiosCatalog.from_json(serialized, self.ttl_seconds) 

386 temporary = self.cache_path.with_name(self.cache_path.name + ".tmp") 

387 try: 

388 self.cache_path.parent.mkdir(parents=True, exist_ok=True) 

389 temporary.write_text(serialized, encoding="utf-8") 

390 os.replace(temporary, self.cache_path) 

391 except OSError as error: 

392 temporary.unlink(missing_ok=True) 

393 raise BiosError(f"Could not save the RetroBIOS catalogue: {error}") from error 

394 LOGGER.info( 

395 "RetroBIOS catalogue cached revision=%s systems=%d path=%s", 

396 revision, 

397 len(catalogue.systems), 

398 self.cache_path, 

399 ) 

400 return catalogue 

401 

402 

403def _download_yaml( 

404 url: str, 

405 timeout_seconds: float, 

406 cancelled: CancellationCheck | None, 

407) -> dict[str, Any]: 

408 content = _download_bytes( 

409 url, 

410 timeout_seconds, 

411 cancelled, 

412 max_bytes=2 * 1024 * 1024, 

413 ) 

414 try: 

415 payload = yaml.safe_load(content) 

416 except yaml.YAMLError as error: 

417 raise BiosError(f"RetroBIOS returned invalid YAML metadata: {error}") from error 

418 if not isinstance(payload, dict): 418 ↛ 420line 418 didn't jump to line 420 because the condition on line 418 was always true

419 raise BiosError("RetroBIOS returned incomplete YAML metadata.") 

420 return payload 

421 

422 

423def _download_json( 

424 url: str, 

425 timeout_seconds: float, 

426 cancelled: CancellationCheck | None, 

427 *, 

428 max_bytes: int, 

429) -> dict[str, Any]: 

430 content = _download_bytes(url, timeout_seconds, cancelled, max_bytes=max_bytes) 

431 try: 

432 payload = json.loads(content) 

433 except (UnicodeDecodeError, json.JSONDecodeError) as error: 

434 raise BiosError(f"RetroBIOS returned invalid JSON metadata: {error}") from error 

435 if not isinstance(payload, dict): 435 ↛ 437line 435 didn't jump to line 437 because the condition on line 435 was always true

436 raise BiosError("RetroBIOS returned incomplete JSON metadata.") 

437 return payload 

438 

439 

440def _download_bytes( 

441 url: str, 

442 timeout_seconds: float, 

443 cancelled: CancellationCheck | None, 

444 *, 

445 max_bytes: int, 

446) -> bytes: 

447 request = urllib.request.Request( 

448 url, 

449 headers={ 

450 "Accept": "application/vnd.github+json", 

451 "User-Agent": "pocket-harbor/retrobios", 

452 }, 

453 ) 

454 content = bytearray() 

455 try: 

456 with urllib.request.urlopen(request, timeout=timeout_seconds) as response: 

457 while chunk := response.read(1024 * 128): 

458 if cancelled is not None and cancelled(): 

459 raise BiosDownloadCancelled("RetroBIOS catalogue update cancelled.") 

460 content.extend(chunk) 

461 if len(content) > max_bytes: 

462 raise BiosError("RetroBIOS metadata exceeded the safety limit.") 

463 except BiosError: 

464 raise 

465 except (OSError, urllib.error.URLError) as error: 

466 raise BiosError(f"Could not reach RetroBIOS: {error}") from error 

467 return bytes(content) 

468 

469 

470def _build_manifest( 

471 revision: str, 

472 platform: dict[str, Any], 

473 profiles: Mapping[str, dict[str, Any]], 

474 database: dict[str, Any], 

475) -> dict[str, Any]: 

476 systems = platform.get("systems") 

477 if not isinstance(systems, dict): 

478 raise BiosError("RetroBIOS platform metadata does not contain systems.") 

479 generated_systems: dict[str, Any] = {} 

480 for system_id, system_data in systems.items(): 

481 if not isinstance(system_id, str) or not isinstance(system_data, dict): 

482 continue 

483 core = _optional_string(system_data.get("core")) 

484 profile = profiles.get(core) if core is not None else None 

485 profile_files = _profile_files(profile) 

486 generated_files: list[dict[str, Any]] = [] 

487 raw_files = system_data.get("files", ()) 

488 for file_data in raw_files if isinstance(raw_files, list) else (): 

489 if not isinstance(file_data, dict) or not isinstance(file_data.get("name"), str): 

490 continue 

491 profile_file = profile_files.get(str(file_data["name"]).casefold(), {}) 

492 stored_file = _database_file(database, file_data) or {} 

493 generated_files.append( 

494 { 

495 "name": file_data["name"], 

496 "destination": file_data.get("destination", file_data["name"]), 

497 "required": ( 

498 profile_file.get("required", False) 

499 if profile is not None 

500 else file_data.get("required", False) 

501 ), 

502 "size": file_data.get("size", stored_file.get("size")), 

503 "sha256": stored_file.get("sha256"), 

504 "sha1": file_data.get("sha1", stored_file.get("sha1")), 

505 "md5": file_data.get("md5", stored_file.get("md5")), 

506 "source_path": stored_file.get("path"), 

507 "description": profile_file.get("description"), 

508 "note": profile_file.get("note"), 

509 "region": profile_file.get("region"), 

510 "hle_fallback": profile_file.get("hle_fallback", False), 

511 } 

512 ) 

513 generated_systems[system_id] = { 

514 "name": system_data.get("native_id", system_id), 

515 "core": core, 

516 "docs": system_data.get("docs"), 

517 "files": generated_files, 

518 } 

519 return { 

520 "schema_version": 1, 

521 "fetched_at": time(), 

522 "source_repository": RETROBIOS_REPOSITORY, 

523 "source_commit": revision, 

524 "source_generated_at": database.get("generated_at"), 

525 "retroarch_version": platform.get("version"), 

526 "systems": generated_systems, 

527 } 

528 

529 

530def _profile_files(profile: dict[str, Any] | None) -> dict[str, dict[str, Any]]: 

531 raw_files = profile.get("files", ()) if profile is not None else () 

532 if not isinstance(raw_files, list): 

533 return {} 

534 return { 

535 str(item["name"]).casefold(): item 

536 for item in raw_files 

537 if isinstance(item, dict) and isinstance(item.get("name"), str) 

538 } 

539 

540 

541def _database_file( 

542 database: dict[str, Any], 

543 file_data: dict[str, Any], 

544) -> dict[str, Any] | None: 

545 raw_files = database.get("files", {}) 

546 if not isinstance(raw_files, dict): 

547 return None 

548 sha1 = _optional_string(file_data.get("sha1")) 

549 if sha1 is not None: 

550 candidate = raw_files.get(sha1.casefold()) 

551 if isinstance(candidate, dict): 551 ↛ 553line 551 didn't jump to line 553 because the condition on line 551 was always true

552 return candidate 

553 indexes = database.get("indexes", {}) 

554 by_name = indexes.get("by_name", {}) if isinstance(indexes, dict) else {} 

555 name = _optional_string(file_data.get("name")) 

556 keys = by_name.get(name.casefold(), ()) if isinstance(by_name, dict) and name else () 

557 for key in keys if isinstance(keys, list) else (): 

558 candidate = raw_files.get(key) 

559 if not isinstance(candidate, dict): 

560 continue 

561 expected_size = file_data.get("size") 

562 if expected_size is None or candidate.get("size") == expected_size: 

563 return candidate 

564 return None 

565 

566 

567def audit_bios( 

568 requirements: Iterable[BiosRequirement], 

569 platform: Platform, 

570 roms_directory: Path, 

571 bios_directory: str = "bios", 

572) -> tuple[BiosCheck, ...]: 

573 """Verify BIOS files on the selected ROM card using pinned checksums.""" 

574 

575 checks: list[BiosCheck] = [] 

576 for requirement in requirements: 

577 paths = bios_destinations(requirement, platform, roms_directory, bios_directory) 

578 existing = tuple(path for path in paths if path.is_file()) 

579 validity = tuple(verify_bios_file(path, requirement) for path in existing) 

580 if len(existing) == len(paths) and all(validity): 

581 state = BiosState.VALID 

582 elif any(not valid for valid in validity): 

583 state = BiosState.INVALID 

584 else: 

585 state = BiosState.MISSING 

586 checks.append(BiosCheck(requirement, state, paths)) 

587 return tuple(checks) 

588 

589 

590def audit_bios_roots( 

591 requirements: Iterable[BiosRequirement], 

592 platform: Platform, 

593 roms_directories: Sequence[Path], 

594 preferred_directory: Path, 

595 bios_directory: str = "bios", 

596) -> tuple[BiosCheck, ...]: 

597 """Treat a valid BIOS on either active memory card as already available.""" 

598 

599 roots = tuple(dict.fromkeys((preferred_directory, *roms_directories))) 

600 preferred_checks = audit_bios(requirements, platform, preferred_directory, bios_directory) 

601 checks_by_root = tuple( 

602 audit_bios(requirements, platform, root, bios_directory) for root in roots 

603 ) 

604 combined: list[BiosCheck] = [] 

605 for index, preferred in enumerate(preferred_checks): 

606 valid = next( 

607 ( 

608 checks[index] 

609 for checks in checks_by_root 

610 if index < len(checks) and checks[index].state is BiosState.VALID 

611 ), 

612 None, 

613 ) 

614 combined.append(valid or preferred) 

615 return tuple(combined) 

616 

617 

618def bios_destinations( 

619 requirement: BiosRequirement, 

620 platform: Platform, 

621 roms_directory: Path, 

622 bios_directory: str = "bios", 

623) -> tuple[Path, ...]: 

624 """Resolve shared and exceptional ROM-local BIOS destinations.""" 

625 

626 relative = PurePosixPath(requirement.destination.replace("\\", "/")) 

627 if ( 

628 relative.is_absolute() 

629 or not relative.parts 

630 or any(part in {"", ".", ".."} for part in relative.parts) 

631 ): 

632 raise BiosError(f"RetroBIOS contains an unsafe destination: {requirement.destination}") 

633 folder = platform.rom_folder or "" 

634 filename = relative.name.casefold() 

635 shared = roms_directory / bios_directory / Path(*relative.parts) 

636 local = roms_directory / folder / Path(*relative.parts) 

637 if filename in ROM_LOCAL_BIOS.get(platform.slug, frozenset()): 

638 return (local,) 

639 if filename in ROM_AND_SHARED_BIOS.get(platform.slug, frozenset()): 

640 return (shared, local) 

641 return (shared,) 

642 

643 

644def verify_bios_file(path: Path, requirement: BiosRequirement) -> bool: 

645 """Validate size and the strongest checksum published by RetroBIOS.""" 

646 

647 try: 

648 if requirement.size is not None and path.stat().st_size != requirement.size: 

649 return False 

650 algorithm, expected = _strongest_checksum(requirement) 

651 if algorithm is None or expected is None: 

652 return requirement.size is not None 

653 digest = hashlib.new(algorithm) 

654 with path.open("rb") as stream: 

655 while chunk := stream.read(1024 * 256): 

656 digest.update(chunk) 

657 return digest.hexdigest().casefold() == expected.casefold() 

658 except OSError: 

659 return False 

660 

661 

662def install_bios( 

663 catalog: RetroBiosCatalog, 

664 requirement: BiosRequirement, 

665 platform: Platform, 

666 roms_directory: Path, 

667 timeout_seconds: float, 

668 progress: BiosProgress | None = None, 

669 cancelled: CancellationCheck | None = None, 

670 bios_directory: str = "bios", 

671) -> tuple[Path, ...]: 

672 """Download one explicitly approved BIOS and atomically install verified copies.""" 

673 

674 url = catalog.source_url(requirement) 

675 if url is None: 

676 raise BiosError(f"RetroBIOS does not provide a downloadable copy of {requirement.name}.") 

677 if requirement.size is not None and requirement.size > MAX_FIRMWARE_BYTES: 

678 raise BiosError(f"RetroBIOS firmware is too large to install safely: {requirement.name}") 

679 destinations = bios_destinations(requirement, platform, roms_directory, bios_directory) 

680 valid = tuple(path for path in destinations if verify_bios_file(path, requirement)) 

681 if len(valid) == len(destinations): 

682 return valid 

683 

684 primary = destinations[0] 

685 partial = primary.with_name(primary.name + ".part") 

686 try: 

687 primary.parent.mkdir(parents=True, exist_ok=True) 

688 request = urllib.request.Request( 

689 url, 

690 headers={"User-Agent": "pocket-harbor/retrobios"}, 

691 ) 

692 with ( 

693 urllib.request.urlopen(request, timeout=timeout_seconds) as response, 

694 partial.open("wb") as output, 

695 ): 

696 header = response.headers.get("Content-Length") 

697 total = int(header) if header and header.isdigit() else requirement.size 

698 downloaded = 0 

699 while chunk := response.read(1024 * 128): 

700 if cancelled is not None and cancelled(): 

701 raise BiosDownloadCancelled("BIOS download cancelled.") 

702 downloaded += len(chunk) 

703 if downloaded > MAX_FIRMWARE_BYTES: 703 ↛ 704line 703 didn't jump to line 704 because the condition on line 703 was never true

704 raise BiosError("RetroBIOS download exceeded the firmware safety limit.") 

705 output.write(chunk) 

706 if progress is not None: 

707 progress(requirement.name, downloaded, total) 

708 if cancelled is not None and cancelled(): 

709 raise BiosDownloadCancelled("BIOS download cancelled.") 

710 if not verify_bios_file(partial, requirement): 

711 raise BiosError(f"Downloaded {requirement.name} does not match the RetroBIOS checksum.") 

712 os.replace(partial, primary) 

713 installed = [primary] 

714 for destination in destinations[1:]: 

715 if verify_bios_file(destination, requirement): 

716 installed.append(destination) 

717 continue 

718 destination.parent.mkdir(parents=True, exist_ok=True) 

719 copy_partial = destination.with_name(destination.name + ".part") 

720 try: 

721 shutil.copyfile(primary, copy_partial) 

722 os.replace(copy_partial, destination) 

723 except OSError: 

724 copy_partial.unlink(missing_ok=True) 

725 raise 

726 installed.append(destination) 

727 LOGGER.info( 

728 "Installed RetroBIOS firmware platform=%s file=%s destinations=%s", 

729 platform.alias, 

730 requirement.name, 

731 installed, 

732 ) 

733 return tuple(installed) 

734 except BiosDownloadCancelled: 

735 partial.unlink(missing_ok=True) 

736 LOGGER.info("RetroBIOS download cancelled file=%s", requirement.name) 

737 raise 

738 except BiosError: 

739 partial.unlink(missing_ok=True) 

740 raise 

741 except (OSError, ValueError, urllib.error.URLError) as error: