Christian Manivong
3365207c98
- driver.py: the logger is defined after the imports (E402). The lazy paramiko client is annotated Any, as _node_ssh already is; the quoted paramiko.SSHClient named a module that is never imported there (F821). - Unused locals are gone: get_facts' hostname, get_users' groups (both read a value nobody used), the ARP flags column is _flags; the output is unchanged. - Long lines are wrapped: the apt-get and snmpd.conf commands are split into adjacent string literals, the power-action error message and the system mixin's docstring are wrapped; the text is unchanged. - get_vm_status re-raises with raise ... from e (B904). - The two except Exception: pass (closing the SSH session, the is_alive probe) now log at debug level (B110), so a swallowed failure leaves a trace. - nosec for the empty "password" field of NAPALM's get_users shape and the ssh_password_auth flag (B105); all three are false positives. - tests: unused loop variables.
270 lines
9.7 KiB
Python
270 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
|