feat: OpenSshServerMixin takes flagged algorithms out of what sshd offers

ssh-audit names algorithms an SSH server should stop offering. Taking them
out is the same on every OpenSSH host, so it lives here once: a driver
mixes in OpenSshServerMixin and supplies run_command.

harden_ssh_server reads the lists sshd really uses (sshd -T), removes the
requested algorithms and writes the rest to
/etc/ssh/sshd_config.d/00-netork-ssh-hardening.conf -- or, where
sshd_config has no Include of that directory, rewrites sshd_config after
a copy. It never removes the last algorithm of a list, nor, given the
managing client's lists, the last one that client speaks. sshd -t must
accept the result and sshd -T must show it; after the reload the caller's
verify opens a new session, and if that fails the still-open one puts the
previous state back and reloads again.

Bumps to 4.5.0. NetOrk/netork#1027

.github/workflows/ci.yml

@@ -0,0 +1,45 @@
+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
+          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/*

README.md

@@ -79,6 +79,53 @@ everywhere, so the command and its parse are concrete here and a driver supplies
 `_run_kernel_facts_command`. `OSDriver` does not carry it — a Windows host is an OS driver
 too, and `hasattr(driver, "get_kernel_facts")` has to stay truthful.
 
+`HostStatusMixin` (`get_host_status`) is mixed in the same way: whether a Linux host needs
+a reboot to finish an update (`/var/run/reboot-required`, `needs-restarting -r`, or a newer
+kernel of the running flavour installed), whether it patches itself (unattended-upgrades,
+dnf-automatic) and when that last succeeded (apt's `upgrade-stamp`), and whether dpkg was
+left halfway (a journal in `/var/lib/dpkg/updates` or a half-configured package; unknown
+while a package manager is running), and how full `/`, `/boot` and `/var` are (`df -Pk`,
+each filesystem once, in bytes; how full is too full is the caller's decision). `package_updates` holds the shared apt and dnf parsers: apt's suites become
+an update's `origin`, a `-security` suite makes it a security update, and dnf's security
+advisories do the same.
+
+`ListeningSocketsMixin` (`get_listening_sockets`) is mixed in the same way: every listening
+TCP and bound UDP socket from `ss -lntup`, with the systemd service or container behind it
+from `/proc/<pid>/cgroup`, in one round trip. A driver supplies
+`_run_listening_sockets_command(command, privileged=)`; the command arrives as one `sh -c`
+argument, so a `sudo -n` prefix covers all of it. Without root `ss` names only the login
+user's processes, and the reading says so (`attributed: false`) instead of failing. A host
+without `ss` is read with `netstat -lntup` (OpenWrt's busybox, old net-tools); on OpenWrt the
+cgroup names the procd service (`/services/<name>/<instance>`). A host with neither raises
+`ListeningSocketsUnavailable`.
+
+`VirtualIpsMixin` (`get_virtual_ips`) reads the addresses a host's HA configuration lets
+float between machines: vip-manager (5.x's `--config` YAML, 1.x's `-ip=${VIP_IP}` resolved
+through the unit's environment) and keepalived (`virtual_ipaddress` of every
+`vrrp_instance`, `include` followed four levels deep). It reports what is *declared*, not
+what the host holds right now -- that is in its interface addresses. The same files hold
+the etcd password and `auth_pass`, so an awk program filters **on the host** and prints
+allowlisted fields only; a secret never reaches the caller. Root first, then without it.
+A mechanism it could not read completely is `None` ("did not look"), a host without it
+`[]`. The command is about 4 kB, so a driver sends it on an exec channel
+(`_run_virtual_ips_command(command, privileged=)`), not typed into a shell.
+
+`OpenSshServerMixin` (`harden_ssh_server`, `get_ssh_server_algorithms`) takes algorithms
+out of what an OpenSSH server offers -- the ones ssh-audit flags. It reads the lists sshd
+really uses (`sshd -T`), takes out what was asked and writes the rest to
+`/etc/ssh/sshd_config.d/00-netork-ssh-hardening.conf`; only where sshd_config has no such
+`Include` is sshd_config itself rewritten, after a copy. It never removes the last algorithm
+of a class, nor -- given the managing client's lists -- the last one that client speaks.
+`sshd -t` must accept the result and `sshd -T` must show it; after the reload the caller's
+`verify` opens a new session, and if that fails the still-open one puts the old state back.
+A driver mixes it in and supplies `run_command`; a platform that regenerates sshd_config
+itself (an appliance with its own database) must not, or opts out with
+`harden_ssh_server = None`.
+
+**Update readers raise when they cannot read.** `get_available_updates` returns an empty
+list only when nothing is pending; netOrk keeps "pending since" per package, and an empty
+list for "don't know" would reset it.
+
 `SystemdServicesMixin` (`get_services`, `manage_service`) is mixed in the same way, by
 the drivers whose host runs systemd. Listing the services, checking a unit name and
 reading an action's exit status are the same on every such host, so they are concrete
@@ -88,6 +135,27 @@ trip (`list-unit-files` plus one `systemctl show` over every loaded unit) instea
 `is-enabled` and a `show` per unit. A host without systemd raises `SystemdUnavailable`,
 a `NotImplementedError`, so a driver can fall back to another init system.
 
+### Access channels: why there is no container model here
+
+`CommandChannelMixin` declares a driver's public channel to its host:
+- `run_command(command, *, privileged=False, timeout=60, stdin=None)` returns a `CommandResult(stdout, stderr, exit_code)`;
+- `open_stream(command, *, privileged=False)` returns a `ByteStream` to the command's stdin and stdout.
+
+The mixin sits in `DeviceTypeDriver` and only declares, so `hasattr(driver, "run_command")` is true exactly where a driver implements it. For SSH drivers, `channel.run_on_transport` and `channel.open_stream_on_transport` are the implementation over a paramiko exec channel: no PTY, stderr kept apart, a real exit code. How a command gains root stays with the driver.
+
+**`ContainerEngineMixin` is deliberately different from `SystemdServicesMixin`.** The systemd mixin owns its command and its parse; this one owns neither.
+- `container_engines()` says which engines the host offers (`[{"engine": "docker", "api": "docker-engine"}]`).
+- `open_container_engine(engine)` returns a `ContainerEngineConnection`:
+  - `open_api()` is a stream to the engine's API (`docker system dial-stdio`);
+  - `run_cli(args)` and `stream_cli(args)` run the engine's CLI with arguments the caller chooses. With `privileged=True` the driver runs it the way `run_command` gains root, for an engine that refuses the login user.
+- The driver decides only *how* the engine is reached. Its hook `_container_engine_binary(engine)` returns, for QNAP, the Container Station path, and the caller never sees it.
+
+**What runs on the engine, and what to do with it, is netOrk's** (NetOrk/netork#765): the container model, the Engine API requests, compose, updates. Above the connection everything is specific to the service, so this package abstracts the connection and nothing more.
+
+A driver mixes `ContainerEngineMixin` in itself; `OSDriver` does not carry it. A Windows host is an OS driver too, and has no `dial-stdio` to offer over WinRM.
+
+**Never half-close early.** `ByteStream.write` does not close anything; only `close_write` does. `dial-stdio` hands the daemon a half-close as "client gone", and the daemon then answers an unfinished request with HTTP 499.
+
 A function class may use the **template form** — public method concrete, the
 device-specific part a `_hook` declared under `if TYPE_CHECKING` — *when the base
 genuinely does work* on the result: normalising, sorting, validating, or orchestrating
@@ -270,6 +338,16 @@ class ProxmoxDriver(HypervisorDriver):
         ...
 ```
 
+A hypervisor that provisions VMs from cloud images declares, per guest OS,
+the agent cloud-init installs so the hypervisor can read the new VM's IP:
+`GUEST_AGENTS = {guest_os: (packages, runcmd)}`. Its keys are the guests the
+driver can provision, and `create_vm_from_cloud_init(guest_os=...)` refuses
+any other. A QEMU-based driver takes `provisioning.QEMU_GUEST_AGENTS`
+(Linux, FreeBSD, OpenBSD). `provisioning.network_config()` writes the
+network-config v2 every cloud-init flavour reads (FreeBSD's nuageinit reads
+no other), and `provisioning.split_compression()` says how a packed image is
+unpacked; both are the same for every hypervisor.
+
 ### OS / Linux
 
 ```python

napalm_device_types/__init__.py

@@ -34,13 +34,17 @@ Role bases -- what a device *is*:
 Function classes -- what a device *can do*. Shared behaviour lives here once
 instead of being restated on every role that happens to need it:
 
+* :class:`~napalm_device_types.channel.CommandChannelMixin`
 * :class:`~napalm_device_types.config_lifecycle.ConfigLifecycleMixin`
+* :class:`~napalm_device_types.container_engine.ContainerEngineMixin`
 * :class:`~napalm_device_types.dhcp.DhcpServerMixin`
 * :class:`~napalm_device_types.firewall_rules.FirewallRuleMixin`
 * :class:`~napalm_device_types.health_metrics.HealthMetricsMixin`
+* :class:`~napalm_device_types.host_status.HostStatusMixin`
 * :class:`~napalm_device_types.host_reboot.HostRebootMixin`
 * :class:`~napalm_device_types.interface_filter.InterfaceFilterMixin`
 * :class:`~napalm_device_types.kernel.KernelFactsMixin`
+* :class:`~napalm_device_types.listening.ListeningSocketsMixin`
 * :class:`~napalm_device_types.mac_acl.MacAclMixin`
 * :class:`~napalm_device_types.nat_vpn.NatVpnMixin`
 * :class:`~napalm_device_types.packages.PackageManagementMixin`
@@ -48,6 +52,7 @@ instead of being restated on every role that happens to need it:
 * :class:`~napalm_device_types.services.ServiceControlMixin`
 * :class:`~napalm_device_types.systemd.SystemdServicesMixin`
 * :class:`~napalm_device_types.updates.UpdateMixin`
+* :class:`~napalm_device_types.virtual_ips.VirtualIpsMixin`
 
 Introspection -- :func:`~napalm_device_types.roles.roles_of`,
 :func:`~napalm_device_types.roles.role_keys_of` and
@@ -56,7 +61,20 @@ Introspection -- :func:`~napalm_device_types.roles.roles_of`,
 
 from napalm_device_types.base import DeviceTypeDriver, FingerprintRule, PortSpec
 from napalm_device_types.access_point import AccessPointDriver
+from napalm_device_types.channel import (
+    ByteStream,
+    CommandChannelMixin,
+    CommandResult,
+    ParamikoExecStream,
+    open_stream_on_transport,
+    run_on_transport,
+)
 from napalm_device_types.config_lifecycle import ConfigLifecycleMixin
+from napalm_device_types.container_engine import (
+    ContainerEngineConnection,
+    ContainerEngineMixin,
+    ContainerEngineUnavailable,
+)
 from napalm_device_types.dhcp import DhcpServerMixin, normalize_cidr, normalize_mac
 from napalm_device_types.firewall import FirewallDriver
 from napalm_device_types.hypervisor import HypervisorDriver
@@ -67,6 +85,12 @@ from napalm_device_types.host_reboot import HostRebootMixin
 from napalm_device_types.interface_filter import InterfaceFilterMixin
 from napalm_device_types.kernel import KERNEL_FACTS_COMMAND, KernelFactsMixin, parse_kernel_facts
 from napalm_device_types.lag import add_lag_interfaces
+from napalm_device_types.listening import (
+    LISTENING_SOCKETS_COMMAND,
+    ListeningSocketsMixin,
+    ListeningSocketsUnavailable,
+    parse_listening_sockets,
+)
 from napalm_device_types.mac_acl import MacAclMixin
 from napalm_device_types.media import MediaDriver
 from napalm_device_types.nat_vpn import NatVpnMixin
@@ -74,7 +98,15 @@ from napalm_device_types.packages import PackageManagementMixin
 from napalm_device_types.phone import PhoneDriver
 from napalm_device_types.ping_sweep import PingSweepMixin, driver_supports_ping
 from napalm_device_types.roles import primary_role_of, role_keys_of, roles_of
+from napalm_device_types.host_status import HOST_STATUS_COMMAND, HostStatusMixin, parse_host_status
+from napalm_device_types.package_updates import (
+    APT_UPGRADABLE_COMMAND,
+    DNF_SECURITY_COMMAND,
+    parse_apt_upgradable,
+    parse_dnf_security,
+)
 from napalm_device_types.services import ServiceControlMixin
+from napalm_device_types.terminal import strip_terminal_codes
 from napalm_device_types.systemd import (
     SYSTEMD_SERVICES_COMMAND,
     SystemdServicesMixin,
@@ -82,12 +114,33 @@ from napalm_device_types.systemd import (
     parse_systemd_services,
 )
 from napalm_device_types.updates import UpdateMixin
+from napalm_device_types.virtual_ips import (
+    VIRTUAL_IPS_COMMAND,
+    VirtualIpsMixin,
+    parse_virtual_ips,
+)
 from napalm_device_types.residential_gateway import ResidentialGatewayDriver
+from napalm_device_types.ssh_server import (
+    SSH_ALGORITHM_CLASSES,
+    OpenSshServerMixin,
+    SshHardeningPlan,
+    SshHardeningResult,
+    plan_ssh_hardening,
+)
 from napalm_device_types.storage import StorageDriver
 from napalm_device_types.switch import SwitchDriver
 
 __all__ = [
     "AccessPointDriver",
+    "ByteStream",
+    "CommandChannelMixin",
+    "CommandResult",
+    "ContainerEngineConnection",
+    "ContainerEngineMixin",
+    "ContainerEngineUnavailable",
+    "ParamikoExecStream",
+    "open_stream_on_transport",
+    "run_on_transport",
     "ConfigLifecycleMixin",
     "DeviceTypeDriver",
     "DhcpServerMixin",
@@ -95,6 +148,8 @@ __all__ = [
     "FirewallDriver",
     "FirewallRuleMixin",
     "HealthMetricsMixin",
+    "HOST_STATUS_COMMAND",
+    "HostStatusMixin",
     "HostRebootMixin",
     "HypervisorDriver",
     "InterfaceFilterMixin",
@@ -103,9 +158,19 @@ __all__ = [
     "NatVpnMixin",
     "OSDriver",
     "PackageManagementMixin",
+    "APT_UPGRADABLE_COMMAND",
+    "DNF_SECURITY_COMMAND",
+    "parse_apt_upgradable",
+    "parse_dnf_security",
+    "parse_host_status",
+    "strip_terminal_codes",
     "KernelFactsMixin",
     "KERNEL_FACTS_COMMAND",
     "parse_kernel_facts",
+    "LISTENING_SOCKETS_COMMAND",
+    "ListeningSocketsMixin",
+    "ListeningSocketsUnavailable",
+    "parse_listening_sockets",
     "PhoneDriver",
     "PingSweepMixin",
     "PortSpec",
@@ -118,6 +183,14 @@ __all__ = [
     "SystemdUnavailable",
     "parse_systemd_services",
     "UpdateMixin",
+    "VIRTUAL_IPS_COMMAND",
+    "VirtualIpsMixin",
+    "OpenSshServerMixin",
+    "SSH_ALGORITHM_CLASSES",
+    "SshHardeningPlan",
+    "SshHardeningResult",
+    "plan_ssh_hardening",
+    "parse_virtual_ips",
     "add_lag_interfaces",
     "driver_supports_ping",
     "normalize_cidr",

napalm_device_types/base.py

@@ -12,6 +12,7 @@ from typing import NamedTuple
 
 from napalm.base import NetworkDriver
 
+from napalm_device_types.channel import CommandChannelMixin
 from napalm_device_types.host_reboot import HostRebootMixin
 from napalm_device_types.ping_sweep import PingSweepMixin
 
@@ -48,7 +49,7 @@ class PortSpec(NamedTuple):
     mandatory: bool = False
 
 
-class DeviceTypeDriver(PingSweepMixin, HostRebootMixin, NetworkDriver):
+class DeviceTypeDriver(PingSweepMixin, HostRebootMixin, CommandChannelMixin, NetworkDriver):
     """Common base for all netOrk device-type drivers.
 
     Sits between napalm.base.NetworkDriver and the type-specific abstract

napalm_device_types/channel.py

@@ -0,0 +1,188 @@
+# -*- coding: utf-8 -*-
+"""The command channel: a public way to run a command on a host, or open a stream.
+
+A driver already reaches its host, over SSH, a REST API or WinRM. What it did
+not offer was a *public* way for the layer above to use that reach: netOrk sent
+shell strings through napalm-linux's private ``_send``, an interactive PTY that
+merges stdout and stderr and has no exit code.
+
+:class:`CommandChannelMixin` declares two methods, and a driver that can
+implements them:
+
+* :meth:`run_command` runs a command and returns stdout, stderr and the exit code.
+* :meth:`open_stream` starts a command and returns a :class:`ByteStream` to its
+  stdin and stdout. That is how netOrk speaks the Docker Engine API, over
+  ``docker system dial-stdio`` (see :mod:`napalm_device_types.container_engine`).
+
+Like the role bases, the mixin only declares (under ``TYPE_CHECKING``), so
+``hasattr(driver, "run_command")`` stays a truthful answer.
+
+:func:`run_on_transport` and :func:`open_stream_on_transport` are the SSH
+implementation over a paramiko ``Transport``, for any SSH driver: an exec
+channel, so no PTY, separate stderr and a real exit status. How a command gains
+root (``sudo -S`` with the password on stdin, ``sudo -n``, or nothing as root)
+stays with the driver.
+"""
+
+from __future__ import annotations
+
+import socket
+import time
+from typing import Any, NamedTuple, Optional, Protocol, TYPE_CHECKING
+
+#: How much of a stream's stderr is kept. Enough to tell "permission denied"
+#: from "no such command"; a stream that writes megabytes there keeps its tail.
+STDERR_TAIL = 64 * 1024
+
+_READ = 32 * 1024
+_POLL = 0.01
+
+
+class CommandResult(NamedTuple):
+    """What a command printed, and how it ended."""
+
+    stdout: str
+    stderr: str
+    exit_code: int
+
+
+class ByteStream(Protocol):
+    """A running command's stdin and stdout, as bytes.
+
+    ``read`` returns ``b""`` at end of stream and raises :class:`TimeoutError`
+    when nothing arrives in time. ``write`` never closes anything:
+    ``dial-stdio`` hands the daemon a half-close as "client gone", and the
+    daemon then answers an unfinished request with HTTP 499 (netork#771). Only
+    ``close_write`` closes the writing side.
+    """
+
+    def read(self, max_bytes: int, timeout: Optional[float] = None) -> bytes: ...
+
+    def write(self, data: bytes) -> None: ...
+
+    def close_write(self) -> None: ...
+
+    def close(self) -> None: ...
+
+    @property
+    def exit_status(self) -> Optional[int]: ...
+
+    @property
+    def stderr(self) -> str: ...
+
+
+class CommandChannelMixin:
+    """Declares the channel; a driver that can reach a shell on its host implements it."""
+
+    if TYPE_CHECKING:  # pragma: no cover - declared for type checkers only
+
+        def run_command(
+            self,
+            command: str,
+            *,
+            privileged: bool = False,
+            timeout: float = 60,
+            stdin: Optional[bytes] = None,
+        ) -> CommandResult:
+            """Run *command* with the host's shell; *stdin*, if given, is written first."""
+            ...
+
+        def open_stream(self, command: str, *, privileged: bool = False) -> ByteStream:
+            """Start *command* and return a stream to its stdin and stdout."""
+            ...
+
+
+class ParamikoExecStream:
+    """A :class:`ByteStream` over a paramiko exec channel.
+
+    stderr is drained on every read: stdout and stderr share one window, and an
+    unread stderr would stall the stream.
+    """
+
+    def __init__(self, channel: Any, *, stderr_limit: int = STDERR_TAIL) -> None:
+        self._channel = channel
+        self._stderr = b""
+        self._limit = stderr_limit
+
+    def _drain_stderr(self) -> None:
+        while self._channel.recv_stderr_ready():
+            self._stderr = (self._stderr + self._channel.recv_stderr(_READ))[-self._limit :]
+
+    def read(self, max_bytes: int, timeout: Optional[float] = None) -> bytes:
+        self._channel.settimeout(timeout)
+        try:
+            return self._channel.recv(max_bytes)
+        except socket.timeout:
+            raise TimeoutError(f"no data within {timeout}s") from None
+        finally:
+            self._drain_stderr()
+
+    def write(self, data: bytes) -> None:
+        self._channel.sendall(data)
+
+    def close_write(self) -> None:
+        self._channel.shutdown_write()
+
+    def close(self) -> None:
+        self._channel.close()
+
+    @property
+    def exit_status(self) -> Optional[int]:
+        if not self._channel.exit_status_ready():
+            return None
+        return self._channel.recv_exit_status()
+
+    @property
+    def stderr(self) -> str:
+        self._drain_stderr()
+        return self._stderr.decode("utf-8", "replace")
+
+
+def _collect(channel: Any, command: str, timeout: float) -> tuple:
+    out, err = b"", b""
+    deadline = time.monotonic() + timeout
+    while True:
+        if channel.recv_ready():
+            out += channel.recv(_READ)
+        elif channel.recv_stderr_ready():
+            err += channel.recv_stderr(_READ)
+        elif channel.exit_status_ready():
+            return out, err
+        elif time.monotonic() > deadline:
+            raise TimeoutError(f"{command!r} did not finish within {timeout}s")
+        else:
+            time.sleep(_POLL)
+
+
+def run_on_transport(
+    transport: Any, command: str, *, stdin: Optional[bytes] = None, timeout: float = 60
+) -> CommandResult:
+    """Run *command* on an exec channel of the paramiko *transport*."""
+    channel = transport.open_session()
+    try:
+        channel.settimeout(timeout)
+        channel.exec_command(command)
+        if stdin is not None:
+            channel.sendall(stdin)
+        channel.shutdown_write()
+        out, err = _collect(channel, command, timeout)
+        return CommandResult(
+            out.decode("utf-8", "replace"), err.decode("utf-8", "replace"), channel.recv_exit_status()
+        )
+    finally:
+        channel.close()
+
+
+def open_stream_on_transport(
+    transport: Any, command: str, *, stdin_prefix: Optional[bytes] = None
+) -> ParamikoExecStream:
+    """Start *command* on an exec channel and return its stream.
+
+    *stdin_prefix* is written before anything else, which is how a sudo password
+    reaches ``sudo -S`` ahead of the stream's own bytes.
+    """
+    channel = transport.open_session()
+    channel.exec_command(command)
+    if stdin_prefix:
+        channel.sendall(stdin_prefix)
+    return ParamikoExecStream(channel)

napalm_device_types/container_engine.py

@@ -0,0 +1,141 @@
+# -*- coding: utf-8 -*-
+"""Access to a host's container engine, and nothing more.
+
+This package abstracts the connection, the driver builds it, and netOrk does
+the talking (NetOrk/netork#765). So there is no container model here and no
+parsing:
+
+* :meth:`ContainerEngineMixin.container_engines` says which engines the host
+  offers, and which API each speaks.
+* :meth:`ContainerEngineMixin.open_container_engine` returns a
+  :class:`ContainerEngineConnection`. It opens a stream to the engine's API
+  (``docker system dial-stdio``), and it runs the engine's CLI with arguments
+  the caller chooses. The CLI is there for what the API cannot do: compose,
+  pulls that need the host user's registry login, and private registry digests.
+
+The driver decides *how* the engine is reached. The hook is
+:meth:`ContainerEngineMixin._container_engine_binary`. QNAP returns its
+Container Station path, so callers never see where the binary lives. The
+command runs through the driver's own :class:`~napalm_device_types.channel.CommandChannelMixin`,
+whose privilege handling applies unchanged; Docker normally needs none, because
+the login user is in the ``docker`` group.
+
+Mixed in by a driver, not by a role base: a Windows host is an
+:class:`~napalm_device_types.os.OSDriver` too, and without a shell channel it
+has no ``dial-stdio`` to offer. ``hasattr(driver, "open_container_engine")``
+stays truthful.
+"""
+
+from __future__ import annotations
+
+import shlex
+from typing import Any, Dict, List, Optional, Sequence, TYPE_CHECKING
+
+from napalm_device_types.channel import ByteStream, CommandResult
+from napalm_device_types.models import ContainerEngineDict
+
+#: Engines this package knows how to reach, and the API each speaks.
+ENGINES: Dict[str, str] = {"docker": "docker-engine"}
+
+_PROBE_TIMEOUT = 15
+_CLI_TIMEOUT = 120
+
+
+class ContainerEngineUnavailable(NotImplementedError):
+    """The host offers no such engine, or this package does not know how to reach it."""
+
+
+class ContainerEngineConnection:
+    """The way to one container engine on one host.
+
+    Built by :meth:`ContainerEngineMixin.open_container_engine`. Callers pass
+    arguments; binary, quoting and the channel are the driver's.
+    """
+
+    def __init__(self, driver: Any, engine: str, binary: str) -> None:
+        self._driver = driver
+        self._binary = binary
+        self.engine = engine
+
+    def _command(self, args: Sequence[str]) -> str:
+        return shlex.join([self._binary, *args])
+
+    def open_api(self) -> ByteStream:
+        """A stream to the engine's API: HTTP/1.1 over ``system dial-stdio``."""
+        return self._driver.open_stream(self._command(["system", "dial-stdio"]))
+
+    def run_cli(
+        self,
+        args: Sequence[str],
+        *,
+        stdin: Optional[bytes] = None,
+        timeout: float = _CLI_TIMEOUT,
+        privileged: bool = False,
+    ) -> CommandResult:
+        """Run the engine's CLI with *args* and wait for it.
+
+        *privileged* runs it as root, the way the driver gains root for any
+        command: for a call the engine refused to the login user.
+        """
+        return self._driver.run_command(
+            self._command(args), privileged=privileged, timeout=timeout, stdin=stdin
+        )
+
+    def stream_cli(
+        self, args: Sequence[str], *, merge_stderr: bool = False, privileged: bool = False
+    ) -> ByteStream:
+        """Start the engine's CLI with *args* and stream its output.
+
+        *merge_stderr* folds stderr into the stream, for tools that report
+        progress there (``compose up``); *privileged* as for :meth:`run_cli`.
+        """
+        command = self._command(args)
+        return self._driver.open_stream(
+            f"{command} 2>&1" if merge_stderr else command, privileged=privileged
+        )
+
+
+class ContainerEngineMixin:
+    """Adds container engine access to a driver that implements the channel."""
+
+    if TYPE_CHECKING:  # pragma: no cover - declared for type checkers only
+
+        def run_command(
+            self,
+            command: str,
+            *,
+            privileged: bool = False,
+            timeout: float = 60,
+            stdin: Optional[bytes] = None,
+        ) -> CommandResult: ...
+
+        def open_stream(self, command: str, *, privileged: bool = False) -> ByteStream: ...
+
+    def _container_engine_binary(self, engine: str) -> str:
+        """The host command that is *engine*'s CLI. Override where it is not on PATH."""
+        return engine
+
+    def container_engines(self) -> List[ContainerEngineDict]:
+        """
+        Returns the container engines the host offers:
+
+        * engine (string) - e.g. "docker"
+        * api (string) - what its API is, e.g. "docker-engine"
+
+        An engine is listed when its CLI exists on the host. Whether the login
+        user may use it is decided by the first API call, which says
+        "permission denied" in the stream's stderr.
+        """
+        found: List[ContainerEngineDict] = []
+        for engine, api in ENGINES.items():
+            probe = shlex.join(["command", "-v", self._container_engine_binary(engine)])
+            result = self.run_command(probe, timeout=_PROBE_TIMEOUT)
+            if result.exit_code == 0 and result.stdout.strip():
+                found.append({"engine": engine, "api": api})
+        return found
+
+    def open_container_engine(self, engine: str) -> ContainerEngineConnection:
+        """The connection to *engine*; :class:`ContainerEngineUnavailable` if unknown."""
+        if engine not in ENGINES:
+            raise ContainerEngineUnavailable(f"no way to reach container engine {engine!r}")
+        return ContainerEngineConnection(self, engine, self._container_engine_binary(engine))

napalm_device_types/host_status.py

@@ -0,0 +1,325 @@
+# -*- coding: utf-8 -*-
+"""Host status: does the host need a reboot, and does it patch itself?
+
+A patch run that installed a new kernel or libc has not closed anything until
+the host restarts, so "reboot required" is part of being patched. Whether the
+host installs updates on its own (unattended-upgrades, dnf-automatic) decides
+how far netOrk's maintenance window reaches. Both are read the same way on every
+Linux host, so the command and its parse live here once and a driver only
+carries the command across.
+
+**Reboot required** is any of:
+
+- ``/var/run/reboot-required`` exists. Ubuntu always writes it; Debian does when
+  update-notifier or unattended-upgrades is installed.
+- ``needs-restarting -r`` exits 1 (dnf-utils).
+- A kernel newer than the running one is installed, of the same flavour. A
+  Raspberry Pi carries ``rpi-v8`` and ``rpi-2712`` builds side by side, and only
+  the running one's counts.
+- On FreeBSD, ``freebsd-version -k`` (the installed kernel) differs from
+  ``freebsd-version -r`` (the running one): a patch replaced the kernel.
+
+It is ``None`` when none of these could be read, for example in a container
+without a ``/lib/modules`` of its own.
+
+**Auto updates** is apt's ``APT::Periodic::Unattended-Upgrade`` (set, not "0",
+and ``apt-daily-upgrade.timer`` not disabled) or an enabled dnf-automatic timer.
+It is ``None`` on a host with neither apt nor dnf-automatic. When they last ran
+successfully is ``/var/lib/apt/periodic/upgrade-stamp``, which
+``apt.systemd.daily`` touches only after unattended-upgrade succeeded.
+``update-success-stamp`` would say nothing: every ``apt-get update`` touches it,
+whoever runs it. How often they are meant to run is the same setting, in apt's
+units (a bare number is days).
+
+**An interrupted package manager** is dpkg left halfway, the way apt checks it:
+numbered files left in ``/var/lib/dpkg/updates``, or a package in a state only
+a run cut off leaves behind (half-installed, unpacked, half-configured, waiting
+for triggers, or flagged for reinstallation). apt then refuses to install
+anything until ``dpkg --configure -a`` has run, while ``apt.systemd.daily``
+exits quietly every day (NetOrk/netork#738). While dpkg, apt, aptitude or
+unattended-upgrade is running, packages are halfway on purpose -- also between
+two of apt's dpkg calls -- so the answer is then ``None``, as on a host without
+dpkg. unattended-upgrade is told from Ubuntu's idle shutdown helper by its
+command line (:data:`BUSY_CHECK`).
+
+**Filesystems** are how full ``/``, ``/boot`` and ``/var`` are, from POSIX
+``df -Pk``: where an upgrade unpacks a kernel, builds its initrd and keeps the
+package files. A kernel upgrade that ran out of room there left dpkg halfway on
+a Proxmox host whose root had been 94 % full for a week (NetOrk/netork#862). A
+directory that is not a mount of its own is reported as the filesystem it lives
+on, so each filesystem appears once. ``available`` is what df says an ordinary
+user can still write, which on ext4 is less than total minus used. How full is
+too full is not decided here.
+"""
+
+from __future__ import annotations
+
+import re
+from typing import Dict, List, Optional, Tuple, TYPE_CHECKING
+
+from napalm_device_types.models import (
+    FilesystemUsageDict,
+    HostStatusDict,
+    InterruptedPackageDict,
+)
+from napalm_device_types.terminal import strip_terminal_codes
+
+_BEGIN = "HSTAT_BEGIN"
+_END = "HSTAT_END"
+_REBOOT_FILE = "/var/run/reboot-required"
+_APT_TIMER = "apt-daily-upgrade.timer"
+_DNF_TIMERS = ("dnf-automatic.timer", "dnf-automatic-install.timer")
+_UPGRADE_STAMP = "/var/lib/apt/periodic/upgrade-stamp"
+_DPKG_JOURNAL = "/var/lib/dpkg/updates"
+#: Where an upgrade writes: the kernel's modules, its initrd, the package files.
+_FILESYSTEMS = "/ /boot /var"
+#: Prints ``[dpkg-busy]`` while a process that leaves packages halfway is working:
+#: dpkg, apt, apt-get or aptitude by name, unattended-upgrade by its command line.
+#: Its name is cut to 15 characters, ``unattended-upgr``, the same as that of
+#: ``unattended-upgrade-shutdown --wait-for-signal``, which Ubuntu keeps running all
+#: the time and which installs nothing.
+BUSY_CHECK = (
+    "{ ps -e -o comm= 2>/dev/null | grep -qxE 'dpkg|apt|apt-get|aptitude' || "
+    "ps -e -o args= 2>/dev/null | grep -qE '(^|[ /])unattended-upgrade( |$)'; } "
+    "&& echo '[dpkg-busy]'; "
+)
+
+#: One line, POSIX ``sh``, read-only, no privileges. The frame markers are
+#: printed in two halves so that an echoing transport does not show them early.
+#: Each timer is asked on its own: older systemd prints nothing for an unknown
+#: unit, which would shift a combined answer. Run through a pipe, so nothing in
+#: it sees a terminal and colours its output.
+HOST_STATUS_COMMAND = (
+    "{ printf '%s%s\\n' HSTAT_ BEGIN; "
+    f"[ -f {_REBOOT_FILE} ] && echo '[reboot-required]'; "
+    "if command -v needs-restarting >/dev/null 2>&1; then echo '[needs-restarting]'; "
+    "needs-restarting -r >/dev/null 2>&1; echo $?; fi; "
+    "echo '[kernel]'; uname -r; echo '[modules]'; ls -1 /lib/modules 2>/dev/null; "
+    "if command -v freebsd-version >/dev/null 2>&1; then "
+    "echo '[freebsd-kernel]'; freebsd-version -k; "
+    "echo '[freebsd-running]'; freebsd-version -r; fi; "
+    "if command -v apt-config >/dev/null 2>&1; then echo '[apt-config]'; "
+    "apt-config dump 2>/dev/null | grep '^APT::Periodic::Unattended-Upgrade '; "
+    f"echo '[upgrade-stamp]'; stat -c %Y {_UPGRADE_STAMP} 2>/dev/null; fi; "
+    "if command -v dpkg-query >/dev/null 2>&1; then "
+    f"{BUSY_CHECK}"
+    f"echo '[dpkg-journal]'; ls -1 {_DPKG_JOURNAL} 2>/dev/null | head -n 20; "
+    "echo '[dpkg-audit]'; dpkg-query -W -f='${db:Status-Abbrev} ${Package}\\n' 2>/dev/null "
+    "| awk 'substr($0, 2, 1) ~ /[HUFWt]/ || substr($0, 3, 1) == \"R\"' | head -n 50; fi; "
+    f"echo '[filesystems]'; df -Pk {_FILESYSTEMS}; "
+    f"echo '[timers]'; for u in {_APT_TIMER} {' '.join(_DNF_TIMERS)}; do "
+    'printf \'%s %s\\n\' "$u" "$(systemctl is-enabled "$u" 2>/dev/null)"; done; '
+    "printf '%s%s\\n' HSTAT_ END; } 2>/dev/null | cat"
+)
+
+_PERIODIC = re.compile(r'^APT::Periodic::Unattended-Upgrade\s+"([^"]*)"')
+_OFF_STATES = frozenset({"disabled", "masked"})
+_INTERVAL = re.compile(r"(\d+)([smhd]?)")
+_UNIT_SECONDS = {"s": 1, "m": 60, "h": 3600, "d": 86400, "": 86400}
+_JOURNAL_ENTRY = re.compile(r"\d+")
+#: One line of ``df -P``: device, 1024-blocks, used, available, capacity, mountpoint.
+#: The device and the mountpoint may hold blanks; the four numbers between them do not.
+_DF_LINE = re.compile(r"^.+?\s+(\d+)\s+(\d+)\s+(\d+)\s+(?:\d+%|-)\s+(/.*)$")
+#: dpkg's second status letter -> its word, for the states a cut-off run leaves.
+_HALFWAY = {
+    "H": "half-installed",
+    "U": "unpacked",
+    "F": "half-configured",
+    "W": "triggers-awaited",
+    "t": "triggers-pending",
+}
+
+
+def _sections(output: str) -> Dict[str, List[str]]:
+    lines = [line.strip() for line in strip_terminal_codes(output).splitlines()]
+    try:
+        start = lines.index(_BEGIN)
+        end = lines.index(_END, start)
+    except ValueError:
+        raise ValueError("no intact host status report in the output") from None
+    sections: Dict[str, List[str]] = {}
+    current: List[str] = []
+    for line in lines[start + 1 : end]:
+        if line.startswith("[") and line.endswith("]"):
+            current = sections.setdefault(line[1:-1], [])
+        elif line:
+            current.append(line)
+    return sections
+
+
+def _version_key(version: str) -> Tuple[object, ...]:
+    """Natural order: 6.8.0-142 after 6.8.0-87, 7.0.14 after 7.0.2."""
+    return tuple(int(part) if part.isdigit() else part for part in re.split(r"(\d+)", version))
+
+
+def kernel_reboot_pending(running: str, installed: List[str]) -> Optional[str]:
+    """The newest installed kernel of the running flavour, if it is newer than the
+    running one; otherwise None.
+
+    The flavour is what follows the last ``-`` (``generic``, ``amd64``, ``pve``,
+    ``v8``); a kernel of another flavour is never a reason to reboot.
+    """
+    flavour = running.rsplit("-", 1)[-1]
+    same = [k for k in installed if k.rsplit("-", 1)[-1] == flavour]
+    if not same:
+        return None
+    newest = max(same, key=lambda k: _version_key(k.rsplit("-", 1)[0]))
+    if _version_key(newest.rsplit("-", 1)[0]) > _version_key(running.rsplit("-", 1)[0]):
+        return newest
+    return None
+
+
+def _freebsd_reboot(sections: Dict[str, List[str]]) -> Optional[Tuple[bool, Optional[str]]]:
+    """FreeBSD's answer from freebsd-version, or None on any other system."""
+    installed = (sections.get("freebsd-kernel") or [""])[0]
+    running = (sections.get("freebsd-running") or [""])[0]
+    if not (installed and running):
+        return None
+    if installed != running:
+        return True, f"kernel {installed} installed, {running} running"
+    return False, None
+
+
+def _reboot(sections: Dict[str, List[str]]) -> Tuple[Optional[bool], Optional[str]]:
+    if "reboot-required" in sections:
+        return True, f"{_REBOOT_FILE} is present"
+    needs = sections.get("needs-restarting")
+    if needs and needs[0] == "1":
+        return True, "needs-restarting -r reports a reboot"
+    freebsd = _freebsd_reboot(sections)
+    if freebsd is not None:
+        return freebsd
+    running = (sections.get("kernel") or [""])[0]
+    modules = sections.get("modules") or []
+    newer = kernel_reboot_pending(running, modules) if running and modules else None
+    if newer:
+        return True, f"kernel {newer} installed, {running} running"
+    if needs or modules:
+        return False, None
+    return None, None
+
+
+def _timer_states(sections: Dict[str, List[str]]) -> Dict[str, str]:
+    states: Dict[str, str] = {}
+    for line in sections.get("timers") or []:
+        unit, _, state = line.partition(" ")
+        states[unit] = state.strip()
+    return states
+
+
+def _periodic(sections: Dict[str, List[str]]) -> Optional[str]:
+    """The value of ``APT::Periodic::Unattended-Upgrade``, or None when it is not set."""
+    lines = sections.get("apt-config") or []
+    match = next(filter(None, (_PERIODIC.match(line) for line in lines)), None)
+    return match.group(1) if match else None
+
+
+def _auto_updates(sections: Dict[str, List[str]]) -> Optional[bool]:
+    timers = _timer_states(sections)
+    if any(timers.get(t) == "enabled" for t in _DNF_TIMERS):
+        return True
+    if "apt-config" not in sections:
+        return None
+    switched_on = _periodic(sections) not in (None, "", "0")
+    return switched_on and timers.get(_APT_TIMER) not in _OFF_STATES
+
+
+def _interval(sections: Dict[str, List[str]]) -> Optional[int]:
+    """apt's interval in seconds, the way ``apt.systemd.daily`` reads it; 0 for "always"."""
+    value = _periodic(sections)
+    if value == "always":
+        return 0
+    match = _INTERVAL.fullmatch(value or "")
+    if not match or int(match.group(1)) == 0:
+        return None
+    return int(match.group(1)) * _UNIT_SECONDS[match.group(2)]
+
+
+def _last_success(sections: Dict[str, List[str]]) -> Optional[int]:
+    stamp = (sections.get("upgrade-stamp") or [""])[0]
+    return int(stamp) if stamp.isdigit() else None
+
+
+def _halfway(line: str) -> Optional[InterruptedPackageDict]:
+    status, _, name = line.partition(" ")
+    name = name.strip()
+    state = "reinstall-required" if status[2:3] == "R" else _HALFWAY.get(status[1:2])
+    return {"name": name, "state": state} if name and state else None
+
+
+def _dpkg(sections: Dict[str, List[str]]) -> Tuple[Optional[bool], List[InterruptedPackageDict]]:
+    if "dpkg-audit" not in sections or "dpkg-busy" in sections:
+        return None, []
+    packages = [p for p in map(_halfway, sections["dpkg-audit"]) if p is not None]
+    journal = any(_JOURNAL_ENTRY.fullmatch(f) for f in sections.get("dpkg-journal") or [])
+    return bool(packages) or journal, packages
+
+
+def _filesystems(sections: Dict[str, List[str]]) -> Optional[List[FilesystemUsageDict]]:
+    """Each filesystem df named, once and in its order; None when it named none."""
+    found: Dict[str, FilesystemUsageDict] = {}
+    for line in sections.get("filesystems") or []:
+        match = _DF_LINE.match(line)
+        if not match or int(match.group(1)) == 0:
+            continue
+        total, used, available = (int(match.group(i)) * 1024 for i in (1, 2, 3))
+        mountpoint = match.group(4)
+        found.setdefault(
+            mountpoint,
+            {"mountpoint": mountpoint, "total": total, "used": used, "available": available},
+        )
+    return list(found.values()) or None
+
+
+def parse_host_status(output: str) -> HostStatusDict:
+    """Parse what :data:`HOST_STATUS_COMMAND` printed.
+
+    :raises ValueError: when the output carries no intact report.
+    """
+    sections = _sections(output)
+    required, reason = _reboot(sections)
+    interrupted, packages = _dpkg(sections)
+    return {
+        "reboot_required": required,
+        "reboot_reason": reason,
+        "auto_updates": _auto_updates(sections),
+        "auto_updates_last_success": _last_success(sections),
+        "auto_updates_interval": _interval(sections),
+        "package_manager_interrupted": interrupted,
+        "interrupted_packages": packages,
+        "filesystems": _filesystems(sections),
+    }
+
+
+class HostStatusMixin:
+    """Adds :meth:`get_host_status` to a driver that can run a command on a Linux host.
+
+    The template form, like :class:`~napalm_device_types.kernel.KernelFactsMixin`:
+    the reading and its parse are the same everywhere, and a driver supplies only
+    :meth:`_run_host_status_command`. Mixed in by the drivers that can, so
+    ``hasattr(driver, "get_host_status")`` stays a truthful answer.
+    """
+
+    if TYPE_CHECKING:  # pragma: no cover - declared for type checkers only
+
+        def _run_host_status_command(self, command: str) -> str:
+            """Run *command* on the host with ``sh`` and return what it printed."""
+            ...
+
+    def get_host_status(self) -> HostStatusDict:
+        """
+        Returns whether the host needs a reboot, whether it patches itself, and
+        whether its package manager was left halfway.
+
+        * reboot_required (bool or None)
+        * reboot_reason (string or None)
+        * auto_updates (bool or None)
+        * auto_updates_last_success (Unix time or None)
+        * auto_updates_interval (seconds or None)
+        * package_manager_interrupted (bool or None)
+        * interrupted_packages (list of {name, state})
+        * filesystems (list of {mountpoint, total, used, available} in bytes, or None)
+
+        :raises ValueError: if the host's output carried no intact report.
+        """
+        return parse_host_status(self._run_host_status_command(HOST_STATUS_COMMAND))

napalm_device_types/hypervisor.py

@@ -14,6 +14,7 @@ from typing import Any, Dict, List, TYPE_CHECKING
 from napalm_device_types.base import DeviceTypeDriver
 from napalm_device_types.packages import PackageManagementMixin
 from napalm_device_types.health_metrics import HealthMetricsMixin
+from napalm_device_types.provisioning import QEMU_GUEST_AGENTS, GuestAgents
 from napalm_device_types.models import (
     NICConfigDict,
     NetworkTargetDict,
@@ -44,11 +45,11 @@ class HypervisorDriver(PackageManagementMixin, HealthMetricsMixin, DeviceTypeDri
     ROLE: str = "hypervisor"
     TYPE_LABEL: str = "Hypervisor"
 
-    #: What cloud-init installs and starts on a VM provisioned through this
-    #: driver, so the hypervisor can read the guest's IP address back.
-    GUEST_AGENT_PACKAGES: tuple[str, ...] = ("qemu-guest-agent",)
-    GUEST_AGENT_RUNCMD: tuple[str, ...] = ("systemctl enable --now qemu-guest-agent",)
-
+    #: What cloud-init installs and runs on a VM provisioned through this
+    #: driver, per guest OS, so the hypervisor can read the guest's IP
+    #: address back. The guest operating systems listed here are the ones
+    #: the driver can provision; netOrk reads this off the class.
+    GUEST_AGENTS: GuestAgents = {"linux": QEMU_GUEST_AGENTS["linux"]}
 
 
     # ------------------------------------------------------------------
@@ -464,6 +465,7 @@ class HypervisorDriver(PackageManagementMixin, HealthMetricsMixin, DeviceTypeDri
             disk_resize_gb: int | None = None,
             storage: str | None = None,
             cpu_type: str | None = None,
+            guest_os: str = "linux",
             download_timeout: int = 300,
             timeout: int = 180,
         ) -> VMProvisionResultDict:
@@ -512,6 +514,13 @@ class HypervisorDriver(PackageManagementMixin, HealthMetricsMixin, DeviceTypeDri
                   value other than None with ValueError; one that has it raises
                   ValueError for a name it does not list or that is not
                   ``available`` on this node, before creating anything.
+                guest_os (string) - the operating system in the image, a key of
+                  ``GUEST_AGENTS`` (``"linux"``, ``"freebsd"``, ``"openbsd"``). The
+                  driver sets up the VM's hardware for it (OS type, how the guest
+                  agent is attached) and, where the guest needs it, writes the
+                  network-config itself (``provisioning.network_config``). A
+                  guest_os the driver does not list raises ValueError before
+                  anything is created. Default ``"linux"``.
                 download_timeout (int) - maximum seconds to wait for the image download
                   (skipped entirely if already cached on the hypervisor). Default 300.
                 timeout (int) - maximum seconds to wait for the remaining provisioning

napalm_device_types/listening.py

@@ -0,0 +1,276 @@
+# -*- coding: utf-8 -*-
+"""Listening sockets: what listens on which address, and which service it is.
+
+Whether a service is reachable from outside its host is decided by what it
+listens on -- ``0.0.0.0:5432`` is, ``127.0.0.1:5432`` is not -- and that is read
+the same way on every Linux host. So the command and its parse live here once,
+and a driver only carries the command across.
+
+**One round trip.** ``ss -lntup`` lists every listening TCP and bound UDP
+socket with the processes holding it; for each of those processes,
+``/proc/<pid>/cgroup`` says which systemd service or container it runs in. The
+report is framed, and a report whose end is missing raises: a list cut short
+must never read as sockets that closed.
+
+**Root, and without it.** Only root sees every process behind a socket.
+The whole script therefore goes to the host as one ``sh -c`` argument -- a
+driver that prefixes ``sudo -n`` would otherwise run only its first command as
+root. When that call brings no report back (no sudo, a wrong password), the
+command runs again without privilege: the sockets are still worth having, and
+the reading says it is not ``attributed``.
+
+**No ``-H``.** iproute2 before 4.10 has no option to leave out the header and
+fails on it, which would read as nothing listening. The parse skips the header
+instead.
+
+**Without ``ss``, ``netstat``.** OpenWrt's busybox and old net-tools hosts have
+no ``ss``; ``netstat -lntup`` lists the same sockets with ``PID/Program``, and
+the cgroups are read for its PIDs alike. On OpenWrt the cgroup names the procd
+service (``/services/<name>/<instance>``), a jailed one too, whose PID is not the
+one procd reports. A host with neither raises :class:`ListeningSocketsUnavailable`.
+"""
+
+from __future__ import annotations
+
+import re
+from shlex import quote
+from typing import Dict, List, Optional, Tuple, TYPE_CHECKING
+
+from napalm_device_types.models import ListeningSocketDict, ListeningSocketsDict
+from napalm_device_types.terminal import strip_terminal_codes
+
+_BEGIN = "SOCK_BEGIN"
+_END = "SOCK_END"
+_NO_SS = "no-ss"
+_RC_RE = re.compile(r"^__SS_RC=(\d+)$")
+
+#: One line, POSIX ``sh``, read-only. The frame markers are printed in two
+#: halves so that a transport which echoes the command does not show them early.
+#: ``ss`` is in ``/usr/sbin`` on some systems, outside a login user's ``PATH``.
+#: ``netstat`` names a socket's process as ``PID/Program``, ``ss`` as ``pid=PID``.
+LISTENING_SOCKETS_COMMAND = (
+    "PATH=$PATH:/usr/sbin:/sbin; "
+    "printf '%s%s\\n' SOCK_ BEGIN; "
+    "if command -v ss >/dev/null 2>&1; then "
+    "t=ss; s=$(ss -lntup 2>&1); r=$?; "
+    "pids=$(printf '%s\\n' \"$s\" | grep -o 'pid=[0-9]*' | cut -d= -f2); "
+    "elif command -v netstat >/dev/null 2>&1; then "
+    "t=netstat; s=$(netstat -lntup 2>&1); r=$?; "
+    "pids=$(printf '%s\\n' \"$s\" | grep -o ' [0-9][0-9]*/' | tr -d ' /'); "
+    "else t=; fi; "
+    "if [ -n \"$t\" ]; then "
+    "echo \"[$t]\"; printf '%s\\n' \"$s\"; echo \"__SS_RC=$r\"; "
+    "echo '[cgroups]'; "
+    "for p in $(printf '%s\\n' \"$pids\" | sort -u); do "
+    "sed \"s|^|$p |\" /proc/$p/cgroup 2>/dev/null; done; "
+    "else echo '[no-ss]'; fi; "
+    "printf '%s%s\\n' SOCK_ END"
+)
+
+_PROTOCOLS = frozenset({"tcp", "udp"})
+#: netstat names the IPv6 sockets of net-tools ``tcp6``/``udp6``; busybox does not.
+_NETSTAT_PROTOCOLS = {"tcp": "tcp", "tcp6": "tcp", "udp": "udp", "udp6": "udp"}
+#: ``1604/dropbear``; net-tools prints ``700/sshd: /usr/sbin``.
+_PROGRAM_RE = re.compile(r"^(\d+)/(\S*)")
+#: ``users:(("nginx",pid=901,fd=6),("nginx",pid=900,fd=6))``
+_USER_RE = re.compile(r'\("((?:[^"\\]|\\.)*)",pid=(\d+),fd=\d+\)')
+#: The service a cgroup path runs in: its deepest ``*.service`` component.
+_SERVICE_RE = re.compile(r"/([^/]+)\.service(?=/|$)")
+#: OpenWrt's procd: ``/services/<name>/<instance>``.
+_PROCD_RE = re.compile(r"^/services/([^/]+)(?:/|$)")
+#: A container's cgroup: ``docker-<id>.scope`` (systemd driver), ``/docker/<id>`` (cgroupfs).
+_CONTAINER_RE = re.compile(
+    r"(?:docker|libpod)-([0-9a-f]{64})\.scope|/(?:docker|libpod)/([0-9a-f]{64})(?=/|$)"
+)
+
+
+class ListeningSocketsUnavailable(NotImplementedError):
+    """The host has neither ``ss`` nor ``netstat``; nothing to read, nothing to retry."""
+
+
+def _frame(output: str) -> List[str]:
+    lines = [line.strip() for line in strip_terminal_codes(output).splitlines()]
+    try:
+        start = lines.index(_BEGIN)
+        end = lines.index(_END, start)
+    except ValueError:
+        raise ValueError("no intact listening socket report in the output") from None
+    return lines[start + 1 : end]
+
+
+def _sections(lines: List[str]) -> Dict[str, List[str]]:
+    sections: Dict[str, List[str]] = {}
+    current: List[str] = []
+    for line in lines:
+        if line.startswith("[") and line.endswith("]") and " " not in line:
+            current = sections.setdefault(line[1:-1], [])
+        else:
+            current.append(line)
+    return sections
+
+
+def _split_local(local: str) -> Optional[Tuple[str, Optional[str], int]]:
+    """``[fe80::1%eth0]:546`` -> ``("fe80::1", "eth0", 546)``; None if no port."""
+    host, sep, port = local.rpartition(":")
+    if not sep or not port.isdigit():
+        return None
+    if host.startswith("[") and host.endswith("]"):
+        host = host[1:-1]
+    address, _, zone = host.partition("%")
+    return address or "*", zone or None, int(port)
+
+
+def _cgroup_paths(lines: List[str]) -> Dict[int, str]:
+    """Each process's cgroup path: the unified hierarchy, or systemd's under v1."""
+    paths: Dict[int, str] = {}
+    for line in lines:
+        pid, _, entry = line.partition(" ")
+        parts = entry.split(":", 2)
+        if not pid.isdigit() or len(parts) != 3:
+            continue
+        hierarchy, controllers, path = parts
+        if (hierarchy == "0" and controllers == "") or controllers == "name=systemd":
+            paths[int(pid)] = path
+    return paths
+
+
+def _unit(path: Optional[str]) -> Optional[str]:
+    services = _SERVICE_RE.findall(path or "")
+    if services:
+        return str(services[-1])
+    procd = _PROCD_RE.match(path or "")
+    return str(procd.group(1)) if procd else None
+
+
+def _container(path: Optional[str]) -> Optional[str]:
+    match = _CONTAINER_RE.search(path or "")
+    return (match.group(1) or match.group(2)) if match else None
+
+
+def _entry(
+    proto: str,
+    local: Tuple[str, Optional[str], int],
+    process: Optional[str],
+    pid: Optional[int],
+    paths: Dict[int, str],
+) -> ListeningSocketDict:
+    address, interface, port = local
+    path = paths.get(pid) if pid is not None else None
+    return {
+        "proto": proto,
+        "address": address,
+        "port": port,
+        "interface": interface,
+        "process": process,
+        "pid": pid,
+        "unit": _unit(path),
+        "container_id": _container(path),
+    }
+
+
+def _ss_socket(line: str, paths: Dict[int, str]) -> Optional[ListeningSocketDict]:
+    parts = line.split()
+    if len(parts) < 5 or parts[0] not in _PROTOCOLS:
+        return None
+    local = _split_local(parts[4])
+    if local is None:
+        return None
+    users = _USER_RE.findall(line)
+    process, pid = (users[0][0], int(users[0][1])) if users else (None, None)
+    return _entry(parts[0], local, process, pid, paths)
+
+
+def _netstat_socket(line: str, paths: Dict[int, str]) -> Optional[ListeningSocketDict]:
+    """``Proto Recv-Q Send-Q Local Foreign [State] PID/Program`` -- a UDP line
+    has no state, and ``-`` is a socket without a process."""
+    parts = line.split()
+    proto = _NETSTAT_PROTOCOLS.get(parts[0]) if parts else None
+    if proto is None or len(parts) < 6:
+        return None
+    local = _split_local(parts[3])
+    if local is None:
+        return None
+    process, pid = None, None
+    for token in parts[5:7]:
+        program = _PROGRAM_RE.match(token)
+        if program:
+            pid, process = int(program.group(1)), program.group(2).rstrip(":") or None
+            break
+    return _entry(proto, local, process, pid, paths)
+
+
+_PARSERS = {"ss": _ss_socket, "netstat": _netstat_socket}
+
+
+def parse_listening_sockets(output: str) -> List[ListeningSocketDict]:
+    """Parse what :data:`LISTENING_SOCKETS_COMMAND` printed, sorted by protocol,
+    port and address.
+
+    :raises ListeningSocketsUnavailable: when the host has neither ``ss`` nor ``netstat``.
+    :raises ValueError: when the output carries no intact report, or the tool failed.
+    """
+    sections = _sections(_frame(output))
+    if _NO_SS in sections:
+        raise ListeningSocketsUnavailable("the host has neither ss nor netstat")
+    tool = "netstat" if "netstat" in sections else "ss"
+    lines = sections.get(tool, [])
+    statuses = [m.group(1) for m in map(_RC_RE.match, lines) if m]
+    if not statuses or statuses[-1] != "0":
+        detail = " ".join(line for line in lines if not _RC_RE.match(line))[:200]
+        raise ValueError(f"{tool} did not list the sockets: {detail or 'no exit status'}")
+    paths = _cgroup_paths(sections.get("cgroups", []))
+    parse = _PARSERS[tool]
+    sockets = [s for s in (parse(line, paths) for line in lines) if s is not None]
+    return sorted(sockets, key=lambda s: (s["proto"], s["port"], s["address"], s["interface"] or ""))
+
+
+class ListeningSocketsMixin:
+    """Adds :meth:`get_listening_sockets` to a driver that can run a command on a Linux host.
+
+    With ``ss``, or ``netstat`` where there is none (OpenWrt's busybox).
+
+    The template form (README, "Function classes"): the command and its parse
+    are the same everywhere, so they are concrete here, and a driver supplies
+    only :meth:`_run_listening_sockets_command` -- how a command reaches its
+    host, and how it gains root there. Mixed in by the drivers that can, not by
+    :class:`~napalm_device_types.os.OSDriver`: a Windows host is an OS driver
+    too, and ``hasattr(driver, "get_listening_sockets")`` has to stay truthful.
+    """
+
+    if TYPE_CHECKING:  # pragma: no cover - declared for type checkers only
+
+        def _run_listening_sockets_command(self, command: str, *, privileged: bool) -> str:
+            """Run *command* on the host and return what it printed; as root
+            when *privileged*. The command is a single ``sh -c`` invocation."""
+            ...
+
+    def get_listening_sockets(self) -> ListeningSocketsDict:
+        """
+        Returns every listening TCP and bound UDP socket, with the process,
+        systemd service and container behind it.
+
+        * attributed (bool) - read as root, so every process is named
+        * sockets (list) - see :class:`~napalm_device_types.models.ListeningSocketDict`
+
+        Example::
+
+            {
+                "attributed": True,
+                "sockets": [
+                    {"proto": "tcp", "address": "0.0.0.0", "port": 5432,
+                     "interface": None, "process": "postgres", "pid": 812,
+                     "unit": "postgresql@16-main", "container_id": None},
+                ],
+            }
+
+        :raises ListeningSocketsUnavailable: if the host has neither ``ss`` nor ``netstat``.
+        :raises ValueError: if neither reading carried an intact report.
+        """
+        command = f"sh -c {quote(LISTENING_SOCKETS_COMMAND)}"
+        try:
+            output = self._run_listening_sockets_command(command, privileged=True)
+            return {"attributed": True, "sockets": parse_listening_sockets(output)}
+        except ValueError:
+            pass
+        output = self._run_listening_sockets_command(command, privileged=False)
+        return {"attributed": False, "sockets": parse_listening_sockets(output)}

