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).

CHANGELOG.md

@@ -7,6 +7,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Security
+- The node's SSH session (the fallback for node commands and the provisioning steps) verifies its host key (NetOrk/netork#1167): the key the caller expects for port 22 in `optional_args["ssh_host_keys"]`, else the first one; every key seen goes to `optional_args["ssh_host_key_observer"]`. A changed key is refused before any login: a provisioning step fails, a best-effort read returns nothing. Both paths now share one session (`_node_ssh`). Replaces paramiko's `AutoAddPolicy`. Requires napalm-device-types 4.19.0; version 0.5.0.
+
 ### Changed
 - **Breaking:** `get_device_warnings` is gone; the node reports facts and the caller
   names the warnings (netOrk #1111). lldpd and the pending updates were already facts

napalm_proxmox/driver.py

@@ -48,6 +48,7 @@ from napalm_device_types import (
     SystemdServicesMixin,
 )
 from napalm.base.exceptions import ConnectionException
+from napalm_device_types.ssh_host_keys import HostKeyCheck, SshHostKeyMismatch
 
 try:
     from proxmoxer import ProxmoxAPI
@@ -318,39 +319,49 @@ class ProxmoxDriver(
 
         # Fallback: direct paramiko SSH
         try:
-            import paramiko
-
-            if self._ssh_client is None:
-                ssh_user = self._ssh_username or self.username
-                ssh_pass = self._ssh_password or self.password
-                ssh_pkey = None
-                if self._ssh_key and not ssh_pass:
-                    from io import StringIO as _StringIO
-                    ssh_pkey = paramiko.RSAKey.from_private_key(_StringIO(self._ssh_key))
-                self._ssh_client = paramiko.SSHClient()
-                self._ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
-                connect_kwargs: _JsonDict = {
-                    "hostname": self.hostname,
-                    "port": 22,
-                    "username": ssh_user,
-                    "timeout": self.timeout,
-                }
-                if ssh_pkey:
-                    connect_kwargs["pkey"] = ssh_pkey
-                else:
-                    connect_kwargs["password"] = ssh_pass
-                self._ssh_client.connect(**connect_kwargs)
-            _, stdout, stderr = self._ssh_client.exec_command(command, timeout=self.timeout)
+            _, stdout, stderr = self._node_ssh().exec_command(command, timeout=self.timeout)
             err = stderr.read().decode().strip()
             out = stdout.read().decode().strip()
             return out or err
         except ImportError:
             logger.warning("paramiko not installed — cannot exec SSH commands")
+        except SshHostKeyMismatch as exc:
+            # Refused before any login, and already reported to the caller's store.
+            logger.warning("%s", exc)
         except Exception as exc:
             logger.debug("SSH exec command failed: %s", exc)
 
         return ""
 
+    def _node_ssh(self) -> Any:  # paramiko.SSHClient; paramiko is imported lazily
+        """The SSH session to the node on port 22, opened once.
+
+        Its host key is the one the caller expects for port 22, or on first use
+        whichever the node presents; either way it is reported (netork#1167).
+        """
+        import paramiko
+
+        if self._ssh_client is None:
+            ssh_pass = self._ssh_password or self.password
+            connect_kwargs: _JsonDict = {
+                "username": self._ssh_username or self.username,
+                "timeout": self.timeout,
+            }
+            if self._ssh_key and not ssh_pass:
+                from io import StringIO as _StringIO
+
+                connect_kwargs["pkey"] = paramiko.RSAKey.from_private_key(_StringIO(self._ssh_key))
+            else:
+                connect_kwargs["password"] = ssh_pass
+            client = paramiko.SSHClient()
+            try:
+                HostKeyCheck(self.optional_args, self.hostname, 22).connect(client, **connect_kwargs)
+            except Exception:
+                client.close()
+                raise
+            self._ssh_client = client
+        return self._ssh_client
+
     # ------------------------------------------------------------------ #
     # Node info helpers
     # ------------------------------------------------------------------ #

napalm_proxmox/vm_provision_mixin.py

@@ -143,31 +143,7 @@ class ProxmoxVMProvisionMixin:
         disk import — where a non-zero exit or a caller-specific timeout must
         surface as a hard failure rather than an empty string.
         """
-        import paramiko
-
-        if self._ssh_client is None:
-            ssh_user = self._ssh_username or self.username
-            ssh_pass = self._ssh_password or self.password
-            ssh_pkey = None
-            if self._ssh_key and not ssh_pass:
-                from io import StringIO as _StringIO
-
-                ssh_pkey = paramiko.RSAKey.from_private_key(_StringIO(self._ssh_key))
-            self._ssh_client = paramiko.SSHClient()
-            self._ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
-            connect_kwargs: Dict[str, Any] = {
-                "hostname": self.hostname,
-                "port": 22,
-                "username": ssh_user,
-                "timeout": self.timeout,
-            }
-            if ssh_pkey:
-                connect_kwargs["pkey"] = ssh_pkey
-            else:
-                connect_kwargs["password"] = ssh_pass
-            self._ssh_client.connect(**connect_kwargs)
-
-        _, stdout, stderr = self._ssh_client.exec_command(command, timeout=timeout)
+        _, stdout, stderr = self._node_ssh().exec_command(command, timeout=timeout)
         exit_status = stdout.channel.recv_exit_status()
         out = stdout.read().decode().strip()
         err = stderr.read().decode().strip()

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "napalm-proxmox"
-version = "0.4.0"
+version = "0.5.0"
 description = "NAPALM driver for Proxmox VE (classic Linux networking, SDN, OVS)"
 readme = "README.md"
 license = { text = "Apache-2.0" }
@@ -25,7 +25,7 @@ classifiers = [
 requires-python = ">=3.9"
 dependencies = [
     "napalm>=5.0.0",
-    "napalm_device_types>=4.14.0",
+    "napalm_device_types>=4.19.0",
     "paramiko>=5.0.0",  # CVE-2026-44405; imported directly for SSH fallback (driver.py)
     "proxmoxer>=2.0.0",
     "netaddr>=0.9.0",

tests/test_ssh_host_key.py

@@ -0,0 +1,133 @@
+"""The node's SSH session verifies its host key, and reports the one it sees (NetOrk/netork#1167).
+
+Both ways onto the node's shell — the best-effort ``_exec_ssh_command`` and the
+provisioning steps' ``_run_node_command`` — share one session, opened on port 22.
+"""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock
+
+import paramiko
+import pytest
+from napalm_device_types.ssh_host_keys import (
+    KNOWN_KEYS_ARG,
+    OBSERVER_ARG,
+    SshHostKey,
+    SshHostKeyMismatch,
+)
+
+from napalm_proxmox.driver import ProxmoxDriver
+
+SERVER_KEY = paramiko.ECDSAKey.generate()
+OTHER_KEY = paramiko.ECDSAKey.generate()
+
+
+class _Stream:
+    def __init__(self, data: bytes) -> None:
+        self._data = data
+        self.channel = MagicMock()
+        self.channel.recv_exit_status.return_value = 0
+
+    def read(self) -> bytes:
+        return self._data
+
+
+class _Client:
+    """paramiko.SSHClient as far as the driver uses it; refuses like paramiko would."""
+
+    def __init__(self) -> None:
+        self.lines: list[str] = []
+        self.policy = None
+        self.connected_with: dict | None = None
+        self.commands: list[str] = []
+
+    def set_missing_host_key_policy(self, policy) -> None:
+        self.policy = policy
+
+    def get_host_keys(self):
+        client = self
+
+        class _Keys:
+            def load(self, path: str) -> None:
+                with open(path) as f:
+                    client.lines += f.read().splitlines()
+
+        return _Keys()
+
+    def connect(self, hostname, **kwargs) -> None:
+        if self.lines and SshHostKey.of(SERVER_KEY).key not in self.lines[0]:
+            raise paramiko.BadHostKeyException(hostname, SERVER_KEY, OTHER_KEY)
+        if not self.lines:
+            self.policy.missing_host_key(self, hostname, SERVER_KEY)
+        self.connected_with = {"hostname": hostname, **kwargs}
+
+    def get_transport(self):
+        transport = MagicMock()
+        transport.get_remote_server_key.return_value = SERVER_KEY
+        return transport
+
+    def exec_command(self, command, timeout=None):
+        self.commands.append(command)
+        return None, _Stream(b"ok"), _Stream(b"")
+
+    def close(self) -> None:
+        pass
+
+
+@pytest.fixture
+def client(monkeypatch) -> _Client:
+    fake = _Client()
+    monkeypatch.setattr(paramiko, "SSHClient", lambda: fake)
+    return fake
+
+
+def _driver(seen: list, known: dict | None = None) -> ProxmoxDriver:
+    args: dict = {"port": 8006, OBSERVER_ARG: seen.append}
+    if known is not None:
+        args[KNOWN_KEYS_ARG] = known
+    driver = ProxmoxDriver("10.0.0.9", "root", "s3cret", optional_args=args)
+    driver._node_api = MagicMock(side_effect=RuntimeError("no exec endpoint"))  # type: ignore[method-assign]
+    return driver
+
+
+def test_first_use_runs_the_command_and_reports_the_key(client):
+    seen: list = []
+    assert _driver(seen)._run_node_command("true", timeout=5) == "ok"
+    assert client.connected_with["port"] == 22
+    assert seen[0]["accepted"] is True
+    assert (seen[0]["host"], seen[0]["port"]) == ("10.0.0.9", 22)
+
+
+def test_the_key_for_the_api_port_is_not_the_ssh_one(client):
+    seen: list = []
+    other = SshHostKey.of(OTHER_KEY).as_dict()
+    _driver(seen, {"8006": other})._run_node_command("true", timeout=5)
+    assert seen[0]["expected"] is None
+
+
+def test_a_changed_key_stops_a_provisioning_step(client):
+    seen: list = []
+    driver = _driver(seen, {"22": SshHostKey.of(OTHER_KEY).as_dict()})
+    with pytest.raises(SshHostKeyMismatch):
+        driver._run_node_command("qm importdisk …", timeout=5)
+    assert client.commands == []
+    assert client.connected_with is None
+    assert seen[0]["accepted"] is False
+
+
+def test_a_changed_key_leaves_the_best_effort_read_empty(client):
+    seen: list = []
+    driver = _driver(seen, {"22": SshHostKey.of(OTHER_KEY).as_dict()})
+    assert driver._exec_ssh_command("cat /etc/os-release") == ""
+    assert client.commands == []
+    assert seen[0]["accepted"] is False
+
+
+def test_the_known_key_reuses_one_session(client):
+    seen: list = []
+    driver = _driver(seen, {"22": SshHostKey.of(SERVER_KEY).as_dict()})
+    driver._exec_ssh_command("uptime")
+    driver._run_node_command("true", timeout=5)
+    assert client.commands == ["uptime", "true"]
+    assert len(seen) == 1 and seen[0]["accepted"] is True

tests/test_vm_provision_mixin.py

@@ -616,8 +616,9 @@ def _mock_ssh_exec(exit_status: int, stdout_text: str = "", stderr_text: str = "
 
 def test_run_node_command_success_returns_stdout():
     mixin = ProxmoxVMProvisionMixin()
-    mixin._ssh_client = MagicMock()
-    mixin._ssh_client.exec_command.return_value = _mock_ssh_exec(0, stdout_text="hello\n")
+    ssh = MagicMock()
+    ssh.exec_command.return_value = _mock_ssh_exec(0, stdout_text="hello\n")
+    mixin._node_ssh = MagicMock(return_value=ssh)  # the driver's verified session
 
     result = mixin._run_node_command("echo hello", timeout=10)
 
@@ -626,10 +627,9 @@ def test_run_node_command_success_returns_stdout():
 
 def test_run_node_command_nonzero_exit_raises():
     mixin = ProxmoxVMProvisionMixin()
-    mixin._ssh_client = MagicMock()
-    mixin._ssh_client.exec_command.return_value = _mock_ssh_exec(
-        1, stderr_text="No such file or directory"
-    )
+    ssh = MagicMock()
+    ssh.exec_command.return_value = _mock_ssh_exec(1, stderr_text="No such file or directory")
+    mixin._node_ssh = MagicMock(return_value=ssh)
 
     with pytest.raises(RuntimeError, match="No such file or directory"):
         mixin._run_node_command("cat /nonexistent", timeout=10)