"""Anonymous, best-effort launcher telemetry with bounded asynchronous delivery.""" from __future__ import annotations from collections.abc import Callable, Mapping, Sequence from dataclasses import dataclass from datetime import datetime, timezone import hashlib import json import math import os from pathlib import Path import platform import queue import re import resource import socket import sys import threading import time import traceback from types import TracebackType from typing import Any from urllib.parse import parse_qsl, urlsplit, urlunsplit from urllib.request import Request, urlopen import uuid from . import __version__ from .paths import AppPaths from .telemetry_environment import collect_extended_environment DEFAULT_TELEMETRY_ENDPOINT = "https://pristontale.eu/api/launcher/telemetry" TELEMETRY_ENDPOINT_VARIABLE = "PTLINUX_TELEMETRY_ENDPOINT" TELEMETRY_KEY_VARIABLE = "PTLINUX_TELEMETRY_KEY" TELEMETRY_DEBUG_VARIABLE = "PTLINUX_TELEMETRY_DEBUG" MAX_REQUEST_BYTES = 64 * 1024 MAX_MESSAGE_BYTES = 10 * 1024 MAX_CONTEXT_BYTES = 16 * 1024 MAX_METADATA_BYTES = 32 * 1024 MAX_STDERR_CHARACTERS = 4000 MAX_STDOUT_CHARACTERS = 2000 MAX_TRACEBACK_CHARACTERS = 10_000 MAX_CONTAINER_ITEMS = 128 MAX_VALUE_DEPTH = 8 QUEUE_CAPACITY = 128 DELIVERY_TIMEOUT_SECONDS = 3.0 SHUTDOWN_WAIT_SECONDS = 0.25 ERROR_SUPPRESSION_WINDOW_SECONDS = 60.0 MAX_ERROR_FINGERPRINTS = 128 _RESOURCE_USAGE_EVENTS = frozenset( { "dependency_fix_success", "dxvk_install_success", "dxvk_repair_success", "game_exited", "import_success", "install_success", "launcher_closed", "launcher_ready", "prefix_reset_success", "repair_success", "runtime_install_success", "runtime_repair_success", "update_success", } ) VALID_SEVERITIES = frozenset({"debug", "info", "warning", "error", "critical"}) VALID_EVENTS = frozenset( { "api_error", "asyncio_exception", "checksum_failed", "compatibility_check_completed", "compatibility_check_failed", "compatibility_check_started", "configuration_error", "dependency_fix_failed", "dependency_fix_requested", "dependency_fix_started", "dependency_fix_success", "dependency_check_failed", "download_completed", "download_failed", "download_resumed", "download_retry", "download_started", "dxvk_install_failed", "dxvk_install_started", "dxvk_install_success", "dxvk_repair_failed", "dxvk_repair_started", "dxvk_repair_success", "file_validation_failed", "filesystem_error", "first_run_completed", "first_run_failed", "first_run_started", "game_exited", "game_launch_failed", "game_launch_requested", "game_launch_started", "game_launch_success", "import_failed", "import_scan_completed", "import_started", "import_success", "install_download_completed", "install_download_started", "install_extract_completed", "install_extract_started", "install_failed", "install_started", "install_success", "launcher_closed", "launcher_exception", "launcher_ready", "launcher_started", "operation_timeout", "permission_error", "prefix_reset_failed", "prefix_reset_requested", "prefix_reset_started", "prefix_reset_success", "repair_failed", "repair_requested", "repair_scan_completed", "repair_started", "repair_success", "repeated_error", "runtime_check_failed", "runtime_install_failed", "runtime_install_started", "runtime_install_success", "runtime_reinstall_requested", "runtime_repair_failed", "runtime_repair_started", "runtime_repair_success", "subprocess_failed", "thread_exception", "update_apply_completed", "update_apply_started", "update_available", "update_check_started", "update_download_completed", "update_download_started", "update_failed", "update_not_required", "update_started", "update_success", "vulkan_32bit_check_failed", "vulkan_check_failed", "wine_32bit_check_failed", } ) _SECRET_KEYS = frozenset( { "api_key", "apikey", "authorization", "client_secret", "cookie", "launcher_key", "password", "passwd", "proxy_authorization", "refresh_token", "secret", "session_id", "sessionid", "set_cookie", "token", "access_token", } ) _SECRET_QUERY_KEYS = _SECRET_KEYS | frozenset( { "expires", "key-pair-id", "policy", "signature", "sig", "x-amz-credential", "x-amz-signature", "x-amz-security-token", } ) _NETWORK_FILESYSTEMS = frozenset( {"9p", "afs", "ceph", "cifs", "nfs", "nfs4", "smb3", "sshfs"} ) _URL_PATTERN = re.compile(r"https?://[^\s<>\"']+", re.IGNORECASE) _EMAIL_PATTERN = re.compile( r"(?|/home/|/root)[\\/]\.ssh(?:[\\/]\S*)?") _AUTHORIZATION_PATTERN = re.compile( r"(?i)(authorization\s*:\s*(?:bearer|basic)\s+)[^\s,;]+" ) _SECRET_ASSIGNMENT_PATTERN = re.compile( r"(?i)\b(password|passwd|token|access_token|refresh_token|api[_-]?key|" r"client_secret|session_id|cookie)\s*([=:]\s*)(?:\"[^\"]*\"|'[^']*'|[^&\s,;]+)" ) _UUID_PATTERN = re.compile( r"\b[0-9a-f]{8}-[0-9a-f]{4}-[1-5][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}\b", re.IGNORECASE, ) _MEMORY_ADDRESS_PATTERN = re.compile(r"\b0x[0-9a-f]{6,16}\b", re.IGNORECASE) _HTTP_STATUS_PATTERN = re.compile(r"\bHTTP\s+([1-5][0-9]{2})\b", re.IGNORECASE) _INSTALL_ID_FILE_MODE = 0o600 TelemetryTransport = Callable[[str, bytes, Mapping[str, str], float], None] @dataclass class _RepeatedErrorState: fingerprint: str event: str first_seen: float last_seen: float suppressed_count: int operation: str | None stage: str | None class TelemetryService: """Own telemetry collection, sanitization, queuing, and delivery. Every public method is best-effort. Callers must never need error handling to protect a launcher operation from telemetry failures. """ def __init__( self, paths: AppPaths, *, enabled: bool = True, endpoint: str | None = None, launcher_key: str | None = None, environment: Mapping[str, str] | None = None, transport: TelemetryTransport | None = None, session_metadata: Mapping[str, object] | None = None, monotonic: Callable[[], float] = time.monotonic, ) -> None: self.paths = paths self._environment = os.environ if environment is None else environment configured_endpoint = self._environment.get(TELEMETRY_ENDPOINT_VARIABLE) self.endpoint = ( configured_endpoint.strip() if configured_endpoint is not None else (endpoint or DEFAULT_TELEMETRY_ENDPOINT) ) configured_key = self._environment.get(TELEMETRY_KEY_VARIABLE) self._launcher_key = launcher_key if launcher_key is not None else configured_key self._transport = transport or _post_json self._session_metadata = dict(session_metadata or {}) self._monotonic = monotonic self._started_at = monotonic() self._enabled = bool(enabled) self._active = False self._accepting = False self._ready = False self._first_run = False self._installation_id: str | None = None self._base_payload: dict[str, object] = {} self._queue: queue.Queue[bytes] = queue.Queue(maxsize=QUEUE_CAPACITY) self._stop = threading.Event() self._worker: threading.Thread | None = None self._environment_worker: threading.Thread | None = None self._state_lock = threading.RLock() self._repeated_errors: dict[str, _RepeatedErrorState] = {} self._previous_sys_excepthook: Any = None self._previous_threading_excepthook: Any = None self._installed_sys_excepthook: Any = None self._installed_threading_excepthook: Any = None self._debug = self._environment.get(TELEMETRY_DEBUG_VARIABLE, "").casefold() in { "1", "true", "yes", } private_values = { self._environment.get("USER", ""), self._environment.get("LOGNAME", ""), socket.gethostname(), } self._private_literals = tuple( sorted( (value for value in private_values if len(value) >= 3), key=len, reverse=True, ) ) @property def enabled(self) -> bool: return self._enabled and self._active @property def installation_id(self) -> str | None: return self._installation_id def initialize(self) -> None: """Initialize the service once without propagating telemetry errors.""" with self._state_lock: if self._active or not self._enabled: return try: _validate_endpoint(self.endpoint) identity, created, first_seen = _load_or_create_installation_id( self.paths.installation_id_file ) self._installation_id = identity self._first_run = created self._base_payload = self._collect_common_payload(first_seen) self._stop.clear() self._accepting = True self._active = True worker = threading.Thread( target=self._worker_loop, name="ptlinux-telemetry", daemon=True, ) self._worker = worker worker.start() self._install_exception_hooks() except Exception as exc: self._enabled = False self._accepting = False self._active = False self._debug_log(f"initialization failed: {type(exc).__name__}") return self.send_event( "launcher_started", metadata={"cold_start": True, **self._session_metadata}, ) if self._first_run: self.send_event( "first_run_started", metadata={"install_type": self._install_type()}, ) self._start_environment_collection() def set_enabled(self, enabled: bool) -> None: """Apply a consent change to the current process.""" if enabled: with self._state_lock: if self._enabled and self._active: return if self._worker is not None and self._worker.is_alive(): return self._enabled = True self.initialize() return self._deactivate() def mark_ready(self, metadata: Mapping[str, object] | None = None) -> None: with self._state_lock: if self._ready or not self._active: return self._ready = True duration_ms = round((self._monotonic() - self._started_at) * 1000) self.send_event( "launcher_ready", metadata={"startup_duration_ms": duration_ms, **dict(metadata or {})}, ) if self._first_run: self.send_event( "first_run_completed", metadata={ "duration_ms": duration_ms, "install_type": self._install_type(), }, ) def send_event( self, event: str, severity: str = "info", message: str | None = None, metadata: Mapping[str, object] | None = None, context: Mapping[str, object] | None = None, ) -> bool: """Sanitize and enqueue one event without blocking on network I/O.""" try: if ( not self._accepting or not self._active or event not in VALID_EVENTS or severity not in VALID_SEVERITIES or self._installation_id is None ): return False suppressed, summaries, repeated_error = self._repeated_error_decision( event, severity, message, metadata, context, ) for summary in summaries: self._send_repeated_error(summary) if suppressed: return True with self._state_lock: base_payload = dict(self._base_payload) payload: dict[str, object] = { "installation_id": self._installation_id, "event": event, "severity": severity, "client_timestamp": datetime.now(timezone.utc).isoformat(), **base_payload, } if message is not None: payload["message"] = _truncate_utf8( self._sanitize_text(str(message)), MAX_MESSAGE_BYTES, ) if context: sanitized_context = self._sanitize_value(dict(context), key=None, depth=0) if isinstance(sanitized_context, dict): payload["context"] = _bound_mapping( sanitized_context, MAX_CONTEXT_BYTES, ) event_metadata = dict(metadata or {}) if event in _RESOURCE_USAGE_EVENTS: for key, value in _resource_usage().items(): event_metadata.setdefault(key, value) if event_metadata: sanitized_metadata = self._sanitize_value( event_metadata, key=None, depth=0, ) if isinstance(sanitized_metadata, dict): payload["metadata"] = _bound_mapping( sanitized_metadata, MAX_METADATA_BYTES, ) body = _encode_payload(payload) if body is None: return False self._queue.put_nowait(body) if repeated_error is not None: evicted = self._remember_repeated_error(repeated_error) if evicted is not None: self._send_repeated_error(evicted) self._debug_log(f"enqueued {event}") return True except (OSError, TypeError, ValueError, OverflowError, queue.Full): return False def report_exception( self, event: str, exception: BaseException, metadata: Mapping[str, object] | None = None, context: Mapping[str, object] | None = None, *, severity: str = "error", ) -> bool: try: frames = traceback.extract_tb(exception.__traceback__) last_frame = frames[-1] if frames else None formatted = "".join( traceback.TracebackException.from_exception(exception).format() ) details: dict[str, object] = dict(metadata or {}) details.update( { "exception_type": type(exception).__name__, "exception_module": type(exception).__module__, "sanitized_traceback": _truncate_characters( formatted, MAX_TRACEBACK_CHARACTERS, tail=True, ), } ) if last_frame is not None: details["source_module"] = Path(last_frame.filename).name details["source_function"] = last_frame.name return self.send_event( event, severity=severity, message=str(exception), metadata=details, context=context, ) except Exception: return False def report_subprocess_failure( self, event: str, result: object, command: Sequence[object] | None = None, metadata: Mapping[str, object] | None = None, ) -> bool: try: returncode = getattr(result, "returncode", None) timed_out = bool(getattr(result, "timed_out", False)) if returncode == 0 and not timed_out: return False details: dict[str, object] = dict(metadata or {}) details.update( { "exit_code": returncode, "timed_out": timed_out, "duration_ms": round( float(getattr(result, "duration_seconds", 0.0)) * 1000 ), "stdout": _truncate_characters( str(getattr(result, "stdout", "")), MAX_STDOUT_CHARACTERS, tail=True, ), "stderr": _truncate_characters( str(getattr(result, "stderr", "")), MAX_STDERR_CHARACTERS, tail=True, ), } ) arguments = command if arguments is None: candidate = getattr(result, "argv", None) if isinstance(candidate, Sequence) and not isinstance(candidate, str): arguments = candidate if arguments: details["command"] = [str(value) for value in arguments] if isinstance(returncode, int) and returncode < 0: details["signal"] = -returncode operation = details.get("operation") stage = details.get("stage") failure_context = { "operation": operation or "subprocess", "stage": stage if isinstance(stage, str) else "execution", } status = "unknown" if returncode is None else str(returncode) return self.send_event( event, severity="error", message=( "Subprocess timed out" if timed_out else f"Subprocess exited with status {status}" ), metadata=details, context=failure_context, ) except Exception: return False def _repeated_error_decision( self, event: str, severity: str, message: str | None, metadata: Mapping[str, object] | None, context: Mapping[str, object] | None, ) -> tuple[ bool, tuple[_RepeatedErrorState, ...], _RepeatedErrorState | None, ]: if event == "repeated_error" or not _is_error_event(event, severity): return False, (), None details = metadata or {} sanitized_message = _truncate_utf8( self._sanitize_text(str(message or "")), MAX_MESSAGE_BYTES, ) fingerprint = error_fingerprint( event, sanitized_message, exception_type=self._sanitize_text( _string_field(details, "exception_type") ), source_module=self._sanitize_text( _string_field(details, "source_module") ), source_function=self._sanitize_text( _string_field(details, "source_function") ), ) now = self._monotonic() summaries: list[_RepeatedErrorState] = [] with self._state_lock: current = self._repeated_errors.get(fingerprint) if ( current is not None and now - current.last_seen <= ERROR_SUPPRESSION_WINDOW_SECONDS ): current.last_seen = now current.suppressed_count += 1 return True, (), None if current is not None: self._repeated_errors.pop(fingerprint, None) if current.suppressed_count > 0: summaries.append(current) operation = _string_field(context or {}, "operation") or None stage = _string_field(context or {}, "stage") or None candidate = _RepeatedErrorState( fingerprint=fingerprint, event=event, first_seen=now, last_seen=now, suppressed_count=0, operation=operation, stage=stage, ) return False, tuple(summaries), candidate def _remember_repeated_error( self, state: _RepeatedErrorState, ) -> _RepeatedErrorState | None: with self._state_lock: if state.fingerprint in self._repeated_errors: return None self._repeated_errors[state.fingerprint] = state if len(self._repeated_errors) <= MAX_ERROR_FINGERPRINTS: return None oldest_fingerprint, oldest = min( self._repeated_errors.items(), key=lambda item: item[1].last_seen, ) self._repeated_errors.pop(oldest_fingerprint, None) return oldest if oldest.suppressed_count > 0 else None def _send_repeated_error(self, state: _RepeatedErrorState) -> None: metadata: dict[str, object] = { "fingerprint": state.fingerprint, "original_event": state.event, "suppressed_count": state.suppressed_count, "window_seconds": round( max(0.0, state.last_seen - state.first_seen), 3, ), } if state.stage: metadata["original_stage"] = state.stage self.send_event( "repeated_error", severity="warning", metadata=metadata, context={ "operation": state.operation or "telemetry", "stage": "error_suppression", }, ) def _flush_repeated_errors(self) -> None: with self._state_lock: summaries = tuple( state for state in self._repeated_errors.values() if state.suppressed_count > 0 ) self._repeated_errors.clear() for state in summaries: self._send_repeated_error(state) def shutdown(self, timeout_seconds: float = SHUTDOWN_WAIT_SECONDS) -> None: """Queue the closing event and give the daemon worker a bounded flush.""" with self._state_lock: if not self._active: return deadline = time.monotonic() + max(0.0, timeout_seconds) environment_worker = self._environment_worker if isinstance(environment_worker, threading.Thread) and ( environment_worker is not threading.current_thread() ): environment_worker.join( timeout=min( 0.5, max(0.0, (deadline - time.monotonic()) / 2), ) ) self._flush_repeated_errors() runtime_seconds = max(0, round(self._monotonic() - self._started_at)) if self._first_run and not self._ready: self.send_event( "first_run_failed", severity="error", message="Launcher did not reach the ready state", metadata={"duration_ms": runtime_seconds * 1000}, ) self.send_event( "launcher_closed", metadata={"runtime_seconds": runtime_seconds, **self._session_metadata}, ) with self._state_lock: self._accepting = False self._stop.set() worker = self._worker if isinstance(worker, threading.Thread) and ( worker is not threading.current_thread() ): worker.join(timeout=max(0.0, deadline - time.monotonic())) self._restore_exception_hooks() with self._state_lock: self._active = False def _deactivate(self) -> None: with self._state_lock: self._enabled = False self._accepting = False self._stop.set() self._repeated_errors.clear() while True: try: self._queue.get_nowait() except queue.Empty: break else: self._queue.task_done() self._restore_exception_hooks() with self._state_lock: self._active = False def _worker_loop(self) -> None: headers = { "Accept": "application/json", "Content-Type": "application/json", "User-Agent": f"PT-Linux/{__version__} (+https://pristontale.eu/)", } if self._launcher_key: headers["X-Launcher-Key"] = self._launcher_key while True: try: body = self._queue.get(timeout=0.1) except queue.Empty: if self._stop.is_set(): return continue try: if self._enabled: self._transport( self.endpoint, body, headers, DELIVERY_TIMEOUT_SECONDS, ) self._debug_log("send succeeded") except Exception as exc: self._debug_log(f"send failed: {type(exc).__name__}") finally: self._queue.task_done() def _start_environment_collection(self) -> None: with self._state_lock: if not self._active or not self._enabled: return worker = threading.Thread( target=self._collect_extended_environment, name="ptlinux-telemetry-environment", daemon=True, ) self._environment_worker = worker worker.start() def _collect_extended_environment(self) -> None: started_at = self._monotonic() try: details = collect_extended_environment(self.paths, self._environment) sanitized = self._sanitize_value(details, key=None, depth=0) if not isinstance(sanitized, dict): return except Exception as exc: self._debug_log( f"extended environment detection failed: {type(exc).__name__}" ) return with self._state_lock: if not self._active or not self._enabled: return for key, value in sanitized.items(): existing = self._base_payload.get(key) if isinstance(existing, dict) and isinstance(value, dict): self._base_payload[key] = {**existing, **value} else: self._base_payload[key] = value runtime = sanitized.get("runtime") dependencies = sanitized.get("dependencies") runtime_details = runtime if isinstance(runtime, dict) else {} dependency_details = dependencies if isinstance(dependencies, dict) else {} metadata = { "automatic": True, "duration_ms": round((self._monotonic() - started_at) * 1000), "dependency_healthy": dependency_details.get("healthy"), "missing_required": dependency_details.get("missing_required", []), "runtime_kind": runtime_details.get("kind"), "dxvk_install_status": runtime_details.get("dxvk_install_status"), "vulkan_64bit": runtime_details.get("vulkan_64bit"), "vulkan_32bit": runtime_details.get("vulkan_32bit"), } self.send_event( "compatibility_check_completed", metadata=metadata, context={"operation": "launcher", "stage": "environment_detection"}, ) missing_required = dependency_details.get("missing_required") if dependency_details.get("healthy") is False: self.send_event( "dependency_check_failed", severity="warning", message="One or more launcher dependencies are unavailable", metadata={ "dependencies": ( missing_required if isinstance(missing_required, list) else [] ), }, context={"operation": "launcher", "stage": "dependency_check"}, ) if dependency_details.get("wine_32bit_available") is False: self.send_event( "wine_32bit_check_failed", severity="warning", metadata={"dependency": "wine_32bit", "check": "32bit_wine"}, context={"operation": "launcher", "stage": "runtime_check"}, ) if runtime_details.get("vulkan_64bit") is False: self.send_event( "vulkan_check_failed", severity="warning", metadata={"dependency": "vulkan", "check": "64bit_vulkan"}, context={"operation": "launcher", "stage": "vulkan_check"}, ) if runtime_details.get("vulkan_32bit") is False: self.send_event( "vulkan_32bit_check_failed", severity="warning", metadata={"dependency": "vulkan_32bit", "check": "32bit_vulkan"}, context={"operation": "launcher", "stage": "vulkan_check"}, ) def _collect_common_payload(self, first_seen: str) -> dict[str, object]: environment = self._environment payload: dict[str, object] = { "launcher_version": __version__, "launcher_build": environment.get("PTLINUX_BUILD", __version__)[:32], "launcher_channel": environment.get("PTLINUX_CHANNEL", "stable")[:16], "python_version": platform.python_version(), "installation": { "first_seen": first_seen, "install_type": self._install_type(), "install_is_external": _install_is_external(self.paths, environment), "configuration_schema_version": _settings_schema_version(), }, "os": _operating_system_info(), "display": _display_info(environment), "filesystem": _filesystem_info(self.paths.data), } commit = environment.get("PTLINUX_COMMIT") if commit and re.fullmatch(r"[0-9a-fA-F]{7,40}", commit): payload["launcher_commit"] = commit[:12].lower() game_version = _current_game_version(self.paths.native_depot_state) if game_version is not None: payload["game_version"] = game_version filesystem = payload["filesystem"] if isinstance(filesystem, dict): installation = payload["installation"] if isinstance(installation, dict): installation["install_is_network_fs"] = bool( filesystem.get("network") ) return self._sanitize_value(payload, key=None, depth=0) # type: ignore[return-value] def _install_type(self) -> str: if self.paths.game_import_manifest.is_file(): return "imported" if self.paths.source_game.is_dir(): return "fresh" return "unknown" def _sanitize_value(self, value: object, *, key: str | None, depth: int) -> object: if key is not None and _is_secret_key(key): return "" if depth >= MAX_VALUE_DEPTH: return "" if value is None or isinstance(value, (bool, int)): return value if isinstance(value, float): return value if math.isfinite(value) else None if isinstance(value, (str, Path)): text = self._sanitize_text(str(value)) limit = _string_limit_for_key(key) return _truncate_characters(text, limit, tail=_keep_string_tail(key)) if isinstance(value, Mapping): result: dict[str, object] = {} for index, (raw_key, item) in enumerate(value.items()): if index >= MAX_CONTAINER_ITEMS: result["_truncated"] = True break name = _truncate_characters(self._sanitize_text(str(raw_key)), 128) result[name] = self._sanitize_value( item, key=name, depth=depth + 1, ) return result if isinstance(value, Sequence) and not isinstance(value, (bytes, bytearray)): items = list(value[:MAX_CONTAINER_ITEMS]) result = [ self._sanitize_value(item, key=key, depth=depth + 1) for item in items ] if len(value) > MAX_CONTAINER_ITEMS: result.append("") return result return f"<{type(value).__name__}>" def _sanitize_text(self, value: str) -> str: text = value.replace("\x00", "") for target, replacement in ( (str(self.paths.source_game), ""), (str(self.paths.runtime_view), ""), (str(self.paths.data), ""), ): if target and target != ".": text = text.replace(target.rstrip("/"), replacement) try: home = str(Path(self._environment.get("HOME", "")).expanduser()) except (OSError, RuntimeError): home = "" if home and home not in {".", "/"}: text = text.replace(home.rstrip("/"), "") text = _HOME_PATTERN.sub("", text) text = _WINDOWS_USER_PATTERN.sub(r"\1", text) text = _SSH_PATH_PATTERN.sub("", text) text = _EMAIL_PATTERN.sub("", text) text = _AUTHORIZATION_PATTERN.sub(r"\1", text) text = _SECRET_ASSIGNMENT_PATTERN.sub(r"\1\2", text) text = _URL_PATTERN.sub(_sanitize_url_match, text) for private in self._private_literals: text = re.sub( rf"(?", text, ) return text def _install_exception_hooks(self) -> None: previous_sys_excepthook = sys.excepthook self._previous_sys_excepthook = previous_sys_excepthook def sys_hook( exception_type: type[BaseException], exception: BaseException, traceback_object: TracebackType | None, ) -> None: _ = exception_type, traceback_object self.report_exception( "launcher_exception", exception, context={"operation": "launcher", "stage": "uncaught_exception"}, severity="critical", ) previous_sys_excepthook(exception_type, exception, traceback_object) sys.excepthook = sys_hook self._installed_sys_excepthook = sys_hook if hasattr(threading, "excepthook"): previous_threading_excepthook = threading.excepthook self._previous_threading_excepthook = previous_threading_excepthook def thread_hook(arguments: Any) -> None: self.report_exception( "thread_exception", arguments.exc_value, metadata={"thread_name": arguments.thread.name}, context={"operation": "launcher", "stage": "worker_thread"}, severity="critical", ) previous_threading_excepthook(arguments) threading.excepthook = thread_hook self._installed_threading_excepthook = thread_hook def _restore_exception_hooks(self) -> None: if ( self._previous_sys_excepthook is not None and sys.excepthook is self._installed_sys_excepthook ): sys.excepthook = self._previous_sys_excepthook if ( self._previous_threading_excepthook is not None and threading.excepthook is self._installed_threading_excepthook ): threading.excepthook = self._previous_threading_excepthook self._previous_sys_excepthook = None self._previous_threading_excepthook = None self._installed_sys_excepthook = None self._installed_threading_excepthook = None def _debug_log(self, message: str) -> None: if not self._debug: return try: print(f"Telemetry: {message}", file=sys.stderr) except Exception: pass _shared_service: TelemetryService | None = None _shared_service_lock = threading.Lock() def initialize_shared_telemetry( paths: AppPaths, *, session_metadata: Mapping[str, object] | None = None, ) -> tuple[TelemetryService, bool]: """Return the process-wide service and whether this caller owns its shutdown.""" global _shared_service with _shared_service_lock: if _shared_service is not None: return _shared_service, False service = TelemetryService( paths, enabled=_telemetry_enabled_from_settings(paths), session_metadata=session_metadata, ) _shared_service = service service.initialize() return service, True def current_telemetry() -> TelemetryService | None: return _shared_service def set_shared_telemetry_enabled(enabled: bool) -> None: service = _shared_service if service is not None: service.set_enabled(enabled) def shutdown_shared_telemetry( service: TelemetryService, *, timeout_seconds: float = SHUTDOWN_WAIT_SECONDS, ) -> None: global _shared_service service.shutdown(timeout_seconds=timeout_seconds) with _shared_service_lock: if _shared_service is service: _shared_service = None def _telemetry_enabled_from_settings(paths: AppPaths) -> bool: try: from .settings import Settings, load_settings if not paths.settings_file.is_file(): return Settings().telemetry_enabled return load_settings(paths).telemetry_enabled except (OSError, ValueError): return False def _post_json( endpoint: str, body: bytes, headers: Mapping[str, str], timeout_seconds: float, ) -> None: request = Request(endpoint, data=body, headers=dict(headers), method="POST") with urlopen(request, timeout=timeout_seconds) as response: status = getattr(response, "status", response.getcode()) response.read(1024) if not isinstance(status, int) or status < 200 or status >= 300: raise OSError(f"telemetry endpoint returned HTTP {status}") def _validate_endpoint(endpoint: str) -> None: parsed = urlsplit(endpoint) if ( parsed.scheme != "https" or not parsed.hostname or parsed.username is not None or parsed.password is not None or parsed.query or parsed.fragment ): raise ValueError("telemetry endpoint must be a credential-free HTTPS URL") def _load_or_create_installation_id(path: Path) -> tuple[str, bool, str]: path.parent.mkdir(parents=True, exist_ok=True) try: identity = _read_installation_id(path) created = False except FileNotFoundError: candidate = str(uuid.uuid4()) temporary = path.with_name( f".{path.name}.{os.getpid()}.{threading.get_ident()}.{uuid.uuid4().hex}.tmp" ) try: with temporary.open("x", encoding="ascii", newline="\n") as handle: os.fchmod(handle.fileno(), _INSTALL_ID_FILE_MODE) handle.write(candidate + "\n") handle.flush() os.fsync(handle.fileno()) try: os.link(temporary, path) except FileExistsError: identity = _read_installation_id(path) created = False else: identity = candidate created = True _fsync_directory(path.parent) finally: temporary.unlink(missing_ok=True) os.chmod(path, _INSTALL_ID_FILE_MODE) first_seen = datetime.fromtimestamp(path.stat().st_mtime, timezone.utc).isoformat() return identity, created, first_seen def _read_installation_id(path: Path) -> str: raw = path.read_text(encoding="ascii").strip() try: value = uuid.UUID(raw) except (ValueError, AttributeError) as exc: raise ValueError("invalid telemetry installation ID") from exc if value.version != 4 or str(value) != raw.casefold(): raise ValueError("invalid telemetry installation ID") return str(value) def _fsync_directory(path: Path) -> None: try: descriptor = os.open(path, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0)) except OSError: return try: os.fsync(descriptor) except OSError: pass finally: os.close(descriptor) def _operating_system_info() -> dict[str, object]: release = _read_os_release() return { "name": release.get("NAME", platform.system() or "unknown"), "id": release.get("ID", "unknown"), "version": release.get("PRETTY_NAME", release.get("VERSION", "unknown")), "version_id": release.get("VERSION_ID", "unknown"), "kernel": platform.release(), "architecture": platform.machine() or "unknown", "machine_architecture": platform.machine() or "unknown", "glibc_version": platform.libc_ver()[1] or None, } def _read_os_release(path: Path = Path("/etc/os-release")) -> dict[str, str]: try: lines = path.read_text(encoding="utf-8", errors="replace").splitlines() except OSError: return {} result: dict[str, str] = {} for line in lines: if not line or line.startswith("#") or "=" not in line: continue key, raw = line.split("=", 1) if not re.fullmatch(r"[A-Z][A-Z0-9_]*", key): continue value = raw.strip() if len(value) >= 2 and value[0] == value[-1] and value[0] in {"'", '"'}: value = value[1:-1] value = value.replace(r"\n", " ").replace(r"\"", '"').replace(r"\\", "\\") result[key] = value[:255] return result def _display_info(environment: Mapping[str, str]) -> dict[str, object]: session_type = environment.get("XDG_SESSION_TYPE", "").casefold() if session_type not in {"wayland", "x11"}: if environment.get("WAYLAND_DISPLAY"): session_type = "wayland" elif environment.get("DISPLAY"): session_type = "x11" else: session_type = "unknown" desktop = environment.get("XDG_CURRENT_DESKTOP", "") desktop = desktop.split(":", 1)[0].strip().casefold() or "unknown" session = environment.get("DESKTOP_SESSION", "").strip().casefold() or "unknown" return { "server": session_type, "desktop": desktop[:64], "session": session[:64], } def _filesystem_info(target: Path) -> dict[str, object]: existing = target while not existing.exists() and existing != existing.parent: existing = existing.parent filesystem_type = "unknown" readonly = False noexec = False try: mounts = _read_mountinfo() resolved = existing.resolve(strict=True) matches = [ mount for mount in mounts if resolved == mount[0] or resolved.is_relative_to(mount[0]) ] if matches: mountpoint, filesystem_type, options = max( matches, key=lambda item: len(item[0].parts), ) _ = mountpoint readonly = "ro" in options noexec = "noexec" in options except (OSError, RuntimeError, ValueError): pass try: writable = os.access(existing, os.W_OK) and not readonly except OSError: writable = False return { "type": filesystem_type, "writable": writable, "readonly": readonly, "noexec": noexec, "network": filesystem_type.casefold() in _NETWORK_FILESYSTEMS, } def _read_mountinfo(path: Path = Path("/proc/self/mountinfo")) -> list[tuple[Path, str, set[str]]]: mounts: list[tuple[Path, str, set[str]]] = [] for line in path.read_text(encoding="utf-8", errors="replace").splitlines(): fields = line.split() try: separator = fields.index("-") except ValueError: continue if separator < 6 or len(fields) <= separator + 3: continue mountpoint = Path(_unescape_mount_field(fields[4])) filesystem_type = fields[separator + 1] options = set(fields[5].split(",")) | set(fields[separator + 3].split(",")) mounts.append((mountpoint, filesystem_type, options)) return mounts def _unescape_mount_field(value: str) -> str: return re.sub( r"\\([0-7]{3})", lambda match: chr(int(match.group(1), 8)), value, ) def _install_is_external(paths: AppPaths, environment: Mapping[str, str]) -> bool: home_value = environment.get("HOME") if not home_value: return False try: home = Path(home_value).expanduser().resolve() data = paths.data.resolve() return data != home and not data.is_relative_to(home) except (OSError, RuntimeError): return False def _current_game_version(path: Path) -> str | None: try: value = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): return None if not isinstance(value, dict): return None version = value.get("version") return version[:32] if isinstance(version, str) and version else None def _settings_schema_version() -> int: try: from .settings import SETTINGS_SCHEMA return SETTINGS_SCHEMA except Exception: return 1 def _is_secret_key(key: str) -> bool: normalized = key.casefold().replace("-", "_").strip() return normalized in _SECRET_KEYS def _string_limit_for_key(key: str | None) -> int: normalized = (key or "").casefold() if normalized == "stderr": return MAX_STDERR_CHARACTERS if normalized == "stdout": return MAX_STDOUT_CHARACTERS if normalized in {"sanitized_traceback", "traceback"}: return MAX_TRACEBACK_CHARACTERS return 8192 def _keep_string_tail(key: str | None) -> bool: return (key or "").casefold() in {"stderr", "stdout", "sanitized_traceback", "traceback"} def _sanitize_url_match(match: re.Match[str]) -> str: raw = match.group(0) trailing = "" while raw and raw[-1] in ".,;:)]": trailing = raw[-1] + trailing raw = raw[:-1] try: parsed = urlsplit(raw) except ValueError: return "" + trailing host = parsed.hostname or "" if not host: return "" + trailing netloc = host if ":" in host and not host.startswith("["): netloc = f"[{host}]" try: if parsed.port is not None: netloc += f":{parsed.port}" except ValueError: return "" + trailing query = parsed.query if parsed.username is not None or any( str(key).casefold() in _SECRET_QUERY_KEYS for key, _value in parse_qsl(query) ): query = "" if query else "" return urlunsplit((parsed.scheme, netloc, parsed.path, query, "")) + trailing def _truncate_characters(value: str, limit: int, *, tail: bool = False) -> str: if len(value) <= limit: return value marker = "" remaining = max(0, limit - len(marker)) return marker + value[-remaining:] if tail else value[:remaining] + marker def _truncate_utf8(value: str, limit: int, *, tail: bool = False) -> str: encoded = value.encode("utf-8") if len(encoded) <= limit: return value marker = b"" remaining = max(0, limit - len(marker)) chunk = encoded[-remaining:] if tail else encoded[:remaining] decoded = chunk.decode("utf-8", errors="ignore") return "" + decoded if tail else decoded + "" def _bound_mapping(value: dict[str, object], limit: int) -> dict[str, object]: if len(_json_bytes(value)) <= limit: return value bounded: dict[str, object] = {} for key, item in value.items(): candidate = {**bounded, key: item, "_truncated": True} if len(_json_bytes(candidate)) <= limit: bounded[key] = item continue if isinstance(item, str): for string_limit in (4096, 2048, 1024, 512, 256, 128): shortened = _truncate_utf8(item, string_limit, tail=_keep_string_tail(key)) candidate = {**bounded, key: shortened, "_truncated": True} if len(_json_bytes(candidate)) <= limit: bounded[key] = shortened break bounded["_truncated"] = True while len(_json_bytes(bounded)) > limit and len(bounded) > 1: last = next(key for key in reversed(bounded) if key != "_truncated") bounded.pop(last) return bounded def _encode_payload(payload: dict[str, object]) -> bytes | None: body = _json_bytes(payload) if len(body) <= MAX_REQUEST_BYTES: return body for key in ("metadata", "context"): if key in payload: payload[key] = {"_truncated": True} body = _json_bytes(payload) if len(body) <= MAX_REQUEST_BYTES: return body if isinstance(payload.get("message"), str): payload["message"] = _truncate_utf8(str(payload["message"]), 2048) body = _json_bytes(payload) return body if len(body) <= MAX_REQUEST_BYTES else None def _json_bytes(value: object) -> bytes: return json.dumps( value, ensure_ascii=False, separators=(",", ":"), sort_keys=True, ).encode("utf-8") def normalized_error_message(message: str) -> str: """Normalize dynamic values for stable server-compatible fingerprints.""" value = _UUID_PATTERN.sub("", message) value = _MEMORY_ADDRESS_PATTERN.sub("
", value) return re.sub(r"\b\d{4}-\d{2}-\d{2}[T ][0-9:.+-Z]+\b", "", value) def _resource_usage() -> dict[str, object]: """Read one bounded process-usage snapshot on the Linux launcher host.""" try: usage = resource.getrusage(resource.RUSAGE_SELF) peak_value = float(usage.ru_maxrss) peak_divisor = 1024 * 1024 if sys.platform == "darwin" else 1024 peak_memory_mb = peak_value / peak_divisor cpu_time_seconds = float(usage.ru_utime) + float(usage.ru_stime) except (OSError, TypeError, ValueError, OverflowError): return {} if ( not math.isfinite(peak_memory_mb) or not math.isfinite(cpu_time_seconds) or peak_memory_mb < 0 or cpu_time_seconds < 0 ): return {} return { "peak_memory_mb": round(peak_memory_mb, 2), "cpu_time_seconds": round(cpu_time_seconds, 3), } def _is_error_event(event: str, severity: str) -> bool: return ( severity in {"error", "critical"} or "failed" in event or "exception" in event or "error" in event ) def _string_field(values: Mapping[str, object], key: str) -> str: value = values.get(key) return value if isinstance(value, str) else "" def error_fingerprint( event: str, message: str, *, exception_type: str = "", source_module: str = "", source_function: str = "", ) -> str: """Create a privacy-safe deterministic fingerprint for tests and server parity.""" material = "\n".join( (event, exception_type, source_module, source_function, normalized_error_message(message)) ) return hashlib.sha256(material.encode("utf-8", errors="replace")).hexdigest() def classify_network_failure(error: object) -> str: """Return one stable low-cardinality class for download failures.""" message = str(error).casefold() name = type(error).__name__.casefold() if "timeout" in name or "timed out" in message or "timeout" in message: return "connection_timeout" if any( value in message for value in ( "name or service not known", "name resolution", "nodename nor servname", "temporary failure in name", ) ): return "dns_failure" if "connection refused" in message: return "connection_refused" if "connection reset" in message or "broken pipe" in message: return "connection_reset" if any(value in message for value in ("certificate", "ssl", "tls")): return "tls_failure" if re.search(r"\bhttp\s+\d{3}\b", message): return "http_error" if any( value in message for value in ("sha-1 mismatch", "sha-256", "checksum", "content-integrity") ): return "content_integrity" if "could not connect" in message or "unreachable" in message: return "cdn_unreachable" return "unknown" def network_http_status(error: object) -> int | None: """Extract an HTTP status without exposing a failed request URL.""" current: object | None = error seen: set[int] = set() while current is not None and id(current) not in seen: seen.add(id(current)) status = getattr(current, "code", None) if isinstance(status, int) and 100 <= status <= 599: return status match = _HTTP_STATUS_PATTERN.search(str(current)) if match is not None: return int(match.group(1)) current = getattr(current, "__cause__", None) or getattr( current, "__context__", None, ) return None