napalm-linux/napalm_linux/linux.py

# Licensed under the Apache License, Version 2.0

"""NAPALM driver for generic Linux systems.

Connects via SSH using netmiko (device_type ``linux``) and supports
auto-detection of the installed package manager:

  * apt  — Debian, Ubuntu, Raspberry Pi OS, …
  * dnf  — RHEL 8+, Rocky Linux, AlmaLinux, Fedora
  * yum  — RHEL 7, CentOS 7
  * apk  — Alpine Linux
  * pacman — Arch Linux, Manjaro

A specific package manager can be forced with
``optional_args={"pkg_manager": "apt"}``.
"""

from __future__ import annotations

import ipaddress
import logging
import re
from shlex import quote as _shlex_quote
from typing import Any

from napalm.base.exceptions import ConnectionClosedException, ConnectionException
from napalm.base.netmiko_helpers import netmiko_args
from napalm_device_types import (
    APT_UPGRADABLE_COMMAND,
    DNF_SECURITY_COMMAND,
    AptProxyMixin,
    ByteStream,
    CommandResult,
    ContainerEngineMixin,
    EnsurePackageMixin,
    FingerprintRule,
    HostFirewallMixin,
    HostStatusMixin,
    KernelFactsMixin,
    ListeningSocketsMixin,
    OpenSshServerMixin,
    OSDriver,
    ServiceClustersMixin,
    ServicePackagesMixin,
    SnmpAgentMixin,
    SystemdServicesMixin,
    SystemdUnavailable,
    SystemIdentityMixin,
    VirtualIpsMixin,
    open_stream_on_transport,
    parse_apt_upgradable,
    parse_dnf_security,
    run_fix_steps,
    run_on_transport,
    strip_terminal_codes,
)
from napalm_device_types.models import (
    ApplyUpdatesResultDict,
    CronJobDict,
    FixResultDict,
    FixStepOutcomeDict,
    PackageDict,
    ProcessDict,
    ServiceDict,
    SnmpAgentSettingsDict,
    SNMPConfigDict,
    UpdateDict,
    UserDict,
)
from napalm_device_types.ssh_host_keys import HostKeyCheck
from netmiko import ConnectHandler
from netmiko.exceptions import (
    NetmikoAuthenticationException,
    NetmikoTimeoutException,
)

logger = logging.getLogger("napalm_linux")

# Package managers in detection order
_PKG_MANAGERS = ["apt", "dnf", "yum", "apk", "pacman"]

#: Printed after a command run through the interactive shell (``_run_privileged``
#: callers) so its exit status survives the trip. Matched only on a line of its own with a
#: number after it — an echoed command line carries the literal ``$?`` instead.
_RC_MARKER = "__NETORK_RC="
_RC_MARKER_RE = re.compile(rf"^{_RC_MARKER}(\d+)\s*$", re.MULTILINE)


#: The end of a command's output when nothing better is known: a line that looks
#: like a shell prompt.
_PROMPT_RE = r"[#$\>]\s*$"
#: A command ending in ``echo __NAME=$?`` reports its exit status on a line of its
#: own: ``_RC_MARKER`` here, ``__APT_RC=`` and ``__SVC_RC=`` in napalm-device-types.
_STATUS_ECHO_RE = re.compile(r"echo\s+(__[A-Z_]+=)\$\?")


def _expect_for(command: str) -> str:
    """The pattern that ends *command*'s output.

    netmiko stops reading as soon as the pattern matches what it has read so far.
    A line the command prints can end in ``#``, ``$`` or ``>`` -- apt's
    ``<ftpmaster@ubuntu.com>`` after a bad signature -- and was taken for the
    prompt: half the output came back, and the rest started the next command's
    (#615). A command that echoes its exit status is read until that marker, with
    a number, and the prompt line after it. The echoed command line carries a
    literal ``$?`` and cannot match.
    """
    markers = _STATUS_ECHO_RE.findall(command)
    if not markers:
        return _PROMPT_RE
    return re.escape(markers[-1]) + r"\d+\s*\n.*" + _PROMPT_RE


def _split_status(raw: str) -> tuple[str, int | None]:
    """``(output, exit_status)`` of a command followed by ``echo {_RC_MARKER}$?``.

    The status is ``None`` when the marker never arrived (output cut short), so
    a caller can tell "unknown" from "succeeded".
    """
    raw = strip_terminal_codes(raw)
    matches = list(_RC_MARKER_RE.finditer(raw))
    if not matches:
        return raw, None
    last = matches[-1]
    return (raw[: last.start()] + raw[last.end() :]).strip(), int(last.group(1))


#: How each package manager refreshes its index. pacman is left out on purpose:
#: ``pacman -Sy`` without ``-u`` invites a partial upgrade on the next install.
_REFRESH = {
    # No LC_ALL=C here: under sudo it is an environment variable sudoers may refuse
    # to set. Only the exit status decides, so the language is merely what is shown.
    "apt": "apt-get update -q 2>&1",
    "dnf": "dnf makecache -q 2>&1",
    "yum": "yum makecache -q 2>&1",
    "apk": "apk update -q 2>&1",
}
#: yum's counterpart of DNF_SECURITY_COMMAND, in the same shape: parse_dnf_security
#: (napalm-device-types 4.7) needs the ``__DNF_RC=`` line, and behind ``| cat`` it is
#: the only place yum's own exit status survives.
_YUM_SECURITY_COMMAND = (
    "{ LC_ALL=C yum updateinfo list security -q 2>/dev/null; echo __DNF_RC=$?; } | cat"
)
#: The exit status ``DNF_SECURITY_COMMAND`` and ``_YUM_SECURITY_COMMAND`` report.
_DNF_RC_RE = re.compile(r"^__DNF_RC=(\d+)\s*$", re.MULTILINE)

#: Restart the host two seconds later, detached from this session: the launcher's
#: exit status comes back before the host goes down, and closing the session
#: cannot take the restart with it.
_REBOOT_DETACHED = "sh -c '(trap \"\" HUP; sleep 2; /sbin/reboot) </dev/null >/dev/null 2>&1 &'"

#: Why ``run_command`` refuses stdin to a privileged command while a sudo
#: password is set (NetOrk/netork#1029).
_PRIVILEGED_STDIN = (
    "run_command(privileged=True) takes no stdin while a sudo password is set: "
    "sudo -S reads the password from stdin, and a sudo that does not prompt "
    "(NOPASSWD, a cached timestamp) leaves it for the command, which would read it "
    "as the first line of the data. Stage the data without privilege (umask 077, "
    "mktemp, cat) and have root install that file."
)

#: Sent before anything else in the interactive session (NetOrk/netork#1039):
#: the shell writes no history file and records nothing more. The leading space
#: keeps this line itself out of a history under ``HISTCONTROL=ignorespace``;
#: a shell without ``set -o history`` (dash) only complains, to /dev/null.
_HISTORY_OFF = " unset HISTFILE; set +o history 2>/dev/null; true"

#: First in every script ``_sudo`` runs: a sudo that does not prompt (NOPASSWD,
#: a cached timestamp) leaves the password on stdin, and nothing may read it.
_NO_STDIN = "exec </dev/null; "

#: Where ``_install_file_as_root`` stages content: a fresh file from ``mktemp``
#: under ``umask 077``, never a fixed name another user could create first.
#: The path is on the device, not a local temp file.
_STAGE_TEMPLATE = "/tmp/netork-stage.XXXXXX"  # nosec B108
_STAGE_PREFIX = _STAGE_TEMPLATE[: -len("XXXXXX")]
_STAGE = (
    f"umask 077 && t=$(mktemp {_STAGE_TEMPLATE}) && "
    '{ cat > "$t" && printf %s "$t" || { s=$?; rm -f -- "$t"; exit $s; }; }'
)

#: What to do when sudo wants a password netOrk does not have (a message, not a password).
_SUDO_PASSWORD_HINT = (  # nosec B105
    "sudo requires a password on this host and no sudo password was given: pass one, "
    "or allow this user passwordless sudo (NOPASSWD)."
)

# DMI field values that carry no useful information (OEM defaults, blanks)
_BAD_DMI: frozenset[str] = frozenset(
    {
        "",
        "none",
        "n/a",
        "not specified",
        "not applicable",
        "to be filled by o.e.m.",
        "default string",
        "unknown",
        "no asset tag",
        "not present",
    }
)

# systemd-detect-virt output → human-readable vendor name
_VIRT_VENDOR_MAP: dict[str, str] = {
    "kvm": "KVM",
    "qemu": "KVM",
    "vmware": "VMware ESXi",
    "microsoft": "Microsoft Hyper-V",
    "xen": "Xen",
    "virtualbox": "Oracle VirtualBox",
    "parallels": "Parallels",
    "docker": "Docker",
    "podman": "Podman",
    "lxc": "LXC",
    "lxc-libvirt": "LXC",
    "systemd-nspawn": "systemd-nspawn",
}

# Container technologies reported by systemd-detect-virt
_CONTAINER_VIRT: frozenset[str] = frozenset(
    {
        "docker",
        "podman",
        "lxc",
        "lxc-libvirt",
        "systemd-nspawn",
    }
)

# DMI sys_vendor strings that indicate a VM when detect-virt is unavailable
_VM_DMI_VENDORS: frozenset[str] = frozenset(
    {
        "qemu",
        "vmware, inc.",
        "microsoft corporation",
        "innotek gmbh",
        "xen",
        "bochs",
        "parallels software international inc.",
    }
)

# Known ARM board model prefixes → canonical vendor name
_ARM_VENDOR_PREFIXES: list[tuple[str, str]] = [
    ("Raspberry Pi", "Raspberry Pi Foundation"),
    ("NVIDIA Jetson", "NVIDIA"),
    ("ODROID", "Hardkernel"),
    ("Hardkernel", "Hardkernel"),
    ("Rock Pi", "Radxa"),
    ("ROCK Pi", "Radxa"),
    ("Radxa", "Radxa"),
    ("Orange Pi", "Xunlong Software"),
    ("Banana Pi", "SinoVoip"),
    ("NanoPi", "FriendlyElec"),
    ("PINE64", "Pine64"),
    ("BeagleBone", "BeagleBoard.org"),
]


def _arm_vendor_from_model(model: str) -> str:
    """Extract vendor from an ARM device tree / cpuinfo model string."""
    for prefix, vendor in _ARM_VENDOR_PREFIXES:
        if model.startswith(prefix):
            return vendor
    # Generic fallback: words before the first numeric token
    brand = []
    for word in model.split():
        if word[0].isdigit():
            break
        brand.append(word)
    return " ".join(brand)


