# 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