คุมสวิตช์ทั้งไซต์จากนอกวง ด้วย cloud connector proxy ของ UniFi

พอดีไซต์ที่ผมดูแลอยู่คนละจังหวัดกับเครื่องที่ผมใช้ทำงาน จะเข้าไปแก้อะไรทีต้องขับรถ หรือไม่ก็โทรบอกให้คนหน้างานกดตาม ซึ่งช้าและพลาดง่ายทั้งสองทาง จะเปิดพอร์ตเข้ามาก็ไม่เอา เพราะการเปิดหน้าจัดการอุปกรณ์ออกอินเทอร์เน็ตคือการยกกุญแจให้คนทั้งโลกลองกด สุดท้ายผมเลยไปทางที่ผู้ผลิตเปิดไว้ให้อยู่แล้ว คือ API ฝั่งคลาวด์ แล้วพบว่ามันทำได้มากกว่าที่เอกสารบอกไว้เยอะมาก

เลข IP ชื่ออุปกรณ์ ชื่อ VLAN และ id ทุกตัวในบทความเป็นค่าตัวอย่าง ส่วน path ของ API ชื่อฟิลด์ และตัวเลขที่เจอคือของจริงจากระบบที่ผมดูแลด้วยคีย์ที่เจ้าของออกให้

สามทางที่คุยกับระบบนี้ได้ กับเหตุผลที่ผมเลือกทางที่สาม

โครงของ client ที่ผมใช้เป็นแบบนี้ ตัวคลาวด์เก็บแค่ session กับคีย์ แล้วมีเมธอด proxy() ตัวเดียวที่ต่อ path เองได้

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()

คีย์ทั้งหมดอยู่ในไฟล์ .env ข้างสคริปต์ ไม่อยู่ในโค้ด และไฟล์นั้นไม่ขึ้น 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

proxy ไม่ได้จำกัดแค่ integration API

จุดที่ทำให้งานนี้เปลี่ยนไปเลยคือผมลองเอา path หลัง /proxy/ เปลี่ยนจาก network/integration/v1 เป็นอย่างอื่นดู แล้วมันไปต่อได้จริง

สิ่งที่ต่อไปถึงได้จริงมีสามชุด ชุดแรกคือ API ตัวเก่าเต็มรูปแบบที่ /proxy/network/api/s/<site>/... ซึ่งมีทั้ง stat/device ที่คืน port_table พร้อมค่า PoE รายพอร์ต rest/networkconf rest/portconf และ list/user ชุดที่สองคือ API รุ่นสองสำหรับไฟร์วอลล์แบบแบ่งโซนที่ /proxy/network/v2/api/site/<site>/firewall-policies และ /firewall/zone ชุดที่สามคือฝั่งกล้องที่ /proxy/protect/api/bootstrap ซึ่งคืนข้อมูลทั้งก้อนละเอียดกว่า integration API มาก

ตัวดูดข้อมูลของผมเลยเป็นแค่ลูปสั้นๆ วนตามรายการ path แล้วเขียนลงไฟล์ JSON แยกตามชื่อ

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

เขียนได้ด้วย และไม่ต้องขอ CSRF

เรื่องที่ผมคาดไม่ถึงคือ proxy ตัวนี้ เขียนได้ ไม่ใช่แค่อ่าน PUT /rest/device/<id> เพื่อแก้ port_overrides ก็ผ่าน POST /rest/networkconf ก็ผ่าน โดยใช้แค่ X-API-KEY ไม่ต้องล็อกอินเก็บ cookie แล้วไปขอ CSRF token มาแนบเหมือนตอนคุยกับ API ตัวเก่าตรงๆ

นั่นแปลว่างานที่เมื่อก่อนต้องอยู่หน้าเครื่อง อย่างการเปลี่ยน VLAN ของพอร์ตหนึ่งพอร์ต ทำจากที่ไหนก็ได้ แต่ก็แปลว่าคีย์ตัวเดียวนี้พลาดทีเดียวคือทั้งไซต์เน็ตหลุด ผมเลยเขียนตัวแก้พอร์ตให้ เริ่มต้นเป็น dry-run เสมอ ต้องพิมพ์ --apply ต่อท้ายเองถึงจะยิงจริง

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 วินาที")

