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

@@ -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:

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",

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)