bf-infra/nautobot/ingest-core-firewall.py

#!/usr/bin/env python3
r"""`/ip firewall filter` von `crs326-core` als Daten nach Nautobot schreiben.

WARUM DATEN UND NICHT TEMPLATEZEILEN
------------------------------------
36 Firewallregeln fest ins Jinja zu schreiben hiesse: jede TP5-Etappe braucht
einen Template-Edit. Genau die Reibung, wegen der die Governance in Verruf
geraten ist. Hier landen die Regeln als **Liste im local_config_context_data**
des Core; das Template rendert sie nur noch. Eine neue Regel = ein Listeneintrag.

⚠️ **Das ist ein RE-BASELINE-Werkzeug, kein Sync.** Es liest den Ist-Stand vom
Geraet und erklaert ihn zum Soll. Blind ausgefuehrt macht es Drift-Erkennung
wertlos — es gaebe nie Drift, weil das Soll immer dem Ist nachlaeuft.
Darum: **nur nach einer bewusst durchgefuehrten, verifizierten Aenderung
laufen lassen**, nie als Cronjob und nie "zum Grunmachen" der Compliance.

Attributreihenfolge = alphabetisch, exakt wie RouterOS `/export` sie schreibt
(gleiche Konvention wie der Rest des Templates, s. dessen Kopfkommentar).

    python3 nautobot/ingest-core-firewall.py            # Dry-Run
    python3 nautobot/ingest-core-firewall.py --apply
"""
import base64
import json
import os
import pathlib
import sys
import urllib.request

NB = os.environ.get("NAUTOBOT_URL", "http://10.40.20.154:8080")
DEVICE = "crs326-core"
SECRETS = pathlib.Path.home() / ".secrets" / "credentials.env"

# Felder, die RouterOS im Export NICHT schreibt (Laufzeit-/Interndaten).
# `hw-offload` gehoert dazu (2026-08-18 aufgeschlagen): die REST-API meldet es
# auf der defconf-fasttrack-Regel als true, `/ip firewall filter export` laesst
# es weg -- am Geraet gegengeprueft. Ohne diesen Eintrag meldet die Compliance
# ewige Falsch-Drift auf genau dieser einen Regel. Gleiche Klasse wie die
# `disabled=false`- und Boolean-Faelle weiter unten.
SKIP = {".id", "dynamic", "invalid", "bytes", "packets", "creation-time",
        "hw-offload"}


def secret(key):
    for line in SECRETS.read_text(errors="replace").splitlines():
        line = line.strip().removeprefix("export ")
        k, sep, v = line.partition("=")
        if sep and k.strip() == key:
            return v.strip().strip("'\"")
    raise SystemExit(f"{key} nicht in {SECRETS}")


def nb(path, method="GET", payload=None):
    tok = os.environ.get("NAUTOBOT_TOKEN") or sys.exit("NAUTOBOT_TOKEN nicht gesetzt")
    req = urllib.request.Request(
        f"{NB}{path}", method=method,
        data=json.dumps(payload).encode() if payload is not None else None,
        headers={"Authorization": f"Token {tok}", "Content-Type": "application/json",
                 "Accept": "application/json"})
    with urllib.request.urlopen(req, timeout=60) as r:
        raw = r.read()
        return json.loads(raw) if raw else {}


def device_rules(host, pw):
    auth = base64.b64encode(f"admin:{pw}".encode()).decode()
    req = urllib.request.Request(f"http://{host}/rest/ip/firewall/filter",
                                 headers={"Authorization": f"Basic {auth}",
                                          "Accept": "application/json"})
    with urllib.request.urlopen(req, timeout=30) as r:
        return json.load(r)


def is_true(val):
    """RouterOS-REST liefert Booleans als STRINGS ("true"/"false"). Ein
    naives `if rule.get("dynamic")` ist damit fuer "false" wahr und filtert
    alles weg — am 2026-08-17 genau so passiert (0 statt 36 Regeln)."""
    return val is True or (isinstance(val, str) and val.lower() == "true")


def render(rule):
    """Ein REST-Objekt -> die `add ...`-Zeile, wie /export sie schreibt."""
    parts = []
    for key in sorted(rule):
        if key in SKIP:
            continue
        val = rule[key]
        # `/export` schreibt Default-Werte NICHT. REST liefert sie trotzdem mit.
        # Konkret `disabled=false`: der Export laesst es weg, `disabled=yes` steht
        # dagegen drin und ist semantisch wichtig. Ohne diese Zeile meldet die
        # Compliance ewige Falsch-Drift auf genau einer Regel (2026-08-17 beim
        # ersten Lauf der neuen Rule aufgeschlagen: 75/76).
        if key == "disabled" and not is_true(val):
            continue
        # REST liefert Booleans als "true"/"false", /export schreibt "yes"/"no".
        # Ohne diese Uebersetzung meldet die Compliance ewige Falsch-Drift auf
        # `hw-offload` (2026-08-17 im Vergleich gegen den echten Export gefunden).
        if isinstance(val, bool):
            val = "yes" if val else "no"
        elif isinstance(val, str) and val.lower() in ("true", "false"):
            val = "yes" if val.lower() == "true" else "no"
        val = str(val)
        # RouterOS quotet NUR bei Leerzeichen. Nicht bei "!" (siehe
        # `connection-nat-state=!dstnat`) und nicht bei Kommas
        # (`dst-port=8123,1883,8883`) — beides steht unquoted im Export.
        if " " in val:
            val = f'"{val}"'
        parts.append(f"{key}={val}")
    return "add " + " ".join(parts)


def main():
    apply_ = "--apply" in sys.argv
    dev = nb(f"/api/dcim/devices/?name={DEVICE}&depth=1")["results"][0]
    host = dev["primary_ip4"]["address"].split("/")[0]

    rules = [r for r in device_rules(host, secret("CRS326_CORE_PASS"))
             if not is_true(r.get("dynamic"))]
    lines = [render(r) for r in rules]

    print(f"== {DEVICE}: {len(lines)} statische Filterregeln ==")
    for ln in lines:
        print("   " + (ln[:150] + " ..." if len(ln) > 150 else ln))

    ctx = dict(dev.get("local_config_context_data") or {})
    if ctx.get("firewall_filter") == lines:
        print("\n  = unveraendert, nichts zu tun.")
        return
    ctx["firewall_filter"] = lines
    if not apply_:
        print(f"\nDRY-RUN — {len(lines)} Zeilen wuerden nach "
              f"local_config_context_data.firewall_filter geschrieben. Mit --apply.")
        return
    nb(f"/api/dcim/devices/{dev['id']}/", "PATCH", {"local_config_context_data": ctx})
    print(f"\n  ~ geschrieben. Jetzt: gc-configplan.py --refresh-compliance")


if __name__ == "__main__":
    main()