fix(ssh): verify the host key netOrk expects, report the one seen (NetOrk/netork#1167)
CI / test (3.10) (push) Successful in 33s
CI / test (3.11) (push) Successful in 32s
CI / test (3.12) (push) Successful in 31s
CI / test (3.10) (pull_request) Successful in 27s
CI / test (3.11) (pull_request) Successful in 24s
CI / test (3.12) (pull_request) Successful in 26s

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).
This commit is contained in:
Christian Manivong committed 2026-10-18 21:23:41 +02:00
1 parent 8062188b12
commit 2de5844f92
3 files changed
+101 -10

No files matched your search

+15 -8
View File
@@ -46,6 +46,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
@@ -126,6 +127,8 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
optional_args.setdefault("banner_timeout", 20)
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: Optional[str] = None
@@ -146,16 +149,20 @@ class NetgearSmartDriver(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}") from exc
except NetmikoAuthenticationException as exc:
+2 -2
View File
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "napalm-netgear"
version = "0.4.0"
version = "0.5.0"
description = "NAPALM driver for Netgear switches (Plus/Smart)"
readme = "README.md"
license = { text = "Apache-2.0" }
@@ -26,7 +26,7 @@ classifiers = [
]
dependencies = [
"napalm>=4.0.0",
"napalm-device-types>=4.16.0",
"napalm-device-types>=4.19.0",
"netmiko>=4.0.0",
"paramiko>=5.0.0", # CVE-2026-44405
"netaddr",
+84
View File
@@ -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_netgear.netgear_smart import NetgearSmartDriver
SERVER_KEY = paramiko.ECDSAKey.generate()
OTHER_KEY = paramiko.ECDSAKey.generate()
HANDLER = "napalm_netgear.netgear_smart.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) -> NetgearSmartDriver:
args: dict = {"port": 2222, OBSERVER_ARG: seen.append}
if known is not None:
args[KNOWN_KEYS_ARG] = {"2222": known.as_dict()}
return NetgearSmartDriver("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