CHANGELOG.md

@@ -7,6 +7,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Security
+- The SSH host key is verified (NetOrk/netork#1167). `open()` hands netmiko the key the caller expects for the port in `optional_args["ssh_host_keys"]` (as a `known_hosts` file that exists only while it connects, with `ssh_strict`), so any other key is refused before the login (`SshHostKeyMismatch`); without one netmiko accepts the first key. Every key seen goes to `optional_args["ssh_host_key_observer"]`. OpenMediaVault and QNAP get this through `LinuxDriver.open()`. Requires napalm-device-types 4.19.0; version 0.11.0.
+
 ### Changed
 - **Breaking:** `get_device_warnings` is gone, and with it the `apt_proxy_url` argument.
   The driver reports facts and the application names warnings (napalm-device-types 4.14):

napalm_linux/linux.py

@@ -59,6 +59,7 @@ from napalm_device_types import (
     run_on_transport,
     strip_terminal_codes,
 )
+from napalm_device_types.ssh_host_keys import HostKeyCheck
 from napalm_device_types.models import (
     ApplyUpdatesResultDict,
     CronJobDict,
@@ -356,6 +357,8 @@ class LinuxDriver(
         # port is passed explicitly in open() — remove it from netmiko_optional_args
         # to avoid "multiple values for keyword argument 'port'"
         self.netmiko_optional_args.pop("port", None)
+        # The host key to expect, and where the one seen is reported (netork#1167).
+        self._host_key = HostKeyCheck(optional_args, hostname, self.port)
 
         # Runtime state
         self._device: ConnectHandler | None = None
@@ -368,6 +371,7 @@ class LinuxDriver(
     def open(self) -> None:
         """Open the SSH connection and detect the package manager."""
         try:
+            with self._host_key.netmiko(self.netmiko_optional_args) as netmiko_args_:
                 self._device = ConnectHandler(
                     device_type=self.NETMIKO_DEVICE_TYPE,
                     host=self.hostname,
@@ -376,12 +380,13 @@ class LinuxDriver(
                     port=self.port,
                     secret=self._secret,
                     timeout=self.timeout,
-                **self.netmiko_optional_args,
+                    **netmiko_args_,
                 )
         except NetmikoAuthenticationException as exc:
             raise ConnectionException(str(exc)) from exc
         except NetmikoTimeoutException as exc:
             raise ConnectionException(str(exc)) from exc
+        self._host_key.observe(self._device.remote_conn_pre.get_transport())
 
         # No history before anything else is typed (NetOrk/netork#1039).
         try:

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "napalm-linux"
-version = "0.10.1"
+version = "0.11.0"
 description = "NAPALM driver for generic Linux systems via SSH"
 readme = "README.md"
 requires-python = ">=3.9"
@@ -37,7 +37,7 @@ classifiers = [
 ]
 dependencies = [
     "napalm>=4.0",
-    "napalm-device-types>=4.18.0",
+    "napalm-device-types>=4.19.0",
     "netmiko>=4.0.0",
     "paramiko>=5.0.0",  # CVE-2026-44405
 ]

tests/test_ssh_host_key.py

@@ -0,0 +1,86 @@
+"""open() hands netmiko the host key to expect, and reports the one it saw (NetOrk/netork#1167).
+
+OpenMediaVault and QNAP open through ``LinuxDriver.open()`` and get this too.
+"""
+
+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_linux.linux import LinuxDriver
+
+SERVER_KEY = paramiko.ECDSAKey.generate()
+OTHER_KEY = paramiko.ECDSAKey.generate()
+
+
+def _device() -> MagicMock:
+    device = MagicMock()
+    device.send_command.return_value = "/usr/bin/apt"
+    device.remote_conn_pre.get_transport.return_value.get_remote_server_key.return_value = SERVER_KEY
+    return device
+
+
+def _driver(seen: list, known: SshHostKey | None = None) -> LinuxDriver:
+    args: dict = {"port": 2222, OBSERVER_ARG: seen.append}
+    if known is not None:
+        args[KNOWN_KEYS_ARG] = {"2222": known.as_dict()}
+    return LinuxDriver("10.0.0.8", "admin", "s3cret", optional_args=args)
+
+
+def test_first_use_lets_netmiko_accept_and_reports_the_key():
+    seen: list = []
+    with patch("napalm_linux.linux.ConnectHandler", return_value=_device()) as handler:
+        _driver(seen).open()
+    kwargs = handler.call_args.kwargs
+    assert kwargs["ssh_strict"] is False
+    assert kwargs["alt_host_keys"] is False
+    assert kwargs["system_host_keys"] is False
+    assert seen[0]["accepted"] is True
+    assert seen[0]["seen"]["fingerprint"] == SshHostKey.of(SERVER_KEY).fingerprint
+    assert (seen[0]["host"], seen[0]["port"]) == ("10.0.0.8", 2222)
+
+
+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.8", 2222)
+        assert kwargs["ssh_strict"] is True and kwargs["alt_host_keys"] is True
+        return _device()
+
+    with patch("napalm_linux.linux.ConnectHandler", side_effect=connect):
+        _driver(seen, known).open()
+    assert not os.path.exists(files[0])
+    assert seen[0]["accepted"] is True
+
+
+def test_a_changed_key_is_refused():
+    seen: list = []
+
+    def refuse(**kwargs):
+        try:
+            raise paramiko.BadHostKeyException("10.0.0.8", SERVER_KEY, OTHER_KEY)
+        except paramiko.SSHException as exc:  # what netmiko does with it
+            raise NetmikoTimeoutException(f"A paramiko SSHException occurred: {exc}")
+
+    driver = _driver(seen, SshHostKey.of(OTHER_KEY))
+    with patch("napalm_linux.linux.ConnectHandler", side_effect=refuse):
+        with pytest.raises(SshHostKeyMismatch):
+            driver.open()
+    assert driver._device is None
+    assert seen[0]["accepted"] is False