1092 lines
32 KiB
Python
1092 lines
32 KiB
Python
"""Native, verified updater for the public PristonTale EU depot."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from base64 import b64decode
|
|
from binascii import Error as BinAsciiError
|
|
from concurrent.futures import FIRST_COMPLETED, Future, ThreadPoolExecutor, wait
|
|
from dataclasses import asdict, dataclass
|
|
from datetime import datetime
|
|
import hashlib
|
|
from http.client import HTTPException
|
|
import json
|
|
import os
|
|
from pathlib import Path, PurePosixPath
|
|
import re
|
|
import threading
|
|
import time
|
|
from typing import Callable, Iterable, Mapping, Protocol
|
|
from urllib.error import HTTPError
|
|
from urllib.parse import quote, urlparse
|
|
from urllib.request import Request, urlopen
|
|
from uuid import uuid4
|
|
|
|
from . import __version__
|
|
from .paths import AppPaths
|
|
|
|
|
|
CDN_CONFIGURATION_URL = "https://cdn-main.chatplex.io/pristontale-eu/cdn_configuration.json"
|
|
DEPOT_ID = 1
|
|
FALLBACK_SERVERS = (
|
|
"cdn-fr-sbg1.chatplex.io/pristontale-eu",
|
|
"cdn-us-ore1.chatplex.io/pristontale-eu",
|
|
)
|
|
READ_SIZE = 128 * 1024
|
|
HASH_READ_SIZE = 1024 * 1024
|
|
DOWNLOAD_WORKERS = 8
|
|
DOWNLOAD_RETRY_DELAYS_SECONDS = (1.0, 2.0, 4.0, 8.0, 16.0)
|
|
PROGRESS_UPDATE_INTERVAL_SECONDS = 0.075
|
|
MAX_CREATE_IF_MISSING_BYTES = 16 * 1024 * 1024
|
|
MAX_LINE_ENDING_RECOVERY_BYTES = 1024 * 1024
|
|
SHA1_PATTERN = re.compile(r"^[0-9a-fA-F]{40}$")
|
|
CONTENT_RANGE_PATTERN = re.compile(r"^bytes\s+(\d+)-(\d+)/(\d+)$", re.IGNORECASE)
|
|
|
|
|
|
class DepotError(RuntimeError):
|
|
"""Raised when a depot response or update operation cannot be trusted."""
|
|
|
|
|
|
class _RetryableDownloadError(DepotError):
|
|
"""One transient network or content-integrity failure."""
|
|
|
|
def __init__(self, message: str, *, discard_partial: bool = False) -> None:
|
|
super().__init__(message)
|
|
self.discard_partial = discard_partial
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class DepotFile:
|
|
"""One exact file in the published depot build."""
|
|
|
|
path: str
|
|
build: int
|
|
size: int
|
|
sha1: str
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class CreateIfMissing:
|
|
"""A small publisher-provided file created only when absent."""
|
|
|
|
path: str
|
|
content_base64: str
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class DepotSnapshot:
|
|
"""The current, verified shape of the public game depot."""
|
|
|
|
region: str
|
|
server: str
|
|
name: str
|
|
build: int
|
|
version: str
|
|
files: tuple[DepotFile, ...]
|
|
deleted_files: tuple[str, ...]
|
|
create_if_missing: tuple[CreateIfMissing, ...]
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class PlannedDownload:
|
|
file: DepotFile
|
|
reason: str
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class UpdatePlan:
|
|
"""The safe changes needed to make a local tree match one depot snapshot."""
|
|
|
|
downloads: tuple[PlannedDownload, ...]
|
|
deletes: tuple[str, ...]
|
|
files_total: int
|
|
bytes_total: int
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class SyncProgress:
|
|
"""A thread-safe progress snapshot emitted by ``sync_managed_game``."""
|
|
|
|
phase: str
|
|
files_completed: int
|
|
files_total: int
|
|
bytes_completed: int
|
|
bytes_total: int
|
|
current_file: str | None
|
|
bytes_transferred: int = 0
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class DepotSyncResult:
|
|
"""Persistent result of one native depot synchronization."""
|
|
|
|
region: str
|
|
server: str
|
|
name: str
|
|
build: int
|
|
version: str
|
|
files_total: int
|
|
files_downloaded: int
|
|
bytes_downloaded: int
|
|
files_deleted: int
|
|
files_created: int
|
|
repair: bool
|
|
dry_run: bool
|
|
completed_at: str
|
|
|
|
|
|
ProgressCallback = Callable[[SyncProgress], None]
|
|
|
|
|
|
class HttpResponse(Protocol):
|
|
"""The small response surface used by the native depot client."""
|
|
|
|
headers: Mapping[str, str]
|
|
|
|
def read(self, size: int = -1) -> bytes: ...
|
|
|
|
def close(self) -> None: ...
|
|
|
|
|
|
HttpOpener = Callable[..., HttpResponse]
|
|
|
|
|
|
class DepotClient:
|
|
"""Small HTTP client for the publisher's documented public depot endpoints."""
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
timeout_seconds: float = 60.0,
|
|
opener: HttpOpener = urlopen,
|
|
retry_delays: tuple[float, ...] = DOWNLOAD_RETRY_DELAYS_SECONDS,
|
|
sleeper: Callable[[float], None] = time.sleep,
|
|
) -> None:
|
|
if timeout_seconds <= 0:
|
|
raise ValueError("timeout_seconds must be positive")
|
|
if any(delay < 0 for delay in retry_delays):
|
|
raise ValueError("retry delays must not be negative")
|
|
self.timeout_seconds = timeout_seconds
|
|
self._opener = opener
|
|
self._retry_delays = retry_delays
|
|
self._sleeper = sleeper
|
|
self._candidate_servers: tuple[str, ...] = ()
|
|
|
|
def fetch_snapshot(self, *, preferred_region: str = "Europe") -> DepotSnapshot:
|
|
"""Fetch the live depot and build manifests, trying trusted regional fallbacks."""
|
|
|
|
configuration: Mapping[str, object] | None = None
|
|
configuration_error: DepotError | None = None
|
|
try:
|
|
configuration = self._get_json(CDN_CONFIGURATION_URL)
|
|
except DepotError as exc:
|
|
configuration_error = exc
|
|
|
|
candidates = _server_candidates(configuration, preferred_region)
|
|
self._candidate_servers = tuple(server for _region, server in candidates)
|
|
last_error = configuration_error
|
|
for region, server in candidates:
|
|
try:
|
|
depot_manifest = self._get_json(_depot_manifest_url(server))
|
|
build = _positive_int(depot_manifest.get("Build"), "depot Build")
|
|
build_manifest = self._get_json(_build_manifest_url(server, build))
|
|
return _parse_snapshot(region, server, depot_manifest, build_manifest)
|
|
except DepotError as exc:
|
|
last_error = exc
|
|
|
|
if last_error is not None:
|
|
raise DepotError(f"could not load the PristonTale EU depot: {last_error}")
|
|
raise DepotError("could not load the PristonTale EU depot")
|
|
|
|
def download_file(
|
|
self,
|
|
server: str,
|
|
file: DepotFile,
|
|
destination: Path,
|
|
*,
|
|
progress: Callable[[int], None] | None = None,
|
|
) -> None:
|
|
"""Download, resume, and verify one file before replacing its destination.
|
|
|
|
The current publisher CDN normalizes a small number of legacy text files
|
|
from CRLF to LF while its manifest still hashes the CRLF bytes. A
|
|
candidate is accepted only when restoring bare LF line endings produces
|
|
the manifest's exact size and SHA-1.
|
|
"""
|
|
|
|
servers = _ordered_download_servers(server, self._candidate_servers)
|
|
temporary = _download_partial_path(destination)
|
|
last_error: _RetryableDownloadError | None = None
|
|
attempts = len(self._retry_delays) + 1
|
|
for attempt_index in range(attempts):
|
|
attempt_server = servers[attempt_index % len(servers)]
|
|
try:
|
|
self._download_file_once(
|
|
attempt_server,
|
|
file,
|
|
destination,
|
|
progress=progress,
|
|
)
|
|
return
|
|
except _RetryableDownloadError as exc:
|
|
last_error = exc
|
|
if exc.discard_partial:
|
|
temporary.unlink(missing_ok=True)
|
|
if attempt_index >= len(self._retry_delays):
|
|
break
|
|
self._sleeper(self._retry_delays[attempt_index])
|
|
|
|
assert last_error is not None
|
|
resume_hint = (
|
|
"; partial data was retained for the next update run"
|
|
if temporary.is_file() and not temporary.is_symlink()
|
|
else ""
|
|
)
|
|
raise DepotError(
|
|
f"download failed for {file.path} after {attempts} attempts: "
|
|
f"{last_error}{resume_hint}"
|
|
) from last_error
|
|
|
|
def _download_file_once(
|
|
self,
|
|
server: str,
|
|
file: DepotFile,
|
|
destination: Path,
|
|
*,
|
|
progress: Callable[[int], None] | None,
|
|
) -> None:
|
|
temporary = _download_partial_path(destination)
|
|
written = 0
|
|
digest = hashlib.sha1(usedforsecurity=False)
|
|
if temporary.is_symlink():
|
|
raise DepotError(f"refusing to use symbolic-link partial file for {file.path}")
|
|
if temporary.exists():
|
|
if not temporary.is_file():
|
|
raise DepotError(f"partial download is not a file for {file.path}")
|
|
try:
|
|
with temporary.open("rb") as partial_handle:
|
|
while True:
|
|
chunk = partial_handle.read(HASH_READ_SIZE)
|
|
if not chunk:
|
|
break
|
|
written += len(chunk)
|
|
digest.update(chunk)
|
|
except OSError as exc:
|
|
raise DepotError(f"cannot inspect partial download for {file.path}: {exc}") from exc
|
|
|
|
if written > file.size:
|
|
temporary.unlink()
|
|
written = 0
|
|
digest = hashlib.sha1(usedforsecurity=False)
|
|
elif written == file.size:
|
|
if digest.hexdigest().casefold() == file.sha1.casefold():
|
|
temporary.replace(destination)
|
|
if progress is not None:
|
|
progress(written)
|
|
return
|
|
temporary.unlink()
|
|
written = 0
|
|
digest = hashlib.sha1(usedforsecurity=False)
|
|
elif written and _recover_crlf_manifest_file(temporary, destination, file):
|
|
temporary.unlink(missing_ok=True)
|
|
if progress is not None:
|
|
progress(file.size)
|
|
return
|
|
if progress is not None:
|
|
progress(written)
|
|
|
|
requested_offset = written
|
|
headers = _headers()
|
|
if requested_offset:
|
|
headers["Range"] = f"bytes={requested_offset}-"
|
|
url = _file_url(server, file)
|
|
request = Request(url, headers=headers, method="GET")
|
|
try:
|
|
response = self._opener(request, timeout=self.timeout_seconds)
|
|
except HTTPError as exc:
|
|
if exc.code == 416 and requested_offset:
|
|
raise _RetryableDownloadError(
|
|
f"server rejected the saved offset for {file.path}",
|
|
discard_partial=True,
|
|
) from exc
|
|
raise _RetryableDownloadError(
|
|
f"HTTP {exc.code} from {server} for {file.path}"
|
|
) from exc
|
|
except (OSError, HTTPException) as exc:
|
|
raise _RetryableDownloadError(
|
|
f"could not connect to {server} for {file.path}: {exc}"
|
|
) from exc
|
|
|
|
try:
|
|
append = False
|
|
if requested_offset:
|
|
content_range = response.headers.get("Content-Range")
|
|
status = getattr(response, "status", None)
|
|
if status == 206 or content_range is not None:
|
|
try:
|
|
range_start, _range_end, range_total = _content_range(content_range)
|
|
except DepotError as exc:
|
|
raise _RetryableDownloadError(
|
|
f"invalid resume response from {server} for {file.path}: {exc}"
|
|
) from exc
|
|
if range_start != requested_offset or range_total != file.size:
|
|
raise _RetryableDownloadError(
|
|
f"resume response did not match {file.path}: "
|
|
f"offset {range_start}/{requested_offset}, size {range_total}/{file.size}"
|
|
)
|
|
append = True
|
|
else:
|
|
# A server may ignore Range and send a normal 200 response. In
|
|
# that case safely restart this file using the full response body.
|
|
written = 0
|
|
digest = hashlib.sha1(usedforsecurity=False)
|
|
if progress is not None:
|
|
progress(written)
|
|
|
|
try:
|
|
content_length = _content_length(response.headers.get("Content-Length"))
|
|
except DepotError as exc:
|
|
raise _RetryableDownloadError(
|
|
f"invalid response length from {server} for {file.path}: {exc}"
|
|
) from exc
|
|
remaining = file.size - written
|
|
if content_length is not None and content_length > remaining:
|
|
raise _RetryableDownloadError(
|
|
f"download exceeds expected size for {file.path}: "
|
|
f"server declared {content_length}, expected at most {remaining}"
|
|
)
|
|
|
|
mode = "ab" if append else "wb"
|
|
try:
|
|
with temporary.open(mode) as handle:
|
|
while True:
|
|
try:
|
|
chunk = response.read(READ_SIZE)
|
|
except (OSError, HTTPException) as exc:
|
|
raise _RetryableDownloadError(
|
|
f"could not read {file.path} from {server}: {exc}"
|
|
) from exc
|
|
if not chunk:
|
|
break
|
|
written += len(chunk)
|
|
if written > file.size:
|
|
raise _RetryableDownloadError(
|
|
f"download exceeds expected size for {file.path}",
|
|
discard_partial=True,
|
|
)
|
|
digest.update(chunk)
|
|
handle.write(chunk)
|
|
if progress is not None:
|
|
progress(written)
|
|
except _RetryableDownloadError:
|
|
raise
|
|
except OSError as exc:
|
|
raise DepotError(f"cannot write partial download for {file.path}: {exc}") from exc
|
|
|
|
if written != file.size or digest.hexdigest().casefold() != file.sha1.casefold():
|
|
if _recover_crlf_manifest_file(temporary, destination, file):
|
|
temporary.unlink(missing_ok=True)
|
|
if progress is not None:
|
|
progress(file.size)
|
|
return
|
|
if written != file.size:
|
|
raise _RetryableDownloadError(
|
|
f"incomplete response from {server} for {file.path}: "
|
|
f"{written}/{file.size} bytes"
|
|
)
|
|
raise _RetryableDownloadError(
|
|
f"SHA-1 mismatch for {file.path} from {server}",
|
|
discard_partial=True,
|
|
)
|
|
temporary.replace(destination)
|
|
finally:
|
|
try:
|
|
response.close()
|
|
except (OSError, HTTPException):
|
|
pass
|
|
|
|
def _get_json(self, url: str) -> Mapping[str, object]:
|
|
request = Request(url, headers=_headers(), method="GET")
|
|
try:
|
|
response = self._opener(request, timeout=self.timeout_seconds)
|
|
except (OSError, HTTPException) as exc:
|
|
raise DepotError(f"request failed for {url}: {exc}") from exc
|
|
payload = b""
|
|
try:
|
|
payload = response.read()
|
|
except (OSError, HTTPException) as exc:
|
|
raise DepotError(f"could not read {url}: {exc}") from exc
|
|
finally:
|
|
try:
|
|
response.close()
|
|
except (OSError, HTTPException):
|
|
pass
|
|
try:
|
|
value = json.loads(payload.decode("utf-8-sig"))
|
|
except (UnicodeDecodeError, json.JSONDecodeError) as exc:
|
|
raise DepotError(f"invalid JSON from {url}") from exc
|
|
if not isinstance(value, dict):
|
|
raise DepotError(f"unexpected JSON root from {url}")
|
|
return value
|
|
|
|
|
|
def plan_update(
|
|
game_directory: Path,
|
|
snapshot: DepotSnapshot,
|
|
*,
|
|
repair: bool = False,
|
|
) -> UpdatePlan:
|
|
"""Inspect a local game tree without writing to it."""
|
|
|
|
root = _prepare_game_root(game_directory, create=False)
|
|
downloads: list[PlannedDownload] = []
|
|
for file in snapshot.files:
|
|
target = _safe_target(root, file.path)
|
|
if target.is_symlink():
|
|
downloads.append(PlannedDownload(file, "symbolic link"))
|
|
continue
|
|
if not target.is_file():
|
|
downloads.append(PlannedDownload(file, "missing"))
|
|
continue
|
|
try:
|
|
if target.stat().st_size != file.size:
|
|
downloads.append(PlannedDownload(file, "size mismatch"))
|
|
continue
|
|
except OSError as exc:
|
|
raise DepotError(f"cannot inspect {file.path}: {exc}") from exc
|
|
if repair or file.build == snapshot.build:
|
|
if _sha1(target).casefold() != file.sha1.casefold():
|
|
downloads.append(PlannedDownload(file, "hash mismatch"))
|
|
|
|
deletes: list[str] = []
|
|
for relative_path in snapshot.deleted_files:
|
|
target = _safe_target(root, relative_path)
|
|
if target.is_symlink() or target.is_file():
|
|
deletes.append(relative_path)
|
|
elif target.exists():
|
|
raise DepotError(
|
|
f"refusing to delete non-file depot target: {relative_path}"
|
|
)
|
|
|
|
return UpdatePlan(
|
|
downloads=tuple(downloads),
|
|
deletes=tuple(deletes),
|
|
files_total=len(snapshot.files),
|
|
bytes_total=sum(item.file.size for item in downloads),
|
|
)
|
|
|
|
|
|
def sync_managed_game(
|
|
paths: AppPaths,
|
|
*,
|
|
repair: bool = False,
|
|
dry_run: bool = False,
|
|
progress: ProgressCallback | None = None,
|
|
client: DepotClient | None = None,
|
|
) -> DepotSyncResult:
|
|
"""Synchronize the managed game directory using only official public data."""
|
|
|
|
paths.create_base_directories()
|
|
depot_client = client or DepotClient()
|
|
_emit_progress(progress, "manifest", 0, 0, 0, 0, None)
|
|
snapshot = depot_client.fetch_snapshot()
|
|
plan = plan_update(paths.source_game, snapshot, repair=repair)
|
|
_emit_progress(
|
|
progress,
|
|
"planned",
|
|
0,
|
|
len(plan.downloads),
|
|
0,
|
|
plan.bytes_total,
|
|
None,
|
|
)
|
|
|
|
if dry_run:
|
|
return _sync_result(snapshot, plan, 0, 0, 0, repair, True)
|
|
|
|
root = _prepare_game_root(paths.source_game, create=True)
|
|
_download_plan(depot_client, snapshot, root, plan, progress)
|
|
deleted = _apply_deletions(root, plan.deletes)
|
|
created = _apply_create_if_missing(root, snapshot.create_if_missing)
|
|
result = _sync_result(
|
|
snapshot,
|
|
plan,
|
|
len(plan.downloads),
|
|
plan.bytes_total,
|
|
deleted,
|
|
repair,
|
|
False,
|
|
created=created,
|
|
)
|
|
_write_json(paths.native_depot_state, asdict(result))
|
|
_emit_progress(
|
|
progress,
|
|
"complete",
|
|
len(plan.downloads),
|
|
len(plan.downloads),
|
|
plan.bytes_total,
|
|
plan.bytes_total,
|
|
None,
|
|
)
|
|
return result
|
|
|
|
|
|
def _download_plan(
|
|
client: DepotClient,
|
|
depot_snapshot: DepotSnapshot,
|
|
root: Path,
|
|
plan: UpdatePlan,
|
|
progress: ProgressCallback | None,
|
|
) -> None:
|
|
if not plan.downloads:
|
|
return
|
|
|
|
lock = threading.Lock()
|
|
emit_lock = threading.Lock()
|
|
current_bytes: dict[str, int] = {}
|
|
completed_files = 0
|
|
completed_bytes = 0
|
|
transferred_bytes = 0
|
|
last_progress_emitted_at = 0.0
|
|
|
|
def emit_download_progress(
|
|
current_file: str,
|
|
*,
|
|
force: bool = False,
|
|
) -> None:
|
|
"""Throttle progress I/O without holding the download-state lock.
|
|
|
|
The workers can produce many chunk and file-completion updates per second.
|
|
The callback writes to a subprocess pipe for the native dashboard, so
|
|
retaining every event would make that pipe and its UI consumer the bottleneck.
|
|
"""
|
|
|
|
nonlocal last_progress_emitted_at
|
|
if progress is None:
|
|
return
|
|
with emit_lock:
|
|
now = time.monotonic()
|
|
if not force and now - last_progress_emitted_at < PROGRESS_UPDATE_INTERVAL_SECONDS:
|
|
return
|
|
last_progress_emitted_at = now
|
|
with lock:
|
|
event = SyncProgress(
|
|
"downloading",
|
|
completed_files,
|
|
len(plan.downloads),
|
|
completed_bytes + sum(current_bytes.values()),
|
|
plan.bytes_total,
|
|
current_file,
|
|
transferred_bytes,
|
|
)
|
|
progress(event)
|
|
|
|
def download(planned_download: PlannedDownload) -> None:
|
|
target = _safe_target(root, planned_download.file.path)
|
|
_ensure_target_parent(root, target)
|
|
|
|
def report_file_bytes(written: int) -> None:
|
|
nonlocal transferred_bytes
|
|
with lock:
|
|
previous = current_bytes.get(planned_download.file.path)
|
|
if previous is not None and written > previous:
|
|
transferred_bytes += written - previous
|
|
current_bytes[planned_download.file.path] = written
|
|
emit_download_progress(planned_download.file.path)
|
|
|
|
client.download_file(
|
|
depot_snapshot.server,
|
|
planned_download.file,
|
|
target,
|
|
progress=report_file_bytes if progress is not None else None,
|
|
)
|
|
|
|
worker_count = min(DOWNLOAD_WORKERS, len(plan.downloads))
|
|
remaining_items = iter(plan.downloads)
|
|
with ThreadPoolExecutor(max_workers=worker_count) as executor:
|
|
future_items: dict[Future[None], PlannedDownload] = {}
|
|
|
|
def submit_next() -> bool:
|
|
try:
|
|
next_item = next(remaining_items)
|
|
except StopIteration:
|
|
return False
|
|
future_items[executor.submit(download, next_item)] = next_item
|
|
return True
|
|
|
|
try:
|
|
for _worker in range(worker_count):
|
|
if not submit_next():
|
|
break
|
|
while future_items:
|
|
done, _pending = wait(
|
|
tuple(future_items),
|
|
return_when=FIRST_COMPLETED,
|
|
)
|
|
completed_items: list[PlannedDownload] = []
|
|
for future in done:
|
|
completed_item = future_items.pop(future)
|
|
future.result()
|
|
completed_items.append(completed_item)
|
|
for completed_item in completed_items:
|
|
with lock:
|
|
completed_files += 1
|
|
completed_bytes += completed_item.file.size
|
|
current_bytes.pop(completed_item.file.path, None)
|
|
is_final_download = completed_files == len(plan.downloads)
|
|
emit_download_progress(
|
|
completed_item.file.path,
|
|
force=is_final_download,
|
|
)
|
|
for _item in completed_items:
|
|
submit_next()
|
|
except BaseException:
|
|
for future in future_items:
|
|
future.cancel()
|
|
raise
|
|
|
|
|
|
def _apply_deletions(root: Path, relative_paths: Iterable[str]) -> int:
|
|
deleted = 0
|
|
for relative_path in relative_paths:
|
|
target = _safe_target(root, relative_path)
|
|
if target.is_symlink() or target.is_file():
|
|
target.unlink()
|
|
deleted += 1
|
|
return deleted
|
|
|
|
|
|
def _apply_create_if_missing(
|
|
root: Path,
|
|
entries: Iterable[CreateIfMissing],
|
|
) -> int:
|
|
created = 0
|
|
for entry in entries:
|
|
target = _safe_target(root, entry.path)
|
|
if target.is_symlink():
|
|
raise DepotError(f"refusing to follow symbolic link for {entry.path}")
|
|
if target.exists():
|
|
continue
|
|
content = _decode_create_if_missing(entry)
|
|
_ensure_target_parent(root, target)
|
|
temporary = target.with_name(f".{target.name}.{uuid4().hex}.partial")
|
|
try:
|
|
with temporary.open("xb") as handle:
|
|
handle.write(content)
|
|
handle.flush()
|
|
os.fsync(handle.fileno())
|
|
temporary.replace(target)
|
|
except BaseException:
|
|
temporary.unlink(missing_ok=True)
|
|
raise
|
|
created += 1
|
|
return created
|
|
|
|
|
|
def _sync_result(
|
|
snapshot: DepotSnapshot,
|
|
plan: UpdatePlan,
|
|
files_downloaded: int,
|
|
bytes_downloaded: int,
|
|
files_deleted: int,
|
|
repair: bool,
|
|
dry_run: bool,
|
|
*,
|
|
created: int = 0,
|
|
) -> DepotSyncResult:
|
|
return DepotSyncResult(
|
|
region=snapshot.region,
|
|
server=snapshot.server,
|
|
name=snapshot.name,
|
|
build=snapshot.build,
|
|
version=snapshot.version,
|
|
files_total=plan.files_total,
|
|
files_downloaded=files_downloaded,
|
|
bytes_downloaded=bytes_downloaded,
|
|
files_deleted=files_deleted,
|
|
files_created=created,
|
|
repair=repair,
|
|
dry_run=dry_run,
|
|
completed_at=datetime.now().astimezone().isoformat(),
|
|
)
|
|
|
|
|
|
def _parse_snapshot(
|
|
region: str,
|
|
server: str,
|
|
depot_manifest: Mapping[str, object],
|
|
build_manifest: Mapping[str, object],
|
|
) -> DepotSnapshot:
|
|
build = _positive_int(depot_manifest.get("Build"), "depot Build")
|
|
build_manifest_build = _positive_int(build_manifest.get("b"), "build manifest b")
|
|
if build_manifest_build != build:
|
|
raise DepotError(
|
|
f"depot build {build} does not match build manifest {build_manifest_build}"
|
|
)
|
|
name = _string(depot_manifest.get("Name"), "depot Name")
|
|
version = _string(depot_manifest.get("BuildVersionText"), "depot BuildVersionText")
|
|
files_value = build_manifest.get("f")
|
|
if not isinstance(files_value, list):
|
|
raise DepotError("build manifest f must be an array")
|
|
files: list[DepotFile] = []
|
|
seen_paths: set[str] = set()
|
|
for value in files_value:
|
|
if not isinstance(value, dict):
|
|
raise DepotError("build manifest contains a non-object file entry")
|
|
path = _normalise_relative_path(value.get("p"), "file path")
|
|
if path.casefold() in seen_paths:
|
|
raise DepotError(f"build manifest has duplicate file path: {path}")
|
|
seen_paths.add(path.casefold())
|
|
file_build = _positive_int(value.get("b"), f"file build for {path}")
|
|
size = _non_negative_int(value.get("s"), f"file size for {path}")
|
|
sha1 = _string(value.get("h"), f"file hash for {path}")
|
|
if not SHA1_PATTERN.fullmatch(sha1):
|
|
raise DepotError(f"invalid SHA-1 hash for {path}")
|
|
files.append(DepotFile(path, file_build, size, sha1.casefold()))
|
|
|
|
deleted = _parse_deleted_paths(build_manifest.get("df"), seen_paths)
|
|
create_if_missing = _parse_create_if_missing(depot_manifest.get("CreateIfMissing"))
|
|
return DepotSnapshot(
|
|
region=region,
|
|
server=server,
|
|
name=name,
|
|
build=build,
|
|
version=version,
|
|
files=tuple(files),
|
|
deleted_files=deleted,
|
|
create_if_missing=create_if_missing,
|
|
)
|
|
|
|
|
|
def _parse_deleted_paths(value: object, known_paths: set[str]) -> tuple[str, ...]:
|
|
if value is None:
|
|
return ()
|
|
if not isinstance(value, list):
|
|
raise DepotError("build manifest df must be an array")
|
|
paths: list[str] = []
|
|
seen: set[str] = set()
|
|
for item in value:
|
|
path = _normalise_relative_path(item, "deleted file path")
|
|
folded = path.casefold()
|
|
if folded in known_paths:
|
|
raise DepotError(f"manifest lists a file as both present and deleted: {path}")
|
|
if folded in seen:
|
|
continue
|
|
seen.add(folded)
|
|
paths.append(path)
|
|
return tuple(paths)
|
|
|
|
|
|
def _parse_create_if_missing(value: object) -> tuple[CreateIfMissing, ...]:
|
|
if value is None:
|
|
return ()
|
|
if not isinstance(value, list):
|
|
raise DepotError("CreateIfMissing must be an array")
|
|
entries: list[CreateIfMissing] = []
|
|
seen: set[str] = set()
|
|
for item in value:
|
|
if not isinstance(item, dict):
|
|
raise DepotError("CreateIfMissing contains a non-object entry")
|
|
path = _normalise_relative_path(item.get("Path"), "CreateIfMissing path")
|
|
folded = path.casefold()
|
|
if folded in seen:
|
|
raise DepotError(f"CreateIfMissing has duplicate path: {path}")
|
|
seen.add(folded)
|
|
content = _string(item.get("Content"), f"CreateIfMissing content for {path}")
|
|
entry = CreateIfMissing(path, content)
|
|
_decode_create_if_missing(entry)
|
|
entries.append(entry)
|
|
return tuple(entries)
|
|
|
|
|
|
def _decode_create_if_missing(entry: CreateIfMissing) -> bytes:
|
|
try:
|
|
content = b64decode(entry.content_base64, validate=True)
|
|
except (ValueError, BinAsciiError) as exc:
|
|
raise DepotError(f"invalid base64 create-if-missing entry: {entry.path}") from exc
|
|
if len(content) > MAX_CREATE_IF_MISSING_BYTES:
|
|
raise DepotError(f"create-if-missing entry is too large: {entry.path}")
|
|
return content
|
|
|
|
|
|
def _server_candidates(
|
|
configuration: Mapping[str, object] | None,
|
|
preferred_region: str,
|
|
) -> tuple[tuple[str, str], ...]:
|
|
entries: list[tuple[str, str]] = []
|
|
if configuration is not None:
|
|
regions = configuration.get("Regions")
|
|
default_region = configuration.get("DefaultRegion")
|
|
if isinstance(regions, list):
|
|
ordered = [preferred_region]
|
|
if isinstance(default_region, str) and default_region not in ordered:
|
|
ordered.append(default_region)
|
|
for region_name in ordered:
|
|
for region in regions:
|
|
if not isinstance(region, dict) or region.get("Name") != region_name:
|
|
continue
|
|
servers = region.get("Servers")
|
|
if isinstance(servers, list):
|
|
for server in servers:
|
|
if not isinstance(server, str):
|
|
continue
|
|
try:
|
|
entries.append((region_name, _normalise_server(server)))
|
|
except DepotError:
|
|
continue
|
|
seen: set[str] = set()
|
|
unique: list[tuple[str, str]] = []
|
|
for region, server in [*entries, *(("Fallback", value) for value in FALLBACK_SERVERS)]:
|
|
if server in seen:
|
|
continue
|
|
seen.add(server)
|
|
unique.append((region, server))
|
|
return tuple(unique)
|
|
|
|
|
|
def _normalise_server(value: str) -> str:
|
|
server = value.strip().strip("/")
|
|
if not server or "://" in server or any(character.isspace() for character in server):
|
|
raise DepotError(f"invalid CDN server: {value!r}")
|
|
parsed = urlparse(f"https://{server}")
|
|
if not parsed.netloc or parsed.query or parsed.fragment or "@" in parsed.netloc:
|
|
raise DepotError(f"invalid CDN server: {value!r}")
|
|
parts = PurePosixPath(parsed.path).parts
|
|
if any(part in (".", "..") for part in parts):
|
|
raise DepotError(f"invalid CDN server path: {value!r}")
|
|
return f"{parsed.netloc}{parsed.path.rstrip('/')}"
|
|
|
|
|
|
def _depot_manifest_url(server: str) -> str:
|
|
return f"https://{server}/depots/{DEPOT_ID}/manifest.json"
|
|
|
|
|
|
def _build_manifest_url(server: str, build: int) -> str:
|
|
return f"https://{server}/depots/{DEPOT_ID}/builds/{build}.json"
|
|
|
|
|
|
def _file_url(server: str, file: DepotFile) -> str:
|
|
return (
|
|
f"https://{server}/depots/{DEPOT_ID}/builds/{file.build}/"
|
|
f"{quote(file.path, safe='/')}"
|
|
)
|
|
|
|
|
|
def _ordered_download_servers(
|
|
primary: str,
|
|
candidates: Iterable[str],
|
|
) -> tuple[str, ...]:
|
|
ordered: list[str] = []
|
|
seen: set[str] = set()
|
|
for value in (primary, *candidates):
|
|
server = _normalise_server(value)
|
|
if server in seen:
|
|
continue
|
|
seen.add(server)
|
|
ordered.append(server)
|
|
return tuple(ordered)
|
|
|
|
|
|
def _download_partial_path(destination: Path) -> Path:
|
|
return destination.with_name(f".{destination.name}.ptlinux.partial")
|
|
|
|
|
|
def _recover_crlf_manifest_file(
|
|
temporary: Path,
|
|
destination: Path,
|
|
file: DepotFile,
|
|
) -> bool:
|
|
"""Restore publisher-normalized legacy text only on an exact hash match."""
|
|
|
|
try:
|
|
if temporary.stat().st_size > MAX_LINE_ENDING_RECOVERY_BYTES:
|
|
return False
|
|
data = temporary.read_bytes()
|
|
except OSError as exc:
|
|
raise DepotError(f"could not inspect downloaded {file.path}: {exc}") from exc
|
|
if b"\r" in data:
|
|
return False
|
|
recovered = data.replace(b"\n", b"\r\n")
|
|
if len(recovered) != file.size:
|
|
return False
|
|
if (
|
|
hashlib.sha1(recovered, usedforsecurity=False).hexdigest().casefold()
|
|
!= file.sha1.casefold()
|
|
):
|
|
return False
|
|
recovered_temporary = destination.with_name(
|
|
f".{destination.name}.{os.getpid()}.{uuid4().hex}.recovered"
|
|
)
|
|
try:
|
|
with recovered_temporary.open("xb") as handle:
|
|
handle.write(recovered)
|
|
handle.flush()
|
|
os.fsync(handle.fileno())
|
|
recovered_temporary.replace(destination)
|
|
except BaseException:
|
|
recovered_temporary.unlink(missing_ok=True)
|
|
raise
|
|
return True
|
|
|
|
|
|
def _prepare_game_root(game_directory: Path, *, create: bool) -> Path:
|
|
root = game_directory.expanduser()
|
|
if root.is_symlink():
|
|
raise DepotError(f"managed game directory must not be a symbolic link: {root}")
|
|
if root.exists() and not root.is_dir():
|
|
raise DepotError(f"managed game path is not a directory: {root}")
|
|
if create:
|
|
root.mkdir(parents=True, exist_ok=True)
|
|
return root.resolve(strict=False)
|
|
|
|
|
|
def _safe_target(root: Path, relative_path: str) -> Path:
|
|
parts = _relative_parts(relative_path, "depot path")
|
|
target = root.joinpath(*parts)
|
|
try:
|
|
target.resolve(strict=False).relative_to(root)
|
|
except ValueError as exc:
|
|
raise DepotError(f"depot path escapes the managed game directory: {relative_path}") from exc
|
|
return target
|
|
|
|
|
|
def _ensure_target_parent(root: Path, target: Path) -> None:
|
|
try:
|
|
relative_parent = target.parent.relative_to(root)
|
|
except ValueError as exc:
|
|
raise DepotError(f"target escapes managed game directory: {target}") from exc
|
|
cursor = root
|
|
for part in relative_parent.parts:
|
|
cursor = cursor / part
|
|
if cursor.is_symlink():
|
|
raise DepotError(f"refusing to traverse symbolic link in game tree: {cursor}")
|
|
if cursor.exists():
|
|
if not cursor.is_dir():
|
|
raise DepotError(f"cannot create game directory over file: {cursor}")
|
|
else:
|
|
try:
|
|
cursor.mkdir()
|
|
except FileExistsError:
|
|
# Another download worker can create a shared parent after
|
|
# ``exists()`` and before ``mkdir()``. Accept only the directory
|
|
# that the losing worker expected; never accept a file or symlink.
|
|
if cursor.is_symlink():
|
|
raise DepotError(
|
|
f"refusing to traverse symbolic link in game tree: {cursor}"
|
|
)
|
|
if not cursor.is_dir():
|
|
raise DepotError(f"cannot create game directory over file: {cursor}")
|
|
except OSError as exc:
|
|
raise DepotError(f"cannot create game directory {cursor}: {exc}") from exc
|
|
|
|
|
|
def _normalise_relative_path(value: object, field: str) -> str:
|
|
if not isinstance(value, str):
|
|
raise DepotError(f"{field} must be a string")
|
|
parts = _relative_parts(value, field)
|
|
return "/".join(parts)
|
|
|
|
|
|
def _relative_parts(value: str, field: str) -> tuple[str, ...]:
|
|
if not value or "\x00" in value:
|
|
raise DepotError(f"invalid {field}")
|
|
normalised = value.replace("\\", "/")
|
|
if normalised.startswith("/") or normalised.startswith("//"):
|
|
raise DepotError(f"{field} must be relative: {value!r}")
|
|
parts = PurePosixPath(normalised).parts
|
|
if not parts or any(part in ("", ".", "..") or ":" in part for part in parts):
|
|
raise DepotError(f"invalid {field}: {value!r}")
|
|
return tuple(parts)
|
|
|
|
|
|
def _positive_int(value: object, field: str) -> int:
|
|
if isinstance(value, bool) or not isinstance(value, int) or value <= 0:
|
|
raise DepotError(f"{field} must be a positive integer")
|
|
return value
|
|
|
|
|
|
def _non_negative_int(value: object, field: str) -> int:
|
|
if isinstance(value, bool) or not isinstance(value, int) or value < 0:
|
|
raise DepotError(f"{field} must be a non-negative integer")
|
|
return value
|
|
|
|
|
|
def _string(value: object, field: str) -> str:
|
|
if not isinstance(value, str) or not value.strip():
|
|
raise DepotError(f"{field} must be a non-empty string")
|
|
return value
|
|
|
|
|
|
def _sha1(path: Path) -> str:
|
|
digest = hashlib.sha1(usedforsecurity=False)
|
|
try:
|
|
with path.open("rb") as handle:
|
|
for chunk in iter(lambda: handle.read(HASH_READ_SIZE), b""):
|
|
digest.update(chunk)
|
|
except OSError as exc:
|
|
raise DepotError(f"cannot hash {path}: {exc}") from exc
|
|
return digest.hexdigest()
|
|
|
|
|
|
def _content_length(value: str | None) -> int | None:
|
|
if value is None:
|
|
return None
|
|
try:
|
|
length = int(value)
|
|
except ValueError as exc:
|
|
raise DepotError(f"invalid Content-Length: {value!r}") from exc
|
|
if length < 0:
|
|
raise DepotError(f"invalid Content-Length: {value!r}")
|
|
return length
|
|
|
|
|
|
def _content_range(value: str | None) -> tuple[int, int, int]:
|
|
if value is None:
|
|
raise DepotError("206 response omitted Content-Range")
|
|
match = CONTENT_RANGE_PATTERN.fullmatch(value.strip())
|
|
if match is None:
|
|
raise DepotError(f"invalid Content-Range: {value!r}")
|
|
start, end, total = (int(part) for part in match.groups())
|
|
if start > end or end >= total:
|
|
raise DepotError(f"invalid Content-Range: {value!r}")
|
|
return start, end, total
|
|
|
|
|
|
def _headers() -> dict[str, str]:
|
|
return {"User-Agent": f"PT-Linux/{__version__} (+https://pristontale.eu/)"}
|
|
|
|
|
|
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()}.tmp")
|
|
try:
|
|
temporary.write_text(
|
|
json.dumps(value, indent=2, sort_keys=True) + "\n",
|
|
encoding="utf-8",
|
|
)
|
|
temporary.replace(path)
|
|
except BaseException:
|
|
temporary.unlink(missing_ok=True)
|
|
raise
|
|
|
|
|
|
def _emit_progress(
|
|
callback: ProgressCallback | None,
|
|
phase: str,
|
|
files_completed: int,
|
|
files_total: int,
|
|
bytes_completed: int,
|
|
bytes_total: int,
|
|
current_file: str | None,
|
|
) -> None:
|
|
if callback is not None:
|
|
callback(
|
|
SyncProgress(
|
|
phase,
|
|
files_completed,
|
|
files_total,
|
|
bytes_completed,
|
|
bytes_total,
|
|
current_file,
|
|
)
|
|
)
|