napalm_device_types/models.py

@@ -60,11 +60,64 @@ class ServiceDict(TypedDict):
 
 
 class UpdateDict(TypedDict):
-    """A software package that has a newer version available in the package repository."""
+    """A software package that has a newer version available in the package repository.
+
+    ``origin`` and ``security`` are optional: a reader that cannot tell leaves
+    them out, and netOrk treats a missing ``security`` as unknown.
+    """
 
     name: str
     current_version: str
     new_version: str
+    #: Where the new version comes from, e.g. apt's suites "noble-updates,noble-security".
+    origin: NotRequired[Optional[str]]
+    #: True for a security update, False for a known other one, None when unknown.
+    security: NotRequired[Optional[bool]]
+
+
+class InterruptedPackageDict(TypedDict):
+    """A package an interrupted package manager run left halfway."""
+
+    name: str
+    #: dpkg's word for it: "half-installed", "unpacked", "half-configured",
+    #: "triggers-awaited", "triggers-pending", or "reinstall-required".
+    state: str
+
+
+class FilesystemUsageDict(TypedDict):
+    """How full one filesystem is, as ``df -Pk`` says, in bytes."""
+
+    #: Where it is mounted, e.g. "/" or "/boot".
+    mountpoint: str
+    total: int
+    used: int
+    #: What an ordinary user can still write: less than total minus used on a
+    #: filesystem that keeps blocks for root (ext4 keeps 5 %).
+    available: int
+
+
+class HostStatusDict(TypedDict):
+    """What a host says about its own patch state (``HostStatusMixin.get_host_status``)."""
+
+    #: True when the host needs a reboot to finish an update, None when it cannot tell.
+    reboot_required: Optional[bool]
+    #: Why, e.g. "kernel 6.8.0-142-generic installed, 6.8.0-139-generic running".
+    reboot_reason: Optional[str]
+    #: True when the host installs updates on its own (unattended-upgrades, dnf-automatic).
+    auto_updates: Optional[bool]
+    #: True when a package manager run was cut off and has to be finished
+    #: (``dpkg --configure -a``) before anything else installs; None when it
+    #: cannot tell, or a package manager is running right now.
+    package_manager_interrupted: NotRequired[Optional[bool]]
+    #: The packages that run left halfway; empty when only its journal says so.
+    interrupted_packages: NotRequired[List[InterruptedPackageDict]]
+    #: When the host's own automatic updates last ran successfully (Unix time).
+    auto_updates_last_success: NotRequired[Optional[int]]
+    #: How often they are meant to run, in seconds; 0 for every time the timer fires.
+    auto_updates_interval: NotRequired[Optional[int]]
+    #: How full ``/``, ``/boot`` and ``/var`` are, each filesystem once; None when
+    #: it could not be read.
+    filesystems: NotRequired[Optional[List[FilesystemUsageDict]]]
 
 
 # ---------------------------------------------------------------------------
