14 Commits

23 changed files with 3314 additions and 446 deletions

View File

@@ -3,7 +3,7 @@
set -euo pipefail set -euo pipefail
SOURCE_ROOT="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd)" SOURCE_ROOT="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd)"
VERSION="${PTLINUX_VERSION:-0.5.2}" VERSION="${PTLINUX_VERSION:-0.5.9}"
if [[ ! "$VERSION" =~ ^[0-9]+\.[0-9]+\.[0-9]+$ ]] if [[ ! "$VERSION" =~ ^[0-9]+\.[0-9]+\.[0-9]+$ ]]
then then
printf 'Invalid launcher version: %s (expected X.Y.Z)\n' "$VERSION" >&2 printf 'Invalid launcher version: %s (expected X.Y.Z)\n' "$VERSION" >&2
@@ -133,10 +133,12 @@ RELEASE_ROOT="${RELEASES_ROOT}/${VERSION}"
if [[ ! -d "${SOURCE_ROOT}/ptlinux" \ if [[ ! -d "${SOURCE_ROOT}/ptlinux" \
|| ! -f "${SOURCE_ROOT}/ptdiag/assbin.py" \ || ! -f "${SOURCE_ROOT}/ptdiag/assbin.py" \
|| ! -f "${SOURCE_ROOT}/ptdiag/__init__.py" \ || ! -f "${SOURCE_ROOT}/ptdiag/__init__.py" \
|| ! -f "${SOURCE_ROOT}/ptdiag/logs.py" \
|| ! -f "${SOURCE_ROOT}/ptlinux/desktop_entry.py" \ || ! -f "${SOURCE_ROOT}/ptlinux/desktop_entry.py" \
|| ! -f "${SOURCE_ROOT}/ptlinux/launcher_wrapper.py" \ || ! -f "${SOURCE_ROOT}/ptlinux/launcher_wrapper.py" \
|| ! -f "${SOURCE_ROOT}/ptlinux/self_update.py" \ || ! -f "${SOURCE_ROOT}/ptlinux/self_update.py" \
|| ! -f "${SOURCE_ROOT}/ptlinux/launcher-update-public.pem" \ || ! -f "${SOURCE_ROOT}/ptlinux/launcher-update-public.pem" \
|| ! -f "${SOURCE_ROOT}/ptlinux/native/pt-world-fix.exe" \
|| ! -f "${SOURCE_ROOT}/packaging/eu.pristontale.PTLinux.desktop" \ || ! -f "${SOURCE_ROOT}/packaging/eu.pristontale.PTLinux.desktop" \
|| ! -f "${SOURCE_ROOT}/packaging/eu.pristontale.PTLinux.png" \ || ! -f "${SOURCE_ROOT}/packaging/eu.pristontale.PTLinux.png" \
|| ! -f "${SOURCE_ROOT}/packaging/pristontale-eu" \ || ! -f "${SOURCE_ROOT}/packaging/pristontale-eu" \
@@ -190,13 +192,20 @@ then
exit 1 exit 1
fi fi
install -d -m 0755 "$STAGING_RELEASE" install -d -m 0755 "$STAGING_RELEASE"
install -d -m 0755 "${STAGING_RELEASE}/ptlinux" "${STAGING_RELEASE}/ptdiag" install -d -m 0755 \
"${STAGING_RELEASE}/ptlinux" \
"${STAGING_RELEASE}/ptlinux/native" \
"${STAGING_RELEASE}/ptdiag"
install -m 0644 "${SOURCE_ROOT}/ptlinux/"*.py "${STAGING_RELEASE}/ptlinux/" install -m 0644 "${SOURCE_ROOT}/ptlinux/"*.py "${STAGING_RELEASE}/ptlinux/"
install -m 0644 \ install -m 0644 \
"${SOURCE_ROOT}/ptlinux/launcher-update-public.pem" \ "${SOURCE_ROOT}/ptlinux/launcher-update-public.pem" \
"${STAGING_RELEASE}/ptlinux/launcher-update-public.pem" "${STAGING_RELEASE}/ptlinux/launcher-update-public.pem"
install -m 0644 \
"${SOURCE_ROOT}/ptlinux/native/pt-world-fix.exe" \
"${STAGING_RELEASE}/ptlinux/native/pt-world-fix.exe"
install -m 0644 "${SOURCE_ROOT}/ptdiag/__init__.py" "${STAGING_RELEASE}/ptdiag/__init__.py" install -m 0644 "${SOURCE_ROOT}/ptdiag/__init__.py" "${STAGING_RELEASE}/ptdiag/__init__.py"
install -m 0644 "${SOURCE_ROOT}/ptdiag/assbin.py" "${STAGING_RELEASE}/ptdiag/assbin.py" install -m 0644 "${SOURCE_ROOT}/ptdiag/assbin.py" "${STAGING_RELEASE}/ptdiag/assbin.py"
install -m 0644 "${SOURCE_ROOT}/ptdiag/logs.py" "${STAGING_RELEASE}/ptdiag/logs.py"
printf '{\n "schema": 1,\n "version": "%s"\n}\n' "$VERSION" > "${STAGING_RELEASE}/release.json" printf '{\n "schema": 1,\n "version": "%s"\n}\n' "$VERSION" > "${STAGING_RELEASE}/release.json"
chmod 0644 "${STAGING_RELEASE}/release.json" chmod 0644 "${STAGING_RELEASE}/release.json"
@@ -301,8 +310,44 @@ then
update-desktop-database "$APPLICATION_HOME" >/dev/null 2>&1 || true update-desktop-database "$APPLICATION_HOME" >/dev/null 2>&1 || true
fi fi
printf '%s\n' "Application installed." printf '%s\n' "Launcher files installed."
DOCTOR_READY=1
if (( RUN_DOCTOR == 1 )) if (( RUN_DOCTOR == 1 ))
then then
"${BIN_HOME}/pristontale-eu" doctor || true if ! "${BIN_HOME}/pristontale-eu" doctor
then
DOCTOR_READY=0
fi
else
printf '%s\n' "Host readiness check skipped."
fi fi
printf '%s\n' "Start PristonTale EU with:"
printf ' %q gui\n' "${BIN_HOME}/pristontale-eu"
case ":${PATH:-}:" in
*":${BIN_HOME}:"*)
;;
*)
printf '%s\n' \
"Note: ${BIN_HOME} is not in PATH for this shell." \
"The application-menu shortcut already uses the full path."
if [[ "${SHELL##*/}" == "fish" ]]
then
printf '%s\n' "Enable the short command in Fish with:"
printf ' fish_add_path %q\n' "$BIN_HOME"
else
printf '%s\n' "Enable the short command in this Bash/Zsh session with:"
printf ' export PATH=%q:"$PATH"\n' "$BIN_HOME"
fi
;;
esac
if (( DOCTOR_READY == 0 ))
then
printf '%s\n' \
"Launcher files were installed, but this host is not ready to run the game." \
"Resolve the diagnostics above, then run the launcher command again." >&2
exit 1
fi
printf '%s\n' "Installation ready."

1179
native/pt_world_fix.c Normal file

File diff suppressed because it is too large Load Diff

View File

@@ -121,6 +121,9 @@ ptlinux_resolve_dependency_packages()
arch:openssl) arch:openssl)
candidates=(openssl) candidates=(openssl)
;; ;;
arch:umu)
candidates=(umu-launcher)
;;
suse:python) suse:python)
candidates=(python3) candidates=(python3)
;; ;;
@@ -210,6 +213,7 @@ ptlinux_has_wine_stack()
ptlinux_detect_missing_dependencies() ptlinux_detect_missing_dependencies()
{ {
local python_bin="${PYTHON_BIN:-python3}" local python_bin="${PYTHON_BIN:-python3}"
local family="${1:-}"
PTLINUX_MISSING_DEPENDENCIES=() PTLINUX_MISSING_DEPENDENCIES=()
if ! command -v "$python_bin" >/dev/null 2>&1 if ! command -v "$python_bin" >/dev/null 2>&1
@@ -227,10 +231,6 @@ ptlinux_detect_missing_dependencies()
then then
PTLINUX_MISSING_DEPENDENCIES+=(winetricks) PTLINUX_MISSING_DEPENDENCIES+=(winetricks)
fi fi
if ! command -v gamescope >/dev/null 2>&1
then
PTLINUX_MISSING_DEPENDENCIES+=(gamescope)
fi
if ! command -v cabextract >/dev/null 2>&1 if ! command -v cabextract >/dev/null 2>&1
then then
PTLINUX_MISSING_DEPENDENCIES+=(cabextract) PTLINUX_MISSING_DEPENDENCIES+=(cabextract)
@@ -239,6 +239,10 @@ ptlinux_detect_missing_dependencies()
then then
PTLINUX_MISSING_DEPENDENCIES+=(openssl) PTLINUX_MISSING_DEPENDENCIES+=(openssl)
fi fi
if [[ "$family" == "arch" ]] && ! command -v umu-run >/dev/null 2>&1
then
PTLINUX_MISSING_DEPENDENCIES+=(umu)
fi
} }
ptlinux_select_privilege_command() ptlinux_select_privilege_command()
@@ -344,15 +348,14 @@ ptlinux_install_host_dependencies()
local package_line local package_line
local -a packages=() local -a packages=()
ptlinux_detect_missing_dependencies ptlinux_read_distribution
family="$(ptlinux_dependency_family "$PTLINUX_DISTRIBUTION_ID" "$PTLINUX_DISTRIBUTION_LIKE")"
ptlinux_detect_missing_dependencies "$family"
if (( ${#PTLINUX_MISSING_DEPENDENCIES[@]} == 0 )) if (( ${#PTLINUX_MISSING_DEPENDENCIES[@]} == 0 ))
then then
printf '%s\n' "All required host dependencies are already available." printf '%s\n' "All required host dependencies are already available."
return 0 return 0
fi fi
ptlinux_read_distribution
family="$(ptlinux_dependency_family "$PTLINUX_DISTRIBUTION_ID" "$PTLINUX_DISTRIBUTION_LIKE")"
if [[ "$family" == "unknown" ]] if [[ "$family" == "unknown" ]]
then then
printf 'Unsupported distribution: %s. Re-run with --skip-deps after installing: %s\n' \ printf 'Unsupported distribution: %s. Re-run with --skip-deps after installing: %s\n' \
@@ -374,7 +377,7 @@ ptlinux_install_host_dependencies()
printf '%s\n' "Installing missing host dependencies (authentication may be requested)." printf '%s\n' "Installing missing host dependencies (authentication may be requested)."
ptlinux_install_resolved_packages "$family" "${packages[@]}" ptlinux_install_resolved_packages "$family" "${packages[@]}"
hash -r hash -r
ptlinux_detect_missing_dependencies ptlinux_detect_missing_dependencies "$family"
if (( ${#PTLINUX_MISSING_DEPENDENCIES[@]} != 0 )) if (( ${#PTLINUX_MISSING_DEPENDENCIES[@]} != 0 ))
then then
printf 'Dependencies remain unavailable after package installation: %s\n' \ printf 'Dependencies remain unavailable after package installation: %s\n' \

View File

@@ -23,6 +23,11 @@ _FAILED_IMPORT_RE = re.compile(
r"Failed to import scene:\s*(?P<scene>.*)$", r"Failed to import scene:\s*(?P<scene>.*)$",
re.IGNORECASE, re.IGNORECASE,
) )
_MODULE_CODE_SYNC_RE = re.compile(
r"Module Code Sync Error\s*\(\[(?P<left>-?\d+)\]\s*!=\s*"
r"\[(?P<right>-?\d+)\]\)",
re.IGNORECASE,
)
_TIMESTAMP_RE = re.compile( _TIMESTAMP_RE = re.compile(
r"^\s*(?:\[)?(?P<timestamp>" r"^\s*(?:\[)?(?P<timestamp>"
r"\d{4}-\d{2}-\d{2}[T ][0-9:.+Z-]+|\d{2}:\d{2}:\d{2}(?:\.\d+)?)" r"\d{4}-\d{2}-\d{2}[T ][0-9:.+Z-]+|\d{2}:\d{2}:\d{2}(?:\.\d+)?)"
@@ -72,6 +77,11 @@ class NativeLogAnalysis:
import_scene_threw: bool = False import_scene_threw: bool = False
stage_without_new_stage: bool = False stage_without_new_stage: bool = False
tex_reclaim_count: int = 0 tex_reclaim_count: int = 0
module_code_sync_error: bool = False
module_code_sync_left: int | None = None
module_code_sync_right: int | None = None
module_code_sync_delta: int | None = None
banned_code_1: bool = False
def latest_scene(self, name: str) -> SceneImport | None: def latest_scene(self, name: str) -> SceneImport | None:
wanted = _scene_key(name) wanted = _scene_key(name)
@@ -95,6 +105,11 @@ class NativeLogAnalysis:
"failed_files": [event.to_dict() for event in self.failed_files], "failed_files": [event.to_dict() for event in self.failed_files],
"stage_without_new_stage": self.stage_without_new_stage, "stage_without_new_stage": self.stage_without_new_stage,
"tex_reclaim_count": self.tex_reclaim_count, "tex_reclaim_count": self.tex_reclaim_count,
"module_code_sync_error": self.module_code_sync_error,
"module_code_sync_left": self.module_code_sync_left,
"module_code_sync_right": self.module_code_sync_right,
"module_code_sync_delta": self.module_code_sync_delta,
"banned_code_1": self.banned_code_1,
} }
for output_key, scene_name in ( for output_key, scene_name in (
("whispering_vale", "WhisperingVale"), ("whispering_vale", "WhisperingVale"),
@@ -243,6 +258,27 @@ def parse_native_log_lines(lines: Iterable[str]) -> NativeLogAnalysis:
analysis.import_scene_threw = True analysis.import_scene_threw = True
analysis.events.append(LogEvent("import_scene_threw", line_number, raw, timestamp)) analysis.events.append(LogEvent("import_scene_threw", line_number, raw, timestamp))
module_sync = _MODULE_CODE_SYNC_RE.search(raw)
if module_sync is not None:
left = int(module_sync.group("left"))
right = int(module_sync.group("right"))
analysis.module_code_sync_error = True
analysis.module_code_sync_left = left
analysis.module_code_sync_right = right
analysis.module_code_sync_delta = abs(left - right)
analysis.events.append(
LogEvent(
"module_code_sync_error",
line_number,
raw,
timestamp,
f"{left} != {right}",
)
)
if "banned code 1" in raw.casefold():
analysis.banned_code_1 = True
analysis.events.append(LogEvent("banned_code_1", line_number, raw, timestamp))
failed_import = _FAILED_IMPORT_RE.search(raw) failed_import = _FAILED_IMPORT_RE.search(raw)
if failed_import is not None: if failed_import is not None:
value = failed_import.group("scene").strip() value = failed_import.group("scene").strip()

View File

@@ -4,4 +4,4 @@ from .paths import AppPaths
__all__ = ["AppPaths", "__version__"] __all__ = ["AppPaths", "__version__"]
__version__ = "0.5.2" __version__ = "0.5.11"

View File

@@ -9,8 +9,16 @@ from datetime import datetime
import json import json
from pathlib import Path from pathlib import Path
import sys import sys
import time
from typing import Sequence from typing import Sequence
from ptdiag.logs import (
LogFileState,
newest_current_native_log,
parse_current_run_log,
snapshot_native_logs,
)
from . import __version__ from . import __version__
from .command import CommandError, CommandRunner from .command import CommandError, CommandRunner
from .depot import SyncProgress, sync_managed_game from .depot import SyncProgress, sync_managed_game
@@ -26,7 +34,8 @@ from .launcher import (
) )
from .locking import launcher_operation_lock from .locking import launcher_operation_lock
from .login_diagnostics import inspect_login_environment from .login_diagnostics import inspect_login_environment
from .overlay import cache_compatibility_binaries, cache_runtime_patch, prepare_runtime_view from .login_trace import DEFAULT_TRACE_TIMEOUT_SECONDS, run_login_trace
from .overlay import cache_compatibility_binaries, prepare_runtime_view
from .paths import AppPaths from .paths import AppPaths
from .runtime import ( from .runtime import (
prepare_game_runtime, prepare_game_runtime,
@@ -122,20 +131,37 @@ def build_parser() -> argparse.ArgumentParser:
help="cache one exact verified Game.exe/game.dll compatibility pair", help="cache one exact verified Game.exe/game.dll compatibility pair",
) )
compatibility.add_argument("source", type=Path) compatibility.add_argument("source", type=Path)
runtime_patch = subparsers.add_parser(
"cache-runtime-patch",
help="cache the exact verified V21 texture-reclaim game.dll",
)
runtime_patch.add_argument("source", type=Path)
prepare = subparsers.add_parser("prepare", help="build the read-only runtime view and scene fix") prepare = subparsers.add_parser("prepare", help="build the read-only runtime view and scene fix")
prepare.add_argument("--game-dir", type=Path) prepare.add_argument("--game-dir", type=Path)
play = subparsers.add_parser("play", help="launch the game in the clean Gamescope box") play = subparsers.add_parser("play", help="launch the game in the managed Proton window")
play.add_argument("--game-dir", type=Path) play.add_argument("--game-dir", type=Path)
play.add_argument("--no-box", action="store_true", help="launch without Gamescope") play.add_argument("--no-box", action="store_true", help="launch without X11 window styling")
play.add_argument(
"--publisher-dll",
"--no-world-fix",
action="store_true",
dest="publisher_dll",
help=(
"compatibility option retained for older scripts; publisher code is "
"always used without the retired world fix"
),
)
play.add_argument("--dry-run", action="store_true") play.add_argument("--dry-run", action="store_true")
play.add_argument("--skip-runtime-setup", action="store_true") play.add_argument("--skip-runtime-setup", action="store_true")
login_trace = subparsers.add_parser(
"login-trace",
help="trace the post-authentication account check without retaining credentials",
)
login_trace.add_argument("--game-dir", type=Path)
login_trace.add_argument("--no-box", action="store_true", help="launch without X11 window styling")
login_trace.add_argument("--skip-runtime-setup", action="store_true")
login_trace.add_argument(
"--timeout",
type=float,
default=DEFAULT_TRACE_TIMEOUT_SECONDS,
help="maximum game runtime in seconds (default: 600)",
)
settings = subparsers.add_parser("settings", help="show or initialize settings") settings = subparsers.add_parser("settings", help="show or initialize settings")
settings.add_argument("--init", action="store_true", help="write conservative defaults") settings.add_argument("--init", action="store_true", help="write conservative defaults")
@@ -229,10 +255,6 @@ def main(argv: Sequence[str] | None = None) -> int:
result = cache_compatibility_binaries(paths, arguments.source) result = cache_compatibility_binaries(paths, arguments.source)
_emit({"ok": True, "compatibility": asdict(result)}, arguments.json) _emit({"ok": True, "compatibility": asdict(result)}, arguments.json)
return 0 return 0
if command == "cache-runtime-patch":
result = cache_runtime_patch(paths, arguments.source)
_emit({"ok": True, "runtime_patch": asdict(result)}, arguments.json)
return 0
if command == "prepare": if command == "prepare":
return _prepare(paths, arguments.game_dir, arguments.json) return _prepare(paths, arguments.game_dir, arguments.json)
if command == "play": if command == "play":
@@ -240,10 +262,20 @@ def main(argv: Sequence[str] | None = None) -> int:
paths, paths,
arguments.game_dir, arguments.game_dir,
arguments.no_box, arguments.no_box,
arguments.publisher_dll,
arguments.dry_run, arguments.dry_run,
arguments.skip_runtime_setup, arguments.skip_runtime_setup,
arguments.json, arguments.json,
) )
if command == "login-trace":
return _login_trace(
paths,
arguments.game_dir,
arguments.no_box,
arguments.skip_runtime_setup,
arguments.timeout,
arguments.json,
)
if command == "settings": if command == "settings":
return _settings(paths, arguments.init, arguments.json) return _settings(paths, arguments.init, arguments.json)
if command == "launcher-update": if command == "launcher-update":
@@ -256,6 +288,13 @@ def main(argv: Sequence[str] | None = None) -> int:
arguments.json, arguments.json,
) )
parser.error(f"unknown command: {command}") parser.error(f"unknown command: {command}")
except KeyboardInterrupt:
message = "Interrupted; PristonTale and its compatibility runtime were stopped."
if arguments.json:
_print_json({"ok": False, "error": message, "interrupted": True})
else:
print(f"\n{message}", file=sys.stderr)
return 130
except (ValueError, OSError, RuntimeError) as exc: except (ValueError, OSError, RuntimeError) as exc:
if arguments.json: if arguments.json:
_print_json({"ok": False, "error": str(exc)}) _print_json({"ok": False, "error": str(exc)})
@@ -300,6 +339,7 @@ def _login_diagnostics(paths: AppPaths, timeout_seconds: float, as_json: bool) -
probe = report.login_server probe = report.login_server
print(f"Launcher: {report.launcher_version} ({report.launcher_module})") print(f"Launcher: {report.launcher_version} ({report.launcher_module})")
print(f"Python: {report.python_executable}") print(f"Python: {report.python_executable}")
print(f"Host: {report.distribution_id} ({report.kernel_release})")
status = "REACHABLE" if probe.reachable else "UNREACHABLE" status = "REACHABLE" if probe.reachable else "UNREACHABLE"
latency = f", {probe.latency_ms:.1f} ms" if probe.latency_ms is not None else "" latency = f", {probe.latency_ms:.1f} ms" if probe.latency_ms is not None else ""
print(f"Login server: {status} ({probe.host}:{probe.port}{latency})") print(f"Login server: {status} ({probe.host}:{probe.port}{latency})")
@@ -313,10 +353,19 @@ def _login_diagnostics(paths: AppPaths, timeout_seconds: float, as_json: bool) -
print(f"game.dll: {report.game_dll}") print(f"game.dll: {report.game_dll}")
if report.game_dll_sha256: if report.game_dll_sha256:
print(f"DLL SHA-256: {report.game_dll_sha256}") print(f"DLL SHA-256: {report.game_dll_sha256}")
world_fix_status = "enabled" if report.world_fix_enabled else "retired"
print(f"World fix: {world_fix_status}")
if report.world_fix_disabled_reason:
print(f"Reason: {report.world_fix_disabled_reason}")
if report.runtime_kind: if report.runtime_kind:
print(f"Runtime: {report.runtime_kind} ({report.runtime_path or 'unknown path'})") print(f"Runtime: {report.runtime_kind} ({report.runtime_path or 'unknown path'})")
else: else:
print(f"Runtime: unavailable ({report.runtime_error or 'selection failed'})") print(f"Runtime: unavailable ({report.runtime_error or 'selection failed'})")
if report.proton_sync_effective is not None:
print(
"Proton sync: "
f"{report.proton_sync_configured} -> {report.proton_sync_effective}"
)
prefix_status = "ready" if report.prefix_ready else "not ready" prefix_status = "ready" if report.prefix_ready else "not ready"
directx_status = "ready" if report.directx_ready else "not ready" directx_status = "ready" if report.directx_ready else "not ready"
print(f"Prefix: {prefix_status}; DirectX helpers {directx_status}") print(f"Prefix: {prefix_status}; DirectX helpers {directx_status}")
@@ -555,10 +604,35 @@ def _play(
paths: AppPaths, paths: AppPaths,
game_dir: Path | None, game_dir: Path | None,
no_box: bool, no_box: bool,
publisher_dll: bool,
dry_run: bool, dry_run: bool,
skip_runtime_setup: bool, skip_runtime_setup: bool,
as_json: bool, as_json: bool,
) -> int: ) -> int:
with _operation_guard(paths, "play", dry_run):
return _play_locked(
paths,
game_dir,
no_box,
publisher_dll,
dry_run,
skip_runtime_setup,
as_json,
)
def _play_locked(
paths: AppPaths,
game_dir: Path | None,
no_box: bool,
publisher_dll: bool,
dry_run: bool,
skip_runtime_setup: bool,
as_json: bool,
) -> int:
# The legacy flag is retained for command-line compatibility. V21 performs
# periodic module-code checks, so every production launch is publisher-pristine.
_ = publisher_dll
settings = _load_or_default_settings(paths, persist=not dry_run) settings = _load_or_default_settings(paths, persist=not dry_run)
source = _source_game(paths, settings, game_dir) source = _source_game(paths, settings, game_dir)
overlay = None overlay = None
@@ -573,13 +647,36 @@ def _play(
_print_operation_status(as_json, 3, 4, "Preparing Proton and DirectX components…") _print_operation_status(as_json, 3, 4, "Preparing Proton and DirectX components…")
setup_results = prepare_game_runtime(paths, selection, runner) setup_results = prepare_game_runtime(paths, selection, runner)
_print_operation_status(as_json, 4, 4, "Starting the PristonTale game client…") _print_operation_status(as_json, 4, 4, "Starting the PristonTale game client…")
outcome = run_game( native_log_directory = paths.game_save / "logs"
paths, native_log_baseline = snapshot_native_logs(
settings, native_log_directory,
selection, patterns=("log_*.txt",),
runner,
use_gamebox=not no_box,
) )
game_started_ns = time.time_ns()
try:
outcome = run_game(
paths,
settings,
selection,
runner,
use_gamebox=not no_box,
use_world_fix=False,
)
except KeyboardInterrupt:
current_game_log = _current_game_log(
native_log_directory,
native_log_baseline,
game_started_ns,
)
if current_game_log is not None and not as_json:
_print_interrupted_game_log(current_game_log)
raise
current_game_log = _current_game_log(
native_log_directory,
native_log_baseline,
game_started_ns,
)
world_fix_status = None
_emit( _emit(
{ {
"ok": outcome.returncode == 0, "ok": outcome.returncode == 0,
@@ -588,12 +685,112 @@ def _play(
"runtime_view": asdict(overlay) if overlay else None, "runtime_view": asdict(overlay) if overlay else None,
"setup_commands": [asdict(result) for result in setup_results], "setup_commands": [asdict(result) for result in setup_results],
"game": asdict(outcome), "game": asdict(outcome),
"current_game_log": current_game_log,
"world_fix": world_fix_status,
}, },
as_json, as_json,
) )
return 0 if outcome.returncode == 0 else 1 return 0 if outcome.returncode == 0 else 1
def _current_game_log(
directory: Path,
baseline: dict[str, LogFileState],
started_ns: int,
) -> dict[str, object] | None:
state = newest_current_native_log(
directory,
baseline,
started_ns,
patterns=("log_*.txt",),
)
if state is None:
return None
analysis = parse_current_run_log(state, baseline)
return {
"path": str(state.path),
"modified_at": datetime.fromtimestamp(
state.mtime_ns / 1_000_000_000
).astimezone().isoformat(),
"size": state.size,
"load_texture_to_material_failed": analysis.load_texture_to_material_failed,
"import_scene_threw": analysis.import_scene_threw,
"failed_import_scenes": analysis.failed_import_scenes,
"stage_without_new_stage": analysis.stage_without_new_stage,
"module_code_sync_error": analysis.module_code_sync_error,
"module_code_sync_left": analysis.module_code_sync_left,
"module_code_sync_right": analysis.module_code_sync_right,
"module_code_sync_delta": analysis.module_code_sync_delta,
"banned_code_1": analysis.banned_code_1,
"stage_without_new_stage_events": [
event.to_dict()
for event in analysis.events
if event.kind == "stage_without_new_stage"
],
"scene_imports": {
name: [scene.to_dict() for scene in imports]
for name, imports in sorted(analysis.scenes.items())
},
"failed_file_count": len(analysis.failed_files),
"failed_files": [event.to_dict() for event in analysis.failed_files[:64]],
}
def _print_interrupted_game_log(report: dict[str, object]) -> None:
print(f"\nCurrent game log: {report['path']}", file=sys.stderr)
if report["module_code_sync_error"]:
print(
"Detected a publisher module-integrity mismatch; the game was launched "
"with publisher-pristine code in this launcher version.",
file=sys.stderr,
)
elif report["load_texture_to_material_failed"]:
print(
"Detected a texture/material failure while loading the game world.",
file=sys.stderr,
)
elif report["import_scene_threw"] or report["failed_import_scenes"]:
print("Detected a game-world scene import failure.", file=sys.stderr)
def _login_trace(
paths: AppPaths,
game_dir: Path | None,
no_box: bool,
skip_runtime_setup: bool,
timeout_seconds: float,
as_json: bool,
) -> int:
settings = _load_or_default_settings(paths, persist=True)
source = _source_game(paths, settings, game_dir)
runner = _runner(paths, "login-trace", False)
with _operation_guard(paths, "login-trace", False):
_print_operation_status(as_json, 1, 4, "Preparing the managed game view…")
prepare_runtime_view(paths, source)
_print_operation_status(as_json, 2, 4, "Selecting the configured Proton runtime…")
selection = select_runtime(settings)
if not skip_runtime_setup:
_print_operation_status(as_json, 3, 4, "Checking Proton and DirectX components…")
prepare_game_runtime(paths, selection, runner)
_print_operation_status(as_json, 4, 4, "Starting the privacy-safe login trace…")
if not as_json:
print(
"Enter the account name with its registered capitalization, press Login, "
"leave the approval screen visible for five seconds, then close the game.",
flush=True,
)
report = run_login_trace(
paths,
settings,
selection,
runner,
use_gamebox=not no_box,
timeout_seconds=timeout_seconds,
)
_emit({"ok": True, "login_trace": report.to_dict()}, as_json)
return 0
def _settings(paths: AppPaths, initialize: bool, as_json: bool) -> int: def _settings(paths: AppPaths, initialize: bool, as_json: bool) -> int:
if initialize: if initialize:
save_settings(paths, Settings()) save_settings(paths, Settings())
@@ -659,14 +856,18 @@ def _launcher_update(
def _source_game( def _source_game(
paths: AppPaths, paths: AppPaths,
settings: Settings, _settings: Settings,
argument: Path | None, argument: Path | None,
) -> Path: ) -> Path:
"""Select the managed game unless an explicit diagnostic path was supplied.
The persisted ``game_source`` setting is an import seed only. It must not
redirect normal Play, Prepare, or Login Trace operations away from the
launcher-managed tree.
"""
if argument is not None: if argument is not None:
return argument.expanduser().resolve() return argument.expanduser().resolve()
source_override = settings.source_override
if source_override is not None:
return source_override
return paths.source_game return paths.source_game

View File

@@ -7,11 +7,15 @@ from datetime import datetime
import json import json
import os import os
from pathlib import Path from pathlib import Path
import signal
import subprocess import subprocess
import sys import sys
import threading import threading
import time import time
from typing import Mapping, Sequence, TextIO from typing import Callable, Mapping, Sequence, TextIO
OutputLineFilter = Callable[[str], str | None]
@dataclass(frozen=True) @dataclass(frozen=True)
@@ -53,6 +57,7 @@ class CommandRunner:
timeout_seconds: float = 300.0, timeout_seconds: float = 300.0,
check: bool = True, check: bool = True,
stream_output: bool = False, stream_output: bool = False,
output_line_filter: OutputLineFilter | None = None,
) -> CommandOutcome: ) -> CommandOutcome:
arguments = [str(value) for value in argv] arguments = [str(value) for value in argv]
if not arguments: if not arguments:
@@ -81,6 +86,7 @@ class CommandRunner:
environment=environment, environment=environment,
timeout_seconds=timeout_seconds, timeout_seconds=timeout_seconds,
started=started, started=started,
output_line_filter=output_line_filter,
) )
else: else:
try: try:
@@ -126,6 +132,7 @@ class CommandRunner:
environment: Mapping[str, str], environment: Mapping[str, str],
timeout_seconds: float, timeout_seconds: float,
started: float, started: float,
output_line_filter: OutputLineFilter | None,
) -> CommandOutcome: ) -> CommandOutcome:
process = subprocess.Popen( process = subprocess.Popen(
arguments, arguments,
@@ -135,6 +142,7 @@ class CommandRunner:
stdout=subprocess.PIPE, stdout=subprocess.PIPE,
stderr=subprocess.PIPE, stderr=subprocess.PIPE,
bufsize=1, bufsize=1,
start_new_session=True,
) )
stdout: list[str] = [] stdout: list[str] = []
stderr: list[str] = [] stderr: list[str] = []
@@ -144,13 +152,21 @@ class CommandRunner:
capture: list[str], capture: list[str],
destination: TextIO, destination: TextIO,
) -> None: ) -> None:
while True: try:
line = stream.readline() while True:
if not line: line = stream.readline()
return if not line:
capture.append(line) return
destination.write(line) if output_line_filter is not None:
destination.flush() line = output_line_filter(line)
if line is None:
continue
capture.append(line)
destination.write(line)
destination.flush()
except (OSError, ValueError):
# Shutdown can close a pipe while its relay thread is leaving readline().
return
assert process.stdout is not None assert process.stdout is not None
assert process.stderr is not None assert process.stderr is not None
@@ -171,12 +187,12 @@ class CommandRunner:
returncode = process.wait(timeout=timeout_seconds) returncode = process.wait(timeout=timeout_seconds)
except subprocess.TimeoutExpired: except subprocess.TimeoutExpired:
timed_out = True timed_out = True
process.kill() returncode = _terminate_process_group(process)
returncode = process.wait() except BaseException:
stdout_thread.join() _terminate_process_group(process)
stderr_thread.join() _finish_stream_relays(process, stdout_thread, stderr_thread)
process.stdout.close() raise
process.stderr.close() _finish_stream_relays(process, stdout_thread, stderr_thread)
return CommandOutcome( return CommandOutcome(
argv=arguments, argv=arguments,
returncode=None if timed_out else returncode, returncode=None if timed_out else returncode,
@@ -210,3 +226,49 @@ def _as_text(value: bytes | str | None) -> str:
if isinstance(value, bytes): if isinstance(value, bytes):
return value.decode("utf-8", errors="replace") return value.decode("utf-8", errors="replace")
return value return value
def _terminate_process_group(
process: subprocess.Popen[str],
*,
grace_seconds: float = 5.0,
) -> int | None:
"""Stop the complete child session, allowing Proton a short clean exit."""
if process.poll() is not None:
return process.returncode
try:
os.killpg(process.pid, signal.SIGTERM)
except ProcessLookupError:
return process.poll()
try:
return process.wait(timeout=grace_seconds)
except subprocess.TimeoutExpired:
pass
try:
os.killpg(process.pid, signal.SIGKILL)
except ProcessLookupError:
return process.poll()
try:
return process.wait(timeout=grace_seconds)
except subprocess.TimeoutExpired:
return process.poll()
def _finish_stream_relays(
process: subprocess.Popen[str],
stdout_thread: threading.Thread,
stderr_thread: threading.Thread,
) -> None:
"""Bound relay cleanup so a second interrupt can never be required."""
stdout_thread.join(timeout=2.0)
stderr_thread.join(timeout=2.0)
for stream in (process.stdout, process.stderr):
if stream is not None:
try:
stream.close()
except OSError:
pass
stdout_thread.join(timeout=0.25)
stderr_thread.join(timeout=0.25)

View File

@@ -70,8 +70,8 @@ def inspect_host(environment: Mapping[str, str] | None = None) -> DoctorReport:
ToolStatus( ToolStatus(
"gamescope", "gamescope",
shutil.which("gamescope"), shutil.which("gamescope"),
True, False,
"borderless X11/Wayland game container", "optional manual game-container diagnostics",
), ),
ToolStatus( ToolStatus(
"cabextract", "cabextract",
@@ -147,22 +147,22 @@ def _python_tk_path() -> str | None:
def _package_hint(distribution: str) -> str | None: def _package_hint(distribution: str) -> str | None:
if distribution in {"ubuntu", "debian", "linuxmint", "pop"}: if distribution in {"ubuntu", "debian", "linuxmint", "pop"}:
return ( return (
"Install Python/Tk, Wine (64- and 32-bit), Winetricks, Gamescope, " "Install Python/Tk, Wine (64- and 32-bit), Winetricks, "
"Cabextract, OpenSSL, and both 64- and 32-bit Vulkan drivers with APT." "Cabextract, OpenSSL, and both 64- and 32-bit Vulkan drivers with APT."
) )
if distribution in {"fedora", "nobara"}: if distribution in {"fedora", "nobara"}:
return ( return (
"Install python3-tkinter, wine, winetricks, gamescope, cabextract, openssl, " "Install python3-tkinter, wine, winetricks, cabextract, openssl, "
"and the matching 32-bit Vulkan driver with DNF." "and the matching 32-bit Vulkan driver with DNF."
) )
if distribution in {"arch", "cachyos", "endeavouros", "manjaro"}: if distribution in {"arch", "cachyos", "endeavouros", "manjaro"}:
return ( return (
"Install python, tk, wine, winetricks, gamescope, cabextract, openssl, and " "Install python, tk, wine, winetricks, umu-launcher, cabextract, "
"the matching lib32 Vulkan driver with Pacman." "openssl, and the matching lib32 Vulkan driver with Pacman."
) )
if distribution in {"opensuse-tumbleweed", "opensuse-leap", "suse"}: if distribution in {"opensuse-tumbleweed", "opensuse-leap", "suse"}:
return ( return (
"Install Python/Tk, Wine, Winetricks, Gamescope, Cabextract, OpenSSL, and " "Install Python/Tk, Wine, Winetricks, Cabextract, OpenSSL, and "
"32-bit Vulkan support with Zypper." "32-bit Vulkan support with Zypper."
) )
return None return None

View File

@@ -23,5 +23,7 @@ def build_gamebox_command(
] ]
if settings.borderless: if settings.borderless:
command.append("--borderless") command.append("--borderless")
if settings.suppress_unresponsive_warnings:
command.append("--suppress-wm-ping")
command.extend(("--", *[str(value) for value in game_command])) command.extend(("--", *[str(value) for value in game_command]))
return command return command

View File

@@ -44,6 +44,14 @@ def _select_game_window(
return None return None
def _protocols_without(
protocols: Sequence[int],
blocked_protocol: int,
) -> tuple[int, ...]:
"""Remove only the selected atom while preserving every other protocol."""
return tuple(protocol for protocol in protocols if protocol != blocked_protocol)
class GameboxRuntimeError(RuntimeError): class GameboxRuntimeError(RuntimeError):
"""Raised when the game command or X11 mapper cannot be started.""" """Raised when the game command or X11 mapper cannot be started."""
@@ -127,6 +135,58 @@ class _X11:
self.library.XMapRaised(self.display, window) self.library.XMapRaised(self.display, window)
self.library.XFlush(self.display) self.library.XFlush(self.display)
def suppress_wm_ping(self, window: int) -> bool:
"""Stop the WM hang prompt for one window, preserving its other protocols."""
ping_atom = int(self.library.XInternAtom(self.display, b"_NET_WM_PING", 0))
if ping_atom == 0:
return False
protocols = self._wm_protocols(window)
if protocols is None:
return False
filtered_protocols = _protocols_without(protocols, ping_atom)
if filtered_protocols == protocols:
return True
if filtered_protocols:
protocol_values = (ctypes.c_ulong * len(filtered_protocols))(
*filtered_protocols
)
status = self.library.XSetWMProtocols(
self.display,
window,
protocol_values,
len(filtered_protocols),
)
else:
status = self.library.XSetWMProtocols(
self.display,
window,
ctypes.POINTER(ctypes.c_ulong)(),
0,
)
if status == 0:
return False
self.library.XSync(self.display, 0)
return self._wm_protocols(window) == filtered_protocols
def _wm_protocols(self, window: int) -> tuple[int, ...] | None:
protocols = ctypes.POINTER(ctypes.c_ulong)()
protocol_count = ctypes.c_int()
status = self.library.XGetWMProtocols(
self.display,
window,
ctypes.byref(protocols),
ctypes.byref(protocol_count),
)
if status == 0:
return None
try:
return tuple(
int(protocols[index]) for index in range(protocol_count.value)
)
finally:
if protocols:
self.library.XFree(ctypes.cast(protocols, ctypes.c_void_p))
def _remove_decorations(self, window: int) -> None: def _remove_decorations(self, window: int) -> None:
atom = self.library.XInternAtom(self.display, b"_MOTIF_WM_HINTS", 0) atom = self.library.XInternAtom(self.display, b"_MOTIF_WM_HINTS", 0)
if atom == 0: if atom == 0:
@@ -205,8 +265,24 @@ class _X11:
ctypes.c_int, ctypes.c_int,
] ]
self.library.XChangeProperty.restype = ctypes.c_int self.library.XChangeProperty.restype = ctypes.c_int
self.library.XGetWMProtocols.argtypes = [
ctypes.c_void_p,
ctypes.c_ulong,
ctypes.POINTER(ctypes.POINTER(ctypes.c_ulong)),
ctypes.POINTER(ctypes.c_int),
]
self.library.XGetWMProtocols.restype = ctypes.c_int
self.library.XSetWMProtocols.argtypes = [
ctypes.c_void_p,
ctypes.c_ulong,
ctypes.POINTER(ctypes.c_ulong),
ctypes.c_int,
]
self.library.XSetWMProtocols.restype = ctypes.c_int
self.library.XMapRaised.argtypes = [ctypes.c_void_p, ctypes.c_ulong] self.library.XMapRaised.argtypes = [ctypes.c_void_p, ctypes.c_ulong]
self.library.XMapRaised.restype = ctypes.c_int self.library.XMapRaised.restype = ctypes.c_int
self.library.XSync.argtypes = [ctypes.c_void_p, ctypes.c_int]
self.library.XSync.restype = ctypes.c_int
self.library.XFlush.argtypes = [ctypes.c_void_p] self.library.XFlush.argtypes = [ctypes.c_void_p]
self.library.XFlush.restype = ctypes.c_int self.library.XFlush.restype = ctypes.c_int
self.library.XFree.argtypes = [ctypes.c_void_p] self.library.XFree.argtypes = [ctypes.c_void_p]
@@ -215,11 +291,46 @@ class _X11:
self.library.XCloseDisplay.restype = ctypes.c_int self.library.XCloseDisplay.restype = ctypes.c_int
def _prepare_game_window(
mapper: _X11,
window: int,
*,
borderless: bool,
suppress_wm_ping: bool,
) -> None:
if suppress_wm_ping:
try:
ping_suppressed = mapper.suppress_wm_ping(window)
except Exception as exc:
print(
f"Warning: could not suppress game-window hang warnings ({exc}); "
"continuing normally",
file=sys.stderr,
flush=True,
)
else:
if ping_suppressed:
print(
"Game-window hang warnings suppressed for this launch.",
file=sys.stderr,
flush=True,
)
else:
print(
"Warning: could not verify game-window hang-warning "
"suppression; continuing normally",
file=sys.stderr,
flush=True,
)
mapper.map_window(window, borderless=borderless)
def run_game_window( def run_game_window(
command: Sequence[str], command: Sequence[str],
window_title: str = WINDOW_TITLE, window_title: str = WINDOW_TITLE,
*, *,
borderless: bool = False, borderless: bool = False,
suppress_wm_ping: bool = False,
) -> int: ) -> int:
if not command: if not command:
raise GameboxRuntimeError("game command cannot be empty") raise GameboxRuntimeError("game command cannot be empty")
@@ -270,7 +381,12 @@ def run_game_window(
and game_window_created_at is not None and game_window_created_at is not None
and now - game_window_created_at >= WINDOW_READY_DELAY_SECONDS and now - game_window_created_at >= WINDOW_READY_DELAY_SECONDS
): ):
mapper.map_window(game_window, borderless=borderless) _prepare_game_window(
mapper,
game_window,
borderless=borderless,
suppress_wm_ping=suppress_wm_ping,
)
print( print(
f"Game window ready: {game_title or window_title}", f"Game window ready: {game_title or window_title}",
file=sys.stderr, file=sys.stderr,
@@ -285,7 +401,12 @@ def run_game_window(
flush=True, flush=True,
) )
if game_window is not None: if game_window is not None:
mapper.map_window(game_window, borderless=borderless) _prepare_game_window(
mapper,
game_window,
borderless=borderless,
suppress_wm_ping=suppress_wm_ping,
)
return int(process.wait()) return int(process.wait())
finally: finally:
mapper.close() mapper.close()
@@ -295,6 +416,7 @@ def main(argv: Sequence[str] | None = None) -> int:
parser = argparse.ArgumentParser() parser = argparse.ArgumentParser()
parser.add_argument("--window-title", default=WINDOW_TITLE) parser.add_argument("--window-title", default=WINDOW_TITLE)
parser.add_argument("--borderless", action="store_true") parser.add_argument("--borderless", action="store_true")
parser.add_argument("--suppress-wm-ping", action="store_true")
parser.add_argument("command", nargs=argparse.REMAINDER) parser.add_argument("command", nargs=argparse.REMAINDER)
arguments = parser.parse_args(argv) arguments = parser.parse_args(argv)
command = arguments.command command = arguments.command
@@ -305,6 +427,7 @@ def main(argv: Sequence[str] | None = None) -> int:
command, command,
arguments.window_title, arguments.window_title,
borderless=arguments.borderless, borderless=arguments.borderless,
suppress_wm_ping=arguments.suppress_wm_ping,
) )
except (GameboxRuntimeError, OSError) as exc: except (GameboxRuntimeError, OSError) as exc:
print(f"Error: {exc}", file=sys.stderr) print(f"Error: {exc}", file=sys.stderr)

