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 time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

RESULTS = os.environ.get("RESULTS_DIR", "/results")
STATE_FILE = os.environ.get("STATE_FILE", "/state/first-seen.json")
# 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")

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 collect():
    lines = []
    now = int(time.time())
    seen_keys = set()
    gelesene_targets = set()   # nur DEREN Erstfunde duerfen geprunt werden
    berichte = 0
    lesefehler = 0
    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", "?")
                    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}")
    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


ThreadingHTTPServer(("0.0.0.0", 9101), Handler).serve_forever()