diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..7a1f6bd --- /dev/null +++ b/.gitignore @@ -0,0 +1,6 @@ +__pycache__/ +*.pyc +*.egg-info/ +.venv/ +build/ +dist/ diff --git a/napalm_hpe_aruba_procurve.egg-info/PKG-INFO b/napalm_hpe_aruba_procurve.egg-info/PKG-INFO deleted file mode 100644 index 40b9e52..0000000 --- a/napalm_hpe_aruba_procurve.egg-info/PKG-INFO +++ /dev/null @@ -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 diff --git a/napalm_hpe_aruba_procurve.egg-info/SOURCES.txt b/napalm_hpe_aruba_procurve.egg-info/SOURCES.txt deleted file mode 100644 index f15b5c3..0000000 --- a/napalm_hpe_aruba_procurve.egg-info/SOURCES.txt +++ /dev/null @@ -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 \ No newline at end of file diff --git a/napalm_hpe_aruba_procurve.egg-info/dependency_links.txt b/napalm_hpe_aruba_procurve.egg-info/dependency_links.txt deleted file mode 100644 index 8b13789..0000000 --- a/napalm_hpe_aruba_procurve.egg-info/dependency_links.txt +++ /dev/null @@ -1 +0,0 @@ - diff --git a/napalm_hpe_aruba_procurve.egg-info/entry_points.txt b/napalm_hpe_aruba_procurve.egg-info/entry_points.txt deleted file mode 100644 index 04560d5..0000000 --- a/napalm_hpe_aruba_procurve.egg-info/entry_points.txt +++ /dev/null @@ -1,2 +0,0 @@ -[napalm.drivers] -procurve = napalm_procurve:ProcurveDriver diff --git a/napalm_hpe_aruba_procurve.egg-info/requires.txt b/napalm_hpe_aruba_procurve.egg-info/requires.txt deleted file mode 100644 index bcdacea..0000000 --- a/napalm_hpe_aruba_procurve.egg-info/requires.txt +++ /dev/null @@ -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 diff --git a/napalm_hpe_aruba_procurve.egg-info/top_level.txt b/napalm_hpe_aruba_procurve.egg-info/top_level.txt deleted file mode 100644 index 51a2d42..0000000 --- a/napalm_hpe_aruba_procurve.egg-info/top_level.txt +++ /dev/null @@ -1 +0,0 @@ -napalm_procurve diff --git a/napalm_procurve/__pycache__/__init__.cpython-312.pyc b/napalm_procurve/__pycache__/__init__.cpython-312.pyc deleted file mode 100644 index 73a8967..0000000 Binary files a/napalm_procurve/__pycache__/__init__.cpython-312.pyc and /dev/null differ diff --git a/napalm_procurve/__pycache__/api_client.cpython-312.pyc b/napalm_procurve/__pycache__/api_client.cpython-312.pyc deleted file mode 100644 index 76db29a..0000000 Binary files a/napalm_procurve/__pycache__/api_client.cpython-312.pyc and /dev/null differ diff --git a/napalm_procurve/__pycache__/parsers.cpython-312.pyc b/napalm_procurve/__pycache__/parsers.cpython-312.pyc deleted file mode 100644 index e7329a4..0000000 Binary files a/napalm_procurve/__pycache__/parsers.cpython-312.pyc and /dev/null differ diff --git a/napalm_procurve/__pycache__/procurve.cpython-312.pyc b/napalm_procurve/__pycache__/procurve.cpython-312.pyc deleted file mode 100644 index c435f0f..0000000 Binary files a/napalm_procurve/__pycache__/procurve.cpython-312.pyc and /dev/null differ 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/pyproject.toml b/pyproject.toml index 9973f20..4c6d6cd 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -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", diff --git a/tests/__pycache__/__init__.cpython-312.pyc b/tests/__pycache__/__init__.cpython-312.pyc deleted file mode 100644 index b33a6ff..0000000 Binary files a/tests/__pycache__/__init__.cpython-312.pyc and /dev/null differ diff --git a/tests/unit/__pycache__/__init__.cpython-312.pyc b/tests/unit/__pycache__/__init__.cpython-312.pyc deleted file mode 100644 index 5e9f8d1..0000000 Binary files a/tests/unit/__pycache__/__init__.cpython-312.pyc and /dev/null differ diff --git a/tests/unit/__pycache__/test_driver.cpython-312-pytest-9.1.1.pyc b/tests/unit/__pycache__/test_driver.cpython-312-pytest-9.1.1.pyc deleted file mode 100644 index e59bb8e..0000000 Binary files a/tests/unit/__pycache__/test_driver.cpython-312-pytest-9.1.1.pyc and /dev/null differ 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