diff --git a/python/vyos/configdict.py b/python/vyos/configdict.py
index 9618ec93e..1205342df 100644
--- a/python/vyos/configdict.py
+++ b/python/vyos/configdict.py
@@ -1,690 +1,624 @@
# Copyright 2019-2022 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.util import dict_search
from vyos.xml import defaults
from vyos.util 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):
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):
"""
Check if a leaf node was altered. If it has been altered - values has been
changed, or it was added/removed, we will return the old value. If nothing
has been changed, None is returned
"""
from vyos.configdiff import get_config_diff, Diff
D = get_config_diff(conf, key_mangling)
# get_child_nodes() will return dict_keys(), mangle this into a list with PEP448
keys = D.get_child_nodes_diff(path, expand_nodes=Diff.DELETE, recursive=recursive)['delete'].keys()
return list(keys)
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 T2665_set_dhcpv6pd_defaults(config_dict):
""" Properly configure DHCPv6 default options in the dictionary. If there is
no DHCPv6 configured at all, it is safe to remove the entire configuration.
"""
# As this is the same for every interface type it is safe to assume this
# for ethernet
pd_defaults = defaults(['interfaces', 'ethernet', 'dhcpv6-options', 'pd'])
# Implant default dictionary for DHCPv6-PD instances
if dict_search('dhcpv6_options.pd.length', config_dict):
del config_dict['dhcpv6_options']['pd']['length']
for pd in (dict_search('dhcpv6_options.pd', config_dict) or []):
config_dict['dhcpv6_options']['pd'][pd] = dict_merge(pd_defaults,
config_dict['dhcpv6_options']['pd'][pd])
return config_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
"""
from vyos.ifconfig import Section
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_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):
"""
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)
# 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' : {}})
# 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):
"""
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.util import get_half_cpus
from vyos.template import is_ipv4
dict = config.get_config_dict(base, key_mangling=('-', '_'),
get_first_key=True,
- no_tag_node_value_mangle=True)
-
- # 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 = defaults(base)
-
- # T2665: defaults include RADIUS server specifics per TAG node which need to
- # be added to individual RADIUS servers instead - so we can simply delete them
- if dict_search('authentication.radius.server', default_values):
- del default_values['authentication']['radius']['server']
-
- # T2665: defaults include static-ip address per TAG node which need to be
- # added to individual local users instead - so we can simply delete them
- if dict_search('authentication.local_users.username', default_values):
- del default_values['authentication']['local_users']['username']
-
- # T2665: defaults include IPv6 client-pool mask per TAG node which need to be
- # added to individual local users instead - so we can simply delete them
- if dict_search('client_ipv6_pool.prefix.mask', default_values):
- del default_values['client_ipv6_pool']['prefix']['mask']
- # delete empty dicts
- if len (default_values['client_ipv6_pool']['prefix']) == 0:
- del default_values['client_ipv6_pool']['prefix']
- if len (default_values['client_ipv6_pool']) == 0:
- del default_values['client_ipv6_pool']
-
- # T2665: IPoE only - it has an interface tag node
- # added to individual local users instead - so we can simply delete them
- if dict_search('authentication.interface', default_values):
- del default_values['authentication']['interface']
- if dict_search('interface', default_values):
- del default_values['interface']
-
- dict = dict_merge(default_values, dict)
+ no_tag_node_value_mangle=True,
+ with_recursive_defaults=True)
# set CPUs cores to process requests
dict.update({'thread_count' : get_half_cpus()})
# 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']
- # T2665: Add individual RADIUS server default values
- if dict_search('authentication.radius.server', dict):
- default_values = defaults(base + ['authentication', 'radius', 'server'])
- for server in dict_search('authentication.radius.server', dict):
- dict['authentication']['radius']['server'][server] = dict_merge(
- default_values, dict['authentication']['radius']['server'][server])
-
- # Check option "disable-accounting" per server and replace default value from '1813' to '0'
- # set vpn sstp authentication radius server x.x.x.x disable-accounting
- if 'disable_accounting' in dict['authentication']['radius']['server'][server]:
- dict['authentication']['radius']['server'][server]['acct_port'] = '0'
-
- # T2665: Add individual local-user default values
- if dict_search('authentication.local_users.username', dict):
- default_values = defaults(base + ['authentication', 'local-users', 'username'])
- for username in dict_search('authentication.local_users.username', dict):
- dict['authentication']['local_users']['username'][username] = dict_merge(
- default_values, dict['authentication']['local_users']['username'][username])
-
- # T2665: Add individual IPv6 client-pool default mask if required
- if dict_search('client_ipv6_pool.prefix', dict):
- default_values = defaults(base + ['client-ipv6-pool', 'prefix'])
- for prefix in dict_search('client_ipv6_pool.prefix', dict):
- dict['client_ipv6_pool']['prefix'][prefix] = dict_merge(
- default_values, dict['client_ipv6_pool']['prefix'][prefix])
-
- # T2665: IPoE only - add individual local-user default values
- if dict_search('authentication.interface', dict):
- default_values = defaults(base + ['authentication', 'interface'])
- for interface in dict_search('authentication.interface', dict):
- dict['authentication']['interface'][interface] = dict_merge(
- default_values, dict['authentication']['interface'][interface])
-
- if dict_search('interface', dict):
- default_values = defaults(base + ['interface'])
- for interface in dict_search('interface', dict):
- dict['interface'][interface] = dict_merge(default_values,
- dict['interface'][interface])
+ # 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
diff --git a/python/vyos/xml_ref/definition.py b/python/vyos/xml_ref/definition.py
index 7fd7a7b77..33a49ca69 100644
--- a/python/vyos/xml_ref/definition.py
+++ b/python/vyos/xml_ref/definition.py
@@ -1,276 +1,276 @@
# Copyright 2023 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/>.
from typing import Optional, Union, Any
class Xml:
def __init__(self):
self.ref = {}
def define(self, ref: dict):
self.ref = ref
def _get_ref_node_data(self, node: dict, data: str) -> Union[bool, str]:
res = node.get('node_data', {})
if not res:
raise ValueError("non-existent node data")
if data not in res:
raise ValueError("non-existent data field")
return res.get(data)
def _get_ref_path(self, path: list) -> dict:
ref_path = path.copy()
d = self.ref
while ref_path and d:
d = d.get(ref_path[0], {})
ref_path.pop(0)
if self._is_tag_node(d) and ref_path:
ref_path.pop(0)
return d
def _is_tag_node(self, node: dict) -> bool:
res = self._get_ref_node_data(node, 'node_type')
return res == 'tag'
def is_tag(self, path: list) -> bool:
ref_path = path.copy()
d = self.ref
while ref_path and d:
d = d.get(ref_path[0], {})
ref_path.pop(0)
if self._is_tag_node(d) and ref_path:
if len(ref_path) == 1:
return False
ref_path.pop(0)
return self._is_tag_node(d)
def is_tag_value(self, path: list) -> bool:
if len(path) < 2:
return False
return self.is_tag(path[:-1])
def _is_multi_node(self, node: dict) -> bool:
b = self._get_ref_node_data(node, 'multi')
assert isinstance(b, bool)
return b
def is_multi(self, path: list) -> bool:
d = self._get_ref_path(path)
return self._is_multi_node(d)
def _is_valueless_node(self, node: dict) -> bool:
b = self._get_ref_node_data(node, 'valueless')
assert isinstance(b, bool)
return b
def is_valueless(self, path: list) -> bool:
d = self._get_ref_path(path)
return self._is_valueless_node(d)
def _is_leaf_node(self, node: dict) -> bool:
res = self._get_ref_node_data(node, 'node_type')
return res == 'leaf'
def is_leaf(self, path: list) -> bool:
d = self._get_ref_path(path)
return self._is_leaf_node(d)
@staticmethod
def _dict_get(d: dict, path: list) -> dict:
for i in path:
d = d.get(i, {})
if not isinstance(d, dict):
return {}
if not d:
break
return d
def _dict_find(self, d: dict, key: str, non_local=False) -> bool:
for k in list(d):
if k in ('node_data', 'component_version'):
continue
if k == key:
return True
if non_local and isinstance(d[k], dict):
if self._dict_find(d[k], key):
return True
return False
def cli_defined(self, path: list, node: str, non_local=False) -> bool:
d = self._dict_get(self.ref, path)
return self._dict_find(d, node, non_local=non_local)
def component_version(self) -> dict:
d = {}
for k, v in self.ref['component_version']:
d[k] = int(v)
return d
def multi_to_list(self, rpath: list, conf: dict) -> dict:
if rpath and rpath[-1] in list(conf):
raise ValueError('rpath should be disjoint from conf keys')
res: Any = {}
for k in list(conf):
d = self._get_ref_path(rpath + [k])
if self._is_leaf_node(d):
if self._is_multi_node(d) and not isinstance(conf[k], list):
res[k] = [conf[k]]
else:
res[k] = conf[k]
else:
res[k] = self.multi_to_list(rpath + [k], conf[k])
return res
def _get_default_value(self, node: dict) -> Optional[str]:
return self._get_ref_node_data(node, "default_value")
def _get_default(self, node: dict) -> Optional[Union[str, list]]:
default = self._get_default_value(node)
if default is None:
return None
- if self._is_multi_node(node) and not isinstance(default, list):
- return [default]
+ if self._is_multi_node(node):
+ return default.split()
return default
def get_defaults(self, path: list, get_first_key=False, recursive=False) -> dict:
"""Return dict containing default values below path
Note that descent below path will not proceed beyond an encountered
tag node, as no tag node value is known. For a default dict relative
to an existing config dict containing tag node values, see function:
'relative_defaults'
"""
res: dict = {}
if self.is_tag(path):
return res
d = self._get_ref_path(path)
if self._is_leaf_node(d):
default_value = self._get_default(d)
if default_value is not None:
return {path[-1]: default_value} if path else {}
for k in list(d):
if k in ('node_data', 'component_version') :
continue
if self._is_leaf_node(d[k]):
default_value = self._get_default(d[k])
if default_value is not None:
res |= {k: default_value}
elif self.is_tag(path + [k]):
# tag node defaults are used as suggestion, not default value;
# should this change, append to path and continue if recursive
pass
else:
if recursive:
pos = self.get_defaults(path + [k], recursive=True)
res |= pos
if res:
if get_first_key or not path:
return res
return {path[-1]: res}
return {}
def _well_defined(self, path: list, conf: dict) -> bool:
# test disjoint path + conf for sensible config paths
def step(c):
return [next(iter(c.keys()))] if c else []
try:
tmp = step(conf)
if tmp and self.is_tag_value(path + tmp):
c = conf[tmp[0]]
if not isinstance(c, dict):
raise ValueError
tmp = tmp + step(c)
self._get_ref_path(path + tmp)
else:
self._get_ref_path(path + tmp)
except ValueError:
return False
return True
# use local copy of function in module configdict, to avoid circular
# import
def _dict_merge(self, source, destination):
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] = self._dict_merge(source[key], tmp[key])
return tmp
def _relative_defaults(self, rpath: list, conf: dict, recursive=False) -> dict:
res: dict = {}
res = self.get_defaults(rpath, recursive=recursive,
get_first_key=True)
for k in list(conf):
if isinstance(conf[k], dict):
step = self._relative_defaults(rpath + [k], conf=conf[k],
recursive=recursive)
res |= step
if res:
return {rpath[-1]: res} if rpath else res
return {}
def relative_defaults(self, path: list, conf: dict, get_first_key=False,
recursive=False) -> dict:
"""Return dict containing defaults along paths of a config dict
"""
if not conf:
return self.get_defaults(path, get_first_key=get_first_key,
recursive=recursive)
if path and path[-1] in list(conf):
conf = conf[path[-1]]
conf = {} if not isinstance(conf, dict) else conf
if not self._well_defined(path, conf):
print('path to config dict does not define full config paths')
return {}
res = self._relative_defaults(path, conf, recursive=recursive)
if get_first_key and path:
if res.values():
res = next(iter(res.values()))
else:
res = {}
return res
def merge_defaults(self, path: list, conf: dict, get_first_key=False,
recursive=False) -> dict:
"""Return config dict with defaults non-destructively merged
This merges non-recursive defaults relative to the config dict.
"""
d = self.relative_defaults(path, conf, get_first_key=get_first_key,
recursive=recursive)
d = self._dict_merge(d, conf)
return d