#!/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//...`. 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())