@@ -316,6 +369,66 @@ class KernelFactsDict(TypedDict):
     config: Optional[Dict[str, str]]  # build configuration, set options only; quotes stripped
 
 
+class ListeningSocketDict(TypedDict):
+    """A TCP socket that listens, or a UDP socket that is bound, on the host.
+
+    One entry per socket as ``ss`` lists it. ``address`` is printed the way ss
+    prints it, without brackets or zone: ``0.0.0.0`` and ``::`` are every
+    address of their family, ``*`` every address of both. Whether that is
+    reachable from outside the host is the consumer's call.
+    """
+
+    proto: str  # "tcp" or "udp"
+    address: str  # "0.0.0.0", "::", "*", "127.0.0.1", "::ffff:127.0.0.1", ...
+    port: int
+    interface: Optional[str]  # the %zone a socket is bound to ("lo", "eth0"), if any
+    process: Optional[str]  # the first process holding it; None without one or without root
+    pid: Optional[int]
+    unit: Optional[str]  # the process's systemd service, without ".service"
+    container_id: Optional[str]  # the full container ID, for a container on the host network
+
+
+class ListeningSocketsDict(TypedDict):
+    """What ``ListeningSocketsMixin.get_listening_sockets`` read.
+
+    ``attributed`` is False when the reading ran without root: ``ss`` then names
+    only the login user's own processes, so a socket without a process means
+    "not told", not "the kernel's".
+    """
+
+    attributed: bool
+    sockets: List[ListeningSocketDict]
+
+
+class VirtualIpDict(TypedDict):
+    """An address the host's HA configuration lets float between machines.
+
+    Declared, not observed: whether the host holds the address right now is in
+    its interface addresses, not here. Only these fields leave the host -- never
+    a password, an etcd endpoint or a notify script.
+    """
+
+    mechanism: str  # "vip-manager" | "keepalived"
+    instance: str  # the vrrp_instance name, or the vip-manager unit without ".service"
+    address: str  # normalised: "10.7.224.10", "fd00::10"
+    prefix_length: Optional[int]  # None where the configuration gives none
+    interface: Optional[str]
+    group_key: Optional[str]  # vip-manager's trigger-key, or "vrid:<n>" for keepalived
+    running: Optional[bool]  # the unit is active
+
+
+class VirtualIpsDict(TypedDict):
+    """What ``VirtualIpsMixin.get_virtual_ips`` read, per mechanism.
+
+    A list is what was looked at -- ``[]`` a host without that mechanism. None is
+    "did not look": a configuration it could not read, an include it could not
+    follow, a host without systemd. A consumer keeps what it knew for a None.
+    """
+
+    vip_manager: Optional[List[VirtualIpDict]]
+    keepalived: Optional[List[VirtualIpDict]]
+
+
 class PortForwardDict(TypedDict):
     """A port the WAN side can reach, forwarded to a host inside.
 
@@ -763,62 +876,15 @@ class ApplyUpdatesResultDict(TypedDict):
     error: NotRequired[str]  # error message if success is False
 
 
-class DockerContainerDict(TypedDict):
-    """A Docker container entry as returned by ``docker ps -a``."""
+class ContainerEngineDict(TypedDict):
+    """One container engine a host offers (``container_engines()``).
 
