8 Commits

31 changed files with 1935 additions and 2166 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.6.5 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-6-5.tar.gz.sig \
pristontale-eu-linux-installer-v0-6-5.tar.gz
sha256sum -c pristontale-eu-linux-installer-v0-6-5.tar.gz.sha256
tar -xzf pristontale-eu-linux-installer-v0-6-5.tar.gz
cd pristontale-eu-linux-installer-v0-6-5
./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

@@ -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,10 @@ 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 and exact package root.
export PYTHONSAFEPATH=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.6.5"

View File

@@ -25,6 +25,7 @@ from .depot import SyncProgress, 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,
@@ -34,7 +35,7 @@ from .launcher import (
)
from .locking import launcher_operation_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 (
@@ -42,12 +43,13 @@ from .runtime import (
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 +78,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")
@@ -151,7 +162,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 +174,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 +197,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 +233,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,7 +293,6 @@ 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,
@@ -277,11 +307,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,12 +357,23 @@ 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)
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())
@@ -340,12 +382,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 +413,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 +528,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 +592,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 +617,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,6 +687,11 @@ def _maybe_import_game(
def _prepare(paths: AppPaths, game_dir: Path | None, as_json: bool) -> int:
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)
@@ -604,7 +703,6 @@ def _play(
paths: AppPaths,
game_dir: Path | None,
no_box: bool,
publisher_dll: bool,
dry_run: bool,
skip_runtime_setup: bool,
as_json: bool,
@@ -614,7 +712,6 @@ def _play(
paths,
game_dir,
no_box,
publisher_dll,
dry_run,
skip_runtime_setup,
as_json,
@@ -625,14 +722,15 @@ def _play_locked(
paths: AppPaths,
game_dir: Path | None,
no_box: bool,
publisher_dll: bool,
dry_run: bool,
skip_runtime_setup: bool,
as_json: bool,
) -> int:
# The legacy flag is retained for command-line compatibility. V21 performs
# periodic module-code checks, so every production launch is publisher-pristine.
_ = publisher_dll
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)
source = _source_game(paths, settings, game_dir)
overlay = None
@@ -660,7 +758,6 @@ def _play_locked(
selection,
runner,
use_gamebox=not no_box,
use_world_fix=False,
)
except KeyboardInterrupt:
current_game_log = _current_game_log(
@@ -676,7 +773,12 @@ 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,
@@ -686,7 +788,7 @@ def _play_locked(
"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 +824,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 +843,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 +881,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 +913,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 +942,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,

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,6 +39,38 @@ 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)
@@ -129,11 +162,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 +496,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 +526,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 +537,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 +552,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,6 +576,7 @@ def sync_managed_game(
repair,
False,
created=created,
quarantine=quarantine,
)
_write_json(paths.native_depot_state, asdict(result))
_emit_progress(
@@ -530,6 +591,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 +857,7 @@ def _sync_result(
dry_run: bool,
*,
created: int = 0,
quarantine: QuarantineResult = QuarantineResult(0, None),
) -> DepotSyncResult:
return DepotSyncResult(
region=snapshot.region,
@@ -704,6 +870,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(),

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

@@ -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 = 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

@@ -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,6 +153,12 @@ def fetch_news(
_write_cache(cache_path, articles)
return articles
except (OSError, NewsError, UnicodeDecodeError, json.JSONDecodeError):
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)
@@ -179,6 +185,12 @@ def fetch_latest_broadcast(
_write_broadcast_cache(cache_path, broadcast)
return broadcast
except (OSError, NewsError, UnicodeDecodeError, json.JSONDecodeError):
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)
@@ -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,6 +144,12 @@ 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."""
@@ -226,6 +250,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,6 +4,7 @@ from __future__ import annotations
from dataclasses import asdict, dataclass
from datetime import datetime
import hashlib
import json
import os
from pathlib import Path
@@ -16,11 +17,6 @@ from .command import CommandOutcome, CommandRunner, OutputLineFilter
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"
@@ -33,6 +29,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 +50,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 +89,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 +98,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 +184,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,11 +264,13 @@ 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),
},
)
@@ -255,8 +284,6 @@ def run_game(
runner: CommandRunner,
*,
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 +291,21 @@ 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")
if selection.kind == "umu":
game_command = [
selection.executable,
str(launch_executable),
str(paths.runtime_view / "Game.exe"),
"PTEU_REBORN",
]
if world_fix_helper is None:
game_command.append("PTEU_REBORN")
elif world_fix_status_target is not None:
game_command.extend(
["--status-file", world_fix_status_target.windows_path]
)
environment = _umu_environment(paths)
elif selection.kind == "proton7-container":
if not selection.proton_path:
raise RuntimeErrorPT("Proton script path is missing")
game_command = [
selection.executable,
str(launch_executable),
str(paths.runtime_view / "Game.exe"),
"PTEU_REBORN",
]
if world_fix_helper is None:
game_command.append("PTEU_REBORN")
elif world_fix_status_target is not None:
game_command.extend(
["--status-file", world_fix_status_target.windows_path]
)
environment = _umu_proton7_environment(
paths,
selection,
@@ -310,14 +320,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_GAME_EXE,
"PTEU_REBORN",
]
if world_fix_helper is None:
game_command.append("PTEU_REBORN")
elif world_fix_status_target is not None:
game_command.extend(
["--status-file", world_fix_status_target.windows_path]
)
environment = _proton_environment(
paths,
selection,
@@ -371,6 +376,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 +497,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 +541,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 +627,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 +655,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 +732,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,6 +316,8 @@ def install_launcher_update(
)
release_directory = _stage_and_install_release(paths, archive, check.manifest)
previous = _activate_release(paths, release_directory)
state_warning = None
try:
_write_json(
paths.launcher_update_state,
{
@@ -229,6 +332,10 @@ def install_launcher_update(
"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,14 +2,21 @@
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:
width: int = 1600
@@ -33,14 +40,98 @@ 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(
width=_integer(value, "width", 1600),
height=_integer(value, "height", 900),
@@ -65,25 +156,62 @@ 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:
@@ -135,3 +263,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))

View File

@@ -6,6 +6,7 @@ import base64
from collections import deque
import json
import math
import os
import queue
import subprocess
import sys
@@ -21,6 +22,7 @@ from .install_status import (
PrimaryAction,
inspect_game_installation,
)
from .import_state import game_import_requires_verification
from .news import (
Broadcast,
NewsArticle,
@@ -28,11 +30,18 @@ from .news import (
fetch_news,
fetch_news_image,
fetch_official_logo,
read_cached_broadcast,
read_cached_news,
)
from .overlay import RuntimeCompatibility, inspect_runtime_compatibility
from .paths import AppPaths
from .settings import Settings, ensure_settings, save_settings
from .self_update import LauncherUpdateCheck, check_for_launcher_update
from .settings import ensure_settings, update_settings
from .self_update import (
LauncherUpdateCheck,
cache_launcher_update_check,
check_for_launcher_update,
read_cached_launcher_update_check,
)
MAX_NEWS_CARDS = 3
@@ -43,6 +52,10 @@ NEWS_IMAGE_HEIGHT = 118
LAUNCH_STAGE_TOTAL = 4
EVENT_BATCH_SIZE = 48
POLL_INTERVAL_MILLISECONDS = 33
ACTIVE_POLL_INTERVAL_MILLISECONDS = 75
IDLE_POLL_INTERVAL_MILLISECONDS = 500
MAX_ACTIVITY_LOG_LINES = 5000
ACTIVITY_LOG_TRIM_LINES = 500
TRANSFER_RATE_WINDOW_SECONDS = 3.0
TRANSFER_RATE_MINIMUM_INTERVAL_SECONDS = 0.2
TRANSFER_RATE_IDLE_SECONDS = 2.0
@@ -52,6 +65,34 @@ class DashboardError(RuntimeError):
"""Raised when the native dashboard cannot be displayed."""
def launcher_restart_spec(
paths: AppPaths,
environment: Mapping[str, str] | None = None,
) -> tuple[list[str], dict[str, str]]:
"""Build an isolated restart that imports the newly activated release."""
current = paths.launcher_current
if not current.is_symlink():
raise DashboardError(f"launcher current path is not a symbolic link: {current}")
try:
release = current.resolve(strict=True)
except OSError as exc:
raise DashboardError(f"launcher current release is unavailable: {exc}") from exc
releases = paths.launcher_releases.resolve(strict=False)
if release.parent != releases or not (release / "ptlinux").is_dir():
raise DashboardError(f"launcher current release is outside the release store: {release}")
env = dict(os.environ if environment is None else environment)
env.update(
{
"PTLINUX_INSTALL_DIR": str(paths.data),
"PTLINUX_APP_HOME": str(paths.app_home),
"PYTHONSAFEPATH": "1",
"PYTHONPATH": str(release),
}
)
return [sys.executable, "-B", "-P", "-m", "ptlinux", "gui"], env
class ProgressCoalescer:
"""Keep only the latest cross-thread progress event for the Tk loop.
@@ -152,12 +193,21 @@ def run_dashboard(paths: AppPaths) -> None:
) from exc
paths.create_base_directories()
ensure_settings(paths)
root = tk.Tk()
root.title("PristonTale EU — Linux Launcher")
root.geometry(WINDOW_SIZE)
root.resizable(False, False)
root.configure(background="#0c0908")
try:
ensure_settings(paths)
except (OSError, ValueError) as error:
messagebox.showerror(
"Launcher settings",
f"The launcher could not load its settings:\n\n{error}",
parent=root,
)
root.destroy()
return
style = ttk.Style(root)
style.theme_use("clam")
@@ -565,6 +615,8 @@ def run_dashboard(paths: AppPaths) -> None:
preloader_progress: ttk.Progressbar | None = None
settings_window: tk.Toplevel | None = None
operation_active = False
game_window_active = False
activity_log_lines = 0
operation_started_at: float | None = None
operation_stage = 0
operation_stage_total = 0
@@ -586,8 +638,16 @@ def run_dashboard(paths: AppPaths) -> None:
)
def append_log(value: str) -> None:
nonlocal activity_log_lines
rendered = value.rstrip() + "\n"
log.configure(state="normal")
log.insert("end", value.rstrip() + "\n")
log.insert("end", rendered)
activity_log_lines += rendered.count("\n")
if activity_log_lines > MAX_ACTIVITY_LOG_LINES:
retained_lines = MAX_ACTIVITY_LOG_LINES - ACTIVITY_LOG_TRIM_LINES
excess = activity_log_lines - retained_lines
log.delete("1.0", f"{excess + 1}.0")
activity_log_lines -= excess
log.see("end")
log.configure(state="disabled")
@@ -759,7 +819,7 @@ def run_dashboard(paths: AppPaths) -> None:
restart_on_success: bool = False,
game_files_may_change: bool = False,
) -> None:
nonlocal operation_active
nonlocal operation_active, game_window_active
nonlocal operation_started_at
nonlocal operation_stage
nonlocal operation_stage_total
@@ -767,6 +827,7 @@ def run_dashboard(paths: AppPaths) -> None:
nonlocal operation_refreshes_compatibility
nonlocal last_elapsed_second
operation_active = True
game_window_active = False
operation_started_at = time.monotonic()
operation_stage = 0
operation_stage_total = 0
@@ -792,6 +853,8 @@ def run_dashboard(paths: AppPaths) -> None:
process = subprocess.Popen(
command,
text=True,
encoding="utf-8",
errors="replace",
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
bufsize=1,
@@ -1127,7 +1190,7 @@ def run_dashboard(paths: AppPaths) -> None:
runtime_compatibility = None
local_status = inspect_game_installation(paths)
refresh_primary_action(update_activity=update_activity)
if not local_status.installed:
if not local_status.installed or game_import_requires_verification(paths):
return
def worker() -> None:
@@ -1145,6 +1208,47 @@ def run_dashboard(paths: AppPaths) -> None:
daemon=True,
).start()
def show_launcher_update_check(
check: LauncherUpdateCheck,
*,
log_reason: bool = True,
) -> None:
nonlocal launcher_update_check
launcher_update_check = check
if check.available and check.supported:
launcher_update_button.configure(
text=f"UPDATE TO {check.latest_version}",
state="normal",
)
elif check.available:
launcher_update_button.configure(
text="MANUAL UPDATE NEEDED",
state="normal",
)
else:
launcher_update_button.configure(
text=f"LAUNCHER {__version__}",
state="normal",
)
if log_reason:
append_log(check.reason)
def refresh_launcher_update_check() -> None:
def worker() -> None:
try:
check = check_for_launcher_update()
except RuntimeError as update_error:
events.put(("launcher_update_error", str(update_error)))
else:
cache_launcher_update_check(paths.launcher_update_check_cache, check)
events.put(("launcher_update_check", check))
threading.Thread(
target=worker,
name="ptlinux-self-update",
daemon=True,
).start()
def refresh_remote_content() -> None:
def broadcast_worker() -> None:
events.put(("broadcast", fetch_latest_broadcast(paths.broadcast_cache)))
@@ -1160,27 +1264,17 @@ def run_dashboard(paths: AppPaths) -> None:
else:
events.put(("version", snapshot))
def launcher_update_worker() -> None:
try:
check = check_for_launcher_update()
except RuntimeError as update_error:
events.put(("launcher_update_error", str(update_error)))
else:
events.put(("launcher_update_check", check))
threading.Thread(target=broadcast_worker, name="ptlinux-broadcast", daemon=True).start()
threading.Thread(target=news_worker, name="ptlinux-news", daemon=True).start()
threading.Thread(target=version_worker, name="ptlinux-version", daemon=True).start()
threading.Thread(
target=launcher_update_worker,
name="ptlinux-self-update",
daemon=True,
).start()
refresh_launcher_update_check()
def finish_operation(returncode: int) -> None:
nonlocal operation_active, operation_refreshes_compatibility, operation_started_at
nonlocal operation_active, game_window_active
nonlocal operation_refreshes_compatibility, operation_started_at
refresh_compatibility = operation_refreshes_compatibility
operation_active = False
game_window_active = False
operation_refreshes_compatibility = False
operation_started_at = None
progress_updates.clear()
@@ -1200,7 +1294,7 @@ def run_dashboard(paths: AppPaths) -> None:
progress_text.set("See the activity log for details.")
def poll_events() -> None:
nonlocal launcher_update_check, runtime_compatibility
nonlocal game_window_active, launcher_update_check, runtime_compatibility
processed = 0
try:
while processed < EVENT_BATCH_SIZE:
@@ -1215,8 +1309,10 @@ def run_dashboard(paths: AppPaths) -> None:
if value == 0:
append_log("Launcher update activated; restarting the dashboard.")
try:
restart_command, restart_environment = launcher_restart_spec(paths)
subprocess.Popen(
command_for("gui"),
restart_command,
env=restart_environment,
close_fds=True,
start_new_session=True,
)
@@ -1236,6 +1332,7 @@ def run_dashboard(paths: AppPaths) -> None:
if step > 0 and total > 0 and isinstance(message, str) and message:
set_operation_stage(step, total, message)
elif kind in {"game_started", "game_window_ready"}:
game_window_active = True
progress_bar.stop()
status.set("Game client started.")
close_game_preloader()
@@ -1285,26 +1382,13 @@ def run_dashboard(paths: AppPaths) -> None:
value,
LauncherUpdateCheck,
):
launcher_update_check = value
if value.available and value.supported:
launcher_update_button.configure(
text=f"UPDATE TO {value.latest_version}",
state="normal",
)
elif value.available:
launcher_update_button.configure(
text="MANUAL UPDATE NEEDED",
state="normal",
)
else:
launcher_update_button.configure(
text=f"LAUNCHER {__version__}",
state="normal",
)
append_log(value.reason)
show_launcher_update_check(value)
elif kind == "launcher_update_error" and isinstance(value, str):
launcher_update_check = None
launcher_update_button.configure(text="CHECK LAUNCHER", state="normal")
if launcher_update_check is None:
launcher_update_button.configure(
text="CHECK LAUNCHER",
state="normal",
)
append_log(f"Launcher update check unavailable: {value}")
except queue.Empty:
pass
@@ -1315,7 +1399,14 @@ def run_dashboard(paths: AppPaths) -> None:
if speed_text.get() != "0 B/s" and transfer_rate.is_idle():
speed_text.set("0 B/s")
update_elapsed_indicator()
root.after(POLL_INTERVAL_MILLISECONDS, poll_events)
root.after(
dashboard_poll_interval(
operation_active=operation_active,
game_window_active=game_window_active,
pending_events=not events.empty(),
),
poll_events,
)
def command_for(*arguments: str) -> list[str]:
return [sys.executable, "-m", "ptlinux", *arguments]
@@ -1327,7 +1418,7 @@ def run_dashboard(paths: AppPaths) -> None:
if required_reason is not None:
status.set("Checking for a compatible launcher update…")
progress_text.set("Play remains disabled until this release is supported.")
refresh_remote_content()
refresh_launcher_update_check()
return
if not check.available:
if required_reason is None:
@@ -1419,7 +1510,15 @@ def run_dashboard(paths: AppPaths) -> None:
settings_window.focus_set()
return
try:
current_settings = ensure_settings(paths)
except (OSError, ValueError) as error:
messagebox.showerror(
"Settings",
f"The launcher could not load its settings:\n\n{error}",
parent=root,
)
return
dialog = tk.Toplevel(root)
settings_window = dialog
dialog.title("PristonTale EU — Settings")
@@ -1554,27 +1653,23 @@ def run_dashboard(paths: AppPaths) -> None:
def save_current_settings() -> None:
try:
updated_settings = Settings(
width=current_settings.width,
height=current_settings.height,
output_width=current_settings.output_width,
output_height=current_settings.output_height,
refresh_rate=current_settings.refresh_rate,
fullscreen=current_settings.fullscreen,
updated_settings = update_settings(
paths,
borderless=borderless_value.get(),
suppress_unresponsive_warnings=suppress_unresponsive_value.get(),
scaler=current_settings.scaler,
filter=current_settings.filter,
runtime=runtime_value.get(),
proton_sync=proton_sync_value.get(),
proton_path=proton_path_value.get().strip() or None,
game_source=game_source_value.get().strip() or None,
)
save_settings(paths, updated_settings)
except (OSError, ValueError) as error:
messagebox.showerror("Settings", str(error), parent=dialog)
return
append_log(f"Saved launcher settings to {paths.settings_file}")
append_log(
f"Saved launcher settings to {paths.settings_file}: "
f"runtime={updated_settings.runtime}, "
f"proton_sync={updated_settings.proton_sync}"
)
status.set("Settings saved. They will be used on the next game launch.")
progress_text.set("Open Settings again at any time to adjust the launcher.")
close_settings()
@@ -1729,6 +1824,13 @@ def run_dashboard(paths: AppPaths) -> None:
news_content.bind("<Configure>", update_news_viewport)
news_canvas.bind("<Configure>", resize_news_content)
show_broadcast(read_cached_broadcast(paths.broadcast_cache))
show_news(read_cached_news(paths.news_cache))
cached_launcher_update = read_cached_launcher_update_check(
paths.launcher_update_check_cache
)
if cached_launcher_update is not None:
show_launcher_update_check(cached_launcher_update, log_reason=False)
refresh_primary_action()
refresh_runtime_compatibility()
poll_events()
@@ -1747,6 +1849,21 @@ def _as_int(value: object) -> int:
return value if isinstance(value, int) and value >= 0 else 0
def dashboard_poll_interval(
*,
operation_active: bool,
game_window_active: bool,
pending_events: bool,
) -> int:
"""Choose a responsive poll rate while avoiding busy idle/game polling."""
if pending_events:
return POLL_INTERVAL_MILLISECONDS
if operation_active and not game_window_active:
return ACTIVE_POLL_INTERVAL_MILLISECONDS
return IDLE_POLL_INTERVAL_MILLISECONDS
def _format_bytes(value: int) -> str:
units = ("B", "KiB", "MiB", "GiB")
amount = float(value)

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.6.5"
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": "c1132f2e2d74579ea29fa59d69cb1df881800445be11d17e0fff7bc5d34c12fa",
"archive_size": 226431,
"archive_url": "https://git.m4ko.net/m4ko/pristontale-eu-linux/releases/download/v0.6.0/pristontale-eu-linux-update-v0.6.0.tar.gz",
"channel": "stable",
"minimum_updater_version": "0.5.0",
"notes_url": "https://git.m4ko.net/m4ko/pristontale-eu-linux/releases/tag/v0.6.0",
"published_at": "2026-08-29T21:16:21Z",
"schema": 1,
"version": "0.6.0"
}

BIN
updates/stable.json.sig Normal file

Binary file not shown.