From 94bf399d34787a10697271b2407da03720621d71 Mon Sep 17 00:00:00 2001 From: Christian Manivong Date: Sun, 18 Oct 2026 21:23:25 +0200 Subject: [PATCH] feat: ssh_host_keys.HostKeyCheck, trust on first use (NetOrk/netork#1167) Drivers get the host keys to expect in optional_args["ssh_host_keys"] (by port) and report what they saw through optional_args["ssh_host_key_observer"]. With a known key, any other key is refused before authentication (SshHostKeyMismatch); without one, the first key is accepted. One implementation for paramiko (HostKeyCheck.connect) and netmiko (HostKeyCheck.netmiko + observe). A caller or driver that ignores both keys keeps working as before. Version 4.19.0. --- README.md | 21 ++ napalm_device_types/__init__.py | 14 ++ napalm_device_types/ssh_host_keys.py | 300 +++++++++++++++++++++++++++ pyproject.toml | 2 +- tests/test_ssh_host_keys.py | 275 ++++++++++++++++++++++++ 5 files changed, 611 insertions(+), 1 deletion(-) create mode 100644 napalm_device_types/ssh_host_keys.py create mode 100644 tests/test_ssh_host_keys.py diff --git a/README.md b/README.md index c12f40d..e4367f6 100644 --- a/README.md +++ b/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": ""}}`, + 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 diff --git a/napalm_device_types/__init__.py b/napalm_device_types/__init__.py index c198e33..b5f0935 100644 --- a/napalm_device_types/__init__.py +++ b/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", diff --git a/napalm_device_types/ssh_host_keys.py b/napalm_device_types/ssh_host_keys.py new file mode 100644 index 0000000..5c86e3f --- /dev/null +++ b/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": ""}}``, 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 diff --git a/pyproject.toml b/pyproject.toml index 552d896..39867e8 100644 --- a/pyproject.toml +++ b/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" diff --git a/tests/test_ssh_host_keys.py b/tests/test_ssh_host_keys.py new file mode 100644 index 0000000..0a56bee --- /dev/null +++ b/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() -- 2.54.0