-    id: str             # short container ID
-    name: str           # container name(s)
-    image: str          # image reference
-    image_version: str  # OCI org.opencontainers.image.version label; empty if absent
-    command: str        # entrypoint / command
-    created: str        # creation timestamp string
-    status: str         # human-readable status (e.g. "Up 3 hours")
-    ports: str          # port mapping string
-    state: str          # "running", "exited", "paused", etc.
+    Only how to reach it: which engine, and which API it speaks. What runs on
+    it is read and modelled above the driver, in netOrk (netork#765).
+    """
 
-
-class DockerImageDict(TypedDict):
-    """A local Docker image entry as returned by ``docker images``."""
-
-    id: str             # short image ID
-    repository: str     # image repository
-    tag: str            # image tag
-    size: str           # human-readable size string (e.g. "187MB")
-    created: str        # creation timestamp string
-    version: str        # OCI org.opencontainers.image.version label; empty if absent
-
-
-class DockerVolumeDict(TypedDict):
-    """A Docker volume entry as returned by ``docker volume ls``."""
-
-    name: str           # volume name
-    driver: str         # volume driver (e.g. "local")
-    mountpoint: str     # host filesystem path
-    scope: str          # "local" or "global"
-
-
-class DockerNetworkDict(TypedDict):
-    """A Docker network entry as returned by ``docker network ls``."""
-
-    id: str             # short network ID
-    name: str           # network name
-    driver: str         # network driver (e.g. "bridge", "host", "overlay")
-    scope: str          # network scope
-    ipv6: str           # "true" if IPv6 is enabled, "false" otherwise
-    internal: str       # "true" if the network is internal, "false" otherwise
-
-
-class DockerInfoDict(TypedDict):
-    """Return value of ``get_docker_info()``."""
-
-    available: bool                              # False if Docker is absent/inaccessible
-    permission_denied: NotRequired[bool]         # True when socket access is denied
-    version: NotRequired[str]                    # Docker Engine version string
-    containers: NotRequired[List[DockerContainerDict]]
-    images: NotRequired[List[DockerImageDict]]
-    volumes: NotRequired[List[DockerVolumeDict]]
-    networks: NotRequired[List[DockerNetworkDict]]
-    outdated_images: NotRequired[List[str]]      # image names with a newer remote digest
+    engine: str     # "docker"
+    api: str        # "docker-engine": the Docker Engine API over ``system dial-stdio``
 
 
 class DeviceActionResultDict(TypedDict):

napalm_device_types/os.py

@@ -19,7 +19,6 @@ from napalm_device_types.health_metrics import HealthMetricsMixin
 from napalm_device_types.models import (
     CronJobDict,
     DeviceActionResultDict,
-    DockerInfoDict,
     ProcessDict,
     SNMPConfigDict,
     UserDict,
@@ -193,97 +192,6 @@ class OSDriver(UpdateMixin, ServiceControlMixin, PackageManagementMixin, HealthM
             """
             ...
 
-        # ------------------------------------------------------------------
-        # Docker
-        # ------------------------------------------------------------------
-
-        def get_docker_info(self) -> DockerInfoDict:
-            """
-            Returns information about the local Docker environment.
-
-            If Docker is not installed or the current user lacks access to the
-            Docker socket, returns ``{"available": False}``.  When the user has
-            no socket permission, ``permission_denied`` is additionally set to
-            ``True``.
-
-            When Docker is available the dict contains:
-
-            * available (bool) - always ``True``
-            * version (string) - Docker Engine version string
-            * containers (list) - all containers (running and stopped), each with:
-
-              * id (string) - short container ID
-              * name (string) - container name(s)
-              * image (string) - image reference
-              * image_version (string) - OCI ``org.opencontainers.image.version`` label; empty if absent
-              * command (string) - entrypoint / command string
-              * created (string) - creation timestamp string
-              * status (string) - human-readable status (e.g. ``"Up 3 hours"``)
-              * ports (string) - port mapping string
-              * state (string) - ``"running"``, ``"exited"``, ``"paused"``, etc.
-
-            * images (list) - local images, each with:
-
-              * id (string) - short image ID
-              * repository (string) - image repository
-              * tag (string) - image tag
-              * size (string) - human-readable size string (e.g. ``"187MB"``)
-              * created (string) - creation timestamp string
-              * version (string) - OCI ``org.opencontainers.image.version`` label; empty if absent
-
-            * volumes (list) - Docker volumes, each with:
-
-              * name (string) - volume name
-              * driver (string) - volume driver
-              * mountpoint (string) - host filesystem path
-              * scope (string) - ``"local"`` or ``"global"``
-
-            * networks (list) - Docker networks, each with:
-
-              * id (string) - short network ID
-              * name (string) - network name
-              * driver (string) - network driver (e.g. ``"bridge"``, ``"host"``, ``"overlay"``)
-              * scope (string) - network scope
-              * ipv6 (string) - ``"true"`` if IPv6 is enabled
-              * internal (string) - ``"true"`` if the network is internal
-
-            * outdated_images (list of strings) - image names where the local digest
-              differs from the latest remote digest; empty list if all images are
-              current or update checks could not be performed.
-
-            Example::
-
-                # Docker not installed:
-                {"available": False}
-
-                # Docker installed, no socket permission:
-                {"available": False, "permission_denied": True}
-
-                # Docker available:
-                {
-                    "available": True,
-                    "version": "Docker version 27.3.1, build ce12230",
-                    "containers": [
-                        {
-                            "id": "a1b2c3d4e5f6",
-                            "name": "my-app",
-                            "image": "nginx:latest",
-                            "image_version": "1.27.0",
-                            "command": "nginx -g 'daemon off;'",
-                            "created": "2026-05-28 10:00:00 +0000 UTC",
-                            "status": "Up 3 days",
-                            "ports": "0.0.0.0:80->80/tcp",
-                            "state": "running",
-                        },
-                    ],
-                    "images": [...],
-                    "volumes": [],
-                    "networks": [...],
-                    "outdated_images": ["nginx:latest"],
-                }
-            """
-            ...
-
         # ------------------------------------------------------------------
         # Generic device actions
         # ------------------------------------------------------------------

napalm_device_types/package_updates.py

@@ -0,0 +1,111 @@
+# -*- coding: utf-8 -*-
+"""Pending package updates: which package, from where, and whether it is a security fix.
+
+A patch deadline -- "security updates within 14 days" -- needs to know which
+pending update is a security update. apt says so in the suite a candidate comes
+from (``noble-security``, ``stable-security``), dnf in its update advisories.
+Reading that is the same for every driver whose host runs apt or dnf, so the
+parsers live here once and a driver only carries the command across.
+
+apt: the suites a candidate comes from are its ``origin``; any suite ending in
+``-security`` makes it a security update. A security fix that a later
+``-updates`` build superseded shows only ``-updates`` and counts as not
+security -- netOrk's CVE matching is what catches those.
+"""
+
+from __future__ import annotations
+
+import re
+from typing import Dict, List, Set
+
+from napalm_device_types.models import UpdateDict
+from napalm_device_types.terminal import strip_terminal_codes
+
+#: Read-only, no root needed, in a fixed language so the parse holds, and
+#: through a pipe: without a terminal apt draws no progress and no terminal
+#: codes, one of which (``ESC >``) a screen-scraping transport took for a shell
+#: prompt and stopped reading at. Its exit status is printed inside the group,
+#: so it is apt's, not cat's.
+APT_UPGRADABLE_COMMAND = "{ LC_ALL=C apt list --upgradable 2>/dev/null; echo __APT_RC=$?; } | cat"
+
+#: Read-only. Lists the packages that a pending security advisory covers.
+DNF_SECURITY_COMMAND = "LC_ALL=C dnf updateinfo list --security --quiet 2>/dev/null"
+
+# openssl/noble-updates,noble-security 3.0.13-0ubuntu3.6 amd64 [upgradable from: 3.0.13-0ubuntu3.5]
+_APT_LINE = re.compile(r"^(\S+)/(\S+)\s+(\S+)\s+\S+\s+\[upgradable from:\s+(\S+)\]")
+_SECURITY_SUITE = "-security"
+_APT_STATUS = re.compile(r"^__APT_RC=(\d+)\s*$", re.MULTILINE)
+
+
+def _joined_lines(output: str) -> List[str]:
+    """Lines as apt printed them: a terminal wraps long ones, and the
+    continuation starts with a space."""
+    lines: List[str] = []
+    for line in output.splitlines():
+        if line.startswith(" ") and lines:
+            lines[-1] += line.strip()
+        else:
+            lines.append(line)
+    return lines
+
+
+def _apt_listing(output: str) -> str:
+    """The listing without its exit status, or ``ValueError`` when apt failed or
+    the output was cut short -- "could not read" must never look like "nothing
+    pending"."""
+    text = strip_terminal_codes(output)
+    statuses = _APT_STATUS.findall(text)
+    if not statuses:
+        raise ValueError("apt list --upgradable reported no exit status; the output was cut short")
+    if statuses[-1] != "0":
+        raise ValueError(f"apt list --upgradable failed with exit status {statuses[-1]}")
+    return _APT_STATUS.sub("", text)
+
+
+def parse_apt_upgradable(output: str) -> List[UpdateDict]:
+    """Parse :data:`APT_UPGRADABLE_COMMAND`'s output, one entry per package.
+
+    A package listed for several architectures (``libc6`` for amd64 and i386)
+    is one entry; it counts as a security update if any of its lines does.
+
+    :raises ValueError: when apt failed or its exit status never arrived.
+    """
+    by_name: Dict[str, UpdateDict] = {}
+    for line in _joined_lines(_apt_listing(output)):
+        match = _APT_LINE.match(line)
+        if not match:
+            continue
+        name, listed, new_version, current_version = match.groups()
+        suites = list(dict.fromkeys(listed.split(",")))  # apt may list a suite twice
+        security = any(suite.endswith(_SECURITY_SUITE) for suite in suites)
+        seen = by_name.get(name)
+        if seen is not None:
+            seen["security"] = bool(seen.get("security")) or security
+            continue
+        by_name[name] = {
+            "name": name,
+            "current_version": current_version,
+            "new_version": new_version,
+            "origin": ",".join(suites),
+            "security": security,
+        }
+    return list(by_name.values())
+
+
+def nevra_name(nevra: str) -> str:
+    """The package name of an RPM ``name-[epoch:]version-release.arch``."""
+    without_arch = nevra.rsplit(".", 1)[0]
+    return without_arch.rsplit("-", 2)[0]
+
+
+def parse_dnf_security(output: str) -> Set[str]:
+    """The names of the packages a pending security advisory covers.
+
+    Parses :data:`DNF_SECURITY_COMMAND`'s ``ADVISORY  SEVERITY/Sec.  NEVRA`` lines.
+    """
+    names: Set[str] = set()
+    for line in output.splitlines():
+        parts = line.split()
+        if len(parts) >= 3 and parts[1].endswith("/Sec."):
+            names.add(nevra_name(parts[-1]))
+    return names

napalm_device_types/provisioning.py

@@ -0,0 +1,64 @@
+"""
+What provisioning a VM from a cloud image needs to know about the guest.
+
+The same for every hypervisor: which agent a guest runs so a QEMU-based
+hypervisor can read its IP, the network-config cloud-init and FreeBSD's
+nuageinit both read, and how a packed cloud image is unpacked. How a
+hypervisor attaches, boots and talks to the VM stays in its driver.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Mapping
+from typing import Any
+
+#: ``{guest_os: (packages, runcmd)}``: what cloud-init installs and runs so
+#: the hypervisor can read the guest's IP address back.
+GuestAgents = Mapping[str, tuple[tuple[str, ...], tuple[str, ...]]]
+
+#: The QEMU guest agent per guest OS, for hypervisors built on QEMU. Package
+#: names and service commands were verified on FreeBSD 15.1 and OpenBSD 7.9
+#: cloud images (NetOrk/netork#793).
+QEMU_GUEST_AGENTS: GuestAgents = {
+    "linux": (("qemu-guest-agent",), ("systemctl enable --now qemu-guest-agent",)),
+    "freebsd": (
+        ("qemu-guest-agent",),
+        ("sysrc qemu_guest_agent_enable=YES", "service qemu-guest-agent start"),
+    ),
+    "openbsd": (("qemu-ga",), ("rcctl enable qemu_ga", "rcctl start qemu_ga")),
+}
+
+#: Suffix of a packed image -> the command that writes it unpacked to stdout.
+_DECOMPRESSORS = {
+    "xz": "xz -dc",
+    "gz": "gzip -dc",
+    "bz2": "bzip2 -dc",
+    "zst": "zstd -dc",
+}
+
+
+def network_config(nics: list[tuple[str, bool]]) -> dict[str, Any] | None:
+    """Network-config v2: DHCP on every ``(mac, dhcp)`` NIC that asks for it.
+
+    Version 2 because FreeBSD's nuageinit reads no other: given the v1 that
+    Proxmox generates, it fails and skips the rest of its first stage,
+    runcmd included. ``None`` when no NIC wants DHCP, so the guest is left
+    alone rather than told to configure nothing.
+    """
+    ethernets = {
+        f"nic{i}": {"match": {"macaddress": mac.lower()}, "dhcp4": True}
+        for i, (mac, dhcp) in enumerate(nics)
+        if dhcp
+    }
+    return {"version": 2, "ethernets": ethernets} if ethernets else None
+
+
+def split_compression(filename: str) -> tuple[str, str | None]:
+    """``(unpacked filename, command that unpacks to stdout)`` for *filename*.
+
+    The command is None for an image that is not packed, which is then
+    imported as it is.
+    """
+    stem, dot, suffix = filename.rpartition(".")
+    command = _DECOMPRESSORS.get(suffix.lower()) if dot and stem else None
+    return (stem, command) if command else (filename, None)

napalm_device_types/ssh_server.py

@@ -0,0 +1,435 @@
+# -*- coding: utf-8 -*-
+"""OpenSSH server hardening: taking algorithms out of what sshd offers.
+
+A scanner such as ssh-audit names algorithms a server should stop offering --
+``hmac-sha1``, the NIST curves. Taking them out is the same on every host that
+runs OpenSSH, so the reading, the plan and the write live here once, and a
+driver only mixes in :class:`OpenSshServerMixin` and supplies ``run_command``.
+
+**What the server offers is read, not assumed.** ``sshd -T`` prints the lists
+sshd actually uses, after its defaults, ``sshd_config``, every ``Include`` and
+first-match-wins. The new list is that one minus what was asked, so nothing the
+host already restricted comes back, and nothing it allows is dropped by
+accident.
+
+**A drop-in, not an edit.** Where ``sshd_config`` includes
+``/etc/ssh/sshd_config.d/*.conf`` -- Debian and Ubuntu do, at the top -- the
+lists go to :data:`DROPIN_PATH`, which sorts first and so wins. sshd_config
+stays as the distribution shipped it, and removing the file undoes the change.
+Only without such an ``Include`` is sshd_config itself rewritten, after a copy.
+
+**Nobody is locked out, by three checks.**
+
+1. The plan never removes the last algorithm of a class, and -- given the
+   lists of the client that manages the host -- never the last one that client
+   speaks. Each one kept back is reported, with why.
+2. ``sshd -t`` must accept the result, and ``sshd -T`` must show the new lists.
+   A file read before ours can set the same directive; ours is then valid and
+   ignored, and that is reported rather than called a success.
+3. After the reload, *verify* -- the caller's -- opens a new session. The
+   running one survives a reload, so if the new one cannot get in, this one
+   puts the old state back and reloads again.
+"""
+
+from __future__ import annotations
+
+import time
+from dataclasses import dataclass, field
+from shlex import quote
+from typing import Callable, Dict, Iterable, List, Mapping, Optional, Sequence, Tuple, TYPE_CHECKING
+
+if TYPE_CHECKING:  # pragma: no cover - for type checkers only
+    from napalm_device_types.channel import CommandResult
+
+#: The algorithm classes: key -> (``sshd -T`` keyword, sshd_config directive).
+SSH_ALGORITHM_CLASSES: Dict[str, Tuple[str, str]] = {
+    "kex": ("kexalgorithms", "KexAlgorithms"),
+    "hostkey": ("hostkeyalgorithms", "HostKeyAlgorithms"),
+    "cipher": ("ciphers", "Ciphers"),
+    "mac": ("macs", "MACs"),
+}
+
+MAIN_CONFIG = "/etc/ssh/sshd_config"
+DROPIN_DIR = "/etc/ssh/sshd_config.d"
+#: ``00-`` so it sorts before every other drop-in: sshd keeps the first value it reads.
+DROPIN_PATH = f"{DROPIN_DIR}/00-netork-ssh-hardening.conf"
+#: What marks a line this module disabled in sshd_config.
+DISABLED_PREFIX = "# netork: "
+
+_HEADER = (
+    "# Written by netOrk: the SSH algorithms a security assessment flagged,\n"
+    "# taken out of what this server offers. Delete this file to undo it.\n"
+)
+
+_RELOAD = (
+    "if systemctl is-active --quiet ssh 2>/dev/null; then systemctl reload ssh;"
+    " elif systemctl is-active --quiet sshd 2>/dev/null; then systemctl reload sshd;"
+    " else for f in /run/sshd.pid /var/run/sshd.pid; do"
+    ' [ -r "$f" ] && kill -HUP "$(cat "$f")" && exit 0; done; exit 0; fi'
+)
+
+
+def _directive_key(line: str) -> Optional[str]:
+    """The algorithm class a global sshd_config line sets, if it sets one."""
+    parts = line.split(None, 1)
+    if not parts:
+        return None
+    word = parts[0].lower()
+    for kind, (keyword, _directive) in SSH_ALGORITHM_CLASSES.items():
+        if word == keyword:
+            return kind
+    return None
+
+
+def parse_sshd_effective(text: str) -> Dict[str, List[str]]:
+    """The four algorithm lists from ``sshd -T`` output, in the server's order.
+
+    :raises ValueError: if a list is missing -- not sshd -T output, or cut short.
+    """
+    found: Dict[str, List[str]] = {}
+    for raw in text.splitlines():
+        parts = raw.strip().split(None, 1)
+        if len(parts) != 2:
+            continue
+        kind = _directive_key(parts[0])
+        if kind and kind not in found:
+            found[kind] = [a for a in parts[1].split(",") if a]
+    missing = [k for k in SSH_ALGORITHM_CLASSES if k not in found]
+    if missing:
+        raise ValueError(f"sshd -T printed no {', '.join(missing)} list")
+    return found
+
+
+@dataclass(frozen=True)
+class Refusal:
+    """One algorithm that was asked to go and stays, with why.
+
+    *reason* is ``"last"`` (the class would be empty), ``"client"`` (the managing
+    client would share none of the class with the server) or ``"not_offered"``
+    (the server does not offer it, so there is nothing to remove).
+    """
+
+    kind: str
+    algorithm: str
+    reason: str
+
+
+@dataclass(frozen=True)
+class SshHardeningPlan:
+    keep: Dict[str, List[str]]
+    remove: Dict[str, List[str]]
+    refused: List[Refusal] = field(default_factory=list)
+    not_offered: List[Refusal] = field(default_factory=list)
+
+    @property
+    def changed_classes(self) -> List[str]:
+        return [k for k in SSH_ALGORITHM_CLASSES if self.remove.get(k)]
+
+
+def plan_ssh_hardening(
+    effective: Mapping[str, Sequence[str]],
+    remove: Mapping[str, Iterable[str]],
+    *,
+    client: Optional[Mapping[str, Iterable[str]]] = None,
+) -> SshHardeningPlan:
+    """What the lists become when *remove* is taken out of *effective*.
+
+    *client* is what the client managing this host speaks, per class. A class it
+    has no list for is not guarded beyond "never empty".
+
+    When a removal would leave a class empty, or without an algorithm *client*
+    speaks, the first of the asked-for algorithms in the server's order that
+    fixes it stays, and is reported in ``refused``.
+
+    :raises ValueError: for a class that is not one of :data:`SSH_ALGORITHM_CLASSES`.
+    """
+    unknown = [k for k in remove if k not in SSH_ALGORITHM_CLASSES]
+    if unknown:
+        raise ValueError(f"unknown SSH algorithm class: {', '.join(unknown)}")
+    keep: Dict[str, List[str]] = {}
+    removed: Dict[str, List[str]] = {}
+    refused: List[Refusal] = []
+    not_offered: List[Refusal] = []
+    for kind in SSH_ALGORITHM_CLASSES:
+        offered = list(effective.get(kind, []))
+        asked = set(remove.get(kind, []))
+        not_offered += [Refusal(kind, a, "not_offered") for a in sorted(asked - set(offered))]
+        going = [a for a in offered if a in asked]
+        staying = [a for a in offered if a not in asked]
+        speaks = set(client[kind]) if client and kind in client else None
+        if going and speaks is not None and not speaks & set(staying):
+            spared = next((a for a in going if a in speaks), None)
+            if spared:
+                going.remove(spared)
+                refused.append(Refusal(kind, spared, "client"))
+        if going and not [a for a in offered if a not in going]:
+            spared = going.pop(0)
+            refused.append(Refusal(kind, spared, "last"))
+        keep[kind] = [a for a in offered if a not in going]
+        if going:
+            removed[kind] = going
+    return SshHardeningPlan(keep=keep, remove=removed, refused=refused, not_offered=not_offered)
+
+
+def render_dropin(keep: Mapping[str, Sequence[str]], kinds: Iterable[str]) -> str:
+    """The drop-in file stating the full list of each class in *kinds*."""
+    lines = [
+        f"{SSH_ALGORITHM_CLASSES[k][1]} {','.join(keep[k])}\n"
+        for k in SSH_ALGORITHM_CLASSES
+        if k in set(kinds)
+    ]
+    return _HEADER + "".join(lines)
+
+
+def dropin_classes(text: str) -> List[str]:
+    """The classes an existing drop-in sets."""
+    return [k for k in (_directive_key(line.strip()) for line in text.splitlines()) if k]
+
+
+def _is_match(line: str) -> bool:
+    parts = line.split(None, 1)
+    return bool(parts) and parts[0].lower() == "match"
+
+
+def has_dropin_include(text: str) -> bool:
+    """Whether sshd_config's global section includes :data:`DROPIN_DIR`."""
+    for raw in text.splitlines():
+        line = raw.strip()
+        if _is_match(line):
+            return False
+        parts = line.split()
+        if len(parts) >= 2 and parts[0].lower() == "include":
+            if any(p.startswith(DROPIN_DIR + "/") for p in parts[1:]):
+                return True
+    return False
+
+
+def _split_our_block(text: str) -> Tuple[List[str], str]:
+    """(the classes an earlier rewrite put above the file, the file without them)."""
+    if not text.startswith(_HEADER):
+        return [], text
+    lines = text[len(_HEADER):].splitlines(keepends=True)
+    ours = 0
+    while ours < len(lines) and _directive_key(lines[ours].strip()):
+        ours += 1
+    return dropin_classes("".join(lines[:ours])), "".join(lines[ours:])
+
+
+def render_main_config(text: str, keep: Mapping[str, Sequence[str]], kinds: Iterable[str]) -> str:
+    """sshd_config with the lists of *kinds* first and their old global lines disabled.
+
+    First because sshd keeps the first value it reads. A ``Match`` block is left
+    alone -- it is not the global setting. A class an earlier rewrite set stays
+    set, and rewriting a file this already rewrote gives the same file.
+    """
+    earlier, body = _split_our_block(text)
+    wanted = set(kinds) | set(earlier)
+    out: List[str] = []
+    in_match = False
+    for raw in body.splitlines():
+        if _is_match(raw.strip()):
+            in_match = True
+        if not in_match and _directive_key(raw.strip()) in wanted:
+            out.append(DISABLED_PREFIX + raw)
+        else:
+            out.append(raw)
+    return render_dropin(keep, wanted) + "\n".join(out) + ("\n" if body.endswith("\n") else "")
+
+
+@dataclass
+class SshHardeningResult:
+    """What :meth:`OpenSshServerMixin.harden_ssh_server` did.
+
+    *applied* is True only when the new lists are live and a new session got
+    in. *rolled_back* says the host was changed and then put back. *backup* is
+    where sshd_config was copied to, when it was the file changed.
+    """
+
+    plan: SshHardeningPlan
+    applied: bool = False
+    rolled_back: bool = False
+    path: Optional[str] = None
+    backup: Optional[str] = None
+    message: str = ""
+
+
+class OpenSshServerMixin:
+    """Adds :meth:`harden_ssh_server` to a driver that can run commands as root on an OpenSSH host.
+
+    Mixed in by the drivers whose host runs a stock OpenSSH, not by a base class,
+    so ``hasattr(driver, "harden_ssh_server")`` stays a truthful answer. A driver
+    whose platform writes sshd_config itself (an appliance that regenerates it
+    from its own database) must not inherit this, or must opt out with
+    ``harden_ssh_server = None``: its file would be overwritten on the next
+    apply.
+    """
+
+    if TYPE_CHECKING:  # pragma: no cover - declared for type checkers only
+
+        def run_command(
+            self,
+            command: str,
+            *,
+            privileged: bool = False,
+            timeout: float = 60,
+            stdin: Optional[bytes] = None,
+        ) -> CommandResult: ...
+
+    # ── the host ────────────────────────────────────────────────────────
+
+    def _ssh_run(self, command: str, stdin: Optional[bytes] = None) -> "CommandResult":
+        return self.run_command(command, privileged=True, timeout=30, stdin=stdin)
+
+    def _ssh_read(self, path: str) -> Optional[str]:
+        result = self._ssh_run(f"cat -- {quote(path)}")
+        return result.stdout if result.exit_code == 0 else None
+
+    def _ssh_write(self, path: str, content: str) -> None:
+        script = (
+            f'tmp={quote(path + ".netork-tmp")} && cat > "$tmp" && chmod 644 "$tmp"'
+            f' && mv -f -- "$tmp" {quote(path)} || {{ s=$?; rm -f -- "$tmp"; exit $s; }}'
+        )
+        result = self._ssh_run(script, stdin=content.encode())
+        if result.exit_code != 0:
+            raise RuntimeError(f"could not write {path}: {(result.stderr or result.stdout).strip()}")
+
+    def _ssh_remove(self, path: str) -> None:
+        self._ssh_run(f"rm -f -- {quote(path)}")
+
+    def _reload_ssh_server(self) -> None:
+        """Make the running sshd read its configuration again; sessions stay up."""
+        result = self._ssh_run(_RELOAD)
+        if result.exit_code != 0:
+            raise RuntimeError(f"could not reload sshd: {(result.stderr or result.stdout).strip()}")
+
+    def get_ssh_server_algorithms(self) -> Dict[str, List[str]]:
+        """The algorithm lists sshd uses now (``sshd -T``), per class.
+
+        :raises RuntimeError: if sshd -T fails -- no OpenSSH, or no root.
+        """
+        result = self._ssh_run("PATH=$PATH:/usr/sbin:/sbin sshd -T")
+        if result.exit_code != 0:
+            raise RuntimeError(f"sshd -T failed: {(result.stderr or result.stdout).strip()}")
+        return parse_sshd_effective(result.stdout)
+
+    def _sshd_accepts(self) -> Optional[str]:
+        """None when ``sshd -t`` accepts the configuration, else what it said."""
+        result = self._ssh_run("PATH=$PATH:/usr/sbin:/sbin sshd -t")
+        return None if result.exit_code == 0 else (result.stderr or result.stdout).strip()
+
+    # ── the change ──────────────────────────────────────────────────────
+
+    def harden_ssh_server(
+        self,
+        remove: Mapping[str, Iterable[str]],
+        *,
+        client: Optional[Mapping[str, Iterable[str]]] = None,
+        verify: Optional[Callable[[], bool]] = None,
+        log: Optional[Callable[[str], None]] = None,
+    ) -> SshHardeningResult:
+        """Stop the server offering the algorithms in *remove* (class -> names).
+
+        *client* guards the managing client's algorithms (see
+        :func:`plan_ssh_hardening`); *verify* opens a new session after the
+        reload and returns whether it got in; *log* receives a line per step.
+
+        Never raises after the host was touched: a failure from there on puts
+        the previous state back and says so in the result.
+
+        :raises RuntimeError: if the current state cannot be read.
+        """
+        say = log or (lambda _line: None)
+        say("Reading the algorithms sshd offers (sshd -T)")
+        plan = plan_ssh_hardening(self.get_ssh_server_algorithms(), remove, client=client)
+        for r in plan.refused:
+            why = "it is the last one" if r.reason == "last" else "netOrk could no longer connect"
+            say(f"Keeping {r.algorithm}: {why}")
+        kinds = plan.changed_classes
+        result = SshHardeningResult(plan=plan)
+        if not kinds:
+            result.message = "Nothing to change: the server offers none of the selected algorithms."
+            say(result.message)
+            return result
+        for kind in kinds:
+            say(f"Removing {', '.join(plan.remove[kind])} from {SSH_ALGORITHM_CLASSES[kind][1]}")
+
+        main = self._ssh_read(MAIN_CONFIG)
+        if main is None:
+            raise RuntimeError(f"cannot read {MAIN_CONFIG}")
+        restore = self._write_change(plan, main, result, say)
+        return self._apply_or_restore(plan, kinds, result, restore, verify, say)
+
+    def _write_change(
+        self, plan: SshHardeningPlan, main: str, result: SshHardeningResult, say: Callable[[str], None]
+    ) -> Callable[[], None]:
+        """Write the new lists; return what puts the previous state back."""
+        if has_dropin_include(main):
+            previous = self._ssh_read(DROPIN_PATH)
+            kinds = set(plan.changed_classes) | set(dropin_classes(previous or ""))
+            self._ssh_run(f"mkdir -p -- {quote(DROPIN_DIR)}")
+            self._ssh_write(DROPIN_PATH, render_dropin(plan.keep, kinds))
+            result.path = DROPIN_PATH
+            say(f"Wrote {DROPIN_PATH}")
+
+            def restore() -> None:
+                if previous is None:
+                    self._ssh_remove(DROPIN_PATH)
+                else:
+                    self._ssh_write(DROPIN_PATH, previous)
+
+            return restore
+        backup = f"{MAIN_CONFIG}.netork-{time.strftime('%Y%m%d-%H%M%S')}.bak"
+        self._ssh_write(backup, main)
+        self._ssh_write(MAIN_CONFIG, render_main_config(main, plan.keep, plan.changed_classes))
+        result.path, result.backup = MAIN_CONFIG, backup
+        say(f"{MAIN_CONFIG} has no Include of {DROPIN_DIR}; rewrote it (copy at {backup})")
+
+        def restore_main() -> None:
+            self._ssh_write(MAIN_CONFIG, main)
+
+        return restore_main
+
+    def _apply_or_restore(
+        self,
+        plan: SshHardeningPlan,
+        kinds: List[str],
+        result: SshHardeningResult,
+        restore: Callable[[], None],
+        verify: Optional[Callable[[], bool]],
+        say: Callable[[str], None],
+    ) -> SshHardeningResult:
+        def back(message: str, reload: bool) -> SshHardeningResult:
+            say(message)
+            say("Putting the previous configuration back")
+            restore()
+            if reload:
+                self._reload_ssh_server()
+            result.rolled_back = True
+            result.message = message
+            return result
+
+        try:
+            say("Checking the configuration (sshd -t)")
+            complaint = self._sshd_accepts()
+            if complaint:
+                return back(f"sshd rejected the new configuration: {complaint}", reload=False)
+            now = self.get_ssh_server_algorithms()
+            ignored = [SSH_ALGORITHM_CLASSES[k][1] for k in kinds if now[k] != plan.keep[k]]
+            if ignored:
+                return back(
+                    f"sshd does not use the new {', '.join(ignored)}: a setting read before"
+                    f" {result.path} decides them",
+                    reload=False,
+                )
+            say("Reloading sshd")
+            self._reload_ssh_server()
+            if verify is not None:
+                say("Opening a new SSH session to check netOrk still gets in")
+                if not verify():
+                    return back("A new SSH session could not log in after the reload", reload=True)
+        except Exception as exc:  # noqa: BLE001 - whatever went wrong, the host goes back
+            return back(f"The change failed: {exc}", reload=True)
+        result.applied = True
+        result.message = "The server no longer offers the selected algorithms."
+        say(result.message)
+        return result

napalm_device_types/systemd.py

@@ -35,6 +35,7 @@ from typing import Any, Dict, List, Set, Tuple, TYPE_CHECKING
 
 from napalm_device_types.models import ServiceDict
 from napalm_device_types.services import ServiceControlMixin
+from napalm_device_types.terminal import strip_terminal_codes
 
 _BEGIN = "SVC_BEGIN"
 _END = "SVC_END"
@@ -88,9 +89,6 @@ _RC_RE = re.compile(rf"^{_RC_MARKER}(\d+)\s*$", re.MULTILINE)
 _UNIT_RE = re.compile(r"(?:[A-Za-z0-9_.:@-]|\\x[0-9A-Fa-f]{2})+")
 _MAX_UNIT_LENGTH = 255
 
-#: Terminal colour codes, which systemctl adds when a transport gives it a terminal.
-_ANSI_RE = re.compile(r"\x1b\[[0-9;?]*[A-Za-z]")
-
 _ENABLED = frozenset({"enabled", "enabled-runtime"})
 #: Unit file states of a service that is installed but need not be loaded.
 _INSTALLED = frozenset({"enabled", "enabled-runtime", "disabled", "indirect"})
@@ -139,7 +137,7 @@ def parse_action_result(output: str) -> Dict[str, Any]:
     Only the exit status decides. A job still running when ``timeout`` gave up
     is not reported as done, and output without a status is no success.
     """
-    output = _ANSI_RE.sub("", output)
+    output = strip_terminal_codes(output)
     statuses = _RC_RE.findall(output)
     text = _RC_RE.sub("", output).strip()
     if not statuses:
@@ -154,7 +152,7 @@ def parse_action_result(output: str) -> Dict[str, Any]:
 
 
 def _frame(output: str) -> List[str]:
-    lines = [line.strip() for line in _ANSI_RE.sub("", output).splitlines()]
+    lines = [line.strip() for line in strip_terminal_codes(output).splitlines()]
     try:
         start = lines.index(_BEGIN)
         end = lines.index(_END, start)

napalm_device_types/terminal.py

@@ -0,0 +1,23 @@
+# -*- coding: utf-8 -*-
+"""What a pseudo-terminal adds to a command's output, taken out again.
+
+A screen-scraping transport (netmiko) gives the remote command a terminal. Tools
+then colour their output and draw progress: systemctl colours its errors, apt
+switches the keypad mode with ``ESC =`` / ``ESC >``. Every parser in this package
+reads the text without them.
+"""
+
+from __future__ import annotations
+
+import re
+
+#: CSI sequences (colours, cursor), OSC sequences (window titles) and the
+#: two-character escapes (``ESC =``, ``ESC >``, ``ESC (B``).
+_TERMINAL_CODES = re.compile(
+    r"\x1b(?:\[[0-?]*[ -/]*[@-~]|\][^\x07\x1b]*(?:\x07|\x1b\\)|\([0-9A-Za-z]|[=>78DEHMNOc])"
+)
+
+
+def strip_terminal_codes(text: str) -> str:
+    """*text* without terminal escape sequences."""
+    return _TERMINAL_CODES.sub("", text)

napalm_device_types/updates.py

@@ -13,7 +13,7 @@ this class in can never shadow a working implementation from a sibling base.
 
 from __future__ import annotations
 
-from typing import List, TYPE_CHECKING
+from typing import Any, Dict, List, TYPE_CHECKING
 
 from napalm_device_types.models import ApplyUpdatesResultDict, UpdateDict
 
@@ -30,6 +30,14 @@ class UpdateMixin:
             * name (string) - package name
             * current_version (string) - currently installed version
             * new_version (string) - version available in the repository
+            * origin (string, optional) - where it comes from, e.g. apt's suites
+            * security (bool or None, optional) - a security update; leave it
+              out or None when the source does not say
+
+            **An empty list means nothing is pending.** A reader that cannot
+            read -- no package index yet, an API that did not answer -- raises
+            instead: netOrk keeps "pending since" per package, and an empty
+            list for "don't know" would reset every one of those clocks.
 
             Example::
 
@@ -43,6 +51,16 @@ class UpdateMixin:
             """
             ...
 
+        def refresh_available_updates(self) -> Dict[str, Any]:
+            """
+            Refreshes the host's package index, so that :meth:`get_available_updates`
+            reports what the repositories offer now (``apt-get update``,
+            ``dnf makecache``, ``opkg update``, a firmware check). Installs nothing.
+
+            :returns: ``{"success": bool, "output": str}``
+            """
+            ...
+
         def apply_updates(self, packages: List[str]) -> ApplyUpdatesResultDict:
             """
             Upgrades the given packages to the newest available version.

napalm_device_types/virtual_ips.py

@@ -0,0 +1,406 @@
+# -*- coding: utf-8 -*-
+"""Virtual IPs: the addresses a host's HA configuration lets float between machines.
+
+vip-manager and keepalived put a service address on whichever member is master
+*right now*. The address list of a standby shows no trace of it; what every
+member has is the configuration that declares it. That is read the same way on
+every Linux host, so the command and its parse live here once, and a driver
+only carries the command across.
+
+**Declared, not held.** The reading says which addresses the configuration
+lets onto this host. Whether the host holds one at the moment is in its
+interface addresses, which the caller already has.
+
+**Filtered on the host.** The same files hold secrets -- vip-manager's etcd
+password, keepalived's ``auth_pass``, notify scripts with their arguments. The
+command therefore never prints a configuration: an awk program reads the unit's
+properties and files and prints allowlisted fields only. A secret cannot reach
+the caller's memory, its log or an exception text, because it never leaves the
+host.
+
+**What is read.**
+
+- *vip-manager*: every ``vip-manager.service`` and ``vip-manager@*.service`` unit
+  systemd knows. The address comes from the first of: a command-line flag
+  (``--ip``; 1.x's ``-ip=${VIP_IP}`` resolved through the unit's environment), the
+  environment (``VIP_IP``, an ``EnvironmentFile`` over ``Environment``), the
+  ``--config`` YAML (5.x's ``ip``). Mask, interface and trigger key alike.
+- *keepalived*: ``keepalived.service``, its ``-f``/``--use-file`` or
+  ``/etc/keepalived/keepalived.conf``. The ``virtual_ipaddress`` and
+  ``virtual_ipaddress_excluded`` entries of every ``vrrp_instance``, with the
+  instance's ``interface`` and ``virtual_router_id``. ``include`` is followed
+  four levels deep, a relative pattern from the including file's directory. A
+  pattern that is not a plain path glob is not followed: it would otherwise be
+  handed to a root shell.
+
+**None is "did not look".** A file that exists but cannot be read, an include
+that cannot be followed, a host without systemd: that mechanism is None, and the
+caller keeps what it knew. ``[]`` is a host that has none.
+
+**Root, and without it.** The configuration may be root's alone. The whole
+script goes to the host as one ``sh -c`` argument; when the privileged call
+brings no report back, it runs again without privilege and reads what it can.
+
+**An exec channel, not a terminal.** With its awk program the command is about
+4 kB, past the line an interactive shell reliably takes. A driver sends it the
+way it runs any command that needs no PTY -- napalm-linux's ``run_command``.
+"""
+
+from __future__ import annotations
+
+import ipaddress
+import re
+from dataclasses import dataclass, field
+from shlex import quote
+from typing import Callable, Dict, List, Optional, Set, Tuple, TYPE_CHECKING
+
+from napalm_device_types.models import VirtualIpDict, VirtualIpsDict
+from napalm_device_types.terminal import strip_terminal_codes
+
+_BEGIN = "VIPS_BEGIN"
+_END = "VIPS_END"
+_NO_SYSTEMD = "no-systemd"
+#: A record that says a mechanism was not fully read.
+_BLIND = frozenset({"unreadable", "too-deep"})
+_KEEPALIVED_CONF = "/etc/keepalived/keepalived.conf"
+_ENV_KEYS = "VIP_(IP|MASK|NETMASK|IFACE|INTERFACE|KEY|TRIGGER_KEY)"
+_CFG_KEYS = "(ip|netmask|mask|interface|iface|trigger-key|key)"
+
+#: One line of awk, fed a unit's ``systemctl show`` output with ``-v m=<mechanism>``.
+#: It prints records, never a line of configuration: ``active <state>``,
+#: ``arg <flag> <value>``, ``env <VIP_*> <value>``, ``cfg <key> <value>``,
+#: ``instance <name>`` / ``interface`` / ``virtual_router_id`` / ``vip <addr> [dev <if>]``
+#: / ``end``, and ``unreadable``/``too-deep``/``missing <path>``. No ``>`` but in
+#: ``2>/dev/null``: comparisons are written the other way round. ``\047`` is a
+#: single quote, so the program itself needs none.
+_AWK = " ".join(
+    (
+        r'function q(p) { return "\047" p "\047" }',
+        r'function ok(f) { if (f == "" || f ~ /\047/) { print "unreadable " f; return 0 }'
+        r' if (system("test -r " q(f)) == 0) return 1;'
+        r' if (system("test -e " q(f)) == 0) print "unreadable " f; else print "missing " f;'
+        r" return 0 }",
+        r"function rf(f, c,   l, r, k) { if (!ok(f)) return;"
+        r" while (0 < (r = (getline l < f))) {"
+        rf' if (c == "cfg" && l ~ /^[ \t]*{_CFG_KEYS}[ \t]*:/) {{ sub(/^[ \t]*/, "", l); k = l;'
+        r' sub(/[ \t]*:.*/, "", k); sub(/^[^:]*:[ \t]*/, "", l); print "cfg " k " " l }'
+        rf' else if (c == "env" && l ~ /^[ \t]*(export[ \t]+)?{_ENV_KEYS}[ \t]*=/) {{'
+        r' sub(/^[ \t]*(export[ \t]+)?/, "", l); k = l; sub(/[ \t]*=.*/, "", k);'
+        r' sub(/^[^=]*=[ \t]*/, "", l); print "env " k " " l } }'
+        r' if (r < 0) print "unreadable " f; close(f) }',
+        r'function push(n, a) { st[++sd] = n; if (n == "vrrp_instance") print "instance " a }',
+        r'function pop() { if (0 < sd) { if (st[sd] == "vrrp_instance") print "end"; sd-- } }',
+        r"function stmt(w, nw, f, d,   x, i) { if (!nw) return;"
+        r' if (w[1] == "include") { if (1 < nw) inc(w[2], f, d); return }'
+        r' if (sd == 1 && st[1] == "vrrp_instance") {'
+        r' if (1 < nw && (w[1] == "interface" || w[1] == "virtual_router_id")) print w[1] " " w[2];'
+        r" return }"
+        r' if (sd == 2 && st[1] == "vrrp_instance" && st[2] ~ /^virtual_ipaddress(_excluded)?$/) {'
+        r' x = "vip " w[1]; for (i = 2; i < nw; i++) if (w[i] == "dev") x = x " dev " w[i + 1];'
+        r" print x } }",
+        r"function rk(f, d,   l, r, n, t, k, w, nw) {"
+        r' if (4 < d || (f in seen)) { print "too-deep " f; return }'
+        r" if (!ok(f)) return; seen[f] = 1;"
+        r' while (0 < (r = (getline l < f))) { sub(/[#!].*/, "", l); gsub(/[{}]/, " & ", l);'
+        r' n = split(l, t, " "); nw = 0; for (k = 1; k <= n; k++) {'
+        r' if (t[k] == "{") { push(nw ? w[1] : "", 1 < nw ? w[2] : ""); nw = 0 }'
+        r' else if (t[k] == "}") { stmt(w, nw, f, d); nw = 0; pop() } else w[++nw] = t[k] }'
+        r" stmt(w, nw, f, d) }"
+        r' if (r < 0) print "unreadable " f; close(f); delete seen[f] }',
+        r"function inc(p, f, d,   c, g, dir, fs, n, i) {"
+        r' if (p !~ /^\//) { dir = f; sub(/[^\/]*$/, "", dir); p = dir p }'
+        r' if (p !~ /^[A-Za-z0-9_.\/*?-]+$/) { print "unreadable " p; return }'
+        r' c = "for g in " p "; do [ -e \"$g\" ] && echo \"$g\"; done"; n = 0;'
+        r" while (0 < (c | getline g)) fs[++n] = g; close(c);"
+        r" for (i = 1; i <= n; i++) rk(fs[i], d + 1) }",
+        r'/^ActiveState=/ { print "active " substr($0, 13); next }',
+        r'/^ExecStart=/ { s = $0; i = index(s, "argv[]="); if (!i) next; s = substr(s, i + 7);'
+        r' j = index(s, " ;"); if (j) s = substr(s, 1, j - 1); n = split(s, t, " ");'
+        r' for (k = 2; k <= n; k++) { f = t[k]; if (f !~ /^-/) continue; v = ""; x = index(f, "=");'
+        r" if (x) { v = substr(f, x + 1); f = substr(f, 1, x - 1) }"
+        r' else if (k < n && t[k + 1] !~ /^-/) v = t[++k]; sub(/^--?/, "", f);'
+        r" if (f ~ /^(config|ip|netmask|mask|interface|iface|trigger-key|key|f|use-file)$/) {"
+        r' print "arg " f " " v; a[f] = v } } next }',
+        r'/^EnvironmentFiles=/ { s = substr($0, 18); sub(/ \(ignore_errors=[a-z]*\)$/, "", s);'
+        r' if (s != "") ef[++ne] = s; next }',
+        r'/^Environment=/ { n = split(substr($0, 13), t, " "); for (k = 1; k <= n; k++) {'
+        rf' x = index(t[k], "="); if (x && substr(t[k], 1, x - 1) ~ /^{_ENV_KEYS}$/)'
+        r' print "env " substr(t[k], 1, x - 1) " " substr(t[k], x + 1) } next }',
+        rf'END {{ if (m == "keepalived") {{ p = "{_KEEPALIVED_CONF}";'
+        r' if ("use-file" in a) p = a["use-file"]; if ("f" in a) p = a["f"]; rk(p, 0) }'
+        r' else { for (k = 1; k <= ne; k++) rf(ef[k], "env");'
+        r' if ("config" in a) rf(a["config"], "cfg") } }',
+    )
+)
+
+#: One line, POSIX ``sh``, read-only. The frame markers are printed in two halves
+#: so that a transport which echoes the command does not show them early.
+VIRTUAL_IPS_COMMAND = (
+    "PATH=$PATH:/usr/sbin:/sbin; "
+    "printf '%s%s\\n' VIPS_ BEGIN; "
+    "if command -v systemctl >/dev/null 2>&1; then echo '[systemd]'; "
+    "for u in $(systemctl list-units --all --plain --no-legend --type=service "
+    "vip-manager.service 'vip-manager@*' keepalived.service 2>/dev/null | awk '{print $1}'); do "
+    "case $u in keepalived*) m=keepalived;; *) m=vip-manager;; esac; "
+    'echo "[$m $u]"; '
+    'systemctl show -p ActiveState -p ExecStart -p EnvironmentFiles -p Environment "$u" '
+    f"2>/dev/null | awk -v m=$m {quote(_AWK)}; done; "
+    "else echo '[no-systemd]'; fi; "
+    "printf '%s%s\\n' VIPS_ END"
+)
+
+_COMMENT_RE = re.compile(r"\s+#.*$")
+_VARIABLE_RE = re.compile(r"^\$\{?(\w+)\}?$")
+_Record = Tuple[str, str]
+
+
+def _frame(output: str) -> List[str]:
+    lines = [line.strip() for line in strip_terminal_codes(output).splitlines()]
+    try:
+        start = lines.index(_BEGIN)
+        end = lines.index(_END, start)
+    except ValueError:
+        raise ValueError("no intact virtual IP report in the output") from None
+    return lines[start + 1 : end]
+
+
+def _sections(lines: List[str]) -> List[Tuple[str, str, List[_Record]]]:
+    """``[<mechanism> <unit>]`` headers, each with its records as ``(kind, rest)``."""
+    sections: List[Tuple[str, str, List[_Record]]] = []
+    for line in lines:
+        if line.startswith("[") and line.endswith("]"):
+            mechanism, _, unit = line[1:-1].partition(" ")
+            sections.append((mechanism, unit, []))
+        elif line and sections:
+            kind, _, rest = line.partition(" ")
+            sections[-1][2].append((kind, rest.strip()))
+    return sections
+
+
+def _clean(value: str) -> str:
+    """Without a trailing ``# comment`` and the quotes around it."""
+    value = _COMMENT_RE.sub("", value.strip())
+    if len(value) >= 2 and value[0] == value[-1] and value[0] in "\"'":
+        value = value[1:-1]
+    return value.strip()
+
+
+def _address(text: str) -> Optional[Tuple[str, Optional[int]]]:
+    """``"10.0.0.10/24"`` -> ``("10.0.0.10", 24)``; None for what is no address."""
+    host, slash, prefix = text.partition("/")
+    try:
+        address = ipaddress.ip_address(host)
+    except ValueError:
+        return None
+    if not slash:
+        return str(address), None
+    if not prefix.isdigit() or int(prefix) > address.max_prefixlen:
+        return None
+    return str(address), int(prefix)
+
+
+def _prefix(mask: Optional[str]) -> Optional[int]:
+    """``"24"`` or ``"255.255.255.0"`` -> 24."""
+    if not mask:
+        return None
+    if mask.isdigit():
+        return int(mask) if int(mask) <= 128 else None
+    try:
+        return ipaddress.IPv4Network(f"0.0.0.0/{mask}").prefixlen
+    except ValueError:
+        return None
+
+
+def _running(records: List[_Record]) -> Optional[bool]:
+    states = [rest for kind, rest in records if kind == "active"]
+    return states[-1] == "active" if states else None
+
+
+def _vip_manager(unit: str, records: List[_Record]) -> List[VirtualIpDict]:
+    values: Dict[str, Dict[str, str]] = {"arg": {}, "env": {}, "cfg": {}}
+    for kind, rest in records:
+        if kind in values:
+            key, _, value = rest.partition(" ")
+            values[kind][key] = value
+    env = values["env"]
+
+    def resolve(value: str) -> Optional[str]:
+        value = _clean(value)
+        variable = _VARIABLE_RE.match(value)
+        if variable:
+            value = _clean(env.get(variable.group(1), ""))
+        return value if value and "$" not in value else None
+
+    def pick(args: Tuple[str, ...], envs: Tuple[str, ...], cfgs: Tuple[str, ...]) -> Optional[str]:
+        for kind, names in (("arg", args), ("env", envs), ("cfg", cfgs)):
+            for name in names:
+                if name in values[kind]:
+                    found = resolve(values[kind][name])
+                    if found:
+                        return found
+        return None
+
+    ip = pick(("ip",), ("VIP_IP",), ("ip",))
+    parsed = _address(ip) if ip else None
+    if parsed is None:
+        return []
+    address, prefix = parsed
+    mask = pick(("netmask", "mask"), ("VIP_NETMASK", "VIP_MASK"), ("netmask", "mask"))
+    return [
+        {
+            "mechanism": "vip-manager",
+            "instance": unit[: -len(".service")] if unit.endswith(".service") else unit,
+            "address": address,
+            "prefix_length": _prefix(mask) if mask else prefix,
+            "interface": pick(
+                ("interface", "iface"), ("VIP_INTERFACE", "VIP_IFACE"), ("interface", "iface")
+            ),
+            "group_key": pick(
+                ("trigger-key", "key"), ("VIP_TRIGGER_KEY", "VIP_KEY"), ("trigger-key", "key")
+            ),
+            "running": _running(records),
+        }
+    ]
+
+
+@dataclass
+class _Instance:
+    name: str
+    interface: Optional[str] = None
+    vrid: str = ""
+    vips: List[str] = field(default_factory=list)
+
+
+def _keepalived(unit: str, records: List[_Record]) -> List[VirtualIpDict]:
+    running = _running(records)
+    instances: List[_Instance] = []
+    current: Optional[_Instance] = None
+    for kind, rest in records:
+        if kind == "instance":
+            current = _Instance(rest or unit)
+            instances.append(current)
+        elif current is None:
+            continue
+        elif kind == "interface":
+            current.interface = rest or None
+        elif kind == "virtual_router_id":
+            current.vrid = rest
+        elif kind == "vip":
+            current.vips.append(rest)
+        elif kind == "end":
+            current = None
+    return [entry for instance in instances for entry in _instance_entries(instance, running)]
+
+
+def _instance_entries(instance: _Instance, running: Optional[bool]) -> List[VirtualIpDict]:
+    entries: List[VirtualIpDict] = []
+    for vip in instance.vips:
+        tokens = vip.split()
+        parsed = _address(tokens[0]) if tokens else None
+        if parsed is None:
+            continue
+        address, prefix = parsed
+        dev = tokens[tokens.index("dev") + 1] if "dev" in tokens[1:-1] else None
+        entries.append(
+            {
+                "mechanism": "keepalived",
+                "instance": instance.name,
+                "address": address,
+                "prefix_length": prefix if prefix is not None else _host_prefix(address),
+                "interface": dev or instance.interface,
+                "group_key": f"vrid:{instance.vrid}" if instance.vrid.isdigit() else None,
+                "running": running,
+            }
+        )
+    return entries
+
+
+def _host_prefix(address: str) -> int:
+    """keepalived's default for an entry without one: the single address."""
+    return ipaddress.ip_address(address).max_prefixlen
+
+
+_READERS: Dict[str, Tuple[str, Callable[[str, List[_Record]], List[VirtualIpDict]]]] = {
+    "vip-manager": ("vip_manager", _vip_manager),
+    "keepalived": ("keepalived", _keepalived),
+}
+
+
+def _order(entry: VirtualIpDict) -> Tuple[int, int, str]:
+    address = ipaddress.ip_address(entry["address"])
+    return address.version, int(address), entry["instance"]
+
+
+def parse_virtual_ips(output: str) -> VirtualIpsDict:
+    """Parse what :data:`VIRTUAL_IPS_COMMAND` printed, each mechanism's entries
+    sorted by address.
+
+    :raises ValueError: when the output carries no intact report.
+    """
+    sections = _sections(_frame(output))
+    if any(mechanism == _NO_SYSTEMD for mechanism, _, _ in sections):
+        return {"vip_manager": None, "keepalived": None}
+    found: Dict[str, List[VirtualIpDict]] = {"vip_manager": [], "keepalived": []}
+    blind: Set[str] = set()
+    for mechanism, unit, records in sections:
+        if mechanism not in _READERS:
+            continue
+        key, read = _READERS[mechanism]
+        if any(kind in _BLIND for kind, _ in records):
+            blind.add(key)
+        else:
+            found[key].extend(read(unit, records))
+    def result(key: str) -> Optional[List[VirtualIpDict]]:
+        return None if key in blind else sorted(found[key], key=_order)
+
+    return {"vip_manager": result("vip_manager"), "keepalived": result("keepalived")}
+
+
+class VirtualIpsMixin:
+    """Adds :meth:`get_virtual_ips` to a driver that can run a command on a Linux host.
+
+    The template form (README, "Function classes"): the command and its parse are
+    the same everywhere, so they are concrete here, and a driver supplies only
+    :meth:`_run_virtual_ips_command` -- how a command reaches its host, and how it
+    gains root there. Mixed in by the drivers that can, not by
+    :class:`~napalm_device_types.os.OSDriver`, so ``hasattr(driver,
+    "get_virtual_ips")`` stays a truthful answer.
+    """
+
+    if TYPE_CHECKING:  # pragma: no cover - declared for type checkers only
+
+        def _run_virtual_ips_command(self, command: str, *, privileged: bool) -> str:
+            """Run *command* on the host and return what it printed; as root
+            when *privileged*. The command is a single ``sh -c`` invocation of
+            about 4 kB: send it on an exec channel, not typed into a shell."""
+            ...
+
+    def get_virtual_ips(self) -> VirtualIpsDict:
+        """
+        Returns the virtual IPs the host's vip-manager and keepalived configuration
+        declares -- whether or not the host holds them at the moment.
+
+        * vip_manager (list or None) - see :class:`~napalm_device_types.models.VirtualIpDict`
+        * keepalived (list or None)
+
+        None for a mechanism means it could not be read; ``[]`` that the host has none.
+
+        Example::
+
+            {
+                "vip_manager": [
+                    {"mechanism": "vip-manager", "instance": "vip-manager",
+                     "address": "10.7.224.10", "prefix_length": 24, "interface": "ens7",
+                     "group_key": "/service/netork-db/leader", "running": True},
+                ],
+                "keepalived": [],
+            }
+
+        :raises ValueError: if neither reading carried an intact report.
+        """
+        command = f"sh -c {quote(VIRTUAL_IPS_COMMAND)}"
+        try:
+            return parse_virtual_ips(self._run_virtual_ips_command(command, privileged=True))
+        except ValueError:
+            pass
+        return parse_virtual_ips(self._run_virtual_ips_command(command, privileged=False))

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "napalm-device-types"
-version = "2.2.0"
+version = "4.5.0"
 description = "Abstract device-type base classes for NAPALM drivers"
 readme = "README.md"
 requires-python = ">=3.10"

