#!/usr/bin/env python3 """FEDEO host-side update engine. Python 3.10+, Docker Compose 2.24+; no pip packages. The CLI and a future host service share Updater. No daemon or HTTP API is installed. All commands operate on an existing local selfhost installation, never a git checkout. """ import argparse import contextlib import datetime import difflib import fcntl import hashlib import json import os from pathlib import Path import re import shutil import signal import subprocess import sys import time import urllib.error import urllib.parse import urllib.request import uuid REPOSITORY = "https://git.federspiel.tech/flfeders/FEDEO" REGISTRY = "git.federspiel.tech/flfeders/fedeo" VERSION = re.compile(r"(0|[1-9][0-9]*)\.(0|[1-9][0-9]*)\.(0|[1-9][0-9]*)\Z") FILES = ("release-manifest.json", "docker-compose.selfhost.yml", ".env.example", "selfhost-setup.sh", "CHANGELOG.md") GIB = 1024 ** 3 LEGACY_MATRIX_SERVICES = {"matrix-db", "matrix-synapse", "matrix-redis", "matrix-element", "matrix-livekit", "matrix-rtc-jwt", "matrix-turn", "matrix-well-known"} class UpdateError(Exception): pass class PlanChangedError(UpdateError): pass def now(): return datetime.datetime.now(datetime.timezone.utc).isoformat() def sha(data): return hashlib.sha256(data).hexdigest() def atomic(path, data): path = Path(path) path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) temp = path.with_name(path.name + ".tmp-" + uuid.uuid4().hex) try: with temp.open("xb") as stream: os.chmod(temp, 0o600) stream.write(data if isinstance(data, bytes) else data.encode()) stream.flush() os.fsync(stream.fileno()) os.replace(temp, path) fd = os.open(path.parent, os.O_RDONLY) try: os.fsync(fd) finally: os.close(fd) finally: temp.unlink(missing_ok=True) def write_json(path, data): atomic(path, json.dumps(data, ensure_ascii=False, indent=2) + "\n") def read_json(path, default=None): return json.loads(Path(path).read_text()) if Path(path).exists() else default def set_version(data, version): # Never source .env as a shell program. Preserve every other byte/setting. value = data.decode() pattern = r"(?m)^(?:export\s+)?FEDEO_VERSION=[^\r\n]*" if re.search(pattern, value): return re.sub(pattern, "FEDEO_VERSION=" + version, value).encode() return (value + ("" if value.endswith("\n") else "\n") + "FEDEO_VERSION=" + version + "\n").encode() class Https: def __init__(self, token=""): self.token = token def get(self, url, limit=8 * 1024 * 1024): class NoRedirect(urllib.request.HTTPRedirectHandler): def redirect_request(self, *args): return None opener = urllib.request.build_opener(NoRedirect) for _ in range(6): parsed = urllib.parse.urlsplit(url) if parsed.scheme != "https" or parsed.username or parsed.password: raise UpdateError("Download benötigt eine HTTPS-Adresse ohne Zugangsdaten.") headers = {"Accept": "application/json", "User-Agent": "FEDEO-Updater/1"} if self.token and parsed.netloc == urllib.parse.urlsplit(REPOSITORY).netloc: headers["Authorization"] = "token " + self.token try: with opener.open(urllib.request.Request(url, headers=headers), timeout=30) as response: data = response.read(limit + 1) if len(data) > limit: raise UpdateError("Release-Datei überschreitet die erlaubte Größe.") return data except urllib.error.HTTPError as error: if error.code in (301, 302, 303, 307, 308) and error.headers.get("Location"): url = urllib.parse.urljoin(url, error.headers["Location"]) continue raise UpdateError(f"Download fehlgeschlagen: HTTP {error.code}.") from None except (OSError, urllib.error.URLError): raise UpdateError("HTTPS-Anfrage fehlgeschlagen; Verbindung und Zertifikat prüfen.") from None raise UpdateError("Zu viele Download-Weiterleitungen.") def json(self, url): try: return json.loads(self.get(url)) except (ValueError, UnicodeError): raise UpdateError("Ungültige JSON-Antwort.") from None class Releases: def __init__(self, http): self.http = http self.api = REPOSITORY.replace("/flfeders/FEDEO", "/api/v1/repos/flfeders/FEDEO") def download(self, requested, destination): if requested: version = requested.removeprefix("v") if not VERSION.fullmatch(version): raise UpdateError("Eine stabile Version wie 3.1.0 angeben.") release = self.http.json(self.api + "/releases/tags/v" + version) else: releases = [] for page in range(1, 101): batch = self.http.json(self.api + f"/releases?limit=50&page={page}") releases.extend(r for r in batch if not r.get("draft") and not r.get("prerelease") and VERSION.fullmatch(r.get("tag_name", "").removeprefix("v"))) if len(batch) < 50: break if not releases: raise UpdateError("Kein veröffentlichter stabiler Release gefunden.") release = max(releases, key=lambda r: tuple(map(int, r["tag_name"].removeprefix("v").split(".")))) version = release["tag_name"].removeprefix("v") if release.get("draft") or release.get("prerelease") or release.get("tag_name") != "v" + version: raise UpdateError("Nur veröffentlichte stabile Releases sind erlaubt.") assets = self.http.json(self.api + f'/releases/{int(release["id"])}/assets') urls = {} for asset in assets: name = asset.get("name") if name in (*FILES, "SHA256SUMS"): url = asset.get("browser_download_url", "") if name in urls or urllib.parse.urlsplit(url).netloc != urllib.parse.urlsplit(REPOSITORY).netloc: raise UpdateError("Doppelte Release-Datei oder unerwartete Download-Herkunft.") urls[name] = url if any(name not in urls for name in (*FILES, "SHA256SUMS")): raise UpdateError("Release enthält nicht alle benötigten Betriebsdateien.") checksum_data = self.http.get(urls["SHA256SUMS"]) sums = {} for line in checksum_data.decode().splitlines(): match = re.fullmatch(r"([a-f0-9]{64}) ([^/\\]+)", line) if not match or match[2] in sums: raise UpdateError("Ungültige oder doppelte Prüfsummen.") sums[match[2]] = match[1] for name in FILES: data = self.http.get(urls[name]) if sha(data) != sums.get(name): raise UpdateError(f"Prüfsumme stimmt nicht: {name}.") atomic(destination / name, data) atomic(destination / "SHA256SUMS", checksum_data) manifest = read_json(destination / "release-manifest.json") tag = self.http.json(self.api + "/tags/v" + version) if (manifest.get("schemaVersion") != 1 or manifest.get("version") != version or manifest.get("tag") != "v" + version or manifest.get("repository") != REPOSITORY or not re.fullmatch(r"[a-f0-9]{40}", manifest.get("commit", "")) or manifest["commit"] != tag.get("commit", {}).get("sha")): raise UpdateError("Release-Manifest, Git-Tag und Version passen nicht zusammen.") for name in ("backend", "frontend"): images = [i for i in manifest.get("images", []) if i.get("name") == name] if (len(images) != 1 or images[0].get("repository") != REGISTRY + "/" + name or not re.fullmatch(r"sha256:[a-f0-9]{64}", images[0].get("digest", ""))): raise UpdateError(f"Ungültiger Image-Digest für {name}.") return manifest class Runner: def __init__(self, log): self.log = log # Compose interpolation must use the installation's .env, not host shell settings. allowed = {"PATH", "HOME", "LANG", "LC_ALL", "DOCKER_CONFIG", "DOCKER_HOST", "DOCKER_CONTEXT", "DOCKER_TLS_VERIFY", "DOCKER_CERT_PATH", "XDG_RUNTIME_DIR", "HTTP_PROXY", "HTTPS_PROXY", "NO_PROXY"} self.env = {k: v for k, v in os.environ.items() if k in allowed} def run(self, args, *, output=None, input_file=None, timeout=1800): with contextlib.ExitStack() as stack: log = stack.enter_context(self.log.open("ab")) target = stack.enter_context(Path(output).open("wb")) if output else subprocess.PIPE source = stack.enter_context(Path(input_file).open("rb")) if input_file else subprocess.DEVNULL # Do not log argv/environment: Compose commands may contain secrets. log.write(f"{now()} Start: {args[0]}\n".encode()) log.flush() proc = subprocess.Popen(args, stdout=target, stderr=log, stdin=source, env=self.env, start_new_session=True, close_fds=True) try: stdout, _ = proc.communicate(timeout=timeout) except BaseException: with contextlib.suppress(ProcessLookupError): os.killpg(proc.pid, signal.SIGTERM) try: proc.wait(timeout=10) except subprocess.TimeoutExpired: with contextlib.suppress(ProcessLookupError): os.killpg(proc.pid, signal.SIGKILL) proc.wait() raise if proc.returncode: if stdout: log.write(stdout) raise UpdateError(f"{args[0]} fehlgeschlagen (Exit {proc.returncode}); geschütztes commands.log prüfen.") return (stdout or b"").decode() class Updater: def __init__(self, root, *, http=None, runner_factory=Runner): self.root = Path(root).resolve() if not self.root.is_dir() or (self.root / ".git").exists(): raise UpdateError("Ein bestehendes Selfhost-Betriebsverzeichnis ohne Git-Checkout angeben.") self.home = self.root / ".updates" self.home.mkdir(mode=0o700, exist_ok=True) if self.home.is_symlink(): raise UpdateError(".updates darf kein symbolischer Link sein.") os.chmod(self.home, 0o700) self.http = http or Https(os.environ.get("FEDEO_RELEASE_TOKEN", "")) self.releases = Releases(self.http) self.runner_factory = runner_factory self.state = None @contextlib.contextmanager def lock(self): with (self.home / "lock").open("a") as lock: try: fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB) except BlockingIOError: raise UpdateError("Ein Update oder eine Prüfung läuft bereits.") from None # Children never inherit the lock; the kernel releases it after process death. yield def status(self): state = read_json(self.home / "status.json", {"status": "idle"}) try: with self.lock(): state = read_json(self.home / "status.json", {"status": "idle"}) if state.get("status") == "running": state.update(status="interrupted", error="Update-Prozess wurde unterbrochen.", updatedAt=now()) write_json(self.home / "status.json", state) write_json(Path(state["job"]) / "status.json", state) except UpdateError: pass return state def save(self, phase=None, **values): self.state.update(values, updatedAt=now()) if phase: self.state["phase"] = phase write_json(self.home / "status.json", self.state) write_json(self.job / "status.json", self.state) with (self.job / "events.jsonl").open("a") as stream: stream.write(json.dumps({"time": now(), "phase": self.state["phase"], "status": self.state["status"]}) + "\n") def compose(self, candidate, *args, **kwargs): prefix = ["docker", "compose", "--project-directory", str(self.root)] if getattr(self, "project", None): prefix += ["--project-name", self.project] prefix += ["--env-file", str(self.job / "target.env" if candidate else self.root / ".env"), "-f", str(self.job / "target.compose.yml" if candidate else self.root / "docker-compose.yml")] if self.override: prefix += ["-f", str(self.override)] return self.runner.run(prefix + list(args), **kwargs) def config(self, candidate): return json.loads(self.compose(candidate, "config", "--format", "json")) def inspect(self, name): return json.loads(self.runner.run(["docker", "inspect", name]))[0] def check_space(self, required, directory): if shutil.disk_usage(directory).free < required: raise UpdateError(f"Zu wenig freier Speicher auf {directory}: mindestens {required // GIB + 1} GiB erforderlich.") def prepare(self, version, *, allow_current=False): for name in (".env", "docker-compose.yml"): path = self.root / name if not path.is_file() or path.is_symlink(): raise UpdateError(f"{name} fehlt oder ist ein symbolischer Link.") if any((self.root / name).exists() for name in ("compose.yaml", "compose.yml")): raise UpdateError("Mehrdeutige Compose-Dateien im Betriebsverzeichnis.") overrides = [self.root / name for name in ("docker-compose.override.yml", "docker-compose.override.yaml") if (self.root / name).exists()] if len(overrides) > 1 or any(p.is_symlink() for p in overrides): raise UpdateError("Genau eine reguläre Compose-Override-Datei verwenden.") self.override = overrides[0] if overrides else None self.runner = self.runner_factory(self.job / "commands.log") compose_version = self.runner.run(["docker", "compose", "version", "--short"]).strip().lstrip("v") parts = re.match(r"(\d+)\.(\d+)", compose_version) if not parts or tuple(map(int, parts.groups())) < (2, 24): raise UpdateError("Docker Compose ab Version 2.24 wird benötigt.") info = json.loads(self.runner.run(["docker", "info", "--format", "{{json .}}"] )) if info.get("OSType") != "linux" or info.get("Architecture") not in ("x86_64", "amd64"): raise UpdateError("Diese Releases unterstützen ausschließlich Linux/amd64.") endpoint = self.runner.run(["docker", "context", "inspect", "--format", "{{.Endpoints.docker.Host}}"] ).strip() if not os.environ.get("DOCKER_HOST", endpoint).startswith("unix://"): raise UpdateError("Der Updater muss direkt auf dem lokalen Docker-Host laufen.") self.check_space(5 * GIB, info["DockerRootDir"]) self.current = self.config(False) self.project = self.current["name"] self.manifest = self.releases.download(version, self.job) self.target = self.manifest["version"] self.save("validating", version=self.target, revision=self.manifest["commit"]) old_version = self.current["services"]["backend"].get("environment", {}).get("FEDEO_VERSION", "dev") update_available = not VERSION.fullmatch(old_version) or tuple(map(int, old_version.split("."))) < tuple(map(int, self.target.split("."))) if not allow_current and not update_available: raise UpdateError("Zielversion muss neuer sein; Downgrades und erneute Installation sind gesperrt.") previous = read_json(self.home / "installed.json", {}) write_json(self.job / "original-installed.json", previous) previous_manifest = previous.get("manifest", {}) old_migrations = {m["path"]: m["sha256"] for m in previous_manifest.get("migrations", [])} new_migrations = {m["path"]: m["sha256"] for m in self.manifest.get("migrations", [])} if any(new_migrations.get(name) != checksum for name, checksum in old_migrations.items()): raise UpdateError("Bereits ausgelieferte Migrationen wurden verändert oder entfernt.") original = (self.root / "docker-compose.yml").read_bytes() self.managed = sha(original) == previous.get("composeSha256") compose = (self.job / "docker-compose.selfhost.yml").read_text() self.images = {} for name in ("backend", "frontend"): image = next(i for i in self.manifest["images"] if i["name"] == name) self.images[name] = image["repository"] + "@" + image["digest"] pattern = r"(?m)^(\s+image:\s*)" + re.escape(image["repository"]) + r":[^\r\n]+$" compose, count = re.subn(pattern, lambda m: m[1] + self.images[name], compose) if count != 1: raise UpdateError(f"Release-Compose enthält kein eindeutiges {name}-Image.") atomic(self.job / "target.compose.yml", compose) atomic(self.job / "target.env", set_version((self.root / ".env").read_bytes(), self.target)) diff = "".join(difflib.unified_diff(original.decode().splitlines(True), compose.splitlines(True), fromfile="installed/docker-compose.yml", tofile="release/docker-compose.yml")) atomic(self.job / "compose.diff", diff) self.candidate = self.config(True) self.validate_configs() ids = self.compose(False, "ps", "--all", "--quiet").split() self.containers = [self.inspect(container) for container in ids] write_json(self.job / "original-containers.json", self.containers) self.retired_containers = [c for c in self.containers if c["Config"]["Labels"].get("com.docker.compose.service") in self.retired_services] if any(c.get("HostConfig", {}).get("AutoRemove") for c in self.retired_containers): raise UpdateError("Matrix-Container mit automatischer Löschung können nicht zur Wiederherstellung erhalten werden.") for container in self.retired_containers: name = container["Config"]["Labels"]["com.docker.compose.service"] expected = {v["target"]: v["source"] for v in self.current["services"][name].get("volumes", [])} actual = {v["Destination"]: v["Source"] for v in container.get("Mounts", []) if v["Type"] == "bind"} if actual != expected: raise UpdateError(f"{name}: Tatsächliche Matrix-Datenpfade weichen von der Compose-Konfiguration ab.") self.save(retiredServices=self.retired_services, retiredContainerIds=[c["Id"] for c in self.retired_containers]) if any(c["State"]["Running"] and c["Config"]["Labels"].get("com.docker.compose.oneoff", "False").lower() == "true" for c in self.containers): raise UpdateError("Ein Einmalcontainer läuft noch; dessen Arbeit zuerst abschließen.") self.running = [c["Config"]["Labels"]["com.docker.compose.service"] for c in self.containers if c["State"]["Running"] and c["Config"]["Labels"].get("com.docker.compose.oneoff", "False").lower() != "true"] if not {"backend", "frontend", "db", "minio"}.issubset(self.running): raise UpdateError("Backend, Frontend, Datenbank und MinIO müssen vor dem Update laufen.") if not set(self.running).issubset(self.current["services"]): raise UpdateError("Nicht konfigurierte Container dieses Compose-Projekts laufen noch.") self.before = {str(p): sha(p.read_bytes()) for p in [self.root / ".env", self.root / "docker-compose.yml", *([self.override] if self.override else [])]} database_size = int(self.compose(False, "exec", "-T", "db", "sh", "-ec", 'exec psql -U "$POSTGRES_USER" -d "$POSTGRES_DB" -Atc "SELECT pg_database_size(current_database())"').strip()) minio_size = int(self.runner.run(["du", "-sb", str(self.root / "minio")]).split()[0]) self.backup_size = database_size * 2 + minio_size + GIB if self.matrix_backup_paths: self.backup_size += int(self.runner.run(["du", "-sb", str(self.root / "matrix")]).split()[0]) self.check_space(self.backup_size, self.home) plan = {"version": self.target, "revision": self.manifest["commit"], "previousVersion": old_version, "images": self.images, "composeDiff": str(self.job / "compose.diff"), "releaseNotes": str(self.job / "CHANGELOG.md"), "requiresComposeAdoption": not self.managed, "estimatedBackupBytes": self.backup_size, "services": list(self.candidate["services"]), "retiredServices": self.retired_services, "retainedDataPaths": self.matrix_backup_paths, "updateAvailable": update_available, # Kept in private job records; the service never exposes configuration fingerprints. "approvalFingerprint": sha(json.dumps({"manifest": self.manifest, "configuration": self.before, "targetCompose": sha((self.job / "target.compose.yml").read_bytes())}, sort_keys=True).encode())} write_json(self.job / "plan.json", plan) return plan def validate_configs(self): old, new = self.current["services"], self.candidate["services"] required = {"backend", "frontend", "db", "minio", "createbuckets", "traefik", "node-exporter"} differences = [] for label, services in (("Bestehende Konfiguration", old), ("Zielkonfiguration", new)): missing, additional = required - set(services), set(services) - required if services is old: additional -= LEGACY_MATRIX_SERVICES details = [] if missing: details.append("fehlend: " + ", ".join(sorted(missing))) if additional: details.append("zusätzlich: " + ", ".join(sorted(additional))) if details: differences.append(label + " (" + "; ".join(details) + ")") if differences: raise UpdateError("Dieser Updater unterstützt den Standard-Selfhost-Stack mit lokalen Daten und MinIO. " + ". ".join(differences) + ". Zusätzliche Dienste benötigen Unterstützung für Erhalt, Backup und Wiederherstellung.") self.retired_services = sorted(set(old) & LEGACY_MATRIX_SERVICES) self.matrix_backup_paths = [] self.validate_legacy_matrix() for name in required: if old[name].get("profiles") or new[name].get("profiles") or new[name].get("build"): raise UpdateError("Build-Kontexte und Compose-Profile werden nicht unterstützt.") old_mounts, new_mounts = old[name].get("volumes", []), new[name].get("volumes", []) # One-time bootstrap of the constrained updater API; never relax data-mount checks. control_mount = {"type": "bind", "source": "/run/fedeo-updater", "target": "/run/fedeo-updater", "read_only": True, "bind": {}} control_variants = (control_mount, {**control_mount, "bind": {"create_host_path": True}}) if name == "backend" and not any(v.get("target") == "/run/fedeo-updater" for v in old_mounts): new_mounts = [v for v in new_mounts if v not in control_variants] if old_mounts != new_mounts: raise UpdateError(f"Datenpfade/Mounts von {name} würden verändert; zuerst manuell migrieren.") if old[name].get("container_name") != new[name].get("container_name"): raise UpdateError(f"Container-Identität von {name} würde verändert.") if self.current.get("networks") != self.candidate.get("networks"): raise UpdateError("Docker-Netzwerke würden verändert.") for name, target, directory in (("db", "/var/lib/postgresql/data", "postgres"), ("minio", "/data", "minio")): volumes = new[name].get("volumes", []) if len(volumes) != 1 or volumes[0].get("type") != "bind" or volumes[0].get("target") != target or volumes[0].get("source") != str(self.root / directory): raise UpdateError(f"{name} benötigt das lokale Standard-Datenverzeichnis.") if (self.root / directory).is_symlink() or not (self.root / directory).is_dir(): raise UpdateError(f"Ungültiges Datenverzeichnis: {directory}.") if old[name].get("environment") != new[name].get("environment"): raise UpdateError(f"Zugangsdaten/Konfiguration von {name} würden verändert.") keys = ("POSTGRES_DB", "POSTGRES_USER", "POSTGRES_PASSWORD") if name == "db" else ("MINIO_ROOT_USER", "MINIO_ROOT_PASSWORD") if any(not new[name].get("environment", {}).get(key) for key in keys): raise UpdateError(f"Pflichteinstellungen von {name} fehlen.") if old["db"]["image"] != new["db"]["image"]: raise UpdateError("Ein Wechsel der PostgreSQL-Version benötigt eine separate Datenbankmigration.") for name in ("backend", "frontend"): healthcheck = new[name].get("healthcheck", {}) if new[name]["image"] != self.images[name] or healthcheck.get("disable") or not healthcheck.get("test") or healthcheck["test"] == ["NONE"]: raise UpdateError(f"{name}: Release-Digest oder Healthcheck wurde überschrieben.") for name in ("minio", "createbuckets"): if not re.search(r"@sha256:[a-f0-9]{64}$", new[name]["image"]): raise UpdateError("MinIO und mc müssen auf feste Digests gepinnt sein.") env = new["backend"].get("environment", {}) for key in ("DATABASE_URL", "S3_ENDPOINT", "S3_BUCKET", "S3_ACCESS_KEY", "S3_SECRET_KEY", "ENCRYPTION_KEY", "COOKIE_SECRET", "JWT_SECRET"): if not env.get(key) or env[key] != old["backend"].get("environment", {}).get(key): raise UpdateError(f"Backend-Einstellung {key} fehlt oder würde verändert.") if (env.get("FEDEO_FILE_BACKEND", "s3") != "s3" or old["backend"].get("environment", {}).get("FEDEO_FILE_BACKEND", "s3") != "s3" or urllib.parse.urlsplit(env["S3_ENDPOINT"]).hostname != "minio" or urllib.parse.urlsplit(env["DATABASE_URL"]).hostname != "db"): raise UpdateError("Externe Datenbanken/Dateispeicher benötigen einen eigenen Backup-Adapter.") database = urllib.parse.urlsplit(env["DATABASE_URL"]) db_env = new["db"]["environment"] if (urllib.parse.unquote(database.path.lstrip("/")) != db_env["POSTGRES_DB"] or urllib.parse.unquote(database.username or "") != db_env["POSTGRES_USER"] or urllib.parse.unquote(database.password or "") != db_env["POSTGRES_PASSWORD"] or database.port not in (None, 5432)): raise UpdateError("DATABASE_URL muss auf die gesicherte lokale PostgreSQL-Datenbank zeigen.") url = new["frontend"].get("environment", {}).get("NUXT_PUBLIC_API_BASE", "") parts = urllib.parse.urlsplit(url) if parts.scheme != "https" or not parts.hostname or parts.username or parts.password or parts.path.rstrip("/") != "/backend" or parts.query or parts.fragment: raise UpdateError("Die öffentliche API-Adresse muss https://DOMAIN/backend entsprechen.") self.public = urllib.parse.urlunsplit((parts.scheme, parts.netloc, "", "", "")) for name, service in new.items(): for key, value in service.get("environment", {}).items(): if value is None or (value == "" and key in ("DOMAIN", "POSTGRES_PASSWORD", "COOKIE_SECRET", "JWT_SECRET", "ENCRYPTION_KEY")): raise UpdateError(f"Fehlende Umgebungsvariable: {name}.{key}.") def validate_legacy_matrix(self): """Only retire the known, local legacy stack; never silently drop arbitrary services.""" if not self.retired_services: return matrix_root = self.root / "matrix" for name in self.retired_services: service = self.current["services"][name] if any(service.get(key) for key in ("profiles", "build", "volumes_from", "configs", "secrets")): raise UpdateError(f"{name}: Abweichende Matrix-Konfiguration benötigt eine manuelle Stilllegung.") volumes = service.get("volumes", []) for volume in volumes: source = Path(volume.get("source", "")) if (volume.get("type") != "bind" or not source.is_relative_to(matrix_root) or source.resolve() != source or not source.exists()): raise UpdateError(f"{name}: Matrix-Daten müssen ohne symbolische Links unter {matrix_root} liegen.") expected = {"matrix-db": ("postgres", "/var/lib/postgresql/data"), "matrix-synapse": ("synapse", "/data")}.get(name) if expected and (not (matrix_root / expected[0]).is_dir() or not any( v.get("source") == str(matrix_root / expected[0]) and v.get("target") == expected[1] for v in volumes)): raise UpdateError(f"{name}: Das lokale Standard-Matrix-Datenverzeichnis fehlt.") if matrix_root.exists() or matrix_root.is_symlink(): if matrix_root.is_symlink() or not matrix_root.is_dir(): raise UpdateError("Das Matrix-Datenverzeichnis muss ein reguläres lokales Verzeichnis sein.") # A cold PostgreSQL backup must include WAL and tablespaces, not external symlink targets. if any(path.is_symlink() for path in matrix_root.rglob("*")): raise UpdateError("Symbolische Links in Matrix-Daten benötigen eine manuelle Sicherung/Stilllegung.") self.matrix_backup_paths = ["matrix"] def verify_matrix_stopped(self, *, retired=False): for container_id in self.state.get("retiredContainerIds", []): container = self.inspect(container_id) state = container.get("State", {}) if state.get("Running") or state.get("Restarting") or state.get("Paused"): raise UpdateError("Ein alter Matrix-Container läuft noch; Stilllegung nicht abgeschlossen.") if retired and container.get("HostConfig", {}).get("RestartPolicy", {}).get("Name") != "no": raise UpdateError("Ein alter Matrix-Container könnte automatisch neu starten; Stilllegung nicht abgeschlossen.") def backup(self): self.verify_matrix_stopped() self.validate_legacy_matrix() directory = self.job / "backup" directory.mkdir(mode=0o700) for name in (".env", "docker-compose.yml", ".env.example", "scripts/selfhost-setup.sh"): if (self.root / name).is_file(): atomic(directory / "config" / name, (self.root / name).read_bytes()) if self.override: atomic(directory / "config" / self.override.name, self.override.read_bytes()) write_json(directory / "containers.json", self.containers) self.compose(False, "exec", "-T", "db", "sh", "-ec", 'exec pg_dump -U "$POSTGRES_USER" -d "$POSTGRES_DB" --format=custom', output=directory / "database.dump") self.compose(False, "exec", "-T", "db", "pg_restore", "--list", input_file=directory / "database.dump") self.compose(False, "exec", "-T", "db", "sh", "-ec", 'exec pg_dumpall -U "$POSTGRES_USER" --globals-only', output=directory / "globals.sql") paths = ["minio", *self.matrix_backup_paths] if (self.root / "traefik/letsencrypt").is_dir(): paths.append("traefik/letsencrypt") self.runner.run(["tar", "-cpf", str(directory / "files.tar"), "-C", str(self.root), *paths], timeout=7200) self.runner.run(["tar", "-tf", str(directory / "files.tar")], output=directory / "files.list", timeout=7200) checksums = {} for path in sorted(directory.rglob("*")): if path.is_file(): digest = hashlib.sha256() with path.open("rb") as stream: for chunk in iter(lambda: stream.read(1024 * 1024), b""): digest.update(chunk) os.fsync(stream.fileno()) checksums[str(path.relative_to(directory))] = digest.hexdigest() self.verify_matrix_stopped() write_json(directory / "complete.json", {"createdAt": now(), "database": "database.dump", "files": "files.tar", "retiredServices": self.retired_services, "retainedContainerIds": self.state["retiredContainerIds"], "matrixDataPaths": self.matrix_backup_paths, "sha256": checksums}) self.save("backed_up", backup=str(directory)) def verify(self, version, revision=None): # Traefik may need a moment to discover newly created healthy containers. for attempt in range(6): try: for path in ("/backend/health", "/api/health"): value = self.http.json(self.public + path) if value.get("status") != "ok" or value.get("product") != "FEDEO" or value.get("version") != version or (revision and value.get("revision") != revision): raise UpdateError("Öffentliche Funktionsprüfung meldet eine falsche Version oder einen Fehler.") return except UpdateError: if attempt == 5: raise time.sleep(2) def apply(self, accept_compose): if not self.managed and not accept_compose: raise UpdateError("Compose-Datei ist noch nicht verwaltet oder wurde bearbeitet. compose.diff prüfen, Anpassungen in die Override-Datei übertragen und --accept-compose-replacement verwenden.") self.save("pulling") self.compose(True, "pull", timeout=3600) self.check_space(self.backup_size, self.home) if any(sha(Path(path).read_bytes()) != value for path, value in self.before.items()): raise UpdateError("Konfiguration wurde während der Vorbereitung verändert; erneut prüfen.") self.save("maintenance", recoveryRequired=True, maintenance=True, runningServices=self.running) # Stop every writer and public entry point. The database stays up for pg_dump. self.compose(False, "stop", "--timeout", "60", *[s for s in self.running if s != "db"]) self.backup() if any(sha(Path(path).read_bytes()) != value for path, value in self.before.items()): raise UpdateError("Konfiguration wurde während des Backups verändert; erneut prüfen.") # From here recovery must be explicit: migrations or infrastructure can alter persisted data. self.save("applying", mutationStarted=True) if self.retired_containers: self.save("retiring_matrix") self.verify_matrix_stopped() # Keep containers, writable layers and volumes for recovery; prevent restart on host reboot. for container in self.retired_containers: self.runner.run(["docker", "update", "--restart=no", container["Id"]]) self.verify_matrix_stopped(retired=True) atomic(self.root / "docker-compose.yml", (self.job / "target.compose.yml").read_bytes()) atomic(self.root / ".env", (self.job / "target.env").read_bytes()) atomic(self.root / ".env.example", (self.job / ".env.example").read_bytes()) atomic(self.root / "scripts/selfhost-setup.sh", (self.job / "selfhost-setup.sh").read_bytes()) os.chmod(self.root / "scripts/selfhost-setup.sh", 0o700) self.compose(True, "up", "-d", "--no-deps", "--wait", "--wait-timeout", "180", "db", "minio", "node-exporter") self.compose(True, "run", "--rm", "--no-deps", "createbuckets") self.save("migrating", migrationContainer="fedeo-update-" + self.job.name) self.compose(True, "run", "--rm", "--no-deps", "--name", self.state["migrationContainer"], "-e", "FEDEO_RUN_MIGRATIONS=false", "--entrypoint", "npm", "backend", "run", "migrate") self.save("starting") self.compose(True, "up", "-d", "--wait", "--wait-timeout", "600", timeout=900) self.save("verifying") self.verify_matrix_stopped(retired=True) for name, image in self.images.items(): container_id = self.compose(True, "ps", "--quiet", name).strip() container = self.inspect(container_id) image_id = self.runner.run(["docker", "image", "inspect", image, "--format", "{{.Id}}"] ).strip() if container["Image"] != image_id or container.get("State", {}).get("Health", {}).get("Status") != "healthy": raise UpdateError(f"{name} verwendet nicht das erwartete gesunde Image.") self.verify(self.target, self.manifest["commit"]) write_json(self.home / "installed.json", {"manifest": self.manifest, "composeSha256": sha((self.root / "docker-compose.yml").read_bytes()), "installedAt": now()}) self.save("complete", status="succeeded", recoveryRequired=False, maintenance=False) def execute(self, command, version=None, accept_compose=False, *, expected_fingerprint=None, request_id=None): with self.lock(): previous = read_json(self.home / "status.json", {}) if previous.get("recoveryRequired"): raise UpdateError("Ein vorheriges Update benötigt Wiederherstellung. status und docs/selfhost-updates.md prüfen; danach resolve ausführen.") self.job = self.home / (datetime.datetime.now().strftime("%Y%m%d-%H%M%S-") + uuid.uuid4().hex[:8]) self.job.mkdir(mode=0o700) self.state = {"status": "running", "phase": "preparing", "job": str(self.job), "command": command, "startedAt": now(), "recoveryRequired": False, "mutationStarted": False} if request_id: self.state["requestId"] = request_id self.save() try: plan = self.prepare(version, allow_current=command == "check") if expected_fingerprint is not None and plan["approvalFingerprint"] != expected_fingerprint: raise PlanChangedError("Release oder Konfiguration haben sich seit der Prüfung geändert; erneut prüfen.") if command == "check": self.save("checked", status="checked") else: self.apply(accept_compose) return {**self.state, "plan": plan} except BaseException as error: if self.state.get("maintenance") and not self.state.get("mutationStarted"): try: self.compose(False, "start", *self.running) self.save(recoveryRequired=False, maintenance=False) except BaseException: pass elif self.state.get("mutationStarted"): try: self.compose(True, "stop", "--timeout", "60", "backend", "frontend", "traefik") except BaseException: pass self.save(status="interrupted" if isinstance(error, (KeyboardInterrupt, SystemExit)) else "failed", error=str(error) if isinstance(error, UpdateError) else type(error).__name__) raise def resolve(self, version): if not version or not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9.+-]{0,100}", version): raise UpdateError("Die manuell wiederhergestellte Version mit --version angeben.") with self.lock(): state = read_json(self.home / "status.json", {}) if not state.get("recoveryRequired"): raise UpdateError("Kein offener Wiederherstellungsfall.") self.job = Path(state["job"]) self.state = state self.runner = self.runner_factory(self.job / "commands.log") self.override = next((self.root / name for name in ("docker-compose.override.yml", "docker-compose.override.yaml") if (self.root / name).exists()), None) current = self.config(False) self.public = current["services"]["frontend"]["environment"]["NUXT_PUBLIC_API_BASE"].removesuffix("/backend") if state.get("migrationContainer"): remaining = self.runner.run(["docker", "ps", "--all", "--quiet", "--filter", "name=^/" + state["migrationContainer"] + "$"]).strip() if remaining: raise UpdateError("Den verbliebenen Migrationscontainer zuerst prüfen und manuell entfernen.") # Older releases have /version and /health, but no frontend health endpoint. version_info = self.http.json(self.public + "/backend/version") if version_info.get("product") != "FEDEO" or version_info.get("version") != version: raise UpdateError("Die wiederhergestellte Backend-Version stimmt nicht.") if self.http.json(self.public + "/backend/health").get("status") != "ok": raise UpdateError("Das wiederhergestellte Backend ist nicht erreichbar.") self.http.get(self.public + "/") original_images = {c["Config"]["Labels"]["com.docker.compose.service"]: c["Image"] for c in read_json(self.job / "original-containers.json", [])} manifest = read_json(self.job / "release-manifest.json") for name in ("backend", "frontend", "db", "minio"): container_id = self.compose(False, "ps", "--quiet", name).strip() container = self.inspect(container_id) if container_id else {} container_state = container.get("State", {}) if not container_state.get("Running") or container_state.get("Health", {}).get("Status") not in (None, "healthy"): raise UpdateError(f"{name} läuft noch nicht.") if name in ("backend", "frontend"): if version == manifest["version"]: image = next(i for i in manifest["images"] if i["name"] == name) expected = self.runner.run(["docker", "image", "inspect", image["repository"] + "@" + image["digest"], "--format", "{{.Id}}"] ).strip() else: expected = original_images.get(name) if not expected or container.get("Image") != expected: raise UpdateError(f"{name} entspricht weder dem gewählten Zielrelease noch dem gesicherten Originalstand.") if version == manifest["version"]: self.verify_matrix_stopped(retired=True) self.verify(version, manifest["commit"]) write_json(self.home / "installed.json", {"manifest": manifest, "composeSha256": sha((self.job / "target.compose.yml").read_bytes()), "installedAt": now()}) else: for original in read_json(self.job / "original-containers.json", []): if original.get("Id") not in state.get("retiredContainerIds", []): continue name = original["Config"]["Labels"]["com.docker.compose.service"] container_id = self.compose(False, "ps", "--all", "--quiet", name).strip() restored = self.inspect(container_id) if container_id else {} if (restored.get("Image") != original["Image"] or restored.get("State", {}).get("Running") != original["State"]["Running"] or (original["State"]["Running"] and restored.get("State", {}).get("Health", {}).get("Status") not in (None, "healthy")) or restored.get("HostConfig", {}).get("RestartPolicy") != original["HostConfig"]["RestartPolicy"]): raise UpdateError("Die ursprünglichen Matrix-Container und Neustartrichtlinien müssen zuerst wiederhergestellt werden.") previous = read_json(self.job / "original-installed.json", {}) if previous: write_json(self.home / "installed.json", previous) else: (self.home / "installed.json").unlink(missing_ok=True) self.save(status="recovered", phase="resolved", recoveryRequired=False, maintenance=False, recoveredVersion=version) return self.state def main(argv=None): if sys.version_info < (3, 10): print("Python ab Version 3.10 wird benötigt.", file=sys.stderr) return 1 parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("command", choices=("check", "apply", "status", "resolve")) parser.add_argument("--directory", default=os.environ.get("FEDEO_DIR", "/opt/fedeo")) parser.add_argument("--version", help="Zielversion, z. B. 3.1.0; check wählt sonst den neuesten stabilen Release") parser.add_argument("--accept-compose-replacement", action="store_true", help="Zuvor geprüften Austausch einer nicht verwalteten Compose-Datei akzeptieren") args = parser.parse_args(argv) if args.command == "apply" and not args.version: parser.error("apply benötigt eine explizite --version; zuerst check ausführen") os.umask(0o077) signal.signal(signal.SIGTERM, lambda *_: sys.exit(143)) try: updater = Updater(args.directory) if args.command == "status": result = updater.status() elif args.command == "resolve": result = updater.resolve(args.version) else: result = updater.execute(args.command, args.version, args.accept_compose_replacement) print(json.dumps(result, ensure_ascii=False, indent=2)) return 0 except (UpdateError, OSError, ValueError, KeyError, subprocess.TimeoutExpired) as error: print(str(error) if isinstance(error, UpdateError) else f"Update abgebrochen: {type(error).__name__}; geschütztes Statusprotokoll prüfen.", file=sys.stderr) return 1 if __name__ == "__main__": sys.exit(main())