From 27770b6ec7608b21dbad576b328955ce29e0d678 Mon Sep 17 00:00:00 2001 From: Thore Cimbal Date: Fri, 21 Aug 2026 12:00:00 +0000 Subject: [PATCH] =?UTF-8?q?cve:=20derive=20the=20target=20set=20and=20meas?= =?UTF-8?q?ure=20coverage=20=E2=80=94=20observing=20only=20(#0106)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The scan loop is untouched and still reads images.txt, so nothing about this deployment behaves differently. What changes is that the exporter now knows what *should* be scanned and can say how much of it is: three sources, none of them new infrastructure. The cluster and the operating host both already sit in the same Prometheus, and the registry answers an anonymous token — the same token dance Trivy performs to pull. The coverage numbers are the point of this slice. Once deployed they have to read 24 missing, 2 orphaned and about 0.52, because that is what was counted by hand on 2026-08-21. A different answer means the derivation is wrong, not the hand count stale. Failure handling is the substance rather than an afterthought. A source that fails costs only its own share; its freshness timestamp keeps ageing instead of disappearing, because a series that vanishes can never fire a rule — the third finding of the 2026-08-01 AAR. When every source fails the target file is left untouched, so a restart during a Prometheus outage cannot clear the estate. Twenty-one assertions cover it, each paired with its counter-proof: names that must collapse and names that must not, a source filter that is shown to matter by removing it, and time-ordered tag selection against the name ordering that would silently drop v0.10.0. Sabotaging the normalisation turns eleven of them red, so the suite demonstrably can fail. --- monitoring/cve/cve-exporter.py | 90 ++++++++++ monitoring/cve/targets.py | 299 +++++++++++++++++++++++++++++++++ monitoring/cve/test_targets.py | 297 ++++++++++++++++++++++++++++++++ monitoring/docker-compose.yml | 17 ++ 4 files changed, 703 insertions(+) create mode 100644 monitoring/cve/targets.py create mode 100644 monitoring/cve/test_targets.py diff --git a/monitoring/cve/cve-exporter.py b/monitoring/cve/cve-exporter.py index 5796d13..1a95f7d 100644 --- a/monitoring/cve/cve-exporter.py +++ b/monitoring/cve/cve-exporter.py @@ -13,11 +13,20 @@ # trivy_last_scan_timestamp{target,target_type,host} 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") @@ -33,6 +42,57 @@ def esc(v): return str(v).replace("\\", "\\\\").replace('"', '\\"').replace("\n", " ") +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) + 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()) @@ -96,6 +156,36 @@ def collect(): # 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}" + ) try: os.makedirs(os.path.dirname(STATE_FILE), exist_ok=True) with open(STATE_FILE, "w") as f: diff --git a/monitoring/cve/targets.py b/monitoring/cve/targets.py new file mode 100644 index 0000000..4b24e30 --- /dev/null +++ b/monitoring/cve/targets.py @@ -0,0 +1,299 @@ +#!/usr/bin/env python3 +# Soll-Menge der CVE-Scan-Ziele (#0106, ADR-0026): abgeleitet statt gepflegt. +# +# Vorher stand die Zielliste in cve/images.txt und wurde von Hand nachgezogen. +# Am 2026-08-21 gemessen: 27 von 51 laufenden Images, also 52 % Deckung - nicht +# gescannt wurden unter anderem der Web-Client, den jeder Nutzer laedt, beide +# Traefik-Schichten und die Registry selbst. Niemand hatte einen Fehler gemacht; +# die Pflege haengt an keinem Schritt, den ein Ausrollen ohnehin durchlaeuft. +# +# Drei Quellen, alle ohne neue Zugangsdaten und ohne neuen Netzweg: +# cluster - kube_pod_container_info (Prometheus) +# operating - container_last_seen{job=operating_*} (Prometheus, cAdvisor) +# registry - rohana /v2, anonymer Token-Tanz +# +# ⚠️ WARUM DIE FEHLER-BEHANDLUNG HIER DER KERN IST: Faellt eine Quelle aus, wird +# die Soll-Menge KLEINER - und die Deckung saehe damit BESSER aus, nicht +# schlechter. Ein Ausfall der Herleitung ist von Erfolg also nicht zu +# unterscheiden, solange niemand die Frische je Quelle misst. Deshalb traegt +# Herleitung.fehler und Herleitung.stand je Quelle, und deshalb wird bei +# vollstaendigem Ausfall NICHTS geschrieben (siehe cve-exporter.py). +import json +import os +import time +import urllib.error +import urllib.parse +import urllib.request +from datetime import datetime, timezone +from typing import NamedTuple, Optional + +# --- Quellen-Ausdruecke ------------------------------------------------------ +# Beide zaehlen nur, um die Beschriftung `image` zu bekommen; der Wert ist egal. +AUSDRUCK_CLUSTER = "count by (image) (kube_pod_container_info)" +# ⚠️ job=operating_cadvisor grenzt gegen gameserver_cadvisor ab (20 Images): +# game-operating liegt ausserhalb des Auftrags. Ein Beschriftungsvergleich, +# keine Ausschlussliste - eine Liste haette denselben Verfall wie images.txt. +AUSDRUCK_OPERATING = ( + 'count by (image) (container_last_seen{job="operating_cadvisor",image!=""})' +) + +# Tags, die nie ausgeliefert werden - Bau-Artefakte der CI. +TAG_PRAEFIXE_IGNORIEREN = ("sha-", "latest") + +QUELLEN = ("cluster", "operating", "registry") + + +# --- Normalisierung ---------------------------------------------------------- +def normalisiere(ref: str) -> Optional[str]: + """Bringt eine Image-Referenz auf eine vergleichbare Form. + + kube_pod_container_info liefert voll qualifiziert + (`docker.io/library/postgres:17-alpine`), cAdvisor und die alte Liste oft + kurz (`postgres:17-alpine`). Ohne diesen Schritt meldet jeder Abgleich + Scheinluecken - beim ersten Handvergleich am 2026-08-21 war das nur deshalb + folgenlos, weil die gemischten Formen zufaellig nicht kollidierten. + + Gibt None zurueck, wenn die Referenz kein Ziel sein kann. ⚠️ Insbesondere + wird ein fehlender Tag NICHT zu `:latest` ergaenzt - das erzeugte ein Ziel, + das es so nicht gibt, und genau davon wollen wir weg. + """ + ref = (ref or "").strip() + if not ref: + return None + # Digest abtrennen, Tag behalten: `foo:v1@sha256:...` -> `foo:v1` + if "@" in ref: + ref = ref.split("@", 1)[0] + + teile = ref.split("/", 1) + if len(teile) == 2 and ("." in teile[0] or ":" in teile[0] or teile[0] == "localhost"): + registry, rest = teile[0], teile[1] + else: + registry, rest = None, ref + + # Docker Hub ist der implizite Standard - beide Schreibweisen sind dasselbe. + if registry in ("docker.io", "index.docker.io"): + registry = None + if registry is None and rest.startswith("library/"): + rest = rest[len("library/") :] + + # Tag pruefen: der Doppelpunkt muss im LETZTEN Pfadteil stehen, sonst ist es + # ein Registry-Port (`myreg:5000/foo`). + if ":" not in rest.rsplit("/", 1)[-1]: + return None + + return rest if registry is None else f"{registry}/{rest}" + + +# --- Prometheus -------------------------------------------------------------- +def promql(basis_url: str, ausdruck: str, timeout: float = 10.0) -> list: + """Eine Instant-Query. Wirft bei allem, was nicht sauber 'success' ist.""" + url = basis_url.rstrip("/") + "/api/v1/query?" + urllib.parse.urlencode( + {"query": ausdruck} + ) + with urllib.request.urlopen(url, timeout=timeout) as antwort: + d = json.load(antwort) + if d.get("status") != "success": + raise RuntimeError(f"Prometheus: {d.get('status')} {d.get('error', '')}".strip()) + return d.get("data", {}).get("result", []) + + +def _images_aus_ergebnis(ergebnis: list) -> set: + ziele = set() + for reihe in ergebnis: + n = normalisiere(reihe.get("metric", {}).get("image", "")) + if n: + ziele.add(n) + return ziele + + +def ziele_cluster(basis_url: str, timeout: float = 10.0) -> set: + return _images_aus_ergebnis(promql(basis_url, AUSDRUCK_CLUSTER, timeout)) + + +def ziele_operating(basis_url: str, timeout: float = 10.0) -> set: + return _images_aus_ergebnis(promql(basis_url, AUSDRUCK_OPERATING, timeout)) + + +# --- Registry (OCI Distribution v2) ----------------------------------------- +# Der Token-Tanz ist Standard und anonym moeglich: /v2/ antwortet immer 401 mit +# einem realm-Verweis, der Client holt dort einen Token und wiederholt. Trivy +# macht genau das beim Ziehen - am 2026-08-01 in der AAR schon gemessen +# ("zieht anonym, keine Credentials noetig"), am 2026-08-21 erneut bestaetigt, +# diesmal auch fuer Katalog und Tag-Listen. +# +# ⚠️ Giteas eigene API (/api/v1/packages) waere bequemer - sie liefert +# created_at je Fassung in einem Aufruf -, verlangt aber einen Token. Das haette +# die Randbedingung "keine neuen Zugangsdaten" gebrochen, deshalb der Umweg +# ueber Manifest + Config-Blob. +MANIFEST_TYPEN = ",".join( + [ + "application/vnd.oci.image.index.v1+json", + "application/vnd.docker.distribution.manifest.list.v2+json", + "application/vnd.oci.image.manifest.v1+json", + "application/vnd.docker.distribution.manifest.v2+json", + ] +) + + +def _hole_json(url: str, token: str = "", accept: str = "", timeout: float = 15.0): + anfrage = urllib.request.Request(url) + if token: + anfrage.add_header("Authorization", f"Bearer {token}") + if accept: + anfrage.add_header("Accept", accept) + with urllib.request.urlopen(anfrage, timeout=timeout) as antwort: + return json.load(antwort) + + +def registry_token(registry: str, scope: str, timeout: float = 15.0) -> str: + url = ( + f"https://{registry}/v2/token?" + + urllib.parse.urlencode({"service": "container_registry", "scope": scope}) + ) + return _hole_json(url, timeout=timeout).get("token", "") + + +def registry_repos(registry: str, timeout: float = 15.0) -> list: + token = registry_token(registry, "registry:catalog:*", timeout) + d = _hole_json(f"https://{registry}/v2/_catalog?n=1000", token, timeout=timeout) + return list(d.get("repositories") or []) + + +def registry_tags(registry: str, repo: str, token: str = "", timeout: float = 15.0) -> list: + token = token or registry_token(registry, f"repository:{repo}:pull", timeout) + d = _hole_json(f"https://{registry}/v2/{repo}/tags/list", token, timeout=timeout) + return list(d.get("tags") or []) + + +def tag_erstellt( + registry: str, repo: str, tag: str, token: str = "", timeout: float = 15.0 +) -> Optional[datetime]: + """Erstellzeitpunkt eines Tags: Manifest -> ggf. Index-Eintrag -> Config-Blob. + + ⚠️ Nach dem NAMEN zu sortieren waere falsch: `v0.10.0` steht lexikografisch + vor `v0.9.0`. Massgeblich ist der Zeitstempel. + """ + token = token or registry_token(registry, f"repository:{repo}:pull", timeout) + basis = f"https://{registry}/v2/{repo}" + manifest = _hole_json(f"{basis}/manifests/{tag}", token, MANIFEST_TYPEN, timeout) + + # Multi-Arch: der Index verweist auf Einzelmanifeste - eines genuegt, die + # Erstellzeit unterscheidet sich zwischen den Architekturen nicht relevant. + if "manifests" in manifest and manifest.get("manifests"): + digest = manifest["manifests"][0]["digest"] + manifest = _hole_json(f"{basis}/manifests/{digest}", token, MANIFEST_TYPEN, timeout) + + digest = (manifest.get("config") or {}).get("digest") + if not digest: + return None + config = _hole_json(f"{basis}/blobs/{digest}", token, timeout=timeout) + roh = config.get("created") + if not roh: + return None + try: + return datetime.fromisoformat(roh.replace("Z", "+00:00")) + except ValueError: + return None + + +def ziele_registry(registry: str, je_repo: int = 3, timeout: float = 15.0) -> set: + """Die juengsten `je_repo` Fassungen jedes Repos. + + Reichweite laut Gate 1 (Entscheidung sorb, Option C): Bestand plus das, was + ein Rollback realistisch treffen kann. Ein Rollback-Ziel mit bekannter + Luecke ist der einzige Fall, in dem ein nicht laufendes Image betrieblich + zaehlt - bei v0.1.0 rollt niemand mehr zurueck. + """ + ziele = set() + for repo in registry_repos(registry, timeout): + token = registry_token(registry, f"repository:{repo}:pull", timeout) + kandidaten = [ + t + for t in registry_tags(registry, repo, token, timeout) + if not t.startswith(TAG_PRAEFIXE_IGNORIEREN) + ] + datiert = [] + for tag in kandidaten: + try: + wann = tag_erstellt(registry, repo, tag, token, timeout) + except (urllib.error.URLError, OSError, ValueError, KeyError): + # Ein einzelner unlesbarer Tag darf das Repo nicht kosten. + wann = None + if wann is not None: + datiert.append((wann, tag)) + datiert.sort(reverse=True) + for _, tag in datiert[:je_repo]: + n = normalisiere(f"{registry}/{repo}:{tag}") + if n: + ziele.add(n) + return ziele + + +# --- Herleitung -------------------------------------------------------------- +class Herleitung(NamedTuple): + ziele: set + je_quelle: dict # Quelle -> set + fehler: dict # Quelle -> Meldung (nur bei Ausfall) + stand: dict # Quelle -> unix-ts des letzten Erfolgs + + @property + def vollstaendig(self) -> bool: + return not self.fehler + + @property + def brauchbar(self) -> bool: + """Mindestens eine Quelle hat geliefert. + + ⚠️ Nur dann darf targets.txt ueberschrieben und duerfen Berichte + geloescht werden. Sonst raeumt ein Neustart waehrend eines + Prometheus-Ausfalls den gesamten Bestand ab. + """ + return bool(self.ziele) + + +def herleiten(cfg: dict, vorheriger_stand: Optional[dict] = None) -> Herleitung: + """Fragt alle drei Quellen. Ein Ausfall kostet nur die eigene Quelle.""" + stand = dict(vorheriger_stand or {}) + je_quelle, fehler = {}, {} + jetzt = time.time() + + aufgaben = ( + ("cluster", lambda: ziele_cluster(cfg["prometheus"], cfg["timeout"])), + ("operating", lambda: ziele_operating(cfg["prometheus"], cfg["timeout"])), + ("registry", lambda: ziele_registry(cfg["registry"], cfg["je_repo"], cfg["timeout"])), + ) + for name, hole in aufgaben: + if not cfg.get(f"{name}_an", True): + je_quelle[name] = set() + continue + try: + je_quelle[name] = hole() + stand[name] = jetzt + except Exception as e: # noqa: BLE001 - bewusst breit + je_quelle[name] = set() + fehler[name] = f"{type(e).__name__}: {e}" + + ziele = set().union(*je_quelle.values()) if je_quelle else set() + return Herleitung(ziele=ziele, je_quelle=je_quelle, fehler=fehler, stand=stand) + + +def cfg_aus_umgebung() -> dict: + return { + "prometheus": os.environ.get("PROMETHEUS_URL", "http://prometheus:9090"), + "registry": os.environ.get("REGISTRY_HOST", "rohana.axion1337.de"), + "je_repo": int(os.environ.get("REGISTRY_TAGS_JE_REPO", "3")), + "timeout": float(os.environ.get("QUELLEN_TIMEOUT", "15")), + "cluster_an": os.environ.get("QUELLE_CLUSTER", "1") != "0", + "operating_an": os.environ.get("QUELLE_OPERATING", "1") != "0", + "registry_an": os.environ.get("QUELLE_REGISTRY", "1") != "0", + } + + +if __name__ == "__main__": + # Handlauf zur Uebergabe: zeigt die Mengen, ohne etwas zu schreiben. + h = herleiten(cfg_aus_umgebung()) + for q in QUELLEN: + print(f"{q:10} {len(h.je_quelle.get(q, ())):3} {h.fehler.get(q, '')}") + print(f"{'gesamt':10} {len(h.ziele):3}") + for z in sorted(h.ziele): + print(" ", z) diff --git a/monitoring/cve/test_targets.py b/monitoring/cve/test_targets.py new file mode 100644 index 0000000..ab9f98a --- /dev/null +++ b/monitoring/cve/test_targets.py @@ -0,0 +1,297 @@ +#!/usr/bin/env python3 +# Zusicherungen fuer targets.py (#0106). Stdlib, laeuft ohne Netz: +# python3 test_targets.py +# +# Die Antworten sind AUFGEZEICHNET, nicht erfunden - abgenommen am 2026-08-21 +# von der laufenden Anlage (Prometheus auf dem Betriebs-Host, rohana /v2). +# +# ⚠️ Jede Zusicherung hat ihre GEGENPROBE. Eine Pruefung, die nur je "gleich" +# gesagt hat, ist eine Vermutung - das ist in diesem Projekt die haeufigste +# Fehlerklasse ("meldet Erfolg, ist aber blind"), zuletzt belegt in der AAR vom +# 2026-08-01. Deshalb steht neben jedem "muss zusammenfallen" ein "darf NICHT +# zusammenfallen". +import unittest +from datetime import datetime, timezone + +import targets + + +def ts(jahr, monat, tag): + return datetime(jahr, monat, tag, tzinfo=timezone.utc) + + +# --- Aufgezeichnete Antworten ------------------------------------------------ +# Auszug aus `count by (image) (kube_pod_container_info)`, 2026-08-21. +PROM_CLUSTER = [ + {"metric": {"image": "docker.io/library/postgres:17-alpine"}}, + {"metric": {"image": "docker.io/rancher/mirrored-library-traefik:3.6.10"}}, + {"metric": {"image": "rohana.axion1337.de/sorb/threadnet-web:v0.6.0"}}, + {"metric": {"image": "oci.element.io/synapse:v1.151.0-ess.1"}}, +] +# Auszug aus `container_last_seen{...}` - BEIDE Jobs, damit die Abgrenzung +# ueberhaupt etwas zu tun hat. +PROM_CADVISOR_GEMISCHT = [ + {"metric": {"image": "gitea/gitea:1.27.0", "job": "operating_cadvisor"}}, + {"metric": {"image": "traefik:v3.7.9", "job": "operating_cadvisor"}}, + {"metric": {"image": "ghcr.io/coollabsio/coolify:4.3.9", "job": "gameserver_cadvisor"}}, + {"metric": {"image": "redis:7-alpine", "job": "gameserver_cadvisor"}}, +] + + +class Normalisierung(unittest.TestCase): + def test_docker_hub_schreibweisen_fallen_zusammen(self): + formen = [ + "docker.io/library/postgres:17-alpine", + "index.docker.io/library/postgres:17-alpine", + "library/postgres:17-alpine", + "postgres:17-alpine", + ] + ergebnisse = {targets.normalisiere(f) for f in formen} + self.assertEqual(ergebnisse, {"postgres:17-alpine"}) + + def test_gegenprobe_verschiedene_fassungen_fallen_NICHT_zusammen(self): + # Ohne diese Zusicherung koennte normalisiere() alles auf denselben Wert + # abbilden und der Test oben waere trotzdem gruen. + self.assertNotEqual( + targets.normalisiere("postgres:15-alpine"), + targets.normalisiere("postgres:17-alpine"), + ) + self.assertNotEqual( + targets.normalisiere("rohana.axion1337.de/sorb/threadnet-web:v0.3.0"), + targets.normalisiere("rohana.axion1337.de/sorb/threadnet-web:v0.6.0"), + ) + + def test_fremde_registry_bleibt_unangetastet(self): + for ref in ( + "rohana.axion1337.de/sorb/threadnet-web:v0.6.0", + "ghcr.io/fluxcd/source-controller:v1.8.2", + "quay.io/jetstack/cert-manager-webhook:v1.14.0", + ): + self.assertEqual(targets.normalisiere(ref), ref) + + def test_ohne_tag_wird_verworfen_und_NICHT_zu_latest(self): + for ref in ("postgres", "docker.io/library/postgres", "ghcr.io/a/b"): + self.assertIsNone(targets.normalisiere(ref)) + + def test_digest_wird_abgetrennt_tag_bleibt(self): + self.assertEqual( + targets.normalisiere("postgres:17-alpine@sha256:" + "ab" * 32), + "postgres:17-alpine", + ) + # Nur Digest, kein Tag -> kein Ziel. + self.assertIsNone(targets.normalisiere("postgres@sha256:" + "ab" * 32)) + + def test_registry_mit_port_wird_nicht_als_tag_gelesen(self): + self.assertEqual(targets.normalisiere("reg.lan:5000/a/b:v1"), "reg.lan:5000/a/b:v1") + self.assertIsNone(targets.normalisiere("reg.lan:5000/a/b")) + + def test_leeres_und_muell(self): + for ref in ("", " ", None): + self.assertIsNone(targets.normalisiere(ref)) + + +class Quellentrennung(unittest.TestCase): + def setUp(self): + self._echt = targets.promql + + def tearDown(self): + targets.promql = self._echt + + def test_operating_nimmt_gameserver_nicht_mit(self): + # Der Ausdruck filtert serverseitig; hier wird geprueft, dass er den + # Filter ueberhaupt traegt - sonst faellt die Abgrenzung still weg. + self.assertIn('job="operating_cadvisor"', targets.AUSDRUCK_OPERATING) + + gesehen = {} + + def fake(basis, ausdruck, timeout=10.0): + gesehen["ausdruck"] = ausdruck + # Prometheus wuerde bereits gefiltert antworten - wir simulieren das, + # indem wir den Filter anwenden. + job = "operating_cadvisor" if 'job="operating_cadvisor"' in ausdruck else None + return [r for r in PROM_CADVISOR_GEMISCHT if r["metric"]["job"] == job] + + targets.promql = fake + ziele = targets.ziele_operating("http://x") + self.assertEqual(ziele, {"gitea/gitea:1.27.0", "traefik:v3.7.9"}) + + def test_gegenprobe_ohne_filter_kaeme_gameserver_mit(self): + # Beweist, dass die Zusicherung oben ueberhaupt etwas prueft. + def fake_ohne_filter(basis, ausdruck, timeout=10.0): + return PROM_CADVISOR_GEMISCHT + + targets.promql = fake_ohne_filter + ziele = targets.ziele_operating("http://x") + self.assertIn("redis:7-alpine", ziele) + + def test_cluster_normalisiert_beim_einlesen(self): + targets.promql = lambda b, a, timeout=10.0: PROM_CLUSTER + ziele = targets.ziele_cluster("http://x") + self.assertIn("postgres:17-alpine", ziele) + self.assertIn("rancher/mirrored-library-traefik:3.6.10", ziele) + self.assertNotIn("docker.io/library/postgres:17-alpine", ziele) + + +class RegistryAuswahl(unittest.TestCase): + def setUp(self): + self._repos, self._tags, self._erstellt = ( + targets.registry_repos, + targets.registry_tags, + targets.tag_erstellt, + ) + targets.registry_token = lambda *a, **k: "egal" + + def tearDown(self): + targets.registry_repos, targets.registry_tags, targets.tag_erstellt = ( + self._repos, + self._tags, + self._erstellt, + ) + + def _stelle(self, tags, zeiten): + targets.registry_repos = lambda reg, timeout=15.0: ["sorb/x"] + targets.registry_tags = lambda reg, repo, token="", timeout=15.0: tags + targets.tag_erstellt = lambda reg, repo, tag, token="", timeout=15.0: zeiten.get(tag) + + def test_juengste_drei_nach_ZEIT_nicht_nach_name(self): + # ⚠️ Der Kern: lexikografisch stuende v0.9.0 vor v0.10.0. + tags = ["v0.9.0", "v0.10.0", "v0.8.0", "v0.7.0"] + zeiten = { + "v0.7.0": ts(2026, 1, 1), + "v0.8.0": ts(2026, 2, 1), + "v0.9.0": ts(2026, 3, 1), + "v0.10.0": ts(2026, 4, 1), + } + self._stelle(tags, zeiten) + ziele = targets.ziele_registry("reg.example", je_repo=3) + self.assertEqual( + ziele, + {"reg.example/sorb/x:v0.10.0", "reg.example/sorb/x:v0.9.0", "reg.example/sorb/x:v0.8.0"}, + ) + + def test_gegenprobe_namenssortierung_haette_anders_gewaehlt(self): + # Belegt, dass der Test oben nicht zufaellig gruen ist: nach Namen + # sortiert faellt die JUENGSTE Fassung heraus und eine alte kommt mit. + tags = ["v0.9.0", "v0.10.0", "v0.8.0", "v0.7.0"] + nach_name = sorted(tags, reverse=True)[:3] + self.assertNotIn("v0.10.0", nach_name) # die juengste - waere verloren + self.assertIn("v0.7.0", nach_name) # die aelteste - waere mitgekommen + + def test_bau_artefakte_fallen_raus(self): + tags = ["v1.0.0", "sha-13547676", "latest-ci", "latest"] + zeiten = {t: ts(2026, 5, 10) for t in tags} + self._stelle(tags, zeiten) + ziele = targets.ziele_registry("reg.example", je_repo=5) + self.assertEqual(ziele, {"reg.example/sorb/x:v1.0.0"}) + + def test_tag_ohne_zeitstempel_wird_uebersprungen_nicht_geraten(self): + tags = ["v1.0.0", "v2.0.0"] + zeiten = {"v1.0.0": ts(2026, 5, 10)} # v2.0.0 liefert None + self._stelle(tags, zeiten) + ziele = targets.ziele_registry("reg.example", je_repo=5) + self.assertEqual(ziele, {"reg.example/sorb/x:v1.0.0"}) + + +class Ausfallverhalten(unittest.TestCase): + """⚠️ Der eigentliche Kern (ADR-0026): Ein Ausfall der Herleitung darf nicht + wie vollstaendige Deckung aussehen.""" + + CFG = { + "prometheus": "http://x", + "registry": "reg.example", + "je_repo": 3, + "timeout": 1.0, + "cluster_an": True, + "operating_an": True, + "registry_an": True, + } + + def setUp(self): + self._c, self._o, self._r = ( + targets.ziele_cluster, + targets.ziele_operating, + targets.ziele_registry, + ) + + def tearDown(self): + targets.ziele_cluster, targets.ziele_operating, targets.ziele_registry = ( + self._c, + self._o, + self._r, + ) + + def _stelle(self, cluster=None, operating=None, registry=None): + def macher(wert): + if isinstance(wert, Exception): + def f(*a, **k): + raise wert + return f + return lambda *a, **k: (wert or set()) + + targets.ziele_cluster = macher(cluster) + targets.ziele_operating = macher(operating) + targets.ziele_registry = macher(registry) + + def test_alles_gut(self): + self._stelle({"a:1"}, {"b:1"}, {"c:1"}) + h = targets.herleiten(self.CFG) + self.assertEqual(h.ziele, {"a:1", "b:1", "c:1"}) + self.assertTrue(h.vollstaendig) + self.assertTrue(h.brauchbar) + self.assertEqual(set(h.stand), {"cluster", "operating", "registry"}) + + def test_eine_quelle_faellt_aus_die_anderen_liefern_weiter(self): + self._stelle({"a:1"}, OSError("Prometheus weg"), {"c:1"}) + h = targets.herleiten(self.CFG) + self.assertEqual(h.ziele, {"a:1", "c:1"}) + self.assertFalse(h.vollstaendig) + self.assertTrue(h.brauchbar) + self.assertIn("operating", h.fehler) + self.assertNotIn("operating", h.stand) # Frische bleibt aus + self.assertIn("cluster", h.stand) + + def test_stand_der_ausgefallenen_quelle_bleibt_auf_dem_ALTEN_wert(self): + # Damit cve_target_source_stale altert, statt zu verschwinden. + self._stelle(OSError("weg"), {"b:1"}, {"c:1"}) + h = targets.herleiten(self.CFG, vorheriger_stand={"cluster": 1000.0}) + self.assertEqual(h.stand["cluster"], 1000.0) + self.assertGreater(h.stand["operating"], 1000.0) + + def test_alle_quellen_weg_ergibt_leere_und_UNBRAUCHBARE_menge(self): + self._stelle(OSError("a"), OSError("b"), OSError("c")) + h = targets.herleiten(self.CFG) + self.assertEqual(h.ziele, set()) + self.assertFalse(h.brauchbar) # -> nichts schreiben, nichts loeschen + self.assertEqual(set(h.fehler), {"cluster", "operating", "registry"}) + + def test_gegenprobe_leere_menge_ist_NICHT_dasselbe_wie_erfolg(self): + # Der Fall, der ohne diese Unterscheidung wie 100 % Deckung aussaehe. + self._stelle(set(), set(), set()) + h = targets.herleiten(self.CFG) + self.assertEqual(h.ziele, set()) + self.assertTrue(h.vollstaendig) # kein Fehler gemeldet ... + self.assertFalse(h.brauchbar) # ... aber trotzdem nicht verwertbar + + +class Deckungsrechnung(unittest.TestCase): + """Die Mengenarithmetik, auf der die Kriterien 2 und 3 beruhen.""" + + def test_zahlen_der_handmessung_vom_2026_08_21(self): + soll = {f"i{n}:v1" for n in range(51)} + ist = {f"i{n}:v1" for n in range(27)} | {"alt:v1", "weg:v1"} + fehlend = soll - ist + verwaist = ist - soll + self.assertEqual(len(fehlend), 24) + self.assertEqual(len(verwaist), 2) + self.assertAlmostEqual(len(soll & ist) / len(soll), 27 / 51, places=4) + + def test_gegenprobe_die_rechnung_kann_rot_werden(self): + soll = {"a:1", "b:1"} + ist = {"a:1", "b:1"} + self.assertEqual(len(soll - ist), 0) + ist_kaputt = {"a:1"} + self.assertEqual(len(soll - ist_kaputt), 1) # merkt es + + +if __name__ == "__main__": + unittest.main(verbosity=2) diff --git a/monitoring/docker-compose.yml b/monitoring/docker-compose.yml index 0c4a8c6..8da4833 100644 --- a/monitoring/docker-compose.yml +++ b/monitoring/docker-compose.yml @@ -98,6 +98,9 @@ services: - ./cve:/config:ro - cve_results:/results - cve_trivy_cache:/root/.cache + # Ab Slice 3 die Quelle der Ziele; in Slice 1 nur eingehaengt, damit + # der Pfad existiert und beim Umschalten kein Rennen entsteht. + - cve_targets:/targets:ro networks: - traefik @@ -109,10 +112,23 @@ services: environment: - RESULTS_DIR=/results - STATE_FILE=/state/first-seen.json + # Soll-Menge der Scan-Ziele (#0106, ADR-0026): abgeleitet, nicht gepflegt. + # Der Exporter bildet sie, weil er fuer die Deckungsmetrik ohnehin BEIDE + # Mengen kennen muss - ein zweiter Ort waere eine zweite Wahrheit. + - TARGETS_FILE=/targets/targets.txt + - PROMETHEUS_URL=http://prometheus:9090 + - REGISTRY_HOST=rohana.axion1337.de + # Reichweite (Entscheidung sorb 2026-08-21, Option C): Bestand plus die + # letzten drei Fassungen je Repo - so weit, wie ein Rollback zielt. + - REGISTRY_TAGS_JE_REPO=3 + # Nicht je Scrape herleiten: bei 15 s waeren das ~21 000 + # Registry-Anfragen am Tag; eine Runde kostet ~90. + - SOLL_INTERVALL_SEKUNDEN=3600 volumes: - ./cve:/app:ro - cve_results:/results:ro - cve_exporter_state:/state + - cve_targets:/targets command: python3 /app/cve-exporter.py networks: - traefik @@ -208,6 +224,7 @@ volumes: cve_results: cve_trivy_cache: cve_exporter_state: + cve_targets: grafana_data: loki_data: alloy_data: