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

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:39 +02:00
1 parent 5951a6100e
commit 4b17e2f3f8
4 files changed
+106 -12

No files matched your search

+15 -10
View File
@@ -59,6 +59,7 @@ from napalm_device_types import (
run_on_transport,
strip_terminal_codes,
)
from napalm_device_types.ssh_host_keys import HostKeyCheck
from napalm_device_types.models import (
ApplyUpdatesResultDict,
CronJobDict,
@@ -356,6 +357,8 @@ class LinuxDriver(
# port is passed explicitly in open() — remove it from netmiko_optional_args
# to avoid "multiple values for keyword argument 'port'"
self.netmiko_optional_args.pop("port", None)
# The host key to expect, and where the one seen is reported (netork#1167).
self._host_key = HostKeyCheck(optional_args, hostname, self.port)
# Runtime state
self._device: ConnectHandler | None = None
@@ -368,20 +371,22 @@ class LinuxDriver(
def open(self) -> None:
"""Open the SSH connection and detect the package manager."""
try:
self._device = ConnectHandler(
device_type=self.NETMIKO_DEVICE_TYPE,
host=self.hostname,
username=self.username,
password=self.password,
port=self.port,
secret=self._secret,
timeout=self.timeout,
**self.netmiko_optional_args,
)
with self._host_key.netmiko(self.netmiko_optional_args) as netmiko_args_:
self._device = ConnectHandler(
device_type=self.NETMIKO_DEVICE_TYPE,
host=self.hostname,
username=self.username,
password=self.password,
port=self.port,
secret=self._secret,
timeout=self.timeout,
**netmiko_args_,
)
except NetmikoAuthenticationException as exc:
raise ConnectionException(str(exc)) from exc
except NetmikoTimeoutException as exc:
raise ConnectionException(str(exc)) from exc
self._host_key.observe(self._device.remote_conn_pre.get_transport())
# No history before anything else is typed (NetOrk/netork#1039).
try: