Merge pull request #4650 from sever-sever/T7697

T7697: Merge vyos-vpp repo into vyos-1x

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