#!/usr/bin/env python3 """BF-Netzwerk -> Nautobot Golden Config: manuelle ConfigPlans anlegen und deployen. WOFUER ------ Bis 2026-08-16 wurden alle 17 ConfigPlans dieses Projekts ad hoc per curl gebaut. Das hat zweimal wehgetan (s. GOVERNANCE unten), deshalb hier einmal sauber als Skript. Ein "manual" ConfigPlan ist der einzige Weg, etwas vom Geraet zu ENTFERNEN: `plan_type=missing` erzeugt ausschliesslich `add`-Zeilen, `intended` waere ein Voll-Push (verboten, s. Regel 1). Alles, was Drift vom Typ `extra` beseitigt, ist also zwangslaeufig ein manueller Plan. GOVERNANCE (docs/teilprojekte.md TP22 -- verbindlich, nicht optional) -------------------------------------------------------------------- 1. Plans NUR feature-/manual-scoped, NIEMALS "full intended". Ein Voll-Push schriebe woertlich `passphrase="<>"` auf die APs (= alle WLAN-Clients raus) bzw. `private-key="<>"` auf den Core (= WG-Hub tot). -> Dieses Skript weigert sich, ein config_set mit `<>` zu senden. 2. Das AP-Template wird NIE zeilenweise auf ein laufendes Geraet gepusht (setzt zwischendurch vlan-filtering=no -> reisst den Mgmt-Pfad ab). 3. Vor jedem Push ein `/system backup` auf dem Zielgeraet. -> Dieses Skript prueft das nicht, es kann nur daran erinnern. Machen. ZWEI TEUER BEZAHLTE LEHREN, DIE HIER FEST EINGEBAUT SIND -------------------------------------------------------- * Nautobot-REST sortiert Sammelabfragen ALPHABETISCH. Am 2026-08-13 wurden 4 Device-UUIDs positional zugeordnet -> der Deploy ging an ap1 statt an den Core. Schadlos nur, weil RouterOS die Kommandos ablehnte. -> `resolve_device()` fragt IMMER genau einen Namen ab und besteht auf genau einem Treffer; nach dem Anlegen wird `plan.device.display` gegen den Zielnamen assertet. * Job-Status ist KEIN Wirksamkeitsnachweis. Der Deploy-Job meldet `SUCCESS`, waehrend der Plan auf `Failed` steht -- und selbst ein `Completed`-Plan beweist nur, dass Netmiko die Zeilen abgesetzt hat, nicht dass das Geraet sie angenommen hat. -> Dieses Skript liest am Ende den PLAN-Status und sagt explizit, dass die Gegenprobe am Geraet noch aussteht. WEITERE ROUTEROS-FALLEN (Memory `ref_gc_push_gotchas_routeros`) --------------------------------------------------------------- * Keine `print`-Zeilen in Config Plans -- `:log info` verwenden. * Selbstprotokollierendes `:do { ... } on-error={ :log info "..." }` statt nackter Kommandos: ein Netmiko-Fehlschlag hinterlaesst sonst Teil-Zustand (anders als NAPALM, das atomar zurueckrollt). * `napalm-ros` ist READ-ONLY. Der Push laeuft nur, weil in den Golden-Config- Settings ein `custom_dispatcher` auf `NetmikoMikrotikRouteros` zeigt. BENUTZUNG --------- export NAUTOBOT_TOKEN=... python3 nautobot/gc-configplan.py --job drift-cleanup # Dry-Run python3 nautobot/gc-configplan.py --job drift-cleanup --apply # anlegen+deployen python3 nautobot/gc-configplan.py --job drift-cleanup --apply --no-deploy """ import argparse import json import os import sys import time import urllib.error import urllib.parse import urllib.request BASE = "http://10.40.20.30:8080/api" TOKEN = os.environ.get("NAUTOBOT_TOKEN", "") # Status-UUIDs fuer nautobot_golden_config.configplan (2026-08-16 aus der API gelesen) STATUS_NOT_APPROVED = "4d2d08bf-3bf2-4370-8c8b-248dff2618ab" STATUS_APPROVED = "7869b4b9-542d-4ff7-a846-92a04943e75d" JOB_DEPLOY_CONFIG_PLANS = "31c3f093-eeec-43fa-ba4a-0aed2f3c67aa" JOB_GENERATE_CONFIG_PLANS = "f93bfa53-2332-4fa2-a6ab-5921c802fcee" # Die Reihenfolge ist NICHT beliebig: die Compliance vergleicht die intended config # gegen das letzte BACKUP, nicht gegen das Live-Geraet. Wer nach einer Geraetaenderung # direkt "Compliance" laufen laesst, misst den Stand von vorgestern (TP22-Falle). REFRESH_SEQUENCE = [ ("Backup Configurations", "a48b8c7b-ad4b-4909-8344-1b4e758e75a8"), ("Generate Intended Configurations", "afe60af2-bcd1-4166-8976-1a78f49bc228"), ("Perform Configuration Compliance", "ed3bcb19-5c84-4ffd-a7ba-564703213997"), ] MASK = "<>" # --------------------------------------------------------------------------- HTTP 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=60) as resp: raw = resp.read() return json.loads(raw) if raw else {} except urllib.error.HTTPError as e: detail = e.read().decode()[:600] print(f"!! {method} {url}\n payload={json.dumps(data)[:300]}\n -> {e.code} {detail}") raise # ------------------------------------------------------------------- Bausteine def resolve_device(name): """Genau EIN Geraet per Name. Nie positional aus einer Sammelabfrage.""" q = urllib.parse.urlencode({"name": name}) res = req("GET", f"/dcim/devices/?{q}").get("results", []) if len(res) != 1: raise SystemExit(f"FEHLER: '{name}' liefert {len(res)} Treffer, erwartet genau 1") dev = res[0] if dev.get("name") != name: raise SystemExit(f"FEHLER: API lieferte '{dev.get('name')}' statt '{name}'") return dev["id"] def guard_config_set(config_set, device_name): """Regel 1: nie eine Maske aufs Geraet schreiben.""" if MASK in config_set: raise SystemExit( f"ABBRUCH: config_set fuer {device_name} enthaelt {MASK}. " "Das wuerde das echte Secret auf dem Geraet ueberschreiben (TP22-Regel 1)." ) for line in config_set.splitlines(): if line.strip().startswith("print") or " print " in f" {line} ": raise SystemExit(f"ABBRUCH: `print` in Config Plan ({device_name}) -- `:log info` verwenden.") def create_plans(device_names, config_set, change_control_id, dry): """Plaene ueber den Job 'Generate Config Plans' anlegen. WICHTIG: `POST /plugins/golden-config/config-plan/` ist **405 Method Not Allowed** -- das ViewSet ist read-only. ConfigPlans entstehen ausschliesslich ueber diesen Job. Bei `plan_type=manual` landet der Inhalt des `commands`-TextVar als `config_set` auf jedem ausgewaehlten Geraet (ein Plan pro Geraet). """ guard_config_set(config_set, ",".join(device_names)) dev_ids = [resolve_device(n) for n in device_names] if dry: print(f" [dry-run] Job 'Generate Config Plans' fuer {', '.join(device_names)} ({change_control_id})") for line in config_set.strip().splitlines(): print(f" | {line}") return [] run = req( "POST", f"/extras/jobs/{JOB_GENERATE_CONFIG_PLANS}/run/", { "data": { "device": dev_ids, "plan_type": "manual", "commands": config_set, "change_control_id": change_control_id, } }, ) jr = (run.get("job_result") or run).get("id") wait_job(jr, "Generate Config Plans") # Plaene ueber die change_control_id einsammeln -- nie positional aus einer # Sammelabfrage zuordnen (Lehre aus TP21.1: REST sortiert alphabetisch). q = urllib.parse.urlencode({"change_control_id": change_control_id, "depth": 1}) found = req("GET", f"/plugins/golden-config/config-plan/?{q}").get("results", []) by_name = {} for p in found: name = (p.get("device") or {}).get("display") by_name[name] = p["id"] plans = [] for name in device_names: pid = by_name.get(name) if not pid: raise SystemExit(f"ABBRUCH: kein Plan fuer '{name}' unter ccid '{change_control_id}' gefunden") # Gegenprobe am Objekt selbst, nicht am Suchergebnis. check = req("GET", f"/plugins/golden-config/config-plan/{pid}/?depth=1") got = (check.get("device") or {}).get("display") if got != name: raise SystemExit(f"ABBRUCH: Plan {pid} haengt an '{got}', erwartet '{name}'. NICHT deployen.") if check.get("config_set", "").strip() != config_set.strip(): raise SystemExit(f"ABBRUCH: config_set von Plan {pid} weicht ab -- nicht deployen.") print(f" Plan {pid} -> '{name}' verifiziert (Device + config_set)") plans.append((name, pid)) return plans def wait_job(jr, label): print(f" Job '{label}' gestartet, JobResult {jr}") for _ in range(60): time.sleep(5) st = req("GET", f"/extras/job-results/{jr}/").get("status") st = st.get("value") if isinstance(st, dict) else st if str(st).upper() in ("SUCCESS", "FAILURE", "ERRORED", "REVOKED", "COMPLETED", "FAILED"): print(f" Job-Status: {st} (!! kein Wirksamkeitsnachweis)") return st print(" !! Job-Timeout nach 5 Min -- Status manuell pruefen") return None def approve(plan_id): req("PATCH", f"/plugins/golden-config/config-plan/{plan_id}/", {"status": STATUS_APPROVED}) def deploy(plan_ids): """Deploy-Job starten und auf das JobResult warten.""" run = req( "POST", f"/extras/jobs/{JOB_DEPLOY_CONFIG_PLANS}/run/", {"data": {"config_plan": plan_ids}}, ) jr = (run.get("job_result") or run).get("id") wait_job(jr, "Deploy Config Plans") return jr def refresh_compliance(): """Backup -> Intended -> Compliance, strikt in dieser Reihenfolge, dann Bericht.""" for label, job_id in REFRESH_SEQUENCE: print(f"\n-- {label}") run = req("POST", f"/extras/jobs/{job_id}/run/", {"data": {}}) jr = (run.get("job_result") or run).get("id") wait_job(jr, label) print("\n== Compliance ==") res = req("GET", "/plugins/golden-config/config-compliance/?limit=500&depth=1") rows = res.get("results", []) bad = [r for r in rows if not r.get("compliance")] print(f" {len(rows) - len(bad)}/{len(rows)} compliant") for r in bad: dev = (r.get("device") or {}).get("display") rule = r.get("rule") rule = rule.get("display") if isinstance(rule, dict) else rule print(f" !! {dev} | {rule}") if r.get("extra"): print(f" extra: {r['extra'].strip()[:200]}") if r.get("missing"): print(f" missing: {r['missing'].strip()[:200]}") if not bad: print(" keine Drifts.") return bad def plan_status(plan_id): p = req("GET", f"/plugins/golden-config/config-plan/{plan_id}/?depth=1") st = p.get("status") return st.get("display") if isinstance(st, dict) else st # ------------------------------------------------------------------------ Jobs def job_drift_cleanup(): """TP22-Compliance-Backlog: die 4 verbliebenen `extra`-Drifts entfernen. ap1/ap2: Werks-`defconf` DNS-Static `router.lan -> 192.168.88.1`. Auf beiden APs der EINZIGE statische Eintrag; die IP existiert im Netz nicht. Der Core hat sein eigenes, korrektes `router.lan -> 10.40.99.1` (im Template, compliant). crs326-branch: Bring-up-Rest `vlan10` + `10.40.10.2`. Reihenfolge Adresse -> Interface. Beruehrt KEINEN Client-Traffic: `/interface bridge vlan` bleibt unangetastet (`bridge` steht dort statisch als tagged-Member von VLAN10), es faellt nur die L3-Praesenz des Geraets in VLAN10 weg. Die Deploy-Session selbst laeuft ueber vlan99/10.40.99.2, also nicht ueber den abgebauten Pfad. """ ap_set = ( ':do { /ip dns static remove [find address=192.168.88.1] } ' 'on-error={ :log info "drift: ap dns-static remove hat geworfen" }\n' ':log info ("drift: rest-defconf-static=" . ' '[:tostr [:len [/ip dns static find address=192.168.88.1]]])' ) branch_set = ( ':do { /ip address remove [find address="10.40.10.2/24"] } ' 'on-error={ :log info "drift: branch address remove hat geworfen" }\n' ':do { /interface vlan remove [find name="vlan10"] } ' 'on-error={ :log info "drift: branch vlan10 remove hat geworfen" }\n' ':log info ("drift: branch rest-vlan10=" . ' '[:tostr [:len [/interface vlan find name="vlan10"]]] . " rest-addr=" . ' '[:tostr [:len [/ip address find address~"10.40.10."]]])' ) # (Geraeteliste, commands, change_control_id) -- ap1/ap2 teilen sich das # config_set, koennen also in einem Job-Lauf erzeugt werden. return [ (["ap1", "ap2"], ap_set, "TP22-drift-aps-defconf-dns"), (["crs326-branch"], branch_set, "TP22-drift-branch-temp-vlan10"), ] def job_mgmt_acl(): """TP16.1 (2026-08-17): Mgmt-Dienste + Address-ACL auf allen vier Geraeten. WARUM: seit dem Cloudflare-Mesh kommt der Remote-Zugriff aus 100.96.0.0/12. Eine ACL, die nur 10.40.0.0/16 kennt, laesst Ping durch und sperrt SSH/REST/MikroMCP aus -- genau das ist auf dem Core passiert. Umgekehrt fuhren branch/ap1/ap2 ssh/www/winbox/api/ftp/ telnet komplett OHNE Address-Beschraenkung; winbox war auf allen vier offen. KEIN LOCKOUT-RISIKO: der Deploy laeuft von Nautobot (10.40.20.30) aus, also aus 10.40.0.0/16 -- die Adresse bleibt in jeder gesetzten ACL enthalten. RouterOS wendet `address=` ohnehin erst auf NEUE Verbindungen an, die laufende Session ueberlebt. SCHREIBWEISE: `/ip service set ssh ...` mit dem Dienstnamen, NICHT `[find name=ssh]` -- letzteres trifft auch den dynamischen Verbindungs-Eintrag (`D c ssh ... local `) und quittiert mit `failure: this is configured elsewhere` (TP16.1-Befund). """ acl = "10.40.0.0/16,100.96.0.0/12" def line(svc, attr): return ( f":do {{ /ip service set {svc} {attr} }} " f'on-error={{ :log info "mgmt-acl: {svc} hat geworfen" }}' ) # Erst die Abschaltungen (Angriffsflaeche runter), dann die ACLs. full_set = "\n".join( [line(s, "disabled=yes") for s in ("ftp", "telnet", "api-ssl")] + [line(s, f"address={acl}") for s in ("ssh", "www", "winbox", "api")] + [ ':log info ("mgmt-acl: ssh=" . [/ip service get [find name=ssh and !dynamic] address]' ' . " winbox=" . [/ip service get [find name=winbox and !dynamic] address]' ' . " telnet-disabled=" . [:tostr [/ip service get [find name=telnet and !dynamic] disabled]])' ] ) # Der Core hat ftp/telnet/api-ssl bereits aus und ssh/www bereits korrekt -- # ihm fehlen nur winbox (war offen) und die Mesh-Range auf api. core_set = "\n".join( [line(s, f"address={acl}") for s in ("winbox", "api")] + [ ':log info ("mgmt-acl: winbox=" . [/ip service get [find name=winbox and !dynamic] address]' ' . " api=" . [/ip service get [find name=api and !dynamic] address])' ] ) return [ (["crs326-branch", "ap1", "ap2"], full_set, "TP16.1-mgmt-acl-mesh"), (["crs326-core"], core_set, "TP16.1-mgmt-acl-mesh-core"), ] def job_tp42_dhcp_domain(): """TP42 (2026-09-02): DHCP-Search-Domain der 7 Netze emsenhuber.net -> bf.emsenhuber.net. WARUM: das GC-Template verteilte `domain=emsenhuber.net` (bare), die internen A-Records leben aber unter `bf.emsenhuber.net`. Folge: ein bare `macmini` loeste NICHT auf (nur der volle FQDN). Zielschema-Entscheid (User 2026-09-02) = `bf.emsenhuber.net`. Das Template (Z.133) ist bereits auf `bf.emsenhuber.net` gehoben; dieser Job zieht das laufende Geraet nach. MODIFY, kein add: das Netz existiert -> `set [find]`, nicht die `add`-Zeile der intended config (die wuerde `already have such address` werfen). `[find]` ohne Filter trifft alle 7 Netze. KEIN LOCKOUT-RISIKO: `domain` ist eine DHCP-Option, sie beruehrt weder Routing noch Mgmt-Zugang; Clients ziehen sie erst beim naechsten Lease-Renew (lease-time 1d). """ core_set = "\n".join([ ':do { /ip dhcp-server network set [find] domain=bf.emsenhuber.net } ' 'on-error={ :log info "tp42-dhcp: set geworfen" }', ':log info ("tp42-dhcp: bf-domains=" . [:tostr [:len [/ip dhcp-server network ' 'find where domain="bf.emsenhuber.net"]]] . "/7")', ]) return [(["crs326-core"], core_set, "TP42-dhcp-domain-bf")] def job_tp38_oob(): """TP38 (2026-09-15): OOB-MGMT-Port (ether23 untagged VLAN99) + Branch-Input-Firewall. Zwei Geraete, zwei config_sets: * branch: ether23 untaggen, VLAN99 um untagged=ether23 erweitern, 8 Input-Regeln (deny-by-default). Die Regeln werden in der Reihenfolge angelegt, in der das Template sie rendert — established zuerst, drop zuletzt — damit kein LOCKOUT: die laufende Nautobot-SSH-Session (10.40.20.30) matcht auf Regel 1 (established), bevor Regel 8 (drop) greift. * core: ether23 untaggen, VLAN99 untagged=ether23, EINE Regel "OOB->core mgmt" mit place-before vor die Log-Regel (sonst landet sie hinter dem finalen drop und waere wirkungslos + non-compliant). KEIN LOCKOUT-RISIKO: ether23 ist auf beiden Geraeten frei (running=false); der VLAN99-Umbau aendert nur `untagged`, nicht `tagged`; der Deploy laeuft von Nautobot (10.40.20.30), das in jeder gesetzten Firewall-Freigabe enthalten bleibt. """ bridge_set = ( ':do { /interface bridge port set [find interface=ether23] pvid=99 ' 'frame-types=admit-only-untagged-and-priority-tagged comment="TP38-OOB-MGMT" } ' 'on-error={ :log info "tp38: ether23 set geworfen" }' ) vlan_set = ( ':do { /interface bridge vlan set [find vlan-ids=99] untagged=ether23 } ' 'on-error={ :log info "tp38: vlan99 untagged geworfen" }' ) branch_fw = "\n".join([ 'add action=accept chain=input comment="TP38: established" connection-state=established,related,untracked', 'add action=drop chain=input comment="TP38: invalid" connection-state=invalid', 'add action=accept chain=input comment="TP38: icmp" protocol=icmp', 'add action=accept chain=input comment="TP38: mndp" dst-port=5678 protocol=udp', 'add action=accept chain=input comment="TP38: mgmt LAN" dst-port=22,80,8291,8728 protocol=tcp src-address=10.40.0.0/16', 'add action=accept chain=input comment="TP38: mgmt mesh" dst-port=22,80,8291,8728 protocol=tcp src-address=100.96.0.0/12', 'add action=log chain=input comment="TP38: input log-only" log-prefix=TP38-BRANCH-WOULD-DROP', 'add action=drop chain=input comment="TP38: input deny-by-default"', ]) branch_set = "\n".join([ bridge_set, vlan_set, *[f"/ip firewall filter {ln}" for ln in branch_fw.splitlines()], ':log info ("tp38: branch pvid=" . [:tostr [/interface bridge port get [find interface=ether23] pvid]]' ' . " fw=" . [:tostr [:len [/ip firewall filter find where comment~"TP38"]]] . "/8")', ]) # ⚠️ `add ... place-before=[find ...]` wird ueber Netmiko IGNORIERT (Regel landet am # Ketten-Ende, also HINTER dem finalen drop = wirkungslos; beim Deploy 2026-09-15 so # aufgeschlagen). Deshalb: erst add (appends), dann `move` vor die Log-Regel. core_set = "\n".join([ bridge_set, vlan_set, '/ip firewall filter add action=accept chain=input comment="TP38: OOB->core mgmt" ' 'dst-port=22,80,8291,8728 in-interface=vlan99 protocol=tcp', '/ip firewall filter move [find where comment="TP38: OOB->core mgmt"] ' 'destination=[find where comment="TP5: input log-only"]', ':log info ("tp38: core pvid=" . [:tostr [/interface bridge port get [find interface=ether23] pvid]]' ' . " oob=" . [:tostr [:len [/ip firewall filter find where comment="TP38: OOB->core mgmt"]]])', ]) return [ (["crs326-branch"], branch_set, "TP38-oob-branch"), (["crs326-core"], core_set, "TP38-oob-core"), ] JOBS = { "drift-cleanup": job_drift_cleanup, "mgmt-acl": job_mgmt_acl, "tp42-dhcp-domain": job_tp42_dhcp_domain, "tp38-oob": job_tp38_oob, } # ------------------------------------------------------------------------ main def main(): ap = argparse.ArgumentParser() ap.add_argument("--job", choices=sorted(JOBS)) ap.add_argument("--apply", action="store_true", help="schreiben (sonst Dry-Run)") ap.add_argument("--no-deploy", action="store_true", help="Plaene nur anlegen, nicht deployen") ap.add_argument( "--deploy-ids", nargs="+", metavar="PLAN_ID", help="bereits angelegte Plaene gestaffelt deployen (Risiko-Reihenfolge selbst waehlen)", ) ap.add_argument( "--refresh-compliance", action="store_true", help="Backup -> Generate Intended -> Compliance (in dieser Reihenfolge) + Ergebnisbericht", ) args = ap.parse_args() if not TOKEN: raise SystemExit("NAUTOBOT_TOKEN nicht gesetzt") # Gestaffelter Deploy bereits angelegter Plaene -- fuer Faelle, in denen man # das riskantere Geraet erst anfassen will, nachdem das harmlose verifiziert ist. if args.deploy_ids: for pid in args.deploy_ids: p = req("GET", f"/plugins/golden-config/config-plan/{pid}/?depth=1") print(f" {pid} -> {(p.get('device') or {}).get('display')} [{plan_status(pid)}]") deploy(args.deploy_ids) print("\n== Plan-Status (massgeblich, nicht der Job-Status) ==") for pid in args.deploy_ids: p = req("GET", f"/plugins/golden-config/config-plan/{pid}/?depth=1") print(f" {(p.get('device') or {}).get('display'):15s} {plan_status(pid)}") print("\nJetzt am GERAET gegenpruefen -- 'Completed' beweist nichts.") return if args.refresh_compliance: refresh_compliance() return if not args.job: raise SystemExit("--job, --deploy-ids oder --refresh-compliance angeben") dry = not args.apply print(f"== {args.job} == {'DRY-RUN' if dry else 'APPLY'}") if not dry: print(" Erinnerung Regel 3: `/system backup` auf allen Zielgeraeten vorhanden?") plans = [] for device_names, config_set, ccid in JOBS[args.job](): print(f"\n-- {', '.join(device_names)} ({ccid})") plans.extend(create_plans(device_names, config_set, ccid, dry)) if dry or not plans: print("\nDry-Run beendet -- nichts geschrieben." if dry else "\nKeine Plaene angelegt.") return for _, pid in plans: approve(pid) print(f"\n{len(plans)} Plaene auf 'Approved' gesetzt") if args.no_deploy: print("--no-deploy: Deploy uebersprungen.") return print("\n== Deploy ==") deploy([pid for _, pid in plans]) print("\n== Plan-Status (massgeblich, nicht der Job-Status) ==") for name, pid in plans: print(f" {name:15s} {plan_status(pid)} ({pid})") print( "\nNOCH NICHT FERTIG:\n" " 1. Am GERAET gegenpruefen (MikroMCP) -- Plan 'Completed' beweist nichts.\n" " 2. Danach in dieser Reihenfolge: Backup -> Generate Intended -> Compliance.\n" " Die Compliance vergleicht gegen das letzte BACKUP, nicht gegen das Geraet." ) if __name__ == "__main__": main()