tests/test_channel.py

@@ -0,0 +1,169 @@
+"""The command channel: how a driver runs a command and opens a byte stream.
+
+netOrk's container runtime driver speaks the Docker Engine API over a stream
+from ``docker system dial-stdio`` (NetOrk/netork#765). The driver only provides
+the way there; these are the SSH pieces every SSH driver can reuse. The
+paramiko channel is faked, so the tests pin behaviour, not paramiko.
+"""
+
+from __future__ import annotations
+
+import socket
+
+import pytest
+
+from napalm_device_types.channel import (
+    CommandResult,
+    ParamikoExecStream,
+    open_stream_on_transport,
+    run_on_transport,
+)
+
+
+class FakeChannel:
+    """Just enough of paramiko.Channel: scripted stdout/stderr chunks, an exit code."""
+
+    def __init__(self, stdout=(), stderr=(), exit_code=0, hang=False):
+        self.out = list(stdout)
+        self.err = list(stderr)
+        self.exit_code = exit_code
+        self.hang = hang
+        self.sent = b""
+        self.command = None
+        self.write_closed = False
+        self.closed = False
+        self.timeout = None
+
+    def exec_command(self, command):
+        self.command = command
+
+    def settimeout(self, timeout):
+        self.timeout = timeout
+
+    def sendall(self, data):
+        self.sent += data
+
+    def shutdown_write(self):
+        self.write_closed = True
+
+    def close(self):
+        self.closed = True
+
+    def recv_ready(self):
+        return bool(self.out)
+
+    def recv(self, n):
+        if self.out:
+            return self.out.pop(0)
+        if self.hang:
+            raise socket.timeout()
+        return b""
+
+    def recv_stderr_ready(self):
+        return bool(self.err)
+
+    def recv_stderr(self, n):
+        return self.err.pop(0) if self.err else b""
+
+    def exit_status_ready(self):
+        return not self.hang and not self.out and not self.err
+
+    def recv_exit_status(self):
+        return self.exit_code
+
+
+class FakeTransport:
+    def __init__(self, channel):
+        self.channel = channel
+
+    def open_session(self):
+        return self.channel
+
+
+def test_run_collects_stdout_stderr_and_the_exit_code():
+    ch = FakeChannel(stdout=[b"hel", b"lo\n"], stderr=[b"warn\n"], exit_code=3)
+
+    result = run_on_transport(FakeTransport(ch), "echo hello", timeout=5)
+
+    assert result == CommandResult(stdout="hello\n", stderr="warn\n", exit_code=3)
+    assert ch.command == "echo hello"
+    assert ch.closed
+
+
+def test_run_sends_stdin_and_then_closes_the_write_side():
+    ch = FakeChannel(stdout=[b"ok"])
+
+    run_on_transport(FakeTransport(ch), "sudo -S true", stdin=b"pw\n", timeout=5)
+
+    assert ch.sent == b"pw\n"
+    assert ch.write_closed
+
+
+def test_run_raises_when_the_command_does_not_finish_in_time():
+    ch = FakeChannel(hang=True)
+
+    with pytest.raises(TimeoutError, match="did not finish"):
+        run_on_transport(FakeTransport(ch), "sleep 999", timeout=0.05)
+    assert ch.closed
+
+
+def test_a_stream_reads_until_eof_and_drains_stderr_on_the_way():
+    ch = FakeChannel(stdout=[b"HTTP/1.1 200 OK\r\n"], stderr=[b"note\n"])
+    stream = open_stream_on_transport(FakeTransport(ch), "docker system dial-stdio")
+
+    assert stream.read(4096, timeout=5) == b"HTTP/1.1 200 OK\r\n"
+    assert stream.read(4096, timeout=5) == b""
+    assert stream.stderr == "note\n"
+    assert ch.command == "docker system dial-stdio"
+
+
+def test_a_stream_writes_and_closes_its_write_side_only_when_asked():
+    """dial-stdio answers HTTP 499 to a request whose writer closed early
+    (netork#771), so write() never implies close_write()."""
+    ch = FakeChannel()
+    stream = open_stream_on_transport(FakeTransport(ch), "docker system dial-stdio")
+
+    stream.write(b"GET /_ping HTTP/1.1\r\n\r\n")
+    assert not ch.write_closed
+
+    stream.close_write()
+    assert ch.write_closed and ch.sent.startswith(b"GET /_ping")
+
+
+def test_a_stream_sends_its_stdin_prefix_first():
+    """How a sudo password reaches `sudo -S` before the stream's own bytes."""
+    ch = FakeChannel()
+
+    open_stream_on_transport(FakeTransport(ch), "sudo -S -p '' cmd", stdin_prefix=b"pw\n")
+
+    assert ch.sent == b"pw\n"
+
+
+def test_a_stream_times_out_as_timeout_error():
+    stream = ParamikoExecStream(FakeChannel(hang=True))
+
+    with pytest.raises(TimeoutError):
+        stream.read(10, timeout=0.01)
+
+
+def test_the_stream_keeps_only_a_bounded_tail_of_stderr():
+    ch = FakeChannel(stderr=[b"x" * 100, b"y" * 100])
+    stream = ParamikoExecStream(ch, stderr_limit=50)
+
+    assert stream.stderr == "y" * 50
+
+
+def test_exit_status_is_none_while_running_and_the_code_after():
+    running = ParamikoExecStream(FakeChannel(hang=True))
+    finished = ParamikoExecStream(FakeChannel(exit_code=1))
+
+    assert running.exit_status is None
+    assert finished.exit_status == 1
+
+
+def test_close_closes_the_channel():
+    ch = FakeChannel()
+
+    ParamikoExecStream(ch).close()
+
+    assert ch.closed

tests/test_container_engine.py

@@ -0,0 +1,178 @@
+"""Access to a container engine: which engines a host has, and a way to each.
+
+This package abstracts the connection, the driver builds it, and netOrk does
+the talking (NetOrk/netork#765, decided 2026-10-07). So nothing here knows what
+a container is. It knows how to reach the engine's API (``dial-stdio``) and how
+to run the engine's CLI with arguments netOrk chooses, and the driver alone
+knows where that CLI lives.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from napalm_device_types import OSDriver
+from napalm_device_types.channel import CommandResult
+from napalm_device_types.container_engine import (
+    ContainerEngineConnection,
+    ContainerEngineMixin,
+    ContainerEngineUnavailable,
+)
+
+
+class FakeDriver(ContainerEngineMixin):
+    """A driver with a channel: records what it is asked to run."""
+
+    def __init__(self, present=("docker",), binary=None):
+        self.present = set(present)
+        self.binary = binary
+        self.commands = []
+        self.streams = []
+
+    def run_command(self, command, *, privileged=False, timeout=60, stdin=None):
+        self.commands.append((command, privileged, timeout, stdin))
+        found = any(command == f"command -v {b}" for b in self._binaries())
+        return CommandResult(stdout="/usr/bin/x\n" if found else "", stderr="", exit_code=0 if found else 1)
+
+    def open_stream(self, command, *, privileged=False):
+        self.streams.append((command, privileged))
+        return object()
+
+    def _binaries(self):
+        return {self._container_engine_binary(e) for e in self.present}
+
+    def _container_engine_binary(self, engine):
+        return self.binary or super()._container_engine_binary(engine)
+
+
+def test_a_host_with_docker_lists_it():
+    assert FakeDriver().container_engines() == [{"engine": "docker", "api": "docker-engine"}]
+
+
+def test_a_host_without_the_cli_lists_nothing():
+    assert FakeDriver(present=()).container_engines() == []
+
+
+def test_the_api_is_reached_through_dial_stdio():
+    driver = FakeDriver()
+
+    driver.open_container_engine("docker").open_api()
+
+    assert driver.streams == [("docker system dial-stdio", False)]
+
+
+def test_the_driver_decides_where_the_binary_lives():
+    """QNAP keeps docker under Container Station's path; netOrk never sees it."""
+    path = "/share/CACHEDEV1_DATA/.qpkg/container-station/bin/docker"
+    driver = FakeDriver(binary=path)
+
+    assert driver.container_engines() == [{"engine": "docker", "api": "docker-engine"}]
+    driver.open_container_engine("docker").open_api()
+
+    assert driver.streams == [(f"{path} system dial-stdio", False)]
+
+
+def test_cli_arguments_are_quoted_and_prefixed_by_the_binary():
+    driver = FakeDriver()
+    conn = driver.open_container_engine("docker")
+
+    conn.run_cli(["compose", "-f", "/srv/my stack/compose.yml", "config", "--format", "json"])
+
+    command, privileged, timeout, stdin = driver.commands[-1]
+    assert command == "docker compose -f '/srv/my stack/compose.yml' config --format json"
+    assert privileged is False and stdin is None
+
+
+def test_cli_stdin_and_timeout_reach_the_channel():
+    driver = FakeDriver()
+
+    driver.open_container_engine("docker").run_cli(["login", "--password-stdin"], stdin=b"x", timeout=30)
+
+    _command, _privileged, timeout, stdin = driver.commands[-1]
+    assert (timeout, stdin) == (30, b"x")
+
+
+def test_a_streamed_cli_call_can_merge_stderr():
+    """compose writes its progress to stderr; a merged stream carries both."""
+    driver = FakeDriver()
+    conn = driver.open_container_engine("docker")
+
+    conn.stream_cli(["pull", "redis:7-alpine"])
+    conn.stream_cli(["compose", "up", "-d", "db"], merge_stderr=True)
+
+    assert driver.streams == [
+        ("docker pull redis:7-alpine", False),
+        ("docker compose up -d db 2>&1", False),
+    ]
+
+
+def test_a_shell_metacharacter_in_an_argument_stays_an_argument():
+    driver = FakeDriver()
+
+    driver.open_container_engine("docker").run_cli(["inspect", "x; rm -rf /"])
+
+    assert driver.commands[-1][0] == "docker inspect 'x; rm -rf /'"
+
+
+def test_an_unknown_engine_is_refused():
+    with pytest.raises(ContainerEngineUnavailable):
+        FakeDriver().open_container_engine("rkt")
+
+
+def test_the_connection_names_its_engine():
+    conn = FakeDriver().open_container_engine("docker")
+
+    assert isinstance(conn, ContainerEngineConnection)
+    assert conn.engine == "docker"
+
+
+def test_an_os_driver_does_not_get_container_engine_access_by_role():
+    """Mixed in by the driver, not by OSDriver: a Windows host is an OS driver
+    too, and `hasattr(driver, "open_container_engine")` has to stay truthful."""
+    assert not issubclass(OSDriver, ContainerEngineMixin)
+    assert not hasattr(OSDriver, "open_container_engine")
+
+
+class TestPrivilegedCli:
+    """A refused engine call can be repeated as root (netork#773): the driver
+    knows the binary and how it gains root, so the connection carries it."""
+
+    def test_run_cli_passes_privileged_to_the_channel(self):
+        driver = FakeDriver()
+
+        driver.open_container_engine("docker").run_cli(["restart", "web"], privileged=True)
+
+        assert driver.commands == [("docker restart web", True, 120, None)]
+
+    def test_run_cli_is_unprivileged_by_default(self):
+        driver = FakeDriver()
+
+        driver.open_container_engine("docker").run_cli(["ps"])
+
+        assert driver.commands[0][1] is False
+
+    def test_stream_cli_passes_privileged_too(self):
+        driver = FakeDriver()
+
+        driver.open_container_engine("docker").stream_cli(["pull", "nginx"], privileged=True)
+
+        assert driver.streams == [("docker pull nginx", True)]
+
+
+class TestNoDockerModel:
+    """There is no container model here (NetOrk/netork#765): netOrk reads and
+    handles containers over the Engine API, so the CLI-shaped types and the
+    ``get_docker_info()`` declaration that served napalm-linux' old reading are gone."""
+
+    def test_the_docker_types_are_gone(self):
+        import napalm_device_types
+        from napalm_device_types import models
+
+        for name in ("DockerInfoDict", "DockerContainerDict", "DockerImageDict"):
+            assert not hasattr(models, name), name
+            assert not hasattr(napalm_device_types, name), name
+
+    def test_an_os_driver_declares_no_docker_reading(self):
+        from napalm_device_types import os as os_role
+
+        assert "get_docker_info" not in os_role.__dict__.get("OSDriver", object).__dict__

tests/test_host_status.py

