diff --git a/CHANGELOG.md b/CHANGELOG.md index a46fa2b..185e73a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Security +- The SSH host key is verified (NetOrk/netork#1167). `open()` takes the key to expect from `optional_args["ssh_host_key"]` and refuses any other one before it logs in (`SshHostKeyMismatch`); without one it accepts the first key. Every key it sees goes to `optional_args["ssh_host_key_observer"]`. Replaces paramiko's `AutoAddPolicy`. Requires napalm-device-types 4.19.0; version 0.4.0. + ## [0.3.0] - 2026-10-17 ### Changed diff --git a/napalm_bsd/base.py b/napalm_bsd/base.py index e0f4bd6..15ee69c 100644 --- a/napalm_bsd/base.py +++ b/napalm_bsd/base.py @@ -20,6 +20,7 @@ from napalm.base.exceptions import ConnectionClosedException, ConnectionExceptio from napalm_device_types import EnsurePackageMixin, OSDriver, SnmpAgentMixin from napalm_device_types.models import FixStepOutcomeDict, SnmpAgentSettingsDict from napalm_device_types.host_status import HostStatusMixin +from napalm_device_types.ssh_host_keys import HostKeyCheck, SshHostKeyMismatch from napalm_device_types.channel import ( ByteStream, CommandResult, @@ -124,6 +125,8 @@ class BsdDriver(SnmpAgentMixin, EnsurePackageMixin, HostStatusMixin, OSDriver): self._sudo_password: Optional[str] = args.get("sudo_password") self._allow_agent: bool = bool(args.get("allow_agent", False)) self._look_for_keys: bool = bool(args.get("look_for_keys", False)) + # The host key to expect, and where the one seen is reported (netork#1167). + self._host_key = HostKeyCheck(args, hostname, self.port) self._client: Optional[paramiko.SSHClient] = None self._root: Optional[bool] = None @@ -131,11 +134,9 @@ class BsdDriver(SnmpAgentMixin, EnsurePackageMixin, HostStatusMixin, OSDriver): def open(self) -> None: client = paramiko.SSHClient() - client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) try: - client.connect( - self.hostname, - port=self.port, + self._host_key.connect( + client, username=self.username, password=self.password or None, key_filename=self._key_file, @@ -143,6 +144,9 @@ class BsdDriver(SnmpAgentMixin, EnsurePackageMixin, HostStatusMixin, OSDriver): look_for_keys=self._look_for_keys, timeout=self.timeout, ) + except SshHostKeyMismatch: + client.close() # refused before any login; nothing was sent + raise except (paramiko.SSHException, socket.error) as exc: client.close() raise ConnectionException(f"SSH to {self.hostname} failed: {exc}") from exc diff --git a/pyproject.toml b/pyproject.toml index 9a9f3d5..75f71aa 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "napalm-bsd" -version = "0.3.0" +version = "0.4.0" description = "NAPALM drivers for FreeBSD and OpenBSD hosts via SSH" readme = "README.md" requires-python = ">=3.10" @@ -26,7 +26,7 @@ classifiers = [ ] dependencies = [ "napalm>=4.0", - "napalm-device-types>=4.13.0", # fixes: SnmpAgentMixin, EnsurePackageMixin + "napalm-device-types>=4.19.0", # fixes: SnmpAgentMixin, EnsurePackageMixin "paramiko>=5.0.0", # CVE-2026-44405 ] diff --git a/tests/test_ssh_host_key.py b/tests/test_ssh_host_key.py new file mode 100644 index 0000000..494e709 --- /dev/null +++ b/tests/test_ssh_host_key.py @@ -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