179 lines
7.8 KiB
Python
179 lines
7.8 KiB
Python
|
"""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)
|