Christian Manivong
20fcf2ebc3
Closes netork#115. The suite had been red long enough that it stopped being read. Four of the fourteen failures were the tests being right. `interfaces_mixin.py` used `re.match` without importing `re`, so `get_mac_address_table` raised NameError against any node with a Linux bridge. The tests never reached that line: they mocked the API call underneath `_exec_ssh_command`, which takes two positional arguments where the doubles accepted one, and which base64-wraps the command — so a fixture keyed on "bridge fdb" appearing in the text matched nothing and the helper returned "". They mock `_exec_ssh_command` itself now, which is the driver's own seam. `is_alive` called `_resolve_node()`, which returns early without touching the API whenever a node was configured through optional_args. A dead connection reported itself alive. It probes `GET /version` now. The documented `realm` optional_arg was read into `self._realm` in `__init__` and then never used. Proxmox authenticates against "<user>@<realm>" and rejects a bare username, so the option had no effect and callers had to know to type the realm themselves. `get_vlans` filtered out entries with no member ports on one return path while the OVS path returned them, so a configured SDN VNet was visible or invisible depending on which branch ran. A VNet exists on the node whether or not anything is attached to it, and netOrk's VLAN discovery reads this. `get_ipv6_neighbors_table` was simply missing and fell through to NAPALM's stub; it is implemented against `ip -6 neigh show`, dropping FAILED entries. The rest were stale tests. The DNS fixture put an FQDN where a search domain belongs, which made `get_facts` build "pve1.pve1.example.com" and look like a driver bug. The LLDP fixture was a simplified shape that real `lldpcli show neighbors summary` does not produce — the parser matches on the ", via: LLDP" that follows the interface name. And `test_bridge_vlan_show_parsing` covered a fallback that was replaced by VM-config scanning, asserting an "interfaces" key this method has never returned; it is now a test of the fallback that exists.
269 lines
9.7 KiB
Python
269 lines
9.7 KiB
Python
# 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
|
|
import re
|
|
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_ipv6_neighbors_table(self) -> list[_JsonDict]:
|
|
"""Return the IPv6 neighbour table, read via ``ip -6 neigh show``.
|
|
|
|
The IPv6 counterpart to :meth:`get_arp_table`. Proxmox exposes no REST
|
|
endpoint for it, so it goes through the node exec helper like the ARP
|
|
table does.
|
|
|
|
Entries in FAILED state are dropped: they record an address the kernel
|
|
could not resolve, so there is no neighbour to report.
|
|
"""
|
|
raw = self._exec_ssh_command("ip -6 neigh show 2>/dev/null || true")
|
|
if not raw:
|
|
return []
|
|
|
|
entries: list[_JsonDict] = []
|
|
for line in raw.splitlines():
|
|
parts = line.split()
|
|
# "<ip> dev <iface> lladdr <mac> <STATE>" — an entry without lladdr
|
|
# never resolved and carries no neighbour.
|
|
if len(parts) < 6 or "lladdr" not in parts:
|
|
continue
|
|
state = parts[-1].upper()
|
|
if state == "FAILED":
|
|
continue
|
|
try:
|
|
iface = parts[parts.index("dev") + 1]
|
|
mac = parts[parts.index("lladdr") + 1]
|
|
except (ValueError, IndexError):
|
|
continue
|
|
entries.append(
|
|
{
|
|
"interface": iface,
|
|
"mac": utils.normalize_mac(mac),
|
|
"ip": parts[0],
|
|
# `ip neigh` reports no age; NAPALM's shape requires the key.
|
|
"age": 0.0,
|
|
"state": state,
|
|
}
|
|
)
|
|
return entries
|
|
|
|
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
|