README.md

@@ -254,6 +254,27 @@ NAPALM's own getters (`get_facts`, `get_interfaces`, `ping`, `get_config`) are n
 wrapped in a template — they belong to NAPALM, and code outside this repo relies on
 their contract.
 
+### SSH host keys: trust on first use
+
+A driver that accepts any host key logs in with the device's credentials at whatever
+answers on the device's address (NetOrk/netork#1167). The store of trusted keys is the
+caller's; `ssh_host_keys.HostKeyCheck` is how every SSH driver verifies against it:
+
+- **In:** `optional_args["ssh_host_keys"]` = `{"22": {"key_type": "ssh-ed25519", "key": "<base64>"}}`,
+  by port. The driver takes the key for the port it connects to. With one, any other key is
+  refused *before* authentication and the connection raises `SshHostKeyMismatch` (a
+  `ConnectionException`). Without one, the first key is accepted.
+- **Out:** `optional_args["ssh_host_key_observer"]`, a callable, gets one dict per
+  connection: `{"host", "port", "seen": {"key_type", "key", "fingerprint"}, "expected", "accepted"}`.
+  `accepted` is False only for a refused key. A failing observer is logged, never raised.
+- **paramiko:** `HostKeyCheck(optional_args, host, port).connect(client, **connect_kwargs)`.
+- **netmiko:** `with check.netmiko(self.netmiko_optional_args) as kwargs: ConnectHandler(..., **kwargs)`,
+  then `check.observe(conn.remote_conn_pre.get_transport())`. The known key reaches netmiko as a
+  `known_hosts` file that exists only inside the block, with `ssh_strict=True`.
+
+A caller that passes neither key gets first use on every connection, as before; a driver
+that ignores both still connects. Old and new versions on either side keep working.
+
 ## Design principle: generic vs. device-specific logic
 
 When adding behavior to a device-type base class, split it along one line: **would

napalm_device_types/__init__.py

@@ -181,6 +181,14 @@ from napalm_device_types.ssh_server import (
     SshHardeningResult,
     plan_ssh_hardening,
 )
+from napalm_device_types.ssh_host_keys import (
+    KNOWN_KEYS_ARG as SSH_HOST_KEYS_ARG,
+    OBSERVER_ARG as SSH_HOST_KEY_OBSERVER_ARG,
+    HostKeyCheck,
+    HostKeyObservation,
+    SshHostKey,
+    SshHostKeyMismatch,
+)
 from napalm_device_types.storage import StorageDriver
 from napalm_device_types.switch import SwitchDriver
 
