#!/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()