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