จุดสำคัญที่ต้องระวังคือ port_overrides เป็น ก้อนเดียวทั้งสวิตช์ ไม่ใช่รายพอร์ต ถ้าส่งไปแค่พอร์ตที่จะแก้พอร์ตเดียว การตั้งค่าเดิมของพอร์ตอื่นหายหมด ผมเลยต้อง deep copy ของเดิมมาทั้งชุด แก้เฉพาะรายการที่ต้องการ แล้วส่งกลับไปทั้งก้อน ตรงนี้เขียนผิดทีเดียวคือคืนค่าโรงงานให้ทุกพอร์ตพร้อมกัน

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

ไล่หาสายเสื่อมจากตัวเลขที่ไม่มีใครเปิดดู

ตอนดูด stat/device มาแล้ว ในนั้นมีฟิลด์ที่มีค่ามากแต่แทบไม่มีใครเปิดดูคือ link_down_count ของแต่ละพอร์ต มันคือจำนวนครั้งที่ลิงก์พอร์ตนั้นตกนับตั้งแต่บูตล่าสุด

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

พอร์ตที่ผมเจอในไซต์นี้พอร์ตหนึ่ง ค่ามันขึ้นมาที่ 2,411 ครั้ง และเจรจาความเร็วได้แค่ 100 Mbps ทั้งที่ปลายทั้งสองข้างเป็นกิกะบิตทั้งคู่ สองอย่างนี้มาด้วยกันเมื่อไหร่ คำตอบคือสายเส้นนั้นเสื่อม ไม่ต้องเดาต่อ

วิธีอ่านค่านี้ให้ถูกคือ ดูส่วนต่าง ไม่ใช่ดูตัวเลขดิบ ผมเก็บสถานะไว้แล้วมาเทียบวันถัดไป รอบนั้นค่ามันขยับจาก 2,411 เป็น 2,608 คือเพิ่ม 197 ซึ่งตรงกับวันที่มีทีมเข้าไปติดตั้งอุปกรณ์เสียบถอดกันทั้งวัน ไม่ใช่ลิงก์กระพริบเอง หลังจากนั้นผมเฝ้าดูต่ออีกยี่สิบสี่นาที ค่าไม่ขยับเลยและ uptime ของอุปกรณ์ทุกตัวเดินหน้าไม่มีตัวไหนรีบูต แปลว่านิ่งแล้ว ถ้าดูแค่เลขดิบ 2,411 ครั้งเดียวจะสรุปผิดได้ทั้งสองทาง

จอง fixed IP แล้วอย่าลืมเช็กว่าไปทับของใคร

งานที่ทำบ่อยอีกอย่างคือจอง IP ให้อุปกรณ์ที่ต้องอ้างด้วยเลขคงที่ อย่าง NVR หรือสวิตช์ตัวรอง ตัวสคริปต์ผมเขียนให้เช็กชนก่อนเสมอ เพราะ API มันยอมให้จอง IP ซ้ำได้โดยไม่เตือนอะไรเลย แล้วผลคือสองเครื่องแย่งกันตอนต่อสาย

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)

เกร็ดที่ใช้ได้จริงคือ IP ที่อยู่ นอกช่วง DHCP pool แต่ยังอยู่ใน subnet เดียวกัน จองได้ปกติ ซึ่งดีกว่าจองทับในช่วง pool เพราะไม่ต้องกลัวชนกับเครื่องที่เพิ่งเข้ามาใหม่ ผมเลยกันช่วงต้น subnet ไว้ให้อุปกรณ์โครงสร้างพื้นฐานล้วน แล้วให้ pool เริ่มหลังจากนั้น

กฎไฟร์วอลล์แบบแบ่งโซน กับกับดักที่ไม่มีใครนึกถึง

ระบบรุ่นใหม่เปลี่ยนจากกฎเรียงเป็นบรรทัดมาเป็นการแบ่งโซนต่อ VLAN แล้วเขียนนโยบายระหว่างโซน ตอนผมต้องเปิดให้วง VPN ของผู้ดูแลเข้าถึงทุกโซนได้ ก็เขียนสคริปต์สร้างนโยบายทีเดียว 12 กฎ ตามจำนวนโซนปลายทางที่มีอยู่จริง แทนที่จะกดในหน้าเว็บทีละอัน

ก่อนเขียนกฎ ต้องรู้ก่อนว่าตอนนี้มีโซนอะไรอยู่บ้างและกฎเดิมเป็นอย่างไร ยิงอ่านสองเส้นนี้ก่อนเสมอ แล้วเก็บผลไว้เทียบหลังแก้

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

เรื่องที่ทำให้เขียนน้อยกว่าที่คิดคือไฟร์วอลล์ตัวนี้เป็นแบบ stateful อยู่แล้ว สร้างแค่ขาไปพอ ขากลับระบบทำให้เอง ไม่ต้องเขียนคู่ ส่วนสองโซนที่ผมตั้งใจไม่แตะคือโซนอินเทอร์เน็ตที่มีกฎบล็อกตั้งใจอยู่ กับโซนของ VPN อีกชุดที่ไม่ใช่ VLAN

วินัยที่ต้องมี เวลาถือคีย์ที่ทำได้ขนาดนี้

1. คีย์อยู่ในไฟล์ที่ไม่ขึ้น git ตั้ง permission ให้เจ้าของอ่านได้คนเดียว และไม่วางไว้ในเครื่องที่คนอื่นใช้ร่วม
2. ทุกสคริปต์ที่เขียนค่าได้ ต้องเป็น dry-run โดยค่าเริ่มต้น ให้พิมพ์ยืนยันเองถึงจะยิงจริง
3. อ่านสถานะเดิมมาเก็บก่อนแก้ทุกครั้ง เผื่อต้องย้อนกลับ และเพื่อให้เห็นส่วนต่างว่าเปลี่ยนอะไรไปจริง
4. ฟิลด์ที่เป็นก้อนรวมอย่าง port_overrides ต้องส่งกลับทั้งก้อน อย่าส่งเฉพาะรายการที่แก้
5. ไฟล์ที่ดูดมาถือเป็นเอกสารลับทั้งโฟลเดอร์ ตัดฟิลด์รหัสออกตั้งแต่ตอนเขียนไฟล์ ไม่ใช่ตอนแสดงผล
6. งานที่ทำให้ตัวเองหลุดได้ อย่างการแก้ VLAN ของ uplink ให้เก็บไว้ทำตอนมีคนอยู่หน้างาน

ส่วนตัวผมคิดว่าการทำงานผ่านคลาวด์ของผู้ผลิตแบบนี้ ปลอดภัยกว่าการเปิดพอร์ตจัดการออกอินเทอร์เน็ตเองเยอะ เพราะเราไม่ได้เพิ่มพื้นที่ให้ใครมาลองกด และคีย์เพิกถอนได้ทันทีที่สงสัย แลกมาด้วยการที่เราต้องเชื่อคลาวด์ของเขาระดับหนึ่ง ใครรับข้อแลกเปลี่ยนนี้ไม่ได้ก็ใช้ VPN เข้าไปเองแล้วยิง API ในวงตรง วิธีเขียนเหมือนกันทั้งหมด ต่างแค่ base URL ตามสะดวกครับ

ที่ยังค้างในแผนคือทำตัวเก็บ link_down_count ทุกพอร์ตทุกวันลงฐานข้อมูลเล็กๆ แล้วแจ้งเตือนเฉพาะตอนส่วนต่างพุ่ง กับเขียนตัวเทียบคอนฟิกจริงกับแผนที่ตั้งใจไว้ให้อัตโนมัติ ตรงไหนผมอธิบายตกหล่นหรือพิมพ์ผิด เดี๋ยวกลับมาแก้เพิ่มให้ภายหลังครับ