377 lines
17 KiB
Python
377 lines
17 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
|
|
|
|
spec = importlib.util.spec_from_file_location("fedeo_updater", Path(__file__).with_name("selfhost-update.py"))
|
|
engine = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(engine)
|
|
|
|
REQUEST_ID = re.compile(r"[A-Za-z0-9][A-Za-z0-9_-]{7,79}\Z")
|
|
PHASES = {"preparing", "validating", "checked", "pulling", "maintenance", "backed_up", "applying",
|
|
"retiring_matrix", "migrating", "starting", "verifying", "complete", "resolved"}
|
|
STATUSES = {"idle", "queued", "running", "checked", "up_to_date", "succeeded", "failed", "interrupted", "recovered"}
|
|
ERRORS = {
|
|
"INVALID_REQUEST": "Ungültige Anfrage.",
|
|
"NOT_FOUND": "Auftrag oder Endpunkt nicht gefunden.",
|
|
"BUSY": "Eine Prüfung oder ein Update läuft bereits.",
|
|
"SHUTTING_DOWN": "Der Dienst wird beendet.",
|
|
"REQUEST_CONFLICT": "Diese Auftrags-ID wurde für eine andere Anfrage verwendet.",
|
|
"CHECK_REQUIRED": "Zuerst genau diese Zielversion erfolgreich prüfen.",
|
|
"COMPOSE_REVIEW_REQUIRED": "Die Compose-Datei muss zuerst auf dem Host geprüft und übernommen werden.",
|
|
"RECOVERY_REQUIRED": "Ein vorheriges Update benötigt Wiederherstellung auf dem Host.",
|
|
"PLAN_CHANGED": "Release oder Konfiguration haben sich geändert. Erneut prüfen.",
|
|
"UPDATE_FAILED": "Auftrag fehlgeschlagen. Geschützte Protokolle auf dem Host prüfen.",
|
|
"INTERRUPTED": "Der Auftrag wurde unterbrochen. Status auf dem Host prüfen.",
|
|
}
|
|
|
|
|
|
class ApiError(Exception):
|
|
def __init__(self, code, status=400):
|
|
self.code, self.status = code, status
|
|
|
|
|
|
def identifier(value):
|
|
if not isinstance(value, str) or not REQUEST_ID.fullmatch(value):
|
|
raise ApiError("INVALID_REQUEST")
|
|
return value
|
|
|
|
|
|
def public_state(state):
|
|
"""An allowlist, never raw command logs, exceptions, host paths, config or diff contents."""
|
|
result = {"status": state.get("status") if state.get("status") in STATUSES else "idle"}
|
|
if state.get("phase") in PHASES:
|
|
result["phase"] = state["phase"]
|
|
for key in ("recoveryRequired", "maintenance", "mutationStarted"):
|
|
if key in state:
|
|
result[key] = bool(state[key])
|
|
for key in ("version", "previousVersion"):
|
|
if isinstance(state.get(key), str) and engine.VERSION.fullmatch(state[key]):
|
|
result[key] = state[key]
|
|
if re.fullmatch(r"[a-f0-9]{40}", str(state.get("revision", ""))):
|
|
result["revision"] = state["revision"]
|
|
for key in ("startedAt", "updatedAt"):
|
|
if isinstance(state.get(key), str) and re.fullmatch(r"[0-9T:+.Z-]{10,40}", state[key]):
|
|
result[key] = state[key]
|
|
if state.get("requestId") and REQUEST_ID.fullmatch(str(state["requestId"])):
|
|
result["requestId"] = state["requestId"]
|
|
if state.get("command") in ("check", "apply"):
|
|
result["command"] = state["command"]
|
|
return result
|
|
|
|
|
|
def public_plan(plan):
|
|
result = {k: v for k, v in public_state(plan).items() if k != "status"}
|
|
for key in ("updateAvailable", "requiresComposeAdoption"):
|
|
result[key] = bool(plan.get(key))
|
|
if type(plan.get("estimatedBackupBytes")) is int:
|
|
result["estimatedBackupBytes"] = plan["estimatedBackupBytes"]
|
|
result["retiredServices"] = sorted(set(plan.get("retiredServices", [])) & engine.LEGACY_MATRIX_SERVICES)
|
|
return result
|
|
|
|
|
|
class Controller:
|
|
def __init__(self, directory, *, updater_factory=engine.Updater):
|
|
self.directory = Path(directory).resolve()
|
|
self.factory = updater_factory
|
|
updater = self.factory(self.directory)
|
|
self.home = updater.home / "service"
|
|
self.home.mkdir(mode=0o700, exist_ok=True)
|
|
if self.home.is_symlink():
|
|
raise engine.UpdateError("Dienstverzeichnis darf kein symbolischer Link sein.")
|
|
os.chmod(self.home, 0o700)
|
|
self.gate = threading.Lock()
|
|
self.worker = None
|
|
self.closing = False
|
|
self.active = None
|
|
self.daemon_lock = (self.home / "daemon.lock").open("a")
|
|
try:
|
|
fcntl.flock(self.daemon_lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
|
except OSError:
|
|
self.daemon_lock.close()
|
|
raise engine.UpdateError("Für diese Instanz läuft bereits ein Update-Dienst.") from None
|
|
# A persisted request is an audit record, not a queue that silently replays updates.
|
|
try:
|
|
for path in self.home.glob("*.json"):
|
|
record = engine.read_json(path)
|
|
if record.get("status") in ("queued", "running"):
|
|
record.update(status="interrupted", errorCode="INTERRUPTED", updatedAt=engine.now())
|
|
engine.write_json(path, record)
|
|
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 status(self):
|
|
updater = self.factory(self.directory)
|
|
result = {"apiVersion": 1, "engine": public_state(updater.status()), "acceptingRequests": not self.closing}
|
|
installed = engine.read_json(updater.home / "installed.json", {}).get("manifest", {})
|
|
result["installed"] = {k: v for k, v in public_state({"version": installed.get("version"),
|
|
"revision": installed.get("commit")}).items() if k != "status"}
|
|
if self.active:
|
|
result["request"] = self.request_status(self.active)
|
|
return result
|
|
|
|
def submit(self, command, body):
|
|
if not isinstance(body, dict) or command not in ("check", "apply"):
|
|
raise ApiError("INVALID_REQUEST")
|
|
allowed = {"requestId", "version"} | ({"checkRequestId"} if command == "apply" else set())
|
|
if set(body) - allowed:
|
|
raise ApiError("INVALID_REQUEST")
|
|
request_id = identifier(body.get("requestId"))
|
|
version = body.get("version")
|
|
if version is not None and (not isinstance(version, str) or not engine.VERSION.fullmatch(version)):
|
|
raise ApiError("INVALID_REQUEST")
|
|
if command == "apply" and not version:
|
|
raise ApiError("INVALID_REQUEST")
|
|
check_id = identifier(body.get("checkRequestId")) if command == "apply" else None
|
|
payload = {"command": command, "version": version, "checkRequestId": check_id}
|
|
with self.gate:
|
|
previous = engine.read_json(self.path(request_id))
|
|
if previous:
|
|
if previous["input"] != payload:
|
|
raise ApiError("REQUEST_CONFLICT", 409)
|
|
return 200, self.request_status(request_id)
|
|
if self.closing:
|
|
raise ApiError("SHUTTING_DOWN", 503)
|
|
if self.worker and self.worker.is_alive():
|
|
raise ApiError("BUSY", 409)
|
|
state = self.factory(self.directory).status()
|
|
if state.get("recoveryRequired"):
|
|
raise ApiError("RECOVERY_REQUIRED", 409)
|
|
if state.get("status") == "running":
|
|
raise ApiError("BUSY", 409)
|
|
fingerprint = None
|
|
if command == "apply":
|
|
try:
|
|
check = self.read(check_id)
|
|
except ApiError:
|
|
raise ApiError("CHECK_REQUIRED", 409) from None
|
|
plan = check.get("plan", {})
|
|
if (check.get("command") != "check" or check.get("status") != "checked"
|
|
or plan.get("version") != version or not plan.get("updateAvailable")
|
|
or not re.fullmatch(r"[a-f0-9]{64}", str(plan.get("approvalFingerprint", "")))):
|
|
raise ApiError("CHECK_REQUIRED", 409)
|
|
if plan.get("requiresComposeAdoption"):
|
|
raise ApiError("COMPOSE_REVIEW_REQUIRED", 409)
|
|
fingerprint = plan["approvalFingerprint"]
|
|
record = {"requestId": request_id, "command": command, "input": payload,
|
|
"status": "queued", "startedAt": engine.now(), "updatedAt": engine.now()}
|
|
engine.write_json(self.path(request_id), record)
|
|
self.active = request_id
|
|
self.worker = threading.Thread(target=self.run, args=(record, fingerprint), daemon=False)
|
|
try:
|
|
self.worker.start()
|
|
except BaseException:
|
|
record.update(status="failed", errorCode="UPDATE_FAILED", updatedAt=engine.now())
|
|
engine.write_json(self.path(request_id), record)
|
|
raise
|
|
return 202, public_state(record)
|
|
|
|
def run(self, record, fingerprint):
|
|
request_id = record["requestId"]
|
|
try:
|
|
record.update(status="running", updatedAt=engine.now())
|
|
engine.write_json(self.path(request_id), record)
|
|
updater = self.factory(self.directory)
|
|
result = updater.execute(record["command"], record["input"]["version"],
|
|
expected_fingerprint=fingerprint, request_id=request_id)
|
|
record.update(status=result["status"], plan=result["plan"])
|
|
if record["command"] == "check" and not result["plan"]["updateAvailable"]:
|
|
record["status"] = "up_to_date"
|
|
except BaseException as error:
|
|
code = "PLAN_CHANGED" if isinstance(error, engine.PlanChangedError) else "UPDATE_FAILED"
|
|
record.update(status="failed", errorCode=code)
|
|
with (self.home / (request_id + ".log")).open("a") as log:
|
|
os.chmod(log.name, 0o600)
|
|
traceback.print_exc(file=log)
|
|
finally:
|
|
record["updatedAt"] = engine.now()
|
|
engine.write_json(self.path(request_id), record)
|
|
|
|
def close(self):
|
|
with self.gate:
|
|
self.closing = True
|
|
worker = self.worker
|
|
if worker:
|
|
worker.join() # Normal service stop drains the active update instead of interrupting a migration.
|
|
self.daemon_lock.close()
|
|
|
|
|
|
class Handler(BaseHTTPRequestHandler):
|
|
protocol_version = "HTTP/1.0"
|
|
server_version = "FEDEO-Updater/1"
|
|
sys_version = ""
|
|
|
|
def setup(self):
|
|
super().setup()
|
|
self.connection.settimeout(5)
|
|
|
|
def log_message(self, *_):
|
|
pass
|
|
|
|
def send_json(self, status, body):
|
|
data = json.dumps(body, ensure_ascii=False).encode()
|
|
self.close_connection = True
|
|
self.send_response(status)
|
|
self.send_header("Content-Type", "application/json; charset=utf-8")
|
|
self.send_header("Content-Length", str(len(data)))
|
|
self.send_header("Cache-Control", "no-store")
|
|
self.send_header("Connection", "close")
|
|
self.end_headers()
|
|
self.wfile.write(data)
|
|
|
|
def dispatch(self):
|
|
try:
|
|
if self.headers.get("Origin") or self.headers.get("Transfer-Encoding"):
|
|
raise ApiError("INVALID_REQUEST")
|
|
controller = self.server.controller
|
|
if self.command == "GET" and self.path == "/v1/status":
|
|
self.send_json(200, controller.status())
|
|
elif self.command == "GET" and self.path.startswith("/v1/requests/"):
|
|
self.send_json(200, controller.request_status(self.path.removeprefix("/v1/requests/")))
|
|
elif self.command == "POST" and self.path in ("/v1/check", "/v1/apply"):
|
|
lengths = self.headers.get_all("Content-Length", [])
|
|
if (len(lengths) != 1 or not lengths[0].isdigit() or not 1 <= int(lengths[0]) <= 4096
|
|
or self.headers.get_content_type() != "application/json"):
|
|
raise ApiError("INVALID_REQUEST")
|
|
body = json.loads(self.rfile.read(int(lengths[0])))
|
|
status, result = controller.submit(self.path.rsplit("/", 1)[-1], body)
|
|
self.send_json(status, result)
|
|
else:
|
|
raise ApiError("NOT_FOUND", 404)
|
|
except ApiError as error:
|
|
self.send_json(error.status, {"error": {"code": error.code, "message": ERRORS[error.code]}})
|
|
except (ValueError, UnicodeError):
|
|
self.send_json(400, {"error": {"code": "INVALID_REQUEST", "message": ERRORS["INVALID_REQUEST"]}})
|
|
except (OSError, engine.UpdateError, KeyError, TypeError):
|
|
with contextlib.suppress(OSError):
|
|
self.send_json(500, {"error": {"code": "UPDATE_FAILED", "message": ERRORS["UPDATE_FAILED"]}})
|
|
|
|
do_GET = dispatch
|
|
do_POST = dispatch
|
|
|
|
|
|
class Server(socketserver.ThreadingMixIn, socketserver.UnixStreamServer):
|
|
daemon_threads = True
|
|
request_queue_size = 8
|
|
|
|
def __init__(self, path, controller):
|
|
self.controller = controller
|
|
self.slots = threading.BoundedSemaphore(16)
|
|
super().__init__(str(path), Handler)
|
|
os.chmod(path, 0o660)
|
|
|
|
def process_request(self, request, client_address):
|
|
if not self.slots.acquire(blocking=False):
|
|
self.shutdown_request(request)
|
|
return
|
|
try:
|
|
super().process_request(request, client_address)
|
|
except BaseException:
|
|
self.slots.release()
|
|
raise
|
|
|
|
def process_request_thread(self, request, client_address):
|
|
try:
|
|
super().process_request_thread(request, client_address)
|
|
finally:
|
|
self.slots.release()
|
|
|
|
def handle_error(self, *_):
|
|
pass # Do not send request data or tracebacks to the journal.
|
|
|
|
|
|
def serve(directory, socket_path):
|
|
controller = None
|
|
path = Path(socket_path)
|
|
stop = threading.Event()
|
|
server = None
|
|
socket_lock = None
|
|
try:
|
|
path.parent.mkdir(mode=0o750, parents=True, exist_ok=True)
|
|
parent = path.parent.stat()
|
|
if path.parent.is_symlink() or parent.st_uid != 0 or parent.st_mode & 0o022:
|
|
raise engine.UpdateError("Socket-Verzeichnis muss root gehören und darf nicht durch andere beschreibbar sein.")
|
|
socket_lock = path.with_name(path.name + ".lock").open("a")
|
|
try:
|
|
fcntl.flock(socket_lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
|
except OSError:
|
|
raise engine.UpdateError("An diesem Socket läuft bereits ein Update-Dienst.") from None
|
|
controller = Controller(directory)
|
|
if path.is_symlink() or (path.exists() and not stat.S_ISSOCK(path.stat().st_mode)):
|
|
raise engine.UpdateError("Socket-Pfad ist bereits anderweitig belegt.")
|
|
# Both the instance and endpoint are locked before replacing a stale socket.
|
|
path.unlink(missing_ok=True)
|
|
server = Server(path, controller)
|
|
for sig in (signal.SIGTERM, signal.SIGINT):
|
|
signal.signal(sig, lambda *_: stop.set())
|
|
server.timeout = 0.5
|
|
while not stop.is_set():
|
|
server.handle_request()
|
|
finally:
|
|
if controller:
|
|
controller.close()
|
|
if server:
|
|
server.server_close()
|
|
path.unlink(missing_ok=True)
|
|
if socket_lock:
|
|
socket_lock.close()
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("command", choices=("serve",))
|
|
parser.add_argument("--directory", default="/opt/fedeo")
|
|
parser.add_argument("--socket", default="/run/fedeo-updater/control.sock")
|
|
args = parser.parse_args()
|
|
if os.geteuid() != 0:
|
|
parser.error("Der Host-Dienst muss als root laufen.")
|
|
os.umask(0o077)
|
|
try:
|
|
serve(args.directory, args.socket)
|
|
except (OSError, engine.UpdateError) as error:
|
|
parser.exit(1, f"Update-Dienst konnte nicht gestartet werden: {error}\n")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|