Publish Ludarium source

2026-09-03 02:08:58 +02:00
commit df869819ce
339 changed files with 46455 additions and 0 deletions
@@ -0,0 +1,609 @@
+#!/usr/bin/env python3
+"""Deterministic tests for the emulator sidecar control surface.
+
+`resolve_game_path` is the last defence between an authenticated control request and a
+process launch inside the sidecar, and `desktop_display` decides whether a launch reaches
+the live desktop at all. Both are covered here so a future sidecar (Azahar, xemu, Cemu,
+Vita3K) inherits proven behaviour instead of a copied one.
+
+Run with: python -m unittest discover -s tests/controllers -t .
+"""
+import json
+import os
+import sys
+import tempfile
+import threading
+import unittest
+import urllib.error
+import urllib.request
+from http.server import ThreadingHTTPServer
+from pathlib import Path
+from unittest import mock
+
+DEPLOY = Path(__file__).resolve().parents[2] / "deploy"
+sys.path.insert(0, str(DEPLOY))
+
+import ludarium_sidecar as sidecar  # noqa: E402
+
+
+def load_controller(module_name, file_name, token_variable, token="test-token"):
+    """Import a controller script under a controlled games root and control token."""
+    import importlib.util
+    root = tempfile.mkdtemp(prefix=f"ludarium-{module_name}-")
+    environment = {token_variable: token, "LUDARIUM_GAMES_ROOT": root}
+    with mock.patch.dict(os.environ, environment, clear=False):
+        spec = importlib.util.spec_from_file_location(module_name, DEPLOY / file_name)
+        module = importlib.util.module_from_spec(spec)
+        spec.loader.exec_module(module)
+    return module, Path(root)
+
+
+class ResolveGamePathTests(unittest.TestCase):
+    def setUp(self):
+        self.root = Path(tempfile.mkdtemp(prefix="ludarium-games-"))
+        (self.root / "gamecube").mkdir()
+        self.game = self.root / "gamecube" / "title.iso"
+        self.game.write_bytes(b"\0" * 16)
+        self.extensions = {".iso", ".rvz"}
+        self.platforms = {"gamecube", "wii"}
+
+    def resolve(self, value, platforms=None):
+        return sidecar.resolve_game_path(value, self.root, self.extensions, platforms)
+
+    def test_accepts_an_allowlisted_file_inside_the_mount(self):
+        self.assertEqual(self.resolve("gamecube/title.iso", self.platforms), self.game.resolve())
+
+    def test_rejects_an_absolute_path(self):
+        with self.assertRaises(ValueError):
+            self.resolve("/etc/passwd")
+
+    def test_rejects_an_empty_path(self):
+        with self.assertRaises(ValueError):
+            self.resolve("")
+
+    def test_rejects_parent_traversal(self):
+        with self.assertRaises(ValueError):
+            self.resolve("gamecube/../../etc/passwd", self.platforms)
+
+    def test_normalises_a_current_directory_segment(self):
+        # PurePosixPath drops "." segments, so the path still resolves inside the mount.
+        self.assertEqual(self.resolve("gamecube/./title.iso", self.platforms), self.game.resolve())
+
+    def test_rejects_a_leading_parent_segment(self):
+        with self.assertRaises(ValueError):
+            self.resolve("../gamecube/title.iso", self.platforms)
+
+    def test_rejects_an_unlisted_platform_directory(self):
+        with self.assertRaises(ValueError):
+            self.resolve("switch/title.iso", self.platforms)
+
+    def test_rejects_a_bare_file_when_a_platform_directory_is_required(self):
+        with self.assertRaises(ValueError):
+            self.resolve("title.iso", self.platforms)
+
+    def test_rejects_a_disallowed_extension(self):
+        (self.root / "gamecube" / "notes.txt").write_text("x")
+        with self.assertRaises(ValueError):
+            self.resolve("gamecube/notes.txt", self.platforms)
+
+    def test_extension_matching_is_case_insensitive(self):
+        upper = self.root / "gamecube" / "Upper.ISO"
+        upper.write_bytes(b"\0" * 16)
+        self.assertEqual(self.resolve("gamecube/Upper.ISO", self.platforms), upper.resolve())
+
+    def test_rejects_a_missing_file(self):
+        with self.assertRaises(ValueError):
+            self.resolve("gamecube/absent.iso", self.platforms)
+
+    def test_rejects_a_directory(self):
+        (self.root / "gamecube" / "disc.iso").mkdir()
+        with self.assertRaises(ValueError):
+            self.resolve("gamecube/disc.iso", self.platforms)
+
+    @unittest.skipUnless(hasattr(os, "symlink") and sys.platform != "win32",
+                         "symlink creation needs privileges on Windows")
+    def test_rejects_a_symlink_that_escapes_the_mount(self):
+        outside = Path(tempfile.mkdtemp(prefix="ludarium-outside-")) / "secret.iso"
+        outside.write_bytes(b"\0" * 16)
+        os.symlink(outside, self.root / "gamecube" / "link.iso")
+        with self.assertRaises(ValueError):
+            self.resolve("gamecube/link.iso", self.platforms)
+
+    def test_accepts_a_bare_file_when_no_platform_directory_is_required(self):
+        game = self.root / "title.rvz"
+        game.write_bytes(b"\0" * 16)
+        self.assertEqual(self.resolve("title.rvz"), game.resolve())
+
+
+class DesktopDisplayTests(unittest.TestCase):
+    """The live Eden defect: after a restart the newest socket is not the active display."""
+
+    def test_prefers_the_display_published_by_the_running_process(self):
+        def run(argv, **kwargs):
+            if argv[0].endswith("pgrep"):
+                return mock.Mock(returncode=0, stdout="4242")
+            return mock.Mock(returncode=0, stdout=b"HOME=/config\0DISPLAY=:7\0LANG=C\0")
+        with mock.patch.object(sidecar.subprocess, "run", side_effect=run):
+            self.assertEqual(sidecar.desktop_display("eden"), ":7")
+
+    def test_falls_back_to_the_newest_socket_when_no_process_is_running(self):
+        with mock.patch.object(sidecar.subprocess, "run",
+                               return_value=mock.Mock(returncode=1, stdout="")), \
+             mock.patch.object(sidecar.Path, "glob", autospec=True) as glob:
+            sockets = []
+            for number, mtime in ((0, 100), (9, 50)):
+                socket = mock.Mock()
+                socket.name = f"X{number}"
+                socket.stat.return_value = mock.Mock(st_mtime_ns=mtime)
+                sockets.append(socket)
+            glob.return_value = sockets
+            # The newest socket wins over the highest number, matching the live fallback.
+            self.assertEqual(sidecar.desktop_display("eden"), ":0")
+
+    def test_falls_back_to_the_environment_when_nothing_is_discoverable(self):
+        with mock.patch.object(sidecar.subprocess, "run",
+                               return_value=mock.Mock(returncode=1, stdout="")), \
+             mock.patch.object(sidecar.Path, "glob", autospec=True, return_value=[]), \
+             mock.patch.dict(os.environ, {"DISPLAY": ":3"}, clear=False):
+            self.assertEqual(sidecar.desktop_display("eden"), ":3")
+
+    def test_ignores_a_malformed_published_display(self):
+        def run(argv, **kwargs):
+            if argv[0].endswith("pgrep"):
+                return mock.Mock(returncode=0, stdout="4242")
+            return mock.Mock(returncode=0, stdout=b"DISPLAY=:not-a-number\0")
+        with mock.patch.object(sidecar.subprocess, "run", side_effect=run), \
+             mock.patch.object(sidecar.Path, "glob", autospec=True, return_value=[]), \
+             mock.patch.dict(os.environ, {"DISPLAY": ":1"}, clear=False):
+            self.assertEqual(sidecar.desktop_display("eden"), ":1")
+
+    def test_survives_a_failing_process_lookup(self):
+        with mock.patch.object(sidecar.subprocess, "run", side_effect=OSError("no pgrep")), \
+             mock.patch.object(sidecar.Path, "glob", autospec=True, return_value=[]), \
+             mock.patch.dict(os.environ, {"DISPLAY": ":2"}, clear=False):
+            self.assertEqual(sidecar.desktop_display("eden"), ":2")
+
+
+class VisibleWindowTests(unittest.TestCase):
+    def test_prefers_the_launched_process_id(self):
+        calls = []
+
+        def run(argv, **kwargs):
+            calls.append(argv)
+            return mock.Mock(returncode=0, stdout="123\n")
+        with mock.patch.object(sidecar.subprocess, "run", side_effect=run):
+            self.assertEqual(sidecar.visible_windows({}, pid=99, name="Dolphin"), ["123"])
+        self.assertEqual(len(calls), 1)
+        self.assertIn("--pid", calls[0])
+
+    def test_falls_back_to_the_window_name(self):
+        def run(argv, **kwargs):
+            return mock.Mock(returncode=0, stdout="" if "--pid" in argv else "456\n")
+        with mock.patch.object(sidecar.subprocess, "run", side_effect=run):
+            self.assertEqual(sidecar.visible_windows({}, pid=99, name="Dolphin"), ["456"])
+
+    def test_returns_nothing_when_no_window_matches(self):
+        with mock.patch.object(sidecar.subprocess, "run",
+                               return_value=mock.Mock(returncode=1, stdout="")):
+            self.assertEqual(sidecar.visible_windows({}, pid=99, name="Dolphin"), [])
+
+
+def build_runtime(root, **overrides):
+    options = dict(
+        name="dolphin", process_name="dolphin-emu",
+        launch_argv=lambda target: ["/usr/bin/dolphin-emu", "--batch", "--exec", str(target)],
+        games=root, extensions={".iso"}, hotkeys={"pause-resume": "F10"},
+        save_mode="native-and-savestate-persistent", platforms={"gamecube"},
+        window_name="Dolphin", ready_seconds=0, save_paths=())
+    options.update(overrides)
+    return sidecar.EmulatorRuntime(**options)
+
+
+class EmulatorRuntimeTests(unittest.TestCase):
+    def setUp(self):
+        self.root = Path(tempfile.mkdtemp(prefix="ludarium-runtime-"))
+        (self.root / "gamecube").mkdir()
+        (self.root / "gamecube" / "title.iso").write_bytes(b"\0" * 16)
+        self.runtime = build_runtime(self.root)
+        self.environment = mock.patch.object(
+            sidecar, "runtime_environment", return_value={"DISPLAY": ":1"})
+        self.environment.start()
+        self.addCleanup(self.environment.stop)
+
+    def test_reports_no_running_title_before_a_launch(self):
+        self.assertEqual(self.runtime.status(),
+                         {"running": False, "pid": None, "saveRoot": None, "savePaths": [],
+                          "saveMode": "native-and-savestate-persistent", "saveData": False})
+
+    def test_launch_returns_the_process_id_and_reports_running(self):
+        process = mock.Mock(pid=4321)
+        process.poll.return_value = None
+        with mock.patch.object(sidecar.subprocess, "Popen", return_value=process) as popen:
+            self.assertEqual(self.runtime.launch(self.root / "gamecube" / "title.iso"), 4321)
+        argv = popen.call_args.args[0]
+        self.assertEqual(argv[:2], [sidecar.SETUIDGID, sidecar.RUNTIME_USER])
+        self.assertIn("--batch", argv)
+        self.assertTrue(self.runtime.status()["running"])
+
+    def test_launch_reports_an_early_exit(self):
+        process = mock.Mock(pid=4321, returncode=1)
+        process.poll.return_value = 1
+        with mock.patch.object(sidecar.subprocess, "Popen", return_value=process):
+            self.assertIsNone(self.runtime.launch(self.root / "gamecube" / "title.iso"))
+
+    def test_launch_terminates_a_previously_running_title(self):
+        first = mock.Mock(pid=1)
+        first.poll.return_value = None
+        second = mock.Mock(pid=2)
+        second.poll.return_value = None
+        with mock.patch.object(sidecar.subprocess, "Popen", side_effect=[first, second]):
+            self.runtime.launch(self.root / "gamecube" / "title.iso")
+            self.runtime.launch(self.root / "gamecube" / "title.iso")
+        first.send_signal.assert_called_once_with(sidecar.signal.SIGTERM)
+
+    def test_stop_reports_false_when_nothing_runs(self):
+        self.assertFalse(self.runtime.stop())
+
+    def test_stop_kills_a_title_that_ignores_termination(self):
+        process = mock.Mock(pid=7)
+        process.poll.return_value = None
+        process.wait.side_effect = [sidecar.subprocess.TimeoutExpired("dolphin-emu", 8), 0]
+        with mock.patch.object(sidecar.subprocess, "Popen", return_value=process):
+            self.runtime.launch(self.root / "gamecube" / "title.iso")
+        self.assertTrue(self.runtime.stop())
+        process.kill.assert_called_once()
+        self.assertFalse(self.runtime.status()["running"])
+
+    def test_hotkey_targets_the_launched_process(self):
+        process = mock.Mock(pid=555)
+        process.poll.return_value = None
+        sent = []
+
+        def run(argv, **kwargs):
+            sent.append(argv)
+            return mock.Mock(returncode=0, stdout="900\n")
+        with mock.patch.object(sidecar.subprocess, "Popen", return_value=process):
+            self.runtime.launch(self.root / "gamecube" / "title.iso")
+        with mock.patch.object(sidecar.subprocess, "run", side_effect=run):
+            self.assertTrue(self.runtime.hotkey("F10"))
+        self.assertIn("555", sent[0])
+        self.assertEqual(sent[-1][-3:], ["900", "key", "F10"])
+
+    def test_hotkey_reports_false_without_a_window(self):
+        with mock.patch.object(sidecar.subprocess, "run",
+                               return_value=mock.Mock(returncode=1, stdout="")):
+            self.assertFalse(self.runtime.hotkey("F10"))
+
+
+class ControlSurfaceTests(unittest.TestCase):
+    """Covers the exact status codes the Ludarium API depends on."""
+
+    def setUp(self):
+        self.root = Path(tempfile.mkdtemp(prefix="ludarium-surface-"))
+        (self.root / "gamecube").mkdir()
+        (self.root / "gamecube" / "title.iso").write_bytes(b"\0" * 16)
+        self.runtime = build_runtime(self.root)
+        self.server = ThreadingHTTPServer(
+            ("127.0.0.1", 0), sidecar.build_handler(self.runtime, "secret-token"))
+        threading.Thread(target=self.server.serve_forever, daemon=True).start()
+        self.addCleanup(self.server.server_close)
+        self.addCleanup(self.server.shutdown)
+        self.base = f"http://127.0.0.1:{self.server.server_address[1]}"
+
+    def call(self, path, method="GET", body=None, token="secret-token", raw=None):
+        payload = raw if raw is not None else (json.dumps(body).encode() if body is not None else None)
+        request = urllib.request.Request(self.base + path, data=payload, method=method)
+        if token:
+            request.add_header("X-Ludarium-Control-Token", token)
+        if payload is not None:
+            request.add_header("Content-Type", "application/json")
+        try:
+            with urllib.request.urlopen(request, timeout=5) as response:
+                return response.status, json.loads(response.read() or b"{}")
+        except urllib.error.HTTPError as error:
+            return error.code, json.loads(error.read() or b"{}")
+
+    def test_health_needs_no_token(self):
+        self.assertEqual(self.call("/health", token=None), (200, {"status": "healthy"}))
+
+    def test_status_needs_a_token(self):
+        self.assertEqual(self.call("/v1/status", token=None)[0], 404)
+        self.assertEqual(self.call("/v1/status", token="wrong")[0], 404)
+
+    def test_status_reports_the_save_mode(self):
+        status, body = self.call("/v1/status")
+        self.assertEqual(status, 200)
+        self.assertEqual(body, {"running": False, "pid": None, "saveRoot": None, "savePaths": [],
+                                "saveMode": "native-and-savestate-persistent", "saveData": False})
+
+    def test_unknown_get_route_is_not_found(self):
+        self.assertEqual(self.call("/v1/anything")[0], 404)
+
+    def test_unauthenticated_post_is_not_found(self):
+        self.assertEqual(self.call("/v1/launch", "POST", {"path": "gamecube/title.iso"},
+                                   token=None)[0], 404)
+
+    def test_launch_accepts_an_allowlisted_game(self):
+        process = mock.Mock(pid=31337)
+        process.poll.return_value = None
+        with mock.patch.object(sidecar, "runtime_environment", return_value={"DISPLAY": ":1"}), \
+             mock.patch.object(sidecar.subprocess, "Popen", return_value=process):
+            status, body = self.call("/v1/launch", "POST", {"path": "gamecube/title.iso"})
+        self.assertEqual(status, 202)
+        self.assertEqual(body["pid"], 31337)
+        self.assertEqual(body["saveMode"], "native-and-savestate-persistent")
+
+    def test_launch_rejects_traversal(self):
+        status, body = self.call("/v1/launch", "POST", {"path": "gamecube/../../etc/passwd"})
+        self.assertEqual(status, 400)
+        self.assertIn("Unsafe", body["message"])
+
+    def test_launch_reports_an_early_exit_as_unavailable(self):
+        process = mock.Mock(pid=1, returncode=3)
+        process.poll.return_value = 3
+        with mock.patch.object(sidecar, "runtime_environment", return_value={"DISPLAY": ":1"}), \
+             mock.patch.object(sidecar.subprocess, "Popen", return_value=process):
+            self.assertEqual(self.call("/v1/launch", "POST", {"path": "gamecube/title.iso"})[0], 503)
+
+    def test_oversized_body_is_rejected(self):
+        self.assertEqual(self.call("/v1/launch", "POST", raw=b"{" + b"a" * 9000)[0], 400)
+
+    def test_empty_body_is_rejected(self):
+        self.assertEqual(self.call("/v1/launch", "POST", raw=b"")[0], 400)
+
+    def test_non_object_body_is_rejected(self):
+        self.assertEqual(self.call("/v1/action", "POST", raw=b'"stop"')[0], 400)
+
+    def test_unsupported_action_is_rejected(self):
+        self.assertEqual(self.call("/v1/action", "POST", {"action": "self-destruct"})[0], 400)
+
+    def test_stop_without_a_running_title_conflicts(self):
+        self.assertEqual(self.call("/v1/action", "POST", {"action": "stop"})[0], 409)
+
+    def test_hotkey_without_a_window_conflicts(self):
+        with mock.patch.object(sidecar, "runtime_environment", return_value={"DISPLAY": ":1"}), \
+             mock.patch.object(sidecar.subprocess, "run",
+                               return_value=mock.Mock(returncode=1, stdout="")):
+            self.assertEqual(self.call("/v1/action", "POST", {"action": "pause-resume"})[0], 409)
+
+
+class ControllerIdentityTests(unittest.TestCase):
+    """Each shipped controller must keep its exact runtime contract."""
+
+    def test_dolphin_controller(self):
+        module, root = load_controller("dolphin_controller", "dolphin-controller.py",
+                                       "LUDARIUM_DOLPHIN_CONTROL_TOKEN")
+        runtime = module.RUNTIME
+        # Dolphin's profile also holds a shader cache; only save-bearing directories are captured.
+        self.assertEqual(runtime.save_paths, ("GC", "Wii", "StateSaves"))
+        self.assertEqual(runtime.name, "dolphin")
+        self.assertEqual(runtime.process_name, "dolphin-emu")
+        self.assertEqual(runtime.platforms, {"gamecube", "wii"})
+        self.assertEqual(runtime.extensions,
+                         {".iso", ".gcm", ".rvz", ".gcz", ".wbfs", ".wia"})
+        self.assertEqual(set(runtime.hotkeys),
+                         {"pause-resume", "fullscreen", "save-state", "load-state"})
+        self.assertEqual(runtime.save_mode, "native-and-savestate-persistent")
+        self.assertEqual(runtime.launch_argv("/games/wii/a.iso"),
+                         ["/usr/bin/dolphin-emu", "--batch", "--exec", "/games/wii/a.iso"])
+        self.assertEqual(Path(runtime.games), root)
+
+    def test_eden_controller(self):
+        module, root = load_controller("eden_controller", "eden-controller.py",
+                                       "LUDARIUM_EDEN_CONTROL_TOKEN")
+        runtime = module.RUNTIME
+        self.assertEqual(runtime.save_paths, ("nand/user/save",))
+        self.assertEqual(runtime.name, "eden")
+        self.assertEqual(runtime.process_name, "eden")
+        self.assertIsNone(runtime.platforms)
+        self.assertEqual(runtime.extensions, {".xci", ".nsp"})
+        self.assertEqual(set(runtime.hotkeys), {"pause-resume", "fullscreen"})
+        self.assertEqual(runtime.save_mode, "native-persistent")
+        self.assertEqual(runtime.launch_argv("/games/a.xci"),
+                         ["/usr/bin/eden", "-f", "-g", "/games/a.xci"])
+        self.assertEqual(Path(runtime.games), root)
+
+    def test_a_controller_refuses_to_serve_without_a_token(self):
+        module, _ = load_controller("eden_untokened", "eden-controller.py",
+                                    "LUDARIUM_EDEN_CONTROL_TOKEN", token="")
+        with self.assertRaises(SystemExit):
+            module.main()
+
+
+if __name__ == "__main__":
+    unittest.main()
+
+
+class SaveDataTests(unittest.TestCase):
+    """Round-tripping an emulator's app-owned save directories through the vault."""
+
+    SAVE_PATHS = ("nand/user/save", "StateSaves")
+
+    def setUp(self):
+        self.root = Path(tempfile.mkdtemp(prefix="ludarium-saves-"))
+        (self.root / "nand" / "user" / "save").mkdir(parents=True)
+        (self.root / "nand" / "user" / "save" / "save.bin").write_bytes(b"progress" * 64)
+        (self.root / "StateSaves").mkdir()
+        (self.root / "StateSaves" / "slot1.sav").write_bytes(b"state")
+        # A profile also holds material a capture must never carry: on a real Dolphin install this
+        # is a shader cache far larger than one bounded revision.
+        (self.root / "Cache").mkdir()
+        (self.root / "Cache" / "shaders.bin").write_bytes(b"x" * 4096)
+        (self.root / "config.ini").write_text("[General]\n")
+
+    def export(self):
+        return sidecar.export_save_data(self.root, self.SAVE_PATHS)
+
+    def test_export_packs_only_the_configured_save_directories(self):
+        import io
+        import tarfile
+        with tarfile.open(fileobj=io.BytesIO(self.export()), mode="r:gz") as archive:
+            names = sorted(member.name for member in archive)
+            self.assertEqual(names, ["StateSaves/slot1.sav", "nand/user/save/save.bin"])
+            self.assertTrue(all(member.mode == 0o644 and member.uid == 0 for member in archive))
+
+    def test_export_packs_the_whole_root_when_no_paths_are_configured(self):
+        import io
+        import tarfile
+        with tarfile.open(fileobj=io.BytesIO(sidecar.export_save_data(self.root)), mode="r:gz") as archive:
+            self.assertIn("Cache/shaders.bin", [member.name for member in archive])
+
+    def test_export_ignores_a_configured_path_that_does_not_exist(self):
+        import io
+        import tarfile
+        payload = sidecar.export_save_data(self.root, ("StateSaves", "GC", "Wii"))
+        with tarfile.open(fileobj=io.BytesIO(payload), mode="r:gz") as archive:
+            self.assertEqual([member.name for member in archive], ["StateSaves/slot1.sav"])
+
+    def test_export_rejects_a_missing_directory(self):
+        with self.assertRaises(ValueError):
+            sidecar.export_save_data(self.root / "absent")
+
+    def test_export_rejects_a_profile_with_no_save_directories(self):
+        with self.assertRaises(ValueError):
+            sidecar.export_save_data(self.root, ("GC", "Wii"))
+
+    def test_export_rejects_an_escaping_configured_path(self):
+        with self.assertRaises(ValueError):
+            sidecar.export_save_data(self.root, ("../..",))
+
+    def test_a_round_trip_restores_the_exact_content(self):
+        payload = self.export()
+        (self.root / "nand" / "user" / "save" / "save.bin").write_bytes(b"overwritten")
+        (self.root / "StateSaves" / "slot1.sav").unlink()
+
+        self.assertEqual(sidecar.import_save_data(self.root, payload, self.SAVE_PATHS), 2)
+        self.assertEqual((self.root / "nand" / "user" / "save" / "save.bin").read_bytes(), b"progress" * 64)
+        self.assertEqual((self.root / "StateSaves" / "slot1.sav").read_bytes(), b"state")
+
+    def test_a_restore_leaves_unrelated_profile_content_alone(self):
+        payload = self.export()
+        sidecar.import_save_data(self.root, payload, self.SAVE_PATHS)
+        self.assertEqual((self.root / "config.ini").read_text(), "[General]\n")
+        self.assertEqual((self.root / "Cache" / "shaders.bin").stat().st_size, 4096)
+
+    def test_restore_rejects_a_member_outside_the_save_directories(self):
+        import io
+        import tarfile
+        buffer = io.BytesIO()
+        with tarfile.open(fileobj=buffer, mode="w:gz") as archive:
+            info = tarfile.TarInfo("config.ini")
+            info.size = 5
+            archive.addfile(info, io.BytesIO(b"evil!"))
+        # A stored revision must never be able to reach an emulator's configuration.
+        with self.assertRaises(ValueError):
+            sidecar.import_save_data(self.root, buffer.getvalue(), self.SAVE_PATHS)
+        self.assertEqual((self.root / "config.ini").read_text(), "[General]\n")
+
+    def test_restore_rejects_an_archive_that_escapes_the_save_directory(self):
+        import io
+        import tarfile
+        buffer = io.BytesIO()
+        with tarfile.open(fileobj=buffer, mode="w:gz") as archive:
+            info = tarfile.TarInfo("../escaped.bin")
+            info.size = 4
+            archive.addfile(info, io.BytesIO(b"evil"))
+        with self.assertRaises(ValueError):
+            sidecar.import_save_data(self.root, buffer.getvalue(), self.SAVE_PATHS)
+        self.assertFalse((self.root.parent / "escaped.bin").exists())
+
+    def test_restore_rejects_a_symlink_member(self):
+        import io
+        import tarfile
+        buffer = io.BytesIO()
+        with tarfile.open(fileobj=buffer, mode="w:gz") as archive:
+            info = tarfile.TarInfo("link")
+            info.type = tarfile.SYMTYPE
+            info.linkname = "/etc/passwd"
+            archive.addfile(info)
+        with self.assertRaises(ValueError):
+            sidecar.import_save_data(self.root, buffer.getvalue(), self.SAVE_PATHS)
+
+    def test_restore_rejects_an_oversized_archive(self):
+        with self.assertRaises(ValueError):
+            sidecar.import_save_data(self.root, b"\0" * (sidecar.MAX_SAVE_DATA_BYTES + 1), self.SAVE_PATHS)
+
+    def test_restore_prevalidates_the_expanded_size_before_writing(self):
+        import io
+        import tarfile
+        buffer = io.BytesIO()
+        with tarfile.open(fileobj=buffer, mode="w:gz") as archive:
+            info = tarfile.TarInfo("save/large.bin")
+            info.size = 5
+            archive.addfile(info, io.BytesIO(b"12345"))
+
+        with mock.patch.object(sidecar, "MAX_SAVE_DATA_EXPANDED_BYTES", 4), self.assertRaises(ValueError):
+            sidecar.import_save_data(self.root, buffer.getvalue(), self.SAVE_PATHS)
+        self.assertFalse((self.root / "save" / "large.bin").exists())
+
+    def test_status_reports_save_data_support_and_the_exact_directories(self):
+        without = build_runtime(self.root)
+        with_saves = build_runtime(self.root, save_root=str(self.root), save_paths=self.SAVE_PATHS)
+
+        self.assertFalse(without.status()["saveData"])
+        status = with_saves.status()
+        self.assertTrue(status["saveData"])
+        self.assertEqual(status["saveRoot"], str(self.root))
+        # A misconfigured path is visible here instead of surfacing at capture time.
+        self.assertEqual(sorted(Path(path).name for path in status["savePaths"]),
+                         ["StateSaves", "save"])
+        self.assertFalse(build_runtime(self.root, save_root=str(self.root / "absent")).status()["saveData"])
+        self.assertFalse(build_runtime(self.root, save_root=str(self.root),
+                                       save_paths=("GC",)).status()["saveData"])
+
+
+class SaveDataSurfaceTests(unittest.TestCase):
+    def setUp(self):
+        self.root = Path(tempfile.mkdtemp(prefix="ludarium-savesurface-"))
+        (self.root / "gamecube").mkdir()
+        (self.root / "gamecube" / "title.iso").write_bytes(b"\0" * 16)
+        self.saves = Path(tempfile.mkdtemp(prefix="ludarium-saveroot-"))
+        (self.saves / "GC").mkdir()
+        (self.saves / "GC" / "card.raw").write_bytes(b"memory-card")
+        self.runtime = build_runtime(self.root, save_root=str(self.saves), save_paths=("GC",))
+        self.server = ThreadingHTTPServer(
+            ("127.0.0.1", 0), sidecar.build_handler(self.runtime, "secret-token"))
+        threading.Thread(target=self.server.serve_forever, daemon=True).start()
+        self.addCleanup(self.server.server_close)
+        self.addCleanup(self.server.shutdown)
+        self.base = f"http://127.0.0.1:{self.server.server_address[1]}"
+
+    def request(self, method, data=None, token="secret-token"):
+        request = urllib.request.Request(self.base + "/v1/save-data", data=data, method=method)
+        if token:
+            request.add_header("X-Ludarium-Control-Token", token)
+        try:
+            with urllib.request.urlopen(request, timeout=5) as response:
+                return response.status, response.read()
+        except urllib.error.HTTPError as error:
+            return error.code, error.read()
+
+    def test_export_needs_a_token(self):
+        self.assertEqual(self.request("GET", token=None)[0], 404)
+
+    def test_export_and_restore_round_trip_over_http(self):
+        status, payload = self.request("GET")
+        self.assertEqual(status, 200)
+        (self.saves / "GC" / "card.raw").write_bytes(b"lost")
+
+        self.assertEqual(self.request("POST", payload)[0], 202)
+        self.assertEqual((self.saves / "GC" / "card.raw").read_bytes(), b"memory-card")
+
+    def test_restore_rejects_a_non_archive(self):
+        self.assertEqual(self.request("POST", b"not-an-archive")[0], 400)
+
+    def test_restore_rejects_an_empty_body(self):
+        self.assertEqual(self.request("POST", b"")[0], 400)
+
+    def test_a_runtime_without_a_save_directory_conflicts(self):
+        runtime = build_runtime(self.root)
+        server = ThreadingHTTPServer(("127.0.0.1", 0), sidecar.build_handler(runtime, "secret-token"))
+        threading.Thread(target=server.serve_forever, daemon=True).start()
+        self.addCleanup(server.server_close)
+        self.addCleanup(server.shutdown)
+        request = urllib.request.Request(
+            f"http://127.0.0.1:{server.server_address[1]}/v1/save-data", method="GET")
+        request.add_header("X-Ludarium-Control-Token", "secret-token")
+        with self.assertRaises(urllib.error.HTTPError) as error:
+            urllib.request.urlopen(request, timeout=5)
+        self.assertEqual(error.exception.code, 409)