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.
This commit is contained in:
Christian Manivong
2026-10-06 18:19:57 +02:00
parent 6c9a6e7017
commit c644519af6
4 changed files with 49 additions and 1 deletions
+37
View File
@@ -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")