napalm-proxmox/napalm_proxmox/utils.py

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