Managing Site Switches Remotely with UniFi's Cloud Connector Proxy

The site I manage happens to be in a different province from my workstation. Whenever I needed to change something, I had to drive there or call someone on site and talk them through the clicks, both of which are slow and prone to mistakes. Opening inbound ports was out of the question, because exposing device management to the internet is like handing everyone in the world a key to try. In the end, I took the route the vendor already provided: the cloud API, which turned out to do far more than the documentation described.

Every IP address, device name, VLAN name, and id in this article is an example value. The API paths, field names, and observed numbers are real, from the system I manage using a key issued by its owner.

Three ways to talk to this system, and why I chose the third

This is the structure of my client. The cloud client holds only the session and key, with a single proxy() method that lets me build the path myself.

class UniFiCloud:
    BASE = "https://api.ui.com/v1"

    def __init__(self, api_key=None):
        self.api_key = api_key or os.environ.get("UI_API_KEY", "")
        self.s = requests.Session()
        self.s.headers.update({"X-API-KEY": self.api_key,
                               "Accept": "application/json"})

    def get(self, path, **params):
        r = self.s.get(self.BASE + path, params=params or None, timeout=30)
        r.raise_for_status()
        return r.json()

    # ยิงคำสั่ง local API ผ่าน cloud proxy (ไม่ต้องเปิดพอร์ต / ไม่ต้อง VPN)
    def proxy(self, console_id, path, **params):
        console_id = console_id or os.environ.get("UNIFI_CONSOLE_ID", "")
        url = (f"{self.BASE}/connector/consoles/{console_id}"
               f"/proxy/network/integration/v1{path}")
        r = self.s.get(url, params=params or None, timeout=30)
        r.raise_for_status()
        return r.json()

All keys live in a .env file beside the script, not in the code, and that file stays out of git.

cat > .env <<'EOF'
UI_API_KEY=<API-KEY>
UNIFI_CONSOLE_ID=<CONSOLE-ID>
UNIFI_SITE_ID=<SITE-ID>
UNIFI_SITE=default
EOF
chmod 600 .env

The proxy is not limited to the integration API

What changed the whole job was trying a different path after /proxy/, replacing network/integration/v1 with something else, and finding that the request actually went through.

I could reach three sets of endpoints. The first was the full legacy API at /proxy/network/api/s/<site>/..., including stat/device, which returns port_table with per-port PoE values, along with rest/networkconf, rest/portconf, and list/user. The second was the v2 API for zone-based firewalls at /proxy/network/v2/api/site/<site>/firewall-policies and /firewall/zone. The third was the camera side at /proxy/protect/api/bootstrap, which returns a complete dataset with far more detail than the integration API.

My data collector therefore became just a short loop over a list of paths, writing each response to a separate named JSON file.

CLASSIC = {
    "devices": "/stat/device",
    "clients": "/stat/sta",
    "health": "/stat/health",
    "networkconf": "/rest/networkconf",
    "portconf": "/rest/portconf",
    "firewallrule": "/rest/firewallrule",
    "user": "/list/user",
    "sysinfo": "/stat/sysinfo",
}

def grab(name, url):
    r = c.s.get(url, timeout=45)
    if r.status_code != 200:
        print(f"  [WARN] {name}: HTTP {r.status_code}")
        return
    data = r.json()
    payload = data.get("data", data) if isinstance(data, dict) else data
    with open(os.path.join(OUT, f"{name}.json"), "w", encoding="utf-8") as fh:
        json.dump(payload, fh, ensure_ascii=False, indent=2)
    n = len(payload) if isinstance(payload, list) else 1
    print(f"  [OK]   {name}: {n} รายการ")

for name, path in CLASSIC.items():
    grab(name, f"{base}/proxy/network/api/s/{site}{path}")

It supports writes too, without requesting a CSRF token

What I did not expect was that this proxy supports writes, not just reads. PUT /rest/device/<id> to change port_overrides went through, as did POST /rest/networkconf, using only X-API-KEY, without logging in, storing cookies, and requesting a CSRF token to attach as I would when calling the legacy API directly.

That means work that previously required being at the machine, such as changing a port's VLAN, can be done from anywhere. But it also means that one mistake with this single key can disconnect the entire site, so I wrote the port-editing tool to always default to dry-run, requiring me to append --apply before it sends the actual request.

overrides = copy.deepcopy(sw.get("port_overrides") or [])
cur = next((o for o in overrides if o.get("port_idx") == a.port), None)
if cur is None:
    cur = {"port_idx": a.port, "setting_preference": "auto"}
    overrides.append(cur)
before = copy.deepcopy(cur)

if a.network:
    n = net_by_name.get(a.network)
    cur["native_networkconf_id"] = n["_id"]
    cur.setdefault("forward", "native")
    cur.setdefault("tagged_vlan_mgmt", "block_all")
if a.poe:
    cur["poe_mode"] = a.poe            # auto | off | pasv24

show(before, "ก่อน")
show(cur, "หลัง")
if before == cur:
    print("[OK] ไม่มีอะไรต้องเปลี่ยน")
    return 0
if not a.apply:
    print("\n[DRY-RUN] ยังไม่ได้แก้จริง - ใส่ --apply เพื่อยืนยัน")
    return 0

r = c.s.put(f"{base}/rest/device/{sw['_id']}",
            json={"port_overrides": overrides}, timeout=45)
if r.status_code == 200:
    print("[OK] แก้แล้ว - สวิตช์จะ provision ใหม่ราว 10-30 วินาที")

The critical detail is that port_overrides is one collection for the entire switch, not a per-port value. If I send only the single port I want to change, the existing settings for every other port disappear. I therefore have to deep copy the entire existing collection, change only the intended entry, and send the whole collection back. One mistake here resets every port to factory defaults at once.

