"""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 from .world_fix import V21_BINARY_HASHES 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 = V21_BINARY_HASHES PATCHED_V21_GAME_DLL_SHA256 = ( "0be11b06ccf9be747e32b195c8d5980062f446d7d3a0a662ebfe0e602a01c538" ) RETIRED_POST_AUTH_GAME_DLL_SHA256 = ( "93cd27284eb294753da1111c99bfc9689f82b6d4a59302aee9fdbd8cb52452dd" ) RUNTIME_VIEW_CONFIGURATION_VERSION = 1 class OverlayError(RuntimeError): """Raised when a safe runtime view cannot be built.""" @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 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, *, use_world_fix: bool = True, ) -> 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. The normal preparation path still performs every authoritative check immediately 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) source_scene = source / WHISPERING_VALE_RELATIVE inspection = inspect_wrapped_assbin(source_scene) source_scene_sha256 = inspection.file_sha256 scene_action = _scene_action(inspection) # Retain the keyword for callers from the 0.5.8 candidate. The runtime # view and loaded process are now always publisher-pristine. _ = use_world_fix 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, *, use_world_fix: bool = True, ) -> 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) source_scene = source / WHISPERING_VALE_RELATIVE inspection = inspect_wrapped_assbin(source_scene) action = _scene_action(inspection) game_save = _prepare_persistent_save(paths, source) # ``use_world_fix`` is retained as a compatibility keyword. The retired # helper is never selected and no code path may replace the publisher DLL. _ = use_world_fix ( binary_action, game_exe_source, game_dll_source, source_binary_hashes, ) = _runtime_game_binary_profile( paths, source, ) reused = _reuse_runtime_view( paths, source, inspection, action, game_save, binary_action, game_exe_source, game_dll_source, source_binary_hashes, ) if reused is not None: return reused 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", "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 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 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"), 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 _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 _reuse_runtime_view( paths: AppPaths, source: Path, inspection: SceneInspection, scene_action: str, game_save: Path, binary_action: str, game_exe_source: Path, game_dll_source: Path, source_binary_hashes: Mapping[str, str], ) -> 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 if result.source_scene_sha256 != inspection.file_sha256: return None if result.scene_action != scene_action or result.binary_action != binary_action: return None if result.game_exe_sha256 != source_binary_hashes.get("Game.exe"): return None if result.game_dll_sha256 != source_binary_hashes.get("game.dll"): return None if not _runtime_layout_ready( paths, source, game_save, game_exe_source, game_dll_source, result.links_created, ): return None runtime_scene = paths.runtime_view / WHISPERING_VALE_RELATIVE try: if _sha256(runtime_scene) != result.runtime_scene_sha256: return None except OSError: return None return result def _runtime_layout_ready( paths: AppPaths, source: Path, game_save: 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", "save"} top_level_special = {"game", "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 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 + 1 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 _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 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"], 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)