diff options
Diffstat (limited to 'python')
29 files changed, 3804 insertions, 0 deletions
diff --git a/python/vyos/vpp/__init__.py b/python/vyos/vpp/__init__.py new file mode 100644 index 000000000..3352914f1 --- /dev/null +++ b/python/vyos/vpp/__init__.py @@ -0,0 +1,20 @@ +# +# Copyright (C) VyOS Inc. +# +# This program is free software; you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation; either version 2 of the License, or +# (at your option) any later version. +# +# 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, write to the Free Software Foundation, Inc., +# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. + +from .control_vpp import VPPControl + +__all__ = ['VPPControl'] diff --git a/python/vyos/vpp/acl/__init__.py b/python/vyos/vpp/acl/__init__.py new file mode 100644 index 000000000..6bd6cdb23 --- /dev/null +++ b/python/vyos/vpp/acl/__init__.py @@ -0,0 +1,3 @@ +from .acl import Acl + +__all__ = ['Acl'] diff --git a/python/vyos/vpp/acl/acl.py b/python/vyos/vpp/acl/acl.py new file mode 100644 index 000000000..57d36ac91 --- /dev/null +++ b/python/vyos/vpp/acl/acl.py @@ -0,0 +1,106 @@ +# +# Copyright (C) VyOS Inc. +# +# This program is free software; you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation; either version 2 of the License, or +# (at your option) any later version. +# +# 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, write to the Free Software Foundation, Inc., +# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. + +from vyos.vpp import VPPControl + + +NO_ACL_INDEX = 0xFFFFFFFF + + +class Acl: + def __init__(self): + self.vpp = VPPControl() + + def get_acl_index_by_tag(self, tag): + """Get ACL index by tag name + https://github.com/FDio/vpp/blob/21c641f9356da5137760cdc799127064c8c1fd31/src/plugins/acl/acl.api + """ + for acl in self.vpp.api.acl_dump(acl_index=NO_ACL_INDEX): + if acl.tag == tag: + return acl.acl_index + return NO_ACL_INDEX + + def add_replace_acl(self, tag, rules): + """Add new ACL or replace existing one""" + self.vpp.api.acl_add_replace( + tag=tag, + acl_index=self.get_acl_index_by_tag(tag), + count=len(rules), + r=rules, + ) + + def delete_acl(self, tag): + """Delete existing ACL""" + self.vpp.api.acl_del(acl_index=self.get_acl_index_by_tag(tag)) + + def add_acl_interface(self, interface, input_tags, output_tags): + """Add or replace ACLs on interface""" + acls = [] + for tag in input_tags: + acl_index = self.get_acl_index_by_tag(tag) + acls.append(acl_index) + for tag in output_tags: + acl_index = self.get_acl_index_by_tag(tag) + acls.append(acl_index) + self.vpp.api.acl_interface_set_acl_list( + sw_if_index=self.vpp.get_sw_if_index(interface), + count=len(acls), + n_input=len(input_tags), + acls=acls, + ) + + def delete_acl_interface(self, interface): + """Delete ACLs from interface""" + self.vpp.api.acl_interface_set_acl_list( + sw_if_index=self.vpp.get_sw_if_index(interface), + count=0, + ) + + def get_macip_acl_index_by_tag(self, tag): + """Get macip ACL by tag name""" + for acl in self.vpp.api.macip_acl_dump(): + if acl.tag == tag: + return acl.acl_index + return NO_ACL_INDEX + + def add_replace_acl_macip(self, tag, rules): + """Add or replace existing macip ACL""" + self.vpp.api.macip_acl_add_replace( + tag=tag, + acl_index=self.get_macip_acl_index_by_tag(tag), + count=len(rules), + r=rules, + ) + + def delete_acl_macip(self, tag): + """Delete existing macip ACL""" + self.vpp.api.macip_acl_del(acl_index=self.get_macip_acl_index_by_tag(tag)) + + def add_acl_macip_interface(self, interface, tag): + """Add or replace macip ACLs on interface""" + self.vpp.api.macip_acl_interface_add_del( + sw_if_index=self.vpp.get_sw_if_index(interface), + acl_index=self.get_macip_acl_index_by_tag(tag), + is_add=True, + ) + + def delete_acl_macip_interface(self, interface): + """Delete macip ACLs from interface""" + self.vpp.api.macip_acl_interface_add_del( + sw_if_index=self.vpp.get_sw_if_index(interface), + is_add=False, + ) diff --git a/python/vyos/vpp/config_deps.py b/python/vyos/vpp/config_deps.py new file mode 100644 index 000000000..dffd865d1 --- /dev/null +++ b/python/vyos/vpp/config_deps.py @@ -0,0 +1,76 @@ +# +# Copyright (C) VyOS Inc. +# +# This program is free software; you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation; either version 2 of the License, or +# (at your option) any later version. +# +# 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, write to the Free Software Foundation, Inc., +# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. + + +def deps_xconnect_dict(conf) -> dict[str, list[str]]: + """Get a dict of all xconnect interface members: + + keys: members + + values: xconnect interfaces + + Args: + conf (config): VyOS config object + + Returns: + dict[str, list[str]]: dict of members + """ + xconn_members_dict: dict[str, list[str]] = {} + config = conf.get_config_dict( + ['vpp', 'interfaces', 'xconnect'], + key_mangling=('-', '_'), + get_first_key=True, + no_tag_node_value_mangle=True, + ) + + for xconn_name, xconn_config in config.items(): + for member_name in xconn_config.get('member', {}).get('interface', []): + xconn_ifaces_list = xconn_members_dict.get(xconn_name, []) + xconn_ifaces_list.append(xconn_name) + xconn_members_dict.update({member_name: xconn_ifaces_list}) + + return xconn_members_dict + + +def deps_bridge_dict(conf) -> dict[str, list[str]]: + """Get a dict of all bridge interface members: + + keys: members + + values: bridge interfaces + + Args: + conf (config): VyOS config object + + Returns: + dict[str, list[str]]: dict of members + """ + bridge_members_dict: dict[str, list[str]] = {} + config = conf.get_config_dict( + ['vpp', 'interfaces', 'bridge'], + key_mangling=('-', '_'), + get_first_key=True, + no_tag_node_value_mangle=True, + ) + + for bridge_name, bridge_config in config.items(): + for member_name in bridge_config.get('member', {}).get('interface', []): + bridge_ifaces_list = bridge_members_dict.get(bridge_name, []) + bridge_ifaces_list.append(bridge_name) + bridge_members_dict.update({member_name: bridge_ifaces_list}) + + return bridge_members_dict diff --git a/python/vyos/vpp/config_filter.py b/python/vyos/vpp/config_filter.py new file mode 100644 index 000000000..dc864930b --- /dev/null +++ b/python/vyos/vpp/config_filter.py @@ -0,0 +1,58 @@ +# +# Copyright (C) VyOS Inc. +# +# This program is free software; you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation; either version 2 of the License, or +# (at your option) any later version. +# +# 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, write to the Free Software Foundation, Inc., +# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. + +from vyos.config import Config + + +def iface_filter_eth(config: Config, iface: str) -> None: + """Filter out unsupported config nodes from Ethernet interface config + + Args: + config (Config): config object + iface (str): Ethernet interface name to filter + """ + allowed_nodes: list[str] = [ + 'address', + 'description', + 'dhcp-options', + 'dhcpv6-options', + 'disable', + 'eapol', + 'hw-id', + 'ip', + 'ipv6', + 'mtu', + 'redirect', + 'vif', + 'vif-s', + 'vrf', + ] + + # get list of config nides in a session configuration + iface_nodes = config._session_config.list_nodes(['interfaces', 'ethernet', iface]) + + # clean cached session config + if False in config._dict_cache: + del config._dict_cache[False] + + # remove unsupported config nodes + for cfg_node in iface_nodes: + if cfg_node not in allowed_nodes: + config._session_config.delete(['interfaces', 'ethernet', iface, cfg_node]) + print( + f'WARNING: {cfg_node} option in {iface} settings is not supported by VPP interfaces. It will be ignored.' + ) diff --git a/python/vyos/vpp/config_resource_checks/__init__.py b/python/vyos/vpp/config_resource_checks/__init__.py new file mode 100644 index 000000000..e69de29bb --- /dev/null +++ b/python/vyos/vpp/config_resource_checks/__init__.py diff --git a/python/vyos/vpp/config_resource_checks/cpu.py b/python/vyos/vpp/config_resource_checks/cpu.py new file mode 100644 index 000000000..b319608e4 --- /dev/null +++ b/python/vyos/vpp/config_resource_checks/cpu.py @@ -0,0 +1,76 @@ +# Used for validating estimated CPU/physical cores use +# +# Copyright (C) VyOS Inc. +# +# This program is free software; you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation; either version 2 of the License, or +# (at your option) any later version. +# +# 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, write to the Free Software Foundation, Inc., +# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. + +from vyos.utils.cpu import get_available_cpus, get_core_count + +from vyos.vpp.config_resource_checks.resource_defaults import default_resource_map + + +# Get default value for reserved cpu cores +reserved_cpus = default_resource_map.get('reserved_cpu_cores') + + +def available_cores_count(cpu_settings: dict) -> int: + core_count = get_core_count() + + if cpu_settings.get('main_core'): + core_count -= 1 + + skip_cores = int(cpu_settings.get('skip_cores', 0)) + # The default settings assume that + # at least 2 CPU cores should remain reserved for system use + # (only in case of current runtime is not smoke test) + if skip_cores < reserved_cpus: + core_count -= reserved_cpus + else: + core_count -= skip_cores + + return core_count + + +def available_cores_list(skip_cores: int) -> list: + # Available cores are all CPU cores without first N skipped cores that will not be used + # Get all available physical cores - use set to filter out unique values + cpu_cores = set(map(lambda el: el['cpu'], get_available_cpus())) + cpu_cores = list(cpu_cores) + + return cpu_cores[skip_cores:] + + +def worker_cores_list(iface: str, worker_ranges: list) -> list: + all_core_numbers = [] + for worker_range in worker_ranges: + core_numbers = worker_range.split('-') + + if int(core_numbers[0]) > int(core_numbers[-1]): + raise ValueError( + f'Range for "{iface} workers {worker_range}" is not correct' + ) + + all_core_numbers.extend(range(int(core_numbers[0]), int(core_numbers[-1]) + 1)) + + # Check for duplicates + duplicates = set( + [str(x) for n, x in enumerate(all_core_numbers) if x in all_core_numbers[:n]] + ) + if duplicates: + raise ValueError( + f'Some workers in "{iface} workers" are duplicated: #{",".join(list(duplicates))}' + ) + + return all_core_numbers diff --git a/python/vyos/vpp/config_resource_checks/memory.py b/python/vyos/vpp/config_resource_checks/memory.py new file mode 100644 index 000000000..bce2df2bb --- /dev/null +++ b/python/vyos/vpp/config_resource_checks/memory.py @@ -0,0 +1,153 @@ +# Used for memory consumption calculations +# +# Copyright (C) VyOS Inc. +# +# This program is free software; you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation; either version 2 of the License, or +# (at your option) any later version. +# +# 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, write to the Free Software Foundation, Inc., +# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. + +import re + +from vyos.utils.process import cmd +from vyos.vpp.utils import ( + human_memory_to_bytes, + human_page_memory_to_bytes, +) +from vyos.vpp.config_resource_checks.resource_defaults import default_resource_map + + +def get_hugepages_info() -> dict: + """ + Returns the information about HugePages for default hugepage size + retrieved from /proc/meminfo + """ + info = {} + with open('/proc/meminfo', 'r') as meminfo: + for line in meminfo: + if line.startswith('Huge'): + key, value, *_ = line.strip().split() + info[key.rstrip(':')] = int(value) + return info + + +def get_total_hugepages_memory() -> int: + """ + Returns the total amount of hugepage memory (in bytes) + """ + info = get_hugepages_info() + hugepages_total = info.get('HugePages_Total') + hugepage_size = info.get('Hugepagesize') * 1024 + + return hugepage_size * hugepages_total + + +def get_total_hugepages_count() -> int: + """ + Returns the total count of hugepages + """ + info = get_hugepages_info() + return info.get('HugePages_Total') + + +def get_numa_count(): + """ + Run `numactl --hardware` and parse the 'available:' line. + """ + out = cmd('numactl --hardware') + # e.g. "available: 2 nodes (0-1)" + m = re.search(r'available:\s*(\d+)\s+nodes', out) + return int(m.group(1)) if m else 0 + + +def buffer_size(settings: dict) -> int: + numa_count = get_numa_count() + buffers_per_numa = int( + settings.get('buffers', {}).get( + 'buffers_per_numa', default_resource_map.get('buffers_per_numa') + ) + ) + data_size = int( + settings.get('buffers', {}).get( + 'data_size', default_resource_map.get('data_size') + ) + ) + buffers_memory = buffers_per_numa * data_size * numa_count + return buffers_memory + + +def main_heap_page_size(settings: dict) -> int: + heap_page_size = settings.get('memory', {}).get( + 'main_heap_page_size', default_resource_map.get('main_heap_page_size') + ) + return human_page_memory_to_bytes(heap_page_size) + + +def memory_main_heap(settings: dict) -> int: + heap_size = settings.get('memory', {}).get( + 'main_heap_size', default_resource_map.get('main_heap_size') + ) + return human_memory_to_bytes(heap_size) + + +def ipv6_heap_size(settings: dict) -> int: + heap_size = settings.get('ipv6', {}).get( + 'heap_size', default_resource_map.get('ipv6_heap_size') + ) + return human_memory_to_bytes(heap_size) + + +def total_heap_size(heap_size: int, heap_page_size: int) -> int: + return (heap_size + heap_page_size - 1) & ~(heap_page_size - 1) + + +def statseg_size(settings: dict) -> int: + statseg_memory = settings.get('statseg', {}).get( + 'size', default_resource_map.get('statseg_heap_size') + ) + return human_memory_to_bytes(statseg_memory) + + +def statseg_page_size(settings: dict) -> int: + page_size = settings.get('statseg', {}).get('page_size', 'default') + return human_page_memory_to_bytes(page_size) + + +def total_statseg_size(_statseg_size: int, _statseg_page: int) -> int: + return (_statseg_size + _statseg_page - 1) & ~(_statseg_page - 1) + + +def total_memory_required(settings: dict) -> int: + mem_required = 0 + + mem_stats = { + 'memory_buffers': buffer_size(settings), + 'netlink_buffer_size': int( + settings.get('lcp', {}).get( + 'rx_buffer_size', default_resource_map.get('netlink_rx_buffer_size') + ) + ), + 'heap_size': total_heap_size( + heap_size=memory_main_heap(settings), + heap_page_size=main_heap_page_size(settings), + ), + 'statseg_size': total_statseg_size( + _statseg_size=statseg_size(settings), + _statseg_page=statseg_page_size(settings), + ), + 'ipv6_heap_size': ipv6_heap_size(settings), + } + + for stat in mem_stats: + mem_required += mem_stats[stat] + + return mem_required diff --git a/python/vyos/vpp/config_resource_checks/resource_defaults.py b/python/vyos/vpp/config_resource_checks/resource_defaults.py new file mode 100644 index 000000000..797b45701 --- /dev/null +++ b/python/vyos/vpp/config_resource_checks/resource_defaults.py @@ -0,0 +1,43 @@ +# Default values for resource consumption checks +# +# Copyright (C) VyOS Inc. +# +# This program is free software; you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation; either version 2 of the License, or +# (at your option) any later version. +# +# 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, write to the Free Software Foundation, Inc., +# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. + + +default_resource_map = { + # Default amount of buffers per NUMA (populated CPU socket) + 'buffers_per_numa': 16384, + # Default size of buffer (in bytes) + 'data_size': 2048, + # Default hugepage size for VPP + 'hugepage_size': '2M', + # Default amount of memory allocated for VPP exclusive usage + 'main_heap_size': '3G', + # Default main heap page size + 'main_heap_page_size': '2M', + # Default size of buffers transferred via netlink + 'netlink_rx_buffer_size': 212992, + # Default amount of memory allocated for VPP stats segment usage + 'statseg_heap_size': '96M', + # Minimal amount of memory required to start VPP + 'min_memory': '8G', + # Minimal number of physical CPU cores required to start VPP + 'min_cpus': 4, + # Reserve at least 2 physical cores + 'reserved_cpu_cores': 2, + # Default heap size for IPv6 + 'ipv6_heap_size': '32M', +} diff --git a/python/vyos/vpp/config_verify.py b/python/vyos/vpp/config_verify.py new file mode 100644 index 000000000..1beb3141b --- /dev/null +++ b/python/vyos/vpp/config_verify.py @@ -0,0 +1,407 @@ +# Used for verifying configuration vpp interfaces +# +# Copyright (C) VyOS Inc. +# +# This program is free software; you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation; either version 2 of the License, or +# (at your option) any later version. +# +# 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, write to the Free Software Foundation, Inc., +# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. + +import psutil + +from vyos import ConfigError +from vyos.base import Warning +from vyos.utils.cpu import get_core_count as total_core_count + +from vyos.vpp.control_host import get_eth_driver +from vyos.vpp.config_resource_checks import cpu as cpu_checks, memory as mem_checks +from vyos.vpp.config_resource_checks.resource_defaults import default_resource_map +from vyos.vpp.utils import human_memory_to_bytes, bytes_to_human_memory + + +def verify_vpp_remove_kernel_interface(config: dict): + """Common verify for removed kernel-interfaces. + Verify that removed kernel interface are not used in 'vpp kernel-interfaces'. + + Example: + delete vpp interfaces gre|vxlan <tag>X kernel-interface vpp-tunX + set vpp kernel-interface vpp-tunX + """ + if ( + 'remove' in config + and 'kernel_interface_removed' in config + and 'vpp_kernel_interfaces' in config + ): + removed_interfaces = config['kernel_interface_removed'] + used_interfaces = config['vpp_kernel_interfaces'] + + for interface in removed_interfaces: + if interface in used_interfaces: + raise ConfigError( + f'"{interface}" is still in use within "vpp kernel-interfaces". ' + 'Please remove it before proceeding.' + ) + + +def verify_vpp_change_kernel_interface(config: dict): + """Common verify for changed kernel-interface + + Example: + set vpp interfaces gre|vxlan <tag> kernel-interface vpp-tunX' + commit + set vpp interfaces gre|vxlan <tag> kernel-interface vpp-tunY' + commit + + check if we have kernel interface config 'vpp kernel-interface vpp-tunX' + """ + kernel_interface_removed = config.get('kernel_interface_removed', []) + vpp_kernel_interfaces = config.get('vpp_kernel_interfaces', {}) + + for interface in kernel_interface_removed: + if interface in vpp_kernel_interfaces: + raise ConfigError( + f'interface "{interface}" is still in use within "vpp kernel-interfaces". ' + f'Please remove it "vpp kernel-interface {interface}" before proceeding.' + ) + + +def verify_vpp_exists_kernel_interface(config: dict): + """Verify is a kernel-interface already created by another VPP LCP pair + + Example: + set vpp interfaces vxlan vxlan10 kernel-interface vpp-tun10' + commit + set vpp interfaces vxlan vxlan20 kernel-interface vpp-tun10' + commit + """ + kernel_interface = config.get('kernel_interface', '') + vpp_interface = config.get('ifname', '') + candidate_kernel_interfaces = config.get('candidate_kernel_interfaces', []) + + for candidate_kernel_iface in candidate_kernel_interfaces: + if ( + vpp_interface != candidate_kernel_iface[0] + and kernel_interface == candidate_kernel_iface[1] + ): + raise ConfigError( + f'Kernel interface "{kernel_interface}" is already configured for {candidate_kernel_iface[0]}. ' + 'Duplicates are not allowed.' + ) + + +def verify_vpp_remove_xconnect_interface(config: dict): + if not config.get('remove'): + return + for xconn_member, xconn_iface in config.get('xconn_members').items(): + if xconn_member == config.get('ifname'): + raise ConfigError( + f'interface "{xconn_member}" is still in use within "vpp interfaces xconnect". ' + f'Please remove it from "vpp interface xconnect {xconn_iface}" before proceeding.' + ) + + +def verify_vpp_tunnel_source_address(config: dict): + from vyos.utils.network import is_intf_addr_assigned + + address = config.get('source_address') + for iface in config.get('vpp_ether_vif_ifaces', []): + if is_intf_addr_assigned(iface, address): + return True + + raise ConfigError( + f'Source address "{address}" is not assigned on any Ethernet or VIF interface!' + ) + + +def verify_dev_driver(iface_name: str, driver_type: str) -> bool: + # Lists of drivers compatible with DPDK and XDP + drivers_dpdk: list[str] = [ + 'atlantic', + 'bnx2x', + 'e1000', + 'ena', + 'gve', + 'hv_netvsc', + 'i40e', + 'ice', + 'igc', + 'ixgbe', + 'liquidio', + 'mlx4_core', + 'mlx5_core', + 'qede', + 'sfc', + 'tap', + 'tun', + 'virtio_net', + 'vmxnet3', + ] + + drivers_xdp: list[str] = [ + 'atlantic', + 'ena', + 'gve', + 'hv_netvsc', + 'i40e', + 'ice', + 'igb', + 'igc', + 'ixgbe', + 'mlx4_core', + 'mlx5_core', + 'qede', + 'sfc', + 'tap', + 'tun', + 'virtio_net', + 'vmxnet3', + ] + + driver: str = get_eth_driver(iface_name) + + if driver_type == 'dpdk': + if driver in drivers_dpdk: + return True + elif driver_type == 'xdp': + if driver in drivers_xdp: + return True + else: + raise ConfigError(f'"Driver type {driver_type} is wrong') + + return False + + +def verify_vpp_minimum_cpus(): + """ + Verify that the host system has enough physical CPU cores + Current minimal requirement is 4 + """ + min_cpus = default_resource_map.get('min_cpus') + if total_core_count() < min_cpus: + raise ConfigError( + 'This system does not meet minimal requirements for VPP. ' + f'Minimum {min_cpus} CPU cores are required.' + ) + + +def verify_vpp_minimum_memory(): + """ + Verify that the host system has enough RAM + Calculate by retrieving the amount of physical memory + And the minimal requirement (currently 8 GB). Round before comparing - + To avoid situations like when a machine nominally has 8192 MB (8 giga/gibibytes) + But the OS sees only 7.75 GB, creating a fail condition for this check + """ + min_mem = default_resource_map.get('min_memory') + total_memory = round(psutil.virtual_memory().total / (1024**3)) + min_memory = round(human_memory_to_bytes(min_mem) / (1024**3)) + + if total_memory < min_memory: + raise ConfigError( + 'This system does not meet minimal requirements for VPP. ' + f'Minimum {min_memory} GB of RAM are required.' + ) + + +def verify_vpp_memory(config: dict): + main_heap_size = mem_checks.memory_main_heap(config['settings']) + main_heap_page_size = mem_checks.main_heap_page_size(config['settings']) + + if main_heap_size < 51 << 20: + raise ConfigError('The main heap size must be greater than or equal to 51M') + + readable_heap_page = bytes_to_human_memory(main_heap_page_size, 'K') + + if main_heap_page_size > main_heap_size: + raise ConfigError( + f'The main heap size must be greater than or equal to page-size ({readable_heap_page})' + ) + + available_memory = mem_checks.get_total_hugepages_memory() + memory_required = mem_checks.total_memory_required(config['settings']) + + if main_heap_size > available_memory: + available_memory_in_mb = bytes_to_human_memory(available_memory, 'M') + raise ConfigError( + f'"memory main-heap-size" must not be greater than hugepages memory. Reduce to {available_memory_in_mb} or less' + ) + + memory_required = round(memory_required / 1024**3, 1) + available_memory = round(available_memory / 1024**3, 1) + + # Allow 10% error margin + allowed_margin = memory_required * 0.1 + + # Compare HugePage memory with required memory for VPP + if memory_required > available_memory + allowed_margin: + raise ConfigError( + f'Not enough free hugepage memory to start VPP: ' + f'available: {available_memory} GB, required: {memory_required} GB. ' + 'Please add kernel memory options for HugePages ' + '"set system option kernel memory hugepage-size ..." and reboot' + ) + + +def verify_vpp_settings_cpu_skip_cores(skip_cores: int): + cpu_cores = total_core_count() + + # The number of skipped cores must not be greater than + # available CPU cores in the system - 1 for main thread + if skip_cores > (cpu_cores - 1): + raise ConfigError( + f'The system does not have enough available CPUs to skip ' + f'(reduce "cpu skip-cores" to {cpu_cores} or less)' + ) + + +def verify_vpp_settings_cpu_and_corelist_workers(settings: dict): + """ + `set vpp settings cpu workers` and `set vpp settings cpu corelist-workers` + are mutually exclusive! + """ + if ( + 'corelist_workers' in settings or 'workers' in settings + ) and 'main_core' not in settings: + raise ConfigError('"cpu main-core" is required but not set!') + + if 'corelist_workers' in settings and 'workers' in settings: + raise ConfigError( + '"cpu corelist-workers" and "cpu workers" cannot be used at the same time!' + ) + + +def verify_vpp_cpu_main_core(cpu_settings: dict) -> None: + """Check that the main core is available""" + skip_cores = int(cpu_settings.get('skip_cores', 0)) + available_cores = cpu_checks.available_cores_list(skip_cores) + main_core = int(cpu_settings['main_core']) + + if main_core not in available_cores: + raise ConfigError( + 'Cannot set main core for VPP process: ' + f'CPU#{main_core} is not available.' + ) + + +def verify_vpp_settings_cpu_workers(cpu_settings: dict) -> int: + """ + Verify that the system has enough available CPU cores + to run a given amount of worker processes (1 worker/core) + """ + workers = int(cpu_settings.get('workers', 0)) + available_cores = cpu_checks.available_cores_count(cpu_settings) + + if workers > available_cores: + raise ConfigError( + f'Not enough free CPU cores for {workers} VPP workers ' + f'(reduce to {available_cores} or less)' + ) + + return workers + + +def verify_vpp_settings_cpu_corelist_workers(cpu_settings: dict) -> int: + """ + Verify that the CPU cores provided to the config are free and can be used by VPP + """ + workers = cpu_settings.get('corelist_workers') + main_core = int(cpu_settings.get('main_core')) + skip_cores = int(cpu_settings.get('skip_cores', 0)) + available_cores = cpu_checks.available_cores_list(skip_cores) + try: + all_core_nums = cpu_checks.worker_cores_list( + iface='cpu corelist', worker_ranges=workers + ) + except ValueError as e: + raise ConfigError(str(e)) + + error_msg = 'Cannot set VPP "cpu corelist-workers"' + + if main_core in all_core_nums: + raise ConfigError( + f'CPU#{main_core} is set as main core and should not ' + 'be included to the corelist-workers' + ) + + invalid_cores = [str(el) for el in all_core_nums if el not in available_cores] + if invalid_cores: + raise ConfigError( + f'{error_msg}: CPU# {",".join(invalid_cores)} are not available.' + ) + + if len(all_core_nums) > cpu_checks.available_cores_count(cpu_settings): + raise ConfigError(f'{error_msg}: Not enough free CPUs in the system.') + + return len(all_core_nums) + + +def verify_vpp_nat44_workers(workers: int, nat44_workers: list): + if workers < 1: + raise ConfigError( + '"nat44 workers" requires cpu workers or corelist-workers to be set!' + ) + try: + nat_workers = cpu_checks.worker_cores_list( + iface='nat44', worker_ranges=nat44_workers + ) + except ValueError as e: + raise ConfigError(str(e)) + + invalid_workers = [str(el) for el in nat_workers if el not in range(workers)] + if invalid_workers: + raise ConfigError( + f'Cannot set VPP "nat44 workers": worker(s) #{",".join(invalid_workers)} not available. ' + f'Available worker ids: {",".join(map(str, range(workers)))}' + ) + + +def verify_vpp_statseg_size(settings: dict): + statseg_size = mem_checks.statseg_size(settings) + + if 'size' in settings.get('statseg'): + if statseg_size < 1 << 20: + raise ConfigError('The statseg size must be greater than or equal to 1M') + + if 'page_size' in settings['statseg']: + statseg_page_size = mem_checks.statseg_page_size(settings) + if statseg_page_size > statseg_size: + readable_statseg_page = bytes_to_human_memory(statseg_page_size, 'K') + raise ConfigError( + f'The statseg size must be greater than or equal to page-size ({readable_statseg_page})' + ) + + +def verify_vpp_interfaces_dpdk_num_queues(qtype: str, num_queues: int, workers: int): + """ + Verify that VPP has enough workers to run the given amount of RX/TX queues + 1 queue per 1 worker is assumed as default + """ + + if num_queues > workers: + raise ConfigError( + f'The number of {qtype} queues cannot be greater than the number of configured VPP workers: ' + f'workers: {workers}, queues: {num_queues}' + ) + + +def verify_vpp_host_resources(config: dict): + max_map_count = int(config['settings']['host_resources']['max_map_count']) + + # Get HugePages total count + hugepages = mem_checks.get_total_hugepages_count() + + if max_map_count < 2 * hugepages: + Warning( + 'The max-map-count should be greater than or equal to (2 * HugePages_Total) ' + 'or VPP could work not properly. Please set up ' + f'"vpp settings host-resources max-map-count" to {2 * hugepages} or higher' + ) diff --git a/python/vyos/vpp/configdb.py b/python/vyos/vpp/configdb.py new file mode 100644 index 000000000..13c1cd3a0 --- /dev/null +++ b/python/vyos/vpp/configdb.py @@ -0,0 +1,206 @@ +# +# Copyright (C) VyOS Inc. +# +# This program is free software; you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation; either version 2 of the License, or +# (at your option) any later version. +# +# 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, write to the Free Software Foundation, Inc., +# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. + +from json import loads as json_loads, dumps as json_dumps +from pathlib import Path +from typing import Any + +STORAGE_LOCATION = '/run/vpp' + + +class JSONStorage: + def __init__(self, name: str = '') -> None: + """Initiate a file storage + + Args: + name (str, optional): Unique storage name. Defaults to '' (generate a name). + + Raises: + err: In case a file for storage cannot be created + """ + # If a name is not provided, this is a temporary one-time storage + # this use case is strange, but let's allow this + if not name: + self.__temporary = True + self.__cache: dict[Any, Any] = {} + self.__locked = False + return + self.__temporary = False + + self.__storage = Path(f'{STORAGE_LOCATION}/{name}.json') + self.__lock_file = Path(f'{STORAGE_LOCATION}/{name}.lock') + + # prepare a folder + storage_dir = Path(STORAGE_LOCATION) + if not storage_dir.exists(): + storage_dir.mkdir(parents=True) + + # initialize lock status + self.__locked = False + if self.__storage_locked(): + raise FileExistsError(f'Cannot open locked storage: {self.__storage}') + self.__lock_file.touch() + + if not self.__storage.exists(): + try: + self.__storage.touch() + except Exception as err: + print(f'Unable to initiate storage: {err}') + raise err + # prepare an empty cache + self.__cache: dict[Any, Any] = {} + else: + # load a cache from file + self.__cache = self.__load_file() + + def __del__(self) -> None: + """Dump data to persistent storage and unlock it""" + if self.__temporary: + return + # dump a cache to storage + if self.__cache: + self.__dump_file() + # or remove a file + else: + self.__storage.unlink() + # unlock a storage + self.__lock_file.unlink() + + def __check_types(self, data: Any) -> None: + """Check if all the data have supported types + + Args: + data (Any): object to validate + + Raises: + TypeError: If a data type is not supported + """ + if isinstance(data, str | int | float | bool | None): + return + if isinstance(data, list): + for item in data: + self.__check_types(item) + return + if isinstance(data, dict): + for item in data.values(): + self.__check_types(item) + return + raise TypeError(f'Object type "{type(data)}" is not allowed') + + def __load_file(self) -> dict[Any, Any]: + """Read a file to a dictionary + + Returns: + dict[Any, Any]: loaded dict object + """ + data: bytes = self.__storage.read_bytes() + return json_loads(data) + + def __dump_file(self) -> None: + """Dump cache to a file""" + data: str = json_dumps(self.__cache) + self.__storage.write_text(data) + + def __lock(self) -> None: + """Lock storage + + Raises: + FileExistsError: Raised if a storage is already locked + """ + if self.__locked: + raise FileExistsError(f'Access is already locked: {self.__storage}') + self.__locked = True + + def __unlock(self) -> None: + """Unlock storage + + Raises: + FileNotFoundError: Raised if a storage is already unlocked + """ + if not self.__locked: + raise FileNotFoundError(f'Access is already unlocked: {self.__storage}') + self.__locked = False + + def __storage_locked(self) -> bool: + """Check if a storage is locked + + Returns: + bool: Lock status + """ + if self.__lock_file.exists(): + return True + return False + + def delete(self, key: Any = None) -> None: + """Delete data from a storage or a full storage + + Raises: + FileExistsError: Raised if a storage is locked + """ + if self.__locked: + raise FileExistsError( + f'Storage locked and delete operation cannot be performed: {self.__storage}' + ) + if key: + if key not in self.__cache: + raise ValueError( + f'Object {key} does not exist in storage {self.__storage}' + ) + del self.__cache[key] + else: + self.__cache = {} + + def write(self, key: Any, value: Any) -> None: + # Check types first + self.__check_types(key) + self.__check_types(value) + # check lock status + if self.__locked: + raise FileExistsError( + f'Storage is locked and cannot be written: {self.__storage}' + ) + # write a data to a cache + self.__lock() + self.__cache[key] = value + self.__unlock() + + def read(self, key: Any, default: Any = None) -> Any: + """Read data from a storage + + Args: + key (Any): key name + default (Any, optional): Value to return if a key does not exist. Defaults to None. + + Raises: + FileExistsError: Raised if a storage is locked + + Returns: + Any: Value to return + """ + # Check types first + self.__check_types(key) + # check lock status + if self.__locked: + raise FileExistsError( + f'Storage is locked and it is not safe to read: {self.__storage}' + ) + # read a data from cache + self.__lock() + data: Any | None = self.__cache.get(key, default) + self.__unlock() + + return data diff --git a/python/vyos/vpp/control_host.py b/python/vyos/vpp/control_host.py new file mode 100644 index 000000000..6a6b9d72f --- /dev/null +++ b/python/vyos/vpp/control_host.py @@ -0,0 +1,375 @@ +# +# Copyright (C) VyOS Inc. +# +# This program is free software; you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation; either version 2 of the License, or +# (at your option) any later version. +# +# 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, write to the Free Software Foundation, Inc., +# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. + + +from pathlib import Path +from re import fullmatch as re_fullmatch +from subprocess import run +from time import sleep + +from pyroute2 import IPRoute + +from vyos.vpp.utils import EthtoolGDrvinfo + + +def pci_rescan(pci_addr: str = '') -> None: + """Rescan PCI device by removing it and rescan PCI bus + + If PCI address is not defined - just rescan PCI bus + + Args: + address (str, optional): PCI address of device. Defaults to ''. + """ + device_file = Path(f'/sys/bus/pci/devices/{pci_addr}/remove') + if pci_addr: + if device_file.exists(): + device_file.write_text('1') + # wait 10 seconds max until device will be removed + attempts = 100 + while device_file.exists() and attempts: + attempts -= 1 + sleep(0.1) + if device_file.exists(): + raise TimeoutError( + f'Timeout was reached for removing PCI device {pci_addr}' + ) + else: + raise FileNotFoundError(f'PCI device {pci_addr} does not exist') + rescan_file = Path('/sys/bus/pci/rescan') + rescan_file.write_text('1') + if pci_addr: + # wait 10 seconds max until device will be installed + attempts = 100 + while not device_file.exists() and attempts: + attempts -= 1 + sleep(0.1) + if not device_file.exists(): + raise TimeoutError( + f'Timeout was reached for installing PCI device {pci_addr}' + ) + + +def unbind_driver(bus_id: str, device_id: str) -> bool: + """Unbind a driver from a device + + Args: + bus_id (str): bus ID (pci, vmbus, etc.) + device_id (str): device id on the bus (PCI address, VMBus UUID) + + Returns: + bool: True if a driver has been unbound, False otherwise + """ + device_resolved: str = ( + Path(f'/sys/bus/{bus_id}/devices/{device_id}').resolve().as_posix() + ) + if not Path(f'{device_resolved}/driver').exists(): + return False + + Path(f'{device_resolved}/driver/unbind').write_text(device_id) + return True + + +def probe_driver(bus_id: str, device_id: str) -> None: + """Probe driver for a device on a bus + + Args: + bus_id (str): bus ID (pci, vmbus, etc.) + device_id (str): device id on the bus (PCI address, VMBus UUID) + """ + Path(f'/sys/bus/{bus_id}/drivers_probe').write_text(device_id) + + +def load_kernel_module(module_name: str) -> None: + """Load a kernel module + + Args: + module_name (str): module name + """ + # check if a module already loaded + if Path(f'/sys/module/{module_name}').exists(): + return + + # execute modprobe with the specified module name + run(['/usr/sbin/modprobe', '-q', module_name], check=True) + + +def override_driver(bus_id: str, device_id: str, driver_name: str = '') -> None: + """Override a driver for a device + + Args: + bus_id (str): bus ID (pci, vmbus, etc.) + device_id (str): device id on the bus (PCI address, VMBus UUID) + driver_name (str, optional): Kernel module (driver) name. Defaults to '' - clear an override. + + Raises: + FileNotFoundError: A device does not support driver override + ChildProcessError: Failed to override a driver + """ + device_resolved: str = ( + Path(f'/sys/bus/{bus_id}/devices/{device_id}').resolve().as_posix() + ) + # check if a device supports driver override + if not Path(f'{device_resolved}/driver_override').exists(): + raise FileNotFoundError(f'{device_resolved} does not support driver override') + + if driver_name: + load_kernel_module(driver_name) + + unbind_driver(bus_id, device_id) + + # vfio-pci needs special approach + if driver_name == 'vfio-pci': + vendor: str = Path(f'{device_resolved}/vendor').read_text() + device: str = Path(f'{device_resolved}/device').read_text() + Path('/sys/module/vfio_pci/drivers/pci:vfio-pci/new_id').write_text( + f'{vendor} {device}' + ) + + # override a driver + Path(f'{device_resolved}/driver_override').write_text(f'{driver_name}\n') + + # probe a driver + probe_driver(bus_id, device_id) + + # check the result + if not Path(f'{device_resolved}/driver').exists(): + raise ChildProcessError( + f'Failed to override a driver to {driver_name} for {bus_id}, {device_id}' + ) + + +def get_bus_name(iface: str) -> str: + """Get bus name + Works for PCI, VMbus, maybe something else. + Does not work for Virtio and other virtual devices + (however, it does not seem we need this for such kind of devices). + + Args: + iface (str): interface name + + Returns: + str: bus name + """ + device_resolved: Path = Path(f'/sys/class/net/{iface}/device').resolve() + + # Iterate upwards until a `bus` directory is found + current_path: Path = device_resolved + while True: + # Check if a bus info is available + subsystem_path = Path(f'{current_path}/subsystem') + if subsystem_path.is_symlink(): + # Read the link to determine the bus type + bus_path = subsystem_path.resolve() + # Check if the parent directory is a 'bus' directory in '/sys/bus/' + if bus_path.parent.name == 'bus': + # Return only the last name of the path, e.g., 'pci' + return bus_path.name + + # Move up one directory level + current_path = current_path.parent + if current_path == Path('/sys'): + break # Stop if we reach the root of /sys without finding a bus type + + return '' # Return None if no bus type was found + + +def get_eth_name(dev_id: str) -> str: + """Find Ethernet interface name by PCI address or UUID + + Args: + dev_id (str): PCI address or UUID + + Raises: + FileNotFoundError: no Ethernet interface was found + + Returns: + str: Ethernet interface name + """ + # find all PCI devices with eth* names + net_devs: dict[str, str] = {} + net_devs_dir = Path('/sys/class/net') + regex_filter = r'^/sys/devices/pci[\w/:\.]+/(?P<pci_addr>\w+:\w+:\w+\.\w+)/[\w/:\.]+/(?P<iface_name>eth\d+)$' + for dir in net_devs_dir.iterdir(): + # PCI devices + real_dir: str = dir.resolve().as_posix() + re_obj = re_fullmatch(regex_filter, real_dir) + if re_obj: + iface_name: str = re_obj.group('iface_name') + iface_addr: str = re_obj.group('pci_addr') + net_devs.update({iface_addr: iface_name}) + # UUID devices + else: + try: + bus_type: str = get_bus_name(dir.name) + iface_addr = EthtoolGDrvinfo(dir.name).bus_info_expand(bus_type) + net_devs.update({iface_addr: dir.name}) + except FileNotFoundError: + pass + + # match to provided PCI address or UUID and return a name if found + if dev_id in net_devs: + return net_devs[dev_id] + # raise error if device was not found + raise FileNotFoundError( + f'A device with ID {dev_id} not found in ethernet interfaces' + ) + + +def get_dev_id(iface: str) -> str: + """Get device ID by its interface name + + Args: + iface (str): interface name + + Raises: + FileNotFoundError: no Ethernet interface was found + + Returns: + str: device ID (PCI address or UUID) + """ + try: + # Try to get details via ethtool first + ethtool_info = EthtoolGDrvinfo(iface) + # For devices represented by UUID we need to expand them + # to their full representation + if ethtool_info.driver == 'hv_netvsc': + return ethtool_info.bus_info_expand('vmbus') + return ethtool_info.bus_info + except Exception: + # raise error if a device ID was not found + raise FileNotFoundError(f'Cannot find device ID for interface {iface}') + + +def get_eth_driver(iface: str) -> str: + """Find kernel module used for Ethernet interface + + Args: + iface (str): Ethernet interface name + + Raises: + FileNotFoundError: no Ethernet interface was found + + Returns: + str: kernel module name + """ + iface_driver: str = '' + driver_dir = Path(f'/sys/class/net/{iface}/device/driver/module') + if not driver_dir.exists(): + # raise error if device was not found + raise FileNotFoundError(f'PCI device {iface} not found in ethernet interfaces') + + iface_driver: str = driver_dir.resolve().name + return iface_driver + + +def unsafe_noiommu_mode(status: bool) -> None: + """Control unsafe_noiommu_mode parameter of vfio module + + Args: + status (bool): Target status + + Raises: + ChildProcessError: Raised if failed to set unsafe_noiommu_mode + """ + param_path = Path('/sys/module/vfio/parameters/enable_unsafe_noiommu_mode') + current_status: str = param_path.read_text().strip() + target_status: str = 'Y' if status else 'N' + if current_status != target_status: + param_path.write_text(target_status) + if param_path.read_text().strip() != target_status: + raise ChildProcessError('Failed to set unsafe_noiommu_mode') + + +def rename_iface(name_old: str, name_new: str) -> None: + """Rename interface + + Args: + name_old (str): old name + name_new (str): new name + """ + run(['ip', 'link', 'set', name_old, 'down']) + rename_cmd: list[str] = ['ip', 'link', 'set', name_old, 'name', name_new] + run(rename_cmd) + + +def set_promisc(iface_name: str, operation: str) -> None: + """Set promisc mode for interface + + Args: + iface_name (str): name of an interface + operation (str): operation (on, off) + """ + run(['ip', 'link', 'set', iface_name, 'promisc', operation]) + + +def set_mtu(iface_name: str, mtu: int) -> None: + """Set MTU for interface + + Args: + iface_name (str): name of an interface + mtu (int): MTU + """ + run(['ip', 'link', 'set', iface_name, 'mtu', str(mtu)]) + + +def get_eth_mac(iface_name: str) -> str: + """Get MAC address of an interface + + Args: + iface_name (str): name of an interface + + Raises: + FileNotFoundError: interface was not found + + Returns: + str: MAC address + """ + dev_addr_path = Path(f'/sys/class/net/{iface_name}/address') + if dev_addr_path.exists(): + return dev_addr_path.read_text().strip() + else: + # raise error if device was not found + raise FileNotFoundError(f'Interface {iface_name} not found') + + +def xdp_remove(iface_name: str) -> None: + """Remove XDP BPF program from an interfce + + Args: + iface_name (str): name of an interface + """ + run(['ip', 'link', 'set', iface_name, 'xdp', 'off']) + + +def set_status(iface_name: str, status: str) -> None: + """Set interface status + + Args: + iface_name (str): name of an interface + status (str): status - "up" or "down" + """ + run(['ip', 'link', 'set', iface_name, status]) + + +def flush_ip(iface_name: str) -> None: + """Flush IP addresses from an interface + + Args: + iface_name (str): name of an interface + """ + iproute = IPRoute() + iproute.flush_addr(label=iface_name) diff --git a/python/vyos/vpp/control_vpp.py b/python/vyos/vpp/control_vpp.py new file mode 100644 index 000000000..596ef4f12 --- /dev/null +++ b/python/vyos/vpp/control_vpp.py @@ -0,0 +1,488 @@ +# +# Copyright (C) VyOS Inc. +# +# This program is free software; you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation; either version 2 of the License, or +# (at your option) any later version. +# +# 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, write to the Free Software Foundation, Inc., +# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. + +from collections.abc import Callable +from functools import wraps +from re import search as re_search, MULTILINE as re_M +from systemd import journal +from time import sleep +from typing import TypeVar, ParamSpec, Literal + +from vpp_papi import VPPApiClient +from vpp_papi import VPPIOError, VPPValueError + +# define types for static type checkers +AnyType = TypeVar('AnyType') +AnyParam = ParamSpec('AnyParam') + + +class VPPControl: + """Control VPP network stack""" + + class _Decorators: + """Decorators for VPPControl""" + + @classmethod + def api_call( + cls, decorated_func: Callable[AnyParam, AnyType] + ) -> Callable[AnyParam, AnyType]: + """Check if API is connected before API call + + Args: + decorated_func: function to decorate + + Raises: + VPPIOError: Connection to API is not established + """ + + @wraps(decorated_func) + def api_safe_wrapper( + cls, *args: AnyParam.args, **kwargs: AnyParam.kwargs + ) -> AnyType: + if not cls.connected: + raise VPPIOError(2, 'VPP API is not connected') + return decorated_func(cls, *args, **kwargs) + + return api_safe_wrapper + + @classmethod + def check_retval( + cls, decorated_func: Callable[AnyParam, AnyType] + ) -> Callable[AnyParam, AnyType]: + """Check retval from API response + + Args: + decorated_func: function to decorate + + Raises: + VPPValueError: raised when retval is not 0 + """ + + @wraps(decorated_func) + def check_retval_wrapper( + cls, *args: AnyParam.args, **kwargs: AnyParam.kwargs + ) -> AnyType: + return_value = decorated_func(cls, *args, **kwargs) + if not return_value.retval == 0: + raise VPPValueError(f'VPP API call failed: {return_value.retval}') + return return_value + + return check_retval_wrapper + + def __init__(self, attempts: int = 5, interval: int = 1000) -> None: + """Create VPP API connection + + Args: + attempts (int, optional): attempts to connect. Defaults to 5. + interval (int, optional): interval between attempts in ms. Defaults to 1000. + + Raises: + VPPIOError: Connection to API cannot be established + """ + self.__vpp_api_client = VPPApiClient() + # connect with interval + while attempts: + try: + attempts -= 1 + self.__vpp_api_client.connect('vpp-vyos') + break + except (ConnectionRefusedError, FileNotFoundError) as err: + error_message = f'VPP API connection timeout: {err}' + journal.send(error_message, priority=journal.LOG_ERR) + sleep(interval / 1000) + # raise exception if connection was not successful in the end + if not self.__vpp_api_client.transport.connected: + raise VPPIOError(2, 'Cannot connect to VPP API') + + def __del__(self) -> None: + """Disconnect from VPP API (destructor)""" + self.disconnect() + + def disconnect(self) -> None: + """Disconnect from VPP API""" + if self.__vpp_api_client.transport.connected: + self.__vpp_api_client.disconnect() + + @_Decorators.check_retval + @_Decorators.api_call + def cli_cmd(self, command: str): + """Send raw CLI command + + Args: + command (str): command to send + + Returns: + vpp_papi.vpp_serializer.cli_inband_reply: CLI reply class + """ + return self.__vpp_api_client.api.cli_inband(cmd=command) + + @_Decorators.api_call + def get_mac(self, ifname: str) -> str: + """Find MAC address by interface name in VPP + + Args: + ifname (str): interface name inside VPP + + Returns: + str: MAC address + """ + for iface in self.__vpp_api_client.api.sw_interface_dump(): + if iface.interface_name == ifname: + return iface.l2_address.mac_string + return '' + + @_Decorators.api_call + def get_sw_if_index(self, ifname: str) -> int | None: + """Find interface index by interface name in VPP + + Args: + ifname (str): interface name inside VPP + + Returns: + int | None: Interface index or None (if was not fount) + """ + for iface in self.__vpp_api_client.api.sw_interface_dump(): + if iface.interface_name == ifname: + return iface.sw_if_index + return None + + @_Decorators.api_call + def get_interface_name(self, index: int) -> str | None: + """Find interface name by interface index in VPP + + Args: + index (int): interface index inside VPP + + Returns: + str | None: Interface name or None (if was not found) + """ + for iface in self.__vpp_api_client.api.sw_interface_dump(): + if iface.sw_if_index == index: + return iface.interface_name + return None + + @_Decorators.check_retval + @_Decorators.api_call + def lcp_pair_add( + self, + iface_name_vpp: str, + iface_name_kernel: str, + iface_type: Literal['tun', 'tap', ''] = '', + ) -> None: + """Create LCP interface pair between VPP and kernel + + Args: + iface_name_vpp (str): interface name in VPP + iface_name_kernel (str): interface name in kernel + iface_type (Literal['tun', 'tap', ''], optional): Use explicit interface type in kernel. Defaults to ''. + """ + iface_index = self.get_sw_if_index(iface_name_vpp) + if iface_index: + api_call_args: dict[str, bool | int | str] = { + 'is_add': True, + 'sw_if_index': iface_index, + 'host_if_name': iface_name_kernel, + } + if iface_type: + iface_type_resolve = {'tun': 1, 'tap': 0} + api_call_args['host_if_type'] = iface_type_resolve[iface_type] + return self.__vpp_api_client.api.lcp_itf_pair_add_del_v2(**api_call_args) + + @_Decorators.check_retval + @_Decorators.api_call + def lcp_pair_del(self, iface_name_vpp: str, iface_name_kernel: str) -> None: + """Delete LCP interface pair between VPP and kernel + + Args: + iface_name_vpp (str): interface name in VPP + iface_name_kernel (str): interface name in kernel + """ + iface_index = self.get_sw_if_index(iface_name_vpp) + if iface_index: + return self.__vpp_api_client.api.lcp_itf_pair_add_del_v2( + is_add=False, sw_if_index=iface_index, host_if_name=iface_name_kernel + ) + + @_Decorators.api_call + def lcp_pair_find( + self, + kernel_name: str = '', + vpp_index_hw: int | None = None, + vpp_index_kernel: int | None = None, + vpp_name_hw: str = '', + vpp_name_kernel: str = '', + ) -> dict[str, str | int] | None: + """Find LCP pair details + + Args: + kernel_name (str, optional): Interface name in the kernel. Defaults to ''. + vpp_index_hw (int | None, optional): Interface index in VPP (hardware). Defaults to None. + vpp_index_kernel (int | None, optional): Interface index in VPP (kernel). Defaults to None. + vpp_name_hw (str, optional): Interface name in VPP (hardware). Defaults to ''. + vpp_name_kernel (str, optional): Interface name in VPP (to kernel). Defaults to ''. + + Returns: + dict[str, str | int] | None: LCP pair details + """ + filter_dict = {} + for filter_name, filter_value in locals().items(): + if filter_value: + filter_dict[filter_name] = filter_value + + # Get list of pairs + lcp_pairs = self.lcp_pairs_list() + + # Check each pair + for pair in lcp_pairs: + pair_found = False + # For each item provided in function arguments + for filter_name, filter_value in filter_dict.items(): + # Stop if filter value is not as in a current pair + if filter_name in pair and pair[filter_name] != filter_value: + pair_found = False + break + # Set flag to True and check the next filter value + pair_found = True + + if pair_found: + return pair + + return None + + @_Decorators.api_call + def lcp_pairs_list(self) -> list[dict[str, str | int]]: + """List all LCP pairs + + Returns: + list[dict[str, str | int]]: LCP pairs details + """ + lcp_pairs_details = [] + + lcp_pairs = self.__vpp_api_client.api.lcp_itf_pair_get()[1] + vpp_ifaces = self.__vpp_api_client.api.sw_interface_dump() + for pair in lcp_pairs: + pair_details = { + 'kernel_name': pair.host_if_name, + 'vpp_index_hw': pair.phy_sw_if_index, + 'vpp_index_kernel': pair.host_sw_if_index, + } + for vpp_iface in vpp_ifaces: + if vpp_iface.sw_if_index == pair_details['vpp_index_hw']: + pair_details['vpp_name_hw'] = vpp_iface.interface_name + if vpp_iface.sw_if_index == pair_details['vpp_index_kernel']: + pair_details['vpp_name_kernel'] = vpp_iface.interface_name + + lcp_pairs_details.append(pair_details) + + return lcp_pairs_details + + @_Decorators.check_retval + @_Decorators.api_call + def lcp_resync(self) -> None: + """Resynchronize objects between kernel and VPP via Netlink + + This clears all routes in VPP configured by LCP and re-creates them + based on the current state of the kernel. + """ + return self.__vpp_api_client.api.lcp_nl_resync() + + @_Decorators.check_retval + @_Decorators.api_call + def iface_rxmode(self, iface_name: str, rx_mode: str) -> None: + """Set interface rx-mode in VPP + + Args: + iface_name (str): interface name in VPP + rx_mode (str): mode (polling, interrupt, adaptive) + """ + modes_dict: dict[str, int] = {'polling': 1, 'interrupt': 2, 'adaptive': 3} + if rx_mode not in modes_dict: + raise VPPValueError(f'Mode {rx_mode} is not known') + iface_index = self.get_sw_if_index(iface_name) + return self.__vpp_api_client.api.sw_interface_set_rx_mode( + sw_if_index=iface_index, mode=modes_dict[rx_mode] + ) + + @_Decorators.api_call + def get_pci_addr(self, ifname: str) -> str: + """Find PCI address of interface by interface name in VPP + + Args: + ifname (str): interface name inside VPP + + Returns: + str: PCI address + """ + hw_info = self.cli_cmd(f'show hardware-interfaces {ifname}').reply + + regex_filter = r'^\s+pci: device (?P<device>\w+:\w+) subsystem (?P<subsystem>\w+:\w+) address (?P<address>\w+:\w+:\w+\.\w+) numa (?P<numa>\w+)$' + re_obj = re_search(regex_filter, hw_info, re_M) + + # return empty string if no interface or no PCI info was found + if not hw_info or not re_obj: + return '' + + address = re_obj.groupdict().get('address', '') + + # we need to modify address to match kernel style + # for example: 0000:06:14.00 -> 0000:06:14.0 + address_chunks: list[str] = address.split('.') + address_normalized: str = f'{address_chunks[0]}.{int(address_chunks[1])}' + + return address_normalized + + @_Decorators.check_retval + @_Decorators.api_call + def xdp_iface_create( + self, + host_if: str, + name: str, + rxq_num: int = 0, + rxq_size: int = 0, + txq_size: int = 0, + mode: Literal['auto', 'copy', 'zero-copy'] = 'auto', + flags: Literal['no_syscall_lock', ''] = '', + ) -> None: + """Create XDP interface + + Args: + host_if (str): name of an interface in kernel + name (str): name of an interface in VPP + rxq_num (int, optional): Number of receive queues to connect to. Defaults to 0 (all). + rxq_size (int, optional): Size of receive queue. Defaults to 0. + txq_size (int, optional): Size of tranceive queue. Defaults to 0. + mode (Literal['auto', 'copy', 'zero-copy', optional): Zero-copy mode. Defaults to 'auto'. + flags (Literal['no_syscall_lock', ''], optional): Syscall lock mode. Defaults to ''. + """ + api_call_args: dict[str, int | str] = { + 'host_if': host_if, + 'name': name, + 'rxq_num': rxq_num, + 'rxq_size': rxq_size, + 'txq_size': txq_size, + } + if mode != 'auto': + mode_resolve: dict[str, int] = {'auto': 0, 'copy': 1, 'zero-copy': 2} + api_call_args['mode'] = mode_resolve[mode] + if flags == 'no_systcall_lock': + api_call_args['flags'] = 1 + return self.__vpp_api_client.api.af_xdp_create_v3(**api_call_args) + + @_Decorators.check_retval + @_Decorators.api_call + def xdp_iface_delete(self, iface_name_vpp: str) -> None: + """Delete XDP interface + + Args: + iface_name_vpp (str): Name of an interface in VPP + """ + iface_index = self.get_sw_if_index(iface_name_vpp) + if iface_index: + api_call_args: dict[str, int] = {'sw_if_index': iface_index} + return self.__vpp_api_client.api.af_xdp_delete(**api_call_args) + + @_Decorators.check_retval + @_Decorators.api_call + def set_iface_mac(self, iface_name_vpp: str, mac_address: str) -> None: + """Set MAC address of an interface + + Args: + iface_name_vpp (str): Name of an interface in VPP + mac_address (str): MAC address + """ + iface_index = self.get_sw_if_index(iface_name_vpp) + api_call_args: dict[str, str | int] = { + 'sw_if_index': iface_index, + 'mac_address': mac_address, + } + return self.__vpp_api_client.api.sw_interface_set_mac_address(**api_call_args) + + @_Decorators.check_retval + @_Decorators.api_call + def set_iface_mtu(self, iface_name_vpp: str, mtu: int) -> None: + """Set MTU for interface + + Args: + iface_name_vpp (str): Name of an interface in VPP + mtu (int): MTU + """ + iface_index = self.get_sw_if_index(iface_name_vpp) + api_call_args: dict[str, str | int] = {'sw_if_index': iface_index, 'mtu': mtu} + return self.__vpp_api_client.api.hw_interface_set_mtu(**api_call_args) + + @_Decorators.api_call + def get_sw_if_dev_type(self, ifname: str) -> int | None: + """Find interface device type by interface name in VPP + + Args: + ifname (str): interface name inside VPP + + Returns: + int | None: Interface device type or None (if was not fount) + """ + for iface in self.__vpp_api_client.api.sw_interface_dump(): + if iface.interface_name == ifname: + return iface.interface_dev_type + return None + + @_Decorators.api_call + def enable_disable_nat44_forwarding(self, enable: bool) -> None: + """Enable/disable NAT44 forwarding + + Args: + enable (bool): True if enable, False if disable + """ + self.__vpp_api_client.api.nat44_forwarding_enable_disable(enable=enable) + + @_Decorators.api_call + def set_nat44_session_limit(self, session_limit: int) -> None: + """Set NAT44 session limit + + Args: + session_limit (int): Maximum number of sessions per thread + """ + self.__vpp_api_client.api.nat44_set_session_limit( + session_limit=session_limit, + ) + + @_Decorators.api_call + def set_nat_workers(self, workers: int) -> None: + """Set NAT44 session limit + + Args: + workers (int): Bitmask of workers list + """ + self.__vpp_api_client.api.nat_set_workers(worker_mask=workers) + + @property + def connected(self) -> bool: + """Check if VPP API is connected + + Returns: + bool: True if connected, False if not + """ + return self.__vpp_api_client.transport.connected + + @property + @_Decorators.api_call + def api(self): + """Call API + + Returns: + Callable[AnyParam, AnyType]: API functions + """ + return self.__vpp_api_client.api diff --git a/python/vyos/vpp/interface/__init__.py b/python/vyos/vpp/interface/__init__.py new file mode 100644 index 000000000..903b7cbfc --- /dev/null +++ b/python/vyos/vpp/interface/__init__.py @@ -0,0 +1,42 @@ +# +# Copyright (C) VyOS Inc. +# +# This program is free software; you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation; either version 2 of the License, or +# (at your option) any later version. +# +# 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, write to the Free Software Foundation, Inc., +# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. + +from .bond import BondInterface +from .bridge import BridgeInterface +from .ethernet import EthernetInterface +from .geneve import GeneveInterface +from .gre import GREInterface +from .interface import Interface +from .ipip import IPIPInterface +from .loopback import LoopbackInterface +from .vxlan import VXLANInterface +from .wireguard import WireguardInterface +from .xconnect import XconnectInterface + +__all__ = [ + 'BondInterface', + 'BridgeInterface', + 'EthernetInterface', + 'GeneveInterface', + 'GREInterface', + 'Interface', + 'IPIPInterface', + 'LoopbackInterface', + 'VXLANInterface', + 'WireguardInterface', + 'XconnectInterface', +] diff --git a/python/vyos/vpp/interface/bond.py b/python/vyos/vpp/interface/bond.py new file mode 100644 index 000000000..604ba2ae5 --- /dev/null +++ b/python/vyos/vpp/interface/bond.py @@ -0,0 +1,118 @@ +# +# Copyright (C) VyOS Inc. +# +# This program is free software; you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation; either version 2 of the License, or +# (at your option) any later version. +# +# 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, write to the Free Software Foundation, Inc., +# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. + +from vyos.vpp.control_host import set_promisc +from vyos.vpp.interface.interface import Interface + + +class BondInterface(Interface): + def __init__( + self, + ifname, + mode: str = '', + load_balance: int = 0, + mac: str = '', + kernel_interface: str = '', + state: str = 'up', + ): + super().__init__(ifname) + self.instance = int(ifname.removeprefix('bond')) + self.ifname = f'BondEthernet{self.instance}' + self.mode = mode + self.load_balance = load_balance + self.mac = mac + self.kernel_interface = kernel_interface + self.state = state + + def add(self): + """Create Bond interface + https://github.com/FDio/vpp/blob/stable/2306/src/vnet/bonding/bond.api + Example: + from vyos.vpp.interface import BondInterface + a = BondInterface(ifname='bond0', mode=5) + a.add() + """ + # Create interface 'bondX' + create_args = { + 'id': self.instance, + 'mode': self.mode, + 'lb': self.load_balance, + } + if self.mac: + create_args.update({'use_custom_mac': True, 'mac_address': self.mac}) + self.vpp.api.bond_create2(**create_args) + if self.kernel_interface: + self.vpp.lcp_pair_add(self.ifname, self.kernel_interface) + # Set interface state + self.set_state(self.state) + + def delete(self): + """Delete Bond interface + Example: + from vyos.vpp.interface import BondInterface + a = BondInterface(ifname='bond0') + a.delete() + """ + bond_if_index = self.vpp.get_sw_if_index(self.ifname) + self.vpp.api.bond_delete(sw_if_index=bond_if_index) + + def add_member(self, interface): + """Add member to Bond interface + Example: + from vyos.vpp.interface import BondInterface + a = BondInterface(ifname='bond0') + a.add_member(interface='eth0') + """ + bond_if_index = self.vpp.get_sw_if_index(f'BondEthernet{self.instance}') + member_if_index = self.vpp.get_sw_if_index(interface) + member_if_type = self.vpp.get_sw_if_dev_type(interface) + self.vpp.api.bond_add_member( + bond_sw_if_index=bond_if_index, sw_if_index=member_if_index + ) + self.vpp.api.sw_interface_set_promisc( + sw_if_index=member_if_index, promisc_on=True + ) + if member_if_type == 'AF_XDP interface': + set_promisc(f'defunct_{interface}', 'on') + + def detach_member(self, interface): + """Detach member from Bond interface + Example: + from vyos.vpp.interface import BondInterface + a = BondInterface(ifname='bond0') + a.detach_member(interface='eth0') + """ + member_if_index = self.vpp.get_sw_if_index(interface) + self.vpp.api.bond_detach_member(sw_if_index=member_if_index) + + def kernel_add(self): + """Add LCP pair + Example: + from vyos.vpp.interface import BondInterface + a = BondInterface(ifname='bond0', mode=5) + a.kernel_add() + """ + self.vpp.lcp_pair_add(self.ifname, self.kernel_interface) + + def kernel_delete(self): + """Delete LCP pair + Example: + from vyos.vpp.interface import BondInterface + a = BondInterface(ifname='bond0', mode=5) + a.kernel_delete() + """ + self.vpp.lcp_pair_del(self.ifname, self.kernel_interface) diff --git a/python/vyos/vpp/interface/bridge.py b/python/vyos/vpp/interface/bridge.py new file mode 100644 index 000000000..25f0b4660 --- /dev/null +++ b/python/vyos/vpp/interface/bridge.py @@ -0,0 +1,128 @@ +# +# Copyright (C) VyOS Inc. +# +# This program is free software; you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation; either version 2 of the License, or +# (at your option) any later version. +# +# 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, write to the Free Software Foundation, Inc., +# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. + +from vyos.vpp import VPPControl + + +class BridgeInterface: + def __init__( + self, + ifname: str, + flood: bool = True, + forward: bool = True, + learn: bool = True, + uu_flood: bool = True, + arp_term: bool = False, + ): + self.ifname = ifname + self.interface_suffix = int(self.ifname.replace('br', '')) + self.flood = flood + self.forward = forward + self.learn = learn + self.uu_flood = uu_flood + self.arp_term = arp_term + self.vpp = VPPControl() + + def add(self): + """Create Bridge interface + https://github.com/FDio/vpp/blob/stable/2306/src/vnet/l2/l2.api + + Bridge-domain 0 is reserved for the default bridge-domain. + + Example: + from vyos.vpp.interface import BridgeInterface + a = BridgeInterface(ifname='br23') + a.add() + """ + self.vpp.api.bridge_domain_add_del_v2( + is_add=True, + bd_id=self.interface_suffix, + flood=self.flood, + forward=self.forward, + learn=self.learn, + uu_flood=self.uu_flood, + arp_term=self.arp_term, + ) + + def delete(self): + """Delete Bridge interface + + Bridge-members must be detached before deleting the bridge interface. + + Example: + from vyos.vpp.interface import BridgeInterface + a = BridgeInterface(ifname='br23') + a.delete() + """ + self.vpp.api.bridge_domain_add_del_v2(is_add=False, bd_id=self.interface_suffix) + + def add_member(self, member: str | int, port_type: int = 0): + """Add member to Bridge interface + + Attaches a VPP interface to the Bridge interface specified by `interface_suffix`. + The `member` parameter can be either the name (str) or the index (int) of the network + VPP interface to be added as a member to the bridge. + + Args: + member (str or int): The name or index of the VPP network interface + to be added as a member to the bridge. + port_type: 0 - Normal port, 1 - BVI port + + Example: + from vyos.vpp.interface import BridgeInterface + a = BridgeInterface(ifname='br23') + a.add_member(member='eth0') + """ + bridge_index = self.interface_suffix + # If the 'member' is an Integer or digit, assume it's an interface index + if isinstance(member, int): + member_if_index = member + elif member.isdigit(): + member_if_index = int(member) + else: + member_if_index = self.vpp.get_sw_if_index(member) + + return self.vpp.api.sw_interface_set_l2_bridge( + rx_sw_if_index=member_if_index, bd_id=bridge_index, port_type=port_type + ) + + def detach_member(self, member: str | int): + """Detach member from Bridge interface. + Bridge-domain 0 is reserved for the default bridge-domain. + The `member` parameter can be either the name (str) or the index (int) + of the network VPP interface + + Args: + member (str or int): The name or index of the VPP network interface + to be detached from the bridge. + + Example: + from vyos.vpp.interface import BridgeInterface + a = BridgeInterface(ifname='br23') + a.detach_member(member='eth0') + """ + # If the 'member' is an Integer or digit, assume it's an interface index + if isinstance(member, int): + member_if_index = member + elif member.isdigit(): + member_if_index = int(member) + else: + member_if_index = self.vpp.get_sw_if_index(member) + + return self.vpp.api.sw_interface_set_l2_bridge( + rx_sw_if_index=member_if_index, bd_id=0, port_type=0 + ) diff --git a/python/vyos/vpp/interface/ethernet.py b/python/vyos/vpp/interface/ethernet.py new file mode 100644 index 000000000..523211db8 --- /dev/null +++ b/python/vyos/vpp/interface/ethernet.py @@ -0,0 +1,53 @@ +# VyOS implementation of VPP Ethernet interface +# +# Copyright (C) VyOS Inc. +# +# This program is free software; you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation; either version 2 of the License, or +# (at your option) any later version. +# +# 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, write to the Free Software Foundation, Inc., +# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. + +from vyos.vpp import VPPControl + + +class EthernetInterface: + """Interface Ethernet""" + + def __init__(self, ifname, kernel_interface: str = ''): + self.instance = int(ifname.removeprefix('eth')) + self.ifname = ifname + self.kernel_interface = kernel_interface + self.vpp = VPPControl() + + def add(self): + pass + + def delete(self): + pass + + def kernel_add(self): + """Add LCP pair + Example: + from vyos.vpp.interface import EthernetInterface + a = EthernetInterface(ifname='eth0') + a.kernel_add() + """ + self.vpp.lcp_pair_add(self.ifname, self.kernel_interface) + + def kernel_delete(self): + """Delete LCP pair + Example: + from vyos.vpp.interface import EthernetInterface + a = EthernetInterface(ifname='eth0') + a.kernel_delete() + """ + self.vpp.lcp_pair_del(self.ifname, self.kernel_interface) diff --git a/python/vyos/vpp/interface/geneve.py b/python/vyos/vpp/interface/geneve.py new file mode 100644 index 000000000..2d6347f12 --- /dev/null +++ b/python/vyos/vpp/interface/geneve.py @@ -0,0 +1,90 @@ +# VyOS implementation of Geneve interface +# +# Copyright (C) VyOS Inc. +# +# This program is free software; you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation; either version 2 of the License, or +# (at your option) any later version. +# +# 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, write to the Free Software Foundation, Inc., +# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. + +from vyos.vpp import VPPControl + + +def show(): + """Show Geneve interface + Example: + from vyos.vpp.interface import geneve + geneve.show() + """ + vpp = VPPControl() + return vpp.api.geneve_tunnel_dump() + + +class GeneveInterface: + def __init__(self, ifname, source_address, remote, vni, kernel_interface: str = ''): + self.instance = int(ifname.removeprefix('geneve')) + self.ifname = f'geneve_tunnel{self.instance}' + self.src_address = source_address + self.dst_address = remote + self.vni = vni + self.kernel_interface = kernel_interface + self.vpp = VPPControl() + + def add(self): + """Create Geneve interface + https://github.com/FDio/vpp/blob/stable/2306/src/plugins/geneve/geneve.api + + Example: + from vyos.vpp.interface import GeneveInterface + a = GeneveInterface(ifname='geneve25', source_address='192.0.2.1', remote='203.0.113.25', vni=25) + a.add() + """ + return self.vpp.api.geneve_add_del_tunnel2( + is_add=True, + local_address=self.src_address, + remote_address=self.dst_address, + vni=self.vni, + l3_mode=False, + ) + + def delete(self): + """Delete Geneve interface + Example: + from vyos.vpp.interface import GeneveInterface + a = GeneveInterface(ifname='vxlan25', source_address='192.0.2.1', remote='203.0.113.25', vni=25) + a.delete() + """ + return self.vpp.api.geneve_add_del_tunnel2( + is_add=False, + local_address=self.src_address, + remote_address=self.dst_address, + vni=self.vni, + l3_mode=False, + ) + + def kernel_add(self): + """Add LCP pair + Example: + from vyos.vpp.interface import GeneveInterface + a = GeneveInterface(ifname='vxlan25', source_address='192.0.2.1', remote='203.0.113.25', vni=25) + a.kernel_add() + """ + self.vpp.lcp_pair_add(self.ifname, self.kernel_interface, 'tun') + + def kernel_delete(self): + """Delete LCP pair + Example: + from vyos.vpp.interface import GeneveInterface + a = GeneveInterface(ifname='vxlan25', source_address='192.0.2.1', remote='203.0.113.25', vni=25) + a.kernel_delete() + """ + self.vpp.lcp_pair_del(self.ifname, self.kernel_interface) diff --git a/python/vyos/vpp/interface/gre.py b/python/vyos/vpp/interface/gre.py new file mode 100644 index 000000000..bc278d2a0 --- /dev/null +++ b/python/vyos/vpp/interface/gre.py @@ -0,0 +1,140 @@ +# VyOS implementation of VPP GRE interface +# +# Copyright (C) VyOS Inc. +# +# This program is free software; you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation; either version 2 of the License, or +# (at your option) any later version. +# +# 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, write to the Free Software Foundation, Inc., +# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. + +from vyos.vpp import VPPControl +from vyos.vpp.interface.interface import Interface + + +def show(): + """Show GRE interface + Example: + from vyos.vpp.interface import gre + gre.show() + """ + vpp = VPPControl() + return vpp.api.gre_tunnel_dump() + + +class GREInterface(Interface): + """ + Class representing a GRE (Generic Routing Encapsulation) interface. + + Attributes: + ifname (str): The interface name. + source_address (str): The source IP address for the GRE tunnel. + remote (str): The remote IP address for the GRE tunnel. + tunnel_type (str): The type of GRE tunnel. Defaults to 'l3'. + mode (str): The mode of the GRE tunnel. Options are 'point-to-point' and 'point-to-multipoint'. Defaults to 'point-to-point'. + kernel_interface (str): The associated kernel interface. Defaults to an empty string. + instance (int): The instance number derived from the interface name. + vpp (VPPControl): An instance of the VPPControl class for interacting with the VPP API. + """ + + # Mapping of tunnel types https://github.com/FDio/vpp/blob/stable/2406/src/plugins/gre/gre.api#L25-L35 + TUNNEL_TYPE_MAP = { + 'l3': 0, + 'teb': 1, + 'erspan': 2, + } + + MODE_MAP = { + 'point-to-point': 0, + 'point-to-multipoint': 1, + } + + def __init__( + self, + ifname, + source_address, + remote, + tunnel_type: str = 'l3', + mode: str = 'point-to-point', + kernel_interface: str = '', + state: str = 'up', + ): + """ + Initialize a GREInterface instance. + + Args: + ifname (str): The interface name. + source_address (str): The source IP address for the GRE tunnel. + remote (str): The remote IP address for the GRE tunnel. + mode (str): The mode of the GRE tunnel. Options are 'point-to-point' and 'point-to-multipoint'. Defaults to 'point-to-point'. + tunnel_type (str): The type of GRE tunnel. Defaults to 'l3'. + kernel_interface (str): The associated kernel interface. Defaults to an empty string. + state (str): The state of the interface. Defaults to 'up'. + """ + super().__init__(ifname) + self.instance = int(ifname.removeprefix('gre')) + self.ifname = ifname + self.src_address = source_address + self.dst_address = remote + self.tunnel_type = self.TUNNEL_TYPE_MAP[tunnel_type] + self.mode = self.MODE_MAP[mode] + self.kernel_interface = kernel_interface + self.initial_state = state + + def add(self): + """Create GRE interface + https://github.com/FDio/vpp/blob/stable/2406/src/plugins/gre/gre.api + Example: + from vyos.vpp.interface import GREInterface + a = GREInterface(ifname='gre0', source_address='192.0.2.1', remote='203.0.113.25', tunnel_type='l3') + a.add() + """ + self.vpp.api.gre_tunnel_add_del( + is_add=True, + tunnel={ + 'src': self.src_address, + 'dst': self.dst_address, + 'instance': self.instance, + 'mode': self.mode, + 'type': self.tunnel_type, + }, + ) + # Set interface state + self.set_state(self.initial_state) + + def delete(self): + """Delete GRE interface + Example: + from vyos.vpp.interface import GREInterface + a = GREInterface(ifname='gre0', source_address='192.0.2.1', remote='203.0.113.25') + a.delete() + """ + return self.vpp.api.gre_tunnel_add_del( + is_add=False, tunnel={'src': self.src_address, 'dst': self.dst_address} + ) + + def kernel_add(self): + """Add LCP pair + Example: + from vyos.vpp.interface import GREInterface + a = GREInterface(ifname='gre0', source_address='192.0.2.1', remote='203.0.113.25') + a.kernel_add() + """ + self.vpp.lcp_pair_add(self.ifname, self.kernel_interface, 'tun') + + def kernel_delete(self): + """Delete LCP pair + Example: + from vyos.vpp.interface import GREInterface + a = GREInterface(ifname='gre0', source_address='192.0.2.1', remote='203.0.113.25') + a.kernel_delete() + """ + self.vpp.lcp_pair_del(self.ifname, self.kernel_interface) diff --git a/python/vyos/vpp/interface/interface.py b/python/vyos/vpp/interface/interface.py new file mode 100644 index 000000000..05c1d14d6 --- /dev/null +++ b/python/vyos/vpp/interface/interface.py @@ -0,0 +1,49 @@ +# +# Copyright (C) VyOS Inc. +# +# This program is free software; you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation; either version 2 of the License, or +# (at your option) any later version. +# +# 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, write to the Free Software Foundation, Inc., +# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. + +from vyos.vpp import VPPControl + + +class Interface: + def __init__(self, ifname): + self.ifname = ifname + self.vpp = VPPControl() + + def set_state(self, state: str): + """Set interface state to UP or DOWN + Args: + state (str): The state of the interface. Options are 'up' and 'down'. + Example: + from vyos.vpp.interface import Interface + a = Interface(ifname='eth0') + a.set_state(state='up') + """ + if state not in ['up', 'down']: + raise ValueError(f"Invalid state: {state}") + state_flag = 1 if state == 'up' else 0 + if_index = self.vpp.get_sw_if_index(self.ifname) + self.vpp.api.sw_interface_set_flags(sw_if_index=if_index, flags=state_flag) + + def get_state(self): + """Get interface state + Example: + from vyos.vpp.interface import Interface + a = Interface(ifname='eth0') + a.get_state() + """ + if_index = self.vpp.get_sw_if_index(self.ifname) + return self.vpp.api.sw_interface_dump(sw_if_index=if_index)[0]['flags'] diff --git a/python/vyos/vpp/interface/ipip.py b/python/vyos/vpp/interface/ipip.py new file mode 100644 index 000000000..e6ed44537 --- /dev/null +++ b/python/vyos/vpp/interface/ipip.py @@ -0,0 +1,94 @@ +# VyOS implementation of VPP IPIP interface +# +# Copyright (C) VyOS Inc. +# +# This program is free software; you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation; either version 2 of the License, or +# (at your option) any later version. +# +# 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, write to the Free Software Foundation, Inc., +# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. + +from vyos.vpp import VPPControl +from vyos.vpp.interface.interface import Interface + + +def show(): + """Show IPIP interface + Example: + from vyos.vpp.interface import ipip + ipip.show() + """ + vpp = VPPControl() + return vpp.api.ipip_tunnel_dump() + + +class IPIPInterface(Interface): + def __init__( + self, + ifname, + source_address, + remote, + kernel_interface: str = '', + state: str = 'up', + ): + super().__init__(ifname) + self.instance = int(ifname.removeprefix('ipip')) + self.ifname = ifname + self.src_address = source_address + self.dst_address = remote + self.kernel_interface = kernel_interface + self.initial_state = state + + def add(self): + """Create IPIP interface + https://github.com/FDio/vpp/blob/stable/2310/src/vnet/ipip/ipip.api + Example: + from vyos.vpp.interface import IPIPInterface + a = IPIPInterface(ifname='ipip0', source_address='192.0.2.1', remote='192.0.2.5') + a.add() + """ + self.vpp.api.ipip_add_tunnel( + tunnel={ + 'src': self.src_address, + 'dst': self.dst_address, + 'instance': self.instance, + }, + ) + # Set interface state + self.set_state(self.initial_state) + + def delete(self): + """Delete IPIP interface + Example: + from vyos.vpp.interface import IPIPInterface + a = IPIPInterface(ifname='ipip0', source_address='192.0.2.1', remote='192.0.2.5') + a.delete() + """ + ipip_if_index = self.vpp.get_sw_if_index(f'ipip{self.instance}') + return self.vpp.api.ipip_del_tunnel(sw_if_index=ipip_if_index) + + def kernel_add(self): + """Add LCP pair + Example: + from vyos.vpp.interface import IPIPInterface + a = IPIPInterface(ifname='ipip0', source_address='192.0.2.1', remote='192.0.2.5') + a.kernel_add() + """ + self.vpp.lcp_pair_add(self.ifname, self.kernel_interface, 'tun') + + def kernel_delete(self): + """Delete LCP pair + Example: + from vyos.vpp.interface import IPIPInterface + a = IPIPInterface(ifname='ipip0', source_address='192.0.2.1', remote='192.0.2.5') + a.kernel_delete() + """ + self.vpp.lcp_pair_del(self.ifname, self.kernel_interface) diff --git a/python/vyos/vpp/interface/loopback.py b/python/vyos/vpp/interface/loopback.py new file mode 100644 index 000000000..357423780 --- /dev/null +++ b/python/vyos/vpp/interface/loopback.py @@ -0,0 +1,72 @@ +# VyOS implementation of VPP Loopback interface +# +# Copyright (C) VyOS Inc. +# +# This program is free software; you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation; either version 2 of the License, or +# (at your option) any later version. +# +# 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, write to the Free Software Foundation, Inc., +# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. + +from vyos.vpp.interface.interface import Interface + + +class LoopbackInterface(Interface): + """Interface Loopback""" + + def __init__(self, ifname, kernel_interface: str = '', state: str = 'up'): + super().__init__(ifname) + self.instance = int(ifname.removeprefix('lo')) + self.ifname = f'loop{self.instance}' + self.kernel_interface = kernel_interface + self.initial_state = state + + def add(self): + """Create Loopback interface + https://github.com/FDio/vpp/blob/stable/2306/src/vnet/interface.api + Example: + from vyos.vpp.interface import LoopbackInterface + a = LoopbackInterface(ifname='lo1') + a.add() + """ + self.vpp.api.create_loopback_instance( + is_specified=True, user_instance=self.instance + ) + # Set interface state + self.set_state(self.initial_state) + + def delete(self): + """Delete Loopback interface + Example: + from vyos.vpp.interface import LoopbackInterface + a = LoopbackInterface(ifname='lo1') + a.delete() + """ + loopback_if_index = self.vpp.get_sw_if_index(f'loop{self.instance}') + return self.vpp.api.delete_loopback(sw_if_index=loopback_if_index) + + def kernel_add(self): + """Add LCP pair + Example: + from vyos.vpp.interface import LoopbackInterface + a = LoopbackInterface(ifname='lo1') + a.kernel_add() + """ + self.vpp.lcp_pair_add(self.ifname, self.kernel_interface) + + def kernel_delete(self): + """Delete LCP pair + Example: + from vyos.vpp.interface import LoopbackInterface + a = LoopbackInterface(ifname='lo1') + a.kernel_delete() + """ + self.vpp.lcp_pair_del(self.ifname, self.kernel_interface) diff --git a/python/vyos/vpp/interface/vxlan.py b/python/vyos/vpp/interface/vxlan.py new file mode 100644 index 000000000..4be5b5ec9 --- /dev/null +++ b/python/vyos/vpp/interface/vxlan.py @@ -0,0 +1,106 @@ +# VyOS implementation of VPP VXLAN interface +# +# Copyright (C) VyOS Inc. +# +# This program is free software; you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation; either version 2 of the License, or +# (at your option) any later version. +# +# 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, write to the Free Software Foundation, Inc., +# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. + +from vyos.vpp import VPPControl +from vyos.vpp.interface.interface import Interface + + +def show(): + """Show VXLAN interface + Example: + from vyos.vpp.interface import vxlan + vxlan.show() + """ + vpp = VPPControl() + return vpp.api.vxlan_tunnel_dump() + + +class VXLANInterface(Interface): + """Interface VXLAN""" + + def __init__( + self, + ifname, + source_address, + remote, + vni, + kernel_interface: str = '', + state: str = 'up', + ): + super().__init__(ifname) + self.instance = int(ifname.removeprefix('vxlan')) + self.ifname = f'vxlan_tunnel{self.instance}' + self.src_address = source_address + self.dst_address = remote + self.vni = vni + self.kernel_interface = kernel_interface + self.initial_state = state + + def add(self): + """Create VXLAN interface + https://github.com/FDio/vpp/blob/stable/2306/src/plugins/vxlan/vxlan.api + + Example: + from vyos.vpp.interface import VXLANInterface + a = VXLANInterface(ifname='vxlan23', source_address='192.0.2.1', remote='203.0.113.23', vni=23) + a.add() + """ + self.vpp.api.vxlan_add_del_tunnel_v3( + is_add=True, + src_address=self.src_address, + dst_address=self.dst_address, + vni=self.vni, + instance=self.instance, + decap_next_index=1, + is_l3=False, + ) + # Set interface state + self.set_state(self.initial_state) + + def delete(self): + """Delete VXLAN interface + Example: + from vyos.vpp.interface import VXLANInterface + a = VXLANInterface(ifname='vxlan23', source_address='192.0.2.1', remote='203.0.113.23', vni=23) + a.delete() + """ + return self.vpp.api.vxlan_add_del_tunnel_v3( + is_add=False, + src_address=self.src_address, + dst_address=self.dst_address, + vni=self.vni, + is_l3=False, + ) + + def kernel_add(self): + """Add LCP pair + Example: + from vyos.vpp.interface import VXLANInterface + a = VXLANInterface(ifname='vxlan23', source_address='192.0.2.1', remote='203.0.113.23', vni=23, kernel_interface='vpptap10') + a.kernel_add() + """ + self.vpp.lcp_pair_add(self.ifname, self.kernel_interface) + + def kernel_delete(self): + """Delete LCP pair + Example: + from vyos.vpp.interface import VXLANInterface + a = VXLANInterface(ifname='vxlan23', source_address='192.0.2.1', remote='203.0.113.23', vni=23) + a.kernel_delete() + """ + self.vpp.lcp_pair_del(self.ifname, self.kernel_interface) diff --git a/python/vyos/vpp/interface/wireguard.py b/python/vyos/vpp/interface/wireguard.py new file mode 100644 index 000000000..c2d36e1b0 --- /dev/null +++ b/python/vyos/vpp/interface/wireguard.py @@ -0,0 +1,55 @@ +# VyOS implementation of VPP Wireguard interface +# +# Copyright (C) VyOS Inc. +# +# This program is free software; you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation; either version 2 of the License, or +# (at your option) any later version. +# +# 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, write to the Free Software Foundation, Inc., +# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. + +from vyos.vpp import VPPControl + + +class WireguardInterface: + """Interface Wireguard""" + + def __init__(self, ifname, listen_port=51820, kernel_interface: str = ''): + self.instance = int(ifname.removeprefix('wg')) + self.ifname = ifname + self.listen_port = listen_port + self.kernel_interface = kernel_interface + self.vpp = VPPControl() + + def add(self): + """Create Wireguard interface + https://github.com/FDio/vpp/blob/stable/2306/src/plugins/wireguard/wireguard.api + Example: + from vyos.vpp.interface import WireguardInterface + a = WireguardInterface(ifname='wg5', listen_port=51820, generate_key=True) + a.add() + """ + self.vpp.api.wireguard_interface_create( + generate_key=True, + interface={'user_instance': self.instance, 'listen_port': self.listen_port}, + ) + if self.kernel_interface: + self.vpp.lcp_pair_add(self.ifname, self.kernel_interface, 'tun') + + def delete(self): + """Delete Wireguard interface + Example: + from vyos.vpp.interface import WireguardInterface + a = WireguardInterface(ifname='wg5') + a.delete() + """ + wg_if_index = self.vpp.get_sw_if_index(f'wg{self.instance}') + return self.vpp.api.wireguard_interface_delete(sw_if_index=wg_if_index) diff --git a/python/vyos/vpp/interface/xconnect.py b/python/vyos/vpp/interface/xconnect.py new file mode 100644 index 000000000..299a61405 --- /dev/null +++ b/python/vyos/vpp/interface/xconnect.py @@ -0,0 +1,95 @@ +# VyOS implementation of VPP bridge interface +# +# Copyright (C) VyOS Inc. +# +# This program is free software; you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation; either version 2 of the License, or +# (at your option) any later version. +# +# 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, write to the Free Software Foundation, Inc., +# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. + +from vyos.vpp import VPPControl +from vyos.vpp.utils import iftunnel_transform + + +class XconnectInterface: + def __init__( + self, + ifname: str, + members: list = [], + state: str = 'up', + ): + self.ifname = ifname + self.members = members + self.vpp = VPPControl() + self.initial_state = state + + def add_l2_xconnect(self): + """Add l2 cross connect + Args: + members (list): The list of the xconnect members + Example: + from vyos.vpp.interface import XconnectInterface + a = XconnectInterface(ifname='xcon0', members=['eth0', 'vxlan0']) + a.add_l2_xconnect() + """ + interface_transform_filter = ('vxlan', 'gre') + first_member = self.members[0] + second_member = self.members[1] + # Check if member in required filter to transform 'vxlanX' => 'vxlan_tunnelX' + if first_member.startswith(interface_transform_filter): + first_member = iftunnel_transform(first_member) + if second_member.startswith(interface_transform_filter): + second_member = iftunnel_transform(second_member) + + member_first_if_index = self.vpp.get_sw_if_index(first_member) + member_second_if_index = self.vpp.get_sw_if_index(second_member) + self.vpp.api.sw_interface_set_l2_xconnect( + rx_sw_if_index=member_first_if_index, + tx_sw_if_index=member_second_if_index, + enable=True, + ) + self.vpp.api.sw_interface_set_l2_xconnect( + rx_sw_if_index=member_second_if_index, + tx_sw_if_index=member_first_if_index, + enable=True, + ) + + def del_l2_xconnect(self): + """Move l2 cross connect member to mode l3 (delte xconnect) + Args: + members (list): The list of the xconnect members + Example: + from vyos.vpp.interface import XconnectInterface + a = XconnectInterface(ifname='xcon0', members=['eth0', 'vxlan0']) + a.del_l2_xconnect() + """ + interface_transform_filter = ('vxlan', 'gre') + first_member = self.members[0] + second_member = self.members[1] + # Check if member in required filter to transform 'vxlanX' => 'vxlan_tunnelX' + if first_member.startswith(interface_transform_filter): + first_member = iftunnel_transform(first_member) + if second_member.startswith(interface_transform_filter): + second_member = iftunnel_transform(second_member) + + member_first_if_index = self.vpp.get_sw_if_index(first_member) + member_second_if_index = self.vpp.get_sw_if_index(second_member) + self.vpp.api.sw_interface_set_l2_xconnect( + rx_sw_if_index=member_first_if_index, + tx_sw_if_index=member_second_if_index, + enable=False, + ) + self.vpp.api.sw_interface_set_l2_xconnect( + rx_sw_if_index=member_second_if_index, + tx_sw_if_index=member_first_if_index, + enable=False, + ) diff --git a/python/vyos/vpp/nat/__init__.py b/python/vyos/vpp/nat/__init__.py new file mode 100644 index 000000000..16685c9cc --- /dev/null +++ b/python/vyos/vpp/nat/__init__.py @@ -0,0 +1,4 @@ +from .nat44 import Nat44 +from .det44 import Det44 + +__all__ = ['Nat44', 'Det44'] diff --git a/python/vyos/vpp/nat/det44.py b/python/vyos/vpp/nat/det44.py new file mode 100644 index 000000000..70f903086 --- /dev/null +++ b/python/vyos/vpp/nat/det44.py @@ -0,0 +1,106 @@ +# +# Copyright (C) VyOS Inc. +# +# This program is free software; you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation; either version 2 of the License, or +# (at your option) any later version. +# +# 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, write to the Free Software Foundation, Inc., +# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. + +from vyos.vpp import VPPControl + + +class Det44: + def __init__(self): + self.vpp = VPPControl() + + def enable_det44_plugin(self): + """Enable DET44 plugin + Example: + from vyos.vpp.nat import Det44 + det44 = Det44() + det44.enable_det44_plugin() + https://github.com/FDio/vpp/blob/stable/2410/src/plugins/nat/det44/det44.api + """ + self.vpp.api.det44_plugin_enable_disable(enable=True) + + def disable_det44_plugin(self): + """Disable DET44 plugin""" + self.vpp.api.det44_plugin_enable_disable(enable=False) + + def add_det44_interface_outside(self, interface_out): + """Add DET44 outside interface""" + self.vpp.api.det44_interface_add_del_feature( + sw_if_index=self.vpp.get_sw_if_index(interface_out), + is_inside=False, + is_add=True, + ) + + def delete_det44_interface_outside(self, interface_out): + """Delete DET44 outside interface""" + self.vpp.api.det44_interface_add_del_feature( + sw_if_index=self.vpp.get_sw_if_index(interface_out), + is_inside=False, + is_add=False, + ) + + def add_det44_interface_inside(self, interface_in): + """Add DET44 inside interface""" + self.vpp.api.det44_interface_add_del_feature( + sw_if_index=self.vpp.get_sw_if_index(interface_in), + is_inside=True, + is_add=True, + ) + + def delete_det44_interface_inside(self, interface_in): + """Delete DET44 inside interface""" + self.vpp.api.det44_interface_add_del_feature( + sw_if_index=self.vpp.get_sw_if_index(interface_in), + is_inside=True, + is_add=False, + ) + + def add_det44_mapping(self, in_addr, in_plen, out_addr, out_plen): + """Add DET44 mapping""" + self.vpp.api.det44_add_del_map( + in_addr=in_addr, + in_plen=in_plen, + out_addr=out_addr, + out_plen=out_plen, + is_add=True, + ) + + def delete_det44_mapping(self, in_addr, in_plen, out_addr, out_plen): + """Delete DET44 mapping""" + self.vpp.api.det44_add_del_map( + in_addr=in_addr, + in_plen=in_plen, + out_addr=out_addr, + out_plen=out_plen, + is_add=False, + ) + + def set_det44_timeouts( + self, icmp: int, udp: int, tcp_established: int, tcp_transitory: int + ): + """Set DET44 timeouts + Args: + tcp_established (int): TCP established timeout + tcp_transitory (int): TCP transitory timeout + udp (int): UDP timeout + icmp (int): ICMP timeout + """ + self.vpp.api.det44_set_timeouts( + icmp=icmp, + udp=udp, + tcp_established=tcp_established, + tcp_transitory=tcp_transitory, + ) diff --git a/python/vyos/vpp/nat/nat44.py b/python/vyos/vpp/nat/nat44.py new file mode 100644 index 000000000..b7e2d7744 --- /dev/null +++ b/python/vyos/vpp/nat/nat44.py @@ -0,0 +1,229 @@ +# +# Copyright (C) VyOS Inc. +# +# This program is free software; you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation; either version 2 of the License, or +# (at your option) any later version. +# +# 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, write to the Free Software Foundation, Inc., +# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. + +from vyos.vpp import VPPControl + + +# NAT44 flags +NAT_IS_NONE = 0x00 +NAT_IS_TWICE_NAT = 0x01 +NAT_IS_SELF_TWICE_NAT = 0x02 +NAT_IS_OUT2IN_ONLY = 0x04 +NAT_IS_ADDR_ONLY = 0x08 +NAT_IS_OUTSIDE = 0x10 +NAT_IS_INSIDE = 0x20 + +NO_INTERFACE = 0xFFFFFFFF + + +class Nat44: + def __init__(self): + self.vpp = VPPControl() + + def enable_nat44_ed(self): + """Enable NAT44 endpoint dependent plugin + Example: + from vyos.vpp.nat import Nat44 + nat44 = Nat44() + nat44.enable_nat44_ed() + https://github.com/FDio/vpp/blob/stable/2410/src/plugins/nat/nat44-ed/nat44_ed.api + """ + self.vpp.api.nat44_ed_plugin_enable_disable(enable=True) + + def disable_nat44_ed(self): + """Disable NAT44 endpoint dependent plugin""" + self.vpp.api.nat44_ed_plugin_enable_disable(enable=False) + + def enable_nat44_ei(self): + """Enable NAT44 endpoint independent plugin + Example: + from vyos.vpp.nat import Nat44 + nat44 = Nat44() + nat44.enable_nat44_ei() + """ + self.vpp.api.nat44_ei_plugin_enable_disable(enable=True) + + def add_nat44_interface_inside(self, interface_in): + """Add NAT44 interface""" + self.vpp.api.nat44_interface_add_del_feature( + flags=NAT_IS_INSIDE, + sw_if_index=self.vpp.get_sw_if_index(interface_in), + is_add=True, + ) + + def delete_nat44_interface_inside(self, interface_in): + """Delete NAT44 interface""" + self.vpp.api.nat44_interface_add_del_feature( + flags=NAT_IS_INSIDE, + sw_if_index=self.vpp.get_sw_if_index(interface_in), + is_add=False, + ) + + def add_nat44_interface_outside(self, interface_out): + """Add NAT44 interface""" + self.vpp.api.nat44_interface_add_del_feature( + flags=NAT_IS_OUTSIDE, + sw_if_index=self.vpp.get_sw_if_index(interface_out), + is_add=True, + ) + + def delete_nat44_interface_outside(self, interface_out): + """Delete NAT44 interface""" + self.vpp.api.nat44_interface_add_del_feature( + flags=NAT_IS_OUTSIDE, + sw_if_index=self.vpp.get_sw_if_index(interface_out), + is_add=False, + ) + + def add_nat44_address_range(self, addresses, twice_nat): + """Add NAT44 address range""" + if '-' not in addresses: + first_ip_address = last_ip_address = addresses + else: + first_ip_address, last_ip_address = addresses.split('-') + self.vpp.api.nat44_add_del_address_range( + flags=NAT_IS_TWICE_NAT if twice_nat else NAT_IS_NONE, + first_ip_address=first_ip_address, + last_ip_address=last_ip_address, + is_add=True, + ) + + def delete_nat44_address_range(self, addresses, twice_nat): + """Delete NAT44 address range""" + if '-' not in addresses: + first_ip_address = last_ip_address = addresses + else: + first_ip_address, last_ip_address = addresses.split('-') + self.vpp.api.nat44_add_del_address_range( + flags=NAT_IS_TWICE_NAT if twice_nat else NAT_IS_NONE, + first_ip_address=first_ip_address, + last_ip_address=last_ip_address, + is_add=False, + ) + + def add_nat44_interface_address(self, interface, twice_nat): + """Add NAT44 interface address""" + self.vpp.api.nat44_add_del_interface_addr( + flags=NAT_IS_TWICE_NAT if twice_nat else NAT_IS_NONE, + sw_if_index=self.vpp.get_sw_if_index(interface), + is_add=True, + ) + + def delete_nat44_interface_address(self, interface, twice_nat): + """Delete NAT44 interface address""" + self.vpp.api.nat44_add_del_interface_addr( + flags=NAT_IS_TWICE_NAT if twice_nat else NAT_IS_NONE, + sw_if_index=self.vpp.get_sw_if_index(interface), + is_add=False, + ) + + def add_nat44_static_mapping( + self, + local_ip, + external_ip, + local_port, + external_port, + protocol, + twice_nat, + self_twice_nat, + out2in, + pool_ip, + ): + """Add NAT44 static mapping""" + flags = NAT_IS_ADDR_ONLY if not (protocol or local_port) else NAT_IS_NONE + flags |= NAT_IS_TWICE_NAT if twice_nat else 0 + flags |= NAT_IS_SELF_TWICE_NAT if self_twice_nat else 0 + flags |= NAT_IS_OUT2IN_ONLY if out2in else 0 + self.vpp.api.nat44_add_del_static_mapping_v2( + local_ip_address=local_ip, + external_ip_address=external_ip, + protocol=protocol, + local_port=local_port, + external_port=external_port, + match_pool=True if pool_ip else False, + pool_ip_address=pool_ip if pool_ip else '', + flags=flags, + is_add=True, + ) + + def delete_nat44_static_mapping( + self, + local_ip, + external_ip, + local_port, + external_port, + protocol, + twice_nat, + self_twice_nat, + out2in, + pool_ip, + ): + """Delete NAT44 static mapping""" + flags = NAT_IS_ADDR_ONLY if not (protocol or local_port) else NAT_IS_NONE + flags |= NAT_IS_TWICE_NAT if twice_nat else 0 + flags |= NAT_IS_SELF_TWICE_NAT if self_twice_nat else 0 + flags |= NAT_IS_OUT2IN_ONLY if out2in else 0 + self.vpp.api.nat44_add_del_static_mapping_v2( + local_ip_address=local_ip, + external_ip_address=external_ip, + protocol=protocol, + local_port=local_port, + external_port=external_port, + match_pool=True if pool_ip else False, + pool_ip_address=pool_ip if pool_ip else '', + flags=flags, + is_add=False, + ) + + def add_nat44_identity_mapping(self, ip_address, protocol, port, interface): + """Add NAT44 identity mapping""" + self.vpp.api.nat44_add_del_identity_mapping( + ip_address=ip_address, + protocol=protocol, + port=port, + sw_if_index=( + self.vpp.get_sw_if_index(interface) if interface else NO_INTERFACE + ), + flags=NAT_IS_ADDR_ONLY if not (protocol or port) else NAT_IS_NONE, + is_add=True, + ) + + def delete_nat44_identity_mapping(self, ip_address, protocol, port, interface): + """Delete NAT44 identity mapping""" + self.vpp.api.nat44_add_del_identity_mapping( + ip_address=ip_address, + protocol=protocol, + port=port, + sw_if_index=( + self.vpp.get_sw_if_index(interface) if interface else NO_INTERFACE + ), + flags=NAT_IS_ADDR_ONLY if not (protocol or port) else NAT_IS_NONE, + is_add=False, + ) + + def set_nat_timeouts(self, icmp, udp, tcp_established, tcp_transitory): + """Set NAT timeouts""" + self.vpp.api.nat_set_timeouts( + icmp=icmp, + udp=udp, + tcp_established=tcp_established, + tcp_transitory=tcp_transitory, + ) + + def enable_ipfix(self): + """Enable NAT44 IPFIX logging""" + self.vpp.api.nat44_ei_ipfix_enable_disable(enable=True) diff --git a/python/vyos/vpp/utils.py b/python/vyos/vpp/utils.py new file mode 100644 index 000000000..8b3a8c038 --- /dev/null +++ b/python/vyos/vpp/utils.py @@ -0,0 +1,412 @@ +# +# Copyright (C) VyOS Inc. +# +# This program is free software; you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation; either version 2 of the License, or +# (at your option) any later version. +# +# 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, write to the Free Software Foundation, Inc., +# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. + +import ctypes +import os +import socket +from fcntl import ioctl +from pathlib import Path +from struct import pack + + +mem_shift = {'K': 10, 'k': 10, 'M': 20, 'm': 20, 'G': 30, 'g': 30} + + +def iftunnel_transform(iface: str) -> str: + """Transform interface name from `xxxNN` to `xxx_tunnelNN` + + Args: + iface (str): original interface name + + Raises: + ValueError: Raised if an interface name does not start with a alpha and ends with decimal digit + + Returns: + str: Transformed interface name + """ + # Check format + if not iface[0].isascii() or not iface[-1].isdecimal(): + raise ValueError(f'Wrong interface name format: {iface}') + # Transform + iface_type: str = iface.rstrip('0123456789') + iface_num: str = iface.removeprefix(iface_type) + # Return transformed + return f'{iface_type}_tunnel{iface_num}' + + +def cli_ifaces_list(config_instance, mode: str = 'candidate') -> list[str]: + """List of all VPP interfaces (CLI names) + + Args: + config_instance (VyOS Config): VyOS Config instance + mode (str, optional): `candidate` or `running`. Defaults to 'candidate'. + + Returns: + list[str]: list of interfaces + """ + + effective_mode: bool = True if mode == 'running' else False + + # Read a config + config = config_instance.get_config_dict( + ['vpp'], + key_mangling=('-', '_'), + effective=effective_mode, + get_first_key=True, + no_tag_node_value_mangle=True, + with_recursive_defaults=True, + ) + + vpp_ifaces: list[str] = [] + + # Get a list of Ethernet interfaces + for iface in config.get('settings', {}).get('interface', {}).keys(): + vpp_ifaces.append(iface) + + # Get a list of VPP interfaces + for iface_type in config.get('interfaces', {}).keys(): + for iface in config.get('interfaces', {}).get(iface_type, {}).keys(): + vpp_ifaces.append(iface) + + return vpp_ifaces + + +def cli_ethernet_with_vifs_ifaces(config_instance) -> list[str]: + """List of all VPP Ethernet interfaces with VIFs + + Args: + config_instance (VyOS Config): VyOS Config instance + + Returns: + list[str]: list of interfaces + """ + from vyos.configdict import get_interface_dict + + # Read a config + config = config_instance.get_config_dict( + ['vpp'], + key_mangling=('-', '_'), + get_first_key=True, + no_tag_node_value_mangle=True, + with_recursive_defaults=True, + ) + + ifaces: list[str] = [] + + # Get a list of Ethernet interfaces + for iface in config.get('settings', {}).get('interface', {}).keys(): + ifaces.append(iface) + + # Add Ethernet interfaces with VIFs + for iface in ifaces: + _, iface_config = get_interface_dict( + config_instance, ['interfaces', 'ethernet'], ifname=iface + ) + ifaces.extend([f'{iface}.{vif}' for vif in iface_config.get('vif', {})]) + ifaces.extend([f'{iface}.{vif_s}' for vif_s in iface_config.get('vif_s', {})]) + + return ifaces + + +def vpp_ifaces_list(vpp_api) -> list[dict]: + """List interfaces in VPP + + Args: + vpp_api (_type_): VPP API object + + Returns: + list[dict]: list of dictionaries with interfaces + """ + ifaces_list: list[dict] = [] + sw_ifaces_dump = vpp_api.sw_interface_dump() + while sw_ifaces_dump: + iface_details = sw_ifaces_dump.pop() + ifaces_list.append(iface_details._asdict()) + + return ifaces_list + + +def vpp_ip_addresses_by_index(vpp_api, index: str) -> list[str]: + """List of IP addresses for interface by its index in VPP + + Args: + vpp_api (_type_): VPP API object + index (str): interface index in vpp + + Returns: + list[str]: list of IP addresses + """ + ip_addresses_list: list[dict] = [] + ip_address_dump = vpp_api.ip_address_dump(sw_if_index=index) + while ip_address_dump: + ip_address_details = ip_address_dump.pop() + ip_addresses_list.append(str(ip_address_details._asdict().get('prefix'))) + return ip_addresses_list + + +def vpp_ifaces_stats( + iface_name: str = '', +) -> dict[str, dict[str, int | dict[str, int]]]: + from re import compile as re_compile + from vpp_papi import vpp_stats + + def total_value(val_list: vpp_stats.SimpleList) -> int | dict[str, int]: + """Helper for aggregation stats from multiple workers + + Args: + val_list (vpp_stats.SimpleList): list of stats for all workers + + Returns: + int | dict[str, int]: Summary stats + """ + # if all items are int return their sum + if all(isinstance(value, int) for value in val_list): + return sum(val_list) + # if all items are tuple + if all(isinstance(value, tuple) for value in val_list): + combined_stats = {} + # process individual workers + for worker_stats in val_list: + packets, octets = worker_stats + sum_packets = combined_stats.get('packets', 0) + packets + sum_octets = combined_stats.get('octets', 0) + octets + combined_stats = {'packets': sum_packets, 'bytes': sum_octets} + return combined_stats + + # items are something unknown, just return what we received + return val_list + + stats = vpp_stats.VPPStats() + + ifaces_stats: dict[str, dict[str, int | dict[str, int]]] = {} + + # prepare parser for stats output + regex_parser = re_compile(r'^/interfaces/(?P<iface>[^/]+)/(?P<param>[^/]+)') + # get list of available stats and dump them + stats_list: list[str] = stats.ls([f'^/interfaces/{iface_name}']) + stats_dump: list[dict[str, int]] = stats.dump(stats_list) + + # parse outputs and convert it to a dictionary + for stats_key, stats_value in stats_dump.items(): + parsed_key = regex_parser.search(stats_key).groupdict() + iface_name = parsed_key['iface'] + param = parsed_key['param'] + stats_item = {param: total_value(stats_value)} + if iface_name in ifaces_stats: + ifaces_stats[iface_name].update(stats_item) + else: + ifaces_stats[iface_name] = stats_item + + return ifaces_stats + + +def cli_ifaces_lcp_kernel_list( + config_instance, mode: str = 'candidate' +) -> list[tuple[str, str]]: + """List of all VPP kernel-interfaces (CLI names, attached VPP interfaces) + + Args: + config_instance (VyOS Config): VyOS Config instance + mode (str, optional): `candidate` or `running`. Defaults to 'candidate'. + + Returns: + list[tuple[str, str]]: list of interfaces ([(vpp_iface, kernel_iface)]) + """ + + effective_mode: bool = True if mode == 'running' else False + + # Read a config + config = config_instance.get_config_dict( + ['vpp'], + key_mangling=('-', '_'), + effective=effective_mode, + get_first_key=True, + no_tag_node_value_mangle=True, + with_recursive_defaults=True, + ) + + lcp_kernel_ifaces: list[tuple[str, str]] = [] + + # Get a list with kernel interfaces + for ifaces_list in config.get('interfaces', {}).values(): + for iface_name, iface_settings in ifaces_list.items(): + if 'kernel_interface' in iface_settings: + lcp_kernel_ifaces.append( + (iface_name, iface_settings['kernel_interface']) + ) + + return lcp_kernel_ifaces + + +def get_default_hugepage_size() -> int: + """ + Retrieve the system's default huge page size. + :return: The default huge page size in bytes. + """ + page_size = None + try: + # default huge page size + memfd = os.memfd_create('tmp', os.MFD_HUGETLB) + st = os.fstat(memfd) + page_size = st.st_blksize + os.close(memfd) + except OSError: + pass + + return page_size + + +def get_default_page_size() -> int: + """ + Retrieve the system's default page size. + :return: The default page size in bytes. + """ + return os.sysconf('SC_PAGESIZE') + + +def get_hugepage_sizes() -> list[int]: + """ + Retrieve all available huge page sizes from the system. + :return: A list of huge page sizes in bytes. + """ + huge_sizes = [] + path = '/sys/kernel/mm/hugepages/' + try: + entries = os.listdir(path) + for entry in entries: + if entry.startswith('hugepages-'): + try: + size_kb = int(entry.replace('hugepages-', '').replace('kB', '')) + huge_sizes.append(size_kb << 10) # Convert KB to bytes + except ValueError: + pass + except FileNotFoundError: + pass + + return huge_sizes + + +def human_memory_to_bytes(value: str) -> int: + """ + Convert a human-readable vpp memory format (K, M, G) to a byte value. + + :param value: The string memory size in vpp human-readable format. + :return: A int representing the value. + """ + try: + return int(value) + except ValueError: + return int(value[:-1]) << mem_shift[value[-1]] + + +def bytes_to_human_memory(value: int, unit: str) -> str | None: + """ + Convert a byte value to a human-readable format (K, M, G). + + :param value: The size in bytes. + :param unit: The unit to convert to ('K', 'M', 'G'). + :return: A string representing the value in the specified unit, or None if zero. + """ + val = value >> mem_shift[unit] + return f'{val}{unit}' if val else None + + +def human_page_memory_to_bytes(value: str) -> int: + """ + Convert a human-readable vpp page size format to a byte value. + + :param value: The string memory size in vpp human-readable format. + :return: A int representing the value. + """ + default = { + 'default': get_default_page_size, + 'default-hugepage': get_default_hugepage_size, + } + try: + return default[value]() + except KeyError: + return human_memory_to_bytes(value) + + +class EthtoolGDrvinfo: + """Return interface details like `ethtol -i` does""" + + # TODO + # this probably need to be replaced with a code generator + # like ctypeslib or C extension + class EthtoolDrvinfo(ctypes.Structure): + _fields_ = [ + ('cmd', ctypes.c_uint32), + ('driver', ctypes.c_char * 32), # Driver short name + ('version', ctypes.c_char * 32), # Driver version + ('fw_version', ctypes.c_char * 32), # Firmware version + # Be careful: bus info can be longer than 32 chars and thus truncated + ('bus_info', ctypes.c_char * 32), # Bus info. + ('erom_version', ctypes.c_char * 32), # Expansion ROM version + ('reserved2', ctypes.c_char * 12), # Reserved for future use + ('n_priv_flags', ctypes.c_uint32), # Number of private flags + ('n_stats', ctypes.c_uint32), # Number of U64 stats + ('testinfo_len', ctypes.c_uint32), # Test info length + ('eedump_len', ctypes.c_uint32), # EEPROM dump length + ('regdump_len', ctypes.c_uint32), # Register dump length + ] + + def __init__(self, iface: str): + # Constants for ethtool + SIOCETHTOOL = 0x8946 # pretend to be ethtool + ETHTOOL_GDRVINFO = 0x00000003 # Command to get driver info + + # Create a dummy socket + sockfd = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + + # Prepare the request for getting driver info + drvinfo = self.EthtoolDrvinfo(cmd=ETHTOOL_GDRVINFO) + ifreq: bytes = pack('16sP', iface.encode('utf-8'), ctypes.addressof(drvinfo)) + + # Make an ioctl call to get the driver info + try: + ioctl(sockfd, SIOCETHTOOL, ifreq) + except OSError: + raise FileNotFoundError(f'There is no Ethernet device: {iface}') + + # Close the socket + sockfd.close() + + # save the information + self.driver: str = drvinfo.driver.decode('utf-8').strip('\x00') + self.version: str = drvinfo.version.decode('utf-8').strip('\x00') + self.fw_version: str = drvinfo.fw_version.decode('utf-8').strip('\x00') + self.bus_info: str = drvinfo.bus_info.decode('utf-8').strip('\x00') + self.erom_version: str = drvinfo.erom_version.decode('utf-8').strip('\x00') + self.reserved2: str = drvinfo.reserved2.decode('utf-8').strip('\x00') + self.n_priv_flags: int = drvinfo.n_priv_flags + self.testinfo_len: int = drvinfo.testinfo_len + self.eedump_len: int = drvinfo.eedump_len + self.regdump_len: int = drvinfo.regdump_len + + def bus_info_expand(self, bus_name: str) -> str: + bus_path = Path(f'/sys/bus/{bus_name}/devices').glob(f'{self.bus_info}*') + dev_ids = list(bus_path) + if not dev_ids: + raise FileNotFoundError( + f'No matching IDs on the bus: {self.bus_info} on {bus_name}' + ) + if len(dev_ids) > 1: + raise FileNotFoundError( + f'There are more than one matching IDs on the bus: {dev_ids} on {bus_name}' + ) + return dev_ids[0].name |