View File

@@ -24,6 +24,8 @@ class PrimaryAction(str, Enum):
INSTALL = "INSTALL" INSTALL = "INSTALL"
UPDATE = "UPDATE" UPDATE = "UPDATE"
PLAY = "PLAY" PLAY = "PLAY"
CHECKING = "CHECKING…"
LAUNCHER_UPDATE = "UPDATE LAUNCHER"
@dataclass(frozen=True) @dataclass(frozen=True)
@@ -40,8 +42,10 @@ class GameInstallationStatus:
def inspect_game_installation( def inspect_game_installation(
paths: AppPaths, paths: AppPaths,
official_build: int | None = None, official_build: int | None = None,
*,
compatibility_ready: bool | None = True,
) -> GameInstallationStatus: ) -> GameInstallationStatus:
"""Select Install, Update, or Play without modifying the installation. """Select the primary launcher action without modifying the installation.
A completed native sync is the verified source for the local depot build. A completed native sync is the verified source for the local depot build.
When the official build cannot be checked, a complete local game remains When the official build cannot be checked, a complete local game remains
@@ -54,6 +58,8 @@ def inspect_game_installation(
or official_build < 0 or official_build < 0
): ):
raise ValueError("official build must be a non-negative integer") raise ValueError("official build must be a non-negative integer")
if compatibility_ready is not None and not isinstance(compatibility_ready, bool):
raise ValueError("compatibility readiness must be a boolean or null")
missing_files = tuple( missing_files = tuple(
relative relative
@@ -68,6 +74,10 @@ def inspect_game_installation(
local_build is None or local_build < official_build local_build is None or local_build < official_build
): ):
action = PrimaryAction.UPDATE action = PrimaryAction.UPDATE
elif compatibility_ready is None:
action = PrimaryAction.CHECKING
elif not compatibility_ready:
action = PrimaryAction.LAUNCHER_UPDATE
else: else:
action = PrimaryAction.PLAY action = PrimaryAction.PLAY
return GameInstallationStatus( return GameInstallationStatus(

View File

@@ -24,7 +24,7 @@ def launcher_operation_lock(lock_path: Path, operation: str) -> Iterator[None]:
handle.seek(0) handle.seek(0)
owner = handle.read().strip() or "unknown operation" owner = handle.read().strip() or "unknown operation"
raise OperationBusyError( raise OperationBusyError(
f"launcher files are busy ({owner}); close the updater and try again" f"launcher files are busy ({owner}); wait for it to finish and try again"
) from exc ) from exc
handle.seek(0) handle.seek(0)
handle.truncate() handle.truncate()

View File

@@ -0,0 +1,391 @@
"""Credential-free diagnostics for the PristonTale EU login path."""
from __future__ import annotations
from dataclasses import asdict, dataclass
from datetime import datetime
import hashlib
from pathlib import Path
import platform
import socket
import sys
import time
from typing import Callable, Protocol
from . import __version__
from .overlay import (
COMPATIBILITY_BINARY_HASHES,
CURRENT_V21_BINARY_HASHES,
PATCHED_V21_GAME_DLL_SHA256,
)
from .paths import AppPaths
from .runtime import (
RuntimeSelection,
detect_distribution_id,
native_d3d_ready,
resolve_proton_sync_mode,
select_runtime,
)
from .settings import Settings
from .world_fix import (
WORLD_FIX_DISABLED_REASON,
WORLD_FIX_ENABLED,
WORLD_FIX_HELPER_SHA256,
packaged_world_fix_helper,
)
LOGIN_SERVER_HOST = "5.135.35.188"
LOGIN_SERVER_PORT = 10019
DEFAULT_LOGIN_TIMEOUT_SECONDS = 5.0
VERIFIED_GAME_PROFILES = frozenset(
{"v21.75.83", "verified-compatibility"}
)
RETIRED_POST_AUTH_GAME_DLL_SHA256 = (
"93cd27284eb294753da1111c99bfc9689f82b6d4a59302aee9fdbd8cb52452dd"
)
class _Closable(Protocol):
def close(self) -> None: ...
LoginConnector = Callable[[tuple[str, int], float], _Closable]
def _create_login_connection(address: tuple[str, int], timeout: float) -> _Closable:
return socket.create_connection(address, timeout)
@dataclass(frozen=True)
class LoginServerProbe:
host: str
port: int
reachable: bool
latency_ms: float | None
error: str | None
@dataclass(frozen=True)
class LoginDiagnosticReport:
preflight_ok: bool
checked_at: str
launcher_version: str
launcher_module: str
python_executable: str
distribution_id: str
kernel_release: str
login_server: LoginServerProbe
authentication_tested: bool
install_directory: str
game_exe: str
game_exe_sha256: str | None
game_dll: str
game_dll_sha256: str | None
game_profile: str
world_fix_helper: str
world_fix_helper_sha256: str | None
world_fix_helper_ready: bool
world_fix_enabled: bool
world_fix_disabled_reason: str | None
runtime_kind: str | None
runtime_path: str | None
runtime_error: str | None
proton_sync_configured: str
proton_sync_effective: str | None
prefix: str | None
prefix_ready: bool
directx_ready: bool
latest_game_log: str | None
latest_game_log_modified_at: str | None
warnings: tuple[str, ...]
def to_dict(self) -> dict[str, object]:
return asdict(self)
def probe_login_server(
*,
host: str = LOGIN_SERVER_HOST,
port: int = LOGIN_SERVER_PORT,
timeout_seconds: float = DEFAULT_LOGIN_TIMEOUT_SECONDS,
connector: LoginConnector = _create_login_connection,
clock: Callable[[], float] = time.monotonic,
) -> LoginServerProbe:
"""Test only the TCP handshake; never send credentials or protocol data."""
if timeout_seconds <= 0:
raise ValueError("login diagnostic timeout must be greater than zero")
started = clock()
try:
connection = connector((host, port), timeout_seconds)
connection.close()
except OSError as exc:
return LoginServerProbe(
host=host,
port=port,
reachable=False,
latency_ms=None,
error=f"{type(exc).__name__}: {exc}",
)
return LoginServerProbe(
host=host,
port=port,
reachable=True,
latency_ms=round((clock() - started) * 1000.0, 1),
error=None,
)
def inspect_login_environment(
paths: AppPaths,
settings: Settings,
*,
timeout_seconds: float = DEFAULT_LOGIN_TIMEOUT_SECONDS,
connector: LoginConnector = _create_login_connection,
) -> LoginDiagnosticReport:
"""Inspect the installed client and its network route without authenticating."""
probe = probe_login_server(
timeout_seconds=timeout_seconds,
connector=connector,
)
selection, runtime_error = _runtime_selection(settings)
distribution_id = detect_distribution_id() or "unknown"
kernel_release = platform.release()
proton_sync_effective = (
resolve_proton_sync_mode(
settings.proton_sync,
distribution_id=distribution_id,
kernel_release=kernel_release,
)
if selection is not None
and selection.kind in {"proton7", "proton7-container"}
else None
)
game_root = _active_game_root(paths, settings)
game_exe = game_root / "Game.exe"
game_dll = game_root / "game.dll"
game_exe_sha256 = _sha256_if_file(game_exe)
game_dll_sha256 = _sha256_if_file(game_dll)
game_profile = classify_game_profile(game_exe_sha256, game_dll_sha256)
world_fix_helper = packaged_world_fix_helper()
world_fix_helper_sha256 = _sha256_if_file(world_fix_helper)
world_fix_helper_ready = (
WORLD_FIX_ENABLED
and world_fix_helper_sha256 == WORLD_FIX_HELPER_SHA256
)
prefix = _runtime_prefix(paths, selection)
prefix_ready = _prefix_ready(prefix)
directx_ready = native_d3d_ready(prefix) if prefix is not None else False
latest_log = latest_native_game_log(paths, settings)
warnings = _diagnostic_warnings(
probe,
game_profile,
runtime_error,
prefix_ready,
directx_ready,
latest_log,
)
return LoginDiagnosticReport(
preflight_ok=(
probe.reachable
and game_profile in VERIFIED_GAME_PROFILES
and selection is not None
and prefix_ready
and directx_ready
),
checked_at=datetime.now().astimezone().isoformat(),
launcher_version=__version__,
launcher_module=str(Path(__file__).resolve()),
python_executable=sys.executable,
distribution_id=distribution_id,
kernel_release=kernel_release,
login_server=probe,
authentication_tested=False,
install_directory=str(paths.data),
game_exe=str(game_exe),
game_exe_sha256=game_exe_sha256,
game_dll=str(game_dll),
game_dll_sha256=game_dll_sha256,
game_profile=game_profile,
world_fix_helper=str(world_fix_helper),
world_fix_helper_sha256=world_fix_helper_sha256,
world_fix_helper_ready=world_fix_helper_ready,
world_fix_enabled=WORLD_FIX_ENABLED,
world_fix_disabled_reason=(
None if WORLD_FIX_ENABLED else WORLD_FIX_DISABLED_REASON
),
runtime_kind=selection.kind if selection is not None else None,
runtime_path=_runtime_path(selection),
runtime_error=runtime_error,
proton_sync_configured=settings.proton_sync,
proton_sync_effective=proton_sync_effective,
prefix=str(prefix) if prefix is not None else None,
prefix_ready=prefix_ready,
directx_ready=directx_ready,
latest_game_log=str(latest_log) if latest_log is not None else None,
latest_game_log_modified_at=_modified_at(latest_log),
warnings=warnings,
)
def classify_game_profile(
game_exe_sha256: str | None,
game_dll_sha256: str | None,
) -> str:
if game_exe_sha256 is None or game_dll_sha256 is None:
return "missing"
pair = {"Game.exe": game_exe_sha256, "game.dll": game_dll_sha256}
if pair == CURRENT_V21_BINARY_HASHES:
return "v21.75.83"
if (
game_exe_sha256 == CURRENT_V21_BINARY_HASHES["Game.exe"]
and game_dll_sha256 == RETIRED_POST_AUTH_GAME_DLL_SHA256
):
return "retired-post-auth-disk-patch"
if (
game_exe_sha256 == CURRENT_V21_BINARY_HASHES["Game.exe"]
and game_dll_sha256 == PATCHED_V21_GAME_DLL_SHA256
):
return "retired-texture-reclaim"
if pair == COMPATIBILITY_BINARY_HASHES:
return "verified-compatibility"
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"],
RETIRED_POST_AUTH_GAME_DLL_SHA256,
PATCHED_V21_GAME_DLL_SHA256,
COMPATIBILITY_BINARY_HASHES["game.dll"],
}
if game_exe_sha256 in known_game_exe_hashes or game_dll_sha256 in known_game_dll_hashes:
return "mismatched"
return "unknown"
def latest_native_game_log(paths: AppPaths, settings: Settings) -> Path | None:
"""Return the newest native log from the active managed source or runtime view."""
directories = (paths.runtime_view / "save/logs", paths.source_game / "save/logs")
logs: dict[Path, Path] = {}
for directory in directories:
try:
candidates = directory.glob("log_*.txt")
for candidate in candidates:
if candidate.is_file():
logs[candidate.resolve()] = candidate
except OSError:
continue
if not logs:
return None
try:
return max(logs.values(), key=lambda item: item.stat().st_mtime_ns)
except OSError:
return None
def _active_game_root(paths: AppPaths, settings: Settings) -> Path:
if paths.runtime_view.is_dir():
return paths.runtime_view
return paths.source_game
def _runtime_selection(settings: Settings) -> tuple[RuntimeSelection | None, str | None]:
try:
return select_runtime(settings), None
except RuntimeError as exc:
return None, str(exc)
def _runtime_path(selection: RuntimeSelection | None) -> str | None:
if selection is None:
return None
return selection.proton_path or selection.executable
def _runtime_prefix(paths: AppPaths, selection: RuntimeSelection | None) -> Path | None:
if selection is None:
return None
return (
paths.proton_prefix
if selection.kind in {"proton7", "proton7-container"}
else paths.umu_prefix
)
def _prefix_ready(prefix: Path | None) -> bool:
if prefix is None:
return False
return (prefix / "user.reg").is_file() and (
prefix / "drive_c/windows/system32/cmd.exe"
).is_file()
def _diagnostic_warnings(
probe: LoginServerProbe,
game_profile: str,
runtime_error: str | None,
prefix_ready: bool,
directx_ready: bool,
latest_log: Path | None,
) -> tuple[str, ...]:
warnings: list[str] = []
if not probe.reachable:
warnings.append(
"The publisher login port is unreachable from this Linux host; check the "
"network, firewall, VPN, or server status."
)
if game_profile == "missing":
warnings.append("Game.exe or game.dll is missing; install or update the game first.")
elif game_profile == "mismatched":
warnings.append(
"Game.exe and game.dll are from different verified releases; run update --repair."
)
elif game_profile == "unknown":
warnings.append("Game.exe/game.dll do not match a launcher-verified client profile.")
elif game_profile == "retired-texture-reclaim":
warnings.append(
"A retired experimental game.dll patch is active and can stall login; "
"run prepare or launch the game once to restore the publisher DLL."
)
elif game_profile == "retired-post-auth-disk-patch":
warnings.append(
"The retired 0.5.8 disk-patched game.dll is active and can stall login; "
"run prepare or launch with the current launcher to restore the publisher DLL."
)
if runtime_error is not None:
warnings.append(runtime_error)
elif not prefix_ready:
warnings.append("The selected game prefix is not initialized; run play once to prepare it.")
elif not directx_ready:
warnings.append("The native DirectX helper DLLs are not ready; run play to repair them.")
if latest_log is None:
warnings.append("No native game log was found yet.")
return tuple(warnings)
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 _sha256_if_file(path: Path) -> str | None:
try:
return _sha256(path) if path.is_file() else None
except OSError:
return None
def _modified_at(path: Path | None) -> str | None:
if path is None:
return None
try:
return datetime.fromtimestamp(path.stat().st_mtime).astimezone().isoformat()
except OSError:
return None

472
ptlinux/login_trace.py Normal file
View File

@@ -0,0 +1,472 @@
"""Privacy-safe instrumentation for the post-authentication login transition."""
from __future__ import annotations
from dataclasses import asdict, dataclass
from datetime import datetime
import hashlib
import json
import os
from pathlib import Path
import platform
import re
import tempfile
from typing import Iterable
from .command import CommandOutcome, CommandRunner
from .login_diagnostics import latest_native_game_log
from .paths import AppPaths
from .runtime import (
RuntimeSelection,
detect_distribution_id,
resolve_proton_sync_mode,
run_game,
run_proton_prefix_command,
)
from .settings import Settings
RELAY_REGISTRY_KEY = r"HKCU\Software\Wine\Debug"
RELAY_REGISTRY_VALUE = "RelayInclude"
RELAY_INCLUDE_FILTER = "KERNEL32.lstrcmpA"
GAME_DLL_LOGIN_COMPARE_RETURN_OFFSET = 0x79F157
DEFAULT_TRACE_TIMEOUT_SECONDS = 10 * 60.0
_LOAD_GAME_DLL = re.compile(
r'Loaded L"[^"\r\n]*(?:\\|/)game\.dll" at (?P<base>[0-9a-f]+):',
re.IGNORECASE,
)
_COMPARE_RETURN = re.compile(
r'(?P<thread>[0-9a-f]{4}):Ret\s+KERNEL32\.lstrcmpA\(\) '
r'retval=(?P<result>[0-9a-f]+) ret=(?P<return_address>[0-9a-f]+)',
re.IGNORECASE,
)
_RELAY_VALUE = re.compile(
rf"^\s*{re.escape(RELAY_REGISTRY_VALUE)}\s+REG_SZ(?:\s+(?P<value>.*))?$",
re.MULTILINE,
)
_RELAY_CALL = re.compile(
r'^(?P<prefix>.*?Call\s+KERNEL32\.lstrcmpA)\(.*\) '
r'ret=(?P<return_address>[0-9a-f]+)\s*$',
re.IGNORECASE,
)
@dataclass(frozen=True)
class RelayTraceResult:
game_dll_base: int | None
target_return_address: int | None
comparison_results: tuple[bool, ...]
log_files: int
@dataclass(frozen=True)
class LoginTraceReport:
started_at: str
finished_at: str
result_directory: str
summary_file: str
report_file: str
distribution_id: str
kernel_release: str
runtime_kind: str
runtime_path: str | None
proton_sync_effective: str | None
game_exe_sha256: str | None
game_dll_sha256: str | None
native_log_before: str | None
native_log_before_modified_at: str | None
native_log_after: str | None
native_log_after_modified_at: str | None
game_returncode: int | None
game_timed_out: bool
game_duration_seconds: float
registry_restored: bool
raw_relay_deleted: bool
relay_log_files: int
game_dll_base: str | None
target_return_address: str | None
comparison_observed: bool
account_id_exact_match: bool | None
comparison_attempts: int
conclusion: str
warnings: tuple[str, ...]
def to_dict(self) -> dict[str, object]:
return asdict(self)
@dataclass(frozen=True)
class _RegistryValue:
existed: bool
value: str | None
def run_login_trace(
paths: AppPaths,
settings: Settings,
selection: RuntimeSelection,
runner: CommandRunner,
*,
use_gamebox: bool = True,
timeout_seconds: float = DEFAULT_TRACE_TIMEOUT_SECONDS,
) -> LoginTraceReport:
"""Launch one instrumented login attempt and retain only a redacted summary."""
if timeout_seconds <= 0:
raise ValueError("login trace timeout must be greater than zero")
if selection.kind not in {"proton7", "proton7-container"}:
raise RuntimeError("login tracing currently requires the pinned Proton 7 runtime")
if not selection.proton_path:
raise RuntimeError("the selected Proton 7 runtime path is incomplete")
started = datetime.now().astimezone()
result_directory = _create_result_directory(paths, started)
game_exe = paths.runtime_view / "Game.exe"
game_dll = paths.runtime_view / "game.dll"
log_before, log_before_modified = _native_log_snapshot(paths, settings)
registry_before = _query_relay_filter(paths, selection, runner)
registry_restored = False
trace = RelayTraceResult(None, None, (), 0)
outcome: CommandOutcome | None = None
raw_directory_path: Path | None = None
paths.cache.mkdir(parents=True, exist_ok=True)
with tempfile.TemporaryDirectory(
prefix="ptlinux-login-relay-",
dir=paths.cache,
) as raw_directory:
raw_directory_path = Path(raw_directory)
try:
_set_relay_filter(paths, selection, runner, RELAY_INCLUDE_FILTER)
outcome = run_game(
paths,
settings,
selection,
runner,
use_gamebox=use_gamebox,
use_world_fix=False,
timeout_seconds=timeout_seconds,
check=False,
extra_environment={
"PROTON_LOG": "1",
"PROTON_LOG_DIR": str(raw_directory_path),
"SteamAppId": "0",
"SteamGameId": "0",
"WINEDEBUG": "+relay,+loaddll",
},
output_line_filter=redact_relay_output_line,
)
trace = parse_relay_logs(raw_directory_path.rglob("*.log"))
finally:
registry_restored = _restore_relay_filter(
paths,
selection,
runner,
registry_before,
)
if outcome is None:
raise RuntimeError("the instrumented game process did not start")
raw_relay_deleted = raw_directory_path is not None and not raw_directory_path.exists()
log_after, log_after_modified = _native_log_snapshot(paths, settings)
comparison_observed = bool(trace.comparison_results)
account_id_exact_match = (
trace.comparison_results[-1] if comparison_observed else None
)
conclusion, warnings = _conclusion(
trace,
registry_restored=registry_restored,
raw_relay_deleted=raw_relay_deleted,
timed_out=outcome.timed_out,
)
finished = datetime.now().astimezone()
summary_file = result_directory / "summary.json"
report_file = result_directory / "report.md"
report = LoginTraceReport(
started_at=started.isoformat(),
finished_at=finished.isoformat(),
result_directory=str(result_directory),
summary_file=str(summary_file),
report_file=str(report_file),
distribution_id=detect_distribution_id() or "unknown",
kernel_release=platform.release(),
runtime_kind=selection.kind,
runtime_path=selection.proton_path,
proton_sync_effective=resolve_proton_sync_mode(settings.proton_sync),
game_exe_sha256=_sha256_if_file(game_exe),
game_dll_sha256=_sha256_if_file(game_dll),
native_log_before=log_before,
native_log_before_modified_at=log_before_modified,
native_log_after=log_after,
native_log_after_modified_at=log_after_modified,
game_returncode=outcome.returncode,
game_timed_out=outcome.timed_out,
game_duration_seconds=round(outcome.duration_seconds, 3),
registry_restored=registry_restored,
raw_relay_deleted=raw_relay_deleted,
relay_log_files=trace.log_files,
game_dll_base=_hex_or_none(trace.game_dll_base),
target_return_address=_hex_or_none(trace.target_return_address),
comparison_observed=comparison_observed,
account_id_exact_match=account_id_exact_match,
comparison_attempts=len(trace.comparison_results),
conclusion=conclusion,
warnings=warnings,
)
_write_report(report)
return report
def parse_relay_logs(log_paths: Iterable[Path]) -> RelayTraceResult:
"""Extract only module addresses and equality results from raw Proton logs."""
game_dll_base: int | None = None
returns: list[tuple[int, int]] = []
log_files = 0
for path in log_paths:
try:
handle = path.open("r", encoding="utf-8", errors="replace")
except OSError:
continue
log_files += 1
with handle:
for line in handle:
if game_dll_base is None:
loaded = _LOAD_GAME_DLL.search(line)
if loaded is not None:
game_dll_base = int(loaded.group("base"), 16)
returned = _COMPARE_RETURN.search(line)
if returned is not None:
returns.append(
(
int(returned.group("return_address"), 16),
int(returned.group("result"), 16),
)
)
target_return_address = (
game_dll_base + GAME_DLL_LOGIN_COMPARE_RETURN_OFFSET
if game_dll_base is not None
else None
)
comparison_results = tuple(
result == 0
for return_address, result in returns
if target_return_address is not None and return_address == target_return_address
)
return RelayTraceResult(
game_dll_base=game_dll_base,
target_return_address=target_return_address,
comparison_results=comparison_results,
log_files=log_files,
)
def redact_relay_output_line(line: str) -> str | None:
"""Remove strings from a relay call before console or command-log output."""
match = _RELAY_CALL.match(line.rstrip("\r\n"))
if match is None:
return line
return (
f"{match.group('prefix')}(<redacted>) "
f"ret={match.group('return_address')}\n"
)
def _query_relay_filter(
paths: AppPaths,
selection: RuntimeSelection,
runner: CommandRunner,
) -> _RegistryValue:
outcome = run_proton_prefix_command(
paths,
selection,
runner,
[
"reg.exe",
"query",
RELAY_REGISTRY_KEY,
"/v",
RELAY_REGISTRY_VALUE,
],
check=False,
)
match = _RELAY_VALUE.search(outcome.stdout)
if match is None:
return _RegistryValue(False, None)
return _RegistryValue(True, match.group("value") or "")
def _set_relay_filter(
paths: AppPaths,
selection: RuntimeSelection,
runner: CommandRunner,
value: str,
) -> None:
run_proton_prefix_command(
paths,
selection,
runner,
[
"reg.exe",
"add",
RELAY_REGISTRY_KEY,
"/v",
RELAY_REGISTRY_VALUE,
"/t",
"REG_SZ",
"/d",
value,
"/f",
],
)
def _restore_relay_filter(
paths: AppPaths,
selection: RuntimeSelection,
runner: CommandRunner,
previous: _RegistryValue,
) -> bool:
if previous.existed:
outcome = run_proton_prefix_command(
paths,
selection,
runner,
[
"reg.exe",
"add",
RELAY_REGISTRY_KEY,
"/v",
RELAY_REGISTRY_VALUE,
"/t",
"REG_SZ",
"/d",
previous.value or "",
"/f",
],
check=False,
)
else:
outcome = run_proton_prefix_command(
paths,
selection,
runner,
[
"reg.exe",
"delete",
RELAY_REGISTRY_KEY,
"/v",
RELAY_REGISTRY_VALUE,
"/f",
],
check=False,
)
return not outcome.timed_out and outcome.returncode == 0
def _create_result_directory(paths: AppPaths, started: datetime) -> Path:
root = paths.state / "login-traces"
root.mkdir(parents=True, exist_ok=True)
stem = f"{started.strftime('%Y%m%d-%H%M%S')}-{os.getpid()}"
result = root / stem
suffix = 1
while result.exists():
result = root / f"{stem}-{suffix}"
suffix += 1
result.mkdir(mode=0o700)
return result
def _native_log_snapshot(
paths: AppPaths,
settings: Settings,
) -> tuple[str | None, str | None]:
path = latest_native_game_log(paths, settings)
if path is None:
return None, None
try:
modified = datetime.fromtimestamp(path.stat().st_mtime).astimezone().isoformat()
except OSError:
modified = None
return str(path), modified
def _conclusion(
trace: RelayTraceResult,
*,
registry_restored: bool,
raw_relay_deleted: bool,
timed_out: bool,
) -> tuple[str, tuple[str, ...]]:
warnings: list[str] = []
if trace.comparison_results:
if trace.comparison_results[-1]:
conclusion = "account-id-match"
else:
conclusion = "account-id-mismatch"
warnings.append(
"The authenticated reply was discarded because its account ID did not "
"exactly match the ID entered in the client."
)
else:
conclusion = "comparison-not-observed"
warnings.append(
"The post-authentication account comparison was not observed; confirm that "
"Login was pressed before closing the game."
)
if trace.log_files == 0:
warnings.append("Proton did not create a relay log for this run.")
elif trace.game_dll_base is None:
warnings.append("The relay log did not record the loaded game.dll base address.")
if timed_out:
warnings.append("The diagnostic reached its timeout and terminated the launcher process.")
if not registry_restored:
warnings.append("The temporary Wine relay registry value could not be restored.")
if not raw_relay_deleted:
warnings.append("The private raw relay directory could not be removed.")
return conclusion, tuple(warnings)
def _write_report(report: LoginTraceReport) -> None:
summary_path = Path(report.summary_file)
report_path = Path(report.report_file)
summary_path.write_text(
json.dumps(report.to_dict(), indent=2, sort_keys=True) + "\n",
encoding="utf-8",
)
match = (
"not observed"
if report.account_id_exact_match is None
else "yes" if report.account_id_exact_match else "no"
)
report_path.write_text(
"# PristonTale EU login trace\n\n"
f"- Started: {report.started_at}\n"
f"- Runtime: {report.runtime_kind}\n"
f"- Account-ID comparison observed: {report.comparison_observed}\n"
f"- Exact account-ID match: {match}\n"
f"- Conclusion: `{report.conclusion}`\n"
f"- Registry restored: {report.registry_restored}\n"
f"- Raw relay data deleted: {report.raw_relay_deleted}\n\n"
"No account names, passwords, or packet payloads are retained in this report.\n",
encoding="utf-8",
)
def _sha256_if_file(path: Path) -> str | None:
try:
if not path.is_file():
return None
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()
except OSError:
return None
def _hex_or_none(value: int | None) -> str | None:
return f"0x{value:08x}" if value is not None else None

BIN
ptlinux/native/pt-world-fix.exe Executable file

Binary file not shown.

View File

@@ -19,7 +19,7 @@ from ptdiag.assbin import (
) )
from .paths import AppPaths from .paths import AppPaths
from .v21_texture_reclaim import TextureReclaimPatchError, build_texture_reclaim_patch from .world_fix import V21_BINARY_HASHES
WHISPERING_VALE_RELATIVE = Path("game/maps/WhisperingVale.rovug") WHISPERING_VALE_RELATIVE = Path("game/maps/WhisperingVale.rovug")
@@ -35,10 +35,7 @@ COMPATIBILITY_BINARY_HASHES = {
"Game.exe": "3cc0980aac0f65b379ff340ec61750498fb9f620fb6c88aa66281e8ab27023a5", "Game.exe": "3cc0980aac0f65b379ff340ec61750498fb9f620fb6c88aa66281e8ab27023a5",
"game.dll": "7ea85b19f452962a39c91f967cad22495f066b2d52e8ac37f4e234463d5c4f09", "game.dll": "7ea85b19f452962a39c91f967cad22495f066b2d52e8ac37f4e234463d5c4f09",
} }
CURRENT_V21_BINARY_HASHES = { CURRENT_V21_BINARY_HASHES = V21_BINARY_HASHES
"Game.exe": "2bc3edd26c5f6d7ae9bbdc23967c5aefd374bdf934fcdfa2ed42c6442e550837",
"game.dll": "4b4e5c28e2bfdaa8d36c8341fad27becfd03faf63aa8ea874f5d43c5b1961868",
}
PATCHED_V21_GAME_DLL_SHA256 = ( PATCHED_V21_GAME_DLL_SHA256 = (
"0be11b06ccf9be747e32b195c8d5980062f446d7d3a0a662ebfe0e602a01c538" "0be11b06ccf9be747e32b195c8d5980062f446d7d3a0a662ebfe0e602a01c538"
) )
@@ -72,16 +69,71 @@ class CompatibilityCacheResult:
@dataclass(frozen=True) @dataclass(frozen=True)
class RuntimePatchCacheResult: class RuntimeCompatibility:
source: str """Read-only compatibility result for the managed game currently on disk."""
destination: str
game_dll_sha256: str compatible: bool
created_at: str 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( def prepare_runtime_view(
paths: AppPaths, paths: AppPaths,
source_game: Path, source_game: Path,
*,
use_world_fix: bool = True,
) -> OverlayResult: ) -> OverlayResult:
"""Build a symlink view and keep the launcher-managed source read-only.""" """Build a symlink view and keep the launcher-managed source read-only."""
@@ -91,6 +143,10 @@ def prepare_runtime_view(
source_scene = source / WHISPERING_VALE_RELATIVE source_scene = source / WHISPERING_VALE_RELATIVE
inspection = inspect_wrapped_assbin(source_scene) inspection = inspect_wrapped_assbin(source_scene)
action = _scene_action(inspection) 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( binary_action, game_exe_source, game_dll_source = _runtime_game_binary_sources(
paths, paths,
source, source,
@@ -101,7 +157,7 @@ def prepare_runtime_view(
links_created = 0 links_created = 0
try: try:
build.mkdir(parents=True, exist_ok=False) build.mkdir(parents=True, exist_ok=False)
top_level_exceptions = {"game"} top_level_exceptions = {"game", "save"}
for name, binary_source in ( for name, binary_source in (
("Game.exe", game_exe_source), ("Game.exe", game_exe_source),
("game.dll", game_dll_source), ("game.dll", game_dll_source),
@@ -113,6 +169,8 @@ def prepare_runtime_view(
build, build,
except_names=top_level_exceptions, 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": if game_exe_source != source / "Game.exe":
(build / "Game.exe").symlink_to(game_exe_source) (build / "Game.exe").symlink_to(game_exe_source)
links_created += 1 links_created += 1
@@ -202,50 +260,17 @@ def cache_compatibility_binaries(
return result return result
def cache_runtime_patch(
paths: AppPaths,
source_dll: Path,
) -> RuntimePatchCacheResult:
source = source_dll.expanduser().resolve(strict=True)
if _sha256(source) != PATCHED_V21_GAME_DLL_SHA256:
raise OverlayError("runtime patch does not match the verified V21 profile")
target = paths.runtime_patch_dll
if target.exists() or target.is_symlink():
if target.is_file() and _sha256(target) == PATCHED_V21_GAME_DLL_SHA256:
return _runtime_patch_result(paths, source)
raise OverlayError(f"runtime patch cache has an unknown profile: {target}")
target.parent.mkdir(parents=True, exist_ok=True)
temporary = target.with_name(f".{target.name}.{uuid.uuid4().hex}.tmp")
try:
shutil.copy2(source, temporary)
if _sha256(temporary) != PATCHED_V21_GAME_DLL_SHA256:
raise OverlayError("copied runtime patch failed hash verification")
temporary.replace(target)
except BaseException:
temporary.unlink(missing_ok=True)
raise
result = _runtime_patch_result(paths, source)
_write_json(paths.runtime_patch_manifest, asdict(result))
return result
def select_binary_action( def select_binary_action(
cache_hashes: dict[str, str] | None, cache_hashes: dict[str, str] | None,
source_hashes: dict[str, str], source_hashes: dict[str, str],
runtime_patch_hash: str | None = None,
) -> str: ) -> str:
if runtime_patch_hash is not None:
if runtime_patch_hash != PATCHED_V21_GAME_DLL_SHA256:
raise OverlayError("runtime patch cache has an unknown binary profile")
if cache_hashes is not None: if cache_hashes is not None:
if cache_hashes != COMPATIBILITY_BINARY_HASHES: if cache_hashes != COMPATIBILITY_BINARY_HASHES:
raise OverlayError("compatibility cache has an unknown binary profile") raise OverlayError("compatibility cache has an unknown binary profile")
if source_hashes == COMPATIBILITY_BINARY_HASHES: if source_hashes == COMPATIBILITY_BINARY_HASHES:
return "verified-source-pair" return "verified-source-pair"
if runtime_patch_hash is not None: if source_hashes == CURRENT_V21_BINARY_HASHES:
if source_hashes != CURRENT_V21_BINARY_HASHES: return "verified-v21-publisher-pair"
raise OverlayError("V21 runtime patch does not match the installed client profile")
return "verified-v21-texture-reclaim"
return "upstream-pair" return "upstream-pair"
@@ -349,26 +374,82 @@ def _link_level(source: Path, target: Path, *, except_names: set[str]) -> int:
return count return count
def _runtime_game_binary_sources(paths: AppPaths, source: Path) -> tuple[str, Path, Path]: 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 = paths.compatibility_binaries
cache_hashes = _binary_hashes(cache) if cache.is_dir() else None cache_hashes = _binary_hashes(cache) if cache.is_dir() else None
source_hashes = _binary_hashes(source) source_hashes = _binary_hashes(source)
patch = paths.runtime_patch_dll action = select_binary_action(cache_hashes, source_hashes)
if source_hashes == CURRENT_V21_BINARY_HASHES and not patch.exists():
try:
result = build_texture_reclaim_patch(source / "game.dll", patch)
except TextureReclaimPatchError as exc:
raise OverlayError(f"could not build the verified V21 runtime patch: {exc}") from exc
_write_json(patch.parent / "patch.json", {"ok": True, **asdict(result)})
patch_hash = _sha256(patch) if patch.is_file() else None
action = select_binary_action(cache_hashes, source_hashes, patch_hash)
return ( return (
action, action,
source / "Game.exe", source / "Game.exe",
patch if action == "verified-v21-texture-reclaim" else source / "game.dll", 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]: def _binary_hashes(directory: Path) -> dict[str, str]:
return { return {
name: _sha256(directory / name) name: _sha256(directory / name)
@@ -387,15 +468,6 @@ def _compatibility_result(paths: AppPaths, source: Path) -> CompatibilityCacheRe
) )
def _runtime_patch_result(paths: AppPaths, source: Path) -> RuntimePatchCacheResult:
return RuntimePatchCacheResult(
source=str(source),
destination=str(paths.runtime_patch_dll),
game_dll_sha256=PATCHED_V21_GAME_DLL_SHA256,
created_at=datetime.now().astimezone().isoformat(),
)
def _remove_runtime_tree(path: Path) -> None: def _remove_runtime_tree(path: Path) -> None:
if path.is_symlink() or path.is_file(): if path.is_symlink() or path.is_file():
path.unlink(missing_ok=True) path.unlink(missing_ok=True)

View File

@@ -162,6 +162,16 @@ class AppPaths:
def runtime_manifest(self) -> Path: def runtime_manifest(self) -> Path:
return self.state / "runtime-view.json" return self.state / "runtime-view.json"
@property
def game_user_data(self) -> Path:
"""Writable game-created data kept outside the replaceable runtime view."""
return self.data / "user-data"
@property
def game_save(self) -> Path:
return self.game_user_data / "save"
@property @property
def compatibility_binaries(self) -> Path: def compatibility_binaries(self) -> Path:
return self.data / "compatibility-binaries" return self.data / "compatibility-binaries"
@@ -172,16 +182,26 @@ class AppPaths:
@property @property
def runtime_patch_directory(self) -> Path: def runtime_patch_directory(self) -> Path:
return self.data / "runtime-patches" / "v21-texture-reclaim" """Retired 0.5.8 disk-patch cache; never selected for play."""
return self.data / "runtime-patches" / "v21-post-auth-reclaim"
@property @property
def runtime_patch_dll(self) -> Path: def runtime_patch_dll(self) -> Path:
"""Retired 0.5.8 DLL retained only so diagnostics can identify it."""
return self.runtime_patch_directory / "game.dll" return self.runtime_patch_directory / "game.dll"
@property @property
def runtime_patch_manifest(self) -> Path: def runtime_patch_manifest(self) -> Path:
return self.state / "runtime-patch.json" return self.state / "runtime-patch.json"
@property
def retired_runtime_patch_dll(self) -> Path:
"""The pre-0.5.7 all-session patch, retained only for diagnostics."""
return self.data / "runtime-patches" / "v21-texture-reclaim" / "game.dll"
@property @property
def compat_data(self) -> Path: def compat_data(self) -> Path:
return self.data / "game-compat" return self.data / "game-compat"

View File

@@ -7,14 +7,20 @@ from datetime import datetime
import json import json
import os import os
from pathlib import Path from pathlib import Path
import platform
import shutil import shutil
import sys import sys
from typing import Mapping from typing import Mapping, Sequence
from .command import CommandOutcome, CommandRunner from .command import CommandOutcome, CommandRunner, OutputLineFilter
from .gamebox import build_gamebox_command from .gamebox import build_gamebox_command
from .paths import AppPaths from .paths import AppPaths
from .settings import Settings from .settings import Settings
from .world_fix import (
WorldFixError,
WorldFixStatusTarget,
resolve_world_fix_helper,
)
WINDOWS_GAME_EXE = r"C:\Program Files (x86)\Pristontale EU\Game.exe" WINDOWS_GAME_EXE = r"C:\Program Files (x86)\Pristontale EU\Game.exe"
@@ -25,6 +31,8 @@ WINETRICKS_CODEPAGE_PROBE = (
'winetricks_early_wine_arch cmd.exe /c "chcp 65001 > nul & echo %$1%"' 'winetricks_early_wine_arch cmd.exe /c "chcp 65001 > nul & echo %$1%"'
) )
WINETRICKS_PROTON7_PROBE = 'winetricks_early_wine_arch cmd.exe /c "echo %$1%"' WINETRICKS_PROTON7_PROBE = 'winetricks_early_wine_arch cmd.exe /c "echo %$1%"'
RUNTIME_DEFAULT_SYNC = "runtime-default"
CACHYOS_MARKER = "cachyos"
class RuntimeErrorPT(RuntimeError): class RuntimeErrorPT(RuntimeError):
@@ -44,21 +52,52 @@ def select_runtime(
environment: Mapping[str, str] | None = None, environment: Mapping[str, str] | None = None,
) -> RuntimeSelection: ) -> RuntimeSelection:
env = os.environ if environment is None else environment env = os.environ if environment is None else environment
umu = shutil.which("umu-run", path=env.get("PATH"))
if settings.proton_path: if settings.proton_path:
proton = Path(settings.proton_path).expanduser().resolve() proton = Path(settings.proton_path).expanduser().resolve()
if not proton.is_file(): if not proton.is_file():
raise RuntimeErrorPT(f"configured Proton script does not exist: {proton}") raise RuntimeErrorPT(f"configured Proton script does not exist: {proton}")
steam_root = _steam_root_for_proton(proton) steam_root = _steam_root_for_proton(proton)
if settings.runtime == "proton7-container":
if not umu:
raise RuntimeErrorPT(
"containerized Proton 7 was requested but umu-run was not found"
)
return RuntimeSelection(
"proton7-container",
umu,
str(proton),
str(steam_root),
)
if settings.runtime == "auto" and umu:
return RuntimeSelection(
"proton7-container",
umu,
str(proton),
str(steam_root),
)
return RuntimeSelection("proton7", sys.executable, str(proton), str(steam_root)) return RuntimeSelection("proton7", sys.executable, str(proton), str(steam_root))
if settings.runtime in {"auto", "proton7"}: if settings.runtime in {"auto", "proton7", "proton7-container"}:
proton_result = find_steam_proton7(env) proton_result = find_steam_proton7(env)
if proton_result is not None: if proton_result is not None:
proton, steam_root = proton_result proton, steam_root = proton_result
if settings.runtime == "proton7-container" or (
settings.runtime == "auto" and umu
):
if not umu:
raise RuntimeErrorPT(
"containerized Proton 7 was requested but umu-run was not found"
)
return RuntimeSelection(
"proton7-container",
umu,
str(proton),
str(steam_root),
)
return RuntimeSelection("proton7", sys.executable, str(proton), str(steam_root)) return RuntimeSelection("proton7", sys.executable, str(proton), str(steam_root))
if settings.runtime == "proton7": if settings.runtime in {"proton7", "proton7-container"}:
raise RuntimeErrorPT("Steam Proton 7.0 was requested but not found") raise RuntimeErrorPT("Steam Proton 7.0 was requested but not found")
if settings.runtime in {"auto", "umu"}: if settings.runtime in {"auto", "umu"}:
umu = shutil.which("umu-run")
if umu: if umu:
return RuntimeSelection("umu", umu, None, None) return RuntimeSelection("umu", umu, None, None)
raise RuntimeErrorPT( raise RuntimeErrorPT(
@@ -117,7 +156,7 @@ def prepare_game_runtime(
timeout_seconds=1800.0, timeout_seconds=1800.0,
) )
) )
elif selection.kind == "proton7": elif selection.kind in {"proton7", "proton7-container"}:
if not selection.proton_path or not selection.steam_root: if not selection.proton_path or not selection.steam_root:
raise RuntimeErrorPT("incomplete Proton 7 runtime selection") raise RuntimeErrorPT("incomplete Proton 7 runtime selection")
if not runner.dry_run: if not runner.dry_run:
@@ -216,16 +255,52 @@ def run_game(
runner: CommandRunner, runner: CommandRunner,
*, *,
use_gamebox: bool = True, use_gamebox: bool = True,
use_world_fix: bool = False,
world_fix_status_target: WorldFixStatusTarget | None = None,
timeout_seconds: float = 12 * 60 * 60,
check: bool = True,
extra_environment: Mapping[str, str] | None = None,
output_line_filter: OutputLineFilter | None = None,
) -> CommandOutcome: ) -> CommandOutcome:
if not paths.runtime_view.is_dir() and not runner.dry_run: if not paths.runtime_view.is_dir() and not runner.dry_run:
raise RuntimeErrorPT("runtime view is missing; prepare the game first") raise RuntimeErrorPT("runtime view is missing; prepare the game first")
world_fix_helper = None
if use_world_fix and not runner.dry_run:
try:
world_fix_helper = resolve_world_fix_helper(paths.runtime_view)
except WorldFixError as exc:
raise RuntimeErrorPT(str(exc)) from exc
launch_executable = world_fix_helper or (paths.runtime_view / "Game.exe")
if selection.kind == "umu": if selection.kind == "umu":
game_command = [ game_command = [
selection.executable, selection.executable,
str(paths.runtime_view / "Game.exe"), str(launch_executable),
"PTEU_REBORN",
] ]
if world_fix_helper is None:
game_command.append("PTEU_REBORN")
elif world_fix_status_target is not None:
game_command.extend(
["--status-file", world_fix_status_target.windows_path]
)
environment = _umu_environment(paths) environment = _umu_environment(paths)
elif selection.kind == "proton7-container":
if not selection.proton_path:
raise RuntimeErrorPT("Proton script path is missing")
game_command = [
selection.executable,
str(launch_executable),
]
if world_fix_helper is None:
game_command.append("PTEU_REBORN")
elif world_fix_status_target is not None:
game_command.extend(
["--status-file", world_fix_status_target.windows_path]
)
environment = _umu_proton7_environment(
paths,
selection,
synchronization=resolve_proton_sync_mode(settings.proton_sync),
)
else: else:
if not selection.proton_path: if not selection.proton_path:
raise RuntimeErrorPT("Proton script path is missing") raise RuntimeErrorPT("Proton script path is missing")
@@ -235,12 +310,23 @@ def run_game(
"ptlinux.proton_compat", "ptlinux.proton_compat",
selection.proton_path, selection.proton_path,
"runinprefix", "runinprefix",
WINDOWS_GAME_EXE, str(world_fix_helper) if world_fix_helper is not None else WINDOWS_GAME_EXE,
"PTEU_REBORN",
] ]
environment = _proton_environment(paths, selection) if world_fix_helper is None:
game_command.append("PTEU_REBORN")
elif world_fix_status_target is not None:
game_command.extend(
["--status-file", world_fix_status_target.windows_path]
)
environment = _proton_environment(
paths,
selection,
synchronization=resolve_proton_sync_mode(settings.proton_sync),
)
if use_gamebox: if use_gamebox:
game_command = build_gamebox_command(game_command, settings) game_command = build_gamebox_command(game_command, settings)
if extra_environment:
environment.update(extra_environment)
environment["PYTHONPATH"] = _runtime_python_path( environment["PYTHONPATH"] = _runtime_python_path(
os.environ.get("PYTHONPATH"), os.environ.get("PYTHONPATH"),
) )
@@ -248,8 +334,40 @@ def run_game(
game_command, game_command,
cwd=paths.runtime_view, cwd=paths.runtime_view,
env_overrides=environment, env_overrides=environment,
timeout_seconds=12 * 60 * 60, timeout_seconds=timeout_seconds,
check=check,
stream_output=True, stream_output=True,
output_line_filter=output_line_filter,
)
def run_proton_prefix_command(
paths: AppPaths,
selection: RuntimeSelection,
runner: CommandRunner,
arguments: Sequence[str],
*,
check: bool = True,
timeout_seconds: float = 60.0,
) -> CommandOutcome:
"""Run a maintenance command in the pinned Proton 7 prefix."""
if selection.kind not in {"proton7", "proton7-container"}:
raise RuntimeErrorPT("the login trace requires the pinned Proton 7 runtime")
if not selection.proton_path or not selection.steam_root:
raise RuntimeErrorPT("incomplete Proton 7 runtime selection")
return runner.run(
[
sys.executable,
"-m",
"ptlinux.proton_compat",
selection.proton_path,
"runinprefix",
*arguments,
],
env_overrides=_proton_environment(paths, selection),
timeout_seconds=timeout_seconds,
check=check,
) )
@@ -271,13 +389,42 @@ def _umu_environment(paths: AppPaths) -> dict[str, str]:
} }
def _umu_proton7_environment(
paths: AppPaths,
selection: RuntimeSelection,
*,
synchronization: str = RUNTIME_DEFAULT_SYNC,
) -> dict[str, str]:
"""Run the retained Proton 7 build inside its Steam Runtime container."""
if not selection.proton_path:
raise RuntimeErrorPT("Proton script path is missing")
environment = {
# UMU expects the Proton compatibility-data root and manages its pfx child.
# Pointing it at proton_prefix would create a nested pfx -> . compatibility
# symlink; compat_data preserves the exact prefix prepared by Proton 7.
"WINEPREFIX": str(paths.compat_data),
"GAMEID": "umu-pristontale-eu",
"STORE": "none",
"PROTONPATH": str(Path(selection.proton_path).parent),
"PROTON_VERB": "waitforexitandrun",
"PROTONFIXES_DISABLE": "1",
"DXVK_STATE_CACHE_PATH": str(paths.dxvk_cache),
"WINEDEBUG": "-all",
}
environment.update(_proton_sync_environment(synchronization))
return environment
def _proton_environment( def _proton_environment(
paths: AppPaths, paths: AppPaths,
selection: RuntimeSelection, selection: RuntimeSelection,
*,
synchronization: str = RUNTIME_DEFAULT_SYNC,
) -> dict[str, str]: ) -> dict[str, str]:
if not selection.steam_root: if not selection.steam_root:
raise RuntimeErrorPT("Steam root is missing for Proton") raise RuntimeErrorPT("Steam root is missing for Proton")
return { environment = {
"STEAM_COMPAT_CLIENT_INSTALL_PATH": selection.steam_root, "STEAM_COMPAT_CLIENT_INSTALL_PATH": selection.steam_root,
"STEAM_COMPAT_DATA_PATH": str(paths.compat_data), "STEAM_COMPAT_DATA_PATH": str(paths.compat_data),
"STEAM_COMPAT_INSTALL_PATH": str(paths.runtime_view), "STEAM_COMPAT_INSTALL_PATH": str(paths.runtime_view),
@@ -288,6 +435,57 @@ def _proton_environment(
"PROTON_LOG": "0", "PROTON_LOG": "0",
"DXVK_STATE_CACHE_PATH": str(paths.dxvk_cache), "DXVK_STATE_CACHE_PATH": str(paths.dxvk_cache),
} }
environment.update(_proton_sync_environment(synchronization))
return environment
def detect_distribution_id() -> str:
"""Return the freedesktop distribution ID without invoking distro tools."""
try:
value = platform.freedesktop_os_release().get("ID", "")
except OSError:
return ""
return value.strip().casefold()
def resolve_proton_sync_mode(
configured: str,
*,
distribution_id: str | None = None,
kernel_release: str | None = None,
) -> str:
"""Resolve the configured mode, applying the narrow CachyOS compatibility fallback."""
if configured not in {"auto", "fsync", "esync"}:
raise ValueError(f"unsupported Proton synchronization mode: {configured}")
if configured != "auto":
return configured
distribution = detect_distribution_id() if distribution_id is None else distribution_id
kernel = platform.release() if kernel_release is None else kernel_release
if CACHYOS_MARKER in distribution.casefold() or CACHYOS_MARKER in kernel.casefold():
return "esync"
return RUNTIME_DEFAULT_SYNC
def _proton_sync_environment(synchronization: str) -> dict[str, str]:
if synchronization == RUNTIME_DEFAULT_SYNC:
return {}
if synchronization == "esync":
return {
"PROTON_NO_FSYNC": "1",
"WINEFSYNC": "0",
"PROTON_NO_ESYNC": "0",
"WINEESYNC": "1",
}
if synchronization == "fsync":
return {
"PROTON_NO_FSYNC": "0",
"WINEFSYNC": "1",
"PROTON_NO_ESYNC": "0",
"WINEESYNC": "1",
}
raise ValueError(f"unsupported resolved Proton synchronization mode: {synchronization}")
def _ensure_proton_game_link(paths: AppPaths) -> None: def _ensure_proton_game_link(paths: AppPaths) -> None:

View File

@@ -639,6 +639,7 @@ def _validate_release_directory(release: Path) -> str:
Path("ptlinux") / PUBLIC_KEY_FILENAME, Path("ptlinux") / PUBLIC_KEY_FILENAME,
Path("ptdiag/__init__.py"), Path("ptdiag/__init__.py"),
Path("ptdiag/assbin.py"), Path("ptdiag/assbin.py"),
Path("ptdiag/logs.py"),
): ):
if not (release / required).is_file(): if not (release / required).is_file():
raise LauncherUpdateError(f"launcher release is missing {required}") raise LauncherUpdateError(f"launcher release is missing {required}")

View File

@@ -19,9 +19,11 @@ class Settings:
refresh_rate: int = 60 refresh_rate: int = 60
fullscreen: bool = False fullscreen: bool = False
borderless: bool = True borderless: bool = True
suppress_unresponsive_warnings: bool = True
scaler: str = "fit" scaler: str = "fit"
filter: str = "linear" filter: str = "linear"
runtime: str = "auto" runtime: str = "auto"
proton_sync: str = "auto"
proton_path: str | None = None proton_path: str | None = None
game_source: str | None = None game_source: str | None = None
@@ -47,9 +49,15 @@ def load_settings(paths: AppPaths) -> Settings:
refresh_rate=_integer(value, "refresh_rate", 60), refresh_rate=_integer(value, "refresh_rate", 60),
fullscreen=_boolean(value, "fullscreen", False), fullscreen=_boolean(value, "fullscreen", False),
borderless=_boolean(value, "borderless", True), borderless=_boolean(value, "borderless", True),
suppress_unresponsive_warnings=_boolean(
value,
"suppress_unresponsive_warnings",
True,
),
scaler=_string(value, "scaler", "fit"), scaler=_string(value, "scaler", "fit"),
filter=_string(value, "filter", "linear"), filter=_string(value, "filter", "linear"),
runtime=_string(value, "runtime", "auto"), runtime=_string(value, "runtime", "auto"),
proton_sync=_string(value, "proton_sync", "auto"),
proton_path=_optional_string(value, "proton_path"), proton_path=_optional_string(value, "proton_path"),
game_source=_optional_string(value, "game_source"), game_source=_optional_string(value, "game_source"),
) )
@@ -90,11 +98,13 @@ def _validate(settings: Settings) -> None:
if settings.refresh_rate < 24 or settings.refresh_rate > 1000: if settings.refresh_rate < 24 or settings.refresh_rate > 1000:
raise ValueError("refresh_rate must be between 24 and 1000") raise ValueError("refresh_rate must be between 24 and 1000")
if settings.scaler not in {"auto", "integer", "fit", "fill", "stretch"}: if settings.scaler not in {"auto", "integer", "fit", "fill", "stretch"}:
raise ValueError(f"unsupported Gamescope scaler: {settings.scaler}") raise ValueError(f"unsupported legacy scaler setting: {settings.scaler}")
if settings.filter not in {"linear", "nearest", "fsr", "nis", "pixel"}: if settings.filter not in {"linear", "nearest", "fsr", "nis", "pixel"}:
raise ValueError(f"unsupported Gamescope filter: {settings.filter}") raise ValueError(f"unsupported legacy filter setting: {settings.filter}")
if settings.runtime not in {"auto", "proton7", "umu"}: if settings.runtime not in {"auto", "proton7", "proton7-container", "umu"}:
raise ValueError(f"unsupported runtime selection: {settings.runtime}") raise ValueError(f"unsupported runtime selection: {settings.runtime}")
if settings.proton_sync not in {"auto", "fsync", "esync"}:
raise ValueError(f"unsupported Proton synchronization mode: {settings.proton_sync}")
def _integer(value: dict[str, object], key: str, default: int) -> int: def _integer(value: dict[str, object], key: str, default: int) -> int:

