HitonabiandClaude Opus 5.5
ad8dbbc1b3
Jede Gerätekachel (Homelab-Cockpit und Übersicht) zeigt eine Live-Zahl, wo es eine gibt: Proxmox-Host „8 von 8 Gästen an“, PBS „letzte Sicherung vor 13 h“, NPMplus „10 Zertifikate gültig“ (aus der Zertifikats-Wache), Arcane „11 von 12 Containern an“ (Arcane-API mit dem vorhandenen Schlüssel) und AdGuard „12 % geblockt“. Für AdGuard gibt es unter Einstellungen → Homelab einen Zugang, der vor dem Speichern gegen AdGuard geprüft wird und nur im Homelab-Teil liegt (0600, das Passwort steht in keiner Antwort). Arcane und AdGuard werden höchstens alle zwei Minuten gefragt. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
254 lines
10 KiB
Python
254 lines
10 KiB
Python
"""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 container_zahlen(url: str) -> dict | None:
|
|
"""Laufende und alle Container über alle Umgebungen, für die Kachel (seit 25.09.2026): {"laufend", "gesamt"}.
|
|
None ohne Schlüssel oder wenn Arcane nicht antwortet."""
|
|
if not schluessel():
|
|
return None
|
|
return _gemerkt(f"container:{url}", lambda: _container_zahlen(url))
|
|
|
|
|
|
def _container_zahlen(url: str) -> dict | None:
|
|
laufend = gesamt = 0
|
|
try:
|
|
for umgebung in umgebungen(url):
|
|
daten = (_get(url, f"/environments/{umgebung['id']}/containers/counts") or {}).get("data") or {}
|
|
laufend += int(daten.get("runningContainers") or 0)
|
|
gesamt += int(daten.get("totalContainers") or 0)
|
|
except (httpx.HTTPError, ValueError, TypeError, AttributeError):
|
|
return None
|
|
return {"laufend": laufend, "gesamt": gesamt}
|
|
|
|
|
|
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
|