63 lines
2.0 KiB
Python
63 lines
2.0 KiB
Python
# services/combocore.py
|
|
import requests
|
|
from requests.exceptions import HTTPError
|
|
|
|
requests.packages.urllib3.disable_warnings()
|
|
|
|
def _fmt_host(host: str) -> str:
|
|
host = host.strip().strip("[]")
|
|
# if IPv6, wrap with []
|
|
if ":" in host and not host.startswith("["):
|
|
return f"[{host}]"
|
|
return host
|
|
|
|
def login(host: str, username="admin", password="Super4dmin!") -> str:
|
|
h = _fmt_host(host)
|
|
url = f"https://{h}/core/pls/api/1/auth/login"
|
|
r = requests.post(url, json={"username": username, "password": password}, verify=False, timeout=10)
|
|
r.raise_for_status()
|
|
j = r.json()
|
|
# API returns "access_token"
|
|
return j["access_token"]
|
|
|
|
def get_routes(host: str, token: str) -> list:
|
|
h = _fmt_host(host)
|
|
url = f"https://{h}/core/ncm/api/1/status/routes"
|
|
r = requests.get(url, headers={"Authorization": f"Bearer {token}"}, verify=False, timeout=10)
|
|
r.raise_for_status()
|
|
return r.json()
|
|
|
|
def extract_eth0_cidr_gw(routes: list) -> tuple[str, str]:
|
|
"""
|
|
From the routes list:
|
|
- CIDR = <prefsrc>/<mask-from-eth0-connected-route>
|
|
- GW = default route gateway on eth0
|
|
"""
|
|
prefsrc = None
|
|
mask = None
|
|
gw = None
|
|
|
|
# default gw via eth0
|
|
for r in routes:
|
|
if r.get("family") == "inet" and r.get("type") == "unicast" and r.get("dst") == "default" and r.get("dev") == "eth0":
|
|
gw = r.get("gateway")
|
|
break
|
|
|
|
# find eth0 connected network to derive mask, and prefsrc for IP
|
|
for r in routes:
|
|
if r.get("family") == "inet" and r.get("dev") == "eth0":
|
|
if not prefsrc and r.get("prefsrc"):
|
|
prefsrc = r.get("prefsrc")
|
|
dst = r.get("dst") or ""
|
|
if "/" in dst:
|
|
try:
|
|
mask = dst.split("/", 1)[1]
|
|
except Exception:
|
|
pass
|
|
if prefsrc and mask and gw:
|
|
break
|
|
|
|
if not (prefsrc and mask and gw):
|
|
raise ValueError("Unable to derive eth0 CIDR/GW from routes payload")
|
|
|
|
return f"{prefsrc}/{mask}", gw |