diff --git a/python/vyos/frrender.py b/python/vyos/frrender.py
index 776e893d9..544983b2c 100644
--- a/python/vyos/frrender.py
+++ b/python/vyos/frrender.py
@@ -1,676 +1,687 @@
# Copyright 2024 VyOS maintainers and contributors <maintainers@vyos.io>
#
# This library is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
# License as published by the Free Software Foundation; either
# version 2.1 of the License, or (at your option) any later version.
#
# This library 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
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with this library. If not, see <http://www.gnu.org/licenses/>.
"""
Library used to interface with FRRs mgmtd introduced in version 10.0
"""
import os
from time import sleep
from vyos.defaults import frr_debug_enable
from vyos.utils.dict import dict_search
from vyos.utils.file import write_file
from vyos.utils.process import cmd
from vyos.utils.process import rc_cmd
from vyos.template import render_to_string
from vyos import ConfigError
def debug(message):
if not os.path.exists(frr_debug_enable):
return
print(message)
frr_protocols = ['babel', 'bfd', 'bgp', 'eigrp', 'isis', 'mpls', 'nhrp',
'openfabric', 'ospf', 'ospfv3', 'pim', 'pim6', 'rip',
'ripng', 'rpki', 'segment_routing', 'static']
babel_daemon = 'babeld'
bfd_daemon = 'bfdd'
bgp_daemon = 'bgpd'
isis_daemon = 'isisd'
ldpd_daemon = 'ldpd'
mgmt_daemon = 'mgmtd'
openfabric_daemon = 'fabricd'
ospf_daemon = 'ospfd'
ospf6_daemon = 'ospf6d'
pim_daemon = 'pimd'
pim6_daemon = 'pim6d'
rip_daemon = 'ripd'
ripng_daemon = 'ripngd'
zebra_daemon = 'zebra'
def get_frrender_dict(conf, argv=None) -> dict:
from copy import deepcopy
from vyos.config import config_dict_merge
from vyos.configdict import get_dhcp_interfaces
from vyos.configdict import get_pppoe_interfaces
# Create an empty dictionary which will be filled down the code path and
# returned to the caller
dict = {}
if argv and len(argv) > 1:
dict['vrf_context'] = argv[1]
def dict_helper_ospf_defaults(ospf, path):
# We have gathered the dict representation of the CLI, but there are default
# options which we need to update into the dictionary retrived.
default_values = conf.get_config_defaults(path, key_mangling=('-', '_'),
get_first_key=True, recursive=True)
# We have to cleanup the default dict, as default values could enable features
# which are not explicitly enabled on the CLI. Example: default-information
# originate comes with a default metric-type of 2, which will enable the
# entire default-information originate tree, even when not set via CLI so we
# need to check this first and probably drop that key.
if dict_search('default_information.originate', ospf) is None:
del default_values['default_information']
if 'mpls_te' not in ospf:
del default_values['mpls_te']
if 'graceful_restart' not in ospf:
del default_values['graceful_restart']
for area_num in default_values.get('area', []):
if dict_search(f'area.{area_num}.area_type.nssa', ospf) is None:
del default_values['area'][area_num]['area_type']['nssa']
for protocol in ['babel', 'bgp', 'connected', 'isis', 'kernel', 'rip', 'static']:
if dict_search(f'redistribute.{protocol}', ospf) is None:
del default_values['redistribute'][protocol]
if not bool(default_values['redistribute']):
del default_values['redistribute']
for interface in ospf.get('interface', []):
# We need to reload the defaults on every pass b/c of
# hello-multiplier dependency on dead-interval
# If hello-multiplier is set, we need to remove the default from
# dead-interval.
if 'hello_multiplier' in ospf['interface'][interface]:
del default_values['interface'][interface]['dead_interval']
ospf = config_dict_merge(default_values, ospf)
return ospf
def dict_helper_ospfv3_defaults(ospfv3, path):
# We have gathered the dict representation of the CLI, but there are default
# options which we need to update into the dictionary retrived.
default_values = conf.get_config_defaults(path, key_mangling=('-', '_'),
get_first_key=True, recursive=True)
# We have to cleanup the default dict, as default values could enable features
# which are not explicitly enabled on the CLI. Example: default-information
# originate comes with a default metric-type of 2, which will enable the
# entire default-information originate tree, even when not set via CLI so we
# need to check this first and probably drop that key.
if dict_search('default_information.originate', ospfv3) is None:
del default_values['default_information']
if 'graceful_restart' not in ospfv3:
del default_values['graceful_restart']
for protocol in ['babel', 'bgp', 'connected', 'isis', 'kernel', 'ripng', 'static']:
if dict_search(f'redistribute.{protocol}', ospfv3) is None:
del default_values['redistribute'][protocol]
if not bool(default_values['redistribute']):
del default_values['redistribute']
default_values.pop('interface', {})
# merge in remaining default values
ospfv3 = config_dict_merge(default_values, ospfv3)
return ospfv3
def dict_helper_pim_defaults(pim, path):
# We have gathered the dict representation of the CLI, but there are default
# options which we need to update into the dictionary retrived.
default_values = conf.get_config_defaults(path, key_mangling=('-', '_'),
get_first_key=True, recursive=True)
# We have to cleanup the default dict, as default values could enable features
# which are not explicitly enabled on the CLI.
for interface in pim.get('interface', []):
if 'igmp' not in pim['interface'][interface]:
del default_values['interface'][interface]['igmp']
pim = config_dict_merge(default_values, pim)
return pim
# Ethernet and bonding interfaces can participate in EVPN which is configured via FRR
tmp = {}
for if_type in ['ethernet', 'bonding']:
interface_path = ['interfaces', if_type]
if not conf.exists(interface_path):
continue
for interface in conf.list_nodes(interface_path):
evpn_path = interface_path + [interface, 'evpn']
if not conf.exists(evpn_path):
continue
evpn = conf.get_config_dict(evpn_path, key_mangling=('-', '_'))
tmp.update({interface : evpn})
# At least one participating EVPN interface found, add to result dict
if tmp: dict['interfaces'] = tmp
# Zebra prefix exchange for Kernel IP/IPv6 and routing protocols
for ip_version in ['ip', 'ipv6']:
ip_cli_path = ['system', ip_version]
ip_dict = conf.get_config_dict(ip_cli_path, key_mangling=('-', '_'),
get_first_key=True, with_recursive_defaults=True)
if ip_dict:
ip_dict['afi'] = ip_version
dict.update({ip_version : ip_dict})
# Enable SNMP agentx support
# SNMP AgentX support cannot be disabled once enabled
if conf.exists(['service', 'snmp']):
dict['snmp'] = {}
# We will always need the policy key
dict['policy'] = conf.get_config_dict(['policy'], key_mangling=('-', '_'),
get_first_key=True,
no_tag_node_value_mangle=True)
# We need to check the CLI if the BABEL node is present and thus load in all the default
# values present on the CLI - that's why we have if conf.exists()
babel_cli_path = ['protocols', 'babel']
if conf.exists(babel_cli_path):
babel = conf.get_config_dict(babel_cli_path, key_mangling=('-', '_'),
get_first_key=True,
with_recursive_defaults=True)
dict.update({'babel' : babel})
# We need to check the CLI if the BFD node is present and thus load in all the default
# values present on the CLI - that's why we have if conf.exists()
bfd_cli_path = ['protocols', 'bfd']
if conf.exists(bfd_cli_path):
bfd = conf.get_config_dict(bfd_cli_path, key_mangling=('-', '_'),
get_first_key=True,
no_tag_node_value_mangle=True,
with_recursive_defaults=True)
dict.update({'bfd' : bfd})
# We need to check the CLI if the BGP node is present and thus load in all the default
# values present on the CLI - that's why we have if conf.exists()
bgp_cli_path = ['protocols', 'bgp']
if conf.exists(bgp_cli_path):
bgp = conf.get_config_dict(bgp_cli_path, key_mangling=('-', '_'),
get_first_key=True,
no_tag_node_value_mangle=True,
with_recursive_defaults=True)
bgp['dependent_vrfs'] = {}
dict.update({'bgp' : bgp})
elif conf.exists_effective(bgp_cli_path):
dict.update({'bgp' : {'deleted' : '', 'dependent_vrfs' : {}}})
# We need to check the CLI if the EIGRP node is present and thus load in all the default
# values present on the CLI - that's why we have if conf.exists()
eigrp_cli_path = ['protocols', 'eigrp']
if conf.exists(eigrp_cli_path):
eigrp = conf.get_config_dict(eigrp_cli_path, key_mangling=('-', '_'),
get_first_key=True,
no_tag_node_value_mangle=True,
with_recursive_defaults=True)
dict.update({'eigrp' : eigrp})
elif conf.exists_effective(eigrp_cli_path):
dict.update({'eigrp' : {'deleted' : ''}})
# We need to check the CLI if the ISIS node is present and thus load in all the default
# values present on the CLI - that's why we have if conf.exists()
isis_cli_path = ['protocols', 'isis']
if conf.exists(isis_cli_path):
isis = conf.get_config_dict(isis_cli_path, key_mangling=('-', '_'),
get_first_key=True,
no_tag_node_value_mangle=True,
with_recursive_defaults=True)
dict.update({'isis' : isis})
elif conf.exists_effective(isis_cli_path):
dict.update({'isis' : {'deleted' : ''}})
# We need to check the CLI if the MPLS node is present and thus load in all the default
# values present on the CLI - that's why we have if conf.exists()
mpls_cli_path = ['protocols', 'mpls']
if conf.exists(mpls_cli_path):
mpls = conf.get_config_dict(mpls_cli_path, key_mangling=('-', '_'),
get_first_key=True)
dict.update({'mpls' : mpls})
elif conf.exists_effective(mpls_cli_path):
dict.update({'mpls' : {'deleted' : ''}})
# We need to check the CLI if the OPENFABRIC node is present and thus load in all the default
# values present on the CLI - that's why we have if conf.exists()
openfabric_cli_path = ['protocols', 'openfabric']
if conf.exists(openfabric_cli_path):
openfabric = conf.get_config_dict(openfabric_cli_path, key_mangling=('-', '_'),
get_first_key=True,
no_tag_node_value_mangle=True)
dict.update({'openfabric' : openfabric})
elif conf.exists_effective(openfabric_cli_path):
dict.update({'openfabric' : {'deleted' : ''}})
# We need to check the CLI if the OSPF node is present and thus load in all the default
# values present on the CLI - that's why we have if conf.exists()
ospf_cli_path = ['protocols', 'ospf']
if conf.exists(ospf_cli_path):
ospf = conf.get_config_dict(ospf_cli_path, key_mangling=('-', '_'),
get_first_key=True)
ospf = dict_helper_ospf_defaults(ospf, ospf_cli_path)
dict.update({'ospf' : ospf})
elif conf.exists_effective(ospf_cli_path):
dict.update({'ospf' : {'deleted' : ''}})
# We need to check the CLI if the OSPFv3 node is present and thus load in all the default
# values present on the CLI - that's why we have if conf.exists()
ospfv3_cli_path = ['protocols', 'ospfv3']
if conf.exists(ospfv3_cli_path):
ospfv3 = conf.get_config_dict(ospfv3_cli_path, key_mangling=('-', '_'),
get_first_key=True)
ospfv3 = dict_helper_ospfv3_defaults(ospfv3, ospfv3_cli_path)
dict.update({'ospfv3' : ospfv3})
elif conf.exists_effective(ospfv3_cli_path):
dict.update({'ospfv3' : {'deleted' : ''}})
# We need to check the CLI if the PIM node is present and thus load in all the default
# values present on the CLI - that's why we have if conf.exists()
pim_cli_path = ['protocols', 'pim']
if conf.exists(pim_cli_path):
pim = conf.get_config_dict(pim_cli_path, key_mangling=('-', '_'),
get_first_key=True)
pim = dict_helper_pim_defaults(pim, pim_cli_path)
dict.update({'pim' : pim})
elif conf.exists_effective(pim_cli_path):
dict.update({'pim' : {'deleted' : ''}})
# We need to check the CLI if the PIM6 node is present and thus load in all the default
# values present on the CLI - that's why we have if conf.exists()
pim6_cli_path = ['protocols', 'pim6']
if conf.exists(pim6_cli_path):
pim6 = conf.get_config_dict(pim6_cli_path, key_mangling=('-', '_'),
get_first_key=True,
with_recursive_defaults=True)
dict.update({'pim6' : pim6})
elif conf.exists_effective(pim6_cli_path):
dict.update({'pim6' : {'deleted' : ''}})
# We need to check the CLI if the RIP node is present and thus load in all the default
# values present on the CLI - that's why we have if conf.exists()
rip_cli_path = ['protocols', 'rip']
if conf.exists(rip_cli_path):
rip = conf.get_config_dict(rip_cli_path, key_mangling=('-', '_'),
get_first_key=True,
with_recursive_defaults=True)
dict.update({'rip' : rip})
elif conf.exists_effective(rip_cli_path):
dict.update({'rip' : {'deleted' : ''}})
# We need to check the CLI if the RIPng node is present and thus load in all the default
# values present on the CLI - that's why we have if conf.exists()
ripng_cli_path = ['protocols', 'ripng']
if conf.exists(ripng_cli_path):
ripng = conf.get_config_dict(ripng_cli_path, key_mangling=('-', '_'),
get_first_key=True,
with_recursive_defaults=True)
dict.update({'ripng' : ripng})
elif conf.exists_effective(ripng_cli_path):
dict.update({'ripng' : {'deleted' : ''}})
# We need to check the CLI if the RPKI node is present and thus load in all the default
# values present on the CLI - that's why we have if conf.exists()
rpki_cli_path = ['protocols', 'rpki']
if conf.exists(rpki_cli_path):
rpki = conf.get_config_dict(rpki_cli_path, key_mangling=('-', '_'),
get_first_key=True, with_pki=True,
with_recursive_defaults=True)
rpki_ssh_key_base = '/run/frr/id_rpki'
for cache, cache_config in rpki.get('cache',{}).items():
if 'ssh' in cache_config:
cache_config['ssh']['public_key_file'] = f'{rpki_ssh_key_base}_{cache}.pub'
cache_config['ssh']['private_key_file'] = f'{rpki_ssh_key_base}_{cache}'
dict.update({'rpki' : rpki})
elif conf.exists_effective(rpki_cli_path):
dict.update({'rpki' : {'deleted' : ''}})
# We need to check the CLI if the Segment Routing node is present and thus load in
# all the default values present on the CLI - that's why we have if conf.exists()
sr_cli_path = ['protocols', 'segment-routing']
if conf.exists(sr_cli_path):
sr = conf.get_config_dict(sr_cli_path, key_mangling=('-', '_'),
get_first_key=True,
no_tag_node_value_mangle=True,
with_recursive_defaults=True)
dict.update({'segment_routing' : sr})
elif conf.exists_effective(sr_cli_path):
dict.update({'segment_routing' : {'deleted' : ''}})
# We need to check the CLI if the static node is present and thus load in
# all the default values present on the CLI - that's why we have if conf.exists()
static_cli_path = ['protocols', 'static']
if conf.exists(static_cli_path):
static = conf.get_config_dict(static_cli_path, key_mangling=('-', '_'),
get_first_key=True,
no_tag_node_value_mangle=True)
-
- # T3680 - get a list of all interfaces currently configured to use DHCP
- tmp = get_dhcp_interfaces(conf)
- if tmp: static.update({'dhcp' : tmp})
- tmp = get_pppoe_interfaces(conf)
- if tmp: static.update({'pppoe' : tmp})
-
dict.update({'static' : static})
elif conf.exists_effective(static_cli_path):
dict.update({'static' : {'deleted' : ''}})
+ # T3680 - get a list of all interfaces currently configured to use DHCP
+ tmp = get_dhcp_interfaces(conf)
+ if tmp:
+ if 'static' in dict:
+ dict['static'].update({'dhcp' : tmp})
+ else:
+ dict.update({'static' : {'dhcp' : tmp}})
+ tmp = get_pppoe_interfaces(conf)
+ if tmp:
+ if 'static' in dict:
+ dict['static'].update({'pppoe' : tmp})
+ else:
+ dict.update({'static' : {'pppoe' : tmp}})
+
# keep a re-usable list of dependent VRFs
dependent_vrfs_default = {}
if 'bgp' in dict:
dependent_vrfs_default = deepcopy(dict['bgp'])
# we do not need to nest the 'dependent_vrfs' key - simply remove it
if 'dependent_vrfs' in dependent_vrfs_default:
del dependent_vrfs_default['dependent_vrfs']
vrf_cli_path = ['vrf', 'name']
if conf.exists(vrf_cli_path):
vrf = conf.get_config_dict(vrf_cli_path, key_mangling=('-', '_'),
get_first_key=False,
no_tag_node_value_mangle=True)
# We do not have any VRF related default values on the CLI. The defaults will only
# come into place under the protocols tree, thus we can safely merge them with the
# appropriate routing protocols
for vrf_name, vrf_config in vrf['name'].items():
bgp_vrf_path = ['vrf', 'name', vrf_name, 'protocols', 'bgp']
if 'bgp' in vrf_config.get('protocols', []):
# We have gathered the dict representation of the CLI, but there are default
# options which we need to update into the dictionary retrived.
default_values = conf.get_config_defaults(bgp_vrf_path, key_mangling=('-', '_'),
get_first_key=True, recursive=True)
# merge in remaining default values
vrf_config['protocols']['bgp'] = config_dict_merge(default_values,
vrf_config['protocols']['bgp'])
# Add this BGP VRF instance as dependency into the default VRF
if 'bgp' in dict:
dict['bgp']['dependent_vrfs'].update({vrf_name : deepcopy(vrf_config)})
vrf_config['protocols']['bgp']['dependent_vrfs'] = conf.get_config_dict(
vrf_cli_path, key_mangling=('-', '_'), get_first_key=True,
no_tag_node_value_mangle=True)
# We can safely delete ourself from the dependent VRF list
if vrf_name in vrf_config['protocols']['bgp']['dependent_vrfs']:
del vrf_config['protocols']['bgp']['dependent_vrfs'][vrf_name]
# Add dependency on possible existing default VRF to this VRF
if 'bgp' in dict:
vrf_config['protocols']['bgp']['dependent_vrfs'].update({'default': {'protocols': {
'bgp': dependent_vrfs_default}}})
elif conf.exists_effective(bgp_vrf_path):
# Add this BGP VRF instance as dependency into the default VRF
tmp = {'deleted' : '', 'dependent_vrfs': deepcopy(vrf['name'])}
# We can safely delete ourself from the dependent VRF list
if vrf_name in tmp['dependent_vrfs']:
del tmp['dependent_vrfs'][vrf_name]
# Add dependency on possible existing default VRF to this VRF
if 'bgp' in dict:
tmp['dependent_vrfs'].update({'default': {'protocols': {
'bgp': dependent_vrfs_default}}})
if 'bgp' in dict:
dict['bgp']['dependent_vrfs'].update({vrf_name : {'protocols': tmp} })
if 'protocols' not in vrf['name'][vrf_name]:
vrf['name'][vrf_name].update({'protocols': {'bgp' : tmp}})
else:
vrf['name'][vrf_name]['protocols'].update({'bgp' : tmp})
# We need to check the CLI if the EIGRP node is present and thus load in all the default
# values present on the CLI - that's why we have if conf.exists()
eigrp_vrf_path = ['vrf', 'name', vrf_name, 'protocols', 'eigrp']
if 'eigrp' in vrf_config.get('protocols', []):
eigrp = conf.get_config_dict(eigrp_vrf_path, key_mangling=('-', '_'), get_first_key=True,
no_tag_node_value_mangle=True)
vrf['name'][vrf_name]['protocols'].update({'eigrp' : isis})
elif conf.exists_effective(eigrp_vrf_path):
vrf['name'][vrf_name]['protocols'].update({'eigrp' : {'deleted' : ''}})
# We need to check the CLI if the ISIS node is present and thus load in all the default
# values present on the CLI - that's why we have if conf.exists()
isis_vrf_path = ['vrf', 'name', vrf_name, 'protocols', 'isis']
if 'isis' in vrf_config.get('protocols', []):
isis = conf.get_config_dict(isis_vrf_path, key_mangling=('-', '_'), get_first_key=True,
no_tag_node_value_mangle=True, with_recursive_defaults=True)
vrf['name'][vrf_name]['protocols'].update({'isis' : isis})
elif conf.exists_effective(isis_vrf_path):
vrf['name'][vrf_name]['protocols'].update({'isis' : {'deleted' : ''}})
# We need to check the CLI if the OSPF node is present and thus load in all the default
# values present on the CLI - that's why we have if conf.exists()
ospf_vrf_path = ['vrf', 'name', vrf_name, 'protocols', 'ospf']
if 'ospf' in vrf_config.get('protocols', []):
ospf = conf.get_config_dict(ospf_vrf_path, key_mangling=('-', '_'), get_first_key=True)
ospf = dict_helper_ospf_defaults(vrf_config['protocols']['ospf'], ospf_vrf_path)
vrf['name'][vrf_name]['protocols'].update({'ospf' : ospf})
elif conf.exists_effective(ospf_vrf_path):
vrf['name'][vrf_name]['protocols'].update({'ospf' : {'deleted' : ''}})
# We need to check the CLI if the OSPFv3 node is present and thus load in all the default
# values present on the CLI - that's why we have if conf.exists()
ospfv3_vrf_path = ['vrf', 'name', vrf_name, 'protocols', 'ospfv3']
if 'ospfv3' in vrf_config.get('protocols', []):
ospfv3 = conf.get_config_dict(ospfv3_vrf_path, key_mangling=('-', '_'), get_first_key=True)
ospfv3 = dict_helper_ospfv3_defaults(vrf_config['protocols']['ospfv3'], ospfv3_vrf_path)
vrf['name'][vrf_name]['protocols'].update({'ospfv3' : ospfv3})
elif conf.exists_effective(ospfv3_vrf_path):
vrf['name'][vrf_name]['protocols'].update({'ospfv3' : {'deleted' : ''}})
# We need to check the CLI if the static node is present and thus load in all the default
# values present on the CLI - that's why we have if conf.exists()
static_vrf_path = ['vrf', 'name', vrf_name, 'protocols', 'static']
if 'static' in vrf_config.get('protocols', []):
static = conf.get_config_dict(static_vrf_path, key_mangling=('-', '_'),
get_first_key=True,
no_tag_node_value_mangle=True)
# T3680 - get a list of all interfaces currently configured to use DHCP
tmp = get_dhcp_interfaces(conf, vrf_name)
if tmp: static.update({'dhcp' : tmp})
tmp = get_pppoe_interfaces(conf, vrf_name)
if tmp: static.update({'pppoe' : tmp})
vrf['name'][vrf_name]['protocols'].update({'static': static})
elif conf.exists_effective(static_vrf_path):
vrf['name'][vrf_name]['protocols'].update({'static': {'deleted' : ''}})
vrf_vni_path = ['vrf', 'name', vrf_name, 'vni']
if conf.exists(vrf_vni_path):
vrf_config.update({'vni': conf.return_value(vrf_vni_path)})
dict.update({'vrf' : vrf})
elif conf.exists_effective(vrf_cli_path):
effective_vrf = conf.get_config_dict(vrf_cli_path, key_mangling=('-', '_'),
get_first_key=False,
no_tag_node_value_mangle=True,
effective=True)
vrf = {'name' : {}}
for vrf_name, vrf_config in effective_vrf.get('name', {}).items():
vrf['name'].update({vrf_name : {}})
for protocol in frr_protocols:
if protocol in vrf_config.get('protocols', []):
# Create initial protocols key if not present
if 'protocols' not in vrf['name'][vrf_name]:
vrf['name'][vrf_name].update({'protocols' : {}})
# All routing protocols are deleted when we pass this point
tmp = {'deleted' : ''}
# Special treatment for BGP routing protocol
if protocol == 'bgp':
tmp['dependent_vrfs'] = {}
if 'name' in vrf:
tmp['dependent_vrfs'] = conf.get_config_dict(
vrf_cli_path, key_mangling=('-', '_'),
get_first_key=True, no_tag_node_value_mangle=True,
effective=True)
# Add dependency on possible existing default VRF to this VRF
if 'bgp' in dict:
tmp['dependent_vrfs'].update({'default': {'protocols': {
'bgp': dependent_vrfs_default}}})
# We can safely delete ourself from the dependent VRF list
if vrf_name in tmp['dependent_vrfs']:
del tmp['dependent_vrfs'][vrf_name]
# Update VRF related dict
vrf['name'][vrf_name]['protocols'].update({protocol : tmp})
dict.update({'vrf' : vrf})
+ if os.path.exists(frr_debug_enable):
+ import pprint
+ pprint.pprint(dict)
+
return dict
class FRRender:
cached_config_dict = {}
def __init__(self):
self._frr_conf = '/run/frr/config/vyos.frr.conf'
def generate(self, config_dict) -> None:
"""
Generate FRR configuration file
Returns False if no changes to configuration were made, otherwise True
"""
if not isinstance(config_dict, dict):
tmp = type(config_dict)
raise ValueError(f'Config must be of type "dict" and not "{tmp}"!')
if self.cached_config_dict == config_dict:
debug('FRR: NO CHANGES DETECTED')
return False
self.cached_config_dict = config_dict
def inline_helper(config_dict) -> str:
output = '!\n'
if 'babel' in config_dict and 'deleted' not in config_dict['babel']:
output += render_to_string('frr/babeld.frr.j2', config_dict['babel'])
output += '\n'
if 'bfd' in config_dict and 'deleted' not in config_dict['bfd']:
output += render_to_string('frr/bfdd.frr.j2', config_dict['bfd'])
output += '\n'
if 'bgp' in config_dict and 'deleted' not in config_dict['bgp']:
output += render_to_string('frr/bgpd.frr.j2', config_dict['bgp'])
output += '\n'
if 'eigrp' in config_dict and 'deleted' not in config_dict['eigrp']:
output += render_to_string('frr/eigrpd.frr.j2', config_dict['eigrp'])
output += '\n'
if 'isis' in config_dict and 'deleted' not in config_dict['isis']:
output += render_to_string('frr/isisd.frr.j2', config_dict['isis'])
output += '\n'
if 'mpls' in config_dict and 'deleted' not in config_dict['mpls']:
output += render_to_string('frr/ldpd.frr.j2', config_dict['mpls'])
output += '\n'
if 'openfabric' in config_dict and 'deleted' not in config_dict['openfabric']:
output += render_to_string('frr/fabricd.frr.j2', config_dict['openfabric'])
output += '\n'
if 'ospf' in config_dict and 'deleted' not in config_dict['ospf']:
output += render_to_string('frr/ospfd.frr.j2', config_dict['ospf'])
output += '\n'
if 'ospfv3' in config_dict and 'deleted' not in config_dict['ospfv3']:
output += render_to_string('frr/ospf6d.frr.j2', config_dict['ospfv3'])
output += '\n'
if 'pim' in config_dict and 'deleted' not in config_dict['pim']:
output += render_to_string('frr/pimd.frr.j2', config_dict['pim'])
output += '\n'
if 'pim6' in config_dict and 'deleted' not in config_dict['pim6']:
output += render_to_string('frr/pim6d.frr.j2', config_dict['pim6'])
output += '\n'
if 'policy' in config_dict and len(config_dict['policy']) > 0:
output += render_to_string('frr/policy.frr.j2', config_dict['policy'])
output += '\n'
if 'rip' in config_dict and 'deleted' not in config_dict['rip']:
output += render_to_string('frr/ripd.frr.j2', config_dict['rip'])
output += '\n'
if 'ripng' in config_dict and 'deleted' not in config_dict['ripng']:
output += render_to_string('frr/ripngd.frr.j2', config_dict['ripng'])
output += '\n'
if 'rpki' in config_dict and 'deleted' not in config_dict['rpki']:
output += render_to_string('frr/rpki.frr.j2', config_dict['rpki'])
output += '\n'
if 'segment_routing' in config_dict and 'deleted' not in config_dict['segment_routing']:
output += render_to_string('frr/zebra.segment_routing.frr.j2', config_dict['segment_routing'])
output += '\n'
if 'static' in config_dict and 'deleted' not in config_dict['static']:
output += render_to_string('frr/staticd.frr.j2', config_dict['static'])
output += '\n'
if 'ip' in config_dict and 'deleted' not in config_dict['ip']:
output += render_to_string('frr/zebra.route-map.frr.j2', config_dict['ip'])
output += '\n'
if 'ipv6' in config_dict and 'deleted' not in config_dict['ipv6']:
output += render_to_string('frr/zebra.route-map.frr.j2', config_dict['ipv6'])
output += '\n'
return output
debug('FRR: START CONFIGURATION RENDERING')
# we can not reload an empty file, thus we always embed the marker
output = '!\n'
# Enable SNMP agentx support
# SNMP AgentX support cannot be disabled once enabled
if 'snmp' in config_dict:
output += 'agentx\n'
# Add routing protocols in global VRF
output += inline_helper(config_dict)
# Interface configuration for EVPN is not VRF related
if 'interfaces' in config_dict:
output += render_to_string('frr/evpn.mh.frr.j2', {'interfaces' : config_dict['interfaces']})
output += '\n'
if 'vrf' in config_dict and 'name' in config_dict['vrf']:
output += render_to_string('frr/zebra.vrf.route-map.frr.j2', config_dict['vrf'])
for vrf, vrf_config in config_dict['vrf']['name'].items():
if 'protocols' not in vrf_config:
continue
for protocol in vrf_config['protocols']:
vrf_config['protocols'][protocol]['vrf'] = vrf
output += inline_helper(vrf_config['protocols'])
# remove any accidently added empty newline to not confuse FRR
output = os.linesep.join([s for s in output.splitlines() if s])
if '!!' in output:
raise ConfigError('FRR configuration contains "!!" which is not allowed')
debug(output)
write_file(self._frr_conf, output)
debug('FRR: RENDERING CONFIG COMPLETE')
return True
def apply(self, count_max=5):
count = 0
emsg = ''
while count < count_max:
count += 1
debug(f'FRR: reloading configuration - tries: {count} | Python class ID: {id(self)}')
cmdline = '/usr/lib/frr/frr-reload.py --reload'
if os.path.exists(frr_debug_enable):
cmdline += ' --debug --stdout'
rc, emsg = rc_cmd(f'{cmdline} {self._frr_conf}')
if rc != 0:
sleep(2)
continue
debug(emsg)
debug('FRR: configuration reload complete')
break
if count >= count_max:
raise ConfigError(emsg)
# T3217: Save FRR configuration to /run/frr/config/frr.conf
return cmd('/usr/bin/vtysh -n --writeconfig')
diff --git a/smoketest/scripts/cli/test_protocols_static.py b/smoketest/scripts/cli/test_protocols_static.py
index a2cde0237..7cfc02e30 100755
--- a/smoketest/scripts/cli/test_protocols_static.py
+++ b/smoketest/scripts/cli/test_protocols_static.py
@@ -1,571 +1,620 @@
#!/usr/bin/env python3
#
# Copyright (C) 2021-2024 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
import unittest
+from time import sleep
from base_vyostest_shim import VyOSUnitTestSHIM
from vyos.configsession import ConfigSessionError
from vyos.template import is_ipv6
+from vyos.template import get_dhcp_router
from vyos.utils.network import get_interface_config
from vyos.utils.network import get_vrf_tableid
+from vyos.utils.process import process_named_running
+from vyos.xml_ref import default_value
base_path = ['protocols', 'static']
vrf_path = ['protocols', 'vrf']
routes = {
'10.0.0.0/8' : {
'next_hop' : {
'192.0.2.100' : { 'distance' : '100' },
'192.0.2.110' : { 'distance' : '110', 'interface' : 'eth0' },
'192.0.2.120' : { 'distance' : '120', 'disable' : '' },
'192.0.2.130' : { 'bfd' : '' },
'192.0.2.131' : { 'bfd' : '',
'bfd_profile' : 'vyos1' },
'192.0.2.140' : { 'bfd' : '',
'bfd_source' : '192.0.2.10',
'bfd_profile' : 'vyos2' },
},
'interface' : {
'eth0' : { 'distance' : '130' },
'eth1' : { 'distance' : '140' },
},
'blackhole' : { 'distance' : '250', 'tag' : '500' },
},
'172.16.0.0/12' : {
'interface' : {
'eth0' : { 'distance' : '50', 'vrf' : 'black' },
'eth1' : { 'distance' : '60', 'vrf' : 'black' },
},
'blackhole' : { 'distance' : '90' },
},
'192.0.2.0/24' : {
'interface' : {
'eth0' : { 'distance' : '50', 'vrf' : 'black' },
'eth1' : { 'disable' : '' },
},
'blackhole' : { 'distance' : '90' },
},
'100.64.0.0/16' : {
'blackhole' : {},
},
'100.65.0.0/16' : {
'reject' : { 'distance' : '10', 'tag' : '200' },
},
'100.66.0.0/16' : {
'blackhole' : {},
'reject' : { 'distance' : '10', 'tag' : '200' },
},
'2001:db8:100::/40' : {
'next_hop' : {
'2001:db8::1' : { 'distance' : '10' },
'2001:db8::2' : { 'distance' : '20', 'interface' : 'eth0' },
'2001:db8::3' : { 'distance' : '30', 'disable' : '' },
'2001:db8::4' : { 'bfd' : '' },
'2001:db8::5' : { 'bfd_source' : '2001:db8::ffff' },
},
'interface' : {
'eth0' : { 'distance' : '40', 'vrf' : 'black' },
'eth1' : { 'distance' : '50', 'disable' : '' },
},
'blackhole' : { 'distance' : '250', 'tag' : '500' },
},
'2001:db8:200::/40' : {
'interface' : {
'eth0' : { 'distance' : '40' },
'eth1' : { 'distance' : '50', 'disable' : '' },
},
'blackhole' : { 'distance' : '250', 'tag' : '500' },
},
'2001:db8:300::/40' : {
'reject' : { 'distance' : '250', 'tag' : '500' },
},
'2001:db8:400::/40' : {
'next_hop' : {
'2001:db8::400' : { 'segments' : '2001:db8:aaaa::400/2002::400/2003::400/2004::400' },
},
},
'2001:db8:500::/40' : {
'next_hop' : {
'2001:db8::500' : { 'segments' : '2001:db8:aaaa::500/2002::500/2003::500/2004::500' },
},
},
'2001:db8:600::/40' : {
'interface' : {
'eth0' : { 'segments' : '2001:db8:aaaa::600/2002::600' },
},
},
'2001:db8:700::/40' : {
'interface' : {
'eth1' : { 'segments' : '2001:db8:aaaa::700' },
},
},
'2001:db8::/32' : {
'blackhole' : { 'distance' : '200', 'tag' : '600' }
},
}
multicast_routes = {
'224.0.0.0/24' : {
'next_hop' : {
'224.203.0.1' : { },
'224.203.0.2' : { 'distance' : '110'},
},
},
'224.1.0.0/24' : {
'next_hop' : {
'224.205.0.1' : { 'disable' : {} },
'224.205.0.2' : { 'distance' : '110'},
},
},
'224.2.0.0/24' : {
'next_hop' : {
'1.2.3.0' : { },
'1.2.3.1' : { 'distance' : '110'},
},
},
'224.10.0.0/24' : {
'interface' : {
'eth1' : { 'disable' : {} },
'eth2' : { 'distance' : '110'},
},
},
'224.11.0.0/24' : {
'interface' : {
'eth0' : { },
'eth1' : { 'distance' : '10'},
},
},
'224.12.0.0/24' : {
'interface' : {
'eth0' : { },
'eth1' : { 'distance' : '200'},
},
},
}
tables = ['80', '81', '82']
class TestProtocolsStatic(VyOSUnitTestSHIM.TestCase):
@classmethod
def setUpClass(cls):
super(TestProtocolsStatic, cls).setUpClass()
+ cls.cli_delete(cls, base_path)
cls.cli_delete(cls, ['vrf'])
- cls.cli_set(cls, ['vrf', 'name', 'black', 'table', '43210'])
@classmethod
def tearDownClass(cls):
+ cls.cli_delete(cls, base_path)
cls.cli_delete(cls, ['vrf'])
super(TestProtocolsStatic, cls).tearDownClass()
def tearDown(self):
self.cli_delete(base_path)
+ self.cli_delete(['vrf'])
self.cli_commit()
v4route = self.getFRRconfig('ip route', end='')
self.assertFalse(v4route)
v6route = self.getFRRconfig('ipv6 route', end='')
self.assertFalse(v6route)
def test_01_static(self):
+ self.cli_set(['vrf', 'name', 'black', 'table', '43210'])
for route, route_config in routes.items():
route_type = 'route'
if is_ipv6(route):
route_type = 'route6'
base = base_path + [route_type, route]
if 'next_hop' in route_config:
for next_hop, next_hop_config in route_config['next_hop'].items():
self.cli_set(base + ['next-hop', next_hop])
if 'disable' in next_hop_config:
self.cli_set(base + ['next-hop', next_hop, 'disable'])
if 'distance' in next_hop_config:
self.cli_set(base + ['next-hop', next_hop, 'distance', next_hop_config['distance']])
if 'interface' in next_hop_config:
self.cli_set(base + ['next-hop', next_hop, 'interface', next_hop_config['interface']])
if 'vrf' in next_hop_config:
self.cli_set(base + ['next-hop', next_hop, 'vrf', next_hop_config['vrf']])
if 'bfd' in next_hop_config:
self.cli_set(base + ['next-hop', next_hop, 'bfd'])
if 'bfd_profile' in next_hop_config:
self.cli_set(base + ['next-hop', next_hop, 'bfd', 'profile', next_hop_config['bfd_profile']])
if 'bfd_source' in next_hop_config:
self.cli_set(base + ['next-hop', next_hop, 'bfd', 'multi-hop', 'source-address', next_hop_config['bfd_source']])
if 'segments' in next_hop_config:
self.cli_set(base + ['next-hop', next_hop, 'segments', next_hop_config['segments']])
if 'interface' in route_config:
for interface, interface_config in route_config['interface'].items():
self.cli_set(base + ['interface', interface])
if 'disable' in interface_config:
self.cli_set(base + ['interface', interface, 'disable'])
if 'distance' in interface_config:
self.cli_set(base + ['interface', interface, 'distance', interface_config['distance']])
if 'vrf' in interface_config:
self.cli_set(base + ['interface', interface, 'vrf', interface_config['vrf']])
if 'segments' in interface_config:
self.cli_set(base + ['interface', interface, 'segments', interface_config['segments']])
if 'blackhole' in route_config:
self.cli_set(base + ['blackhole'])
if 'distance' in route_config['blackhole']:
self.cli_set(base + ['blackhole', 'distance', route_config['blackhole']['distance']])
if 'tag' in route_config['blackhole']:
self.cli_set(base + ['blackhole', 'tag', route_config['blackhole']['tag']])
if 'reject' in route_config:
self.cli_set(base + ['reject'])
if 'distance' in route_config['reject']:
self.cli_set(base + ['reject', 'distance', route_config['reject']['distance']])
if 'tag' in route_config['reject']:
self.cli_set(base + ['reject', 'tag', route_config['reject']['tag']])
if {'blackhole', 'reject'} <= set(route_config):
# Can not use blackhole and reject at the same time
with self.assertRaises(ConfigSessionError):
self.cli_commit()
self.cli_delete(base + ['blackhole'])
self.cli_delete(base + ['reject'])
# commit changes
self.cli_commit()
# Verify FRR bgpd configuration
frrconfig = self.getFRRconfig('ip route', end='')
# Verify routes
for route, route_config in routes.items():
ip_ipv6 = 'ip'
if is_ipv6(route):
ip_ipv6 = 'ipv6'
if 'next_hop' in route_config:
for next_hop, next_hop_config in route_config['next_hop'].items():
tmp = f'{ip_ipv6} route {route} {next_hop}'
if 'interface' in next_hop_config:
tmp += ' ' + next_hop_config['interface']
if 'distance' in next_hop_config:
tmp += ' ' + next_hop_config['distance']
if 'vrf' in next_hop_config:
tmp += ' nexthop-vrf ' + next_hop_config['vrf']
if 'bfd' in next_hop_config:
tmp += ' bfd'
if 'bfd_source' in next_hop_config:
tmp += ' multi-hop source ' + next_hop_config['bfd_source']
if 'bfd_profile' in next_hop_config:
tmp += ' profile ' + next_hop_config['bfd_profile']
if 'segments' in next_hop_config:
tmp += ' segments ' + next_hop_config['segments']
if 'disable' in next_hop_config:
self.assertNotIn(tmp, frrconfig)
else:
self.assertIn(tmp, frrconfig)
if 'interface' in route_config:
for interface, interface_config in route_config['interface'].items():
tmp = f'{ip_ipv6} route {route} {interface}'
if 'interface' in interface_config:
tmp += ' ' + interface_config['interface']
if 'distance' in interface_config:
tmp += ' ' + interface_config['distance']
if 'vrf' in interface_config:
tmp += ' nexthop-vrf ' + interface_config['vrf']
if 'segments' in interface_config:
tmp += ' segments ' + interface_config['segments']
if 'disable' in interface_config:
self.assertNotIn(tmp, frrconfig)
else:
self.assertIn(tmp, frrconfig)
if {'blackhole', 'reject'} <= set(route_config):
# Can not use blackhole and reject at the same time
# Config error validated above - skip this route
continue
if 'blackhole' in route_config:
tmp = f'{ip_ipv6} route {route} blackhole'
if 'tag' in route_config['blackhole']:
tmp += ' tag ' + route_config['blackhole']['tag']
if 'distance' in route_config['blackhole']:
tmp += ' ' + route_config['blackhole']['distance']
self.assertIn(tmp, frrconfig)
if 'reject' in route_config:
tmp = f'{ip_ipv6} route {route} reject'
if 'tag' in route_config['reject']:
tmp += ' tag ' + route_config['reject']['tag']
if 'distance' in route_config['reject']:
tmp += ' ' + route_config['reject']['distance']
self.assertIn(tmp, frrconfig)
def test_02_static_table(self):
+ self.cli_set(['vrf', 'name', 'black', 'table', '43210'])
for table in tables:
for route, route_config in routes.items():
route_type = 'route'
if is_ipv6(route):
route_type = 'route6'
base = base_path + ['table', table, route_type, route]
if 'next_hop' in route_config:
for next_hop, next_hop_config in route_config['next_hop'].items():
self.cli_set(base + ['next-hop', next_hop])
if 'disable' in next_hop_config:
self.cli_set(base + ['next-hop', next_hop, 'disable'])
if 'distance' in next_hop_config:
self.cli_set(base + ['next-hop', next_hop, 'distance', next_hop_config['distance']])
if 'interface' in next_hop_config:
self.cli_set(base + ['next-hop', next_hop, 'interface', next_hop_config['interface']])
if 'vrf' in next_hop_config:
self.cli_set(base + ['next-hop', next_hop, 'vrf', next_hop_config['vrf']])
if 'interface' in route_config:
for interface, interface_config in route_config['interface'].items():
self.cli_set(base + ['interface', interface])
if 'disable' in interface_config:
self.cli_set(base + ['interface', interface, 'disable'])
if 'distance' in interface_config:
self.cli_set(base + ['interface', interface, 'distance', interface_config['distance']])
if 'vrf' in interface_config:
self.cli_set(base + ['interface', interface, 'vrf', interface_config['vrf']])
if 'blackhole' in route_config:
self.cli_set(base + ['blackhole'])
if 'distance' in route_config['blackhole']:
self.cli_set(base + ['blackhole', 'distance', route_config['blackhole']['distance']])
if 'tag' in route_config['blackhole']:
self.cli_set(base + ['blackhole', 'tag', route_config['blackhole']['tag']])
# commit changes
self.cli_commit()
# Verify FRR bgpd configuration
frrconfig = self.getFRRconfig('ip route', end='')
for table in tables:
# Verify routes
for route, route_config in routes.items():
ip_ipv6 = 'ip'
if is_ipv6(route):
ip_ipv6 = 'ipv6'
if 'next_hop' in route_config:
for next_hop, next_hop_config in route_config['next_hop'].items():
tmp = f'{ip_ipv6} route {route} {next_hop}'
if 'interface' in next_hop_config:
tmp += ' ' + next_hop_config['interface']
if 'distance' in next_hop_config:
tmp += ' ' + next_hop_config['distance']
if 'vrf' in next_hop_config:
tmp += ' nexthop-vrf ' + next_hop_config['vrf']
tmp += ' table ' + table
if 'disable' in next_hop_config:
self.assertNotIn(tmp, frrconfig)
else:
self.assertIn(tmp, frrconfig)
if 'interface' in route_config:
for interface, interface_config in route_config['interface'].items():
tmp = f'{ip_ipv6} route {route} {interface}'
if 'interface' in interface_config:
tmp += ' ' + interface_config['interface']
if 'distance' in interface_config:
tmp += ' ' + interface_config['distance']
if 'vrf' in interface_config:
tmp += ' nexthop-vrf ' + interface_config['vrf']
tmp += ' table ' + table
if 'disable' in interface_config:
self.assertNotIn(tmp, frrconfig)
else:
self.assertIn(tmp, frrconfig)
if 'blackhole' in route_config:
tmp = f'{ip_ipv6} route {route} blackhole'
if 'tag' in route_config['blackhole']:
tmp += ' tag ' + route_config['blackhole']['tag']
if 'distance' in route_config['blackhole']:
tmp += ' ' + route_config['blackhole']['distance']
tmp += ' table ' + table
self.assertIn(tmp, frrconfig)
def test_03_static_vrf(self):
+ self.cli_set(['vrf', 'name', 'black', 'table', '43210'])
# Create VRF instances and apply the static routes from above to FRR.
# Re-read the configured routes and match them if they are programmed
# properly. This also includes VRF leaking
vrfs = {
'red' : { 'table' : '1000' },
'green' : { 'table' : '2000' },
'blue' : { 'table' : '3000' },
}
for vrf, vrf_config in vrfs.items():
vrf_base_path = ['vrf', 'name', vrf]
self.cli_set(vrf_base_path + ['table', vrf_config['table']])
for route, route_config in routes.items():
route_type = 'route'
if is_ipv6(route):
route_type = 'route6'
route_base_path = vrf_base_path + ['protocols', 'static', route_type, route]
if 'next_hop' in route_config:
for next_hop, next_hop_config in route_config['next_hop'].items():
self.cli_set(route_base_path + ['next-hop', next_hop])
if 'disable' in next_hop_config:
self.cli_set(route_base_path + ['next-hop', next_hop, 'disable'])
if 'distance' in next_hop_config:
self.cli_set(route_base_path + ['next-hop', next_hop, 'distance', next_hop_config['distance']])
if 'interface' in next_hop_config:
self.cli_set(route_base_path + ['next-hop', next_hop, 'interface', next_hop_config['interface']])
if 'vrf' in next_hop_config:
self.cli_set(route_base_path + ['next-hop', next_hop, 'vrf', next_hop_config['vrf']])
if 'segments' in next_hop_config:
self.cli_set(route_base_path + ['next-hop', next_hop, 'segments', next_hop_config['segments']])
if 'interface' in route_config:
for interface, interface_config in route_config['interface'].items():
self.cli_set(route_base_path + ['interface', interface])
if 'disable' in interface_config:
self.cli_set(route_base_path + ['interface', interface, 'disable'])
if 'distance' in interface_config:
self.cli_set(route_base_path + ['interface', interface, 'distance', interface_config['distance']])
if 'vrf' in interface_config:
self.cli_set(route_base_path + ['interface', interface, 'vrf', interface_config['vrf']])
if 'segments' in interface_config:
self.cli_set(route_base_path + ['interface', interface, 'segments', interface_config['segments']])
if 'blackhole' in route_config:
self.cli_set(route_base_path + ['blackhole'])
if 'distance' in route_config['blackhole']:
self.cli_set(route_base_path + ['blackhole', 'distance', route_config['blackhole']['distance']])
if 'tag' in route_config['blackhole']:
self.cli_set(route_base_path + ['blackhole', 'tag', route_config['blackhole']['tag']])
# commit changes
self.cli_commit()
for vrf, vrf_config in vrfs.items():
tmp = get_interface_config(vrf)
# Compare VRF table ID
self.assertEqual(get_vrf_tableid(vrf), int(vrf_config['table']))
self.assertEqual(tmp['linkinfo']['info_kind'], 'vrf')
# Verify FRR bgpd configuration
frrconfig = self.getFRRconfig(f'vrf {vrf}', endsection='^exit-vrf')
self.assertIn(f'vrf {vrf}', frrconfig)
# Verify routes
for route, route_config in routes.items():
ip_ipv6 = 'ip'
if is_ipv6(route):
ip_ipv6 = 'ipv6'
if 'next_hop' in route_config:
for next_hop, next_hop_config in route_config['next_hop'].items():
tmp = f'{ip_ipv6} route {route} {next_hop}'
if 'interface' in next_hop_config:
tmp += ' ' + next_hop_config['interface']
if 'distance' in next_hop_config:
tmp += ' ' + next_hop_config['distance']
if 'vrf' in next_hop_config:
tmp += ' nexthop-vrf ' + next_hop_config['vrf']
if 'segments' in next_hop_config:
tmp += ' segments ' + next_hop_config['segments']
if 'disable' in next_hop_config:
self.assertNotIn(tmp, frrconfig)
else:
self.assertIn(tmp, frrconfig)
if 'interface' in route_config:
for interface, interface_config in route_config['interface'].items():
tmp = f'{ip_ipv6} route {route} {interface}'
if 'interface' in interface_config:
tmp += ' ' + interface_config['interface']
if 'distance' in interface_config:
tmp += ' ' + interface_config['distance']
if 'vrf' in interface_config:
tmp += ' nexthop-vrf ' + interface_config['vrf']
if 'segments' in interface_config:
tmp += ' segments ' + interface_config['segments']
if 'disable' in interface_config:
self.assertNotIn(tmp, frrconfig)
else:
self.assertIn(tmp, frrconfig)
if 'blackhole' in route_config:
tmp = f'{ip_ipv6} route {route} blackhole'
if 'tag' in route_config['blackhole']:
tmp += ' tag ' + route_config['blackhole']['tag']
if 'distance' in route_config['blackhole']:
tmp += ' ' + route_config['blackhole']['distance']
self.assertIn(tmp, frrconfig)
def test_04_static_multicast(self):
for route, route_config in multicast_routes.items():
if 'next_hop' in route_config:
base = base_path + ['mroute', route]
for next_hop, next_hop_config in route_config['next_hop'].items():
self.cli_set(base + ['next-hop', next_hop])
if 'distance' in next_hop_config:
self.cli_set(base + ['next-hop', next_hop, 'distance', next_hop_config['distance']])
if 'disable' in next_hop_config:
self.cli_set(base + ['next-hop', next_hop, 'disable'])
if 'interface' in route_config:
base = base_path + ['mroute', route]
for next_hop, next_hop_config in route_config['interface'].items():
self.cli_set(base + ['interface', next_hop])
if 'distance' in next_hop_config:
self.cli_set(base + ['interface', next_hop, 'distance', next_hop_config['distance']])
self.cli_commit()
# Verify FRR configuration
frrconfig = self.getFRRconfig('ip mroute', end='')
for route, route_config in multicast_routes.items():
if 'next_hop' in route_config:
for next_hop, next_hop_config in route_config['next_hop'].items():
tmp = f'ip mroute {route} {next_hop}'
if 'distance' in next_hop_config:
tmp += ' ' + next_hop_config['distance']
if 'disable' in next_hop_config:
self.assertNotIn(tmp, frrconfig)
else:
self.assertIn(tmp, frrconfig)
if 'next_hop_interface' in route_config:
for next_hop, next_hop_config in route_config['next_hop_interface'].items():
tmp = f'ip mroute {route} {next_hop}'
if 'distance' in next_hop_config:
tmp += ' ' + next_hop_config['distance']
if 'disable' in next_hop_config:
self.assertNotIn(tmp, frrconfig)
else:
self.assertIn(tmp, frrconfig)
+ def test_05_dhcp_default_route(self):
+ # When running via vyos-build under the QEmu environment a local DHCP
+ # server is available. This test verifies that the default route is set.
+ # When not running under the VyOS QEMU environment, this test is skipped.
+ if not os.path.exists('/tmp/vyos.smoketests.hint'):
+ self.skipTest('Not running under VyOS CI/CD QEMU environment!')
+
+ interface = 'eth0'
+ interface_path = ['interfaces', 'ethernet', interface]
+ default_distance = default_value(interface_path + ['dhcp-options', 'default-route-distance'])
+ self.cli_set(interface_path + ['address', 'dhcp'])
+ self.cli_commit()
+
+ # Wait for dhclient to receive IP address and default gateway
+ sleep(5)
+
+ router = get_dhcp_router(interface)
+ frrconfig = self.getFRRconfig('')
+ self.assertIn(rf'ip route 0.0.0.0/0 {router} {interface} tag 210 {default_distance}', frrconfig)
+
+ # T6991: Default route is missing when there is no "protocols static"
+ # CLI node entry
+ self.cli_delete(base_path)
+ # We can trigger a FRR reconfiguration and config re-rendering when
+ # we simply disable IPv6 forwarding
+ self.cli_set(['system', 'ipv6', 'disable-forwarding'])
+ self.cli_commit()
+
+ # Re-check FRR configuration that default route is still present
+ frrconfig = self.getFRRconfig('')
+ self.assertIn(rf'ip route 0.0.0.0/0 {router} {interface} tag 210 {default_distance}', frrconfig)
+
+ self.cli_delete(interface_path + ['address'])
+ self.cli_commit()
+
+ # Wait for dhclient to stop
+ while process_named_running('dhclient', cmdline=interface, timeout=10):
+ sleep(0.250)
+
if __name__ == '__main__':
unittest.main(verbosity=2)