import base64 import copy import hashlib import json from pathlib import Path import unittest from unittest.mock import patch import urllib.error import test_agent as fixtures module = fixtures.module AgentError = fixtures.AgentError IMAGE = "ghcr.io/ayymoss/plutainer:latest" BOOTSTRAPPER = "bin/plutonium-bootstrapper-win32.exe" BASE_URL = "https://cdn.plutoniummod.com/updater/prod/files/" class PlutoniumCacheTests(unittest.TestCase): tearDown = fixtures.AgentTests.tearDown operation = fixtures.AgentTests.operation create = fixtures.AgentTests.create def setUp(self): fixtures.AgentTests.setUp(self) network = patch.object(self.agent, "plutonium_open", side_effect=AssertionError("Unexpected vendor request"), create=True) network.start() self.addCleanup(network.stop) def release(self, revision=5354, contents=None): contents = contents or { BOOTSTRAPPER: b"bootstrapper", "bin/plutonium.dll": b"runtime", "storage/t5/zone/plutonium.ff": b"zone-data", "storage/iw5/images/~weapon_ak47_iw5_spc-rgb&weap~fa48e615.iwi": b"texture-data", "launcher/plutonium.exe": b"launcher-not-needed", } manifest = { "product": "plutonium-core-prod", "revision": revision, "baseUrl": BASE_URL, "files": [ {"name": name, "size": len(content), "hash": hashlib.sha1(content).hexdigest()} for name, content in contents.items() ], } return manifest, contents def opener(self, manifest, contents, calls=None): calls = calls if calls is not None else [] def open_url(url): calls.append(url) if url == module.PLUTONIUM_MANIFEST_URL: return fixtures.Response(json.dumps(manifest).encode()) self.assertTrue(url.startswith(BASE_URL), url) name = next(entry["name"] for entry in manifest["files"] if url == BASE_URL + entry["hash"]) return fixtures.Response(contents[name]) return open_url def file_url(self, content): return BASE_URL + hashlib.sha1(content).hexdigest() def prepare(self, revision=5354, contents=None): manifest, contents = self.release(revision, contents) calls = [] with patch.object(self.agent, "plutonium_open", side_effect=self.opener(manifest, contents, calls)): result = self.agent.prepare_plutonium({"revision": revision}) return manifest, contents, calls, result def cache_content(self): root = self.agent.root / "agent" / "plutonium-cache" pointer = json.loads((root / "current.json").read_text()) return root / pointer["key"] / "content" def app(self, name="app"): app = Path(self.temp.name) / name app.mkdir(exist_ok=True) return app def assert_runtime(self, app, manifest, contents): runtime = app / "runtime" / "plutonium" self.assertEqual(json.loads((runtime / "cdn_info.json").read_text()), manifest) for name, content in contents.items(): if name.startswith("launcher"): self.assertFalse((runtime / name).exists()) else: self.assertEqual((runtime / name).read_bytes(), content) def test_cold_preparation_verifies_downloads_and_reports_inventory(self): self.assertIsNone(self.agent.plutonium_cache_inventory()) manifest, contents, calls, result = self.prepare() expected_bytes = sum(len(content) for name, content in contents.items() if not name.startswith("launcher")) self.assertEqual(result, {"revision": 5354, "bytes": expected_bytes}) self.assertEqual(self.agent.plutonium_cache_inventory(), result) self.assertEqual(calls[0], module.PLUTONIUM_MANIFEST_URL) self.assertEqual(set(calls[1:]), {self.file_url(content) for name, content in contents.items() if not name.startswith("launcher")}) self.assertEqual(json.loads((self.cache_content() / "cdn_info.json").read_text()), manifest) self.assertEqual((self.agent.root / "agent" / "plutonium-cache").stat().st_mode & 0o077, 0) def test_warm_preparation_only_fetches_manifest_and_dispatches(self): manifest, contents, _, first = self.prepare() calls = [] with patch.object(self.agent, "plutonium_open", side_effect=self.opener(manifest, contents, calls)): result = self.agent.dispatch({"kind": "plutonium.prepare", "payload": {"revision": 5354}}) self.assertEqual(result, first) self.assertEqual(calls, [module.PLUTONIUM_MANIFEST_URL]) def test_refresh_downloads_changed_files_and_reuses_verified_unchanged_files(self): _, contents, _, _ = self.prepare() changed = {**contents, "bin/plutonium.dll": b"new-runtime", "storage/t6/new.ff": b"new-zone"} manifest, changed, calls, result = self.prepare(5355, changed) self.assertEqual(result["revision"], 5355) self.assertEqual(set(calls[1:]), {self.file_url(changed["bin/plutonium.dll"]), self.file_url(changed["storage/t6/new.ff"])}) app = self.app() self.assertTrue(self.agent.seed_plutonium(app, IMAGE)) self.assert_runtime(app, manifest, changed) def test_changed_manifest_is_refreshed_even_when_revision_number_is_unchanged(self): _, contents, _, _ = self.prepare() previous = self.cache_content() manifest, changed, calls, result = self.prepare(5354, {**contents, "bin/plutonium.dll": b"corrected-runtime"}) self.assertEqual(result["revision"], 5354) self.assertNotEqual(self.cache_content(), previous) self.assertEqual(calls[1:], [self.file_url(changed["bin/plutonium.dll"])]) app = self.app() self.assertTrue(self.agent.seed_plutonium(app, IMAGE)) self.assert_runtime(app, manifest, changed) def test_failed_refresh_preserves_the_older_complete_cache(self): manifest, contents, _, result = self.prepare() previous = self.cache_content() next_manifest, next_contents = self.release(5355, {**contents, "bin/plutonium.dll": b"new-runtime"}) next_contents["bin/plutonium.dll"] = b"bad-runtime" with patch.object(self.agent, "plutonium_open", side_effect=self.opener(next_manifest, next_contents)): with self.assertRaises(AgentError): self.agent.prepare_plutonium({"revision": 5355}) self.assertEqual(self.cache_content(), previous) self.assertEqual(self.agent.plutonium_cache_inventory(), result) app = self.app() self.assertTrue(self.agent.seed_plutonium(app, IMAGE)) self.assert_runtime(app, manifest, contents) def test_interrupted_preparation_resumes_without_advertising_a_partial_cache(self): manifest, contents = self.release() open_url = self.opener(manifest, contents) def interrupted_open(url): if url == self.file_url(contents["bin/plutonium.dll"]): raise urllib.error.URLError("simulated network interruption") return open_url(url) with patch.object(self.agent, "plutonium_open", side_effect=interrupted_open): with self.assertRaises(AgentError): self.agent.prepare_plutonium({}) self.assertIsNone(self.agent.plutonium_cache_inventory()) self.assertFalse((self.agent.root / "agent/plutonium-cache/current.json").exists()) _, _, calls, result = self.prepare() self.assertNotIn(self.file_url(contents[BOOTSTRAPPER]), calls) self.assertEqual(result["revision"], manifest["revision"]) def test_agent_restart_reconciles_preparation_instead_of_accepting_an_older_cache(self): _, contents, _, _ = self.prepare() operation = {"id": "prepare-new-release", "kind": "plutonium.prepare", "payload": {"revision": 5355}} self.agent.journal.accept(operation) self.agent.journal.update(operation["id"], "running") self.agent.journal.close() self.agent = fixtures.Agent(self.config, self.docker) self.assertEqual(self.agent.journal.get(operation["id"])["state"], "unknown") manifest, contents = self.release(5355, {**contents, "bin/plutonium.dll": b"new-runtime"}) calls = [] with patch.object(self.agent, "plutonium_open", side_effect=self.opener(manifest, contents, calls)): result = self.agent.execute(operation) self.assertEqual(result["state"], "completed", result) self.assertEqual(json.loads(result["result"])["revision"], 5355) self.assertTrue(json.loads(result["result"])["reconciled"]) self.assertEqual(self.agent.plutonium_cache_inventory()["revision"], 5355) self.assertEqual(calls, [module.PLUTONIUM_MANIFEST_URL, self.file_url(contents["bin/plutonium.dll"])]) with patch.object(self.agent, "plutonium_open", side_effect=AssertionError("Completed operation repeated")): self.assertEqual(self.agent.execute(operation)["result"], result["result"]) def test_heartbeat_defers_preparation_only_for_explicit_true(self): self.assertFalse(self.agent.defer_plutonium_preparation) for response, expected in ( ({"deferPlutoniumPreparation": True}, True), ({}, False), ({"deferPlutoniumPreparation": False}, False), ({"deferPlutoniumPreparation": 1}, False), ({"deferPlutoniumPreparation": "true"}, False), ({"deferPlutoniumPreparation": None}, False), ): self.agent.defer_plutonium_preparation = not expected with self.subTest(response=response), patch.object(self.agent, "request", return_value=response), patch.object(self.agent, "inventory", return_value={}), patch.object(self.agent, "record_health"): self.agent.heartbeat() self.assertIs(self.agent.defer_plutonium_preparation, expected) def assert_preparation_yields_and_resumes(self, *, restarted): manifest, contents = self.release() operation = {"id": "cooperative-prepare", "kind": "plutonium.prepare", "payload": {"revision": 5354}} if restarted: self.agent.journal.accept(operation) self.agent.journal.update(operation["id"], "running") self.agent.journal.close() self.agent = fixtures.Agent(self.config, self.docker) self.assertEqual(self.agent.journal.get(operation["id"])["state"], "unknown") agent = self.agent open_url = self.opener(manifest, contents) class BusyResponse(fixtures.Response): def read(self, size=-1): chunk = super().read(size) if chunk: agent.defer_plutonium_preparation = True return chunk def become_busy_during_download(url): if url == self.file_url(contents["bin/plutonium.dll"]): return BusyResponse(contents["bin/plutonium.dll"]) return open_url(url) with patch.object(self.agent, "plutonium_open", side_effect=become_busy_during_download): result = self.agent.execute(operation) self.assertEqual(result["state"], "accepted", result) self.assertIn("waiting for other host operations", json.loads(result["result"])["message"]) self.assertIsNone(self.agent.executing_id) self.assertIsNone(self.agent.plutonium_cache_inventory()) self.assertTrue(self.agent.defer_plutonium_preparation) # A yielded preparation must let other mutations finish before retrying. self.create() self.assertTrue(self.agent.inspect(fixtures.INSTANCE, 1)["running"]) self.agent.defer_plutonium_preparation = False calls = [] with patch.object(self.agent, "plutonium_open", side_effect=self.opener(manifest, contents, calls)): result = self.agent.execute(operation) self.assertEqual(result["state"], "completed", result) self.assertEqual(result["id"], operation["id"]) self.assertNotIn(self.file_url(contents[BOOTSTRAPPER]), calls) self.assertIn(self.file_url(contents["bin/plutonium.dll"]), calls) self.assertEqual(self.agent.plutonium_cache_inventory()["revision"], 5354) def test_busy_host_yields_mid_download_and_resumes_completed_files(self): self.assert_preparation_yields_and_resumes(restarted=False) def test_reconciled_preparation_yields_mid_download_and_resumes_same_operation(self): self.assert_preparation_yields_and_resumes(restarted=True) def test_resumption_removes_orphans_before_storage_check_and_publication(self): manifest, contents = self.release() raw = json.dumps(manifest).encode() staging = self.agent.root / "agent/plutonium-cache" / (module.digest(raw) + ".partial/content") bootstrapper = staging / BOOTSTRAPPER bootstrapper.parent.mkdir(parents=True) bootstrapper.write_bytes(contents[BOOTSTRAPPER]) orphans = [staging / ".download-abandoned", staging / "bin/.ec-abandoned", staging / "unlisted/extra.dat", staging / "cdn_info.json"] for orphan in orphans: orphan.parent.mkdir(parents=True, exist_ok=True) orphan.write_bytes(b"orphaned bytes") outside = Path(self.temp.name) / "outside-cache" outside.mkdir() (outside / "untouched").write_bytes(b"private outside bytes") link = staging / "outside-link" link.symlink_to(outside, target_is_directory=True) original_disk_usage = module.shutil.disk_usage def checked_storage(path): for orphan in orphans: self.assertFalse(orphan.exists(), "orphan must be removed before storage accounting") self.assertFalse(link.is_symlink()) return original_disk_usage(path) calls = [] with patch.object(module.shutil, "disk_usage", side_effect=checked_storage), patch.object(self.agent, "plutonium_open", side_effect=self.opener(manifest, contents, calls)): result = self.agent.prepare_plutonium({}) content = self.cache_content() expected = {name for name in contents if not name.startswith("launcher")} actual = {path.relative_to(content).as_posix() for path in content.rglob("*") if path.is_file()} self.assertEqual(actual, expected | {"cdn_info.json"}) self.assertEqual(result["bytes"], sum(len(contents[name]) for name in expected)) self.assertEqual(sum(path.stat().st_size for path in content.rglob("*") if path.is_file()), result["bytes"] + len(raw)) self.assertNotIn(self.file_url(contents[BOOTSTRAPPER]), calls) self.assertEqual((outside / "untouched").read_bytes(), b"private outside bytes") app = self.app() self.assertTrue(self.agent.seed_plutonium(app, IMAGE)) self.assert_runtime(app, manifest, contents) self.assertFalse((app / "runtime/plutonium/unlisted/extra.dat").exists()) def test_public_downloads_reject_other_origins_and_never_send_host_credentials(self): # Call the class method to bypass the fixture's default network blocker; # the underlying opener is mocked, so no request reaches the network. with patch.object(module.urllib.request, "build_opener") as build: for url in ( "http://cdn.plutoniummod.com/updater/prod/info.json", "https://attacker.example/updater/prod/info.json", "https://cdn.plutoniummod.com.attacker.example/updater/prod/info.json", "https://secret@cdn.plutoniummod.com/updater/prod/info.json", "https://cdn.plutoniummod.com/updater/dev/info.json", ): with self.subTest(url=url), self.assertRaises(AgentError): fixtures.Agent.plutonium_open(self.agent, url) build.assert_not_called() fixtures.Agent.plutonium_open(self.agent, module.PLUTONIUM_MANIFEST_URL) build.return_value.open.assert_called_once_with(module.PLUTONIUM_MANIFEST_URL, timeout=30) handler = build.call_args.args[0] for redirect in (module.PLUTONIUM_MANIFEST_URL, "https://attacker.example/collect"): with self.subTest(redirect=redirect), self.assertRaises(AgentError): handler.redirect_request(None, None, 302, "Found", {}, redirect) def test_rejects_revision_downgrade_and_invalid_preparation_arguments(self): self.prepare() manifest, contents = self.release(5353) with patch.object(self.agent, "plutonium_open", side_effect=self.opener(manifest, contents)): with self.assertRaises(AgentError): self.agent.prepare_plutonium({"revision": 5354}) for payload in ({"revision": -1}, {"revision": True}, {"revision": "5354"}, {"minimumFreeBytes": -1}, {"minimumFreeBytes": True}, {"minimumFreeBytes": "10"}): with self.subTest(payload=payload), self.assertRaises(AgentError): self.agent.prepare_plutonium(payload) def test_disk_reserve_failure_leaves_existing_cache_usable(self): _, contents, _, result = self.prepare() manifest, contents = self.release(5355, {**contents, "bin/plutonium.dll": b"new-runtime"}) calls = [] usage = type(module.shutil.disk_usage(self.agent.root))(10000, 9900, 100) with patch.object(module.shutil, "disk_usage", return_value=usage), patch.object(self.agent, "plutonium_open", side_effect=self.opener(manifest, contents, calls)): with self.assertRaises(AgentError): self.agent.prepare_plutonium({"minimumFreeBytes": 1000}) self.assertEqual(calls, [module.PLUTONIUM_MANIFEST_URL]) self.assertEqual(self.agent.plutonium_cache_inventory(), result) def test_malformed_manifest_never_publishes_a_cache(self): good, contents = self.release() malformed = [ None, [], {}, {**good, "product": "other-product"}, {**good, "revision": 0}, {**good, "revision": True}, {**good, "revision": "5354"}, {**good, "baseUrl": "http://cdn.plutoniummod.com/updater/prod/"}, {**good, "baseUrl": "https://attacker.example/"}, {**good, "files": []}, {**good, "files": {}}, {**good, "files": good["files"][1:]}, {**good, "files": good["files"] + [good["files"][0]]}, ] for manifest in malformed: with self.subTest(manifest=manifest), patch.object(self.agent, "plutonium_open", side_effect=self.opener(manifest, contents)): with self.assertRaises(AgentError): self.agent.prepare_plutonium({}) self.assertIsNone(self.agent.plutonium_cache_inventory()) def test_manifest_rejects_unsafe_paths_sizes_and_hashes_before_downloading(self): good, contents = self.release() bad_entries = [ {"name": name} for name in ( ".", "../escape", "/absolute", "bin/../../escape", "bin//file", "bin/./file", "bin\\file", "C:/escape", "bin/file\x00", "cdn_info.json", ) ] + [ {"size": size} for size in (-1, True, "5", None) ] + [ {"hash": value} for value in ("", "0" * 39, "z" * 40, None) ] for override in bad_entries: manifest = copy.deepcopy(good) manifest["files"][1].update(override) calls = [] with self.subTest(override=override), patch.object(self.agent, "plutonium_open", side_effect=self.opener(manifest, contents, calls)): with self.assertRaises(AgentError): self.agent.prepare_plutonium({}) self.assertEqual(calls, [module.PLUTONIUM_MANIFEST_URL]) self.assertIsNone(self.agent.plutonium_cache_inventory()) def test_short_oversize_and_wrong_hash_downloads_are_not_published(self): manifest, contents = self.release() for body in (b"short", contents[BOOTSTRAPPER] + b"extra", b"X" * len(contents[BOOTSTRAPPER])): with self.subTest(body=body), patch.object(self.agent, "plutonium_open", side_effect=self.opener(manifest, {**contents, BOOTSTRAPPER: body})): with self.assertRaises(AgentError): self.agent.prepare_plutonium({}) self.assertIsNone(self.agent.plutonium_cache_inventory()) def test_seeds_are_private_and_preserve_existing_lobby_state(self): manifest, contents, _, _ = self.prepare() first, second = self.app("first"), self.app("second") private = { "configs/server.cfg": b"secret config", "runtime/plutonium/plugins/private.dll": b"private plugin", "runtime/plutonium/storage/t5/logs/games_mp.log": b"lobby log", } for name, body in private.items(): target = first / name target.parent.mkdir(parents=True, exist_ok=True) target.write_bytes(body) self.assertTrue(self.agent.seed_plutonium(first, IMAGE)) self.assertTrue(self.agent.seed_plutonium(second, IMAGE)) self.assert_runtime(first, manifest, contents) self.assert_runtime(second, manifest, contents) for name, body in private.items(): self.assertEqual((first / name).read_bytes(), body) self.assertFalse((second / name).exists()) first_file = first / "runtime/plutonium" / BOOTSTRAPPER second_file = second / "runtime/plutonium" / BOOTSTRAPPER cached_file = self.cache_content() / BOOTSTRAPPER self.assertNotEqual(first_file.stat().st_ino, second_file.stat().st_ino) self.assertNotEqual(first_file.stat().st_ino, cached_file.stat().st_ino) first_file.write_bytes(b"container-updated") self.assertEqual(second_file.read_bytes(), contents[BOOTSTRAPPER]) self.assertEqual(cached_file.read_bytes(), contents[BOOTSTRAPPER]) def test_absent_or_corrupt_cache_and_unsupported_image_fall_back(self): app = self.app() self.assertFalse(self.agent.seed_plutonium(app, IMAGE)) self.assertFalse((app / "runtime/plutonium/cdn_info.json").exists()) self.prepare() for image in ("example.invalid/plutainer:latest", "ghcr.io/ayymoss/plutainer-other:latest", "ghcr.io/attacker/plutainer:latest"): with self.subTest(image=image): self.assertFalse(self.agent.seed_plutonium(app, image)) (self.cache_content() / BOOTSTRAPPER).write_bytes(b"broken-cache") self.assertFalse(self.agent.seed_plutonium(app, IMAGE)) self.assertFalse((app / "runtime/plutonium/cdn_info.json").exists()) self.assertIsNone(self.agent.plutonium_cache_inventory()) def test_interrupted_seed_does_not_install_revision_marker_and_can_retry(self): manifest, contents, _, _ = self.prepare() app = self.app() marker = app / "runtime/plutonium/cdn_info.json" marker.parent.mkdir(parents=True) marker.write_text(json.dumps({"revision": 5353})) copy_file = module.managed_copy copied = [] def interrupted_copy(root, name, source, *args, **kwargs): self.assertFalse(marker.exists(), "revision marker must be installed only after all files") if copied: raise OSError("simulated full disk") copied.append(name) return copy_file(root, name, source, *args, **kwargs) with patch.object(module, "managed_copy", side_effect=interrupted_copy): with self.assertRaises((AgentError, OSError)): self.agent.seed_plutonium(app, IMAGE) self.assertFalse(marker.exists()) self.assertTrue(copied) self.assertTrue(self.agent.seed_plutonium(app, IMAGE)) self.assert_runtime(app, manifest, contents) def test_failed_seed_prevents_container_start(self): self.prepare() with patch.object(module, "managed_copy", side_effect=OSError("simulated full disk")): payload = {"game": "t5", "libraryId": self.library_id, "image": IMAGE, "port": 28960, "env": {}} result = self.agent.execute(self.operation("instance.create", payload, "failed-seed")) self.assertEqual(result["state"], "failed", result) self.assertFalse(any(call[:2] == ("container", "start") for call in self.docker.calls)) self.assertFalse((self.agent.instance_dir(fixtures.INSTANCE, 1) / "app/runtime/plutonium/cdn_info.json").exists()) def test_symlink_in_cache_cannot_seed_outside_data(self): _, contents, _, _ = self.prepare() cached = self.cache_content() / BOOTSTRAPPER outside = Path(self.temp.name) / "outside" outside.write_bytes(contents[BOOTSTRAPPER]) cached.unlink() cached.symlink_to(outside) app = self.app() self.assertFalse(self.agent.seed_plutonium(app, IMAGE)) self.assertFalse((app / "runtime/plutonium/cdn_info.json").exists()) def test_create_applies_initial_overlay_after_seed_and_starts_without_cache(self): self.create() self.assertTrue(self.agent.inspect(fixtures.INSTANCE, 1)["running"]) self.assertFalse((self.agent.instance_dir(fixtures.INSTANCE, 1) / "app/runtime/plutonium/cdn_info.json").exists()) manifest, contents, _, _ = self.prepare() payload = self.agent.spec(fixtures.INSTANCE, 1) payload["port"] += 1 payload["files"].append({"path": "runtime/plutonium/" + BOOTSTRAPPER, "contentBase64": base64.b64encode(b"custom-overlay").decode()}) other_id = "00000000-0000-4000-8000-000000000002" result = self.agent.execute({**self.operation("instance.create", payload, "second-create"), "instanceId": other_id}) self.assertEqual(result["state"], "completed", result) app = self.agent.instance_dir(other_id, 1) / "app" self.assertEqual((app / "runtime/plutonium" / BOOTSTRAPPER).read_bytes(), b"custom-overlay") self.assertEqual(json.loads((app / "runtime/plutonium/cdn_info.json").read_text()), manifest) self.assertEqual((app / "configs/server.cfg").read_bytes(), b"original") def test_rebuild_seeds_the_new_generation_from_current_cache(self): payload = self.create() manifest, contents, _, _ = self.prepare() self.agent.rcon = lambda *_: {"ok": True, "players": 0} result = self.agent.execute(self.operation("instance.rebuild", {**payload, "nextGeneration": 2}, "rebuild")) self.assertEqual(result["state"], "completed", result) self.assertTrue(self.agent.inspect(fixtures.INSTANCE, 2)["running"]) self.assert_runtime(self.agent.instance_dir(fixtures.INSTANCE, 2) / "app", manifest, contents) def test_restart_preserves_the_instances_downloaded_runtime(self): self.prepare() self.create() runtime_file = self.agent.instance_dir(fixtures.INSTANCE, 1) / "app/runtime/plutonium" / BOOTSTRAPPER runtime_file.write_bytes(b"container-updated-runtime") self.agent.rcon = lambda *_: {"ok": True, "players": 0} with patch.object(self.agent, "seed_plutonium", side_effect=AssertionError("Restart must retain its runtime")): result = self.agent.execute(self.operation("instance.restart", op_id="restart")) self.assertEqual(result["state"], "completed", result) self.assertEqual(runtime_file.read_bytes(), b"container-updated-runtime") if __name__ == "__main__": unittest.main()