Christian Manivong
6179607137
CI / test (3.10) (push) Successful in 27s
CI / test (3.11) (push) Successful in 30s
CI / test (3.12) (push) Successful in 37s
CI / test (3.10) (pull_request) Successful in 32s
CI / test (3.11) (pull_request) Successful in 34s
CI / test (3.12) (pull_request) Successful in 29s
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).
110 lines
3.3 KiB
Python
110 lines
3.3 KiB
Python
"""open() verifies the host key it is given, and reports the one it sees (NetOrk/netork#1167)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import paramiko
|
|
import pytest
|
|
from napalm_device_types.ssh_host_keys import (
|
|
KNOWN_KEYS_ARG,
|
|
OBSERVER_ARG,
|
|
SshHostKey,
|
|
SshHostKeyMismatch,
|
|
)
|
|
|
|
from napalm_bsd import FreeBSDDriver, base
|
|
|
|
SERVER_KEY = paramiko.ECDSAKey.generate()
|
|
OTHER_KEY = paramiko.ECDSAKey.generate()
|
|
|
|
|
|
class _HostKeys:
|
|
def __init__(self) -> None:
|
|
self.lines: list[str] = []
|
|
|
|
def load(self, path: str) -> None:
|
|
with open(path) as f:
|
|
self.lines += f.read().splitlines()
|
|
|
|
|
|
class _Transport:
|
|
def get_remote_server_key(self):
|
|
return SERVER_KEY
|
|
|
|
def is_active(self) -> bool:
|
|
return True
|
|
|
|
|
|
class _Client:
|
|
"""paramiko.SSHClient as far as open() uses it; refuses like paramiko would."""
|
|
|
|
def __init__(self) -> None:
|
|
self.host_keys = _HostKeys()
|
|
self.policy = None
|
|
self.connected_with: dict | None = None
|
|
self.closed = False
|
|
|
|
def set_missing_host_key_policy(self, policy) -> None:
|
|
self.policy = policy
|
|
|
|
def get_host_keys(self) -> _HostKeys:
|
|
return self.host_keys
|
|
|
|
def connect(self, hostname, **kwargs) -> None:
|
|
if self.host_keys.lines and SshHostKey.of(SERVER_KEY).key not in self.host_keys.lines[0]:
|
|
raise paramiko.BadHostKeyException(hostname, SERVER_KEY, OTHER_KEY)
|
|
if not self.host_keys.lines:
|
|
self.policy.missing_host_key(self, hostname, SERVER_KEY)
|
|
self.connected_with = {"hostname": hostname, **kwargs}
|
|
|
|
def get_transport(self) -> _Transport:
|
|
return _Transport()
|
|
|
|
def close(self) -> None:
|
|
self.closed = True
|
|
|
|
|
|
@pytest.fixture
|
|
def client(monkeypatch) -> _Client:
|
|
fake = _Client()
|
|
monkeypatch.setattr(base.paramiko, "SSHClient", lambda: fake)
|
|
return fake
|
|
|
|
|
|
def _driver(seen: list, known: SshHostKey | None = None) -> FreeBSDDriver:
|
|
args: dict = {"port": 2222, OBSERVER_ARG: seen.append}
|
|
if known is not None:
|
|
args[KNOWN_KEYS_ARG] = {"2222": known.as_dict()}
|
|
return FreeBSDDriver("10.0.0.7", "netork", "s3cret", optional_args=args)
|
|
|
|
|
|
def test_first_use_connects_and_reports_the_key(client):
|
|
seen: list = []
|
|
_driver(seen).open()
|
|
assert client.connected_with["password"] == "s3cret"
|
|
assert client.connected_with["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.7", 2222)
|
|
|
|
|
|
def test_the_known_key_connects(client):
|
|
seen: list = []
|
|
_driver(seen, SshHostKey.of(SERVER_KEY)).open()
|
|
assert client.host_keys.lines == [SshHostKey.of(SERVER_KEY).known_hosts_line("10.0.0.7", 2222)]
|
|
assert client.connected_with is not None
|
|
assert seen[0]["accepted"] is True
|
|
|
|
|
|
def test_a_changed_key_is_refused_and_the_client_closed(client):
|
|
seen: list = []
|
|
with pytest.raises(SshHostKeyMismatch):
|
|
_driver(seen, SshHostKey.of(OTHER_KEY)).open()
|
|
assert client.connected_with is None
|
|
assert client.closed is True
|
|
assert seen[0]["accepted"] is False
|
|
|
|
|
|
def test_without_host_key_args_it_connects_as_before(client):
|
|
FreeBSDDriver("10.0.0.7", "netork", "s3cret").open()
|
|
assert client.connected_with is not None
|