diff --git a/napalm_proxmox/driver.py b/napalm_proxmox/driver.py index 13e737e..5c57f0e 100644 --- a/napalm_proxmox/driver.py +++ b/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.""" diff --git a/napalm_proxmox/system_mixin.py b/napalm_proxmox/system_mixin.py index 05f8fc2..d8d3d65 100644 --- a/napalm_proxmox/system_mixin.py +++ b/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" + 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:]} + + 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: + 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" ) - 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" + + 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", ) - 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]}") - # 4. Open firewall if ufw is present - if netork_ip: - 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" - ) - lines.append(f"[firewall/ufw] {fw_out.strip()[:200]}") - except Exception as exc: - lines.append(f"[firewall] skipped - {exc}") + 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)}") - # 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 _start_snmp_agent(self) -> dict: + return self._step( + "systemctl enable snmpd && systemctl restart snmpd && systemctl is-active snmpd", + "snmpd enabled and running", ) - import time as _time - _time.sleep(1) - start_out = self._exec_ssh_command( - "service snmpd start 2>&1 || systemctl start snmpd 2>&1 || true" - ) - 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. diff --git a/pyproject.toml b/pyproject.toml index 8587146..81cc0c7 100644 --- a/pyproject.toml +++ b/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", diff --git a/tests/test_fix_templates.py b/tests/test_fix_templates.py new file mode 100644 index 0000000..dfaa5af --- /dev/null +++ b/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