Merge pull request 'ci: run the tests and build the package on every push and pull request' (#7) from ci/workflow into master

The same job as napalm-fritzbox and napalm-opnsense: Python 3.10, 3.11 and
3.12, pytest, then wheel and sdist. napalm-device-types comes from
git.netork.io first, because PyPI has an unrelated package of that name.

.github/workflows/ci.yml

@@ -0,0 +1,48 @@
+name: CI
+
+on:
+  push:
+    branches: ["**"]
+  pull_request:
+    branches: ["**"]
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+    strategy:
+      fail-fast: false
+      matrix:
+        python-version: ["3.10", "3.11", "3.12"]
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v4
+
+      - name: Setup Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: ${{ matrix.python-version }}
+          cache: pip
+
+      - name: Install package with dev extras
+        run: |
+          python -m pip install --upgrade pip
+          # napalm-device-types lives in git.netork.io/NAPALM, not on PyPI: without this
+          # pip looks there, finds an unrelated 0.1.0 and the job dies before any test.
+          python -m pip install "napalm-device-types @ git+https://git.netork.io/NAPALM/napalm-device-types.git"
+          python -m pip install -e ".[dev]"
+
+      - name: Run unit tests
+        run: |
+          python -m pytest -q --tb=short
+
+      - name: Build wheel and sdist
+        run: |
+          python -m pip install build
+          python -m build
+
+      - name: Upload dist artifacts
+        # v4 refuses to run on Gitea ("not currently supported on GHES").
+        uses: actions/upload-artifact@v3
+        with:
+          name: dist-${{ matrix.python-version }}
+          path: dist/*

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")