"""Load the permanent runtime-resource selection.""" from __future__ import annotations import json import os import shutil import urllib.parse from pathlib import Path from common import ROOT, ProbeError, safe_relative, sha256, write_json RUNTIME_RESOURCE_ROOT = ROOT / 'private-assets' RUNTIME_RESOURCE_MANIFEST = ROOT / 'config/runtime-resources.json' _SCHEMA = 'kam-runtime-resources/1' _KEYS = {'schema', 'entries'} _ENTRY_KEYS = {'kind', 'path'} def _load_runtime_resource_manifest() -> tuple[dict[str, str], ...]: """Return the validated authored resource selection, without resolving it.""" try: value = json.loads(RUNTIME_RESOURCE_MANIFEST.read_text(encoding='utf-8')) except (OSError, UnicodeError, json.JSONDecodeError) as error: raise ProbeError(f'Cannot read runtime-resource manifest: {error}') from error if not isinstance(value, dict) or set(value) != _KEYS: raise ProbeError('Invalid runtime-resource manifest fields') if value['schema'] != _SCHEMA: raise ProbeError('Invalid runtime-resource manifest schema') if not isinstance(value['entries'], list) or not value['entries']: raise ProbeError('Runtime-resource entries must be a non-empty array') seen_paths: set[str] = set() entries = [] for entry in value['entries']: if not isinstance(entry, dict) or set(entry) != _ENTRY_KEYS: raise ProbeError('Invalid runtime-resource entry fields') kind, path = entry['kind'], entry['path'] if not isinstance(kind, str) or kind not in {'file', 'directory'} or not isinstance(path, str) or not path: raise ProbeError('Invalid runtime-resource entry') safe_relative(path) if path in seen_paths: raise ProbeError(f'Duplicate runtime-resource path: {path!r}') seen_paths.add(path) entries.append({'kind': kind, 'path': path}) return tuple(entries) def _resolve_runtime_resource_files(entries: tuple[dict[str, str], ...]) -> tuple[str, ...]: """Expand parsed resource entries into exact, safe file identities.""" root = RUNTIME_RESOURCE_ROOT if root.is_symlink() or not root.is_dir(): raise ProbeError(f'Runtime-resource root is not a real directory: {root}') try: fixed_root = root.resolve(strict=True) except OSError as error: raise ProbeError(f'Cannot resolve runtime-resource root: {error}') from error def checked(path): if path.is_symlink(): raise ProbeError(f'Runtime-resource symlink is not allowed: {path}') try: resolved = path.resolve(strict=True) except OSError as error: raise ProbeError(f'Cannot resolve runtime resource: {path}') from error if not resolved.is_relative_to(fixed_root): raise ProbeError(f'Runtime resource escapes fixed root: {path}') return resolved def exact(path: str): current = root for component in path.split('/'): if not current.is_dir(): raise ProbeError(f'Runtime-resource path component is not a directory: {current}') try: with os.scandir(current) as scanned: child = next((Path(entry.path) for entry in scanned if entry.name == component), None) except OSError as error: raise ProbeError(f'Cannot inspect runtime-resource path: {current}') from error if child is None: raise ProbeError(f'Runtime-resource path not found: {path}') checked(child) current = child return current selected: set[str] = set() for entry in entries: kind, authored = entry['kind'], entry['path'] target = exact(authored) if kind == 'file': if target.name == '.gitkeep' or not target.is_file() or target.is_symlink(): raise ProbeError(f'Runtime-resource file is not a regular selectable file: {authored}') selected.add(authored) continue if not target.is_dir() or target.is_symlink(): raise ProbeError(f'Runtime-resource directory is not a real directory: {authored}') produced = False pending = [target] while pending: directory = pending.pop() try: children = list(os.scandir(directory)) except OSError as error: raise ProbeError(f'Cannot inspect runtime-resource directory: {directory}') from error for child_entry in children: child = Path(child_entry.path) checked(child) if child_entry.is_symlink(): raise ProbeError(f'Runtime-resource symlink is not allowed: {child}') if child_entry.is_dir(follow_symlinks=False): pending.append(child) elif child_entry.is_file(follow_symlinks=False): if child.name != '.gitkeep': produced = True selected.add(child.relative_to(root).as_posix()) else: raise ProbeError(f'Runtime-resource node is not a regular file or directory: {child}') if not produced: raise ProbeError(f'Runtime-resource directory is empty: {authored}') folded: dict[str, str] = {} for path in selected: prior = folded.setdefault(path.casefold(), path) if prior != path: raise ProbeError(f'Case-colliding runtime-resource paths: {prior!r}, {path!r}') return tuple(sorted(selected, key=lambda path: (path.casefold(), path))) def stage_runtime_resources(castle_data_root: Path) -> tuple[dict[str, object], ...]: """Stage the selected runtime resources into a Castle data directory.""" paths = _resolve_runtime_resource_files(_load_runtime_resource_manifest()) records = [] for path in paths: source = RUNTIME_RESOURCE_ROOT / path if source.is_symlink() or not source.is_file(): raise ProbeError(f'Runtime-resource file disappeared or changed type: {path}') try: digest = sha256(source) records.append({'path': path, 'size': source.stat().st_size, 'sha256': digest, 'url': 'castle-data:/game/' + urllib.parse.quote(path, safe='/')}) except OSError as error: raise ProbeError(f'Cannot inventory runtime resource: {path}: {error}') from error game = castle_data_root / 'game' manifest = castle_data_root / 'assets-manifest.json' try: if game.is_symlink() or game.is_file(): game.unlink() elif game.exists(): shutil.rmtree(game) if manifest.exists() or manifest.is_symlink(): manifest.unlink() for record in records: path = record['path'] source = RUNTIME_RESOURCE_ROOT / path if source.is_symlink() or not source.is_file(): raise ProbeError(f'Runtime-resource file disappeared or changed type: {path}') destination = game / path destination.parent.mkdir(parents=True, exist_ok=True) shutil.copyfile(source, destination) if sha256(destination) != record['sha256']: raise ProbeError(f'Runtime-resource copy hash mismatch: {path}') write_json(manifest, {'schema': 'kam-runtime-assets/1', 'files': records}) except ProbeError: shutil.rmtree(game, ignore_errors=True) try: manifest.unlink(missing_ok=True) except OSError: pass raise except OSError as error: shutil.rmtree(game, ignore_errors=True) try: manifest.unlink(missing_ok=True) except OSError: pass raise ProbeError(f'Cannot stage runtime resources: {error}') from error return tuple(records)