From 81d709cad225e7f4a48019e6ea7c41b36d624e4f Mon Sep 17 00:00:00 2001 From: pavel Date: Tue, 4 Nov 2025 16:19:22 +0200 Subject: [PATCH] 16:19 --- files/nb_onedevice_update.py | 57 ++++++++++++++++++++++++++++++------ 1 file changed, 48 insertions(+), 9 deletions(-) diff --git a/files/nb_onedevice_update.py b/files/nb_onedevice_update.py index 91ca9cd..3c41018 100644 --- a/files/nb_onedevice_update.py +++ b/files/nb_onedevice_update.py @@ -9,15 +9,17 @@ Flow 2) NetBox: lookup device by name -> read custom_fields.cloud_id. 3) Cloud: GET /v1/devices/{cloud_id}/liveness; if isConnected == false -> exit with "device is not online". 4) Cloud: GET /v1/devices/{cloud_id} -> read firmwareVersion, ipAddress, nodeName, sectorName, smallCellName, serialNumber. -5) NetBox updates (idempotent; change-only by default): +5) NetBox updates (idempotent by default): - custom_fields.fw_version ← firmwareVersion - custom_fields.nodeName ← nodeName - custom_fields.sectorName ← sectorName - custom_fields.smallCellName ← smallCellName - serial ← serialNumber 6) IP handling (only when online): - - If Cloud ipAddress is valid (not None/""/"0.0.0.0"): ensure eth0, ensure/create IP, MOVE if needed, set primary_ip4. - - If Cloud ipAddress is placeholder/invalid: skip IP changes; keep NetBox IP as-is. + - If Cloud ipAddress is valid (not None/""/"0.0.0.0"): + ensure eth0, ensure/create IP, MOVE if needed, set primary_ip4, + then PRUNE all other IPs on the device (default behavior). + - If Cloud ipAddress is placeholder/invalid: skip IP changes and keep NetBox IPs as-is. 7) Single-line OK/FAIL and proper exit code. Exit codes: @@ -30,7 +32,7 @@ Exit codes: import sys import json import argparse -from typing import Optional, Union +from typing import Optional, Union, List, Dict import requests from requests.adapters import HTTPAdapter @@ -181,6 +183,33 @@ class NetBox: return (r.json().get("device") or {}).get("id") return None + # --- Helpers for pruning --- + def list_device_ips(self, dev_id: int) -> List[Dict]: + # Try device_id filter (NetBox ≥3.6); otherwise fall back via interface scan + r = S_NB.get(self._url("/api/ipam/ip-addresses/"), headers=self._h(), + params={"device_id": dev_id, "limit": 1000}, timeout=REQ_TIMEOUT) + if r.status_code == 200: + return r.json().get("results") or [] + # Fallback path + ips: List[Dict] = [] + r2 = S_NB.get(self._url("/api/dcim/interfaces/"), headers=self._h(), + params={"device_id": dev_id, "limit": 1000}, timeout=REQ_TIMEOUT) + if r2.status_code == 200: + for iface in (r2.json().get("results") or []): + ifid = iface.get("id") + r3 = S_NB.get(self._url("/api/ipam/ip-addresses/"), headers=self._h(), + params={"assigned_object_type": "dcim.interface", + "assigned_object_id": ifid, "limit": 1000}, timeout=REQ_TIMEOUT) + if r3.status_code == 200: + ips.extend(r3.json().get("results") or []) + return ips + + def delete_ip(self, ip_id: int) -> None: + log(f"NB: delete IP id={ip_id}") + r = S_NB.delete(self._url(f"/api/ipam/ip-addresses/{ip_id}/"), headers=self._h(), timeout=REQ_TIMEOUT) + if not (200 <= r.status_code < 300 or r.status_code == 204): + die(4, f"FAIL delete IP id={ip_id} HTTP={r.status_code} body={r.text[:200]}") + # ------------ Cloud API ------------ class Cloud: def __init__(self, base: str, bearer: str): @@ -229,14 +258,14 @@ def run(hostname: str) -> None: if cloud_id in (None, "", "null"): die(1, f"FAIL {hostname} has no custom_fields.cloud_id in NetBox") - # 1) Liveness gate: proceed only if online + # 1) Liveness gate live = cl.device_liveness(cloud_id) if not bool(live.get("isConnected")): die(1, f"FAIL {hostname} cloud_id={cloud_id} device is not online") current_nb_ip = _primary_ip4_text(dev_full) - # 2) Cloud detail (now that we know it's online) + # 2) Cloud detail d = cl.device_detail(cloud_id) fw = d.get("firmwareVersion") or d.get("version") ip_from_cloud = (d.get("ipAddress") or "").strip() if isinstance(d.get("ipAddress"), str) else d.get("ipAddress") @@ -248,7 +277,7 @@ def run(hostname: str) -> None: if not fw: die(1, f"FAIL {hostname} cloud_id={cloud_id}: missing firmwareVersion") - # 3) NetBox patch (change-only) + # 3) NetBox patch (idempotent) cf_patch = {} if cf.get("fw_version") != fw: cf_patch["fw_version"] = fw @@ -276,7 +305,7 @@ def run(hostname: str) -> None: ip_out_for_status = current_nb_ip # default to current NB IP if ip_is_placeholder: - log(f"IP: cloud reported placeholder '{ip_from_cloud}', skipping IP changes; keeping NetBox ip={current_nb_ip}") + log(f"IP: cloud reported placeholder '{ip_from_cloud}', skipping IP changes; keeping NetBox ip(s) as-is") else: iface_id = nb.ensure_eth0(dev_id) @@ -312,6 +341,16 @@ def run(hostname: str) -> None: nb.device_set_primary_ip4(dev_id, ip_id) ip_out_for_status = ip_from_cloud + # PRUNE all other IPs on this device (default behavior) + all_ips = nb.list_device_ips(dev_id) + for rec in all_ips: + rid = rec.get("id") + if str(rid) == str(ip_id): + continue + addr = rec.get("address") + log(f"IP: pruning stale {addr} (id={rid}) from device {dev_id}") + nb.delete_ip(rid) + print(f"OK {hostname} ip={ip_out_for_status or 'NONE'} fw={fw} node={node} sector={sector} small={small}") # ------------ CLI ------------ @@ -321,7 +360,7 @@ if __name__ == "__main__": ap.add_argument("--chatty", action="store_true", help="Verbose step-by-step logging to stderr") args = ap.parse_args() - CHATTY = bool(args.chatty) # module-scope assignment (no global needed) + CHATTY = bool(args.chatty) # module-scope assignment try: run(args.hostname) except requests.RequestException as e: