Cleanup and security updates.

ptlinux/depot.py

@@ -13,6 +13,7 @@ import json
 import os
 from pathlib import Path, PurePosixPath
 import re
+import stat
 import threading
 import time
 from typing import Callable, Iterable, Mapping, Protocol
@@ -38,6 +39,35 @@ 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
+UNMANAGED_EXECUTABLE_SUFFIXES = frozenset(
+	{
+		".asi",
+		".bat",
+		".cmd",
+		".com",
+		".cpl",
+		".desktop",
+		".dll",
+		".drv",
+		".exe",
+		".jar",
+		".js",
+		".lnk",
+		".lua",
+		".msi",
+		".ocx",
+		".ps1",
+		".py",
+		".reg",
+		".scr",
+		".sh",
+		".so",
+		".url",
+		".vbe",
+		".vbs",
+		".wsf",
+	}
+)
 ALWAYS_VERIFY_PATHS = frozenset(
 	{"game.exe", "game.dll", "gamese.exe", "gamese.dll"}
 )
@@ -132,11 +162,21 @@ class DepotSyncResult:
 	bytes_downloaded: int
 	files_deleted: int
 	files_created: int
+	files_quarantined: int
+	quarantine_directory: str | None
 	repair: bool
 	dry_run: bool
 	completed_at: str
 
 
+@dataclass(frozen=True)
+class QuarantineResult:
+	"""Recoverable unmanaged entries removed before authenticating an import."""
+
+	files: int
+	directory: str | None
+
+
 ProgressCallback = Callable[[SyncProgress], None]
 
 
