Christian Manivong
3b873d91fe
Mechanical: ruff format, and ruff check --fix with the rule set every NAPALM driver uses (E, F, I, UP, B at line length 100, Python 3.10). What ruff cannot fix comes in the next commit.
250 lines
7.9 KiB
Python
250 lines
7.9 KiB
Python
"""Utility helpers for the Proxmox NAPALM driver.
|
|
|
|
Covers:
|
|
- Classic Linux networking (/etc/network/interfaces)
|
|
- Proxmox SDN (zones, vnets, subnets)
|
|
- Open vSwitch (OVS bridges, bonds, internal ports)
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from ipaddress import IPv4Interface, IPv6Interface, ip_interface
|
|
from typing import Any
|
|
|
|
from netaddr import EUI, AddrFormatError
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# MAC address helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_MAC_RE = re.compile(
|
|
r"(?:[0-9a-fA-F]{2}[:\-]){5}[0-9a-fA-F]{2}"
|
|
r"|[0-9a-fA-F]{12}"
|
|
)
|
|
|
|
|
|
def normalize_mac(mac: str) -> str:
|
|
"""Return lower-case colon-separated MAC or empty string."""
|
|
if not mac:
|
|
return ""
|
|
try:
|
|
eui = EUI(mac)
|
|
eui.dialect = type(
|
|
"_ColonDialect",
|
|
(),
|
|
{"word_sep": ":", "word_fmt": "%.2x", "word_size": 8, "num_words": 6},
|
|
)
|
|
return str(eui).lower()
|
|
except (AddrFormatError, ValueError):
|
|
# Best-effort normalisation
|
|
cleaned = mac.replace("-", ":").strip().lower()
|
|
return cleaned
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# IP / prefix helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def prefix_length(cidr: str) -> int:
|
|
"""Return prefix length from a CIDR string such as '192.168.1.0/24' → 24."""
|
|
try:
|
|
return int(cidr.split("/")[1])
|
|
except (IndexError, ValueError):
|
|
# No slash present — treat as host route
|
|
return 128 if ":" in cidr else 32
|
|
|
|
|
|
def parse_cidr(cidr: str) -> tuple[str, int]:
|
|
"""Return (ip_str, prefix_len) from a CIDR address."""
|
|
try:
|
|
iface = ip_interface(cidr)
|
|
return str(iface.ip), iface.network.prefixlen
|
|
except ValueError:
|
|
return cidr, -1
|
|
|
|
|
|
def addresses_from_node_network(iface_data: dict[str, Any]) -> dict[str, dict]:
|
|
"""Build an interfaces_ip sub-dict from a Proxmox node/network entry.
|
|
|
|
Proxmox returns separate fields: ``address``, ``netmask``, ``address6``,
|
|
``netmask6``. We convert those to the NAPALM ipv4/ipv6 keyed format.
|
|
"""
|
|
result: dict[str, dict] = {}
|
|
|
|
addr4 = iface_data.get("address", "")
|
|
mask4 = iface_data.get("netmask", "")
|
|
if addr4 and mask4:
|
|
try:
|
|
prefix = IPv4Interface(f"{addr4}/{mask4}").network.prefixlen
|
|
result.setdefault("ipv4", {})[addr4] = {"prefix_length": prefix}
|
|
except ValueError:
|
|
pass
|
|
|
|
addr6 = iface_data.get("address6", "")
|
|
mask6 = iface_data.get("netmask6", "")
|
|
if addr6 and mask6:
|
|
try:
|
|
prefix = IPv6Interface(f"{addr6}/{mask6}").network.prefixlen
|
|
result.setdefault("ipv6", {})[addr6] = {"prefix_length": prefix}
|
|
except ValueError:
|
|
pass
|
|
|
|
# Also handle CIDR-style fields (e.g. from SDN)
|
|
cidr4 = iface_data.get("cidr", "")
|
|
if cidr4 and "/" in cidr4 and "ipv4" not in result:
|
|
ip, plen = parse_cidr(cidr4)
|
|
if plen >= 0:
|
|
result.setdefault("ipv4", {})[ip] = {"prefix_length": plen}
|
|
|
|
cidr6 = iface_data.get("cidr6", "")
|
|
if cidr6 and "/" in cidr6 and "ipv6" not in result:
|
|
ip, plen = parse_cidr(cidr6)
|
|
if plen >= 0:
|
|
result.setdefault("ipv6", {})[ip] = {"prefix_length": plen}
|
|
|
|
return result
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Interface type helpers (classic Linux vs OVS vs SDN bridge)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
OVS_TYPES = {"OVSBridge", "OVSBond", "OVSIntPort", "OVSPort"}
|
|
|
|
SDN_BRIDGE_RE = re.compile(r"^vnet\d+$|^vmbr\d+$|^sdnbr\d+$")
|
|
|
|
|
|
def is_ovs_interface(iface_data: dict[str, Any]) -> bool:
|
|
return iface_data.get("type", "") in OVS_TYPES
|
|
|
|
|
|
def is_sdn_vnet(name: str) -> bool:
|
|
return bool(re.fullmatch(r"vnet\d+", name))
|
|
|
|
|
|
def speed_mbps(iface_data: dict[str, Any]) -> float:
|
|
"""Extract link speed (Mbit/s) from Proxmox interface data."""
|
|
raw = iface_data.get("speed", -1)
|
|
if raw is None or raw == "":
|
|
return -1.0
|
|
try:
|
|
val = float(raw)
|
|
# Proxmox reports speed in Mbit/s already for physical interfaces
|
|
return val if val > 0 else -1.0
|
|
except (TypeError, ValueError):
|
|
return -1.0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# ARP / neighbour table helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def parse_arp_output(raw: str) -> list[dict[str, Any]]:
|
|
"""Parse Linux ``arp -n`` text output into NAPALM ARP dicts.
|
|
|
|
Expected format::
|
|
|
|
Address HWtype HWaddress Flags Iface
|
|
192.168.1.1 ether aa:bb:cc:dd:ee:ff C eth0
|
|
"""
|
|
entries = []
|
|
for line in raw.splitlines():
|
|
line = line.strip()
|
|
if not line or line.startswith("Address") or line.startswith("?"):
|
|
continue
|
|
parts = line.split()
|
|
if len(parts) < 5:
|
|
continue
|
|
ip_addr, _, mac, _, iface = parts[0], parts[1], parts[2], parts[3], parts[4]
|
|
if mac in ("(incomplete)", "<incomplete>"):
|
|
continue
|
|
entries.append(
|
|
{
|
|
"interface": iface,
|
|
"mac": normalize_mac(mac),
|
|
"ip": ip_addr,
|
|
"age": -1.0,
|
|
}
|
|
)
|
|
return entries
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# SDN helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
ZONE_TYPE_MAP = {
|
|
"simple": "L3VRF",
|
|
"vlan": "L2VPN",
|
|
"qinq": "L2VPN",
|
|
"vxlan": "L2VPN",
|
|
"evpn": "L3VRF",
|
|
}
|
|
|
|
|
|
def sdn_zone_to_network_instance(zone: dict[str, Any]) -> dict[str, Any]:
|
|
"""Convert a Proxmox SDN zone entry to a NAPALM network instance dict."""
|
|
zone_id = zone.get("zone", zone.get("name", "unknown"))
|
|
zone_type = zone.get("type", "simple")
|
|
return {
|
|
"name": zone_id,
|
|
"type": ZONE_TYPE_MAP.get(zone_type, "DEFAULT_INSTANCE"),
|
|
"state": {
|
|
"route_distinguisher": zone.get("vrf-vxlan", None),
|
|
},
|
|
"interfaces": {"interface": {}},
|
|
}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# OVS helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def parse_ovs_vsctl_show(raw: str) -> dict[str, Any]:
|
|
"""Very lightweight parser for ``ovs-vsctl show`` output.
|
|
|
|
Returns a mapping of bridge_name → {ports: [port_name, …]}.
|
|
"""
|
|
bridges: dict[str, Any] = {}
|
|
current_bridge: str | None = None
|
|
for line in raw.splitlines():
|
|
stripped = line.strip()
|
|
m_bridge = re.match(r'^Bridge\s+"?([^\s"]+)"?', stripped)
|
|
if m_bridge:
|
|
current_bridge = m_bridge.group(1)
|
|
bridges[current_bridge] = {"ports": []}
|
|
continue
|
|
m_port = re.match(r'^Port\s+"?([^\s"]+)"?', stripped)
|
|
if m_port and current_bridge:
|
|
port_name = m_port.group(1)
|
|
if port_name != current_bridge: # skip the bridge self-port
|
|
bridges[current_bridge]["ports"].append(port_name)
|
|
return bridges
|
|
|
|
|
|
def parse_ovs_ofctl_dump_flows(raw: str) -> list[dict[str, Any]]:
|
|
"""Parse ``ovs-ofctl dump-flows <bridge>`` into a list of flow dicts.
|
|
|
|
Used to populate get_mac_address_table for OVS bridges.
|
|
"""
|
|
flows = []
|
|
for line in raw.splitlines():
|
|
line = line.strip()
|
|
if not line or line.startswith("NXST_FLOW") or line.startswith("OFPST_FLOW"):
|
|
continue
|
|
# Extract dl_src / dl_dst / in_port / output
|
|
mac_m = re.search(r"dl_src=([\da-f:]+)", line)
|
|
port_m = re.search(r"in_port=(\d+)", line)
|
|
if mac_m:
|
|
flows.append(
|
|
{
|
|
"mac": normalize_mac(mac_m.group(1)),
|
|
"in_port": port_m.group(1) if port_m else "",
|
|
}
|
|
)
|
|
return flows
|