fix(ssh): verify the host key netOrk expects, report the one seen (NetOrk/netork#1167)
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
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).
This commit is contained in:
1 parent
68dd818da6
commit
6179607137
4 files changed
+122
-6
No files matched your search
@@ -0,0 +1,109 @@
|
||||
"""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
|
||||
Reference in new issue
Block a user