Welle 1 · Zertifikats-Wache: NPMplus, Proxmox-Host und PBS mit Grenzen nach Laufzeit

Der Ausführer liest alle 30 Minuten die https-Zertifikate (NPMplus samt dem, was es für den Namen ausliefert,
Proxmox-Host, PBS). Weil NPMplus das kurzlebige Profil von Let's Encrypt nutzt (6,7 Tage), gilt keine feste
30-Tage-Grenze: gelb „wird nicht erneuert“, sobald weniger als ein Drittel der Laufzeit übrig ist (langlebige: 30 Tage
vorher), rot unter einem Sechstel (1–7 Tage vor Ablauf), gelb auch, wenn NPMplus noch das alte ausliefert. Je Quelle
ein Hinweis; das Homelab-Cockpit zeigt die Liste.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Hitonabi
2026-09-25 16:45:57 +02:00
co-authored by Claude Opus 5.5
parent afa3a476b5
commit 53ea0cd9b6
11 changed files with 610 additions and 6 deletions
+8
View File
@@ -11,6 +11,7 @@
GET /api/homelab/snapshots Snapshots und Sicherungen je Gerät (Aktionen: siehe unten)
POST /api/homelab/snapshots/rueckwege-loeschen Alle Rückwege von Updates löschen (nur per Klick)
POST /api/homelab/sicherungen/probe Probe-Wiederherstellung jetzt (nur lesend, sonst einmal die Woche)
GET /api/homelab/zertifikate https-Zertifikate (NPMplus, Proxmox-Host, PBS) mit Stufe
GET /api/homelab/hinweise Hinweise des Wächters dieser Instanz
GET /api/homelab/ausfuehrer Lebenszeichen des Ausführers
GET /api/homelab/ausfuehrer/auftrag ┐
@@ -156,6 +157,13 @@ def speicher_lage() -> dict:
return speicher.lage()
@router.get("/zertifikate")
def zertifikate_lage() -> dict:
"""Die https-Zertifikate mit Ablauf und Stufe (Zertifikats-Wache, seit 25.09.2026)."""
from services.homelab import zertifikate
return zertifikate.lage()
@router.post("/sicherungen/probe")
def sicherungen_probe() -> dict:
"""Knopf „Jetzt prüfen“: die jüngste Sicherung jedes Geräts probeweise öffnen (seit 25.09.2026)."""
+139
View File
@@ -0,0 +1,139 @@
"""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]
return {"quelle": z.get("quelle"), "wo": QUELLE.get(z.get("quelle"), z.get("quelle")), "gast": z.get("gast"),
"name": namen[0] if namen else z.get("inhaber") or z.get("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
+2
View File
@@ -622,6 +622,8 @@ PRUEFUNGEN += (__import__("services.homelab.pflege").homelab.pflege.pruefe_paket
PRUEFUNGEN += (__import__("services.homelab.messwerte").homelab.messwerte.pruefe_host,) if ROLLE == "homelab" else ()
# Speicherpool, NAS und Sicherungen (seit 25.09.2026): gelb ab 80 %, rot bei 10 % frei; Sicherung älter als erlaubt.
PRUEFUNGEN += (__import__("services.homelab.speicher").homelab.speicher.pruefe_speicher,) if ROLLE == "homelab" else ()
# Zertifikats-Wache (seit 25.09.2026): gelb, wenn die Erneuerung hängt; rot kurz vor Ablauf.
PRUEFUNGEN += (__import__("services.homelab.zertifikate").homelab.zertifikate.pruefe_zertifikate,) if ROLLE == "homelab" else ()
BETREFF_PROBLEM, BETREFF_OK = {"box": ("[Box-Problem]", "[Box wieder ok]"),
"homelab": ("[Homelab-Problem]", "[Homelab wieder ok]")}[ROLLE]
+2 -1
View File
@@ -220,7 +220,8 @@ from services import waechter
print(json.dumps({"pruefungen": [p.__name__ for p in waechter.PRUEFUNGEN], "daten": str(waechter.DATEN_DIR)}))
""")
assert ergebnis["pruefungen"] == ["pruefe_platte", "pruefe_partner", "pruefe_ausfuehrer", "pruefe_gaeste",
"pruefe_gast_platten", "pruefe_paketlisten", "pruefe_host", "pruefe_speicher"]
"pruefe_gast_platten", "pruefe_paketlisten", "pruefe_host", "pruefe_speicher",
"pruefe_zertifikate"]
def test_homelab_rolle_hat_die_homelab_schnittstellen(tmp_path):
+160
View File
@@ -0,0 +1,160 @@
"""Zertifikats-Wache (Ausbauplan Welle 1, 25.09.2026): der Ausführer liest die Zertifikate (mit echten Ausgaben von
openssl auf dem Proxmox-PC), der Homelab-Teil bewertet sie nach ihrer Laufzeit — NPMplus nutzt das kurzlebige Profil
von Let's Encrypt (6,7 Tage), der Proxmox-Host hat zwei Jahre, der PBS tausend."""
import importlib.util
from datetime import datetime, timezone
from pathlib import Path
import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient
from kern import einstellungen as einstellungen_mod
from services.homelab import kanal, zertifikate
AUSFUEHRER = Path(__file__).resolve().parents[2] / "deploy" / "homelab" / "ausfuehrer.py"
TAG = 86400
def _utc(*teile: int) -> int:
return int(datetime(*teile, tzinfo=timezone.utc).timestamp())
@pytest.fixture(scope="module")
def a():
spec = importlib.util.spec_from_file_location("ausfuehrer_zertifikate", AUSFUEHRER)
modul = importlib.util.module_from_spec(spec)
spec.loader.exec_module(modul)
return modul
NPM_AUSGABE = """notBefore=Sep 23 04:45:02 2026 GMT
notAfter=Sep 29 20:45:01 2026 GMT
issuer=C=US, O=Let's Encrypt, CN=YE2
subject=CN=git.tobisniceshomelab.ddnsfree.com
X509v3 Subject Alternative Name: critical
DNS:git.tobisniceshomelab.ddnsfree.com
"""
PBS_AUSGABE = """notBefore=Jul 9 08:51:30 2026 GMT
notAfter=Nov 9 08:51:30 3025 GMT
issuer=O=Proxmox Backup Server, OU=6C03AEB2-FC61-470D-9C84-E039725B97DE, CN=proxmox-backup-server.fritz.box
subject=O=Proxmox Backup Server, OU=6C03AEB2-FC61-470D-9C84-E039725B97DE, CN=proxmox-backup-server.fritz.box
No extensions in certificate
"""
def test_ausgabe_von_openssl(a):
assert a.openssl_zeit("Sep 29 20:45:01 2026 GMT") == _utc(2026, 9, 29, 20, 45, 1)
assert a.openssl_zeit("Oct 1 17:46:16 2026 GMT") == _utc(2026, 10, 1, 17, 46, 16)
assert a.openssl_zeit("kaputt") is None
assert a.x509_auswerten(NPM_AUSGABE) == {
"von": _utc(2026, 9, 23, 4, 45, 2), "bis": _utc(2026, 9, 29, 20, 45, 1), "aussteller": "Let's Encrypt",
"inhaber": "git.tobisniceshomelab.ddnsfree.com", "namen": ["git.tobisniceshomelab.ddnsfree.com"]}
pbs = a.x509_auswerten(PBS_AUSGABE)
assert (pbs["namen"], pbs["inhaber"], pbs["aussteller"]) == ([], "proxmox-backup-server.fritz.box", "Proxmox Backup Server")
platzhalter = a.x509_auswerten("notAfter=Oct 1 17:46:16 2026 GMT\nX509v3 Subject Alternative Name:\n"
" DNS:*.tobisniceshomelab.ddnsfree.com, DNS:tobisniceshomelab.ddnsfree.com\n")
assert platzhalter["namen"] == ["*.tobisniceshomelab.ddnsfree.com", "tobisniceshomelab.ddnsfree.com"]
assert a.x509_auswerten("unable to load certificate") is None
def test_pem_bloecke(a):
text = "### /opt/npmplus/tls/certbot/live/npm-11/fullchain.pem\n-----BEGIN CERTIFICATE-----\nAAA\n" \
"### /opt/npmplus/tls/certbot/live/npm-4/fullchain.pem\n-----BEGIN CERTIFICATE-----\nBBB\n"
assert a.pem_bloecke(text) == {"/opt/npmplus/tls/certbot/live/npm-11/fullchain.pem": "-----BEGIN CERTIFICATE-----\nAAA\n",
"/opt/npmplus/tls/certbot/live/npm-4/fullchain.pem": "-----BEGIN CERTIFICATE-----\nBBB\n"}
def test_zertifikate_lesen(a, monkeypatch):
gaeste = [{"vmid": 101, "art": "lxc", "status": "running", "name": "npmplus", "ip": "192.168.178.110",
"app": {"kennung": "npmplus"}},
{"vmid": 105, "art": "lxc", "status": "running", "name": "proxmox-backup-server", "ip": "192.168.178.156"},
{"vmid": 102, "art": "lxc", "status": "stopped", "name": "npmplus"}]
befehle: list[str] = []
def im_gast(vmid, befehl, zeitlimit=30):
befehle.append(f"{vmid}: {befehl}")
if vmid == 101:
return 0, "### /opt/npmplus/tls/certbot/live/npm-11/fullchain.pem\nNPM\n"
return 0, "PBS"
monkeypatch.setattr(a, "_im_gast", im_gast)
monkeypatch.setattr(a, "_lesen", lambda pfad: "PVE")
monkeypatch.setattr(a.os.path, "exists", lambda pfad: pfad.endswith("pve-ssl.pem"))
monkeypatch.setattr(a, "_zertifikat", lambda pem: a.x509_auswerten(NPM_AUSGABE if "NPM" in pem else PBS_AUSGABE))
monkeypatch.setattr(a, "_ausgeliefert", lambda ip, namen: 12345 if ip == "192.168.178.110" else None)
liste = a._zertifikate_lesen(gaeste)
assert [(z["quelle"], z["gast"], z["name"]) for z in liste] == [
("pve", None, "pve-ssl.pem"), ("npm", 101, "npm-11"), ("pbs", 105, "proxy.pem")]
assert liste[1]["ausgeliefert_bis"] == 12345
assert len(befehle) == 2 and "/opt/npmplus/tls/certbot/live/*/fullchain.pem" in befehle[0]
# --- Bewertung -----------------------------------------------------------------------------------------------------
NPM = {"quelle": "npm", "gast": 101, "name": "npm-11", "von": _utc(2026, 9, 23, 4, 45, 2),
"bis": _utc(2026, 9, 29, 20, 45, 1), "aussteller": "Let's Encrypt", "namen": ["git.tobisniceshomelab.ddnsfree.com"]}
PVE = {"quelle": "pve", "gast": None, "name": "pve-ssl.pem", "von": _utc(2025, 8, 5, 16, 5, 14),
"bis": _utc(2027, 8, 5, 16, 5, 14), "aussteller": "PVE Cluster Manager CA", "inhaber": "pve.fritz.box", "namen": []}
def test_bewertung_nach_laufzeit():
frisch = zertifikate.bewerten(NPM, _utc(2026, 9, 25, 14, 40)) # 4,25 Tage übrig von 6,7
assert (frisch["stufe"], frisch["name"], frisch["laufzeit_s"]) == ("ok", "git.tobisniceshomelab.ddnsfree.com", 576_000 - 1)
haengt = zertifikate.bewerten(NPM, NPM["bis"] - 2 * TAG) # unter einem Drittel
assert (haengt["stufe"], haengt["art"]) == ("gelb", "haengt")
assert haengt["problem"].startswith("wird nicht erneuert, noch 2 Tage gültig (bis ")
knapp = zertifikate.bewerten(NPM, NPM["bis"] - 20 * 3600) # unter einem Sechstel (1,1 Tage)
assert (knapp["stufe"], knapp["art"]) == ("rot", "ablauf") and knapp["problem"].startswith("läuft in 20 Stunden ab")
weg = zertifikate.bewerten(NPM, NPM["bis"] + 60)
assert (weg["stufe"], weg["art"]) == ("rot", "abgelaufen")
# Der Proxmox-Host: zwei Jahre Laufzeit — ein Drittel wären acht Monate; gewarnt wird 30 Tage vorher.
assert zertifikate.bewerten(PVE, PVE["bis"] - 200 * TAG)["stufe"] == "ok"
bald = zertifikate.bewerten(PVE, PVE["bis"] - 20 * TAG)
assert (bald["stufe"], bald["name"], bald["problem"][:22]) == ("gelb", "pve.fritz.box", "läuft in 20 Tagen ab (")
assert zertifikate.bewerten(PVE, PVE["bis"] - 3 * TAG)["stufe"] == "rot"
# NPMplus liefert noch das alte aus, obwohl das neue da ist.
alt = zertifikate.bewerten({**NPM, "ausgeliefert_bis": NPM["bis"] - 3 * TAG}, _utc(2026, 9, 25, 14, 40))
assert (alt["stufe"], alt["art"]) == ("gelb", "alt")
assert zertifikate.bewerten({**NPM, "ausgeliefert_bis": NPM["bis"]}, _utc(2026, 9, 25, 14, 40))["stufe"] == "ok"
def test_texte():
assert [zertifikate.bis_ablauf(s) for s in (1800, 20 * 3600, TAG, 5 * TAG, 90 * TAG, 1000 * 365 * TAG)] == [
"in einer Stunde", "in 20 Stunden", "in einem Tag", "in 5 Tagen", "in 3 Monaten", "in 1000 Jahren"]
assert [zertifikate.dauer(s) for s in (3600, TAG, 2 * TAG)] == ["eine Stunde", "einen Tag", "2 Tage"]
def test_hinweise_je_quelle(monkeypatch):
jetzt = NPM["bis"] - 2 * TAG
zweites = {**NPM, "name": "npm-12", "namen": ["mc.tobisniceshomelab.ddnsfree.com"]}
monkeypatch.setattr(kanal, "bericht", lambda: {"bericht": {"host": {"zertifikate": [NPM, zweites, PVE]}}})
befunde = {b.id: b for b in zertifikate.pruefe_zertifikate(jetzt)}
assert set(befunde) == {"zertifikat:npm"} # der Proxmox-Host ist in Ordnung
npm = befunde["zertifikat:npm"]
assert (npm.stufe, npm.titel, npm.quelle) == ("gelb", "NPMplus erneuert 2 Zertifikate nicht", "ct-101")
assert "SSL Certificates" in npm.text and "Container 101" in npm.text
monkeypatch.setattr(kanal, "bericht", lambda: {"bericht": {"host": {"zertifikate": [
{**NPM, "ausgeliefert_bis": NPM["bis"] - 3 * TAG}, {**PVE, "bis": jetzt + 20 * TAG}]}}})
befunde = {b.id: b for b in zertifikate.pruefe_zertifikate(NPM["von"] + TAG)}
assert befunde["zertifikat:npm"].titel == ("NPMplus liefert für git.tobisniceshomelab.ddnsfree.com noch das alte "
"Zertifikat aus")
assert befunde["zertifikat:npm"].text.endswith("NPMplus einmal neu starten, dann liefert es das neue aus.")
assert befunde["zertifikat:pve"].titel.startswith("Zertifikat des Proxmox-Hosts läuft in ")
def test_schnittstelle(tmp_path, monkeypatch):
from routers import homelab
monkeypatch.setenv("MC_DATEN_DIR", str(tmp_path))
einstellungen_mod.einstellungen.cache_clear()
try:
app = FastAPI()
app.include_router(homelab.router)
client = TestClient(app)
assert client.get("/api/homelab/zertifikate").json() == {"gelesen": False, "zertifikate": []}
kanal.bericht_speichern({"host": {"zertifikate": [PVE, NPM, {"kaputt": True}]}, "gaeste": []})
daten = client.get("/api/homelab/zertifikate").json()
assert daten["gelesen"] is True and [z["quelle"] for z in daten["zertifikate"]] == ["npm", "pve"]
finally:
einstellungen_mod.einstellungen.cache_clear()