No OneTemporary
Actions

Size

109 KB

Referenced Files

None

Subscribers

None

View Options

	diff --git a/data/config-mode-dependencies/vyos-vpp.json b/data/config-mode-dependencies/vyos-vpp.json
	index 4729ae95c..eac0f56d6 100644
	--- a/data/config-mode-dependencies/vyos-vpp.json
	+++ b/data/config-mode-dependencies/vyos-vpp.json
	@@ -1,70 +1,73 @@
	{
	"vpp": {
	"ethernet": ["interfaces_ethernet"],
	"vpp_interfaces_bonding": ["vpp_interfaces_bonding"],
	"vpp_interfaces_bridge": ["vpp_interfaces_bridge"],
	"vpp_interfaces_ethernet": ["vpp_interfaces_ethernet"],
	"vpp_interfaces_geneve": ["vpp_interfaces_geneve"],
	"vpp_interfaces_gre": ["vpp_interfaces_gre"],
	"vpp_interfaces_ipip": ["vpp_interfaces_ipip"],
	"vpp_interfaces_loopback": ["vpp_interfaces_loopback"],
	"vpp_interfaces_vxlan": ["vpp_interfaces_vxlan"],
	"vpp_interfaces_xconnect": ["vpp_interfaces_xconnect"],
	"vpp_acl": ["vpp_acl"],
	"vpp_nat": ["vpp_nat"],
	"vpp_nat_cgnat": ["vpp_nat_cgnat"],
	"vpp_kernel_interface": ["vpp_kernel-interfaces"],
	"vpp_sflow": ["vpp_sflow"],
	"pppoe_server": ["service_pppoe-server"]
	},
	"vpp_interfaces_bonding": {
	"vpp_interfaces_xconnect": ["vpp_interfaces_xconnect"],
	"vpp_acl": ["vpp_acl"],
	"vpp_nat": ["vpp_nat"],
	"vpp_nat_cgnat": ["vpp_nat_cgnat"],
	"vpp_kernel_interface": ["vpp_kernel-interfaces"]
	},
	"vpp_interfaces_ethernet": {
	"vpp_interfaces_xconnect": ["vpp_interfaces_xconnect"],
	"vpp_acl": ["vpp_acl"],
	"vpp_nat": ["vpp_nat"],
	"vpp_nat_cgnat": ["vpp_nat_cgnat"],
	"vpp_kernel_interface": ["vpp_kernel-interfaces"]
	},
	"vpp_interfaces_geneve": {
	"vpp_interfaces_xconnect": ["vpp_interfaces_xconnect"],
	"vpp_acl": ["vpp_acl"],
	"vpp_nat": ["vpp_nat"],
	"vpp_nat_cgnat": ["vpp_nat_cgnat"],
	"vpp_kernel_interface": ["vpp_kernel-interfaces"]
	},
	"vpp_interfaces_gre": {
	"vpp_interfaces_xconnect": ["vpp_interfaces_xconnect"],
	"vpp_acl": ["vpp_acl"],
	"vpp_nat": ["vpp_nat"],
	"vpp_nat_cgnat": ["vpp_nat_cgnat"],
	"vpp_kernel_interface": ["vpp_kernel-interfaces"]
	},
	"vpp_interfaces_ipip": {
	"vpp_interfaces_xconnect": ["vpp_interfaces_xconnect"],
	"vpp_acl": ["vpp_acl"],
	"vpp_nat": ["vpp_nat"],
	"vpp_nat_cgnat": ["vpp_nat_cgnat"],
	"vpp_kernel_interface": ["vpp_kernel-interfaces"]
	},
	"vpp_interfaces_loopback": {
	"vpp_acl": ["vpp_acl"],
	"vpp_nat": ["vpp_nat"],
	"vpp_nat_cgnat": ["vpp_nat_cgnat"],
	"vpp_kernel_interface": ["vpp_kernel-interfaces"]
	},
	"vpp_interfaces_vxlan": {
	"vpp_interfaces_bridge": ["vpp_interfaces_bridge"],
	"vpp_interfaces_xconnect": ["vpp_interfaces_xconnect"],
	"vpp_acl": ["vpp_acl"],
	"vpp_nat": ["vpp_nat"],
	"vpp_nat_cgnat": ["vpp_nat_cgnat"],
	"vpp_kernel_interface": ["vpp_kernel-interfaces"]
	+ },
	+ "vpp_kernel_interfaces": {
	+ "vpp_nat_cgnat": ["vpp_nat_cgnat"]
	}
	}

	diff --git a/python/vyos/vpp/nat/det44.py b/python/vyos/vpp/nat/det44.py
	index 70f903086..51ad41dc8 100644
	--- a/python/vyos/vpp/nat/det44.py
	+++ b/python/vyos/vpp/nat/det44.py
	@@ -1,106 +1,120 @@
	#
	# 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):
	+ def delete_det44_interface_outside(self, interface_out_index):
	"""Delete DET44 outside interface"""
	self.vpp.api.det44_interface_add_del_feature(
	- sw_if_index=self.vpp.get_sw_if_index(interface_out),
	+ sw_if_index=interface_out_index,
	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):
	+ def delete_det44_interface_inside(self, interface_in_index):
	"""Delete DET44 inside interface"""
	self.vpp.api.det44_interface_add_del_feature(
	- sw_if_index=self.vpp.get_sw_if_index(interface_in),
	+ sw_if_index=interface_in_index,
	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,
	)
	+
	+ def get_det44_interfaces_outside(self):
	+ ifaces_outside = []
	+ for iface in self.vpp.api.det44_interface_dump():
	+ if iface.is_outside:
	+ ifaces_outside.append(iface.sw_if_index)
	+ return ifaces_outside
	+
	+ def get_det44_interfaces_inside(self):
	+ ifaces_inside = []
	+ for iface in self.vpp.api.det44_interface_dump():
	+ if iface.is_inside:
	+ ifaces_inside.append(iface.sw_if_index)
	+ return ifaces_inside
	diff --git a/smoketest/scripts/cli/test_vpp.py b/smoketest/scripts/cli/test_vpp.py
	index 4738db485..19e24e7f4 100755
	--- a/smoketest/scripts/cli/test_vpp.py
	+++ b/smoketest/scripts/cli/test_vpp.py
	@@ -1,1516 +1,1560 @@
	#!/usr/bin/env python3
	#
	# 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 os
	import re
	import unittest
	from collections import defaultdict

	from json import loads

	from base_vyostest_shim import VyOSUnitTestSHIM

	from vyos.configsession import ConfigSessionError
	from vyos.utils.process import process_named_running
	from vyos.utils.file import read_file
	from vyos.utils.process import rc_cmd
	from vyos.utils.system import sysctl_read
	from vyos.system import image

	PROCESS_NAME = 'vpp_main'
	VPP_CONF = '/run/vpp/vpp.conf'
	base_path = ['vpp']
	driver = 'dpdk'
	interface = 'eth1'


	def get_vpp_config():
	config = defaultdict(dict)
	current_section = None

	with open(VPP_CONF, 'r') as f:
	for line in f:
	line = line.strip()

	if not line or line.startswith('#'): # Ignore empty lines and comments
	continue

	section_match = re.match(r'([a-zA-Z0-9_-]+)\s*{', line)
	if section_match:
	current_section = section_match.group(1)
	config[current_section] = {}
	continue

	if line == '}': # End of section
	current_section = None
	continue

	key_value_match = re.match(r'([a-zA-Z0-9_-]+)\s+(.+)', line)
	if key_value_match:
	key, value = key_value_match.groups()
	if current_section:
	config[current_section][key] = value
	else:
	config[key] = value

	return config


	def get_address(interface):
	rc, data = rc_cmd(f'ip --json address show dev {interface}')
	if rc == 0:
	data = loads(data)
	if isinstance(data, list) and len(data) > 0:
	ip_address = data[0]['addr_info'][0]['local']
	return ip_address


	class TestVPP(VyOSUnitTestSHIM.TestCase):
	@classmethod
	def setUpClass(cls):
	super(TestVPP, cls).setUpClass()

	# ensure we can also run this test on a live system - so lets clean
	# out the current configuration :)
	cls.cli_delete(cls, base_path)

	def setUp(self):
	self.cli_set(base_path + ['settings', 'interface', interface, 'driver', driver])
	self.cli_set(base_path + ['settings', 'unix', 'poll-sleep-usec', '10'])

	def tearDown(self):
	try:
	# Check for running process
	self.assertTrue(process_named_running(PROCESS_NAME))
	finally:
	# Ensure these cleanup operations always run
	self.cli_delete(base_path)
	self.cli_commit()

	# delete address for Ethernet interface
	self.cli_delete(['interfaces', 'ethernet', interface, 'address'])
	self.cli_commit()

	self.assertFalse(os.path.exists(VPP_CONF))
	self.assertFalse(process_named_running(PROCESS_NAME))

	def test_01_vpp_basic(self):
	main_core = '0'
	poll_sleep = '0'

	# Main core must be verified
	# expect raise ConfigError
	self.cli_set(base_path + ['settings', 'cpu', 'main-core', '99'])

	with self.assertRaises(ConfigSessionError):
	self.cli_commit()

	self.cli_set(base_path + ['settings', 'cpu', 'main-core', main_core])
	self.cli_set(base_path + ['settings', 'unix', 'poll-sleep-usec', poll_sleep])

	# commit changes
	self.cli_commit()

	config_entries = (
	f'poll-sleep-usec {poll_sleep}',
	f'main-core {main_core}',
	'plugin default { disable }',
	'plugin dpdk_plugin.so { enable }',
	'plugin linux_cp_plugin.so { enable }',
	'dev 0000:00:00.0',
	'uio-bind-force',
	)

	# Check configured options
	config = read_file(VPP_CONF)
	for config_entry in config_entries:
	self.assertIn(config_entry, config)

	# route-no-paths is not present in the output
	# looks like vpp bug
	_, out = rc_cmd('sudo vppctl show lcp')
	required_str = 'lcp route-no-paths on'
	self.assertIn(required_str, out)

	self.cli_set(base_path + ['settings', 'lcp', 'ignore-kernel-routes'])
	self.cli_commit()

	# check disabled 'route no path'
	_, out = rc_cmd('sudo vppctl show lcp')
	required_str = 'lcp route-no-paths off'
	self.assertIn(required_str, out)

	def test_02_vpp_vxlan(self):
	vni = '23'
	interface_vxlan = f'vxlan{vni}'
	interface_kernel = f'vpptap{vni}'
	new_interface_kernel = f'vpptap1{vni}'
	source_address = '192.0.2.1'
	new_source_address = '192.0.2.3'
	remote_address = '192.0.2.254'
	kernel_address = '203.0.113.1'

	self.cli_set(['interfaces', 'ethernet', interface, 'address', '192.0.2.1/24'])
	self.cli_set(
	base_path
	+ ['interfaces', 'vxlan', interface_vxlan, 'source-address', source_address]
	)
	self.cli_set(base_path + ['interfaces', 'vxlan', interface_vxlan, 'vni', vni])

	# remote and source address must not be the same
	# expect raise ConfigError
	self.cli_set(
	base_path
	+ ['interfaces', 'vxlan', interface_vxlan, 'remote', source_address]
	)
	with self.assertRaises(ConfigSessionError):
	self.cli_commit()

	self.cli_set(
	base_path
	+ ['interfaces', 'vxlan', interface_vxlan, 'remote', remote_address]
	)
	self.cli_set(
	base_path
	+ [
	'interfaces',
	'vxlan',
	interface_vxlan,
	'kernel-interface',
	interface_kernel,
	]
	)
	self.cli_set(
	base_path
	+ ['kernel-interfaces', interface_kernel, 'address', f'{kernel_address}/24']
	)

	# commit changes
	self.cli_commit()

	self.assertTrue(os.path.isdir(f'/sys/class/net/{interface_kernel}'))

	current_address = get_address(interface_kernel)
	self.assertEqual(kernel_address, current_address)

	# check vxlan interface
	_, out = rc_cmd('sudo vppctl show vxlan tunnel')
	required_str = f'[0] instance 23 src {source_address} dst {remote_address} src_port 4789 dst_port 4789 vni {vni}'
	self.assertIn(required_str, out)

	# update vxlan interface
	self.cli_set(
	base_path
	+ [
	'interfaces',
	'vxlan',
	interface_vxlan,
	'source-address',
	new_source_address,
	]
	)

	# source address of the tunnel interface should be configured
	# expect raise ConfigError
	with self.assertRaises(ConfigSessionError):
	self.cli_commit()

	self.cli_set(
	[
	'interfaces',
	'ethernet',
	interface,
	'vif',
	vni,
	'address',
	f'{new_source_address}/24',
	]
	)
	self.cli_commit()

	# check gre interface after update
	_, out = rc_cmd('sudo vppctl show vxlan tunnel')
	required_str = (
	f'[0] instance {vni} src {new_source_address} dst {remote_address}'
	)
	self.assertIn(required_str, out)
	self.assertTrue(os.path.isdir(f'/sys/class/net/{interface_kernel}'))
	self.assertEqual(kernel_address, current_address)

	# change vpp settings
	self.cli_set(base_path + ['settings', 'unix', 'poll-sleep-usec', '5'])
	self.cli_commit()

	config = read_file(VPP_CONF)
	self.assertIn('poll-sleep-usec 5', config)

	# delete vxlan kernel-interface but do not delete 'vpp kernel-interface'
	# expect raise ConfigError
	self.cli_delete(
	base_path
	+ [
	'interfaces',
	'vxlan',
	interface_vxlan,
	'kernel-interface',
	interface_kernel,
	]
	)
	with self.assertRaises(ConfigSessionError):
	self.cli_commit()

	# update vxlan kernel-interface but do not change 'vpp kernel-interface'
	# expect raise ConfigError
	self.cli_set(
	base_path
	+ [
	'interfaces',
	'vxlan',
	interface_vxlan,
	'kernel-interface',
	new_interface_kernel,
	]
	)
	with self.assertRaises(ConfigSessionError):
	self.cli_commit()

	# delete vpp kernel-interface
	self.cli_delete(base_path + ['kernel-interfaces', interface_kernel])
	self.cli_commit()

	# delete vxlan kernel-interface
	self.cli_delete(
	base_path + ['interfaces', 'vxlan', interface_vxlan, 'kernel-interface']
	)
	self.cli_commit()
	self.assertFalse(os.path.isdir(f'/sys/class/net/{interface_kernel}'))

	# delete vxlan interface
	self.cli_delete(base_path + ['interfaces', 'vxlan', interface_vxlan])
	self.cli_commit()

	# delete vif Ethernet interface
	self.cli_delete(['interfaces', 'ethernet', interface, 'vif'])
	self.cli_commit()

	def test_03_vpp_gre(self):
	interface_gre = 'gre12'
	interface_kernel = 'vpptun12'
	new_interface_kernel = 'vpptun123'
	source_address = '192.0.2.1'
	new_source_address = '192.0.2.2'
	remote_address = '192.0.2.254'
	kernel_address = '10.0.0.0'

	self.cli_set(
	base_path
	+ ['interfaces', 'gre', interface_gre, 'source-address', source_address]
	)
	self.cli_set(
	base_path + ['interfaces', 'gre', interface_gre, 'remote', remote_address]
	)
	self.cli_set(
	base_path
	+ ['interfaces', 'gre', interface_gre, 'kernel-interface', interface_kernel]
	)
	self.cli_set(
	base_path
	+ ['kernel-interfaces', interface_kernel, 'address', f'{kernel_address}/31']
	)

	# source address of the tunnel interface should be configured
	# expect raise ConfigError
	with self.assertRaises(ConfigSessionError):
	self.cli_commit()

	self.cli_set(
	['interfaces', 'ethernet', interface, 'address', f'{source_address}/24']
	)

	# commit changes
	self.cli_commit()

	self.assertTrue(os.path.isdir(f'/sys/class/net/{interface_kernel}'))
	current_address = get_address(interface_kernel)
	self.assertEqual(kernel_address, current_address)

	# check gre interface
	_, out = rc_cmd('sudo vppctl show gre tunnel')
	required_str = f'[0] instance 12 src {source_address} dst {remote_address}'
	self.assertIn(required_str, out)

	# update gre interface
	self.cli_set(
	base_path
	+ ['interfaces', 'gre', interface_gre, 'source-address', new_source_address]
	)

	self.cli_set(
	['interfaces', 'ethernet', interface, 'address', f'{new_source_address}/24']
	)
	self.cli_commit()

	# check gre interface after update
	_, out = rc_cmd('sudo vppctl show gre tunnel')
	required_str = f'[0] instance 12 src {new_source_address} dst {remote_address}'
	self.assertIn(required_str, out)
	self.assertTrue(os.path.isdir(f'/sys/class/net/{interface_kernel}'))
	self.assertEqual(kernel_address, current_address)

	# delete gre kernel-interface but do not delete 'vpp kernel-interface'
	# expect raise ConfigError
	self.cli_delete(
	base_path
	+ ['interfaces', 'gre', interface_gre, 'kernel-interface', interface_kernel]
	)
	with self.assertRaises(ConfigSessionError):
	self.cli_commit()

	# update gre kernel-interface but do not change 'vpp kernel-interface'
	# expect raise ConfigError
	self.cli_set(
	base_path
	+ [
	'interfaces',
	'gre',
	interface_gre,
	'kernel-interface',
	new_interface_kernel,
	]
	)
	with self.assertRaises(ConfigSessionError):
	self.cli_commit()

	# delete kernel interface
	self.cli_delete(base_path + ['kernel-interfaces', interface_kernel])
	self.cli_commit()

	# delete gre kernel-interface
	self.cli_delete(
	base_path + ['interfaces', 'gre', interface_gre, 'kernel-interface']
	)
	self.cli_commit()
	self.assertFalse(os.path.isdir(f'/sys/class/net/{interface_kernel}'))

	# delete gre interface
	self.cli_delete(base_path + ['interfaces', 'gre', interface_gre])
	self.cli_commit()

	@unittest.skip('Skipping this test geneve index always is 0')
	def test_04_vpp_geneve(self):
	vni = '2'
	# Must be 'geneve0' to pass smoketest
	# As geneve interfaces cannot be named with "instance" suffix
	interface_geneve = 'geneve0'
	interface_kernel = f'vpptun{vni}'
	new_interface_kernel = f'vpptun1{vni}'
	source_address = '192.0.2.1'
	new_source_address = '192.0.2.2'
	remote_address = '203.0.113.10'
	kernel_address = '10.0.0.1'

	self.cli_set(
	base_path
	+ [
	'interfaces',
	'geneve',
	interface_geneve,
	'source-address',
	source_address,
	]
	)
	self.cli_set(
	base_path
	+ ['interfaces', 'geneve', interface_geneve, 'remote', remote_address]
	)
	self.cli_set(base_path + ['interfaces', 'geneve', interface_geneve, 'vni', vni])
	self.cli_set(
	base_path
	+ [
	'interfaces',
	'geneve',
	interface_geneve,
	'kernel-interface',
	interface_kernel,
	]
	)
	self.cli_set(
	base_path
	+ ['kernel-interfaces', interface_kernel, 'address', f'{kernel_address}/31']
	)

	# commit changes
	self.cli_commit()

	self.assertTrue(os.path.isdir(f'/sys/class/net/{interface_kernel}'))
	current_address = get_address(interface_kernel)
	self.assertEqual(kernel_address, current_address)

	# check geneve interface
	_, out = rc_cmd('sudo vppctl show geneve tunnel')
	required_str = f'[0] lcl {source_address} rmt {remote_address} vni {vni}'
	self.assertIn(required_str, out)

	# update geneve interface
	self.cli_set(
	base_path
	+ [
	'interfaces',
	'geneve',
	interface_geneve,
	'source-address',
	new_source_address,
	]
	)
	self.cli_commit()

	# check geneve interface after update
	_, out = rc_cmd('sudo vppctl show geneve tunnel')
	required_str = f'[0] lcl {new_source_address} rmt {remote_address} vni {vni}'
	self.assertIn(required_str, out)
	self.assertTrue(os.path.isdir(f'/sys/class/net/{interface_kernel}'))
	self.assertEqual(kernel_address, current_address)

	# delete geneve kernel-interface but do not delete 'vpp kernel-interface'
	# expect raise ConfigError
	self.cli_delete(
	base_path
	+ [
	'interfaces',
	'geneve',
	interface_geneve,
	'kernel-interface',
	interface_kernel,
	]
	)
	with self.assertRaises(ConfigSessionError):
	self.cli_commit()

	# update gemeve kernel-interface but do not change 'vpp kernel-interface'
	# expect raise ConfigError
	self.cli_set(
	base_path
	+ [
	'interfaces',
	'geneve',
	interface_geneve,
	'kernel-interface',
	new_interface_kernel,
	]
	)
	with self.assertRaises(ConfigSessionError):
	self.cli_commit()

	# delete vpp kernel-interface
	self.cli_delete(base_path + ['kernel-interfaces', interface_kernel])
	self.cli_commit()

	# delete geneve kernel-interface
	self.cli_delete(
	base_path + ['interfaces', 'geneve', interface_geneve, 'kernel-interface']
	)
	self.cli_commit()

	self.assertFalse(os.path.isdir(f'/sys/class/net/{interface_kernel}'))

	# delete geneve interface
	self.cli_set(base_path + ['interfaces', 'geneve', interface_geneve])
	self.cli_commit()

	def test_05_vpp_loopback(self):
	interface_loopback = 'lo11'
	interface_kernel = 'vpptun11'
	new_interface_kernel = 'vpptun12'
	kernel_address = '192.0.2.54'

	self.cli_set(base_path + ['interfaces', 'loopback', interface_loopback])
	self.cli_set(
	base_path
	+ [
	'interfaces',
	'loopback',
	interface_loopback,
	'kernel-interface',
	interface_kernel,
	]
	)
	self.cli_set(
	base_path
	+ ['kernel-interfaces', interface_kernel, 'address', f'{kernel_address}/25']
	)

	# commit changes
	self.cli_commit()

	self.assertTrue(os.path.isdir(f'/sys/class/net/{interface_kernel}'))

	current_address = get_address(interface_kernel)
	self.assertEqual(kernel_address, current_address)

	# check loopback interface
	_, out = rc_cmd('sudo vppctl show interface loop11')
	required_str = 'loop11'
	self.assertIn(required_str, out)

	# delete loopback kernel-interface but do not delete 'vpp kernel-interface'
	# expect raise ConfigError
	self.cli_delete(
	base_path
	+ [
	'interfaces',
	'loopback',
	interface_loopback,
	'kernel-interface',
	interface_kernel,
	]
	)
	with self.assertRaises(ConfigSessionError):
	self.cli_commit()

	# update loopback kernel-interface but do not change 'vpp kernel-interface'
	# expect raise ConfigError
	self.cli_set(
	base_path
	+ [
	'interfaces',
	'loopback',
	interface_loopback,
	'kernel-interface',
	new_interface_kernel,
	]
	)
	with self.assertRaises(ConfigSessionError):
	self.cli_commit()

	# delete vpp kernel-interface
	self.cli_delete(base_path + ['kernel-interfaces', interface_kernel])
	self.cli_commit()

	# delete loopback kernel-interface
	self.cli_delete(
	base_path
	+ ['interfaces', 'loopback', interface_loopback, 'kernel-interface']
	)
	self.cli_commit()
	self.assertFalse(os.path.isdir(f'/sys/class/net/{interface_kernel}'))

	# delete loopback interface
	self.cli_delete(base_path + ['interfaces', 'loopback', interface_loopback])
	self.cli_commit()

	def test_06_vpp_bonding(self):
	interface_bond = 'bond23'
	interface_kernel = 'vpptun23'
	hash = 'layer3+4'
	mode = '802.3ad'
	description = 'Interface-Bonding'
	vlans = ['123', '456']
	vlan_description = 'My-vlan-123'

	self.cli_set(
	base_path
	+ [
	'interfaces',
	'bonding',
	interface_bond,
	'member',
	'interface',
	interface,
	]
	)
	self.cli_set(
	base_path + ['interfaces', 'bonding', interface_bond, 'hash-policy', hash]
	)
	self.cli_set(
	base_path + ['interfaces', 'bonding', interface_bond, 'mode', mode]
	)

	# commit changes
	self.cli_commit()

	# Check for interface state "BondEthernet23 up"
	_, out = rc_cmd('sudo vppctl show interface')
	# Normalize the output for consistent whitespace
	normalized_out = re.sub(r'\s+', ' ', out)
	self.assertRegex(
	normalized_out,
	r'BondEthernet23\s+\d+\s+up',
	"Interface BondEthernet23 is not in the expected state 'up'.",
	)

	# set kernel interface
	self.cli_set(
	base_path
	+ [
	'interfaces',
	'bonding',
	interface_bond,
	'kernel-interface',
	interface_kernel,
	]
	)
	self.cli_set(
	base_path
	+ ['kernel-interfaces', interface_kernel, 'description', description]
	)
	for vlan in vlans:
	self.cli_set(
	base_path
	+ [
	'kernel-interfaces',
	interface_kernel,
	'vif',
	vlan,
	'description',
	vlan_description,
	]
	)

	# commit changes
	self.cli_commit()

	self.assertTrue(os.path.isdir(f'/sys/class/net/{interface_kernel}'))
	self.assertTrue(os.path.isdir(f'/sys/class/net/{interface_kernel}.{vlan}'))

	current_alias = read_file(f'/sys/class/net/{interface_kernel}/ifalias')
	vlan_alias = read_file(f'/sys/class/net/{interface_kernel}.{vlan}/ifalias')
	self.assertEqual(current_alias, description)
	self.assertEqual(vlan_alias, vlan_description)

	# check bonding interface
	_, out = rc_cmd('sudo vppctl show bond details')
	required_enries = (
	'BondEthernet23',
	'mode: lacp',
	'load balance: l34',
	'number of active members: 0',
	'number of members: 1',
	f'{interface}',
	'device instance: 0',
	'interface id: 23',
	)
	for entry in required_enries:
	self.assertIn(entry, out)

	# check interface state
	_, out = rc_cmd('sudo vppctl show interface')
	# Normalize the output for consistent whitespace
	normalized_out = re.sub(r'\s+', ' ', out)
	# Check for interface state "BondEthernet23 up"
	self.assertRegex(
	normalized_out,
	r'BondEthernet23\s+\d+\s+up',
	"Interface BondEthernet23 is not in the expected state 'up'.",
	)

	# delete vpp kernel-interface vlan
	self.cli_delete(base_path + ['kernel-interfaces', interface_kernel, 'vif'])
	self.cli_commit()
	self.assertFalse(os.path.isdir(f'/sys/class/net/{interface_kernel}.{vlan}'))

	# delete vpp kernel-interface
	self.cli_delete(base_path + ['kernel-interfaces', interface_kernel])
	self.cli_commit()

	# delete bonding kernel-interface
	self.cli_delete(
	base_path + ['interfaces', 'bonding', interface_bond, 'kernel-interface']
	)
	self.cli_commit()
	self.assertFalse(os.path.isdir(f'/sys/class/net/{interface_kernel}'))

	# delete bonding interface
	self.cli_delete(base_path + ['interfaces', 'bonding'])
	self.cli_commit()

	# check deleting bonding interface
	_, out = rc_cmd('sudo vppctl show interface')
	self.assertNotIn('BondEthernet23', out)

	def test_07_vpp_bridge(self):
	fake_member = 'eth2'
	members = [interface]
	interface_bridge = 'br10'
	vni = '23'
	interface_vxlan = f'vxlan{vni}'
	source_address = '192.0.2.1'
	remote_address = '192.0.2.254'

	self.cli_set(['interfaces', 'ethernet', interface, 'address', '192.0.2.1/24'])
	for member in members:
	self.cli_set(
	base_path
	+ [
	'interfaces',
	'bridge',
	interface_bridge,
	'member',
	'interface',
	member,
	]
	)

	# commit changes
	self.cli_commit()

	# check bridge interface
	_, out = rc_cmd('sudo vppctl show bridge-domain 10 detail')

	# Normalize the output for consistent whitespace
	normalized_out = re.sub(r'\s+', ' ', out)

	# Perform assertions based on the normalized output
	self.assertIn('BD-ID Index BSN Age(min)', normalized_out)
	self.assertIn('10 1 0 off', normalized_out)
	self.assertIn('Learning U-Forwrd UU-Flood Flooding', normalized_out)
	self.assertIn('on on flood on', normalized_out)
	self.assertIn('Interface If-idx ISN', normalized_out)
	# Check Interface, If-idx, ISN
	self.assertRegex(out, r'\s*eth1\s+\d+\s+\d+')

	# Set non exist member
	# expect raise ConfigError
	self.cli_set(
	base_path
	+ [
	'interfaces',
	'bridge',
	interface_bridge,
	'member',
	'interface',
	fake_member,
	]
	)
	with self.assertRaises(ConfigSessionError):
	self.cli_commit()

	self.cli_delete(
	base_path
	+ [
	'interfaces',
	'bridge',
	interface_bridge,
	'member',
	'interface',
	fake_member,
	]
	)

	# Add VXLAN to the bridge
	self.cli_set(
	base_path
	+ ['interfaces', 'vxlan', interface_vxlan, 'source-address', source_address]
	)
	self.cli_set(
	base_path
	+ ['interfaces', 'vxlan', interface_vxlan, 'remote', remote_address]
	)
	self.cli_set(base_path + ['interfaces', 'vxlan', interface_vxlan, 'vni', vni])
	self.cli_set(
	base_path
	+ [
	'interfaces',
	'bridge',
	interface_bridge,
	'member',
	'interface',
	interface_vxlan,
	]
	)

	# commit changes
	self.cli_commit()

	# check bridge interface
	_, out = rc_cmd('sudo vppctl show bridge-domain 10 detail')
	# Normalize the output for consistent whitespace
	normalized_out = re.sub(r'\s+', ' ', out)

	# Perform assertions based on the normalized output
	self.assertIn('BD-ID Index BSN Age(min)', normalized_out)
	self.assertIn('10 1 0 off', normalized_out)
	self.assertIn('Learning U-Forwrd UU-Flood Flooding', normalized_out)
	self.assertIn('on on flood on', normalized_out)
	self.assertIn('Interface If-idx ISN', normalized_out)
	# Check Interface, If-idx, ISN
	self.assertRegex(out, r'\s*eth1\s+\d+\s+\d+')
	self.assertRegex(out, r'\s*vxlan_tunnel23\s+\d+\s+\d+')

	# Add check dependency ethernet => bridge
	self.cli_set(
	base_path + ['settings', 'interface', interface, 'dpdk-options', 'promisc']
	)
	self.cli_commit()
	# check bridge interface
	_, out = rc_cmd('sudo vppctl show bridge-domain 10 detail')
	# Normalize the output for consistent whitespace
	normalized_out = re.sub(r'\s+', ' ', out)
	self.assertRegex(out, r'\s*eth1\s+\d+\s+\d+')
	self.assertRegex(out, r'\s*vxlan_tunnel23\s+\d+\s+\d+')

	# Add Loopback BVI to the bridge
	self.cli_set(base_path + ['interfaces', 'loopback', f'lo{vni}'])
	self.cli_set(
	base_path
	+ [
	'interfaces',
	'bridge',
	interface_bridge,
	'member',
	'interface',
	f'lo{vni}',
	'bvi',
	]
	)
	# commit changes
	self.cli_commit()

	# check bridge interface
	_, out = rc_cmd('sudo vppctl show bridge-domain 10 detail')
	# Normalize the output for consistent whitespace
	normalized_out = re.sub(r'\s+', ' ', out)

	self.assertIn('10 1 0 off', normalized_out)
	self.assertRegex(out, r'\bloop23\s+\d+\s+\d+\s+\d+\s+\*\s+')

	def test_08_vpp_ipip(self):
	interface_ipip = 'ipip12'
	interface_kernel = 'vpptun12'
	new_interface_kernel = 'vpptun123'
	source_address = '192.0.2.1'
	new_source_address = '192.0.2.2'
	remote_address = '192.0.2.5'
	kernel_address = '10.0.0.0'

	self.cli_set(['interfaces', 'ethernet', interface, 'address', '192.0.2.1/24'])
	self.cli_set(
	base_path
	+ ['interfaces', 'ipip', interface_ipip, 'source-address', source_address]
	)
	self.cli_set(
	base_path + ['interfaces', 'ipip', interface_ipip, 'remote', remote_address]
	)
	self.cli_set(
	base_path
	+ [
	'interfaces',
	'ipip',
	interface_ipip,
	'kernel-interface',
	interface_kernel,
	]
	)
	self.cli_set(
	base_path
	+ ['kernel-interfaces', interface_kernel, 'address', f'{kernel_address}/31']
	)

	# commit changes
	self.cli_commit()

	self.assertTrue(os.path.isdir(f'/sys/class/net/{interface_kernel}'))
	current_address = get_address(interface_kernel)
	self.assertEqual(kernel_address, current_address)

	# check ipip interface
	_, out = rc_cmd('sudo vppctl show ipip tunnel')
	required_str = f'[0] instance 12 src {source_address} dst {remote_address}'
	self.assertIn(required_str, out)

	# update ipip interface
	self.cli_set(
	base_path
	+ [
	'interfaces',
	'ipip',
	interface_ipip,
	'source-address',
	new_source_address,
	]
	)

	# source address of the tunnel interface should be configured
	# expect raise ConfigError
	with self.assertRaises(ConfigSessionError):
	self.cli_commit()

	self.cli_set(
	['interfaces', 'ethernet', interface, 'address', f'{new_source_address}/24']
	)
	self.cli_commit()

	# check ipip interface after update
	_, out = rc_cmd('sudo vppctl show ipip tunnel')
	required_str = f'[0] instance 12 src {new_source_address} dst {remote_address}'
	self.assertIn(required_str, out)
	self.assertTrue(os.path.isdir(f'/sys/class/net/{interface_kernel}'))
	self.assertEqual(kernel_address, current_address)

	# delete ipip kernel-interface but do not delete 'vpp kernel-interface'
	# expect raise ConfigError
	self.cli_delete(
	base_path
	+ [
	'interfaces',
	'ipip',
	interface_ipip,
	'kernel-interface',
	interface_kernel,
	]
	)
	with self.assertRaises(ConfigSessionError):
	self.cli_commit()

	# update ipip kernel-interface but do not change 'vpp kernel-interface'
	# expect raise ConfigError
	self.cli_set(
	base_path
	+ [
	'interfaces',
	'ipip',
	interface_ipip,
	'kernel-interface',
	new_interface_kernel,
	]
	)
	with self.assertRaises(ConfigSessionError):
	self.cli_commit()

	# delete kernel interface
	self.cli_delete(base_path + ['kernel-interfaces', interface_kernel])
	self.cli_commit()

	# delete ipip kernel-interface
	self.cli_delete(
	base_path + ['interfaces', 'ipip', interface_ipip, 'kernel-interface']
	)
	self.cli_commit()
	self.assertFalse(os.path.isdir(f'/sys/class/net/{interface_kernel}'))

	# delete ipip interface
	self.cli_delete(base_path + ['interfaces', 'ipip', interface_ipip])
	self.cli_commit()

	def test_09_vpp_xconnect(self):
	vni = '23'
	interface_vxlan = f'vxlan{vni}'
	interface_xconnect = f'xcon{vni}'
	source_address = '192.0.2.1'
	remote_address = '192.0.2.254'

	self.cli_set(['interfaces', 'ethernet', interface, 'address', '192.0.2.1/24'])
	self.cli_set(
	base_path
	+ ['interfaces', 'vxlan', interface_vxlan, 'source-address', source_address]
	)
	self.cli_set(
	base_path
	+ ['interfaces', 'vxlan', interface_vxlan, 'remote', remote_address]
	)
	self.cli_set(base_path + ['interfaces', 'vxlan', interface_vxlan, 'vni', vni])

	# Add xconneect
	self.cli_set(
	base_path
	+ [
	'interfaces',
	'xconnect',
	interface_xconnect,
	'member',
	'interface',
	interface,
	]
	)

	# Cross connect interfaces require 2 interfaces
	# expect raise ConfigError
	with self.assertRaises(ConfigSessionError):
	self.cli_commit()

	self.cli_set(
	base_path
	+ [
	'interfaces',
	'xconnect',
	interface_xconnect,
	'member',
	'interface',
	interface_vxlan,
	]
	)

	# commit changes
	self.cli_commit()

	# check interface mode
	_, out = rc_cmd('sudo vppctl show mode')
	required_str_list = [
	f'l2 xconnect {interface} vxlan_tunnel{vni}',
	f'l2 xconnect vxlan_tunnel{vni} {interface}',
	]
	for required_string in required_str_list:
	self.assertIn(required_string, out)

	# delete xconnect interface
	self.cli_delete(base_path + ['interfaces', 'xconnect', interface_xconnect])
	self.cli_commit()

	# check delete xconnect interface
	_, out = rc_cmd('sudo vppctl show mode')
	for required_string in required_str_list:
	self.assertNotIn(required_string, out)

	def test_10_vpp_driver_options(self):
	dpdk_options = {
	'num-rx-desc': '512',
	'num-tx-desc': '512',
	'num-rx-queues': '1',
	'num-tx-queues': '1',
	}
	main_core = '0'
	workers = '1'

	base_interface_path = base_path + ['settings', 'interface', interface]

	for option, value in dpdk_options.items():
	self.cli_set(base_interface_path + ['dpdk-options', option, value])

	# rx/tx queue configuration expect VPP workers to be set
	# expect raise ConfigError
	with self.assertRaises(ConfigSessionError):
	self.cli_commit()

	self.cli_set(base_path + ['settings', 'cpu', 'main-core', main_core])
	self.cli_set(base_path + ['settings', 'cpu', 'workers', workers])

	# DPDK driver expect only dpdk-options and not xdp-options to be set
	# expect raise ConfigError
	self.cli_set(base_interface_path + ['xdp-options', 'no-syscall-lock'])

	with self.assertRaises(ConfigSessionError):
	self.cli_commit()

	# delete xdp-options and apply commit
	self.cli_delete(base_interface_path + ['xdp-options'])
	self.cli_commit()

	# check dpdk options in config file
	config = read_file(VPP_CONF)

	for option, value in dpdk_options.items():
	self.assertIn(f'{option} {value}', config)

	def test_11_vpp_cpu_settings(self):
	main_core = '2'
	workers = '1'
	skip_cores = '1'

	self.cli_set(base_path + ['settings', 'cpu', 'workers', workers])

	# "cpu workers" reqiures main-core to be set
	# expect raise ConfigError
	with self.assertRaises(ConfigSessionError):
	self.cli_commit()

	self.cli_set(base_path + ['settings', 'cpu', 'main-core', main_core])

	self.cli_set(base_path + ['settings', 'cpu', 'skip-cores', '99'])

	# "cpu skip-cores" cannot be more than number of available CPUs - 1
	# expect raise ConfigError
	with self.assertRaises(ConfigSessionError):
	self.cli_commit()

	self.cli_set(base_path + ['settings', 'cpu', 'skip-cores', skip_cores])

	self.cli_commit()

	config_entries = (
	f'skip-cores {skip_cores}',
	f'main-core {main_core}',
	f'workers {workers}',
	'dev 0000:00:00.0',
	)

	# Check configured options
	config = read_file(VPP_CONF)
	for config_entry in config_entries:
	self.assertIn(config_entry, config)

	def test_12_vpp_cpu_corelist_workers(self):
	main_core = '0'
	corelist_workers = ['3']

	for worker in corelist_workers:
	self.cli_set(base_path + ['settings', 'cpu', 'corelist-workers', worker])

	# "cpu corelist-workers" reqiures main-core to be set
	# expect raise ConfigError
	with self.assertRaises(ConfigSessionError):
	self.cli_commit()

	self.cli_set(base_path + ['settings', 'cpu', 'main-core', main_core])

	# corelist-workers and workers cannot be used at the same time
	# expect raise ConfigError
	self.cli_set(base_path + ['settings', 'cpu', 'workers', '2'])
	with self.assertRaises(ConfigSessionError):
	self.cli_commit()
	self.cli_delete(base_path + ['settings', 'cpu', 'workers'])

	# verify corelist-workers are set not correctly
	# expect raise ConfigError
	self.cli_set(base_path + ['settings', 'cpu', 'corelist-workers', '99-101'])
	with self.assertRaises(ConfigSessionError):
	self.cli_commit()

	self.cli_delete(base_path + ['settings', 'cpu', 'corelist-workers', '99-101'])

	self.cli_commit()

	config_entries = (
	f'main-core {main_core}',
	f'corelist-workers {",".join(corelist_workers)}',
	'dev 0000:00:00.0',
	)

	# Check configured options
	config = read_file(VPP_CONF)
	for config_entry in config_entries:
	self.assertIn(config_entry, config)

	def test_13_1_buffer_page_size(self):
	sizes = ['4K', '2M']
	for size in sizes:
	self.cli_set(base_path + ['settings', 'buffers', 'page-size', size])
	self.cli_commit()

	conf = get_vpp_config()
	self.assertEqual(conf['buffers']['page-size'], size)

	def test_13_2_statseg_page_size(self):
	sizes = ['4K', '2M']
	for size in sizes:
	self.cli_set(base_path + ['settings', 'statseg', 'page-size', size])
	self.cli_commit()

	conf = get_vpp_config()
	self.assertEqual(conf['statseg']['page-size'], size)

	def test_13_3_mem_page_size(self):
	sizes = ['4K', '2M']
	for size in sizes:
	self.cli_set(
	base_path + ['settings', 'memory', 'main-heap-page-size', size]
	)
	self.cli_commit()

	conf = get_vpp_config()
	self.assertEqual(conf['memory']['main-heap-page-size'], size)

	def test_14_vpp_ipsec_xfrm_nl(self):
	base_ipsec = base_path + ['settings', 'ipsec']
	batch_delay = '250'
	batch_size = '150'
	rx_buffer_zise = '1024'

	self.cli_set(base_ipsec + ['netlink', 'batch-delay-ms', batch_delay])
	self.cli_set(base_ipsec + ['netlink', 'batch-size', batch_size])
	self.cli_set(base_ipsec + ['netlink', 'rx-buffer-size', rx_buffer_zise])
	self.cli_commit()

	config_entries = (
	'linux-xfrm-nl',
	'enable-route-mode-ipsec',
	'interface ipsec',
	f'nl-batch-delay-ms {batch_delay}',
	f'nl-batch-size {batch_size}',
	f'nl-rx-buffer-size {rx_buffer_zise}',
	)

	# Check configured options
	config = read_file(VPP_CONF)
	for config_entry in config_entries:
	self.assertIn(config_entry, config)

	# set IPsec tunnel-type ipip
	self.cli_set(base_ipsec + ['interface-type', 'ipip'])
	self.cli_commit()

	config = read_file(VPP_CONF)
	self.assertIn('interface ipip', config)

	- def test_15_vpp_cgnat(self):
	+ def test_15_1_vpp_cgnat(self):
	base_cgnat = base_path + ['nat', 'cgnat']
	iface_out = 'eth0'
	iface_inside = 'eth1'
	timeout_udp = '150'
	timeout_icmp = '30'
	timeout_tcp_est = '600'
	timeout_tcp_trans = '120'
	inside_prefix = '100.64.0.0/24'
	outside_prefix = '192.0.2.1/32'

	self.cli_set(base_path + ['settings', 'interface', iface_out, 'driver', driver])
	self.cli_set(base_cgnat + ['interface', 'inside', iface_inside])
	self.cli_set(base_cgnat + ['interface', 'outside', iface_out])
	self.cli_set(base_cgnat + ['rule', '100', 'inside-prefix', inside_prefix])
	self.cli_set(base_cgnat + ['rule', '100', 'outside-prefix', outside_prefix])
	self.cli_set(base_cgnat + ['timeout', 'icmp', timeout_icmp])
	self.cli_set(base_cgnat + ['timeout', 'tcp-established', timeout_tcp_est])
	self.cli_set(base_cgnat + ['timeout', 'tcp-transitory', timeout_tcp_trans])
	self.cli_set(base_cgnat + ['timeout', 'udp', timeout_udp])
	self.cli_commit()

	# Check interfaces
	_, out = rc_cmd('sudo vppctl show det44 interfaces')
	self.assertIn(f'{iface_inside} in', out)
	self.assertIn(f'{iface_out} out', out)

	# Check mappings
	_, out = rc_cmd('sudo vppctl show det44 mappings')
	self.assertIn(inside_prefix, out)
	self.assertIn(outside_prefix, out)

	# Check timeouts
	_, out = rc_cmd('sudo vppctl show det44 timeouts')
	self.assertIn(f'udp timeout: {timeout_udp}sec', out)
	self.assertIn(f'tcp established timeout: {timeout_tcp_est}sec', out)
	self.assertIn(f'tcp transitory timeout: {timeout_tcp_trans}sec', out)
	self.assertIn(f'icmp timeout: {timeout_icmp}sec', out)

	+ def test_15_2_vpp_cgnat_bond_with_vifs(self):
	+ base_cgnat = base_path + ['nat', 'cgnat']
	+ base_kernel = base_path + ['kernel-interfaces']
	+ base_bond = base_path + ['interfaces', 'bonding']
	+ iface_kernel = 'vpptun0'
	+ iface_bond = 'bond0'
	+ vif_1 = '23'
	+ vif_2 = '24'
	+ iface_out = f'{iface_bond}.{vif_1}'
	+ iface_inside = f'{iface_bond}.{vif_2}'
	+ address_1 = '100.64.0.23/32'
	+ address_2 = '192.0.2.1/32'
	+
	+ self.cli_set(base_bond + [iface_bond, 'kernel-interface', iface_kernel])
	+ self.cli_set(base_bond + [iface_bond, 'member', 'interface', interface])
	+
	+ self.cli_set(base_kernel + [iface_kernel, 'vif', vif_1, 'address', address_1])
	+ self.cli_set(base_kernel + [iface_kernel, 'vif', vif_2, 'address', address_2])
	+
	+ self.cli_set(base_cgnat + ['interface', 'inside', iface_inside])
	+ self.cli_set(base_cgnat + ['interface', 'outside', iface_out])
	+ self.cli_set(base_cgnat + ['rule', '100', 'inside-prefix', address_1])
	+ self.cli_set(base_cgnat + ['rule', '100', 'outside-prefix', address_2])
	+ self.cli_commit()
	+
	+ # Check interfaces
	+ _, out = rc_cmd('sudo vppctl show det44 interfaces')
	+ self.assertIn(f'BondEthernet0.{vif_2} in', out)
	+ self.assertIn(f'BondEthernet0.{vif_1} out', out)
	+
	+ # Change bonding interface configuration
	+ self.cli_set(base_bond + [iface_bond, 'mode', '802.3ad'])
	+ self.cli_commit()
	+
	+ # Check interfaces
	+ _, out = rc_cmd('sudo vppctl show det44 interfaces')
	+ self.assertIn(f'BondEthernet0.{vif_2} in', out)
	+ self.assertIn(f'BondEthernet0.{vif_1} out', out)
	+
	+ # Verify only expected interfaces are shown:
	+ # header + inside + outside = 3 lines total
	+ lines = out.split('\n')
	+ self.assertTrue(len(lines) == 3)
	+
	def test_16_vpp_nat(self):
	base_nat = base_path + ['nat44']
	base_nat_settings = base_path + ['settings', 'nat44']
	exclude_local_addr = '100.64.0.52'
	exclude_local_port = '22'
	iface_out = 'eth0'
	iface_inside = 'eth1'
	timeout_udp = '150'
	timeout_icmp = '30'
	timeout_tcp_est = '600'
	timeout_tcp_trans = '120'
	translation_pool = '192.0.2.1-192.0.2.2'
	static_ext_addr = '192.0.2.55'
	static_local_addr = '100.64.0.55'
	sess_limit = '64000'

	self.cli_set(base_path + ['settings', 'interface', iface_out, 'driver', driver])
	self.cli_set(base_nat + ['interface', 'inside', iface_inside])
	self.cli_set(base_nat + ['interface', 'outside', iface_out])
	self.cli_set(
	base_nat + ['address-pool', 'translation', 'address', translation_pool]
	)
	self.cli_set(
	base_nat + ['exclude', 'rule', '100', 'local-address', exclude_local_addr]
	)
	self.cli_set(
	base_nat + ['exclude', 'rule', '100', 'local-port', exclude_local_port]
	)

	# cannot set local-port without specifying protocol
	# expect raise ConfigError
	with self.assertRaises(ConfigSessionError):
	self.cli_commit()

	self.cli_set(base_nat + ['exclude', 'rule', '100', 'protocol', 'tcp'])
	self.cli_set(
	base_nat + ['static', 'rule', '100', 'external', 'address', static_ext_addr]
	)
	self.cli_set(
	base_nat + ['static', 'rule', '100', 'local', 'address', static_local_addr]
	)

	self.cli_set(base_nat_settings + ['no-forwarding'])
	self.cli_set(base_nat_settings + ['session-limit', sess_limit])
	self.cli_set(base_nat_settings + ['timeout', 'icmp', timeout_icmp])
	self.cli_set(
	base_nat_settings + ['timeout', 'tcp-established', timeout_tcp_est]
	)
	self.cli_set(
	base_nat_settings + ['timeout', 'tcp-transitory', timeout_tcp_trans]
	)
	self.cli_set(base_nat_settings + ['timeout', 'udp', timeout_udp])
	self.cli_commit()

	# Check addresses
	_, out = rc_cmd('sudo vppctl show nat44 addresses')
	self.assertIn(translation_pool.split('-')[0], out)
	self.assertIn(translation_pool.split('-')[1], out)

	# Check interfaces
	_, out = rc_cmd('sudo vppctl show nat44 interfaces')
	self.assertIn(f'{iface_inside} in', out)
	self.assertIn(f'{iface_out} out', out)

	# Check mappings
	_, out = rc_cmd('sudo vppctl show nat44 static mappings')
	self.assertIn(
	f'local {static_local_addr} external {static_ext_addr} vrf 0', out
	)
	self.assertIn(f'{exclude_local_addr}:{exclude_local_port} vrf 0', out)

	# Check timeouts
	_, out = rc_cmd('sudo vppctl show nat timeouts')
	self.assertIn(f'udp timeout: {timeout_udp}sec', out)
	self.assertIn(f'tcp-established timeout: {timeout_tcp_est}sec', out)
	self.assertIn(f'tcp-transitory timeout: {timeout_tcp_trans}sec', out)
	self.assertIn(f'icmp timeout: {timeout_icmp}sec', out)

	# Summary
	_, out = rc_cmd('sudo vppctl show nat44 summary')
	self.assertIn(f'max translations per thread: {sess_limit} fib 0', out)

	def test_17_vpp_sflow(self):
	base_sflow = ['system', 'sflow']

	self.cli_set(base_path + ['sflow', 'interface', interface])
	self.cli_set(base_sflow + ['interface', interface])
	self.cli_set(base_sflow + ['server', '127.0.0.1'])
	self.cli_set(base_sflow + ['vpp'])
	self.cli_commit()

	# Check sFlow
	_, out = rc_cmd('sudo vppctl show sflow')

	expected_entries = (
	'sflow sampling-direction ingress',
	f'sflow enable {interface}',
	'interfaces enabled: 1',
	)

	for expected_entry in expected_entries:
	self.assertIn(expected_entry, out)

	self.cli_delete(base_sflow)
	self.cli_commit()

	def test_18_resource_limits(self):
	max_map_count = '100000'
	shmmax = '55555555555555'
	hr_path = ['system', 'option', 'resource-limits']

	# Check if max-map-count has default auto calculated value
	# but not less than '65530'
	self.assertEqual(sysctl_read('vm.max_map_count'), '65530')
	# The same is with: kernel.shmmax = '8589934592'
	self.assertEqual(sysctl_read('kernel.shmmax'), '8589934592')

	# Change max-map-count, shmmax and check
	self.cli_set(hr_path + ['max-map-count', max_map_count])
	self.cli_set(hr_path + ['shmmax', shmmax])
	self.cli_commit()

	self.assertEqual(sysctl_read('vm.max_map_count'), max_map_count)
	self.assertEqual(sysctl_read('kernel.shmmax'), shmmax)

	# We expect max-map-count and shmmax will return auto calculated values
	self.cli_delete(hr_path + ['max-map-count'])
	self.cli_delete(hr_path + ['shmmax'])
	self.cli_commit()

	self.assertEqual(sysctl_read('vm.max_map_count'), '65530')
	self.assertEqual(sysctl_read('kernel.shmmax'), '8589934592')

	def test_19_vpp_pppoe_mapping(self):
	config_file = '/run/accel-pppd/pppoe.conf'
	pool = "TEST-POOL"
	vni = '23'
	pppoe_base = ['service', 'pppoe-server']

	self.cli_set(['interfaces', 'ethernet', interface, 'vif', vni])

	# Basic pppoe-server config
	self.cli_set(pppoe_base + ['authentication', 'mode', 'noauth'])
	self.cli_set(pppoe_base + ['gateway-address', '192.0.2.1'])
	self.cli_set(pppoe_base + ['client-ip-pool', pool, 'range', '192.0.2.0/24'])
	self.cli_set(pppoe_base + ['default-pool', pool])

	# Enable PPPoE control-plane integration with VPP
	self.cli_set(pppoe_base + ['interface', interface, 'vpp-cp'])
	self.cli_set(pppoe_base + ['interface', f'{interface}.{vni}', 'vpp-cp'])

	self.cli_commit()

	# Validate configuration values
	config = read_file(config_file)

	# Validate configuration
	self.assertIn(f'interface={interface},vpp-cp=true', config)
	self.assertIn(f'interface={interface}.{vni},vpp-cp=true', config)

	# Check pppoe mapping
	_, out = rc_cmd('sudo vppctl show pppoe control-plane binding')
	self.assertRegex(out, rf'{interface}\s+tap4096')
	self.assertRegex(out, rf'{interface}.{vni}\s+tap4096.23')

	# check if dependency is called and mapping is correct after changes in vpp script
	self.cli_set(
	base_path + ['settings', 'interface', interface, 'dpdk-options', 'promisc']
	)
	self.cli_commit()

	# Check pppoe mapping
	_, out = rc_cmd('sudo vppctl show pppoe control-plane binding')
	self.assertRegex(out, rf'{interface}\s+tap4096')
	self.assertRegex(out, rf'{interface}.{vni}\s+tap4096.23')

	# delete PPPoE config
	self.cli_delete(pppoe_base)

	# delete vif Ethernet interface
	self.cli_delete(['interfaces', 'ethernet', interface, 'vif'])
	self.cli_commit()

	def test_20_kernel_options_hugepages(self):
	default_hp_size = '2M'
	hp_size_1g = '1G'
	hp_size_2m = '2M'
	hp_count_1g = '2'
	hp_count_2m = '512'
	memory_path = ['system', 'option', 'kernel', 'memory']

	self.cli_set(memory_path + ['default-hugepage-size', default_hp_size])
	self.cli_set(
	memory_path + ['hugepage-size', hp_size_2m, 'hugepage-count', hp_count_2m]
	)
	self.cli_set(
	memory_path + ['hugepage-size', hp_size_1g, 'hugepage-count', '2000']
	)
	# very big number of 1G hugepages, not enough memory for configuring them
	# expect raise ConfigError
	with self.assertRaises(ConfigSessionError):
	self.cli_commit()

	self.cli_set(
	memory_path + ['hugepage-size', hp_size_1g, 'hugepage-count', hp_count_1g]
	)
	self.cli_commit()

	# Read GRUB config file for current running image
	tmp = read_file(
	f'{image.grub.GRUB_DIR_VYOS_VERS}/{image.get_running_image()}.cfg'
	)
	self.assertIn(f' default_hugepagesz={default_hp_size}', tmp)
	self.assertIn(f' hugepagesz={hp_size_1g} hugepages={hp_count_1g}', tmp)
	self.assertIn(f' hugepagesz={hp_size_2m} hugepages={hp_count_2m}', tmp)


	if __name__ == '__main__':
	unittest.main(verbosity=2, failfast=VyOSUnitTestSHIM.TestCase.debug_on())
	diff --git a/src/conf_mode/vpp.py b/src/conf_mode/vpp.py
	index 81502c0b4..2d8c17e4f 100755
	--- a/src/conf_mode/vpp.py
	+++ b/src/conf_mode/vpp.py
	@@ -1,795 +1,800 @@
	#!/usr/bin/env python3
	#
	# 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 pyroute2.iproute import IPRoute

	try:
	from vpp_papi import VPPIOError, VPPValueError
	except ImportError: # pylint: disable=import-error
	VPPIOError = VPPValueError = None

	from vyos import ConfigError
	from vyos import airbag
	from vyos.base import Warning
	from vyos.config import Config, config_dict_merge
	from vyos.configdep import set_dependents, call_dependents
	from vyos.configdict import node_changed, leaf_node_changed
	from vyos.ifconfig import Section
	from vyos.logger import getLogger
	from vyos.template import render
	from vyos.utils.boot import boot_configuration_complete
	from vyos.utils.kernel import check_kmod
	from vyos.utils.kernel import unload_kmod
	from vyos.utils.kernel import list_loaded_modules
	from vyos.utils.process import call

	from vyos.vpp import VPPControl
	from vyos.vpp import control_host
	from vyos.vpp.config_deps import deps_xconnect_dict
	from vyos.vpp.config_verify import (
	verify_dev_driver,
	verify_vpp_minimum_cpus,
	verify_vpp_minimum_memory,
	verify_vpp_settings_cpu,
	verify_vpp_settings_cpu_corelist_workers,
	verify_vpp_cpu_main_core,
	verify_vpp_settings_cpu_skip_cores,
	verify_vpp_settings_cpu_workers,
	verify_vpp_nat44_workers,
	verify_vpp_memory,
	verify_vpp_statseg_size,
	verify_vpp_interfaces_dpdk_num_queues,
	verify_routes_count,
	verify_vpp_main_heap_size,
	verify_vpp_buffers,
	)
	from vyos.vpp.config_resource_checks import memory
	from vyos.vpp.config_filter import iface_filter_eth
	from vyos.vpp.utils import EthtoolGDrvinfo
	from vyos.vpp.configdb import JSONStorage

	airbag.enable()

	service_name = 'vpp'
	service_conf = Path(f'/run/vpp/{service_name}.conf')
	systemd_override = '/run/systemd/system/vpp.service.d/10-override.conf'

	vpp_log = getLogger(
	service_name, format='%(filename)s[%(process)d]: %(message)s', address='/dev/log'
	)

	dependency_interface_type_map = {
	'vpp_interfaces_bonding': 'bonding',
	'vpp_interfaces_bridge': 'bridge',
	'vpp_interfaces_ethernet': 'ethernet',
	'vpp_interfaces_geneve': 'geneve',
	'vpp_interfaces_gre': 'gre',
	'vpp_interfaces_ipip': 'ipip',
	'vpp_interfaces_loopback': 'loopback',
	'vpp_interfaces_vxlan': 'vxlan',
	'vpp_interfaces_xconnect': 'xconnect',
	}

	# dict of drivers that needs to be overrided
	override_drivers: dict[str, str] = {
	'hv_netvsc': 'uio_hv_generic',
	'ena': 'vfio-pci',
	}

	# drivers that does not use PCIe addresses
	not_pci_drv: list[str] = ['hv_netvsc']

	# drivers that support interrupt RX mode for DPDK and XDP
	drivers_support_interrupt: dict[str, list] = {
	'atlantic': ['dpdk', 'xdp'],
	'bnx2x': ['dpdk'],
	'e1000': ['dpdk'],
	'ena': ['dpdk', 'xdp'],
	'i40e': ['dpdk', 'xdp'],
	'ice': ['dpdk', 'xdp'],
	'igb': ['xdp'],
	'igc': ['dpdk', 'xdp'],
	'ixgbe': ['dpdk', 'xdp'],
	'ixgbevf': ['dpdk'],
	'qede': ['dpdk', 'xdp'],
	'vmxnet3': ['xdp'],
	'virtio_net': ['xdp'],
	}


	def _load_module(module_name: str):
	"""
	Load a kernel module

	Args:
	module_name (str): Name of the module to load.
	"""
	if module_name in list_loaded_modules():
	vpp_log.info(f"Module '{module_name}' is alrady loaded")
	return
	try:
	check_kmod(module_name)
	vpp_log.info(f"Module '{module_name}' loaded successfully")
	except Exception as e:
	vpp_log.error(f"Failed to load module '{module_name}': {e}")
	raise


	def _unload_module(module_name: str):
	"""
	Unload a kernel module

	Args:
	module_name (str): Name of the module to unload.
	"""
	if module_name not in list_loaded_modules():
	vpp_log.info(f"Module '{module_name}' is not loaded")
	return
	try:
	unload_kmod(module_name)
	vpp_log.info(f"Module '{module_name}' unloaded successfully")
	except Exception as e:
	vpp_log.error(f"Failed to unload module '{module_name}': {e}")
	raise


	def _get_workers_count(cpu_settings: dict) -> int:
	if 'corelist_workers' in cpu_settings:
	corelist_workers = []
	for worker_range in cpu_settings['corelist_workers']:
	core_numbers = worker_range.split('-')
	corelist_workers.extend(
	range(int(core_numbers[0]), int(core_numbers[-1]) + 1)
	)

	return len(corelist_workers)

	return int(cpu_settings.get('workers', 0))


	def _normalize_buffers(config: dict):
	"""Replace 'auto' buffers_per_numa with calculated value"""
	if config['settings']['buffers']['buffers_per_numa'] == 'auto':
	workers = _get_workers_count(config['settings'].get('cpu', {}))
	buffers = memory.buffers_required(config['settings'], workers)
	config['settings']['buffers']['buffers_per_numa'] = str(buffers)


	def get_config(config=None):
	# use persistent config to store interfaces data between executions
	# this is required because some interfaces after they are connected
	# to VPP is really hard or impossible to restore without knowing
	# their original parameters (like IDs)
	persist_config = JSONStorage('vpp_conf')
	eth_ifaces_persist: dict[str, dict[str, str]] = persist_config.read(
	'eth_ifaces', {}
	)

	if config:
	conf = config
	else:
	conf = Config()

	base = ['vpp']
	base_settings = ['vpp', 'settings']

	# find interfaces removed from VPP
	effective_config = conf.get_config_dict(
	base,
	key_mangling=('-', '_'),
	effective=True,
	get_first_key=True,
	no_tag_node_value_mangle=True,
	)

	xconn_members = deps_xconnect_dict(conf)

	removed_ifaces = []
	tmp = node_changed(conf, base_settings + ['interface'])
	if tmp:
	for removed_iface in tmp:
	to_append = {
	'iface_name': removed_iface,
	'driver': effective_config['settings']['interface'][removed_iface][
	'driver'
	],
	}
	removed_ifaces.append(to_append)
	# add an interface to a list of interfaces that need
	# to be reinitialized after the commit
	set_dependents('ethernet', conf, removed_iface)

	# Get interfaces that are used in PPPoe for control-plane integration
	pppoe_conf = conf.get_config_dict(
	['service', 'pppoe-server'],
	key_mangling=('-', '_'),
	get_first_key=True,
	no_tag_node_value_mangle=True,
	)
	pppoe_map_ifaces = [
	ifname
	for ifname, iface_conf in pppoe_conf.get('interface', {}).items()
	if 'vpp_cp' in iface_conf
	]

	if not conf.exists(base):
	return {
	'removed_ifaces': removed_ifaces,
	'xconn_members': xconn_members,
	'persist_config': eth_ifaces_persist,
	**({'pppoe_ifaces': pppoe_map_ifaces} if pppoe_map_ifaces else {}),
	}

	config = conf.get_config_dict(
	base,
	get_first_key=True,
	key_mangling=('-', '_'),
	no_tag_node_value_mangle=True,
	)

	# Get default values which we need to conditionally update into the
	# dictionary retrieved.
	default_values = conf.get_config_defaults(**config.kwargs, recursive=True)

	# delete driver-incompatible defaults
	for iface, iface_config in config.get('settings', {}).get('interface', {}).items():
	if iface_config.get('driver') == 'dpdk':
	del default_values['settings']['interface'][iface]['xdp_options']
	elif iface_config.get('driver') == 'xdp':
	del default_values['settings']['interface'][iface]['dpdk_options']

	config = config_dict_merge(default_values, config)

	# Ignore default XML values if config doesn't exists
	if not conf.exists(base_settings + ['ipsec']):
	del config['settings']['ipsec']

	# add running config
	if effective_config:
	default_values_effective = conf.get_config_defaults(
	**effective_config.kwargs, recursive=True
	)
	effective_config = config_dict_merge(default_values_effective, effective_config)
	# Buffer normalization (auto → computed)
	_normalize_buffers(effective_config)
	config['effective'] = effective_config

	if 'settings' in config:
	if 'interface' in config['settings']:
	for iface, iface_config in config['settings']['interface'].items():
	# Driver must be configured to continue
	if 'driver' not in iface_config:
	raise ConfigError(
	f'"driver" must be configured for {iface} interface!'
	)

	old_driver = leaf_node_changed(
	conf, base_settings + ['interface', iface, 'driver']
	)

	# Get current kernel module, required for extra verification and
	# logic for VMBus interfaces
	config['settings']['interface'][iface]['kernel_module'] = (
	EthtoolGDrvinfo(iface).driver
	)

	# filter unsupported config nodes
	iface_filter_eth(conf, iface)
	set_dependents('ethernet', conf, iface)
	# Interfaces with changed driver should be removed/readded
	if old_driver and old_driver[0] == 'dpdk':
	removed_ifaces.append(
	{
	'iface_name': iface,
	'driver': 'dpdk',
	}
	)

	# Get PCI address or device ID
	if iface_config['driver'] == 'dpdk':
	if 'dpdk_options' not in iface_config:
	iface_config['dpdk_options'] = {}
	# Check in a persistent config first
	id_from_persisten_conf = eth_ifaces_persist.get(iface, {}).get(
	'dev_id'
	)
	if id_from_persisten_conf:
	iface_config['dpdk_options']['dev_id'] = id_from_persisten_conf
	else:
	try:
	iface_to_search = iface
	if old_driver and old_driver[0] == 'xdp':
	iface_to_search = f'defunct_{iface}'
	iface_config['dpdk_options']['dev_id'] = (
	control_host.get_dev_id(iface_to_search)
	)
	except Exception:
	# Return empty address if all attempts failed
	# We will catch this in verify()
	iface_config['dpdk_options']['dev_id'] = ''
	# prepare XDP interface parameters
	if iface_config['driver'] == 'xdp':
	xdp_api_params = {
	'rxq_size': int(iface_config['xdp_options']['rx_queue_size']),
	'txq_size': int(iface_config['xdp_options']['tx_queue_size']),
	}
	if iface_config['xdp_options']['num_rx_queues'] == 'all':
	xdp_api_params['rxq_num'] = 0
	else:
	xdp_api_params['rxq_num'] = int(
	iface_config['xdp_options']['num_rx_queues']
	)
	if 'zero-copy' in iface_config['xdp_options']:
	xdp_api_params['mode'] = 'zero-copy'
	if 'zero-copy' in iface_config['xdp_options']:
	xdp_api_params['flags'] = 'no_syscall_lock'
	iface_config['xdp_api_params'] = xdp_api_params

	# Buffer normalization (auto → computed)
	_normalize_buffers(config)

	if removed_ifaces:
	config['removed_ifaces'] = removed_ifaces
	config['xconn_members'] = xconn_members

	# Dependencies
	for dependency, interface_type in dependency_interface_type_map.items():
	# if conf.exists(base + ['interfaces', interface_type]):
	if effective_config.get('interfaces', {}).get(interface_type):
	for iface, iface_config in (
	config.get('interfaces', {}).get(interface_type, {}).items()
	):
	# filter unsupported config nodes
	if interface_type == 'ethernet':
	iface_filter_eth(conf, iface)
	set_dependents(dependency, conf, iface)

	# Save important info about all interfaces that cannot be retrieved later
	# Add new interfaces (only if they are first time seen in a config)
	for iface, iface_config in config.get('settings', {}).get('interface', {}).items():
	if iface not in effective_config.get('settings', {}).get('interface', {}):
	eth_ifaces_persist[iface] = {
	'original_driver': config['settings']['interface'][iface][
	'kernel_module'
	],
	}
	eth_ifaces_persist[iface]['bus_id'] = control_host.get_bus_name(iface)
	eth_ifaces_persist[iface]['dev_id'] = control_host.get_dev_id(iface)

	+ # kernel-interfaces dependency
	+ if effective_config.get('kernel_interfaces'):
	+ for iface in config.get('kernel_interfaces', {}):
	+ set_dependents('vpp_kernel_interface', conf, iface)
	+
	# NAT dependency
	if conf.exists(['vpp', 'nat44']):
	set_dependents('vpp_nat', conf)
	if conf.exists(['vpp', 'nat', 'cgnat']):
	set_dependents('vpp_nat_cgnat', conf)

	# sFlow dependency
	if conf.exists(['vpp', 'sflow']):
	set_dependents('vpp_sflow', conf)

	# ACL dependency
	if conf.exists(['vpp', 'acl']):
	set_dependents('vpp_acl', conf)

	# PPPoE dependency
	if pppoe_map_ifaces:
	config['pppoe_ifaces'] = pppoe_map_ifaces
	set_dependents('pppoe_server', conf)

	# Return to config dictionary
	config['persist_config'] = eth_ifaces_persist

	return config


	def verify(config):
	# Cannot remove interface if PPPoE control-plane is still enabled
	removed_ifaces = [iface['iface_name'] for iface in config.get('removed_ifaces', [])]
	pppoe_removed_ifaces = [
	p
	for p in config.get('pppoe_ifaces', [])
	for r in removed_ifaces
	if p == r or p.startswith(f'{r}.')
	]
	if pppoe_removed_ifaces:
	raise ConfigError(
	f'{", ".join(pppoe_removed_ifaces)} still in use by the PPPoE server. '
	'Disable PPPoE control-plane integration with VPP before proceeding.'
	)

	# bail out early - looks like removal from running config
	if not config or ('removed_ifaces' in config and 'settings' not in config):
	return None

	if 'settings' not in config:
	raise ConfigError('"settings interface" is required but not set!')

	if 'interface' not in config['settings']:
	raise ConfigError('"settings interface" is required but not set!')

	# check if the system meets minimal requirements
	verify_vpp_minimum_memory()

	# check if Ethernet interfaces exist
	ethernet_ifaces = Section.interfaces('ethernet')
	for iface in config['settings']['interface'].keys():
	if iface not in ethernet_ifaces:
	raise ConfigError(f'Interface {iface} does not exist or is not Ethernet!')

	# Resource usage checks
	cpu_settings = config['settings'].get('cpu', {})

	if not any(key in cpu_settings for key in ('workers', 'corelist_workers')):
	verify_vpp_minimum_cpus()

	if 'cpu' in config['settings']:
	# Check whether the workers and corelist-workers are configured properly
	verify_vpp_settings_cpu(cpu_settings)

	# Check if there are enough CPU cores to skip according to config
	if 'skip_cores' in cpu_settings:
	skip_cores = int(cpu_settings['skip_cores'])
	verify_vpp_settings_cpu_skip_cores(skip_cores)

	# Check if there are enough CPU cores to add workers
	if 'workers' in cpu_settings:
	verify_vpp_settings_cpu_workers(cpu_settings)

	if 'main_core' in cpu_settings:
	verify_vpp_cpu_main_core(cpu_settings)

	# Check the CPU main core not falling to the corelist-workers
	if 'corelist_workers' in cpu_settings:
	verify_vpp_settings_cpu_corelist_workers(cpu_settings)

	workers = _get_workers_count(config['settings'].get('cpu', {}))

	if 'workers' in config['settings']['nat44']:
	verify_vpp_nat44_workers(
	workers=workers, nat44_workers=config['settings']['nat44']['workers']
	)

	verify_vpp_main_heap_size(config['settings'])
	verify_vpp_statseg_size(config['settings'])

	# Check buffers
	verify_vpp_buffers(config['settings'], workers)

	# Check if available memory is enough for current VPP config
	verify_vpp_memory(config)

	# ensure DPDK/XDP settings are properly configured
	for iface, iface_config in config['settings']['interface'].items():
	# check if selected driver is supported, but only for new interfaces
	if iface not in config.get('effective', {}).get('settings', {}).get(
	'interface', {}
	):
	if not verify_dev_driver(iface, iface_config['driver']):
	raise ConfigError(
	f'Driver {iface_config["driver"]} is not compatible with interface {iface}!'
	)
	if iface_config['driver'] == 'xdp' and 'xdp_options' in iface_config:
	if iface_config['xdp_options']['num_rx_queues'] != 'all':
	Warning(f'Not all RX queues will be connected to VPP for {iface}!')

	if iface_config['driver'] == 'xdp' and 'dpdk_options' in iface_config:
	raise ConfigError('DPDK options are not applicable for XDP driver!')

	if iface_config['driver'] == 'dpdk' and 'xdp_options' in iface_config:
	raise ConfigError('XDP options are not applicable for DPDK driver!')

	if iface_config['driver'] == 'dpdk' and 'dpdk_options' in iface_config:
	if 'num_rx_queues' in iface_config['dpdk_options']:
	rx_queues = int(iface_config['dpdk_options']['num_rx_queues'])
	verify_vpp_interfaces_dpdk_num_queues(
	qtype='receive', num_queues=rx_queues, workers=workers
	)

	if 'num_tx_queues' in iface_config['dpdk_options']:
	tx_queues = int(iface_config['dpdk_options']['num_tx_queues'])
	verify_vpp_interfaces_dpdk_num_queues(
	qtype='transmit', num_queues=tx_queues, workers=workers
	)

	# RX-mode verification
	rx_mode = iface_config.get('rx_mode')
	if rx_mode and rx_mode != 'polling':
	# By default drivers operate in polling mode. Not all NIC drivers support
	# RX mode interrupt and adaptive
	driver = config.get('persist_config').get(iface).get('original_driver')
	if (
	driver not in drivers_support_interrupt
	or iface_config['driver'] not in drivers_support_interrupt[driver]
	):
	raise ConfigError(
	f'RX mode {rx_mode} is not supported for interface {iface}'
	)

	# check GRE tunnels as part of the bridge, only tunnel-type teb is allowed
	# set vpp interfaces bridge br1 member interface gre1
	# set vpp interfaces gre gre1 tunnel-type teb
	if 'interfaces' in config:
	if 'bridge' in config['interfaces']:
	for iface, iface_config in config['interfaces']['bridge'].items():
	if 'member' in iface_config:
	for member in iface_config['member'].get('interface', []):
	if member.startswith('gre'):
	if (
	'gre' in config['interfaces']
	and config['interfaces']['gre']
	.get(member, {})
	.get('tunnel_type')
	!= 'teb'
	):
	raise ConfigError(
	f'Only tunnel-type teb is allowed for GRE interfaces in bridge {iface}'
	)

	# Only one multipoint GRE tunnel is allowed from the same source address
	# set vpp interfaces gre gre0 mode 'point-to-multipoint'
	# set vpp interfaces gre gre0 remote '0.0.0.0'
	# set vpp interfaces gre gre0 source-address '192.0.2.1'
	# set vpp interfaces gre gre1 mode 'point-to-multipoint'
	# set vpp interfaces gre gre1 remote '0.0.0.0'
	# set vpp interfaces gre gre1 source-address '192.0.2.1'
	if 'gre' in config['interfaces']:
	for iface, iface_config in config['interfaces']['gre'].items():
	if iface_config['mode'] == 'point-to-multipoint':
	for other_iface, other_iface_config in config['interfaces'][
	'gre'
	].items():
	if (
	other_iface_config['mode'] == 'point-to-multipoint'
	and other_iface_config['source_address']
	== iface_config['source_address']
	and iface != other_iface
	):
	raise ConfigError(
	'Only one multipoint GRE tunnel is allowed from the same source address'
	)

	# Check if deleted interfaces are not xconnect memebrs
	for iface_config in config.get('removed_ifaces', []):
	if iface_config['iface_name'] in config.get('xconn_members', {}):
	raise ConfigError(
	f'Interface {iface_config["iface_name"]} is an xconnect member and cannot be removed'
	)

	verify_routes_count(config['settings'], workers)


	def generate(config):
	if not config or ('removed_ifaces' in config and 'settings' not in config):
	# Remove old config and return
	service_conf.unlink(missing_ok=True)
	return None

	render(service_conf, 'vpp/startup.conf.j2', config['settings'])
	render(systemd_override, 'vpp/override.conf.j2', config)

	return None


	def initialize_interface(iface, driver, iface_config) -> None:
	# DPDK - rescan PCI to use a proper driver
	if driver == 'dpdk' and iface_config['original_driver'] not in not_pci_drv:
	control_host.pci_rescan(iface_config['dev_id'])
	# rename to the proper name
	iface_new_name: str = control_host.get_eth_name(iface_config['dev_id'])
	control_host.rename_iface(iface_new_name, iface)

	# XDP - rename an interface, disable promisc and XDP
	if driver == 'xdp':
	control_host.set_promisc(f'defunct_{iface}', 'off')
	control_host.rename_iface(f'defunct_{iface}', iface)
	control_host.xdp_remove(iface)

	# Rename Mellanox NIC to a normal name
	try:
	if control_host.get_eth_driver(f'defunct_{iface}') == 'mlx5_core':
	control_host.rename_iface(f'defunct_{iface}', iface)
	except FileNotFoundError:
	pass

	# Replace a driver with original for VMBus interfaces and rename it
	if driver == 'dpdk' and iface_config['original_driver'] in override_drivers:
	control_host.override_driver(iface_config['bus_id'], iface_config['dev_id'])
	iface_new_name: str = control_host.get_eth_name(iface_config['dev_id'])
	control_host.rename_iface(iface_new_name, iface)


	def apply(config):
	# modrpobe modules
	modules = ('vfio_iommu_type1', 'vfio_pci', 'vfio_pci_core', 'vfio')
	# Open persistent config
	# It is required for operations with interfaces
	persist_config = JSONStorage('vpp_conf')
	if not config or ('removed_ifaces' in config and 'settings' not in config):
	# Cleanup persistent config
	persist_config.delete()
	# And stop the service
	call(f'systemctl stop {service_name}.service')
	# Unlod modules (modprobe -r)
	for module in modules:
	_unload_module(module)
	else:
	# Some interfaces required extra preparation before VPP can be started
	if 'settings' in config and 'interface' in config.get('settings'):
	# modprobe vfio
	if any(
	iface_config.get('driver') == 'dpdk'
	for iface_config in config['settings']['interface'].values()
	):
	for module in modules:
	_load_module(module)

	for iface, iface_config in config['settings']['interface'].items():
	if iface_config['driver'] == 'dpdk':
	# ena interfaces require noiommu mode
	if iface_config['kernel_module'] == 'ena':
	control_host.unsafe_noiommu_mode(True)

	if iface_config['kernel_module'] in override_drivers:
	control_host.override_driver(
	config['persist_config'][iface]['bus_id'],
	config['persist_config'][iface]['dev_id'],
	override_drivers[iface_config['kernel_module']],
	)

	call('systemctl daemon-reload')
	call(f'systemctl restart {service_name}.service')

	# Initialize interfaces removed from VPP
	for iface in config.get('removed_ifaces', []):
	initialize_interface(
	iface['iface_name'],
	iface['driver'],
	config['persist_config'][iface['iface_name']],
	)

	# Remove what is not in the config anymore
	if iface['iface_name'] not in config.get('settings', {}).get('interface', {}):
	del config['persist_config'][iface['iface_name']]

	if 'settings' in config and 'interface' in config.get('settings'):
	# connect to VPP
	# must be performed multiple attempts because API is not available
	# immediately after the service restart
	try:
	vpp_control = VPPControl(attempts=20, interval=500)

	# preconfigure LCP plugin
	if 'ignore_kernel_routes' in config.get('settings', {}).get('lcp', {}):
	vpp_control.cli_cmd('lcp param route-no-paths off')
	else:
	vpp_control.cli_cmd('lcp param route-no-paths on')
	# add interfaces
	iproute = IPRoute()
	for iface, iface_config in config['settings']['interface'].items():
	# promisc option for DPDK interfaces
	if iface_config['driver'] == 'dpdk':
	if 'promisc' in iface_config['dpdk_options']:
	if_index = vpp_control.get_sw_if_index(iface)
	vpp_control.api.sw_interface_set_promisc(
	sw_if_index=if_index, promisc_on=True
	)
	# add XDP interfaces
	if iface_config['driver'] == 'xdp':
	control_host.rename_iface(iface, f'defunct_{iface}')
	vpp_control.xdp_iface_create(
	host_if=f'defunct_{iface}',
	name=iface,
	**iface_config['xdp_api_params'],
	)
	# replicate MAC address of a real interface
	real_mac = control_host.get_eth_mac(f'defunct_{iface}')
	vpp_control.set_iface_mac(iface, real_mac)
	if 'promisc' in iface_config['xdp_options']:
	control_host.set_promisc(f'defunct_{iface}', 'on')
	control_host.set_status(f'defunct_{iface}', 'up')
	control_host.flush_ip(f'defunct_{iface}')
	# Rename Mellanox interfaces to hide them and create LCP properly
	if (
	iface in Section.interfaces()
	and control_host.get_eth_driver(iface) == 'mlx5_core'
	):
	control_host.rename_iface(iface, f'defunct_{iface}')
	control_host.set_status(f'defunct_{iface}', 'up')
	control_host.flush_ip(f'defunct_{iface}')
	# Create lcp
	if iface not in Section.interfaces():
	vpp_control.lcp_pair_add(iface, iface)

	# For unknown reasons, if multiple interfaces later try to be
	# initialized by configuration scripts, some of them may stuck
	# in an endless UP/DOWN loop
	# We found two workarounds - pause initialization (requires
	# main code modifications).
	# And this one
	dev_index = iproute.link_lookup(ifname=iface)[0]
	iproute.link('set', index=dev_index, state='up')

	# Set rx-mode. Should be configured after interface state set to UP
	rx_mode = iface_config.get('rx_mode')
	if rx_mode:
	# to hardware side
	vpp_control.iface_rxmode(iface, rx_mode)
	# to kernel side
	lcp_name = vpp_control.lcp_pair_find(vpp_name_hw=iface).get(
	'vpp_name_kernel'
	)
	vpp_control.iface_rxmode(lcp_name, rx_mode)

	# Syncronize routes via LCP
	vpp_control.lcp_resync()

	# NAT44 settings
	nat44_settings = config['settings'].get('nat44', {})

	enable_forwarding = True
	if 'no_forwarding' in nat44_settings:
	enable_forwarding = False
	vpp_control.enable_disable_nat44_forwarding(enable_forwarding)

	vpp_control.set_nat44_session_limit(
	int(nat44_settings.get('session_limit'))
	)

	if nat44_settings.get('workers'):
	bitmask = 0
	for worker_range in nat44_settings['workers']:
	worker_numbers = worker_range.split('-')
	for wid in range(
	int(worker_numbers[0]), int(worker_numbers[-1]) + 1
	):
	bitmask \|= 1 << wid
	vpp_control.set_nat_workers(bitmask)

	except (VPPIOError, VPPValueError) as e:
	# if cannot connect to VPP or an error occurred then
	# we need to stop vpp service and initialize interfaces
	call(f'systemctl stop {service_name}.service')
	for iface, iface_config in config['settings']['interface'].items():
	initialize_interface(
	iface, iface_config['driver'], config['persist_config'][iface]
	)

	raise ConfigError(
	f'An error occurred: {e}. '
	'VPP service will be restarted with the previous configuration'
	)

	# Save persistent config
	if 'persist_config' in config and config['persist_config']:
	persist_config.write('eth_ifaces', config['persist_config'])

	# reinitialize interfaces, but not during the first boot
	if boot_configuration_complete():
	call_dependents()


	if __name__ == '__main__':
	try:
	c = get_config()
	verify(c)
	generate(c)
	apply(c)
	except ConfigError as e:
	print(e)
	exit(1)
	diff --git a/src/conf_mode/vpp_kernel-interfaces.py b/src/conf_mode/vpp_kernel-interfaces.py
	index 85706094e..d10449eec 100644
	--- a/src/conf_mode/vpp_kernel-interfaces.py
	+++ b/src/conf_mode/vpp_kernel-interfaces.py
	@@ -1,199 +1,205 @@
	#!/usr/bin/env python3
	#
	# 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 os

	from vyos.config import Config
	+from vyos.configdep import set_dependents, call_dependents
	from vyos.configdict import leaf_node_changed
	from vyos.configdict import node_changed
	from vyos import ConfigError
	from vyos.ifconfig import Interface
	from vyos.utils.network import interface_exists
	from vyos.utils.process import call
	from vyos.vpp import VPPControl


	def get_config(config=None) -> dict:
	if config:
	conf = config
	else:
	conf = Config()

	base = ['vpp', 'kernel-interfaces']

	ifname = os.environ['VYOS_TAGNODE_VALUE']

	# Get config_dict with default values
	config = conf.get_config_dict(
	base + [ifname],
	key_mangling=('-', '_'),
	get_first_key=True,
	no_tag_node_value_mangle=True,
	with_defaults=True,
	)

	if not conf.exists(['vpp']):
	config['remove_vpp'] = True
	return config

	# Get effective config as we need full dicitonary per interface delete
	if __name__ == '__main__':
	effective_config = conf.get_config_dict(
	base + [ifname],
	key_mangling=('-', '_'),
	effective=True,
	get_first_key=True,
	no_tag_node_value_mangle=True,
	)
	# if a file was started as dependency, we are starting from empty config
	else:
	effective_config = {}

	if effective_config:
	config.update({'effective': effective_config})

	address_removed = leaf_node_changed(conf, base + [ifname, 'address'])
	if address_removed:
	config['address_removed'] = address_removed

	description_removed = leaf_node_changed(conf, base + [ifname, 'description'])
	if description_removed:
	config['description_removed'] = {}

	vlans_removed = node_changed(conf, base + [ifname, 'vif'])
	if vlans_removed:
	config['vlans_removed'] = vlans_removed

	+ if conf.exists(['vpp', 'nat', 'cgnat']):
	+ set_dependents('vpp_nat_cgnat', conf)
	+
	config['ifname'] = ifname

	return config


	def verify(config):
	if 'remove' in config or 'remove_vpp' in config:
	return None

	# Interface must exists before it is configured
	if not interface_exists(config['ifname']):
	raise ConfigError(
	f'Interface {config["ifname"]} must be created before using in configuration'
	)


	def generate(config):
	pass


	def apply(config):
	if 'remove_vpp' in config:
	return None

	ifname = config.get('ifname')
	i = Interface(ifname)
	# update/remove addresses
	if 'address_removed' in config:
	for address in config['address_removed']:
	i.del_addr(address)
	# remove description
	if 'description_removed' in config:
	i.set_alias('')

	# remove VLANs
	if 'vlans_removed' in config:
	for vlan in config['vlans_removed']:
	call(f'ip link del dev {ifname}.{vlan}')

	# Delete
	if 'remove' in config:
	pass
	else:
	# Add address
	if 'address' in config:
	for address in config['address']:
	i.add_addr(address)
	# Set MTU
	if 'mtu' in config:
	i.set_mtu(config.get('mtu'))
	# Set description
	if 'description' in config:
	i.set_alias(config.get('description'))
	# Admin state down
	if 'disable' in config:
	i.set_admin_state('down')
	else:
	i.set_admin_state('up')

	for vlan, vlan_config in config.get('vif', {}).items():
	if vlan not in config.get('effective', {}).get('vif', {}).keys():
	call(
	f'ip link add link {ifname} name {ifname}.{vlan} type vlan id {vlan}'
	)
	call(f'ip link set dev {ifname}.{vlan} up')
	v = Interface(f'{ifname}.{vlan}')

	# VLAN address
	addresses_effective = (
	config.get('effective', {})
	.get('vif', {})
	.get(vlan, {})
	.get('address', [])
	)
	addresses_candidate = vlan_config.get('address', [])

	for ipaddr in addresses_effective:
	if ipaddr not in addresses_candidate:
	v.del_addr(ipaddr)
	for ipaddr in addresses_candidate:
	if ipaddr not in addresses_effective:
	v.add_addr(ipaddr)

	# VLAN description
	description_effective = (
	config.get('effective', {})
	.get('vif', {})
	.get(vlan, {})
	.get('description', '')
	)
	description_candidate = vlan_config.get('description', '')

	if description_candidate:
	v.set_alias(description_candidate)
	elif description_effective and not description_candidate:
	v.set_alias('')

	v.set_admin_state('up')

	# Set rx-mode
	rx_mode = config.get('rx_mode')
	if rx_mode:
	vpp_control = VPPControl()
	lcp_name = vpp_control.lcp_pair_find(kernel_name=ifname).get('vpp_name_kernel')
	vpp_control.iface_rxmode(lcp_name, rx_mode)

	+ call_dependents()
	+
	return None


	if __name__ == '__main__':
	try:
	c = get_config()
	verify(c)
	generate(c)
	apply(c)
	except ConfigError as e:
	print(e)
	exit(1)
	diff --git a/src/conf_mode/vpp_nat_cgnat.py b/src/conf_mode/vpp_nat_cgnat.py
	index be2386ba2..56d11b5a1 100644
	--- a/src/conf_mode/vpp_nat_cgnat.py
	+++ b/src/conf_mode/vpp_nat_cgnat.py
	@@ -1,218 +1,225 @@
	#!/usr/bin/env python3
	#
	# 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 import ConfigError
	from vyos.config import Config, config_dict_merge
	from vyos.configdict import node_changed
	from vyos.configdiff import Diff
	from vyos.vpp.utils import cli_ifaces_list

	from vyos.vpp.nat.det44 import Det44
	+from vyos.vpp.control_vpp import VPPControl
	+
	+
	+def _vpp_iface_name_transform(iface_name):
	+ vpp_iface_name = iface_name
	+ if vpp_iface_name.startswith('bond'):
	+ # interface name in VPP is BondEthernetX
	+ vpp_iface_name = vpp_iface_name.replace('bond', 'BondEthernet')
	+ return vpp_iface_name


	def get_config(config=None) -> dict:
	if config:
	conf = config
	else:
	conf = Config()

	base = ['vpp', 'nat', 'cgnat']

	# Get config_dict with default values
	config = conf.get_config_dict(
	base,
	key_mangling=('-', '_'),
	get_first_key=True,
	no_tag_node_value_mangle=True,
	)

	if not conf.exists(['vpp']):
	config['remove_vpp'] = True
	return config

	# Get effective config as we need full dictionary to delete
	effective_config = conf.get_config_dict(
	base,
	key_mangling=('-', '_'),
	effective=True,
	get_first_key=True,
	no_tag_node_value_mangle=True,
	)

	if not config:
	config['remove'] = True

	# Get default values which we need to conditionally update into the
	# dictionary retrieved.
	default_values = conf.get_config_defaults(**config.kwargs, recursive=True)
	config = config_dict_merge(default_values, config)

	config_changed = node_changed(
	conf,
	base,
	key_mangling=('-', '_'),
	recursive=True,
	expand_nodes=Diff.DELETE \| Diff.ADD,
	)

	changed_rules = node_changed(
	conf,
	base + ['rule'],
	key_mangling=('-', '_'),
	recursive=True,
	expand_nodes=Diff.DELETE \| Diff.ADD,
	)

	if not config_changed:
	changed_rules = list(config.get('rule', {}).keys())

	config.update(
	{
	'changed_rules': changed_rules,
	'vpp_ifaces': cli_ifaces_list(conf),
	}
	)

	if effective_config:
	config.update({'effective': effective_config})

	return config


	def verify(config):
	if 'remove' in config or 'remove_vpp' in config:
	return None

	if 'interface' not in config:
	raise ConfigError('Interfaces must be configured for CGNAT')
	if 'rule' not in config:
	raise ConfigError('Rules must be configured for CGNAT')

	required_keys = {'inside', 'outside'}
	missing_keys = required_keys - set(config['interface'].keys())
	if missing_keys:
	raise ConfigError(
	f'Both inside and outside interfaces must be configured. '
	f'Please add: {", ".join(missing_keys)}'
	)

	conflict_ifaces = set(config['interface']['inside']).intersection(
	set(config['interface']['outside'])
	)
	if conflict_ifaces:
	raise ConfigError(
	f'Interface cannot be both inside and outside. '
	f'Please choose a side for: {", ".join(conflict_ifaces)} '
	)

	- for interface in config['interface']['inside']:
	- if interface not in config['vpp_ifaces']:
	- raise ConfigError(
	- f'{interface} must be a VPP interface for inside CGNAT interface'
	- )
	- for interface in config['interface']['outside']:
	- if interface not in config['vpp_ifaces']:
	- raise ConfigError(
	- f'{interface} must be a VPP interface for outside CGNAT interface'
	- )
	+ vpp = VPPControl()
	+ for direction in ['inside', 'outside']:
	+ for interface in config['interface'][direction]:
	+ vpp_iface_name = _vpp_iface_name_transform(interface)
	+ if vpp.get_sw_if_index(vpp_iface_name) is None:
	+ raise ConfigError(
	+ f'{interface} must be a VPP interface for {direction} CGNAT interface'
	+ )

	required_keys = {'outside_prefix', 'inside_prefix'}
	for rule in config['rule']:
	missing_keys = required_keys - set(config['rule'][rule].keys())
	if missing_keys:
	raise ConfigError(
	f'Both inside-prefix and outside-prefix must be configured in rule {rule}. '
	f'Please add: {", ".join(missing_keys).replace("_", "-")}'
	)


	def generate(config):
	pass


	def apply(config):
	if 'remove_vpp' in config:
	return None

	cgnat = Det44()

	if 'remove' in config:
	cgnat.disable_det44_plugin()
	return None

	if 'effective' in config:
	remove_config = config.get('effective')
	# Delete inside interfaces
	- for interface in remove_config['interface']['inside']:
	- if interface not in config.get('interface', {}).get('inside', []):
	- cgnat.delete_det44_interface_inside(interface)
	+ for interface in cgnat.get_det44_interfaces_inside():
	+ cgnat.delete_det44_interface_inside(interface)
	# Delete outside interfaces
	- for interface in remove_config['interface']['outside']:
	- if interface not in config.get('interface', {}).get('outside', []):
	- cgnat.delete_det44_interface_outside(interface)
	+ for interface in cgnat.get_det44_interfaces_outside():
	+ cgnat.delete_det44_interface_outside(interface)
	# Delete CGNAT rules
	for rule in config['changed_rules']:
	if rule in remove_config.get('rule', {}):
	rule_config = remove_config['rule'][rule]
	in_addr, in_plen = rule_config['inside_prefix'].split('/')
	out_addr, out_plen = rule_config['outside_prefix'].split('/')
	cgnat.delete_det44_mapping(
	in_addr=in_addr,
	in_plen=int(in_plen),
	out_addr=out_addr,
	out_plen=int(out_plen),
	)

	# Add DET44
	cgnat.enable_det44_plugin()
	# Add inside interfaces
	for interface in config['interface']['inside']:
	- cgnat.add_det44_interface_inside(interface)
	+ vpp_iface_name = _vpp_iface_name_transform(interface)
	+ cgnat.add_det44_interface_inside(vpp_iface_name)
	# Add outside interfaces
	for interface in config['interface']['outside']:
	- cgnat.add_det44_interface_outside(interface)
	+ vpp_iface_name = _vpp_iface_name_transform(interface)
	+ cgnat.add_det44_interface_outside(vpp_iface_name)
	# Add CGNAT rules
	for rule in config['changed_rules']:
	if rule in config.get('rule', {}):
	rule_config = config['rule'][rule]
	in_addr, in_plen = rule_config['inside_prefix'].split('/')
	out_addr, out_plen = rule_config['outside_prefix'].split('/')
	cgnat.add_det44_mapping(
	in_addr=in_addr,
	in_plen=int(in_plen),
	out_addr=out_addr,
	out_plen=int(out_plen),
	)
	# Set CGNAT timeouts
	cgnat.set_det44_timeouts(
	icmp=int(config['timeout']['icmp']),
	udp=int(config['timeout']['udp']),
	tcp_established=int(config['timeout']['tcp_established']),
	tcp_transitory=int(config['timeout']['tcp_transitory']),
	)


	if __name__ == '__main__':
	try:
	c = get_config()
	verify(c)
	generate(c)
	apply(c)
	except ConfigError as e:
	print(e)
	exit(1)

File Metadata

Mime Type: text/x-diff
Expires: Sat, Oct 17, 12:09 AM (1 d, 5 h)
Storage Engine: blob
Storage Format: Raw Data
Storage Handle: 4416432
Default Alt Text: (109 KB)

No OneTemporaryActions

View Options

File Metadata

Event Timeline

No OneTemporary
Actions