mission-control-v2/backend/services/homelab/arcane.py

"""Arcane und Docker auf der Arcane-VM (Phase 3/4, 24.09.2026).

Arcane (getarcaneapp/arcane) verwaltet die Docker-Container der VM (NerdQuiz, Rippy …) und hat eine
eigene Schnittstelle (OpenAPI unter /api/openapi.json). Öffentlich ist nur die eigene Version
(/api/app-version mit currentVersion, newestVersion, updateAvailable). Alles über Images braucht einen
API-Schlüssel (Kopfzeile X-API-Key), den man in Arcane anlegt.

Der Schlüssel (seit 24.09.2026 auch über die Oberfläche): MC_ARCANE_KEY in /etc/mc2/homelab.env geht vor, sonst
<Datenordner>/arcane.key (0600, schreibt nur die Oberfläche nach einer Prüfung gegen Arcane, einstellungen.py).
Beides wird bei jedem Zugriff gelesen: Ein neuer Schlüssel gilt ohne Neustart. Er steht nie in einer Antwort,
einem Protokoll oder einer Fehlermeldung.

Docker-Updates laufen über Arcanes eigenen Updater (POST /environments/{id}/updater/run). User-Entscheid
24.09.: zuerst nur als Probelauf (dryRun); echt erst, wenn der Schalter in den Einstellungen an ist
(<Datenordner>/homelab-einstellungen.json). Setzt die Umgebung MC_ARCANE_ECHT, gewinnt sie („1“ = echt).
"""

import json
import os
import threading
import time
from datetime import datetime
from pathlib import Path

import httpx
from kern.einstellungen import einstellungen
from kern.zeit import LOCAL_TZ

ZEITLIMIT = httpx.Timeout(20.0, connect=5.0)
CACHE_S = 120
_cache: dict[str, tuple[float, object]] = {}
_lock = threading.Lock()
_datei_lock = threading.Lock()


def _gemerkt(schluessel: str, holen):
    """Arcane merkt sich den Update-Stand selbst; die Übersicht fragt ihn höchstens alle zwei Minuten ab."""
    jetzt = time.time()
    with _lock:
        if (eintrag := _cache.get(schluessel)) and jetzt - eintrag[0] < CACHE_S:
            return eintrag[1]
    wert = holen()
    with _lock:
        _cache[schluessel] = (jetzt, wert)
    return wert


def cache_leeren() -> None:
    """Nach einem neuen oder gelöschten Schlüssel: Die Übersicht fragt Arcane gleich neu."""
    with _lock:
        _cache.clear()


# --- Schlüssel ----------------------------------------------------------------------------------------

def _schluessel_pfad() -> Path:
    return einstellungen().daten_dir / "arcane.key"


def _schluessel_datei() -> str:
    try:
        return _schluessel_pfad().read_text(encoding="utf-8").strip()
    except (OSError, UnicodeDecodeError):
        return ""


def schluessel() -> str:
    """Der geltende Schlüssel: MC_ARCANE_KEY, sonst die Datei, sonst leer."""
    return os.environ.get("MC_ARCANE_KEY", "").strip() or _schluessel_datei()


def schluessel_quelle() -> str:
    """„umgebung“ (MC_ARCANE_KEY), „hinterlegt“ (Datei im Datenordner) oder „fehlt“."""
    if os.environ.get("MC_ARCANE_KEY", "").strip():
        return "umgebung"
    return "hinterlegt" if _schluessel_datei() else "fehlt"


def schluessel_speichern(wert: str) -> None:
    """Atomar und von Anfang an nur für den Dienst lesbar (0600): erst eine Nachbardatei, dann ersetzen."""
    pfad = _schluessel_pfad()
    tmp = pfad.with_name(pfad.name + ".tmp")
    with _datei_lock:
        pfad.parent.mkdir(parents=True, exist_ok=True)
        tmp.unlink(missing_ok=True)
        fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
        with os.fdopen(fd, "w", encoding="utf-8") as f:
            f.write(wert.strip() + "\n")
        os.replace(tmp, pfad)
    cache_leeren()


def schluessel_loeschen() -> bool:
    """Die Datei weg. True, wenn es eine gab."""
    with _datei_lock:
        pfad = _schluessel_pfad()
        da = pfad.exists()
        pfad.unlink(missing_ok=True)
    cache_leeren()
    return da


# --- Echt statt Probelauf -----------------------------------------------------------------------------

def _einstellungen_pfad() -> Path:
    return einstellungen().daten_dir / "homelab-einstellungen.json"


def _einstellungen_lesen() -> dict:
    try:
        daten = json.loads(_einstellungen_pfad().read_text(encoding="utf-8"))
    except (OSError, ValueError):
        return {}
    return daten if isinstance(daten, dict) else {}


def _echt_eingestellt() -> bool | None:
    arcane = _einstellungen_lesen().get("arcane")
    wert = arcane.get("echt") if isinstance(arcane, dict) else None
    return wert if isinstance(wert, bool) else None


def _echt_umgebung() -> str:
    return os.environ.get("MC_ARCANE_ECHT", "").strip()


def echt() -> bool:
    """Echte Docker-Updates statt Probelauf? Die Umgebung (MC_ARCANE_ECHT) gewinnt, dann der Schalter, sonst nein."""
    if umgebung := _echt_umgebung():
        return umgebung == "1"
    return bool(_echt_eingestellt())


