feat(vm_provision_mixin): expose fixed_vlan_tag for SDN vnets

vnet's SDN tag (VLAN ID) is now surfaced in get_network_targets() output
instead of being silently discarded. Bridges never set this field.

napalm_proxmox/config_mixin.py

@@ -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/interfaces_mixin.py

@@ -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

@@ -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

@@ -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

@@ -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

@@ -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

napalm_proxmox/utils.py

@@ -26,28 +26,6 @@ _MAC_RE = re.compile(
 
 
 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."""
     if not mac:
         return ""

napalm_proxmox/vm_mixin.py

@@ -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)}

napalm_proxmox/vm_provision_mixin.py

@@ -0,0 +1,429 @@
+"""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 NetworkTargetDict, 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}")
+
+    def get_network_targets(self) -> List[NetworkTargetDict]:
+        """
+        List selectable network targets (bridges + SDN vnets) for a new VM's NIC.
+
+        Excludes physical NICs, bonds, and other non-bridge interface types —
+        those are never valid ``NICConfigDict.bridge`` values on Proxmox.
+        """
+        targets: List[NetworkTargetDict] = []
+
+        for iface in self._get_node_network():
+            iface_type = iface.get("type")
+            name = iface.get("iface", "")
+            if not name:
+                continue
+
+            if iface_type == "bridge":
+                vlan_aware = bool(int(iface.get("bridge_vlan_aware", 0) or 0))
+                targets.append({"name": name, "kind": "bridge", "vlan_aware": vlan_aware})
+            elif iface_type == "OVSBridge":
+                # OVS bridges tag per-port regardless of a dedicated "VLAN aware" setting.
+                targets.append({"name": name, "kind": "bridge", "vlan_aware": True})
+
+        for vnet in self._get_sdn_vnets():
+            name = vnet.get("vnet", "")
+            if not name:
+                continue
+            # A vnet's VLAN is already fixed by its zone/tag — no separate vlan_tag applies.
+            tag = vnet.get("tag")
+            fixed_vlan_tag = int(tag) if tag is not None else None
+            targets.append(
+                {
+                    "name": name,
+                    "kind": "vnet",
+                    "vlan_aware": False,
+                    "fixed_vlan_tag": fixed_vlan_tag,
+                }
+            )
+
+        return targets

pyproject.toml

@@ -26,6 +26,7 @@ requires-python = ">=3.9"
 dependencies = [
     "napalm>=5.0.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",
     "netaddr>=0.9.0",
     "requests>=2.31.0",

tests/test_vm_provision_mixin.py

@@ -0,0 +1,513 @@
+"""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
+
+
+# ---------------------------------------------------------------------------
+# get_network_targets
+# ---------------------------------------------------------------------------
+
+
+def test_get_network_targets_filters_to_bridges_and_vnets():
+    """get_network_targets excludes physical NICs/bonds; includes bridges + vnets."""
+    mixin = ProxmoxVMProvisionMixin()
+    mixin._get_node_network = MagicMock(
+        return_value=[
+            {"iface": "eth0", "type": "eth"},
+            {"iface": "bond0", "type": "bond"},
+            {"iface": "vmbr0", "type": "bridge"},
+            {"iface": "vmbr1", "type": "OVSBridge"},
+        ]
+    )
+    mixin._get_sdn_vnets = MagicMock(
+        return_value=[{"vnet": "vnet0", "zone": "zone-vlan", "tag": 10}]
+    )
+
+    targets = mixin.get_network_targets()
+    names = {t["name"] for t in targets}
+
+    assert names == {"vmbr0", "vmbr1", "vnet0"}
+    assert "eth0" not in names
+    assert "bond0" not in names
+
+
+def test_get_network_targets_linux_bridge_vlan_aware_flag():
+    """A Linux bridge's vlan_aware reflects its bridge_vlan_aware config flag."""
+    mixin = ProxmoxVMProvisionMixin()
+    mixin._get_node_network = MagicMock(
+        return_value=[
+            {"iface": "vmbr0", "type": "bridge", "bridge_vlan_aware": 1},
+            {"iface": "vmbr2", "type": "bridge"},  # no flag -> not vlan aware
+        ]
+    )
+    mixin._get_sdn_vnets = MagicMock(return_value=[])
+
+    targets = {t["name"]: t for t in mixin.get_network_targets()}
+
+    assert targets["vmbr0"]["kind"] == "bridge"
+    assert targets["vmbr0"]["vlan_aware"] is True
+    assert targets["vmbr2"]["vlan_aware"] is False
+
+
+def test_get_network_targets_ovs_bridge_always_vlan_aware():
+    """An OVS bridge is always vlan_aware, regardless of bridge_vlan_aware."""
+    mixin = ProxmoxVMProvisionMixin()
+    mixin._get_node_network = MagicMock(
+        return_value=[{"iface": "vmbr1", "type": "OVSBridge"}]
+    )
+    mixin._get_sdn_vnets = MagicMock(return_value=[])
+
+    targets = mixin.get_network_targets()
+
+    assert targets[0]["kind"] == "bridge"
+    assert targets[0]["vlan_aware"] is True
+
+
+def test_get_network_targets_sdn_vnet_never_vlan_aware():
+    """An SDN vnet is never vlan_aware — its VLAN is already fixed by zone/tag."""
+    mixin = ProxmoxVMProvisionMixin()
+    mixin._get_node_network = MagicMock(return_value=[])
+    mixin._get_sdn_vnets = MagicMock(
+        return_value=[
+            {"vnet": "vnet0", "zone": "zone-vlan", "tag": 10},
+            {"vnet": "vnet1", "zone": "zone-vlan", "tag": 20},
+        ]
+    )
+
+    targets = mixin.get_network_targets()
+
+    assert len(targets) == 2
+    assert all(t["kind"] == "vnet" for t in targets)
+    assert all(t["vlan_aware"] is False for t in targets)
+
+
+def test_get_network_targets_vnet_exposes_fixed_vlan_tag():
+    """A vnet's fixed_vlan_tag surfaces its SDN tag for display purposes."""
+    mixin = ProxmoxVMProvisionMixin()
+    mixin._get_node_network = MagicMock(return_value=[])
+    mixin._get_sdn_vnets = MagicMock(
+        return_value=[{"vnet": "vnet0", "zone": "zone-vlan", "tag": 10}]
+    )
+
+    targets = mixin.get_network_targets()
+
+    assert targets[0]["fixed_vlan_tag"] == 10
+
+
+def test_get_network_targets_vnet_without_tag_has_none_fixed_vlan_tag():
+    """A vnet with no tag (e.g. VXLAN/EVPN zone) reports fixed_vlan_tag=None."""
+    mixin = ProxmoxVMProvisionMixin()
+    mixin._get_node_network = MagicMock(return_value=[])
+    mixin._get_sdn_vnets = MagicMock(
+        return_value=[{"vnet": "vnet2", "zone": "zone-vxlan"}]
+    )
+
+    targets = mixin.get_network_targets()
+
+    assert targets[0]["fixed_vlan_tag"] is None
+
+
+def test_get_network_targets_bridge_has_no_fixed_vlan_tag():
+    """Bridges never carry a single fixed VLAN tag — only vnets do."""
+    mixin = ProxmoxVMProvisionMixin()
+    mixin._get_node_network = MagicMock(
+        return_value=[{"iface": "vmbr0", "type": "bridge", "bridge_vlan_aware": 1}]
+    )
+    mixin._get_sdn_vnets = MagicMock(return_value=[])
+
+    targets = mixin.get_network_targets()
+
+    assert "fixed_vlan_tag" not in targets[0] or targets[0]["fixed_vlan_tag"] is None