#!/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, laufend: Optional[set] = None, 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. ⚠️ EINENGUNG (#0051, Entscheidung sorb 2026-08-21): Beruecksichtigt werden nur Repos, von denen gerade MINDESTENS EIN Tag laeuft. Ohne diese Bedingung kamen reine Bau-Artefakte in die Zielmenge - `element-desktop-build` und `windows-vm` trugen zusammen 61 CRITICAL, davon 1 behebbar. Befunde ueber Images, die niemand betreibt, sind Rauschen, das wie Signal aussieht. Die Einengung ist ABGELEITET, nicht gepflegt: `laufend` ist dieselbe Menge, die die Herleitung ohnehin bildet. Keine zweite Liste (ADR-0026). """ ziele = set() for repo in registry_repos(registry, timeout): if laufend is not None and not any( z.startswith(f"{registry}/{repo}:") for z in laufend ): # Kein Tag dieses Repos laeuft - Bau-Artefakt, kein Rollback-Ziel. continue 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. ⚠️ REIHENFOLGE IST BEDEUTSAM: Die Registry-Auswahl braucht den laufenden Bestand, um Bau-Artefakte auszuschliessen (#0051). Faellt der Bestand aus, faellt die Registry-Auswahl mit - und das ist richtig so: Eine Registry-Auswahl ohne Bestandswissen koennte nicht unterscheiden, was ein Rollback-Ziel ist und was ein Bau-Artefakt. """ stand = dict(vorheriger_stand or {}) je_quelle, fehler = {}, {} jetzt = time.time() def versuche(name, hole): if not cfg.get(f"{name}_an", True): je_quelle[name] = set() return 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}" versuche("cluster", lambda: ziele_cluster(cfg["prometheus"], cfg["timeout"])) versuche("operating", lambda: ziele_operating(cfg["prometheus"], cfg["timeout"])) laufend = je_quelle.get("cluster", set()) | je_quelle.get("operating", set()) if laufend: versuche("registry", lambda: ziele_registry( cfg["registry"], cfg["je_repo"], laufend, cfg["timeout"])) else: # Ohne Bestandswissen keine Registry-Auswahl - lieber eine kleinere # Menge als eine, die Bau-Artefakte mitschleppt. Als Fehler gemeldet, # damit es nicht wie Erfolg aussieht. je_quelle["registry"] = set() if "registry" not in fehler: fehler["registry"] = "uebersprungen: kein laufender Bestand bekannt" 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)