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

initial commit

Browse Source
This commit is contained in:
Christian Manivong
2026-05-29 09:13:59 +02:00
commit a8ce5a8033
10 changed files with 3058 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+28
View File
@@ -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
README.md
+112
View File
@@ -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).
napalm_netgear/__init__.py
+6
View File
@@ -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"]
napalm_netgear/netgear_plus.py
+787
View File
@@ -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 ``<input class="portName" value="...">``
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'<input[^>]+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: <span vid-4>...<i ...></i>24</span><span vnm-4 ...>Guest</span>
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>',
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 <span pvid-str ...>1*, 24, 25,</span> 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'<span\s[^>]*pvid-str[^>]*>([^<]+)</span>')
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
napalm_netgear/netgear_smart.py
+1423
View File
File diff suppressed because it is too large Load Diff
pyproject.toml
+55
View File
@@ -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"
requirements.txt
+3
View File
@@ -0,0 +1,3 @@
napalm>=4.0.0
netmiko>=4.0.0
netaddr
tests/__init__.py
View File
tests/unit/__init__.py
View File
tests/unit/test_driver.py
+644
View File
@@ -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")