#!/usr/bin/env python3 """BF-Netzwerk -> Nautobot: TP22 Schritt 5 -- Datenmodell-Nachtraege (2026-08-13). WAS DAS FIXT ------------ Das DCIM-Modell stand noch auf dem Stand VOR dem Core-Cutover (2026-08-08): * `crs326-core` hatte 5 von 26 Ports modelliert, davon 3 als "GEPLANT (tentativ)" -- das Geraet ist seit dem Cutover der produktive L3-Router mit voller Belegung. * `crs326-branch` ether1 zeigte auf `sw1 #8`; sw1 existiert nicht mehr, der Uplink haengt an `crs326-core ether8`. * `crs326-branch` + `ap2` standen in "Technik UG"; beide stehen real in "Technik OG" (TP15-Standort-Mapping: UG = core, OG = branch + ap2). * sw1/sw2/cudy2 waren "Decommissioning" (= Prozess laeuft), sind aber physisch weg. * sw1 trug 4 doppelt modellierte Ports (`gi8/gi11/gi15/gi16` neben `GE8/GE11/GE15/GE16`). DATENQUELLE = DAS GERAET, NICHT DIE DOKU ---------------------------------------- Die Portbelegung unten ist am 2026-08-13 live erhoben worden (MikroMCP): `/interface/bridge/port` (pvid), `/interface/bridge/vlan` (tagged/untagged), `/interface/bridge/host` (gelernte MACs je Port+VID) und `/ip/arp` + DHCP-Leases zur Identifikation der Gegenstellen. Wo kein MAC gelernt wurde, steht "frei" bzw. die Unsicherheit explizit im Description-Text -- nicht geraten. COMPLIANCE-SICHERHEIT (wichtig) ------------------------------- Der core-Zweig in `mikrotik_routeros.j2` iteriert NUR Interfaces mit Praefix `vlan` (GW-IPs) und haelt die Bridge-VLAN-Tabelle als Jinja-Konstante. Ether-Ports sind damit compliance-neutral -- dieses Skript kann die intended config nicht verschieben. UMGEKEHRT GILT: die `vlanXX`-Descriptions auf dem Core NICHT anfassen, die landen als `comment=` in der intended config. Dieses Skript ruehrt sie nicht an. export NAUTOBOT_TOKEN=... python3 nautobot/ingest-tp22-datamodell.py # Dry-Run python3 nautobot/ingest-tp22-datamodell.py --apply # schreiben """ import json, os, sys, urllib.request, urllib.error, urllib.parse BASE = "http://10.40.20.154:8080/api" TOKEN = os.environ.get("NAUTOBOT_TOKEN", "") DRY = "--apply" not in sys.argv log = [] 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 payload={json.dumps(data)}\n -> {e.code} {e.read().decode()[:500]}") 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 goc(path, match, payload, label=None): name = label or payload.get("name") or str(match) obj = find(path, **match) if obj: return obj if DRY: log.append(f" + {name} (dry-run)") return {"id": f"DRY-{name}"} obj = req("POST", path + "/", payload) log.append(f" + {name}") return obj def patch(path, obj, payload, label): if str(obj["id"]).startswith("DRY"): return obj delta = {} for k, v in payload.items(): cur = obj.get(k) # Relationen kommen als {"id": ...}, Choice-Felder (type/mode) als {"value": ...}. if isinstance(cur, dict): cur = cur.get("id", cur.get("value")) if cur != v: delta[k] = v if not delta: return obj if DRY: log.append(f" ~ {label}: {sorted(delta)} (dry-run)") return obj out = req("PATCH", f"{path}/{obj['id']}/", delta) log.append(f" ~ {label}: {sorted(delta)}") return out def sid(name): return find("/extras/statuses", name=name)["id"] ETH, SFP, BR = "1000base-t", "10gbase-x-sfpp", "bridge" S_ACTIVE = sid("Active") S_OFFLINE = sid("Offline") # Fuer Geraete liefert Nautobot ab Werk nur "Decommissioning" (= Prozess laeuft). # sw1/sw2/cudy2 sind physisch ausgebaut -> Endzustand. "Decommissioned" existiert bereits # als mitgelieferter Status, aber nur fuer `circuits.circuit` -> content_types erweitern # statt einen zweiten, gleichnamigen Status anzulegen. ("Inventory" waere falsch, das # hiesse "auf Lager, einsatzbereit".) st_decom = find("/extras/statuses", name="Decommissioned") if not st_decom: st_decom = goc("/extras/statuses", {"name": "Decommissioned"}, {"name": "Decommissioned", "color": "9e9e9e", "content_types": ["dcim.device"], "description": "Ausgebaut/ausser Betrieb (Endzustand, kein laufender Prozess)"}, label="Status Decommissioned") elif "dcim.device" not in st_decom["content_types"]: patch("/extras/statuses", st_decom, {"content_types": sorted(set(st_decom["content_types"]) | {"dcim.device"})}, "Status Decommissioned +dcim.device") S_DECOM_DONE = st_decom["id"] vg = find("/ipam/vlan-groups", name="BF") V = {vid: find("/ipam/vlans", vid=vid, vlan_group=vg["id"]) for vid in (1, 10, 20, 30, 40, 50, 90, 99, 100)} LOC_OG = find("/dcim/locations", name="Technik OG")["id"] DEVS = ("crs326-core", "crs326-branch", "ap2", "sw1", "sw2", "cudy1", "cudy2") D = {n: find("/dcim/devices", name=n) for n in DEVS} def iface(dev, name, typ, desc, mode=None, untagged=None, tagged=None, enabled=True): obj = goc("/dcim/interfaces", {"device": dev, "name": name}, {"device": D[dev]["id"], "name": name, "type": typ, "status": S_ACTIVE, "description": desc, "enabled": enabled}, label=f"{dev}:{name}") p = {"description": desc, "type": typ, "enabled": enabled} if mode: p["mode"] = mode if untagged is not None: p["untagged_vlan"] = V[untagged]["id"] if mode == "access": # Nautobot verweigert mode=access, solange noch tagged_vlans haengen # ("Clear tagged_vlans to set mode to access") -> im selben PATCH mitleeren. p["tagged_vlans"] = [] obj = patch("/dcim/interfaces", obj, p, f"{dev}:{name}") if tagged is not None and not str(obj["id"]).startswith("DRY"): # ACHTUNG: `tagged_vlans` liest sich per REST als null, auch wenn korrekt gespeichert # (bekannte Nautobot-Falle, s. nautobot/README.md) -> dieser Vergleich schlaegt IMMER # an und der PATCH laeuft bei jedem Durchlauf. Der Write ist idempotent, die Zeile im # Log ist also kein Drift-Signal. Verifikation nur per GraphQL. cur = {t["id"] for t in (obj.get("tagged_vlans") or [])} want = {V[t]["id"] for t in tagged} if cur != want: if DRY: log.append(f" ~ {dev}:{name}: tagged_vlans -> {sorted(tagged)} (dry-run)") else: req("PATCH", f"/dcim/interfaces/{obj['id']}/", {"tagged_vlans": sorted(want)}) log.append(f" ~ {dev}:{name}: tagged_vlans -> {sorted(tagged)}") return obj # --- crs326-core: volle As-built-Belegung (live erhoben 2026-08-13) ------------------- # Trunk-Sets exakt wie in der Bridge-VLAN-Tabelle des Geraets -- ether1 traegt z.B. KEIN # VLAN20/99 (Bring-up-Rest), ether13 nur VLAN20. Bewusst nicht "geradegezogen". T_FULL = [10, 20, 30, 40, 50, 90, 99] # ether8 (branch), ether15 (ap1) T_EX_SW1 = [10, 30, 40, 50, 90, 100] # ether1, Ex-sw1-Uplink FREI = "frei (pvid=1, kein Link, keine VLAN-Membership)" CORE = [ ("bridge", BR, "vlan-filtering=yes; L3-Router-Bridge (SVIs vlan10-100 haengen hier)", None, None, None, True), ("ether1", ETH, "frei -- Ex-Uplink zu sw1 GE12 (sw1 ausser Betrieb 2026-08-08). Trunk-Reste " "+ frame-types=admit-only-vlan-tagged aus dem Bring-up, kein Link.", "tagged", 1, T_EX_SW1, True), ("ether2", ETH, "frei (Access VLAN40 vorbereitet, kein Link). " "WARN zusaetzlich statisches untagged-Member von VLAN1 (Bring-up-Altlast).", "access", 40, None, True), ("ether3", ETH, "Wohnzimmer Sideboard (untagged VLAN40) -- kein Link. " "WARN pvid=1 (Ex-Hybrid-VLAN1): Ingress landet in VLAN1, Egress untagged VLAN40.", "access", 40, None, True), # Descriptions sind auf 255 Zeichen begrenzt -> Langfassung steht in teilprojekte.md TP21.1/4. ("ether4", ETH, "Wohnzimmer-Sideboard: Smart-Managed-Switch mit den VLAN40-Multimediageraeten " "(u.a. TV 10.40.40.11). 2026-08-13 von Trunk auf untagged VLAN40 gezogen " "(GC ConfigPlan) -- vorher Egress tagged = Einbahnstrasse.", "access", 40, None, True), ("ether5", ETH, "Sonos SYMFONISK Schlafzimmer (R), 10.40.40.195 (Access VLAN40)", "access", 40, None, True), ("ether6", ETH, "Sonos SYMFONISK Schlafzimmer (L), 10.40.40.193 (Access VLAN40); " "untagged-Membership nur dynamisch aus pvid=40", "access", 40, None, True), ("ether7", ETH, FREI, None, None, None, True), ("ether8", ETH, "Uplink-Trunk -> crs326-branch ether1 (native VLAN1 + tagged 10/20/30/40/50/90/99)", "tagged", 1, T_FULL, True), ("ether9", ETH, FREI, None, None, None, True), ("ether10", ETH, FREI, None, None, None, True), ("ether11", ETH, "cudy1 (Access VLAN10, Nachfolger sw1 GE11) -- kein Link. " "WARN pvid=10 ohne VLAN10-Membership -> Port aktuell funktionslos.", "access", 10, None, True), ("ether12", ETH, FREI, None, None, None, True), ("ether13", ETH, "pve / Proxmox 10.40.20.100 -- Trunk: native VLAN1 + tagged VLAN20", "tagged", 1, [20], True), ("ether14", ETH, "Sonoff Dongle-M 10.40.30.250 (Access VLAN30)", "access", 30, None, True), ("ether15", ETH, "ap1 Uplink-Trunk (PoE-Injektor) -- native VLAN1 + tagged 10/20/30/40/50/90/99", "tagged", 1, T_FULL, True), ("ether16", ETH, "Arris-Kabelmodem = WAN (Access VLAN100); vlan100-SVI zieht die Liwest-Public-IP", "access", 100, None, True), ] CORE += [(f"ether{n}", ETH, FREI, None, None, None, True) for n in range(17, 25)] CORE += [("sfp-sfpplus1", SFP, "frei (disabled)", None, None, None, False), ("sfp-sfpplus2", SFP, "frei (disabled)", None, None, None, False)] print("== crs326-core: As-built-Portbelegung (26 Ports, live 2026-08-13)") for name, typ, desc, mode, unt, tag, en in CORE: iface("crs326-core", name, typ, desc, mode=mode, untagged=unt, tagged=tag, enabled=en) # --- crs326-branch: Korrekturen ------------------------------------------------------ print("== crs326-branch: Uplink-Ziel + Leerport-Realitaet") iface("crs326-branch", "ether1", ETH, "Uplink-Trunk -> crs326-core ether8 (native VLAN1 + tagged 10/20/30/40/50/90/99, kein 100)", mode="tagged", untagged=1, tagged=[10, 20, 30, 40, 50, 90, 99]) for n in (5, 6): iface("crs326-branch", f"ether{n}", ETH, "Sonos Arbeitszimmer (10.40.40.191/.192, Access VLAN40) -- aktuell kein Link", mode="access", untagged=40) for n in range(9, 24): iface("crs326-branch", f"ether{n}", ETH, "frei (kein Link; pvid=20 aus Bootstrap Rev4, aber keine VLAN-Membership -> " "funktionslos. NICHT disabled -- die alte 'disabled'-Notiz war stale.)") for n in (1, 2): iface("crs326-branch", f"sfp-sfpplus{n}", SFP, "frei (kein Link, nicht disabled)") # --- Standorte + Geraetestatus ------------------------------------------------------- print("== Standorte + Status") patch("/dcim/devices", D["crs326-branch"], {"location": LOC_OG}, "crs326-branch location") patch("/dcim/devices", D["ap2"], {"location": LOC_OG}, "ap2 location") patch("/dcim/devices", D["sw1"], {"status": S_DECOM_DONE, "comments": "ALLNET Core-Switch, ausser Betrieb 2026-08-08 -- ersetzt durch crs326-core " "(L3-Router + Core-Switch, Technik UG). Portbelegung GE1-GE16 bleibt als " "historischer Beleg stehen."}, "sw1") patch("/dcim/devices", D["sw2"], {"status": S_DECOM_DONE, "comments": "Omada ES208G, physisch abgebaut 2026-07-23 (TP15 Branch-Tausch) -- " "ersetzt durch crs326-branch."}, "sw2") patch("/dcim/devices", D["cudy2"], {"status": S_DECOM_DONE, "comments": "Cudy WR3600E V1.0, ausser Betrieb, wird nicht reaktiviert (TP19)."}, "cudy2") patch("/dcim/devices", D["cudy1"], {"status": S_OFFLINE, "comments": "Cudy WR3600E V1.0, Uebergangs-WLAN 'elektrosmog.c' waehrend des WLAN-Umbaus. " "Seit dem TP3-Rebuild (2026-08-08) nicht mehr noetig; core ether11 hat kein Link " "(Stand 2026-08-13) -> Offline. Endgueltiger Teardown = TP19."}, "cudy1") # --- sw1: doppelt modellierte Ports entfernen ---------------------------------------- # gi8/gi11/gi15/gi16 stammen aus einem zweiten Ingest-Lauf mit anderer Namenskonvention # und duplizieren GE8/GE11/GE15/GE16 (die die reichere Description tragen). print("== sw1: Duplikat-Ports (gi*) entfernen") for gi in ("gi8", "gi11", "gi15", "gi16"): obj = find("/dcim/interfaces", device="sw1", name=gi) if not obj: continue if DRY: log.append(f" - sw1:{gi} (dry-run)") else: req("DELETE", f"/dcim/interfaces/{obj['id']}/") log.append(f" - sw1:{gi}") print("\n".join(log) or " (keine Aenderungen)") print("\nDRY-RUN -- mit --apply schreiben." if DRY else "\nGeschrieben.")