16 Commits

34 changed files with 3986 additions and 2287 deletions

View File

@@ -1,17 +1,25 @@
# PristonTale EU for Linux
Version 0.5.2 installs, updates, repairs, and launches PristonTale EU through a
Version 0.7.0 installs, updates, repairs, and launches PristonTale EU through a
native Linux dashboard. Game files come directly from the publisher and are not
included in this archive.
## Install
Keep the archive and checksum file together, then run:
Keep the archive, detached signature, checksum, and
`launcher-update-public.pem` together. First confirm the public-key fingerprint
through an independent trusted channel; the expected DER SHA-256 fingerprint is
`3ae6e8dc483b93796b32e3ca270c367d247caf8d047eb796826c8e558a4a9e34`.
Then run:
```text
sha256sum -c pristontale-eu-linux-installer-v0-5-2.tar.gz.sha256
tar -xzf pristontale-eu-linux-installer-v0-5-2.tar.gz
cd pristontale-eu-linux-installer-v0-5-2
openssl pkey -pubin -in launcher-update-public.pem -outform DER | sha256sum
openssl dgst -sha256 -verify launcher-update-public.pem \
-signature pristontale-eu-linux-installer-v0-7-0.tar.gz.sig \
pristontale-eu-linux-installer-v0-7-0.tar.gz
sha256sum -c pristontale-eu-linux-installer-v0-7-0.tar.gz.sha256
tar -xzf pristontale-eu-linux-installer-v0-7-0.tar.gz
cd pristontale-eu-linux-installer-v0-7-0
./install.sh
```
@@ -37,11 +45,6 @@ pristontale-eu gui
The main launcher button automatically shows **Install**, **Update**, or
**Play** according to the managed game's state.
Version 0.5.0 is the self-update bootstrap. Future signed launcher releases are
checked in the background and appear as an **Update Launcher** action. Launcher
updates are staged separately from game files and retain the previous version
for rollback.
## Commands
```text
@@ -49,16 +52,22 @@ pristontale-eu --help Show all commands and options
pristontale-eu gui Open the native dashboard
pristontale-eu doctor Check required host components
pristontale-eu login-diagnostics Check the login route without credentials
pristontale-eu login-diagnostics --runtime proton7
Compare the direct-Proton prefix read-only
pristontale-eu login-trace Retired; refuses without starting the game
pristontale-eu update Install or update official files
pristontale-eu update --repair Verify and repair every game file
pristontale-eu update --dry-run Preview an update
pristontale-eu import-game "/path/to/game" Import an existing installation
pristontale-eu prepare Build the safe runtime view
pristontale-eu play Launch through Gamescope/Proton
pristontale-eu play --no-box Launch without Gamescope
pristontale-eu play Launch through the managed Proton window
pristontale-eu play --no-box Launch without X11 window styling
pristontale-eu play --no-world-fix Compatibility alias for the pristine launch
pristontale-eu settings Show current settings
pristontale-eu settings --init Write default settings
pristontale-eu settings --init Create/migrate settings without resetting choices
pristontale-eu settings --reset Explicitly restore every launcher default
pristontale-eu launcher-update --check Check the signed launcher channel
pristontale-eu launcher-update --verify Validate the full release without installing it
pristontale-eu launcher-update Install and activate an update
pristontale-eu launcher-update --rollback Restore the previous launcher
pristontale-eu sync Sync without import discovery
@@ -66,7 +75,6 @@ pristontale-eu wine-update Run the legacy WPF updater
pristontale-eu fetch Fetch the publisher launcher
pristontale-eu setup Prepare its Wine environment
pristontale-eu cache-compatibility PATH Cache a verified compatibility pair
pristontale-eu cache-runtime-patch PATH Cache a verified runtime patch
```
Use `./install.sh --dry-run` to preview dependency installation or

View File

@@ -1 +0,0 @@
cd0314c9fba1779dac5bbe24190c174941dbed89c371b68808fefa37de7750bf pristontale-eu-linux-installer-v0-5-2.tar.gz

Binary file not shown.

Binary file not shown.

View File

@@ -1 +0,0 @@
e391accd5448f3239aeda2561e5faf2eb17b5aa5c61837da609f456815398002 pristontale-eu-linux-update-v0.5.2.tar.gz

11
dist/stable.json vendored
View File

@@ -1,11 +0,0 @@
{
"archive_sha256": "e391accd5448f3239aeda2561e5faf2eb17b5aa5c61837da609f456815398002",
"archive_size": 71737,
"archive_url": "https://git.m4ko.net/m4ko/pristontale-eu-linux/releases/download/v0.5.2/pristontale-eu-linux-update-v0.5.2.tar.gz",
"channel": "stable",
"minimum_updater_version": "0.5.0",
"notes_url": "https://git.m4ko.net/m4ko/pristontale-eu-linux/releases/tag/v0.5.2",
"published_at": "2026-08-27T19:01:33Z",
"schema": 1,
"version": "0.5.2"
}

BIN
dist/stable.json.sig vendored

Binary file not shown.

View File

