josh
e5c0fda1ed
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.
1232 lines
68 KiB
Python
1232 lines
68 KiB
Python
#!/usr/bin/env python3
|
|
"""Electric Cherry host agent. Python 3 standard library; no remote shell API.
|
|
|
|
Each operation is recorded before execution. An interrupted operation becomes
|
|
unknown and is reported for reconciliation, never automatically executed twice.
|
|
"""
|
|
|
|
import argparse
|
|
import base64
|
|
import contextlib
|
|
import hashlib
|
|
import fcntl
|
|
import json
|
|
import os
|
|
from pathlib import Path, PurePosixPath
|
|
import platform
|
|
import re
|
|
import shutil
|
|
import signal
|
|
import socket
|
|
import sqlite3
|
|
import stat as stat_module
|
|
import subprocess
|
|
import sys
|
|
import threading
|
|
import time
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
import uuid
|
|
|
|
VERSION = "1.0.0"
|
|
LABEL = "io.electric-cherry"
|
|
MAX_FILE = 32 * 1024 * 1024
|
|
ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_-]{0,127}$")
|
|
|
|
|
|
class AgentError(Exception):
|
|
pass
|
|
|
|
|
|
class Held(AgentError):
|
|
"""A reversible prerequisite prevented any disruptive action."""
|
|
|
|
|
|
class Rejected(AgentError):
|
|
"""A definitive precondition failure, before any destructive action."""
|
|
|
|
|
|
def digest(data):
|
|
return hashlib.sha256(data).hexdigest()
|
|
|
|
|
|
def file_digest(path):
|
|
h = hashlib.sha256()
|
|
with open(path, "rb") as stream:
|
|
for block in iter(lambda: stream.read(1024 * 1024), b""):
|
|
h.update(block)
|
|
return h.hexdigest()
|
|
|
|
|
|
def status_players(text):
|
|
lines = text.splitlines()
|
|
separator = next((index for index, line in enumerate(lines) if re.match(r"^\s*-+\s+-+", line)), None)
|
|
if separator is not None and any(re.search(r"\bnum\s+score\s+ping\b", line, re.I) for line in lines[:separator]):
|
|
rows = [line.strip() for line in lines[separator + 1:] if line.strip()]
|
|
if any(not re.match(r"^\d+\s+", row) for row in rows):
|
|
return None
|
|
return len(rows)
|
|
info = next((index for index, line in enumerate(lines) if line.startswith("\\") and "\\mapname\\" in line), None)
|
|
if info is not None:
|
|
return sum(bool(line.strip()) for line in lines[info + 1:])
|
|
# A truncated header or an unfamiliar dialect is never proof of emptiness.
|
|
return None
|
|
|
|
|
|
def identifier(value):
|
|
if not isinstance(value, str) or not ID_RE.fullmatch(value):
|
|
raise AgentError("Invalid identifier")
|
|
return value
|
|
|
|
|
|
def instance_identity(instance_id, generation):
|
|
try:
|
|
if str(uuid.UUID(instance_id)) != instance_id.lower():
|
|
raise ValueError()
|
|
except (ValueError, AttributeError, TypeError):
|
|
raise AgentError("Instance ID must be a UUID") from None
|
|
if isinstance(generation, bool) or not isinstance(generation, int) or generation < 1:
|
|
raise AgentError("Invalid instance generation")
|
|
return instance_id.lower(), generation
|
|
|
|
|
|
def scoped(root, value, allow_root=False):
|
|
"""Reject traversal AND existing symlinks, including dangling symlinks."""
|
|
if not isinstance(value, str) or "\\" in value or "\x00" in value:
|
|
raise AgentError("Invalid managed path")
|
|
rel = PurePosixPath(value)
|
|
if rel.is_absolute() or any(p in ("..", "") for p in rel.parts):
|
|
raise AgentError("Path is outside the managed directory")
|
|
if value not in ("", ".") and str(rel) != value:
|
|
raise AgentError("Managed paths must use their canonical relative form")
|
|
if (not rel.parts or str(rel) == ".") and not allow_root:
|
|
raise AgentError("A file path is required")
|
|
root = Path(root)
|
|
current = root
|
|
if current.is_symlink():
|
|
raise AgentError("Managed root must not be a symlink")
|
|
for part in rel.parts:
|
|
current = current / part
|
|
if current.is_symlink():
|
|
raise AgentError("Symlinks are not allowed in managed file operations")
|
|
return current
|
|
|
|
|
|
def atomic_bytes(path, data, mode=0o600):
|
|
path = Path(path)
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
tmp = path.with_name(path.name + "." + uuid.uuid4().hex + ".tmp")
|
|
try:
|
|
with open(tmp, "xb") as stream:
|
|
os.chmod(tmp, mode)
|
|
stream.write(data)
|
|
stream.flush()
|
|
os.fsync(stream.fileno())
|
|
os.replace(tmp, path)
|
|
fd = os.open(path.parent, os.O_RDONLY)
|
|
try:
|
|
os.fsync(fd)
|
|
finally:
|
|
os.close(fd)
|
|
finally:
|
|
with contextlib.suppress(FileNotFoundError):
|
|
tmp.unlink()
|
|
|
|
|
|
def atomic_json(path, value):
|
|
atomic_bytes(path, json.dumps(value, sort_keys=True).encode())
|
|
|
|
|
|
@contextlib.contextmanager
|
|
def managed_parent(root, value, create=False, uid=1000, gid=1000):
|
|
"""Use dirfds so an instance cannot race a symlink into a host-root write."""
|
|
scoped(root, value)
|
|
parts = PurePosixPath(value).parts
|
|
flags = os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW
|
|
fd = os.open(root, flags)
|
|
try:
|
|
for part in parts[:-1]:
|
|
if create:
|
|
with contextlib.suppress(FileExistsError):
|
|
os.mkdir(part, 0o755, dir_fd=fd)
|
|
if os.geteuid() == 0:
|
|
os.chown(part, uid, gid, dir_fd=fd, follow_symlinks=False)
|
|
child = os.open(part, flags, dir_fd=fd)
|
|
os.close(fd)
|
|
fd = child
|
|
yield fd, parts[-1]
|
|
finally:
|
|
os.close(fd)
|
|
|
|
|
|
@contextlib.contextmanager
|
|
def managed_open(root, value):
|
|
with managed_parent(root, value) as (parent, name):
|
|
fd = os.open(name, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK, dir_fd=parent)
|
|
try:
|
|
if not stat_module.S_ISREG(os.fstat(fd).st_mode):
|
|
raise AgentError("Requested managed path is not a regular file")
|
|
with os.fdopen(fd, "rb", closefd=False) as stream:
|
|
yield stream
|
|
finally:
|
|
os.close(fd)
|
|
|
|
|
|
def managed_read(root, value):
|
|
try:
|
|
with managed_open(root, value) as stream:
|
|
if os.fstat(stream.fileno()).st_size > MAX_FILE:
|
|
raise Rejected("Managed file exceeds the 32 MiB read limit")
|
|
content = stream.read(MAX_FILE + 1)
|
|
if len(content) > MAX_FILE:
|
|
raise Rejected("Managed file exceeds the 32 MiB read limit")
|
|
return content
|
|
except FileNotFoundError:
|
|
return None
|
|
|
|
|
|
def managed_write(root, value, content, uid=1000, gid=1000):
|
|
if content is None:
|
|
try:
|
|
with managed_parent(root, value) as (parent, name):
|
|
with contextlib.suppress(FileNotFoundError):
|
|
os.unlink(name, dir_fd=parent)
|
|
os.fsync(parent)
|
|
except FileNotFoundError:
|
|
pass
|
|
return
|
|
with managed_parent(root, value, create=True, uid=uid, gid=gid) as (parent, name):
|
|
prior = None
|
|
try:
|
|
prior = os.stat(name, dir_fd=parent, follow_symlinks=False)
|
|
if not stat_module.S_ISREG(prior.st_mode):
|
|
raise Rejected("Managed target is not a regular file")
|
|
except FileNotFoundError:
|
|
pass
|
|
temp_name = ".ec-" + uuid.uuid4().hex
|
|
fd = os.open(temp_name, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600, dir_fd=parent)
|
|
try:
|
|
os.fchmod(fd, prior.st_mode & 0o777 if prior else 0o644)
|
|
if os.geteuid() == 0:
|
|
os.fchown(fd, prior.st_uid if prior else uid, prior.st_gid if prior else gid)
|
|
with os.fdopen(fd, "wb", closefd=False) as stream:
|
|
stream.write(content)
|
|
stream.flush()
|
|
os.fsync(fd)
|
|
os.replace(temp_name, name, src_dir_fd=parent, dst_dir_fd=parent)
|
|
os.fsync(parent)
|
|
finally:
|
|
os.close(fd)
|
|
with contextlib.suppress(FileNotFoundError):
|
|
os.unlink(temp_name, dir_fd=parent)
|
|
|
|
|
|
class Journal:
|
|
def __init__(self, path):
|
|
self.db = sqlite3.connect(path)
|
|
self.db.row_factory = sqlite3.Row
|
|
self.db.execute("PRAGMA journal_mode=WAL")
|
|
self.db.execute("PRAGMA synchronous=FULL")
|
|
self.db.execute("CREATE TABLE IF NOT EXISTS operations (id TEXT PRIMARY KEY, fingerprint TEXT NOT NULL, request TEXT NOT NULL, state TEXT NOT NULL, result TEXT, error TEXT, updated REAL NOT NULL)")
|
|
if "checkpoint" not in {row[1] for row in self.db.execute("PRAGMA table_info(operations)")}:
|
|
self.db.execute("ALTER TABLE operations ADD COLUMN checkpoint TEXT")
|
|
self.db.execute("UPDATE operations SET state='unknown', error='Agent restarted during execution; inspect before reconciliation', updated=? WHERE state='running'", (time.time(),))
|
|
self.db.commit()
|
|
|
|
def accept(self, request):
|
|
op_id = identifier(request.get("id"))
|
|
immutable = {key: request.get(key) for key in ("id", "kind", "instanceId", "generation", "payload")}
|
|
canonical = json.dumps(immutable, sort_keys=True, separators=(",", ":"))
|
|
fingerprint = digest(canonical.encode())
|
|
row = self.db.execute("SELECT * FROM operations WHERE id=?", (op_id,)).fetchone()
|
|
if row:
|
|
if row["fingerprint"] != fingerprint:
|
|
raise AgentError("Operation ID was reused with different content")
|
|
return dict(row), False
|
|
self.db.execute("INSERT INTO operations(id,fingerprint,request,state,updated) VALUES(?,?,?,'accepted',?)", (op_id, fingerprint, canonical, time.time()))
|
|
self.db.commit()
|
|
return self.get(op_id), True
|
|
|
|
def get(self, op_id):
|
|
row = self.db.execute("SELECT * FROM operations WHERE id=?", (op_id,)).fetchone()
|
|
if row is None:
|
|
raise AgentError("Unknown operation")
|
|
return dict(row)
|
|
|
|
def update(self, op_id, state, result=None, error=None):
|
|
self.db.execute("UPDATE operations SET state=?, result=?, error=?, updated=? WHERE id=?", (state, json.dumps(result) if result is not None else None, error, time.time(), op_id))
|
|
self.db.commit()
|
|
return self.get(op_id)
|
|
|
|
def close(self):
|
|
self.db.close()
|
|
|
|
def checkpoint(self, op_id, value):
|
|
self.db.execute("UPDATE operations SET checkpoint=?, updated=? WHERE id=?", (json.dumps(value), time.time(), op_id))
|
|
self.db.commit()
|
|
|
|
|
|
class Docker:
|
|
def call(self, *args, timeout=120, tolerate_missing=False, merge_stderr=False):
|
|
try:
|
|
result = subprocess.run(["docker", *args], stdout=subprocess.PIPE, stderr=subprocess.STDOUT if merge_stderr else subprocess.PIPE, timeout=timeout, check=False)
|
|
except (OSError, subprocess.TimeoutExpired):
|
|
raise AgentError("Docker command unavailable or timed out") from None
|
|
if result.returncode:
|
|
# Never include arbitrary Docker stderr: create errors may echo secrets.
|
|
if tolerate_missing and (b"No such object" in (result.stderr or b"") or b"No such container" in (result.stderr or b"")):
|
|
return None
|
|
raise AgentError("Docker rejected the requested operation")
|
|
return result.stdout.decode("utf-8", "replace")
|
|
|
|
def inspect(self, name):
|
|
raw = self.call("container", "inspect", name, tolerate_missing=True)
|
|
return json.loads(raw)[0] if raw is not None else None
|
|
|
|
|
|
class Agent:
|
|
def __init__(self, config, docker=None):
|
|
self.config = config
|
|
self.root = Path(config.get("root", "/var/lib/electric-cherry")).resolve()
|
|
for directory in ("agent", "instances", "libraries"):
|
|
(self.root / directory).mkdir(parents=True, exist_ok=True)
|
|
os.chmod(self.root / "agent", 0o700)
|
|
self.journal = Journal(self.root / "agent" / "operations.sqlite3")
|
|
self.docker = docker or Docker()
|
|
self.stopping = threading.Event()
|
|
self.current_operation = None
|
|
self.library_progress = {}
|
|
self.config_path = config.get("configPath", "/etc/electric-cherry/agent.json")
|
|
self.executing_id = None
|
|
|
|
def instance_dir(self, instance_id, generation):
|
|
instance_id, generation = instance_identity(instance_id, generation)
|
|
return self.root / "instances" / instance_id / str(generation)
|
|
|
|
def container_name(self, instance_id, generation):
|
|
instance_id, generation = instance_identity(instance_id, generation)
|
|
return f"ec-{instance_id}-g{generation}"
|
|
|
|
def inspect(self, instance_id, generation):
|
|
name = self.container_name(instance_id, generation)
|
|
info = self.docker.inspect(name)
|
|
if info is None:
|
|
return {"exists": False, "running": False, "healthy": False, "state": "missing", "instanceId": instance_id, "generation": generation}
|
|
labels = info.get("Config", {}).get("Labels", {}) or {}
|
|
if labels.get(LABEL + ".managed") != "true" or labels.get(LABEL + ".instance") != instance_id or labels.get(LABEL + ".generation") != str(generation):
|
|
raise AgentError("Container ownership or generation does not match")
|
|
state = info.get("State", {})
|
|
status = state.get("Status", "unknown")
|
|
healthy = state.get("Running") is True and state.get("Health", {}).get("Status", "healthy") == "healthy"
|
|
if status == "running" and not healthy:
|
|
status = state.get("Health", {}).get("Status", "unknown")
|
|
return {"exists": True, "running": state.get("Running") is True, "healthy": healthy, "state": status, "containerId": info.get("Id"), "instanceId": instance_id, "generation": generation, "startedAt": state.get("StartedAt"), "creationOperation": labels.get(LABEL + ".operation"), "specSha256": labels.get(LABEL + ".spec-sha256")}
|
|
|
|
def checkpoint(self, value):
|
|
if self.executing_id:
|
|
self.journal.checkpoint(self.executing_id, value)
|
|
|
|
def mark_removed(self, instance_id, generation):
|
|
marker = self.instance_dir(instance_id, generation).parent / "current.json"
|
|
if not marker.exists() or json.loads(marker.read_text())["generation"] <= generation:
|
|
atomic_json(marker, {"generation": generation, "removed": True})
|
|
|
|
def make_writable(self, path, directory=False):
|
|
if os.geteuid() == 0:
|
|
os.chown(path, int(self.config.get("runtimeUid", 1000)), int(self.config.get("runtimeGid", 1000)))
|
|
os.chmod(path, 0o755 if directory else 0o644)
|
|
elif directory:
|
|
os.chmod(path, 0o777)
|
|
|
|
def spec(self, instance_id, generation):
|
|
path = self.instance_dir(instance_id, generation) / "spec.json"
|
|
try:
|
|
return json.loads(path.read_text())
|
|
except (OSError, ValueError):
|
|
raise AgentError("Managed instance metadata is unavailable") from None
|
|
|
|
def request(self, method, path, payload=None, credential=True):
|
|
url = self.config["controller"].rstrip("/") + path
|
|
if urllib.parse.urlsplit(url).scheme != "https" and not self.config.get("allowInsecureLocalhost", False):
|
|
raise AgentError("Controller URL must use HTTPS")
|
|
if urllib.parse.urlsplit(url).scheme != "https" and urllib.parse.urlsplit(url).hostname not in ("127.0.0.1", "localhost", "::1"):
|
|
raise AgentError("Insecure development transport is limited to loopback")
|
|
headers = {"Content-Type": "application/json", "User-Agent": "electric-cherry-agent/" + VERSION}
|
|
if credential:
|
|
headers["Authorization"] = "Bearer " + self.config["credential"]
|
|
data = None if payload is None else json.dumps(payload).encode()
|
|
req = urllib.request.Request(url, data=data, headers=headers, method=method)
|
|
try:
|
|
class NoRedirect(urllib.request.HTTPRedirectHandler):
|
|
def redirect_request(self, req, fp, code, msg, hdrs, newurl):
|
|
raise AgentError("Controller redirects are not allowed")
|
|
with urllib.request.build_opener(NoRedirect()).open(req, timeout=35) as response:
|
|
return json.load(response)
|
|
except (OSError, ValueError):
|
|
raise AgentError("Controller request failed") from None
|
|
|
|
def inventory(self):
|
|
usage = shutil.disk_usage(self.root)
|
|
memory = {}
|
|
try:
|
|
for line in Path("/proc/meminfo").read_text().splitlines():
|
|
key, value = line.split(":", 1)
|
|
memory[key] = int(value.strip().split()[0]) * 1024
|
|
except (OSError, ValueError):
|
|
pass
|
|
docker_version, docker_running = "", False
|
|
try:
|
|
info = json.loads(self.docker.call("info", "--format", "{{json .}}", timeout=15))
|
|
docker_version, docker_running = info.get("ServerVersion", ""), True
|
|
except (AgentError, ValueError):
|
|
pass
|
|
libraries, caches = [], []
|
|
for marker in (self.root / "libraries").glob("*/*/manifest.json"):
|
|
try:
|
|
value = json.loads(marker.read_text())
|
|
libraries.append({"id": value["libraryId"], "state": "ready"})
|
|
caches.append({"libraryVersionId": value["libraryId"], "status": "ready", "bytesDownloaded": value["size"], "bytesTotal": value["size"], "manifestSha256": value.get("manifestSha256")})
|
|
except (OSError, ValueError, KeyError):
|
|
continue
|
|
progress = list(self.library_progress.values())
|
|
libraries.extend({"id": item["libraryVersionId"], "state": item["status"]} for item in progress)
|
|
caches.extend(progress)
|
|
images = []
|
|
if docker_running:
|
|
with contextlib.suppress(AgentError):
|
|
lines = self.docker.call("image", "ls", "--no-trunc", "--digests", "--format", "{{.Repository}}|{{.Tag}}|{{.Digest}}|{{.ID}}", timeout=15).splitlines()
|
|
image_ids = set()
|
|
for line in lines:
|
|
parts = line.split("|")
|
|
if len(parts) != 4:
|
|
continue
|
|
repository, tag, image_digest, image_id = parts
|
|
image_ids.add(image_id)
|
|
if repository != "<none>" and tag != "<none>":
|
|
images.append(repository + ":" + tag)
|
|
if repository != "<none>" and image_digest != "<none>":
|
|
images.append(repository + "@" + image_digest)
|
|
aliases = self.root / "agent" / "runtime-images.json"
|
|
if aliases.exists():
|
|
with contextlib.suppress(OSError, ValueError):
|
|
images.extend(alias for alias, image_id in json.loads(aliases.read_text()).items() if image_id in image_ids)
|
|
os_name = platform.system()
|
|
with contextlib.suppress(OSError):
|
|
for line in Path("/etc/os-release").read_text().splitlines():
|
|
if line.startswith("PRETTY_NAME="):
|
|
os_name = line.split("=", 1)[1].strip('"')
|
|
return {"hostname": socket.gethostname(), "agentVersion": VERSION, "protocolVersion": 1, "os": os_name, "osVersion": platform.release(), "architecture": platform.machine(), "cpu": platform.processor() or platform.machine(), "cpuCount": os.cpu_count() or 0, "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)), "libraries": libraries, "libraryCaches": caches, "currentOperation": self.current_operation, "portRange": self.config.get("portRange", {"min": 28960, "max": 29060})}
|
|
|
|
def safe_to_disrupt(self, instance_id, generation):
|
|
health = self.inspect(instance_id, generation)
|
|
if health["state"] in ("missing", "stopped", "exited", "created", "dead"):
|
|
return
|
|
if not health["running"]:
|
|
raise Held("Container state is unknown; disruption held")
|
|
result = self.rcon(instance_id, generation, "status")
|
|
if not result.get("ok") or result.get("players") is None:
|
|
raise Held("Fresh local player occupancy is unavailable; disruption held")
|
|
if result["players"] != 0:
|
|
raise Held("Players are present; disruption held")
|
|
|
|
def rcon(self, instance_id, generation, command):
|
|
if not isinstance(command, str) or not command or len(command) > 2048 or any(ch in command for ch in ("\x00", "\r", "\n")):
|
|
raise AgentError("Invalid RCON command")
|
|
saved = self.spec(instance_id, generation)
|
|
password = saved.get("rconPassword") or saved.get("env", {}).get("PLUTAINER_RCON_PASSWORD")
|
|
if not password:
|
|
return {"ok": False, "reason": "unconfigured", "players": None, "text": ""}
|
|
bind = saved.get("bindAddress", "0.0.0.0")
|
|
host = "127.0.0.1" if bind in ("0.0.0.0", "") else "::1" if bind == "::" else bind
|
|
family = socket.AF_INET6 if ":" in host else socket.AF_INET
|
|
chunks = []
|
|
with socket.socket(family, socket.SOCK_DGRAM) as sock:
|
|
sock.settimeout(3)
|
|
try:
|
|
sock.connect((host, int(saved["port"])))
|
|
sock.send(b"\xff\xff\xff\xffrcon " + password.encode() + b" " + command.encode() + b"\n")
|
|
while len(chunks) < 64 and sum(map(len, chunks)) < 1024 * 1024:
|
|
chunks.append(sock.recv(65535))
|
|
sock.settimeout(0.15)
|
|
except socket.timeout:
|
|
pass
|
|
except OSError:
|
|
return {"ok": False, "reason": "unreachable", "players": None, "text": ""}
|
|
if any(not chunk.startswith(b"\xff\xff\xff\xff") for chunk in chunks):
|
|
return {"ok": False, "reason": "invalid-reply", "players": None, "text": ""}
|
|
text = "\n".join(chunk[4:].decode("utf-8", "replace").removeprefix("print\n") for chunk in chunks)
|
|
if not text or re.search(r"(bad|invalid|wrong).{0,20}(password|rcon)", text, re.I):
|
|
return {"ok": False, "reason": "timeout" if not text else "authentication", "players": None, "text": text}
|
|
players = status_players(text) if command == "status" else None
|
|
return {"ok": True, "text": text, "players": players}
|
|
|
|
def validate_spec(self, payload):
|
|
game = payload.get("game")
|
|
if game not in ("t5", "t6"):
|
|
raise AgentError("Only T5 and T6 are supported")
|
|
library = self.library_dir(game, payload.get("libraryId"))
|
|
self.verify_library(library)
|
|
image = payload.get("image", "")
|
|
if not isinstance(image, str) or not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._/:@-]{0,511}", image):
|
|
raise AgentError("Invalid runtime image")
|
|
port = payload.get("port")
|
|
allowed = self.config.get("portRange", {"min": 28960, "max": 29060})
|
|
if isinstance(port, bool) or not isinstance(port, int) or not allowed["min"] <= port <= allowed["max"]:
|
|
raise AgentError("Port is outside the configured host range")
|
|
bind = payload.get("bindAddress", "0.0.0.0")
|
|
try:
|
|
import ipaddress
|
|
ipaddress.ip_address(bind)
|
|
except (ValueError, TypeError):
|
|
raise AgentError("Invalid local bind address") from None
|
|
env = dict(payload.get("env", {}))
|
|
if payload.get("rconPassword"):
|
|
env["PLUTAINER_RCON_PASSWORD"] = payload["rconPassword"]
|
|
if not isinstance(env, dict) or any(not re.fullmatch(r"[A-Z][A-Z0-9_]*", key) or not isinstance(value, str) or "\x00" in value or "\n" in value for key, value in env.items()):
|
|
raise AgentError("Invalid runtime environment")
|
|
# Runtime download is a separate preparation operation. A deployment
|
|
# uses the already prepared image even during a registry outage.
|
|
self.docker.call("image", "inspect", image)
|
|
files = payload.get("files", [])
|
|
if not isinstance(files, list) or len(files) > 10000:
|
|
raise Rejected("Invalid initial file list")
|
|
total, paths = 0, set()
|
|
for item in files:
|
|
scoped(self.root / "agent" / "validation", item["path"])
|
|
if item["path"] in paths:
|
|
raise Rejected("Duplicate initial file path")
|
|
paths.add(item["path"])
|
|
total += len(self.decode_file(item))
|
|
if total > 64 * 1024 * 1024:
|
|
raise Rejected("Initial files exceed the 64 MiB operation limit")
|
|
return game, library, image, port, bind, env
|
|
|
|
def create(self, instance_id, generation, payload):
|
|
directory = self.instance_dir(instance_id, generation)
|
|
current_marker = directory.parent / "current.json"
|
|
if current_marker.exists():
|
|
current = json.loads(current_marker.read_text())
|
|
if current["generation"] > generation or current["generation"] == generation and current.get("removed"):
|
|
raise AgentError("Instance generation has been superseded or removed")
|
|
existing = self.inspect(instance_id, generation)
|
|
if existing["exists"]:
|
|
if self.spec(instance_id, generation) != payload:
|
|
raise AgentError("Instance already exists with a different specification")
|
|
return existing
|
|
game, library, image, port, bind, env = self.validate_spec(payload)
|
|
app = directory / "app"
|
|
app.mkdir(parents=True, exist_ok=True)
|
|
self.make_writable(app, directory=True)
|
|
self.write_initial_files(app, payload.get("files", []))
|
|
atomic_json(directory / "spec.json", payload)
|
|
env_path = directory / "environment"
|
|
atomic_bytes(env_path, "".join(f"{key}={value}\n" for key, value in env.items()).encode())
|
|
name = self.container_name(instance_id, generation)
|
|
spec_sha = digest(json.dumps(payload, sort_keys=True, separators=(",", ":")).encode())
|
|
args = ["container", "create", "--name", name, "--restart", "unless-stopped", "--label", LABEL + ".managed=true", "--label", LABEL + ".instance=" + instance_id, "--label", LABEL + ".generation=" + str(generation), "--label", LABEL + ".operation=" + (self.executing_id or "direct"), "--label", LABEL + ".spec-sha256=" + spec_sha, "--mount", f"type=bind,src={app},dst=/home/plutainer/app", "--mount", f"type=bind,src={library / 'content'},dst=/home/plutainer/gamefiles,readonly", "--env-file", str(env_path)]
|
|
bind_arg = "[" + bind + "]" if ":" in bind else bind
|
|
for protocol in ("udp", "tcp"):
|
|
args.extend(["--publish", f"{bind_arg}:{port}:{port}/{protocol}"])
|
|
args.append(image)
|
|
self.checkpoint({"phase": "creating", "generation": generation, "specSha256": spec_sha})
|
|
self.docker.call(*args, timeout=120)
|
|
self.docker.call("container", "start", name)
|
|
atomic_json(current_marker, {"generation": generation})
|
|
return self.inspect(instance_id, generation)
|
|
|
|
def write_initial_files(self, root, files):
|
|
if not isinstance(files, list) or len(files) > 10000:
|
|
raise AgentError("Invalid file list")
|
|
for item in files:
|
|
content = self.decode_file(item)
|
|
managed_write(root, item["path"], content, int(self.config.get("runtimeUid", 1000)), int(self.config.get("runtimeGid", 1000)))
|
|
|
|
def decode_file(self, item):
|
|
try:
|
|
content = base64.b64decode(item["contentBase64"], validate=True)
|
|
except (ValueError, KeyError, TypeError):
|
|
raise AgentError("Invalid encoded file content") from None
|
|
if len(content) > MAX_FILE:
|
|
raise AgentError("File exceeds staging size limit")
|
|
if item.get("sha256") and item["sha256"] != digest(content):
|
|
raise AgentError("Staged file checksum mismatch")
|
|
return content
|
|
|
|
def lifecycle(self, kind, instance_id, generation, payload):
|
|
if kind == "instance.create":
|
|
return self.create(instance_id, generation, payload)
|
|
if kind == "instance.inspect":
|
|
return self.inspect(instance_id, generation)
|
|
health = self.inspect(instance_id, generation)
|
|
name = self.container_name(instance_id, generation)
|
|
before = []
|
|
if kind == "instance.start":
|
|
if not health["exists"]:
|
|
raise AgentError("Instance does not exist")
|
|
if payload.get("stageId"):
|
|
self.safe_to_disrupt(instance_id, generation)
|
|
self.checkpoint({"phase": "activating", "containerId": health.get("containerId"), "startedAt": health.get("startedAt"), "restartRequired": health["running"]})
|
|
before = self.activate(instance_id, generation, payload["stageId"])
|
|
health = self.inspect(instance_id, generation)
|
|
self.safe_to_disrupt(instance_id, generation)
|
|
restart_required = bool(payload.get("stageId") and health["running"])
|
|
self.checkpoint({"phase": "starting", "containerId": health.get("containerId"), "startedAt": health.get("startedAt"), "restartRequired": restart_required})
|
|
if restart_required:
|
|
self.docker.call("container", "restart", "--time", "30", name)
|
|
else:
|
|
self.docker.call("container", "start", name)
|
|
elif kind in ("instance.stop", "instance.restart", "instance.remove", "instance.rebuild"):
|
|
if kind == "instance.rebuild":
|
|
next_generation = payload.get("nextGeneration", payload.get("newGeneration"))
|
|
if isinstance(next_generation, bool) or not isinstance(next_generation, int) or next_generation <= generation:
|
|
raise AgentError("Rebuild must use a newer immutable generation")
|
|
spec = payload.get("spec") or {key: value for key, value in payload.items() if key not in ("nextGeneration", "newGeneration")}
|
|
self.validate_spec(spec)
|
|
self.safe_to_disrupt(instance_id, generation)
|
|
self.checkpoint({"phase": "disrupting", "containerId": health.get("containerId"), "startedAt": health.get("startedAt")})
|
|
if payload.get("stageId") and kind == "instance.restart":
|
|
before = self.activate(instance_id, generation, payload["stageId"])
|
|
if kind in ("instance.remove", "instance.rebuild"):
|
|
if health["exists"]:
|
|
self.docker.call("container", "stop", "--time", "30", name)
|
|
self.docker.call("container", "rm", name)
|
|
if kind == "instance.rebuild":
|
|
result = self.create(instance_id, next_generation, spec)
|
|
shutil.rmtree(self.instance_dir(instance_id, generation), ignore_errors=False)
|
|
return result
|
|
directory = self.instance_dir(instance_id, generation)
|
|
if directory.exists():
|
|
shutil.rmtree(directory)
|
|
self.mark_removed(instance_id, generation)
|
|
elif health["exists"]:
|
|
if kind == "instance.restart" and payload.get("stageId"):
|
|
self.safe_to_disrupt(instance_id, generation)
|
|
self.docker.call("container", "stop" if kind.endswith("stop") else "restart", "--time", "30", name)
|
|
else:
|
|
raise AgentError("Unsupported instance operation")
|
|
return {**self.inspect(instance_id, generation), "before": before}
|
|
|
|
def stage(self, instance_id, generation, payload, operation_id):
|
|
directory = self.instance_dir(instance_id, generation)
|
|
self.spec(instance_id, generation)
|
|
stage_id = identifier(payload.get("stageId", operation_id))
|
|
stage = directory / "stages" / stage_id
|
|
files = payload.get("files")
|
|
if not isinstance(files, list) or not files or len(files) > 10000:
|
|
raise AgentError("Invalid staged file list")
|
|
planned = []
|
|
paths = set()
|
|
total_size = 0
|
|
for item in files:
|
|
scoped(directory / "app", item["path"])
|
|
if item["path"] in paths:
|
|
raise AgentError("Duplicate staged file path")
|
|
paths.add(item["path"])
|
|
deleting = item.get("delete") is True or item.get("contentBase64") is None
|
|
content = None if deleting else self.decode_file(item)
|
|
total_size += len(content) if content is not None else 0
|
|
if total_size > 64 * 1024 * 1024:
|
|
raise Rejected("Staged file content exceeds the 64 MiB operation limit")
|
|
planned.append({"id": item.get("id"), "path": item["path"], "delete": deleting, "sha256": None if deleting else digest(content), "checkBefore": item.get("checkBefore", "expectedSha256" in item), "expectedBeforeSha": item.get("expectedBeforeSha", item.get("expectedSha256")), "contentBase64": None if deleting else base64.b64encode(content).decode()})
|
|
manifest = {"stageId": stage_id, "files": planned, "state": "staged"}
|
|
marker = stage / "manifest.json"
|
|
if marker.exists():
|
|
existing = json.loads(marker.read_text())
|
|
if existing["files"] != planned:
|
|
raise AgentError("Stage ID already contains different files")
|
|
return {"stageId": stage_id, "files": len(planned), "state": existing["state"]}
|
|
atomic_json(marker, manifest)
|
|
return {"stageId": stage_id, "files": len(planned), "state": "staged"}
|
|
|
|
def activate(self, instance_id, generation, stage_id):
|
|
directory = self.instance_dir(instance_id, generation)
|
|
stage = directory / "stages" / identifier(stage_id)
|
|
marker = stage / "manifest.json"
|
|
saved = json.loads(marker.read_text())
|
|
if saved["state"] == "activated":
|
|
return saved["before"]
|
|
if saved["state"] != "staged":
|
|
raise AgentError("Interrupted activation requires explicit rollback")
|
|
before = []
|
|
for item in saved["files"]:
|
|
content = managed_read(directory / "app", item["path"])
|
|
actual_sha = None if content is None else digest(content)
|
|
if item.get("checkBefore") and actual_sha != item.get("expectedBeforeSha"):
|
|
raise Rejected("Target changed after campaign review")
|
|
before.append({"id": item.get("id"), "path": item["path"], "contentBase64": None if content is None else base64.b64encode(content).decode()})
|
|
# Snapshot preparation may take time. Occupancy is rechecked at the
|
|
# local activation boundary, after all file I/O and integrity checks.
|
|
self.safe_to_disrupt(instance_id, generation)
|
|
saved.update(state="activating", before=before)
|
|
atomic_json(marker, saved)
|
|
try:
|
|
for item in saved["files"]:
|
|
managed_write(directory / "app", item["path"], None if item["delete"] else self.decode_file(item), int(self.config.get("runtimeUid", 1000)), int(self.config.get("runtimeGid", 1000)))
|
|
saved["state"] = "activated"
|
|
atomic_json(marker, saved)
|
|
except Exception:
|
|
self.restore_files(directory / "app", before)
|
|
saved["state"] = "rolled-back"
|
|
atomic_json(marker, saved)
|
|
raise
|
|
return before
|
|
|
|
def restore_files(self, app, before):
|
|
for item in before:
|
|
managed_write(app, item["path"], None if item["contentBase64"] is None else self.decode_file(item), int(self.config.get("runtimeUid", 1000)), int(self.config.get("runtimeGid", 1000)))
|
|
|
|
def file_operation(self, kind, instance_id, generation, payload, operation_id):
|
|
directory = self.instance_dir(instance_id, generation)
|
|
app = directory / "app"
|
|
if kind == "files.stage":
|
|
return self.stage(instance_id, generation, payload, operation_id)
|
|
if kind in ("files.activate", "files.rollback"):
|
|
self.safe_to_disrupt(instance_id, generation)
|
|
stage_id = identifier(payload["stageId"])
|
|
health = self.inspect(instance_id, generation)
|
|
self.checkpoint({"phase": "activating" if kind == "files.activate" else "rolling-back", "containerId": health.get("containerId"), "startedAt": health.get("startedAt")})
|
|
if kind == "files.activate":
|
|
before = self.activate(instance_id, generation, stage_id)
|
|
result = {"activated": True, "before": before}
|
|
else:
|
|
marker = directory / "stages" / stage_id / "manifest.json"
|
|
saved = json.loads(marker.read_text())
|
|
if saved["state"] not in ("activated", "activating", "rolled-back"):
|
|
raise AgentError("Stage has no activation to roll back")
|
|
if saved["state"] != "rolled-back":
|
|
self.restore_files(app, saved["before"])
|
|
saved["state"] = "rolled-back"
|
|
atomic_json(marker, saved)
|
|
result = {"rolledBack": True}
|
|
if payload.get("restart", True) and self.inspect(instance_id, generation)["exists"]:
|
|
self.safe_to_disrupt(instance_id, generation)
|
|
self.docker.call("container", "restart", "--time", "30", self.container_name(instance_id, generation))
|
|
return result
|
|
if kind == "logs.read":
|
|
return self.read_log(instance_id, generation, payload)
|
|
path = scoped(app, payload.get("path", ""), allow_root=kind in ("files.list", "files.stat"))
|
|
if kind == "files.read":
|
|
content = managed_read(app, payload["path"])
|
|
if content is None:
|
|
return {"contentBase64": None}
|
|
return {"contentBase64": base64.b64encode(content).decode(), "sha256": digest(content), "size": len(content)}
|
|
if kind == "files.stat":
|
|
if not path.exists():
|
|
return None
|
|
stat = path.stat()
|
|
return {"size": stat.st_size, "mtime": stat.st_mtime, "isDirectory": path.is_dir(), "isFile": path.is_file()}
|
|
if kind == "files.list":
|
|
if not path.exists():
|
|
return {"entries": []}
|
|
return {"entries": [{"name": child.name, "path": child.relative_to(app).as_posix(), "isDirectory": child.is_dir(), "size": child.stat().st_size} for child in sorted(path.iterdir()) if not child.is_symlink()][:10000]}
|
|
raise AgentError("Unsupported file operation")
|
|
|
|
def read_log(self, instance_id, generation, payload):
|
|
app = self.instance_dir(instance_id, generation) / "app"
|
|
value = payload["path"]
|
|
count = min(2000, max(1, int(payload.get("lines", 200))))
|
|
# Validate the original relative spelling independently of symlinks.
|
|
scoped(self.root / "agent" / "log-validation", value)
|
|
try:
|
|
resolved = (app / value).resolve(strict=True)
|
|
relative = resolved.relative_to(app.resolve(strict=True)).as_posix()
|
|
except FileNotFoundError:
|
|
if value != "logs/console.log" or not self.inspect(instance_id, generation)["exists"]:
|
|
return None
|
|
text = self.docker.call("container", "logs", "--tail", str(count), "--timestamps", self.container_name(instance_id, generation), timeout=30, merge_stderr=True)
|
|
lines = text.splitlines()
|
|
return {"lines": lines[-count:], "size": len(text.encode()), "mtime": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "truncated": len(lines) >= count, "source": "container"}
|
|
except ValueError:
|
|
raise AgentError("Log symlink resolves outside the managed instance") from None
|
|
# Reopen the resolved relative target through no-follow directory FDs;
|
|
# swapping either the original link or a target component cannot escape.
|
|
with managed_open(app, relative) as stream:
|
|
stat = os.fstat(stream.fileno())
|
|
start = max(0, stat.st_size - 256 * 1024)
|
|
stream.seek(start)
|
|
lines = stream.read(256 * 1024).decode("utf-8", "replace").splitlines()
|
|
if start:
|
|
lines = lines[1:]
|
|
return {"lines": lines[-count:], "size": stat.st_size, "mtime": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(stat.st_mtime)), "truncated": start > 0}
|
|
|
|
def library_dir(self, game, library_id):
|
|
if game not in ("t5", "t6"):
|
|
raise AgentError("Unsupported library game")
|
|
return self.root / "libraries" / game / identifier(library_id)
|
|
|
|
def verify_library(self, directory):
|
|
marker = directory / "manifest.json"
|
|
if not marker.exists():
|
|
raise AgentError("Required game library is not ready")
|
|
manifest = json.loads(marker.read_text())
|
|
for item in manifest["files"]:
|
|
path = scoped(directory / "content", item["path"])
|
|
if not path.is_file() or path.stat().st_size != item["size"] or file_digest(path) != item["sha256"]:
|
|
raise AgentError("Game library integrity verification failed")
|
|
return manifest
|
|
|
|
def artifact_url(self, value):
|
|
controller = self.config["controller"].rstrip("/")
|
|
url = urllib.parse.urljoin(controller + "/", value)
|
|
parsed = urllib.parse.urlsplit(url)
|
|
origin = urllib.parse.urlsplit(controller)
|
|
if (parsed.scheme, parsed.netloc) != (origin.scheme, origin.netloc) or parsed.username or parsed.password:
|
|
raise AgentError("Library transport must remain on the authenticated controller origin")
|
|
if parsed.scheme != "https" and not (self.config.get("allowInsecureLocalhost") and parsed.hostname in ("127.0.0.1", "localhost", "::1")):
|
|
raise AgentError("Library transport must use HTTPS")
|
|
return url
|
|
|
|
def artifact_open(self, method, path, data=None, extra_headers=None):
|
|
headers = {"Authorization": "Bearer " + self.config["credential"], **(extra_headers or {})}
|
|
req = urllib.request.Request(self.artifact_url(path), data=data, method=method, headers=headers)
|
|
# Redirects cannot forward host credentials to another origin.
|
|
class NoRedirect(urllib.request.HTTPRedirectHandler):
|
|
def redirect_request(self, req, fp, code, msg, hdrs, newurl):
|
|
raise AgentError("Artifact redirects are not allowed")
|
|
return urllib.request.build_opener(NoRedirect()).open(req, timeout=60)
|
|
|
|
def library_download(self, payload):
|
|
library_id = identifier(payload.get("libraryVersionId", payload.get("libraryId")))
|
|
with self.artifact_open("GET", payload["manifestUrl"]) as response:
|
|
manifest = json.load(response)
|
|
if manifest.get("id") != library_id or not isinstance(manifest.get("files"), list) or not manifest["files"]:
|
|
raise AgentError("Library manifest identity mismatch")
|
|
files = manifest["files"]
|
|
canonical = json.dumps([{ "path": item["path"], "size": item["size"], "sha256": item["sha256"] } for item in files], separators=(",", ":"), ensure_ascii=False)
|
|
manifest_sha = manifest.get("manifestSha256")
|
|
if not isinstance(manifest_sha, str) or not re.fullmatch("[0-9a-f]{64}", manifest_sha):
|
|
raise AgentError("Library manifest checksum is missing")
|
|
if digest(canonical.encode("utf-8")) != manifest_sha:
|
|
raise AgentError("Library manifest checksum mismatch")
|
|
if payload.get("manifestSha256") and payload["manifestSha256"] != manifest_sha:
|
|
raise AgentError("Library manifest differs from the requested immutable version")
|
|
directory = self.library_dir(manifest["game"], library_id)
|
|
if directory.exists():
|
|
saved = self.verify_library(directory)
|
|
if saved.get("manifestSha256") != manifest_sha:
|
|
raise AgentError("Published library changed unexpectedly")
|
|
return {"libraryId": library_id, "state": "ready", "size": saved["size"]}
|
|
staging = directory.with_name("." + library_id + ".partial")
|
|
staging.mkdir(parents=True, exist_ok=True)
|
|
source_marker = staging / "source.json"
|
|
if source_marker.exists() and json.loads(source_marker.read_text()).get("manifestSha256") != manifest_sha:
|
|
raise Rejected("Partial library download belongs to a different manifest")
|
|
atomic_json(source_marker, {"libraryId": library_id, "manifestSha256": manifest_sha})
|
|
size = sum(item["size"] for item in files)
|
|
downloaded = 0
|
|
paths = set()
|
|
progress = {"libraryVersionId": library_id, "status": "downloading", "bytesDownloaded": 0, "bytesTotal": size, "manifestSha256": manifest_sha}
|
|
self.library_progress[library_id] = progress
|
|
try:
|
|
if shutil.disk_usage(staging).free < max(0, size - sum(path.stat().st_size for path in staging.rglob("*") if path.is_file())):
|
|
raise AgentError("Insufficient library storage")
|
|
for item in files:
|
|
if item["path"] in paths or not isinstance(item["size"], int) or item["size"] < 0 or not re.fullmatch("[0-9a-f]{64}", item["sha256"]):
|
|
raise AgentError("Invalid library file manifest")
|
|
paths.add(item["path"])
|
|
target = scoped(staging / "content", item["path"])
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
offset = target.stat().st_size if target.exists() else 0
|
|
if offset > item["size"]:
|
|
target.unlink()
|
|
offset = 0
|
|
if offset == item["size"] and target.exists() and file_digest(target) != item["sha256"]:
|
|
target.unlink()
|
|
offset = 0
|
|
if offset < item["size"] or not target.exists():
|
|
url = payload["filesBaseUrl"].rstrip("/") + "/" + urllib.parse.quote(item["path"], safe="/")
|
|
with self.artifact_open("GET", url, extra_headers={"Range": f"bytes={offset}-"} if offset else {}) as response:
|
|
if offset and response.status == 206:
|
|
content_range = response.headers.get("Content-Range", "")
|
|
if not content_range.startswith(f"bytes {offset}-"):
|
|
raise AgentError("Invalid resumed download response")
|
|
elif offset:
|
|
offset = 0
|
|
with target.open("ab" if offset else "wb") as output:
|
|
received = offset
|
|
while True:
|
|
chunk = response.read(1024 * 1024)
|
|
if not chunk:
|
|
break
|
|
received += len(chunk)
|
|
if received > item["size"]:
|
|
raise AgentError("Library download exceeded its manifest size")
|
|
output.write(chunk)
|
|
progress["bytesDownloaded"] = downloaded + received
|
|
output.flush()
|
|
os.fsync(output.fileno())
|
|
progress["status"] = "verifying"
|
|
if target.stat().st_size != item["size"] or file_digest(target) != item["sha256"]:
|
|
raise AgentError("Downloaded library file failed integrity verification")
|
|
downloaded += item["size"]
|
|
progress.update(status="downloading", bytesDownloaded=downloaded)
|
|
saved = {"libraryId": library_id, "game": manifest["game"], "size": size, "files": files, "manifestSha256": manifest_sha}
|
|
atomic_json(staging / "manifest.json", saved)
|
|
for path in (staging / "content").rglob("*"):
|
|
os.chmod(path, 0o555 if path.is_dir() else 0o444)
|
|
os.chmod(staging / "content", 0o555)
|
|
os.replace(staging, directory)
|
|
self.library_progress.pop(library_id, None)
|
|
return {"libraryId": library_id, "state": "ready", "size": size, "manifestSha256": manifest_sha}
|
|
except Exception:
|
|
progress.update(status="failed", error="Library download or integrity verification failed")
|
|
raise
|
|
|
|
def library_import(self, payload):
|
|
game = payload.get("game")
|
|
volume = payload.get("sourceVolume")
|
|
if game not in ("t5", "t6") or volume != game + "_game_files":
|
|
raise AgentError("Only the explicitly approved game-file Docker volumes can be imported")
|
|
raw = self.docker.call("volume", "inspect", volume)
|
|
source = Path(json.loads(raw)[0]["Mountpoint"])
|
|
if not source.is_dir() or source.is_symlink():
|
|
raise AgentError("Import volume is unavailable")
|
|
files = []
|
|
for path in sorted(source.rglob("*")):
|
|
if path.is_symlink():
|
|
raise AgentError("Import source contains a symlink")
|
|
if path.is_file():
|
|
files.append({"path": path.relative_to(source).as_posix(), "size": path.stat().st_size, "sha256": file_digest(path)})
|
|
elif not path.is_dir():
|
|
raise AgentError("Import source contains a special file")
|
|
if not files:
|
|
raise AgentError("Import volume is empty")
|
|
with self.artifact_open("PUT", payload["manifestUrl"], json.dumps({"files": files}).encode(), {"Content-Type": "application/json"}) as response:
|
|
response.read()
|
|
for item in files:
|
|
path = scoped(source, item["path"])
|
|
if path.stat().st_size != item["size"] or file_digest(path) != item["sha256"]:
|
|
raise AgentError("Import source changed while publishing")
|
|
url = payload["uploadBaseUrl"].rstrip("/") + "/" + urllib.parse.quote(item["path"], safe="/")
|
|
with path.open("rb") as stream, self.artifact_open("PUT", url, stream, {"Content-Type": "application/octet-stream", "Content-Length": str(item["size"])}) as response:
|
|
response.read()
|
|
self.checkpoint({"phase": "publishing-import"})
|
|
return self.complete_import(payload, start=True)
|
|
|
|
def complete_import(self, payload, start=False):
|
|
method = "POST" if start else "GET"
|
|
while not self.stopping.is_set():
|
|
with self.artifact_open(method, payload["completeUrl"], b"{}" if method == "POST" else None, {"Content-Type": "application/json"}) as response:
|
|
result = json.load(response)
|
|
if result.get("status") == "published":
|
|
return result
|
|
if result.get("status") == "failed":
|
|
raise Rejected("Controller rejected the game-file import after integrity verification")
|
|
if result.get("status") not in ("pending", "verifying"):
|
|
raise AgentError("Controller returned an invalid import publication state")
|
|
method = "POST" if result["status"] == "pending" else "GET"
|
|
self.stopping.wait(3)
|
|
raise AgentError("Agent stopped while the library import was being published")
|
|
|
|
def dispatch(self, operation):
|
|
kind, payload = operation.get("kind"), operation.get("payload", {})
|
|
if not isinstance(kind, str) or not isinstance(payload, dict):
|
|
raise AgentError("Invalid operation")
|
|
if kind == "credentials.rotate":
|
|
pending = "ec_host_" + base64.urlsafe_b64encode(os.urandom(32)).rstrip(b"=").decode()
|
|
self.config["pendingCredential"] = pending
|
|
atomic_json(self.config_path, self.config)
|
|
self.checkpoint({"phase": "rotating"})
|
|
try:
|
|
self.request("POST", "/api/agent/credentials", {"credential": pending})
|
|
except AgentError:
|
|
if not self.recover_credential():
|
|
raise
|
|
self.config["credential"] = pending
|
|
self.config.pop("pendingCredential", None)
|
|
atomic_json(self.config_path, self.config)
|
|
return {"rotated": True}
|
|
if kind == "runtime.pull":
|
|
image = payload.get("image")
|
|
if not isinstance(image, str) or not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._/:@-]{0,511}", image):
|
|
raise AgentError("Invalid runtime image")
|
|
self.docker.call("image", "pull", image, timeout=3600)
|
|
image_id = self.docker.call("image", "inspect", "--format", "{{.Id}}", image).strip()
|
|
aliases_file = self.root / "agent" / "runtime-images.json"
|
|
aliases = json.loads(aliases_file.read_text()) if aliases_file.exists() else {}
|
|
aliases[image] = image_id
|
|
atomic_json(aliases_file, aliases)
|
|
return {"image": image, "ready": True}
|
|
if kind == "host.inspect":
|
|
return self.inventory()
|
|
if kind == "library.download":
|
|
return self.library_download(payload)
|
|
if kind == "library.import":
|
|
return self.library_import(payload)
|
|
if kind == "library.inspect":
|
|
return self.verify_library(self.library_dir(payload["game"], payload["libraryId"]))
|
|
instance_id, generation = instance_identity(operation.get("instanceId"), operation.get("generation"))
|
|
current_marker = self.instance_dir(instance_id, generation).parent / "current.json"
|
|
if current_marker.exists():
|
|
current = json.loads(current_marker.read_text())
|
|
if current["generation"] > generation or current["generation"] == generation and current.get("removed") and kind not in ("instance.inspect", "instance.remove"):
|
|
raise AgentError("Instance generation has been superseded or removed")
|
|
if kind.startswith("instance."):
|
|
return self.lifecycle(kind, instance_id, generation, payload)
|
|
if kind.startswith("files.") or kind == "logs.read":
|
|
return self.file_operation(kind, instance_id, generation, payload, operation["id"])
|
|
if kind == "rcon.execute":
|
|
if payload["command"] not in ("status", "getstatus", "version"):
|
|
self.checkpoint({"phase": "rcon-command"})
|
|
return self.rcon(instance_id, generation, payload["command"])
|
|
raise AgentError("Unsupported operation kind")
|
|
|
|
def execute(self, operation, report_running=False):
|
|
row, _ = self.journal.accept(operation)
|
|
if row["state"] == "unknown":
|
|
return self.reconcile(operation, row)
|
|
if row["state"] != "accepted":
|
|
return row
|
|
running = self.journal.update(operation["id"], "running")
|
|
if report_running:
|
|
try:
|
|
self.report(running)
|
|
except AgentError:
|
|
return self.journal.update(operation["id"], "accepted", {"blocked": "controller-unavailable", "message": "Waiting for controller contact before executing the operation"})
|
|
self.executing_id = operation["id"]
|
|
try:
|
|
result = self.dispatch(operation)
|
|
except Held as err:
|
|
return self.journal.update(operation["id"], "accepted", {"blocked": "occupied" if "Players" in str(err) else "unknown", "message": str(err)})
|
|
except Exception as err:
|
|
# The command may already have changed Docker or files. Never infer
|
|
# failure means nothing happened, and never retry without inspection.
|
|
message = str(err) if isinstance(err, AgentError) else "Operation could not be completed; inspect its current state"
|
|
checkpoint = self.journal.get(operation["id"]).get("checkpoint")
|
|
return self.journal.update(operation["id"], "failed" if isinstance(err, Rejected) or not checkpoint else "unknown", error=message)
|
|
finally:
|
|
self.executing_id = None
|
|
return self.journal.update(operation["id"], "completed", result)
|
|
|
|
def recover_credential(self):
|
|
pending = self.config.get("pendingCredential")
|
|
if not pending:
|
|
return False
|
|
previous = self.config["credential"]
|
|
self.config["credential"] = pending
|
|
try:
|
|
self.request("POST", "/api/agent/credentials", {"credential": pending})
|
|
except AgentError:
|
|
self.config["credential"] = previous
|
|
try:
|
|
self.request("POST", "/api/agent/credentials", {"credential": pending})
|
|
except AgentError:
|
|
return False
|
|
self.config["credential"] = pending
|
|
self.config.pop("pendingCredential", None)
|
|
atomic_json(self.config_path, self.config)
|
|
return True
|
|
|
|
def reconcile(self, operation, row):
|
|
"""Prove desired state with immutable ownership; never resubmit a command."""
|
|
kind, payload = operation["kind"], operation.get("payload", {})
|
|
instance_id, generation = operation.get("instanceId"), operation.get("generation")
|
|
checkpoint = json.loads(row["checkpoint"]) if row.get("checkpoint") else {}
|
|
try:
|
|
if kind == "credentials.rotate":
|
|
if not self.config.get("pendingCredential") or self.recover_credential():
|
|
return self.journal.update(operation["id"], "completed", {"rotated": True, "reconciled": True})
|
|
return row
|
|
if kind == "library.import":
|
|
try:
|
|
with self.artifact_open("GET", payload["completeUrl"]) as response:
|
|
status = json.load(response)
|
|
if status.get("status") == "published":
|
|
return self.journal.update(operation["id"], "completed", status)
|
|
if checkpoint.get("phase") == "publishing-import" or status.get("status") == "verifying":
|
|
return self.journal.update(operation["id"], "completed", self.complete_import(payload))
|
|
# Import uploads use an immutable import ID and per-file
|
|
# checksums. Reconcile their staged files after interruption.
|
|
self.journal.update(operation["id"], "running")
|
|
self.executing_id = operation["id"]
|
|
return self.journal.update(operation["id"], "completed", self.library_import(payload))
|
|
except Rejected as err:
|
|
return self.journal.update(operation["id"], "failed", error=str(err))
|
|
except Exception:
|
|
return self.journal.update(operation["id"], "unknown", error="Import reconciliation is waiting for controller contact")
|
|
finally:
|
|
self.executing_id = None
|
|
if kind == "library.download":
|
|
library_id = payload.get("libraryVersionId", payload.get("libraryId"))
|
|
for game in ("t5", "t6"):
|
|
directory = self.library_dir(game, library_id)
|
|
if directory.exists():
|
|
saved = self.verify_library(directory)
|
|
return self.journal.update(operation["id"], "completed", {"libraryId": library_id, "state": "ready", "size": saved["size"], "manifestSha256": saved.get("manifestSha256"), "reconciled": True})
|
|
# This is a resumable immutable transfer, not a repeated
|
|
# container command: verified completed files remain untouched.
|
|
self.journal.update(operation["id"], "running")
|
|
try:
|
|
return self.journal.update(operation["id"], "completed", self.library_download(payload))
|
|
except Exception:
|
|
return self.journal.update(operation["id"], "failed", error="Interrupted library download could not be resumed")
|
|
if kind == "files.stage":
|
|
return self.journal.update(operation["id"], "completed", self.stage(instance_id, generation, payload, operation["id"]))
|
|
if kind in ("host.inspect", "instance.inspect", "files.read", "files.list", "files.stat", "logs.read") or kind == "rcon.execute" and payload.get("command") in ("status", "getstatus", "version"):
|
|
return self.journal.update(operation["id"], "completed", self.dispatch(operation))
|
|
if kind in ("files.activate", "files.rollback") and checkpoint:
|
|
stage = self.instance_dir(instance_id, generation) / "stages" / identifier(payload["stageId"]) / "manifest.json"
|
|
saved = json.loads(stage.read_text())
|
|
expected = "activated" if kind == "files.activate" else "rolled-back"
|
|
if saved["state"] == expected:
|
|
health = self.inspect(instance_id, generation)
|
|
restarted = health["running"] and health.get("containerId") == checkpoint.get("containerId") and bool(checkpoint.get("startedAt")) and health.get("startedAt") != checkpoint["startedAt"]
|
|
if not payload.get("restart", True) or restarted:
|
|
return self.journal.update(operation["id"], "completed", {"activated": expected == "activated", "rolledBack": expected == "rolled-back", "before": saved["before"], "reconciled": True})
|
|
return row
|
|
if not kind.startswith("instance.") or not checkpoint:
|
|
return row
|
|
observed_generation = payload.get("nextGeneration", payload.get("newGeneration")) if kind == "instance.rebuild" else generation
|
|
health = self.inspect(instance_id, observed_generation)
|
|
before = []
|
|
if payload.get("stageId"):
|
|
stage = self.instance_dir(instance_id, generation) / "stages" / identifier(payload["stageId"]) / "manifest.json"
|
|
saved = json.loads(stage.read_text())
|
|
if saved["state"] != "activated":
|
|
return row
|
|
before = saved["before"]
|
|
proven = False
|
|
if kind in ("instance.create", "instance.rebuild"):
|
|
proven = health["running"] and health.get("creationOperation") == operation["id"] and health.get("specSha256") == checkpoint.get("specSha256")
|
|
if proven:
|
|
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"]:
|
|
old = self.instance_dir(instance_id, generation)
|
|
if old.exists():
|
|
shutil.rmtree(old)
|
|
elif kind in ("instance.start", "instance.stop"):
|
|
proven = health.get("containerId") == checkpoint.get("containerId") and (health["running"] if kind == "instance.start" else health["state"] in ("stopped", "exited", "created", "dead"))
|
|
if kind == "instance.start":
|
|
proven = proven and checkpoint.get("phase") == "starting"
|
|
if checkpoint.get("restartRequired"):
|
|
proven = proven and bool(checkpoint.get("startedAt")) and bool(health.get("startedAt")) and health["startedAt"] != checkpoint["startedAt"]
|
|
elif kind == "instance.restart":
|
|
proven = health["running"] and health.get("containerId") == checkpoint.get("containerId") and bool(checkpoint.get("startedAt")) and bool(health.get("startedAt")) and health["startedAt"] != checkpoint["startedAt"]
|
|
elif kind == "instance.remove":
|
|
proven = not health["exists"]
|
|
if proven:
|
|
directory = self.instance_dir(instance_id, generation)
|
|
if directory.exists():
|
|
shutil.rmtree(directory)
|
|
self.mark_removed(instance_id, generation)
|
|
if proven:
|
|
return self.journal.update(operation["id"], "completed", {**health, "before": before, "reconciled": True})
|
|
except (AgentError, OSError, ValueError, KeyError):
|
|
pass
|
|
return row
|
|
|
|
def report(self, row):
|
|
payload = {"state": row["state"]}
|
|
if row.get("result"):
|
|
payload["result"] = json.loads(row["result"])
|
|
if row["state"] == "accepted" and isinstance(payload["result"], dict) and payload["result"].get("blocked"):
|
|
# The controller sees a durable running/waiting operation;
|
|
# locally accepted means its mutation gate can safely retry.
|
|
payload["state"] = "running"
|
|
if row.get("error"):
|
|
payload["error"] = row["error"]
|
|
return self.request("POST", "/api/agent/operations/" + urllib.parse.quote(row["id"], safe=""), payload)
|
|
|
|
def heartbeat(self):
|
|
response = self.request("POST", "/api/agent/heartbeat", {"inventory": self.inventory(), "operationId": self.current_operation})
|
|
host = response.get("host", {})
|
|
if isinstance(host.get("portMin"), int) and isinstance(host.get("portMax"), int):
|
|
self.config["portRange"] = {"min": host["portMin"], "max": host["portMax"]}
|
|
|
|
def heartbeat_loop(self):
|
|
while not self.stopping.wait(15):
|
|
try:
|
|
self.heartbeat()
|
|
except AgentError:
|
|
pass
|
|
|
|
def run(self):
|
|
if self.config.get("pendingCredential"):
|
|
self.recover_credential()
|
|
# Obtain the configured host port range before accepting any deployment.
|
|
# Enrollment's default values must never race the first heartbeat.
|
|
while not self.stopping.is_set():
|
|
try:
|
|
self.heartbeat()
|
|
break
|
|
except AgentError:
|
|
self.stopping.wait(3)
|
|
heartbeats = threading.Thread(target=self.heartbeat_loop, daemon=True)
|
|
heartbeats.start()
|
|
while not self.stopping.is_set():
|
|
try:
|
|
if self.config.get("pendingCredential"):
|
|
self.recover_credential()
|
|
response = self.request("GET", "/api/agent/operations")
|
|
for operation in response.get("operations", []):
|
|
if self.stopping.is_set():
|
|
break
|
|
self.current_operation = operation["id"]
|
|
accepted, _ = self.journal.accept(operation)
|
|
self.report(accepted)
|
|
self.report(self.execute(operation, report_running=True))
|
|
self.current_operation = None
|
|
except (AgentError, OSError, ValueError, KeyError):
|
|
# Logs deliberately omit URLs, tokens, payloads and raw errors.
|
|
print("Host controller contact unavailable; existing containers continue running", file=sys.stderr)
|
|
self.stopping.wait(3)
|
|
heartbeats.join(timeout=2)
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--config", default="/etc/electric-cherry/agent.json")
|
|
parser.add_argument("--enroll", action="store_true")
|
|
parser.add_argument("--controller")
|
|
parser.add_argument("--token-file")
|
|
parser.add_argument("--root", default="/var/lib/electric-cherry")
|
|
parser.add_argument("--force-enroll", action="store_true")
|
|
parser.add_argument("--version", action="version", version=VERSION)
|
|
args = parser.parse_args()
|
|
if args.enroll:
|
|
if not args.controller or not args.token_file:
|
|
parser.error("Enrollment needs --controller and --token-file")
|
|
if Path(args.config).exists() and not args.force_enroll:
|
|
existing = json.loads(Path(args.config).read_text())
|
|
if existing.get("credential"):
|
|
if existing.get("controller", "").rstrip("/") != args.controller.rstrip("/") or Path(existing.get("root", "")).resolve() != Path(args.root).resolve():
|
|
raise AgentError("Existing enrollment uses a different controller or data directory")
|
|
print("Host is already enrolled; retaining existing credentials")
|
|
return
|
|
agent = Agent({"controller": args.controller.rstrip("/"), "root": args.root})
|
|
token = Path(args.token_file).read_text().strip()
|
|
enrolled = agent.request("POST", "/api/agent/enroll", {"token": token, "inventory": agent.inventory()}, credential=False)
|
|
if not enrolled.get("hostId") or not enrolled.get("credential"):
|
|
raise AgentError("Controller returned invalid enrollment credentials")
|
|
atomic_json(args.config, {"controller": args.controller.rstrip("/"), "root": args.root, "hostId": enrolled["hostId"], "credential": enrolled["credential"], "portRange": enrolled.get("portRange", {"min": 28960, "max": 29060})})
|
|
agent.journal.close()
|
|
print("Host enrollment completed")
|
|
return
|
|
config = json.loads(Path(args.config).read_text())
|
|
config["configPath"] = args.config
|
|
lock_root = Path(config.get("root", "/var/lib/electric-cherry")) / "agent"
|
|
lock_root.mkdir(parents=True, exist_ok=True)
|
|
lock = open(lock_root / "process.lock", "a")
|
|
try:
|
|
fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
|
except BlockingIOError:
|
|
raise AgentError("Another host agent already owns this data directory") from None
|
|
agent = Agent(config)
|
|
for sig in (signal.SIGINT, signal.SIGTERM):
|
|
signal.signal(sig, lambda *_: agent.stopping.set())
|
|
try:
|
|
agent.run()
|
|
finally:
|
|
agent.journal.close()
|
|
lock.close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
main()
|
|
except (AgentError, OSError, ValueError, KeyError):
|
|
print("Electric Cherry host agent failed; check configuration and controller availability", file=sys.stderr)
|
|
sys.exit(1)
|