Merge pull request 'feat: report the running kernel's modules and build configuration' (#8) from feat/kernel-facts into master

LinuxDriver mixes in KernelFactsMixin from napalm-device-types and supplies
only the transport: the shared read-only command over the existing SSH
session, no sudo, one round trip. OpenMediaVault and QNAP inherit it.

Requires napalm-device-types 2.1.0.

napalm_linux/linux.py

@@ -31,7 +31,7 @@ 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 FingerprintRule, KernelFactsMixin, OSDriver
 from napalm_device_types.models import (
     ApplyUpdatesResultDict,
     CronJobDict,
@@ -50,6 +50,12 @@ 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)
+
 # 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 +122,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, OSDriver):
     """NAPALM driver for generic Linux systems.
 
     Connects via SSH (netmiko ``linux`` device type) and auto-detects the
@@ -126,6 +148,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 +273,25 @@ 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".
+        """
+        raw = self._sudo(f"{command}; echo {_RC_MARKER}$?", read_timeout=read_timeout)
+        matches = list(_RC_MARKER_RE.finditer(raw))
+        if not matches:
+            return raw, None
+        last = matches[-1]
+        output = (raw[: last.start()] + raw[last.end():]).strip()
+        return output, int(last.group(1))
+
     def _detect_pkg_manager(self) -> str | None:
         """Return the first package manager binary found on PATH."""
         for pm in _PKG_MANAGERS:
@@ -381,6 +425,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 +438,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 +840,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 +867,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 +904,8 @@ class LinuxDriver(OSDriver):
                 "description": description,
                 "size": size,
                 "source": "apt",
+                "source_package": source_package,
+                "source_version": source_version,
             })
         return packages
 
@@ -1000,26 +1078,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 +1172,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 +1188,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 +1199,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:
@@ -1484,6 +1616,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 +1644,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 +1687,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 +1749,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 +1844,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 +1873,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 +1892,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 +1920,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:
@@ -1884,8 +2088,48 @@ class LinuxDriver(OSDriver):
             return self._action_fix_snmp()
         if action == "fix_apt_proxy":
             return self._action_fix_apt_proxy()
+        if action == "apt_update_upgrade":
+            return self._action_apt_update_upgrade()
         raise NotImplementedError(f"Unknown action: {action!r}")
 
+    def _action_apt_update_upgrade(self) -> DeviceActionResultDict:
+        """Refresh the apt cache and fully upgrade all packages (apt-based systems only).
+
+        Uses full-upgrade (not plain upgrade) — plain "apt-get upgrade" refuses
+        to install/remove packages even when required to satisfy a newer
+        version's dependencies, silently leaving those updates pending.
+        """
+        if self._pkg_manager != "apt":
+            return {
+                "success": True,
+                "output": f"Skipped — package manager is {self._pkg_manager!r}, not apt.",
+            }
+
+        sudo_check = self._send("sudo -n true 2>&1 || echo __SUDO_NEEDS_PW__")
+        if "__SUDO_NEEDS_PW__" in sudo_check or "password is required" in sudo_check.lower():
+            if not self._sudo_password:
+                return {
+                    "success": False,
+                    "output": (
+                        "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] = []
+        try:
+            out = self._sudo("apt-get update -y 2>&1", read_timeout=90)
+            lines.append(f"[update] {out.strip()[-300:]}")
+            out = self._sudo(
+                "DEBIAN_FRONTEND=noninteractive apt-get full-upgrade -y 2>&1", read_timeout=240
+            )
+            lines.append(f"[upgrade] {out.strip()[-300:]}")
+            return {"success": True, "output": "\n".join(lines)}
+        except Exception as exc:
+            lines.append(f"[error] {exc}")
+            return {"success": False, "output": "\n".join(lines)}
+
     def _action_fix_snmp(self) -> DeviceActionResultDict:
         """Install, configure and start snmpd with community 'public'."""
         lines: list[str] = []
@@ -1908,6 +2152,16 @@ class LinuxDriver(OSDriver):
         if not pkg_mgr:
             return {"success": False, "output": "Package manager not detected — cannot install snmpd."}
 
+        # Refresh the package index first — a freshly provisioned (or simply
+        # long-untouched) system's cache can be stale/empty, which makes the
+        # install below fail outright rather than just being slow.
+        if pkg_mgr == "apt":
+            try:
+                update_out = self._sudo("apt-get update -y 2>&1", read_timeout=90)
+                lines.append(f"[update] {update_out.strip()[-200:]}")
+            except Exception as exc:
+                lines.append(f"[warn] apt-get update failed: {exc}")
+
         # Install both snmpd (daemon) and snmp (client tools incl. snmpget for probing)
         install_cmd: dict[str, str] = {
             "apt":    "DEBIAN_FRONTEND=noninteractive apt-get install -y snmpd snmp 2>&1",
@@ -1918,8 +2172,12 @@ class LinuxDriver(OSDriver):
         }
         cmd = install_cmd.get(pkg_mgr)
         if cmd:
-            out = self._sudo(cmd, read_timeout=120)
+            try:
-            lines.append(f"[install] {out.strip()[-200:]}")
+                out = self._sudo(cmd, read_timeout=120)
+                lines.append(f"[install] {out.strip()[-200:]}")
+            except Exception as exc:
+                lines.append(f"[error] install failed: {exc}")
+                return {"success": False, "output": "\n".join(lines)}
 
         # 2. Determine the IP netOrk is connecting from by checking the established SSH connection
         netork_ip = ""

pyproject.toml

@@ -37,7 +37,7 @@ classifiers = [
 ]
 dependencies = [
     "napalm>=4.0",
-    "napalm-device-types>=0.3.0",
+    "napalm-device-types>=2.1.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)
 # ---------------------------------------------------------------------------
 
 
@@ -177,10 +229,10 @@ APT_UPGRADABLE = (
 )
 
 
-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 +364,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 +686,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 +705,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,6 +717,466 @@ 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
+
+
+# ---------------------------------------------------------------------------
+# _action_fix_snmp / _action_apt_update_upgrade
+# ---------------------------------------------------------------------------
+
+
+def _fix_snmp_send_side_effect(command: str, read_timeout: float = 100) -> str:
+    """Canned responses covering every _send() call _action_fix_snmp makes."""
+    if command.startswith("sudo -n true"):
+        return ""  # passwordless sudo works
+    if command.startswith("command -v apt"):
+        return "/usr/bin/apt"
+    if command.startswith("command -v"):
+        return ""  # ufw/iptables not found, other pkg managers not found
+    if command.startswith("ss -tnp"):
+        return ""  # no netork_ip detected — skips firewall step
+    if command.startswith("cat /etc/snmp/snmpd.conf"):
+        return "agentAddress udp:161\nrocommunity public\n"
+    if command.startswith("snmpget"):
+        return "STRING: Linux test"
+    return ""
+
+
+class TestActionFixSnmp:
+    def test_runs_apt_get_update_before_install(self, driver):
+        driver._pkg_manager = "apt"
+        driver._sudo_password = None
+        sudo_calls: list[str] = []
+
+        def _sudo_side_effect(command: str, read_timeout: float = 100) -> str:
+            sudo_calls.append(command)
+            return ""
+
+        with (
+            patch.object(driver, "_send", side_effect=_fix_snmp_send_side_effect),
+            patch.object(driver, "_sudo", side_effect=_sudo_side_effect),
+        ):
+            driver._action_fix_snmp()
+
+        update_idx = next(i for i, c in enumerate(sudo_calls) if "apt-get update" in c)
+        install_idx = next(i for i, c in enumerate(sudo_calls) if "apt-get install" in c)
+        assert update_idx < install_idx
+
+    def test_install_exception_returns_failure_instead_of_raising(self, driver):
+        driver._pkg_manager = "apt"
+        driver._sudo_password = None
+
+        def _sudo_side_effect(command: str, read_timeout: float = 100) -> str:
+            if "apt-get install" in command:
+                raise TimeoutError("connection timed out")
+            return ""
+
+        with (
+            patch.object(driver, "_send", side_effect=_fix_snmp_send_side_effect),
+            patch.object(driver, "_sudo", side_effect=_sudo_side_effect),
+        ):
+            result = driver._action_fix_snmp()
+
+        assert result["success"] is False
+        assert "install failed" in result["output"]
+
+    def test_apt_get_update_failure_is_non_fatal(self, driver):
+        """apt-get update failing (e.g. transient network issue) must not
+        abort the whole action — install is still attempted."""
+        driver._pkg_manager = "apt"
+        driver._sudo_password = None
+
+        def _sudo_side_effect(command: str, read_timeout: float = 100) -> str:
+            if "apt-get update" in command:
+                raise TimeoutError("network unreachable")
+            return ""
+
+        with (
+            patch.object(driver, "_send", side_effect=_fix_snmp_send_side_effect),
+            patch.object(driver, "_sudo", side_effect=_sudo_side_effect),
+        ):
+            result = driver._action_fix_snmp()
+
+        assert "apt-get update failed" in result["output"]
+
+
+class TestActionAptUpdateUpgrade:
+    def test_skips_when_not_apt(self, driver):
+        driver._pkg_manager = "dnf"
+        result = driver._action_apt_update_upgrade()
+        assert result["success"] is True
+        assert "Skipped" in result["output"]
+
+    def test_fails_when_sudo_needs_password_and_none_configured(self, driver):
+        driver._pkg_manager = "apt"
+        driver._sudo_password = None
+        with patch.object(driver, "_send", return_value="__SUDO_NEEDS_PW__"):
+            result = driver._action_apt_update_upgrade()
+        assert result["success"] is False
+        assert "sudo requires a password" in result["output"]
+
+    def test_runs_update_then_upgrade_successfully(self, driver):
+        driver._pkg_manager = "apt"
+        driver._sudo_password = "secret"  # noqa: S105
+        with (
+            patch.object(driver, "_send", return_value=""),
+            patch.object(
+                driver,
+                "_sudo",
+                side_effect=["Reading package lists... Done", "0 upgraded, 0 newly installed"],
+            ) as mock_sudo,
+        ):
+            result = driver._action_apt_update_upgrade()
+
+        assert result["success"] is True
+        assert "[update]" in result["output"]
+        assert "[upgrade]" in result["output"]
+        update_call, upgrade_call = mock_sudo.call_args_list
+        assert "apt-get update" in update_call.args[0]
+        # full-upgrade (not plain upgrade) — plain upgrade silently holds back
+        # packages whose newer version needs to install/remove dependencies.
+        assert "apt-get full-upgrade" in upgrade_call.args[0]
+
+    def test_exception_during_upgrade_returns_failure(self, driver):
+        driver._pkg_manager = "apt"
+        driver._sudo_password = "secret"  # noqa: S105
+        with (
+            patch.object(driver, "_send", return_value=""),
+            patch.object(
+                driver, "_sudo", side_effect=["update ok", TimeoutError("connection lost")]
+            ),
+        ):
+            result = driver._action_apt_update_upgrade()
+
+        assert result["success"] is False
+        assert "[error]" in result["output"]
+
+
+class TestRunDeviceActionDispatch:
+    def test_apt_update_upgrade_action_dispatches(self, driver):
+        with patch.object(
+            driver, "_action_apt_update_upgrade", return_value={"success": True, "output": ""}
+        ) 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()