feat!: configure_snmp_agent with the caller's values replaces fix_snmp (0.3.0) #11
No files matched your search
@@ -36,7 +36,7 @@ import netaddr
|
||||
from netmiko import ConnectHandler
|
||||
from netmiko.exceptions import NetmikoAuthenticationException, NetmikoTimeoutException
|
||||
|
||||
from napalm_device_types import ConfigLifecycleMixin, FingerprintRule, SwitchDriver
|
||||
from napalm_device_types import ConfigLifecycleMixin, FingerprintRule, SnmpAgentMixin, SwitchDriver
|
||||
from napalm.base import helpers as napalm_helpers
|
||||
from napalm.base.exceptions import (
|
||||
CommandErrorException,
|
||||
@@ -49,7 +49,7 @@ from napalm.base.netmiko_helpers import netmiko_args
|
||||
import napalm.base.constants as C
|
||||
|
||||
|
||||
class NetgearSmartDriver(ConfigLifecycleMixin, SwitchDriver):
|
||||
class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
|
||||
"""NAPALM driver for Netgear Smart Managed switches (GS-series)."""
|
||||
|
||||
VENDOR = "Netgear"
|
||||
@@ -74,7 +74,10 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SwitchDriver):
|
||||
#: CPU is text ("5 Secs ( 3.2%) 60 Secs ( …"); the five-second figure counts.
|
||||
SNMP_HEALTH_PROFILE: ClassVar[Dict[str, Any]] = {
|
||||
"uptime": {"oids": ["1.3.6.1.2.1.1.3.0"]},
|
||||
"cpu": {"oids": ["1.3.6.1.4.1.4526.100.1.1.4.9.0"], "pattern": r"5\s+Secs\s*\(\s*([\d.]+)\s*%\)"},
|
||||
"cpu": {
|
||||
"oids": ["1.3.6.1.4.1.4526.100.1.1.4.9.0"],
|
||||
"pattern": r"5\s+Secs\s*\(\s*([\d.]+)\s*%\)",
|
||||
},
|
||||
"memory": {
|
||||
"total": {"oids": ["1.3.6.1.4.1.4526.100.1.1.4.2.0"]},
|
||||
"free": {"oids": ["1.3.6.1.4.1.4526.100.1.1.4.1.0"]},
|
||||
@@ -1877,50 +1880,61 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SwitchDriver):
|
||||
|
||||
return SNMPConfigDict(running=True, community=community, port=161, version="2c")
|
||||
|
||||
def run_device_action(self, action: str) -> Dict:
|
||||
"""Execute a named action on the switch."""
|
||||
if action == "fix_snmp":
|
||||
return self._action_fix_snmp()
|
||||
raise NotImplementedError(f"Unknown action: {action!r}")
|
||||
# ------------------------------------------------------------------
|
||||
# SNMP agent (napalm-device-types SnmpAgentMixin): the caller's values
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def _action_fix_snmp(self) -> Dict:
|
||||
"""Enable SNMP with community 'public' (read-only) on the Netgear Smart switch."""
|
||||
def _install_snmp_agent(self) -> Dict[str, Any]:
|
||||
return {"ok": True, "output": "the SNMP agent is built in"}
|
||||
|
||||
def _write_snmp_agent_config(self, settings: Dict[str, Any]) -> Dict[str, Any]:
|
||||
if self._detect_cli_v7():
|
||||
# The "v7" Cisco-like CLI (e.g. GS110TPv3 7.x) has no
|
||||
# "snmp-server"/"show snmp" commands at all — SNMP can only be
|
||||
# enabled via the web UI on this firmware generation.
|
||||
# The "v7" Cisco-like CLI (e.g. GS110TPv3 7.x) has no "snmp-server"
|
||||
# commands at all: SNMP is set in the web UI on this firmware.
|
||||
return {
|
||||
"success": False,
|
||||
"output": (
|
||||
"SNMP cannot be configured via CLI on this switch's firmware "
|
||||
"(no 'snmp-server' commands available). Enable SNMP manually "
|
||||
"via the switch's web UI, or ignore this warning."
|
||||
),
|
||||
"ok": False,
|
||||
"output": "this firmware has no SNMP commands in its CLI; "
|
||||
"SNMP is set in its web UI",
|
||||
}
|
||||
|
||||
lines: list = []
|
||||
|
||||
community = settings["community"]
|
||||
if not re.match(r'^[^\s"#]+$', community):
|
||||
raise ValueError("the community must be one word without quotes or #")
|
||||
lines = [f"snmp-server community {community} ro"]
|
||||
for key in ("contact", "location"):
|
||||
value = settings.get(key, "")
|
||||
if value:
|
||||
if '"' in value or "\n" in value:
|
||||
raise ValueError(f"{key} must be one line without double quotes")
|
||||
lines.append(f'snmp-server {key} "{value}"')
|
||||
self._enter_config_mode()
|
||||
# Netgear Smart CLI: snmp-server community <name> ro
|
||||
self._send_command("snmp-server community public ro")
|
||||
lines.append("[config] SNMP community 'public' (read-only) configured.")
|
||||
self._exit_config_mode()
|
||||
self._save_config()
|
||||
lines.append("[config] Configuration saved.")
|
||||
|
||||
# Verify
|
||||
try:
|
||||
out = self._send_command("show snmp")
|
||||
success = "public" in out
|
||||
except Exception:
|
||||
success = False
|
||||
out = ""
|
||||
if success:
|
||||
lines.append("[ok] SNMP active with community 'public'.")
|
||||
else:
|
||||
lines.append(f"[warn] Verification failed: {out[:100]}")
|
||||
errors = self._apply_config_lines("\n".join(lines))
|
||||
finally:
|
||||
self._exit_config_mode()
|
||||
if errors:
|
||||
return {"ok": False, "output": "the switch refused:\n" + "\n".join(errors)}
|
||||
return {"ok": True, "output": "community (read-only), contact and location set"}
|
||||
|
||||
return {"success": success, "output": "\n".join(lines)}
|
||||
@staticmethod
|
||||
def _allow_snmp_sources(settings: Dict[str, Any]) -> Dict[str, Any]:
|
||||
sources = ", ".join(settings.get("allowed_sources", [])) or "none given"
|
||||
note = f"no host firewall on the switch; sources not enforced ({sources})"
|
||||
return {"ok": True, "output": note}
|
||||
|
||||
def _start_snmp_agent(self) -> Dict[str, Any]:
|
||||
self._save_config()
|
||||
return {"ok": True, "output": "configuration saved"}
|
||||
|
||||
def configure_snmp_agent(self, settings: Dict[str, Any]) -> Dict[str, Any]:
|
||||
"""The template's steps, then a check that the switch lists the community."""
|
||||
result = SnmpAgentMixin.configure_snmp_agent(self, settings)
|
||||
if not result["success"]:
|
||||
return result
|
||||
listed = self._send_command("show snmp")
|
||||
if re.search(rf"(?<!\S){re.escape(settings['community'])}(?!\S)", listed):
|
||||
return result
|
||||
missing = f"[verify] the switch does not list the community: {listed[:200]}"
|
||||
return {**result, "success": False, "output": f"{result['output']}\n{missing}"}
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# PoE
|
||||
|
||||
+2
-1
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
|
||||
|
||||
[project]
|
||||
name = "napalm-netgear"
|
||||
version = "0.2.0"
|
||||
version = "0.3.0"
|
||||
description = "NAPALM driver for Netgear switches (Plus/Smart)"
|
||||
readme = "README.md"
|
||||
license = { text = "Apache-2.0" }
|
||||
@@ -26,6 +26,7 @@ classifiers = [
|
||||
]
|
||||
dependencies = [
|
||||
"napalm>=4.0.0",
|
||||
"napalm-device-types>=4.13.0",
|
||||
"netmiko>=4.0.0",
|
||||
"paramiko>=5.0.0", # CVE-2026-44405
|
||||
"netaddr",
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
napalm>=4.0.0
|
||||
napalm-device-types>=4.13.0
|
||||
netmiko>=4.0.0
|
||||
netaddr
|
||||
@@ -0,0 +1,77 @@
|
||||
"""The SNMP agent through napalm-device-types' configure_snmp_agent (netOrk #1105).
|
||||
|
||||
fix_snmp hard-coded community "public" and sent the config-mode command through
|
||||
_send_command, which waits for the exec prompt and so ran into its timeout in
|
||||
config mode (#6). The caller's values go through _apply_config_lines now, which
|
||||
waits for any prompt and reports a rejected line.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
import pytest
|
||||
|
||||
from napalm_netgear.netgear_smart import NetgearSmartDriver
|
||||
|
||||
SETTINGS = {"community": "c0mm", "allowed_sources": [], "contact": "ops", "location": "rack 1"}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def driver():
|
||||
d = NetgearSmartDriver("10.0.0.30", "admin", "pw")
|
||||
d.device = MagicMock()
|
||||
d.applied: list[str] = []
|
||||
d._detect_cli_v7 = lambda: False
|
||||
d._enter_config_mode = MagicMock()
|
||||
d._exit_config_mode = MagicMock()
|
||||
d._save_config = MagicMock()
|
||||
d._apply_config_lines = lambda text: d.applied.extend(text.splitlines()) or []
|
||||
d._send_command = MagicMock(
|
||||
return_value="Community-String Community-Access\nc0mm Read Only\n"
|
||||
)
|
||||
return d
|
||||
|
||||
|
||||
def test_community_contact_and_location_through_the_config_helper(driver):
|
||||
result = driver.configure_snmp_agent(SETTINGS)
|
||||
|
||||
assert result["success"] is True
|
||||
assert driver.applied == [
|
||||
"snmp-server community c0mm ro",
|
||||
'snmp-server contact "ops"',
|
||||
'snmp-server location "rack 1"',
|
||||
]
|
||||
driver._exit_config_mode.assert_called_once()
|
||||
driver._save_config.assert_called_once()
|
||||
|
||||
|
||||
def test_config_mode_is_left_even_when_a_line_is_refused(driver):
|
||||
driver._apply_config_lines = lambda text: [" 'snmp-server community c0mm ro': Invalid input"]
|
||||
|
||||
result = driver.configure_snmp_agent(SETTINGS)
|
||||
|
||||
assert result["success"] is False
|
||||
driver._exit_config_mode.assert_called_once()
|
||||
driver._save_config.assert_not_called()
|
||||
|
||||
|
||||
def test_v7_firmware_says_it_has_no_snmp_cli(driver):
|
||||
driver._detect_cli_v7 = lambda: True
|
||||
|
||||
result = driver.configure_snmp_agent(SETTINGS)
|
||||
|
||||
assert result["success"] is False
|
||||
assert "web UI" in result["output"] and "warning" not in result["output"]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("community", ["two words", 'q"uote', ""])
|
||||
def test_a_community_that_would_break_the_cli_is_refused(driver, community):
|
||||
result = driver.configure_snmp_agent({"community": community, "allowed_sources": []})
|
||||
assert result["success"] is False
|
||||
assert driver.applied == []
|
||||
|
||||
|
||||
def test_the_old_action_is_gone():
|
||||
assert not hasattr(NetgearSmartDriver, "run_device_action")
|
||||
assert not hasattr(NetgearSmartDriver, "_action_fix_snmp")
|
||||
Reference in new issue
Block a user