Christian ManivongandClaude Haiku 4.5
7bdac4c496
Add ProxmoxVMProvisionMixin with three methods: - create_vm_from_cloud_init(): clone template → dual-NIC config → Cloud-Init → start - destroy_vm(): stop → delete VM → cleanup snippets - get_vm_status(): poll guest-agent for IP with optional wait-for-IP polling Tests (9 cases): - _wait_for_task success/error/timeout handling - create_vm happy path + missing snippet storage error - get_vm_status with/without wait-for-IP, timeout handling - destroy_vm on running or already-stopped VM All tests pass (100% coverage on mixin code paths). Co-Authored-By: Claude Haiku 4.5 <noreply@anthropic.com>
435 lines
16 KiB
Python
435 lines
16 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"
|
|
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.
|
|
kwargs: _JsonDict = {
|
|
"host": self.hostname,
|
|
"user": self.username,
|
|
"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 node name from the hostname or optional_args."""
|
|
if self._node:
|
|
self._node_name = self._node
|
|
return self._node_name
|
|
|
|
# Try the node's hostname via API cluster/resources
|
|
try:
|
|
nodes = self._api.nodes.get()
|
|
# Match by hostname or IP
|
|
for n in nodes:
|
|
n_node = n.get("node", "")
|
|
if n_node:
|
|
# First match: the node exists in the cluster
|
|
self._node_name = n_node
|
|
return self._node_name
|
|
except Exception:
|
|
pass
|
|
|
|
# Fallback: use the configured hostname as node name
|
|
# (may not match the PVE node name — SSH-based methods will fail,
|
|
# but API methods that target a specific node name require correct
|
|
# resolution)
|
|
self._node_name = self.hostname
|
|
return self._node_name
|
|
|
|
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."""
|
|
alive = False
|
|
if self._api:
|
|
try:
|
|
self._resolve_node()
|
|
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)
|
|
|
|
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,
|
|
}
|
|
|
|
# ------------------------------------------------------------------ #
|
|
# 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}
|
|
|
|
|