mirror of
https://github.com/kernelkit/infix.git
synced 2026-07-29 20:23:01 +02:00
yanger: Break up ietf-interfaces implementation into modules
This commit is contained in:
Tobias Waldekranz
@@ -1,15 +1,5 @@
|
||||
LOG = None
|
||||
|
||||
def lookup(obj, *keys):
|
||||
"""This function returns a value from a nested json object"""
|
||||
curr = obj
|
||||
for key in keys:
|
||||
if isinstance(curr, dict) and key in curr:
|
||||
curr = curr[key]
|
||||
else:
|
||||
return None
|
||||
return curr
|
||||
|
||||
|
||||
def insert(obj, *path_and_value):
|
||||
""""This function inserts a value into a nested json object"""
|
||||
|
||||
@@ -33,7 +33,7 @@ class Host(abc.ABC):
|
||||
def run_multiline(self, cmd, default=None):
|
||||
"""Get lines of stdout of cmd"""
|
||||
try:
|
||||
txt = self.run(cmd, log=(default is None))
|
||||
txt = self.run(tuple(cmd), log=(default is None))
|
||||
return txt.splitlines()
|
||||
except:
|
||||
if default is not None:
|
||||
@@ -43,7 +43,7 @@ class Host(abc.ABC):
|
||||
def run_json(self, cmd, default=None):
|
||||
"""Get JSON object from stdout of cmd"""
|
||||
try:
|
||||
txt = self.run(cmd, log=(default is None))
|
||||
txt = self.run(tuple(cmd), log=(default is None))
|
||||
return json.loads(txt)
|
||||
except:
|
||||
if default is not None:
|
||||
@@ -74,7 +74,7 @@ class Localhost(Host):
|
||||
def now(self):
|
||||
return datetime.datetime.now(tz=datetime.timezone.utc)
|
||||
|
||||
# @functools.cache
|
||||
@functools.cache
|
||||
def run(self, cmd, default=None, log=True):
|
||||
try:
|
||||
result = subprocess.run(cmd, check=True, text=True,
|
||||
@@ -107,7 +107,7 @@ class Localhost(Host):
|
||||
class Remotehost(Localhost):
|
||||
def __init__(self, wrap, basedir):
|
||||
super().__init__()
|
||||
self.wrap = wrap.split()
|
||||
self.wrap = tuple(wrap.split())
|
||||
self.basedir = basedir
|
||||
if basedir:
|
||||
for subdir in ("rootfs", "run"):
|
||||
@@ -146,7 +146,7 @@ class Remotehost(Localhost):
|
||||
return out
|
||||
|
||||
def read(self, path):
|
||||
out = self._run(["cat", path], default="", log=False)
|
||||
out = self._run(("cat", path), default="", log=False)
|
||||
|
||||
if self.basedir:
|
||||
dirname = os.path.join(self.basedir, "rootfs", os.path.dirname(path[1:]))
|
||||
@@ -154,6 +154,8 @@ class Remotehost(Localhost):
|
||||
with open(os.path.join(self.basedir, "rootfs", path[1:]), "w") as f:
|
||||
f.write(out)
|
||||
|
||||
return out
|
||||
|
||||
|
||||
class Testhost(Host):
|
||||
def SlugOf(cmd):
|
||||
|
||||
@@ -1,564 +1,11 @@
|
||||
from ..common import insert, lookup, LOG
|
||||
from ..host import HOST
|
||||
|
||||
|
||||
def json_get_yang_type(iface_in):
|
||||
if iface_in['link_type'] == "loopback":
|
||||
return "infix-if-type:loopback"
|
||||
|
||||
if iface_in['link_type'] in ("gre", "gre6"):
|
||||
return "infix-if-type:gre"
|
||||
|
||||
if iface_in['link_type'] != "ether":
|
||||
return "infix-if-type:other"
|
||||
|
||||
if 'parentbus' in iface_in and iface_in['parentbus'] == "virtio":
|
||||
return "infix-if-type:etherlike"
|
||||
|
||||
if 'linkinfo' not in iface_in:
|
||||
return "infix-if-type:ethernet"
|
||||
|
||||
if 'info_kind' not in iface_in['linkinfo']:
|
||||
return "infix-if-type:ethernet"
|
||||
|
||||
if iface_in['linkinfo']['info_kind'] == "veth":
|
||||
return "infix-if-type:veth"
|
||||
|
||||
if iface_in['linkinfo']['info_kind'] in ("gretap", "ip6gretap"):
|
||||
return "infix-if-type:gretap"
|
||||
|
||||
if iface_in['linkinfo']['info_kind'] == "vlan":
|
||||
return "infix-if-type:vlan"
|
||||
|
||||
if iface_in['linkinfo']['info_kind'] == "bridge":
|
||||
return "infix-if-type:bridge"
|
||||
|
||||
if iface_in['linkinfo']['info_kind'] == "dsa":
|
||||
return "infix-if-type:ethernet"
|
||||
|
||||
if iface_in['linkinfo']['info_kind'] == "dummy":
|
||||
return "infix-if-type:dummy"
|
||||
|
||||
# Fallback
|
||||
return "infix-if-type:ethernet"
|
||||
|
||||
|
||||
def json_get_yang_origin(addr):
|
||||
"""Translate kernel IP address origin to YANG"""
|
||||
xlate = {
|
||||
"kernel_ll": "link-layer",
|
||||
"kernel_ra": "link-layer",
|
||||
"static": "static",
|
||||
"dhcp": "dhcp",
|
||||
"random": "random",
|
||||
}
|
||||
proto = addr['protocol']
|
||||
|
||||
if proto in ("kernel_ll", "kernel_ra"):
|
||||
if "stable-privacy" in addr:
|
||||
return "random"
|
||||
|
||||
return xlate.get(proto, "other")
|
||||
|
||||
|
||||
|
||||
|
||||
def iface_is_dsa(iface_in):
|
||||
"""Check if interface is a DSA/intra-switch port"""
|
||||
if "linkinfo" not in iface_in:
|
||||
return False
|
||||
if "info_kind" not in iface_in['linkinfo']:
|
||||
return False
|
||||
if iface_in['linkinfo']['info_kind'] != "dsa":
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
def get_bridge_port_pvid(ifname):
|
||||
data = HOST.run_json(['bridge', '-j', 'vlan', 'show', 'dev', ifname])
|
||||
if len(data) != 1:
|
||||
return None
|
||||
|
||||
iface = data[0]
|
||||
|
||||
for vlan in iface['vlans']:
|
||||
if 'flags' in vlan and 'PVID' in vlan['flags']:
|
||||
return vlan['vlan']
|
||||
|
||||
return None
|
||||
|
||||
|
||||
def get_bridge_port_stp_state(ifname):
|
||||
data = HOST.run_json(['bridge', '-j', 'link', 'show', 'dev', ifname])
|
||||
if len(data) != 1:
|
||||
return None
|
||||
|
||||
iface = data[0]
|
||||
|
||||
states = ['disabled', 'listening', 'learning', 'forwarding', 'blocking']
|
||||
if 'state' in iface and iface['state'] in states:
|
||||
return iface['state']
|
||||
|
||||
return None
|
||||
|
||||
|
||||
def get_brport_multicast(ifname):
|
||||
"""Check if multicast snooping is enabled on bridge, default: nope"""
|
||||
data = HOST.run_json(['mctl', '-p', 'show', 'igmp', 'json'], default={})
|
||||
multicast = {}
|
||||
|
||||
if ifname in data.get('fast-leave-ports', []):
|
||||
multicast["fast-leave"] = True
|
||||
else:
|
||||
multicast["fast-leave"] = False
|
||||
|
||||
if ifname in data.get('multicast-router-ports', []):
|
||||
multicast["router"] = "permanent"
|
||||
else:
|
||||
multicast["router"] = "auto"
|
||||
|
||||
return multicast
|
||||
|
||||
|
||||
# We always get all interfaces for two reasons.
|
||||
# 1) To increase speed on large iron with many ports.
|
||||
# 2) To simplify testing (single dummy file ip-link-show.json).
|
||||
def get_ip_link():
|
||||
"""Fetch interface link information from kernel"""
|
||||
return HOST.run_json(['ip', '-s', '-d', '-j', 'link', 'show'])
|
||||
|
||||
def netns_get_ip_link(netns):
|
||||
"""Fetch interface link information from within a network namespace"""
|
||||
return HOST.run_json(['ip', 'netns', 'exec', netns, 'ip', '-s', '-d', '-j', 'link', 'show'])
|
||||
|
||||
def get_ip_addr():
|
||||
"""Fetch interface address information from kernel"""
|
||||
return HOST.run_json(['ip', '-j', 'addr', 'show'])
|
||||
|
||||
def netns_get_ip_addr(netns):
|
||||
"""Fetch interface address information from within a network namespace"""
|
||||
return HOST.run_json(['ip', 'netns', 'exec', netns, 'ip', '-j', 'addr', 'show'])
|
||||
|
||||
def get_netns_list():
|
||||
"""Fetch a list of network namespaces"""
|
||||
return HOST.run_json(['ip', '-j', 'netns', 'list'], [])
|
||||
|
||||
def netns_find_ifname(ifname):
|
||||
"""Find which network namespace owns ifname (if any)"""
|
||||
for netns in get_netns_list():
|
||||
for iface in netns_get_ip_link(netns['name']):
|
||||
if 'ifalias' in iface and iface['ifalias'] == ifname:
|
||||
return netns['name']
|
||||
return None
|
||||
|
||||
def netns_ifindex_to_ifname(ifindex):
|
||||
"""Look through all network namespaces for an interface index and return its name"""
|
||||
for netns in get_netns_list():
|
||||
for iface in netns_get_ip_link(netns['name']):
|
||||
if iface['ifindex'] == ifindex:
|
||||
if 'ifalias' in iface:
|
||||
return iface['ifalias']
|
||||
if 'ifname' in iface:
|
||||
return iface['ifname']
|
||||
return None
|
||||
|
||||
return None
|
||||
|
||||
def add_bridge_port_common(ifname, iface_in, iface_out):
|
||||
li = iface_in.get("linkinfo", {})
|
||||
if not (li.get("info_slave_kind") == "bridge" or \
|
||||
li.get("info_kind") == "bridge"):
|
||||
return
|
||||
|
||||
pvid = get_bridge_port_pvid(ifname)
|
||||
if pvid is not None:
|
||||
insert(iface_out, "infix-interfaces:bridge-port", "pvid", pvid)
|
||||
|
||||
def add_bridge_port_lower(ifname, iface_in, iface_out):
|
||||
li = iface_in.get("linkinfo", {})
|
||||
if not li.get("info_slave_kind") == "bridge":
|
||||
return
|
||||
|
||||
insert(iface_out, "infix-interfaces:bridge-port", "bridge", iface_in['master'])
|
||||
|
||||
stp_state = get_bridge_port_stp_state(ifname)
|
||||
if stp_state is not None:
|
||||
insert(iface_out, "infix-interfaces:bridge-port", "stp-state", stp_state)
|
||||
|
||||
multicast = get_brport_multicast(ifname)
|
||||
insert(iface_out, "infix-interfaces:bridge-port", "multicast", multicast)
|
||||
|
||||
|
||||
def add_gre(iface_in, iface_out):
|
||||
if 'link_type' in iface_in:
|
||||
val = json_get_yang_type(iface_in)
|
||||
if val != "infix-if-type:gre" and val != "infix-if-type:gretap":
|
||||
return;
|
||||
gre={}
|
||||
info_data=iface_in.get("linkinfo", {}).get("info_data", {})
|
||||
gre["local"] = info_data.get("local")
|
||||
gre["remote"] = info_data.get("remote")
|
||||
insert(iface_out, "infix-interfaces:gre", gre)
|
||||
|
||||
|
||||
def add_ip_link(ifname, iface_in, iface_out):
|
||||
if 'ifname' in iface_in:
|
||||
iface_out['name'] = ifname
|
||||
|
||||
if 'ifindex' in iface_in:
|
||||
iface_out['if-index'] = iface_in['ifindex']
|
||||
|
||||
if 'ifalias' in iface_in:
|
||||
iface_out['description'] = iface_in['ifalias']
|
||||
|
||||
if 'address' in iface_in and not "POINTOPOINT" in iface_in["flags"]:
|
||||
iface_out['phys-address'] = iface_in['address']
|
||||
|
||||
add_bridge_port_common(ifname, iface_in, iface_out)
|
||||
add_bridge_port_lower(ifname, iface_in, iface_out)
|
||||
|
||||
if not iface_is_dsa(iface_in):
|
||||
if iface_in.get('link'):
|
||||
insert(iface_out, "infix-interfaces:vlan", "lower-layer-if", iface_in['link'])
|
||||
elif 'link_index' in iface_in:
|
||||
# 'link_index' is the only reference we have if the link iface is in a namespace
|
||||
lower = netns_ifindex_to_ifname(iface_in['link_index'])
|
||||
if lower:
|
||||
insert(iface_out, "infix-interfaces:vlan", "lower-layer-if", lower)
|
||||
|
||||
if 'flags' in iface_in:
|
||||
iface_out['admin-status'] = "up" if "UP" in iface_in['flags'] else "down"
|
||||
|
||||
if 'operstate' in iface_in:
|
||||
xlate = {
|
||||
"DOWN": "down",
|
||||
"UP": "up",
|
||||
"DORMANT": "dormant",
|
||||
"TESTING": "testing",
|
||||
"LOWERLAYERDOWN": "lower-layer-down",
|
||||
"NOTPRESENT": "not-present"
|
||||
}
|
||||
val = xlate.get(iface_in['operstate'], "unknown")
|
||||
iface_out['oper-status'] = val
|
||||
|
||||
if 'link_type' in iface_in:
|
||||
val = json_get_yang_type(iface_in)
|
||||
iface_out['type'] = val
|
||||
add_gre(iface_in, iface_out)
|
||||
|
||||
val = lookup(iface_in, "stats64", "rx", "bytes")
|
||||
if val is not None:
|
||||
insert(iface_out, "statistics", "out-octets", str(val))
|
||||
|
||||
val = lookup(iface_in, "stats64", "tx", "bytes")
|
||||
if val is not None:
|
||||
insert(iface_out, "statistics", "in-octets", str(val))
|
||||
|
||||
|
||||
def add_ip_addr(ifname, iface_in, iface_out):
|
||||
if 'mtu' in iface_in and ifname != "lo":
|
||||
insert(iface_out, "ietf-ip:ipv4", "mtu", iface_in['mtu'])
|
||||
|
||||
val = HOST.read(f"/proc/sys/net/ipv6/conf/{ifname}/mtu")
|
||||
if val is not None:
|
||||
insert(iface_out, "ietf-ip:ipv6", "mtu", int(val.strip()))
|
||||
|
||||
if 'addr_info' in iface_in:
|
||||
inet = []
|
||||
inet6 = []
|
||||
|
||||
for addr in iface_in['addr_info']:
|
||||
new = {}
|
||||
|
||||
if 'family' not in addr:
|
||||
LOG.error("'family' missing from 'addr_info'")
|
||||
continue
|
||||
|
||||
if 'local' in addr:
|
||||
new['ip'] = addr['local']
|
||||
if 'prefixlen' in addr:
|
||||
new['prefix-length'] = addr['prefixlen']
|
||||
if 'protocol' in addr:
|
||||
new['origin'] = json_get_yang_origin(addr)
|
||||
|
||||
if addr['family'] == "inet":
|
||||
inet.append(new)
|
||||
elif addr['family'] == "inet6":
|
||||
inet6.append(new)
|
||||
else:
|
||||
LOG.error("invalid 'family' in 'addr_info'")
|
||||
sys.exit(1)
|
||||
|
||||
insert(iface_out, "ietf-ip:ipv4", "address", inet)
|
||||
insert(iface_out, "ietf-ip:ipv6", "address", inet6)
|
||||
|
||||
|
||||
def add_ethtool_groups(ifname, iface_out):
|
||||
"""Fetch interface counters from kernel (need new JSON format!)"""
|
||||
cmd = ['ethtool', '--json', '-S', ifname, '--all-groups']
|
||||
try:
|
||||
data = HOST.run_json(cmd)
|
||||
if len(data) != 1:
|
||||
LOG.warning("%s: no counters available, skipping.", ifname)
|
||||
return
|
||||
except subprocess.CalledProcessError:
|
||||
# Allow comand to fail, not all NICs support --json yet
|
||||
return
|
||||
|
||||
iface_in = data[0]
|
||||
|
||||
# TODO: room for improvement, the "frame" creation could be more dynamic.
|
||||
if "eth-mac" in iface_in or "rmon" in iface_in:
|
||||
insert(iface_out, "ieee802-ethernet-interface:ethernet", "statistics", "frame", {})
|
||||
frame = iface_out['ieee802-ethernet-interface:ethernet']['statistics']['frame']
|
||||
|
||||
if "eth-mac" in iface_in:
|
||||
mac_in = iface_in['eth-mac']
|
||||
|
||||
if "FramesTransmittedOK" in mac_in:
|
||||
frame['out-frames'] = str(mac_in['FramesTransmittedOK'])
|
||||
if "MulticastFramesXmittedOK" in mac_in:
|
||||
frame['out-multicast-frames'] = str(mac_in['MulticastFramesXmittedOK'])
|
||||
if "BroadcastFramesXmittedOK" in mac_in:
|
||||
frame['out-broadcast-frames'] = str(mac_in['BroadcastFramesXmittedOK'])
|
||||
if "FramesReceivedOK" in mac_in:
|
||||
frame['in-frames'] = str(mac_in['FramesReceivedOK'])
|
||||
if "MulticastFramesReceivedOK" in mac_in:
|
||||
frame['in-multicast-frames'] = str(mac_in['MulticastFramesReceivedOK'])
|
||||
if "BroadcastFramesReceivedOK" in mac_in:
|
||||
frame['in-broadcast-frames'] = str(mac_in['BroadcastFramesReceivedOK'])
|
||||
if "FrameCheckSequenceErrors" in mac_in:
|
||||
frame['in-error-fcs-frames'] = str(mac_in['FrameCheckSequenceErrors'])
|
||||
if "FramesLostDueToIntMACRcvError" in mac_in:
|
||||
frame['in-error-mac-internal-frames'] = str(mac_in['FramesLostDueToIntMACRcvError'])
|
||||
|
||||
if "OctetsTransmittedOK" in mac_in:
|
||||
frame['infix-ethernet-interface:out-good-octets'] = str(mac_in['OctetsTransmittedOK'])
|
||||
if "OctetsReceivedOK" in mac_in:
|
||||
frame['infix-ethernet-interface:in-good-octets'] = str(mac_in['OctetsReceivedOK'])
|
||||
|
||||
tot = 0
|
||||
found = False
|
||||
if "FramesReceivedOK" in mac_in:
|
||||
tot += mac_in['FramesReceivedOK']
|
||||
found = True
|
||||
if "FrameCheckSequenceErrors" in mac_in:
|
||||
tot += mac_in['FrameCheckSequenceErrors']
|
||||
found = True
|
||||
if "FramesLostDueToIntMACRcvError" in mac_in:
|
||||
tot += mac_in['FramesLostDueToIntMACRcvError']
|
||||
found = True
|
||||
if "AlignmentErrors" in mac_in:
|
||||
tot += mac_in['AlignmentErrors']
|
||||
found = True
|
||||
if "etherStatsOversizePkts" in mac_in:
|
||||
tot += mac_in['etherStatsOversizePkts']
|
||||
found = True
|
||||
if "etherStatsJabbers" in mac_in:
|
||||
tot += mac_in['etherStatsJabbers']
|
||||
found = True
|
||||
if found:
|
||||
frame['in-total-frames'] = str(tot)
|
||||
|
||||
if "rmon" in iface_in:
|
||||
rmon_in = iface_in['rmon']
|
||||
|
||||
if "undersize_pkts" in rmon_in:
|
||||
frame['in-error-undersize-frames'] = str(rmon_in['undersize_pkts'])
|
||||
|
||||
tot = 0
|
||||
found = False
|
||||
if "etherStatsJabbers" in rmon_in:
|
||||
tot += rmon_in['etherStatsJabbers']
|
||||
found = True
|
||||
if "etherStatsOversizePkts" in rmon_in:
|
||||
tot += rmon_in['etherStatsOversizePkts']
|
||||
found = True
|
||||
if found:
|
||||
frame['in-error-oversize-frames'] = str(tot)
|
||||
|
||||
def add_ethtool_std(ifname, iface_out):
|
||||
"""Fetch interface speed/duplex/autoneg from kernel"""
|
||||
keys = ['Speed', 'Duplex', 'Auto-negotiation']
|
||||
result = {}
|
||||
|
||||
lines = HOST.run_multiline(['ethtool', ifname])
|
||||
for line in lines:
|
||||
line = line.strip()
|
||||
key = line.split(':', 1)[0].strip()
|
||||
if key in keys:
|
||||
key, value = line.split(':', 1)
|
||||
result[key.strip()] = value.strip()
|
||||
|
||||
if "Auto-negotiation" in result:
|
||||
if result['Auto-negotiation'] == "on":
|
||||
insert(iface_out, "ieee802-ethernet-interface:ethernet", "auto-negotiation", "enable", True)
|
||||
else:
|
||||
insert(iface_out, "ieee802-ethernet-interface:ethernet", "auto-negotiation", "enable", False)
|
||||
|
||||
if "Duplex" in result:
|
||||
if result['Duplex'] == "Half":
|
||||
insert(iface_out, "ieee802-ethernet-interface:ethernet", "duplex", "half")
|
||||
elif result['Duplex'] == "Full":
|
||||
insert(iface_out, "ieee802-ethernet-interface:ethernet", "duplex", "full")
|
||||
else:
|
||||
insert(iface_out, "ieee802-ethernet-interface:ethernet", "duplex", "unknown")
|
||||
|
||||
if "Speed" in result and result['Speed'] != "Unknown!":
|
||||
# Avoid importing re (performance)
|
||||
num = ''.join(filter(str.isdigit, result['Speed']))
|
||||
if num:
|
||||
num = round((int(num) / 1000), 3)
|
||||
insert(iface_out, "ieee802-ethernet-interface:ethernet", "speed", str(num))
|
||||
|
||||
def get_querier_data(querier_data):
|
||||
multicast={}
|
||||
|
||||
if not querier_data:
|
||||
multicast["snooping"] = False
|
||||
return multicast
|
||||
multicast["snooping"] = True
|
||||
if(querier_data.get("query-interval")):
|
||||
multicast["query-interval"] = querier_data["query-interval"]
|
||||
|
||||
return multicast
|
||||
|
||||
def get_multicast_filters(filters):
|
||||
multicast_filters=[]
|
||||
for f in filters:
|
||||
multicast_filter={}
|
||||
multicast_filter["group"] = f["group"]
|
||||
multicast_filter["ports"] = []
|
||||
for p in f["ports"]:
|
||||
port={}
|
||||
port["port"] = p
|
||||
multicast_filter["ports"].append(port)
|
||||
multicast_filters.append(multicast_filter)
|
||||
return multicast_filters
|
||||
|
||||
def add_mdb_to_bridge(brname, iface_out, mc_status):
|
||||
filters = [entry for entry in mc_status.get("multicast-groups", []) if entry.get('vid') == None and entry.get('bridge') == brname]
|
||||
querier = next((querier for querier in mc_status.get('multicast-queriers', []) if (querier.get("interface", "") == brname) and (querier.get("vid") is None)), None)
|
||||
multicast = get_querier_data(querier)
|
||||
multicast_filters = get_multicast_filters(filters)
|
||||
insert(iface_out, "infix-interfaces:bridge", "multicast", multicast)
|
||||
insert(iface_out, "infix-interfaces:bridge", "multicast-filters", "multicast-filter", multicast_filters)
|
||||
|
||||
def add_container_ifaces(yang_ifaces):
|
||||
"""Add all podman interfaces with limited data"""
|
||||
interfaces={}
|
||||
try:
|
||||
containers = HOST.run_json(['podman', 'ps', '-a', '--format=json'], default=[])
|
||||
except Exception as e:
|
||||
logging.error(f"Error, unable to run podman: {e}")
|
||||
return
|
||||
|
||||
for container in containers:
|
||||
name = container.get('Names', ['Unknown'])[0]
|
||||
networks = container.get('Networks', [])
|
||||
|
||||
for network in networks:
|
||||
if not network in interfaces:
|
||||
interfaces[network] = []
|
||||
if name not in interfaces[network]:
|
||||
interfaces[network].append(name)
|
||||
|
||||
for ifname, containers in interfaces.items():
|
||||
iface_out = {}
|
||||
iface_out['name'] = ifname
|
||||
iface_out['type'] = "infix-if-type:other" # Fallback
|
||||
insert(iface_out, "infix-interfaces:container-network", "containers", containers)
|
||||
|
||||
netns = netns_find_ifname(ifname)
|
||||
if netns is not None:
|
||||
ip_link_data = netns_get_ip_link(netns)
|
||||
ip_link_data = next((d for d in ip_link_data if d.get('ifalias') == ifname), None)
|
||||
add_ip_link(ifname, ip_link_data, iface_out)
|
||||
|
||||
yang_ifaces.append(iface_out)
|
||||
|
||||
# Helper function to add tagged/untagged interfaces to a vlan dict in a list
|
||||
def _add_vlan_iface(vlans, multicast_filter, multicast, vid, key, val):
|
||||
for d in vlans:
|
||||
if d['vid'] == vid:
|
||||
if key in d:
|
||||
d[key].append(val)
|
||||
else:
|
||||
d[key] = [val]
|
||||
return
|
||||
|
||||
vlans.append({'vid': vid, "multicast-filters": {"multicast-filter": multicast_filter}, "multicast": multicast, key: [val]})
|
||||
|
||||
def add_vlans_to_bridge(brname, iface_out, mc_status):
|
||||
slaves = [] # Contains all interfaces that has this bridge as 'master'
|
||||
for iface in HOST.run_json(['bridge', '-j', 'link']):
|
||||
if "master" in iface and iface['master'] == brname:
|
||||
slaves.append(iface['ifname'])
|
||||
|
||||
vlans = [] # Contains all vlans and slaves belonging to this bridge
|
||||
for iface in HOST.run_json(['bridge', '-j', 'vlan']):
|
||||
if iface['ifname'] not in slaves and iface['ifname'] != brname:
|
||||
continue
|
||||
for vlan in iface['vlans']:
|
||||
querier = next((querier for querier in mc_status.get('multicast-queriers', []) if (querier.get("vid") == vlan['vlan'])), None)
|
||||
filters = [entry for entry in mc_status.get("multicast-groups", []) if (entry.get("vid") == vlan["vlan"])]
|
||||
if 'flags' in vlan and 'Egress Untagged' in vlan['flags']:
|
||||
_add_vlan_iface(vlans, get_multicast_filters(filters), get_querier_data(querier), vlan['vlan'], 'untagged', iface['ifname'])
|
||||
else:
|
||||
_add_vlan_iface(vlans, get_multicast_filters(filters), get_querier_data(querier), vlan['vlan'], 'tagged', iface['ifname'])
|
||||
|
||||
insert(iface_out, "infix-interfaces:bridge", "vlans", "vlan", vlans)
|
||||
|
||||
def get_iface_data(ifname, ip_link_data, ip_addr_data):
|
||||
iface_out = {}
|
||||
|
||||
add_ip_link(ifname, ip_link_data, iface_out)
|
||||
add_ip_addr(ifname, ip_addr_data, iface_out)
|
||||
|
||||
if 'type' in iface_out and iface_out['type'] == "infix-if-type:ethernet":
|
||||
add_ethtool_groups(ifname, iface_out)
|
||||
add_ethtool_std(ifname, iface_out)
|
||||
|
||||
if 'type' in iface_out and iface_out['type'] == "infix-if-type:bridge":
|
||||
# Fail silent, multicast snooping may not be enabled on bridge
|
||||
mc_status = HOST.run_json(['mctl', '-p', 'show', 'igmp', 'json'], default={})
|
||||
|
||||
add_vlans_to_bridge(ifname, iface_out, mc_status)
|
||||
add_mdb_to_bridge(ifname, iface_out, mc_status)
|
||||
|
||||
return iface_out
|
||||
|
||||
def _add_interface(ifname, ip_link_data, ip_addr_data, yang_ifaces):
|
||||
# We expect both ip addr and link data to exist.
|
||||
if not ip_link_data or not ip_addr_data:
|
||||
return
|
||||
|
||||
# Skip internal interfaces.
|
||||
if 'group' in ip_link_data and ip_link_data['group'] == "internal":
|
||||
return
|
||||
|
||||
yang_ifaces.append(get_iface_data(ifname, ip_link_data, ip_addr_data))
|
||||
from . import container
|
||||
from . import link
|
||||
|
||||
def operational(ifname=None):
|
||||
out = {
|
||||
return {
|
||||
"ietf-interfaces:interfaces": {
|
||||
"interface": []
|
||||
}
|
||||
"interface":
|
||||
link.interfaces(ifname) +
|
||||
container.interfaces(ifname),
|
||||
},
|
||||
}
|
||||
out_ifaces = out["ietf-interfaces:interfaces"]["interface"]
|
||||
|
||||
ip_link_data = get_ip_link()
|
||||
ip_addr_data = get_ip_addr()
|
||||
|
||||
if ifname:
|
||||
ip_link_data = next((d for d in ip_link_data if d.get('ifname') == ifname), None)
|
||||
ip_addr_data = next((d for d in ip_addr_data if d.get('ifname') == ifname), None)
|
||||
_add_interface(ifname, ip_link_data, ip_addr_data, out_ifaces)
|
||||
else:
|
||||
for link in ip_link_data:
|
||||
addr = next((d for d in ip_addr_data if d.get('ifname') == link["ifname"]), None)
|
||||
_add_interface(link["ifname"], link, addr, out_ifaces)
|
||||
|
||||
add_container_ifaces(out_ifaces)
|
||||
|
||||
return out
|
||||
|
||||
@@ -0,0 +1,171 @@
|
||||
from functools import cache
|
||||
|
||||
from ..common import LOG
|
||||
from ..host import HOST
|
||||
|
||||
from .common import iplinks, iplinks_lower_of
|
||||
|
||||
from . import vlan
|
||||
|
||||
|
||||
@cache
|
||||
def bridge_vlan():
|
||||
return { v["ifname"]: v for v in HOST.run_json("bridge -j vlan show".split(), []) }
|
||||
|
||||
|
||||
def lower(iplink):
|
||||
lower = {}
|
||||
|
||||
if brpvlans := bridge_vlan().get(iplink["ifname"]):
|
||||
for brpvlan in brpvlans["vlans"]:
|
||||
if "PVID" in brpvlan.get("flags", []):
|
||||
lower["pvid"] = brpvlan["vlan"]
|
||||
|
||||
if iplink.get("linkinfo", {}).get("info_kind") == "bridge":
|
||||
return lower
|
||||
|
||||
info = iplink["linkinfo"]["info_slave_data"]
|
||||
|
||||
return lower | {
|
||||
"bridge": iplink["master"],
|
||||
"flood": {
|
||||
"broadcast": info["bcast_flood"],
|
||||
"unicast": info["flood"],
|
||||
"multicast": info["mcast_flood"],
|
||||
},
|
||||
|
||||
"multicast": {
|
||||
"fast-leave": info["fastleave"],
|
||||
"router": {
|
||||
0: "off",
|
||||
1: "auto",
|
||||
2: "permanent",
|
||||
}.get(info["multicast_router"], "UNKNOWN"),
|
||||
},
|
||||
|
||||
"stp-state": info["state"],
|
||||
}
|
||||
|
||||
|
||||
def mctlq2yang_mode(mctlq):
|
||||
if state := mctlq.get("state"):
|
||||
return "proxy" if state == "proxy" else "auto"
|
||||
|
||||
return "off"
|
||||
|
||||
|
||||
def mctl(ifname, vid):
|
||||
mctl = HOST.run_json(["mctl", "-p", "show", "igmp", "json"], default={})
|
||||
|
||||
for q in mctl.get("multicast-queriers", []):
|
||||
# TODO: Also need to match against VLAN uppers (e.g. br0.1337)
|
||||
if q.get("interface") == ifname and q.get("vid") == vid:
|
||||
return q
|
||||
|
||||
return {}
|
||||
|
||||
|
||||
def multicast_filters(iplink, vid):
|
||||
filt = ["dev", iplink["ifname"]] + (["vid", str(vid)] if vid else [])
|
||||
brmdb = HOST.run_json(["bridge", "-j", "mdb", "show"] + filt)[0]["mdb"]
|
||||
|
||||
mdb = {}
|
||||
for brentry in brmdb:
|
||||
if not (entry := mdb.get(brentry["grp"])):
|
||||
mdb[brentry["grp"]] = {
|
||||
"group": brentry["grp"],
|
||||
"ports": [],
|
||||
}
|
||||
entry = mdb[brentry["grp"]]
|
||||
|
||||
entry["ports"].append({
|
||||
"port": brentry["port"],
|
||||
"state": {
|
||||
"temp": "temporary",
|
||||
"permanent": "permanent"
|
||||
}.get(brentry["state"], "UNKNOWN"),
|
||||
})
|
||||
|
||||
return { "multicast-filter": list(mdb.values()) }
|
||||
|
||||
|
||||
def multicast(iplink, info):
|
||||
mctlq = mctl(iplink["ifname"], info.get("vlan"))
|
||||
|
||||
mcast = {
|
||||
"snooping": bool(info.get("mcast_snooping")),
|
||||
"querier": mctlq2yang_mode(mctlq),
|
||||
}
|
||||
|
||||
if interval := mctlq.get("interval"):
|
||||
mcast["query-interval"] = interval
|
||||
|
||||
return mcast
|
||||
|
||||
|
||||
def vlans_add_memberships(iplink, vlans):
|
||||
brvlans = bridge_vlan()
|
||||
ports = [iplink["ifname"]] + [link["ifname"] for link in iplinks_lower_of(iplink["ifname"]).values()]
|
||||
|
||||
for port in ports:
|
||||
if not (brpvlans := brvlans.get(port)):
|
||||
continue
|
||||
|
||||
for brpvlan in brpvlans["vlans"]:
|
||||
if not (vlan := vlans.get(brpvlan["vlan"])):
|
||||
LOG.error(f"Unexpected vlan {brpvlans['vlan']} on {port}")
|
||||
continue
|
||||
|
||||
if "Egress Untagged" in brpvlan.get("flags", []):
|
||||
vlan["untagged"].append(port)
|
||||
else:
|
||||
vlan["tagged"].append(port)
|
||||
|
||||
|
||||
def vlans(iplink):
|
||||
if not (brgvlans := HOST.run_json(f"bridge -j vlan global show dev {iplink['ifname']}".split())):
|
||||
return []
|
||||
|
||||
vlans = {
|
||||
v["vlan"]: {
|
||||
"vid": v["vlan"],
|
||||
"untagged": [],
|
||||
"tagged": [],
|
||||
|
||||
"multicast": multicast(iplink, v),
|
||||
"multicast-filters": multicast_filters(iplink, v["vlan"]),
|
||||
}
|
||||
for v in brgvlans[0]["vlans"]
|
||||
}
|
||||
|
||||
vlans_add_memberships(iplink, vlans)
|
||||
return list(vlans.values())
|
||||
|
||||
|
||||
def qbridge(iplink):
|
||||
info = iplink["linkinfo"]["info_data"]
|
||||
|
||||
return {
|
||||
"vlans": {
|
||||
"proto": vlan.proto2yang(info["vlan_protocol"]),
|
||||
"vlan": vlans(iplink),
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
def dbridge(iplink):
|
||||
info = iplink["linkinfo"]["info_data"]
|
||||
|
||||
return {
|
||||
"multicast": multicast(iplink, info),
|
||||
"multicast-filters": multicast_filters(iplink, None),
|
||||
}
|
||||
|
||||
|
||||
def bridge(iplink):
|
||||
info = iplink["linkinfo"]["info_data"]
|
||||
|
||||
if info.get("vlan_filtering"):
|
||||
return qbridge(iplink)
|
||||
else:
|
||||
return dbridge(iplink)
|
||||
@@ -0,0 +1,30 @@
|
||||
from functools import cache
|
||||
|
||||
from ..host import HOST
|
||||
|
||||
|
||||
@cache
|
||||
def iplinks(ifname=None, netns=None):
|
||||
def _iplinks(ifname, netns):
|
||||
pre = [ "ip", "netns", "exec", netns ] if netns else []
|
||||
filt = ["dev", ifname] if ifname else []
|
||||
return HOST.run_json(pre + ["ip", "-s", "-d", "-j", "link", "show"] + filt)
|
||||
|
||||
return { link["ifname"]: link for link in _iplinks(ifname, netns) }
|
||||
|
||||
|
||||
def iplinks_lower_of(upper):
|
||||
return {
|
||||
link["ifname"]: link for link in
|
||||
filter(lambda link: link.get("master") == upper, iplinks().values())
|
||||
}
|
||||
|
||||
|
||||
@cache
|
||||
def ipaddrs(ifname=None, netns=None):
|
||||
def _ipaddrs(ifname, netns):
|
||||
pre = [ "ip", "netns", "exec", netns ] if netns else []
|
||||
filt = ["dev", ifname] if ifname else []
|
||||
return HOST.run_json(pre + ["ip", "-j", "addr", "show"] + filt)
|
||||
|
||||
return { addr["ifname"]: addr for addr in _ipaddrs(ifname, netns) }
|
||||
@@ -0,0 +1,57 @@
|
||||
from ..host import HOST
|
||||
from ..infix_containers import podman_ps
|
||||
|
||||
from . import common
|
||||
from . import link
|
||||
|
||||
|
||||
def ip_netns_list():
|
||||
return HOST.run_json(["ip", "-j", "netns", "list"], [])
|
||||
|
||||
|
||||
def find_interface(cifname):
|
||||
for ns in ip_netns_list():
|
||||
for iplink in common.iplinks(netns=ns["name"]).values():
|
||||
if iplink.get("ifalias") == cifname:
|
||||
ipaddrs = common.ipaddrs(ifname=iplink["ifname"], netns=ns["name"])
|
||||
return (iplink, next(iter(ipaddrs.values())))
|
||||
|
||||
|
||||
def podman_interfaces():
|
||||
interfaces = {}
|
||||
|
||||
for container in podman_ps():
|
||||
containername = container.get("Names", ["Unknown"])[0]
|
||||
|
||||
for ifname in container.get("Networks", []):
|
||||
if ifname not in interfaces:
|
||||
interfaces[ifname] = []
|
||||
if containername not in interfaces[ifname]:
|
||||
interfaces[ifname].append(containername)
|
||||
|
||||
return interfaces
|
||||
|
||||
def interfaces(ifname):
|
||||
interfaces = []
|
||||
|
||||
for cifname, cnames in podman_interfaces().items():
|
||||
if ifname and cifname != ifname:
|
||||
continue
|
||||
|
||||
iplink, ipaddr = find_interface(cifname)
|
||||
interface = link.interface_common(iplink, ipaddr)
|
||||
|
||||
# The original interface name is stored in ifalias by podman -
|
||||
# which we then translate to the description. We need to
|
||||
# reverse these since the "name" from the user's perspective
|
||||
# is the one set in running-config, not whatever the container
|
||||
# has renamed it to.
|
||||
interface["description"] = interface["name"]
|
||||
interface["name"] = cifname
|
||||
|
||||
interface["infix-interfaces:container-network"] = {
|
||||
"containers": cnames,
|
||||
}
|
||||
interfaces.append(interface)
|
||||
|
||||
return interfaces
|
||||
@@ -0,0 +1,110 @@
|
||||
from ..host import HOST
|
||||
|
||||
|
||||
def frame_statistics(etstats):
|
||||
STAT_MAP = {
|
||||
"eth-mac": {
|
||||
"out-frames": "FramesTransmittedOK",
|
||||
"out-multicast-frames": "MulticastFramesXmittedOK",
|
||||
"out-broadcast-frames": "BroadcastFramesXmittedOK",
|
||||
"in-frames": "FramesReceivedOK",
|
||||
"in-multicast-frames": "MulticastFramesReceivedOK",
|
||||
"in-broadcast-frames": "BroadcastFramesReceivedOK",
|
||||
"in-error-fcs-frames": "FrameCheckSequenceErrors",
|
||||
"in-error-mac-internal-frames": "FramesLostDueToIntMACRcvError",
|
||||
"infix-ethernet-interface:out-good-octets": "OctetsTransmittedOK",
|
||||
"infix-ethernet-interface:in-good-octets": "OctetsReceivedOK",
|
||||
|
||||
"in-total-frames": (
|
||||
"FramesReceivedOK",
|
||||
"FrameCheckSequenceErrors",
|
||||
"FramesLostDueToIntMACRcvError",
|
||||
"AlignmentErrors",
|
||||
"etherStatsOversizePkts",
|
||||
"etherStatsJabbers",
|
||||
),
|
||||
},
|
||||
|
||||
"rmon": {
|
||||
"in-error-undersize-frames": "undersize_pkts",
|
||||
|
||||
"in-error-oversize-frames": (
|
||||
"etherStatsJabbers",
|
||||
"etherStatsOversizePkts",
|
||||
),
|
||||
},
|
||||
}
|
||||
|
||||
fstats = {}
|
||||
|
||||
for group, mapping in STAT_MAP.items():
|
||||
etgroup = etstats.get(group)
|
||||
if not etgroup:
|
||||
continue
|
||||
|
||||
for name, source in mapping.items():
|
||||
if type(source) == str:
|
||||
counter = etgroup.get(source)
|
||||
if counter is not None:
|
||||
fstats[name] = str(counter)
|
||||
elif type(source) == tuple:
|
||||
inputs = [etgroup.get(src) for src in source]
|
||||
if inputs := filter(lambda i: i is not None, inputs):
|
||||
fstats[name] = str(sum(inputs))
|
||||
|
||||
return fstats
|
||||
|
||||
|
||||
def statistics(ifname):
|
||||
if etstats := HOST.run_json(["ethtool", "--json", "-S", ifname, "--all-groups"], []):
|
||||
etstats = etstats[0]
|
||||
else:
|
||||
return None
|
||||
|
||||
statistics = {}
|
||||
|
||||
if fstats := frame_statistics(etstats):
|
||||
statistics["frame"] = fstats
|
||||
|
||||
return statistics
|
||||
|
||||
|
||||
def link(ethtool):
|
||||
"""Parse speed/duplex/autoneg from ethtool output"""
|
||||
eth = {}
|
||||
|
||||
for line in ethtool:
|
||||
kv = [s.strip() for s in line.split(":")]
|
||||
if len(kv) != 2:
|
||||
continue
|
||||
|
||||
key, val = kv
|
||||
match key:
|
||||
case "Auto-negotiation":
|
||||
eth["auto-negotiation"] = { "enable": val == "on" }
|
||||
case "Duplex":
|
||||
match val:
|
||||
case "Half":
|
||||
eth["duplex"] = "half"
|
||||
case "Full":
|
||||
eth["duplex"] = "full"
|
||||
case _:
|
||||
eth["duplex"] = "unknown"
|
||||
case "Speed":
|
||||
mbps = "".join(filter(str.isdigit, val))
|
||||
if mbps:
|
||||
gbps = round((int(mbps) / 1000), 3)
|
||||
eth["speed"] = str(gbps)
|
||||
|
||||
return eth
|
||||
|
||||
|
||||
def ethernet(iplink):
|
||||
ethtool = HOST.run_multiline(["ethtool", iplink["ifname"]])
|
||||
|
||||
eth = link(ethtool)
|
||||
|
||||
if stats := statistics(iplink["ifname"]):
|
||||
eth["statistics"] = stats
|
||||
|
||||
return eth
|
||||
@@ -0,0 +1,56 @@
|
||||
from ..host import HOST
|
||||
|
||||
def inet2yang_origin(inet):
|
||||
"""Translate kernel IP address origin to YANG"""
|
||||
xlate = {
|
||||
"kernel_ll": "link-layer",
|
||||
"kernel_ra": "link-layer",
|
||||
"static": "static",
|
||||
"dhcp": "dhcp",
|
||||
"random": "random",
|
||||
}
|
||||
proto = inet.get("protocol")
|
||||
|
||||
if proto in ("kernel_ll", "kernel_ra"):
|
||||
if "stable-privacy" in inet:
|
||||
return "random"
|
||||
|
||||
return xlate.get(proto, "other")
|
||||
|
||||
|
||||
def addresses(ipaddr, proto):
|
||||
addrs = []
|
||||
for inet in ipaddr.get("addr_info", []):
|
||||
if inet.get("family") != proto:
|
||||
continue
|
||||
|
||||
addrs.append({
|
||||
"ip": inet.get("local"),
|
||||
"prefix-length": inet.get("prefixlen"),
|
||||
"origin": inet2yang_origin(inet),
|
||||
})
|
||||
|
||||
return addrs
|
||||
|
||||
def ipv4(ipaddr):
|
||||
ipv4 = {}
|
||||
|
||||
mtu = ipaddr.get("mtu")
|
||||
if mtu and ipaddr.get("ifname") != "lo":
|
||||
ipv4["mtu"] = mtu
|
||||
|
||||
if addrs := addresses(ipaddr, "inet"):
|
||||
ipv4["address"] = addrs
|
||||
|
||||
return ipv4
|
||||
|
||||
def ipv6(ipaddr):
|
||||
ipv6 = {}
|
||||
|
||||
if mtu := HOST.read(f"/proc/sys/net/ipv6/conf/{ipaddr['ifname']}/mtu"):
|
||||
ipv6["mtu"] = int(mtu.strip())
|
||||
|
||||
if addrs := addresses(ipaddr, "inet6"):
|
||||
ipv6["address"] = addrs
|
||||
|
||||
return ipv6
|
||||
@@ -0,0 +1,158 @@
|
||||
from . import common
|
||||
|
||||
from . import bridge
|
||||
from . import ethernet
|
||||
from . import ip
|
||||
from . import tun
|
||||
from . import veth
|
||||
from . import vlan
|
||||
|
||||
|
||||
def statistics(iplink):
|
||||
statistics = {}
|
||||
|
||||
if rx := iplink.get("stats64", {}).get("rx"):
|
||||
if octets := rx.get("bytes"):
|
||||
statistics["in-octets"] = str(octets)
|
||||
|
||||
if tx := iplink.get("stats64", {}).get("tx"):
|
||||
if octets := tx.get("bytes"):
|
||||
statistics["out-octets"] = str(octets)
|
||||
|
||||
return statistics
|
||||
|
||||
|
||||
def iplink2yang_type(iplink):
|
||||
match iplink["link_type"]:
|
||||
case "loopback":
|
||||
return "infix-if-type:loopback"
|
||||
case "gre"|"gre6":
|
||||
return "infix-if-type:gre"
|
||||
case "ether":
|
||||
pass
|
||||
case _:
|
||||
return "infix-if-type:other"
|
||||
|
||||
if iplink.get("parentbus") == "virtio":
|
||||
return "infix-if-type:etherlike"
|
||||
|
||||
match iplink.get("linkinfo", {}).get("info_kind"):
|
||||
case "bond":
|
||||
return "infix-if-type:lag"
|
||||
case "bridge":
|
||||
return "infix-if-type:bridge"
|
||||
case "dummy":
|
||||
return "infix-if-type:dummy"
|
||||
case "gretap"|"ip6gretap":
|
||||
return "infix-if-type:gretap"
|
||||
case "veth":
|
||||
return "infix-if-type:veth"
|
||||
case "vlan":
|
||||
return "infix-if-type:vlan"
|
||||
|
||||
return "infix-if-type:ethernet"
|
||||
|
||||
|
||||
def iplink2yang_lower(iplink):
|
||||
if not (kind := iplink.get("linkinfo",{}).get("info_slave_kind")):
|
||||
return None
|
||||
|
||||
match kind:
|
||||
case "bridge":
|
||||
return "infix-interfaces:bridge-port"
|
||||
case "bond":
|
||||
return "infix-interfaces:lag-port"
|
||||
|
||||
return None
|
||||
|
||||
|
||||
def iplink2yang_operstate(iplink):
|
||||
xlate = {
|
||||
"DOWN": "down",
|
||||
"UP": "up",
|
||||
"DORMANT": "dormant",
|
||||
"TESTING": "testing",
|
||||
"LOWERLAYERDOWN": "lower-layer-down",
|
||||
"NOTPRESENT": "not-present"
|
||||
}
|
||||
return xlate.get(iplink["operstate"], "unknown")
|
||||
|
||||
|
||||
def interface_common(iplink, ipaddr):
|
||||
interface = {
|
||||
"type": iplink2yang_type(iplink),
|
||||
"name": iplink.get("ifname"),
|
||||
"if-index": iplink.get("ifindex"),
|
||||
|
||||
"admin-status": "up" if "UP" in iplink.get("flags", "") else "down",
|
||||
"oper-status": iplink2yang_operstate(iplink),
|
||||
}
|
||||
|
||||
if "ifalias" in iplink:
|
||||
interface["description"] = iplink["ifalias"]
|
||||
|
||||
if "address" in iplink and not "POINTOPOINT" in iplink["flags"]:
|
||||
interface["phys-address"] = iplink["address"]
|
||||
|
||||
if stats := statistics(iplink):
|
||||
interface["statistics"] = stats
|
||||
|
||||
if ipv4 := ip.ipv4(ipaddr):
|
||||
interface["ietf-ip:ipv4"] = ipv4
|
||||
|
||||
if ipv6 := ip.ipv6(ipaddr):
|
||||
interface["ietf-ip:ipv6"] = ipv6
|
||||
|
||||
return interface
|
||||
|
||||
|
||||
def interface(iplink, ipaddr):
|
||||
interface = interface_common(iplink, ipaddr)
|
||||
|
||||
match interface["type"]:
|
||||
case "infix-if-type:bridge":
|
||||
if br := bridge.bridge(iplink):
|
||||
interface["infix-interfaces:bridge"] = br
|
||||
if brport := bridge.lower(iplink):
|
||||
interface["infix-interfaces:bridge-port"] = brport
|
||||
# case "infix-if-type:lag":
|
||||
# if l := lag.lag(iplink):
|
||||
# interface["infix-interfaces:lag"] = l
|
||||
case "infix-if-type:ethernet":
|
||||
if eth := ethernet.ethernet(iplink):
|
||||
interface["ieee802-ethernet-interface:ethernet"] = eth
|
||||
case "infix-if-type:gre"|"infix-if-type:gretap":
|
||||
if gre := tun.gre(iplink):
|
||||
interface["infix-interfaces:gre"] = gre
|
||||
case "infix-if-type:veth":
|
||||
if ve := veth.veth(iplink):
|
||||
interface["infix-interfaces:veth"] = ve
|
||||
case "infix-if-type:vlan":
|
||||
if v := vlan.vlan(iplink):
|
||||
interface["infix-interfaces:vlan"] = v
|
||||
|
||||
match iplink2yang_lower(iplink):
|
||||
case "infix-interfaces:bridge-port":
|
||||
if brport := bridge.lower(iplink):
|
||||
interface["infix-interfaces:bridge-port"] = brport
|
||||
# case "infix-interfaces:lag-port":
|
||||
# if lagport := lag.lower(iplink):
|
||||
# interface["infix-interfaces:lag-port"] = lagport
|
||||
|
||||
return interface
|
||||
|
||||
|
||||
def interfaces(ifname=None):
|
||||
links = common.iplinks(ifname)
|
||||
addrs = common.ipaddrs(ifname)
|
||||
|
||||
interfaces = []
|
||||
for ifname, iplink in links.items():
|
||||
if iplink.get("group") == "internal":
|
||||
continue
|
||||
|
||||
ipaddr = addrs.get(ifname, {})
|
||||
|
||||
interfaces.append(interface(iplink, ipaddr))
|
||||
|
||||
return interfaces
|
||||
@@ -0,0 +1,7 @@
|
||||
def gre(iplink):
|
||||
info=iplink.get("linkinfo", {}).get("info_data", {})
|
||||
|
||||
return {
|
||||
"local": info["local"],
|
||||
"remote": info["remote"],
|
||||
}
|
||||
@@ -0,0 +1,7 @@
|
||||
def veth(iplink):
|
||||
veth = {}
|
||||
|
||||
if peer := iplink.get("link"):
|
||||
veth["peer"] = peer
|
||||
|
||||
return veth
|
||||
@@ -0,0 +1,21 @@
|
||||
def proto2yang(proto):
|
||||
return {
|
||||
"802.1Q": "ieee802-dot1q-types:c-vlan",
|
||||
"802.1ad": "ieee802-dot1q-types:s-vlan",
|
||||
}.get(proto, "other")
|
||||
|
||||
|
||||
def vlan(iplink):
|
||||
info = iplink["linkinfo"]["info_data"]
|
||||
|
||||
vlan = {
|
||||
"tag-type": proto2yang(info["protocol"]),
|
||||
"id": info["id"],
|
||||
}
|
||||
|
||||
# Lower could be in a different namespace, and thus might not be
|
||||
# available to us
|
||||
if lower := iplink.get("link"):
|
||||
vlan["lower-layer-if"] = lower
|
||||
|
||||
return vlan
|
||||
Reference in New Issue
Block a user