napalm_proxmox/driver.py

@@ -35,12 +35,14 @@ from typing import Any
 logger = logging.getLogger(__name__)
 
 from napalm_device_types import (
+    EnsurePackageMixin,
     FingerprintRule,
     HostStatusMixin,
     HypervisorDriver,
     KernelFactsMixin,
     ListeningSocketsMixin,
     PortSpec,
+    SnmpAgentMixin,
     SystemIdentityMixin,
     SystemdServicesMixin,
 )
@@ -90,6 +92,8 @@ class ProxmoxDriver(
     SystemIdentityMixin,
     SystemdServicesMixin,
     HostStatusMixin,
+    SnmpAgentMixin,
+    EnsurePackageMixin,
     HypervisorDriver,
 ):
     """NAPALM driver for Proxmox VE nodes."""

napalm_proxmox/system_mixin.py

@@ -16,6 +16,8 @@
 
 from __future__ import annotations
 
+import base64
+import ipaddress
 import logging
 import re
 from typing import Any
@@ -29,6 +31,33 @@ logger = logging.getLogger(__name__)
 _JsonDict = dict[str, Any]
 
 
+_PACKAGE_NAME = re.compile(r"^[a-zA-Z0-9][a-zA-Z0-9_.+:~-]*$")
+#: What may stand in an snmpd.conf value: no whitespace, no quote, no comment.
+_SNMP_WORD = re.compile(r'^[^\s"#]+$')
+
+
+def _net(value: str) -> str:
+    return str(ipaddress.ip_network(value.strip(), strict=False))
+
+
+def render_snmpd_conf(settings: dict) -> str:
+    """net-snmp's configuration for *settings*: read-only v2c, per allowed source."""
+    community = settings["community"]
+    if not _SNMP_WORD.match(community):
+        raise ValueError("the community must be one word without quotes or #")
+    listen = settings.get("listen")
+    lines = [f"agentAddress udp:{listen + ':' if listen else ''}161"]
+    sources = [_net(s) for s in settings.get("allowed_sources", [])]
+    lines += [f"rocommunity {community} {s}" for s in sources] or [f"rocommunity {community}"]
+    for key, directive in (("location", "sysLocation"), ("contact", "sysContact")):
+        value = settings.get(key, "")
+        if value:
+            if "\n" in value:
+                raise ValueError(f"{key} must be one line")
+            lines.append(f"{directive}    {value}")
+    return "\n".join(lines) + "\n"
+
+
 class ProxmoxSystemMixin:
     """Mixin providing system-level NAPALM methods."""
 
@@ -474,131 +503,69 @@ class ProxmoxSystemMixin:
 
         return SNMPConfigDict(running=True, community=community, port=port, version="2c")
 
-    def run_device_action(self, action: str) -> dict:
-        """Execute a named administrative action on the Proxmox node."""
-        if action == "fix_snmp":
-            return self._action_fix_snmp()
-        if action == "install_lldpd":
-            return self._action_install_lldpd()
-        raise NotImplementedError(f"Unknown action: {action!r}")
+    # ------------------------------------------------------------------
+    # Fixes (napalm-device-types SnmpAgentMixin, EnsurePackageMixin): the
+    # caller's values. The node's shell runs as root.
+    # ------------------------------------------------------------------
 
