#!/usr/bin/env python3 """ sync_to_netbox.py Reads the lldp-mapper SQLite DB and syncs devices/links to NetBox. Additive only - adds new devices/cables, never deletes anything. Run manually: python3 sync_to_netbox.py """ import sqlite3 import requests import logging logging.basicConfig(level=logging.INFO, format='%(asctime)s %(levelname)s %(message)s') logger = logging.getLogger(__name__) # --- Configuration --- DB_PATH = "/opt/lldp-mapper/data/network.db" NETBOX_URL = "http://192.168.16.137:8000" NETBOX_TOKEN = "nbt_LYkI6iSsflIU.9N4ziQxpbrV1AGPmpk0fwhfhJOSrSr0nGXEF4Ze2" # --- Helpers --- NB_HEADERS = { "Authorization": f"Bearer {NETBOX_TOKEN}", "Content-Type": "application/json", "Accept": "application/json", } def nb_get(path, params=None): r = requests.get(f"{NETBOX_URL}/api/{path}", headers=NB_HEADERS, params=params) r.raise_for_status() return r.json() def nb_post(path, data): r = requests.post(f"{NETBOX_URL}/api/{path}", headers=NB_HEADERS, json=data) if r.status_code not in (200, 201): logger.error(f"NetBox POST {path} failed: {r.status_code} {r.text}") return None return r.json() def nb_get_or_create(path, lookup_params, create_data): results = nb_get(path, params=lookup_params).get("results", []) if results: return results[0] logger.info(f"Creating {path}: {create_data.get('name', create_data)}") return nb_post(path, create_data) # --- Read from SQLite --- def load_db(): conn = sqlite3.connect(DB_PATH) conn.row_factory = sqlite3.Row switches = [dict(r) for r in conn.execute("SELECT * FROM switches ORDER BY hostname").fetchall()] links = [dict(r) for r in conn.execute(""" SELECT l.*, sa.hostname as hn_a, sb.hostname as hn_b FROM links l LEFT JOIN switches sa ON l.chassis_a = sa.chassis_id LEFT JOIN switches sb ON l.chassis_b = sb.chassis_id """).fetchall()] conn.close() return switches, links # --- NetBox Sync --- # # This NetBox instance is hand-curated (real sites, device types, roles) — # this script never creates devices. It only looks up devices that already # exist by hostname and syncs their interfaces/cables/IP. A switch lldp-mapper # knows about but that has no matching device in NetBox is skipped with a # warning so it can be added properly (correct site/manufacturer) by hand. def ensure_device(switch): hostname = switch["hostname"] or switch["mgmt_ip"] results = nb_get("dcim/devices/", params={"name": hostname}).get("results", []) if results: logger.info(f" Device exists: {hostname}") return results[0] logger.warning(f" Skipping {hostname} — no matching device in NetBox (add it manually first)") return None def ensure_interface(device_id, port_name): results = nb_get("dcim/interfaces/", params={ "device_id": device_id, "name": port_name }).get("results", []) if results: return results[0] return nb_post("dcim/interfaces/", { "device": device_id, "name": port_name, "type": "1000base-t", }) def ensure_serial(switch, device): serial = switch.get("serial") or "" if not serial or device.get("serial") == serial: return logger.info(f" Updating serial: {serial}") requests.patch( f"{NETBOX_URL}/api/dcim/devices/{device['id']}/", headers=NB_HEADERS, json={"serial": serial} ) def ensure_mac_comment(switch, device): mac = switch.get("chassis_id") or "" if not mac or "-" not in mac: # only real MACs, not IP-based chassis IDs return line = f"Chassis MAC: {mac}" comments = device.get("comments") or "" if line in comments: return logger.info(f" Recording chassis MAC in comments: {mac}") new_comments = f"{comments}\n{line}".strip() if comments else line requests.patch( f"{NETBOX_URL}/api/dcim/devices/{device['id']}/", headers=NB_HEADERS, json={"comments": new_comments} ) def ensure_ip(switch, device_id): if not switch.get("mgmt_ip"): return ip_addr = f"{switch['mgmt_ip']}/24" results = nb_get("ipam/ip-addresses/", params={"address": ip_addr}).get("results", []) if results: ip_obj = results[0] else: logger.info(f" Creating IP: {ip_addr}") ip_obj = nb_post("ipam/ip-addresses/", { "address": ip_addr, "status": "active", }) if not ip_obj: return requests.patch( f"{NETBOX_URL}/api/dcim/devices/{device_id}/", headers=NB_HEADERS, json={"primary_ip4": ip_obj["id"]} ) def ensure_cable(device_map, link): hn_a = link["hn_a"] hn_b = link["hn_b"] port_a = link["port_a"] port_b = link["port_b"] if hn_a not in device_map or hn_b not in device_map: logger.warning(f" Skipping cable {hn_a}:{port_a} <-> {hn_b}:{port_b} — device not in NetBox") return dev_a = device_map[hn_a] dev_b = device_map[hn_b] iface_a = ensure_interface(dev_a["id"], port_a) iface_b = ensure_interface(dev_b["id"], port_b) if not iface_a or not iface_b: return result_a = nb_get("dcim/interfaces/", params={"device_id": dev_a["id"], "name": port_a}).get("results", []) if result_a and result_a[0].get("cable"): logger.info(f" Cable already exists: {hn_a}:{port_a} <-> {hn_b}:{port_b}") return logger.info(f" Creating cable: {hn_a}:{port_a} <-> {hn_b}:{port_b}") nb_post("dcim/cables/", { "a_terminations": [{"object_type": "dcim.interface", "object_id": iface_a["id"]}], "b_terminations": [{"object_type": "dcim.interface", "object_id": iface_b["id"]}], "status": "connected", "label": f"{hn_a}→{hn_b}", "type": "smf-os2", }) def sync_netbox(switches, links): logger.info("=== Syncing to NetBox ===") device_map = {} for sw in switches: hostname = sw["hostname"] or sw["mgmt_ip"] logger.info(f"Processing device: {hostname}") device = ensure_device(sw) if device: device_map[hostname] = device ensure_serial(sw, device) ensure_mac_comment(sw, device) ensure_ip(sw, device["id"]) logger.info(f"Devices synced: {len(device_map)}") logger.info("Processing cables...") for link in links: ensure_cable(device_map, link) logger.info("NetBox sync complete.") # --- Main --- if __name__ == "__main__": switches, links = load_db() logger.info(f"Loaded {len(switches)} switches and {len(links)} links from DB") sync_netbox(switches, links) logger.info("=== All done ===")