From e4f57d3c3ab2f175e953caef651fb3dcf617f905 Mon Sep 17 00:00:00 2001 From: Christian Manivong Date: Sun, 18 Oct 2026 21:23:42 +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_tplink_jetstream/tplink_jetstream.py | 23 ++++-- pyproject.toml | 4 +- tests/unit/test_ssh_host_key.py | 84 +++++++++++++++++++++ 3 files changed, 101 insertions(+), 10 deletions(-) create mode 100644 tests/unit/test_ssh_host_key.py diff --git a/napalm_tplink_jetstream/tplink_jetstream.py b/napalm_tplink_jetstream/tplink_jetstream.py index 9ce92c5..76921fd 100644 --- a/napalm_tplink_jetstream/tplink_jetstream.py +++ b/napalm_tplink_jetstream/tplink_jetstream.py @@ -38,6 +38,7 @@ from napalm.base.exceptions import ( ReplaceConfigException, ) from napalm.base.netmiko_helpers import netmiko_args +from napalm_device_types.ssh_host_keys import HostKeyCheck, SshHostKeyMismatch import napalm.base.constants as C @@ -99,6 +100,8 @@ class TPLinkJetstreamDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver): self.port = optional_args.get("port", 22) self.use_canonical_interface = optional_args.get("canonical_int_fmt", False) 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 @@ -112,16 +115,20 @@ class TPLinkJetstreamDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver): 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, - **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, + **netmiko_kwargs, + ) + self._host_key.observe(self.device.remote_conn_pre.get_transport()) if not self.force_no_enable: self.device.enable() + except SshHostKeyMismatch: + raise # refused before the login; nothing was sent except NetmikoTimeoutException as exc: raise ConnectionException( f"Cannot connect to {self.hostname}: {exc}" diff --git a/pyproject.toml b/pyproject.toml index a2152b5..7f85cc4 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "napalm-tplink-jetstream" -version = "0.3.0" +version = "0.4.0" description = "NAPALM driver for TP-Link Jetstream managed switches" readme = "README.md" license = { text = "Apache-2.0" } @@ -24,7 +24,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..c6a1afb --- /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_tplink_jetstream.tplink_jetstream import TPLinkJetstreamDriver + +SERVER_KEY = paramiko.ECDSAKey.generate() +OTHER_KEY = paramiko.ECDSAKey.generate() +HANDLER = "napalm_tplink_jetstream.tplink_jetstream.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) -> TPLinkJetstreamDriver: + args: dict = {"port": 2222, OBSERVER_ARG: seen.append} + if known is not None: + args[KNOWN_KEYS_ARG] = {"2222": known.as_dict()} + return TPLinkJetstreamDriver("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