Merge pull request 'refactor!: drop the Docker CLI methods' (#20) from chore/drop-legacy-docker into master

get_docker_info(), get_docker_outdated(), reconstruct_docker_run() and
the _docker_bin() hook go, with the image-ID helpers. netOrk reads and
handles containers over the Engine API through open_container_engine()
since NetOrk/netork#765 and no longer calls them (phase 9). The
fix_docker_permissions action stays: it is an OS action.

Version 0.3.0.

.github/workflows/ci.yml

@@ -0,0 +1,48 @@
+name: CI
+
+on:
+  push:
+    branches: ["**"]
+  pull_request:
+    branches: ["**"]
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+    strategy:
+      fail-fast: false
+      matrix:
+        python-version: ["3.10", "3.11", "3.12"]
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v4
+
+      - name: Setup Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: ${{ matrix.python-version }}
+          cache: pip
+
+      - name: Install package with dev extras
+        run: |
+          python -m pip install --upgrade pip
+          # napalm-device-types lives in git.netork.io/NAPALM, not on PyPI: without this
+          # pip looks there, finds an unrelated 0.1.0 and the job dies before any test.
+          python -m pip install "napalm-device-types @ git+https://git.netork.io/NAPALM/napalm-device-types.git"
+          python -m pip install -e ".[dev]"
+
+      - name: Run unit tests
+        run: |
+          python -m pytest -q --tb=short
+
+      - name: Build wheel and sdist
+        run: |
+          python -m pip install build
+          python -m build
+
+      - name: Upload dist artifacts
+        # v4 refuses to run on Gitea ("not currently supported on GHES").
+        uses: actions/upload-artifact@v3
+        with:
+          name: dist-${{ matrix.python-version }}
+          path: dist/*

CHANGELOG.md

@@ -7,6 +7,34 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+## [0.3.0] – 2026-10-08
+
+### Removed
+
+- `get_docker_info()`, `get_docker_outdated()`, `reconstruct_docker_run()` and
+  the `_docker_bin()` hook, with the image-ID helpers they used. netOrk reads
+  and handles containers itself over the Engine API, through
+  `open_container_engine()` (NetOrk/netork#765), and no longer calls them.
+  `run_device_action("fix_docker_permissions")` stays: letting the login user
+  use Docker is an OS action.
+
+## [0.2.0] – 2026-10-07
+
+### Added
+
+- `run_command()` and `open_stream()`, the public command channel from
+  napalm-device-types 2.6.0: an exec channel on the existing SSH transport, so
+  no PTY, separate stderr and a real exit code. `privileged=True` runs as root
+  directly, through `sudo -S` with the password on stdin (never on the command
+  line), or through `sudo -n`, which fails instead of prompting.
+- `ContainerEngineMixin`: `container_engines()` and `open_container_engine()`,
+  whose `open_api()` streams the Docker Engine API over `docker system
+  dial-stdio` (NetOrk/netork#765). The existing Docker methods are unchanged.
+
+### Changed
+
+- Requires napalm-device-types >= 2.6.0.
+
 ## [0.1.0] – 2026-05-29
 
 ### Added

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,12 +31,29 @@ 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,
+    ByteStream,
+    CommandResult,
+    ContainerEngineMixin,
+    FingerprintRule,
+    HostStatusMixin,
+    KernelFactsMixin,
+    ListeningSocketsMixin,
+    OSDriver,
+    SystemdServicesMixin,
+    SystemdUnavailable,
+    parse_apt_upgradable,
+    open_stream_on_transport,
+    parse_dnf_security,
+    run_on_transport,
+    strip_terminal_codes,
+)
 from napalm_device_types.models import (
     ApplyUpdatesResultDict,
     CronJobDict,
     DeviceActionResultDict,
-    DockerInfoDict,
     PackageDict,
     ProcessDict,
     ServiceDict,
@@ -50,6 +67,76 @@ 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)
+
+
+#: The end of a command's output when nothing better is known: a line that looks
+#: like a shell prompt.
+_PROMPT_RE = r"[#$\>]\s*$"
+#: A command ending in ``echo __NAME=$?`` reports its exit status on a line of its
+#: own: ``_RC_MARKER`` here, ``__APT_RC=`` and ``__SVC_RC=`` in napalm-device-types.
+_STATUS_ECHO_RE = re.compile(r"echo\s+(__[A-Z_]+=)\$\?")
+
+
+def _expect_for(command: str) -> str:
+    """The pattern that ends *command*'s output.
+
+    netmiko stops reading as soon as the pattern matches what it has read so far.
+    A line the command prints can end in ``#``, ``$`` or ``>`` -- apt's
+    ``<ftpmaster@ubuntu.com>`` after a bad signature -- and was taken for the
+    prompt: half the output came back, and the rest started the next command's
+    (#615). A command that echoes its exit status is read until that marker, with
+    a number, and the prompt line after it. The echoed command line carries a
+    literal ``$?`` and cannot match.
+    """
+    markers = _STATUS_ECHO_RE.findall(command)
+    if not markers:
+        return _PROMPT_RE
+    return re.escape(markers[-1]) + r"\d+\s*\n.*" + _PROMPT_RE
+
+
+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"
+
+#: Restart the host two seconds later, detached from this session: the launcher's
+#: exit status comes back before the host goes down, and closing the session
+#: cannot take the restart with it.
+_REBOOT_DETACHED = "sh -c '(trap \"\" HUP; sleep 2; /sbin/reboot) </dev/null >/dev/null 2>&1 &'"
+
+#: 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 +203,14 @@ def _arm_vendor_from_model(model: str) -> str:
     return " ".join(brand)
 
 
-class LinuxDriver(OSDriver):
+class LinuxDriver(
+    KernelFactsMixin,
+    ListeningSocketsMixin,
+    SystemdServicesMixin,
+    HostStatusMixin,
+    ContainerEngineMixin,
+    OSDriver,
+):
     """NAPALM driver for generic Linux systems.
 
     Connects via SSH (netmiko ``linux`` device type) and auto-detects the
@@ -126,6 +220,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),
     ]
@@ -235,7 +332,7 @@ class LinuxDriver(OSDriver):
             command,
             read_timeout=read_timeout,
             cmd_verify=False,
-            expect_string=r'[#$\>]\s*$',
+            expect_string=_expect_for(command),
         ).strip()
 
     def _sudo(self, command: str, read_timeout: float = 100) -> str:
@@ -248,6 +345,114 @@ 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)
+
+    # ------------------------------------------------------------------
+    # Command channel (napalm-device-types CommandChannelMixin)
+    # ------------------------------------------------------------------
+
+    def _transport(self) -> Any:
+        """The SSH transport netmiko already holds, for exec channels next to its PTY."""
+        if not self._device:
+            raise ConnectionClosedException("Not connected")
+        return self._device.remote_conn_pre.get_transport()
+
+    def _privileged(self, command: str, privileged: bool) -> tuple[str, bytes | None]:
+        """The command line that runs *command* with the privileges asked for,
+        and what has to reach its stdin first.
+
+        Root runs it directly. With a sudo password, ``sudo -S`` reads it from
+        stdin, so it never appears in a process list. Without one, ``sudo -n``
+        fails at once where a prompt would hang. The command goes to ``sh -c``
+        as one argument, so the privilege covers every part of it.
+        """
+        if not privileged or self._is_root():
+            return command, None
+        if self._sudo_password:
+            line = f"sudo -S -p '' sh -c {_shlex_quote(command)}"
+            return line, f"{self._sudo_password}\n".encode()
+        return f"sudo -n sh -c {_shlex_quote(command)}", None
+
+    def run_command(
+        self,
+        command: str,
+        *,
+        privileged: bool = False,
+        timeout: float = 60,
+        stdin: bytes | None = None,
+    ) -> CommandResult:
+        """Run *command* on an exec channel: no PTY, stderr apart, a real exit code."""
+        line, prefix = self._privileged(command, privileged)
+        data = (prefix or b"") + (stdin or b"") if (prefix or stdin) else None
+        return run_on_transport(self._transport(), line, stdin=data, timeout=timeout)
+
+    def open_stream(self, command: str, *, privileged: bool = False) -> ByteStream:
+        """Start *command* on an exec channel and return a stream to it."""
+        line, prefix = self._privileged(command, privileged)
+        return open_stream_on_transport(self._transport(), line, stdin_prefix=prefix)
+
+    def reboot_host(self) -> None:
+        """Restart the host (``HostRebootMixin``); returns once the restart is under way.
+
+        :raises RuntimeError: when the host refuses -- sudo without a password,
+            no ``reboot`` -- or its answer carried no exit status.
+        """
+        output, status = _split_status(
+            self._run_privileged(f"{_REBOOT_DETACHED}; echo {_RC_MARKER}$?", 30)
+        )
+        if status != 0:
+            reason = output or f"the reboot command exited with status {status}"
+            if "password is required" in output:
+                reason = f"{reason}\n{_SUDO_PASSWORD_HINT}"
+            raise RuntimeError(reason)
+
+    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 +586,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 +599,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 +1001,27 @@ 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)
+
+    # ------------------------------------------------------------------
+    # ListeningSocketsMixin – the transport for get_listening_sockets
+    # ------------------------------------------------------------------
+
+    def _run_listening_sockets_command(self, command: str, *, privileged: bool) -> str:
+        """The transport for ``ListeningSocketsMixin.get_listening_sockets``.
+
+        Read-only either way; root only so that ``ss`` names every process.
+        """
+        if privileged:
+            return self._run_privileged(command, 60)
+        return self._send(command, read_timeout=60)
+
     # ------------------------------------------------------------------
     # OSDriver – package management
     # ------------------------------------------------------------------
@@ -810,17 +1041,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 +1078,8 @@ class LinuxDriver(OSDriver):
                 "description": description,
                 "size": size,
                 "source": "apt",
+                "source_package": source_package,
+                "source_version": source_version,
             })
         return packages
 
@@ -1000,26 +1252,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 +1346,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 +1362,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 +1373,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 +1382,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 +1554,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")
@@ -1480,349 +1762,6 @@ class LinuxDriver(OSDriver):
             job["description"] = comment
         return job
 
-    # ------------------------------------------------------------------
-    # Docker
-    # ------------------------------------------------------------------
-
-    def get_docker_info(self) -> DockerInfoDict:
-        """Return information about the local Docker environment.
-
-        Uses a single SSH call to collect all Docker data at once, eliminating
-        per-section round-trip overhead.  Labels from ``docker images`` are used
-        directly for the OCI version field — no separate ``docker image inspect``
-        needed.
-
-        Returns a dict with keys:
-          - ``available`` (bool)  — False if docker is not installed/accessible
-          - ``version``   (str)   — Docker Engine version string
-          - ``containers`` (list) — list of container dicts
-          - ``images``    (list)  — list of image dicts
-          - ``volumes``   (list)  — list of volume dicts
-          - ``networks``  (list)  — list of network dicts
-        """
-        import json as _json
-
-        # Check docker binary first (docker --version doesn't need socket access)
-        if not self._send("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")
-        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()
-
-        combined = self._send(
-            "echo '---CONTAINERS---'; "
-            "docker ps -a --format '{{json .}}' 2>/dev/null; "
-            "echo '---IMAGES---'; "
-            "docker images --format '{{json .}}' 2>/dev/null; "
-            "echo '---VOLUMES---'; "
-            "docker volume ls --format '{{json .}}' 2>/dev/null; "
-            "echo '---NETWORKS---'; "
-            "docker network ls --format '{{json .}}' 2>/dev/null",
-            read_timeout=60,
-        )
-
-        if "---CONTAINERS---" not in combined:
-            return {"available": False}
-
-        # Split into sections
-        def _section(text: str, marker: str, next_marker: str) -> str:
-            start = text.find(marker)
-            if start == -1:
-                return ""
-            start += len(marker)
-            end = text.find(next_marker, start)
-            return text[start:end] if end != -1 else text[start:]
-
-        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
-
-        def _parse_labels(raw: Any) -> Dict[str, str]:
-            """Parse Docker labels — may be a dict (JSON map) or comma-sep string."""
-            if isinstance(raw, dict):
-                return {str(k): str(v) for k, v in raw.items()}
-            if isinstance(raw, str) and raw:
-                result: Dict[str, str] = {}
-                for part in raw.split(","):
-                    if "=" in part:
-                        k, _, v = part.partition("=")
-                        result[k.strip()] = v.strip()
-                return result
-            return {}
-
-        # Containers
-        containers: List[dict[str, Any]] = []
-        for line in raw_containers.splitlines():
-            line = line.strip()
-            if not line:
-                continue
-            try:
-                obj = _json.loads(line)
-                labels = _parse_labels(obj.get("Labels", ""))
-                containers.append({
-                    "id": obj.get("ID", ""),
-                    "name": obj.get("Names", ""),
-                    "image": obj.get("Image", ""),
-                    "image_version": labels.get("org.opencontainers.image.version", ""),
-                    "command": obj.get("Command", ""),
-                    "created": obj.get("CreatedAt", ""),
-                    "status": obj.get("Status", ""),
-                    "ports": obj.get("Ports", ""),
-                    "state": obj.get("State", ""),
-                    "compose_project": labels.get("com.docker.compose.project", ""),
-                    "compose_service": labels.get("com.docker.compose.service", ""),
-                    "compose_file": labels.get("com.docker.compose.project.config_files", ""),
-                })
-            except Exception:
-                pass
-
-        # Images — OCI version comes from Labels, no separate inspect needed
-        images: List[dict[str, Any]] = []
-        for line in raw_images.splitlines():
-            line = line.strip()
-            if not line:
-                continue
-            try:
-                obj = _json.loads(line)
-                labels = _parse_labels(obj.get("Labels", ""))
-                images.append({
-                    "id": obj.get("ID", ""),
-                    "repository": obj.get("Repository", ""),
-                    "tag": obj.get("Tag", ""),
-                    "size": obj.get("Size", ""),
-                    "created": obj.get("CreatedAt", ""),
-                    "version": labels.get("org.opencontainers.image.version", ""),
-                })
-            except Exception:
-                pass
-
-        # Volumes
-        volumes: List[dict[str, Any]] = []
-        for line in raw_volumes.splitlines():
-            line = line.strip()
-            if not line:
-                continue
-            try:
-                obj = _json.loads(line)
-                volumes.append({
-                    "name": obj.get("Name", ""),
-                    "driver": obj.get("Driver", ""),
-                    "mountpoint": obj.get("Mountpoint", ""),
-                    "scope": obj.get("Scope", ""),
-                })
-            except Exception:
-                pass
-
-        # Networks
-        networks: List[dict[str, Any]] = []
-        for line in raw_networks.splitlines():
-            line = line.strip()
-            if not line:
-                continue
-            try:
-                obj = _json.loads(line)
-                networks.append({
-                    "id": obj.get("ID", ""),
-                    "name": obj.get("Name", ""),
-                    "driver": obj.get("Driver", ""),
-                    "scope": obj.get("Scope", ""),
-                    "ipv6": obj.get("IPv6", ""),
-                    "internal": obj.get("Internal", ""),
-                })
-            except Exception:
-                pass
-
-        return {
-            "available": True,
-            "version": version,
-            "containers": containers,
-            "images": images,
-            "volumes": volumes,
-            "networks": networks,
-            "outdated_images": [],  # populated by separate check_docker_outdated task
-        }
-
-    def get_docker_outdated(self, containers: List[Dict]) -> List[str]:
-        """Check registry for available updates for all container images.
-
-        Runs ``docker buildx imagetools inspect`` (metadata-only, no download)
-        for each unique image referenced by a container.  Intended to be called
-        from a separate Celery task on a long interval (e.g. every 3 hours) so
-        it never blocks the main device poll.
-
-        Returns a list of image references that have a newer digest available.
-        """
-        outdated_images: List[str] = []
-        candidate_images: List[str] = list({
-            c["image"] for c in containers
-            if c.get("image")
-            and "@sha256:" not in c.get("image", "")  # skip digest-pinned
-        })
-        for img_name in candidate_images:
-            try:
-                local_raw = self._send(
-                    f"docker inspect {img_name!r} --format '{{{{index .RepoDigests 0}}}}' 2>/dev/null",
-                    read_timeout=5,
-                ).strip()
-                if not local_raw or "@" not in local_raw:
-                    continue  # locally built or not yet pulled
-                local_digest = local_raw.split("@", 1)[1]
-
-                remote_full = self._send(
-                    f"docker buildx imagetools inspect {img_name!r} 2>&1",
-                    read_timeout=30,
-                ).strip()
-                if ("429" in remote_full
-                        or "Too Many Requests" in remote_full
-                        or "toomanyrequests" in remote_full):
-                    logger.warning(
-                        "Docker Hub rate limit hit for %s — run "
-                        "'docker login' on the device to avoid this",
-                        img_name,
-                    )
-                    continue
-                remote_digest = ""
-                for _line in remote_full.splitlines():
-                    _ls = _line.strip()
-                    if _ls.startswith("Digest:"):
-                        remote_digest = _ls[7:].strip()
-                        break
-                if not remote_digest or not remote_digest.startswith("sha256:"):
-                    continue
-                if local_digest != remote_digest:
-                    outdated_images.append(img_name)
-            except Exception as exc:
-                logger.warning("image update check for %s: %s", img_name, exc)
-        return outdated_images
-
-    def reconstruct_docker_run(self, container_id: str) -> dict | None:
-        """Return the information needed to recreate a standalone container.
-
-        Parses ``docker inspect`` JSON and returns a dict with:
-        - ``name``      — container name (without leading slash)
-        - ``image``     — current image reference
-        - ``run_args``  — list of CLI args for ``docker run`` (without image/cmd)
-        - ``cmd``       — command override (may be empty list)
-        - ``entrypoint`` — entrypoint override (may be empty list)
-
-        Returns None if the container does not exist or inspect fails.
-        """
-        import json as _json
-        import shlex as _shlex
-
-        raw = self._send(
-            f"docker inspect {_shlex.quote(container_id)} 2>/dev/null",
-            read_timeout=10,
-        ).strip()
-        if not raw:
-            return None
-        try:
-            data = _json.loads(raw)
-        except Exception:
-            return None
-        if not data:
-            return None
-        c = data[0]
-
-        name = c.get("Name", "").lstrip("/")
-        cfg = c.get("Config", {})
-        hcfg = c.get("HostConfig", {})
-        net_settings = c.get("NetworkSettings", {})
-
-        args: List[str] = ["--name", name]
-
-        # Restart policy
-        rp = hcfg.get("RestartPolicy", {})
-        rp_name = rp.get("Name", "no")
-        if rp_name and rp_name != "no":
-            max_retry = rp.get("MaximumRetryCount", 0)
-            if rp_name == "on-failure" and max_retry:
-                args += ["--restart", f"on-failure:{max_retry}"]
-            else:
-                args += ["--restart", rp_name]
-
-        # Hostname
-        hostname = cfg.get("Hostname", "")
-        if hostname and hostname != name[:12]:
-            args += ["--hostname", hostname]
-
-        # Environment (skip vars that look like Docker-injected metadata)
-        _skip_prefixes = ("PATH=", "HOME=", "TERM=", "HOSTNAME=")
-        for env in cfg.get("Env") or []:
-            if not any(env.startswith(p) for p in _skip_prefixes):
-                args += ["-e", env]
-
-        # Volume binds
-        for bind in hcfg.get("Binds") or []:
-            args += ["-v", bind]
-
-        # Port bindings
-        for container_port, host_bindings in (hcfg.get("PortBindings") or {}).items():
-            for hb in (host_bindings or []):
-                host_ip = hb.get("HostIp", "")
-                host_port = hb.get("HostPort", "")
-                if host_ip:
-                    args += ["-p", f"{host_ip}:{host_port}:{container_port}"]
-                else:
-                    args += ["-p", f"{host_port}:{container_port}"]
-
-        # Network mode
-        net_mode = hcfg.get("NetworkMode", "default")
-        if net_mode not in ("default", "bridge"):
-            args += ["--network", net_mode]
-        else:
-            # Check for custom networks from NetworkSettings
-            for net_name in (net_settings.get("Networks") or {}):
-                if net_name not in ("bridge", "host", "none"):
-                    args += ["--network", net_name]
-                    break
-
-        # Privileged
-        if hcfg.get("Privileged"):
-            args.append("--privileged")
-
-        # Cap-add
-        for cap in hcfg.get("CapAdd") or []:
-            args += ["--cap-add", cap]
-
-        # Devices
-        for dev in hcfg.get("Devices") or []:
-            host_p = dev.get("PathOnHost", "")
-            ctr_p = dev.get("PathInContainer", "")
-            perms = dev.get("CgroupPermissions", "rwm")
-            if host_p:
-                args += ["--device", f"{host_p}:{ctr_p}:{perms}"]
-
-        # Extra hosts
-        for eh in hcfg.get("ExtraHosts") or []:
-            args += ["--add-host", eh]
-
-        # DNS
-        for dns in hcfg.get("Dns") or []:
-            args += ["--dns", dns]
-
-        # Labels (skip Docker-internal labels)
-        _skip_label_prefixes = ("com.docker.compose.", "org.opencontainers.")
-        for k, v in (cfg.get("Labels") or {}).items():
-            if not any(k.startswith(p) for p in _skip_label_prefixes):
-                args += ["--label", f"{k}={v}"]
-
-        # Detach always
-        args.append("-d")
-
-        return {
-            "name": name,
-            "image": cfg.get("Image", ""),
-            "run_args": args,
-            "cmd": cfg.get("Cmd") or [],
-            "entrypoint": cfg.get("Entrypoint") or [],
-        }
-
     # ── Device actions ────────────────────────────────────────────────────────
 
     def get_snmp_config(self) -> Optional[SNMPConfigDict]:
@@ -1906,11 +1845,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 +1871,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

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "napalm-linux"
-version = "0.1.0"
+version = "0.3.0"
 description = "NAPALM driver for generic Linux systems via SSH"
 readme = "README.md"
 requires-python = ">=3.9"
@@ -37,7 +37,7 @@ classifiers = [
 ]
 dependencies = [
     "napalm>=4.0",
-    "napalm-device-types>=0.3.0",
+    "napalm-device-types>=2.6.0",
     "netmiko>=4.0.0",
     "paramiko>=5.0.0",  # CVE-2026-44405
 ]

tests/test_command_channel.py

@@ -0,0 +1,140 @@
+"""The public command channel and container engine access (napalm-device-types 2.6.0).
+
+netOrk used to reach a Linux host's shell through the private ``_send``: an
+interactive PTY, stdout and stderr merged, no exit code. ``run_command`` and
+``open_stream`` go through an exec channel on the same SSH transport instead,
+and ``open_container_engine`` builds on them (NetOrk/netork#765). Privileges
+work as they do everywhere else in this driver: root runs directly, a sudo
+password goes to ``sudo -S`` on stdin and never onto a command line, and
+without one ``sudo -n`` fails at once instead of hanging.
+"""
+
+from __future__ import annotations
+
+from types import SimpleNamespace
+
+import pytest
+from napalm.base.exceptions import ConnectionClosedException
+
+from napalm_linux import LinuxDriver
+
+
+class FakeChannel:
+    def __init__(self):
+        self.command = None
+        self.sent = b""
+
+    def exec_command(self, command):
+        self.command = command
+
+    def settimeout(self, timeout):
+        pass
+
+    def sendall(self, data):
+        self.sent += data
+
+    def shutdown_write(self):
+        pass
+
+    def close(self):
+        pass
+
+    def recv_ready(self):
+        return False
+
+    def recv_stderr_ready(self):
+        return False
+
+    def exit_status_ready(self):
+        return True
+
+    def recv_exit_status(self):
+        return 0
+
+
+class FakeTransport:
+    def __init__(self):
+        self.channels = []
+
+    def open_session(self):
+        self.channels.append(FakeChannel())
+        return self.channels[-1]
+
+
+def _driver(*, root=False, sudo_password=None):
+    driver = LinuxDriver("h", "u", "p", optional_args={"sudo_password": sudo_password})
+    transport = FakeTransport()
+    driver._device = SimpleNamespace(remote_conn_pre=SimpleNamespace(get_transport=lambda: transport))
+    driver._root = root
+    return driver, transport
+
+
+def test_an_unprivileged_command_runs_as_given():
+    driver, transport = _driver()
+
+    result = driver.run_command("docker version", timeout=5)
+
+    assert transport.channels[-1].command == "docker version"
+    assert transport.channels[-1].sent == b""
+    assert result.exit_code == 0
+
+
+def test_a_privileged_command_with_a_sudo_password_reads_it_from_stdin():
+    driver, transport = _driver(sudo_password="s3cr3t")
+
+    driver.run_command("usermod -aG docker u", privileged=True, timeout=5)
+
+    channel = transport.channels[-1]
+    assert channel.command == "sudo -S -p '' sh -c 'usermod -aG docker u'"
+    assert channel.sent == b"s3cr3t\n"
+    assert "s3cr3t" not in channel.command
+
+
+def test_a_privileged_command_without_a_password_fails_fast_instead_of_prompting():
+    driver, transport = _driver()
+
+    driver.run_command("id", privileged=True, timeout=5)
+
+    assert transport.channels[-1].command == "sudo -n sh -c id"
+
+
+def test_a_root_login_needs_no_sudo():
+    driver, transport = _driver(root=True, sudo_password="s3cr3t")
+
+    driver.run_command("id", privileged=True, timeout=5)
+
+    assert transport.channels[-1].command == "id"
+    assert transport.channels[-1].sent == b""
+
+
+def test_stdin_follows_the_sudo_password():
+    driver, transport = _driver(sudo_password="pw")
+
+    driver.run_command("tee /etc/x", privileged=True, stdin=b"data", timeout=5)
+
+    assert transport.channels[-1].sent == b"pw\ndata"
+
+
+def test_a_privileged_stream_gets_the_password_first():
+    driver, transport = _driver(sudo_password="pw")
+
+    driver.open_stream("cat > /tmp/x", privileged=True)
+
+    channel = transport.channels[-1]
+    assert channel.command == "sudo -S -p '' sh -c 'cat > /tmp/x'"
+    assert channel.sent == b"pw\n"
+
+
+def test_the_container_engine_api_is_a_stream_over_dial_stdio():
+    driver, transport = _driver()
+
+    driver.open_container_engine("docker").open_api()
+
+    assert transport.channels[-1].command == "docker system dial-stdio"
+
+
+def test_the_channel_needs_an_open_connection():
+    driver = LinuxDriver("h", "u", "p")
+
+    with pytest.raises(ConnectionClosedException):
+        driver.run_command("true")

tests/test_linux.py

@@ -1,5 +1,7 @@
 """Unit tests for LinuxDriver – parsing helpers (no real SSH connection needed)."""
 
+import re
+
 import pytest
 from unittest.mock import MagicMock, patch
 from napalm_linux.linux import LinuxDriver, _arm_vendor_from_model
@@ -22,6 +24,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 +155,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 +178,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 +228,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 +367,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 +689,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 +708,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 +720,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 +864,714 @@ class TestRunDeviceActionDispatch:
         ) as mock_action:
             driver.run_device_action("apt_update_upgrade")
         mock_action.assert_called_once()
+
+
+class TestNoDockerOfItsOwn:
+    """Containers are read and handled by netOrk over the Engine API, through
+    ``open_container_engine`` (NetOrk/netork#765). The old CLI methods are gone;
+    only the OS action that lets the login user use Docker stays."""
+
+    def test_the_cli_methods_are_gone(self):
+        for name in ("get_docker_info", "get_docker_outdated", "reconstruct_docker_run", "_docker_bin"):
+            assert not hasattr(LinuxDriver, name), name
+
+    def test_the_engine_access_and_the_permission_fix_stay(self):
+        assert hasattr(LinuxDriver, "open_container_engine")
+        assert hasattr(LinuxDriver, "_action_fix_docker_permissions")
+
+
+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 _Channel:
+    """A netmiko connection that hands out its output in chunks and stops where
+    netmiko does: at the first chunk after which ``expect_string`` matches all
+    that was read so far."""
+
+    def __init__(self, chunks):
+        self.chunks = list(chunks)
+        self.patterns: list = []
+
+    def send_command(self, command, *, expect_string, **_kwargs):
+        self.patterns.append(expect_string)
+        output = ""
+        while self.chunks:
+            output += self.chunks.pop(0)
+            if re.search(expect_string, output):
+                return output
+        raise TimeoutError(f"pattern not detected: {expect_string!r}")
+
+
+#: What vault-01 sent on 2026-10-06 while its apt proxy served a corrupted
+#: InRelease: the signature line ends in ">", which looks like a prompt (#615).
+_BADSIG_CHUNKS = [
+    "sudo -n apt-get update -q 2>&1; echo __NETORK_RC=$?\n",
+    "Fehl:2 http://archive.ubuntu.com/ubuntu noble-updates InRelease\n"
+    "  Die folgenden Signaturen waren ungültig: BADSIG 871920D1991BC93C "
+    "Ubuntu Archive Automatic Signing Key (2018) <ftpmaster@ubuntu.com>\n",
+    "W: Fehler beim Holen von http://archive.ubuntu.com/ubuntu/dists/noble-updates/InRelease\n"
+    "E: Das Depot ist nicht signiert.\n__NETORK_RC=100\n",
+    "chris@vault-01:~$ ",
+]
+
+
+class TestReadToTheEnd:
+    """A line the command prints can end in ``>``, ``#`` or ``$`` -- apt's
+    ``<ftpmaster@ubuntu.com>`` after a bad signature. Taken for the prompt, it
+    ended the read while the command still ran, and the rest arrived as the next
+    command's output (#615). A command that echoes its exit status is read until
+    that marker and the prompt after it."""
+
+    def test_a_signature_line_does_not_end_the_refresh(self, driver):
+        driver._root = False
+        driver._device = _Channel(_BADSIG_CHUNKS)
+
+        result = driver.refresh_available_updates()
+
+        assert result["success"] is False
+        assert "E: Das Depot ist nicht signiert." in result["output"]
+
+    def test_the_session_stays_in_step(self, driver):
+        """Everything up to the prompt is consumed, so the next command reads its own output."""
+        driver._root = False
+        driver._device = _Channel(_BADSIG_CHUNKS + ["true\n", "__NETORK_RC=0\nchris@vault-01:~$ "])
+
+        driver.refresh_available_updates()
+        output, status = driver._sudo_status("true")
+
+        assert status == 0
+        assert "BADSIG" not in output
+
+    def test_the_echoed_command_does_not_count_as_the_marker(self, driver):
+        """Its literal ``$?`` is no number."""
+        channel = _Channel(["sudo true; echo __NETORK_RC=$?\n", "__NETORK_RC=0\nchris@vault-01:~$ "])
+        driver._device = channel
+
+        assert driver._sudo_status("true")[1] == 0
+        assert channel.chunks == []
+
+    def test_the_marker_alone_is_not_the_end(self, driver):
+        """The prompt after it has to be read too, or it would start the next output."""
+        channel = _Channel(["out\n__NETORK_RC=0\n", "chris@vault-01:~$ "])
+        driver._device = channel
+
+        driver._sudo_status("true")
+
+        assert channel.chunks == []
+
+    @pytest.mark.parametrize(
+        "command",
+        [
+            "{ LC_ALL=C apt list --upgradable 2>/dev/null; echo __APT_RC=$?; } | cat",
+            "timeout 45 systemctl restart -- cron.service; echo __SVC_RC=$?",
+        ],
+    )
+    def test_every_status_marker_is_waited_for(self, driver, command):
+        marker = re.search(r"echo (__[A-Z_]+=)", command).group(1)
+        channel = _Channel([f"x <a@b>\n", f"{marker}0\nchris@host:~$ "])
+        driver._device = channel
+
+        output = driver._send(command)
+
+        assert f"{marker}0" in output
+
+    def test_a_command_without_a_marker_still_ends_at_the_prompt(self, driver):
+        channel = _Channel(["6.8.0-142-generic\nchris@host:~$ "])
+        driver._device = channel
+
+        assert driver._send("uname -r").startswith("6.8.0-142-generic")
+        assert channel.patterns == [r"[#$\>]\s*$"]
+
+
+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
+
+
+class TestRebootHost:
+    """``reboot_host`` (napalm-device-types' ``HostRebootMixin``) restarts the host.
+
+    Without it netOrk could not restart a Linux host at all: its capability check
+    looks for ``reboot_host`` and found nothing (netOrk #637). The restart is
+    detached and a moment late, so the launcher's exit status comes back before
+    the host goes down, and closing the session cannot take it along.
+    """
+
+    def test_the_driver_can_restart_its_host(self):
+        assert callable(getattr(LinuxDriver, "reboot_host", None))
+
+    def test_the_restart_is_detached_and_privileged(self, driver):
+        driver._root = False
+        driver._device.send_command.return_value = "\n__NETORK_RC=0"
+
+        driver.reboot_host()
+
+        sent = driver._device.send_command.call_args[0][0]
+        assert sent.startswith("sudo -n sh -c ")
+        assert "/sbin/reboot" in sent and "trap" in sent and "&" in sent
+        assert sent.endswith("echo __NETORK_RC=$?")
+
+    def test_a_refusal_is_raised_with_what_the_host_said(self, driver):
+        driver._root = False
+        driver._device.send_command.return_value = "sudo: a password is required\n__NETORK_RC=1"
+
+        with pytest.raises(RuntimeError, match="password is required"):
+            driver.reboot_host()
+
+    def test_no_exit_status_is_no_success(self, driver):
+        driver._root = True
+        driver._device.send_command.return_value = "something else"
+
+        with pytest.raises(RuntimeError):
+            driver.reboot_host()
+
+
+# ---------------------------------------------------------------------------
+# Listening sockets: ss and the processes' cgroups, as root when it can
+# ---------------------------------------------------------------------------
+
+_SOCKETS = (
+    "SOCK_BEGIN\n[ss]\n"
+    'tcp LISTEN 0 128 0.0.0.0:5432 0.0.0.0:* users:(("postgres",pid=812,fd=6))\n'
+    "__SS_RC=0\n[cgroups]\n"
+    "812 0::/system.slice/system-postgresql.slice/postgresql@16-main.service\n"
+    "SOCK_END\n"
+)
+
+
+class TestGetListeningSockets:
+    def _sent(self, driver) -> list[str]:
+        return [c[0][0] for c in driver._device.send_command.call_args_list]
+
+    def test_it_reads_with_the_shared_command(self, driver):
+        from napalm_device_types import ListeningSocketsMixin
+
+        assert isinstance(driver, ListeningSocketsMixin)
+        driver._root = True
+        driver._device.send_command.return_value = _SOCKETS
+
+        reading = driver.get_listening_sockets()
+
+        assert reading["attributed"] is True
+        [socket] = reading["sockets"]
+        assert (socket["port"], socket["unit"]) == (5432, "postgresql@16-main")
+        assert self._sent(driver)[0].startswith("sh -c '")
+
+    def test_without_a_sudo_password_the_whole_script_runs_under_sudo_n(self, driver):
+        driver._root = False
+        driver._device.send_command.return_value = _SOCKETS
+
+        driver.get_listening_sockets()
+
+        assert self._sent(driver)[0].startswith("sudo -n sh -c '")
+
+    def test_with_a_sudo_password_it_goes_through_sudo(self, driver):
+        driver._root = False
+        driver._sudo_password = "pw"  # noqa: S105
+        driver._device.send_command.return_value = _SOCKETS
+
+        driver.get_listening_sockets()
+
+        sent = self._sent(driver)[0]
+        assert sent.startswith("echo pw | sudo -S")
+        assert "sh -c '" in sent
+
+    def test_when_sudo_refuses_it_reads_what_the_user_may_see(self, driver):
+        driver._root = False
+        driver._device.send_command.side_effect = ["sudo: a password is required", _SOCKETS]
+
+        reading = driver.get_listening_sockets()
+
+        assert reading["attributed"] is False
+        refused, plain = self._sent(driver)
+        assert refused.startswith("sudo -n sh -c '")
+        assert plain.startswith("sh -c '")