bf-infra/nautobot/ingest-veth-rule.py

#!/usr/bin/env python3
r"""ComplianceRule `interface-veth` + zwei Normalisierungs-Regeln (TP16.3 §5, 2026-08-31).

DIE LUECKE, DIE DAS SCHLIESST
-----------------------------
Seit dem 2026-08-31 traegt jeder AP ein `veth-cf` — das Container-Netz, an dem der
`cloudflared`-Connector fuer `bf-public` haengt. Es ist das tragende Stueck: ohne das
veth laeuft kein Container, und mit ihm haengt der oeffentliche Weg von vier Diensten.

Fuer `/interface veth` gab es bis heute **keine ComplianceRule** — deshalb hat das veth
auch nie eine Drift erzeugt. Das ist genau die Sorte Ruhe, die nichts beweist: wer es
loescht, bekommt keine Meldung. Dieselbe blinde Stelle wie bei `/queue` (TP6),
`/ip firewall filter` (TP26) und `/system logging` (TP25 Fund E).

WARUM DAS SKRIPT MEHR MACHT ALS EINE REGEL ANZULEGEN
----------------------------------------------------
Der rohe Export traegt ZWEI geraetevergebene MAC-Adressen, auf jedem AP verschieden:

    /interface veth add address=172.17.0.2/24 container-mac-address=5E:85:94:E7:F9:51 \
        dhcp=no gateway=172.17.0.1 gateway6="" mac-address=5E:85:94:E7:F9:50 name=veth-cf

Eine Regel ohne Vorbehandlung erzeugte damit auf allen drei APs **Dauer-Drift** — die
Regel wuerde also genau das produzieren, was sie verhindern soll: eine Zahl, die
immer rot ist und die deshalb niemand mehr liest.

Zwei Auswege, und warum es dieser geworden ist:
  (a) MACs je Geraet in Nautobot modellieren. Verworfen: `ingest-ap-configcontext.py`
      begruendet ausfuehrlich, warum geraetevergebene MACs NICHT ins Modell gehoeren
      (dieselbe Entscheidung wie bei den VAP-Interfaces, s. Template-Kommentar [1]).
      Ein veth-Neubau vergibt ausserdem neue MACs -> das Modell waere sofort wieder alt.
  (b) Die zwei Felder im Backup normalisieren, wie es fuer Secrets laengst geschieht.
      -> dieser Weg.

⚠️ EHRLICH DAZUGESAGT: die Maske kostet Backup-Treue. In `bf-golden-config` stehen die
   veth-MACs danach nicht mehr im Klartext. Das ist vertretbar, weil RouterOS sie beim
   Neuanlegen ohnehin neu vergibt — ein Restore braucht sie nicht. Was die Regel
   beweist, ist die EXISTENZ des veth mit der richtigen Adressierung, nicht seine MAC.

🔑 WARUM DIESELBE MASKE WIE FUER SECRETS (`<<REDACTED>>`) UND KEINE EIGENE
--------------------------------------------------------------------------
Hier ist sie kein Geheimnis-Hinweis, sondern ein **Push-Schutz**: `gc-configplan.py`
weigert sich, ein `config_set` mit `<<REDACTED>>` zu senden (TP22-Regel 1). Eine
eigene Maske wie `<<MAC>>` haette diesen Schutz NICHT — und eine gepushte Zeile
`mac-address=<<MAC>>` waere Muell auf dem Geraet. Die Maske als "nicht pushbar"-Marke
zu lesen ist die allgemeinere und hier die richtige Bedeutung.

Die Secrets-Regeln in `ingest-gc-scrubbing.py` bleiben davon **unberuehrt** — dieses
Skript legt nur zusaetzliche, eng verankerte Regeln an. Beide Skripte sind additiv und
raeumen nichts weg, was ihnen nicht gehoert.

REGEX-VERANKERUNG
-----------------
Beide Muster beginnen mit `/interface veth add ` und benutzen `[^\n]*?` — sie koennen
die Zeile also nicht verlassen. `mac-address=` allein waere gefaehrlich (das Feld gibt
es auch an VAPs, Bridges und Ethernet-Ports); der Lookbehind `(?<![-\w])` haelt
ausserdem `container-mac-address=` heraus, damit die zwei Regeln sich nicht ins
Gehege kommen — in beiden Reihenfolgen.

    export NAUTOBOT_TOKEN=...
    python3 nautobot/ingest-veth-rule.py          # Dry-Run + Selbsttest
    python3 nautobot/ingest-veth-rule.py --apply
"""
import json
import os
import re
import sys
import urllib.error
import urllib.parse
import urllib.request

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

PLATFORM = "RouterOS"
FEATURE = "interface-veth"
MATCH = "/interface veth"
MASK = "<<REDACTED>>"

# (name, regex, beschreibung) - Namen numerisch praefixiert, GC ordnet nach (platform, name).
# 60/61 schliessen an die Secret-Regeln 10-50 aus ingest-gc-scrubbing.py an.
REPLACES = [
    (
        "60-veth-container-mac",
        r"(/interface veth add [^\n]*?container-mac-address=)\S+",
        "Geraetevergebene Container-MAC des veth (TP16.3). Kein Secret - Normalisierung, "
        "damit die ComplianceRule interface-veth nicht auf allen APs Dauer-Drift meldet.",
    ),
    (
        "61-veth-mac",
        r"(/interface veth add [^\n]*?(?<![-\w])mac-address=)\S+",
        "Geraetevergebene MAC des veth (TP16.3). Lookbehind haelt container-mac-address= "
        "heraus; die Verankerung auf die veth-Zeile haelt VAP-/Bridge-MACs heraus.",
    ),
]

# Selbsttest: (eingabe, erwartete ausgabe oder None fuer 'unveraendert').
AP1 = ('/interface veth add address=172.17.0.2/24 container-mac-address=5E:85:94:E7:F9:51 '
       'dhcp=no gateway=172.17.0.1 gateway6="" mac-address=5E:85:94:E7:F9:50 name=veth-cf')
AP1_SOLL = ('/interface veth add address=172.17.0.2/24 container-mac-address=<<REDACTED>> '
            'dhcp=no gateway=172.17.0.1 gateway6="" mac-address=<<REDACTED>> name=veth-cf')

SELFTEST = [
    (AP1, AP1_SOLL),
    # Bereits maskiert -> idempotent, kein zweiter Durchlauf-Schaden.
    (AP1_SOLL, AP1_SOLL),
    # MACs ausserhalb der veth-Zeile bleiben unangetastet - das ist der eigentliche Test.
    ('/interface wifi add configuration=cfg-main mac-address=D0:EA:11:4D:40:E0 name=w5-iot', None),
    ('/interface bridge set bridge admin-mac=04:F4:1C:B4:17:76 auto-mac=no', None),
    ('/interface ethernet set ether1 mac-address=AA:BB:CC:DD:EE:FF', None),
    # Zwei Zeilen: die Verankerung darf nicht ueber den Zeilenumbruch greifen.
    ('/interface veth add address=172.17.0.2/24 name=veth-cf\n'
     '/interface wifi add mac-address=D0:EA:11:4D:40:3A name=w2-iot',
     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   -> {e.code} {e.read().decode()[:400]}")
        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 anwenden(text):
    for _, rx, _ in REPLACES:
        text = re.sub(rx, r"\1" + MASK, text)
    return text


def selftest():
    fails = []
    for src, want in SELFTEST:
        got = anwenden(src)
        erwartet = src if want is None else want
        ok = got == erwartet
        if not ok:
            fails.append((src, erwartet, got))
        print(f"    [{'ok ' if ok else 'FAIL'}] {src.splitlines()[0][:78]}")
    return fails


def main():
    print("== Regex-Selbsttest (laeuft auch im Dry-Run) ==")
    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:
        sys.exit("!! NAUTOBOT_TOKEN nicht gesetzt")

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

    print(f"== ConfigReplace (Platform {PLATFORM}) ==")
    for name, rx, desc in REPLACES:
        nutzlast = {
            "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 nutzlast.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/", nutzlast)
            print(f"  + {name}")

    print(f"\n== ComplianceFeature/-Rule {FEATURE!r} ==")
    feat = find("/plugins/golden-config/compliance-feature", name=FEATURE)
    if feat:
        print(f"  = Feature {FEATURE}")
    elif DRY:
        print(f"  + Feature {FEATURE}  (dry-run)")
        feat = None
    else:
        feat = req("POST", "/plugins/golden-config/compliance-feature/",
                   {"name": FEATURE, "slug": FEATURE,
                    "description": "Container-Netz der AP-cloudflared-Connectoren (TP16.3)."})
        print(f"  + Feature {FEATURE}")

    if feat:
        rule = find("/plugins/golden-config/compliance-rule",
                    feature=feat["id"], platform=plat["id"])
    else:
        rule = None
    if rule:
        print(f"  = Rule    {FEATURE}  match_config={rule.get('match_config')!r}")
        if rule.get("match_config") != MATCH:
            print(f"            ⚠️ weicht ab - erwartet {MATCH!r}")
    elif DRY:
        print(f"  + Rule    {FEATURE}  match_config={MATCH!r}  (dry-run)")
    else:
        req("POST", "/plugins/golden-config/compliance-rule/", {
            "feature": feat["id"],
            "platform": plat["id"],
            "description": "Ohne diese Regel meldet ein geloeschtes veth-cf gar nichts (TP16.3).",
            "config_ordered": False,
            "config_type": "cli",
            "match_config": MATCH,
        })
        print(f"  + Rule    {FEATURE}")

    print("\n🔴 Danach ZWINGEND: python3 nautobot/gc-configplan.py --refresh-compliance")
    print("   (Backup -> Intended -> Compliance). Die Check-Zahl MUSS steigen: 27x5=135 -> 28x5=140.")
    print("   Bleibt sie bei 135, greift die Regel nicht - das ist ein Befund, kein Erfolg.")
    if DRY:
        print("\nDRY-RUN - nichts geschrieben. Mit --apply ausfuehren.")


if __name__ == "__main__":
    main()