def echt_quelle() -> str:
    """„umgebung“, „einstellung“ oder „standard“ (Probelauf)."""
    if _echt_umgebung():
        return "umgebung"
    return "einstellung" if _echt_eingestellt() is not None else "standard"


def echt_speichern(an: bool) -> None:
    pfad = _einstellungen_pfad()
    with _datei_lock:
        daten = _einstellungen_lesen()
        arcane = daten.get("arcane") if isinstance(daten.get("arcane"), dict) else {}
        daten.update(arcane={**arcane, "echt": bool(an)},
                     geaendert=datetime.now(LOCAL_TZ).isoformat(timespec="seconds"))
        pfad.parent.mkdir(parents=True, exist_ok=True)
        tmp = pfad.with_name(pfad.name + ".tmp")
        tmp.write_text(json.dumps(daten, ensure_ascii=False, indent=1), encoding="utf-8")
        os.replace(tmp, pfad)


# --- Schnittstelle ------------------------------------------------------------------------------------

def app_version(url: str) -> dict | None:
    """Arcanes eigene Version (öffentlich): {"installiert", "neu", "update"} oder None."""
    return _gemerkt(f"version:{url}", lambda: _app_version(url))


def _app_version(url: str) -> dict | None:
    try:
        d = httpx.get(f"{url}/api/app-version", timeout=ZEITLIMIT).json()
    except (httpx.HTTPError, ValueError):
        return None
    if not isinstance(d, dict) or not d.get("currentVersion"):
        return None
    return {"installiert": str(d["currentVersion"]).lstrip("v"), "neu": str(d.get("newestVersion") or "").lstrip("v") or None,
            "update": bool(d.get("updateAvailable"))}


def _get(url: str, pfad: str, mit_schluessel: str | None = None) -> object:
    r = httpx.get(f"{url}/api{pfad}", headers={"X-API-Key": mit_schluessel or schluessel()}, timeout=ZEITLIMIT)
    r.raise_for_status()
    return r.json()


def umgebungen(url: str, mit_schluessel: str | None = None) -> list[dict]:
    daten = _get(url, "/environments?limit=50", mit_schluessel)
    return [e for e in (daten or {}).get("data") or [] if isinstance(e, dict) and e.get("id") is not None]


def images_mit_update(url: str) -> list[dict]:
    """Images mit einem neueren Stand in der Registry, über alle Umgebungen: {"name", "aktuell", "neu", "benutzt"}."""
    return _gemerkt(f"images:{url}", lambda: _images_mit_update(url))


def _images_mit_update(url: str) -> list[dict]:
    ergebnis = []
    for umgebung in umgebungen(url):
        daten = _get(url, f"/environments/{umgebung['id']}/images?limit=500")
        for image in (daten or {}).get("data") or []:
            info = image.get("updateInfo") or {}
            if not info.get("hasUpdate"):
                continue
            name = f"{image.get('repo') or '?'}:{image.get('tag') or 'latest'}"
            ergebnis.append({"name": name, "aktuell": info.get("currentVersion") or None,
                             "neu": info.get("latestVersion") or None, "umgebung": umgebung["id"],
                             "benutzt": [str(u.get("name") if isinstance(u, dict) else u) for u in image.get("usedBy") or []]})
    return ergebnis


def schluessel_pruefen(url: str, wert: str) -> dict:
    """Nimmt Arcane diesen Schlüssel an? Geprüft wird mit denselben Aufrufen, die die Übersicht braucht: die
    Umgebungen, dann ihre Images. {"ok": True, "umgebungen", "images", "mit_update"} oder {"ok": False, "status":
    HTTP-Code bzw. None, "art": "abgelehnt" | "http" | "netz" | "antwort"}. Nichts davon enthält den Schlüssel."""
    try:
        liste = umgebungen(url, wert)
        images = mit_update = 0
        for umgebung in liste:
            daten = _get(url, f"/environments/{umgebung['id']}/images?limit=500", wert)
            eintraege = [i for i in (daten or {}).get("data") or [] if isinstance(i, dict)]
            images += len(eintraege)
            mit_update += sum(1 for i in eintraege if (i.get("updateInfo") or {}).get("hasUpdate"))
    except httpx.HTTPStatusError as exc:
        status = exc.response.status_code
        return {"ok": False, "status": status, "art": "abgelehnt" if status in (401, 403) else "http"}
    except httpx.HTTPError:
        return {"ok": False, "status": None, "art": "netz"}
    except (ValueError, AttributeError, TypeError):
        return {"ok": False, "status": None, "art": "antwort"}
    return {"ok": True, "umgebungen": len(liste), "images": images, "mit_update": mit_update}


def updater(url: str, probelauf: bool) -> dict:
    """Arcanes Updater für alle Umgebungen. Rückgabe: je Umgebung das Ergebnis (geprüft, aktualisiert, Fehler)."""
    ergebnisse = {}
    for umgebung in umgebungen(url):
        r = httpx.post(f"{url}/api/environments/{umgebung['id']}/updater/run", headers={"X-API-Key": schluessel()},
                       json={"dryRun": probelauf}, timeout=httpx.Timeout(600.0, connect=5.0))
        r.raise_for_status()
        ergebnisse[str(umgebung.get("name") or umgebung["id"])] = (r.json() or {}).get("data") or {}
    return ergebnisse