bf-infra/nautobot/tp45-agh-firewall.py

#!/usr/bin/env python3
r"""TP45 Schicht A — die AGH-DNS-Freigaben im Core-`firewall_filter`-Soll pflegen.

WAS HIER PASSIERT
-----------------
Die vier `TP5-E2: LAN->AGH dns*`-Regeln des Cores stehen als LITERALE im
`local_config_context_data.firewall_filter` (Nautobot) und werden vom GC-Template
nur gerendert. Zieht der Resolver um, muss das Soll mit — sonst meldet die
Compliance Drift, sobald das Geraet die neue Adresse traegt.

  add <ip> [...]      zwei Regeln (udp+tcp) fuer jede IP, direkt hinter den
                      bestehenden AGH-Block einsortiert
  remove <ip> [...]   genau diese zwei Regeln wieder entfernen (= Rueckbau)

WARUM KEIN `ingest-core-firewall.py`
------------------------------------
Das ist ein RE-BASELINE-Werkzeug: es liest den Ist-Stand vom Geraet und erklaert
ihn zum Soll. Fuer eine *geplante* Aenderung ist es das falsche Werkzeug — es
wuerde die Aenderung mit sich selbst absegnen. Hier wird das Soll explizit
gesetzt, das Geraet zieht per `gc-configplan.py --job tp45-agh-fw` nach.

    python3 nautobot/tp45-agh-firewall.py add 10.40.20.30 10.40.20.31      # Dry-Run
    python3 nautobot/tp45-agh-firewall.py add 10.40.20.30 10.40.20.31 --apply

REIHENFOLGE IST SEMANTISCH — die ComplianceRule steht auf `config_ordered`. Neue
Regeln landen hinter der bestehenden `.37`-tcp-Zeile, also weiterhin VOR der
`TP5: log-only`-Zeile und damit vor dem finalen `drop`.
"""
import difflib
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"
COMMENT_UDP = "TP5-E2: LAN->AGH dns"
COMMENT_TCP = "TP5-E2: LAN->AGH dns-tcp"
ANCHOR = 'comment="TP5-E2: LAN->AGH dns-tcp" dst-address='


def token():
    tok = os.environ.get("NAUTOBOT_TOKEN")
    if tok:
        return tok
    # Fallback: aus dem (gitignored) Stack-File, Wert wird nie ausgegeben.
    yml = (pathlib.Path(__file__).resolve().parent.parent
           / "docs/nas-vlan20/stacks/210-nautobot.yml")
    for line in yml.read_text(errors="replace").splitlines():
        key, sep, val = line.strip().lstrip("- ").partition("=")
        if sep and key.strip() == "NAUTOBOT_SUPERUSER_API_TOKEN":
            return val.strip().strip("'\"")
    raise SystemExit("NAUTOBOT_TOKEN nicht gesetzt und nicht in 210-nautobot.yml")


def nb(path, method="GET", payload=None):
    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 {token()}", "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 rules_for(ip):
    # Attributreihenfolge alphabetisch, exakt wie RouterOS /export sie schreibt —
    # dieselbe Konvention wie ingest-core-firewall.py und das Template.
    return [
        f'add action=accept chain=forward comment="{COMMENT_UDP}" '
        f'dst-address={ip} dst-port=53 protocol=udp',
        f'add action=accept chain=forward comment="{COMMENT_TCP}" '
        f'dst-address={ip} dst-port=53 protocol=tcp',
    ]


def is_agh(rule, ip):
    return f"dst-address={ip}" in rule and "TP5-E2: LAN->AGH dns" in rule


def main():
    argv = sys.argv[1:]
    if len(argv) < 3 or argv[0] not in ("add", "remove"):
        raise SystemExit(__doc__.strip().splitlines()[-1])
    mode = argv[0]
    ips = [a for a in argv[1:] if not a.startswith("--")]
    apply_ = "--apply" in argv

    dev = nb(f"/api/dcim/devices/?name={DEVICE}")["results"][0]
    ctx = dict(dev.get("local_config_context_data") or {})
    lines = list(ctx.get("firewall_filter") or [])
    if not lines:
        raise SystemExit("firewall_filter ist leer — falsches Geraet?")

    new = list(lines)
    for ip in ips:
        present = [l for l in new if is_agh(l, ip)]
        if mode == "add":
            if present:
                print(f"   {ip}: schon vorhanden ({len(present)} Regeln) — uebersprungen")
                continue
            # hinter die letzte bestehende AGH-Zeile einsortieren
            idx = max((i for i, l in enumerate(new) if ANCHOR in l), default=None)
            if idx is None:
                raise SystemExit(f"   kein AGH-Anker ({ANCHOR}...) im Soll gefunden")
            new[idx + 1:idx + 1] = rules_for(ip)
            print(f"   {ip}: +2 Regeln nach Index {idx}")
        else:
            if len(present) != 2:
                raise SystemExit(f"   {ip}: {len(present)} statt 2 Regeln gefunden — Abbruch")
            new = [l for l in new if not is_agh(l, ip)]
            print(f"   {ip}: -2 Regeln entfernt")

    if new == lines:
        print("\n  = unveraendert, nichts zu tun.")
        return

    print(f"\n== Diff ({len(lines)} -> {len(new)} Zeilen) ==")
    for d in difflib.unified_diff(lines, new, "IST", "SOLL", lineterm="", n=1):
        print("   " + d[:170])

    if not apply_:
        print(f"\nDRY-RUN — mit --apply nach "
              f"local_config_context_data.firewall_filter schreiben.")
        return
    ctx["firewall_filter"] = new
    nb(f"/api/dcim/devices/{dev['id']}/", "PATCH", {"local_config_context_data": ctx})
    print("\n  ~ geschrieben. Geraet zieht nach per: "
          "python3 nautobot/gc-configplan.py --job tp45-agh-fw --apply")


if __name__ == "__main__":
    main()