python set_port.py --switch "SW-FLOOR2" --port 12 \
    --network "IoT" --name "CCTV-SW1" --poe off        # dry-run
python set_port.py --switch "SW-FLOOR2" --port 12 \
    --network "IoT" --name "CCTV-SW1" --poe off --apply

Tracking down failing cables with numbers nobody checks

After downloading stat/device, I found a valuable field that hardly anyone checks: each port's link_down_count. It is the number of times that port's link has gone down since the last boot.

port = next((p for p in sw.get("port_table", []) if p.get("port_idx") == a.port), {})
print(f"  สถานะพอร์ต: up={port.get('up')} speed={port.get('speed')} "
      f"link_down_count={port.get('link_down_count')} poe={port.get('poe_power')}W")

One port I found at this site had reached 2,411 drops and negotiated only 100 Mbps, even though both ends were gigabit. When these two things appear together, the answer is a failing cable, with no further guessing needed.

The right way to read this value is to look at the difference, not the raw count. I saved the state and compared it the next day. That time, the count moved from 2,411 to 2,608, an increase of 197, which coincided with a team installing equipment and plugging and unplugging cables all day, rather than the link flapping on its own. I then watched for another twenty-four minutes: the count did not move, and every device's uptime kept increasing without any reboots, meaning things had settled down. Looking at the raw count of 2,411 just once could lead to the wrong conclusion in either direction.

Reserve a fixed IP, then check whose address it might overlap

Another common task is reserving IPs for devices that need to be referenced by a fixed address, such as an NVR or a secondary switch. I wrote my script to always check for conflicts first, because the API allows duplicate IP reservations without any warning, leaving two machines fighting over the address when they are plugged in.

users = c.s.get(base + "/rest/user", timeout=60).json()["data"]
by_mac = {u.get("mac"): u for u in users}

# กันชน: IP ที่จะจอง ต้องไม่ถูกจองให้ MAC อื่นอยู่แล้ว
taken = {}
for u in users:
    if u.get("use_fixedip") and u.get("fixed_ip"):
        taken.setdefault(u["fixed_ip"], []).append(u["mac"])

for mac, (ip, name, net) in TARGETS.items():
    u = by_mac.get(mac)
    if not u:
        print(f"[ERR]  {mac}: ไม่มี user record - อุปกรณ์ต้องเคยต่อเข้ามาก่อน")
        continue
    clash = [m for m in taken.get(ip, []) if m != mac]
    if clash:
        print(f"[ERR]  {ip} ถูกจองให้ {clash} อยู่แล้ว - ข้าม")
        continue
    body = {"name": name, "usergroup_id": u.get("usergroup_id", ""),
            "use_fixedip": True, "fixed_ip": ip, "network_id": net, "noted": True}
    if args.apply:
        r = c.s.put(f"{base}/rest/user/{u['_id']}", json=body, timeout=45)

One useful detail is that an IP outside the DHCP pool but still within the same subnet can be reserved normally. This is better than reserving an address inside the pool because I do not have to worry about a conflict with a newly connected machine. I therefore reserve the beginning of the subnet exclusively for infrastructure devices and start the pool after that.

Zone-based firewall rules and an unexpected trap

The newer system replaces a sequential list of rules with zones assigned to VLANs and policies between zones. When I needed to let the administrators' VPN network reach every zone, I wrote a script to create 12 rules in one go, matching the actual number of destination zones, instead of clicking through them one by one in the web interface.

Before writing rules, I need to know which zones exist and what the current rules look like. I always read these two endpoints first and save the results for comparison after the changes.

BASE="https://api.ui.com/v1/connector/consoles/$UNIFI_CONSOLE_ID/proxy/network"

# รายชื่อโซนที่มีอยู่จริง ใช้กำหนดว่าต้องเขียนกี่กฎ
curl -s -H "X-API-KEY: $UI_API_KEY" \
  "$BASE/v2/api/site/default/firewall/zone" > zones-before.json

# กฎที่มีอยู่ตอนนี้ เก็บไว้เป็นตัวเทียบก่อนแก้
curl -s -H "X-API-KEY: $UI_API_KEY" \
  "$BASE/v2/api/site/default/firewall-policies" > policies-before.json

jq -r '.[] | "\(.name)  \(.action)"' policies-before.json | sort

What made this require less work than expected was that the firewall is already stateful: only the outbound direction needs a rule; the system handles the return traffic. There is no need to write a pair. The two zones I deliberately left alone were the internet zone, which had intentional blocking rules, and the zone for another VPN that was not a VLAN.

The discipline needed when holding a key this powerful

1. Keep the key in a file excluded from git, set permissions so only its owner can read it, and do not store it on a shared machine.
2. Every script that can write settings must default to dry-run and require typed confirmation before sending the actual request.
3. Read and save the existing state before every change, both for rollback and to see exactly what changed.
4. Fields that hold a whole collection, such as port_overrides, must be sent back in full; do not send only the entries being changed.
5. Treat the entire downloaded folder as confidential, and strip password fields when writing the files, not when displaying them.
6. Save work that could disconnect my own access, such as changing an uplink's VLAN, for a time when someone is on site.

Personally, I think working through the vendor's cloud this way is much safer than exposing management ports to the internet myself, because I am not adding an entry point for people to probe, and I can revoke the key immediately if anything looks suspicious. The tradeoff is that I have to trust their cloud to some extent. Anyone who cannot accept that tradeoff can connect through a VPN and call the local API directly. The code is the same throughout; only the base URL changes, so use whichever suits you.

Still on my list is a collector that stores every port's link_down_count in a small database each day and alerts only when the difference spikes, along with an automated comparison of the actual configuration against the intended plan. If I have left anything out or made a typo, I will come back and correct it later.