From a6f9a17858354995085ab2f1f33fb0c47fc7242b Mon Sep 17 00:00:00 2001 From: Christian Manivong Date: Tue, 6 Oct 2026 00:20:07 +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 (apt) runs the shared APT_UPGRADABLE_COMMAND and parse_apt_upgradable: origin and security from apt's suites, and a ValueError instead of [] when apt failed or its output was cut short. - dnf/yum: check-update's exit status decides (0 none, 100 updates, anything else raises); security comes from `updateinfo list --security`, and is None when dnf cannot say. The repository column becomes the origin. - refresh_available_updates(): apt-get update, dnf/yum makecache, apk update; pacman is left out (-Sy without -u invites a partial upgrade). - HostStatusMixin: reboot required and self-patching, read over SSH. - _run_privileged(): root runs directly, a sudo password goes through _sudo, otherwise sudo -n. Shared by service control and the refresh. - _split_status() drops terminal codes before it looks for the exit status; a pseudo-terminal left keypad codes in front of the marker. OpenMediaVault inherits all of it. --- napalm_linux/linux.py | 118 +++++++++++++++++++++++++------------ pyproject.toml | 2 +- tests/test_linux.py | 133 ++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 214 insertions(+), 39 deletions(-) diff --git a/napalm_linux/linux.py b/napalm_linux/linux.py index d195318..c6248bf 100644 --- a/napalm_linux/linux.py +++ b/napalm_linux/linux.py @@ -32,11 +32,17 @@ from netmiko.exceptions import ( from napalm.base.exceptions import ConnectionException, ConnectionClosedException from napalm.base.netmiko_helpers import netmiko_args from napalm_device_types import ( + APT_UPGRADABLE_COMMAND, + DNF_SECURITY_COMMAND, FingerprintRule, + HostStatusMixin, KernelFactsMixin, OSDriver, SystemdServicesMixin, SystemdUnavailable, + parse_apt_upgradable, + parse_dnf_security, + strip_terminal_codes, ) from napalm_device_types.models import ( ApplyUpdatesResultDict, @@ -62,6 +68,33 @@ _PKG_MANAGERS = ["apt", "dnf", "yum", "apk", "pacman"] _RC_MARKER = "__NETORK_RC=" _RC_MARKER_RE = re.compile(rf"^{_RC_MARKER}(\d+)\s*$", re.MULTILINE) + +def _split_status(raw: str) -> tuple[str, int | None]: + """``(output, exit_status)`` of a command followed by ``echo {_RC_MARKER}$?``. + + The status is ``None`` when the marker never arrived (output cut short), so + a caller can tell "unknown" from "succeeded". + """ + raw = strip_terminal_codes(raw) + matches = list(_RC_MARKER_RE.finditer(raw)) + if not matches: + return raw, None + last = matches[-1] + return (raw[: last.start()] + raw[last.end():]).strip(), int(last.group(1)) + + +#: How each package manager refreshes its index. pacman is left out on purpose: +#: ``pacman -Sy`` without ``-u`` invites a partial upgrade on the next install. +_REFRESH = { + # No LC_ALL=C here: under sudo it is an environment variable sudoers may refuse + # to set. Only the exit status decides, so the language is merely what is shown. + "apt": "apt-get update -q 2>&1", + "dnf": "dnf makecache -q 2>&1", + "yum": "yum makecache -q 2>&1", + "apk": "apk update -q 2>&1", +} +_YUM_SECURITY_COMMAND = "LC_ALL=C yum updateinfo list security -q 2>/dev/null" + #: What to do when sudo wants a password netOrk does not have. _SUDO_PASSWORD_HINT = ( "sudo requires a password on this device but none is configured in netOrk. " @@ -151,7 +184,7 @@ def _short_image_id(raw: str) -> str: return raw.strip().removeprefix("sha256:")[:12] -class LinuxDriver(KernelFactsMixin, SystemdServicesMixin, OSDriver): +class LinuxDriver(KernelFactsMixin, SystemdServicesMixin, HostStatusMixin, OSDriver): """NAPALM driver for generic Linux systems. Connects via SSH (netmiko ``linux`` device type) and auto-detects the @@ -297,13 +330,9 @@ class LinuxDriver(KernelFactsMixin, SystemdServicesMixin, OSDriver): The status is ``None`` when the marker never arrived (output cut short), so a caller can tell "unknown" from "succeeded". """ - raw = self._sudo(f"{command}; echo {_RC_MARKER}$?", read_timeout=read_timeout) - matches = list(_RC_MARKER_RE.finditer(raw)) - if not matches: - return raw, None - last = matches[-1] - output = (raw[: last.start()] + raw[last.end():]).strip() - return output, int(last.group(1)) + return _split_status( + self._sudo(f"{command}; echo {_RC_MARKER}$?", read_timeout=read_timeout) + ) def _is_root(self) -> bool: """Whether the SSH user is root, asked once per session. @@ -321,12 +350,24 @@ class LinuxDriver(KernelFactsMixin, SystemdServicesMixin, OSDriver): Without a sudo password, ``sudo -n`` fails at once where a prompt would otherwise hang the session until the read timeout. """ - if not privileged or self._is_root(): + if not privileged: + return self._send(command, read_timeout=timeout) + return self._run_privileged(command, timeout) + + def _run_privileged(self, command: str, timeout: float = 100) -> str: + """Run *command* as root: directly for a root login, through ``_sudo`` + with a sudo password, and through ``sudo -n`` without one -- which fails at + once where a password prompt would hang the session until the timeout.""" + if self._is_root(): return self._send(command, read_timeout=timeout) if self._sudo_password: return self._sudo(command, read_timeout=timeout) return self._send(f"sudo -n {command}", read_timeout=timeout) + def _run_host_status_command(self, command: str) -> str: + """The transport for ``HostStatusMixin.get_host_status``: read-only, no sudo.""" + return self._send(command, read_timeout=60) + def _detect_pkg_manager(self) -> str | None: """Return the first package manager binary found on PATH.""" for pm in _PKG_MANAGERS: @@ -1243,48 +1284,49 @@ class LinuxDriver(KernelFactsMixin, SystemdServicesMixin, OSDriver): def _get_updates_apt(self) -> list[UpdateDict]: # apt list --upgradable does not need root; avoid sudo so it works even # without a configured sudo password. - out = self._send( - "LC_ALL=C apt list --upgradable 2>/dev/null | grep -v '^Listing'", - read_timeout=60, - ) - # Join wrapped lines: netmiko's 80-col pseudo-TTY causes long apt lines to - # break; continuation lines start with a space. - raw_lines: List[str] = [] - for line in out.splitlines(): - if line.startswith(" ") and raw_lines: - raw_lines[-1] += line.strip() - else: - raw_lines.append(line) - updates: list[UpdateDict] = [] - for line in raw_lines: - # openssh-server/stable 1:9.2p1-2+deb12u2 amd64 [upgradable from: 1:9.2p1-2+deb12u1] - m = re.match( - r"^(\S+)/\S+\s+(\S+)\s+\S+\s+\[upgradable from:\s+(\S+)\]", line - ) - if m: - updates.append({ - "name": m.group(1), - "current_version": m.group(3), - "new_version": m.group(2), - }) - return updates + # Raises ValueError when apt failed or the output was cut short. + return parse_apt_upgradable(self._send(APT_UPGRADABLE_COMMAND, read_timeout=60)) def _get_updates_rpm(self) -> list[UpdateDict]: + """dnf/yum check-update: exit 100 means updates, 0 none, anything else failed.""" cmd = "dnf check-update --quiet 2>/dev/null" if self._pkg_manager == "dnf" else "yum check-update -q 2>/dev/null" - out = self._sudo(cmd) + output, status = _split_status(self._run_privileged(f"{cmd}; echo {_RC_MARKER}$?", 120)) + if status not in (0, 100): + raise RuntimeError(f"{self._pkg_manager} check-update failed (exit {status}): {output[-200:]}") + security = self._rpm_security_names() updates: list[UpdateDict] = [] - for line in out.splitlines(): + for line in output.splitlines(): parts = line.split() if len(parts) >= 2 and not line.startswith(" ") and "." in parts[0]: - name_arch = parts[0] - name = name_arch.rsplit(".", 1)[0] if "." in name_arch else name_arch + name = parts[0].rsplit(".", 1)[0] updates.append({ "name": name, "current_version": "", "new_version": parts[1], + "origin": parts[2] if len(parts) >= 3 else None, + "security": None if security is None else name in security, }) return updates + def _rpm_security_names(self) -> set[str] | None: + """Packages a pending security advisory covers; None when dnf/yum cannot say.""" + cmd = DNF_SECURITY_COMMAND if self._pkg_manager == "dnf" else _YUM_SECURITY_COMMAND + output, status = _split_status(self._run_privileged(f"{cmd}; echo {_RC_MARKER}$?", 120)) + return parse_dnf_security(output) if status == 0 else None + + def refresh_available_updates(self) -> dict[str, Any]: + """Refresh the package index (apt-get update, dnf makecache, apk update).""" + cmd = _REFRESH.get(self._pkg_manager or "") + if cmd is None: + return { + "success": False, + "output": f"Refreshing the index is not supported for {self._pkg_manager!r}", + } + output, status = _split_status(self._run_privileged(f"{cmd}; echo {_RC_MARKER}$?", 180)) + if status != 0 and "password is required" in output: + output = f"{output}\n{_SUDO_PASSWORD_HINT}" + return {"success": status == 0, "output": output} + def _get_updates_apk(self) -> list[UpdateDict]: out = self._send("apk version -l '<' 2>/dev/null") updates: list[UpdateDict] = [] diff --git a/pyproject.toml b/pyproject.toml index ca07c17..962e007 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -37,7 +37,7 @@ classifiers = [ ] dependencies = [ "napalm>=4.0", - "napalm-device-types>=2.2.0", + "napalm-device-types>=2.3.0", "netmiko>=4.0.0", "paramiko>=5.0.0", # CVE-2026-44405 ] diff --git a/tests/test_linux.py b/tests/test_linux.py index c09ac00..d0b074e 100644 --- a/tests/test_linux.py +++ b/tests/test_linux.py @@ -226,6 +226,7 @@ APT_UPGRADABLE = ( "Listing... Done\n" "openssh-server/stable 1:9.2p1-2+deb12u2 amd64 [upgradable from: 1:9.2p1-2+deb12u1]\n" "curl/stable 7.88.1-10+deb12u6 amd64 [upgradable from: 7.88.1-10+deb12u5]\n" + "__APT_RC=0\n" ) @@ -1282,3 +1283,135 @@ class TestManageService: driver.manage_service("cron; reboot", "stop") assert driver._device.send_command.call_count == 0 + + +# --------------------------------------------------------------------------- +# Updates: origin and security, refresh, host status (netOrk MVP 5) +# --------------------------------------------------------------------------- + +APT_WITH_SECURITY = ( + "openssl/noble-updates,noble-security 3.0.13-0ubuntu3.6 amd64 [upgradable from: 3.0.13-0ubuntu3.5]\n" + "docker-compose-plugin/noble 5.6.0-1~ubuntu.24.04~noble amd64 [upgradable from: 5.5.1-1~ubuntu.24.04~noble]\n" + "__APT_RC=0\n" +) + + +class TestAvailableUpdates: + def test_apt_reports_origin_and_security(self, driver): + driver._pkg_manager = "apt" + _mock_send(driver, APT_WITH_SECURITY) + + updates = {u["name"]: u for u in driver.get_available_updates()} + + assert updates["openssl"]["security"] is True + assert updates["openssl"]["origin"] == "noble-updates,noble-security" + assert updates["docker-compose-plugin"]["security"] is False + + def test_apt_that_could_not_read_raises_instead_of_reporting_nothing(self, driver): + driver._pkg_manager = "apt" + _mock_send(driver, "E: Could not open lock file\n__APT_RC=100\n") + + with pytest.raises(ValueError): + driver.get_available_updates() + + def test_apt_without_an_exit_status_raises(self, driver): + driver._pkg_manager = "apt" + _mock_send(driver, "openssl/noble-security 3.0.13-0ubuntu3.6 amd64 [upgradable fro") + + with pytest.raises(ValueError): + driver.get_available_updates() + + def test_dnf_marks_what_a_security_advisory_covers(self, driver): + driver._pkg_manager = "dnf" + driver._device.send_command.side_effect = [ + "0", # id -u + "openssl-libs.x86_64 1:3.1.4-2.fc40 updates\n" + "vim-enhanced.x86_64 2:9.1.083-1.fc40 updates\n__NETORK_RC=100", + "FEDORA-2024-1a2b3c4d5e Important/Sec. openssl-libs-1:3.1.4-2.fc40.x86_64\n__NETORK_RC=0", + ] + + updates = {u["name"]: u for u in driver.get_available_updates()} + + assert updates["openssl-libs"]["security"] is True + assert updates["vim-enhanced"]["security"] is False + + def test_dnf_without_advisories_leaves_security_unknown(self, driver): + driver._pkg_manager = "dnf" + driver._device.send_command.side_effect = [ + "0", + "vim-enhanced.x86_64 2:9.1.083-1.fc40 updates\n__NETORK_RC=100", + "Error: updateinfo metadata missing\n__NETORK_RC=1", + ] + + assert driver.get_available_updates()[0]["security"] is None + + def test_dnf_that_failed_raises(self, driver): + driver._pkg_manager = "dnf" + driver._device.send_command.side_effect = ["0", "Error: Failed to download metadata\n__NETORK_RC=1"] + + with pytest.raises(RuntimeError): + driver.get_available_updates() + + +class TestRefreshAvailableUpdates: + def _sent(self, driver) -> list: + return [c[0][0] for c in driver._device.send_command.call_args_list] + + def test_apt_refreshes_its_index_as_root(self, driver): + driver._pkg_manager = "apt" + driver._device.send_command.side_effect = ["0", "Hit:1 http://archive.ubuntu.com noble InRelease\n__NETORK_RC=0"] + + result = driver.refresh_available_updates() + + assert result["success"] is True + assert "apt-get update" in self._sent(driver)[1] + + def test_without_a_sudo_password_it_never_waits_for_one(self, driver): + driver._pkg_manager = "apt" + driver._device.send_command.side_effect = ["1000", "sudo: a password is required\n__NETORK_RC=1"] + + result = driver.refresh_available_updates() + + assert result["success"] is False + assert self._sent(driver)[1].startswith("sudo -n apt-get update") + + def test_dnf_refreshes_its_metadata(self, driver): + driver._pkg_manager = "dnf" + driver._device.send_command.side_effect = ["0", "Metadata cache created.\n__NETORK_RC=0"] + + assert driver.refresh_available_updates()["success"] is True + assert "dnf makecache" in self._sent(driver)[1] + + def test_pacman_is_not_refreshed_on_its_own(self, driver): + """pacman -Sy without -u invites a partial upgrade on the next install.""" + driver._pkg_manager = "pacman" + + result = driver.refresh_available_updates() + + assert result["success"] is False + driver._device.send_command.assert_not_called() + + +class TestHostStatus: + def test_the_driver_carries_the_shared_command(self, driver): + from napalm_device_types.host_status import HOST_STATUS_COMMAND + + _mock_send( + driver, + "HSTAT_BEGIN\n[reboot-required]\n[kernel]\n6.8.0-142-generic\n[modules]\n" + "6.8.0-142-generic\n[timers]\napt-daily-upgrade.timer enabled\nHSTAT_END\n", + ) + + status = driver.get_host_status() + + assert driver._device.send_command.call_args[0][0] == HOST_STATUS_COMMAND + assert status["reboot_required"] is True + + +class TestTerminalCodes: + def test_a_status_marker_behind_a_terminal_code_is_still_read(self, driver): + """apt-get on a pseudo-terminal leaves keypad codes in front of the marker.""" + driver._pkg_manager = "apt" + driver._device.send_command.side_effect = ["0", "Hit:1 noble InRelease\n\x1b>__NETORK_RC=0"] + + assert driver.refresh_available_updates()["success"] is True -- 2.54.0