diff options
Diffstat (limited to 'python')
| -rw-r--r-- | python/vyos/vpp/__init__.py | 20 | ||||
| -rw-r--r-- | python/vyos/vpp/config_deps.py | 76 | ||||
| -rw-r--r-- | python/vyos/vpp/config_filter.py | 58 | ||||
| -rw-r--r-- | python/vyos/vpp/config_verify.py | 160 | ||||
| -rw-r--r-- | python/vyos/vpp/configdb.py | 206 | ||||
| -rw-r--r-- | python/vyos/vpp/control_host.py | 363 | ||||
| -rw-r--r-- | python/vyos/vpp/control_vpp.py | 444 | ||||
| -rw-r--r-- | python/vyos/vpp/interface/__init__.py | 40 | ||||
| -rw-r--r-- | python/vyos/vpp/interface/bond.py | 113 | ||||
| -rw-r--r-- | python/vyos/vpp/interface/bridge.py | 128 | ||||
| -rw-r--r-- | python/vyos/vpp/interface/ethernet.py | 54 | ||||
| -rw-r--r-- | python/vyos/vpp/interface/geneve.py | 90 | ||||
| -rw-r--r-- | python/vyos/vpp/interface/gre.py | 85 | ||||
| -rw-r--r-- | python/vyos/vpp/interface/ipip.py | 83 | ||||
| -rw-r--r-- | python/vyos/vpp/interface/loopback.py | 68 | ||||
| -rw-r--r-- | python/vyos/vpp/interface/vxlan.py | 94 | ||||
| -rw-r--r-- | python/vyos/vpp/interface/wireguard.py | 56 | ||||
| -rw-r--r-- | python/vyos/vpp/interface/xconnect.py | 94 | ||||
| -rw-r--r-- | python/vyos/vpp/nat/__init__.py | 0 | ||||
| -rw-r--r-- | python/vyos/vpp/utils.py | 281 |
20 files changed, 2513 insertions, 0 deletions
diff --git a/python/vyos/vpp/__init__.py b/python/vyos/vpp/__init__.py new file mode 100644 index 000000000..a320ac7d1 --- /dev/null +++ b/python/vyos/vpp/__init__.py @@ -0,0 +1,20 @@ +# +# Copyright (C) 2023 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/config_deps.py b/python/vyos/vpp/config_deps.py new file mode 100644 index 000000000..d826dedbb --- /dev/null +++ b/python/vyos/vpp/config_deps.py @@ -0,0 +1,76 @@ +# +# Copyright (C) 2023-2024 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..4ab9ffda2 --- /dev/null +++ b/python/vyos/vpp/config_filter.py @@ -0,0 +1,58 @@ +# +# Copyright (C) 2024 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_verify.py b/python/vyos/vpp/config_verify.py new file mode 100644 index 000000000..6b35858d2 --- /dev/null +++ b/python/vyos/vpp/config_verify.py @@ -0,0 +1,160 @@ +# Used for verifying configuration vpp interfaces +# +# Copyright (C) 2023 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 import ConfigError + +from vyos.vpp.control_host import get_eth_driver + + +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_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 diff --git a/python/vyos/vpp/configdb.py b/python/vyos/vpp/configdb.py new file mode 100644 index 000000000..1830937ad --- /dev/null +++ b/python/vyos/vpp/configdb.py @@ -0,0 +1,206 @@ +# +# Copyright (C) 2024 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..ead9f6968 --- /dev/null +++ b/python/vyos/vpp/control_host.py @@ -0,0 +1,363 @@ +# +# Copyright (C) 2023 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 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]) diff --git a/python/vyos/vpp/control_vpp.py b/python/vyos/vpp/control_vpp.py new file mode 100644 index 000000000..39ecac906 --- /dev/null +++ b/python/vyos/vpp/control_vpp.py @@ -0,0 +1,444 @@ +# +# Copyright (C) 2023 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.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_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 + + @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..14659ed7b --- /dev/null +++ b/python/vyos/vpp/interface/__init__.py @@ -0,0 +1,40 @@ +# +# Copyright (C) 2023 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 .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', + '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..1cc306670 --- /dev/null +++ b/python/vyos/vpp/interface/bond.py @@ -0,0 +1,113 @@ +# +# Copyright (C) 2023 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.control_host import set_promisc + +vpp = VPPControl() + + +class BondInterface: + def __init__( + self, + ifname, + mode: str = '', + load_balance: int = 0, + mac: str = '', + kernel_interface: str = '', + ): + 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 + + 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}) + vpp.api.bond_create2(**create_args) + if self.kernel_interface: + vpp.lcp_pair_add(self.ifname, self.kernel_interface) + + def delete(self): + """Delete Bond interface + Example: + from vyos.vpp.interface import BondInterface + a = BondInterface(ifname='bond0') + a.delete() + """ + bond_if_index = vpp.get_sw_if_index(self.ifname) + 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 = vpp.get_sw_if_index(f'BondEthernet{self.instance}') + member_if_index = vpp.get_sw_if_index(interface) + member_if_type = vpp.get_sw_if_dev_type(interface) + vpp.api.bond_add_member( + bond_sw_if_index=bond_if_index, sw_if_index=member_if_index + ) + 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 = vpp.get_sw_if_index(interface) + 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() + """ + 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() + """ + 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..e220547de --- /dev/null +++ b/python/vyos/vpp/interface/bridge.py @@ -0,0 +1,128 @@ +# +# Copyright (C) 2023-2024 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 + +vpp = 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 + + 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() + """ + 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() + """ + vpp.api.bridge_domain_add_del_v2(is_add=False, bd_id=self.interface_suffix) + + def add_member(self, member: str | int): + """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. + + 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 = vpp.get_sw_if_index(member) + + return vpp.api.sw_interface_set_l2_bridge( + rx_sw_if_index=member_if_index, bd_id=bridge_index, port_type=0 + ) + + 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 = vpp.get_sw_if_index(member) + + return 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..ddd3dc42f --- /dev/null +++ b/python/vyos/vpp/interface/ethernet.py @@ -0,0 +1,54 @@ +# VyOS implementation of VPP Ethernet interface +# +# Copyright (C) 2023 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 + +vpp = 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 + + 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() + """ + 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() + """ + 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..3d5e75003 --- /dev/null +++ b/python/vyos/vpp/interface/geneve.py @@ -0,0 +1,90 @@ +# VyOS implementation of Geneve interface +# +# Copyright (C) 2023 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 + +vpp = VPPControl() + + +def show(): + """Show Geneve interface + Example: + from vyos.vpp.interface import geneve + geneve.show() + """ + 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 + + 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 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 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() + """ + 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() + """ + 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..0017127ec --- /dev/null +++ b/python/vyos/vpp/interface/gre.py @@ -0,0 +1,85 @@ +# VyOS implementation of VPP GRE interface +# +# Copyright (C) 2023 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 + +vpp = VPPControl() + + +def show(): + """Show GRE interface + Example: + from vyos.vpp.interface import gre + gre.show() + """ + return vpp.api.gre_tunnel_dump() + + +class GREInterface: + def __init__(self, ifname, source_address, remote, kernel_interface: str = ''): + self.instance = int(ifname.removeprefix('gre')) + self.ifname = ifname + self.src_address = source_address + self.dst_address = remote + self.kernel_interface = kernel_interface + + def add(self): + """Create GRE interface + https://github.com/FDio/vpp/blob/stable/2306/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') + a.add() + """ + vpp.api.gre_tunnel_add_del( + is_add=True, + tunnel={ + 'src': self.src_address, + 'dst': self.dst_address, + 'instance': self.instance, + }, + ) + + 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 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() + """ + 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() + """ + vpp.lcp_pair_del(self.ifname, self.kernel_interface) diff --git a/python/vyos/vpp/interface/ipip.py b/python/vyos/vpp/interface/ipip.py new file mode 100644 index 000000000..e1f19735b --- /dev/null +++ b/python/vyos/vpp/interface/ipip.py @@ -0,0 +1,83 @@ +# VyOS implementation of VPP IPIP interface +# +# Copyright (C) 2023 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 + +vpp = VPPControl() + + +def show(): + """Show IPIP interface + Example: + from vyos.vpp.interface import ipip + ipip.show() + """ + return vpp.api.ipip_tunnel_dump() + + +class IPIPInterface: + def __init__(self, ifname, source_address, remote, kernel_interface: str = ''): + self.instance = int(ifname.removeprefix('ipip')) + self.ifname = ifname + self.src_address = source_address + self.dst_address = remote + self.kernel_interface = kernel_interface + + 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() + """ + vpp.api.ipip_add_tunnel( + tunnel={ + 'src': self.src_address, + 'dst': self.dst_address, + 'instance': self.instance, + }, + ) + + 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 = vpp.get_sw_if_index(f'ipip{self.instance}') + return 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() + """ + 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() + """ + 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..df7b82356 --- /dev/null +++ b/python/vyos/vpp/interface/loopback.py @@ -0,0 +1,68 @@ +# VyOS implementation of VPP Loopback interface +# +# Copyright (C) 2023 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 + +vpp = VPPControl() + + +class LoopbackInterface: + """Interface Loopback""" + + def __init__(self, ifname, kernel_interface: str = ''): + self.instance = int(ifname.removeprefix('lo')) + self.ifname = f'loop{self.instance}' + self.kernel_interface = kernel_interface + + 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() + """ + vpp.api.create_loopback_instance(is_specified=True, user_instance=self.instance) + + def delete(self): + """Delete Loopback interface + Example: + from vyos.vpp.interface import LoopbackInterface + a = LoopbackInterface(ifname='lo1') + a.delete() + """ + loopback_if_index = vpp.get_sw_if_index(f'loop{self.instance}') + return 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() + """ + 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() + """ + 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..907fcbee0 --- /dev/null +++ b/python/vyos/vpp/interface/vxlan.py @@ -0,0 +1,94 @@ +# VyOS implementation of VPP VXLAN interface +# +# Copyright (C) 2023-2024 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 + +vpp = VPPControl() + + +def show(): + """Show VXLAN interface + Example: + from vyos.vpp.interface import vxlan + vxlan.show() + """ + return vpp.api.vxlan_tunnel_dump() + + +class VXLANInterface: + """Interface VXLAN""" + + def __init__(self, ifname, source_address, remote, vni, kernel_interface: str = ''): + 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 + + 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() + """ + 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, + ) + + 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 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) + a.kernel_add() + """ + 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() + """ + 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..582929851 --- /dev/null +++ b/python/vyos/vpp/interface/wireguard.py @@ -0,0 +1,56 @@ +# VyOS implementation of VPP Wireguard interface +# +# Copyright (C) 2023 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 + +vpp = 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 + + 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() + """ + vpp.api.wireguard_interface_create( + generate_key=True, + interface={'user_instance': self.instance, 'listen_port': self.listen_port}, + ) + if self.kernel_interface: + 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 = vpp.get_sw_if_index(f'wg{self.instance}') + return 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..b4bec3321 --- /dev/null +++ b/python/vyos/vpp/interface/xconnect.py @@ -0,0 +1,94 @@ +# VyOS implementation of VPP bridge interface +# +# Copyright (C) 2023 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 + +vpp = VPPControl() + + +class XconnectInterface: + def __init__( + self, + ifname: str, + members: list = [], + ): + self.ifname = ifname + self.members = members + + 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 = vpp.get_sw_if_index(first_member) + member_second_if_index = vpp.get_sw_if_index(second_member) + 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, + ) + 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 = vpp.get_sw_if_index(first_member) + member_second_if_index = vpp.get_sw_if_index(second_member) + 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, + ) + 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..e69de29bb --- /dev/null +++ b/python/vyos/vpp/nat/__init__.py diff --git a/python/vyos/vpp/utils.py b/python/vyos/vpp/utils.py new file mode 100644 index 000000000..08d69bba2 --- /dev/null +++ b/python/vyos/vpp/utils.py @@ -0,0 +1,281 @@ +# +# Copyright (C) 2023 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 socket +from fcntl import ioctl +from pathlib import Path +from struct import pack + + +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 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 + + +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 |