cve: derive the target set and measure coverage — observing only (#0106)
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.
This commit is contained in:
Thore Cimbal
@@ -13,11 +13,20 @@
|
||||
# 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")
|
||||
@@ -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:
|
||||
|
||||
@@ -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)
|
||||
@@ -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)
|
||||
@@ -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:
|
||||
|
||||
Reference in New Issue
Block a user