Christian Manivong
3365207c98
- driver.py: the logger is defined after the imports (E402). The lazy paramiko client is annotated Any, as _node_ssh already is; the quoted paramiko.SSHClient named a module that is never imported there (F821). - Unused locals are gone: get_facts' hostname, get_users' groups (both read a value nobody used), the ARP flags column is _flags; the output is unchanged. - Long lines are wrapped: the apt-get and snmpd.conf commands are split into adjacent string literals, the power-action error message and the system mixin's docstring are wrapped; the text is unchanged. - get_vm_status re-raises with raise ... from e (B904). - The two except Exception: pass (closing the SSH session, the is_alive probe) now log at debug level (B110), so a swallowed failure leaves a trace. - nosec for the empty "password" field of NAPALM's get_users shape and the ssh_password_auth flag (B105); all three are false positives. - tests: unused loop variables.
512 lines
20 KiB
Python
512 lines
20 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
|
|
|
|
from napalm.base.exceptions import ConnectionException
|
|
from napalm_device_types import (
|
|
AptProxyMixin,
|
|
EnsurePackageMixin,
|
|
FingerprintRule,
|
|
HostStatusMixin,
|
|
HypervisorDriver,
|
|
KernelFactsMixin,
|
|
LicenseMixin,
|
|
ListeningSocketsMixin,
|
|
PortSpec,
|
|
ServicePackagesMixin,
|
|
SnmpAgentMixin,
|
|
SystemdServicesMixin,
|
|
SystemIdentityMixin,
|
|
)
|
|
from napalm_device_types.ssh_host_keys import HostKeyCheck, SshHostKeyMismatch
|
|
|
|
try:
|
|
from proxmoxer import ProxmoxAPI
|
|
except ImportError as exc:
|
|
raise ImportError("proxmoxer is required: pip install proxmoxer") from exc
|
|
|
|
from napalm_proxmox.config_mixin import ProxmoxConfigMixin
|
|
from napalm_proxmox.interfaces_mixin import ProxmoxInterfaceMixin
|
|
from napalm_proxmox.lldp_mixin import ProxmoxLLDPMixin
|
|
from napalm_proxmox.routing_mixin import ProxmoxRoutingMixin
|
|
from napalm_proxmox.sdn_mixin import ProxmoxSDNMixin
|
|
from napalm_proxmox.system_mixin import ProxmoxSystemMixin
|
|
from napalm_proxmox.vm_contract_mixin import ProxmoxVMContractMixin
|
|
from napalm_proxmox.vm_mixin import ProxmoxVMMixin
|
|
from napalm_proxmox.vm_provision_mixin import ProxmoxVMProvisionMixin
|
|
from napalm_proxmox.vm_snapshot_mixin import ProxmoxVMSnapshotMixin
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Type aliases
|
|
# --------------------------------------------------------------------------- #
|
|
_JsonDict = dict[str, Any]
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Driver
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
class ProxmoxDriver(
|
|
ProxmoxInterfaceMixin,
|
|
ProxmoxSDNMixin,
|
|
ProxmoxLLDPMixin,
|
|
ProxmoxConfigMixin,
|
|
ProxmoxVMMixin,
|
|
ProxmoxVMContractMixin,
|
|
ProxmoxVMSnapshotMixin,
|
|
ProxmoxVMProvisionMixin,
|
|
ProxmoxRoutingMixin,
|
|
ProxmoxSystemMixin,
|
|
KernelFactsMixin,
|
|
ServicePackagesMixin,
|
|
ListeningSocketsMixin,
|
|
SystemIdentityMixin,
|
|
SystemdServicesMixin,
|
|
HostStatusMixin,
|
|
LicenseMixin,
|
|
SnmpAgentMixin,
|
|
EnsurePackageMixin,
|
|
AptProxyMixin,
|
|
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
|
|
#: "Upgrade everything" must be a full upgrade on Proxmox VE: a plain
|
|
#: ``apt-get upgrade`` holds back what needs new or removed packages and can
|
|
#: leave the node half updated. netOrk reads this when it upgrades the host.
|
|
FULL_UPGRADE = True
|
|
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 = ""
|
|
# paramiko.SSHClient; paramiko is imported lazily (see _node_ssh).
|
|
self._ssh_client: Any = 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:
|
|
logger.debug("%s: closing the SSH session: ignored", self.hostname, exc_info=True)
|
|
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:
|
|
logger.debug("%s: liveness probe failed", self.hostname, exc_info=True)
|
|
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
|
|
|
|
# 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:
|
|
_, stdout, stderr = self._node_ssh().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 SshHostKeyMismatch as exc:
|
|
# Refused before any login, and already reported to the caller's store.
|
|
logger.warning("%s", exc)
|
|
except Exception as exc:
|
|
logger.debug("SSH exec command failed: %s", exc)
|
|
|
|
return ""
|
|
|
|
def _node_ssh(self) -> Any: # paramiko.SSHClient; paramiko is imported lazily
|
|
"""The SSH session to the node on port 22, opened once.
|
|
|
|
Its host key is the one the caller expects for port 22, or on first use
|
|
whichever the node presents; either way it is reported (netork#1167).
|
|
"""
|
|
import paramiko
|
|
|
|
if self._ssh_client is None:
|
|
ssh_pass = self._ssh_password or self.password
|
|
connect_kwargs: _JsonDict = {
|
|
"username": self._ssh_username or self.username,
|
|
"timeout": self.timeout,
|
|
}
|
|
if self._ssh_key and not ssh_pass:
|
|
from io import StringIO as _StringIO
|
|
|
|
connect_kwargs["pkey"] = paramiko.RSAKey.from_private_key(_StringIO(self._ssh_key))
|
|
else:
|
|
connect_kwargs["password"] = ssh_pass
|
|
client = paramiko.SSHClient()
|
|
try:
|
|
HostKeyCheck(self.optional_args, self.hostname, 22).connect(
|
|
client, **connect_kwargs
|
|
)
|
|
except Exception:
|
|
client.close()
|
|
raise
|
|
self._ssh_client = client
|
|
return self._ssh_client
|
|
|
|
# ------------------------------------------------------------------ #
|
|
# 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", "")
|
|
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}
|