NAPALM/napalm-netgear
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 aef5eb1487
commit 36a239692b
2 changed files with 86 additions and 133 deletions
Download Patch File Download Diff File Expand all files Collapse all files
napalm_netgear/netgear_plus.py
+61 -47
View File
@@ -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 ``<input class="portName" value="...">``
@@ -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: <span vid-4>...<i ...></i>24</span><span vnm-4 ...>Guest</span>
vlan_names: Dict[str, str] = {}
vlan_names: dict[str, str] = {}
vlan_entry_re = re.compile(
r'<span\s+vid-4[^>]*>.*?(\d+)\s*</span>\s*'
r'<span\s+vnm-4[^>]*>([^<]*)</span>',
@@ -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.
napalm_netgear/netgear_smart.py
+25 -86
View File
@@ -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).