From fe0aec9432bf824ff237f4357ac38e4848911633 Mon Sep 17 00:00:00 2001 From: Christian Manivong Date: Sun, 18 Oct 2026 21:23:43 +0200 Subject: [PATCH] fix(ssh): verify the host key netOrk expects, report the one seen (NetOrk/netork#1167) The driver accepted any SSH host key. It now verifies through napalm-device-types' HostKeyCheck (>= 4.19.0): with a known key in optional_args["ssh_host_keys"] any other key is refused before login, and every key seen is reported to optional_args["ssh_host_key_observer"]. Without either argument it connects as before (first use). --- napalm_procurve/procurve.py | 40 ++++++++++-- pyproject.toml | 4 +- tests/unit/test_ssh_host_key.py | 112 ++++++++++++++++++++++++++++++++ 3 files changed, 149 insertions(+), 7 deletions(-) create mode 100644 tests/unit/test_ssh_host_key.py diff --git a/napalm_procurve/procurve.py b/napalm_procurve/procurve.py index 6df66fa..ee1d49d 100644 --- a/napalm_procurve/procurve.py +++ b/napalm_procurve/procurve.py @@ -44,6 +44,12 @@ from napalm.base.exceptions import ( ReplaceConfigException, ) from napalm.base.netmiko_helpers import netmiko_args +from napalm_device_types.ssh_host_keys import ( + KNOWN_KEYS_ARG, + OBSERVER_ARG, + HostKeyCheck, + SshHostKeyMismatch, +) from napalm_procurve.api_client import ProcurveApiClient from napalm_procurve import parsers @@ -153,6 +159,10 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver): logger.setLevel(logging.DEBUG) self.netmiko_optional_args = netmiko_args(optional_args) + # The host keys to expect and where the one seen is reported (netork#1167). + self._host_key_args: Dict[str, Any] = { + k: optional_args[k] for k in (KNOWN_KEYS_ARG, OBSERVER_ARG) if k in optional_args + } # --- runtime state --- # Active transport: "api" | "ssh" | "ssh_legacy" | "telnet" @@ -318,17 +328,36 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver): **extra, ) + def _ssh_connect(self, legacy: bool = False, port: Optional[int] = None) -> ConnectHandler: + """A netmiko SSH session, its host key verified and reported (netork#1167). + + Every SSH session of this driver opens here. A refused key raises + :class:`SshHostKeyMismatch` before the login. + """ + kwargs = self._netmiko_kwargs(legacy) + if port is not None: + kwargs["port"] = port + check = HostKeyCheck(self._host_key_args, self.hostname, kwargs["port"]) + with check.netmiko(kwargs) as verified: + conn = ConnectHandler(**verified) + check.observe(conn.remote_conn_pre.get_transport()) + return conn + def _try_ssh(self, legacy: bool = False) -> bool: """Probe and connect via SSH. Returns True on success.""" label = "SSH-legacy" if legacy else "SSH" transport = "ssh_legacy" if legacy else "ssh" logger.debug("Trying %s for %s", label, self.hostname) try: - conn = ConnectHandler(**self._netmiko_kwargs(legacy)) + conn = self._ssh_connect(legacy) self._device = conn self._transport = transport logger.info("Connected to %s via %s", self.hostname, label) return True + except SshHostKeyMismatch: + # Not another transport: the legacy KEX or Telnet would reach the same + # host, and Telnet would hand it the password in clear text. + raise except NetmikoAuthenticationException as exc: raise self._auth_failure(transport, exc) from exc except NetmikoTimeoutException: @@ -638,9 +667,7 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver): self.port is the REST API port (443) in API transport mode. """ try: - kwargs = self._netmiko_kwargs() - kwargs["port"] = 22 # always SSH — self.port is the REST API port - conn = ConnectHandler(**kwargs) + conn = self._ssh_connect(port=22) # always SSH — self.port is the REST API port try: return conn.send_command("show running-config") finally: @@ -1248,9 +1275,12 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver): ssh_error = "" for legacy in (False, True): try: - self._device = ConnectHandler(**self._netmiko_kwargs(legacy)) + self._device = self._ssh_connect(legacy) self._transport = "ssh_legacy" if legacy else "ssh" break + except SshHostKeyMismatch as exc: # the same host either way + ssh_error = str(exc) + break except Exception as exc: # noqa: BLE001 - the next form, then report ssh_error = str(exc) try: diff --git a/pyproject.toml b/pyproject.toml index 4c6d6cd..0f8e0e5 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "napalm-hpe-aruba-procurve" -version = "0.4.0" +version = "0.5.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,7 +26,7 @@ classifiers = [ ] dependencies = [ "napalm>=4.0.0", - "napalm-device-types>=4.13.0", + "napalm-device-types>=4.19.0", "netmiko>=4.0.0", "paramiko>=5.0.0", # CVE-2026-44405 "netaddr", diff --git a/tests/unit/test_ssh_host_key.py b/tests/unit/test_ssh_host_key.py new file mode 100644 index 0000000..ec90ab4 --- /dev/null +++ b/tests/unit/test_ssh_host_key.py @@ -0,0 +1,112 @@ +"""Every SSH session verifies the switch's host key and reports it (NetOrk/netork#1167). + +The CLI over SSH (standard and legacy), the running-config fallback on port 22 +and the SNMP fix's temporary session all connect through ``_ssh_connect``. A +refused key ends the attempt: no legacy retry, and above all no fall-back to +Telnet, which would hand the same host the password in clear text. +""" + +from __future__ import annotations + +import os +from unittest.mock import MagicMock, patch + +import paramiko +import pytest +from napalm_device_types.ssh_host_keys import ( + KNOWN_KEYS_ARG, + OBSERVER_ARG, + SshHostKey, + SshHostKeyMismatch, +) +from netmiko.exceptions import NetmikoTimeoutException + +from napalm_procurve.procurve import ProcurveDriver + +SERVER_KEY = paramiko.ECDSAKey.generate() +OTHER_KEY = paramiko.ECDSAKey.generate() +HANDLER = "napalm_procurve.procurve.ConnectHandler" + + +def _device() -> MagicMock: + device = MagicMock() + device.remote_conn_pre.get_transport.return_value.get_remote_server_key.return_value = SERVER_KEY + device.send_command.return_value = "hostname x" + return device + + +def _refuse(**kwargs): + if kwargs["device_type"] == ProcurveDriver.NETMIKO_DEVICE_TYPE_TELNET: + raise AssertionError("fell back to Telnet after a refused host key") + try: + raise paramiko.BadHostKeyException("10.0.0.4", SERVER_KEY, OTHER_KEY) + except paramiko.SSHException as exc: + raise NetmikoTimeoutException(f"A paramiko SSHException occurred: {exc}") + + +def _driver(seen: list, known: dict | None = None, **extra) -> ProcurveDriver: + args: dict = {OBSERVER_ARG: seen.append, **extra} + if known is not None: + args[KNOWN_KEYS_ARG] = known + return ProcurveDriver("10.0.0.4", "manager", "s3cret", optional_args=args) + + +def test_first_use_lets_netmiko_accept_and_reports_the_key(): + seen: list = [] + with patch(HANDLER, return_value=_device()) as handler: + _driver(seen, transport="ssh").open() + assert handler.call_args.kwargs["ssh_strict"] is False + assert seen[0]["accepted"] is True + assert (seen[0]["host"], seen[0]["port"]) == ("10.0.0.4", 22) + + +def test_a_known_key_is_the_only_one_netmiko_accepts(): + seen: list = [] + known = SshHostKey.of(SERVER_KEY) + files: list[str] = [] + + def connect(**kwargs): + files.append(kwargs["alt_key_file"]) + with open(kwargs["alt_key_file"]) as f: + assert f.read().strip() == known.known_hosts_line("10.0.0.4", 22) + assert kwargs["ssh_strict"] is True + return _device() + + with patch(HANDLER, side_effect=connect): + _driver(seen, {"22": known.as_dict()}, transport="ssh").open() + assert not os.path.exists(files[0]) + assert seen[0]["accepted"] is True + + +def test_a_changed_key_stops_open_without_legacy_or_telnet(): + seen: list = [] + driver = _driver(seen, {"22": SshHostKey.of(OTHER_KEY).as_dict()}) + with patch("napalm_procurve.procurve.ProcurveApiClient.probe", return_value=(None, None)): + with patch(HANDLER, side_effect=_refuse) as handler: + with pytest.raises(SshHostKeyMismatch): + driver.open() + assert handler.call_count == 1 + assert driver._device is None + assert [s["accepted"] for s in seen] == [False] + + +def test_the_config_fallback_checks_the_key_of_port_22(): + seen: list = [] + driver = _driver(seen, {"22": SshHostKey.of(SERVER_KEY).as_dict()}, port=443) + with patch(HANDLER, return_value=_device()) as handler: + assert driver._get_config_via_ssh() == "hostname x" + assert handler.call_args.kwargs["port"] == 22 + assert handler.call_args.kwargs["ssh_strict"] is True + assert seen[0]["port"] == 22 and seen[0]["accepted"] is True + + +def test_the_snmp_fix_over_rest_stops_at_a_changed_key(): + seen: list = [] + driver = _driver(seen, {"443": SshHostKey.of(OTHER_KEY).as_dict()}, port=443) + driver._transport = "api" + driver._api = MagicMock() + with patch(HANDLER, side_effect=_refuse) as handler: + result = driver.configure_snmp_agent({"community": "c0mm", "allowed_sources": []}) + assert result["success"] is False + assert handler.call_count == 1 + assert "host key" in str(result)