Christian Manivong
bef2f23ab2
- set_interface: the except Exception: pass around reading the old untagged VLANs now logs at debug level (B110), with a module logger. - nosec for the empty password field of get_users (B105, now a named constant); a false positive. - Two docstrings quoting tabular device output keep their lines (noqa E501); a long docstring and an error message are shortened or split; the unused duplex column is named _duplex. - tests: a get_facts test named SHOW_INTERFACE, which does not exist; it is SHOW_INTERFACE_STATUS. raise ... from exc.
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_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}") 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
|