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.

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",

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))

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]
 
 
 # ---------------------------------------------------------------------------

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

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)

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)

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.