Unify Iron Coast mission names and cinematic mission previews

tools/assemble_runtime.py

@@ -28,6 +28,90 @@ def checked_file(root, relative):
     return resolved
 
 
+def default_data_source(package):
+    """The portable package is authoritative when it has its own catalog."""
+    bundled = package / "data"
+    return bundled if (bundled / "INDEX").is_file() else package.parent / "data"
+
+
+def checked_tree(root):
+    """Select a complete contained file tree; never silently follow/skip symlink directories."""
+    root = root.resolve(strict=True)
+    if not root.is_dir():
+        raise ValueError(f"not a runtime data directory: {root}")
+    def raise_walk_error(error):
+        raise error
+
+    files = {}
+    for directory, directories, names in os.walk(root, followlinks=False, onerror=raise_walk_error):
+        parent = Path(directory)
+        for name in directories:
+            if (parent / name).is_symlink():
+                raise ValueError(f"symlink directory in runtime data: {parent / name}")
+        for name in names:
+            relative = (parent / name).relative_to(root).as_posix()
+            files[relative] = checked_file(root, relative)
+    return files
+
+
+def select_data_files(package, source=None, hd_source=None):
+    """Validate and select the complete runtime data without copying or mutating it."""
+    source = (source or default_data_source(package)).resolve(strict=True)
+    files = checked_tree(source)
+    entries = checked_file(source, "INDEX").read_text().splitlines()
+    catalog = {"INDEX"}
+    for row in entries:
+        if not row.strip():
+            continue
+        _, length, name = row.split()
+        original = checked_file(source, name)
+        if original.stat().st_size != int(length):
+            raise ValueError(f"catalog size mismatch: {name}")
+        catalog.add(Path(name).as_posix())
+
+    if hd_source is not None:
+        hd_source = hd_source.resolve(strict=True)
+        replacement = checked_tree(hd_source)
+        checked_file(hd_source, "hd.json")
+        if any(Path(name).parts[0] == "hd" for name in catalog):
+            raise ValueError("HD pack paths conflict with the original catalog")
+        files = {name: path for name, path in files.items() if Path(name).parts[0] != "hd"}
+        files.update({"hd/" + name: path for name, path in replacement.items()})
+
+    hd_frames = 0
+    if "hd/hd.json" in files:
+        hd = json.loads(files["hd/hd.json"].read_bytes())
+        hd_root = hd_source if hd_source is not None else source / "hd"
+        for sequence in hd["sequences"]:
+            for frame in sequence["frames"]:
+                payload = checked_file(hd_root, frame["file"])
+                name = "hd/" + Path(frame["file"]).as_posix()
+                if name not in files or files[name] != payload:
+                    raise ValueError(f"HD frame is missing from runtime selection: {name}")
+                if name in catalog:
+                    raise ValueError("HD pack paths conflict with the original catalog")
+                hd_frames += 1
+
+    # Original missions register these 27 SD/HD pairs in src/content.rs.
+    # An external legacy catalog need not carry new menu art; the bundled art
+    # is the only fallback, and existing external briefing files keep priority.
+    bundled = (package / "data").resolve()
+    required_briefings = [
+        f"briefings/sector_{sector}_wave_{wave}_{mode}.png"
+        for sector in range(1, 4) for wave in range(1, 10) for mode in ("sd", "hd")
+    ]
+    if source != bundled and any(name not in files for name in required_briefings):
+        briefings = bundled / "briefings"
+        if briefings.is_symlink():
+            raise ValueError(f"symlink directory in bundled mission briefings: {briefings}")
+        for name, path in checked_tree(briefings).items():
+            files.setdefault("briefings/" + name, path)
+    for name in required_briefings:
+        if name not in files:
+            raise ValueError(f"required mission briefing is missing: {name}")
+    return files, entries, hd_frames
+
+
 def iter_archive_files(package, archive, skipped, excluded_files=(), required_files=()):
     """Keep manifest payloads even when a parent has a build-cache name."""
     required = {Path(name) for name in required_files}
@@ -69,9 +153,6 @@ def main():
     parser.add_argument("--zip", type=Path, help="export runtime, source, docs and mods; exclude build caches")
     args = parser.parse_args()
     package = Path(__file__).resolve().parents[1]
-    working_data = package.parent / "data"
-    default_data = working_data if (working_data / "INDEX").is_file() else package / "data"
-    source = (args.source_data or default_data).resolve()
     environment = os.environ.copy()
     local_lib = Path.home() / ".local/lib"
     environment["LD_LIBRARY_PATH"] = str(local_lib) + ":" + environment.get("LD_LIBRARY_PATH", "")
@@ -86,49 +167,14 @@ def main():
     if not executable.is_file():
         raise ValueError("release executable is missing; run without --skip-build")
 
