diff --git a/smoketest/scripts/cli/test_system_ip.py b/smoketest/scripts/cli/test_system_ip.py
index bd2531084..567416774 100755
--- a/smoketest/scripts/cli/test_system_ip.py
+++ b/smoketest/scripts/cli/test_system_ip.py
@@ -1,113 +1,124 @@
#!/usr/bin/env python3
#
# Copyright (C) 2020-2023 VyOS maintainers and contributors
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License version 2 or later as
# published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import unittest
from base_vyostest_shim import VyOSUnitTestSHIM
from vyos.configsession import ConfigSessionError
from vyos.utils.file import read_file
base_path = ['system', 'ip']
class TestSystemIP(VyOSUnitTestSHIM.TestCase):
def tearDown(self):
self.cli_delete(base_path)
self.cli_commit()
def test_system_ip_forwarding(self):
# Test if IPv4 forwarding can be disabled globally, default is '1'
# which means forwarding enabled
all_forwarding = '/proc/sys/net/ipv4/conf/all/forwarding'
self.assertEqual(read_file(all_forwarding), '1')
self.cli_set(base_path + ['disable-forwarding'])
self.cli_commit()
self.assertEqual(read_file(all_forwarding), '0')
def test_system_ip_directed_broadcast_forwarding(self):
# Test if IPv4 directed broadcast forwarding can be disabled globally,
# default is '1' which means forwarding enabled
bc_forwarding = '/proc/sys/net/ipv4/conf/all/bc_forwarding'
self.assertEqual(read_file(bc_forwarding), '1')
self.cli_set(base_path + ['disable-directed-broadcast'])
self.cli_commit()
self.assertEqual(read_file(bc_forwarding), '0')
def test_system_ip_multipath(self):
# Test IPv4 multipathing options, options default to off -> '0'
use_neigh = '/proc/sys/net/ipv4/fib_multipath_use_neigh'
hash_policy = '/proc/sys/net/ipv4/fib_multipath_hash_policy'
self.assertEqual(read_file(use_neigh), '0')
self.assertEqual(read_file(hash_policy), '0')
self.cli_set(base_path + ['multipath', 'ignore-unreachable-nexthops'])
self.cli_set(base_path + ['multipath', 'layer4-hashing'])
self.cli_commit()
self.assertEqual(read_file(use_neigh), '1')
self.assertEqual(read_file(hash_policy), '1')
def test_system_ip_arp_table_size(self):
# Maximum number of entries to keep in the ARP cache, the
# default is 8k
gc_thresh3 = '/proc/sys/net/ipv4/neigh/default/gc_thresh3'
gc_thresh2 = '/proc/sys/net/ipv4/neigh/default/gc_thresh2'
gc_thresh1 = '/proc/sys/net/ipv4/neigh/default/gc_thresh1'
self.assertEqual(read_file(gc_thresh3), '8192')
self.assertEqual(read_file(gc_thresh2), '4096')
self.assertEqual(read_file(gc_thresh1), '1024')
for size in [1024, 2048, 4096, 8192, 16384, 32768]:
self.cli_set(base_path + ['arp', 'table-size', str(size)])
self.cli_commit()
self.assertEqual(read_file(gc_thresh3), str(size))
self.assertEqual(read_file(gc_thresh2), str(size // 2))
self.assertEqual(read_file(gc_thresh1), str(size // 8))
def test_system_ip_protocol_route_map(self):
protocols = ['any', 'babel', 'bgp', 'connected', 'eigrp', 'isis',
'kernel', 'ospf', 'rip', 'static', 'table']
for protocol in protocols:
self.cli_set(['policy', 'route-map', f'route-map-{protocol}', 'rule', '10', 'action', 'permit'])
self.cli_set(base_path + ['protocol', protocol, 'route-map', f'route-map-{protocol}'])
self.cli_commit()
# Verify route-map properly applied to FRR
frrconfig = self.getFRRconfig('ip protocol', end='', daemon='zebra')
for protocol in protocols:
self.assertIn(f'ip protocol {protocol} route-map route-map-{protocol}', frrconfig)
+ # Delete route-maps
+ self.cli_delete(['policy', 'route-map'])
+ self.cli_delete(base_path + ['protocol'])
+
+ self.cli_commit()
+
+ # Verify route-map properly applied to FRR
+ frrconfig = self.getFRRconfig('ip protocol', end='', daemon='zebra')
+ self.assertNotIn(f'ip protocol', frrconfig)
+
def test_system_ip_protocol_non_existing_route_map(self):
non_existing = 'non-existing'
self.cli_set(base_path + ['protocol', 'static', 'route-map', non_existing])
# VRF does yet not exist - an error must be thrown
with self.assertRaises(ConfigSessionError):
self.cli_commit()
self.cli_set(['policy', 'route-map', non_existing, 'rule', '10', 'action', 'deny'])
+
# Commit again
self.cli_commit()
if __name__ == '__main__':
unittest.main(verbosity=2)
diff --git a/smoketest/scripts/cli/test_system_ipv6.py b/smoketest/scripts/cli/test_system_ipv6.py
index 978af3766..225c2d666 100755
--- a/smoketest/scripts/cli/test_system_ipv6.py
+++ b/smoketest/scripts/cli/test_system_ipv6.py
@@ -1,124 +1,135 @@
#!/usr/bin/env python3
#
# Copyright (C) 2021-2023 VyOS maintainers and contributors
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License version 2 or later as
# published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import unittest
from base_vyostest_shim import VyOSUnitTestSHIM
from vyos.configsession import ConfigSessionError
from vyos.template import is_ipv4
from vyos.utils.file import read_file
from vyos.utils.network import get_interface_config
from vyos.utils.network import is_intf_addr_assigned
base_path = ['system', 'ipv6']
file_forwarding = '/proc/sys/net/ipv6/conf/all/forwarding'
file_disable = '/proc/sys/net/ipv6/conf/all/disable_ipv6'
file_dad = '/proc/sys/net/ipv6/conf/all/accept_dad'
file_multipath = '/proc/sys/net/ipv6/fib_multipath_hash_policy'
class TestSystemIPv6(VyOSUnitTestSHIM.TestCase):
def tearDown(self):
self.cli_delete(base_path)
self.cli_commit()
def test_system_ipv6_forwarding(self):
# Test if IPv6 forwarding can be disabled globally, default is '1'
# which means forwearding enabled
self.assertEqual(read_file(file_forwarding), '1')
self.cli_set(base_path + ['disable-forwarding'])
self.cli_commit()
self.assertEqual(read_file(file_forwarding), '0')
def test_system_ipv6_strict_dad(self):
# This defaults to 1
self.assertEqual(read_file(file_dad), '1')
# Do not assign any IPv6 address on interfaces, this requires a reboot
# which can not be tested, but we can read the config file :)
self.cli_set(base_path + ['strict-dad'])
self.cli_commit()
# Verify configuration file
self.assertEqual(read_file(file_dad), '2')
def test_system_ipv6_multipath(self):
# This defaults to 0
self.assertEqual(read_file(file_multipath), '0')
# Do not assign any IPv6 address on interfaces, this requires a reboot
# which can not be tested, but we can read the config file :)
self.cli_set(base_path + ['multipath', 'layer4-hashing'])
self.cli_commit()
# Verify configuration file
self.assertEqual(read_file(file_multipath), '1')
def test_system_ipv6_neighbor_table_size(self):
# Maximum number of entries to keep in the ARP cache, the
# default is 8192
gc_thresh3 = '/proc/sys/net/ipv6/neigh/default/gc_thresh3'
gc_thresh2 = '/proc/sys/net/ipv6/neigh/default/gc_thresh2'
gc_thresh1 = '/proc/sys/net/ipv6/neigh/default/gc_thresh1'
self.assertEqual(read_file(gc_thresh3), '8192')
self.assertEqual(read_file(gc_thresh2), '4096')
self.assertEqual(read_file(gc_thresh1), '1024')
for size in [1024, 2048, 4096, 8192, 16384, 32768]:
self.cli_set(base_path + ['neighbor', 'table-size', str(size)])
self.cli_commit()
self.assertEqual(read_file(gc_thresh3), str(size))
self.assertEqual(read_file(gc_thresh2), str(size // 2))
self.assertEqual(read_file(gc_thresh1), str(size // 8))
def test_system_ipv6_protocol_route_map(self):
protocols = ['any', 'babel', 'bgp', 'connected', 'isis',
'kernel', 'ospfv3', 'ripng', 'static', 'table']
for protocol in protocols:
route_map = 'route-map-' + protocol.replace('ospfv3', 'ospf6')
self.cli_set(['policy', 'route-map', route_map, 'rule', '10', 'action', 'permit'])
self.cli_set(base_path + ['protocol', protocol, 'route-map', route_map])
self.cli_commit()
# Verify route-map properly applied to FRR
frrconfig = self.getFRRconfig('ipv6 protocol', end='', daemon='zebra')
for protocol in protocols:
# VyOS and FRR use a different name for OSPFv3 (IPv6)
if protocol == 'ospfv3':
protocol = 'ospf6'
self.assertIn(f'ipv6 protocol {protocol} route-map route-map-{protocol}', frrconfig)
+ # Delete route-maps
+ self.cli_delete(['policy', 'route-map'])
+ self.cli_delete(base_path + ['protocol'])
+
+ self.cli_commit()
+
+ # Verify route-map properly applied to FRR
+ frrconfig = self.getFRRconfig('ipv6 protocol', end='', daemon='zebra')
+ self.assertNotIn(f'ipv6 protocol', frrconfig)
+
def test_system_ipv6_protocol_non_existing_route_map(self):
non_existing = 'non-existing6'
self.cli_set(base_path + ['protocol', 'static', 'route-map', non_existing])
# VRF does yet not exist - an error must be thrown
with self.assertRaises(ConfigSessionError):
self.cli_commit()
self.cli_set(['policy', 'route-map', non_existing, 'rule', '10', 'action', 'deny'])
+
# Commit again
self.cli_commit()
if __name__ == '__main__':
unittest.main(verbosity=2)
diff --git a/smoketest/scripts/cli/test_vrf.py b/smoketest/scripts/cli/test_vrf.py
index 5fb599a87..0f658f366 100755
--- a/smoketest/scripts/cli/test_vrf.py
+++ b/smoketest/scripts/cli/test_vrf.py
@@ -1,466 +1,492 @@
#!/usr/bin/env python3
#
# Copyright (C) 2020-2023 VyOS maintainers and contributors
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License version 2 or later as
# published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import re
import os
import json
import unittest
from netifaces import interfaces
from base_vyostest_shim import VyOSUnitTestSHIM
from vyos.configsession import ConfigSessionError
from vyos.ifconfig import Interface
from vyos.ifconfig import Section
from vyos.template import is_ipv4
from vyos.utils.process import cmd
from vyos.utils.file import read_file
from vyos.utils.network import get_interface_config
from vyos.utils.network import is_intf_addr_assigned
base_path = ['vrf']
vrfs = ['red', 'green', 'blue', 'foo-bar', 'baz_foo']
v4_protocols = ['any', 'babel', 'bgp', 'connected', 'eigrp', 'isis', 'kernel', 'ospf', 'rip', 'static', 'table']
v6_protocols = ['any', 'babel', 'bgp', 'connected', 'isis', 'kernel', 'ospfv3', 'ripng', 'static', 'table']
class VRFTest(VyOSUnitTestSHIM.TestCase):
_interfaces = []
@classmethod
def setUpClass(cls):
# we need to filter out VLAN interfaces identified by a dot (.)
# in their name - just in case!
if 'TEST_ETH' in os.environ:
tmp = os.environ['TEST_ETH'].split()
cls._interfaces = tmp
else:
for tmp in Section.interfaces('ethernet', vlan=False):
cls._interfaces.append(tmp)
# call base-classes classmethod
super(VRFTest, cls).setUpClass()
def tearDown(self):
# delete all VRFs
self.cli_delete(base_path)
self.cli_commit()
for vrf in vrfs:
self.assertNotIn(vrf, interfaces())
def test_vrf_vni_and_table_id(self):
base_table = '1000'
table = base_table
for vrf in vrfs:
base = base_path + ['name', vrf]
description = f'VyOS-VRF-{vrf}'
self.cli_set(base + ['description', description])
# check validate() - a table ID is mandatory
with self.assertRaises(ConfigSessionError):
self.cli_commit()
self.cli_set(base + ['table', table])
self.cli_set(base + ['vni', table])
if vrf == 'green':
self.cli_set(base + ['disable'])
table = str(int(table) + 1)
# commit changes
self.cli_commit()
# Verify VRF configuration
table = base_table
iproute2_config = read_file('/etc/iproute2/rt_tables.d/vyos-vrf.conf')
for vrf in vrfs:
description = f'VyOS-VRF-{vrf}'
self.assertTrue(vrf in interfaces())
vrf_if = Interface(vrf)
# validate proper interface description
self.assertEqual(vrf_if.get_alias(), description)
# validate admin up/down state of VRF
state = 'up'
if vrf == 'green':
state = 'down'
self.assertEqual(vrf_if.get_admin_state(), state)
# Test the iproute2 lookup file, syntax is as follows:
#
# # id vrf name comment
# 1000 red # VyOS-VRF-red
# 1001 green # VyOS-VRF-green
# ...
regex = f'{table}\s+{vrf}\s+#\s+{description}'
self.assertTrue(re.findall(regex, iproute2_config))
frrconfig = self.getFRRconfig(f'vrf {vrf}')
self.assertIn(f' vni {table}', frrconfig)
tmp = get_interface_config(vrf)
self.assertEqual(int(table), tmp['linkinfo']['info_data']['table'])
# Increment table ID for the next run
table = str(int(table) + 1)
def test_vrf_loopbacks_ips(self):
table = '2000'
for vrf in vrfs:
base = base_path + ['name', vrf]
self.cli_set(base + ['table', str(table)])
table = str(int(table) + 1)
# commit changes
self.cli_commit()
# Verify VRF configuration
loopbacks = ['127.0.0.1', '::1']
for vrf in vrfs:
# Ensure VRF was created
self.assertIn(vrf, interfaces())
# Verify IP forwarding is 1 (enabled)
self.assertEqual(read_file(f'/proc/sys/net/ipv4/conf/{vrf}/forwarding'), '1')
self.assertEqual(read_file(f'/proc/sys/net/ipv6/conf/{vrf}/forwarding'), '1')
# Test for proper loopback IP assignment
for addr in loopbacks:
self.assertTrue(is_intf_addr_assigned(vrf, addr))
def test_vrf_bind_all(self):
table = '2000'
for vrf in vrfs:
base = base_path + ['name', vrf]
self.cli_set(base + ['table', str(table)])
table = str(int(table) + 1)
self.cli_set(base_path + ['bind-to-all'])
# commit changes
self.cli_commit()
# Verify VRF configuration
tmp = read_file('/proc/sys/net/ipv4/tcp_l3mdev_accept')
self.assertIn(tmp, '1')
tmp = read_file('/proc/sys/net/ipv4/udp_l3mdev_accept')
self.assertIn(tmp, '1')
def test_vrf_table_id_is_unalterable(self):
# Linux Kernel prohibits the change of a VRF table on the fly.
# VRF must be deleted and recreated!
table = '1000'
vrf = vrfs[0]
base = base_path + ['name', vrf]
self.cli_set(base + ['table', table])
# commit changes
self.cli_commit()
# Check if VRF has been created
self.assertTrue(vrf in interfaces())
table = str(int(table) + 1)
self.cli_set(base + ['table', table])
# check validate() - table ID can not be altered!
with self.assertRaises(ConfigSessionError):
self.cli_commit()
def test_vrf_assign_interface(self):
vrf = vrfs[0]
table = '5000'
self.cli_set(['vrf', 'name', vrf, 'table', table])
for interface in self._interfaces:
section = Section.section(interface)
self.cli_set(['interfaces', section, interface, 'vrf', vrf])
# commit changes
self.cli_commit()
# Verify VRF assignmant
for interface in self._interfaces:
tmp = get_interface_config(interface)
self.assertEqual(vrf, tmp['master'])
# cleanup
section = Section.section(interface)
self.cli_delete(['interfaces', section, interface, 'vrf'])
def test_vrf_static_route(self):
base_table = '100'
table = base_table
for vrf in vrfs:
next_hop = f'192.0.{table}.1'
prefix = f'10.0.{table}.0/24'
base = base_path + ['name', vrf]
self.cli_set(base + ['vni', table])
# check validate() - a table ID is mandatory
with self.assertRaises(ConfigSessionError):
self.cli_commit()
self.cli_set(base + ['table', table])
self.cli_set(base + ['protocols', 'static', 'route', prefix, 'next-hop', next_hop])
table = str(int(table) + 1)
# commit changes
self.cli_commit()
# Verify VRF configuration
table = base_table
for vrf in vrfs:
next_hop = f'192.0.{table}.1'
prefix = f'10.0.{table}.0/24'
self.assertTrue(vrf in interfaces())
frrconfig = self.getFRRconfig(f'vrf {vrf}')
self.assertIn(f' vni {table}', frrconfig)
self.assertIn(f' ip route {prefix} {next_hop}', frrconfig)
# Increment table ID for the next run
table = str(int(table) + 1)
def test_vrf_link_local_ip_addresses(self):
# Testcase for issue T4331
table = '100'
vrf = 'orange'
interface = 'dum9998'
addresses = ['192.0.2.1/26', '2001:db8:9998::1/64', 'fe80::1/64']
for address in addresses:
self.cli_set(['interfaces', 'dummy', interface, 'address', address])
# Create dummy interfaces
self.cli_commit()
# ... and verify IP addresses got assigned
for address in addresses:
self.assertTrue(is_intf_addr_assigned(interface, address))
# Move interface to VRF
self.cli_set(base_path + ['name', vrf, 'table', table])
self.cli_set(['interfaces', 'dummy', interface, 'vrf', vrf])
# Apply VRF config
self.cli_commit()
# Ensure VRF got created
self.assertIn(vrf, interfaces())
# ... and IP addresses are still assigned
for address in addresses:
self.assertTrue(is_intf_addr_assigned(interface, address))
# Verify VRF table ID
tmp = get_interface_config(vrf)
self.assertEqual(int(table), tmp['linkinfo']['info_data']['table'])
# Verify interface is assigned to VRF
tmp = get_interface_config(interface)
self.assertEqual(vrf, tmp['master'])
# Delete Interface
self.cli_delete(['interfaces', 'dummy', interface])
self.cli_commit()
def test_vrf_disable_forwarding(self):
table = '2000'
for vrf in vrfs:
base = base_path + ['name', vrf]
self.cli_set(base + ['table', table])
self.cli_set(base + ['ip', 'disable-forwarding'])
self.cli_set(base + ['ipv6', 'disable-forwarding'])
table = str(int(table) + 1)
# commit changes
self.cli_commit()
# Verify VRF configuration
loopbacks = ['127.0.0.1', '::1']
for vrf in vrfs:
# Ensure VRF was created
self.assertIn(vrf, interfaces())
# Verify IP forwarding is 0 (disabled)
self.assertEqual(read_file(f'/proc/sys/net/ipv4/conf/{vrf}/forwarding'), '0')
self.assertEqual(read_file(f'/proc/sys/net/ipv6/conf/{vrf}/forwarding'), '0')
def test_vrf_ip_protocol_route_map(self):
table = '6000'
for vrf in vrfs:
base = base_path + ['name', vrf]
self.cli_set(base + ['table', table])
for protocol in v4_protocols:
self.cli_set(['policy', 'route-map', f'route-map-{vrf}-{protocol}', 'rule', '10', 'action', 'permit'])
self.cli_set(base + ['ip', 'protocol', protocol, 'route-map', f'route-map-{vrf}-{protocol}'])
table = str(int(table) + 1)
self.cli_commit()
# Verify route-map properly applied to FRR
for vrf in vrfs:
frrconfig = self.getFRRconfig(f'vrf {vrf}', daemon='zebra')
self.assertIn(f'vrf {vrf}', frrconfig)
for protocol in v4_protocols:
self.assertIn(f' ip protocol {protocol} route-map route-map-{vrf}-{protocol}', frrconfig)
+ # Delete route-maps
+ for vrf in vrfs:
+ base = base_path + ['name', vrf]
+ self.cli_delete(['policy', 'route-map', f'route-map-{vrf}-{protocol}'])
+ self.cli_delete(base + ['ip', 'protocol'])
+
+ self.cli_commit()
+
+ # Verify route-map properly is removed from FRR
+ for vrf in vrfs:
+ frrconfig = self.getFRRconfig(f'vrf {vrf}', daemon='zebra')
+ self.assertNotIn(f'vrf {vrf}', frrconfig)
+
def test_vrf_ip_ipv6_protocol_non_existing_route_map(self):
table = '6100'
non_existing = 'non-existing'
for vrf in vrfs:
base = base_path + ['name', vrf]
self.cli_set(base + ['table', table])
for protocol in v4_protocols:
self.cli_set(base + ['ip', 'protocol', protocol, 'route-map', f'v4-{non_existing}'])
for protocol in v6_protocols:
self.cli_set(base + ['ipv6', 'protocol', protocol, 'route-map', f'v6-{non_existing}'])
table = str(int(table) + 1)
# Both v4 and v6 route-maps do not exist yet
with self.assertRaises(ConfigSessionError):
self.cli_commit()
self.cli_set(['policy', 'route-map', f'v4-{non_existing}', 'rule', '10', 'action', 'deny'])
# v6 route-map does not exist yet
with self.assertRaises(ConfigSessionError):
self.cli_commit()
self.cli_set(['policy', 'route-map', f'v6-{non_existing}', 'rule', '10', 'action', 'deny'])
# Commit again
self.cli_commit()
def test_vrf_ipv6_protocol_route_map(self):
table = '6200'
for vrf in vrfs:
base = base_path + ['name', vrf]
self.cli_set(base + ['table', table])
for protocol in v6_protocols:
route_map = f'route-map-{vrf}-{protocol.replace("ospfv3", "ospf6")}'
self.cli_set(['policy', 'route-map', route_map, 'rule', '10', 'action', 'permit'])
self.cli_set(base + ['ipv6', 'protocol', protocol, 'route-map', route_map])
table = str(int(table) + 1)
self.cli_commit()
# Verify route-map properly applied to FRR
for vrf in vrfs:
frrconfig = self.getFRRconfig(f'vrf {vrf}', daemon='zebra')
self.assertIn(f'vrf {vrf}', frrconfig)
for protocol in v6_protocols:
# VyOS and FRR use a different name for OSPFv3 (IPv6)
if protocol == 'ospfv3':
protocol = 'ospf6'
route_map = f'route-map-{vrf}-{protocol}'
self.assertIn(f' ipv6 protocol {protocol} route-map {route_map}', frrconfig)
+ # Delete route-maps
+ for vrf in vrfs:
+ base = base_path + ['name', vrf]
+ self.cli_delete(['policy', 'route-map', f'route-map-{vrf}-{protocol}'])
+ self.cli_delete(base + ['ipv6', 'protocol'])
+
+ self.cli_commit()
+
+ # Verify route-map properly is removed from FRR
+ for vrf in vrfs:
+ frrconfig = self.getFRRconfig(f'vrf {vrf}', daemon='zebra')
+ self.assertNotIn(f'vrf {vrf}', frrconfig)
+
def test_vrf_vni_duplicates(self):
base_table = '6300'
table = base_table
for vrf in vrfs:
base = base_path + ['name', vrf]
self.cli_set(base + ['table', str(table)])
self.cli_set(base + ['vni', '100'])
table = str(int(table) + 1)
# L3VNIs can only be used once
with self.assertRaises(ConfigSessionError):
self.cli_commit()
table = base_table
for vrf in vrfs:
base = base_path + ['name', vrf]
self.cli_set(base + ['vni', str(table)])
table = str(int(table) + 1)
# commit changes
self.cli_commit()
# Verify VRF configuration
table = base_table
for vrf in vrfs:
self.assertTrue(vrf in interfaces())
frrconfig = self.getFRRconfig(f'vrf {vrf}')
self.assertIn(f' vni {table}', frrconfig)
# Increment table ID for the next run
table = str(int(table) + 1)
def test_vrf_vni_add_change_remove(self):
base_table = '6300'
table = base_table
for vrf in vrfs:
base = base_path + ['name', vrf]
self.cli_set(base + ['table', str(table)])
self.cli_set(base + ['vni', str(table)])
table = str(int(table) + 1)
# commit changes
self.cli_commit()
# Verify VRF configuration
table = base_table
for vrf in vrfs:
self.assertTrue(vrf in interfaces())
frrconfig = self.getFRRconfig(f'vrf {vrf}')
self.assertIn(f' vni {table}', frrconfig)
# Increment table ID for the next run
table = str(int(table) + 1)
# Now change all L3VNIs (increment 2)
# We must also change the base_table number as we probably could get
# duplicate VNI's during the test as VNIs are applied 1:1 to FRR
base_table = '5000'
table = base_table
for vrf in vrfs:
base = base_path + ['name', vrf]
self.cli_set(base + ['vni', str(table)])
table = str(int(table) + 2)
# commit changes
self.cli_commit()
# Verify VRF configuration
table = base_table
for vrf in vrfs:
self.assertTrue(vrf in interfaces())
frrconfig = self.getFRRconfig(f'vrf {vrf}')
self.assertIn(f' vni {table}', frrconfig)
# Increment table ID for the next run
table = str(int(table) + 2)
# Now delete all the VNIs
for vrf in vrfs:
base = base_path + ['name', vrf]
self.cli_delete(base + ['vni'])
# commit changes
self.cli_commit()
# Verify no VNI is defined
for vrf in vrfs:
self.assertTrue(vrf in interfaces())
frrconfig = self.getFRRconfig(f'vrf {vrf}')
self.assertNotIn('vni', frrconfig)
if __name__ == '__main__':
- unittest.main(verbosity=2)
+ unittest.main(verbosity=2, failfast=True)
diff --git a/src/conf_mode/system-ip.py b/src/conf_mode/system-ip.py
index 9ed34c735..5e4e5ec28 100755
--- a/src/conf_mode/system-ip.py
+++ b/src/conf_mode/system-ip.py
@@ -1,139 +1,137 @@
#!/usr/bin/env python3
#
# Copyright (C) 2019-2023 VyOS maintainers and contributors
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License version 2 or later as
# published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
from sys import exit
from vyos.config import Config
from vyos.configdict import dict_merge
from vyos.configverify import verify_route_map
from vyos.template import render_to_string
from vyos.utils.process import call
from vyos.utils.dict import dict_search
from vyos.utils.file import write_file
from vyos.utils.system import sysctl_write
from vyos import ConfigError
from vyos import frr
from vyos import airbag
airbag.enable()
def get_config(config=None):
if config:
conf = config
else:
conf = Config()
base = ['system', 'ip']
opt = conf.get_config_dict(base, key_mangling=('-', '_'),
get_first_key=True,
with_recursive_defaults=True)
# When working with FRR we need to know the corresponding address-family
opt['afi'] = 'ip'
# We also need the route-map information from the config
#
# XXX: one MUST always call this without the key_mangling() option! See
# vyos.configverify.verify_common_route_maps() for more information.
tmp = {'policy' : {'route-map' : conf.get_config_dict(['policy', 'route-map'],
get_first_key=True)}}
# Merge policy dict into "regular" config dict
opt = dict_merge(tmp, opt)
return opt
def verify(opt):
if 'protocol' in opt:
for protocol, protocol_options in opt['protocol'].items():
if 'route_map' in protocol_options:
verify_route_map(protocol_options['route_map'], opt)
return
def generate(opt):
- if 'protocol' in opt:
- opt['frr_zebra_config'] = render_to_string('frr/zebra.route-map.frr.j2', opt)
+ opt['frr_zebra_config'] = render_to_string('frr/zebra.route-map.frr.j2', opt)
return
def apply(opt):
# Apply ARP threshold values
# table_size has a default value - thus the key always exists
size = int(dict_search('arp.table_size', opt))
# Amount upon reaching which the records begin to be cleared immediately
sysctl_write('net.ipv4.neigh.default.gc_thresh3', size)
# Amount after which the records begin to be cleaned after 5 seconds
sysctl_write('net.ipv4.neigh.default.gc_thresh2', size // 2)
# Minimum number of stored records is indicated which is not cleared
sysctl_write('net.ipv4.neigh.default.gc_thresh1', size // 8)
# enable/disable IPv4 forwarding
tmp = dict_search('disable_forwarding', opt)
value = '0' if (tmp != None) else '1'
write_file('/proc/sys/net/ipv4/conf/all/forwarding', value)
# enable/disable IPv4 directed broadcast forwarding
tmp = dict_search('disable_directed_broadcast', opt)
value = '0' if (tmp != None) else '1'
write_file('/proc/sys/net/ipv4/conf/all/bc_forwarding', value)
# configure multipath
tmp = dict_search('multipath.ignore_unreachable_nexthops', opt)
value = '1' if (tmp != None) else '0'
sysctl_write('net.ipv4.fib_multipath_use_neigh', value)
tmp = dict_search('multipath.layer4_hashing', opt)
value = '1' if (tmp != None) else '0'
sysctl_write('net.ipv4.fib_multipath_hash_policy', value)
# configure TCP options (defaults as of Linux 6.4)
tmp = dict_search('tcp.mss.probing', opt)
if tmp is None:
value = 0
elif tmp == 'on-icmp-black-hole':
value = 1
elif tmp == 'force':
value = 2
else:
# Shouldn't happen
raise ValueError("TCP MSS probing is neither 'on-icmp-black-hole' nor 'force'!")
sysctl_write('net.ipv4.tcp_mtu_probing', value)
tmp = dict_search('tcp.mss.base', opt)
value = '1024' if (tmp is None) else tmp
sysctl_write('net.ipv4.tcp_base_mss', value)
tmp = dict_search('tcp.mss.floor', opt)
value = '48' if (tmp is None) else tmp
sysctl_write('net.ipv4.tcp_mtu_probe_floor', value)
- if 'protocol' in opt:
- zebra_daemon = 'zebra'
- # Save original configuration prior to starting any commit actions
- frr_cfg = frr.FRRConfig()
-
- # The route-map used for the FIB (zebra) is part of the zebra daemon
- frr_cfg.load_configuration(zebra_daemon)
- frr_cfg.modify_section(r'ip protocol \w+ route-map [-a-zA-Z0-9.]+', stop_pattern='(\s|!)')
- if 'frr_zebra_config' in opt:
- frr_cfg.add_before(frr.default_add_before, opt['frr_zebra_config'])
- frr_cfg.commit_configuration(zebra_daemon)
+ zebra_daemon = 'zebra'
+ # Save original configuration prior to starting any commit actions
+ frr_cfg = frr.FRRConfig()
+
+ # The route-map used for the FIB (zebra) is part of the zebra daemon
+ frr_cfg.load_configuration(zebra_daemon)
+ frr_cfg.modify_section(r'ip protocol \w+ route-map [-a-zA-Z0-9.]+', stop_pattern='(\s|!)')
+ if 'frr_zebra_config' in opt:
+ frr_cfg.add_before(frr.default_add_before, opt['frr_zebra_config'])
+ frr_cfg.commit_configuration(zebra_daemon)
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/system-ipv6.py b/src/conf_mode/system-ipv6.py
index 8a4df11fa..e40ed38e2 100755
--- a/src/conf_mode/system-ipv6.py
+++ b/src/conf_mode/system-ipv6.py
@@ -1,117 +1,115 @@
#!/usr/bin/env python3
#
# Copyright (C) 2019-2023 VyOS maintainers and contributors
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License version 2 or later as
# published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import os
from sys import exit
from vyos.config import Config
from vyos.configdict import dict_merge
from vyos.configverify import verify_route_map
from vyos.template import render_to_string
from vyos.utils.dict import dict_search
from vyos.utils.system import sysctl_write
from vyos.utils.file import write_file
from vyos import ConfigError
from vyos import frr
from vyos import airbag
airbag.enable()
def get_config(config=None):
if config:
conf = config
else:
conf = Config()
base = ['system', 'ipv6']
opt = conf.get_config_dict(base, key_mangling=('-', '_'),
get_first_key=True,
with_recursive_defaults=True)
# When working with FRR we need to know the corresponding address-family
opt['afi'] = 'ipv6'
# We also need the route-map information from the config
#
# XXX: one MUST always call this without the key_mangling() option! See
# vyos.configverify.verify_common_route_maps() for more information.
tmp = {'policy' : {'route-map' : conf.get_config_dict(['policy', 'route-map'],
get_first_key=True)}}
# Merge policy dict into "regular" config dict
opt = dict_merge(tmp, opt)
return opt
def verify(opt):
if 'protocol' in opt:
for protocol, protocol_options in opt['protocol'].items():
if 'route_map' in protocol_options:
verify_route_map(protocol_options['route_map'], opt)
return
def generate(opt):
- if 'protocol' in opt:
- opt['frr_zebra_config'] = render_to_string('frr/zebra.route-map.frr.j2', opt)
+ opt['frr_zebra_config'] = render_to_string('frr/zebra.route-map.frr.j2', opt)
return
def apply(opt):
# configure multipath
tmp = dict_search('multipath.layer4_hashing', opt)
value = '1' if (tmp != None) else '0'
sysctl_write('net.ipv6.fib_multipath_hash_policy', value)
# Apply ND threshold values
# table_size has a default value - thus the key always exists
size = int(dict_search('neighbor.table_size', opt))
# Amount upon reaching which the records begin to be cleared immediately
sysctl_write('net.ipv6.neigh.default.gc_thresh3', size)
# Amount after which the records begin to be cleaned after 5 seconds
sysctl_write('net.ipv6.neigh.default.gc_thresh2', size // 2)
# Minimum number of stored records is indicated which is not cleared
sysctl_write('net.ipv6.neigh.default.gc_thresh1', size // 8)
# enable/disable IPv6 forwarding
tmp = dict_search('disable_forwarding', opt)
value = '0' if (tmp != None) else '1'
write_file('/proc/sys/net/ipv6/conf/all/forwarding', value)
# configure IPv6 strict-dad
tmp = dict_search('strict_dad', opt)
value = '2' if (tmp != None) else '1'
for root, dirs, files in os.walk('/proc/sys/net/ipv6/conf'):
for name in files:
if name == 'accept_dad':
write_file(os.path.join(root, name), value)
- if 'protocol' in opt:
- zebra_daemon = 'zebra'
- # Save original configuration prior to starting any commit actions
- frr_cfg = frr.FRRConfig()
+ zebra_daemon = 'zebra'
+ # Save original configuration prior to starting any commit actions
+ frr_cfg = frr.FRRConfig()
- # The route-map used for the FIB (zebra) is part of the zebra daemon
- frr_cfg.load_configuration(zebra_daemon)
- frr_cfg.modify_section(r'ipv6 protocol \w+ route-map [-a-zA-Z0-9.]+', stop_pattern='(\s|!)')
- if 'frr_zebra_config' in opt:
- frr_cfg.add_before(frr.default_add_before, opt['frr_zebra_config'])
- frr_cfg.commit_configuration(zebra_daemon)
+ # The route-map used for the FIB (zebra) is part of the zebra daemon
+ frr_cfg.load_configuration(zebra_daemon)
+ frr_cfg.modify_section(r'ipv6 protocol \w+ route-map [-a-zA-Z0-9.]+', stop_pattern='(\s|!)')
+ if 'frr_zebra_config' in opt:
+ frr_cfg.add_before(frr.default_add_before, opt['frr_zebra_config'])
+ frr_cfg.commit_configuration(zebra_daemon)
if __name__ == '__main__':
try:
c = get_config()
verify(c)
generate(c)
apply(c)
except ConfigError as e:
print(e)
exit(1)