@@ -3,12 +3,18 @@
set -euo pipefail
SOURCE_ROOT="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd)"
VERSION="${PTLINUX_VERSION:-0.5.9}"
SOURCE_VERSION="$(sed -n 's/^__version__ = "\([0-9][0-9]*\.[0-9][0-9]*\.[0-9][0-9]*\)"$/\1/p' "${SOURCE_ROOT}/ptlinux/__init__.py")"
VERSION="${PTLINUX_VERSION:-$SOURCE_VERSION}"
if [[ ! "$VERSION" =~ ^[0-9]+\.[0-9]+\.[0-9]+$ ]]
then
printf 'Invalid launcher version: %s (expected X.Y.Z)\n' "$VERSION" >&2
exit 2
fi
if [[ "$VERSION" != "$SOURCE_VERSION" ]]
then
printf 'Launcher version %s does not match source version %s.\n' "$VERSION" "$SOURCE_VERSION" >&2
exit 2
fi
DATA_HOME="${XDG_DATA_HOME:-${HOME}/.local/share}"
BIN_HOME="${HOME}/.local/bin"
APPLICATION_HOME="${DATA_HOME}/applications"
@@ -138,7 +144,7 @@ if [[ ! -d "${SOURCE_ROOT}/ptlinux" \
|| ! -f "${SOURCE_ROOT}/ptlinux/launcher_wrapper.py" \
|| ! -f "${SOURCE_ROOT}/ptlinux/self_update.py" \
|| ! -f "${SOURCE_ROOT}/ptlinux/launcher-update-public.pem" \
|| ! -f "${SOURCE_ROOT}/ptlinux/native/pt-world-fix.exe" \
|| ! -f "${SOURCE_ROOT}/ptlinux/native/pt-identity-probe.exe" \
|| ! -f "${SOURCE_ROOT}/packaging/eu.pristontale.PTLinux.desktop" \
|| ! -f "${SOURCE_ROOT}/packaging/eu.pristontale.PTLinux.png" \
|| ! -f "${SOURCE_ROOT}/packaging/pristontale-eu" \
@@ -196,20 +202,28 @@ install -d -m 0755 \
"${STAGING_RELEASE}/ptlinux" \
"${STAGING_RELEASE}/ptlinux/native" \
"${STAGING_RELEASE}/ptdiag"
install -m 0644 "${SOURCE_ROOT}/ptlinux/"*.py "${STAGING_RELEASE}/ptlinux/"
for SOURCE_MODULE in "${SOURCE_ROOT}/ptlinux/"*.py
do
case "${SOURCE_MODULE##*/}" in
desktop_entry.py|launcher_wrapper.py)
continue
;;
esac
install -m 0644 "$SOURCE_MODULE" "${STAGING_RELEASE}/ptlinux/"
done
install -m 0644 \
"${SOURCE_ROOT}/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"
"${SOURCE_ROOT}/ptlinux/native/pt-identity-probe.exe" \
"${STAGING_RELEASE}/ptlinux/native/pt-identity-probe.exe"
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/logs.py" "${STAGING_RELEASE}/ptdiag/logs.py"
printf '{\n "schema": 1,\n "version": "%s"\n}\n' "$VERSION" > "${STAGING_RELEASE}/release.json"
chmod 0644 "${STAGING_RELEASE}/release.json"
if ! "$PYTHON_BIN" -I -c \
if ! "$PYTHON_BIN" -I -B -c \
'import sys; sys.path.insert(0, sys.argv[1]); import ptlinux, ptlinux.cli, ptlinux.self_update; raise SystemExit(0 if ptlinux.__version__ == sys.argv[2] else 1)' \
"$STAGING_RELEASE" "$VERSION"
then
@@ -286,7 +300,7 @@ CURRENT_LINK="${APP_ROOT}/.current.$$.tmp"
ln -s -- "releases/${VERSION}" "$CURRENT_LINK"
mv -Tf -- "$CURRENT_LINK" "${APP_ROOT}/current"
"$PYTHON_BIN" "${RELEASE_ROOT}/ptlinux/launcher_wrapper.py" \
"$PYTHON_BIN" "${SOURCE_ROOT}/ptlinux/launcher_wrapper.py" \
"${SOURCE_ROOT}/packaging/pristontale-eu" \
"$APP_ROOT" \
"$INSTALL_DIRECTORY" \
@@ -294,7 +308,7 @@ mv -Tf -- "$CURRENT_LINK" "${APP_ROOT}/current"
install -m 0644 \
"${SOURCE_ROOT}/packaging/eu.pristontale.PTLinux.png" \
"$ICON_PATH"
"$PYTHON_BIN" "${RELEASE_ROOT}/ptlinux/desktop_entry.py" \
"$PYTHON_BIN" "${SOURCE_ROOT}/ptlinux/desktop_entry.py" \
"${SOURCE_ROOT}/packaging/eu.pristontale.PTLinux.desktop" \
"${BIN_HOME}/pristontale-eu" \
"$ICON_PATH" \

File diff suppressed because it is too large Load Diff

View File

@@ -8,3 +8,4 @@ Terminal=false
Categories=Game;RolePlaying;
Keywords=PristonTale;MMORPG;Wine;Proton;
StartupNotify=true
StartupWMClass=PristonTaleEU

View File

@@ -16,7 +16,7 @@ do
then
continue
fi
if "$python_bin" -I -c \
if "$python_bin" -I -B -c \
'import sys; sys.path.insert(0, sys.argv[1]); import ptlinux, ptlinux.cli' \
"${candidate}" >/dev/null 2>&1
then
@@ -31,11 +31,11 @@ then
exit 1
fi
if [ -n "${PYTHONPATH:-}" ]
then
export PYTHONPATH="${release_root}:${PYTHONPATH}"
else
export PYTHONPATH="${release_root}"
fi
# Never let an unpacked older installer in the caller's current directory
# shadow the selected installed release. Children started by the dashboard
# inherit the same safe-path policy, package root, and no-bytecode policy.
export PYTHONSAFEPATH=1
export PYTHONDONTWRITEBYTECODE=1
export PYTHONPATH="${release_root}"
exec "$python_bin" -m ptlinux "$@"
exec "$python_bin" -B -P -m ptlinux "$@"

View File

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

View File

@@ -10,7 +10,7 @@ import json
from pathlib import Path
import sys
import time
from typing import Sequence
from typing import Callable, Sequence
from ptdiag.logs import (
LogFileState,
@@ -21,10 +21,16 @@ from ptdiag.logs import (
from . import __version__
from .command import CommandError, CommandRunner
from .depot import SyncProgress, sync_managed_game
from .depot import (
SyncProgress,
read_cached_depot_launch_catalog,
select_depot_start_option,
sync_managed_game,
)
from .doctor import inspect_host
from .download import OFFICIAL_LAUNCHER_URL, fetch_official_launcher
from .importer import GameImportResult, discover_existing_game, import_existing_game
from .import_state import game_import_requires_verification, mark_game_import_verified
from .launcher import (
LauncherError,
configure_launcher_prefix,
@@ -32,22 +38,24 @@ from .launcher import (
launch_official_updater,
prepare_launcher_binary,
)
from .locking import launcher_operation_lock
from .locking import LauncherPlayLease, launcher_operation_lock, launcher_play_lock
from .login_diagnostics import inspect_login_environment
from .login_trace import DEFAULT_TRACE_TIMEOUT_SECONDS, run_login_trace
from .login_trace import DEFAULT_TRACE_TIMEOUT_SECONDS, LOGIN_TRACE_DISABLED_REASON
from .overlay import cache_compatibility_binaries, prepare_runtime_view
from .paths import AppPaths
from .runtime import (
game_runtime_ready,
prepare_game_runtime,
run_game,
select_runtime,
)
from .settings import Settings, ensure_settings, load_settings, save_settings
from .settings import Settings, ensure_settings, load_settings, reset_settings
from .self_update import (
check_for_launcher_update,
install_launcher_update,
rollback_launcher_update,
update_check_to_dict,
verify_launcher_update_channel,
)
@@ -76,6 +84,15 @@ def build_parser() -> argparse.ArgumentParser:
default=5.0,
help="TCP connection timeout in seconds (default: 5)",
)
login_diagnostics.add_argument(
"--runtime",
choices=("configured", "auto", "proton7", "proton7-container", "umu"),
default="configured",
help=(
"probe one runtime without changing settings or starting the game "
"(default: configured)"
),
)
fetch = subparsers.add_parser("fetch", help="fetch and safely prepare the latest official launcher")
fetch.add_argument("--dry-run", action="store_true")
@@ -136,6 +153,10 @@ def build_parser() -> argparse.ArgumentParser:
play = subparsers.add_parser("play", help="launch the game in the managed Proton window")
play.add_argument("--game-dir", type=Path)
play.add_argument(
"--realm",
help="launch an available realm from the last official depot manifest",
)
play.add_argument("--no-box", action="store_true", help="launch without X11 window styling")
play.add_argument(
"--publisher-dll",
@@ -151,7 +172,7 @@ def build_parser() -> argparse.ArgumentParser:
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",
help="retired unsafe authentication trace (always refuses to launch)",
)
login_trace.add_argument("--game-dir", type=Path)
login_trace.add_argument("--no-box", action="store_true", help="launch without X11 window styling")
@@ -163,12 +184,22 @@ def build_parser() -> argparse.ArgumentParser:
help="maximum game runtime in seconds (default: 600)",
)
settings = subparsers.add_parser("settings", help="show or initialize settings")
settings.add_argument("--init", action="store_true", help="write conservative defaults")
settings = subparsers.add_parser("settings", help="show, initialize, or reset settings")
settings_action = settings.add_mutually_exclusive_group()
settings_action.add_argument(
"--init",
action="store_true",
help="create missing settings and migrate legacy files without resetting choices",
)
settings_action.add_argument(
"--reset",
action="store_true",
help="explicitly replace all launcher settings with defaults",
)
launcher_update = subparsers.add_parser(
"launcher-update",
help="check, install, or roll back the native Linux launcher",
help="check, verify, install, or roll back the native Linux launcher",
)
launcher_action = launcher_update.add_mutually_exclusive_group()
launcher_action.add_argument(
@@ -176,6 +207,11 @@ def build_parser() -> argparse.ArgumentParser:
action="store_true",
help="authenticate the update manifest without installing it",
)
launcher_action.add_argument(
"--verify",
action="store_true",
help="download and fully validate the published release without installing it",
)
launcher_action.add_argument(
"--rollback",
action="store_true",
@@ -207,7 +243,12 @@ def main(argv: Sequence[str] | None = None) -> int:
if command == "doctor":
return _doctor(arguments.json)
if command == "login-diagnostics":
return _login_diagnostics(paths, arguments.timeout, arguments.json)
return _login_diagnostics(
paths,
arguments.timeout,
arguments.runtime,
arguments.json,
)
if command == "fetch":
return _fetch(paths, arguments.dry_run, arguments.json)
if command == "setup":
@@ -262,10 +303,10 @@ def main(argv: Sequence[str] | None = None) -> int:
paths,
arguments.game_dir,
arguments.no_box,
arguments.publisher_dll,
arguments.dry_run,
arguments.skip_runtime_setup,
arguments.json,
realm=arguments.realm,
)
if command == "login-trace":
return _login_trace(
@@ -277,11 +318,12 @@ def main(argv: Sequence[str] | None = None) -> int:
arguments.json,
)
if command == "settings":
return _settings(paths, arguments.init, arguments.json)
return _settings(paths, arguments.init, arguments.reset, arguments.json)
if command == "launcher-update":
return _launcher_update(
paths,
arguments.check,
arguments.verify,
arguments.rollback,
arguments.manifest_url,
arguments.public_key,
@@ -326,13 +368,24 @@ def _doctor(as_json: bool) -> int:
return 0 if report.ok else 1
def _login_diagnostics(paths: AppPaths, timeout_seconds: float, as_json: bool) -> int:
def _login_diagnostics(
paths: AppPaths,
timeout_seconds: float,
runtime_choice: str,
as_json: bool,
) -> int:
settings = _load_or_default_settings(paths, persist=False)
report = inspect_login_environment(
paths,
settings,
timeout_seconds=timeout_seconds,
)
runner = _runner(paths, "login-diagnostics", False)
with _operation_guard(paths, "login-diagnostics", False):
report = inspect_login_environment(
paths,
settings,
timeout_seconds=timeout_seconds,
runner=runner,
runtime_override=(
None if runtime_choice == "configured" else runtime_choice
),
)
if as_json:
_print_json(report.to_dict())
else:
@@ -340,12 +393,30 @@ def _login_diagnostics(paths: AppPaths, timeout_seconds: float, as_json: bool) -
print(f"Launcher: {report.launcher_version} ({report.launcher_module})")
print(f"Python: {report.python_executable}")
print(f"Host: {report.distribution_id} ({report.kernel_release})")
print(
"Runtime test: "
f"{report.runtime_requested} "
f"(configured: {report.runtime_configured}; settings unchanged)"
)
status = "REACHABLE" if probe.reachable else "UNREACHABLE"
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})")
if probe.error:
print(f"Network error: {probe.error}")
print("Authentication: not tested; no credentials or game-protocol data were sent")
print(f"Login responses: {report.login_response_ownership}")
identity_status = (
"ready"
if report.client_identity_ready is True
else "NOT READY"
if report.client_identity_ready is False
else "not probed"
)
print(f"Client identity: {identity_status}; raw identity values were not collected")
print(f"Privacy: {report.client_identity_privacy}")
if not report.login_trace_enabled:
print("Login trace: retired (will not launch or instrument the game)")
print(f"Publisher test: {report.publisher_comparison_command}")
print(f"Game profile: {report.game_profile}")
print(f"Game.exe: {report.game_exe}")
if report.game_exe_sha256:
@@ -353,10 +424,15 @@ def _login_diagnostics(paths: AppPaths, timeout_seconds: float, as_json: bool) -
print(f"game.dll: {report.game_dll}")
if 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}")
print(
"Managed source: "
+ ("matches runtime" if report.runtime_matches_managed_source else "DIFFERS")
)
if report.depot_build is not None:
print(
f"Depot: build {report.depot_build} "
f"({report.depot_version or 'unknown version'})"
)
if report.runtime_kind:
print(f"Runtime: {report.runtime_kind} ({report.runtime_path or 'unknown path'})")
else:
@@ -463,13 +539,22 @@ def _native_update(
runner,
announce=not as_json,
)
verify_import = game_import is not None or game_import_requires_verification(paths)
_print_operation_status(as_json, 2, 3, "Synchronizing official game files…")
result = sync_managed_game(
paths,
repair=repair,
repair=repair or verify_import,
quarantine_unmanaged=verify_import,
dry_run=dry_run,
progress=None if as_json else _print_sync_progress,
)
if verify_import and not dry_run:
mark_game_import_verified(
paths,
build=result.build,
files_quarantined=result.files_quarantined,
quarantine_directory=result.quarantine_directory,
)
overlay = None
if not dry_run and paths.source_game.is_dir():
_print_operation_status(as_json, 3, 3, "Preparing the clean game runtime view…")
@@ -518,6 +603,23 @@ def _wine_update(
announce=not as_json,
)
updater = launch_official_updater(paths, tools, runner)
native_verification = None
verify_import = not dry_run and (
game_import is not None or game_import_requires_verification(paths)
)
if verify_import:
native_verification = sync_managed_game(
paths,
repair=True,
quarantine_unmanaged=True,
progress=None if as_json else _print_sync_progress,
)
mark_game_import_verified(
paths,
build=native_verification.build,
files_quarantined=native_verification.files_quarantined,
quarantine_directory=native_verification.quarantine_directory,
)
overlay = None
if not dry_run and paths.source_game.is_dir():
overlay = asdict(prepare_runtime_view(paths, paths.source_game))
@@ -526,6 +628,9 @@ def _wine_update(
"dry_run": dry_run,
"game_import": asdict(game_import) if game_import else None,
"updater": asdict(updater),
"native_verification": (
asdict(native_verification) if native_verification is not None else None
),
"runtime_view": overlay,
}
_emit(value, as_json)
@@ -593,9 +698,15 @@ def _maybe_import_game(
def _prepare(paths: AppPaths, game_dir: Path | None, as_json: bool) -> int:
settings = ensure_settings(paths)
source = _source_game(paths, settings, game_dir)
result = prepare_runtime_view(paths, source)
with _operation_guard(paths, "prepare", False):
if game_dir is None and game_import_requires_verification(paths):
raise LauncherError(
"the imported game has not completed its full publisher verification; "
"run pristontale-eu update"
)
settings = ensure_settings(paths)
source = _source_game(paths, settings, game_dir)
result = prepare_runtime_view(paths, source)
_emit({"ok": True, "runtime_view": asdict(result)}, as_json)
return 0
@@ -604,20 +715,23 @@ def _play(
paths: AppPaths,
game_dir: Path | None,
no_box: bool,
publisher_dll: bool,
dry_run: bool,
skip_runtime_setup: bool,
as_json: bool,
*,
realm: str | None = None,
) -> int:
with _operation_guard(paths, "play", dry_run):
with _play_guard(paths, dry_run) as play_lease:
return _play_locked(
paths,
game_dir,
no_box,
publisher_dll,
dry_run,
skip_runtime_setup,
as_json,
realm=realm,
reuse_prepared=not play_lease.preparation_allowed,
before_game=play_lease.allow_shared_clients,
)
@@ -625,27 +739,69 @@ 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,
*,
realm: str | None = None,
reuse_prepared: bool = False,
before_game: Callable[[], None] | None = None,
) -> 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
if game_dir is None and game_import_requires_verification(paths):
raise LauncherError(
"the imported game has not completed its full publisher verification; "
"run pristontale-eu update before playing"
)
settings = _load_or_default_settings(paths, persist=not dry_run)
start_option = select_depot_start_option(
read_cached_depot_launch_catalog(paths.depot_launch_catalog_cache),
realm or settings.realm,
)
source = _source_game(paths, settings, game_dir)
overlay = None
_print_operation_status(as_json, 1, 4, "Preparing the managed game view…")
if not dry_run:
_print_operation_status(
as_json,
1,
4,
(
"Reusing the active managed game view…"
if reuse_prepared
else "Preparing the managed game view…"
),
)
if reuse_prepared:
if game_dir is not None:
raise LauncherError(
"an additional client cannot replace the active managed game view; "
"close every client before using --game-dir"
)
if not paths.runtime_view.is_dir():
raise LauncherError(
"another game client is active, but the prepared runtime view is missing"
)
elif not dry_run:
overlay = prepare_runtime_view(paths, source)
_print_operation_status(as_json, 2, 4, "Selecting the configured game runtime…")
selection = select_runtime(settings)
runner = _runner(paths, "play", dry_run)
setup_results = []
if not skip_runtime_setup:
if reuse_prepared:
_print_operation_status(
as_json,
3,
4,
"Reusing the active Proton and DirectX runtime…",
)
if not game_runtime_ready(paths, selection):
raise LauncherError(
"another game client is active, but its prepared runtime no longer "
"matches the current settings; close every client and launch again"
)
elif not skip_runtime_setup:
_print_operation_status(as_json, 3, 4, "Preparing Proton and DirectX components…")
setup_results = prepare_game_runtime(paths, selection, runner)
else:
_print_operation_status(as_json, 3, 4, "Using the existing Proton runtime…")
_print_operation_status(as_json, 4, 4, "Starting the PristonTale game client…")
native_log_directory = paths.game_save / "logs"
native_log_baseline = snapshot_native_logs(
@@ -653,14 +809,16 @@ def _play_locked(
patterns=("log_*.txt",),
)
game_started_ns = time.time_ns()
if before_game is not None:
before_game()
try:
outcome = run_game(
paths,
settings,
selection,
runner,
start_option=start_option,
use_gamebox=not no_box,
use_world_fix=False,
)
except KeyboardInterrupt:
current_game_log = _current_game_log(
@@ -676,17 +834,23 @@ def _play_locked(
native_log_baseline,
game_started_ns,
)
world_fix_status = None
runtime_issues = _runtime_issue_summary(outcome.stderr)
if (
selection.kind == "proton7-container"
and runtime_issues["host_futex_fatal_count"]
):
runtime_issues["recommended_one_variable_comparison"] = "proton7"
_emit(
{
"ok": outcome.returncode == 0,
"dry_run": dry_run,
"runtime": asdict(selection),
"realm": asdict(start_option),
"runtime_view": asdict(overlay) if overlay else None,
"setup_commands": [asdict(result) for result in setup_results],
"game": asdict(outcome),
"current_game_log": current_game_log,
"world_fix": world_fix_status,
"runtime_issues": runtime_issues,
},
as_json,
)
@@ -722,6 +886,11 @@ def _current_game_log(
"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,
"banned_code_1_in_native_log": analysis.banned_code_1,
"banned_code_1_observation_scope": (
"native-game-log-only; server-provided login-screen text is not "
"written to this log"
),
"stage_without_new_stage_events": [
event.to_dict()
for event in analysis.events
@@ -736,6 +905,19 @@ def _current_game_log(
}
def _runtime_issue_summary(stderr: str) -> dict[str, object]:
folded = stderr.casefold()
futex_message = "the futex facility returned an unexpected error code"
wrong_elf_message = "wrong elf class: elfclass64"
return {
"host_futex_fatal_count": folded.count(futex_message),
"gstreamer_wrong_elf_class_count": folded.count(wrong_elf_message),
"gstreamer_stream_error": "winegstreamer error" in folded,
"mesa_egl_initialization_failed": "egl: failed to create dri2 screen" in folded,
"steam_api_app_id_missing": "steamapi_init() failed; no appid found" in folded,
}
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"]:
@@ -761,40 +943,24 @@ def _login_trace(
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
# Preserve the command name so old instructions fail safely before touching
# the prefix, preparing the runtime view, or starting Game.exe.
_ = paths, game_dir, no_box, skip_runtime_setup, timeout_seconds, as_json
raise RuntimeError(LOGIN_TRACE_DISABLED_REASON)
def _settings(paths: AppPaths, initialize: bool, as_json: bool) -> int:
if initialize:
save_settings(paths, Settings())
settings = ensure_settings(paths) if initialize else load_settings(paths)
def _settings(
paths: AppPaths,
initialize: bool,
reset: bool,
as_json: bool,
) -> int:
if reset:
settings = reset_settings(paths)
elif initialize:
settings = ensure_settings(paths)
else:
settings = load_settings(paths)
_emit(
{
"ok": True,
@@ -809,6 +975,7 @@ def _settings(paths: AppPaths, initialize: bool, as_json: bool) -> int:
def _launcher_update(
paths: AppPaths,
check_only: bool,
verify: bool,
rollback: bool,
manifest_url: str | None,
public_key: Path | None,
@@ -837,6 +1004,21 @@ def _launcher_update(
as_json,
)
return 0
if verify:
_print_operation_status(as_json, 1, 2, "Authenticating the launcher release…")
result = verify_launcher_update_channel(
manifest_url=manifest_url,
public_key_path=public_key,
)
_print_operation_status(as_json, 2, 2, "Published launcher release verified.")
_emit(
{
"ok": True,
"launcher_update_verification": asdict(result),
},
as_json,
)
return 0
_print_operation_status(as_json, 1, 2, "Verifying the signed launcher release…")
result = install_launcher_update(
paths,
@@ -880,10 +1062,19 @@ def _load_or_default_settings(paths: AppPaths, *, persist: bool) -> Settings:
def _runner(paths: AppPaths, operation: str, dry_run: bool) -> CommandRunner:
timestamp = datetime.now().astimezone().strftime("%Y%m%d-%H%M%S")
timestamp = datetime.now().astimezone().strftime("%Y%m%d-%H%M%S-%f")
return CommandRunner(paths.logs / f"{timestamp}-{operation}.jsonl", dry_run=dry_run)
def _play_guard(
paths: AppPaths,
dry_run: bool,
) -> AbstractContextManager[LauncherPlayLease]:
if dry_run:
return nullcontext(LauncherPlayLease.dry_run())
return launcher_play_lock(paths.launcher_operation_lock)
def _operation_guard(
paths: AppPaths,
operation: str,

View File

@@ -2,6 +2,7 @@
from __future__ import annotations
from collections import deque
from dataclasses import asdict, dataclass
from datetime import datetime
import json
@@ -16,6 +17,46 @@ from typing import Callable, Mapping, Sequence, TextIO
OutputLineFilter = Callable[[str], str | None]
MAX_CAPTURE_CHARACTERS = 1024 * 1024
SUBPROCESS_ENCODING = "utf-8"
SUBPROCESS_DECODE_ERRORS = "replace"
class BoundedTextCapture:
"""Retain the newest subprocess output without unbounded memory growth."""
def __init__(self, max_characters: int = MAX_CAPTURE_CHARACTERS) -> None:
if max_characters <= 0:
raise ValueError("capture size must be positive")
self._max_characters = max_characters
self._parts: deque[str] = deque()
self._characters = 0
self._lock = threading.Lock()
def append(self, value: str) -> None:
if not value:
return
with self._lock:
if len(value) >= self._max_characters:
self._parts.clear()
self._parts.append(value[-self._max_characters:])
self._characters = self._max_characters
return
self._parts.append(value)
self._characters += len(value)
while self._characters > self._max_characters:
overflow = self._characters - self._max_characters
oldest = self._parts[0]
if len(oldest) <= overflow:
self._parts.popleft()
self._characters -= len(oldest)
else:
self._parts[0] = oldest[overflow:]
self._characters -= overflow
def getvalue(self) -> str:
with self._lock:
return "".join(self._parts)
@dataclass(frozen=True)
@@ -95,6 +136,8 @@ class CommandRunner:
cwd=str(cwd) if cwd is not None else None,
env=environment,
text=True,
encoding=SUBPROCESS_ENCODING,
errors=SUBPROCESS_DECODE_ERRORS,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
timeout=timeout_seconds,
@@ -103,8 +146,8 @@ class CommandRunner:
outcome = CommandOutcome(
argv=arguments,
returncode=completed.returncode,
stdout=completed.stdout,
stderr=completed.stderr,
stdout=_bounded_text(completed.stdout),
stderr=_bounded_text(completed.stderr),
duration_seconds=time.monotonic() - started,
timed_out=False,
dry_run=False,
@@ -113,8 +156,8 @@ class CommandRunner:
outcome = CommandOutcome(
argv=arguments,
returncode=None,
stdout=_as_text(exc.stdout),
stderr=_as_text(exc.stderr),
stdout=_bounded_text(_as_text(exc.stdout)),
stderr=_bounded_text(_as_text(exc.stderr)),
duration_seconds=time.monotonic() - started,
timed_out=True,
dry_run=False,
@@ -139,17 +182,19 @@ class CommandRunner:
cwd=str(cwd) if cwd is not None else None,
env=dict(environment),
text=True,
encoding=SUBPROCESS_ENCODING,
errors=SUBPROCESS_DECODE_ERRORS,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
bufsize=1,
start_new_session=True,
)
stdout: list[str] = []
stderr: list[str] = []
stdout = BoundedTextCapture()
stderr = BoundedTextCapture()
def relay(
stream: TextIO,
capture: list[str],
capture: BoundedTextCapture,
destination: TextIO,
) -> None:
try:
@@ -196,8 +241,8 @@ class CommandRunner:
return CommandOutcome(
argv=arguments,
returncode=None if timed_out else returncode,
stdout="".join(stdout),
stderr="".join(stderr),
stdout=stdout.getvalue(),
stderr=stderr.getvalue(),
duration_seconds=time.monotonic() - started,
timed_out=timed_out,
dry_run=False,
@@ -228,6 +273,10 @@ def _as_text(value: bytes | str | None) -> str:
return value
def _bounded_text(value: str) -> str:
return value[-MAX_CAPTURE_CHARACTERS:]
def _terminate_process_group(
process: subprocess.Popen[str],
*,

View File

@@ -13,6 +13,7 @@ import json
import os
from pathlib import Path, PurePosixPath
import re
import stat
import threading
import time
from typing import Callable, Iterable, Mapping, Protocol
@@ -38,8 +39,41 @@ DOWNLOAD_RETRY_DELAYS_SECONDS = (1.0, 2.0, 4.0, 8.0, 16.0)
PROGRESS_UPDATE_INTERVAL_SECONDS = 0.075
MAX_CREATE_IF_MISSING_BYTES = 16 * 1024 * 1024
MAX_LINE_ENDING_RECOVERY_BYTES = 1024 * 1024
UNMANAGED_EXECUTABLE_SUFFIXES = frozenset(
{
".asi",
".bat",
".cmd",
".com",
".cpl",
".desktop",
".dll",
".drv",
".exe",
".jar",
".js",
".lnk",
".lua",
".msi",
".ocx",
".ps1",
".py",
".reg",
".scr",
".sh",
".so",
".url",
".vbe",
".vbs",
".wsf",
}
)
ALWAYS_VERIFY_PATHS = frozenset(
{"game.exe", "game.dll", "gamese.exe", "gamese.dll"}
)
SHA1_PATTERN = re.compile(r"^[0-9a-fA-F]{40}$")
CONTENT_RANGE_PATTERN = re.compile(r"^bytes\s+(\d+)-(\d+)/(\d+)$", re.IGNORECASE)
DEPOT_LAUNCH_CATALOG_SCHEMA = 1
class DepotError(RuntimeError):
@@ -72,6 +106,22 @@ class CreateIfMissing:
content_base64: str
@dataclass(frozen=True)
class DepotStartOption:
"""One publisher-advertised realm launch target."""
name: str
path: str
arguments: str
DEFAULT_DEPOT_START_OPTION = DepotStartOption(
name="Babel",
path="Game.exe",
arguments="PTEU_REBORN",
)
@dataclass(frozen=True)
class DepotSnapshot:
"""The current, verified shape of the public game depot."""
@@ -84,6 +134,101 @@ class DepotSnapshot:
files: tuple[DepotFile, ...]
deleted_files: tuple[str, ...]
create_if_missing: tuple[CreateIfMissing, ...]
start_options: tuple[DepotStartOption, ...] = (DEFAULT_DEPOT_START_OPTION,)
default_start_option: str = DEFAULT_DEPOT_START_OPTION.name
@dataclass(frozen=True)
class DepotLaunchCatalog:
"""Small cached subset of a depot manifest needed for offline Play."""
build: int
start_options: tuple[DepotStartOption, ...]
default_start_option: str
def launch_catalog_from_snapshot(snapshot: DepotSnapshot) -> DepotLaunchCatalog:
return DepotLaunchCatalog(
build=snapshot.build,
start_options=snapshot.start_options,
default_start_option=snapshot.default_start_option,
)
def select_depot_start_option(
catalog: DepotLaunchCatalog | None,
preferred_realm: str | None = None,
) -> DepotStartOption:
"""Choose a published realm, falling back to Babel when possible."""
current = catalog or DepotLaunchCatalog(
build=0,
start_options=(DEFAULT_DEPOT_START_OPTION,),
default_start_option=DEFAULT_DEPOT_START_OPTION.name,
)
if preferred_realm:
preferred = preferred_realm.casefold()
for option in current.start_options:
if option.name.casefold() == preferred:
return option
for option in current.start_options:
if option.name.casefold() == DEFAULT_DEPOT_START_OPTION.name.casefold():
return option
for option in current.start_options:
if option.name.casefold() == current.default_start_option.casefold():
return option
return current.start_options[0]
def cache_depot_launch_catalog(path: Path, snapshot: DepotSnapshot) -> None:
"""Cache only publisher-advertised launch metadata for network-free Play."""
catalog = launch_catalog_from_snapshot(snapshot)
_write_json(
path,
{
"schema": DEPOT_LAUNCH_CATALOG_SCHEMA,
"build": catalog.build,
"default_start_option": catalog.default_start_option,
"start_options": [asdict(option) for option in catalog.start_options],
},
)
def read_cached_depot_launch_catalog(path: Path) -> DepotLaunchCatalog | None:
"""Read the last valid launch catalog without performing network I/O."""
try:
value = json.loads(path.read_text(encoding="utf-8"))
if not isinstance(value, dict):
return None
if value.get("schema") != DEPOT_LAUNCH_CATALOG_SCHEMA:
return None
build = _positive_int(value.get("build"), "cached depot build")
raw_options = value.get("start_options")
if not isinstance(raw_options, list):
return None
manifest_options = [
{
"Name": item.get("name"),
"Path": item.get("path"),
"Arguments": item.get("arguments"),
}
if isinstance(item, dict)
else item
for item in raw_options
]
options, default_name = _parse_start_options(
manifest_options,
value.get("default_start_option"),
)
except (OSError, UnicodeError, json.JSONDecodeError, DepotError):
return None
return DepotLaunchCatalog(
build=build,
start_options=options,
default_start_option=default_name,
)
@dataclass(frozen=True)
@@ -129,11 +274,21 @@ class DepotSyncResult:
bytes_downloaded: int
files_deleted: int
files_created: int
files_quarantined: int
quarantine_directory: str | None
repair: bool
dry_run: bool
completed_at: str
@dataclass(frozen=True)
class QuarantineResult:
"""Recoverable unmanaged entries removed before authenticating an import."""
files: int
directory: str | None
ProgressCallback = Callable[[SyncProgress], None]
@@ -453,7 +608,11 @@ def plan_update(
continue
except OSError as exc:
raise DepotError(f"cannot inspect {file.path}: {exc}") from exc
if repair or file.build == snapshot.build:
if (
repair
or file.build == snapshot.build
or file.path.casefold() in ALWAYS_VERIFY_PATHS
):
if _sha1(target).casefold() != file.sha1.casefold():
downloads.append(PlannedDownload(file, "hash mismatch"))
@@ -479,6 +638,7 @@ def sync_managed_game(
paths: AppPaths,
*,
repair: bool = False,
quarantine_unmanaged: bool = False,
dry_run: bool = False,
progress: ProgressCallback | None = None,
client: DepotClient | None = None,
@@ -489,6 +649,9 @@ def sync_managed_game(
depot_client = client or DepotClient()
_emit_progress(progress, "manifest", 0, 0, 0, 0, None)
snapshot = depot_client.fetch_snapshot()
quarantine = QuarantineResult(0, None)
if quarantine_unmanaged and not dry_run:
quarantine = _quarantine_unmanaged_files(paths, snapshot)
plan = plan_update(paths.source_game, snapshot, repair=repair)
_emit_progress(
progress,
@@ -501,7 +664,16 @@ def sync_managed_game(
)
if dry_run:
return _sync_result(snapshot, plan, 0, 0, 0, repair, True)
return _sync_result(
snapshot,
plan,
0,
0,
0,
repair,
True,
quarantine=quarantine,
)
root = _prepare_game_root(paths.source_game, create=True)
_download_plan(depot_client, snapshot, root, plan, progress)
@@ -516,8 +688,10 @@ def sync_managed_game(
repair,
False,
created=created,
quarantine=quarantine,
)
_write_json(paths.native_depot_state, asdict(result))
cache_depot_launch_catalog(paths.depot_launch_catalog_cache, snapshot)
_emit_progress(
progress,
"complete",
@@ -530,6 +704,110 @@ def sync_managed_game(
return result
def _quarantine_unmanaged_files(
paths: AppPaths,
snapshot: DepotSnapshot,
) -> QuarantineResult:
"""Move non-depot import contents aside without following links."""
root = _prepare_game_root(paths.source_game, create=False)
if not root.exists():
return QuarantineResult(0, None)
allowed = {
file.path for file in snapshot.files
} | {
entry.path for entry in snapshot.create_if_missing
}
candidates: list[tuple[Path, str]] = []
def walk_error(error: OSError) -> None:
raise DepotError(f"cannot inspect imported game tree: {error}") from error
for directory, directory_names, file_names in os.walk(
root,
topdown=True,
followlinks=False,
onerror=walk_error,
):
parent = Path(directory)
for name in tuple(directory_names):
path = parent / name
try:
status = path.lstat()
except OSError as exc:
raise DepotError(f"cannot inspect imported entry {path}: {exc}") from exc
if stat.S_ISDIR(status.st_mode):
continue
directory_names.remove(name)
candidates.append((path, path.relative_to(root).as_posix()))
for name in file_names:
path = parent / name
relative = path.relative_to(root).as_posix()
try:
status = path.lstat()
except OSError as exc:
raise DepotError(f"cannot inspect imported entry {path}: {exc}") from exc
if not stat.S_ISREG(status.st_mode):
candidates.append((path, relative))
continue
if relative in allowed:
continue
if _safe_imported_user_file(path, relative):
continue
candidates.append((path, relative))
if not candidates:
return QuarantineResult(0, None)
base = paths.game_import_quarantine.resolve(strict=False)
data_root = paths.data.resolve(strict=False)
try:
base.relative_to(data_root)
except ValueError as exc:
raise DepotError(f"import quarantine is outside the data root: {base}") from exc
stamp = datetime.now().astimezone().strftime("%Y%m%d-%H%M%S")
quarantine = base / f"{stamp}-{uuid4().hex}"
files_root = quarantine / "files"
files_root.mkdir(parents=True, exist_ok=False)
manifest = {
"schema": 1,
"complete": False,
"created_at": datetime.now().astimezone().isoformat(),
"source": str(root),
"files": [relative for _path, relative in candidates],
}
_write_json(quarantine / "quarantine.json", manifest)
for source, relative in candidates:
destination = files_root.joinpath(*PurePosixPath(relative).parts)
destination.parent.mkdir(parents=True, exist_ok=True)
try:
source.replace(destination)
except OSError as exc:
raise DepotError(
f"could not quarantine imported entry {relative}: {exc}"
) from exc
manifest["complete"] = True
_write_json(quarantine / "quarantine.json", manifest)
return QuarantineResult(len(candidates), str(quarantine))
def _safe_imported_user_file(path: Path, relative: str) -> bool:
parts = PurePosixPath(relative).parts
if not parts or parts[0].casefold() != "save":
return False
if path.suffix.casefold() in UNMANAGED_EXECUTABLE_SUFFIXES:
return False
try:
with path.open("rb") as handle:
prefix = handle.read(4)
except OSError as exc:
raise DepotError(f"cannot inspect imported user file {path}: {exc}") from exc
return not (
prefix.startswith(b"MZ")
or prefix.startswith(b"\x7fELF")
or prefix.startswith(b"#!")
)
def _download_plan(
client: DepotClient,
depot_snapshot: DepotSnapshot,
@@ -692,6 +970,7 @@ def _sync_result(
dry_run: bool,
*,
created: int = 0,
quarantine: QuarantineResult = QuarantineResult(0, None),
) -> DepotSyncResult:
return DepotSyncResult(
region=snapshot.region,
@@ -704,6 +983,8 @@ def _sync_result(
bytes_downloaded=bytes_downloaded,
files_deleted=files_deleted,
files_created=created,
files_quarantined=quarantine.files,
quarantine_directory=quarantine.directory,
repair=repair,
dry_run=dry_run,
completed_at=datetime.now().astimezone().isoformat(),
@@ -745,6 +1026,19 @@ def _parse_snapshot(
deleted = _parse_deleted_paths(build_manifest.get("df"), seen_paths)
create_if_missing = _parse_create_if_missing(depot_manifest.get("CreateIfMissing"))
start_options, default_start_option = _parse_start_options(
depot_manifest.get("StartOptions"),
depot_manifest.get("DefaultStartOption"),
)
canonical_file_paths = {file.path.casefold(): file.path for file in files}
start_options = tuple(
DepotStartOption(
name=option.name,
path=canonical_file_paths.get(option.path.casefold(), option.path),
arguments=option.arguments,
)
for option in start_options
)
return DepotSnapshot(
region=region,
server=server,
@@ -754,6 +1048,8 @@ def _parse_snapshot(
files=tuple(files),
deleted_files=deleted,
create_if_missing=create_if_missing,
start_options=start_options,
default_start_option=default_start_option,
)
@@ -798,6 +1094,74 @@ def _parse_create_if_missing(value: object) -> tuple[CreateIfMissing, ...]:
return tuple(entries)
def _parse_start_options(
value: object,
default_value: object,
) -> tuple[tuple[DepotStartOption, ...], str]:
"""Validate the launch targets published by the official depot manifest."""
# Older test fixtures and historical manifests did not expose StartOptions.
# Babel remains the launcher-safe compatibility default for that case.
if value is None:
return (DEFAULT_DEPOT_START_OPTION,), DEFAULT_DEPOT_START_OPTION.name
if not isinstance(value, list) or not value:
raise DepotError("StartOptions must be a non-empty array")
options: list[DepotStartOption] = []
seen_names: set[str] = set()
for item in value:
if not isinstance(item, dict):
raise DepotError("StartOptions contains a non-object entry")
name = _start_option_text(item.get("Name"), "StartOptions Name", 64)
folded_name = name.casefold()
if folded_name in seen_names:
raise DepotError(f"StartOptions has duplicate realm name: {name}")
seen_names.add(folded_name)
path = _normalise_relative_path(item.get("Path"), f"StartOptions path for {name}")
if PurePosixPath(path).suffix.casefold() != ".exe":
raise DepotError(f"StartOptions path is not an executable for {name}: {path}")
arguments = _start_option_text(
item.get("Arguments"),
f"StartOptions arguments for {name}",
2048,
)
options.append(DepotStartOption(name=name, path=path, arguments=arguments))
if default_value is None:
default_name = next(
(
option.name
for option in options
if option.name.casefold() == DEFAULT_DEPOT_START_OPTION.name.casefold()
),
options[0].name,
)
else:
default_name = _start_option_text(
default_value,
"DefaultStartOption",
64,
)
matching_default = next(
(option.name for option in options if option.name.casefold() == default_name.casefold()),
None,
)
if matching_default is None:
raise DepotError(
f"DefaultStartOption does not match an advertised realm: {default_name}"
)
default_name = matching_default
return tuple(options), default_name
def _start_option_text(value: object, field: str, maximum_length: int) -> str:
if not isinstance(value, str):
raise DepotError(f"{field} must be a string")
text = value.strip()
if not text or len(text) > maximum_length or any(ord(character) < 32 for character in text):
raise DepotError(f"invalid {field}")
return text
def _decode_create_if_missing(entry: CreateIfMissing) -> bytes:
try:
content = b64decode(entry.content_base64, validate=True)

View File

@@ -13,6 +13,11 @@ from typing import Mapping
from .launcher import find_wine_tools
ARCH_FAMILY_DISTRIBUTIONS = frozenset(
{"arch", "cachyos", "endeavouros", "garuda", "manjaro"}
)
@dataclass(frozen=True)
class ToolStatus:
name: str
@@ -88,8 +93,8 @@ def inspect_host(environment: Mapping[str, str] | None = None) -> DoctorReport:
ToolStatus(
"umu-run",
shutil.which("umu-run"),
False,
"portable Proton runtime when Steam Proton 7 is unavailable",
distribution in ARCH_FAMILY_DISTRIBUTIONS,
"current portable Proton runtime; automatic on Arch-family hosts",
),
ToolStatus(
"vulkaninfo",
@@ -98,6 +103,7 @@ def inspect_host(environment: Mapping[str, str] | None = None) -> DoctorReport:
"Vulkan diagnostics",
),
]
tools.extend(_platform_optional_tools(distribution, architecture=platform.machine()))
warnings: list[str] = []
if platform.machine().casefold() not in {"x86_64", "amd64"}:
warnings.append("PristonTale EU requires an x86-64 host with 32-bit Vulkan support.")
@@ -107,6 +113,7 @@ def inspect_host(environment: Mapping[str, str] | None = None) -> DoctorReport:
warnings.append(
"No game runtime was found: install umu-launcher or Steam Proton 7.0."
)
warnings.extend(_arch_runtime_warnings(distribution, platform.machine()))
required_ok = all(item.path is not None for item in tools if item.required)
return DoctorReport(
ok=required_ok and not any("No game runtime" in warning for warning in warnings),
@@ -155,10 +162,10 @@ def _package_hint(distribution: str) -> str | None:
"Install python3-tkinter, wine, winetricks, cabextract, openssl, "
"and the matching 32-bit Vulkan driver with DNF."
)
if distribution in {"arch", "cachyos", "endeavouros", "manjaro"}:
if distribution in ARCH_FAMILY_DISTRIBUTIONS:
return (
"Install python, tk, wine, winetricks, umu-launcher, cabextract, "
"openssl, and the matching lib32 Vulkan driver with Pacman."
"openssl, and the matching lib32 Vulkan and gstreamer dependencies with Pacman."
)
if distribution in {"opensuse-tumbleweed", "opensuse-leap", "suse"}:
return (
@@ -182,3 +189,55 @@ def _has_steam_proton7(environment: Mapping[str, str]) -> bool:
if (steam_root / "steamapps/common/Proton 7.0/proton").is_file():
return True
return False
def _platform_optional_tools(distribution: str, *, architecture: str) -> list[ToolStatus]:
if architecture.casefold() not in {"x86_64", "amd64"}:
return []
if distribution in ARCH_FAMILY_DISTRIBUTIONS:
return [
_optional_tool_status(
"lib32-vulkan",
"/usr/lib32/libvulkan.so.1",
"32-bit Vulkan runtime compatibility layer",
),
_optional_tool_status(
"lib32-gstreamer",
"/usr/lib32/gstreamer-1.0",
"32-bit gstreamer runtime compatibility modules",
),
]
return []
def _arch_runtime_warnings(distribution: str, architecture: str) -> list[str]:
if distribution not in ARCH_FAMILY_DISTRIBUTIONS:
return []
if architecture.casefold() not in {"x86_64", "amd64"}:
return [
"Non-x86_64 Linux hosts are not a supported runtime architecture for this game."
]
warnings: list[str] = []
if not _path_exists("/usr/lib32/libvulkan.so.1"):
warnings.append(
"Install lib32-vulkan-icd-loader and the GPU vendor's matching 32-bit "
"Vulkan driver; this 32-bit D3D9 client otherwise cannot initialize DXVK."
)
if not _path_exists("/usr/lib32/gstreamer-1.0"):
warnings.append(
"Install lib32-gstreamer and lib32-gst-plugins-base if the login video or "
"other media does not play. Missing media plugins are not evidence of an "
"authentication failure."
)
return warnings
def _optional_tool_status(name: str, path: str, purpose: str) -> ToolStatus:
if _path_exists(path):
return ToolStatus(name, path, required=False, purpose=purpose)
return ToolStatus(name, None, required=False, purpose=purpose)
def _path_exists(path: str) -> bool:
entry = Path(path)
return entry.is_file() or entry.is_dir()

View File

@@ -10,8 +10,13 @@ import sys
import time
from collections.abc import Collection, Iterable, Sequence
from .desktop_identity import (
APPLICATION_NAME,
APPLICATION_WM_CLASS,
APPLICATION_WM_INSTANCE,
)
WINDOW_TITLE = "PristonTale EU"
WINDOW_TITLE = APPLICATION_NAME
WINDOW_DISCOVERY_TIMEOUT_SECONDS = 60.0
WINDOW_READY_DELAY_SECONDS = 0.5
_X_ERROR_HANDLER = ctypes.CFUNCTYPE(ctypes.c_int, ctypes.c_void_p, ctypes.c_void_p)
@@ -66,6 +71,13 @@ class _MotifWmHints(ctypes.Structure):
]
class _XClassHint(ctypes.Structure):
_fields_ = [
("res_name", ctypes.c_char_p),
("res_class", ctypes.c_char_p),
]
class _X11:
def __init__(self) -> None:
library_path = ctypes.util.find_library("X11") or "libX11.so.6"
@@ -135,6 +147,33 @@ class _X11:
self.library.XMapRaised(self.display, window)
self.library.XFlush(self.display)
def set_window_identity(
self,
window: int,
*,
resource_name: str = APPLICATION_WM_INSTANCE,
resource_class: str = APPLICATION_WM_CLASS,
) -> None:
"""Replace Proton's generic steamapp identity before taskbar matching."""
for label, value in (
("resource name", resource_name),
("resource class", resource_class),
):
if not value or "\x00" in value:
raise GameboxRuntimeError(f"invalid X11 {label}")
hints = _XClassHint(
res_name=resource_name.encode("utf-8"),
res_class=resource_class.encode("utf-8"),
)
if self.library.XSetClassHint(
self.display,
window,
ctypes.byref(hints),
) == 0:
raise GameboxRuntimeError("could not set the game window identity")
self.library.XSync(self.display, 0)
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))
@@ -265,6 +304,12 @@ class _X11:
ctypes.c_int,
]
self.library.XChangeProperty.restype = ctypes.c_int
self.library.XSetClassHint.argtypes = [
ctypes.c_void_p,
ctypes.c_ulong,
ctypes.POINTER(_XClassHint),
]
self.library.XSetClassHint.restype = ctypes.c_int
self.library.XGetWMProtocols.argtypes = [
ctypes.c_void_p,
ctypes.c_ulong,
@@ -325,6 +370,18 @@ def _prepare_game_window(
mapper.map_window(window, borderless=borderless)
def _apply_game_window_identity(mapper: _X11, window: int) -> None:
try:
mapper.set_window_identity(window)
except Exception as exc:
print(
f"Warning: could not assign the PristonTale EU taskbar identity ({exc}); "
"continuing normally",
file=sys.stderr,
flush=True,
)
def run_game_window(
command: Sequence[str],
window_title: str = WINDOW_TITLE,
@@ -371,6 +428,7 @@ def run_game_window(
game_window = window
game_title = title
game_window_created_at = now
_apply_game_window_identity(mapper, window)
print(
f"Game window created; preparing it: {title}",
file=sys.stderr,

View File

@@ -9,6 +9,7 @@ import json
import os
from pathlib import Path
import shutil
import stat
from typing import Mapping
import uuid
@@ -43,6 +44,7 @@ class GameImportResult:
installed_build: int | None
started_at: str
completed_at: str
verification_required: bool
def import_existing_game(
@@ -87,10 +89,13 @@ def import_existing_game(
),
started_at=started_at,
completed_at=started_at,
verification_required=True,
)
destination.parent.mkdir(parents=True, exist_ok=True)
temporary = destination.with_name(f".{destination.name}.import-{uuid.uuid4().hex}")
committed = False
pending_recorded = False
try:
temporary.mkdir(parents=False, exist_ok=False)
runner.run(
@@ -111,8 +116,31 @@ def import_existing_game(
):
cache_compatibility_binaries(paths, source)
temporary.replace(destination)
committed = True
provisional = GameImportResult(
source=str(source),
destination=str(destination),
copy_mode="reflink-or-copy",
files=source_before.files,
bytes=source_before.bytes,
core_sha256=source_before.core_sha256,
depot_metadata_source=(
str(depot_metadata.path) if depot_metadata is not None else None
),
depot_metadata_destination=None,
installed_build=(
depot_metadata.installed_build if depot_metadata is not None else None
),
started_at=started_at,
completed_at=datetime.now().astimezone().isoformat(),
verification_required=True,
)
_write_import_manifest(paths, provisional)
pending_recorded = True
except BaseException:
_remove_staging_tree(temporary, destination.parent)
if committed and not pending_recorded:
_remove_staging_tree(destination, destination.parent)
raise
metadata_destination = None
@@ -140,8 +168,9 @@ def import_existing_game(
),
started_at=started_at,
completed_at=datetime.now().astimezone().isoformat(),
verification_required=True,
)
_write_json(paths.game_import_manifest, asdict(result))
_write_import_manifest(paths, result)
return result
@@ -202,22 +231,48 @@ def inventory_game_tree(source_game: Path) -> GameTreeInventory:
source = source_game.expanduser().resolve(strict=True)
for relative in REQUIRED_GAME_FILES:
path = source / relative
if not path.is_file():
try:
status = path.lstat()
except OSError:
raise GameImportError(f"game installation is incomplete: missing {relative}")
if stat.S_ISLNK(status.st_mode):
raise GameImportError(f"game installation contains a symbolic link: {relative}")
if not stat.S_ISREG(status.st_mode):
raise GameImportError(f"game installation contains a non-file: {relative}")
files = 0
total_bytes = 0
for root, _directories, filenames in os.walk(source, followlinks=False):
def walk_error(error: OSError) -> None:
raise GameImportError(f"cannot inventory game tree: {error}") from error
for root, directories, filenames in os.walk(
source,
followlinks=False,
onerror=walk_error,
):
root_path = Path(root)
for directory in directories:
path = root_path / directory
try:
status = path.lstat()
except OSError as exc:
raise GameImportError(f"cannot inventory game directory {path}: {exc}") from exc
if stat.S_ISLNK(status.st_mode):
raise GameImportError(f"game installation contains a symbolic link: {path}")
if not stat.S_ISDIR(status.st_mode):
raise GameImportError(f"game installation contains a special entry: {path}")
for filename in filenames:
path = root_path / filename
if path.is_symlink():
files += 1
continue
try:
total_bytes += path.stat().st_size
status = path.lstat()
except OSError as exc:
raise GameImportError(f"cannot inventory game file {path}: {exc}") from exc
if stat.S_ISLNK(status.st_mode):
raise GameImportError(f"game installation contains a symbolic link: {path}")
if not stat.S_ISREG(status.st_mode):
raise GameImportError(f"game installation contains a special entry: {path}")
total_bytes += status.st_size
files += 1
return GameTreeInventory(
files=files,
@@ -327,6 +382,21 @@ def _sha256(path: Path) -> str:
def _write_json(path: Path, value: object) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
temporary = path.with_name(f".{path.name}.{os.getpid()}.tmp")
temporary.write_text(json.dumps(value, indent=2, sort_keys=True) + "\n", encoding="utf-8")
temporary.replace(path)
temporary = path.with_name(f".{path.name}.{os.getpid()}.{uuid.uuid4().hex}.tmp")
try:
with temporary.open("x", encoding="utf-8") as handle:
json.dump(value, handle, indent=2, sort_keys=True)
handle.write("\n")
handle.flush()
os.fsync(handle.fileno())
temporary.replace(path)
except BaseException:
temporary.unlink(missing_ok=True)
raise
def _write_import_manifest(paths: AppPaths, result: GameImportResult) -> None:
_write_json(
paths.game_import_manifest,
{"schema": 2, **asdict(result)},
)

View File

@@ -7,6 +7,7 @@ from enum import Enum
import json
from pathlib import Path
from .import_state import game_import_requires_verification
from .paths import AppPaths
@@ -70,6 +71,8 @@ def inspect_game_installation(
local_build = _read_verified_local_build(paths.native_depot_state)
if not installed:
action = PrimaryAction.INSTALL
elif game_import_requires_verification(paths):
action = PrimaryAction.UPDATE
elif official_build is not None and (
local_build is None or local_build < official_build
):

View File

@@ -7,13 +7,51 @@ from datetime import datetime
import fcntl
import os
from pathlib import Path
from typing import Iterator
from typing import Iterator, TextIO
class OperationBusyError(RuntimeError):
"""Raised when another launcher-prefix operation is still active."""
class LauncherPlayLease:
"""Exclusive preparation lease that can become a shared game lease."""
def __init__(
self,
handle: TextIO | None,
*,
preparation_allowed: bool,
record_owned: bool,
) -> None:
self._handle = handle
self.preparation_allowed = preparation_allowed
self._record_owned = record_owned
self._shared = not preparation_allowed
@classmethod
def dry_run(cls) -> LauncherPlayLease:
return cls(None, preparation_allowed=True, record_owned=False)
@property
def shared(self) -> bool:
return self._shared
def allow_shared_clients(self) -> None:
"""Downgrade after all runtime mutations have completed."""
if self._shared:
return
if self._handle is not None:
_write_owner(self._handle, "play")
fcntl.flock(self._handle.fileno(), fcntl.LOCK_SH)
self._shared = True
@property
def record_owned(self) -> bool:
return self._record_owned
@contextmanager
def launcher_operation_lock(lock_path: Path, operation: str) -> Iterator[None]:
lock_path.parent.mkdir(parents=True, exist_ok=True)
@@ -26,13 +64,7 @@ def launcher_operation_lock(lock_path: Path, operation: str) -> Iterator[None]:
raise OperationBusyError(
f"launcher files are busy ({owner}); wait for it to finish and try again"
) from exc
handle.seek(0)
handle.truncate()
handle.write(
f"{operation}, pid={os.getpid()}, "
f"started={datetime.now().astimezone().isoformat()}\n"
)
handle.flush()
_write_owner(handle, operation)
try:
yield
finally:
@@ -40,3 +72,67 @@ def launcher_operation_lock(lock_path: Path, operation: str) -> Iterator[None]:
handle.truncate()
handle.flush()
fcntl.flock(handle.fileno(), fcntl.LOCK_UN)
@contextmanager
def launcher_play_lock(
lock_path: Path,
operation: str = "play",
) -> Iterator[LauncherPlayLease]:
"""Serialize preparation, then allow multiple active game clients.
The first play process acquires an exclusive lease so it may safely rebuild
the runtime view or prefix. It downgrades that lease immediately before
starting Game.exe. Later play processes then receive shared, read-only leases,
while updates and repairs remain excluded until every client exits.
"""
lock_path.parent.mkdir(parents=True, exist_ok=True)
with lock_path.open("a+", encoding="utf-8") as handle:
try:
fcntl.flock(handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
except BlockingIOError:
try:
fcntl.flock(handle.fileno(), fcntl.LOCK_SH | fcntl.LOCK_NB)
except BlockingIOError as exc:
owner = _read_owner(handle)
raise OperationBusyError(
f"launcher files are busy ({owner}); wait for it to finish and try again"
) from exc
lease = LauncherPlayLease(
handle,
preparation_allowed=False,
record_owned=False,
)
else:
_write_owner(handle, f"{operation} preparation")
lease = LauncherPlayLease(
handle,
preparation_allowed=True,
record_owned=True,
)
try:
yield lease
finally:
# A downgraded owner record remains useful while any secondary shared
# lease is active. The next exclusive operation atomically replaces it.
if lease.record_owned and not lease.shared:
handle.seek(0)
handle.truncate()
handle.flush()
fcntl.flock(handle.fileno(), fcntl.LOCK_UN)
def _read_owner(handle: TextIO) -> str:
handle.seek(0)
return handle.read().strip() or "unknown operation"
def _write_owner(handle: TextIO, operation: str) -> None:
handle.seek(0)
handle.truncate()
handle.write(
f"{operation}, pid={os.getpid()}, "
f"started={datetime.now().astimezone().isoformat()}\n"
)
handle.flush()

View File

@@ -2,9 +2,10 @@
from __future__ import annotations
from dataclasses import asdict, dataclass
from dataclasses import asdict, dataclass, replace
from datetime import datetime
import hashlib
import json
from pathlib import Path
import platform
import socket
@@ -13,10 +14,18 @@ import time
from typing import Callable, Protocol
from . import __version__
from .command import CommandRunner
from .identity_probe import (
IDENTITY_PROBE_PRIVACY,
IdentityProbeResult,
probe_client_identity,
)
from .login_trace import LOGIN_TRACE_DISABLED_REASON, LOGIN_TRACE_ENABLED
from .overlay import (
COMPATIBILITY_BINARY_HASHES,
CURRENT_V21_BINARY_HASHES,
PATCHED_V21_GAME_DLL_SHA256,
RETIRED_POST_AUTH_GAME_DLL_SHA256,
)
from .paths import AppPaths
from .runtime import (
@@ -27,12 +36,6 @@ from .runtime import (
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"
@@ -41,8 +44,35 @@ DEFAULT_LOGIN_TIMEOUT_SECONDS = 5.0
VERIFIED_GAME_PROFILES = frozenset(
{"v21.75.83", "verified-compatibility"}
)
RETIRED_POST_AUTH_GAME_DLL_SHA256 = (
"93cd27284eb294753da1111c99bfc9689f82b6d4a59302aee9fdbd8cb52452dd"
SERVER_LOGIN_RESPONSE_OWNERSHIP = (
"Banned Code 1 and Banned Code 3 are server-provided login response text; "
"neither string exists in the verified V21.75.83 Game.exe/game.dll pair"
)
CLIENT_IDENTITY_INPUTS = (
"Win32_Processor",
"Win32_BIOS",
"Win32_BaseBoard",
"Win32_VideoController",
"network-adapter identity",
"C: volume information",
"computer name",
)
PUBLISHER_COMPARISON_COMMAND = "pristontale-eu wine-update"
RUNTIME_COMPONENT_PRIVACY = (
"Only SHA-256 hashes of public runtime executables and DLLs plus expected DLL-"
"override flags are reported. Registry contents and machine identifiers are not "
"hashed or returned."
)
PREFIX_COMPONENT_PATHS = (
"drive_c/windows/syswow64/d3d9.dll",
"drive_c/windows/syswow64/d3dx9_43.dll",
"drive_c/windows/syswow64/d3dcompiler_43.dll",
"drive_c/windows/syswow64/iphlpapi.dll",
"drive_c/windows/syswow64/kernel32.dll",
"drive_c/windows/syswow64/ole32.dll",
"drive_c/windows/syswow64/oleaut32.dll",
"drive_c/windows/syswow64/wbemprox.dll",
"drive_c/windows/syswow64/wbem/wbemprox.dll",
)
@@ -66,6 +96,20 @@ class LoginServerProbe:
error: str | None
@dataclass(frozen=True)
class RuntimeComponentFingerprint:
privacy: str
runtime_wrapper_sha256: str | None
proton_script_sha256: str | None
proton_version_sha256: str | None
proton_toolmanifest_sha256: str | None
proton_wine_sha256: str | None
proton_wineserver_sha256: str | None
prefix_dll_sha256: dict[str, str | None]
registry_overrides: dict[str, bool]
runtime_marker_matches_selection: bool | None
@dataclass(frozen=True)
class LoginDiagnosticReport:
preflight_ok: bool
@@ -77,20 +121,35 @@ class LoginDiagnosticReport:
kernel_release: str
login_server: LoginServerProbe
authentication_tested: bool
login_response_ownership: str
client_identity_inputs: tuple[str, ...]
client_identity_values_collected: bool
client_identity_shape_collected: bool
client_identity_ready: bool | None
client_identity_privacy: str
client_identity_probe: IdentityProbeResult | None
login_trace_enabled: bool
login_trace_disabled_reason: str | None
publisher_comparison_command: str
install_directory: str
game_exe: str
game_exe_sha256: str | None
game_dll: str
game_dll_sha256: str | None
source_game_exe_sha256: str | None
source_game_dll_sha256: str | None
runtime_matches_managed_source: bool
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
depot_build: int | None
depot_version: str | None
depot_completed_at: str | None
runtime_configured: str
runtime_requested: str
runtime_override_applied: bool
runtime_kind: str | None
runtime_path: str | None
runtime_error: str | None
runtime_components: RuntimeComponentFingerprint | None
proton_sync_configured: str
proton_sync_effective: str | None
prefix: str | None
@@ -143,6 +202,8 @@ def inspect_login_environment(
*,
timeout_seconds: float = DEFAULT_LOGIN_TIMEOUT_SECONDS,
connector: LoginConnector = _create_login_connection,
runner: CommandRunner | None = None,
runtime_override: str | None = None,
) -> LoginDiagnosticReport:
"""Inspect the installed client and its network route without authenticating."""
@@ -150,7 +211,9 @@ def inspect_login_environment(
timeout_seconds=timeout_seconds,
connector=connector,
)
selection, runtime_error = _runtime_selection(settings)
runtime_requested = runtime_override or settings.runtime
diagnostic_settings = replace(settings, runtime=runtime_requested)
selection, runtime_error = _runtime_selection(diagnostic_settings)
distribution_id = detect_distribution_id() or "unknown"
kernel_release = platform.release()
proton_sync_effective = (
@@ -163,22 +226,35 @@ def inspect_login_environment(
and selection.kind in {"proton7", "proton7-container"}
else None
)
game_root = _active_game_root(paths, settings)
game_root = _active_game_root(paths)
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
source_game_exe_sha256 = _sha256_if_file(paths.source_game / "Game.exe")
source_game_dll_sha256 = _sha256_if_file(paths.source_game / "game.dll")
runtime_matches_managed_source = (
game_exe_sha256 is not None
and game_dll_sha256 is not None
and game_exe_sha256 == source_game_exe_sha256
and game_dll_sha256 == source_game_dll_sha256
)
game_profile = classify_game_profile(game_exe_sha256, game_dll_sha256)
depot_state = _read_json_object(paths.native_depot_state)
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)
runtime_components = (
inspect_runtime_components(paths, selection, prefix)
if selection is not None and prefix is not None
else None
)
identity_probe = (
probe_client_identity(paths, selection, runner)
if selection is not None and prefix_ready and runner is not None
else None
)
latest_log = latest_native_game_log(paths)
warnings = _diagnostic_warnings(
probe,
game_profile,
@@ -186,14 +262,23 @@ def inspect_login_environment(
prefix_ready,
directx_ready,
latest_log,
runtime_matches_managed_source,
identity_probe,
)
identity_preflight_ok = (
True
if runner is None
else identity_probe is not None and identity_probe.ready is True
)
return LoginDiagnosticReport(
preflight_ok=(
probe.reachable
and game_profile in VERIFIED_GAME_PROFILES
and runtime_matches_managed_source
and selection is not None
and prefix_ready
and directx_ready
and identity_preflight_ok
),
checked_at=datetime.now().astimezone().isoformat(),
launcher_version=__version__,
@@ -203,22 +288,41 @@ def inspect_login_environment(
kernel_release=kernel_release,
login_server=probe,
authentication_tested=False,
login_response_ownership=SERVER_LOGIN_RESPONSE_OWNERSHIP,
client_identity_inputs=CLIENT_IDENTITY_INPUTS,
client_identity_values_collected=False,
client_identity_shape_collected=(
identity_probe is not None and identity_probe.report is not None
),
client_identity_ready=(
identity_probe.ready if identity_probe is not None else None
),
client_identity_privacy=IDENTITY_PROBE_PRIVACY,
client_identity_probe=identity_probe,
login_trace_enabled=LOGIN_TRACE_ENABLED,
login_trace_disabled_reason=(
None if LOGIN_TRACE_ENABLED else LOGIN_TRACE_DISABLED_REASON
),
publisher_comparison_command=PUBLISHER_COMPARISON_COMMAND,
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,
source_game_exe_sha256=source_game_exe_sha256,
source_game_dll_sha256=source_game_dll_sha256,
runtime_matches_managed_source=runtime_matches_managed_source,
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
),
depot_build=_integer_or_none(depot_state.get("build")),
depot_version=_string_or_none(depot_state.get("version")),
depot_completed_at=_string_or_none(depot_state.get("completed_at")),
runtime_configured=settings.runtime,
runtime_requested=runtime_requested,
runtime_override_applied=runtime_override is not None,
runtime_kind=selection.kind if selection is not None else None,
runtime_path=_runtime_path(selection),
runtime_error=runtime_error,
runtime_components=runtime_components,
proton_sync_configured=settings.proton_sync,
proton_sync_effective=proton_sync_effective,
prefix=str(prefix) if prefix is not None else None,
@@ -230,6 +334,85 @@ def inspect_login_environment(
)
def inspect_runtime_components(
paths: AppPaths,
selection: RuntimeSelection,
prefix: Path,
) -> RuntimeComponentFingerprint:
"""Fingerprint non-private runtime components for side-by-side reports."""
proton_script_sha256: str | None = None
proton_version_sha256: str | None = None
proton_toolmanifest_sha256: str | None = None
proton_wine_sha256: str | None = None
proton_wineserver_sha256: str | None = None
if selection.proton_path:
proton = Path(selection.proton_path)
proton_script_sha256 = _sha256_if_file(proton)
proton_version_sha256 = _sha256_if_file(proton.parent / "version")
proton_toolmanifest_sha256 = _sha256_if_file(
proton.parent / "toolmanifest.vdf"
)
for directory in (proton.parent / "files/bin", proton.parent / "dist/bin"):
wine = directory / "wine"
wineserver = directory / "wineserver"
if wine.is_file() and wineserver.is_file():
proton_wine_sha256 = _sha256_if_file(wine)
proton_wineserver_sha256 = _sha256_if_file(wineserver)
break
prefix_dll_sha256 = {
relative: _sha256_if_file(prefix / relative)
for relative in PREFIX_COMPONENT_PATHS
}
runtime_marker = _read_json_object(paths.game_runtime_marker)
marker_matches = (
runtime_marker.get("runtime") == asdict(selection)
if runtime_marker
else None
)
return RuntimeComponentFingerprint(
privacy=RUNTIME_COMPONENT_PRIVACY,
runtime_wrapper_sha256=_sha256_if_file(Path(selection.executable)),
proton_script_sha256=proton_script_sha256,
proton_version_sha256=proton_version_sha256,
proton_toolmanifest_sha256=proton_toolmanifest_sha256,
proton_wine_sha256=proton_wine_sha256,
proton_wineserver_sha256=proton_wineserver_sha256,
prefix_dll_sha256=prefix_dll_sha256,
registry_overrides=_runtime_registry_override_flags(prefix),
runtime_marker_matches_selection=marker_matches,
)
def _runtime_registry_override_flags(prefix: Path) -> dict[str, bool]:
try:
registry = (prefix / "user.reg").read_text(
encoding="utf-8",
errors="replace",
).casefold()
except OSError:
registry = ""
return {
"d3d9_native_builtin": '"d3d9"="native,builtin"' in registry,
"d3dx9_43_native": _native_override_present(registry, "d3dx9_43"),
"d3dcompiler_43_native": _native_override_present(
registry,
"d3dcompiler_43",
),
}
def _native_override_present(registry: str, name: str) -> bool:
return any(
entry in registry
for entry in (
f'"{name}"="native,builtin"',
f'"{name}"="native"',
f'"*{name}"="native"',
)
)
def classify_game_profile(
game_exe_sha256: str | None,
game_dll_sha256: str | None,
@@ -266,7 +449,7 @@ def classify_game_profile(
return "unknown"
def latest_native_game_log(paths: AppPaths, settings: Settings) -> Path | None:
def latest_native_game_log(paths: AppPaths) -> 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")
@@ -287,7 +470,7 @@ def latest_native_game_log(paths: AppPaths, settings: Settings) -> Path | None:
return None
def _active_game_root(paths: AppPaths, settings: Settings) -> Path:
def _active_game_root(paths: AppPaths) -> Path:
if paths.runtime_view.is_dir():
return paths.runtime_view
return paths.source_game
@@ -331,6 +514,8 @@ def _diagnostic_warnings(
prefix_ready: bool,
directx_ready: bool,
latest_log: Path | None,
runtime_matches_managed_source: bool,
identity_probe: IdentityProbeResult | None,
) -> tuple[str, ...]:
warnings: list[str] = []
if not probe.reachable:
@@ -356,6 +541,11 @@ def _diagnostic_warnings(
"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 not runtime_matches_managed_source:
warnings.append(
"The runtime Game.exe/game.dll pair does not match the managed updater "
"source; run prepare after update --repair before testing login again."
)
if runtime_error is not None:
warnings.append(runtime_error)
elif not prefix_ready:
@@ -364,9 +554,29 @@ def _diagnostic_warnings(
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.")
if identity_probe is not None:
warnings.extend(identity_probe.warnings)
return tuple(warnings)
def _read_json_object(path: Path) -> dict[str, object]:
try:
value = json.loads(path.read_text(encoding="utf-8"))
except (OSError, UnicodeError, json.JSONDecodeError):
return {}
return value if isinstance(value, dict) else {}
def _integer_or_none(value: object) -> int | None:
if isinstance(value, bool) or not isinstance(value, int):
return None
return value
def _string_or_none(value: object) -> str | None:
return value if isinstance(value, str) else None
def _sha256(path: Path) -> str:
digest = hashlib.sha256()
with path.open("rb") as handle:

View File

@@ -1,472 +1,20 @@
"""Privacy-safe instrumentation for the post-authentication login transition."""
"""Retired login instrumentation retained only for command compatibility."""
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,
LOGIN_TRACE_ENABLED = False
LOGIN_TRACE_DISABLED_REASON = (
"login-trace is retired and will not start the game: it enabled Wine relay "
"instrumentation during authentication, while the current client performs "
"anti-debug and module-integrity checks. Use login-diagnostics for a "
"credential-free check, then use wine-update for one controlled comparison "
"through the unmodified publisher launcher."
)
@dataclass(frozen=True)
class RelayTraceResult:
game_dll_base: int | None
target_return_address: int | None
comparison_results: tuple[bool, ...]
log_files: int
def run_login_trace(*_args: object, **_kwargs: object) -> None:
"""Refuse to reactivate the retired relay-based authentication trace."""
@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
raise RuntimeError(LOGIN_TRACE_DISABLED_REASON)

Binary file not shown.

View File

@@ -153,7 +153,13 @@ def fetch_news(
_write_cache(cache_path, articles)
return articles
except (OSError, NewsError, UnicodeDecodeError, json.JSONDecodeError):
return _read_cache(cache_path)
return read_cached_news(cache_path)
def read_cached_news(cache_path: Path) -> tuple[NewsArticle, ...]:
"""Read the last valid news feed without performing network I/O."""
return _read_cache(cache_path)
def fetch_latest_broadcast(
@@ -179,7 +185,13 @@ def fetch_latest_broadcast(
_write_broadcast_cache(cache_path, broadcast)
return broadcast
except (OSError, NewsError, UnicodeDecodeError, json.JSONDecodeError):
return _read_broadcast_cache(cache_path)
return read_cached_broadcast(cache_path)
def read_cached_broadcast(cache_path: Path) -> Broadcast | None:
"""Read the last valid broadcast without performing network I/O."""
return _read_broadcast_cache(cache_path)
def _parse_news(payload: bytes) -> tuple[NewsArticle, ...]:
@@ -288,7 +300,7 @@ def _read_cache(path: Path) -> tuple[NewsArticle, ...]:
if not isinstance(value, list):
return ()
return tuple(NewsArticle(**item) for item in value if isinstance(item, dict))
except (OSError, TypeError, json.JSONDecodeError):
except (OSError, UnicodeError, TypeError, json.JSONDecodeError):
return ()
@@ -309,7 +321,7 @@ def _read_broadcast_cache(path: Path) -> Broadcast | None:
try:
value = json.loads(path.read_text(encoding="utf-8"))
return Broadcast(**value) if isinstance(value, dict) else None
except (OSError, TypeError, json.JSONDecodeError):
except (OSError, UnicodeError, TypeError, json.JSONDecodeError):
return None

View File

@@ -19,7 +19,6 @@ from ptdiag.assbin import (
)
from .paths import AppPaths
from .world_fix import V21_BINARY_HASHES
WHISPERING_VALE_RELATIVE = Path("game/maps/WhisperingVale.rovug")
@@ -35,10 +34,17 @@ COMPATIBILITY_BINARY_HASHES = {
"Game.exe": "3cc0980aac0f65b379ff340ec61750498fb9f620fb6c88aa66281e8ab27023a5",
"game.dll": "7ea85b19f452962a39c91f967cad22495f066b2d52e8ac37f4e234463d5c4f09",
}
CURRENT_V21_BINARY_HASHES = V21_BINARY_HASHES
CURRENT_V21_BINARY_HASHES = {
"Game.exe": "2bc3edd26c5f6d7ae9bbdc23967c5aefd374bdf934fcdfa2ed42c6442e550837",
"game.dll": "4b4e5c28e2bfdaa8d36c8341fad27becfd03faf63aa8ea874f5d43c5b1961868",
}
PATCHED_V21_GAME_DLL_SHA256 = (
"0be11b06ccf9be747e32b195c8d5980062f446d7d3a0a662ebfe0e602a01c538"
)
RETIRED_POST_AUTH_GAME_DLL_SHA256 = (
"93cd27284eb294753da1111c99bfc9689f82b6d4a59302aee9fdbd8cb52452dd"
)
RUNTIME_VIEW_CONFIGURATION_VERSION = 1
class OverlayError(RuntimeError):
@@ -83,8 +89,6 @@ class RuntimeCompatibility:
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.
@@ -105,9 +109,6 @@ def inspect_runtime_compatibility(
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__
@@ -132,8 +133,6 @@ def inspect_runtime_compatibility(
def prepare_runtime_view(
paths: AppPaths,
source_game: Path,
*,
use_world_fix: bool = True,
) -> OverlayResult:
"""Build a symlink view and keep the launcher-managed source read-only."""
@@ -144,13 +143,28 @@ def prepare_runtime_view(
inspection = inspect_wrapped_assbin(source_scene)
action = _scene_action(inspection)
game_save = _prepare_persistent_save(paths, source)
# ``use_world_fix`` is retained as a compatibility keyword. The retired
# helper is never selected and no code path may replace the publisher DLL.
_ = use_world_fix
binary_action, game_exe_source, game_dll_source = _runtime_game_binary_sources(
(
binary_action,
game_exe_source,
game_dll_source,
source_binary_hashes,
) = _runtime_game_binary_profile(
paths,
source,
)
reused = _reuse_runtime_view(
paths,
source,
inspection,
action,
game_save,
binary_action,
game_exe_source,
game_dll_source,
source_binary_hashes,
)
if reused is not None:
return reused
build = paths.data / f".{paths.runtime_view.name}.build-{uuid.uuid4().hex}"
backup = paths.data / f".{paths.runtime_view.name}.old-{uuid.uuid4().hex}"
@@ -223,7 +237,13 @@ def prepare_runtime_view(
created_at=datetime.now().astimezone().isoformat(),
links_created=links_created,
)
_write_json(paths.runtime_manifest, asdict(result))
_write_json(
paths.runtime_manifest,
{
"configuration_version": RUNTIME_VIEW_CONFIGURATION_VERSION,
**asdict(result),
},
)
return result
@@ -271,6 +291,32 @@ def select_binary_action(
return "verified-source-pair"
if source_hashes == CURRENT_V21_BINARY_HASHES:
return "verified-v21-publisher-pair"
game_exe_sha256 = source_hashes.get("Game.exe")
game_dll_sha256 = source_hashes.get("game.dll")
if game_dll_sha256 in {
PATCHED_V21_GAME_DLL_SHA256,
RETIRED_POST_AUTH_GAME_DLL_SHA256,
}:
raise OverlayError(
"a retired experimental game.dll is installed; run update --repair "
"before launching"
)
known_game_exe_hashes = {
CURRENT_V21_BINARY_HASHES["Game.exe"],
COMPATIBILITY_BINARY_HASHES["Game.exe"],
}
known_game_dll_hashes = {
CURRENT_V21_BINARY_HASHES["game.dll"],
COMPATIBILITY_BINARY_HASHES["game.dll"],
}
if (
game_exe_sha256 in known_game_exe_hashes
or game_dll_sha256 in known_game_dll_hashes
):
raise OverlayError(
"Game.exe and game.dll are from different verified releases; run "
"update --repair before launching"
)
return "upstream-pair"
@@ -392,13 +438,14 @@ def _prepare_persistent_save(paths: AppPaths, source: Path) -> Path:
candidate = path.resolve(strict=True)
break
if candidate is not None:
selected = candidate
allowed_roots = (paths.data.resolve(strict=False), source.resolve(strict=True))
if not any(
candidate == root or candidate.is_relative_to(root)
selected == root or selected.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)
raise OverlayError(f"refusing to migrate game save outside managed roots: {selected}")
_reject_save_symlinks(selected)
target.parent.mkdir(parents=True, exist_ok=True)
staging = target.parent / f".{target.name}.build-{uuid.uuid4().hex}"
@@ -423,10 +470,213 @@ def _reject_save_symlinks(root: Path) -> None:
raise OverlayError(f"persistent game save contains a symbolic link: {path}")
def _reuse_runtime_view(
paths: AppPaths,
source: Path,
inspection: SceneInspection,
scene_action: str,
game_save: Path,
binary_action: str,
game_exe_source: Path,
game_dll_source: Path,
source_binary_hashes: Mapping[str, str],
) -> OverlayResult | None:
result = _read_overlay_result(paths.runtime_manifest)
if result is None:
return None
if result.source_game != str(source):
return None
if result.runtime_view != str(paths.runtime_view):
return None
if result.source_scene_sha256 != inspection.file_sha256:
return None
if result.scene_action != scene_action or result.binary_action != binary_action:
return None
if result.game_exe_sha256 != source_binary_hashes.get("Game.exe"):
return None
if result.game_dll_sha256 != source_binary_hashes.get("game.dll"):
return None
if not _runtime_layout_ready(
paths,
source,
game_save,
game_exe_source,
game_dll_source,
result.links_created,
):
return None
runtime_scene = paths.runtime_view / WHISPERING_VALE_RELATIVE
try:
if _sha256(runtime_scene) != result.runtime_scene_sha256:
return None
except OSError:
return None
return result
def _runtime_layout_ready(
paths: AppPaths,
source: Path,
game_save: Path,
game_exe_source: Path,
game_dll_source: Path,
expected_links: int,
) -> bool:
runtime = paths.runtime_view
if runtime.is_symlink() or not runtime.is_dir():
return False
top_level_exceptions = {"game", "save"}
top_level_special = {"game", "save"}
for name, binary_source in (
("Game.exe", game_exe_source),
("game.dll", game_dll_source),
):
if binary_source != source / name:
top_level_exceptions.add(name)
top_level_special.add(name)
top_level_ready, top_level_links = _linked_level_ready(
source,
runtime,
except_names=top_level_exceptions,
special_names=top_level_special,
)
if not top_level_ready:
return False
if not _symlink_points_to(runtime / "save", game_save):
return False
if game_exe_source != source / "Game.exe" and not _symlink_points_to(
runtime / "Game.exe",
game_exe_source,
):
return False
if game_dll_source != source / "game.dll" and not _symlink_points_to(
runtime / "game.dll",
game_dll_source,
):
return False
game_source = source / "game"
game_target = runtime / "game"
if game_target.is_symlink() or not game_target.is_dir():
return False
game_ready, game_links = _linked_level_ready(
game_source,
game_target,
except_names={"maps"},
special_names={"maps"},
)
if not game_ready:
return False
maps_source = game_source / "maps"
maps_target = game_target / "maps"
if maps_target.is_symlink() or not maps_target.is_dir():
return False
maps_ready, maps_links = _linked_level_ready(
maps_source,
maps_target,
except_names={"WhisperingVale.rovug"},
special_names={"WhisperingVale.rovug"},
)
if not maps_ready:
return False
runtime_scene = maps_target / "WhisperingVale.rovug"
if runtime_scene.is_symlink() or not runtime_scene.is_file():
return False
actual_links = top_level_links + game_links + maps_links + 1
if game_exe_source != source / "Game.exe":
actual_links += 1
if game_dll_source != source / "game.dll":
actual_links += 1
return actual_links == expected_links
def _linked_level_ready(
source: Path,
target: Path,
*,
except_names: set[str],
special_names: set[str],
) -> tuple[bool, int]:
try:
source_children = {
child.name: child
for child in source.iterdir()
if child.name not in except_names
}
target_names = {child.name for child in target.iterdir()}
except OSError:
return False, 0
if target_names != set(source_children) | special_names:
return False, 0
for name, child in source_children.items():
if not _symlink_points_to(target / name, child):
return False, 0
return True, len(source_children)
def _symlink_points_to(link: Path, expected_target: Path) -> bool:
if not link.is_symlink():
return False
try:
return link.resolve(strict=True) == expected_target.resolve(strict=True)
except (OSError, RuntimeError):
return False
def _read_overlay_result(path: Path) -> OverlayResult | None:
try:
value = json.loads(path.read_text(encoding="utf-8"))
except (OSError, UnicodeError, json.JSONDecodeError):
return None
if not isinstance(value, dict):
return None
if value.get("configuration_version") != RUNTIME_VIEW_CONFIGURATION_VERSION:
return None
string_fields = (
"source_game",
"runtime_view",
"source_scene_sha256",
"runtime_scene_sha256",
"scene_action",
"binary_action",
"game_exe_sha256",
"game_dll_sha256",
"created_at",
)
if not all(isinstance(value.get(name), str) for name in string_fields):
return None
links_created = value.get("links_created")
if isinstance(links_created, bool) or not isinstance(links_created, int):
return None
if links_created < 0:
return None
return OverlayResult(
source_game=value["source_game"],
runtime_view=value["runtime_view"],
source_scene_sha256=value["source_scene_sha256"],
runtime_scene_sha256=value["runtime_scene_sha256"],
scene_action=value["scene_action"],
binary_action=value["binary_action"],
game_exe_sha256=value["game_exe_sha256"],
game_dll_sha256=value["game_dll_sha256"],
created_at=value["created_at"],
links_created=links_created,
)
def _runtime_game_binary_sources(
paths: AppPaths,
source: Path,
) -> tuple[str, Path, Path]:
action, game_exe, game_dll, _ = _runtime_game_binary_profile(paths, source)
return action, game_exe, game_dll
def _runtime_game_binary_profile(
paths: AppPaths,
source: Path,
) -> tuple[str, Path, Path, dict[str, str]]:
cache = paths.compatibility_binaries
cache_hashes = _binary_hashes(cache) if cache.is_dir() else None
source_hashes = _binary_hashes(source)
@@ -435,6 +685,7 @@ def _runtime_game_binary_sources(
action,
source / "Game.exe",
source / "game.dll",
source_hashes,
)

View File

@@ -20,7 +20,13 @@ def _xdg_path(
fallback: Path,
) -> Path:
value = environment.get(variable)
return Path(value).expanduser() if value else fallback
if not value:
return fallback
path = Path(value).expanduser()
# The XDG Base Directory specification requires absolute paths. Ignoring an
# invalid relative value keeps settings anchored to HOME instead of making
# their location depend on the launcher's current working directory.
return path if path.is_absolute() else fallback
@dataclass(frozen=True)
@@ -91,6 +97,12 @@ class AppPaths:
def launcher_update_lock(self) -> Path:
return self.state / "launcher-update.lock"
@property
def launcher_update_check_cache(self) -> Path:
"""Display-only cache of the last authenticated update check."""
return self.cache / "launcher-update-check.json"
def launcher_update_archive(self, version: str) -> Path:
return self.downloads / f"pristontale-eu-linux-update-v{version}.tar.gz"
@@ -118,6 +130,12 @@ class AppPaths:
def launcher_operation_lock(self) -> Path:
return self.state / "launcher-operation.lock"
@property
def game_runtime_marker(self) -> Path:
"""Verified Proton/D3D preparation state for the game prefix."""
return self.state / "game-runtime.json"
@property
def source_game(self) -> Path:
return self.launcher_prefix / GAME_RELATIVE_PATH
@@ -126,16 +144,34 @@ class AppPaths:
def game_import_manifest(self) -> Path:
return self.state / "game-import.json"
@property
def game_import_quarantine(self) -> Path:
"""Recoverable files excluded while authenticating an imported game."""
return self.data / "import-quarantine"
@property
def native_depot_state(self) -> Path:
"""Last completed native depot synchronization."""
return self.state / "native-depot.json"
@property
def depot_launch_catalog_cache(self) -> Path:
"""Last official realm choices used for network-free game launches."""
return self.cache / "depot-launch-catalog.json"
@property
def news_cache(self) -> Path:
return self.cache / "news.json"
@property
def patches_cache(self) -> Path:
"""Cached official patch changelog shown beside the news feed."""
return self.cache / "patches.json"
@property
def broadcast_cache(self) -> Path:
"""Cached latest announcement from the broadcast service."""
@@ -226,6 +262,10 @@ class AppPaths:
def settings_file(self) -> Path:
return self.config / "settings.json"
@property
def settings_lock(self) -> Path:
return self.config / "settings.lock"
def create_base_directories(self) -> None:
for path in (
self.data,

View File

@@ -4,26 +4,25 @@ from __future__ import annotations
from dataclasses import asdict, dataclass
from datetime import datetime
import hashlib
import json
import os
from pathlib import Path
from pathlib import Path, PurePosixPath
import platform
import shlex
import shutil
import sys
from typing import Mapping, Sequence
from .command import CommandOutcome, CommandRunner, OutputLineFilter
from .depot import DEFAULT_DEPOT_START_OPTION, DepotStartOption
from .gamebox import build_gamebox_command
from .paths import AppPaths
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_DIRECTORY = r"C:\Program Files (x86)\Pristontale EU"
WINDOWS_GAME_EXE = rf"{WINDOWS_GAME_DIRECTORY}\Game.exe"
NATIVE_D3D43_DLLS = ("d3dx9_43", "d3dcompiler_43")
DXVK_D3D9_OVERRIDE = "native,builtin"
DLL_OVERRIDES_KEY = r"HKCU\Software\Wine\DllOverrides"
@@ -33,6 +32,10 @@ WINETRICKS_CODEPAGE_PROBE = (
WINETRICKS_PROTON7_PROBE = 'winetricks_early_wine_arch cmd.exe /c "echo %$1%"'
RUNTIME_DEFAULT_SYNC = "runtime-default"
CACHYOS_MARKER = "cachyos"
CURRENT_UMU_DISTRIBUTIONS = frozenset(
{"arch", "cachyos", "endeavouros", "garuda", "manjaro"}
)
GAME_RUNTIME_CONFIGURATION_VERSION = 1
class RuntimeErrorPT(RuntimeError):
@@ -50,10 +53,30 @@ class RuntimeSelection:
def select_runtime(
settings: Settings,
environment: Mapping[str, str] | None = None,
*,
distribution_id: str | None = None,
kernel_release: str | None = None,
) -> RuntimeSelection:
env = os.environ if environment is None else environment
umu = shutil.which("umu-run", path=env.get("PATH"))
if settings.proton_path:
current_umu_host = prefers_current_umu_host(
distribution_id=distribution_id,
kernel_release=kernel_release,
)
auto_container = (
settings.runtime == "auto"
and bool(umu)
and not current_umu_host
)
if settings.runtime == "umu":
if umu:
return RuntimeSelection("umu", umu, None, None)
raise RuntimeErrorPT("UMU was requested but umu-run was not found")
if settings.proton_path and settings.runtime in {
"auto",
"proton7",
"proton7-container",
}:
proton = Path(settings.proton_path).expanduser().resolve()
if not proton.is_file():
raise RuntimeErrorPT(f"configured Proton script does not exist: {proton}")
@@ -69,7 +92,8 @@ def select_runtime(
str(proton),
str(steam_root),
)
if settings.runtime == "auto" and umu:
if auto_container:
assert umu is not None
return RuntimeSelection(
"proton7-container",
umu,
@@ -77,13 +101,17 @@ def select_runtime(
str(steam_root),
)
return RuntimeSelection("proton7", sys.executable, str(proton), str(steam_root))
# Current UMU avoids several rolling-distribution container/media failures and
# remains the broad Arch-family compatibility default. Direct Proton 7 is also
# known to authenticate on CachyOS, so this is a compatibility policy rather
# than an assertion that Proton 7 itself causes server-side Code 1/Code 3.
if settings.runtime == "auto" and current_umu_host and umu:
return RuntimeSelection("umu", umu, None, None)
if settings.runtime in {"auto", "proton7", "proton7-container"}:
proton_result = find_steam_proton7(env)
if proton_result is not None:
proton, steam_root = proton_result
if settings.runtime == "proton7-container" or (
settings.runtime == "auto" and umu
):
if settings.runtime == "proton7-container" or auto_container:
if not umu:
raise RuntimeErrorPT(
"containerized Proton 7 was requested but umu-run was not found"
@@ -159,6 +187,8 @@ def prepare_game_runtime(
elif selection.kind in {"proton7", "proton7-container"}:
if not selection.proton_path or not selection.steam_root:
raise RuntimeErrorPT("incomplete Proton 7 runtime selection")
if not runner.dry_run and _game_runtime_ready(paths, selection):
return results
if not runner.dry_run:
paths.compat_data.mkdir(parents=True, exist_ok=True)
proton = Path(selection.proton_path)
@@ -237,26 +267,37 @@ def prepare_game_runtime(
raise RuntimeErrorPT(f"unsupported runtime kind: {selection.kind}")
if not runner.dry_run:
_write_json(
paths.state / "game-runtime.json",
paths.game_runtime_marker,
{
"configuration_version": GAME_RUNTIME_CONFIGURATION_VERSION,
"configured_at": datetime.now().astimezone().isoformat(),
"native_d3d43": True,
"dxvk_d3d9_override": DXVK_D3D9_OVERRIDE,
"proton_identity": _proton_identity(selection),
"runtime": asdict(selection),
},
)
return results
def game_runtime_ready(paths: AppPaths, selection: RuntimeSelection) -> bool:
"""Return whether another client may reuse the prepared runtime read-only."""
if selection.kind == "umu":
return native_d3d_ready(paths.umu_prefix)
if selection.kind in {"proton7", "proton7-container"}:
return _game_runtime_ready(paths, selection)
return False
def run_game(
paths: AppPaths,
settings: Settings,
selection: RuntimeSelection,
runner: CommandRunner,
*,
start_option: DepotStartOption = DEFAULT_DEPOT_START_OPTION,
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,
@@ -264,38 +305,25 @@ def run_game(
) -> CommandOutcome:
if not paths.runtime_view.is_dir() and not runner.dry_run:
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")
native_executable, windows_executable, game_arguments = _game_start_command(
paths,
start_option,
)
if selection.kind == "umu":
game_command = [
selection.executable,
str(launch_executable),
native_executable,
*game_arguments,
]
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)
elif selection.kind == "proton7-container":
if not selection.proton_path:
raise RuntimeErrorPT("Proton script path is missing")
game_command = [
selection.executable,
str(launch_executable),
native_executable,
*game_arguments,
]
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,
@@ -310,14 +338,9 @@ def run_game(
"ptlinux.proton_compat",
selection.proton_path,
"runinprefix",
str(world_fix_helper) if world_fix_helper is not None else WINDOWS_GAME_EXE,
windows_executable,
*game_arguments,
]
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,
@@ -341,6 +364,57 @@ def run_game(
)
def _game_start_command(
paths: AppPaths,
option: DepotStartOption,
) -> tuple[str, str, tuple[str, ...]]:
relative = PurePosixPath(option.path)
if (
relative.is_absolute()
or not relative.parts
or any(part in {"", ".", ".."} for part in relative.parts)
or relative.suffix.casefold() != ".exe"
):
raise RuntimeErrorPT(f"invalid realm executable path: {option.path!r}")
try:
arguments = tuple(shlex.split(option.arguments))
except ValueError as exc:
raise RuntimeErrorPT(
f"invalid launch arguments for realm {option.name}: {exc}"
) from exc
if not arguments:
raise RuntimeErrorPT(f"realm {option.name} has no launch arguments")
native = str(_resolve_runtime_path_case(paths.runtime_view, relative))
windows_relative = "\\".join(relative.parts)
return native, rf"{WINDOWS_GAME_DIRECTORY}\{windows_relative}", arguments
def _resolve_runtime_path_case(root: Path, relative: PurePosixPath) -> Path:
"""Match Windows manifest casing against the case-sensitive runtime view."""
direct = root.joinpath(*relative.parts)
if direct.exists():
return direct
cursor = root
for part in relative.parts:
if not cursor.is_dir():
return direct
try:
matches = tuple(
entry for entry in cursor.iterdir() if entry.name.casefold() == part.casefold()
)
except OSError:
return direct
if not matches:
return direct
if len(matches) > 1:
raise RuntimeErrorPT(
f"realm executable path has ambiguous casing: {relative.as_posix()}"
)
cursor = matches[0]
return cursor
def run_proton_prefix_command(
paths: AppPaths,
selection: RuntimeSelection,
@@ -371,6 +445,49 @@ def run_proton_prefix_command(
)
def run_runtime_prefix_command(
paths: AppPaths,
selection: RuntimeSelection,
runner: CommandRunner,
arguments: Sequence[str],
*,
check: bool = True,
timeout_seconds: float = 60.0,
) -> CommandOutcome:
"""Run a credential-free helper inside the selected game's Wine prefix."""
if not arguments:
raise ValueError("runtime-prefix command arguments cannot be empty")
if selection.kind == "umu":
command = [selection.executable, *arguments]
environment = _umu_environment(paths)
elif selection.kind == "proton7-container":
if not selection.proton_path or not selection.steam_root:
raise RuntimeErrorPT("incomplete Proton 7 runtime selection")
command = [selection.executable, *arguments]
environment = _umu_proton7_environment(paths, selection)
elif selection.kind == "proton7":
if not selection.proton_path or not selection.steam_root:
raise RuntimeErrorPT("incomplete Proton 7 runtime selection")
command = [
sys.executable,
"-m",
"ptlinux.proton_compat",
selection.proton_path,
"runinprefix",
*arguments,
]
environment = _proton_environment(paths, selection)
else:
raise RuntimeErrorPT(f"unsupported runtime kind: {selection.kind}")
return runner.run(
command,
env_overrides=environment,
timeout_seconds=timeout_seconds,
check=check,
)
def _runtime_python_path(existing: str | None) -> str:
"""Make the installed package importable after changing into runtime_view."""
package_root = str(Path(__file__).resolve().parent.parent)
@@ -449,6 +566,38 @@ def detect_distribution_id() -> str:
return value.strip().casefold()
def is_cachyos_host(
*,
distribution_id: str | None = None,
kernel_release: str | None = None,
) -> bool:
"""Identify CachyOS narrowly for runtime compatibility policy."""
distribution = detect_distribution_id() if distribution_id is None else distribution_id
kernel = platform.release() if kernel_release is None else kernel_release
return (
CACHYOS_MARKER in distribution.casefold()
or CACHYOS_MARKER in kernel.casefold()
)
def prefers_current_umu_host(
*,
distribution_id: str | None = None,
kernel_release: str | None = None,
) -> bool:
"""Use current UMU automatically on rolling Arch-family hosts."""
distribution = detect_distribution_id() if distribution_id is None else distribution_id
return (
distribution.strip().casefold() in CURRENT_UMU_DISTRIBUTIONS
or is_cachyos_host(
distribution_id=distribution,
kernel_release=kernel_release,
)
)
def resolve_proton_sync_mode(
configured: str,
*,
@@ -461,9 +610,10 @@ def resolve_proton_sync_mode(
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():
if is_cachyos_host(
distribution_id=distribution_id,
kernel_release=kernel_release,
):
return "esync"
return RUNTIME_DEFAULT_SYNC
@@ -546,21 +696,23 @@ def prepare_proton7_winetricks(
dry_run: bool = False,
) -> Path:
try:
content = source.read_text(encoding="utf-8")
content = source.read_bytes()
except OSError as exc:
raise RuntimeErrorPT(f"could not read Winetricks script {source}: {exc}") from exc
if WINETRICKS_CODEPAGE_PROBE not in content:
probe = WINETRICKS_CODEPAGE_PROBE.encode("utf-8")
replacement = WINETRICKS_PROTON7_PROBE.encode("utf-8")
if probe not in content:
return source
if dry_run:
return destination
destination.parent.mkdir(parents=True, exist_ok=True)
patched = content.replace(
WINETRICKS_CODEPAGE_PROBE,
WINETRICKS_PROTON7_PROBE,
probe,
replacement,
1,
)
temporary = destination.with_name(f".{destination.name}.{os.getpid()}.tmp")
temporary.write_text(patched, encoding="utf-8")
temporary.write_bytes(patched)
temporary.chmod(source.stat().st_mode & 0o777)
temporary.replace(destination)
return destination
@@ -572,6 +724,69 @@ def _proton_prefix_ready(prefix: Path) -> bool:
).exists()
def _game_runtime_ready(paths: AppPaths, selection: RuntimeSelection) -> bool:
marker = _read_json(paths.game_runtime_marker)
identity = _proton_identity(selection)
if marker is None or identity is None:
return False
if marker.get("configuration_version") != GAME_RUNTIME_CONFIGURATION_VERSION:
return False
if marker.get("runtime") != asdict(selection):
return False
if marker.get("proton_identity") != identity:
return False
if marker.get("native_d3d43") is not True:
return False
if marker.get("dxvk_d3d9_override") != DXVK_D3D9_OVERRIDE:
return False
if not _proton_prefix_ready(paths.proton_prefix):
return False
if not native_d3d_ready(paths.proton_prefix):
return False
if not _d3d9_override_ready(paths.proton_prefix):
return False
return _proton_game_link_ready(paths)
def _d3d9_override_ready(prefix: Path) -> bool:
try:
registry = (prefix / "user.reg").read_text(
encoding="utf-8",
errors="replace",
).casefold()
except OSError:
return False
return f'"d3d9"="{DXVK_D3D9_OVERRIDE}"' in registry
def _proton_game_link_ready(paths: AppPaths) -> bool:
link = paths.proton_prefix / "drive_c/Program Files (x86)/Pristontale EU"
if not link.is_symlink():
return False
try:
return link.resolve(strict=True) == paths.runtime_view.resolve(strict=True)
except (OSError, RuntimeError):
return False
def _proton_identity(selection: RuntimeSelection) -> dict[str, str] | None:
if not selection.proton_path:
return None
proton = Path(selection.proton_path)
identity_paths = {
"proton": proton,
"version": proton.parent / "version",
"toolmanifest": proton.parent / "toolmanifest.vdf",
}
identity: dict[str, str] = {}
for name, path in identity_paths.items():
try:
identity[name] = _sha256(path)
except OSError:
return None
return identity
def _find_proton_wine_tools(proton: Path) -> tuple[Path, Path] | None:
for directory in (proton.parent / "files/bin", proton.parent / "dist/bin"):
wine = directory / "wine"
@@ -586,3 +801,19 @@ def _write_json(path: Path, value: object) -> None:
temporary = path.with_name(f".{path.name}.{os.getpid()}.tmp")
temporary.write_text(json.dumps(value, indent=2, sort_keys=True) + "\n", encoding="utf-8")
temporary.replace(path)
def _read_json(path: Path) -> dict[str, object] | None:
try:
value = json.loads(path.read_text(encoding="utf-8"))
except (OSError, UnicodeError, json.JSONDecodeError):
return None
return value if isinstance(value, dict) else None
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()

View File

@@ -45,6 +45,7 @@ UPDATE_ARCHIVE_PREFIX = "pristontale-eu-linux-update-v"
_VERSION_PATTERN = re.compile(r"(?:0|[1-9][0-9]*)\.(?:0|[1-9][0-9]*)\.(?:0|[1-9][0-9]*)\Z")
_SHA256_PATTERN = re.compile(r"[0-9a-f]{64}\Z")
_ALLOWED_RELEASE_ROOTS = {"ptlinux", "ptdiag", "release.json"}
UPDATE_CHECK_CACHE_SCHEMA = 1
class LauncherUpdateError(RuntimeError):
@@ -84,7 +85,20 @@ class LauncherUpdateResult:
previous_release: str | None
archive: str | None
archive_sha256: str | None
restarted_required: bool
restart_required: bool
state_recorded: bool
warning: str | None
@dataclass(frozen=True)
class LauncherUpdateVerification:
current_version: str
release_version: str
manifest_url: str
archive_url: str
archive_size: int
archive_sha256: str
payload_files: int
@dataclass(frozen=True)
@@ -143,6 +157,89 @@ def check_for_launcher_update(
)
_verify_manifest_signature(manifest_bytes, signature_bytes, key)
manifest = parse_update_manifest(manifest_bytes)
return _evaluate_launcher_update(manifest, current_version, url)
def read_cached_launcher_update_check(
cache_path: Path,
*,
current_version: str = __version__,
manifest_url: str | None = None,
public_key_path: Path | None = None,
environment: Mapping[str, str] | None = None,
) -> LauncherUpdateCheck | None:
"""Read a display-only result from the last authenticated live check."""
url = manifest_url or configured_manifest_url(environment)
key = public_key_path or configured_public_key_path(environment)
try:
_parse_version(current_version, "current launcher version")
_require_https_url(url, "update manifest URL")
key_sha256 = _sha256_file(key.expanduser().resolve())
value = json.loads(cache_path.read_text(encoding="utf-8"))
except (
OSError,
UnicodeError,
json.JSONDecodeError,
LauncherUpdateError,
RuntimeError,
):
return None
if not isinstance(value, dict):
return None
if value.get("schema") != UPDATE_CHECK_CACHE_SCHEMA:
return None
if value.get("current_version") != current_version:
return None
if value.get("manifest_url") != url:
return None
if value.get("public_key_sha256") != key_sha256:
return None
manifest_value = value.get("manifest")
if not isinstance(manifest_value, dict):
return None
try:
manifest = parse_update_manifest(
(json.dumps(manifest_value, sort_keys=True) + "\n").encode("utf-8")
)
except (LauncherUpdateError, TypeError, ValueError):
return None
return _evaluate_launcher_update(manifest, current_version, url)
def cache_launcher_update_check(
cache_path: Path,
check: LauncherUpdateCheck,
*,
public_key_path: Path | None = None,
environment: Mapping[str, str] | None = None,
) -> bool:
"""Cache an authenticated result for startup display, never for installation."""
key = public_key_path or configured_public_key_path(environment)
try:
key_sha256 = _sha256_file(key.expanduser().resolve())
_write_json(
cache_path,
{
"schema": UPDATE_CHECK_CACHE_SCHEMA,
"checked_at": datetime.now().astimezone().isoformat(),
"current_version": check.current_version,
"manifest_url": check.manifest_url,
"public_key_sha256": key_sha256,
"manifest": asdict(check.manifest),
},
)
except (OSError, RuntimeError):
return False
return True
def _evaluate_launcher_update(
manifest: UpdateManifest,
current_version: str,
manifest_url: str,
) -> LauncherUpdateCheck:
current = _parse_version(current_version, "current launcher version")
latest = _parse_version(manifest.version, "manifest version")
minimum = _parse_version(
@@ -170,7 +267,7 @@ def check_for_launcher_update(
available=available,
supported=supported,
reason=reason,
manifest_url=url,
manifest_url=manifest_url,
manifest=manifest,
)
@@ -178,6 +275,7 @@ def check_for_launcher_update(
def install_launcher_update(
paths: AppPaths,
*,
current_version: str = __version__,
manifest_url: str | None = None,
public_key_path: Path | None = None,
environment: Mapping[str, str] | None = None,
@@ -186,6 +284,7 @@ def install_launcher_update(
"""Authenticate, stage, validate, and atomically activate one update."""
check = check_for_launcher_update(
current_version=current_version,
manifest_url=manifest_url,
public_key_path=public_key_path,
environment=environment,
@@ -200,7 +299,9 @@ def install_launcher_update(
previous_release=None,
archive=None,
archive_sha256=None,
restarted_required=False,
restart_required=False,
state_recorded=False,
warning=None,
)
if not check.supported:
raise LauncherUpdateError(check.reason)
@@ -215,20 +316,26 @@ def install_launcher_update(
)
release_directory = _stage_and_install_release(paths, archive, check.manifest)
previous = _activate_release(paths, release_directory)
_write_json(
paths.launcher_update_state,
{
"schema": 1,
"updated_at": datetime.now().astimezone().isoformat(),
"from_version": check.current_version,
"to_version": check.manifest.version,
"manifest_url": check.manifest_url,
"archive_url": check.manifest.archive_url,
"archive_sha256": check.manifest.archive_sha256,
"release_directory": str(release_directory),
"previous_release": str(previous) if previous else None,
},
)
state_warning = None
try:
_write_json(
paths.launcher_update_state,
{
"schema": 1,
"updated_at": datetime.now().astimezone().isoformat(),
"from_version": check.current_version,
"to_version": check.manifest.version,
"manifest_url": check.manifest_url,
"archive_url": check.manifest.archive_url,
"archive_sha256": check.manifest.archive_sha256,
"release_directory": str(release_directory),
"previous_release": str(previous) if previous else None,
},
)
except OSError as exc:
# Activation is the durable user-facing operation. A diagnostics-only
# state record must not strand the dashboard on the old release.
state_warning = f"launcher updated, but update history could not be saved: {exc}"
return LauncherUpdateResult(
changed=True,
current_version=check.current_version,
@@ -237,7 +344,57 @@ def install_launcher_update(
previous_release=str(previous) if previous else None,
archive=str(archive),
archive_sha256=check.manifest.archive_sha256,
restarted_required=True,
restart_required=True,
state_recorded=state_warning is None,
warning=state_warning,
)
def verify_launcher_update_channel(
*,
current_version: str = __version__,
manifest_url: str | None = None,
public_key_path: Path | None = None,
environment: Mapping[str, str] | None = None,
timeout_seconds: float = 60.0,
) -> LauncherUpdateVerification:
"""Authenticate, download, extract, and import a release without activating it."""
check = check_for_launcher_update(
current_version=current_version,
manifest_url=manifest_url,
public_key_path=public_key_path,
environment=environment,
timeout_seconds=min(timeout_seconds, 30.0),
)
with tempfile.TemporaryDirectory(prefix="ptlinux-update-verification-") as temporary:
root = Path(temporary)
archive = root / f"{UPDATE_ARCHIVE_PREFIX}{check.manifest.version}.tar.gz"
_download_archive(
check.manifest,
archive,
timeout_seconds=timeout_seconds,
)
staging = root / "release"
staging.mkdir(mode=0o700)
_extract_release_archive(archive, staging, check.manifest.version)
staged_version = _validate_release_directory(staging)
if staged_version != check.manifest.version:
raise LauncherUpdateError(
f"verified launcher version is {staged_version}, "
f"expected {check.manifest.version}"
)
_self_check_release(staging, check.manifest.version)
_validate_release_directory(staging)
payload_files = sum(1 for item in staging.rglob("*") if item.is_file())
return LauncherUpdateVerification(
current_version=current_version,
release_version=check.manifest.version,
manifest_url=check.manifest_url,
archive_url=check.manifest.archive_url,
archive_size=check.manifest.archive_size,
archive_sha256=check.manifest.archive_sha256,
payload_files=payload_files,
)
@@ -313,6 +470,7 @@ def parse_update_manifest(data: bytes) -> UpdateManifest:
f"update archive size must be between 1 and {MAX_ARCHIVE_BYTES} bytes"
)
published_at = _required_string(value, "published_at")
_parse_published_at(published_at)
notes_value = value.get("notes_url")
if notes_value is not None and not isinstance(notes_value, str):
raise LauncherUpdateError("notes_url must be a string or null")
@@ -344,6 +502,8 @@ def _fetch_small_file(
"User-Agent": USER_AGENT,
"Accept": "application/octet-stream, application/json",
"Accept-Encoding": "identity",
"Cache-Control": "no-cache",
"Pragma": "no-cache",
},
method="GET",
)
@@ -354,7 +514,13 @@ def _fetch_small_file(
except (URLError, OSError) as exc:
raise LauncherUpdateError(f"{label} request failed: {exc}") from exc
with response:
_require_https_url(response.geturl(), f"final {label} URL")
final_url = response.geturl()
_require_https_url(final_url, f"final {label} URL")
if _resource_url(final_url) != _resource_url(url):
raise LauncherUpdateError(
f"{label} redirected unexpectedly to {final_url}; "
"the launcher update channel must be publicly readable"
)
length = _content_length(response.headers.get("Content-Length"), label)
if length is not None and length > maximum_bytes:
raise LauncherUpdateError(f"{label} exceeds its size limit")
@@ -363,6 +529,8 @@ def _fetch_small_file(
raise LauncherUpdateError(f"{label} exceeds its size limit")
if length is not None and len(data) != length:
raise LauncherUpdateError(f"{label} response was incomplete")
if not data:
raise LauncherUpdateError(f"{label} response was empty")
return data
@@ -399,12 +567,14 @@ def _verify_manifest_signature(
],
capture_output=True,
text=True,
encoding="utf-8",
errors="replace",
timeout=10,
check=False,
)
except (OSError, subprocess.TimeoutExpired) as exc:
raise LauncherUpdateError(f"could not verify update signature: {exc}") from exc
if result.returncode != 0 or "Verified OK" not in result.stdout:
if result.returncode != 0:
raise LauncherUpdateError("launcher update manifest signature is invalid")
@@ -501,6 +671,7 @@ def _stage_and_install_release(
f"staged launcher version is {staged_version}, expected {manifest.version}"
)
_self_check_release(staging, manifest.version)
_validate_release_directory(staging)
staging.chmod(0o755)
destination = paths.launcher_releases / manifest.version
if destination.exists() or destination.is_symlink():
@@ -655,9 +826,11 @@ def _self_check_release(release: Path, expected_version: str) -> None:
)
try:
result = subprocess.run(
[sys.executable, "-I", "-c", check, str(release), expected_version],
[sys.executable, "-I", "-B", "-c", check, str(release), expected_version],
capture_output=True,
text=True,
encoding="utf-8",
errors="replace",
timeout=15,
check=False,
)
@@ -769,6 +942,13 @@ def _signature_url(manifest_url: str) -> str:
return urlunsplit((parts.scheme, parts.netloc, parts.path + ".sig", parts.query, ""))
def _resource_url(value: str) -> str:
parts = urlsplit(value)
return urlunsplit(
(parts.scheme.casefold(), parts.netloc.casefold(), parts.path, parts.query, "")
)
def _require_https_url(value: str, label: str) -> None:
parts = urlsplit(value)
if (
@@ -788,6 +968,18 @@ def _parse_version(value: str, label: str) -> tuple[int, int, int]:
return int(major), int(minor), int(patch)
def _parse_published_at(value: str) -> datetime:
try:
parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
except ValueError as exc:
raise LauncherUpdateError(
f"manifest published_at must be an ISO-8601 timestamp: {value!r}"
) from exc
if parsed.tzinfo is None or parsed.utcoffset() is None:
raise LauncherUpdateError("manifest published_at must include a timezone")
return parsed
def _required_string(value: dict[str, object], key: str) -> str:
item = value.get(key)
if not isinstance(item, str) or not item:

View File

@@ -2,16 +2,24 @@
from __future__ import annotations
from dataclasses import asdict, dataclass
from dataclasses import asdict, dataclass, fields, replace
import json
import os
from pathlib import Path
import secrets
import threading
from typing import Any
from .locking import launcher_operation_lock
from .paths import AppPaths
SETTINGS_SCHEMA = 1
@dataclass(frozen=True)
class Settings:
realm: str = "Babel"
width: int = 1600
height: int = 900
output_width: int = 1600
@@ -33,15 +41,100 @@ class Settings:
def load_settings(paths: AppPaths) -> Settings:
try:
value = json.loads(paths.settings_file.read_text(encoding="utf-8"))
except FileNotFoundError:
value = _read_settings_object(paths.settings_file)
if value is None:
return Settings()
except (OSError, json.JSONDecodeError) as exc:
raise ValueError(f"invalid settings file {paths.settings_file}: {exc}") from exc
return _settings_from_object(value)
def save_settings(paths: AppPaths, settings: Settings) -> Settings:
"""Atomically save a complete settings snapshot.
Unknown fields from a newer launcher are retained when the existing file is
valid. Callers editing only part of the configuration should use
``update_settings`` so an old in-memory snapshot cannot undo another save.
"""
_validate(settings)
paths.config.mkdir(parents=True, exist_ok=True)
with launcher_operation_lock(paths.settings_lock, "settings-save"):
existing = _read_settings_object(paths.settings_file)
_write_settings(paths, settings, existing)
return settings
def update_settings(paths: AppPaths, **changes: Any) -> Settings:
"""Merge selected settings into the newest on-disk configuration."""
unknown = set(changes).difference(_SETTINGS_FIELD_NAMES)
if unknown:
names = ", ".join(sorted(unknown))
raise ValueError(f"unsupported launcher setting: {names}")
paths.config.mkdir(parents=True, exist_ok=True)
with launcher_operation_lock(paths.settings_lock, "settings-update"):
existing = _read_settings_object(paths.settings_file)
current = (
_settings_from_object(existing)
if existing is not None
else Settings()
)
updated = replace(current, **changes)
_validate(updated)
_write_settings(paths, updated, existing)
return updated
def ensure_settings(paths: AppPaths) -> Settings:
"""Create settings once and materialize defaults added by newer releases."""
paths.config.mkdir(parents=True, exist_ok=True)
with launcher_operation_lock(paths.settings_lock, "settings-migrate"):
existing = _read_settings_object(paths.settings_file)
if existing is None:
settings = Settings()
_write_settings(paths, settings, None)
return settings
settings = _settings_from_object(existing)
if _settings_need_migration(existing):
_write_settings(paths, settings, existing)
return settings
def reset_settings(paths: AppPaths) -> Settings:
"""Replace all launcher settings with the documented defaults."""
settings = Settings()
paths.config.mkdir(parents=True, exist_ok=True)
with launcher_operation_lock(paths.settings_lock, "settings-reset"):
_write_settings(paths, settings, None)
return settings
def _read_settings_object(path: Path) -> dict[str, object] | None:
try:
content = path.read_bytes()
except FileNotFoundError:
return None
except OSError as exc:
raise ValueError(f"invalid settings file {path}: {exc}") from exc
try:
value = json.loads(_decode_settings(content))
except (UnicodeError, json.JSONDecodeError) as exc:
raise ValueError(f"invalid settings file {path}: {exc}") from exc
if not isinstance(value, dict):
raise ValueError(f"settings file must contain an object: {paths.settings_file}")
raise ValueError(f"settings file must contain an object: {path}")
schema = value.get("schema")
if schema is not None:
if isinstance(schema, bool) or not isinstance(schema, int):
raise ValueError(f"settings schema must be an integer: {path}")
if schema != SETTINGS_SCHEMA:
raise ValueError(f"unsupported settings schema {schema}: {path}")
return value
def _settings_from_object(value: dict[str, object]) -> Settings:
settings = Settings(
realm=_string(value, "realm", "Babel").strip(),
width=_integer(value, "width", 1600),
height=_integer(value, "height", 900),
output_width=_integer(value, "output_width", 1600),
@@ -65,28 +158,71 @@ def load_settings(paths: AppPaths) -> Settings:
return settings
def save_settings(paths: AppPaths, settings: Settings) -> None:
_validate(settings)
paths.config.mkdir(parents=True, exist_ok=True)
def _decode_settings(content: bytes) -> str:
"""Accept native UTF-8 plus settings saved by common Windows editors."""
try:
return content.decode("utf-8-sig")
except UnicodeDecodeError as utf8_error:
try:
return content.decode("cp1252")
except UnicodeDecodeError:
raise utf8_error
def _settings_need_migration(value: dict[str, object]) -> bool:
return (
value.get("schema") != SETTINGS_SCHEMA
or not _SETTINGS_FIELD_NAMES.issubset(value)
)
def _write_settings(
paths: AppPaths,
settings: Settings,
existing: dict[str, object] | None,
) -> None:
payload = dict(existing or {})
payload.update(asdict(settings))
payload["schema"] = SETTINGS_SCHEMA
temporary = paths.settings_file.with_name(
f".{paths.settings_file.name}.{os.getpid()}.tmp"
f".{paths.settings_file.name}.{os.getpid()}.{threading.get_ident()}."
f"{secrets.token_hex(8)}.tmp"
)
temporary.write_text(
json.dumps(asdict(settings), indent=2, sort_keys=True) + "\n",
encoding="utf-8",
)
temporary.replace(paths.settings_file)
try:
with temporary.open("x", encoding="utf-8", newline="\n") as handle:
json.dump(payload, handle, indent=2, sort_keys=True)
handle.write("\n")
handle.flush()
os.fsync(handle.fileno())
temporary.chmod(0o600)
os.replace(temporary, paths.settings_file)
_fsync_directory(paths.config)
except BaseException:
temporary.unlink(missing_ok=True)
raise
def ensure_settings(paths: AppPaths) -> Settings:
if paths.settings_file.exists():
return load_settings(paths)
settings = Settings()
save_settings(paths, settings)
return settings
def _fsync_directory(path: Path) -> None:
try:
descriptor = os.open(path, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0))
except OSError:
return
try:
os.fsync(descriptor)
except OSError:
pass
finally:
os.close(descriptor)
def _validate(settings: Settings) -> None:
if (
len(settings.realm) > 64
or not settings.realm.strip()
or any(ord(character) < 32 for character in settings.realm)
):
raise ValueError("realm must be a non-empty display name")
for name, value in (
("width", settings.width),
("height", settings.height),
@@ -135,3 +271,6 @@ def _optional_string(value: dict[str, object], key: str) -> str | None:
if not isinstance(item, str) or not item:
raise ValueError(f"{key} must be null or a non-empty string")
return item
_SETTINGS_FIELD_NAMES = frozenset(field.name for field in fields(Settings))

File diff suppressed because it is too large Load Diff

View File

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

View File

@@ -1,6 +1,6 @@
[project]
name = "ptdiag"
version = "0.5.2"
name = "PTEU-Linux-Launcher"
version = "0.7.0"
description = "PristonTale EU Linux installer, clean launcher, and diagnostic harness"
requires-python = ">=3.11"
dependencies = []

11
updates/stable.json Normal file
View File

@@ -0,0 +1,11 @@
{
"archive_sha256": "a74403e6db36f88ee5fe8eef731b49b976eab5842428dd8dac82c6361f52ec44",
"archive_size": 241062,
"archive_url": "https://git.m4ko.net/m4ko/pristontale-eu-linux/releases/download/v0.7.0/pristontale-eu-linux-update-v0.7.0.tar.gz",
"channel": "stable",
"minimum_updater_version": "0.5.0",
"notes_url": "https://git.m4ko.net/m4ko/pristontale-eu-linux/releases/tag/v0.7.0",
"published_at": "2026-08-30T20:49:27Z",
"schema": 1,
"version": "0.7.0"
}

BIN
updates/stable.json.sig Normal file

Binary file not shown.