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

get_listening_sockets reads /api/diagnostics/interface/get_socket_statistics:
netstat's sockets with sockstat's user, command and PID, collected by
configd as root, so no SSH is needed. A listening socket is one without a
peer (*:*); "*" is the any-address of the socket's family.

Which service: the socket's unit is a name of the firewall's own service
list, so netOrk can match it to the service. The command (cut to ten
characters by FreeBSD) matches a service name, the start of one, or one
of the daemons whose service is called otherwise (sshd is openssh, the
FRR daemons are frr, kea-ctrl-agent is kea-dhcp). lighttpd is the web UI,
or the captive portal when it runs as www. WireGuard's sockets belong to
the kernel and are told by the listen ports of /api/wireguard/service/show.

The fixture is a real OPNsense 26.7.5 firewall's answer, cut down, with
documentation addresses. For netOrk#673.

napalm_opnsense/listening.py

@@ -0,0 +1,128 @@
+"""What listens on the firewall, and which OPNsense service it is. Pure: no I/O.
+
+``/api/diagnostics/interface/get_socket_statistics`` is netstat's socket list
+with sockstat's user, command and PID merged in, read by configd as root. The
+contract this serves -- ``napalm_device_types.models.ListeningSocketsDict`` --
+wants the sockets that listen: a TCP socket in LISTEN and a bound UDP socket,
+both without a peer (``*:*``).
+
+**Which service.** netOrk keeps a firewall's services under the names of its
+service list (``/api/core/service/search``), so a socket's ``unit`` is one of
+those names, or None. The statistics name the process, cut to ten characters
+by FreeBSD (``mdns-repea``): a name the service list has wins, then one the
+command starts with or -- cut short -- is the start of, then the few daemons
+whose service is called otherwise (``sshd`` is ``openssh``, ``ospfd`` is
+``frr``). The admin UI and the captive portal are both lighttpd; the portal's
+runs as ``www``.
+
+**WireGuard has no process.** Its sockets belong to the kernel (``??``) and are
+told by the listen ports of the WireGuard interfaces.
+"""
+
+from __future__ import annotations
+
+from typing import Any, Dict, Iterable, List, Optional, Set
+
+from napalm_device_types.models import ListeningSocketDict
+
+#: FRR's daemons, all under the one service ``frr``.
+_FRR_DAEMONS = "zebra mgmtd staticd ospfd ospf6d bgpd bfdd ripd ripngd isisd pimd watchfrr".split()
+#: Daemons whose OPNsense service has another name, by the command as reported.
+_SERVICE_OF: Dict[str, str] = {
+    "sshd": "openssh",
+    "kea-ctrl-a": "kea-dhcp",
+    "kea-ctrl-agent": "kea-dhcp",
+    "charon": "strongswan",
+    **{daemon: "frr" for daemon in _FRR_DAEMONS},
+}
+#: FreeBSD cuts a command at this many characters in the socket list.
+_COMMAND_LENGTH = 10
+#: Below this length a service name is too short to match a command's start (``pf``).
+_PREFIX_MIN = 4
+_NOBODY = "??"
+_PEERLESS = "*"
+
+
+def _service_of(command: str, user: str, services: Set[str]) -> Optional[str]:
+    if command == "lighttpd":
+        name = "captiveportal" if user == "www" else "webgui"
+        return name if name in services else None
+    if command in services:
+        return command
+    alias = _SERVICE_OF.get(command)
+    if alias in services:
+        return alias
+    matches = [
+        name
+        for name in services
+        if (len(name) >= _PREFIX_MIN and command.startswith(name))
+        or (len(command) == _COMMAND_LENGTH and name.startswith(command))
+    ]
+    return matches[0] if len(matches) == 1 else None
+
+
+def _address(protocol: str, address: str) -> tuple:
+    """``(address, interface)``: ``*`` is every address of the socket's family."""
+    if address == "*":
+        return ("::" if protocol.endswith("6") else "0.0.0.0"), None
+    host, _, zone = address.partition("%")
+    return host, zone or None
+
+
+def _socket(
+    entry: Dict[str, Any], services: Set[str], wireguard_ports: Set[int]
+) -> Optional[ListeningSocketDict]:
+    protocol = str(entry.get("protocol") or "")
+    proto = protocol[:3]
+    local, remote = entry.get("local") or {}, entry.get("remote") or {}
+    port = str(local.get("port") or "")
+    if proto not in ("tcp", "udp") or not port.isdigit():
+        return None
+    if remote.get("address") != _PEERLESS or remote.get("port") != _PEERLESS:
+        return None
+    command, pid = str(entry.get("command") or _NOBODY), str(entry.get("pid") or _NOBODY)
+    process = None if command == _NOBODY else command
+    if process is not None:
+        unit = _service_of(process, str(entry.get("user") or ""), services)
+    elif proto == "udp" and int(port) in wireguard_ports and "wireguard" in services:
+        unit = "wireguard"
+    else:
+        unit = None
+    address, interface = _address(protocol, str(local.get("address") or ""))
+    return {
+        "proto": proto,
+        "address": address,
+        "port": int(port),
+        "interface": interface,
+        "process": process,
+        "pid": int(pid) if pid.isdigit() else None,
+        "unit": unit,
+        "container_id": None,
+    }
+
+
+def wireguard_listen_ports(show: Dict[str, Any]) -> Set[int]:
+    """The listen ports of the WireGuard interfaces in ``/api/wireguard/service/show``."""
+    ports: Set[int] = set()
+    for row in show.get("rows") or []:
+        port = str(row.get("listen-port") or "")
+        if row.get("type") == "interface" and port.isdigit():
+            ports.add(int(port))
+    return ports
+
+
+def parse_socket_statistics(
+    statistics: Dict[str, Any], services: Iterable[str], wireguard_ports: Set[int]
+) -> List[ListeningSocketDict]:
+    """The listening sockets of ``get_socket_statistics``, sorted by protocol,
+    port and address. Every section is read: the UNIX sockets have no ``local``."""
+    names = set(services)
+    sockets: List[ListeningSocketDict] = []
+    for section in (statistics.get("statistics") or {}).values():
+        for entry in (section or {}).values() if isinstance(section, dict) else ():
+            socket = _socket(entry, names, wireguard_ports) if isinstance(entry, dict) else None
+            if socket is not None:
+                sockets.append(socket)
+    return sorted(
+        sockets, key=lambda s: (s["proto"], s["port"], s["address"], s["interface"] or "")
+    )

