diff --git a/CHANGELOG.md b/CHANGELOG.md index 45e2192..905284b 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 node's SSH session (the fallback for node commands and the provisioning steps) verifies its host key (NetOrk/netork#1167): the key the caller expects for port 22 in `optional_args["ssh_host_keys"]`, else the first one; every key seen goes to `optional_args["ssh_host_key_observer"]`. A changed key is refused before any login: a provisioning step fails, a best-effort read returns nothing. Both paths now share one session (`_node_ssh`). Replaces paramiko's `AutoAddPolicy`. Requires napalm-device-types 4.19.0; version 0.5.0. + ### Changed - **Breaking:** `get_device_warnings` is gone; the node reports facts and the caller names the warnings (netOrk #1111). lldpd and the pending updates were already facts diff --git a/napalm_proxmox/driver.py b/napalm_proxmox/driver.py index 1b2f0fa..684f706 100644 --- a/napalm_proxmox/driver.py +++ b/napalm_proxmox/driver.py @@ -48,6 +48,7 @@ from napalm_device_types import ( SystemdServicesMixin, ) from napalm.base.exceptions import ConnectionException +from napalm_device_types.ssh_host_keys import HostKeyCheck, SshHostKeyMismatch try: from proxmoxer import ProxmoxAPI @@ -318,39 +319,49 @@ class ProxmoxDriver( # Fallback: direct paramiko SSH try: - import paramiko - - if self._ssh_client is None: - ssh_user = self._ssh_username or self.username - ssh_pass = self._ssh_password or self.password - ssh_pkey = None - if self._ssh_key and not ssh_pass: - from io import StringIO as _StringIO - ssh_pkey = paramiko.RSAKey.from_private_key(_StringIO(self._ssh_key)) - self._ssh_client = paramiko.SSHClient() - self._ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) - connect_kwargs: _JsonDict = { - "hostname": self.hostname, - "port": 22, - "username": ssh_user, - "timeout": self.timeout, - } - if ssh_pkey: - connect_kwargs["pkey"] = ssh_pkey - else: - connect_kwargs["password"] = ssh_pass - self._ssh_client.connect(**connect_kwargs) - _, stdout, stderr = self._ssh_client.exec_command(command, timeout=self.timeout) + _, stdout, stderr = self._node_ssh().exec_command(command, timeout=self.timeout) err = stderr.read().decode().strip() out = stdout.read().decode().strip() return out or err except ImportError: logger.warning("paramiko not installed — cannot exec SSH commands") + except SshHostKeyMismatch as exc: + # Refused before any login, and already reported to the caller's store. + logger.warning("%s", exc) except Exception as exc: logger.debug("SSH exec command failed: %s", exc) return "" + def _node_ssh(self) -> Any: # paramiko.SSHClient; paramiko is imported lazily + """The SSH session to the node on port 22, opened once. + + Its host key is the one the caller expects for port 22, or on first use + whichever the node presents; either way it is reported (netork#1167). + """ + import paramiko + + if self._ssh_client is None: + ssh_pass = self._ssh_password or self.password + connect_kwargs: _JsonDict = { + "username": self._ssh_username or self.username, + "timeout": self.timeout, + } + if self._ssh_key and not ssh_pass: + from io import StringIO as _StringIO + + connect_kwargs["pkey"] = paramiko.RSAKey.from_private_key(_StringIO(self._ssh_key)) + else: + connect_kwargs["password"] = ssh_pass + client = paramiko.SSHClient() + try: + HostKeyCheck(self.optional_args, self.hostname, 22).connect(client, **connect_kwargs) + except Exception: + client.close() + raise + self._ssh_client = client + return self._ssh_client + # ------------------------------------------------------------------ # # Node info helpers # ------------------------------------------------------------------ # diff --git a/napalm_proxmox/vm_provision_mixin.py b/napalm_proxmox/vm_provision_mixin.py index 695cf2a..e29024c 100644 --- a/napalm_proxmox/vm_provision_mixin.py +++ b/napalm_proxmox/vm_provision_mixin.py @@ -143,31 +143,7 @@ class ProxmoxVMProvisionMixin: disk import — where a non-zero exit or a caller-specific timeout must surface as a hard failure rather than an empty string. """ - import paramiko - - if self._ssh_client is None: - ssh_user = self._ssh_username or self.username - ssh_pass = self._ssh_password or self.password - ssh_pkey = None - if self._ssh_key and not ssh_pass: - from io import StringIO as _StringIO - - ssh_pkey = paramiko.RSAKey.from_private_key(_StringIO(self._ssh_key)) - self._ssh_client = paramiko.SSHClient() - self._ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) - connect_kwargs: Dict[str, Any] = { - "hostname": self.hostname, - "port": 22, - "username": ssh_user, - "timeout": self.timeout, - } - if ssh_pkey: - connect_kwargs["pkey"] = ssh_pkey - else: - connect_kwargs["password"] = ssh_pass - self._ssh_client.connect(**connect_kwargs) - - _, stdout, stderr = self._ssh_client.exec_command(command, timeout=timeout) + _, stdout, stderr = self._node_ssh().exec_command(command, timeout=timeout) exit_status = stdout.channel.recv_exit_status() out = stdout.read().decode().strip() err = stderr.read().decode().strip() diff --git a/pyproject.toml b/pyproject.toml index 000e64e..97b806e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "napalm-proxmox" -version = "0.4.0" +version = "0.5.0" description = "NAPALM driver for Proxmox VE (classic Linux networking, SDN, OVS)" readme = "README.md" license = { text = "Apache-2.0" } @@ -25,7 +25,7 @@ classifiers = [ requires-python = ">=3.9" dependencies = [ "napalm>=5.0.0", - "napalm_device_types>=4.14.0", + "napalm_device_types>=4.19.0", "paramiko>=5.0.0", # CVE-2026-44405; imported directly for SSH fallback (driver.py) "proxmoxer>=2.0.0", "netaddr>=0.9.0", diff --git a/tests/test_ssh_host_key.py b/tests/test_ssh_host_key.py new file mode 100644 index 0000000..e977084 --- /dev/null +++ b/tests/test_ssh_host_key.py @@ -0,0 +1,133 @@ +"""The node's SSH session verifies its host key, and reports the one it sees (NetOrk/netork#1167). + +Both ways onto the node's shell — the best-effort ``_exec_ssh_command`` and the +provisioning steps' ``_run_node_command`` — share one session, opened on port 22. +""" + +from __future__ import annotations + +from unittest.mock import MagicMock + +import paramiko +import pytest +from napalm_device_types.ssh_host_keys import ( + KNOWN_KEYS_ARG, + OBSERVER_ARG, + SshHostKey, + SshHostKeyMismatch, +) + +from napalm_proxmox.driver import ProxmoxDriver + +SERVER_KEY = paramiko.ECDSAKey.generate() +OTHER_KEY = paramiko.ECDSAKey.generate() + + +class _Stream: + def __init__(self, data: bytes) -> None: + self._data = data + self.channel = MagicMock() + self.channel.recv_exit_status.return_value = 0 + + def read(self) -> bytes: + return self._data + + +class _Client: + """paramiko.SSHClient as far as the driver uses it; refuses like paramiko would.""" + + def __init__(self) -> None: + self.lines: list[str] = [] + self.policy = None + self.connected_with: dict | None = None + self.commands: list[str] = [] + + def set_missing_host_key_policy(self, policy) -> None: + self.policy = policy + + def get_host_keys(self): + client = self + + class _Keys: + def load(self, path: str) -> None: + with open(path) as f: + client.lines += f.read().splitlines() + + return _Keys() + + def connect(self, hostname, **kwargs) -> None: + if self.lines and SshHostKey.of(SERVER_KEY).key not in self.lines[0]: + raise paramiko.BadHostKeyException(hostname, SERVER_KEY, OTHER_KEY) + if not self.lines: + self.policy.missing_host_key(self, hostname, SERVER_KEY) + self.connected_with = {"hostname": hostname, **kwargs} + + def get_transport(self): + transport = MagicMock() + transport.get_remote_server_key.return_value = SERVER_KEY + return transport + + def exec_command(self, command, timeout=None): + self.commands.append(command) + return None, _Stream(b"ok"), _Stream(b"") + + def close(self) -> None: + pass + + +@pytest.fixture +def client(monkeypatch) -> _Client: + fake = _Client() + monkeypatch.setattr(paramiko, "SSHClient", lambda: fake) + return fake + + +def _driver(seen: list, known: dict | None = None) -> ProxmoxDriver: + args: dict = {"port": 8006, OBSERVER_ARG: seen.append} + if known is not None: + args[KNOWN_KEYS_ARG] = known + driver = ProxmoxDriver("10.0.0.9", "root", "s3cret", optional_args=args) + driver._node_api = MagicMock(side_effect=RuntimeError("no exec endpoint")) # type: ignore[method-assign] + return driver + + +def test_first_use_runs_the_command_and_reports_the_key(client): + seen: list = [] + assert _driver(seen)._run_node_command("true", timeout=5) == "ok" + assert client.connected_with["port"] == 22 + assert seen[0]["accepted"] is True + assert (seen[0]["host"], seen[0]["port"]) == ("10.0.0.9", 22) + + +def test_the_key_for_the_api_port_is_not_the_ssh_one(client): + seen: list = [] + other = SshHostKey.of(OTHER_KEY).as_dict() + _driver(seen, {"8006": other})._run_node_command("true", timeout=5) + assert seen[0]["expected"] is None + + +def test_a_changed_key_stops_a_provisioning_step(client): + seen: list = [] + driver = _driver(seen, {"22": SshHostKey.of(OTHER_KEY).as_dict()}) + with pytest.raises(SshHostKeyMismatch): + driver._run_node_command("qm importdisk …", timeout=5) + assert client.commands == [] + assert client.connected_with is None + assert seen[0]["accepted"] is False + + +def test_a_changed_key_leaves_the_best_effort_read_empty(client): + seen: list = [] + driver = _driver(seen, {"22": SshHostKey.of(OTHER_KEY).as_dict()}) + assert driver._exec_ssh_command("cat /etc/os-release") == "" + assert client.commands == [] + assert seen[0]["accepted"] is False + + +def test_the_known_key_reuses_one_session(client): + seen: list = [] + driver = _driver(seen, {"22": SshHostKey.of(SERVER_KEY).as_dict()}) + driver._exec_ssh_command("uptime") + driver._run_node_command("true", timeout=5) + assert client.commands == ["uptime", "true"] + assert len(seen) == 1 and seen[0]["accepted"] is True diff --git a/tests/test_vm_provision_mixin.py b/tests/test_vm_provision_mixin.py index ed9ad13..b1b5f97 100644 --- a/tests/test_vm_provision_mixin.py +++ b/tests/test_vm_provision_mixin.py @@ -616,8 +616,9 @@ def _mock_ssh_exec(exit_status: int, stdout_text: str = "", stderr_text: str = " def test_run_node_command_success_returns_stdout(): mixin = ProxmoxVMProvisionMixin() - mixin._ssh_client = MagicMock() - mixin._ssh_client.exec_command.return_value = _mock_ssh_exec(0, stdout_text="hello\n") + ssh = MagicMock() + ssh.exec_command.return_value = _mock_ssh_exec(0, stdout_text="hello\n") + mixin._node_ssh = MagicMock(return_value=ssh) # the driver's verified session result = mixin._run_node_command("echo hello", timeout=10) @@ -626,10 +627,9 @@ def test_run_node_command_success_returns_stdout(): def test_run_node_command_nonzero_exit_raises(): mixin = ProxmoxVMProvisionMixin() - mixin._ssh_client = MagicMock() - mixin._ssh_client.exec_command.return_value = _mock_ssh_exec( - 1, stderr_text="No such file or directory" - ) + ssh = MagicMock() + ssh.exec_command.return_value = _mock_ssh_exec(1, stderr_text="No such file or directory") + mixin._node_ssh = MagicMock(return_value=ssh) with pytest.raises(RuntimeError, match="No such file or directory"): mixin._run_node_command("cat /nonexistent", timeout=10)