753 lines
46 KiB
Python
753 lines
46 KiB
Python
#!/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")
|
|
# Exact digests are immutable: reuse local copies even when their registry is offline.
|
|
# Mutable infrastructure tags must still be refreshed before entering maintenance.
|
|
for policy in ("missing", "always"):
|
|
services = [name for name, service in self.candidate["services"].items()
|
|
if bool(re.search(r"@sha256:[a-f0-9]{64}$", service["image"])) == (policy == "missing")]
|
|
if services:
|
|
self.compose(True, "pull", "--policy", policy, *services, 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", "--pull", "never", "-d", "--no-deps", "--wait", "--wait-timeout", "180", "db", "minio", "node-exporter")
|
|
self.compose(True, "run", "--pull", "never", "--rm", "--no-deps", "createbuckets")
|
|
self.save("migrating", migrationContainer="fedeo-update-" + self.job.name)
|
|
self.compose(True, "run", "--pull", "never", "--rm", "--no-deps", "--name", self.state["migrationContainer"],
|
|
"-e", "FEDEO_RUN_MIGRATIONS=false", "--entrypoint", "npm", "backend", "run", "migrate")
|
|
self.save("starting")
|
|
self.compose(True, "up", "--pull", "never", "-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())
|