napalm_opnsense/opnsense.py

@@ -53,8 +53,10 @@ import requests
 from requests.exceptions import RequestException
 
 from napalm_device_types import FingerprintRule, FirewallDriver
+from napalm_device_types.models import ListeningSocketsDict
 from napalm.base.exceptions import ConnectionException, ConnectionClosedException, MergeConfigException
 
+from napalm_opnsense.listening import parse_socket_statistics, wireguard_listen_ports
 from napalm_opnsense.ping_mixin import OPNsensePingMixin
 from napalm_opnsense.port_forwards import alias_index, port_forwards, wan_interfaces
 
@@ -1789,6 +1791,26 @@ class OPNsenseDriver(OPNsensePingMixin, FirewallDriver):
             })
         return sorted(result, key=lambda x: x["name"].lower())
 
+    def get_listening_sockets(self) -> ListeningSocketsDict:
+        """Every listening TCP and bound UDP socket, with the OPNsense service behind it.
+
+        Read from ``/api/diagnostics/interface/get_socket_statistics``, which
+        configd collects as root, so every socket names its process where it has
+        one (``attributed``). Which service: :mod:`napalm_opnsense.listening`.
+        """
+        statistics = self._get("/api/diagnostics/interface/get_socket_statistics")
+        services = [service["name"] for service in self.get_services()]
+        try:
+            wireguard = wireguard_listen_ports(self._get("/api/wireguard/service/show"))
+        except Exception as exc:
+            # No WireGuard here: its kernel sockets stay nobody's.
+            logger.debug("WireGuard listen ports not read: %s", exc)
+            wireguard = set()
+        return {
+            "attributed": True,
+            "sockets": parse_socket_statistics(statistics, services, wireguard),
+        }
+
     def manage_service(self, name: str, action: str) -> dict[str, Any]:
         """Execute a lifecycle action on an OPNsense service.