diff --git a/README.md b/README.md index ac5c6c7..615dd88 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,6 @@ # NetBox VM Import Desktop -Windows-Desktopanwendung zum Importieren und Synchronisieren virtueller Maschinen aus VMware ESXi/vSphere und Proxmox VE nach NetBox. +Windows-Desktopanwendung zum Synchronisieren virtueller Maschinen sowie HPE-/Aruba-Switche nach NetBox. ## Funktionen @@ -16,6 +16,10 @@ Windows-Desktopanwendung zum Importieren und Synchronisieren virtueller Maschine - Optionale Unterstützung selbstsignierter TLS-Zertifikate - Synchronisierung von VMs, Hardwaredaten, Interfaces und IP-Adressen - Übernahme primärer IPv4- und IPv6-Adressen +- Getrennte Haupt-Tabs für Virtualisierung und Switche +- Verschlüsselte SSH-Zugänge pro bestehendem NetBox-Switch +- SSH-Erkennung für Aruba CX, ArubaOS-Switch/ProCurve und HPE Comware +- Synchronisierung von Switch-Schnittstellen und optionalen LLDP-Verbindungen als NetBox-Kabel ## Verwendung diff --git a/app.py b/app.py index a924979..ed55322 100644 --- a/app.py +++ b/app.py @@ -13,6 +13,7 @@ from pyVim.connect import Disconnect, SmartConnect from pyVmomi import vim from profile_store import ProfileStore +from switch_sync import PLATFORMS, discover_switch, sync_snapshot_to_netbox class SourceDialog(tk.Toplevel): @@ -134,6 +135,112 @@ class SourceDialog(tk.Toplevel): self.destroy() +class SwitchSourceDialog(tk.Toplevel): + """Editor for one existing NetBox switch and its SSH connection.""" + + def __init__(self, parent, devices, switch=None): + super().__init__(parent) + self.title("Switch-Zugang") + self.geometry("560x500") + self.resizable(False, False) + self.transient(parent) + self.grab_set() + self.result = None + switch = switch or {} + self.switch_id = switch.get("id", uuid.uuid4().hex) + self.enabled = tk.BooleanVar(value=switch.get("enabled", True)) + self.platform = tk.StringVar(value=switch.get("platform", "aruba_cx")) + self.device_map = {} + + device_values = [] + selected_device = "" + for device in devices: + site = getattr(getattr(device, "site", None), "name", "") + display = f"{device.name} · {site}" if site else device.name + self.device_map[display] = (device.id, device.name) + device_values.append(display) + if device.id == switch.get("device_id"): + selected_device = display + + body = ttk.Frame(self, padding=22) + body.pack(fill="both", expand=True) + body.columnconfigure(1, weight=1) + ttk.Label(body, text="NetBox-Gerät").grid(row=0, column=0, sticky="w", padx=(0, 12), pady=7) + self.device_selection = tk.StringVar(value=selected_device) + ttk.Combobox( + body, + textvariable=self.device_selection, + values=device_values, + state="readonly", + ).grid(row=0, column=1, sticky="ew", pady=7) + + ttk.Label(body, text="Plattform").grid(row=1, column=0, sticky="w", padx=(0, 12), pady=7) + platform_map = {label: key for key, label in PLATFORMS.items()} + self.platform_map = platform_map + platform_display = tk.StringVar(value=PLATFORMS[self.platform.get()]) + self.platform_display = platform_display + ttk.Combobox( + body, + textvariable=platform_display, + values=list(platform_map), + state="readonly", + ).grid(row=1, column=1, sticky="ew", pady=7) + + self.host_entry = self.add_entry(body, 2, "Host / IP", switch.get("host", "")) + self.port_entry = self.add_entry(body, 3, "SSH-Port", str(switch.get("port", 22))) + self.username_entry = self.add_entry(body, 4, "Benutzer", switch.get("username", "")) + self.password_entry = self.add_entry(body, 5, "Passwort", switch.get("password", ""), secret=True) + ttk.Checkbutton(body, text="Switch aktivieren", variable=self.enabled).grid( + row=6, column=0, columnspan=2, sticky="w", pady=(12, 4) + ) + ttk.Label( + body, + text="Das Passwort wird im Kundenprofil mit Windows DPAPI verschlüsselt gespeichert.", + foreground="#52606d", + wraplength=500, + ).grid(row=7, column=0, columnspan=2, sticky="w", pady=(12, 0)) + buttons = ttk.Frame(body) + buttons.grid(row=8, column=0, columnspan=2, sticky="e", pady=(22, 0)) + ttk.Button(buttons, text="Abbrechen", command=self.destroy).pack(side="left", padx=4) + ttk.Button(buttons, text="Übernehmen", command=self.accept, style="Primary.TButton").pack(side="left", padx=4) + self.protocol("WM_DELETE_WINDOW", self.destroy) + self.wait_window(self) + + @staticmethod + def add_entry(parent, row, label, value, secret=False): + ttk.Label(parent, text=label).grid(row=row, column=0, sticky="w", padx=(0, 12), pady=7) + entry = ttk.Entry(parent, show="•" if secret else "") + entry.grid(row=row, column=1, sticky="ew", pady=7) + entry.insert(0, value) + return entry + + def accept(self): + device = self.device_map.get(self.device_selection.get()) + host = self.host_entry.get().strip() + username = self.username_entry.get().strip() + password = self.password_entry.get() + try: + port = int(self.port_entry.get()) + except ValueError: + port = 0 + if not device or not host or not username or not password or not 1 <= port <= 65535: + messagebox.showwarning("Fehlende Angaben", "Bitte alle Switch-Zugangsdaten korrekt ausfüllen.", parent=self) + return + device_id, device_name = device + self.result = { + "id": self.switch_id, + "name": device_name, + "device_id": device_id, + "platform": self.platform_map[self.platform_display.get()], + "host": host, + "port": port, + "username": username, + "password": password, + "enabled": self.enabled.get(), + } + self.destroy() + + class TenantConfirmationDialog(tk.Toplevel): """Require explicit tenant confirmation before changing NetBox VMs.""" @@ -726,6 +833,9 @@ class NetBoxVMImporter: self.vms = [] self.profiles = {} self.current_sources = [] + self.current_switches = [] + self.switch_snapshots = {} + self.netbox_devices = [] self.cluster_map = {} self.cluster_tenants = {} self.tenant_map = {} @@ -766,7 +876,14 @@ class NetBoxVMImporter: pass def create_gui(self): - main = ttk.Frame(self.root, padding=18) + self.main_tabs = ttk.Notebook(self.root) + self.main_tabs.pack(fill="both", expand=True) + virtualization_tab = ttk.Frame(self.main_tabs) + switches_tab = ttk.Frame(self.main_tabs) + self.main_tabs.add(virtualization_tab, text="Virtualisierung") + self.main_tabs.add(switches_tab, text="Switche") + + main = ttk.Frame(virtualization_tab, padding=18) main.pack(fill="both", expand=True) main.columnconfigure(0, weight=1) main.rowconfigure(4, weight=3) @@ -854,6 +971,81 @@ class NetBoxVMImporter: self.log_text = tk.Text(log_card, height=7, borderwidth=0, bg="#f7f9fb", foreground="#243b53", font=("Consolas", 9), state="disabled") self.log_text.pack(fill="both", expand=True) ttk.Label(main, textvariable=self.status_text, style="Status.TLabel", anchor="w").grid(row=7, column=0, sticky="ew", pady=(6, 0)) + self.create_switch_gui(switches_tab) + + def create_switch_gui(self, parent): + main = ttk.Frame(parent, padding=18) + main.pack(fill="both", expand=True) + main.columnconfigure(0, weight=1) + main.rowconfigure(2, weight=2) + main.rowconfigure(4, weight=3) + + header = ttk.Frame(main) + header.grid(row=0, column=0, sticky="ew", pady=(0, 12)) + ttk.Label(header, text="HPE & Aruba Switche", style="Title.TLabel").pack(anchor="w") + ttk.Label( + header, + text="Bestehende NetBox-Geräte per SSH auslesen · Profil: ", + style="Subtitle.TLabel", + ).pack(side="left") + ttk.Label(header, textvariable=self.profile_name, style="Subtitle.TLabel").pack(side="left") + + source_card = ttk.LabelFrame(main, text="Switch-Zugänge", style="Card.TLabelframe") + source_card.grid(row=1, column=0, sticky="nsew", pady=(0, 10)) + switch_columns = ("device", "platform", "host", "active") + self.switch_tree = ttk.Treeview(source_card, columns=switch_columns, show="headings", height=6) + for column, title, width in ( + ("device", "NetBox-Gerät", 230), + ("platform", "Plattform", 210), + ("host", "SSH-Host", 260), + ("active", "Aktiv", 60), + ): + self.switch_tree.heading(column, text=title) + self.switch_tree.column(column, width=width, anchor="center" if column == "active" else "w") + self.switch_tree.pack(fill="both", expand=True) + self.switch_tree.bind("", lambda _event: self.edit_switch()) + source_buttons = ttk.Frame(source_card) + source_buttons.pack(fill="x", pady=(8, 0)) + ttk.Button(source_buttons, text="NetBox-Geräte laden", command=self.load_switch_devices).pack(side="left") + ttk.Button(source_buttons, text="Switch hinzufügen", command=self.add_switch).pack(side="left", padx=(8, 4)) + ttk.Button(source_buttons, text="Bearbeiten", command=self.edit_switch).pack(side="left", padx=4) + ttk.Button(source_buttons, text="Entfernen", command=self.remove_switch).pack(side="left", padx=4) + ttk.Button( + source_buttons, + text="Ausgewählte auslesen", + command=self.discover_selected_switches, + style="Primary.TButton", + ).pack(side="right") + + result_card = ttk.LabelFrame(main, text="Schnittstellen & LLDP-Nachbarn", style="Card.TLabelframe") + result_card.grid(row=2, column=0, sticky="nsew", pady=(0, 10)) + result_columns = ("switch", "interface", "state", "speed", "neighbor", "remote_port") + self.switch_result_tree = ttk.Treeview(result_card, columns=result_columns, show="headings") + result_headings = { + "switch": "Switch", "interface": "Schnittstelle", "state": "Status", + "speed": "Geschwindigkeit", "neighbor": "LLDP-Nachbar", "remote_port": "Nachbar-Port", + } + result_widths = {"switch": 180, "interface": 150, "state": 80, "speed": 110, "neighbor": 210, "remote_port": 170} + for column in result_columns: + self.switch_result_tree.heading(column, text=result_headings[column]) + self.switch_result_tree.column(column, width=result_widths[column]) + result_scroll = ttk.Scrollbar(result_card, orient="vertical", command=self.switch_result_tree.yview) + self.switch_result_tree.configure(yscrollcommand=result_scroll.set) + self.switch_result_tree.pack(side="left", fill="both", expand=True) + result_scroll.pack(side="right", fill="y") + + actions = ttk.Frame(main) + actions.grid(row=3, column=0, sticky="ew", pady=(0, 10)) + self.create_neighbor_cables = tk.BooleanVar(value=True) + ttk.Checkbutton(actions, text="LLDP-Verbindungen als Kabel anlegen", variable=self.create_neighbor_cables).pack(side="left") + ttk.Button(actions, text="Nach NetBox synchronisieren", command=self.sync_switches, style="Primary.TButton").pack(side="right") + + log_card = ttk.LabelFrame(main, text="Switch-Protokoll", style="Card.TLabelframe") + log_card.grid(row=4, column=0, sticky="nsew") + self.switch_log_text = tk.Text( + log_card, height=8, borderwidth=0, bg="#f7f9fb", foreground="#243b53", font=("Consolas", 9), state="disabled" + ) + self.switch_log_text.pack(fill="both", expand=True) @staticmethod def add_field(parent, row, label, secret=False): @@ -901,6 +1093,7 @@ class NetBoxVMImporter: "netbox_token": self.netbox_token.get(), "netbox_ignore_ssl": self.netbox_ignore_ssl.get(), "sources": [dict(source) for source in self.current_sources], + "switches": [dict(switch) for switch in self.current_switches], } def populate_profile(self, name): @@ -909,7 +1102,12 @@ class NetBoxVMImporter: self.set_entry(self.netbox_token, profile.get("netbox_token", "")) self.netbox_ignore_ssl.set(bool(profile.get("netbox_ignore_ssl", False))) self.current_sources = [dict(source) for source in profile.get("sources", [])] + self.current_switches = [dict(switch) for switch in profile.get("switches", [])] + self.switch_snapshots = {} self.refresh_source_tree() + self.refresh_switch_tree() + if hasattr(self, "switch_result_tree"): + self.refresh_switch_results() self.clear_loaded_data() self.status_text.set(f"Profil „{name}“ geladen") @@ -933,7 +1131,9 @@ class NetBoxVMImporter: self.set_entry(self.netbox_token, "") self.netbox_ignore_ssl.set(False) self.current_sources = [] + self.current_switches = [] self.refresh_source_tree() + self.refresh_switch_tree() self.profiles[name] = self.profile_from_form(name) self.refresh_profile_list() @@ -971,7 +1171,9 @@ class NetBoxVMImporter: self.set_entry(self.netbox_url, "") self.set_entry(self.netbox_token, "") self.current_sources = [] + self.current_switches = [] self.refresh_source_tree() + self.refresh_switch_tree() def refresh_source_tree(self): self.source_tree.delete(*self.source_tree.get_children()) @@ -1012,6 +1214,164 @@ class NetBoxVMImporter: self.current_sources.remove(source) self.refresh_source_tree() + def switch_log(self, text): + self.switch_log_text.configure(state="normal") + self.switch_log_text.insert(tk.END, f"{text}\n") + self.switch_log_text.see(tk.END) + self.switch_log_text.configure(state="disabled") + self.status_text.set(text) + self.root.update() + + def refresh_switch_tree(self): + if not hasattr(self, "switch_tree"): + return + self.switch_tree.delete(*self.switch_tree.get_children()) + for switch in self.current_switches: + self.switch_tree.insert("", "end", iid=switch["id"], values=( + switch.get("name", ""), + PLATFORMS.get(switch.get("platform"), switch.get("platform", "")), + f"{switch.get('host', '')}:{switch.get('port', 22)}", + "Ja" if switch.get("enabled", True) else "Nein", + )) + + def load_switch_devices(self): + try: + if not self.netbox_url.get().strip() or not self.netbox_token.get(): + raise ValueError("Bitte NetBox-URL und API-Token im Tab Virtualisierung eintragen.") + if not self.save_profile(show_message=False): + return False + self.connect_netbox() + self.switch_log("Lade vorhandene NetBox-Geräte …") + self.netbox_devices = sorted(self.nb.dcim.devices.all(), key=lambda item: item.name.casefold()) + self.switch_log(f"{len(self.netbox_devices)} NetBox-Geräte geladen") + return True + except Exception as error: + messagebox.showerror("NetBox-Verbindung", str(error), parent=self.root) + return False + + def add_switch(self): + if not self.netbox_devices and not self.load_switch_devices(): + return + dialog = SwitchSourceDialog(self.root, self.netbox_devices) + if dialog.result: + self.current_switches.append(dialog.result) + self.refresh_switch_tree() + + def selected_switches(self): + selected_ids = set(self.switch_tree.selection()) + return [switch for switch in self.current_switches if switch["id"] in selected_ids] + + def edit_switch(self): + selected = self.selected_switches() + if len(selected) != 1: + messagebox.showinfo("Switch", "Bitte genau einen Switch auswählen.", parent=self.root) + return + if not self.netbox_devices and not self.load_switch_devices(): + return + current = selected[0] + dialog = SwitchSourceDialog(self.root, self.netbox_devices, current) + if dialog.result: + index = self.current_switches.index(current) + self.current_switches[index] = dialog.result + self.refresh_switch_tree() + self.switch_tree.selection_set(dialog.result["id"]) + + def remove_switch(self): + selected = self.selected_switches() + if not selected: + return + if not messagebox.askyesno( + "Switch entfernen", + f"{len(selected)} Switch-Zugang/Zugänge aus dem Profil entfernen?", + parent=self.root, + ): + return + selected_ids = {switch["id"] for switch in selected} + self.current_switches = [switch for switch in self.current_switches if switch["id"] not in selected_ids] + for switch_id in selected_ids: + self.switch_snapshots.pop(switch_id, None) + self.refresh_switch_tree() + self.refresh_switch_results() + + def discover_selected_switches(self): + switches = self.selected_switches() + if not switches: + messagebox.showinfo("Switche", "Bitte mindestens einen Switch auswählen.", parent=self.root) + return + if not self.save_profile(show_message=False): + return + failures = [] + for switch in switches: + if not switch.get("enabled", True): + continue + try: + self.switch_log(f"Lese {switch['name']} ({switch['host']}) aus …") + snapshot = discover_switch(switch) + self.switch_snapshots[switch["id"]] = snapshot + self.switch_log( + f"{switch['name']}: {len(snapshot['interfaces'])} Schnittstellen, " + f"{len(snapshot['neighbors'])} LLDP-Nachbarn" + ) + except Exception as error: + failures.append(f"{switch['name']}: {error}") + self.switch_log(f"FEHLER {switch['name']}: {error}") + self.refresh_switch_results() + if failures: + messagebox.showwarning("Teilweise ausgelesen", "\n".join(failures), parent=self.root) + + def refresh_switch_results(self): + self.switch_result_tree.delete(*self.switch_result_tree.get_children()) + switch_map = {switch["id"]: switch for switch in self.current_switches} + for switch_id, snapshot in self.switch_snapshots.items(): + switch = switch_map.get(switch_id) + if not switch: + continue + for interface in snapshot["interfaces"]: + neighbor = interface.get("neighbor") or {} + speed = interface.get("speed_mbps") + speed_text = f"{speed / 1000:g} Gbit/s" if speed and speed >= 1000 else (f"{speed} Mbit/s" if speed else "") + self.switch_result_tree.insert("", "end", values=( + switch["name"], + interface["name"], + "Up" if interface.get("connected") else "Down", + speed_text, + neighbor.get("system_name", ""), + neighbor.get("remote_port", ""), + )) + + def sync_switches(self): + if not self.switch_snapshots: + messagebox.showinfo("Switche", "Bitte zuerst mindestens einen Switch auslesen.", parent=self.root) + return + if self.nb is None and not self.load_switch_devices(): + return + switch_map = {switch["id"]: switch for switch in self.current_switches} + totals = {"created": 0, "updated": 0, "cables": 0} + errors = [] + for switch_id, snapshot in self.switch_snapshots.items(): + switch = switch_map.get(switch_id) + if not switch: + continue + self.switch_log(f"Synchronisiere {switch['name']} nach NetBox …") + result = sync_snapshot_to_netbox( + self.nb, + switch, + snapshot, + create_cables=self.create_neighbor_cables.get(), + ) + for key in totals: + totals[key] += result[key] + errors.extend(f"{switch['name']}: {error}" for error in result["errors"]) + text = ( + f"{totals['created']} Schnittstellen erstellt, {totals['updated']} aktualisiert, " + f"{totals['cables']} LLDP-Kabel angelegt." + ) + self.switch_log(text) + if errors: + messagebox.showwarning("Synchronisierung teilweise abgeschlossen", text + "\n\n" + "\n".join(errors), parent=self.root) + else: + messagebox.showinfo("Switch-Synchronisierung abgeschlossen", text, parent=self.root) + def clear_loaded_data(self): self.vms = [] if hasattr(self, "vm_tree"): diff --git a/dist/NetBox VM Import.exe b/dist/NetBox VM Import.exe index be92ddd..b3ed623 100644 Binary files a/dist/NetBox VM Import.exe and b/dist/NetBox VM Import.exe differ diff --git a/profile_store.py b/profile_store.py index 2756cd2..82a509f 100644 --- a/profile_store.py +++ b/profile_store.py @@ -83,6 +83,7 @@ class ProfileStore: "netbox_token": unprotect(record.get("netbox_token_encrypted", "")), "netbox_ignore_ssl": bool(record.get("netbox_ignore_ssl", record.get("ignore_ssl_errors", False))), "sources": [], + "switches": [], } for source_record in record.get("sources", []): source = dict(source_record) @@ -91,6 +92,13 @@ class ProfileStore: if encrypted_key in source: source[key] = unprotect(source.pop(encrypted_key)) profile["sources"].append(source) + for switch_record in record.get("switches", []): + switch = dict(switch_record) + for key in SECRET_KEYS: + encrypted_key = f"{key}_encrypted" + if encrypted_key in switch: + switch[key] = unprotect(switch.pop(encrypted_key)) + profile["switches"].append(switch) # Migration from v2: one VMware source embedded in each profile. if not profile["sources"] and record.get("esxi_host"): @@ -123,6 +131,7 @@ class ProfileStore: "password": data.get("esxi_password", ""), "ignore_ssl": bool(data.get("ignore_ssl_errors", False)), }], + "switches": [], } return profiles, data.get("active_profile", "") @@ -137,15 +146,23 @@ class ProfileStore: if key in source_data: source[f"{key}_encrypted"] = protect(source_data.get(key, "")) sources.append(source) + switches = [] + for switch_data in profile.get("switches", []): + switch = {key: value for key, value in switch_data.items() if key not in SECRET_KEYS} + for key in SECRET_KEYS: + if key in switch_data: + switch[f"{key}_encrypted"] = protect(switch_data.get(key, "")) + switches.append(switch) records.append({ "name": name, "netbox_url": profile.get("netbox_url", ""), "netbox_token_encrypted": protect(profile.get("netbox_token", "")), "netbox_ignore_ssl": bool(profile.get("netbox_ignore_ssl", False)), "sources": sources, + "switches": switches, }) - payload = {"version": 3, "active_profile": active_profile, "profiles": records} + payload = {"version": 4, "active_profile": active_profile, "profiles": records} self.path.parent.mkdir(parents=True, exist_ok=True) temp_path = self.path.with_suffix(".tmp") with temp_path.open("w", encoding="utf-8") as stream: diff --git a/requirements.txt b/requirements.txt index f26374d..bc1ca6b 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,4 +1,5 @@ pyvmomi pynetbox requests +paramiko pyinstaller diff --git a/switch_sync.py b/switch_sync.py new file mode 100644 index 0000000..eda71af --- /dev/null +++ b/switch_sync.py @@ -0,0 +1,298 @@ +"""SSH discovery and NetBox synchronization for HPE and Aruba switches.""" + +import re +import time + +import paramiko + + +PLATFORMS = { + "aruba_cx": "Aruba CX", + "aruba_aos": "ArubaOS-Switch / ProCurve", + "hpe_comware": "HPE Comware", +} + +COMMANDS = { + "aruba_cx": ("no page", "show interface brief", "show lldp neighbor-info"), + "aruba_aos": ("no page", "show interfaces brief", "show lldp info remote-device"), + "hpe_comware": ("screen-length disable", "display interface brief", "display lldp neighbor-information verbose"), +} + +ANSI_ESCAPE = re.compile(r"\x1b(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])") +MAC_PATTERN = re.compile(r"(?i)\b(?:[0-9a-f]{2}[:-]){5}[0-9a-f]{2}\b|\b[0-9a-f]{4}(?:-[0-9a-f]{4}){2}\b") + + +def _clean_output(value): + value = ANSI_ESCAPE.sub("", value).replace("\r", "") + value = re.sub(r"--\s*MORE\s*--|Press any key to continue.*", "", value, flags=re.I) + return value + + +def _read_available(channel, quiet_seconds=0.35, timeout=12): + chunks = [] + started = time.monotonic() + last_data = started + while time.monotonic() - started < timeout: + if channel.recv_ready(): + chunks.append(channel.recv(65535).decode("utf-8", errors="replace")) + last_data = time.monotonic() + elif time.monotonic() - last_data >= quiet_seconds: + break + else: + time.sleep(0.05) + return _clean_output("".join(chunks)) + + +def run_switch_commands(config): + platform = config["platform"] + if platform not in COMMANDS: + raise ValueError(f"Nicht unterstützte Switch-Plattform: {platform}") + client = paramiko.SSHClient() + client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) + try: + client.connect( + hostname=config["host"], + port=int(config.get("port", 22)), + username=config["username"], + password=config["password"], + look_for_keys=False, + allow_agent=False, + timeout=12, + auth_timeout=12, + banner_timeout=12, + ) + channel = client.invoke_shell(width=240, height=1000) + _read_available(channel, timeout=3) + outputs = [] + for command in COMMANDS[platform]: + channel.send(command + "\n") + outputs.append(_read_available(channel)) + return {"interfaces": outputs[1], "neighbors": outputs[2]} + finally: + client.close() + + +def _normalize_mac(value): + if not value: + return None + compact = re.sub(r"[^0-9a-fA-F]", "", value).upper() + if len(compact) != 12: + return None + return ":".join(compact[index:index + 2] for index in range(0, 12, 2)) + + +def _speed_mbps(line): + matches = re.findall(r"(?i)\b(\d+(?:\.\d+)?)\s*(T|G|M)(?:b(?:it)?|bps)?\b", line) + if not matches: + duplex = re.search(r"(?i)\b(\d+)(?:FDx|HDx)\b", line) + if duplex: + return int(duplex.group(1)) + trailing = re.search(r"\b(\d{2,6})\s*$", line) + return int(trailing.group(1)) if trailing else None + value, unit = matches[-1] + factor = {"M": 1, "G": 1000, "T": 1_000_000}[unit.upper()] + return int(float(value) * factor) + + +def _interface_name(token): + token = token.strip(" |:") + return bool(re.match( + r"(?i)^(?:\d+(?:/\d+){0,3}|(?:GE|XGE|HGE|FGE|FortyGigE|Ten-GigabitEthernet|GigabitEthernet)\S+|" + r"Bridge-Aggregation\d+|Trk\d+|lag\d+)$", + token, + )) + + +def parse_interfaces(output, platform): + interfaces = [] + seen = set() + for raw_line in _clean_output(output).splitlines(): + line = raw_line.strip() + if not line or set(line) <= {"-", "=", " "}: + continue + tokens = re.split(r"\s+", line.replace("|", " | ").strip()) + name_index = next((index for index, token in enumerate(tokens[:3]) if _interface_name(token)), None) + if name_index is None: + continue + name = tokens[name_index].strip(" |:") + if name in seen: + continue + lower = line.lower() + if not re.search(r"\b(up|down|administratively down|disabled|enabled)\b", lower): + continue + enabled = not bool(re.search(r"administratively down|admin.*down|disabled|\bno\s+(?:up|down)\b", lower)) + connected = bool(re.search(r"\bup\b", lower)) and not bool(re.search(r"\bdown\b", lower)) + mac_match = MAC_PATTERN.search(line) + description = "" + if platform == "aruba_aos" and "|" in raw_line: + columns = [column.strip() for column in raw_line.split("|")] + if len(columns) > 1: + description = columns[1] + interfaces.append({ + "name": name, + "enabled": enabled, + "connected": connected, + "speed_mbps": _speed_mbps(line), + "mac_address": _normalize_mac(mac_match.group(0)) if mac_match else None, + "description": description, + "neighbor": None, + }) + seen.add(name) + return interfaces + + +def _field(block, patterns): + for pattern in patterns: + match = re.search(pattern, block, flags=re.I | re.M) + if match: + return match.group(1).strip() + return "" + + +def parse_neighbors(output): + clean = _clean_output(output) + starts = list(re.finditer( + r"(?im)^(?:\s*(?:Local (?:Port|Interface|Intf)|LLDP neighbor-information of port)\s*[: ]\s*.+)$", + clean, + )) + if not starts: + return _parse_neighbor_table(clean) + neighbors = [] + for index, start in enumerate(starts): + end = starts[index + 1].start() if index + 1 < len(starts) else len(clean) + block = clean[start.start():end] + local_port = _field(block, ( + r"^\s*Local (?:Port|Interface|Intf)\s*[: ]\s*(\S+)", + r"^\s*LLDP neighbor-information of port\s+(?:\d+\[)?([^\]\s:]+)\]?", + )) + system_name = _field(block, ( + r"^\s*(?:System Name|SysName)\s*[: ]\s*(.+)$", + r"^\s*Neighbor name\s*[: ]\s*(.+)$", + )) + remote_port = _field(block, ( + r"^\s*(?:Port ID|PortId|Port ID subtype.*\n\s*Port ID)\s*[: ]\s*(.+)$", + r"^\s*(?:Port Description|PortDesc)\s*[: ]\s*(.+)$", + )) + management_ip = _field(block, ( + r"^\s*(?:Management Address|Management address|Management IP)\s*[: ]\s*([0-9a-fA-F:.]+)", + )) + if local_port and (system_name or remote_port): + neighbors.append({ + "local_port": local_port, + "system_name": system_name, + "remote_port": remote_port, + "management_ip": management_ip, + }) + return neighbors + + +def _parse_neighbor_table(output): + neighbors = [] + for raw_line in output.splitlines(): + line = raw_line.strip() + if not line or set(line) <= {"-", "=", " "}: + continue + tokens = re.split(r"\s+", line.replace("|", " | ")) + if not tokens or not _interface_name(tokens[0]): + continue + if "|" in tokens: + separator = tokens.index("|") + remaining = tokens[separator + 1:] + if len(remaining) < 3: + continue + remote_port = remaining[1] + system_name = remaining[-1] + else: + if len(tokens) < 5: + continue + remote_port = tokens[2] + system_name = tokens[-1] + neighbors.append({ + "local_port": tokens[0], + "system_name": system_name, + "remote_port": remote_port, + "management_ip": "", + }) + return neighbors + + +def discover_switch(config): + outputs = run_switch_commands(config) + interfaces = parse_interfaces(outputs["interfaces"], config["platform"]) + neighbors = parse_neighbors(outputs["neighbors"]) + interface_map = {interface["name"].casefold(): interface for interface in interfaces} + for neighbor in neighbors: + local = interface_map.get(neighbor["local_port"].casefold()) + if local: + local["neighbor"] = neighbor + return {"interfaces": interfaces, "neighbors": neighbors, "raw": outputs} + + +def _netbox_interface_type(speed_mbps): + if not speed_mbps: + return "other" + if speed_mbps >= 100_000: + return "100gbase-x-qsfp28" + if speed_mbps >= 40_000: + return "40gbase-x-qsfpp" + if speed_mbps >= 25_000: + return "25gbase-x-sfp28" + if speed_mbps >= 10_000: + return "10gbase-x-sfpp" + if speed_mbps >= 1000: + return "1000base-t" + return "100base-tx" + + +def sync_snapshot_to_netbox(nb, config, snapshot, create_cables=True): + device_id = int(config["device_id"]) + interface_records = {} + created = updated = cables = 0 + errors = [] + for interface in snapshot["interfaces"]: + payload = { + "device": device_id, + "name": interface["name"], + "type": _netbox_interface_type(interface.get("speed_mbps")), + "enabled": interface.get("enabled", True), + } + if interface.get("description"): + payload["description"] = interface["description"][:200] + if interface.get("mac_address"): + payload["mac_address"] = interface["mac_address"] + try: + record = nb.dcim.interfaces.get(device_id=device_id, name=interface["name"]) + if record: + record.update(payload) + updated += 1 + else: + record = nb.dcim.interfaces.create(payload) + created += 1 + interface_records[interface["name"].casefold()] = record + except Exception as error: + errors.append(f"Interface {interface['name']}: {error}") + + if create_cables: + for neighbor in snapshot["neighbors"]: + try: + local = interface_records.get(neighbor["local_port"].casefold()) + if not local or getattr(local, "cable", None): + continue + remote_device = nb.dcim.devices.get(name=neighbor["system_name"]) + if not remote_device: + continue + remote = nb.dcim.interfaces.get(device_id=remote_device.id, name=neighbor["remote_port"]) + if not remote or getattr(remote, "cable", None): + continue + nb.dcim.cables.create({ + "a_terminations": [{"object_type": "dcim.interface", "object_id": local.id}], + "b_terminations": [{"object_type": "dcim.interface", "object_id": remote.id}], + "status": "connected", + }) + cables += 1 + except Exception as error: + errors.append( + f"Nachbar {neighbor.get('local_port')} → {neighbor.get('system_name')} " + f"{neighbor.get('remote_port')}: {error}" + ) + return {"created": created, "updated": updated, "cables": cables, "errors": errors}