NAPALM/napalm-proxmox
1
0
Fork 0
Code Issues 1 Pull Requests 1 Actions Packages Projects Releases Wiki Activity

fix: accept ssl_verify and verify aliases alongside verify_ssl

Browse Source 
Mirrors the robust fallback pattern from napalm-opnsense so the driver
works regardless of which key name the caller passes in optional_args.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Christian Manivong
2026-06-04 22:40:35 +02:00
co-authored by Claude Sonnet 4.6
parent f3ecf14c8d
commit 90d6592159
1 changed files with 287 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
napalm_proxmox/driver.py
+287 -17
View File
@@ -85,7 +85,9 @@ class ProxmoxDriver(HypervisorDriver):
self.optional_args: _JsonDict = optional_args or {}
self._port: int = self.optional_args.get("port", 8006)
self._verify_ssl: bool = self.optional_args.get("verify_ssl", True)
self._verify_ssl: bool = self.optional_args.get(
"verify_ssl", self.optional_args.get("ssl_verify", self.optional_args.get("verify", True))
)
self._realm: str = self.optional_args.get("realm", "pam")
self._token_name: str | None = self.optional_args.get("token_name")
self._token_value: str | None = self.optional_args.get("token_value")
@@ -93,6 +95,7 @@ class ProxmoxDriver(HypervisorDriver):
self._api: ProxmoxAPI | None = None
self._node_name: str = ""
self._ssh_client: "paramiko.SSHClient | None" = None
# Candidate config (merge/replace)
self._candidate_config: str = ""
@@ -136,6 +139,27 @@ class ProxmoxDriver(HypervisorDriver):
f"Cannot connect to Proxmox at {self.hostname}:{self._port} — {exc}"
) from exc
# Establish a persistent SSH connection for shell commands (used by
# _exec_ssh_command as a fast path, avoiding per-call connect overhead).
if self.username and self.password:
try:
import paramiko # noqa: PLC0415
ssh_port: int = self.optional_args.get("ssh_port", 22)
client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
client.connect(
hostname=self.hostname,
port=ssh_port,
username=self.username,
password=self.password,
timeout=self.timeout,
look_for_keys=False,
allow_agent=False,
)
self._ssh_client = client
except Exception:
self._ssh_client = None
def _resolve_node(self) -> str:
"""Return the Proxmox node name for this host."""
if self._node:
@@ -160,6 +184,12 @@ class ProxmoxDriver(HypervisorDriver):
def close(self) -> None:
"""Close the session (Proxmox REST is stateless; nothing to tear down)."""
self._api = None
if self._ssh_client is not None:
try:
self._ssh_client.close()
except Exception:
pass
self._ssh_client = None
def is_alive(self) -> _JsonDict:
"""Return connection state."""
@@ -201,28 +231,53 @@ class ProxmoxDriver(HypervisorDriver):
except Exception:
pass
# 2. Fall back to SSH if we have credentials
if not self.username or not self.password:
# 2. Fall back to SSH using explicit SSH credentials from optional_args,
# or the driver's own username/password as a last resort.
ssh_user = self.optional_args.get("ssh_username") or self.username
ssh_pass = self.optional_args.get("ssh_password") or self.password
ssh_key_str: str | None = self.optional_args.get("ssh_private_key_str")
if not ssh_user or (not ssh_pass and not ssh_key_str):
return ""
try:
import paramiko # noqa: PLC0415
ssh_port: int = self.optional_args.get("ssh_port", 22)
client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
client.connect(
hostname=self.hostname,
port=ssh_port,
username=self.username,
password=self.password,
timeout=self.timeout,
look_for_keys=False,
allow_agent=False,
)
import tempfile, os as _os # noqa: PLC0415
client = self._ssh_client
if client is None or not client.get_transport() or not client.get_transport().is_active():
ssh_port: int = self.optional_args.get("ssh_port", 22)
client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
connect_kwargs: dict = dict(
hostname=self.hostname,
port=ssh_port,
username=ssh_user,
timeout=self.timeout,
look_for_keys=False,
allow_agent=False,
)
_tmp_key = None
if ssh_key_str:
_tmp_key = tempfile.NamedTemporaryFile(mode="w", suffix=".pem", delete=False)
_tmp_key.write(ssh_key_str)
_tmp_key.flush()
_tmp_key.close()
connect_kwargs["key_filename"] = _tmp_key.name
else:
connect_kwargs["password"] = ssh_pass
try:
client.connect(**connect_kwargs)
finally:
if _tmp_key:
try:
_os.unlink(_tmp_key.name)
except OSError:
pass
self._ssh_client = client
_, stdout, _ = client.exec_command(command, timeout=self.timeout)
output = stdout.read().decode("utf-8", errors="replace")
client.close()
return output
except Exception:
self._ssh_client = None
return ""
def _get_version_info(self) -> _JsonDict:
@@ -681,7 +736,13 @@ class ProxmoxDriver(HypervisorDriver):
if current_iface not in entry["tagged"]:
entry["tagged"].append(current_iface)
return result
# Only return VLANs that are actually assigned to at least one interface.
# Linux bridge vlan show reports all 4094 possible VIDs per port —
# filtering here avoids polluting the VLAN table with phantom entries.
return {
vid: entry for vid, entry in result.items()
if entry.get("tagged") or entry.get("untagged")
}
# ------------------------------------------------------------------ #
# get_network_instances (SDN Zones as VRF-like instances)
@@ -2107,3 +2168,212 @@ class ProxmoxDriver(HypervisorDriver):
self._api.cluster.sdn.put()
except Exception:
pass # Best effort
# ------------------------------------------------------------------ #
# SNMP / Health
# ------------------------------------------------------------------ #
def get_snmp_config(self):
"""Return SNMP agent config if snmpd is installed and running on the node.
Uses _exec_ssh_command (Proxmox API exec or SSH) to inspect the node.
Returns a SNMPConfigDict or None.
"""
try:
from napalm_device_types.models import SNMPConfigDict
except ImportError:
return None
running = (
self._exec_ssh_command("systemctl is-active snmpd 2>/dev/null || true").strip()
== "active"
)
if not running:
return None
community = "public"
port = 161
try:
conf = self._exec_ssh_command(
"grep -E '^[[:space:]]*(ro|rw)?community' /etc/snmp/snmpd.conf 2>/dev/null | head -5"
)
for line in conf.splitlines():
parts = line.split()
if not parts:
continue
kw = parts[0].lower()
if kw in ("rocommunity", "rwcommunity") and len(parts) >= 2:
community = parts[1]
break
elif kw == "com2sec" and len(parts) >= 4:
community = parts[3]
break
except Exception:
pass
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()
raise NotImplementedError(f"Unknown action: {action!r}")
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: [].__class__ = []
# 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 = ""
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:
pass
# 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]}")
# 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}")
# 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"
)
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)}
def get_disk_smart(self) -> dict:
"""Return SMART health data for all disks on this Proxmox node.
Combines the disk list (model, size, wearout) with per-disk SMART
data (health, temperature, percentage used).
Returns a dict keyed by device path, e.g. {"/dev/nvme0n1": {...}}.
"""
import re as _re
result: dict = {}
try:
disks = self._node_api().disks.list.get() or []
except Exception:
return result
for disk in disks:
dev = disk.get("devpath") or disk.get("dev")
if not dev:
continue
entry: dict = {
"model": disk.get("model", ""),
"serial": disk.get("serial", ""),
"type": disk.get("type", ""),
"size": disk.get("size", 0),
"health": disk.get("health", "unknown").lower(),
"wearout": disk.get("wearout"), # NVMe wear indicator 0-100
"temperature": None,
"percentage_used": None,
"available_spare": None,
"reallocated_sectors": None,
"power_on_hours": None,
}
try:
smart = self._node_api().disks.smart.get(disk=dev) or {}
# health from SMART endpoint may be more accurate
if smart.get("health"):
entry["health"] = smart["health"].lower()
text = smart.get("text", "")
# Parse temperature
m = _re.search(r"Temperature[^:]*:\s*(\d+)\s*Celsius", text)
if m:
entry["temperature"] = int(m.group(1))
# NVMe-specific
m = _re.search(r"Percentage Used:\s*(\d+)%", text)
if m:
entry["percentage_used"] = int(m.group(1))
m = _re.search(r"Available Spare:\s*(\d+)%", text)
if m:
entry["available_spare"] = int(m.group(1))
m = _re.search(r"Power On Hours:\s*([\d,]+)", text)
if m:
entry["power_on_hours"] = int(m.group(1).replace(",", ""))
# HDD-specific SMART attributes
for attr in smart.get("attributes", []):
name = attr.get("name", "").lower()
raw = attr.get("raw", "")
try:
raw_int = int(str(raw).split()[0])
except (ValueError, TypeError):
raw_int = None
if "temperature" in name and raw_int is not None:
entry["temperature"] = raw_int
elif "reallocated" in name and "sector" in name and raw_int is not None:
entry["reallocated_sectors"] = raw_int
elif "power_on" in name and raw_int is not None:
entry["power_on_hours"] = raw_int
except Exception:
pass
result[dev] = entry
return result