-    def _action_install_lldpd(self) -> dict:
-        """Install, enable and start lldpd on the Proxmox node.
-
-        Proxmox runs Debian/Linux underneath, so this is a plain apt install
-        followed by enabling the systemd service. lldpd's defaults (listen on
-        all interfaces) are sufficient to discover the directly-connected
-        switch via the management bridge (e.g. vmbr0).
-        """
-        lines: list[str] = []
-
-        install_out = self._exec_ssh_command(
-            "DEBIAN_FRONTEND=noninteractive apt-get install -y lldpd 2>&1 | tail -5"
-        )
-        lines.append(f"[install] {install_out.strip()[-200:]}")
-
-        enable_out = self._exec_ssh_command(
-            "systemctl enable --now lldpd 2>&1 || service lldpd start 2>&1 || true"
-        )
-        lines.append(f"[service] {enable_out.strip()[-200:]}")
-
-        verify = self._exec_ssh_command("command -v lldpd 2>/dev/null").strip()
-        success = bool(verify)
-        if success:
-            lines.append("[ok] lldpd installed. Neighbors will appear after a short warm-up period.")
-        else:
-            lines.append("[warn] lldpd does not appear to be installed after the attempt.")
-
-        return {"success": success, "output": "\n".join(lines)}
-
-    def _action_fix_snmp(self) -> dict:
-        """Install, configure and start snmpd on the Proxmox node.
-
-        Proxmox runs Debian/Linux underneath. _exec_ssh_command runs as root
-        (either via Proxmox API execute endpoint or SSH with root credentials),
-        so no sudo is needed.
-        """
-        import base64 as _b64
-        lines: list[str] = []
-
-        # 1. Install snmpd and snmp client tools
-        install_out = self._exec_ssh_command(
-            "DEBIAN_FRONTEND=noninteractive apt-get install -y snmpd snmp 2>&1 | tail -5"
-        )
-        lines.append(f"[install] {install_out.strip()[-200:]}")
-
-        # 2. Detect the IP this connection comes from (for firewall rule)
-        netork_ip = ""
+    def _rc(self, command: str) -> tuple[str, int]:
+        """Run *command* and return its output and exit status (-1 when unknown)."""
+        out = self._exec_ssh_command(f"{{ {command}; }} 2>&1; echo __PVE_RC=$?")
+        text, _, tail = out.rpartition("__PVE_RC=")
         try:
-            raw = self._exec_ssh_command(
-                "ss -tnp 2>/dev/null | awk '/sshd/{print $5}' | head -1 | cut -d: -f1"
-            ).strip()
-            if raw and raw not in ("", "0.0.0.0", "::", "127.0.0.1"):
-                netork_ip = raw
-        except Exception as exc:
-            logger.debug("Failed to detect network origin IP: %s", exc)
+            return text.strip(), int(tail.strip())
+        except ValueError:
+            return out.strip(), -1
 
-        # 3. Write snmpd.conf via /tmp (no permission issues)
-        conf_str = (
-            "agentAddress udp:161\n"
-            "rocommunity public\n"
-            "sysLocation    Managed by netOrk\n"
-            "sysContact     netork@localhost\n"
-        )
-        conf_b64 = _b64.b64encode(conf_str.encode()).decode()
-        self._exec_ssh_command(f"echo {conf_b64} | base64 -d > /tmp/netork_snmpd.conf")
-        self._exec_ssh_command(
-            "mv /tmp/netork_snmpd.conf /etc/snmp/snmpd.conf && "
-            "chown root:root /etc/snmp/snmpd.conf && chmod 644 /etc/snmp/snmpd.conf"
-        )
-        verify = self._exec_ssh_command("cat /etc/snmp/snmpd.conf 2>/dev/null").strip()
-        if "agentAddress" in verify and "rocommunity" in verify:
-            lines.append("[config] Wrote /etc/snmp/snmpd.conf - agentAddress udp:161, rocommunity public.")
-        else:
-            lines.append(f"[warn] snmpd.conf write may have failed: {verify[:100]}")
+    def _step(self, command: str, done: str) -> dict:
+        output, status = self._rc(command)
+        return {"ok": status == 0, "output": done if status == 0 else output[-300:]}
 
-        # 4. Open firewall if ufw is present
-        if netork_ip:
+    def install_package(self, name: str, version: str = "") -> None:
+        """apt-get install *name* (``PackageManagementMixin``); raises when apt fails."""
+        if not _PACKAGE_NAME.match(name) or (version and not _PACKAGE_NAME.match(version)):
+            raise ValueError(f"not a package name: {name!r}")
+        target = f"{name}={version}" if version else name
+        output, status = self._rc(f"DEBIAN_FRONTEND=noninteractive apt-get install -y {target}")
+        if status != 0:
+            raise RuntimeError(f"apt-get install {target} failed: {output[-300:]}")
+
+    def get_management_peer(self) -> str | None:
+        """The address the node sees the SSH connection come from (``SSH_CONNECTION``)."""
+        out = self._exec_ssh_command('printf "%s\\n" "$SSH_CONNECTION"').split()
         try:
-                ufw = self._exec_ssh_command("command -v ufw 2>/dev/null").strip()
-                if ufw:
-                    parts = netork_ip.rsplit(".", 1)
-                    subnet = f"{parts[0]}.0/24" if len(parts) == 2 else netork_ip
-                    fw_out = self._exec_ssh_command(
-                        f"ufw allow from {subnet} to any port 161 proto udp 2>&1"
+            return str(ipaddress.ip_address(out[0])) if out else None
+        except ValueError:
+            return None
+
+    def _install_snmp_agent(self) -> dict:
+        return self._step(
+            "DEBIAN_FRONTEND=noninteractive apt-get install -y snmpd snmp", "net-snmp installed"
         )
-                    lines.append(f"[firewall/ufw] {fw_out.strip()[:200]}")
-            except Exception as exc:
-                lines.append(f"[firewall] skipped - {exc}")
 
-        # 5. Stop and restart snmpd cleanly (no DBus needed for stop+start)
-        self._exec_ssh_command(
-            "service snmpd stop 2>/dev/null; pkill -9 snmpd 2>/dev/null; true"
+    def _write_snmp_agent_config(self, settings: dict) -> dict:
+        encoded = base64.b64encode(render_snmpd_conf(settings).encode()).decode()
+        return self._step(
+            f"umask 077 && t=$(mktemp) && echo {encoded} | base64 -d > \"$t\" && "
+            f"install -m 644 -o root -g root \"$t\" /etc/snmp/snmpd.conf; "
+            "s=$?; rm -f \"$t\"; exit $s",
+            "wrote /etc/snmp/snmpd.conf",
         )
-        import time as _time
-        _time.sleep(1)
-        start_out = self._exec_ssh_command(
-            "service snmpd start 2>&1 || systemctl start snmpd 2>&1 || true"
+
+    def _allow_snmp_sources(self, settings: dict) -> dict:
+        sources = [_net(s) for s in settings.get("allowed_sources", [])]
+        if not sources:
+            return {"ok": True, "output": "no sources given; firewall unchanged"}
+        if self._rc("command -v ufw")[1] != 0:
+            return {"ok": True, "output": "no ufw; pve-firewall rules are the operator's"}
+        rules = " && ".join(f"ufw allow from {s} to any port 161 proto udp" for s in sources)
+        return self._step(rules, f"udp/161 open for {', '.join(sources)}")
+
+    def _start_snmp_agent(self) -> dict:
+        return self._step(
+            "systemctl enable snmpd && systemctl restart snmpd && systemctl is-active snmpd",
+            "snmpd enabled and running",
         )
-        lines.append(f"[service] {start_out.strip()[-200:]}")
-
-        # 6. Verify via local probe
-        _time.sleep(2)
-        probe_out = self._exec_ssh_command(
-            "snmpget -v2c -cpublic -t2 -r0 -Ov 127.0.0.1 1.3.6.1.2.1.1.1.0 2>&1 || true"
-        ).strip()
-        _snmp_types = ("STRING:", "INTEGER:", "OID:", "Timeticks:", "Hex-STRING:", "IpAddress:")
-        success = any(t in probe_out for t in _snmp_types)
-        if success:
-            lines.append("[ok] SNMP probe successful - community 'public' is working.")
-        else:
-            lines.append(f"[warn] SNMP probe failed - output: {probe_out[:200]}")
-
-        return {"success": success, "output": "\n".join(lines)}
-
-    # ------------------------------------------------------------------ #
-    # set_hostname
-    # ------------------------------------------------------------------ #
 
     def set_hostname(self, new_hostname: str) -> None:
         """Set the system hostname on the Proxmox node.

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "napalm-proxmox"
-version = "0.2.0"
+version = "0.3.0"
 description = "NAPALM driver for Proxmox VE (classic Linux networking, SDN, OVS)"
 readme = "README.md"
 license = { text = "Apache-2.0" }
@@ -25,7 +25,7 @@ classifiers = [
 requires-python = ">=3.9"
 dependencies = [
     "napalm>=5.0.0",
-    "napalm_device_types>=4.9.0",
+    "napalm_device_types>=4.13.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_fix_templates.py

@@ -0,0 +1,118 @@
+"""The fixes as napalm-device-types templates, with the caller's values (netOrk #1105).
+
+``fix_snmp`` wrote an application's snmpd.conf (community "public", "Managed by
+netOrk") and opened ufw for the caller's /24 guessed from ``ss``;
+``install_lldpd`` was an action of its own. Now ``configure_snmp_agent(settings)``
+takes every value from the caller, and ``ensure_package`` installs through the
+contract's ``install_package``. Every step reads its exit status.
+"""
+
+from __future__ import annotations
+
+import re
+
+import pytest
+
+from napalm_proxmox import ProxmoxDriver
+
+SETTINGS = {"community": "c0mm", "allowed_sources": ["10.7.224.20"], "location": "rack 1"}
+
+
+class _Node:
+    """The node's shell: answers by command fragment, with an exit status."""
+
+    def __init__(self, **status):
+        self.status = status  # fragment -> exit status
+        self.ran: list[str] = []
+
+    def __call__(self, command: str) -> str:
+        self.ran.append(command)
+        if "SSH_CONNECTION" in command:
+            return "10.7.224.20 51234 10.0.0.9 22\n"
+        if command.startswith("dpkg-query"):
+            return "curl\tinstalled\n"
+        rc = next((code for frag, code in self.status.items() if frag in command), 0)
+        out = "ok" if rc == 0 else "E: failed"
+        return f"{out}\n__PVE_RC={rc}" if "__PVE_RC" in command else out
+
+
+@pytest.fixture
+def node():
+    return _Node()
+
+
+@pytest.fixture
+def driver(node):
+    d = ProxmoxDriver("pve", "root@pam", "pw")
+    d._exec_ssh_command = node
+    return d
+
+
+class TestSnmpAgent:
+    def test_install_write_open_start_with_the_callers_values(self, driver, node):
+        result = driver.configure_snmp_agent(SETTINGS)
+
+        assert result["success"] is True
+        assert [s["step"] for s in result["steps"]] == ["install", "configure", "firewall", "start"]
+        assert any("apt-get install -y snmpd snmp" in c for c in node.ran)
+        write = next(c for c in node.ran if "/etc/snmp/snmpd.conf" in c)
+        assert "install -m 644" in write
+        b64 = re.search(r"echo ([A-Za-z0-9+/=]+) \| base64 -d", write).group(1)
+        import base64
+
+        conf = base64.b64decode(b64).decode()
+        assert "rocommunity c0mm 10.7.224.20/32" in conf
+        assert "sysLocation    rack 1" in conf
+        assert "public" not in conf and "netork" not in conf.lower()
+        assert any("systemctl restart snmpd" in c for c in node.ran)
+
+    def test_no_temp_file_with_an_application_name(self, driver, node):
+        driver.configure_snmp_agent(SETTINGS)
+
+        assert not any("netork" in c for c in node.ran)
+
+    def test_ufw_opens_only_for_the_allowed_sources(self, driver, node):
+        node.status["command -v ufw"] = 0
+
+        driver.configure_snmp_agent(SETTINGS)
+
+        assert any("ufw allow from 10.7.224.20/32 to any port 161 proto udp" in c for c in node.ran)
+        assert not any(".0/24" in c for c in node.ran)
+
+    def test_a_failed_install_stops_the_fix(self, driver, node):
+        node.status["apt-get install -y snmpd"] = 100
+
+        result = driver.configure_snmp_agent(SETTINGS)
+
+        assert result["success"] is False
+        assert not any("/etc/snmp/snmpd.conf" in c for c in node.ran)
+
+    def test_the_management_peer(self, driver):
+        assert driver.get_management_peer() == "10.7.224.20"
+
+
+class TestPackages:
+    def test_ensure_package_installs_what_is_missing(self, driver, node):
+        result = driver.ensure_package("lldpd")
+
+        assert result["success"] is True
+        assert any("apt-get install -y lldpd" in c for c in node.ran)
+
+    def test_ensure_package_leaves_alone_what_is_there(self, driver, node):
+        assert driver.ensure_package("curl")["success"] is True
+        assert not any("apt-get install" in c for c in node.ran)
+
+    def test_install_package_raises_when_apt_fails(self, driver, node):
+        node.status["apt-get install -y lldpd"] = 100
+
+        with pytest.raises(RuntimeError, match="lldpd"):
+            driver.install_package("lldpd")
+
+    def test_a_name_that_is_no_package_name_is_refused(self, driver):
+        with pytest.raises(ValueError):
+            driver.install_package("lldpd; reboot")
+
+
+def test_the_old_actions_are_gone():
+    for name in ("run_device_action", "_action_fix_snmp", "_action_install_lldpd"):
+        assert not hasattr(ProxmoxDriver, name), name