pristontale-eu-linux/ptlinux/overlay.py

"""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"
)


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 = _runtime_game_binary_sources(
		paths,
		source,
	)

	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],
) -> 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"
	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,
) -> tuple[str, Path, Path]:
	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",
	)


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)