16:19

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 valid (not None/""/"0.0.0.0"):
-   - If Cloud ipAddress is placeholder/invalid: skip IP changes; keep NetBox IP as-is.
+       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: