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

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:40 +02:00
1 parent fddb5c6772
commit cb14f1f00f
3 files changed
+102 -11

No files matched your search

+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_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