@@ -288,6 +296,12 @@ __all__ = [
     "SshHardeningPlan",
     "SshHardeningResult",
     "plan_ssh_hardening",
+    "SSH_HOST_KEYS_ARG",
+    "SSH_HOST_KEY_OBSERVER_ARG",
+    "HostKeyCheck",
+    "HostKeyObservation",
+    "SshHostKey",
+    "SshHostKeyMismatch",
     "parse_virtual_ips",
     "add_lag_interfaces",
     "driver_supports_ping",

napalm_device_types/ssh_host_keys.py

@@ -0,0 +1,300 @@
+# -*- coding: utf-8 -*-
+"""SSH host keys: trust on first use, and a changed key refused before any login.
+
+A driver that accepts whatever host key answers at a device's address logs in
+there with the device's credentials, whoever that is (NetOrk/netork#1167). The
+layer above keeps the store of the keys it trusts; a driver only verifies:
+
+* It receives the keys to expect in ``optional_args["ssh_host_keys"]``
+  (:data:`KNOWN_KEYS_ARG`), by the port they were seen on:
+  ``{"22": {"key_type": "ssh-ed25519", "key": "<base64>"}}``, the two fields of
+  a ``known_hosts`` line. A driver takes the one for the port it connects to, so
+  the caller need not know which port that is (Proxmox: its API on 8006, SSH on
+  22). With a key, any other key is refused before a password or a private key
+  leaves the caller. Without one, the first key is accepted: trust on first use.
+* It reports every key it sees through ``optional_args["ssh_host_key_observer"]``
+  (:data:`OBSERVER_ARG`), a callable that receives one dict per connection::
+
+      {"host": str, "port": int,
+       "seen": {"key_type", "key", "fingerprint"},
+       "expected": {...} | None,
+       "accepted": bool}
+
+  ``accepted`` is False only for a refused key; the connection then raises
+  :class:`SshHostKeyMismatch`. A failing observer is logged and ignored, so the
+  store being away never stops a connection that verified.
+
+A caller that passes neither keeps working as before (first use, every time),
+and a driver that ignores both still connects for a caller that passes them:
+old and new versions on either side do not break each other.
+
+:class:`HostKeyCheck` is the one implementation, for a paramiko ``SSHClient``
+(:meth:`~HostKeyCheck.connect`) and for netmiko (:meth:`~HostKeyCheck.netmiko`,
+then :meth:`~HostKeyCheck.observe`). Like :mod:`napalm_device_types.channel` it
+works on the objects paramiko hands it and imports no SSH library itself.
+"""
+
+from __future__ import annotations
+
+import base64
+import hashlib
+import logging
+import os
+import tempfile
+from contextlib import contextmanager
+from dataclasses import dataclass
+from typing import Any, Dict, Iterator, List, Mapping, Optional, Set
+
+from napalm.base.exceptions import ConnectionException
+
+logger = logging.getLogger(__name__)
+
+#: The ``optional_args`` key that carries the host keys to expect, by port.
+KNOWN_KEYS_ARG = "ssh_host_keys"
+#: The ``optional_args`` key that carries the callable every seen key goes to.
+OBSERVER_ARG = "ssh_host_key_observer"
+
+#: The netmiko arguments this module decides; a caller's own values give way.
+_NETMIKO_HOST_KEY_ARGS = ("ssh_strict", "system_host_keys", "alt_host_keys", "alt_key_file")
+
+_DEFAULT_SSH_PORT = 22
+
+
+@dataclass(frozen=True)
+class SshHostKey:
+    """A host's public key, as the two fields of a ``known_hosts`` line."""
+
+    key_type: str
+    #: The public key blob, base64.
+    key: str
+
+    @property
+    def fingerprint(self) -> str:
+        """``SHA256:…``, as ``ssh-keygen -l`` and the ssh client print it."""
+        digest = hashlib.sha256(base64.b64decode(self.key)).digest()
+        return "SHA256:" + base64.b64encode(digest).decode("ascii").rstrip("=")
+
+    def as_dict(self) -> Dict[str, str]:
+        return {"key_type": self.key_type, "key": self.key, "fingerprint": self.fingerprint}
+
+    @classmethod
+    def from_value(cls, value: Any) -> Optional["SshHostKey"]:
+        """The key in *value* (a mapping with ``key_type`` and ``key``), or None for none."""
+        if not value:
+            return None
+        if not isinstance(value, Mapping) or not value.get("key_type") or not value.get("key"):
+            raise ValueError(f"not an SSH host key: {value!r}")
+        return cls(str(value["key_type"]), str(value["key"]))
+
+    @classmethod
+    def of(cls, pkey: Any) -> "SshHostKey":
+        """The key of a paramiko ``PKey`` (what a transport says the server presented)."""
+        return cls(pkey.get_name(), pkey.get_base64())
+
+    def known_hosts_line(self, host: str, port: int) -> str:
+        """The line under which an ssh client looks this host up."""
+        name = host if port == _DEFAULT_SSH_PORT else f"[{host}]:{port}"
+        return f"{name} {self.key_type} {self.key}"
+
+
+@dataclass(frozen=True)
+class HostKeyObservation:
+    """The key one connection saw, against the key it was to expect."""
+
+    host: str
+    port: int
+    seen: SshHostKey
+    expected: Optional[SshHostKey]
+
+    @property
+    def accepted(self) -> bool:
+        """True on first use and for the expected key; False for a refused one."""
+        return self.expected is None or self.expected == self.seen
+
+    def as_dict(self) -> Dict[str, Any]:
+        return {
+            "host": self.host,
+            "port": self.port,
+            "seen": self.seen.as_dict(),
+            "expected": self.expected.as_dict() if self.expected else None,
+            "accepted": self.accepted,
+        }
+
+
+class SshHostKeyMismatch(ConnectionException):  # type: ignore[misc]
+    """The device presented another host key than the one it is known by.
+
+    Raised before any authentication: no password and no private key went to it.
+    """
+
+    def __init__(self, observation: HostKeyObservation) -> None:
+        self.observation = observation
+        expected = observation.expected
+        trusted = f"{expected.fingerprint} ({expected.key_type})" if expected else "none"
+        super().__init__(
+            f"SSH host key mismatch at {observation.host}:{observation.port}: it presented "
+            f"{observation.seen.fingerprint} ({observation.seen.key_type}), the trusted key is "
+            f"{trusted}. Nothing was sent to it."
+        )
+
+
+class _AcceptFirst:
+    """A paramiko missing-host-key policy for first use: accept, keep nothing.
+
+    The key is reported once the connection stands (:meth:`HostKeyCheck.observe`);
+    nothing is written to a ``known_hosts`` file.
+    """
+
+    def missing_host_key(self, client: Any, hostname: str, key: Any) -> None:
+        return None
+
+
+class _RefuseUnknown:
+    """A paramiko policy for a host with a known key: refuse anything else.
+
+    paramiko compares a loaded key itself and raises ``BadHostKeyException``
+    for another one; this only answers if the loaded key was not found under
+    the name connected to, which must not turn into an accepted key.
+    """
+
+    def missing_host_key(self, client: Any, hostname: str, key: Any) -> None:
+        raise ConnectionException(f"SSH host key of {hostname} is not the one expected")
+
+
+def _expected_key(known: Any, port: int) -> Optional[SshHostKey]:
+    """The key in *known* (``{port: key}``, the port as text or number) for *port*."""
+    if not known:
+        return None
+    if not isinstance(known, Mapping):
+        raise ValueError(f"SSH host keys must be given by port, not as {type(known).__name__}")
+    return SshHostKey.from_value(known.get(str(port)) or known.get(port))
+
+
+def _refused_key(exc: BaseException) -> Optional[Any]:
+    """paramiko's ``BadHostKeyException`` in *exc*'s chain (netmiko and drivers wrap it)."""
+    seen: Set[int] = set()
+    stack: List[Optional[BaseException]] = [exc]
+    while stack:
+        current = stack.pop()
+        if current is None or id(current) in seen:
+            continue
+        seen.add(id(current))
+        if hasattr(current, "expected_key") and hasattr(current, "key"):
+            return current
+        stack.extend([current.__cause__, current.__context__])
+    return None
+
+
+class HostKeyCheck:
+    """How one driver connection verifies its host's key.
+
+    *optional_args* are the driver's (both keys optional), *host* and *port* the
+    address the connection goes to, as given to paramiko or netmiko.
+    """
+
+    def __init__(self, optional_args: Optional[Mapping[str, Any]], host: str, port: int) -> None:
+        args = optional_args or {}
+        self.host = host
+        self.port = int(port)
+        self.expected = _expected_key(args.get(KNOWN_KEYS_ARG), self.port)
+        self._observer = args.get(OBSERVER_ARG)
+
+    # ------------------------------------------------------------------
+    # paramiko
+    # ------------------------------------------------------------------
+
+    def prepare(self, client: Any) -> None:
+        """Before ``client.connect()``: the known key, or acceptance of the first one.
+
+        The known key goes into the client's host keys, so paramiko asks the
+        server for that key type first and refuses any other key itself.
+        """
+        if self.expected is None:
+            client.set_missing_host_key_policy(_AcceptFirst())
+            return
+        with self._known_hosts_file() as path:
+            client.get_host_keys().load(path)
+        client.set_missing_host_key_policy(_RefuseUnknown())
+
+    def connect(self, client: Any, **kwargs: Any) -> HostKeyObservation:
+        """``client.connect()`` to this host with *kwargs*, verified and reported."""
+        self.prepare(client)
+        with self.refusing():
+            client.connect(hostname=self.host, port=self.port, **kwargs)
+        return self.observe(client.get_transport())
+
+    # ------------------------------------------------------------------
+    # netmiko
+    # ------------------------------------------------------------------
+
+    @contextmanager
+    def netmiko(self, base: Mapping[str, Any]) -> Iterator[Dict[str, Any]]:
+        """``ConnectHandler`` keyword arguments: *base* with the host key decided.
+
+        Open the connection inside the block and call :meth:`observe` on its
+        transport after it. A known key goes to netmiko in a ``known_hosts``
+        file that exists only for the block, with ``ssh_strict``: should the
+        file not be read, netmiko refuses rather than accepts.
+        """
+        kwargs = {k: v for k, v in base.items() if k not in _NETMIKO_HOST_KEY_ARGS}
+        kwargs["system_host_keys"] = False
+        if self.expected is None:
+            kwargs.update(ssh_strict=False, alt_host_keys=False)
+            with self.refusing():
+                yield kwargs
+            return
+        with self._known_hosts_file() as path:
+            kwargs.update(ssh_strict=True, alt_host_keys=True, alt_key_file=path)
+            with self.refusing():
+                yield kwargs
+
+    # ------------------------------------------------------------------
+    # both
+    # ------------------------------------------------------------------
+
+    @contextmanager
+    def refusing(self) -> Iterator[None]:
+        """Around a connect: a refused host key becomes :class:`SshHostKeyMismatch`."""
+        try:
+            yield
+        except Exception as exc:
+            refused = _refused_key(exc)
+            if refused is None:
+                raise
+            expected = self.expected or SshHostKey.of(refused.expected_key)
+            offered = SshHostKey.of(refused.key)
+            observation = HostKeyObservation(self.host, self.port, offered, expected)
+            self._report(observation)
+            raise SshHostKeyMismatch(observation) from exc
+
+    def observe(self, transport: Any) -> HostKeyObservation:
+        """After a connect: the key the server presented on *transport*, reported."""
+        seen = SshHostKey.of(transport.get_remote_server_key())
+        observation = HostKeyObservation(self.host, self.port, seen, self.expected)
+        self._report(observation)
+        return observation
+
+    def _report(self, observation: HostKeyObservation) -> None:
+        if not callable(self._observer):
+            return
+        try:
+            self._observer(observation.as_dict())
+        except Exception:  # the store is the caller's; a connection does not hang on it
+            logger.warning(
+                "Could not report the SSH host key of %s:%s", self.host, self.port, exc_info=True
+            )
+
+    @contextmanager
+    def _known_hosts_file(self) -> Iterator[str]:
+        """A private ``known_hosts`` file holding only the expected key."""
+        assert self.expected is not None
+        fd, path = tempfile.mkstemp(prefix="known_hosts-")
+        try:
+            with os.fdopen(fd, "w") as f:
+                f.write(self.expected.known_hosts_line(self.host, self.port) + "\n")
+            yield path
+        finally:
+            try:
+                os.unlink(path)
+            except OSError:
+                pass

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "napalm-device-types"
-version = "4.18.0"
+version = "4.19.0"
 description = "Abstract device-type base classes for NAPALM drivers"
 readme = "README.md"
 requires-python = ">=3.10"

tests/test_ssh_host_keys.py

@@ -0,0 +1,275 @@
+"""SSH host keys: trust on first use, and a changed key refused before any login.
+
+The layer above keeps the store (NetOrk/netork#1167). A driver gets the key it
+should expect through ``optional_args["ssh_host_keys"]`` (by port) and reports every key it
+sees through ``optional_args["ssh_host_key_observer"]``. These tests run real
+paramiko against an in-process SSH server, because the property that matters —
+nothing is sent to a host with the wrong key — is paramiko's behaviour once the
+key is loaded, not something a fake could show.
+"""
+
+from __future__ import annotations
+
+import os
+import socket
+import threading
+
+import paramiko
+import pytest
+from napalm.base.exceptions import ConnectionException
+
+from napalm_device_types.ssh_host_keys import (
+    KNOWN_KEYS_ARG,
+    OBSERVER_ARG,
+    HostKeyCheck,
+    SshHostKey,
+    SshHostKeyMismatch,
+)
+
+#: An ed25519 public key and the fingerprint ``ssh-keygen -lf`` prints for it.
+ED25519 = SshHostKey(
+    "ssh-ed25519", "AAAAC3NzaC1lZDI1NTE5AAAAIJH2oO06vroJT2ooSB3/2tu8HufXQng4xQlxAk/4raee"
+)
+ED25519_FP = "SHA256:26dxfTuve20LGGTpJ1T0ns11/ID3XMhbOahQq3k75TU"
+
+
+class _Server(paramiko.ServerInterface):
+    """Accepts any password and remembers each one it was offered."""
+
+    def __init__(self) -> None:
+        self.passwords: list[str] = []
+
+    def get_allowed_auths(self, username):
+        return "password"
+
+    def check_auth_password(self, username, password):
+        self.passwords.append(password)
+        return paramiko.AUTH_SUCCESSFUL
+
+
+class _Host:
+    """An SSH server on one end of a socket pair, with *host_key*."""
+
+    def __init__(self, host_key: paramiko.PKey) -> None:
+        self.client_sock, server_sock = socket.socketpair()
+        self.transport = paramiko.Transport(server_sock)
+        self.transport.add_server_key(host_key)
+        self.server = _Server()
+        self._thread = threading.Thread(target=self._serve, daemon=True)
+        self._thread.start()
+
+    def _serve(self) -> None:
+        try:
+            self.transport.start_server(server=self.server)
+        except Exception:  # the client hung up on a refused key
+            pass
+
+    def close(self) -> None:
+        self.transport.close()
+        self.client_sock.close()
+        self._thread.join(timeout=5)
+
+
+@pytest.fixture
+def host_key() -> paramiko.PKey:
+    return paramiko.ECDSAKey.generate()
+
+
+@pytest.fixture
+def host(host_key):
+    h = _Host(host_key)
+    yield h
+    h.close()
+
+
+def _connect(check: HostKeyCheck, host: _Host) -> paramiko.SSHClient:
+    client = paramiko.SSHClient()
+    check.connect(
+        client,
+        username="admin",
+        password="s3cret",
+        sock=host.client_sock,
+        look_for_keys=False,
+        allow_agent=False,
+        timeout=5,
+    )
+    return client
+
+
+def _check(expected: SshHostKey | None, seen: list, port: int = 22) -> HostKeyCheck:
+    args: dict = {OBSERVER_ARG: seen.append}
+    if expected is not None:
+        args[KNOWN_KEYS_ARG] = {str(port): expected.as_dict()}
+    return HostKeyCheck(args, "10.0.0.5", port)
+
+
+class TestSshHostKey:
+    def test_fingerprint_is_the_one_ssh_keygen_prints(self):
+        assert ED25519.fingerprint == ED25519_FP
+
+    def test_of_reads_a_paramiko_key(self, host_key):
+        key = SshHostKey.of(host_key)
+        assert key.key_type == host_key.get_name()
+        assert key.key == host_key.get_base64()
+
+    def test_as_dict_round_trips(self):
+        d = ED25519.as_dict()
+        assert d == {"key_type": "ssh-ed25519", "key": ED25519.key, "fingerprint": ED25519_FP}
+        assert SshHostKey.from_value(d) == ED25519
+
+    @pytest.mark.parametrize("value", [None, {}, ""])
+    def test_no_value_is_no_key(self, value):
+        assert SshHostKey.from_value(value) is None
+
+    @pytest.mark.parametrize(
+        "value", [{"key_type": "ssh-ed25519"}, {"key": "AAAA"}, "ssh-ed25519 AAAA"]
+    )
+    def test_a_malformed_value_is_refused(self, value):
+        with pytest.raises(ValueError):
+            SshHostKey.from_value(value)
+
+    def test_known_hosts_line_names_a_port_other_than_22(self):
+        assert ED25519.known_hosts_line("10.0.0.5", 22) == f"10.0.0.5 ssh-ed25519 {ED25519.key}"
+        line = ED25519.known_hosts_line("10.0.0.5", 2222)
+        assert line == f"[10.0.0.5]:2222 ssh-ed25519 {ED25519.key}"
+
+
+class TestKnownKeys:
+    def test_the_key_is_the_one_for_the_port_connected_to(self):
+        args = {KNOWN_KEYS_ARG: {"22": ED25519.as_dict()}}
+        assert HostKeyCheck(args, "10.0.0.5", 22).expected == ED25519
+        assert HostKeyCheck(args, "10.0.0.5", 2222).expected is None
+
+    def test_a_port_may_be_given_as_a_number(self):
+        args = {KNOWN_KEYS_ARG: {2222: ED25519.as_dict()}}
+        assert HostKeyCheck(args, "10.0.0.5", 2222).expected == ED25519
+
+    def test_a_malformed_mapping_is_refused(self):
+        with pytest.raises(ValueError):
+            HostKeyCheck({KNOWN_KEYS_ARG: [ED25519.as_dict()]}, "10.0.0.5", 22)
+
+
+class TestParamiko:
+    def test_first_use_accepts_the_key_and_reports_it(self, host, host_key):
+        seen: list = []
+        _connect(_check(None, seen), host).close()
+        assert host.server.passwords == ["s3cret"]
+        assert seen == [
+            {
+                "host": "10.0.0.5",
+                "port": 22,
+                "seen": SshHostKey.of(host_key).as_dict(),
+                "expected": None,
+                "accepted": True,
+            }
+        ]
+
+    def test_the_known_key_connects(self, host, host_key):
+        seen: list = []
+        known = SshHostKey.of(host_key)
+        _connect(_check(known, seen), host).close()
+        assert host.server.passwords == ["s3cret"]
+        assert seen[0]["accepted"] is True
+        assert seen[0]["expected"] == known.as_dict()
+
+    def test_a_changed_key_is_refused_before_any_login(self, host, host_key):
+        seen: list = []
+        other = SshHostKey.of(paramiko.ECDSAKey.generate())
+        with pytest.raises(SshHostKeyMismatch) as caught:
+            _connect(_check(other, seen), host)
+        assert host.server.passwords == []
+        assert caught.value.observation.seen == SshHostKey.of(host_key)
+        assert caught.value.observation.expected == other
+        assert seen[0]["accepted"] is False
+        assert seen[0]["seen"]["fingerprint"] == SshHostKey.of(host_key).fingerprint
+
+    def test_a_key_of_another_type_is_refused(self, host):
+        seen: list = []
+        with pytest.raises(SshHostKeyMismatch):
+            _connect(_check(ED25519, seen), host)
+        assert host.server.passwords == []
+
+    def test_the_mismatch_is_a_connection_failure_that_names_both_keys(self, host, host_key):
+        with pytest.raises(ConnectionException) as caught:
+            _connect(_check(ED25519, []), host)
+        message = str(caught.value)
+        assert "10.0.0.5:22" in message
+        assert SshHostKey.of(host_key).fingerprint in message
+        assert ED25519_FP in message
+
+    def test_a_failing_observer_does_not_break_the_connection(self, host):
+        def boom(_observation):
+            raise RuntimeError("store down")
+
+        check = HostKeyCheck({OBSERVER_ARG: boom}, "10.0.0.5", 22)
+        _connect(check, host).close()
+        assert host.server.passwords == ["s3cret"]
+
+    def test_without_optional_args_it_still_connects(self, host):
+        _connect(HostKeyCheck(None, "10.0.0.5", 22), host).close()
+        assert host.server.passwords == ["s3cret"]
+
+
+class TestNetmiko:
+    BASE = {
+        "ssh_strict": False,
+        "alt_host_keys": True,
+        "alt_key_file": "/x",
+        "system_host_keys": True,
+        "conn_timeout": 9,
+    }
+
+    def test_first_use_lets_netmiko_accept_the_key(self):
+        with HostKeyCheck({}, "10.0.0.5", 22).netmiko(self.BASE) as kwargs:
+            assert kwargs == {
+                "ssh_strict": False,
+                "alt_host_keys": False,
+                "system_host_keys": False,
+                "conn_timeout": 9,
+            }
+
+    def test_a_known_key_is_the_only_one_netmiko_takes(self):
+        check = HostKeyCheck({KNOWN_KEYS_ARG: {"2222": ED25519.as_dict()}}, "10.0.0.5", 2222)
+        with check.netmiko(self.BASE) as kwargs:
+            path = kwargs["alt_key_file"]
+            assert kwargs["ssh_strict"] is True
+            assert kwargs["alt_host_keys"] is True
+            assert kwargs["system_host_keys"] is False
+            assert kwargs["conn_timeout"] == 9
+            with open(path) as f:
+                assert f.read().strip() == ED25519.known_hosts_line("10.0.0.5", 2222)
+        assert not os.path.exists(path)
+
+    def test_a_refusal_inside_netmiko_becomes_the_mismatch(self, host_key):
+        # netmiko re-raises paramiko's BadHostKeyException as its own exception,
+        # inside the except block, and a driver wraps that once more.
+        seen: list = []
+        check = _check(ED25519, seen)
+        with pytest.raises(SshHostKeyMismatch) as caught:
+            with check.netmiko({}):
+                try:
+                    try:
+                        other = paramiko.ECDSAKey.generate()
+                        raise paramiko.BadHostKeyException("10.0.0.5", host_key, other)
+                    except paramiko.SSHException as exc:
+                        raise RuntimeError(f"A paramiko SSHException occurred: {exc}")
+                except RuntimeError as exc:
+                    raise ConnectionException(str(exc)) from exc
+        assert caught.value.observation.seen == SshHostKey.of(host_key)
+        assert caught.value.observation.expected == ED25519
+        assert seen[0]["accepted"] is False
+
+    def test_any_other_failure_passes_unchanged(self):
+        with pytest.raises(ConnectionException, match="timed out"):
+            with HostKeyCheck({KNOWN_KEYS_ARG: {"22": ED25519.as_dict()}}, "10.0.0.5", 22).netmiko({}):
+                raise ConnectionException("timed out")
+
+    def test_observe_reports_the_key_netmiko_saw(self, host_key):
+        class _Transport:
+            def get_remote_server_key(self):
+                return host_key
+
+        seen: list = []
+        observation = _check(None, seen).observe(_Transport())
+        assert observation.accepted is True
+        assert seen[0]["seen"] == SshHostKey.of(host_key).as_dict()