@@ -0,0 +1,508 @@
+"""Host status: does the host need a reboot, and does it patch itself?
+
+A patch run that installed a new kernel has not closed anything until the host
+boots it, so "reboot required" is part of being patched. Whether the host
+installs updates on its own (unattended-upgrades, dnf-automatic) is what netOrk
+shows next to the window it governs. Both are read the same way on every Linux
+host, so the command and its parse live here once (netOrk MVP 5).
+
+The fixtures are the real states of six hosts on netOrk's test server.
+"""
+
+from __future__ import annotations
+
+import re
+import subprocess
+
+import pytest
+
+from napalm_device_types import OSDriver
+from napalm_device_types.host_status import (
+    BUSY_CHECK,
+    HOST_STATUS_COMMAND,
+    HostStatusMixin,
+    kernel_reboot_pending,
+    parse_host_status,
+)
+
+
+def _wire(
+    *,
+    reboot_file=False,
+    running="6.8.0-142-generic",
+    modules=("6.8.0-139-generic", "6.8.0-142-generic"),
+    needs_restarting=None,
+    periodic=None,
+    stamp=None,
+    timer="enabled",
+    dnf_timers=("not-found", "not-found"),
+    dpkg=None,
+    journal=(),
+    busy=False,
+):
+    lines = ["HSTAT_BEGIN"]
+    if reboot_file:
+        lines.append("[reboot-required]")
+    if needs_restarting is not None:
+        lines += ["[needs-restarting]", str(needs_restarting)]
+    lines += ["[kernel]", running, "[modules]", *modules]
+    if periodic is not None:
+        lines += ["[apt-config]", *periodic, "[upgrade-stamp]", *([str(stamp)] if stamp else [])]
+    if dpkg is not None:
+        if busy:
+            lines.append("[dpkg-busy]")
+        lines += ["[dpkg-journal]", *journal, "[dpkg-audit]", *dpkg]
+    lines += [
+        "[timers]",
+        f"apt-daily-upgrade.timer {timer}",
+        f"dnf-automatic.timer {dnf_timers[0]}",
+        f"dnf-automatic-install.timer {dnf_timers[1]}",
+        "HSTAT_END",
+    ]
+    return "\n".join(lines) + "\n"
+
+
+UNATTENDED = ['APT::Periodic::Update-Package-Lists "1";', 'APT::Periodic::Unattended-Upgrade "1";']
+
+
+class TestRebootRequired:
+    def test_the_reboot_required_file_says_so(self):
+        status = parse_host_status(_wire(reboot_file=True))
+
+        assert status["reboot_required"] is True
+        assert "reboot-required" in status["reboot_reason"]
+
+    def test_a_newer_installed_kernel_than_the_running_one(self):
+        """z2m-garden: running 6.8.0-139, 6.8.0-142 installed."""
+        status = parse_host_status(
+            _wire(
+                running="6.8.0-139-generic",
+                modules=("6.8.0-87-generic", "6.8.0-139-generic", "6.8.0-142-generic"),
+            )
+        )
+
+        assert status["reboot_required"] is True
+        assert "6.8.0-142-generic" in status["reboot_reason"]
+
+    def test_the_newest_kernel_running_needs_none(self):
+        """vault-01: 6.8.0-142 running and newest; 6.8.0-94 sorts below it."""
+        status = parse_host_status(
+            _wire(running="6.8.0-142-generic", modules=("6.8.0-94-generic", "6.8.0-142-generic"))
+        )
+
+        assert status["reboot_required"] is False
+        assert status["reboot_reason"] is None
+
+    def test_needs_restarting_exit_1_means_reboot(self):
+        assert parse_host_status(_wire(needs_restarting=1))["reboot_required"] is True
+
+    def test_needs_restarting_exit_0_does_not(self):
+        assert parse_host_status(_wire(needs_restarting=0))["reboot_required"] is False
+
+    def test_no_kernel_information_is_unknown(self):
+        """A container has no /lib/modules of its own."""
+        status = parse_host_status(_wire(modules=()))
+
+        assert status["reboot_required"] is None
+
+
+class TestKernelRebootPending:
+    @pytest.mark.parametrize(
+        ("running", "installed", "newer"),
+        [
+            # Raspberry Pi: two flavours side by side; only the running one counts.
+            (
+                "6.18.33+rpt-rpi-v8",
+                [
+                    "6.12.75+rpt-rpi-2712",
+                    "6.12.75+rpt-rpi-v8",
+                    "6.18.33+rpt-rpi-2712",
+                    "6.18.33+rpt-rpi-v8",
+                ],
+                None,
+            ),
+            # Debian (OMV): 7.1.8 installed while 7.1.3 runs.
+            (
+                "7.1.3+deb13-amd64",
+                ["6.12.57+deb13-amd64", "7.1.3+deb13-amd64", "7.1.8+deb13-amd64"],
+                "7.1.8+deb13-amd64",
+            ),
+            # Proxmox: 7.0.14-19 is newer than 7.0.2-6, numerically.
+            ("7.0.14-19-pve", ["7.0.14-19-pve", "7.0.2-6-pve"], None),
+            # Arch: the running kernel's modules were replaced by the upgrade.
+            ("6.10.5-arch1-1", ["6.10.9-arch1-1"], "6.10.9-arch1-1"),
+        ],
+    )
+    def test_the_newer_kernel_of_the_running_flavour(self, running, installed, newer):
+        assert kernel_reboot_pending(running, installed) == newer
+
+
+class TestAutoUpdates:
+    def test_unattended_upgrades_switched_on(self):
+        assert parse_host_status(_wire(periodic=UNATTENDED))["auto_updates"] is True
+
+    def test_apt_without_the_setting_does_not_patch_itself(self):
+        """Proxmox and Raspberry Pi OS: apt-config answers, the setting is absent."""
+        assert parse_host_status(_wire(periodic=[]))["auto_updates"] is False
+
+    def test_switched_off_by_zero(self):
+        off = ['APT::Periodic::Unattended-Upgrade "0";']
+
+        assert parse_host_status(_wire(periodic=off))["auto_updates"] is False
+
+    def test_a_disabled_timer_stops_it_even_when_configured(self):
+        status = parse_host_status(_wire(periodic=UNATTENDED, timer="disabled"))
+
+        assert status["auto_updates"] is False
+
+    def test_dnf_automatic(self):
+        status = parse_host_status(_wire(dnf_timers=("enabled", "not-found")))
+
+        assert status["auto_updates"] is True
+
+    def test_neither_apt_nor_dnf_is_unknown(self):
+        assert parse_host_status(_wire())["auto_updates"] is None
+
+
+class TestTheReport:
+    def test_a_cut_short_report_raises(self):
+        with pytest.raises(ValueError):
+            parse_host_status(_wire().replace("HSTAT_END\n", ""))
+
+    def test_the_frame_is_not_in_the_command_itself(self):
+        assert "HSTAT_BEGIN" not in HOST_STATUS_COMMAND
+        assert "HSTAT_END" not in HOST_STATUS_COMMAND
+
+    def test_it_runs_without_a_terminal(self):
+        """No colour codes from ls, nothing a screen scraper could take for a prompt."""
+        assert HOST_STATUS_COMMAND.rstrip().endswith("| cat")
+
+    def test_terminal_codes_are_dropped(self):
+        status = parse_host_status(
+            "\x1b[0m" + _wire(reboot_file=True).replace("HSTAT_END", "\x1b>HSTAT_END")
+        )
+
+        assert status["reboot_required"] is True
+
+    def test_it_changes_nothing(self):
+        for word in ("rm ", "apt-get ", "dnf install", "systemctl start", "reboot"):
+            assert word not in HOST_STATUS_COMMAND.replace("reboot-required", "")
+
+
+class _Driver(HostStatusMixin):
+    def __init__(self, reply: str) -> None:
+        self.reply = reply
+        self.commands: list = []
+
+    def _run_host_status_command(self, command: str) -> str:
+        self.commands.append(command)
+        return self.reply
+
+
+class TestHostStatusMixin:
+    def test_a_driver_supplies_only_the_transport(self):
+        driver = _Driver(_wire(reboot_file=True, periodic=UNATTENDED))
+
+        status = driver.get_host_status()
+
+        assert driver.commands == [HOST_STATUS_COMMAND]
+        assert (status["reboot_required"], status["auto_updates"]) == (True, True)
+
+    def test_not_every_os_driver_has_it(self):
+        assert not hasattr(OSDriver, "get_host_status")
+
+
+class TestFreeBSD:
+    """freebsd-version -k names the installed kernel, -r the running one; a
+    patch that replaced the kernel shows as the two differing until the host
+    boots it. Recorded on a FreeBSD 15.1 host (NetOrk/netork#799)."""
+
+    #: What the command printed there before this section existed.
+    BEFORE = (
+        "HSTAT_BEGIN\n[kernel]\n15.1-RELEASE\n[modules]\n[timers]\n"
+        "apt-daily-upgrade.timer \ndnf-automatic.timer \ndnf-automatic-install.timer \n"
+        "HSTAT_END\n"
+    )
+
+    @staticmethod
+    def _report(installed: str, running: str) -> str:
+        return (
+            f"HSTAT_BEGIN\n[kernel]\n{running}\n[modules]\n"
+            f"[freebsd-kernel]\n{installed}\n[freebsd-running]\n{running}\n"
+            "[timers]\napt-daily-upgrade.timer \nHSTAT_END\n"
+        )
+
+    def test_without_the_section_it_is_unknown(self):
+        assert parse_host_status(self.BEFORE)["reboot_required"] is None
+
+    def test_the_running_kernel_is_the_installed_one(self):
+        status = parse_host_status(self._report("15.1-RELEASE-p4", "15.1-RELEASE-p4"))
+        assert status["reboot_required"] is False
+        assert status["reboot_reason"] is None
+
+    def test_a_patched_kernel_waits_for_a_reboot(self):
+        status = parse_host_status(self._report("15.1-RELEASE-p5", "15.1-RELEASE-p4"))
+        assert status["reboot_required"] is True
+        assert status["reboot_reason"] == (
+            "kernel 15.1-RELEASE-p5 installed, 15.1-RELEASE-p4 running"
+        )
+
+    def test_freebsd_does_not_patch_itself_as_far_as_netork_can_tell(self):
+        status = parse_host_status(self._report("15.1-RELEASE", "15.1-RELEASE"))
+        assert status["auto_updates"] is None
+
+    def test_the_command_asks_only_where_freebsd_version_exists(self):
+        assert "command -v freebsd-version" in HOST_STATUS_COMMAND
+        assert "freebsd-version -k" in HOST_STATUS_COMMAND
+        assert "freebsd-version -r" in HOST_STATUS_COMMAND
+
+
+class TestInterruptedDpkg:
+    """dpkg left halfway: apt refuses to install anything until
+    ``dpkg --configure -a`` has run, and the daily apt timers report success
+    all the same (NetOrk/netork#738).
+
+    The fixture is aris (Ubuntu 24.04): unattended-upgrades was cut off on
+    2026-07-03 while update-initramfs built the initrd for 6.8.0-134, and the
+    kernel package stayed half-configured for three months.
+    """
+
+    ARIS = ("iF  linux-image-6.8.0-134-generic",)
+
+    def test_a_half_configured_kernel_is_an_interrupted_dpkg(self):
+        status = parse_host_status(_wire(dpkg=self.ARIS))
+
+        assert status["package_manager_interrupted"] is True
+        assert status["interrupted_packages"] == [
+            {"name": "linux-image-6.8.0-134-generic", "state": "half-configured"}
+        ]
+
+    def test_a_clean_dpkg_is_not_interrupted(self):
+        status = parse_host_status(_wire(dpkg=()))
+
+        assert status["package_manager_interrupted"] is False
+        assert status["interrupted_packages"] == []
+
+    @pytest.mark.parametrize(
+        ("line", "state"),
+        [
+            ("iH  base-files", "half-installed"),
+            ("iU  base-files", "unpacked"),
+            ("iF  base-files", "half-configured"),
+            ("iW  base-files", "triggers-awaited"),
+            ("it  base-files", "triggers-pending"),
+            ("iUR base-files", "reinstall-required"),
+            ("hF  base-files", "half-configured"),
+        ],
+    )
+    def test_every_state_an_interrupted_run_leaves(self, line, state):
+        status = parse_host_status(_wire(dpkg=(line,)))
+
+        assert status["interrupted_packages"] == [{"name": "base-files", "state": state}]
+
+    def test_a_journal_left_behind_is_interrupted_on_its_own(self):
+        """What apt itself checks: numbered files in /var/lib/dpkg/updates."""
+        status = parse_host_status(_wire(dpkg=(), journal=("0000", "0001")))
+
+        assert status["package_manager_interrupted"] is True
+        assert status["interrupted_packages"] == []
+
+    def test_only_numbered_journal_files_count(self):
+        """apt ignores anything else there, such as dpkg's tmp.i."""
+        status = parse_host_status(_wire(dpkg=(), journal=("tmp.i",)))
+
+        assert status["package_manager_interrupted"] is False
+
+    def test_a_running_package_manager_makes_it_unknown(self):
+        """Mid-upgrade, packages are unpacked and not yet configured, also
+        between two of apt's dpkg calls. That is work in progress, not an
+        interruption."""
+        status = parse_host_status(_wire(dpkg=("iU  base-files",), busy=True))
+
+        assert status["package_manager_interrupted"] is None
+        assert status["interrupted_packages"] == []
+
+    def test_without_dpkg_it_is_unknown(self):
+        status = parse_host_status(_wire())
+
+        assert status["package_manager_interrupted"] is None
+        assert status["interrupted_packages"] == []
+
+    def test_the_command_asks_dpkg_only_where_it_exists(self):
+        assert "command -v dpkg-query" in HOST_STATUS_COMMAND
+        assert "/var/lib/dpkg/updates" in HOST_STATUS_COMMAND
+
+    def test_it_never_repairs(self):
+        assert "--configure" not in HOST_STATUS_COMMAND
+
+
+class TestAutoUpdatesLastSuccess:
+    """apt.systemd.daily touches upgrade-stamp only after unattended-upgrade
+    succeeded. update-success-stamp says nothing: netOrk's own daily
+    ``apt-get update`` touches it as well (NetOrk/netork#738)."""
+
+    def test_the_upgrade_stamp_is_the_last_successful_run(self):
+        status = parse_host_status(_wire(periodic=UNATTENDED, stamp=1751525820))
+
+        assert status["auto_updates_last_success"] == 1751525820
+        assert status["auto_updates_interval"] == 86400
+
+    def test_no_stamp_is_unknown(self):
+        status = parse_host_status(_wire(periodic=UNATTENDED))
+
+        assert status["auto_updates_last_success"] is None
+
+    def test_without_apt_both_are_unknown(self):
+        status = parse_host_status(_wire())
+
+        assert status["auto_updates_last_success"] is None
+        assert status["auto_updates_interval"] is None
+
+    @pytest.mark.parametrize(
+        ("value", "seconds"),
+        [
+            ('"1"', 86400),
+            ('"7d"', 7 * 86400),
+            ('"12h"', 43200),
+            ('"30m"', 1800),
+            ('"90s"', 90),
+            ('"always"', 0),
+            ('"0"', None),
+            ('"weekly"', None),
+        ],
+    )
+    def test_the_interval_in_apts_units(self, value, seconds):
+        periodic = [f"APT::Periodic::Unattended-Upgrade {value};"]
+
+        assert parse_host_status(_wire(periodic=periodic))["auto_updates_interval"] == seconds
+
+    def test_the_command_reads_the_upgrade_stamp_only(self):
+        assert "/var/lib/apt/periodic/upgrade-stamp" in HOST_STATUS_COMMAND
+        assert "update-success-stamp" not in HOST_STATUS_COMMAND
+
+
+class TestBusyCheck:
+    """Whether a package manager is working right now, run through ``sh`` with a
+    stand-in ``ps``. On Ubuntu, unattended-upgrades keeps
+    ``unattended-upgrade-shutdown --wait-for-signal`` running all the time; its
+    process name is cut to 15 characters, ``unattended-upgr``, the same as a real
+    run's. Taken for a run, it made every such host's dpkg state unknown -- aris
+    among them (NetOrk/netork#738)."""
+
+    UBUNTU_IDLE = [
+        ("systemd", "/sbin/init"),
+        ("unattended-upgr", "/usr/bin/python3 /usr/share/unattended-upgrades/unattended-upgrade-shutdown --wait-for-signal"),
+        ("sshd", "sshd: netork [priv]"),
+    ]
+
+    @staticmethod
+    def _busy(tmp_path, processes) -> bool:
+        comm = "\n".join(c for c, _ in processes)
+        args = "\n".join(a for _, a in processes)
+        ps = tmp_path / "ps"
+        ps.write_text(
+            "#!/bin/sh\n"
+            f"case \"$*\" in *comm=*) printf '%s\\n' '{comm}';; *) printf '%s\\n' '{args}';; esac\n"
+        )
+        ps.chmod(0o755)
+        env = {"PATH": f"{tmp_path}:/usr/bin:/bin"}
+        out = subprocess.run(["sh", "-c", BUSY_CHECK], capture_output=True, text=True, env=env)
+        return "[dpkg-busy]" in out.stdout
+
+    def test_the_shutdown_helper_of_an_idle_ubuntu_host_is_no_run(self, tmp_path):
+        assert self._busy(tmp_path, self.UBUNTU_IDLE) is False
+
+    def test_unattended_upgrade_running(self, tmp_path):
+        run = ("unattended-upgr", "/usr/bin/python3 /usr/bin/unattended-upgrade")
+
+        assert self._busy(tmp_path, [*self.UBUNTU_IDLE, run]) is True
+
+    @pytest.mark.parametrize("name", ["dpkg", "apt", "apt-get", "aptitude"])
+    def test_dpkg_and_apt(self, tmp_path, name):
+        assert self._busy(tmp_path, [*self.UBUNTU_IDLE, (name, f"/usr/bin/{name} upgrade")]) is True
+
+    def test_it_is_part_of_the_command(self):
+        assert BUSY_CHECK in HOST_STATUS_COMMAND
+
+    def test_the_command_does_not_match_itself(self):
+        """Run over an exec channel, the command is the command line of the shell
+        that runs it, and ``ps -o args`` lists it."""
+        assert not re.search(r"(^|[ /])unattended-upgrade( |$)", HOST_STATUS_COMMAND)
+
+
+class TestFilesystems:
+    """How full the filesystems an upgrade writes to are: ``/``, ``/boot`` and
+    ``/var``, read with ``df -Pk`` and no root.
+
+    pve-dual's root filesystem stood at 94 % with about 400 MB free for a week;
+    a kernel upgrade failed on it with "No space left on device" and left dpkg
+    halfway, and nothing had said the disk was full (NetOrk/netork#862). The
+    figures below are shaped after it (6.5 GiB, 400 MiB available): ext4 keeps
+    5 % for root, so what is available is less than total minus used.
+    """
+
+    HEADER = "Filesystem 1024-blocks Used Available Capacity Mounted on"
+    PVE_ROOT = "/dev/mapper/pve-root 6815744 6065357 409600 94% /"
+    BOOT = "/dev/sda2 1046512 104860 941652 11% /boot"
+
+    @staticmethod
+    def _report(*df_lines: str) -> str:
+        return _wire().replace("[timers]", "\n".join(["[filesystems]", *df_lines, "[timers]"]))
+
+    def test_the_root_filesystem_in_bytes(self):
+        status = parse_host_status(self._report(self.HEADER, self.PVE_ROOT))
+
+        assert status["filesystems"] == [
+            {
+                "mountpoint": "/",
+                "total": 6815744 * 1024,
+                "used": 6065357 * 1024,
+                "available": 409600 * 1024,
+            }
+        ]
+
+    def test_a_separate_boot_partition(self):
+        status = parse_host_status(self._report(self.HEADER, self.PVE_ROOT, self.BOOT))
+
+        assert [fs["mountpoint"] for fs in status["filesystems"]] == ["/", "/boot"]
+
+    def test_a_directory_on_root_is_root_once(self):
+        """``df /boot /var`` names the filesystem they live on, here ``/`` three times."""
+        status = parse_host_status(
+            self._report(self.HEADER, self.PVE_ROOT, self.PVE_ROOT, self.PVE_ROOT)
+        )
+
+        assert [fs["mountpoint"] for fs in status["filesystems"]] == ["/"]
+
+    def test_a_long_device_name_and_a_blank_in_the_mountpoint(self):
+        line = "/dev/disk/by-id/a very long name 1000 600 400 60% /mnt/my disk"
+
+        (fs,) = parse_host_status(self._report(self.HEADER, line))["filesystems"]
+
+        assert (fs["mountpoint"], fs["available"]) == ("/mnt/my disk", 400 * 1024)
+
+    def test_a_filesystem_without_size_is_left_out(self):
+        """What a pseudo filesystem reports; it cannot run full."""
+        status = parse_host_status(self._report(self.HEADER, "none 0 0 0 - /"))
+
+        assert status["filesystems"] is None
+
+    def test_without_the_section_it_is_unknown(self):
+        """An older command, or no df: None, not "nothing is full"."""
+        assert parse_host_status(_wire())["filesystems"] is None
+
+    def test_df_that_printed_nothing_is_unknown(self):
+        assert parse_host_status(self._report())["filesystems"] is None
+
+    def test_the_command_asks_for_the_three_without_root(self):
+        assert "df -Pk / /boot /var" in HOST_STATUS_COMMAND
+        assert "sudo" not in HOST_STATUS_COMMAND
+
+    def test_the_command_reads_them_on_a_real_shell(self):
+        """POSIX df the way a host's sh runs it: ``/`` is always there, whatever it is."""
+        out = subprocess.run(
+            ["sh", "-c", HOST_STATUS_COMMAND], capture_output=True, text=True, timeout=30
+        )
+
+        mountpoints = [fs["mountpoint"] for fs in parse_host_status(out.stdout)["filesystems"]]
+        assert "/" in mountpoints

tests/test_listening_sockets.py

@@ -0,0 +1,396 @@
+"""get_listening_sockets: what listens on which address, and which service it is.
+
+Whether a service is reachable from outside its host is decided by what it
+listens on -- ``0.0.0.0:5432`` is, ``127.0.0.1:5432`` is not. Reading that is
+the same on every Linux host: ``ss`` for the sockets, ``/proc/<pid>/cgroup``
+for the systemd unit or container a process belongs to. So both live here once,
+and a driver only carries the command across (#658 in netOrk).
+
+A host without ``ss`` -- OpenWrt's busybox, an old net-tools box -- is read
+with ``netstat -lntup`` instead, and on OpenWrt the cgroup names the procd
+service (#673 in netOrk).
+"""
+
+from __future__ import annotations
+
+import os
+import shutil
+import subprocess
+
+import pytest
+
+from napalm_device_types import OSDriver
+from napalm_device_types.listening import (
+    LISTENING_SOCKETS_COMMAND,
+    ListeningSocketsMixin,
+    ListeningSocketsUnavailable,
+    parse_listening_sockets,
+)
+
+CID = "4f1c" + "0" * 60
+SS = """Netid State  Recv-Q Send-Q          Local Address:Port  Peer Address:PortProcess
+udp   UNCONN 0      0               127.0.0.53%lo:53         0.0.0.0:*    users:(("systemd-resolve",pid=612,fd=13))
+udp   UNCONN 0      0              0.0.0.0%ens18:68          0.0.0.0:*    users:(("systemd-network",pid=590,fd=22))
+tcp   LISTEN 0      4096                  0.0.0.0:5432        0.0.0.0:*    users:(("postgres",pid=812,fd=6))
+tcp   LISTEN 0      128                      [::]:22             [::]:*    users:(("sshd",pid=700,fd=4))
+tcp   LISTEN 0      511                         *:80                *:*    users:(("nginx",pid=901,fd=6),("nginx",pid=900,fd=6))
+tcp   LISTEN 0      4096       [::ffff:127.0.0.1]:8125              *:*    users:(("statsd",pid=950,fd=3))
+tcp   LISTEN 0      4096                  0.0.0.0:8080        0.0.0.0:*    users:(("docker-proxy",pid=1200,fd=4))
+tcp   LISTEN 0      4096                  0.0.0.0:9100        0.0.0.0:*    users:(("node_exporter",pid=1300,fd=3))
+tcp   LISTEN 0      64                    0.0.0.0:2049        0.0.0.0:*
+"""
+CGROUPS = f"""612 0::/system.slice/systemd-resolved.service
+590 0::/system.slice/systemd-networkd.service
+812 0::/system.slice/system-postgresql.slice/postgresql@16-main.service
+700 0::/system.slice/ssh.service
+900 0::/system.slice/nginx.service
+901 0::/system.slice/nginx.service
+950 0::/user.slice/user-1000.slice/session-3.scope
+1200 0::/system.slice/docker.service
+1300 0::/system.slice/docker-{CID}.scope
+"""
+
+
+def _wire(ss: str = SS, cgroups: str = CGROUPS, *, rc: int = 0, noise: str = "") -> str:
+    """The report as the command prints it, framed."""
+    return f"{noise}SOCK_BEGIN\n[ss]\n{ss}__SS_RC={rc}\n[cgroups]\n{cgroups}SOCK_END\n"
+
+
+def _by_port(sockets: list) -> dict:
+    return {(s["proto"], s["port"], s["address"]): s for s in sockets}
+
+
+class TestParsing:
+    def test_a_socket_comes_with_its_process_and_unit(self):
+        sockets = _by_port(parse_listening_sockets(_wire()))
+
+        assert sockets[("tcp", 5432, "0.0.0.0")] == {
+            "proto": "tcp",
+            "address": "0.0.0.0",
+            "port": 5432,
+            "interface": None,
+            "process": "postgres",
+            "pid": 812,
+            "unit": "postgresql@16-main",
+            "container_id": None,
+        }
+
+    @pytest.mark.parametrize(
+        "local, address, interface",
+        [
+            ("[::]:22", "::", None),
+            ("*:80", "*", None),
+            ("127.0.0.53%lo:53", "127.0.0.53", "lo"),
+            ("0.0.0.0%ens18:68", "0.0.0.0", "ens18"),
+            ("[::ffff:127.0.0.1]:8125", "::ffff:127.0.0.1", None),
+            ("[fe80::1%eth0]:546", "fe80::1", "eth0"),
+            (":::22", "::", None),  # iproute2 4.9 prints IPv6 without brackets
+        ],
+    )
+    def test_every_address_form(self, local: str, address: str, interface):
+        line = f"tcp LISTEN 0 128 {local} *:* users:((\"x\",pid=1,fd=3))\n"
+
+        [socket] = parse_listening_sockets(_wire(line, "1 0::/system.slice/x.service\n"))
+
+        assert (socket["address"], socket["interface"]) == (address, interface)
+
+    def test_the_first_of_several_processes_names_the_socket(self):
+        nginx = _by_port(parse_listening_sockets(_wire()))[("tcp", 80, "*")]
+
+        assert (nginx["process"], nginx["pid"], nginx["unit"]) == ("nginx", 901, "nginx")
+
+    def test_a_socket_without_a_process_is_kept(self):
+        """The kernel's nfsd, or every socket of another user without root."""
+        nfs = _by_port(parse_listening_sockets(_wire()))[("tcp", 2049, "0.0.0.0")]
+
+        assert (nfs["process"], nfs["pid"], nfs["unit"], nfs["container_id"]) == (
+            None,
+            None,
+            None,
+            None,
+        )
+
+    def test_the_header_is_no_socket(self):
+        sockets = parse_listening_sockets(_wire())
+
+        assert len(sockets) == 9
+        assert all(s["proto"] in ("tcp", "udp") for s in sockets)
+
+    def test_sorted_by_protocol_port_and_address(self):
+        keys = [(s["proto"], s["port"], s["address"]) for s in parse_listening_sockets(_wire())]
+
+        assert keys == sorted(keys)
+
+    def test_nothing_listening_is_an_empty_list(self):
+        assert parse_listening_sockets(_wire(SS.splitlines(True)[0], "")) == []
+
+
+class TestCgroups:
+    def _unit_and_container(self, cgroup_lines: str) -> tuple:
+        line = 'tcp LISTEN 0 128 0.0.0.0:1 0.0.0.0:* users:(("p",pid=5,fd=3))\n'
+        [socket] = parse_listening_sockets(_wire(line, cgroup_lines))
+        return socket["unit"], socket["container_id"]
+
+    def test_a_container_on_the_host_network(self):
+        assert self._unit_and_container(f"5 0::/system.slice/docker-{CID}.scope\n") == (
+            None,
+            CID,
+        )
+
+    def test_a_container_under_the_cgroupfs_driver(self):
+        assert self._unit_and_container(f"5 0::/docker/{CID}\n") == (None, CID)
+
+    def test_cgroup_v1_reads_the_systemd_hierarchy(self):
+        lines = "5 12:pids:/user.slice\n5 1:name=systemd:/system.slice/ssh.service\n"
+
+        assert self._unit_and_container(lines) == ("ssh", None)
+
+    def test_an_escaped_template_instance(self):
+        lines = "5 0::/system.slice/system-wg\\x2dquick.slice/wg-quick@wg0.service\n"
+
+        assert self._unit_and_container(lines) == ("wg-quick@wg0", None)
+
+    def test_a_procd_service_on_openwrt(self):
+        """procd puts every instance into /services/<name>/<instance>, a jailed
+        one too -- its PID is not the one procd reports, its cgroup is."""
+        assert self._unit_and_container("5 0::/services/dnsmasq/cfg01411c\n") == (
+            "dnsmasq",
+            None,
+        )
+
+    def test_a_login_session_is_no_unit(self):
+        assert self._unit_and_container("5 0::/user.slice/user-1000.slice/session-3.scope\n") == (
+            None,
+            None,
+        )
+
+    def test_a_process_gone_before_its_cgroup_was_read(self):
+        assert self._unit_and_container("") == (None, None)
+
+    def test_the_docker_daemons_own_proxy_stays_raw(self):
+        """docker-proxy runs in docker.service. Telling it apart from a real
+        listener is the consumer's business; the reading reports what is there."""
+        proxy = _by_port(parse_listening_sockets(_wire()))[("tcp", 8080, "0.0.0.0")]
+
+        assert (proxy["process"], proxy["unit"]) == ("docker-proxy", "docker")
+
+
+class TestFailures:
+    def test_whatever_surrounds_the_frame_is_ignored(self):
+        noisy = _wire(noise="$ sh -c '...'\nWelcome to Ubuntu\n")
+
+        assert len(parse_listening_sockets(noisy)) == 9
+
+    def test_output_without_the_frame_raises(self):
+        with pytest.raises(ValueError):
+            parse_listening_sockets("sudo: a password is required\n")
+
+    def test_a_report_cut_short_raises(self):
+        with pytest.raises(ValueError):
+            parse_listening_sockets(_wire()[:-len("SOCK_END\n")])
+
+    def test_a_host_without_ss_or_netstat_is_unavailable(self):
+        with pytest.raises(ListeningSocketsUnavailable):
+            parse_listening_sockets("SOCK_BEGIN\n[no-ss]\nSOCK_END\n")
+
+    def test_ss_failing_raises(self):
+        with pytest.raises(ValueError):
+            parse_listening_sockets(_wire("ss: invalid option -- 'p'\n", "", rc=1))
+
+    def test_unavailable_is_not_a_value_error(self):
+        """A caller retries a broken report without privilege; a host without
+        ss has nothing to retry."""
+        assert not issubclass(ListeningSocketsUnavailable, ValueError)
+
+
+# busybox netstat on OpenWrt 25.12, addresses replaced by documentation ones.
+BUSYBOX = """Active Internet connections (only servers)
+Proto Recv-Q Send-Q Local Address           Foreign Address         State       PID/Program name    
+tcp        0      0 0.0.0.0:22              0.0.0.0:*               LISTEN      1604/dropbear
+tcp        0      0 0.0.0.0:443             0.0.0.0:*               LISTEN      1969/uhttpd
+tcp        0      0 192.0.2.15:53           0.0.0.0:*               LISTEN      1495/dnsmasq
+tcp        0      0 127.0.0.1:53            0.0.0.0:*               LISTEN      1495/dnsmasq
+tcp        0      0 :::22                   :::*                    LISTEN      1604/dropbear
+tcp        0      0 fe80::1:53              :::*                    LISTEN      1495/dnsmasq
+tcp        0      0 ::1:53                  :::*                    LISTEN      1495/dnsmasq
+udp        0      0 192.0.2.15:53           0.0.0.0:*                           1495/dnsmasq
+udp        0      0 0.0.0.0:161             0.0.0.0:*                           3173/snmpd
+udp        0      0 0.0.0.0:5353            0.0.0.0:*                           -
+"""
+PROCD = """1495 0::/services/dnsmasq/cfg01411c
+1604 0::/services/dropbear/instance1
+1969 0::/services/uhttpd/instance1
+3173 0::/services/snmpd/instance1
+"""
+# net-tools netstat on an old Debian: tcp6/udp6, and a program name with a space.
+NET_TOOLS = """Active Internet connections (only servers)
+Proto Recv-Q Send-Q Local Address           Foreign Address         State       PID/Program name    
+tcp        0      0 0.0.0.0:22              0.0.0.0:*               LISTEN      700/sshd: /usr/sbin 
+tcp6       0      0 :::22                   :::*                    LISTEN      700/sshd: /usr/sbin 
+udp6       0      0 :::5353                 :::*                                -                   
+"""
+
+
+def _netstat(out: str = BUSYBOX, cgroups: str = PROCD, *, rc: int = 0) -> str:
+    return f"SOCK_BEGIN\n[netstat]\n{out}__SS_RC={rc}\n[cgroups]\n{cgroups}SOCK_END\n"
+
+
+class TestNetstat:
+    """A host without ss (#673 in netOrk): OpenWrt's busybox, old net-tools."""
+
+    def test_a_socket_comes_with_its_process_and_procd_service(self):
+        sockets = _by_port(parse_listening_sockets(_netstat()))
+
+        assert sockets[("tcp", 22, "0.0.0.0")] == {
+            "proto": "tcp",
+            "address": "0.0.0.0",
+            "port": 22,
+            "interface": None,
+            "process": "dropbear",
+            "pid": 1604,
+            "unit": "dropbear",
+            "container_id": None,
+        }
+
+    @pytest.mark.parametrize(
+        "proto, port, address",
+        [("tcp", 22, "::"), ("tcp", 53, "fe80::1"), ("tcp", 53, "::1"), ("tcp", 53, "192.0.2.15")],
+    )
+    def test_every_address_form(self, proto, port, address):
+        assert (proto, port, address) in _by_port(parse_listening_sockets(_netstat()))
+
+    def test_a_udp_socket_has_no_state_column(self):
+        snmpd = _by_port(parse_listening_sockets(_netstat()))[("udp", 161, "0.0.0.0")]
+
+        assert (snmpd["process"], snmpd["pid"], snmpd["unit"]) == ("snmpd", 3173, "snmpd")
+
+    def test_a_socket_without_a_process_is_kept(self):
+        mdns = _by_port(parse_listening_sockets(_netstat()))[("udp", 5353, "0.0.0.0")]
+
+        assert (mdns["process"], mdns["pid"], mdns["unit"]) == (None, None, None)
+
+    def test_the_headers_are_no_sockets(self):
+        assert len(parse_listening_sockets(_netstat())) == 10
+
+    def test_net_tools_names_ipv6_and_programs_its_own_way(self):
+        sockets = _by_port(parse_listening_sockets(_netstat(NET_TOOLS, "")))
+
+        assert sockets[("tcp", 22, "::")]["process"] == "sshd"
+        assert sockets[("tcp", 22, "::")]["pid"] == 700
+        assert ("udp", 5353, "::") in sockets
+
+    def test_netstat_failing_raises(self):
+        with pytest.raises(ValueError):
+            parse_listening_sockets(_netstat("netstat: invalid option -- 'p'\n", "", rc=1))
+
+
+class TestTheNetstatFallback:
+    """The command itself, on a host where ss is missing and netstat is not."""
+
+    @pytest.mark.skipif(
+        any(os.path.exists(f"{d}/ss") for d in ("/usr/sbin", "/sbin")),
+        reason="ss sits on the PATH the command always adds",
+    )
+    def test_it_reads_netstat_when_there_is_no_ss(self, tmp_path):
+        for tool in ("grep", "cut", "sort", "sed", "tr", "cat"):
+            (tmp_path / tool).symlink_to(shutil.which(tool))
+        stub = tmp_path / "netstat"
+        stub.write_text(f"#!/bin/sh\ncat <<'EOF'\n{BUSYBOX}EOF\n")
+        stub.chmod(0o755)
+
+        out = subprocess.run(
+            ["/bin/sh", "-c", LISTENING_SOCKETS_COMMAND],
+            capture_output=True,
+            text=True,
+            env={"PATH": str(tmp_path)},
+            timeout=30,
+        ).stdout
+
+        assert "[netstat]" in out
+        sockets = _by_port(parse_listening_sockets(out))
+        assert sockets[("tcp", 443, "0.0.0.0")]["process"] == "uhttpd"
+        assert len(sockets) == 10
+
+
+class TestTheCommand:
+    def test_the_frame_is_not_in_the_command_itself(self):
+        """An echoing transport prints the command back; the markers must only
+        appear once the command has run."""
+        assert "SOCK_BEGIN" not in LISTENING_SOCKETS_COMMAND
+        assert "SOCK_END" not in LISTENING_SOCKETS_COMMAND
+
+    def test_it_writes_and_changes_nothing(self):
+        for verb in ("sudo", " > ", ">>", "kill", "rm ", "systemctl "):
+            assert verb not in LISTENING_SOCKETS_COMMAND
+
+    def test_it_is_posix_sh(self):
+        result = subprocess.run(
+            ["sh", "-n", "-c", LISTENING_SOCKETS_COMMAND], capture_output=True, text=True
+        )
+        assert result.returncode == 0, result.stderr
+
+    @pytest.mark.skipif(shutil.which("ss") is None, reason="needs ss on this host")
+    def test_it_runs_and_parses_on_this_host(self):
+        out = subprocess.run(
+            ["sh", "-c", LISTENING_SOCKETS_COMMAND], capture_output=True, text=True, timeout=60
+        ).stdout
+
+        sockets = parse_listening_sockets(out)
+
+        assert all(s["proto"] in ("tcp", "udp") and 0 < s["port"] < 65536 for s in sockets)
+
+
+class _Driver(ListeningSocketsMixin):
+    def __init__(self, privileged: str, unprivileged: str = "") -> None:
+        self.answers = {True: privileged, False: unprivileged}
+        self.calls: list = []
+
+    def _run_listening_sockets_command(self, command: str, *, privileged: bool) -> str:
+        self.calls.append((command, privileged))
+        return self.answers[privileged]
+
+
+class TestTheTemplate:
+    def test_a_driver_supplies_only_the_transport(self):
+        driver = _Driver(_wire())
+
+        reading = driver.get_listening_sockets()
+
+        assert reading["attributed"] is True
+        assert len(reading["sockets"]) == 9
+        [(command, privileged)] = driver.calls
+        assert privileged is True
+
+    def test_the_command_reaches_root_as_one_shell(self):
+        """``sudo -n a; b`` runs only ``a`` as root: the whole script goes as
+        one ``sh -c`` argument."""
+        driver = _Driver(_wire())
+        driver.get_listening_sockets()
+
+        [(command, _privileged)] = driver.calls
+        assert command.startswith("sh -c '")
+        assert subprocess.run(["sh", "-n", "-c", command]).returncode == 0
+
+    def test_without_root_it_reads_what_it_can(self):
+        """Without sudo, ss names only the user's own processes: the sockets
+        are still worth having, marked as not attributed."""
+        driver = _Driver("sudo: a password is required\n", _wire(cgroups=""))
+
+        reading = driver.get_listening_sockets()
+
+        assert reading["attributed"] is False
+        assert len(reading["sockets"]) == 9
+        assert [p for _c, p in driver.calls] == [True, False]
+
+    def test_a_host_without_either_is_not_asked_twice(self):
+        driver = _Driver("SOCK_BEGIN\n[no-ss]\nSOCK_END\n")
+
+        with pytest.raises(ListeningSocketsUnavailable):
+            driver.get_listening_sockets()
+        assert len(driver.calls) == 1
+
+    def test_not_every_os_driver_has_it(self):
+        """A Windows host is an OSDriver too and has no ss: ``hasattr`` has to
+        stay a truthful answer, so the drivers that can mix this in themselves."""
+        assert not issubclass(OSDriver, ListeningSocketsMixin)
+        assert not hasattr(OSDriver, "get_listening_sockets")

