leistungsbilanz-ts/scripts/security/stamp.py

#!/usr/bin/env python3
"""Prüfstempel: merkt sich, welche Baseline-Prüfung für welchen Commit grün war.

Damit läuft vor dem Deploy nicht noch einmal, was für genau diesen Commit grün war.

Aufruf (aus dem Repo-Root, so wie das Makefile es tut):
  stamp.py begin  NAME            vor dem Lauf: Ausgangszustand merken
  stamp.py record NAME [DETAIL]   nach grünem Lauf: Stempel für HEAD schreiben
  stamp.py status [NAME[=DETAIL]] Exit 0, wenn alle genannten Stempel für HEAD gelten
                                  und der Arbeitsbaum sauber ist; ohne NAME Übersicht
  stamp.py report [NAME ...]      Markdown-Tabelle fürs Release-Protokoll
  stamp.py clean                  Exit 0, wenn der Arbeitsbaum sauber ist
  stamp.py pending ZIEL ...       gibt die make-Ziele aus, die kein gültiger Stempel
                                  abdeckt (für `make prepush`)

Ein Stempel gilt nur, wenn genau der Commit geprüft wurde:
  - Arbeitsbaum beim Start sauber (keine geänderten, keine neuen Dateien),
  - HEAD am Ende gleich HEAD am Start,
  - keine versionierte Datei vom Lauf verändert (neue Artefakte wie Logs erlaubt).
Ein Stempel eines älteren Commits gilt weiter, wenn danach nur Doku kam
(Einteilung wie `make prepush`, siehe changed_scope.py).
`prepush` mit vollem Programm (secrets lint test sast audit) zählt auch als `check`.

Ablage: reports/stamps/<commit>.json (STAMP_DIR, relativ zum Repo-Root).
Nur lokal, nie committen.
"""

import datetime
import json
import os
import re
import shutil
import subprocess  # nosec B404 - nur feste git-Aufrufe ohne Shell
import sys

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from changed_scope import classify  # noqa: E402, RUF100

GIT = shutil.which("git") or "git"
STAMP_DIR = os.environ.get("STAMP_DIR") or os.path.join("reports", "stamps")
NAME_RE = re.compile(r"^[a-z][a-z0-9-]{0,39}$")
FULL_CHECK = {"secrets", "lint", "test", "sast", "audit"}
DETAIL_NAMES = {"container", "zap", "fuzz"}
# Welche prepush-Ziele ein gültiger Stempel abdeckt.
COVERS = {
    "check": {"secrets", "secrets-diff", "lint", "test", "sast", "sast-diff", "audit"},
    "mutation": {"mutation"},
}
SEARCH_DEPTH = 50


def _git(*args):
    cmd = [GIT, *args]
    # Feste git-Aufrufe ohne Shell, Argumente nur aus diesem Skript bzw. Commit-Hashes.
    r = subprocess.run(cmd, capture_output=True, text=True, check=False)  # noqa: S603, RUF100  # nosec B603
    return r.returncode, r.stdout.strip()


def _outside_stamps():
    return ["--", ".", f":(exclude){STAMP_DIR}"]


def head():
    rc, out = _git("rev-parse", "HEAD")
    return out if rc == 0 else None


def tree_state():
    """Geänderte versionierte und neue Dateien, ohne den Stempelordner selbst."""
    args = ("status", "--porcelain=v1", "--untracked-files=all")
    rc, out = _git(*args, *_outside_stamps())
    tracked, untracked = [], []
    for line in out.splitlines() if rc == 0 else []:
        (untracked if line.startswith("??") else tracked).append(line[3:])
    return tracked, untracked


def _now():
    return datetime.datetime.now().astimezone().isoformat(timespec="seconds")


def _check_name(name):
    if not NAME_RE.match(name):
        print(f"stamp: ungültiger Name {name!r}", file=sys.stderr)
        sys.exit(2)


def _stamp_file(commit):
    return os.path.join(STAMP_DIR, commit + ".json")


def _begin_file(name):
    return os.path.join(STAMP_DIR, f".begin-{name}.json")


def _load(path, default):
    try:
        with open(path, encoding="utf-8") as f:
            return json.load(f)
    except (OSError, ValueError):
        return default


def _entries(commit):
    return _load(_stamp_file(commit), {}).get("entries", [])


def _save(path, data):
    os.makedirs(STAMP_DIR, exist_ok=True)
    tmp = path + ".tmp"
    with open(tmp, "w", encoding="utf-8") as f:
        json.dump(data, f, ensure_ascii=False, indent=1)
    os.replace(tmp, path)


def begin(name):
    tracked, untracked = tree_state()
    state = {"head": head(), "tracked": tracked, "untracked": untracked}
    _save(_begin_file(name), state)
    return 0


def _entry(name, detail, start, end_head, end_tracked, end_untracked):
    notes, valid = [], True
    if start is None:
        valid, notes = False, ["kein Startzustand (begin fehlt)"]
    else:
        if start.get("tracked") or start.get("untracked"):
            valid = False
            notes.append("Arbeitsbaum beim Start nicht sauber")
        if start.get("head") != end_head:
            valid = False
            notes.append("HEAD hat sich während des Laufs geändert")
        if end_tracked:
            valid = False
            changed = ", ".join(end_tracked[:5])
            notes.append("Lauf hat versionierte Dateien verändert: " + changed)
        new = sorted(set(end_untracked) - set(start.get("untracked") or []))
        if new:
            notes.append("neue Dateien aus dem Lauf: " + ", ".join(new[:5]))
    return {
        "name": name,
        "detail": detail,
        "valid": valid,
        "at": _now(),
        "notes": notes,
    }


