From bef8d04d72ff8ad502e2ae764f0607b89b414608 Mon Sep 17 00:00:00 2001 From: m4ko Date: Sun, 30 Aug 2026 00:08:41 +0200 Subject: [PATCH] Fixed issues with encodings and wine / proton compatibilities. --- install.sh | 16 ++- packaging/pristontale-eu | 4 +- ptlinux/__init__.py | 2 +- ptlinux/cli.py | 157 +++++++++++++++++++--------- ptlinux/command.py | 6 ++ ptlinux/doctor.py | 15 ++- ptlinux/login_diagnostics.py | 192 ++++++++++++++++++++++++++++++++++- ptlinux/paths.py | 12 ++- ptlinux/runtime.py | 132 +++++++++++++++++++++--- ptlinux/self_update.py | 148 +++++++++++++++++++++++---- ptlinux/settings.py | 175 +++++++++++++++++++++++++++---- ptlinux/ui.py | 43 +++++--- pyproject.toml | 2 +- 13 files changed, 772 insertions(+), 132 deletions(-) diff --git a/install.sh b/install.sh index fd08b90..7c53ac0 100755 --- a/install.sh +++ b/install.sh @@ -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" \ @@ -201,15 +207,15 @@ 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 diff --git a/packaging/pristontale-eu b/packaging/pristontale-eu index 9012f58..fe35fc0 100755 --- a/packaging/pristontale-eu +++ b/packaging/pristontale-eu @@ -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 @@ -37,4 +37,4 @@ fi export PYTHONSAFEPATH=1 export PYTHONPATH="${release_root}" -exec "$python_bin" -P -m ptlinux "$@" +exec "$python_bin" -B -P -m ptlinux "$@" diff --git a/ptlinux/__init__.py b/ptlinux/__init__.py index 14db81f..bb2068c 100644 --- a/ptlinux/__init__.py +++ b/ptlinux/__init__.py @@ -4,4 +4,4 @@ from .paths import AppPaths __all__ = ["AppPaths", "__version__"] -__version__ = "0.5.13" +__version__ = "0.6.0" diff --git a/ptlinux/cli.py b/ptlinux/cli.py index 82d0a37..551b012 100644 --- a/ptlinux/cli.py +++ b/ptlinux/cli.py @@ -34,7 +34,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 +42,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 +77,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 +161,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 +173,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 +196,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 +232,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": @@ -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,13 +357,24 @@ def _doctor(as_json: bool) -> int: return 0 if report.ok else 1 -def _login_diagnostics(paths: AppPaths, timeout_seconds: float, as_json: bool) -> int: +def _login_diagnostics( + paths: AppPaths, + timeout_seconds: float, + runtime_choice: str, + as_json: bool, +) -> int: settings = _load_or_default_settings(paths, persist=False) - report = inspect_login_environment( - paths, - settings, - timeout_seconds=timeout_seconds, - ) + runner = _runner(paths, "login-diagnostics", False) + with _operation_guard(paths, "login-diagnostics", False): + report = inspect_login_environment( + paths, + settings, + timeout_seconds=timeout_seconds, + runner=runner, + runtime_override=( + None if runtime_choice == "configured" else runtime_choice + ), + ) if as_json: _print_json(report.to_dict()) else: @@ -340,12 +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: @@ -738,6 +798,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 @@ -790,40 +855,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, @@ -838,6 +887,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, @@ -866,6 +916,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, diff --git a/ptlinux/command.py b/ptlinux/command.py index 4118237..e7bd633 100644 --- a/ptlinux/command.py +++ b/ptlinux/command.py @@ -18,6 +18,8 @@ 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: @@ -134,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, @@ -178,6 +182,8 @@ 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, diff --git a/ptlinux/doctor.py b/ptlinux/doctor.py index a4980aa..4fa2db7 100644 --- a/ptlinux/doctor.py +++ b/ptlinux/doctor.py @@ -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", @@ -157,7 +162,7 @@ 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 and gstreamer dependencies with Pacman." @@ -189,7 +194,7 @@ def _has_steam_proton7(environment: Mapping[str, str]) -> bool: def _platform_optional_tools(distribution: str, *, architecture: str) -> list[ToolStatus]: if architecture.casefold() not in {"x86_64", "amd64"}: return [] - if distribution in {"arch", "cachyos", "endeavouros", "manjaro", "garuda"}: + if distribution in ARCH_FAMILY_DISTRIBUTIONS: return [ _optional_tool_status( "lib32-vulkan", @@ -206,7 +211,7 @@ def _platform_optional_tools(distribution: str, *, architecture: str) -> list[To def _arch_runtime_warnings(distribution: str, architecture: str) -> list[str]: - if distribution not in {"arch", "cachyos", "endeavouros", "manjaro", "garuda"}: + if distribution not in ARCH_FAMILY_DISTRIBUTIONS: return [] if architecture.casefold() not in {"x86_64", "amd64"}: return [ diff --git a/ptlinux/login_diagnostics.py b/ptlinux/login_diagnostics.py index c56b734..8f85459 100644 --- a/ptlinux/login_diagnostics.py +++ b/ptlinux/login_diagnostics.py @@ -2,7 +2,7 @@ from __future__ import annotations -from dataclasses import asdict, dataclass +from dataclasses import asdict, dataclass, replace from datetime import datetime import hashlib import json @@ -14,6 +14,13 @@ 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, @@ -43,6 +50,36 @@ DEFAULT_LOGIN_TIMEOUT_SECONDS = 5.0 VERIFIED_GAME_PROFILES = frozenset( {"v21.75.83", "verified-compatibility"} ) +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", +) class _Closable(Protocol): @@ -65,6 +102,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 @@ -76,6 +127,16 @@ 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 @@ -93,9 +154,13 @@ class LoginDiagnosticReport: world_fix_helper_ready: bool world_fix_enabled: bool world_fix_disabled_reason: 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 @@ -148,6 +213,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.""" @@ -155,7 +222,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 = ( @@ -192,6 +261,16 @@ def inspect_login_environment( prefix = _runtime_prefix(paths, selection) prefix_ready = _prefix_ready(prefix) directx_ready = native_d3d_ready(prefix) if prefix is not None else False + 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, settings) warnings = _diagnostic_warnings( probe, @@ -201,6 +280,12 @@ def inspect_login_environment( 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=( @@ -210,6 +295,7 @@ def inspect_login_environment( and selection is not None and prefix_ready and directx_ready + and identity_preflight_ok ), checked_at=datetime.now().astimezone().isoformat(), launcher_version=__version__, @@ -219,6 +305,22 @@ 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, @@ -238,9 +340,13 @@ def inspect_login_environment( world_fix_disabled_reason=( None if WORLD_FIX_ENABLED else WORLD_FIX_DISABLED_REASON ), + 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, @@ -252,6 +358,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, @@ -354,6 +539,7 @@ def _diagnostic_warnings( 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: @@ -392,6 +578,8 @@ 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) diff --git a/ptlinux/paths.py b/ptlinux/paths.py index 9afb1a5..256e9b7 100644 --- a/ptlinux/paths.py +++ b/ptlinux/paths.py @@ -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) @@ -238,6 +244,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, diff --git a/ptlinux/runtime.py b/ptlinux/runtime.py index c5ead23..10013e3 100644 --- a/ptlinux/runtime.py +++ b/ptlinux/runtime.py @@ -34,6 +34,9 @@ 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 @@ -52,10 +55,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}") @@ -71,7 +94,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, @@ -79,13 +103,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" @@ -377,6 +405,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) @@ -455,6 +526,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, *, @@ -467,9 +570,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 @@ -552,21 +656,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 diff --git a/ptlinux/self_update.py b/ptlinux/self_update.py index 7bb3d7e..cf6c26b 100644 --- a/ptlinux/self_update.py +++ b/ptlinux/self_update.py @@ -85,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) @@ -262,6 +275,7 @@ def _evaluate_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, @@ -270,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, @@ -284,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) @@ -299,20 +316,26 @@ def install_launcher_update( ) release_directory = _stage_and_install_release(paths, archive, check.manifest) previous = _activate_release(paths, release_directory) - _write_json( - paths.launcher_update_state, - { - "schema": 1, - "updated_at": datetime.now().astimezone().isoformat(), - "from_version": check.current_version, - "to_version": check.manifest.version, - "manifest_url": check.manifest_url, - "archive_url": check.manifest.archive_url, - "archive_sha256": check.manifest.archive_sha256, - "release_directory": str(release_directory), - "previous_release": str(previous) if previous else None, - }, - ) + state_warning = None + try: + _write_json( + paths.launcher_update_state, + { + "schema": 1, + "updated_at": datetime.now().astimezone().isoformat(), + "from_version": check.current_version, + "to_version": check.manifest.version, + "manifest_url": check.manifest_url, + "archive_url": check.manifest.archive_url, + "archive_sha256": check.manifest.archive_sha256, + "release_directory": str(release_directory), + "previous_release": str(previous) if previous else None, + }, + ) + except OSError as exc: + # Activation is the durable user-facing operation. A diagnostics-only + # state record must not strand the dashboard on the old release. + state_warning = f"launcher updated, but update history could not be saved: {exc}" return LauncherUpdateResult( changed=True, current_version=check.current_version, @@ -321,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, ) @@ -397,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") @@ -428,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", ) @@ -438,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") @@ -447,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 @@ -483,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") @@ -585,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(): @@ -739,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, ) @@ -853,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 ( @@ -872,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: diff --git a/ptlinux/settings.py b/ptlinux/settings.py index 22a47b9..709dee2 100644 --- a/ptlinux/settings.py +++ b/ptlinux/settings.py @@ -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)) diff --git a/ptlinux/ui.py b/ptlinux/ui.py index 13bd435..81270e9 100644 --- a/ptlinux/ui.py +++ b/ptlinux/ui.py @@ -33,7 +33,7 @@ from .news import ( ) from .overlay import RuntimeCompatibility, inspect_runtime_compatibility from .paths import AppPaths -from .settings import Settings, ensure_settings, save_settings +from .settings import ensure_settings, update_settings from .self_update import ( LauncherUpdateCheck, cache_launcher_update_check, @@ -163,12 +163,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") @@ -814,6 +823,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, @@ -1467,7 +1478,15 @@ def run_dashboard(paths: AppPaths) -> None: settings_window.focus_set() return - current_settings = ensure_settings(paths) + 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") @@ -1602,27 +1621,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() diff --git a/pyproject.toml b/pyproject.toml index 4f1436c..188419f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "ptdiag" -version = "0.5.2" +version = "0.6.0" description = "PristonTale EU Linux installer, clean launcher, and diagnostic harness" requires-python = ">=3.11" dependencies = []