18 Commits

Author SHA1 Message Date
55b2292bfe Released v0.8.3 2026-09-01 22:26:19 +02:00
15dbd90dbb Optimizations and performance boost to pre-loader for launcher and game client. 2026-09-01 22:21:54 +02:00
c2400db19e Changed how borderless setting works. 2026-09-01 22:06:00 +02:00
1a17c1a197 Fixed a gamebox issue where resolution would act up on launch. 2026-09-01 22:05:33 +02:00
25f490c8b8 Hotfixed a issue with resolution in borderless mode. 2026-09-01 20:51:45 +02:00
e8d77f3779 Released v0.8.1. 2026-09-01 20:17:37 +02:00
6900c20734 Prepare release v0.8.1 2026-09-01 20:09:16 +02:00
f96aba9d99 Fixed multiple clients bug in umu. 2026-09-01 20:05:54 +02:00
78dade0d86 Fixed some proton 7 path issues and the override in settings. 2026-09-01 19:07:00 +02:00
d819809197 New release: v0.8.0 2026-08-31 23:30:39 +02:00
dadae8f60a Added ui changes to modals, pre-loader and some fixes for other ui elements and performance improvements all over. 2026-08-31 23:23:37 +02:00
99982b12d9 Added debugger class for dashboard and other launcher util classes. 2026-08-31 23:13:04 +02:00
75b21acf87 Added windows identity checker. 2026-08-31 23:12:20 +02:00
b16540099d Added changelog class 2026-08-31 23:11:21 +02:00
b0855ccd9b Added import class. 2026-08-31 23:10:38 +02:00
87dee93b86 Reworked the modal ui's for launcher, settings, updates and game pre-loader. 2026-08-31 22:03:47 +02:00
fe883047e0 Added a debug system and fixed thread issues with update / repair. 2026-08-31 20:23:23 +02:00
20a0bbcd75 Removed player counter. 2026-08-31 17:46:08 +02:00
25 changed files with 4407 additions and 430 deletions

23
.gitignore vendored
View File

