From cb14f1f00ffcf67dcd2c27be093d89b80c8c181c Mon Sep 17 00:00:00 2001 From: Christian Manivong Date: Sun, 18 Oct 2026 21:23:40 +0200 Subject: [PATCH] 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). --- napalm_openwrt/openwrt.py | 25 ++++++---- pyproject.toml | 4 +- tests/unit/test_ssh_host_key.py | 84 +++++++++++++++++++++++++++++++++ 3 files changed, 102 insertions(+), 11 deletions(-) create mode 100644 tests/unit/test_ssh_host_key.py diff --git a/napalm_openwrt/openwrt.py b/napalm_openwrt/openwrt.py index c02a1dd..08434cd 100644 --- a/napalm_openwrt/openwrt.py +++ b/napalm_openwrt/openwrt.py @@ -42,6 +42,7 @@ from napalm.base.exceptions import ( ConnectionClosedException, ) from napalm.base.netmiko_helpers import netmiko_args +from napalm_device_types.ssh_host_keys import HostKeyCheck, SshHostKeyMismatch from napalm_openwrt.interfaces_mixin import OpenWrtInterfaceMixin from napalm_openwrt.wireless_mixin import OpenWrtWirelessMixin @@ -117,6 +118,8 @@ class OpenWrtDriver( self.port = optional_args.pop("port", 22) self.netmiko_optional_args = netmiko_args(optional_args) + # The host key to expect, and where the one seen is reported (netork#1167). + self._host_key = HostKeyCheck(optional_args, hostname, self.port) # Config management state self._candidate_config: str | None = None @@ -130,15 +133,18 @@ class OpenWrtDriver( def open(self) -> None: """Open an SSH connection to the device.""" try: - self.device = ConnectHandler( - device_type=self.NETMIKO_DEVICE_TYPE, - host=self.hostname, - username=self.username, - password=self.password, - timeout=self.timeout, - port=self.port, - **self.netmiko_optional_args, - ) + with self._host_key.netmiko(self.netmiko_optional_args) as netmiko_kwargs: + self.device = ConnectHandler( + device_type=self.NETMIKO_DEVICE_TYPE, + host=self.hostname, + username=self.username, + password=self.password, + timeout=self.timeout, + port=self.port, + **netmiko_kwargs, + ) + except SshHostKeyMismatch: + raise # refused before the login; nothing was sent except NetmikoTimeoutException as exc: raise ConnectionException( f"Cannot connect to {self.hostname}: {exc}" @@ -147,6 +153,7 @@ class OpenWrtDriver( raise ConnectionException( f"Authentication failed for {self.hostname}: {exc}" ) from exc + self._host_key.observe(self.device.remote_conn_pre.get_transport()) def close(self) -> None: """Close the SSH connection.""" diff --git a/pyproject.toml b/pyproject.toml index 52e591e..ae770cb 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "napalm-openwrt" -version = "0.5.0" +version = "0.6.0" description = "NAPALM driver for OpenWrt routers/access-points" readme = "README.md" license = { text = "Apache-2.0" } @@ -26,7 +26,7 @@ classifiers = [ ] dependencies = [ "napalm>=4.0.0", - "napalm_device_types>=4.13.0", + "napalm_device_types>=4.19.0", "netmiko>=4.0.0", "paramiko>=5.0.0", # CVE-2026-44405 "netaddr", diff --git a/tests/unit/test_ssh_host_key.py b/tests/unit/test_ssh_host_key.py new file mode 100644 index 0000000..5193dfa --- /dev/null +++ b/tests/unit/test_ssh_host_key.py @@ -0,0 +1,84 @@ +"""open() hands netmiko the host key to expect, and reports the one it saw (NetOrk/netork#1167).""" + +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_openwrt.openwrt import OpenWrtDriver + +SERVER_KEY = paramiko.ECDSAKey.generate() +OTHER_KEY = paramiko.ECDSAKey.generate() +HANDLER = "napalm_openwrt.openwrt.ConnectHandler" + + +def _device() -> MagicMock: + device = MagicMock() + 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) -> OpenWrtDriver: + args: dict = {"port": 2222, OBSERVER_ARG: seen.append} + if known is not None: + args[KNOWN_KEYS_ARG] = {"2222": known.as_dict()} + return OpenWrtDriver("10.0.0.8", "admin", "s3cret", optional_args=args) + + +def test_first_use_lets_netmiko_accept_and_reports_the_key(): + seen: list = [] + with patch(HANDLER, 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 kwargs["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.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(HANDLER, 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(HANDLER, side_effect=refuse): + with pytest.raises(SshHostKeyMismatch): + driver.open() + assert driver.device is None + assert seen[0]["accepted"] is False