diff --git a/napalm_netgear/netgear_plus.py b/napalm_netgear/netgear_plus.py index d845514..54c3a83 100644 --- a/napalm_netgear/netgear_plus.py +++ b/napalm_netgear/netgear_plus.py @@ -20,11 +20,13 @@ These switches use password-only web authentication (no username). The ``username`` parameter is accepted but ignored. """ +from __future__ import annotations + import threading import time -from typing import Any, Dict, List, Optional, Tuple +from typing import Any -from napalm_device_types import SwitchDriver +from napalm_device_types import FingerprintRule, SwitchDriver from napalm.base.exceptions import ConnectionException # --------------------------------------------------------------------------- @@ -38,7 +40,7 @@ _FW_CDN_BASE = "https://www.downloads.netgear.com/files/GDC" # Known-latest firmware versions per CDN model ID. # Verified via HEAD requests against downloads.netgear.com on 2026-05-24. # Update this table when Netgear releases new firmware. -_FW_KNOWN_LATEST: Dict[str, str] = { +_FW_KNOWN_LATEST: dict[str, str] = { "GS305EP": "V2.0.0.10", "GS305EPP": "V2.0.0.10", "GS308EP": "V2.0.0.11", @@ -46,14 +48,14 @@ _FW_KNOWN_LATEST: Dict[str, str] = { } # In-memory cache: model_id -> (latest_version, unix_timestamp_of_check) -_fw_cache: Dict[str, Tuple[str, float]] = {} +_fw_cache: dict[str, tuple[str, float]] = {} _fw_cache_lock = threading.Lock() # How long (seconds) to reuse a cached firmware-version result (24 hours). _FW_CACHE_TTL = 86400 -def _fw_version_tuple(version: str) -> Tuple[int, ...]: +def _fw_version_tuple(version: str) -> tuple[int, ...]: """Parse 'V1.0.1.4' or '1.0.1.4' into (1, 0, 1, 4). Raises ValueError on bad input.""" v = version.strip().lstrip("Vv") return tuple(int(x) for x in v.split(".")) @@ -94,7 +96,7 @@ def _fw_probe_newer(model: str, baseline: str) -> str: best_t = base_t best_v = baseline - def _try(candidate: Tuple[int, ...]) -> None: + def _try(candidate: tuple[int, ...]) -> None: nonlocal best_t, best_v v_str = "V" + ".".join(str(x) for x in candidate) if _fw_cdn_exists(model, v_str) and candidate > best_t: @@ -114,7 +116,7 @@ def _fw_probe_newer(model: str, baseline: str) -> str: return best_v -def _fw_get_latest(model: str) -> Optional[str]: +def _fw_get_latest(model: str) -> str | None: """Return the latest known firmware version for *model*, using a 24-hour in-memory cache. Returns ``None`` when the model is not in the known table. """ @@ -146,6 +148,18 @@ class NetgearPlusDriver(SwitchDriver): """HTTP-based NAPALM driver for Netgear Plus Smart Managed switches.""" VENDOR = "Netgear" + DRIVER_NAME = "netgear_plus" + SNMP_OBJECT_ID_PREFIX = "1.3.6.1.4.1.4526" + HTTP_FINGERPRINT = [ + FingerprintRule("netgear gs1", weight=8.0), + FingerprintRule("netgear gs2", weight=8.0), + FingerprintRule("netgear gs3", weight=8.0), + FingerprintRule("netgear ms", weight=8.0), + FingerprintRule("netgear", weight=3.0), + ] + SNMP_FINGERPRINT = [ + FingerprintRule("netgear", weight=4.0), + ] SNMP_CAPABLE = False # HTTP-only device — no SNMP daemon def __init__( @@ -154,14 +168,14 @@ class NetgearPlusDriver(SwitchDriver): username: str, password: str, timeout: int = 60, - optional_args: Optional[Dict] = None, + optional_args: dict | None = None, ) -> None: # username is accepted but ignored — these switches have no username self.hostname = hostname self.password = password self.timeout = timeout self._connector = None - self._switch_infos_cache: Optional[Dict[str, Any]] = None + self._switch_infos_cache: dict[str, Any] | None = None if optional_args is None: optional_args = {} @@ -206,20 +220,20 @@ class NetgearPlusDriver(SwitchDriver): self._connector = None self._switch_infos_cache = None - def is_alive(self) -> Dict[str, bool]: + def is_alive(self) -> dict[str, bool]: return {"is_alive": self._connector is not None} # ------------------------------------------------------------------ # Internal helpers # ------------------------------------------------------------------ - def _get_switch_infos(self) -> Dict[str, Any]: + def _get_switch_infos(self) -> dict[str, Any]: """Fetch and cache switch infos for this poll cycle.""" if self._switch_infos_cache is None: self._switch_infos_cache = self._connector.get_switch_infos() return self._switch_infos_cache - def _get_port_names(self) -> Dict[int, str]: + def _get_port_names(self) -> dict[int, str]: """Fetch port names/descriptions from dashboard.cgi. The GS30x series embeds ```` @@ -263,7 +277,7 @@ class NetgearPlusDriver(SwitchDriver): # NAPALM getters # ------------------------------------------------------------------ - def get_facts(self) -> Dict[str, Any]: + def get_facts(self) -> dict[str, Any]: """Return a dictionary of general device facts.""" infos = self._get_switch_infos() model = getattr(self._connector.switch_model, "MODEL_NAME", "") @@ -284,7 +298,7 @@ class NetgearPlusDriver(SwitchDriver): "interface_list": interface_list, } - def get_interfaces(self) -> Dict[str, Dict]: + def get_interfaces(self) -> dict[str, dict]: """Return a dictionary of interface details. Each interface entry contains: @@ -293,7 +307,7 @@ class NetgearPlusDriver(SwitchDriver): """ infos = self._get_switch_infos() port_names = self._get_port_names() - interfaces: Dict[str, Dict] = {} + interfaces: dict[str, dict] = {} for port in range(1, self._connector.ports + 1): name = self._port_name(port) status = infos.get(f"port_{port}_status", "off") == "on" @@ -310,7 +324,7 @@ class NetgearPlusDriver(SwitchDriver): } return interfaces - def get_interfaces_ip(self) -> Dict[str, Dict]: + def get_interfaces_ip(self) -> dict[str, dict]: """Return IP addresses per interface. Netgear Plus switches are L2 only — no routed interfaces. @@ -318,14 +332,14 @@ class NetgearPlusDriver(SwitchDriver): """ return {} - def get_interfaces_counters(self) -> Dict[str, Dict]: + def get_interfaces_counters(self) -> dict[str, dict]: """Return interface traffic counters. Uses cumulative RX/TX totals (``sum_rx_mbytes`` / ``sum_tx_mbytes``) from ``get_switch_infos()`` converted to octets, plus CRC errors. """ infos = self._get_switch_infos() - counters: Dict[str, Dict] = {} + counters: dict[str, dict] = {} for port in range(1, self._connector.ports + 1): name = self._port_name(port) # sum_*_mbytes are lifetime totals in megabytes; convert to bytes @@ -348,7 +362,7 @@ class NetgearPlusDriver(SwitchDriver): } return counters - def get_environment(self) -> Dict[str, Any]: + def get_environment(self) -> dict[str, Any]: """Return environment data. Reports PoE power status per PoE port. Temperature, fans, CPU and @@ -356,7 +370,7 @@ class NetgearPlusDriver(SwitchDriver): dicts. """ infos = self._get_switch_infos() - power: Dict[str, Any] = {} + power: dict[str, Any] = {} poe_max_single = getattr( self._connector.switch_model, "POE_MAX_POWER_SINGLE_PORT", None ) @@ -376,7 +390,7 @@ class NetgearPlusDriver(SwitchDriver): "memory": {}, } - def get_mac_address_table(self) -> List[Dict]: + def get_mac_address_table(self) -> list[dict]: """Return the MAC address table. Not available via the HTTP web interface of Plus switches. @@ -384,7 +398,7 @@ class NetgearPlusDriver(SwitchDriver): """ return [] - def get_arp_table(self, vrf: str = "") -> List[Dict]: + def get_arp_table(self, vrf: str = "") -> list[dict]: """Return the ARP table. Netgear Plus switches are L2 only and have no ARP table. @@ -392,7 +406,7 @@ class NetgearPlusDriver(SwitchDriver): """ return [] - def get_lldp_neighbors(self) -> Dict[str, Any]: + def get_lldp_neighbors(self) -> dict[str, Any]: """Return LLDP neighbors. Not supported on Netgear Plus switches. @@ -400,7 +414,7 @@ class NetgearPlusDriver(SwitchDriver): """ return {} - def get_lldp_neighbors_detail(self, interface: str = "") -> Dict[str, Any]: + def get_lldp_neighbors_detail(self, interface: str = "") -> dict[str, Any]: """Return detailed LLDP neighbor info. Not supported on Netgear Plus switches. @@ -408,7 +422,7 @@ class NetgearPlusDriver(SwitchDriver): """ return {} - def get_vlans_detail(self) -> Dict[str, Any]: + def get_vlans_detail(self) -> dict[str, Any]: """Return VLAN configuration by scraping the switch's /vlan.cgi page. Returns a dict keyed by VLAN ID string: @@ -433,7 +447,7 @@ class NetgearPlusDriver(SwitchDriver): # Parse VLAN IDs and names from vid-4 / vnm-4 span pairs. # Example: ...24Guest - vlan_names: Dict[str, str] = {} + vlan_names: dict[str, str] = {} vlan_entry_re = re.compile( r']*>.*?(\d+)\s*\s*' r']*>([^<]*)', @@ -447,7 +461,7 @@ class NetgearPlusDriver(SwitchDriver): if not vlan_names: return {} - vlans: Dict[str, Any] = { + vlans: dict[str, Any] = { vid: {"name": name, "tagged": [], "untagged": []} for vid, name in vlan_names.items() } @@ -477,7 +491,7 @@ class NetgearPlusDriver(SwitchDriver): return vlans - def get_vlans(self) -> Dict[str, Any]: + def get_vlans(self) -> dict[str, Any]: """Return VLAN information (standard NAPALM format). Delegates to get_vlans_detail() and merges tagged and untagged ports @@ -492,7 +506,7 @@ class NetgearPlusDriver(SwitchDriver): for vid, info in detail.items() } - def set_vlan(self, vlan_id: int, config: Dict) -> None: + def set_vlan(self, vlan_id: int, config: dict) -> None: """Create or update a VLAN. Not supported via HTTP API. @@ -516,7 +530,7 @@ class NetgearPlusDriver(SwitchDriver): full: bool = False, sanitized: bool = False, format: str = "text", - ) -> Dict[str, str]: + ) -> dict[str, str]: """Return device configuration. Netgear Plus switches do not expose running/startup config via HTTP. @@ -524,12 +538,12 @@ class NetgearPlusDriver(SwitchDriver): """ return {"running": "", "startup": "", "candidate": ""} - def load_merge_candidate(self, filename: Optional[str] = None, config: Optional[str] = None) -> None: + def load_merge_candidate(self, filename: str | None = None, config: str | None = None) -> None: raise NotImplementedError( "Config management is not supported for Netgear Plus switches." ) - def load_replace_candidate(self, filename: Optional[str] = None, config: Optional[str] = None) -> None: + def load_replace_candidate(self, filename: str | None = None, config: str | None = None) -> None: raise NotImplementedError( "Config management is not supported for Netgear Plus switches." ) @@ -567,7 +581,7 @@ class NetgearPlusDriver(SwitchDriver): count: int = 5, vrf: str = "", source_interface: str = "", - ) -> Dict[str, Any]: + ) -> dict[str, Any]: """Execute ping. Not supported on Netgear Plus switches via HTTP. @@ -583,7 +597,7 @@ class NetgearPlusDriver(SwitchDriver): ttl: int = 255, timeout: int = 2, vrf: str = "", - ) -> Dict[str, Any]: + ) -> dict[str, Any]: """Execute traceroute. Not supported on Netgear Plus switches via HTTP. @@ -592,7 +606,7 @@ class NetgearPlusDriver(SwitchDriver): "traceroute() is not supported for Netgear Plus switches." ) - def cli(self, commands: List[str], encoding: str = "text") -> Dict[str, Any]: + def cli(self, commands: list[str], encoding: str = "text") -> dict[str, Any]: """Send CLI commands. Netgear Plus switches have no SSH CLI — only HTTP. @@ -601,35 +615,35 @@ class NetgearPlusDriver(SwitchDriver): "cli() is not supported for Netgear Plus switches (HTTP-only device)." ) - def get_users(self) -> Dict[str, Any]: + def get_users(self) -> dict[str, Any]: """Return user accounts. Not available via HTTP API. Returns an empty dict. """ return {} - def get_snmp_information(self) -> Dict[str, Any]: + def get_snmp_information(self) -> dict[str, Any]: """Return SNMP configuration. Not available via HTTP API. Returns an empty dict. """ return {} - def get_ntp_servers(self) -> Dict[str, Any]: + def get_ntp_servers(self) -> dict[str, Any]: """Return configured NTP servers. Not available via HTTP API. Returns an empty dict. """ return {} - def get_ntp_peers(self) -> Dict[str, Any]: + def get_ntp_peers(self) -> dict[str, Any]: """Return NTP peers. Not available via HTTP API. Returns an empty dict. """ return {} - def get_ntp_stats(self) -> List[Dict]: + def get_ntp_stats(self) -> list[dict]: """Return NTP stats. Not available via HTTP API. Returns an empty list. @@ -640,35 +654,35 @@ class NetgearPlusDriver(SwitchDriver): # SwitchDriver abstract methods (not applicable for this device) # ------------------------------------------------------------------ - def get_spanning_tree(self) -> Dict[str, Any]: + def get_spanning_tree(self) -> dict[str, Any]: """Return spanning tree status. Not available via HTTP API. Returns an empty dict. """ return {} - def get_port_channels(self) -> Dict[str, Any]: + def get_port_channels(self) -> dict[str, Any]: """Return port-channel information. Not available via HTTP API. Returns an empty dict. """ return {} - def get_dot1x_config(self) -> Dict[str, Any]: + def get_dot1x_config(self) -> dict[str, Any]: """Return 802.1X port configuration. Not available via HTTP API. Returns an empty dict. """ return {} - def get_mac_acl(self) -> Dict[str, Any]: + def get_mac_acl(self) -> dict[str, Any]: """Return MAC ACL information. Not available via HTTP API. Returns an empty dict. """ return {} - def get_poe_status(self) -> Dict[str, Any]: + def get_poe_status(self) -> dict[str, Any]: """Return PoE status. Reports PoE power status from switch infos. @@ -680,7 +694,7 @@ class NetgearPlusDriver(SwitchDriver): poe_max_single = getattr( self._connector.switch_model, "POE_MAX_POWER_SINGLE_PORT", None ) - ports: Dict[str, Any] = {} + ports: dict[str, Any] = {} for p in self._connector.poe_ports: poe_active = infos.get(f"port_{p}_poe_power_active") is_delivering = bool(poe_active) if poe_active is not None else False @@ -722,7 +736,7 @@ class NetgearPlusDriver(SwitchDriver): raise ValueError(f"{interface} is not a PoE port on this switch.") self._connector.power_cycle_poe_port(port_number) - def set_interface(self, interface: str, config: Dict) -> None: + def set_interface(self, interface: str, config: dict) -> None: """Configure an interface. Not supported via HTTP API. diff --git a/napalm_netgear/netgear_smart.py b/napalm_netgear/netgear_smart.py index 634ed0d..5d72bc9 100644 --- a/napalm_netgear/netgear_smart.py +++ b/napalm_netgear/netgear_smart.py @@ -27,7 +27,6 @@ CLI notes * Config is saved with ``write memory``. """ -import difflib import re import socket import time @@ -37,7 +36,7 @@ import netaddr from netmiko import ConnectHandler from netmiko.exceptions import NetmikoAuthenticationException, NetmikoTimeoutException -from napalm_device_types import SwitchDriver +from napalm_device_types import ConfigLifecycleMixin, FingerprintRule, SwitchDriver from napalm.base import helpers as napalm_helpers from napalm.base.exceptions import ( CommandErrorException, @@ -50,10 +49,23 @@ from napalm.base.netmiko_helpers import netmiko_args import napalm.base.constants as C -class NetgearSmartDriver(SwitchDriver): +class NetgearSmartDriver(ConfigLifecycleMixin, SwitchDriver): """NAPALM driver for Netgear Smart Managed switches (GS-series).""" VENDOR = "Netgear" + DRIVER_NAME = "netgear_smart" + SNMP_OBJECT_ID_PREFIX = "1.3.6.1.4.1.4526" + HTTP_FINGERPRINT = [ + FingerprintRule("netgear xsm", weight=8.0), + FingerprintRule("netgear m4", weight=8.0), + FingerprintRule("netgear gs1", weight=6.0), + FingerprintRule("netgear gs2", weight=6.0), + FingerprintRule("netgear gs3", weight=6.0), + FingerprintRule("netgear", weight=3.0), + ] + SNMP_FINGERPRINT = [ + FingerprintRule("netgear", weight=4.0), + ] # Netmiko device type – handles prompt ``(hostname) >`` / ``(hostname) #`` NETMIKO_DEVICE_TYPE = "netgear_prosafe" @@ -1224,10 +1236,16 @@ class NetgearSmartDriver(SwitchDriver): vid_s, name, untagged, tagged = parts[0], parts[1], parts[2], parts[3] if not vid_s.isdigit(): continue + if name.lower().startswith("auto-"): + # Built-in "Auto-Voice/Video/Camera/Wifi" pseudo-VLANs are + # always listed even when unused — not real user VLANs. + continue + untagged_ports = self._expand_port_list_v7(untagged) + tagged_ports = self._expand_port_list_v7(tagged) vlans[vid_s] = { "name": name, - "untagged": self._expand_port_list_v7(untagged), - "tagged": self._expand_port_list_v7(tagged), + "untagged": untagged_ports, + "tagged": tagged_ports, } return vlans @@ -1235,78 +1253,8 @@ class NetgearSmartDriver(SwitchDriver): # NAPALM configuration management # ------------------------------------------------------------------ - def load_merge_candidate( - self, filename: Optional[str] = None, config: Optional[str] = None - ) -> None: - """Stage CLI commands to be merged into the running configuration. - - *config* is a plain-text string of CLI commands as typed in config - mode (one command per line). Sub-mode entry/exit lines are supported. - Blank lines and lines starting with ``!`` or ``#`` are ignored. - - The configuration is **not** applied until :meth:`commit_config`. - - :raises MergeConfigException: on invalid input. - """ - if filename is not None: - try: - with open(filename) as fh: - config = fh.read() - except OSError as exc: - raise MergeConfigException(str(exc)) from exc - if config is None: - raise MergeConfigException("Either 'filename' or 'config' must be provided.") - self._candidate_config = config - self._candidate_mode = "merge" - - def load_replace_candidate( - self, filename: Optional[str] = None, config: Optional[str] = None - ) -> None: - """Stage a full running-config replacement candidate. - - Netgear Smart Managed does not support atomic config replace; - commit applies the candidate lines additively (same as merge). - :meth:`compare_config` still shows a full unified diff. - - :raises ReplaceConfigException: on invalid input. - """ - if filename is not None: - try: - with open(filename) as fh: - config = fh.read() - except OSError as exc: - raise ReplaceConfigException(str(exc)) from exc - if config is None: - raise ReplaceConfigException("Either 'filename' or 'config' must be provided.") - self._candidate_config = config - self._candidate_mode = "replace" - - def compare_config(self) -> str: - """Return a human-readable diff of the pending candidate vs running config. - - Merge candidate → lines prefixed with ``+``. - Replace candidate → unified diff against current running config. - Returns an empty string when no candidate is staged. - """ - if self._candidate_config is None: - return "" - - if self._candidate_mode == "merge": - lines = [] - for line in self._candidate_config.splitlines(): - if line.strip() and not line.strip().startswith(("!", "#")): - lines.append(f"+{line}") - return "\n".join(lines) - - running = self._send_command("show running-config") - diff = difflib.unified_diff( - running.splitlines(), - self._candidate_config.splitlines(), - fromfile="running-config", - tofile="candidate-config", - lineterm="", - ) - return "\n".join(diff) + def _get_running_config(self) -> str: + return self._send_command("show running-config") def commit_config(self, message: str = "", revert_in: Optional[int] = None) -> None: """Apply the staged candidate configuration to the device and persist it. @@ -1344,11 +1292,6 @@ class NetgearSmartDriver(SwitchDriver): self._candidate_config = None self._candidate_mode = None - def discard_config(self) -> None: - """Discard the staged candidate configuration without applying it.""" - self._candidate_config = None - self._candidate_mode = None - def rollback(self) -> None: """Revert the running config to the state before the last commit. @@ -1462,10 +1405,6 @@ class NetgearSmartDriver(SwitchDriver): # Additional NAPALM getters # ------------------------------------------------------------------ - def has_pending_commit(self) -> bool: - """Return True when a candidate configuration is staged.""" - return self._candidate_config is not None - def get_environment(self) -> Dict: """Return device environment data (CPU, memory).