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

"""Einstellungen des Homelab-Teils (24.09.2026): was sich unter „Einstellungen“ für das Homelab einstellen lässt.

  • Arcane-API-Schlüssel  eintragen oder löschen. Vor dem Speichern prüft Arcane ihn mit denselben Aufrufen, die die
                          Übersicht braucht; lehnt es ab oder antwortet nicht, wird nichts gespeichert. Er liegt in
                          <Datenordner>/arcane.key (0600); MC_ARCANE_KEY in /etc/mc2/homelab.env geht vor (arcane.py).
  • Docker echt           echte Docker-Updates über Arcane statt Probelauf (<Datenordner>/homelab-einstellungen.json);
                          MC_ARCANE_ECHT in der Umgebung gewinnt.
stand() zeigt dazu die Wartezeit nach Skriptänderungen (karenz.py) und den Ausführer (verbunden, Nur-Lesen,
Sicherungsspeicher). Der Schlüssel verlässt diesen Teil nie: Keine Antwort, kein Protokolleintrag und keine
Fehlermeldung enthält ihn, auch kein Stück davon.
"""

import json
import logging
import re
import time

from services.homelab import apps, arcane, inventar, kanal, karenz

log = logging.getLogger(__name__)

UMGEBUNG_SCHLUESSEL = ("Der Arcane-Schlüssel kommt aus der Umgebung (MC_ARCANE_KEY in /etc/mc2/homelab.env) und "
                       "gilt vor einem hier eingetragenen. Erst dort entfernen, dann hier eintragen.")
_FORM = re.compile(r"[\x21-\x7e]{8,512}")     # druckbares ASCII ohne Leerzeichen, wie ihn Arcane ausgibt


def arcane_url() -> str | None:
    """Adresse der Arcane-Oberfläche aus dem Bericht des Ausführers (die VM, deren Name „arcane“ ist)."""
    eingang = kanal.bericht()
    for gast in ((eingang or {}).get("bericht") or {}).get("gaeste") or []:
        app = apps.app_fuer((gast.get("app") or {}).get("kennung"), gast.get("name"))
        if app and app.kennung == "arcane" and (url := apps.adresse(app, gast.get("ip"))):
            return url.rstrip("/")
    return None


def stand() -> dict:
    """GET /api/homelab/einstellungen."""
    zuletzt = kanal.zuletzt()
    host = ((kanal.bericht() or {}).get("bericht") or {}).get("host") or {}
    sicherung = host.get("sicherung") if isinstance(host.get("sicherung"), dict) else {}
    return {
        "arcane": {"url": arcane_url() or "", "schluessel": arcane.schluessel_quelle(), "echt": arcane.echt(),
                   "echt_quelle": arcane.echt_quelle()},
        "karenz_h": round(karenz.karenz_s() / 3600),
        "ausfuehrer": {"verbunden": bool(zuletzt and time.time() - zuletzt < inventar.BERICHT_ALT_S),
                       "zuletzt": zuletzt, "nur_lesen": bool(host.get("nur_lesen")),
                       "sicherung_speicher": sicherung.get("speicher") or None},
    }


def schluessel_aus_body(roh: bytes) -> str | None:
    """Der Schlüssel aus dem Body {"schluessel": "…"} — ohne Pydantic, dessen Fehlermeldung (422) die Eingabe
    wiederholen würde. None, wenn der Body nicht passt."""
    try:
        daten = json.loads(roh.decode("utf-8")) if roh else None
    except (UnicodeDecodeError, ValueError):
        return None
    wert = daten.get("schluessel") if isinstance(daten, dict) else None
    return wert if isinstance(wert, str) else None


def _abgelehnt(ergebnis: dict, url: str) -> str:
    status, art = ergebnis.get("status"), ergebnis.get("art")
    if art == "abgelehnt":
        text = f"Arcane lehnt den Schlüssel ab (HTTP {status})."
        if status == 403:
            text += " Ihm fehlt das Recht, Umgebungen und Images zu lesen."
        return text
    if art == "http":
        return f"Arcane antwortet mit HTTP {status}. Der Schlüssel ist nicht gespeichert."
    if art == "netz":
        return f"Arcane ist nicht erreichbar ({url}). Der Schlüssel ist nicht gespeichert."
    return "Arcane antwortet nicht wie erwartet. Der Schlüssel ist nicht gespeichert."


def arcane_schluessel_setzen(wert: str | None) -> dict:
    """POST /api/homelab/einstellungen/arcane-schluessel: erst gegen Arcane prüfen, dann speichern."""
    if arcane.schluessel_quelle() == "umgebung":
        return {"ok": False, "detail": UMGEBUNG_SCHLUESSEL}
    wert = (wert or "").strip()
    if not _FORM.fullmatch(wert):
        return {"ok": False, "detail": "Das sieht nicht wie ein Arcane-API-Schlüssel aus (leer, zu kurz oder mit "
                                       "Leerzeichen). Nichts gespeichert."}
    url = arcane_url()
    if not url:
        return {"ok": False, "detail": "Arcane steht nicht im Bericht des Ausführers, die Adresse ist unbekannt. "
                                       "Nichts gespeichert."}
    ergebnis = arcane.schluessel_pruefen(url, wert)
    if not ergebnis["ok"]:
        log.info("homelab: Arcane-Schlüssel nicht übernommen (%s, HTTP %s)", ergebnis.get("art"), ergebnis.get("status"))
        return {"ok": False, "detail": _abgelehnt(ergebnis, url)}
    try:
        arcane.schluessel_speichern(wert)
    except OSError as exc:
        return {"ok": False, "detail": f"Arcane nimmt den Schlüssel an, aber er ließ sich nicht speichern "
                                       f"({exc.__class__.__name__})."}
    images, neu, anzahl = ergebnis["images"], ergebnis["mit_update"], ergebnis["umgebungen"]
    log.info("homelab: Arcane-Schlüssel gespeichert (%d Images geprüft)", images)
    return {"ok": True, "text": (f"Arcane nimmt den Schlüssel an: {images} Images in {anzahl} "
                                 f"{'Umgebung' if anzahl == 1 else 'Umgebungen'} geprüft, {neu} davon mit neuerem "
                                 "Stand. Die Übersicht zeigt die Docker-Images ab sofort.")}


def arcane_schluessel_loeschen() -> dict:
    """DELETE /api/homelab/einstellungen/arcane-schluessel: die Datei weg."""
    try:
        da = arcane.schluessel_loeschen()
    except OSError as exc:
        return {"ok": False, "detail": f"Der Schlüssel ließ sich nicht löschen ({exc.__class__.__name__})."}
    text = "Der Arcane-Schlüssel ist gelöscht." if da else "Es war kein Arcane-Schlüssel hinterlegt."
    if arcane.schluessel_quelle() == "umgebung":
        text += " Die Umgebung setzt weiter einen (MC_ARCANE_KEY); der gilt."
    return {"ok": True, "text": text}


def arcane_echt_setzen(an: bool) -> dict:
    """POST /api/homelab/einstellungen/arcane-echt: echte Docker-Updates statt Probelauf."""
    if arcane.echt_quelle() == "umgebung":
        return {"ok": False, "detail": f"Die Umgebung legt das fest (MC_ARCANE_ECHT in /etc/mc2/homelab.env: Docker-"
                                       f"Updates laufen {'echt' if arcane.echt() else 'als Probelauf'}). Sie gewinnt "
                                       "vor diesem Schalter."}
    try:
        arcane.echt_speichern(an)
    except OSError as exc:
        return {"ok": False, "detail": f"Der Schalter ließ sich nicht speichern ({exc.__class__.__name__})."}
    return {"ok": True, "text": ("Docker-Updates über Arcane laufen ab jetzt echt: Arcane holt die neuen Images und "
                                 "startet die Container neu." if an else
                                 "Docker-Updates über Arcane laufen wieder nur als Probelauf; geändert wird nichts.")}