From 18676a573fb1112096d891c2b81f7a850944ba92 Mon Sep 17 00:00:00 2001 From: Christian Manivong Date: Tue, 6 Oct 2026 00:19:24 +0200 Subject: [PATCH] feat: say where a pending update comes from, whether it is a security fix, and whether the host needs a reboot netOrk MVP 5 measures "security updates applied within N days" and starts patch runs inside agreed windows. That needs three things every Linux driver reads the same way, so they live here once: - UpdateDict gains optional `origin` and `security` (None = unknown). - package_updates: APT_UPGRADABLE_COMMAND and parse_apt_upgradable(). apt's suites are the origin; a "-security" suite makes it a security update; several architectures of one package are one entry. The command runs through a pipe with its exit status printed inside the group: through a pseudo-terminal apt drew progress and keypad codes, one of which (ESC >) a screen-scraping read took for a prompt and stopped at. The parser raises ValueError when apt failed or its status never arrived. DNF_SECURITY_COMMAND / parse_dnf_security() / nevra_name() for dnf/yum. - host_status: HOST_STATUS_COMMAND, parse_host_status(), HostStatusMixin (template form, hook _run_host_status_command). reboot_required from /var/run/reboot-required, needs-restarting -r, or a newer kernel of the running flavour (a Raspberry Pi carries two flavours side by side); auto_updates from APT::Periodic::Unattended-Upgrade with its timer, or dnf-automatic. HostStatusDict in models.py. - terminal.strip_terminal_codes(): CSI, OSC and two-character escapes, now also used by the systemd parser. - UpdateMixin: contract refresh_available_updates(); get_available_updates' docstring now states the rule that a reader raises when it cannot read and never returns [] for "don't know". Fixtures are real output from Ubuntu 24.04, Debian 13 / OMV, Raspberry Pi OS and Proxmox VE 9. Version 2.3.0. --- README.md | 11 ++ napalm_device_types/__init__.py | 17 +++ napalm_device_types/host_status.py | 177 ++++++++++++++++++++++ napalm_device_types/models.py | 21 ++- napalm_device_types/package_updates.py | 111 ++++++++++++++ napalm_device_types/systemd.py | 8 +- napalm_device_types/terminal.py | 23 +++ napalm_device_types/updates.py | 20 ++- pyproject.toml | 2 +- tests/test_host_status.py | 200 +++++++++++++++++++++++++ tests/test_package_updates.py | 149 ++++++++++++++++++ 11 files changed, 731 insertions(+), 8 deletions(-) create mode 100644 napalm_device_types/host_status.py create mode 100644 napalm_device_types/package_updates.py create mode 100644 napalm_device_types/terminal.py create mode 100644 tests/test_host_status.py create mode 100644 tests/test_package_updates.py diff --git a/README.md b/README.md index 09949af..6af00f5 100644 --- a/README.md +++ b/README.md @@ -79,6 +79,17 @@ everywhere, so the command and its parse are concrete here and a driver supplies `_run_kernel_facts_command`. `OSDriver` does not carry it — a Windows host is an OS driver too, and `hasattr(driver, "get_kernel_facts")` has to stay truthful. +`HostStatusMixin` (`get_host_status`) is mixed in the same way: whether a Linux host needs +a reboot to finish an update (`/var/run/reboot-required`, `needs-restarting -r`, or a newer +kernel of the running flavour installed) and whether it patches itself (unattended-upgrades, +dnf-automatic). `package_updates` holds the shared apt and dnf parsers: apt's suites become +an update's `origin`, a `-security` suite makes it a security update, and dnf's security +advisories do the same. + +**Update readers raise when they cannot read.** `get_available_updates` returns an empty +list only when nothing is pending; netOrk keeps "pending since" per package, and an empty +list for "don't know" would reset it. + `SystemdServicesMixin` (`get_services`, `manage_service`) is mixed in the same way, by the drivers whose host runs systemd. Listing the services, checking a unit name and reading an action's exit status are the same on every such host, so they are concrete diff --git a/napalm_device_types/__init__.py b/napalm_device_types/__init__.py index ecabd65..c83a3b1 100644 --- a/napalm_device_types/__init__.py +++ b/napalm_device_types/__init__.py @@ -38,6 +38,7 @@ instead of being restated on every role that happens to need it: * :class:`~napalm_device_types.dhcp.DhcpServerMixin` * :class:`~napalm_device_types.firewall_rules.FirewallRuleMixin` * :class:`~napalm_device_types.health_metrics.HealthMetricsMixin` +* :class:`~napalm_device_types.host_status.HostStatusMixin` * :class:`~napalm_device_types.host_reboot.HostRebootMixin` * :class:`~napalm_device_types.interface_filter.InterfaceFilterMixin` * :class:`~napalm_device_types.kernel.KernelFactsMixin` @@ -74,7 +75,15 @@ from napalm_device_types.packages import PackageManagementMixin from napalm_device_types.phone import PhoneDriver from napalm_device_types.ping_sweep import PingSweepMixin, driver_supports_ping from napalm_device_types.roles import primary_role_of, role_keys_of, roles_of +from napalm_device_types.host_status import HOST_STATUS_COMMAND, HostStatusMixin, parse_host_status +from napalm_device_types.package_updates import ( + APT_UPGRADABLE_COMMAND, + DNF_SECURITY_COMMAND, + parse_apt_upgradable, + parse_dnf_security, +) from napalm_device_types.services import ServiceControlMixin +from napalm_device_types.terminal import strip_terminal_codes from napalm_device_types.systemd import ( SYSTEMD_SERVICES_COMMAND, SystemdServicesMixin, @@ -95,6 +104,8 @@ __all__ = [ "FirewallDriver", "FirewallRuleMixin", "HealthMetricsMixin", + "HOST_STATUS_COMMAND", + "HostStatusMixin", "HostRebootMixin", "HypervisorDriver", "InterfaceFilterMixin", @@ -103,6 +114,12 @@ __all__ = [ "NatVpnMixin", "OSDriver", "PackageManagementMixin", + "APT_UPGRADABLE_COMMAND", + "DNF_SECURITY_COMMAND", + "parse_apt_upgradable", + "parse_dnf_security", + "parse_host_status", + "strip_terminal_codes", "KernelFactsMixin", "KERNEL_FACTS_COMMAND", "parse_kernel_facts", diff --git a/napalm_device_types/host_status.py b/napalm_device_types/host_status.py new file mode 100644 index 0000000..97c6525 --- /dev/null +++ b/napalm_device_types/host_status.py @@ -0,0 +1,177 @@ +# -*- coding: utf-8 -*- +"""Host status: does the host need a reboot, and does it patch itself? + +A patch run that installed a new kernel or libc has not closed anything until +the host restarts, so "reboot required" is part of being patched. Whether the +host installs updates on its own (unattended-upgrades, dnf-automatic) decides +how far netOrk's maintenance window reaches. Both are read the same way on every +Linux host, so the command and its parse live here once and a driver only +carries the command across. + +**Reboot required** is any of: + +- ``/var/run/reboot-required`` exists. Ubuntu always writes it; Debian does when + update-notifier or unattended-upgrades is installed. +- ``needs-restarting -r`` exits 1 (dnf-utils). +- A kernel newer than the running one is installed, of the same flavour. A + Raspberry Pi carries ``rpi-v8`` and ``rpi-2712`` builds side by side, and only + the running one's counts. + +It is ``None`` when none of these could be read, for example in a container +without a ``/lib/modules`` of its own. + +**Auto updates** is apt's ``APT::Periodic::Unattended-Upgrade`` (set, not "0", +and ``apt-daily-upgrade.timer`` not disabled) or an enabled dnf-automatic timer. +It is ``None`` on a host with neither apt nor dnf-automatic. +""" + +from __future__ import annotations + +import re +from typing import Dict, List, Optional, Tuple, TYPE_CHECKING + +from napalm_device_types.models import HostStatusDict +from napalm_device_types.terminal import strip_terminal_codes + +_BEGIN = "HSTAT_BEGIN" +_END = "HSTAT_END" +_REBOOT_FILE = "/var/run/reboot-required" +_APT_TIMER = "apt-daily-upgrade.timer" +_DNF_TIMERS = ("dnf-automatic.timer", "dnf-automatic-install.timer") + +#: One line, POSIX ``sh``, read-only, no privileges. The frame markers are +#: printed in two halves so that an echoing transport does not show them early. +#: Each timer is asked on its own: older systemd prints nothing for an unknown +#: unit, which would shift a combined answer. Run through a pipe, so nothing in +#: it sees a terminal and colours its output. +HOST_STATUS_COMMAND = ( + "{ printf '%s%s\\n' HSTAT_ BEGIN; " + f"[ -f {_REBOOT_FILE} ] && echo '[reboot-required]'; " + "if command -v needs-restarting >/dev/null 2>&1; then echo '[needs-restarting]'; " + "needs-restarting -r >/dev/null 2>&1; echo $?; fi; " + "echo '[kernel]'; uname -r; echo '[modules]'; ls -1 /lib/modules 2>/dev/null; " + "if command -v apt-config >/dev/null 2>&1; then echo '[apt-config]'; " + "apt-config dump 2>/dev/null | grep '^APT::Periodic::Unattended-Upgrade '; fi; " + f"echo '[timers]'; for u in {_APT_TIMER} {' '.join(_DNF_TIMERS)}; do " + 'printf \'%s %s\\n\' "$u" "$(systemctl is-enabled "$u" 2>/dev/null)"; done; ' + "printf '%s%s\\n' HSTAT_ END; } 2>/dev/null | cat" +) + +_PERIODIC = re.compile(r'^APT::Periodic::Unattended-Upgrade\s+"([^"]*)"') +_OFF_STATES = frozenset({"disabled", "masked"}) + + +def _sections(output: str) -> Dict[str, List[str]]: + lines = [line.strip() for line in strip_terminal_codes(output).splitlines()] + try: + start = lines.index(_BEGIN) + end = lines.index(_END, start) + except ValueError: + raise ValueError("no intact host status report in the output") from None + sections: Dict[str, List[str]] = {} + current: List[str] = [] + for line in lines[start + 1 : end]: + if line.startswith("[") and line.endswith("]"): + current = sections.setdefault(line[1:-1], []) + elif line: + current.append(line) + return sections + + +def _version_key(version: str) -> Tuple[object, ...]: + """Natural order: 6.8.0-142 after 6.8.0-87, 7.0.14 after 7.0.2.""" + return tuple(int(part) if part.isdigit() else part for part in re.split(r"(\d+)", version)) + + +def kernel_reboot_pending(running: str, installed: List[str]) -> Optional[str]: + """The newest installed kernel of the running flavour, if it is newer than the + running one; otherwise None. + + The flavour is what follows the last ``-`` (``generic``, ``amd64``, ``pve``, + ``v8``); a kernel of another flavour is never a reason to reboot. + """ + flavour = running.rsplit("-", 1)[-1] + same = [k for k in installed if k.rsplit("-", 1)[-1] == flavour] + if not same: + return None + newest = max(same, key=lambda k: _version_key(k.rsplit("-", 1)[0])) + if _version_key(newest.rsplit("-", 1)[0]) > _version_key(running.rsplit("-", 1)[0]): + return newest + return None + + +def _reboot(sections: Dict[str, List[str]]) -> Tuple[Optional[bool], Optional[str]]: + if "reboot-required" in sections: + return True, f"{_REBOOT_FILE} is present" + needs = sections.get("needs-restarting") + if needs and needs[0] == "1": + return True, "needs-restarting -r reports a reboot" + running = (sections.get("kernel") or [""])[0] + modules = sections.get("modules") or [] + newer = kernel_reboot_pending(running, modules) if running and modules else None + if newer: + return True, f"kernel {newer} installed, {running} running" + if needs or modules: + return False, None + return None, None + + +def _timer_states(sections: Dict[str, List[str]]) -> Dict[str, str]: + states: Dict[str, str] = {} + for line in sections.get("timers") or []: + unit, _, state = line.partition(" ") + states[unit] = state.strip() + return states + + +def _auto_updates(sections: Dict[str, List[str]]) -> Optional[bool]: + timers = _timer_states(sections) + if any(timers.get(t) == "enabled" for t in _DNF_TIMERS): + return True + if "apt-config" not in sections: + return None + match = next(filter(None, (_PERIODIC.match(line) for line in sections["apt-config"])), None) + switched_on = match is not None and match.group(1) not in ("", "0") + return switched_on and timers.get(_APT_TIMER) not in _OFF_STATES + + +def parse_host_status(output: str) -> HostStatusDict: + """Parse what :data:`HOST_STATUS_COMMAND` printed. + + :raises ValueError: when the output carries no intact report. + """ + sections = _sections(output) + required, reason = _reboot(sections) + return { + "reboot_required": required, + "reboot_reason": reason, + "auto_updates": _auto_updates(sections), + } + + +class HostStatusMixin: + """Adds :meth:`get_host_status` to a driver that can run a command on a Linux host. + + The template form, like :class:`~napalm_device_types.kernel.KernelFactsMixin`: + the reading and its parse are the same everywhere, and a driver supplies only + :meth:`_run_host_status_command`. Mixed in by the drivers that can, so + ``hasattr(driver, "get_host_status")`` stays a truthful answer. + """ + + if TYPE_CHECKING: # pragma: no cover - declared for type checkers only + + def _run_host_status_command(self, command: str) -> str: + """Run *command* on the host with ``sh`` and return what it printed.""" + ... + + def get_host_status(self) -> HostStatusDict: + """ + Returns whether the host needs a reboot and whether it patches itself. + + * reboot_required (bool or None) + * reboot_reason (string or None) + * auto_updates (bool or None) + + :raises ValueError: if the host's output carried no intact report. + """ + return parse_host_status(self._run_host_status_command(HOST_STATUS_COMMAND)) diff --git a/napalm_device_types/models.py b/napalm_device_types/models.py index 349dba3..80e5fa8 100644 --- a/napalm_device_types/models.py +++ b/napalm_device_types/models.py @@ -60,11 +60,30 @@ class ServiceDict(TypedDict): class UpdateDict(TypedDict): - """A software package that has a newer version available in the package repository.""" + """A software package that has a newer version available in the package repository. + + ``origin`` and ``security`` are optional: a reader that cannot tell leaves + them out, and netOrk treats a missing ``security`` as unknown. + """ name: str current_version: str new_version: str + #: Where the new version comes from, e.g. apt's suites "noble-updates,noble-security". + origin: NotRequired[Optional[str]] + #: True for a security update, False for a known other one, None when unknown. + security: NotRequired[Optional[bool]] + + +class HostStatusDict(TypedDict): + """What a host says about its own patch state (``HostStatusMixin.get_host_status``).""" + + #: True when the host needs a reboot to finish an update, None when it cannot tell. + reboot_required: Optional[bool] + #: Why, e.g. "kernel 6.8.0-142-generic installed, 6.8.0-139-generic running". + reboot_reason: Optional[str] + #: True when the host installs updates on its own (unattended-upgrades, dnf-automatic). + auto_updates: Optional[bool] # --------------------------------------------------------------------------- diff --git a/napalm_device_types/package_updates.py b/napalm_device_types/package_updates.py new file mode 100644 index 0000000..4c3e4ce --- /dev/null +++ b/napalm_device_types/package_updates.py @@ -0,0 +1,111 @@ +# -*- coding: utf-8 -*- +"""Pending package updates: which package, from where, and whether it is a security fix. + +A patch deadline -- "security updates within 14 days" -- needs to know which +pending update is a security update. apt says so in the suite a candidate comes +from (``noble-security``, ``stable-security``), dnf in its update advisories. +Reading that is the same for every driver whose host runs apt or dnf, so the +parsers live here once and a driver only carries the command across. + +apt: the suites a candidate comes from are its ``origin``; any suite ending in +``-security`` makes it a security update. A security fix that a later +``-updates`` build superseded shows only ``-updates`` and counts as not +security -- netOrk's CVE matching is what catches those. +""" + +from __future__ import annotations + +import re +from typing import Dict, List, Set + +from napalm_device_types.models import UpdateDict +from napalm_device_types.terminal import strip_terminal_codes + +#: Read-only, no root needed, in a fixed language so the parse holds, and +#: through a pipe: without a terminal apt draws no progress and no terminal +#: codes, one of which (``ESC >``) a screen-scraping transport took for a shell +#: prompt and stopped reading at. Its exit status is printed inside the group, +#: so it is apt's, not cat's. +APT_UPGRADABLE_COMMAND = "{ LC_ALL=C apt list --upgradable 2>/dev/null; echo __APT_RC=$?; } | cat" + +#: Read-only. Lists the packages that a pending security advisory covers. +DNF_SECURITY_COMMAND = "LC_ALL=C dnf updateinfo list --security --quiet 2>/dev/null" + +# openssl/noble-updates,noble-security 3.0.13-0ubuntu3.6 amd64 [upgradable from: 3.0.13-0ubuntu3.5] +_APT_LINE = re.compile(r"^(\S+)/(\S+)\s+(\S+)\s+\S+\s+\[upgradable from:\s+(\S+)\]") +_SECURITY_SUITE = "-security" +_APT_STATUS = re.compile(r"^__APT_RC=(\d+)\s*$", re.MULTILINE) + + +def _joined_lines(output: str) -> List[str]: + """Lines as apt printed them: a terminal wraps long ones, and the + continuation starts with a space.""" + lines: List[str] = [] + for line in output.splitlines(): + if line.startswith(" ") and lines: + lines[-1] += line.strip() + else: + lines.append(line) + return lines + + +def _apt_listing(output: str) -> str: + """The listing without its exit status, or ``ValueError`` when apt failed or + the output was cut short -- "could not read" must never look like "nothing + pending".""" + text = strip_terminal_codes(output) + statuses = _APT_STATUS.findall(text) + if not statuses: + raise ValueError("apt list --upgradable reported no exit status; the output was cut short") + if statuses[-1] != "0": + raise ValueError(f"apt list --upgradable failed with exit status {statuses[-1]}") + return _APT_STATUS.sub("", text) + + +def parse_apt_upgradable(output: str) -> List[UpdateDict]: + """Parse :data:`APT_UPGRADABLE_COMMAND`'s output, one entry per package. + + A package listed for several architectures (``libc6`` for amd64 and i386) + is one entry; it counts as a security update if any of its lines does. + + :raises ValueError: when apt failed or its exit status never arrived. + """ + by_name: Dict[str, UpdateDict] = {} + for line in _joined_lines(_apt_listing(output)): + match = _APT_LINE.match(line) + if not match: + continue + name, listed, new_version, current_version = match.groups() + suites = list(dict.fromkeys(listed.split(","))) # apt may list a suite twice + security = any(suite.endswith(_SECURITY_SUITE) for suite in suites) + seen = by_name.get(name) + if seen is not None: + seen["security"] = bool(seen.get("security")) or security + continue + by_name[name] = { + "name": name, + "current_version": current_version, + "new_version": new_version, + "origin": ",".join(suites), + "security": security, + } + return list(by_name.values()) + + +def nevra_name(nevra: str) -> str: + """The package name of an RPM ``name-[epoch:]version-release.arch``.""" + without_arch = nevra.rsplit(".", 1)[0] + return without_arch.rsplit("-", 2)[0] + + +def parse_dnf_security(output: str) -> Set[str]: + """The names of the packages a pending security advisory covers. + + Parses :data:`DNF_SECURITY_COMMAND`'s ``ADVISORY SEVERITY/Sec. NEVRA`` lines. + """ + names: Set[str] = set() + for line in output.splitlines(): + parts = line.split() + if len(parts) >= 3 and parts[1].endswith("/Sec."): + names.add(nevra_name(parts[-1])) + return names diff --git a/napalm_device_types/systemd.py b/napalm_device_types/systemd.py index 3733c32..d181d4b 100644 --- a/napalm_device_types/systemd.py +++ b/napalm_device_types/systemd.py @@ -35,6 +35,7 @@ from typing import Any, Dict, List, Set, Tuple, TYPE_CHECKING from napalm_device_types.models import ServiceDict from napalm_device_types.services import ServiceControlMixin +from napalm_device_types.terminal import strip_terminal_codes _BEGIN = "SVC_BEGIN" _END = "SVC_END" @@ -88,9 +89,6 @@ _RC_RE = re.compile(rf"^{_RC_MARKER}(\d+)\s*$", re.MULTILINE) _UNIT_RE = re.compile(r"(?:[A-Za-z0-9_.:@-]|\\x[0-9A-Fa-f]{2})+") _MAX_UNIT_LENGTH = 255 -#: Terminal colour codes, which systemctl adds when a transport gives it a terminal. -_ANSI_RE = re.compile(r"\x1b\[[0-9;?]*[A-Za-z]") - _ENABLED = frozenset({"enabled", "enabled-runtime"}) #: Unit file states of a service that is installed but need not be loaded. _INSTALLED = frozenset({"enabled", "enabled-runtime", "disabled", "indirect"}) @@ -139,7 +137,7 @@ def parse_action_result(output: str) -> Dict[str, Any]: Only the exit status decides. A job still running when ``timeout`` gave up is not reported as done, and output without a status is no success. """ - output = _ANSI_RE.sub("", output) + output = strip_terminal_codes(output) statuses = _RC_RE.findall(output) text = _RC_RE.sub("", output).strip() if not statuses: @@ -154,7 +152,7 @@ def parse_action_result(output: str) -> Dict[str, Any]: def _frame(output: str) -> List[str]: - lines = [line.strip() for line in _ANSI_RE.sub("", output).splitlines()] + lines = [line.strip() for line in strip_terminal_codes(output).splitlines()] try: start = lines.index(_BEGIN) end = lines.index(_END, start) diff --git a/napalm_device_types/terminal.py b/napalm_device_types/terminal.py new file mode 100644 index 0000000..c5e4c00 --- /dev/null +++ b/napalm_device_types/terminal.py @@ -0,0 +1,23 @@ +# -*- coding: utf-8 -*- +"""What a pseudo-terminal adds to a command's output, taken out again. + +A screen-scraping transport (netmiko) gives the remote command a terminal. Tools +then colour their output and draw progress: systemctl colours its errors, apt +switches the keypad mode with ``ESC =`` / ``ESC >``. Every parser in this package +reads the text without them. +""" + +from __future__ import annotations + +import re + +#: CSI sequences (colours, cursor), OSC sequences (window titles) and the +#: two-character escapes (``ESC =``, ``ESC >``, ``ESC (B``). +_TERMINAL_CODES = re.compile( + r"\x1b(?:\[[0-?]*[ -/]*[@-~]|\][^\x07\x1b]*(?:\x07|\x1b\\)|\([0-9A-Za-z]|[=>78DEHMNOc])" +) + + +def strip_terminal_codes(text: str) -> str: + """*text* without terminal escape sequences.""" + return _TERMINAL_CODES.sub("", text) diff --git a/napalm_device_types/updates.py b/napalm_device_types/updates.py index b952164..56e7934 100644 --- a/napalm_device_types/updates.py +++ b/napalm_device_types/updates.py @@ -13,7 +13,7 @@ this class in can never shadow a working implementation from a sibling base. from __future__ import annotations -from typing import List, TYPE_CHECKING +from typing import Any, Dict, List, TYPE_CHECKING from napalm_device_types.models import ApplyUpdatesResultDict, UpdateDict @@ -30,6 +30,14 @@ class UpdateMixin: * name (string) - package name * current_version (string) - currently installed version * new_version (string) - version available in the repository + * origin (string, optional) - where it comes from, e.g. apt's suites + * security (bool or None, optional) - a security update; leave it + out or None when the source does not say + + **An empty list means nothing is pending.** A reader that cannot + read -- no package index yet, an API that did not answer -- raises + instead: netOrk keeps "pending since" per package, and an empty + list for "don't know" would reset every one of those clocks. Example:: @@ -43,6 +51,16 @@ class UpdateMixin: """ ... + def refresh_available_updates(self) -> Dict[str, Any]: + """ + Refreshes the host's package index, so that :meth:`get_available_updates` + reports what the repositories offer now (``apt-get update``, + ``dnf makecache``, ``opkg update``, a firmware check). Installs nothing. + + :returns: ``{"success": bool, "output": str}`` + """ + ... + def apply_updates(self, packages: List[str]) -> ApplyUpdatesResultDict: """ Upgrades the given packages to the newest available version. diff --git a/pyproject.toml b/pyproject.toml index 21062d9..3335977 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "napalm-device-types" -version = "2.2.0" +version = "2.3.0" description = "Abstract device-type base classes for NAPALM drivers" readme = "README.md" requires-python = ">=3.10" diff --git a/tests/test_host_status.py b/tests/test_host_status.py new file mode 100644 index 0000000..c216b69 --- /dev/null +++ b/tests/test_host_status.py @@ -0,0 +1,200 @@ +"""Host status: does the host need a reboot, and does it patch itself? + +A patch run that installed a new kernel has not closed anything until the host +boots it, so "reboot required" is part of being patched. Whether the host +installs updates on its own (unattended-upgrades, dnf-automatic) is what netOrk +shows next to the window it governs. Both are read the same way on every Linux +host, so the command and its parse live here once (netOrk MVP 5). + +The fixtures are the real states of six hosts on netOrk's test server. +""" + +from __future__ import annotations + +import pytest + +from napalm_device_types import OSDriver +from napalm_device_types.host_status import ( + HOST_STATUS_COMMAND, + HostStatusMixin, + kernel_reboot_pending, + parse_host_status, +) + + +def _wire( + *, + reboot_file=False, + running="6.8.0-142-generic", + modules=("6.8.0-139-generic", "6.8.0-142-generic"), + needs_restarting=None, + periodic=None, + timer="enabled", + dnf_timers=("not-found", "not-found"), +): + lines = ["HSTAT_BEGIN"] + if reboot_file: + lines.append("[reboot-required]") + if needs_restarting is not None: + lines += ["[needs-restarting]", str(needs_restarting)] + lines += ["[kernel]", running, "[modules]", *modules] + if periodic is not None: + lines += ["[apt-config]", *periodic] + lines += [ + "[timers]", + f"apt-daily-upgrade.timer {timer}", + f"dnf-automatic.timer {dnf_timers[0]}", + f"dnf-automatic-install.timer {dnf_timers[1]}", + "HSTAT_END", + ] + return "\n".join(lines) + "\n" + + +UNATTENDED = ['APT::Periodic::Update-Package-Lists "1";', 'APT::Periodic::Unattended-Upgrade "1";'] + + +class TestRebootRequired: + def test_the_reboot_required_file_says_so(self): + status = parse_host_status(_wire(reboot_file=True)) + + assert status["reboot_required"] is True + assert "reboot-required" in status["reboot_reason"] + + def test_a_newer_installed_kernel_than_the_running_one(self): + """z2m-garden: running 6.8.0-139, 6.8.0-142 installed.""" + status = parse_host_status( + _wire( + running="6.8.0-139-generic", + modules=("6.8.0-87-generic", "6.8.0-139-generic", "6.8.0-142-generic"), + ) + ) + + assert status["reboot_required"] is True + assert "6.8.0-142-generic" in status["reboot_reason"] + + def test_the_newest_kernel_running_needs_none(self): + """vault-01: 6.8.0-142 running and newest; 6.8.0-94 sorts below it.""" + status = parse_host_status( + _wire(running="6.8.0-142-generic", modules=("6.8.0-94-generic", "6.8.0-142-generic")) + ) + + assert status["reboot_required"] is False + assert status["reboot_reason"] is None + + def test_needs_restarting_exit_1_means_reboot(self): + assert parse_host_status(_wire(needs_restarting=1))["reboot_required"] is True + + def test_needs_restarting_exit_0_does_not(self): + assert parse_host_status(_wire(needs_restarting=0))["reboot_required"] is False + + def test_no_kernel_information_is_unknown(self): + """A container has no /lib/modules of its own.""" + status = parse_host_status(_wire(modules=())) + + assert status["reboot_required"] is None + + +class TestKernelRebootPending: + @pytest.mark.parametrize( + ("running", "installed", "newer"), + [ + # Raspberry Pi: two flavours side by side; only the running one counts. + ( + "6.18.33+rpt-rpi-v8", + [ + "6.12.75+rpt-rpi-2712", + "6.12.75+rpt-rpi-v8", + "6.18.33+rpt-rpi-2712", + "6.18.33+rpt-rpi-v8", + ], + None, + ), + # Debian (OMV): 7.1.8 installed while 7.1.3 runs. + ( + "7.1.3+deb13-amd64", + ["6.12.57+deb13-amd64", "7.1.3+deb13-amd64", "7.1.8+deb13-amd64"], + "7.1.8+deb13-amd64", + ), + # Proxmox: 7.0.14-19 is newer than 7.0.2-6, numerically. + ("7.0.14-19-pve", ["7.0.14-19-pve", "7.0.2-6-pve"], None), + # Arch: the running kernel's modules were replaced by the upgrade. + ("6.10.5-arch1-1", ["6.10.9-arch1-1"], "6.10.9-arch1-1"), + ], + ) + def test_the_newer_kernel_of_the_running_flavour(self, running, installed, newer): + assert kernel_reboot_pending(running, installed) == newer + + +class TestAutoUpdates: + def test_unattended_upgrades_switched_on(self): + assert parse_host_status(_wire(periodic=UNATTENDED))["auto_updates"] is True + + def test_apt_without_the_setting_does_not_patch_itself(self): + """Proxmox and Raspberry Pi OS: apt-config answers, the setting is absent.""" + assert parse_host_status(_wire(periodic=[]))["auto_updates"] is False + + def test_switched_off_by_zero(self): + off = ['APT::Periodic::Unattended-Upgrade "0";'] + + assert parse_host_status(_wire(periodic=off))["auto_updates"] is False + + def test_a_disabled_timer_stops_it_even_when_configured(self): + status = parse_host_status(_wire(periodic=UNATTENDED, timer="disabled")) + + assert status["auto_updates"] is False + + def test_dnf_automatic(self): + status = parse_host_status(_wire(dnf_timers=("enabled", "not-found"))) + + assert status["auto_updates"] is True + + def test_neither_apt_nor_dnf_is_unknown(self): + assert parse_host_status(_wire())["auto_updates"] is None + + +class TestTheReport: + def test_a_cut_short_report_raises(self): + with pytest.raises(ValueError): + parse_host_status(_wire().replace("HSTAT_END\n", "")) + + def test_the_frame_is_not_in_the_command_itself(self): + assert "HSTAT_BEGIN" not in HOST_STATUS_COMMAND + assert "HSTAT_END" not in HOST_STATUS_COMMAND + + def test_it_runs_without_a_terminal(self): + """No colour codes from ls, nothing a screen scraper could take for a prompt.""" + assert HOST_STATUS_COMMAND.rstrip().endswith("| cat") + + def test_terminal_codes_are_dropped(self): + status = parse_host_status( + "\x1b[0m" + _wire(reboot_file=True).replace("HSTAT_END", "\x1b>HSTAT_END") + ) + + assert status["reboot_required"] is True + + def test_it_changes_nothing(self): + for word in ("rm ", "apt-get ", "dnf install", "systemctl start", "reboot"): + assert word not in HOST_STATUS_COMMAND.replace("reboot-required", "") + + +class _Driver(HostStatusMixin): + def __init__(self, reply: str) -> None: + self.reply = reply + self.commands: list = [] + + def _run_host_status_command(self, command: str) -> str: + self.commands.append(command) + return self.reply + + +class TestHostStatusMixin: + def test_a_driver_supplies_only_the_transport(self): + driver = _Driver(_wire(reboot_file=True, periodic=UNATTENDED)) + + status = driver.get_host_status() + + assert driver.commands == [HOST_STATUS_COMMAND] + assert (status["reboot_required"], status["auto_updates"]) == (True, True) + + def test_not_every_os_driver_has_it(self): + assert not hasattr(OSDriver, "get_host_status") diff --git a/tests/test_package_updates.py b/tests/test_package_updates.py new file mode 100644 index 0000000..8a10b2c --- /dev/null +++ b/tests/test_package_updates.py @@ -0,0 +1,149 @@ +"""Pending updates: which package, from where, and whether it closes a security hole. + +A patch deadline ("security updates within 14 days") needs to know which pending +update is a security update. apt says so in the suite a candidate comes from +(``noble-security``, ``stable-security``); dnf says so in its update advisories. +Reading that is the same for every driver whose host runs apt or dnf, so the +parsers live here once (netOrk MVP 5, #556). + +Fixture lines are from real hosts (Ubuntu 24.04, Debian 13 / OMV, Proxmox VE 9). +""" + +from __future__ import annotations + +import pytest + +from napalm_device_types.package_updates import ( + APT_UPGRADABLE_COMMAND, + nevra_name, + parse_apt_upgradable, + parse_dnf_security, +) + +UBUNTU = """\ +docker-compose-plugin/noble 5.6.0-1~ubuntu.24.04~noble amd64 [upgradable from: 5.5.1-1~ubuntu.24.04~noble] +openssl/noble-updates,noble-security 3.0.13-0ubuntu3.6 amd64 [upgradable from: 3.0.13-0ubuntu3.5] +__APT_RC=0 +""" + +DEBIAN = """\ +linux-image-amd64/stable-backports 7.1.13-1~bpo13+1 amd64 [upgradable from: 7.1.8-1~bpo13+1] +libssl3t64/stable-security 3.5.1-1+deb13u2 amd64 [upgradable from: 3.5.1-1+deb13u1] +__APT_RC=0 +""" + + +def _by_name(updates): + return {u["name"]: u for u in updates} + + +class TestAptUpgradable: + def test_each_line_is_a_package_with_both_versions(self): + update = _by_name(parse_apt_upgradable(UBUNTU))["docker-compose-plugin"] + + assert update["current_version"] == "5.5.1-1~ubuntu.24.04~noble" + assert update["new_version"] == "5.6.0-1~ubuntu.24.04~noble" + + def test_the_suites_are_its_origin(self): + updates = _by_name(parse_apt_upgradable(UBUNTU)) + + assert updates["openssl"]["origin"] == "noble-updates,noble-security" + assert updates["docker-compose-plugin"]["origin"] == "noble" + + def test_a_suite_apt_lists_twice_is_named_once(self): + line = ( + "fonts-opensymbol/noble-updates,noble-updates,noble-security,noble-security " + "4:102.12+LibO24.2.7-0ubuntu0.24.04.7 all [upgradable from: 4:102.12+LibO24.2.7-0ubuntu0.24.04.6]\n" + "__APT_RC=0\n" + ) + + assert parse_apt_upgradable(line)[0]["origin"] == "noble-updates,noble-security" + + @pytest.mark.parametrize( + ("output", "name", "security"), + [ + (UBUNTU, "openssl", True), + (UBUNTU, "docker-compose-plugin", False), + (DEBIAN, "libssl3t64", True), + (DEBIAN, "linux-image-amd64", False), + ], + ) + def test_a_security_suite_makes_it_a_security_update(self, output, name, security): + assert _by_name(parse_apt_upgradable(output))[name]["security"] is security + + def test_a_line_the_terminal_wrapped_is_joined(self): + wrapped = ( + "openssl/noble-updates,noble-security 3.0.13-0ubuntu3.6 amd64 [upgradable fro\n" + " m: 3.0.13-0ubuntu3.5]\n__APT_RC=0\n" + ) + + assert _by_name(parse_apt_upgradable(wrapped))["openssl"]["current_version"] == ( + "3.0.13-0ubuntu3.5" + ) + + def test_one_package_for_several_architectures_is_one_entry(self): + multiarch = ( + "libc6/noble-updates 2.39-0ubuntu8.5 amd64 [upgradable from: 2.39-0ubuntu8.4]\n" + "libc6/noble-updates,noble-security 2.39-0ubuntu8.5 i386 [upgradable from: 2.39-0ubuntu8.4]\n" + "__APT_RC=0\n" + ) + + updates = parse_apt_upgradable(multiarch) + + assert [u["name"] for u in updates] == ["libc6"] + assert updates[0]["security"] is True + + def test_noise_is_ignored(self): + noisy = "Listing... Done\nWARNING: apt does not have a stable CLI interface.\n" + UBUNTU + + assert len(parse_apt_upgradable(noisy)) == 2 + + def test_nothing_pending_is_an_empty_list(self): + assert parse_apt_upgradable("__APT_RC=0\n") == [] + + def test_a_list_without_its_exit_status_raises(self): + """Cut short: a transport stopped reading early, so nothing can be concluded.""" + with pytest.raises(ValueError): + parse_apt_upgradable(UBUNTU.replace("__APT_RC=0\n", "")) + + def test_a_failed_apt_raises(self): + with pytest.raises(ValueError, match="100"): + parse_apt_upgradable("E: Could not get lock\n__APT_RC=100\n") + + def test_terminal_codes_around_the_status_are_dropped(self): + """What a pseudo-terminal left on a Raspberry Pi OS host.""" + raw = "Listing... 0%\n\x1b[?1h\x1b=\n" + UBUNTU.replace("__APT_RC=0", "\x1b>__APT_RC=0") + + assert len(parse_apt_upgradable(raw)) == 2 + + def test_the_command_reads_without_root_or_a_terminal(self): + """Through a pipe apt draws no progress and no terminal codes; one of + those, ESC >, ended a screen-scraping read at a false prompt.""" + assert "LC_ALL=C apt list --upgradable" in APT_UPGRADABLE_COMMAND + assert APT_UPGRADABLE_COMMAND.rstrip().endswith("| cat") + assert "sudo" not in APT_UPGRADABLE_COMMAND + + +class TestDnfSecurity: + ADVISORIES = """\ +FEDORA-2024-1a2b3c4d5e Important/Sec. openssl-libs-1:3.1.4-2.fc40.x86_64 +FEDORA-2024-1a2b3c4d5e Important/Sec. openssl-1:3.1.4-2.fc40.x86_64 +RLSA-2024:1234 Moderate/Sec. kernel-core-5.14.0-427.13.1.el9_4.x86_64 +""" + + def test_the_names_of_packages_with_a_security_advisory(self): + assert parse_dnf_security(self.ADVISORIES) == {"openssl-libs", "openssl", "kernel-core"} + + def test_nothing_is_an_empty_set(self): + assert parse_dnf_security("") == set() + + @pytest.mark.parametrize( + ("nevra", "name"), + [ + ("openssl-libs-1:3.1.4-2.fc40.x86_64", "openssl-libs"), + ("kernel-core-5.14.0-427.13.1.el9_4.x86_64", "kernel-core"), + ("python3-dnf-4.14.0-9.el9.noarch", "python3-dnf"), + ], + ) + def test_the_name_of_a_nevra(self, nevra, name): + assert nevra_name(nevra) == name