-    files = {"INDEX"}
-    entries = (source / "INDEX").read_text().splitlines()
-    for row in entries:
-        if not row.strip():
-            continue
-        _, length, name = row.split()
-        original = checked_file(source, name)
-        if original.stat().st_size != int(length):
-            raise ValueError(f"catalog size mismatch: {name}")
-        files.add(name)
-    hd_source = (args.hd_source or source / "hd").resolve()
-    hd_path = hd_source / "hd.json"
-    if args.hd_source and not hd_path.is_file():
-        raise ValueError(f"approved HD manifest is missing: {hd_path}")
-    hd_bytes = hd_path.read_bytes() if hd_path.is_file() else None
-    hd_frames = 0
-    hd_files = set()
-    if hd_bytes:
-        hd = json.loads(hd_bytes)
-        hd_files.add("hd/hd.json")
-        if (hd_source / "provenance.json").is_file():
-            checked_file(hd_source, "provenance.json")
-            hd_files.add("hd/provenance.json")
-        for sequence in hd["sequences"]:
-            for frame in sequence["frames"]:
-                checked_file(hd_source, frame["file"])
-                hd_files.add("hd/" + frame["file"])
-                hd_frames += 1
-        if files.intersection(hd_files):
-            raise ValueError("HD pack paths conflict with the original catalog")
-        files.update(hd_files)
+    files, entries, hd_frames = select_data_files(package, args.source_data, args.hd_source)
 
     staging = Path(tempfile.mkdtemp(prefix=".runtime-data-", dir=package))
     try:
         for name in sorted(files):
             destination = staging / name
             destination.parent.mkdir(parents=True, exist_ok=True)
-            if name == "hd/hd.json":
-                destination.write_bytes(hd_bytes)
-            elif name in hd_files:
-                shutil.copy2(checked_file(hd_source, name.removeprefix("hd/")), destination)
-            else:
-                shutil.copy2(checked_file(source, name), destination)
+            shutil.copy2(files[name], destination)
         previous = package / ".runtime-data-previous"
         if previous.exists():
             raise ValueError("previous interrupted package exists; preserve/recover .runtime-data-previous before retrying")

tools/test_runtime_data_selection.py

@@ -0,0 +1,179 @@
+"""Synthetic packing checks; no production copies or compiler required."""
+import json
+import os
+from pathlib import Path
+import shutil
+import tempfile
+import unittest
+from unittest.mock import patch
+
+from assemble_runtime import default_data_source, select_data_files
+
+
+class RuntimeDataSelectionTests(unittest.TestCase):
+    def setUp(self):
+        self.temporary = tempfile.TemporaryDirectory()
+        self.addCleanup(self.temporary.cleanup)
+        self.root = Path(self.temporary.name)
+        self.package = self.root / "RustRaptor"
+        self.data = self.package / "data"
+        self.catalog(self.data)
+        self.briefings(self.data)
+
+    def write(self, root, name, payload):
+        path = root / name
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_bytes(payload)
+        return path
+
+    def catalog(self, root):
+        self.write(root, "INDEX", b"00001 4 sprites/a.bin\n")
+        self.write(root, "sprites/a.bin", b"base")
+
+    def briefings(self, root):
+        for sector in range(1, 4):
+            for wave in range(1, 10):
+                for mode in ("sd", "hd"):
+                    self.write(root, f"briefings/sector_{sector}_wave_{wave}_{mode}.png",
+                               f"briefing {sector}/{wave}/{mode}".encode())
+
+    def hd(self, root, name="frames/new.png", payload=b"new frame"):
+        self.write(root, name, payload)
+        self.write(root, "hd.json", json.dumps({"sequences": [
+            {"frames": [{"file": name}]}]}).encode())
+
+    def test_bundled_catalog_wins_over_parent_legacy_catalog(self):
+        external = self.root / "data"
+        self.catalog(external)
+        self.write(external, "legacy-only", b"obsolete")
+        self.assertEqual(default_data_source(self.package), self.data)
+        files, entries, count = select_data_files(self.package)
+        self.assertNotIn("legacy-only", files)
+        self.assertEqual((len(entries), count), (1, 0))
+        (self.data / "INDEX").unlink()
+        self.assertEqual(default_data_source(self.package), external)
+
+    def test_complete_tree_roundtrips_briefings_provenance_and_hd_jobs(self):
+        self.hd(self.data / "hd")
+        extras = {
+            "briefings/provenance.json": b"briefing provenance",
+            "hd/provenance.json": b"HD lineage",
+            "hd/jobs/family/derived/accepted.png": b"derived payload",
+            "hd/jobs/family/raw/source.png": b"raw lineage",
+            "hd/jobs/family/recipe.json": b"recipe",
+            "misc/runtime-extra": b"extra data",
+        }
+        for name, payload in extras.items():
+            self.write(self.data, name, payload)
+        files, _, count = select_data_files(self.package)
+        self.assertEqual(count, 1)
+        self.assertEqual(len([n for n in files if n.startswith("briefings/") and n.endswith(".png")]), 54)
+        self.assertTrue(set(extras) <= set(files))
+        relocated = self.root / "Portable Runtime Ω" / "data"
+        for name, path in files.items():
+            destination = relocated / name
+            destination.parent.mkdir(parents=True, exist_ok=True)
+            shutil.copy2(path, destination)
+        self.assertEqual({p.relative_to(self.data).as_posix(): p.read_bytes()
+                          for p in self.data.rglob("*") if p.is_file()},
+                         {p.relative_to(relocated).as_posix(): p.read_bytes()
+                          for p in relocated.rglob("*") if p.is_file()})
+
+    def test_external_catalog_fallback_preserves_existing_briefing(self):
+        external = self.root / "external"
+        self.catalog(external)
+        override = "briefings/sector_1_wave_1_sd.png"
+        self.write(external, override, b"external briefing")
+        self.write(external, "provenance.json", b"external metadata")
+        self.write(self.data, "briefings/provenance.json", b"bundled briefing lineage")
+        files, _, _ = select_data_files(self.package, external)
+        self.assertEqual(files[override].read_bytes(), b"external briefing")
+        self.assertEqual(files["briefings/sector_3_wave_9_hd.png"].read_bytes(), b"briefing 3/9/hd")
+        self.assertEqual(files["provenance.json"].read_bytes(), b"external metadata")
+        self.assertEqual(len([n for n in files if n.startswith("briefings/") and n.endswith(".png")]), 54)
+        self.assertEqual(files["briefings/provenance.json"].read_bytes(), b"bundled briefing lineage")
+
+    def test_explicit_hd_replaces_entire_old_subtree(self):
+        self.hd(self.data / "hd", "old.png", b"obsolete frame")
+        self.write(self.data, "hd/jobs/old/receipt.json", b"obsolete metadata")
+        replacement = self.root / "approved-HD"
+        self.hd(replacement)
+        self.write(replacement, "jobs/new/raw/source.png", b"new lineage")
+        self.write(replacement, "provenance.json", b"new provenance")
+        files, _, count = select_data_files(self.package, hd_source=replacement)
+        self.assertEqual(count, 1)
+        self.assertEqual({n for n in files if n.startswith("hd/")},
+                         {"hd/hd.json", "hd/frames/new.png", "hd/jobs/new/raw/source.png", "hd/provenance.json"})
+        self.assertEqual(files["hd/provenance.json"].read_bytes(), b"new provenance")
+
+    def test_missing_required_briefing_or_explicit_manifest_fails(self):
+        (self.data / "briefings/sector_3_wave_9_hd.png").unlink()
+        with self.assertRaisesRegex(ValueError, "required mission briefing"):
+            select_data_files(self.package)
+        self.briefings(self.data)
+        empty = self.root / "empty-hd"
+        empty.mkdir()
+        with self.assertRaises(FileNotFoundError):
+            select_data_files(self.package, hd_source=empty)
+
+    def test_unlisted_file_and_directory_symlinks_cannot_escape_source(self):
+        outside = self.write(self.root, "outside", b"private")
+        link = self.data / "unlisted-link"
+        link.symlink_to(outside)
+        with self.assertRaises(ValueError):
+            select_data_files(self.package)
+        link.unlink()
+        link.symlink_to(self.root, target_is_directory=True)
+        with self.assertRaisesRegex(ValueError, "symlink directory"):
+            select_data_files(self.package)
+
+    def test_catalog_and_hd_manifest_traversal_and_size_are_checked(self):
+        for catalog in (b"00001 4 ../outside\n", b"00001 999 sprites/a.bin\n"):
+            self.write(self.data, "INDEX", catalog)
+            with self.subTest(catalog=catalog), self.assertRaises(ValueError):
+                select_data_files(self.package)
+        self.catalog(self.data)
+        replacement = self.root / "unsafe-HD"
+        self.hd(replacement, "../outside", b"private")
+        with self.assertRaises(ValueError):
+            select_data_files(self.package, hd_source=replacement)
+
+    def test_external_fallback_briefing_symlink_cannot_escape_bundled_data(self):
+        external = self.root / "external"
+        self.catalog(external)
+        missing = self.data / "briefings/sector_1_wave_1_sd.png"
+        missing.unlink()
+        missing.symlink_to(self.write(self.root, "outside", b"private"))
+        with self.assertRaises(ValueError):
+            select_data_files(self.package, external)
+
+
+    def test_external_fallback_briefing_root_symlink_is_rejected(self):
+        external = self.root / "external"
+        self.catalog(external)
+        outside = self.root / "outside-private"
+        self.briefings(outside)
+        bundled_briefings = self.data / "briefings"
+        shutil.rmtree(bundled_briefings)
+        bundled_briefings.symlink_to(outside / "briefings", target_is_directory=True)
+        with self.assertRaisesRegex(ValueError, "symlink directory in bundled mission briefings"):
+            select_data_files(self.package, external)
+
+    def test_unreadable_unlisted_lineage_directory_fails_instead_of_being_omitted(self):
+        proof = self.write(self.data, "hd/jobs/family/proofs/lineage.json", b"required lineage")
+        denied = proof.parent
+        real_scandir = os.scandir
+
+        def checked_scandir(path):
+            if Path(path) == denied:
+                raise PermissionError("synthetic unreadable lineage directory")
+            return real_scandir(path)
+
+        with patch("assemble_runtime.os.scandir", side_effect=checked_scandir):
+            with self.assertRaisesRegex(PermissionError, "synthetic unreadable lineage directory"):
+                select_data_files(self.package)
+
+
+
+if __name__ == "__main__":
+    unittest.main()