Refresh RackPeek and UniFi network inventory

automation/bin/unifi_network.py

@@ -0,0 +1,190 @@
+#!/usr/bin/env python3
+"""Minimal Doris helper for UniFi OS / Network inventory reads.
+
+Reads credentials from automation/.env by default and performs a cookie-based
+login against UniFi OS, then calls Network app endpoints through
+`/proxy/network/api/s/<site>/...`.
+
+Default action prints a compact JSON summary suitable for RackPeek refresh work.
+No secrets are printed.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import ssl
+import sys
+from http.cookiejar import CookieJar
+from pathlib import Path
+from typing import Any
+from urllib import error, request
+
+DEFAULT_TIMEOUT = 20
+SCRIPT_DIR = Path(__file__).resolve().parent
+REPO_ROOT = SCRIPT_DIR.parent.parent
+DEFAULT_ENV_PATH = REPO_ROOT / 'automation' / '.env'
+
+
+def load_env_file(path: Path) -> None:
+    if not path.exists():
+        return
+    for raw_line in path.read_text(encoding='utf-8').splitlines():
+        line = raw_line.strip()
+        if not line or line.startswith('#') or '=' not in line:
+            continue
+        key, value = line.split('=', 1)
+        os.environ.setdefault(key.strip(), value.strip().strip('"').strip("'"))
+
+
+def env_bool(name: str, default: bool) -> bool:
+    value = os.environ.get(name)
+    if value is None:
+        return default
+    return value.strip().lower() in {'1', 'true', 'yes', 'on'}
+
+
+class UniFiClient:
+    def __init__(self, base_url: str, username: str, password: str, *, verify_tls: bool = True, timeout: int = DEFAULT_TIMEOUT) -> None:
+        self.base_url = base_url.rstrip('/')
+        self.username = username
+        self.password = password
+        self.timeout = timeout
+        self.cookie_jar = CookieJar()
+        handlers: list[Any] = [request.HTTPCookieProcessor(self.cookie_jar)]
+        if not verify_tls:
+            handlers.append(request.HTTPSHandler(context=ssl._create_unverified_context()))
+        self.opener = request.build_opener(*handlers)
+
+    def _json_request(self, method: str, path: str, body: dict[str, Any] | None = None) -> tuple[int, Any]:
+        url = self.base_url + path
+        data = None
+        headers = {
+            'Accept': 'application/json',
+            'Content-Type': 'application/json',
+            'User-Agent': 'doris-unifi-helper/1.0',
+        }
+        if body is not None:
+            data = json.dumps(body).encode('utf-8')
+        req = request.Request(url, data=data, method=method.upper(), headers=headers)
+        try:
+            with self.opener.open(req, timeout=self.timeout) as resp:
+                raw = resp.read().decode('utf-8', 'replace')
+                ctype = resp.headers.get('Content-Type', '')
+                payload = json.loads(raw) if 'json' in ctype else raw
+                return resp.status, payload
+        except error.HTTPError as exc:
+            raw = exc.read().decode('utf-8', 'replace')
+            try:
+                payload = json.loads(raw)
+            except Exception:
+                payload = raw
+            return exc.code, payload
+
+    def login(self) -> None:
+        status, payload = self._json_request('POST', '/api/auth/login', {
+            'username': self.username,
+            'password': self.password,
+            'rememberMe': True,
+        })
+        if status >= 300:
+            raise SystemExit(f'UniFi login failed: HTTP {status} {payload!r}')
+
+    def get(self, path: str) -> Any:
+        status, payload = self._json_request('GET', path)
+        if status >= 300:
+            raise SystemExit(f'UniFi GET {path} failed: HTTP {status} {payload!r}')
+        return payload
+
+    def site_path(self, site: str, suffix: str) -> str:
+        suffix = suffix.lstrip('/')
+        return f'/proxy/network/api/s/{site}/{suffix}'
+
+
+def response_data(payload: Any) -> Any:
+    if isinstance(payload, dict) and 'data' in payload:
+        return payload['data']
+    return payload
+
+
+def device_summary(devices: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    out = []
+    for dev in devices:
+        out.append({
+            'name': dev.get('name') or dev.get('hostname') or dev.get('mac'),
+            'model': dev.get('model'),
+            'type': dev.get('type') or dev.get('device_type'),
+            'state': dev.get('state'),
+            'ip': dev.get('ip'),
+            'mac': dev.get('mac'),
+            'version': dev.get('version'),
+            'uplink': ((dev.get('uplink') or {}).get('name') if isinstance(dev.get('uplink'), dict) else None),
+        })
+    return out
+
+
+def client_summary(clients: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    out = []
+    for client in clients:
+        out.append({
+            'name': client.get('name') or client.get('hostname') or client.get('mac'),
+            'ip': client.get('ip'),
+            'mac': client.get('mac'),
+            'network': client.get('network'),
+            'ap_mac': client.get('ap_mac'),
+            'oui': client.get('oui'),
+        })
+    return out
+
+
+def build_summary(client: UniFiClient, site: str) -> dict[str, Any]:
+    devices = response_data(client.get(client.site_path(site, 'stat/device')))
+    health = response_data(client.get(client.site_path(site, 'stat/health')))
+    clients = response_data(client.get(client.site_path(site, 'stat/sta')))
+    return {
+        'site': site,
+        'device_count': len(devices),
+        'client_count': len(clients),
+        'devices': device_summary(devices),
+        'health': health,
+        'clients': client_summary(clients),
+    }
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description='Read UniFi Network inventory for Doris.')
+    parser.add_argument('--env-file', default=str(DEFAULT_ENV_PATH), help='Path to .env containing UNIFI_* vars')
+    parser.add_argument('--site', help='UniFi site name; defaults to UNIFI_SITE or default')
+    parser.add_argument('--raw', action='store_true', help='Print raw device/client/health payloads instead of compact summary')
+    parser.add_argument('--timeout', type=int, default=DEFAULT_TIMEOUT)
+    args = parser.parse_args()
+
+    load_env_file(Path(args.env_file))
+    base_url = os.environ.get('UNIFI_BASE_URL', 'https://10.5.0.1')
+    username = os.environ.get('UNIFI_USERNAME')
+    password = os.environ.get('UNIFI_PASSWORD')
+    site = args.site or os.environ.get('UNIFI_SITE', 'default')
+    verify_tls = env_bool('UNIFI_VERIFY_TLS', False)
+
+    if not username or not password:
+        raise SystemExit('Missing UNIFI_USERNAME or UNIFI_PASSWORD in env file/environment.')
+
+    client = UniFiClient(base_url, username, password, verify_tls=verify_tls, timeout=args.timeout)
+    client.login()
+
+    if args.raw:
+        payload = {
+            'devices': response_data(client.get(client.site_path(site, 'stat/device'))),
+            'health': response_data(client.get(client.site_path(site, 'stat/health'))),
+            'clients': response_data(client.get(client.site_path(site, 'stat/sta'))),
+        }
+    else:
+        payload = build_summary(client, site)
+
+    print(json.dumps(payload, indent=2, sort_keys=True))
+    return 0
+
+
+if __name__ == '__main__':
+    raise SystemExit(main())