Files
FEDEO/scripts/selfhost-update-service.py
flfeders 29155f916a
All checks were successful
Build and Release FEDEO / verify (push) Successful in 48s
Build and Release FEDEO / build-and-release (push) Successful in 5m45s
feat(selfhost): show sanitized update logs in admin UI
2026-09-25 18:42:26 +02:00

490 lines
22 KiB
Python

#!/usr/bin/env python3
"""Local HTTP/Unix-socket control service for the FEDEO host updater. No TCP listener."""
import argparse
import contextlib
import fcntl
from http.server import BaseHTTPRequestHandler
import importlib.util
import json
import os
from pathlib import Path
import re
import signal
import socketserver
import stat
import threading
import traceback
import urllib.parse
spec = importlib.util.spec_from_file_location("fedeo_updater", Path(__file__).with_name("selfhost-update.py"))
engine = importlib.util.module_from_spec(spec)
spec.loader.exec_module(engine)
REQUEST_ID = re.compile(r"[A-Za-z0-9][A-Za-z0-9_-]{7,79}\Z")
PHASES = {"preparing", "validating", "checked", "pulling", "maintenance", "backed_up", "applying",
"retiring_matrix", "migrating", "starting", "verifying", "complete", "resolved"}
STATUSES = {"idle", "queued", "running", "checked", "up_to_date", "succeeded", "failed", "interrupted", "recovered"}
ERRORS = {
"INVALID_REQUEST": "Ungültige Anfrage.",
"NOT_FOUND": "Auftrag oder Endpunkt nicht gefunden.",
"BUSY": "Eine Prüfung oder ein Update läuft bereits.",
"SHUTTING_DOWN": "Der Dienst wird beendet.",
"REQUEST_CONFLICT": "Diese Auftrags-ID wurde für eine andere Anfrage verwendet.",
"CHECK_REQUIRED": "Zuerst genau diese Zielversion erfolgreich prüfen.",
"COMPOSE_REVIEW_REQUIRED": "Die Compose-Datei muss zuerst auf dem Host geprüft und übernommen werden.",
"RECOVERY_REQUIRED": "Ein vorheriges Update benötigt Wiederherstellung auf dem Host.",
"PLAN_CHANGED": "Release oder Konfiguration haben sich geändert. Erneut prüfen.",
"UPDATE_FAILED": "Auftrag fehlgeschlagen.",
"LOGS_UNAVAILABLE": "Das Update-Protokoll kann gerade nicht gelesen werden.",
"INTERRUPTED": "Der Auftrag wurde unterbrochen. Status auf dem Host prüfen.",
}
class ApiError(Exception):
def __init__(self, code, status=400):
self.code, self.status = code, status
def identifier(value):
if not isinstance(value, str) or not REQUEST_ID.fullmatch(value):
raise ApiError("INVALID_REQUEST")
return value
def public_state(state):
"""An allowlist, never raw command logs, exceptions, host paths, config or diff contents."""
result = {"status": state.get("status") if state.get("status") in STATUSES else "idle"}
if state.get("phase") in PHASES:
result["phase"] = state["phase"]
for key in ("recoveryRequired", "maintenance", "mutationStarted"):
if key in state:
result[key] = bool(state[key])
for key in ("version", "previousVersion"):
if isinstance(state.get(key), str) and engine.VERSION.fullmatch(state[key]):
result[key] = state[key]
if re.fullmatch(r"[a-f0-9]{40}", str(state.get("revision", ""))):
result["revision"] = state["revision"]
for key in ("startedAt", "updatedAt"):
if isinstance(state.get(key), str) and re.fullmatch(r"[0-9T:+.Z-]{10,40}", state[key]):
result[key] = state[key]
if state.get("requestId") and REQUEST_ID.fullmatch(str(state["requestId"])):
result["requestId"] = state["requestId"]
if state.get("command") in ("check", "apply"):
result["command"] = state["command"]
return result
def public_plan(plan):
result = {k: v for k, v in public_state(plan).items() if k != "status"}
for key in ("updateAvailable", "requiresComposeAdoption"):
result[key] = bool(plan.get(key))
if type(plan.get("estimatedBackupBytes")) is int:
result["estimatedBackupBytes"] = plan["estimatedBackupBytes"]
result["retiredServices"] = sorted(set(plan.get("retiredServices", [])) & engine.LEGACY_MATRIX_SERVICES)
return result
LOG_BYTES = 6000
SECRET_KEY = r"[\w.-]*(?:password|passwd|secret|token|credential|authorization|(?:api|private|access|encryption|signing)[_-]?key|database[_-]?url|dsn|auth)[\w.-]*"
QUOTED_VALUE = r"\"(?:\\.|[^\"\\])*\"|'[^']*'"
SECRET_VALUE = re.compile(r"(?im)(?<![\w.-])([\"']?" + SECRET_KEY + r"[\"']?\s*[:=]\s*)(" + QUOTED_VALUE + r"|[^\r\n,}\]]+)")
CONFIG_SECRET = re.compile(r"(?im)^\s*(?:export\s+|-\s*)?[\"']?" + SECRET_KEY + r"[\"']?\s*[:=]\s*(" + QUOTED_VALUE + r"|[^\r\n]+)")
def read_private(path, *, tail=False):
"""Read only regular files, never follow a log/config symlink or block on a FIFO."""
try:
fd = os.open(path, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK)
except FileNotFoundError:
return "", False
with os.fdopen(fd, "rb") as stream:
size = os.fstat(stream.fileno()).st_size
if not stat.S_ISREG(os.fstat(stream.fileno()).st_mode):
raise OSError("Not a regular file")
limit = 32768 if tail else 1024 * 1024
if size > limit and not tail:
raise OSError("Configuration too large to redact safely")
truncated = size > limit
if truncated:
stream.seek(-limit, os.SEEK_END)
stream.readline(limit) # Never expose a suffix of a secret on a cut line.
return stream.read(limit).decode("utf-8", errors="replace"), truncated
def log_redactor(directory, job):
# Include original values after an update changed the active configuration.
paths = [directory / name for name in (".env", "docker-compose.yml", "docker-compose.override.yml", "docker-compose.override.yaml")]
if job:
paths += [job / "target.env", job / "target.compose.yml"]
paths += [job / "backup/config" / name for name in (".env", "docker-compose.yml", "docker-compose.override.yml", "docker-compose.override.yaml")]
values = {value for key, value in os.environ.items() if re.fullmatch(SECRET_KEY, key, re.I) and value}
for path in paths:
content, _ = read_private(path)
for match in CONFIG_SECRET.finditer(content):
value = match[1].strip()
if value[:1] in ("'", '"'):
if value.startswith('"'):
try:
values.add(json.loads(value))
except ValueError:
pass
value = value[1:-1]
else:
value = value.split(" #", 1)[0].strip()
if value:
values.add(value)
for match in re.finditer(r"[a-z][a-z0-9+.-]*://[^\s\"']+", content, re.I):
password = urllib.parse.urlsplit(match[0]).password
if password:
values.update((password, urllib.parse.unquote(password)))
values |= {urllib.parse.quote(value, safe="") for value in values}
def redact(content):
content = re.sub(r"\x1b(?:\[[0-?]*[ -/]*[@-~]|\][^\x07]*(?:\x07|\x1b\\))", "", content)
for value in sorted(values, key=len, reverse=True):
content = content.replace(value, "[REDACTED]")
content = re.sub(r"(?is)-----BEGIN [^-]*PRIVATE KEY-----.*?(?:-----END [^-]*PRIVATE KEY-----|\Z)", "[REDACTED]", content)
content = re.sub(r"(?i)([a-z][a-z0-9+.-]*://)[^\s/]*@", r"\1[REDACTED]@", content)
content = re.sub(r"(?i)\b(Bearer|Basic|token)\s+[A-Za-z0-9._~+/=-]+", r"\1 [REDACTED]", content)
content = SECRET_VALUE.sub(lambda m: m[1] + "[REDACTED]", content)
content = re.sub(r"(?i)(--" + SECRET_KEY + r"\s+)(" + QUOTED_VALUE + r"|\S+)", r"\1[REDACTED]", content)
return re.sub(r"[\x00-\x08\x0b-\x1f\x7f]", "", content)
return redact
class Controller:
def __init__(self, directory, *, updater_factory=engine.Updater):
self.directory = Path(directory).resolve()
self.factory = updater_factory
updater = self.factory(self.directory)
self.home = updater.home / "service"
self.home.mkdir(mode=0o700, exist_ok=True)
if self.home.is_symlink():
raise engine.UpdateError("Dienstverzeichnis darf kein symbolischer Link sein.")
os.chmod(self.home, 0o700)
self.gate = threading.Lock()
self.worker = None
self.closing = False
self.active = None
self.daemon_lock = (self.home / "daemon.lock").open("a")
try:
fcntl.flock(self.daemon_lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
except OSError:
self.daemon_lock.close()
raise engine.UpdateError("Für diese Instanz läuft bereits ein Update-Dienst.") from None
# A persisted request is an audit record, not a queue that silently replays updates.
try:
for path in self.home.glob("*.json"):
record = engine.read_json(path)
if record.get("status") in ("queued", "running"):
record.update(status="interrupted", errorCode="INTERRUPTED", updatedAt=engine.now())
engine.write_json(path, record)
if (REQUEST_ID.fullmatch(str(record.get("requestId", "")))
and (self.active is None or record.get("startedAt", "") > self.read(self.active).get("startedAt", ""))):
self.active = record["requestId"]
updater.status() # Detect an abandoned engine job without touching an active CLI job.
except BaseException:
self.daemon_lock.close()
raise
def path(self, request_id):
return self.home / (identifier(request_id) + ".json")
def read(self, request_id):
record = engine.read_json(self.path(request_id))
if not record:
raise ApiError("NOT_FOUND", 404)
return record
def request_status(self, request_id):
record = self.read(request_id)
result = public_state(record)
if record.get("plan"):
result["plan"] = public_plan(record["plan"])
code = record.get("errorCode")
if code in ERRORS:
result["error"] = {"code": code, "message": ERRORS[code]}
if record["status"] == "running":
state = self.factory(self.directory).status()
if state.get("requestId") == request_id:
result["progress"] = public_state(state)
return result
def request_logs(self, request_id):
record = self.read(request_id)
try:
home = self.home.parent
state = engine.read_json(home / "status.json", {})
candidates = [Path(record["job"])] if record.get("job") else []
if state.get("requestId") == request_id and state.get("job"):
candidates.append(Path(state["job"]))
# Older service versions did not persist the engine-job association.
if not candidates:
candidates = [path.parent for path in home.glob("*/status.json")]
job = None
for candidate in candidates:
if candidate.parent != home or candidate.is_symlink() or not candidate.is_dir():
continue
content, _ = read_private(candidate / "status.json")
if content and json.loads(content).get("requestId") == request_id:
job = candidate
break
redact = log_redactor(self.directory, job)
files = [("service", self.home / (request_id + ".log"))]
if job:
files += [("events", job / "events.jsonl"), ("commands", job / "commands.log")]
logs = []
for source, path in files:
content, truncated = read_private(path, tail=True)
content = redact(content)
encoded = content.encode("utf-8")
if len(encoded) > LOG_BYTES:
content = encoded[-LOG_BYTES:].decode("utf-8", errors="ignore")
truncated = True
if content or truncated:
logs.append({"source": source, "content": content, "truncated": truncated})
return {"requestId": request_id, "logs": logs}
except (OSError, ValueError, TypeError):
raise ApiError("LOGS_UNAVAILABLE", 503) from None
def status(self):
updater = self.factory(self.directory)
result = {"apiVersion": 1, "engine": public_state(updater.status()), "acceptingRequests": not self.closing,
"capabilities": {"requestLogs": True}}
installed = engine.read_json(updater.home / "installed.json", {}).get("manifest", {})
result["installed"] = {k: v for k, v in public_state({"version": installed.get("version"),
"revision": installed.get("commit")}).items() if k != "status"}
if self.active:
result["request"] = self.request_status(self.active)
return result
def submit(self, command, body):
if not isinstance(body, dict) or command not in ("check", "apply"):
raise ApiError("INVALID_REQUEST")
allowed = {"requestId", "version"} | ({"checkRequestId"} if command == "apply" else set())
if set(body) - allowed:
raise ApiError("INVALID_REQUEST")
request_id = identifier(body.get("requestId"))
version = body.get("version")
if version is not None and (not isinstance(version, str) or not engine.VERSION.fullmatch(version)):
raise ApiError("INVALID_REQUEST")
if command == "apply" and not version:
raise ApiError("INVALID_REQUEST")
check_id = identifier(body.get("checkRequestId")) if command == "apply" else None
payload = {"command": command, "version": version, "checkRequestId": check_id}
with self.gate:
previous = engine.read_json(self.path(request_id))
if previous:
if previous["input"] != payload:
raise ApiError("REQUEST_CONFLICT", 409)
return 200, self.request_status(request_id)
if self.closing:
raise ApiError("SHUTTING_DOWN", 503)
if self.worker and self.worker.is_alive():
raise ApiError("BUSY", 409)
state = self.factory(self.directory).status()
if state.get("recoveryRequired"):
raise ApiError("RECOVERY_REQUIRED", 409)
if state.get("status") == "running":
raise ApiError("BUSY", 409)
fingerprint = None
if command == "apply":
try:
check = self.read(check_id)
except ApiError:
raise ApiError("CHECK_REQUIRED", 409) from None
plan = check.get("plan", {})
if (check.get("command") != "check" or check.get("status") != "checked"
or plan.get("version") != version or not plan.get("updateAvailable")
or not re.fullmatch(r"[a-f0-9]{64}", str(plan.get("approvalFingerprint", "")))):
raise ApiError("CHECK_REQUIRED", 409)
if plan.get("requiresComposeAdoption"):
raise ApiError("COMPOSE_REVIEW_REQUIRED", 409)
fingerprint = plan["approvalFingerprint"]
record = {"requestId": request_id, "command": command, "input": payload,
"status": "queued", "startedAt": engine.now(), "updatedAt": engine.now()}
engine.write_json(self.path(request_id), record)
self.active = request_id
self.worker = threading.Thread(target=self.run, args=(record, fingerprint), daemon=False)
try:
self.worker.start()
except BaseException:
record.update(status="failed", errorCode="UPDATE_FAILED", updatedAt=engine.now())
engine.write_json(self.path(request_id), record)
raise
return 202, public_state(record)
def run(self, record, fingerprint):
request_id = record["requestId"]
updater = None
try:
record.update(status="running", updatedAt=engine.now())
engine.write_json(self.path(request_id), record)
updater = self.factory(self.directory)
result = updater.execute(record["command"], record["input"]["version"],
expected_fingerprint=fingerprint, request_id=request_id)
record.update(status=result["status"], plan=result["plan"])
if record["command"] == "check" and not result["plan"]["updateAvailable"]:
record["status"] = "up_to_date"
except BaseException as error:
code = "PLAN_CHANGED" if isinstance(error, engine.PlanChangedError) else "UPDATE_FAILED"
record.update(status="failed", errorCode=code)
with (self.home / (request_id + ".log")).open("a") as log:
os.chmod(log.name, 0o600)
traceback.print_exc(file=log)
finally:
if updater and updater.state and updater.state.get("requestId") == request_id:
record["job"] = str(updater.job)
record["updatedAt"] = engine.now()
engine.write_json(self.path(request_id), record)
def close(self):
with self.gate:
self.closing = True
worker = self.worker
if worker:
worker.join() # Normal service stop drains the active update instead of interrupting a migration.
self.daemon_lock.close()
class Handler(BaseHTTPRequestHandler):
protocol_version = "HTTP/1.0"
server_version = "FEDEO-Updater/1"
sys_version = ""
def setup(self):
super().setup()
self.connection.settimeout(5)
def log_message(self, *_):
pass
def send_json(self, status, body):
data = json.dumps(body, ensure_ascii=False).encode()
self.close_connection = True
self.send_response(status)
self.send_header("Content-Type", "application/json; charset=utf-8")
self.send_header("Content-Length", str(len(data)))
self.send_header("Cache-Control", "no-store")
self.send_header("Connection", "close")
self.end_headers()
self.wfile.write(data)
def dispatch(self):
try:
if self.headers.get("Origin") or self.headers.get("Transfer-Encoding"):
raise ApiError("INVALID_REQUEST")
controller = self.server.controller
if self.command == "GET" and self.path == "/v1/status":
self.send_json(200, controller.status())
elif self.command == "GET" and self.path.startswith("/v1/requests/"):
suffix = self.path.removeprefix("/v1/requests/")
self.send_json(200, controller.request_logs(suffix.removesuffix("/logs"))
if suffix.endswith("/logs") else controller.request_status(suffix))
elif self.command == "POST" and self.path in ("/v1/check", "/v1/apply"):
lengths = self.headers.get_all("Content-Length", [])
if (len(lengths) != 1 or not lengths[0].isdigit() or not 1 <= int(lengths[0]) <= 4096
or self.headers.get_content_type() != "application/json"):
raise ApiError("INVALID_REQUEST")
body = json.loads(self.rfile.read(int(lengths[0])))
status, result = controller.submit(self.path.rsplit("/", 1)[-1], body)
self.send_json(status, result)
else:
raise ApiError("NOT_FOUND", 404)
except ApiError as error:
self.send_json(error.status, {"error": {"code": error.code, "message": ERRORS[error.code]}})
except (ValueError, UnicodeError):
self.send_json(400, {"error": {"code": "INVALID_REQUEST", "message": ERRORS["INVALID_REQUEST"]}})
except (OSError, engine.UpdateError, KeyError, TypeError):
with contextlib.suppress(OSError):
self.send_json(500, {"error": {"code": "UPDATE_FAILED", "message": ERRORS["UPDATE_FAILED"]}})
do_GET = dispatch
do_POST = dispatch
class Server(socketserver.ThreadingMixIn, socketserver.UnixStreamServer):
daemon_threads = True
request_queue_size = 8
def __init__(self, path, controller):
self.controller = controller
self.slots = threading.BoundedSemaphore(16)
super().__init__(str(path), Handler)
os.chmod(path, 0o660)
def process_request(self, request, client_address):
if not self.slots.acquire(blocking=False):
self.shutdown_request(request)
return
try:
super().process_request(request, client_address)
except BaseException:
self.slots.release()
raise
def process_request_thread(self, request, client_address):
try:
super().process_request_thread(request, client_address)
finally:
self.slots.release()
def handle_error(self, *_):
pass # Do not send request data or tracebacks to the journal.
def serve(directory, socket_path):
controller = None
path = Path(socket_path)
stop = threading.Event()
server = None
socket_lock = None
try:
path.parent.mkdir(mode=0o750, parents=True, exist_ok=True)
parent = path.parent.stat()
if path.parent.is_symlink() or parent.st_uid != 0 or parent.st_mode & 0o022:
raise engine.UpdateError("Socket-Verzeichnis muss root gehören und darf nicht durch andere beschreibbar sein.")
socket_lock = path.with_name(path.name + ".lock").open("a")
try:
fcntl.flock(socket_lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
except OSError:
raise engine.UpdateError("An diesem Socket läuft bereits ein Update-Dienst.") from None
controller = Controller(directory)
if path.is_symlink() or (path.exists() and not stat.S_ISSOCK(path.stat().st_mode)):
raise engine.UpdateError("Socket-Pfad ist bereits anderweitig belegt.")
# Both the instance and endpoint are locked before replacing a stale socket.
path.unlink(missing_ok=True)
server = Server(path, controller)
for sig in (signal.SIGTERM, signal.SIGINT):
signal.signal(sig, lambda *_: stop.set())
server.timeout = 0.5
while not stop.is_set():
server.handle_request()
finally:
if controller:
controller.close()
if server:
server.server_close()
path.unlink(missing_ok=True)
if socket_lock:
socket_lock.close()
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("command", choices=("serve",))
parser.add_argument("--directory", default="/opt/fedeo")
parser.add_argument("--socket", default="/run/fedeo-updater/control.sock")
args = parser.parse_args()
if os.geteuid() != 0:
parser.error("Der Host-Dienst muss als root laufen.")
os.umask(0o077)
try:
serve(args.directory, args.socket)
except (OSError, engine.UpdateError) as error:
parser.exit(1, f"Update-Dienst konnte nicht gestartet werden: {error}\n")
if __name__ == "__main__":
main()