Christian Manivong
fc24495b83
- The six except Exception: pass (REST logout, the extra-ports lookup, the per-port detail in get_interfaces, closing CLI and REST sessions) now log at debug level (B110), so a swallowed failure leaves a trace. - nosec for comparing a reported address with 0.0.0.0 (B104) and for the empty password field of get_users (B105, now a named constant); both are false positives. - A long error message is split into adjacent f-strings; the unused ARP entry type is named _entry_type. - tests: the failed-open test expects ConnectionException instead of any Exception; raise ... from exc.
115 lines
4.1 KiB
Python
115 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}") from 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)
|