From 02a441ff09b5aa788ffa18c9fc08683088760f40 Mon Sep 17 00:00:00 2001 From: Christian Manivong Date: Tue, 6 Oct 2026 00:20:08 +0200 Subject: [PATCH] feat: report where an update comes from and whether it is a security fix, refresh the index, read the host status For netOrk MVP 5, on napalm-device-types 2.3.0: - get_available_updates reads `apt list --upgradable` over the exec path (APT_UPGRADABLE_COMMAND), so each update carries its suite and security status; the APT API, which names only "Debian"/"Proxmox", is the fallback with security unknown. It raises when neither answers instead of returning [] -- it used to swallow every error. - refresh_available_updates(): POST nodes/{n}/apt/update. - HostStatusMixin over the exec path (reboot required, self-patching). --- napalm_proxmox/driver.py | 2 + napalm_proxmox/system_mixin.py | 53 +++++++++++---- pyproject.toml | 2 +- tests/test_updates.py | 120 +++++++++++++++++++++++++++++++++ 4 files changed, 163 insertions(+), 14 deletions(-) create mode 100644 tests/test_updates.py diff --git a/napalm_proxmox/driver.py b/napalm_proxmox/driver.py index 11eb40e..051762d 100644 --- a/napalm_proxmox/driver.py +++ b/napalm_proxmox/driver.py @@ -36,6 +36,7 @@ logger = logging.getLogger(__name__) from napalm_device_types import ( FingerprintRule, + HostStatusMixin, HypervisorDriver, KernelFactsMixin, PortSpec, @@ -84,6 +85,7 @@ class ProxmoxDriver( ProxmoxSystemMixin, KernelFactsMixin, SystemdServicesMixin, + HostStatusMixin, HypervisorDriver, ): """NAPALM driver for Proxmox VE nodes.""" diff --git a/napalm_proxmox/system_mixin.py b/napalm_proxmox/system_mixin.py index 26f6d77..9f6317e 100644 --- a/napalm_proxmox/system_mixin.py +++ b/napalm_proxmox/system_mixin.py @@ -20,6 +20,8 @@ import logging import re from typing import Any +from napalm_device_types import APT_UPGRADABLE_COMMAND, parse_apt_upgradable + from napalm_proxmox import utils logger = logging.getLogger(__name__) @@ -362,22 +364,47 @@ class ProxmoxSystemMixin: # ------------------------------------------------------------------ # def get_available_updates(self) -> list[_JsonDict]: - """Return list of upgradable packages from the Proxmox APT API.""" - updates: list[_JsonDict] = [] + """Return the node's upgradable packages, with origin and security status. + + ``apt list --upgradable`` over the exec path names each candidate's suite + (``trixie-security``); the APT API names only an Origin ("Debian", + "Proxmox") and is the fallback, with the security status unknown. + + :raises Exception: when neither answers -- never an empty list for + "could not read". + """ try: - for upd in self._api.nodes(self._node_name).apt.update.get(): - pkg = upd.get("Package", "") - if not pkg: - continue - updates.append({ - "name": pkg, - "current_version": upd.get("OldVersion", ""), - "new_version": upd.get("Version", ""), - }) - except Exception as exc: - logger.debug("Failed to fetch available updates: %s", exc) + updates = parse_apt_upgradable(self._exec_ssh_command(APT_UPGRADABLE_COMMAND) or "") + except ValueError as exc: + logger.debug("apt list over the exec path failed, using the API: %s", exc) + updates = self._updates_from_api() return sorted(updates, key=lambda u: u["name"]) + def _updates_from_api(self) -> list[_JsonDict]: + return [ + { + "name": upd["Package"], + "current_version": upd.get("OldVersion", ""), + "new_version": upd.get("Version", ""), + "origin": upd.get("Origin"), + "security": None, + } + for upd in self._api.nodes(self._node_name).apt.update.get() # type: ignore[union-attr] + if upd.get("Package") + ] + + def refresh_available_updates(self) -> _JsonDict: + """Resynchronise the node's package index (``POST nodes/{n}/apt/update``).""" + try: + task = self._api.nodes(self._node_name).apt.update.post() # type: ignore[union-attr] + except Exception as exc: + return {"success": False, "output": str(exc)} + return {"success": True, "output": f"Package index refresh started ({task})"} + + def _run_host_status_command(self, command: str) -> str: + """The transport for ``HostStatusMixin.get_host_status``: the exec path.""" + return str(self._exec_ssh_command(command)) + def apply_updates(self, packages: list[str]) -> _JsonDict: """Upgrade the given packages via ``apt-get install`` over SSH.""" for pkg in packages: diff --git a/pyproject.toml b/pyproject.toml index 9bc0199..d1f99bf 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -25,7 +25,7 @@ classifiers = [ requires-python = ">=3.9" dependencies = [ "napalm>=5.0.0", - "napalm_device_types>=2.2.0", + "napalm_device_types>=2.3.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_updates.py b/tests/test_updates.py new file mode 100644 index 0000000..1ca67d5 --- /dev/null +++ b/tests/test_updates.py @@ -0,0 +1,120 @@ +"""Pending updates on a Proxmox node: from where, whether they are security fixes, +and whether the node needs a reboot. + +The node's APT API names only an Origin ("Debian", "Proxmox"), the same for the +main and the security archive. ``apt list --upgradable`` over the exec path +names the suite (``trixie-security``), so that is read first; the API remains +the fallback, with the security status left unknown. A reader that cannot read +raises: an empty list would tell netOrk that nothing is pending. +""" + +from __future__ import annotations + +from unittest.mock import MagicMock, patch + +import pytest +from napalm_device_types import HostStatusMixin +from napalm_device_types.host_status import HOST_STATUS_COMMAND +from napalm_device_types.package_updates import APT_UPGRADABLE_COMMAND + +from napalm_proxmox.driver import ProxmoxDriver + +APT = ( + "libssl3t64/stable-security 3.5.1-1+deb13u2 amd64 [upgradable from: 3.5.1-1+deb13u1]\n" + "ceph-common/stable 20.2.4-pve5 amd64 [upgradable from: 20.2.4-pve4]\n" +) +API_ENTRY = { + "Package": "librados2", + "OldVersion": "20.2.4-pve4", + "Version": "20.2.4-pve5", + "Origin": "Proxmox", +} + + +def _api_updates(driver, entries=None, error=None): + api = MagicMock() + getter = api.nodes.return_value.apt.update.get + if error: + getter.side_effect = error + else: + getter.return_value = entries or [] + driver._api = api + return api + + +class TestAvailableUpdates: + def test_apt_over_the_exec_path_names_the_suite(self, driver): + with patch.object(driver, "_exec_ssh_command", return_value=APT + "__APT_RC=0\n") as exec_: + updates = {u["name"]: u for u in driver.get_available_updates()} + + exec_.assert_called_once_with(APT_UPGRADABLE_COMMAND) + assert updates["libssl3t64"]["security"] is True + assert updates["ceph-common"]["security"] is False + assert updates["ceph-common"]["origin"] == "stable" + + def test_the_api_is_the_fallback_with_security_unknown(self, driver): + _api_updates(driver, [API_ENTRY]) + with patch.object(driver, "_exec_ssh_command", return_value=""): + updates = driver.get_available_updates() + + assert updates == [ + { + "name": "librados2", + "current_version": "20.2.4-pve4", + "new_version": "20.2.4-pve5", + "origin": "Proxmox", + "security": None, + } + ] + + def test_a_failed_apt_falls_back_too(self, driver): + _api_updates(driver, [API_ENTRY]) + with patch.object(driver, "_exec_ssh_command", return_value="E: lock\n__APT_RC=100\n"): + assert [u["name"] for u in driver.get_available_updates()] == ["librados2"] + + def test_nothing_readable_raises_instead_of_reporting_nothing(self, driver): + _api_updates(driver, error=RuntimeError("API timeout")) + with patch.object(driver, "_exec_ssh_command", return_value=""): + with pytest.raises(RuntimeError): + driver.get_available_updates() + + def test_nothing_pending_is_an_empty_list(self, driver): + with patch.object(driver, "_exec_ssh_command", return_value="__APT_RC=0\n"): + assert driver.get_available_updates() == [] + + +class TestRefresh: + def test_the_node_refreshes_its_index_through_the_api(self, driver): + api = _api_updates(driver) + api.nodes.return_value.apt.update.post.return_value = "UPID:pve1:0001" + + result = driver.refresh_available_updates() + + assert result["success"] is True + api.nodes.return_value.apt.update.post.assert_called_once() + + def test_a_refused_refresh_says_why(self, driver): + api = _api_updates(driver) + api.nodes.return_value.apt.update.post.side_effect = RuntimeError( + "403 Permission check failed" + ) + + result = driver.refresh_available_updates() + + assert result == {"success": False, "output": "403 Permission check failed"} + + +class TestHostStatus: + def test_the_node_is_read_over_the_exec_path(self, driver): + report = ( + "HSTAT_BEGIN\n[kernel]\n7.0.14-19-pve\n[modules]\n7.0.14-19-pve\n7.0.2-6-pve\n" + "[timers]\napt-daily-upgrade.timer enabled\nHSTAT_END\n" + ) + with patch.object(driver, "_exec_ssh_command", return_value=report) as exec_: + status = driver.get_host_status() + + exec_.assert_called_once_with(HOST_STATUS_COMMAND) + assert status["reboot_required"] is False + + def test_the_driver_declares_the_contract(self): + assert issubclass(ProxmoxDriver, HostStatusMixin) -- 2.54.0