Welle 2 · Sicherheits-Markierung: Debian-Sicherheitsupdates, Alpine über secdb, Lücken-Scans aus Arcane
Der Ausführer zählt je Gast, wie viele offene Updates aus einer Sicherheits-Suite kommen („-security“ in apt list), beim Proxmox-Host ebenso; für Alpine (NPMplus) schickt er die Paketliste, und der Homelab-Teil gleicht sie mit Alpines Sicherheitsdatenbank ab (im Hintergrund geladen, zwölf Stunden gemerkt, Alpine-Versionsvergleich). Die Update-Tabelle, die Kacheln und die Übersicht zeigen „· 3 Sicherheit“, die Rückfrage nennt es. Je Docker-Container steht, was Arcanes Lücken-Scan (Trivy, sonntags 02:00) im Image fand, kritische und hohe. Der Ausführer braucht ausfuehrer-einrichten.sh. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 5.5
parent
995a0f34ee
commit
1c66daad0b
@@ -219,6 +219,32 @@ def _container_liste(url: str) -> list[dict] | None:
|
||||
return ergebnis
|
||||
|
||||
|
||||
def luecken(url: str, image_ids: tuple[str, ...]) -> dict[str, dict]:
|
||||
"""Arcanes Lücken-Scan (Trivy) je Image: {imageId: {"kritisch", "hoch", "mittel", "gesamt", "zeit"}} — nur für
|
||||
gescannte Images. Leer ohne Schlüssel, ohne Scan oder wenn Arcane nicht antwortet (seit 25.09.2026)."""
|
||||
if not schluessel() or not image_ids:
|
||||
return {}
|
||||
return _gemerkt(f"luecken:{url}:{','.join(image_ids)}", lambda: _luecken(url, list(image_ids))) or {}
|
||||
|
||||
|
||||
def _luecken(url: str, image_ids: list[str]) -> dict[str, dict]:
|
||||
ergebnis: dict[str, dict] = {}
|
||||
try:
|
||||
for umgebung in umgebungen(url):
|
||||
r = httpx.post(f"{url}/api/environments/{umgebung['id']}/images/vulnerabilities/summaries",
|
||||
headers={"X-API-Key": schluessel()}, json={"imageIds": image_ids}, timeout=ZEITLIMIT)
|
||||
r.raise_for_status()
|
||||
for image_id, s in (((r.json() or {}).get("data") or {}).get("summaries") or {}).items():
|
||||
zahlen = (s or {}).get("summary") or {}
|
||||
if s and s.get("scanTime") and zahlen:
|
||||
ergebnis[image_id] = {"kritisch": int(zahlen.get("critical") or 0), "hoch": int(zahlen.get("high") or 0),
|
||||
"mittel": int(zahlen.get("medium") or 0), "gesamt": int(zahlen.get("total") or 0),
|
||||
"zeit": s.get("scanTime")}
|
||||
except (httpx.HTTPError, ValueError, TypeError, AttributeError):
|
||||
return {}
|
||||
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."""
|
||||
|
||||
@@ -7,6 +7,8 @@ Stack (Compose-Projekt) Zustand, Gesundheit, Image und Update-Stand:
|
||||
• „lokal gebaut“, wenn das Image aus keiner Registry kommt (Arcane: updateType „local“, oder die Registry kennt es
|
||||
nicht — NerdQuiz und Rippy baut der User selbst) — dort gibt es keinen Abgleich,
|
||||
• sonst „aktuell“.
|
||||
Seit 25.09.2026 (Punkt 11) steht je Container auch, was Arcanes Lücken-Scan (Trivy, sonntags 02:00) im Image gefunden
|
||||
hat — kritische und hohe zuerst; ungescannte Images haben keine Angabe.
|
||||
Der Wächter meldet gelb, wenn ein Container ungesund ist (Healthcheck) oder immer wieder neu startet; beendete
|
||||
Container (etwa einmalige Aufgaben) sind kein Befund. Neue Stacks erscheinen unter „Neu entdeckt“ (entdecken.py).
|
||||
"""
|
||||
@@ -37,7 +39,8 @@ def container_zeile(c: dict) -> dict:
|
||||
else:
|
||||
update = "aktuell"
|
||||
name = str((c.get("names") or ["?"])[0]).lstrip("/")
|
||||
return {"id": str(c.get("id") or name)[:12], "name": name, "stack": labels.get("com.docker.compose.project"),
|
||||
return {"id": str(c.get("id") or name)[:12], "image_id": c.get("imageId"), "luecken": None,
|
||||
"name": name, "stack": labels.get("com.docker.compose.project"),
|
||||
"dienst": labels.get("com.docker.compose.service"), "image": c.get("image"),
|
||||
"zustand": str(c.get("state") or "unbekannt"), "status": c.get("status"),
|
||||
"gesund": {"healthy": True, "unhealthy": False}.get(gesund.group(1)) if gesund else None,
|
||||
@@ -55,8 +58,10 @@ def lage() -> dict:
|
||||
if roh is None:
|
||||
return {"verfuegbar": False, "grund": "Arcane antwortet gerade nicht.", "stacks": []}
|
||||
stacks: dict[str, list[dict]] = {}
|
||||
for c in roh:
|
||||
zeile = container_zeile(c)
|
||||
zeilen = [container_zeile(c) for c in roh]
|
||||
befunde = arcane.luecken(url, tuple(sorted({z["image_id"] for z in zeilen if z["image_id"]})))
|
||||
for zeile in zeilen:
|
||||
zeile["luecken"] = befunde.get(zeile["image_id"] or "")
|
||||
stacks.setdefault(zeile["stack"] or zeile["name"], []).append(zeile)
|
||||
return {"verfuegbar": True, "grund": None,
|
||||
"stacks": [{"name": n, "container": sorted(cs, key=lambda z: z["name"])} for n, cs in sorted(stacks.items())]}
|
||||
|
||||
@@ -23,7 +23,7 @@ from kern.github import neueste_version
|
||||
from kern.zeit import LOCAL_TZ
|
||||
from kern.ziele import Aktion, BausteinStand, ZielStand
|
||||
|
||||
from services.homelab import apps, arcane, kanal, karenz
|
||||
from services.homelab import apps, arcane, kanal, karenz, sicherheit
|
||||
|
||||
BERICHT_ALT_S = 30 * 60
|
||||
LISTEN_ALT_S = 14 * 24 * 3600
|
||||
@@ -168,9 +168,13 @@ def _os_baustein(gast: dict, zid: str, jetzt: float) -> BausteinStand:
|
||||
stand.zustand, stand.aktion = "unbekannt", suchen
|
||||
stand.grund = f"Die Paketlisten sind vom {_datum(listen)}; was fehlt, weiß erst eine neue Suche."
|
||||
elif anzahl > 0:
|
||||
# Seit 25.09.2026: wie viele davon eine Sicherheitslücke schließen (Debian: Suite „-security“, Alpine: secdb).
|
||||
stand.sicherheit = os_stand.get("sicherheit") if "sicherheit" in os_stand else sicherheit.alpine_sicherheit(os_stand)
|
||||
davon = f" (davon {stand.sicherheit} Sicherheitsupdate{'s' if stand.sicherheit != 1 else ''})" \
|
||||
if stand.sicherheit else ""
|
||||
stand.zustand, stand.kurz = "neu", f"{anzahl} Pakete"
|
||||
stand.aktion = Aktion("POST", f"/api/homelab/ziele/{zid}/os-update",
|
||||
f"{anzahl} Pakete im Gast jetzt einspielen? {_rueckweg_frage(gast)}")
|
||||
f"{anzahl} Pakete{davon} im Gast jetzt einspielen? {_rueckweg_frage(gast)}")
|
||||
return stand
|
||||
|
||||
|
||||
@@ -269,6 +273,8 @@ def _host_ziel(host: dict) -> ZielStand:
|
||||
pakete.zustand, pakete.grund = "unbekannt", host.get("fehler") or "Die Update-Liste fehlt im Bericht."
|
||||
elif updates:
|
||||
pakete.zustand, pakete.kurz = "neu", f"{len(updates)} Pakete"
|
||||
if isinstance(host.get("updates_sicherheit"), list):
|
||||
pakete.sicherheit = len(host["updates_sicherheit"])
|
||||
pakete.aktion = Aktion("POST", "/api/homelab/ziele/pve/update",
|
||||
f"{len(updates)} Pakete auf dem Proxmox-Host einspielen? Alle Gäste laufen weiter. "
|
||||
"Einen Neustart löst das nicht aus — der ist ein eigener Knopf.")
|
||||
|
||||
@@ -0,0 +1,120 @@
|
||||
"""Sicherheits-Markierung (Ausbauplan Welle 2, Punkt 11, seit 25.09.2026).
|
||||
|
||||
Welche offenen Paket-Updates schließen eine Sicherheitslücke?
|
||||
• Debian und Ubuntu (Gäste und Proxmox-Host) liefern sie aus eigenen Suiten („trixie-security“); das zählt der
|
||||
Ausführer selbst (os_updates.sicherheit, host.updates_sicherheit).
|
||||
• Alpine (NPMplus) kennt keine solche Suite. Dafür führt Alpine eine Sicherheitsdatenbank (secdb.alpinelinux.org,
|
||||
je Zweig main und community): je Paket die Versionen, die Lücken schließen. Ein Update zählt, wenn eine davon
|
||||
nach der installierten und höchstens bei der neuen Version liegt. Die Datenbank lädt dieser Teil im Hintergrund
|
||||
und merkt sie zwölf Stunden (<Datenordner>/alpine-secdb-v<zweig>.json); bis sie da ist, heißt es „unbekannt“.
|
||||
Docker-Images prüft Arcane selbst (Trivy, sonntags 02:00): docker.py holt dessen Befunde je Image.
|
||||
"""
|
||||
|
||||
import json
|
||||
import logging
|
||||
import re
|
||||
import threading
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
import httpx
|
||||
from kern.einstellungen import einstellungen
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
SECDB = "https://secdb.alpinelinux.org/v{zweig}/{repo}.json"
|
||||
NEU_LADEN_S = 12 * 3600
|
||||
_SUFFIX = {"alpha": -4, "beta": -3, "pre": -2, "rc": -1, "cvs": 1, "svn": 1, "git": 1, "hg": 1, "p": 2}
|
||||
_lock = threading.Lock()
|
||||
_laedt: set[str] = set()
|
||||
_gemerkt: dict[str, tuple[float, dict]] = {}
|
||||
|
||||
|
||||
def apk_teile(version: str) -> tuple:
|
||||
"""Eine Alpine-Version („1.37.0-r18“, „3.3.2_rc1-r0“, „1.2a-r1“) als vergleichbares Tupel: Zahlen (Buchstaben
|
||||
als kleine Zahlen), Zusätze (_alpha < _beta < _pre < _rc < ohne < _p), zuletzt die Paketrevision -rN."""
|
||||
haupt, _, revision = version.partition("-r")
|
||||
kern, *zusaetze = haupt.split("_")
|
||||
teile = tuple(int(x) if x.isdigit() else ord(x[0]) - 96 for x in re.findall(r"\d+|[a-z]+", kern.lower()))
|
||||
suffixe = []
|
||||
for z in zusaetze:
|
||||
m = re.fullmatch(r"([a-z]+)(\d*)", z)
|
||||
suffixe.append((_SUFFIX.get(m.group(1), 0), int(m.group(2) or 0)) if m else (0, 0))
|
||||
return (teile, tuple(suffixe) or ((0, 0),), int(revision) if revision.isdigit() else 0)
|
||||
|
||||
|
||||
def _pfad(zweig: str) -> Path:
|
||||
return einstellungen().daten_dir / f"alpine-secdb-v{zweig}.json"
|
||||
|
||||
|
||||
def _laden(zweig: str) -> None:
|
||||
"""Beide Repos des Zweigs holen und als {paket: [Versionen mit Fixes]} ablegen (eigener Faden)."""
|
||||
try:
|
||||
fixes: dict[str, list[str]] = {}
|
||||
for repo in ("main", "community"):
|
||||
r = httpx.get(SECDB.format(zweig=zweig, repo=repo), timeout=30, follow_redirects=True)
|
||||
r.raise_for_status()
|
||||
for eintrag in r.json().get("packages") or []:
|
||||
pkg = (eintrag or {}).get("pkg") or {}
|
||||
if pkg.get("name") and isinstance(pkg.get("secfixes"), dict):
|
||||
fixes.setdefault(pkg["name"], []).extend(v for v in pkg["secfixes"] if v != "0")
|
||||
pfad = _pfad(zweig)
|
||||
pfad.parent.mkdir(parents=True, exist_ok=True)
|
||||
tmp = pfad.with_suffix(".tmp")
|
||||
tmp.write_text(json.dumps({"geholt": time.time(), "fixes": fixes}), encoding="utf-8")
|
||||
tmp.replace(pfad)
|
||||
log.info("sicherheit: Alpine-Sicherheitsdatenbank v%s geladen (%d Pakete)", zweig, len(fixes))
|
||||
except Exception:
|
||||
log.warning("sicherheit: Alpine-Sicherheitsdatenbank v%s nicht geladen", zweig, exc_info=True)
|
||||
finally:
|
||||
with _lock:
|
||||
_laedt.discard(zweig)
|
||||
|
||||
|
||||
def secdb(zweig: str, jetzt: float | None = None) -> dict | None:
|
||||
"""{paket: [Versionen mit Fixes]} — aus der Datei, im Speicher gemerkt; ist sie alt oder fehlt, lädt ein Faden
|
||||
sie nach (bis dahin die alte oder None). Blockiert nie."""
|
||||
jetzt = time.time() if jetzt is None else jetzt
|
||||
pfad = _pfad(zweig)
|
||||
daten = None
|
||||
try:
|
||||
mtime = pfad.stat().st_mtime
|
||||
with _lock:
|
||||
alt = _gemerkt.get(zweig)
|
||||
if alt and alt[0] == mtime:
|
||||
daten = alt[1]
|
||||
else:
|
||||
daten = json.loads(pfad.read_text(encoding="utf-8"))
|
||||
with _lock:
|
||||
_gemerkt[zweig] = (mtime, daten)
|
||||
except (OSError, ValueError):
|
||||
daten = None
|
||||
if daten is None or jetzt - float(daten.get("geholt") or 0) > NEU_LADEN_S:
|
||||
with _lock:
|
||||
starten = zweig not in _laedt
|
||||
_laedt.add(zweig)
|
||||
if starten:
|
||||
threading.Thread(target=_laden, args=(zweig,), name=f"secdb-{zweig}", daemon=True).start()
|
||||
return daten.get("fixes") if daten else None
|
||||
|
||||
|
||||
def alpine_sicherheit(os_stand: dict, fixes: dict | None = None) -> int | None:
|
||||
"""Wie viele der offenen Alpine-Updates eine Lücke schließen (None, solange die Datenbank fehlt)."""
|
||||
pakete = os_stand.get("pakete")
|
||||
if not isinstance(pakete, list):
|
||||
return None
|
||||
if not pakete:
|
||||
return 0
|
||||
if fixes is None:
|
||||
zweig = ".".join(str(os_stand.get("alpine") or "").split(".")[:2])
|
||||
if not re.fullmatch(r"\d+\.\d+", zweig):
|
||||
return None
|
||||
fixes = secdb(zweig)
|
||||
if fixes is None:
|
||||
return None
|
||||
anzahl = 0
|
||||
for p in pakete:
|
||||
alt, neu = apk_teile(str(p.get("alt") or "")), apk_teile(str(p.get("neu") or ""))
|
||||
if any(alt < apk_teile(v) <= neu for v in fixes.get(str(p.get("name")), [])):
|
||||
anzahl += 1
|
||||
return anzahl
|
||||
Reference in New Issue
Block a user