NAPALM/napalm-zyxel
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: Fingerprint-Attribute für Discovery-Scoring

Browse Source 
Ergänzt DRIVER_NAME, HTTP_FINGERPRINT, SNMP_FINGERPRINT, SSH_FINGERPRINT,
PORT_SPECS und SNMP_OBJECT_ID_PREFIX gemäß docs/DISCOVERY_FINGERPRINTING.md.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Christian Manivong
2026-06-24 14:47:08 +02:00
co-authored by Claude Sonnet 4.6
parent 13e1aba5aa
commit 23e68ecd87
1 changed files with 75 additions and 68 deletions
Download Patch File Download Diff File Expand all files Collapse all files
napalm_zyxel/zyxel_driver.py
+75 -68
View File
@@ -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<hostname>\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<rssi>-?\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: