feat: report what listens, and which procd service it is

Mixes in napalm-device-types' ListeningSocketsMixin (2.5.0): without ss
the shared command reads busybox netstat, and each process's cgroup names
its procd service. The driver only carries the command across.

Over an SSH exec channel of its own, not the interactive shell: busybox
ash cuts a typed line at 512 characters, the command is longer, and the
cut line left the shell waiting for a closing quote. OpenWrt logs in as
root; a login other than root cannot become root, so its reading says it
is not attributed.

Checked against a real OpenWrt 25.12.2 access point. For netOrk#673.

README.md

@@ -64,6 +64,7 @@ device.close()
 | `get_environment` | ✅ | CPU from `/proc/stat`, memory from `/proc/meminfo` |
 | `get_lldp_neighbors` | ✅ | Requires `lldpd` package installed on device |
 | `get_lldp_neighbors_detail` | ✅ | Requires `lldpd` package installed on device |
+| `get_listening_sockets` | ✅ | busybox `netstat -lntup`, procd service from the process's cgroup (napalm-device-types `ListeningSocketsMixin`); over an SSH exec channel |
 
 ## Configuration management
 

napalm_openwrt/openwrt.py

@@ -27,6 +27,7 @@ from netmiko import ConnectHandler
 from netmiko.exceptions import NetmikoTimeoutException, NetmikoAuthenticationException
 
 from napalm_device_types import AccessPointDriver, FingerprintRule
+from napalm_device_types.listening import ListeningSocketsMixin
 from napalm.base.exceptions import (
     ConnectionException,
     ConnectionClosedException,
@@ -52,6 +53,7 @@ class OpenWrtDriver(
     OpenWrtSystemMixin,
     OpenWrtPackageMixin,
     OpenWrtRoutingMixin,
+    ListeningSocketsMixin,
     AccessPointDriver,
 ):
     """NAPALM driver for OpenWrt routers and access-points."""
@@ -161,6 +163,22 @@ class OpenWrtDriver(
         except (socket.error, EOFError) as exc:
             raise ConnectionClosedException(str(exc)) from exc
 
+    def _run_listening_sockets_command(self, command: str, *, privileged: bool) -> str:
+        """Run the listening-socket command for ``ListeningSocketsMixin``.
+
+        Over an SSH exec channel of its own, not the interactive shell: busybox
+        ash cuts a typed line at 512 characters, and the command is longer. As
+        root it names every socket's process; a login other than root has no way
+        to become root here, so a privileged reading comes back empty and the
+        mixin reads again without attributing.
+        """
+        if privileged and self.username != "root":
+            return ""
+        _stdin, stdout, _stderr = self.device.remote_conn_pre.exec_command(
+            command, timeout=self.timeout
+        )
+        return stdout.read().decode("utf-8", "replace")
+
     @staticmethod
     def _parse_openwrt_release(output: str) -> dict[str, str]:
         """Parse ``/etc/openwrt_release`` key=value pairs."""

pyproject.toml

@@ -26,7 +26,7 @@ classifiers = [
 ]
 dependencies = [
     "napalm>=4.0.0",
-    "napalm_device_types>=0.1.0",
+    "napalm_device_types>=2.5.0",
     "netmiko>=4.0.0",
     "paramiko>=5.0.0",  # CVE-2026-44405
     "netaddr",

tests/unit/test_listening_sockets.py

@@ -0,0 +1,89 @@
+"""What listens on an OpenWrt device, and which procd service it is (netOrk #673).
+
+The command and its parse are napalm-device-types' (``ListeningSocketsMixin``):
+OpenWrt has no ``ss``, so the command falls back to busybox ``netstat``, and the
+cgroup of each process names its procd service. The driver only carries the
+command across -- over an SSH exec channel of its own, because busybox ash cuts
+an interactive line at 512 characters and the command is longer.
+
+The netstat output is a real OpenWrt 25.12.2 access point's, with documentation
+addresses.
+"""
+
+from __future__ import annotations
+
+import io
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from napalm_openwrt.openwrt import OpenWrtDriver
+
+REPORT = """SOCK_BEGIN
+[netstat]
+Active Internet connections (only servers)
+Proto Recv-Q Send-Q Local Address           Foreign Address         State       PID/Program name
+tcp        0      0 0.0.0.0:22              0.0.0.0:*               LISTEN      1604/dropbear
+tcp        0      0 0.0.0.0:443             0.0.0.0:*               LISTEN      1969/uhttpd
+tcp        0      0 192.0.2.15:53           0.0.0.0:*               LISTEN      1495/dnsmasq
+tcp        0      0 :::443                  :::*                    LISTEN      1969/uhttpd
+udp        0      0 0.0.0.0:161             0.0.0.0:*                           3173/snmpd
+__SS_RC=0
+[cgroups]
+1495 0::/services/dnsmasq/cfg01411c
+1604 0::/services/dropbear/instance1
+1969 0::/services/uhttpd/instance1
+3173 0::/services/snmpd/instance1
+SOCK_END
+"""
+
+
+def _driver(username: str = "root", report: str = REPORT) -> OpenWrtDriver:
+    with patch("napalm_openwrt.openwrt.ConnectHandler"):
+        drv = OpenWrtDriver(hostname="192.0.2.15", username=username, password="")
+    drv.device = MagicMock()
+    exec_command = drv.device.remote_conn_pre.exec_command
+    exec_command.side_effect = lambda *_a, **_k: (None, io.BytesIO(report.encode()), None)
+    return drv
+
+
+def test_every_socket_with_its_procd_service():
+    reading = _driver().get_listening_sockets()
+
+    assert reading["attributed"] is True
+    services = {(s["proto"], s["port"], s["address"]): s["unit"] for s in reading["sockets"]}
+    assert services == {
+        ("tcp", 22, "0.0.0.0"): "dropbear",
+        ("tcp", 53, "192.0.2.15"): "dnsmasq",
+        ("tcp", 443, "0.0.0.0"): "uhttpd",
+        ("tcp", 443, "::"): "uhttpd",
+        ("udp", 161, "0.0.0.0"): "snmpd",
+    }
+
+
+def test_the_command_goes_over_an_exec_channel_not_the_shell():
+    drv = _driver()
+    drv.get_listening_sockets()
+
+    [call] = drv.device.remote_conn_pre.exec_command.call_args_list
+    assert call.args[0].startswith("sh -c '")
+    drv.device.send_command.assert_not_called()
+
+
+def test_a_login_other_than_root_reads_without_attributing():
+    drv = _driver(username="admin")
+
+    reading = drv.get_listening_sockets()
+
+    assert reading["attributed"] is False
+    assert len(drv.device.remote_conn_pre.exec_command.call_args_list) == 1
+
+
+def test_a_report_cut_short_raises():
+    with pytest.raises(ValueError):
+        _driver(report=REPORT.replace("SOCK_END\n", "")).get_listening_sockets()
+
+
+def test_netork_finds_it():
+    """netOrk asks ``hasattr(driver, "get_listening_sockets")``."""
+    assert hasattr(OpenWrtDriver, "get_listening_sockets")