749 lines
29 KiB
Python
Executable file
749 lines
29 KiB
Python
Executable file
#!/usr/bin/env python3
|
||
"""
|
||
uisp — CLI for UISP at $UISP_URL (default https://uisp.vntx.net).
|
||
|
||
Auth: export UISP_KEY=<x-auth-token from UISP user settings>.
|
||
|
||
Subcommands:
|
||
list [--unauthorized] [--upgradable] [--json] [--raw]
|
||
approve <id>... | --all [--site vntx] [--port 443] [--dry-run]
|
||
upgrade <id>... | --all [--yes] [--force]
|
||
prune [--days 1] [--yes] delete devices offline longer than N days
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import datetime as dt
|
||
import json
|
||
import os
|
||
import ssl
|
||
import sys
|
||
import time
|
||
import urllib.error
|
||
import urllib.parse
|
||
import urllib.request
|
||
from collections import Counter
|
||
from typing import Any, Iterable
|
||
|
||
DEFAULT_BASE = os.environ.get("UISP_URL", "https://uisp.vntx.net").rstrip("/")
|
||
TOKEN = os.environ.get("UISP_KEY") or os.environ.get("UISP_TOKEN")
|
||
|
||
# Ordered list — first cred that lets UISP adopt a device wins.
|
||
STANDARD_CREDS = [
|
||
("ubnt", "fngckewl"),
|
||
("ubnt", "vntx1830"),
|
||
("ubnt", "Vntx1830"),
|
||
("ubnt", "vntx1830vntx"),
|
||
("ubnt", "H8xd9tkryg"),
|
||
("ubnt", "h8xd9tkryg"),
|
||
]
|
||
|
||
# discovery.status values that mean "trying again with a new password might help".
|
||
RETRYABLE_DISCOVERY_STATUSES = {
|
||
"authentication-failed",
|
||
"authenticating",
|
||
"starting",
|
||
}
|
||
|
||
|
||
def die(msg: str, code: int = 1) -> None:
|
||
print(f"error: {msg}", file=sys.stderr)
|
||
sys.exit(code)
|
||
|
||
|
||
def api(method: str, path: str, body: Any = None) -> Any:
|
||
if not TOKEN:
|
||
die("UISP_KEY env var is not set")
|
||
url = f"{DEFAULT_BASE}/nms/api/v2.1{path}"
|
||
data = None
|
||
if body is not None:
|
||
data = json.dumps(body).encode()
|
||
req = urllib.request.Request(url, data=data, method=method)
|
||
req.add_header("X-Auth-Token", TOKEN)
|
||
req.add_header("Accept", "application/json")
|
||
if body is not None:
|
||
req.add_header("Content-Type", "application/json")
|
||
ctx = ssl.create_default_context()
|
||
try:
|
||
with urllib.request.urlopen(req, context=ctx, timeout=60) as resp:
|
||
raw = resp.read()
|
||
except urllib.error.HTTPError as e:
|
||
raw = e.read()
|
||
try:
|
||
payload = json.loads(raw)
|
||
msg = payload.get("message") or payload
|
||
except Exception:
|
||
msg = raw.decode("utf-8", "replace")[:200]
|
||
die(f"{method} {url}: HTTP {e.code}: {msg}")
|
||
if not raw:
|
||
return None
|
||
return json.loads(raw)
|
||
|
||
|
||
# ---------- Helpers ----------
|
||
|
||
def disp_name(d: dict) -> str:
|
||
ident = d.get("identification") or {}
|
||
return ident.get("hostname") or ident.get("name") or ident.get("displayName") or ident.get("mac") or "?"
|
||
|
||
|
||
def parse_version(v: str) -> tuple[int, ...]:
|
||
"""Loose semver parse: '6.3.24-cs' -> (6, 3, 24)."""
|
||
if not v:
|
||
return ()
|
||
v = v.lstrip("v")
|
||
head = ""
|
||
for ch in v:
|
||
if ch in "-+ ":
|
||
break
|
||
head += ch
|
||
parts = []
|
||
for chunk in head.split("."):
|
||
n = 0
|
||
for ch in chunk:
|
||
if not ch.isdigit():
|
||
break
|
||
n = n * 10 + int(ch)
|
||
parts.append(n)
|
||
return tuple(parts)
|
||
|
||
|
||
def version_lt(a: str, b: str) -> bool:
|
||
return parse_version(a) < parse_version(b)
|
||
|
||
|
||
def latest_firmware_for(fws: list[dict], d: dict) -> str:
|
||
ident = d.get("identification") or {}
|
||
plat = ident.get("platformId")
|
||
model = ident.get("model")
|
||
if not plat or not model:
|
||
return ""
|
||
best = ""
|
||
for fw in fws:
|
||
ident_fw = fw.get("identification") or {}
|
||
if ident_fw.get("platformId") != plat:
|
||
continue
|
||
models = ident_fw.get("models") or []
|
||
if models and model not in models:
|
||
continue
|
||
if not ident_fw.get("stable", True):
|
||
continue
|
||
v = ident_fw.get("version") or ""
|
||
if not best or version_lt(best, v):
|
||
best = v
|
||
return best
|
||
|
||
|
||
def upgrade_in_flight(d: dict, target_v: str = "") -> bool:
|
||
"""True if UISP already has a pending/running upgrade for this device.
|
||
Re-POSTing /tasks for a device with one queued returns a misleading
|
||
422 "Firmware not found" error; skip these instead.
|
||
"""
|
||
up = d.get("upgrade") or {}
|
||
status = (up.get("status") or "").lower()
|
||
if status not in {"queued", "running", "in-progress", "started", "preparing", "downloading", "installing"}:
|
||
return False
|
||
if not target_v:
|
||
return True
|
||
# Only skip if the queued upgrade is for the same target we'd request.
|
||
return up.get("firmwareVersion") == target_v
|
||
|
||
|
||
def needs_upgrade(d: dict, fws: list[dict]) -> bool:
|
||
ident = d.get("identification") or {}
|
||
if ident.get("type") == "blackBox":
|
||
return False
|
||
model = ident.get("model")
|
||
if not model or model == "UNKNOWN":
|
||
return False
|
||
cur = ident.get("firmwareVersion") or (d.get("firmware") or {}).get("current") or ""
|
||
if not cur:
|
||
return False
|
||
latest = latest_firmware_for(fws, d)
|
||
if not latest:
|
||
return False
|
||
if not version_lt(cur, latest):
|
||
return False
|
||
if upgrade_in_flight(d, latest):
|
||
return False
|
||
return True
|
||
|
||
|
||
def is_unauthorized(d: dict) -> bool:
|
||
ident = d.get("identification") or {}
|
||
overview = d.get("overview") or {}
|
||
return not ident.get("authorized", True) or overview.get("status") == "discovered"
|
||
|
||
|
||
# ---------- Commands ----------
|
||
|
||
def cmd_list(args: argparse.Namespace) -> None:
|
||
if args.unauthorized:
|
||
devs = api("GET", "/devices/discovered") or []
|
||
# SNMP-discovered stubs come back as model=UNKNOWN and aren't actually
|
||
# adoptable. Hide by default to keep the pending-adoption view useful.
|
||
if not args.include_unknown:
|
||
devs = [d for d in devs if (d.get("identification") or {}).get("model") not in ("", None, "UNKNOWN")]
|
||
else:
|
||
devs = api("GET", "/devices?withInterfaces=false")
|
||
|
||
if args.raw:
|
||
json.dump(devs, sys.stdout, indent=2)
|
||
print()
|
||
return
|
||
|
||
fws = api("GET", "/firmwares") or []
|
||
|
||
rows = []
|
||
for d in devs:
|
||
if args.upgradable and not needs_upgrade(d, fws):
|
||
continue
|
||
rows.append(d)
|
||
rows.sort(key=disp_name)
|
||
|
||
if args.json:
|
||
json.dump(rows, sys.stdout, indent=2)
|
||
print()
|
||
return
|
||
|
||
headers = ["ID", "NAME", "MODEL", "SITE", "FW", "STATUS", "AUTH", "UPGRADE"]
|
||
table = []
|
||
for d in rows:
|
||
ident = d.get("identification") or {}
|
||
ov = d.get("overview") or {}
|
||
cur = ident.get("firmwareVersion") or (d.get("firmware") or {}).get("current") or ""
|
||
latest = latest_firmware_for(fws, d)
|
||
fw = cur or "-"
|
||
if latest and latest != cur and version_lt(cur, latest):
|
||
fw = f"{cur} → {latest}"
|
||
site_name = ""
|
||
site = ident.get("site")
|
||
if isinstance(site, dict):
|
||
site_name = site.get("name") or ""
|
||
table.append([
|
||
ident.get("id", ""),
|
||
disp_name(d),
|
||
ident.get("modelName") or ident.get("model") or "",
|
||
site_name,
|
||
fw,
|
||
ov.get("status") or "",
|
||
"yes" if ident.get("authorized") else "NO",
|
||
"yes" if needs_upgrade(d, fws) else "",
|
||
])
|
||
widths = [max(len(h), *(len(r[i]) for r in table)) if table else len(h) for i, h in enumerate(headers)]
|
||
print(" ".join(h.ljust(w) for h, w in zip(headers, widths)))
|
||
for r in table:
|
||
print(" ".join(c.ljust(w) for c, w in zip(r, widths)))
|
||
|
||
|
||
def lookup_site_id(name: str) -> str:
|
||
sites = api("GET", "/sites")
|
||
want = name.strip().lower()
|
||
for s in sites:
|
||
ident = s.get("identification") or {}
|
||
n = ident.get("name") or s.get("name") or ""
|
||
if n.lower() == want:
|
||
return ident.get("id") or s.get("id")
|
||
die(f"site {name!r} not found")
|
||
|
||
|
||
def wait_settle(ids: list[str], timeout_sec: int = 60) -> dict[str, dict]:
|
||
"""Poll /devices/discovered until none of the given IDs report
|
||
isProcessing=true (or timeout). Returns the latest discovery dict per ID."""
|
||
want = set(ids)
|
||
deadline = time.time() + timeout_sec
|
||
last: dict[str, dict] = {}
|
||
while True:
|
||
discovered = api("GET", "/devices/discovered") or []
|
||
last = {}
|
||
still = 0
|
||
for d in discovered:
|
||
ident = d.get("identification") or {}
|
||
did = ident.get("id")
|
||
if did not in want:
|
||
continue
|
||
disc = d.get("discovery") or {}
|
||
last[did] = {**disc, "_device": d}
|
||
if disc.get("isProcessing"):
|
||
still += 1
|
||
if still == 0 or time.time() > deadline:
|
||
print(f" settled (still processing: {still})")
|
||
return last
|
||
print(f" ...waiting on {still} device(s)")
|
||
time.sleep(5)
|
||
|
||
|
||
def cmd_approve(args: argparse.Namespace) -> None:
|
||
discovered = api("GET", "/devices/discovered") or []
|
||
by_id = {d["identification"]["id"]: d for d in discovered}
|
||
|
||
if args.all:
|
||
queue = list(discovered)
|
||
else:
|
||
if not args.ids:
|
||
die("usage: uisp approve <id>... | --all")
|
||
queue = []
|
||
for did in args.ids:
|
||
if did not in by_id:
|
||
print(f"skip {did}: not in discovered list", file=sys.stderr)
|
||
continue
|
||
queue.append(by_id[did])
|
||
|
||
# Devices with model "UNKNOWN" are SNMP-discovered stubs UISP can't manage —
|
||
# connect/ubnt always fails on them, so don't burn cycles trying.
|
||
if not args.include_unknown:
|
||
before = len(queue)
|
||
queue = [d for d in queue if (d.get("identification") or {}).get("model") not in ("", None, "UNKNOWN")]
|
||
skipped = before - len(queue)
|
||
if skipped:
|
||
print(f"skipping {skipped} device(s) with model=UNKNOWN (use --include-unknown to override)")
|
||
|
||
if not queue:
|
||
print("no devices pending adoption.")
|
||
return
|
||
|
||
queue_ids = [d["identification"]["id"] for d in queue]
|
||
print(f"queue ({len(queue)} device(s)):")
|
||
for d in queue:
|
||
ident = d["identification"]
|
||
print(f" {ident['id']} {disp_name(d):<32} {ident.get('modelName') or ident.get('model','')}")
|
||
|
||
site_id = ""
|
||
if args.site:
|
||
site_id = lookup_site_id(args.site)
|
||
print(f"site {args.site!r} resolved to id {site_id}")
|
||
|
||
if args.dry_run:
|
||
print(f"\ndry-run: would try {len(STANDARD_CREDS)} credential combo(s)" +
|
||
(f" then assign to site {args.site!r}" if site_id else ""))
|
||
return
|
||
|
||
# Cred sweep — each pass only triggers connect on still-retryable IDs.
|
||
retry_ids = list(queue_ids)
|
||
for i, (user, pw) in enumerate(STANDARD_CREDS, 1):
|
||
if not retry_ids:
|
||
break
|
||
print(f"\n[{i}/{len(STANDARD_CREDS)}] trying user={user} pw={pw} on {len(retry_ids)} device(s)...")
|
||
try:
|
||
api("POST", "/discovery/connect/ubnt", {
|
||
"deviceIds": retry_ids,
|
||
"username": user,
|
||
"password": pw,
|
||
"httpsPort": args.port,
|
||
})
|
||
except SystemExit:
|
||
print(" connect call failed; moving on", file=sys.stderr)
|
||
continue
|
||
statuses = wait_settle(retry_ids, timeout_sec=args.poll_timeout)
|
||
# Keep only IDs that are still retryable for the next cred.
|
||
next_retry = []
|
||
for did in retry_ids:
|
||
st = statuses.get(did, {}).get("status", "")
|
||
if st in RETRYABLE_DISCOVERY_STATUSES:
|
||
next_retry.append(did)
|
||
retry_ids = next_retry
|
||
|
||
# Final reconciliation.
|
||
fleet = api("GET", "/devices?withInterfaces=false") or []
|
||
fleet_by_id = {d["identification"]["id"]: d for d in fleet}
|
||
final_disc = {d["identification"]["id"]: d for d in (api("GET", "/devices/discovered") or [])}
|
||
|
||
adopted, failed = [], []
|
||
for did in queue_ids:
|
||
d = fleet_by_id.get(did)
|
||
if d and d["identification"].get("type") != "blackBox":
|
||
adopted.append(did)
|
||
else:
|
||
failed.append(did)
|
||
|
||
if adopted:
|
||
print(f"\nadopted {len(adopted)} device(s):")
|
||
for did in adopted:
|
||
print(f" {did} {disp_name(fleet_by_id[did])}")
|
||
|
||
if failed:
|
||
# Group by status for a tidy summary.
|
||
groups: dict[str, list[tuple[str, str, str]]] = {}
|
||
for did in failed:
|
||
d = final_disc.get(did) or by_id.get(did) or {}
|
||
disc = d.get("discovery") or {}
|
||
status = disc.get("status") or "unknown"
|
||
err = disc.get("error") or ""
|
||
groups.setdefault(status, []).append((did, disp_name(d), err))
|
||
print(f"\n{len(failed)} device(s) still pending:")
|
||
for status in sorted(groups):
|
||
entries = groups[status]
|
||
print(f" [{status}] ({len(entries)})")
|
||
for did, name, err in entries:
|
||
line = f" {did} {name}"
|
||
if err:
|
||
line += f" — {err}"
|
||
print(line)
|
||
|
||
if site_id and adopted:
|
||
print(f"\nassigning {len(adopted)} adopted device(s) to site {args.site!r}...")
|
||
api("POST", "/devices/authorize", {"siteId": site_id, "deviceIds": adopted})
|
||
print(" done.")
|
||
|
||
|
||
def parse_uisp_time(s: str | None) -> dt.datetime | None:
|
||
"""UISP returns ISO 8601 like '2026-04-15T00:08:42.587Z'. Returns aware UTC dt."""
|
||
if not s:
|
||
return None
|
||
s = s.replace("Z", "+00:00")
|
||
try:
|
||
return dt.datetime.fromisoformat(s)
|
||
except ValueError:
|
||
return None
|
||
|
||
|
||
def cmd_prune(args: argparse.Namespace) -> None:
|
||
devs = api("GET", "/devices?withInterfaces=false") or []
|
||
now = dt.datetime.now(dt.timezone.utc)
|
||
cutoff = now - dt.timedelta(days=args.days)
|
||
|
||
stale = []
|
||
for d in devs:
|
||
ident = d.get("identification") or {}
|
||
ov = d.get("overview") or {}
|
||
last_seen = parse_uisp_time(ov.get("lastSeen"))
|
||
if last_seen is not None and last_seen >= cutoff:
|
||
continue
|
||
if last_seen is None and args.skip_never_seen:
|
||
continue
|
||
# Skip backbone routers / switches by default — losing one of these from
|
||
# UISP just because it's not reporting in would be an own-goal.
|
||
if not args.include_infra and ident.get("type") in {"erouter", "eswitch", "olt"}:
|
||
continue
|
||
if not args.include_infra and ident.get("role") in {"router", "switch", "gateway"}:
|
||
continue
|
||
stale.append((d, last_seen))
|
||
|
||
if not stale:
|
||
print(f"no devices offline longer than {args.days} day(s).")
|
||
return
|
||
|
||
stale.sort(key=lambda t: t[1] or dt.datetime.min.replace(tzinfo=dt.timezone.utc))
|
||
|
||
# Summary buckets so the user can sanity-check before approving.
|
||
summary: Counter = Counter()
|
||
type_summary: Counter = Counter()
|
||
for d, last_seen in stale:
|
||
if last_seen is None:
|
||
bucket = "never seen"
|
||
else:
|
||
age_days = (now - last_seen).days
|
||
if age_days > 30:
|
||
bucket = ">30d offline"
|
||
elif age_days > 7:
|
||
bucket = "7-30d offline"
|
||
else:
|
||
bucket = "1-7d offline"
|
||
summary[bucket] += 1
|
||
type_summary[(bucket, (d.get("identification") or {}).get("type") or "?")] += 1
|
||
|
||
print(f"{len(stale)} device(s) eligible for prune (offline > {args.days}d, including never-seen={'no' if args.skip_never_seen else 'yes'}):")
|
||
for bucket, n in summary.most_common():
|
||
print(f" {n:>4} {bucket}")
|
||
print()
|
||
print("by bucket × type:")
|
||
for (bucket, t), n in sorted(type_summary.items()):
|
||
print(f" {n:>4} {bucket:<14} type={t}")
|
||
print()
|
||
|
||
if args.verbose:
|
||
print("detail:")
|
||
for d, last_seen in stale:
|
||
ident = d["identification"]
|
||
site_name = ""
|
||
if isinstance(ident.get("site"), dict):
|
||
site_name = ident["site"].get("name") or ""
|
||
age = "never seen" if last_seen is None else f"{(now - last_seen).days}d"
|
||
print(f" {ident['id']} {disp_name(d):<30} "
|
||
f"{ident.get('modelName') or ident.get('model',''):<22} "
|
||
f"site={site_name:<14} offline={age}")
|
||
print()
|
||
|
||
if not args.yes:
|
||
print(f"dry-run: pass --yes to delete these {len(stale)} device(s) (use --verbose to see each one).")
|
||
return
|
||
|
||
ids = [d["identification"]["id"] for d, _ in stale]
|
||
api("POST", "/devices/bulkdelete", {"ids": ids})
|
||
print(f"deleted {len(ids)} device(s).")
|
||
|
||
|
||
# Whitelist of fields UISP's PUT /nms/settings will accept. Extracted from the
|
||
# UI bundle (`hZr` array). Anything else triggers a 400 like "X is not allowed".
|
||
SETTINGS_PUT_ALLOWED = {
|
||
"allowAutoUpdateUbntFirmwares", "allowBetaFirmwares", "allowLoggingToLogentries",
|
||
"allowLoggingToSentry", "allowNewFirmware", "autoUpdatePlatforms", "country",
|
||
"dateFormat", "defaultGracePeriod", "defaultQosPropagation", "devicePingAddress",
|
||
"devicePingAddressMode", "devicePingIntervalNormal", "devicePingIntervalOutage",
|
||
"deviceTransmissionFrequencies", "deviceTransmissionProfile", "deviceUpdateNotification",
|
||
"discoveryAllowLocalScan", "discoveryAllowRemoteScan", "discoveryAllowUnsecuredChannels",
|
||
"discoveryAutoConfiguration", "discoveryBlacklist", "discoveryHideBlackBox",
|
||
"discoveryNotification", "discoverySnmpCommunity", "googleMapsApiKey", "homePage",
|
||
"hostname", "isOnuDisabledOnSubscriberSuspend", "maintenanceWindowFriday",
|
||
"maintenanceWindowFromTime", "maintenanceWindowMonday", "maintenanceWindowSaturday",
|
||
"maintenanceWindowSunday", "maintenanceWindowThursday", "maintenanceWindowToTime",
|
||
"maintenanceWindowTuesday", "maintenanceWindowWednesday", "mapsProvider",
|
||
"migrationForceModeEnabled", "migrationWithBackupEnabled", "migrationUispKey",
|
||
"migrationHostname", "migrationModeEnabled", "migrationPort", "outageMailablePeriod",
|
||
"parallelUpgradeLimit", "restartGracePeriod", "tableDensity", "timeFormat",
|
||
"timezone", "trafficShapingAdjustment", "upgradeGracePeriod", "useLetsEncrypt",
|
||
}
|
||
|
||
|
||
def put_nms_settings(updates: dict) -> None:
|
||
"""GET current settings, merge updates, filter to PUT-allowed fields, PUT."""
|
||
current = api("GET", "/nms/settings") or {}
|
||
merged = {**current, **updates}
|
||
body = {k: v for k, v in merged.items() if k in SETTINGS_PUT_ALLOWED}
|
||
api("PUT", "/nms/settings", body)
|
||
|
||
|
||
def cmd_ignore(args: argparse.Namespace) -> None:
|
||
"""Add IPs of model=UNKNOWN devices to UISP's discoveryBlacklist so they
|
||
stop reappearing, then delete the stub records."""
|
||
# Pull from both /devices/discovered (pending) and /devices (adopted but
|
||
# blackBox) since the SNMP-stub population shows up in both lists.
|
||
discovered = api("GET", "/devices/discovered") or []
|
||
fleet = api("GET", "/devices?withInterfaces=false") or []
|
||
pool = []
|
||
for d in discovered + fleet:
|
||
ident = d.get("identification") or {}
|
||
if ident.get("model") in ("", None, "UNKNOWN") or ident.get("type") == "blackBox":
|
||
pool.append(d)
|
||
# Dedupe by id (a device can appear in both lists).
|
||
seen_ids: set[str] = set()
|
||
targets = []
|
||
for d in pool:
|
||
did = (d.get("identification") or {}).get("id")
|
||
if did and did not in seen_ids:
|
||
seen_ids.add(did)
|
||
targets.append(d)
|
||
|
||
ip_set: dict[str, dict] = {} # ip -> first device that has it
|
||
no_ip = []
|
||
for d in targets:
|
||
ip = d.get("ipAddress") or ""
|
||
ip = ip.split("/")[0].strip()
|
||
if ip:
|
||
ip_set.setdefault(ip, d)
|
||
else:
|
||
no_ip.append(d)
|
||
|
||
if not ip_set and not no_ip:
|
||
print("no UNKNOWN/blackBox devices found.")
|
||
return
|
||
|
||
settings = api("GET", "/nms/settings")
|
||
current_bl = settings.get("discoveryBlacklist") or []
|
||
if isinstance(current_bl, str):
|
||
current_bl = [x.strip() for x in current_bl.split(",") if x.strip()]
|
||
current_set = set(current_bl)
|
||
new_ips = sorted(ip for ip in ip_set if ip not in current_set)
|
||
|
||
print(f"found {len(targets)} UNKNOWN/blackBox device(s) "
|
||
f"({len(ip_set)} unique IP(s), {len(no_ip)} without IP)")
|
||
print(f"already blacklisted: {len(current_set)}")
|
||
print(f"new IPs to blacklist: {len(new_ips)}")
|
||
if new_ips and args.verbose:
|
||
for ip in new_ips:
|
||
d = ip_set[ip]
|
||
ident = d["identification"]
|
||
print(f" {ip:<16} {ident.get('hostname') or ident.get('name') or '?'}")
|
||
|
||
if args.dry_run:
|
||
print(f"\ndry-run: would add {len(new_ips)} IP(s) to discoveryBlacklist "
|
||
f"and delete {len(targets)} stub record(s).")
|
||
return
|
||
|
||
if new_ips:
|
||
merged = sorted(current_set | set(new_ips))
|
||
put_nms_settings({"discoveryBlacklist": merged})
|
||
print(f"blacklist updated: {len(current_set)} → {len(merged)} entries")
|
||
|
||
if targets:
|
||
ids = [d["identification"]["id"] for d in targets]
|
||
# bulkdelete handles batches comfortably; UISP returns 200 with no body.
|
||
api("POST", "/devices/bulkdelete", {"ids": ids})
|
||
print(f"deleted {len(ids)} stub record(s).")
|
||
|
||
|
||
def cmd_upgrade(args: argparse.Namespace) -> None:
|
||
devs = api("GET", "/devices?withInterfaces=false") or []
|
||
fws = api("GET", "/firmwares") or []
|
||
by_id = {d["identification"]["id"]: d for d in devs}
|
||
|
||
targets = []
|
||
if args.all:
|
||
for d in devs:
|
||
if needs_upgrade(d, fws):
|
||
targets.append(d)
|
||
else:
|
||
if not args.ids:
|
||
die("usage: uisp upgrade <id>... | --all")
|
||
for did in args.ids:
|
||
d = by_id.get(did)
|
||
if not d:
|
||
print(f"skip {did}: not found", file=sys.stderr)
|
||
continue
|
||
if not needs_upgrade(d, fws) and not args.force:
|
||
ident = d["identification"]
|
||
cur = ident.get("firmwareVersion") or "?"
|
||
latest = latest_firmware_for(fws, d) or "?"
|
||
print(f"skip {did} ({disp_name(d)}): already on {cur} (latest {latest}) — pass --force to override",
|
||
file=sys.stderr)
|
||
continue
|
||
targets.append(d)
|
||
|
||
if not targets:
|
||
print("no devices need upgrade.")
|
||
return
|
||
|
||
print(f"upgrade plan ({len(targets)} device(s)):")
|
||
for d in targets:
|
||
ident = d["identification"]
|
||
cur = ident.get("firmwareVersion") or "?"
|
||
latest = latest_firmware_for(fws, d) or "?"
|
||
print(f" {ident['id']} {disp_name(d):<30} {cur} → {latest}")
|
||
|
||
if not args.yes:
|
||
print(f"\ndry-run: pass --yes to actually trigger upgrades.")
|
||
return
|
||
|
||
batch_size = args.batch_size if args.batch_size > 0 else len(targets)
|
||
n_batches = (len(targets) + batch_size - 1) // batch_size
|
||
failed_total = 0
|
||
|
||
for batch_idx in range(n_batches):
|
||
batch = targets[batch_idx * batch_size:(batch_idx + 1) * batch_size]
|
||
print(f"\n=== batch {batch_idx + 1}/{n_batches}: {len(batch)} device(s) ===")
|
||
|
||
# POST /tasks one device at a time. Bulk submission (all device
|
||
# upgrades in a single /tasks payload) is atomic: any one device
|
||
# whose target firmwareVersion doesn't exactly match a catalog
|
||
# entry causes the whole call to 422 with "Firmware not found".
|
||
# Per-device /tasks isolates failures.
|
||
triggered: list[tuple[str, str, str]] = [] # (id, name, target_version)
|
||
for d in batch:
|
||
did = d["identification"]["id"]
|
||
target_v = latest_firmware_for(fws, d)
|
||
if not target_v:
|
||
print(f" skip {did} {disp_name(d):<30}: no catalog firmware match", file=sys.stderr)
|
||
failed_total += 1
|
||
continue
|
||
if upgrade_in_flight(d, target_v):
|
||
up = d.get("upgrade") or {}
|
||
print(f" skip {did} {disp_name(d):<30}: already {up.get('status')} → {up.get('firmwareVersion')}")
|
||
continue
|
||
try:
|
||
api("POST", "/tasks", {
|
||
"upgrades": [{"deviceId": did, "firmwareVersion": target_v}],
|
||
"upgradeInMaintenanceWindow": False,
|
||
})
|
||
print(f" trigger OK {did} {disp_name(d):<30} → {target_v}")
|
||
triggered.append((did, disp_name(d), target_v))
|
||
except SystemExit:
|
||
print(f" trigger FAIL {did} {disp_name(d):<30} → {target_v}", file=sys.stderr)
|
||
failed_total += 1
|
||
|
||
if not args.wait:
|
||
continue
|
||
|
||
# Wait for every device in this batch to land on its target version.
|
||
deadline = time.time() + args.wait_timeout
|
||
pending = {did: (name, tv) for did, name, tv in triggered}
|
||
while pending and time.time() < deadline:
|
||
time.sleep(20)
|
||
current = api("GET", "/devices?withInterfaces=false") or []
|
||
cur_by_id = {x["identification"]["id"]: x for x in current}
|
||
for did in list(pending.keys()):
|
||
name, tv = pending[did]
|
||
d = cur_by_id.get(did)
|
||
if not d:
|
||
continue
|
||
cur = d["identification"].get("firmwareVersion") or ""
|
||
if cur and tv != "?" and not version_lt(cur, tv):
|
||
print(f" done {did} {name:<30} on {cur}")
|
||
del pending[did]
|
||
if pending:
|
||
print(f" ...waiting on {len(pending)} device(s) "
|
||
f"({int(deadline - time.time())}s left)")
|
||
|
||
if pending:
|
||
print(f" batch timeout: {len(pending)} device(s) didn't finish in {args.wait_timeout}s; "
|
||
f"continuing anyway")
|
||
for did, (name, tv) in pending.items():
|
||
print(f" still pending {did} {name} → {tv}")
|
||
failed_total += 1
|
||
|
||
if failed_total:
|
||
print(f"\n{failed_total} device(s) had problems (failed trigger or didn't complete in time).")
|
||
else:
|
||
print(f"\nall {len(targets)} device(s) processed.")
|
||
|
||
|
||
# ---------- argparse wiring ----------
|
||
|
||
def main() -> None:
|
||
p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
|
||
sub = p.add_subparsers(dest="cmd", required=True)
|
||
|
||
pl = sub.add_parser("list", help="list devices")
|
||
pl.add_argument("--unauthorized", action="store_true", help="show only pending-adoption devices")
|
||
pl.add_argument("--upgradable", action="store_true", help="show only devices with a firmware update")
|
||
pl.add_argument("--json", action="store_true", help="emit parsed JSON")
|
||
pl.add_argument("--raw", action="store_true", help="dump raw API JSON (debug)")
|
||
pl.add_argument("--include-unknown", action="store_true",
|
||
help="include model=UNKNOWN devices in --unauthorized output (hidden by default)")
|
||
pl.set_defaults(func=cmd_list)
|
||
|
||
pa = sub.add_parser("approve", help="adopt pending-adoption devices")
|
||
pa.add_argument("ids", nargs="*", help="device IDs to adopt")
|
||
pa.add_argument("--all", action="store_true", help="adopt every pending-adoption device")
|
||
pa.add_argument("--site", default="vntx", help="site name to assign adopted devices to (empty = skip)")
|
||
pa.add_argument("--port", type=int, default=443, help="HTTPS port UISP uses to reach the device")
|
||
pa.add_argument("--poll-timeout", type=int, default=60, help="seconds to wait for each cred attempt to settle")
|
||
pa.add_argument("--dry-run", action="store_true")
|
||
pa.add_argument("--include-unknown", action="store_true",
|
||
help="also try to adopt devices with model=UNKNOWN (SNMP stubs; almost never adoptable)")
|
||
pa.set_defaults(func=cmd_approve)
|
||
|
||
pi = sub.add_parser("ignore",
|
||
help="blacklist IPs of UNKNOWN/blackBox devices and delete the stubs so they stop coming back")
|
||
pi.add_argument("--dry-run", action="store_true")
|
||
pi.add_argument("--verbose", "-v", action="store_true", help="list every IP being added")
|
||
pi.set_defaults(func=cmd_ignore)
|
||
|
||
pp = sub.add_parser("prune", help="delete devices offline longer than N days")
|
||
pp.add_argument("--days", type=float, default=1.0, help="offline threshold in days (default: 1)")
|
||
pp.add_argument("--yes", action="store_true", help="actually delete (default: dry-run)")
|
||
pp.add_argument("--include-infra", action="store_true",
|
||
help="also prune routers/switches/OLTs (default: skipped)")
|
||
pp.add_argument("--skip-never-seen", action="store_true",
|
||
help="exclude devices with no lastSeen timestamp (SNMP stubs that never adopted)")
|
||
pp.add_argument("--verbose", "-v", action="store_true", help="list every device, not just summary")
|
||
pp.set_defaults(func=cmd_prune)
|
||
|
||
pu = sub.add_parser("upgrade", help="upgrade firmware")
|
||
pu.add_argument("ids", nargs="*", help="device IDs to upgrade")
|
||
pu.add_argument("--all", action="store_true", help="upgrade every device with a newer firmware available")
|
||
pu.add_argument("--yes", action="store_true", help="actually trigger upgrades (default: dry-run)")
|
||
pu.add_argument("--force", action="store_true", help="upgrade even if already on latest")
|
||
pu.add_argument("--batch-size", type=int, default=0,
|
||
help="devices per batch; 0 means all in one batch (default). Only useful with --wait.")
|
||
pu.add_argument("--wait", action="store_true",
|
||
help="wait for each batch to land on the new firmware before starting the next")
|
||
pu.add_argument("--wait-timeout", type=int, default=900,
|
||
help="seconds to wait per batch when --wait is set (default: 900)")
|
||
pu.set_defaults(func=cmd_upgrade)
|
||
|
||
args = p.parse_args()
|
||
args.func(args)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|