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",
@@ -116,7 +168,23 @@ def _arm_vendor_from_model(model: str) -> str:
     return " ".join(brand)
 
 
-class LinuxDriver(OSDriver):
+#: A bare image ID (``d626d04934cd``), not a registry reference. ``docker ps``
+#: falls back to this whenever the tag a container was created from has since
+#: been moved to a newer image — i.e. exactly after a pull without a recreate.
+_IMAGE_ID_RE = re.compile(r"^(sha256:)?[0-9a-f]{12,64}$")
+
+
+def _looks_like_image_id(ref: str) -> bool:
+    """True if *ref* is an image ID rather than something a registry can resolve."""
+    return bool(_IMAGE_ID_RE.match(ref.strip()))
+
+
+def _short_image_id(raw: str) -> str:
+    """Normalise ``sha256:<64hex>`` and ``<12hex>`` to a comparable 12-char form."""
+    return raw.strip().removeprefix("sha256:")[:12]
+
+
+class LinuxDriver(KernelFactsMixin, SystemdServicesMixin, HostStatusMixin, OSDriver):
     """NAPALM driver for generic Linux systems.
 
     Connects via SSH (netmiko ``linux`` device type) and auto-detects the
@@ -126,6 +194,9 @@ class LinuxDriver(OSDriver):
     TYPE_LABEL = "Linux"
     VENDOR = "Linux"
     DRIVER_NAME = "linux"
+    # A general-purpose host runs through a full init sequence; NAS derivatives
+    # (OpenMediaVault, QNAP) inherit this and are, if anything, slower.
+    REBOOT_SETTLE_SECONDS = 90
     SNMP_FINGERPRINT = [
         FingerprintRule("linux", weight=5.0),
     ]
@@ -248,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:
@@ -381,6 +501,10 @@ class LinuxDriver(OSDriver):
         iface_out = self._send("ip -o link show | awk -F': ' '{print $2}' | cut -d@ -f1")
         interface_list = [i.strip() for i in iface_out.splitlines() if i.strip() and i.strip() != "lo"]
 
+        # Currently-booted kernel release, distinct from an installed-but-not-yet-
+        # booted newer kernel (used for kernel CVE relevance).
+        running_kernel = self._send("uname -r").strip()
+
         return {
             "hostname": hostname,
             "fqdn": fqdn,
@@ -390,6 +514,7 @@ class LinuxDriver(OSDriver):
             "os_version": os_version,
             "uptime": uptime_secs,
             "interface_list": interface_list,
+            "running_kernel": running_kernel,
         }
 
     def _parse_uptime(self) -> int:
@@ -791,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
     # ------------------------------------------------------------------
@@ -810,17 +943,36 @@ class LinuxDriver(OSDriver):
 
     def _get_packages_apt(self) -> list[PackageDict]:
         out = self._send(
-            "dpkg-query -W -f='${Package}\\t${Version}\\t${Installed-Size}\\t${binary:Summary}\\n' 2>/dev/null"
+            "dpkg-query -W -f='${Package}\\t${Version}\\t${Installed-Size}"
+            "\\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", 3)
+            # 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()
             version = parts[1].strip()
             size = int(parts[2].strip()) * 1024 if len(parts) > 2 and parts[2].strip().isdigit() else 0
-            description = parts[3].strip() if len(parts) > 3 else ""
+            # 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
+            # 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,
@@ -828,6 +980,8 @@ class LinuxDriver(OSDriver):
                 "description": description,
                 "size": size,
                 "source": "apt",
+                "source_package": source_package,
+                "source_version": source_version,
             })
         return packages
 
@@ -1000,26 +1154,88 @@ 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"))
-        return {"success": success, "output": raw.strip()}
 
-    def get_pending_updates(self) -> list[UpdateDict]:
+        raw, rc = self._sudo_status(cmd)
+        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":
             return self._get_updates_apt()
         if self._pkg_manager in ("dnf", "yum"):
@@ -1032,10 +1248,6 @@ class LinuxDriver(OSDriver):
             f"Package manager '{self._pkg_manager}' is not supported"
         )
 
-    def get_available_updates(self) -> list[UpdateDict]:
-        """Alias for get_pending_updates(); called by the netork API backend."""
-        return self.get_pending_updates()
-
     def get_device_warnings(self) -> List[dict[str, Any]]:
         """Return warning dicts for issues detected on this device.
 
@@ -1052,8 +1264,6 @@ class LinuxDriver(OSDriver):
         if updates:
             warnings.append({
                 "code": "updates_available",
-                "severity": "warning",
-                "action": None,
                 "meta": {
                     "count": len(updates),
                     "packages": [u.get("name", "") for u in updates],
@@ -1065,8 +1275,6 @@ class LinuxDriver(OSDriver):
                 if self._apt_proxy_url not in current:
                     warnings.append({
                         "code": "apt_proxy_missing",
-                        "severity": "warning",
-                        "action": "fix_apt_proxy",
                         "meta": {"expected_url": self._apt_proxy_url},
                     })
             except Exception as exc:
@@ -1076,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] = []
@@ -1247,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")
@@ -1484,6 +1668,16 @@ class LinuxDriver(OSDriver):
     # Docker
     # ------------------------------------------------------------------
 
+    def _docker_bin(self) -> str:
+        """Path to the docker binary.
+
+        A hook rather than a literal because the Docker *logic* is the same
+        everywhere while the *location* is not: QTS ships Container Station's
+        docker under /share/<pool>/.qpkg/ and never puts it on PATH. Subclasses
+        override this one method instead of reimplementing the surface.
+        """
+        return "docker"
+
     def get_docker_info(self) -> DockerInfoDict:
         """Return information about the local Docker environment.
 
@@ -1502,26 +1696,31 @@ class LinuxDriver(OSDriver):
         """
         import json as _json
 
+        docker = self._docker_bin()
+
         # Check docker binary first (docker --version doesn't need socket access)
-        if not self._send("command -v docker 2>/dev/null").strip():
+        if not self._send(f"command -v {docker} 2>/dev/null").strip():
             return {"available": False}
 
         # Verify socket access — docker ps is cheaper and fails immediately on permission errors
-        ps_check = self._send("docker ps 2>&1")
+        ps_check = self._send(f"{docker} ps 2>&1")
         if "permission denied" in ps_check.lower() or "cannot connect" in ps_check.lower():
             return {"available": False, "permission_denied": True}
 
-        version = self._send("docker --version 2>/dev/null").strip()
+        version = self._send(f"{docker} --version 2>/dev/null").strip()
 
         combined = self._send(
             "echo '---CONTAINERS---'; "
-            "docker ps -a --format '{{json .}}' 2>/dev/null; "
+            f"{docker} ps -a --format '{{{{json .}}}}' 2>/dev/null; "
             "echo '---IMAGES---'; "
-            "docker images --format '{{json .}}' 2>/dev/null; "
+            f"{docker} images --format '{{{{json .}}}}' 2>/dev/null; "
             "echo '---VOLUMES---'; "
-            "docker volume ls --format '{{json .}}' 2>/dev/null; "
+            f"{docker} volume ls --format '{{{{json .}}}}' 2>/dev/null; "
             "echo '---NETWORKS---'; "
-            "docker network ls --format '{{json .}}' 2>/dev/null",
+            f"{docker} network ls --format '{{{{json .}}}}' 2>/dev/null; "
+            "echo '---CONFIGIMAGES---'; "
+            f"{docker} ps -aq 2>/dev/null | xargs -r {docker} inspect "
+            f"--format '{{{{.Id}}}}|{{{{.Config.Image}}}}|{{{{.Image}}}}' 2>/dev/null",
             read_timeout=60,
         )
 
@@ -1540,7 +1739,8 @@ class LinuxDriver(OSDriver):
         raw_containers = _section(combined, "---CONTAINERS---", "---IMAGES---")
         raw_images     = _section(combined, "---IMAGES---",     "---VOLUMES---")
         raw_volumes    = _section(combined, "---VOLUMES---",    "---NETWORKS---")
-        raw_networks   = _section(combined, "---NETWORKS---",   "\x00")  # sentinel
+        raw_networks   = _section(combined, "---NETWORKS---",   "---CONFIGIMAGES---")
+        raw_cfgimages  = _section(combined, "---CONFIGIMAGES---", "\x00")  # sentinel
 
         def _parse_labels(raw: Any) -> Dict[str, str]:
             """Parse Docker labels — may be a dict (JSON map) or comma-sep string."""
@@ -1601,6 +1801,44 @@ class LinuxDriver(OSDriver):
             except Exception:
                 pass
 
+        # Stable image reference + restart-pending detection.
+        #
+        # ``docker ps`` only reports a usable tag while that tag still resolves to
+        # the running image. Pull a newer image without recreating the container and
+        # it degrades to a bare image ID — useless as a registry reference, and the
+        # very state in which an update is waiting. ``.Config.Image`` is the
+        # reference the container was created from and never degrades.
+        cfg_by_cid: dict[str, tuple] = {}
+        for line in raw_cfgimages.splitlines():
+            parts = line.strip().split("|")
+            if len(parts) != 3 or not parts[0]:
+                continue
+            cid, cfg_ref, run_id = parts
+            cfg_by_cid[cid[:12]] = (cfg_ref.strip(), run_id.strip())
+
+        tag_index: dict[str, tuple] = {}
+        for im in images:
+            repo, tag = im.get("repository", ""), im.get("tag", "")
+            if not repo or not tag or "<none>" in (repo, tag):
+                continue
+            tag_index[f"{repo}:{tag}"] = (_short_image_id(im.get("id", "")), im.get("version", ""))
+
+        for c in containers:
+            cfg_ref, run_id = cfg_by_cid.get(c.get("id", "")[:12], ("", ""))
+            if not cfg_ref:
+                continue
+            c["image_ref"] = cfg_ref
+            c["running_image_id"] = _short_image_id(run_id)
+            c["restart_pending"] = False
+            c["pending_version"] = ""
+            # A stopped container is not "pending a restart" in any useful sense.
+            if c.get("state") != "running":
+                continue
+            tag_id, tag_version = tag_index.get(cfg_ref, ("", ""))
+            if tag_id and c["running_image_id"] and tag_id != c["running_image_id"]:
+                c["restart_pending"] = True
+                c["pending_version"] = tag_version
+
         # Volumes
         volumes: List[dict[str, Any]] = []
         for line in raw_volumes.splitlines():
@@ -1658,15 +1896,28 @@ class LinuxDriver(OSDriver):
         Returns a list of image references that have a newer digest available.
         """
         outdated_images: List[str] = []
-        candidate_images: List[str] = list({
+        # Prefer the reference the container was created from. `image` is whatever
-            c["image"] for c in containers
+        # `docker ps` displayed, which collapses to a bare image ID once the tag has
-            if c.get("image")
+        # moved on — and an image ID is not something a registry can resolve.
-            and "@sha256:" not in c.get("image", "")  # skip digest-pinned
+        candidate_images: List[str] = []
-        })
+        for c in containers:
+            ref = (c.get("image_ref") or c.get("image") or "").strip()
+            if not ref or "@sha256:" in ref:  # skip digest-pinned
+                continue
+            if _looks_like_image_id(ref):
+                logger.warning(
+                    "container %s reports image ID %r instead of a tag — cannot ask the "
+                    "registry about it; skipping update check",
+                    c.get("name", "?"), ref,
+                )
+                continue
+            if ref not in candidate_images:
+                candidate_images.append(ref)
         for img_name in candidate_images:
             try:
                 local_raw = self._send(
-                    f"docker inspect {img_name!r} --format '{{{{index .RepoDigests 0}}}}' 2>/dev/null",
+                    f"{self._docker_bin()} inspect {img_name!r} "
+                    f"--format '{{{{index .RepoDigests 0}}}}' 2>/dev/null",
                     read_timeout=5,
                 ).strip()
                 if not local_raw or "@" not in local_raw:
@@ -1674,7 +1925,7 @@ class LinuxDriver(OSDriver):
                 local_digest = local_raw.split("@", 1)[1]
 
                 remote_full = self._send(
-                    f"docker buildx imagetools inspect {img_name!r} 2>&1",
+                    f"{self._docker_bin()} buildx imagetools inspect {img_name!r} 2>&1",
                     read_timeout=30,
                 ).strip()
                 if ("429" in remote_full
@@ -1693,6 +1944,11 @@ class LinuxDriver(OSDriver):
                         remote_digest = _ls[7:].strip()
                         break
                 if not remote_digest or not remote_digest.startswith("sha256:"):
+                    # Silence here is indistinguishable from "up to date" — say so.
+                    logger.warning(
+                        "no digest returned for %s; skipping update check. Registry said: %s",
+                        img_name, remote_full[:200].replace("\n", " ") or "(nothing)",
+                    )
                     continue
                 if local_digest != remote_digest:
                     outdated_images.append(img_name)
@@ -1716,7 +1972,7 @@ class LinuxDriver(OSDriver):
         import shlex as _shlex
 
         raw = self._send(
-            f"docker inspect {_shlex.quote(container_id)} 2>/dev/null",
+            f"{self._docker_bin()} inspect {_shlex.quote(container_id)} 2>/dev/null",
             read_timeout=10,
         ).strip()
         if not raw:
@@ -1906,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] = []
@@ -1936,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

@@ -22,6 +22,11 @@ def driver():
     d._secret = "pass"  # noqa: S105
     d._forced_pkg_manager = None
     d._pkg_manager = "apt"
+    # Set by __init__, which this fixture bypasses via __new__. Without it every
+    # call through _sudo() raises AttributeError, which the callers' broad
+    # `except Exception` turns into a plain {"success": False} -- so the tests
+    # failed for a reason that had nothing to do with what they were testing.
+    d._sudo_password = None
     d.netmiko_optional_args = {}
     d._device = MagicMock()
     return d
@@ -148,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"
 )
 
 
@@ -158,15 +176,49 @@ 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"
 
 
 # ---------------------------------------------------------------------------
-# get_pending_updates (apt)
+# get_available_updates (apt)
 # ---------------------------------------------------------------------------
 
 
@@ -174,13 +226,14 @@ 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"
 )
 
 
-def test_get_pending_updates_apt(driver):
+def test_get_available_updates_apt(driver):
     driver._pkg_manager = "apt"
-    with patch.object(driver, "_send", side_effect=["", APT_UPGRADABLE]):
+    with patch.object(driver, "_send", return_value=APT_UPGRADABLE):
-        updates = driver.get_pending_updates()
+        updates = driver.get_available_updates()
     assert len(updates) == 2
     assert updates[0]["name"] == "openssh-server"
     assert updates[0]["current_version"] == "1:9.2p1-2+deb12u1"
@@ -312,7 +365,10 @@ def test_apply_updates_apt_all_packages(driver):
     driver._pkg_manager = "apt"
     sent_commands = []
 
-    def capture_send(cmd):
+    def capture_send(cmd, **kwargs):
+        # _sudo() passes read_timeout as a keyword; without **kwargs this raises
+        # TypeError, which the caller's `except Exception` reports as a failed
+        # upgrade rather than a broken test double.
         sent_commands.append(cmd)
         return APT_UPGRADE_SUCCESS
 
@@ -631,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):
@@ -647,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"
@@ -658,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
 
@@ -802,3 +862,556 @@ class TestRunDeviceActionDispatch:
         ) as mock_action:
             driver.run_device_action("apt_update_upgrade")
         mock_action.assert_called_once()
+
+
+class TestDockerBinHook:
+    """Where the docker binary lives is device-specific; what to do with it is not.
+
+    QTS puts Container Station's docker under /share/<pool>/.qpkg/ and not on
+    PATH. Rather than duplicating the whole Docker surface in the QNAP driver,
+    the path is a one-method hook here and the logic stays generic. See
+    docs/ARCHITECTURE.md §4.4, "Generisch vs. treiberspezifisch".
+    """
+
+    def test_defaults_to_docker_on_path(self, driver):
+        assert driver._docker_bin() == "docker"
+
+    def test_detection_uses_the_hook(self, driver):
+        """A subclass pointing elsewhere must not be probed for a PATH docker."""
+        sent = []
+
+        def _record(cmd, **kwargs):
+            sent.append(cmd)
+            return ""
+
+        with patch.object(driver, "_docker_bin", return_value="/opt/cs/docker"):
+            with patch.object(driver, "_send", side_effect=_record):
+                result = driver.get_docker_info()
+
+        assert result == {"available": False}
+        assert any("/opt/cs/docker" in cmd for cmd in sent)
+        assert not any("command -v docker " in cmd for cmd in sent)
+
+    def test_all_docker_subcommands_use_the_hook(self, driver):
+        """Half-converted call sites are the failure mode here: detection would
+        find the binary and the actual queries would still miss it."""
+        sent = []
+
+        def _record(cmd, **kwargs):
+            sent.append(cmd)
+            if "command -v" in cmd:
+                return "/opt/cs/docker"
+            if "---CONTAINERS---" in cmd:
+                return "---CONTAINERS---\n---IMAGES---\n---VOLUMES---\n---NETWORKS---\n"
+            return ""
+
+        with patch.object(driver, "_docker_bin", return_value="/opt/cs/docker"):
+            with patch.object(driver, "_send", side_effect=_record):
+                driver.get_docker_info()
+
+        docker_cmds = [c for c in sent if "docker" in c]
+        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