Author SHA1 Message Date
Christian ManivongandClaude Haiku 4.5 ae23208eac test(vm_provision): cover generic NIC config, dual-NIC trunk, disk resize
Test cases: single/dual NIC with VLAN tags or trunk config, per-NIC DHCP control,
disk resize parameter, snippet storage validation, IP wait timeout, destroy paths.
All TDD cases green.

Co-Authored-By: Claude Haiku 4.5 <noreply@anthropic.com>
2026-07-07 00:00:09 +02:00
Christian Manivong d2c361937e feat(vm_provision_mixin): generalize create_vm_from_cloud_init for arbitrary NIC configs
- Replace fixed mgmt/capture dual-NIC parameters with generic nics list
- Loop over NICs to build net0, net1, ... config strings (access VLAN or trunk)
- Remove hardcoded 'ip link set eth1 up' Cloud-Init hack (caller responsibility)
- Add disk_resize_gb parameter for post-clone disk expansion (scsi0/virtio0/ide0/sata0)
- Make DHCP configuration per-NIC with sensible defaults (primary NIC only)
- Update docstrings and logging to reflect generic NIC architecture
2026-07-06 23:29:33 +02:00
Christian ManivongandClaude Haiku 4.5 7bdac4c496 feat(provisioning): implement VM provisioning mixin for Proxmox
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>
2026-07-06 22:03:41 +02:00
Christian Manivong ede2a97770 fix(deps): pin paramiko>=5.0.0 (CVE-2026-44405) 2026-07-02 12:22:55 +02:00
Christian ManivongandClaude Sonnet 4.6 243e66a893 feat: Fingerprint-Attribute für Discovery-Scoring
Ergänzt DRIVER_NAME, HTTP_FINGERPRINT, SNMP_FINGERPRINT, SSH_FINGERPRINT,
PORT_SPECS und SNMP_OBJECT_ID_PREFIX gemäß docs/DISCOVERY_FINGERPRINTING.md.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-06-24 14:47:08 +02:00
Christian ManivongandClaude Sonnet 4.6 08d02d285a feat: collect individual disk config and onboot flag per VM
_get_vm_disk_and_boot() now returns a list of disk dicts (name, size_mb)
instead of a single total. Each disk entry is read from the VM config
via /qemu/{vmid}/config or /lxc/{vmid}/config. The onboot flag is also
read from the same config endpoint.

Both QEMU VMs and LXC containers are covered. The 'disks' and 'onboot'
keys are added to every entry in the vms_snapshot.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-06-23 00:44:17 +02:00
Christian ManivongandClaude Sonnet 4.6 50ccf654ba fix: prefer product_version only when it contains a space (marketing name)
product_version is used as model only when it contains a space, indicating
a human-readable marketing name (e.g. "ThinkCentre M910x"). Part numbers
like "J26843-409" have no space and are skipped — product_name is used
instead (e.g. "NUC6CAYH" for Intel NUC).

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-06-22 22:09:06 +02:00
Christian ManivongandClaude Sonnet 4.6 d9b3ac12ea fix: read DMI fields separately to avoid shell quoting issues
The combined printf approach silently produced empty values when
product_name/version contained special chars or the shell split tokens
incorrectly. Read each /sys/class/dmi/id/ file via a separate cat,
collect lines, then apply vendor-specific model name selection:

- Intel NUC: product_name='NUC6CAYH' (marketing) preferred over
  product_version='J26843-409' (part number)
- Lenovo: product_name='10MYS03U00' (type code, all-caps+digits) →
  prefer product_version='ThinkCentre M910x' (marketing name)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-06-22 22:02:24 +02:00
Christian ManivongandClaude Sonnet 4.6 a3b0414d99 fix: prefer product_version over product_name for DMI model name
On Lenovo (and some other vendors) product_name contains the machine-type
code (e.g. "10MYS03U00") while product_version holds the marketing name
(e.g. "ThinkCentre M910x"). Read both and prefer product_version when it
is set and different from product_name.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-06-22 20:08:54 +02:00
Christian ManivongandClaude Sonnet 4.6 9631275d6b fix: get_facts() reads physical hardware info from DMI sysfs via SSH
vendor, model and serial_number now come from /sys/class/dmi/id/
(sys_vendor, product_name, product_serial) via SSH, reflecting the
actual physical server rather than the Proxmox software layer.

Falls back to "Proxmox Server Solutions GmbH" / status.model if SSH
or DMI files are unavailable (e.g. bare-metal without SSH creds).

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-06-22 20:03:14 +02:00
Christian ManivongandClaude Sonnet 4.6 f288c01629 feat: add get_system_config() — cluster name, timezone, node list
Implements get_system_config() in ProxmoxSystemMixin:
- cluster_name from /cluster/status (type=cluster entry)
- cluster_nodes list of online node hostnames
- timezone from /nodes/{node}/time
- hostname, ssh_port, ssh_password_auth with safe defaults

Used by NetOrk's sync task to name the NetBox Cluster after the
actual Proxmox cluster rather than falling back to the site name.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-06-22 19:47:38 +02:00
Christian ManivongandClaude Sonnet 4.6 f6e694f028 fix: migrate /etc/pve/nodes/<old>/ before reboot in set_hostname
After renaming /etc/hostname, Proxmox boots under the new node name and
looks for VM/CT configs in /etc/pve/nodes/<new>/qemu-server/. Without
migrating the directory first, all VMs appear missing after the reboot.

Now renames /etc/pve/nodes/<old>/ to /etc/pve/nodes/<new>/ while
pve-cluster is running (pmxcfs supports live rename). Also replaces
pvecm updatecerts -f with pvenode cert create --overwrite — pvecm
updatecerts restarts pve-cluster, unmounts /etc/pve and can crash VMs.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-06-19 21:15:30 +02:00
Christian ManivongandClaude Sonnet 4.6 054a866d37 fix: add _send_command delegating to _exec_ssh_command
netOrk's generic reboot helper (_do_reboot) calls conn._send_command()
which did not exist on ProxmoxDriver — the resulting AttributeError was
silently swallowed, so reboot commands never executed on Proxmox nodes.
Delegates to the existing _exec_ssh_command so all generic SSH-based
actions (reboot, dns_port check, etc.) work correctly.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-06-19 16:39:58 +02:00
Christian ManivongandClaude Sonnet 4.6 b1ba991e6d feat: set_hostname — update /etc/hostname, /etc/hosts, cert, Postfix
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>
2026-06-19 16:30:45 +02:00
12 changed files with 3418 additions and 2279 deletions
+145
View File
@@ -0,0 +1,145 @@
# Copyright 2025 The NetOrk Project Authors. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Configuration management NAPALM methods for Proxmox VE nodes."""
from __future__ import annotations
import re
from typing import Any
_JsonDict = dict[str, Any]
class ProxmoxConfigMixin:
"""Mixin providing configuration-management NAPALM methods."""
def get_config(
self,
retrieve: str = "all",
full: bool = False,
sanitized: bool = False,
format: str = "text",
) -> _JsonDict:
"""Return the node network configuration.
``running`` config is the contents of ``/etc/network/interfaces``
(and the SDN config directory). ``startup`` is identical (PVE
applies on boot). ``candidate`` is what was loaded via
``load_merge_candidate`` / ``load_replace_candidate`` but not yet
committed.
"""
configs: _JsonDict = {"running": "", "candidate": "", "startup": ""}
if retrieve in ("running", "all", "startup"):
raw = self._exec_ssh_command(
"cat /etc/network/interfaces 2>/dev/null || true"
)
# Append SDN config if available
sdn_raw = self._exec_ssh_command(
"cat /etc/pve/sdn/vnets.cfg 2>/dev/null || true"
)
running = raw
if sdn_raw:
running += "\n# === SDN VNets ===\n" + sdn_raw
if sanitized:
running = re.sub(r"password\s+\S+", "password ****", running)
configs["running"] = running
self._running_config = running
if retrieve == "startup":
configs["startup"] = running
elif retrieve == "all":
configs["startup"] = running
if retrieve in ("candidate", "all"):
configs["candidate"] = self._candidate_config
return configs
def load_merge_candidate(
self,
filename: str | None = None,
config: str | None = None,
) -> None:
"""Load a candidate configuration (merge mode)."""
if filename:
with open(filename) as fh:
config = fh.read()
if config is None:
raise ValueError("Either filename or config must be provided")
# In merge mode we append / overlay
self._candidate_config = config
def load_replace_candidate(
self,
filename: str | None = None,
config: str | None = None,
) -> None:
"""Load a candidate configuration (replace mode)."""
if filename:
with open(filename) as fh:
config = fh.read()
if config is None:
raise ValueError("Either filename or config must be provided")
self._candidate_config = config
def compare_config(self) -> str:
"""Return a unified diff between running and candidate config."""
import difflib
if not self._running_config:
self.get_config(retrieve="running")
running_lines = self._running_config.splitlines(keepends=True)
candidate_lines = self._candidate_config.splitlines(keepends=True)
diff = difflib.unified_diff(
running_lines,
candidate_lines,
fromfile="running",
tofile="candidate",
)
return "".join(diff)
def commit_config(self, message: str = "", revert_in: int | None = None) -> None:
"""Commit the candidate configuration to the Proxmox node.
This writes the candidate config to ``/etc/network/interfaces``
via the Proxmox node/network PUT API (which applies it live).
.. note::
Full programmatic apply requires the Proxmox API to accept raw
interface configs. This implementation uses ``pvesh`` via exec
which requires the node exec endpoint to be available.
"""
if not self._candidate_config:
return
# Write via exec endpoint
escaped = self._candidate_config.replace("'", "'\\''")
self._exec_ssh_command(
f"printf '%s' '{escaped}' > /etc/network/interfaces && "
"ifreload -a 2>&1 || ifup -a 2>&1 || true"
)
self._running_config = self._candidate_config
self._candidate_config = ""
def discard_config(self) -> None:
"""Discard the candidate configuration."""
self._candidate_config = ""
def rollback(self) -> None:
"""Revert to the stored running configuration."""
if self._running_config:
self._candidate_config = self._running_config
self.commit_config()
self._candidate_config = ""
+225 -2257
View File
File diff suppressed because it is too large Load Diff
+226
View File
@@ -0,0 +1,226 @@
# Copyright 2025 The NetOrk Authors
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Interface-related NAPALM getters for Proxmox VE nodes."""
from __future__ import annotations
import logging
from typing import Any
from napalm_proxmox import utils
logger = logging.getLogger(__name__)
_JsonDict = dict[str, Any]
class ProxmoxInterfaceMixin:
"""Mixin providing interface-related NAPALM getters."""
def get_interfaces(self) -> dict[str, _JsonDict]:
"""Return a dict of interfaces keyed by interface name."""
result: dict[str, _JsonDict] = {}
for iface in self._get_node_network():
name = iface.get("iface", "")
if not name:
continue
active = iface.get("active", 0)
autostart = iface.get("autostart", 0)
result[name] = {
"is_up": bool(active),
"is_enabled": bool(autostart) or bool(active),
"description": iface.get("comments", "").strip(),
"last_flapped": -1.0,
"speed": utils.speed_mbps(iface),
"mtu": int(iface.get("mtu") or 1500),
"mac_address": utils.normalize_mac(iface.get("hwaddr", "")),
}
return result
def get_interfaces_ip(self) -> dict[str, _JsonDict]:
"""Return IP addresses per interface."""
result: dict[str, _JsonDict] = {}
for iface in self._get_node_network():
name = iface.get("iface", "")
if not name:
continue
addrs = utils.addresses_from_node_network(iface)
if addrs:
result[name] = addrs
for vnet in self._get_sdn_vnets():
vnet_id = vnet.get("vnet", "")
if not vnet_id:
continue
for subnet in self._get_sdn_subnets(vnet_id):
cidr = subnet.get("cidr", "")
gateway = subnet.get("gateway", "")
if gateway and cidr:
ip, plen = utils.parse_cidr(cidr)
if "." in gateway:
result.setdefault(vnet_id, {}).setdefault("ipv4", {})[gateway] = {
"prefix_length": plen
}
else:
result.setdefault(vnet_id, {}).setdefault("ipv6", {})[gateway] = {
"prefix_length": plen
}
return result
def get_interfaces_counters(self) -> dict[str, _JsonDict]:
"""Return per-interface traffic counters."""
result: dict[str, _JsonDict] = {}
try:
rrd_data = self._node_api().netstat.get() or []
except Exception as exc:
logger.debug("Failed to fetch netstat counters: %s", exc)
rrd_data = []
latest: dict[str, _JsonDict] = {}
for entry in rrd_data:
iface = entry.get("dev", "")
if iface:
latest[iface] = entry
for iface, data in latest.items():
result[iface] = {
"tx_errors": int(data.get("tx_errs", 0) or 0),
"rx_errors": int(data.get("rx_errs", 0) or 0),
"tx_discards": int(data.get("tx_drop", 0) or 0),
"rx_discards": int(data.get("rx_drop", 0) or 0),
"tx_octets": int(data.get("tx_bytes", 0) or 0),
"rx_octets": int(data.get("rx_bytes", 0) or 0),
"tx_unicast_packets": int(data.get("tx_packets", 0) or 0),
"rx_unicast_packets": int(data.get("rx_packets", 0) or 0),
"tx_multicast_packets": 0,
"rx_multicast_packets": 0,
"tx_broadcast_packets": 0,
"rx_broadcast_packets": 0,
}
return result
def get_arp_table(self, vrf: str = "") -> list[_JsonDict]:
"""Return ARP table.
Proxmox does not expose ARP via the REST API directly. We attempt
to read it via the node's ``/proc/net/arp`` through the Proxmox
exec endpoint. If that is unavailable, an empty list is returned.
"""
raw = self._exec_ssh_command("cat /proc/net/arp")
if not raw:
return []
entries = []
for line in raw.splitlines():
line = line.strip()
if not line or line.startswith("IP"):
continue
parts = line.split()
if len(parts) < 6:
continue
ip_addr, _, flags, mac, _, iface = (
parts[0], parts[1], parts[2], parts[3], parts[4], parts[5]
)
if mac in ("00:00:00:00:00:00", ""):
continue
if vrf and iface != vrf:
continue
entries.append(
{
"interface": iface,
"mac": utils.normalize_mac(mac),
"ip": ip_addr,
"age": -1.0,
}
)
return entries
def get_mac_address_table(self) -> list[_JsonDict]:
"""Return MAC address table from Linux bridges and OVS bridges."""
result: list[_JsonDict] = []
network = self._get_node_network()
linux_bridges = [
iface["iface"]
for iface in network
if iface.get("type") in ("bridge",) and iface.get("iface")
]
for bridge in linux_bridges:
raw = self._exec_ssh_command(
f"bridge fdb show br {bridge} 2>/dev/null || true"
)
for line in raw.splitlines():
parts = line.split()
if len(parts) < 3:
continue
mac_str = parts[0]
if not re.match(r"([0-9a-f]{2}:){5}[0-9a-f]{2}", mac_str):
continue
dev = ""
vlan_id = 1
for i, tok in enumerate(parts):
if tok == "dev" and i + 1 < len(parts):
dev = parts[i + 1]
if tok == "vlan" and i + 1 < len(parts):
try:
vlan_id = int(parts[i + 1])
except ValueError:
pass
result.append(
{
"mac": utils.normalize_mac(mac_str),
"interface": dev or bridge,
"vlan": vlan_id,
"static": "permanent" in line,
"active": True,
"moves": 0,
"last_move": 0.0,
}
)
ovs_bridges = [
iface["iface"]
for iface in network
if iface.get("type") in ("OVSBridge",) and iface.get("iface")
]
for bridge in ovs_bridges:
raw = self._exec_ssh_command(
f"ovs-appctl fdb/show {bridge} 2>/dev/null || true"
)
for line in raw.splitlines():
parts = line.split()
if len(parts) < 4:
continue
try:
_port = int(parts[0])
vlan_id = int(parts[1])
mac_str = parts[2]
except (ValueError, IndexError):
continue
result.append(
{
"mac": utils.normalize_mac(mac_str),
"interface": bridge,
"vlan": vlan_id,
"static": False,
"active": True,
"moves": 0,
"last_move": 0.0,
}
)
return result
+107
View File
@@ -0,0 +1,107 @@
# Copyright 2025 The NetOrk Authors
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""LLDP-related NAPALM getters for Proxmox VE nodes."""
from __future__ import annotations
import re
from typing import Any
_JsonDict = dict[str, Any]
class ProxmoxLLDPMixin:
"""Mixin providing LLDP NAPALM methods."""
def get_lldp_neighbors(self) -> dict[str, list[_JsonDict]]:
"""Return LLDP neighbours (requires lldpd on the Proxmox node)."""
result: dict[str, list[_JsonDict]] = {}
raw = self._exec_ssh_command(
"lldpcli show neighbors summary 2>/dev/null || true"
)
current_iface = ""
for line in raw.splitlines():
m_iface = re.match(r"^\s*Interface:\s+(\S+?),?\s", line)
if m_iface:
current_iface = m_iface.group(1)
result.setdefault(current_iface, [])
continue
m_sys = re.match(r"^\s*SysName:\s+(.+)", line)
m_port = re.match(r"^\s*PortID:\s+\S+\s+(.+)", line)
if m_sys and current_iface:
hostname = m_sys.group(1).strip()
if result[current_iface]:
result[current_iface][-1]["hostname"] = hostname
else:
result[current_iface].append({"hostname": hostname, "port": ""})
if m_port and current_iface and result[current_iface]:
result[current_iface][-1]["port"] = m_port.group(1).strip()
return result
def get_lldp_neighbors_detail(self, interface: str = "") -> dict[str, list[_JsonDict]]:
"""Return detailed LLDP neighbour information."""
result: dict[str, list[_JsonDict]] = {}
raw = self._exec_ssh_command(
"lldpcli show neighbors details 2>/dev/null || true"
)
current_iface = ""
current_entry: _JsonDict = {}
def _flush():
if current_iface and current_entry:
result.setdefault(current_iface, []).append(current_entry.copy())
for line in raw.splitlines():
m_iface = re.match(r"^\s*Interface:\s+(\S+?),?\s", line)
if m_iface:
_flush()
current_iface = m_iface.group(1)
if interface and current_iface != interface:
current_iface = ""
current_entry = {
"parent_interface": "",
"remote_chassis_id": "",
"remote_system_name": "",
"remote_port": "",
"remote_port_description": "",
"remote_system_description": "",
"remote_system_capab": [],
"remote_system_enable_capab": [],
}
continue
if not current_iface:
continue
for key, pattern in (
("remote_chassis_id", r"ChassisID:\s+\S+\s+(.+)"),
("remote_system_name", r"SysName:\s+(.+)"),
("remote_port", r"PortID:\s+\S+\s+(.+)"),
("remote_port_description", r"PortDescr:\s+(.+)"),
("remote_system_description", r"SysDescr:\s+(.+)"),
):
m = re.match(rf"^\s*{pattern}", line)
if m:
current_entry[key] = m.group(1).strip()
m_cap = re.match(r"^\s*Capability:\s+(\S+),\s+(\w+)", line)
if m_cap:
cap = m_cap.group(1).lower()
enabled = m_cap.group(2).lower() == "on"
current_entry["remote_system_capab"].append(cap)
if enabled:
current_entry["remote_system_enable_capab"].append(cap)
_flush()
return result
+225
View File
@@ -0,0 +1,225 @@
# Copyright 2025 The NetOrk Project Authors. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Routing-related NAPALM getters for Proxmox VE nodes."""
from __future__ import annotations
import re
from typing import Any
import napalm.base.constants as C
_JsonDict = dict[str, Any]
class ProxmoxRoutingMixin:
"""Mixin providing routing NAPALM methods (get_route_to, ping, traceroute)."""
def ping(
self,
destination: str,
source: str = C.PING_SOURCE,
ttl: int = C.PING_TTL,
timeout: int = C.PING_TIMEOUT,
size: int = C.PING_SIZE,
count: int = C.PING_COUNT,
vrf: str = C.PING_VRF,
source_interface: str = C.PING_SOURCE_INTERFACE,
) -> _JsonDict:
"""Execute ping on the Proxmox node and return results."""
cmd_parts = [
f"ping -c {count}",
f"-W {timeout}",
f"-s {size}",
f"-t {ttl}",
]
if source:
cmd_parts.append(f"-I {source}")
elif source_interface:
cmd_parts.append(f"-I {source_interface}")
cmd_parts.append(destination)
cmd = " ".join(cmd_parts)
raw = self._exec_ssh_command(f"{cmd} 2>&1 || true")
if not raw:
return {"error": "Ping command not available via exec endpoint"}
# Detect common failure strings before parsing statistics
_error_patterns = (
"Name or service not known",
"Network is unreachable",
"connect: No route to host",
"unknown host",
)
for _pat in _error_patterns:
if _pat.lower() in raw.lower():
return {"error": raw.strip()}
# Parse statistics line: "5 packets transmitted, 5 received, 0% packet loss"
m_stat = re.search(
r"(\d+) packets transmitted,\s+(\d+) received.*?([\d.]+)% packet loss",
raw,
)
if not m_stat:
return {"error": raw.strip()}
sent = int(m_stat.group(1))
received = int(m_stat.group(2))
loss = sent - received
# RTT line: "rtt min/avg/max/mdev = 0.123/0.456/0.789/0.100 ms"
m_rtt = re.search(
r"rtt min/avg/max/mdev = ([\d.]+)/([\d.]+)/([\d.]+)/([\d.]+)",
raw,
)
rtt_min = float(m_rtt.group(1)) if m_rtt else 0.0
rtt_avg = float(m_rtt.group(2)) if m_rtt else 0.0
rtt_max = float(m_rtt.group(3)) if m_rtt else 0.0
rtt_std = float(m_rtt.group(4)) if m_rtt else 0.0
# Individual probe lines
probes = []
for m_probe in re.finditer(
r"icmp_seq=\d+.*?time=([\d.]+) ms.*?from ([\d.a-fA-F:]+)", raw
):
probes.append(
{"ip_address": m_probe.group(2), "rtt": float(m_probe.group(1))}
)
return {
"success": {
"probes_sent": sent,
"packet_loss": loss,
"rtt_min": rtt_min,
"rtt_max": rtt_max,
"rtt_avg": rtt_avg,
"rtt_stddev": rtt_std,
"results": probes,
}
}
def traceroute(
self,
destination: str,
source: str = "",
ttl: int = 255,
timeout: int = 2,
vrf: str = "",
) -> _JsonDict:
"""Execute traceroute on the Proxmox node and return results."""
cmd_parts = [f"traceroute -m {ttl}", f"-w {timeout}", "-n"]
if source:
cmd_parts.append(f"-s {source}")
cmd_parts.append(destination)
raw = self._exec_ssh_command(" ".join(cmd_parts) + " 2>&1 || true")
if not raw:
return {"error": "traceroute not available via exec endpoint"}
hops: _JsonDict = {}
for line in raw.splitlines():
m = re.match(
r"^\s*(\d+)\s+([\d.a-fA-F:]+|\*)\s+([\d.]+|[\d.]+\s+ms|\*)",
line,
)
if not m:
continue
hop_id = int(m.group(1))
ip_addr = m.group(2)
if ip_addr == "*":
continue
# Parse RTT probes: each hop can have up to 3
rtts = re.findall(r"([\d.]+)\s+ms", line)
probes_dict = {}
for idx, rtt in enumerate(rtts, start=1):
probes_dict[idx] = {
"rtt": float(rtt),
"ip_address": ip_addr,
"host_name": ip_addr,
}
if probes_dict:
hops[hop_id] = {"probes": probes_dict}
if not hops:
return {"error": raw.strip()}
return {"success": hops}
def get_bgp_config(self, group: str = "", neighbor: str = "") -> _JsonDict:
raise NotImplementedError("BGP configuration is not managed via Proxmox API")
def get_bgp_neighbors(self) -> _JsonDict:
raise NotImplementedError("BGP is not managed via Proxmox API")
def get_bgp_neighbors_detail(self, neighbor_address: str = "") -> _JsonDict:
raise NotImplementedError("BGP is not managed via Proxmox API")
def get_route_to(
self, destination: str = "", protocol: str = "", longer: bool = False
) -> _JsonDict:
"""Return routing table entries for the given destination."""
cmd = f"ip route show {destination} 2>/dev/null || true"
raw = self._exec_ssh_command(cmd)
routes: _JsonDict = {}
for line in raw.splitlines():
line = line.strip()
if not line:
continue
parts = line.split()
if not parts:
continue
prefix = parts[0]
next_hop = ""
out_iface = ""
proto = "static"
for i, tok in enumerate(parts):
if tok == "via" and i + 1 < len(parts):
next_hop = parts[i + 1]
if tok == "dev" and i + 1 < len(parts):
out_iface = parts[i + 1]
if tok == "proto" and i + 1 < len(parts):
proto = parts[i + 1]
if protocol and protocol.lower() not in proto.lower():
continue
routes.setdefault(prefix, []).append(
{
"protocol": proto,
"current_active": True,
"last_active": True,
"age": -1,
"next_hop": next_hop,
"outgoing_interface": out_iface,
"selected_next_hop": True,
"preference": 1,
"inactive_reason": "",
"routing_table": "default",
"protocol_attributes": {},
}
)
return routes
def get_optics(self) -> _JsonDict:
raise NotImplementedError("Optics not available via Proxmox API")
def get_probes_config(self) -> _JsonDict:
raise NotImplementedError
def get_probes_results(self) -> _JsonDict:
raise NotImplementedError
def get_firewall_policies(self) -> _JsonDict:
raise NotImplementedError("Use the Proxmox firewall API directly")
+442
View File
@@ -0,0 +1,442 @@
# Copyright 2025 The NetOrk Authors
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""SDN, VLAN and network-instance NAPALM getters for Proxmox VE nodes."""
from __future__ import annotations
import logging
import re
from typing import Any
from napalm.base.exceptions import ConnectionException
from proxmoxer.core import ResourceException
from napalm_proxmox import utils
logger = logging.getLogger(__name__)
_JsonDict = dict[str, Any]
class ProxmoxSDNMixin:
"""Mixin providing SDN, VLAN and network-instance NAPALM methods."""
def _get_sdn_zones(self) -> list[_JsonDict]:
try:
return self._api.cluster.sdn.zones.get() or [] # type: ignore[union-attr]
except Exception as exc:
logger.debug("Failed to fetch SDN zones: %s", exc)
return []
def _get_sdn_vnets(self) -> list[_JsonDict]:
try:
return self._api.cluster.sdn.vnets.get() or [] # type: ignore[union-attr]
except Exception as exc:
logger.debug("Failed to fetch SDN VNets: %s", exc)
return []
def _get_sdn_subnets(self, vnet: str) -> list[_JsonDict]:
try:
return self._api.cluster.sdn.vnets(vnet).subnets.get() or []
except Exception as exc:
logger.debug("Failed to fetch SDN subnets for %s: %s", vnet, exc)
return []
def get_vlans(self) -> dict[str, _JsonDict]:
"""Return VLAN table.
For OVS+SDN nodes: reads SDN VNets for VLAN IDs/names, then maps
OVSIntPort (access ports with ovs_tag) → untagged membership, and
OVSPort / OVSBridge (trunk ports) → tagged membership.
Falls back to ``bridge vlan show`` for classic Linux-bridge nodes.
"""
result: dict[str, _JsonDict] = {}
node_network = self._get_node_network()
for vnet in self._get_sdn_vnets():
tag = vnet.get("tag")
vnet_id = vnet.get("vnet", "")
if tag is None:
continue
try:
tag_int = int(tag)
except (ValueError, TypeError):
continue
result[str(tag_int)] = {
"name": vnet_id,
"tagged": [],
"untagged": [],
}
trunk_ports: list[str] = []
access_by_vlan: dict[str, list[str]] = {}
for iface in node_network:
ovs_type = iface.get("ovs_type", "")
iface_name = iface.get("iface", "")
if not iface_name:
continue
if ovs_type == "OVSIntPort":
ovs_tag = iface.get("ovs_tag")
if ovs_tag is not None:
vid = str(int(ovs_tag))
access_by_vlan.setdefault(vid, []).append(iface_name)
elif ovs_type in ("OVSPort", "OVSBridge"):
trunk_ports.append(iface_name)
if trunk_ports or access_by_vlan:
for vid, vlan_entry in result.items():
vlan_entry["tagged"] = list(trunk_ports)
vlan_entry["untagged"] = list(access_by_vlan.get(vid, []))
return result
for entry in result.values():
entry.setdefault("tagged", [])
entry.setdefault("untagged", [])
for tag, bridges in self._get_vm_vlan_tags().items():
entry = result.setdefault(tag, {"name": "", "tagged": [], "untagged": []})
for bridge in bridges:
if bridge not in entry["untagged"]:
entry["untagged"].append(bridge)
return {
vid: entry for vid, entry in result.items()
if entry.get("tagged") or entry.get("untagged")
}
def _get_vm_vlan_tags(self) -> dict[str, set[str]]:
"""Return ``{vlan_tag: {bridge_names}}`` derived from VM/container net configs.
Scans every QEMU VM and LXC container on this node for ``netN`` config
entries of the form ``bridge=vmbrX,tag=N,...`` and groups the bridges
each VLAN tag is used on.
"""
tags: dict[str, set[str]] = {}
net_re = re.compile(r"^net\d+$")
def _collect(vmid: int, config: _JsonDict) -> None:
for key, val in config.items():
if not net_re.match(key):
continue
bridge = ""
tag: int | None = None
for part in str(val).split(","):
if "=" not in part:
continue
k, v = part.split("=", 1)
k = k.strip().lower()
if k == "bridge":
bridge = v.strip()
elif k == "tag":
try:
tag = int(v.strip())
except ValueError:
pass
if tag is not None and bridge:
tags.setdefault(str(tag), set()).add(bridge)
try:
for vm in (self._node_api().qemu.get() or []):
vmid = int(vm.get("vmid", 0))
try:
config = self._node_api().qemu(vmid).config.get() or {}
_collect(vmid, config)
except Exception as exc:
logger.debug("_get_vm_vlan_tags: QEMU %s config failed: %s", vmid, exc)
except Exception as exc:
logger.warning("_get_vm_vlan_tags: failed to list QEMU VMs: %s", exc)
try:
for ct in (self._node_api().lxc.get() or []):
vmid = int(ct.get("vmid", 0))
try:
config = self._node_api().lxc(vmid).config.get() or {}
_collect(vmid, config)
except Exception as exc:
logger.debug("_get_vm_vlan_tags: LXC %s config failed: %s", vmid, exc)
except Exception as exc:
logger.warning("_get_vm_vlan_tags: failed to list LXC containers: %s", exc)
return tags
def get_network_instances(self, name: str = "") -> dict[str, _JsonDict]:
"""Return SDN zones as network instances."""
result: dict[str, _JsonDict] = {}
result["default"] = {
"name": "default",
"type": "DEFAULT_INSTANCE",
"state": {"route_distinguisher": None},
"interfaces": {"interface": {}},
}
for iface in self._get_node_network():
iface_name = iface.get("iface", "")
if iface_name:
result["default"]["interfaces"]["interface"][iface_name] = {}
for zone in self._get_sdn_zones():
zone_id = zone.get("zone", zone.get("name", ""))
if not zone_id:
continue
if name and zone_id != name:
continue
instance = utils.sdn_zone_to_network_instance(zone)
for vnet in self._get_sdn_vnets():
if vnet.get("zone") == zone_id:
vnet_id = vnet.get("vnet", "")
if vnet_id:
instance["interfaces"]["interface"][vnet_id] = {}
result[zone_id] = instance
if name:
return {k: v for k, v in result.items() if k == name}
return result
def _is_physical_uplink(self, iface_name: str, network: dict) -> bool:
"""Return True if *iface_name* is a physical Ethernet port usable as uplink.
Rules:
- Must not match any known virtual interface name prefix.
- Must appear in the Proxmox node network config (runtime-only virtual
interfaces such as ``fwpr*`` or ``tap*`` will not be listed there).
- Must have a physical-compatible type:
- ``"eth"`` — regular physical NIC
- ``"OVSPort"`` — physical NIC attached directly to an OVS bridge
- ``""`` — untyped (e.g. OVS bond slave, still physical)
"""
if any(iface_name.startswith(p) for p in self._VIRTUAL_IFACE_PREFIXES):
return False
iface_info = network.get(iface_name)
if iface_info is None:
return False
return iface_info.get("type", "") in ("eth", "OVSPort", "")
def _find_switch_uplink(self) -> str | None:
"""Return the name of the physical interface connected to a switch.
Detection order:
1. LLDP detailed: physical port whose neighbour advertises Bridge
capability.
2. LLDP basic fallback: first physical port with any LLDP neighbour.
"""
network = {
iface["iface"]: iface
for iface in self._get_node_network()
if iface.get("iface")
}
try:
for iface_name, neighbour_list in self.get_lldp_neighbors_detail().items():
if not self._is_physical_uplink(iface_name, network):
continue
for nb in neighbour_list:
caps = nb.get("remote_system_capab", [])
if any("bridge" in str(c).lower() for c in caps):
return iface_name
except Exception as exc:
logger.debug("LLDP detailed neighbor discovery failed: %s", exc)
try:
for iface_name, neighbour_list in self.get_lldp_neighbors().items():
if self._is_physical_uplink(iface_name, network) and neighbour_list:
return iface_name
except Exception as exc:
logger.debug("LLDP basic neighbor discovery failed: %s", exc)
return None
def _get_ovs_bridge_for_port(self, port_name: str) -> str | None:
"""Return the OVS bridge name that *port_name* belongs to, or ``None``.
Checks (in order):
1. Port listed in an OVSBridge's ``ovs_ports``.
2. Port is a slave of an OVSBond which has an ``ovs_bridge`` reference.
3. Port itself carries an ``ovs_bridge`` field.
"""
network = self._get_node_network()
by_name: dict[str, _JsonDict] = {
iface["iface"]: iface for iface in network if iface.get("iface")
}
for iface in network:
if iface.get("type") == "OVSBridge":
ports = (iface.get("ovs_ports") or "").split()
if port_name in ports:
return iface["iface"]
for iface in network:
if iface.get("type") == "OVSBond":
slaves = (iface.get("slaves") or "").split()
if port_name in slaves:
bridge = iface.get("ovs_bridge", "")
if bridge:
return bridge
port_info = by_name.get(port_name, {})
return port_info.get("ovs_bridge") or None
def _is_cluster_master(self) -> bool:
"""Return ``True`` if this node is the Corosync quorum coordinator.
The coordinator is the online cluster node with the lowest ``nodeid``.
On standalone (non-clustered) nodes this always returns ``True``.
"""
try:
status = self._api.cluster.status.get() or []
node_entries = [e for e in status if e.get("type") == "node"]
if not node_entries:
return True
online_nodes = [n for n in node_entries if n.get("online", 0)]
if not online_nodes:
return True
min_id = min(int(n.get("nodeid", 9999)) for n in online_nodes)
for n in online_nodes:
if (
n.get("name") == self._node_name
and int(n.get("nodeid", 9999)) == min_id
):
return True
return False
except Exception as exc:
logger.warning("Cannot determine cluster status, assuming standalone: %s", exc)
return True
def _get_sdn_zone_for_bridge(self, bridge_name: str) -> str | None:
"""Return the SDN zone ID whose ``bridge`` field matches *bridge_name*.
In Proxmox SDN each zone is linked to exactly one OVS bridge via the
``bridge`` property. We look for that mapping so the VNet is always
created in the correct zone instead of guessing by type order.
Falls back to the first zone if no bridge match is found.
"""
try:
zones = self._get_sdn_zones()
for zone in zones:
if zone.get("bridge") == bridge_name:
return zone.get("zone")
if zones:
return zones[0].get("zone")
except Exception as exc:
logger.debug("Failed to get SDN zone for bridge: %s", exc)
return None
def set_vlan(self, vlan_id: int, config) -> None:
"""Create (or update) a VLAN via an SDN VNet on this Proxmox node.
Pre-flight checks (all must pass to proceed):
1. Finds the physical uplink port connected to a switch via LLDP.
Physical ports are those with type ``eth``, ``OVSPort``, or ``""``
in the Proxmox network config (excludes runtime virtuals like
``fwpr*``, ``tap*``, etc.).
2. Verifies that uplink is part of an OVS bridge or OVS bond.
3. Confirms this node is the Corosync quorum master (lowest node-id).
Non-master nodes return silently — the master handles VNet creation.
The SDN VNet is named ``vlan{vid:04d}`` (e.g. ``vlan0007`` for VID 7).
If the VNet already exists its alias is updated. After creating /
updating the VNet the SDN configuration is reloaded via
``PUT /cluster/sdn``.
Args:
vlan_id: VLAN identifier (1-4094).
config: Dict that may contain ``"name"`` for the VLAN alias.
"""
name: str = (
(config.get("name") or f"VLAN{vlan_id}") if config else f"VLAN{vlan_id}"
)
vnet_id = f"vlan{vlan_id:04d}"
uplink = self._find_switch_uplink()
if uplink is None:
raise ConnectionException(
f"set_vlan({vlan_id}): no LLDP-detected switch uplink found"
f" on node {self._node_name!r}"
)
ovs_bridge = self._get_ovs_bridge_for_port(uplink)
if ovs_bridge is None:
raise ConnectionException(
f"set_vlan({vlan_id}): uplink {uplink!r} is not part of an OVS bridge"
)
if not self._is_cluster_master():
return
zone = self._get_sdn_zone_for_bridge(ovs_bridge)
if not zone:
raise ConnectionException(
f"set_vlan({vlan_id}): no SDN zone found for bridge {ovs_bridge!r}"
)
try:
self._api.cluster.sdn.vnets.post(
vnet=vnet_id,
zone=zone,
tag=vlan_id,
alias=name,
)
except ResourceException as exc:
err_str = str(exc).lower()
if "already exists" in err_str or "duplicate" in err_str or "500" in err_str:
try:
self._api.cluster.sdn.vnets(vnet_id).put(alias=name)
except Exception as exc:
logger.debug("Best-effort VNet alias update failed: %s", exc)
else:
raise ConnectionException(
f"set_vlan({vlan_id}): failed to create VNet {vnet_id!r}: {exc}"
) from exc
try:
self._api.cluster.sdn.put()
except Exception as exc:
logger.debug("Best-effort SDN reload failed (may not be needed on older PVE): %s", exc)
def delete_vlan(self, vlan_id: int) -> None:
"""Delete the SDN VNet corresponding to *vlan_id*.
The VNet is identified by the canonical name ``vlan{vid:04d}``.
Only the Corosync quorum master performs the deletion — non-master
nodes return silently.
After deletion the SDN configuration is reloaded via
``PUT /cluster/sdn``.
Args:
vlan_id: VLAN identifier to delete.
"""
if not self._is_cluster_master():
return
vnet_id = f"vlan{vlan_id:04d}"
try:
self._api.cluster.sdn.vnets(vnet_id).delete()
except ResourceException as exc:
err_str = str(exc).lower()
if "does not exist" in err_str or "404" in str(exc):
return
raise ConnectionException(
f"delete_vlan({vlan_id}): failed to delete VNet {vnet_id!r}: {exc}"
) from exc
try:
self._api.cluster.sdn.put()
except Exception as exc:
logger.debug("Best-effort SDN reload after delete failed: %s", exc)
+866
View File
@@ -0,0 +1,866 @@
# Copyright 2025 The NetOrk Project Authors. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""System-level NAPALM getters for Proxmox VE nodes (environment, NTP, SNMP, users, packages, services, updates, disk SMART)."""
from __future__ import annotations
import logging
import re
from typing import Any
from napalm_proxmox import utils
logger = logging.getLogger(__name__)
_JsonDict = dict[str, Any]
class ProxmoxSystemMixin:
"""Mixin providing system-level NAPALM methods."""
# ------------------------------------------------------------------ #
# get_environment
# ------------------------------------------------------------------ #
def get_environment(self) -> _JsonDict:
"""Return environment status (CPU, memory, temperature)."""
status = self._get_node_status()
env: _JsonDict = {
"fans": {},
"temperature": {},
"power": {},
"cpu": {},
"memory": {"available_ram": 0, "used_ram": 0},
}
# CPU
cpu_usage = status.get("cpu", 0.0)
env["cpu"]["0"] = {"%usage": round(float(cpu_usage) * 100, 2)}
# Memory (Proxmox reports in bytes)
mem = status.get("memory", {})
total = int(mem.get("total", 0) or 0)
used = int(mem.get("used", 0) or 0)
env["memory"]["available_ram"] = total
env["memory"]["used_ram"] = used
# Temperature (from node sensors if available)
try:
sensors = self._node_api().hardware.sensors.get() or []
except Exception as exc:
logger.debug("Failed to fetch hardware sensors: %s", exc)
sensors = []
for sensor in sensors:
name = sensor.get("name", "unknown")
value = sensor.get("value", None)
if value is not None:
try:
temp_c = float(value)
env["temperature"][name] = {
"temperature": temp_c,
"is_alert": temp_c >= 80.0,
"is_critical": temp_c >= 95.0,
}
except (TypeError, ValueError):
pass
return env
# ------------------------------------------------------------------ #
# get_ntp_servers / get_ntp_stats
# ------------------------------------------------------------------ #
def get_ntp_servers(self) -> dict[str, _JsonDict]:
"""Return configured NTP servers."""
ntp = self._get_node_ntp()
servers: dict[str, _JsonDict] = {}
# Proxmox reports a comma-separated or space-separated server list
raw = ntp.get("server", "") or ntp.get("servers", "")
for srv in re.split(r"[\s,]+", raw):
srv = srv.strip()
if srv:
servers[srv] = {}
return servers
def get_ntp_stats(self) -> list[_JsonDict]:
"""Return NTP synchronisation statistics from chronyc/ntpq output."""
raw = self._exec_ssh_command(
"chronyc -n tracking 2>/dev/null || ntpq -pn 2>/dev/null || true"
)
stats: list[_JsonDict] = []
for line in raw.splitlines():
line = line.strip()
# ntpq -pn format: *remote refid st t when poll reach delay offset jitter
m = re.match(
r"^([\*\+\-\s])([\d.]+)\s+([\d.]+)\s+(\d+)\s+\S+\s+(\S+)\s+(\d+)\s+(\d+)\s+([\d.]+)\s+([-\d.]+)\s+([\d.]+)",
line,
)
if m:
synced = m.group(1).strip() == "*"
stats.append(
{
"remote": m.group(2),
"referenceid": m.group(3),
"synchronized": synced,
"stratum": int(m.group(4)),
"type": "",
"when": m.group(5),
"hostpoll": int(m.group(6)),
"reachability": int(m.group(7)),
"delay": float(m.group(8)),
"offset": float(m.group(9)),
"jitter": float(m.group(10)),
}
)
return stats
# ------------------------------------------------------------------ #
# get_snmp_information
# ------------------------------------------------------------------ #
def get_snmp_information(self) -> _JsonDict:
"""Return SNMP information.
Proxmox does not expose SNMP configuration via the REST API.
We read /etc/snmp/snmpd.conf via exec if available.
"""
raw = self._exec_ssh_command(
"cat /etc/snmp/snmpd.conf 2>/dev/null || true"
)
communities: dict[str, _JsonDict] = {}
location = ""
contact = ""
for line in raw.splitlines():
line = line.strip()
if line.startswith("#") or not line:
continue
# rocommunity <community> [source]
m = re.match(r"^(ro|rw)community\s+(\S+)", line)
if m:
mode = "ro" if m.group(1) == "ro" else "rw"
community = m.group(2)
communities[community] = {"acl": "N/A", "mode": mode}
m_loc = re.match(r"^sysLocation\s+(.+)", line)
if m_loc:
location = m_loc.group(1).strip()
m_con = re.match(r"^sysContact\s+(.+)", line)
if m_con:
contact = m_con.group(1).strip()
return {
"chassis_id": self._node_name,
"community": communities,
"contact": contact,
"location": location,
}
# ------------------------------------------------------------------ #
# get_users
# ------------------------------------------------------------------ #
def get_users(self) -> dict[str, _JsonDict]:
"""Return users configured on the Proxmox node.
Reads from both the Proxmox access/users API and local /etc/passwd.
"""
result: dict[str, _JsonDict] = {}
try:
pve_users = self._api.access.users.get() or [] # type: ignore[union-attr]
except Exception as exc:
logger.debug("Failed to fetch Proxmox users: %s", exc)
pve_users = []
for user in pve_users:
uid = user.get("userid", "")
if not uid:
continue
# Proxmox roles: Administrator -> 15, otherwise 1
groups = user.get("groups", "") or ""
level = 1
try:
roles = self._api.access.users(uid).get() or {} # type: ignore[union-attr]
if "Administrator" in str(roles):
level = 15
except Exception as exc:
logger.debug("Failed to fetch roles for user %s: %s", uid, exc)
result[uid] = {
"level": level,
"password": "",
"sshkeys": [],
}
# Merge local OS users from /etc/passwd
raw = self._exec_ssh_command("getent passwd 2>/dev/null || cat /etc/passwd")
for line in raw.splitlines():
parts = line.split(":")
if len(parts) < 7:
continue
uname, _, uid_str, *_ = parts
try:
uid_int = int(uid_str)
except ValueError:
continue
if uname not in result and uid_int < 1000 or uid_int == 0:
result[uname] = {
"level": 15 if uid_int == 0 else 0,
"password": "",
"sshkeys": [],
}
return result
# ------------------------------------------------------------------ #
# Packages (Debian APT)
# ------------------------------------------------------------------ #
def get_packages(self) -> list[_JsonDict]:
"""Return installed Debian packages with available-update info.
Installed list comes from ``dpkg-query`` via SSH (the Proxmox API
``/apt/installed`` endpoint is not implemented on PVE 8.x).
Available updates come from the Proxmox API ``/apt/update``.
"""
# Available updates from Proxmox API (keyed by package name)
upgradable: dict[str, str] = {}
try:
for upd in self._api.nodes(self._node_name).apt.update.get():
pkg = upd.get("Package", "")
if pkg:
upgradable[pkg] = upd.get("Version", "")
except Exception as exc:
logger.debug("Failed to fetch available APT updates: %s", exc)
# Installed packages via SSH dpkg-query
raw = self._exec_ssh_command(
"dpkg-query -W -f='${Package}\\t${Version}\\t${db:Status-Status}\\t${Installed-Size}\\n'"
" 2>/dev/null"
)
result: list[_JsonDict] = []
for line in raw.splitlines():
parts = line.strip().split("\t")
if len(parts) < 2:
continue
name = parts[0]
version = parts[1] if len(parts) > 1 else ""
status = parts[2] if len(parts) > 2 else "installed"
size_kb = parts[3] if len(parts) > 3 else "0"
if not name or status != "installed":
continue
size_bytes = int(size_kb) * 1024 if size_kb.isdigit() else 0
result.append({
"name": name,
"version": version,
"installed": True,
"description": "",
"size": size_bytes,
"source": "pve",
"upgrade_version": upgradable.get(name, ""),
})
return result
# ------------------------------------------------------------------ #
# Device warnings
# ------------------------------------------------------------------ #
def get_device_warnings(self) -> list[_JsonDict]:
"""Return warnings for the Proxmox node.
Currently detects:
- lldpd not installed (LLDP neighbor discovery unavailable)
- Available package updates (via Proxmox APT API)
- Missing / invalid subscription
"""
warnings: list[_JsonDict] = []
# 0. LLDP daemon
try:
lldpd_path = self._exec_ssh_command("command -v lldpd 2>/dev/null").strip()
if not lldpd_path:
warnings.append({
"code": "lldpd_not_installed",
"severity": "warning",
"action": "install_lldpd",
})
except Exception as exc:
logger.debug("Failed to check for lldpd: %s", exc)
# 1. Available package updates
try:
updates = self.get_available_updates()
if updates:
warnings.append({
"code": "updates_available",
"severity": "warning",
"title": (
f"{len(updates)} package update"
f"{'s' if len(updates) != 1 else ''} available"
),
"message": None,
"action": None,
"meta": {
"count": len(updates),
"packages": [u["name"] for u in updates],
},
})
except Exception as exc:
logger.debug("Failed to check available updates: %s", exc)
# 2. Subscription status
try:
sub = self._get_node_subscription()
status = sub.get("status", "")
if status in ("NotFound", "Invalid", "Expired"):
warnings.append({
"code": "no_subscription",
"severity": "warning",
"action": None,
"meta": {"status": status},
})
except Exception as exc:
logger.debug("Failed to check subscription status: %s", exc)
return warnings
# ------------------------------------------------------------------ #
# Services (systemd)
# ------------------------------------------------------------------ #
def get_services(self) -> list[_JsonDict]:
"""Return systemd services with running and enabled state.
Uses two ``systemctl`` invocations combined in a single SSH command:
- ``list-unit-files`` for the static enabled/disabled state
- ``list-units`` for the live running state
"""
raw = self._exec_ssh_command(
"{ systemctl list-unit-files --type=service --no-pager --no-legend --full 2>/dev/null;"
" echo '---UNITS---';"
" systemctl list-units --type=service --all --no-pager --no-legend --full 2>/dev/null;"
" } || true"
)
# Parse enabled state from list-unit-files
enabled_map: dict[str, bool] = {}
section = "files"
for line in raw.splitlines():
if line.strip() == "---UNITS---":
section = "units"
continue
parts = line.strip().split(None, 1)
if len(parts) < 1:
continue
unit = parts[0].lstrip("\u25cf").strip()
if not unit.endswith(".service"):
continue
name = unit[: -len(".service")]
if section == "files":
state = parts[1].strip() if len(parts) > 1 else ""
enabled_map[name] = state in ("enabled", "enabled-runtime", "static")
# Parse running state from list-units
running_map: dict[str, bool] = {}
section = "files"
for line in raw.splitlines():
if line.strip() == "---UNITS---":
section = "units"
continue
if section != "units":
continue
parts = line.strip().lstrip("\u25cf").strip().split(None, 4)
if len(parts) < 4:
continue
unit = parts[0]
if not unit.endswith(".service"):
continue
name = unit[: -len(".service")]
sub_state = parts[3]
running_map[name] = sub_state == "running"
all_names = sorted(set(enabled_map) | set(running_map))
return [
{
"name": name,
"running": running_map.get(name, False),
"enabled": enabled_map.get(name, False),
"pid": 0,
}
for name in all_names
]
def manage_service(self, name: str, action: str) -> _JsonDict:
"""Start / stop / restart / enable / disable a systemd service."""
if not re.match(r'^[a-zA-Z0-9_\-\.@]+$', name):
raise ValueError(f"Invalid service name: {name!r}")
if action not in ('start', 'stop', 'restart', 'enable', 'disable'):
raise ValueError(f"Invalid action: {action!r}")
output = self._exec_ssh_command(f"systemctl {action} {name}.service 2>&1 || true")
return {"success": True, "output": output}
# ------------------------------------------------------------------ #
# Available updates
# ------------------------------------------------------------------ #
def get_available_updates(self) -> list[_JsonDict]:
"""Return list of upgradable packages from the Proxmox APT API."""
updates: list[_JsonDict] = []
try:
for upd in self._api.nodes(self._node_name).apt.update.get():
pkg = upd.get("Package", "")
if not pkg:
continue
updates.append({
"name": pkg,
"current_version": upd.get("OldVersion", ""),
"new_version": upd.get("Version", ""),
})
except Exception as exc:
logger.debug("Failed to fetch available updates: %s", exc)
return sorted(updates, key=lambda u: u["name"])
def apply_updates(self, packages: list[str]) -> _JsonDict:
"""Upgrade the given packages via ``apt-get install`` over SSH."""
for pkg in packages:
if not re.match(r'^[a-zA-Z0-9_\-\+\.]+$', pkg):
raise ValueError(f"Invalid package name: {pkg!r}")
pkg_args = " ".join(packages)
output = self._exec_ssh_command(
f"DEBIAN_FRONTEND=noninteractive apt-get install --only-upgrade -y {pkg_args} 2>&1 || true"
)
return {"success": True, "output": output}
# ------------------------------------------------------------------ #
# 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 as exc:
logger.debug("Failed to parse snmpd.conf: %s", exc)
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()
if action == "install_lldpd":
return self._action_install_lldpd()
raise NotImplementedError(f"Unknown action: {action!r}")
def _action_install_lldpd(self) -> dict:
"""Install, enable and start lldpd on the Proxmox node.
Proxmox runs Debian/Linux underneath, so this is a plain apt install
followed by enabling the systemd service. lldpd's defaults (listen on
all interfaces) are sufficient to discover the directly-connected
switch via the management bridge (e.g. vmbr0).
"""
lines: list[str] = []
install_out = self._exec_ssh_command(
"DEBIAN_FRONTEND=noninteractive apt-get install -y lldpd 2>&1 | tail -5"
)
lines.append(f"[install] {install_out.strip()[-200:]}")
enable_out = self._exec_ssh_command(
"systemctl enable --now lldpd 2>&1 || service lldpd start 2>&1 || true"
)
lines.append(f"[service] {enable_out.strip()[-200:]}")
verify = self._exec_ssh_command("command -v lldpd 2>/dev/null").strip()
success = bool(verify)
if success:
lines.append("[ok] lldpd installed. Neighbors will appear after a short warm-up period.")
else:
lines.append("[warn] lldpd does not appear to be installed after the attempt.")
return {"success": success, "output": "\n".join(lines)}
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: list[str] = []
# 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 as exc:
logger.debug("Failed to detect network origin IP: %s", exc)
# 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)}
# ------------------------------------------------------------------ #
# set_hostname
# ------------------------------------------------------------------ #
def set_hostname(self, new_hostname: str) -> None:
"""Set the system hostname on the Proxmox node.
Performs all steps required for a clean rename on a Debian/Proxmox host:
1. ``/etc/hostname`` — short hostname only.
2. ``/etc/hosts`` — old hostname replaced by new (short or FQDN).
3. ``/etc/mailname`` — FQDN, if the file exists.
4. Postfix ``myhostname`` — updated via ``postconf -e`` if Postfix is
installed.
5. Runtime hostname — applied immediately via ``hostname(1)`` so SSH
and the Proxmox API see the new name without a reboot.
6. ``/etc/pve/nodes/<old>`` → ``/etc/pve/nodes/<new>`` rename so
VM/CT configs are found after the reboot under the new node name.
7. Proxmox node TLS certificate — regenerated via
``pvenode cert create --overwrite``.
**Important**: the Proxmox *node name* (shown in the web UI and stored
in ``/etc/pve/nodes/<name>/``) only changes after a full reboot.
Until then the web UI still shows the old node name, but the OS-level
hostname and the SSL certificate already reflect the new value.
Accepts a bare hostname (``pve-see``) or an FQDN
(``pve-see.see.local``).
"""
import base64 as _b64
if "." in new_hostname:
short_hostname, domain = new_hostname.split(".", 1)
fqdn = new_hostname
else:
short_hostname = new_hostname
domain = None
fqdn = new_hostname
# Derive old short hostname for substitution in /etc/hosts
old_short = self._exec_ssh_command(
"cat /etc/hostname 2>/dev/null || hostname -s 2>/dev/null"
).strip().split(".")[0]
# ── 1. /etc/hostname ─────────────────────────────────────────────
hostname_b64 = _b64.b64encode(f"{short_hostname}\n".encode()).decode()
self._exec_ssh_command(f"echo {hostname_b64} | base64 -d > /etc/hostname")
# ── 2. /etc/hosts ────────────────────────────────────────────────
# Read the file, substitute in Python (safe for all characters),
# write back via base64 to avoid shell-escaping issues.
if old_short and old_short != short_hostname:
hosts_raw = self._exec_ssh_command("cat /etc/hosts 2>/dev/null")
# Replace whole-word occurrences of the old short hostname.
# Both "hostname" and "hostname.domain" forms are covered because
# the new value is the full fqdn when a domain was given.
import re as _re
new_replace = fqdn if domain else short_hostname
hosts_new = _re.sub(
r"\b" + _re.escape(old_short) + r"\b",
new_replace,
hosts_raw,
)
hosts_b64 = _b64.b64encode(hosts_new.encode()).decode()
self._exec_ssh_command(
f"echo {hosts_b64} | base64 -d > /tmp/_netork_hosts"
f" && mv /tmp/_netork_hosts /etc/hosts"
f" && chmod 644 /etc/hosts"
)
# ── 3. /etc/mailname ─────────────────────────────────────────────
mailname_b64 = _b64.b64encode(f"{fqdn}\n".encode()).decode()
self._exec_ssh_command(
f"[ -f /etc/mailname ]"
f" && echo {mailname_b64} | base64 -d > /etc/mailname"
f" || true"
)
# ── 4. Postfix myhostname ─────────────────────────────────────────
self._exec_ssh_command(
f"command -v postconf >/dev/null 2>&1"
f" && postconf -e 'myhostname = {fqdn}'"
f" || true"
)
# ── 5. Apply hostname at runtime ─────────────────────────────────
self._exec_ssh_command(f"hostname '{short_hostname}'")
# ── 6. Migrate /etc/pve/nodes/<old> → /etc/pve/nodes/<new> ──────
# Proxmox stores VM and CT configs under /etc/pve/nodes/<nodename>/.
# This rename MUST happen before the reboot while pve-cluster is
# running and pmxcfs is mounted — pmxcfs supports the rename live.
# Without this step the node boots under the new hostname, finds an
# empty /etc/pve/nodes/<new>/ and all VMs appear missing.
if old_short and old_short != short_hostname:
pve_old = f"/etc/pve/nodes/{old_short}"
pve_new = f"/etc/pve/nodes/{short_hostname}"
migrate_out = self._exec_ssh_command(
f"if [ -d '{pve_old}' ] && [ ! -e '{pve_new}' ]; then"
f" mv '{pve_old}' '{pve_new}'"
f" && echo 'migrated {pve_old} -> {pve_new}';"
f"elif [ -d '{pve_new}' ]; then"
f" echo 'target already exists, skipping migration';"
f"else"
f" echo 'source {pve_old} not found, skipping migration';"
f"fi"
)
logger.info("Proxmox node dir migration: %s", migrate_out)
# ── 7. Regenerate Proxmox node TLS certificate ───────────────────
# Use pvenode cert create --overwrite rather than pvecm updatecerts -f.
# pvecm updatecerts -f restarts pve-cluster which briefly unmounts the
# cluster filesystem (/etc/pve) and can crash running VMs.
# pvenode cert create --overwrite only touches the node's own cert and
# restarts pveproxy (safe while VMs are running).
cert_out = self._exec_ssh_command(
"pvenode cert create --overwrite 1 2>&1"
" || { systemctl restart pveproxy 2>&1; echo 'restarted pveproxy'; }"
" || true"
)
logger.info("Proxmox cert regeneration: %s", cert_out[:200])
logger.info(
"Proxmox set_hostname: %s → %s (fqdn: %s). "
"Reboot required for node name to update in web UI / cluster.",
old_short, short_hostname, fqdn,
)
# ------------------------------------------------------------------ #
# _send_command — generic SSH command helper used by netOrk core
# ------------------------------------------------------------------ #
def _send_command(self, command: str) -> str:
"""Execute *command* on the node and return its output.
Delegates to _exec_ssh_command so that netOrk's generic helpers
(e.g. the reboot action) work without knowing the driver internals.
"""
return self._exec_ssh_command(command)
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 as exc:
logger.debug("Failed to list disks: %s", exc)
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 as exc:
logger.debug("Failed to fetch SMART data for %s: %s", dev, exc)
result[dev] = entry
return result
def get_system_config(self) -> dict[str, Any]:
"""Return system-level configuration for this Proxmox node.
Collected from ``/cluster/status`` and the node config API:
* ``cluster_name`` (str | None) — Proxmox cluster name; ``None`` when
the node is standalone (not part of a cluster).
* ``cluster_nodes`` (list[str]) — hostnames of all online cluster nodes.
* ``hostname`` (str) — this node's name as reported by the cluster.
* ``timezone`` (str | None) — e.g. ``"Europe/Berlin"``.
* ``ntp_servers`` (list[str]).
* ``ssh_port`` (int) — always 22 for Proxmox nodes.
* ``ssh_password_auth`` (bool) — ``False`` as a safe default.
"""
cfg: dict[str, Any] = {
"cluster_name": None,
"cluster_nodes": [],
"hostname": self._node_name,
"timezone": None,
"ntp_servers": [],
"ssh_port": 22,
"ssh_password_auth": False,
}
# Cluster name and node list from /cluster/status
try:
status = self._api.cluster.status.get() or []
for entry in status:
if entry.get("type") == "cluster":
cfg["cluster_name"] = entry.get("name")
elif entry.get("type") == "node" and entry.get("online"):
cfg["cluster_nodes"].append(entry.get("name", ""))
except Exception as exc:
logger.debug("Failed to fetch cluster status for system config: %s", exc)
# Timezone from node time API
try:
time_cfg = self._node_api().time.get() or {}
cfg["timezone"] = time_cfg.get("timezone") or None
except Exception as exc:
logger.debug("Failed to fetch node time config: %s", exc)
return cfg
-22
View File
@@ -26,28 +26,6 @@ _MAC_RE = re.compile(
def normalize_mac(mac: str) -> str: def normalize_mac(mac: str) -> str:
"""Return a colon-separated, lower-case MAC address, or '' on failure."""
if not mac:
return ""
try:
return str(EUI(mac, dialect=_ColonDialect()))
except (AddrFormatError, ValueError):
return mac.lower()
class _ColonDialect(EUI):
"""EUI dialect that formats as aa:bb:cc:dd:ee:ff."""
word_sep = ":"
word_fmt = "%.2x"
def __new__(cls): # type: ignore[override]
from netaddr import mac_unix_expanded
return mac_unix_expanded
def normalize_mac(mac: str) -> str: # noqa: F811
"""Return lower-case colon-separated MAC or empty string.""" """Return lower-case colon-separated MAC or empty string."""
if not mac: if not mac:
return "" return ""
+396
View File
@@ -0,0 +1,396 @@
# Copyright 2025 The NetOrk Project Authors. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""VM/container NAPALM getters for Proxmox VE nodes."""
from __future__ import annotations
import logging
import re
from typing import Any
from napalm_proxmox import utils
logger = logging.getLogger(__name__)
_JsonDict = dict[str, Any]
class ProxmoxVMMixin:
"""Mixin providing VM and container NAPALM methods."""
def get_vm_interfaces(
self, vmid: int, vm_type: str
) -> tuple[dict[str, _JsonDict], bool, bool]:
"""Return network interfaces for a single VM or LXC container.
Returns a 3-tuple ``(interfaces, agent_running, agent_enabled)``:
* ``interfaces`` – dict keyed by interface name, each with
NAPALM-compatible fields plus ``ipv4``, ``bridge``, and ``tag``.
* ``agent_running`` – True if the QEMU Guest Agent responded during
this call (always False for LXC).
* ``agent_enabled`` – True if the QEMU Guest Agent is enabled in the
VM's Proxmox config (always False for LXC).
LXC : uses ``/nodes/{node}/lxc/{vmid}/interfaces`` + LXC config
QEMU : tries QEMU guest agent first, falls back to VM config parsing.
In both paths the VM config is fetched to derive bridge/tag.
"""
_NET_MODELS = {"virtio", "e1000", "e1000e", "vmxnet3", "rtl8139", "ne2k_pci"}
def _parse_net_entry(val_str: str) -> tuple[str, str, int | None]:
"""Parse a Proxmox net config value → (mac_upper, bridge, tag|None)."""
mac = bridge = ""
tag: int | None = None
for part in str(val_str).split(","):
if "=" not in part:
continue
k, v = part.split("=", 1)
k = k.strip().lower()
if k in _NET_MODELS:
mac = v.strip()
elif k == "bridge":
bridge = v.strip()
elif k == "tag":
try:
tag = int(v.strip())
except ValueError:
pass
return mac.upper() if mac else "", bridge, tag
interfaces: dict[str, _JsonDict] = {}
if vm_type == "container":
# Build iface_name → (bridge, tag) from LXC config
# LXC net entries look like: net0=name=eth0,bridge=vmbr40,tag=40,...
lxc_net_map: dict[str, tuple[str, int | None]] = {}
try:
config = self._node_api().lxc(vmid).config.get() or {}
net_re = re.compile(r"^net(\d+)$")
for key, val in config.items():
if not net_re.match(key):
continue
iface_name = ""
bridge = ""
tag: int | None = None
for part in str(val).split(","):
if "=" not in part:
continue
k, v = part.split("=", 1)
k = k.strip().lower()
if k == "name":
iface_name = v.strip()
elif k == "bridge":
bridge = v.strip()
elif k == "tag":
try:
tag = int(v.strip())
except ValueError:
pass
if iface_name:
lxc_net_map[iface_name] = (bridge, tag)
except Exception as exc:
logger.debug("get_vm_interfaces: LXC %s config failed: %s", vmid, exc)
try:
for iface in (self._node_api().lxc(vmid).interfaces.get() or []):
name = iface.get("name", "")
if not name or name == "lo":
continue
mac = iface.get("hwaddr", "")
ipv4 = ""
inet = iface.get("inet", "")
if inet:
ipv4 = inet.split("/")[0]
bridge, tag = lxc_net_map.get(name, ("", None))
interfaces[name] = {
"is_up": True,
"is_enabled": True,
"description": bridge,
"mac_address": mac.upper() if mac else "",
"speed": -1,
"mtu": 1500,
"last_flapped": -1.0,
"ipv4": ipv4,
"bridge": bridge,
"tag": tag,
}
except Exception as exc:
logger.debug("get_vm_interfaces: LXC %s ifaces failed: %s", vmid, exc)
# LXC containers do not use QEMU Guest Agent
return interfaces, False, False
else:
# QEMU: pre-fetch VM config to build MAC → (bridge, tag) map and
# to check whether the QEMU Guest Agent is enabled.
mac_to_net: dict[str, tuple[str, int | None]] = {} # mac_upper → (bridge, tag)
net_idx_map: dict[str, tuple[str, str, int | None]] = {} # "netN" → (mac, bridge, tag)
agent_enabled = False
try:
config = self._node_api().qemu(vmid).config.get() or {}
# Proxmox stores the agent setting as agent=1, agent=0, or
# agent=enabled=1[,fstrim_cloned_disks=1,...]
raw_agent = str(config.get("agent", "0"))
# Treat any truthy value ("1", "enabled=1", ...) as enabled
agent_enabled = bool(
raw_agent.strip() in ("1", "true")
or raw_agent.startswith("enabled=1")
or raw_agent.startswith("1,")
)
net_re = re.compile(r"^net(\d+)$")
for key, val in config.items():
m = net_re.match(key)
if not m:
continue
mac, bridge, tag = _parse_net_entry(val)
iface_key = f"net{m.group(1)}"
net_idx_map[iface_key] = (mac, bridge, tag)
if mac:
mac_to_net[mac] = (bridge, tag)
except Exception as exc:
logger.debug("get_vm_interfaces: QEMU %s config fetch failed: %s", vmid, exc)
# Try guest agent first
agent_ok = False
try:
agent_result = self._node_api().qemu(vmid).agent("network-get-interfaces").get()
for iface in (agent_result or {}).get("result", []):
name = iface.get("name", "")
if not name or name == "lo":
continue
mac = (iface.get("hardware-address", "") or "").upper()
ipv4 = ""
for addr in iface.get("ip-addresses", []):
if addr.get("ip-address-type") == "ipv4":
ipv4 = addr.get("ip-address", "")
break
bridge, tag = mac_to_net.get(mac, ("", None))
interfaces[name] = {
"is_up": True,
"is_enabled": True,
"description": bridge,
"mac_address": mac,
"speed": -1,
"mtu": 1500,
"last_flapped": -1.0,
"ipv4": ipv4,
"bridge": bridge,
"tag": tag,
}
agent_ok = bool(interfaces)
except Exception as exc:
logger.debug("QEMU guest agent network-get-interfaces failed: %s", exc)
if not agent_ok:
# Fall back to config-only (gives MAC + bridge + tag, no IP)
for iface_key, (mac, bridge, tag) in net_idx_map.items():
interfaces[iface_key] = {
"is_up": False,
"is_enabled": True,
"description": bridge,
"mac_address": mac,
"speed": -1,
"mtu": 1500,
"last_flapped": -1.0,
"ipv4": "",
"bridge": bridge,
"tag": tag,
}
return interfaces, agent_ok, agent_enabled
def get_vms(self) -> list[_JsonDict]:
"""Return all VMs (QEMU) and containers (LXC) on this node.
Each entry contains:
* vmid (int) - Proxmox VM/container ID
* name (str) - display name
* type (str) - ``"vm"`` or ``"container"``
* status (str) - ``"running"``, ``"stopped"``, etc.
* vcpus (int) - allocated vCPUs
* memory (int) - configured RAM in megabytes
* cpu_usage (float) - current CPU utilisation 0.0–1.0 (from last stats cycle)
* memory_usage (int)- current RSS in megabytes
* uptime (int) - uptime in seconds (0 if stopped)
* node (str) - cluster node name
* interfaces (dict) - network interfaces (NAPALM format + ipv4 field)
* ipv4 (str) - primary IPv4 address (empty string if unknown)
"""
result: list[_JsonDict] = []
# QEMU VMs
try:
for vm in (self._node_api().qemu.get() or []):
vmid = int(vm.get("vmid", 0))
name = vm.get("name", f"vm-{vmid}")
status = vm.get("status", "unknown")
cpu_usage = float(vm.get("cpu", 0.0) or 0.0)
uptime = int(vm.get("uptime", 0) or 0)
# mem/maxmem are in bytes
maxmem_bytes = int(vm.get("maxmem", 0) or 0)
mem_bytes = int(vm.get("mem", 0) or 0)
memory_mb = maxmem_bytes // (1024 * 1024)
memory_usage_mb = mem_bytes // (1024 * 1024)
# vcpus can be in "cpus" key for running VMs
vcpus = int(vm.get("cpus", vm.get("vcpus", 0)) or 0)
interfaces, agent_running, agent_enabled = self.get_vm_interfaces(vmid, "vm")
ipv4 = next(
(iface["ipv4"] for iface in interfaces.values() if iface.get("ipv4")),
"",
)
disks, onboot = self._get_vm_disk_and_boot(vmid, "qemu")
result.append({
"vmid": vmid,
"name": name,
"type": "vm",
"status": status,
"vcpus": vcpus,
"memory": memory_mb,
"cpu_usage": round(cpu_usage, 4),
"memory_usage": memory_usage_mb,
"uptime": uptime,
"node": self._node_name,
"interfaces": interfaces,
"ipv4": ipv4,
"agent_enabled": agent_enabled,
"agent_running": agent_running,
"disks": disks,
"onboot": onboot,
})
except Exception as exc:
logger.warning("get_vms: failed to list QEMU VMs: %s", exc)
# LXC containers
try:
for ct in (self._node_api().lxc.get() or []):
vmid = int(ct.get("vmid", 0))
name = ct.get("name", f"ct-{vmid}")
status = ct.get("status", "unknown")
cpu_usage = float(ct.get("cpu", 0.0) or 0.0)
uptime = int(ct.get("uptime", 0) or 0)
maxmem_bytes = int(ct.get("maxmem", 0) or 0)
mem_bytes = int(ct.get("mem", 0) or 0)
memory_mb = maxmem_bytes // (1024 * 1024)
memory_usage_mb = mem_bytes // (1024 * 1024)
vcpus = int(ct.get("cpus", 0) or 0)
interfaces, agent_running, agent_enabled = self.get_vm_interfaces(vmid, "container")
ipv4 = next(
(iface["ipv4"] for iface in interfaces.values() if iface.get("ipv4")),
"",
)
disks, onboot = self._get_vm_disk_and_boot(vmid, "lxc")
result.append({
"vmid": vmid,
"name": name,
"type": "container",
"status": status,
"vcpus": vcpus,
"memory": memory_mb,
"cpu_usage": round(cpu_usage, 4),
"memory_usage": memory_usage_mb,
"uptime": uptime,
"node": self._node_name,
"interfaces": interfaces,
"ipv4": ipv4,
"agent_enabled": agent_enabled,
"agent_running": agent_running,
"disks": disks,
"onboot": onboot,
})
except Exception as exc:
logger.warning("get_vms: failed to list LXC containers: %s", exc)
return sorted(result, key=lambda x: x["vmid"])
# Disk-key prefixes for QEMU: scsi, virtio, ide, sata (exclude cdrom/none entries)
_DISK_KEYS_VM = re.compile(r"^(scsi|virtio|ide|sata)\d+$")
# Disk-key prefix for LXC: rootfs, mp (mount points)
_DISK_KEYS_CT = re.compile(r"^(rootfs|mp\d+)$")
def _get_vm_disk_and_boot(
self, vmid: int, vm_type: str
) -> tuple[list[dict], bool]:
"""Return (disks, onboot) for a VM or container.
Fetches config from /nodes/{node}/{qemu|lxc}/{vmid}/config.
Each disk entry is a dict with 'name' and 'size_mb'.
Returns ([], False) on any error.
"""
try:
if vm_type == "qemu":
cfg = self._node_api().qemu(vmid).config.get() or {}
disk_re = self._DISK_KEYS_VM
else:
cfg = self._node_api().lxc(vmid).config.get() or {}
disk_re = self._DISK_KEYS_CT
onboot = bool(int(cfg.get("onboot", 0) or 0))
disks: list[dict] = []
for key, value in sorted(cfg.items()):
if not disk_re.match(key):
continue
val_str = str(value or "")
# Skip CD-ROM / none entries
if "media=cdrom" in val_str or val_str.startswith("none"):
continue
# Extract size=NNX from the value string
m = re.search(r"\bsize=(\d+)([GMK]?)", val_str, re.I)
if not m:
continue
num = int(m.group(1))
unit = m.group(2).upper()
if unit == "G" or unit == "":
size_mb = num * 1024
elif unit == "M":
size_mb = num
elif unit == "K":
size_mb = num // 1024
else:
size_mb = num * 1024
disks.append({"name": key, "size_mb": size_mb})
return disks, onboot
except Exception as exc:
logger.debug("Failed to fetch disk/boot config for vmid %s: %s", vmid, exc)
return [], False
_POWER_ACTIONS_VM = {'start', 'stop', 'shutdown', 'reboot', 'reset'}
_POWER_ACTIONS_CT = {'start', 'stop', 'shutdown', 'reboot'}
def power_vm(self, vmid: int, vm_type: str, action: str) -> dict:
"""Send a power action to a VM or container on this node.
Supported actions for VMs: start, stop, shutdown, reboot, reset
Supported actions for containers: start, stop, shutdown, reboot
"""
allowed = self._POWER_ACTIONS_VM if vm_type == 'vm' else self._POWER_ACTIONS_CT
if action not in allowed:
return {"success": False, "error": f"Action '{action}' not supported for {vm_type} (allowed: {sorted(allowed)})"}
try:
vm_api = self._node_api().qemu(vmid) if vm_type == 'vm' else self._node_api().lxc(vmid)
task_id = getattr(vm_api.status, action).post()
return {"success": True, "task_id": task_id or ""}
except Exception as exc:
return {"success": False, "error": str(exc)}
+389
View File
@@ -0,0 +1,389 @@
"""VM provisioning mixin for Proxmox — creates, destroys, and monitors VMs via Cloud-Init."""
from __future__ import annotations
import logging
import time
import yaml
from typing import Any, Dict, List
from urllib.parse import quote
from napalm_device_types.models import VMProvisionResultDict, VMStatusDict
_logger = logging.getLogger(__name__)
class ProxmoxVMProvisionMixin:
"""Mixin to add VM provisioning to ProxmoxDriver."""
def _wait_for_task(self, upid: str, timeout: int = 120) -> None:
"""
Poll a Proxmox task until completion.
Polls /nodes/{node}/tasks/{upid}/status until status == 'stopped'.
Raises RuntimeError if exitstatus != 'OK' or timeout exceeded.
"""
start_time = time.time()
while True:
elapsed = time.time() - start_time
if elapsed > timeout:
raise RuntimeError(f"Task {upid} timed out after {timeout}s")
try:
task_status = self._node_api().tasks(upid).status.get()
except Exception as e:
_logger.debug(f"Error polling task {upid}: {e}")
time.sleep(2)
continue
if task_status.get("status") == "stopped":
exitstatus = task_status.get("exitstatus", "UNKNOWN")
if exitstatus != "OK":
raise RuntimeError(
f"Task {upid} failed with exitstatus='{exitstatus}': "
f"{task_status.get('exitstatus_text', 'no error message')}"
)
return
time.sleep(2)
def create_vm_from_cloud_init(
self,
name: str,
*,
template: str,
cpu: int,
memory: int,
nics: List[Dict[str, Any]],
cloud_init_config: Dict[str, Any],
ssh_public_keys: List[str] | None = None,
disk_resize_gb: int | None = None,
timeout: int = 180,
) -> VMProvisionResultDict:
"""
Create a new VM from a Cloud-Init template via Proxmox API.
Steps:
1. Get next available VMID from cluster
2. Clone template VM (full clone, new VMID)
3. Configure CPU, memory, and network interfaces
4. Verify snippet storage exists
5. Render cloud-init config to YAML and upload
6. Set Cloud-Init config references and SSH keys
7. Optionally resize root disk
8. Start the VM
9. Return VMID, name, node
Args:
name: new VM display name
template: template VMID/name to clone from
cpu: number of vCPUs
memory: RAM in MB
nics: list of NIC config dicts (bridge, vlan_tag/trunk_vlan_tags, dhcp flag)
cloud_init_config: user-data dict (will be YAML-rendered)
ssh_public_keys: SSH public keys to inject
disk_resize_gb: resize root disk to this size (None = no resize)
timeout: max seconds for provisioning
Returns:
{"vmid": str, "name": str, "node": str}
Raises:
RuntimeError: provisioning failure (clone, config, timeout, etc.)
ValueError: invalid storage or configuration
"""
try:
_logger.info(f"Creating VM '{name}' from template {template}")
# Step 1: Get next VMID
next_vmid = self._api.cluster.nextid.get()
vmid = int(next_vmid)
_logger.info(f"Allocated VMID {vmid}")
# Step 2: Clone template
_logger.info(f"Cloning template {template} → VMID {vmid}")
clone_upid = self._node_api().qemu(template).clone.post(
newid=vmid,
full=1,
name=name,
)
self._wait_for_task(clone_upid, timeout=timeout)
# Step 3: Configure CPU, memory, and NICs
_logger.info(f"Configuring VM {vmid}: {cpu} CPU, {memory}MB RAM, {len(nics)} NIC(s)")
config_args = {"cores": cpu, "memory": memory}
# Build NIC config strings generically
for i, nic in enumerate(nics):
bridge = nic.get("bridge")
if not bridge:
raise ValueError(f"NIC {i}: bridge is required")
# Build base config: model + bridge
net_config = f"virtio,bridge={bridge}"
# Add VLAN configuration (access vs trunk)
if "trunk_vlan_tags" in nic and nic["trunk_vlan_tags"]:
vlan_list = ";".join(str(v) for v in nic["trunk_vlan_tags"])
net_config += f",trunks={vlan_list}"
elif "vlan_tag" in nic and nic["vlan_tag"] is not None:
net_config += f",tag={nic['vlan_tag']}"
config_args[f"net{i}"] = net_config
self._node_api().qemu(vmid).config.post(**config_args)
# Step 4: Verify snippet storage exists
_logger.info("Checking for snippet storage...")
storages = self._api.storage.get()
snippet_storage = None
for storage in storages:
content = storage.get("content", "")
if "snippets" in content and storage.get("enabled"):
snippet_storage = storage["storage"]
break
if not snippet_storage:
raise ValueError(
"No storage with content='snippets' found. "
"Configure a snippet-capable storage (e.g. local, nfs dir) "
"and enable it."
)
_logger.info(f"Using snippet storage: {snippet_storage}")
# Step 5: Render and upload Cloud-Init config
_logger.info(f"Rendering Cloud-Init config for VMID {vmid}")
user_data_yaml = "#cloud-config\n" + yaml.dump(
cloud_init_config, default_flow_style=False
)
filename = f"{vmid}-user-data.yaml"
_logger.debug(f"Uploading Cloud-Init snippet {filename} to {snippet_storage}")
# Upload to snippet storage
self._node_api().storage(snippet_storage).upload.post(
content="snippets",
filename=filename,
data=user_data_yaml,
)
# Step 6: Configure Cloud-Init references and SSH keys
_logger.info(f"Setting Cloud-Init config for VM {vmid}")
cloud_init_args = {
"ide2": f"{snippet_storage}:cloudinit",
"citype": "nocloud",
"cicustom": f"user={snippet_storage}:snippets/{filename}",
}
# Configure DHCP for NICs where enabled (default True for index 0, False otherwise)
for i, nic in enumerate(nics):
dhcp_enabled = nic.get("dhcp", i == 0) # Default DHCP for first NIC only
if dhcp_enabled:
cloud_init_args[f"ipconfig{i}"] = "ip=dhcp"
if ssh_public_keys:
# URL-encode SSH keys for Proxmox API
sshkeys = ";".join(ssh_public_keys)
cloud_init_args["sshkeys"] = quote(sshkeys)
self._node_api().qemu(vmid).config.post(**cloud_init_args)
# Step 7: Optionally resize root disk
if disk_resize_gb is not None:
_logger.info(f"Resizing root disk to {disk_resize_gb}GB")
# Find root disk (scsi0, virtio0, ide0, sata0 — whichever is first)
try:
config = self._node_api().qemu(vmid).config.get()
root_disk = None
for prefix in ("scsi", "virtio", "ide", "sata"):
if f"{prefix}0" in config:
root_disk = f"{prefix}0"
break
if root_disk:
self._node_api().qemu(vmid).resize.put(
disk=root_disk,
size=f"{disk_resize_gb}G",
)
else:
_logger.warning(f"Could not find root disk for VM {vmid}, skipping resize")
except Exception as e:
_logger.warning(f"Failed to resize disk: {e}, continuing anyway")
# Step 8: Start the VM
_logger.info(f"Starting VM {vmid}")
start_upid = self._node_api().qemu(vmid).status.start.post()
self._wait_for_task(start_upid, timeout=timeout)
_logger.info(f"VM {vmid} ('{name}') provisioned successfully on {self._node_name}")
return {
"vmid": str(vmid),
"name": name,
"node": self._node_name,
}
except Exception as e:
_logger.exception(f"Failed to create VM '{name}': {e}")
raise
def destroy_vm(
self,
vmid: str,
*,
remove_disk: bool = True,
timeout: int = 60,
) -> None:
"""
Destroy a virtual machine and optionally remove its storage.
Steps:
1. Stop the VM if running
2. Delete VM configuration and optionally disks
3. Clean up Cloud-Init snippets
Args:
vmid: hypervisor VMID (string, e.g. "101")
remove_disk: if True, also delete disks and storage
timeout: max seconds for stop/delete operations
Raises:
RuntimeError: VM doesn't exist or destruction fails
"""
try:
vmid_int = int(vmid)
_logger.info(f"Destroying VM {vmid}")
# Step 1: Stop the VM if running
try:
_logger.debug(f"Stopping VM {vmid}")
stop_upid = self._node_api().qemu(vmid_int).status.stop.post()
self._wait_for_task(stop_upid, timeout=timeout)
except Exception as e:
_logger.debug(f"VM {vmid} stop failed (may already be stopped): {e}")
# Step 2: Delete VM
_logger.debug(f"Deleting VM {vmid} configuration and disks")
self._node_api().qemu(vmid_int).delete(
purge=1,
destroy_unreferenced_disks=1 if remove_disk else 0,
)
# Step 3: Clean up Cloud-Init snippets
# (This is best-effort; snippet files may be unreachable if storage is unavailable)
try:
config = self._node_api().qemu(vmid_int).config.get()
cicustom = config.get("cicustom", "")
if "snippets/" in cicustom:
parts = cicustom.split("=")
if len(parts) >= 2:
snippet_ref = parts[1] # e.g. "snippets:snippets/101-user-data.yaml"
storage, filepath = snippet_ref.split(":", 1)
_logger.debug(f"Deleting snippet {filepath} from {storage}")
try:
self._node_api().storage(storage).content(filepath).delete()
except Exception as e:
_logger.warning(f"Failed to delete snippet {filepath}: {e}")
except Exception as e:
_logger.debug(f"Could not clean up snippets for VM {vmid}: {e}")
_logger.info(f"VM {vmid} destroyed successfully")
except Exception as e:
_logger.exception(f"Failed to destroy VM {vmid}: {e}")
raise
def get_vm_status(
self,
vmid: str,
*,
wait_for_ip: bool = False,
timeout: int = 300,
poll_interval: int = 5,
) -> VMStatusDict:
"""
Get the runtime status of a virtual machine.
Optionally waits for the guest-agent to report an IP address on the
management NIC (net0), useful after provisioning.
Args:
vmid: hypervisor VMID (string)
wait_for_ip: if True, poll until IP appears on net0
timeout: max seconds to wait for IP (if wait_for_ip=True)
poll_interval: seconds between status polls
Returns:
{"status": str, "ip_address": str, "hostname": str, "mac_address": str}
(ip_address, hostname, mac_address only if VM is running and has network info)
Raises:
RuntimeError: VM doesn't exist or wait_for_ip times out
"""
try:
vmid_int = int(vmid)
_logger.debug(f"Getting status for VM {vmid}")
# Get VM config to infer net0 MAC (for matching guest-agent results)
try:
config = self._node_api().qemu(vmid_int).config.get()
except Exception:
# VM may not exist yet or config not readable
return {"status": "unknown"}
# Parse net0 MAC from config (if present)
net0_line = config.get("net0", "")
expected_mac = None
# Example: "virtio,bridge=vmbr0,tag=10" — no explicit MAC
# Proxmox auto-generates MACs in a deterministic pattern, but we'll
# match by looking for the first NIC's IP in guest-agent results
# Polling loop
start_time = time.time()
while True:
elapsed = time.time() - start_time
if wait_for_ip and elapsed > timeout:
raise RuntimeError(
f"VM {vmid} failed to acquire IP within {timeout}s"
)
try:
# Query guest-agent network interfaces
agent_info = self._node_api().qemu(vmid_int).agent.network_get_interfaces.get()
interfaces = agent_info.get("result", [])
# Find net0 (first interface with IP)
if interfaces:
net0_iface = interfaces[0] # Assumes net0 is first in list
net0_mac = net0_iface.get("hardware-address", "")
ip_addresses = net0_iface.get("ip-addresses", [])
if ip_addresses:
# Found IP
ip_info = ip_addresses[0]
ip_addr = ip_info.get("ip-address", "")
if ip_addr:
_logger.info(f"VM {vmid} acquired IP {ip_addr}")
return {
"status": "running",
"ip_address": ip_addr,
"hostname": net0_iface.get("name", ""),
"mac_address": net0_mac,
}
except Exception as e:
_logger.debug(f"Error querying guest-agent for VM {vmid}: {e}")
if not wait_for_ip:
# Return immediate status without IP
return {"status": "running"}
# Wait before next poll
time.sleep(poll_interval)
except RuntimeError:
raise
except Exception as e:
_logger.exception(f"Error getting status for VM {vmid}: {e}")
raise RuntimeError(f"Failed to get VM {vmid} status: {e}")
+1
View File
@@ -26,6 +26,7 @@ requires-python = ">=3.9"
dependencies = [ dependencies = [
"napalm>=5.0.0", "napalm>=5.0.0",
"napalm_device_types>=0.1.0", "napalm_device_types>=0.1.0",
"paramiko>=5.0.0", # CVE-2026-44405; imported directly for SSH fallback (driver.py)
"proxmoxer>=2.0.0", "proxmoxer>=2.0.0",
"netaddr>=0.9.0", "netaddr>=0.9.0",
"requests>=2.31.0", "requests>=2.31.0",
+396
View File
@@ -0,0 +1,396 @@
"""Tests for ProxmoxVMProvisionMixin — VM creation, destruction, status polling."""
from __future__ import annotations
import pytest
from unittest.mock import MagicMock, patch
from napalm_proxmox.driver import ProxmoxDriver
from napalm_proxmox.vm_provision_mixin import ProxmoxVMProvisionMixin
# ---------------------------------------------------------------------------
# Tests
# ---------------------------------------------------------------------------
def test_wait_for_task_success():
"""_wait_for_task succeeds when task reaches stopped status with exitstatus OK."""
mixin = ProxmoxVMProvisionMixin()
# Mock the _node_api() to return a mock that supports task polling
mock_node = MagicMock()
mock_task_status = MagicMock()
mock_task_status.status.get.side_effect = [
{"status": "running", "exitstatus": None},
{"status": "stopped", "exitstatus": "OK"},
]
mock_node.tasks.return_value = mock_task_status
mixin._node_api = MagicMock(return_value=mock_node)
# Should complete without raising
with patch("time.sleep"): # Speed up polling
result = mixin._wait_for_task("UPID:pve1:123:456:789:clone:100:root@pam:", timeout=30)
assert result is None
def test_wait_for_task_error():
"""_wait_for_task raises RuntimeError when task exits with non-OK status."""
mixin = ProxmoxVMProvisionMixin()
mock_node = MagicMock()
mock_task_status = MagicMock()
mock_task_status.status.get.return_value = {"status": "stopped", "exitstatus": "FAILED"}
mock_node.tasks.return_value = mock_task_status
mixin._node_api = MagicMock(return_value=mock_node)
with pytest.raises(RuntimeError, match="FAILED"):
with patch("time.sleep"):
mixin._wait_for_task("UPID:pve1:123:456:789:clone:100:root@pam:", timeout=30)
def test_wait_for_task_timeout():
"""_wait_for_task raises RuntimeError on timeout."""
mixin = ProxmoxVMProvisionMixin()
mock_node = MagicMock()
mock_task_status = MagicMock()
mock_task_status.status.get.return_value = {"status": "running"} # Always running
mock_node.tasks.return_value = mock_task_status
mixin._node_api = MagicMock(return_value=mock_node)
with pytest.raises(RuntimeError, match="timed out"):
with patch("napalm_proxmox.vm_provision_mixin.time.time") as mock_time:
# Simulate time passing: return incremented values to exceed timeout quickly
mock_time.side_effect = [0, 2, 4] # After 2 iterations, time > timeout=1
with patch("napalm_proxmox.vm_provision_mixin.time.sleep"):
mixin._wait_for_task("UPID:pve1:123:456:789:clone:100:root@pam:", timeout=1)
def test_create_vm_from_cloud_init_single_nic():
"""create_vm_from_cloud_init with single NIC, DHCP enabled (generic happy path)."""
mixin = ProxmoxVMProvisionMixin()
mixin._node_name = "pve1"
# Mock API hierarchy
mock_api = MagicMock()
mock_api.cluster.nextid.get.return_value = 101
mock_api.storage.get.return_value = [
{"storage": "snippets", "type": "dir", "content": "snippets", "enabled": 1}
]
mock_node = MagicMock()
mixin._api = mock_api
mixin._node_api = MagicMock(return_value=mock_node)
# Mock VM operations
mock_vm = MagicMock()
mock_node.qemu.return_value = mock_vm
mock_vm.clone.post.return_value = "UPID:pve1:123:clone"
mock_vm.config.post.return_value = None
mock_vm.config.get.return_value = {"scsi0": "local:100/vm-101-disk-0.raw"}
mock_vm.status.start.post.return_value = "UPID:pve1:124:start"
# Mock task completion
mock_task = MagicMock()
mock_task.status.get.return_value = {"status": "stopped", "exitstatus": "OK"}
mock_node.tasks.return_value = mock_task
# Mock storage upload
mock_storage = MagicMock()
mock_storage.upload.post.return_value = {"filename": "snippets:snippets/101-user-data.yaml"}
mock_node.storage.return_value = mock_storage
with patch("time.sleep"):
result = mixin.create_vm_from_cloud_init(
name="test-vm",
template="100",
cpu=2,
memory=2048,
nics=[{"bridge": "vmbr0", "vlan_tag": 10}],
cloud_init_config={"hostname": "test-vm"},
timeout=120,
)
assert result["vmid"] == "101"
assert result["name"] == "test-vm"
assert result["node"] == "pve1"
assert mock_vm.clone.post.called
# Verify NIC config was set correctly: net0 with tag=10, DHCP enabled
config_call_args = mock_vm.config.post.call_args_list[0] # First call (cores/memory/net0)
assert "net0" in config_call_args[1]
assert "tag=10" in config_call_args[1]["net0"]
assert "vmbr0" in config_call_args[1]["net0"]
def test_create_vm_from_cloud_init_dual_nic_trunk():
"""create_vm_from_cloud_init with dual NICs: net0 DHCP + net1 trunk (no DHCP)."""
mixin = ProxmoxVMProvisionMixin()
mixin._node_name = "pve1"
# Mock API hierarchy
mock_api = MagicMock()
mock_api.cluster.nextid.get.return_value = 102
mock_api.storage.get.return_value = [
{"storage": "snippets", "type": "dir", "content": "snippets", "enabled": 1}
]
mock_node = MagicMock()
mixin._api = mock_api
mixin._node_api = MagicMock(return_value=mock_node)
# Mock VM operations
mock_vm = MagicMock()
mock_node.qemu.return_value = mock_vm
mock_vm.clone.post.return_value = "UPID:pve1:125:clone"
mock_vm.config.post.return_value = None
mock_vm.config.get.return_value = {"scsi0": "local:100/vm-102-disk-0.raw"}
mock_vm.status.start.post.return_value = "UPID:pve1:126:start"
# Mock task completion
mock_task = MagicMock()
mock_task.status.get.return_value = {"status": "stopped", "exitstatus": "OK"}
mock_node.tasks.return_value = mock_task
# Mock storage upload
mock_storage = MagicMock()
mock_storage.upload.post.return_value = {"filename": "snippets:snippets/102-user-data.yaml"}
mock_node.storage.return_value = mock_storage
with patch("time.sleep"):
result = mixin.create_vm_from_cloud_init(
name="wireshark-sat-1",
template="100",
cpu=4,
memory=4096,
nics=[
{"bridge": "vmbr0", "vlan_tag": 10}, # net0: mgmt with DHCP
{"bridge": "vmbr1", "trunk_vlan_tags": [20, 30], "dhcp": False}, # net1: trunk, no DHCP
],
cloud_init_config={"hostname": "sat-1", "runcmd": ["custom cmd"]},
timeout=120,
)
assert result["vmid"] == "102"
assert result["name"] == "wireshark-sat-1"
# Verify both NICs configured
config_call_args = mock_vm.config.post.call_args_list[0]
assert "net0" in config_call_args[1]
assert "net1" in config_call_args[1]
assert "tag=10" in config_call_args[1]["net0"]
assert "trunks=20;30" in config_call_args[1]["net1"]
assert "vmbr1" in config_call_args[1]["net1"]
# Verify DHCP config: ipconfig0 yes, ipconfig1 no
cloud_init_call_args = mock_vm.config.post.call_args_list[1] # Second call (ipconfig)
assert "ipconfig0" in cloud_init_call_args[1]
assert cloud_init_call_args[1]["ipconfig0"] == "ip=dhcp"
assert "ipconfig1" not in cloud_init_call_args[1] # net1 has no DHCP
def test_create_vm_missing_snippet_storage():
"""create_vm_from_cloud_init raises ValueError if snippet storage unavailable."""
mixin = ProxmoxVMProvisionMixin()
mixin._node_name = "pve1"
# Mock API with no snippet storage
mock_api = MagicMock()
mock_api.cluster.nextid.get.return_value = 101
mock_api.storage.get.return_value = [
{"storage": "local-lvm", "type": "lvmthin", "content": "images,rootdir", "enabled": 1}
]
mixin._api = mock_api
# Mock node for clone operation (so we get to the storage check)
mock_node = MagicMock()
mixin._node_api = MagicMock(return_value=mock_node)
mock_vm = MagicMock()
mock_node.qemu.return_value = mock_vm
mock_vm.clone.post.return_value = "UPID:pve1:123:clone"
mock_vm.config.post.return_value = None
# Mock task to allow clone to complete
mock_task = MagicMock()
mock_task.status.get.return_value = {"status": "stopped", "exitstatus": "OK"}
mock_node.tasks.return_value = mock_task
with pytest.raises(ValueError, match="snippet"):
with patch("napalm_proxmox.vm_provision_mixin.time.sleep"):
mixin.create_vm_from_cloud_init(
name="test-vm",
template="100",
cpu=2,
memory=2048,
nics=[{"bridge": "vmbr0"}],
cloud_init_config={},
)
def test_create_vm_with_disk_resize():
"""create_vm_from_cloud_init resizes disk when disk_resize_gb is set."""
mixin = ProxmoxVMProvisionMixin()
mixin._node_name = "pve1"
# Mock API hierarchy
mock_api = MagicMock()
mock_api.cluster.nextid.get.return_value = 103
mock_api.storage.get.return_value = [
{"storage": "snippets", "type": "dir", "content": "snippets", "enabled": 1}
]
mock_node = MagicMock()
mixin._api = mock_api
mixin._node_api = MagicMock(return_value=mock_node)
# Mock VM operations
mock_vm = MagicMock()
mock_node.qemu.return_value = mock_vm
mock_vm.clone.post.return_value = "UPID:pve1:127:clone"
mock_vm.config.post.return_value = None
mock_vm.config.get.return_value = {"scsi0": "local:100/vm-103-disk-0.raw"}
mock_vm.resize.put.return_value = None
mock_vm.status.start.post.return_value = "UPID:pve1:128:start"
# Mock task completion
mock_task = MagicMock()
mock_task.status.get.return_value = {"status": "stopped", "exitstatus": "OK"}
mock_node.tasks.return_value = mock_task
# Mock storage upload
mock_storage = MagicMock()
mock_storage.upload.post.return_value = {"filename": "snippets:snippets/103-user-data.yaml"}
mock_node.storage.return_value = mock_storage
with patch("time.sleep"):
result = mixin.create_vm_from_cloud_init(
name="big-vm",
template="100",
cpu=2,
memory=2048,
nics=[{"bridge": "vmbr0"}],
cloud_init_config={"hostname": "big-vm"},
disk_resize_gb=100,
timeout=120,
)
assert result["vmid"] == "103"
# Verify resize was called
mock_vm.resize.put.assert_called_once()
call_kwargs = mock_vm.resize.put.call_args[1]
assert call_kwargs["disk"] == "scsi0"
assert call_kwargs["size"] == "100G"
def test_get_vm_status_with_wait_for_ip():
"""get_vm_status(wait_for_ip=True) polls guest-agent until IP acquired."""
mixin = ProxmoxVMProvisionMixin()
mock_node = MagicMock()
mixin._node_api = MagicMock(return_value=mock_node)
# Mock VM config
mock_vm = MagicMock()
mock_node.qemu.return_value = mock_vm
mock_vm.config.get.return_value = {"net0": "virtio,bridge=vmbr0,tag=10"}
# Mock guest-agent: first no IP, then with IP
mock_agent = MagicMock()
mock_agent.network_get_interfaces.get.side_effect = [
{"result": [{"name": "eth0", "hardware-address": "aa:bb:cc:dd:ee:00"}]},
{
"result": [
{
"name": "eth0",
"hardware-address": "aa:bb:cc:dd:ee:00",
"ip-addresses": [{"ip-address": "10.0.0.100", "ip-address-type": "ipv4"}],
}
]
},
]
mock_vm.agent = mock_agent
with patch("time.sleep"):
result = mixin.get_vm_status("101", wait_for_ip=True, timeout=30, poll_interval=1)
assert result["status"] == "running"
assert result["ip_address"] == "10.0.0.100"
assert result["mac_address"] == "aa:bb:cc:dd:ee:00"
def test_get_vm_status_timeout_waiting_for_ip():
"""get_vm_status raises RuntimeError if IP acquisition times out."""
mixin = ProxmoxVMProvisionMixin()
mock_node = MagicMock()
mixin._node_api = MagicMock(return_value=mock_node)
# Mock VM config
mock_vm = MagicMock()
mock_node.qemu.return_value = mock_vm
mock_vm.config.get.return_value = {"net0": "virtio,bridge=vmbr0"}
# Mock guest-agent that never returns IP
mock_agent = MagicMock()
mock_agent.network_get_interfaces.get.return_value = {
"result": [{"name": "eth0", "hardware-address": "aa:bb:cc:dd:ee:00"}]
}
mock_vm.agent = mock_agent
with pytest.raises(RuntimeError, match="timeout|IP"):
with patch("time.sleep"):
mixin.get_vm_status("101", wait_for_ip=True, timeout=1, poll_interval=0.5)
def test_destroy_vm_success():
"""destroy_vm stops running VM, deletes it, cleans up snippets."""
mixin = ProxmoxVMProvisionMixin()
mock_node = MagicMock()
mixin._node_api = MagicMock(return_value=mock_node)
# Mock VM operations
mock_vm = MagicMock()
mock_node.qemu.return_value = mock_vm
mock_vm.config.get.return_value = {"cicustom": "user=snippets:snippets/101-user-data.yaml"}
mock_vm.status.stop.post.return_value = "UPID:pve1:125:stop"
mock_vm.delete.return_value = None
# Mock task completion
mock_task = MagicMock()
mock_task.status.get.return_value = {"status": "stopped", "exitstatus": "OK"}
mock_node.tasks.return_value = mock_task
# Mock snippet deletion (best-effort)
mock_storage = MagicMock()
mock_node.storage.return_value = mock_storage
mock_content = MagicMock()
mock_storage.content.return_value = mock_content
mock_content.delete.return_value = None
with patch("time.sleep"):
mixin.destroy_vm("101", remove_disk=True, timeout=60)
assert mock_vm.delete.called
def test_destroy_vm_already_stopped():
"""destroy_vm succeeds even if VM already stopped."""
mixin = ProxmoxVMProvisionMixin()
mock_node = MagicMock()
mixin._node_api = MagicMock(return_value=mock_node)
# Mock VM operations
mock_vm = MagicMock()
mock_node.qemu.return_value = mock_vm
mock_vm.config.get.return_value = {}
mock_vm.delete.return_value = None
with patch("time.sleep"):
mixin.destroy_vm("101", remove_disk=True, timeout=60)
# Verify delete was called (stop may fail or not be needed)
assert mock_vm.delete.called