Christian Manivong
abce85d6ef
_resolve_node() took the first entry of GET /nodes. In a cluster that lists every member, so a node polled without an explicit `node` driver argument talked to whichever member came first: pve-dual reported pve-02's name and VMs, and netOrk's VM sync moved pve-02's VM devices over to it. Resolve through GET /cluster/status instead: the entry marked local, then a match by IP or (short) name, then the sole node of a standalone host, and otherwise raise rather than guess. The lookup no longer swallows API errors either. A TLS verification failure used to leave the IP as the node name, so open() succeeded and every getter failed quietly while the poll reported success with empty data. It now surfaces as a ConnectionException from open(). Refs NetOrk/netork#417, NetOrk/netork#418
477 lines
18 KiB
Python
477 lines
18 KiB
Python
"""NAPALM driver for Proxmox VE.
|
|
|
|
Supports:
|
|
- Classic Linux networking (/etc/network/interfaces via Proxmox API)
|
|
- Software-Defined Networking (SDN): zones, VNets, subnets
|
|
- Open vSwitch (OVS) bridges, bonds, and internal ports
|
|
|
|
Connection is made via the Proxmox REST API (``proxmoxer`` library).
|
|
The driver targets the *node* level: each Proxmox node is treated as a
|
|
network device. Cluster-wide SDN information is also exposed where the
|
|
NAPALM API allows it.
|
|
|
|
Optional args
|
|
-------------
|
|
verify_ssl : bool
|
|
Verify TLS certificates (default: True).
|
|
port : int
|
|
Proxmox API port (default: 8006).
|
|
node : str
|
|
Override the target node name (default: auto-detected from hostname).
|
|
realm : str
|
|
PAM realm (default: ``pam``).
|
|
token_name : str
|
|
API token name (e.g. ``napalm@pam!mytoken``).
|
|
token_value : str
|
|
API token secret. When both token_name and token_value are provided,
|
|
token-based auth is used instead of password auth.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from typing import Any
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
from napalm_device_types import FingerprintRule, HypervisorDriver, PortSpec
|
|
from napalm.base.exceptions import ConnectionException
|
|
|
|
try:
|
|
from proxmoxer import ProxmoxAPI
|
|
except ImportError as exc:
|
|
raise ImportError(
|
|
"proxmoxer is required: pip install proxmoxer"
|
|
) from exc
|
|
|
|
from napalm_proxmox.interfaces_mixin import ProxmoxInterfaceMixin
|
|
from napalm_proxmox.sdn_mixin import ProxmoxSDNMixin
|
|
from napalm_proxmox.lldp_mixin import ProxmoxLLDPMixin
|
|
from napalm_proxmox.config_mixin import ProxmoxConfigMixin
|
|
from napalm_proxmox.vm_mixin import ProxmoxVMMixin
|
|
from napalm_proxmox.vm_provision_mixin import ProxmoxVMProvisionMixin
|
|
from napalm_proxmox.routing_mixin import ProxmoxRoutingMixin
|
|
from napalm_proxmox.system_mixin import ProxmoxSystemMixin
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Type aliases
|
|
# --------------------------------------------------------------------------- #
|
|
_JsonDict = dict[str, Any]
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Driver
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
class ProxmoxDriver(
|
|
ProxmoxInterfaceMixin,
|
|
ProxmoxSDNMixin,
|
|
ProxmoxLLDPMixin,
|
|
ProxmoxConfigMixin,
|
|
ProxmoxVMMixin,
|
|
ProxmoxVMProvisionMixin,
|
|
ProxmoxRoutingMixin,
|
|
ProxmoxSystemMixin,
|
|
HypervisorDriver,
|
|
):
|
|
"""NAPALM driver for Proxmox VE nodes."""
|
|
|
|
VENDOR = "Proxmox"
|
|
DRIVER_NAME = "proxmox"
|
|
# Everything runs over the Proxmox REST API; there is no SSH session.
|
|
USES_SSH = False
|
|
# A PVE node reboots through a full init sequence plus storage checks.
|
|
REBOOT_SETTLE_SECONDS = 90
|
|
PORT_SPECS = [
|
|
PortSpec("https", 8006, weight=8.0),
|
|
]
|
|
SSH_FINGERPRINT = [
|
|
FingerprintRule("debian", weight=3.0),
|
|
]
|
|
HTTP_FINGERPRINT = [
|
|
FingerprintRule("proxmox virtual environment", weight=9.0, mandatory=True),
|
|
FingerprintRule("proxmox", weight=5.0),
|
|
FingerprintRule("pve", weight=2.0),
|
|
]
|
|
platform = "proxmox"
|
|
|
|
def __init__(
|
|
self,
|
|
hostname: str,
|
|
username: str,
|
|
password: str,
|
|
timeout: int = 60,
|
|
optional_args: _JsonDict | None = None,
|
|
) -> None:
|
|
self.hostname = hostname
|
|
self.username = username
|
|
self.password = password
|
|
self.timeout = timeout
|
|
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", 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")
|
|
self._node: str | None = self.optional_args.get("node")
|
|
|
|
self._ssh_username: str | None = self.optional_args.get("ssh_username")
|
|
self._ssh_password: str | None = self.optional_args.get("ssh_password")
|
|
self._ssh_key: str | None = self.optional_args.get("ssh_private_key_str")
|
|
|
|
self._api: ProxmoxAPI | None = None
|
|
self._node_name: str = ""
|
|
self._ssh_client: "paramiko.SSHClient | None" = None
|
|
|
|
# Candidate config (merge/replace)
|
|
self._candidate_config: str = ""
|
|
self._running_config: str = ""
|
|
|
|
# ------------------------------------------------------------------ #
|
|
# Connection management
|
|
# ------------------------------------------------------------------ #
|
|
|
|
def open(self) -> None:
|
|
"""Open the connection to the Proxmox API."""
|
|
try:
|
|
# Use the HTTPS backend for proper REST API support.
|
|
# The openssh backend tunnels all kwargs through to
|
|
# openssh_wrapper.CommandBaseSession, which does not
|
|
# accept password/verify_ssl/token params.
|
|
# Proxmox authenticates against "<user>@<realm>" and rejects a bare
|
|
# username outright. A caller who typed a realm keeps it; one who
|
|
# did not gets self._realm, which is what the documented `realm`
|
|
# optional_arg is for -- it was read in __init__ and then never
|
|
# used, so the option had no effect and a bare username failed.
|
|
user = self.username or ""
|
|
if user and "@" not in user:
|
|
user = f"{user}@{self._realm}"
|
|
|
|
kwargs: _JsonDict = {
|
|
"host": self.hostname,
|
|
"user": user,
|
|
"password": self.password,
|
|
"port": self._port,
|
|
"verify_ssl": self._verify_ssl,
|
|
"backend": "https",
|
|
}
|
|
|
|
if self._token_name and self._token_value:
|
|
kwargs.pop("password", None)
|
|
kwargs["token_value"] = self._token_value
|
|
# token_name may be in "<user>!<tokenid>" format (e.g.
|
|
# "root@pam!netork"). proxmoxer expects them split:
|
|
# user="root@pam", token_name="netork"
|
|
user_part, _, token_id = self._token_name.partition("!")
|
|
if token_id:
|
|
kwargs["user"] = user_part
|
|
kwargs["token_name"] = token_id
|
|
else:
|
|
kwargs["token_name"] = self._token_name
|
|
|
|
self._api = ProxmoxAPI(**kwargs)
|
|
|
|
# Validate connection by fetching node status
|
|
self._resolve_node()
|
|
|
|
except Exception as exc:
|
|
raise ConnectionException(
|
|
f"Cannot connect to {self.hostname}: {exc}"
|
|
) from exc
|
|
|
|
def _resolve_node(self) -> str:
|
|
"""Resolve the PVE node this connection talks to.
|
|
|
|
In a cluster, ``GET /nodes`` lists every member, so its first entry is
|
|
just some node, not necessarily the one at ``self.hostname``. That used
|
|
to be taken blindly, and a device then reported another node's name and
|
|
VMs. ``GET /cluster/status`` marks the node the session landed on with
|
|
``local: 1``; failing that, the node is matched by IP or name.
|
|
|
|
API errors propagate so that ``open()`` fails with the real cause (e.g.
|
|
a TLS verification error) rather than succeeding on a guessed name.
|
|
"""
|
|
if self._node:
|
|
self._node_name = self._node
|
|
return self._node_name
|
|
|
|
members = [
|
|
s for s in (self._api.cluster.status.get() or []) if s.get("type") == "node"
|
|
]
|
|
short_host = self.hostname.split(".")[0]
|
|
for pick in (
|
|
lambda s: s.get("local"),
|
|
lambda s: s.get("ip") == self.hostname,
|
|
lambda s: s.get("name") in (self.hostname, short_host),
|
|
):
|
|
match = next((s for s in members if pick(s)), None)
|
|
if match:
|
|
self._node_name = match["name"]
|
|
return self._node_name
|
|
|
|
names = [n.get("node") for n in (self._api.nodes.get() or []) if n.get("node")]
|
|
if len(names) == 1:
|
|
self._node_name = names[0]
|
|
return self._node_name
|
|
|
|
raise ConnectionException(
|
|
f"Cannot tell which PVE node {self.hostname} is among {names}; "
|
|
f"set the 'node' driver argument"
|
|
)
|
|
|
|
def close(self) -> None:
|
|
"""Close the connection."""
|
|
self._api = None
|
|
if self._ssh_client:
|
|
try:
|
|
self._ssh_client.close()
|
|
except Exception:
|
|
pass
|
|
self._ssh_client = None
|
|
|
|
def is_alive(self) -> _JsonDict:
|
|
"""Return connection liveness.
|
|
|
|
Probes ``GET /version``, the cheapest endpoint that proves the session
|
|
still authenticates. It used to call ``_resolve_node()``, which returns
|
|
early without touching the API whenever a node was configured via
|
|
optional_args — so a dead connection reported itself alive.
|
|
"""
|
|
alive = False
|
|
if self._api:
|
|
try:
|
|
self._api.version.get()
|
|
alive = True
|
|
except Exception:
|
|
pass
|
|
return {"is_alive": alive}
|
|
|
|
# ------------------------------------------------------------------ #
|
|
# Internal API helpers
|
|
# ------------------------------------------------------------------ #
|
|
|
|
def _node_api(self):
|
|
"""Return the API resource for the current node."""
|
|
return self._api.nodes(self._node_name)
|
|
|
|
def _get_node_network(self) -> list[_JsonDict]:
|
|
"""Return the node's network interface list from the Proxmox API."""
|
|
try:
|
|
return self._node_api().network.get() or []
|
|
except Exception as exc:
|
|
logger.debug("Failed to fetch node network: %s", exc)
|
|
return []
|
|
|
|
def _exec_ssh_command(self, command: str) -> str:
|
|
"""Execute a shell command on the Proxmox node and return output.
|
|
|
|
Tries the Proxmox API exec endpoint first. If that fails,
|
|
falls back to a direct paramiko SSH connection.
|
|
"""
|
|
import base64 as _b64
|
|
import time as _time
|
|
|
|
# Try API exec endpoint
|
|
try:
|
|
encoded = _b64.b64encode(command.encode()).decode()
|
|
result = self._node_api().execute.post("command", f"echo {encoded} | base64 -d | sh")
|
|
# On PVE 8.x the exec endpoint returns a dict with 'data' key
|
|
if isinstance(result, dict):
|
|
raw = result.get("data", result.get("output", ""))
|
|
else:
|
|
raw = result
|
|
if raw:
|
|
return str(raw).strip()
|
|
except Exception as exc:
|
|
logger.debug("API exec failed, falling back to SSH: %s", exc)
|
|
|
|
# Fallback: direct paramiko SSH
|
|
try:
|
|
import paramiko
|
|
|
|
if self._ssh_client is None:
|
|
ssh_user = self._ssh_username or self.username
|
|
ssh_pass = self._ssh_password or self.password
|
|
ssh_pkey = None
|
|
if self._ssh_key and not ssh_pass:
|
|
from io import StringIO as _StringIO
|
|
ssh_pkey = paramiko.RSAKey.from_private_key(_StringIO(self._ssh_key))
|
|
self._ssh_client = paramiko.SSHClient()
|
|
self._ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
|
|
connect_kwargs: _JsonDict = {
|
|
"hostname": self.hostname,
|
|
"port": 22,
|
|
"username": ssh_user,
|
|
"timeout": self.timeout,
|
|
}
|
|
if ssh_pkey:
|
|
connect_kwargs["pkey"] = ssh_pkey
|
|
else:
|
|
connect_kwargs["password"] = ssh_pass
|
|
self._ssh_client.connect(**connect_kwargs)
|
|
_, stdout, stderr = self._ssh_client.exec_command(command, timeout=self.timeout)
|
|
err = stderr.read().decode().strip()
|
|
out = stdout.read().decode().strip()
|
|
return out or err
|
|
except ImportError:
|
|
logger.warning("paramiko not installed — cannot exec SSH commands")
|
|
except Exception as exc:
|
|
logger.debug("SSH exec command failed: %s", exc)
|
|
|
|
return ""
|
|
|
|
# ------------------------------------------------------------------ #
|
|
# Node info helpers
|
|
# ------------------------------------------------------------------ #
|
|
|
|
def _get_version_info(self) -> _JsonDict:
|
|
"""Return Proxmox VE version info from the API."""
|
|
try:
|
|
return self._api.version.get() or {}
|
|
except Exception as exc:
|
|
logger.debug("Failed to fetch version info: %s", exc)
|
|
return {}
|
|
|
|
def _get_node_status(self) -> _JsonDict:
|
|
"""Return the node's status from the Proxmox API."""
|
|
try:
|
|
return self._node_api().status.get() or {}
|
|
except Exception as exc:
|
|
logger.debug("Failed to fetch node status: %s", exc)
|
|
return {}
|
|
|
|
def _get_node_subscription(self) -> _JsonDict:
|
|
"""Return subscription status for this node."""
|
|
try:
|
|
return self._node_api().subscription.get() or {}
|
|
except Exception as exc:
|
|
logger.debug("Failed to fetch node subscription: %s", exc)
|
|
return {}
|
|
|
|
def _get_node_dns(self) -> _JsonDict:
|
|
"""Return DNS configuration for this node."""
|
|
try:
|
|
return self._node_api().dns.get() or {}
|
|
except Exception as exc:
|
|
logger.debug("Failed to fetch node DNS: %s", exc)
|
|
return {}
|
|
|
|
def _get_node_time(self) -> _JsonDict:
|
|
"""Return time configuration for this node."""
|
|
try:
|
|
return self._node_api().time.get() or {}
|
|
except Exception as exc:
|
|
logger.debug("Failed to fetch node time: %s", exc)
|
|
return {}
|
|
|
|
def _get_node_ntp(self) -> _JsonDict:
|
|
"""Return NTP configuration for this node."""
|
|
try:
|
|
return self._node_api().ntp.get() or {}
|
|
except Exception as exc:
|
|
logger.debug("Failed to fetch node NTP: %s", exc)
|
|
return {}
|
|
|
|
# ------------------------------------------------------------------ #
|
|
# NAPALM getters kept in driver
|
|
# ------------------------------------------------------------------ #
|
|
|
|
def get_facts(self) -> _JsonDict:
|
|
"""Return basic facts about the Proxmox node.
|
|
|
|
Hardware vendor, model and serial are read from the Linux DMI sysfs
|
|
entries (``/sys/class/dmi/id/``) via SSH so they reflect the physical
|
|
machine, not the Proxmox software layer.
|
|
"""
|
|
status = self._get_node_status()
|
|
version = self._get_version_info()
|
|
network = self._get_node_network()
|
|
dns = self._get_node_dns()
|
|
|
|
uptime = float(status.get("uptime", 0))
|
|
dns_search = dns.get("search", "")
|
|
hostname = self._node_name
|
|
fqdn = f"{self._node_name}.{dns_search}" if dns_search else self.hostname
|
|
|
|
iface_list = sorted(
|
|
iface["iface"] for iface in network if iface.get("iface")
|
|
)
|
|
|
|
pve_version = version.get("version", "")
|
|
release = version.get("release", "")
|
|
os_version = f"Proxmox VE {pve_version}" if pve_version else f"Proxmox VE {release}"
|
|
|
|
# Physical hardware info from Linux DMI sysfs.
|
|
# Read each field separately to avoid shell quoting issues with printf.
|
|
# Field priority for model:
|
|
# product_name — human-readable name on most vendors (e.g. "NUC6CAYH",
|
|
# "ThinkCentre M910x")
|
|
# product_version — sometimes the marketing name on Lenovo; on Intel NUC
|
|
# it is the board part number (less useful as model name)
|
|
# We prefer product_name; fall back to product_version only when
|
|
# product_name looks like a raw type code (all uppercase + digits, no spaces).
|
|
vendor = ""
|
|
model = ""
|
|
serial = ""
|
|
try:
|
|
dmi_cmd = (
|
|
"v=$(cat /sys/class/dmi/id/sys_vendor 2>/dev/null); "
|
|
"n=$(cat /sys/class/dmi/id/product_name 2>/dev/null); "
|
|
"r=$(cat /sys/class/dmi/id/product_version 2>/dev/null); "
|
|
"s=$(cat /sys/class/dmi/id/product_serial 2>/dev/null); "
|
|
"printf '%s\\n%s\\n%s\\n%s\\n' \"$v\" \"$n\" \"$r\" \"$s\""
|
|
)
|
|
lines = self._exec_ssh_command(dmi_cmd).splitlines()
|
|
if len(lines) >= 4:
|
|
vendor = lines[0].strip()
|
|
product_name = lines[1].strip()
|
|
product_version = lines[2].strip()
|
|
serial = lines[3].strip()
|
|
|
|
_bad = {"none", "n/a", "not specified", "to be filled by o.e.m."}
|
|
pv_usable = (
|
|
product_version
|
|
and product_version.lower() not in _bad
|
|
and product_version != product_name
|
|
# Only prefer product_version when it contains a space —
|
|
# that indicates a human-readable marketing name like
|
|
# "ThinkCentre M910x" rather than a part code like "J26843-409".
|
|
and " " in product_version
|
|
)
|
|
model = product_version if pv_usable else product_name
|
|
except Exception as exc:
|
|
logger.debug("Failed to read DMI info via SSH: %s", exc)
|
|
|
|
# Currently-booted kernel release (uname -r) — distinct from any newer
|
|
# kernel that is merely installed and pending a reboot.
|
|
running_kernel = ""
|
|
try:
|
|
running_kernel = self._exec_ssh_command("uname -r").strip()
|
|
except Exception as exc:
|
|
logger.debug("Failed to read running kernel via SSH: %s", exc)
|
|
|
|
return {
|
|
"uptime": uptime,
|
|
"vendor": vendor or "Proxmox Server Solutions GmbH",
|
|
"model": model or status.get("model") or "Proxmox VE Node",
|
|
"hostname": self._node_name,
|
|
"fqdn": fqdn or self.hostname,
|
|
"os_version": os_version,
|
|
"serial_number": serial,
|
|
"interface_list": iface_list,
|
|
"running_kernel": running_kernel,
|
|
}
|
|
|
|
# ------------------------------------------------------------------ #
|
|
# CLI passthrough
|
|
# ------------------------------------------------------------------ #
|
|
|
|
def cli(self, commands: list[str], encoding: str = "text") -> dict[str, str]:
|
|
"""Execute a list of shell commands and return their outputs."""
|
|
return {cmd: self._exec_ssh_command(cmd) for cmd in commands}
|
|
|
|
|