feat(selfhost): add persistent host update service

scripts/selfhost-update-service.py

@@ -0,0 +1,373 @@
+#!/usr/bin/env python3
+"""Local HTTP/Unix-socket control service for the FEDEO host updater. No TCP listener."""
+import argparse
+import contextlib
+import fcntl
+from http.server import BaseHTTPRequestHandler
+import importlib.util
+import json
+import os
+from pathlib import Path
+import re
+import signal
+import socketserver
+import stat
+import threading
+import traceback
+
+spec = importlib.util.spec_from_file_location("fedeo_updater", Path(__file__).with_name("selfhost-update.py"))
+engine = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(engine)
+
+REQUEST_ID = re.compile(r"[A-Za-z0-9][A-Za-z0-9_-]{7,79}\Z")
+PHASES = {"preparing", "validating", "checked", "pulling", "maintenance", "backed_up", "applying",
+          "retiring_matrix", "migrating", "starting", "verifying", "complete", "resolved"}
+STATUSES = {"idle", "queued", "running", "checked", "up_to_date", "succeeded", "failed", "interrupted", "recovered"}
+ERRORS = {
+    "INVALID_REQUEST": "Ungültige Anfrage.",
+    "NOT_FOUND": "Auftrag oder Endpunkt nicht gefunden.",
+    "BUSY": "Eine Prüfung oder ein Update läuft bereits.",
+    "SHUTTING_DOWN": "Der Dienst wird beendet.",
+    "REQUEST_CONFLICT": "Diese Auftrags-ID wurde für eine andere Anfrage verwendet.",
+    "CHECK_REQUIRED": "Zuerst genau diese Zielversion erfolgreich prüfen.",
+    "COMPOSE_REVIEW_REQUIRED": "Die Compose-Datei muss zuerst auf dem Host geprüft und übernommen werden.",
+    "RECOVERY_REQUIRED": "Ein vorheriges Update benötigt Wiederherstellung auf dem Host.",
+    "PLAN_CHANGED": "Release oder Konfiguration haben sich geändert. Erneut prüfen.",
+    "UPDATE_FAILED": "Auftrag fehlgeschlagen. Geschützte Protokolle auf dem Host prüfen.",
+    "INTERRUPTED": "Der Auftrag wurde unterbrochen. Status auf dem Host prüfen.",
+}
+
+
+class ApiError(Exception):
+    def __init__(self, code, status=400):
+        self.code, self.status = code, status
+
+
+def identifier(value):
+    if not isinstance(value, str) or not REQUEST_ID.fullmatch(value):
+        raise ApiError("INVALID_REQUEST")
+    return value
+
+
+def public_state(state):
+    """An allowlist, never raw command logs, exceptions, host paths, config or diff contents."""
+    result = {"status": state.get("status") if state.get("status") in STATUSES else "idle"}
+    if state.get("phase") in PHASES:
+        result["phase"] = state["phase"]
+    for key in ("recoveryRequired", "maintenance", "mutationStarted"):
+        if key in state:
+            result[key] = bool(state[key])
+    for key in ("version", "previousVersion"):
+        if isinstance(state.get(key), str) and engine.VERSION.fullmatch(state[key]):
+            result[key] = state[key]
+    if re.fullmatch(r"[a-f0-9]{40}", str(state.get("revision", ""))):
+        result["revision"] = state["revision"]
+    for key in ("startedAt", "updatedAt"):
+        if isinstance(state.get(key), str) and re.fullmatch(r"[0-9T:+.Z-]{10,40}", state[key]):
+            result[key] = state[key]
+    if state.get("requestId") and REQUEST_ID.fullmatch(str(state["requestId"])):
+        result["requestId"] = state["requestId"]
+    if state.get("command") in ("check", "apply"):
+        result["command"] = state["command"]
+    return result
+
+
+def public_plan(plan):
+    result = {k: v for k, v in public_state(plan).items() if k != "status"}
+    for key in ("updateAvailable", "requiresComposeAdoption"):
+        result[key] = bool(plan.get(key))
+    if type(plan.get("estimatedBackupBytes")) is int:
+        result["estimatedBackupBytes"] = plan["estimatedBackupBytes"]
+    result["retiredServices"] = sorted(set(plan.get("retiredServices", [])) & engine.LEGACY_MATRIX_SERVICES)
+    return result
+
+
+class Controller:
+    def __init__(self, directory, *, updater_factory=engine.Updater):
+        self.directory = Path(directory).resolve()
+        self.factory = updater_factory
+        updater = self.factory(self.directory)
+        self.home = updater.home / "service"
+        self.home.mkdir(mode=0o700, exist_ok=True)
+        if self.home.is_symlink():
+            raise engine.UpdateError("Dienstverzeichnis darf kein symbolischer Link sein.")
+        os.chmod(self.home, 0o700)
+        self.gate = threading.Lock()
+        self.worker = None
+        self.closing = False
+        self.active = None
+        self.daemon_lock = (self.home / "daemon.lock").open("a")
+        try:
+            fcntl.flock(self.daemon_lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        except OSError:
+            self.daemon_lock.close()
+            raise engine.UpdateError("Für diese Instanz läuft bereits ein Update-Dienst.") from None
+        # A persisted request is an audit record, not a queue that silently replays updates.
+        try:
+            for path in self.home.glob("*.json"):
+                record = engine.read_json(path)
+                if record.get("status") in ("queued", "running"):
+                    record.update(status="interrupted", errorCode="INTERRUPTED", updatedAt=engine.now())
+                    engine.write_json(path, record)
+            updater.status()  # Detect an abandoned engine job without touching an active CLI job.
+        except BaseException:
+            self.daemon_lock.close()
+            raise
+
+    def path(self, request_id):
+        return self.home / (identifier(request_id) + ".json")
+
+    def read(self, request_id):
+        record = engine.read_json(self.path(request_id))
+        if not record:
+            raise ApiError("NOT_FOUND", 404)
+        return record
+
+    def request_status(self, request_id):
+        record = self.read(request_id)
+        result = public_state(record)
+        if record.get("plan"):
+            result["plan"] = public_plan(record["plan"])
+        code = record.get("errorCode")
+        if code in ERRORS:
+            result["error"] = {"code": code, "message": ERRORS[code]}
+        if record["status"] == "running":
+            state = self.factory(self.directory).status()
+            if state.get("requestId") == request_id:
+                result["progress"] = public_state(state)
+        return result
+
+    def status(self):
+        updater = self.factory(self.directory)
+        result = {"apiVersion": 1, "engine": public_state(updater.status()), "acceptingRequests": not self.closing}
+        installed = engine.read_json(updater.home / "installed.json", {}).get("manifest", {})
+        result["installed"] = {k: v for k, v in public_state({"version": installed.get("version"),
+                              "revision": installed.get("commit")}).items() if k != "status"}
+        if self.active:
+            result["request"] = self.request_status(self.active)
+        return result
+
+    def submit(self, command, body):
+        if not isinstance(body, dict) or command not in ("check", "apply"):
+            raise ApiError("INVALID_REQUEST")
+        allowed = {"requestId", "version"} | ({"checkRequestId"} if command == "apply" else set())
+        if set(body) - allowed:
+            raise ApiError("INVALID_REQUEST")
+        request_id = identifier(body.get("requestId"))
+        version = body.get("version")
+        if version is not None and (not isinstance(version, str) or not engine.VERSION.fullmatch(version)):
+            raise ApiError("INVALID_REQUEST")
+        if command == "apply" and not version:
+            raise ApiError("INVALID_REQUEST")
+        check_id = identifier(body.get("checkRequestId")) if command == "apply" else None
+        payload = {"command": command, "version": version, "checkRequestId": check_id}
+        with self.gate:
+            previous = engine.read_json(self.path(request_id))
+            if previous:
+                if previous["input"] != payload:
+                    raise ApiError("REQUEST_CONFLICT", 409)
+                return 200, self.request_status(request_id)
+            if self.closing:
+                raise ApiError("SHUTTING_DOWN", 503)
+            if self.worker and self.worker.is_alive():
+                raise ApiError("BUSY", 409)
+            state = self.factory(self.directory).status()
+            if state.get("recoveryRequired"):
+                raise ApiError("RECOVERY_REQUIRED", 409)
+            if state.get("status") == "running":
+                raise ApiError("BUSY", 409)
+            fingerprint = None
+            if command == "apply":
+                try:
+                    check = self.read(check_id)
+                except ApiError:
+                    raise ApiError("CHECK_REQUIRED", 409) from None
+                plan = check.get("plan", {})
+                if (check.get("command") != "check" or check.get("status") != "checked"
+                        or plan.get("version") != version or not plan.get("updateAvailable")
+                        or not re.fullmatch(r"[a-f0-9]{64}", str(plan.get("approvalFingerprint", "")))):
+                    raise ApiError("CHECK_REQUIRED", 409)
+                if plan.get("requiresComposeAdoption"):
+                    raise ApiError("COMPOSE_REVIEW_REQUIRED", 409)
+                fingerprint = plan["approvalFingerprint"]
+            record = {"requestId": request_id, "command": command, "input": payload,
+                      "status": "queued", "startedAt": engine.now(), "updatedAt": engine.now()}
+            engine.write_json(self.path(request_id), record)
+            self.active = request_id
+            self.worker = threading.Thread(target=self.run, args=(record, fingerprint), daemon=False)
+            try:
+                self.worker.start()
+            except BaseException:
+                record.update(status="failed", errorCode="UPDATE_FAILED", updatedAt=engine.now())
+                engine.write_json(self.path(request_id), record)
+                raise
+            return 202, public_state(record)
+
+    def run(self, record, fingerprint):
+        request_id = record["requestId"]
+        try:
+            record.update(status="running", updatedAt=engine.now())
+            engine.write_json(self.path(request_id), record)
+            updater = self.factory(self.directory)
+            result = updater.execute(record["command"], record["input"]["version"],
+                                     expected_fingerprint=fingerprint, request_id=request_id)
+            record.update(status=result["status"], plan=result["plan"])
+            if record["command"] == "check" and not result["plan"]["updateAvailable"]:
+                record["status"] = "up_to_date"
+        except BaseException as error:
+            code = "PLAN_CHANGED" if isinstance(error, engine.PlanChangedError) else "UPDATE_FAILED"
+            record.update(status="failed", errorCode=code)
+            with (self.home / (request_id + ".log")).open("a") as log:
+                os.chmod(log.name, 0o600)
+                traceback.print_exc(file=log)
+        finally:
+            record["updatedAt"] = engine.now()
+            engine.write_json(self.path(request_id), record)
+
+    def close(self):
+        with self.gate:
+            self.closing = True
+            worker = self.worker
+        if worker:
+            worker.join()  # Normal service stop drains the active update instead of interrupting a migration.
+        self.daemon_lock.close()
+
+
+class Handler(BaseHTTPRequestHandler):
+    protocol_version = "HTTP/1.0"
+    server_version = "FEDEO-Updater/1"
+    sys_version = ""
+
+    def setup(self):
+        super().setup()
+        self.connection.settimeout(5)
+
+    def log_message(self, *_):
+        pass
+
+    def send_json(self, status, body):
+        data = json.dumps(body, ensure_ascii=False).encode()
+        self.close_connection = True
+        self.send_response(status)
+        self.send_header("Content-Type", "application/json; charset=utf-8")
+        self.send_header("Content-Length", str(len(data)))
+        self.send_header("Cache-Control", "no-store")
+        self.send_header("Connection", "close")
+        self.end_headers()
+        self.wfile.write(data)
+
+    def dispatch(self):
+        try:
+            if self.headers.get("Origin") or self.headers.get("Transfer-Encoding"):
+                raise ApiError("INVALID_REQUEST")
+            controller = self.server.controller
+            if self.command == "GET" and self.path == "/v1/status":
+                self.send_json(200, controller.status())
+            elif self.command == "GET" and self.path.startswith("/v1/requests/"):
+                self.send_json(200, controller.request_status(self.path.removeprefix("/v1/requests/")))
+            elif self.command == "POST" and self.path in ("/v1/check", "/v1/apply"):
+                lengths = self.headers.get_all("Content-Length", [])
+                if (len(lengths) != 1 or not lengths[0].isdigit() or not 1 <= int(lengths[0]) <= 4096
+                        or self.headers.get_content_type() != "application/json"):
+                    raise ApiError("INVALID_REQUEST")
+                body = json.loads(self.rfile.read(int(lengths[0])))
+                status, result = controller.submit(self.path.rsplit("/", 1)[-1], body)
+                self.send_json(status, result)
+            else:
+                raise ApiError("NOT_FOUND", 404)
+        except ApiError as error:
+            self.send_json(error.status, {"error": {"code": error.code, "message": ERRORS[error.code]}})
+        except (ValueError, UnicodeError):
+            self.send_json(400, {"error": {"code": "INVALID_REQUEST", "message": ERRORS["INVALID_REQUEST"]}})
+        except (OSError, engine.UpdateError, KeyError, TypeError):
+            with contextlib.suppress(OSError):
+                self.send_json(500, {"error": {"code": "UPDATE_FAILED", "message": ERRORS["UPDATE_FAILED"]}})
+
+    do_GET = dispatch
+    do_POST = dispatch
+
+
+class Server(socketserver.ThreadingMixIn, socketserver.UnixStreamServer):
+    daemon_threads = True
+    request_queue_size = 8
+
+    def __init__(self, path, controller):
+        self.controller = controller
+        self.slots = threading.BoundedSemaphore(16)
+        super().__init__(str(path), Handler)
+        os.chmod(path, 0o660)
+
+    def process_request(self, request, client_address):
+        if not self.slots.acquire(blocking=False):
+            self.shutdown_request(request)
+            return
+        try:
+            super().process_request(request, client_address)
+        except BaseException:
+            self.slots.release()
+            raise
+
+    def process_request_thread(self, request, client_address):
+        try:
+            super().process_request_thread(request, client_address)
+        finally:
+            self.slots.release()
+
+    def handle_error(self, *_):
+        pass  # Do not send request data or tracebacks to the journal.
+
+
+def serve(directory, socket_path):
+    controller = None
+    path = Path(socket_path)
+    stop = threading.Event()
+    server = None
+    socket_lock = None
+    try:
+        path.parent.mkdir(mode=0o750, parents=True, exist_ok=True)
+        parent = path.parent.stat()
+        if path.parent.is_symlink() or parent.st_uid != 0 or parent.st_mode & 0o022:
+            raise engine.UpdateError("Socket-Verzeichnis muss root gehören und darf nicht durch andere beschreibbar sein.")
+        socket_lock = path.with_name(path.name + ".lock").open("a")
+        try:
+            fcntl.flock(socket_lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        except OSError:
+            raise engine.UpdateError("An diesem Socket läuft bereits ein Update-Dienst.") from None
+        controller = Controller(directory)
+        if path.is_symlink() or (path.exists() and not stat.S_ISSOCK(path.stat().st_mode)):
+            raise engine.UpdateError("Socket-Pfad ist bereits anderweitig belegt.")
+        # Both the instance and endpoint are locked before replacing a stale socket.
+        path.unlink(missing_ok=True)
+        server = Server(path, controller)
+        for sig in (signal.SIGTERM, signal.SIGINT):
+            signal.signal(sig, lambda *_: stop.set())
+        server.timeout = 0.5
+        while not stop.is_set():
+            server.handle_request()
+    finally:
+        if controller:
+            controller.close()
+        if server:
+            server.server_close()
+            path.unlink(missing_ok=True)
+        if socket_lock:
+            socket_lock.close()
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("command", choices=("serve",))
+    parser.add_argument("--directory", default="/opt/fedeo")
+    parser.add_argument("--socket", default="/run/fedeo-updater/control.sock")
+    args = parser.parse_args()
+    if os.geteuid() != 0:
+        parser.error("Der Host-Dienst muss als root laufen.")
+    os.umask(0o077)
+    try:
+        serve(args.directory, args.socket)
+    except (OSError, engine.UpdateError) as error:
+        parser.exit(1, f"Update-Dienst konnte nicht gestartet werden: {error}\n")
+
+
+if __name__ == "__main__":
+    main()