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

"""Zertifikats-Wache (Ausbauplan Welle 1, seit 25.09.2026).

Der Ausführer liest höchstens alle 30 Minuten die https-Zertifikate (host.zertifikate im Bericht): die von NPMplus —
Let's Encrypt im kurzlebigen Profil, am 25.09.2026 gut 6,7 Tage Laufzeit — samt dem, was NPMplus für den Namen gerade
ausliefert, das des Proxmox-Hosts (vom Cluster ausgestellt, zwei Jahre) und das des PBS (selbst signiert). Eine feste
30-Tage-Grenze passt zu so verschiedenen Laufzeiten nicht; die Grenzen richten sich nach der Laufzeit:
  • gelb „wird nicht erneuert“: Bei kurzlebigen Zertifikaten (bis 100 Tage) ist weniger als ein Drittel der Laufzeit
    übrig — NPMplus erneuert lange vorher. Bei langlebigen ab 30 Tagen vor Ablauf.
  • rot: weniger als ein Sechstel übrig, dabei höchstens 7 und mindestens 1 Tag vor Ablauf; abgelaufen sowieso.
  • gelb: NPMplus liefert ein älteres Zertifikat aus, als es hat (nach der Erneuerung nicht neu geladen).
Hinweise gehen über den Wächter (pruefe_zertifikate), je Quelle einer; die Liste zeigt das Homelab-Cockpit.
"""

import time
from datetime import datetime

from kern.zeit import LOCAL_TZ

from services.homelab import kanal

TAG_S = 86400
KURZLEBIG_S = 100 * TAG_S
LANG_GELB_S = 30 * TAG_S
ROT_MAX_S, ROT_MIN_S = 7 * TAG_S, TAG_S
AUSGELIEFERT_TOLERANZ_S = 60
RANG = {"ok": 0, "gelb": 1, "rot": 2}
QUELLE = {"npm": "NPMplus", "pve": "Proxmox-Host", "pbs": "Proxmox Backup Server"}
WESSEN = {"pve": "des Proxmox-Hosts", "pbs": "des PBS"}


def _datum(ts: float) -> str:
    return datetime.fromtimestamp(ts, LOCAL_TZ).strftime("%d.%m. %H:%M")


def _zeitraum(sekunden: float) -> tuple[int, str, str]:
    """(Anzahl, Einheit im Akkusativ, Einheit im Dativ) — 20 Stunden, 5 Tage, 3 Monate, 2 Jahre."""
    if sekunden < TAG_S:
        return max(1, round(sekunden / 3600)), "Stunden", "Stunden"
    tage = round(sekunden / TAG_S)
    if tage < 60:
        return tage, "Tage", "Tagen"
    return (round(tage / 30), "Monate", "Monaten") if tage < 730 else (round(tage / 365), "Jahre", "Jahren")


def bis_ablauf(sekunden: float) -> str:
    """„in 20 Stunden“, „in einem Tag“, „in 5 Tagen“, „in 3 Monaten“."""
    n, _, dativ = _zeitraum(sekunden)
    if n == 1:
        return {"Stunden": "in einer Stunde", "Tagen": "in einem Tag", "Monaten": "in einem Monat",
                "Jahren": "in einem Jahr"}[dativ]
    return f"in {n} {dativ}"


def dauer(sekunden: float) -> str:
    """„20 Stunden“, „einen Tag“, „5 Tage“ — für „noch … gültig“."""
    n, akkusativ, _ = _zeitraum(sekunden)
    if n == 1:
        return {"Stunden": "eine Stunde", "Tage": "einen Tag", "Monate": "einen Monat", "Jahre": "ein Jahr"}[akkusativ]
    return f"{n} {akkusativ}"


