1104 lines
70 KiB
Python
1104 lines
70 KiB
Python
#!/usr/bin/env python3
|
|
"""FEDEO host-side update engine. Python 3.10+, Docker Compose 2.24+; no pip packages.
|
|
|
|
The CLI and the optional local host service share Updater and the exclusive lock.
|
|
All commands operate on an existing local selfhost installation, never a git checkout.
|
|
"""
|
|
import argparse
|
|
import contextlib
|
|
import copy
|
|
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"}
|
|
CORE_SERVICES = {"backend", "frontend", "db", "traefik", "node-exporter"}
|
|
IMAGE_ID = re.compile(r"sha256:[a-f0-9]{64}\Z")
|
|
|
|
|
|
def compose_strings(value, *, escape=False):
|
|
"""Round-trip resolved Compose output, including literal dollars in secrets."""
|
|
if isinstance(value, str):
|
|
return value.replace("$", "$$") if escape else value.replace("$$", "$")
|
|
if isinstance(value, list):
|
|
return [compose_strings(v, escape=escape) for v in value]
|
|
if isinstance(value, dict):
|
|
return {k: compose_strings(v, escape=escape) for k, v in value.items()}
|
|
return value
|
|
|
|
|
|
def file_sha(path):
|
|
digest = hashlib.sha256()
|
|
with Path(path).open("rb") as stream:
|
|
for chunk in iter(lambda: stream.read(1024 * 1024), b""):
|
|
digest.update(chunk)
|
|
return digest.hexdigest()
|
|
|
|
|
|
# Runs with the actual application's SDK. Never prints keys, URLs or credentials.
|
|
S3_READ_CHECK = r"""
|
|
const {S3Client, HeadBucketCommand, ListObjectsV2Command, GetObjectCommand} = require('@aws-sdk/client-s3');
|
|
const client = new S3Client({endpoint:process.env.S3_ENDPOINT, region:process.env.S3_REGION,
|
|
forcePathStyle:true, maxAttempts:8, credentials:{accessKeyId:process.env.S3_ACCESS_KEY,secretAccessKey:process.env.S3_SECRET_KEY}});
|
|
const timer = setTimeout(()=>process.exit(1), 180000);
|
|
(async()=>{
|
|
const Bucket=process.env.S3_BUCKET;
|
|
await client.send(new HeadBucketCommand({Bucket}));
|
|
const listed=await client.send(new ListObjectsV2Command({Bucket,MaxKeys:1}));
|
|
const first=listed.Contents?.[0];
|
|
if(first) {
|
|
const object=await client.send(new GetObjectCommand({Bucket,Key:first.Key,...(first.Size ? {Range:'bytes=0-0'} : {})}));
|
|
await object.Body.transformToByteArray();
|
|
}
|
|
})().catch(()=>{console.error('S3-Zugriffsprüfung fehlgeschlagen');process.exitCode=1})
|
|
.finally(()=>{clearTimeout(timer);client.destroy()});
|
|
"""
|
|
|
|
S3_UPGRADE_CHECK = r"""
|
|
const s3 = require('@aws-sdk/client-s3');
|
|
const {getSignedUrl} = require('@aws-sdk/s3-request-presigner');
|
|
const {Readable} = require('node:stream');
|
|
const assert = require('node:assert/strict');
|
|
const client = new s3.S3Client({endpoint:process.env.S3_ENDPOINT, region:process.env.S3_REGION,
|
|
forcePathStyle:true, maxAttempts:8,
|
|
credentials:{accessKeyId:process.env.S3_ACCESS_KEY,secretAccessKey:process.env.S3_SECRET_KEY}});
|
|
const Bucket=process.env.S3_BUCKET, Key=process.env.FEDEO_UPDATE_PROBE_KEY;
|
|
const timer=setTimeout(()=>process.exit(1), 180000);
|
|
(async()=>{
|
|
assert.match(Key, /^\.fedeo-update-check\/[a-f0-9]{32}\.pdf$/);
|
|
try {
|
|
await client.send(new s3.HeadBucketCommand({Bucket}));
|
|
const body=Buffer.from('FEDEO S3 update check');
|
|
await client.send(new s3.PutObjectCommand({Bucket,Key,Body:Readable.from([body]),ContentLength:body.length,
|
|
ContentType:'application/octet-stream',Metadata:{fedeo:'update-check'}}));
|
|
const head=await client.send(new s3.HeadObjectCommand({Bucket,Key}));
|
|
assert.equal(head.ContentType,'application/octet-stream');
|
|
assert.equal(head.Metadata.fedeo,'update-check');
|
|
assert.ok(!head.ContentEncoding);
|
|
const result=await client.send(new s3.GetObjectCommand({Bucket,Key}));
|
|
assert.deepEqual(Buffer.from(await result.Body.transformToByteArray()),body);
|
|
const signed=await getSignedUrl(client,new s3.GetObjectCommand({Bucket,Key}),{expiresIn:86400});
|
|
const response=await fetch(signed);
|
|
assert.equal(response.status,200);
|
|
assert.deepEqual(Buffer.from(await response.arrayBuffer()),body);
|
|
const anonymous=new URL(signed); anonymous.search='';
|
|
const denied=await fetch(anonymous);
|
|
assert.equal(denied.status,403); await denied.body?.cancel();
|
|
} finally { await client.send(new s3.DeleteObjectCommand({Bucket,Key})); }
|
|
try { await client.send(new s3.HeadObjectCommand({Bucket,Key})); throw new Error('probe remains'); }
|
|
catch(error) { if(error.$metadata?.httpStatusCode!==404) throw error; }
|
|
})().catch(()=>{console.error('S3-Vertrag des Zielbackends fehlgeschlagen');process.exitCode=1})
|
|
.finally(()=>{clearTimeout(timer);client.destroy()});
|
|
"""
|
|
|
|
S3_CLEANUP_CHECK = r"""
|
|
const s3=require('@aws-sdk/client-s3'), assert=require('node:assert/strict');
|
|
const client=new s3.S3Client({endpoint:process.env.S3_ENDPOINT,region:process.env.S3_REGION,forcePathStyle:true,
|
|
credentials:{accessKeyId:process.env.S3_ACCESS_KEY,secretAccessKey:process.env.S3_SECRET_KEY}});
|
|
const Bucket=process.env.S3_BUCKET, Key=process.env.FEDEO_UPDATE_PROBE_KEY;
|
|
const timer=setTimeout(()=>process.exit(1),60000);
|
|
(async()=>{
|
|
assert.match(Key,/^\.fedeo-update-check\/[a-f0-9]{32}\.pdf$/);
|
|
let head;
|
|
try { head=await client.send(new s3.HeadObjectCommand({Bucket,Key})); }
|
|
catch(error) { if(error.$metadata?.httpStatusCode===404) return; throw error; }
|
|
assert.equal(head.Metadata?.fedeo,'update-check');
|
|
await client.send(new s3.DeleteObjectCommand({Bucket,Key}));
|
|
try { await client.send(new s3.HeadObjectCommand({Bucket,Key})); throw new Error('probe remains'); }
|
|
catch(error) { if(error.$metadata?.httpStatusCode!==404) throw error; }
|
|
})().catch(()=>{console.error('Protokollierten S3-Prüfschlüssel manuell prüfen');process.exitCode=1})
|
|
.finally(()=>{clearTimeout(timer);client.destroy()});
|
|
"""
|
|
|
|
|
|
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 compose_strings(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 storage_record(self):
|
|
path = self.home / "storage-provider.json"
|
|
if path.is_symlink():
|
|
raise UpdateError("Speicherbeschreibung darf kein symbolischer Link sein.")
|
|
record = read_json(path, {})
|
|
if record and (record.get("schemaVersion") != 1 or record.get("provider") != "seaweedfs"
|
|
or not IMAGE_ID.fullmatch(str(record.get("image", "")))):
|
|
raise UpdateError("Unbekannte Speicherbeschreibung; Speichermigration auf dem Host prüfen.")
|
|
return record
|
|
|
|
def validate_storage(self, config, record, *, running=True):
|
|
"""The adopted storage service is immutable across application releases."""
|
|
store = config.get("services", {}).get("objectstore", {})
|
|
if store != record.get("service") or store.get("image") != record.get("image"):
|
|
raise UpdateError("SeaweedFS-Konfiguration oder Image verändert; separate Speichermigration erforderlich.")
|
|
for directory in ("storage", "storage/seaweedfs", "storage/config"):
|
|
path = self.root / directory
|
|
if not path.is_dir() or path.resolve() != path:
|
|
raise UpdateError("SeaweedFS benötigt reguläre lokale storage/-Verzeichnisse.")
|
|
if any(p.is_symlink() for p in (self.root / "storage").rglob("*")):
|
|
raise UpdateError("Symbolische Links in storage/ können nicht vollständig gesichert werden.")
|
|
expected = {"/data": str(self.root / "storage/seaweedfs"), "/etc/fedeo": str(self.root / "storage/config")}
|
|
volumes = store.get("volumes", [])
|
|
if (len(volumes) != 2 or any(v.get("type") != "bind" for v in volumes)
|
|
or {v.get("target"): v.get("source") for v in volumes} != expected
|
|
or not next(v for v in volumes if v["target"] == "/etc/fedeo").get("read_only")):
|
|
raise UpdateError("SeaweedFS-Datenmounts entsprechen nicht der übernommenen lokalen Ablage.")
|
|
for name in ("s3.json", "start-seaweedfs.sh"):
|
|
path = self.root / "storage/config" / name
|
|
if not path.is_file() or file_sha(path) != record.get("configSha256", {}).get(name):
|
|
raise UpdateError("Geschützte SeaweedFS-Konfiguration verändert oder unvollständig.")
|
|
if store.get("environment", {}).get("FEDEO_S3_CONFIG_SHA256") != record["configSha256"]["s3.json"]:
|
|
raise UpdateError("SeaweedFS-Konfigurationsprüfsumme stimmt nicht.")
|
|
network = config.get("networks", {}).get("internal", {}).get("name")
|
|
if (network != record.get("network") or store.get("ports") or set(store.get("networks", {})) != {"internal"}
|
|
or "minio" not in store["networks"]["internal"].get("aliases", [])):
|
|
raise UpdateError("SeaweedFS muss ausschließlich am bisherigen internen S3-Anschluss liegen.")
|
|
old = self.inspect(record["retainedMinio"]["id"])
|
|
if (old.get("Image") != record["retainedMinio"]["image"]
|
|
or any(old.get("State", {}).get(k) for k in ("Running", "Restarting", "Paused"))
|
|
or old.get("HostConfig", {}).get("RestartPolicy", {}).get("Name") != "no"
|
|
or network in old.get("NetworkSettings", {}).get("Networks", {})):
|
|
raise UpdateError("Der erhaltene MinIO-Container ist nicht unverändert stillgelegt.")
|
|
if running:
|
|
cid = self.compose(False, "ps", "--quiet", "objectstore").strip()
|
|
actual = self.inspect(cid) if cid and "\n" not in cid else {}
|
|
mounts = actual.get("Mounts", [])
|
|
if (actual.get("Image") != record["image"] or not actual.get("State", {}).get("Running")
|
|
or actual.get("State", {}).get("Health", {}).get("Status") != "healthy"
|
|
or actual.get("HostConfig", {}).get("PortBindings")
|
|
or set(actual.get("NetworkSettings", {}).get("Networks", {})) != {network}
|
|
or "minio" not in actual["NetworkSettings"]["Networks"][network].get("Aliases", [])
|
|
or len(mounts) != 2 or any(v.get("Type") != "bind" for v in mounts)
|
|
or {v.get("Destination"): v.get("Source") for v in mounts} != expected
|
|
or any(v.get("RW") for v in mounts if v.get("Destination") == "/etc/fedeo")
|
|
or ("FEDEO_S3_CONFIG_SHA256=" + record["configSha256"]["s3.json"]) not in actual.get("Config", {}).get("Env", [])):
|
|
raise UpdateError("Laufender SeaweedFS-Container weicht vom geprüften Speicher ab.")
|
|
|
|
def adopt_storage(self):
|
|
"""Explicit host-only handover; cannot resolve an interrupted migration/update."""
|
|
with self.lock():
|
|
guard = read_json(self.home / "status.json", {})
|
|
if guard.get("phase") != "storage_migration" or not guard.get("recoveryRequired"):
|
|
if self.storage_record() and not guard.get("recoveryRequired"):
|
|
return {"status": "already_adopted", "storageProvider": "seaweedfs"}
|
|
raise UpdateError("Keine abgeschlossene Pilotmigration zur Übernahme vorhanden.")
|
|
migration = self.home / "storage-migration"
|
|
if migration.is_symlink() or guard.get("job") != str(migration):
|
|
raise UpdateError("Speichermigrationsauftrag stimmt nicht.")
|
|
state = read_json(migration / "state.json", {})
|
|
if (state.get("schemaVersion") != 1 or state.get("phase") != "complete"
|
|
or state.get("writesReleased") is not True or state.get("provider") != "seaweedfs"
|
|
or not state.get("restoreVerifiedAt") or not IMAGE_ID.fullmatch(str(state.get("weedImage", "")))):
|
|
raise UpdateError("Speichermigration und geprüften Restore zuerst vollständig abschließen; kein resolve verwenden.")
|
|
if any((self.root / n).exists() for n in ("docker-compose.override.yml", "docker-compose.override.yaml", "compose.yml", "compose.yaml")):
|
|
raise UpdateError("Pilotübernahme unterstützt keine zusätzlichen Compose-Dateien.")
|
|
before = {n: file_sha(self.root / n) for n in (".env", "docker-compose.yml")}
|
|
if any((self.root / n).is_symlink() for n in before):
|
|
raise UpdateError("Konfigurationsdateien dürfen keine symbolischen Links sein.")
|
|
if (before["docker-compose.yml"] != file_sha(migration / "target.json")
|
|
or before[".env"] != state.get("fingerprint", {}).get(".env")):
|
|
raise UpdateError("Installation nach der Speichermigration verändert; Übernahme manuell prüfen.")
|
|
verified = read_json(migration / "copy/verified.json", {})
|
|
if (not verified.get("verifiedAt") or verified.get("count") != state.get("count")
|
|
or verified.get("bytes") != state.get("bytes")
|
|
or len(verified.get("objects", [])) != state.get("count")):
|
|
raise UpdateError("Vollständiger Kopiernachweis der Speichermigration fehlt.")
|
|
for name in ("restore-contract", "acceptance", "backend-acceptance"):
|
|
if not {"PUT", "HEAD", "GET", "DELETE", "SigV4", "presigned-24h", "anonymous-denied", "wrong-key-denied"} <= set(
|
|
read_json(migration / name / "result.json", {}).get("operations", [])):
|
|
raise UpdateError("S3-Abnahme der Speichermigration fehlt: " + name)
|
|
if (not (migration / "backup/complete.json").is_file()
|
|
or file_sha(migration / "backup/seaweedfs.tar") != state.get("seaweedBackupSha256")):
|
|
raise UpdateError("Geprüfte Speichersicherung fehlt oder wurde verändert.")
|
|
self.job, self.override = migration, None
|
|
self.runner = self.runner_factory(migration / "adoption.log")
|
|
endpoint = self.runner.run(["docker", "context", "inspect", "--format", "{{.Endpoints.docker.Host}}"] ).strip()
|
|
if not os.environ.get("DOCKER_HOST", endpoint).startswith("unix://"):
|
|
raise UpdateError("Speicherübernahme muss direkt auf dem lokalen Docker-Host laufen.")
|
|
config = self.config(False)
|
|
self.project = config["name"]
|
|
required = CORE_SERVICES | {"objectstore"}
|
|
if not required <= set(config["services"]) or set(config["services"]) - required - LEGACY_MATRIX_SERVICES:
|
|
raise UpdateError("Speicherübernahme benötigt den Standard-Stack; nur bekannte alte Matrix-Dienste sind zusätzlich erlaubt.")
|
|
original = read_json(migration / "original.json")
|
|
record = {"schemaVersion": 1, "provider": "seaweedfs", "adoptedAt": now(),
|
|
"image": state["weedImage"], "network": state["network"],
|
|
"service": config["services"]["objectstore"],
|
|
"databaseSourceImage": original["services"]["db"]["image"],
|
|
"databaseImage": state["containers"]["db"]["image"],
|
|
"retainedMinio": {k: state["containers"]["minio"][k] for k in ("id", "image")},
|
|
"configSha256": {n: file_sha(self.root / "storage/config" / n) for n in ("s3.json", "start-seaweedfs.sh")}}
|
|
self.validate_storage(config, record)
|
|
for name in CORE_SERVICES:
|
|
cid = self.compose(False, "ps", "--quiet", name).strip()
|
|
actual = self.inspect(cid) if cid and "\n" not in cid else {}
|
|
if (actual.get("Image") != state["containers"][name]["image"]
|
|
or not actual.get("State", {}).get("Running")
|
|
or actual.get("State", {}).get("Health", {}).get("Status") not in (None, "healthy")):
|
|
raise UpdateError("Anwendung nach Speichermigration nicht gesund/unverändert: " + name)
|
|
self.compose(False, "exec", "-T", "backend", "node", "-e", S3_READ_CHECK, timeout=210)
|
|
if any(file_sha(self.root / n) != digest for n, digest in before.items()):
|
|
raise UpdateError("Konfiguration während Speicherübernahme verändert.")
|
|
# The old guard remains until both records are durable; retry after a crash is safe.
|
|
write_json(self.home / "storage-provider.json", record)
|
|
installed = read_json(self.home / "installed.json", {})
|
|
installed.update(composeSha256=before["docker-compose.yml"], storageProvider="seaweedfs")
|
|
write_json(self.home / "installed.json", installed)
|
|
write_json(migration / "updater-adoption.json", {"adoptedAt": record["adoptedAt"], "provider": "seaweedfs"})
|
|
write_json(self.home / "status.json", {"status": "recovered", "phase": "storage_adopted",
|
|
"updatedAt": now(), "recoveryRequired": False, "maintenance": False, "mutationStarted": False})
|
|
return {"status": "adopted", "storageProvider": "seaweedfs", "version": state["version"]}
|
|
|
|
def adapt_storage_candidate(self):
|
|
self.storage = self.storage_record()
|
|
self.storage_service = "objectstore" if self.storage else "minio"
|
|
self.storage_paths = ["storage"] if self.storage else ["minio"]
|
|
if not self.storage:
|
|
if "objectstore" in self.current["services"]:
|
|
raise UpdateError("SeaweedFS zuerst mit adopt-storage aus der abgeschlossenen Pilotmigration übernehmen.")
|
|
return
|
|
self.validate_storage(self.current, self.storage)
|
|
services = self.candidate["services"]
|
|
if set(services) != CORE_SERVICES | {"minio", "createbuckets"}:
|
|
raise UpdateError("Release verwendet einen unbekannten Speicheraufbau; Updater-Unterstützung erforderlich.")
|
|
if services["db"]["image"] != self.storage["databaseSourceImage"]:
|
|
raise UpdateError("Ein Wechsel der PostgreSQL-Version benötigt eine separate Datenbankmigration.")
|
|
services["db"]["image"] = self.storage["databaseImage"]
|
|
del services["minio"], services["createbuckets"]
|
|
services["objectstore"] = copy.deepcopy(self.storage["service"])
|
|
dependencies = services["backend"].setdefault("depends_on", {})
|
|
dependencies.pop("minio", None)
|
|
dependencies.pop("createbuckets", None)
|
|
dependencies["objectstore"] = {"condition": "service_healthy", "required": True}
|
|
services["backend"].setdefault("environment", {}).update(
|
|
AWS_REQUEST_CHECKSUM_CALCULATION="WHEN_REQUIRED", FEDEO_RUN_MIGRATIONS="false", FEDEO_STORAGE_PROVIDER="seaweedfs")
|
|
write_json(self.job / "target.compose.yml", compose_strings(self.candidate, escape=True))
|
|
self.candidate = self.config(True)
|
|
self.validate_storage(self.candidate, self.storage)
|
|
|
|
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))
|
|
self.candidate = self.config(True)
|
|
self.adapt_storage_candidate()
|
|
diff = "".join(difflib.unified_diff(original.decode().splitlines(True),
|
|
(self.job / "target.compose.yml").read_text().splitlines(True),
|
|
fromfile="installed/docker-compose.yml", tofile="release/docker-compose.yml"))
|
|
atomic(self.job / "compose.diff", diff)
|
|
self.validate_configs()
|
|
# Compose ps can omit services removed from the current file. Retained
|
|
# Matrix/MinIO containers still belong to the project and its backup plan.
|
|
ids = self.runner.run(["docker", "ps", "--all", "--quiet", "--filter",
|
|
"label=com.docker.compose.project=" + self.project]).split()
|
|
self.containers = [self.inspect(container) for container in ids]
|
|
write_json(self.job / "original-containers.json", self.containers)
|
|
retained = {}
|
|
seen = set()
|
|
for container in self.containers:
|
|
labels = container.get("Config", {}).get("Labels", {})
|
|
name = labels.get("com.docker.compose.service")
|
|
if labels.get("com.docker.compose.oneoff", "False").lower() == "true":
|
|
if container.get("State", {}).get("Running"):
|
|
raise UpdateError("Ein Einmalcontainer läuft noch; dessen Arbeit zuerst abschließen.")
|
|
continue
|
|
if not name or name in seen:
|
|
raise UpdateError("Unbekannte oder doppelte Projektcontainer vorhanden.")
|
|
seen.add(name)
|
|
if name not in self.current["services"]:
|
|
if self.storage and name in ("minio", "createbuckets"):
|
|
continue # MinIO's exact identity and retirement are checked by validate_storage.
|
|
if name not in LEGACY_MATRIX_SERVICES:
|
|
raise UpdateError("Unbekannter Projektcontainer: " + name)
|
|
state, host = container.get("State", {}), container.get("HostConfig", {})
|
|
if (any(state.get(k) for k in ("Running", "Restarting", "Paused"))
|
|
or host.get("RestartPolicy", {}).get("Name") != "no" or host.get("AutoRemove")):
|
|
raise UpdateError("Alter Matrix-Container ist nicht dauerhaft stillgelegt: " + name)
|
|
volumes = []
|
|
for mount in container.get("Mounts", []):
|
|
if mount["Type"] == "bind":
|
|
volumes.append({"type": "bind", "source": mount["Source"], "target": mount["Destination"]})
|
|
elif mount["Type"] != "volume" or (name, mount["Destination"]) not in {
|
|
("matrix-redis", "/data"), ("matrix-turn", "/var/lib/coturn")}:
|
|
raise UpdateError("Abweichender Datenmount eines alten Matrix-Containers: " + name)
|
|
retained[name] = {"volumes": volumes}
|
|
self.current["services"].update(retained)
|
|
self.retired_services = sorted(set(self.retired_services) | set(retained))
|
|
self.validate_legacy_matrix()
|
|
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", self.storage_service}.issubset(self.running):
|
|
raise UpdateError("Backend, Frontend, Datenbank und Dateispeicher 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 [])]}
|
|
if self.storage:
|
|
for path in [self.home / "storage-provider.json", *[self.root / "storage/config" / n for n in ("s3.json", "start-seaweedfs.sh")]]:
|
|
self.before[str(path)] = file_sha(path)
|
|
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())
|
|
storage_size = sum(int(self.runner.run(["du", "-sb", str(self.root / path)]).split()[0]) for path in self.storage_paths)
|
|
self.backup_size = database_size * 2 + storage_size + GIB
|
|
if self.storage:
|
|
image = json.loads(self.runner.run(["docker", "image", "inspect", self.storage["image"]]))[0]
|
|
self.backup_size += image["Size"] * 2
|
|
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,
|
|
"storageProvider": "seaweedfs" if self.storage else "minio",
|
|
"storageImageRetained": bool(self.storage), "storageBackupPaths": self.storage_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 = CORE_SERVICES | ({"objectstore"} if self.storage else {"minio", "createbuckets"})
|
|
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 lokalem Dateispeicher. "
|
|
+ ". ".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.")
|
|
data_services = [("db", "/var/lib/postgresql/data", "postgres")]
|
|
if not self.storage:
|
|
data_services.append(("minio", "/data", "minio"))
|
|
for name, target, directory in data_services:
|
|
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 (() if self.storage else ("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())
|
|
if self.storage:
|
|
self.validate_storage(self.current, self.storage, running=False)
|
|
cid = self.compose(False, "ps", "--all", "--quiet", "objectstore").strip()
|
|
store_state = self.inspect(cid).get("State", {}) if cid else {"Running": True}
|
|
if any(store_state.get(k) for k in ("Running", "Restarting", "Paused")):
|
|
raise UpdateError("SeaweedFS muss vor der kalten Sicherung gestoppt sein.")
|
|
atomic(directory / "config/.updates/storage-provider.json", (self.home / "storage-provider.json").read_bytes())
|
|
self.runner.run(["docker", "image", "save", "-o", str(directory / "storage-image.tar"), self.storage["image"]], timeout=3600)
|
|
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 = [*self.storage_paths, *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()
|
|
if self.storage:
|
|
cid = self.compose(False, "ps", "--all", "--quiet", "objectstore").strip()
|
|
if not cid or any(self.inspect(cid).get("State", {}).get(k) for k in ("Running", "Restarting", "Paused")):
|
|
raise UpdateError("SeaweedFS wurde während der kalten Sicherung gestartet.")
|
|
write_json(directory / "complete.json", {"createdAt": now(), "database": "database.dump", "files": "files.tar",
|
|
"storageProvider": "seaweedfs" if self.storage else "minio", "storageDataPaths": self.storage_paths,
|
|
"storageImage": self.storage["image"] if self.storage else None,
|
|
"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 not IMAGE_ID.fullmatch(service["image"])
|
|
and bool(re.search(r"@sha256:[a-f0-9]{64}$", service["image"])) == (policy == "missing")]
|
|
if services:
|
|
self.compose(True, "pull", "--policy", policy, *services, timeout=3600)
|
|
for service in self.candidate["services"].values():
|
|
if IMAGE_ID.fullmatch(service["image"]):
|
|
self.runner.run(["docker", "image", "inspect", service["image"]])
|
|
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.")
|
|
if self.storage:
|
|
self.validate_storage(self.current, self.storage)
|
|
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", self.storage_service, "node-exporter")
|
|
if self.storage:
|
|
self.save("checking_storage", storageCheckContainer="fedeo-storage-check-" + self.job.name,
|
|
storageProbeKey=".fedeo-update-check/" + uuid.uuid4().hex + ".pdf")
|
|
self.compose(True, "run", "--pull", "never", "--rm", "--no-deps", "--name", self.state["storageCheckContainer"],
|
|
"-e", "FEDEO_UPDATE_PROBE_KEY=" + self.state["storageProbeKey"],
|
|
"--entrypoint", "node", "backend", "-e", S3_UPGRADE_CHECK, timeout=210)
|
|
self.save(storageProbeKey=None)
|
|
else:
|
|
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)
|
|
if self.storage:
|
|
self.validate_storage(self.candidate, self.storage)
|
|
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("phase") == "storage_migration":
|
|
raise UpdateError("Speichermigration prüfen und nach vollständigem Abschluss adopt-storage ausführen; kein resolve verwenden.")
|
|
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 state.get("phase") == "storage_migration":
|
|
raise UpdateError("Speichermigration kann nicht mit resolve freigegeben werden; Status prüfen und nach Abschluss adopt-storage verwenden.")
|
|
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.storage = self.storage_record()
|
|
self.storage_service = "objectstore" if self.storage else "minio"
|
|
if self.storage:
|
|
self.validate_storage(current, self.storage)
|
|
self.public = current["services"]["frontend"]["environment"]["NUXT_PUBLIC_API_BASE"].removesuffix("/backend")
|
|
for key in ("migrationContainer", "storageCheckContainer"):
|
|
if state.get(key):
|
|
remaining = self.runner.run(["docker", "ps", "--all", "--quiet", "--filter", "name=^/" + state[key] + "$"]).strip()
|
|
if remaining:
|
|
raise UpdateError("Den verbliebenen Prüf-/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 + "/")
|
|
if state.get("storageProbeKey"):
|
|
if not self.storage or not re.fullmatch(r"\.fedeo-update-check/[a-f0-9]{32}\.pdf", state["storageProbeKey"]):
|
|
raise UpdateError("Ungültiger protokollierter S3-Prüfschlüssel; manuell prüfen.")
|
|
self.compose(False, "exec", "-T", "-e", "FEDEO_UPDATE_PROBE_KEY=" + state["storageProbeKey"],
|
|
"backend", "node", "-e", S3_CLEANUP_CHECK, timeout=90)
|
|
self.save(storageProbeKey=None)
|
|
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", self.storage_service):
|
|
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
|
|
restored = self.inspect(original["Id"])
|
|
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", "adopt-storage"))
|
|
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)
|
|
elif args.command == "adopt-storage":
|
|
result = updater.adopt_storage()
|
|
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())
|