bf-infra/nautobot/ingest-gc-scrubbing.py

#!/usr/bin/env python3
r"""BF-Netzwerk -> Nautobot Golden Config: Backup-Scrubbing (TP22 Schritt 0, 2026-08-13).

HINTERGRUND / WARUM DAS EXISTIERT
---------------------------------
Der GC-Backup-Job schreibt den rohen RouterOS-`/export` ins Gitea-Repo `bf-golden-config`.
Ohne ConfigReplace-Regeln landen dort Secrets im KLARTEXT. Genau das ist am 2026-07-22
passiert: `backups/ap1.cfg` enthielt den (bis heute live genutzten) WG-Hub-private-key und
6 WLAN-PSKs. Diese Regeln maskieren solche Werte, BEVOR sie das Repo erreichen.

  -> MUSS scharf sein, bevor der Backup-Job fuer crs326-core/-branch/ap2 laeuft,
     sonst leakt der Core-Backup denselben WG-Key ein zweites Mal.

REGEX-DESIGN (gegen echten ap1-Export kalibriert, 2026-08-13)
-------------------------------------------------------------
RouterOS quotet Werte nur, wenn noetig -> im ap1-Export standen 4 von 6 `passphrase=`
UNQUOTIERT und 2 quotiert (mit `\\$`-Escapes). Beide Formen muessen fallen, deshalb:

    (feld=)("(?:[^"\\]|\\.)*"|\S+)

Die quotierte Alternative steht ZWINGEND VORNE: `\\S+` ist greedy und wuerde bei einem
PSK mit Leerzeichen (`passphrase="my psk"`) nur `"my` fassen und den Rest exponieren.
Python-`re` bevorzugt die linke Alternative -> quotiert zuerst.

Der Replace normalisiert IMMER auf die quotierte Form, damit die Maske unabhaengig von
der Ursprungsform byte-identisch ist. Das ist Voraussetzung dafuer, dass das
Jinja-Template denselben String emittieren kann (sonst Dauer-Diff in der Compliance).

  export NAUTOBOT_TOKEN=...
  python3 nautobot/ingest-gc-scrubbing.py          # Dry-Run
  python3 nautobot/ingest-gc-scrubbing.py --apply  # schreiben
"""
import json, os, re, sys, urllib.error, urllib.parse, urllib.request

BASE = "http://10.40.20.154:8080/api"
TOKEN = os.environ.get("NAUTOBOT_TOKEN", "")
DRY = "--apply" not in sys.argv

# Der Masken-String. Muss identisch im Jinja-Template stehen (TP22 Schritt 1).
MASK = '"<<REDACTED>>"'

# Wert-Matcher: quotiert (mit Escape-Support) ODER unquotiert. Quotiert MUSS zuerst.
VAL = r'("(?:[^"\\]|\\.)*"|\S+)'

# Platform-Name, auf den die Regeln gelten.
PLATFORM = "RouterOS"

# (name, feld-regex, beschreibung)
# Namen numerisch praefixiert: GC ordnet nach (platform, name), die Reihenfolge ist
# damit deterministisch und nachvollziehbar.
RULES = [
    (
        "10-wifi-passphrase",
        r"(passphrase=)" + VAL,
        "WLAN-PSKs (/interface wifi security). Im ap1-Export 6x, davon 4 unquotiert.",
    ),
    (
        "20-wireguard-private-key",
        r"(private-key=)" + VAL,
        "WireGuard-private-key (wg-hub auf crs326-core). Der kritischste Wert im Fleet.",
    ),
    (
        "30-preshared-key",
        r"((?:wpa2?-)?pre-?shared-key=)" + VAL,
        "WG-preshared-key / wpa-pre-shared-key. Defensiv - im ap1-Export nicht vorhanden.",
    ),
    (
        "40-password",
        r"(password=)" + VAL,
        "Generisch password= (faengt auch authentication-/encryption-password=). Defensiv.",
    ),
    (
        "50-secret",
        r"(\bsecret=)" + VAL,
        "RADIUS-/IPsec-/PPP-secret=. \\b verhindert Treffer auf secret-artige Praefixe. Defensiv.",
    ),
]

