management/scripts/stillstandspruefung.py

#!/usr/bin/env python3
"""Stillstandspruefung — sucht Dinge, die leise aufgehoert haben zu funktionieren.

Hintergrund: Retro 2026-08-09. In neun Tagen sind sechs Fehler aufgefallen, die
alle dasselbe Merkmal hatten — sie sahen erfolgreich aus, ohne es zu sein. Keiner
wurde durch eine Ueberwachung gefunden, vier durch Zufall beim Suchen nach etwas
anderem.

Jede Pruefung hier bildet einen konkreten, real passierten Fall ab. Neue Pruefungen
kommen dazu, wenn wieder etwas still ausfaellt — nicht auf Vorrat.

Beendet sich mit 1, wenn etwas gefunden wurde. In der CI heisst das: rote Pipeline,
und die ist bei uns die Alarmanlage (siehe gitops/CLAUDE.md zur TURN-Rotation).

Braucht:
  GITLAB_TOKEN   Lesezugriff auf die Gruppe axion1337.chat (read_api)
  LAB_CA         optional, Pfad zur Lab-CA-Kette
  GITEA_TOKEN    optional, noetig fuer private Spiegel (Lesezugriff genuegt)
  AUTHENTIK_URL  optional, sonst wird die Blueprint-Pruefung uebersprungen
  AUTHENTIK_TOKEN
"""

import json
import os
import subprocess
import sys
import urllib.parse
from datetime import datetime, timedelta, timezone

GITLAB = "https://git.lab/api/v4"
GITEA = "https://rohana.axion1337.de/api/v1"
GROUP = "axion1337.chat"

# Wird zur Laufzeit aus der Gruppe gelesen statt gepflegt — eine Liste im Code
# waere genau die Stelle, an der ein neues Repo jahrelang durchrutscht.
PROJEKTE = []

befunde = []
uebersprungen = []
MIT_MIRROR = []


def http(url, token=None, header="PRIVATE-TOKEN", schema=""):
    """schema ist das Praefix VOR dem Token im Header-Wert (Gitea: "token ",
    Authentik: "Bearer "). Ohne diese Trennung entsteht ein doppelter Doppelpunkt
    wie "Authorization: token: abc" — der Header ist dann still ungueltig, und die
    Pruefung meldet "nicht lesbar" statt eines Zugriffsfehlers."""
    cmd = ["curl", "-s", "--max-time", "25"]
    if token:
        cmd += ["-H", f"{header}: {schema}{token}"]
    ca = os.environ.get("LAB_CA")
    if ca and "git.lab" in url:
        cmd += ["--cacert", ca]
    cmd.append(url)
    out = subprocess.run(cmd, capture_output=True, text=True).stdout
    try:
        return json.loads(out)
    except json.JSONDecodeError:
        return None


def gl(path, token):
    return http(f"{GITLAB}{path}", token)


def enc(project):
    return urllib.parse.quote(f"{GROUP}/{project}", safe="")


def lade_projekte(token):
    d = gl(f"/groups/{urllib.parse.quote(GROUP, safe='')}/projects?per_page=100&archived=false", token)
    if not isinstance(d, list):
        return []
    return sorted(p["path"] for p in d)


def pruefe_mirror_vorhanden(token):
    """Real passiert 2026-08-09: game-operating wurde in einer Session angelegt und
    nie gespiegelt — auf Gitea existierte es gar nicht. Die Repo-Topologie sagt,
    dass die Gruppe gespiegelt wird; ein Repo ohne Mirror widerspricht ihr still."""
    for p in PROJEKTE:
        m = gl(f"/projects/{enc(p)}/remote_mirrors", token)
        if not isinstance(m, list):
            uebersprungen.append(f"Mirror-Konfiguration {p}: nicht lesbar")
            continue
        aktiv = [x for x in m if x.get("enabled")]
        if aktiv:
            MIT_MIRROR.append(p)
        else:
            befunde.append(
                f"{p}: kein aktiver Push-Mirror. Laut Repo-Topologie wird die Gruppe "
                f"gespiegelt — entweder nachziehen oder die Ausnahme begruenden.")
        for x in aktiv:
            if x.get("last_error"):
                befunde.append(f"{p}: Push-Mirror meldet Fehler: {str(x['last_error'])[:120]}")


