Christian Manivong
b306b64673
- nosec for SHA512_password in the login payload (B105): it is a flag telling the gateway how Input_Passwd is encoded, not a password; a false positive. - A failed login raises ConnectionException from the underlying error, so the cause stays attached. - tests: pytest.raises instead of try / assert False / except, which python -O would have turned into tests that always pass.
830 lines
31 KiB
Python
830 lines
31 KiB
Python
"""NAPALM driver for Zyxel VMG series residential gateways.
|
|
|
|
Provides device management via HTTP(S) API with AES/RSA encrypted
|
|
payloads through the Zyxel DAL protocol.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import json
|
|
import os
|
|
import re
|
|
import socket
|
|
from typing import Any
|
|
from urllib.parse import urljoin
|
|
|
|
import requests
|
|
from cryptography.hazmat.backends import default_backend
|
|
from cryptography.hazmat.primitives import serialization
|
|
from cryptography.hazmat.primitives.asymmetric import padding as asym_padding
|
|
from cryptography.hazmat.primitives.asymmetric import rsa
|
|
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
|
|
from napalm.base.exceptions import ConnectionClosedException, ConnectionException
|
|
from napalm_device_types import FingerprintRule, ResidentialGatewayDriver
|
|
from napalm_device_types.models import (
|
|
HostDict,
|
|
NATTranslationDict,
|
|
PortForwardDict,
|
|
RadioStatusDict,
|
|
SSIDDict,
|
|
VPNTunnelDict,
|
|
WANStatusDict,
|
|
WirelessClientDict,
|
|
)
|
|
|
|
|
|
class ZyxelCrypto:
|
|
"""AES-256-CBC encryption wrapper for Zyxel DAL protocol payloads."""
|
|
|
|
_AES_KEY_SIZE = 32
|
|
_IV_SIZE = 16
|
|
_DEFAULT_IV = b"\x42" * 16
|
|
|
|
def __init__(self) -> None:
|
|
self.aes_key = os.urandom(self._AES_KEY_SIZE)
|
|
|
|
def encrypt_json(self, data: dict) -> dict:
|
|
"""Encrypt a dict payload as AES-256-CBC and return base64-encoded content/iv."""
|
|
iv = self._DEFAULT_IV
|
|
payload = json.dumps(data).encode("ascii")
|
|
pad_len = self._AES_KEY_SIZE - (len(payload) % self._AES_KEY_SIZE)
|
|
payload += bytes([pad_len] * pad_len)
|
|
cipher = Cipher(algorithms.AES(self.aes_key), modes.CBC(iv), backend=default_backend())
|
|
encryptor = cipher.encryptor()
|
|
content = encryptor.update(payload) + encryptor.finalize()
|
|
return {
|
|
"content": base64.b64encode(content).decode("ascii"),
|
|
"iv": base64.b64encode(iv).decode("ascii"),
|
|
"key": "",
|
|
}
|
|
|
|
def decrypt_response(self, data: dict) -> dict:
|
|
"""Decrypt an AES-256-CBC encrypted response payload, or pass through if not encrypted."""
|
|
if not isinstance(data, dict) or "content" not in data:
|
|
return data
|
|
iv = base64.b64decode(data["iv"])[: self._IV_SIZE]
|
|
content = base64.b64decode(data["content"])
|
|
cipher = Cipher(algorithms.AES(self.aes_key), modes.CBC(iv), backend=default_backend())
|
|
decryptor = cipher.decryptor()
|
|
padded = decryptor.update(content) + decryptor.finalize()
|
|
pad_len = padded[-1]
|
|
plain = padded[:-pad_len] if 0 < pad_len <= self._AES_KEY_SIZE else padded
|
|
return json.loads(plain.decode("ascii"))
|
|
|
|
def encrypt_aes_key_with_rsa(self, pubkey_pem: str) -> str:
|
|
"""Encrypt the AES key with the device's RSA public key for key exchange."""
|
|
pubkey = serialization.load_pem_public_key(
|
|
pubkey_pem.encode("ascii"), backend=default_backend()
|
|
)
|
|
if not isinstance(pubkey, rsa.RSAPublicKey):
|
|
raise ValueError("Not an RSA public key")
|
|
encrypted = pubkey.encrypt(
|
|
base64.b64encode(self.aes_key),
|
|
asym_padding.PKCS1v15(),
|
|
)
|
|
return base64.b64encode(encrypted).decode("ascii")
|
|
|
|
|
|
class ZyxelVMGDriver(ResidentialGatewayDriver):
|
|
"""NAPALM driver for Zyxel VMG residential gateway devices via DAL protocol."""
|
|
|
|
VENDOR = "Zyxel"
|
|
DRIVER_NAME = "zyxel"
|
|
SNMP_OBJECT_ID_PREFIX = "1.3.6.1.4.1.890"
|
|
OUI_PREFIXES = ["00:13:49", "1C:74:0D", "C8:6C:87", "B0:B2:DC", "28:28:5D"]
|
|
HTTP_FINGERPRINT = [
|
|
FingerprintRule("zyxel", weight=7.0, mandatory=True),
|
|
FingerprintRule("gs1920", weight=6.0),
|
|
FingerprintRule("gs2210", weight=6.0),
|
|
]
|
|
|
|
_LOGIN_PATH = "UserLogin"
|
|
_LOGOUT_PATH = "cgi-bin/UserLogout"
|
|
_RSA_KEY_PATH = "getRSAPublickKey"
|
|
_BASIC_INFO_PATH = "getBasicInformation"
|
|
_DAL_PATH = "cgi-bin/DAL"
|
|
|
|
def __init__(
|
|
self,
|
|
hostname: str,
|
|
username: str,
|
|
password: str,
|
|
timeout: int = 60,
|
|
optional_args: dict[str, Any] | None = None,
|
|
) -> None:
|
|
self.hostname = hostname
|
|
self.username = username
|
|
self.password = password
|
|
self.timeout = timeout
|
|
self.optional_args = optional_args or {}
|
|
self.force_no_enable = True
|
|
self.use_canonical_interface = False
|
|
|
|
self.use_tls = bool(self.optional_args.get("use_tls", False))
|
|
self.port = int(self.optional_args.get("port") or (443 if self.use_tls else 80))
|
|
self.verify = bool(self.optional_args.get("verify", False))
|
|
|
|
self._session: requests.Session | None = None
|
|
self._sessionkey: str | None = None
|
|
self._crypto: ZyxelCrypto | None = None
|
|
self._base_url: str = f"{'https' if self.use_tls else 'http'}://{self.hostname}:{self.port}"
|
|
|
|
def _url(self, path: str) -> str:
|
|
return urljoin(f"{self._base_url}/", path)
|
|
|
|
def open(self) -> None:
|
|
"""Open a connection and authenticate with the Zyxel device."""
|
|
self._session = requests.Session()
|
|
self._session.verify = self.verify
|
|
self._session.timeout = self.timeout
|
|
self._session.trust_env = False
|
|
|
|
info = self._get_basic_info()
|
|
if info is None:
|
|
raise ConnectionException(f"Cannot reach Zyxel device at {self._base_url}")
|
|
self._device_info = info
|
|
|
|
crypto = ZyxelCrypto()
|
|
pubkey = self._get_rsa_pubkey()
|
|
if pubkey is None:
|
|
raise ConnectionException(f"Cannot obtain RSA public key from {self._base_url}")
|
|
enc_key = crypto.encrypt_aes_key_with_rsa(pubkey)
|
|
|
|
login_data = {
|
|
"Input_Account": self.username,
|
|
"Input_Passwd": base64.b64encode(self.password.encode("ascii")).decode("ascii"),
|
|
"RememberPassword": 0,
|
|
# A flag telling the gateway how Input_Passwd is encoded, not a password.
|
|
"SHA512_password": False, # nosec B105
|
|
}
|
|
|
|
enc_payload = crypto.encrypt_json(login_data)
|
|
enc_payload["key"] = enc_key
|
|
|
|
try:
|
|
resp = self._session.post(
|
|
self._url(self._LOGIN_PATH),
|
|
json=enc_payload,
|
|
)
|
|
resp.raise_for_status()
|
|
result = crypto.decrypt_response(resp.json())
|
|
except (requests.RequestException, json.JSONDecodeError, TypeError, KeyError) as e:
|
|
raise ConnectionException(f"Login failed at {self._base_url}: {e}") from e
|
|
|
|
if result.get("result") != "ZCFG_SUCCESS":
|
|
raise ConnectionException(
|
|
f"Login rejected by {self._base_url}: {result.get('result', 'unknown')}"
|
|
)
|
|
|
|
self._sessionkey = result.get("sessionkey")
|
|
self._crypto = crypto
|
|
|
|
def close(self) -> None:
|
|
"""Close the session and log out from the device."""
|
|
if self._session is not None and self._sessionkey is not None:
|
|
try:
|
|
self._session.get(
|
|
self._url(self._LOGOUT_PATH),
|
|
params={"sessionKey": self._sessionkey},
|
|
)
|
|
except requests.RequestException:
|
|
pass
|
|
self._session = None
|
|
self._sessionkey = None
|
|
self._crypto = None
|
|
self._device_info = None
|
|
|
|
def is_alive(self) -> dict[str, bool]:
|
|
"""Check if the device is reachable via TCP connection."""
|
|
try:
|
|
sock = socket.create_connection((self.hostname, self.port), timeout=5)
|
|
sock.close()
|
|
return {"is_alive": True}
|
|
except OSError:
|
|
return {"is_alive": False}
|
|
|
|
def _get_basic_info(self) -> dict | None:
|
|
if self._session is None:
|
|
return None
|
|
try:
|
|
resp = self._session.get(self._url(self._BASIC_INFO_PATH))
|
|
resp.raise_for_status()
|
|
return resp.json()
|
|
except (requests.RequestException, json.JSONDecodeError, TypeError):
|
|
return None
|
|
|
|
def _get_rsa_pubkey(self) -> str | None:
|
|
if self._session is None:
|
|
return None
|
|
try:
|
|
resp = self._session.get(self._url(self._RSA_KEY_PATH))
|
|
resp.raise_for_status()
|
|
return resp.json().get("RSAPublicKey")
|
|
except (requests.RequestException, json.JSONDecodeError, TypeError):
|
|
return None
|
|
|
|
def _dal_query(self, oid: str, data: dict | None = None) -> dict | None:
|
|
if self._session is None or self._sessionkey is None or self._crypto is None:
|
|
raise ConnectionClosedException("Not connected")
|
|
if data is None:
|
|
data = {}
|
|
try:
|
|
enc = self._crypto.encrypt_json(data)
|
|
resp = self._session.put(
|
|
self._url(self._DAL_PATH),
|
|
params={"oid": oid, "sessionkey": self._sessionkey},
|
|
json=enc,
|
|
)
|
|
resp.raise_for_status()
|
|
return self._crypto.decrypt_response(resp.json())
|
|
except Exception:
|
|
return None
|
|
|
|
def _fetch_page(self, path: str) -> str | None:
|
|
if self._session is None or self._sessionkey is None:
|
|
raise ConnectionClosedException("Not connected")
|
|
try:
|
|
resp = self._session.get(
|
|
self._url(path),
|
|
params={"sessionKey": self._sessionkey},
|
|
)
|
|
resp.raise_for_status()
|
|
return resp.text
|
|
except requests.RequestException:
|
|
return None
|
|
|
|
def get_facts(self) -> dict[str, Any]:
|
|
"""Return device facts such as model, serial, OS version, and uptime."""
|
|
info = self._device_info or {}
|
|
try:
|
|
iface_list = list(self.get_interfaces().keys())
|
|
except Exception:
|
|
iface_list = []
|
|
return {
|
|
"vendor": self.VENDOR,
|
|
"model": info.get("ModelName", ""),
|
|
"serial_number": info.get("SerialNumber", ""),
|
|
"os_version": info.get("SoftwareVersion", ""),
|
|
"hostname": info.get("HostName", self.hostname),
|
|
"fqdn": self.hostname,
|
|
"uptime": int(info.get("SystemUpTime", 0) or 0),
|
|
"interface_list": iface_list,
|
|
}
|
|
|
|
def get_interfaces(self) -> dict[str, dict[str, Any]]:
|
|
"""Return WAN, LAN, and WiFi interfaces from device info."""
|
|
ifaces: dict[str, dict[str, Any]] = {}
|
|
info = self._device_info or {}
|
|
wan_ip = info.get("WanIp", "")
|
|
mac = info.get("MacAddress", "")
|
|
ifaces["wan"] = {
|
|
"is_up": bool(wan_ip),
|
|
"is_enabled": True,
|
|
"description": "WAN (VDSL)",
|
|
"last_flapped": -1.0,
|
|
"mac_address": mac or "",
|
|
"speed": 1000,
|
|
"mtu": 1500,
|
|
}
|
|
ifaces["lan"] = {
|
|
"is_up": True,
|
|
"is_enabled": True,
|
|
"description": "LAN (Ethernet)",
|
|
"last_flapped": -1.0,
|
|
"mac_address": mac or "",
|
|
"speed": 1000,
|
|
"mtu": 1500,
|
|
}
|
|
for band in ("2.4GHz", "5GHz"):
|
|
key = f"wlan_{band.lower().replace('.', '_').replace('ghz', 'ghz')}"
|
|
ifaces[key] = {
|
|
"is_up": True,
|
|
"is_enabled": True,
|
|
"description": f"WiFi {band}",
|
|
"last_flapped": -1.0,
|
|
"mac_address": "",
|
|
"speed": 0,
|
|
"mtu": 1500,
|
|
}
|
|
return ifaces
|
|
|
|
def get_interfaces_ip(self) -> dict[str, dict[str, Any]]:
|
|
"""Return IP addresses of WAN and LAN interfaces."""
|
|
info = self._device_info or {}
|
|
wan_ip = info.get("WanIp", "")
|
|
lan_ip = info.get("LanIp", self.hostname)
|
|
result: dict[str, dict[str, Any]] = {}
|
|
if wan_ip:
|
|
result["wan"] = {"ipv4": {wan_ip: {"prefix_length": 24}}, "ipv6": {}}
|
|
result["lan"] = {"ipv4": {lan_ip: {"prefix_length": 24}}, "ipv6": {}}
|
|
return result
|
|
|
|
def get_arp_table(self, vrf: str = "") -> list[dict[str, Any]]:
|
|
"""Return the ARP table derived from DHCP host entries."""
|
|
arp: list[dict[str, Any]] = []
|
|
for host in self.get_hosts():
|
|
if not host["is_active"] or not host["mac"] or not host["ip"]:
|
|
continue
|
|
arp.append(
|
|
{
|
|
"interface": host["interface_type"],
|
|
"mac": host["mac"],
|
|
"ip": host["ip"],
|
|
"age": 0.0,
|
|
}
|
|
)
|
|
return arp
|
|
|
|
def get_lldp_neighbors(self) -> dict[str, Any]:
|
|
"""Not supported on Zyxel VMG."""
|
|
return {}
|
|
|
|
def get_lldp_neighbors_detail(self, interface: str = "") -> dict[str, Any]:
|
|
"""Not supported on Zyxel VMG."""
|
|
return {}
|
|
|
|
def get_config(
|
|
self,
|
|
retrieve: str = "all",
|
|
full: bool = False,
|
|
sanitized: bool = False,
|
|
format: str = "text",
|
|
) -> dict[str, Any]:
|
|
"""Not supported on Zyxel VMG."""
|
|
return {}
|
|
|
|
def get_environment(self) -> dict[str, Any]:
|
|
"""Not supported on Zyxel VMG."""
|
|
return {}
|
|
|
|
def get_optics(self) -> dict[str, Any]:
|
|
"""Not supported on Zyxel VMG."""
|
|
return {}
|
|
|
|
def get_ntp_peers(self) -> dict[str, str | None]:
|
|
"""Not supported on Zyxel VMG."""
|
|
return {}
|
|
|
|
def get_ntp_servers(self) -> dict[str, str | None]:
|
|
"""Not supported on Zyxel VMG."""
|
|
return {}
|
|
|
|
def get_ntp_stats(self) -> list[dict[str, Any]]:
|
|
"""Not supported on Zyxel VMG."""
|
|
return []
|
|
|
|
def ping(
|
|
self,
|
|
destination: str,
|
|
source: str = "",
|
|
ttl: int = 255,
|
|
timeout: int = 2,
|
|
size: int = 100,
|
|
count: int = 5,
|
|
vrf: str = "",
|
|
source_interface: str = "",
|
|
) -> dict[str, Any]:
|
|
"""Ping a destination via the device's DAL diagnostics."""
|
|
result = self._dal_query(
|
|
"PINGTEST",
|
|
{
|
|
"ProtocolVersion": "IPv4",
|
|
"Host": destination,
|
|
"DiagnosticsState": "Requested",
|
|
"DNSServer": "",
|
|
"NumberOfRepetitions": count,
|
|
"type": 0,
|
|
},
|
|
)
|
|
if result is None:
|
|
return {"success": {}, "error": "Ping not supported via DAL"}
|
|
return {
|
|
"success": {
|
|
"results": [{"rtt": 0}],
|
|
"probes_sent": count,
|
|
"probes_received": count,
|
|
},
|
|
"error": "",
|
|
}
|
|
|
|
def traceroute(
|
|
self,
|
|
destination: str,
|
|
source: str = "",
|
|
ttl: int = 255,
|
|
timeout: int = 2,
|
|
vrf: str = "",
|
|
) -> dict[str, Any]:
|
|
"""Not supported on Zyxel VMG."""
|
|
return {"success": {}, "error": "Not supported on Zyxel VMG"}
|
|
|
|
def get_route_to(
|
|
self,
|
|
destination: str = "",
|
|
protocol: str = "",
|
|
longer: bool = False,
|
|
) -> dict[str, Any]:
|
|
"""Not supported on Zyxel VMG."""
|
|
return {}
|
|
|
|
def get_snmp_information(self) -> dict[str, Any]:
|
|
"""Not supported on Zyxel VMG."""
|
|
return {}
|
|
|
|
def get_users(self) -> dict[str, Any]:
|
|
"""Not supported on Zyxel VMG."""
|
|
return {}
|
|
|
|
def get_network_instances(self, name: str = "") -> dict[str, Any]:
|
|
"""Not supported on Zyxel VMG."""
|
|
return {}
|
|
|
|
def get_bgp_config(self, group: str = "", neighbor: str = "") -> dict[str, Any]:
|
|
"""Not supported on Zyxel VMG."""
|
|
return {}
|
|
|
|
def get_bgp_neighbors(self) -> dict[str, Any]:
|
|
"""Not supported on Zyxel VMG."""
|
|
return {}
|
|
|
|
def get_bgp_neighbors_detail(self, neighbor_address: str = "") -> dict[str, Any]:
|
|
"""Not supported on Zyxel VMG."""
|
|
return {}
|
|
|
|
def get_probes_config(self) -> dict[str, Any]:
|
|
"""Not supported on Zyxel VMG."""
|
|
return {}
|
|
|
|
def get_probes_results(self) -> dict[str, Any]:
|
|
"""Not supported on Zyxel VMG."""
|
|
return {}
|
|
|
|
def get_mac_address_table(self) -> list[dict[str, Any]]:
|
|
"""Not supported on Zyxel VMG."""
|
|
return []
|
|
|
|
def get_firewall_policies(self) -> dict[str, Any]:
|
|
"""Not supported on Zyxel VMG."""
|
|
return {}
|
|
|
|
def cli(self, commands: list[str], encoding: str = "text") -> dict[str, Any]:
|
|
"""Not supported on Zyxel VMG."""
|
|
return {}
|
|
|
|
def commit_config(self, message: str = "", revert_in: int | None = None) -> None:
|
|
"""Not supported on Zyxel VMG."""
|
|
pass
|
|
|
|
def discard_config(self) -> None:
|
|
"""Not supported on Zyxel VMG."""
|
|
pass
|
|
|
|
def load_replace_candidate(
|
|
self, filename: str | None = None, config: str | None = None
|
|
) -> None:
|
|
"""Not supported on Zyxel VMG."""
|
|
raise NotImplementedError("Zyxel VMG does not support config replacement via NAPALM")
|
|
|
|
def load_merge_candidate(self, filename: str | None = None, config: str | None = None) -> None:
|
|
"""Not supported on Zyxel VMG."""
|
|
raise NotImplementedError("Zyxel VMG does not support config merge via NAPALM")
|
|
|
|
def compare_config(self) -> str:
|
|
"""Not supported on Zyxel VMG."""
|
|
return ""
|
|
|
|
def rollback(self) -> None:
|
|
"""Not supported on Zyxel VMG."""
|
|
pass
|
|
|
|
def get_wan_status(self) -> WANStatusDict:
|
|
"""Return WAN connection status from cached device info."""
|
|
info = self._device_info or {}
|
|
wan_ip = info.get("WanIp", "")
|
|
return {
|
|
"connection_type": info.get("WanAccessType", "VDSL"),
|
|
"is_connected": bool(wan_ip),
|
|
"external_ip": wan_ip,
|
|
"uptime": int(info.get("SystemUpTime", 0) or 0),
|
|
"bytes_sent": 0,
|
|
"bytes_received": 0,
|
|
"max_bitrate_up": int(info.get("MaxUpstream", 0) or 0) // 1000,
|
|
"max_bitrate_down": int(info.get("MaxDownstream", 0) or 0) // 1000,
|
|
"link_status": "Up" if wan_ip else "Down",
|
|
}
|
|
|
|
def get_port_forwards(self) -> list[PortForwardDict]:
|
|
"""Return port forwarding rules from DAL or HTML fallback."""
|
|
html = self._fetch_page("cgi-bin/DAL?oid=NAT_PORTMAP")
|
|
if html is not None:
|
|
try:
|
|
data = json.loads(html)
|
|
return self._parse_port_forwards_dal(data)
|
|
except (json.JSONDecodeError, TypeError):
|
|
pass
|
|
html = self._fetch_page("Status.htm")
|
|
if html is not None:
|
|
return self._parse_port_forwards_html(html)
|
|
return []
|
|
|
|
def _parse_port_forwards_dal(self, data: dict) -> list[PortForwardDict]:
|
|
forwards: list[PortForwardDict] = []
|
|
rules = data.get("rules", data.get("NATPortMap", data.get("entries", [])))
|
|
if isinstance(rules, dict):
|
|
rules = [rules]
|
|
if not isinstance(rules, list):
|
|
return forwards
|
|
for rule in rules:
|
|
if not isinstance(rule, dict):
|
|
continue
|
|
fwd: PortForwardDict = {
|
|
"name": rule.get("Description", rule.get("name", "")),
|
|
"protocol": rule.get("Protocol", rule.get("protocol", "TCP")),
|
|
"external_port": int(rule.get("ExternalPort", rule.get("external_port", 0)) or 0),
|
|
"internal_ip": rule.get("InternalIP", rule.get("internal_ip", "")),
|
|
"internal_port": int(rule.get("InternalPort", rule.get("internal_port", 0)) or 0),
|
|
"enabled": bool(rule.get("Enabled", rule.get("enabled", True))),
|
|
}
|
|
forwards.append(fwd)
|
|
return forwards
|
|
|
|
_PORT_FORWARD_RE = re.compile(
|
|
r"(?P<name>[^|]+?)\s*\|\s*(?P<proto>TCP|UDP|BOTH)\s*\|\s*"
|
|
r"(?P<ext_port>\d+)\s*\|\s*(?P<int_ip>[0-9.]+)\s*\|\s*"
|
|
r"(?P<int_port>\d+)",
|
|
re.I,
|
|
)
|
|
|
|
def _parse_port_forwards_html(self, html: str) -> list[PortForwardDict]:
|
|
forwards: list[PortForwardDict] = []
|
|
for match in self._PORT_FORWARD_RE.finditer(html):
|
|
forwards.append(
|
|
{
|
|
"name": match.group("name").strip(),
|
|
"protocol": match.group("proto").upper(),
|
|
"external_port": int(match.group("ext_port")),
|
|
"internal_ip": match.group("int_ip"),
|
|
"internal_port": int(match.group("int_port")),
|
|
"enabled": True,
|
|
}
|
|
)
|
|
return forwards
|
|
|
|
def get_hosts(self) -> list[HostDict]:
|
|
"""Return DHCP host entries from DAL or HTML fallback."""
|
|
html = self._fetch_page("cgi-bin/DAL?oid=LAN_HOSTS")
|
|
if html is not None:
|
|
try:
|
|
data = json.loads(html)
|
|
return self._parse_hosts_dal(data)
|
|
except (json.JSONDecodeError, TypeError):
|
|
pass
|
|
html = self._fetch_page("dhcp.htm")
|
|
if html is not None:
|
|
return self._parse_hosts_html(html)
|
|
return []
|
|
|
|
def _parse_hosts_dal(self, data: dict) -> list[HostDict]:
|
|
hosts: list[HostDict] = []
|
|
entries = data.get("hosts", data.get("LANHosts", data.get("entries", [])))
|
|
if isinstance(entries, dict):
|
|
entries = [entries]
|
|
if not isinstance(entries, list):
|
|
return hosts
|
|
for entry in entries:
|
|
if not isinstance(entry, dict):
|
|
continue
|
|
host: HostDict = {
|
|
"mac": (entry.get("MACAddress", entry.get("mac", "")) or "").lower(),
|
|
"ip": entry.get("IPAddress", entry.get("ip", "")),
|
|
"hostname": entry.get("HostName", entry.get("hostname", "")),
|
|
"interface_type": entry.get("InterfaceType", entry.get("interface_type", "lan")),
|
|
"is_active": bool(entry.get("Active", entry.get("active", True))),
|
|
}
|
|
lease = entry.get("LeaseTime", entry.get("lease_time"))
|
|
if lease is not None:
|
|
host["lease_time_remaining"] = int(lease)
|
|
hosts.append(host)
|
|
return hosts
|
|
|
|
_HOST_RE = re.compile(
|
|
r"(?P<ip>[0-9]+\.[0-9]+\.[0-9]+\.[0-9]+)"
|
|
r"\s+"
|
|
r"(?P<mac>[0-9A-Fa-f]{2}(?::[0-9A-Fa-f]{2}){5})"
|
|
r"\s+"
|
|
r"(?P<hostname>\S+)?",
|
|
)
|
|
|
|
def _parse_hosts_html(self, html: str) -> list[HostDict]:
|
|
hosts: list[HostDict] = []
|
|
for match in self._HOST_RE.finditer(html):
|
|
hosts.append(
|
|
{
|
|
"mac": match.group("mac").lower(),
|
|
"ip": match.group("ip"),
|
|
"hostname": match.group("hostname") or "",
|
|
"interface_type": "lan",
|
|
"is_active": True,
|
|
}
|
|
)
|
|
return hosts
|
|
|
|
def get_nat_translations(self) -> list[NATTranslationDict]:
|
|
"""Return NAT translations derived from enabled port forwards."""
|
|
wan_ip = (self._device_info or {}).get("WanIp", "")
|
|
translations: list[NATTranslationDict] = []
|
|
for forward in self.get_port_forwards():
|
|
if not forward["enabled"]:
|
|
continue
|
|
translations.append(
|
|
{
|
|
"protocol": forward["protocol"].lower(),
|
|
"inside_local": f"{forward['internal_ip']}:{forward['internal_port']}",
|
|
"inside_global": f"{wan_ip}:{forward['external_port']}",
|
|
"outside_local": "0.0.0.0:0",
|
|
"outside_global": "0.0.0.0:0",
|
|
"age": 0.0,
|
|
}
|
|
)
|
|
return translations
|
|
|
|
def get_vpn_tunnels(self) -> dict[str, VPNTunnelDict]:
|
|
"""Not supported on Zyxel VMG."""
|
|
return {}
|
|
|
|
def get_wireless_clients(self) -> list[WirelessClientDict]:
|
|
"""Return connected wireless clients from DAL or HTML fallback."""
|
|
html = self._fetch_page("cgi-bin/DAL?oid=WIFI_CLIENTS")
|
|
if html is not None:
|
|
try:
|
|
data = json.loads(html)
|
|
return self._parse_wireless_clients_dal(data)
|
|
except (json.JSONDecodeError, TypeError):
|
|
pass
|
|
html = self._fetch_page("wireless.htm")
|
|
if html is not None:
|
|
return self._parse_wireless_clients_html(html)
|
|
return []
|
|
|
|
def _parse_wireless_clients_dal(self, data: dict) -> list[WirelessClientDict]:
|
|
clients: list[WirelessClientDict] = []
|
|
entries = data.get("clients", data.get("assoc", data.get("entries", [])))
|
|
if isinstance(entries, dict):
|
|
entries = [entries]
|
|
if not isinstance(entries, list):
|
|
return clients
|
|
for entry in entries:
|
|
if not isinstance(entry, dict):
|
|
continue
|
|
client: WirelessClientDict = {
|
|
"mac": (entry.get("MACAddress", entry.get("mac", "")) or "").lower(),
|
|
"ssid": entry.get("SSID", entry.get("ssid", "")),
|
|
"radio": entry.get("Radio", entry.get("radio", "")),
|
|
"signal": int(entry.get("Signal", entry.get("signal", 0)) or 0),
|
|
"noise": int(entry.get("Noise", entry.get("noise", 0)) or 0),
|
|
"tx_rate": float(entry.get("TxRate", entry.get("tx_rate", 0)) or 0),
|
|
"rx_rate": float(entry.get("RxRate", entry.get("rx_rate", 0)) or 0),
|
|
"uptime": int(entry.get("Uptime", entry.get("uptime", 0)) or 0),
|
|
}
|
|
ip = entry.get("IPAddress", entry.get("ip", ""))
|
|
if ip:
|
|
client["ip"] = ip
|
|
clients.append(client)
|
|
return clients
|
|
|
|
_WIFI_CLIENT_RE = re.compile(
|
|
r"(?P<mac>[0-9A-Fa-f]{2}(?::[0-9A-Fa-f]{2}){5})"
|
|
r"\s+"
|
|
r"(?P<ip>[0-9]+\.[0-9]+\.[0-9]+\.[0-9]+)"
|
|
r"\s+"
|
|
r"(?P<rssi>-?\d+)",
|
|
)
|
|
|
|
def _parse_wireless_clients_html(self, html: str) -> list[WirelessClientDict]:
|
|
clients: list[WirelessClientDict] = []
|
|
for match in self._WIFI_CLIENT_RE.finditer(html):
|
|
clients.append(
|
|
{
|
|
"mac": match.group("mac").lower(),
|
|
"ssid": "",
|
|
"radio": "wlan0",
|
|
"signal": int(match.group("rssi")),
|
|
"noise": 0,
|
|
"tx_rate": 0.0,
|
|
"rx_rate": 0.0,
|
|
"uptime": 0,
|
|
"ip": match.group("ip"),
|
|
}
|
|
)
|
|
return clients
|
|
|
|
def get_ssids(self) -> dict[str, SSIDDict]:
|
|
"""Return WiFi SSID configurations from DAL or HTML fallback."""
|
|
html = self._fetch_page("cgi-bin/DAL?oid=WIFI_SSIDS")
|
|
if html is not None:
|
|
try:
|
|
data = json.loads(html)
|
|
return self._parse_ssids_dal(data)
|
|
except (json.JSONDecodeError, TypeError):
|
|
pass
|
|
html = self._fetch_page("wireless.htm")
|
|
if html is not None:
|
|
return self._parse_ssids_html(html)
|
|
return {}
|
|
|
|
def _parse_ssids_dal(self, data: dict) -> dict[str, SSIDDict]:
|
|
ssids: dict[str, SSIDDict] = {}
|
|
entries = data.get("ssids", data.get("entries", []))
|
|
if isinstance(entries, dict):
|
|
entries = [entries]
|
|
if not isinstance(entries, list):
|
|
return ssids
|
|
for entry in entries:
|
|
if not isinstance(entry, dict):
|
|
continue
|
|
ssid_name = (entry.get("SSID", entry.get("ssid", "")) or "").strip()
|
|
if not ssid_name:
|
|
continue
|
|
ssids[ssid_name] = {
|
|
"enabled": bool(entry.get("Enabled", entry.get("enabled", True))),
|
|
"radio": entry.get("Radio", entry.get("radio", "")),
|
|
"bssid": (entry.get("BSSID", entry.get("bssid", "")) or "").lower(),
|
|
"encryption": entry.get("Encryption", entry.get("encryption", "")),
|
|
"hidden": bool(entry.get("Hidden", entry.get("hidden", False))),
|
|
"clients": int(entry.get("Clients", entry.get("clients", 0)) or 0),
|
|
}
|
|
return ssids
|
|
|
|
def _parse_ssids_html(self, html: str) -> dict[str, SSIDDict]:
|
|
ssids: dict[str, SSIDDict] = {}
|
|
pattern = re.compile(r"SSID[^:]*:\s*(\S+)", re.I)
|
|
for match in pattern.finditer(html):
|
|
name = match.group(1)
|
|
if name and name not in ssids:
|
|
ssids[name] = {
|
|
"enabled": True,
|
|
"radio": "wlan0",
|
|
"bssid": "",
|
|
"encryption": "",
|
|
"hidden": False,
|
|
"clients": 0,
|
|
}
|
|
return ssids
|
|
|
|
def get_radio_status(self) -> dict[str, RadioStatusDict]:
|
|
"""Return WiFi radio status from DAL or HTML fallback."""
|
|
radios: dict[str, RadioStatusDict] = {}
|
|
html = self._fetch_page("cgi-bin/DAL?oid=WIFI_RADIO")
|
|
if html is not None:
|
|
try:
|
|
data = json.loads(html)
|
|
entries = data.get("radios", data.get("entries", []))
|
|
if isinstance(entries, dict):
|
|
entries = [entries]
|
|
for entry in entries if isinstance(entries, list) else []:
|
|
name = entry.get("Name", entry.get("name", ""))
|
|
channel = int(entry.get("Channel", entry.get("channel", 0)) or 0)
|
|
radios[name] = {
|
|
"enabled": bool(entry.get("Enabled", entry.get("enabled", True))),
|
|
"band": "2.4GHz" if channel <= 14 else "5GHz",
|
|
"channel": channel,
|
|
"channel_width": int(
|
|
entry.get("ChannelWidth", entry.get("channel_width", 0)) or 0
|
|
),
|
|
"tx_power": int(entry.get("TxPower", entry.get("tx_power", 0)) or 0),
|
|
"frequency": float(entry.get("Frequency", entry.get("frequency", 0)) or 0),
|
|
}
|
|
return radios
|
|
except (json.JSONDecodeError, TypeError, ValueError):
|
|
pass
|
|
html = self._fetch_page("wireless.htm")
|
|
if html is not None:
|
|
for band, pat in [("2.4GHz", r"2\.4[Gg]"), ("5GHz", r"5[Gg]")]:
|
|
if re.search(pat, html):
|
|
radios[band] = {
|
|
"enabled": True,
|
|
"band": band,
|
|
"channel": 0,
|
|
"channel_width": 0,
|
|
"tx_power": 0,
|
|
"frequency": 0.0,
|
|
}
|
|
return radios
|
|
|
|
def reboot_host(self) -> None:
|
|
"""Restart the gateway (``HostRebootMixin``); raises when it refuses."""
|
|
result = self._dal_query("REBOOT", {"Command": "reboot"})
|
|
if result is None or result.get("result") != "ZCFG_SUCCESS":
|
|
raise RuntimeError(f"the gateway refused to reboot: {result}")
|
|
|
|
def restart_wifi(self, band: str = "both") -> None:
|
|
"""Restart the radios of *band* (``2.4G``, ``5G`` or ``both``); raises when refused."""
|
|
result = self._dal_query("WIFI_RESTART", {"Band": band})
|
|
if result is None or result.get("result") != "ZCFG_SUCCESS":
|
|
raise RuntimeError(f"the gateway refused to restart Wi-Fi ({band}): {result}")
|
|
|
|
def get_diagnostics(self) -> dict:
|
|
"""What the gateway's diagnostics page reports; empty when it reports nothing."""
|
|
return self._dal_query("DIAGNOSTICS", {}) or {}
|