Merge pull request 'feat!: configure_snmp_agent with the caller's values replaces fix_snmp (0.4.0)' (#15) from feat/snmp-agent-template into master

2026-10-17 06:48:19 +00:00
commit 382519f780
@@ -0,0 +1,6 @@
+__pycache__/
+*.pyc
+*.egg-info/
+.venv/
+build/
+dist/
@@ -1,142 +0,0 @@
-Metadata-Version: 2.4
-Name: napalm-hpe-aruba-procurve
-Version: 0.2.0
-Summary: NAPALM driver for HPE/Aruba ProCurve switches with auto-detecting transport (REST API, SSH, legacy SSH, Telnet)
-Author: Christian Manivong
-License: Apache-2.0
-Project-URL: Repository, https://github.com/chrismanivong/napalm-hpe-aruba-procurve
-Classifier: Topic :: Utilities
-Classifier: License :: OSI Approved :: Apache Software License
-Classifier: Programming Language :: Python :: 3
-Classifier: Programming Language :: Python :: 3.8
-Classifier: Programming Language :: Python :: 3.9
-Classifier: Programming Language :: Python :: 3.10
-Classifier: Programming Language :: Python :: 3.11
-Classifier: Programming Language :: Python :: 3.12
-Classifier: Operating System :: POSIX :: Linux
-Classifier: Operating System :: MacOS
-Requires-Python: >=3.8
-Description-Content-Type: text/markdown
-Requires-Dist: napalm>=4.0.0
-Requires-Dist: netmiko>=4.0.0
-Requires-Dist: paramiko>=5.0.0
-Requires-Dist: netaddr
-Requires-Dist: requests>=2.25.0
-Requires-Dist: urllib3
-Provides-Extra: dev
-Requires-Dist: pytest; extra == "dev"
-Requires-Dist: pytest-cov; extra == "dev"
-Requires-Dist: black; extra == "dev"
-Requires-Dist: ruff; extra == "dev"
-
-# napalm-hpe-aruba-procurve
-
-NAPALM driver for **HPE / Aruba ProCurve** switches — from ancient 2520G-8-PoE
-to modern 2530/2540 series.
-
-## Transport auto-detection
-
-The driver probes transports in order and uses the first one that succeeds:
-
-| Priority | Transport | Notes |
-|----------|-----------|-------|
-| 1 | REST API (HTTPS → HTTP, v7 → v6) | Newer 2530/2540 with `rest-interface` enabled |
-| 2 | SSH (standard) | hp_procurve netmiko driver |
-| 3 | SSH (legacy KEX) | Forces older kex/cipher negotiation for 2520G-8-PoE etc. |
-| 4 | Telnet | Oldest devices without SSH or with broken SSH |
-
-Override the transport with `optional_args={"transport": "ssh_legacy"}` to skip
-auto-detection.
-
-## Switch prerequisites
-
-### REST API (newer switches, e.g. 2530, 2540)
-
-```
-web-management ssl
-rest-interface
-rest-interface session-idle-timeout 120
-```
-
-### SSH (all ProCurve)
-
-```
-crypto key generate ssh rsa
-ip ssh
-```
-
-Telnet is enabled by default on most ProCurve switches.
-
-## Installation
-
-```bash
-pip install napalm napalm-hpe-aruba-procurve
-```
-
-Or from source:
-
-```bash
-git clone https://github.com/chrismanivong/napalm-hpe-aruba-procurve
-pip install -e napalm-hpe-aruba-procurve/
-```
-
-## Quick start
-
-```python
-from napalm import get_network_driver
-
-Driver = get_network_driver("procurve")
-
-with Driver(
-    "10.0.0.1",
-    "manager",
-    "secret",
-    optional_args={
-        # "transport": "ssh",          # force transport (api/ssh/ssh_legacy/telnet)
-        # "port": 22,
-        # "ssl_verify": False,         # disable SSL cert check for API
-        # "api_version": "v7",         # API version hint (v3/v6/v7)
-        # "secret": "enablepassword",  # enable password for CLI
-    },
-) as dev:
-    print(dev.get_facts())
-    print(dev.get_interfaces())
-```
-
-## Supported NAPALM methods
-
-| Method | API | SSH/Telnet |
-|--------|-----|------------|
-| `open()` | ✅ | ✅ |
-| `close()` | ✅ | ✅ |
-| `is_alive()` | ✅ | ✅ |
-| `get_facts()` | ✅ | ✅ |
-| `get_interfaces()` | ✅ | ✅ |
-| `get_interfaces_ip()` | ✅ | ✅ |
-| `get_arp_table()` | ✅ | ✅ |
-| `get_mac_address_table()` | ✅ | ✅ |
-| `get_lldp_neighbors()` | ✅ | ✅ |
-| `get_lldp_neighbors_detail()` | ✅ | ✅ |
-| `get_config()` | ✅ | ✅ |
-| `get_ntp_servers()` | ✅ | ✅ |
-| `get_environment()` | ❌ | ✅ |
-| `get_users()` | ❌ | ✅ |
-| `get_snmp_information()` | ❌ | ✅ |
-| `ping()` | ✅ | ✅ |
-| `cli()` | ✅ | ✅ |
-| `load_merge_candidate()` | ❌ | ✅ |
-| `load_replace_candidate()` | ❌ | ✅ |
-| `compare_config()` | ❌ | ✅ |
-| `commit_config()` | ❌ | ✅ |
-| `discard_config()` | ❌ | ✅ |
-| `rollback()` | ❌ | ✅ |
-
-## Tested devices
-
- HP ProCurve 2520G-8-PoE (J9565A) — SSH legacy / Telnet
- HP ProCurve 2920-48G — SSH
- Aruba 2530-8-PoE+ — SSH + REST API
-
-## License
-
-Apache 2.0
@@ -1,12 +0,0 @@
-README.md
-pyproject.toml
-napalm_hpe_aruba_procurve.egg-info/PKG-INFO
-napalm_hpe_aruba_procurve.egg-info/SOURCES.txt
-napalm_hpe_aruba_procurve.egg-info/dependency_links.txt
-napalm_hpe_aruba_procurve.egg-info/entry_points.txt
-napalm_hpe_aruba_procurve.egg-info/requires.txt
-napalm_hpe_aruba_procurve.egg-info/top_level.txt
-napalm_procurve/__init__.py
-napalm_procurve/api_client.py
-napalm_procurve/parsers.py
-napalm_procurve/procurve.py
@@ -1 +0,0 @@
-
@@ -1,2 +0,0 @@
-[napalm.drivers]
-procurve = napalm_procurve:ProcurveDriver
@@ -1,12 +0,0 @@
-napalm>=4.0.0
-netmiko>=4.0.0
-paramiko>=5.0.0
-netaddr
-requests>=2.25.0
-urllib3
-
-[dev]
-pytest
-pytest-cov
-black
-ruff
@@ -1 +0,0 @@
-napalm_procurve
@@ -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]}"}


 # ---------------------------------------------------------------------------
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"

 [project]
 name = "napalm-hpe-aruba-procurve"
-version = "0.3.0"
+version = "0.4.0"
 description = "NAPALM driver for HPE/Aruba ProCurve switches with auto-detecting transport (REST API, SSH, legacy SSH, Telnet)"
 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",
@@ -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