feat: report where an update comes from and whether it is a security fix, refresh the index, read the host status

For netOrk MVP 5, on napalm-device-types 2.3.0:

- get_available_updates (apt) runs the shared APT_UPGRADABLE_COMMAND and
  parse_apt_upgradable: origin and security from apt's suites, and a
  ValueError instead of [] when apt failed or its output was cut short.
- dnf/yum: check-update's exit status decides (0 none, 100 updates, anything
  else raises); security comes from `updateinfo list --security`, and is None
  when dnf cannot say. The repository column becomes the origin.
- refresh_available_updates(): apt-get update, dnf/yum makecache, apk update;
  pacman is left out (-Sy without -u invites a partial upgrade).
- HostStatusMixin: reboot required and self-patching, read over SSH.
- _run_privileged(): root runs directly, a sudo password goes through _sudo,
  otherwise sudo -n. Shared by service control and the refresh.
- _split_status() drops terminal codes before it looks for the exit status;
  a pseudo-terminal left keypad codes in front of the marker.

OpenMediaVault inherits all of it.

README.md

@@ -48,7 +48,8 @@ with Driver(
     optional_args={
         # "port": 22,
         # "pkg_manager": "apt",   # force package manager; auto-detected by default
-        # "secret": "sudo-pass",  # password for sudo / enable (defaults to login password)
+        # "sudo_password": "sudo-pass",  # for commands that need root; without it,
+        #                                # `sudo -n` (passwordless sudo) is tried
         # "debugging": True,      # enable verbose logging
     },
 ) as dev:
