From c2f0f7c20aa43d7de48f7e3a5818c3d923557f64 Mon Sep 17 00:00:00 2001 From: Christian Manivong Date: Sat, 17 Oct 2026 08:36:52 +0200 Subject: [PATCH] feat!: configure_snmp_agent with the caller's values replaces fix_snmp (0.3.0) fix_snmp hard-coded community "public" and found the management VLAN for the source interface with a substring match, so 10.0.0.1 matched 10.0.0.10 and a DNS name never matched (#3). The driver implements napalm-device-types 4.13's configure_snmp_agent steps: the caller's community (read-only, viewDefault), the source interface set to the VLAN whose address is exactly the listen address or the management address, configuration saved, and the community checked in show snmp community. Contact and location are reported as not set; there is no verified CLI for them on this switch. A value that would break the CLI is refused. The requirements now name napalm-device-types 4.13 in both files. BREAKING CHANGE: run_device_action is removed. Part of #3 and #4. --- napalm_tplink_jetstream/tplink_jetstream.py | 126 +++++++++----------- pyproject.toml | 4 +- requirements.txt | 2 +- tests/unit/test_snmp_agent.py | 84 +++++++++++++ 4 files changed, 142 insertions(+), 74 deletions(-) create mode 100644 tests/unit/test_snmp_agent.py diff --git a/napalm_tplink_jetstream/tplink_jetstream.py b/napalm_tplink_jetstream/tplink_jetstream.py index 8825dd8..9ce92c5 100644 --- a/napalm_tplink_jetstream/tplink_jetstream.py +++ b/napalm_tplink_jetstream/tplink_jetstream.py @@ -27,7 +27,7 @@ import netaddr from netmiko import ConnectHandler from netmiko.exceptions import NetmikoTimeoutException, NetmikoAuthenticationException -from napalm_device_types import ConfigLifecycleMixin, FingerprintRule, SwitchDriver +from napalm_device_types import ConfigLifecycleMixin, FingerprintRule, SnmpAgentMixin, SwitchDriver from napalm_device_types.models import InterfaceConfigDict, VlanConfigDict from napalm.base import helpers as napalm_helpers from napalm.base.exceptions import ( @@ -41,7 +41,7 @@ from napalm.base.netmiko_helpers import netmiko_args import napalm.base.constants as C -class TPLinkJetstreamDriver(ConfigLifecycleMixin, SwitchDriver): +class TPLinkJetstreamDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver): """NAPALM driver for TP-Link Jetstream managed switches.""" VENDOR = "TP-Link" @@ -1935,81 +1935,65 @@ class TPLinkJetstreamDriver(ConfigLifecycleMixin, SwitchDriver): """Return device warnings. Currently delegates to SNMP detection only.""" return [] - 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 _detect_management_vlan(self) -> int | None: - """Detect which VLAN interface carries the management IP (self.hostname).""" - try: - out = self._send_command("show ip interface") - import re as _re - # Match "VLAN is up" followed by IP line containing self.hostname - vlan = None - current_vlan = None - for line in out.splitlines(): - m = _re.match(r"VLAN(\d+)\s+is", line, _re.IGNORECASE) - if m: - current_vlan = int(m.group(1)) - if current_vlan and self.hostname in line and "IP address" in line: - vlan = current_vlan - break - return vlan - except Exception: - return None + def _management_vlan(self, address: str) -> int | None: + """The VLAN interface whose IP address is exactly *address*.""" + current = None + pattern = re.compile(rf"IP address[:\s]+{re.escape(address)}(?![\d.])", re.I) + for line in self._send_command("show ip interface").splitlines(): + m = re.match(r"VLAN(\d+)\s+is", line, re.I) + if m: + current = int(m.group(1)) + elif current is not None and pattern.search(line): + return current + return None - def _action_fix_snmp(self) -> dict: - """Enable SNMP and configure community 'public' on the switch. + def _install_snmp_agent(self) -> dict: + return {"ok": True, "output": "the SNMP agent is built in"} - TP-Link JetStream uses CLI commands — SNMP is built-in, no install needed. - Also sets snmp-server source-interface to the management VLAN so that - SNMP responses are sent from the correct interface. - """ - lines: list = [] - - # Detect management VLAN (the VLAN interface carrying this device's IP) - mgmt_vlan = self._detect_management_vlan() - if mgmt_vlan: - lines.append(f"[detect] Management VLAN: {mgmt_vlan}") - else: - lines.append("[detect] Could not auto-detect management VLAN.") - - # Configure SNMP in config mode + def _write_snmp_agent_config(self, settings: dict) -> dict: + community = settings["community"] + if not re.match(r'^[^\s"#]+$', community): + raise ValueError("the community must be one word without quotes or #") + vlan = self._management_vlan(settings.get("listen") or self.hostname) conf_prompt = self._conf_prompt() self._enter_config_mode() - self.device.send_command("snmp-server", expect_string=conf_prompt, read_timeout=self.timeout) - lines.append("[config] SNMP enabled.") + try: + for command in ( + "snmp-server", + f'snmp-server community "{community}" read-only "viewDefault"', + *([f"snmp-server source-interface vlan {vlan}"] if vlan else []), + ): + self.device.send_command( + command, expect_string=conf_prompt, read_timeout=self.timeout + ) + finally: + self._exit_config_mode() + notes = [f"source interface vlan {vlan}" if vlan else "management VLAN not found"] + if settings.get("contact") or settings.get("location"): + notes.append("contact and location not set (no verified CLI on this switch)") + return {"ok": True, "output": "community set (read-only); " + "; ".join(notes)} - # Community: TP-Link syntax is snmp-server community "" read-only "" - self.device.send_command( - 'snmp-server community "public" read-only "viewDefault"', - expect_string=conf_prompt, - read_timeout=self.timeout, - ) - lines.append("[config] Community 'public' (read-only, viewDefault) configured.") + @staticmethod + def _allow_snmp_sources(settings: dict) -> dict: + 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} - # Set source-interface so responses come from the management VLAN - if mgmt_vlan: - self.device.send_command( - f"snmp-server source-interface vlan {mgmt_vlan}", - expect_string=conf_prompt, - read_timeout=self.timeout, - ) - lines.append(f"[config] Source-interface set to vlan {mgmt_vlan}.") - - # Exit config mode and save - self._exit_config_mode() + def _start_snmp_agent(self) -> dict: self._save_config() - lines.append("[config] Configuration saved.") + return {"ok": True, "output": "configuration saved"} - # Verify - out = self._send_command("show snmp community") - success = "public" in out and "disabled" not in out.lower() - if success: - lines.append("[ok] SNMP is active with community 'public'.") - else: - lines.append(f"[warn] Verification failed: {out[:100]}") - - return {"success": success, "output": "\n".join(lines)} + def configure_snmp_agent(self, settings: dict) -> dict: + """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 community") + if re.search(rf"(?=4.0.0", - "napalm_device_types>=0.2.0", + "napalm_device_types>=4.13.0", "netmiko>=4.0.0", "paramiko>=5.0.0", # CVE-2026-44405 "netaddr", diff --git a/requirements.txt b/requirements.txt index a6587eb..b34b5f9 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,4 +1,4 @@ napalm>=4.0.0 -napalm_device_types>=0.1.0 +napalm_device_types>=4.13.0 netmiko>=4.0.0 netaddr diff --git a/tests/unit/test_snmp_agent.py b/tests/unit/test_snmp_agent.py new file mode 100644 index 0000000..c1e8fce --- /dev/null +++ b/tests/unit/test_snmp_agent.py @@ -0,0 +1,84 @@ +"""The SNMP agent through napalm-device-types' configure_snmp_agent (netOrk #1105). + +fix_snmp hard-coded community "public" and found the management VLAN by a +substring match (10.0.0.1 matched 10.0.0.10, #3). The community is the caller's +now, and the source interface is the VLAN that carries exactly the management +address. +""" + +from __future__ import annotations + +from unittest.mock import MagicMock + +import pytest + +from napalm_tplink_jetstream.tplink_jetstream import TPLinkJetstreamDriver + +SETTINGS = {"community": "c0mm", "allowed_sources": ["10.7.224.20"], "contact": "ops"} + +SHOW_IP_INTERFACE = """\ +VLAN1 is up, line protocol is up + Primary IP address: 10.0.0.10/24 +VLAN8 is up, line protocol is up + Primary IP address: 10.0.0.1/24 +""" + + +@pytest.fixture +def driver(): + d = TPLinkJetstreamDriver("10.0.0.1", "admin", "pw") + d.device = MagicMock() + d.sent: list[str] = [] + d.device.send_command.side_effect = lambda cmd, **kw: d.sent.append(cmd) or "" + d._conf_prompt = lambda: r"\(config\)#" + d._enter_config_mode = MagicMock() + d._exit_config_mode = MagicMock() + d._save_config = MagicMock() + d._send_command = MagicMock( + side_effect=lambda cmd: SHOW_IP_INTERFACE + if cmd == "show ip interface" + else "Index Community-Name Access\n1 c0mm read-only\n" + ) + return d + + +def test_the_callers_community(driver): + result = driver.configure_snmp_agent(SETTINGS) + + assert result["success"] is True + assert 'snmp-server community "c0mm" read-only "viewDefault"' in driver.sent + assert not any("public" in c for c in driver.sent) + driver._save_config.assert_called_once() + + +def test_the_source_interface_is_the_vlan_of_exactly_the_management_address(driver): + driver.configure_snmp_agent(SETTINGS) + + assert "snmp-server source-interface vlan 8" in driver.sent + + +def test_contact_and_location_are_reported_as_not_set(driver): + steps = {s["step"]: s for s in driver.configure_snmp_agent(SETTINGS)["steps"]} + + assert "contact" in steps["configure"]["output"] + + +def test_a_community_the_switch_does_not_list_fails(driver): + driver._send_command.side_effect = lambda cmd: ( + SHOW_IP_INTERFACE if cmd == "show ip interface" else "1 public read-only\n" + ) + + assert driver.configure_snmp_agent(SETTINGS)["success"] is False + + +@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.sent == [] + + +def test_the_old_action_is_gone(): + for name in ("run_device_action", "_action_fix_snmp", "_detect_management_vlan"): + assert not hasattr(TPLinkJetstreamDriver, name), name -- 2.54.0