No OneTemporary
Actions

Size

65 KB

Referenced Files

None

Subscribers

None

View Options

	diff --git a/python/vyos/utils/dict.py b/python/vyos/utils/dict.py
	index ff21f0677..a43430eed 100644
	--- a/python/vyos/utils/dict.py
	+++ b/python/vyos/utils/dict.py
	@@ -1,373 +1,405 @@
	# Copyright VyOS maintainers and contributors <maintainers@vyos.io>
	#
	# This library is free software; you can redistribute it and/or
	# modify it under the terms of the GNU Lesser General Public
	# License as published by the Free Software Foundation; either
	# version 2.1 of the License, or (at your option) any later version.
	#
	# This library is distributed in the hope that it will be useful,
	# but WITHOUT ANY WARRANTY; without even the implied warranty of
	# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
	# Lesser General Public License for more details.
	#
	# You should have received a copy of the GNU Lesser General Public
	# License along with this library. If not, see <http://www.gnu.org/licenses/>.

	def colon_separated_to_dict(data_string, uniquekeys=False):
	""" Converts a string containing newline-separated entries
	of colon-separated key-value pairs into a dict.

	Such files are common in Linux /proc filesystem

	Args:
	data_string (str): data string
	uniquekeys (bool): whether to insist that keys are unique or not

	Returns: dict

	Raises:
	ValueError: if uniquekeys=True and the data string has
	duplicate keys.

	Note:
	If uniquekeys=True, then dict entries are always strings,
	otherwise they are always lists of strings.
	"""
	import re
	key_value_re = re.compile(r'([^:]+)\s\:\s(.*)')

	data_raw = re.split('\n', data_string)

	data = {}

	for l in data_raw:
	l = l.strip()
	if l:
	match = re.match(key_value_re, l)
	if match and (len(match.groups()) == 2):
	key = match.groups()[0].strip()
	value = match.groups()[1].strip()
	else:
	raise ValueError(f"""Line "{l}" could not be parsed a colon-separated pair """, l)
	if key in data.keys():
	if uniquekeys:
	raise ValueError("Data string has duplicate keys: {0}".format(key))
	else:
	data[key].append(value)
	else:
	if uniquekeys:
	data[key] = value
	else:
	data[key] = [value]
	else:
	pass

	return data

	def mangle_dict_keys(data, regex, replacement, abs_path=None, no_tag_node_value_mangle=False):
	""" Mangles dict keys according to a regex and replacement character.
	Some libraries like Jinja2 do not like certain characters in dict keys.
	This function can be used for replacing all offending characters
	with something acceptable.

	Args:
	data (dict): Original dict to mangle
	regex, replacement (str): arguments to re.sub(regex, replacement, ...)
	abs_path (list): if data is a config dict and no_tag_node_value_mangle is True
	then abs_path should be the absolute config path to the first
	keys of data, non-inclusive
	no_tag_node_value_mangle (bool): do not mangle keys of tag node values

	Returns: dict
	"""
	import re
	from vyos.xml_ref import is_tag_value

	if abs_path is None:
	abs_path = []

	new_dict = type(data)()

	for k in data.keys():
	if no_tag_node_value_mangle and is_tag_value(abs_path + [k]):
	new_key = k
	else:
	new_key = re.sub(regex, replacement, k)

	value = data[k]

	if isinstance(value, dict):
	new_dict[new_key] = mangle_dict_keys(value, regex, replacement,
	abs_path=abs_path + [k],
	no_tag_node_value_mangle=no_tag_node_value_mangle)
	else:
	new_dict[new_key] = value

	return new_dict

	def _get_sub_dict(d, lpath):
	k = lpath[0]
	if k not in d.keys():
	return {}
	c = {k: d[k]}
	lpath = lpath[1:]
	if not lpath:
	return c
	elif not isinstance(c[k], dict):
	return {}
	return _get_sub_dict(c[k], lpath)

	def get_sub_dict(source, lpath, get_first_key=False):
	""" Returns the sub-dict of a nested dict, defined by path of keys.

	Args:
	source (dict): Source dict to extract from
	lpath (list[str]): sequence of keys

	Returns: source, if lpath is empty, else
	{key : source[..]..[key]} for key the last element of lpath, if exists
	{} otherwise
	"""
	if not isinstance(source, dict):
	raise TypeError("source must be of type dict")
	if not isinstance(lpath, list):
	raise TypeError("path must be of type list")
	if not lpath:
	return source

	ret = _get_sub_dict(source, lpath)

	if get_first_key and lpath and ret:
	tmp = next(iter(ret.values()))
	if not isinstance(tmp, dict):
	raise TypeError("Data under node is not of type dict")
	ret = tmp

	return ret

	def dict_search(path, dict_object):
	""" Traverse Python dictionary (dict_object) delimited by dot (.).
	Return value of key if found, None otherwise.

	This is faster implementation then jmespath.search('foo.bar', dict_object)"""
	if not isinstance(dict_object, dict) or not path:
	return None

	parts = path.split('.')
	inside = parts[:-1]
	if not inside:
	if path not in dict_object:
	return None
	return dict_object[path]
	c = dict_object
	for p in parts[:-1]:
	c = c.get(p, {})
	return c.get(parts[-1], None)

	def dict_search_args(dict_object, *path):
	# Traverse dictionary using variable arguments
	# Added due to above function not allowing for '.' in the key names
	# Example: dict_search_args(some_dict, 'key', 'subkey', 'subsubkey', ...)
	if not isinstance(dict_object, dict) or not path:
	return None

	for item in path:
	if item not in dict_object:
	return None
	dict_object = dict_object[item]
	return dict_object

	def dict_search_recursive(dict_object, key, path=[]):
	""" Traverse a dictionary recurisvely and return the value of the key
	we are looking for.

	Thankfully copied from https://stackoverflow.com/a/19871956

	Modified to yield optional path to found keys
	"""
	if isinstance(dict_object, list):
	for i in dict_object:
	new_path = path + [i]
	for x in dict_search_recursive(i, key, new_path):
	yield x
	elif isinstance(dict_object, dict):
	if key in dict_object:
	new_path = path + [key]
	yield dict_object[key], new_path
	for k, j in dict_object.items():
	new_path = path + [k]
	for x in dict_search_recursive(j, key, new_path):
	yield x


	def dict_set(key_path, value, dict_object):
	""" Set value to Python dictionary (dict_object) using path to key delimited by dot (.).
	The key will be added if it does not exist.
	"""
	path_list = key_path.split(".")
	dynamic_dict = dict_object
	if len(path_list) > 0:
	for i in range(0, len(path_list)-1):
	dynamic_dict = dynamic_dict[path_list[i]]
	dynamic_dict[path_list[len(path_list)-1]] = value

	+def dict_set_nested(key_path, value, dict_object):
	+ """
	+ Set value to Python dictionary (dict_object) using a path to the key
	+ delimited by dot ('.'). The key will be added if it does not exist.
	+ Missing keys along the path will be created as nested dictionaries.
	+
	+ Parameters
	+ ----------
	+ key_path : str
	+ Dot-delimited path to the key (e.g. "this.is.a.path").
	+ value : any
	+ The value to set at the final key in the path.
	+ dict_object : dict
	+ Dictionary to modify. Will be updated in place.
	+
	+ Examples
	+ --------
	+ d = {}
	+ dict_set_nested("this.is.a.path", 42, d)
	+ # {'this': {'is': {'a': {'path': 42}}}}
	+
	+ d = {"existing": {"branch": {}}}
	+ dict_set_nested("existing.branch.leaf", "value", d)
	+ # {'existing': {'branch': {'leaf': 'value'}}}
	+ """
	+ path_list = key_path.split(".")
	+ dynamic_dict = dict_object
	+ for i in range(0, len(path_list) - 1):
	+ if path_list[i] not in dynamic_dict or not isinstance(dynamic_dict[path_list[i]], dict):
	+ dynamic_dict[path_list[i]] = {}
	+ dynamic_dict = dynamic_dict[path_list[i]]
	+ dynamic_dict[path_list[-1]] = value
	+
	def dict_delete(key_path, dict_object):
	""" Delete key in Python dictionary (dict_object) using path to key delimited by dot (.).
	"""
	path_dict = dict_object
	path_list = key_path.split('.')
	inside = path_list[:-1]
	if not inside:
	del dict_object[path_list]
	else:
	for key in path_list[:-1]:
	path_dict = path_dict[key]
	del path_dict[path_list[len(path_list)-1]]

	def dict_to_list(d, save_key_to=None):
	""" Convert a dict to a list of dicts.

	Optionally, save the original key of the dict inside
	dicts stores in that list.
	"""
	def save_key(i, k):
	if isinstance(i, dict):
	i[save_key_to] = k
	return
	elif isinstance(i, list):
	for _i in i:
	save_key(_i, k)
	else:
	raise ValueError(f"Cannot save the key: the item is {type(i)}, not a dict")

	collect = []

	for k,_ in d.items():
	item = d[k]
	if save_key_to is not None:
	save_key(item, k)
	if isinstance(item, list):
	collect += item
	else:
	collect.append(item)

	return collect

	def dict_to_paths_values(conf: dict) -> dict:
	"""
	Convert nested dictionary to simple dictionary, where key is a path is delimited by dot (.).
	"""
	list_of_paths = []
	dict_of_options ={}
	for path in dict_to_key_paths(conf):
	str_path = '.'.join(path)
	list_of_paths.append(str_path)

	for path in list_of_paths:
	dict_of_options[path] = dict_search(path,conf)

	return dict_of_options

	def dict_to_key_paths(d: dict) -> list:
	""" Generator to return list of key paths from dict of list[str]\|str
	"""
	def func(d, path):
	if isinstance(d, dict):
	if not d:
	yield path
	for k, v in d.items():
	for r in func(v, path + [k]):
	yield r
	elif isinstance(d, list):
	yield path
	elif isinstance(d, str):
	yield path
	else:
	raise ValueError('object is not a dict of strings/list of strings')
	for r in func(d, []):
	yield r

	def dict_to_paths(d: dict) -> list:
	""" Generator to return list of paths from dict of list[str]\|str
	"""
	def func(d, path):
	if isinstance(d, dict):
	if not d:
	yield path
	for k, v in d.items():
	for r in func(v, path + [k]):
	yield r
	elif isinstance(d, list):
	for i in d:
	for r in func(i, path):
	yield r
	elif isinstance(d, str):
	yield path + [d]
	else:
	raise ValueError('object is not a dict of strings/list of strings')
	for r in func(d, []):
	yield r

	def embed_dict(p: list[str], d: dict) -> dict:
	path = p.copy()
	ret = d
	while path:
	ret = {path.pop(): ret}
	return ret

	def check_mutually_exclusive_options(d, keys, required=False):
	""" Checks if a dict has at most one or only one of
	mutually exclusive keys.
	"""
	present_keys = []

	for k in d:
	if k in keys:
	present_keys.append(k)

	# Un-mangle the keys to make them match CLI option syntax
	from re import sub
	orig_keys = list(map(lambda s: sub(r'_', '-', s), keys))
	orig_present_keys = list(map(lambda s: sub(r'_', '-', s), present_keys))

	if len(present_keys) > 1:
	raise ValueError(f"Options {orig_keys} are mutually-exclusive but more than one of them is present: {orig_present_keys}")

	if required and (len(present_keys) < 1):
	raise ValueError(f"At least one of the following options is required: {orig_keys}")

	class FixedDict(dict):
	"""
	FixedDict: A dictionnary not allowing new keys to be created after initialisation.

	>>> f = FixedDict(**{'count':1})
	>>> f['count'] = 2
	>>> f['king'] = 3
	File "...", line ..., in __setitem__
	raise ConfigError(f'Option "{k}" has no defined default')
	"""

	def __init__(self, **options):
	self._allowed = options.keys()
	super().__init__(**options)

	def __setitem__(self, k, v):
	"""
	__setitem__ is a builtin which is called by python when setting dict values:
	>>> d = dict()
	>>> d['key'] = 'value'
	>>> d
	{'key': 'value'}

	is syntaxic sugar for

	>>> d = dict()
	>>> d.__setitem__('key','value')
	>>> d
	{'key': 'value'}
	"""
	from vyos import ConfigError
	if k not in self._allowed:
	raise ConfigError(f'Option "{k}" has no defined default')
	super().__setitem__(k, v)
	-
	diff --git a/python/vyos/utils/network.py b/python/vyos/utils/network.py
	index 0e2cc58cf..93f780d8b 100644
	--- a/python/vyos/utils/network.py
	+++ b/python/vyos/utils/network.py
	@@ -1,665 +1,668 @@
	# Copyright VyOS maintainers and contributors <maintainers@vyos.io>
	#
	# This library is free software; you can redistribute it and/or
	# modify it under the terms of the GNU Lesser General Public
	# License as published by the Free Software Foundation; either
	# version 2.1 of the License, or (at your option) any later version.
	#
	# This library is distributed in the hope that it will be useful,
	# but WITHOUT ANY WARRANTY; without even the implied warranty of
	# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
	# Lesser General Public License for more details.
	#
	# You should have received a copy of the GNU Lesser General Public
	# License along with this library. If not, see <http://www.gnu.org/licenses/>.

	from socket import AF_INET
	from socket import AF_INET6

	def _are_same_ip(one, two):
	from socket import inet_pton
	from vyos.template import is_ipv4
	# compare the binary representation of the IP
	f_one = AF_INET if is_ipv4(one) else AF_INET6
	s_two = AF_INET if is_ipv4(two) else AF_INET6
	return inet_pton(f_one, one) == inet_pton(f_one, two)

	def get_protocol_by_name(protocol_name):
	"""Get protocol number by protocol name

	% get_protocol_by_name('tcp')
	% 6
	"""
	import socket
	try:
	protocol_number = socket.getprotobyname(protocol_name)
	return protocol_number
	except socket.error:
	return protocol_name

	def interface_exists(interface) -> bool:
	import os
	return os.path.exists(f'/sys/class/net/{interface}')

	def is_netns_interface(interface, netns):
	from vyos.utils.process import rc_cmd
	rc, out = rc_cmd(f'sudo ip netns exec {netns} ip link show dev {interface}')
	if rc == 0:
	return True
	return False

	def get_netns_all() -> list:
	from json import loads
	from vyos.utils.process import cmd
	tmp = loads(cmd('ip --json netns ls'))
	return [ netns['name'] for netns in tmp ]

	def get_vrf_members(vrf: str) -> list:
	"""
	Get list of interface VRF members
	:param vrf: str
	:return: list
	"""
	import json
	from vyos.utils.process import cmd
	interfaces = []
	try:
	if not interface_exists(vrf):
	raise ValueError(f'VRF "{vrf}" does not exist!')
	output = cmd(f'ip --json --brief link show vrf {vrf}')
	answer = json.loads(output)
	for data in answer:
	if 'ifname' in data:
	# Skip PIM interfaces which appears in VRF
	if 'pim' not in data.get('ifname'):
	interfaces.append(data.get('ifname'))
	except:
	pass
	return interfaces

	def get_interface_vrf(interface):
	""" Returns VRF of given interface """
	from vyos.utils.dict import dict_search
	from vyos.utils.network import get_interface_config
	- tmp = get_interface_config(interface)
	+ if isinstance(interface, str):
	+ tmp = get_interface_config(interface)
	+ elif isinstance(interface, dict):
	+ tmp = interface
	if dict_search('linkinfo.info_slave_kind', tmp) == 'vrf':
	return tmp['master']
	return 'default'

	def get_vrf_tableid(interface: str):
	""" Return VRF table ID for given interface name or None """
	from vyos.utils.dict import dict_search
	table = None
	tmp = get_interface_config(interface)
	# Check if we are "the" VRF interface
	if dict_search('linkinfo.info_kind', tmp) == 'vrf':
	table = tmp['linkinfo']['info_data']['table']
	# or an interface bound to a VRF
	elif dict_search('linkinfo.info_slave_kind', tmp) == 'vrf':
	table = tmp['linkinfo']['info_slave_data']['table']
	return table

	def get_interface_config(interface):
	""" Returns the used encapsulation protocol for given interface.
	If interface does not exist, None is returned.
	"""
	if not interface_exists(interface):
	return None
	from json import loads
	from vyos.utils.process import cmd
	tmp = loads(cmd(f'ip --detail --json link show dev {interface}'))[0]
	return tmp

	def get_interface_address(interface):
	""" Returns the used encapsulation protocol for given interface.
	If interface does not exist, None is returned.
	"""
	if not interface_exists(interface):
	return None
	from json import loads
	from vyos.utils.process import cmd
	tmp = loads(cmd(f'ip --detail --json addr show dev {interface}'))[0]
	return tmp

	def get_interface_namespace(interface: str):
	"""
	Returns wich netns the interface belongs to
	"""
	from json import loads
	from vyos.utils.process import cmd

	# Bail out early if netns does not exist
	tmp = cmd(f'ip --json netns ls')
	if not tmp: return None

	for ns in loads(tmp):
	netns = f'{ns["name"]}'
	# Search interface in each netns
	data = loads(cmd(f'ip netns exec {netns} ip --json link show'))
	for tmp in data:
	if interface == tmp["ifname"]:
	return netns

	def is_ipv6_tentative(iface: str, ipv6_address: str) -> bool:
	"""Check if IPv6 address is in tentative state.

	This function checks if an IPv6 address on a specific network interface is
	in the tentative state. IPv6 tentative addresses are not fully configured
	and are undergoing Duplicate Address Detection (DAD) to ensure they are
	unique on the network.

	Args:
	iface (str): The name of the network interface.
	ipv6_address (str): The IPv6 address to check.

	Returns:
	bool: True if the IPv6 address is tentative, False otherwise.
	"""
	import json
	from vyos.utils.process import rc_cmd

	rc, out = rc_cmd(f'ip -6 --json address show dev {iface}')
	if rc:
	return False

	data = json.loads(out)
	for addr_info in data[0]['addr_info']:
	if (
	addr_info.get('local') == ipv6_address and
	addr_info.get('tentative', False)
	):
	return True
	return False

	def is_wwan_connected(interface):
	""" Determine if a given WWAN interface, e.g. wwan0 is connected to the
	carrier network or not """
	import json
	from vyos.utils.dict import dict_search
	from vyos.utils.process import cmd
	from vyos.utils.process import is_systemd_service_active

	if not interface.startswith('wwan'):
	raise ValueError(f'Specified interface "{interface}" is not a WWAN interface')

	# ModemManager is required for connection(s) - if service is not running,
	# there won't be any connection at all!
	if not is_systemd_service_active('ModemManager.service'):
	return False

	modem = interface.lstrip('wwan')

	tmp = cmd(f'mmcli --modem {modem} --output-json')
	tmp = json.loads(tmp)

	# return True/False if interface is in connected state
	return dict_search('modem.generic.state', tmp) == 'connected'

	def get_bridge_fdb(interface):
	""" Returns the forwarding database entries for a given interface """
	if not interface_exists(interface):
	return None
	from json import loads
	from vyos.utils.process import cmd
	tmp = loads(cmd(f'bridge -j fdb show dev {interface}'))
	return tmp

	def get_all_vrfs():
	""" Return a dictionary of all system wide known VRF instances """
	from json import loads
	from vyos.utils.process import cmd
	tmp = loads(cmd('ip --json vrf list'))
	# Result is of type [{"name":"red","table":1000},{"name":"blue","table":2000}]
	# so we will re-arrange it to a more nicer representation:
	# {'red': {'table': 1000}, 'blue': {'table': 2000}}
	data = {}
	for entry in tmp:
	name = entry.pop('name')
	data[name] = entry
	return data

	def interface_list() -> list:
	from vyos.ifconfig import Section
	"""
	Get list of interfaces in system
	:rtype: list
	"""
	return Section.interfaces()


	def vrf_list() -> list:
	"""
	Get list of VRFs in system
	:rtype: list
	"""
	return list(get_all_vrfs().keys())

	def mac2eui64(mac, prefix=None):
	"""
	Convert a MAC address to a EUI64 address or, with prefix provided, a full
	IPv6 address.
	Thankfully copied from https://gist.github.com/wido/f5e32576bb57b5cc6f934e177a37a0d3
	"""
	import re
	from ipaddress import ip_network
	# http://tools.ietf.org/html/rfc4291#section-2.5.1
	eui64 = re.sub(r'[.:-]', '', mac).lower()
	eui64 = eui64[0:6] + 'fffe' + eui64[6:]
	eui64 = hex(int(eui64[0:2], 16) ^ 2)[2:].zfill(2) + eui64[2:]

	if prefix is None:
	return ':'.join(re.findall(r'.{4}', eui64))
	else:
	try:
	net = ip_network(prefix, strict=False)
	euil = int('0x{0}'.format(eui64), 16)
	return str(net[euil])
	except: # pylint: disable=bare-except
	return

	def check_port_availability(address: str=None, port: int=0, protocol: str='tcp') -> bool:
	"""
	Check if given port is available and not used by any service.

	Should be used carefully for services that can start listening
	dynamically, because IP address may be dynamic too

	Args:
	address: IPv4 or IPv6 address - if None, checks on all interfaces
	port: TCP/UDP port number.


	Returns:
	False if a port is busy or IP address does not exists
	True if a port is free and IP address exists
	"""
	import socket
	from ipaddress import ip_address

	# treat None as "any address"
	address = address or '::'

	# verify arguments
	try:
	address = ip_address(address).compressed
	except ValueError:
	raise ValueError(f'{address} is not a valid IPv4 or IPv6 address')
	if port not in range(1, 65536):
	raise ValueError(f'Port {port} is not in range 1-65535')
	if protocol not in ['tcp', 'udp']:
	raise ValueError(f'{protocol} is not supported - only tcp and udp are allowed')

	protocol = socket.SOCK_STREAM if protocol == 'tcp' else socket.SOCK_DGRAM
	try:
	addr_info = socket.getaddrinfo(address, port, socket.AF_UNSPEC, protocol)
	except socket.gaierror as e:
	print(f'Invalid address: {address}')
	return False

	for family, socktype, proto, canonname, sockaddr in addr_info:
	try:
	with socket.socket(family, socktype, proto) as s:
	s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
	s.bind(sockaddr)
	# port is free to use
	return True
	except OSError:
	# port is already in use
	return False

	# if we reach this point, no socket was tested and we assume the port is
	# already in use - better safe then sorry
	return False


	def is_listen_port_bind_service(port: int, service: str) -> bool:
	"""Check if listen port bound to expected program name
	:param port: Bind port
	:param service: Program name
	:return: bool

	Example:
	% is_listen_port_bind_service(443, 'nginx')
	True
	% is_listen_port_bind_service(443, 'ocserv-main')
	False
	"""
	from psutil import net_connections as connections
	from psutil import Process as process
	for connection in connections():
	addr = connection.laddr
	pid = connection.pid
	pid_name = process(pid).name()
	pid_port = addr.port
	if service == pid_name and port == pid_port:
	return True
	return False

	def is_ipv6_link_local(addr):
	""" Check if addrsss is an IPv6 link-local address. Returns True/False """
	from ipaddress import ip_interface
	from vyos.template import is_ipv6
	addr = addr.split('%')[0]
	if is_ipv6(addr):
	if ip_interface(addr).is_link_local:
	return True

	return False

	def is_addr_assigned(ip_address, vrf=None, return_ifname=False, include_vrf=False) -> bool \| str:
	""" Verify if the given IPv4/IPv6 address is assigned to any interface """
	from netifaces import interfaces # pylint: disable = no-name-in-module
	from vyos.utils.network import get_interface_config
	from vyos.utils.dict import dict_search

	for interface in interfaces():
	# Check if interface belongs to the requested VRF, if this is not the
	# case there is no need to proceed with this data set - continue loop
	# with next element
	tmp = get_interface_config(interface)
	if dict_search('master', tmp) != vrf and not include_vrf:
	continue

	if is_intf_addr_assigned(interface, ip_address):
	return interface if return_ifname else True

	return False

	def is_intf_addr_assigned(ifname: str, addr: str, netns: str=None) -> bool:
	"""
	Verify if the given IPv4/IPv6 address is assigned to specific interface.
	It can check both a single IP address (e.g. 192.0.2.1 or a assigned CIDR
	address 192.0.2.1/24.
	"""
	import json
	import jmespath

	from vyos.utils.process import rc_cmd
	from ipaddress import ip_interface

	netns_cmd = f'ip netns exec {netns}' if netns else ''
	rc, out = rc_cmd(f'{netns_cmd} ip --json address show dev {ifname}')
	if rc == 0:
	json_out = json.loads(out)
	addresses = jmespath.search("[].addr_info[].{family: family, address: local, prefixlen: prefixlen}", json_out)
	for address_info in addresses:
	family = address_info['family']
	address = address_info['address']
	prefixlen = address_info['prefixlen']
	# Remove the interface name if present in the given address
	if '%' in addr:
	addr = addr.split('%')[0]
	interface = ip_interface(f"{address}/{prefixlen}")
	if ip_interface(addr) == interface or address == addr:
	return True

	return False

	def is_loopback_addr(addr):
	""" Check if supplied IPv4/IPv6 address is a loopback address """
	from ipaddress import ip_address
	return ip_address(addr).is_loopback

	def is_wireguard_key_pair(private_key: str, public_key:str) -> bool:
	"""
	Checks if public/private keys are keypair
	:param private_key: Wireguard private key
	:type private_key: str
	:param public_key: Wireguard public key
	:type public_key: str
	:return: If public/private keys are keypair returns True else False
	:rtype: bool
	"""
	from vyos.utils.process import cmd
	gen_public_key = cmd('wg pubkey', input=private_key)
	if gen_public_key == public_key:
	return True
	else:
	return False

	def get_wireguard_peers(ifname: str) -> list:
	"""
	Return list of configured Wireguard peers for interface
	:param ifname: Interface name
	:type ifname: str
	:return: list of public keys
	:rtype: list
	"""
	if not interface_exists(ifname):
	return []

	from vyos.utils.process import cmd
	peers = cmd(f'wg show {ifname} peers')
	return peers.splitlines()

	def is_subnet_connected(subnet, primary=False):
	"""
	Verify is the given IPv4/IPv6 subnet is connected to any interface on this
	system.

	primary check if the subnet is reachable via the primary IP address of this
	interface, or in other words has a broadcast address configured. ISC DHCP
	for instance will complain if it should listen on non broadcast interfaces.

	Return True/False
	"""
	from ipaddress import ip_address
	from ipaddress import ip_network

	from netifaces import ifaddresses # pylint: disable = no-name-in-module
	from netifaces import interfaces # pylint: disable = no-name-in-module

	from vyos.template import is_ipv6

	# determine IP version (AF_INET or AF_INET6) depending on passed address
	addr_type = AF_INET
	if is_ipv6(subnet):
	addr_type = AF_INET6

	for interface in interfaces():
	# check if the requested address type is configured at all
	if addr_type not in ifaddresses(interface).keys():
	continue

	# An interface can have multiple addresses, but some software components
	# only support the primary address :(
	if primary:
	ip = ifaddresses(interface)[addr_type][0]['addr']
	if ip_address(ip) in ip_network(subnet):
	return True
	else:
	# Check every assigned IP address if it is connected to the subnet
	# in question
	for ip in ifaddresses(interface)[addr_type]:
	# remove interface extension (e.g. %eth0) that gets thrown on the end of _some_ addrs
	addr = ip['addr'].split('%')[0]
	if ip_address(addr) in ip_network(subnet):
	return True

	return False

	def is_afi_configured(interface: str, afi):
	""" Check if given address family is configured, or in other words - an IP
	address is assigned to the interface. """
	from netifaces import ifaddresses # pylint: disable = no-name-in-module

	if afi not in [AF_INET, AF_INET6]:
	raise ValueError('Address family must be in [AF_INET, AF_INET6]')

	try:
	addresses = ifaddresses(interface)
	except ValueError as e:
	print(e)
	return False

	return afi in addresses

	def get_vxlan_vlan_tunnels(interface: str) -> list:
	""" Return a list of strings with VLAN IDs configured in the Kernel """
	from json import loads
	from vyos.utils.process import cmd

	if not interface.startswith('vxlan'):
	raise ValueError('Only applicable for VXLAN interfaces!')

	# Determine current OS Kernel configured VLANs
	#
	# $ bridge -j -p vlan tunnelshow dev vxlan0
	# [ {
	# "ifname": "vxlan0",
	# "tunnels": [ {
	# "vlan": 10,
	# "vlanEnd": 11,
	# "tunid": 10010,
	# "tunidEnd": 10011
	# },{
	# "vlan": 20,
	# "tunid": 10020
	# } ]
	# } ]
	#
	os_configured_vlan_ids = []
	tmp = loads(cmd(f'bridge --json vlan tunnelshow dev {interface}'))
	if tmp:
	for tunnel in tmp[0].get('tunnels', {}):
	vlanStart = tunnel['vlan']
	if 'vlanEnd' in tunnel:
	vlanEnd = tunnel['vlanEnd']
	# Build a real list for user VLAN IDs
	vlan_list = list(range(vlanStart, vlanEnd +1))
	# Convert list of integers to list or strings
	os_configured_vlan_ids.extend(map(str, vlan_list))
	# Proceed with next tunnel - this one is complete
	continue

	# Add single tunel id - not part of a range
	os_configured_vlan_ids.append(str(vlanStart))

	return os_configured_vlan_ids

	def get_vxlan_vni_filter(interface: str) -> list:
	""" Return a list of strings with VNIs configured in the Kernel"""
	from json import loads
	from vyos.utils.process import cmd

	if not interface.startswith('vxlan'):
	raise ValueError('Only applicable for VXLAN interfaces!')

	# Determine current OS Kernel configured VNI filters in VXLAN interface
	#
	# $ bridge -j vni show dev vxlan1
	# [{"ifname":"vxlan1","vnis":[{"vni":100},{"vni":200},{"vni":300,"vniEnd":399}]}]
	#
	# Example output: ['10010', '10020', '10021', '10022']
	os_configured_vnis = []
	tmp = loads(cmd(f'bridge --json vni show dev {interface}'))
	if tmp:
	for tunnel in tmp[0].get('vnis', {}):
	vniStart = tunnel['vni']
	if 'vniEnd' in tunnel:
	vniEnd = tunnel['vniEnd']
	# Build a real list for user VNIs
	vni_list = list(range(vniStart, vniEnd +1))
	# Convert list of integers to list or strings
	os_configured_vnis.extend(map(str, vni_list))
	# Proceed with next tunnel - this one is complete
	continue

	# Add single tunel id - not part of a range
	os_configured_vnis.append(str(vniStart))

	return os_configured_vnis

	# Calculate prefix length of an IPv6 range, where possible
	# Python-ified from source: https://gitlab.isc.org/isc-projects/dhcp/-/blob/master/keama/confparse.c#L4591
	def ipv6_prefix_length(low, high):
	import socket

	bytemasks = [0x80, 0xc0, 0xe0, 0xf0, 0xf8, 0xfc, 0xfe, 0xff]

	try:
	lo = bytearray(socket.inet_pton(socket.AF_INET6, low))
	hi = bytearray(socket.inet_pton(socket.AF_INET6, high))
	except:
	return None

	xor = bytearray(a ^ b for a, b in zip(lo, hi))

	plen = 0
	while plen < 128 and xor[plen // 8] == 0:
	plen += 8

	if plen == 128:
	return plen

	for i in range((plen // 8) + 1, 16):
	if xor[i] != 0:
	return None

	for i in range(8):
	msk = ~xor[plen // 8] & 0xff

	if msk == bytemasks[i]:
	return plen + i + 1

	return None

	def get_nft_vrf_zone_mapping() -> dict:
	"""
	Retrieve current nftables conntrack mapping list from Kernel

	returns: [{'interface': 'red', 'vrf_tableid': 1000},
	{'interface': 'eth2', 'vrf_tableid': 1000},
	{'interface': 'blue', 'vrf_tableid': 2000}]
	"""
	from json import loads
	from jmespath import search
	from vyos.utils.process import cmd
	output = []
	tmp = loads(cmd('sudo nft -j list table inet vrf_zones'))
	# {'nftables': [{'metainfo': {'json_schema_version': 1,
	# 'release_name': 'Old Doc Yak #3',
	# 'version': '1.0.9'}},
	# {'table': {'family': 'inet', 'handle': 6, 'name': 'vrf_zones'}},
	# {'map': {'elem': [['eth0', 666],
	# ['dum0', 666],
	# ['wg500', 666],
	# ['bond10.666', 666]],
	vrf_list = search('nftables[].map.elem \| [0]', tmp)
	if not vrf_list:
	return output
	for (vrf_name, vrf_id) in vrf_list:
	output.append({'interface' : vrf_name, 'vrf_tableid' : vrf_id})
	return output

	def is_valid_ipv4_address_or_range(addr: str) -> bool:
	"""
	Validates if the provided address is a valid IPv4, CIDR or IPv4 range
	:param addr: address to test
	:return: bool: True if provided address is valid
	"""
	from ipaddress import ip_network
	try:
	if '-' in addr: # If we are checking a range, validate both address's individually
	split = addr.split('-')
	return is_valid_ipv4_address_or_range(split[0]) and is_valid_ipv4_address_or_range(split[1])
	else:
	return ip_network(addr).version == 4
	except:
	return False

	def is_valid_ipv6_address_or_range(addr: str) -> bool:
	"""
	Validates if the provided address is a valid IPv4, CIDR or IPv4 range
	:param addr: address to test
	:return: bool: True if provided address is valid
	"""
	from ipaddress import ip_network
	try:
	if '-' in addr: # If we are checking a range, validate both address's individually
	split = addr.split('-')
	return is_valid_ipv6_address_or_range(split[0]) and is_valid_ipv6_address_or_range(split[1])
	else:
	return ip_network(addr).version == 6
	except:
	return False
	diff --git a/src/op_mode/interfaces.py b/src/op_mode/interfaces.py
	index b6bbcbbc2..faeb846eb 100755
	--- a/src/op_mode/interfaces.py
	+++ b/src/op_mode/interfaces.py
	@@ -1,730 +1,737 @@
	#!/usr/bin/env python3
	#
	# Copyright VyOS maintainers and contributors <maintainers@vyos.io>
	#
	# This program is free software; you can redistribute it and/or modify
	# it under the terms of the GNU General Public License version 2 or later as
	# published by the Free Software Foundation.
	#
	# This program is distributed in the hope that it will be useful,
	# but WITHOUT ANY WARRANTY; without even the implied warranty of
	# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
	# GNU General Public License for more details.
	#
	# You should have received a copy of the GNU General Public License
	# along with this program. If not, see <http://www.gnu.org/licenses/>.
	#

	import os
	import re
	import sys
	import glob
	import json
	import typing
	import textwrap
	from datetime import datetime
	from tabulate import tabulate

	import vyos.opmode
	from vyos.ifconfig import Section
	from vyos.ifconfig import Interface
	from vyos.ifconfig import VRRP
	-from vyos.utils.process import cmd
	+from vyos.utils.dict import dict_set_nested
	+from vyos.utils.network import get_interface_vrf
	from vyos.utils.network import interface_exists
	+from vyos.utils.process import cmd
	from vyos.utils.process import rc_cmd
	from vyos.utils.process import call
	from vyos.configquery import op_mode_config_dict

	def catch_broken_pipe(func):
	def wrapped(args, *kwargs):
	try:
	func(args, *kwargs)
	except (BrokenPipeError, KeyboardInterrupt):
	# Flush output to /dev/null and bail out.
	os.dup2(os.open(os.devnull, os.O_WRONLY), sys.stdout.fileno()) # pylint: disable = no-member
	return wrapped

	# The original implementation of filtered_interfaces has signature:
	# (ifnames: list, iftypes: typing.Union[str, list], vif: bool, vrrp: bool) -> intf: Interface:
	# Arg types allowed in CLI (ifnames: str, iftypes: str) were manually
	# re-typed from argparse args.
	# We include the function in a general form, however op-mode standard
	# functions will restrict to the CLI-allowed arg types, wrapped in Optional.
	def filtered_interfaces(ifnames: typing.Union[str, list],
	iftypes: typing.Union[str, list],
	vif: bool, vrrp: bool) -> Interface:
	"""
	get all interfaces from the OS and return them; ifnames can be used to
	filter which interfaces should be considered

	ifnames: a list of interface names to consider, empty do not filter

	return an instance of the Interface class
	"""
	if isinstance(ifnames, str):
	ifnames = [ifnames] if ifnames else []
	if isinstance(iftypes, list):
	for iftype in iftypes:
	yield from filtered_interfaces(ifnames, iftype, vif, vrrp)

	for ifname in Section.interfaces(iftypes):
	# Bail out early if interface name not part of our search list
	if ifnames and ifname not in ifnames:
	continue

	# As we are only "reading" from the interface - we must use the
	# generic base class which exposes all the data via a common API
	interface = Interface(ifname, create=False, debug=False)

	# VLAN interfaces have a '.' in their name by convention
	if vif and not '.' in ifname:
	continue

	if vrrp:
	vrrp_interfaces = VRRP.active_interfaces()
	if ifname not in vrrp_interfaces:
	continue

	yield interface

	def is_interface_has_mac(interface_name):
	interface_no_mac = ('tun', 'wg')
	return not any(interface_name.startswith(prefix) for prefix in interface_no_mac)

	def detailed_output(dataset, headers):
	for data in dataset:
	adjusted_rule = data + [""] * (len(headers) - len(data)) # account for different header length, like default-action
	transformed_rule = [[header, adjusted_rule[i]] for i, header in enumerate(headers) if i < len(adjusted_rule)] # create key-pair list from headers and rules lists; wrap at 100 char

	print(tabulate(transformed_rule, tablefmt="presto"))
	print()

	def _split_text(text, used=0):
	"""
	take a string and attempt to split it to fit with the width of the screen

	text: the string to split
	used: number of characted already used in the screen
	"""
	no_tty = call('tty -s')

	returned = cmd('stty size') if not no_tty else ''
	returned = returned.split()
	if len(returned) == 2:
	_, columns = tuple(int(_) for _ in returned)
	else:
	_, columns = (40, 80)

	desc_len = columns - used

	line = ''
	for word in text.split():
	if len(line) + len(word) < desc_len:
	line = f'{line} {word}'
	continue
	if line:
	yield line[1:]
	else:
	line = f'{line} {word}'

	yield line[1:]

	def _get_counter_val(prev, now):
	"""
	attempt to correct a counter if it wrapped, copied from perl

	prev: previous counter
	now: the current counter
	"""
	# This function has to deal with both 32 and 64 bit counters
	if prev == 0:
	return now

	# device is using 64 bit values assume they never wrap
	value = now - prev
	if (now >> 32) != 0:
	return value

	# The counter has rolled. If the counter has rolled
	# multiple times since the prev value, then this math
	# is meaningless.
	if value < 0:
	value = (4294967296 - prev) + now

	return value

	def _pppoe(ifname):
	out = cmd('ps -C pppd -f')
	if ifname in out:
	return 'C'
	if ifname in [_.split('/')[-1] for _ in glob.glob('/etc/ppp/peers/pppoe*')]:
	return 'D'
	return ''

	def _find_intf_by_ifname(intf_l: list, name: str):
	for d in intf_l:
	if d['ifname'] == name:
	return d
	return {}

	# lifted out of operational.py to separate formatting from data
	def _format_stats(stats, indent=4):
	stat_names = {
	'rx': ['bytes', 'packets', 'errors', 'dropped', 'overrun', 'mcast'],
	'tx': ['bytes', 'packets', 'errors', 'dropped', 'carrier', 'collisions'],
	}

	stats_dir = {
	'rx': ['rx_bytes', 'rx_packets', 'rx_errors', 'rx_dropped', 'rx_over_errors', 'multicast'],
	'tx': ['tx_bytes', 'tx_packets', 'tx_errors', 'tx_dropped', 'tx_carrier_errors', 'collisions'],
	}
	tabs = []
	for rtx in list(stats_dir):
	tabs.append([f'{rtx.upper()}:', ] + stat_names[rtx])
	tabs.append(['', ] + [stats[_] for _ in stats_dir[rtx]])

	s = tabulate(
	tabs,
	stralign="right",
	numalign="right",
	tablefmt="plain"
	)

	p = ' '*indent
	return f'{p}' + s.replace('\n', f'\n{p}')

	def _get_raw_data(ifname: typing.Optional[str],
	iftype: typing.Optional[str],
	vif: bool, vrrp: bool) -> list:
	if ifname is None:
	ifname = ''
	if iftype is None:
	iftype = ''
	ret =[]
	for interface in filtered_interfaces(ifname, iftype, vif, vrrp):
	res_intf = {}
	cache = interface.operational.load_counters()

	out = cmd(f'ip -json addr show {interface.ifname}')
	res_intf_l = json.loads(out)
	res_intf = res_intf_l[0]

	if res_intf['link_type'] == 'tunnel6':
	# Note that 'ip -6 tun show {interface.ifname}' is not json
	# aware, so find in list
	out = cmd('ip -json -6 tun show')
	tunnel = json.loads(out)
	res_intf['tunnel6'] = _find_intf_by_ifname(tunnel,
	interface.ifname)
	if 'ip6_tnl_f_use_orig_tclass' in res_intf['tunnel6']:
	res_intf['tunnel6']['tclass'] = 'inherit'
	del res_intf['tunnel6']['ip6_tnl_f_use_orig_tclass']

	res_intf['counters_last_clear'] = int(cache.get('timestamp', 0))

	res_intf['description'] = interface.get_alias()

	stats = interface.operational.get_stats()
	for k in list(stats):
	stats[k] = _get_counter_val(cache[k], stats[k])

	res_intf['stats'] = stats

	ret.append(res_intf)

	# find pppoe interfaces that are in a transitional/dead state
	if ifname.startswith('pppoe') and not _find_intf_by_ifname(ret, ifname):
	pppoe_intf = {}
	pppoe_intf['unhandled'] = None
	pppoe_intf['ifname'] = ifname
	pppoe_intf['state'] = _pppoe(ifname)
	ret.append(pppoe_intf)

	return ret

	def _get_summary_data(ifname: typing.Optional[str],
	iftype: typing.Optional[str],
	vif: bool, vrrp: bool) -> list:
	if ifname is None:
	ifname = ''
	if iftype is None:
	iftype = ''
	ret = []

	for interface in filtered_interfaces(ifname, iftype, vif, vrrp):
	res_intf = {}

	res_intf['ifname'] = interface.ifname
	res_intf['oper_state'] = interface.operational.get_state()
	res_intf['admin_state'] = interface.get_admin_state()
	res_intf['addr'] = [_ for _ in interface.get_addr() if not _.startswith('fe80::')]
	res_intf['description'] = interface.get_alias()
	res_intf['mtu'] = interface.get_mtu()
	res_intf['mac'] = interface.get_mac() if is_interface_has_mac(interface.ifname) else 'n/a'
	res_intf['vrf'] = interface.get_vrf()

	ret.append(res_intf)

	# find pppoe interfaces that are in a transitional/dead state
	if ifname.startswith('pppoe') and not _find_intf_by_ifname(ret, ifname):
	pppoe_intf = {}
	pppoe_intf['unhandled'] = None
	pppoe_intf['ifname'] = ifname
	pppoe_intf['state'] = _pppoe(ifname)
	ret.append(pppoe_intf)

	return ret

	def _get_counter_data(ifname: typing.Optional[str],
	iftype: typing.Optional[str],
	vif: bool, vrrp: bool) -> list:
	if ifname is None:
	ifname = ''
	if iftype is None:
	iftype = ''
	ret = []
	for interface in filtered_interfaces(ifname, iftype, vif, vrrp):
	res_intf = {}

	oper = interface.operational.get_state()

	if oper not in ('up','unknown'):
	continue

	stats = interface.operational.get_stats()
	cache = interface.operational.load_counters()
	res_intf['ifname'] = interface.ifname
	res_intf['rx_packets'] = _get_counter_val(cache['rx_packets'], stats['rx_packets'])
	res_intf['rx_bytes'] = _get_counter_val(cache['rx_bytes'], stats['rx_bytes'])
	res_intf['tx_packets'] = _get_counter_val(cache['tx_packets'], stats['tx_packets'])
	res_intf['tx_bytes'] = _get_counter_val(cache['tx_bytes'], stats['tx_bytes'])
	res_intf['rx_dropped'] = _get_counter_val(cache['rx_dropped'], stats['rx_dropped'])
	res_intf['tx_dropped'] = _get_counter_val(cache['tx_dropped'], stats['tx_dropped'])
	res_intf['rx_over_errors'] = _get_counter_val(cache['rx_over_errors'], stats['rx_over_errors'])
	res_intf['tx_carrier_errors'] = _get_counter_val(cache['tx_carrier_errors'], stats['tx_carrier_errors'])

	ret.append(res_intf)

	return ret

	def _get_kernel_data(raw, ifname = None, detail = False):
	if ifname:
	# Check if the interface exists
	if not interface_exists(ifname):
	raise vyos.opmode.IncorrectValue(f"{ifname} does not exist!")
	int_name = f'dev {ifname}'
	else:
	int_name = ''

	kernel_interface = json.loads(cmd(f'ip -j -d -s address show {int_name}'))

	# Return early if raw
	if raw:
	return kernel_interface, None

	# Format the kernel data
	kernel_interface_out = _format_kernel_data(kernel_interface, detail)

	return kernel_interface, kernel_interface_out

	def _format_kernel_data(data, detail):
	output_list = []
	podman_vrf = {}
	tmpInfo = {}

	# Sort interfaces by name
	for interface in sorted(data, key=lambda x: x.get('ifname', '')):
	+ interface_name = interface.get('ifname', '')
	+
	+ # Skip VRF interfaces
	if interface.get('linkinfo', {}).get('info_kind') == 'vrf':
	continue
	- elif interface.get('ifname').startswith(('tunl', 'gre', 'erspan', 'pim6reg')):
	+ # Skip spawned interfaces
	+ elif interface_name.startswith(('tunl', 'gre', 'erspan', 'pim6reg')):
	continue

	+ master = interface.get('master', 'default')
	+ vrf = get_interface_vrf(interface)
	+
	# Get the device model; ex. Intel Corporation Ethernet Controller I225-V
	dev_model = interface.get('parentdev', '')
	if 'parentdev' in interface:
	parentdev = interface['parentdev']
	if re.match(r'^[0-9a-fA-F]{4}:', parentdev):
	dev_model = cmd(f'lspci -nn -s {parentdev}').split(']:')[1].strip()

	# Get the IP addresses on interface
	ip_list = []
	has_global = False
	- vrf = 'default'

	for ip in interface['addr_info']:
	if ip.get('scope') in ('global', 'host'):
	has_global = True
	local = ip.get('local', '-')
	prefixlen = ip.get('prefixlen', '')
	ip_list.append(f"{local}/{prefixlen}")

	- if interface.get('ifname').startswith('pod-'):
	- podman_vrf[interface.get('ifname')] = {}
	- podman_vrf[interface.get('ifname')]['vrf'] = interface.get('master', 'default')
	-
	- if interface.get('master', '').startswith('pod-'):
	- vrf = podman_vrf.get(interface.get('master', '')).get('vrf', 'default')
	- elif interface.get('linkinfo', {}).get('info_slave_kind', '') == 'vrf':
	- vrf = interface.get('master', 'default')
	-
	# If no global IP address, add '-'; indicates no IP address on interface
	if not has_global:
	ip_list.append('-')

	+ # Generate a mapping of podman interfaces to their VRF
	+ if interface_name.startswith('pod-'):
	+ dict_set_nested(f'{interface_name}.vrf', master, podman_vrf)
	+
	+ # If the veth interface's master is a podman interface, the VRF is the VRF of the podman interface
	+ if master.startswith('pod-'):
	+ vrf = podman_vrf.get(master).get('vrf', 'default')
	+
	sl_status = ('A' if not 'UP' in interface['flags'] else 'u') + '/' + ('D' if interface['operstate'] == 'DOWN' else 'u')

	# Generate temporary dict to hold data
	- tmpInfo['ifname'] = interface.get('ifname', '')
	+ tmpInfo['ifname'] = interface_name
	tmpInfo['ip'] = ip_list
	- tmpInfo['mac'] = interface.get('address', 'n/a') if is_interface_has_mac(interface.get('ifname', '')) else 'n/a'
	+ tmpInfo['mac'] = interface.get('address', 'n/a') if is_interface_has_mac(interface_name) else 'n/a'
	tmpInfo['mtu'] = interface.get('mtu', '')
	tmpInfo['vrf'] = vrf
	tmpInfo['status'] = sl_status
	tmpInfo['description'] = "\n".join(textwrap.wrap(interface.get('ifalias', ''), width=50))
	tmpInfo['device'] = dev_model
	tmpInfo['alternate_names'] = interface.get('altnames', '')
	tmpInfo['minimum_mtu'] = interface.get('min_mtu', '')
	tmpInfo['maximum_mtu'] = interface.get('max_mtu', '')
	rx_stats = interface.get('stats64', {}).get('rx')
	tx_stats = interface.get('stats64', {}).get('tx')
	tmpInfo['rx_packets'] = rx_stats.get('packets', "")
	tmpInfo['rx_bytes'] = rx_stats.get('bytes', "")
	tmpInfo['rx_errors'] = rx_stats.get('errors', "")
	tmpInfo['rx_dropped'] = rx_stats.get('dropped', "")
	tmpInfo['rx_over_errors'] = rx_stats.get('over_errors', '')
	tmpInfo['multicast'] = rx_stats.get('multicast', "")
	tmpInfo['tx_packets'] = tx_stats.get('packets', "")
	tmpInfo['tx_bytes'] = tx_stats.get('bytes', "")
	tmpInfo['tx_errors'] = tx_stats.get('errors', "")
	tmpInfo['tx_dropped'] = tx_stats.get('dropped', "")
	tmpInfo['tx_carrier_errors'] = tx_stats.get('carrier_errors', "")
	tmpInfo['tx_collisions'] = tx_stats.get('collisions', "")

	# Generate output list; detail adds more fields
	output_list.append([tmpInfo['ifname'],
	'\n'.join(tmpInfo['ip']),
	tmpInfo['mac'],
	tmpInfo['vrf'],
	tmpInfo['mtu'],
	tmpInfo['status'],
	tmpInfo['description'],
	*([tmpInfo['device']] if detail else []),
	*(['\n'.join(tmpInfo['alternate_names'])] if detail else []),
	*([tmpInfo['minimum_mtu']] if detail else []),
	*([tmpInfo['maximum_mtu']] if detail else []),
	*([tmpInfo['rx_packets']] if detail else []),
	*([tmpInfo['rx_bytes']] if detail else []),
	*([tmpInfo['rx_errors']] if detail else []),
	*([tmpInfo['rx_dropped']] if detail else []),
	*([tmpInfo['rx_over_errors']] if detail else []),
	*([tmpInfo['multicast']] if detail else []),
	*([tmpInfo['tx_packets']] if detail else []),
	*([tmpInfo['tx_bytes']] if detail else []),
	*([tmpInfo['tx_errors']] if detail else []),
	*([tmpInfo['tx_dropped']] if detail else []),
	*([tmpInfo['tx_carrier_errors']] if detail else []),
	*([tmpInfo['tx_collisions']] if detail else [])])

	return output_list

	@catch_broken_pipe
	def _format_show_data(data: list):
	unhandled = []
	for intf in data:
	if 'unhandled' in intf:
	unhandled.append(intf)
	continue
	# instead of reformatting data, call non-json output:
	rc, out = rc_cmd(f"ip addr show {intf['ifname']}")
	if rc != 0:
	continue
	out = re.sub('^\d+:\s+','',out)
	# add additional data already collected
	if 'tunnel6' in intf:
	t6_d = intf['tunnel6']
	t6_str = 'encaplimit %s hoplimit %s tclass %s flowlabel %s (flowinfo %s)' % (
	t6_d.get('encap_limit', ''), t6_d.get('hoplimit', ''),
	t6_d.get('tclass', ''), t6_d.get('flowlabel', ''),
	t6_d.get('flowinfo', ''))
	out = re.sub('(\n\s+)(link/tunnel6)', f'\g<1>{t6_str}\g<1>\g<2>', out)
	print(out)
	ts = intf.get('counters_last_clear', 0)
	if ts:
	when = datetime.fromtimestamp(ts).strftime("%a %b %d %R:%S %Z %Y")
	print(f' Last clear: {when}')
	description = intf.get('description', '')
	if description:
	print(f' Description: {description}')

	stats = intf.get('stats', {})
	if stats:
	print()
	print(_format_stats(stats))

	for intf in unhandled:
	string = {
	'C': 'Coming up',
	'D': 'Link down'
	}[intf['state']]
	print(f"{intf['ifname']}: {string}")

	return 0

	@catch_broken_pipe
	def _format_show_summary(data):
	format1 = '%-16s %-33s %-4s %s'
	format2 = '%-16s %s'

	print('Codes: S - State, L - Link, u - Up, D - Down, A - Admin Down')
	print(format1 % ("Interface", "IP Address", "S/L", "Description"))
	print(format1 % ("---------", "----------", "---", "-----------"))

	unhandled = []
	for intf in data:
	if 'unhandled' in intf:
	unhandled.append(intf)
	continue
	ifname = [intf['ifname'],]
	oper = ['u',] if intf['oper_state'] in ('up', 'unknown') else ['D',]
	admin = ['u',] if intf['admin_state'] in ('up', 'unknown') else ['A',]
	addrs = intf['addr'] or ['-',]
	descs = list(_split_text(intf['description'], 0))

	while ifname or oper or admin or addrs or descs:
	i = ifname.pop(0) if ifname else ''
	a = addrs.pop(0) if addrs else ''
	d = descs.pop(0) if descs else ''
	s = [admin.pop(0)] if admin else []
	l = [oper.pop(0)] if oper else []
	if len(a) < 33:
	print(format1 % (i, a, '/'.join(s+l), d))
	else:
	print(format2 % (i, a))
	print(format1 % ('', '', '/'.join(s+l), d))

	for intf in unhandled:
	string = {
	'C': 'u/D',
	'D': 'A/D'
	}[intf['state']]
	print(format1 % (ifname, '', string, ''))

	return 0

	@catch_broken_pipe
	def _format_show_summary_extended(data):
	headers = ["Interface", "IP Address", "MAC", "VRF", "MTU", "S/L", "Description"]
	table_data = []

	print('Codes: S - State, L - Link, u - Up, D - Down, A - Admin Down')

	for intf in data:
	if 'unhandled' in intf:
	continue

	ifname = intf['ifname']
	oper_state = 'u' if intf['oper_state'] in ('up', 'unknown') else 'D'
	admin_state = 'u' if intf['admin_state'] in ('up', 'unknown') else 'A'
	addrs = intf['addr'] or ['-']
	description = '\n'.join(_split_text(intf['description'], 0))
	mac = intf['mac'] if intf['mac'] else 'n/a'
	mtu = intf['mtu'] if intf['mtu'] else 'n/a'
	vrf = intf['vrf'] if intf['vrf'] else 'default'

	ip_addresses = '\n'.join(ip for ip in addrs)

	# Create a row for the table
	row = [
	ifname,
	ip_addresses,
	mac,
	vrf,
	mtu,
	f"{admin_state}/{oper_state}",
	description,
	]

	# Append the row to the table data
	table_data.append(row)

	for intf in data:
	if 'unhandled' in intf:
	string = {'C': 'u/D', 'D': 'A/D'}[intf['state']]
	table_data.append([intf['ifname'], '', '', '', '', string, ''])

	print(tabulate(table_data, headers))

	return 0

	@catch_broken_pipe
	def _format_show_counters(data: list):
	data_entries = []
	for entry in data:
	interface = entry.get('ifname')
	rx_packets = entry.get('rx_packets')
	rx_bytes = entry.get('rx_bytes')
	tx_packets = entry.get('tx_packets')
	tx_bytes = entry.get('tx_bytes')
	rx_dropped = entry.get('rx_dropped')
	tx_dropped = entry.get('tx_dropped')
	rx_errors = entry.get('rx_over_errors')
	tx_errors = entry.get('tx_carrier_errors')
	data_entries.append([interface, rx_packets, rx_bytes, tx_packets, tx_bytes, rx_dropped, tx_dropped, rx_errors, tx_errors])

	headers = ['Interface', 'Rx Packets', 'Rx Bytes', 'Tx Packets', 'Tx Bytes', 'Rx Dropped', 'Tx Dropped', 'Rx Errors', 'Tx Errors']
	output = tabulate(data_entries, headers, numalign="left")
	print (output)
	return output

	def show_kernel(raw: bool, intf_name: typing.Optional[str], detail: bool):
	raw_data, data = _get_kernel_data(raw, intf_name, detail)

	# Return early if raw
	if raw:
	return raw_data

	# Normal headers; show interfaces kernel
	headers = ['Interface', 'IP Address', 'MAC', 'VRF', 'MTU', 'S/L', 'Description']

	# Detail headers; show interfaces kernel detail
	detail_header = ['Interface', 'IP Address', 'MAC', 'VRF', 'MTU', 'S/L', 'Description',
	'Device', 'Alternate Names','Minimum MTU', 'Maximum MTU', 'RX_Packets',
	'RX_Bytes', 'RX_Errors', 'RX_Dropped', 'Receive Overrun Errors', 'Received Multicast',
	'TX_Packets', 'TX_Bytes', 'TX_Errors', 'TX_Dropped', 'Transmit Carrier Errors',
	'Transmit Collisions']

	if detail:
	detailed_output(data, detail_header)
	else:
	print('Codes: S - State, L - Link, u - Up, D - Down, A - Admin Down')
	print(tabulate(data, headers))

	def _show_raw(data: list, intf_name: str):
	if intf_name is not None and len(data) <= 1:
	try:
	return data[0]
	except IndexError:
	raise vyos.opmode.UnconfiguredObject(
	f"Interface {intf_name} does not exist")
	else:
	return data


	def show(raw: bool, intf_name: typing.Optional[str],
	intf_type: typing.Optional[str],
	vif: bool, vrrp: bool):
	data = _get_raw_data(intf_name, intf_type, vif, vrrp)
	if raw:
	return _show_raw(data, intf_name)
	return _format_show_data(data)

	def show_summary(raw: bool, intf_name: typing.Optional[str],
	intf_type: typing.Optional[str],
	vif: bool, vrrp: bool):
	data = _get_summary_data(intf_name, intf_type, vif, vrrp)
	if raw:
	return _show_raw(data, intf_name)
	return _format_show_summary(data)

	def show_summary_extended(raw: bool, intf_name: typing.Optional[str],
	intf_type: typing.Optional[str],
	vif: bool, vrrp: bool):
	data = _get_summary_data(intf_name, intf_type, vif, vrrp)
	if raw:
	return _show_raw(data, intf_name)
	return _format_show_summary_extended(data)

	def show_counters(raw: bool, intf_name: typing.Optional[str],
	intf_type: typing.Optional[str],
	vif: bool, vrrp: bool):
	data = _get_counter_data(intf_name, intf_type, vif, vrrp)
	if raw:
	return _show_raw(data, intf_name)
	return _format_show_counters(data)

	def show_vlan_to_vni(raw: bool, intf_name: typing.Optional[str], vid: typing.Optional[str], detail: bool):
	if not interface_exists(intf_name):
	raise vyos.opmode.UnconfiguredObject(f"Interface {intf_name} does not exist\n")

	if not vid:
	vid = "all"

	tunnel_data = json.loads(cmd(f"bridge -j vlan tunnelshow dev {intf_name} vid {vid}"))

	if not tunnel_data:
	if vid == "all":
	raise vyos.opmode.UnconfiguredObject(f"No VLAN-to-VNI mapping found for interface {intf_name}\n")
	else:
	raise vyos.opmode.UnconfiguredObject(f"No VLAN-to-VNI mapping found for VLAN {vid}\n")

	statistics_data = json.loads(cmd(f"bridge -j -s vlan tunnelshow dev {intf_name} vid {vid}"))[0]

	mapping_config = op_mode_config_dict(['interfaces', 'vxlan', intf_name, 'vlan-to-vni'],
	get_first_key=True)

	raw_data = {intf_name: {}}
	output_list = []

	for tunnel in tunnel_data[0].get("tunnels", []):
	tunnel_id = tunnel.get("tunid")
	tunnel_dict = raw_data[intf_name][tunnel_id] = {}

	for vlan in statistics_data.get("vlans", []):
	if vlan.get("vid") == tunnel.get("vlan"):
	vlan_id = str(vlan.get("vid"))
	description = mapping_config.get(vlan_id, {}).get("description", "")

	if raw:
	tunnel_dict["vlan"] = vlan_id
	tunnel_dict["rx_bytes"] = vlan.get("rx_bytes")
	tunnel_dict["tx_bytes"] = vlan.get("tx_bytes")
	tunnel_dict["rx_packets"] = vlan.get("rx_packets")
	tunnel_dict["tx_packets"] = vlan.get("tx_packets")
	tunnel_dict["description"] = description
	else:
	#Generate output list; detail adds more fields
	output_list.append([
	*([intf_name] if not detail else []),
	vlan_id,
	tunnel_id,
	description,
	*([vlan.get("rx_bytes")] if detail else []),
	*([vlan.get("tx_bytes")] if detail else []),
	*([vlan.get("rx_packets")] if detail else []),
	*([vlan.get("tx_packets")] if detail else [])
	])

	if raw:
	return raw_data

	if detail:
	# Detail headers; ex. show interfaces vxlan vxlan1 vlan-to-vni detail
	detail_header = ['VLAN', 'VNI', 'Description', 'Rx Bytes', 'Tx Bytes', 'Rx Packets', 'Tx Packets']
	print('-' * 35)
	print(f"Interface: {intf_name}\n")
	detailed_output(output_list, detail_header)
	else:
	# Normal headers; ex. show interfaces vxlan vxlan1 vlan-to-vni
	headers = ['Interface', 'VLAN', 'VNI', 'Description']
	print(tabulate(output_list, headers))

	def clear_counters(intf_name: typing.Optional[str],
	intf_type: typing.Optional[str],
	vif: bool, vrrp: bool):
	for interface in filtered_interfaces(intf_name, intf_type, vif, vrrp):
	interface.operational.clear_counters()

	def reset_counters(intf_name: typing.Optional[str],
	intf_type: typing.Optional[str],
	vif: bool, vrrp: bool):
	for interface in filtered_interfaces(intf_name, intf_type, vif, vrrp):
	interface.operational.reset_counters()

	if __name__ == '__main__':
	try:
	res = vyos.opmode.run(sys.modules[__name__])
	if res:
	print(res)
	except (ValueError, vyos.opmode.Error) as e:
	print(e)
	sys.exit(1)

File Metadata

Mime Type: text/x-diff
Expires: Fri, Nov 6, 9:37 AM (10 h, 17 m)
Storage Engine: blob
Storage Format: Raw Data
Storage Handle: 4601474
Default Alt Text: (65 KB)

No OneTemporaryActions

View Options

File Metadata

Event Timeline

No OneTemporary
Actions