@@ -486,6 +526,7 @@ def sync_managed_game(
 	paths: AppPaths,
 	*,
 	repair: bool = False,
+	quarantine_unmanaged: bool = False,
 	dry_run: bool = False,
 	progress: ProgressCallback | None = None,
 	client: DepotClient | None = None,
@@ -496,6 +537,9 @@ def sync_managed_game(
 	depot_client = client or DepotClient()
 	_emit_progress(progress, "manifest", 0, 0, 0, 0, None)
 	snapshot = depot_client.fetch_snapshot()
+	quarantine = QuarantineResult(0, None)
+	if quarantine_unmanaged and not dry_run:
+		quarantine = _quarantine_unmanaged_files(paths, snapshot)
 	plan = plan_update(paths.source_game, snapshot, repair=repair)
 	_emit_progress(
 		progress,
@@ -508,7 +552,16 @@ def sync_managed_game(
 	)
 
 	if dry_run:
-		return _sync_result(snapshot, plan, 0, 0, 0, repair, True)
+		return _sync_result(
+			snapshot,
+			plan,
+			0,
+			0,
+			0,
+			repair,
+			True,
+			quarantine=quarantine,
+		)
 
 	root = _prepare_game_root(paths.source_game, create=True)
 	_download_plan(depot_client, snapshot, root, plan, progress)
@@ -523,6 +576,7 @@ def sync_managed_game(
 		repair,
 		False,
 		created=created,
+		quarantine=quarantine,
 	)
 	_write_json(paths.native_depot_state, asdict(result))
 	_emit_progress(
@@ -537,6 +591,110 @@ def sync_managed_game(
 	return result
 
 
+def _quarantine_unmanaged_files(
+	paths: AppPaths,
+	snapshot: DepotSnapshot,
+) -> QuarantineResult:
+	"""Move non-depot import contents aside without following links."""
+
+	root = _prepare_game_root(paths.source_game, create=False)
+	if not root.exists():
+		return QuarantineResult(0, None)
+	allowed = {
+		file.path for file in snapshot.files
+	} | {
+		entry.path for entry in snapshot.create_if_missing
+	}
+	candidates: list[tuple[Path, str]] = []
+
+	def walk_error(error: OSError) -> None:
+		raise DepotError(f"cannot inspect imported game tree: {error}") from error
+
+	for directory, directory_names, file_names in os.walk(
+		root,
+		topdown=True,
+		followlinks=False,
+		onerror=walk_error,
+	):
+		parent = Path(directory)
+		for name in tuple(directory_names):
+			path = parent / name
+			try:
+				status = path.lstat()
+			except OSError as exc:
+				raise DepotError(f"cannot inspect imported entry {path}: {exc}") from exc
+			if stat.S_ISDIR(status.st_mode):
+				continue
+			directory_names.remove(name)
+			candidates.append((path, path.relative_to(root).as_posix()))
+		for name in file_names:
+			path = parent / name
+			relative = path.relative_to(root).as_posix()
+			try:
+				status = path.lstat()
+			except OSError as exc:
+				raise DepotError(f"cannot inspect imported entry {path}: {exc}") from exc
+			if not stat.S_ISREG(status.st_mode):
+				candidates.append((path, relative))
+				continue
+			if relative in allowed:
+				continue
+			if _safe_imported_user_file(path, relative):
+				continue
+			candidates.append((path, relative))
+
+	if not candidates:
+		return QuarantineResult(0, None)
+	base = paths.game_import_quarantine.resolve(strict=False)
+	data_root = paths.data.resolve(strict=False)
+	try:
+		base.relative_to(data_root)
+	except ValueError as exc:
+		raise DepotError(f"import quarantine is outside the data root: {base}") from exc
+	stamp = datetime.now().astimezone().strftime("%Y%m%d-%H%M%S")
+	quarantine = base / f"{stamp}-{uuid4().hex}"
+	files_root = quarantine / "files"
+	files_root.mkdir(parents=True, exist_ok=False)
+	manifest = {
+		"schema": 1,
+		"complete": False,
+		"created_at": datetime.now().astimezone().isoformat(),
+		"source": str(root),
+		"files": [relative for _path, relative in candidates],
+	}
+	_write_json(quarantine / "quarantine.json", manifest)
+	for source, relative in candidates:
+		destination = files_root.joinpath(*PurePosixPath(relative).parts)
+		destination.parent.mkdir(parents=True, exist_ok=True)
+		try:
+			source.replace(destination)
+		except OSError as exc:
+			raise DepotError(
+				f"could not quarantine imported entry {relative}: {exc}"
+			) from exc
+	manifest["complete"] = True
+	_write_json(quarantine / "quarantine.json", manifest)
+	return QuarantineResult(len(candidates), str(quarantine))
+
+
+def _safe_imported_user_file(path: Path, relative: str) -> bool:
+	parts = PurePosixPath(relative).parts
+	if not parts or parts[0].casefold() != "save":
+		return False
+	if path.suffix.casefold() in UNMANAGED_EXECUTABLE_SUFFIXES:
+		return False
+	try:
+		with path.open("rb") as handle:
+			prefix = handle.read(4)
+	except OSError as exc:
+		raise DepotError(f"cannot inspect imported user file {path}: {exc}") from exc
+	return not (
+		prefix.startswith(b"MZ")
+		or prefix.startswith(b"\x7fELF")
+		or prefix.startswith(b"#!")
+	)
+
+
 def _download_plan(
 	client: DepotClient,
 	depot_snapshot: DepotSnapshot,
@@ -699,6 +857,7 @@ def _sync_result(
 	dry_run: bool,
 	*,
 	created: int = 0,
+	quarantine: QuarantineResult = QuarantineResult(0, None),
 ) -> DepotSyncResult:
 	return DepotSyncResult(
 		region=snapshot.region,
@@ -711,6 +870,8 @@ def _sync_result(
 		bytes_downloaded=bytes_downloaded,
 		files_deleted=files_deleted,
 		files_created=created,
+		files_quarantined=quarantine.files,
+		quarantine_directory=quarantine.directory,
 		repair=repair,
 		dry_run=dry_run,
 		completed_at=datetime.now().astimezone().isoformat(),