commit 13e1aba5aa2d235894d052f3001f354bc04707db Author: Christian Manivong Date: Sat Jun 13 03:18:28 2026 +0200 init diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..64c75f7 --- /dev/null +++ b/.gitignore @@ -0,0 +1,4 @@ +.venv/ +__pycache__/ +*.egg-info/ +.pytest_cache/ diff --git a/napalm_zyxel/__init__.py b/napalm_zyxel/__init__.py new file mode 100644 index 0000000..b53b928 --- /dev/null +++ b/napalm_zyxel/__init__.py @@ -0,0 +1,3 @@ +from napalm_zyxel.zyxel_driver import ZyxelVMGDriver + +__all__ = ["ZyxelVMGDriver"] diff --git a/napalm_zyxel/zyxel_driver.py b/napalm_zyxel/zyxel_driver.py new file mode 100644 index 0000000..63af0d0 --- /dev/null +++ b/napalm_zyxel/zyxel_driver.py @@ -0,0 +1,839 @@ +"""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, Dict, List, Optional +from urllib.parse import urljoin + +import requests +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric import padding as asym_padding, rsa +from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes +from cryptography.hazmat.backends import default_backend + +from napalm.base.exceptions import ConnectionClosedException, ConnectionException + +from napalm_device_types import ResidentialGatewayDriver +from napalm_device_types.models import ( + DeviceActionResultDict, + 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" + + _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: Optional[Dict[str, Any]] = 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: Optional[requests.Session] = None + self._sessionkey: Optional[str] = None + self._crypto: Optional[ZyxelCrypto] = 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, + "SHA512_password": False, + } + + 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}") + + 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) -> Optional[dict]: + 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) -> Optional[str]: + 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: Optional[dict] = None) -> Optional[dict]: + 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) -> Optional[str]: + 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, Optional[str]]: + """Not supported on Zyxel VMG.""" + return {} + + def get_ntp_servers(self) -> Dict[str, Optional[str]]: + """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: Optional[int] = 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: Optional[str] = None, config: Optional[str] = None + ) -> None: + """Not supported on Zyxel VMG.""" + raise NotImplementedError("Zyxel VMG does not support config replacement via NAPALM") + + def load_merge_candidate( + self, filename: Optional[str] = None, config: Optional[str] = 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[^|]+?)\s*\|\s*(?PTCP|UDP|BOTH)\s*\|\s*" + r"(?P\d+)\s*\|\s*(?P[0-9.]+)\s*\|\s*" + r"(?P\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[0-9]+\.[0-9]+\.[0-9]+\.[0-9]+)" + r"\s+" + r"(?P[0-9A-Fa-f]{2}(?::[0-9A-Fa-f]{2}){5})" + r"\s+" + r"(?P\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[0-9A-Fa-f]{2}(?::[0-9A-Fa-f]{2}){5})" + r"\s+" + r"(?P[0-9]+\.[0-9]+\.[0-9]+\.[0-9]+)" + r"\s+" + r"(?P-?\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(self) -> DeviceActionResultDict: + """Reboot the device via DAL.""" + result = self._dal_query("REBOOT", {"Command": "reboot"}) + if result is not None and result.get("result") == "ZCFG_SUCCESS": + return {"success": True, "action": "reboot", "message": "Device is rebooting"} + return {"success": False, "action": "reboot", "message": "Reboot command failed"} + + def run_device_action(self, action: str) -> DeviceActionResultDict: + """Run a device action such as reboot or WiFi restart.""" + result: DeviceActionResultDict = { + "success": False, + "action": action, + "message": "", + } + try: + if action == "reboot": + return self.reboot() + elif action.startswith("restart_wifi"): + band = "both" + if ":" in action: + band = action.split(":", 1)[1] + dal = self._dal_query("WIFI_RESTART", {"Band": band}) + if dal is not None and dal.get("result") == "ZCFG_SUCCESS": + result["success"] = True + result["message"] = f"WiFi restarting ({band})" + else: + result["message"] = "WiFi restart command failed" + elif action == "get_diagnostics": + diag = self._dal_query("DIAGNOSTICS", {}) + if diag: + result["success"] = True + result["message"] = json.dumps(diag) + else: + result["message"] = "Diagnostics not available" + else: + result["message"] = f"Unknown action: {action}" + except Exception as e: + result["message"] = f"Action '{action}' failed: {e}" + return result diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..fcc6099 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,55 @@ +[build-system] +requires = ["setuptools>=68", "wheel"] +build-backend = "setuptools.build_meta" + +[project] +name = "napalm-zyxel" +version = "0.1.0" +description = "NAPALM driver for Zyxel VMG series residential gateways (VMG4005-B50A and similar)." +readme = "README.md" +license = { text = "Apache-2.0" } +requires-python = ">=3.9" +authors = [ + { name = "Christian Manivong" }, +] +classifiers = [ + "Topic :: Utilities", + "License :: OSI Approved :: Apache Software License", + "Programming Language :: Python :: 3", + "Programming Language :: Python :: 3.9", + "Programming Language :: Python :: 3.10", + "Programming Language :: Python :: 3.11", + "Programming Language :: Python :: 3.12", + "Operating System :: POSIX :: Linux", + "Operating System :: MacOS", +] +dependencies = [ + "napalm>=4.0.0", + "napalm_device_types>=0.4.0", + "requests>=2.28.0", +] + +[project.optional-dependencies] +dev = [ + "pytest", + "pytest-cov", + "black", + "ruff", +] + +[project.entry-points."napalm.drivers"] +zyxel = "napalm_zyxel.zyxel_driver:ZyxelVMGDriver" + +[project.urls] +Repository = "https://github.com/napalm-automation-community/napalm-zyxel" + +[tool.setuptools.packages.find] +where = ["."] +include = ["napalm_zyxel*"] + +[tool.ruff] +line-length = 100 +target-version = "py39" + +[tool.pytest.ini_options] +testpaths = ["tests"] diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/test_zyxel.py b/tests/test_zyxel.py new file mode 100644 index 0000000..a937137 --- /dev/null +++ b/tests/test_zyxel.py @@ -0,0 +1,421 @@ +import json +from unittest.mock import MagicMock, patch + +import requests + +from napalm.base.exceptions import ConnectionException + +from napalm_zyxel import ZyxelVMGDriver + + +MOCK_BASIC_INFO = { + "ModelName": "VMG4005-B50A", + "SoftwareVersion": "V5.13(ABKS.1)C0", + "SerialNumber": "S220L12345678", + "HostName": "ZyxelVMG", + "WanIp": "85.2.34.56", + "LanIp": "192.168.1.1", + "MacAddress": "00:1A:2B:3C:4D:5E", + "SystemUpTime": "1234567", + "WanAccessType": "VDSL", + "MaxUpstream": "30000000", + "MaxDownstream": "100000000", +} + +MOCK_RSA_KEY = ( + "-----BEGIN PUBLIC KEY-----\n" + "MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAw8kFmWQhFfTfL8eP\n" + "yGsqQw5g3n4R7pG9vLxKZJtRq0pQ+kCqWzXHmTZvLxKZJtRq0pQ+kCqWzX\n" + "HmTZvLxKZJtRq0pQ+kCqWzXHmTZvLxKZJtRq0pQ+kCqWzXHmTZvLxKZJtR\n" + "q0pQ+kCqWzXHmTZvLxKZJtRq0pQ+kCqWzXHmTZvLxKZJtRq0pQ+kCqWzXH\n" + "mTZvLxKZJtRq0pQ+kCqWzXHmTZvLxKZJtRq0pQ+kCqWzXHmTZvLxKZJtRq\n" + "0pQ+kCqWzXHmTZvLxKZJtRq0pQ+kCqWzXHmTZvLxKZJtRq0pQ+kCqWzXHm\n" + "TZvLxKZJtRq0pQ+kCqWzXHmTZvLxKZJtRq0pQ+kCqWzXHmTZvLxKZJtRq0\n" + "pQIDAQAB\n" + "-----END PUBLIC KEY-----\n" +) + +MOCK_LOGIN_RESPONSE = { + "content": "", + "iv": "QkJCQkJCQkJCQkJCQkJCQg==", + "key": "", +} + + +class MockResponse: + def __init__(self, json_data, status_code=200, text=""): + self._json = json_data + self.status_code = status_code + self._text = text or json.dumps(json_data) + + def json(self): + return self._json + + @property + def text(self): + return self._text + + def raise_for_status(self): + if self.status_code >= 400: + raise requests.HTTPError(f"HTTP {self.status_code}") + + +class TestZyxelVMGDriver: + def setup_method(self): + self.driver = ZyxelVMGDriver( + "192.168.1.1", + "admin", + "password", + ) + + def _setup_mocked_http(self, monkeypatch_crypto=True): + session = MagicMock(spec=requests.Session) + session.verify = False + session.timeout = 60 + session.trust_env = False + self.driver._session = session + self.driver._device_info = MOCK_BASIC_INFO + + def mock_get(url, **kwargs): + if "getBasicInformation" in url: + return MockResponse(MOCK_BASIC_INFO) + if "getRSAPublickKey" in url: + return MockResponse({"RSAPublicKey": MOCK_RSA_KEY}) + if "UserLogout" in url: + return MockResponse({}) + return MockResponse({}) + + def mock_post(url, **kwargs): + if "UserLogin" in url: + return MockResponse(MOCK_LOGIN_RESPONSE) + return MockResponse({}) + + session.get.side_effect = mock_get + session.post.side_effect = mock_post + + if monkeypatch_crypto: + crypto = MagicMock() + crypto.encrypt_json.return_value = {"content": "ENC", "iv": "IV", "key": ""} + crypto.decrypt_response.return_value = {"result": "ZCFG_SUCCESS", "sessionkey": "12345"} + crypto.encrypt_aes_key_with_rsa.return_value = "ENC_KEY" + crypto.aes_key = b"\x42" * 32 + self.driver._crypto = crypto + + self.driver._sessionkey = "12345" + + def test_init(self): + assert self.driver.hostname == "192.168.1.1" + assert self.driver.username == "admin" + assert self.driver.password == "password" + assert self.driver.port == 80 + assert self.driver.use_tls is False + assert self.driver.VENDOR == "Zyxel" + + def test_open_success(self): + with patch("napalm_zyxel.zyxel_driver.ZyxelCrypto") as mock_crypto_cls: + mock_crypto = MagicMock() + mock_crypto.encrypt_json.return_value = {"content": "ENC", "iv": "IV", "key": ""} + mock_crypto.decrypt_response.return_value = { + "result": "ZCFG_SUCCESS", "sessionkey": "12345" + } + mock_crypto.encrypt_aes_key_with_rsa.return_value = "ENC_KEY" + mock_crypto_cls.return_value = mock_crypto + + session = MagicMock() + session.get.return_value = MockResponse(MOCK_BASIC_INFO) + session.post.return_value = MockResponse(MOCK_LOGIN_RESPONSE) + + with patch.object(self.driver, "_session", session, create=False): + with patch.object( + self.driver, "_get_rsa_pubkey", return_value=MOCK_RSA_KEY + ): + with patch.object( + self.driver, "_get_basic_info", return_value=MOCK_BASIC_INFO + ): + self.driver._session = None + self.driver._crypto = None + self.driver._sessionkey = None + with patch("requests.Session", return_value=session): + self.driver.open() + assert self.driver._sessionkey == "12345" + assert self.driver._device_info == MOCK_BASIC_INFO + + def test_open_failure_unreachable(self): + with patch("napalm_zyxel.zyxel_driver.ZyxelCrypto") as mock_crypto_cls: + mock_crypto = MagicMock() + mock_crypto_cls.return_value = mock_crypto + self.driver._session = None + + with patch("requests.Session") as mock_session: + sess = MagicMock() + sess.get.side_effect = requests.ConnectionError("unreachable") + mock_session.return_value = sess + try: + self.driver.open() + assert False, "Expected ConnectionException" + except ConnectionException: + pass + + def test_close(self): + self._setup_mocked_http() + self.driver.close() + assert self.driver._session is None + assert self.driver._sessionkey is None + + def test_is_alive(self): + with patch("socket.create_connection") as mock_conn: + mock_conn.return_value.__enter__.return_value = MagicMock() + assert self.driver.is_alive()["is_alive"] is True + + def test_is_alive_fail(self): + with patch("socket.create_connection") as mock_conn: + mock_conn.side_effect = OSError("refused") + assert self.driver.is_alive()["is_alive"] is False + + def test_get_facts(self): + self._setup_mocked_http() + facts = self.driver.get_facts() + assert facts["vendor"] == "Zyxel" + assert facts["model"] == "VMG4005-B50A" + assert facts["serial_number"] == "S220L12345678" + assert facts["os_version"] == "V5.13(ABKS.1)C0" + assert facts["hostname"] == "ZyxelVMG" + + def test_get_interfaces(self): + self._setup_mocked_http() + ifaces = self.driver.get_interfaces() + assert "wan" in ifaces + assert "lan" in ifaces + assert "wlan_2_4ghz" in ifaces + assert "wlan_5ghz" in ifaces + + def test_get_interfaces_ip(self): + self._setup_mocked_http() + ips = self.driver.get_interfaces_ip() + assert "85.2.34.56" in ips["wan"]["ipv4"] + assert "192.168.1.1" in ips["lan"]["ipv4"] + + def test_get_wan_status(self): + self._setup_mocked_http() + status = self.driver.get_wan_status() + assert status["is_connected"] is True + assert status["external_ip"] == "85.2.34.56" + assert status["connection_type"] == "VDSL" + + def test_get_port_forwards_empty(self): + self._setup_mocked_http() + self.driver._session.get.side_effect = None + + def mock_get(url, **kwargs): + if "getBasicInformation" in url: + return MockResponse(MOCK_BASIC_INFO) + return MockResponse({}, text="no forwards here") + + self.driver._session.get.side_effect = mock_get + forwards = self.driver.get_port_forwards() + assert forwards == [] + + def test_get_hosts_empty(self): + self._setup_mocked_http() + self.driver._session.get.side_effect = None + + def mock_get(url, **kwargs): + if "getBasicInformation" in url: + return MockResponse(MOCK_BASIC_INFO) + return MockResponse({}, text="no dhcp here") + + self.driver._session.get.side_effect = mock_get + hosts = self.driver.get_hosts() + assert hosts == [] + + def test_get_nat_translations_empty(self): + self._setup_mocked_http() + self.driver._session.get.side_effect = None + + def mock_get(url, **kwargs): + if "getBasicInformation" in url: + return MockResponse(MOCK_BASIC_INFO) + return MockResponse({}, text="") + + self.driver._session.get.side_effect = mock_get + nat = self.driver.get_nat_translations() + assert nat == [] + + def test_get_vpn_tunnels(self): + self._setup_mocked_http() + assert self.driver.get_vpn_tunnels() == {} + + def test_get_wireless_clients_empty(self): + self._setup_mocked_http() + self.driver._session.get.side_effect = None + + def mock_get(url, **kwargs): + if "getBasicInformation" in url: + return MockResponse(MOCK_BASIC_INFO) + return MockResponse({}, text="") + + self.driver._session.get.side_effect = mock_get + assert self.driver.get_wireless_clients() == [] + + def test_get_ssids_empty(self): + self._setup_mocked_http() + self.driver._session.get.side_effect = None + + def mock_get(url, **kwargs): + if "getBasicInformation" in url: + return MockResponse(MOCK_BASIC_INFO) + return MockResponse({}, text="") + + self.driver._session.get.side_effect = mock_get + assert self.driver.get_ssids() == {} + + def test_get_radio_status_empty(self): + self._setup_mocked_http() + self.driver._session.get.side_effect = None + + def mock_get(url, **kwargs): + if "getBasicInformation" in url: + return MockResponse(MOCK_BASIC_INFO) + return MockResponse({}, text="") + + self.driver._session.get.side_effect = mock_get + assert self.driver.get_radio_status() == {} + + def test_get_arp_table(self): + self._setup_mocked_http() + assert self.driver.get_arp_table() == [] + + def test_reboot(self): + self._setup_mocked_http() + result = self.driver.run_device_action("reboot") + assert result["action"] == "reboot" + + def test_run_device_action_unknown(self): + self._setup_mocked_http() + result = self.driver.run_device_action("foobar") + assert result["success"] is False + assert "Unknown" in result["message"] + + def test_load_replace_candidate_raises(self): + self._setup_mocked_http() + try: + self.driver.load_replace_candidate() + assert False, "Expected NotImplementedError" + except NotImplementedError: + pass + + def test_load_merge_candidate_raises(self): + self._setup_mocked_http() + try: + self.driver.load_merge_candidate() + assert False, "Expected NotImplementedError" + except NotImplementedError: + pass + + def test_ping_not_implemented(self): + self._setup_mocked_http() + result = self.driver.ping("8.8.8.8") + assert "error" in result + + def test_get_environment(self): + self._setup_mocked_http() + assert self.driver.get_environment() == {} + + def test_dal_query_not_connected(self): + from napalm.base.exceptions import ConnectionClosedException + + self.driver._session = None + try: + self.driver._dal_query("PINGTEST", {}) + assert False, "Expected ConnectionClosedException" + except ConnectionClosedException: + pass + + def test_parse_port_forwards_dal(self): + self._setup_mocked_http() + data = { + "NATPortMap": [ + { + "Description": "Web Server", + "Protocol": "TCP", + "ExternalPort": 80, + "InternalIP": "192.168.1.10", + "InternalPort": 8080, + "Enabled": True, + } + ] + } + result = self.driver._parse_port_forwards_dal(data) + assert len(result) == 1 + assert result[0]["name"] == "Web Server" + assert result[0]["external_port"] == 80 + assert result[0]["internal_ip"] == "192.168.1.10" + + def test_parse_hosts_dal(self): + self._setup_mocked_http() + data = { + "LANHosts": [ + { + "MACAddress": "aa:bb:cc:dd:ee:ff", + "IPAddress": "192.168.1.100", + "HostName": "laptop", + "InterfaceType": "lan", + "Active": True, + } + ] + } + result = self.driver._parse_hosts_dal(data) + assert len(result) == 1 + assert result[0]["mac"] == "aa:bb:cc:dd:ee:ff" + assert result[0]["ip"] == "192.168.1.100" + assert result[0]["hostname"] == "laptop" + + def test_parse_wireless_clients_dal(self): + self._setup_mocked_http() + data = { + "assoc": [ + { + "MACAddress": "11:22:33:44:55:66", + "SSID": "MyWiFi", + "Radio": "wlan0", + "Signal": -45, + "TxRate": 300.0, + "RxRate": 150.0, + "Uptime": 3600, + "IPAddress": "192.168.1.50", + } + ] + } + result = self.driver._parse_wireless_clients_dal(data) + assert len(result) == 1 + assert result[0]["mac"] == "11:22:33:44:55:66" + assert result[0]["signal"] == -45 + assert result[0]["ip"] == "192.168.1.50" + + def test_parse_ssids_dal(self): + self._setup_mocked_http() + data = { + "ssids": [ + { + "SSID": "MyWiFi", + "Radio": "wlan0", + "BSSID": "00:1a:2b:3c:4d:5e", + "Encryption": "WPA2", + "Enabled": True, + "Hidden": False, + "Clients": 3, + } + ] + } + result = self.driver._parse_ssids_dal(data) + assert "MyWiFi" in result + assert result["MyWiFi"]["encryption"] == "WPA2" + assert result["MyWiFi"]["clients"] == 3 + + def test_get_config(self): + self._setup_mocked_http() + assert self.driver.get_config() == {} + + def test_cli(self): + self._setup_mocked_http() + assert self.driver.cli(["show version"]) == {}