# Selbsttest-Faelle: (eingabe, erwartete ausgabe) - laufen bei JEDEM Aufruf, auch Dry-Run.
SELFTEST = [
    ("passphrase=jgp4ap3cwcnz", 'passphrase="<<REDACTED>>"'),
    (r'passphrase="74c9L283\$669"', 'passphrase="<<REDACTED>>"'),
    ('passphrase="psk mit space"', 'passphrase="<<REDACTED>>"'),
    (
        "add name=sec-main ft=yes passphrase=abc123 disabled=no",
        'add name=sec-main ft=yes passphrase="<<REDACTED>>" disabled=no',
    ),
    ('private-key="9AFG+aGpk3/gnzka46oVvFfcU="', 'private-key="<<REDACTED>>"'),
    # Public-Keys duerfen NICHT maskiert werden (sie stehen legitim im Template):
    ('public-key="dzRY0lnnUygijChWe5wXZmXPeNcbdALpqBptA3fyiVE="', None),
    # Nicht-Secrets duerfen nicht angefasst werden:
    ("name=dp-main vlan-id=10", None),
]


def req(method, path, data=None):
    url = path if path.startswith("http") else f"{BASE}{path}"
    body = json.dumps(data).encode() if data is not None else None
    r = urllib.request.Request(url, data=body, method=method)
    r.add_header("Authorization", f"Token {TOKEN}")
    r.add_header("Content-Type", "application/json")
    r.add_header("Accept", "application/json")
    try:
        with urllib.request.urlopen(r, timeout=30) as resp:
            raw = resp.read()
            return json.loads(raw) if raw else {}
    except urllib.error.HTTPError as e:
        print(f"!! {method} {url}\n   payload={json.dumps(data)}\n   -> {e.code} {e.read().decode()[:500]}")
        raise


def find(path, **params):
    q = "&".join(f"{k}={urllib.parse.quote(str(v))}" for k, v in params.items())
    res = req("GET", f"{path}?{q}&limit=2")
    return res["results"][0] if res.get("count") else None


def selftest():
    """Beweist die Regexe lokal, bevor irgendetwas nach Nautobot geht."""
    fails = []
    for src, want in SELFTEST:
        got = src
        for _, rx, _ in RULES:
            got = re.sub(rx, r"\1" + MASK, got)
        expected = src if want is None else want
        status = "ok " if got == expected else "FAIL"
        if got != expected:
            fails.append((src, expected, got))
        # Werte nie ausgeben - nur ob die Maske sitzt.
        shown = re.sub(r"=(\"[^\"]*\"|\S+)", "=…", src)
        print(f"    [{status}] {shown}")
    return fails


def main():
    print("== Regex-Selbsttest ==")
    fails = selftest()
    if fails:
        print("\n!! Selbsttest fehlgeschlagen - NICHTS geschrieben:")
        for src, exp, got in fails:
            print(f"   input={src!r}\n   want ={exp!r}\n   got  ={got!r}")
        sys.exit(1)
    print("   alle Faelle ok\n")

    if not TOKEN:
        print("!! NAUTOBOT_TOKEN nicht gesetzt")
        sys.exit(1)

    plat = find("/dcim/platforms", name=PLATFORM)
    if not plat:
        print(f"!! Platform {PLATFORM!r} nicht gefunden")
        sys.exit(1)
    print(f"== ConfigReplace-Regeln (Platform {PLATFORM}) ==")

    for name, rx, desc in RULES:
        payload = {
            "name": name,
            "platform": plat["id"],
            "regex": rx,
            "replace": r"\1" + MASK,
            "description": desc,
        }
        obj = find("/plugins/golden-config/config-replace", name=name)
        if obj:
            delta = {k: v for k, v in payload.items() if k in ("regex", "replace", "description") and obj.get(k) != v}
            if not delta:
                print(f"  = {name}")
            elif DRY:
                print(f"  ~ {name}  {list(delta)}  (dry-run)")
            else:
                req("PATCH", f"/plugins/golden-config/config-replace/{obj['id']}/", delta)
                print(f"  ~ {name}  {list(delta)}")
        elif DRY:
            print(f"  + {name}  (dry-run)")
        else:
            req("POST", "/plugins/golden-config/config-replace/", payload)
            print(f"  + {name}")

    print("\nMaske: " + MASK + "   <- muss identisch im Jinja-Template stehen (TP22 Schritt 1)")
    if DRY:
        print("DRY-RUN - nichts geschrieben. Mit --apply ausfuehren.")


if __name__ == "__main__":
    main()