def record(name, detail):
    start = _load(_begin_file(name), None)
    end_head = head()
    if end_head is None:
        print("stamp: kein Commit (HEAD fehlt)", file=sys.stderr)
        return 1
    tracked, untracked = tree_state()
    entries = [_entry(name, detail, start, end_head, tracked, untracked)]
    if name == "prepush" and FULL_CHECK <= set(detail.split()):
        entries.append(dict(entries[0], name="check", detail="über prepush"))
    data = _load(_stamp_file(end_head), {"commit": end_head, "entries": []})
    data["entries"].extend(entries)
    _save(_stamp_file(end_head), data)
    try:
        os.remove(_begin_file(name))
    except OSError:
        pass
    e = entries[0]
    if e["valid"]:
        print(f"stamp: {name} gültig für {end_head[:7]}")
    else:
        why = "; ".join(e["notes"])
        print(f"stamp: {name} UNGÜLTIG für {end_head[:7]} ({why})")
    return 0


def _latest(commit, name, detail):
    """Jüngster Eintrag für name (und detail, falls angegeben) in einem Commit."""
    for e in reversed(_entries(commit)):
        if e.get("name") != name:
            continue
        if detail is None or e.get("detail") == detail:
            return e
    return None


def _docs_only(old, new):
    rc, out = _git("diff", "--name-only", "--no-renames", old, new, *_outside_stamps())
    if rc != 0:
        return False
    files = [f for f in out.splitlines() if f]
    return len(classify(files)["docs"]) == len(files)


def lookup(name, detail):
    """(Zustand, Eintrag, Commit): Zustand ist gültig, ungültig oder fehlt."""
    cur = head()
    if cur is None:
        return "fehlt", None, None
    rc, out = _git("rev-list", f"--max-count={SEARCH_DEPTH}", "HEAD")
    commits = out.splitlines() if rc == 0 else [cur]
    first_seen = None
    for c in commits:
        if c != cur and not _docs_only(c, cur):
            break
        e = _latest(c, name, detail)
        if e is None:
            continue
        if e.get("valid"):
            return "gültig", e, c
        first_seen = first_seen or (e, c)
    if first_seen:
        return "ungültig", first_seen[0], first_seen[1]
    return "fehlt", None, None


def _parse(spec):
    name, _, detail = spec.partition("=")
    _check_name(name)
    return name, (detail or None)


def _describe(name, detail, state, e, c, cur):
    label = name + (f" ({detail})" if detail else "")
    if state == "fehlt":
        return f"  {label:<24} fehlt"
    via = "" if c == cur else f", von {c[:7]}, danach nur Doku"
    notes = "; ".join(e.get("notes") or [])
    line = f"  {label:<24} {state}  {e.get('at', '')}{via}"
    return line + (f"  [{notes}]" if notes else "")


def _known_names():
    cur = head()
    names = [("check", None), ("mutation", None)]
    for e in _entries(cur) if cur else []:
        name = e.get("name")
        key = (name, e.get("detail") if name in DETAIL_NAMES else None)
        if key not in names:
            names.append(key)
    return names


def status(specs):
    cur = head()
    tracked, untracked = tree_state()
    dirty = tracked + untracked
    tree = "sauber" if not dirty else f"nicht sauber ({len(dirty)} Dateien)"
    print(f"Stempel für {cur[:7] if cur else '-'}: Arbeitsbaum {tree}")
    wanted = [_parse(s) for s in specs]
    ok = True
    for name, detail in wanted or _known_names():
        state, e, c = lookup(name, detail)
        print(_describe(name, detail, state, e, c, cur))
        ok = ok and state == "gültig"
    if not wanted:
        return 0
    return 0 if ok and not dirty else 1


def _report_row(name, cur):
    state, e, c = lookup(name, None)
    if state == "fehlt":
        return f"| {name} | nicht ausgeführt | | | |"
    result = "grün" if state == "gültig" else "ungültig"
    via = c[:7] + (" (danach nur Doku)" if c != cur else "")
    hints = ([e["detail"]] if e.get("detail") else []) + (e.get("notes") or [])
    return f"| {name} | {result} | {e.get('at', '')} | {via} | {'; '.join(hints)} |"


def report(names):
    for n in names:
        _check_name(n)
    cur = head()
    rows = [
        "| Prüfung | Ergebnis | Zeit | Commit | Hinweise |",
        "|---|---|---|---|---|",
    ]
    for name in names or [n for n, _ in _known_names()]:
        rows.append(_report_row(name, cur))
    print("\n".join(rows))
    return 0


def clean():
    tracked, untracked = tree_state()
    for path in (tracked + untracked)[:10]:
        print(f"  nicht committet: {path}")
    return 0 if not tracked + untracked else 1


def pending(targets):
    """Ziele ohne gültigen Stempel; bei unsauberem Arbeitsbaum alle."""
    tracked, untracked = tree_state()
    covered = set()
    if not tracked + untracked:
        for name, targets_covered in COVERS.items():
            if lookup(name, None)[0] == "gültig":
                covered |= targets_covered
    skipped = [t for t in targets if t in covered]
    if skipped:
        print("  Stempel gültig, entfällt: " + " ".join(skipped), file=sys.stderr)
    print(" ".join(t for t in targets if t not in covered))
    return 0


def main(argv):
    modes = {"begin", "record", "status", "report", "clean", "pending"}
    if len(argv) < 2 or argv[1] not in modes:
        print(__doc__, file=sys.stderr)
        return 2
    mode, args = argv[1], argv[2:]
    if mode == "clean":
        return clean()
    if mode == "status":
        return status(args)
    if mode == "report":
        return report(args)
    if mode == "pending":
        return pending(args)
    if not args:
        print(__doc__, file=sys.stderr)
        return 2
    _check_name(args[0])
    if mode == "begin":
        return begin(args[0])
    return record(args[0], " ".join(args[1:]))


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