feat: ssh_host_keys.HostKeyCheck, trust on first use (NetOrk/netork#1167)

Drivers get the host keys to expect in optional_args["ssh_host_keys"] (by port)
and report what they saw through optional_args["ssh_host_key_observer"].
With a known key, any other key is refused before authentication
(SshHostKeyMismatch); without one, the first key is accepted. One
implementation for paramiko (HostKeyCheck.connect) and netmiko
(HostKeyCheck.netmiko + observe). A caller or driver that ignores both keys
keeps working as before.

Version 4.19.0.

tests/test_ssh_host_keys.py

@@ -0,0 +1,275 @@
+"""SSH host keys: trust on first use, and a changed key refused before any login.
+
+The layer above keeps the store (NetOrk/netork#1167). A driver gets the key it
+should expect through ``optional_args["ssh_host_keys"]`` (by port) and reports every key it
+sees through ``optional_args["ssh_host_key_observer"]``. These tests run real
+paramiko against an in-process SSH server, because the property that matters —
+nothing is sent to a host with the wrong key — is paramiko's behaviour once the
+key is loaded, not something a fake could show.
+"""
+
+from __future__ import annotations
+
+import os
+import socket
+import threading
+
+import paramiko
+import pytest
+from napalm.base.exceptions import ConnectionException
+
+from napalm_device_types.ssh_host_keys import (
+    KNOWN_KEYS_ARG,
+    OBSERVER_ARG,
+    HostKeyCheck,
+    SshHostKey,
+    SshHostKeyMismatch,
+)
+
+#: An ed25519 public key and the fingerprint ``ssh-keygen -lf`` prints for it.
+ED25519 = SshHostKey(
+    "ssh-ed25519", "AAAAC3NzaC1lZDI1NTE5AAAAIJH2oO06vroJT2ooSB3/2tu8HufXQng4xQlxAk/4raee"
+)
+ED25519_FP = "SHA256:26dxfTuve20LGGTpJ1T0ns11/ID3XMhbOahQq3k75TU"
+
+
+class _Server(paramiko.ServerInterface):
+    """Accepts any password and remembers each one it was offered."""
+
+    def __init__(self) -> None:
+        self.passwords: list[str] = []
+
+    def get_allowed_auths(self, username):
+        return "password"
+
+    def check_auth_password(self, username, password):
+        self.passwords.append(password)
+        return paramiko.AUTH_SUCCESSFUL
+
+
+class _Host:
+    """An SSH server on one end of a socket pair, with *host_key*."""
+
+    def __init__(self, host_key: paramiko.PKey) -> None:
+        self.client_sock, server_sock = socket.socketpair()
+        self.transport = paramiko.Transport(server_sock)
+        self.transport.add_server_key(host_key)
+        self.server = _Server()
+        self._thread = threading.Thread(target=self._serve, daemon=True)
+        self._thread.start()
+
+    def _serve(self) -> None:
+        try:
+            self.transport.start_server(server=self.server)
+        except Exception:  # the client hung up on a refused key
+            pass
+
+    def close(self) -> None:
+        self.transport.close()
+        self.client_sock.close()
+        self._thread.join(timeout=5)
+
+
+@pytest.fixture
+def host_key() -> paramiko.PKey:
+    return paramiko.ECDSAKey.generate()
+
+
+@pytest.fixture
+def host(host_key):
+    h = _Host(host_key)
+    yield h
+    h.close()
+
+
+def _connect(check: HostKeyCheck, host: _Host) -> paramiko.SSHClient:
+    client = paramiko.SSHClient()
+    check.connect(
+        client,
+        username="admin",
+        password="s3cret",
+        sock=host.client_sock,
+        look_for_keys=False,
+        allow_agent=False,
+        timeout=5,
+    )
+    return client
+
+
+def _check(expected: SshHostKey | None, seen: list, port: int = 22) -> HostKeyCheck:
+    args: dict = {OBSERVER_ARG: seen.append}
+    if expected is not None:
+        args[KNOWN_KEYS_ARG] = {str(port): expected.as_dict()}
+    return HostKeyCheck(args, "10.0.0.5", port)
+
+
+class TestSshHostKey:
+    def test_fingerprint_is_the_one_ssh_keygen_prints(self):
+        assert ED25519.fingerprint == ED25519_FP
+
+    def test_of_reads_a_paramiko_key(self, host_key):
+        key = SshHostKey.of(host_key)
+        assert key.key_type == host_key.get_name()
+        assert key.key == host_key.get_base64()
+
+    def test_as_dict_round_trips(self):
+        d = ED25519.as_dict()
+        assert d == {"key_type": "ssh-ed25519", "key": ED25519.key, "fingerprint": ED25519_FP}
+        assert SshHostKey.from_value(d) == ED25519
+
+    @pytest.mark.parametrize("value", [None, {}, ""])
+    def test_no_value_is_no_key(self, value):
+        assert SshHostKey.from_value(value) is None
+
+    @pytest.mark.parametrize(
+        "value", [{"key_type": "ssh-ed25519"}, {"key": "AAAA"}, "ssh-ed25519 AAAA"]
+    )
+    def test_a_malformed_value_is_refused(self, value):
+        with pytest.raises(ValueError):
+            SshHostKey.from_value(value)
+
+    def test_known_hosts_line_names_a_port_other_than_22(self):
+        assert ED25519.known_hosts_line("10.0.0.5", 22) == f"10.0.0.5 ssh-ed25519 {ED25519.key}"
+        line = ED25519.known_hosts_line("10.0.0.5", 2222)
+        assert line == f"[10.0.0.5]:2222 ssh-ed25519 {ED25519.key}"
+
+
+class TestKnownKeys:
+    def test_the_key_is_the_one_for_the_port_connected_to(self):
+        args = {KNOWN_KEYS_ARG: {"22": ED25519.as_dict()}}
+        assert HostKeyCheck(args, "10.0.0.5", 22).expected == ED25519
+        assert HostKeyCheck(args, "10.0.0.5", 2222).expected is None
+
+    def test_a_port_may_be_given_as_a_number(self):
+        args = {KNOWN_KEYS_ARG: {2222: ED25519.as_dict()}}
+        assert HostKeyCheck(args, "10.0.0.5", 2222).expected == ED25519
+
+    def test_a_malformed_mapping_is_refused(self):
+        with pytest.raises(ValueError):
+            HostKeyCheck({KNOWN_KEYS_ARG: [ED25519.as_dict()]}, "10.0.0.5", 22)
+
+
+class TestParamiko:
+    def test_first_use_accepts_the_key_and_reports_it(self, host, host_key):
+        seen: list = []
+        _connect(_check(None, seen), host).close()
+        assert host.server.passwords == ["s3cret"]
+        assert seen == [
+            {
+                "host": "10.0.0.5",
+                "port": 22,
+                "seen": SshHostKey.of(host_key).as_dict(),
+                "expected": None,
+                "accepted": True,
+            }
+        ]
+
+    def test_the_known_key_connects(self, host, host_key):
+        seen: list = []
+        known = SshHostKey.of(host_key)
+        _connect(_check(known, seen), host).close()
+        assert host.server.passwords == ["s3cret"]
+        assert seen[0]["accepted"] is True
+        assert seen[0]["expected"] == known.as_dict()
+
+    def test_a_changed_key_is_refused_before_any_login(self, host, host_key):
+        seen: list = []
+        other = SshHostKey.of(paramiko.ECDSAKey.generate())
+        with pytest.raises(SshHostKeyMismatch) as caught:
+            _connect(_check(other, seen), host)
+        assert host.server.passwords == []
+        assert caught.value.observation.seen == SshHostKey.of(host_key)
+        assert caught.value.observation.expected == other
+        assert seen[0]["accepted"] is False
+        assert seen[0]["seen"]["fingerprint"] == SshHostKey.of(host_key).fingerprint
+
+    def test_a_key_of_another_type_is_refused(self, host):
+        seen: list = []
+        with pytest.raises(SshHostKeyMismatch):
+            _connect(_check(ED25519, seen), host)
+        assert host.server.passwords == []
+
+    def test_the_mismatch_is_a_connection_failure_that_names_both_keys(self, host, host_key):
+        with pytest.raises(ConnectionException) as caught:
+            _connect(_check(ED25519, []), host)
+        message = str(caught.value)
+        assert "10.0.0.5:22" in message
+        assert SshHostKey.of(host_key).fingerprint in message
+        assert ED25519_FP in message
+
+    def test_a_failing_observer_does_not_break_the_connection(self, host):
+        def boom(_observation):
+            raise RuntimeError("store down")
+
+        check = HostKeyCheck({OBSERVER_ARG: boom}, "10.0.0.5", 22)
+        _connect(check, host).close()
+        assert host.server.passwords == ["s3cret"]
+
+    def test_without_optional_args_it_still_connects(self, host):
+        _connect(HostKeyCheck(None, "10.0.0.5", 22), host).close()
+        assert host.server.passwords == ["s3cret"]
+
+
+class TestNetmiko:
+    BASE = {
+        "ssh_strict": False,
+        "alt_host_keys": True,
+        "alt_key_file": "/x",
+        "system_host_keys": True,
+        "conn_timeout": 9,
+    }
+
+    def test_first_use_lets_netmiko_accept_the_key(self):
+        with HostKeyCheck({}, "10.0.0.5", 22).netmiko(self.BASE) as kwargs:
+            assert kwargs == {
+                "ssh_strict": False,
+                "alt_host_keys": False,
+                "system_host_keys": False,
+                "conn_timeout": 9,
+            }
+
+    def test_a_known_key_is_the_only_one_netmiko_takes(self):
+        check = HostKeyCheck({KNOWN_KEYS_ARG: {"2222": ED25519.as_dict()}}, "10.0.0.5", 2222)
+        with check.netmiko(self.BASE) as kwargs:
+            path = kwargs["alt_key_file"]
+            assert kwargs["ssh_strict"] is True
+            assert kwargs["alt_host_keys"] is True
+            assert kwargs["system_host_keys"] is False
+            assert kwargs["conn_timeout"] == 9
+            with open(path) as f:
+                assert f.read().strip() == ED25519.known_hosts_line("10.0.0.5", 2222)
+        assert not os.path.exists(path)
+
+    def test_a_refusal_inside_netmiko_becomes_the_mismatch(self, host_key):
+        # netmiko re-raises paramiko's BadHostKeyException as its own exception,
+        # inside the except block, and a driver wraps that once more.
+        seen: list = []
+        check = _check(ED25519, seen)
+        with pytest.raises(SshHostKeyMismatch) as caught:
+            with check.netmiko({}):
+                try:
+                    try:
+                        other = paramiko.ECDSAKey.generate()
+                        raise paramiko.BadHostKeyException("10.0.0.5", host_key, other)
+                    except paramiko.SSHException as exc:
+                        raise RuntimeError(f"A paramiko SSHException occurred: {exc}")
+                except RuntimeError as exc:
+                    raise ConnectionException(str(exc)) from exc
+        assert caught.value.observation.seen == SshHostKey.of(host_key)
+        assert caught.value.observation.expected == ED25519
+        assert seen[0]["accepted"] is False
+
+    def test_any_other_failure_passes_unchanged(self):
+        with pytest.raises(ConnectionException, match="timed out"):
+            with HostKeyCheck({KNOWN_KEYS_ARG: {"22": ED25519.as_dict()}}, "10.0.0.5", 22).netmiko({}):
+                raise ConnectionException("timed out")
+
+    def test_observe_reports_the_key_netmiko_saw(self, host_key):
+        class _Transport:
+            def get_remote_server_key(self):
+                return host_key
+
+        seen: list = []
+        observation = _check(None, seen).observe(_Transport())
+        assert observation.accepted is True
+        assert seen[0]["seen"] == SshHostKey.of(host_key).as_dict()