191 lines
6.9 KiB
Python
191 lines
6.9 KiB
Python
#!/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())
|