tests/test_package_updates.py

@@ -0,0 +1,149 @@
+"""Pending updates: which package, from where, and whether it closes a security hole.
+
+A patch deadline ("security updates within 14 days") needs to know which pending
+update is a security update. apt says so in the suite a candidate comes from
+(``noble-security``, ``stable-security``); dnf says so in its update advisories.
+Reading that is the same for every driver whose host runs apt or dnf, so the
+parsers live here once (netOrk MVP 5, #556).
+
+Fixture lines are from real hosts (Ubuntu 24.04, Debian 13 / OMV, Proxmox VE 9).
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from napalm_device_types.package_updates import (
+    APT_UPGRADABLE_COMMAND,
+    nevra_name,
+    parse_apt_upgradable,
+    parse_dnf_security,
+)
+
+UBUNTU = """\
+docker-compose-plugin/noble 5.6.0-1~ubuntu.24.04~noble amd64 [upgradable from: 5.5.1-1~ubuntu.24.04~noble]
+openssl/noble-updates,noble-security 3.0.13-0ubuntu3.6 amd64 [upgradable from: 3.0.13-0ubuntu3.5]
+__APT_RC=0
+"""
+
+DEBIAN = """\
+linux-image-amd64/stable-backports 7.1.13-1~bpo13+1 amd64 [upgradable from: 7.1.8-1~bpo13+1]
+libssl3t64/stable-security 3.5.1-1+deb13u2 amd64 [upgradable from: 3.5.1-1+deb13u1]
+__APT_RC=0
+"""
+
+
+def _by_name(updates):
+    return {u["name"]: u for u in updates}
+
+
+class TestAptUpgradable:
+    def test_each_line_is_a_package_with_both_versions(self):
+        update = _by_name(parse_apt_upgradable(UBUNTU))["docker-compose-plugin"]
+
+        assert update["current_version"] == "5.5.1-1~ubuntu.24.04~noble"
+        assert update["new_version"] == "5.6.0-1~ubuntu.24.04~noble"
+
+    def test_the_suites_are_its_origin(self):
+        updates = _by_name(parse_apt_upgradable(UBUNTU))
+
+        assert updates["openssl"]["origin"] == "noble-updates,noble-security"
+        assert updates["docker-compose-plugin"]["origin"] == "noble"
+
+    def test_a_suite_apt_lists_twice_is_named_once(self):
+        line = (
+            "fonts-opensymbol/noble-updates,noble-updates,noble-security,noble-security "
+            "4:102.12+LibO24.2.7-0ubuntu0.24.04.7 all [upgradable from: 4:102.12+LibO24.2.7-0ubuntu0.24.04.6]\n"
+            "__APT_RC=0\n"
+        )
+
+        assert parse_apt_upgradable(line)[0]["origin"] == "noble-updates,noble-security"
+
+    @pytest.mark.parametrize(
+        ("output", "name", "security"),
+        [
+            (UBUNTU, "openssl", True),
+            (UBUNTU, "docker-compose-plugin", False),
+            (DEBIAN, "libssl3t64", True),
+            (DEBIAN, "linux-image-amd64", False),
+        ],
+    )
+    def test_a_security_suite_makes_it_a_security_update(self, output, name, security):
+        assert _by_name(parse_apt_upgradable(output))[name]["security"] is security
+
+    def test_a_line_the_terminal_wrapped_is_joined(self):
+        wrapped = (
+            "openssl/noble-updates,noble-security 3.0.13-0ubuntu3.6 amd64 [upgradable fro\n"
+            " m: 3.0.13-0ubuntu3.5]\n__APT_RC=0\n"
+        )
+
+        assert _by_name(parse_apt_upgradable(wrapped))["openssl"]["current_version"] == (
+            "3.0.13-0ubuntu3.5"
+        )
+
+    def test_one_package_for_several_architectures_is_one_entry(self):
+        multiarch = (
+            "libc6/noble-updates 2.39-0ubuntu8.5 amd64 [upgradable from: 2.39-0ubuntu8.4]\n"
+            "libc6/noble-updates,noble-security 2.39-0ubuntu8.5 i386 [upgradable from: 2.39-0ubuntu8.4]\n"
+            "__APT_RC=0\n"
+        )
+
+        updates = parse_apt_upgradable(multiarch)
+
+        assert [u["name"] for u in updates] == ["libc6"]
+        assert updates[0]["security"] is True
+
+    def test_noise_is_ignored(self):
+        noisy = "Listing... Done\nWARNING: apt does not have a stable CLI interface.\n" + UBUNTU
+
+        assert len(parse_apt_upgradable(noisy)) == 2
+
+    def test_nothing_pending_is_an_empty_list(self):
+        assert parse_apt_upgradable("__APT_RC=0\n") == []
+
+    def test_a_list_without_its_exit_status_raises(self):
+        """Cut short: a transport stopped reading early, so nothing can be concluded."""
+        with pytest.raises(ValueError):
+            parse_apt_upgradable(UBUNTU.replace("__APT_RC=0\n", ""))
+
+    def test_a_failed_apt_raises(self):
+        with pytest.raises(ValueError, match="100"):
+            parse_apt_upgradable("E: Could not get lock\n__APT_RC=100\n")
+
+    def test_terminal_codes_around_the_status_are_dropped(self):
+        """What a pseudo-terminal left on a Raspberry Pi OS host."""
+        raw = "Listing... 0%\n\x1b[?1h\x1b=\n" + UBUNTU.replace("__APT_RC=0", "\x1b>__APT_RC=0")
+
+        assert len(parse_apt_upgradable(raw)) == 2
+
+    def test_the_command_reads_without_root_or_a_terminal(self):
+        """Through a pipe apt draws no progress and no terminal codes; one of
+        those, ESC >, ended a screen-scraping read at a false prompt."""
+        assert "LC_ALL=C apt list --upgradable" in APT_UPGRADABLE_COMMAND
+        assert APT_UPGRADABLE_COMMAND.rstrip().endswith("| cat")
+        assert "sudo" not in APT_UPGRADABLE_COMMAND
+
+
+class TestDnfSecurity:
+    ADVISORIES = """\
+FEDORA-2024-1a2b3c4d5e Important/Sec. openssl-libs-1:3.1.4-2.fc40.x86_64
+FEDORA-2024-1a2b3c4d5e Important/Sec. openssl-1:3.1.4-2.fc40.x86_64
+RLSA-2024:1234         Moderate/Sec.  kernel-core-5.14.0-427.13.1.el9_4.x86_64
+"""
+
+    def test_the_names_of_packages_with_a_security_advisory(self):
+        assert parse_dnf_security(self.ADVISORIES) == {"openssl-libs", "openssl", "kernel-core"}
+
+    def test_nothing_is_an_empty_set(self):
+        assert parse_dnf_security("") == set()
+
+    @pytest.mark.parametrize(
+        ("nevra", "name"),
+        [
+            ("openssl-libs-1:3.1.4-2.fc40.x86_64", "openssl-libs"),
+            ("kernel-core-5.14.0-427.13.1.el9_4.x86_64", "kernel-core"),
+            ("python3-dnf-4.14.0-9.el9.noarch", "python3-dnf"),
+        ],
+    )
+    def test_the_name_of_a_nevra(self, nevra, name):
+        assert nevra_name(nevra) == name

tests/test_provisioning.py

@@ -0,0 +1,102 @@
+"""What provisioning a VM from a cloud image needs to know about the guest,
+the same for every hypervisor (NetOrk/netork#794)."""
+
+from __future__ import annotations
+
+import pytest
+from napalm_device_types import HypervisorDriver
+from napalm_device_types.provisioning import (
+    QEMU_GUEST_AGENTS,
+    network_config,
+    split_compression,
+)
+
+
+class TestGuestAgentDeclaration:
+    """netOrk's cloud-init installs the agent through which the hypervisor
+    reads the new VM's IP. Which agent depends on the hypervisor and on the
+    guest; the guests a driver lists are the ones it can provision."""
+
+    def test_the_base_provisions_linux_with_qemu_guest_agent(self):
+        assert HypervisorDriver.GUEST_AGENTS == {
+            "linux": (("qemu-guest-agent",), ("systemctl enable --now qemu-guest-agent",)),
+        }
+
+    def test_the_declaration_is_data_netork_reads_off_the_class(self):
+        assert not callable(vars(HypervisorDriver)["GUEST_AGENTS"])
+
+    def test_the_per_hypervisor_attributes_are_gone(self):
+        assert not hasattr(HypervisorDriver, "GUEST_AGENT_PACKAGES")
+        assert not hasattr(HypervisorDriver, "GUEST_AGENT_RUNCMD")
+
+
+class TestQemuGuestAgents:
+    """Verified on FreeBSD 15.1 and OpenBSD 7.9 cloud images (#793)."""
+
+    def test_linux_is_the_base_default(self):
+        assert QEMU_GUEST_AGENTS["linux"] == HypervisorDriver.GUEST_AGENTS["linux"]
+
+    def test_freebsd(self):
+        assert QEMU_GUEST_AGENTS["freebsd"] == (
+            ("qemu-guest-agent",),
+            ("sysrc qemu_guest_agent_enable=YES", "service qemu-guest-agent start"),
+        )
+
+    def test_openbsd(self):
+        assert QEMU_GUEST_AGENTS["openbsd"] == (
+            ("qemu-ga",),
+            ("rcctl enable qemu_ga", "rcctl start qemu_ga"),
+        )
+
+    @pytest.mark.parametrize("guest_os", ["freebsd", "openbsd"])
+    def test_no_bsd_guest_is_told_to_use_systemd(self, guest_os):
+        _, runcmd = QEMU_GUEST_AGENTS[guest_os]
+        assert not any("systemctl" in command for command in runcmd)
+
+
+class TestNetworkConfig:
+    """cloud-init's network-config v2. FreeBSD's nuageinit reads only this
+    version; the v1 Proxmox generates makes it skip runcmd (#793)."""
+
+    def test_dhcp_nics_matched_by_mac(self):
+        cfg = network_config([("00:50:56:aa:bb:cc", True), ("00:50:56:aa:bb:dd", False)])
+        assert cfg == {
+            "version": 2,
+            "ethernets": {
+                "nic0": {"match": {"macaddress": "00:50:56:aa:bb:cc"}, "dhcp4": True},
+            },
+        }
+
+    def test_no_dhcp_nic_means_no_network_config(self):
+        assert network_config([("00:50:56:aa:bb:cc", False)]) is None
+
+    def test_macs_are_written_in_lower_case(self):
+        """Proxmox reports them upper-case; nuageinit compares them as given."""
+        cfg = network_config([("BC:24:11:AA:BB:02", True)])
+        assert cfg["ethernets"]["nic0"]["match"]["macaddress"] == "bc:24:11:aa:bb:02"
+
+
+class TestSplitCompression:
+    """Official FreeBSD images come packed; hypervisors import unpacked ones."""
+
+    def test_xz(self):
+        assert split_compression("FreeBSD-15.1-RELEASE-amd64-BASIC-CLOUDINIT-ufs.qcow2.xz") == (
+            "FreeBSD-15.1-RELEASE-amd64-BASIC-CLOUDINIT-ufs.qcow2",
+            "xz -dc",
+        )
+
+    @pytest.mark.parametrize(
+        ("name", "command"),
+        [("a.raw.gz", "gzip -dc"), ("a.raw.bz2", "bzip2 -dc"), ("a.qcow2.zst", "zstd -dc")],
+    )
+    def test_other_packers(self, name, command):
+        assert split_compression(name) == (name.rsplit(".", 1)[0], command)
+
+    @pytest.mark.parametrize(
+        "name", ["debian-13-genericcloud-amd64.qcow2", "ubuntu-24.04-server-cloudimg-amd64.img"]
+    )
+    def test_an_unpacked_image_is_left_as_it_is(self, name):
+        assert split_compression(name) == (name, None)
+
+    def test_the_suffix_is_matched_case_insensitively(self):
+        assert split_compression("IMAGE.QCOW2.XZ") == ("IMAGE.QCOW2", "xz -dc")

tests/test_ssh_server.py

@@ -0,0 +1,303 @@
+"""OpenSSH server hardening: removing algorithms a scanner flagged, without locking anyone out.
+
+The plan is pure and tested on its own. The mixin is tested against a fake host
+that keeps files and answers ``sshd -T`` from them, so every path -- the
+drop-in, the sshd_config fallback, a refused config, a reload the new session
+cannot get through -- runs without a machine.
+"""
+
+from __future__ import annotations
+
+import re
+from typing import Dict, List, Optional
+
+import pytest
+
+from napalm_device_types.channel import CommandResult
+from napalm_device_types.ssh_server import (
+    DROPIN_PATH,
+    MAIN_CONFIG,
+    OpenSshServerMixin,
+    has_dropin_include,
+    parse_sshd_effective,
+    plan_ssh_hardening,
+    render_dropin,
+    render_main_config,
+)
+
+SSHD_T = """\
+port 22
+kexalgorithms sntrup761x25519-sha512@openssh.com,curve25519-sha256,curve25519-sha256@libssh.org,ecdh-sha2-nistp256,ecdh-sha2-nistp384,diffie-hellman-group14-sha256
+hostkeyalgorithms ssh-ed25519,ecdsa-sha2-nistp256,rsa-sha2-512,rsa-sha2-256
+ciphers chacha20-poly1305@openssh.com,aes128-ctr,aes256-gcm@openssh.com
+macs umac-64-etm@openssh.com,hmac-sha2-256-etm@openssh.com,hmac-sha1,hmac-sha2-256
+passwordauthentication no
+"""
+
+PARAMIKO = {
+    "kex": ["curve25519-sha256@libssh.org", "ecdh-sha2-nistp256", "diffie-hellman-group14-sha256"],
+    "hostkey": ["ssh-ed25519", "ecdsa-sha2-nistp256", "rsa-sha2-512"],
+    "cipher": ["aes128-ctr", "aes256-gcm@openssh.com"],
+    "mac": ["hmac-sha2-256", "hmac-sha2-256-etm@openssh.com", "hmac-sha1"],
+}
+
+
+# ── parse ────────────────────────────────────────────────────────────────
+
+
+def test_parse_reads_the_four_lists_in_server_order():
+    eff = parse_sshd_effective(SSHD_T)
+    assert eff["kex"][:2] == ["sntrup761x25519-sha512@openssh.com", "curve25519-sha256"]
+    assert eff["hostkey"] == ["ssh-ed25519", "ecdsa-sha2-nistp256", "rsa-sha2-512", "rsa-sha2-256"]
+    assert eff["cipher"] == ["chacha20-poly1305@openssh.com", "aes128-ctr", "aes256-gcm@openssh.com"]
+    assert eff["mac"][-1] == "hmac-sha2-256"
+
+
+def test_parse_refuses_output_without_the_lists():
+    with pytest.raises(ValueError):
+        parse_sshd_effective("port 22\n")
+
+
+# ── plan ─────────────────────────────────────────────────────────────────
+
+
+def test_plan_removes_what_was_asked_and_keeps_the_order():
+    plan = plan_ssh_hardening(parse_sshd_effective(SSHD_T), {"mac": ["hmac-sha1", "umac-64-etm@openssh.com"]})
+    assert plan.remove == {"mac": ["umac-64-etm@openssh.com", "hmac-sha1"]}
+    assert plan.keep["mac"] == ["hmac-sha2-256-etm@openssh.com", "hmac-sha2-256"]
+    assert plan.changed_classes == ["mac"]
+    assert plan.refused == []
+
+
+def test_plan_ignores_an_algorithm_the_server_no_longer_offers():
+    plan = plan_ssh_hardening(parse_sshd_effective(SSHD_T), {"cipher": ["3des-cbc"]})
+    assert plan.changed_classes == []
+    assert [r.algorithm for r in plan.not_offered] == ["3des-cbc"]
+
+
+def test_plan_never_removes_the_last_algorithm_of_a_class():
+    plan = plan_ssh_hardening(
+        parse_sshd_effective(SSHD_T),
+        {"cipher": ["chacha20-poly1305@openssh.com", "aes128-ctr", "aes256-gcm@openssh.com"]},
+    )
+    # The first in the server's order stays, the rest go.
+    assert plan.keep["cipher"] == ["chacha20-poly1305@openssh.com"]
+    assert plan.remove["cipher"] == ["aes128-ctr", "aes256-gcm@openssh.com"]
+    assert [(r.kind, r.algorithm, r.reason) for r in plan.refused] == [
+        ("cipher", "chacha20-poly1305@openssh.com", "last")
+    ]
+
+
+def test_plan_keeps_one_algorithm_the_managing_client_still_speaks():
+    """paramiko knows no sntrup or plain curve25519: removing everything it does
+    know would leave the server reachable, but not by netOrk."""
+    plan = plan_ssh_hardening(
+        parse_sshd_effective(SSHD_T),
+        {"kex": ["curve25519-sha256@libssh.org", "ecdh-sha2-nistp256", "ecdh-sha2-nistp384",
+                 "diffie-hellman-group14-sha256"]},
+        client=PARAMIKO,
+    )
+    assert "curve25519-sha256@libssh.org" in plan.keep["kex"]
+    assert [(r.algorithm, r.reason) for r in plan.refused] == [("curve25519-sha256@libssh.org", "client")]
+    assert plan.remove["kex"] == ["ecdh-sha2-nistp256", "ecdh-sha2-nistp384", "diffie-hellman-group14-sha256"]
+
+
+def test_plan_without_a_client_list_only_guards_the_last_algorithm():
+    plan = plan_ssh_hardening(
+        parse_sshd_effective(SSHD_T),
+        {"kex": ["curve25519-sha256@libssh.org", "ecdh-sha2-nistp256", "ecdh-sha2-nistp384",
+                 "diffie-hellman-group14-sha256"]},
+    )
+    assert plan.refused == []
+    assert plan.keep["kex"] == ["sntrup761x25519-sha512@openssh.com", "curve25519-sha256"]
+
+
+def test_plan_rejects_an_unknown_class():
+    with pytest.raises(ValueError):
+        plan_ssh_hardening(parse_sshd_effective(SSHD_T), {"compression": ["zlib"]})
+
+
+# ── render ───────────────────────────────────────────────────────────────
+
+
+def test_dropin_states_full_lists_for_the_changed_classes_only():
+    plan = plan_ssh_hardening(parse_sshd_effective(SSHD_T), {"mac": ["hmac-sha1"]})
+    text = render_dropin(plan.keep, ["mac"])
+    assert "MACs umac-64-etm@openssh.com,hmac-sha2-256-etm@openssh.com,hmac-sha2-256\n" in text
+    assert "KexAlgorithms" not in text
+    assert text.startswith("#")  # says who wrote it
+
+
+def test_include_is_found_only_in_the_global_section():
+    assert has_dropin_include("Include /etc/ssh/sshd_config.d/*.conf\nPort 22\n")
+    assert has_dropin_include("  include   /etc/ssh/sshd_config.d/*.conf\n")
+    assert not has_dropin_include("Port 22\n")
+    assert not has_dropin_include("Port 22\nMatch User x\n  Include /etc/ssh/sshd_config.d/*.conf\n")
+
+
+def test_main_config_gets_the_lists_first_and_the_old_lines_disabled():
+    main = "# defaults\nPort 22\nMACs hmac-sha1,hmac-sha2-256\nMatch User backup\n  MACs hmac-sha1\n"
+    text = render_main_config(main, {"mac": ["hmac-sha2-256"]}, ["mac"])
+    lines = text.splitlines()
+    first_directive = next(line for line in lines if line and not line.startswith("#"))
+    assert first_directive == "MACs hmac-sha2-256"
+    assert "# netork: MACs hmac-sha1,hmac-sha2-256" in lines
+    # A Match block is left alone: it is not the global setting.
+    assert "  MACs hmac-sha1" in lines
+
+
+def test_main_config_rewrite_is_idempotent():
+    main = "Port 22\nMACs hmac-sha1,hmac-sha2-256\n"
+    once = render_main_config(main, {"mac": ["hmac-sha2-256"]}, ["mac"])
+    twice = render_main_config(once, {"mac": ["hmac-sha2-256"]}, ["mac"])
+    assert once == twice
+
+
+# ── the mixin, against a fake host ───────────────────────────────────────
+
+_DIRECTIVE = {"kexalgorithms": "kex", "hostkeyalgorithms": "hostkey", "ciphers": "cipher", "macs": "mac"}
+
+
+class FakeHost(OpenSshServerMixin):
+    """A host with files, an sshd that reads them first-match-wins, and a log of what ran."""
+
+    def __init__(self, main: str = "Include /etc/ssh/sshd_config.d/*.conf\nPort 22\n") -> None:
+        self.files: Dict[str, str] = {MAIN_CONFIG: main}
+        self.defaults = parse_sshd_effective(SSHD_T)
+        self.ran: List[str] = []
+        self.syntax_ok = True
+        self.reloads = 0
+
+    def _effective(self) -> Dict[str, List[str]]:
+        eff = {k: list(v) for k, v in self.defaults.items()}
+        seen = set()
+        lines: List[str] = []
+        for line in self.files[MAIN_CONFIG].splitlines():
+            if re.match(r"\s*include\s", line, re.I):
+                lines += self.files.get(DROPIN_PATH, "").splitlines()
+            elif re.match(r"\s*match\s", line, re.I):
+                break
+            else:
+                lines.append(line)
+        for line in lines:
+            parts = line.split()
+            if len(parts) == 2 and parts[0].lower() in _DIRECTIVE:
+                kind = _DIRECTIVE[parts[0].lower()]
+                if kind not in seen:
+                    seen.add(kind)
+                    eff[kind] = parts[1].split(",")
+        return eff
+
+    def run_command(self, command: str, *, privileged: bool = False, timeout: float = 60,
+                    stdin: Optional[bytes] = None) -> CommandResult:
+        self.ran.append(command)
+        assert privileged, command
+        if command.startswith("sshd -T") or "sshd -T" in command:
+            eff = self._effective()
+            out = "".join(f"{d} {','.join(eff[k])}\n" for d, k in _DIRECTIVE.items())
+            return CommandResult(out, "", 0)
+        if "sshd -t" in command:
+            return CommandResult("", "" if self.syntax_ok else "bad line 3", 0 if self.syntax_ok else 255)
+        m = re.match(r"cat -- '?([^' ]+)'?$", command)
+        if m:
+            path = m.group(1)
+            if path in self.files:
+                return CommandResult(self.files[path], "", 0)
+            return CommandResult("", "No such file", 1)
+        m = re.search(r"mv -f -- \"\$tmp\" '?([^' ]+)'?", command)
+        if m:
+            self.files[m.group(1)] = (stdin or b"").decode()
+            return CommandResult("", "", 0)
+        m = re.match(r"rm -f -- '?([^' ]+)'?$", command)
+        if m:
+            self.files.pop(m.group(1), None)
+            return CommandResult("", "", 0)
+        if "reload" in command or "HUP" in command:
+            self.reloads += 1
+            return CommandResult("", "", 0)
+        if command.startswith("mkdir"):
+            return CommandResult("", "", 0)
+        raise AssertionError(f"unexpected command: {command}")
+
+
+def test_mixin_writes_the_dropin_checks_it_and_reloads():
+    host = FakeHost()
+    lines: List[str] = []
+    result = host.harden_ssh_server({"mac": ["hmac-sha1"]}, client=PARAMIKO, log=lines.append)
+    assert result.applied and not result.rolled_back
+    assert result.path == DROPIN_PATH
+    assert "hmac-sha1" not in host._effective()["mac"]
+    assert host.reloads == 1
+    assert any("sshd -t" in c for c in host.ran)
+    assert lines  # it says what it does
+
+
+def test_mixin_keeps_what_an_earlier_fix_wrote():
+    host = FakeHost()
+    host.harden_ssh_server({"hostkey": ["ecdsa-sha2-nistp256"]})
+    host.harden_ssh_server({"mac": ["hmac-sha1"]})
+    eff = host._effective()
+    assert "ecdsa-sha2-nistp256" not in eff["hostkey"]
+    assert "hmac-sha1" not in eff["mac"]
+
+
+def test_mixin_falls_back_to_sshd_config_without_an_include():
+    host = FakeHost(main="Port 22\nMACs hmac-sha1,hmac-sha2-256\n")
+    result = host.harden_ssh_server({"mac": ["hmac-sha1"]})
+    assert result.path == MAIN_CONFIG
+    assert result.backup and host.files[result.backup] == "Port 22\nMACs hmac-sha1,hmac-sha2-256\n"
+    assert host._effective()["mac"] == ["hmac-sha2-256"]
+
+
+def test_mixin_restores_when_sshd_rejects_the_config():
+    host = FakeHost()
+    host.syntax_ok = False
+    result = host.harden_ssh_server({"mac": ["hmac-sha1"]})
+    assert not result.applied and result.rolled_back
+    assert DROPIN_PATH not in host.files
+    assert host.reloads == 0
+    assert "bad line 3" in result.message
+
+
+def test_mixin_restores_when_the_effective_lists_do_not_change():
+    """Another file read before ours sets the list: ours is valid and ignored."""
+    host = FakeHost(main="MACs hmac-sha1,hmac-sha2-256\nInclude /etc/ssh/sshd_config.d/*.conf\n")
+    result = host.harden_ssh_server({"mac": ["hmac-sha1"]})
+    assert not result.applied and result.rolled_back
+    assert DROPIN_PATH not in host.files
+    assert "MACs" in result.message
+
+
+def test_mixin_rolls_back_when_a_new_session_cannot_get_in():
+    host = FakeHost()
+    host.files[DROPIN_PATH] = "# netOrk\nHostKeyAlgorithms ssh-ed25519\n"
+    before = host.files[DROPIN_PATH]
+    result = host.harden_ssh_server({"mac": ["hmac-sha1"]}, verify=lambda: False)
+    assert not result.applied and result.rolled_back
+    assert host.files[DROPIN_PATH] == before
+    assert host.reloads == 2  # the change, and the way back
+
+
+def test_mixin_does_nothing_when_nothing_would_change():
+    host = FakeHost()
+    result = host.harden_ssh_server({"cipher": ["3des-cbc"]})
+    assert not result.applied and not result.rolled_back
+    assert host.reloads == 0
+    assert DROPIN_PATH not in host.files
+
+
+def test_declared_on_no_base_class():
+    """hasattr is the capability probe: only a driver that mixes it in answers True."""
+    from napalm_device_types import DeviceTypeDriver, OSDriver
+
+    assert not hasattr(DeviceTypeDriver, "harden_ssh_server")
+    assert not hasattr(OSDriver, "harden_ssh_server")
+
+
+def test_main_config_second_fix_keeps_what_the_first_set():
+    host = FakeHost(main="Port 22\n")
+    host.harden_ssh_server({"hostkey": ["ecdsa-sha2-nistp256"]})
+    host.harden_ssh_server({"mac": ["hmac-sha1"]})
+    eff = host._effective()
+    assert "ecdsa-sha2-nistp256" not in eff["hostkey"]
+    assert "hmac-sha1" not in eff["mac"]

tests/test_virtual_ips.py

@@ -0,0 +1,617 @@
+"""get_virtual_ips: the addresses a host's HA configuration lets float between machines.
+
+A virtual IP is not visible from the address list alone: vip-manager and
+keepalived put it on whichever node is master *right now*, and a standby has no
+trace of it in ``ip addr``. What every member has is the configuration that
+declares it. Reading that is the same on every Linux host, so the command and
+its parse live here once, and a driver only carries the command across
+(netOrk #829).
+
+The configuration also holds secrets -- vip-manager's etcd password, keepalived's
+``auth_pass`` -- so the command filters on the host: only allowlisted fields
+ever leave it. The tests below run the command for real to hold it to that.
+"""
+
+from __future__ import annotations
+
+import os
+import shutil
+import subprocess
+from pathlib import Path
+
+import pytest
+
+from napalm_device_types import OSDriver
+from napalm_device_types.virtual_ips import (
+    VIRTUAL_IPS_COMMAND,
+    VirtualIpsMixin,
+    parse_virtual_ips,
+)
+
+SECRET = "s3cr3t-do-not-leak"
+
+
+def _wire(*sections: str, noise: str = "") -> str:
+    """The report as the command prints it, framed."""
+    return f"{noise}VIPS_BEGIN\n{''.join(sections)}VIPS_END\n"
+
+
+def _section(mechanism: str, unit: str, *records: str) -> str:
+    return f"[{mechanism} {unit}]\n" + "".join(f"{r}\n" for r in records)
+
+
+def _vip_manager(*records: str, unit: str = "vip-manager.service") -> str:
+    return _section("vip-manager", unit, *records)
+
+
+def _keepalived(*records: str, unit: str = "keepalived.service") -> str:
+    return _section("keepalived", unit, *records)
+
+
+DB01 = _vip_manager(
+    "active active",
+    "arg config /etc/default/vip-manager.yml",
+    "cfg ip 10.7.224.10",
+    "cfg netmask 24",
+    "cfg interface ens7",
+    'cfg trigger-key "/service/netork-db/leader"',
+)
+
+
+class TestVipManager:
+    def test_the_5x_yaml_declares_the_address(self):
+        reading = parse_virtual_ips(_wire(DB01))
+
+        assert reading == {
+            "vip_manager": [
+                {
+                    "mechanism": "vip-manager",
+                    "instance": "vip-manager",
+                    "address": "10.7.224.10",
+                    "prefix_length": 24,
+                    "interface": "ens7",
+                    "group_key": "/service/netork-db/leader",
+                    "running": True,
+                }
+            ],
+            "keepalived": [],
+        }
+
+    def test_the_1x_unit_takes_its_values_from_the_environment_file(self):
+        """Ubuntu's 1.0.2 package: ``-ip=${VIP_IP}`` on the command line,
+        the values in /etc/default/vip-manager."""
+        section = _vip_manager(
+            "active active",
+            "arg ip ${VIP_IP}",
+            "arg mask $VIP_MASK",
+            "arg iface ${VIP_IFACE}",
+            "arg key ${VIP_KEY}",
+            "env VIP_IP 10.0.0.10",
+            "env VIP_MASK 255.255.255.0",
+            "env VIP_IFACE eth0",
+            'env VIP_KEY "/service/pg/leader"',
+        )
+
+        [entry] = parse_virtual_ips(_wire(section))["vip_manager"]
+
+        assert (entry["address"], entry["prefix_length"], entry["interface"]) == (
+            "10.0.0.10",
+            24,
+            "eth0",
+        )
+        assert entry["group_key"] == "/service/pg/leader"
+
+    def test_a_flag_beats_the_environment_which_beats_the_config_file(self):
+        section = _vip_manager(
+            "active active",
+            "arg ip 10.0.0.1",
+            "env VIP_IP 10.0.0.2",
+            "env VIP_NETMASK 16",
+            "cfg ip 10.0.0.3",
+            "cfg netmask 24",
+            "cfg interface eth1",
+        )
+
+        [entry] = parse_virtual_ips(_wire(section))["vip_manager"]
+
+        assert (entry["address"], entry["prefix_length"], entry["interface"]) == (
+            "10.0.0.1",
+            16,
+            "eth1",
+        )
+
+    @pytest.mark.parametrize(
+        "value, expected",
+        [
+            ("10.0.0.10 # the database", "10.0.0.10"),
+            ("'10.0.0.10'", "10.0.0.10"),
+            ('"fd00::10"', "fd00::10"),
+            ("FD00:0:0::10", "fd00::10"),
+        ],
+    )
+    def test_values_lose_quotes_and_comments(self, value, expected):
+        section = _vip_manager("active active", f"cfg ip {value}")
+
+        [entry] = parse_virtual_ips(_wire(section))["vip_manager"]
+
+        assert entry["address"] == expected
+
+    def test_a_stopped_unit_still_declares_its_address(self):
+        """A standby whose vip-manager is down is still a member -- one that
+        cannot take the address over, which is worth knowing."""
+        section = _vip_manager("active inactive", "cfg ip 10.0.0.10")
+
+        [entry] = parse_virtual_ips(_wire(section))["vip_manager"]
+
+        assert entry["running"] is False
+        assert entry["prefix_length"] is None
+
+    def test_a_unit_without_an_address_declares_nothing(self):
+        section = _vip_manager("active active", "arg ip ${VIP_IP}", "cfg netmask 24")
+
+        assert parse_virtual_ips(_wire(section))["vip_manager"] == []
+
+    def test_a_template_unit_is_named_after_its_instance(self):
+        section = _vip_manager(
+            "active active", "cfg ip 10.0.0.10", unit="vip-manager@pg16.service"
+        )
+
+        [entry] = parse_virtual_ips(_wire(section))["vip_manager"]
+
+        assert entry["instance"] == "vip-manager@pg16"
+
+    def test_an_unreadable_config_is_not_looked_at(self):
+        """None, not []: the reading must not say the address is gone."""
+        section = _vip_manager("active active", "unreadable /etc/default/vip-manager.yml")
+
+        reading = parse_virtual_ips(_wire(section, _keepalived("active inactive")))
+
+        assert reading["vip_manager"] is None
+        assert reading["keepalived"] == []
+
+
+class TestKeepalived:
+    TWO_INSTANCES = _keepalived(
+        "active active",
+        "instance VI_DB",
+        "interface eth0",
+        "virtual_router_id 51",
+        "vip 10.0.0.10/24 dev eth0",
+        "vip 10.0.0.11",
+        "end",
+        "instance web",
+        "vip fd00::80",
+        "interface eth1",
+        "end",
+    )
+
+    def test_each_address_of_each_instance(self):
+        reading = parse_virtual_ips(_wire(self.TWO_INSTANCES))
+
+        assert reading["vip_manager"] == []
+        assert reading["keepalived"] == [
+            {
+                "mechanism": "keepalived",
+                "instance": "VI_DB",
+                "address": "10.0.0.10",
+                "prefix_length": 24,
+                "interface": "eth0",
+                "group_key": "vrid:51",
+                "running": True,
+            },
+            {
+                "mechanism": "keepalived",
+                "instance": "VI_DB",
+                "address": "10.0.0.11",
+                "prefix_length": 32,
+                "interface": "eth0",
+                "group_key": "vrid:51",
+                "running": True,
+            },
+            {
+                "mechanism": "keepalived",
+                "instance": "web",
+                "address": "fd00::80",
+                "prefix_length": 128,
+                "interface": "eth1",
+                "group_key": None,
+                "running": True,
+            },
+        ]
+
+    def test_dev_beats_the_instance_interface(self):
+        section = _keepalived(
+            "active active", "instance a", "interface eth0", "vip 10.0.0.10/24 dev eth9", "end"
+        )
+
+        [entry] = parse_virtual_ips(_wire(section))["keepalived"]
+
+        assert entry["interface"] == "eth9"
+
+    @pytest.mark.parametrize("line", ["vip $VIP", "vip @node1", "vip not-an-address"])
+    def test_what_is_no_address_is_skipped(self, line):
+        """keepalived's ``$VAR`` substitution and ``@host`` conditionals are
+        not resolved here; an entry that is no address is no entry."""
+        section = _keepalived("active active", "instance a", line, "vip 10.0.0.12", "end")
+
+        assert [e["address"] for e in parse_virtual_ips(_wire(section))["keepalived"]] == [
+            "10.0.0.12"
+        ]
+
+    def test_an_instance_cut_off_by_the_end_of_the_file_still_counts(self):
+        section = _keepalived("active active", "instance a", "vip 10.0.0.10")
+
+        assert len(parse_virtual_ips(_wire(section))["keepalived"]) == 1
+
+    def test_without_a_config_file_it_declares_nothing(self):
+        section = _keepalived("active inactive", "missing /etc/keepalived/keepalived.conf")
+
+        assert parse_virtual_ips(_wire(section))["keepalived"] == []
+
+    @pytest.mark.parametrize(
+        "record", ["unreadable /etc/keepalived/conf.d/x.conf", "too-deep /etc/keepalived/a.conf"]
+    )
+    def test_an_include_it_could_not_follow_is_not_looked_at(self, record):
+        section = _keepalived("active active", "instance a", "vip 10.0.0.10", "end", record)
+
+        assert parse_virtual_ips(_wire(section))["keepalived"] is None
+
+
+class TestTheReport:
+    def test_no_units_is_looked_and_found_nothing(self):
+        assert parse_virtual_ips(_wire("[systemd]\n")) == {"vip_manager": [], "keepalived": []}
+
+    def test_a_host_without_systemd_is_not_looked_at(self):
+        assert parse_virtual_ips(_wire("[no-systemd]\n")) == {
+            "vip_manager": None,
+            "keepalived": None,
+        }
+
+    def test_noise_before_the_report_is_ignored(self):
+        reading = parse_virtual_ips(_wire(DB01, noise="$ sh -c '...'\nWelcome to Ubuntu\n"))
+
+        assert reading["vip_manager"][0]["address"] == "10.7.224.10"
+
+    def test_a_report_cut_short_raises_without_quoting_it(self):
+        """The output may carry configuration; an error message must not."""
+        with pytest.raises(ValueError) as caught:
+            parse_virtual_ips(f"VIPS_BEGIN\n{DB01}")
+
+        assert "10.7.224.10" not in str(caught.value)
+
+    def test_entries_are_sorted(self):
+        reading = parse_virtual_ips(
+            _wire(
+                _vip_manager("active active", "cfg ip 10.0.0.20", unit="vip-manager@b.service"),
+                _vip_manager("active active", "cfg ip 10.0.0.3", unit="vip-manager@a.service"),
+            )
+        )
+
+        assert [e["address"] for e in reading["vip_manager"]] == ["10.0.0.3", "10.0.0.20"]
+
+
+# ---------------------------------------------------------------------------
+# The command itself, run against a stub systemctl and real files
+# ---------------------------------------------------------------------------
+
+DB01_YAML = f"""# managed by hand, see infrastructure/docs/DATABASE_CLUSTER.md
+ip: 10.7.224.10
+netmask: 24
+interface: ens7
+trigger-key: "/service/netork-db/leader"
+trigger-value: "db-01"
+dcs-type: etcd
+dcs-endpoints:
+  - http://10.7.234.12:2379
+etcd-user: patroni
+etcd-password: {SECRET}
+"""
+
+KEEPALIVED_CONF = f"""! Configuration File for keepalived
+global_defs {{
+   notification_email {{ ops@example.org }}
+   router_id LVS_{SECRET}
+}}
+
+vrrp_script chk_haproxy {{
+    script "/usr/local/bin/check {SECRET}"
+}}
+
+vrrp_instance VI_WEB {{
+    state MASTER
+    interface eth0
+    virtual_router_id 51
+    priority 101
+    authentication {{
+        auth_type PASS
+        auth_pass {SECRET}
+    }}
+    virtual_ipaddress {{
+        192.0.2.10/24 dev eth0 label eth0:1
+    }}
+    notify_master "/etc/keepalived/master.sh {SECRET}"
+}}
+
+include conf.d/*.conf
+"""
+
+INCLUDED_CONF = """vrrp_instance VI_DB { interface eth1
+  virtual_router_id 52
+  virtual_ipaddress { 192.0.2.20 }
+  virtual_ipaddress_excluded {
+    2001:db8::20/64
+  }
+}
+"""
+
+
+def _awks() -> list:
+    found = []
+    if shutil.which("mawk"):
+        found.append(pytest.param(["mawk"], id="mawk"))
+    if shutil.which("busybox"):
+        found.append(pytest.param(["busybox", "awk"], id="busybox"))
+    if shutil.which("gawk"):
+        found.append(pytest.param(["gawk"], id="gawk"))
+    return found or [pytest.param(None, marks=pytest.mark.skip(reason="no awk to run"))]
+
+
+class _Host:
+    """A host as far as the command can tell: units, their properties, files."""
+
+    def __init__(self, root: Path, awk: list) -> None:
+        self.root = root
+        self.bin = root / "bin"
+        self.bin.mkdir()
+        self.state = root / "systemd"
+        self.state.mkdir()
+        (self.state / "units").write_text("")
+        stub = self.bin / "systemctl"
+        stub.write_text(
+            "#!/bin/sh\n"
+            f"d={self.state}\n"
+            'case "$1" in\n'
+            f'  list-units) {shutil.which("cat")} "$d/units" ;;\n'
+            f'  show) for last; do :; done; {shutil.which("cat")} "$d/show-$last" ;;\n'
+            "esac\n"
+        )
+        stub.chmod(0o755)
+        wrapper = self.bin / "awk"
+        program = " ".join([shutil.which(awk[0]) or awk[0], *awk[1:]])
+        wrapper.write_text(f'#!/bin/sh\nexec {program} "$@"\n')
+        wrapper.chmod(0o755)
+
+    def unit(self, name: str, *properties: str) -> None:
+        with (self.state / "units").open("a") as units:
+            units.write(f"{name} loaded active running {name}\n")
+        (self.state / f"show-{name}").write_text("".join(f"{p}\n" for p in properties))
+
+    def file(self, relative: str, content: str) -> Path:
+        path = self.root / relative
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_text(content)
+        return path
+
+    def run(self, *, systemctl: bool = True) -> str:
+        if not systemctl:
+            (self.bin / "systemctl").unlink()
+        return subprocess.run(
+            ["/bin/sh", "-c", VIRTUAL_IPS_COMMAND],
+            capture_output=True,
+            text=True,
+            env={"PATH": str(self.bin)},
+            timeout=30,
+        ).stdout
+
+
+def _exec_start(*argv: str) -> str:
+    return (
+        f"ExecStart={{ path={argv[0]} ; argv[]={' '.join(argv)} ; ignore_errors=no ; "
+        "start_time=[Mon 2026-10-05 06:50:00 UTC] ; stop_time=[n/a] ; pid=758 ; "
+        "code=(null) ; status=0/0 }"
+    )
+
+
+@pytest.fixture(params=_awks())
+def host(request, tmp_path) -> _Host:
+    return _Host(tmp_path, request.param)
+
+
+class TestTheCommandOnAHost:
+    def test_vip_manager_5x_as_on_db_01(self, host):
+        config = host.file("etc/default/vip-manager.yml", DB01_YAML)
+        host.unit(
+            "vip-manager.service",
+            "ActiveState=active",
+            _exec_start("/usr/bin/vip-manager", f"--config={config}"),
+            f"Environment=VIP_ETCD_PASSWORD={SECRET} VIP_INTERFACE=ens7",
+        )
+
+        out = host.run()
+
+        assert SECRET not in out
+        assert "10.7.234.12" not in out  # the etcd endpoints stay on the host
+        assert parse_virtual_ips(out)["vip_manager"] == [
+            {
+                "mechanism": "vip-manager",
+                "instance": "vip-manager",
+                "address": "10.7.224.10",
+                "prefix_length": 24,
+                "interface": "ens7",
+                "group_key": "/service/netork-db/leader",
+                "running": True,
+            }
+        ]
+
+    def test_vip_manager_1x_with_its_environment_file(self, host):
+        envfile = host.file(
+            "etc/default/vip-manager",
+            f'VIP_IP="10.0.0.10"\nexport VIP_MASK=24\nVIP_IFACE=eth0\nVIP_KEY=/service/pg/leader\n'
+            f"VIP_ETCD_PASSWORD={SECRET}\n",
+        )
+        host.unit(
+            "vip-manager.service",
+            "ActiveState=active",
+            _exec_start(
+                "/usr/bin/vip-manager",
+                "-ip=${VIP_IP}",
+                "-mask=${VIP_MASK}",
+                "-iface=${VIP_IFACE}",
+                "-key=${VIP_KEY}",
+                "-etcd_password",
+                SECRET,
+            ),
+            f"EnvironmentFiles={envfile} (ignore_errors=yes)",
+            f"EnvironmentFiles={host.root}/etc/default/missing (ignore_errors=yes)",
+        )
+
+        out = host.run()
+
+        assert SECRET not in out
+        [entry] = parse_virtual_ips(out)["vip_manager"]
+        assert (entry["address"], entry["prefix_length"], entry["interface"]) == (
+            "10.0.0.10",
+            24,
+            "eth0",
+        )
+
+    def test_keepalived_with_includes(self, host):
+        config = host.file("etc/keepalived/keepalived.conf", KEEPALIVED_CONF)
+        host.file("etc/keepalived/conf.d/db.conf", INCLUDED_CONF)
+        host.unit(
+            "keepalived.service",
+            "ActiveState=active",
+            _exec_start("/usr/sbin/keepalived", "--dont-fork", "-f", str(config)),
+        )
+
+        out = host.run()
+
+        assert SECRET not in out
+        assert "MASTER" not in out and "priority" not in out
+        reading = parse_virtual_ips(out)
+        assert [(e["instance"], e["address"], e["prefix_length"], e["interface"], e["group_key"])
+                for e in reading["keepalived"]] == [
+            ("VI_WEB", "192.0.2.10", 24, "eth0", "vrid:51"),
+            ("VI_DB", "192.0.2.20", 32, "eth1", "vrid:52"),
+            ("VI_DB", "2001:db8::20", 64, "eth1", "vrid:52"),
+        ]
+        assert reading["vip_manager"] == []
+
+    def test_an_include_that_includes_itself_is_not_followed_forever(self, host):
+        config = host.file(
+            "etc/keepalived/keepalived.conf",
+            "vrrp_instance a {\n virtual_ipaddress {\n 192.0.2.30\n }\n}\n"
+            "include keepalived.conf\n",
+        )
+        host.unit(
+            "keepalived.service",
+            "ActiveState=active",
+            _exec_start("/usr/sbin/keepalived", f"--use-file={config}"),
+        )
+
+        assert parse_virtual_ips(host.run())["keepalived"] is None
+
+    def test_an_include_pattern_is_never_run_as_shell(self, host):
+        """The include line comes from a file and the command runs as root."""
+        marker = host.root / "PWNED"
+        config = host.file(
+            "etc/keepalived/keepalived.conf",
+            f"include /etc/x.conf;touch${{IFS}}{marker}\ninclude $(touch {marker})\n",
+        )
+        host.unit(
+            "keepalived.service",
+            "ActiveState=active",
+            _exec_start("/usr/sbin/keepalived", "-f", str(config)),
+        )
+
+        out = host.run()
+
+        assert not marker.exists()
+        assert parse_virtual_ips(out)["keepalived"] is None
+
+    @pytest.mark.skipif(os.geteuid() == 0, reason="root reads every file")
+    def test_a_config_it_may_not_read_is_reported_as_such(self, host):
+        config = host.file("etc/default/vip-manager.yml", DB01_YAML)
+        config.chmod(0o000)
+        host.unit(
+            "vip-manager.service",
+            "ActiveState=active",
+            _exec_start("/usr/bin/vip-manager", "--config", str(config)),
+        )
+
+        out = host.run()
+
+        assert parse_virtual_ips(out)["vip_manager"] is None
+
+    def test_no_units_at_all(self, host):
+        assert parse_virtual_ips(host.run()) == {"vip_manager": [], "keepalived": []}
+
+    def test_a_host_without_systemd(self, host):
+        assert parse_virtual_ips(host.run(systemctl=False)) == {
+            "vip_manager": None,
+            "keepalived": None,
+        }
+
+
+class TestTheCommand:
+    def test_the_frame_is_not_in_the_command_itself(self):
+        assert "VIPS_BEGIN" not in VIRTUAL_IPS_COMMAND
+        assert "VIPS_END" not in VIRTUAL_IPS_COMMAND
+
+    def test_it_writes_and_changes_nothing(self):
+        for verb in ("sudo", " > ", ">>", "kill", "rm ", "start", "stop", "restart", "reload"):
+            assert verb not in VIRTUAL_IPS_COMMAND
+
+    def test_it_is_posix_sh(self):
+        result = subprocess.run(
+            ["sh", "-n", "-c", VIRTUAL_IPS_COMMAND], capture_output=True, text=True
+        )
+        assert result.returncode == 0, result.stderr
+
+    def test_it_is_one_line(self):
+        """About 4 kB with its awk program -- past the line a terminal takes, so a
+        driver sends it on an exec channel. One argument there, one line."""
+        driver = _Driver(_wire())
+        driver.get_virtual_ips()
+
+        [(command, _privileged)] = driver.calls
+        assert "\n" not in command
+
+
+class _Driver(VirtualIpsMixin):
+    def __init__(self, privileged: str, unprivileged: str = "") -> None:
+        self.answers = {True: privileged, False: unprivileged}
+        self.calls: list = []
+
+    def _run_virtual_ips_command(self, command: str, *, privileged: bool) -> str:
+        self.calls.append((command, privileged))
+        return self.answers[privileged]
+
+
+class TestTheTemplate:
+    def test_a_driver_supplies_only_the_transport(self):
+        driver = _Driver(_wire(DB01))
+
+        reading = driver.get_virtual_ips()
+
+        assert reading["vip_manager"][0]["address"] == "10.7.224.10"
+        [(command, privileged)] = driver.calls
+        assert privileged is True
+        assert command.startswith("sh -c '")
+        assert subprocess.run(["sh", "-n", "-c", command]).returncode == 0
+
+    def test_without_root_it_reads_what_it_can(self):
+        """The config may be root's alone; what is readable is still worth having."""
+        driver = _Driver(
+            "sudo: a password is required\n",
+            _wire(_vip_manager("active active", "unreadable /etc/default/vip-manager.yml")),
+        )
+
+        reading = driver.get_virtual_ips()
+
+        assert reading["vip_manager"] is None
+        assert [p for _, p in driver.calls] == [True, False]
+
+    def test_not_every_os_driver_has_it(self):
+        assert not issubclass(OSDriver, VirtualIpsMixin)
+        assert not hasattr(OSDriver, "get_virtual_ips")

tests/test_vm_identifiers.py

@@ -51,20 +51,3 @@ class TestVMConfigCarriesWhatAHardwareViewShows:
         from napalm_device_types.models import VMPassthroughDict
 
         assert get_type_hints(VMPassthroughDict) == {"slot": str, "kind": str, "config": str}
-
-
-class TestGuestAgentDeclaration:
-    """netOrk's cloud-init installed qemu-guest-agent on every new VM. A VMware
-    guest reports its IP through open-vm-tools instead; the hypervisor says
-    which, and netOrk stops hard-coding one of them."""
-
-    def test_default_is_qemu_guest_agent(self):
-        from napalm_device_types import HypervisorDriver
-
-        assert HypervisorDriver.GUEST_AGENT_PACKAGES == ("qemu-guest-agent",)
-        assert HypervisorDriver.GUEST_AGENT_RUNCMD == ("systemctl enable --now qemu-guest-agent",)
-
-    def test_attributes_are_not_methods(self):
-        from napalm_device_types import HypervisorDriver
-
-        assert not callable(vars(HypervisorDriver)["GUEST_AGENT_PACKAGES"])