Christian ManivongandClaude Sonnet 4.6
2c0a5e94bf
Split from monolithic openwrt.py into separate mixin files for interfaces, VLANs, packages, routing, wireless, LLDP, and config. Already in use by the main driver; just missing from git tracking. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
380 lines
12 KiB
Python
380 lines
12 KiB
Python
# -*- coding: utf-8 -*-
|
|
# 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 OpenWrt devices."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from typing import Any
|
|
|
|
import netaddr
|
|
from napalm.base import helpers as napalm_helpers
|
|
|
|
|
|
class OpenWrtInterfaceMixin:
|
|
"""Mixin providing interface-related NAPALM getters."""
|
|
|
|
def _get_interface_list(self) -> list[str]:
|
|
"""Return a sorted list of interface names from ``ip link show``."""
|
|
output = self._send_command("ip link show")
|
|
interfaces = []
|
|
for line in output.splitlines():
|
|
m = re.match(r"^\d+:\s+(\S+?)[@:]", line)
|
|
if m:
|
|
name = m.group(1)
|
|
if (
|
|
name not in self._EXCLUDED_INTERFACES
|
|
and not name.startswith(self._EXCLUDED_INTERFACE_PREFIXES)
|
|
):
|
|
interfaces.append(name)
|
|
return sorted(set(interfaces))
|
|
|
|
def get_interfaces(self) -> dict[str, Any]:
|
|
"""Return interface details, excluding loopback and raw radio (phy*) interfaces."""
|
|
output = self._send_command("ip link show")
|
|
return self._filter_interfaces(self._parse_ip_link(output))
|
|
|
|
@staticmethod
|
|
def _parse_ip_link(output: str) -> dict[str, Any]:
|
|
"""Parse ``ip link show`` output into NAPALM interface dicts."""
|
|
interfaces: dict[str, Any] = {}
|
|
current: str | None = None
|
|
|
|
for line in output.splitlines():
|
|
# New interface block: "2: eth0: <FLAGS> mtu 1500 ..."
|
|
m = re.match(
|
|
r"^\d+:\s+(\S+?)[@:].*<([^>]*)>.*\bmtu\s+(\d+).*\bstate\s+(\S+)",
|
|
line,
|
|
)
|
|
if m:
|
|
name = m.group(1)
|
|
flags = m.group(2).upper()
|
|
mtu = int(m.group(3))
|
|
state = m.group(4).upper()
|
|
|
|
is_up = state in ("UP", "UNKNOWN") and "UP" in flags.split(",")
|
|
is_enabled = "UP" in flags.split(",")
|
|
|
|
interfaces[name] = {
|
|
"is_up": is_up,
|
|
"is_enabled": is_enabled,
|
|
"description": "",
|
|
"last_flapped": -1.0,
|
|
"speed": -1.0,
|
|
"mtu": mtu,
|
|
"mac_address": "",
|
|
}
|
|
current = name
|
|
continue
|
|
|
|
# MAC address line: " link/ether aa:bb:cc:dd:ee:ff ..."
|
|
if current and "link/ether" in line:
|
|
m2 = re.search(r"link/ether\s+(\S+)", line)
|
|
if m2:
|
|
try:
|
|
interfaces[current]["mac_address"] = napalm_helpers.mac(m2.group(1))
|
|
except Exception:
|
|
interfaces[current]["mac_address"] = m2.group(1)
|
|
|
|
return interfaces
|
|
|
|
def get_interfaces_ip(self) -> dict[str, Any]:
|
|
"""Return all configured IP addresses grouped by interface.
|
|
|
|
Uses ``ip addr show``.
|
|
|
|
Example output::
|
|
|
|
2: eth0: <BROADCAST,MULTICAST,UP,LOWER_UP> mtu 1500 ...
|
|
inet 192.168.1.1/24 brd 192.168.1.255 scope global eth0
|
|
inet6 fd00::1/64 scope global
|
|
"""
|
|
output = self._send_command("ip addr show")
|
|
interfaces_ip: dict[str, Any] = {}
|
|
current_iface: str | None = None
|
|
|
|
for line in output.splitlines():
|
|
# Interface line
|
|
m = re.match(r"^\d+:\s+(\S+?)[@:]", line)
|
|
if m:
|
|
current_iface = m.group(1)
|
|
continue
|
|
|
|
if current_iface is None:
|
|
continue
|
|
|
|
# IPv4
|
|
m = re.match(r"^\s+inet\s+(\S+)", line)
|
|
if m:
|
|
cidr = m.group(1)
|
|
try:
|
|
ip_net = netaddr.IPNetwork(cidr)
|
|
except (netaddr.AddrFormatError, ValueError):
|
|
continue
|
|
if current_iface not in interfaces_ip:
|
|
interfaces_ip[current_iface] = {}
|
|
interfaces_ip[current_iface].setdefault("ipv4", {})[str(ip_net.ip)] = {
|
|
"prefix_length": ip_net.prefixlen
|
|
}
|
|
continue
|
|
|
|
# IPv6
|
|
m = re.match(r"^\s+inet6\s+(\S+)", line)
|
|
if m:
|
|
cidr = m.group(1)
|
|
try:
|
|
ip_net = netaddr.IPNetwork(cidr)
|
|
except (netaddr.AddrFormatError, ValueError):
|
|
continue
|
|
if current_iface not in interfaces_ip:
|
|
interfaces_ip[current_iface] = {}
|
|
interfaces_ip[current_iface].setdefault("ipv6", {})[str(ip_net.ip)] = {
|
|
"prefix_length": ip_net.prefixlen
|
|
}
|
|
|
|
return interfaces_ip
|
|
|
|
def get_interfaces_counters(self) -> dict[str, Any]:
|
|
"""Return per-interface packet and byte counters from ``/proc/net/dev``.
|
|
|
|
``/proc/net/dev`` columns (Receive | Transmit)::
|
|
|
|
face |bytes packets errs drop fifo frame compressed multicast| \
|
|
bytes packets errs drop fifo colls carrier compressed
|
|
"""
|
|
output = self._send_command("cat /proc/net/dev")
|
|
counters: dict[str, Any] = {}
|
|
|
|
for line in output.splitlines():
|
|
# Skip header lines
|
|
if "|" in line or "Inter" in line:
|
|
continue
|
|
line_s = line.strip()
|
|
if not line_s:
|
|
continue
|
|
|
|
parts = line_s.replace(":", " ").split()
|
|
if len(parts) < 17:
|
|
continue
|
|
|
|
iface = parts[0]
|
|
try:
|
|
counters[iface] = {
|
|
"tx_errors": int(parts[10]),
|
|
"rx_errors": int(parts[3]),
|
|
"tx_discards": int(parts[11]),
|
|
"rx_discards": int(parts[4]),
|
|
"tx_octets": int(parts[9]),
|
|
"rx_octets": int(parts[1]),
|
|
"tx_unicast_packets": int(parts[10 - 1]), # packets field
|
|
"rx_unicast_packets": int(parts[2]),
|
|
"tx_multicast_packets": 0,
|
|
"rx_multicast_packets": int(parts[8]),
|
|
"tx_broadcast_packets": 0,
|
|
"rx_broadcast_packets": 0,
|
|
}
|
|
except (IndexError, ValueError):
|
|
continue
|
|
|
|
return counters
|
|
|
|
def get_arp_table(self, vrf: str = "") -> list[dict[str, Any]]:
|
|
"""Return the ARP/neighbour table.
|
|
|
|
Uses ``ip neigh show`` (preferred) which produces::
|
|
|
|
192.168.1.100 dev br-lan lladdr aa:bb:cc:dd:ee:ff REACHABLE
|
|
192.168.1.1 dev br-lan lladdr 00:11:22:33:44:55 STALE
|
|
"""
|
|
output = self._send_command(["ip neigh show", "cat /proc/net/arp"])
|
|
arp_table = []
|
|
|
|
# ip neigh show format
|
|
for line in output.splitlines():
|
|
line_s = line.strip()
|
|
if not line_s:
|
|
continue
|
|
|
|
# Skip incomplete/failed entries
|
|
if "FAILED" in line_s or "INCOMPLETE" in line_s:
|
|
continue
|
|
|
|
# ip neigh show: "192.168.1.100 dev br-lan lladdr aa:bb:cc:dd:ee:ff REACHABLE"
|
|
m = re.match(
|
|
r"^(\S+)\s+dev\s+(\S+)\s+lladdr\s+(\S+)",
|
|
line_s,
|
|
re.I,
|
|
)
|
|
if m:
|
|
ip_addr = m.group(1)
|
|
interface = m.group(2)
|
|
mac_raw = m.group(3)
|
|
|
|
try:
|
|
netaddr.IPAddress(ip_addr)
|
|
except (netaddr.AddrFormatError, ValueError):
|
|
continue
|
|
|
|
try:
|
|
mac_addr = napalm_helpers.mac(mac_raw)
|
|
except Exception:
|
|
mac_addr = mac_raw
|
|
|
|
arp_table.append(
|
|
{
|
|
"interface": interface,
|
|
"mac": mac_addr,
|
|
"ip": ip_addr,
|
|
"age": 0.0,
|
|
}
|
|
)
|
|
continue
|
|
|
|
# /proc/net/arp fallback: "IP address HW type Flags HW address Mask Device"
|
|
# skip header
|
|
if line_s.startswith("IP address"):
|
|
continue
|
|
parts = line_s.split()
|
|
if len(parts) >= 6:
|
|
ip_addr = parts[0]
|
|
mac_raw = parts[3]
|
|
interface = parts[5]
|
|
|
|
try:
|
|
netaddr.IPAddress(ip_addr)
|
|
except (netaddr.AddrFormatError, ValueError):
|
|
continue
|
|
|
|
if mac_raw in ("00:00:00:00:00:00", ""):
|
|
continue
|
|
|
|
try:
|
|
mac_addr = napalm_helpers.mac(mac_raw)
|
|
except Exception:
|
|
mac_addr = mac_raw
|
|
|
|
arp_table.append(
|
|
{
|
|
"interface": interface,
|
|
"mac": mac_addr,
|
|
"ip": ip_addr,
|
|
"age": 0.0,
|
|
}
|
|
)
|
|
|
|
return arp_table
|
|
|
|
def get_mac_address_table(self) -> list[dict[str, Any]]:
|
|
"""Return the bridge forwarding database (MAC address table).
|
|
|
|
Uses ``bridge fdb show`` which produces::
|
|
|
|
aa:bb:cc:dd:ee:ff dev br-lan master br-lan permanent
|
|
11:22:33:44:55:66 dev eth0.1 vlan 1 master br-lan
|
|
"""
|
|
output = self._send_command("bridge fdb show")
|
|
mac_table = []
|
|
|
|
for line in output.splitlines():
|
|
line_s = line.strip()
|
|
if not line_s:
|
|
continue
|
|
|
|
m = re.match(r"^(\S+)\s+dev\s+(\S+)", line_s)
|
|
if not m:
|
|
continue
|
|
|
|
mac_raw = m.group(1)
|
|
interface = m.group(2)
|
|
|
|
# Skip broadcast/multicast self-entries that are always present
|
|
if mac_raw.lower() in ("ff:ff:ff:ff:ff:ff", "33:33:00:00:00:01"):
|
|
continue
|
|
|
|
static = "permanent" in line_s or "static" in line_s
|
|
|
|
# Extract VLAN if present: "vlan 10"
|
|
vlan = 0
|
|
vlan_m = re.search(r"\bvlan\s+(\d+)", line_s)
|
|
if vlan_m:
|
|
vlan = int(vlan_m.group(1))
|
|
|
|
try:
|
|
mac_addr = napalm_helpers.mac(mac_raw)
|
|
except Exception:
|
|
mac_addr = mac_raw
|
|
|
|
mac_table.append(
|
|
{
|
|
"mac": mac_addr,
|
|
"interface": interface,
|
|
"vlan": vlan,
|
|
"static": static,
|
|
"active": True,
|
|
"moves": None,
|
|
"last_move": None,
|
|
}
|
|
)
|
|
|
|
return mac_table
|
|
|
|
def get_ipv6_neighbors_table(self) -> list[dict[str, Any]]:
|
|
"""Return the IPv6 neighbour table from ``ip -6 neigh show``.
|
|
|
|
Example output::
|
|
|
|
2001:db8::1 dev eth0 lladdr aa:bb:cc:dd:ee:ff REACHABLE
|
|
fe80::1 dev br-lan lladdr 11:22:33:44:55:66 STALE
|
|
"""
|
|
output = self._send_command("ip -6 neigh show")
|
|
table = []
|
|
|
|
for line in output.splitlines():
|
|
line_s = line.strip()
|
|
if not line_s or "FAILED" in line_s or "INCOMPLETE" in line_s:
|
|
continue
|
|
|
|
m = re.match(
|
|
r"^(\S+)\s+dev\s+(\S+)\s+lladdr\s+(\S+)\s+(\S+)",
|
|
line_s,
|
|
re.I,
|
|
)
|
|
if not m:
|
|
continue
|
|
|
|
ip_addr = m.group(1)
|
|
interface = m.group(2)
|
|
mac_raw = m.group(3)
|
|
state = m.group(4)
|
|
|
|
try:
|
|
netaddr.IPAddress(ip_addr, version=6)
|
|
except (netaddr.AddrFormatError, ValueError):
|
|
continue
|
|
|
|
try:
|
|
mac_addr = napalm_helpers.mac(mac_raw)
|
|
except Exception:
|
|
mac_addr = mac_raw
|
|
|
|
table.append({
|
|
"interface": interface,
|
|
"mac": mac_addr,
|
|
"ip": ip_addr,
|
|
"age": -1.0,
|
|
"state": state,
|
|
})
|
|
|
|
return table
|