Merge pull request 'feat: read what listens on which address, and which service it is, once for every driver' (#7) from feat/listening-sockets into main

Whether a service is reachable from outside its host is decided by the
address it listens on: 0.0.0.0:5432 is, 127.0.0.1:5432 is not. Reading
that is the same on every Linux host, so the command and its parse live
here once and a driver only carries the command across
(ListeningSocketsMixin, _run_listening_sockets_command).

One framed round trip: ss -lntup for every listening TCP and bound UDP
socket, then /proc/<pid>/cgroup for each process holding one, which names
the systemd service (v2, nested slices, v1's name=systemd hierarchy) or
the container (docker-<id>.scope, /docker/<id>) it runs in.

- Root: only root sees every process. The script goes as one sh -c
  argument, so a sudo -n prefix covers all of it; when that brings no
  report back the reading runs again unprivileged and says it is not
  attributed.
- No -H: iproute2 before 4.10 fails on it, which would read as nothing
  listening. The header is skipped instead.
- A host without ss raises ListeningSocketsUnavailable; a report cut short
  or a failing ss raises ValueError.
- The reading is raw: docker-proxy shows up as docker.service, loopback as
  loopback. What counts as reachable is the consumer's call.

2.4.0. For netOrk#658.

README.md

@@ -79,6 +79,34 @@ 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.
+
+`ListeningSocketsMixin` (`get_listening_sockets`) is mixed in the same way: every listening
+TCP and bound UDP socket from `ss -lntup`, with the systemd service or container behind it
+from `/proc/<pid>/cgroup`, in one round trip. A driver supplies
+`_run_listening_sockets_command(command, privileged=)`; the command arrives as one `sh -c`
+argument, so a `sudo -n` prefix covers all of it. Without root `ss` names only the login
+user's processes, and the reading says so (`attributed: false`) instead of failing. A host
+without `ss` raises `ListeningSocketsUnavailable`.
+
+**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
+here, and a driver supplies only `_run_service_command(command, *, privileged, timeout)`
+— how a command reaches its host and how it gains root there. The listing is one round
+trip (`list-unit-files` plus one `systemctl show` over every loaded unit) instead of an
+`is-enabled` and a `show` per unit. A host without systemd raises `SystemdUnavailable`,
+a `NotImplementedError`, so a driver can fall back to another init system.
+
 A function class may use the **template form** — public method concrete, the
 device-specific part a `_hook` declared under `if TYPE_CHECKING` — *when the base
 genuinely does work* on the result: normalising, sorting, validating, or orchestrating

napalm_device_types/__init__.py

@@ -38,14 +38,17 @@ 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`
+* :class:`~napalm_device_types.listening.ListeningSocketsMixin`
 * :class:`~napalm_device_types.mac_acl.MacAclMixin`
 * :class:`~napalm_device_types.nat_vpn.NatVpnMixin`
 * :class:`~napalm_device_types.packages.PackageManagementMixin`
 * :class:`~napalm_device_types.ping_sweep.PingSweepMixin`
 * :class:`~napalm_device_types.services.ServiceControlMixin`
+* :class:`~napalm_device_types.systemd.SystemdServicesMixin`
 * :class:`~napalm_device_types.updates.UpdateMixin`
 
 Introspection -- :func:`~napalm_device_types.roles.roles_of`,
@@ -66,6 +69,12 @@ from napalm_device_types.host_reboot import HostRebootMixin
 from napalm_device_types.interface_filter import InterfaceFilterMixin
 from napalm_device_types.kernel import KERNEL_FACTS_COMMAND, KernelFactsMixin, parse_kernel_facts
 from napalm_device_types.lag import add_lag_interfaces
+from napalm_device_types.listening import (
+    LISTENING_SOCKETS_COMMAND,
+    ListeningSocketsMixin,
+    ListeningSocketsUnavailable,
+    parse_listening_sockets,
+)
 from napalm_device_types.mac_acl import MacAclMixin
 from napalm_device_types.media import MediaDriver
 from napalm_device_types.nat_vpn import NatVpnMixin
@@ -73,7 +82,21 @@ 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,
+    SystemdUnavailable,
+    parse_systemd_services,
+)
 from napalm_device_types.updates import UpdateMixin
 from napalm_device_types.residential_gateway import ResidentialGatewayDriver
 from napalm_device_types.storage import StorageDriver
@@ -88,6 +111,8 @@ __all__ = [
     "FirewallDriver",
     "FirewallRuleMixin",
     "HealthMetricsMixin",
+    "HOST_STATUS_COMMAND",
+    "HostStatusMixin",
     "HostRebootMixin",
     "HypervisorDriver",
     "InterfaceFilterMixin",
@@ -96,9 +121,19 @@ __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",
+    "LISTENING_SOCKETS_COMMAND",
+    "ListeningSocketsMixin",
+    "ListeningSocketsUnavailable",
+    "parse_listening_sockets",
     "PhoneDriver",
     "PingSweepMixin",
     "PortSpec",
@@ -106,6 +141,10 @@ __all__ = [
     "ServiceControlMixin",
     "StorageDriver",
     "SwitchDriver",
+    "SYSTEMD_SERVICES_COMMAND",
+    "SystemdServicesMixin",
+    "SystemdUnavailable",
+    "parse_systemd_services",
     "UpdateMixin",
     "add_lag_interfaces",
     "driver_supports_ping",

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/listening.py

@@ -0,0 +1,218 @@
+# -*- coding: utf-8 -*-
+"""Listening sockets: what listens on which address, and which service it is.
+
+Whether a service is reachable from outside its host is decided by what it
+listens on -- ``0.0.0.0:5432`` is, ``127.0.0.1:5432`` is not -- and that is 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.
+
+**One round trip.** ``ss -lntup`` lists every listening TCP and bound UDP
+socket with the processes holding it; for each of those processes,
+``/proc/<pid>/cgroup`` says which systemd service or container it runs in. The
+report is framed, and a report whose end is missing raises: a list cut short
+must never read as sockets that closed.
+
+**Root, and without it.** Only root sees every process behind a socket.
+The whole script therefore goes to the host as one ``sh -c`` argument -- a
+driver that prefixes ``sudo -n`` would otherwise run only its first command as
+root. When that call brings no report back (no sudo, a wrong password), the
+command runs again without privilege: the sockets are still worth having, and
+the reading says it is not ``attributed``.
+
+**No ``-H``.** iproute2 before 4.10 has no option to leave out the header and
+fails on it, which would read as nothing listening. The parse skips the header
+instead. A host without ``ss`` at all (busybox, QNAP) raises
+:class:`ListeningSocketsUnavailable`.
+"""
+
+from __future__ import annotations
+
+import re
+from shlex import quote
+from typing import Dict, List, Optional, Tuple, TYPE_CHECKING
+
+from napalm_device_types.models import ListeningSocketDict, ListeningSocketsDict
+from napalm_device_types.terminal import strip_terminal_codes
+
+_BEGIN = "SOCK_BEGIN"
+_END = "SOCK_END"
+_NO_SS = "no-ss"
+_RC_RE = re.compile(r"^__SS_RC=(\d+)$")
+
+#: One line, POSIX ``sh``, read-only. The frame markers are printed in two
+#: halves so that a transport which echoes the command does not show them early.
+#: ``ss`` is in ``/usr/sbin`` on some systems, outside a login user's ``PATH``.
+LISTENING_SOCKETS_COMMAND = (
+    "PATH=$PATH:/usr/sbin:/sbin; "
+    "printf '%s%s\\n' SOCK_ BEGIN; "
+    "if command -v ss >/dev/null 2>&1; then "
+    "s=$(ss -lntup 2>&1); r=$?; echo '[ss]'; printf '%s\\n' \"$s\"; echo \"__SS_RC=$r\"; "
+    "echo '[cgroups]'; "
+    "for p in $(printf '%s\\n' \"$s\" | grep -o 'pid=[0-9]*' | cut -d= -f2 | sort -u); do "
+    "sed \"s|^|$p |\" /proc/$p/cgroup 2>/dev/null; done; "
+    "else echo '[no-ss]'; fi; "
+    "printf '%s%s\\n' SOCK_ END"
+)
+
+_PROTOCOLS = frozenset({"tcp", "udp"})
+#: ``users:(("nginx",pid=901,fd=6),("nginx",pid=900,fd=6))``
+_USER_RE = re.compile(r'\("((?:[^"\\]|\\.)*)",pid=(\d+),fd=\d+\)')
+#: The service a cgroup path runs in: its deepest ``*.service`` component.
+_SERVICE_RE = re.compile(r"/([^/]+)\.service(?=/|$)")
+#: A container's cgroup: ``docker-<id>.scope`` (systemd driver), ``/docker/<id>`` (cgroupfs).
+_CONTAINER_RE = re.compile(
+    r"(?:docker|libpod)-([0-9a-f]{64})\.scope|/(?:docker|libpod)/([0-9a-f]{64})(?=/|$)"
+)
+
+
+class ListeningSocketsUnavailable(NotImplementedError):
+    """The host has no ``ss``; there is nothing to read and nothing to retry."""
+
+
+def _frame(output: 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 listening socket report in the output") from None
+    return lines[start + 1 : end]
+
+
+def _sections(lines: List[str]) -> Dict[str, List[str]]:
+    sections: Dict[str, List[str]] = {}
+    current: List[str] = []
+    for line in lines:
+        if line.startswith("[") and line.endswith("]") and " " not in line:
+            current = sections.setdefault(line[1:-1], [])
+        else:
+            current.append(line)
+    return sections
+
+
+def _split_local(local: str) -> Optional[Tuple[str, Optional[str], int]]:
+    """``[fe80::1%eth0]:546`` -> ``("fe80::1", "eth0", 546)``; None if no port."""
+    host, sep, port = local.rpartition(":")
+    if not sep or not port.isdigit():
+        return None
+    if host.startswith("[") and host.endswith("]"):
+        host = host[1:-1]
+    address, _, zone = host.partition("%")
+    return address or "*", zone or None, int(port)
+
+
+def _cgroup_paths(lines: List[str]) -> Dict[int, str]:
+    """Each process's cgroup path: the unified hierarchy, or systemd's under v1."""
+    paths: Dict[int, str] = {}
+    for line in lines:
+        pid, _, entry = line.partition(" ")
+        parts = entry.split(":", 2)
+        if not pid.isdigit() or len(parts) != 3:
+            continue
+        hierarchy, controllers, path = parts
+        if (hierarchy == "0" and controllers == "") or controllers == "name=systemd":
+            paths[int(pid)] = path
+    return paths
+
+
+def _unit(path: Optional[str]) -> Optional[str]:
+    services = _SERVICE_RE.findall(path or "")
+    return services[-1] if services else None
+
+
+def _container(path: Optional[str]) -> Optional[str]:
+    match = _CONTAINER_RE.search(path or "")
+    return (match.group(1) or match.group(2)) if match else None
+
+
+def _socket(line: str, paths: Dict[int, str]) -> Optional[ListeningSocketDict]:
+    parts = line.split()
+    if len(parts) < 5 or parts[0] not in _PROTOCOLS:
+        return None
+    local = _split_local(parts[4])
+    if local is None:
+        return None
+    address, interface, port = local
+    users = _USER_RE.findall(line)
+    process, pid = (users[0][0], int(users[0][1])) if users else (None, None)
+    path = paths.get(pid) if pid is not None else None
+    return {
+        "proto": parts[0],
+        "address": address,
+        "port": port,
+        "interface": interface,
+        "process": process,
+        "pid": pid,
+        "unit": _unit(path),
+        "container_id": _container(path),
+    }
+
+
+def parse_listening_sockets(output: str) -> List[ListeningSocketDict]:
+    """Parse what :data:`LISTENING_SOCKETS_COMMAND` printed, sorted by protocol,
+    port and address.
+
+    :raises ListeningSocketsUnavailable: when the host has no ``ss``.
+    :raises ValueError: when the output carries no intact report, or ``ss`` failed.
+    """
+    sections = _sections(_frame(output))
+    if _NO_SS in sections:
+        raise ListeningSocketsUnavailable("the host has no ss")
+    ss_lines = sections.get("ss", [])
+    statuses = [m.group(1) for m in map(_RC_RE.match, ss_lines) if m]
+    if not statuses or statuses[-1] != "0":
+        detail = " ".join(line for line in ss_lines if not _RC_RE.match(line))[:200]
+        raise ValueError(f"ss did not list the sockets: {detail or 'no exit status'}")
+    paths = _cgroup_paths(sections.get("cgroups", []))
+    sockets = [s for s in (_socket(line, paths) for line in ss_lines) if s is not None]
+    return sorted(sockets, key=lambda s: (s["proto"], s["port"], s["address"], s["interface"] or ""))
+
+
+class ListeningSocketsMixin:
+    """Adds :meth:`get_listening_sockets` to a driver that can run a command on a Linux host.
+
+    The template form (README, "Function classes"): the command and its parse
+    are the same everywhere, so they are concrete here, and a driver supplies
+    only :meth:`_run_listening_sockets_command` -- how a command reaches its
+    host, and how it gains root there. Mixed in by the drivers that can, not by
+    :class:`~napalm_device_types.os.OSDriver`: a Windows host is an OS driver
+    too, and ``hasattr(driver, "get_listening_sockets")`` has to stay truthful.
+    """
+
+    if TYPE_CHECKING:  # pragma: no cover - declared for type checkers only
+
+        def _run_listening_sockets_command(self, command: str, *, privileged: bool) -> str:
+            """Run *command* on the host and return what it printed; as root
+            when *privileged*. The command is a single ``sh -c`` invocation."""
+            ...
+
+    def get_listening_sockets(self) -> ListeningSocketsDict:
+        """
+        Returns every listening TCP and bound UDP socket, with the process,
+        systemd service and container behind it.
+
+        * attributed (bool) - read as root, so every process is named
+        * sockets (list) - see :class:`~napalm_device_types.models.ListeningSocketDict`
+
+        Example::
+
+            {
+                "attributed": True,
+                "sockets": [
+                    {"proto": "tcp", "address": "0.0.0.0", "port": 5432,
+                     "interface": None, "process": "postgres", "pid": 812,
+                     "unit": "postgresql@16-main", "container_id": None},
+                ],
+            }
+
+        :raises ListeningSocketsUnavailable: if the host has no ``ss``.
+        :raises ValueError: if neither reading carried an intact report.
+        """
+        command = f"sh -c {quote(LISTENING_SOCKETS_COMMAND)}"
+        try:
+            output = self._run_listening_sockets_command(command, privileged=True)
+            return {"attributed": True, "sockets": parse_listening_sockets(output)}
+        except ValueError:
+            pass
+        output = self._run_listening_sockets_command(command, privileged=False)
+        return {"attributed": False, "sockets": parse_listening_sockets(output)}

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]
 
 
 # ---------------------------------------------------------------------------
@@ -316,6 +335,37 @@ class KernelFactsDict(TypedDict):
     config: Optional[Dict[str, str]]  # build configuration, set options only; quotes stripped
 
 
+class ListeningSocketDict(TypedDict):
+    """A TCP socket that listens, or a UDP socket that is bound, on the host.
+
+    One entry per socket as ``ss`` lists it. ``address`` is printed the way ss
+    prints it, without brackets or zone: ``0.0.0.0`` and ``::`` are every
+    address of their family, ``*`` every address of both. Whether that is
+    reachable from outside the host is the consumer's call.
+    """
+
+    proto: str  # "tcp" or "udp"
+    address: str  # "0.0.0.0", "::", "*", "127.0.0.1", "::ffff:127.0.0.1", ...
+    port: int
+    interface: Optional[str]  # the %zone a socket is bound to ("lo", "eth0"), if any
+    process: Optional[str]  # the first process holding it; None without one or without root
+    pid: Optional[int]
+    unit: Optional[str]  # the process's systemd service, without ".service"
+    container_id: Optional[str]  # the full container ID, for a container on the host network
+
+
+class ListeningSocketsDict(TypedDict):
+    """What ``ListeningSocketsMixin.get_listening_sockets`` read.
+
+    ``attributed`` is False when the reading ran without root: ``ss`` then names
+    only the login user's own processes, so a socket without a process means
+    "not told", not "the kernel's".
+    """
+
+    attributed: bool
+    sockets: List[ListeningSocketDict]
+
+
 class PortForwardDict(TypedDict):
     """A port the WAN side can reach, forwarded to a host inside.
 

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

@@ -0,0 +1,329 @@
+# -*- coding: utf-8 -*-
+"""systemd services: listing them in one round trip, and starting and stopping them.
+
+What systemd reports about its services, and how one is started or stopped, is
+the same on every host that runs it. So the command, its parse, the check of a
+unit name and the reading of an action's exit status live here once, and a
+driver only carries a command across: SSH, an API's exec endpoint, whatever it
+has.
+
+**Listing.** One command prints the installed unit files and, for every loaded
+service unit, what ``systemctl show`` knows about it -- state, boot state and
+main PID together, instead of asking ``systemctl is-enabled`` and ``systemctl
+show`` once per unit (two hundred round trips on an ordinary Linux host). The
+report is framed, and a report whose end is missing raises: a list cut short
+must never read as services that went away.
+
+**What counts as enabled.** A unit file state of ``enabled`` or
+``enabled-runtime``. ``static`` does not: such a unit starts only when
+something else pulls it in, and calling it enabled made every one of them look
+like a service of the host. The state is read from ``UnitFileState``, never
+from a column of ``list-unit-files``, whose second column has been followed by
+a preset column since systemd 245.
+
+**Starting and stopping.** ``systemctl`` runs bounded by ``timeout`` and never
+asks for a password, and its exit status is printed after it. The marker also
+keeps the output from ever being empty, which a transport that retries on an
+empty answer would otherwise take as a reason to run the action twice.
+"""
+
+from __future__ import annotations
+
+import re
+from shlex import quote
+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"
+_NO_SYSTEMD = "no-systemd"
+_SUFFIX = ".service"
+
+#: What ``systemctl show`` prints per unit. It prints them in its own order.
+_PROPERTIES = "Id,Names,LoadState,ActiveState,SubState,UnitFileState,MainPID"
+
+#: Picks the units whose file state is ``generated`` out of ``systemctl show``'s
+#: output, whatever order it prints the properties in.
+_GENERATED_AWK = (
+    'awk -F= \'NF<2{id="";g=0;next} $1=="Id"{id=$2} '
+    '$1=="UnitFileState"{g=($2=="generated")} id!=""&&g{print id;id="";g=0}\''
+)
+
+#: One line, POSIX ``sh``, read-only. The frame markers are printed in two
+#: halves so that a transport which echoes the command does not show them early.
+#: ``xargs -0`` passes escaped names such as ``foo\x2dbar.service`` unchanged.
+#: A generated unit -- the wrapper systemd makes for a SysV script -- has no unit
+#: file whose state says whether it starts at boot; ``systemctl is-enabled``
+#: asks the script's rc links instead, for those few units only.
+SYSTEMD_SERVICES_COMMAND = (
+    "printf '%s%s\\n' SVC_ BEGIN; "
+    "[ -d /run/systemd/system ] || echo '[no-systemd]'; "
+    "echo '[files]'; systemctl list-unit-files --type=service --no-legend --no-pager 2>/dev/null; "
+    "echo '[units]'; s=$(systemctl list-units --type=service --all --no-legend --no-pager --plain "
+    "2>/dev/null | awk '{print $1}' | tr '\\n' '\\0' | xargs -0 -r systemctl show --no-pager "
+    f"-p {_PROPERTIES} -- 2>/dev/null); printf '%s\\n' \"$s\"; "
+    f"echo '[generated]'; printf '%s\\n' \"$s\" | {_GENERATED_AWK} | while read -r u; do "
+    'printf \'%s %s\\n\' "$u" "$(systemctl is-enabled -- "$u" 2>/dev/null)"; done; '
+    "printf '%s%s\\n' SVC_ END"
+)
+
+#: The lifecycle actions :meth:`SystemdServicesMixin.manage_service` accepts.
+SERVICE_ACTIONS = ("start", "stop", "restart", "enable", "disable")
+
+#: Seconds an action may run on the host before ``timeout`` stops waiting for
+#: it. systemd itself carries on with the job.
+ACTION_TIMEOUT = 45
+
+#: What a transport should allow for one command: the action's own bound plus
+#: the round trip around it.
+_TRANSPORT_TIMEOUT = ACTION_TIMEOUT + 15
+
+_TIMED_OUT = 124  # timeout(1)'s exit status when the time ran out
+_RC_MARKER = "__SVC_RC="
+_RC_RE = re.compile(rf"^{_RC_MARKER}(\d+)\s*$", re.MULTILINE)
+
+#: The characters systemd allows in a unit name, with ``\xHH`` for any other byte.
+_UNIT_RE = re.compile(r"(?:[A-Za-z0-9_.:@-]|\\x[0-9A-Fa-f]{2})+")
+_MAX_UNIT_LENGTH = 255
+
+_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"})
+
+
+class SystemdUnavailable(NotImplementedError):
+    """The host does not run systemd; a driver may fall back to another init system."""
+
+
+def unit_name(name: str) -> str:
+    """*name* as a service unit's name without ``.service``, or ``ValueError``.
+
+    Accepts template instances (``wg-quick@wg0``), dots (``snapd.apparmor``),
+    colons and systemd's ``\\xHH`` escapes. Refuses a bare template
+    (``getty@``), a leading ``-`` that a command would read as an option, and
+    anything a shell would read.
+    """
+    base = name[: -len(_SUFFIX)] if name.endswith(_SUFFIX) else name
+    if (
+        not _UNIT_RE.fullmatch(base)
+        or base.startswith("-")
+        or base.endswith("@")
+        or len(base) + len(_SUFFIX) > _MAX_UNIT_LENGTH
+    ):
+        raise ValueError(f"Invalid service name: {name!r}")
+    return base
+
+
+def service_action_command(name: str, action: str) -> str:
+    """The shell command that applies *action* to the service *name*.
+
+    :raises ValueError: for an unknown action or an invalid name.
+    """
+    if action not in SERVICE_ACTIONS:
+        raise ValueError(f"Invalid action {action!r}; use one of {', '.join(SERVICE_ACTIONS)}")
+    unit = quote(unit_name(name) + _SUFFIX)
+    return (
+        f"timeout {ACTION_TIMEOUT} systemctl --no-ask-password {action} -- {unit} 2>&1; "
+        f"echo {_RC_MARKER}$?"
+    )
+
+
+def parse_action_result(output: str) -> Dict[str, Any]:
+    """``{"success", "output"}`` from what :func:`service_action_command` printed.
+
+    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 = strip_terminal_codes(output)
+    statuses = _RC_RE.findall(output)
+    text = _RC_RE.sub("", output).strip()
+    if not statuses:
+        return {"success": False, "output": text or "No exit status came back from the host."}
+    status = int(statuses[-1])
+    if status == 0:
+        return {"success": True, "output": text}
+    if status == _TIMED_OUT:
+        note = f"Still running after {ACTION_TIMEOUT} s; systemd carries on with the job."
+        return {"success": False, "output": f"{text}\n{note}".strip()}
+    return {"success": False, "output": text or f"systemctl exited with status {status}."}
+
+
+def _frame(output: 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 systemd service report in the output") from None
+    return lines[start + 1 : end]
+
+
+def _sections(lines: List[str]) -> Dict[str, List[str]]:
+    sections: Dict[str, List[str]] = {}
+    current: List[str] = []
+    for line in lines:
+        if line.startswith("[") and line.endswith("]"):
+            current = sections.setdefault(line[1:-1], [])
+        else:
+            current.append(line)
+    return sections
+
+
+def _unit_blocks(lines: List[str]) -> List[Dict[str, str]]:
+    """``systemctl show``'s output, one dict per unit.
+
+    Units are separated by a blank line -- except where ``xargs`` split the
+    list over two runs and the blocks meet, so a key seen twice starts the next
+    unit as well.
+    """
+    blocks: List[Dict[str, str]] = []
+    current: Dict[str, str] = {}
+    for line in lines:
+        key, sep, value = line.partition("=")
+        if not sep or key in current:
+            if current:
+                blocks.append(current)
+            current = {}
+        if sep:
+            current[key] = value
+    if current:
+        blocks.append(current)
+    return blocks
+
+
+def _base(unit: str) -> str:
+    return unit[: -len(_SUFFIX)]
+
+
+def _main_pid(block: Dict[str, str]) -> int:
+    try:
+        return int(block.get("MainPID") or 0)
+    except ValueError:
+        return 0
+
+
+def _loaded(blocks: List[Dict[str, str]]) -> Tuple[Dict[str, ServiceDict], Set[str]]:
+    """The loaded services, and every name they go by (aliases included)."""
+    services: Dict[str, ServiceDict] = {}
+    names: Set[str] = set()
+    for block in blocks:
+        unit = block.get("Id", "")
+        if not unit.endswith(_SUFFIX) or block.get("LoadState") == "not-found":
+            continue
+        names.update(block.get("Names", unit).split())
+        running = block.get("ActiveState") == "active" and block.get("SubState") == "running"
+        services[_base(unit)] = {
+            "name": _base(unit),
+            "running": running,
+            "enabled": block.get("UnitFileState") in _ENABLED,
+            "pid": _main_pid(block) if running else 0,
+        }
+    return services, names
+
+
+def _installed(lines: List[str], known: Set[str]) -> Dict[str, ServiceDict]:
+    """Installed services that are not loaded: neither running nor starting now.
+
+    Templates, static units and aliases are left out -- the last also when an
+    older systemd lists an alias as ``enabled``, which is why every name a
+    loaded unit goes by is skipped.
+    """
+    services: Dict[str, ServiceDict] = {}
+    for line in lines:
+        parts = line.split()
+        if len(parts) < 2:
+            continue
+        unit, state = parts[0], parts[1]
+        if (
+            not unit.endswith(_SUFFIX)
+            or unit.endswith("@" + _SUFFIX)
+            or unit in known
+            or state not in _INSTALLED
+        ):
+            continue
+        services[_base(unit)] = {
+            "name": _base(unit),
+            "running": False,
+            "enabled": state in _ENABLED,
+            "pid": 0,
+        }
+    return services
+
+
+def _apply_generated(services: Dict[str, ServiceDict], lines: List[str]) -> None:
+    """Take a generated unit's boot state from ``is-enabled``'s answer."""
+    for line in lines:
+        parts = line.split()
+        if len(parts) == 2 and parts[0].endswith(_SUFFIX) and _base(parts[0]) in services:
+            services[_base(parts[0])]["enabled"] = parts[1] in _ENABLED
+
+
+def parse_systemd_services(output: str) -> List[ServiceDict]:
+    """Parse what :data:`SYSTEMD_SERVICES_COMMAND` printed, sorted by name.
+
+    Lists every loaded service unit but those that are not found, and every
+    installed one that is not loaded.
+
+    :raises SystemdUnavailable: when the host does not run systemd.
+    :raises ValueError: when the output carries no intact report.
+    """
+    sections = _sections(_frame(output))
+    if _NO_SYSTEMD in sections:
+        raise SystemdUnavailable("the host does not run systemd")
+    loaded, known = _loaded(_unit_blocks(sections.get("units", [])))
+    _apply_generated(loaded, sections.get("generated", []))
+    merged = {**_installed(sections.get("files", []), known), **loaded}
+    return [merged[name] for name in sorted(merged)]
+
+
+class SystemdServicesMixin(ServiceControlMixin):
+    """Implements :class:`ServiceControlMixin` for a driver whose host runs systemd.
+
+    The template form (README, "Function classes"): the command, the parse,
+    the check of the name and the reading of the exit status are the same
+    everywhere, so they are concrete here, and a driver supplies only
+    :meth:`_run_service_command` -- how a command reaches its host, and how it
+    gains root there when it needs to.
+    """
+
+    if TYPE_CHECKING:  # pragma: no cover - declared for type checkers only
+
+        def _run_service_command(self, command: str, *, privileged: bool, timeout: int) -> str:
+            """Run *command* with ``sh`` on the host and return what it printed.
+
+            *privileged* commands change the system and need root; *timeout*
+            is how long the transport should wait for the output, in seconds.
+            """
+            ...
+
+    def get_services(self) -> List[ServiceDict]:
+        """
+        Returns the services systemd knows, in one round trip.
+
+        * name (string) - the unit name without ``.service``
+        * running (bool) - active and running
+        * enabled (bool) - the unit file is enabled
+        * pid (int) - the main process; 0 when not running
+
+        :raises SystemdUnavailable: if the host does not run systemd.
+        :raises ValueError: if the host's output carried no intact report.
+        """
+        output = self._run_service_command(
+            SYSTEMD_SERVICES_COMMAND, privileged=False, timeout=_TRANSPORT_TIMEOUT
+        )
+        return parse_systemd_services(output)
+
+    def manage_service(self, name: str, action: str) -> Dict[str, Any]:
+        """
+        Applies *action* (start, stop, restart, enable, disable) to the service *name*.
+
+        :returns: ``{"success": bool, "output": str}``
+        :raises ValueError: for an unknown action or an invalid name, before
+            anything is sent.
+        """
+        command = service_action_command(name, action)
+        output = self._run_service_command(command, privileged=True, timeout=_TRANSPORT_TIMEOUT)
+        return parse_action_result(output)

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.

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "napalm-device-types"
-version = "2.1.0"
+version = "2.4.0"
 description = "Abstract device-type base classes for NAPALM drivers"
 readme = "README.md"
 requires-python = ">=3.10"

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

tests/test_listening_sockets.py

@@ -0,0 +1,274 @@
+"""get_listening_sockets: what listens on which address, and which service it is.
+
+Whether a service is reachable from outside its host is decided by what it
+listens on -- ``0.0.0.0:5432`` is, ``127.0.0.1:5432`` is not. Reading that is
+the same on every Linux host: ``ss`` for the sockets, ``/proc/<pid>/cgroup``
+for the systemd unit or container a process belongs to. So both live here once,
+and a driver only carries the command across (#658 in netOrk).
+"""
+
+from __future__ import annotations
+
+import shutil
+import subprocess
+
+import pytest
+
+from napalm_device_types import OSDriver
+from napalm_device_types.listening import (
+    LISTENING_SOCKETS_COMMAND,
+    ListeningSocketsMixin,
+    ListeningSocketsUnavailable,
+    parse_listening_sockets,
+)
+
+CID = "4f1c" + "0" * 60
+SS = """Netid State  Recv-Q Send-Q          Local Address:Port  Peer Address:PortProcess
+udp   UNCONN 0      0               127.0.0.53%lo:53         0.0.0.0:*    users:(("systemd-resolve",pid=612,fd=13))
+udp   UNCONN 0      0              0.0.0.0%ens18:68          0.0.0.0:*    users:(("systemd-network",pid=590,fd=22))
+tcp   LISTEN 0      4096                  0.0.0.0:5432        0.0.0.0:*    users:(("postgres",pid=812,fd=6))
+tcp   LISTEN 0      128                      [::]:22             [::]:*    users:(("sshd",pid=700,fd=4))
+tcp   LISTEN 0      511                         *:80                *:*    users:(("nginx",pid=901,fd=6),("nginx",pid=900,fd=6))
+tcp   LISTEN 0      4096       [::ffff:127.0.0.1]:8125              *:*    users:(("statsd",pid=950,fd=3))
+tcp   LISTEN 0      4096                  0.0.0.0:8080        0.0.0.0:*    users:(("docker-proxy",pid=1200,fd=4))
+tcp   LISTEN 0      4096                  0.0.0.0:9100        0.0.0.0:*    users:(("node_exporter",pid=1300,fd=3))
+tcp   LISTEN 0      64                    0.0.0.0:2049        0.0.0.0:*
+"""
+CGROUPS = f"""612 0::/system.slice/systemd-resolved.service
+590 0::/system.slice/systemd-networkd.service
+812 0::/system.slice/system-postgresql.slice/postgresql@16-main.service
+700 0::/system.slice/ssh.service
+900 0::/system.slice/nginx.service
+901 0::/system.slice/nginx.service
+950 0::/user.slice/user-1000.slice/session-3.scope
+1200 0::/system.slice/docker.service
+1300 0::/system.slice/docker-{CID}.scope
+"""
+
+
+def _wire(ss: str = SS, cgroups: str = CGROUPS, *, rc: int = 0, noise: str = "") -> str:
+    """The report as the command prints it, framed."""
+    return f"{noise}SOCK_BEGIN\n[ss]\n{ss}__SS_RC={rc}\n[cgroups]\n{cgroups}SOCK_END\n"
+
+
+def _by_port(sockets: list) -> dict:
+    return {(s["proto"], s["port"], s["address"]): s for s in sockets}
+
+
+class TestParsing:
+    def test_a_socket_comes_with_its_process_and_unit(self):
+        sockets = _by_port(parse_listening_sockets(_wire()))
+
+        assert sockets[("tcp", 5432, "0.0.0.0")] == {
+            "proto": "tcp",
+            "address": "0.0.0.0",
+            "port": 5432,
+            "interface": None,
+            "process": "postgres",
+            "pid": 812,
+            "unit": "postgresql@16-main",
+            "container_id": None,
+        }
+
+    @pytest.mark.parametrize(
+        "local, address, interface",
+        [
+            ("[::]:22", "::", None),
+            ("*:80", "*", None),
+            ("127.0.0.53%lo:53", "127.0.0.53", "lo"),
+            ("0.0.0.0%ens18:68", "0.0.0.0", "ens18"),
+            ("[::ffff:127.0.0.1]:8125", "::ffff:127.0.0.1", None),
+            ("[fe80::1%eth0]:546", "fe80::1", "eth0"),
+            (":::22", "::", None),  # iproute2 4.9 prints IPv6 without brackets
+        ],
+    )
+    def test_every_address_form(self, local: str, address: str, interface):
+        line = f"tcp LISTEN 0 128 {local} *:* users:((\"x\",pid=1,fd=3))\n"
+
+        [socket] = parse_listening_sockets(_wire(line, "1 0::/system.slice/x.service\n"))
+
+        assert (socket["address"], socket["interface"]) == (address, interface)
+
+    def test_the_first_of_several_processes_names_the_socket(self):
+        nginx = _by_port(parse_listening_sockets(_wire()))[("tcp", 80, "*")]
+
+        assert (nginx["process"], nginx["pid"], nginx["unit"]) == ("nginx", 901, "nginx")
+
+    def test_a_socket_without_a_process_is_kept(self):
+        """The kernel's nfsd, or every socket of another user without root."""
+        nfs = _by_port(parse_listening_sockets(_wire()))[("tcp", 2049, "0.0.0.0")]
+
+        assert (nfs["process"], nfs["pid"], nfs["unit"], nfs["container_id"]) == (
+            None,
+            None,
+            None,
+            None,
+        )
+
+    def test_the_header_is_no_socket(self):
+        sockets = parse_listening_sockets(_wire())
+
+        assert len(sockets) == 9
+        assert all(s["proto"] in ("tcp", "udp") for s in sockets)
+
+    def test_sorted_by_protocol_port_and_address(self):
+        keys = [(s["proto"], s["port"], s["address"]) for s in parse_listening_sockets(_wire())]
+
+        assert keys == sorted(keys)
+
+    def test_nothing_listening_is_an_empty_list(self):
+        assert parse_listening_sockets(_wire(SS.splitlines(True)[0], "")) == []
+
+
+class TestCgroups:
+    def _unit_and_container(self, cgroup_lines: str) -> tuple:
+        line = 'tcp LISTEN 0 128 0.0.0.0:1 0.0.0.0:* users:(("p",pid=5,fd=3))\n'
+        [socket] = parse_listening_sockets(_wire(line, cgroup_lines))
+        return socket["unit"], socket["container_id"]
+
+    def test_a_container_on_the_host_network(self):
+        assert self._unit_and_container(f"5 0::/system.slice/docker-{CID}.scope\n") == (
+            None,
+            CID,
+        )
+
+    def test_a_container_under_the_cgroupfs_driver(self):
+        assert self._unit_and_container(f"5 0::/docker/{CID}\n") == (None, CID)
+
+    def test_cgroup_v1_reads_the_systemd_hierarchy(self):
+        lines = "5 12:pids:/user.slice\n5 1:name=systemd:/system.slice/ssh.service\n"
+
+        assert self._unit_and_container(lines) == ("ssh", None)
+
+    def test_an_escaped_template_instance(self):
+        lines = "5 0::/system.slice/system-wg\\x2dquick.slice/wg-quick@wg0.service\n"
+
+        assert self._unit_and_container(lines) == ("wg-quick@wg0", None)
+
+    def test_a_login_session_is_no_unit(self):
+        assert self._unit_and_container("5 0::/user.slice/user-1000.slice/session-3.scope\n") == (
+            None,
+            None,
+        )
+
+    def test_a_process_gone_before_its_cgroup_was_read(self):
+        assert self._unit_and_container("") == (None, None)
+
+    def test_the_docker_daemons_own_proxy_stays_raw(self):
+        """docker-proxy runs in docker.service. Telling it apart from a real
+        listener is the consumer's business; the reading reports what is there."""
+        proxy = _by_port(parse_listening_sockets(_wire()))[("tcp", 8080, "0.0.0.0")]
+
+        assert (proxy["process"], proxy["unit"]) == ("docker-proxy", "docker")
+
+
+class TestFailures:
+    def test_whatever_surrounds_the_frame_is_ignored(self):
+        noisy = _wire(noise="$ sh -c '...'\nWelcome to Ubuntu\n")
+
+        assert len(parse_listening_sockets(noisy)) == 9
+
+    def test_output_without_the_frame_raises(self):
+        with pytest.raises(ValueError):
+            parse_listening_sockets("sudo: a password is required\n")
+
+    def test_a_report_cut_short_raises(self):
+        with pytest.raises(ValueError):
+            parse_listening_sockets(_wire()[:-len("SOCK_END\n")])
+
+    def test_a_host_without_ss_is_unavailable(self):
+        with pytest.raises(ListeningSocketsUnavailable):
+            parse_listening_sockets("SOCK_BEGIN\n[no-ss]\nSOCK_END\n")
+
+    def test_ss_failing_raises(self):
+        with pytest.raises(ValueError):
+            parse_listening_sockets(_wire("ss: invalid option -- 'p'\n", "", rc=1))
+
+    def test_unavailable_is_not_a_value_error(self):
+        """A caller retries a broken report without privilege; a host without
+        ss has nothing to retry."""
+        assert not issubclass(ListeningSocketsUnavailable, ValueError)
+
+
+class TestTheCommand:
+    def test_the_frame_is_not_in_the_command_itself(self):
+        """An echoing transport prints the command back; the markers must only
+        appear once the command has run."""
+        assert "SOCK_BEGIN" not in LISTENING_SOCKETS_COMMAND
+        assert "SOCK_END" not in LISTENING_SOCKETS_COMMAND
+
+    def test_it_writes_and_changes_nothing(self):
+        for verb in ("sudo", " > ", ">>", "kill", "rm ", "systemctl "):
+            assert verb not in LISTENING_SOCKETS_COMMAND
+
+    def test_it_is_posix_sh(self):
+        result = subprocess.run(
+            ["sh", "-n", "-c", LISTENING_SOCKETS_COMMAND], capture_output=True, text=True
+        )
+        assert result.returncode == 0, result.stderr
+
+    @pytest.mark.skipif(shutil.which("ss") is None, reason="needs ss on this host")
+    def test_it_runs_and_parses_on_this_host(self):
+        out = subprocess.run(
+            ["sh", "-c", LISTENING_SOCKETS_COMMAND], capture_output=True, text=True, timeout=60
+        ).stdout
+
+        sockets = parse_listening_sockets(out)
+
+        assert all(s["proto"] in ("tcp", "udp") and 0 < s["port"] < 65536 for s in sockets)
+
+
+class _Driver(ListeningSocketsMixin):
+    def __init__(self, privileged: str, unprivileged: str = "") -> None:
+        self.answers = {True: privileged, False: unprivileged}
+        self.calls: list = []
+
+    def _run_listening_sockets_command(self, command: str, *, privileged: bool) -> str:
+        self.calls.append((command, privileged))
+        return self.answers[privileged]
+
+
+class TestTheTemplate:
+    def test_a_driver_supplies_only_the_transport(self):
+        driver = _Driver(_wire())
+
+        reading = driver.get_listening_sockets()
+
+        assert reading["attributed"] is True
+        assert len(reading["sockets"]) == 9
+        [(command, privileged)] = driver.calls
+        assert privileged is True
+
+    def test_the_command_reaches_root_as_one_shell(self):
+        """``sudo -n a; b`` runs only ``a`` as root: the whole script goes as
+        one ``sh -c`` argument."""
+        driver = _Driver(_wire())
+        driver.get_listening_sockets()
+
+        [(command, _privileged)] = driver.calls
+        assert command.startswith("sh -c '")
+        assert subprocess.run(["sh", "-n", "-c", command]).returncode == 0
+
+    def test_without_root_it_reads_what_it_can(self):
+        """Without sudo, ss names only the user's own processes: the sockets
+        are still worth having, marked as not attributed."""
+        driver = _Driver("sudo: a password is required\n", _wire(cgroups=""))
+
+        reading = driver.get_listening_sockets()
+
+        assert reading["attributed"] is False
+        assert len(reading["sockets"]) == 9
+        assert [p for _c, p in driver.calls] == [True, False]
+
+    def test_a_host_without_ss_is_not_asked_twice(self):
+        driver = _Driver("SOCK_BEGIN\n[no-ss]\nSOCK_END\n")
+
+        with pytest.raises(ListeningSocketsUnavailable):
+            driver.get_listening_sockets()
+        assert len(driver.calls) == 1
+
+    def test_not_every_os_driver_has_it(self):
+        """A Windows host is an OSDriver too and has no ss: ``hasattr`` has to
+        stay a truthful answer, so the drivers that can mix this in themselves."""
+        assert not issubclass(OSDriver, ListeningSocketsMixin)
+        assert not hasattr(OSDriver, "get_listening_sockets")

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

tests/test_systemd_services.py

@@ -0,0 +1,412 @@
+"""systemd services: listing them in one round trip, and starting and stopping them.
+
+What systemd reports, and how a unit is started or stopped, is the same on
+every host that runs it -- so the command, its parse, the name check and the
+reading of the exit status live here once, and a driver only carries a command
+across (napalm-linux#7, napalm-proxmox#6).
+"""
+
+from __future__ import annotations
+
+import os
+import subprocess
+
+import pytest
+
+from napalm_device_types import OSDriver
+from napalm_device_types.systemd import (
+    ACTION_TIMEOUT,
+    SERVICE_ACTIONS,
+    SYSTEMD_SERVICES_COMMAND,
+    SystemdServicesMixin,
+    SystemdUnavailable,
+    parse_action_result,
+    parse_systemd_services,
+    service_action_command,
+    unit_name,
+)
+
+FILES = """\
+apparmor.service                             enabled         enabled
+ssh.service                                  enabled         enabled
+sshd.service                                 alias           -
+getty@.service                               enabled         enabled
+rsync.service                                disabled        enabled
+cups.service                                 indirect        enabled
+plymouth-quit.service                        static          -
+systemd-networkd-wait-online.service         enabled-runtime enabled
+nfs-server.service                           masked          enabled
+"""
+
+UNITS = """\
+MainPID=812
+Id=ssh.service
+Names=ssh.service sshd.service
+LoadState=loaded
+ActiveState=active
+SubState=running
+UnitFileState=enabled
+
+MainPID=0
+Id=apparmor.service
+Names=apparmor.service
+LoadState=loaded
+ActiveState=active
+SubState=exited
+UnitFileState=enabled
+
+MainPID=900
+Id=getty@tty1.service
+Names=getty@tty1.service
+LoadState=loaded
+ActiveState=active
+SubState=running
+UnitFileState=enabled
+
+MainPID=0
+Id=systemd-fsck@dev-disk-by\\x2dlabel-BOOT.service
+Names=systemd-fsck@dev-disk-by\\x2dlabel-BOOT.service
+LoadState=loaded
+ActiveState=inactive
+SubState=dead
+UnitFileState=static
+
+MainPID=0
+Id=display-manager.service
+Names=display-manager.service
+LoadState=not-found
+ActiveState=inactive
+SubState=dead
+UnitFileState=
+
+MainPID=0
+Id=nfs-server.service
+Names=nfs-server.service
+LoadState=masked
+ActiveState=inactive
+SubState=dead
+UnitFileState=masked
+
+MainPID=0
+Id=systemd-networkd-wait-online.service
+Names=systemd-networkd-wait-online.service
+LoadState=loaded
+ActiveState=active
+SubState=exited
+UnitFileState=enabled-runtime
+"""
+
+
+GENERATED_UNIT = """
+MainPID=0
+Id=rrdcached.service
+Names=rrdcached.service
+LoadState=loaded
+ActiveState=active
+SubState=running
+UnitFileState=generated
+"""
+
+
+def _wire(
+    files: str = FILES,
+    units: str = UNITS,
+    *,
+    generated: str = "",
+    noise: str = "",
+    end: bool = True,
+) -> str:
+    """The report as the command prints it, framed."""
+    tail = "SVC_END\n" if end else ""
+    return f"{noise}SVC_BEGIN\n[files]\n{files}[units]\n{units}[generated]\n{generated}{tail}"
+
+
+def _by_name(services):
+    return {s["name"]: s for s in services}
+
+
+class TestParseSystemdServices:
+    def test_a_loaded_unit_is_read_with_its_state(self):
+        services = _by_name(parse_systemd_services(_wire()))
+
+        assert services["ssh"] == {"name": "ssh", "running": True, "enabled": True, "pid": 812}
+        assert services["apparmor"] == {
+            "name": "apparmor",
+            "running": False,
+            "enabled": True,
+            "pid": 0,
+        }
+
+    def test_enabled_means_enabled_now_not_merely_installed(self):
+        services = _by_name(parse_systemd_services(_wire()))
+
+        assert services["systemd-networkd-wait-online"]["enabled"] is True
+        assert services[r"systemd-fsck@dev-disk-by\x2dlabel-BOOT"]["enabled"] is False
+        assert services["nfs-server"]["enabled"] is False
+
+    def test_a_generated_unit_takes_its_boot_state_from_is_enabled(self):
+        """A SysV script's unit is generated; only is-enabled knows its rc links."""
+        raw = _wire(units=UNITS + GENERATED_UNIT, generated="rrdcached.service enabled\n")
+
+        assert _by_name(parse_systemd_services(raw))["rrdcached"]["enabled"] is True
+
+    def test_a_generated_unit_is_not_enabled_unless_is_enabled_says_so(self):
+        raw = _wire(units=UNITS + GENERATED_UNIT, generated="rrdcached.service disabled\n")
+
+        assert _by_name(parse_systemd_services(raw))["rrdcached"]["enabled"] is False
+
+    def test_a_unit_that_is_not_there_is_left_out(self):
+        assert "display-manager" not in _by_name(parse_systemd_services(_wire()))
+
+    def test_an_installed_unit_that_is_not_loaded_is_listed(self):
+        services = _by_name(parse_systemd_services(_wire()))
+
+        assert services["rsync"] == {"name": "rsync", "running": False, "enabled": False, "pid": 0}
+        assert services["cups"]["enabled"] is False
+
+    def test_templates_and_static_files_that_are_not_loaded_are_not(self):
+        services = _by_name(parse_systemd_services(_wire()))
+
+        assert "getty@" not in services
+        assert "plymouth-quit" not in services
+        assert services["getty@tty1"]["running"] is True
+
+    def test_an_alias_never_appears_beside_its_unit(self):
+        assert "sshd" not in _by_name(parse_systemd_services(_wire()))
+
+    def test_an_alias_that_older_systemd_calls_enabled_does_not_either(self):
+        files = (
+            "\n".join(
+                "sshd.service enabled enabled" if line.startswith("sshd.service") else line
+                for line in FILES.splitlines()
+            )
+            + "\n"
+        )
+
+        assert "sshd" not in _by_name(parse_systemd_services(_wire(files=files)))
+
+    def test_an_escaped_name_survives(self):
+        assert r"systemd-fsck@dev-disk-by\x2dlabel-BOOT" in _by_name(
+            parse_systemd_services(_wire())
+        )
+
+    def test_blocks_run_together_are_still_told_apart(self):
+        """xargs may split the unit list across two systemctl runs."""
+        units = UNITS.replace(
+            "UnitFileState=enabled\n\nMainPID=0\nId=apparmor",
+            "UnitFileState=enabled\nMainPID=0\nId=apparmor",
+        )
+
+        services = _by_name(parse_systemd_services(_wire(units=units)))
+
+        assert services["ssh"]["pid"] == 812
+        assert services["apparmor"]["running"] is False
+
+    def test_the_list_is_sorted_by_name(self):
+        names = [s["name"] for s in parse_systemd_services(_wire())]
+
+        assert names == sorted(names)
+
+    def test_terminal_colours_in_the_report_are_dropped(self):
+        files = FILES.replace(
+            "rsync.service                                disabled",
+            "rsync.service \x1b[0;1;31mdisabled\x1b[0m",
+        )
+
+        assert "rsync" in _by_name(parse_systemd_services(_wire(files=files)))
+
+    def test_whatever_surrounds_the_frame_is_ignored(self):
+        noisy = _wire(noise="user@host:~$ systemctl ...\n") + "user@host:~$ "
+
+        assert "ssh" in _by_name(parse_systemd_services(noisy))
+
+    def test_a_cut_short_report_raises(self):
+        """A missing tail must not read as services that went away."""
+        with pytest.raises(ValueError):
+            parse_systemd_services(_wire(end=False))
+
+    def test_output_without_the_frame_raises(self):
+        with pytest.raises(ValueError):
+            parse_systemd_services("bash: systemctl: command not found\n")
+
+    def test_a_host_without_systemd_says_so(self):
+        raw = "SVC_BEGIN\n[no-systemd]\n[files]\n[units]\nSVC_END\n"
+
+        with pytest.raises(SystemdUnavailable):
+            parse_systemd_services(raw)
+
+    def test_no_systemd_is_a_not_implemented_error(self):
+        assert issubclass(SystemdUnavailable, NotImplementedError)
+
+
+class TestTheCommand:
+    def test_the_frame_is_not_in_the_command_itself(self):
+        """An echoing transport must not show the end marker early."""
+        assert "SVC_END" not in SYSTEMD_SERVICES_COMMAND
+        assert "SVC_BEGIN" not in SYSTEMD_SERVICES_COMMAND
+
+    def test_it_changes_nothing(self):
+        for verb in ("start", "stop", "restart", "enable", "disable", "mask"):
+            assert f"systemctl {verb}" not in SYSTEMD_SERVICES_COMMAND
+
+    @pytest.mark.skipif(not os.path.isdir("/run/systemd/system"), reason="needs systemd")
+    def test_it_runs_and_parses_on_this_host(self):
+        out = subprocess.run(
+            ["sh", "-c", SYSTEMD_SERVICES_COMMAND], capture_output=True, text=True, timeout=60
+        ).stdout
+
+        services = _by_name(parse_systemd_services(out))
+
+        assert "systemd-journald" in services
+        assert services["systemd-journald"]["running"] is True
+
+
+class TestUnitName:
+    @pytest.mark.parametrize(
+        ("raw", "name"),
+        [
+            ("ssh", "ssh"),
+            ("ssh.service", "ssh"),
+            ("getty@tty1", "getty@tty1"),
+            ("wg-quick@wg0", "wg-quick@wg0"),
+            ("snapd.apparmor", "snapd.apparmor"),
+            ("systemd-backlight@backlight:acpi_video0", "systemd-backlight@backlight:acpi_video0"),
+            (r"systemd-fsck@dev-disk-by\x2dlabel-BOOT", r"systemd-fsck@dev-disk-by\x2dlabel-BOOT"),
+        ],
+    )
+    def test_a_unit_name_is_accepted(self, raw, name):
+        assert unit_name(raw) == name
+
+    @pytest.mark.parametrize(
+        "raw",
+        [
+            "",
+            "-x",
+            "foo@",
+            "foo@.service",
+            "a b",
+            "a;b",
+            "$(id)",
+            "a/b",
+            r"bad\x2",
+            "ssh\n",
+            "x" * 256,
+        ],
+    )
+    def test_anything_else_is_refused(self, raw):
+        with pytest.raises(ValueError):
+            unit_name(raw)
+
+
+class TestServiceActionCommand:
+    def test_the_command_is_bounded_and_never_asks(self):
+        cmd = service_action_command("getty@tty1", "restart")
+
+        assert cmd.startswith(f"timeout {ACTION_TIMEOUT} systemctl --no-ask-password restart -- ")
+        assert "getty@tty1.service" in cmd
+
+    def test_an_escaped_name_is_quoted_for_the_shell(self):
+        cmd = service_action_command(r"systemd-fsck@dev-disk-by\x2dlabel-BOOT", "stop")
+
+        assert r"'systemd-fsck@dev-disk-by\x2dlabel-BOOT.service'" in cmd
+
+    def test_its_exit_status_is_printed_after_it(self):
+        assert service_action_command("ssh", "start").endswith("; echo __SVC_RC=$?")
+
+    def test_the_actions(self):
+        assert SERVICE_ACTIONS == ("start", "stop", "restart", "enable", "disable")
+
+    def test_an_unknown_action_is_refused(self):
+        with pytest.raises(ValueError):
+            service_action_command("ssh", "mask")
+
+    def test_an_invalid_name_is_refused(self):
+        with pytest.raises(ValueError):
+            service_action_command("ssh; reboot", "stop")
+
+
+class TestParseActionResult:
+    def test_exit_status_zero_is_success(self):
+        assert parse_action_result("__SVC_RC=0\n") == {"success": True, "output": ""}
+
+    def test_what_systemctl_printed_comes_back_without_the_marker(self):
+        raw = (
+            "Created symlink /etc/systemd/system/multi-user.target.wants/cron.service.\n__SVC_RC=0"
+        )
+
+        result = parse_action_result(raw)
+
+        assert result["success"] is True
+        assert result["output"].startswith("Created symlink")
+        assert "__SVC_RC" not in result["output"]
+
+    def test_terminal_colours_are_dropped(self):
+        """systemctl colours its errors when a transport gives it a terminal."""
+        raw = (
+            "\x1b[0;1;31mFailed to restart x.service: Unit x.service not found.\x1b[0m\n"
+            "__SVC_RC=5\n"
+        )
+
+        assert parse_action_result(raw)["output"] == (
+            "Failed to restart x.service: Unit x.service not found."
+        )
+
+    def test_a_failure_keeps_its_message(self):
+        raw = "Failed to start foo.service: Unit foo.service not found.\n__SVC_RC=5\n"
+
+        assert parse_action_result(raw) == {
+            "success": False,
+            "output": "Failed to start foo.service: Unit foo.service not found.",
+        }
+
+    def test_a_job_still_running_at_the_timeout_is_not_called_done(self):
+        result = parse_action_result("__SVC_RC=124\n")
+
+        assert result["success"] is False
+        assert str(ACTION_TIMEOUT) in result["output"]
+
+    def test_no_exit_status_is_no_success(self):
+        assert parse_action_result("Connection reset\n")["success"] is False
+
+    def test_the_echoed_command_is_not_taken_for_the_status(self):
+        raw = "timeout 45 systemctl restart -- cron.service 2>&1; echo __SVC_RC=$?\n__SVC_RC=1\n"
+
+        assert parse_action_result(raw)["success"] is False
+
+
+class _Driver(SystemdServicesMixin):
+    def __init__(self, reply: str) -> None:
+        self.reply = reply
+        self.calls: list = []
+
+    def _run_service_command(self, command: str, *, privileged: bool, timeout: int) -> str:
+        self.calls.append((command, privileged, timeout))
+        return self.reply
+
+
+class TestSystemdServicesMixin:
+    def test_listing_runs_the_command_unprivileged(self):
+        driver = _Driver(_wire())
+
+        assert "ssh" in _by_name(driver.get_services())
+        assert driver.calls == [(SYSTEMD_SERVICES_COMMAND, False, ACTION_TIMEOUT + 15)]
+
+    def test_an_action_runs_privileged_and_reports_its_outcome(self):
+        driver = _Driver("__SVC_RC=0\n")
+
+        assert driver.manage_service("cron", "restart") == {"success": True, "output": ""}
+        command, privileged, timeout = driver.calls[0]
+        assert command == service_action_command("cron", "restart")
+        assert privileged is True
+        assert timeout > ACTION_TIMEOUT
+
+    def test_an_invalid_request_is_refused_before_anything_is_sent(self):
+        driver = _Driver("__SVC_RC=0\n")
+
+        with pytest.raises(ValueError):
+            driver.manage_service("cron;reboot", "stop")
+        assert driver.calls == []
+
+    def test_not_every_os_driver_has_it(self):
+        assert not hasattr(OSDriver, "manage_service")
+        assert callable(getattr(SystemdServicesMixin, "manage_service"))