bf-infra/nautobot/gc-snapshot.py

#!/usr/bin/env python3
"""
TP36-Live — Live-Gerätedaten als CustomFields auf den RouterOS-dcim.Device.

Legt (idempotent, get-or-create) fünf CustomFields auf `dcim.device` an und
befüllt sie für jedes RouterOS-Gerät. Die Werte kommen AUSSCHLIESSLICH aus
Nautobot selbst — es wird KEIN Gerät kontaktiert:

  ros_version         aus der GoldenConfig-`backup_config` ("by RouterOS X.Y.Z")
  dhcp_server_count   gezählte Einträge der Sektion "/ip dhcp-server" im Backup
  dns_upstream_mode   aus der "/ip dns"-Sektion des Backups: doh | plain | none
  compliance_pct      GC config-compliance: compliant / total * 100 (gerundet)
  mgmt_ip             device.primary_ip4 (ohne /prefix)

Warum aus Nautobot statt live vom Gerät: der MGMT-/REST-Weg (VLAN99:80) ist
nicht aus jedem Vantage erreichbar; die GC-Backups werden ohnehin 2×/Tag per
SSH gezogen und sind die Doku-SSoT. Damit ist der Snapshot cron-fest und
braucht keine Geräte-Credentials. Aktualität steht in `backup_last_success_date`.

Der Backup-Config-Text wird nur INTERN geparst (Zähler/Modus abgeleitet) und
NIE ausgegeben — er kann gescrubte Reste enthalten (PSK-Leak-Regel).

Token-Quelle: docs/nas-vlan20/stacks/210-nautobot.yml (gitignored),
Schlüssel NAUTOBOT_SUPERUSER_API_TOKEN.

Aufruf:
  python3 nautobot/gc-snapshot.py            # Dry-Run (default) — zeigt nur an
  python3 nautobot/gc-snapshot.py --commit   # legt CustomFields an + patcht Devices
"""
import json
import os
import re
import sys
import urllib.request
import urllib.parse
import urllib.error
from pathlib import Path

NB = os.environ.get("NAUTOBOT_URL", "http://10.40.20.30:8080")
REPO = Path(__file__).resolve().parent.parent
YML = REPO / "docs" / "nas-vlan20" / "stacks" / "210-nautobot.yml"
COMMIT = "--commit" in sys.argv


def load_token() -> str:
    for line in YML.read_text().splitlines():
        m = re.search(r"NAUTOBOT_SUPERUSER_API_TOKEN[=:]\s*([^\s\"']+)", line)
        if m:
            return m.group(1)
    sys.exit(f"NAUTOBOT_SUPERUSER_API_TOKEN nicht in {YML} gefunden")


TOKEN = load_token()
HDRS = {"Authorization": f"Token {TOKEN}", "Accept": "application/json",
        "Content-Type": "application/json"}


def req(method: str, path: str, body=None):
    data = json.dumps(body).encode() if body is not None else None
    r = urllib.request.Request(f"{NB}{path}", data=data, headers=HDRS, method=method)
    try:
        with urllib.request.urlopen(r, timeout=20) as resp:
            return resp.status, json.loads(resp.read().decode() or "{}")
    except urllib.error.HTTPError as e:
        return e.code, json.loads(e.read().decode() or "{}")


# --------------------------------------------------------------- CustomFields
# (key, label, type, description). content_types wird ergänzt.
CUSTOM_FIELDS = [
    ("ros_version", "RouterOS-Version", "text",
     "RouterOS-Version aus der GC-backup_config (by RouterOS …)."),
    ("compliance_pct", "GC-Compliance %", "integer",
     "Golden-Config config-compliance: compliant/total in Prozent."),
    ("dhcp_server_count", "DHCP-Server", "integer",
     "Anzahl Einträge der Sektion /ip dhcp-server im GC-Backup."),
    ("mgmt_ip", "Mgmt-IP", "text",
     "device.primary_ip4 ohne Prefix (VLAN99-Management)."),
    ("dns_upstream_mode", "DNS-Upstream-Modus", "text",
     "Aus /ip dns des GC-Backups: doh | plain | none."),
]
CF_KEYS = [c[0] for c in CUSTOM_FIELDS]