@@ -69,6 +70,10 @@ with Driver(
 
     # Upgrade everything with pending updates
     result = dev.apply_updates([])
+
+    # Restart a service (start, stop, restart, enable, disable)
+    result = dev.manage_service("cron", "restart")
+    print(result)  # {"success": True, "output": ""}
 ```
 
 ## Supported NAPALM methods
@@ -99,7 +104,8 @@ with Driver(
 | `get_packages()` | ✅ | apt, dnf, yum, apk, pacman |
 | `get_pending_updates()` | ✅ | apt, dnf, yum, apk, pacman |
 | `apply_updates(packages)` | ✅ | apt, dnf, yum, apk, pacman |
-| `get_services()` | ✅ | systemd (fallback: SysV `service`) |
+| `get_services()` | ✅ | systemd, one round trip (fallback: SysV `service`) |
+| `manage_service(name, action)` | ✅ | systemd: start, stop, restart, enable, disable |
 | `get_users()` | ✅ | `/etc/passwd` + `/etc/group` |
 | `get_processes()` | ✅ | `ps axo` |
 | `get_cron_jobs()` | ✅ | user crontabs + `/etc/cron.d/` |
@@ -127,13 +133,25 @@ The SSH user needs read access to:
 | `/etc/passwd`, `/etc/group` | world-readable (default) |
 | `/proc/uptime`, `/sys/class/dmi/…` | world-readable (default) |
 | User crontabs (`/var/spool/cron/…`) | `root` or `sudo` required |
-| `systemctl is-enabled <unit>` | unprivileged on most distros |
+| `systemctl list-unit-files`, `systemctl show` | unprivileged |
+| `systemctl start/stop/restart/enable/disable` | `root`, or `sudo` (with `sudo_password`, or passwordless) |
 | `apt list --upgradable` | may require `apt-get update` (root) |
 | `dnf check-update` / `yum check-update` | unprivileged, but slower without cache |
 
 For full functionality it is recommended to run as `root` or grant passwordless `sudo` for
 the above commands.
 
+`get_services()` and `manage_service()` come from napalm-device-types'
+`SystemdServicesMixin`; this driver supplies only the transport. An action runs as
+`timeout 45 systemctl --no-ask-password <action> -- <unit>.service`, so a unit that hangs
+on its way up or down cannot hold the session, and only the exit status decides whether it
+succeeded. Without a sudo password it uses `sudo -n`, which fails at once instead of
+waiting for a password prompt.
+
+On OpenMediaVault (napalm-openmediavault inherits this driver), enabling or disabling a
+unit that OMV manages itself — Samba, NFS, SSH — may be reverted the next time OMV applies
+its configuration.
+
 ## Tested distributions
 
 | Distribution | Version | Package manager | Tested |

napalm_linux/linux.py

@@ -31,7 +31,19 @@ from netmiko.exceptions import (
 )
 from napalm.base.exceptions import ConnectionException, ConnectionClosedException
 from napalm.base.netmiko_helpers import netmiko_args
-from napalm_device_types import FingerprintRule, OSDriver
+from napalm_device_types import (
+    APT_UPGRADABLE_COMMAND,
+    DNF_SECURITY_COMMAND,
+    FingerprintRule,
+    HostStatusMixin,
+    KernelFactsMixin,
+    OSDriver,
+    SystemdServicesMixin,
+    SystemdUnavailable,
+    parse_apt_upgradable,
+    parse_dnf_security,
+    strip_terminal_codes,
+)
 from napalm_device_types.models import (
     ApplyUpdatesResultDict,
     CronJobDict,
@@ -50,6 +62,46 @@ logger = logging.getLogger("napalm_linux")
 # Package managers in detection order
 _PKG_MANAGERS = ["apt", "dnf", "yum", "apk", "pacman"]
 
+#: Printed after a command by ``_sudo_status`` so its exit status survives the
+#: trip through an interactive shell. Matched only on a line of its own with a
+#: number after it — an echoed command line carries the literal ``$?`` instead.
+_RC_MARKER = "__NETORK_RC="
+_RC_MARKER_RE = re.compile(rf"^{_RC_MARKER}(\d+)\s*$", re.MULTILINE)
+
+
+def _split_status(raw: str) -> tuple[str, int | None]:
+    """``(output, exit_status)`` of a command followed by ``echo {_RC_MARKER}$?``.
+
+    The status is ``None`` when the marker never arrived (output cut short), so
+    a caller can tell "unknown" from "succeeded".
+    """
+    raw = strip_terminal_codes(raw)
+    matches = list(_RC_MARKER_RE.finditer(raw))
+    if not matches:
+        return raw, None
+    last = matches[-1]
+    return (raw[: last.start()] + raw[last.end():]).strip(), int(last.group(1))
+
+
+#: How each package manager refreshes its index. pacman is left out on purpose:
+#: ``pacman -Sy`` without ``-u`` invites a partial upgrade on the next install.
+_REFRESH = {
+    # No LC_ALL=C here: under sudo it is an environment variable sudoers may refuse
+    # to set. Only the exit status decides, so the language is merely what is shown.
+    "apt": "apt-get update -q 2>&1",
+    "dnf": "dnf makecache -q 2>&1",
+    "yum": "yum makecache -q 2>&1",
+    "apk": "apk update -q 2>&1",
+}
+_YUM_SECURITY_COMMAND = "LC_ALL=C yum updateinfo list security -q 2>/dev/null"
+
+#: What to do when sudo wants a password netOrk does not have.
+_SUDO_PASSWORD_HINT = (
+    "sudo requires a password on this device but none is configured in netOrk. "
+    "Please add the sudo password to a Credential Profile assigned to this device, "
+    "or configure passwordless sudo (NOPASSWD) for this user."
+)
+
 # DMI field values that carry no useful information (OEM defaults, blanks)
 _BAD_DMI: frozenset[str] = frozenset({
     "", "none", "n/a", "not specified", "not applicable",
@@ -132,7 +184,7 @@ def _short_image_id(raw: str) -> str:
     return raw.strip().removeprefix("sha256:")[:12]
 
 
-class LinuxDriver(OSDriver):
+class LinuxDriver(KernelFactsMixin, SystemdServicesMixin, HostStatusMixin, OSDriver):
     """NAPALM driver for generic Linux systems.
 
     Connects via SSH (netmiko ``linux`` device type) and auto-detects the
@@ -267,6 +319,55 @@ class LinuxDriver(OSDriver):
             return self._send(wrapped, read_timeout=read_timeout)
         return self._send(f'sudo {command}', read_timeout=read_timeout)
 
+    def _sudo_status(self, command: str, read_timeout: float = 100) -> tuple[str, int | None]:
+        """Run *command* via sudo and return ``(output, exit_status)``.
+
+        ``_sudo`` callers append ``|| true`` so a failing command yields output
+        instead of an error, which throws the exit status away. This variant
+        echoes ``$?`` straight after the sudo pipeline instead — sudo passes
+        the command's status through, and a failed password is non-zero too.
+
+        The status is ``None`` when the marker never arrived (output cut short),
+        so a caller can tell "unknown" from "succeeded".
+        """
+        return _split_status(
+            self._sudo(f"{command}; echo {_RC_MARKER}$?", read_timeout=read_timeout)
+        )
+
+    def _is_root(self) -> bool:
+        """Whether the SSH user is root, asked once per session.
+
+        A root login on a box without sudo (an LXC container, a minimal Debian)
+        must not have its commands prefixed with a sudo that is not there.
+        """
+        if getattr(self, "_root", None) is None:
+            self._root = self._send("id -u") == "0"
+        return bool(self._root)
+
+    def _run_service_command(self, command: str, *, privileged: bool, timeout: int) -> str:
+        """The transport for :class:`SystemdServicesMixin`.
+
+        Without a sudo password, ``sudo -n`` fails at once where a prompt would
+        otherwise hang the session until the read timeout.
+        """
+        if not privileged:
+            return self._send(command, read_timeout=timeout)
+        return self._run_privileged(command, timeout)
+
+    def _run_privileged(self, command: str, timeout: float = 100) -> str:
+        """Run *command* as root: directly for a root login, through ``_sudo``
+        with a sudo password, and through ``sudo -n`` without one -- which fails at
+        once where a password prompt would hang the session until the timeout."""
+        if self._is_root():
+            return self._send(command, read_timeout=timeout)
+        if self._sudo_password:
+            return self._sudo(command, read_timeout=timeout)
+        return self._send(f"sudo -n {command}", read_timeout=timeout)
+
+    def _run_host_status_command(self, command: str) -> str:
+        """The transport for ``HostStatusMixin.get_host_status``: read-only, no sudo."""
+        return self._send(command, read_timeout=60)
+
     def _detect_pkg_manager(self) -> str | None:
         """Return the first package manager binary found on PATH."""
         for pm in _PKG_MANAGERS:
@@ -815,6 +916,14 @@ class LinuxDriver(OSDriver):
             }
         }
 
+    # ------------------------------------------------------------------
+    # KernelFactsMixin – the transport for get_kernel_facts
+    # ------------------------------------------------------------------
+
+    def _run_kernel_facts_command(self, command: str) -> str:
+        """The transport for ``KernelFactsMixin.get_kernel_facts``: read-only, no sudo."""
+        return self._send(command, read_timeout=60)
+
     # ------------------------------------------------------------------
     # OSDriver – package management
     # ------------------------------------------------------------------
@@ -835,11 +944,13 @@ class LinuxDriver(OSDriver):
     def _get_packages_apt(self) -> list[PackageDict]:
         out = self._send(
             "dpkg-query -W -f='${Package}\\t${Version}\\t${Installed-Size}"
-            "\\t${source:Package}\\t${binary:Summary}\\n' 2>/dev/null"
+            "\\t${source:Package}\\t${source:Version}\\t${binary:Summary}\\n' 2>/dev/null"
         )
         packages: list[PackageDict] = []
         for line in out.splitlines():
-            parts = line.split("\t", 4)
+            # Summary stays last and keeps whatever it contains: maxsplit must
+            # equal the number of tabs the format writes, not the field count.
+            parts = line.split("\t", 5)
             if len(parts) < 2:
                 continue
             name = parts[0].strip()
@@ -847,7 +958,21 @@ class LinuxDriver(OSDriver):
             size = int(parts[2].strip()) * 1024 if len(parts) > 2 and parts[2].strip().isdigit() else 0
             # Debian source package (e.g. openssh-server → openssh) for OSV matching.
             source_package = parts[3].strip() if len(parts) > 3 and parts[3].strip() else name
-            description = parts[4].strip() if len(parts) > 4 else ""
+            # And its version, which is a different number from this package's.
+            #
+            # OSV states Debian ranges in *source* versions. A source package
+            # that ships several binaries gives each its own upstream version:
+            # libldb2 is 2:2.11.0+samba4.22.11+dfsg-… while its source, samba,
+            # is 2:4.22.11+dfsg-…. A consumer matching on source_package and
+            # comparing `version` compares two unrelated numbers — dpkg reads
+            # ldb's 2.11.0 as older than the 2:4.17.4+dfsg-1 that fixed
+            # CVE-2022-44640, and a host five releases past the fix was reported
+            # vulnerable on four packages at once.
+            #
+            # dpkg leaves this empty when it equals `Version`; so does an older
+            # dpkg that does not know the field at all.
+            source_version = parts[4].strip() if len(parts) > 4 and parts[4].strip() else version
+            description = parts[5].strip() if len(parts) > 5 else ""
             packages.append({
                 "name": name,
                 "version": version,
@@ -856,6 +981,7 @@ class LinuxDriver(OSDriver):
                 "size": size,
                 "source": "apt",
                 "source_package": source_package,
+                "source_version": source_version,
             })
         return packages
 
@@ -1028,24 +1154,86 @@ class LinuxDriver(OSDriver):
         success = not any(kw in low for kw in ("error:", "failed", "no packages", "not found", "unable to locate", "no match"))
         return {"success": success, "output": raw.strip()}
 
-    def uninstall_package(self, name: str) -> dict[str, Any]:
+    #: Words in a package manager's output that mean it did not do the job.
-        """Remove a package by name. Returns ``{"success": bool, "output": str}``."""
+    #: Only consulted when the exit status is unknown; see ``_uninstall_failed``.
+    _UNINSTALL_FAILED = ("error:", "failed", "not found", "is not installed", "no packages")
+
+    def uninstall_package(self, name: str, purge: bool = False) -> dict[str, Any]:
+        """Remove a package by name. Returns ``{"success": bool, "output": str}``.
+
+        ``purge`` also removes the package's configuration where the package
+        manager distinguishes the two. Off by default: configuration somebody
+        may want back is not this function's to delete unless it was asked for.
+
+        It matters for more than tidiness. A package's apt source survives a
+        plain ``remove``, so the repository keeps being fetched on every
+        ``apt-get update`` long after the package itself is gone — which is what
+        the Wazuh agent left behind on thirteen hosts.
+
+        **The dpkg fallback.** A package whose ``postinst`` failed sits at
+        ``install ok unpacked``, and apt cannot remove it: it configures a
+        package before removing it, and configuring is precisely what is broken.
+        Seven of those thirteen hosts were in that state after an upgrade whose
+        postinst could not reach a manager that had been decommissioned, and on
+        one of them only ``dpkg --purge --force-all`` got it out.
+
+        So the fallback runs **only after apt has failed**, never as a routine
+        second step: forcing dpkg past its own consistency checks is a bigger
+        hammer than apt, and a caller who reaches for it every time will
+        eventually break something apt would have refused to.
+        """
         from shlex import quote as _q
         safe = _q(name)
         pm = self._pkg_manager
         if pm == "apt":
-            raw = self._sudo(f"DEBIAN_FRONTEND=noninteractive apt-get remove -y {safe} 2>&1 || true")
+            action = "purge" if purge else "remove"
+            cmd = f"DEBIAN_FRONTEND=noninteractive apt-get {action} -y {safe} 2>&1"
         elif pm in ("dnf", "yum"):
-            raw = self._sudo(f"{pm} remove -y {safe} 2>&1 || true")
+            cmd = f"{pm} remove -y {safe} 2>&1"
         elif pm == "apk":
-            raw = self._sudo(f"apk del {safe} 2>&1 || true")
+            # apk and pacman have no separate purge; asking for one is not an
+            # error, it simply has nothing extra to do.
+            cmd = f"apk del {safe} 2>&1"
         elif pm == "pacman":
-            raw = self._sudo(f"pacman -R --noconfirm {safe} 2>&1 || true")
+            cmd = f"pacman -R --noconfirm {safe} 2>&1"
         else:
             return {"success": False, "output": f"Unsupported package manager: {pm}"}
-        low = raw.lower()
+
-        success = not any(kw in low for kw in ("error:", "failed", "not found", "is not installed", "no packages"))
+        raw, rc = self._sudo_status(cmd)
-        return {"success": success, "output": raw.strip()}
+        failed = self._uninstall_failed(raw, rc)
+
+        if failed and pm == "apt":
+            forced, forced_rc = self._sudo_status(f"dpkg --purge --force-all {safe} 2>&1")
+            raw = f"{raw.strip()}\n--- dpkg --purge --force-all ---\n{forced.strip()}"
+            failed = self._uninstall_failed(forced, forced_rc)
+
+        return {"success": not failed, "output": raw.strip()}
+
+    def _uninstall_failed(self, output: str, rc: int | None = None) -> bool:
+        """Whether the package manager did not do the job.
+
+        The exit status decides whenever there is one (netork#267): it is the
+        answer the package manager actually gives, where the output is prose
+        that every tool phrases differently. A prerm printing "Failed to stop
+        …" while the removal completes is a success; a non-zero exit with
+        nothing alarming in the output is not.
+
+        Only when the status is unknown (``rc is None``) is the output read,
+        as the best answer left. apt prefixes its own errors with ``E: `` at
+        the start of a line, and the commonest of them — ``E: Sub-process
+        /usr/bin/dpkg returned an error code (1)`` — contains neither "error:"
+        nor "failed". The keyword list alone therefore read a failed removal
+        as a success, which is the worst direction for this particular answer
+        to be wrong in.
+
+        Matched at line start rather than anywhere: "note: " ends in "e: ".
+        """
+        if rc is not None:
+            return rc != 0
+        low = output.lower()
+        if any(line.lstrip().startswith("e: ") for line in low.splitlines()):
+            return True
+        return any(kw in low for kw in self._UNINSTALL_FAILED)
 
     def get_available_updates(self) -> list[UpdateDict]:
         if self._pkg_manager == "apt":
@@ -1096,48 +1284,49 @@ class LinuxDriver(OSDriver):
     def _get_updates_apt(self) -> list[UpdateDict]:
         # apt list --upgradable does not need root; avoid sudo so it works even
         # without a configured sudo password.
-        out = self._send(
+        # Raises ValueError when apt failed or the output was cut short.
-            "LC_ALL=C apt list --upgradable 2>/dev/null | grep -v '^Listing'",
+        return parse_apt_upgradable(self._send(APT_UPGRADABLE_COMMAND, read_timeout=60))
-            read_timeout=60,
-        )
-        # Join wrapped lines: netmiko's 80-col pseudo-TTY causes long apt lines to
-        # break; continuation lines start with a space.
-        raw_lines: List[str] = []
-        for line in out.splitlines():
-            if line.startswith(" ") and raw_lines:
-                raw_lines[-1] += line.strip()
-            else:
-                raw_lines.append(line)
-        updates: list[UpdateDict] = []
-        for line in raw_lines:
-            # openssh-server/stable 1:9.2p1-2+deb12u2 amd64 [upgradable from: 1:9.2p1-2+deb12u1]
-            m = re.match(
-                r"^(\S+)/\S+\s+(\S+)\s+\S+\s+\[upgradable from:\s+(\S+)\]", line
-            )
-            if m:
-                updates.append({
-                    "name": m.group(1),
-                    "current_version": m.group(3),
-                    "new_version": m.group(2),
-                })
-        return updates
 
     def _get_updates_rpm(self) -> list[UpdateDict]:
+        """dnf/yum check-update: exit 100 means updates, 0 none, anything else failed."""
         cmd = "dnf check-update --quiet 2>/dev/null" if self._pkg_manager == "dnf" else "yum check-update -q 2>/dev/null"
-        out = self._sudo(cmd)
+        output, status = _split_status(self._run_privileged(f"{cmd}; echo {_RC_MARKER}$?", 120))
+        if status not in (0, 100):
+            raise RuntimeError(f"{self._pkg_manager} check-update failed (exit {status}): {output[-200:]}")
+        security = self._rpm_security_names()
         updates: list[UpdateDict] = []
-        for line in out.splitlines():
+        for line in output.splitlines():
             parts = line.split()
             if len(parts) >= 2 and not line.startswith(" ") and "." in parts[0]:
-                name_arch = parts[0]
+                name = parts[0].rsplit(".", 1)[0]
-                name = name_arch.rsplit(".", 1)[0] if "." in name_arch else name_arch
                 updates.append({
                     "name": name,
                     "current_version": "",
                     "new_version": parts[1],
+                    "origin": parts[2] if len(parts) >= 3 else None,
+                    "security": None if security is None else name in security,
                 })
         return updates
 
+    def _rpm_security_names(self) -> set[str] | None:
+        """Packages a pending security advisory covers; None when dnf/yum cannot say."""
+        cmd = DNF_SECURITY_COMMAND if self._pkg_manager == "dnf" else _YUM_SECURITY_COMMAND
+        output, status = _split_status(self._run_privileged(f"{cmd}; echo {_RC_MARKER}$?", 120))
+        return parse_dnf_security(output) if status == 0 else None
+
+    def refresh_available_updates(self) -> dict[str, Any]:
+        """Refresh the package index (apt-get update, dnf makecache, apk update)."""
+        cmd = _REFRESH.get(self._pkg_manager or "")
+        if cmd is None:
+            return {
+                "success": False,
+                "output": f"Refreshing the index is not supported for {self._pkg_manager!r}",
+            }
+        output, status = _split_status(self._run_privileged(f"{cmd}; echo {_RC_MARKER}$?", 180))
+        if status != 0 and "password is required" in output:
+            output = f"{output}\n{_SUDO_PASSWORD_HINT}"
+        return {"success": status == 0, "output": output}
+
     def _get_updates_apk(self) -> list[UpdateDict]:
         out = self._send("apk version -l '<' 2>/dev/null")
         updates: list[UpdateDict] = []
@@ -1267,50 +1456,25 @@ class LinuxDriver(OSDriver):
             return {"success": False, "output": "", "error": str(exc)}
 
     # ------------------------------------------------------------------
-    # OSDriver – services (systemd)
+    # OSDriver – services (systemd, through SystemdServicesMixin)
     # ------------------------------------------------------------------
 
     def get_services(self) -> list[ServiceDict]:
-        """Return systemd service units (falls back to service --status-all on SysV)."""
+        """systemd's services in one round trip; ``service --status-all`` without systemd."""
-        out = self._send(
+        try:
-            "systemctl list-units --type=service --all --no-legend --no-pager "
+            return super().get_services()
-            "--plain 2>/dev/null"
+        except SystemdUnavailable:
-        )
-        if not out:
             return self._get_services_sysv()
 
-        services: list[ServiceDict] = []
+    def manage_service(self, name: str, action: str) -> dict[str, Any]:
-        for line in out.splitlines():
+        """Start, stop, restart, enable or disable a systemd service.
-            # ssh.service loaded active running OpenBSD Secure Shell server
-            parts = line.split(None, 4)
-            if len(parts) < 4:
-                continue
-            unit, load, active, sub = parts[0], parts[1], parts[2], parts[3]
-            name = unit.removesuffix(".service")
-            running = active == "active" and sub == "running"
-            enabled_out = self._send(
-                f"systemctl is-enabled {unit} 2>/dev/null"
-            )
-            enabled = enabled_out.strip() == "enabled"
 
-            # Retrieve main PID for running services
+        :raises ValueError: for an unknown action or an invalid name.
-            pid = 0
+        """
-            if running:
+        result = super().manage_service(name, action)
-                pid_out = self._send(
+        if not result["success"] and "password is required" in result["output"]:
-                    f"systemctl show -p MainPID --value {unit} 2>/dev/null"
+            result["output"] = f"{result['output']}\n{_SUDO_PASSWORD_HINT}"
-                )
+        return result
-                try:
-                    pid = int(pid_out.strip())
-                except ValueError:
-                    pid = 0
-
-            services.append({
-                "name": name,
-                "running": running,
-                "enabled": enabled,
-                "pid": pid,
-            })
-        return services
 
     def _get_services_sysv(self) -> list[ServiceDict]:
         out = self._send("service --status-all 2>/dev/null")
@@ -1998,11 +2162,7 @@ class LinuxDriver(OSDriver):
             if not self._sudo_password:
                 return {
                     "success": False,
-                    "output": (
+                    "output": _SUDO_PASSWORD_HINT,
-                        "sudo requires a password on this device but none is configured in "
-                        "netOrk. Please add the sudo password to a Credential Profile assigned "
-                        "to this device, or configure passwordless sudo (NOPASSWD) for this user."
-                    ),
                 }
 
         lines: list[str] = []
@@ -2028,11 +2188,7 @@ class LinuxDriver(OSDriver):
             if not self._sudo_password:
                 return {
                     "success": False,
-                    "output": (
+                    "output": _SUDO_PASSWORD_HINT,
-                        "sudo requires a password on this device but none is configured in netOrk. "
-                        "Please add the sudo password to a Credential Profile assigned to this device, "
-                        "or configure passwordless sudo (NOPASSWD) for this user."
-                    ),
                 }
 
         # 1. Install snmpd if missing

pyproject.toml

@@ -37,7 +37,7 @@ classifiers = [
 ]
 dependencies = [
     "napalm>=4.0",
-    "napalm-device-types>=0.3.0",
+    "napalm-device-types>=2.3.0",
     "netmiko>=4.0.0",
     "paramiko>=5.0.0",  # CVE-2026-44405
 ]

tests/test_linux.py

@@ -153,9 +153,22 @@ def test_parse_uptime_invalid(driver):
 # ---------------------------------------------------------------------------
 
 
+#: What `dpkg-query` actually returns for the format this driver asks for.
+#:
+#: The previous fixture carried four fields against a format string asking for
+#: five, so `source_package` was silently receiving the description and no
+#: assertion noticed. A fixture simpler than the data cannot fail the way the
+#: data does.
 APT_PKG_OUTPUT = (
-    "openssh-server\t1:9.2p1-2+deb12u2\t512\tsecure shell server\n"
+    "openssh-server\t1:9.2p1-2+deb12u2\t512\topenssh\t1:9.2p1-2+deb12u2"
-    "curl\t7.88.1-10+deb12u5\t1024\tcommand line tool for transferring data\n"
+    "\tsecure shell server\n"
+    "curl\t7.88.1-10+deb12u5\t1024\tcurl\t7.88.1-10+deb12u5"
+    "\tcommand line tool for transferring data\n"
+    # The shape that matters: a binary package whose own upstream version has
+    # nothing to do with its source package's. ldb 2.11.0 is built from samba
+    # 4.22.11, and OSV states Debian ranges in source versions.
+    "libldb2\t2:2.11.0+samba4.22.11+dfsg-0+deb13u1\t2048\tsamba"
+    "\t2:4.22.11+dfsg-0+deb13u1\tLDB shared library\n"
 )
 
 
@@ -163,11 +176,45 @@ def test_get_packages_apt(driver):
     driver._pkg_manager = "apt"
     with patch.object(driver, "_send", return_value=APT_PKG_OUTPUT):
         pkgs = driver.get_packages()
-    assert len(pkgs) == 2
+    assert len(pkgs) == 3
     assert pkgs[0]["name"] == "openssh-server"
     assert pkgs[0]["version"] == "1:9.2p1-2+deb12u2"
     assert pkgs[0]["installed"] is True
     assert pkgs[0]["source"] == "apt"
+    assert pkgs[0]["description"] == "secure shell server"
+
+
+def test_get_packages_apt_keeps_the_source_package_and_its_version(driver):
+    """OSV states Debian ranges in *source* package versions.
+
+    A consumer that matches on the source package and then compares the binary
+    package's version is comparing two unrelated numbers. On a Debian 13 host
+    that reported four Samba libraries as vulnerable to CVE-2022-44640 while
+    running samba 4.22.11 — five releases past the fix — because dpkg reads
+    ldb's own `2.11.0` as older than samba's `2:4.17.4+dfsg-1`.
+
+    The driver cannot fix the comparison, but it is the only place that can
+    supply the number to compare.
+    """
+    driver._pkg_manager = "apt"
+    with patch.object(driver, "_send", return_value=APT_PKG_OUTPUT):
+        pkgs = {p["name"]: p for p in driver.get_packages()}
+
+    assert pkgs["libldb2"]["source_package"] == "samba"
+    assert pkgs["libldb2"]["source_version"] == "2:4.22.11+dfsg-0+deb13u1"
+    assert pkgs["libldb2"]["version"] == "2:2.11.0+samba4.22.11+dfsg-0+deb13u1"
+    assert pkgs["libldb2"]["description"] == "LDB shared library"
+
+
+def test_get_packages_apt_falls_back_when_dpkg_gives_no_source(driver):
+    """`source:Package` is empty for a package whose source name equals its own.
+    Older dpkg builds leave `source:Version` empty in that case too."""
+    driver._pkg_manager = "apt"
+    with patch.object(driver, "_send", return_value="curl\t7.88.1-10\t1024\t\t\ttool\n"):
+        (pkg,) = driver.get_packages()
+
+    assert pkg["source_package"] == "curl"
+    assert pkg["source_version"] == "7.88.1-10"
 
 
 # ---------------------------------------------------------------------------
@@ -179,6 +226,7 @@ APT_UPGRADABLE = (
     "Listing... Done\n"
     "openssh-server/stable 1:9.2p1-2+deb12u2 amd64 [upgradable from: 1:9.2p1-2+deb12u1]\n"
     "curl/stable 7.88.1-10+deb12u6 amd64 [upgradable from: 7.88.1-10+deb12u5]\n"
+    "__APT_RC=0\n"
 )
 
 
@@ -639,13 +687,16 @@ def test_get_facts_baremetal_vendor_model_serial(driver):
     platform = {"vendor": "Dell Inc.", "model": "PowerEdge R720", "serial": "ABC123", "is_vm": False}
     with patch.object(driver, "_collect_platform_info", return_value=platform), \
          patch.object(driver, "_parse_uptime", return_value=86400), \
-         patch.object(driver, "_send", side_effect=["myhost", "myhost.example.com", "Debian GNU/Linux 12", "eth0\neth1"]):
+         patch.object(driver, "_send", side_effect=["myhost", "myhost.example.com", "Debian GNU/Linux 12", "eth0\neth1",
+                                                   "6.1.0-18-amd64"]):
         facts = driver.get_facts()
     assert facts["vendor"] == "Dell Inc."
     assert facts["model"] == "PowerEdge R720"
     assert facts["serial_number"] == "ABC123"
     assert facts["hostname"] == "myhost"
     assert facts["uptime"] == 86400
+    # Booted kernel, not the newest installed one — kernel CVE relevance needs it.
+    assert facts["running_kernel"] == "6.1.0-18-amd64"
 
 
 def test_get_facts_vm_kvm(driver):
@@ -655,7 +706,8 @@ def test_get_facts_vm_kvm(driver):
     }
     with patch.object(driver, "_collect_platform_info", return_value=platform), \
          patch.object(driver, "_parse_uptime", return_value=3600), \
-         patch.object(driver, "_send", side_effect=["vmhost", "vmhost.local", "Ubuntu 22.04 LTS", "eth0"]):
+         patch.object(driver, "_send", side_effect=["vmhost", "vmhost.local", "Ubuntu 22.04 LTS", "eth0",
+                                                   "5.15.0-91-generic"]):
         facts = driver.get_facts()
     assert facts["vendor"] == "KVM"
     assert facts["model"] == "Virtual Machine"
@@ -666,7 +718,7 @@ def test_get_facts_fallback_vendor_when_dmi_empty(driver):
     platform = {"vendor": "", "model": "", "serial": "", "is_vm": False}
     with patch.object(driver, "_collect_platform_info", return_value=platform), \
          patch.object(driver, "_parse_uptime", return_value=0), \
-         patch.object(driver, "_send", side_effect=["host", "host.local", "Alpine Linux 3.19", "eth0"]):
+         patch.object(driver, "_send", side_effect=["host", "host.local", "Alpine Linux 3.19", "eth0", "6.6.7-0-lts"]):
         facts = driver.get_facts()
     assert facts["vendor"] == "Linux"  # fallback to VENDOR class attribute
 
@@ -861,3 +913,505 @@ class TestDockerBinHook:
         assert docker_cmds
         for cmd in docker_cmds:
             assert "/opt/cs/docker" in cmd, f"unconverted call site: {cmd}"
+
+
+# ---------------------------------------------------------------------------
+# uninstall_package – purge, and getting out of `install ok unpacked`
+# ---------------------------------------------------------------------------
+
+
+class TestUninstallPackage:
+    """Removing a package that does not want to go.
+
+    Both cases here were found during a fleet-wide Wazuh rollback. Of thirteen
+    hosts carrying the agent, seven sat at `install ok unpacked` with the unit
+    failed — an upgrade whose postinst could not reach a manager that no longer
+    existed. `apt-get remove` cannot help there: apt configures a package before
+    removing it, and configuring is exactly what was broken.
+
+    And `remove` leaves the configuration behind by design, which for the Wazuh
+    agent means its apt source keeps being fetched on every update, long after
+    the package is gone.
+    """
+
+    def test_remove_is_still_the_default(self, driver):
+        """Callers that did not ask for a purge must not get one: configuration
+        somebody may want back is not this function's to delete."""
+        _mock_send(driver, "Removing wazuh-agent ...")
+
+        driver.uninstall_package("wazuh-agent")
+
+        sent = driver._device.send_command.call_args[0][0]
+        assert "apt-get remove" in sent
+        assert "purge" not in sent
+
+    def test_purge_is_asked_for_explicitly(self, driver):
+        _mock_send(driver, "Purging configuration files for wazuh-agent ...")
+
+        driver.uninstall_package("wazuh-agent", purge=True)
+
+        assert "apt-get purge" in driver._device.send_command.call_args[0][0]
+
+    def test_a_half_configured_package_falls_back_to_dpkg(self, driver):
+        """`install ok unpacked` is the state apt cannot get out of. On one host
+        only `dpkg --purge --force-all` removed it."""
+        driver._device.send_command.side_effect = [
+            "E: Sub-process /usr/bin/dpkg returned an error code (1)",
+            "Removing wazuh-agent (4.14.7-1) ...",
+        ]
+
+        result = driver.uninstall_package("wazuh-agent", purge=True)
+
+        assert result["success"] is True
+        second = driver._device.send_command.call_args_list[1][0][0]
+        assert "dpkg --purge --force-all" in second
+
+    def test_the_fallback_is_not_tried_when_the_first_pass_worked(self, driver):
+        """A forced dpkg purge is a bigger hammer than apt and must stay a last
+        resort, not a routine second step."""
+        _mock_send(driver, "Removing wazuh-agent ...")
+
+        driver.uninstall_package("wazuh-agent", purge=True)
+
+        assert driver._device.send_command.call_count == 1
+
+    def test_a_package_manager_without_purge_still_removes(self, driver):
+        """apk and pacman have no separate purge; asking for one must not turn
+        into a failure or a command they do not understand."""
+        driver._pkg_manager = "apk"
+        _mock_send(driver, "(1/1) Purging wazuh-agent")
+
+        result = driver.uninstall_package("wazuh-agent", purge=True)
+
+        assert result["success"] is True
+        assert "apk del" in driver._device.send_command.call_args[0][0]
+
+
+# ---------------------------------------------------------------------------
+# uninstall_package – success from the exit status, not from prose (netork#267)
+# ---------------------------------------------------------------------------
+
+
+def _with_rc(output: str, rc: int) -> str:
+    """What the shell prints for a command run through ``_sudo_status``."""
+    return f"{output}\n__NETORK_RC={rc}"
+
+
+class TestSudoStatus:
+    """``_sudo_status`` keeps the exit status that ``|| true`` throws away."""
+
+    def test_returns_output_and_exit_status(self, driver):
+        _mock_send(driver, _with_rc("Removing wazuh-agent ...", 0))
+
+        assert driver._sudo_status("apt-get remove -y wazuh-agent") == (
+            "Removing wazuh-agent ...",
+            0,
+        )
+
+    def test_a_non_zero_exit_status_is_reported(self, driver):
+        _mock_send(driver, _with_rc("E: Unable to locate package nope", 100))
+
+        assert driver._sudo_status("apt-get remove -y nope")[1] == 100
+
+    def test_the_status_is_read_right_after_sudo_returns(self, driver):
+        """``$?`` must be read straight after the sudo pipeline — with an
+        ``|| true`` in between, every command would report 0."""
+        driver._sudo_password = "pw"  # noqa: S105
+        _mock_send(driver, _with_rc("", 0))
+
+        driver._sudo_status("apt-get remove -y x 2>&1")
+
+        sent = driver._device.send_command.call_args[0][0]
+        assert sent.startswith("echo pw | sudo -S")
+        assert sent.endswith("apt-get remove -y x 2>&1; echo __NETORK_RC=$?")
+        assert "|| true" not in sent
+
+    def test_a_missing_marker_means_unknown_not_success(self, driver):
+        """Output cut short before the marker arrived says nothing about the
+        exit status; ``None`` says so instead of guessing 0."""
+        _mock_send(driver, "Removing wazuh-agent ...")
+
+        assert driver._sudo_status("apt-get remove -y wazuh-agent") == (
+            "Removing wazuh-agent ...",
+            None,
+        )
+
+    def test_the_command_echo_is_not_mistaken_for_the_marker(self, driver):
+        """A terminal may echo the command line back; its literal ``$?`` is not
+        a number, and only the marker on a line of its own counts."""
+        _mock_send(
+            driver,
+            "sudo apt-get remove -y x; echo __NETORK_RC=$?\nRemoving x ...\n__NETORK_RC=1",
+        )
+
+        output, rc = driver._sudo_status("apt-get remove -y x")
+
+        assert rc == 1
+        assert "__NETORK_RC=1" not in output
+
+
+class TestUninstallExitStatus:
+    """Whether a removal worked is what the package manager's exit status says.
+
+    Reading it out of human-readable output was guesswork in both directions:
+    apt's commonest failure (``E: Sub-process /usr/bin/dpkg returned an error
+    code (1)``) read as success until #240, and a successful removal whose
+    prerm merely *mentions* a failure read as a failure.
+    """
+
+    def test_a_non_zero_exit_is_a_failure_whatever_the_output_says(self, driver):
+        """Nothing in this output matches a failure keyword; only the exit
+        status knows."""
+        driver._pkg_manager = "dnf"
+        _mock_send(driver, _with_rc("Removing: wazuh-agent", 1))
+
+        result = driver.uninstall_package("wazuh-agent")
+
+        assert result["success"] is False
+
+    def test_a_zero_exit_is_a_success_even_if_the_output_mentions_failure(self, driver):
+        """A prerm that cannot stop an already-dead unit prints "Failed" and
+        still lets the removal complete."""
+        _mock_send(
+            driver,
+            _with_rc(
+                "Removing wazuh-agent (4.14.7-1) ...\n"
+                "Failed to stop wazuh-agent.service: Unit wazuh-agent.service not loaded.",
+                0,
+            ),
+        )
+
+        result = driver.uninstall_package("wazuh-agent")
+
+        assert result["success"] is True
+
+    def test_the_marker_does_not_reach_the_caller(self, driver):
+        _mock_send(driver, _with_rc("Removing wazuh-agent ...", 0))
+
+        result = driver.uninstall_package("wazuh-agent")
+
+        assert result["output"] == "Removing wazuh-agent ..."
+
+    def test_the_uninstall_command_keeps_its_exit_status(self, driver):
+        _mock_send(driver, _with_rc("Removing wazuh-agent ...", 0))
+
+        driver.uninstall_package("wazuh-agent")
+
+        sent = driver._device.send_command.call_args[0][0]
+        assert "|| true" not in sent
+        assert sent.endswith("; echo __NETORK_RC=$?")
+
+    def test_apt_failing_by_exit_status_falls_back_to_dpkg(self, driver):
+        driver._device.send_command.side_effect = [
+            _with_rc("E: Sub-process /usr/bin/dpkg returned an error code (1)", 100),
+            _with_rc("Removing wazuh-agent (4.14.7-1) ...", 0),
+        ]
+
+        result = driver.uninstall_package("wazuh-agent", purge=True)
+
+        assert result["success"] is True
+        second = driver._device.send_command.call_args_list[1][0][0]
+        assert "dpkg --purge --force-all" in second
+        assert "|| true" not in second
+        assert "__NETORK_RC" not in result["output"]
+
+    def test_the_dpkg_fallback_failing_is_a_failure(self, driver):
+        driver._device.send_command.side_effect = [
+            _with_rc("E: Sub-process /usr/bin/dpkg returned an error code (1)", 100),
+            _with_rc("dpkg: error processing package wazuh-agent (--purge):", 1),
+        ]
+
+        result = driver.uninstall_package("wazuh-agent", purge=True)
+
+        assert result["success"] is False
+        assert "dpkg --purge --force-all" in result["output"]
+
+    def test_a_zero_exit_does_not_trigger_the_fallback(self, driver):
+        """Even when the output contains words that used to mean failure: apt
+        exits 0 for a package that is already gone, which is the state the
+        caller asked for."""
+        _mock_send(driver, _with_rc("Package 'x' is not installed, so not removed", 0))
+
+        result = driver.uninstall_package("x", purge=True)
+
+        assert result["success"] is True
+        assert driver._device.send_command.call_count == 1
+
+    def test_without_an_exit_status_the_output_is_read_as_before(self, driver):
+        """If the marker never arrived, the keyword check is still the best
+        answer available — and it errs towards failure on apt's ``E:``."""
+        driver._pkg_manager = "dnf"
+        _mock_send(driver, "E: Sub-process /usr/bin/dpkg returned an error code (1)")
+
+        result = driver.uninstall_package("wazuh-agent")
+
+        assert result["success"] is False
+
+
+# ---------------------------------------------------------------------------
+# get_kernel_facts -- the command and its parse live in napalm-device-types
+# ---------------------------------------------------------------------------
+
+
+def _kernel_wire(report: str) -> str:
+    import base64
+    import gzip
+
+    return "KFACTS_BEGIN\n" + base64.encodebytes(gzip.compress(report.encode())).decode() + "KFACTS_END"
+
+
+def test_get_kernel_facts_carries_the_shared_command_across(driver):
+    from napalm_device_types import KernelFactsMixin
+    from napalm_device_types.kernel import KERNEL_FACTS_COMMAND
+
+    assert isinstance(driver, KernelFactsMixin)
+
+    report = "[release]\n6.1.0-25-amd64\n[loaded]\ntipc\n[available]\nkernel/net/tipc/tipc.ko.xz\n"
+    with patch.object(driver, "_send", return_value=_kernel_wire(report)) as send:
+        facts = driver.get_kernel_facts()
+
+    assert send.call_args.args[0] == KERNEL_FACTS_COMMAND
+    assert facts["release"] == "6.1.0-25-amd64"
+    assert facts["loaded"] == ["tipc"]
+    assert facts["available"] == ["tipc"]
+    assert facts["builtin"] is None
+
+
+def test_get_kernel_facts_raises_on_output_without_a_report(driver):
+    with patch.object(driver, "_send", return_value="sh: base64: not found"):
+        with pytest.raises(ValueError):
+            driver.get_kernel_facts()
+
+
+# ---------------------------------------------------------------------------
+# Services: listed in one round trip, controlled through systemctl (#7)
+# ---------------------------------------------------------------------------
+
+_REPORT = (
+    "SVC_BEGIN\n[files]\ncron.service enabled enabled\n[units]\n"
+    "MainPID=640\nId=cron.service\nNames=cron.service\nLoadState=loaded\n"
+    "ActiveState=active\nSubState=running\nUnitFileState=enabled\n"
+    "[generated]\nSVC_END\n"
+)
+
+
+class TestGetServices:
+    def test_one_command_lists_every_service(self, driver):
+        driver._device.send_command.return_value = _REPORT
+
+        services = driver.get_services()
+
+        assert services == [{"name": "cron", "running": True, "enabled": True, "pid": 640}]
+        assert driver._device.send_command.call_count == 1
+        assert "systemctl show" in driver._device.send_command.call_args[0][0]
+
+    def test_a_host_without_systemd_falls_back_to_service(self, driver):
+        driver._device.send_command.side_effect = [
+            "SVC_BEGIN\n[no-systemd]\n[files]\n[units]\n[generated]\nSVC_END\n",
+            " [ + ]  cron\n [ - ]  rsync\n",
+        ]
+
+        services = driver.get_services()
+
+        assert {s["name"]: s["running"] for s in services} == {"cron": True, "rsync": False}
+        assert "service --status-all" in driver._device.send_command.call_args[0][0]
+
+
+class TestManageService:
+    def _sent(self, driver) -> list[str]:
+        return [c[0][0] for c in driver._device.send_command.call_args_list]
+
+    def test_as_root_the_command_runs_as_it_is(self, driver):
+        driver._device.send_command.side_effect = ["0", "__SVC_RC=0"]
+
+        assert driver.manage_service("cron", "restart") == {"success": True, "output": ""}
+        uid, action = self._sent(driver)
+        assert uid == "id -u"
+        assert action.startswith("timeout 45 systemctl --no-ask-password restart -- cron.service")
+
+    def test_with_a_sudo_password_it_goes_through_sudo(self, driver):
+        driver._sudo_password = "pw"  # noqa: S105
+        driver._device.send_command.side_effect = ["1000", "__SVC_RC=0"]
+
+        assert driver.manage_service("cron", "stop")["success"] is True
+        action = self._sent(driver)[1]
+        assert action.startswith("echo pw | sudo -S")
+        assert "timeout 45 systemctl --no-ask-password stop -- cron.service" in action
+
+    def test_without_one_sudo_never_waits_for_a_password(self, driver):
+        driver._device.send_command.side_effect = ["1000", "__SVC_RC=0"]
+
+        driver.manage_service("cron", "enable")
+
+        assert self._sent(driver)[1].startswith("sudo -n timeout 45 systemctl")
+
+    def test_a_missing_sudo_password_is_explained(self, driver):
+        driver._device.send_command.side_effect = [
+            "1000",
+            "sudo: a password is required\n__SVC_RC=1",
+        ]
+
+        result = driver.manage_service("cron", "restart")
+
+        assert result["success"] is False
+        assert "sudo password" in result["output"]
+        assert "NOPASSWD" in result["output"]
+
+    def test_a_failure_keeps_systemctls_message(self, driver):
+        driver._device.send_command.side_effect = [
+            "0",
+            "Failed to start nope.service: Unit nope.service not found.\n__SVC_RC=5",
+        ]
+
+        result = driver.manage_service("nope", "start")
+
+        assert result == {
+            "success": False,
+            "output": "Failed to start nope.service: Unit nope.service not found.",
+        }
+
+    def test_who_the_user_is_is_asked_once(self, driver):
+        driver._device.send_command.side_effect = ["0", "__SVC_RC=0", "__SVC_RC=0"]
+
+        driver.manage_service("cron", "stop")
+        driver.manage_service("cron", "start")
+
+        assert self._sent(driver).count("id -u") == 1
+
+    def test_an_invalid_name_is_refused_before_anything_is_sent(self, driver):
+        with pytest.raises(ValueError):
+            driver.manage_service("cron; reboot", "stop")
+
+        assert driver._device.send_command.call_count == 0
+
+
+# ---------------------------------------------------------------------------
+# Updates: origin and security, refresh, host status (netOrk MVP 5)
+# ---------------------------------------------------------------------------
+
+APT_WITH_SECURITY = (
+    "openssl/noble-updates,noble-security 3.0.13-0ubuntu3.6 amd64 [upgradable from: 3.0.13-0ubuntu3.5]\n"
+    "docker-compose-plugin/noble 5.6.0-1~ubuntu.24.04~noble amd64 [upgradable from: 5.5.1-1~ubuntu.24.04~noble]\n"
+    "__APT_RC=0\n"
+)
+
+
+class TestAvailableUpdates:
+    def test_apt_reports_origin_and_security(self, driver):
+        driver._pkg_manager = "apt"
+        _mock_send(driver, APT_WITH_SECURITY)
+
+        updates = {u["name"]: u for u in driver.get_available_updates()}
+
+        assert updates["openssl"]["security"] is True
+        assert updates["openssl"]["origin"] == "noble-updates,noble-security"
+        assert updates["docker-compose-plugin"]["security"] is False
+
+    def test_apt_that_could_not_read_raises_instead_of_reporting_nothing(self, driver):
+        driver._pkg_manager = "apt"
+        _mock_send(driver, "E: Could not open lock file\n__APT_RC=100\n")
+
+        with pytest.raises(ValueError):
+            driver.get_available_updates()
+
+    def test_apt_without_an_exit_status_raises(self, driver):
+        driver._pkg_manager = "apt"
+        _mock_send(driver, "openssl/noble-security 3.0.13-0ubuntu3.6 amd64 [upgradable fro")
+
+        with pytest.raises(ValueError):
+            driver.get_available_updates()
+
+    def test_dnf_marks_what_a_security_advisory_covers(self, driver):
+        driver._pkg_manager = "dnf"
+        driver._device.send_command.side_effect = [
+            "0",  # id -u
+            "openssl-libs.x86_64   1:3.1.4-2.fc40   updates\n"
+            "vim-enhanced.x86_64   2:9.1.083-1.fc40 updates\n__NETORK_RC=100",
+            "FEDORA-2024-1a2b3c4d5e Important/Sec. openssl-libs-1:3.1.4-2.fc40.x86_64\n__NETORK_RC=0",
+        ]
+
+        updates = {u["name"]: u for u in driver.get_available_updates()}
+
+        assert updates["openssl-libs"]["security"] is True
+        assert updates["vim-enhanced"]["security"] is False
+
+    def test_dnf_without_advisories_leaves_security_unknown(self, driver):
+        driver._pkg_manager = "dnf"
+        driver._device.send_command.side_effect = [
+            "0",
+            "vim-enhanced.x86_64   2:9.1.083-1.fc40 updates\n__NETORK_RC=100",
+            "Error: updateinfo metadata missing\n__NETORK_RC=1",
+        ]
+
+        assert driver.get_available_updates()[0]["security"] is None
+
+    def test_dnf_that_failed_raises(self, driver):
+        driver._pkg_manager = "dnf"
+        driver._device.send_command.side_effect = ["0", "Error: Failed to download metadata\n__NETORK_RC=1"]
+
+        with pytest.raises(RuntimeError):
+            driver.get_available_updates()
+
+
+class TestRefreshAvailableUpdates:
+    def _sent(self, driver) -> list:
+        return [c[0][0] for c in driver._device.send_command.call_args_list]
+
+    def test_apt_refreshes_its_index_as_root(self, driver):
+        driver._pkg_manager = "apt"
+        driver._device.send_command.side_effect = ["0", "Hit:1 http://archive.ubuntu.com noble InRelease\n__NETORK_RC=0"]
+
+        result = driver.refresh_available_updates()
+
+        assert result["success"] is True
+        assert "apt-get update" in self._sent(driver)[1]
+
+    def test_without_a_sudo_password_it_never_waits_for_one(self, driver):
+        driver._pkg_manager = "apt"
+        driver._device.send_command.side_effect = ["1000", "sudo: a password is required\n__NETORK_RC=1"]
+
+        result = driver.refresh_available_updates()
+
+        assert result["success"] is False
+        assert self._sent(driver)[1].startswith("sudo -n apt-get update")
+
+    def test_dnf_refreshes_its_metadata(self, driver):
+        driver._pkg_manager = "dnf"
+        driver._device.send_command.side_effect = ["0", "Metadata cache created.\n__NETORK_RC=0"]
+
+        assert driver.refresh_available_updates()["success"] is True
+        assert "dnf makecache" in self._sent(driver)[1]
+
+    def test_pacman_is_not_refreshed_on_its_own(self, driver):
+        """pacman -Sy without -u invites a partial upgrade on the next install."""
+        driver._pkg_manager = "pacman"
+
+        result = driver.refresh_available_updates()
+
+        assert result["success"] is False
+        driver._device.send_command.assert_not_called()
+
+
+class TestHostStatus:
+    def test_the_driver_carries_the_shared_command(self, driver):
+        from napalm_device_types.host_status import HOST_STATUS_COMMAND
+
+        _mock_send(
+            driver,
+            "HSTAT_BEGIN\n[reboot-required]\n[kernel]\n6.8.0-142-generic\n[modules]\n"
+            "6.8.0-142-generic\n[timers]\napt-daily-upgrade.timer enabled\nHSTAT_END\n",
+        )
+
+        status = driver.get_host_status()
+
+        assert driver._device.send_command.call_args[0][0] == HOST_STATUS_COMMAND
+        assert status["reboot_required"] is True
+
+
+class TestTerminalCodes:
+    def test_a_status_marker_behind_a_terminal_code_is_still_read(self, driver):
+        """apt-get on a pseudo-terminal leaves keypad codes in front of the marker."""
+        driver._pkg_manager = "apt"
+        driver._device.send_command.side_effect = ["0", "Hit:1 noble InRelease\n\x1b>__NETORK_RC=0"]
+
+        assert driver.refresh_available_updates()["success"] is True