344 lines
12 KiB
Python
Executable file
344 lines
12 KiB
Python
Executable file
#!/usr/bin/env python3
|
|
"""PPPoE-side counterpart to freeloaders.py.
|
|
|
|
Pulls /ppp/active/print from every tower router and classifies each active
|
|
PPPoE session against Gaiia inventory + accounts, the same way freeloaders.py
|
|
does for DHCP leases. The goal is to identify which active PPPoE customers
|
|
won't be mapped in Preseem because Gaiia can't tie their session to a paying
|
|
account.
|
|
|
|
Match order for each active session:
|
|
1. caller-id MAC -> InventoryItem (mac field) -> Account
|
|
2. PPPoE username -> Account displayName (fuzzy smushed-name match;
|
|
"judydevine" -> "Judy Devine")
|
|
|
|
Categories:
|
|
OK_PAYING - resolved to ACTIVE/TRIAL account (would be mapped in Preseem)
|
|
UNPAID - resolved to account whose subscriptions are all
|
|
CANCELLED/SUSPENDED/etc. (Preseem sync presumably skips)
|
|
UNASSIGNED - MAC is in Gaiia inventory but not assigned to an Account
|
|
UNMATCHED - neither MAC nor name resolves to anything in Gaiia
|
|
|
|
Usage:
|
|
GAIIA_KEY=... ./scripts/pppoe_freeloaders.py
|
|
GAIIA_KEY=... ./scripts/pppoe_freeloaders.py --json > out.json
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
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 # noqa: E402
|
|
|
|
TOWER_ROUTERS = ["verona", "climax", "culleoka", "newhope", "lowry", "982", "494", "core"]
|
|
MTT = os.path.join(REPO, "mikrotik-tool", "mikrotik-tool")
|
|
|
|
PAYING_STATUSES = {"ACTIVE", "TRIAL"}
|
|
|
|
|
|
def norm_mac(s: str) -> str:
|
|
return re.sub(r"[^0-9a-f]", "", (s or "").lower())
|
|
|
|
|
|
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
|
|
|
|
|
|
def fetch_ppp_active(router: str) -> list[dict[str, Any]]:
|
|
proc = subprocess.run(
|
|
[MTT, "api", router, "/ppp/active/print"],
|
|
capture_output=True, text=True, timeout=90,
|
|
cwd=os.path.join(REPO, "mikrotik-tool"),
|
|
)
|
|
if proc.returncode != 0:
|
|
print(f"!! {router}: {proc.stderr.strip()[:200]}", file=sys.stderr)
|
|
return []
|
|
out: list[dict[str, Any]] = []
|
|
for line in proc.stdout.splitlines():
|
|
if not line.startswith(".id="):
|
|
continue
|
|
kv = parse_kv_line(line)
|
|
if kv.get("service") != "pppoe":
|
|
continue
|
|
out.append({
|
|
"router": router,
|
|
"username": kv.get("name", ""),
|
|
"address": kv.get("address", ""),
|
|
"caller_mac": norm_mac(kv.get("caller-id", "")),
|
|
"caller_mac_raw": kv.get("caller-id", ""),
|
|
"uptime": kv.get("uptime", ""),
|
|
})
|
|
return out
|
|
|
|
|
|
def paginate(client: GaiiaClient, q_with: str, q_first: str, path: str):
|
|
cursor: str | None = None
|
|
while True:
|
|
data = client.query(q_first) if cursor is None else client.query(q_with, {"after": cursor})
|
|
node = data
|
|
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
|
|
cursor = pi.get("endCursor")
|
|
if not cursor:
|
|
return
|
|
|
|
|
|
def fetch_gaiia() -> tuple[dict[str, dict], dict[str, dict]]:
|
|
"""Return (mac -> inventory record, account_id -> account record)."""
|
|
mac_to_item: dict[str, dict] = {}
|
|
account_info: dict[str, dict] = {}
|
|
|
|
accts_with = """
|
|
query Q($after: String!) {
|
|
accounts(first: 100, after: $after) {
|
|
nodes {
|
|
id readableId displayName
|
|
status { name }
|
|
billingSubscriptions(first: 50) { nodes { status } }
|
|
}
|
|
pageInfo { hasNextPage endCursor }
|
|
}
|
|
}"""
|
|
accts_first = """
|
|
query {
|
|
accounts(first: 100) {
|
|
nodes {
|
|
id readableId displayName
|
|
status { name }
|
|
billingSubscriptions(first: 50) { nodes { status } }
|
|
}
|
|
pageInfo { hasNextPage endCursor }
|
|
}
|
|
}"""
|
|
|
|
items_with = """
|
|
query Q($after: String!) {
|
|
inventoryItems(first: 100, after: $after) {
|
|
nodes {
|
|
id ipAddressV4
|
|
model { name }
|
|
fields { nodes { data modelField { name } } }
|
|
assignation {
|
|
assigneeType
|
|
assignee { __typename ... on Account { id readableId displayName } }
|
|
}
|
|
}
|
|
pageInfo { hasNextPage endCursor }
|
|
}
|
|
}"""
|
|
items_first = items_with.replace("$after: String!", "").replace("after: $after", "").replace(", )", ")")
|
|
# simpler: hand-roll the first query
|
|
items_first = """
|
|
query {
|
|
inventoryItems(first: 100) {
|
|
nodes {
|
|
id ipAddressV4
|
|
model { name }
|
|
fields { nodes { data modelField { name } } }
|
|
assignation {
|
|
assigneeType
|
|
assignee { __typename ... on Account { id readableId displayName } }
|
|
}
|
|
}
|
|
pageInfo { hasNextPage endCursor }
|
|
}
|
|
}"""
|
|
|
|
with GaiiaClient(timezone="America/Chicago") as g:
|
|
for a in paginate(g, accts_with, accts_first, "accounts"):
|
|
subs = (a.get("billingSubscriptions") or {}).get("nodes") or []
|
|
statuses = [s["status"] for s in subs]
|
|
account_info[a["id"]] = {
|
|
"readable": a["readableId"],
|
|
"name": a["displayName"],
|
|
"status": a["status"]["name"],
|
|
"paying": any(s in PAYING_STATUSES for s in statuses),
|
|
"sub_statuses": statuses,
|
|
}
|
|
|
|
for it in paginate(g, items_with, items_first, "inventoryItems"):
|
|
macs = []
|
|
for f in ((it.get("fields") or {}).get("nodes") or []):
|
|
mf = f.get("modelField") or {}
|
|
if mf.get("name", "").lower().startswith("mac") and f.get("data"):
|
|
macs.append(norm_mac(f["data"]))
|
|
assn = it.get("assignation") or {}
|
|
assignee = assn.get("assignee") or {}
|
|
atype = assignee.get("__typename")
|
|
for m in macs:
|
|
if len(m) != 12:
|
|
continue
|
|
rec = {
|
|
"item_id": it["id"],
|
|
"model": (it.get("model") or {}).get("name", ""),
|
|
"ip": it.get("ipAddressV4"),
|
|
"assignee_type": atype,
|
|
}
|
|
if atype == "Account":
|
|
rec["account_id"] = assignee["id"]
|
|
# Prefer Account-assigned items if MAC is duplicated
|
|
if m in mac_to_item and mac_to_item[m].get("assignee_type") == "Account":
|
|
continue
|
|
mac_to_item[m] = rec
|
|
|
|
return mac_to_item, account_info
|
|
|
|
|
|
def smush(s: str) -> str:
|
|
"""Lowercase + drop non-alphanumeric. 'Judy Devine' -> 'judydevine'."""
|
|
return re.sub(r"[^a-z0-9]", "", (s or "").lower())
|
|
|
|
|
|
def build_username_index(accounts: dict[str, dict]) -> dict[str, list[str]]:
|
|
"""Smushed displayName -> [account_id, ...]. Many accounts can collide."""
|
|
idx: dict[str, list[str]] = {}
|
|
for aid, a in accounts.items():
|
|
key = smush(a["name"])
|
|
if key:
|
|
idx.setdefault(key, []).append(aid)
|
|
return idx
|
|
|
|
|
|
def classify(sessions, mac_to_item, accounts):
|
|
user_idx = build_username_index(accounts)
|
|
|
|
results = []
|
|
for s in sessions:
|
|
username = s["username"]
|
|
mac = s["caller_mac"]
|
|
match_via = None
|
|
item = None
|
|
account = None
|
|
|
|
# 1. MAC match
|
|
if mac:
|
|
item = mac_to_item.get(mac)
|
|
if item and item.get("assignee_type") == "Account":
|
|
account = accounts.get(item["account_id"])
|
|
match_via = "mac"
|
|
|
|
# 2. Username -> displayName fuzzy match
|
|
if account is None:
|
|
cands = user_idx.get(smush(username), [])
|
|
if len(cands) == 1:
|
|
account = accounts.get(cands[0])
|
|
match_via = "username"
|
|
elif len(cands) > 1:
|
|
# ambiguous; prefer a paying one
|
|
paying = [accounts[a] for a in cands if accounts[a]["paying"]]
|
|
if paying:
|
|
account = paying[0]
|
|
match_via = "username-ambig-paying"
|
|
else:
|
|
account = accounts[cands[0]]
|
|
match_via = "username-ambig"
|
|
|
|
# Classify
|
|
if account is None and item is None:
|
|
cat = "UNMATCHED"
|
|
elif account is None:
|
|
cat = "UNASSIGNED" # MAC in inventory but no Account
|
|
elif account["paying"]:
|
|
cat = "OK_PAYING"
|
|
else:
|
|
cat = "UNPAID"
|
|
|
|
results.append({
|
|
**s,
|
|
"category": cat,
|
|
"match_via": match_via,
|
|
"account": account,
|
|
"inventory_item": item,
|
|
})
|
|
return results
|
|
|
|
|
|
def main():
|
|
ap = argparse.ArgumentParser()
|
|
ap.add_argument("--json", action="store_true")
|
|
ap.add_argument("--routers", help="comma-separated subset")
|
|
ap.add_argument("--show-ok", action="store_true",
|
|
help="also list OK_PAYING sessions (default: hidden)")
|
|
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 inventory + accounts...", file=sys.stderr)
|
|
mac_to_item, accounts = fetch_gaiia()
|
|
paying = sum(1 for a in accounts.values() if a["paying"])
|
|
print(f" {len(accounts)} accounts ({paying} paying), {len(mac_to_item)} MACs in inventory "
|
|
f"({sum(1 for v in mac_to_item.values() if v.get('assignee_type')=='Account')} account-assigned)",
|
|
file=sys.stderr)
|
|
|
|
print(f"Fetching active PPPoE sessions from {len(routers)} routers...", file=sys.stderr)
|
|
with ThreadPoolExecutor(max_workers=len(routers)) as ex:
|
|
per_router = list(ex.map(fetch_ppp_active, routers))
|
|
sessions = [s for sub in per_router for s in sub]
|
|
print(f" {len(sessions)} active PPPoE sessions", file=sys.stderr)
|
|
|
|
results = classify(sessions, mac_to_item, accounts)
|
|
by_cat: dict[str, list] = {}
|
|
for r in results:
|
|
by_cat.setdefault(r["category"], []).append(r)
|
|
|
|
if args.json:
|
|
json.dump({"summary": {k: len(v) for k, v in by_cat.items()},
|
|
"results": results}, sys.stdout, indent=2, default=str)
|
|
return
|
|
|
|
counts = {k: len(by_cat.get(k, [])) for k in ("OK_PAYING","UNPAID","UNASSIGNED","UNMATCHED")}
|
|
print()
|
|
print(f"=== PPPoE classification summary ({len(results)} sessions) ===")
|
|
for k, v in counts.items():
|
|
print(f" {k:12} {v}")
|
|
|
|
for cat in ("UNPAID", "UNASSIGNED", "UNMATCHED"):
|
|
rows = by_cat.get(cat, [])
|
|
if not rows:
|
|
continue
|
|
print()
|
|
print(f"--- {cat} ({len(rows)}) ---")
|
|
for r in sorted(rows, key=lambda x: (x["router"], x["address"])):
|
|
a = r["account"]
|
|
if a:
|
|
acct_str = (f"#{a['readable']} {a['name']!r} "
|
|
f"status={a['status']} "
|
|
f"subs=[{','.join(a['sub_statuses']) or '<none>'}] "
|
|
f"via={r['match_via']}")
|
|
else:
|
|
acct_str = "(no account match)"
|
|
print(f" {r['router']:9} {r['address']:16} user={r['username']!r:25} "
|
|
f"caller={r['caller_mac_raw']:17} {acct_str}")
|
|
|
|
if args.show_ok:
|
|
print()
|
|
print(f"--- OK_PAYING ({len(by_cat.get('OK_PAYING',[]))}) ---")
|
|
for r in sorted(by_cat.get("OK_PAYING", []), key=lambda x: (x["router"], x["address"])):
|
|
a = r["account"]
|
|
print(f" {r['router']:9} {r['address']:16} user={r['username']!r:25} "
|
|
f"acct=#{a['readable']} {a['name']!r} via={r['match_via']}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|