def ensure_custom_fields():
    """get-or-create die fünf CFs; bei Bestand: content_type dcim.device sicherstellen."""
    st, d = req("GET", "/api/extras/custom-fields/?limit=0")
    have = {r["key"]: r for r in d.get("results", [])}
    for key, label, typ, desc in CUSTOM_FIELDS:
        ex = have.get(key)
        if ex:
            cts = ex.get("content_types", [])
            ok = "dcim.device" in cts
            print(f"  = CF '{key}' existiert ({ex['id']}){'' if ok else '  ⚠ ohne dcim.device!'}")
            if not ok and COMMIT:
                req("PATCH", f"/api/extras/custom-fields/{ex['id']}/",
                    {"content_types": sorted(set(cts) | {"dcim.device"})})
                print(f"    + dcim.device ergänzt")
            continue
        payload = {"key": key, "label": label, "type": typ, "description": desc,
                   "content_types": ["dcim.device"], "grouping": "TP36-Live",
                   "required": False}
        if not COMMIT:
            print(f"  + CF '{key}' ({typ})  [DRY-RUN]")
            continue
        st, r = req("POST", "/api/extras/custom-fields/", payload)
        print(f"  {'+' if st in (200,201) else '!'} CF '{key}'"
              + (f" angelegt ({r.get('id')})" if st in (200, 201)
                 else f" FEHLER {st}: {json.dumps(r)[:200]}"))


# --------------------------------------------------------------- Backup-Parser
def parse_backup(cfg: str):
    """Werte aus dem flachen RouterOS-Export (eine '/…'-Zeile pro Kommando)."""
    ver = ""
    m = re.search(r"by RouterOS ([0-9][0-9A-Za-z.\-]*)", cfg)
    if m:
        ver = m.group(1)
    # DHCP-Server = Zeilen '/ip dhcp-server add …' (NICHT lease/network/option)
    dhcp = len(re.findall(r"(?m)^\s*/ip dhcp-server add\b", cfg))
    # DNS-Modus aus der '/ip dns set …'-Zeile
    mode = "none"
    md = re.search(r"(?m)^\s*/ip dns set\b.*", cfg)
    if md:
        line = md.group(0)
        doh = re.search(r"use-doh-server=([^\s]+)", line)
        srv = re.search(r"\bservers=([^\s]+)", line)
        if doh and doh.group(1) not in ("", '""'):
            mode = "doh"
        elif srv and srv.group(1) not in ("", '""'):
            mode = "plain"
    return ver, dhcp, mode


def compliance_pct(device_name: str):
    q = urllib.parse.urlencode({"device": device_name, "limit": 0, "depth": 0})
    st, d = req("GET", f"/api/plugins/golden-config/config-compliance/?{q}")
    rows = d.get("results", [])
    if not rows:
        return None
    ok = sum(1 for r in rows if r.get("compliance"))
    return round(ok / len(rows) * 100)


# ---------------------------------------------------------------------- main
def routeros_devices():
    q = urllib.parse.urlencode({"limit": 0, "depth": 1, "platform": "RouterOS"})
    st, d = req("GET", f"/api/dcim/devices/?{q}")
    devs = []
    for r in d.get("results", []):
        ip = r.get("primary_ip4")
        ip = ip.get("address") if isinstance(ip, dict) else ip
        if not ip:
            continue  # tote Altgeräte (cudy/sw) ohne IP überspringen
        devs.append((r["id"], r["name"], ip.split("/")[0]))
    return sorted(devs, key=lambda x: x[1])


def main():
    print(f"== TP36-Live gc-snapshot  ({'COMMIT' if COMMIT else 'DRY-RUN'})  @ {NB} ==")

    print("\n[1] CustomFields sicherstellen")
    ensure_custom_fields()

    print("\n[2] Golden-Config-Backups laden")
    st, gc = req("GET", "/api/plugins/golden-config/golden-config/?depth=1&limit=0")
    backups = {}
    for r in gc.get("results", []):
        dev = r.get("device")
        dev = dev.get("name") if isinstance(dev, dict) else dev
        backups[dev] = (r.get("backup_config") or "", r.get("backup_last_success_date") or "?")

    print("\n[3] Snapshot je Gerät")
    print(f"    {'Gerät':16} {'ros':9} {'cmpl%':>5} {'dhcp':>4} {'dns':6} {'mgmt-ip':15} backup")
    for dev_id, name, mgmt in routeros_devices():
        cfg, when = backups.get(name, ("", "—"))
        ver, dhcp, dns = parse_backup(cfg) if cfg else ("?", 0, "?")
        pct = compliance_pct(name)
        vals = {
            "ros_version": ver or None,
            "compliance_pct": pct,
            "dhcp_server_count": dhcp,
            "mgmt_ip": mgmt,
            "dns_upstream_mode": dns,
        }
        pct_s = "—" if pct is None else str(pct)
        print(f"    {name:16} {ver or '?':9} {pct_s:>5} {dhcp:>4} {dns:6} {mgmt:15} {when[:10]}")
        if COMMIT:
            st, r = req("PATCH", f"/api/dcim/devices/{dev_id}/", {"custom_fields": vals})
            if st not in (200, 201):
                print(f"      ! PATCH FEHLER {st}: {json.dumps(r)[:200]}")

    print(f"\n== fertig ({'COMMIT — geschrieben' if COMMIT else 'DRY-RUN — nichts geschrieben'}) ==")


if __name__ == "__main__":
    main()