threadnet-operating/monitoring/cve/cve-exporter.py

#!/usr/bin/env python3
# Trivy-JSON -> Prometheus-Metriken (gitops#47). Stdlib-only, Machart wie die
# uebrigen Monitoring-Helfer. Liest die Reports des cve-scan-Sidecars und
# serviert /metrics; merkt sich je (CVE, Target) den Erstfund (first_seen),
# damit der geforderte Zeitstrahl (erstmals gesehen / geschlossen) abbildbar
# ist - "geschlossen" = Serie verschwindet, resolved kommt via Alertmanager.
#
# Schema (Pflichtfelder-Vorgabe sorb: CVE-ID, Mitigation, Zeitstrahl, Ort, Typ):
#   trivy_vuln_info{cve,severity,target,target_type,host,pkg,installed,fixed_version} 1
#       (nur HIGH/CRITICAL als Einzelserien - Kardinalitaet)
#   trivy_vuln_count{target,target_type,host,severity} <n>   (alle Severities)
#   trivy_vuln_first_seen_timestamp{cve,target} <unix>
#   trivy_last_scan_timestamp{target,target_type,host} <unix>
import json
import os
import threading
import time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

import targets as ziel_herleitung

RESULTS = os.environ.get("RESULTS_DIR", "/results")
STATE_FILE = os.environ.get("STATE_FILE", "/state/first-seen.json")
# Soll-Menge (#0106): hierhin schreibt der Exporter, von hier liest scan-loop.sh.
TARGETS_FILE = os.environ.get("TARGETS_FILE", "/targets/targets.txt")
# ⚠️ Die Herleitung laeuft NICHT je Scrape. Bei 15 s Scrape-Intervall waeren das
# rund 21 000 Registry-Anfragen am Tag; eine Runde kostet ~90.
SOLL_INTERVALL = float(os.environ.get("SOLL_INTERVALL_SEKUNDEN", "3600"))
PROZESSSTART = time.time()
# Ort: Container-Images laufen (bis auf Weiteres) alle auf dem MATRIX-Host;
# Host-rootfs-Scans (Ausbaustufe) bringen ihren Hostnamen im Dateinamen mit.
DEFAULT_HOST = os.environ.get("DEFAULT_HOST", "matrix")
# Entschiedene Befunde (#0051): CRITICAL, zu dem es nichts mehr zu tun gibt oder
# zu dem die Arbeit noch aussteht - mit Begruendung und Pruefdatum.
ENTSCHEIDUNGEN_FILE = os.environ.get("ENTSCHEIDUNGEN_FILE", "/app/entscheidungen.json")

try:
    with open(STATE_FILE) as f:
        first_seen = json.load(f)  # "cve|target" -> unix-ts
except Exception:
    first_seen = {}


def esc(v):
    return str(v).replace("\\", "\\\\").replace('"', '\\"').replace("\n", " ")


def lade_entscheidungen(pfad=None):
    """Entscheidungen -> {(ziel, cve): (art, ablauf_unix)} plus Fehlertext.

    ⚠️ Bewusst KEINE Unterdrueckung. Trivy koennte die Befunde ueber eine
    Ignore-Datei einfach weglassen - dann saehe "0 CRITICAL, weil behoben"
    genauso aus wie "0 CRITICAL, weil weggeschaut". Hier bleibt jeder Befund in
    trivy_vuln_info stehen; entschieden wird er nur zusaetzlich gezaehlt.

    ⚠️ Ein Lesefehler gibt eine LEERE Menge zurueck und meldet ihn. Damit zaehlt
    im Zweifel alles als offen - die Richtung, die auffaellt. Umgekehrt saehe
    eine kaputte Datei aus wie "alles entschieden".
    """
    pfad = pfad or ENTSCHEIDUNGEN_FILE
    getroffen = {}
    try:
        with open(pfad) as fh:
            doc = json.load(fh)
    except FileNotFoundError:
        return getroffen, "keine Datei"
    except Exception as e:
        return getroffen, f"unlesbar: {e}"
    for e in doc.get("eintraege") or []:
        ziel = ziel_herleitung.normalisiere(e.get("ziel", ""))
        art = e.get("art", "?")
        datum = e.get("pruefen_am", "")
        try:
            ablauf = time.mktime(time.strptime(datum, "%Y-%m-%d"))
        except Exception:
            return {}, f"Pruefdatum unbrauchbar bei {e.get('ziel')}: {datum!r}"
        if not ziel or not e.get("cves"):
            return {}, f"Eintrag ohne Ziel oder ohne CVE-Kennung: {e.get('ziel')!r}"
        for cve in e["cves"]:
            getroffen[(ziel, cve)] = (art, ablauf)
    return getroffen, None


class ZielCache:
    """Haelt die zuletzt gebildete Soll-Menge und erneuert sie im eigenen Takt.

    ⚠️ Der Stand je Quelle ueberlebt einen Ausfall absichtlich: Faellt eine
    Quelle aus, bleibt ihr alter Zeitstempel stehen und ALTERT, statt zu
    verschwinden. Eine verschwundene Serie kann keine Alarmregel ausloesen
    (AAR 2026-08-01, Befund 3) - genau daran waere die Absicherung sonst
    gescheitert.
    """

    def __init__(self):
        self._sperre = threading.Lock()
        self._stand = {}
        self._letzte = None
        self._geholt_um = 0.0

    def hole(self):
        with self._sperre:
            if self._letzte is not None and time.time() - self._geholt_um < SOLL_INTERVALL:
                return self._letzte
            cfg = ziel_herleitung.cfg_aus_umgebung()
            h = ziel_herleitung.herleiten(
                cfg, vorheriger_stand=self._stand,
                vorherige_ziele=(self._letzte.je_quelle if self._letzte else None))
            self._stand = dict(h.stand)
            self._letzte, self._geholt_um = h, time.time()
            for quelle, meldung in h.fehler.items():
                print(f"soll-menge: quelle {quelle} ausgefallen: {meldung}", flush=True)
            if h.brauchbar:
                schreibe_targets(h.ziele)
            else:
                # NICHTS ueberschreiben. Sonst raeumt ein Neustart waehrend eines
                # Prometheus-Ausfalls den gesamten Bestand ab.
                print("soll-menge: alle Quellen aus - targets.txt bleibt unberuehrt", flush=True)
            return h


def schreibe_targets(ziele):
    try:
        os.makedirs(os.path.dirname(TARGETS_FILE), exist_ok=True)
        tmp = TARGETS_FILE + ".tmp"
        with open(tmp, "w") as f:
            f.write("# erzeugt von cve-exporter (#0106) - NICHT von Hand pflegen\n")
            for z in sorted(ziele):
                f.write(z + "\n")
        os.replace(tmp, TARGETS_FILE)      # atomar, der Scanner liest nebenlaeufig
    except Exception as e:
        print(f"targets.txt nicht schreibbar: {e}", flush=True)


ziel_cache = ZielCache()


def collect():
    lines = []
    now = int(time.time())
    seen_keys = set()
    gelesene_targets = set()   # nur DEREN Erstfunde duerfen geprunt werden
    berichte = 0
    lesefehler = 0
    kritisch = set()          # (normalisiertes Ziel, CVE) - Grundlage fuer "offen"
    for fn in sorted(os.listdir(RESULTS)):
        if not fn.endswith(".json"):
            continue
        berichte += 1
        path = os.path.join(RESULTS, fn)
        try:
            rep = json.load(open(path))
        except Exception as e:
            # Frueher: stilles 'continue'. Ein einmaliger Lesefehler (Datei wird
            # gerade geschrieben, kurzer I/O-Fehler) liess die Findings dieses
            # Targets aus seen_keys verschwinden - und der Prune unten loeschte
            # ihre Erstfund-Zeitstempel DAUERHAFT. Der Zeitstrahl war damit weg,
            # ohne dass irgendetwas gemeldet haette. Jetzt zaehlbar und sichtbar.
            lesefehler += 1
            print(f"report unlesbar: {fn}: {e}", flush=True)
            continue
        target = rep.get("ArtifactName", fn[:-5])
        gelesene_targets.add(target)
        ttype = "host" if rep.get("ArtifactType") in ("filesystem", "rootfs") else "image"
        host = fn.split("__host__")[1].split(".json")[0] if "__host__" in fn else DEFAULT_HOST
        base = f'target="{esc(target)}",target_type="{ttype}",host="{esc(host)}"'
        lines.append(f'trivy_last_scan_timestamp{{{base}}} {int(os.path.getmtime(path))}')
        counts = {}
        for res in rep.get("Results") or []:
            for v in res.get("Vulnerabilities") or []:
                sev = v.get("Severity", "UNKNOWN")
                counts[sev] = counts.get(sev, 0) + 1
                if sev in ("HIGH", "CRITICAL"):
                    cve = v.get("VulnerabilityID", "?")
                    if sev == "CRITICAL":
                        kritisch.add((ziel_herleitung.normalisiere(target), cve))
                    key = f"{cve}|{target}"
                    if key not in first_seen:
                        first_seen[key] = now
                    seen_keys.add(key)
                    lines.append(
                        'trivy_vuln_info{cve="%s",severity="%s",%s,pkg="%s",installed="%s",fixed_version="%s"} 1'
                        % (esc(cve), sev, base, esc(v.get("PkgName", "?")),
                           esc(v.get("InstalledVersion", "?")), esc(v.get("FixedVersion", ""))))
                    lines.append(f'trivy_vuln_first_seen_timestamp{{cve="{esc(cve)}",target="{esc(target)}"}} {first_seen[key]}')
        for sev, n in sorted(counts.items()):
            lines.append(f'trivy_vuln_count{{{base},severity="{sev}"}} {n}')
    # State kompakt halten: verschwundene Findings raus (= "geschlossen").
    # ABER nur fuer Targets, deren Bericht in DIESEM Durchgang auch wirklich
    # gelesen wurde. Sonst loescht ein einzelner Lesefehler den Zeitstrahl eines
    # Targets, das es noch gibt - unwiederbringlich, weil "erstmals gesehen"
    # danach auf 'jetzt' neu anfaengt.
    for k in list(first_seen):
        ziel = k.split("|", 1)[1] if "|" in k else ""
        if ziel in gelesene_targets and k not in seen_keys:
            del first_seen[k]

    # Sichtbarkeit des Scanners selbst: ein still gestorbener Scanner macht blind,
    # und TrivyScanStale kann ein Target, das NIE einen Bericht hatte, nicht melden
    # (ohne Serie kein time()-Vergleich). Diese beiden Zahlen schliessen die Luecke
    # so weit, wie sie ohne Soll-Liste zu schliessen ist.
    lines.append(f"trivy_reports_total {berichte}")
    lines.append(f"trivy_report_read_errors {lesefehler}")

    # --- Deckung gegen die Soll-Menge (#0106) --------------------------------
    # Bis hierher konnte der Exporter nur sagen, WAS er gescannt hat. Was er nie
    # angesehen hat, war unsichtbar - am 2026-08-21 waren das 24 von 51 Images.
    # Erst der Vergleich gegen eine Soll-Menge macht die Luecke zur Zahl.
    h = ziel_cache.hole()
    ist = {ziel_herleitung.normalisiere(t) for t in gelesene_targets}
    ist.discard(None)
    soll = h.ziele

    for quelle in ziel_herleitung.QUELLEN:
        for z in sorted(h.je_quelle.get(quelle, ())):
            lines.append(f'cve_target_desired{{target="{esc(z)}",quelle="{quelle}"}} 1')
        # ⚠️ IMMER ausgeben, auch wenn die Quelle noch nie geliefert hat: Eine
        # fehlende Serie kann keine Regel ausloesen, und eine ausgefallene
        # Herleitung sieht sonst aus wie vollstaendige Deckung. Ohne je einen
        # Erfolg zaehlt die Zeit seit Prozessstart - der Wert waechst also und
        # schlaegt irgendwann an, statt still zu fehlen.
        seit = h.stand.get(quelle, PROZESSSTART)
        lines.append(f'cve_target_source_stale{{quelle="{quelle}"}} {int(now - seit)}')

    fehlend, verwaist = soll - ist, ist - soll
    lines.append(f"cve_targets_missing {len(fehlend)}")
    lines.append(f"cve_targets_orphaned {len(verwaist)}")
    lines.append(f"cve_targets_desired_total {len(soll)}")
    # Bei leerer Soll-Menge waere 0/0 rechnerisch 1,0 - also "alles gedeckt",
    # ausgerechnet im Ausfall. Deshalb 0 statt einer Division.
    lines.append(
        f"cve_target_coverage_ratio {len(soll & ist) / len(soll) if soll else 0}"
    )

    # --- Entschieden oder offen (#0051, Kriterien 1 und 5) -------------------
    # Kriterium 1 lautet "kein CRITICAL in einem laufenden Image OHNE
    # Entscheidung". Das ist erst eine Zahl, wenn beide Seiten zaehlbar sind:
    # die Befunde (oben) und die Entscheidungen (hier).
    getroffen, fehler = lade_entscheidungen()
    if fehler:
        print(f"entscheidungen: {fehler}", flush=True)
    # ⚠️ Nur Befunde auf Zielen, die WIRKLICH betrieben werden. Ein verwaister
    # Bericht (Image abgeloest, Runde noch nicht gelaufen) darf die Zahl nicht
    # hochhalten - sonst sieht ein erfolgreiches Update wie Stillstand aus.
    #
    # ⚠️ ABER: Ist die Soll-Menge leer, ist sie nicht "leer", sondern UNBEKANNT -
    # die Herleitung ist ausgefallen. Dann darf hier nicht 0 herauskommen, denn
    # 0 liest sich wie "nichts offen". In dem Fall zaehlen alle Berichte, auch
    # verwaiste: lieber zu viel melden als im Ausfall Entwarnung geben.
    laufend = {p for p in kritisch if p[0] in soll} if soll else set(kritisch)
    offen = laufend - set(getroffen)
    lines.append(f"cve_critical_offen {len(offen)}")
    lines.append(f"cve_critical_entschieden {len(laufend & set(getroffen))}")
    lines.append(f"cve_entscheidungen_lesefehler {1 if fehler else 0}")

    abgelaufen = 0
    ohne_befund = 0
    for (ziel, cve), (art, ablauf) in sorted(getroffen.items()):
        lines.append(
            f'cve_entscheidung_gueltig_bis{{target="{esc(ziel)}",cve="{esc(cve)}",art="{esc(art)}"}} {int(ablauf)}'
        )
        if now > ablauf:
            abgelaufen += 1
        # ⚠️ Eine Entscheidung ohne Befund ist Altpapier: das Image ist ersetzt
        # oder der Befund weg. Sie zaehlt hier, damit die Datei nicht zuwaechst
        # und irgendwann Befunde deckt, die niemand mehr geprueft hat.
        #
        # ⚠️ KEIN "und ziel in soll" mehr. Genau diese Bedingung stand hier bis
        # zum 2026-08-22 und machte den Zaehler blind fuer den haeufigsten Fall:
        # Ist das Image ganz aus dem Bestand gefallen (Fassungssprung), faellt
        # auch seine Entscheidung aus der Pruefung - und bleibt fuer immer
        # unbemerkt in der Datei stehen. Belegt am selben Tag: die Eintraege zu
        # grafana 12.0.0 und 12.4.9 ueberlebten den Sprung auf 13.2.0, und der
        # Zaehler meldete 0.
        #
        # Der Preis ist bekannt und tragbar: Wer eine Entscheidung trifft, BEVOR
        # der Scanner das Image erreicht, sieht sie bis zur naechsten Runde hier
        # auftauchen. Die Alarmregel wartet 24h und deckt das ab.
        if (ziel, cve) not in laufend:
            ohne_befund += 1
    lines.append(f"cve_entscheidungen_abgelaufen {abgelaufen}")
    lines.append(f"cve_entscheidungen_ohne_befund {ohne_befund}")
    lines.append(f"cve_entscheidungen_total {len(getroffen)}")
    try:
        os.makedirs(os.path.dirname(STATE_FILE), exist_ok=True)
        with open(STATE_FILE, "w") as f:
            json.dump(first_seen, f)
    except Exception as e:
        print(f"state save failed: {e}", flush=True)
    return "\n".join(lines) + "\n"


class Handler(BaseHTTPRequestHandler):
    def do_GET(self):
        if self.path != "/metrics":
            self.send_response(404); self.end_headers(); return
        body = collect().encode()
        self.send_response(200)
        self.send_header("Content-Type", "text/plain; version=0.0.4")
        self.end_headers()
        self.wfile.write(body)

    def log_message(self, fmt, *args):
        pass


# ⚠️ Unter __main__, damit test_entscheidungen.py die ECHTE Datei importieren
# kann statt eine zurechtgeschnittene Kopie. Ohne die Wache startet beim
# Import der Server, und der Test haette einen Nachbau geprueft - genau die
# Fehlerklasse, die scan-loop.sh am 2026-08-21 schon einmal gekostet hat.
if __name__ == "__main__":
    ThreadingHTTPServer(("0.0.0.0", 9101), Handler).serve_forever()