#!/usr/bin/env python3 """Audit CPE MAC addresses: compare Gaiia assignments against live DHCP leases. For every active (paying) customer in Gaiia, find their assigned CPE MAC(s), then cross-reference against live DHCP leases on all tower routers. Flags: MAC_MISMATCH -- DHCP host-name matches a Gaiia account, but the lease MAC belongs to a *different* account in Gaiia (or no account). MAC_UNKNOWN -- DHCP lease MAC not found in Gaiia inventory at all, but host-name fuzzy-matches an active account (inventory gap). GAIIA_ORPHAN -- Active customer's CPE MAC is in Gaiia but NOT in any DHCP lease (device offline / replaced / MAC wrong in Gaiia). Usage: GAIIA_KEY=... ./scripts/cpe_mac_audit.py GAIIA_KEY=... ./scripts/cpe_mac_audit.py --routers 982,core GAIIA_KEY=... ./scripts/cpe_mac_audit.py --json > audit.json GAIIA_KEY=... ./scripts/cpe_mac_audit.py --verbose # also show OK entries """ from __future__ import annotations import argparse import json import os import ipaddress import re import subprocess import sys import time from concurrent.futures import ThreadPoolExecutor from typing import Any REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) sys.path.insert(0, os.path.join(REPO, "gaiia", "src")) from gaiia import GaiiaClient, GaiiaTransportError # noqa: E402 TOWER_ROUTERS = ["verona", "climax", "culleoka", "newhope", "lowry", "982", "494", "core"] MTT = os.path.join(REPO, "mikrotik-tool", "mikrotik-tool") CPE_NET = ipaddress.IPv4Network("10.10.0.0/16") PAYING = {"ACTIVE", "TRIAL"} # Indoor routers / WiFi APs — not CPE radios, skip them. _INDOOR_PATTERNS = ("AC1200", "AC1000", "Vilo", "hc220", "GWN") # MAC OUI → vendor / device type hints for "needs creation" reporting. # First 3 bytes of the MAC (upper case, colon-separated). _OUI_HINTS: dict[str, str] = {} for _prefix, _hint in [ # Ubiquiti ("00:04:56", "Ubiquiti (legacy)"), ("00:27:22", "Ubiquiti (legacy)"), ("00:01:9F", "Ubiquiti AC1000/AC1200"), ("04:18:D6", "Ubiquiti Loco M5"), ("24:A4:3C", "Ubiquiti Loco M5"), ("24:2F:D0", "Ubiquiti AC1200-Mesh"), ("24:5A:4C", "Ubiquiti"), ("5C:62:88", "Ubiquiti"), ("68:72:51", "Ubiquiti"), ("68:D7:9A", "Ubiquiti Power Beam AC 2.4"), ("70:A7:41", "Ubiquiti Nanobeam AC 2.4"), ("74:83:C2", "Ubiquiti Loco M5"), ("74:AC:B9", "Ubiquiti PowerBeam M5"), ("74:FE:CE", "Ubiquiti AC1200-Mesh"), ("78:45:58", "Ubiquiti PowerBeam AC Gen2"), ("78:8A:20", "Ubiquiti Loco M5"), ("80:2A:A8", "Ubiquiti Loco M5"), ("8C:90:2D", "Ubiquiti AC1200-Mesh"), ("98:03:8E", "Ubiquiti AC1200-Mesh"), ("98:25:4A", "Ubiquiti AC1200-Mesh"), ("A8:5E:45", "Ubiquiti"), ("B0:19:21", "Ubiquiti AC1200-Mesh"), ("B4:FB:E4", "Ubiquiti Loco M5"), ("CC:2D:E0", "Ubiquiti wAP 60G"), ("D0:21:F9", "Ubiquiti"), ("D8:44:89", "Ubiquiti AC1200-Mesh"), ("DC:9F:DB", "Ubiquiti"), ("E0:63:DA", "Ubiquiti PowerBeam M5 / Loco M5"), ("E4:38:83", "Ubiquiti PowerBeam / LiteBeam"), ("E8:DA:00", "Ubiquiti"), ("EC:74:D7", "Ubiquiti GWN"), ("F0:9F:C2", "Ubiquiti"), ("F0:A7:31", "Ubiquiti AC1200-Mesh"), ("F4:92:BF", "Ubiquiti Loco M5 / PowerBeam"), ("F4:E2:C6", "Ubiquiti PowerBeam AC Gen2"), ("FC:EC:DA", "Ubiquiti LiteBeam / Loco"), # Cambium ("58:C1:7A", "Cambium ePMP"), # MikroTik ("18:E8:29", "MikroTik"), ("00:15:6D", "Ubiquiti NanoBridge M900"), ]: _OUI_HINTS[_prefix] = _hint def guess_device(mac: str) -> str: """Best-effort device type from MAC OUI prefix.""" n = fmt_mac(mac) if not n: return "unknown" prefix = n[:8] # first 3 bytes: AA:BB:CC return _OUI_HINTS.get(prefix, n[:8]) # ---------- helpers ---------- def norm_mac(s: str) -> str: return re.sub(r"[^0-9a-f]", "", (s or "").lower()) def fmt_mac(s: str) -> str: n = norm_mac(s) return ":".join(n[i:i+2] for i in range(0, 12, 2)).upper() if len(n) == 12 else s def name_tokens(s: str) -> set[str]: return {w for w in re.split(r"[^a-z0-9]+", s.lower()) if len(w) >= 3} def parse_kv_line(line: str) -> dict[str, str]: kv: dict[str, str] = {} for part in re.split(r" +", line.strip()): if "=" in part: k, v = part.split("=", 1) kv[k] = v return kv # ---------- MikroTik ---------- def fetch_dhcp_leases(router: str) -> tuple[list[dict[str, Any]], bool]: """Return (leases, reachable). reachable=False means the router timed out or errored.""" p = subprocess.run( [MTT, "api", router, "/ip/dhcp-server/lease/print"], capture_output=True, text=True, timeout=90, cwd=os.path.join(REPO, "mikrotik-tool"), ) if p.returncode != 0: print(f"!! {router}: {p.stderr.strip()[:200]}", file=sys.stderr) return [], False out: list[dict[str, Any]] = [] for line in p.stdout.splitlines(): if not line.startswith(".id="): continue kv = parse_kv_line(line) if kv.get("status") != "bound": continue mac = kv.get("active-mac-address") or kv.get("mac-address") or "" if not mac: continue addr = kv.get("active-address") or kv.get("address") or "" if addr: try: if ipaddress.IPv4Address(addr) not in CPE_NET: continue except ValueError: continue out.append({ "router": router, "mac": norm_mac(mac), "mac_raw": mac, "address": addr, "host": kv.get("host-name", ""), }) return out, True # ---------- Gaiia ---------- def _retry_query(client: GaiiaClient, query: str, variables: dict | None = None, max_tries: int = 5): """Run a Gaiia query with retries on transport errors.""" for attempt in range(1, max_tries + 1): try: return client.query(query, variables) except GaiiaTransportError: if attempt == max_tries: raise delay = min(2 ** attempt, 30) print(f" (transport error, retrying in {delay}s — attempt {attempt}/{max_tries})", file=sys.stderr) time.sleep(delay) def paginate(client: GaiiaClient, q_with: str, q_first: str, path: str): cur: str | None = None while True: d = _retry_query(client, q_first) if cur is None else _retry_query(client, q_with, {"after": cur}) node = d for part in path.split("."): node = node[part] for n in node.get("nodes") or []: yield n pi = node.get("pageInfo") or {} if not pi.get("hasNextPage"): return cur = pi.get("endCursor") if not cur: return def fetch_gaiia() -> tuple[dict[str, dict], dict[str, dict], dict[str, dict]]: """Return (accounts_by_id, mac_to_cpe, mac_to_any). mac_to_cpe: outdoor CPE radios only (used for the main audit). mac_to_any: ALL inventory items including indoor routers (used to resolve unknown MACs).""" accounts: dict[str, dict] = {} q_acct_with = """query Q($after:String!){ accounts(first:100,after:$after){ nodes{ id readableId displayName status{ name } billingSubscriptions(first:50){ nodes{ status } } } pageInfo{ hasNextPage endCursor } } }""" q_acct_first = """query{ accounts(first:100){ nodes{ id readableId displayName status{ name } billingSubscriptions(first:50){ nodes{ status } } } pageInfo{ hasNextPage endCursor } } }""" mac_to_cpe: dict[str, dict] = {} mac_to_any: dict[str, dict] = {} q_inv_with = """query Q($after:String!){ inventoryItems(first:100,after:$after){ nodes{ id model{ name } fields{ nodes{ data modelField{ name } } } assignation{ assigneeType assignee{ __typename ... on Account{ id } } } } pageInfo{ hasNextPage endCursor } } }""" q_inv_first = q_inv_with.replace("query Q($after:String!)", "query").replace(",after:$after", "") def _add_macs(macs, item_id, model, acct_id, assignee_type, target_map): for m in macs: if m in target_map and target_map[m].get("account_id"): continue target_map[m] = { "item_id": item_id, "model": model, "account_id": acct_id, "assignee_type": assignee_type, } with GaiiaClient(timezone="America/Chicago") as g: for a in paginate(g, q_acct_with, q_acct_first, "accounts"): subs = [s["status"] for s in ((a.get("billingSubscriptions") or {}).get("nodes") or [])] accounts[a["id"]] = { "id": a["id"], "readable": a["readableId"], "name": a["displayName"], "status": a["status"]["name"], "paying": any(s in PAYING for s in subs), "subs": subs, } for it in paginate(g, q_inv_with, q_inv_first, "inventoryItems"): model = (it.get("model") or {}).get("name", "") is_indoor = any(p.lower() in model.lower() for p in _INDOOR_PATTERNS) macs = [] for f in ((it.get("fields") or {}).get("nodes") or []): mf = f.get("modelField") or {} if "mac" in mf.get("name", "").lower() and f.get("data"): m = norm_mac(f["data"]) if len(m) == 12: macs.append(m) assn = it.get("assignation") or {} assignee = assn.get("assignee") or {} assignee_type = assn.get("assigneeType") or None # None = truly unassigned acct_id = assignee["id"] if assignee.get("__typename") == "Account" else None _add_macs(macs, it["id"], model, acct_id, assignee_type, mac_to_any) if not is_indoor: _add_macs(macs, it["id"], model, acct_id, assignee_type, mac_to_cpe) return accounts, mac_to_cpe, mac_to_any # ---------- matching ---------- def fuzzy_match_host(host: str, accounts: dict[str, dict]) -> dict | None: """Match a DHCP host-name to a Gaiia account by displayName tokens. Requires >=2 token overlap to reduce false positives.""" if not host: return None htoks = name_tokens(host) if len(htoks) < 2: return None best, best_score = None, 0 for aid, a in accounts.items(): atoks = name_tokens(a["name"]) score = len(htoks & atoks) if score >= 2 and score > best_score: best_score = score best = {**a, "id": aid, "match_score": score} return best # ---------- audit ---------- def audit(leases, accounts, mac_to_cpe, mac_to_any): """Compare DHCP leases against Gaiia assignments. Returns categorized results.""" paying = {aid: a for aid, a in accounts.items() if a["paying"]} mismatches: list[dict] = [] # MAC belongs to different account than host suggests unknowns: list[dict] = [] # MAC not in mac_to_cpe, but host matches paying account unassigned: list[dict] = [] # MAC in Gaiia but not assigned to any account ok_entries: list[dict] = [] # everything matches unmatched_host: list[dict] = [] # MAC in Gaiia, assigned, but host doesn't match any account # MAC_UNKNOWN sub-categories, resolved against mac_to_any (full inventory) fixable: list[dict] = [] # item exists, unassigned → can auto-assign exists_assigned: list[dict] = [] # item exists, assigned to a different account needs_creation: list[dict] = [] # not in Gaiia inventory at all # Track which Gaiia MACs we've seen in DHCP seen_gaiia_macs: set[str] = set() for lease in leases: mac = lease["mac"] gaiia = mac_to_cpe.get(mac) if gaiia: seen_gaiia_macs.add(mac) gaiia_acct_id = gaiia.get("account_id") gaiia_acct = accounts.get(gaiia_acct_id) if gaiia_acct_id else None gaiia_has_assignee = bool(gaiia.get("assignee_type")) else: gaiia_acct_id = None gaiia_acct = None gaiia_has_assignee = False host_match = fuzzy_match_host(lease["host"], accounts) entry = { "lease": lease, "gaiia_mac_owner": gaiia_acct, "host_match": host_match, } if gaiia_acct and host_match: if gaiia_acct["id"] == host_match["id"]: ok_entries.append(entry) else: mismatches.append(entry) elif gaiia_acct and not host_match: unmatched_host.append(entry) elif not gaiia_acct and host_match: if host_match["paying"]: unknowns.append(entry) # Resolve against full inventory (including indoor routers) any_item = mac_to_any.get(mac) if not any_item: # MAC genuinely absent from Gaiia inventory. needs_creation.append(entry) elif any_item.get("account_id"): # Exists but assigned to a *different* account. owner = accounts.get(any_item["account_id"]) exists_assigned.append({ **entry, "existing_item": any_item, "existing_owner_name": owner["name"] if owner else None, }) else: # Exists but not on any account: truly unassigned OR parked at an # inventory location/site. Either way, reassign to matched account. fixable.append({**entry, "existing_item": any_item}) elif gaiia and not gaiia_acct and not gaiia_has_assignee: unassigned.append(entry) # else: lease with no Gaiia MAC match and no host match → truly unknown, skip # Items assigned to sites/locations are not flagged (silently correct). # Gaiia orphans: paying customers whose CPE MACs are NOT in any DHCP lease orphans: list[dict] = [] for mac, info in mac_to_cpe.items(): acct_id = info.get("account_id") if not acct_id or acct_id not in paying: continue if mac not in seen_gaiia_macs: orphans.append({ "mac": mac, "mac_formatted": fmt_mac(mac), "account": paying[acct_id], "item_id": info["item_id"], "model": info["model"], }) return { "mismatches": mismatches, "unknowns": unknowns, "fixable": fixable, "exists_assigned": exists_assigned, "needs_creation": needs_creation, "unassigned": unassigned, "unmatched_host": unmatched_host, "orphans": orphans, "ok": ok_entries, "mac_to_any": mac_to_any, } # ---------- output ---------- def print_report(results, verbose=False): m = results["mismatches"] u = results["unknowns"] fixable = results["fixable"] exists_assigned = results["exists_assigned"] needs_creation = results["needs_creation"] na = results["unassigned"] uh = results["unmatched_host"] orph = results["orphans"] stale = results.get("stale_orphans", []) ok = results["ok"] print(f"\n=== CPE MAC AUDIT ===") print(f" {len(ok)} OK | {len(m)} MAC_MISMATCH | {len(u)} MAC_UNKNOWN | " f"{len(orph)} GAIIA_ORPHAN | {len(stale)} STALE") print(f" ({len(fixable)} fixable | {len(exists_assigned)} conflict | " f"{len(needs_creation)} needs-creation)") print(f" ({len(na)} unassigned | {len(uh)} unmatched host)") if m: print(f"\n--- {len(m)} MAC MISMATCH (DHCP host-name matches account A, " f"but lease MAC belongs to account B) ---") for e in sorted(m, key=lambda x: (x["lease"]["router"], x["lease"]["address"])): l = e["lease"] g_owner = e["gaiia_mac_owner"] h_match = e["host_match"] print(f" {l['router']:9} {l['address']:16} {fmt_mac(l['mac']):17} " f"host={l['host']!r}") print(f" -> host suggests: #{h_match['readable']} {h_match['name']!r} " f"(paying={h_match['paying']})") print(f" -> MAC belongs to: #{g_owner['readable']} {g_owner['name']!r} " f"(paying={g_owner['paying']})") if fixable: print(f"\n--- {len(fixable)} FIXABLE (MAC exists in Gaiia — unassigned or parked " f"at a location — auto-fix with --fix) ---") for e in sorted(fixable, key=lambda x: (x["lease"]["router"], x["lease"]["address"])): l = e["lease"] h = e["host_match"] item = e["existing_item"] cur = item.get("assignee_type") or "UNASSIGNED" print(f" {l['router']:9} {l['address']:16} {fmt_mac(l['mac']):17} " f"host={l['host']!r}") print(f" -> assign to: #{h['readable']} {h['name']!r} " f"item={item['item_id']} model={item['model']} currently={cur}") if exists_assigned: print(f"\n--- {len(exists_assigned)} CONFLICT (MAC exists in Gaiia but assigned to different account) ---") for e in sorted(exists_assigned, key=lambda x: (x["lease"]["router"], x["lease"]["address"])): l = e["lease"] h = e["host_match"] item = e["existing_item"] owner_name = e.get("existing_owner_name") or "unassigned" print(f" {l['router']:9} {l['address']:16} {fmt_mac(l['mac']):17} " f"host={l['host']!r}") print(f" -> host suggests: #{h['readable']} {h['name']!r}") print(f" -> item {item['item_id']} ({item['model']}) already assigned to: {owner_name!r}") if needs_creation: print(f"\n--- {len(needs_creation)} NEEDS CREATION (MAC not in Gaiia — " f"add device and assign to account) ---") for e in sorted(needs_creation, key=lambda x: (x["lease"]["router"], x["lease"]["address"])): l = e["lease"] h = e["host_match"] dev = guess_device(l["mac"]) print(f" router={l['router']:9} ip={l['address']:16} " f"mac={fmt_mac(l['mac']):17} device={dev}") print(f" account=#{h['readable']} name={h['name']!r} " f"host={l['host']!r}") if orph: print(f"\n--- {len(orph)} GAIIA ORPHAN (not in DHCP, tower may be unreachable) ---") for e in sorted(orph, key=lambda x: x["account"]["name"]): a = e["account"] print(f" {e['mac_formatted']:17} #{a['readable']:6} {a['name']!r} " f"model={e['model']} item={e['item_id']}") if stale: print(f"\n--- {len(stale)} STALE (on reachable tower but not in DHCP — " f"unassign with --fix) ---") for e in sorted(stale, key=lambda x: x["account"]["name"]): a = e["account"] print(f" {e['mac_formatted']:17} #{a['readable']:6} {a['name']!r} " f"model={e['model']} item={e['item_id']}") if na: print(f"\n--- {len(na)} UNASSIGNED (MAC in Gaiia inventory but not assigned to any account) ---") for e in sorted(na, key=lambda x: (x["lease"]["router"], x["lease"]["address"])): l = e["lease"] print(f" {l['router']:9} {l['address']:16} {fmt_mac(l['mac']):17} " f"host={l['host']!r}") if uh: print(f"\n--- {len(uh)} UNMATCHED HOST (MAC assigned in Gaiia, host-name doesn't match any account) ---") for e in sorted(uh, key=lambda x: (x["lease"]["router"], x["lease"]["address"])): l = e["lease"] g = e["gaiia_mac_owner"] print(f" {l['router']:9} {l['address']:16} {fmt_mac(l['mac']):17} " f"host={l['host']!r}") print(f" -> MAC belongs to: #{g['readable']} {g['name']!r} (paying={g['paying']})") if verbose and ok: print(f"\n--- {len(ok)} OK (MAC matches expected account) ---") for e in sorted(ok, key=lambda x: (x["lease"]["router"], x["lease"]["address"])): l = e["lease"] a = e["gaiia_mac_owner"] print(f" {l['router']:9} {l['address']:16} {fmt_mac(l['mac']):17} " f"host={l['host']!r} acct=#{a['readable']} {a['name']!r}") if not any([m, u, orph, stale, na, uh]): print("\nNo issues found.") def json_report(results): def strip(obj): if isinstance(obj, dict): return {k: strip(v) for k, v in obj.items()} if isinstance(obj, list): return [strip(v) for v in obj] return obj out = { "summary": { "ok": len(results["ok"]), "mac_mismatch": len(results["mismatches"]), "mac_unknown": len(results["unknowns"]), "fixable": len(results["fixable"]), "exists_assigned": len(results["exists_assigned"]), "needs_creation": len(results["needs_creation"]), "gaiia_orphan": len(results["orphans"]), "unassigned": len(results["unassigned"]), "unmatched_host": len(results["unmatched_host"]), }, "mismatches": [ { "lease": {**e["lease"], "mac": fmt_mac(e["lease"]["mac"])}, "gaiia_mac_owner": e["gaiia_mac_owner"], "host_match": e["host_match"], } for e in results["mismatches"] ], "fixable": [ { "lease": {**e["lease"], "mac": fmt_mac(e["lease"]["mac"])}, "host_match": e["host_match"], "existing_item": e["existing_item"], } for e in results["fixable"] ], "exists_assigned": [ { "lease": {**e["lease"], "mac": fmt_mac(e["lease"]["mac"])}, "host_match": e["host_match"], "existing_item": e["existing_item"], "existing_owner_name": e.get("existing_owner_name"), } for e in results["exists_assigned"] ], "needs_creation": [ { "lease": {**e["lease"], "mac": fmt_mac(e["lease"]["mac"])}, "host_match": e["host_match"], "device_guess": guess_device(e["lease"]["mac"]), } for e in results["needs_creation"] ], "orphans": results["orphans"], "unassigned": [ { "lease": {**e["lease"], "mac": fmt_mac(e["lease"]["mac"])}, } for e in results["unassigned"] ], "unmatched_host": [ { "lease": {**e["lease"], "mac": fmt_mac(e["lease"]["mac"])}, "gaiia_mac_owner": e["gaiia_mac_owner"], } for e in results["unmatched_host"] ], } json.dump(strip(out), sys.stdout, indent=2, default=str) # ---------- main ---------- def do_fixes(client, fixable): """Call assignInventoryItem for each fixable entry.""" mut = """mutation M($i: AssignInventoryItemInput!) { assignInventoryItem(input: $i) { errors { code message } } }""" ok = 0 for e in fixable: h = e["host_match"] item = e["existing_item"] l = e["lease"] try: client.mutate(mut, "assignInventoryItem", {"i": {"inventoryItemId": item["item_id"], "assigneeType": "ACCOUNT", "assigneeId": h["id"]}}) print(f" assigned {item['item_id']} ({item['model']}) -> #{h['readable']} " f"{h['name']!r} [MAC {fmt_mac(l['mac'])}]", file=sys.stderr) ok += 1 except Exception as exc: print(f" FAILED {item['item_id']}: {exc}", file=sys.stderr) return ok def do_reassigns(client, conflicts): """Reassign inventory items from the wrong account to the matched account.""" mut = """mutation M($i: AssignInventoryItemInput!) { assignInventoryItem(input: $i) { errors { code message } } }""" ok = 0 for e in conflicts: h = e["host_match"] item = e["existing_item"] l = e["lease"] old = e.get("existing_owner_name") or "unknown" try: client.mutate(mut, "assignInventoryItem", {"i": {"inventoryItemId": item["item_id"], "assigneeType": "ACCOUNT", "assigneeId": h["id"]}}) print(f" reassigned {item['item_id']} ({item['model']}) " f"from {old!r} -> #{h['readable']} {h['name']!r} " f"[MAC {fmt_mac(l['mac'])}]", file=sys.stderr) ok += 1 except Exception as exc: print(f" FAILED {item['item_id']}: {exc}", file=sys.stderr) return ok def do_unassigns(client, stale_orphans, leases, accounts): """Reassign stale CPEs to their tower's inventory location.""" # Fetch all inventory locations locs = {} q_loc = """query { inventoryLocations(first: 100) { nodes { id name } pageInfo { hasNextPage endCursor } }}""" r = client.query(q_loc) for loc in (r.get("inventoryLocations") or {}).get("nodes") or []: locs[loc["name"].lower()] = loc["id"] # Map account ID -> router from DHCP leases acct_to_router: dict[str, str] = {} for l in leases: host_match = fuzzy_match_host(l["host"], accounts) if host_match and host_match["paying"]: if host_match["id"] not in acct_to_router: acct_to_router[host_match["id"]] = l["router"] # Map router -> location (fuzzy: router name should match location name) # Known mappings from the fleet ROUTER_TO_LOC: dict[str, str] = { "verona": "verona", "climax": "climax", "culleoka": "culleoka", "newhope": "newhope", "lowry": "lowrycrossing", "982": "982", "494": "494", "core": "380", } mut = """mutation M($i: AssignInventoryItemInput!) { assignInventoryItem(input: $i) { errors { code message } } }""" ok = 0 for o in stale_orphans: acct_id = o["account"]["id"] router = acct_to_router.get(acct_id) loc_key = ROUTER_TO_LOC.get(router) if router else None loc_id = locs.get(loc_key) if loc_key else None if not loc_id: print(f" SKIP {o['item_id']} ({o['model']}): " f"can't determine location (router={router})", file=sys.stderr) continue try: client.mutate(mut, "assignInventoryItem", {"i": {"inventoryItemId": o["item_id"], "assigneeType": "INVENTORY_LOCATION", "assigneeId": loc_id}}) a = o["account"] print(f" unassigned {o['item_id']} ({o['model']}) from #{a['readable']} " f"{a['name']!r} -> {loc_key}", file=sys.stderr) ok += 1 except Exception as exc: print(f" FAILED {o['item_id']}: {exc}", file=sys.stderr) return ok def main(): ap = argparse.ArgumentParser(description="Audit CPE MACs: Gaiia vs live DHCP leases") ap.add_argument("--routers", help="comma-separated subset of tower routers") ap.add_argument("--json", action="store_true", help="emit machine-readable JSON") ap.add_argument("--verbose", "-v", action="store_true", help="also show OK entries") ap.add_argument("--fix", action="store_true", help="auto-assign unassigned inventory items to matched accounts") args = ap.parse_args() if not os.environ.get("GAIIA_KEY"): sys.exit("GAIIA_KEY not set") routers = args.routers.split(",") if args.routers else TOWER_ROUTERS print("Fetching Gaiia accounts + inventory...", file=sys.stderr) accounts, mac_to_cpe, mac_to_any = fetch_gaiia() paying = sum(1 for a in accounts.values() if a["paying"]) cpe_assigned = sum(1 for v in mac_to_cpe.values() if v.get("account_id")) print(f" {len(accounts)} accounts ({paying} paying), " f"{len(mac_to_cpe)} outdoor CPEs in inventory ({cpe_assigned} account-assigned), " f"{len(mac_to_any)} total items", file=sys.stderr) print(f"Fetching DHCP leases from {len(routers)} routers...", file=sys.stderr) with ThreadPoolExecutor(max_workers=len(routers)) as ex: results_list = list(ex.map(fetch_dhcp_leases, routers)) lease_lists = [r[0] for r in results_list] reachable = {router: ok for router, (_, ok) in zip(routers, results_list)} leases = [l for sub in lease_lists for l in sub] print(f" {len(leases)} bound leases in 10.10.0.0/16 " f"({sum(1 for v in reachable.values() if not v)} unreachable)", file=sys.stderr) results = audit(leases, accounts, mac_to_cpe, mac_to_any) # Separate orphans into stale (on reachable towers) vs unknown # An orphan is stale if the account has at least one DHCP lease on a reachable router. accts_with_leases = {e["host_match"]["id"] for e in results["ok"] + results["mismatches"] if e.get("host_match")} accts_with_leases |= {e["host_match"]["id"] for e in results["unknowns"] if e.get("host_match")} # Also include account IDs from unmatched_host (MAC is in Gaiia, assigned) for e in results["unmatched_host"]: if e.get("gaiia_mac_owner"): accts_with_leases.add(e["gaiia_mac_owner"]["id"]) stale_orphans = [o for o in results["orphans"] if o["account"]["id"] in accts_with_leases] unknown_orphans = [o for o in results["orphans"] if o["account"]["id"] not in accts_with_leases] results["stale_orphans"] = stale_orphans results["unknown_orphans"] = unknown_orphans # Replace orphans with unknown-only for display results["orphans"] = unknown_orphans if args.fix: with GaiiaClient(timezone="America/Chicago") as g: if results["fixable"]: print(f"\nApplying {len(results['fixable'])} assignments...", file=sys.stderr) n = do_fixes(g, results["fixable"]) print(f" {n}/{len(results['fixable'])} assigned", file=sys.stderr) if results["exists_assigned"]: print(f"\nReassigning {len(results['exists_assigned'])} conflicting items...", file=sys.stderr) n = do_reassigns(g, results["exists_assigned"]) print(f" {n}/{len(results['exists_assigned'])} reassigned", file=sys.stderr) if stale_orphans: print(f"\nUnassigning {len(stale_orphans)} stale CPEs " f"(on reachable towers but not in DHCP)...", file=sys.stderr) n = do_unassigns(g, stale_orphans, leases, accounts) print(f" {n}/{len(stale_orphans)} unassigned", file=sys.stderr) if args.json: json_report(results) else: print_report(results, verbose=args.verbose) if __name__ == "__main__": main()