feat: harden the host's SSH server through napalm-device-types

LinuxDriver mixes in OpenSshServerMixin, so netOrk can take algorithms
ssh-audit flagged out of what sshd offers. The work is the base's; the
driver brings run_command, which already runs as root over an exec channel.

Requires napalm-device-types 4.5.0; bumps to 0.5.0. NetOrk/netork#1027

.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,44 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+## [0.4.0] – 2026-10-08
+
+### Added
+
+- `get_virtual_ips()` through napalm-device-types' `VirtualIpsMixin` (4.2.0): the
+  addresses vip-manager and keepalived configuration lets float onto this host,
+  whether or not it holds them right now (NetOrk/netork#829). The command goes on
+  the exec channel (`run_command`): with its awk filter it is about 4 kB, past the
+  line the PTY reliably takes. Root first, then without it.
+
+## [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

napalm_linux/linux.py

@@ -32,17 +32,30 @@ from netmiko.exceptions import (
 from napalm.base.exceptions import ConnectionException, ConnectionClosedException
 from napalm.base.netmiko_helpers import netmiko_args
 from napalm_device_types import (
+    APT_UPGRADABLE_COMMAND,
+    DNF_SECURITY_COMMAND,
+    ByteStream,
+    CommandResult,
+    ContainerEngineMixin,
     FingerprintRule,
+    HostStatusMixin,
     KernelFactsMixin,
+    ListeningSocketsMixin,
+    OpenSshServerMixin,
     OSDriver,
     SystemdServicesMixin,
     SystemdUnavailable,
+    VirtualIpsMixin,
+    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,
@@ -62,6 +75,63 @@ _PKG_MANAGERS = ["apt", "dnf", "yum", "apk", "pacman"]
 _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. "
@@ -135,23 +205,16 @@ def _arm_vendor_from_model(model: str) -> str:
     return " ".join(brand)
 
 
-#: A bare image ID (``d626d04934cd``), not a registry reference. ``docker ps``
-#: falls back to this whenever the tag a container was created from has since
-#: been moved to a newer image — i.e. exactly after a pull without a recreate.
-_IMAGE_ID_RE = re.compile(r"^(sha256:)?[0-9a-f]{12,64}$")
-
-
-def _looks_like_image_id(ref: str) -> bool:
-    """True if *ref* is an image ID rather than something a registry can resolve."""
-    return bool(_IMAGE_ID_RE.match(ref.strip()))
-
-
-def _short_image_id(raw: str) -> str:
-    """Normalise ``sha256:<64hex>`` and ``<12hex>`` to a comparable 12-char form."""
-    return raw.strip().removeprefix("sha256:")[:12]
-
-
-class LinuxDriver(KernelFactsMixin, SystemdServicesMixin, OSDriver):
+class LinuxDriver(
+    KernelFactsMixin,
+    ListeningSocketsMixin,
+    VirtualIpsMixin,
+    OpenSshServerMixin,
+    SystemdServicesMixin,
+    HostStatusMixin,
+    ContainerEngineMixin,
+    OSDriver,
+):
     """NAPALM driver for generic Linux systems.
 
     Connects via SSH (netmiko ``linux`` device type) and auto-detects the
@@ -273,7 +336,7 @@ class LinuxDriver(KernelFactsMixin, SystemdServicesMixin, 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:
@@ -297,13 +360,9 @@ class LinuxDriver(KernelFactsMixin, SystemdServicesMixin, OSDriver):
         The status is ``None`` when the marker never arrived (output cut short),
         so a caller can tell "unknown" from "succeeded".
         """
-        raw = self._sudo(f"{command}; echo {_RC_MARKER}$?", read_timeout=read_timeout)
-        matches = list(_RC_MARKER_RE.finditer(raw))
-        if not matches:
-            return raw, None
-        last = matches[-1]
-        output = (raw[: last.start()] + raw[last.end():]).strip()
-        return output, int(last.group(1))
+        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.
@@ -321,12 +380,83 @@ class LinuxDriver(KernelFactsMixin, SystemdServicesMixin, OSDriver):
         Without a sudo password, ``sudo -n`` fails at once where a prompt would
         otherwise hang the session until the read timeout.
         """
-        if not privileged or self._is_root():
+        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:
@@ -883,6 +1013,28 @@ class LinuxDriver(KernelFactsMixin, SystemdServicesMixin, OSDriver):
         """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)
+
+    def _run_virtual_ips_command(self, command: str, *, privileged: bool) -> str:
+        """The transport for ``VirtualIpsMixin.get_virtual_ips``: an exec channel.
+
+        The command carries its awk program and is about 4 kB, past the line the
+        PTY netmiko holds reliably takes. Read-only either way; root only because
+        the configuration may be root's alone.
+        """
+        return self.run_command(command, privileged=privileged, timeout=60).stdout
+
     # ------------------------------------------------------------------
     # OSDriver – package management
     # ------------------------------------------------------------------
@@ -1243,48 +1395,49 @@ class LinuxDriver(KernelFactsMixin, SystemdServicesMixin, 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(
-            "LC_ALL=C apt list --upgradable 2>/dev/null | grep -v '^Listing'",
-            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
+        # Raises ValueError when apt failed or the output was cut short.
+        return parse_apt_upgradable(self._send(APT_UPGRADABLE_COMMAND, read_timeout=60))
 
     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 = name_arch.rsplit(".", 1)[0] if "." in name_arch else name_arch
+                name = parts[0].rsplit(".", 1)[0]
                 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] = []
@@ -1622,421 +1775,6 @@ class LinuxDriver(KernelFactsMixin, SystemdServicesMixin, OSDriver):
             job["description"] = comment
         return job
 
-    # ------------------------------------------------------------------
-    # Docker
-    # ------------------------------------------------------------------
-
-    def _docker_bin(self) -> str:
-        """Path to the docker binary.
-
-        A hook rather than a literal because the Docker *logic* is the same
-        everywhere while the *location* is not: QTS ships Container Station's
-        docker under /share/<pool>/.qpkg/ and never puts it on PATH. Subclasses
-        override this one method instead of reimplementing the surface.
-        """
-        return "docker"
-
-    def get_docker_info(self) -> DockerInfoDict:
-        """Return information about the local Docker environment.
-
-        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
-
-        docker = self._docker_bin()
-
-        # Check docker binary first (docker --version doesn't need socket access)
-        if not self._send(f"command -v {docker} 2>/dev/null").strip():
-            return {"available": False}
-
-        # Verify socket access — docker ps is cheaper and fails immediately on permission errors
-        ps_check = self._send(f"{docker} ps 2>&1")
-        if "permission denied" in ps_check.lower() or "cannot connect" in ps_check.lower():
-            return {"available": False, "permission_denied": True}
-
-        version = self._send(f"{docker} --version 2>/dev/null").strip()
-
-        combined = self._send(
-            "echo '---CONTAINERS---'; "
-            f"{docker} ps -a --format '{{{{json .}}}}' 2>/dev/null; "
-            "echo '---IMAGES---'; "
-            f"{docker} images --format '{{{{json .}}}}' 2>/dev/null; "
-            "echo '---VOLUMES---'; "
-            f"{docker} volume ls --format '{{{{json .}}}}' 2>/dev/null; "
-            "echo '---NETWORKS---'; "
-            f"{docker} network ls --format '{{{{json .}}}}' 2>/dev/null; "
-            "echo '---CONFIGIMAGES---'; "
-            f"{docker} ps -aq 2>/dev/null | xargs -r {docker} inspect "
-            f"--format '{{{{.Id}}}}|{{{{.Config.Image}}}}|{{{{.Image}}}}' 2>/dev/null",
-            read_timeout=60,
-        )
-
-        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---",   "---CONFIGIMAGES---")
-        raw_cfgimages  = _section(combined, "---CONFIGIMAGES---", "\x00")  # sentinel
-
-        def _parse_labels(raw: Any) -> Dict[str, str]:
-            """Parse Docker labels — may be a dict (JSON map) or comma-sep string."""
-            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
-
-        # Stable image reference + restart-pending detection.
-        #
-        # ``docker ps`` only reports a usable tag while that tag still resolves to
-        # the running image. Pull a newer image without recreating the container and
-        # it degrades to a bare image ID — useless as a registry reference, and the
-        # very state in which an update is waiting. ``.Config.Image`` is the
-        # reference the container was created from and never degrades.
-        cfg_by_cid: dict[str, tuple] = {}
-        for line in raw_cfgimages.splitlines():
-            parts = line.strip().split("|")
-            if len(parts) != 3 or not parts[0]:
-                continue
-            cid, cfg_ref, run_id = parts
-            cfg_by_cid[cid[:12]] = (cfg_ref.strip(), run_id.strip())
-
-        tag_index: dict[str, tuple] = {}
-        for im in images:
-            repo, tag = im.get("repository", ""), im.get("tag", "")
-            if not repo or not tag or "<none>" in (repo, tag):
-                continue
-            tag_index[f"{repo}:{tag}"] = (_short_image_id(im.get("id", "")), im.get("version", ""))
-
-        for c in containers:
-            cfg_ref, run_id = cfg_by_cid.get(c.get("id", "")[:12], ("", ""))
-            if not cfg_ref:
-                continue
-            c["image_ref"] = cfg_ref
-            c["running_image_id"] = _short_image_id(run_id)
-            c["restart_pending"] = False
-            c["pending_version"] = ""
-            # A stopped container is not "pending a restart" in any useful sense.
-            if c.get("state") != "running":
-                continue
-            tag_id, tag_version = tag_index.get(cfg_ref, ("", ""))
-            if tag_id and c["running_image_id"] and tag_id != c["running_image_id"]:
-                c["restart_pending"] = True
-                c["pending_version"] = tag_version
-
-        # Volumes
-        volumes: List[dict[str, Any]] = []
-        for line in raw_volumes.splitlines():
-            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] = []
-        # Prefer the reference the container was created from. `image` is whatever
-        # `docker ps` displayed, which collapses to a bare image ID once the tag has
-        # moved on — and an image ID is not something a registry can resolve.
-        candidate_images: List[str] = []
-        for c in containers:
-            ref = (c.get("image_ref") or c.get("image") or "").strip()
-            if not ref or "@sha256:" in ref:  # skip digest-pinned
-                continue
-            if _looks_like_image_id(ref):
-                logger.warning(
-                    "container %s reports image ID %r instead of a tag — cannot ask the "
-                    "registry about it; skipping update check",
-                    c.get("name", "?"), ref,
-                )
-                continue
-            if ref not in candidate_images:
-                candidate_images.append(ref)
-        for img_name in candidate_images:
-            try:
-                local_raw = self._send(
-                    f"{self._docker_bin()} inspect {img_name!r} "
-                    f"--format '{{{{index .RepoDigests 0}}}}' 2>/dev/null",
-                    read_timeout=5,
-                ).strip()
-                if not local_raw or "@" not in local_raw:
-                    continue  # locally built or not yet pulled
-                local_digest = local_raw.split("@", 1)[1]
-
-                remote_full = self._send(
-                    f"{self._docker_bin()} 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:"):
-                    # Silence here is indistinguishable from "up to date" — say so.
-                    logger.warning(
-                        "no digest returned for %s; skipping update check. Registry said: %s",
-                        img_name, remote_full[:200].replace("\n", " ") or "(nothing)",
-                    )
-                    continue
-                if local_digest != remote_digest:
-                    outdated_images.append(img_name)
-            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"{self._docker_bin()} 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]:

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "napalm-linux"
-version = "0.1.0"
+version = "0.5.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>=2.2.0",
+    "napalm-device-types>=4.5.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
@@ -226,6 +228,7 @@ APT_UPGRADABLE = (
     "Listing... Done\n"
     "openssh-server/stable 1:9.2p1-2+deb12u2 amd64 [upgradable from: 1:9.2p1-2+deb12u1]\n"
     "curl/stable 7.88.1-10+deb12u6 amd64 [upgradable from: 7.88.1-10+deb12u5]\n"
+    "__APT_RC=0\n"
 )
 
 
@@ -863,60 +866,18 @@ class TestRunDeviceActionDispatch:
         mock_action.assert_called_once()
 
 
-class TestDockerBinHook:
-    """Where the docker binary lives is device-specific; what to do with it is not.
+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."""
 
-    QTS puts Container Station's docker under /share/<pool>/.qpkg/ and not on
-    PATH. Rather than duplicating the whole Docker surface in the QNAP driver,
-    the path is a one-method hook here and the logic stays generic. See
-    docs/ARCHITECTURE.md §4.4, "Generisch vs. treiberspezifisch".
-    """
+    def test_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_defaults_to_docker_on_path(self, driver):
-        assert driver._docker_bin() == "docker"
-
-    def test_detection_uses_the_hook(self, driver):
-        """A subclass pointing elsewhere must not be probed for a PATH docker."""
-        sent = []
-
-        def _record(cmd, **kwargs):
-            sent.append(cmd)
-            return ""
-
-        with patch.object(driver, "_docker_bin", return_value="/opt/cs/docker"):
-            with patch.object(driver, "_send", side_effect=_record):
-                result = driver.get_docker_info()
-
-        assert result == {"available": False}
-        assert any("/opt/cs/docker" in cmd for cmd in sent)
-        assert not any("command -v docker " in cmd for cmd in sent)
-
-    def test_all_docker_subcommands_use_the_hook(self, driver):
-        """Half-converted call sites are the failure mode here: detection would
-        find the binary and the actual queries would still miss it."""
-        sent = []
-
-        def _record(cmd, **kwargs):
-            sent.append(cmd)
-            if "command -v" in cmd:
-                return "/opt/cs/docker"
-            if "---CONTAINERS---" in cmd:
-                return "---CONTAINERS---\n---IMAGES---\n---VOLUMES---\n---NETWORKS---\n"
-            return ""
-
-        with patch.object(driver, "_docker_bin", return_value="/opt/cs/docker"):
-            with patch.object(driver, "_send", side_effect=_record):
-                driver.get_docker_info()
-
-        docker_cmds = [c for c in sent if "docker" in c]
-        assert docker_cmds
-        for cmd in docker_cmds:
-            assert "/opt/cs/docker" in cmd, f"unconverted call site: {cmd}"
-
-
-# ---------------------------------------------------------------------------
-# uninstall_package – purge, and getting out of `install ok unpacked`
-# ---------------------------------------------------------------------------
+    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:
@@ -1049,6 +1010,106 @@ class TestSudoStatus:
         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.
 
@@ -1282,3 +1343,353 @@ class TestManageService:
             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 '")
+
+
+# ---------------------------------------------------------------------------
+# get_virtual_ips (napalm-device-types VirtualIpsMixin)
+# ---------------------------------------------------------------------------
+
+_VIPS = (
+    "VIPS_BEGIN\n[systemd]\n[vip-manager vip-manager.service]\nactive active\n"
+    "cfg ip 10.7.224.10\ncfg netmask 24\ncfg interface ens7\nVIPS_END\n"
+)
+
+
+class TestGetVirtualIps:
+    """The command is about 4 kB -- past the line a terminal takes -- so it goes
+    on an exec channel, never typed into the PTY netmiko holds."""
+
+    @staticmethod
+    def _exec(driver, *answers):
+        from napalm_device_types import CommandResult
+
+        calls = []
+        replies = iter(answers)
+
+        def run_command(command, *, privileged=False, timeout=60, stdin=None):
+            calls.append((command, privileged))
+            return CommandResult(*next(replies))
+
+        driver.run_command = run_command
+        return calls
+
+    def test_it_reads_with_the_shared_command_on_an_exec_channel(self, driver):
+        from napalm_device_types import VirtualIpsMixin
+
+        assert isinstance(driver, VirtualIpsMixin)
+        calls = self._exec(driver, (_VIPS, "", 0))
+
+        reading = driver.get_virtual_ips()
+
+        assert reading["vip_manager"][0]["address"] == "10.7.224.10"
+        [(command, privileged)] = calls
+        assert privileged is True
+        assert command.startswith("sh -c '")
+        driver._device.send_command.assert_not_called()
+
+    def test_when_sudo_refuses_it_reads_what_the_user_may_see(self, driver):
+        calls = self._exec(driver, ("", "sudo: a password is required\n", 1), (_VIPS, "", 0))
+
+        reading = driver.get_virtual_ips()
+
+        assert reading["vip_manager"][0]["interface"] == "ens7"
+        assert [privileged for _, privileged in calls] == [True, False]
+
+    def test_without_a_sudo_password_the_whole_script_runs_under_sudo_n(self):
+        from types import SimpleNamespace
+
+        channels = []
+
+        class Channel:
+            def __init__(self):
+                channels.append(self)
+
+            def exec_command(self, command):
+                self.command = command
+
+            def settimeout(self, timeout):
+                pass
+
+            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 1
+
+        transport = SimpleNamespace(open_session=Channel)
+        driver = LinuxDriver("h", "u", "p", optional_args={})
+        driver._device = SimpleNamespace(
+            remote_conn_pre=SimpleNamespace(get_transport=lambda: transport)
+        )
+        driver._root = False
+
+        with pytest.raises(ValueError):
+            driver.get_virtual_ips()
+
+        privileged, plain = (c.command for c in channels)
+        assert privileged.startswith("sudo -n sh -c 'sh -c '")
+        assert plain.startswith("sh -c '")
+
+
+class TestHardensItsSshServer:
+    """A stock Linux host runs a stock OpenSSH: the hardening netOrk offers for
+    ssh-audit findings comes from napalm-device-types, carried over run_command
+    (NetOrk/netork#1027)."""
+
+    def test_mixes_in_the_openssh_hardening(self):
+        from napalm_device_types import OpenSshServerMixin
+
+        assert issubclass(LinuxDriver, OpenSshServerMixin)
+        assert callable(getattr(LinuxDriver, "harden_ssh_server", None))
+
+    def test_reads_the_algorithms_as_root_over_the_command_channel(self, driver):
+        from napalm_device_types import CommandResult
+
+        out = "kexalgorithms a\nhostkeyalgorithms b\nciphers c\nmacs d\n"
+        with patch.object(driver, "run_command", return_value=CommandResult(out, "", 0)) as run:
+            assert driver.get_ssh_server_algorithms()["mac"] == ["d"]
+        assert run.call_args.kwargs["privileged"] is True