Christian Manivong
8b8905f0c7
- Long lines are wrapped: the get_environment awk command is split into adjacent string literals, the rollback message and the system mixin's docstring are wrapped; the command and the text are unchanged. - The two except Exception: pass (the snmpd community in get_snmp_config, the network VLANs in get_ssids) now log at debug level (B110), so a swallowed failure leaves a trace. - tests: raise ... from exc, unused loop variable.
87 lines
2.8 KiB
Python
87 lines
2.8 KiB
Python
"""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}") from 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
|