Thore Cimbal
d68991629e
I wrote 'and ziel in soll' into that condition this evening, and it means a decision only counts as stale while its image is still deployed. The moment an image leaves the inventory — a version bump, which is exactly when decisions go stale — the entry stops being checked and sits in the file forever. Grafana 12.0.0 and 12.4.9 proved it hours later: both entries survived the jump to 13.2.0 and the counter reported zero. The cost of removing the guard is that a decision made before the scanner reaches its image shows up here until the next round. The alert waits 24 hours, which covers it. The existing arithmetic test could not have caught this, because it reimplements the calculation and the faulty condition was never in the copy. New tests drive the real collect() instead, including one that the old behaviour fails. Both dead grafana entries removed: 28 decisions down to 26.
237 lines
11 KiB
Python
237 lines
11 KiB
Python
#!/usr/bin/env python3
|
|
# Zusicherungen fuer die Entscheidungen zu CRITICAL-Befunden (#0051,
|
|
# Abnahmekriterien 1 und 5). Stdlib, laeuft ohne Netz:
|
|
# python3 test_entscheidungen.py
|
|
#
|
|
# ⚠️ Geprueft wird die ECHTE cve-exporter.py, nicht eine Kopie ihrer Logik. Der
|
|
# Modulname traegt einen Bindestrich, deshalb der Umweg ueber importlib; der
|
|
# Serverstart steht dort unter __main__, damit dieser Import moeglich ist. Am
|
|
# 2026-08-21 hat ein nachgebauter Test fuer scan-loop.sh eine Sabotage gruen
|
|
# durchgelassen - dieselbe Falle soll hier nicht noch einmal stehen.
|
|
#
|
|
# ⚠️ Jede Zusicherung hat ihre GEGENPROBE: neben jedem "muss zaehlen" steht ein
|
|
# "darf NICHT zaehlen".
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
import tempfile
|
|
import unittest
|
|
|
|
os.environ.setdefault("RESULTS_DIR", tempfile.mkdtemp())
|
|
os.environ.setdefault("STATE_FILE", os.path.join(tempfile.mkdtemp(), "s.json"))
|
|
|
|
_spec = importlib.util.spec_from_file_location(
|
|
"cve_exporter", os.path.join(os.path.dirname(__file__) or ".", "cve-exporter.py")
|
|
)
|
|
exporter = importlib.util.module_from_spec(_spec)
|
|
_spec.loader.exec_module(exporter)
|
|
|
|
ECHT = os.path.join(os.path.dirname(__file__) or ".", "entscheidungen.json")
|
|
|
|
|
|
def schreibe(doc):
|
|
fd, pfad = tempfile.mkstemp(suffix=".json")
|
|
with os.fdopen(fd, "w") as f:
|
|
json.dump(doc, f)
|
|
return pfad
|
|
|
|
|
|
GUELTIG = {
|
|
"eintraege": [
|
|
{"ziel": "coturn/coturn:4.17.2", "art": "hingenommen",
|
|
"grund": "neueste Fassung", "pruefen_am": "2026-11-21",
|
|
"cves": ["CVE-1", "CVE-2"]},
|
|
]
|
|
}
|
|
|
|
|
|
class LaderTest(unittest.TestCase):
|
|
def test_gueltige_datei_wird_zu_paaren(self):
|
|
getroffen, fehler = exporter.lade_entscheidungen(schreibe(GUELTIG))
|
|
self.assertIsNone(fehler)
|
|
self.assertEqual(set(getroffen), {("coturn/coturn:4.17.2", "CVE-1"),
|
|
("coturn/coturn:4.17.2", "CVE-2")})
|
|
self.assertEqual(getroffen[("coturn/coturn:4.17.2", "CVE-1")][0], "hingenommen")
|
|
|
|
def test_fehlende_datei_meldet_sich_und_entscheidet_nichts(self):
|
|
getroffen, fehler = exporter.lade_entscheidungen("/gibt/es/nicht.json")
|
|
self.assertEqual(getroffen, {})
|
|
self.assertEqual(fehler, "keine Datei")
|
|
|
|
def test_kaputte_datei_entscheidet_NICHTS_statt_alles(self):
|
|
# ⚠️ Die Richtung ist der Punkt: Eine unlesbare Datei muss dazu fuehren,
|
|
# dass alles als OFFEN zaehlt. Umgekehrt saehe ein Tippfehler in der
|
|
# Datei aus wie "alles entschieden" - und niemand merkte es.
|
|
fd, pfad = tempfile.mkstemp(suffix=".json")
|
|
with os.fdopen(fd, "w") as f:
|
|
f.write("{kein json")
|
|
getroffen, fehler = exporter.lade_entscheidungen(pfad)
|
|
self.assertEqual(getroffen, {})
|
|
self.assertIn("unlesbar", fehler)
|
|
# Gegenprobe: dieselbe Datei mit gueltigem Inhalt entscheidet sehr wohl.
|
|
self.assertEqual(len(exporter.lade_entscheidungen(schreibe(GUELTIG))[0]), 2)
|
|
|
|
def test_unbrauchbares_pruefdatum_verwirft_die_GANZE_datei(self):
|
|
doc = json.loads(json.dumps(GUELTIG))
|
|
doc["eintraege"][0]["pruefen_am"] = "irgendwann"
|
|
getroffen, fehler = exporter.lade_entscheidungen(schreibe(doc))
|
|
self.assertEqual(getroffen, {})
|
|
self.assertIn("Pruefdatum", fehler)
|
|
|
|
def test_eintrag_ohne_cve_kennungen_ist_ein_fehler(self):
|
|
# Eine Pauschale je Image ("alles an diesem Ziel") wuerde auch jeden
|
|
# NEUEN Befund verschlucken. Deshalb ist sie nicht ausdrueckbar.
|
|
doc = json.loads(json.dumps(GUELTIG))
|
|
doc["eintraege"][0]["cves"] = []
|
|
getroffen, fehler = exporter.lade_entscheidungen(schreibe(doc))
|
|
self.assertEqual(getroffen, {})
|
|
self.assertIn("CVE-Kennung", fehler)
|
|
|
|
def test_ziel_wird_normalisiert_wie_die_soll_menge(self):
|
|
# Sonst passt eine Entscheidung fuer "postgres:15-alpine" nicht auf
|
|
# einen Bericht, dessen ArtifactName "docker.io/library/postgres:..."
|
|
# lautet - und die Entscheidung liefe ins Leere, ohne aufzufallen.
|
|
doc = {"eintraege": [dict(GUELTIG["eintraege"][0],
|
|
ziel="docker.io/library/postgres:15-alpine")]}
|
|
getroffen, _ = exporter.lade_entscheidungen(schreibe(doc))
|
|
self.assertIn(("postgres:15-alpine", "CVE-1"), getroffen)
|
|
|
|
|
|
class RechnungTest(unittest.TestCase):
|
|
"""Die Zahl aus Kriterium 1: CRITICAL auf laufenden Zielen ohne Entscheidung."""
|
|
|
|
def rechne(self, kritisch, soll, getroffen):
|
|
laufend = {p for p in kritisch if p[0] in soll} if soll else set(kritisch)
|
|
return len(laufend - set(getroffen)), len(laufend & set(getroffen))
|
|
|
|
def test_entschiedenes_ist_nicht_offen_unentschiedenes_schon(self):
|
|
kritisch = {("a:1", "CVE-1"), ("a:1", "CVE-2")}
|
|
offen, entschieden = self.rechne(kritisch, {"a:1"}, {("a:1", "CVE-1"): ("x", 0)})
|
|
self.assertEqual((offen, entschieden), (1, 1))
|
|
|
|
def test_verwaister_bericht_haelt_die_zahl_NICHT_hoch(self):
|
|
# Nach einem Update liegt der alte Bericht noch da, bis die naechste
|
|
# Scan-Runde ihn entfernt. Zaehlte er mit, saehe ein erfolgreiches
|
|
# Update wie Stillstand aus.
|
|
kritisch = {("alt:1", "CVE-9"), ("neu:1", "CVE-1")}
|
|
offen, _ = self.rechne(kritisch, {"neu:1"}, {})
|
|
self.assertEqual(offen, 1)
|
|
# Gegenprobe: gehoert das alte Ziel noch zum Soll, zaehlt es sehr wohl.
|
|
offen_beide, _ = self.rechne(kritisch, {"neu:1", "alt:1"}, {})
|
|
self.assertEqual(offen_beide, 2)
|
|
|
|
def test_ausgefallene_soll_menge_gibt_KEINE_entwarnung(self):
|
|
# ⚠️ Beim Bauen zuerst falsch herum: eine leere Soll-Menge liess
|
|
# cve_critical_offen auf 0 fallen - waehrend die Herleitung ausgefallen
|
|
# war, also genau dann, wenn man sich am wenigsten darauf verlassen darf.
|
|
kritisch = {("a:1", "CVE-1"), ("b:1", "CVE-2")}
|
|
offen, _ = self.rechne(kritisch, set(), {})
|
|
self.assertEqual(offen, 2)
|
|
# Gegenprobe: mit bekannter Soll-Menge wird sehr wohl gefiltert.
|
|
offen_gefiltert, _ = self.rechne(kritisch, {"a:1"}, {})
|
|
self.assertEqual(offen_gefiltert, 1)
|
|
|
|
def test_entscheidung_fuer_ein_anderes_ziel_deckt_nicht(self):
|
|
# Derselbe CVE in einem anderen Image bleibt offen - genau der Grund,
|
|
# warum die Entscheidung am Paar (Ziel, CVE) haengt und nicht am CVE.
|
|
offen, _ = self.rechne({("b:1", "CVE-1")}, {"b:1"}, {("a:1", "CVE-1"): ("x", 0)})
|
|
self.assertEqual(offen, 1)
|
|
|
|
|
|
class CollectTest(unittest.TestCase):
|
|
"""⚠️ Diese Klasse prueft die ECHTE collect(), nicht die nachgebaute Rechnung
|
|
in RechnungTest. Der Unterschied ist an diesem Projekt schon einmal teuer
|
|
geworden: Am 2026-08-22 stand in collect() ein `and ziel in soll`, das den
|
|
Zaehler fuer verwaiste Entscheidungen blind machte, sobald ein Image ganz aus
|
|
dem Bestand fiel - und RechnungTest konnte das nicht sehen, weil dort die
|
|
Bedingung gar nicht stand.
|
|
"""
|
|
|
|
def laufe(self, berichte, entscheidungen, soll):
|
|
res = tempfile.mkdtemp()
|
|
for i, (ziel, cves) in enumerate(berichte):
|
|
json.dump({"ArtifactName": ziel, "ArtifactType": "container_image",
|
|
"Results": [{"Vulnerabilities": [
|
|
{"VulnerabilityID": c, "Severity": "CRITICAL", "PkgName": "p",
|
|
"InstalledVersion": "1", "FixedVersion": ""} for c in cves]}]},
|
|
open(os.path.join(res, f"b{i}.json"), "w"))
|
|
pfad = schreibe({"eintraege": entscheidungen})
|
|
alt_res, alt_ent = exporter.RESULTS, exporter.ENTSCHEIDUNGEN_FILE
|
|
exporter.RESULTS, exporter.ENTSCHEIDUNGEN_FILE = res, pfad
|
|
exporter.ziel_cache._letzte = exporter.ziel_herleitung.Herleitung(
|
|
ziele=set(soll), je_quelle={"cluster": set(soll)}, fehler={},
|
|
stand={"cluster": exporter.time.time()})
|
|
exporter.ziel_cache._geholt_um = exporter.time.time()
|
|
try:
|
|
zeilen = exporter.collect().splitlines()
|
|
finally:
|
|
exporter.RESULTS, exporter.ENTSCHEIDUNGEN_FILE = alt_res, alt_ent
|
|
self.zeilen = zeilen
|
|
return {z.split(" ")[0]: z.split(" ")[1] for z in zeilen if " " in z and "{" not in z}
|
|
|
|
E = [{"ziel": "a:1", "art": "hingenommen", "grund": "x" * 70,
|
|
"pruefen_am": "2099-01-01", "cves": ["CVE-1"]}]
|
|
|
|
def test_entschieden_und_offen_werden_getrennt_gezaehlt(self):
|
|
m = self.laufe([("a:1", ["CVE-1", "CVE-2"])], self.E, ["a:1"])
|
|
self.assertEqual(m["cve_critical_entschieden"], "1")
|
|
self.assertEqual(m["cve_critical_offen"], "1")
|
|
self.assertEqual(m["cve_entscheidungen_ohne_befund"], "0")
|
|
|
|
def test_entscheidung_zu_einem_VERSCHWUNDENEN_image_faellt_auf(self):
|
|
# Der Fall vom 2026-08-22: das Image ist ersetzt, das Ziel steht nicht
|
|
# mehr im Soll, der Bericht ist weg. Die Entscheidung ist Altpapier und
|
|
# muss sich melden - vorher meldete sie sich NICHT.
|
|
m = self.laufe([("b:1", [])], self.E, ["b:1"])
|
|
self.assertEqual(m["cve_entscheidungen_ohne_befund"], "1")
|
|
|
|
def test_gegenprobe_eine_gebrauchte_entscheidung_meldet_sich_nicht(self):
|
|
m = self.laufe([("a:1", ["CVE-1"])], self.E, ["a:1"])
|
|
self.assertEqual(m["cve_entscheidungen_ohne_befund"], "0")
|
|
self.assertEqual(m["cve_critical_offen"], "0")
|
|
|
|
def test_der_entschiedene_befund_bleibt_SICHTBAR(self):
|
|
# Der Kern der Entscheidung gegen eine Trivy-Ignore-Datei: entschieden
|
|
# heisst gezaehlt, nicht versteckt. Waere er unterdrueckt, fehlte diese
|
|
# Serie - und "0, weil behoben" saehe aus wie "0, weil weggeschaut".
|
|
m = self.laufe([("a:1", ["CVE-1"])], self.E, ["a:1"])
|
|
self.assertEqual(m["cve_critical_entschieden"], "1")
|
|
treffer = [z for z in self.zeilen if z.startswith("trivy_vuln_info{") and "CVE-1" in z]
|
|
self.assertEqual(len(treffer), 1, "der entschiedene Befund fehlt in trivy_vuln_info")
|
|
# Gegenprobe: ein Ziel ohne Befund erzeugt auch keine Serie.
|
|
self.laufe([("a:1", [])], self.E, ["a:1"])
|
|
self.assertFalse([z for z in self.zeilen if z.startswith("trivy_vuln_info{")])
|
|
|
|
|
|
class EchteDateiTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.getroffen, self.fehler = exporter.lade_entscheidungen(ECHT)
|
|
with open(ECHT) as f:
|
|
self.doc = json.load(f)
|
|
|
|
def test_sie_laedt_ueberhaupt(self):
|
|
self.assertIsNone(self.fehler)
|
|
self.assertGreater(len(self.getroffen), 0)
|
|
|
|
def test_jeder_eintrag_traegt_eine_begruendung_die_etwas_sagt(self):
|
|
# Kriterium 1 verlangt eine Begruendung. "siehe oben" waere keine.
|
|
for e in self.doc["eintraege"]:
|
|
self.assertGreaterEqual(len(e.get("grund", "")), 60, e["ziel"])
|
|
|
|
def test_jeder_eintrag_traegt_ein_pruefdatum(self):
|
|
# Kriterium 5: kein Suppress ohne Ablauf.
|
|
for e in self.doc["eintraege"]:
|
|
self.assertRegex(e.get("pruefen_am", ""), r"^\d{4}-\d{2}-\d{2}$", e["ziel"])
|
|
|
|
def test_nur_bekannte_arten(self):
|
|
for e in self.doc["eintraege"]:
|
|
self.assertIn(e.get("art"), ("hingenommen", "geplant"), e["ziel"])
|
|
|
|
def test_keine_doppelten_paare(self):
|
|
paare = [(e["ziel"], c) for e in self.doc["eintraege"] for c in e["cves"]]
|
|
self.assertEqual(len(paare), len(set(paare)))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main(verbosity=2)
|