graham/network
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
network/scripts/cpe_mac_audit.py
Graham McIntire d0d1667991
update
2026-07-19 14:42:02 -05:00

774 lines
31 KiB
Python
Executable file
Raw Permalink Blame History

#!/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()
Reference in a new issue View git blame Copy permalink