Provision instance bank credentials automatically (Refs #146)

agent/ec_agent.py

@@ -35,7 +35,7 @@ import urllib.parse
 import urllib.request
 import uuid
 
-VERSION = "1.8.7"
+VERSION = "1.9.0"
 LABEL = "io.electric-cherry"
 MAX_FILE = 32 * 1024 * 1024
 MANIFEST_MAX_ENTRIES = 10000
@@ -1503,7 +1503,7 @@ class Agent:
                 if line.startswith("PRETTY_NAME="):
                     os_name = line.split("=", 1)[1].strip('"')
         update = self.update_inventory()
-        capabilities = ["library.remove", "plutonium.cache", "instance.cancel-create", "instance.safety-override", "instance.unhealthy-recovery", "instance.autoscaling-remove", "operations.read-lane", "operations.health-lane", "rcon.status-batch", "instance.heartbeat-state", "files.manifest", "files.blobs", "logs.read-offset"] + (["host.decommission"] if self.can_uninstall() else []) + (["agent.update"] if update["enabled"] else [])
+        capabilities = ["library.remove", "plutonium.cache", "instance.cancel-create", "instance.safety-override", "instance.unhealthy-recovery", "instance.autoscaling-remove", "operations.read-lane", "operations.health-lane", "rcon.status-batch", "instance.heartbeat-state", "files.manifest", "files.blobs", "logs.read-offset", "bank.settlement"] + (["host.decommission"] if self.can_uninstall() else []) + (["agent.update"] if update["enabled"] else [])
         plutonium_cache = self.plutonium_cache_inventory()
         return {"hostname": socket.gethostname(), "agentVersion": VERSION, "protocolVersion": 1, "capabilities": capabilities, "agentUpdate": update, "agentLanes": self.lane_inventory(), "os": os_name, "osVersion": platform.release(), "architecture": platform.machine(), "cpu": platform.processor() or platform.machine(), "cpuCount": os.cpu_count() or 0, **resources, **({"cpuUsagePercent": cpu_usage} if cpu_usage is not None else {}), "memoryTotal": memory.get("MemTotal", 0), "memoryAvailable": memory.get("MemAvailable", 0), "storageTotal": usage.total, "storageAvailable": usage.free, "dockerVersion": docker_version, "dockerRunning": docker_running, "directoriesWritable": os.access(self.root, os.W_OK), "runtimeImages": sorted(set(images)), "preparedRuntimeImages": sorted(set(prepared_images)), **({"plutoniumCache": plutonium_cache} if plutonium_cache else {}), "libraries": libraries, "libraryCaches": caches, "currentOperation": self.current_operation, "portRange": self.config.get("portRange", {"min": 28960, "max": 29060})}
 
@@ -1657,7 +1657,7 @@ class Agent:
         # record. Whenever preparation was recorded, enforce that exact image.
         if image in prepared and prepared[image] != image_id:
             raise AgentError("Prepared runtime image has changed. An admin must prepare it again before launching.")
-        files = payload.get("files", [])
+        files = self.initial_files(payload)
         if not isinstance(files, list) or len(files) > 10000:
             raise Rejected("Invalid initial file list")
         total, paths = 0, set()
@@ -1911,7 +1911,7 @@ class Agent:
         self.operation_phase = "seeding-runtime"
         self.seed_plutonium(app, image)
         self.operation_phase = "writing-create-files"
-        self.write_initial_files(app, payload.get("files", []))
+        self.write_initial_files(app, self.initial_files(payload))
         self.operation_phase = "persisting-create-metadata"
         atomic_json(directory / "spec.json", payload)
         env_path = directory / "environment"
@@ -1944,6 +1944,25 @@ class Agent:
         if self.executing_id and self.executing_id in self.cancelled_creations:
             raise AgentError("Creation cancelled; waiting for instance cleanup")
 
+    def initial_files(self, payload):
+        files = payload.get("files", [])
+        if not isinstance(files, list):
+            raise Rejected("Invalid initial file list")
+        bank = payload.get("bank")
+        if bank is None:
+            return files
+        if not isinstance(bank, dict) or not isinstance(bank.get("token"), str) or not re.fullmatch(r"ec_bank_[A-Za-z0-9_-]{32,128}", bank["token"]):
+            raise Rejected("Invalid instance bank credential")
+        path = bank.get("configPath")
+        scoped(self.root / "agent" / "validation", path)
+        if PurePosixPath(path).name != "electric_cherry_bank.cfg":
+            raise Rejected("Invalid private bank config path")
+        controller = self.config["controller"].rstrip("/")
+        if any(char in controller for char in ('"', "\\", "\n", "\r", "\x00")):
+            raise Rejected("Invalid bank controller URL")
+        content = f'set tws_bank_token "{bank["token"]}"\nset tws_bank_url "{controller}/api/bank"\n'.encode()
+        return files + [{"path": path, "contentBase64": base64.b64encode(content).decode()}]
+
     def write_initial_files(self, root, files):
         if not isinstance(files, list) or len(files) > 10000:
             raise AgentError("Invalid file list")
@@ -1992,6 +2011,72 @@ class Agent:
         if (time.monotonic() - started) * 1000 >= ttl:
             raise AutoscalingInvalidated("Autoscaling removal authorization expired before mutation")
 
+    def settle_bank(self, instance_id, generation, operation_id=None):
+        """Drain stopped runtime journals before discarding any generation's storage.
+
+        Use only our configured controller and host credential. Never follow an
+        endpoint or symlink supplied by a game journal. Lost replies are retried
+        with the original request identity; files remain until teardown succeeds.
+        """
+        app = scoped(self.instance_dir(instance_id, generation), "app")
+        if not app.exists():
+            return
+        if self.inspect(instance_id, generation)["running"]:
+            raise AgentError("Bank settlement requires a stopped instance")
+        operation_id = operation_id or self.executing_id
+        visited = 0
+        deadline = time.monotonic() + 30
+        journals = []
+
+        def unreadable(_error):
+            raise AgentError("Bank journal discovery failed; runtime storage retained")
+
+        for directory, directories, files in os.walk(app, followlinks=False, onerror=unreadable):
+            relative = Path(directory).relative_to(app)
+            inside_bank = "tws-bank" in relative.parts
+            visited += 1
+            if visited > 100000 or time.monotonic() > deadline:
+                raise AgentError("Bank journal discovery limit exceeded; runtime storage retained")
+            for name in directories + files:
+                if (inside_bank or name == "tws-bank") and (Path(directory) / name).is_symlink():
+                    raise AgentError("Bank journal paths must not be symlinks")
+            if not inside_bank:
+                continue
+            for name in files:
+                if name.endswith(".json"):
+                    journals.append((relative / name).as_posix())
+                    if len(journals) > 4096:
+                        raise AgentError("Too many bank journals; runtime storage retained")
+        for path in sorted(journals):
+            try:
+                raw = managed_read(app, path)
+            except Rejected:
+                raise AgentError("Bank journal cannot be read; runtime storage retained") from None
+            if raw is None or len(raw) > 16 * 1024 * 1024:
+                raise AgentError("Bank journal is unavailable or oversized")
+            try:
+                journal = json.loads(raw)
+                reports = journal["reports"]
+                if journal["endpoint"].rstrip("/") != self.config["controller"].rstrip("/") + "/api/bank":
+                    raise AgentError("Bank journal belongs to a different controller; runtime storage retained")
+                if type(journal["version"]) is not int or journal["version"] != 1 or not isinstance(reports, list) or len(reports) > 4096:
+                    raise ValueError()
+                # Controller independently binds this operation to the instance/server.
+                for report in reports:
+                    if not isinstance(report, str) or len(report.encode()) > 20000:
+                        raise ValueError()
+                    transfer = json.loads(report)
+                    action = transfer["decision"]
+                    if action not in ("commit", "cancel") or transfer["server"] != journal["server"] or not operation_id:
+                        raise ValueError()
+                    response = self.request("POST", "/api/agent/bank/settle", {
+                        "operationId": operation_id, "action": action, "transfer": transfer})
+                    expected = ("committed",) if action == "commit" else ("cancelled", "rejected")
+                    if not isinstance(response, dict) or response.get("request_id") != transfer["request_id"] or response.get("state") not in expected:
+                        raise AgentError("Bank settlement was not acknowledged; runtime storage retained")
+            except (ValueError, TypeError, KeyError, AttributeError):
+                raise AgentError("Bank journal is invalid; runtime storage retained") from None
+
     def lifecycle(self, kind, instance_id, generation, payload):
         unhealthy_recovery = payload.get("unhealthyRecovery") is True
         if unhealthy_recovery and (kind not in ("instance.restart", "instance.rebuild") or payload.get("stageId") or "autoscaling" in payload):
@@ -2043,6 +2128,7 @@ class Agent:
                 if health["exists"]:
                     self.docker.call("container", "stop", "--time", "30", name)
                     self.docker.call("container", "rm", name)
+                self.settle_bank(instance_id, generation)
                 if kind == "instance.rebuild":
                     result = self.create(instance_id, next_generation, spec)
                     self.operation_phase = "removing-old-instance"
@@ -2062,7 +2148,7 @@ class Agent:
             raise AgentError("Unsupported instance operation")
         return {**self.inspect(instance_id, generation), "before": before}
 
-    def cancel_creation(self, instance_id, generation, payload):
+    def cancel_creation(self, instance_id, generation, payload, operation_id=None):
         """Explicit launch cancellation, fenced to its immutable creation identity.
 
         The operation loop serializes this with create. Once delivered, removal
@@ -2106,6 +2192,7 @@ class Agent:
                 self.docker.call("container", "rm", "--force", "--volumes", self.container_name(instance_id, target_generation))
             if self.inspect(instance_id, target_generation)["exists"]:
                 raise AgentError("Cancelled creation container still exists")
+            self.settle_bank(instance_id, target_generation, operation_id)
             directory = self.instance_dir(instance_id, target_generation)
             if directory.exists():
                 shutil.rmtree(directory)
@@ -2747,6 +2834,7 @@ class Agent:
 
         if not unchanged_generation() or not both_missing():
             return row
+        self.settle_bank(instance_id, generation, operation["id"])
         # Preparation can leave runtime/configuration files before create's
         # checkpoint. A fresh operation must not inherit those partial files.
         expected_spec = payload.get("spec") or {key: value for key, value in payload.items() if key not in ("nextGeneration", "newGeneration", "unhealthyRecovery")}
@@ -2793,7 +2881,7 @@ class Agent:
             if kind == "instance.remove" and "autoscaling" in payload and not checkpoint:
                 return self.journal.update(operation["id"], "failed", {"autoscalingInvalidated": True}, error="Interrupted autoscaling removal requires a new decision")
             if kind == "instance.cancel-create":
-                return self.journal.update(operation["id"], "completed", self.cancel_creation(instance_id, generation, payload))
+                return self.journal.update(operation["id"], "completed", self.cancel_creation(instance_id, generation, payload, operation["id"]))
             if kind == "instance.create":
                 marker = self.instance_dir(instance_id, generation).parent / "current.json"
                 if marker.exists():
@@ -2885,6 +2973,7 @@ class Agent:
                     directory = self.instance_dir(instance_id, observed_generation)
                     atomic_json(directory.parent / "current.json", {"generation": observed_generation})
                     if kind == "instance.rebuild" and not self.inspect(instance_id, generation)["exists"]:
+                        self.settle_bank(instance_id, generation, operation["id"])
                         old = self.instance_dir(instance_id, generation)
                         if old.exists():
                             shutil.rmtree(old)
@@ -2899,6 +2988,7 @@ class Agent:
             elif kind == "instance.remove":
                 proven = not health["exists"]
                 if proven:
+                    self.settle_bank(instance_id, generation, operation["id"])
                     directory = self.instance_dir(instance_id, generation)
                     if directory.exists():
                         shutil.rmtree(directory)

agent/tests/test_agent.py

@@ -1644,6 +1644,109 @@ class AgentTests(unittest.TestCase):
         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_create_materializes_private_bank_config_from_instance_credential(self):
+        secret = "ec_bank_" + "a" * 43
+        payload = {"name": "server", "game": "t5", "libraryId": self.library_id,
+                   "image": "ghcr.io/ayymoss/plutainer:latest", "port": 28960,
+                   "bank": {"token": secret, "configPath": "configs/electric_cherry_bank.cfg"}}
+        operation = self.operation("instance.create", payload, "create")
+        self.assertEqual(self.agent.execute(operation)["state"], "completed")
+        path = self.agent.instance_dir(INSTANCE, 1) / "app/configs/electric_cherry_bank.cfg"
+        self.assertEqual(path.read_text(), f'set tws_bank_token "{secret}"\nset tws_bank_url "https://controller.example/api/bank"\n')
+        self.assertIn("bank.settlement", self.agent.inventory()["capabilities"])
+        payload["bank"]["token"] = secret + '\nset bad "1"'
+        with self.assertRaises(AgentError):
+            self.agent.initial_files(payload)
+
+    def bank_journal(self, decisions=("commit", "cancel"), game="t5"):
+        path = self.agent.instance_dir(INSTANCE, 1) / f"app/runtime/tws-bank/{game}/scope.json"
+        reports = [json.dumps({"server": "t5_zombie_test_1", "guid": "ace", "match": "match-one", "request_id": f"{game}-{i}", "operation": "deposit", "amount": 2000, "capacity": 1000000, "decision": decision}) for i, decision in enumerate(decisions)]
+        module.atomic_json(path, {"version": 1, "server": "t5_zombie_test_1", "endpoint": "https://controller.example/api/bank", "reports": reports})
+        return path
+
+    def bank_acknowledgment(self, method, path, payload):
+        self.assertEqual((method, path), ("POST", "/api/agent/bank/settle"))
+        self.assertFalse(self.agent.inspect(INSTANCE, 1)["exists"])
+        return {"state": "committed" if payload["action"] == "commit" else "cancelled", "request_id": payload["transfer"]["request_id"]}
+
+    def test_remove_settles_both_game_journals_after_stop_before_deleting_storage(self):
+        self.create()
+        paths = [self.bank_journal(game=game) for game in ("t5", "t6")]
+        def acknowledge(*args):
+            self.assertTrue(all(path.exists() for path in paths))
+            self.assertEqual(args[2]["operationId"], "remove")
+            return self.bank_acknowledgment(*args)
+        with patch.object(self.agent, "request", side_effect=acknowledge) as request:
+            result = self.agent.execute(self.operation("instance.remove", {"safetyOverride": True}, "remove"))
+        self.assertEqual(result["state"], "completed", result)
+        self.assertEqual(request.call_count, 4)
+        self.assertTrue(all(not path.exists() for path in paths))
+
+    def test_remove_retains_bank_journal_during_outage_and_resumes_after_agent_restart(self):
+        self.create()
+        path = self.bank_journal()
+        operation = self.operation("instance.remove", {"safetyOverride": True}, "remove")
+        with patch.object(self.agent, "request", side_effect=AgentError("Controller request failed")):
+            self.assertEqual(self.agent.execute(operation)["state"], "unknown")
+            self.assertEqual(self.agent.execute(operation)["state"], "unknown")
+        self.assertTrue(path.exists())
+        self.agent.journal.close()
+        self.agent = Agent(self.config, self.docker)
+        calls = list(self.docker.calls)
+        with patch.object(self.agent, "request", side_effect=self.bank_acknowledgment):
+            self.assertEqual(self.agent.execute(operation)["state"], "completed")
+        self.assertFalse(path.exists())
+        self.assertEqual(self.docker.calls, calls)
+
+    def test_rebuild_settles_before_starting_replacement_and_retains_unknown_decisions(self):
+        payload = self.create()
+        path = self.bank_journal(("commit",))
+        operation = self.operation("instance.rebuild", {"spec": payload, "nextGeneration": 2, "safetyOverride": True}, "rebuild")
+        with patch.object(self.agent, "request", return_value={"state": "committed", "request_id": "wrong"}):
+            self.assertEqual(self.agent.execute(operation)["state"], "unknown")
+        self.assertTrue(path.exists())
+        self.assertFalse(self.agent.inspect(INSTANCE, 2)["exists"])
+        with patch.object(self.agent, "request", side_effect=self.bank_acknowledgment):
+            # Existing rebuild recovery records the interrupted create as failed;
+            # a fresh operation can then rebuild without replaying the old command.
+            self.assertEqual(self.agent.execute(operation)["state"], "failed")
+            retry = self.operation("instance.rebuild", operation["payload"], "replacement")
+            self.assertEqual(self.agent.execute(retry)["state"], "completed")
+        self.assertFalse(path.exists())
+        self.assertTrue(self.agent.inspect(INSTANCE, 2)["running"])
+
+    def test_cancel_creation_retries_bank_settlement_before_deleting_runtime(self):
+        self.create()
+        path = self.bank_journal(("cancel",))
+        operation = self.operation("instance.cancel-create", {"createOperationId": "create"}, "cancel")
+        with patch.object(self.agent, "request", side_effect=AgentError("offline")):
+            self.assertEqual(self.agent.execute(operation)["state"], "unknown")
+        self.assertTrue(path.exists())
+        with patch.object(self.agent, "request", side_effect=self.bank_acknowledgment):
+            self.assertEqual(self.agent.execute(operation)["state"], "completed")
+        self.assertFalse(path.exists())
+
+    def test_malformed_or_symlinked_bank_journal_blocks_storage_deletion(self):
+        self.create()
+        path = self.bank_journal()
+        path.write_text('{broken')
+        operation = self.operation("instance.remove", {"safetyOverride": True}, "remove")
+        with patch.object(self.agent, "request") as request:
+            self.assertEqual(self.agent.execute(operation)["state"], "unknown")
+            self.assertTrue(path.exists())
+            outside = Path(self.temp.name) / "outside.json"
+            outside.write_text('{"reports": []}')
+            self.bank_journal()
+            journal = json.loads(path.read_text())
+            journal["endpoint"] = "https://untrusted.example/api/bank"
+            path.write_text(json.dumps(journal))
+            self.assertEqual(self.agent.execute(operation)["state"], "unknown")
+            path.unlink()
+            path.symlink_to(outside)
+            self.assertEqual(self.agent.execute(operation)["state"], "unknown")
+            self.assertTrue(path.is_symlink())
+            request.assert_not_called()
+
     def test_rebuild_moves_generation_and_rejects_old_commands(self):
         payload = self.create()
         self.agent.rcon = lambda *_: {"ok": True, "players": 0}