diff --git a/python/vyos/configdict.py b/python/vyos/configdict.py
index baffd94dd..f5e84267e 100644
--- a/python/vyos/configdict.py
+++ b/python/vyos/configdict.py
@@ -1,1139 +1,1147 @@
# Copyright 2019-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/>.
"""
A library for retrieving value dicts from VyOS configs in a declarative fashion.
"""
import os
import json
from vyos.defaults import frr_debug_enable
from vyos.utils.dict import dict_search
from vyos.utils.process import cmd
def retrieve_config(path_hash, base_path, config):
"""
Retrieves a VyOS config as a dict according to a declarative description
The description dict, passed in the first argument, must follow this format:
``field_name : <path, type, [inner_options_dict]>``.
Supported types are: ``str`` (for normal nodes),
``list`` (returns a list of strings, for multi nodes),
``bool`` (returns True if valueless node exists),
``dict`` (for tag nodes, returns a dict indexed by node names,
according to description in the third item of the tuple).
Args:
path_hash (dict): Declarative description of the config to retrieve
base_path (list): A base path to prepend to all option paths
config (vyos.config.Config): A VyOS config object
Returns:
dict: config dict
"""
config_hash = {}
for k in path_hash:
if type(path_hash[k]) != tuple:
raise ValueError("In field {0}: expected a tuple, got a value {1}".format(k, str(path_hash[k])))
if len(path_hash[k]) < 2:
raise ValueError("In field {0}: field description must be a tuple of at least two items, path (list) and type".format(k))
path = path_hash[k][0]
if type(path) != list:
raise ValueError("In field {0}: path must be a list, not a {1}".format(k, type(path)))
typ = path_hash[k][1]
if type(typ) != type:
raise ValueError("In field {0}: type must be a type, not a {1}".format(k, type(typ)))
path = base_path + path
path_str = " ".join(path)
if typ == str:
config_hash[k] = config.return_value(path_str)
elif typ == list:
config_hash[k] = config.return_values(path_str)
elif typ == bool:
config_hash[k] = config.exists(path_str)
elif typ == dict:
try:
inner_hash = path_hash[k][2]
except IndexError:
raise ValueError("The type of the \'{0}\' field is dict, but inner options hash is missing from the tuple".format(k))
config_hash[k] = {}
nodes = config.list_nodes(path_str)
for node in nodes:
config_hash[k][node] = retrieve_config(inner_hash, path + [node], config)
return config_hash
def dict_merge(source, destination):
""" Merge two dictionaries. Only keys which are not present in destination
will be copied from source, anything else will be kept untouched. Function
will return a new dict which has the merged key/value pairs. """
from copy import deepcopy
tmp = deepcopy(destination)
for key, value in source.items():
if key not in tmp:
tmp[key] = value
elif isinstance(source[key], dict):
tmp[key] = dict_merge(source[key], tmp[key])
return tmp
def list_diff(first, second):
""" Diff two dictionaries and return only unique items """
second = set(second)
return [item for item in first if item not in second]
def is_node_changed(conf, path):
"""
Check if any key under path has been changed and return True.
If nothing changed, return false
"""
from vyos.configdiff import get_config_diff
D = get_config_diff(conf, key_mangling=('-', '_'))
return D.is_node_changed(path)
def leaf_node_changed(conf, path):
"""
Check if a leaf node was altered. If it has been altered - values has been
changed, or it was added/removed, we will return a list containing the old
value(s). If nothing has been changed, None is returned.
NOTE: path must use the real CLI node name (e.g. with a hyphen!)
"""
from vyos.configdiff import get_config_diff
D = get_config_diff(conf, key_mangling=('-', '_'))
(new, old) = D.get_value_diff(path)
if new != old:
if isinstance(old, dict):
# valueLess nodes return {} if node is deleted
return True
if old is None and isinstance(new, dict):
# valueLess nodes return {} if node was added
return True
if old is None:
return []
if isinstance(old, str):
return [old]
if isinstance(old, list):
if isinstance(new, str):
new = [new]
elif isinstance(new, type(None)):
new = []
return list_diff(old, new)
return None
def node_changed(conf, path, key_mangling=None, recursive=False, expand_nodes=None) -> list:
"""
Check if node under path (or anything under path if recursive=True) was changed. By default
we only check if a node or subnode (recursive) was deleted from path. If expand_nodes
is set to Diff.ADD we can also check if something was added to the path.
If nothing changed, an empty list is returned.
"""
from vyos.configdiff import get_config_diff
from vyos.configdiff import Diff
# to prevent circular dependencies we assign the default here
if not expand_nodes: expand_nodes = Diff.DELETE
D = get_config_diff(conf, key_mangling)
# get_child_nodes_diff() will return dict_keys()
tmp = D.get_child_nodes_diff(path, expand_nodes=expand_nodes, recursive=recursive)
output = []
if expand_nodes & Diff.DELETE:
output.extend(list(tmp['delete'].keys()))
if expand_nodes & Diff.ADD:
output.extend(list(tmp['add'].keys()))
# remove duplicate keys from list, this happens when a node (e.g. description) is altered
output = list(dict.fromkeys(output))
return output
def get_removed_vlans(conf, path, dict):
"""
Common function to parse a dictionary retrieved via get_config_dict() and
determine any added/removed VLAN interfaces - be it 802.1q or Q-in-Q.
"""
from vyos.configdiff import get_config_diff, Diff
# Check vif, vif-s/vif-c VLAN interfaces for removal
D = get_config_diff(conf, key_mangling=('-', '_'))
D.set_level(conf.get_level())
# get_child_nodes() will return dict_keys(), mangle this into a list with PEP448
keys = D.get_child_nodes_diff(path + ['vif'], expand_nodes=Diff.DELETE)['delete'].keys()
if keys: dict['vif_remove'] = [*keys]
# get_child_nodes() will return dict_keys(), mangle this into a list with PEP448
keys = D.get_child_nodes_diff(path + ['vif-s'], expand_nodes=Diff.DELETE)['delete'].keys()
if keys: dict['vif_s_remove'] = [*keys]
for vif in dict.get('vif_s', {}).keys():
keys = D.get_child_nodes_diff(path + ['vif-s', vif, 'vif-c'], expand_nodes=Diff.DELETE)['delete'].keys()
if keys: dict['vif_s'][vif]['vif_c_remove'] = [*keys]
return dict
def is_member(conf, interface, intftype=None):
"""
Checks if passed interface is member of other interface of specified type.
intftype is optional, if not passed it will search all known types
(currently bridge and bonding)
Returns: dict
empty -> Interface is not a member
key -> Interface is a member of this interface
"""
ret_val = {}
intftypes = ['bonding', 'bridge']
if intftype not in intftypes + [None]:
raise ValueError((
f'unknown interface type "{intftype}" or it cannot '
f'have member interfaces'))
intftype = intftypes if intftype == None else [intftype]
for iftype in intftype:
base = ['interfaces', iftype]
for intf in conf.list_nodes(base):
member = base + [intf, 'member', 'interface', interface]
if conf.exists(member):
tmp = conf.get_config_dict(member, key_mangling=('-', '_'),
get_first_key=True,
no_tag_node_value_mangle=True)
ret_val.update({intf : tmp})
return ret_val
def is_mirror_intf(conf, interface, direction=None):
"""
Check whether the passed interface is used for port mirroring. Direction
is optional, if not passed it will search all known direction
(currently ingress and egress)
Returns:
None -> Interface is not a monitor interface
Array() -> This interface is a monitor interface of interfaces
"""
from vyos.ifconfig import Section
directions = ['ingress', 'egress']
if direction not in directions + [None]:
raise ValueError(f'Unknown interface mirror direction "{direction}"')
direction = directions if direction == None else [direction]
ret_val = None
base = ['interfaces']
for dir in direction:
for iftype in conf.list_nodes(base):
iftype_base = base + [iftype]
for intf in conf.list_nodes(iftype_base):
mirror = iftype_base + [intf, 'mirror', dir, interface]
if conf.exists(mirror):
path = ['interfaces', Section.section(intf), intf]
tmp = conf.get_config_dict(path, key_mangling=('-', '_'),
get_first_key=True)
ret_val = {intf : tmp}
return ret_val
def has_address_configured(conf, intf):
"""
Checks if interface has an address configured.
Checks the following config nodes:
'address', 'ipv6 address eui64', 'ipv6 address autoconf'
Returns True if interface has address configured, False if it doesn't.
"""
from vyos.ifconfig import Section
ret = False
old_level = conf.get_level()
conf.set_level([])
intfpath = ['interfaces', Section.get_config_path(intf)]
if (conf.exists([intfpath, 'address']) or
conf.exists([intfpath, 'ipv6', 'address', 'autoconf']) or
conf.exists([intfpath, 'ipv6', 'address', 'eui64'])):
ret = True
conf.set_level(old_level)
return ret
def has_vrf_configured(conf, intf):
"""
Checks if interface has a VRF configured.
Returns True if interface has VRF configured, False if it doesn't.
"""
from vyos.ifconfig import Section
ret = False
old_level = conf.get_level()
conf.set_level([])
if conf.exists(['interfaces', Section.get_config_path(intf), 'vrf']):
ret = True
conf.set_level(old_level)
return ret
def has_vlan_subinterface_configured(conf, intf):
"""
Checks if interface has an VLAN subinterface configured.
Checks the following config nodes:
'vif', 'vif-s'
Return True if interface has VLAN subinterface configured.
"""
from vyos.ifconfig import Section
ret = False
intfpath = ['interfaces', Section.section(intf), intf]
if (conf.exists(intfpath + ['vif']) or conf.exists(intfpath + ['vif-s'])):
ret = True
return ret
def is_source_interface(conf, interface, intftype=None):
"""
Checks if passed interface is configured as source-interface of other
interfaces of specified type. intftype is optional, if not passed it will
search all known types (currently pppoe, macsec, pseudo-ethernet, tunnel
and vxlan)
Returns:
None -> Interface is not a member
interface name -> Interface is a member of this interface
False -> interface type cannot have members
"""
ret_val = None
intftypes = ['macsec', 'pppoe', 'pseudo-ethernet', 'tunnel', 'vxlan']
if not intftype:
intftype = intftypes
if isinstance(intftype, str):
intftype = [intftype]
elif not isinstance(intftype, list):
raise ValueError(f'Interface type "{type(intftype)}" must be either str or list!')
if not all(x in intftypes for x in intftype):
raise ValueError(f'unknown interface type "{intftype}" or it can not '
'have a source-interface')
for it in intftype:
base = ['interfaces', it]
for intf in conf.list_nodes(base):
src_intf = base + [intf, 'source-interface']
if conf.exists(src_intf) and interface in conf.return_values(src_intf):
ret_val = intf
break
return ret_val
def get_dhcp_interfaces(conf, vrf=None):
""" Common helper functions to retrieve all interfaces from current CLI
sessions that have DHCP configured. """
dhcp_interfaces = {}
dict = conf.get_config_dict(['interfaces'], get_first_key=True)
if not dict:
return dhcp_interfaces
def check_dhcp(config):
ifname = config['ifname']
tmp = {}
if 'address' in config and 'dhcp' in config['address']:
options = {}
if dict_search('dhcp_options.default_route_distance', config) != None:
options.update({'dhcp_options' : config['dhcp_options']})
if 'vrf' in config:
if vrf == config['vrf']: tmp.update({ifname : options})
else:
if vrf is None: tmp.update({ifname : options})
return tmp
for section, interface in dict.items():
for ifname in interface:
# always reset config level, as get_interface_dict() will alter it
conf.set_level([])
# we already have a dict representation of the config from get_config_dict(),
# but with the extended information from get_interface_dict() we also
# get the DHCP client default-route-distance default option if not specified.
_, ifconfig = get_interface_dict(conf, ['interfaces', section], ifname)
tmp = check_dhcp(ifconfig)
dhcp_interfaces.update(tmp)
# check per VLAN interfaces
for vif, vif_config in ifconfig.get('vif', {}).items():
tmp = check_dhcp(vif_config)
dhcp_interfaces.update(tmp)
# check QinQ VLAN interfaces
for vif_s, vif_s_config in ifconfig.get('vif_s', {}).items():
tmp = check_dhcp(vif_s_config)
dhcp_interfaces.update(tmp)
for vif_c, vif_c_config in vif_s_config.get('vif_c', {}).items():
tmp = check_dhcp(vif_c_config)
dhcp_interfaces.update(tmp)
return dhcp_interfaces
def get_pppoe_interfaces(conf, vrf=None):
""" Common helper functions to retrieve all interfaces from current CLI
sessions that have DHCP configured. """
pppoe_interfaces = {}
conf.set_level([])
for ifname in conf.list_nodes(['interfaces', 'pppoe']):
# always reset config level, as get_interface_dict() will alter it
conf.set_level([])
# we already have a dict representation of the config from get_config_dict(),
# but with the extended information from get_interface_dict() we also
# get the DHCP client default-route-distance default option if not specified.
_, ifconfig = get_interface_dict(conf, ['interfaces', 'pppoe'], ifname)
options = {}
if 'default_route_distance' in ifconfig:
options.update({'default_route_distance' : ifconfig['default_route_distance']})
if 'no_default_route' in ifconfig:
options.update({'no_default_route' : {}})
if 'vrf' in ifconfig:
if vrf == ifconfig['vrf']: pppoe_interfaces.update({ifname : options})
else:
if vrf is None: pppoe_interfaces.update({ifname : options})
return pppoe_interfaces
def get_interface_dict(config, base, ifname='', recursive_defaults=True, with_pki=False):
"""
Common utility function to retrieve and mangle the interfaces configuration
from the CLI input nodes. All interfaces have a common base where value
retrival is identical. This function must be used whenever possible when
working on the interfaces node!
Return a dictionary with the necessary interface config keys.
"""
if not ifname:
from vyos import ConfigError
# determine tagNode instance
if 'VYOS_TAGNODE_VALUE' not in os.environ:
raise ConfigError('Interface (VYOS_TAGNODE_VALUE) not specified')
ifname = os.environ['VYOS_TAGNODE_VALUE']
# Check if interface has been removed. We must use exists() as
# get_config_dict() will always return {} - even when an empty interface
# node like the following exists.
# +macsec macsec1 {
# +}
if not config.exists(base + [ifname]):
dict = config.get_config_dict(base + [ifname], key_mangling=('-', '_'),
get_first_key=True,
no_tag_node_value_mangle=True)
dict.update({'deleted' : {}})
else:
# Get config_dict with default values
dict = config.get_config_dict(base + [ifname], key_mangling=('-', '_'),
get_first_key=True,
no_tag_node_value_mangle=True,
with_defaults=True,
with_recursive_defaults=recursive_defaults,
with_pki=with_pki)
# If interface does not request an IPv4 DHCP address there is no need
# to keep the dhcp-options key
if 'address' not in dict or 'dhcp' not in dict['address']:
if 'dhcp_options' in dict:
del dict['dhcp_options']
# Add interface instance name into dictionary
dict.update({'ifname': ifname})
# Check if QoS policy applied on this interface - See ifconfig.interface.set_mirror_redirect()
if config.exists(['qos', 'interface', ifname]):
dict.update({'traffic_policy': {}})
address = leaf_node_changed(config, base + [ifname, 'address'])
if address: dict.update({'address_old' : address})
# Check if we are a member of a bridge device
bridge = is_member(config, ifname, 'bridge')
if bridge: dict.update({'is_bridge_member' : bridge})
# Check if it is a monitor interface
mirror = is_mirror_intf(config, ifname)
if mirror: dict.update({'is_mirror_intf' : mirror})
# Check if we are a member of a bond device
bond = is_member(config, ifname, 'bonding')
if bond: dict.update({'is_bond_member' : bond})
# Check if any DHCP options changed which require a client restat
dhcp = is_node_changed(config, base + [ifname, 'dhcp-options'])
if dhcp: dict.update({'dhcp_options_changed' : {}})
# Changine interface VRF assignemnts require a DHCP restart, too
dhcp = is_node_changed(config, base + [ifname, 'vrf'])
if dhcp: dict.update({'dhcp_options_changed' : {}})
# Some interfaces come with a source_interface which must also not be part
# of any other bond or bridge interface as it is exclusivly assigned as the
# Kernels "lower" interface to this new "virtual/upper" interface.
if 'source_interface' in dict:
# Check if source interface is member of another bridge
tmp = is_member(config, dict['source_interface'], 'bridge')
if tmp: dict.update({'source_interface_is_bridge_member' : tmp})
# Check if source interface is member of another bridge
tmp = is_member(config, dict['source_interface'], 'bonding')
if tmp: dict.update({'source_interface_is_bond_member' : tmp})
mac = leaf_node_changed(config, base + [ifname, 'mac'])
if mac: dict.update({'mac_old' : mac})
eui64 = leaf_node_changed(config, base + [ifname, 'ipv6', 'address', 'eui64'])
if eui64:
tmp = dict_search('ipv6.address', dict)
if not tmp:
dict.update({'ipv6': {'address': {'eui64_old': eui64}}})
else:
dict['ipv6']['address'].update({'eui64_old': eui64})
for vif, vif_config in dict.get('vif', {}).items():
# Add subinterface name to dictionary
dict['vif'][vif].update({'ifname' : f'{ifname}.{vif}'})
if config.exists(['qos', 'interface', f'{ifname}.{vif}']):
dict['vif'][vif].update({'traffic_policy': {}})
if 'deleted' not in dict:
address = leaf_node_changed(config, base + [ifname, 'vif', vif, 'address'])
if address: dict['vif'][vif].update({'address_old' : address})
# If interface does not request an IPv4 DHCP address there is no need
# to keep the dhcp-options key
if 'address' not in dict['vif'][vif] or 'dhcp' not in dict['vif'][vif]['address']:
if 'dhcp_options' in dict['vif'][vif]:
del dict['vif'][vif]['dhcp_options']
# Check if we are a member of a bridge device
bridge = is_member(config, f'{ifname}.{vif}', 'bridge')
if bridge: dict['vif'][vif].update({'is_bridge_member' : bridge})
# Check if any DHCP options changed which require a client restat
dhcp = is_node_changed(config, base + [ifname, 'vif', vif, 'dhcp-options'])
if dhcp: dict['vif'][vif].update({'dhcp_options_changed' : {}})
for vif_s, vif_s_config in dict.get('vif_s', {}).items():
# Add subinterface name to dictionary
dict['vif_s'][vif_s].update({'ifname' : f'{ifname}.{vif_s}'})
if config.exists(['qos', 'interface', f'{ifname}.{vif_s}']):
dict['vif_s'][vif_s].update({'traffic_policy': {}})
if 'deleted' not in dict:
address = leaf_node_changed(config, base + [ifname, 'vif-s', vif_s, 'address'])
if address: dict['vif_s'][vif_s].update({'address_old' : address})
# If interface does not request an IPv4 DHCP address there is no need
# to keep the dhcp-options key
if 'address' not in dict['vif_s'][vif_s] or 'dhcp' not in \
dict['vif_s'][vif_s]['address']:
if 'dhcp_options' in dict['vif_s'][vif_s]:
del dict['vif_s'][vif_s]['dhcp_options']
# Check if we are a member of a bridge device
bridge = is_member(config, f'{ifname}.{vif_s}', 'bridge')
if bridge: dict['vif_s'][vif_s].update({'is_bridge_member' : bridge})
# Check if any DHCP options changed which require a client restat
dhcp = is_node_changed(config, base + [ifname, 'vif-s', vif_s, 'dhcp-options'])
if dhcp: dict['vif_s'][vif_s].update({'dhcp_options_changed' : {}})
for vif_c, vif_c_config in vif_s_config.get('vif_c', {}).items():
# Add subinterface name to dictionary
dict['vif_s'][vif_s]['vif_c'][vif_c].update({'ifname' : f'{ifname}.{vif_s}.{vif_c}'})
if config.exists(['qos', 'interface', f'{ifname}.{vif_s}.{vif_c}']):
dict['vif_s'][vif_s]['vif_c'][vif_c].update({'traffic_policy': {}})
if 'deleted' not in dict:
address = leaf_node_changed(config, base + [ifname, 'vif-s', vif_s, 'vif-c', vif_c, 'address'])
if address: dict['vif_s'][vif_s]['vif_c'][vif_c].update(
{'address_old' : address})
# If interface does not request an IPv4 DHCP address there is no need
# to keep the dhcp-options key
if 'address' not in dict['vif_s'][vif_s]['vif_c'][vif_c] or 'dhcp' \
not in dict['vif_s'][vif_s]['vif_c'][vif_c]['address']:
if 'dhcp_options' in dict['vif_s'][vif_s]['vif_c'][vif_c]:
del dict['vif_s'][vif_s]['vif_c'][vif_c]['dhcp_options']
# Check if we are a member of a bridge device
bridge = is_member(config, f'{ifname}.{vif_s}.{vif_c}', 'bridge')
if bridge: dict['vif_s'][vif_s]['vif_c'][vif_c].update(
{'is_bridge_member' : bridge})
# Check if any DHCP options changed which require a client restat
dhcp = is_node_changed(config, base + [ifname, 'vif-s', vif_s, 'vif-c', vif_c, 'dhcp-options'])
if dhcp: dict['vif_s'][vif_s]['vif_c'][vif_c].update({'dhcp_options_changed' : {}})
# Check vif, vif-s/vif-c VLAN interfaces for removal
dict = get_removed_vlans(config, base + [ifname], dict)
return ifname, dict
def get_vlan_ids(interface):
"""
Get the VLAN ID of the interface bound to the bridge
"""
vlan_ids = set()
bridge_status = cmd('bridge -j vlan show', shell=True)
vlan_filter_status = json.loads(bridge_status)
if vlan_filter_status is not None:
for interface_status in vlan_filter_status:
ifname = interface_status['ifname']
if interface == ifname:
vlans_status = interface_status['vlans']
for vlan_status in vlans_status:
vlan_id = vlan_status['vlan']
vlan_ids.add(vlan_id)
return vlan_ids
def get_accel_dict(config, base, chap_secrets, with_pki=False):
"""
Common utility function to retrieve and mangle the Accel-PPP configuration
from different CLI input nodes. All Accel-PPP services have a common base
where value retrival is identical. This function must be used whenever
possible when working with Accel-PPP services!
Return a dictionary with the necessary interface config keys.
"""
from vyos.utils.cpu import get_core_count
from vyos.template import is_ipv4
dict = config.get_config_dict(base, key_mangling=('-', '_'),
get_first_key=True,
no_tag_node_value_mangle=True,
with_recursive_defaults=True,
with_pki=with_pki)
# set CPUs cores to process requests
dict.update({'thread_count' : get_core_count()})
# we need to store the path to the secrets file
dict.update({'chap_secrets_file' : chap_secrets})
# We can only have two IPv4 and three IPv6 nameservers - also they are
# configured in a different way in the configuration, this is why we split
# the configuration
if 'name_server' in dict:
ns_v4 = []
ns_v6 = []
for ns in dict['name_server']:
if is_ipv4(ns): ns_v4.append(ns)
else: ns_v6.append(ns)
dict.update({'name_server_ipv4' : ns_v4, 'name_server_ipv6' : ns_v6})
del dict['name_server']
# Check option "disable-accounting" per server and replace default value from '1813' to '0'
for server in (dict_search('authentication.radius.server', dict) or []):
if 'disable_accounting' in dict['authentication']['radius']['server'][server]:
dict['authentication']['radius']['server'][server]['acct_port'] = '0'
return dict
-def get_frrender_dict(conf) -> dict:
+def get_frrender_dict(conf, argv=None) -> dict:
from copy import deepcopy
from vyos.config import config_dict_merge
from vyos.frrender import frr_protocols
# 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
# 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):
isis = 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' : isis})
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)
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' : ''}})
# 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} })
- vrf['name'][vrf_name]['protocols'].update({'bgp' : 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_effective(vrf_vni_path):
vrf_config.update({'vni': conf.return_effective_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):
+ print('======== < BEGIN > ==========')
+ import pprint
+ pprint.pprint(dict)
+ print('========= < END > ===========')
+
# Use singleton instance of the FRR render class
if hasattr(conf, 'frrender_cls'):
frrender = getattr(conf, 'frrender_cls')
dict.update({'frrender_cls' : frrender})
frrender.generate(dict)
- if os.path.exists(frr_debug_enable):
- print('======== < BEGIN > ==========')
- import pprint
- pprint.pprint(dict)
- print('========= < END > ===========')
return dict
diff --git a/python/vyos/configverify.py b/python/vyos/configverify.py
index 4450dc16b..4084425b1 100644
--- a/python/vyos/configverify.py
+++ b/python/vyos/configverify.py
@@ -1,546 +1,549 @@
# Copyright 2020-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/>.
# The sole purpose of this module is to hold common functions used in
# all kinds of implementations to verify the CLI configuration.
# It is started by migrating the interfaces to the new get_config_dict()
# approach which will lead to a lot of code that can be reused.
# NOTE: imports should be as local as possible to the function which
# makes use of it!
from vyos import ConfigError
from vyos.utils.dict import dict_search
# pattern re-used in ipsec migration script
dynamic_interface_pattern = r'(ppp|pppoe|sstpc|l2tp|ipoe)[0-9]+'
def verify_mtu(config):
"""
Common helper function used by interface implementations to perform
recurring validation if the specified MTU can be used by the underlaying
hardware.
"""
from vyos.ifconfig import Interface
if 'mtu' in config:
mtu = int(config['mtu'])
tmp = Interface(config['ifname'])
# Not all interfaces support min/max MTU
# https://vyos.dev/T5011
try:
min_mtu = tmp.get_min_mtu()
max_mtu = tmp.get_max_mtu()
except: # Fallback to defaults
min_mtu = 68
max_mtu = 9000
if mtu < min_mtu:
raise ConfigError(f'Interface MTU too low, ' \
f'minimum supported MTU is {min_mtu}!')
if mtu > max_mtu:
raise ConfigError(f'Interface MTU too high, ' \
f'maximum supported MTU is {max_mtu}!')
def verify_mtu_parent(config, parent):
if 'mtu' not in config or 'mtu' not in parent:
return
mtu = int(config['mtu'])
parent_mtu = int(parent['mtu'])
if mtu > parent_mtu:
raise ConfigError(f'Interface MTU "{mtu}" too high, ' \
f'parent interface MTU is "{parent_mtu}"!')
def verify_mtu_ipv6(config):
"""
Common helper function used by interface implementations to perform
recurring validation if the specified MTU can be used when IPv6 is
configured on the interface. IPv6 requires a 1280 bytes MTU.
"""
from vyos.template import is_ipv6
if 'mtu' in config:
# IPv6 minimum required link mtu
min_mtu = 1280
if int(config['mtu']) < min_mtu:
interface = config['ifname']
error_msg = f'IPv6 address will be configured on interface "{interface}",\n' \
f'the required minimum MTU is "{min_mtu}"!'
if 'address' in config:
for address in config['address']:
if address in ['dhcpv6'] or is_ipv6(address):
raise ConfigError(error_msg)
tmp = dict_search('ipv6.address.no_default_link_local', config)
if tmp == None: raise ConfigError('link-local ' + error_msg)
tmp = dict_search('ipv6.address.autoconf', config)
if tmp != None: raise ConfigError(error_msg)
tmp = dict_search('ipv6.address.eui64', config)
if tmp != None: raise ConfigError(error_msg)
def verify_vrf(config):
"""
Common helper function used by interface implementations to perform
recurring validation of VRF configuration.
"""
from vyos.utils.network import interface_exists
if 'vrf' in config:
vrfs = config['vrf']
if isinstance(vrfs, str):
vrfs = [vrfs]
for vrf in vrfs:
if vrf == 'default':
continue
if not interface_exists(vrf):
raise ConfigError(f'VRF "{vrf}" does not exist!')
if 'is_bridge_member' in config:
raise ConfigError(
'Interface "{ifname}" cannot be both a member of VRF "{vrf}" '
'and bridge "{is_bridge_member}"!'.format(**config))
def verify_bond_bridge_member(config):
"""
Checks if interface has a VRF configured and is also part of a bond or
bridge, which is not allowed!
"""
if 'vrf' in config:
ifname = config['ifname']
if 'is_bond_member' in config:
raise ConfigError(f'Can not add interface "{ifname}" to bond, it has a VRF assigned!')
if 'is_bridge_member' in config:
raise ConfigError(f'Can not add interface "{ifname}" to bridge, it has a VRF assigned!')
def verify_tunnel(config):
"""
This helper is used to verify the common part of the tunnel
"""
from vyos.template import is_ipv4
from vyos.template import is_ipv6
if 'encapsulation' not in config:
raise ConfigError('Must configure the tunnel encapsulation for '\
'{ifname}!'.format(**config))
if 'source_address' not in config and 'source_interface' not in config:
raise ConfigError('source-address or source-interface required for tunnel!')
if 'remote' not in config and config['encapsulation'] != 'gre':
raise ConfigError('remote ip address is mandatory for tunnel')
if config['encapsulation'] in ['ipip6', 'ip6ip6', 'ip6gre', 'ip6gretap', 'ip6erspan']:
error_ipv6 = 'Encapsulation mode requires IPv6'
if 'source_address' in config and not is_ipv6(config['source_address']):
raise ConfigError(f'{error_ipv6} source-address')
if 'remote' in config and not is_ipv6(config['remote']):
raise ConfigError(f'{error_ipv6} remote')
else:
error_ipv4 = 'Encapsulation mode requires IPv4'
if 'source_address' in config and not is_ipv4(config['source_address']):
raise ConfigError(f'{error_ipv4} source-address')
if 'remote' in config and not is_ipv4(config['remote']):
raise ConfigError(f'{error_ipv4} remote address')
if config['encapsulation'] in ['sit', 'gretap', 'ip6gretap']:
if 'source_interface' in config:
encapsulation = config['encapsulation']
raise ConfigError(f'Option source-interface can not be used with ' \
f'encapsulation "{encapsulation}"!')
elif config['encapsulation'] == 'gre':
if 'source_address' in config and is_ipv6(config['source_address']):
raise ConfigError('Can not use local IPv6 address is for mGRE tunnels')
def verify_mirror_redirect(config):
"""
Common helper function used by interface implementations to perform
recurring validation of mirror and redirect interface configuration via tc(8)
It makes no sense to mirror traffic back at yourself!
"""
from vyos.utils.network import interface_exists
if {'mirror', 'redirect'} <= set(config):
raise ConfigError('Mirror and redirect can not be enabled at the same time!')
if 'mirror' in config:
for direction, mirror_interface in config['mirror'].items():
if not interface_exists(mirror_interface):
raise ConfigError(f'Requested mirror interface "{mirror_interface}" '\
'does not exist!')
if mirror_interface == config['ifname']:
raise ConfigError(f'Can not mirror "{direction}" traffic back '\
'the originating interface!')
if 'redirect' in config:
redirect_ifname = config['redirect']
if not interface_exists(redirect_ifname):
raise ConfigError(f'Requested redirect interface "{redirect_ifname}" '\
'does not exist!')
if ('mirror' in config or 'redirect' in config) and dict_search('traffic_policy.in', config) is not None:
# XXX: support combination of limiting and redirect/mirror - this is an
# artificial limitation
raise ConfigError('Can not use ingress policy together with mirror or redirect!')
def verify_authentication(config):
"""
Common helper function used by interface implementations to perform
recurring validation of authentication for either PPPoE or WWAN interfaces.
If authentication CLI option is defined, both username and password must
be set!
"""
if 'authentication' not in config:
return
if not {'username', 'password'} <= set(config['authentication']):
raise ConfigError('Authentication requires both username and ' \
'password to be set!')
def verify_address(config):
"""
Common helper function used by interface implementations to perform
recurring validation of IP address assignment when interface is part
of a bridge or bond.
"""
if {'is_bridge_member', 'address'} <= set(config):
interface = config['ifname']
bridge_name = next(iter(config['is_bridge_member']))
raise ConfigError(f'Cannot assign address to interface "{interface}" '
f'as it is a member of bridge "{bridge_name}"!')
def verify_bridge_delete(config):
"""
Common helper function used by interface implementations to
perform recurring validation of IP address assignmenr
when interface also is part of a bridge.
"""
if 'is_bridge_member' in config:
interface = config['ifname']
bridge_name = next(iter(config['is_bridge_member']))
raise ConfigError(f'Interface "{interface}" cannot be deleted as it '
f'is a member of bridge "{bridge_name}"!')
def verify_interface_exists(config, ifname, state_required=False, warning_only=False):
"""
Common helper function used by interface implementations to perform
recurring validation if an interface actually exists. We first probe
if the interface is defined on the CLI, if it's not found we try if
it exists at the OS level.
"""
from vyos.base import Warning
from vyos.utils.dict import dict_search_recursive
from vyos.utils.network import interface_exists
if not state_required:
# Check if interface is present in CLI config
tmp = getattr(config, 'interfaces_root', {})
if bool(list(dict_search_recursive(tmp, ifname))):
return True
# Interface not found on CLI, try Linux Kernel
if interface_exists(ifname):
return True
message = f'Interface "{ifname}" does not exist!'
if warning_only:
Warning(message)
return False
raise ConfigError(message)
def verify_source_interface(config):
"""
Common helper function used by interface implementations to
perform recurring validation of the existence of a source-interface
required by e.g. peth/MACvlan, MACsec ...
"""
import re
from vyos.utils.network import interface_exists
ifname = config['ifname']
if 'source_interface' not in config:
raise ConfigError(f'Physical source-interface required for "{ifname}"!')
src_ifname = config['source_interface']
# We do not allow sourcing other interfaces (e.g. tunnel) from dynamic interfaces
tmp = re.compile(dynamic_interface_pattern)
if tmp.match(src_ifname):
raise ConfigError(f'Can not source "{ifname}" from dynamic interface "{src_ifname}"!')
if not interface_exists(src_ifname):
raise ConfigError(f'Specified source-interface {src_ifname} does not exist')
if 'source_interface_is_bridge_member' in config:
bridge_name = next(iter(config['source_interface_is_bridge_member']))
raise ConfigError(f'Invalid source-interface "{src_ifname}". Interface '
f'is already a member of bridge "{bridge_name}"!')
if 'source_interface_is_bond_member' in config:
bond_name = next(iter(config['source_interface_is_bond_member']))
raise ConfigError(f'Invalid source-interface "{src_ifname}". Interface '
f'is already a member of bond "{bond_name}"!')
if 'is_source_interface' in config:
tmp = config['is_source_interface']
raise ConfigError(f'Can not use source-interface "{src_ifname}", it already ' \
f'belongs to interface "{tmp}"!')
def verify_dhcpv6(config):
"""
Common helper function used by interface implementations to perform
recurring validation of DHCPv6 options which are mutually exclusive.
"""
if 'dhcpv6_options' in config:
if {'parameters_only', 'temporary'} <= set(config['dhcpv6_options']):
raise ConfigError('DHCPv6 temporary and parameters-only options '
'are mutually exclusive!')
# It is not allowed to have duplicate SLA-IDs as those identify an
# assigned IPv6 subnet from a delegated prefix
for pd in (dict_search('dhcpv6_options.pd', config) or []):
sla_ids = []
interfaces = dict_search(f'dhcpv6_options.pd.{pd}.interface', config)
if not interfaces:
raise ConfigError('DHCPv6-PD requires an interface where to assign '
'the delegated prefix!')
for count, interface in enumerate(interfaces):
if 'sla_id' in interfaces[interface]:
sla_ids.append(interfaces[interface]['sla_id'])
else:
sla_ids.append(str(count))
# Check for duplicates
duplicates = [x for n, x in enumerate(sla_ids) if x in sla_ids[:n]]
if duplicates:
raise ConfigError('Site-Level Aggregation Identifier (SLA-ID) '
'must be unique per prefix-delegation!')
def verify_vlan_config(config):
"""
Common helper function used by interface implementations to perform
recurring validation of interface VLANs
"""
# VLAN and Q-in-Q IDs are not allowed to overlap
if 'vif' in config and 'vif_s' in config:
duplicate = list(set(config['vif']) & set(config['vif_s']))
if duplicate:
raise ConfigError(f'Duplicate VLAN id "{duplicate[0]}" used for vif and vif-s interfaces!')
parent_ifname = config['ifname']
# 802.1q VLANs
for vlan_id in config.get('vif', {}):
vlan = config['vif'][vlan_id]
vlan['ifname'] = f'{parent_ifname}.{vlan_id}'
verify_dhcpv6(vlan)
verify_address(vlan)
verify_vrf(vlan)
verify_mirror_redirect(vlan)
verify_mtu_parent(vlan, config)
# 802.1ad (Q-in-Q) VLANs
for s_vlan_id in config.get('vif_s', {}):
s_vlan = config['vif_s'][s_vlan_id]
s_vlan['ifname'] = f'{parent_ifname}.{s_vlan_id}'
verify_dhcpv6(s_vlan)
verify_address(s_vlan)
verify_vrf(s_vlan)
verify_mirror_redirect(s_vlan)
verify_mtu_parent(s_vlan, config)
for c_vlan_id in s_vlan.get('vif_c', {}):
c_vlan = s_vlan['vif_c'][c_vlan_id]
c_vlan['ifname'] = f'{parent_ifname}.{s_vlan_id}.{c_vlan_id}'
verify_dhcpv6(c_vlan)
verify_address(c_vlan)
verify_vrf(c_vlan)
verify_mirror_redirect(c_vlan)
verify_mtu_parent(c_vlan, config)
verify_mtu_parent(c_vlan, s_vlan)
def verify_diffie_hellman_length(file, min_keysize):
""" Verify Diffie-Hellamn keypair length given via file. It must be greater
then or equal to min_keysize """
import os
import re
from vyos.utils.process import cmd
try:
keysize = str(min_keysize)
except:
return False
if os.path.exists(file):
out = cmd(f'openssl dhparam -inform PEM -in {file} -text')
prog = re.compile('\d+\s+bit')
if prog.search(out):
bits = prog.search(out)[0].split()[0]
if int(bits) >= int(min_keysize):
return True
return False
def verify_common_route_maps(config):
"""
Common helper function used by routing protocol implementations to perform
recurring validation if the specified route-map for either zebra to kernel
installation exists (this is the top-level route_map key) or when a route
is redistributed with a route-map that it exists!
"""
# XXX: This function is called in combination with a previous call to:
# tmp = conf.get_config_dict(['policy']) - see protocols_ospf.py as example.
# We should NOT call this with the key_mangling option as this would rename
# route-map hypens '-' to underscores '_' and one could no longer distinguish
# what should have been the "proper" route-map name, as foo-bar and foo_bar
# are two entire different route-map instances!
for route_map in ['route-map', 'route_map']:
if route_map not in config:
continue
tmp = config[route_map]
# Check if the specified route-map exists, if not error out
if dict_search(f'policy.route_map.{tmp}', config) == None:
raise ConfigError(f'Specified route-map "{tmp}" does not exist!')
if 'redistribute' in config:
for protocol, protocol_config in config['redistribute'].items():
if 'route_map' in protocol_config:
verify_route_map(protocol_config['route_map'], config)
def verify_route_map(route_map_name, config):
"""
Common helper function used by routing protocol implementations to perform
recurring validation if a specified route-map exists!
"""
# Check if the specified route-map exists, if not error out
if dict_search(f'policy.route_map.{route_map_name}', config) == None:
raise ConfigError(f'Specified route-map "{route_map_name}" does not exist!')
def verify_prefix_list(prefix_list, config, version=''):
"""
Common helper function used by routing protocol implementations to perform
recurring validation if a specified prefix-list exists!
"""
# Check if the specified prefix-list exists, if not error out
if dict_search(f'policy.prefix_list{version}.{prefix_list}', config) == None:
raise ConfigError(f'Specified prefix-list{version} "{prefix_list}" does not exist!')
def verify_access_list(access_list, config, version=''):
"""
Common helper function used by routing protocol implementations to perform
recurring validation if a specified prefix-list exists!
"""
# Check if the specified ACL exists, if not error out
if dict_search(f'policy.access_list{version}.{access_list}', config) == None:
raise ConfigError(f'Specified access-list{version} "{access_list}" does not exist!')
def verify_pki_certificate(config: dict, cert_name: str, no_password_protected: bool=False):
"""
Common helper function user by PKI consumers to perform recurring
validation functions for PEM based certificates
"""
if 'pki' not in config:
raise ConfigError('PKI is not configured!')
if 'certificate' not in config['pki']:
raise ConfigError('PKI does not contain any certificates!')
if cert_name not in config['pki']['certificate']:
raise ConfigError(f'Certificate "{cert_name}" not found in configuration!')
pki_cert = config['pki']['certificate'][cert_name]
if 'certificate' not in pki_cert:
raise ConfigError(f'PEM certificate for "{cert_name}" missing in configuration!')
if 'private' not in pki_cert or 'key' not in pki_cert['private']:
raise ConfigError(f'PEM private key for "{cert_name}" missing in configuration!')
if no_password_protected and 'password_protected' in pki_cert['private']:
raise ConfigError('Password protected PEM private key is not supported!')
def verify_pki_ca_certificate(config: dict, ca_name: str):
"""
Common helper function user by PKI consumers to perform recurring
validation functions for PEM based CA certificates
"""
if 'pki' not in config:
raise ConfigError('PKI is not configured!')
if 'ca' not in config['pki']:
raise ConfigError('PKI does not contain any CA certificates!')
if ca_name not in config['pki']['ca']:
raise ConfigError(f'CA Certificate "{ca_name}" not found in configuration!')
pki_cert = config['pki']['ca'][ca_name]
if 'certificate' not in pki_cert:
raise ConfigError(f'PEM CA certificate for "{cert_name}" missing in configuration!')
def verify_pki_dh_parameters(config: dict, dh_name: str, min_key_size: int=0):
"""
Common helper function user by PKI consumers to perform recurring
validation functions on DH parameters
"""
from vyos.pki import load_dh_parameters
if 'pki' not in config:
raise ConfigError('PKI is not configured!')
if 'dh' not in config['pki']:
raise ConfigError('PKI does not contain any DH parameters!')
if dh_name not in config['pki']['dh']:
raise ConfigError(f'DH parameter "{dh_name}" not found in configuration!')
if min_key_size:
pki_dh = config['pki']['dh'][dh_name]
dh_params = load_dh_parameters(pki_dh['parameters'])
dh_numbers = dh_params.parameter_numbers()
dh_bits = dh_numbers.p.bit_length()
if dh_bits < min_key_size:
raise ConfigError(f'Minimum DH key-size is {min_key_size} bits!')
def verify_eapol(config: dict):
"""
Common helper function used by interface implementations to perform
recurring validation of EAPoL configuration.
"""
if 'eapol' not in config:
return
if 'certificate' not in config['eapol']:
raise ConfigError('Certificate must be specified when using EAPoL!')
verify_pki_certificate(config, config['eapol']['certificate'], no_password_protected=True)
if 'ca_certificate' in config['eapol']:
for ca_cert in config['eapol']['ca_certificate']:
verify_pki_ca_certificate(config, ca_cert)
-def has_frr_protocol_in_dict(config_dict: dict, protocol: str, vrf: str=None) -> bool:
+def has_frr_protocol_in_dict(config_dict: dict, protocol: str) -> bool:
+ vrf = None
+ if config_dict and 'vrf_context' in config_dict:
+ vrf = config_dict['vrf_context']
if vrf and protocol in (dict_search(f'vrf.name.{vrf}.protocols', config_dict) or []):
return True
if config_dict and protocol in config_dict:
return True
return False
diff --git a/python/vyos/frrender.py b/python/vyos/frrender.py
index e02094bbb..f1bb39094 100644
--- a/python/vyos/frrender.py
+++ b/python/vyos/frrender.py
@@ -1,156 +1,157 @@
# 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 vyos.defaults import frr_debug_enable
from vyos.utils.file import write_file
from vyos.utils.process import rc_cmd
from vyos.template import render_to_string
from vyos import ConfigError
DEBUG_ON = os.path.exists(frr_debug_enable)
def debug(message):
if not DEBUG_ON:
return
print(message)
pim_daemon = 'pimd'
frr_protocols = ['babel', 'bfd', 'bgp', 'eigrp', 'isis', 'mpls', 'nhrp',
'openfabric', 'ospf', 'ospfv3', 'pim', 'pim6', 'rip',
'ripng', 'rpki', 'segment_routing', 'static']
class FRRender:
def __init__(self):
self._frr_conf = '/run/frr/config/frr.conf'
def generate(self, config):
if not isinstance(config, dict):
- raise ValueError('config must be of type dict')
+ tmp = type(config)
+ raise ValueError(f'Config must be of type "dict" and not "{tmp}"!')
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'
return output
debug('======< RENDERING CONFIG >======')
# 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:
output += 'agentx\n'
# Add routing protocols in global VRF
output += inline_helper(config)
# Interface configuration for EVPN is not VRF related
if 'interfaces' in config:
output += render_to_string('frr/evpn.mh.frr.j2', {'interfaces' : config['interfaces']})
output += '\n'
if 'vrf' in config and 'name' in config['vrf']:
output += render_to_string('frr/zebra.vrf.route-map.frr.j2', config['vrf']) + '\n'
for vrf, vrf_config in config['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'])
debug(output)
debug('======< RENDERING CONFIG COMPLETE >======')
write_file(self._frr_conf, output)
if DEBUG_ON: write_file('/tmp/frr.conf.debug', output)
def apply(self):
count = 0
count_max = 5
emsg = ''
while count < count_max:
count += 1
- print('FRR: Reloading configuration - tries:', count, 'Python class ID:', id(self))
+ debug(f'FRR: Reloading configuration - tries: {count} | Python class ID: {id(self)}')
cmdline = '/usr/lib/frr/frr-reload.py --reload'
if DEBUG_ON:
cmdline += ' --debug'
rc, emsg = rc_cmd(f'{cmdline} {self._frr_conf}')
if rc != 0:
debug('FRR configuration reload failed, retrying')
continue
debug(emsg)
debug('======< DONE APPLYING CONFIG >======')
break
if count >= count_max:
raise ConfigError(emsg)
def save_configuration():
""" T3217: Save FRR configuration to /run/frr/config/frr.conf """
return cmd('/usr/bin/vtysh -n --writeconfig')
diff --git a/src/conf_mode/policy.py b/src/conf_mode/policy.py
index 2122cb032..4abb150ac 100755
--- a/src/conf_mode/policy.py
+++ b/src/conf_mode/policy.py
@@ -1,282 +1,282 @@
#!/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/>.
from sys import exit
from vyos.config import Config
from vyos.configdict import get_frrender_dict
from vyos.configverify import has_frr_protocol_in_dict
from vyos.frrender import FRRender
from vyos.frrender import frr_protocols
from vyos.utils.dict import dict_search
from vyos import ConfigError
from vyos import airbag
airbag.enable()
def community_action_compatibility(actions: dict) -> bool:
"""
Check compatibility of values in community and large community sections
:param actions: dictionary with community
:type actions: dict
:return: true if compatible, false if not
:rtype: bool
"""
if ('none' in actions) and ('replace' in actions or 'add' in actions):
return False
if 'replace' in actions and 'add' in actions:
return False
if ('delete' in actions) and ('none' in actions or 'replace' in actions):
return False
return True
def extcommunity_action_compatibility(actions: dict) -> bool:
"""
Check compatibility of values in extended community sections
:param actions: dictionary with community
:type actions: dict
:return: true if compatible, false if not
:rtype: bool
"""
if ('none' in actions) and (
'rt' in actions or 'soo' in actions or 'bandwidth' in actions or 'bandwidth_non_transitive' in actions):
return False
if ('bandwidth_non_transitive' in actions) and ('bandwidth' not in actions):
return False
return True
def routing_policy_find(key, dictionary):
# Recursively traverse a dictionary and extract the value assigned to
# a given key as generator object. This is made for routing policies,
# thus also import/export is checked
for k, v in dictionary.items():
if k == key:
if isinstance(v, dict):
for a, b in v.items():
if a in ['import', 'export']:
yield b
else:
yield v
elif isinstance(v, dict):
for result in routing_policy_find(key, v):
yield result
elif isinstance(v, list):
for d in v:
if isinstance(d, dict):
for result in routing_policy_find(key, d):
yield result
def get_config(config=None):
if config:
conf = config
else:
conf = Config()
return get_frrender_dict(conf)
def verify(config_dict):
if not has_frr_protocol_in_dict(config_dict, 'policy'):
return None
policy_types = ['access_list', 'access_list6', 'as_path_list',
'community_list', 'extcommunity_list',
'large_community_list', 'prefix_list',
'prefix_list6', 'route_map']
policy = config_dict['policy']
for protocol in frr_protocols:
if protocol not in config_dict:
continue
if 'protocol' not in policy:
policy.update({'protocol': {}})
policy['protocol'].update({protocol : config_dict[protocol]})
for policy_type in policy_types:
# Bail out early and continue with next policy type
if policy_type not in policy:
continue
# instance can be an ACL name/number, prefix-list name or route-map name
for instance, instance_config in policy[policy_type].items():
# If no rule was found within the instance ... sad, but we can leave
# early as nothing needs to be verified
if 'rule' not in instance_config:
continue
# human readable instance name (hypen instead of underscore)
policy_hr = policy_type.replace('_', '-')
entries = []
for rule, rule_config in instance_config['rule'].items():
mandatory_error = f'must be specified for "{policy_hr} {instance} rule {rule}"!'
if 'action' not in rule_config:
raise ConfigError(f'Action {mandatory_error}')
if policy_type == 'access_list':
if 'source' not in rule_config:
raise ConfigError(f'A source {mandatory_error}')
if int(instance) in range(100, 200) or int(
instance) in range(2000, 2700):
if 'destination' not in rule_config:
raise ConfigError(
f'A destination {mandatory_error}')
if policy_type == 'access_list6':
if 'source' not in rule_config:
raise ConfigError(f'A source {mandatory_error}')
if policy_type in ['as_path_list', 'community_list',
'extcommunity_list',
'large_community_list']:
if 'regex' not in rule_config:
raise ConfigError(f'A regex {mandatory_error}')
if policy_type in ['prefix_list', 'prefix_list6']:
if 'prefix' not in rule_config:
raise ConfigError(f'A prefix {mandatory_error}')
if rule_config in entries:
raise ConfigError(
f'Rule "{rule}" contains a duplicate prefix definition!')
entries.append(rule_config)
# route-maps tend to be a bit more complex so they get their own verify() section
if 'route_map' in policy:
for route_map, route_map_config in policy['route_map'].items():
if 'rule' not in route_map_config:
continue
for rule, rule_config in route_map_config['rule'].items():
# Action 'deny' cannot be used with "continue" or "on-match"
# FRR does not validate it T4827, T6676
if rule_config['action'] == 'deny' and ('continue' in rule_config or 'on_match' in rule_config):
raise ConfigError(f'rule {rule} "continue" or "on-match" cannot be used with action deny!')
# Specified community-list must exist
tmp = dict_search('match.community.community_list',
rule_config)
if tmp and tmp not in policy.get('community_list', []):
raise ConfigError(f'community-list {tmp} does not exist!')
# Specified extended community-list must exist
tmp = dict_search('match.extcommunity', rule_config)
if tmp and tmp not in policy.get('extcommunity_list', []):
raise ConfigError(
f'extcommunity-list {tmp} does not exist!')
# Specified large-community-list must exist
tmp = dict_search('match.large_community.large_community_list',
rule_config)
if tmp and tmp not in policy.get('large_community_list', []):
raise ConfigError(
f'large-community-list {tmp} does not exist!')
# Specified prefix-list must exist
tmp = dict_search('match.ip.address.prefix_list', rule_config)
if tmp and tmp not in policy.get('prefix_list', []):
raise ConfigError(f'prefix-list {tmp} does not exist!')
# Specified prefix-list must exist
tmp = dict_search('match.ipv6.address.prefix_list',
rule_config)
if tmp and tmp not in policy.get('prefix_list6', []):
raise ConfigError(f'prefix-list6 {tmp} does not exist!')
# Specified access_list6 in nexthop must exist
tmp = dict_search('match.ipv6.nexthop.access_list',
rule_config)
if tmp and tmp not in policy.get('access_list6', []):
raise ConfigError(f'access_list6 {tmp} does not exist!')
# Specified prefix-list6 in nexthop must exist
tmp = dict_search('match.ipv6.nexthop.prefix_list',
rule_config)
if tmp and tmp not in policy.get('prefix_list6', []):
raise ConfigError(f'prefix-list6 {tmp} does not exist!')
tmp = dict_search('set.community.delete', rule_config)
if tmp and tmp not in policy.get('community_list', []):
raise ConfigError(f'community-list {tmp} does not exist!')
tmp = dict_search('set.large_community.delete',
rule_config)
if tmp and tmp not in policy.get('large_community_list', []):
raise ConfigError(
f'large-community-list {tmp} does not exist!')
if 'set' in rule_config:
rule_action = rule_config['set']
if 'community' in rule_action:
if not community_action_compatibility(
rule_action['community']):
raise ConfigError(
f'Unexpected combination between action replace, add, delete or none in community')
if 'large_community' in rule_action:
if not community_action_compatibility(
rule_action['large_community']):
raise ConfigError(
f'Unexpected combination between action replace, add, delete or none in large-community')
if 'extcommunity' in rule_action:
if not extcommunity_action_compatibility(
rule_action['extcommunity']):
raise ConfigError(
f'Unexpected combination between none, rt, soo, bandwidth, bandwidth-non-transitive in extended-community')
# When routing protocols are active some use prefix-lists, route-maps etc.
# to apply the systems routing policy to the learned or redistributed routes.
# When the "routing policy" changes and policies, route-maps etc. are deleted,
# it is our responsibility to verify that the policy can not be deleted if it
# is used by any routing protocol
# Check if any routing protocol is activated
if 'protocol' in policy:
for policy_type in policy_types:
for policy_name in list(set(routing_policy_find(policy_type, policy['protocol']))):
found = False
if policy_type in policy and policy_name in policy[policy_type]:
found = True
# BGP uses prefix-list for selecting both an IPv4 or IPv6 AFI related
# list - we need to go the extra mile here and check both prefix-lists
if policy_type == 'prefix_list' and 'prefix_list6' in policy and policy_name in \
policy['prefix_list6']:
found = True
if not found:
tmp = policy_type.replace('_', '-')
raise ConfigError(
f'Can not delete {tmp} "{policy_name}", still in use!')
return None
def generate(config_dict):
- if 'frrender_cls' not in config_dict:
+ if config_dict and 'frrender_cls' not in config_dict:
FRRender().generate(config_dict)
return None
def apply(config_dict):
- if 'frrender_cls' not in config_dict:
+ if config_dict and 'frrender_cls' not in config_dict:
FRRender().apply()
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/protocols_babel.py b/src/conf_mode/protocols_babel.py
index f458493c2..af0751e02 100755
--- a/src/conf_mode/protocols_babel.py
+++ b/src/conf_mode/protocols_babel.py
@@ -1,109 +1,109 @@
#!/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/>.
from sys import exit
from vyos.config import Config
from vyos.configdict import get_frrender_dict
from vyos.configverify import has_frr_protocol_in_dict
from vyos.configverify import verify_access_list
from vyos.configverify import verify_prefix_list
from vyos.frrender import FRRender
from vyos.utils.dict import dict_search
from vyos import ConfigError
from vyos import airbag
airbag.enable()
def get_config(config=None):
if config:
conf = config
else:
conf = Config()
return get_frrender_dict(conf)
def verify(config_dict):
if not has_frr_protocol_in_dict(config_dict, 'babel'):
return None
babel = config_dict['babel']
babel['policy'] = config_dict['policy']
# verify distribute_list
if "distribute_list" in babel:
acl_keys = {
"ipv4": [
"distribute_list.ipv4.access_list.in",
"distribute_list.ipv4.access_list.out",
],
"ipv6": [
"distribute_list.ipv6.access_list.in",
"distribute_list.ipv6.access_list.out",
]
}
prefix_list_keys = {
"ipv4": [
"distribute_list.ipv4.prefix_list.in",
"distribute_list.ipv4.prefix_list.out",
],
"ipv6":[
"distribute_list.ipv6.prefix_list.in",
"distribute_list.ipv6.prefix_list.out",
]
}
for address_family in ["ipv4", "ipv6"]:
for iface_key in babel["distribute_list"].get(address_family, {}).get("interface", {}).keys():
acl_keys[address_family].extend([
f"distribute_list.{address_family}.interface.{iface_key}.access_list.in",
f"distribute_list.{address_family}.interface.{iface_key}.access_list.out"
])
prefix_list_keys[address_family].extend([
f"distribute_list.{address_family}.interface.{iface_key}.prefix_list.in",
f"distribute_list.{address_family}.interface.{iface_key}.prefix_list.out"
])
for address_family, keys in acl_keys.items():
for key in keys:
acl = dict_search(key, babel)
if acl:
verify_access_list(acl, babel, version='6' if address_family == 'ipv6' else '')
for address_family, keys in prefix_list_keys.items():
for key in keys:
prefix_list = dict_search(key, babel)
if prefix_list:
verify_prefix_list(prefix_list, babel, version='6' if address_family == 'ipv6' else '')
def generate(config_dict):
- if 'frrender_cls' not in config_dict:
+ if config_dict and 'frrender_cls' not in config_dict:
FRRender().generate(config_dict)
return None
def apply(config_dict):
- if 'frrender_cls' not in config_dict:
+ if config_dict and 'frrender_cls' not in config_dict:
FRRender().apply()
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/protocols_bfd.py b/src/conf_mode/protocols_bfd.py
index d8b19fa0e..623801897 100755
--- a/src/conf_mode/protocols_bfd.py
+++ b/src/conf_mode/protocols_bfd.py
@@ -1,96 +1,96 @@
#!/usr/bin/env python3
#
# Copyright (C) 2019-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/>.
from vyos.config import Config
from vyos.configdict import get_frrender_dict
from vyos.configverify import verify_vrf
from vyos.configverify import has_frr_protocol_in_dict
from vyos.frrender import FRRender
from vyos.template import is_ipv6
from vyos.utils.network import is_ipv6_link_local
from vyos import ConfigError
from vyos import airbag
airbag.enable()
def get_config(config=None):
if config:
conf = config
else:
conf = Config()
return get_frrender_dict(conf)
def verify(config_dict):
if not has_frr_protocol_in_dict(config_dict, 'bfd'):
return None
bfd = config_dict['bfd']
if 'peer' in bfd:
for peer, peer_config in bfd['peer'].items():
# IPv6 link local peers require an explicit local address/interface
if is_ipv6_link_local(peer):
if 'source' not in peer_config or len(peer_config['source']) < 2:
raise ConfigError('BFD IPv6 link-local peers require explicit local address and interface setting')
# IPv6 peers require an explicit local address
if is_ipv6(peer):
if 'source' not in peer_config or 'address' not in peer_config['source']:
raise ConfigError('BFD IPv6 peers require explicit local address setting')
if 'multihop' in peer_config:
# multihop require source address
if 'source' not in peer_config or 'address' not in peer_config['source']:
raise ConfigError('BFD multihop require source address')
# multihop and echo-mode cannot be used together
if 'echo_mode' in peer_config:
raise ConfigError('BFD multihop and echo-mode cannot be used together')
# multihop doesn't accept interface names
if 'source' in peer_config and 'interface' in peer_config['source']:
raise ConfigError('BFD multihop and source interface cannot be used together')
if 'minimum_ttl' in peer_config and 'multihop' not in peer_config:
raise ConfigError('Minimum TTL is only available for multihop BFD sessions!')
if 'profile' in peer_config:
profile_name = peer_config['profile']
if 'profile' not in bfd or profile_name not in bfd['profile']:
raise ConfigError(f'BFD profile "{profile_name}" does not exist!')
if 'vrf' in peer_config:
verify_vrf(peer_config)
return None
def generate(config_dict):
- if 'frrender_cls' not in config_dict:
+ if config_dict and 'frrender_cls' not in config_dict:
FRRender().generate(config_dict)
def apply(config_dict):
- if 'frrender_cls' not in config_dict:
+ if config_dict and 'frrender_cls' not in config_dict:
FRRender().apply()
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/protocols_bgp.py b/src/conf_mode/protocols_bgp.py
index 2a4bcbadf..db3123bd3 100755
--- a/src/conf_mode/protocols_bgp.py
+++ b/src/conf_mode/protocols_bgp.py
@@ -1,576 +1,574 @@
#!/usr/bin/env python3
#
# Copyright (C) 2020-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/>.
from sys import exit
from sys import argv
from vyos.base import Warning
from vyos.config import Config
from vyos.configdict import get_frrender_dict
from vyos.configverify import has_frr_protocol_in_dict
from vyos.configverify import verify_prefix_list
from vyos.configverify import verify_route_map
from vyos.configverify import verify_vrf
from vyos.frrender import FRRender
from vyos.template import is_ip
from vyos.template import is_interface
from vyos.utils.dict import dict_search
from vyos.utils.network import get_interface_vrf
from vyos.utils.network import is_addr_assigned
from vyos.utils.process import process_named_running
from vyos import ConfigError
from vyos import airbag
airbag.enable()
-vrf = None
-if len(argv) > 1:
- vrf = argv[1]
-
def get_config(config=None):
if config:
conf = config
else:
conf = Config()
- return get_frrender_dict(conf)
+ return get_frrender_dict(conf, argv)
def verify_vrf_as_import(search_vrf_name: str, afi_name: str, vrfs_config: dict) -> bool:
"""
:param search_vrf_name: search vrf name in import list
:type search_vrf_name: str
:param afi_name: afi/safi name
:type afi_name: str
:param vrfs_config: configuration dependents vrfs
:type vrfs_config: dict
:return: if vrf in import list retrun true else false
:rtype: bool
"""
for vrf_name, vrf_config in vrfs_config.items():
import_list = dict_search(
f'protocols.bgp.address_family.{afi_name}.import.vrf',
vrf_config)
if import_list:
if search_vrf_name in import_list:
return True
return False
def verify_vrf_import_options(afi_config: dict) -> bool:
"""
Search if afi contains one of options
:param afi_config: afi/safi
:type afi_config: dict
:return: if vrf contains rd and route-target options return true else false
:rtype: bool
"""
options = [
f'rd.vpn.export',
f'route_target.vpn.import',
f'route_target.vpn.export',
f'route_target.vpn.both'
]
for option in options:
if dict_search(option, afi_config):
return True
return False
def verify_vrf_import(vrf_name: str, vrfs_config: dict, afi_name: str) -> bool:
"""
Verify if vrf exists and contain options
:param vrf_name: name of VRF
:type vrf_name: str
:param vrfs_config: dependent vrfs config
:type vrfs_config: dict
:param afi_name: afi/safi name
:type afi_name: str
:return: if vrf contains rd and route-target options return true else false
:rtype: bool
"""
if vrf_name != 'default':
verify_vrf({'vrf': vrf_name})
if dict_search(f'{vrf_name}.protocols.bgp.address_family.{afi_name}',
vrfs_config):
afi_config = \
vrfs_config[vrf_name]['protocols']['bgp']['address_family'][
afi_name]
if verify_vrf_import_options(afi_config):
return True
return False
def verify_vrflist_import(afi_name: str, afi_config: dict, vrfs_config: dict) -> bool:
"""
Call function to verify
if scpecific vrf contains rd and route-target
options return true else false
:param afi_name: afi/safi name
:type afi_name: str
:param afi_config: afi/safi configuration
:type afi_config: dict
:param vrfs_config: dependent vrfs config
:type vrfs_config:dict
:return: if vrf contains rd and route-target options return true else false
:rtype: bool
"""
for vrf_name in afi_config['import']['vrf']:
if verify_vrf_import(vrf_name, vrfs_config, afi_name):
return True
return False
def verify_remote_as(peer_config, bgp_config):
if 'remote_as' in peer_config:
return peer_config['remote_as']
if 'peer_group' in peer_config:
peer_group_name = peer_config['peer_group']
tmp = dict_search(f'peer_group.{peer_group_name}.remote_as', bgp_config)
if tmp: return tmp
if 'interface' in peer_config:
if 'remote_as' in peer_config['interface']:
return peer_config['interface']['remote_as']
if 'peer_group' in peer_config['interface']:
peer_group_name = peer_config['interface']['peer_group']
tmp = dict_search(f'peer_group.{peer_group_name}.remote_as', bgp_config)
if tmp: return tmp
if 'v6only' in peer_config['interface']:
if 'remote_as' in peer_config['interface']['v6only']:
return peer_config['interface']['v6only']['remote_as']
if 'peer_group' in peer_config['interface']['v6only']:
peer_group_name = peer_config['interface']['v6only']['peer_group']
tmp = dict_search(f'peer_group.{peer_group_name}.remote_as', bgp_config)
if tmp: return tmp
return None
def verify_afi(peer_config, bgp_config):
# If address_family configured under neighboor
if 'address_family' in peer_config:
return True
# If address_family configured under peer-group
# if neighbor interface configured
peer_group_name = None
if dict_search('interface.peer_group', peer_config):
peer_group_name = peer_config['interface']['peer_group']
elif dict_search('interface.v6only.peer_group', peer_config):
peer_group_name = peer_config['interface']['v6only']['peer_group']
# if neighbor IP configured.
if 'peer_group' in peer_config:
peer_group_name = peer_config['peer_group']
if peer_group_name:
tmp = dict_search(f'peer_group.{peer_group_name}.address_family', bgp_config)
if tmp: return True
return False
def verify(config_dict):
- global vrf
- if not has_frr_protocol_in_dict(config_dict, 'bgp', vrf):
+
+ print('====== verify() ======')
+ import pprint
+ pprint.pprint(config_dict)
+
+ if not has_frr_protocol_in_dict(config_dict, 'bgp'):
return None
+ vrf = None
+ if 'vrf_context' in config_dict:
+ vrf = config_dict['vrf_context']
+
# eqivalent of the C foo ? 'a' : 'b' statement
bgp = vrf and config_dict['vrf']['name'][vrf]['protocols']['bgp'] or config_dict['bgp']
bgp['policy'] = config_dict['policy']
- if vrf:
- bgp['vrf'] = vrf
-
if 'deleted' in bgp:
- if 'vrf' in bgp:
+ if vrf:
# Cannot delete vrf if it exists in import vrf list in other vrfs
for tmp_afi in ['ipv4_unicast', 'ipv6_unicast']:
- if verify_vrf_as_import(bgp['vrf'], tmp_afi, bgp['dependent_vrfs']):
- raise ConfigError(f'Cannot delete VRF instance "{bgp["vrf"]}", ' \
+ if verify_vrf_as_import(vrf, tmp_afi, bgp['dependent_vrfs']):
+ raise ConfigError(f'Cannot delete VRF instance "{vrf}", ' \
'unconfigure "import vrf" commands!')
else:
# We are running in the default VRF context, thus we can not delete
# our main BGP instance if there are dependent BGP VRF instances.
if 'dependent_vrfs' in bgp:
for vrf, vrf_options in bgp['dependent_vrfs'].items():
if vrf != 'default':
if dict_search('protocols.bgp', vrf_options):
dependent_vrfs = ', '.join(bgp['dependent_vrfs'].keys())
raise ConfigError(f'Cannot delete default BGP instance, ' \
f'dependent VRF instance(s): {dependent_vrfs}')
if 'vni' in vrf_options:
raise ConfigError('Cannot delete default BGP instance, ' \
'dependent L3VNI exists!')
return None
if 'system_as' not in bgp:
raise ConfigError('BGP system-as number must be defined!')
# Verify BMP
if 'bmp' in bgp:
# check bmp flag "bgpd -d -F traditional --daemon -A 127.0.0.1 -M rpki -M bmp"
if not process_named_running('bgpd', 'bmp'):
raise ConfigError(
f'"bmp" flag is not found in bgpd. Configure "set system frr bmp" and restart bgp process'
)
# check bmp target
if 'target' in bgp['bmp']:
for target, target_config in bgp['bmp']['target'].items():
if 'address' not in target_config:
raise ConfigError(f'BMP target "{target}" address must be defined!')
# Verify vrf on interface and bgp section
if 'interface' in bgp:
for interface in bgp['interface']:
error_msg = f'Interface "{interface}" belongs to different VRF instance'
tmp = get_interface_vrf(interface)
- if 'vrf' in bgp:
- if bgp['vrf'] != tmp:
- vrf = bgp['vrf']
+ if vrf:
+ if vrf != tmp:
raise ConfigError(f'{error_msg} "{vrf}"!')
elif tmp != 'default':
raise ConfigError(f'{error_msg} "{tmp}"!')
peer_groups_context = dict()
# Common verification for both peer-group and neighbor statements
for neighbor in ['neighbor', 'peer_group']:
# bail out early if there is no neighbor or peer-group statement
# this also saves one indention level
if neighbor not in bgp:
continue
for peer, peer_config in bgp[neighbor].items():
# Only regular "neighbor" statement can have a peer-group set
# Check if the configure peer-group exists
if 'peer_group' in peer_config:
peer_group = peer_config['peer_group']
if 'peer_group' not in bgp or peer_group not in bgp['peer_group']:
raise ConfigError(f'Specified peer-group "{peer_group}" for '\
f'neighbor "{neighbor}" does not exist!')
if 'remote_as' in peer_config:
is_ibgp = True
if peer_config['remote_as'] != 'internal' and \
peer_config['remote_as'] != bgp['system_as']:
is_ibgp = False
if peer_group not in peer_groups_context:
peer_groups_context[peer_group] = is_ibgp
elif peer_groups_context[peer_group] != is_ibgp:
raise ConfigError(f'Peer-group members must be '
f'all internal or all external')
if 'local_role' in peer_config:
#Ensure Local Role has only one value.
if len(peer_config['local_role']) > 1:
raise ConfigError(f'Only one local role can be specified for peer "{peer}"!')
if 'local_as' in peer_config:
if len(peer_config['local_as']) > 1:
raise ConfigError(f'Only one local-as number can be specified for peer "{peer}"!')
# Neighbor local-as override can not be the same as the local-as
# we use for this BGP instane!
asn = list(peer_config['local_as'].keys())[0]
if asn == bgp['system_as']:
raise ConfigError('Cannot have local-as same as system-as number')
# Neighbor AS specified for local-as and remote-as can not be the same
if dict_search('remote_as', peer_config) == asn and neighbor != 'peer_group':
raise ConfigError(f'Neighbor "{peer}" has local-as specified which is '\
'the same as remote-as, this is not allowed!')
# ttl-security and ebgp-multihop can't be used in the same configration
if 'ebgp_multihop' in peer_config and 'ttl_security' in peer_config:
raise ConfigError('You can not set both ebgp-multihop and ttl-security hops')
# interface and ebgp-multihop can't be used in the same configration
if 'ebgp_multihop' in peer_config and 'interface' in peer_config:
raise ConfigError(f'Ebgp-multihop can not be used with directly connected '\
f'neighbor "{peer}"')
# Check if neighbor has both override capability and strict capability match
# configured at the same time.
if 'override_capability' in peer_config and 'strict_capability_match' in peer_config:
raise ConfigError(f'Neighbor "{peer}" cannot have both override-capability and '\
'strict-capability-match configured at the same time!')
# Check spaces in the password
if 'password' in peer_config and ' ' in peer_config['password']:
raise ConfigError('Whitespace is not allowed in passwords!')
# Some checks can/must only be done on a neighbor and not a peer-group
if neighbor == 'neighbor':
# remote-as must be either set explicitly for the neighbor
# or for the entire peer-group
if not verify_remote_as(peer_config, bgp):
raise ConfigError(f'Neighbor "{peer}" remote-as must be set!')
if not verify_afi(peer_config, bgp):
Warning(f'BGP neighbor "{peer}" requires address-family!')
# Peer-group member cannot override remote-as of peer-group
if 'peer_group' in peer_config:
peer_group = peer_config['peer_group']
if 'remote_as' in peer_config and 'remote_as' in bgp['peer_group'][peer_group]:
raise ConfigError(f'Peer-group member "{peer}" cannot override remote-as of peer-group "{peer_group}"!')
if 'interface' in peer_config:
if 'peer_group' in peer_config['interface']:
peer_group = peer_config['interface']['peer_group']
if 'remote_as' in peer_config['interface'] and 'remote_as' in bgp['peer_group'][peer_group]:
raise ConfigError(f'Peer-group member "{peer}" cannot override remote-as of peer-group "{peer_group}"!')
if 'v6only' in peer_config['interface']:
if 'peer_group' in peer_config['interface']['v6only']:
peer_group = peer_config['interface']['v6only']['peer_group']
if 'remote_as' in peer_config['interface']['v6only'] and 'remote_as' in bgp['peer_group'][peer_group]:
raise ConfigError(f'Peer-group member "{peer}" cannot override remote-as of peer-group "{peer_group}"!')
# Only checks for ipv4 and ipv6 neighbors
# Check if neighbor address is assigned as system interface address
- vrf = None
vrf_error_msg = f' in default VRF!'
- if 'vrf' in bgp:
- vrf = bgp['vrf']
+ if vrf:
vrf_error_msg = f' in VRF "{vrf}"!'
if is_ip(peer) and is_addr_assigned(peer, vrf):
raise ConfigError(f'Can not configure local address as neighbor "{peer}"{vrf_error_msg}')
elif is_interface(peer):
if 'peer_group' in peer_config:
raise ConfigError(f'peer-group must be set under the interface node of "{peer}"')
if 'remote_as' in peer_config:
raise ConfigError(f'remote-as must be set under the interface node of "{peer}"')
if 'source_interface' in peer_config['interface']:
raise ConfigError(f'"source-interface" option not allowed for neighbor "{peer}"')
# Local-AS allowed only for EBGP peers
if 'local_as' in peer_config:
remote_as = verify_remote_as(peer_config, bgp)
if remote_as == bgp['system_as']:
raise ConfigError(f'local-as configured for "{peer}", allowed only for eBGP peers!')
for afi in ['ipv4_unicast', 'ipv4_multicast', 'ipv4_labeled_unicast', 'ipv4_flowspec',
'ipv6_unicast', 'ipv6_multicast', 'ipv6_labeled_unicast', 'ipv6_flowspec',
'l2vpn_evpn']:
# Bail out early if address family is not configured
if 'address_family' not in peer_config or afi not in peer_config['address_family']:
continue
# Check if neighbor has both ipv4 unicast and ipv4 labeled unicast configured at the same time.
if 'ipv4_unicast' in peer_config['address_family'] and 'ipv4_labeled_unicast' in peer_config['address_family']:
raise ConfigError(f'Neighbor "{peer}" cannot have both ipv4-unicast and ipv4-labeled-unicast configured at the same time!')
# Check if neighbor has both ipv6 unicast and ipv6 labeled unicast configured at the same time.
if 'ipv6_unicast' in peer_config['address_family'] and 'ipv6_labeled_unicast' in peer_config['address_family']:
raise ConfigError(f'Neighbor "{peer}" cannot have both ipv6-unicast and ipv6-labeled-unicast configured at the same time!')
afi_config = peer_config['address_family'][afi]
if 'conditionally_advertise' in afi_config:
if 'advertise_map' not in afi_config['conditionally_advertise']:
raise ConfigError('Must speficy advertise-map when conditionally-advertise is in use!')
# Verify advertise-map (which is a route-map) exists
verify_route_map(afi_config['conditionally_advertise']['advertise_map'], bgp)
if ('exist_map' not in afi_config['conditionally_advertise'] and
'non_exist_map' not in afi_config['conditionally_advertise']):
raise ConfigError('Must either speficy exist-map or non-exist-map when ' \
'conditionally-advertise is in use!')
if {'exist_map', 'non_exist_map'} <= set(afi_config['conditionally_advertise']):
raise ConfigError('Can not specify both exist-map and non-exist-map for ' \
'conditionally-advertise!')
if 'exist_map' in afi_config['conditionally_advertise']:
verify_route_map(afi_config['conditionally_advertise']['exist_map'], bgp)
if 'non_exist_map' in afi_config['conditionally_advertise']:
verify_route_map(afi_config['conditionally_advertise']['non_exist_map'], bgp)
# T4332: bgp deterministic-med cannot be disabled while addpath-tx-bestpath-per-AS is in use
if 'addpath_tx_per_as' in afi_config:
if dict_search('parameters.deterministic_med', bgp) == None:
raise ConfigError('addpath-tx-per-as requires BGP deterministic-med paramtere to be set!')
# Validate if configured Prefix list exists
if 'prefix_list' in afi_config:
for tmp in ['import', 'export']:
if tmp not in afi_config['prefix_list']:
# bail out early
continue
if afi == 'ipv4_unicast':
verify_prefix_list(afi_config['prefix_list'][tmp], bgp)
elif afi == 'ipv6_unicast':
verify_prefix_list(afi_config['prefix_list'][tmp], bgp, version='6')
if 'route_map' in afi_config:
for tmp in ['import', 'export']:
if tmp in afi_config['route_map']:
verify_route_map(afi_config['route_map'][tmp], bgp)
if 'route_reflector_client' in afi_config:
peer_group_as = peer_config.get('remote_as')
if peer_group_as is None or (peer_group_as != 'internal' and peer_group_as != bgp['system_as']):
raise ConfigError('route-reflector-client only supported for iBGP peers')
else:
if 'peer_group' in peer_config:
peer_group_as = dict_search(f'peer_group.{peer_group}.remote_as', bgp)
if peer_group_as is None or (peer_group_as != 'internal' and peer_group_as != bgp['system_as']):
raise ConfigError('route-reflector-client only supported for iBGP peers')
# T5833 not all AFIs are supported for VRF
if 'vrf' in bgp and 'address_family' in peer_config:
unsupported_vrf_afi = {
'ipv4_flowspec',
'ipv6_flowspec',
'ipv4_labeled_unicast',
'ipv6_labeled_unicast',
'ipv4_vpn',
'ipv6_vpn',
}
for afi in peer_config['address_family']:
if afi in unsupported_vrf_afi:
raise ConfigError(
f"VRF is not allowed for address-family '{afi.replace('_', '-')}'"
)
# Throw an error if a peer group is not configured for allow range
for prefix in dict_search('listen.range', bgp) or []:
# we can not use dict_search() here as prefix contains dots ...
if 'peer_group' not in bgp['listen']['range'][prefix]:
raise ConfigError(f'Listen range for prefix "{prefix}" has no peer group configured.')
peer_group = bgp['listen']['range'][prefix]['peer_group']
if 'peer_group' not in bgp or peer_group not in bgp['peer_group']:
raise ConfigError(f'Peer-group "{peer_group}" for listen range "{prefix}" does not exist!')
if not verify_remote_as(bgp['listen']['range'][prefix], bgp):
raise ConfigError(f'Peer-group "{peer_group}" requires remote-as to be set!')
# Throw an error if the global administrative distance parameters aren't all filled out.
if dict_search('parameters.distance.global', bgp) != None:
for key in ['external', 'internal', 'local']:
if dict_search(f'parameters.distance.global.{key}', bgp) == None:
raise ConfigError('Missing mandatory configuration option for '\
f'global administrative distance {key}!')
# TCP keepalive requires all three parameters to be set
if dict_search('parameters.tcp_keepalive', bgp) != None:
if not {'idle', 'interval', 'probes'} <= set(bgp['parameters']['tcp_keepalive']):
raise ConfigError('TCP keepalive incomplete - idle, keepalive and probes must be set')
# Address Family specific validation
if 'address_family' in bgp:
for afi, afi_config in bgp['address_family'].items():
if 'distance' in afi_config:
# Throw an error if the address family specific administrative
# distance parameters aren't all filled out.
for key in ['external', 'internal', 'local']:
if key not in afi_config['distance']:
raise ConfigError('Missing mandatory configuration option for '\
f'{afi} administrative distance {key}!')
if afi in ['ipv4_unicast', 'ipv6_unicast']:
- vrf_name = bgp['vrf'] if dict_search('vrf', bgp) else 'default'
+ vrf_name = vrf if vrf else 'default'
# Verify if currant VRF contains rd and route-target options
# and does not exist in import list in other VRFs
if dict_search(f'rd.vpn.export', afi_config):
if verify_vrf_as_import(vrf_name, afi, bgp['dependent_vrfs']):
raise ConfigError(
'Command "import vrf" conflicts with "rd vpn export" command!')
if not dict_search('parameters.router_id', bgp):
Warning(f'BGP "router-id" is required when using "rd" and "route-target"!')
if dict_search('route_target.vpn.both', afi_config):
if verify_vrf_as_import(vrf_name, afi, bgp['dependent_vrfs']):
raise ConfigError(
'Command "import vrf" conflicts with "route-target vpn both" command!')
if dict_search('route_target.vpn.export', afi_config):
raise ConfigError(
'Command "route-target vpn export" conflicts '\
'with "route-target vpn both" command!')
if dict_search('route_target.vpn.import', afi_config):
raise ConfigError(
'Command "route-target vpn import" conflicts '\
'with "route-target vpn both" command!')
if dict_search('route_target.vpn.import', afi_config):
if verify_vrf_as_import(vrf_name, afi, bgp['dependent_vrfs']):
raise ConfigError(
'Command "import vrf conflicts" with "route-target vpn import" command!')
if dict_search('route_target.vpn.export', afi_config):
if verify_vrf_as_import(vrf_name, afi, bgp['dependent_vrfs']):
raise ConfigError(
'Command "import vrf" conflicts with "route-target vpn export" command!')
# Verify if VRFs in import do not contain rd
# and route-target options
if dict_search('import.vrf', afi_config) is not None:
# Verify if VRF with import does not contain rd
# and route-target options
if verify_vrf_import_options(afi_config):
raise ConfigError(
'Please unconfigure "import vrf" commands before using vpn commands in the same VRF!')
# Verify if VRFs in import list do not contain rd
# and route-target options
if verify_vrflist_import(afi, afi_config, bgp['dependent_vrfs']):
raise ConfigError(
'Please unconfigure import vrf commands before using vpn commands in dependent VRFs!')
# FRR error: please unconfigure vpn to vrf commands before
# using import vrf commands
if 'vpn' in afi_config['import'] or dict_search('export.vpn', afi_config) != None:
raise ConfigError('Please unconfigure VPN to VRF commands before '\
'using "import vrf" commands!')
# Verify that the export/import route-maps do exist
for export_import in ['export', 'import']:
tmp = dict_search(f'route_map.vpn.{export_import}', afi_config)
if tmp: verify_route_map(tmp, bgp)
# per-vrf sid and per-af sid are mutually exclusive
if 'sid' in afi_config and 'sid' in bgp:
raise ConfigError('SID per VRF and SID per address-family are mutually exclusive!')
# Checks only required for L2VPN EVPN
if afi in ['l2vpn_evpn']:
if 'vni' in afi_config:
for vni, vni_config in afi_config['vni'].items():
if 'rd' in vni_config and 'advertise_all_vni' not in afi_config:
raise ConfigError('BGP EVPN "rd" requires "advertise-all-vni" to be set!')
if 'route_target' in vni_config and 'advertise_all_vni' not in afi_config:
raise ConfigError('BGP EVPN "route-target" requires "advertise-all-vni" to be set!')
return None
def generate(config_dict):
- if 'frrender_cls' not in config_dict:
+ if config_dict and 'frrender_cls' not in config_dict:
FRRender().generate(config_dict)
return None
def apply(config_dict):
- if 'frrender_cls' not in config_dict:
+ if config_dict and 'frrender_cls' not in config_dict:
FRRender().apply()
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/protocols_eigrp.py b/src/conf_mode/protocols_eigrp.py
index 4f56d2b94..33812d350 100755
--- a/src/conf_mode/protocols_eigrp.py
+++ b/src/conf_mode/protocols_eigrp.py
@@ -1,74 +1,73 @@
#!/usr/bin/env python3
#
# Copyright (C) 2022-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/>.
from sys import exit
from sys import argv
from vyos.config import Config
from vyos.configdict import get_frrender_dict
from vyos.configverify import has_frr_protocol_in_dict
from vyos.configverify import verify_vrf
from vyos.frrender import FRRender
from vyos import ConfigError
from vyos import airbag
airbag.enable()
-vrf = None
-if len(argv) > 1:
- vrf = argv[1]
-
def get_config(config=None):
if config:
conf = config
else:
conf = Config()
- return get_frrender_dict(conf)
+ return get_frrender_dict(conf, argv)
def verify(config_dict):
- global vrf
- if not has_frr_protocol_in_dict(config_dict, 'eigrp', vrf):
+ if not has_frr_protocol_in_dict(config_dict, 'eigrp'):
return None
+ vrf = None
+ if 'vrf_context' in config_dict:
+ vrf = config_dict['vrf_context']
+
# eqivalent of the C foo ? 'a' : 'b' statement
eigrp = vrf and config_dict['vrf']['name'][vrf]['protocols']['eigrp'] or config_dict['eigrp']
eigrp['policy'] = config_dict['policy']
if 'system_as' not in eigrp:
raise ConfigError('EIGRP system-as must be defined!')
- if 'vrf' in eigrp:
- verify_vrf(eigrp)
+ if vrf:
+ verify_vrf({'vrf': vrf})
def generate(config_dict):
- if 'frrender_cls' not in config_dict:
+ if config_dict and 'frrender_cls' not in config_dict:
FRRender().generate(config_dict)
return None
def apply(config_dict):
- if 'frrender_cls' not in config_dict:
+ if config_dict and 'frrender_cls' not in config_dict:
FRRender().apply()
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/protocols_isis.py b/src/conf_mode/protocols_isis.py
index 9e494ecc8..7a54685bb 100755
--- a/src/conf_mode/protocols_isis.py
+++ b/src/conf_mode/protocols_isis.py
@@ -1,299 +1,252 @@
#!/usr/bin/env python3
#
# Copyright (C) 2020-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/>.
from sys import exit
from sys import argv
from vyos.config import Config
from vyos.configdict import get_frrender_dict
from vyos.configverify import has_frr_protocol_in_dict
from vyos.configverify import verify_common_route_maps
from vyos.configverify import verify_interface_exists
from vyos.frrender import FRRender
from vyos.ifconfig import Interface
from vyos.utils.dict import dict_search
from vyos.utils.network import get_interface_config
from vyos import ConfigError
from vyos import airbag
airbag.enable()
-vrf = None
-if len(argv) > 1:
- vrf = argv[1]
-
def get_config(config=None):
if config:
conf = config
else:
conf = Config()
- return get_frrender_dict(conf)
-
-
- # base_path = ['protocols', 'isis']
-
- # # eqivalent of the C foo ? 'a' : 'b' statement
- # base = vrf and ['vrf', 'name', vrf, 'protocols', 'isis'] or base_path
- # isis = conf.get_config_dict(base, key_mangling=('-', '_'),
- # get_first_key=True,
- # no_tag_node_value_mangle=True)
-
- # # Assign the name of our VRF context. This MUST be done before the return
- # # statement below, else on deletion we will delete the default instance
- # # instead of the VRF instance.
- # if vrf: isis['vrf'] = vrf
-
- # # FRR has VRF support for different routing daemons. As interfaces belong
- # # to VRFs - or the global VRF, we need to check for changed interfaces so
- # # that they will be properly rendered for the FRR config. Also this eases
- # # removal of interfaces from the running configuration.
- # interfaces_removed = node_changed(conf, base + ['interface'])
- # if interfaces_removed:
- # isis['interface_removed'] = list(interfaces_removed)
-
- # # Bail out early if configuration tree does no longer exist. this must
- # # be done after retrieving the list of interfaces to be removed.
- # if not conf.exists(base):
- # isis.update({'deleted' : ''})
- # return isis
-
- # # merge in default values
- # isis = conf.merge_defaults(isis, recursive=True)
-
- # # We also need some additional information from the config, prefix-lists
- # # and route-maps for instance. They will be used in verify().
- # #
- # # XXX: one MUST always call this without the key_mangling() option! See
- # # vyos.configverify.verify_common_route_maps() for more information.
- # tmp = conf.get_config_dict(['policy'])
- # # Merge policy dict into "regular" config dict
- # isis = dict_merge(tmp, isis)
-
- return isis
+ return get_frrender_dict(conf, argv)
def verify(config_dict):
- global vrf
- if not has_frr_protocol_in_dict(config_dict, 'isis', vrf):
+ if not has_frr_protocol_in_dict(config_dict, 'isis'):
return None
+ vrf = None
+ if 'vrf_context' in config_dict:
+ vrf = config_dict['vrf_context']
+
# eqivalent of the C foo ? 'a' : 'b' statement
isis = vrf and config_dict['vrf']['name'][vrf]['protocols']['isis'] or config_dict['isis']
isis['policy'] = config_dict['policy']
if 'deleted' in isis:
return None
- if vrf:
- isis['vrf'] = vrf
-
if 'net' not in isis:
raise ConfigError('Network entity is mandatory!')
# last byte in IS-IS area address must be 0
tmp = isis['net'].split('.')
if int(tmp[-1]) != 0:
raise ConfigError('Last byte of IS-IS network entity title must always be 0!')
verify_common_route_maps(isis)
# If interface not set
if 'interface' not in isis:
raise ConfigError('Interface used for routing updates is mandatory!')
for interface in isis['interface']:
verify_interface_exists(isis, interface)
# Interface MTU must be >= configured lsp-mtu
mtu = Interface(interface).get_mtu()
area_mtu = isis['lsp_mtu']
# Recommended maximum PDU size = interface MTU - 3 bytes
recom_area_mtu = mtu - 3
if mtu < int(area_mtu) or int(area_mtu) > recom_area_mtu:
raise ConfigError(f'Interface {interface} has MTU {mtu}, ' \
f'current area MTU is {area_mtu}! \n' \
f'Recommended area lsp-mtu {recom_area_mtu} or less ' \
'(calculated on MTU size).')
- if 'vrf' in isis:
+ if vrf:
# If interface specific options are set, we must ensure that the
# interface is bound to our requesting VRF. Due to the VyOS
# priorities the interface is bound to the VRF after creation of
# the VRF itself, and before any routing protocol is configured.
- vrf = isis['vrf']
tmp = get_interface_config(interface)
if 'master' not in tmp or tmp['master'] != vrf:
raise ConfigError(f'Interface "{interface}" is not a member of VRF "{vrf}"!')
# If md5 and plaintext-password set at the same time
for password in ['area_password', 'domain_password']:
if password in isis:
if {'md5', 'plaintext_password'} <= set(isis[password]):
tmp = password.replace('_', '-')
raise ConfigError(f'Can use either md5 or plaintext-password for {tmp}!')
# If one param from delay set, but not set others
if 'spf_delay_ietf' in isis:
required_timers = ['holddown', 'init_delay', 'long_delay', 'short_delay', 'time_to_learn']
exist_timers = []
for elm_timer in required_timers:
if elm_timer in isis['spf_delay_ietf']:
exist_timers.append(elm_timer)
exist_timers = set(required_timers).difference(set(exist_timers))
if len(exist_timers) > 0:
raise ConfigError('All types of spf-delay must be configured. Missing: ' + ', '.join(exist_timers).replace('_', '-'))
# If Redistribute set, but level don't set
if 'redistribute' in isis:
proc_level = isis.get('level','').replace('-','_')
for afi in ['ipv4', 'ipv6']:
if afi not in isis['redistribute']:
continue
for proto, proto_config in isis['redistribute'][afi].items():
if 'level_1' not in proto_config and 'level_2' not in proto_config:
raise ConfigError(f'Redistribute level-1 or level-2 should be specified in ' \
f'"protocols isis redistribute {afi} {proto}"!')
for redistr_level, redistr_config in proto_config.items():
if proc_level and proc_level != 'level_1_2' and proc_level != redistr_level:
raise ConfigError(f'"protocols isis redistribute {afi} {proto} {redistr_level}" ' \
f'can not be used with \"protocols isis level {proc_level}\"!')
# Segment routing checks
if dict_search('segment_routing.global_block', isis):
g_high_label_value = dict_search('segment_routing.global_block.high_label_value', isis)
g_low_label_value = dict_search('segment_routing.global_block.low_label_value', isis)
# If segment routing global block high or low value is blank, throw error
if not (g_low_label_value or g_high_label_value):
raise ConfigError('Segment routing global-block requires both low and high value!')
# If segment routing global block low value is higher than the high value, throw error
if int(g_low_label_value) > int(g_high_label_value):
raise ConfigError('Segment routing global-block low value must be lower than high value')
if dict_search('segment_routing.local_block', isis):
if dict_search('segment_routing.global_block', isis) == None:
raise ConfigError('Segment routing local-block requires global-block to be configured!')
l_high_label_value = dict_search('segment_routing.local_block.high_label_value', isis)
l_low_label_value = dict_search('segment_routing.local_block.low_label_value', isis)
# If segment routing local-block high or low value is blank, throw error
if not (l_low_label_value or l_high_label_value):
raise ConfigError('Segment routing local-block requires both high and low value!')
# If segment routing local-block low value is higher than the high value, throw error
if int(l_low_label_value) > int(l_high_label_value):
raise ConfigError('Segment routing local-block low value must be lower than high value')
# local-block most live outside global block
global_range = range(int(g_low_label_value), int(g_high_label_value) +1)
local_range = range(int(l_low_label_value), int(l_high_label_value) +1)
# Check for overlapping ranges
if list(set(global_range) & set(local_range)):
raise ConfigError(f'Segment-Routing Global Block ({g_low_label_value}/{g_high_label_value}) '\
f'conflicts with Local Block ({l_low_label_value}/{l_high_label_value})!')
# Check for a blank or invalid value per prefix
if dict_search('segment_routing.prefix', isis):
for prefix, prefix_config in isis['segment_routing']['prefix'].items():
if 'absolute' in prefix_config:
if prefix_config['absolute'].get('value') is None:
raise ConfigError(f'Segment routing prefix {prefix} absolute value cannot be blank.')
elif 'index' in prefix_config:
if prefix_config['index'].get('value') is None:
raise ConfigError(f'Segment routing prefix {prefix} index value cannot be blank.')
# Check for explicit-null and no-php-flag configured at the same time per prefix
if dict_search('segment_routing.prefix', isis):
for prefix, prefix_config in isis['segment_routing']['prefix'].items():
if 'absolute' in prefix_config:
if ("explicit_null" in prefix_config['absolute']) and ("no_php_flag" in prefix_config['absolute']):
raise ConfigError(f'Segment routing prefix {prefix} cannot have both explicit-null '\
f'and no-php-flag configured at the same time.')
elif 'index' in prefix_config:
if ("explicit_null" in prefix_config['index']) and ("no_php_flag" in prefix_config['index']):
raise ConfigError(f'Segment routing prefix {prefix} cannot have both explicit-null '\
f'and no-php-flag configured at the same time.')
# Check for index ranges being larger than the segment routing global block
if dict_search('segment_routing.global_block', isis):
g_high_label_value = dict_search('segment_routing.global_block.high_label_value', isis)
g_low_label_value = dict_search('segment_routing.global_block.low_label_value', isis)
g_label_difference = int(g_high_label_value) - int(g_low_label_value)
if dict_search('segment_routing.prefix', isis):
for prefix, prefix_config in isis['segment_routing']['prefix'].items():
if 'index' in prefix_config:
index_size = isis['segment_routing']['prefix'][prefix]['index']['value']
if int(index_size) > int(g_label_difference):
raise ConfigError(f'Segment routing prefix {prefix} cannot have an '\
f'index base size larger than the SRGB label base.')
# Check for LFA tiebreaker index duplication
if dict_search('fast_reroute.lfa.local.tiebreaker', isis):
comparison_dictionary = {}
for item, item_options in isis['fast_reroute']['lfa']['local']['tiebreaker'].items():
for index, index_options in item_options.items():
for index_value, index_value_options in index_options.items():
if index_value not in comparison_dictionary.keys():
comparison_dictionary[index_value] = [item]
else:
comparison_dictionary[index_value].append(item)
for index, index_length in comparison_dictionary.items():
if int(len(index_length)) > 1:
raise ConfigError(f'LFA index {index} cannot have more than one tiebreaker configured.')
# Check for LFA priority-limit configured multiple times per level
if dict_search('fast_reroute.lfa.local.priority_limit', isis):
comparison_dictionary = {}
for priority, priority_options in isis['fast_reroute']['lfa']['local']['priority_limit'].items():
for level, level_options in priority_options.items():
if level not in comparison_dictionary.keys():
comparison_dictionary[level] = [priority]
else:
comparison_dictionary[level].append(priority)
for level, level_length in comparison_dictionary.items():
if int(len(level_length)) > 1:
raise ConfigError(f'LFA priority-limit on {level.replace("_", "-")} cannot have more than one priority configured.')
# Check for LFA remote prefix list configured with more than one list
if dict_search('fast_reroute.lfa.remote.prefix_list', isis):
if int(len(isis['fast_reroute']['lfa']['remote']['prefix_list'].items())) > 1:
raise ConfigError(f'LFA remote prefix-list has more than one configured. Cannot have more than one configured.')
return None
def generate(config_dict):
- if 'frrender_cls' not in config_dict:
+ if config_dict and 'frrender_cls' not in config_dict:
FRRender().generate(config_dict)
return None
def apply(config_dict):
- if 'frrender_cls' not in config_dict:
+ if config_dict and 'frrender_cls' not in config_dict:
FRRender().apply()
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/protocols_mpls.py b/src/conf_mode/protocols_mpls.py
index 12899f0b2..bab9648c4 100755
--- a/src/conf_mode/protocols_mpls.py
+++ b/src/conf_mode/protocols_mpls.py
@@ -1,139 +1,139 @@
#!/usr/bin/env python3
#
# Copyright (C) 2020-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
from sys import exit
from glob import glob
from vyos.config import Config
from vyos.configdict import get_frrender_dict
from vyos.configverify import has_frr_protocol_in_dict
from vyos.frrender import FRRender
from vyos.utils.dict import dict_search
from vyos.utils.file import read_file
from vyos.utils.system import sysctl_write
from vyos.configverify import verify_interface_exists
from vyos import ConfigError
from vyos import airbag
airbag.enable()
def get_config(config=None):
if config:
conf = config
else:
conf = Config()
return get_frrender_dict(conf)
def verify(config_dict):
if not has_frr_protocol_in_dict(config_dict, 'mpls'):
return None
mpls = config_dict['mpls']
if 'interface' in mpls:
for interface in mpls['interface']:
verify_interface_exists(mpls, interface)
# Checks to see if LDP is properly configured
if 'ldp' in mpls:
# If router ID not defined
if 'router_id' not in mpls['ldp']:
raise ConfigError('Router ID missing. An LDP router id is mandatory!')
# If interface not set
if 'interface' not in mpls['ldp']:
raise ConfigError('LDP interfaces are missing. An LDP interface is mandatory!')
# If transport addresses are not set
if not dict_search('ldp.discovery.transport_ipv4_address', mpls) and \
not dict_search('ldp.discovery.transport_ipv6_address', mpls):
raise ConfigError('LDP transport address missing!')
return None
def generate(config_dict):
- if 'frrender_cls' not in config_dict:
+ if config_dict and 'frrender_cls' not in config_dict:
FRRender().generate(config_dict)
return None
def apply(config_dict):
- if 'frrender_cls' not in config_dict:
+ if config_dict and 'frrender_cls' not in config_dict:
FRRender().apply()
if not has_frr_protocol_in_dict(config_dict, 'mpls'):
return None
mpls = config_dict['mpls']
# Set number of entries in the platform label tables
labels = '0'
if 'interface' in mpls:
labels = '1048575'
sysctl_write('net.mpls.platform_labels', labels)
# Check for changes in global MPLS options
if 'parameters' in mpls:
# Choose whether to copy IP TTL to MPLS header TTL
if 'no_propagate_ttl' in mpls['parameters']:
sysctl_write('net.mpls.ip_ttl_propagate', 0)
# Choose whether to limit maximum MPLS header TTL
if 'maximum_ttl' in mpls['parameters']:
ttl = mpls['parameters']['maximum_ttl']
sysctl_write('net.mpls.default_ttl', ttl)
else:
# Set default global MPLS options if not defined.
sysctl_write('net.mpls.ip_ttl_propagate', 1)
sysctl_write('net.mpls.default_ttl', 255)
# Enable and disable MPLS processing on interfaces per configuration
if 'interface' in mpls:
system_interfaces = []
# Populate system interfaces list with local MPLS capable interfaces
for interface in glob('/proc/sys/net/mpls/conf/*'):
system_interfaces.append(os.path.basename(interface))
# This is where the comparison is done on if an interface needs to be enabled/disabled.
for system_interface in system_interfaces:
interface_state = read_file(f'/proc/sys/net/mpls/conf/{system_interface}/input')
if '1' in interface_state:
if system_interface not in mpls['interface']:
system_interface = system_interface.replace('.', '/')
sysctl_write(f'net.mpls.conf.{system_interface}.input', 0)
elif '0' in interface_state:
if system_interface in mpls['interface']:
system_interface = system_interface.replace('.', '/')
sysctl_write(f'net.mpls.conf.{system_interface}.input', 1)
else:
system_interfaces = []
# If MPLS interfaces are not configured, set MPLS processing disabled
for interface in glob('/proc/sys/net/mpls/conf/*'):
system_interfaces.append(os.path.basename(interface))
for system_interface in system_interfaces:
system_interface = system_interface.replace('.', '/')
sysctl_write(f'net.mpls.conf.{system_interface}.input', 0)
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/protocols_openfabric.py b/src/conf_mode/protocols_openfabric.py
index 9fdcf4b50..48c89d742 100644
--- a/src/conf_mode/protocols_openfabric.py
+++ b/src/conf_mode/protocols_openfabric.py
@@ -1,109 +1,109 @@
#!/usr/bin/env python3
#
# Copyright (C) 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/>.
from sys import exit
from vyos.base import Warning
from vyos.config import Config
from vyos.configdict import get_frrender_dict
from vyos.configverify import verify_interface_exists
from vyos.configverify import has_frr_protocol_in_dict
from vyos.frrender import FRRender
from vyos import ConfigError
from vyos import airbag
airbag.enable()
def get_config(config=None):
if config:
conf = config
else:
conf = Config()
return get_frrender_dict(conf)
def verify(config_dict):
if not has_frr_protocol_in_dict(config_dict, 'openfabric'):
return None
openfabric = config_dict['openfabric']
if 'deleted' in openfabric:
return None
if 'net' not in openfabric:
raise ConfigError('Network entity is mandatory!')
# last byte in OpenFabric area address must be 0
tmp = openfabric['net'].split('.')
if int(tmp[-1]) != 0:
raise ConfigError('Last byte of OpenFabric network entity title must always be 0!')
if 'domain' not in openfabric:
raise ConfigError('OpenFabric domain name is mandatory!')
interfaces_used = []
for domain, domain_config in openfabric['domain'].items():
# If interface not set
if 'interface' not in domain_config:
raise ConfigError(f'Interface used for routing updates in OpenFabric "{domain}" is mandatory!')
for iface, iface_config in domain_config['interface'].items():
verify_interface_exists(openfabric, iface)
# interface can be activated only on one OpenFabric instance
if iface in interfaces_used:
raise ConfigError(f'Interface {iface} is already used in different OpenFabric instance!')
if 'address_family' not in iface_config or len(iface_config['address_family']) < 1:
raise ConfigError(f'Need to specify address family for the interface "{iface}"!')
# If md5 and plaintext-password set at the same time
if 'password' in iface_config:
if {'md5', 'plaintext_password'} <= set(iface_config['password']):
raise ConfigError(f'Can use either md5 or plaintext-password for password for the interface!')
if iface == 'lo' and 'passive' not in iface_config:
Warning('For loopback interface passive mode is implied!')
interfaces_used.append(iface)
# If md5 and plaintext-password set at the same time
password = 'domain_password'
if password in domain_config:
if {'md5', 'plaintext_password'} <= set(domain_config[password]):
raise ConfigError(f'Can use either md5 or plaintext-password for domain-password!')
return None
def generate(config_dict):
- if 'frrender_cls' not in config_dict:
+ if config_dict and 'frrender_cls' not in config_dict:
FRRender().generate(config_dict)
return None
def apply(config_dict):
- if 'frrender_cls' not in config_dict:
+ if config_dict and 'frrender_cls' not in config_dict:
FRRender().apply()
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/protocols_ospf.py b/src/conf_mode/protocols_ospf.py
index 07e6a5860..9d35aa007 100755
--- a/src/conf_mode/protocols_ospf.py
+++ b/src/conf_mode/protocols_ospf.py
@@ -1,197 +1,196 @@
#!/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/>.
from sys import exit
from sys import argv
from vyos.config import Config
from vyos.configdict import get_frrender_dict
from vyos.configverify import verify_common_route_maps
from vyos.configverify import verify_route_map
from vyos.configverify import verify_interface_exists
from vyos.configverify import verify_access_list
from vyos.configverify import has_frr_protocol_in_dict
from vyos.frrender import FRRender
from vyos.utils.dict import dict_search
from vyos.utils.network import get_interface_config
from vyos import ConfigError
from vyos import airbag
airbag.enable()
-vrf = None
-if len(argv) > 1:
- vrf = argv[1]
-
def get_config(config=None):
if config:
conf = config
else:
conf = Config()
- return get_frrender_dict(conf)
+ return get_frrender_dict(conf, argv)
def verify(config_dict):
- global vrf
- if not has_frr_protocol_in_dict(config_dict, 'ospf', vrf):
+ if not has_frr_protocol_in_dict(config_dict, 'ospf'):
return None
+ vrf = None
+ if 'vrf_context' in config_dict:
+ vrf = config_dict['vrf_context']
+
# eqivalent of the C foo ? 'a' : 'b' statement
ospf = vrf and config_dict['vrf']['name'][vrf]['protocols']['ospf'] or config_dict['ospf']
ospf['policy'] = config_dict['policy']
verify_common_route_maps(ospf)
# As we can have a default-information route-map, we need to validate it!
route_map_name = dict_search('default_information.originate.route_map', ospf)
if route_map_name: verify_route_map(route_map_name, ospf)
# Validate if configured Access-list exists
if 'area' in ospf:
networks = []
for area, area_config in ospf['area'].items():
if 'import_list' in area_config:
acl_import = area_config['import_list']
if acl_import: verify_access_list(acl_import, ospf)
if 'export_list' in area_config:
acl_export = area_config['export_list']
if acl_export: verify_access_list(acl_export, ospf)
if 'network' in area_config:
for network in area_config['network']:
if network in networks:
raise ConfigError(f'Network "{network}" already defined in different area!')
networks.append(network)
if 'interface' in ospf:
for interface, interface_config in ospf['interface'].items():
verify_interface_exists(ospf, interface)
# One can not use dead-interval and hello-multiplier at the same
# time. FRR will only activate the last option set via CLI.
if {'hello_multiplier', 'dead_interval'} <= set(interface_config):
raise ConfigError(f'Can not use hello-multiplier and dead-interval ' \
f'concurrently for {interface}!')
# One can not use the "network <prefix> area <id>" command and an
# per interface area assignment at the same time. FRR will error
# out using: "Please remove all network commands first."
if 'area' in ospf and 'area' in interface_config:
for area, area_config in ospf['area'].items():
if 'network' in area_config:
raise ConfigError('Can not use OSPF interface area and area ' \
'network configuration at the same time!')
# If interface specific options are set, we must ensure that the
# interface is bound to our requesting VRF. Due to the VyOS
# priorities the interface is bound to the VRF after creation of
# the VRF itself, and before any routing protocol is configured.
if vrf:
tmp = get_interface_config(interface)
if 'master' not in tmp or tmp['master'] != vrf:
raise ConfigError(f'Interface "{interface}" is not a member of VRF "{vrf}"!')
# Segment routing checks
if dict_search('segment_routing.global_block', ospf):
g_high_label_value = dict_search('segment_routing.global_block.high_label_value', ospf)
g_low_label_value = dict_search('segment_routing.global_block.low_label_value', ospf)
# If segment routing global block high or low value is blank, throw error
if not (g_low_label_value or g_high_label_value):
raise ConfigError('Segment routing global-block requires both low and high value!')
# If segment routing global block low value is higher than the high value, throw error
if int(g_low_label_value) > int(g_high_label_value):
raise ConfigError('Segment routing global-block low value must be lower than high value')
if dict_search('segment_routing.local_block', ospf):
if dict_search('segment_routing.global_block', ospf) == None:
raise ConfigError('Segment routing local-block requires global-block to be configured!')
l_high_label_value = dict_search('segment_routing.local_block.high_label_value', ospf)
l_low_label_value = dict_search('segment_routing.local_block.low_label_value', ospf)
# If segment routing local-block high or low value is blank, throw error
if not (l_low_label_value or l_high_label_value):
raise ConfigError('Segment routing local-block requires both high and low value!')
# If segment routing local-block low value is higher than the high value, throw error
if int(l_low_label_value) > int(l_high_label_value):
raise ConfigError('Segment routing local-block low value must be lower than high value')
# local-block most live outside global block
global_range = range(int(g_low_label_value), int(g_high_label_value) +1)
local_range = range(int(l_low_label_value), int(l_high_label_value) +1)
# Check for overlapping ranges
if list(set(global_range) & set(local_range)):
raise ConfigError(f'Segment-Routing Global Block ({g_low_label_value}/{g_high_label_value}) '\
f'conflicts with Local Block ({l_low_label_value}/{l_high_label_value})!')
# Check for a blank or invalid value per prefix
if dict_search('segment_routing.prefix', ospf):
for prefix, prefix_config in ospf['segment_routing']['prefix'].items():
if 'index' in prefix_config:
if prefix_config['index'].get('value') is None:
raise ConfigError(f'Segment routing prefix {prefix} index value cannot be blank.')
# Check for explicit-null and no-php-flag configured at the same time per prefix
if dict_search('segment_routing.prefix', ospf):
for prefix, prefix_config in ospf['segment_routing']['prefix'].items():
if 'index' in prefix_config:
if ("explicit_null" in prefix_config['index']) and ("no_php_flag" in prefix_config['index']):
raise ConfigError(f'Segment routing prefix {prefix} cannot have both explicit-null '\
f'and no-php-flag configured at the same time.')
# Check for index ranges being larger than the segment routing global block
if dict_search('segment_routing.global_block', ospf):
g_high_label_value = dict_search('segment_routing.global_block.high_label_value', ospf)
g_low_label_value = dict_search('segment_routing.global_block.low_label_value', ospf)
g_label_difference = int(g_high_label_value) - int(g_low_label_value)
if dict_search('segment_routing.prefix', ospf):
for prefix, prefix_config in ospf['segment_routing']['prefix'].items():
if 'index' in prefix_config:
index_size = ospf['segment_routing']['prefix'][prefix]['index']['value']
if int(index_size) > int(g_label_difference):
raise ConfigError(f'Segment routing prefix {prefix} cannot have an '\
f'index base size larger than the SRGB label base.')
# Check route summarisation
if 'summary_address' in ospf:
for prefix, prefix_options in ospf['summary_address'].items():
if {'tag', 'no_advertise'} <= set(prefix_options):
raise ConfigError(f'Can not set both "tag" and "no-advertise" for Type-5 '\
f'and Type-7 route summarisation of "{prefix}"!')
return None
def generate(config_dict):
- if 'frrender_cls' not in config_dict:
+ if config_dict and 'frrender_cls' not in config_dict:
FRRender().generate(config_dict)
return None
def apply(config_dict):
- if 'frrender_cls' not in config_dict:
+ if config_dict and 'frrender_cls' not in config_dict:
FRRender().apply()
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/protocols_ospfv3.py b/src/conf_mode/protocols_ospfv3.py
index 9af85cabf..c6b042b54 100755
--- a/src/conf_mode/protocols_ospfv3.py
+++ b/src/conf_mode/protocols_ospfv3.py
@@ -1,108 +1,107 @@
#!/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/>.
from sys import exit
from sys import argv
from vyos.config import Config
from vyos.configdict import get_frrender_dict
from vyos.configverify import verify_common_route_maps
from vyos.configverify import verify_route_map
from vyos.configverify import verify_interface_exists
from vyos.configverify import has_frr_protocol_in_dict
from vyos.frrender import FRRender
from vyos.ifconfig import Interface
from vyos.utils.dict import dict_search
from vyos.utils.network import get_interface_config
from vyos import ConfigError
from vyos import airbag
airbag.enable()
-vrf = None
-if len(argv) > 1:
- vrf = argv[1]
-
def get_config(config=None):
if config:
conf = config
else:
conf = Config()
- return get_frrender_dict(conf)
+ return get_frrender_dict(conf, argv)
def verify(config_dict):
- global vrf
- if not has_frr_protocol_in_dict(config_dict, 'ospfv3', vrf):
+ if not has_frr_protocol_in_dict(config_dict, 'ospfv3'):
return None
+ vrf = None
+ if 'vrf_context' in config_dict:
+ vrf = config_dict['vrf_context']
+
# eqivalent of the C foo ? 'a' : 'b' statement
ospfv3 = vrf and config_dict['vrf']['name'][vrf]['protocols']['ospfv3'] or config_dict['ospfv3']
ospfv3['policy'] = config_dict['policy']
verify_common_route_maps(ospfv3)
# As we can have a default-information route-map, we need to validate it!
route_map_name = dict_search('default_information.originate.route_map', ospfv3)
if route_map_name: verify_route_map(route_map_name, ospfv3)
if 'area' in ospfv3:
for area, area_config in ospfv3['area'].items():
if 'area_type' in area_config:
if len(area_config['area_type']) > 1:
raise ConfigError(f'Can only configure one area-type for OSPFv3 area "{area}"!')
if 'range' in area_config:
for range, range_config in area_config['range'].items():
if {'not_advertise', 'advertise'} <= range_config.keys():
raise ConfigError(f'"not-advertise" and "advertise" for "range {range}" cannot be both configured at the same time!')
if 'interface' in ospfv3:
for interface, interface_config in ospfv3['interface'].items():
verify_interface_exists(ospfv3, interface)
if 'ifmtu' in interface_config:
mtu = Interface(interface).get_mtu()
if int(interface_config['ifmtu']) > int(mtu):
raise ConfigError(f'OSPFv3 ifmtu can not exceed physical MTU of "{mtu}"')
# If interface specific options are set, we must ensure that the
# interface is bound to our requesting VRF. Due to the VyOS
# priorities the interface is bound to the VRF after creation of
# the VRF itself, and before any routing protocol is configured.
if vrf:
tmp = get_interface_config(interface)
if 'master' not in tmp or tmp['master'] != vrf:
raise ConfigError(f'Interface "{interface}" is not a member of VRF "{vrf}"!')
return None
def generate(config_dict):
- if 'frrender_cls' not in config_dict:
+ if config_dict and 'frrender_cls' not in config_dict:
FRRender().generate(config_dict)
return None
def apply(config_dict):
- if 'frrender_cls' not in config_dict:
+ if config_dict and 'frrender_cls' not in config_dict:
FRRender().apply()
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/protocols_pim.py b/src/conf_mode/protocols_pim.py
index df0e82d69..c40b9d86a 100755
--- a/src/conf_mode/protocols_pim.py
+++ b/src/conf_mode/protocols_pim.py
@@ -1,118 +1,118 @@
#!/usr/bin/env python3
#
# Copyright (C) 2020-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
from ipaddress import IPv4Address
from ipaddress import IPv4Network
from signal import SIGTERM
from sys import exit
from vyos.config import Config
from vyos.configdict import get_frrender_dict
from vyos.configverify import verify_interface_exists
from vyos.configverify import has_frr_protocol_in_dict
from vyos.frrender import FRRender
from vyos.frrender import pim_daemon
from vyos.utils.process import process_named_running
from vyos.utils.process import call
from vyos import ConfigError
from vyos import airbag
airbag.enable()
def get_config(config=None):
if config:
conf = config
else:
conf = Config()
return get_frrender_dict(conf)
def verify(config_dict):
if not has_frr_protocol_in_dict(config_dict, 'pim'):
return None
pim = config_dict['pim']
if 'deleted' in pim:
return None
if 'igmp_proxy_enabled' in pim:
raise ConfigError('IGMP proxy and PIM cannot be configured at the same time!')
if 'interface' not in pim:
raise ConfigError('PIM require defined interfaces!')
RESERVED_MC_NET = '224.0.0.0/24'
for interface, interface_config in pim['interface'].items():
verify_interface_exists(pim, interface)
# Check join group in reserved net
if 'igmp' in interface_config and 'join' in interface_config['igmp']:
for join_addr in interface_config['igmp']['join']:
if IPv4Address(join_addr) in IPv4Network(RESERVED_MC_NET):
raise ConfigError(f'Groups within {RESERVED_MC_NET} are reserved and cannot be joined!')
if 'rp' in pim:
if 'address' not in pim['rp']:
raise ConfigError('PIM rendezvous point needs to be defined!')
# Check unique multicast groups
unique = []
pim_base_error = 'PIM rendezvous point group'
for address, address_config in pim['rp']['address'].items():
if 'group' not in address_config:
raise ConfigError(f'{pim_base_error} should be defined for "{address}"!')
# Check if it is a multicast group
for gr_addr in address_config['group']:
if not IPv4Network(gr_addr).is_multicast:
raise ConfigError(f'{pim_base_error} "{gr_addr}" is not a multicast group!')
if gr_addr in unique:
raise ConfigError(f'{pim_base_error} must be unique!')
unique.append(gr_addr)
def generate(config_dict):
- if 'frrender_cls' not in config_dict:
+ if config_dict and 'frrender_cls' not in config_dict:
FRRender().generate(config_dict)
return None
def apply(config_dict):
if not has_frr_protocol_in_dict(config_dict, 'pim'):
return None
pim_pid = process_named_running(pim_daemon)
pim = config_dict['pim']
if 'deleted' in pim:
os.kill(int(pim_pid), SIGTERM)
return None
if not pim_pid:
call('/usr/lib/frr/pimd -d -F traditional --daemon -A 127.0.0.1')
- if 'frrender_cls' not in config_dict:
+ if config_dict and 'frrender_cls' not in config_dict:
FRRender().apply()
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/protocols_pim6.py b/src/conf_mode/protocols_pim6.py
index a5d612814..ff8fa38c3 100755
--- a/src/conf_mode/protocols_pim6.py
+++ b/src/conf_mode/protocols_pim6.py
@@ -1,95 +1,95 @@
#!/usr/bin/env python3
#
# Copyright (C) 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 ipaddress import IPv6Address
from ipaddress import IPv6Network
from sys import exit
from vyos.config import Config
from vyos.configdict import get_frrender_dict
from vyos.configverify import has_frr_protocol_in_dict
from vyos.configverify import verify_interface_exists
from vyos.frrender import FRRender
from vyos import ConfigError
from vyos import airbag
airbag.enable()
def get_config(config=None):
if config:
conf = config
else:
conf = Config()
return get_frrender_dict(conf)
def verify(config_dict):
if not has_frr_protocol_in_dict(config_dict, 'pim6'):
return None
pim6 = config_dict['pim6']
if 'deleted' in pim6:
return None
for interface, interface_config in pim6.get('interface', {}).items():
verify_interface_exists(pim6, interface)
if 'mld' in interface_config:
mld = interface_config['mld']
for group in mld.get('join', {}).keys():
# Validate multicast group address
if not IPv6Address(group).is_multicast:
raise ConfigError(f"{group} is not a multicast group")
if 'rp' in pim6:
if 'address' not in pim6['rp']:
raise ConfigError('PIM6 rendezvous point needs to be defined!')
# Check unique multicast groups
unique = []
pim_base_error = 'PIM6 rendezvous point group'
if {'address', 'prefix-list6'} <= set(pim6['rp']):
raise ConfigError(f'{pim_base_error} supports either address or a prefix-list!')
for address, address_config in pim6['rp']['address'].items():
if 'group' not in address_config:
raise ConfigError(f'{pim_base_error} should be defined for "{address}"!')
# Check if it is a multicast group
for gr_addr in address_config['group']:
if not IPv6Network(gr_addr).is_multicast:
raise ConfigError(f'{pim_base_error} "{gr_addr}" is not a multicast group!')
if gr_addr in unique:
raise ConfigError(f'{pim_base_error} must be unique!')
unique.append(gr_addr)
def generate(config_dict):
- if 'frrender_cls' not in config_dict:
+ if config_dict and 'frrender_cls' not in config_dict:
FRRender().generate(config_dict)
return None
def apply(config_dict):
- if 'frrender_cls' not in config_dict:
+ if config_dict and 'frrender_cls' not in config_dict:
FRRender().apply()
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/protocols_rip.py b/src/conf_mode/protocols_rip.py
index 7eb060504..38108c9f5 100755
--- a/src/conf_mode/protocols_rip.py
+++ b/src/conf_mode/protocols_rip.py
@@ -1,88 +1,88 @@
#!/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/>.
from sys import exit
from vyos.config import Config
from vyos.configdict import get_frrender_dict
from vyos.configverify import has_frr_protocol_in_dict
from vyos.configverify import verify_common_route_maps
from vyos.configverify import verify_access_list
from vyos.configverify import verify_prefix_list
from vyos.frrender import FRRender
from vyos.utils.dict import dict_search
from vyos import ConfigError
from vyos import airbag
airbag.enable()
def get_config(config=None):
if config:
conf = config
else:
conf = Config()
return get_frrender_dict(conf)
def verify(config_dict):
if not has_frr_protocol_in_dict(config_dict, 'rip'):
return None
rip = config_dict['rip']
rip['policy'] = config_dict['policy']
verify_common_route_maps(rip)
acl_in = dict_search('distribute_list.access_list.in', rip)
if acl_in: verify_access_list(acl_in, rip)
acl_out = dict_search('distribute_list.access_list.out', rip)
if acl_out: verify_access_list(acl_out, rip)
prefix_list_in = dict_search('distribute_list.prefix-list.in', rip)
if prefix_list_in: verify_prefix_list(prefix_list_in, rip)
prefix_list_out = dict_search('distribute_list.prefix_list.out', rip)
if prefix_list_out: verify_prefix_list(prefix_list_out, rip)
if 'interface' in rip:
for interface, interface_options in rip['interface'].items():
if 'authentication' in interface_options:
if {'md5', 'plaintext_password'} <= set(interface_options['authentication']):
raise ConfigError('Can not use both md5 and plaintext-password at the same time!')
if 'split_horizon' in interface_options:
if {'disable', 'poison_reverse'} <= set(interface_options['split_horizon']):
raise ConfigError(f'You can not have "split-horizon poison-reverse" enabled ' \
f'with "split-horizon disable" for "{interface}"!')
def generate(config_dict):
- if 'frrender_cls' not in config_dict:
+ if config_dict and 'frrender_cls' not in config_dict:
FRRender().generate(config_dict)
return None
def apply(config_dict):
- if 'frrender_cls' not in config_dict:
+ if config_dict and 'frrender_cls' not in config_dict:
FRRender().apply()
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/protocols_ripng.py b/src/conf_mode/protocols_ripng.py
index 5884e61f9..46f3ce014 100755
--- a/src/conf_mode/protocols_ripng.py
+++ b/src/conf_mode/protocols_ripng.py
@@ -1,88 +1,88 @@
#!/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/>.
from sys import exit
from vyos.config import Config
from vyos.configdict import get_frrender_dict
from vyos.configverify import has_frr_protocol_in_dict
from vyos.configverify import verify_common_route_maps
from vyos.configverify import verify_access_list
from vyos.configverify import verify_prefix_list
from vyos.frrender import FRRender
from vyos.utils.dict import dict_search
from vyos import ConfigError
from vyos import airbag
airbag.enable()
def get_config(config=None):
if config:
conf = config
else:
conf = Config()
return get_frrender_dict(conf)
def verify(config_dict):
if not has_frr_protocol_in_dict(config_dict, 'ripng'):
return None
ripng = config_dict['ripng']
ripng['policy'] = config_dict['policy']
verify_common_route_maps(ripng)
acl_in = dict_search('distribute_list.access_list.in', ripng)
if acl_in: verify_access_list(acl_in, ripng, version='6')
acl_out = dict_search('distribute_list.access_list.out', ripng)
if acl_out: verify_access_list(acl_out, ripng, version='6')
prefix_list_in = dict_search('distribute_list.prefix_list.in', ripng)
if prefix_list_in: verify_prefix_list(prefix_list_in, ripng, version='6')
prefix_list_out = dict_search('distribute_list.prefix_list.out', ripng)
if prefix_list_out: verify_prefix_list(prefix_list_out, ripng, version='6')
if 'interface' in ripng:
for interface, interface_options in ripng['interface'].items():
if 'authentication' in interface_options:
if {'md5', 'plaintext_password'} <= set(interface_options['authentication']):
raise ConfigError('Can not use both md5 and plaintext-password at the same time!')
if 'split_horizon' in interface_options:
if {'disable', 'poison_reverse'} <= set(interface_options['split_horizon']):
raise ConfigError(f'You can not have "split-horizon poison-reverse" enabled ' \
f'with "split-horizon disable" for "{interface}"!')
def generate(config_dict):
- if 'frrender_cls' not in config_dict:
+ if config_dict and 'frrender_cls' not in config_dict:
FRRender().generate(config_dict)
return None
def apply(config_dict):
- if 'frrender_cls' not in config_dict:
+ if config_dict and 'frrender_cls' not in config_dict:
FRRender().apply()
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/protocols_rpki.py b/src/conf_mode/protocols_rpki.py
index 33696a742..d3f515feb 100755
--- a/src/conf_mode/protocols_rpki.py
+++ b/src/conf_mode/protocols_rpki.py
@@ -1,114 +1,114 @@
#!/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
from glob import glob
from sys import exit
from vyos.config import Config
from vyos.configdict import get_frrender_dict
from vyos.configverify import has_frr_protocol_in_dict
from vyos.frrender import FRRender
from vyos.pki import wrap_openssh_public_key
from vyos.pki import wrap_openssh_private_key
from vyos.utils.dict import dict_search_args
from vyos.utils.file import write_file
from vyos import ConfigError
from vyos import airbag
airbag.enable()
rpki_ssh_key_base = '/run/frr/id_rpki'
def get_config(config=None):
if config:
conf = config
else:
conf = Config()
return get_frrender_dict(conf)
def verify(config_dict):
if not has_frr_protocol_in_dict(config_dict, 'rpki'):
return None
rpki = config_dict['rpki']
if 'cache' in rpki:
preferences = []
for peer, peer_config in rpki['cache'].items():
for mandatory in ['port', 'preference']:
if mandatory not in peer_config:
raise ConfigError(f'RPKI cache "{peer}" {mandatory} must be defined!')
if 'preference' in peer_config:
preference = peer_config['preference']
if preference in preferences:
raise ConfigError(f'RPKI cache with preference {preference} already configured!')
preferences.append(preference)
if 'ssh' in peer_config:
if 'username' not in peer_config['ssh']:
raise ConfigError('RPKI+SSH requires username to be defined!')
if 'key' not in peer_config['ssh'] or 'openssh' not in rpki['pki']:
raise ConfigError('RPKI+SSH requires key to be defined!')
if peer_config['ssh']['key'] not in rpki['pki']['openssh']:
raise ConfigError('RPKI+SSH key not found on PKI subsystem!')
return None
def generate(config_dict):
for key in glob(f'{rpki_ssh_key_base}*'):
os.unlink(key)
if not has_frr_protocol_in_dict(config_dict, 'rpki'):
return None
rpki = config_dict['rpki']
if 'cache' in rpki:
for cache, cache_config in rpki['cache'].items():
if 'ssh' in cache_config:
key_name = cache_config['ssh']['key']
public_key_data = dict_search_args(rpki['pki'], 'openssh', key_name, 'public', 'key')
public_key_type = dict_search_args(rpki['pki'], 'openssh', key_name, 'public', 'type')
private_key_data = dict_search_args(rpki['pki'], 'openssh', key_name, 'private', 'key')
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}'
write_file(cache_config['ssh']['public_key_file'], wrap_openssh_public_key(public_key_data, public_key_type))
write_file(cache_config['ssh']['private_key_file'], wrap_openssh_private_key(private_key_data))
- if 'frrender_cls' not in config_dict:
+ if config_dict and 'frrender_cls' not in config_dict:
FRRender().generate(config_dict)
return None
def apply(config_dict):
- if 'frrender_cls' not in config_dict:
+ if config_dict and 'frrender_cls' not in config_dict:
FRRender().apply()
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/protocols_segment-routing.py b/src/conf_mode/protocols_segment-routing.py
index a776d1038..cc42e5ac7 100755
--- a/src/conf_mode/protocols_segment-routing.py
+++ b/src/conf_mode/protocols_segment-routing.py
@@ -1,103 +1,103 @@
#!/usr/bin/env python3
#
# Copyright (C) 2023-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/>.
from sys import exit
from vyos.config import Config
from vyos.configdict import get_frrender_dict
from vyos.configdict import list_diff
from vyos.configverify import has_frr_protocol_in_dict
from vyos.frrender import FRRender
from vyos.ifconfig import Section
from vyos.utils.dict import dict_search
from vyos.utils.system import sysctl_write
from vyos import ConfigError
from vyos import airbag
airbag.enable()
def get_config(config=None):
if config:
conf = config
else:
conf = Config()
return get_frrender_dict(conf)
def verify(config_dict):
if not has_frr_protocol_in_dict(config_dict, 'segment_routing'):
return None
sr = config_dict['segment_routing']
if 'srv6' in sr:
srv6_enable = False
if 'interface' in sr:
for interface, interface_config in sr['interface'].items():
if 'srv6' in interface_config:
srv6_enable = True
break
if not srv6_enable:
raise ConfigError('SRv6 should be enabled on at least one interface!')
return None
def generate(config_dict):
- if 'frrender_cls' not in config_dict:
+ if config_dict and 'frrender_cls' not in config_dict:
FRRender().generate(config_dict)
return None
def apply(config_dict):
if not has_frr_protocol_in_dict(config_dict, 'segment_routing'):
return None
sr = config_dict['segment_routing']
current_interfaces = Section.interfaces()
sr_interfaces = list(sr.get('interface', {}).keys())
for interface in list_diff(current_interfaces, sr_interfaces):
# Disable processing of IPv6-SR packets
sysctl_write(f'net.ipv6.conf.{interface}.seg6_enabled', '0')
for interface, interface_config in sr.get('interface', {}).items():
# Accept or drop SR-enabled IPv6 packets on this interface
if 'srv6' in interface_config:
sysctl_write(f'net.ipv6.conf.{interface}.seg6_enabled', '1')
# Define HMAC policy for ingress SR-enabled packets on this interface
# It's a redundant check as HMAC has a default value - but better safe
# then sorry
tmp = dict_search('srv6.hmac', interface_config)
if tmp == 'accept':
sysctl_write(f'net.ipv6.conf.{interface}.seg6_require_hmac', '0')
elif tmp == 'drop':
sysctl_write(f'net.ipv6.conf.{interface}.seg6_require_hmac', '1')
elif tmp == 'ignore':
sysctl_write(f'net.ipv6.conf.{interface}.seg6_require_hmac', '-1')
else:
sysctl_write(f'net.ipv6.conf.{interface}.seg6_enabled', '0')
- if 'frrender_cls' not in config_dict:
+ if config_dict and 'frrender_cls' not in config_dict:
FRRender().apply()
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/protocols_static.py b/src/conf_mode/protocols_static.py
index 69500377c..29fa530f4 100755
--- a/src/conf_mode/protocols_static.py
+++ b/src/conf_mode/protocols_static.py
@@ -1,112 +1,114 @@
#!/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/>.
from ipaddress import IPv4Network
from sys import exit
from sys import argv
from vyos.config import Config
from vyos.configdict import get_frrender_dict
from vyos.configverify import has_frr_protocol_in_dict
from vyos.configverify import verify_common_route_maps
from vyos.configverify import verify_vrf
from vyos.frrender import FRRender
from vyos.template import render
from vyos import ConfigError
from vyos import airbag
airbag.enable()
-vrf = None
-if len(argv) > 1:
- vrf = argv[1]
-
config_file = '/etc/iproute2/rt_tables.d/vyos-static.conf'
def get_config(config=None):
if config:
conf = config
else:
conf = Config()
- return get_frrender_dict(conf)
+ return get_frrender_dict(conf, argv)
def verify(config_dict):
- global vrf
- if not has_frr_protocol_in_dict(config_dict, 'static', vrf):
+ if not has_frr_protocol_in_dict(config_dict, 'static'):
return None
+ vrf = None
+ if 'vrf_context' in config_dict:
+ vrf = config_dict['vrf_context']
+
# eqivalent of the C foo ? 'a' : 'b' statement
static = vrf and config_dict['vrf']['name'][vrf]['protocols']['static'] or config_dict['static']
static['policy'] = config_dict['policy']
verify_common_route_maps(static)
for route in ['route', 'route6']:
# if there is no route(6) key in the dictionary we can immediately
# bail out early
if route not in static:
continue
# When leaking routes to other VRFs we must ensure that the destination
# VRF exists
for prefix, prefix_options in static[route].items():
# both the interface and next-hop CLI node can have a VRF subnode,
# thus we check this using a for loop
for type in ['interface', 'next_hop']:
if type in prefix_options:
for interface, interface_config in prefix_options[type].items():
verify_vrf(interface_config)
if {'blackhole', 'reject'} <= set(prefix_options):
raise ConfigError(f'Can not use both blackhole and reject for '\
f'prefix "{prefix}"!')
if 'multicast' in static and 'route' in static['multicast']:
for prefix, prefix_options in static['multicast']['route'].items():
if not IPv4Network(prefix).is_multicast:
raise ConfigError(f'{prefix} is not a multicast network!')
return None
def generate(config_dict):
- global vrf
- if not has_frr_protocol_in_dict(config_dict, 'static', vrf):
+ if not has_frr_protocol_in_dict(config_dict, 'static'):
return None
+ vrf = None
+ if 'vrf_context' in config_dict:
+ vrf = config_dict['vrf_context']
+
# eqivalent of the C foo ? 'a' : 'b' statement
static = vrf and config_dict['vrf']['name'][vrf]['protocols']['static'] or config_dict['static']
# Put routing table names in /etc/iproute2/rt_tables
render(config_file, 'iproute2/static.conf.j2', static)
- if 'frrender_cls' not in config_dict:
+ if config_dict and 'frrender_cls' not in config_dict:
FRRender().generate(config_dict)
return None
-def apply(static):
- if 'frrender_cls' not in config_dict:
+def apply(config_dict):
+ if config_dict and 'frrender_cls' not in config_dict:
FRRender().apply()
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/services/vyos-configd b/src/services/vyos-configd
index 21d91005a..ecad85801 100755
--- a/src/services/vyos-configd
+++ b/src/services/vyos-configd
@@ -1,338 +1,338 @@
#!/usr/bin/env python3
#
# Copyright (C) 2020-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/>.
# pylint: disable=redefined-outer-name
import os
import sys
import grp
import re
import json
import typing
import logging
import signal
import traceback
import importlib.util
import io
from contextlib import redirect_stdout
import zmq
from vyos.defaults import directories
from vyos.utils.boot import boot_configuration_complete
from vyos.configsource import ConfigSourceString
from vyos.configsource import ConfigSourceError
from vyos.configdiff import get_commit_scripts
from vyos.config import Config
from vyos.frrender import FRRender
from vyos import ConfigError
CFG_GROUP = 'vyattacfg'
script_stdout_log = '/tmp/vyos-configd-script-stdout'
debug = True
logger = logging.getLogger(__name__)
logs_handler = logging.StreamHandler()
logger.addHandler(logs_handler)
if debug:
logger.setLevel(logging.DEBUG)
else:
logger.setLevel(logging.INFO)
SOCKET_PATH = 'ipc:///run/vyos-configd.sock'
MAX_MSG_SIZE = 65535
PAD_MSG_SIZE = 6
# Response error codes
R_SUCCESS = 1
R_ERROR_COMMIT = 2
R_ERROR_DAEMON = 4
R_PASS = 8
vyos_conf_scripts_dir = directories['conf_mode']
configd_include_file = os.path.join(directories['data'], 'configd-include.json')
configd_env_set_file = os.path.join(directories['data'], 'vyos-configd-env-set')
configd_env_unset_file = os.path.join(directories['data'], 'vyos-configd-env-unset')
# sourced on entering config session
configd_env_file = '/etc/default/vyos-configd-env'
def key_name_from_file_name(f):
return os.path.splitext(f)[0]
def module_name_from_key(k):
return k.replace('-', '_')
def path_from_file_name(f):
return os.path.join(vyos_conf_scripts_dir, f)
# opt-in to be run by daemon
with open(configd_include_file) as f:
try:
include = json.load(f)
except OSError as e:
logger.critical(f'configd include file error: {e}')
sys.exit(1)
except json.JSONDecodeError as e:
logger.critical(f'JSON load error: {e}')
sys.exit(1)
# import conf_mode scripts
(_, _, filenames) = next(iter(os.walk(vyos_conf_scripts_dir)))
filenames.sort()
load_filenames = [f for f in filenames if f in include]
imports = [key_name_from_file_name(f) for f in load_filenames]
module_names = [module_name_from_key(k) for k in imports]
paths = [path_from_file_name(f) for f in load_filenames]
to_load = list(zip(module_names, paths))
modules = []
for x in to_load:
spec = importlib.util.spec_from_file_location(x[0], x[1])
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
modules.append(module)
conf_mode_scripts = dict(zip(imports, modules))
exclude_set = {key_name_from_file_name(f) for f in filenames if f not in include}
include_set = {key_name_from_file_name(f) for f in filenames if f in include}
def write_stdout_log(file_name, msg):
if boot_configuration_complete():
return
with open(file_name, 'a') as f:
f.write(msg)
def run_script(script_name, config, args) -> tuple[int, str]:
# pylint: disable=broad-exception-caught
script = conf_mode_scripts[script_name]
script.argv = args
config.set_level([])
try:
c = script.get_config(config)
script.verify(c)
script.generate(c)
script.apply(c)
except ConfigError as e:
logger.error(e)
return R_ERROR_COMMIT, str(e)
except Exception:
tb = traceback.format_exc()
logger.error(tb)
return R_ERROR_COMMIT, tb
return R_SUCCESS, ''
def initialization(socket):
# pylint: disable=broad-exception-caught,too-many-locals
# Reset config strings:
active_string = ''
session_string = ''
# check first for resent init msg, in case of client timeout
while True:
msg = socket.recv().decode('utf-8', 'ignore')
try:
message = json.loads(msg)
if message['type'] == 'init':
resp = 'init'
socket.send(resp.encode())
except Exception:
break
# zmq synchronous for ipc from single client:
active_string = msg
resp = 'active'
socket.send(resp.encode())
session_string = socket.recv().decode('utf-8', 'ignore')
resp = 'session'
socket.send(resp.encode())
pid_string = socket.recv().decode('utf-8', 'ignore')
resp = 'pid'
socket.send(resp.encode())
sudo_user_string = socket.recv().decode('utf-8', 'ignore')
resp = 'sudo_user'
socket.send(resp.encode())
temp_config_dir_string = socket.recv().decode('utf-8', 'ignore')
resp = 'temp_config_dir'
socket.send(resp.encode())
changes_only_dir_string = socket.recv().decode('utf-8', 'ignore')
resp = 'changes_only_dir'
socket.send(resp.encode())
logger.debug(f'config session pid is {pid_string}')
logger.debug(f'config session sudo_user is {sudo_user_string}')
os.environ['SUDO_USER'] = sudo_user_string
if temp_config_dir_string:
os.environ['VYATTA_TEMP_CONFIG_DIR'] = temp_config_dir_string
if changes_only_dir_string:
os.environ['VYATTA_CHANGES_ONLY_DIR'] = changes_only_dir_string
try:
configsource = ConfigSourceString(running_config_text=active_string,
session_config_text=session_string)
except ConfigSourceError as e:
logger.debug(e)
return None
config = Config(config_source=configsource)
dependent_func: dict[str, list[typing.Callable]] = {}
setattr(config, 'dependent_func', dependent_func)
commit_scripts = get_commit_scripts(config)
logger.debug(f'commit_scripts: {commit_scripts}')
scripts_called = []
setattr(config, 'scripts_called', scripts_called)
if not hasattr(config, 'frrender_cls'):
setattr(config, 'frrender_cls', FRRender())
return config
def process_node_data(config, data, _last: bool = False) -> tuple[int, str]:
if not config:
out = 'Empty config'
logger.critical(out)
return R_ERROR_DAEMON, out
script_name = None
os.environ['VYOS_TAGNODE_VALUE'] = ''
args = []
config.dependency_list.clear()
res = re.match(r'^(VYOS_TAGNODE_VALUE=[^/]+)?.*\/([^/]+).py(.*)', data)
if res.group(1):
env = res.group(1).split('=')
os.environ[env[0]] = env[1]
if res.group(2):
script_name = res.group(2)
if not script_name:
out = 'Missing script_name'
logger.critical(out)
return R_ERROR_DAEMON, out
if res.group(3):
args = res.group(3).split()
args.insert(0, f'{script_name}.py')
tag_value = os.getenv('VYOS_TAGNODE_VALUE', '')
tag_ext = f'_{tag_value}' if tag_value else ''
script_record = f'{script_name}{tag_ext}'
scripts_called = getattr(config, 'scripts_called', [])
scripts_called.append(script_record)
if script_name not in include_set:
return R_PASS, ''
with redirect_stdout(io.StringIO()) as o:
result, err_out = run_script(script_name, config, args)
amb_out = o.getvalue()
o.close()
out = amb_out + err_out
return result, out
def send_result(sock, err, msg):
msg = msg if msg else ''
msg_size = min(MAX_MSG_SIZE, len(msg))
err_rep = err.to_bytes(1)
msg_size_rep = f'{msg_size:#0{PAD_MSG_SIZE}x}'
logger.debug(f'Sending reply: error_code {err} with output')
sock.send_multipart([err_rep, msg_size_rep.encode(), msg.encode()])
write_stdout_log(script_stdout_log, msg)
def remove_if_file(f: str):
try:
os.remove(f)
except FileNotFoundError:
pass
def shutdown():
remove_if_file(configd_env_file)
os.symlink(configd_env_unset_file, configd_env_file)
sys.exit(0)
if __name__ == '__main__':
context = zmq.Context()
socket = context.socket(zmq.REP)
# Set the right permissions on the socket, then change it back
o_mask = os.umask(0)
socket.bind(SOCKET_PATH)
os.umask(o_mask)
cfg_group = grp.getgrnam(CFG_GROUP)
os.setgid(cfg_group.gr_gid)
os.environ['VYOS_CONFIGD'] = 't'
def sig_handler(signum, frame):
# pylint: disable=unused-argument
shutdown()
signal.signal(signal.SIGTERM, sig_handler)
signal.signal(signal.SIGINT, sig_handler)
# Define the vyshim environment variable
remove_if_file(configd_env_file)
os.symlink(configd_env_set_file, configd_env_file)
config = None
while True:
# Wait for next request from client
msg = socket.recv().decode()
logger.debug(f'Received message: {msg}')
message = json.loads(msg)
if message['type'] == 'init':
resp = 'init'
socket.send(resp.encode())
config = initialization(socket)
elif message['type'] == 'node':
res, out = process_node_data(config, message['data'], message['last'])
send_result(socket, res, out)
if message['last'] and config:
scripts_called = getattr(config, 'scripts_called', [])
logger.debug(f'scripts_called: {scripts_called}')
- if hasattr(config, 'frrender_cls'):
+ if hasattr(config, 'frrender_cls') and res == R_SUCCESS:
frrender_cls = getattr(config, 'frrender_cls')
frrender_cls.apply()
else:
logger.critical(f'Unexpected message: {message}')