#: Where net-snmp reads its configuration, and how each package manager installs it.
_SNMPD_CONF = "/etc/snmp/snmpd.conf"
_SNMP_INSTALL = {
    "apt": (
        "apt-get update -y >/dev/null 2>&1; "
        "DEBIAN_FRONTEND=noninteractive apt-get install -y snmpd snmp 2>&1"
    ),
    "dnf": "dnf install -y net-snmp net-snmp-utils 2>&1",
    "yum": "yum install -y net-snmp net-snmp-utils 2>&1",
    "apk": "apk add --no-cache net-snmp net-snmp-tools 2>&1",
    "pacman": "pacman -Sy --noconfirm net-snmp 2>&1",
}
_USER_NAME = re.compile(r"^[a-z_][a-z0-9_.\-]{0,31}$")
#: What may stand in an snmpd.conf value: no whitespace, no quote, no comment.
_SNMP_WORD = re.compile(r"^[^\s\"#]+$")


def _source(value: str) -> str:
    """An allowed source as an address or network, or ValueError."""
    return str(ipaddress.ip_network(value.strip(), strict=False))


def render_snmpd_conf(settings: SnmpAgentSettingsDict) -> str:
    """net-snmp's configuration for *settings*: read-only v2c, per allowed source."""
    community = settings["community"]
    if not _SNMP_WORD.match(community):
        raise ValueError("the community must be one word without quotes or #")
    listen = settings.get("listen")
    lines = [f"agentAddress udp:{listen + ':' if listen else ''}161"]
    sources = [_source(s) for s in settings.get("allowed_sources", [])]
    lines += [f"rocommunity {community} {s}" for s in sources] or [f"rocommunity {community}"]
    for key, directive in (("location", "sysLocation"), ("contact", "sysContact")):
        value = settings.get(key, "")
        if value:
            if "\n" in value:
                raise ValueError(f"{key} must be one line")
            lines.append(f"{directive}    {value}")
    return "\n".join(lines) + "\n"


