From c644519af699e277edcea9c6785999814f6fb08a Mon Sep 17 00:00:00 2001
From: Christian Manivong <christian.manivong@gmail.com>
Date: Tue, 6 Oct 2026 18:19:57 +0200
Subject: [PATCH] feat: read the node's listening sockets through
 napalm-device-types

ProxmoxDriver mixes in ListeningSocketsMixin (napalm-device-types 2.4.0)
and carries its command over the exec path, which runs as root either
way: whether pveproxy, a Ceph manager or anything else on the node listens
on an address reachable from outside it.

For netOrk#658.
---
 napalm_proxmox/driver.py        |  2 ++
 napalm_proxmox/system_mixin.py  |  9 ++++++++
 pyproject.toml                  |  2 +-
 tests/test_listening_sockets.py | 37 +++++++++++++++++++++++++++++++++
 4 files changed, 49 insertions(+), 1 deletion(-)
 create mode 100644 tests/test_listening_sockets.py

diff --git a/napalm_proxmox/driver.py b/napalm_proxmox/driver.py
index 42a8f50..49b53ad 100644
--- a/napalm_proxmox/driver.py
+++ b/napalm_proxmox/driver.py
@@ -39,6 +39,7 @@ from napalm_device_types import (
     HostStatusMixin,
     HypervisorDriver,
     KernelFactsMixin,
+    ListeningSocketsMixin,
     PortSpec,
     SystemdServicesMixin,
 )
@@ -84,6 +85,7 @@ class ProxmoxDriver(
     ProxmoxRoutingMixin,
     ProxmoxSystemMixin,
     KernelFactsMixin,
+    ListeningSocketsMixin,
     SystemdServicesMixin,
     HostStatusMixin,
     HypervisorDriver,
diff --git a/napalm_proxmox/system_mixin.py b/napalm_proxmox/system_mixin.py
index 9f6317e..f476e35 100644
--- a/napalm_proxmox/system_mixin.py
+++ b/napalm_proxmox/system_mixin.py
@@ -231,6 +231,15 @@ class ProxmoxSystemMixin:
         """The transport for ``KernelFactsMixin.get_kernel_facts``: the exec path."""
         return self._exec_ssh_command(command)
 
+    # ------------------------------------------------------------------ #
+    # Listening sockets (ListeningSocketsMixin supplies get_listening_sockets)
+    # ------------------------------------------------------------------ #
+
+    def _run_listening_sockets_command(self, command: str, *, privileged: bool) -> str:
+        """The transport for ``ListeningSocketsMixin.get_listening_sockets``:
+        the exec path, which runs as root either way."""
+        return str(self._exec_ssh_command(command))
+
     # ------------------------------------------------------------------ #
     # Packages (Debian APT)
     # ------------------------------------------------------------------ #
diff --git a/pyproject.toml b/pyproject.toml
index d1f99bf..f873068 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -25,7 +25,7 @@ classifiers = [
 requires-python = ">=3.9"
 dependencies = [
     "napalm>=5.0.0",
-    "napalm_device_types>=2.3.0",
+    "napalm_device_types>=2.4.0",
     "paramiko>=5.0.0",  # CVE-2026-44405; imported directly for SSH fallback (driver.py)
     "proxmoxer>=2.0.0",
     "netaddr>=0.9.0",
diff --git a/tests/test_listening_sockets.py b/tests/test_listening_sockets.py
new file mode 100644
index 0000000..58359b3
--- /dev/null
+++ b/tests/test_listening_sockets.py
@@ -0,0 +1,37 @@
+"""`get_listening_sockets`: what listens on the node, and which service it is.
+
+Whether pveproxy, a Ceph manager or a guest-facing service is reachable from
+outside the node is decided by the address it listens on. The command and its
+parse are napalm-device-types'; the driver only carries the command over its
+exec path, which already runs as root.
+"""
+
+from __future__ import annotations
+
+from unittest.mock import patch
+
+from napalm_device_types import ListeningSocketsMixin
+from napalm_proxmox.driver import ProxmoxDriver
+
+WIRE = (
+    "SOCK_BEGIN\n[ss]\n"
+    'tcp LISTEN 0 4096 *:8006 *:* users:(("pveproxy worker",pid=2101,fd=6))\n'
+    "__SS_RC=0\n[cgroups]\n"
+    "2101 0::/system.slice/pveproxy.service\n"
+    "SOCK_END\n"
+)
+
+
+def test_the_driver_declares_the_contract():
+    assert issubclass(ProxmoxDriver, ListeningSocketsMixin)
+
+
+def test_it_reads_as_root_over_the_exec_path(driver):
+    with patch.object(driver, "_exec_ssh_command", return_value=WIRE) as exec_:
+        reading = driver.get_listening_sockets()
+
+    [command] = [c.args[0] for c in exec_.call_args_list]
+    assert command.startswith("sh -c '")
+    assert reading["attributed"] is True
+    [socket] = reading["sockets"]
+    assert (socket["address"], socket["port"], socket["unit"]) == ("*", 8006, "pveproxy")