@@ -3,8 +3,25 @@ __pycache__/
*.py[cod] *.py[cod]
.pytest_cache/ .pytest_cache/
.idea/ .idea/
dist/ .run/
release-keys/ release-keys/
config/local.toml config/local.toml
results/*
!results/.gitkeep # Local development, investigation, and release tooling. The production
# repository intentionally contains only the installer/runtime sources.
AGENTS.md
AIO_INSTALLER.md
BOOTSTRAP_WORKPLAN.md
DEBUGGING.md
GITEA_UPDATES.md
README.md
config/
configs/
findings/
replay_pid*.log
results/
testcases/
tests/
tools/
dist/
/native/

View File

@@ -1,6 +1,6 @@
# PristonTale EU for Linux # PristonTale EU for Linux
Version 0.7.0 installs, updates, repairs, and launches PristonTale EU through a Version 0.8.3 installs, updates, repairs, and launches PristonTale EU through a
native Linux dashboard. Game files come directly from the publisher and are not native Linux dashboard. Game files come directly from the publisher and are not
included in this archive. included in this archive.
@@ -15,11 +15,11 @@ Then run:
```text ```text
openssl pkey -pubin -in launcher-update-public.pem -outform DER | sha256sum openssl pkey -pubin -in launcher-update-public.pem -outform DER | sha256sum
openssl dgst -sha256 -verify launcher-update-public.pem \ openssl dgst -sha256 -verify launcher-update-public.pem \
-signature pristontale-eu-linux-installer-v0-7-0.tar.gz.sig \ -signature pristontale-eu-linux-installer-v0-8-3.tar.gz.sig \
pristontale-eu-linux-installer-v0-7-0.tar.gz pristontale-eu-linux-installer-v0-8-3.tar.gz
sha256sum -c pristontale-eu-linux-installer-v0-7-0.tar.gz.sha256 sha256sum -c pristontale-eu-linux-installer-v0-8-3.tar.gz.sha256
tar -xzf pristontale-eu-linux-installer-v0-7-0.tar.gz tar -xzf pristontale-eu-linux-installer-v0-8-3.tar.gz
cd pristontale-eu-linux-installer-v0-7-0 cd pristontale-eu-linux-installer-v0-8-3
./install.sh ./install.sh
``` ```
@@ -34,6 +34,11 @@ Existing installations at the previous default location are preserved during
upgrades. The installer adds **PristonTale EU** to the application menu and upgrades. The installer adds **PristonTale EU** to the application menu and
installs the `pristontale-eu` command in `~/.local/bin`. installs the `pristontale-eu` command in `~/.local/bin`.
Steam Proton 7.0 is discovered in both the default Steam library and secondary
libraries registered with the Steam client. If Proton uses an unregistered
manual layout, open **Settings → Optional Overrides** and select the exact
`Proton 7.0/proton` script.
## Use ## Use
Open **PristonTale EU** from the application menu or run: Open **PristonTale EU** from the application menu or run:

View File

@@ -12,17 +12,20 @@ export PTLINUX_APP_HOME="${app_home}"
release_root="" release_root=""
for candidate in "${app_home}/current" "${app_home}/previous" "${app_home}" for candidate in "${app_home}/current" "${app_home}/previous" "${app_home}"
do do
if [ ! -d "${candidate}/ptlinux" ] # Installed releases already pass an isolated import self-check before their
# activation link is changed. Keep only a cheap corruption/fallback check on
# every boot instead of starting Python twice and importing the full CLI twice.
if [ ! -r "${candidate}/ptlinux/__init__.py" ] || \
[ ! -r "${candidate}/ptlinux/__main__.py" ] || \
[ ! -r "${candidate}/ptlinux/cli.py" ] || \
[ ! -r "${candidate}/ptdiag/__init__.py" ] || \
[ ! -r "${candidate}/ptdiag/assbin.py" ] || \
[ ! -r "${candidate}/ptdiag/logs.py" ]
then then
continue continue
fi fi
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
release_root="${candidate}" release_root="${candidate}"
break break
fi
done done
if [ -z "${release_root}" ] if [ -z "${release_root}" ]

View File

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

View File

@@ -1,4 +1,28 @@
from .cli import main from __future__ import annotations
import sys
def main() -> int:
"""Dispatch the common GUI path without importing every CLI operation."""
arguments = sys.argv[1:]
if not arguments or arguments == ["gui"]:
from .paths import AppPaths
from .ui import run_dashboard
try:
run_dashboard(AppPaths.from_environment())
except KeyboardInterrupt:
return 130
except (ValueError, OSError, RuntimeError) as exc:
print(f"Error: {exc}", file=sys.stderr)
return 1
return 0
from .cli import main as cli_main
return cli_main(arguments)
raise SystemExit(main()) raise SystemExit(main())

332
ptlinux/changelog.py Normal file
View File

@@ -0,0 +1,332 @@
"""Official patch changelog support for the native Linux dashboard."""
from __future__ import annotations
from dataclasses import asdict, dataclass
from datetime import UTC, datetime
import json
import math
import os
from pathlib import Path
import re
from typing import Callable, Protocol
from urllib.request import Request, urlopen
PATCHES_ENDPOINT = "https://pristontale.eu/api/changelog/patches"
PATCH_NEWS_URL = "https://pristontale.eu/news#"
MAX_PATCHES = 10
MAX_PATCH_BYTES = 2 * 1024 * 1024
MAX_PATCH_SECTIONS = 32
MAX_PATCH_CHANGES = 200
_MAX_PATCH_CANDIDATES = 100
_SLUG_PATTERN = re.compile(r"[a-z0-9]+(?:-[a-z0-9]+)*")
class ChangelogError(RuntimeError):
"""Raised when the official patch feed cannot be parsed safely."""
class ChangelogResponse(Protocol):
"""The minimal response interface used by the changelog client."""
def read(self, size: int = -1) -> bytes: ...
def __enter__(self) -> ChangelogResponse: ...
def __exit__(
self,
exception_type: type[BaseException] | None,
exception: BaseException | None,
traceback: object | None,
) -> bool | None: ...
ChangelogOpener = Callable[..., ChangelogResponse]
@dataclass(frozen=True)
class PatchChange:
text: str
type_key: str
@property
def tag(self) -> str:
return patch_change_tag(self.type_key)
@dataclass(frozen=True)
class PatchSection:
title: str
items: tuple[PatchChange, ...]
@dataclass(frozen=True)
class PatchNote:
id: int
slug: str
title: str
version: str | None
tags: tuple[str, ...]
summary: str
sections: tuple[PatchSection, ...]
published_at: str | None
@property
def url(self) -> str:
return f"{PATCH_NEWS_URL}{self.slug}"
def patch_change_tag(type_key: str) -> str:
"""Turn API keys such as ``new-feature`` into compact UI tags."""
label = " ".join(type_key.replace("_", "-").split("-"))
return label.upper() if label else "CHANGE"
def fetch_patch_notes(
cache_path: Path,
*,
timeout_seconds: float = 15.0,
opener: ChangelogOpener = urlopen,
) -> tuple[PatchNote, ...]:
"""Fetch the ten latest patches, falling back to the last valid cache."""
request = Request(
PATCHES_ENDPOINT,
headers={"User-Agent": "PT-Linux/0.6 (+https://pristontale.eu/)"},
method="GET",
)
try:
response = opener(request, timeout=timeout_seconds)
with response:
payload = response.read(MAX_PATCH_BYTES + 1)
if len(payload) > MAX_PATCH_BYTES:
raise ChangelogError("official patch response exceeds the 2 MiB limit")
patches = _parse_patch_notes(payload)
_write_patch_cache(cache_path, patches)
return patches
except (OSError, ChangelogError, UnicodeDecodeError, json.JSONDecodeError):
return read_cached_patch_notes(cache_path)
def read_cached_patch_notes(cache_path: Path) -> tuple[PatchNote, ...]:
"""Read the last valid patch feed without performing network I/O."""
try:
value = json.loads(cache_path.read_text(encoding="utf-8"))
except (OSError, UnicodeError, json.JSONDecodeError):
return ()
if not isinstance(value, list):
return ()
patches: list[PatchNote] = []
for item in value[:MAX_PATCHES]:
patch = _patch_note_from_cache(item)
if patch is not None:
patches.append(patch)
return tuple(patches)
def _parse_patch_notes(payload: bytes) -> tuple[PatchNote, ...]:
value = json.loads(payload.decode("utf-8-sig"))
if isinstance(value, dict):
value = value.get("data")
if not isinstance(value, list):
raise ChangelogError("official patch response is not a patch list")
parsed: list[tuple[float, int, PatchNote]] = []
for item in value[:_MAX_PATCH_CANDIDATES]:
result = _parse_patch_note(item)
if result is not None:
timestamp, patch = result
parsed.append((timestamp, patch.id, patch))
parsed.sort(key=lambda entry: (entry[0], entry[1]), reverse=True)
patches = tuple(entry[2] for entry in parsed[:MAX_PATCHES])
if not patches:
raise ChangelogError("official patch response contains no usable patches")
return patches
def _parse_patch_note(value: object) -> tuple[float, PatchNote] | None:
if not isinstance(value, dict):
return None
patch_id = value.get("id")
slug = _clean_text(value.get("slug"), 180)
title = _clean_text(value.get("title"), 200)
summary = _clean_text(value.get("summary"), 3000, allow_empty=True)
if (
isinstance(patch_id, bool)
or not isinstance(patch_id, int)
or slug is None
or _SLUG_PATTERN.fullmatch(slug) is None
or title is None
or summary is None
):
return None
version = _clean_text(value.get("version"), 64, allow_none=True)
if value.get("version") is not None and version is None:
return None
timestamp = _timestamp(value.get("date"))
patch = PatchNote(
id=patch_id,
slug=slug,
title=title,
version=version,
tags=_parse_tags(value.get("tags")),
summary=summary,
sections=_parse_sections(value.get("changes")),
published_at=_format_timestamp(timestamp),
)
return (timestamp if timestamp is not None else -1.0, patch)
def _parse_tags(value: object) -> tuple[str, ...]:
if not isinstance(value, list):
return ()
tags: list[str] = []
seen: set[str] = set()
for item in value[:16]:
tag = _clean_text(item, 48)
if tag is None or tag.casefold() in seen:
continue
seen.add(tag.casefold())
tags.append(tag)
return tuple(tags)
def _parse_sections(value: object) -> tuple[PatchSection, ...]:
if not isinstance(value, dict):
return ()
sections_value = value.get("sections")
if not isinstance(sections_value, list):
return ()
sections: list[PatchSection] = []
changes_seen = 0
for section_value in sections_value[:MAX_PATCH_SECTIONS]:
if not isinstance(section_value, dict):
continue
title = _clean_text(section_value.get("title"), 100) or "Changes"
items_value = section_value.get("items")
if not isinstance(items_value, list):
continue
items: list[PatchChange] = []
for item in items_value:
if changes_seen >= MAX_PATCH_CHANGES:
break
if not isinstance(item, dict):
continue
text = _clean_text(item.get("text"), 4000)
type_key = _clean_text(item.get("type_key"), 64)
if text is None:
continue
items.append(
PatchChange(
text=text,
type_key=type_key or _type_key_from_title(title),
)
)
changes_seen += 1
if items:
sections.append(PatchSection(title=title, items=tuple(items)))
if changes_seen >= MAX_PATCH_CHANGES:
break
return tuple(sections)
def _patch_note_from_cache(value: object) -> PatchNote | None:
if not isinstance(value, dict):
return None
patch_id = value.get("id")
slug = _clean_text(value.get("slug"), 180)
title = _clean_text(value.get("title"), 200)
summary = _clean_text(value.get("summary"), 3000, allow_empty=True)
version = _clean_text(value.get("version"), 64, allow_none=True)
published_at = _clean_text(value.get("published_at"), 64, allow_none=True)
if (
isinstance(patch_id, bool)
or not isinstance(patch_id, int)
or slug is None
or _SLUG_PATTERN.fullmatch(slug) is None
or title is None
or summary is None
):
return None
sections_value = value.get("sections")
sections: list[PatchSection] = []
if isinstance(sections_value, list):
for section_value in sections_value[:MAX_PATCH_SECTIONS]:
if not isinstance(section_value, dict):
continue
section_title = _clean_text(section_value.get("title"), 100)
items_value = section_value.get("items")
if section_title is None or not isinstance(items_value, list):
continue
items: list[PatchChange] = []
for item in items_value:
if not isinstance(item, dict):
continue
text = _clean_text(item.get("text"), 4000)
type_key = _clean_text(item.get("type_key"), 64)
if text is not None and type_key is not None:
items.append(PatchChange(text=text, type_key=type_key))
if items:
sections.append(PatchSection(section_title, tuple(items)))
return PatchNote(
id=patch_id,
slug=slug,
title=title,
version=version,
tags=_parse_tags(value.get("tags")),
summary=summary,
sections=tuple(sections),
published_at=published_at,
)
def _clean_text(
value: object,
maximum_length: int,
*,
allow_empty: bool = False,
allow_none: bool = False,
) -> str | None:
if value is None and allow_none:
return None
if not isinstance(value, str):
return None
text = " ".join(value.split())
if len(text) > maximum_length or (not text and not allow_empty):
return None
return text
def _timestamp(value: object) -> float | None:
if isinstance(value, bool) or not isinstance(value, (int, float)):
return None
timestamp = float(value)
return timestamp if math.isfinite(timestamp) else None
def _format_timestamp(value: float | None) -> str | None:
if value is None:
return None
try:
return datetime.fromtimestamp(value, UTC).astimezone().strftime("%d %b %Y")
except (OverflowError, OSError, ValueError):
return None
def _type_key_from_title(title: str) -> str:
return re.sub(r"[^a-z0-9]+", "-", title.casefold()).strip("-") or "change"
def _write_patch_cache(path: Path, patches: tuple[PatchNote, ...]) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
temporary = path.with_name(f".{path.name}.{os.getpid()}.tmp")
try:
temporary.write_text(
json.dumps([asdict(patch) for patch in patches], indent=2) + "\n",
encoding="utf-8",
)
temporary.replace(path)
except OSError:
temporary.unlink(missing_ok=True)

View File

@@ -819,6 +819,7 @@ def _play_locked(
runner, runner,
start_option=start_option, start_option=start_option,
use_gamebox=not no_box, use_gamebox=not no_box,
concurrent_prefix_launch=reuse_prepared,
) )
except KeyboardInterrupt: except KeyboardInterrupt:
current_game_log = _current_game_log( current_game_log = _current_game_log(

213
ptlinux/dashboard_debug.py Normal file
View File

@@ -0,0 +1,213 @@
"""Read-only state snapshots for the launcher's hidden debug view."""
from __future__ import annotations
from dataclasses import asdict, dataclass
from datetime import datetime
import json
from pathlib import Path
from typing import Callable
from .import_state import game_import_requires_verification
from .paths import AppPaths
from .runtime import (
RuntimeErrorPT,
RuntimeSelection,
resolve_proton_sync_mode,
select_runtime,
)
from .settings import Settings
@dataclass(frozen=True)
class RuntimeDebugInfo:
"""Configured, selected, and last-prepared runtime state."""
requested_kind: str
selected_kind: str | None
executable: str | None
proton_path: str | None
prefix: str | None
sync_configured: str
sync_effective: str | None
prepared_kind: str | None
prepared_proton_path: str | None
prepared_at: str | None
marker_matches_selection: bool | None
dxvk_d3d9_override: str | None
error: str | None
@dataclass(frozen=True)
class VerificationDebugInfo:
"""Cached trust evidence from completed depot and import operations."""
import_state: str
import_verified_at: str | None
import_build: int | None
last_sync_at: str | None
last_full_verification_at: str | None
build: int | None
version: str | None
mode: str | None
files_total: int | None
files_downloaded: int | None
files_deleted: int | None
files_created: int | None
files_quarantined: int | None
RuntimeSelector = Callable[[Settings], RuntimeSelection]
SyncResolver = Callable[[str], str]
def inspect_runtime_debug(
paths: AppPaths,
settings: Settings,
*,
selector: RuntimeSelector = select_runtime,
sync_resolver: SyncResolver = resolve_proton_sync_mode,
) -> RuntimeDebugInfo:
"""Resolve the next runtime and compare it with the preparation marker."""
marker = _read_json_object(paths.game_runtime_marker)
prepared_runtime: dict[str, object] | None = None
if marker is not None:
raw_prepared_runtime = marker.get("runtime")
if isinstance(raw_prepared_runtime, dict):
prepared_runtime = raw_prepared_runtime
prepared_kind = _text(prepared_runtime.get("kind")) if prepared_runtime else None
prepared_proton_path = (
_text(prepared_runtime.get("proton_path")) if prepared_runtime else None
)
prepared_at = _text(marker.get("configured_at")) if marker else None
dxvk_override = _text(marker.get("dxvk_d3d9_override")) if marker else None
selection: RuntimeSelection | None = None
error: str | None = None
try:
selection = 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
prefix_kind = selected_kind or prepared_kind
prefix = None
if prefix_kind == "umu":
prefix = str(paths.umu_prefix)
elif prefix_kind in {"proton7", "proton7-container"}:
prefix = str(paths.proton_prefix)
sync_effective = None
if prefix_kind in {"proton7", "proton7-container"}:
try:
sync_effective = sync_resolver(settings.proton_sync)
except ValueError as exc:
error = error or (str(exc).strip() or type(exc).__name__)
marker_matches = (
prepared_runtime == asdict(selection)
if prepared_runtime is not None and selection is not None
else None
)
return RuntimeDebugInfo(
requested_kind=settings.runtime,
selected_kind=selected_kind,
executable=selection.executable if selection is not None else None,
proton_path=selection.proton_path if selection is not None else None,
prefix=prefix,
sync_configured=settings.proton_sync,
sync_effective=sync_effective,
prepared_kind=prepared_kind,
prepared_proton_path=prepared_proton_path,
prepared_at=prepared_at,
marker_matches_selection=marker_matches,
dxvk_d3d9_override=dxvk_override,
error=error,
)
def read_verification_debug(paths: AppPaths) -> VerificationDebugInfo:
"""Read cached verification state without hashing or changing game files."""
depot = _read_json_object(paths.native_depot_state)
if depot is not None and depot.get("dry_run") is not False:
depot = None
import_manifest_exists = paths.game_import_manifest.exists()
import_manifest = _read_json_object(paths.game_import_manifest)
if not import_manifest_exists:
import_state = "not applicable"
elif game_import_requires_verification(paths):
import_state = "pending"
else:
import_state = "verified"
import_verified_at = (
_text(import_manifest.get("verification_completed_at"))
if import_state == "verified" and import_manifest is not None
else None
)
import_build = (
_non_negative_int(import_manifest.get("verification_build"))
if import_state == "verified" and import_manifest is not None
else None
)
last_sync_at = _text(depot.get("completed_at")) if depot else None
full_verifications = [
value
for value in (
last_sync_at if depot is not None and depot.get("repair") is True else None,
import_verified_at,
)
if value is not None
]
last_full_verification_at = (
max(full_verifications, key=_iso_timestamp)
if full_verifications
else None
)
return VerificationDebugInfo(
import_state=import_state,
import_verified_at=import_verified_at,
import_build=import_build,
last_sync_at=last_sync_at,
last_full_verification_at=last_full_verification_at,
build=_non_negative_int(depot.get("build")) if depot else None,
version=_text(depot.get("version")) if depot else None,
mode=(
"full repair"
if depot is not None and depot.get("repair") is True
else "update"
if depot is not None and depot.get("repair") is False
else None
),
files_total=_non_negative_int(depot.get("files_total")) if depot else None,
files_downloaded=(
_non_negative_int(depot.get("files_downloaded")) if depot else None
),
files_deleted=_non_negative_int(depot.get("files_deleted")) if depot else None,
files_created=_non_negative_int(depot.get("files_created")) if depot else None,
files_quarantined=(
_non_negative_int(depot.get("files_quarantined")) if depot else None
),
)
def _read_json_object(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 _text(value: object) -> str | None:
return value.strip() if isinstance(value, str) and value.strip() else None
def _non_negative_int(value: object) -> int | None:
if isinstance(value, bool) or not isinstance(value, int) or value < 0:
return None
return value
def _iso_timestamp(value: str) -> float:
try:
return datetime.fromisoformat(value.replace("Z", "+00:00")).timestamp()
except (OverflowError, ValueError):
return float("-inf")

View File

@@ -0,0 +1,59 @@
"""Stable desktop identity shared by the launcher and Wine game window."""
from __future__ import annotations
import os
from pathlib import Path
from typing import Mapping
APPLICATION_NAME = "PristonTale EU"
APPLICATION_WM_CLASS = "PristonTaleEU"
APPLICATION_WM_INSTANCE = "pristontale-eu"
APPLICATION_ICON_FILENAME = "eu.pristontale.PTLinux.png"
APPLICATION_ICON_VARIABLE = "PTLINUX_ICON_PATH"
def find_application_icon(
environment: Mapping[str, str] | None = None,
) -> Path | None:
"""Find the installed icon, with source-tree fallbacks for development."""
env = os.environ if environment is None else environment
candidates: list[Path] = []
explicit = env.get(APPLICATION_ICON_VARIABLE)
if explicit:
candidates.append(Path(explicit).expanduser())
# Some distributions bundle the icon beside this module. The installer
# publishes the same asset through the user's XDG data directory.
candidates.append(Path(__file__).with_name(APPLICATION_ICON_FILENAME))
home_value = env.get("HOME")
if home_value:
home = Path(home_value).expanduser()
data_home_value = env.get("XDG_DATA_HOME")
data_home = (
Path(data_home_value).expanduser()
if data_home_value
else home / ".local/share"
)
candidates.append(data_home / "pixmaps" / APPLICATION_ICON_FILENAME)
default_icon = home / ".local/share/pixmaps" / APPLICATION_ICON_FILENAME
if default_icon not in candidates:
candidates.append(default_icon)
# The source checkout keeps installer assets under packaging/.
candidates.append(
Path(__file__).resolve().parent.parent
/ "packaging"
/ APPLICATION_ICON_FILENAME
)
seen: set[Path] = set()
for candidate in candidates:
resolved = candidate.resolve(strict=False)
if resolved in seen:
continue
seen.add(resolved)
if resolved.is_file():
return resolved
return None

View File

@@ -11,6 +11,7 @@ import shutil
from typing import Mapping from typing import Mapping
from .launcher import find_wine_tools from .launcher import find_wine_tools
from .runtime import find_steam_proton7
ARCH_FAMILY_DISTRIBUTIONS = frozenset( ARCH_FAMILY_DISTRIBUTIONS = frozenset(
@@ -176,19 +177,7 @@ def _package_hint(distribution: str) -> str | None:
def _has_steam_proton7(environment: Mapping[str, str]) -> bool: def _has_steam_proton7(environment: Mapping[str, str]) -> bool:
home_value = environment.get("HOME") return find_steam_proton7(environment) is not None
if not home_value:
return False
home = Path(home_value)
for steam_root in (
home / ".local/share/Steam",
home / ".steam/steam",
home / "snap/steam/common/.local/share/Steam",
home / ".var/app/com.valvesoftware.Steam/.local/share/Steam",
):
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]: def _platform_optional_tools(distribution: str, *, architecture: str) -> list[ToolStatus]:

View File

@@ -1,4 +1,4 @@
"""Native X11 command construction for the clean borderless game window.""" """Native X11 command construction for the managed game window."""
from __future__ import annotations from __future__ import annotations

View File

@@ -19,6 +19,20 @@ from .desktop_identity import (
WINDOW_TITLE = APPLICATION_NAME WINDOW_TITLE = APPLICATION_NAME
WINDOW_DISCOVERY_TIMEOUT_SECONDS = 60.0 WINDOW_DISCOVERY_TIMEOUT_SECONDS = 60.0
WINDOW_READY_DELAY_SECONDS = 0.5 WINDOW_READY_DELAY_SECONDS = 0.5
WINDOW_BORDERLESS_SETTLE_SECONDS = 0.1
WINDOW_SURFACE_REFRESH_DELAY_SECONDS = 0.15
WINDOW_FOCUS_DELAY_SECONDS = 0.2
WINDOW_STARTUP_GUARD_SECONDS = 12.0
WINDOW_STARTUP_POLL_SECONDS = 0.1
WINDOW_ACTIVATION_DELAY_SECONDS = 0.2
WINDOW_STARTUP_RECOVERY_LIMIT = 3
_X_REVERT_TO_PARENT = 2
_X_CURRENT_TIME = 0
_X_CLIENT_MESSAGE = 33
_X_SUBSTRUCTURE_NOTIFY_MASK = 1 << 19
_X_SUBSTRUCTURE_REDIRECT_MASK = 1 << 20
_X_WM_STATE_ICONIC = 3
_MOTIF_HINTS_DECORATIONS = 1 << 1
_X_ERROR_HANDLER = ctypes.CFUNCTYPE(ctypes.c_int, ctypes.c_void_p, ctypes.c_void_p) _X_ERROR_HANDLER = ctypes.CFUNCTYPE(ctypes.c_int, ctypes.c_void_p, ctypes.c_void_p)
@@ -78,6 +92,35 @@ class _XClassHint(ctypes.Structure):
] ]
class _XClientMessageData(ctypes.Union):
_fields_ = [
("bytes", ctypes.c_char * 20),
("shorts", ctypes.c_short * 10),
("longs", ctypes.c_long * 5),
]
class _XClientMessageEvent(ctypes.Structure):
_fields_ = [
("type", ctypes.c_int),
("serial", ctypes.c_ulong),
("send_event", ctypes.c_int),
("display", ctypes.c_void_p),
("window", ctypes.c_ulong),
("message_type", ctypes.c_ulong),
("format", ctypes.c_int),
("data", _XClientMessageData),
]
class _XEvent(ctypes.Union):
_fields_ = [
("type", ctypes.c_int),
("client_message", _XClientMessageEvent),
("padding", ctypes.c_long * 24),
]
class _X11: class _X11:
def __init__(self) -> None: def __init__(self) -> None:
library_path = ctypes.util.find_library("X11") or "libX11.so.6" library_path = ctypes.util.find_library("X11") or "libX11.so.6"
@@ -94,6 +137,7 @@ class _X11:
if not self.display: if not self.display:
self._restore_error_handler() self._restore_error_handler()
raise GameboxRuntimeError("could not open the host X11 display") raise GameboxRuntimeError("could not open the host X11 display")
self._atoms: dict[bytes, int] = {}
def close(self) -> None: def close(self) -> None:
try: try:
@@ -117,17 +161,24 @@ class _X11:
title = self._window_title(window) title = self._window_title(window)
if title is not None: if title is not None:
windows.append((window, title)) windows.append((window, title))
return _select_game_window(windows, title_fragment, excluded_window_ids) selected = _select_game_window(windows, title_fragment, excluded_window_ids)
if selected is None:
return None
window, title = selected
return self._client_window(window), title
def _root_windows(self) -> list[int]: def _root_windows(self) -> list[int]:
root = self.library.XDefaultRootWindow(self.display) root = self.library.XDefaultRootWindow(self.display)
return self._child_windows(root)
def _child_windows(self, window: int) -> list[int]:
root_return = ctypes.c_ulong() root_return = ctypes.c_ulong()
parent_return = ctypes.c_ulong() parent_return = ctypes.c_ulong()
children = ctypes.POINTER(ctypes.c_ulong)() children = ctypes.POINTER(ctypes.c_ulong)()
child_count = ctypes.c_uint() child_count = ctypes.c_uint()
status = self.library.XQueryTree( status = self.library.XQueryTree(
self.display, self.display,
root, window,
ctypes.byref(root_return), ctypes.byref(root_return),
ctypes.byref(parent_return), ctypes.byref(parent_return),
ctypes.byref(children), ctypes.byref(children),
@@ -141,11 +192,267 @@ class _X11:
if children: if children:
self.library.XFree(ctypes.cast(children, ctypes.c_void_p)) self.library.XFree(ctypes.cast(children, ctypes.c_void_p))
def map_window(self, window: int, *, borderless: bool) -> None: def _client_window(self, window: int) -> int:
"""Resolve a window-manager frame to the ICCCM client it contains."""
queue = [window]
visited: set[int] = set()
while queue and len(visited) < 64:
candidate = queue.pop(0)
if candidate in visited:
continue
visited.add(candidate)
if self._has_wm_state(candidate):
return candidate
queue.extend(self._child_windows(candidate))
return window
def _intern_atom(self, name: bytes, *, only_if_exists: bool) -> int:
"""Return one atom ID without repeating X-server lookups."""
try:
atoms = self._atoms
except AttributeError:
# Keep helpers usable in isolated tests that construct the mapper without
# opening a real X11 display.
atoms = {}
self._atoms = atoms
cached_atom = atoms.get(name)
if cached_atom is not None:
return cached_atom
atom = int(
self.library.XInternAtom(
self.display,
name,
int(only_if_exists),
)
)
# A missing only-if-exists lookup must remain retryable: a later call may
# create the atom or another client may publish it.
if atom != 0:
atoms[name] = atom
return atom
def _has_wm_state(self, window: int) -> bool:
wm_state = self._intern_atom(b"WM_STATE", only_if_exists=True)
if wm_state == 0:
return False
actual_type = ctypes.c_ulong()
actual_format = ctypes.c_int()
item_count = ctypes.c_ulong()
bytes_after = ctypes.c_ulong()
value = ctypes.POINTER(ctypes.c_ubyte)()
status = self.library.XGetWindowProperty(
self.display,
window,
wm_state,
0,
0,
0,
0,
ctypes.byref(actual_type),
ctypes.byref(actual_format),
ctypes.byref(item_count),
ctypes.byref(bytes_after),
ctypes.byref(value),
)
try:
return status == 0 and actual_type.value != 0
finally:
if value:
self.library.XFree(ctypes.cast(value, ctypes.c_void_p))
def window_state(self, window: int) -> int | None:
"""Return the ICCCM WM_STATE value for a managed client."""
values = self._long_property(window, b"WM_STATE", 2)
if not values:
return None
return values[0]
def window_is_borderless(self, window: int) -> bool | None:
"""Return whether the client currently requests no Motif decorations."""
hints = self._long_property(window, b"_MOTIF_WM_HINTS", 5)
if hints is None or len(hints) < 3:
return None
if hints[0] & _MOTIF_HINTS_DECORATIONS == 0:
return False
return hints[2] == 0
def _long_property(
self,
window: int,
name: bytes,
maximum_items: int,
) -> tuple[int, ...] | None:
atom = self._intern_atom(name, only_if_exists=True)
if atom == 0:
return None
actual_type = ctypes.c_ulong()
actual_format = ctypes.c_int()
item_count = ctypes.c_ulong()
bytes_after = ctypes.c_ulong()
value = ctypes.POINTER(ctypes.c_ubyte)()
status = self.library.XGetWindowProperty(
self.display,
window,
atom,
0,
maximum_items,
0,
0,
ctypes.byref(actual_type),
ctypes.byref(actual_format),
ctypes.byref(item_count),
ctypes.byref(bytes_after),
ctypes.byref(value),
)
try:
if (
status != 0
or actual_type.value == 0
or actual_format.value != 32
or not value
):
return None
longs = ctypes.cast(value, ctypes.POINTER(ctypes.c_ulong))
return tuple(
int(longs[index] & 0xFFFFFFFF)
for index in range(min(item_count.value, maximum_items))
)
finally:
if value:
self.library.XFree(ctypes.cast(value, ctypes.c_void_p))
def window_size(self, window: int) -> tuple[int, int] | None:
"""Return the current X11 client dimensions."""
root = ctypes.c_ulong()
x = ctypes.c_int()
y = ctypes.c_int()
width = ctypes.c_uint()
height = ctypes.c_uint()
border_width = ctypes.c_uint()
depth = ctypes.c_uint()
if self.library.XGetGeometry(
self.display,
window,
ctypes.byref(root),
ctypes.byref(x),
ctypes.byref(y),
ctypes.byref(width),
ctypes.byref(height),
ctypes.byref(border_width),
ctypes.byref(depth),
) == 0:
return None
return width.value, height.value
def map_window(
self,
window: int,
*,
borderless: bool,
) -> tuple[int, int] | None:
target_size = self.window_size(window) if borderless else None
if borderless: if borderless:
self._remove_decorations(window) self._remove_decorations(window)
self.library.XMapRaised(self.display, window) self.library.XMapRaised(self.display, window)
self.library.XFlush(self.display) self.library.XFlush(self.display)
return target_size
def refresh_render_surface(
self,
window: int,
*,
target_size: tuple[int, int] | None = None,
) -> bool:
"""Nudge the mapped client size once after borderless styling.
Some Wine/DXVK combinations retain the host desktop-sized render surface
after the top-level game window becomes borderless. A one-pixel resize followed
by an exact restore makes Wine reconcile the surface with the real window.
Never iconify or unmap a Wine client for this refresh. Wine can process the
resulting state change after startup and minimize the game at its login screen.
"""
if target_size is None:
target_size = self.window_size(window)
if target_size is None:
return False
original_width, original_height = target_size
if original_width > 1:
nudged_width = original_width - 1
nudged_height = original_height
elif original_height > 1:
nudged_width = original_width
nudged_height = original_height - 1
else:
return False
self.library.XResizeWindow(
self.display,
window,
nudged_width,
nudged_height,
)
self.library.XSync(self.display, 0)
try:
time.sleep(WINDOW_SURFACE_REFRESH_DELAY_SECONDS)
finally:
# Always restore the exact starting dimensions if the short wait is
# interrupted. The game remains mapped for the entire operation.
self.library.XResizeWindow(
self.display,
window,
original_width,
original_height,
)
self.library.XSync(self.display, 0)
return True
def focus_window(self, window: int) -> None:
"""Raise the mapped game client and give it keyboard focus."""
self.library.XMapRaised(self.display, window)
self.library.XSetInputFocus(
self.display,
window,
_X_REVERT_TO_PARENT,
_X_CURRENT_TIME,
)
self.library.XSync(self.display, 0)
def activate_window(self, window: int) -> bool:
"""Ask the window manager to restore and activate one client."""
active_window = self._intern_atom(
b"_NET_ACTIVE_WINDOW",
only_if_exists=False,
)
if active_window == 0:
return False
event = _XEvent()
event.client_message.type = _X_CLIENT_MESSAGE
event.client_message.serial = 0
event.client_message.send_event = 1
event.client_message.display = self.display
event.client_message.window = window
event.client_message.message_type = active_window
event.client_message.format = 32
event.client_message.data.longs[0] = 1
event.client_message.data.longs[1] = _X_CURRENT_TIME
status = self.library.XSendEvent(
self.display,
self.library.XDefaultRootWindow(self.display),
0,
_X_SUBSTRUCTURE_NOTIFY_MASK | _X_SUBSTRUCTURE_REDIRECT_MASK,
ctypes.byref(event),
)
self.library.XSync(self.display, 0)
return status != 0
def set_window_identity( def set_window_identity(
self, self,
@@ -176,7 +483,7 @@ class _X11:
def suppress_wm_ping(self, window: int) -> bool: def suppress_wm_ping(self, window: int) -> bool:
"""Stop the WM hang prompt for one window, preserving its other protocols.""" """Stop the WM hang prompt for one window, preserving its other protocols."""
ping_atom = int(self.library.XInternAtom(self.display, b"_NET_WM_PING", 0)) ping_atom = self._intern_atom(b"_NET_WM_PING", only_if_exists=False)
if ping_atom == 0: if ping_atom == 0:
return False return False
protocols = self._wm_protocols(window) protocols = self._wm_protocols(window)
@@ -227,7 +534,7 @@ class _X11:
self.library.XFree(ctypes.cast(protocols, ctypes.c_void_p)) self.library.XFree(ctypes.cast(protocols, ctypes.c_void_p))
def _remove_decorations(self, window: int) -> None: def _remove_decorations(self, window: int) -> None:
atom = self.library.XInternAtom(self.display, b"_MOTIF_WM_HINTS", 0) atom = self._intern_atom(b"_MOTIF_WM_HINTS", only_if_exists=False)
if atom == 0: if atom == 0:
return return
hints = _MotifWmHints( hints = _MotifWmHints(
@@ -304,6 +611,21 @@ class _X11:
ctypes.c_int, ctypes.c_int,
] ]
self.library.XChangeProperty.restype = ctypes.c_int self.library.XChangeProperty.restype = ctypes.c_int
self.library.XGetWindowProperty.argtypes = [
ctypes.c_void_p,
ctypes.c_ulong,
ctypes.c_ulong,
ctypes.c_long,
ctypes.c_long,
ctypes.c_int,
ctypes.c_ulong,
ctypes.POINTER(ctypes.c_ulong),
ctypes.POINTER(ctypes.c_int),
ctypes.POINTER(ctypes.c_ulong),
ctypes.POINTER(ctypes.c_ulong),
ctypes.POINTER(ctypes.POINTER(ctypes.c_ubyte)),
]
self.library.XGetWindowProperty.restype = ctypes.c_int
self.library.XSetClassHint.argtypes = [ self.library.XSetClassHint.argtypes = [
ctypes.c_void_p, ctypes.c_void_p,
ctypes.c_ulong, ctypes.c_ulong,
@@ -326,6 +648,40 @@ class _X11:
self.library.XSetWMProtocols.restype = ctypes.c_int self.library.XSetWMProtocols.restype = ctypes.c_int
self.library.XMapRaised.argtypes = [ctypes.c_void_p, ctypes.c_ulong] self.library.XMapRaised.argtypes = [ctypes.c_void_p, ctypes.c_ulong]
self.library.XMapRaised.restype = ctypes.c_int self.library.XMapRaised.restype = ctypes.c_int
self.library.XGetGeometry.argtypes = [
ctypes.c_void_p,
ctypes.c_ulong,
ctypes.POINTER(ctypes.c_ulong),
ctypes.POINTER(ctypes.c_int),
ctypes.POINTER(ctypes.c_int),
ctypes.POINTER(ctypes.c_uint),
ctypes.POINTER(ctypes.c_uint),
ctypes.POINTER(ctypes.c_uint),
ctypes.POINTER(ctypes.c_uint),
]
self.library.XGetGeometry.restype = ctypes.c_int
self.library.XResizeWindow.argtypes = [
ctypes.c_void_p,
ctypes.c_ulong,
ctypes.c_uint,
ctypes.c_uint,
]
self.library.XResizeWindow.restype = ctypes.c_int
self.library.XSetInputFocus.argtypes = [
ctypes.c_void_p,
ctypes.c_ulong,
ctypes.c_int,
ctypes.c_ulong,
]
self.library.XSetInputFocus.restype = ctypes.c_int
self.library.XSendEvent.argtypes = [
ctypes.c_void_p,
ctypes.c_ulong,
ctypes.c_int,
ctypes.c_long,
ctypes.POINTER(_XEvent),
]
self.library.XSendEvent.restype = ctypes.c_int
self.library.XSync.argtypes = [ctypes.c_void_p, ctypes.c_int] self.library.XSync.argtypes = [ctypes.c_void_p, ctypes.c_int]
self.library.XSync.restype = ctypes.c_int self.library.XSync.restype = ctypes.c_int
self.library.XFlush.argtypes = [ctypes.c_void_p] self.library.XFlush.argtypes = [ctypes.c_void_p]
@@ -367,7 +723,40 @@ def _prepare_game_window(
file=sys.stderr, file=sys.stderr,
flush=True, flush=True,
) )
mapper.map_window(window, borderless=borderless) target_size = mapper.map_window(window, borderless=borderless)
if not borderless:
return
if target_size is not None:
time.sleep(WINDOW_BORDERLESS_SETTLE_SECONDS)
try:
if target_size is None:
refreshed = mapper.refresh_render_surface(window)
else:
refreshed = mapper.refresh_render_surface(
window,
target_size=target_size,
)
except Exception as exc:
print(
f"Warning: could not synchronize the borderless game surface ({exc}); "
"continuing normally",
file=sys.stderr,
flush=True,
)
else:
if refreshed:
print(
"Borderless game surface synchronized with its window.",
file=sys.stderr,
flush=True,
)
else:
print(
"Warning: the window manager declined the borderless surface "
"refresh; continuing normally",
file=sys.stderr,
flush=True,
)
def _apply_game_window_identity(mapper: _X11, window: int) -> None: def _apply_game_window_identity(mapper: _X11, window: int) -> None:
@@ -382,6 +771,107 @@ def _apply_game_window_identity(mapper: _X11, window: int) -> None:
) )
def _focus_game_window(mapper: _X11, window: int) -> None:
"""Let the dashboard process its ready event before focusing the game."""
time.sleep(WINDOW_FOCUS_DELAY_SECONDS)
try:
mapper.focus_window(window)
except Exception as exc:
print(
f"Warning: could not focus the game window ({exc}); continuing normally",
file=sys.stderr,
flush=True,
)
def _guard_game_window_startup(
mapper: _X11,
process: subprocess.Popen[bytes],
window: int,
*,
borderless: bool,
) -> None:
"""Recover Wine's login-screen window transition without hiding the client."""
deadline = time.monotonic() + WINDOW_STARTUP_GUARD_SECONDS
try:
initial_borderless = (
mapper.window_is_borderless(window) if borderless else None
)
# A readable decorated value can already be Wine's fast login reset. The
# initial preparation immediately before this guard requested borderless.
borderless_was_applied = borderless and initial_borderless is not None
except Exception as exc:
print(
f"Warning: could not monitor the game startup window ({exc}); "
"continuing normally",
file=sys.stderr,
flush=True,
)
return
recovery_count = 0
while process.poll() is None and time.monotonic() < deadline:
try:
state = mapper.window_state(window)
borderless_now = (
mapper.window_is_borderless(window) if borderless else None
)
except Exception as exc:
print(
f"Warning: could not inspect the game startup window ({exc}); "
"continuing normally",
file=sys.stderr,
flush=True,
)
return
became_iconic = state == _X_WM_STATE_ICONIC
login_style_reset = borderless_was_applied and borderless_now is False
if not became_iconic and not login_style_reset:
time.sleep(WINDOW_STARTUP_POLL_SECONDS)
continue
recovery_count += 1
reason = "was minimized" if became_iconic else "changed its window layout"
print(
f"Game startup window {reason}; restoring and refocusing it.",
file=sys.stderr,
flush=True,
)
try:
mapper.activate_window(window)
time.sleep(WINDOW_ACTIVATION_DELAY_SECONDS)
_prepare_game_window(
mapper,
window,
borderless=borderless,
suppress_wm_ping=False,
)
mapper.activate_window(window)
mapper.focus_window(window)
if borderless:
borderless_was_applied = (
mapper.window_is_borderless(window) is True
)
except Exception as exc:
print(
f"Warning: could not recover the game startup window ({exc}); "
"continuing normally",
file=sys.stderr,
flush=True,
)
return
if recovery_count >= WINDOW_STARTUP_RECOVERY_LIMIT:
print(
"Warning: the game repeatedly changed its startup window; "
"stopped automatic recovery.",
file=sys.stderr,
flush=True,
)
return
time.sleep(WINDOW_STARTUP_POLL_SECONDS)
def run_game_window( def run_game_window(
command: Sequence[str], command: Sequence[str],
window_title: str = WINDOW_TITLE, window_title: str = WINDOW_TITLE,
@@ -450,6 +940,13 @@ def run_game_window(
file=sys.stderr, file=sys.stderr,
flush=True, flush=True,
) )
_focus_game_window(mapper, game_window)
_guard_game_window_startup(
mapper,
process,
game_window,
borderless=borderless,
)
return int(process.wait()) return int(process.wait())
time.sleep(0.1) time.sleep(0.1)
if process.poll() is None: if process.poll() is None:

393
ptlinux/identity_probe.py Normal file
View File

@@ -0,0 +1,393 @@
"""Privacy-safe checks for the Windows identity inputs used by the game client."""
from __future__ import annotations
from dataclasses import dataclass
import hashlib
import json
from pathlib import Path
import re
from .command import CommandRunner
from .paths import AppPaths
from .runtime import RuntimeSelection, run_runtime_prefix_command
IDENTITY_PROBE_SHA256 = "677cb5aa19f8dac65b982462bb30974feed05399fdd759765c3b6e802d4c838b"
IDENTITY_PROBE_TIMEOUT_SECONDS = 45.0
IDENTITY_PROBE_PRIVACY = (
"Only success flags, row counts, value lengths, VARIANT types, adapter type, "
"and generic-value quality flags are collected. Computer names, serial numbers, "
"MAC addresses, and WMI values never leave the Windows helper."
)
WMI_PROPERTY_NAMES = (
"processor_id",
"processor_unique_id",
"processor_manufacturer",
"processor_max_clock_speed",
"bios_release_date",
"bios_serial_number",
"bios_identification_code",
"bios_smbios_version",
"bios_manufacturer",
"baseboard_serial_number",
"baseboard_name",
"baseboard_manufacturer",
"baseboard_model",
"video_name",
"video_driver_version",
)
HRESULT_PATTERN = re.compile(r"0x[0-9A-F]{8}\Z")
@dataclass(frozen=True)
class IdentityProbeResult:
helper: str
helper_sha256: str | None
helper_verified: bool
ran: bool
ready: bool | None
error: str | None
report: dict[str, object] | None
observations: tuple[str, ...]
warnings: tuple[str, ...]
def packaged_identity_probe() -> Path:
return Path(__file__).resolve().parent / "native" / "pt-identity-probe.exe"
def probe_client_identity(
paths: AppPaths,
selection: RuntimeSelection,
runner: CommandRunner,
*,
timeout_seconds: float = IDENTITY_PROBE_TIMEOUT_SECONDS,
) -> IdentityProbeResult:
"""Check identity API availability without returning the underlying values."""
helper = packaged_identity_probe()
helper_sha256 = _sha256_if_file(helper)
helper_verified = helper_sha256 == IDENTITY_PROBE_SHA256
if not helper_verified:
error = (
"the packaged privacy-safe identity helper is missing"
if helper_sha256 is None
else "the packaged privacy-safe identity helper failed verification"
)
return IdentityProbeResult(
helper=str(helper),
helper_sha256=helper_sha256,
helper_verified=False,
ran=False,
ready=None,
error=error,
report=None,
observations=(),
warnings=(error,),
)
try:
outcome = run_runtime_prefix_command(
paths,
selection,
runner,
[str(helper)],
check=False,
timeout_seconds=timeout_seconds,
)
except (OSError, RuntimeError) as exc:
error = f"the client-identity helper could not start: {exc}"
return IdentityProbeResult(
helper=str(helper),
helper_sha256=helper_sha256,
helper_verified=True,
ran=False,
ready=None,
error=error,
report=None,
observations=(),
warnings=(error,),
)
if outcome.timed_out:
error = "the client-identity helper timed out"
return _failed_result(helper, helper_sha256, error, ran=True)
if outcome.returncode != 0:
status = "unknown" if outcome.returncode is None else str(outcome.returncode)
error = f"the client-identity helper exited with status {status}"
return _failed_result(helper, helper_sha256, error, ran=True)
try:
report = _parse_privacy_safe_report(outcome.stdout)
except (ValueError, json.JSONDecodeError) as exc:
error = f"the client-identity helper returned invalid data: {exc}"
return _failed_result(helper, helper_sha256, error, ran=True)
warnings = _identity_warnings(report)
return IdentityProbeResult(
helper=str(helper),
helper_sha256=helper_sha256,
helper_verified=True,
ran=True,
ready=not warnings,
error=None,
report=report,
observations=_identity_observations(report),
warnings=warnings,
)
def _failed_result(
helper: Path,
helper_sha256: str | None,
error: str,
*,
ran: bool,
) -> IdentityProbeResult:
return IdentityProbeResult(
helper=str(helper),
helper_sha256=helper_sha256,
helper_verified=True,
ran=ran,
ready=None,
error=error,
report=None,
observations=(),
warnings=(error,),
)
def _parse_privacy_safe_report(stdout: str) -> dict[str, object]:
lines = [line.strip() for line in stdout.splitlines() if line.strip()]
if not lines:
raise ValueError("empty output")
value = json.loads(lines[-1])
root = _object(value, "root")
if _integer(root.get("schema"), "schema") != 2:
raise ValueError("unsupported schema")
computer = _object(root.get("computer_name"), "computer_name")
volume = _object(root.get("volume"), "volume")
adapters = _object(root.get("adapters"), "adapters")
firmware = _object(root.get("firmware"), "firmware")
wmi = _object(root.get("wmi"), "wmi")
properties = _object(wmi.get("properties"), "wmi.properties")
safe_properties: dict[str, object] = {}
for name in WMI_PROPERTY_NAMES:
entry = _object(properties.get(name), f"wmi.properties.{name}")
safe_properties[name] = {
"result": _hresult(entry.get("result"), f"{name}.result"),
"variant_type": _integer(entry.get("variant_type"), f"{name}.variant_type"),
"rows": _integer(entry.get("rows"), f"{name}.rows"),
"nonempty_rows": _integer(
entry.get("nonempty_rows"),
f"{name}.nonempty_rows",
),
"maximum_length": _integer(
entry.get("maximum_length"),
f"{name}.maximum_length",
),
"placeholder_rows": _integer(
entry.get("placeholder_rows"),
f"{name}.placeholder_rows",
),
"wine_rows": _integer(
entry.get("wine_rows"),
f"{name}.wine_rows",
),
}
return {
"schema": 2,
"computer_name": {
"success": _boolean(computer.get("success"), "computer_name.success"),
"nonempty": _boolean(computer.get("nonempty"), "computer_name.nonempty"),
"length": _integer(computer.get("length"), "computer_name.length"),
},
"volume": {
"success": _boolean(volume.get("success"), "volume.success"),
"serial_nonzero": _boolean(
volume.get("serial_nonzero"),
"volume.serial_nonzero",
),
},
"adapters": {
"status": _integer(adapters.get("status"), "adapters.status"),
"count": _integer(adapters.get("count"), "adapters.count"),
"first_address_length": _integer(
adapters.get("first_address_length"),
"adapters.first_address_length",
),
"first_address_nonzero": _boolean(
adapters.get("first_address_nonzero"),
"adapters.first_address_nonzero",
),
"first_address_locally_administered": _boolean(
adapters.get("first_address_locally_administered"),
"adapters.first_address_locally_administered",
),
"first_address_multicast": _boolean(
adapters.get("first_address_multicast"),
"adapters.first_address_multicast",
),
"first_type": _integer(adapters.get("first_type"), "adapters.first_type"),
},
"firmware": {
"raw_smbios_bytes": _integer(
firmware.get("raw_smbios_bytes"),
"firmware.raw_smbios_bytes",
),
},
"wmi": {
"com_initialize": _hresult(wmi.get("com_initialize"), "wmi.com_initialize"),
"com_security": _hresult(wmi.get("com_security"), "wmi.com_security"),
"connect": _hresult(wmi.get("connect"), "wmi.connect"),
"properties": safe_properties,
},
}
def _identity_warnings(report: dict[str, object]) -> tuple[str, ...]:
warnings: list[str] = []
computer = _object(report["computer_name"], "computer_name")
volume = _object(report["volume"], "volume")
adapters = _object(report["adapters"], "adapters")
wmi = _object(report["wmi"], "wmi")
properties = _object(wmi["properties"], "wmi.properties")
computer_success = _boolean(computer["success"], "computer_name.success")
computer_nonempty = _boolean(computer["nonempty"], "computer_name.nonempty")
volume_success = _boolean(volume["success"], "volume.success")
volume_serial_nonzero = _boolean(volume["serial_nonzero"], "volume.serial_nonzero")
adapter_status = _integer(adapters["status"], "adapters.status")
adapter_count = _integer(adapters["count"], "adapters.count")
first_address_length = _integer(
adapters["first_address_length"],
"adapters.first_address_length",
)
first_address_nonzero = _boolean(
adapters["first_address_nonzero"],
"adapters.first_address_nonzero",
)
if not computer_success or not computer_nonempty:
warnings.append("Wine did not provide the computer name used by the login request.")
if not volume_success or not volume_serial_nonzero:
warnings.append("Wine did not provide a usable C: volume identity.")
if (
adapter_status != 0
or adapter_count == 0
or first_address_length < 6
or not first_address_nonzero
):
warnings.append(
"Wine did not provide a usable first network-adapter identity; the game "
"would send its UUID fallback instead."
)
if _hresult(wmi["connect"], "wmi.connect") != "0x00000000":
warnings.append("Wine could not connect to ROOT\\CIMV2 for the login identity query.")
else:
if not _any_nonempty(properties, ("processor_id", "processor_unique_id")):
warnings.append("Wine returned no processor identifier through WMI.")
if not _any_nonempty(
properties,
(
"bios_release_date",
"bios_serial_number",
"bios_identification_code",
"bios_smbios_version",
"bios_manufacturer",
),
):
warnings.append("Wine returned no BIOS identity fields through WMI.")
if not _any_nonempty(
properties,
(
"baseboard_serial_number",
"baseboard_name",
"baseboard_manufacturer",
"baseboard_model",
),
):
warnings.append("Wine returned no baseboard identity fields through WMI.")
if not _any_nonempty(properties, ("video_name",)):
warnings.append("Wine returned no video-controller name through WMI.")
return tuple(warnings)
def _identity_observations(report: dict[str, object]) -> tuple[str, ...]:
observations: list[str] = []
adapters = _object(report["adapters"], "adapters")
if _boolean(
adapters["first_address_locally_administered"],
"adapters.first_address_locally_administered",
):
observations.append(
"The first network-adapter address is locally administered."
)
if _boolean(
adapters["first_address_multicast"],
"adapters.first_address_multicast",
):
observations.append("The first network-adapter address has the multicast bit set.")
wmi = _object(report["wmi"], "wmi")
properties = _object(wmi["properties"], "wmi.properties")
placeholder_names: list[str] = []
wine_names: list[str] = []
for name in WMI_PROPERTY_NAMES:
entry = _object(properties[name], name)
if _integer(entry["placeholder_rows"], f"{name}.placeholder_rows") > 0:
placeholder_names.append(name)
if _integer(entry["wine_rows"], f"{name}.wine_rows") > 0:
wine_names.append(name)
if placeholder_names:
observations.append(
"Common placeholder WMI values were observed in: "
+ ", ".join(placeholder_names)
)
if wine_names:
observations.append(
"Wine-labelled WMI values were observed in: " + ", ".join(wine_names)
)
return tuple(observations)
def _any_nonempty(properties: dict[str, object], names: tuple[str, ...]) -> bool:
for name in names:
entry = _object(properties[name], name)
if (
_hresult(entry["result"], f"{name}.result") == "0x00000000"
and _integer(entry["nonempty_rows"], f"{name}.nonempty_rows") > 0
):
return True
return False
def _object(value: object, label: str) -> dict[str, object]:
if not isinstance(value, dict) or not all(isinstance(key, str) for key in value):
raise ValueError(f"{label} must be an object")
return value
def _boolean(value: object, label: str) -> bool:
if not isinstance(value, bool):
raise ValueError(f"{label} must be a boolean")
return value
def _integer(value: object, label: str) -> int:
if isinstance(value, bool) or not isinstance(value, int) or value < 0 or value > 0xFFFFFFFF:
raise ValueError(f"{label} must be an unsigned 32-bit integer")
return value
def _hresult(value: object, label: str) -> str:
if not isinstance(value, str) or HRESULT_PATTERN.fullmatch(value) is None:
raise ValueError(f"{label} must be a hexadecimal HRESULT")
return value
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

85
ptlinux/import_state.py Normal file
View File

@@ -0,0 +1,85 @@
"""Persistent trust state for game trees seeded from another installation."""
from __future__ import annotations
from datetime import datetime
import json
import os
from pathlib import Path
from uuid import uuid4
from .paths import AppPaths
IMPORT_STATE_SCHEMA = 2
def game_import_requires_verification(paths: AppPaths) -> bool:
"""Return whether a managed imported tree still needs a full depot repair."""
manifest = paths.game_import_manifest
if not manifest.exists():
return False
try:
value = json.loads(manifest.read_text(encoding="utf-8"))
except (OSError, UnicodeError, json.JSONDecodeError):
return True
if not isinstance(value, dict):
return True
build = value.get("verification_build")
completed_at = value.get("verification_completed_at")
return not (
value.get("verification_required") is False
and isinstance(build, int)
and not isinstance(build, bool)
and build > 0
and isinstance(completed_at, str)
and bool(completed_at.strip())
)
def mark_game_import_verified(
paths: AppPaths,
*,
build: int,
files_quarantined: int,
quarantine_directory: str | None,
) -> None:
"""Record a completed full-manifest verification of an imported tree."""
if isinstance(build, bool) or build <= 0:
raise ValueError("verified import build must be positive")
if files_quarantined < 0:
raise ValueError("quarantined file count cannot be negative")
try:
value = json.loads(paths.game_import_manifest.read_text(encoding="utf-8"))
except (OSError, UnicodeError, json.JSONDecodeError):
value = {}
if not isinstance(value, dict):
value = {}
value.update(
{
"schema": IMPORT_STATE_SCHEMA,
"verification_required": False,
"verification_build": build,
"verification_completed_at": datetime.now().astimezone().isoformat(),
"files_quarantined": files_quarantined,
"quarantine_directory": quarantine_directory,
}
)
_write_json(paths.game_import_manifest, value)
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()}.{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

Binary file not shown.

View File

@@ -20,6 +20,8 @@ OFFICIAL_LOGO_URL = "https://pristontale.eu/public/img/pristontale/logo/logo_sma
BROADCAST_ENDPOINT = "http://5.135.35.188:90/broadcasts" BROADCAST_ENDPOINT = "http://5.135.35.188:90/broadcasts"
MAX_NEWS_BYTES = 2 * 1024 * 1024 MAX_NEWS_BYTES = 2 * 1024 * 1024
MAX_NEWS_IMAGE_BYTES = 12 * 1024 * 1024 MAX_NEWS_IMAGE_BYTES = 12 * 1024 * 1024
MAX_BROADCAST_AUTHOR_CHARACTERS = 100
MAX_BROADCAST_CONTENT_CHARACTERS = 2000
PNG_SIGNATURE = b"\x89PNG\r\n\x1a\n" PNG_SIGNATURE = b"\x89PNG\r\n\x1a\n"
@@ -241,26 +243,41 @@ def _parse_latest_broadcast(payload: bytes) -> Broadcast:
value = json.loads(payload.decode("utf-8-sig")) value = json.loads(payload.decode("utf-8-sig"))
if not isinstance(value, list) or not value or not isinstance(value[0], dict): if not isinstance(value, list) or not value or not isinstance(value[0], dict):
raise NewsError("broadcast response does not contain a latest broadcast") raise NewsError("broadcast response does not contain a latest broadcast")
latest = value[0] return _broadcast_from_mapping(value[0])
author = latest.get("author")
content = latest.get("content")
def _broadcast_from_mapping(latest: dict[object, object]) -> Broadcast:
author = _bounded_text(latest.get("author"), MAX_BROADCAST_AUTHOR_CHARACTERS)
content = _bounded_text(latest.get("content"), MAX_BROADCAST_CONTENT_CHARACTERS)
url = _broadcast_url(latest.get("url")) url = _broadcast_url(latest.get("url"))
if ( if (
not isinstance(author, str) author is None
or not author.strip() or content is None
or not isinstance(content, str)
or not content.strip()
or url is None or url is None
): ):
raise NewsError("latest broadcast is incomplete") raise NewsError("latest broadcast is incomplete")
published_at = (
_format_timestamp(latest.get("timestamp"))
if "timestamp" in latest
else _bounded_text(latest.get("published_at"), 64)
)
return Broadcast( return Broadcast(
author=author.strip(), author=author,
content=content.strip(), content=content,
published_at=_format_timestamp(latest.get("timestamp")), published_at=published_at,
url=url, url=url,
) )
def _bounded_text(value: object, maximum_characters: int) -> str | None:
if not isinstance(value, str):
return None
text = value.strip()
if not text or len(text) > maximum_characters:
return None
return text
def _format_timestamp(value: object) -> str | None: def _format_timestamp(value: object) -> str | None:
if isinstance(value, bool) or not isinstance(value, (int, float)): if isinstance(value, bool) or not isinstance(value, (int, float)):
return None return None
@@ -288,10 +305,16 @@ def _image_url(value: object) -> str | None:
def _broadcast_url(value: object) -> str | None: def _broadcast_url(value: object) -> str | None:
if not isinstance(value, str) or not value.strip(): if not isinstance(value, str) or not value.strip():
return None return None
parsed = urlparse(value) url = value.strip()
if parsed.scheme not in {"http", "https"} or not parsed.netloc: parsed = urlparse(url)
if (
parsed.scheme.casefold() != "https"
or not parsed.netloc
or parsed.username is not None
or parsed.password is not None
):
return None return None
return value.strip() return url
def _read_cache(path: Path) -> tuple[NewsArticle, ...]: def _read_cache(path: Path) -> tuple[NewsArticle, ...]:
@@ -320,8 +343,8 @@ def _write_cache(path: Path, articles: tuple[NewsArticle, ...]) -> None:
def _read_broadcast_cache(path: Path) -> Broadcast | None: def _read_broadcast_cache(path: Path) -> Broadcast | None:
try: try:
value = json.loads(path.read_text(encoding="utf-8")) value = json.loads(path.read_text(encoding="utf-8"))
return Broadcast(**value) if isinstance(value, dict) else None return _broadcast_from_mapping(value) if isinstance(value, dict) else None
except (OSError, UnicodeError, TypeError, json.JSONDecodeError): except (OSError, UnicodeError, NewsError, json.JSONDecodeError):
return None return None

View File

@@ -44,15 +44,46 @@ PATCHED_V21_GAME_DLL_SHA256 = (
RETIRED_POST_AUTH_GAME_DLL_SHA256 = ( RETIRED_POST_AUTH_GAME_DLL_SHA256 = (
"93cd27284eb294753da1111c99bfc9689f82b6d4a59302aee9fdbd8cb52452dd" "93cd27284eb294753da1111c99bfc9689f82b6d4a59302aee9fdbd8cb52452dd"
) )
RUNTIME_VIEW_CONFIGURATION_VERSION = 1 RUNTIME_VIEW_CONFIGURATION_VERSION = 2
class OverlayError(RuntimeError): class OverlayError(RuntimeError):
"""Raised when a safe runtime view cannot be built.""" """Raised when a safe runtime view cannot be built."""
@dataclass(frozen=True)
class RuntimeFileIdentity:
"""Cheap change detector for one previously hash-verified runtime input."""
name: str
device: int
inode: int
mode: int
size: int
mtime_ns: int
ctime_ns: int
@dataclass(frozen=True) @dataclass(frozen=True)
class OverlayResult: class OverlayResult:
source_game: str
runtime_view: str
source_scene_sha256: str
runtime_scene_sha256: str
scene_action: str
binary_action: str
game_exe_sha256: str
game_dll_sha256: str
input_identities: tuple[RuntimeFileIdentity, ...]
runtime_scene_identity: RuntimeFileIdentity
created_at: str
links_created: int
@dataclass(frozen=True)
class LegacyOverlayResult:
"""Version-one runtime manifest retained only for a one-time fast migration."""
source_game: str source_game: str
runtime_view: str runtime_view: str
source_scene_sha256: str source_scene_sha256: str
@@ -94,8 +125,9 @@ def inspect_runtime_compatibility(
This preflight is intentionally read-only: it never normalizes a scene, This preflight is intentionally read-only: it never normalizes a scene,
builds the verified runtime DLL, prepares save data, or replaces the runtime builds the verified runtime DLL, prepares save data, or replaces the runtime
view. The normal preparation path still performs every authoritative check view. Preparation reuses a prior authoritative result only while every input
immediately before launch. identity and runtime-view link remains unchanged; otherwise it repeats the
full checks before launch.
""" """
requested_source = source_game.expanduser() requested_source = source_game.expanduser()
@@ -105,6 +137,9 @@ def inspect_runtime_compatibility(
try: try:
source = requested_source.resolve(strict=True) source = requested_source.resolve(strict=True)
_validate_game_tree(source) _validate_game_tree(source)
cached = _cached_runtime_compatibility(paths, source)
if cached is not None:
return cached
source_scene = source / WHISPERING_VALE_RELATIVE source_scene = source / WHISPERING_VALE_RELATIVE
inspection = inspect_wrapped_assbin(source_scene) inspection = inspect_wrapped_assbin(source_scene)
source_scene_sha256 = inspection.file_sha256 source_scene_sha256 = inspection.file_sha256
@@ -139,10 +174,14 @@ def prepare_runtime_view(
source = source_game.expanduser().resolve(strict=True) source = source_game.expanduser().resolve(strict=True)
_validate_game_tree(source) _validate_game_tree(source)
_validate_runtime_target(paths) _validate_runtime_target(paths)
game_save = _prepare_persistent_save(paths, source)
reused = _reuse_runtime_view(paths, source, game_save)
if reused is not None:
return reused
input_identities = _runtime_input_identities(paths, source)
source_scene = source / WHISPERING_VALE_RELATIVE source_scene = source / WHISPERING_VALE_RELATIVE
inspection = inspect_wrapped_assbin(source_scene) inspection = inspect_wrapped_assbin(source_scene)
action = _scene_action(inspection) action = _scene_action(inspection)
game_save = _prepare_persistent_save(paths, source)
( (
binary_action, binary_action,
game_exe_source, game_exe_source,
@@ -152,7 +191,7 @@ def prepare_runtime_view(
paths, paths,
source, source,
) )
reused = _reuse_runtime_view( legacy_reuse = _reuse_legacy_runtime_view(
paths, paths,
source, source,
inspection, inspection,
@@ -162,9 +201,10 @@ def prepare_runtime_view(
game_exe_source, game_exe_source,
game_dll_source, game_dll_source,
source_binary_hashes, source_binary_hashes,
input_identities,
) )
if reused is not None: if legacy_reuse is not None:
return reused return legacy_reuse
build = paths.data / f".{paths.runtime_view.name}.build-{uuid.uuid4().hex}" build = paths.data / f".{paths.runtime_view.name}.build-{uuid.uuid4().hex}"
backup = paths.data / f".{paths.runtime_view.name}.old-{uuid.uuid4().hex}" backup = paths.data / f".{paths.runtime_view.name}.old-{uuid.uuid4().hex}"
@@ -213,6 +253,10 @@ def prepare_runtime_view(
validate_scene_normalization(normalization) validate_scene_normalization(normalization)
else: else:
shutil.copy2(source_scene, runtime_scene) shutil.copy2(source_scene, runtime_scene)
if _runtime_input_identities(paths, source) != input_identities:
raise OverlayError(
"game files changed while the managed runtime view was being prepared"
)
if paths.runtime_view.exists() or paths.runtime_view.is_symlink(): if paths.runtime_view.exists() or paths.runtime_view.is_symlink():
paths.runtime_view.replace(backup) paths.runtime_view.replace(backup)
@@ -234,6 +278,11 @@ def prepare_runtime_view(
binary_action=binary_action, binary_action=binary_action,
game_exe_sha256=_sha256(paths.runtime_view / "Game.exe"), game_exe_sha256=_sha256(paths.runtime_view / "Game.exe"),
game_dll_sha256=_sha256(paths.runtime_view / "game.dll"), game_dll_sha256=_sha256(paths.runtime_view / "game.dll"),
input_identities=input_identities,
runtime_scene_identity=_runtime_file_identity(
"runtime-scene",
paths.runtime_view / WHISPERING_VALE_RELATIVE,
),
created_at=datetime.now().astimezone().isoformat(), created_at=datetime.now().astimezone().isoformat(),
links_created=links_created, links_created=links_created,
) )
@@ -470,7 +519,99 @@ def _reject_save_symlinks(root: Path) -> None:
raise OverlayError(f"persistent game save contains a symbolic link: {path}") raise OverlayError(f"persistent game save contains a symbolic link: {path}")
def _reuse_runtime_view( def _runtime_file_identity(
name: str,
path: Path,
*,
follow_symlinks: bool = True,
) -> RuntimeFileIdentity:
state = path.stat(follow_symlinks=follow_symlinks)
return RuntimeFileIdentity(
name=name,
device=state.st_dev,
inode=state.st_ino,
mode=state.st_mode,
size=state.st_size,
mtime_ns=state.st_mtime_ns,
ctime_ns=state.st_ctime_ns,
)
def _runtime_input_identities(
paths: AppPaths,
source: Path,
) -> tuple[RuntimeFileIdentity, ...]:
"""Snapshot inputs whose hashes established the cached runtime profile.
Nanosecond mtime plus ctime/inode/device prevents ordinary edits, replacements,
or timestamp restoration from bypassing the next authoritative hash check.
"""
identities = [
_runtime_file_identity("source-game-exe", source / "Game.exe"),
_runtime_file_identity("source-game-dll", source / "game.dll"),
_runtime_file_identity("source-scene", source / WHISPERING_VALE_RELATIVE),
]
cache = paths.compatibility_binaries
if cache.exists() or cache.is_symlink():
identities.append(
_runtime_file_identity(
"compatibility-cache",
cache,
follow_symlinks=False,
)
)
if cache.is_dir():
for name in COMPATIBILITY_BINARY_HASHES:
path = cache / name
if path.exists() or path.is_symlink():
identities.append(
_runtime_file_identity(
f"compatibility-cache-{name.casefold()}",
path,
)
)
return tuple(identities)
def _matching_cached_runtime_result(
paths: AppPaths,
source: Path,
) -> 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
try:
current_identities = _runtime_input_identities(paths, source)
except OSError:
return None
if current_identities != result.input_identities:
return None
return result
def _cached_runtime_compatibility(
paths: AppPaths,
source: Path,
) -> RuntimeCompatibility | None:
result = _matching_cached_runtime_result(paths, source)
if result is None:
return None
return RuntimeCompatibility(
compatible=True,
source_game=str(source),
source_scene_sha256=result.source_scene_sha256,
scene_action=result.scene_action,
binary_action=result.binary_action,
reason=None,
)
def _reuse_legacy_runtime_view(
paths: AppPaths, paths: AppPaths,
source: Path, source: Path,
inspection: SceneInspection, inspection: SceneInspection,
@@ -480,22 +621,82 @@ def _reuse_runtime_view(
game_exe_source: Path, game_exe_source: Path,
game_dll_source: Path, game_dll_source: Path,
source_binary_hashes: Mapping[str, str], source_binary_hashes: Mapping[str, str],
input_identities: tuple[RuntimeFileIdentity, ...],
) -> OverlayResult | None: ) -> OverlayResult | None:
result = _read_overlay_result(paths.runtime_manifest) """Upgrade an unchanged v1 view after its original hash checks pass once."""
legacy = _read_legacy_overlay_result(paths.runtime_manifest)
if legacy is None:
return None
if legacy.source_game != str(source):
return None
if legacy.runtime_view != str(paths.runtime_view):
return None
if legacy.source_scene_sha256 != inspection.file_sha256:
return None
if legacy.scene_action != scene_action or legacy.binary_action != binary_action:
return None
if legacy.game_exe_sha256 != source_binary_hashes.get("Game.exe"):
return None
if legacy.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,
legacy.links_created,
):
return None
runtime_scene = paths.runtime_view / WHISPERING_VALE_RELATIVE
try:
if _sha256(runtime_scene) != legacy.runtime_scene_sha256:
return None
if _runtime_input_identities(paths, source) != input_identities:
raise OverlayError(
"game files changed while the legacy runtime view was being verified"
)
runtime_scene_identity = _runtime_file_identity(
"runtime-scene",
runtime_scene,
)
except OSError:
return None
result = OverlayResult(
source_game=legacy.source_game,
runtime_view=legacy.runtime_view,
source_scene_sha256=legacy.source_scene_sha256,
runtime_scene_sha256=legacy.runtime_scene_sha256,
scene_action=legacy.scene_action,
binary_action=legacy.binary_action,
game_exe_sha256=legacy.game_exe_sha256,
game_dll_sha256=legacy.game_dll_sha256,
input_identities=input_identities,
runtime_scene_identity=runtime_scene_identity,
created_at=legacy.created_at,
links_created=legacy.links_created,
)
_write_json(
paths.runtime_manifest,
{
"configuration_version": RUNTIME_VIEW_CONFIGURATION_VERSION,
**asdict(result),
},
)
return result
def _reuse_runtime_view(
paths: AppPaths,
source: Path,
game_save: Path,
) -> OverlayResult | None:
result = _matching_cached_runtime_result(paths, source)
if result is None: if result is None:
return None return None
if result.source_game != str(source): game_exe_source = source / "Game.exe"
return None game_dll_source = source / "game.dll"
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( if not _runtime_layout_ready(
paths, paths,
source, source,
@@ -507,7 +708,12 @@ def _reuse_runtime_view(
return None return None
runtime_scene = paths.runtime_view / WHISPERING_VALE_RELATIVE runtime_scene = paths.runtime_view / WHISPERING_VALE_RELATIVE
try: try:
if _sha256(runtime_scene) != result.runtime_scene_sha256: if (
_runtime_file_identity("runtime-scene", runtime_scene)
!= result.runtime_scene_identity
):
return None
if _runtime_input_identities(paths, source) != result.input_identities:
return None return None
except OSError: except OSError:
return None return None
@@ -651,7 +857,101 @@ def _read_overlay_result(path: Path) -> OverlayResult | None:
return None return None
if links_created < 0: if links_created < 0:
return None return None
input_values = value.get("input_identities")
if not isinstance(input_values, list) or not input_values:
return None
input_identities: list[RuntimeFileIdentity] = []
for input_value in input_values:
identity = _parse_runtime_file_identity(input_value)
if identity is None:
return None
input_identities.append(identity)
input_names = {identity.name for identity in input_identities}
if len(input_names) != len(input_identities) or not {
"source-game-exe",
"source-game-dll",
"source-scene",
}.issubset(input_names):
return None
runtime_scene_identity = _parse_runtime_file_identity(
value.get("runtime_scene_identity")
)
if (
runtime_scene_identity is None
or runtime_scene_identity.name != "runtime-scene"
):
return None
return OverlayResult( 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"],
input_identities=tuple(input_identities),
runtime_scene_identity=runtime_scene_identity,
created_at=value["created_at"],
links_created=links_created,
)
def _parse_runtime_file_identity(value: object) -> RuntimeFileIdentity | None:
if not isinstance(value, dict):
return None
name = value.get("name")
if not isinstance(name, str) or not name:
return None
fields: dict[str, int] = {}
for field in ("device", "inode", "mode", "size", "mtime_ns", "ctime_ns"):
field_value = value.get(field)
if (
isinstance(field_value, bool)
or not isinstance(field_value, int)
or field_value < 0
):
return None
fields[field] = field_value
return RuntimeFileIdentity(
name=name,
device=fields["device"],
inode=fields["inode"],
mode=fields["mode"],
size=fields["size"],
mtime_ns=fields["mtime_ns"],
ctime_ns=fields["ctime_ns"],
)
def _read_legacy_overlay_result(path: Path) -> LegacyOverlayResult | None:
try:
value = json.loads(path.read_text(encoding="utf-8"))
except (OSError, UnicodeError, json.JSONDecodeError):
return None
if not isinstance(value, dict) or value.get("configuration_version") != 1:
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)
or links_created < 0
):
return None
return LegacyOverlayResult(
source_game=value["source_game"], source_game=value["source_game"],
runtime_view=value["runtime_view"], runtime_view=value["runtime_view"],
source_scene_sha256=value["source_scene_sha256"], source_scene_sha256=value["source_scene_sha256"],

View File

@@ -9,6 +9,7 @@ import json
import os import os
from pathlib import Path, PurePosixPath from pathlib import Path, PurePosixPath
import platform import platform
import re
import shlex import shlex
import shutil import shutil
import sys import sys
@@ -36,6 +37,12 @@ CURRENT_UMU_DISTRIBUTIONS = frozenset(
{"arch", "cachyos", "endeavouros", "garuda", "manjaro"} {"arch", "cachyos", "endeavouros", "garuda", "manjaro"}
) )
GAME_RUNTIME_CONFIGURATION_VERSION = 1 GAME_RUNTIME_CONFIGURATION_VERSION = 1
PROTON7_RELATIVE_PATH = Path("steamapps/common/Proton 7.0/proton")
STEAM_LIBRARY_FOLDERS_RELATIVE_PATH = Path("steamapps/libraryfolders.vdf")
STEAM_LIBRARY_FOLDERS_MAX_BYTES = 4 * 1024 * 1024
_VDF_QUOTED_PAIR = re.compile(
r'"((?:\\.|[^"\\])*)"\s*"((?:\\.|[^"\\])*)"'
)
class RuntimeErrorPT(RuntimeError): class RuntimeErrorPT(RuntimeError):
@@ -80,7 +87,7 @@ def select_runtime(
proton = Path(settings.proton_path).expanduser().resolve() proton = Path(settings.proton_path).expanduser().resolve()
if not proton.is_file(): if not proton.is_file():
raise RuntimeErrorPT(f"configured Proton script does not exist: {proton}") raise RuntimeErrorPT(f"configured Proton script does not exist: {proton}")
steam_root = _steam_root_for_proton(proton) steam_root = _steam_root_for_proton(proton, env)
if settings.runtime == "proton7-container": if settings.runtime == "proton7-container":
if not umu: if not umu:
raise RuntimeErrorPT( raise RuntimeErrorPT(
@@ -137,33 +144,108 @@ def select_runtime(
def find_steam_proton7( def find_steam_proton7(
environment: Mapping[str, str] | None = None, environment: Mapping[str, str] | None = None,
) -> tuple[Path, Path] | None: ) -> tuple[Path, Path] | None:
"""Find Proton 7 in a Steam client root or one of its registered libraries.
The returned Steam root is always the client installation, even when the
Proton tool itself lives in a secondary library. Proton 7 uses that client
root for Steam's bundled media compatibility libraries.
"""
env = os.environ if environment is None else environment env = os.environ if environment is None else environment
home_value = env.get("HOME") for steam_root in _steam_client_roots(env):
if not home_value: for library_root in _steam_library_roots(steam_root):
proton = library_root / PROTON7_RELATIVE_PATH
if proton.is_file():
return proton.resolve(), steam_root
return None return None
home = Path(home_value).expanduser()
roots: list[Path] = []
explicit = env.get("STEAM_COMPAT_CLIENT_INSTALL_PATH") def _steam_client_roots(environment: Mapping[str, str]) -> tuple[Path, ...]:
candidates: list[Path] = []
explicit = environment.get("STEAM_COMPAT_CLIENT_INSTALL_PATH")
if explicit: if explicit:
roots.append(Path(explicit).expanduser()) candidates.append(Path(explicit).expanduser())
roots.extend( home_value = environment.get("HOME")
if home_value:
home = Path(home_value).expanduser()
candidates.extend(
( (
home / ".local/share/Steam", home / ".local/share/Steam",
home / ".steam/steam", home / ".steam/steam",
home / ".steam/root",
home / ".steam/debian-installation",
home / "snap/steam/common/.local/share/Steam", home / "snap/steam/common/.local/share/Steam",
home / "snap/steam/common/.steam/steam",
home / ".var/app/com.valvesoftware.Steam/.local/share/Steam", home / ".var/app/com.valvesoftware.Steam/.local/share/Steam",
) )
) )
return _unique_resolved_paths(candidates)
def _steam_library_roots(steam_root: Path) -> tuple[Path, ...]:
registered = _read_steam_library_folders(
steam_root / STEAM_LIBRARY_FOLDERS_RELATIVE_PATH
)
return _unique_resolved_paths((steam_root, *registered))
def _read_steam_library_folders(path: Path) -> tuple[Path, ...]:
"""Read current and legacy Steam libraryfolders.vdf path records."""
try:
with path.open("rb") as handle:
content = handle.read(STEAM_LIBRARY_FOLDERS_MAX_BYTES + 1)
except OSError:
return ()
if len(content) > STEAM_LIBRARY_FOLDERS_MAX_BYTES:
return ()
text = content.decode("utf-8-sig", errors="replace")
result: list[Path] = []
for raw_key, raw_value in _VDF_QUOTED_PAIR.findall(text):
key = _decode_vdf_string(raw_key)
value = _decode_vdf_string(raw_value)
# Current files use a named "path" property. Older Steam clients used
# numeric top-level keys, which are distinguishable from app-size pairs
# because a Linux library path must be absolute.
if key.casefold() != "path" and not key.isdecimal():
continue
if not value or "\x00" in value:
continue
library = Path(value).expanduser()
if library.is_absolute():
result.append(library)
return tuple(result)
def _decode_vdf_string(value: str) -> str:
result: list[str] = []
index = 0
while index < len(value):
character = value[index]
if character == "\\" and index + 1 < len(value):
next_character = value[index + 1]
if next_character in {'"', "\\"}:
result.append(next_character)
index += 2
continue
result.append(character)
index += 1
return "".join(result)
def _unique_resolved_paths(paths: Sequence[Path]) -> tuple[Path, ...]:
result: list[Path] = []
seen: set[Path] = set() seen: set[Path] = set()
for root in roots: for path in paths:
resolved = root.resolve() try:
resolved = path.resolve()
except (OSError, RuntimeError):
continue
if resolved in seen: if resolved in seen:
continue continue
seen.add(resolved) seen.add(resolved)
proton = resolved / "steamapps/common/Proton 7.0/proton" result.append(resolved)
if proton.is_file(): return tuple(result)
return proton, resolved
return None
def prepare_game_runtime( def prepare_game_runtime(
@@ -298,6 +380,7 @@ def run_game(
*, *,
start_option: DepotStartOption = DEFAULT_DEPOT_START_OPTION, start_option: DepotStartOption = DEFAULT_DEPOT_START_OPTION,
use_gamebox: bool = True, use_gamebox: bool = True,
concurrent_prefix_launch: bool = False,
timeout_seconds: float = 12 * 60 * 60, timeout_seconds: float = 12 * 60 * 60,
check: bool = True, check: bool = True,
extra_environment: Mapping[str, str] | None = None, extra_environment: Mapping[str, str] | None = None,
@@ -346,6 +429,13 @@ def run_game(
selection, selection,
synchronization=resolve_proton_sync_mode(settings.proton_sync), synchronization=resolve_proton_sync_mode(settings.proton_sync),
) )
if selection.kind in {"umu", "proton7-container"}:
# Keep every UMU launch identical except for the verb, as required when
# several games share one prefix. The first client owns prefix startup;
# every later shared-lease client joins the active wineserver directly.
environment["PROTON_VERB"] = (
"runinprefix" if concurrent_prefix_launch else "waitforexitandrun"
)
if use_gamebox: if use_gamebox:
game_command = build_gamebox_command(game_command, settings) game_command = build_gamebox_command(game_command, settings)
if extra_environment: if extra_environment:
@@ -653,15 +743,27 @@ def _ensure_proton_game_link(paths: AppPaths) -> None:
link.symlink_to(paths.runtime_view, target_is_directory=True) link.symlink_to(paths.runtime_view, target_is_directory=True)
def _steam_root_for_proton(proton: Path) -> Path: def _steam_root_for_proton(
parts = proton.parts proton: Path,
try: environment: Mapping[str, str] | None = None,
index = parts.index("steamapps") ) -> Path:
except ValueError as exc: steamapps = next(
(parent for parent in proton.parents if parent.name.casefold() == "steamapps"),
None,
)
if steamapps is None:
raise RuntimeErrorPT( raise RuntimeErrorPT(
"configured Proton path must be inside a Steam steamapps directory" "configured Proton path must be inside a Steam steamapps directory"
) from exc )
return Path(*parts[:index]) library_root = steamapps.parent.resolve()
env = os.environ if environment is None else environment
for steam_root in _steam_client_roots(env):
if library_root in _steam_library_roots(steam_root):
return steam_root
# Preserve support for an otherwise valid, manually managed Steam layout.
# Without a registered client root, the library root is the best available
# compatibility value and matches the launcher's historical behaviour.
return library_root
def native_d3d_ready(prefix: Path) -> bool: def native_d3d_ready(prefix: Path) -> bool:

157
ptlinux/server_status.py Normal file
View File

@@ -0,0 +1,157 @@
"""Official realm health used by the native launcher dashboard."""
from __future__ import annotations
from dataclasses import dataclass
import json
from typing import Callable, Mapping, Protocol
from urllib.request import Request, urlopen
from . import __version__
# This is the same publisher endpoint embedded in Windows launcher 1.0.0.64.
SERVER_STATUS_ENDPOINT = (
"https://pristontale.eu/api/pristontale/developer"
"?api_key=d9OlDSvworitU4ajDiH6dU6cxXApc4M0"
)
MAX_SERVER_STATUS_BYTES = 1024 * 1024
class ServerStatusError(RuntimeError):
"""Raised when the official status response cannot be interpreted safely."""
class ServerStatusResponse(Protocol):
def read(self, size: int = -1) -> bytes: ...
def __enter__(self) -> ServerStatusResponse: ...
def __exit__(
self,
exception_type: type[BaseException] | None,
exception: BaseException | None,
traceback: object | None,
) -> bool | None: ...
ServerStatusOpener = Callable[..., ServerStatusResponse]
@dataclass(frozen=True)
class RealmServerStatus:
name: str
online: bool | None
players_online: int | None
login_online: bool | None = None
game_online: bool | None = None
def fetch_server_status(
*,
timeout_seconds: float = 10.0,
opener: ServerStatusOpener = urlopen,
) -> tuple[RealmServerStatus, ...]:
request = Request(
SERVER_STATUS_ENDPOINT,
headers={"User-Agent": f"PT-Linux/{__version__} (+https://pristontale.eu/)"},
method="GET",
)
try:
response = opener(request, timeout=timeout_seconds)
with response:
payload = response.read(MAX_SERVER_STATUS_BYTES + 1)
except OSError as exc:
raise ServerStatusError("official server status is unavailable") from exc
if len(payload) > MAX_SERVER_STATUS_BYTES:
raise ServerStatusError("official server status exceeds the 1 MiB limit")
return parse_server_status(payload)
def parse_server_status(payload: bytes) -> tuple[RealmServerStatus, ...]:
try:
value = json.loads(payload.decode("utf-8-sig"))
except (UnicodeError, json.JSONDecodeError) as exc:
raise ServerStatusError("official server status is not valid JSON") from exc
# Historical versions returned an array containing the status object. The
# current endpoint returns that object directly.
if isinstance(value, list) and value:
value = value[0]
if not isinstance(value, dict):
raise ServerStatusError("official server status is not a realm object")
statuses: list[RealmServerStatus] = []
for raw_name, raw_components in value.items():
if not isinstance(raw_name, str) or raw_name.casefold() == "website":
continue
name = raw_name.strip()
if not name or len(name) > 64 or not isinstance(raw_components, Mapping):
continue
status = _parse_realm(name, raw_components)
if status is not None:
statuses.append(status)
if not statuses:
raise ServerStatusError("official server status contains no realms")
return tuple(statuses)
def _parse_realm(
name: str,
components: Mapping[object, object],
) -> RealmServerStatus | None:
service_states: list[bool] = []
login_online: bool | None = None
game_online: bool | None = None
players_online: int | None = None
for raw_key, raw_value in components.items():
if not isinstance(raw_key, str):
continue
key = raw_key.casefold()
if key == "users.online":
players_online = _non_negative_integer(raw_value)
continue
if not key.startswith("server."):
continue
state = _online_state(raw_value)
if state is None:
continue
service_states.append(state)
if "login" in key:
login_online = state
elif "game" in key:
game_online = state
if not service_states and players_online is None:
return None
online = all(service_states) if service_states else None
display_name = name[0].upper() + name[1:]
return RealmServerStatus(
name=display_name,
online=online,
players_online=players_online,
login_online=login_online,
game_online=game_online,
)
def _online_state(value: object) -> bool | None:
if isinstance(value, bool):
return value
if isinstance(value, (int, float)):
return value > 0
if isinstance(value, str):
text = value.strip().casefold()
if text in {"online", "up", "true", "1"}:
return True
if text in {"offline", "down", "false", "0"}:
return False
return None
def _non_negative_integer(value: object) -> int | None:
if isinstance(value, bool):
return None
if isinstance(value, int) and value >= 0:
return value
if isinstance(value, str) and value.isascii() and value.isdecimal():
return int(value)
return None

View File

@@ -15,6 +15,7 @@ from .paths import AppPaths
SETTINGS_SCHEMA = 1 SETTINGS_SCHEMA = 1
_TITLE_FRAME_DEFAULT_MIGRATION_KEY = "title_frame_default_migrated"
@dataclass(frozen=True) @dataclass(frozen=True)
@@ -26,7 +27,7 @@ class Settings:
output_height: int = 900 output_height: int = 900
refresh_rate: int = 60 refresh_rate: int = 60
fullscreen: bool = False fullscreen: bool = False
borderless: bool = True borderless: bool = False
suppress_unresponsive_warnings: bool = True suppress_unresponsive_warnings: bool = True
scaler: str = "fit" scaler: str = "fit"
filter: str = "linear" filter: str = "linear"
@@ -141,7 +142,7 @@ def _settings_from_object(value: dict[str, object]) -> Settings:
output_height=_integer(value, "output_height", 900), output_height=_integer(value, "output_height", 900),
refresh_rate=_integer(value, "refresh_rate", 60), refresh_rate=_integer(value, "refresh_rate", 60),
fullscreen=_boolean(value, "fullscreen", False), fullscreen=_boolean(value, "fullscreen", False),
borderless=_boolean(value, "borderless", True), borderless=_migrated_borderless(value),
suppress_unresponsive_warnings=_boolean( suppress_unresponsive_warnings=_boolean(
value, value,
"suppress_unresponsive_warnings", "suppress_unresponsive_warnings",
@@ -158,6 +159,22 @@ def _settings_from_object(value: dict[str, object]) -> Settings:
return settings return settings
def _migrated_borderless(value: dict[str, object]) -> bool:
"""Retire the old implicit borderless default exactly once.
Older releases wrote ``borderless=true`` for every default installation, so
they cannot distinguish an inherited value from an explicit choice. The
migration marker keeps schema 1 backward compatible while letting users opt
back into Borderless after the decorated default has been applied.
"""
borderless = _boolean(value, "borderless", False)
migrated = _boolean(value, _TITLE_FRAME_DEFAULT_MIGRATION_KEY, False)
if not migrated:
return False
return borderless
def _decode_settings(content: bytes) -> str: def _decode_settings(content: bytes) -> str:
"""Accept native UTF-8 plus settings saved by common Windows editors.""" """Accept native UTF-8 plus settings saved by common Windows editors."""
@@ -174,6 +191,7 @@ def _settings_need_migration(value: dict[str, object]) -> bool:
return ( return (
value.get("schema") != SETTINGS_SCHEMA value.get("schema") != SETTINGS_SCHEMA
or not _SETTINGS_FIELD_NAMES.issubset(value) or not _SETTINGS_FIELD_NAMES.issubset(value)
or value.get(_TITLE_FRAME_DEFAULT_MIGRATION_KEY) is not True
) )
@@ -185,6 +203,7 @@ def _write_settings(
payload = dict(existing or {}) payload = dict(existing or {})
payload.update(asdict(settings)) payload.update(asdict(settings))
payload["schema"] = SETTINGS_SCHEMA payload["schema"] = SETTINGS_SCHEMA
payload[_TITLE_FRAME_DEFAULT_MIGRATION_KEY] = True
temporary = paths.settings_file.with_name( temporary = paths.settings_file.with_name(
f".{paths.settings_file.name}.{os.getpid()}.{threading.get_ident()}." f".{paths.settings_file.name}.{os.getpid()}.{threading.get_ident()}."
f"{secrets.token_hex(8)}.tmp" f"{secrets.token_hex(8)}.tmp"

705
ptlinux/tk_chrome.py Normal file
View File

@@ -0,0 +1,705 @@
from __future__ import annotations
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Callable, Literal
from .desktop_identity import APPLICATION_WM_CLASS, APPLICATION_WM_INSTANCE
DialogKind = Literal["info", "success", "warning", "error", "question"]
@dataclass(frozen=True)
class DialogPalette:
"""Colors and glyph used by one launcher dialog intent."""
accent: str
accent_soft: str
glyph: str
_DIALOG_PALETTES: dict[str, DialogPalette] = {
"info": DialogPalette("#d98927", "#3a2819", "i"),
"success": DialogPalette("#55b878", "#173122", "✓"),
"warning": DialogPalette("#e0a248", "#3b2a18", "!"),
"error": DialogPalette("#cf604c", "#3a1d19", "×"),
"question": DialogPalette("#d98927", "#3a2819", "?"),
}
def dialog_palette(kind: DialogKind) -> DialogPalette:
"""Return the stable visual palette for a themed launcher dialog."""
try:
return _DIALOG_PALETTES[kind]
except KeyError as error:
raise ValueError(f"unsupported launcher dialog kind: {kind}") from error
def square_icon_layout(width: int, height: int) -> tuple[int, int, int]:
"""Return square canvas size and centered offsets for a window icon."""
if width <= 0 or height <= 0:
raise ValueError("icon dimensions must be positive")
size = max(width, height)
return size, (size - width) // 2, (size - height) // 2
def icon_subsample_factor(width: int, height: int, max_dimension: int) -> int:
"""Return the integer Tk subsample needed to fit an image within a square."""
if width <= 0 or height <= 0:
raise ValueError("icon dimensions must be positive")
if max_dimension <= 0:
raise ValueError("maximum icon dimension must be positive")
largest_dimension = max(width, height)
return max(1, (largest_dimension + max_dimension - 1) // max_dimension)
def load_square_window_icon(master: Any, path: Path) -> Any:
"""Load a PNG into a square transparent Tk image for desktop shells."""
import tkinter as tk
source = tk.PhotoImage(master=master, file=str(path))
size, offset_x, offset_y = square_icon_layout(source.width(), source.height())
if source.width() == source.height():
return source
icon = tk.PhotoImage(master=master, width=size, height=size)
icon.tk.call(
str(icon),
"copy",
str(source),
"-to",
offset_x,
offset_y,
)
return icon
def centered_window_geometry(
width: int,
height: int,
*,
parent_x: int,
parent_y: int,
parent_width: int,
parent_height: int,
) -> str:
"""Return Tk geometry centered over a parent window."""
if width <= 0 or height <= 0:
raise ValueError("window dimensions must be positive")
x = parent_x + (max(1, parent_width) - width) // 2
y = parent_y + (max(1, parent_height) - height) // 2
return f"{width}x{height}{_geometry_offset(x)}{_geometry_offset(y)}"
def dragged_window_geometry(
pointer_x: int,
pointer_y: int,
*,
offset_x: int,
offset_y: int,
) -> str:
"""Return the Tk position for a pointer-driven custom title bar."""
return (
f"{_geometry_offset(pointer_x - offset_x)}"
f"{_geometry_offset(pointer_y - offset_y)}"
)
def _geometry_offset(value: int) -> str:
return f"+{value}" if value >= 0 else str(value)
def _bind_window_drag(handle: Any, *widgets: Any) -> None:
position = [0, 0]
def begin(event: Any) -> None:
position[0] = int(event.x_root) - int(handle.winfo_x())
position[1] = int(event.y_root) - int(handle.winfo_y())
def move(event: Any) -> None:
handle.geometry(
dragged_window_geometry(
int(event.x_root),
int(event.y_root),
offset_x=position[0],
offset_y=position[1],
)
)
for widget in widgets:
widget.bind("<ButtonPress-1>", begin, add="+")
widget.bind("<B1-Motion>", move, add="+")
class DashboardChrome:
"""Controller for one frameless launcher window and its controls."""
def __init__(
self,
root: Any,
content: Any,
close_button: Any,
minimize_button: Any,
*,
managed: bool,
identity_configured: bool,
allow_close: bool,
) -> None:
self.root = root
self.content = content
self.managed = managed
self.identity_configured = identity_configured
self._close_button = close_button
self._minimize_button = minimize_button
self._allow_close = allow_close
self._close_command: Callable[[], None] = root.destroy
if close_button is not None:
close_button.configure(command=self.close)
if minimize_button is not None:
minimize_button.configure(command=self.minimize)
root.bind("<Alt-F4>", self._close_from_event, add="+")
def set_close_command(self, command: Callable[[], None]) -> None:
self._close_command = command
def close(self) -> None:
if self._allow_close:
self._close_command()
def minimize(self) -> None:
"""Ask the window manager to iconify the managed dashboard."""
import tkinter as tk
try:
self.root.iconify()
except tk.TclError:
pass
def _close_from_event(self, _event: Any) -> str:
self.close()
return "break"
def _configure_x11_window(
root: Any,
*,
resource_name: str,
resource_class: str,
) -> tuple[bool, bool]:
"""Hide the X11 frame and give desktop shells a stable window identity."""
import ctypes.util
library_path = ctypes.util.find_library("X11")
if library_path is None:
return False, False
try:
x11 = ctypes.CDLL(library_path)
except OSError:
return False, False
class XClassHint(ctypes.Structure):
_fields_ = [
("res_name", ctypes.c_char_p),
("res_class", ctypes.c_char_p),
]
x11.XOpenDisplay.argtypes = [ctypes.c_char_p]
x11.XOpenDisplay.restype = ctypes.c_void_p
x11.XQueryTree.argtypes = [
ctypes.c_void_p,
ctypes.c_ulong,
ctypes.POINTER(ctypes.c_ulong),
ctypes.POINTER(ctypes.c_ulong),
ctypes.POINTER(ctypes.POINTER(ctypes.c_ulong)),
ctypes.POINTER(ctypes.c_uint),
]
x11.XQueryTree.restype = ctypes.c_int
x11.XInternAtom.argtypes = [ctypes.c_void_p, ctypes.c_char_p, ctypes.c_int]
x11.XInternAtom.restype = ctypes.c_ulong
x11.XChangeProperty.argtypes = [
ctypes.c_void_p,
ctypes.c_ulong,
ctypes.c_ulong,
ctypes.c_ulong,
ctypes.c_int,
ctypes.c_int,
ctypes.POINTER(ctypes.c_ubyte),
ctypes.c_int,
]
x11.XChangeProperty.restype = ctypes.c_int
x11.XSetClassHint.argtypes = [
ctypes.c_void_p,
ctypes.c_ulong,
ctypes.POINTER(XClassHint),
]
x11.XSetClassHint.restype = ctypes.c_int
x11.XFlush.argtypes = [ctypes.c_void_p]
x11.XFlush.restype = ctypes.c_int
x11.XFree.argtypes = [ctypes.c_void_p]
x11.XFree.restype = ctypes.c_int
x11.XCloseDisplay.argtypes = [ctypes.c_void_p]
x11.XCloseDisplay.restype = ctypes.c_int
root.update_idletasks()
display = x11.XOpenDisplay(None)
if display is None:
return False, False
children = ctypes.POINTER(ctypes.c_ulong)()
try:
root_window = ctypes.c_ulong()
wrapper_window = ctypes.c_ulong()
child_count = ctypes.c_uint()
queried = x11.XQueryTree(
display,
root.winfo_id(),
ctypes.byref(root_window),
ctypes.byref(wrapper_window),
ctypes.byref(children),
ctypes.byref(child_count),
)
if not queried or not wrapper_window.value:
return False, False
atom = x11.XInternAtom(display, b"_MOTIF_WM_HINTS", 0)
if not atom:
return False, False
motif_hints_decorations = 1 << 1
hints = (ctypes.c_ulong * 5)(motif_hints_decorations, 0, 0, 0, 0)
changed = x11.XChangeProperty(
display,
wrapper_window.value,
atom,
atom,
32,
0,
ctypes.cast(hints, ctypes.POINTER(ctypes.c_ubyte)),
len(hints),
)
identity = XClassHint(
res_name=resource_name.encode("utf-8"),
res_class=resource_class.encode("utf-8"),
)
identity_changed = x11.XSetClassHint(
display,
wrapper_window.value,
ctypes.byref(identity),
)
x11.XFlush(display)
return bool(changed), bool(identity_changed)
finally:
if children:
x11.XFree(children)
x11.XCloseDisplay(display)
def create_window_chrome(
root: Any,
*,
title: str,
subtitle: str,
window_icon: Any | None = None,
show_title_bar: bool = True,
show_close: bool = True,
show_minimize: bool = False,
allow_close: bool = True,
) -> DashboardChrome:
"""Install polished managed chrome around one launcher-owned Tk window."""
import tkinter as tk
managed, identity_configured = _configure_x11_window(
root,
resource_name=APPLICATION_WM_INSTANCE,
resource_class=APPLICATION_WM_CLASS,
)
if not managed:
root.overrideredirect(True)
border = tk.Frame(
root,
background="#6a482d",
borderwidth=0,
highlightthickness=0,
)
border.pack(fill="both", expand=True)
surface = tk.Frame(
border,
background="#0c0908",
borderwidth=0,
highlightthickness=0,
)
surface.pack(fill="both", expand=True, padx=1, pady=1)
title_bar = tk.Frame(
surface,
background="#17100d",
height=32,
borderwidth=0,
highlightthickness=0,
cursor="fleur",
)
if show_title_bar:
title_bar.pack(fill="x")
title_bar.pack_propagate(False)
tk_accent = tk.Frame(title_bar, background="#d98927", width=3)
tk_accent.pack(side="left", fill="y")
title_icon: Any | None = None
if window_icon is not None:
try:
factor = icon_subsample_factor(
window_icon.width(),
window_icon.height(),
20,
)
title_icon = window_icon.subsample(factor, factor)
except (tk.TclError, ValueError):
title_icon = None
if title_icon is not None:
mark = tk.Label(
title_bar,
image=title_icon,
width=22,
height=22,
background="#17100d",
borderwidth=0,
highlightthickness=0,
cursor="fleur",
)
mark.image = title_icon
else:
mark = tk.Frame(
title_bar,
width=22,
height=22,
background="#17100d",
borderwidth=0,
highlightthickness=0,
cursor="fleur",
)
mark.pack(side="left", padx=(9, 7))
identity = tk.Frame(title_bar, background="#17100d", cursor="fleur")
identity.pack(side="left", fill="y")
title_label = tk.Label(
identity,
text=title,
background="#17100d",
foreground="#f2dfbd",
font=("Sans", 8, "bold"),
cursor="fleur",
)
title_label.pack(side="left", pady=(0, 1))
subtitle_label = tk.Label(
identity,
text=f" · {subtitle}",
background="#17100d",
foreground="#8f7b69",
font=("Sans", 7),
cursor="fleur",
)
subtitle_label.pack(side="left", pady=(0, 1))
close_button: Any = None
if show_close:
close_button = tk.Button(
title_bar,
text="×",
background="#17100d",
foreground="#c9b7a4",
activebackground="#a63f2f",
activeforeground="#ffffff",
font=("Sans", 12),
width=4,
borderwidth=0,
highlightthickness=0,
relief="flat",
cursor="hand2",
takefocus=False,
)
close_button.pack(side="right", fill="y")
minimize_button: Any = None
if show_minimize:
minimize_button = tk.Button(
title_bar,
text="—",
background="#17100d",
foreground="#c9b7a4",
activebackground="#312219",
activeforeground="#ffffff",
font=("Sans", 10, "bold"),
width=4,
borderwidth=0,
highlightthickness=0,
relief="flat",
cursor="hand2",
takefocus=False,
)
minimize_button.pack(side="right", fill="y")
content = tk.Frame(
surface,
background="#0c0908",
borderwidth=0,
highlightthickness=0,
)
content.pack(fill="both", expand=True)
if show_title_bar:
_bind_window_drag(root, title_bar, mark, identity, title_label, subtitle_label)
return DashboardChrome(
root,
content,
close_button,
minimize_button,
managed=managed,
identity_configured=identity_configured,
allow_close=allow_close,
)
def create_dashboard_chrome(
root: Any,
*,
title: str,
subtitle: str,
window_icon: Any | None = None,
) -> DashboardChrome:
"""Install the draggable frame and full controls used by the dashboard."""
return create_window_chrome(
root,
title=title,
subtitle=subtitle,
window_icon=window_icon,
show_close=True,
show_minimize=True,
allow_close=True,
)
def show_themed_dialog(
parent: Any,
*,
title: str,
message: str,
kind: DialogKind = "info",
confirm: bool = False,
accept_text: str = "OK",
cancel_text: str = "CANCEL",
window_icon: Any | None = None,
) -> bool:
"""Show one frameless modal dialog matching the native launcher palette."""
import tkinter as tk
palette = dialog_palette(kind)
dialog = tk.Toplevel(parent)
dialog.withdraw()
dialog.title(title)
dialog.configure(background=palette.accent)
dialog.resizable(False, False)
dialog.overrideredirect(True)
if window_icon is not None:
try:
dialog.iconphoto(False, window_icon)
except tk.TclError:
pass
try:
dialog.transient(parent)
except tk.TclError:
pass
result = False
def finish(value: bool) -> None:
nonlocal result
result = value
try:
dialog.grab_release()
except tk.TclError:
pass
dialog.destroy()
border = tk.Frame(dialog, background=palette.accent)
border.pack(fill="both", expand=True)
surface = tk.Frame(border, background="#191311")
surface.pack(fill="both", expand=True, padx=1, pady=1)
header = tk.Frame(surface, background="#24170f", height=66)
header.pack(fill="x")
header.pack_propagate(False)
tk_accent = tk.Frame(header, background=palette.accent, width=5)
tk_accent.pack(side="left", fill="y")
header_copy = tk.Frame(header, background="#24170f")
header_copy.pack(side="left", fill="both", expand=True, padx=16, pady=10)
eyebrow = tk.Label(
header_copy,
text="PRISTONTALE EU · LINUX LAUNCHER",
background="#24170f",
foreground=palette.accent,
font=("Sans", 7, "bold"),
anchor="w",
)
eyebrow.pack(fill="x")
title_label = tk.Label(
header_copy,
text=title,
background="#24170f",
foreground="#fff0d2",
font=("Georgia", 15, "bold"),
anchor="w",
)
title_label.pack(fill="x", pady=(1, 0))
close_button = tk.Button(
header,
text="×",
command=lambda: finish(False),
background="#24170f",
foreground="#c9b7a4",
activebackground="#a63f2f",
activeforeground="#ffffff",
font=("Sans", 12),
width=4,
borderwidth=0,
highlightthickness=0,
relief="flat",
cursor="hand2",
takefocus=False,
)
close_button.pack(side="right", fill="y")
body = tk.Frame(surface, background="#191311", padx=20, pady=20)
body.pack(fill="both", expand=True)
modal_icon: Any | None = None
if window_icon is not None:
try:
factor = icon_subsample_factor(
window_icon.width(),
window_icon.height(),
46,
)
modal_icon = window_icon.subsample(factor, factor)
except (tk.TclError, ValueError):
modal_icon = None
if modal_icon is not None:
icon_host = tk.Frame(
body,
width=46,
height=46,
background="#191311",
borderwidth=0,
highlightthickness=0,
)
icon_host.pack(side="left", anchor="n", padx=(0, 16))
icon_host.pack_propagate(False)
icon = tk.Label(
icon_host,
image=modal_icon,
background="#191311",
borderwidth=0,
highlightthickness=0,
)
icon.place(relx=0.5, rely=0.5, anchor="center")
else:
icon = tk.Canvas(
body,
width=46,
height=46,
background="#191311",
borderwidth=0,
highlightthickness=0,
)
icon.create_oval(
3,
3,
43,
43,
fill=palette.accent_soft,
outline=palette.accent,
width=2,
)
icon.create_text(
23,
23,
text=palette.glyph,
fill=palette.accent,
font=("Sans", 17, "bold"),
)
icon.pack(side="left", anchor="n", padx=(0, 16))
message_label = tk.Label(
body,
text=message,
background="#191311",
foreground="#ded1c4",
font=("Sans", 10),
justify="left",
anchor="w",
wraplength=420,
)
message_label.pack(side="left", fill="both", expand=True, anchor="n")
footer = tk.Frame(surface, background="#100d0c", padx=18, pady=12)
footer.pack(fill="x")
def button(
container: Any,
*,
label: str,
command: Callable[[], None],
primary: bool,
) -> Any:
background = palette.accent if primary else "#2a201c"
foreground = "#17100d" if primary else "#ead8c2"
active_background = "#efae57" if primary else "#3a2b25"
return tk.Button(
container,
text=label,
command=command,
background=background,
foreground=foreground,
activebackground=active_background,
activeforeground="#ffffff" if not primary else "#17100d",
font=("Sans", 8, "bold"),
padx=16,
pady=7,
borderwidth=0,
highlightthickness=0,
relief="flat",
cursor="hand2",
)
accept_button = button(
footer,
label=accept_text,
command=lambda: finish(True),
primary=True,
)
accept_button.pack(side="right")
if confirm:
cancel_button = button(
footer,
label=cancel_text,
command=lambda: finish(False),
primary=False,
)
cancel_button.pack(side="right", padx=(0, 8))
_bind_window_drag(dialog, header, header_copy, eyebrow, title_label)
dialog.bind("<Escape>", lambda _event: finish(False))
dialog.bind("<Return>", lambda _event: finish(True))
parent.update_idletasks()
dialog.update_idletasks()
width = 560
height = max(218, dialog.winfo_reqheight())
dialog.geometry(
centered_window_geometry(
width,
height,
parent_x=parent.winfo_rootx(),
parent_y=parent.winfo_rooty(),
parent_width=parent.winfo_width(),
parent_height=parent.winfo_height(),
)
)
dialog.deiconify()
dialog.lift()
dialog.grab_set()
accept_button.focus_set()
parent.wait_window(dialog)
return result

File diff suppressed because it is too large Load Diff

View File

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

View File

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

Binary file not shown.