def bewerten(z: dict, jetzt: float) -> dict:
    """Ein Zertifikat aus dem Bericht mit Stufe, Art des Problems (abgelaufen, ablauf, haengt, alt) und einem Satz."""
    von, bis = z.get("von"), z["bis"]
    laufzeit = bis - von if von else None
    rest = bis - jetzt
    kurzlebig = laufzeit is not None and laufzeit <= KURZLEBIG_S
    rot_ab = max(ROT_MIN_S, min(ROT_MAX_S, laufzeit / 6)) if laufzeit else ROT_MAX_S
    ausgeliefert = z.get("ausgeliefert_bis")
    alt = bool(ausgeliefert and ausgeliefert < bis - AUSGELIEFERT_TOLERANZ_S)
    stufe, art, satz = "ok", None, None
    if rest <= 0:
        stufe, art, satz = "rot", "abgelaufen", f"abgelaufen am {_datum(bis)}"
    elif rest < rot_ab or (not kurzlebig and rest < LANG_GELB_S):
        stufe, art, satz = "rot" if rest < rot_ab else "gelb", "ablauf", f"läuft {bis_ablauf(rest)} ab ({_datum(bis)})"
    elif kurzlebig and rest < laufzeit / 3:
        stufe, art, satz = "gelb", "haengt", f"wird nicht erneuert, noch {dauer(rest)} gültig (bis {_datum(bis)})"
    elif alt:
        stufe, art, satz = "gelb", "alt", f"ausgeliefert wird noch das alte (bis {_datum(ausgeliefert)})"
    namen = [n for n in z.get("namen") or [] if n]
    # Anzeigename: bei Proxmox-Host und PBS der Rechnername (ihr erster Name ist „localhost“), sonst der erste Name.
    eigener = z.get("inhaber") if z.get("quelle") in WESSEN else None
    name = eigener or next((n for n in namen if n != "localhost"), None) or z.get("inhaber") or z.get("name")
    return {"quelle": z.get("quelle"), "wo": QUELLE.get(z.get("quelle"), z.get("quelle")), "gast": z.get("gast"),
            "name": name, "namen": namen, "datei": z.get("name"),
            "aussteller": z.get("aussteller"), "von": von, "bis": bis, "rest_s": int(rest),
            "laufzeit_s": int(laufzeit) if laufzeit else None, "stufe": stufe, "art": art, "problem": satz}


def lage(jetzt: float | None = None) -> dict:
    """Für GET /api/homelab/zertifikate: die Zertifikate, Probleme zuerst, dann das am frühesten ablaufende.
    „gelesen“: ob der Ausführer schon Zertifikate meldet (einer von vor dem 25.09.2026 tut es nicht)."""
    jetzt = time.time() if jetzt is None else jetzt
    host = ((kanal.bericht() or {}).get("bericht") or {}).get("host") or {}
    roh = host.get("zertifikate")
    liste = [bewerten(z, jetzt) for z in roh or [] if isinstance(z, dict) and isinstance(z.get("bis"), int)]
    liste.sort(key=lambda z: (-RANG[z["stufe"]], z["bis"]))
    return {"gelesen": isinstance(roh, list), "zertifikate": liste}


RAT = {
    "npm": ("In NPMplus: SSL Certificates → beim Zertifikat „…“ → Renew. Klappt das nicht, steht der Grund im Protokoll "
            "von NPMplus (Container {gast}) — oft ist die Domain von außen nicht erreichbar."),
    "pve": "In Proxmox: pve → System → Zertifikate. Das vom Cluster ausgestellte erneuert `pvecm updatecerts --force`.",
    "pbs": "Im PBS: Konfiguration → Zertifikate — dort ein neues ausstellen oder hochladen.",
}


def _titel(quelle: str, faelle: list[dict], stufe: str) -> str:
    if len(faelle) > 1:
        return (f"{len(faelle)} Zertifikate von NPMplus laufen bald ab" if stufe == "rot"
                else f"NPMplus erneuert {len(faelle)} Zertifikate nicht")
    z = faelle[0]
    wessen = f"für {z['name']}" if quelle == "npm" else WESSEN.get(quelle, z["name"])
    if z["art"] == "alt":
        return f"NPMplus liefert für {z['name']} noch das alte Zertifikat aus"
    if z["art"] == "abgelaufen":
        return f"Zertifikat {wessen} ist abgelaufen"
    if z["art"] == "haengt":
        return f"Zertifikat {wessen} wird nicht erneuert"
    return f"Zertifikat {wessen} läuft {bis_ablauf(z['rest_s'])} ab"


def pruefe_zertifikate(jetzt: float | None = None) -> list:
    """Befunde für den Wächter: je Quelle (NPMplus, Proxmox-Host, PBS) einer mit der ernstesten Stufe."""
    from services.waechter import Befund  # der Wächter registriert diese Prüfung, also erst hier
    befunde = []
    liste = lage(jetzt)["zertifikate"]
    for quelle in QUELLE:
        faelle = [z for z in liste if z["quelle"] == quelle and z["stufe"] != "ok"]
        if not faelle:
            continue
        stufe = max((z["stufe"] for z in faelle), key=RANG.get)
        nur_alt = all(z["art"] == "alt" for z in faelle)
        rat = ("NPMplus einmal neu starten, dann liefert es das neue aus." if nur_alt
               else "Läuft es ab, warnen Browser, und Apps verbinden sich nicht mehr. "
               + RAT[quelle].format(gast=faelle[0]["gast"]))
        einzeln = "; ".join(f"{z['name']} {z['problem']}" for z in faelle[:6])
        befunde.append(Befund(id=f"zertifikat:{quelle}", stufe=stufe, titel=_titel(quelle, faelle, stufe),
                              text=f"{einzeln}. {rat}",
                              quelle=f"ct-{faelle[0]['gast']}" if faelle[0]["gast"] else "pve"))
    return befunde