diff --git a/CHANGELOG.md b/CHANGELOG.md index 63c7904..3a07fa6 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()` hands netmiko the key the caller expects for the port in `optional_args["ssh_host_keys"]` (as a `known_hosts` file that exists only while it connects, with `ssh_strict`), so any other key is refused before the login (`SshHostKeyMismatch`); without one netmiko accepts the first key. Every key seen goes to `optional_args["ssh_host_key_observer"]`. OpenMediaVault and QNAP get this through `LinuxDriver.open()`. Requires napalm-device-types 4.19.0; version 0.11.0. + ### Changed - **Breaking:** `get_device_warnings` is gone, and with it the `apt_proxy_url` argument. The driver reports facts and the application names warnings (napalm-device-types 4.14): diff --git a/napalm_linux/linux.py b/napalm_linux/linux.py index 015b8b4..d18fb45 100644 --- a/napalm_linux/linux.py +++ b/napalm_linux/linux.py @@ -59,6 +59,7 @@ from napalm_device_types import ( run_on_transport, strip_terminal_codes, ) +from napalm_device_types.ssh_host_keys import HostKeyCheck from napalm_device_types.models import ( ApplyUpdatesResultDict, CronJobDict, @@ -356,6 +357,8 @@ class LinuxDriver( # port is passed explicitly in open() — remove it from netmiko_optional_args # to avoid "multiple values for keyword argument 'port'" self.netmiko_optional_args.pop("port", None) + # The host key to expect, and where the one seen is reported (netork#1167). + self._host_key = HostKeyCheck(optional_args, hostname, self.port) # Runtime state self._device: ConnectHandler | None = None @@ -368,20 +371,22 @@ class LinuxDriver( def open(self) -> None: """Open the SSH connection and detect the package manager.""" try: - self._device = ConnectHandler( - device_type=self.NETMIKO_DEVICE_TYPE, - host=self.hostname, - username=self.username, - password=self.password, - port=self.port, - secret=self._secret, - timeout=self.timeout, - **self.netmiko_optional_args, - ) + with self._host_key.netmiko(self.netmiko_optional_args) as netmiko_args_: + self._device = ConnectHandler( + device_type=self.NETMIKO_DEVICE_TYPE, + host=self.hostname, + username=self.username, + password=self.password, + port=self.port, + secret=self._secret, + timeout=self.timeout, + **netmiko_args_, + ) except NetmikoAuthenticationException as exc: raise ConnectionException(str(exc)) from exc except NetmikoTimeoutException as exc: raise ConnectionException(str(exc)) from exc + self._host_key.observe(self._device.remote_conn_pre.get_transport()) # No history before anything else is typed (NetOrk/netork#1039). try: diff --git a/pyproject.toml b/pyproject.toml index 49e9adf..952a2a2 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "napalm-linux" -version = "0.10.1" +version = "0.11.0" description = "NAPALM driver for generic Linux systems via SSH" readme = "README.md" requires-python = ">=3.9" @@ -37,7 +37,7 @@ classifiers = [ ] dependencies = [ "napalm>=4.0", - "napalm-device-types>=4.18.0", + "napalm-device-types>=4.19.0", "netmiko>=4.0.0", "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..bd1cdf0 --- /dev/null +++ b/tests/test_ssh_host_key.py @@ -0,0 +1,86 @@ +"""open() hands netmiko the host key to expect, and reports the one it saw (NetOrk/netork#1167). + +OpenMediaVault and QNAP open through ``LinuxDriver.open()`` and get this too. +""" + +from __future__ import annotations + +import os +from unittest.mock import MagicMock, patch + +import paramiko +import pytest +from napalm_device_types.ssh_host_keys import ( + KNOWN_KEYS_ARG, + OBSERVER_ARG, + SshHostKey, + SshHostKeyMismatch, +) +from netmiko.exceptions import NetmikoTimeoutException + +from napalm_linux.linux import LinuxDriver + +SERVER_KEY = paramiko.ECDSAKey.generate() +OTHER_KEY = paramiko.ECDSAKey.generate() + + +def _device() -> MagicMock: + device = MagicMock() + device.send_command.return_value = "/usr/bin/apt" + device.remote_conn_pre.get_transport.return_value.get_remote_server_key.return_value = SERVER_KEY + return device + + +def _driver(seen: list, known: SshHostKey | None = None) -> LinuxDriver: + args: dict = {"port": 2222, OBSERVER_ARG: seen.append} + if known is not None: + args[KNOWN_KEYS_ARG] = {"2222": known.as_dict()} + return LinuxDriver("10.0.0.8", "admin", "s3cret", optional_args=args) + + +def test_first_use_lets_netmiko_accept_and_reports_the_key(): + seen: list = [] + with patch("napalm_linux.linux.ConnectHandler", return_value=_device()) as handler: + _driver(seen).open() + kwargs = handler.call_args.kwargs + assert kwargs["ssh_strict"] is False + assert kwargs["alt_host_keys"] is False + assert kwargs["system_host_keys"] is False + 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.8", 2222) + + +def test_a_known_key_is_the_only_one_netmiko_accepts(): + seen: list = [] + known = SshHostKey.of(SERVER_KEY) + files: list[str] = [] + + def connect(**kwargs): + files.append(kwargs["alt_key_file"]) + with open(kwargs["alt_key_file"]) as f: + assert f.read().strip() == known.known_hosts_line("10.0.0.8", 2222) + assert kwargs["ssh_strict"] is True and kwargs["alt_host_keys"] is True + return _device() + + with patch("napalm_linux.linux.ConnectHandler", side_effect=connect): + _driver(seen, known).open() + assert not os.path.exists(files[0]) + assert seen[0]["accepted"] is True + + +def test_a_changed_key_is_refused(): + seen: list = [] + + def refuse(**kwargs): + try: + raise paramiko.BadHostKeyException("10.0.0.8", SERVER_KEY, OTHER_KEY) + except paramiko.SSHException as exc: # what netmiko does with it + raise NetmikoTimeoutException(f"A paramiko SSHException occurred: {exc}") + + driver = _driver(seen, SshHostKey.of(OTHER_KEY)) + with patch("napalm_linux.linux.ConnectHandler", side_effect=refuse): + with pytest.raises(SshHostKeyMismatch): + driver.open() + assert driver._device is None + assert seen[0]["accepted"] is False