Merge pull request 'feat: read the node's listening sockets through napalm-device-types' (#11) from feat/listening-sockets into master

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

@@ -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,
@@ -96,6 +98,10 @@ class ProxmoxDriver(
     USES_SSH = False
     # A PVE node reboots through a full init sequence plus storage checks.
     REBOOT_SETTLE_SECONDS = 90
+    #: "Upgrade everything" must be a full upgrade on Proxmox VE: a plain
+    #: ``apt-get upgrade`` holds back what needs new or removed packages and can
+    #: leave the node half updated. netOrk reads this when it upgrades the host.
+    FULL_UPGRADE = True
     PORT_SPECS = [
         PortSpec("https", 8006, weight=8.0),
     ]

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

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

tests/test_full_upgrade.py

@@ -0,0 +1,11 @@
+"""Proxmox VE says that "upgrade everything" must be a full upgrade on it.
+
+A plain ``apt-get upgrade`` can leave a node half updated; Proxmox documents
+``apt full-upgrade``. netOrk reads ``FULL_UPGRADE`` to choose.
+"""
+
+from napalm_proxmox.driver import ProxmoxDriver
+
+
+def test_a_full_upgrade_is_declared():
+    assert ProxmoxDriver.FULL_UPGRADE is True

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