Merge pull request 'fix: decide uninstall success by exit status, not by keywords' (#2) from fix/uninstall-exit-status into master

uninstall_package judged success by searching apt/dnf/apk/pacman output
for failure words. That is guesswork in both directions: apt's commonest
failure ("E: Sub-process /usr/bin/dpkg returned an error code (1)") read
as success until the previous change, and a prerm that prints "Failed to
stop ..." while the removal completes still reads as failure. The exit
status is the answer the package manager actually gives, but every
command went through `_sudo(... || true)`, which throws it away.

Add `_sudo_status()`, which runs the command via `_sudo` followed by
`; echo __NETORK_RC=$?` and returns `(output, exit_status)` with the
marker stripped. The `|| true` of other `_sudo` callers is untouched:
they still want output rather than a status. The marker is matched only
on a line of its own with digits, so an echoed command line (literal
`$?`) is never mistaken for it. If the marker never arrives the status
is None -- unknown, not success.

uninstall_package and its dpkg fallback now use it, and
`_uninstall_failed(output, rc)` lets rc decide whenever it is known,
falling back to the keyword check only when it is not.

Behaviour change worth knowing: removing a package that is not installed
exits 0 on apt (and dnf), so it now reports success where the keyword
"is not installed" used to report failure. The package is absent
afterwards, which is what the caller asked for, and netOrk dropping it
from the installed record is then correct.

Refs christianmanivong/netork#267

napalm_linux/linux.py

@@ -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,6 +122,22 @@ def _arm_vendor_from_model(model: str) -> str:
     return " ".join(brand)
 
 
+#: 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(OSDriver):
     """NAPALM driver for generic Linux systems.
 
@@ -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:
@@ -810,17 +859,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 +896,8 @@ class LinuxDriver(OSDriver):
                 "description": description,
                 "size": size,
                 "source": "apt",
+                "source_package": source_package,
+                "source_version": source_version,
             })
         return packages
 
@@ -1000,26 +1070,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 +1164,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.
 
@@ -1529,7 +1657,10 @@ class LinuxDriver(OSDriver):
             "echo '---VOLUMES---'; "
             f"{docker} volume ls --format '{{{{json .}}}}' 2>/dev/null; "
             "echo '---NETWORKS---'; "
-            f"{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,
         )
 
@@ -1548,7 +1679,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."""
@@ -1609,6 +1741,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():
@@ -1666,11 +1836,23 @@ 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(
@@ -1702,6 +1884,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)

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,7 +717,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
 
@@ -853,3 +912,236 @@ class TestDockerBinHook:
         assert docker_cmds
         for cmd in docker_cmds:
             assert "/opt/cs/docker" in cmd, f"unconverted call site: {cmd}"
+
+
+# ---------------------------------------------------------------------------
+# uninstall_package – purge, and getting out of `install ok unpacked`
+# ---------------------------------------------------------------------------
+
+
+class TestUninstallPackage:
+    """Removing a package that does not want to go.
+
+    Both cases here were found during a fleet-wide Wazuh rollback. Of thirteen
+    hosts carrying the agent, seven sat at `install ok unpacked` with the unit
+    failed — an upgrade whose postinst could not reach a manager that no longer
+    existed. `apt-get remove` cannot help there: apt configures a package before
+    removing it, and configuring is exactly what was broken.
+
+    And `remove` leaves the configuration behind by design, which for the Wazuh
+    agent means its apt source keeps being fetched on every update, long after
+    the package is gone.
+    """
+
+    def test_remove_is_still_the_default(self, driver):
+        """Callers that did not ask for a purge must not get one: configuration
+        somebody may want back is not this function's to delete."""
+        _mock_send(driver, "Removing wazuh-agent ...")
+
+        driver.uninstall_package("wazuh-agent")
+
+        sent = driver._device.send_command.call_args[0][0]
+        assert "apt-get remove" in sent
+        assert "purge" not in sent
+
+    def test_purge_is_asked_for_explicitly(self, driver):
+        _mock_send(driver, "Purging configuration files for wazuh-agent ...")
+
+        driver.uninstall_package("wazuh-agent", purge=True)
+
+        assert "apt-get purge" in driver._device.send_command.call_args[0][0]
+
+    def test_a_half_configured_package_falls_back_to_dpkg(self, driver):
+        """`install ok unpacked` is the state apt cannot get out of. On one host
+        only `dpkg --purge --force-all` removed it."""
+        driver._device.send_command.side_effect = [
+            "E: Sub-process /usr/bin/dpkg returned an error code (1)",
+            "Removing wazuh-agent (4.14.7-1) ...",
+        ]
+
+        result = driver.uninstall_package("wazuh-agent", purge=True)
+
+        assert result["success"] is True
+        second = driver._device.send_command.call_args_list[1][0][0]
+        assert "dpkg --purge --force-all" in second
+
+    def test_the_fallback_is_not_tried_when_the_first_pass_worked(self, driver):
+        """A forced dpkg purge is a bigger hammer than apt and must stay a last
+        resort, not a routine second step."""
+        _mock_send(driver, "Removing wazuh-agent ...")
+
+        driver.uninstall_package("wazuh-agent", purge=True)
+
+        assert driver._device.send_command.call_count == 1
+
+    def test_a_package_manager_without_purge_still_removes(self, driver):
+        """apk and pacman have no separate purge; asking for one must not turn
+        into a failure or a command they do not understand."""
+        driver._pkg_manager = "apk"
+        _mock_send(driver, "(1/1) Purging wazuh-agent")
+
+        result = driver.uninstall_package("wazuh-agent", purge=True)
+
+        assert result["success"] is True
+        assert "apk del" in driver._device.send_command.call_args[0][0]
+
+
+# ---------------------------------------------------------------------------
+# uninstall_package – success from the exit status, not from prose (netork#267)
+# ---------------------------------------------------------------------------
+
+
+def _with_rc(output: str, rc: int) -> str:
+    """What the shell prints for a command run through ``_sudo_status``."""
+    return f"{output}\n__NETORK_RC={rc}"
+
+
+class TestSudoStatus:
+    """``_sudo_status`` keeps the exit status that ``|| true`` throws away."""
+
+    def test_returns_output_and_exit_status(self, driver):
+        _mock_send(driver, _with_rc("Removing wazuh-agent ...", 0))
+
+        assert driver._sudo_status("apt-get remove -y wazuh-agent") == (
+            "Removing wazuh-agent ...",
+            0,
+        )
+
+    def test_a_non_zero_exit_status_is_reported(self, driver):
+        _mock_send(driver, _with_rc("E: Unable to locate package nope", 100))
+
+        assert driver._sudo_status("apt-get remove -y nope")[1] == 100
+
+    def test_the_status_is_read_right_after_sudo_returns(self, driver):
+        """``$?`` must be read straight after the sudo pipeline — with an
+        ``|| true`` in between, every command would report 0."""
+        driver._sudo_password = "pw"  # noqa: S105
+        _mock_send(driver, _with_rc("", 0))
+
+        driver._sudo_status("apt-get remove -y x 2>&1")
+
+        sent = driver._device.send_command.call_args[0][0]
+        assert sent.startswith("echo pw | sudo -S")
+        assert sent.endswith("apt-get remove -y x 2>&1; echo __NETORK_RC=$?")
+        assert "|| true" not in sent
+
+    def test_a_missing_marker_means_unknown_not_success(self, driver):
+        """Output cut short before the marker arrived says nothing about the
+        exit status; ``None`` says so instead of guessing 0."""
+        _mock_send(driver, "Removing wazuh-agent ...")
+
+        assert driver._sudo_status("apt-get remove -y wazuh-agent") == (
+            "Removing wazuh-agent ...",
+            None,
+        )
+
+    def test_the_command_echo_is_not_mistaken_for_the_marker(self, driver):
+        """A terminal may echo the command line back; its literal ``$?`` is not
+        a number, and only the marker on a line of its own counts."""
+        _mock_send(
+            driver,
+            "sudo apt-get remove -y x; echo __NETORK_RC=$?\nRemoving x ...\n__NETORK_RC=1",
+        )
+
+        output, rc = driver._sudo_status("apt-get remove -y x")
+
+        assert rc == 1
+        assert "__NETORK_RC=1" not in output
+
+
+class TestUninstallExitStatus:
+    """Whether a removal worked is what the package manager's exit status says.
+
+    Reading it out of human-readable output was guesswork in both directions:
+    apt's commonest failure (``E: Sub-process /usr/bin/dpkg returned an error
+    code (1)``) read as success until #240, and a successful removal whose
+    prerm merely *mentions* a failure read as a failure.
+    """
+
+    def test_a_non_zero_exit_is_a_failure_whatever_the_output_says(self, driver):
+        """Nothing in this output matches a failure keyword; only the exit
+        status knows."""
+        driver._pkg_manager = "dnf"
+        _mock_send(driver, _with_rc("Removing: wazuh-agent", 1))
+
+        result = driver.uninstall_package("wazuh-agent")
+
+        assert result["success"] is False
+
+    def test_a_zero_exit_is_a_success_even_if_the_output_mentions_failure(self, driver):
+        """A prerm that cannot stop an already-dead unit prints "Failed" and
+        still lets the removal complete."""
+        _mock_send(
+            driver,
+            _with_rc(
+                "Removing wazuh-agent (4.14.7-1) ...\n"
+                "Failed to stop wazuh-agent.service: Unit wazuh-agent.service not loaded.",
+                0,
+            ),
+        )
+
+        result = driver.uninstall_package("wazuh-agent")
+
+        assert result["success"] is True
+
+    def test_the_marker_does_not_reach_the_caller(self, driver):
+        _mock_send(driver, _with_rc("Removing wazuh-agent ...", 0))
+
+        result = driver.uninstall_package("wazuh-agent")
+
+        assert result["output"] == "Removing wazuh-agent ..."
+
+    def test_the_uninstall_command_keeps_its_exit_status(self, driver):
+        _mock_send(driver, _with_rc("Removing wazuh-agent ...", 0))
+
+        driver.uninstall_package("wazuh-agent")
+
+        sent = driver._device.send_command.call_args[0][0]
+        assert "|| true" not in sent
+        assert sent.endswith("; echo __NETORK_RC=$?")
+
+    def test_apt_failing_by_exit_status_falls_back_to_dpkg(self, driver):
+        driver._device.send_command.side_effect = [
+            _with_rc("E: Sub-process /usr/bin/dpkg returned an error code (1)", 100),
+            _with_rc("Removing wazuh-agent (4.14.7-1) ...", 0),
+        ]
+
+        result = driver.uninstall_package("wazuh-agent", purge=True)
+
+        assert result["success"] is True
+        second = driver._device.send_command.call_args_list[1][0][0]
+        assert "dpkg --purge --force-all" in second
+        assert "|| true" not in second
+        assert "__NETORK_RC" not in result["output"]
+
+    def test_the_dpkg_fallback_failing_is_a_failure(self, driver):
+        driver._device.send_command.side_effect = [
+            _with_rc("E: Sub-process /usr/bin/dpkg returned an error code (1)", 100),
+            _with_rc("dpkg: error processing package wazuh-agent (--purge):", 1),
+        ]
+
+        result = driver.uninstall_package("wazuh-agent", purge=True)
+
+        assert result["success"] is False
+        assert "dpkg --purge --force-all" in result["output"]
+
+    def test_a_zero_exit_does_not_trigger_the_fallback(self, driver):
+        """Even when the output contains words that used to mean failure: apt
+        exits 0 for a package that is already gone, which is the state the
+        caller asked for."""
+        _mock_send(driver, _with_rc("Package 'x' is not installed, so not removed", 0))
+
+        result = driver.uninstall_package("x", purge=True)
+
+        assert result["success"] is True
+        assert driver._device.send_command.call_count == 1
+
+    def test_without_an_exit_status_the_output_is_read_as_before(self, driver):
+        """If the marker never arrived, the keyword check is still the best
+        answer available — and it errs towards failure on apt's ``E:``."""
+        driver._pkg_manager = "dnf"
+        _mock_send(driver, "E: Sub-process /usr/bin/dpkg returned an error code (1)")
+
+        result = driver.uninstall_package("wazuh-agent")
+
+        assert result["success"] is False