def pruefe_mirror(token, gespiegelt):
    """Real passiert: nie — aber MIRROR-01 (management#28) beschreibt genau das.
    Faellt ein Push-Mirror aus, liefert Flux weiter den letzten gespiegelten Stand
    aus und alles sieht gesund aus.

    Nur fuer Repos MIT konfiguriertem Mirror. Ohne Mirror sind zwei gleichnamige
    Repos schlicht zwei verschiedene Repos — sie zu vergleichen erzeugt einen
    Befund, der nach Drift aussieht und keiner ist."""
    for p in gespiegelt:
        lab = gl(f"/projects/{enc(p)}/repository/commits?ref_name=main&per_page=1", token)
        if not isinstance(lab, list) or not lab:
            uebersprungen.append(f"Mirror {p}: git.lab nicht lesbar")
            continue
        git = http(f"{GITEA}/repos/sorb/{p}/commits?limit=1",
                   os.environ.get("GITEA_TOKEN"), header="Authorization", schema="token ")
        if not isinstance(git, list) or not git:
            # KEIN "uebersprungen": Dieses Repo HAT einen Mirror. Koennen wir die
            # Gegenseite nicht lesen, wissen wir nicht, ob er auseinanderlaeuft —
            # und "nicht geprueft" darf nicht wie "in Ordnung" aussehen. Real
            # aufgefallen am 2026-08-09, als game-operating auf Gitea privat
            # gestellt wurde und die Pruefung es klaglos uebersprang.
            befunde.append(
                f"Mirror {p}: Gegenseite auf Gitea nicht lesbar — Abgleich nicht "
                f"moeglich. Bei privaten Spiegeln GITEA_TOKEN hinterlegen.")
            continue
        if lab[0]["id"] != git[0]["sha"]:
            befunde.append(
                f"Mirror {p} auseinander: git.lab {lab[0]['id'][:8]} / "
                f"gitea {git[0]['sha'][:8]}. Flux liest die Gitea-Seite.")


def pruefe_leere_pipelines(token):
    """Real passiert 2026-08-06 (ThreadNet-Web 203/204) und bei threadnet-call 187:
    Pipelines ohne einen einzigen Job gelten als failed. Rot, das nichts bedeutet,
    gewoehnt einem das Hinsehen ab."""
    seit = (datetime.now(timezone.utc) - timedelta(days=8)).strftime("%Y-%m-%dT%H:%M:%SZ")
    for p in PROJEKTE:
        pls = gl(f"/projects/{enc(p)}/pipelines?status=failed&updated_after={seit}&per_page=20", token)
        if not isinstance(pls, list):
            continue
        for pl in pls:
            jobs = gl(f"/projects/{enc(p)}/pipelines/{pl['id']}/jobs?per_page=1", token)
            if isinstance(jobs, list) and len(jobs) == 0:
                befunde.append(
                    f"{p}: Pipeline {pl['id']} ({pl['ref']}) ist rot, hat aber NULL Jobs — "
                    f"kein Fehler, sondern eine Regel, die keine Jobs erzeugt.")


def pruefe_artefakte(token):
    """Real passiert bei threadnet-call: build_embedded lief seit jeher gruen und
    lud NICHTS hoch, weil der artifacts-Pfad nicht zum Build-Output passte. Der
    nachgelagerte publish_npm haette ein leeres Paket veroeffentlicht."""
    seit = (datetime.now(timezone.utc) - timedelta(days=8)).strftime("%Y-%m-%dT%H:%M:%SZ")
    for p in PROJEKTE:
        jobs = gl(f"/projects/{enc(p)}/jobs?scope[]=success&per_page=40", token)
        if not isinstance(jobs, list):
            continue
        for j in jobs:
            if (j.get("finished_at") or "") < seit:
                continue
            # Nur Jobs, von denen andere per needs Artefakte erwarten koennen
            if not j.get("artifacts") and j["name"] in ("build_embedded", "web"):
                befunde.append(
                    f"{p}: Job '{j['name']}' (#{j['id']}) war erfolgreich, hat aber "
                    f"kein Artefakt — nachgelagerte Jobs bekommen nichts.")


