52 lines
1.9 KiB
Python
52 lines
1.9 KiB
Python
|
"""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}"}
|