fc31013ceb
This caused a real outage. unifi-reserve-all.py recomputed its plan at --apply time by re-reading stat/sta, so a client that renewed between the dry run and the apply was pinned to whatever transient address it happened to hold at that instant. worker1-k8s0 was reviewed at 192.168.8.13 and written as 192.168.8.242. On its next reboot it could not get an address at all, taking a k8s node down. A plan that gets reviewed and a plan that gets applied must be the same object. The dry run now WRITES the plan to a file and --apply READS it and applies exactly that, reporting any client whose current address has since drifted rather than silently preferring the new value. 1 of 51 diverged; the rest were verified against the reviewed list and were correct. worker1 has been restored to .13 and confirmed: DHCPOFFER for its own MAC returns 192.168.8.13, and the node is up with a full lease and working internet. The second half of the outage was drift between controller and device: the USG was still running config from ~16h before these changes, so the controller looked perfectly correct while the gateway handed out something else. Writing the controller is only half the job, so the script now says so explicitly and gives the force-provision and DHCP-probe commands to verify with. `nmap --script broadcast-dhcp-discover --script-args broadcast-dhcp-discover.mac=...` is the way to prove a specific reservation is live without disturbing the client -- it elicits an OFFER without ever sending a REQUEST. _unifi.py gains post() for device commands. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01DMVzWZgiKW2wquf5z8S1yH
85 lines
3.4 KiB
Python
Executable File
85 lines
3.4 KiB
Python
Executable File
"""Shared UniFi API client for the migration tooling.
|
|
|
|
The controller is a CLASSIC self-hosted UniFi Network app (server_version
|
|
10.4.x), not UniFi OS: login is /api/login and data lives under
|
|
/api/s/<site>/... . UniFi OS would use /api/auth/login + /proxy/network/api.
|
|
Credentials come from the mcpctl server definition so they are not duplicated
|
|
here.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json, re, ssl, subprocess, urllib.request, http.cookiejar
|
|
|
|
|
|
def client():
|
|
raw = subprocess.run(["mcpctl", "describe", "server", "unifi-network"],
|
|
capture_output=True, text=True).stdout
|
|
m = re.search(r"UNIFI_TARGETS\s+(\[.*)", raw)
|
|
if not m:
|
|
raise SystemExit("could not read UNIFI_TARGETS from mcpctl")
|
|
blob = m.group(1).strip()
|
|
try:
|
|
targets = json.loads(blob)
|
|
except json.JSONDecodeError:
|
|
targets = json.loads(blob + "}" * (blob.count("{") - blob.count("}")))
|
|
t = targets[0]
|
|
base = t["base_url"].rstrip("/")
|
|
auth = t.get("auth", {})
|
|
site = t.get("default_site", "default")
|
|
|
|
ctx = ssl.create_default_context()
|
|
ctx.check_hostname = False
|
|
ctx.verify_mode = ssl.CERT_NONE
|
|
opener = urllib.request.build_opener(
|
|
urllib.request.HTTPCookieProcessor(http.cookiejar.CookieJar()),
|
|
urllib.request.HTTPSHandler(context=ctx))
|
|
req = urllib.request.Request(
|
|
f"{base}/api/login",
|
|
data=json.dumps({"username": auth.get("username"),
|
|
"password": auth.get("password")}).encode(),
|
|
headers={"Content-Type": "application/json"})
|
|
opener.open(req, timeout=20).read()
|
|
return opener, base, site
|
|
|
|
|
|
def get(opener, base, site, path):
|
|
"""GET /api/s/<site>/<path>, returning the `data` list (never raising)."""
|
|
try:
|
|
body = opener.open(f"{base}/api/s/{site}/{path}", timeout=30).read()
|
|
return json.loads(body).get("data", [])
|
|
except Exception as exc:
|
|
return {"__error__": f"{type(exc).__name__}: {exc}"}
|
|
|
|
|
|
def post(opener, base, site, path, payload):
|
|
"""POST to /api/s/<site>/<path> -- used for device commands (cmd/devmgr)."""
|
|
req = urllib.request.Request(
|
|
f"{base}/api/s/{site}/{path}",
|
|
data=json.dumps(payload).encode(),
|
|
headers={"Content-Type": "application/json"}, method="POST")
|
|
try:
|
|
return json.loads(opener.open(req, timeout=30).read()).get("data", [])
|
|
except urllib.error.HTTPError as exc:
|
|
return {"__error__": f"HTTP {exc.code}: {exc.read()[:300].decode(errors='replace')}"}
|
|
except Exception as exc:
|
|
return {"__error__": f"{type(exc).__name__}: {exc}"}
|
|
|
|
|
|
def put(opener, base, site, path, payload):
|
|
"""PUT to /api/s/<site>/<path>. Returns the `data` list or an __error__ dict.
|
|
|
|
Classic controllers accept the session cookie alone -- no CSRF token, which
|
|
UniFi OS would require. Errors are returned rather than raised so a caller
|
|
changing production config can report and stop rather than traceback.
|
|
"""
|
|
req = urllib.request.Request(
|
|
f"{base}/api/s/{site}/{path}",
|
|
data=json.dumps(payload).encode(),
|
|
headers={"Content-Type": "application/json"}, method="PUT")
|
|
try:
|
|
return json.loads(opener.open(req, timeout=30).read()).get("data", [])
|
|
except urllib.error.HTTPError as exc:
|
|
return {"__error__": f"HTTP {exc.code}: {exc.read()[:300].decode(errors='replace')}"}
|
|
except Exception as exc:
|
|
return {"__error__": f"{type(exc).__name__}: {exc}"}
|