From 7ce6ac3dd731e210d5899c598b385f2ef4e5c611 Mon Sep 17 00:00:00 2001 From: m4ko Date: Wed, 2 Sep 2026 21:42:49 +0200 Subject: [PATCH] Added umu download and repair for missing components on install, update and manual repair. --- install.sh | 30 ++- ptlinux/__main__.py | 10 +- ptlinux/cli.py | 208 ++++++++++++++++++-- ptlinux/dashboard_debug.py | 8 +- ptlinux/doctor.py | 27 ++- ptlinux/login_diagnostics.py | 9 +- ptlinux/paths.py | 17 ++ ptlinux/runtime.py | 35 +++- ptlinux/umu.py | 356 +++++++++++++++++++++++++++++++++++ 9 files changed, 660 insertions(+), 40 deletions(-) create mode 100644 ptlinux/umu.py diff --git a/install.sh b/install.sh index 827af45..b1def8e 100755 --- a/install.sh +++ b/install.sh @@ -35,14 +35,17 @@ PYTHON_BIN="${PTLINUX_PYTHON:-python3}" DRY_RUN=0 RUN_DOCTOR=1 INSTALL_DEPENDENCIES=1 +RUN_RUNTIME_BOOTSTRAP=1 usage() { printf '%s\n' \ - "Usage: ./install.sh [--install-dir PATH] [--dry-run] [--skip-deps] [--skip-doctor]" \ + "Usage: ./install.sh [--install-dir PATH] [--dry-run] [--skip-deps] [--skip-doctor] [--skip-runtime-bootstrap]" \ "" \ " --install-dir PATH Store the app, managed game, and runtime data at PATH." \ - " Default: ~/Games/PristontaleEU" + " Default: ~/Games/PristontaleEU" \ + " --skip-runtime-bootstrap" \ + " Do not prepare the verified UMU helper during install." } while (( $# > 0 )) @@ -75,6 +78,10 @@ do INSTALL_DEPENDENCIES=0 shift ;; + --skip-runtime-bootstrap) + RUN_RUNTIME_BOOTSTRAP=0 + shift + ;; -h|--help) usage exit 0 @@ -325,6 +332,17 @@ then fi printf '%s\n' "Launcher files installed." +RUNTIME_BOOTSTRAP_READY=1 +if (( RUN_RUNTIME_BOOTSTRAP == 1 )) +then + if ! "${BIN_HOME}/pristontale-eu" runtime-bootstrap + then + RUNTIME_BOOTSTRAP_READY=0 + fi +else + printf '%s\n' "Steam Runtime helper preparation skipped." +fi + DOCTOR_READY=1 if (( RUN_DOCTOR == 1 )) then @@ -356,6 +374,14 @@ case ":${PATH:-}:" in ;; esac +if (( RUNTIME_BOOTSTRAP_READY == 0 )) +then + printf '%s\n' \ + "Launcher files were installed, but the Steam Runtime helper could not be prepared." \ + "Retry with: ${BIN_HOME}/pristontale-eu runtime-bootstrap" >&2 + exit 1 +fi + if (( DOCTOR_READY == 0 )) then printf '%s\n' \ diff --git a/ptlinux/__main__.py b/ptlinux/__main__.py index 44f5523..9c70f6d 100644 --- a/ptlinux/__main__.py +++ b/ptlinux/__main__.py @@ -9,10 +9,16 @@ def main() -> int: arguments = sys.argv[1:] if not arguments or arguments == ["gui"]: from .paths import AppPaths - from .ui import run_dashboard + from .umu import provision_umu_for_startup try: - run_dashboard(AppPaths.from_environment()) + paths = AppPaths.from_environment() + # Older installed releases cannot run a bootstrap step that only exists + # in their update payload. Their automatic restart enters here instead. + provision_umu_for_startup(paths) + from .ui import run_dashboard + + run_dashboard(paths) except KeyboardInterrupt: return 130 except (ValueError, OSError, RuntimeError) as exc: diff --git a/ptlinux/cli.py b/ptlinux/cli.py index 78dc5c5..1f2420c 100644 --- a/ptlinux/cli.py +++ b/ptlinux/cli.py @@ -4,10 +4,11 @@ from __future__ import annotations import argparse from contextlib import AbstractContextManager, nullcontext -from dataclasses import asdict +from dataclasses import asdict, replace from datetime import datetime import json from pathlib import Path +import re import sys import time from typing import Callable, Sequence @@ -20,7 +21,7 @@ from ptdiag.logs import ( ) from . import __version__ -from .command import CommandError, CommandRunner +from .command import CommandError, CommandOutcome, CommandRunner from .depot import ( SyncProgress, read_cached_depot_launch_catalog, @@ -44,6 +45,7 @@ from .login_trace import DEFAULT_TRACE_TIMEOUT_SECONDS, LOGIN_TRACE_DISABLED_REA from .overlay import cache_compatibility_binaries, prepare_runtime_view from .paths import AppPaths from .runtime import ( + RuntimeErrorPT, game_runtime_ready, prepare_game_runtime, run_game, @@ -57,6 +59,12 @@ from .self_update import ( update_check_to_dict, verify_launcher_update_channel, ) +from .umu import ( + UmuBootstrapResult, + find_umu_launcher, + provision_umu, + provision_umu_for_startup, +) def build_parser() -> argparse.ArgumentParser: @@ -74,6 +82,10 @@ def build_parser() -> argparse.ArgumentParser: subparsers.add_parser("gui", help="open the native Linux dashboard") subparsers.add_parser("doctor", help="check cross-distribution dependencies") + subparsers.add_parser( + "runtime-bootstrap", + help="download and verify the Steam Runtime launcher helper", + ) login_diagnostics = subparsers.add_parser( "login-diagnostics", help="check the client, runtime, and publisher login route without credentials", @@ -236,12 +248,15 @@ def main(argv: Sequence[str] | None = None) -> int: paths = AppPaths.from_environment() try: if command == "gui": + provision_umu_for_startup(paths) from .ui import run_dashboard run_dashboard(paths) return 0 if command == "doctor": return _doctor(arguments.json) + if command == "runtime-bootstrap": + return _runtime_bootstrap(paths, arguments.json) if command == "login-diagnostics": return _login_diagnostics( paths, @@ -368,6 +383,25 @@ def _doctor(as_json: bool) -> int: return 0 if report.ok else 1 +def _runtime_bootstrap(paths: AppPaths, as_json: bool) -> int: + _print_operation_status(as_json, 1, 1, "Preparing the Steam Runtime helper…") + result = provision_umu(paths) + executable = find_umu_launcher(paths) + if executable is None: + raise LauncherError("the Steam Runtime helper is still unavailable") + _emit( + { + "ok": True, + "runtime_bootstrap": { + "executable": executable, + "managed_download": asdict(result) if result else None, + }, + }, + as_json, + ) + return 0 + + def _login_diagnostics( paths: AppPaths, timeout_seconds: float, @@ -781,8 +815,13 @@ def _play_locked( ) elif not dry_run: overlay = prepare_runtime_view(paths, source) - _print_operation_status(as_json, 2, 4, "Selecting the configured game runtime…") - selection = select_runtime(settings) + _print_operation_status(as_json, 2, 4, "Ensuring the compatibility runtime is available…") + umu_bootstrap = ( + None + if reuse_prepared + else _ensure_portable_umu_for_play(paths, settings, dry_run=dry_run) + ) + selection = select_runtime(settings, paths=paths) runner = _runner(paths, "play", dry_run) setup_results = [] if reuse_prepared: @@ -820,6 +859,7 @@ def _play_locked( start_option=start_option, use_gamebox=not no_box, concurrent_prefix_launch=reuse_prepared, + check=False, ) except KeyboardInterrupt: current_game_log = _current_game_log( @@ -841,23 +881,63 @@ def _play_locked( and runtime_issues["host_futex_fatal_count"] ): runtime_issues["recommended_one_variable_comparison"] = "proton7" - _emit( - { - "ok": outcome.returncode == 0, - "dry_run": dry_run, - "runtime": asdict(selection), - "realm": asdict(start_option), - "runtime_view": asdict(overlay) if overlay else None, - "setup_commands": [asdict(result) for result in setup_results], - "game": asdict(outcome), - "current_game_log": current_game_log, - "runtime_issues": runtime_issues, - }, - as_json, - ) + elif selection.kind == "proton7" and runtime_issues["gstreamer_missing_libraries"]: + runtime_issues["recommended_one_variable_comparison"] = "proton7-container" + elif runtime_issues["dbghelp_symdeferred_assertion_count"]: + runtime_issues["recommended_one_variable_comparison"] = "umu" + report = { + "ok": outcome.returncode == 0, + "dry_run": dry_run, + "runtime": asdict(selection), + "umu_bootstrap": asdict(umu_bootstrap) if umu_bootstrap else None, + "realm": asdict(start_option), + "runtime_view": asdict(overlay) if overlay else None, + "setup_commands": [asdict(result) for result in setup_results], + "game": asdict(outcome), + "current_game_log": current_game_log, + "runtime_issues": runtime_issues, + } + if as_json or dry_run: + _emit(report, as_json) + else: + _print_game_outcome(outcome, current_game_log, runtime_issues) return 0 if outcome.returncode == 0 else 1 +def _ensure_portable_umu_for_play( + paths: AppPaths, + settings: Settings, + *, + dry_run: bool, +) -> UmuBootstrapResult | None: + """Provide a pinned UMU when the selected policy may use a container.""" + + if settings.runtime == "proton7": + return None + if dry_run: + return None + try: + return provision_umu(paths) + except (OSError, RuntimeError) as exc: + if settings.runtime != "auto": + raise LauncherError( + f"the verified portable UMU runtime helper could not be installed: {exc}" + ) from exc + try: + select_runtime(replace(settings, runtime="proton7"), paths=paths) + except RuntimeErrorPT: + raise LauncherError( + "no usable game runtime is available and the verified portable UMU " + f"runtime helper could not be installed: {exc}" + ) from exc + print( + "Warning: the Steam Runtime helper could not be installed; falling back " + f"to direct Proton 7 for this launch ({exc}).", + file=sys.stderr, + ) + return None + + def _current_game_log( directory: Path, baseline: dict[str, LogFileState], @@ -910,15 +990,76 @@ 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" + dbghelp_assertion = "assertion failed: module->module.symtype != symdeferred" + missing_libraries = sorted( + { + library + for line in folded.splitlines() + if "gstreamer-warning" in line and "failed to load plugin" in line + for library in re.findall( + r"([a-z0-9_.+-]+\.so(?:\.\d+)*): cannot open shared object file", + line, + ) + } + ) return { "host_futex_fatal_count": folded.count(futex_message), "gstreamer_wrong_elf_class_count": folded.count(wrong_elf_message), + "gstreamer_missing_library_count": len(missing_libraries), + "gstreamer_missing_libraries": missing_libraries, "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, + "dbghelp_symdeferred_assertion_count": folded.count(dbghelp_assertion), } +def _print_game_outcome( + outcome: CommandOutcome, + current_game_log: dict[str, object] | None, + runtime_issues: dict[str, object], +) -> None: + # Keep the human-facing stream concise. CommandRunner already relayed every + # process line live; replaying its capture here made each Proton failure appear + # twice in the dashboard. + if outcome.timed_out: + print("Game client timed out and was stopped.", file=sys.stderr) + elif outcome.returncode == 0: + print("Game client exited normally.") + else: + print(f"Game client exited with status {outcome.returncode}.", file=sys.stderr) + if current_game_log is not None: + _print_interrupted_game_log(current_game_log) + wrong_elf_count = runtime_issues["gstreamer_wrong_elf_class_count"] + if isinstance(wrong_elf_count, int) and wrong_elf_count: + print( + "The repeated GStreamer wrong-ELF-class messages are plugin-scan noise; " + "they do not mean that Proton itself failed to start.", + file=sys.stderr, + ) + missing_libraries = runtime_issues["gstreamer_missing_libraries"] + if isinstance(missing_libraries, list) and missing_libraries: + print( + "Media plugins could not load these shared libraries: " + + ", ".join(str(value) for value in missing_libraries) + + ".", + file=sys.stderr, + ) + assertion_count = runtime_issues["dbghelp_symdeferred_assertion_count"] + if isinstance(assertion_count, int) and assertion_count: + print( + "Proton 7's older Wine build hit its dbghelp SymDeferred assertion; this " + "can mask the client failure that occurred immediately before it.", + file=sys.stderr, + ) + recommendation = runtime_issues.get("recommended_one_variable_comparison") + if isinstance(recommendation, str): + print( + f"Recommended single-variable runtime comparison: {recommendation}.", + file=sys.stderr, + ) + + 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"]: @@ -1020,17 +1161,44 @@ def _launcher_update( as_json, ) return 0 - _print_operation_status(as_json, 1, 2, "Verifying the signed launcher release…") + _print_operation_status(as_json, 1, 3, "Verifying the signed launcher release…") result = install_launcher_update( paths, manifest_url=manifest_url, public_key_path=public_key, ) - _print_operation_status(as_json, 2, 2, "Launcher update installed safely.") + _print_operation_status( + as_json, + 2, + 3, + "Launcher update installed; preparing the Steam Runtime helper…", + ) + umu_result: UmuBootstrapResult | None = None + umu_executable: str | None = None + umu_warning: str | None = None + try: + umu_result = provision_umu(paths) + umu_executable = find_umu_launcher(paths) + except (OSError, RuntimeError) as exc: + # Activation is already durable. Keep the update successful so the GUI can + # restart into the new release, whose startup hook and Play path both retry. + warning_message = ( + "the launcher update was installed, but the Steam Runtime helper could " + f"not be prepared yet: {exc}" + ) + umu_warning = warning_message + if not as_json: + print(f"Warning: {warning_message}", file=sys.stderr) + _print_operation_status(as_json, 3, 3, "Launcher update and runtime check complete.") _emit( { "ok": True, "launcher_update": asdict(result), + "runtime_bootstrap": { + "executable": umu_executable, + "managed_download": asdict(umu_result) if umu_result else None, + "warning": umu_warning, + }, }, as_json, ) diff --git a/ptlinux/dashboard_debug.py b/ptlinux/dashboard_debug.py index b36ad95..922485b 100644 --- a/ptlinux/dashboard_debug.py +++ b/ptlinux/dashboard_debug.py @@ -65,7 +65,7 @@ def inspect_runtime_debug( paths: AppPaths, settings: Settings, *, - selector: RuntimeSelector = select_runtime, + selector: RuntimeSelector | None = None, sync_resolver: SyncResolver = resolve_proton_sync_mode, ) -> RuntimeDebugInfo: """Resolve the next runtime and compare it with the preparation marker.""" @@ -85,7 +85,11 @@ def inspect_runtime_debug( selection: RuntimeSelection | None = None error: str | None = None try: - selection = selector(settings) + selection = ( + select_runtime(settings, paths=paths) + if selector is None + else selector(settings) + ) except (OSError, RuntimeErrorPT, ValueError) as exc: error = str(exc).strip() or type(exc).__name__ selected_kind = selection.kind if selection is not None else None diff --git a/ptlinux/doctor.py b/ptlinux/doctor.py index 5245d32..3bd9522 100644 --- a/ptlinux/doctor.py +++ b/ptlinux/doctor.py @@ -11,7 +11,9 @@ import shutil from typing import Mapping from .launcher import find_wine_tools +from .paths import AppPaths from .runtime import find_steam_proton7 +from .umu import managed_umu_ready ARCH_FAMILY_DISTRIBUTIONS = frozenset( @@ -48,6 +50,7 @@ def inspect_host(environment: Mapping[str, str] | None = None) -> DoctorReport: distribution = os_release.get("ID", "unknown") version = os_release.get("VERSION_ID", "unknown") wine_tools = find_wine_tools() + umu_path = _umu_path(env) tools = [ ToolStatus( "python3", @@ -93,9 +96,9 @@ def inspect_host(environment: Mapping[str, str] | None = None) -> DoctorReport: ), ToolStatus( "umu-run", - shutil.which("umu-run"), + umu_path, distribution in ARCH_FAMILY_DISTRIBUTIONS, - "current portable Proton runtime; automatic on Arch-family hosts", + "Steam Runtime container; managed automatically when needed", ), ToolStatus( "vulkaninfo", @@ -110,14 +113,15 @@ def inspect_host(environment: Mapping[str, str] | None = None) -> DoctorReport: warnings.append("PristonTale EU requires an x86-64 host with 32-bit Vulkan support.") if not env.get("WAYLAND_DISPLAY") and not env.get("DISPLAY"): warnings.append("No graphical X11 or Wayland session is visible.") - if shutil.which("umu-run") is None and not _has_steam_proton7(env): + if umu_path is None and not _has_steam_proton7(env): warnings.append( - "No game runtime was found: install umu-launcher or Steam Proton 7.0." + "No local game runtime is installed yet; run runtime-bootstrap to fetch the " + "pinned official UMU helper. Startup and Play also retry automatically." ) 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), + ok=required_ok, distribution=distribution, version=version, architecture=platform.machine(), @@ -180,6 +184,19 @@ def _has_steam_proton7(environment: Mapping[str, str]) -> bool: return find_steam_proton7(environment) is not None +def _umu_path(environment: Mapping[str, str]) -> str | None: + system = shutil.which("umu-run", path=environment.get("PATH")) + if system: + return system + try: + paths = AppPaths.from_environment(environment) + except RuntimeError: + return None + if managed_umu_ready(paths.managed_umu_executable): + return str(paths.managed_umu_executable.resolve()) + return None + + def _platform_optional_tools(distribution: str, *, architecture: str) -> list[ToolStatus]: if architecture.casefold() not in {"x86_64", "amd64"}: return [] diff --git a/ptlinux/login_diagnostics.py b/ptlinux/login_diagnostics.py index a59cc4a..ca958d5 100644 --- a/ptlinux/login_diagnostics.py +++ b/ptlinux/login_diagnostics.py @@ -213,7 +213,7 @@ def inspect_login_environment( ) runtime_requested = runtime_override or settings.runtime diagnostic_settings = replace(settings, runtime=runtime_requested) - selection, runtime_error = _runtime_selection(diagnostic_settings) + selection, runtime_error = _runtime_selection(paths, diagnostic_settings) distribution_id = detect_distribution_id() or "unknown" kernel_release = platform.release() proton_sync_effective = ( @@ -476,9 +476,12 @@ def _active_game_root(paths: AppPaths) -> Path: return paths.source_game -def _runtime_selection(settings: Settings) -> tuple[RuntimeSelection | None, str | None]: +def _runtime_selection( + paths: AppPaths, + settings: Settings, +) -> tuple[RuntimeSelection | None, str | None]: try: - return select_runtime(settings), None + return select_runtime(settings, paths=paths), None except RuntimeError as exc: return None, str(exc) diff --git a/ptlinux/paths.py b/ptlinux/paths.py index c1604d8..6904ed2 100644 --- a/ptlinux/paths.py +++ b/ptlinux/paths.py @@ -71,6 +71,23 @@ class AppPaths: def downloads(self) -> Path: return self.cache / "downloads" + @property + def runtime_tools(self) -> Path: + """Launcher-managed helpers that are independent of app releases.""" + + return self.data / "runtime-tools" + + @property + def managed_umu_executable(self) -> Path: + return self.runtime_tools / "bin/umu-run" + + @property + def managed_umu_metadata(self) -> Path: + return self.state / "umu-launcher.json" + + def managed_umu_archive(self, version: str) -> Path: + return self.downloads / f"umu-launcher-{version}-zipapp.tar" + @property def app_home(self) -> Path: """Stable home containing versioned native-launcher releases.""" diff --git a/ptlinux/runtime.py b/ptlinux/runtime.py index ec56f21..6e186a5 100644 --- a/ptlinux/runtime.py +++ b/ptlinux/runtime.py @@ -20,6 +20,7 @@ from .depot import DEFAULT_DEPOT_START_OPTION, DepotStartOption from .gamebox import build_gamebox_command from .paths import AppPaths from .settings import Settings +from .umu import find_umu_launcher WINDOWS_GAME_DIRECTORY = r"C:\Program Files (x86)\Pristontale EU" @@ -61,11 +62,12 @@ def select_runtime( settings: Settings, environment: Mapping[str, str] | None = None, *, + paths: AppPaths | 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")) + umu = find_umu_launcher(paths, env) current_umu_host = prefers_current_umu_host( distribution_id=distribution_id, kernel_release=kernel_release, @@ -257,7 +259,7 @@ def prepare_game_runtime( paths.create_base_directories() results: list[CommandOutcome] = [] if selection.kind == "umu": - environment = _umu_environment(paths) + environment = _umu_environment(paths, selection) if not native_d3d_ready(paths.umu_prefix): results.append( runner.run( @@ -398,7 +400,7 @@ def run_game( native_executable, *game_arguments, ] - environment = _umu_environment(paths) + environment = _umu_environment(paths, selection) elif selection.kind == "proton7-container": if not selection.proton_path: raise RuntimeErrorPT("Proton script path is missing") @@ -550,7 +552,7 @@ def run_runtime_prefix_command( raise ValueError("runtime-prefix command arguments cannot be empty") if selection.kind == "umu": command = [selection.executable, *arguments] - environment = _umu_environment(paths) + environment = _umu_environment(paths, selection) elif selection.kind == "proton7-container": if not selection.proton_path or not selection.steam_root: raise RuntimeErrorPT("incomplete Proton 7 runtime selection") @@ -586,14 +588,19 @@ def _runtime_python_path(existing: str | None) -> str: return os.pathsep.join((package_root, existing)) -def _umu_environment(paths: AppPaths) -> dict[str, str]: - return { +def _umu_environment( + paths: AppPaths, + selection: RuntimeSelection, +) -> dict[str, str]: + environment = { "WINEPREFIX": str(paths.umu_prefix), "GAMEID": "umu-pristontale-eu", "STORE": "none", "DXVK_STATE_CACHE_PATH": str(paths.dxvk_cache), "WINEDEBUG": "-all", } + environment.update(_portable_umu_environment(paths, selection)) + return environment def _umu_proton7_environment( @@ -619,10 +626,26 @@ def _umu_proton7_environment( "DXVK_STATE_CACHE_PATH": str(paths.dxvk_cache), "WINEDEBUG": "-all", } + environment.update(_portable_umu_environment(paths, selection)) environment.update(_proton_sync_environment(synchronization)) return environment +def _portable_umu_environment( + paths: AppPaths, + selection: RuntimeSelection, +) -> dict[str, str]: + """Keep downloads made by our zipapp inside the selected installation.""" + + try: + selected = Path(selection.executable).expanduser().resolve() + except (OSError, RuntimeError): + return {} + if selected != paths.managed_umu_executable.resolve(): + return {} + return {"UMU_FOLDERS_PATH": str(paths.runtime_tools)} + + def _proton_environment( paths: AppPaths, selection: RuntimeSelection, diff --git a/ptlinux/umu.py b/ptlinux/umu.py new file mode 100644 index 0000000..1a21955 --- /dev/null +++ b/ptlinux/umu.py @@ -0,0 +1,356 @@ +"""Pinned, user-level bootstrap for the official UMU launcher zipapp.""" + +from __future__ import annotations + +from dataclasses import asdict, dataclass +from datetime import datetime +import hashlib +import json +import os +from pathlib import Path +import secrets +import shutil +import sys +import tarfile +import threading +from typing import Callable, Mapping, Protocol +from urllib.error import HTTPError +from urllib.request import Request, urlopen +import zipfile + +from . import __version__ +from .paths import AppPaths + + +UMU_LAUNCHER_VERSION = "1.4.4" +UMU_LAUNCHER_URL = ( + "https://github.com/Open-Wine-Components/umu-launcher/releases/download/" + f"{UMU_LAUNCHER_VERSION}/umu-launcher-{UMU_LAUNCHER_VERSION}-zipapp.tar" +) +UMU_ARCHIVE_SHA256 = "eb590691841f7fad3fc3ad8fd5db4ccb87849fe7948e62b28ece7a4ee48cc851" +UMU_EXECUTABLE_SHA256 = "d0005a58602041229cc467dab03dc0c0b9e8cce09a8145b16b7683244cf17804" +UMU_ARCHIVE_MEMBER = "umu/umu-run" +MAX_UMU_ARCHIVE_BYTES = 4 * 1024 * 1024 +MAX_UMU_EXECUTABLE_BYTES = 2 * 1024 * 1024 +READ_SIZE = 256 * 1024 +USER_AGENT = f"PT-Linux/{__version__} (+https://pristontale.eu/)" + + +class UmuBootstrapError(RuntimeError): + """Raised when the pinned portable UMU launcher cannot be prepared.""" + + +class _HttpResponse(Protocol): + headers: Mapping[str, str] + + def read(self, size: int = -1) -> bytes: + ... + + def __enter__(self) -> _HttpResponse: + ... + + def __exit__( + self, + exception_type: object, + exception: object, + traceback: object, + ) -> object: + ... + + +UmuOpener = Callable[..., _HttpResponse] + + +@dataclass(frozen=True) +class UmuBootstrapResult: + path: str + version: str + source_url: str + archive_sha256: str + executable_sha256: str + changed: bool + + +def find_umu_launcher( + paths: AppPaths | None = None, + environment: Mapping[str, str] | None = None, +) -> str | None: + """Find a system UMU first, then the launcher's verified portable copy.""" + + env = os.environ if environment is None else environment + located = shutil.which("umu-run", path=env.get("PATH")) + if located: + candidate = Path(located).expanduser().resolve() + if ( + paths is None + or candidate != paths.managed_umu_executable.resolve() + or managed_umu_ready(candidate) + ): + return str(candidate) + if paths is not None and managed_umu_ready(paths.managed_umu_executable): + return str(paths.managed_umu_executable.resolve()) + return None + + +def managed_umu_ready(path: Path) -> bool: + """Accept only the exact launcher zipapp pinned by this release.""" + + try: + return ( + path.is_file() + and not path.is_symlink() + and os.access(path, os.X_OK) + and _sha256(path) == UMU_EXECUTABLE_SHA256 + ) + except OSError: + return False + + +def provision_umu( + paths: AppPaths, + *, + environment: Mapping[str, str] | None = None, + timeout_seconds: float = 60.0, + opener: UmuOpener = urlopen, +) -> UmuBootstrapResult | None: + """Ensure that either a system or launcher-managed UMU is available.""" + + if find_umu_launcher(paths, environment) is not None: + return None + return ensure_managed_umu( + paths, + timeout_seconds=timeout_seconds, + opener=opener, + ) + + +def provision_umu_for_startup( + paths: AppPaths, + *, + timeout_seconds: float = 15.0, +) -> UmuBootstrapResult | None: + """Repair a missing helper without preventing dashboard startup on failure.""" + + try: + return provision_umu(paths, timeout_seconds=timeout_seconds) + except (OSError, RuntimeError) as exc: + print( + "Warning: the Steam Runtime helper could not be prepared during launcher " + f"startup; Play will retry ({exc}).", + file=sys.stderr, + ) + return None + + +def ensure_managed_umu( + paths: AppPaths, + *, + timeout_seconds: float = 60.0, + opener: UmuOpener = urlopen, +) -> UmuBootstrapResult: + """Install the pinned official zipapp without modifying the host Python.""" + + destination = paths.managed_umu_executable + if managed_umu_ready(destination): + return _result(destination, changed=False) + + archive = paths.managed_umu_archive(UMU_LAUNCHER_VERSION) + if not _verified_archive(archive): + _download_archive( + archive, + timeout_seconds=timeout_seconds, + opener=opener, + ) + _install_zipapp(archive, destination) + result = _result(destination, changed=True) + _write_json( + paths.managed_umu_metadata, + { + **asdict(result), + "installed_at": datetime.now().astimezone().isoformat(), + }, + ) + return result + + +def _download_archive( + destination: Path, + *, + timeout_seconds: float, + opener: UmuOpener, +) -> None: + request = Request( + UMU_LAUNCHER_URL, + headers={"User-Agent": USER_AGENT}, + method="GET", + ) + try: + response = opener(request, timeout=timeout_seconds) + except HTTPError as exc: + raise UmuBootstrapError( + f"portable UMU request failed: HTTP {exc.code}" + ) from exc + except OSError as exc: + raise UmuBootstrapError(f"portable UMU request failed: {exc}") from exc + + with response: + length = _content_length(response.headers.get("Content-Length")) + if length is not None and length > MAX_UMU_ARCHIVE_BYTES: + raise UmuBootstrapError( + f"portable UMU response is unexpectedly large: {length} bytes" + ) + destination.parent.mkdir(parents=True, exist_ok=True) + temporary = _temporary_path(destination, "part") + digest = hashlib.sha256() + written = 0 + try: + with temporary.open("xb") as handle: + while True: + chunk = response.read(READ_SIZE) + if not chunk: + break + written += len(chunk) + if written > MAX_UMU_ARCHIVE_BYTES: + raise UmuBootstrapError( + "portable UMU response exceeded the 4 MiB safety limit" + ) + digest.update(chunk) + handle.write(chunk) + handle.flush() + os.fsync(handle.fileno()) + if length is not None and written != length: + raise UmuBootstrapError( + f"incomplete portable UMU response: {written}/{length} bytes" + ) + actual_hash = digest.hexdigest() + if actual_hash != UMU_ARCHIVE_SHA256: + raise UmuBootstrapError( + "portable UMU archive failed SHA-256 verification: " + f"expected {UMU_ARCHIVE_SHA256}, received {actual_hash}" + ) + temporary.replace(destination) + except BaseException: + temporary.unlink(missing_ok=True) + raise + + +def _install_zipapp(archive: Path, destination: Path) -> None: + destination.parent.mkdir(parents=True, exist_ok=True) + temporary = _temporary_path(destination, "tmp") + try: + try: + with tarfile.open(archive, "r:") as bundle: + member = bundle.getmember(UMU_ARCHIVE_MEMBER) + if not member.isfile(): + raise UmuBootstrapError( + "portable UMU archive member is not a regular file" + ) + if member.size <= 0 or member.size > MAX_UMU_EXECUTABLE_BYTES: + raise UmuBootstrapError( + f"portable UMU executable has invalid size: {member.size} bytes" + ) + source = bundle.extractfile(member) + if source is None: + raise UmuBootstrapError( + "portable UMU executable could not be read from its archive" + ) + with source, temporary.open("xb") as handle: + digest = hashlib.sha256() + written = 0 + while True: + chunk = source.read(READ_SIZE) + if not chunk: + break + written += len(chunk) + digest.update(chunk) + handle.write(chunk) + handle.flush() + os.fsync(handle.fileno()) + except (KeyError, OSError, tarfile.TarError) as exc: + raise UmuBootstrapError(f"invalid portable UMU archive: {exc}") from exc + if written != member.size or digest.hexdigest() != UMU_EXECUTABLE_SHA256: + raise UmuBootstrapError( + "portable UMU executable failed content verification" + ) + if not zipfile.is_zipfile(temporary): + raise UmuBootstrapError("portable UMU executable is not a valid zipapp") + with zipfile.ZipFile(temporary) as application: + if "__main__.py" not in application.namelist(): + raise UmuBootstrapError( + "portable UMU zipapp has no __main__.py entry point" + ) + bad_member = application.testzip() + if bad_member is not None: + raise UmuBootstrapError( + f"portable UMU zipapp contains a corrupt member: {bad_member}" + ) + temporary.chmod(0o755) + temporary.replace(destination) + except BaseException: + temporary.unlink(missing_ok=True) + raise + + +def _verified_archive(path: Path) -> bool: + try: + return ( + path.is_file() + and not path.is_symlink() + and path.stat().st_size <= MAX_UMU_ARCHIVE_BYTES + and _sha256(path) == UMU_ARCHIVE_SHA256 + ) + except OSError: + return False + + +def _result(path: Path, *, changed: bool) -> UmuBootstrapResult: + return UmuBootstrapResult( + path=str(path), + version=UMU_LAUNCHER_VERSION, + source_url=UMU_LAUNCHER_URL, + archive_sha256=UMU_ARCHIVE_SHA256, + executable_sha256=UMU_EXECUTABLE_SHA256, + changed=changed, + ) + + +def _content_length(value: str | None) -> int | None: + if value is None: + return None + try: + length = int(value) + except ValueError as exc: + raise UmuBootstrapError(f"invalid portable UMU Content-Length: {value!r}") from exc + if length < 0: + raise UmuBootstrapError(f"invalid portable UMU Content-Length: {value!r}") + return length + + +def _sha256(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as handle: + for chunk in iter(lambda: handle.read(READ_SIZE), b""): + digest.update(chunk) + return digest.hexdigest() + + +def _temporary_path(path: Path, suffix: str) -> Path: + return path.with_name( + f".{path.name}.{os.getpid()}.{threading.get_ident()}." + f"{secrets.token_hex(8)}.{suffix}" + ) + + +def _write_json(path: Path, value: object) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + temporary = _temporary_path(path, "tmp") + try: + with temporary.open("x", encoding="utf-8", newline="\n") 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