diff --git a/napalm_procurve/procurve.py b/napalm_procurve/procurve.py index 24750c7..6df66fa 100644 --- a/napalm_procurve/procurve.py +++ b/napalm_procurve/procurve.py @@ -32,6 +32,7 @@ from napalm_device_types import ( FingerprintRule, SwitchDriver, add_lag_interfaces, + run_fix_steps, ) from napalm_device_types.models import InterfaceConfigDict, VlanConfigDict from napalm.base import helpers as napalm_helpers @@ -76,6 +77,10 @@ _SSH_DISABLED_LEGACY = { } +#: What may stand in a quoted CLI value: no whitespace, quote or #. +_CLI_WORD = re.compile(r'^[^\s"#]+$') + + class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver): """NAPALM driver for HPE/Aruba ProCurve switches. @@ -1220,93 +1225,79 @@ class ProcurveDriver(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 _detect_management_vlan(self) -> Optional[int]: - """Find the VLAN that carries this switch's management IP.""" - try: - out = self._send_command("show ip") - for line in out.splitlines(): - # " VLAN8 | 172.22.8.x ..." or similar - m = re.search(r"VLAN\s*(\d+)[^|]*\|\s*" + re.escape(self.hostname), line, re.I) - if m: - return int(m.group(1)) - # Alternative: "VLAN 8 172.22.8.x" - m2 = re.search(r"\bVLAN\s+(\d+)\b.*\b" + re.escape(self.hostname) + r"\b", line, re.I) - if m2: - return int(m2.group(1)) - except Exception: - pass - return None + def configure_snmp_agent(self, settings: Dict[str, Any]) -> Dict[str, Any]: + """The agent is built in; configure community, contact and location, then save. - def _action_fix_snmp(self) -> Dict: - """Enable SNMP with community 'public' (read-only) on the ProCurve switch. - - ProCurve/Aruba syntax: - snmp-server community "public" manager restricted - - When connected via REST API, opens a temporary SSH session for the - config-mode CLI commands (REST API has no writable SNMP endpoint). + The REST API has no writable SNMP endpoint, so over REST a temporary SSH + session runs the config-mode CLI. """ - if self._transport == "api": - # REST API cannot run config-mode CLI. Temporarily open SSH. - old_api, old_device, old_transport = self._api, self._device, self._transport - self._api = None - self._device = None - self._transport = None - - ssh_error: str = "" - for legacy in (False, True): - try: - conn = ConnectHandler(**self._netmiko_kwargs(legacy)) - self._device = conn - self._transport = "ssh_legacy" if legacy else "ssh" - break - except Exception as exc: - ssh_error = str(exc) - - if self._device is None: - self._api, self._device, self._transport = old_api, old_device, old_transport - return { - "success": False, - "output": f"REST transport active; SSH also failed: {ssh_error}", - } + steps = [ + ("install", lambda: {"ok": True, "output": "the SNMP agent is built in"}), + ("configure", lambda: self._snmp_configure_cli(settings)), + ("firewall", lambda: self._snmp_sources_note(settings)), + ("start", lambda: self._snmp_save_and_verify(settings)), + ] + if self._transport != "api": + return run_fix_steps(steps) + old_api, old_device, old_transport = self._api, self._device, self._transport + self._api = self._device = self._transport = None + ssh_error = "" + for legacy in (False, True): try: - return self._action_fix_snmp_cli() - finally: + self._device = ConnectHandler(**self._netmiko_kwargs(legacy)) + self._transport = "ssh_legacy" if legacy else "ssh" + break + except Exception as exc: # noqa: BLE001 - the next form, then report + ssh_error = str(exc) + try: + if self._device is None: + failed = {"ok": False, "output": f"no SSH for the CLI: {ssh_error}"} + return run_fix_steps([("configure", lambda: failed)]) + return run_fix_steps(steps) + finally: + if self._device is not None: try: self._device.disconnect() - except Exception: + except Exception: # noqa: BLE001 - closing a temporary session pass - self._api, self._device, self._transport = old_api, old_device, old_transport + self._api, self._device, self._transport = old_api, old_device, old_transport - return self._action_fix_snmp_cli() - - def _action_fix_snmp_cli(self) -> Dict: - """Run the SNMP fix via CLI (SSH / Telnet, self._device must be open).""" - lines: list = [] + def _snmp_configure_cli(self, settings: Dict[str, Any]) -> Dict[str, Any]: + community = settings["community"] + if not _CLI_WORD.match(community): + raise ValueError("the community must be one word without quotes or #") + lines = [f'snmp-server community "{community}" manager restricted'] + 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() try: - errors = self._apply_config_lines('snmp-server community "public" manager restricted') - if errors: - lines.append(f"[warn] Config errors: {errors}") - return {"success": False, "output": "\n".join(lines)} - lines.append("[config] SNMP community 'public' (manager restricted / read-only) configured.") + errors = self._apply_config_lines("\n".join(lines)) finally: self._exit_config_mode() + if errors: + return {"ok": False, "output": f"the switch refused: {errors}"} + return {"ok": True, "output": "community (read-only), contact and location set"} + @staticmethod + def _snmp_sources_note(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 _snmp_save_and_verify(self, settings: Dict[str, Any]) -> Dict[str, Any]: self._save_config() - lines.append("[config] Configuration saved.") - - out = self._send_command("show snmp-server") - success = "public" in out - lines.append("[ok] SNMP is active with community 'public'." if success - else f"[warn] Verification — community not found: {out[:200]}") - return {"success": success, "output": "\n".join(lines)} + shown = self._send_command("show snmp-server") + if re.search(rf"^\s*{re.escape(settings['community'])}\s", shown, re.M): + return {"ok": True, "output": "configuration saved; the community is active"} + return {"ok": False, "output": f"saved, but the community is not shown: {shown[:200]}"} # --------------------------------------------------------------------------- diff --git a/tests/unit/test_snmp_agent.py b/tests/unit/test_snmp_agent.py new file mode 100644 index 0000000..2d6e5c5 --- /dev/null +++ b/tests/unit/test_snmp_agent.py @@ -0,0 +1,93 @@ +"""The SNMP agent through napalm-device-types' configure_snmp_agent (netOrk #1105). + +fix_snmp hard-coded community "public". The switch's agent is built in and the +switch filters nothing for its own management plane, so of the template's steps +only "configure" (community, contact, location) and "start" (save, verify) do +work here; the caller's values are used as given. +""" + +from __future__ import annotations + +from unittest.mock import MagicMock + +import pytest + +from napalm_procurve.procurve import ProcurveDriver + +SETTINGS = { + "community": "c0mm", + "allowed_sources": ["10.7.224.20"], + "contact": "ops", + "location": "rack 1", +} + + +@pytest.fixture +def driver(): + d = ProcurveDriver("10.0.0.40", "admin", "pw") + d._transport = "ssh" + d._device = MagicMock() + d.applied: list[str] = [] + d._enter_config_mode = MagicMock() + d._exit_config_mode = MagicMock() + d._save_config = MagicMock() + d._apply_config_lines = lambda text: d.applied.append(text) or [] + d._send_command = MagicMock( + return_value=" Community Name MIB View Write Access\n c0mm Manager Restricted\n" + ) + return d + + +def test_community_contact_and_location_are_the_callers(driver): + result = driver.configure_snmp_agent(SETTINGS) + + assert result["success"] is True + config = "\n".join(driver.applied) + assert 'snmp-server community "c0mm" manager restricted' in config + assert 'snmp-server contact "ops"' in config + assert 'snmp-server location "rack 1"' in config + assert "public" not in config + driver._save_config.assert_called_once() + + +def test_the_steps_say_what_the_switch_does_not_do(driver): + result = driver.configure_snmp_agent(SETTINGS) + + steps = {s["step"]: s for s in result["steps"]} + assert list(steps) == ["install", "configure", "firewall", "start"] + assert "built in" in steps["install"]["output"] + assert "10.7.224.20" in steps["firewall"]["output"] # not enforced, and said so + + +def test_a_community_the_switch_does_not_show_afterwards_fails(driver): + driver._send_command.return_value = " Community Name MIB View\n public Manager\n" + + assert driver.configure_snmp_agent(SETTINGS)["success"] is False + + +def test_a_rejected_line_fails_and_nothing_is_saved(driver): + driver._apply_config_lines = lambda text: ["Invalid input: community"] + + result = driver.configure_snmp_agent(SETTINGS) + + assert result["success"] is False + driver._save_config.assert_not_called() + + +@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_and_its_dead_helper_are_gone(): + gone = ( + "run_device_action", + "_action_fix_snmp", + "_action_fix_snmp_cli", + "_detect_management_vlan", + ) + for name in gone: + assert not hasattr(ProcurveDriver, name), name