"""Updater-safe runtime view with one verified normalized scene cache.""" from __future__ import annotations from dataclasses import asdict, dataclass from datetime import datetime import hashlib import json import os from pathlib import Path import shutil from typing import Mapping import uuid from ptdiag.assbin import ( SceneInspection, inspect_wrapped_assbin, normalize_wrapped_assbin_textures, ) from .paths import AppPaths WHISPERING_VALE_RELATIVE = Path("game/maps/WhisperingVale.rovug") KNOWN_SOURCE_SHA256 = ( "1c9655945882b6b2189bbc6961018fa84295f895330987bad34e8428e9b5ce90" ) TEXTURE_INDICES = (37, 38) KNOWN_NORMALIZED_SCANLINES_SHA256 = { 37: "8fcc54e8d30fbfb725bb4c917a123e4f28a228397b4939d2fcfc7e8f3edc84f5", 38: "a8301700ee76b0cbf9d974c1395093623390244f2f92658b352a1c3e19e20f60", } COMPATIBILITY_BINARY_HASHES = { "Game.exe": "3cc0980aac0f65b379ff340ec61750498fb9f620fb6c88aa66281e8ab27023a5", "game.dll": "7ea85b19f452962a39c91f967cad22495f066b2d52e8ac37f4e234463d5c4f09", } CURRENT_V21_BINARY_HASHES = { "Game.exe": "2bc3edd26c5f6d7ae9bbdc23967c5aefd374bdf934fcdfa2ed42c6442e550837", "game.dll": "4b4e5c28e2bfdaa8d36c8341fad27becfd03faf63aa8ea874f5d43c5b1961868", } PATCHED_V21_GAME_DLL_SHA256 = ( "0be11b06ccf9be747e32b195c8d5980062f446d7d3a0a662ebfe0e602a01c538" ) RETIRED_POST_AUTH_GAME_DLL_SHA256 = ( "93cd27284eb294753da1111c99bfc9689f82b6d4a59302aee9fdbd8cb52452dd" ) RUNTIME_VIEW_CONFIGURATION_VERSION = 2 WINDOWED_GAME_CONFIG_VALUES = ( (b"windowed", b"Windowed", b"1"), (b"startmaximized", b"StartMaximized", b"False"), ) class OverlayError(RuntimeError): """Raised when a safe runtime view cannot be built.""" @dataclass(frozen=True) class RuntimeFileIdentity: """Cheap change detector for one previously hash-verified runtime input.""" name: str device: int inode: int mode: int size: int mtime_ns: int ctime_ns: int @dataclass(frozen=True) class OverlayResult: source_game: str runtime_view: str source_scene_sha256: str runtime_scene_sha256: str scene_action: str binary_action: str game_exe_sha256: str game_dll_sha256: str input_identities: tuple[RuntimeFileIdentity, ...] runtime_scene_identity: RuntimeFileIdentity created_at: str links_created: int @dataclass(frozen=True) class LegacyOverlayResult: """Version-one runtime manifest retained only for a one-time fast migration.""" source_game: str runtime_view: str source_scene_sha256: str runtime_scene_sha256: str scene_action: str binary_action: str game_exe_sha256: str game_dll_sha256: str created_at: str links_created: int @dataclass(frozen=True) class CompatibilityCacheResult: source: str destination: str game_exe_sha256: str game_dll_sha256: str created_at: str @dataclass(frozen=True) class RuntimeCompatibility: """Read-only compatibility result for the managed game currently on disk.""" compatible: bool source_game: str source_scene_sha256: str | None scene_action: str | None binary_action: str | None reason: str | None def inspect_runtime_compatibility( paths: AppPaths, source_game: Path, ) -> RuntimeCompatibility: """Check whether ``prepare_runtime_view`` can recognize the installed release. This preflight is intentionally read-only: it never normalizes a scene, builds the verified runtime DLL, prepares save data, or replaces the runtime view. Preparation reuses a prior authoritative result only while every input identity and runtime-view link remains unchanged; otherwise it repeats the full checks before launch. """ requested_source = source_game.expanduser() source_scene_sha256 = None scene_action = None binary_action = None try: source = requested_source.resolve(strict=True) _validate_game_tree(source) cached = _cached_runtime_compatibility(paths, source) if cached is not None: return cached source_scene = source / WHISPERING_VALE_RELATIVE inspection = inspect_wrapped_assbin(source_scene) source_scene_sha256 = inspection.file_sha256 scene_action = _scene_action(inspection) binary_action = _inspect_runtime_binary_action(paths, source) except (OSError, ValueError, OverlayError) as exc: reason = str(exc).strip() or type(exc).__name__ return RuntimeCompatibility( compatible=False, source_game=str(requested_source.resolve(strict=False)), source_scene_sha256=source_scene_sha256, scene_action=scene_action, binary_action=binary_action, reason=reason, ) return RuntimeCompatibility( compatible=True, source_game=str(source), source_scene_sha256=source_scene_sha256, scene_action=scene_action, binary_action=binary_action, reason=None, ) def prepare_runtime_view( paths: AppPaths, source_game: Path, ) -> OverlayResult: """Build a symlink view and keep the launcher-managed source read-only.""" source = source_game.expanduser().resolve(strict=True) _validate_game_tree(source) _validate_runtime_target(paths) game_save = _prepare_persistent_save(paths, source) game_config = _prepare_persistent_game_config(paths, source) reused = _reuse_runtime_view(paths, source, game_save, game_config) if reused is not None: return reused input_identities = _runtime_input_identities(paths, source) source_scene = source / WHISPERING_VALE_RELATIVE inspection = inspect_wrapped_assbin(source_scene) action = _scene_action(inspection) ( binary_action, game_exe_source, game_dll_source, source_binary_hashes, ) = _runtime_game_binary_profile( paths, source, ) legacy_reuse = _reuse_legacy_runtime_view( paths, source, inspection, action, game_save, game_config, binary_action, game_exe_source, game_dll_source, source_binary_hashes, input_identities, ) if legacy_reuse is not None: return legacy_reuse build = paths.data / f".{paths.runtime_view.name}.build-{uuid.uuid4().hex}" backup = paths.data / f".{paths.runtime_view.name}.old-{uuid.uuid4().hex}" links_created = 0 try: build.mkdir(parents=True, exist_ok=False) top_level_exceptions = {"game", "game.ini", "save"} for name, binary_source in ( ("Game.exe", game_exe_source), ("game.dll", game_dll_source), ): if binary_source != source / name: top_level_exceptions.add(name) links_created += _link_level( source, build, except_names=top_level_exceptions, ) (build / "save").symlink_to(game_save, target_is_directory=True) links_created += 1 (build / "game.ini").symlink_to(game_config) links_created += 1 if game_exe_source != source / "Game.exe": (build / "Game.exe").symlink_to(game_exe_source) links_created += 1 if game_dll_source != source / "game.dll": (build / "game.dll").symlink_to(game_dll_source) links_created += 1 game_source = source / "game" game_target = build / "game" game_target.mkdir() links_created += _link_level(game_source, game_target, except_names={"maps"}) maps_source = game_source / "maps" maps_target = game_target / "maps" maps_target.mkdir() links_created += _link_level( maps_source, maps_target, except_names={"WhisperingVale.rovug"}, ) runtime_scene = maps_target / "WhisperingVale.rovug" if action == "normalize-known-profile": normalization = normalize_wrapped_assbin_textures( source_scene, runtime_scene, TEXTURE_INDICES, ) validate_scene_normalization(normalization) else: shutil.copy2(source_scene, runtime_scene) if _runtime_input_identities(paths, source) != input_identities: raise OverlayError( "game files changed while the managed runtime view was being prepared" ) if paths.runtime_view.exists() or paths.runtime_view.is_symlink(): paths.runtime_view.replace(backup) build.replace(paths.runtime_view) if backup.exists() or backup.is_symlink(): _remove_runtime_tree(backup) except BaseException: _remove_runtime_tree(build) if backup.exists() and not paths.runtime_view.exists(): backup.replace(paths.runtime_view) raise result = OverlayResult( source_game=str(source), runtime_view=str(paths.runtime_view), source_scene_sha256=inspection.file_sha256, runtime_scene_sha256=_sha256(paths.runtime_view / WHISPERING_VALE_RELATIVE), scene_action=action, binary_action=binary_action, game_exe_sha256=_sha256(paths.runtime_view / "Game.exe"), game_dll_sha256=_sha256(paths.runtime_view / "game.dll"), input_identities=input_identities, runtime_scene_identity=_runtime_file_identity( "runtime-scene", paths.runtime_view / WHISPERING_VALE_RELATIVE, ), created_at=datetime.now().astimezone().isoformat(), links_created=links_created, ) _write_json( paths.runtime_manifest, { "configuration_version": RUNTIME_VIEW_CONFIGURATION_VERSION, **asdict(result), }, ) return result def cache_compatibility_binaries( paths: AppPaths, source_game: Path, ) -> CompatibilityCacheResult: source = source_game.expanduser().resolve(strict=True) hashes = _binary_hashes(source) if hashes != COMPATIBILITY_BINARY_HASHES: raise OverlayError( "compatibility source does not match the verified Game.exe/game.dll profile: " f"{source}" ) target = paths.compatibility_binaries if target.exists() or target.is_symlink(): if target.is_dir() and _binary_hashes(target) == COMPATIBILITY_BINARY_HASHES: return _compatibility_result(paths, source) raise OverlayError(f"compatibility cache has an unknown profile: {target}") paths.data.mkdir(parents=True, exist_ok=True) build = paths.data / f".{target.name}.build-{uuid.uuid4().hex}" try: build.mkdir(parents=False, exist_ok=False) for name in COMPATIBILITY_BINARY_HASHES: shutil.copy2(source / name, build / name) if _binary_hashes(build) != COMPATIBILITY_BINARY_HASHES: raise OverlayError("copied compatibility binaries failed hash verification") build.replace(target) except BaseException: _remove_runtime_tree(build) raise result = _compatibility_result(paths, source) _write_json(paths.compatibility_manifest, asdict(result)) return result def select_binary_action( cache_hashes: dict[str, str] | None, source_hashes: dict[str, str], ) -> str: if cache_hashes is not None: if cache_hashes != COMPATIBILITY_BINARY_HASHES: raise OverlayError("compatibility cache has an unknown binary profile") if source_hashes == COMPATIBILITY_BINARY_HASHES: return "verified-source-pair" if source_hashes == CURRENT_V21_BINARY_HASHES: return "verified-v21-publisher-pair" game_exe_sha256 = source_hashes.get("Game.exe") game_dll_sha256 = source_hashes.get("game.dll") if game_dll_sha256 in { PATCHED_V21_GAME_DLL_SHA256, RETIRED_POST_AUTH_GAME_DLL_SHA256, }: raise OverlayError( "a retired experimental game.dll is installed; run update --repair " "before launching" ) known_game_exe_hashes = { CURRENT_V21_BINARY_HASHES["Game.exe"], COMPATIBILITY_BINARY_HASHES["Game.exe"], } known_game_dll_hashes = { CURRENT_V21_BINARY_HASHES["game.dll"], COMPATIBILITY_BINARY_HASHES["game.dll"], } if ( game_exe_sha256 in known_game_exe_hashes or game_dll_sha256 in known_game_dll_hashes ): raise OverlayError( "Game.exe and game.dll are from different verified releases; run " "update --repair before launching" ) return "upstream-pair" def _scene_action(inspection: SceneInspection) -> str: selected = [] for index in TEXTURE_INDICES: if index >= len(inspection.embedded_textures): raise OverlayError(f"scene no longer has embedded texture {index}") selected.append(inspection.embedded_textures[index]) depths = [texture.png.bit_depth if texture.png is not None else None for texture in selected] colors = [texture.png.color_type if texture.png is not None else None for texture in selected] return select_scene_action(inspection.file_sha256, depths, colors) def select_scene_action( file_sha256: str, depths: list[int | None], colors: list[int | None], ) -> str: """Choose a repair only from the exact source hash and decoded PNG profile.""" if file_sha256 == KNOWN_SOURCE_SHA256 and depths == [16, 16] and colors == [2, 2]: return "normalize-known-profile" if depths == [8, 8] and colors == [2, 2]: return "upstream-compatible-copy" raise OverlayError( "WhisperingVale.rovug is not a supported release profile; refusing to " f"modify unknown textures (sha256={file_sha256}, " f"depths={depths}, colors={colors})" ) def validate_scene_normalization(report: Mapping[str, object]) -> None: """Verify the known repair by decoded pixels, not zlib's encoded bytes. zlib is allowed to emit different valid DEFLATE streams across versions and compatible implementations. The old whole-file hash therefore rejected a semantically identical scene on some distributions. The source file remains pinned exactly, while these hashes pin the decoded scanlines produced for the two repaired textures. """ if report.get("source_sha256") != KNOWN_SOURCE_SHA256: raise OverlayError("normalized scene source did not match the verified release profile") verified = report.get("verified") if not isinstance(verified, bool) or not verified: raise OverlayError("normalized scene did not pass structural verification") if report.get("texture_indices") != list(TEXTURE_INDICES): raise OverlayError("normalized scene changed an unexpected texture set") changes = report.get("changes") if not isinstance(changes, list): raise OverlayError("normalized scene report did not contain texture changes") change_hashes: dict[int, str] = {} for change in changes: if not isinstance(change, dict): raise OverlayError("normalized scene report contained an invalid texture change") index = change.get("index") scanlines_sha256 = change.get("new_scanlines_sha256") if isinstance(index, bool) or not isinstance(index, int): raise OverlayError("normalized scene report contained an invalid texture index") if not isinstance(scanlines_sha256, str): raise OverlayError("normalized scene report omitted a decoded texture hash") if index in change_hashes: raise OverlayError("normalized scene report repeated a texture change") change_hashes[index] = scanlines_sha256 if change_hashes != KNOWN_NORMALIZED_SCANLINES_SHA256: raise OverlayError( "normalized scene pixels did not match the verified release profile" ) def _validate_game_tree(source: Path) -> None: for relative in ( Path("Game.exe"), Path("game.dll"), Path("game.ini"), WHISPERING_VALE_RELATIVE, ): if not (source / relative).is_file(): raise OverlayError(f"game installation is incomplete: missing {relative}") def _validate_runtime_target(paths: AppPaths) -> None: data = paths.data.resolve() target = paths.runtime_view if target.parent.resolve() != data: raise OverlayError(f"runtime view is outside the application data root: {target}") if target == data: raise OverlayError("runtime view cannot be the application data root") def _link_level(source: Path, target: Path, *, except_names: set[str]) -> int: count = 0 for child in source.iterdir(): if child.name in except_names: continue (target / child.name).symlink_to(child, target_is_directory=child.is_dir()) count += 1 return count def _prepare_persistent_save(paths: AppPaths, source: Path) -> Path: """Migrate game-created settings and logs before replacing the runtime view.""" target = paths.game_save if target.is_symlink(): raise OverlayError(f"persistent game save must not be a symbolic link: {target}") if target.exists(): if not target.is_dir(): raise OverlayError(f"persistent game save is not a directory: {target}") _reject_save_symlinks(target) return target candidate = None for path in (paths.runtime_view / "save", source / "save"): if path.is_dir(): candidate = path.resolve(strict=True) break if candidate is not None: selected = candidate allowed_roots = (paths.data.resolve(strict=False), source.resolve(strict=True)) if not any( selected == root or selected.is_relative_to(root) for root in allowed_roots ): raise OverlayError(f"refusing to migrate game save outside managed roots: {selected}") _reject_save_symlinks(selected) target.parent.mkdir(parents=True, exist_ok=True) staging = target.parent / f".{target.name}.build-{uuid.uuid4().hex}" try: if candidate is None: staging.mkdir(parents=False, exist_ok=False) else: shutil.copytree(candidate, staging, symlinks=True) staging.replace(target) except BaseException: _remove_runtime_tree(staging) raise return target def _prepare_persistent_game_config(paths: AppPaths, source: Path) -> Path: """Keep game settings writable while forcing a true windowed startup.""" target = paths.game_config if target.is_symlink(): raise OverlayError(f"persistent game config must not be a symbolic link: {target}") if target.exists() and not target.is_file(): raise OverlayError(f"persistent game config is not a file: {target}") source_config = target if target.is_file() else source / "game.ini" try: content = source_config.read_bytes() mode = source_config.stat().st_mode & 0o777 except OSError as exc: raise OverlayError(f"could not read game config {source_config}: {exc}") from exc windowed_content = _force_windowed_game_config(content) if source_config == target and windowed_content == content: return target target.parent.mkdir(parents=True, exist_ok=True) temporary = target.with_name(f".{target.name}.{uuid.uuid4().hex}.tmp") try: with temporary.open("xb") as handle: handle.write(windowed_content) handle.flush() os.fsync(handle.fileno()) temporary.chmod(mode) temporary.replace(target) except BaseException: temporary.unlink(missing_ok=True) raise return target def _force_windowed_game_config(content: bytes) -> bytes: """Set only the two Screen keys that can trigger a desktop-sized backbuffer.""" if b"\x00" in content: raise OverlayError("game.ini uses an unsupported encoding") newline = b"\r\n" if b"\r\n" in content else b"\n" desired = { key: (display_name, value) for key, display_name, value in WINDOWED_GAME_CONFIG_VALUES } result: list[bytes] = [] in_screen = False found_screen = False found_settings: set[bytes] = set() def append_missing_settings() -> None: if result and not result[-1].endswith((b"\r", b"\n")): result[-1] += newline for key, display_name, value in WINDOWED_GAME_CONFIG_VALUES: if key in found_settings: continue result.append(display_name + b"=" + value + newline) found_settings.add(key) for line in content.splitlines(keepends=True): if line.endswith(b"\r\n"): body, line_ending = line[:-2], b"\r\n" elif line.endswith((b"\r", b"\n")): body, line_ending = line[:-1], line[-1:] else: body, line_ending = line, b"" stripped = body.strip() if stripped.startswith(b"[") and b"]" in stripped: if in_screen: append_missing_settings() section_end = stripped.find(b"]") in_screen = stripped[1:section_end].strip().lower() == b"screen" found_screen = found_screen or in_screen if in_screen and b"=" in body and not stripped.startswith((b";", b"#")): equals = body.find(b"=") key = body[:equals].strip().lower() setting = desired.get(key) if setting is not None: leading_value_space = body[equals + 1 :] leading_value_space = leading_value_space[ : len(leading_value_space) - len(leading_value_space.lstrip(b" \t")) ] result.append( body[: equals + 1] + leading_value_space + setting[1] + line_ending ) found_settings.add(key) continue result.append(line) if in_screen: append_missing_settings() elif not found_screen: if result and not result[-1].endswith((b"\r", b"\n")): result[-1] += newline if result and result[-1].strip(): result.append(newline) result.append(b"[Screen]" + newline) append_missing_settings() return b"".join(result) def _reject_save_symlinks(root: Path) -> None: for directory, directory_names, file_names in os.walk(root, followlinks=False): parent = Path(directory) for name in (*directory_names, *file_names): path = parent / name if path.is_symlink(): raise OverlayError(f"persistent game save contains a symbolic link: {path}") def _runtime_file_identity( name: str, path: Path, *, follow_symlinks: bool = True, ) -> RuntimeFileIdentity: state = path.stat(follow_symlinks=follow_symlinks) return RuntimeFileIdentity( name=name, device=state.st_dev, inode=state.st_ino, mode=state.st_mode, size=state.st_size, mtime_ns=state.st_mtime_ns, ctime_ns=state.st_ctime_ns, ) def _runtime_input_identities( paths: AppPaths, source: Path, ) -> tuple[RuntimeFileIdentity, ...]: """Snapshot inputs whose hashes established the cached runtime profile. Nanosecond mtime plus ctime/inode/device prevents ordinary edits, replacements, or timestamp restoration from bypassing the next authoritative hash check. """ identities = [ _runtime_file_identity("source-game-exe", source / "Game.exe"), _runtime_file_identity("source-game-dll", source / "game.dll"), _runtime_file_identity("source-scene", source / WHISPERING_VALE_RELATIVE), ] cache = paths.compatibility_binaries if cache.exists() or cache.is_symlink(): identities.append( _runtime_file_identity( "compatibility-cache", cache, follow_symlinks=False, ) ) if cache.is_dir(): for name in COMPATIBILITY_BINARY_HASHES: path = cache / name if path.exists() or path.is_symlink(): identities.append( _runtime_file_identity( f"compatibility-cache-{name.casefold()}", path, ) ) return tuple(identities) def _matching_cached_runtime_result( paths: AppPaths, source: Path, ) -> OverlayResult | None: result = _read_overlay_result(paths.runtime_manifest) if result is None: return None if result.source_game != str(source): return None if result.runtime_view != str(paths.runtime_view): return None try: current_identities = _runtime_input_identities(paths, source) except OSError: return None if current_identities != result.input_identities: return None return result def _cached_runtime_compatibility( paths: AppPaths, source: Path, ) -> RuntimeCompatibility | None: result = _matching_cached_runtime_result(paths, source) if result is None: return None return RuntimeCompatibility( compatible=True, source_game=str(source), source_scene_sha256=result.source_scene_sha256, scene_action=result.scene_action, binary_action=result.binary_action, reason=None, ) def _reuse_legacy_runtime_view( paths: AppPaths, source: Path, inspection: SceneInspection, scene_action: str, game_save: Path, game_config: Path, binary_action: str, game_exe_source: Path, game_dll_source: Path, source_binary_hashes: Mapping[str, str], input_identities: tuple[RuntimeFileIdentity, ...], ) -> OverlayResult | None: """Upgrade an unchanged v1 view after its original hash checks pass once.""" legacy = _read_legacy_overlay_result(paths.runtime_manifest) if legacy is None: return None if legacy.source_game != str(source): return None if legacy.runtime_view != str(paths.runtime_view): return None if legacy.source_scene_sha256 != inspection.file_sha256: return None if legacy.scene_action != scene_action or legacy.binary_action != binary_action: return None if legacy.game_exe_sha256 != source_binary_hashes.get("Game.exe"): return None if legacy.game_dll_sha256 != source_binary_hashes.get("game.dll"): return None _migrate_runtime_game_config_link( paths.runtime_view / "game.ini", source / "game.ini", game_config, ) if not _runtime_layout_ready( paths, source, game_save, game_config, game_exe_source, game_dll_source, legacy.links_created, ): return None runtime_scene = paths.runtime_view / WHISPERING_VALE_RELATIVE try: if _sha256(runtime_scene) != legacy.runtime_scene_sha256: return None if _runtime_input_identities(paths, source) != input_identities: raise OverlayError( "game files changed while the legacy runtime view was being verified" ) runtime_scene_identity = _runtime_file_identity( "runtime-scene", runtime_scene, ) except OSError: return None result = OverlayResult( source_game=legacy.source_game, runtime_view=legacy.runtime_view, source_scene_sha256=legacy.source_scene_sha256, runtime_scene_sha256=legacy.runtime_scene_sha256, scene_action=legacy.scene_action, binary_action=legacy.binary_action, game_exe_sha256=legacy.game_exe_sha256, game_dll_sha256=legacy.game_dll_sha256, input_identities=input_identities, runtime_scene_identity=runtime_scene_identity, created_at=legacy.created_at, links_created=legacy.links_created, ) _write_json( paths.runtime_manifest, { "configuration_version": RUNTIME_VIEW_CONFIGURATION_VERSION, **asdict(result), }, ) return result def _reuse_runtime_view( paths: AppPaths, source: Path, game_save: Path, game_config: Path, ) -> OverlayResult | None: result = _matching_cached_runtime_result(paths, source) if result is None: return None game_exe_source = source / "Game.exe" game_dll_source = source / "game.dll" _migrate_runtime_game_config_link( paths.runtime_view / "game.ini", source / "game.ini", game_config, ) if not _runtime_layout_ready( paths, source, game_save, game_config, game_exe_source, game_dll_source, result.links_created, ): return None runtime_scene = paths.runtime_view / WHISPERING_VALE_RELATIVE try: if ( _runtime_file_identity("runtime-scene", runtime_scene) != result.runtime_scene_identity ): return None if _runtime_input_identities(paths, source) != result.input_identities: return None except OSError: return None return result def _runtime_layout_ready( paths: AppPaths, source: Path, game_save: Path, game_config: Path, game_exe_source: Path, game_dll_source: Path, expected_links: int, ) -> bool: runtime = paths.runtime_view if runtime.is_symlink() or not runtime.is_dir(): return False top_level_exceptions = {"game", "game.ini", "save"} top_level_special = {"game", "game.ini", "save"} for name, binary_source in ( ("Game.exe", game_exe_source), ("game.dll", game_dll_source), ): if binary_source != source / name: top_level_exceptions.add(name) top_level_special.add(name) top_level_ready, top_level_links = _linked_level_ready( source, runtime, except_names=top_level_exceptions, special_names=top_level_special, ) if not top_level_ready: return False if not _symlink_points_to(runtime / "save", game_save): return False if not _symlink_points_to(runtime / "game.ini", game_config): return False if game_exe_source != source / "Game.exe" and not _symlink_points_to( runtime / "Game.exe", game_exe_source, ): return False if game_dll_source != source / "game.dll" and not _symlink_points_to( runtime / "game.dll", game_dll_source, ): return False game_source = source / "game" game_target = runtime / "game" if game_target.is_symlink() or not game_target.is_dir(): return False game_ready, game_links = _linked_level_ready( game_source, game_target, except_names={"maps"}, special_names={"maps"}, ) if not game_ready: return False maps_source = game_source / "maps" maps_target = game_target / "maps" if maps_target.is_symlink() or not maps_target.is_dir(): return False maps_ready, maps_links = _linked_level_ready( maps_source, maps_target, except_names={"WhisperingVale.rovug"}, special_names={"WhisperingVale.rovug"}, ) if not maps_ready: return False runtime_scene = maps_target / "WhisperingVale.rovug" if runtime_scene.is_symlink() or not runtime_scene.is_file(): return False actual_links = top_level_links + game_links + maps_links + 2 if game_exe_source != source / "Game.exe": actual_links += 1 if game_dll_source != source / "game.dll": actual_links += 1 return actual_links == expected_links def _linked_level_ready( source: Path, target: Path, *, except_names: set[str], special_names: set[str], ) -> tuple[bool, int]: try: source_children = { child.name: child for child in source.iterdir() if child.name not in except_names } target_names = {child.name for child in target.iterdir()} except OSError: return False, 0 if target_names != set(source_children) | special_names: return False, 0 for name, child in source_children.items(): if not _symlink_points_to(target / name, child): return False, 0 return True, len(source_children) def _symlink_points_to(link: Path, expected_target: Path) -> bool: if not link.is_symlink(): return False try: return link.resolve(strict=True) == expected_target.resolve(strict=True) except (OSError, RuntimeError): return False def _migrate_runtime_game_config_link( link: Path, source_config: Path, persistent_config: Path, ) -> bool: """Atomically redirect the previous source link without rebuilding the view.""" if _symlink_points_to(link, persistent_config): return True if not _symlink_points_to(link, source_config): return False temporary = link.with_name(f".{link.name}.{uuid.uuid4().hex}.tmp") try: temporary.symlink_to(persistent_config) temporary.replace(link) except OSError: temporary.unlink(missing_ok=True) return False return _symlink_points_to(link, persistent_config) def _read_overlay_result(path: Path) -> OverlayResult | None: try: value = json.loads(path.read_text(encoding="utf-8")) except (OSError, UnicodeError, json.JSONDecodeError): return None if not isinstance(value, dict): return None if value.get("configuration_version") != RUNTIME_VIEW_CONFIGURATION_VERSION: return None string_fields = ( "source_game", "runtime_view", "source_scene_sha256", "runtime_scene_sha256", "scene_action", "binary_action", "game_exe_sha256", "game_dll_sha256", "created_at", ) if not all(isinstance(value.get(name), str) for name in string_fields): return None links_created = value.get("links_created") if isinstance(links_created, bool) or not isinstance(links_created, int): return None if links_created < 0: return None input_values = value.get("input_identities") if not isinstance(input_values, list) or not input_values: return None input_identities: list[RuntimeFileIdentity] = [] for input_value in input_values: identity = _parse_runtime_file_identity(input_value) if identity is None: return None input_identities.append(identity) input_names = {identity.name for identity in input_identities} if len(input_names) != len(input_identities) or not { "source-game-exe", "source-game-dll", "source-scene", }.issubset(input_names): return None runtime_scene_identity = _parse_runtime_file_identity( value.get("runtime_scene_identity") ) if ( runtime_scene_identity is None or runtime_scene_identity.name != "runtime-scene" ): return None return OverlayResult( source_game=value["source_game"], runtime_view=value["runtime_view"], source_scene_sha256=value["source_scene_sha256"], runtime_scene_sha256=value["runtime_scene_sha256"], scene_action=value["scene_action"], binary_action=value["binary_action"], game_exe_sha256=value["game_exe_sha256"], game_dll_sha256=value["game_dll_sha256"], input_identities=tuple(input_identities), runtime_scene_identity=runtime_scene_identity, created_at=value["created_at"], links_created=links_created, ) def _parse_runtime_file_identity(value: object) -> RuntimeFileIdentity | None: if not isinstance(value, dict): return None name = value.get("name") if not isinstance(name, str) or not name: return None fields: dict[str, int] = {} for field in ("device", "inode", "mode", "size", "mtime_ns", "ctime_ns"): field_value = value.get(field) if ( isinstance(field_value, bool) or not isinstance(field_value, int) or field_value < 0 ): return None fields[field] = field_value return RuntimeFileIdentity( name=name, device=fields["device"], inode=fields["inode"], mode=fields["mode"], size=fields["size"], mtime_ns=fields["mtime_ns"], ctime_ns=fields["ctime_ns"], ) def _read_legacy_overlay_result(path: Path) -> LegacyOverlayResult | None: try: value = json.loads(path.read_text(encoding="utf-8")) except (OSError, UnicodeError, json.JSONDecodeError): return None if not isinstance(value, dict) or value.get("configuration_version") != 1: return None string_fields = ( "source_game", "runtime_view", "source_scene_sha256", "runtime_scene_sha256", "scene_action", "binary_action", "game_exe_sha256", "game_dll_sha256", "created_at", ) if not all(isinstance(value.get(name), str) for name in string_fields): return None links_created = value.get("links_created") if ( isinstance(links_created, bool) or not isinstance(links_created, int) or links_created < 0 ): return None return LegacyOverlayResult( source_game=value["source_game"], runtime_view=value["runtime_view"], source_scene_sha256=value["source_scene_sha256"], runtime_scene_sha256=value["runtime_scene_sha256"], scene_action=value["scene_action"], binary_action=value["binary_action"], game_exe_sha256=value["game_exe_sha256"], game_dll_sha256=value["game_dll_sha256"], created_at=value["created_at"], links_created=links_created, ) def _runtime_game_binary_sources( paths: AppPaths, source: Path, ) -> tuple[str, Path, Path]: action, game_exe, game_dll, _ = _runtime_game_binary_profile(paths, source) return action, game_exe, game_dll def _runtime_game_binary_profile( paths: AppPaths, source: Path, ) -> tuple[str, Path, Path, dict[str, str]]: cache = paths.compatibility_binaries cache_hashes = _binary_hashes(cache) if cache.is_dir() else None source_hashes = _binary_hashes(source) action = select_binary_action(cache_hashes, source_hashes) return ( action, source / "Game.exe", source / "game.dll", source_hashes, ) def _inspect_runtime_binary_action( paths: AppPaths, source: Path, ) -> str: """Mirror publisher binary selection without writing any runtime cache.""" cache = paths.compatibility_binaries cache_hashes = _binary_hashes(cache) if cache.is_dir() else None source_hashes = _binary_hashes(source) return select_binary_action(cache_hashes, source_hashes) def _binary_hashes(directory: Path) -> dict[str, str]: return { name: _sha256(directory / name) for name in COMPATIBILITY_BINARY_HASHES if (directory / name).is_file() } def _compatibility_result(paths: AppPaths, source: Path) -> CompatibilityCacheResult: return CompatibilityCacheResult( source=str(source), destination=str(paths.compatibility_binaries), game_exe_sha256=COMPATIBILITY_BINARY_HASHES["Game.exe"], game_dll_sha256=COMPATIBILITY_BINARY_HASHES["game.dll"], created_at=datetime.now().astimezone().isoformat(), ) def _remove_runtime_tree(path: Path) -> None: if path.is_symlink() or path.is_file(): path.unlink(missing_ok=True) elif path.is_dir(): shutil.rmtree(path) def _sha256(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as handle: for chunk in iter(lambda: handle.read(1024 * 1024), b""): digest.update(chunk) return digest.hexdigest() def _write_json(path: Path, value: object) -> None: path.parent.mkdir(parents=True, exist_ok=True) temporary = path.with_name(f".{path.name}.{os.getpid()}.tmp") temporary.write_text(json.dumps(value, indent=2, sort_keys=True) + "\n", encoding="utf-8") temporary.replace(path)