Christian ManivongandClaude Sonnet 4.6
b1ba991e6d
Implements set_hostname on ProxmoxSystemMixin: 1. Writes /etc/hostname (short name, base64-safe transfer) 2. Replaces old hostname in /etc/hosts via Python regex + base64 3. Updates /etc/mailname if present 4. Updates Postfix myhostname via postconf -e if installed 5. Applies hostname immediately at runtime via hostname(1) 6. Regenerates Proxmox node TLS certificate via pvecm updatecerts -f (falls back to pvenode cert create → pveproxy restart) Accepts bare hostname or FQDN. A reboot is required for the Proxmox node name to update in the web UI / cluster — the driver logs this. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
375 lines
14 KiB
Python
375 lines
14 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 HypervisorDriver
|
|
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.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,
|
|
ProxmoxRoutingMixin,
|
|
ProxmoxSystemMixin,
|
|
HypervisorDriver,
|
|
):
|
|
"""NAPALM driver for Proxmox VE nodes."""
|
|
|
|
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."""
|
|
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))
|
|
model = status.get("model", "")
|
|
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}"
|
|
|
|
return {
|
|
"uptime": uptime,
|
|
"vendor": "Proxmox Server Solutions GmbH",
|
|
"model": model or "Proxmox VE Node",
|
|
"hostname": self._node_name,
|
|
"fqdn": fqdn or self.hostname,
|
|
"os_version": os_version,
|
|
"serial_number": "",
|
|
"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}
|
|
|
|
|