From a8ce5a803364a3b0064efa25c4a0f918c768c7aa Mon Sep 17 00:00:00 2001 From: Christian Manivong Date: Fri, 29 May 2026 09:13:59 +0200 Subject: [PATCH] initial commit --- .gitignore | 28 + README.md | 112 +++ napalm_netgear/__init__.py | 6 + napalm_netgear/netgear_plus.py | 787 +++++++++++++++++ napalm_netgear/netgear_smart.py | 1423 +++++++++++++++++++++++++++++++ pyproject.toml | 55 ++ requirements.txt | 3 + tests/__init__.py | 0 tests/unit/__init__.py | 0 tests/unit/test_driver.py | 644 ++++++++++++++ 10 files changed, 3058 insertions(+) create mode 100644 .gitignore create mode 100644 README.md create mode 100644 napalm_netgear/__init__.py create mode 100644 napalm_netgear/netgear_plus.py create mode 100644 napalm_netgear/netgear_smart.py create mode 100644 pyproject.toml create mode 100644 requirements.txt create mode 100644 tests/__init__.py create mode 100644 tests/unit/__init__.py create mode 100644 tests/unit/test_driver.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..eead559 --- /dev/null +++ b/.gitignore @@ -0,0 +1,28 @@ +# Byte-compiled / optimized / DLL files +__pycache__/ +*.py[cod] +*$py.class + +# Distribution / packaging +.eggs/ +dist/ +build/ +*.egg-info/ +*.egg + +# Virtual environments +.venv/ +venv/ +env/ + +# Testing +.pytest_cache/ +.coverage +htmlcov/ + +# IDEs +.vscode/ +.idea/ + +# OS +.DS_Store diff --git a/README.md b/README.md new file mode 100644 index 0000000..2a7c61b --- /dev/null +++ b/README.md @@ -0,0 +1,112 @@ +# napalm-netgear + +NAPALM community driver for **Netgear Smart Managed** and **Netgear Plus Managed** switches +(GS108EPP, GS110TP, GS316EPP and compatible Smart Managed Pro series). + +## Tested devices + +| Model | Series | Tested | +|---|---|---| +| GS110TP | Smart Managed Pro | ✅ | +| GS108EPP | Smart Managed Pro | ✅ | +| GS316EPP | Smart Managed Pro | ✅ | + +> Other models from the Netgear Smart Managed Pro series should work as well — +> contributions welcome. + +## Requirements + +| Dependency | Minimum version | +|---|---| +| Python | 3.8 | +| NAPALM | 4.0 | +| Netmiko | 4.0 | + +## Installation + +```bash +pip install napalm-netgear +``` + +Or from source: + +```bash +git clone https://github.com/napalm-automation-community/napalm-netgear +cd napalm-netgear +pip install -e . +``` + +## Quick start + +```python +from napalm import get_network_driver + +driver = get_network_driver("netgear_smart") +with driver("192.168.0.239", "admin", "password") as device: + facts = device.get_facts() + print(facts) +``` + +## Implemented getters + +| Getter | Status | Notes | +|---|---|---| +| `get_facts` | ✅ | model, hostname, os_version, serial, uptime | +| `get_interfaces` | ✅ | status, speed, description, MTU | +| `get_interfaces_ip` | ✅ | management VLAN IP/mask | +| `get_config` | ✅ | running + startup | +| `get_arp_table` | ✅ | | +| `get_mac_address_table` | ✅ | | +| `get_lldp_neighbors` | ✅ | | +| `get_lldp_neighbors_detail` | ✅ | | +| `get_vlans` | ✅ | | +| `get_environment` | ✅ | CPU + memory only | +| `get_interfaces_counters` | ✅ | | +| `get_users` | ✅ | | +| `get_snmp_information` | ✅ | | +| `get_ntp_servers` | ✅ | | +| `get_ntp_peers` | ✅ | same as servers | +| `get_route_to` | ✅ | static/connected only | +| `ping` | ✅ | | +| `cli` | ✅ | | +| `is_alive` | ✅ | | +| `load_merge_candidate` | ✅ | | +| `load_replace_candidate` | ✅ | additive — no atomic replace | +| `compare_config` | ✅ | | +| `commit_config` | ✅ | | +| `discard_config` | ✅ | | +| `rollback` | ✅ | session-scoped backup | +| `get_bgp_neighbors` | ❌ | not applicable | +| `get_optics` | ❌ | DDM not exposed via CLI | +| `get_ipv6_neighbors_table` | ❌ | not exposed via CLI | +| `get_ntp_stats` | ❌ | not exposed via CLI | + +## Optional arguments + +| Argument | Default | Description | +|---|---|---| +| `port` | `22` | SSH port | +| `force_no_enable` | `False` | Skip `enable` after login | +| `canonical_int_fmt` | `False` | Use canonical interface names | + +Any additional keyword arguments are forwarded to Netmiko. + +## CLI notes + +* Netmiko device type: `netgear_prosafe` +* Prompt pattern: `(hostname) >` / `(hostname) #` / `(hostname) (Config)#` +* Interface naming: `0/1`, `0/2`, … (slot/port); LAG channels: `ch1`, `ch2`, … +* Config is persisted with `write memory` +* Key-value output uses `Key: value` **or** `Key..... value` notation + depending on firmware generation — both are handled automatically + +## Development + +```bash +pip install -e ".[dev]" +pytest tests/ +``` + +## License + +Apache 2.0 — see [LICENSE](LICENSE). diff --git a/napalm_netgear/__init__.py b/napalm_netgear/__init__.py new file mode 100644 index 0000000..d1fc944 --- /dev/null +++ b/napalm_netgear/__init__.py @@ -0,0 +1,6 @@ +"""NAPALM driver for Netgear Smart Managed switches.""" + +from napalm_netgear_plus.netgear_smart import NetgearSmartDriver +from napalm_netgear_plus.netgear_plus import NetgearPlusDriver + +__all__ = ["NetgearSmartDriver", "NetgearPlusDriver"] diff --git a/napalm_netgear/netgear_plus.py b/napalm_netgear/netgear_plus.py new file mode 100644 index 0000000..c6dc49b --- /dev/null +++ b/napalm_netgear/netgear_plus.py @@ -0,0 +1,787 @@ +# -*- coding: utf-8 -*- +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""NAPALM driver for Netgear Plus Smart Managed switches (HTTP-based). + +Tested against: GS308EP and compatible Plus Smart Managed series. +Transport: HTTP web interface via py-netgear-plus library. + +These switches use password-only web authentication (no username). +The ``username`` parameter is accepted but ignored. +""" + +import threading +import time +from typing import Any, Dict, List, Optional, Tuple + +from napalm_device_types import SwitchDriver +from napalm.base.exceptions import ConnectionException + +# --------------------------------------------------------------------------- +# Firmware update check — module-level constants and cache +# --------------------------------------------------------------------------- + +# Netgear CDN base URL for firmware zip files. +# Pattern: {_FW_CDN_BASE}/{MODEL}/{MODEL}_{VERSION}.zip +_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] = { + "GS305EP": "V2.0.0.10", + "GS305EPP": "V2.0.0.10", + "GS308EP": "V2.0.0.11", + "GS308EPP": "V2.0.0.11", +} + +# In-memory cache: model_id -> (latest_version, unix_timestamp_of_check) +_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, ...]: + """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(".")) + + +def _fw_cdn_exists(model: str, version: str) -> bool: + """Return True if a firmware zip exists on the Netgear CDN (HEAD request).""" + import urllib.request + import ssl + + url = f"{_FW_CDN_BASE}/{model}/{model}_{version}.zip" + ctx = ssl.create_default_context() + req = urllib.request.Request(url, method="HEAD", + headers={"User-Agent": "Mozilla/5.0"}) + try: + with urllib.request.urlopen(req, timeout=3, context=ctx): + return True + except Exception: + return False + + +def _fw_probe_newer(model: str, baseline: str) -> str: + """Probe the Netgear CDN for firmware newer than *baseline*. + + Checks up to ~22 candidate versions (patch increments, next micro, next + major). Returns the highest confirmed version string, which may equal + *baseline* if nothing newer is found. + """ + try: + base_t = _fw_version_tuple(baseline) + except ValueError: + return baseline + + if len(base_t) != 4: + return baseline + + major, minor, micro, patch = base_t + best_t = base_t + best_v = baseline + + 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: + best_t = candidate + best_v = v_str + + # Probe the next 20 patch versions + for p in range(patch + 1, patch + 21): + _try((major, minor, micro, p)) + + # Probe next micro version (reset patch to 0) + _try((major, minor, micro + 1, 0)) + + # Probe next major version (e.g. V2.0.0.0 when current is V1.x.x.x) + _try((major + 1, 0, 0, 0)) + + return best_v + + +def _fw_get_latest(model: str) -> Optional[str]: + """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. + """ + model_upper = model.upper() + baseline = _FW_KNOWN_LATEST.get(model_upper) + if baseline is None: + return None + + now = time.monotonic() + + with _fw_cache_lock: + cached = _fw_cache.get(model_upper) + if cached is not None: + version, ts = cached + if now - ts < _FW_CACHE_TTL: + return version + + # Cache miss or expired — probe CDN (outside the lock to avoid blocking + # other threads during potentially slow HTTP requests). + latest = _fw_probe_newer(model_upper, baseline) + + with _fw_cache_lock: + _fw_cache[model_upper] = (latest, now) + + return latest + + +class NetgearPlusDriver(SwitchDriver): + """HTTP-based NAPALM driver for Netgear Plus Smart Managed switches.""" + + VENDOR = "Netgear" + + def __init__( + self, + hostname: str, + username: str, + password: str, + timeout: int = 60, + optional_args: Optional[Dict] = 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 + + if optional_args is None: + optional_args = {} + + # ------------------------------------------------------------------ + # Connection management + # ------------------------------------------------------------------ + + def open(self) -> None: + """Authenticate to the switch via HTTP.""" + try: + from py_netgear_plus import NetgearSwitchConnector + except ImportError as exc: + raise ConnectionException( + "py-netgear-plus is not installed. " + "Add 'py-netgear-plus' to your dependencies." + ) from exc + + try: + connector = NetgearSwitchConnector(self.hostname, self.password) + connector.autodetect_model() + if not connector.get_login_cookie(): + raise ConnectionException( + f"Login failed for {self.hostname}: bad password or unreachable." + ) + self._connector = connector + self._switch_infos_cache = None + except Exception as exc: + if isinstance(exc, ConnectionException): + raise + raise ConnectionException( + f"Cannot connect to {self.hostname}: {exc}" + ) from exc + + def close(self) -> None: + """Log out from the switch.""" + if self._connector is not None: + try: + self._connector.delete_login_cookie() + except Exception: + pass + self._connector = None + self._switch_infos_cache = None + + 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]: + """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]: + """Fetch port names/descriptions from dashboard.cgi. + + The GS30x series embeds ```` + hidden inputs in the dashboard HTML. py-netgear-plus does not + parse them, so we do it here. Returns a mapping of + {port_number: name_string} (1-indexed, empty string if not set). + """ + import re + + try: + url = f"http://{self.hostname}/dashboard.cgi" + resp = self._connector.fetch_page("get", url, {}) + except Exception: + return {} + if not resp or resp.status_code != 200 or not resp.text: + return {} + portname_re = re.compile( + r']+class="portName"[^>]+value="([^"]*)"', + re.IGNORECASE, + ) + names = portname_re.findall(resp.text) + return {i + 1: name for i, name in enumerate(names)} + + def _port_name(self, port_number: int) -> str: + return f"port{port_number}" + + def _interface_to_port_number(self, interface: str) -> int: + """Convert an interface name like 'port3' to the integer port number 3.""" + name = interface.strip().lower() + if name.startswith("port"): + try: + return int(name[4:]) + except ValueError: + pass + raise ValueError( + f"Cannot resolve interface {interface!r} to a port number. " + "Expected 'portN' (e.g. 'port1')." + ) + + # ------------------------------------------------------------------ + # NAPALM getters + # ------------------------------------------------------------------ + + 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", "") + hostname = infos.get("switch_name", self.hostname) + os_version = infos.get("switch_firmware", "") + serial_number = infos.get("switch_serial_number", "") + interface_list = [ + self._port_name(i) for i in range(1, self._connector.ports + 1) + ] + return { + "vendor": self.VENDOR, + "model": str(model), + "hostname": hostname, + "fqdn": hostname, + "os_version": os_version, + "serial_number": serial_number, + "uptime": 0.0, + "interface_list": interface_list, + } + + def get_interfaces(self) -> Dict[str, Dict]: + """Return a dictionary of interface details. + + Each interface entry contains: + ``is_up``, ``is_enabled``, ``description``, ``last_flapped``, + ``speed``, ``mtu``, ``mac_address``. + """ + infos = self._get_switch_infos() + port_names = self._get_port_names() + 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" + speed = float(infos.get(f"port_{port}_connection_speed", 0) or 0) + description = port_names.get(port, "") or infos.get(f"port_{port}_description", "") or "" + interfaces[name] = { + "is_up": status, + "is_enabled": True, + "description": description, + "last_flapped": -1.0, + "speed": speed, + "mtu": 1518, + "mac_address": "", + } + return interfaces + + def get_interfaces_ip(self) -> Dict[str, Dict]: + """Return IP addresses per interface. + + Netgear Plus switches are L2 only — no routed interfaces. + Always returns an empty dict. + """ + return {} + + 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] = {} + for port in range(1, self._connector.ports + 1): + name = self._port_name(port) + # sum_*_mbytes are lifetime totals in megabytes; convert to bytes + rx_mb = float(infos.get(f"port_{port}_sum_rx_mbytes", 0) or 0) + tx_mb = float(infos.get(f"port_{port}_sum_tx_mbytes", 0) or 0) + crc = int(infos.get(f"port_{port}_crc_errors", 0) or 0) + counters[name] = { + "tx_errors": 0, + "rx_errors": crc, + "tx_discards": 0, + "rx_discards": 0, + "tx_octets": int(tx_mb * 1_000_000), + "rx_octets": int(rx_mb * 1_000_000), + "tx_unicast_packets": 0, + "rx_unicast_packets": 0, + "tx_multicast_packets": 0, + "rx_multicast_packets": 0, + "tx_broadcast_packets": 0, + "rx_broadcast_packets": 0, + } + return counters + + def get_environment(self) -> Dict[str, Any]: + """Return environment data. + + Reports PoE power status per PoE port. Temperature, fans, CPU and + memory are not available on these switches and are returned as empty + dicts. + """ + infos = self._get_switch_infos() + power: Dict[str, Any] = {} + poe_max_single = getattr( + self._connector.switch_model, "POE_MAX_POWER_SINGLE_PORT", None + ) + for port in self._connector.poe_ports: + poe_active = infos.get(f"port_{port}_poe_power_active") + poe_status = bool(poe_active) if poe_active is not None else False + power[self._port_name(port)] = { + "status": poe_status, + "capacity": float(poe_max_single) if poe_max_single else 0.0, + "used": 0.0, + } + return { + "fans": {}, + "temperature": {}, + "power": power, + "cpu": {}, + "memory": {}, + } + + def get_mac_address_table(self) -> List[Dict]: + """Return the MAC address table. + + Not available via the HTTP web interface of Plus switches. + Returns an empty list. + """ + return [] + + def get_arp_table(self, vrf: str = "") -> List[Dict]: + """Return the ARP table. + + Netgear Plus switches are L2 only and have no ARP table. + Returns an empty list. + """ + return [] + + def get_lldp_neighbors(self) -> Dict[str, Any]: + """Return LLDP neighbors. + + Not supported on Netgear Plus switches. + Returns an empty dict. + """ + return {} + + def get_lldp_neighbors_detail(self, interface: str = "") -> Dict[str, Any]: + """Return detailed LLDP neighbor info. + + Not supported on Netgear Plus switches. + Returns an empty dict. + """ + return {} + + 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: + {"1": {"name": "Default", "tagged": ["port1"], "untagged": ["port2"]}} + + Parses 802.1Q Advanced VLAN data from the switch web interface. + Each port's PVID (native VLAN) is listed with a trailing "*" in the + per-port VLAN membership spans. + """ + import re + + url = f"http://{self.hostname}/vlan.cgi" + try: + resp = self._connector.fetch_page("get", url, {}) + except Exception: + return {} + + if not resp or resp.status_code != 200 or not resp.text: + return {} + + text = resp.text + + # Parse VLAN IDs and names from vid-4 / vnm-4 span pairs. + # Example: ...24Guest + vlan_names: Dict[str, str] = {} + vlan_entry_re = re.compile( + r']*>.*?(\d+)\s*\s*' + r']*>([^<]*)', + re.DOTALL, + ) + for m in vlan_entry_re.finditer(text): + vid = m.group(1).strip() + name = m.group(2).strip() + vlan_names[vid] = name + + if not vlan_names: + return {} + + vlans: Dict[str, Any] = { + vid: {"name": name, "tagged": [], "untagged": []} + for vid, name in vlan_names.items() + } + + # Parse per-port PVID membership. + # Each 1*, 24, 25, corresponds to one port + # in order (port 1, port 2, ...). A VID followed by "*" is the native + # (untagged) VLAN for that port; all others are tagged. + pvid_span_re = re.compile(r']*pvid-str[^>]*>([^<]+)') + for port_idx, pvid_str in enumerate(pvid_span_re.finditer(text)): + port_num = port_idx + 1 + if port_num > self._connector.ports: + break + port_name = self._port_name(port_num) + for token in pvid_str.group(1).split(","): + token = token.strip() + if not token: + continue + if token.endswith("*"): + vid = token[:-1] + role = "untagged" + else: + vid = token + role = "tagged" + if vid in vlans: + vlans[vid][role].append(port_name) + + return vlans + + def get_vlans(self) -> Dict[str, Any]: + """Return VLAN information (standard NAPALM format). + + Delegates to get_vlans_detail() and merges tagged and untagged ports + into a single ``interfaces`` list. + """ + detail = self.get_vlans_detail() + return { + vid: { + "name": info["name"], + "interfaces": info["tagged"] + info["untagged"], + } + for vid, info in detail.items() + } + + def set_vlan(self, vlan_id: int, config: Dict) -> None: + """Create or update a VLAN. + + Not supported via HTTP API. + """ + raise NotImplementedError( + "VLAN management is not supported for Netgear Plus switches via HTTP." + ) + + def delete_vlan(self, vlan_id: int) -> None: + """Delete a VLAN. + + Not supported via HTTP API. + """ + raise NotImplementedError( + "VLAN management is not supported for Netgear Plus switches via HTTP." + ) + + def get_config( + self, + retrieve: str = "all", + full: bool = False, + sanitized: bool = False, + format: str = "text", + ) -> Dict[str, str]: + """Return device configuration. + + Netgear Plus switches do not expose running/startup config via HTTP. + Returns empty strings for all config slots. + """ + return {"running": "", "startup": "", "candidate": ""} + + def load_merge_candidate(self, filename: Optional[str] = None, config: Optional[str] = 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: + raise NotImplementedError( + "Config management is not supported for Netgear Plus switches." + ) + + def compare_config(self) -> str: + raise NotImplementedError( + "Config management is not supported for Netgear Plus switches." + ) + + def commit_config(self, message: str = "") -> None: + raise NotImplementedError( + "Config management is not supported for Netgear Plus switches." + ) + + def discard_config(self) -> None: + raise NotImplementedError( + "Config management is not supported for Netgear Plus switches." + ) + + def rollback(self) -> None: + raise NotImplementedError( + "Config management is not supported for Netgear Plus switches." + ) + + def has_pending_commit(self) -> bool: + return False + + 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]: + """Execute ping. + + Not supported on Netgear Plus switches via HTTP. + """ + raise NotImplementedError( + "ping() is not supported for Netgear Plus switches." + ) + + def traceroute( + self, + destination: str, + source: str = "", + ttl: int = 255, + timeout: int = 2, + vrf: str = "", + ) -> Dict[str, Any]: + """Execute traceroute. + + Not supported on Netgear Plus switches via HTTP. + """ + raise NotImplementedError( + "traceroute() is not supported for Netgear Plus switches." + ) + + def cli(self, commands: List[str], encoding: str = "text") -> Dict[str, Any]: + """Send CLI commands. + + Netgear Plus switches have no SSH CLI — only HTTP. + """ + raise NotImplementedError( + "cli() is not supported for Netgear Plus switches (HTTP-only device)." + ) + + 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]: + """Return SNMP configuration. + + Not available via HTTP API. Returns an empty dict. + """ + return {} + + 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]: + """Return NTP peers. + + Not available via HTTP API. Returns an empty dict. + """ + return {} + + def get_ntp_stats(self) -> List[Dict]: + """Return NTP stats. + + Not available via HTTP API. Returns an empty list. + """ + return [] + + # ------------------------------------------------------------------ + # SwitchDriver abstract methods (not applicable for this device) + # ------------------------------------------------------------------ + + 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]: + """Return port-channel information. + + Not available via HTTP API. Returns an empty dict. + """ + return {} + + 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]: + """Return MAC ACL information. + + Not available via HTTP API. Returns an empty dict. + """ + return {} + + def get_poe_status(self) -> Dict[str, Any]: + """Return PoE status. + + Reports PoE power status from switch infos. + """ + infos = self._get_switch_infos() + poe_max_all = getattr( + self._connector.switch_model, "POE_MAX_POWER_ALL_PORTS", None + ) + poe_max_single = getattr( + self._connector.switch_model, "POE_MAX_POWER_SINGLE_PORT", None + ) + 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 + ports[self._port_name(p)] = { + "enabled": True, + "status": "delivering" if is_delivering else "searching", + "poe_class": "unknown", + "power_draw": 0.0, + "power_budget": float(poe_max_single) if poe_max_single else 0.0, + "voltage": 0.0, + "current": 0.0, + } + return { + "total_power_budget": float(poe_max_all) if poe_max_all else 0.0, + "total_power_draw": 0.0, + "ports": ports, + } + + def set_poe_enabled(self, interface: str, enabled: bool) -> None: + """Enable or disable PoE on a port. + + Uses py-netgear-plus to switch the PoE port state. + """ + port_number = self._interface_to_port_number(interface) + if port_number not in self._connector.poe_ports: + raise ValueError(f"{interface} is not a PoE port on this switch.") + if enabled: + self._connector.turn_on_poe_port(port_number) + else: + self._connector.turn_off_poe_port(port_number) + + def power_cycle_port(self, interface: str, delay: int = 5) -> None: + """Power-cycle a PoE port. + + Uses py-netgear-plus power cycle functionality. + """ + port_number = self._interface_to_port_number(interface) + if port_number not in self._connector.poe_ports: + 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: + """Configure an interface. + + Not supported via HTTP API. + """ + raise NotImplementedError( + "set_interface() is not supported for Netgear Plus switches." + ) + + def get_device_warnings(self) -> list: + """Check for available firmware updates and return a warning if the + device is not running the latest known firmware. + + The latest version is looked up once per 24 hours (per model, per + worker process) via lightweight HEAD requests against Netgear's + download CDN. Returns an empty list when the model is unknown or + when the check fails. + """ + warnings: list = [] + try: + model_name: str = getattr( + self._connector.switch_model, "MODEL_NAME", "" + ) or self._connector.switch_model.__class__.__name__ + current_fw: str = self._get_switch_infos().get("switch_firmware", "") + + if not model_name or not current_fw: + return warnings + + latest_fw = _fw_get_latest(model_name) + if latest_fw is None: + return warnings + + try: + current_t = _fw_version_tuple(current_fw) + latest_t = _fw_version_tuple(latest_fw) + except ValueError: + return warnings + + if latest_t > current_t: + warnings.append( + { + "code": "firmware_update_available", + "severity": "warning", + "title": "Firmware update available", + "message": ( + f"Version {latest_fw} is available for {model_name} " + f"(installed: {current_fw})." + ), + "action": None, + "action_label": None, + "meta": { + "download_url": ( + f"{_FW_CDN_BASE}/{model_name}" + f"/{model_name}_{latest_fw}.zip" + ), + "latest_version": latest_fw, + "current_version": current_fw, + }, + } + ) + except Exception: + pass + + return warnings diff --git a/napalm_netgear/netgear_smart.py b/napalm_netgear/netgear_smart.py new file mode 100644 index 0000000..ea1fac3 --- /dev/null +++ b/napalm_netgear/netgear_smart.py @@ -0,0 +1,1423 @@ +# -*- coding: utf-8 -*- +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""NAPALM driver for Netgear Smart Managed switches. + +Tested against: GS108EPP, GS110TP, GS316EPP and compatible Smart Managed Pro series. +Netmiko device_type: ``netgear_prosafe`` + +CLI notes +--------- +* Prompt format: ``(hostname) >`` (user) / ``(hostname) #`` (privileged) / + ``(hostname) (Config)#`` (config mode). +* Key-value output uses either ``Key: value`` or ``Key..... value`` notation + depending on firmware generation. +* Interface names are ``0/1``, ``0/2``, … (slot/port) on all Smart Managed Pro + models. Link-Aggregation channels appear as ``ch1``, ``ch2``, etc. +* Config is saved with ``write memory``. +""" + +import difflib +import re +import socket +from typing import Any, Dict, List, Optional, Union + +import netaddr +from netmiko import ConnectHandler +from netmiko.exceptions import NetmikoAuthenticationException, NetmikoTimeoutException + +from napalm_device_types import SwitchDriver +from napalm.base import helpers as napalm_helpers +from napalm.base.exceptions import ( + CommandErrorException, + ConnectionClosedException, + ConnectionException, + MergeConfigException, + ReplaceConfigException, +) +from napalm.base.netmiko_helpers import netmiko_args +import napalm.base.constants as C + + +class NetgearSmartDriver(SwitchDriver): + """NAPALM driver for Netgear Smart Managed switches (GS-series).""" + + VENDOR = "Netgear" + # Netmiko device type – handles prompt ``(hostname) >`` / ``(hostname) #`` + NETMIKO_DEVICE_TYPE = "netgear_prosafe" + + def __init__( + self, + hostname: str, + username: str, + password: str, + timeout: int = 60, + optional_args: Optional[Dict] = None, + ) -> None: + self.hostname = hostname + self.username = username + self.password = password + self.timeout = timeout + self.device: Optional[ConnectHandler] = None + + if optional_args is None: + optional_args = {} + + self.force_no_enable = optional_args.get("force_no_enable", False) + self.port = optional_args.get("port", 22) + self.use_canonical_interface = optional_args.get("canonical_int_fmt", False) + + # Netgear Smart switches run older SSH stacks that advertise ssh-rsa but + # reject connections from clients that negotiate rsa-sha2-256/512 first + # (the Paramiko 3+ default). Force legacy pubkey algorithms unless the + # caller has already supplied their own disabled_algorithms override. + if "disabled_algorithms" not in optional_args: + optional_args.setdefault( + "disabled_algorithms", + {"pubkeys": ["rsa-sha2-256", "rsa-sha2-512"]}, + ) + + # Use a generous conn_timeout for slow-to-respond embedded SSH daemons. + optional_args.setdefault("conn_timeout", 20) + optional_args.setdefault("banner_timeout", 20) + + self.netmiko_optional_args = netmiko_args(optional_args) + + # Config management state + self._candidate_config: Optional[str] = None + self._candidate_mode: Optional[str] = None # 'merge' or 'replace' + self._backup_config: Optional[str] = None + + # ------------------------------------------------------------------ + # Connection management + # ------------------------------------------------------------------ + + def open(self) -> None: + """Open an SSH connection to the device.""" + try: + self.device = ConnectHandler( + device_type=self.NETMIKO_DEVICE_TYPE, + host=self.hostname, + username=self.username, + password=self.password, + timeout=self.timeout, + **self.netmiko_optional_args, + ) + if not self.force_no_enable: + self.device.enable() + except NetmikoTimeoutException as exc: + raise ConnectionException( + f"Cannot connect to {self.hostname}: {exc}" + ) from exc + except NetmikoAuthenticationException as exc: + raise ConnectionException( + f"Authentication failed for {self.hostname}: {exc}" + ) from exc + except Exception as exc: + raise ConnectionException( + f"Verbindungsaufbau fehlgeschlagen (ConnectionException): " + f"Cannot connect to {self.hostname}:\n{exc}" + ) from exc + + def close(self) -> None: + """Close the SSH connection.""" + if self.device: + self.device.disconnect() + self.device = None + + def is_alive(self) -> Dict[str, bool]: + """Return connection liveness without writing to the channel.""" + if self.device is None: + return {"is_alive": False} + try: + return {"is_alive": self.device.remote_conn.transport.is_active()} + except (socket.error, EOFError, AttributeError): + return {"is_alive": False} + + # ------------------------------------------------------------------ + # Internal helpers + # ------------------------------------------------------------------ + + def _send_command(self, command: Union[str, List[str]]) -> str: + """Send a command (or ordered list of fallback commands) to the device. + + Uses an explicit ``expect_string`` so that Netmiko never confuses a + data line inside the output with the device prompt. + """ + prompt_pattern = rf"{re.escape(self.device.base_prompt)}[>#]" + + def _do_send(cmd: str) -> str: + return self.device.send_command( + cmd, + expect_string=prompt_pattern, + read_timeout=self.timeout, + ).strip() + + try: + if isinstance(command, list): + output = "" + for cmd in command: + output = _do_send(cmd) + if "Invalid" not in output and "Error" not in output: + return output + return output + return _do_send(command) + except (socket.error, EOFError) as exc: + raise ConnectionClosedException(str(exc)) from exc + + # ------------------------------------------------------------------ + # Configuration-mode helpers + # ------------------------------------------------------------------ + + def _exec_prompt(self) -> str: + return rf"{re.escape(self.device.base_prompt)}[>#]" + + def _conf_prompt(self) -> str: + """Matches ``(hostname) (Config)#`` config-mode prompt.""" + return rf"{re.escape(self.device.base_prompt)}\s*\(Config\)[>#]" + + def _any_prompt(self) -> str: + """Matches exec prompt and any config sub-mode prompt.""" + return rf"{re.escape(self.device.base_prompt)}(?:\s*\([^)]*\))?[>#]" + + def _enter_config_mode(self) -> None: + self.device.send_command( + "configure", + expect_string=self._conf_prompt(), + read_timeout=self.timeout, + ) + + def _exit_config_mode(self) -> None: + self.device.send_command( + "exit", + expect_string=self._exec_prompt(), + read_timeout=self.timeout, + ) + + def _save_config(self) -> None: + self.device.send_command( + "write memory", + expect_string=self._exec_prompt(), + read_timeout=self.timeout, + ) + + def _apply_config_lines(self, config_text: str) -> List[str]: + """Send config lines while in config mode; return list of rejected lines.""" + ep_any = self._any_prompt() + errors: List[str] = [] + for line in config_text.splitlines(): + stripped = line.strip() + if not stripped or stripped.startswith(("!", "#")): + continue + out = self.device.send_command( + stripped, + expect_string=ep_any, + read_timeout=self.timeout, + ).strip() + if out and ("Error" in out or "Invalid" in out or "Unknown" in out): + errors.append(f" {stripped!r}: {out}") + return errors + + @staticmethod + def _parse_key_value(output: str, key: str) -> str: + """Extract a value from Netgear ``key.... value`` or ``key: value`` output. + + Both dot-separator (``show version``) and colon-separator + (``show sysinfo``) forms are handled automatically. + """ + for line in output.splitlines(): + if key.lower() in line.lower(): + # dots form: "Software Version............................. 6.6.3" + m = re.match(r"[^.\n]+\.{2,}\s*(.*)", line.strip()) + if m: + return m.group(1).strip() + # colon form: "System Name: myswitch" + m = re.match(r"[^:\n]+:\s*(.*)", line.strip()) + if m: + return m.group(1).strip() + return "" + + @staticmethod + def _parse_uptime_seconds(uptime_str: str) -> float: + """Convert Netgear uptime string to seconds. + + Expected format: ``0 days 2 hrs 43 mins 18 secs`` + Also handles ``day(s) hour(s) min(s) sec(s)`` variants. + """ + days = hours = minutes = seconds = 0 + m = re.search(r"(\d+)\s+day", uptime_str, re.I) + if m: + days = int(m.group(1)) + m = re.search(r"(\d+)\s+hr", uptime_str, re.I) + if m: + hours = int(m.group(1)) + m = re.search(r"(\d+)\s+min", uptime_str, re.I) + if m: + minutes = int(m.group(1)) + m = re.search(r"(\d+)\s+sec", uptime_str, re.I) + if m: + seconds = int(m.group(1)) + return float(days * 86400 + hours * 3600 + minutes * 60 + seconds) + + # ------------------------------------------------------------------ + # NAPALM getters + # ------------------------------------------------------------------ + + def get_facts(self) -> Dict: + """Return general device facts. + + Combines ``show sysinfo`` and ``show version``. + + Example ``show sysinfo`` output:: + + System Description: GS110TP Gigabit Smart Managed Pro Switch + Machine Model: GS110TP + Machine Type: Ethernet Switch + Burned In MAC Address: C4:AD:34:AB:12:34 + System OID String: 1.3.6.1.4.1.4526.100.5.4 + System Uptime: 0 days 2 hrs 43 mins 18 secs + System Location: Server Room + System Contact: admin@example.com + System Name: myswitch + + Example ``show version`` output:: + + Software Version............................. 6.6.3 + Loader Version............................... 1.0.0.15 + Boot ROM Version............................. B1.0.0.15 + Hardware Version............................. V1 + Serial Number................................ 1FE2345678 + """ + sysinfo = self._send_command("show sysinfo") + version = self._send_command("show version") + + hostname = self._parse_key_value(sysinfo, "System Name") + + # "Machine Model" is the most reliable field for the switch SKU + model = self._parse_key_value(sysinfo, "Machine Model") + if not model: + desc = self._parse_key_value(sysinfo, "System Description") + model = desc.split()[0] if desc else "" + + os_version = ( + self._parse_key_value(version, "Software Version") + or self._parse_key_value(sysinfo, "Software Version") + ) + + uptime_str = self._parse_key_value(sysinfo, "System Uptime") + uptime = self._parse_uptime_seconds(uptime_str) + + serial_number = ( + self._parse_key_value(version, "Serial Number") + or self._parse_key_value(sysinfo, "Serial Number") + or "" + ) + + interface_list = self._get_interface_list() + + return { + "vendor": self.VENDOR, + "model": model, + "hostname": hostname, + "fqdn": hostname, + "os_version": os_version, + "serial_number": serial_number, + "uptime": uptime, + "interface_list": interface_list, + } + + def _get_interface_list(self) -> List[str]: + """Return sorted list of slot/port interface names from ``show port all``.""" + output = self._send_command("show port all") + interfaces: List[str] = [] + for line in output.splitlines(): + # Physical ports: "0/1 ..." and LAG channels: "ch1 ..." + m = re.match(r"^\s*(\d+/\d+|ch\d+)\s+", line) + if m: + interfaces.append(m.group(1)) + return sorted( + set(interfaces), + key=lambda s: [int(x) if x.isdigit() else x for x in re.split(r"(\d+)", s)], + ) + + def get_interfaces(self) -> Dict[str, Dict]: + """Return per-interface details. + + Combines ``show port all`` (link status, speed) with + ``show interface all`` (description, MTU). + + Example ``show port all`` columns:: + + Intf Type Admin Physical Physical Link + Mode Mode Status Status + ------ ---- ------ ---------- ---------- ------ + 0/1 Enable Auto 100M/Full Up + 0/2 Enable Auto - Down + + Example ``show interface all`` (block per port):: + + Interface................................ 0/1 + Description.............................. uplink + MTU...................................... 1518 + """ + port_out = self._send_command("show port all") + intf_out = self._send_command("show interface all") + return self._parse_interfaces(port_out, intf_out) + + def _parse_interfaces( + self, port_output: str, intf_output: str = "" + ) -> Dict[str, Dict]: + """Parse ``show port all`` and ``show interface all`` into NAPALM format.""" + interfaces: Dict[str, Dict] = {} + + # --- parse 'show port all' --- + # Skip header lines until we hit a separator row, then parse data rows. + # Data row: "0/1 Enable Auto 100M/Full Up ..." + # The "Type" column is often empty; splitting by whitespace is more + # robust than relying on fixed-width column positions. + in_table = False + for line in port_output.splitlines(): + line_s = line.strip() + if re.match(r"^-{4,}", line_s): + in_table = True + continue + if not in_table: + continue + parts = line_s.split() + # Must start with a port name (slot/port or channel) + if not parts or not re.match(r"^(\d+/\d+|ch\d+)$", parts[0]): + continue + # With an empty Type column the fields map to: + # parts[0]=port parts[1]=admin parts[2]=phys_mode + # parts[3]=phys_status parts[4]=link_status + # When Type is present (non-empty token) the indices shift by +1; + # detect this by checking whether parts[1] looks like a type token + # (not Enable/Disable). + offset = 0 + if len(parts) > 1 and parts[1].lower() not in ("enable", "disable"): + offset = 1 + if len(parts) < 5 + offset: + continue + port = parts[0] + admin = parts[1 + offset] # Enable / Disable + phys_status = parts[3 + offset] # e.g. "100M/Full", "1G/Full", or "-" + link_status = parts[4 + offset] # Up / Down + + speed_val = 0.0 + speed_m = re.search(r"(\d+)([MG])", phys_status) + if speed_m: + speed_val = float(speed_m.group(1)) + if speed_m.group(2) == "G": + speed_val *= 1000.0 + + interfaces[port] = { + "is_up": link_status.lower() == "up", + "is_enabled": admin.lower() == "enable", + "description": "", + "last_flapped": -1.0, + "speed": speed_val, + "mtu": 1518, + "mac_address": "", + } + + # --- parse 'show interface all' (block format) --- + current_port: Optional[str] = None + for line in intf_output.splitlines(): + # Block header: "Interface................................ 0/1" + m = re.match(r"^\s*Interface[.:\s]+(\d+/\d+|ch\d+)", line, re.I) + if m: + current_port = m.group(1) + continue + if current_port is None: + continue + # Description + dm = re.match(r"^\s*(?:Description|Port Description)[.:\s]+(.*)", line, re.I) + if dm: + desc = dm.group(1).strip() + if current_port in interfaces and desc: + interfaces[current_port]["description"] = desc + continue + # MTU + mm = re.match(r"^\s*MTU[.:\s]+(\d+)", line, re.I) + if mm: + if current_port in interfaces: + interfaces[current_port]["mtu"] = int(mm.group(1)) + + return interfaces + + def get_interfaces_ip(self) -> Dict[str, Dict]: + """Return all configured IP addresses grouped by interface. + + Parses ``show ip interface`` output. On Smart Managed switches + only the management VLAN typically has an IP address. + + Example output block:: + + IP Address....................................... 192.168.0.239 + Subnet Mask...................................... 255.255.255.0 + Default Gateway.................................. 192.168.0.1 + """ + output = self._send_command("show ip interface") + interfaces_ip: Dict[str, Dict] = {} + + # Smart Managed switches expose a single management IP block; + # some firmwares add an "Interface" header for the routing VLAN. + current_iface = "vlan1" + ip_addr = "" + + for line in output.splitlines(): + line_s = line.strip() + + # Optional interface header + m = re.match( + r"(?:Interface|Routing Interface)[.:\s]+(vlan\d+|\d+/\d+)", + line_s, + re.I, + ) + if m: + current_iface = m.group(1).lower() + ip_addr = "" + continue + + m = re.match(r"IP Address[.:\s]+([\d.]+)", line_s, re.I) + if m: + ip_addr = m.group(1) + continue + + m = re.match(r"Subnet Mask[.:\s]+([\d.]+)", line_s, re.I) + if m and ip_addr: + try: + net = netaddr.IPNetwork(f"{ip_addr}/{m.group(1)}") + if current_iface not in interfaces_ip: + interfaces_ip[current_iface] = {} + interfaces_ip[current_iface].setdefault("ipv4", {})[ip_addr] = { + "prefix_length": net.prefixlen + } + except (netaddr.AddrFormatError, ValueError): + pass + ip_addr = "" + + return interfaces_ip + + def get_config( + self, + retrieve: str = "all", + full: bool = False, + sanitized: bool = False, + format: str = "text", + ) -> Dict[str, str]: + """Return running and/or startup configuration. + + Netgear Smart Managed does not support a candidate config; + that slot is always returned as an empty string. + """ + configs = {"running": "", "startup": "", "candidate": ""} + + if retrieve in ("all", "running"): + configs["running"] = self._send_command("show running-config") + + if retrieve in ("all", "startup"): + configs["startup"] = self._send_command("show startup-config") + + if sanitized: + configs = napalm_helpers.sanitize_configs(configs, C.CISCO_SANITIZE_FILTERS) + + return configs + + def get_arp_table(self, vrf: str = "") -> List[Dict]: + """Return the ARP table. + + Example ``show arp`` output:: + + IP Address MAC Address Interface Age (min) Type + --------- ---------------- --------- --------- ------- + 192.168.0.1 00:11:22:33:44:55 0/0 - Local + 192.168.0.100 00:aa:bb:cc:dd:ee 0/0 5 Dynamic + """ + output = self._send_command("show arp") + arp_table = [] + in_table = False + + for line in output.splitlines(): + line_s = line.strip() + if not line_s: + continue + if re.match(r"^-{3,}", line_s): + in_table = True + continue + if not in_table: + continue + + parts = line_s.split() + if len(parts) < 3: + continue + + ip_addr = parts[0] + mac_raw = parts[1] + interface = parts[2] + + try: + netaddr.IPAddress(ip_addr) + except (netaddr.AddrFormatError, ValueError): + continue + + try: + mac_addr = napalm_helpers.mac(mac_raw) + except Exception: + mac_addr = mac_raw + + arp_table.append( + { + "interface": interface, + "mac": mac_addr, + "ip": ip_addr, + "age": 0.0, + } + ) + + return arp_table + + def get_mac_address_table(self) -> List[Dict]: + """Return the MAC address table. + + Example ``show mac-addr-table`` output:: + + VLAN ID MAC Address Type Port + ------- ------------------ ---------- ------ + 1 00:11:22:33:44:55 Dynamic 0/1 + 1 ff:ff:ff:ff:ff:ff Management CPU + """ + output = self._send_command("show mac-addr-table") + mac_table = [] + in_table = False + + for line in output.splitlines(): + line_s = line.strip() + if not line_s: + continue + if re.match(r"^-{3,}", line_s): + in_table = True + continue + if not in_table: + continue + + parts = line_s.split() + if len(parts) < 4: + continue + + try: + vlan = int(parts[0]) + except ValueError: + continue + + mac_raw = parts[1] + entry_type = parts[2].lower() + interface = parts[3] + + try: + mac_addr = napalm_helpers.mac(mac_raw) + except Exception: + mac_addr = mac_raw + + mac_table.append( + { + "mac": mac_addr, + "interface": interface, + "vlan": vlan, + "static": entry_type in ("static", "management"), + "active": True, + "moves": None, + "last_move": None, + } + ) + + return mac_table + + def get_lldp_neighbors(self) -> Dict[str, List[Dict]]: + """Return a dict of LLDP neighbors keyed by local port. + + Example ``show lldp remote-device all`` output:: + + Local RemID Chassis ID Port ID System Name + Interface + --------- ----- -------------------- --------------- --------------- + 0/1 1 00:1a:2b:3c:4d:5e 0/1 router-1 + """ + neighbors: Dict[str, List[Dict]] = {} + for row in self._get_lldp_table(): + neighbors.setdefault(row["local_port"], []).append( + {"hostname": row["system_name"], "port": row["port_id"]} + ) + return neighbors + + def _get_lldp_table(self) -> List[Dict]: + """Parse ``show lldp remote-device all`` into a list of row dicts.""" + output = self._send_command("show lldp remote-device all") + rows: List[Dict] = [] + in_table = False + + for line in output.splitlines(): + line_s = line.strip() + if not line_s: + continue + if re.match(r"^-{4,}", line_s): + in_table = True + continue + if not in_table: + continue + + parts = line_s.split() + # Columns: Local Interface, RemID, Chassis ID, Port ID, System Name + if len(parts) < 3: + continue + if not re.match(r"^\d+/\d+$", parts[0]): + continue + + rows.append( + { + "local_port": parts[0], + "remote_id": parts[1] if len(parts) > 1 else "", + "remote_chassis_id": parts[2] if len(parts) > 2 else "", + "port_id": parts[3] if len(parts) > 3 else "", + "system_name": " ".join(parts[4:]) if len(parts) > 4 else "", + } + ) + + return rows + + def get_lldp_neighbors_detail(self, interface: str = "") -> Dict[str, List[Dict]]: + """Return detailed LLDP neighbor information. + + Fetches per-port detail via ``show lldp remote-device detail `` + and falls back to the summary table values for missing fields. + """ + details: Dict[str, List[Dict]] = {} + + for row in self._get_lldp_table(): + if interface and row["local_port"] != interface: + continue + + detail_out = self._send_command( + f"show lldp remote-device detail {row['local_port']}" + ) + parsed = self._parse_lldp_detail(detail_out) + + # Fill in from summary table if the detail command returned nothing + if not parsed["remote_chassis_id"]: + parsed["remote_chassis_id"] = row["remote_chassis_id"] + if not parsed["remote_port"]: + parsed["remote_port"] = row["port_id"] + if not parsed["remote_system_name"]: + parsed["remote_system_name"] = row["system_name"] + + details.setdefault(row["local_port"], []).append(parsed) + + return details + + @staticmethod + def _parse_lldp_detail(output: str) -> Dict: + """Parse ``show lldp remote-device detail`` block output.""" + result: Dict = { + "parent_interface": "", + "remote_port": "", + "remote_port_description": "", + "remote_chassis_id": "", + "remote_system_name": "", + "remote_system_description": "", + "remote_system_capab": [], + "remote_system_enable_capab": [], + } + + for line in output.splitlines(): + line_s = line.strip() + # Handle both dot-separator and colon-separator + m = re.match(r"([^.\n]+)\.{2,}\s*(.*)", line_s) + if not m: + m = re.match(r"([^:\n]+):\s*(.*)", line_s) + if not m: + continue + + key = m.group(1).strip().lower() + val = m.group(2).strip() + + if "chassis id" in key: + result["remote_chassis_id"] = val + elif "port id" in key and "desc" not in key: + result["remote_port"] = val + elif "port desc" in key or "port description" in key: + result["remote_port_description"] = val + elif "system name" in key: + result["remote_system_name"] = val + elif "system desc" in key or "system description" in key: + result["remote_system_description"] = val + elif "system cap" in key and "enabl" not in key: + result["remote_system_capab"] = [ + c.strip().lower() for c in val.split(",") if c.strip() + ] + elif "enabled cap" in key or ("system cap" in key and "enabl" in key): + result["remote_system_enable_capab"] = [ + c.strip().lower() for c in val.split(",") if c.strip() + ] + + return result + + def get_vlans(self) -> Dict[str, Dict]: + """Return VLAN information. + + Example ``show vlan`` output:: + + VLAN ID VLAN Name VLAN Type Interface(s) + ------- --------------- ---------- ----------------------------------- + 1 Default Default 0/1-0/8 + 10 Management Static 0/1, 0/3 + """ + output = self._send_command("show vlan") + vlans: Dict[str, Dict] = {} + current_id: Optional[str] = None + in_table = False + + for line in output.splitlines(): + line_s = line.strip() + if not line_s: + continue + if re.match(r"^-{3,}", line_s): + in_table = True + continue + if not in_table: + continue + + m = re.match(r"^(\d+)\s+(\S+)\s+\S+\s*(.*)", line_s) + if m: + current_id = str(int(m.group(1))) + vlans[current_id] = { + "name": m.group(2), + "interfaces": self._parse_vlan_ports(m.group(3).strip()), + } + elif current_id is not None and line_s: + # Continuation line with additional ports + vlans[current_id]["interfaces"].extend(self._parse_vlan_ports(line_s)) + + return vlans + + @staticmethod + def _parse_vlan_ports(ports_raw: str) -> List[str]: + """Parse Netgear VLAN port string and expand ranges. + + Input examples:: + "0/1-0/8" → ["0/1", "0/2", …, "0/8"] + "0/1, 0/3, 0/5" → ["0/1", "0/3", "0/5"] + "0/1-4" → ["0/1", "0/2", "0/3", "0/4"] + """ + interfaces: List[str] = [] + for token in ports_raw.split(","): + token = token.strip() + if not token: + continue + # Range: "0/1-0/8" (fully qualified) or "0/1-8" (short form) + range_m = re.match(r"^(\d+)/(\d+)-(?:(\d+)/)?(\d+)$", token) + if range_m: + slot = range_m.group(1) + start = int(range_m.group(2)) + end = int(range_m.group(4)) + interfaces.extend(f"{slot}/{i}" for i in range(start, end + 1)) + elif re.match(r"^\d+/\d+$", token): + interfaces.append(token) + return interfaces + + # ------------------------------------------------------------------ + # 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 commit_config(self, message: str = "", revert_in: Optional[int] = None) -> None: + """Apply the staged candidate configuration to the device and persist it. + + 1. Saves current running config as rollback backup. + 2. Enters config mode and pushes the candidate lines. + 3. Returns to exec mode. + 4. Persists with ``write memory``. + + :raises MergeConfigException / ReplaceConfigException: if no candidate + is staged or if lines are rejected by the device. + """ + if self._candidate_config is None: + raise MergeConfigException("No candidate configuration is staged.") + + ex_cls = ( + ReplaceConfigException + if self._candidate_mode == "replace" + else MergeConfigException + ) + + self._backup_config = self._send_command("show running-config") + + errors: List[str] = [] + try: + self._enter_config_mode() + errors = self._apply_config_lines(self._candidate_config) + finally: + self._exit_config_mode() + + if errors: + raise ex_cls("The following commands were rejected:\n" + "\n".join(errors)) + + self._save_config() + 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. + + :raises CommandErrorException: if :meth:`commit_config` has not been + called in this session. + """ + if self._backup_config is None: + raise CommandErrorException( + "No backup configuration available – " + "commit_config has not been called in this session." + ) + + current = self._send_command("show running-config") + rollback_cmds = NetgearSmartDriver._diff_to_commands(self._backup_config, current) + + if rollback_cmds: + try: + self._enter_config_mode() + self._apply_config_lines("\n".join(rollback_cmds)) + finally: + self._exit_config_mode() + self._save_config() + + self._backup_config = None + + @staticmethod + def _parse_config_blocks(config_text: str) -> Dict[str, List[str]]: + """Parse a running-config into ``context → [command lines]`` mapping. + + The special key ``"__global__"`` holds top-level commands. + Sub-mode blocks (``interface …``, ``vlan …``) are stored under the + opening line that introduces them. + """ + blocks: Dict[str, List[str]] = {"__global__": []} + ctx = "__global__" + + for line in config_text.splitlines(): + stripped = line.strip() + + if not stripped or stripped == "!": + ctx = "__global__" + continue + + if stripped.startswith("!"): + continue + + if re.match(r"^(interface|vlan)\s+\S+", stripped, re.I): + ctx = stripped + if ctx not in blocks: + blocks[ctx] = [] + continue + + if stripped.lower() == "exit": + ctx = "__global__" + continue + + blocks.setdefault(ctx, []).append(stripped) + + return blocks + + @staticmethod + def _negate_command(cmd: str) -> Optional[str]: + """Return the ``no`` form of *cmd*, or ``None`` if not known.""" + for kw in ( + "description", + "name", + "spanning-tree", + "lldp", + "ip address", + ): + if re.match(rf"^{re.escape(kw)}\b", cmd, re.I): + return f"no {kw}" + return None + + @staticmethod + def _diff_to_commands(backup: str, current: str) -> List[str]: + """Generate the minimal set of config commands to revert *current* to *backup*.""" + backup_blocks = NetgearSmartDriver._parse_config_blocks(backup) + current_blocks = NetgearSmartDriver._parse_config_blocks(current) + + cmds: List[str] = [] + all_ctxs = set(backup_blocks.keys()) | set(current_blocks.keys()) + + for ctx in sorted(all_ctxs): + backup_lines = set(backup_blocks.get(ctx, [])) + current_lines = set(current_blocks.get(ctx, [])) + + if backup_lines == current_lines: + continue + + in_block = ctx != "__global__" + if in_block: + cmds.append(ctx) + + # Lines added since backup → negate them + for line in current_lines - backup_lines: + neg = NetgearSmartDriver._negate_command(line) + if neg: + cmds.append(f" {neg}" if in_block else neg) + + # Lines removed since backup → restore them + for line in backup_lines - current_lines: + cmds.append(f" {line}" if in_block else line) + + if in_block: + cmds.append("exit") + + return cmds + + # ------------------------------------------------------------------ + # 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). + + Netgear Smart Managed switches do not expose fan, temperature, or + power rail data through the CLI; those fields are returned as assumed + healthy (``status: True``) with ``-1.0`` for numeric values. + + Tries ``show process cpu`` and ``show memory`` for utilisation data. + """ + cpu_out = self._send_command(["show process cpu", "show cpu"]) + mem_out = self._send_command(["show memory", "show memory cpu"]) + + cpu_pct = 0.0 + m = re.search(r"(?:CPU Utilization|utilization)[.:\s]+(\d+)%?", cpu_out, re.I) + if m: + cpu_pct = float(m.group(1)) + else: + # Some firmwares: "5 percent" + m = re.search(r"(\d+)\s+percent", cpu_out, re.I) + if m: + cpu_pct = float(m.group(1)) + + mem_pct = 0.0 + m = re.search(r"(?:Memory Utilization|utilization)[.:\s]+(\d+)%?", mem_out, re.I) + if m: + mem_pct = float(m.group(1)) + + return { + "fans": {}, + "temperature": {}, + "power": {}, + "cpu": {0: {"%usage": cpu_pct}}, + "memory": { + "available_ram": int((1 - mem_pct / 100) * 100), + "used_ram": int(mem_pct), + }, + } + + def get_interfaces_counters(self) -> Dict[str, Dict]: + """Return per-interface packet and byte counters. + + Parses ``show interface counters`` (or ``show interface ethernet 0/1``). + Each block starts with ``Interface................................ 0/1``. + """ + output = self._send_command( + ["show interface counters", "show interface ethernet all"] + ) + + def _int(s: str) -> int: + return int(s.replace(",", "")) if s.strip().replace(",", "").isdigit() else 0 + + counters: Dict[str, Dict] = {} + current_port: Optional[str] = None + current: Optional[Dict] = None + + _zero: Dict = { + "tx_errors": 0, "rx_errors": 0, + "tx_discards": 0, "rx_discards": 0, + "tx_octets": 0, "rx_octets": 0, + "tx_unicast_packets": 0, "rx_unicast_packets": 0, + "tx_multicast_packets": 0, "rx_multicast_packets": 0, + "tx_broadcast_packets": 0, "rx_broadcast_packets": 0, + } + + _mapping = { + "tx error": "tx_errors", + "rx error": "rx_errors", + "tx discard": "tx_discards", + "rx discard": "rx_discards", + "tx byte": "tx_octets", + "rx byte": "rx_octets", + "tx unicast": "tx_unicast_packets", + "rx unicast": "rx_unicast_packets", + "tx multicast": "tx_multicast_packets", + "rx multicast": "rx_multicast_packets", + "tx broadcast": "tx_broadcast_packets", + "rx broadcast": "rx_broadcast_packets", + } + + for line in output.splitlines(): + m = re.match(r"^\s*Interface[.:\s]+(\d+/\d+|ch\d+)", line, re.I) + if m: + if current_port and current: + counters[current_port] = current + current_port = m.group(1) + current = dict(_zero) + continue + + if current is None: + continue + + # Parse "Key.... value" or "Key: value" + kv = re.match(r"([^.\n]+)[.:\.]{2,}\s*(.*)", line.strip()) + if not kv: + kv = re.match(r"([^:\n]+):\s*(.*)", line.strip()) + if not kv: + continue + key = kv.group(1).strip().lower() + val = kv.group(2).strip() + + for pat, field in _mapping.items(): + if pat in key: + current[field] = _int(val) + break + + if current_port and current: + counters[current_port] = current + + return counters + + def get_users(self) -> Dict[str, Dict]: + """Return local user accounts. + + Example ``show users`` output:: + + User Access Level Session Timeout Password Strength + -------- ------------ --------------- ----------------- + admin Read/Write 5 Enabled + """ + output = self._send_command("show users") + users: Dict[str, Dict] = {} + in_table = False + + for line in output.splitlines(): + line_s = line.strip() + if not line_s: + continue + if re.match(r"^-{4,}", line_s): + in_table = True + continue + if not in_table: + continue + + parts = line_s.split() + if len(parts) < 2: + continue + + username = parts[0] + role = parts[1].lower() + level = 15 if "write" in role or "admin" in role else 1 + users[username] = {"level": level, "password": "", "sshkeys": []} + + return users + + def get_snmp_information(self) -> Dict: + """Return SNMP configuration. + + Parses ``show sysinfo`` for contact/location/chassis-ID and + ``show snmp`` for community strings. + """ + sysinfo = self._send_command("show sysinfo") + contact = self._parse_key_value(sysinfo, "System Contact") + location = self._parse_key_value(sysinfo, "System Location") + mac = self._parse_key_value(sysinfo, "Burned In MAC Address").replace("-", ":").upper() + + snmp_out = self._send_command("show snmp") + communities: Dict[str, Dict] = {} + + in_comm = False + for line in snmp_out.splitlines(): + line_s = line.strip() + if "Community" in line_s and ("Name" in line_s or "String" in line_s): + in_comm = True + continue + if not in_comm or not line_s: + continue + if re.match(r"^-{3,}", line_s): + continue + parts = line_s.split() + if len(parts) >= 2: + comm_name = parts[0] + # Access mode may be "Read Only" (2 tokens) or "Read Write" (2 tokens) + access = " ".join(parts[1:3]).lower() + mode = "rw" if "write" in access or "rw" in access else "ro" + communities[comm_name] = {"acl": "", "mode": mode} + + return { + "contact": contact, + "location": location, + "community": communities, + "chassis_id": mac, + } + + def get_ntp_servers(self) -> Dict[str, Dict]: + """Return configured NTP/SNTP servers from ``show sntp server``.""" + output = self._send_command(["show sntp server", "show ntp server"]) + servers: Dict[str, Dict] = {} + + for line in output.splitlines(): + line_s = line.strip() + # Key/value line: "SNTP Server Address..... 192.168.0.1" + # Use re.search so "SNTP Server Address" is also matched. + m = re.search( + r"Server\s+(?:Address|IP|Host)?[.:\s]+((?:\d{1,3}\.){3}\d{1,3}|[\w][\w.-]+\.\w+)", + line_s, + re.I, + ) + if m: + servers[m.group(1)] = {} + continue + # Plain IP in a table row (first column) + m = re.match(r"^((?:\d{1,3}\.){3}\d{1,3})", line_s) + if m: + servers[m.group(1)] = {} + + return servers + + def get_ntp_peers(self) -> Dict[str, Dict]: + """Return NTP peers (same as servers on Netgear Smart Managed).""" + return self.get_ntp_servers() + + def get_ntp_stats(self) -> List[Dict]: + """Netgear Smart Managed CLI does not expose per-peer NTP statistics.""" + return [] + + def get_optics(self) -> Dict: + """SFP DDM/DOM data is not accessible via the Smart Managed CLI.""" + return {} + + def get_ipv6_neighbors_table(self) -> List[Dict]: + """IPv6 ND table is not exposed via the Smart Managed CLI.""" + return [] + + def get_route_to( + self, + destination: str = "", + protocol: str = "", + longer: bool = False, + ) -> Dict[str, List[Dict]]: + """Return routing table entries. + + Smart Managed Pro switches support only a default gateway; + ``show ip route`` is parsed where available. + """ + output = self._send_command("show ip route") + routes: Dict[str, List[Dict]] = {} + proto_map = {"c": "connected", "s": "static"} + + for line in output.splitlines(): + line_s = line.strip() + if not line_s or re.match(r"^[Cc]ode", line_s): + continue + + m = re.match( + r"^([A-Za-z])\*?\s+([\d./]+)" + r"(?:\s+\[(\d+)/(\d+)\])?" + r"(?:\s+via\s+([\d.]+))?", + line_s, + ) + if not m: + continue + + code = m.group(1).lower() + prefix = m.group(2) + preference = int(m.group(3)) if m.group(3) else 0 + next_hop = m.group(5) or "" + prot = proto_map.get(code, code) + connected = code == "c" + + if destination and prefix != destination: + continue + if protocol and prot != protocol.lower(): + continue + + entry = { + "protocol": prot, + "current_active": True, + "last_active": False, + "age": -1, + "next_hop": next_hop if not connected else "", + "outgoing_interface": "", + "selected_next_hop": True, + "preference": preference, + "inactive_reason": "", + "routing_table": "global", + "protocol_attributes": {}, + } + routes.setdefault(prefix, []).append(entry) + + return routes + + 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: + """Ping *destination* from the device. + + Uses ``ping count ``. Source IP and packet size selection + are not supported by the Smart Managed CLI. + + :returns: NAPALM-standard ping result dict. + """ + cmd = f"ping {destination} count {count}" + output = self._send_command(cmd) + + if "Error" in output or "Invalid" in output: + return {"error": output.strip()} + + sent = received = 0 + m = re.search( + r"(\d+)\s+packets?\s+transmitted.*?(\d+)\s+(?:packets?\s+)?received", + output, + re.I | re.S, + ) + if m: + sent = int(m.group(1)) + received = int(m.group(2)) + + rtt_min = rtt_max = rtt_avg = 0.0 + m = re.search( + r"min/avg/max\s*=\s*([\d.]+)/([\d.]+)/([\d.]+)", + output, + re.I, + ) + if m: + rtt_min = float(m.group(1)) + rtt_avg = float(m.group(2)) + rtt_max = float(m.group(3)) + + return { + "success": { + "probes_sent": sent, + "packet_loss": sent - received, + "rtt_min": rtt_min, + "rtt_max": rtt_max, + "rtt_avg": rtt_avg, + "rtt_stddev": 0.0, + "results": [], + } + } + + def cli( + self, + commands: List[str], + encoding: str = "text", + ) -> Dict[str, Union[str, Dict[str, Any]]]: + """Execute a list of CLI commands and return their raw output.""" + if encoding != "text": + raise NotImplementedError( + f"Encoding '{encoding}' is not supported by this driver." + ) + result: Dict[str, Union[str, Dict[str, Any]]] = {} + for cmd in commands: + result[cmd] = self._send_command(cmd) + return result diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..1f0da5c --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,55 @@ +[build-system] +requires = ["setuptools>=64", "wheel"] +build-backend = "setuptools.build_meta" + +[project] +name = "napalm-netgear" +version = "0.1.0" +description = "NAPALM driver for Netgear switches (Plus/Smart)" +readme = "README.md" +license = { text = "Apache-2.0" } +requires-python = ">=3.8" +authors = [ + { name = "Christian Manivong" }, +] +classifiers = [ + "Topic :: Utilities", + "License :: OSI Approved :: Apache Software License", + "Programming Language :: Python :: 3", + "Programming Language :: Python :: 3.8", + "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", + "netmiko>=4.0.0", + "netaddr", + "py-netgear-plus>=0.6.4", +] + +[project.optional-dependencies] +dev = [ + "pytest", + "pytest-cov", + "black", + "ruff", +] + +[project.entry-points."napalm.drivers"] +netgear_smart = "napalm_netgear:NetgearSmartDriver" +netgear_plus = "napalm_netgear:NetgearPlusDriver" + +[project.urls] +Repository = "https://github.com/napalm-automation-community/napalm-netgear" + +[tool.setuptools.packages.find] +where = ["."] +include = ["napalm_netgear*"] + +[tool.ruff] +line-length = 100 +target-version = "py38" diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..dc8d5c2 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,3 @@ +napalm>=4.0.0 +netmiko>=4.0.0 +netaddr diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/unit/__init__.py b/tests/unit/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/unit/test_driver.py b/tests/unit/test_driver.py new file mode 100644 index 0000000..8c93e31 --- /dev/null +++ b/tests/unit/test_driver.py @@ -0,0 +1,644 @@ +"""Unit tests for NetgearSmartDriver — no real device required.""" + +import pytest +from unittest.mock import MagicMock, patch + +from napalm_netgear_plus.netgear_smart import NetgearSmartDriver + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + + +@pytest.fixture +def driver(): + """Return a driver instance with a mocked Netmiko connection.""" + with patch("napalm_netgear_plus.netgear_smart.ConnectHandler"): + drv = NetgearSmartDriver( + hostname="192.168.0.239", + username="admin", + password="password", + ) + drv.device = MagicMock() + drv.device.base_prompt = "(GS110TP) " + yield drv + + +# --------------------------------------------------------------------------- +# Sample CLI output fixtures +# --------------------------------------------------------------------------- + +SHOW_SYSINFO = """\ +System Description: GS110TP Gigabit Smart Managed Pro Switch +Machine Model: GS110TP +Machine Type: Ethernet Switch +Burned In MAC Address: C4:AD:34:AB:12:34 +System OID String: 1.3.6.1.4.1.4526.100.5.4 +System Uptime: 2 days 3 hrs 15 mins 42 secs +System Location: Server Room +System Contact: admin@example.com +System Name: myswitch +""" + +SHOW_VERSION = """\ +Software Version............................. 6.6.3 +Loader Version............................... 1.0.0.15 +Boot ROM Version............................. B1.0.0.15 +Hardware Version............................. V1 +Serial Number................................ 1FE2A0B1C2 +""" + +SHOW_PORT_ALL = """\ +Intf Type Admin Physical Physical Link Link LACP Actor Partner Admin + Mode Mode Status Status Trap Mode Port Port LACP + Priority Priority Timeout +--------- ------ ------- ---------- ---------- ------ ----- ----- ------ ------ ------- +0/1 Enable Auto 1G/Full Up Enable Disable 128 0 Long +0/2 Enable Auto - Down Enable Disable 128 0 Long +0/3 Disable Auto - Down Enable Disable 128 0 Long +""" + +SHOW_INTERFACE_ALL = """\ +Interface................................ 0/1 +Description.............................. uplink-to-router +MTU...................................... 1518 + +Interface................................ 0/2 +Description.............................. server-1 +MTU...................................... 1518 + +Interface................................ 0/3 +MTU...................................... 1518 +""" + +SHOW_IP_INTERFACE = """\ +IP Address....................................... 192.168.0.239 +Subnet Mask...................................... 255.255.255.0 +Default Gateway.................................. 192.168.0.1 +""" + +SHOW_ARP = """\ +IP Address MAC Address Interface Age (min) Type +--------- ------------------ --------- --------- ------- +192.168.0.1 00:11:22:33:44:55 0/0 - Local +192.168.0.100 aa:bb:cc:dd:ee:ff 0/0 5 Dynamic +""" + +SHOW_MAC = """\ +VLAN ID MAC Address Type Port +------- ------------------ ---------- ------ +1 00:11:22:33:44:55 Dynamic 0/1 +1 aa:bb:cc:dd:ee:ff Static 0/2 +1 ff:ff:ff:ff:ff:ff Management CPU +""" + +SHOW_VLAN = """\ +VLAN ID VLAN Name VLAN Type Interface(s) +------- --------------- ---------- ----------------------------------- +1 Default Default 0/1, 0/2, 0/3 +10 Management Static 0/1 +20 Servers Static 0/2, 0/3 +""" + +SHOW_LLDP_ALL = """\ +Local RemID Chassis ID Port ID System Name +Interface +--------- ----- -------------------- --------------- --------------- +0/1 1 00:1a:2b:3c:4d:5e Gi0/1 core-router +""" + +SHOW_LLDP_DETAIL = """\ +Chassis ID Subtype............................. MAC Address +Chassis ID................................... 00:1a:2b:3c:4d:5e +Port ID Subtype.............................. Interface Name +Port ID...................................... Gi0/1 +Port Description............................. WAN uplink +System Name.................................. core-router +System Description........................... Cisco IOS XE +System Capabilities.......................... Bridge, Router +Enabled Capabilities......................... Router +""" + +SHOW_SNMP = """\ +SNMP Community + +Community Name Access Mode Status +-------------- ----------- ------ +public Read Only Enable +private Read Write Enable +""" + +SHOW_USERS = """\ +User Access Level Session Timeout Password Strength +-------- ------------ --------------- ----------------- +admin Read/Write 5 Enabled +guest Read Only 5 Enabled +""" + +SHOW_SNTP = """\ +SNTP Server +SNTP Server Address....................... 192.168.0.1 +SNTP Server Port......................... 123 +""" + + +# --------------------------------------------------------------------------- +# Tests: get_facts +# --------------------------------------------------------------------------- + + +class TestGetFacts: + def test_returns_required_keys(self, driver): + driver._send_command = lambda cmd: SHOW_SYSINFO + driver._get_interface_list = lambda: [] + facts = driver.get_facts() + for key in ("vendor", "model", "hostname", "os_version", "serial_number", + "uptime", "interface_list", "fqdn"): + assert key in facts + + def test_vendor(self, driver): + driver._send_command = lambda cmd: ( + SHOW_SYSINFO if "sysinfo" in cmd else SHOW_VERSION + ) + driver._get_interface_list = lambda: [] + facts = driver.get_facts() + assert facts["vendor"] == "Netgear" + + def test_model(self, driver): + driver._send_command = lambda cmd: ( + SHOW_SYSINFO if "sysinfo" in cmd else SHOW_VERSION + ) + driver._get_interface_list = lambda: [] + facts = driver.get_facts() + assert facts["model"] == "GS110TP" + + def test_hostname(self, driver): + driver._send_command = lambda cmd: ( + SHOW_SYSINFO if "sysinfo" in cmd else SHOW_VERSION + ) + driver._get_interface_list = lambda: [] + facts = driver.get_facts() + assert facts["hostname"] == "myswitch" + + def test_os_version(self, driver): + driver._send_command = lambda cmd: ( + SHOW_SYSINFO if "sysinfo" in cmd else SHOW_VERSION + ) + driver._get_interface_list = lambda: [] + facts = driver.get_facts() + assert facts["os_version"] == "6.6.3" + + def test_serial_number(self, driver): + driver._send_command = lambda cmd: ( + SHOW_SYSINFO if "sysinfo" in cmd else SHOW_VERSION + ) + driver._get_interface_list = lambda: [] + facts = driver.get_facts() + assert facts["serial_number"] == "1FE2A0B1C2" + + def test_uptime_parsing(self, driver): + driver._send_command = lambda cmd: ( + SHOW_SYSINFO if "sysinfo" in cmd else SHOW_VERSION + ) + driver._get_interface_list = lambda: [] + facts = driver.get_facts() + expected = 2 * 86400 + 3 * 3600 + 15 * 60 + 42 + assert facts["uptime"] == float(expected) + + +# --------------------------------------------------------------------------- +# Tests: _parse_uptime_seconds +# --------------------------------------------------------------------------- + + +class TestParseUptimeSeconds: + def test_full(self): + assert NetgearSmartDriver._parse_uptime_seconds( + "2 days 3 hrs 15 mins 42 secs" + ) == float(2 * 86400 + 3 * 3600 + 15 * 60 + 42) + + def test_zero(self): + assert NetgearSmartDriver._parse_uptime_seconds( + "0 days 0 hrs 0 mins 0 secs" + ) == 0.0 + + def test_hours_only(self): + assert NetgearSmartDriver._parse_uptime_seconds("1 hrs 0 mins 0 secs") == 3600.0 + + +# --------------------------------------------------------------------------- +# Tests: _parse_key_value +# --------------------------------------------------------------------------- + + +class TestParseKeyValue: + def test_colon_separator(self): + output = "System Name: myswitch" + assert NetgearSmartDriver._parse_key_value(output, "System Name") == "myswitch" + + def test_dot_separator(self): + output = "Software Version............................. 6.6.3" + assert NetgearSmartDriver._parse_key_value(output, "Software Version") == "6.6.3" + + def test_missing_key_returns_empty(self): + assert NetgearSmartDriver._parse_key_value("some output", "Nonexistent Key") == "" + + +# --------------------------------------------------------------------------- +# Tests: get_interfaces +# --------------------------------------------------------------------------- + + +class TestGetInterfaces: + def _mock(self, driver): + def _send(cmd): + if "port" in cmd: + return SHOW_PORT_ALL + return SHOW_INTERFACE_ALL + driver._send_command = _send + + def test_all_ports_present(self, driver): + self._mock(driver) + ifaces = driver.get_interfaces() + assert "0/1" in ifaces + assert "0/2" in ifaces + assert "0/3" in ifaces + + def test_link_up(self, driver): + self._mock(driver) + assert driver.get_interfaces()["0/1"]["is_up"] is True + + def test_link_down(self, driver): + self._mock(driver) + assert driver.get_interfaces()["0/2"]["is_up"] is False + + def test_admin_disabled(self, driver): + self._mock(driver) + assert driver.get_interfaces()["0/3"]["is_enabled"] is False + + def test_speed_1g(self, driver): + self._mock(driver) + assert driver.get_interfaces()["0/1"]["speed"] == 1000.0 + + def test_description_from_interface_all(self, driver): + self._mock(driver) + assert driver.get_interfaces()["0/1"]["description"] == "uplink-to-router" + + +# --------------------------------------------------------------------------- +# Tests: get_interfaces_ip +# --------------------------------------------------------------------------- + + +class TestGetInterfacesIp: + def test_management_ip(self, driver): + driver._send_command = lambda cmd: SHOW_IP_INTERFACE + result = driver.get_interfaces_ip() + assert "vlan1" in result + assert "192.168.0.239" in result["vlan1"]["ipv4"] + + def test_prefix_length(self, driver): + driver._send_command = lambda cmd: SHOW_IP_INTERFACE + result = driver.get_interfaces_ip() + assert result["vlan1"]["ipv4"]["192.168.0.239"]["prefix_length"] == 24 + + +# --------------------------------------------------------------------------- +# Tests: get_arp_table +# --------------------------------------------------------------------------- + + +class TestGetArpTable: + def test_entry_count(self, driver): + driver._send_command = lambda cmd: SHOW_ARP + arp = driver.get_arp_table() + assert len(arp) == 2 + + def test_entry_keys(self, driver): + driver._send_command = lambda cmd: SHOW_ARP + entry = driver.get_arp_table()[0] + for k in ("interface", "mac", "ip", "age"): + assert k in entry + + def test_mac_normalised(self, driver): + driver._send_command = lambda cmd: SHOW_ARP + macs = {e["mac"] for e in driver.get_arp_table()} + # napalm_helpers.mac normalises to "AA:BB:CC:DD:EE:FF" format + assert any(":" in mac for mac in macs) + + +# --------------------------------------------------------------------------- +# Tests: get_mac_address_table +# --------------------------------------------------------------------------- + + +class TestGetMacAddressTable: + def test_entry_count(self, driver): + driver._send_command = lambda cmd: SHOW_MAC + mac_table = driver.get_mac_address_table() + assert len(mac_table) == 3 + + def test_static_flag(self, driver): + driver._send_command = lambda cmd: SHOW_MAC + statics = [e for e in driver.get_mac_address_table() if e["static"]] + # "Static" and "Management" rows are static + assert len(statics) == 2 + + def test_dynamic_flag(self, driver): + driver._send_command = lambda cmd: SHOW_MAC + dynamics = [e for e in driver.get_mac_address_table() if not e["static"]] + assert len(dynamics) == 1 + + +# --------------------------------------------------------------------------- +# Tests: get_vlans +# --------------------------------------------------------------------------- + + +class TestGetVlans: + def test_vlan_ids(self, driver): + driver._send_command = lambda cmd: SHOW_VLAN + vlans = driver.get_vlans() + assert set(vlans.keys()) == {"1", "10", "20"} + + def test_vlan_name(self, driver): + driver._send_command = lambda cmd: SHOW_VLAN + assert driver.get_vlans()["10"]["name"] == "Management" + + def test_vlan_interfaces(self, driver): + driver._send_command = lambda cmd: SHOW_VLAN + assert "0/1" in driver.get_vlans()["1"]["interfaces"] + + +# --------------------------------------------------------------------------- +# Tests: _parse_vlan_ports +# --------------------------------------------------------------------------- + + +class TestParseVlanPorts: + def test_single_port(self): + assert NetgearSmartDriver._parse_vlan_ports("0/1") == ["0/1"] + + def test_range(self): + assert NetgearSmartDriver._parse_vlan_ports("0/1-0/4") == [ + "0/1", "0/2", "0/3", "0/4" + ] + + def test_short_range(self): + assert NetgearSmartDriver._parse_vlan_ports("0/1-4") == [ + "0/1", "0/2", "0/3", "0/4" + ] + + def test_comma_separated(self): + result = NetgearSmartDriver._parse_vlan_ports("0/1, 0/3, 0/5") + assert result == ["0/1", "0/3", "0/5"] + + +# --------------------------------------------------------------------------- +# Tests: get_lldp_neighbors +# --------------------------------------------------------------------------- + + +class TestGetLldpNeighbors: + def test_neighbor_present(self, driver): + driver._send_command = lambda cmd: SHOW_LLDP_ALL + neighbors = driver.get_lldp_neighbors() + assert "0/1" in neighbors + + def test_neighbor_hostname(self, driver): + driver._send_command = lambda cmd: SHOW_LLDP_ALL + neighbors = driver.get_lldp_neighbors() + assert neighbors["0/1"][0]["hostname"] == "core-router" + + +# --------------------------------------------------------------------------- +# Tests: _parse_lldp_detail +# --------------------------------------------------------------------------- + + +class TestParseLldpDetail: + def test_chassis_id(self): + r = NetgearSmartDriver._parse_lldp_detail(SHOW_LLDP_DETAIL) + assert r["remote_chassis_id"] == "00:1a:2b:3c:4d:5e" + + def test_port_id(self): + r = NetgearSmartDriver._parse_lldp_detail(SHOW_LLDP_DETAIL) + assert r["remote_port"] == "Gi0/1" + + def test_system_name(self): + r = NetgearSmartDriver._parse_lldp_detail(SHOW_LLDP_DETAIL) + assert r["remote_system_name"] == "core-router" + + def test_capabilities(self): + r = NetgearSmartDriver._parse_lldp_detail(SHOW_LLDP_DETAIL) + assert "bridge" in r["remote_system_capab"] + assert "router" in r["remote_system_capab"] + + +# --------------------------------------------------------------------------- +# Tests: get_snmp_information +# --------------------------------------------------------------------------- + + +class TestGetSnmpInformation: + def test_communities(self, driver): + def _send(cmd): + if "sysinfo" in cmd: + return SHOW_SYSINFO + return SHOW_SNMP + driver._send_command = _send + snmp = driver.get_snmp_information() + assert "public" in snmp["community"] + assert snmp["community"]["public"]["mode"] == "ro" + assert snmp["community"]["private"]["mode"] == "rw" + + def test_location(self, driver): + def _send(cmd): + if "sysinfo" in cmd: + return SHOW_SYSINFO + return SHOW_SNMP + driver._send_command = _send + snmp = driver.get_snmp_information() + assert snmp["location"] == "Server Room" + + +# --------------------------------------------------------------------------- +# Tests: get_users +# --------------------------------------------------------------------------- + + +class TestGetUsers: + def test_users_present(self, driver): + driver._send_command = lambda cmd: SHOW_USERS + users = driver.get_users() + assert "admin" in users + assert "guest" in users + + def test_admin_level(self, driver): + driver._send_command = lambda cmd: SHOW_USERS + assert driver.get_users()["admin"]["level"] == 15 + + def test_guest_level(self, driver): + driver._send_command = lambda cmd: SHOW_USERS + assert driver.get_users()["guest"]["level"] == 1 + + +# --------------------------------------------------------------------------- +# Tests: get_ntp_servers +# --------------------------------------------------------------------------- + + +class TestGetNtpServers: + def test_server_found(self, driver): + driver._send_command = lambda cmd: SHOW_SNTP + servers = driver.get_ntp_servers() + assert "192.168.0.1" in servers + + +# --------------------------------------------------------------------------- +# Tests: config management +# --------------------------------------------------------------------------- + + +class TestConfigManagement: + def test_load_merge_from_string(self, driver): + driver.load_merge_candidate(config="interface 0/1\n description test\nexit") + assert driver._candidate_config is not None + assert driver._candidate_mode == "merge" + + def test_load_replace_from_string(self, driver): + driver.load_replace_candidate(config="! running config\nvlan 10\n name test\nexit") + assert driver._candidate_mode == "replace" + + def test_discard_clears_candidate(self, driver): + driver.load_merge_candidate(config="vlan 99\n name test\nexit") + driver.discard_config() + assert driver._candidate_config is None + assert driver._candidate_mode is None + + def test_has_pending_commit_false(self, driver): + assert driver.has_pending_commit() is False + + def test_has_pending_commit_true(self, driver): + driver.load_merge_candidate(config="vlan 99\n name test\nexit") + assert driver.has_pending_commit() is True + + def test_compare_config_merge(self, driver): + driver.load_merge_candidate(config="vlan 99\n name test\nexit") + diff = driver.compare_config() + assert diff.startswith("+") + + def test_compare_config_empty_when_no_candidate(self, driver): + assert driver.compare_config() == "" + + def test_load_merge_from_file(self, driver, tmp_path): + cfg_file = tmp_path / "candidate.txt" + cfg_file.write_text("vlan 99\n name test\nexit\n") + driver.load_merge_candidate(filename=str(cfg_file)) + assert "vlan 99" in driver._candidate_config + + def test_load_merge_raises_without_input(self, driver): + from napalm.base.exceptions import MergeConfigException + with pytest.raises(MergeConfigException): + driver.load_merge_candidate() + + def test_load_replace_raises_without_input(self, driver): + from napalm.base.exceptions import ReplaceConfigException + with pytest.raises(ReplaceConfigException): + driver.load_replace_candidate() + + def test_rollback_raises_without_backup(self, driver): + from napalm.base.exceptions import CommandErrorException + with pytest.raises(CommandErrorException): + driver.rollback() + + +# --------------------------------------------------------------------------- +# Tests: _parse_config_blocks / _diff_to_commands +# --------------------------------------------------------------------------- + + +class TestConfigDiff: + BACKUP = """\ +vlan database +vlan 10 + name Management +exit +interface 0/1 + description uplink +exit +""" + CURRENT = """\ +vlan database +vlan 10 + name Management +exit +interface 0/1 + description changed +exit +""" + + def test_diff_detects_change(self): + cmds = NetgearSmartDriver._diff_to_commands( + TestConfigDiff.BACKUP, TestConfigDiff.CURRENT + ) + # Should contain a "no description" and "description uplink" line + joined = " ".join(cmds) + assert "description" in joined + + +# --------------------------------------------------------------------------- +# Tests: get_config +# --------------------------------------------------------------------------- + + +class TestGetConfig: + def test_running_retrieved(self, driver): + driver._send_command = lambda cmd: "! running config" + cfg = driver.get_config(retrieve="running") + assert cfg["running"] == "! running config" + assert cfg["startup"] == "" + + def test_startup_retrieved(self, driver): + driver._send_command = lambda cmd: "! startup config" + cfg = driver.get_config(retrieve="startup") + assert cfg["startup"] == "! startup config" + assert cfg["running"] == "" + + def test_candidate_always_empty(self, driver): + driver._send_command = lambda cmd: "" + cfg = driver.get_config() + assert cfg["candidate"] == "" + + +# --------------------------------------------------------------------------- +# Tests: is_alive +# --------------------------------------------------------------------------- + + +class TestIsAlive: + def test_alive_when_connected(self, driver): + driver.device.remote_conn.transport.is_active.return_value = True + assert driver.is_alive() == {"is_alive": True} + + def test_dead_when_no_device(self, driver): + driver.device = None + assert driver.is_alive() == {"is_alive": False} + + +# --------------------------------------------------------------------------- +# Tests: cli +# --------------------------------------------------------------------------- + + +class TestCli: + def test_returns_output_per_command(self, driver): + driver._send_command = lambda cmd: f"output of {cmd}" + result = driver.cli(["show sysinfo", "show version"]) + assert result["show sysinfo"] == "output of show sysinfo" + assert result["show version"] == "output of show version" + + def test_raises_on_non_text_encoding(self, driver): + with pytest.raises(NotImplementedError): + driver.cli(["show sysinfo"], encoding="json")