View File

@@ -29,6 +29,7 @@ from .news import (
fetch_news_image, fetch_news_image,
fetch_official_logo, fetch_official_logo,
) )
from .overlay import RuntimeCompatibility, inspect_runtime_compatibility
from .paths import AppPaths from .paths import AppPaths
from .settings import Settings, ensure_settings, save_settings from .settings import Settings, ensure_settings, save_settings
from .self_update import LauncherUpdateCheck, check_for_launcher_update from .self_update import LauncherUpdateCheck, check_for_launcher_update
@@ -568,9 +569,12 @@ def run_dashboard(paths: AppPaths) -> None:
operation_stage = 0 operation_stage = 0
operation_stage_total = 0 operation_stage_total = 0
operation_message = "" operation_message = ""
operation_refreshes_compatibility = False
last_elapsed_second = -1 last_elapsed_second = -1
official_snapshot: DepotSnapshot | None = None official_snapshot: DepotSnapshot | None = None
launcher_update_check: LauncherUpdateCheck | None = None launcher_update_check: LauncherUpdateCheck | None = None
runtime_compatibility: RuntimeCompatibility | None = None
compatibility_check_generation = 0
buttons = ( buttons = (
play_button, play_button,
update_button, update_button,
@@ -753,18 +757,21 @@ def run_dashboard(paths: AppPaths) -> None:
*, *,
show_preloader: bool = False, show_preloader: bool = False,
restart_on_success: bool = False, restart_on_success: bool = False,
game_files_may_change: bool = False,
) -> None: ) -> None:
nonlocal operation_active nonlocal operation_active
nonlocal operation_started_at nonlocal operation_started_at
nonlocal operation_stage nonlocal operation_stage
nonlocal operation_stage_total nonlocal operation_stage_total
nonlocal operation_message nonlocal operation_message
nonlocal operation_refreshes_compatibility
nonlocal last_elapsed_second nonlocal last_elapsed_second
operation_active = True operation_active = True
operation_started_at = time.monotonic() operation_started_at = time.monotonic()
operation_stage = 0 operation_stage = 0
operation_stage_total = 0 operation_stage_total = 0
operation_message = "" operation_message = ""
operation_refreshes_compatibility = game_files_may_change
last_elapsed_second = -1 last_elapsed_second = -1
progress_updates.clear() progress_updates.clear()
transfer_rate.reset() transfer_rate.reset()
@@ -1068,8 +1075,21 @@ def run_dashboard(paths: AppPaths) -> None:
installation = inspect_game_installation( installation = inspect_game_installation(
paths, paths,
official_snapshot.build if official_snapshot is not None else None, official_snapshot.build if official_snapshot is not None else None,
compatibility_ready=(
runtime_compatibility.compatible
if runtime_compatibility is not None
else None
),
) )
play_button.configure(text=installation.action.value) play_button.configure(text=installation.action.value)
if not operation_active:
play_button.configure(
state=(
"disabled"
if installation.action is PrimaryAction.CHECKING
else "normal"
),
)
if not update_activity or operation_active: if not update_activity or operation_active:
return installation return installation
if installation.action is PrimaryAction.INSTALL: if installation.action is PrimaryAction.INSTALL:
@@ -1089,11 +1109,42 @@ def run_dashboard(paths: AppPaths) -> None:
f"Installed depot {installation.local_build} · " f"Installed depot {installation.local_build} · "
f"official depot {official_build}" f"official depot {official_build}"
) )
elif installation.action is PrimaryAction.CHECKING:
status.set("Checking game compatibility…")
progress_text.set("The installed release is being inspected without modifying it.")
elif installation.action is PrimaryAction.LAUNCHER_UPDATE:
status.set("Launcher compatibility update required.")
progress_text.set("This installed game release is not recognized by the launcher.")
else: else:
status.set("Ready") status.set("Ready")
progress_text.set("Click Play to start your journey") progress_text.set("Click Play to start your journey")
return installation return installation
def refresh_runtime_compatibility(*, update_activity: bool = True) -> None:
nonlocal compatibility_check_generation, runtime_compatibility
compatibility_check_generation += 1
check_generation = compatibility_check_generation
runtime_compatibility = None
local_status = inspect_game_installation(paths)
refresh_primary_action(update_activity=update_activity)
if not local_status.installed:
return
def worker() -> None:
result = inspect_runtime_compatibility(paths, paths.source_game)
events.put(
(
"compatibility",
(check_generation, result, update_activity),
)
)
threading.Thread(
target=worker,
name="ptlinux-compatibility",
daemon=True,
).start()
def refresh_remote_content() -> None: def refresh_remote_content() -> None:
def broadcast_worker() -> None: def broadcast_worker() -> None:
events.put(("broadcast", fetch_latest_broadcast(paths.broadcast_cache))) events.put(("broadcast", fetch_latest_broadcast(paths.broadcast_cache)))
@@ -1127,8 +1178,10 @@ def run_dashboard(paths: AppPaths) -> None:
).start() ).start()
def finish_operation(returncode: int) -> None: def finish_operation(returncode: int) -> None:
nonlocal operation_active, operation_started_at nonlocal operation_active, operation_refreshes_compatibility, operation_started_at
refresh_compatibility = operation_refreshes_compatibility
operation_active = False operation_active = False
operation_refreshes_compatibility = False
operation_started_at = None operation_started_at = None
progress_updates.clear() progress_updates.clear()
transfer_rate.reset() transfer_rate.reset()
@@ -1138,13 +1191,16 @@ def run_dashboard(paths: AppPaths) -> None:
root.deiconify() root.deiconify()
root.lift() root.lift()
set_busy(False) set_busy(False)
refresh_primary_action(update_activity=returncode == 0) if refresh_compatibility:
refresh_runtime_compatibility(update_activity=returncode == 0)
else:
refresh_primary_action(update_activity=returncode == 0)
if returncode != 0: if returncode != 0:
status.set(f"Operation failed (exit {returncode})") status.set(f"Operation failed (exit {returncode})")
progress_text.set("See the activity log for details.") progress_text.set("See the activity log for details.")
def poll_events() -> None: def poll_events() -> None:
nonlocal launcher_update_check nonlocal launcher_update_check, runtime_compatibility
processed = 0 processed = 0
try: try:
while processed < EVENT_BATCH_SIZE: while processed < EVENT_BATCH_SIZE:
@@ -1192,6 +1248,29 @@ def run_dashboard(paths: AppPaths) -> None:
install_news_image(value) install_news_image(value)
elif kind == "logo_image": elif kind == "logo_image":
install_official_logo(value) install_official_logo(value)
elif (
kind == "compatibility"
and isinstance(value, tuple)
and len(value) == 3
and isinstance(value[0], int)
and isinstance(value[1], RuntimeCompatibility)
and isinstance(value[2], bool)
):
check_generation, checked, update_activity = value
if check_generation == compatibility_check_generation:
runtime_compatibility = checked
if checked.compatible:
append_log(
"Compatibility check passed: "
f"scene={checked.scene_action}, "
f"binaries={checked.binary_action}"
)
else:
append_log(
"Launcher compatibility update required: "
f"{checked.reason or 'unknown release profile'}"
)
refresh_primary_action(update_activity=update_activity)
elif kind == "version" and isinstance(value, DepotSnapshot): elif kind == "version" and isinstance(value, DepotSnapshot):
version.set( version.set(
f"Launcher {__version__} · {value.version} · depot {value.build}" f"Launcher {__version__} · {value.version} · depot {value.build}"
@@ -1241,18 +1320,29 @@ def run_dashboard(paths: AppPaths) -> None:
def command_for(*arguments: str) -> list[str]: def command_for(*arguments: str) -> list[str]:
return [sys.executable, "-m", "ptlinux", *arguments] return [sys.executable, "-m", "ptlinux", *arguments]
def run_launcher_update() -> None: def run_launcher_update(required_reason: str | None = None) -> None:
check = launcher_update_check check = launcher_update_check
if check is None: if check is None:
launcher_update_button.configure(text="CHECKING…", state="disabled") launcher_update_button.configure(text="CHECKING…", state="disabled")
if required_reason is not None:
status.set("Checking for a compatible launcher update…")
progress_text.set("Play remains disabled until this release is supported.")
refresh_remote_content() refresh_remote_content()
return return
if not check.available: if not check.available:
messagebox.showinfo( if required_reason is None:
"Launcher update", messagebox.showinfo(
check.reason, "Launcher update",
parent=root, check.reason,
) parent=root,
)
else:
messagebox.showwarning(
"Compatibility update required",
"The installed game needs a launcher compatibility update:\n\n"
f"{required_reason}\n\n{check.reason}",
parent=root,
)
return return
if not check.supported: if not check.supported:
messagebox.showwarning( messagebox.showwarning(
@@ -1287,11 +1377,19 @@ def run_dashboard(paths: AppPaths) -> None:
run_operation( run_operation(
"Installing official game files…", "Installing official game files…",
command_for("update"), command_for("update"),
game_files_may_change=True,
) )
else: elif installation.action is PrimaryAction.UPDATE:
run_operation( run_operation(
"Synchronizing official game files…", "Synchronizing official game files…",
command_for("update"), command_for("update"),
game_files_may_change=True,
)
elif installation.action is PrimaryAction.LAUNCHER_UPDATE:
run_launcher_update(
runtime_compatibility.reason
if runtime_compatibility is not None
else "The installed release profile is not recognized.",
) )
play_button.configure(command=run_primary_action) play_button.configure(command=run_primary_action)
@@ -1299,12 +1397,14 @@ def run_dashboard(paths: AppPaths) -> None:
command=lambda: run_operation( command=lambda: run_operation(
"Synchronizing official game files…", "Synchronizing official game files…",
command_for("update"), command_for("update"),
game_files_may_change=True,
) )
) )
repair_button.configure( repair_button.configure(
command=lambda: run_operation( command=lambda: run_operation(
"Checking every game file against the official depot…", "Checking every game file against the official depot…",
command_for("update", "--no-import", "--repair"), command_for("update", "--no-import", "--repair"),
game_files_may_change=True,
) )
) )
doctor_button.configure( doctor_button.configure(
@@ -1350,7 +1450,7 @@ def run_dashboard(paths: AppPaths) -> None:
).pack(anchor="w", pady=(1, 0)) ).pack(anchor="w", pady=(1, 0))
tk.Label( tk.Label(
header_copy, header_copy,
text="Tune your display, rendering, and compatibility preferences.", text="Tune the game window, runtime, and compatibility preferences.",
background="#24170f", background="#24170f",
foreground="#c9baaa", foreground="#c9baaa",
font=("Sans", 9), font=("Sans", 9),
@@ -1361,54 +1461,15 @@ def run_dashboard(paths: AppPaths) -> None:
content.columnconfigure(0, weight=1, uniform="settings-card") content.columnconfigure(0, weight=1, uniform="settings-card")
content.columnconfigure(1, weight=1, uniform="settings-card") content.columnconfigure(1, weight=1, uniform="settings-card")
width_value = tk.StringVar(value=str(current_settings.width))
height_value = tk.StringVar(value=str(current_settings.height))
output_width_value = tk.StringVar(value=str(current_settings.output_width))
output_height_value = tk.StringVar(value=str(current_settings.output_height))
refresh_rate_value = tk.StringVar(value=str(current_settings.refresh_rate))
fullscreen_value = tk.BooleanVar(value=current_settings.fullscreen)
borderless_value = tk.BooleanVar(value=current_settings.borderless) borderless_value = tk.BooleanVar(value=current_settings.borderless)
scaler_value = tk.StringVar(value=current_settings.scaler) suppress_unresponsive_value = tk.BooleanVar(
filter_value = tk.StringVar(value=current_settings.filter) value=current_settings.suppress_unresponsive_warnings
)
runtime_value = tk.StringVar(value=current_settings.runtime) runtime_value = tk.StringVar(value=current_settings.runtime)
proton_sync_value = tk.StringVar(value=current_settings.proton_sync)
proton_path_value = tk.StringVar(value=current_settings.proton_path or "") proton_path_value = tk.StringVar(value=current_settings.proton_path or "")
game_source_value = tk.StringVar(value=current_settings.game_source or "") game_source_value = tk.StringVar(value=current_settings.game_source or "")
def add_dimensions(
parent: ttk.LabelFrame,
row: int,
label: str,
description: str,
width_variable: tk.StringVar,
height_variable: tk.StringVar,
) -> None:
ttk.Label(parent, text=label, style="SettingsText.TLabel").grid(
row=row,
column=0,
sticky="w",
)
ttk.Label(parent, text=description, style="SettingsHint.TLabel").grid(
row=row,
column=1,
sticky="e",
)
fields = ttk.Frame(parent, style="SettingsCard.TFrame")
fields.grid(row=row + 1, column=0, columnspan=2, sticky="ew", pady=(3, 9))
ttk.Label(fields, text="Width", style="SettingsHint.TLabel").pack(side="left")
ttk.Entry(
fields,
textvariable=width_variable,
width=8,
style="Settings.TEntry",
).pack(side="left", padx=(5, 12))
ttk.Label(fields, text="Height", style="SettingsHint.TLabel").pack(side="left")
ttk.Entry(
fields,
textvariable=height_variable,
width=8,
style="Settings.TEntry",
).pack(side="left", padx=(5, 0))
def add_choice( def add_choice(
parent: ttk.LabelFrame, parent: ttk.LabelFrame,
row: int, row: int,
@@ -1491,26 +1552,21 @@ def run_dashboard(paths: AppPaths) -> None:
settings_window.destroy() settings_window.destroy()
settings_window = None settings_window = None
def integer_value(variable: tk.StringVar, name: str) -> int:
value = variable.get().strip()
try:
return int(value)
except ValueError as error:
raise ValueError(f"{name} must be a whole number") from error
def save_current_settings() -> None: def save_current_settings() -> None:
try: try:
updated_settings = Settings( updated_settings = Settings(
width=integer_value(width_value, "Client width"), width=current_settings.width,
height=integer_value(height_value, "Client height"), height=current_settings.height,
output_width=integer_value(output_width_value, "Output width"), output_width=current_settings.output_width,
output_height=integer_value(output_height_value, "Output height"), output_height=current_settings.output_height,
refresh_rate=integer_value(refresh_rate_value, "Refresh rate"), refresh_rate=current_settings.refresh_rate,
fullscreen=fullscreen_value.get(), fullscreen=current_settings.fullscreen,
borderless=borderless_value.get(), borderless=borderless_value.get(),
scaler=scaler_value.get(), suppress_unresponsive_warnings=suppress_unresponsive_value.get(),
filter=filter_value.get(), scaler=current_settings.scaler,
filter=current_settings.filter,
runtime=runtime_value.get(), runtime=runtime_value.get(),
proton_sync=proton_sync_value.get(),
proton_path=proton_path_value.get().strip() or None, proton_path=proton_path_value.get().strip() or None,
game_source=game_source_value.get().strip() or None, game_source=game_source_value.get().strip() or None,
) )
@@ -1525,7 +1581,7 @@ def run_dashboard(paths: AppPaths) -> None:
display_card = ttk.LabelFrame( display_card = ttk.LabelFrame(
content, content,
text="DISPLAY", text="GAME WINDOW",
style="SettingsCard.TLabelframe", style="SettingsCard.TLabelframe",
padding=14, padding=14,
) )
@@ -1534,66 +1590,31 @@ def run_dashboard(paths: AppPaths) -> None:
display_card.columnconfigure(1, weight=1) display_card.columnconfigure(1, weight=1)
ttk.Label( ttk.Label(
display_card, display_card,
text="Choose the game canvas and the Gamescope output it is presented on.", text=(
"The launcher uses the proven client resolution and can remove the "
"new game window's host decorations."
),
style="SettingsHint.TLabel", style="SettingsHint.TLabel",
wraplength=360, wraplength=360,
).grid(row=0, column=0, columnspan=2, sticky="w", pady=(0, 11)) ).grid(row=0, column=0, columnspan=2, sticky="w", pady=(0, 11))
add_dimensions(
display_card,
1,
"Game resolution",
"Internal client size",
width_value,
height_value,
)
add_dimensions(
display_card,
3,
"Output resolution",
"Gamescope desktop",
output_width_value,
output_height_value,
)
ttk.Label(display_card, text="Refresh rate", style="SettingsText.TLabel").grid(
row=5,
column=0,
sticky="w",
)
ttk.Label(
display_card,
text="Output refresh",
style="SettingsHint.TLabel",
).grid(row=5, column=1, sticky="e")
refresh_fields = ttk.Frame(display_card, style="SettingsCard.TFrame")
refresh_fields.grid(row=6, column=0, columnspan=2, sticky="w", pady=(3, 8))
ttk.Entry(
refresh_fields,
textvariable=refresh_rate_value,
width=8,
style="Settings.TEntry",
).pack(side="left")
ttk.Label(refresh_fields, text="Hz", style="SettingsHint.TLabel").pack(
side="left",
padx=(5, 0),
)
presentation = ttk.Frame(display_card, style="SettingsCard.TFrame") presentation = ttk.Frame(display_card, style="SettingsCard.TFrame")
presentation.grid(row=7, column=0, columnspan=2, sticky="w") presentation.grid(row=1, column=0, columnspan=2, sticky="w")
ttk.Checkbutton(
presentation,
text="Fullscreen",
variable=fullscreen_value,
style="Settings.TCheckbutton",
).pack(side="left")
ttk.Checkbutton( ttk.Checkbutton(
presentation, presentation,
text="Borderless", text="Borderless",
variable=borderless_value, variable=borderless_value,
style="Settings.TCheckbutton", style="Settings.TCheckbutton",
).pack(side="left", padx=(18, 0)) ).pack(anchor="w")
ttk.Checkbutton(
presentation,
text="Suppress false 'not responding' warnings",
variable=suppress_unresponsive_value,
style="Settings.TCheckbutton",
).pack(anchor="w", pady=(7, 0))
rendering_card = ttk.LabelFrame( rendering_card = ttk.LabelFrame(
content, content,
text="RENDERING & RUNTIME", text="RUNTIME",
style="SettingsCard.TLabelframe", style="SettingsCard.TLabelframe",
padding=14, padding=14,
) )
@@ -1602,40 +1623,25 @@ def run_dashboard(paths: AppPaths) -> None:
rendering_card.columnconfigure(1, weight=1) rendering_card.columnconfigure(1, weight=1)
ttk.Label( ttk.Label(
rendering_card, rendering_card,
text="Control how the client is scaled and which compatible runtime launches it.", text="Choose which compatibility runtime launches the proven game setup.",
style="SettingsHint.TLabel", style="SettingsHint.TLabel",
wraplength=360, wraplength=360,
).grid(row=0, column=0, columnspan=2, sticky="w", pady=(0, 11)) ).grid(row=0, column=0, columnspan=2, sticky="w", pady=(0, 11))
add_choice( add_choice(
rendering_card, rendering_card,
1, 1,
"Scaling mode", "Runtime",
"Gamescope layout", "Compatibility layer",
scaler_value, runtime_value,
("auto", "integer", "fit", "fill", "stretch"), ("auto", "proton7-container", "proton7", "umu"),
) )
add_choice( add_choice(
rendering_card, rendering_card,
3, 3,
"Image filter", "Proton synchronization",
"Upscaling filter", "Auto uses esync on CachyOS",
filter_value, proton_sync_value,
("linear", "nearest", "fsr", "nis", "pixel"), ("auto", "fsync", "esync"),
)
ttk.Separator(rendering_card).grid(
row=5,
column=0,
columnspan=2,
sticky="ew",
pady=(2, 9),
)
add_choice(
rendering_card,
6,
"Runtime",
"Compatibility layer",
runtime_value,
("auto", "proton7", "umu"),
) )
overrides_card = ttk.LabelFrame( overrides_card = ttk.LabelFrame(
@@ -1662,8 +1668,8 @@ def run_dashboard(paths: AppPaths) -> None:
add_path( add_path(
overrides_card, overrides_card,
3, 3,
"Game source", "Import source",
"Import from this existing game folder", "Seed a new managed installation only",
game_source_value, game_source_value,
choose_game_source, choose_game_source,
) )
@@ -1673,7 +1679,7 @@ def run_dashboard(paths: AppPaths) -> None:
footer.columnconfigure(0, weight=1) footer.columnconfigure(0, weight=1)
ttk.Label( ttk.Label(
footer, footer,
text="Settings are stored locally and used the next time you press Play.", text="Only the settings shown here affect the next launcher operation.",
style="SettingsFooter.TLabel", style="SettingsFooter.TLabel",
).grid(row=0, column=0, sticky="w") ).grid(row=0, column=0, sticky="w")
buttons_row = ttk.Frame(footer, style="SettingsFooter.TFrame") buttons_row = ttk.Frame(footer, style="SettingsFooter.TFrame")
@@ -1707,6 +1713,7 @@ def run_dashboard(paths: AppPaths) -> None:
run_operation( run_operation(
"Importing the existing game into the managed installation…", "Importing the existing game into the managed installation…",
command_for("import-game", source), command_for("import-game", source),
game_files_may_change=True,
) )
import_button.configure(command=choose_import_source) import_button.configure(command=choose_import_source)
@@ -1723,6 +1730,7 @@ def run_dashboard(paths: AppPaths) -> None:
news_content.bind("<Configure>", update_news_viewport) news_content.bind("<Configure>", update_news_viewport)
news_canvas.bind("<Configure>", resize_news_content) news_canvas.bind("<Configure>", resize_news_content)
refresh_primary_action() refresh_primary_action()
refresh_runtime_compatibility()
poll_events() poll_events()
load_official_logo() load_official_logo()
refresh_remote_content() refresh_remote_content()

View File

@@ -1,156 +0,0 @@
"""Build the verified V21 texture-reclaim runtime DLL from a local game DLL."""
from __future__ import annotations
from dataclasses import dataclass
from datetime import datetime
import hashlib
from pathlib import Path
import shutil
import struct
SOURCE_SHA256 = "4b4e5c28e2bfdaa8d36c8341fad27becfd03faf63aa8ea874f5d43c5b1961868"
PATCHED_SHA256 = "0be11b06ccf9be747e32b195c8d5980062f446d7d3a0a662ebfe0e602a01c538"
TEXT_VA = 0x10001000
TEXT_FILE_OFFSET = 0x400
ENTRY_VA = 0x107C0B40
ENTRY_FILE_OFFSET = TEXT_FILE_OFFSET + (ENTRY_VA - TEXT_VA)
ENTRY_ORIGINAL = bytes.fromhex("55 8b ec 6a ff")
CAVE_VA = 0x11195CE0
CAVE_FILE_OFFSET = TEXT_FILE_OFFSET + (CAVE_VA - TEXT_VA)
TEXT_RAW_SIZE = 0x1194E00
MAX_HOOK_SIZE = TEXT_FILE_OFFSET + TEXT_RAW_SIZE - CAVE_FILE_OFFSET
HOOK = bytes.fromhex(
"9c60e8490136ff85c0744a89c38d70088dbe505d010039fe743b80be58030000"
"00742a8b068b4c242885c974083b81d41f000074188b4c242c85c974083b81d4"
"1f000074085089d9e8a32036ff81c650040000ebc1619d5589e56affe904ae62"
"ff"
)
class TextureReclaimPatchError(RuntimeError):
"""Raised when the exact V21 binary profile cannot be patched safely."""
@dataclass(frozen=True)
class TextureReclaimPatchResult:
source: str
destination: str
source_sha256: str
destination_sha256: str
entry_va: str
cave_va: str
hook_size: int
old_text_virtual_size: int
new_text_virtual_size: int
commands: tuple[tuple[str, ...], ...]
created_at: str
def build_texture_reclaim_patch(
source: Path,
destination: Path,
) -> TextureReclaimPatchResult:
"""Create the V21 patch in a separate runtime cache, never in the game tree."""
source = source.expanduser().resolve(strict=True)
if _sha256(source) != SOURCE_SHA256:
raise TextureReclaimPatchError(
"source game.dll does not match the verified V21 profile"
)
hook = HOOK
commands: list[tuple[str, ...]] = []
if not hook or len(hook) > MAX_HOOK_SIZE:
raise TextureReclaimPatchError(
f"compiled hook size {len(hook)} exceeds the {MAX_HOOK_SIZE}-byte code cave"
)
image = bytearray(source.read_bytes())
if bytes(image[ENTRY_FILE_OFFSET : ENTRY_FILE_OFFSET + len(ENTRY_ORIGINAL)]) != ENTRY_ORIGINAL:
raise TextureReclaimPatchError(
"V21 map-loader entry bytes do not match the verified profile"
)
if any(image[CAVE_FILE_OFFSET : CAVE_FILE_OFFSET + len(hook)]):
raise TextureReclaimPatchError("V21 code cave is not empty")
text_header = _find_pe_section(image, b".text")
old_virtual_size, text_rva, raw_size, raw_offset = struct.unpack_from(
"<IIII", image, text_header + 8
)
if (text_rva, raw_size, raw_offset) != (
TEXT_VA - 0x10000000,
TEXT_RAW_SIZE,
TEXT_FILE_OFFSET,
):
raise TextureReclaimPatchError(
"V21 executable section layout does not match the verified profile"
)
new_virtual_size = CAVE_VA - TEXT_VA + len(hook)
if new_virtual_size > TEXT_RAW_SIZE:
raise TextureReclaimPatchError("compiled hook extends beyond the executable section")
if old_virtual_size > new_virtual_size:
raise TextureReclaimPatchError(
"verified executable section is already larger than the patch target"
)
entry_displacement = CAVE_VA - (ENTRY_VA + 5)
image[ENTRY_FILE_OFFSET : ENTRY_FILE_OFFSET + 5] = b"\xe9" + struct.pack(
"<i", entry_displacement
)
image[CAVE_FILE_OFFSET : CAVE_FILE_OFFSET + len(hook)] = hook
struct.pack_into("<I", image, text_header + 8, new_virtual_size)
destination = destination.expanduser().absolute()
destination.parent.mkdir(parents=True, exist_ok=True)
temporary = destination.with_name(f".{destination.name}.tmp")
try:
temporary.write_bytes(image)
if _sha256(temporary) != PATCHED_SHA256:
raise TextureReclaimPatchError(
"built runtime patch does not match the verified V21 profile"
)
temporary.replace(destination)
shutil.copystat(source, destination)
except BaseException:
temporary.unlink(missing_ok=True)
raise
return TextureReclaimPatchResult(
source=str(source),
destination=str(destination),
source_sha256=SOURCE_SHA256,
destination_sha256=PATCHED_SHA256,
entry_va=f"0x{ENTRY_VA:08x}",
cave_va=f"0x{CAVE_VA:08x}",
hook_size=len(hook),
old_text_virtual_size=old_virtual_size,
new_text_virtual_size=new_virtual_size,
commands=tuple(commands),
created_at=datetime.now().astimezone().isoformat(),
)
def _find_pe_section(image: bytes | bytearray, section_name: bytes) -> int:
if len(image) < 0x40:
raise TextureReclaimPatchError("source is too small to be a PE image")
pe_offset = struct.unpack_from("<I", image, 0x3C)[0]
if image[pe_offset : pe_offset + 4] != b"PE\0\0":
raise TextureReclaimPatchError("source does not have a valid PE signature")
number_of_sections = struct.unpack_from("<H", image, pe_offset + 6)[0]
optional_header_size = struct.unpack_from("<H", image, pe_offset + 20)[0]
section_table = pe_offset + 24 + optional_header_size
for index in range(number_of_sections):
header = section_table + index * 40
name = bytes(image[header : header + 8]).rstrip(b"\0")
if name == section_name:
return header
raise TextureReclaimPatchError(f"PE section is missing: {section_name!r}")
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()

191
ptlinux/world_fix.py Normal file
View File

@@ -0,0 +1,191 @@
"""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