Merge pull request 'fix(ssh): verify the host key netOrk expects, report the one seen (NetOrk/netork#1167)' (#18) from fix/ssh-host-key-tofu-1167 into main

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()` takes the key to expect from `optional_args["ssh_host_key"]` and refuses any other one before it logs in (`SshHostKeyMismatch`); without one it accepts the first key. Every key it sees goes to `optional_args["ssh_host_key_observer"]`. Replaces paramiko's `AutoAddPolicy`. Requires napalm-device-types 4.19.0; version 0.4.0.
+
 ## [0.3.0] - 2026-10-17
 
 ### Changed

napalm_bsd/base.py

@@ -20,6 +20,7 @@ from napalm.base.exceptions import ConnectionClosedException, ConnectionExceptio
 from napalm_device_types import EnsurePackageMixin, OSDriver, SnmpAgentMixin
 from napalm_device_types.models import FixStepOutcomeDict, SnmpAgentSettingsDict
 from napalm_device_types.host_status import HostStatusMixin
+from napalm_device_types.ssh_host_keys import HostKeyCheck, SshHostKeyMismatch
 from napalm_device_types.channel import (
     ByteStream,
     CommandResult,
@@ -124,6 +125,8 @@ class BsdDriver(SnmpAgentMixin, EnsurePackageMixin, HostStatusMixin, OSDriver):
         self._sudo_password: Optional[str] = args.get("sudo_password")
         self._allow_agent: bool = bool(args.get("allow_agent", False))
         self._look_for_keys: bool = bool(args.get("look_for_keys", False))
+        # The host key to expect, and where the one seen is reported (netork#1167).
+        self._host_key = HostKeyCheck(args, hostname, self.port)
         self._client: Optional[paramiko.SSHClient] = None
         self._root: Optional[bool] = None
 
@@ -131,11 +134,9 @@ class BsdDriver(SnmpAgentMixin, EnsurePackageMixin, HostStatusMixin, OSDriver):
 
     def open(self) -> None:
         client = paramiko.SSHClient()
-        client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
         try:
-            client.connect(
-                self.hostname,
-                port=self.port,
+            self._host_key.connect(
+                client,
                 username=self.username,
                 password=self.password or None,
                 key_filename=self._key_file,
@@ -143,6 +144,9 @@ class BsdDriver(SnmpAgentMixin, EnsurePackageMixin, HostStatusMixin, OSDriver):
                 look_for_keys=self._look_for_keys,
                 timeout=self.timeout,
             )
+        except SshHostKeyMismatch:
+            client.close()  # refused before any login; nothing was sent
+            raise
         except (paramiko.SSHException, socket.error) as exc:
             client.close()
             raise ConnectionException(f"SSH to {self.hostname} failed: {exc}") from exc

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "napalm-bsd"
-version = "0.3.0"
+version = "0.4.0"
 description = "NAPALM drivers for FreeBSD and OpenBSD hosts via SSH"
 readme = "README.md"
 requires-python = ">=3.10"
@@ -26,7 +26,7 @@ classifiers = [
 ]
 dependencies = [
     "napalm>=4.0",
-    "napalm-device-types>=4.13.0",  # fixes: SnmpAgentMixin, EnsurePackageMixin
+    "napalm-device-types>=4.19.0",  # fixes: SnmpAgentMixin, EnsurePackageMixin
     "paramiko>=5.0.0",  # CVE-2026-44405
 ]
 

tests/test_ssh_host_key.py

@@ -0,0 +1,109 @@
+"""open() verifies the host key it is given, and reports the one it sees (NetOrk/netork#1167)."""
+
+from __future__ import annotations
+
+import paramiko
+import pytest
+from napalm_device_types.ssh_host_keys import (
+    KNOWN_KEYS_ARG,
+    OBSERVER_ARG,
+    SshHostKey,
+    SshHostKeyMismatch,
+)
+
+from napalm_bsd import FreeBSDDriver, base
+
+SERVER_KEY = paramiko.ECDSAKey.generate()
+OTHER_KEY = paramiko.ECDSAKey.generate()
+
+
+class _HostKeys:
+    def __init__(self) -> None:
+        self.lines: list[str] = []
+
+    def load(self, path: str) -> None:
+        with open(path) as f:
+            self.lines += f.read().splitlines()
+
+
+class _Transport:
+    def get_remote_server_key(self):
+        return SERVER_KEY
+
+    def is_active(self) -> bool:
+        return True
+
+
+class _Client:
+    """paramiko.SSHClient as far as open() uses it; refuses like paramiko would."""
+
+    def __init__(self) -> None:
+        self.host_keys = _HostKeys()
+        self.policy = None
+        self.connected_with: dict | None = None
+        self.closed = False
+
+    def set_missing_host_key_policy(self, policy) -> None:
+        self.policy = policy
+
+    def get_host_keys(self) -> _HostKeys:
+        return self.host_keys
+
+    def connect(self, hostname, **kwargs) -> None:
+        if self.host_keys.lines and SshHostKey.of(SERVER_KEY).key not in self.host_keys.lines[0]:
+            raise paramiko.BadHostKeyException(hostname, SERVER_KEY, OTHER_KEY)
+        if not self.host_keys.lines:
+            self.policy.missing_host_key(self, hostname, SERVER_KEY)
+        self.connected_with = {"hostname": hostname, **kwargs}
+
+    def get_transport(self) -> _Transport:
+        return _Transport()
+
+    def close(self) -> None:
+        self.closed = True
+
+
+@pytest.fixture
+def client(monkeypatch) -> _Client:
+    fake = _Client()
+    monkeypatch.setattr(base.paramiko, "SSHClient", lambda: fake)
+    return fake
+
+
+def _driver(seen: list, known: SshHostKey | None = None) -> FreeBSDDriver:
+    args: dict = {"port": 2222, OBSERVER_ARG: seen.append}
+    if known is not None:
+        args[KNOWN_KEYS_ARG] = {"2222": known.as_dict()}
+    return FreeBSDDriver("10.0.0.7", "netork", "s3cret", optional_args=args)
+
+
+def test_first_use_connects_and_reports_the_key(client):
+    seen: list = []
+    _driver(seen).open()
+    assert client.connected_with["password"] == "s3cret"
+    assert client.connected_with["port"] == 2222
+    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.7", 2222)
+
+
+def test_the_known_key_connects(client):
+    seen: list = []
+    _driver(seen, SshHostKey.of(SERVER_KEY)).open()
+    assert client.host_keys.lines == [SshHostKey.of(SERVER_KEY).known_hosts_line("10.0.0.7", 2222)]
+    assert client.connected_with is not None
+    assert seen[0]["accepted"] is True
+
+
+def test_a_changed_key_is_refused_and_the_client_closed(client):
+    seen: list = []
+    with pytest.raises(SshHostKeyMismatch):
+        _driver(seen, SshHostKey.of(OTHER_KEY)).open()
+    assert client.connected_with is None
+    assert client.closed is True
+    assert seen[0]["accepted"] is False
+
+
+def test_without_host_key_args_it_connects_as_before(client):
+    FreeBSDDriver("10.0.0.7", "netork", "s3cret").open()
+    assert client.connected_with is not None