NAPALM/napalm-proxmox
1
0
Fork 0
Code Issues 1 Pull Requests 1 Actions Packages Projects Releases Wiki Activity

feat: set_hostname — update /etc/hostname, /etc/hosts, cert, Postfix

Browse Source 
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>
This commit is contained in:
Christian Manivong
2026-06-19 16:30:45 +02:00
co-authored by Claude Sonnet 4.6
parent d1e6931ec5
commit b1ba991e6d
9 changed files with 2428 additions and 2273 deletions
Download Patch File Download Diff File Expand all files Collapse all files
napalm_proxmox/config_mixin.py
+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 = ""
napalm_proxmox/driver.py
+154 -2246
View File
File diff suppressed because it is too large Load Diff
napalm_proxmox/interfaces_mixin.py
+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
napalm_proxmox/lldp_mixin.py
+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
napalm_proxmox/routing_mixin.py
+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")
napalm_proxmox/sdn_mixin.py
+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)
napalm_proxmox/system_mixin.py
+787
View File
@@ -0,0 +1,787 @@
# 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. Proxmox node TLS certificate — regenerated via
``pvecm updatecerts -f``.
**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. Regenerate Proxmox node TLS certificate ───────────────────
# pvecm updatecerts -f is the canonical command for both standalone
# and clustered nodes (PVE 6+). Fall back to restarting pveproxy,
# which regenerates the certificate automatically when it detects a
# hostname change.
cert_out = self._exec_ssh_command(
"pvecm updatecerts -f 2>&1"
" || 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,
)
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
napalm_proxmox/utils.py
-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 ""
napalm_proxmox/vm_mixin.py
+337
View File
@@ -0,0 +1,337 @@
# 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")),
"",
)
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,
})
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")),
"",
)
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,
})
except Exception as exc:
logger.warning("get_vms: failed to list LXC containers: %s", exc)
return sorted(result, key=lambda x: x["vmid"])
_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)}