1595 lines
46 KiB
Python
1595 lines
46 KiB
Python
"""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"
|
|
TELEMETRY_PROCESS_ROLE_VARIABLE = "PTLINUX_TELEMETRY_PROCESS_ROLE"
|
|
TELEMETRY_OPERATION_PROCESS_ROLE = "operation"
|
|
|
|
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"(?<![\w.+-])[\w.+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}(?![\w.-])"
|
|
)
|
|
_HOME_PATTERN = re.compile(r"(?<![A-Za-z0-9_])/(?:home/[^/\s]+|root)(?=/|\b)")
|
|
_WINDOWS_USER_PATTERN = re.compile(
|
|
r"(?i)(?<![A-Za-z0-9_])([A-Z]:[\\/]users[\\/])[^\\/\s]+"
|
|
)
|
|
_SSH_PATH_PATTERN = re.compile(r"(?i)(?:<HOME>|/home/<user>|/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._reports_launcher_lifecycle = (
|
|
self._environment.get(TELEMETRY_PROCESS_ROLE_VARIABLE, "").strip().casefold()
|
|
!= TELEMETRY_OPERATION_PROCESS_ROLE
|
|
)
|
|
self._enabled = bool(enabled)
|
|
self._active = False
|
|
self._accepting = False
|
|
self._closing = 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._closing = False
|
|
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
|
|
|
|
if self._reports_launcher_lifecycle:
|
|
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
|
|
if not self._reports_launcher_lifecycle:
|
|
return
|
|
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:
|
|
"""Finish this process role and give the daemon worker a bounded flush."""
|
|
|
|
with self._state_lock:
|
|
if not self._active or self._closing:
|
|
return
|
|
self._closing = True
|
|
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._reports_launcher_lifecycle:
|
|
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._closing = True
|
|
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
|
|
if not self._reports_launcher_lifecycle:
|
|
# Dashboard operation subprocesses enrich their own events with the
|
|
# detected environment, while the dashboard remains the sole source
|
|
# of launcher lifecycle and automatic compatibility events.
|
|
return
|
|
|
|
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")
|
|
generic_dependency_failures = (
|
|
[
|
|
str(value)
|
|
for value in missing_required
|
|
if isinstance(value, str) and value
|
|
]
|
|
if isinstance(missing_required, list)
|
|
else []
|
|
)
|
|
if (
|
|
dependency_details.get("container_ready") is False
|
|
and "container_host" not in generic_dependency_failures
|
|
):
|
|
generic_dependency_failures.append("container_host")
|
|
if (
|
|
dependency_details.get("healthy") is False
|
|
and generic_dependency_failures
|
|
):
|
|
self.send_event(
|
|
"dependency_check_failed",
|
|
severity="warning",
|
|
message="One or more launcher dependencies are unavailable",
|
|
metadata={
|
|
"dependencies": generic_dependency_failures,
|
|
"required_shared_libraries_present": dependency_details.get(
|
|
"required_shared_libraries_present"
|
|
),
|
|
"vulkan_64bit_available": dependency_details.get(
|
|
"vulkan_64bit_available"
|
|
),
|
|
"vulkan_32bit_available": dependency_details.get(
|
|
"vulkan_32bit_available"
|
|
),
|
|
},
|
|
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",
|
|
"loader_available": runtime_details.get(
|
|
"vulkan_32bit_loader_available"
|
|
),
|
|
"icd_available": runtime_details.get(
|
|
"vulkan_32bit_icd_available"
|
|
),
|
|
"detection": runtime_details.get("vulkan_32bit_detection"),
|
|
},
|
|
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 "<redacted>"
|
|
if depth >= MAX_VALUE_DEPTH:
|
|
return "<max-depth>"
|
|
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("<truncated>")
|
|
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), "<GAME_DIR>"),
|
|
(str(self.paths.runtime_view), "<RUNTIME_DIR>"),
|
|
(str(self.paths.data), "<INSTALL_DIR>"),
|
|
):
|
|
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("/"), "<HOME>")
|
|
text = _HOME_PATTERN.sub("<HOME>", text)
|
|
text = _WINDOWS_USER_PATTERN.sub(r"\1<user>", text)
|
|
text = _SSH_PATH_PATTERN.sub("<SSH_PATH>", text)
|
|
text = _EMAIL_PATTERN.sub("<email>", text)
|
|
text = _AUTHORIZATION_PATTERN.sub(r"\1<redacted>", text)
|
|
text = _SECRET_ASSIGNMENT_PATTERN.sub(r"\1\2<redacted>", text)
|
|
text = _URL_PATTERN.sub(_sanitize_url_match, text)
|
|
for private in self._private_literals:
|
|
text = re.sub(
|
|
rf"(?<![A-Za-z0-9_.-]){re.escape(private)}(?![A-Za-z0-9_.-])",
|
|
"<redacted>",
|
|
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 "<redacted-url>" + trailing
|
|
host = parsed.hostname or ""
|
|
if not host:
|
|
return "<redacted-url>" + 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 "<redacted-url>" + 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 = "<redacted>" 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 = "<truncated>"
|
|
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"<truncated>"
|
|
remaining = max(0, limit - len(marker))
|
|
chunk = encoded[-remaining:] if tail else encoded[:remaining]
|
|
decoded = chunk.decode("utf-8", errors="ignore")
|
|
return "<truncated>" + decoded if tail else decoded + "<truncated>"
|
|
|
|
|
|
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("<UUID>", message)
|
|
value = _MEMORY_ADDRESS_PATTERN.sub("<ADDRESS>", value)
|
|
return re.sub(r"\b\d{4}-\d{2}-\d{2}[T ][0-9:.+-Z]+\b", "<TIMESTAMP>", 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
|