Files
pristontale-eu-linux/ptlinux/overlay.py

451 lines
15 KiB
Python

"""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 .v21_post_auth_reclaim import (
PATCHED_SHA256 as POST_AUTH_RECLAIM_GAME_DLL_SHA256,
PostAuthReclaimPatchError,
build_post_auth_reclaim_patch,
)
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"
)
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
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)
binary_action, game_exe_source, game_dll_source = _runtime_game_binary_sources(
paths,
source,
use_world_fix=use_world_fix,
)
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, 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],
runtime_patch_hash: str | None = None,
) -> str:
if runtime_patch_hash is not None:
if runtime_patch_hash != POST_AUTH_RECLAIM_GAME_DLL_SHA256:
raise OverlayError("post-authentication runtime patch has an unknown profile")
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 runtime_patch_hash is not None:
if source_hashes != CURRENT_V21_BINARY_HASHES:
raise OverlayError(
"post-authentication runtime patch does not match the installed client profile"
)
return "verified-v21-post-auth-world-fix"
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:
allowed_roots = (paths.data.resolve(strict=False), source.resolve(strict=True))
if not any(
candidate == root or candidate.is_relative_to(root)
for root in allowed_roots
):
raise OverlayError(f"refusing to migrate game save outside managed roots: {candidate}")
_reject_save_symlinks(candidate)
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 _runtime_game_binary_sources(
paths: AppPaths,
source: Path,
*,
use_world_fix: bool = True,
) -> tuple[str, Path, Path]:
cache = paths.compatibility_binaries
cache_hashes = _binary_hashes(cache) if cache.is_dir() else None
source_hashes = _binary_hashes(source)
patch = paths.runtime_patch_dll
patch_hash = None
if source_hashes == CURRENT_V21_BINARY_HASHES and use_world_fix:
if patch.exists() or patch.is_symlink():
if not patch.is_file() or patch.is_symlink():
raise OverlayError(f"runtime patch cache is not a regular file: {patch}")
patch_hash = _sha256(patch)
else:
try:
result = build_post_auth_reclaim_patch(source / "game.dll", patch)
except PostAuthReclaimPatchError as exc:
raise OverlayError(
f"could not build the verified post-authentication world fix: {exc}"
) from exc
patch_hash = result.destination_sha256
payload = {"ok": True, **asdict(result)}
_write_json(patch.parent / "patch.json", payload)
_write_json(paths.runtime_patch_manifest, payload)
action = select_binary_action(cache_hashes, source_hashes, patch_hash)
return (
action,
source / "Game.exe",
patch if action == "verified-v21-post-auth-world-fix" else source / "game.dll",
)
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)