class LinuxDriver(
    SnmpAgentMixin,
    EnsurePackageMixin,
    AptProxyMixin,
    KernelFactsMixin,
    ServicePackagesMixin,
    ListeningSocketsMixin,
    HostFirewallMixin,
    VirtualIpsMixin,
    ServiceClustersMixin,
    SystemIdentityMixin,
    OpenSshServerMixin,
    SystemdServicesMixin,
    HostStatusMixin,
    ContainerEngineMixin,
    OSDriver,
):
    """NAPALM driver for generic Linux systems.

    Connects via SSH (netmiko ``linux`` device type) and auto-detects the
    package manager unless overridden by ``optional_args["pkg_manager"]``.
    """

    TYPE_LABEL = "Linux"
    VENDOR = "Linux"
    DRIVER_NAME = "linux"
    # A general-purpose host runs through a full init sequence; NAS derivatives
    # (OpenMediaVault, QNAP) inherit this and are, if anything, slower.
    REBOOT_SETTLE_SECONDS = 90
    SNMP_FINGERPRINT = [
        FingerprintRule("linux", weight=5.0),
    ]
    SSH_FINGERPRINT = [
        FingerprintRule("ubuntu", weight=4.0),
        FingerprintRule("debian", weight=4.0),
        FingerprintRule("alpine", weight=4.0),
        FingerprintRule("openssh", weight=1.0),
    ]
    NETMIKO_DEVICE_TYPE = "linux"

    def __init__(
        self,
        hostname: str,
        username: str,
        password: str,
        timeout: int = 60,
        optional_args: dict | None = None,
    ) -> None:
        self.hostname = hostname
        self.username = username
        self.password = password
        self.timeout = timeout

        if optional_args is None:
            optional_args = {}

        self.port: int = optional_args.get("port", 22)
        self._forced_pkg_manager: str | None = optional_args.get("pkg_manager")
        self._secret: str = optional_args.get("secret", password)
        # Optional sudo password for privilege escalation (e.g. apt-get update)
        self._sudo_password: str | None = optional_args.get("sudo_password")
        # Expected apt proxy URL — checked as a device warning on apt systems.

        if optional_args.get("debugging"):
            logger.setLevel(logging.DEBUG)

        self.netmiko_optional_args = netmiko_args(optional_args)
        # port is passed explicitly in open() — remove it from netmiko_optional_args
        # to avoid "multiple values for keyword argument 'port'"
        self.netmiko_optional_args.pop("port", None)
        # The host key to expect, and where the one seen is reported (netork#1167).
        self._host_key = HostKeyCheck(optional_args, hostname, self.port)

        # Runtime state
        self._device: ConnectHandler | None = None
        self._pkg_manager: str | None = None  # set after open()

    # ------------------------------------------------------------------
    # Connection management
    # ------------------------------------------------------------------

    def open(self) -> None:
        """Open the SSH connection and detect the package manager."""
        try:
            with self._host_key.netmiko(self.netmiko_optional_args) as netmiko_args_:
                self._device = ConnectHandler(
                    device_type=self.NETMIKO_DEVICE_TYPE,
                    host=self.hostname,
                    username=self.username,
                    password=self.password,
                    port=self.port,
                    secret=self._secret,
                    timeout=self.timeout,
                    **netmiko_args_,
                )
        except NetmikoAuthenticationException as exc:
            raise ConnectionException(str(exc)) from exc
        except NetmikoTimeoutException as exc:
            raise ConnectionException(str(exc)) from exc
        self._host_key.observe(self._device.remote_conn_pre.get_transport())

        # No history before anything else is typed (NetOrk/netork#1039).
        try:
            self._device.send_command(_HISTORY_OFF, expect_string=_PROMPT_RE)
        except Exception:
            logger.debug(
                "Could not switch the shell history off on %s", self.hostname, exc_info=True
            )

        # Prevent PTY from wrapping long output lines (e.g. docker JSON).
        try:
            self._device.send_command(
                "stty cols 10000 2>/dev/null || true", expect_string=r"[#$>]\s*$"
            )
        except Exception:
            logger.debug("%s: ignored", self.hostname, exc_info=True)

        self._pkg_manager = self._forced_pkg_manager or self._detect_pkg_manager()
        logger.debug("Connected to %s, pkg_manager=%s", self.hostname, self._pkg_manager)

    def close(self) -> None:
        """Close the SSH connection."""
        if self._device:
            try:
                self._device.disconnect()
            except Exception:
                logger.debug("%s: ignored", self.hostname, exc_info=True)
            self._device = None
        self._pkg_manager = None

    def is_alive(self) -> dict[str, bool]:
        if self._device:
            try:
                return {"is_alive": self._device.remote_conn.transport.is_active()}
            except (OSError, AttributeError, EOFError):
                return {"is_alive": False}
        return {"is_alive": False}

    # ------------------------------------------------------------------
    # Internal helpers
    # ------------------------------------------------------------------

    def _send(self, command: str, read_timeout: float = 100) -> str:
        """Send a command and return stripped output."""
        if not self._device:
            raise ConnectionClosedException("Not connected")
        return self._device.send_command(
            command,
            read_timeout=read_timeout,
            cmd_verify=False,
            expect_string=_expect_for(command),
        ).strip()

    def _sudo(self, command: str, read_timeout: float = 100) -> str:
        """Run *command* as root and return its output, stdout and stderr together.

        It runs on an exec channel, not in netmiko's interactive shell
        (NetOrk/netork#1039): that shell keeps a history, and ``echo <password> |
        sudo -S`` typed into it wrote the password into ``~/.bash_history`` on
        every poll, and into the process list while it ran. On the exec channel
        the password reaches ``sudo -S`` on stdin (``_as_root``), and there is no
        PTY to echo it back into the output.

        The whole of *command* runs as root in one ``sh -c``, ``;`` and ``&&``
        included. Without a sudo password ``sudo -n`` fails at once; a root
        login runs it directly.
        """
        return self._sudo_result(command, read_timeout).stdout.strip()

    def _sudo_result(self, command: str, timeout: float) -> CommandResult:
        line, prefix = self._as_root(f"{_NO_STDIN}{command}")
        return run_on_transport(self._transport(), f"{line} 2>&1", stdin=prefix, timeout=timeout)

    def _install_file_as_root(self, path: str, content: str) -> CommandResult:
        """Replace *path* with *content*, as a new file owned by root, mode 644.

        The content is staged without privilege in a fresh ``mktemp`` file only
        the login user may read; its stdin is the content's, not a privileged
        command's (``_PRIVILEGED_STDIN``). Root then ``install``s that file as a
        new regular file next to *path* and renames it over *path*: nothing root
        runs follows a symlink or changes the owner or mode of an existing path
        (NetOrk/netork#1039). Nothing runs as root when staging failed.
        """
        staged = self.run_command(_STAGE, timeout=30, stdin=content.encode())
        lines = staged.stdout.strip().splitlines()
        tmp = lines[-1] if lines else ""
        if (
            staged.exit_code != 0
            or not tmp.startswith(_STAGE_PREFIX)
            or "/" in tmp[len(_STAGE_PREFIX) :]
        ):
            reason = (staged.stderr or staged.stdout).strip() or f"could not stage {path}"
            return CommandResult(reason, "", staged.exit_code or 1)
        tmp_q, part_q, path_q = (_shlex_quote(p) for p in (tmp, f"{path}.netork-tmp", path))
        return self._sudo_result(
            f"install -m 644 -o root -g root -- {tmp_q} {part_q} && mv -f -- {part_q} {path_q};"
            f" s=$?; rm -f -- {tmp_q} {part_q}; exit $s",
            30,
        )

    def _sudo_status(self, command: str, read_timeout: float = 100) -> tuple[str, int | None]:
        """Run *command* as root and return ``(output, exit_status)``.

        ``_sudo`` callers append ``|| true`` so a failing command yields output
        instead of an error, which throws the exit status away. The exec channel
        reports the status itself: the command's, or sudo's when it refused.

        The status is ``None`` when the channel closed without one (paramiko's
        -1), so a caller can tell "unknown" from "succeeded".
        """
        result = self._sudo_result(command, read_timeout)
        status = result.exit_code if result.exit_code >= 0 else None
        return result.stdout.strip(), status

    def _is_root(self) -> bool:
        """Whether the SSH user is root, asked once per session.

        A root login on a box without sudo (an LXC container, a minimal Debian)
        must not have its commands prefixed with a sudo that is not there.
        """
        if getattr(self, "_root", None) is None:
            self._root = self._send("id -u") == "0"
        return bool(self._root)

    def _run_service_command(self, command: str, *, privileged: bool, timeout: int) -> str:
        """The transport for :class:`SystemdServicesMixin`.

        Without a sudo password, ``sudo -n`` fails at once where a prompt would
        otherwise hang the session until the read timeout.
        """
        if not privileged:
            return self._send(command, read_timeout=timeout)
        return self._run_privileged(command, timeout)

    def _run_privileged(self, command: str, timeout: float = 100) -> str:
        """Run *command* as root: directly for a root login, through ``_sudo``
        with a sudo password, and through ``sudo -n`` without one -- which fails at
        once where a password prompt would hang the session until the timeout."""
        if self._is_root():
            return self._send(command, read_timeout=timeout)
        if self._sudo_password:
            return self._sudo(command, read_timeout=timeout)  # never typed into the shell
        return self._send(f"sudo -n {command}", read_timeout=timeout)

    # ------------------------------------------------------------------
    # Command channel (napalm-device-types CommandChannelMixin)
    # ------------------------------------------------------------------

    def _transport(self) -> Any:
        """The SSH transport netmiko already holds, for exec channels next to its PTY."""
        if not self._device:
            raise ConnectionClosedException("Not connected")
        return self._device.remote_conn_pre.get_transport()

    def _privileged(self, command: str, privileged: bool) -> tuple[str, bytes | None]:
        """The command line that runs *command* with the privileges asked for,
        and what has to reach its stdin first.

        Root runs it directly. With a sudo password, ``sudo -S`` reads it from
        stdin, so it never appears in a process list. Without one, ``sudo -n``
        fails at once where a prompt would hang. The command goes to ``sh -c``
        as one argument, so the privilege covers every part of it.
        """
        if not privileged or self._is_root():
            return command, None
        return self._as_root(command)

    def _as_root(self, command: str) -> tuple[str, bytes | None]:
        """The ``sh -c`` line that runs *command* as root, and its stdin prefix.

        Root runs it directly; with a sudo password, ``sudo -S`` reads it from
        stdin; without one, ``sudo -n`` fails at once instead of prompting.
        """
        script = _shlex_quote(command)
        if self._is_root():
            return f"sh -c {script}", None
        if self._sudo_password:
            return f"sudo -S -p '' sh -c {script}", f"{self._sudo_password}\n".encode()
        return f"sudo -n sh -c {script}", None

    def run_command(
        self,
        command: str,
        *,
        privileged: bool = False,
        timeout: float = 60,
        stdin: bytes | None = None,
    ) -> CommandResult:
        """Run *command* on an exec channel: no PTY, stderr apart, a real exit code.

        :raises ValueError: for *stdin* to a privileged command while a sudo
            password is set (see ``_PRIVILEGED_STDIN``).
        """
        line, prefix = self._privileged(command, privileged)
        if prefix and stdin is not None:
            raise ValueError(_PRIVILEGED_STDIN)
        data = prefix or stdin or None
        return run_on_transport(self._transport(), line, stdin=data, timeout=timeout)

    def open_stream(self, command: str, *, privileged: bool = False) -> ByteStream:
        """Start *command* on an exec channel and return a stream to it.

        A privileged stream with a sudo password starts with that password, for
        ``sudo -S``; a sudo that does not prompt leaves it for *command*. Write
        nothing to such a stream that must not follow the password, as
        ``_PRIVILEGED_STDIN`` says for ``run_command``.
        """
        line, prefix = self._privileged(command, privileged)
        return open_stream_on_transport(self._transport(), line, stdin_prefix=prefix)

    def reboot_host(self) -> None:
        """Restart the host (``HostRebootMixin``); returns once the restart is under way.

        :raises RuntimeError: when the host refuses -- sudo without a password,
            no ``reboot`` -- or its answer carried no exit status.
        """
        output, status = _split_status(
            self._run_privileged(f"{_REBOOT_DETACHED}; echo {_RC_MARKER}$?", 30)
        )
        if status != 0:
            reason = output or f"the reboot command exited with status {status}"
            if "password is required" in output:
                reason = f"{reason}\n{_SUDO_PASSWORD_HINT}"
            raise RuntimeError(reason)

    def _run_host_status_command(self, command: str) -> str:
        """The transport for ``HostStatusMixin.get_host_status``: read-only, no sudo."""
        return self._send(command, read_timeout=60)

    def _detect_pkg_manager(self) -> str | None:
        """Return the first package manager binary found on PATH."""
        for pm in _PKG_MANAGERS:
            result = self._send(f"command -v {pm} 2>/dev/null")
            if result:
                return pm
        return None

    # ------------------------------------------------------------------
    # Standard NAPALM – read-only
    # ------------------------------------------------------------------

    def _collect_platform_info(self) -> dict[str, Any]:
        """Collect hardware/virtualisation info in a single SSH round-trip.

        Returns a dict with keys:
          - vendor  (str)  — hardware vendor or hypervisor name; "" if unknown
          - model   (str)  — product model or "Virtual Machine"/"Container"; "" if unknown
          - serial  (str)  — product serial, or VM UUID as fallback; "" if unknown
          - is_vm   (bool) — True for VMs and containers
        """
        dmi_cmd = (
            "v=$(cat /sys/class/dmi/id/sys_vendor 2>/dev/null); "
            "n=$(cat /sys/class/dmi/id/product_name 2>/dev/null); "
            "r=$(cat /sys/class/dmi/id/product_version 2>/dev/null); "
            "s=$(cat /sys/class/dmi/id/product_serial 2>/dev/null); "
            "u=$(cat /sys/class/dmi/id/product_uuid 2>/dev/null); "
            "d=$(systemd-detect-virt 2>/dev/null); d=${d:-none}; "
            "dt=$(tr -d '\\0' </sys/firmware/devicetree/base/model 2>/dev/null); "
            "cs=$(grep '^Serial' /proc/cpuinfo 2>/dev/null | head -1 | cut -d: -f2 "
            "| xargs 2>/dev/null); "
            "cm=$(grep '^Model' /proc/cpuinfo 2>/dev/null | head -1 | cut -d: -f2 "
            "| xargs 2>/dev/null); "
            # DMIBEGIN sentinel: _send() strips leading blank lines (ARM has no DMI
            # files, so fields 0-4 are empty). The sentinel anchors the output so
            # splitlines()[start+N] always maps to the correct field index.
            "printf 'DMIBEGIN\\n%s\\n%s\\n%s\\n%s\\n%s\\n%s\\n%s\\n%s\\n%s\\n' "
            '"$v" "$n" "$r" "$s" "$u" "$d" "$dt" "$cs" "$cm"'
        )
        try:
            raw_lines = self._send(dmi_cmd).splitlines()
            try:
                start = raw_lines.index("DMIBEGIN") + 1
            except ValueError:
                start = 0
            lines = raw_lines[start:]
        except Exception:
            return {"vendor": "", "model": "", "serial": "", "is_vm": False}

        def _clean(idx: int) -> str:
            val = lines[idx].strip() if idx < len(lines) else ""
            return "" if val.lower() in _BAD_DMI else val

        sys_vendor = _clean(0)
        product_name = _clean(1)
        product_ver = _clean(2)
        product_ser = _clean(3)
        product_uuid = _clean(4)
        detect_virt = lines[5].strip().lower() if len(lines) > 5 else "none"
        dt_model = _clean(6)
        cpuinfo_ser = _clean(7)
        cpuinfo_mdl = _clean(8)

        is_container = detect_virt in _CONTAINER_VIRT
        is_vm = detect_virt not in ("none", "") or sys_vendor.lower() in _VM_DMI_VENDORS

        if is_container:
            return {
                "vendor": _VIRT_VENDOR_MAP.get(detect_virt, sys_vendor or "Container"),
                "model": "Container",
                "serial": product_uuid,
                "is_vm": True,
            }

        if is_vm:
            vendor = _VIRT_VENDOR_MAP.get(detect_virt, "")
            if not vendor:
                sv = sys_vendor.lower()
                if "vmware" in sv:
                    vendor = "VMware ESXi"
                elif "microsoft" in sv:
                    vendor = "Microsoft Hyper-V"
                elif "qemu" in sv or "kvm" in sv:
                    vendor = "KVM"
                elif "xen" in sv:
                    vendor = "Xen"
                elif "innotek" in sv or "virtualbox" in sv:
                    vendor = "Oracle VirtualBox"
                else:
                    vendor = sys_vendor
            return {
                "vendor": vendor,
                "model": "Virtual Machine",
                "serial": product_ser or product_uuid,
                "is_vm": True,
            }

        # Bare-metal: prefer product_version when it reads like a marketing name
        if sys_vendor or product_name:
            pv_usable = product_ver and product_ver != product_name and " " in product_ver
            return {
                "vendor": sys_vendor,
                "model": product_ver if pv_usable else product_name,
                "serial": product_ser,
                "is_vm": False,
            }

        # ARM/embedded fallback: no DMI, try device tree and /proc/cpuinfo
        arm_model = dt_model or cpuinfo_mdl
        if arm_model:
            return {
                "vendor": _arm_vendor_from_model(arm_model),
                "model": arm_model,
                "serial": cpuinfo_ser,
                "is_vm": False,
            }

        return {"vendor": "", "model": "", "serial": "", "is_vm": False}

    def get_facts(self) -> dict[str, Any]:
        """Return basic system facts."""
        hostname = self._send("hostname -s 2>/dev/null || hostname")
        fqdn = self._send("hostname -f 2>/dev/null || hostname")
        os_version = self._send(
            "cat /etc/os-release 2>/dev/null | grep '^PRETTY_NAME' | cut -d= -f2 | tr -d '\"'"
        ) or self._send("uname -r")
        uptime_secs = self._parse_uptime()
        platform = self._collect_platform_info()

        iface_out = self._send("ip -o link show | awk -F': ' '{print $2}' | cut -d@ -f1")
        interface_list = [
            i.strip() for i in iface_out.splitlines() if i.strip() and i.strip() != "lo"
        ]

        # Currently-booted kernel release, distinct from an installed-but-not-yet-
        # booted newer kernel (used for kernel CVE relevance).
        running_kernel = self._send("uname -r").strip()

        return {
            "hostname": hostname,
            "fqdn": fqdn,
            "vendor": platform["vendor"] or self.VENDOR,
            "model": platform["model"],
            "serial_number": platform["serial"],
            "os_version": os_version,
            "uptime": uptime_secs,
            "interface_list": interface_list,
            "running_kernel": running_kernel,
        }

    def _parse_uptime(self) -> int:
        """Return uptime in seconds from ``/proc/uptime``."""
        raw = self._send("cat /proc/uptime 2>/dev/null")
        try:
            return int(float(raw.split()[0]))
        except (IndexError, ValueError):
            return 0

    def get_lldp_neighbors(self) -> dict[str, list[dict[str, Any]]]:
        """Return LLDP neighbors if lldpd is installed and currently running.

        Uses ``lldpctl -f keyvalue``. Returns an empty dict when lldpd is
        absent or stopped — does NOT attempt to start the daemon.
        """
        # Check lldpctl is available
        if not self._send("command -v lldpctl 2>/dev/null").strip():
            return {}

        # Check lldpd is active (systemd or fallback to pgrep)
        running = self._send(
            "systemctl is-active lldpd 2>/dev/null || "
            "service lldpd status 2>/dev/null | grep -q running && echo active || "
            "pgrep -x lldpd >/dev/null 2>&1 && echo active || true"
        ).strip()
        if "active" not in running:
            return {}

        output = self._send("lldpctl -f keyvalue 2>/dev/null || true")
        neighbors: dict[str, list[dict[str, Any]]] = {}
        entries: dict[str, dict[str, str]] = {}

        for line in output.splitlines():
            line = line.strip()
            if "=" not in line:
                continue
            key, _, value = line.partition("=")
            parts = key.split(".")
            if len(parts) < 3 or parts[0] != "lldp":
                continue
            iface = parts[1]
            subkey = ".".join(parts[2:])
            entries.setdefault(iface, {})[subkey] = value

        for iface, data in entries.items():
            entry = {
                "hostname": data.get("chassis.name", ""),
                "port": data.get("port.ifname", data.get("port.id.value", "")),
            }
            if data.get("chassis.id.subtype") == "mac":
                mac = data.get("chassis.id.value", "")
                if re.match(r"^([0-9A-Fa-f]{2}:){5}[0-9A-Fa-f]{2}$", mac):
                    entry["mac"] = mac.lower()
            neighbors.setdefault(iface, []).append(entry)

        return neighbors

    def get_interfaces(self) -> dict[str, Any]:
        """Return interface operational data."""
        interfaces: dict[str, Any] = {}

        # ip -o link show: one line per interface
        link_out = self._send("ip -o link show")
        for line in link_out.splitlines():
            # 2: eth0: <BROADCAST,MULTICAST,UP,LOWER_UP> mtu 1500 ... state UP
            m = re.match(r"^\d+:\s+(\S+?)(?:@\S+)?:\s+<([^>]*)>.*mtu\s+(\d+).*state\s+(\S+)", line)
            if not m:
                continue
            name, flags, mtu, state = m.group(1), m.group(2), int(m.group(3)), m.group(4)
            mac_m = re.search(r"link/ether\s+([\da-f:]+)", line)
            mac = mac_m.group(1) if mac_m else ""
            is_up = "UP" in flags.split(",") or state == "UP"
            interfaces[name] = {
                "is_up": is_up,
                "is_enabled": "UP" in flags.split(","),
                "description": "",
                "last_flapped": -1.0,
                "speed": -1.0,
                "mtu": mtu,
                "mac_address": mac,
            }

        return interfaces

    def get_interfaces_ip(self) -> dict[str, Any]:
        """Return IP addresses per interface."""
        result: dict[str, Any] = {}

        addr_out = self._send("ip -o addr show")
        for line in addr_out.splitlines():
            # 2: eth0    inet 192.168.1.10/24 brd ...
            m = re.match(r"^\d+:\s+(\S+)\s+(inet6?)\s+([\da-f.:]+)/(\d+)", line)
            if not m:
                continue
            iface, family, addr, prefix = m.group(1), m.group(2), m.group(3), int(m.group(4))
            af = "ipv4" if family == "inet" else "ipv6"
            result.setdefault(iface, {"ipv4": {}, "ipv6": {}})
            result[iface][af][addr] = {"prefix_length": prefix}

        return result

    def get_networks(self) -> list[dict[str, Any]]:
        """Return IP networks derived from interface addresses.

        Excludes loopback, link-local, /32 host-only addresses, and
        container-internal interfaces (docker*, br-*, veth*, virbr*).

        Each entry matches the OPNsense get_networks() schema::

            {
                "network": "10.7.224.0/24",
                "interface": "ens7",
                "gateway": "10.7.224.11",
                "family": "ipv4",
                "prefix_length": 24,
                "vlan_id": None,
            }
        """
        import ipaddress

        _SKIP_PREFIXES = ("lo", "docker", "br-", "veth", "virbr", "tun", "tap")
        networks: list[dict[str, Any]] = []

        for iface_name, af_data in self.get_interfaces_ip().items():
            if any(iface_name.startswith(p) for p in _SKIP_PREFIXES):
                continue
            for family, addrs in af_data.items():
                for addr, info in addrs.items():
                    prefix = info.get("prefix_length", 0)
                    try:
                        iface_obj = ipaddress.ip_interface(f"{addr}/{prefix}")
                        net = iface_obj.network
                        if net.is_loopback or net.is_link_local:
                            continue
                        # Skip host-only addresses (/32 IPv4, /128 IPv6)
                        if (net.version == 4 and net.prefixlen >= 32) or (
                            net.version == 6 and net.prefixlen >= 128
                        ):
                            continue
                        networks.append(
                            {
                                "network": str(net),
                                "interface": iface_name,
                                "gateway": str(iface_obj.ip),
                                "family": family,
                                "prefix_length": net.prefixlen,
                                "vlan_id": None,
                            }
                        )
                    except ValueError:
                        pass

        return networks

    def get_route_to(
        self,
        destination: str = "",
        protocol: str = "",
        longer: bool = False,
    ) -> dict[str, list[dict[str, Any]]]:
        """Return the routing table via ``ip route show``.

        OSPF routes (from FRR/Quagga) are included via ``ip route show proto ospf``
        if any are present.  The result is keyed by network prefix.
        """
        routes: dict[str, list[dict[str, Any]]] = {}

        proto_map = {
            "kernel": "connected",
            "dhcp": "dhcp",
            "static": "static",
            "ospf": "ospf",
            "bgp": "bgp",
            "bird": "bgp",
            "ra": "connected",
            "boot": "connected",
            "zebra": "zebra",
        }

        def _make_entry(
            proto: str, nexthop: str, iface: str, metric: int, network: str
        ) -> dict[str, Any]:
            family = "ipv6" if (":" in network or (nexthop and ":" in nexthop)) else "ipv4"
            return {
                "protocol": proto,
                "family": family,
                "current_active": True,
                "last_active": False,
                "age": -1,
                "next_hop": nexthop,
                "outgoing_interface": iface,
                "selected_next_hop": True,
                "preference": metric,
                "routing_table": "global",
                "protocol_attributes": {},
            }

        def _add(network: str, proto: str, nexthop: str, iface: str, metric: int) -> None:
            if destination and network != destination:
                return
            mapped = proto_map.get(proto, proto)
            if protocol and mapped != protocol.lower():
                return
            routes.setdefault(network, []).append(
                _make_entry(mapped, nexthop, iface, metric, network)
            )

        out = self._send("ip -4 route show && ip -6 route show")
        for line in out.splitlines():
            line = line.strip()
            if not line or line.startswith("#"):
                continue
            dest_m = re.match(r"^(\S+)", line)
            if not dest_m:
                continue
            raw_dest = dest_m.group(1)
            network = (
                "0.0.0.0/0"
                if raw_dest == "default"
                else ("::/0" if raw_dest == "default6" else raw_dest)
            )
            if "/" not in network:
                network += "/32"

            nexthop = ""
            nh_m = re.search(r"via\s+(\S+)", line)
            if nh_m:
                nexthop = nh_m.group(1)

            iface = ""
            dev_m = re.search(r"dev\s+(\S+)", line)
            if dev_m:
                iface = dev_m.group(1)

            proto = "kernel"
            proto_m = re.search(r"proto\s+(\S+)", line)
            if proto_m:
                proto = proto_m.group(1)

            metric = 0
            metric_m = re.search(r"metric\s+(\d+)", line)
            if metric_m:
                metric = int(metric_m.group(1))

            _add(network, proto, nexthop, iface, metric)

        # FRR/Zebra enrichment via vtysh — properly attributes OSPF/BGP/RIP protocols.
        # FRR installs routes into the kernel as "proto zebra"; vtysh gives the real source.
        _frr_code: dict[str, str] = {
            "O": "ospf",
            "B": "bgp",
            "R": "rip",
            "I": "isis",
            "S": "static",
            "K": "connected",
            "C": "connected",
        }
        if self._send("command -v vtysh 2>/dev/null").strip():
            try:
                vtysh_out = self._send(
                    "vtysh -c 'show ip route' 2>/dev/null; vtysh -c 'show ipv6 route' 2>/dev/null",
                    read_timeout=15,
                )
                for vline in vtysh_out.splitlines():
                    # "O>* 10.10.0.0/24 [110/20] via 10.255.255.2, wg0, ..."
                    vm = re.match(
                        r"^([OBSCRIKEF])[>*\s]{0,3}([\d.:a-fA-F/]+)\s+\[(\d+)/(\d+)\]"
                        r"(?:\s+via\s+([\d.:a-fA-F]+),\s*(\S+?)(?:,|$))?",
                        vline.strip(),
                    )
                    if not vm:
                        continue
                    code = vm.group(1)
                    prefix = vm.group(2)
                    metric = int(vm.group(4))
                    nexthop = vm.group(5) or ""
                    iface = (vm.group(6) or "").rstrip(",")
                    if "/" not in prefix:
                        prefix += "/32"
                    mapped_proto = _frr_code.get(code, code.lower())
                    if prefix in routes:
                        for entry in routes[prefix]:
                            entry["protocol"] = mapped_proto
                    else:
                        if not (destination and prefix != destination):
                            routes.setdefault(prefix, []).append(
                                _make_entry(mapped_proto, nexthop, iface, metric, prefix)
                            )
            except Exception:
                logger.debug("%s: ignored", self.hostname, exc_info=True)

        return routes

    def get_arp_table(self, vrf: str = "") -> list[dict[str, Any]]:
        """Return the ARP/neighbour table."""
        entries = []
        neigh_out = self._send("ip -4 neigh show")
        for line in neigh_out.splitlines():
            # 192.168.1.1 dev eth0 lladdr aa:bb:cc:dd:ee:ff REACHABLE
            m = re.match(r"^([\d.]+)\s+dev\s+(\S+)\s+lladdr\s+([\da-f:]+)\s+(\S+)", line)
            if not m:
                continue
            entries.append(
                {
                    "interface": m.group(2),
                    "mac": m.group(3),
                    "ip": m.group(1),
                    "age": 0.0,
                }
            )
        return entries

    def get_config(
        self, retrieve: str = "all", full: bool = False, sanitized: bool = False
    ) -> dict[str, Any]:
        """Return minimal config representation (network interfaces only).

        ``valid_lft`` / ``preferred_lft`` fields from DHCP leases are stripped
        so the output is stable across polls and does not produce false-positive
        config-change events in the config-backup feature. ``veth*`` interfaces
        are stripped entirely — Docker creates/destroys them with a fresh index
        and random name on every container restart, which would otherwise flag
        a config change on nearly every poll of a Docker host.
        """
        import re

        running = self._send("ip addr show && ip route show")
        # Strip volatile DHCP lease timer fields — they decrement every poll
        running = re.sub(r"\s+valid_lft\s+\S+\s+preferred_lft\s+\S+", "", running)
        # Strip veth interface blocks (line + indented sub-lines) — ephemeral
        # Docker container network endpoints, not intentional host config
        running = re.sub(r"^\d+: veth\S*:.*\n(?:[ \t]+.*\n?)*", "", running, flags=re.MULTILINE)
        return {"running": running, "startup": "", "candidate": ""}

    # ------------------------------------------------------------------
    # Config management – not applicable for generic Linux
    # ------------------------------------------------------------------

    def load_merge_candidate(self, filename: str = None, config: str = None) -> None:  # type: ignore[override]
        raise NotImplementedError("Config management is not supported for Linux hosts")

    def load_replace_candidate(self, filename: str = None, config: str = None) -> None:  # type: ignore[override]
        raise NotImplementedError("Config management is not supported for Linux hosts")

    def compare_config(self) -> str:
        raise NotImplementedError("Config management is not supported for Linux hosts")

    def commit_config(self, message: str = "") -> None:
        raise NotImplementedError("Config management is not supported for Linux hosts")

    def discard_config(self) -> None:
        raise NotImplementedError("Config management is not supported for Linux hosts")

    def rollback(self) -> None:
        raise NotImplementedError("Config management is not supported for Linux hosts")

    # ------------------------------------------------------------------
    # Optional NAPALM methods
    # ------------------------------------------------------------------

    def ping(
        self,
        destination: str,
        source: str = "",
        ttl: int = 255,
        timeout: int = 2,
        size: int = 100,
        count: int = 5,
        vrf: str = "",
    ) -> dict[str, Any]:
        """Execute ping from the remote host."""
        src_opt = f"-I {source}" if source else ""
        cmd = f"ping -c {count} -W {timeout} -s {size} -t {ttl} {src_opt} {destination} 2>&1"
        output = self._send(cmd)

        # Parse summary line: "5 packets transmitted, 5 received, 0% packet loss"
        m = re.search(
            r"(\d+) packets transmitted,\s*(\d+) received,\s*([\d.]+)% packet loss",
            output,
        )
        if not m:
            return {"error": output}

        sent, received = int(m.group(1)), int(m.group(2))

        # Parse rtt line: "rtt min/avg/max/mdev = 0.123/0.456/0.789/0.100 ms"
        rtt_m = re.search(r"rtt .* = ([\d.]+)/([\d.]+)/([\d.]+)/([\d.]+) ms", output)

        results = []
        for line in output.splitlines():
            icmp_m = re.search(
                r"bytes from ([\d.]+).*icmp_seq=\d+ ttl=(\d+) time=([\d.]+) ms", line
            )
            if icmp_m:
                results.append(
                    {
                        "ip_address": icmp_m.group(1),
                        "rtt": float(icmp_m.group(3)),
                    }
                )

        return {
            "success": {
                "probes_sent": sent,
                "packet_loss": sent - received,
                "rtt_min": float(rtt_m.group(1)) if rtt_m else 0.0,
                "rtt_avg": float(rtt_m.group(2)) if rtt_m else 0.0,
                "rtt_max": float(rtt_m.group(3)) if rtt_m else 0.0,
                "rtt_stddev": float(rtt_m.group(4)) if rtt_m else 0.0,
                "results": results,
            }
        }

    # ------------------------------------------------------------------
    # KernelFactsMixin – the transport for get_kernel_facts
    # ------------------------------------------------------------------

    def _run_kernel_facts_command(self, command: str) -> str:
        """The transport for ``KernelFactsMixin.get_kernel_facts``: read-only, no sudo."""
        return self._send(command, read_timeout=60)

    def _run_service_packages_command(self, command: str) -> str:
        """The transport for ``ServicePackagesMixin.get_service_packages``.

        An exec channel rather than the PTY -- the command is about 1 kB -- under
        ``sh``, which it is written for, whatever the login shell. Read-only, so
        never through sudo.
        """
        return self.run_command(f"sh -c {_shlex_quote(command)}", timeout=120).stdout

    # ------------------------------------------------------------------
    # ListeningSocketsMixin – the transport for get_listening_sockets
    # ------------------------------------------------------------------

    def _run_listening_sockets_command(self, command: str, *, privileged: bool) -> str:
        """The transport for ``ListeningSocketsMixin.get_listening_sockets``.

        Read-only either way; root only so that ``ss`` names every process.
        """
        if privileged:
            return self._run_privileged(command, 60)
        return self._send(command, read_timeout=60)

    def _run_host_firewall_command(self, command: str, *, privileged: bool) -> str:
        """The transport for ``HostFirewallMixin.get_host_firewall``: an exec channel.

        ``nft -j`` prints the whole ruleset as one line of JSON, longer than the
        PTY netmiko holds reliably takes. Read-only; root because both nft and
        iptables-save need ``CAP_NET_ADMIN`` (NetOrk/netork#1344).
        """
        return self.run_command(command, privileged=privileged, timeout=60).stdout

    def _run_virtual_ips_command(self, command: str, *, privileged: bool) -> str:
        """The transport for ``VirtualIpsMixin.get_virtual_ips``: an exec channel.

        The command carries its awk program and is about 4 kB, past the line the
        PTY netmiko holds reliably takes. Read-only either way; root only because
        the configuration may be root's alone.
        """
        return self.run_command(command, privileged=privileged, timeout=60).stdout

    def _run_service_clusters_command(self, command: str, *, privileged: bool) -> str:
        """The transport for ``ServiceClustersMixin.get_service_clusters``: an exec
        channel, like the virtual IPs' command, which it resembles in size.
        Read-only either way; root only because Patroni's configuration is the
        postgres user's alone."""
        return self.run_command(command, privileged=privileged, timeout=60).stdout

    def _run_system_identity_command(self, command: str, *, privileged: bool) -> str:
        """The transport for ``SystemIdentityMixin.get_system_identity``: an exec channel,
        as root, since ``product_uuid`` and ``product_serial`` are root's."""
        return self.run_command(command, privileged=privileged, timeout=30).stdout

    # ------------------------------------------------------------------
    # OSDriver – package management
    # ------------------------------------------------------------------

    def get_packages(self) -> list[PackageDict]:
        if self._pkg_manager == "apt":
            return self._get_packages_apt()
        if self._pkg_manager in ("dnf", "yum"):
            return self._get_packages_rpm()
        if self._pkg_manager == "apk":
            return self._get_packages_apk()
        if self._pkg_manager == "pacman":
            return self._get_packages_pacman()
        raise NotImplementedError(f"Package manager '{self._pkg_manager}' is not supported")

    def _get_packages_apt(self) -> list[PackageDict]:
        out = self._send(
            "dpkg-query -W -f='${Package}\\t${Version}\\t${Installed-Size}"
            "\\t${source:Package}\\t${source:Version}\\t${binary:Summary}\\n' 2>/dev/null"
        )
        packages: list[PackageDict] = []
        for line in out.splitlines():
            # Summary stays last and keeps whatever it contains: maxsplit must
            # equal the number of tabs the format writes, not the field count.
            parts = line.split("\t", 5)
            if len(parts) < 2:
                continue
            name = parts[0].strip()
            version = parts[1].strip()
            size = (
                int(parts[2].strip()) * 1024 if len(parts) > 2 and parts[2].strip().isdigit() else 0
            )
            # Debian source package (e.g. openssh-server → openssh) for OSV matching.
            source_package = parts[3].strip() if len(parts) > 3 and parts[3].strip() else name
            # And its version, which is a different number from this package's.
            #
            # OSV states Debian ranges in *source* versions. A source package
            # that ships several binaries gives each its own upstream version:
            # libldb2 is 2:2.11.0+samba4.22.11+dfsg-… while its source, samba,
            # is 2:4.22.11+dfsg-…. A consumer matching on source_package and
            # comparing `version` compares two unrelated numbers — dpkg reads
            # ldb's 2.11.0 as older than the 2:4.17.4+dfsg-1 that fixed
            # CVE-2022-44640, and a host five releases past the fix was reported
            # vulnerable on four packages at once.
            #
            # dpkg leaves this empty when it equals `Version`; so does an older
            # dpkg that does not know the field at all.
            source_version = parts[4].strip() if len(parts) > 4 and parts[4].strip() else version
            description = parts[5].strip() if len(parts) > 5 else ""
            packages.append(
                {
                    "name": name,
                    "version": version,
                    "installed": True,
                    "description": description,
                    "size": size,
                    "source": "apt",
                    "source_package": source_package,
                    "source_version": source_version,
                }
            )
        return packages

    def _get_packages_rpm(self) -> list[PackageDict]:
        out = self._send(
            "rpm -qa --queryformat "
            "'%{NAME}\\t%{VERSION}-%{RELEASE}\\t%{SIZE}\\t%{SUMMARY}\\n' 2>/dev/null"
        )
        packages: list[PackageDict] = []
        for line in out.splitlines():
            parts = line.split("\t", 3)
            if len(parts) < 2:
                continue
            packages.append(
                {
                    "name": parts[0].strip(),
                    "version": parts[1].strip(),
                    "installed": True,
                    "description": parts[3].strip() if len(parts) > 3 else "",
                    "size": int(parts[2].strip())
                    if len(parts) > 2 and parts[2].strip().isdigit()
                    else 0,
                    "source": self._pkg_manager or "rpm",
                }
            )
        return packages

    def _get_packages_apk(self) -> list[PackageDict]:
        out = self._send("apk info -v 2>/dev/null")
        packages: list[PackageDict] = []
        for line in out.splitlines():
            # openssh-9.3_p2-r4 OpenSSH
            m = re.match(r"^(\S+)-(\d[\S]*)\s*(.*)", line)
            if not m:
                continue
            packages.append(
                {
                    "name": m.group(1),
                    "version": m.group(2),
                    "installed": True,
                    "description": m.group(3).strip(),
                    "size": 0,
                    "source": "apk",
                }
            )
        return packages

    def _get_packages_pacman(self) -> list[PackageDict]:
        out = self._send("pacman -Q 2>/dev/null")
        packages: list[PackageDict] = []
        for line in out.splitlines():
            parts = line.split(None, 1)
            if len(parts) < 2:
                continue
            packages.append(
                {
                    "name": parts[0],
                    "version": parts[1],
                    "installed": True,
                    "description": "",
                    "size": 0,
                    "source": "pacman",
                }
            )
        return packages

    def search_packages(self, query: str) -> list[dict[str, Any]]:
        """Search available (installable) packages matching *query*."""
        from shlex import quote as _q

        safe_q = _q(query)
        installed = {p["name"] for p in self.get_packages()}
        packages: list[dict[str, Any]] = []

        if self._pkg_manager == "apt":
            out = self._send(f"apt-cache search {safe_q} 2>/dev/null")
            versions: dict[str, str] = {}
            ver_out = self._send(
                f"apt-cache show {safe_q} 2>/dev/null | grep -E '^(Package|Version):' || true"
            )
            cur_pkg = ""
            for line in ver_out.splitlines():
                if line.startswith("Package:"):
                    cur_pkg = line.split(":", 1)[1].strip()
                elif line.startswith("Version:") and cur_pkg:
                    versions[cur_pkg] = line.split(":", 1)[1].strip()
            for line in out.splitlines():
                if " - " not in line:
                    continue
                name, _, description = line.partition(" - ")
                name = name.strip()
                packages.append(
                    {
                        "name": name,
                        "version": versions.get(name, ""),
                        "installed": name in installed,
                        "description": description.strip(),
                        "size": 0,
                        "source": "apt",
                    }
                )

        elif self._pkg_manager in ("dnf", "yum"):
            cmd = "dnf" if self._pkg_manager == "dnf" else "yum"
            out = self._send(f"{cmd} search {safe_q} 2>/dev/null || true")
            for line in out.splitlines():
                if " : " not in line:
                    continue
                pkg_ver, _, description = line.partition(" : ")
                name = pkg_ver.split(".")[0].strip()
                version = ""
                packages.append(
                    {
                        "name": name,
                        "version": version,
                        "installed": name in installed,
                        "description": description.strip(),
                        "size": 0,
                        "source": self._pkg_manager or "rpm",
                    }
                )

        elif self._pkg_manager == "apk":
            out = self._send(f"apk search {safe_q} 2>/dev/null")
            for line in out.splitlines():
                line = line.strip()
                if not line:
                    continue
                m = re.match(r"^(.*?)-(\d\S*)(?:\s+(.*))?$", line)
                if m:
                    name, version, description = m.group(1), m.group(2), (m.group(3) or "")
                else:
                    name, version, description = line, "", ""
                packages.append(
                    {
                        "name": name,
                        "version": version,
                        "installed": name in installed,
                        "description": description,
                        "size": 0,
                        "source": "apk",
                    }
                )

        elif self._pkg_manager == "pacman":
            out = self._send(f"pacman -Ss {safe_q} 2>/dev/null || true")
            lines = out.splitlines()
            i = 0
            while i < len(lines):
                line = lines[i].strip()
                if "/" in line and " " in line:
                    parts = line.split()
                    name_ver = parts[0].split("/")[-1] if "/" in parts[0] else parts[0]
                    name_parts = name_ver.rsplit(" ", 1)
                    name = name_parts[0]
                    version = parts[1] if len(parts) > 1 else ""
                    description = lines[i + 1].strip() if i + 1 < len(lines) else ""
                    packages.append(
                        {
                            "name": name,
                            "version": version,
                            "installed": name in installed,
                            "description": description,
                            "size": 0,
                            "source": "pacman",
                        }
                    )
                    i += 2
                    continue
                i += 1

        return packages

    def install_package(self, name: str) -> dict[str, Any]:
        """Install a package by name. Returns ``{"success": bool, "output": str}``."""
        from shlex import quote as _q

        safe = _q(name)
        pm = self._pkg_manager
        if pm == "apt":
            raw = self._sudo(
                f"DEBIAN_FRONTEND=noninteractive apt-get install -y {safe} 2>&1 || true"
            )
        elif pm in ("dnf", "yum"):
            raw = self._sudo(f"{pm} install -y {safe} 2>&1 || true")
        elif pm == "apk":
            raw = self._sudo(f"apk add {safe} 2>&1 || true")
        elif pm == "pacman":
            raw = self._sudo(f"pacman -S --noconfirm {safe} 2>&1 || true")
        else:
            return {"success": False, "output": f"Unsupported package manager: {pm}"}
        low = raw.lower()
        success = not any(
            kw in low
            for kw in (
                "error:",
                "failed",
                "no packages",
                "not found",
                "unable to locate",
                "no match",
            )
        )
        return {"success": success, "output": raw.strip()}

    #: Words in a package manager's output that mean it did not do the job.
    #: Only consulted when the exit status is unknown; see ``_uninstall_failed``.
    _UNINSTALL_FAILED = ("error:", "failed", "not found", "is not installed", "no packages")

    def uninstall_package(self, name: str, purge: bool = False) -> dict[str, Any]:
        """Remove a package by name. Returns ``{"success": bool, "output": str}``.

        ``purge`` also removes the package's configuration where the package
        manager distinguishes the two. Off by default: configuration somebody
        may want back is not this function's to delete unless it was asked for.

        It matters for more than tidiness. A package's apt source survives a
        plain ``remove``, so the repository keeps being fetched on every
        ``apt-get update`` long after the package itself is gone — which is what
        the Wazuh agent left behind on thirteen hosts.

        **The dpkg fallback.** A package whose ``postinst`` failed sits at
        ``install ok unpacked``, and apt cannot remove it: it configures a
        package before removing it, and configuring is precisely what is broken.
        Seven of those thirteen hosts were in that state after an upgrade whose
        postinst could not reach a manager that had been decommissioned, and on
        one of them only ``dpkg --purge --force-all`` got it out.

        So the fallback runs **only after apt has failed**, never as a routine
        second step: forcing dpkg past its own consistency checks is a bigger
        hammer than apt, and a caller who reaches for it every time will
        eventually break something apt would have refused to.
        """
        from shlex import quote as _q

        safe = _q(name)
        pm = self._pkg_manager
        if pm == "apt":
            action = "purge" if purge else "remove"
            cmd = f"DEBIAN_FRONTEND=noninteractive apt-get {action} -y {safe} 2>&1"
        elif pm in ("dnf", "yum"):
            cmd = f"{pm} remove -y {safe} 2>&1"
        elif pm == "apk":
            # apk and pacman have no separate purge; asking for one is not an
            # error, it simply has nothing extra to do.
            cmd = f"apk del {safe} 2>&1"
        elif pm == "pacman":
            cmd = f"pacman -R --noconfirm {safe} 2>&1"
        else:
            return {"success": False, "output": f"Unsupported package manager: {pm}"}

        raw, rc = self._sudo_status(cmd)
        failed = self._uninstall_failed(raw, rc)

        if failed and pm == "apt":
            forced, forced_rc = self._sudo_status(f"dpkg --purge --force-all {safe} 2>&1")
            raw = f"{raw.strip()}\n--- dpkg --purge --force-all ---\n{forced.strip()}"
            failed = self._uninstall_failed(forced, forced_rc)

        return {"success": not failed, "output": raw.strip()}

    def _uninstall_failed(self, output: str, rc: int | None = None) -> bool:
        """Whether the package manager did not do the job.

        The exit status decides whenever there is one (netork#267): it is the
        answer the package manager actually gives, where the output is prose
        that every tool phrases differently. A prerm printing "Failed to stop
        …" while the removal completes is a success; a non-zero exit with
        nothing alarming in the output is not.

        Only when the status is unknown (``rc is None``) is the output read,
        as the best answer left. apt prefixes its own errors with ``E: `` at
        the start of a line, and the commonest of them — ``E: Sub-process
        /usr/bin/dpkg returned an error code (1)`` — contains neither "error:"
        nor "failed". The keyword list alone therefore read a failed removal
        as a success, which is the worst direction for this particular answer
        to be wrong in.

        Matched at line start rather than anywhere: "note: " ends in "e: ".
        """
        if rc is not None:
            return rc != 0
        low = output.lower()
        if any(line.lstrip().startswith("e: ") for line in low.splitlines()):
            return True
        return any(kw in low for kw in self._UNINSTALL_FAILED)

    def get_available_updates(self) -> list[UpdateDict]:
        if self._pkg_manager == "apt":
            return self._get_updates_apt()
        if self._pkg_manager in ("dnf", "yum"):
            return self._get_updates_rpm()
        if self._pkg_manager == "apk":
            return self._get_updates_apk()
        if self._pkg_manager == "pacman":
            return self._get_updates_pacman()
        raise NotImplementedError(f"Package manager '{self._pkg_manager}' is not supported")

    def _get_updates_apt(self) -> list[UpdateDict]:
        # apt list --upgradable does not need root; avoid sudo so it works even
        # without a configured sudo password.
        # Raises ValueError when apt failed or the output was cut short.
        return parse_apt_upgradable(self._send(APT_UPGRADABLE_COMMAND, read_timeout=60))

    def _get_updates_rpm(self) -> list[UpdateDict]:
        """dnf/yum check-update: exit 100 means updates, 0 none, anything else failed."""
        cmd = (
            "dnf check-update --quiet 2>/dev/null"
            if self._pkg_manager == "dnf"
            else "yum check-update -q 2>/dev/null"
        )
        output, status = _split_status(self._run_privileged(f"{cmd}; echo {_RC_MARKER}$?", 120))
        if status not in (0, 100):
            raise RuntimeError(
                f"{self._pkg_manager} check-update failed (exit {status}): {output[-200:]}"
            )
        security = self._rpm_security_names()
        updates: list[UpdateDict] = []
        for line in output.splitlines():
            parts = line.split()
            if len(parts) >= 2 and not line.startswith(" ") and "." in parts[0]:
                name = parts[0].rsplit(".", 1)[0]
                updates.append(
                    {
                        "name": name,
                        "current_version": "",
                        "new_version": parts[1],
                        "origin": parts[2] if len(parts) >= 3 else None,
                        "security": None if security is None else name in security,
                    }
                )
        return updates

    def _rpm_security_names(self) -> set[str] | None:
        """Packages a pending security advisory covers; None when dnf/yum cannot say."""
        cmd = DNF_SECURITY_COMMAND if self._pkg_manager == "dnf" else _YUM_SECURITY_COMMAND
        output, status = _split_status(self._run_privileged(f"{cmd}; echo {_RC_MARKER}$?", 120))
        # Both commands pipe through cat, so the outer status is cat's: dnf/yum's own
        # is in __DNF_RC. A failed run prints nothing on stdout and must not read as
        # "no advisories"; neither may output cut short before the marker.
        rc = _DNF_RC_RE.findall(output)
        if status != 0 or not rc or rc[-1] != "0":
            return None
        try:
            return parse_dnf_security(output)
        except ValueError:
            return None

    def refresh_available_updates(self) -> dict[str, Any]:
        """Refresh the package index (apt-get update, dnf makecache, apk update)."""
        cmd = _REFRESH.get(self._pkg_manager or "")
        if cmd is None:
            return {
                "success": False,
                "output": f"Refreshing the index is not supported for {self._pkg_manager!r}",
            }
        output, status = _split_status(self._run_privileged(f"{cmd}; echo {_RC_MARKER}$?", 180))
        if status != 0 and "password is required" in output:
            output = f"{output}\n{_SUDO_PASSWORD_HINT}"
        return {"success": status == 0, "output": output}

    def _get_updates_apk(self) -> list[UpdateDict]:
        out = self._send("apk version -l '<' 2>/dev/null")
        updates: list[UpdateDict] = []
        for line in out.splitlines():
            # openssh-9.3_p2-r3 < 9.3_p2-r4
            m = re.match(r"^(\S+)-(\S+)\s+<\s+(\S+)", line)
            if m:
                updates.append(
                    {
                        "name": m.group(1),
                        "current_version": m.group(2),
                        "new_version": m.group(3),
                    }
                )
        return updates

    def _get_updates_pacman(self) -> list[UpdateDict]:
        out = self._send("pacman -Qu 2>/dev/null")
        updates: list[UpdateDict] = []
        for line in out.splitlines():
            # openssh 9.3p2-1 -> 9.4p1-1
            m = re.match(r"^(\S+)\s+(\S+)\s+->\s+(\S+)", line)
            if m:
                updates.append(
                    {
                        "name": m.group(1),
                        "current_version": m.group(2),
                        "new_version": m.group(3),
                    }
                )
        return updates

    # ------------------------------------------------------------------
    # OSDriver – apply updates
    # ------------------------------------------------------------------

    # Allowlist for package names – same pattern used by napalm-proxmox
    _PKG_NAME_RE = re.compile(r"^[a-zA-Z0-9_\-\+\.]+$")

    def apply_updates(self, packages: list[str]) -> ApplyUpdatesResultDict:
        """Upgrade *packages* (or all pending updates when the list is empty).

        Package names are validated against ``^[a-zA-Z0-9_\\-\\+\\.]+$`` before
        being passed to the package manager to prevent shell injection.
        """
        for pkg in packages:
            if not self._PKG_NAME_RE.match(pkg):
                raise ValueError(f"Invalid package name: {pkg!r}")

        if self._pkg_manager == "apt":
            return self._apply_updates_apt(packages)
        if self._pkg_manager in ("dnf", "yum"):
            return self._apply_updates_rpm(packages)
        if self._pkg_manager == "apk":
            return self._apply_updates_apk(packages)
        if self._pkg_manager == "pacman":
            return self._apply_updates_pacman(packages)
        raise NotImplementedError(f"Package manager '{self._pkg_manager}' is not supported")

    def _apply_updates_apt(self, packages: list[str]) -> ApplyUpdatesResultDict:
        pkg_args = " ".join(packages) if packages else "--with-new-pkgs"
        cmd = (
            f"DEBIAN_FRONTEND=noninteractive apt-get install --only-upgrade -y {pkg_args} 2>&1"
            if packages
            else "DEBIAN_FRONTEND=noninteractive apt-get upgrade -y 2>&1"
        )
        try:
            output = self._sudo(cmd, read_timeout=600)
            success = not re.search(r"^E:", output, re.MULTILINE)
            result: ApplyUpdatesResultDict = {"success": success, "output": output}
            if not success:
                m = re.search(r"^E:.*", output, re.MULTILINE)
                result["error"] = m.group(0) if m else "apt-get exited with errors"
            return result
        except Exception as exc:
            return {"success": False, "output": "", "error": str(exc)}

    def _apply_updates_rpm(self, packages: list[str]) -> ApplyUpdatesResultDict:
        bin_ = self._pkg_manager  # "dnf" or "yum"
        if packages:
            pkg_args = " ".join(packages)
            cmd = f"{bin_} upgrade -y {pkg_args} 2>&1"
        else:
            cmd = f"{bin_} upgrade -y 2>&1"
        try:
            output = self._sudo(cmd, read_timeout=600)
            # dnf/yum signal failure via "Error:" lines or non-zero exit;
            # since we can't check the exit code directly, look for error markers.
            success = not re.search(r"^Error:", output, re.MULTILINE | re.IGNORECASE)
            result: ApplyUpdatesResultDict = {"success": success, "output": output}
            if not success:
                m = re.search(r"^Error:.*", output, re.MULTILINE | re.IGNORECASE)
                result["error"] = m.group(0) if m else f"{bin_} exited with errors"
            return result
        except Exception as exc:
            return {"success": False, "output": "", "error": str(exc)}

    def _apply_updates_apk(self, packages: list[str]) -> ApplyUpdatesResultDict:
        if packages:
            pkg_args = " ".join(packages)
            cmd = f"apk upgrade {pkg_args} 2>&1"
        else:
            cmd = "apk upgrade 2>&1"
        try:
            output = self._sudo(cmd, read_timeout=300)
            success = "ERROR" not in output.upper().split("\n")[0] if output else True
            result: ApplyUpdatesResultDict = {"success": success, "output": output}
            if not success:
                result["error"] = "apk upgrade reported an error"
            return result
        except Exception as exc:
            return {"success": False, "output": "", "error": str(exc)}

    def _apply_updates_pacman(self, packages: list[str]) -> ApplyUpdatesResultDict:
        if packages:
            pkg_args = " ".join(packages)
            cmd = f"pacman --noconfirm -S {pkg_args} 2>&1"
        else:
            cmd = "pacman --noconfirm -Syu 2>&1"
        try:
            output = self._sudo(cmd, read_timeout=300)
            success = "error" not in output.lower()
            result: ApplyUpdatesResultDict = {"success": success, "output": output}
            if not success:
                result["error"] = "pacman reported an error"
            return result
        except Exception as exc:
            return {"success": False, "output": "", "error": str(exc)}

    # ------------------------------------------------------------------
    # OSDriver – services (systemd, through SystemdServicesMixin)
    # ------------------------------------------------------------------

    def get_services(self) -> list[ServiceDict]:
        """systemd's services in one round trip; ``service --status-all`` without systemd."""
        try:
            return super().get_services()
        except SystemdUnavailable:
            return self._get_services_sysv()

    def manage_service(self, name: str, action: str) -> dict[str, Any]:
        """Start, stop, restart, enable or disable a systemd service.

        :raises ValueError: for an unknown action or an invalid name.
        """
        result = super().manage_service(name, action)
        if not result["success"] and "password is required" in result["output"]:
            result["output"] = f"{result['output']}\n{_SUDO_PASSWORD_HINT}"
        return result

    def _get_services_sysv(self) -> list[ServiceDict]:
        out = self._send("service --status-all 2>/dev/null")
        services: list[ServiceDict] = []
        for line in out.splitlines():
            m = re.match(r"^\s*\[\s*([+\-?])\s*\]\s+(\S+)", line)
            if not m:
                continue
            services.append(
                {
                    "name": m.group(2),
                    "running": m.group(1) == "+",
                    "enabled": False,
                    "pid": 0,
                }
            )
        return services

    # ------------------------------------------------------------------
    # OSDriver – users
    # ------------------------------------------------------------------

    def get_os_users(self) -> list[UserDict]:
        """Return local user accounts from /etc/passwd plus supplementary groups."""
        passwd_out = self._send("getent passwd 2>/dev/null || cat /etc/passwd")
        groups_out = self._send("getent group 2>/dev/null || cat /etc/group")

        # username → supplementary groups, joined with passwd below
        username_to_groups: dict[str, list[str]] = {}
        for line in groups_out.splitlines():
            parts = line.split(":")
            if len(parts) < 4:
                continue
            gname = parts[0]
            members = [m.strip() for m in parts[3].split(",") if m.strip()]
            for member in members:
                username_to_groups.setdefault(member, []).append(gname)

        users: list[UserDict] = []
        for line in passwd_out.splitlines():
            parts = line.split(":")
            if len(parts) < 7:
                continue
            username, _, uid_s, gid_s, _, home, shell = parts[:7]
            try:
                uid, gid = int(uid_s), int(gid_s)
            except ValueError:
                continue
            users.append(
                {
                    "username": username,
                    "uid": uid,
                    "gid": gid,
                    "home": home,
                    "shell": shell,
                    "groups": username_to_groups.get(username, []),
                }
            )
        return users

    # ------------------------------------------------------------------
    # OSDriver – processes
    # ------------------------------------------------------------------

    def get_processes(self) -> list[ProcessDict]:
        """Return running processes via ``ps axo``."""
        out = self._send(
            "ps axo pid,ppid,user:20,pcpu,pmem,vsz,rss,tty,stat,lstart,args "
            "--no-headers 2>/dev/null"
        )
        processes: list[ProcessDict] = []
        for line in out.splitlines():
            parts = line.split(None, 10)
            if len(parts) < 11:
                continue
            try:
                pid = int(parts[0])
                ppid = int(parts[1])
                user = parts[2]
                cpu = float(parts[3])
                mem = float(parts[4])
                vsz = int(parts[5])
                rss = int(parts[6])
                tty = parts[7] if parts[7] != "?" else ""
                state = parts[8][0] if parts[8] else "?"
                # lstart is 5 tokens: "Mon May 27 12:34:56 2024" → parts[9..13]
                # args starts at parts[14] but we merged from 10 onward
                # With --no-headers and ps axo, lstart takes 5 parts
                # Rebuild: parts[9] is start, args is parts[10]
                started = parts[9]
                command = parts[10]
            except (ValueError, IndexError):
                continue
            processes.append(
                {
                    "pid": pid,
                    "ppid": ppid,
                    "user": user,
                    "cpu": cpu,
                    "memory": mem,
                    "vsz": vsz,
                    "rss": rss,
                    "tty": tty,
                    "state": state,
                    "started": started,
                    "command": command,
                }
            )
        return processes

    # ------------------------------------------------------------------
    # OSDriver – cron jobs
    # ------------------------------------------------------------------

    def get_cron_jobs(self) -> list[CronJobDict]:
        """Return cron entries from user crontabs and /etc/cron.d."""
        jobs: list[CronJobDict] = []

        # /etc/cron.d/* — system-wide cron fragments (include user field)
        cron_d_files = self._send("ls /etc/cron.d/ 2>/dev/null").splitlines()
        for fname in cron_d_files:
            fname = fname.strip()
            if not fname:
                continue
            content = self._send(f"cat /etc/cron.d/{fname} 2>/dev/null")
            for line in content.splitlines():
                job = self._parse_cron_line(line, source_user="root", has_user_field=True)
                if job:
                    jobs.append(job)

        # Per-user crontabs from /var/spool/cron/crontabs (Debian) or /var/spool/cron (RHEL)
        for spool_dir in ("/var/spool/cron/crontabs", "/var/spool/cron"):
            ls_out = self._send(f"ls {spool_dir} 2>/dev/null")
            for uname in ls_out.splitlines():
                uname = uname.strip()
                if not uname:
                    continue
                content = self._send(f"cat {spool_dir}/{uname} 2>/dev/null")
                for line in content.splitlines():
                    job = self._parse_cron_line(line, source_user=uname, has_user_field=False)
                    if job:
                        jobs.append(job)

        return jobs

    @staticmethod
    def _parse_cron_line(line: str, source_user: str, has_user_field: bool) -> CronJobDict | None:
        """Parse a single crontab line; returns ``None`` for comments/blanks."""
        stripped = line.strip()
        # Remove trailing comment
        comment = ""
        if "#" in stripped:
            idx = stripped.index("#")
            comment = stripped[idx + 1 :].strip()
            stripped = stripped[:idx].strip()

        if not stripped or stripped.startswith("@") or stripped.startswith("MAILTO"):
            return None

        parts = stripped.split(None, 6 if has_user_field else 5)
        expected = 6 if has_user_field else 5
        if len(parts) < expected:
            return None

        schedule = " ".join(parts[:5])
        if has_user_field:
            user = parts[5]
            command = parts[6] if len(parts) > 6 else ""
        else:
            user = source_user
            command = parts[5] if len(parts) > 5 else ""

        job: CronJobDict = {
            "user": user,
            "schedule": schedule,
            "command": command,
        }
        if comment:
            job["description"] = comment
        return job

    # ── Device actions ────────────────────────────────────────────────────────

    def get_snmp_config(self) -> SNMPConfigDict | None:
        """Return SNMP agent config if snmpd is installed and running."""
        try:
            running = (
                self._send("systemctl is-active snmpd 2>/dev/null || true").strip() == "active"
            )
            if not running:
                return None

            # Parse community string from snmpd.conf
            community = "public"
            port = 161
            try:
                conf = self._send(
                    "grep -E '^[[:space:]]*(ro|rw)?community' /etc/snmp/snmpd.conf 2>/dev/null"
                    " | head -5"
                )
                for line in conf.splitlines():
                    parts = line.split()
                    if not parts:
                        continue
                    kw = parts[0].lower()
                    if kw in ("rocommunity", "rwcommunity", "rocommunity6", "rwcommunity6"):
                        if len(parts) >= 2:
                            community = parts[1]
                            break
                    elif kw == "com2sec" and len(parts) >= 4:
                        # com2sec notConfigUser  default   <community>
                        community = parts[3]
                        break
            except Exception:
                logger.debug("%s: ignored", self.hostname, exc_info=True)

            # Detect port override
            try:
                port_line = self._send(
                    "grep -E '^agentAddress' /etc/snmp/snmpd.conf 2>/dev/null | head -1"
                ).strip()
                if port_line:
                    m = re.search(r":(\d+)", port_line)
                    if m:
                        port = int(m.group(1))
            except Exception:
                logger.debug("%s: ignored", self.hostname, exc_info=True)

            return SNMPConfigDict(running=True, community=community, port=port, version="2c")
        except Exception as exc:
            logger.debug("get_snmp_config() failed: %s", exc)
            return None

    # ------------------------------------------------------------------
    # Fix templates (napalm-device-types ``fixes``): the caller supplies every value
    # ------------------------------------------------------------------

    def _root_outcome(self, command: str, timeout: float, done: str) -> FixStepOutcomeDict:
        """Run *command* as root and report it as a step: its exit status decides."""
        output, status = self._sudo_status(command, timeout)
        if status != 0 and "password is required" in output:
            output = f"{output}\n{_SUDO_PASSWORD_HINT}"
        return {"ok": status == 0, "output": (output.strip()[-300:] if status != 0 else done)}

    def get_management_peer(self) -> str | None:
        """The address this host sees the SSH connection come from (``SSH_CONNECTION``)."""
        out = self.run_command('printf "%s\\n" "$SSH_CONNECTION"', timeout=15).stdout.split()
        try:
            return str(ipaddress.ip_address(out[0])) if out else None
        except ValueError:
            return None

    def _install_snmp_agent(self) -> FixStepOutcomeDict:
        pm = self._pkg_manager or self._detect_pkg_manager()
        command = _SNMP_INSTALL.get(pm or "")
        if not command:
            return {"ok": False, "output": f"no supported package manager ({pm or 'none found'})"}
        return self._root_outcome(command, 300, f"net-snmp installed ({pm})")

    def _write_snmp_agent_config(self, settings: SnmpAgentSettingsDict) -> FixStepOutcomeDict:
        written = self._install_file_as_root(_SNMPD_CONF, render_snmpd_conf(settings))
        if written.exit_code:
            return {"ok": False, "output": (written.stdout or written.stderr).strip()[-300:]}
        return {"ok": True, "output": f"wrote {_SNMPD_CONF}"}

    def _allow_snmp_sources(self, settings: SnmpAgentSettingsDict) -> FixStepOutcomeDict:
        sources = [_source(s) for s in settings.get("allowed_sources", [])]
        if not sources:
            return {"ok": True, "output": "no sources given; firewall unchanged"}
        if self._send("command -v ufw 2>/dev/null"):
            rules = [f"ufw allow from {s} to any port 161 proto udp" for s in sources]
        elif self._send("command -v iptables 2>/dev/null"):
            rules = [
                f"{{ iptables -C INPUT -s {s} -p udp --dport 161 -j ACCEPT 2>/dev/null"
                f" || iptables -I INPUT -s {s} -p udp --dport 161 -j ACCEPT; }}"
                for s in sources
            ]
        else:
            return {"ok": True, "output": "no host firewall (ufw, iptables); nothing to open"}
        return self._root_outcome(" && ".join(rules), 30, f"udp/161 open for {', '.join(sources)}")

    def _start_snmp_agent(self) -> FixStepOutcomeDict:
        return self._root_outcome(
            "systemctl enable snmpd && systemctl restart snmpd && systemctl is-active snmpd",
            60,
            "snmpd enabled and running",
        )

    # apt's proxy (napalm-device-types AptProxyMixin): the transports

    def _is_apt_host(self) -> bool:
        return self._pkg_manager == "apt"

    def _run_apt_command(self, command: str) -> str:
        """On the exec channel: no prompt to mistake a line of apt's output for."""
        return self.run_command(command, timeout=60).stdout

    def _run_apt_root_command(self, command: str, done: str) -> FixStepOutcomeDict:
        return self._root_outcome(command, 30, done)

    def _write_apt_file(self, path: str, content: str) -> FixStepOutcomeDict:
        written = self._install_file_as_root(path, content)
        if written.exit_code:
            return {"ok": False, "output": (written.stdout or written.stderr).strip()[-300:]}
        return {"ok": True, "output": f"wrote {path}"}

    def grant_container_engine_access(self, user: str) -> FixResultDict:
        """Add *user* to the ``docker`` group, so the engine answers without sudo."""

        def grant() -> FixStepOutcomeDict:
            if not _USER_NAME.match(user):
                raise ValueError(f"not a user name: {user!r}")
            return self._root_outcome(
                f"usermod -aG docker {user}",
                30,
                f"{user} is in the docker group from the next login",
            )

        return run_fix_steps([("grant", grant)])

    def get_vpn_tunnels(self) -> dict[str, Any]:
        """Return WireGuard status via ``wg show all dump`` (requires root/sudo).

        Falls back to interface-level data from ``ip link`` + ``/proc/net/dev``
        when root access is unavailable.

        Full data keyed by ``wireguard-<iface>-<pubkey[:8]>`` (one entry per peer).
        Fallback keyed by ``wireguard-<iface>`` (one entry per WireGuard interface).
        """
        import time as _time

        tunnels: dict[str, Any] = {}

        # ── Attempt 1: wg show all dump as root ──────────────────────────────
        # On the exec channel, which keeps the tabs the PTY expands to spaces.
        # It went through /tmp/.netork_wg_<hash> before: a predictable name,
        # holding the interfaces' private keys (#1039).
        raw = self._sudo("wg show all dump 2>&1 || true")

        _perm_errors = (
            "operation not permitted",
            "permission denied",
            "a password is required",
            "a terminal is required",
            "command not found",
            "not found",
            "no such file",
        )
        has_peer_data = raw and not any(e in raw.lower() for e in _perm_errors)

        if has_peer_data:
            # Build listen-port map from interface lines (5 tab-separated fields)
            listen_ports: dict[str, str] = {}
            for line in raw.splitlines():
                parts = line.split("\t")
                if len(parts) == 5:
                    iface, _priv, _pub, port, _fwmark = parts
                    listen_ports[iface.strip()] = port.strip()

            now = int(_time.time())
            for line in raw.splitlines():
                parts = line.split("\t")
                if len(parts) != 9:
                    continue
                (
                    iface,
                    pubkey,
                    _psk,
                    endpoint,
                    allowed_ips,
                    latest_hs,
                    rx_bytes,
                    tx_bytes,
                    _keepalive,
                ) = parts

                iface = iface.strip()
                pubkey = pubkey.strip()
                endpoint = endpoint.strip()

                remote_ip = ""
                if endpoint and endpoint != "(none)":
                    remote_ip = endpoint.rsplit(":", 1)[0].strip("[]")

                try:
                    hs_ts = int(latest_hs)
                except (ValueError, TypeError):
                    hs_ts = 0

                # "up" if last handshake within 3 min (WireGuard re-handshake every 2 min)
                is_up = hs_ts > 0 and (now - hs_ts) < 180

                try:
                    bytes_in = int(rx_bytes)
                except (ValueError, TypeError):
                    bytes_in = 0
                try:
                    bytes_out = int(tx_bytes)
                except (ValueError, TypeError):
                    bytes_out = 0

                local_port = listen_ports.get(iface, "")
                local_ep = f":{local_port}" if local_port and local_port != "0" else ""

                key = f"wireguard-{iface}-{pubkey[:8]}"
                tunnels[key] = {
                    "type": "WireGuard",
                    "local_endpoint": local_ep,
                    "remote_endpoint": remote_ip,
                    "is_up": is_up,
                    "uptime": (now - hs_ts) if hs_ts > 0 else 0,
                    "bytes_in": bytes_in,
                    "bytes_out": bytes_out,
                    "description": f"{iface} — peer {pubkey[:16]}…",
                    "public_key": pubkey,
                    "allowed_ips": allowed_ips.strip(),
                    "interface": iface,
                }
            return tunnels

        # ── Fallback: interface-level data without root ──────────────────────
        # ip -j link show type wireguard → list of WireGuard interface objects
        ip_raw = self._send("ip -j link show type wireguard 2>/dev/null || true").strip()
        if not ip_raw or ip_raw.startswith("[") is False:
            # Try stripping shell noise before the JSON
            start = ip_raw.find("[")
            ip_raw = ip_raw[start:] if start != -1 else ""

        if not ip_raw:
            return tunnels

        try:
            import json as _json

            iface_list = _json.loads(ip_raw)
        except Exception:
            return tunnels

        # /proc/net/dev for total RX/TX bytes per interface
        proc_dev = self._send("cat /proc/net/dev 2>/dev/null || true")
        proc_bytes: dict[str, tuple] = {}
        for line in proc_dev.splitlines()[2:]:
            line = line.strip()
            if ":" not in line:
                continue
            iface_name, rest = line.split(":", 1)
            fields = rest.split()
            try:
                proc_bytes[iface_name.strip()] = (int(fields[0]), int(fields[8]))
            except (IndexError, ValueError):
                pass

        for iface_obj in iface_list:
            iface = iface_obj.get("ifname", "")
            if not iface:
                continue
            flags = iface_obj.get("flags", [])
            is_up = "UP" in flags and "LOWER_UP" in flags
            rx, tx = proc_bytes.get(iface, (0, 0))
            tunnels[f"wireguard-{iface}"] = {
                "type": "WireGuard",
                "local_endpoint": "",
                "remote_endpoint": "",
                "is_up": is_up,
                "uptime": 0,
                "bytes_in": rx,
                "bytes_out": tx,
                "description": f"{iface} (peer data requires root/sudo)",
                "interface": iface,
            }

        return tunnels