def pruefe_npm_paketgroessen():
    """Real passiert 2026-08-06: 0.19.2-threadnet.6 lag mit 12,5 KB statt 12,8 MB
    in der Registry — package.json und Lizenzen, kein dist/. Eine npm-Version laesst
    sich nicht ueberschreiben, die Nummer war dauerhaft verbrannt."""
    meta = http("https://rohana.axion1337.de/api/packages/sorb/npm/"
                "@sorb%2Fthreadnet-call-embedded")
    if not isinstance(meta, dict) or "versions" not in meta:
        uebersprungen.append("npm-Paket: Registry nicht lesbar")
        return
    MINDEST = 1_000_000  # ein Widget-Bundle liegt bei ~12 MB
    for ver, v in meta["versions"].items():
        url = v.get("dist", {}).get("tarball")
        if not url:
            continue
        out = subprocess.run(
            ["curl", "-sIL", "--max-time", "25", "-o", "/dev/null",
             "-w", "%{size_download} %{header_json}", url],
            capture_output=True, text=True).stdout
        try:
            size = int(json.loads(out.split(" ", 1)[1])["content-length"][0])
        except Exception:
            continue
        if size < MINDEST:
            befunde.append(
                f"npm @sorb/threadnet-call-embedded@{ver}: nur {size:,} Bytes — "
                f"das Paket enthaelt vermutlich kein dist/.")


def pruefe_authentik_blueprints():
    """Real passiert: matrix-recovery-flow wurde tagelang bei JEDEM Lauf verworfen,
    waehrend Flux gruen war und die ConfigMap aktuell. Gefunden nur, weil jemand
    fuer etwas anderes in die Datenbank sah."""
    url, tok = os.environ.get("AUTHENTIK_URL"), os.environ.get("AUTHENTIK_TOKEN")
    if not (url and tok):
        uebersprungen.append(
            "Authentik-Blueprints: AUTHENTIK_URL/AUTHENTIK_TOKEN fehlen — "
            "genau der Fall, der uns am laengsten unbemerkt lief")
        return
    d = http(f"{url.rstrip('/')}/api/v3/managed/blueprints/?page_size=100",
             tok, header="Authorization", schema="Bearer ")
    if not isinstance(d, dict):
        uebersprungen.append("Authentik-Blueprints: API nicht lesbar")
        return
    for b in d.get("results", []):
        if b.get("status") != "successful":
            befunde.append(
                f"Authentik-Blueprint '{b.get('name')}': Status "
                f"'{b.get('status')}' — wird bei jedem Lauf verworfen.")


def main():
    token = os.environ.get("GITLAB_TOKEN")
    if not token:
        print("GITLAB_TOKEN fehlt — ohne Lesezugriff ist keine Pruefung moeglich.")
        return 2
    global PROJEKTE
    PROJEKTE = lade_projekte(token)
    if not PROJEKTE:
        print("Gruppe nicht lesbar — Token pruefen.")
        return 2
    print(f"Geprueft: {len(PROJEKTE)} Projekte der Gruppe {GROUP}\n")
    pruefe_mirror_vorhanden(token)
    pruefe_mirror(token, MIT_MIRROR)
    pruefe_leere_pipelines(token)
    pruefe_artefakte(token)
    pruefe_npm_paketgroessen()
    pruefe_authentik_blueprints()

    if uebersprungen:
        print("Uebersprungen:")
        for u in uebersprungen:
            print(f"  - {u}")
        print()
    if befunde:
        print(f"BEFUNDE ({len(befunde)}):")
        for b in befunde:
            print(f"  ! {b}")
        print("\nJeder Befund gehoert als Issue erfasst, nicht weggeklickt.")
        return 1
    print("Keine Befunde.")
    return 0


if __name__ == "__main__":
    sys.exit(main())