diff --git a/napalm_zyxel/zyxel_driver.py b/napalm_zyxel/zyxel_driver.py index 63af0d0..1d47210 100644 --- a/napalm_zyxel/zyxel_driver.py +++ b/napalm_zyxel/zyxel_driver.py @@ -11,7 +11,7 @@ import json import os import re import socket -from typing import Any, Dict, List, Optional +from typing import Any from urllib.parse import urljoin import requests @@ -22,7 +22,7 @@ from cryptography.hazmat.backends import default_backend from napalm.base.exceptions import ConnectionClosedException, ConnectionException -from napalm_device_types import ResidentialGatewayDriver +from napalm_device_types import FingerprintRule, ResidentialGatewayDriver from napalm_device_types.models import ( DeviceActionResultDict, HostDict, @@ -90,6 +90,13 @@ 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" + 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" @@ -103,7 +110,7 @@ class ZyxelVMGDriver(ResidentialGatewayDriver): username: str, password: str, timeout: int = 60, - optional_args: Optional[Dict[str, Any]] = None, + optional_args: dict[str, Any] | None = None, ) -> None: self.hostname = hostname self.username = username @@ -117,9 +124,9 @@ class ZyxelVMGDriver(ResidentialGatewayDriver): 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._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}" ) @@ -194,7 +201,7 @@ class ZyxelVMGDriver(ResidentialGatewayDriver): self._crypto = None self._device_info = None - def is_alive(self) -> Dict[str, bool]: + 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) @@ -203,7 +210,7 @@ class ZyxelVMGDriver(ResidentialGatewayDriver): except OSError: return {"is_alive": False} - def _get_basic_info(self) -> Optional[dict]: + def _get_basic_info(self) -> dict | None: if self._session is None: return None try: @@ -213,7 +220,7 @@ class ZyxelVMGDriver(ResidentialGatewayDriver): except (requests.RequestException, json.JSONDecodeError, TypeError): return None - def _get_rsa_pubkey(self) -> Optional[str]: + def _get_rsa_pubkey(self) -> str | None: if self._session is None: return None try: @@ -223,7 +230,7 @@ class ZyxelVMGDriver(ResidentialGatewayDriver): except (requests.RequestException, json.JSONDecodeError, TypeError): return None - def _dal_query(self, oid: str, data: Optional[dict] = None) -> Optional[dict]: + 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: @@ -240,7 +247,7 @@ class ZyxelVMGDriver(ResidentialGatewayDriver): except Exception: return None - def _fetch_page(self, path: str) -> Optional[str]: + def _fetch_page(self, path: str) -> str | None: if self._session is None or self._sessionkey is None: raise ConnectionClosedException("Not connected") try: @@ -253,7 +260,7 @@ class ZyxelVMGDriver(ResidentialGatewayDriver): except requests.RequestException: return None - def get_facts(self) -> Dict[str, Any]: + def get_facts(self) -> dict[str, Any]: """Return device facts such as model, serial, OS version, and uptime.""" info = self._device_info or {} try: @@ -271,9 +278,9 @@ class ZyxelVMGDriver(ResidentialGatewayDriver): "interface_list": iface_list, } - def get_interfaces(self) -> Dict[str, Dict[str, Any]]: + def get_interfaces(self) -> dict[str, dict[str, Any]]: """Return WAN, LAN, and WiFi interfaces from device info.""" - ifaces: Dict[str, Dict[str, Any]] = {} + ifaces: dict[str, dict[str, Any]] = {} info = self._device_info or {} wan_ip = info.get("WanIp", "") mac = info.get("MacAddress", "") @@ -308,20 +315,20 @@ class ZyxelVMGDriver(ResidentialGatewayDriver): } return ifaces - def get_interfaces_ip(self) -> Dict[str, Dict[str, Any]]: + 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]] = {} + 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]]: + 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]] = [] + 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 @@ -333,11 +340,11 @@ class ZyxelVMGDriver(ResidentialGatewayDriver): }) return arp - def get_lldp_neighbors(self) -> Dict[str, Any]: + 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]: + def get_lldp_neighbors_detail(self, interface: str = "") -> dict[str, Any]: """Not supported on Zyxel VMG.""" return {} @@ -347,27 +354,27 @@ class ZyxelVMGDriver(ResidentialGatewayDriver): full: bool = False, sanitized: bool = False, format: str = "text", - ) -> Dict[str, Any]: + ) -> dict[str, Any]: """Not supported on Zyxel VMG.""" return {} - def get_environment(self) -> Dict[str, Any]: + def get_environment(self) -> dict[str, Any]: """Not supported on Zyxel VMG.""" return {} - def get_optics(self) -> Dict[str, Any]: + def get_optics(self) -> dict[str, Any]: """Not supported on Zyxel VMG.""" return {} - def get_ntp_peers(self) -> Dict[str, Optional[str]]: + def get_ntp_peers(self) -> dict[str, str | None]: """Not supported on Zyxel VMG.""" return {} - def get_ntp_servers(self) -> Dict[str, Optional[str]]: + def get_ntp_servers(self) -> dict[str, str | None]: """Not supported on Zyxel VMG.""" return {} - def get_ntp_stats(self) -> List[Dict[str, Any]]: + def get_ntp_stats(self) -> list[dict[str, Any]]: """Not supported on Zyxel VMG.""" return [] @@ -381,7 +388,7 @@ class ZyxelVMGDriver(ResidentialGatewayDriver): count: int = 5, vrf: str = "", source_interface: str = "", - ) -> Dict[str, Any]: + ) -> dict[str, Any]: """Ping a destination via the device's DAL diagnostics.""" result = self._dal_query("PINGTEST", { "ProtocolVersion": "IPv4", @@ -409,7 +416,7 @@ class ZyxelVMGDriver(ResidentialGatewayDriver): ttl: int = 255, timeout: int = 2, vrf: str = "", - ) -> Dict[str, Any]: + ) -> dict[str, Any]: """Not supported on Zyxel VMG.""" return {"success": {}, "error": "Not supported on Zyxel VMG"} @@ -418,55 +425,55 @@ class ZyxelVMGDriver(ResidentialGatewayDriver): destination: str = "", protocol: str = "", longer: bool = False, - ) -> Dict[str, Any]: + ) -> dict[str, Any]: """Not supported on Zyxel VMG.""" return {} - def get_snmp_information(self) -> Dict[str, Any]: + def get_snmp_information(self) -> dict[str, Any]: """Not supported on Zyxel VMG.""" return {} - def get_users(self) -> Dict[str, Any]: + def get_users(self) -> dict[str, Any]: """Not supported on Zyxel VMG.""" return {} - def get_network_instances(self, name: str = "") -> Dict[str, Any]: + 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]: + 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]: + 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]: + 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]: + def get_probes_config(self) -> dict[str, Any]: """Not supported on Zyxel VMG.""" return {} - def get_probes_results(self) -> Dict[str, Any]: + def get_probes_results(self) -> dict[str, Any]: """Not supported on Zyxel VMG.""" return {} - def get_mac_address_table(self) -> List[Dict[str, Any]]: + def get_mac_address_table(self) -> list[dict[str, Any]]: """Not supported on Zyxel VMG.""" return [] - def get_firewall_policies(self) -> Dict[str, Any]: + 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]: + 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: + def commit_config(self, message: str = "", revert_in: int | None = None) -> None: """Not supported on Zyxel VMG.""" pass @@ -475,13 +482,13 @@ class ZyxelVMGDriver(ResidentialGatewayDriver): pass def load_replace_candidate( - self, filename: Optional[str] = None, config: Optional[str] = None + 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: Optional[str] = None, config: Optional[str] = None + 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") @@ -510,7 +517,7 @@ class ZyxelVMGDriver(ResidentialGatewayDriver): "link_status": "Up" if wan_ip else "Down", } - def get_port_forwards(self) -> List[PortForwardDict]: + 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: @@ -524,8 +531,8 @@ class ZyxelVMGDriver(ResidentialGatewayDriver): return self._parse_port_forwards_html(html) return [] - def _parse_port_forwards_dal(self, data: dict) -> List[PortForwardDict]: - forwards: List[PortForwardDict] = [] + 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] @@ -552,8 +559,8 @@ class ZyxelVMGDriver(ResidentialGatewayDriver): re.I, ) - def _parse_port_forwards_html(self, html: str) -> List[PortForwardDict]: - forwards: List[PortForwardDict] = [] + 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(), @@ -565,7 +572,7 @@ class ZyxelVMGDriver(ResidentialGatewayDriver): }) return forwards - def get_hosts(self) -> List[HostDict]: + 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: @@ -579,8 +586,8 @@ class ZyxelVMGDriver(ResidentialGatewayDriver): return self._parse_hosts_html(html) return [] - def _parse_hosts_dal(self, data: dict) -> List[HostDict]: - hosts: List[HostDict] = [] + 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] @@ -610,8 +617,8 @@ class ZyxelVMGDriver(ResidentialGatewayDriver): r"(?P\S+)?", ) - def _parse_hosts_html(self, html: str) -> List[HostDict]: - hosts: List[HostDict] = [] + 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(), @@ -622,10 +629,10 @@ class ZyxelVMGDriver(ResidentialGatewayDriver): }) return hosts - def get_nat_translations(self) -> List[NATTranslationDict]: + 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] = [] + translations: list[NATTranslationDict] = [] for forward in self.get_port_forwards(): if not forward["enabled"]: continue @@ -639,11 +646,11 @@ class ZyxelVMGDriver(ResidentialGatewayDriver): }) return translations - def get_vpn_tunnels(self) -> Dict[str, VPNTunnelDict]: + def get_vpn_tunnels(self) -> dict[str, VPNTunnelDict]: """Not supported on Zyxel VMG.""" return {} - def get_wireless_clients(self) -> List[WirelessClientDict]: + 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: @@ -657,8 +664,8 @@ class ZyxelVMGDriver(ResidentialGatewayDriver): return self._parse_wireless_clients_html(html) return [] - def _parse_wireless_clients_dal(self, data: dict) -> List[WirelessClientDict]: - clients: List[WirelessClientDict] = [] + 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] @@ -691,8 +698,8 @@ class ZyxelVMGDriver(ResidentialGatewayDriver): r"(?P-?\d+)", ) - def _parse_wireless_clients_html(self, html: str) -> List[WirelessClientDict]: - clients: List[WirelessClientDict] = [] + 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(), @@ -707,7 +714,7 @@ class ZyxelVMGDriver(ResidentialGatewayDriver): }) return clients - def get_ssids(self) -> Dict[str, SSIDDict]: + 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: @@ -721,8 +728,8 @@ class ZyxelVMGDriver(ResidentialGatewayDriver): return self._parse_ssids_html(html) return {} - def _parse_ssids_dal(self, data: dict) -> Dict[str, SSIDDict]: - ssids: Dict[str, SSIDDict] = {} + 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] @@ -744,8 +751,8 @@ class ZyxelVMGDriver(ResidentialGatewayDriver): } return ssids - def _parse_ssids_html(self, html: str) -> Dict[str, SSIDDict]: - ssids: Dict[str, SSIDDict] = {} + 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) @@ -760,9 +767,9 @@ class ZyxelVMGDriver(ResidentialGatewayDriver): } return ssids - def get_radio_status(self) -> Dict[str, RadioStatusDict]: + def get_radio_status(self) -> dict[str, RadioStatusDict]: """Return WiFi radio status from DAL or HTML fallback.""" - radios: Dict[str, RadioStatusDict] = {} + radios: dict[str, RadioStatusDict] = {} html = self._fetch_page("cgi-bin/DAL?oid=WIFI_RADIO") if html is not None: try: