pristontale-eu-linux/ptlinux/world_fix.py

"""Retired V21 world-helper metadata and legacy status-file parsing."""

from __future__ import annotations

from dataclasses import dataclass
import json
import os
from pathlib import Path, PureWindowsPath


V21_BINARY_HASHES = {
	"Game.exe": "2bc3edd26c5f6d7ae9bbdc23967c5aefd374bdf934fcdfa2ed42c6442e550837",
	"game.dll": "4b4e5c28e2bfdaa8d36c8341fad27becfd03faf63aa8ea874f5d43c5b1961868",
}
WORLD_FIX_HELPER_FILENAME = "pt-world-fix.exe"
# Filled from the reproducible stripped PE after the native helper is built.
WORLD_FIX_HELPER_SHA256 = (
	"a7a17d29ef4c11f9603790f0110a4d6823b91cce03f70e466b7e0bdbd6fa313e"
)
WORLD_FIX_ENABLED = False
WORLD_FIX_DISABLED_REASON = (
	"the experimental world fix is retired because PristonTale V21 periodically "
	"checks the loaded game.dll code section; its five-byte map-loader detour "
	"causes Module Code Sync Error / Banned Code 1"
)
WORLD_FIX_STATUS_PROTOCOL = 1
WORLD_FIX_STATUS_MAX_BYTES = 64 * 1024


@dataclass(frozen=True)
class WorldFixStatusTarget:
	path: Path
	windows_path: str


class WorldFixError(RuntimeError):
	"""Raised when the retired helper is requested."""


def packaged_world_fix_helper() -> Path:
	return Path(__file__).resolve().parent / "native" / WORLD_FIX_HELPER_FILENAME


def resolve_world_fix_helper(
	game_directory: Path,
	*,
	helper: Path | None = None,
) -> Path | None:
	"""Reject the retired helper before it can alter publisher module memory."""

	_ = game_directory, helper
	raise WorldFixError(WORLD_FIX_DISABLED_REASON)


def create_world_fix_status_target(
	game_save: Path,
	started_ns: int,
) -> WorldFixStatusTarget:
	"""Allocate a unique launcher-owned status path without pre-creating it."""

	if started_ns <= 0:
		raise WorldFixError("world-fix status start time must be positive")
	directory = game_save / "launcher"
	if directory.is_symlink():
		raise WorldFixError(
			"the world-fix status directory must not be a symbolic link"
		)
	directory.mkdir(parents=True, exist_ok=True)
	if not directory.is_dir():
		raise WorldFixError("the world-fix status directory is unavailable")
	filename = f"ptworld-{started_ns}-{os.getpid()}.jsonl"
	path = directory / filename
	if path.exists() or path.is_symlink():
		raise WorldFixError(f"world-fix status target already exists: {path}")
	windows_path = str(PureWindowsPath("save", "launcher", filename))
	return WorldFixStatusTarget(path=path, windows_path=windows_path)


def read_world_fix_status(
	target: WorldFixStatusTarget | None,
	started_ns: int,
) -> dict[str, object] | None:
	"""Read only the bounded, fresh status stream belonging to this game run."""

	if target is None:
		return None
	report: dict[str, object] = {
		"path": str(target.path),
		"status": "missing",
		"fresh": False,
		"events": [],
	}
	try:
		if target.path.is_symlink() or not target.path.is_file():
			return report
		state = target.path.stat()
	except OSError as exc:
		report["status"] = "unreadable"
		report["error"] = str(exc)
		return report
	report["modified_at_ns"] = state.st_mtime_ns
	report["size"] = state.st_size
	if state.st_mtime_ns < started_ns:
		report["status"] = "stale"
		return report
	report["fresh"] = True
	if state.st_size > WORLD_FIX_STATUS_MAX_BYTES:
		report["status"] = "oversized"
		return report
	try:
		lines = target.path.read_text(encoding="utf-8").splitlines()
	except OSError as exc:
		report["status"] = "unreadable"
		report["error"] = str(exc)
		return report

	events: list[dict[str, object]] = []
	invalid_lines = 0
	for line in lines[:64]:
		try:
			value = json.loads(line)
		except (json.JSONDecodeError, TypeError):
			invalid_lines += 1
			continue
		clean = _validated_status_event(value)
		if clean is None:
			invalid_lines += 1
			continue
		events.append(clean)
	report["events"] = events
	if invalid_lines:
		report["invalid_lines"] = invalid_lines
	event_names = {
		event["event"]
		for event in events
		if isinstance(event.get("event"), str)
	}
	for name in (
		"helper_started",
		"game_started",
		"game_dll_verified",
		"hook_armed",
		"login_accepted",
		"world_fix_activated",
	):
		report[name] = name in event_names
	if "world_fix_activated" in event_names:
		report["status"] = "activated"
	elif "world_fix_activation_failed" in event_names:
		report["status"] = "activation_failed"
	elif "world_fix_not_activated" in event_names:
		report["status"] = "not_activated"
	elif "hook_armed" in event_names:
		report["status"] = "armed"
	elif event_names & {
		"game_path_failed",
		"game_command_failed",
		"game_start_failed",
		"game_dll_unavailable",
		"game_dll_rejected",
		"hook_arm_failed",
	}:
		report["status"] = "helper_failed"
	elif events:
		report["status"] = "started"
	else:
		report["status"] = "invalid"
	return report


def _validated_status_event(value: object) -> dict[str, object] | None:
	if not isinstance(value, dict):
		return None
	if value.get("protocol") != WORLD_FIX_STATUS_PROTOCOL:
		return None
	event = value.get("event")
	tick_ms = value.get("tick_ms")
	if not isinstance(event, str) or not event or len(event) > 64:
		return None
	if isinstance(tick_ms, bool) or not isinstance(tick_ms, int) or tick_ms < 0:
		return None
	clean: dict[str, object] = {
		"event": event,
		"tick_ms": tick_ms,
	}
	for key in ("process_id", "base", "size", "windows_error", "exit_code"):
		item = value.get(key)
		if isinstance(item, bool) or not isinstance(item, int) or item < 0:
			continue
		clean[key] = item
	return clean