Christian Manivong
6bb3035857
CI / test (3.10) (push) Successful in 54s
CI / test (3.11) (push) Successful in 40s
CI / test (3.12) (push) Successful in 51s
CI / test (3.10) (pull_request) Successful in 42s
CI / test (3.11) (pull_request) Successful in 35s
CI / test (3.12) (pull_request) Successful in 49s
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).
134 lines
4.2 KiB
Python
134 lines
4.2 KiB
Python
"""The node's SSH session verifies its host key, and reports the one it sees (NetOrk/netork#1167).
|
|
|
|
Both ways onto the node's shell — the best-effort ``_exec_ssh_command`` and the
|
|
provisioning steps' ``_run_node_command`` — share one session, opened on port 22.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from unittest.mock import MagicMock
|
|
|
|
import paramiko
|
|
import pytest
|
|
from napalm_device_types.ssh_host_keys import (
|
|
KNOWN_KEYS_ARG,
|
|
OBSERVER_ARG,
|
|
SshHostKey,
|
|
SshHostKeyMismatch,
|
|
)
|
|
|
|
from napalm_proxmox.driver import ProxmoxDriver
|
|
|
|
SERVER_KEY = paramiko.ECDSAKey.generate()
|
|
OTHER_KEY = paramiko.ECDSAKey.generate()
|
|
|
|
|
|
class _Stream:
|
|
def __init__(self, data: bytes) -> None:
|
|
self._data = data
|
|
self.channel = MagicMock()
|
|
self.channel.recv_exit_status.return_value = 0
|
|
|
|
def read(self) -> bytes:
|
|
return self._data
|
|
|
|
|
|
class _Client:
|
|
"""paramiko.SSHClient as far as the driver uses it; refuses like paramiko would."""
|
|
|
|
def __init__(self) -> None:
|
|
self.lines: list[str] = []
|
|
self.policy = None
|
|
self.connected_with: dict | None = None
|
|
self.commands: list[str] = []
|
|
|
|
def set_missing_host_key_policy(self, policy) -> None:
|
|
self.policy = policy
|
|
|
|
def get_host_keys(self):
|
|
client = self
|
|
|
|
class _Keys:
|
|
def load(self, path: str) -> None:
|
|
with open(path) as f:
|
|
client.lines += f.read().splitlines()
|
|
|
|
return _Keys()
|
|
|
|
def connect(self, hostname, **kwargs) -> None:
|
|
if self.lines and SshHostKey.of(SERVER_KEY).key not in self.lines[0]:
|
|
raise paramiko.BadHostKeyException(hostname, SERVER_KEY, OTHER_KEY)
|
|
if not self.lines:
|
|
self.policy.missing_host_key(self, hostname, SERVER_KEY)
|
|
self.connected_with = {"hostname": hostname, **kwargs}
|
|
|
|
def get_transport(self):
|
|
transport = MagicMock()
|
|
transport.get_remote_server_key.return_value = SERVER_KEY
|
|
return transport
|
|
|
|
def exec_command(self, command, timeout=None):
|
|
self.commands.append(command)
|
|
return None, _Stream(b"ok"), _Stream(b"")
|
|
|
|
def close(self) -> None:
|
|
pass
|
|
|
|
|
|
@pytest.fixture
|
|
def client(monkeypatch) -> _Client:
|
|
fake = _Client()
|
|
monkeypatch.setattr(paramiko, "SSHClient", lambda: fake)
|
|
return fake
|
|
|
|
|
|
def _driver(seen: list, known: dict | None = None) -> ProxmoxDriver:
|
|
args: dict = {"port": 8006, OBSERVER_ARG: seen.append}
|
|
if known is not None:
|
|
args[KNOWN_KEYS_ARG] = known
|
|
driver = ProxmoxDriver("10.0.0.9", "root", "s3cret", optional_args=args)
|
|
driver._node_api = MagicMock(side_effect=RuntimeError("no exec endpoint")) # type: ignore[method-assign]
|
|
return driver
|
|
|
|
|
|
def test_first_use_runs_the_command_and_reports_the_key(client):
|
|
seen: list = []
|
|
assert _driver(seen)._run_node_command("true", timeout=5) == "ok"
|
|
assert client.connected_with["port"] == 22
|
|
assert seen[0]["accepted"] is True
|
|
assert (seen[0]["host"], seen[0]["port"]) == ("10.0.0.9", 22)
|
|
|
|
|
|
def test_the_key_for_the_api_port_is_not_the_ssh_one(client):
|
|
seen: list = []
|
|
other = SshHostKey.of(OTHER_KEY).as_dict()
|
|
_driver(seen, {"8006": other})._run_node_command("true", timeout=5)
|
|
assert seen[0]["expected"] is None
|
|
|
|
|
|
def test_a_changed_key_stops_a_provisioning_step(client):
|
|
seen: list = []
|
|
driver = _driver(seen, {"22": SshHostKey.of(OTHER_KEY).as_dict()})
|
|
with pytest.raises(SshHostKeyMismatch):
|
|
driver._run_node_command("qm importdisk …", timeout=5)
|
|
assert client.commands == []
|
|
assert client.connected_with is None
|
|
assert seen[0]["accepted"] is False
|
|
|
|
|
|
def test_a_changed_key_leaves_the_best_effort_read_empty(client):
|
|
seen: list = []
|
|
driver = _driver(seen, {"22": SshHostKey.of(OTHER_KEY).as_dict()})
|
|
assert driver._exec_ssh_command("cat /etc/os-release") == ""
|
|
assert client.commands == []
|
|
assert seen[0]["accepted"] is False
|
|
|
|
|
|
def test_the_known_key_reuses_one_session(client):
|
|
seen: list = []
|
|
driver = _driver(seen, {"22": SshHostKey.of(SERVER_KEY).as_dict()})
|
|
driver._exec_ssh_command("uptime")
|
|
driver._run_node_command("true", timeout=5)
|
|
assert client.commands == ["uptime", "true"]
|
|
assert len(seen) == 1 and seen[0]["accepted"] is True
|