Christian Manivong
fe0aec9432
CI / test (3.10) (push) Successful in 32s
CI / test (3.11) (push) Successful in 37s
CI / test (3.12) (push) Successful in 35s
CI / test (3.10) (pull_request) Successful in 29s
CI / test (3.11) (pull_request) Successful in 28s
CI / test (3.12) (pull_request) Successful in 28s
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).
113 lines
4.1 KiB
Python
113 lines
4.1 KiB
Python
"""Every SSH session verifies the switch's host key and reports it (NetOrk/netork#1167).
|
|
|
|
The CLI over SSH (standard and legacy), the running-config fallback on port 22
|
|
and the SNMP fix's temporary session all connect through ``_ssh_connect``. A
|
|
refused key ends the attempt: no legacy retry, and above all no fall-back to
|
|
Telnet, which would hand the same host the password in clear text.
|
|
"""
|
|
|
|
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_procurve.procurve import ProcurveDriver
|
|
|
|
SERVER_KEY = paramiko.ECDSAKey.generate()
|
|
OTHER_KEY = paramiko.ECDSAKey.generate()
|
|
HANDLER = "napalm_procurve.procurve.ConnectHandler"
|
|
|
|
|
|
def _device() -> MagicMock:
|
|
device = MagicMock()
|
|
device.remote_conn_pre.get_transport.return_value.get_remote_server_key.return_value = SERVER_KEY
|
|
device.send_command.return_value = "hostname x"
|
|
return device
|
|
|
|
|
|
def _refuse(**kwargs):
|
|
if kwargs["device_type"] == ProcurveDriver.NETMIKO_DEVICE_TYPE_TELNET:
|
|
raise AssertionError("fell back to Telnet after a refused host key")
|
|
try:
|
|
raise paramiko.BadHostKeyException("10.0.0.4", SERVER_KEY, OTHER_KEY)
|
|
except paramiko.SSHException as exc:
|
|
raise NetmikoTimeoutException(f"A paramiko SSHException occurred: {exc}")
|
|
|
|
|
|
def _driver(seen: list, known: dict | None = None, **extra) -> ProcurveDriver:
|
|
args: dict = {OBSERVER_ARG: seen.append, **extra}
|
|
if known is not None:
|
|
args[KNOWN_KEYS_ARG] = known
|
|
return ProcurveDriver("10.0.0.4", "manager", "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, transport="ssh").open()
|
|
assert handler.call_args.kwargs["ssh_strict"] is False
|
|
assert seen[0]["accepted"] is True
|
|
assert (seen[0]["host"], seen[0]["port"]) == ("10.0.0.4", 22)
|
|
|
|
|
|
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.4", 22)
|
|
assert kwargs["ssh_strict"] is True
|
|
return _device()
|
|
|
|
with patch(HANDLER, side_effect=connect):
|
|
_driver(seen, {"22": known.as_dict()}, transport="ssh").open()
|
|
assert not os.path.exists(files[0])
|
|
assert seen[0]["accepted"] is True
|
|
|
|
|
|
def test_a_changed_key_stops_open_without_legacy_or_telnet():
|
|
seen: list = []
|
|
driver = _driver(seen, {"22": SshHostKey.of(OTHER_KEY).as_dict()})
|
|
with patch("napalm_procurve.procurve.ProcurveApiClient.probe", return_value=(None, None)):
|
|
with patch(HANDLER, side_effect=_refuse) as handler:
|
|
with pytest.raises(SshHostKeyMismatch):
|
|
driver.open()
|
|
assert handler.call_count == 1
|
|
assert driver._device is None
|
|
assert [s["accepted"] for s in seen] == [False]
|
|
|
|
|
|
def test_the_config_fallback_checks_the_key_of_port_22():
|
|
seen: list = []
|
|
driver = _driver(seen, {"22": SshHostKey.of(SERVER_KEY).as_dict()}, port=443)
|
|
with patch(HANDLER, return_value=_device()) as handler:
|
|
assert driver._get_config_via_ssh() == "hostname x"
|
|
assert handler.call_args.kwargs["port"] == 22
|
|
assert handler.call_args.kwargs["ssh_strict"] is True
|
|
assert seen[0]["port"] == 22 and seen[0]["accepted"] is True
|
|
|
|
|
|
def test_the_snmp_fix_over_rest_stops_at_a_changed_key():
|
|
seen: list = []
|
|
driver = _driver(seen, {"443": SshHostKey.of(OTHER_KEY).as_dict()}, port=443)
|
|
driver._transport = "api"
|
|
driver._api = MagicMock()
|
|
with patch(HANDLER, side_effect=_refuse) as handler:
|
|
result = driver.configure_snmp_agent({"community": "c0mm", "allowed_sources": []})
|
|
assert result["success"] is False
|
|
assert handler.call_count == 1
|
|
assert "host key" in str(result)
|