Add native host management and role-based operator authentication
Introduce an outbound Python host agent, Debian installer, durable remote operations, immutable game-file libraries, host-aware deployments and ports, and occupancy-safe campaigns and draining. Native hosts do not require Komodo. Add administrator, operator and viewer accounts, protected management routes, secure sessions, password recovery, audit history and separate telemetry tokens. Document first-admin setup, reporter migration and live-host acceptance checks. Validation: 1,343 TypeScript tests and 39 agent tests passed, along with type checking, the production build, compiled SSR/API smoke tests and administrator recovery. Real Debian/Docker T5/T6 acceptance remains unrun.
This commit is contained in:
@@ -0,0 +1,547 @@
|
||||
import base64
|
||||
import importlib.util
|
||||
import io
|
||||
import json
|
||||
from pathlib import Path
|
||||
import tempfile
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
|
||||
spec = importlib.util.spec_from_file_location("ec_agent", Path(__file__).resolve().parents[1] / "ec_agent.py")
|
||||
module = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(module)
|
||||
Agent, AgentError, Held, Journal = module.Agent, module.AgentError, module.Held, module.Journal
|
||||
INSTANCE = "a53d99da-a1eb-4f01-9b10-a6268b556542"
|
||||
|
||||
|
||||
class FakeDocker:
|
||||
def __init__(self):
|
||||
self.containers = {}
|
||||
self.calls = []
|
||||
self.fail_restart = False
|
||||
self.lose_reply_after = None
|
||||
self.started = 0
|
||||
|
||||
def inspect(self, name):
|
||||
return self.containers.get(name)
|
||||
|
||||
def call(self, *args, **kwargs):
|
||||
self.calls.append(args)
|
||||
if args[0] == "info":
|
||||
return json.dumps({"ServerVersion": "27.5.1"})
|
||||
if args[:2] == ("image", "ls"):
|
||||
return "ghcr.io/ayymoss/plutainer:latest\n"
|
||||
if args[:2] == ("container", "create"):
|
||||
labels = {args[index + 1].split("=", 1)[0]: args[index + 1].split("=", 1)[1] for index, arg in enumerate(args) if arg == "--label"}
|
||||
name = args[args.index("--name") + 1]
|
||||
self.containers[name] = {"Id": "container-123", "Config": {"Labels": labels}, "State": {"Status": "created", "Running": False}}
|
||||
if args[:2] in (("container", "start"), ("container", "restart")):
|
||||
if self.fail_restart and args[1] == "restart":
|
||||
raise AgentError("Docker command unavailable or timed out")
|
||||
self.started += 1
|
||||
self.containers[args[-1]]["State"] = {"Status": "running", "Running": True, "StartedAt": f"started-{self.started}"}
|
||||
if args[:2] == ("container", "stop"):
|
||||
self.containers[args[-1]]["State"] = {"Status": "exited", "Running": False}
|
||||
if args[:2] == ("container", "rm"):
|
||||
self.containers.pop(args[-1])
|
||||
if args[0] == "container" and args[1] == self.lose_reply_after:
|
||||
raise AgentError("Docker command unavailable or timed out")
|
||||
return ""
|
||||
|
||||
|
||||
class Response(io.BytesIO):
|
||||
def __init__(self, content, status=200, headers=None):
|
||||
super().__init__(content)
|
||||
self.status = status
|
||||
self.headers = headers or {}
|
||||
|
||||
|
||||
class AgentTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.temp = tempfile.TemporaryDirectory()
|
||||
self.config = {"root": self.temp.name, "controller": "https://controller.example", "credential": "secret", "portRange": {"min": 28960, "max": 28980}}
|
||||
self.docker = FakeDocker()
|
||||
self.agent = Agent(self.config, self.docker)
|
||||
self.library_id = "library-one"
|
||||
self.library = self.agent.library_dir("t5", self.library_id)
|
||||
(self.library / "content").mkdir(parents=True)
|
||||
(self.library / "content" / "game.ff").write_bytes(b"game-data")
|
||||
files = [{"path": "game.ff", "size": 9, "sha256": module.digest(b"game-data")}]
|
||||
module.atomic_json(self.library / "manifest.json", {"libraryId": self.library_id, "game": "t5", "size": 9, "files": files, "manifestSha256": module.digest(json.dumps(files, separators=(",", ":")).encode())})
|
||||
|
||||
def tearDown(self):
|
||||
self.agent.journal.close()
|
||||
self.temp.cleanup()
|
||||
|
||||
def operation(self, kind, payload=None, op_id="operation-1", generation=1):
|
||||
return {"id": op_id, "kind": kind, "instanceId": INSTANCE, "generation": generation, "payload": payload or {}}
|
||||
|
||||
def create(self):
|
||||
payload = {"name": "t5_zombie_test_1", "game": "t5", "libraryId": self.library_id, "image": "ghcr.io/ayymoss/plutainer:latest", "port": 28960, "env": {"PLUTAINER_PORT": "28960"}, "rconPassword": "rcon-secret", "files": [{"path": "configs/server.cfg", "contentBase64": base64.b64encode(b"original").decode()}]}
|
||||
result = self.agent.execute(self.operation("instance.create", payload, "create"))
|
||||
self.assertEqual(result["state"], "completed", result)
|
||||
return payload
|
||||
|
||||
def test_same_id_reports_saved_result_without_executing_again(self):
|
||||
self.create()
|
||||
operation = self.operation("instance.restart")
|
||||
self.agent.rcon = lambda *_: {"ok": True, "players": 0}
|
||||
first = self.agent.execute(operation)
|
||||
self.assertEqual(first["state"], "completed")
|
||||
second = self.agent.execute({**operation, "state": "running", "updatedAt": "later", "result": None})
|
||||
self.assertEqual(second["result"], first["result"])
|
||||
self.assertEqual(sum(call[:2] == ("container", "restart") for call in self.docker.calls), 1)
|
||||
|
||||
def test_reusing_id_with_different_payload_is_rejected(self):
|
||||
self.agent.execute(self.operation("host.inspect"))
|
||||
with self.assertRaisesRegex(AgentError, "different content"):
|
||||
self.agent.execute(self.operation("host.inspect", {"changed": True}))
|
||||
|
||||
def test_agent_restart_marks_running_unknown_and_never_replays(self):
|
||||
operation = self.operation("instance.restart")
|
||||
self.agent.journal.accept(operation)
|
||||
self.agent.journal.update(operation["id"], "running")
|
||||
self.agent.journal.close()
|
||||
self.agent = Agent(self.config, self.docker)
|
||||
result = self.agent.execute(operation)
|
||||
self.assertEqual(result["state"], "unknown")
|
||||
self.assertEqual(self.docker.calls, [])
|
||||
|
||||
def test_occupied_and_unknown_wait_then_revalidate_same_operation(self):
|
||||
self.create()
|
||||
operation = self.operation("instance.restart")
|
||||
for players in (2, None, 7):
|
||||
self.agent.rcon = lambda *_, count=players: {"ok": True, "players": count}
|
||||
result = self.agent.execute(operation)
|
||||
self.assertEqual(result["state"], "accepted")
|
||||
self.assertFalse(any(call[:2] == ("container", "restart") for call in self.docker.calls))
|
||||
self.agent.rcon = lambda *_: {"ok": True, "players": 0}
|
||||
self.assertEqual(self.agent.execute(operation)["state"], "completed")
|
||||
|
||||
def test_unknown_container_health_cannot_disrupt(self):
|
||||
self.create()
|
||||
self.docker.containers[self.agent.container_name(INSTANCE, 1)]["State"] = {"Status": "unknown", "Running": False}
|
||||
self.agent.rcon = lambda *_: {"ok": True, "players": 0}
|
||||
self.assertEqual(self.agent.execute(self.operation("instance.stop"))["state"], "accepted")
|
||||
self.assertFalse(any(call[:2] == ("container", "stop") for call in self.docker.calls))
|
||||
|
||||
def test_files_stage_is_nondisruptive_activation_snapshots_and_rollback(self):
|
||||
self.create()
|
||||
target = self.agent.instance_dir(INSTANCE, 1) / "app/configs/server.cfg"
|
||||
staged = {"stageId": "campaign", "files": [{"id": 9, "path": "configs/server.cfg", "contentBase64": base64.b64encode(b"new").decode(), "sha256": module.digest(b"new"), "checkBefore": True, "expectedBeforeSha": module.digest(b"original")}]}
|
||||
self.agent.execute(self.operation("files.stage", staged, "stage"))
|
||||
self.assertEqual(target.read_bytes(), b"original")
|
||||
self.agent.rcon = lambda *_: {"ok": True, "players": 3}
|
||||
operation = self.operation("instance.restart", {"stageId": "campaign"})
|
||||
self.assertEqual(self.agent.execute(operation)["state"], "accepted")
|
||||
self.assertEqual(target.read_bytes(), b"original")
|
||||
self.agent.rcon = lambda *_: {"ok": True, "players": 0}
|
||||
result = json.loads(self.agent.execute(operation)["result"])
|
||||
self.assertEqual(base64.b64decode(result["before"][0]["contentBase64"]), b"original")
|
||||
self.assertEqual(target.read_bytes(), b"new")
|
||||
rollback = self.agent.execute(self.operation("files.rollback", {"stageId": "campaign", "restart": False}, "rollback"))
|
||||
self.assertEqual(rollback["state"], "completed")
|
||||
self.assertEqual(target.read_bytes(), b"original")
|
||||
|
||||
def test_changed_target_fails_before_activation_or_restart(self):
|
||||
self.create()
|
||||
target = self.agent.instance_dir(INSTANCE, 1) / "app/configs/server.cfg"
|
||||
self.agent.execute(self.operation("files.stage", {"stageId": "campaign", "files": [{"path": "configs/server.cfg", "contentBase64": base64.b64encode(b"new").decode(), "checkBefore": True, "expectedBeforeSha": "0" * 64}]}, "stage"))
|
||||
self.agent.rcon = lambda *_: {"ok": True, "players": 0}
|
||||
result = self.agent.execute(self.operation("instance.restart", {"stageId": "campaign"}))
|
||||
self.assertEqual(result["state"], "failed")
|
||||
self.assertEqual(target.read_bytes(), b"original")
|
||||
self.assertFalse(any(call[:2] == ("container", "restart") for call in self.docker.calls))
|
||||
|
||||
def test_ambiguous_restart_keeps_before_images_on_disk_and_is_not_replayed(self):
|
||||
self.create()
|
||||
self.agent.rcon = lambda *_: {"ok": True, "players": 0}
|
||||
self.agent.execute(self.operation("files.stage", {"stageId": "campaign", "files": [{"path": "configs/server.cfg", "contentBase64": base64.b64encode(b"new").decode()}]}, "stage"))
|
||||
self.docker.fail_restart = True
|
||||
operation = self.operation("instance.restart", {"stageId": "campaign"})
|
||||
self.assertEqual(self.agent.execute(operation)["state"], "unknown")
|
||||
self.agent.execute(operation)
|
||||
self.assertEqual(sum(call[:2] == ("container", "restart") for call in self.docker.calls), 1)
|
||||
saved = json.loads((self.agent.instance_dir(INSTANCE, 1) / "stages/campaign/manifest.json").read_text())
|
||||
self.assertEqual(saved["state"], "activated")
|
||||
self.assertEqual(base64.b64decode(saved["before"][0]["contentBase64"]), b"original")
|
||||
|
||||
def test_remove_only_own_instance_and_preserves_shared_library(self):
|
||||
self.create()
|
||||
self.agent.rcon = lambda *_: {"ok": True, "players": 0}
|
||||
result = self.agent.execute(self.operation("instance.remove"))
|
||||
self.assertEqual(result["state"], "completed")
|
||||
self.assertTrue((self.library / "content/game.ff").exists())
|
||||
self.assertFalse(self.agent.instance_dir(INSTANCE, 1).exists())
|
||||
|
||||
def test_unowned_container_never_mutated(self):
|
||||
self.create()
|
||||
self.docker.containers[self.agent.container_name(INSTANCE, 1)]["Config"]["Labels"] = {}
|
||||
self.assertEqual(self.agent.execute(self.operation("instance.remove"))["state"], "failed")
|
||||
self.assertFalse(any(call[:2] == ("container", "stop") for call in self.docker.calls))
|
||||
|
||||
def test_rebuild_moves_generation_and_rejects_old_commands(self):
|
||||
payload = self.create()
|
||||
self.agent.rcon = lambda *_: {"ok": True, "players": 0}
|
||||
result = self.agent.execute(self.operation("instance.rebuild", {**payload, "nextGeneration": 2}))
|
||||
self.assertEqual(result["state"], "completed", result)
|
||||
self.assertEqual(json.loads(result["result"])["generation"], 2)
|
||||
self.assertEqual(self.agent.execute(self.operation("instance.stop", op_id="old-stop"))["state"], "failed")
|
||||
self.assertTrue(self.agent.inspect(INSTANCE, 2)["running"])
|
||||
|
||||
def test_traversal_and_symlinks_are_rejected(self):
|
||||
self.create()
|
||||
app = self.agent.instance_dir(INSTANCE, 1) / "app"
|
||||
(app / "escape").symlink_to(self.temp.name)
|
||||
for path in ("../spec.json", "/etc/passwd", "escape/secret", "configs/../../spec.json"):
|
||||
with self.assertRaises(AgentError):
|
||||
module.scoped(app, path)
|
||||
|
||||
def manifest(self, content=b"complete-game-file"):
|
||||
files = [{"path": "zone/test.ff", "size": len(content), "sha256": module.digest(content)}]
|
||||
return {"id": "downloaded", "game": "t6", "files": files, "sizeBytes": len(content), "manifestSha256": module.digest(json.dumps(files, separators=(",", ":")).encode())}
|
||||
|
||||
def test_resume_verifies_and_atomically_publishes_library(self):
|
||||
content = b"complete-game-file"
|
||||
manifest = self.manifest(content)
|
||||
staging = self.agent.root / "libraries/t6/.downloaded.partial/content/zone"
|
||||
staging.mkdir(parents=True)
|
||||
(staging / "test.ff").write_bytes(content[:8])
|
||||
calls = []
|
||||
def fetch(method, path, data=None, extra_headers=None):
|
||||
calls.append((path, extra_headers))
|
||||
if path == "/manifest":
|
||||
return Response(json.dumps(manifest).encode())
|
||||
return Response(content[8:], 206, {"Content-Range": f"bytes 8-{len(content)-1}/{len(content)}"})
|
||||
self.agent.artifact_open = fetch
|
||||
result = self.agent.library_download({"libraryVersionId": "downloaded", "manifestUrl": "/manifest", "filesBaseUrl": "/files"})
|
||||
self.assertEqual(result["state"], "ready")
|
||||
self.assertEqual(calls[1][1], {"Range": "bytes=8-"})
|
||||
self.assertEqual((self.agent.library_dir("t6", "downloaded") / "content/zone/test.ff").read_bytes(), content)
|
||||
self.assertEqual(self.agent.inventory()["libraryCaches"][-1]["status"], "ready")
|
||||
|
||||
def test_corrupt_or_incomplete_library_never_promoted(self):
|
||||
manifest = self.manifest()
|
||||
def fetch(method, path, data=None, extra_headers=None):
|
||||
return Response(json.dumps(manifest).encode() if path == "/manifest" else b"corrupt")
|
||||
self.agent.artifact_open = fetch
|
||||
with self.assertRaisesRegex(AgentError, "integrity"):
|
||||
self.agent.library_download({"libraryVersionId": "downloaded", "manifestUrl": "/manifest", "filesBaseUrl": "/files"})
|
||||
self.assertFalse(self.agent.library_dir("t6", "downloaded").exists())
|
||||
self.assertEqual(self.agent.library_progress["downloaded"]["status"], "failed")
|
||||
|
||||
def test_manifest_tampering_rejected_before_download(self):
|
||||
manifest = self.manifest()
|
||||
manifest["files"][0]["size"] = 999
|
||||
self.agent.artifact_open = lambda *_: Response(json.dumps(manifest).encode())
|
||||
with self.assertRaisesRegex(AgentError, "manifest checksum"):
|
||||
self.agent.library_download({"libraryVersionId": "downloaded", "manifestUrl": "/manifest", "filesBaseUrl": "/files"})
|
||||
|
||||
def test_artifact_urls_never_leak_credentials_to_other_origins(self):
|
||||
for value in ("https://attacker.example/files", "//attacker.example/files", "http://controller.example/files", "https://user:pass@controller.example/files"):
|
||||
with self.assertRaises(AgentError):
|
||||
self.agent.artifact_url(value)
|
||||
|
||||
def test_unsupported_operation_does_not_spawn_shell(self):
|
||||
result = self.agent.execute(self.operation("shell.execute", {"command": "touch /tmp/unsafe"}))
|
||||
self.assertEqual(result["state"], "failed")
|
||||
self.assertEqual(self.docker.calls, [])
|
||||
|
||||
def test_restart_lost_acknowledgement_recovers_from_changed_start_timestamp(self):
|
||||
self.create()
|
||||
self.agent.rcon = lambda *_: {"ok": True, "players": 0}
|
||||
self.docker.lose_reply_after = "restart"
|
||||
operation = self.operation("instance.restart")
|
||||
self.assertEqual(self.agent.execute(operation)["state"], "unknown")
|
||||
self.agent.journal.close()
|
||||
self.agent = Agent(self.config, self.docker)
|
||||
recovered = self.agent.execute(operation)
|
||||
self.assertEqual(recovered["state"], "completed")
|
||||
self.assertTrue(json.loads(recovered["result"])["reconciled"])
|
||||
self.assertEqual(sum(call[:2] == ("container", "restart") for call in self.docker.calls), 1)
|
||||
|
||||
def test_created_container_lost_start_reply_reconciles_immutable_operation_label(self):
|
||||
payload = {"name": "t5_zombie_test_1", "game": "t5", "libraryId": self.library_id, "image": "ghcr.io/ayymoss/plutainer:latest", "port": 28960, "env": {}, "rconPassword": "secret", "files": []}
|
||||
self.docker.lose_reply_after = "start"
|
||||
operation = self.operation("instance.create", payload)
|
||||
self.assertEqual(self.agent.execute(operation)["state"], "unknown")
|
||||
result = self.agent.execute(operation)
|
||||
self.assertEqual(result["state"], "completed")
|
||||
self.assertEqual(sum(call[:2] == ("container", "create") for call in self.docker.calls), 1)
|
||||
|
||||
def test_remove_lost_reply_reconciles_absence_without_duplicate_delete(self):
|
||||
self.create()
|
||||
self.agent.rcon = lambda *_: {"ok": True, "players": 0}
|
||||
self.docker.lose_reply_after = "rm"
|
||||
operation = self.operation("instance.remove")
|
||||
self.assertEqual(self.agent.execute(operation)["state"], "unknown")
|
||||
self.assertEqual(self.agent.execute(operation)["state"], "completed")
|
||||
self.assertFalse(self.agent.instance_dir(INSTANCE, 1).exists())
|
||||
self.assertEqual(sum(call[:2] == ("container", "rm") for call in self.docker.calls), 1)
|
||||
|
||||
def test_rotation_recovers_lost_reply_using_durably_saved_pending_credential(self):
|
||||
self.agent.config_path = str(Path(self.temp.name) / "agent.json")
|
||||
accepted = {}
|
||||
def request(method, path, payload):
|
||||
if not accepted:
|
||||
accepted["credential"] = payload["credential"]
|
||||
persisted = json.loads(Path(self.agent.config_path).read_text())
|
||||
self.assertEqual(persisted["pendingCredential"], payload["credential"])
|
||||
raise AgentError("Controller request failed")
|
||||
self.assertEqual(self.agent.config["credential"], accepted["credential"])
|
||||
return {"ok": True}
|
||||
self.agent.request = request
|
||||
result = self.agent.execute(self.operation("credentials.rotate"))
|
||||
self.assertEqual(result["state"], "completed")
|
||||
saved = json.loads(Path(self.agent.config_path).read_text())
|
||||
self.assertNotIn("pendingCredential", saved)
|
||||
self.assertEqual(saved["credential"], accepted["credential"])
|
||||
|
||||
def test_status_parser_never_treats_partial_or_unfamiliar_rows_as_empty(self):
|
||||
header = "map: zombie_cosmodrome\nnum score ping guid name\n"
|
||||
self.assertIsNone(module.status_players(header))
|
||||
self.assertEqual(module.status_players(header + "--- ----- ---- ---- ----\n"), 0)
|
||||
for row in ("0 0 CNCT abc player", "1 90 ZMBI abc player", "2 4 ???? abc player"):
|
||||
self.assertEqual(module.status_players(header + "--- ----- ---- ---- ----\n" + row), 1)
|
||||
self.assertIsNone(module.status_players(header + "--- ----- ---- ---- ----\nconnection in progress"))
|
||||
self.assertEqual(module.status_players('statusResponse\n\\mapname\\mp_kino\\sv_maxclients\\4\n10 45 "joe"'), 1)
|
||||
|
||||
def test_udp_rcon_uses_host_local_endpoint_and_returns_structured_occupancy(self):
|
||||
self.create()
|
||||
class FakeSocket:
|
||||
def __enter__(self): return self
|
||||
def __exit__(self, *_): pass
|
||||
def settimeout(self, value): pass
|
||||
def connect(self, endpoint): self.endpoint = endpoint
|
||||
def send(self, packet): self.packet = packet
|
||||
def recv(self, _size):
|
||||
if getattr(self, "done", False): raise module.socket.timeout()
|
||||
self.done = True
|
||||
return b"\xff\xff\xff\xffprint\nmap: mp_kino\nnum score ping guid name\n--- ----- ---- ---- ----\n0 42 CNCT abc player\n"
|
||||
sock = FakeSocket()
|
||||
with patch.object(module.socket, "socket", return_value=sock):
|
||||
result = self.agent.rcon(INSTANCE, 1, "status")
|
||||
self.assertEqual(sock.endpoint, ("127.0.0.1", 28960))
|
||||
self.assertTrue(sock.packet.startswith(b"\xff\xff\xff\xffrcon rcon-secret status"))
|
||||
self.assertEqual(result["players"], 1)
|
||||
self.assertTrue(result["ok"])
|
||||
|
||||
def test_invalid_rebuild_preflight_preserves_running_original(self):
|
||||
payload = self.create()
|
||||
result = self.agent.execute(self.operation("instance.rebuild", {**payload, "image": "--privileged", "nextGeneration": 2}))
|
||||
self.assertEqual(result["state"], "failed")
|
||||
self.assertTrue(self.agent.inspect(INSTANCE, 1)["running"])
|
||||
self.assertFalse(any(call[:2] == ("container", "stop") for call in self.docker.calls))
|
||||
|
||||
def test_file_operation_resists_parent_symlink_swap_after_validation(self):
|
||||
self.create()
|
||||
app = self.agent.instance_dir(INSTANCE, 1) / "app"
|
||||
outside = Path(self.temp.name) / "outside"
|
||||
outside.mkdir()
|
||||
(outside / "server.cfg").write_bytes(b"host-data")
|
||||
original = module.scoped
|
||||
def swap(root, path, **kwargs):
|
||||
result = original(root, path, **kwargs)
|
||||
(app / "configs").rename(app / "saved-configs")
|
||||
(app / "configs").symlink_to(outside)
|
||||
return result
|
||||
with patch.object(module, "scoped", side_effect=swap):
|
||||
with self.assertRaises(OSError):
|
||||
module.managed_write(app, "configs/server.cfg", b"escape-attempt")
|
||||
self.assertEqual((outside / "server.cfg").read_bytes(), b"host-data")
|
||||
|
||||
def test_deleting_absent_file_with_absent_parent_is_idempotent(self):
|
||||
self.create()
|
||||
app = self.agent.instance_dir(INSTANCE, 1) / "app"
|
||||
module.managed_write(app, "missing/subdirectory/file.gsc", None)
|
||||
self.assertFalse((app / "missing").exists())
|
||||
|
||||
def test_controller_contact_failure_holds_before_running_any_docker_command(self):
|
||||
self.create()
|
||||
count = len(self.docker.calls)
|
||||
self.agent.report = lambda *_: (_ for _ in ()).throw(AgentError("Controller request failed"))
|
||||
result = self.agent.execute(self.operation("instance.stop"), report_running=True)
|
||||
self.assertEqual(result["state"], "accepted")
|
||||
self.assertEqual(len(self.docker.calls), count)
|
||||
|
||||
def test_async_import_publication_resumes_after_controller_restart(self):
|
||||
responses = iter([{"status": "verifying"}, {"status": "pending"}, {"status": "verifying"}, {"status": "published", "version": {"id": "library-one"}}])
|
||||
methods = []
|
||||
def request(method, path, *args):
|
||||
methods.append(method)
|
||||
return Response(json.dumps(next(responses)).encode())
|
||||
self.agent.artifact_open = request
|
||||
with patch.object(self.agent.stopping, "wait", return_value=False):
|
||||
result = self.agent.complete_import({"completeUrl": "/complete"}, start=True)
|
||||
self.assertEqual(result["status"], "published")
|
||||
self.assertEqual(methods, ["POST", "GET", "POST", "GET"])
|
||||
|
||||
def test_digest_pinned_runtime_inventory_and_exact_aliases(self):
|
||||
image_digest = "sha256:" + "1" * 64
|
||||
image_id = "sha256:" + "2" * 64
|
||||
original = self.docker.call
|
||||
def call(*args, **kwargs):
|
||||
if args[:2] == ("image", "ls"):
|
||||
return f"ghcr.io/ayymoss/plutainer|latest|{image_digest}|{image_id}\n"
|
||||
return original(*args, **kwargs)
|
||||
self.docker.call = call
|
||||
module.atomic_json(self.agent.root / "agent/runtime-images.json", {"ghcr.io/ayymoss/plutainer": image_id, "removed-image": "missing-id"})
|
||||
images = self.agent.inventory()["runtimeImages"]
|
||||
self.assertIn("ghcr.io/ayymoss/plutainer@" + image_digest, images)
|
||||
self.assertIn("ghcr.io/ayymoss/plutainer:latest", images)
|
||||
self.assertIn("ghcr.io/ayymoss/plutainer", images)
|
||||
self.assertNotIn("removed-image", images)
|
||||
|
||||
def test_heartbeat_applies_configured_host_port_range_before_deployment(self):
|
||||
self.agent.request = lambda *_: {"host": {"portMin": 30000, "portMax": 30100}}
|
||||
self.agent.heartbeat()
|
||||
self.assertEqual(self.agent.config["portRange"], {"min": 30000, "max": 30100})
|
||||
|
||||
def test_download_content_permissions_allow_container_read_access_with_private_umask(self):
|
||||
content = b"complete-game-file"
|
||||
manifest = self.manifest(content)
|
||||
self.agent.artifact_open = lambda method, path, **kwargs: Response(json.dumps(manifest).encode() if path == "/manifest" else content)
|
||||
previous = module.os.umask(0o077)
|
||||
try:
|
||||
self.agent.library_download({"libraryVersionId": "downloaded", "manifestUrl": "/manifest", "filesBaseUrl": "/files"})
|
||||
finally:
|
||||
module.os.umask(previous)
|
||||
content_root = self.agent.library_dir("t6", "downloaded") / "content"
|
||||
self.assertEqual(content_root.stat().st_mode & 0o777, 0o555)
|
||||
self.assertEqual((content_root / "zone").stat().st_mode & 0o777, 0o555)
|
||||
self.assertEqual((content_root / "zone/test.ff").stat().st_mode & 0o777, 0o444)
|
||||
|
||||
def test_agent_restart_resumes_same_immutable_library_operation(self):
|
||||
content = b"complete-game-file"
|
||||
manifest = self.manifest(content)
|
||||
payload = {"libraryVersionId": "downloaded", "manifestUrl": "/manifest", "filesBaseUrl": "/files", "manifestSha256": manifest["manifestSha256"]}
|
||||
operation = self.operation("library.download", payload)
|
||||
self.agent.journal.accept(operation)
|
||||
self.agent.journal.update(operation["id"], "running")
|
||||
staging = self.agent.root / "libraries/t6/.downloaded.partial/content/zone"
|
||||
staging.mkdir(parents=True)
|
||||
(staging / "test.ff").write_bytes(content[:8])
|
||||
self.agent.journal.close()
|
||||
self.agent = Agent(self.config, self.docker)
|
||||
def fetch(method, path, data=None, extra_headers=None):
|
||||
if path == "/manifest":
|
||||
return Response(json.dumps(manifest).encode())
|
||||
self.assertEqual(extra_headers, {"Range": "bytes=8-"})
|
||||
return Response(content[8:], 206, {"Content-Range": f"bytes 8-{len(content)-1}/{len(content)}"})
|
||||
self.agent.artifact_open = fetch
|
||||
result = self.agent.execute(operation)
|
||||
self.assertEqual(result["state"], "completed")
|
||||
self.assertEqual((self.agent.library_dir("t6", "downloaded") / "content/zone/test.ff").read_bytes(), content)
|
||||
|
||||
def test_player_join_during_activation_holds_restart_without_reapplying_files(self):
|
||||
self.create()
|
||||
target = self.agent.instance_dir(INSTANCE, 1) / "app/configs/server.cfg"
|
||||
original_write = module.managed_write
|
||||
for index, kind in enumerate(("instance.restart", "files.activate", "instance.start")):
|
||||
with self.subTest(kind=kind):
|
||||
before = target.read_bytes()
|
||||
stage_id = "late-join-" + str(index)
|
||||
content = ("new-" + str(index)).encode()
|
||||
self.agent.execute(self.operation("files.stage", {"stageId": stage_id, "files": [{"id": index, "path": "configs/server.cfg", "contentBase64": base64.b64encode(content).decode()}]}, "stage-late-" + str(index)))
|
||||
joined, writes = [False], []
|
||||
self.agent.rcon = lambda *_: {"ok": True, "players": 1 if joined[0] else 0}
|
||||
def write_then_join(*args, **kwargs):
|
||||
original_write(*args, **kwargs)
|
||||
writes.append(args)
|
||||
joined[0] = True
|
||||
operation = self.operation(kind, {"stageId": stage_id}, "action-late-" + str(index))
|
||||
restarts_before = sum(call[:2] == ("container", "restart") for call in self.docker.calls)
|
||||
with patch.object(module, "managed_write", side_effect=write_then_join):
|
||||
held = self.agent.execute(operation)
|
||||
self.assertEqual(held["state"], "accepted")
|
||||
self.assertEqual(json.loads(held["result"])["blocked"], "occupied")
|
||||
self.assertEqual(target.read_bytes(), content)
|
||||
self.assertEqual(sum(call[:2] == ("container", "restart") for call in self.docker.calls), restarts_before)
|
||||
joined[0] = False
|
||||
completed = self.agent.execute(operation)
|
||||
self.assertEqual(completed["state"], "completed")
|
||||
self.assertEqual(len(writes), 1)
|
||||
self.assertEqual(base64.b64decode(json.loads(completed["result"])["before"][0]["contentBase64"]), before)
|
||||
self.assertEqual(sum(call[:2] == ("container", "restart") for call in self.docker.calls), restarts_before + 1)
|
||||
|
||||
def test_player_join_during_rollback_holds_restart_without_repeating_restore(self):
|
||||
self.create()
|
||||
self.agent.rcon = lambda *_: {"ok": True, "players": 0}
|
||||
self.agent.execute(self.operation("files.stage", {"stageId": "rollback-join", "files": [{"path": "configs/server.cfg", "contentBase64": base64.b64encode(b"new").decode()}]}, "stage-rollback-join"))
|
||||
self.agent.execute(self.operation("files.activate", {"stageId": "rollback-join", "restart": False}, "activate-rollback-join"))
|
||||
joined, writes = [False], []
|
||||
original_write = module.managed_write
|
||||
self.agent.rcon = lambda *_: {"ok": True, "players": 1 if joined[0] else 0}
|
||||
def restore_then_join(*args, **kwargs):
|
||||
original_write(*args, **kwargs)
|
||||
writes.append(args)
|
||||
joined[0] = True
|
||||
operation = self.operation("files.rollback", {"stageId": "rollback-join"}, "rollback-join")
|
||||
with patch.object(module, "managed_write", side_effect=restore_then_join):
|
||||
self.assertEqual(self.agent.execute(operation)["state"], "accepted")
|
||||
self.assertFalse(any(call[:2] == ("container", "restart") for call in self.docker.calls))
|
||||
joined[0] = False
|
||||
self.assertEqual(self.agent.execute(operation)["state"], "completed")
|
||||
self.assertEqual(len(writes), 1)
|
||||
self.assertEqual((self.agent.instance_dir(INSTANCE, 1) / "app/configs/server.cfg").read_bytes(), b"original")
|
||||
|
||||
def test_start_with_stage_does_not_reconcile_success_before_required_restart(self):
|
||||
self.create()
|
||||
self.agent.rcon = lambda *_: {"ok": True, "players": 0}
|
||||
self.agent.execute(self.operation("files.stage", {"stageId": "start-stage", "files": [{"path": "configs/server.cfg", "contentBase64": base64.b64encode(b"new").decode()}]}, "stage-start"))
|
||||
self.docker.fail_restart = True
|
||||
operation = self.operation("instance.start", {"stageId": "start-stage"})
|
||||
self.assertEqual(self.agent.execute(operation)["state"], "unknown")
|
||||
self.assertEqual(self.agent.execute(operation)["state"], "unknown")
|
||||
self.assertEqual(sum(call[:2] == ("container", "restart") for call in self.docker.calls), 1)
|
||||
|
||||
def test_running_unhealthy_container_uses_fresh_occupancy_before_recovery(self):
|
||||
self.create()
|
||||
self.docker.containers[self.agent.container_name(INSTANCE, 1)]["State"]["Health"] = {"Status": "unhealthy"}
|
||||
operation = self.operation("instance.restart")
|
||||
self.agent.rcon = lambda *_: {"ok": True, "players": 2}
|
||||
self.assertEqual(self.agent.execute(operation)["state"], "accepted")
|
||||
self.agent.rcon = lambda *_: {"ok": True, "players": 0}
|
||||
self.assertEqual(self.agent.execute(operation)["state"], "completed")
|
||||
|
||||
def test_game_log_can_follow_internal_relative_plutainer_symlink(self):
|
||||
self.create()
|
||||
app = self.agent.instance_dir(INSTANCE, 1) / "app"
|
||||
(app / "logs").mkdir()
|
||||
(app / "runtime").mkdir()
|
||||
(app / "runtime/game.log").write_text("first\nsecond\nthird\n")
|
||||
(app / "logs/games_sp.log").symlink_to("../runtime/game.log")
|
||||
result = self.agent.execute(self.operation("logs.read", {"path": "logs/games_sp.log", "lines": 2}))
|
||||
self.assertEqual(result["state"], "completed")
|
||||
self.assertEqual(json.loads(result["result"])["lines"], ["second", "third"])
|
||||
with self.assertRaises(AgentError):
|
||||
module.managed_write(app, "logs/games_sp.log", b"must-not-write-through-link")
|
||||
|
||||
def test_game_log_rejects_symlink_escaping_instance(self):
|
||||
self.create()
|
||||
app = self.agent.instance_dir(INSTANCE, 1) / "app"
|
||||
(app / "logs").mkdir()
|
||||
outside = Path(self.temp.name) / "host-secret"
|
||||
outside.write_text("secret")
|
||||
(app / "logs/games_sp.log").symlink_to(outside)
|
||||
result = self.agent.execute(self.operation("logs.read", {"path": "logs/games_sp.log"}))
|
||||
self.assertEqual(result["state"], "failed")
|
||||
self.assertIn("outside", result["error"])
|
||||
|
||||
def test_console_log_falls_back_to_owned_docker_output_including_stderr(self):
|
||||
self.create()
|
||||
original = self.docker.call
|
||||
def call(*args, **kwargs):
|
||||
if args[:2] == ("container", "logs"):
|
||||
self.assertTrue(kwargs["merge_stderr"])
|
||||
self.assertEqual(args[-1], self.agent.container_name(INSTANCE, 1))
|
||||
return "stdout-line\nstderr-line\n"
|
||||
return original(*args, **kwargs)
|
||||
self.docker.call = call
|
||||
result = self.agent.execute(self.operation("logs.read", {"path": "logs/console.log"}))
|
||||
self.assertEqual(result["state"], "completed")
|
||||
self.assertEqual(json.loads(result["result"])["lines"], ["stdout-line", "stderr-line"])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user