feat: read the host's listening sockets through napalm-device-types

LinuxDriver mixes in ListeningSocketsMixin (napalm-device-types 2.4.0) and
supplies its transport: privileged readings go through _run_privileged --
as they are for a root login, through sudo with a password, sudo -n
without one -- and the rest through _send. The command arrives as one
sh -c argument, so sudo covers the whole script; when sudo refuses, the
mixin reads again without it.

OpenMediaVault and QNAP QTS inherit it; a host without ss raises
ListeningSocketsUnavailable.

For netOrk#658.

napalm_linux/linux.py

@@ -37,6 +37,7 @@ from napalm_device_types import (
     FingerprintRule,
     HostStatusMixin,
     KernelFactsMixin,
+    ListeningSocketsMixin,
     OSDriver,
     SystemdServicesMixin,
     SystemdUnavailable,
@@ -214,7 +215,9 @@ def _short_image_id(raw: str) -> str:
     return raw.strip().removeprefix("sha256:")[:12]
 
 
-class LinuxDriver(KernelFactsMixin, SystemdServicesMixin, HostStatusMixin, OSDriver):
+class LinuxDriver(
+    KernelFactsMixin, ListeningSocketsMixin, SystemdServicesMixin, HostStatusMixin, OSDriver
+):
     """NAPALM driver for generic Linux systems.
 
     Connects via SSH (netmiko ``linux`` device type) and auto-detects the
@@ -969,6 +972,19 @@ class LinuxDriver(KernelFactsMixin, SystemdServicesMixin, HostStatusMixin, OSDri
         """The transport for ``KernelFactsMixin.get_kernel_facts``: read-only, no sudo."""
         return self._send(command, read_timeout=60)
 
+    # ------------------------------------------------------------------
+    # ListeningSocketsMixin – the transport for get_listening_sockets
+    # ------------------------------------------------------------------
+
+    def _run_listening_sockets_command(self, command: str, *, privileged: bool) -> str:
+        """The transport for ``ListeningSocketsMixin.get_listening_sockets``.
+
+        Read-only either way; root only so that ``ss`` names every process.
+        """
+        if privileged:
+            return self._run_privileged(command, 60)
+        return self._send(command, read_timeout=60)
+
     # ------------------------------------------------------------------
     # OSDriver – package management
     # ------------------------------------------------------------------

pyproject.toml

@@ -37,7 +37,7 @@ classifiers = [
 ]
 dependencies = [
     "napalm>=4.0",
-    "napalm-device-types>=2.3.0",
+    "napalm-device-types>=2.4.0",
     "netmiko>=4.0.0",
     "paramiko>=5.0.0",  # CVE-2026-44405
 ]

tests/test_linux.py

@@ -1555,3 +1555,65 @@ class TestRebootHost:
 
         with pytest.raises(RuntimeError):
             driver.reboot_host()
+
+
+# ---------------------------------------------------------------------------
+# Listening sockets: ss and the processes' cgroups, as root when it can
+# ---------------------------------------------------------------------------
+
+_SOCKETS = (
+    "SOCK_BEGIN\n[ss]\n"
+    'tcp LISTEN 0 128 0.0.0.0:5432 0.0.0.0:* users:(("postgres",pid=812,fd=6))\n'
+    "__SS_RC=0\n[cgroups]\n"
+    "812 0::/system.slice/system-postgresql.slice/postgresql@16-main.service\n"
+    "SOCK_END\n"
+)
+
+
+class TestGetListeningSockets:
+    def _sent(self, driver) -> list[str]:
+        return [c[0][0] for c in driver._device.send_command.call_args_list]
+
+    def test_it_reads_with_the_shared_command(self, driver):
+        from napalm_device_types import ListeningSocketsMixin
+
+        assert isinstance(driver, ListeningSocketsMixin)
+        driver._root = True
+        driver._device.send_command.return_value = _SOCKETS
+
+        reading = driver.get_listening_sockets()
+
+        assert reading["attributed"] is True
+        [socket] = reading["sockets"]
+        assert (socket["port"], socket["unit"]) == (5432, "postgresql@16-main")
+        assert self._sent(driver)[0].startswith("sh -c '")
+
+    def test_without_a_sudo_password_the_whole_script_runs_under_sudo_n(self, driver):
+        driver._root = False
+        driver._device.send_command.return_value = _SOCKETS
+
+        driver.get_listening_sockets()
+
+        assert self._sent(driver)[0].startswith("sudo -n sh -c '")
+
+    def test_with_a_sudo_password_it_goes_through_sudo(self, driver):
+        driver._root = False
+        driver._sudo_password = "pw"  # noqa: S105
+        driver._device.send_command.return_value = _SOCKETS
+
+        driver.get_listening_sockets()
+
+        sent = self._sent(driver)[0]
+        assert sent.startswith("echo pw | sudo -S")
+        assert "sh -c '" in sent
+
+    def test_when_sudo_refuses_it_reads_what_the_user_may_see(self, driver):
+        driver._root = False
+        driver._device.send_command.side_effect = ["sudo: a password is required", _SOCKETS]
+
+        reading = driver.get_listening_sockets()
+
+        assert reading["attributed"] is False
+        refused, plain = self._sent(driver)
+        assert refused.startswith("sudo -n sh -c '")
+        assert plain.startswith("sh -c '")