lab/migration/_unifi.py

"""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}"}