diff --git a/python/vyos/configdict.py b/python/vyos/configdict.py
index 075ffe466..6a421485f 100644
--- a/python/vyos/configdict.py
+++ b/python/vyos/configdict.py
@@ -1,649 +1,663 @@
# 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.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):
+def node_changed(conf, path, key_mangling=None, recursive=False, expand_nodes=None) -> list:
"""
- 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
+ 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, Diff
+ 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() 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)
+ # 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()))
+ 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
"""
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_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):
"""
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' : {}})
# 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):
"""
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.system 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,
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']
# 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/src/conf_mode/firewall.py b/src/conf_mode/firewall.py
index ceed0cf31..da6724fde 100755
--- a/src/conf_mode/firewall.py
+++ b/src/conf_mode/firewall.py
@@ -1,508 +1,508 @@
#!/usr/bin/env python3
#
# Copyright (C) 2021-2023 VyOS maintainers and contributors
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License version 2 or later as
# published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import os
import re
from glob import glob
from json import loads
from sys import exit
from vyos.base import Warning
from vyos.config import Config
-from vyos.configdict import node_changed
+from vyos.configdict import is_node_changed
from vyos.configdiff import get_config_diff, Diff
from vyos.configdep import set_dependents, call_dependents
from vyos.configverify import verify_interface_exists
from vyos.ethtool import Ethtool
from vyos.firewall import fqdn_config_parse
from vyos.firewall import geoip_update
from vyos.template import render
from vyos.utils.process import call
from vyos.utils.process import cmd
from vyos.utils.dict import dict_search_args
from vyos.utils.dict import dict_search_recursive
from vyos.utils.process import process_named_running
from vyos.utils.process import rc_cmd
from vyos import ConfigError
from vyos import airbag
airbag.enable()
nat_conf_script = 'nat.py'
policy_route_conf_script = 'policy-route.py'
nftables_conf = '/run/nftables.conf'
sysfs_config = {
'all_ping': {'sysfs': '/proc/sys/net/ipv4/icmp_echo_ignore_all', 'enable': '0', 'disable': '1'},
'broadcast_ping': {'sysfs': '/proc/sys/net/ipv4/icmp_echo_ignore_broadcasts', 'enable': '0', 'disable': '1'},
'ip_src_route': {'sysfs': '/proc/sys/net/ipv4/conf/*/accept_source_route'},
'ipv6_receive_redirects': {'sysfs': '/proc/sys/net/ipv6/conf/*/accept_redirects'},
'ipv6_src_route': {'sysfs': '/proc/sys/net/ipv6/conf/*/accept_source_route', 'enable': '0', 'disable': '-1'},
'log_martians': {'sysfs': '/proc/sys/net/ipv4/conf/all/log_martians'},
'receive_redirects': {'sysfs': '/proc/sys/net/ipv4/conf/*/accept_redirects'},
'send_redirects': {'sysfs': '/proc/sys/net/ipv4/conf/*/send_redirects'},
'syn_cookies': {'sysfs': '/proc/sys/net/ipv4/tcp_syncookies'},
'twa_hazards_protection': {'sysfs': '/proc/sys/net/ipv4/tcp_rfc1337'}
}
valid_groups = [
'address_group',
'domain_group',
'network_group',
'port_group',
'interface_group'
]
nested_group_types = [
'address_group', 'network_group', 'mac_group',
'port_group', 'ipv6_address_group', 'ipv6_network_group'
]
snmp_change_type = {
'unknown': 0,
'add': 1,
'delete': 2,
'change': 3
}
snmp_event_source = 1
snmp_trap_mib = 'VYATTA-TRAP-MIB'
snmp_trap_name = 'mgmtEventTrap'
def geoip_updated(conf, firewall):
diff = get_config_diff(conf)
node_diff = diff.get_child_nodes_diff(['firewall'], expand_nodes=Diff.DELETE, recursive=True)
out = {
'name': [],
'ipv6_name': [],
'deleted_name': [],
'deleted_ipv6_name': []
}
updated = False
for key, path in dict_search_recursive(firewall, 'geoip'):
set_name = f'GEOIP_CC_{path[1]}_{path[2]}_{path[4]}'
if (path[0] == 'ipv4'):
out['name'].append(set_name)
elif (path[0] == 'ipv6'):
set_name = f'GEOIP_CC6_{path[1]}_{path[2]}_{path[4]}'
out['ipv6_name'].append(set_name)
updated = True
if 'delete' in node_diff:
for key, path in dict_search_recursive(node_diff['delete'], 'geoip'):
set_name = f'GEOIP_CC_{path[1]}_{path[2]}_{path[4]}'
if (path[0] == 'ipv4'):
out['deleted_name'].append(set_name)
elif (path[0] == 'ipv6'):
set_name = f'GEOIP_CC_{path[1]}_{path[2]}_{path[4]}'
out['deleted_ipv6_name'].append(set_name)
updated = True
if updated:
return out
return False
def get_config(config=None):
if config:
conf = config
else:
conf = Config()
base = ['firewall']
firewall = conf.get_config_dict(base, key_mangling=('-', '_'),
no_tag_node_value_mangle=True,
get_first_key=True,
with_recursive_defaults=True)
- firewall['group_resync'] = bool('group' in firewall or node_changed(conf, base + ['group']))
+ firewall['group_resync'] = bool('group' in firewall or is_node_changed(conf, base + ['group']))
if firewall['group_resync']:
# Update nat and policy-route as firewall groups were updated
set_dependents('group_resync', conf)
firewall['geoip_updated'] = geoip_updated(conf, firewall)
fqdn_config_parse(firewall)
set_dependents('conntrack', conf)
return firewall
def verify_rule(firewall, rule_conf, ipv6):
if 'action' not in rule_conf:
raise ConfigError('Rule action must be defined')
if 'jump' in rule_conf['action'] and 'jump_target' not in rule_conf:
raise ConfigError('Action set to jump, but no jump-target specified')
if 'jump_target' in rule_conf:
if 'jump' not in rule_conf['action']:
raise ConfigError('jump-target defined, but action jump needed and it is not defined')
target = rule_conf['jump_target']
if not ipv6:
if target not in dict_search_args(firewall, 'ipv4', 'name'):
raise ConfigError(f'Invalid jump-target. Firewall name {target} does not exist on the system')
else:
if target not in dict_search_args(firewall, 'ipv6', 'name'):
raise ConfigError(f'Invalid jump-target. Firewall ipv6 name {target} does not exist on the system')
if rule_conf['action'] == 'offload':
if 'offload_target' not in rule_conf:
raise ConfigError('Action set to offload, but no offload-target specified')
offload_target = rule_conf['offload_target']
if not dict_search_args(firewall, 'flowtable', offload_target):
raise ConfigError(f'Invalid offload-target. Flowtable "{offload_target}" does not exist on the system')
if rule_conf['action'] != 'synproxy' and 'synproxy' in rule_conf:
raise ConfigError('"synproxy" option allowed only for action synproxy')
if rule_conf['action'] == 'synproxy':
if 'state' in rule_conf:
raise ConfigError('For action "synproxy" state cannot be defined')
if not rule_conf.get('synproxy', {}).get('tcp'):
raise ConfigError('synproxy TCP MSS is not defined')
if rule_conf.get('protocol', {}) != 'tcp':
raise ConfigError('For action "synproxy" the protocol must be set to TCP')
if 'queue_options' in rule_conf:
if 'queue' not in rule_conf['action']:
raise ConfigError('queue-options defined, but action queue needed and it is not defined')
if 'fanout' in rule_conf['queue_options'] and ('queue' not in rule_conf or '-' not in rule_conf['queue']):
raise ConfigError('queue-options fanout defined, then queue needs to be defined as a range')
if 'queue' in rule_conf and 'queue' not in rule_conf['action']:
raise ConfigError('queue defined, but action queue needed and it is not defined')
if 'fragment' in rule_conf:
if {'match_frag', 'match_non_frag'} <= set(rule_conf['fragment']):
raise ConfigError('Cannot specify both "match-frag" and "match-non-frag"')
if 'limit' in rule_conf:
if 'rate' in rule_conf['limit']:
rate_int = re.sub(r'\D', '', rule_conf['limit']['rate'])
if int(rate_int) < 1:
raise ConfigError('Limit rate integer cannot be less than 1')
if 'ipsec' in rule_conf:
if {'match_ipsec', 'match_non_ipsec'} <= set(rule_conf['ipsec']):
raise ConfigError('Cannot specify both "match-ipsec" and "match-non-ipsec"')
if 'recent' in rule_conf:
if not {'count', 'time'} <= set(rule_conf['recent']):
raise ConfigError('Recent "count" and "time" values must be defined')
tcp_flags = dict_search_args(rule_conf, 'tcp', 'flags')
if tcp_flags:
if dict_search_args(rule_conf, 'protocol') != 'tcp':
raise ConfigError('Protocol must be tcp when specifying tcp flags')
not_flags = dict_search_args(rule_conf, 'tcp', 'flags', 'not')
if not_flags:
duplicates = [flag for flag in tcp_flags if flag in not_flags]
if duplicates:
raise ConfigError(f'Cannot match a tcp flag as set and not set')
if 'protocol' in rule_conf:
if rule_conf['protocol'] == 'icmp' and ipv6:
raise ConfigError(f'Cannot match IPv4 ICMP protocol on IPv6, use ipv6-icmp')
if rule_conf['protocol'] == 'ipv6-icmp' and not ipv6:
raise ConfigError(f'Cannot match IPv6 ICMP protocol on IPv4, use icmp')
for side in ['destination', 'source']:
if side in rule_conf:
side_conf = rule_conf[side]
if len({'address', 'fqdn', 'geoip'} & set(side_conf)) > 1:
raise ConfigError('Only one of address, fqdn or geoip can be specified')
if 'group' in side_conf:
if len({'address_group', 'network_group', 'domain_group'} & set(side_conf['group'])) > 1:
raise ConfigError('Only one address-group, network-group or domain-group can be specified')
for group in valid_groups:
if group in side_conf['group']:
group_name = side_conf['group'][group]
fw_group = f'ipv6_{group}' if ipv6 and group in ['address_group', 'network_group'] else group
error_group = fw_group.replace("_", "-")
if group in ['address_group', 'network_group', 'domain_group']:
types = [t for t in ['address', 'fqdn', 'geoip'] if t in side_conf]
if types:
raise ConfigError(f'{error_group} and {types[0]} cannot both be defined')
if group_name and group_name[0] == '!':
group_name = group_name[1:]
group_obj = dict_search_args(firewall, 'group', fw_group, group_name)
if group_obj is None:
raise ConfigError(f'Invalid {error_group} "{group_name}" on firewall rule')
if not group_obj:
Warning(f'{error_group} "{group_name}" has no members!')
if 'port' in side_conf or dict_search_args(side_conf, 'group', 'port_group'):
if 'protocol' not in rule_conf:
raise ConfigError('Protocol must be defined if specifying a port or port-group')
if rule_conf['protocol'] not in ['tcp', 'udp', 'tcp_udp']:
raise ConfigError('Protocol must be tcp, udp, or tcp_udp when specifying a port or port-group')
if 'port' in side_conf and dict_search_args(side_conf, 'group', 'port_group'):
raise ConfigError(f'{side} port-group and port cannot both be defined')
if 'log_options' in rule_conf:
if 'log' not in rule_conf:
raise ConfigError('log-options defined, but log is not enable')
if 'snapshot_length' in rule_conf['log_options'] and 'group' not in rule_conf['log_options']:
raise ConfigError('log-options snapshot-length defined, but log group is not define')
if 'queue_threshold' in rule_conf['log_options'] and 'group' not in rule_conf['log_options']:
raise ConfigError('log-options queue-threshold defined, but log group is not define')
for direction in ['inbound_interface','outbound_interface']:
if direction in rule_conf:
if 'name' in rule_conf[direction] and 'group' in rule_conf[direction]:
raise ConfigError(f'Cannot specify both interface group and interface name for {direction}')
def verify_nested_group(group_name, group, groups, seen):
if 'include' not in group:
return
seen.append(group_name)
for g in group['include']:
if g not in groups:
raise ConfigError(f'Nested group "{g}" does not exist')
if g in seen:
raise ConfigError(f'Group "{group_name}" has a circular reference')
if 'include' in groups[g]:
verify_nested_group(g, groups[g], groups, seen)
def verify_hardware_offload(ifname):
ethtool = Ethtool(ifname)
enabled, fixed = ethtool.get_hw_tc_offload()
if not enabled and fixed:
raise ConfigError(f'Interface "{ifname}" does not support hardware offload')
if not enabled:
raise ConfigError(f'Interface "{ifname}" requires "offload hw-tc-offload"')
def verify(firewall):
if 'flowtable' in firewall:
for flowtable, flowtable_conf in firewall['flowtable'].items():
if 'interface' not in flowtable_conf:
raise ConfigError(f'Flowtable "{flowtable}" requires at least one interface')
for ifname in flowtable_conf['interface']:
verify_interface_exists(ifname)
if dict_search_args(flowtable_conf, 'offload') == 'hardware':
interfaces = flowtable_conf['interface']
for ifname in interfaces:
verify_hardware_offload(ifname)
if 'group' in firewall:
for group_type in nested_group_types:
if group_type in firewall['group']:
groups = firewall['group'][group_type]
for group_name, group in groups.items():
verify_nested_group(group_name, group, groups, [])
if 'ipv4' in firewall:
for name in ['name','forward','input','output']:
if name in firewall['ipv4']:
for name_id, name_conf in firewall['ipv4'][name].items():
if 'jump' in name_conf['default_action'] and 'default_jump_target' not in name_conf:
raise ConfigError('default-action set to jump, but no default-jump-target specified')
if 'default_jump_target' in name_conf:
target = name_conf['default_jump_target']
if 'jump' not in name_conf['default_action']:
raise ConfigError('default-jump-target defined, but default-action jump needed and it is not defined')
if name_conf['default_jump_target'] == name_id:
raise ConfigError(f'Loop detected on default-jump-target.')
## Now need to check that default-jump-target exists (other firewall chain/name)
if target not in dict_search_args(firewall['ipv4'], 'name'):
raise ConfigError(f'Invalid jump-target. Firewall name {target} does not exist on the system')
if 'rule' in name_conf:
for rule_id, rule_conf in name_conf['rule'].items():
verify_rule(firewall, rule_conf, False)
if 'ipv6' in firewall:
for name in ['name','forward','input','output']:
if name in firewall['ipv6']:
for name_id, name_conf in firewall['ipv6'][name].items():
if 'jump' in name_conf['default_action'] and 'default_jump_target' not in name_conf:
raise ConfigError('default-action set to jump, but no default-jump-target specified')
if 'default_jump_target' in name_conf:
target = name_conf['default_jump_target']
if 'jump' not in name_conf['default_action']:
raise ConfigError('default-jump-target defined, but default-action jump needed and it is not defined')
if name_conf['default_jump_target'] == name_id:
raise ConfigError(f'Loop detected on default-jump-target.')
## Now need to check that default-jump-target exists (other firewall chain/name)
if target not in dict_search_args(firewall['ipv6'], 'name'):
raise ConfigError(f'Invalid jump-target. Firewall name {target} does not exist on the system')
if 'rule' in name_conf:
for rule_id, rule_conf in name_conf['rule'].items():
verify_rule(firewall, rule_conf, True)
#### ZONESSSS
local_zone = False
zone_interfaces = []
if 'zone' in firewall:
for zone, zone_conf in firewall['zone'].items():
if 'local_zone' not in zone_conf and 'interface' not in zone_conf:
raise ConfigError(f'Zone "{zone}" has no interfaces and is not the local zone')
if 'local_zone' in zone_conf:
if local_zone:
raise ConfigError('There cannot be multiple local zones')
if 'interface' in zone_conf:
raise ConfigError('Local zone cannot have interfaces assigned')
if 'intra_zone_filtering' in zone_conf:
raise ConfigError('Local zone cannot use intra-zone-filtering')
local_zone = True
if 'interface' in zone_conf:
found_duplicates = [intf for intf in zone_conf['interface'] if intf in zone_interfaces]
if found_duplicates:
raise ConfigError(f'Interfaces cannot be assigned to multiple zones')
zone_interfaces += zone_conf['interface']
if 'intra_zone_filtering' in zone_conf:
intra_zone = zone_conf['intra_zone_filtering']
if len(intra_zone) > 1:
raise ConfigError('Only one intra-zone-filtering action must be specified')
if 'firewall' in intra_zone:
v4_name = dict_search_args(intra_zone, 'firewall', 'name')
if v4_name and not dict_search_args(firewall, 'ipv4', 'name', v4_name):
raise ConfigError(f'Firewall name "{v4_name}" does not exist')
v6_name = dict_search_args(intra_zone, 'firewall', 'ipv6_name')
if v6_name and not dict_search_args(firewall, 'ipv6', 'name', v6_name):
raise ConfigError(f'Firewall ipv6-name "{v6_name}" does not exist')
if not v4_name and not v6_name:
raise ConfigError('No firewall names specified for intra-zone-filtering')
if 'from' in zone_conf:
for from_zone, from_conf in zone_conf['from'].items():
if from_zone not in firewall['zone']:
raise ConfigError(f'Zone "{zone}" refers to a non-existent or deleted zone "{from_zone}"')
v4_name = dict_search_args(from_conf, 'firewall', 'name')
if v4_name and not dict_search_args(firewall, 'ipv4', 'name', v4_name):
raise ConfigError(f'Firewall name "{v4_name}" does not exist')
v6_name = dict_search_args(from_conf, 'firewall', 'ipv6_name')
if v6_name and not dict_search_args(firewall, 'ipv6', 'name', v6_name):
raise ConfigError(f'Firewall ipv6-name "{v6_name}" does not exist')
return None
def generate(firewall):
if not os.path.exists(nftables_conf):
firewall['first_install'] = True
if 'zone' in firewall:
for local_zone, local_zone_conf in firewall['zone'].items():
if 'local_zone' not in local_zone_conf:
continue
local_zone_conf['from_local'] = {}
for zone, zone_conf in firewall['zone'].items():
if zone == local_zone or 'from' not in zone_conf:
continue
if local_zone in zone_conf['from']:
local_zone_conf['from_local'][zone] = zone_conf['from'][local_zone]
render(nftables_conf, 'firewall/nftables.j2', firewall)
return None
def apply_sysfs(firewall):
for name, conf in sysfs_config.items():
paths = glob(conf['sysfs'])
value = None
if name in firewall['global_options']:
conf_value = firewall['global_options'][name]
if conf_value in conf:
value = conf[conf_value]
elif conf_value == 'enable':
value = '1'
elif conf_value == 'disable':
value = '0'
if value:
for path in paths:
with open(path, 'w') as f:
f.write(value)
def apply(firewall):
install_result, output = rc_cmd(f'nft -f {nftables_conf}')
if install_result == 1:
raise ConfigError(f'Failed to apply firewall: {output}')
apply_sysfs(firewall)
call_dependents()
# T970 Enable a resolver (systemd daemon) that checks
# domain-group/fqdn addresses and update entries for domains by timeout
# If router loaded without internet connection or for synchronization
domain_action = 'stop'
if dict_search_args(firewall, 'group', 'domain_group') or firewall['ip_fqdn'] or firewall['ip6_fqdn']:
domain_action = 'restart'
call(f'systemctl {domain_action} vyos-domain-resolver.service')
if firewall['geoip_updated']:
# Call helper script to Update set contents
if 'name' in firewall['geoip_updated'] or 'ipv6_name' in firewall['geoip_updated']:
print('Updating GeoIP. Please wait...')
geoip_update(firewall)
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/interfaces-bridge.py b/src/conf_mode/interfaces-bridge.py
index 31508a3c5..29991e2da 100755
--- a/src/conf_mode/interfaces-bridge.py
+++ b/src/conf_mode/interfaces-bridge.py
@@ -1,186 +1,186 @@
#!/usr/bin/env python3
#
# Copyright (C) 2019-2023 VyOS maintainers and contributors
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License version 2 or later as
# published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
from sys import exit
from vyos.config import Config
from vyos.configdict import get_interface_dict
from vyos.configdict import node_changed
from vyos.configdict import is_member
from vyos.configdict import is_source_interface
from vyos.configdict import has_vlan_subinterface_configured
from vyos.configverify import verify_dhcpv6
from vyos.configverify import verify_mirror_redirect
from vyos.configverify import verify_vrf
from vyos.ifconfig import BridgeIf
from vyos.configdict import has_address_configured
from vyos.configdict import has_vrf_configured
from vyos.configdep import set_dependents
from vyos.configdep import call_dependents
from vyos.utils.dict import dict_search
from vyos import ConfigError
from vyos import airbag
airbag.enable()
def get_config(config=None):
"""
Retrive CLI config as dictionary. Dictionary can never be empty, as at least the
interface name will be added or a deleted flag
"""
if config:
conf = config
else:
conf = Config()
base = ['interfaces', 'bridge']
ifname, bridge = get_interface_dict(conf, base)
# determine which members have been removed
- tmp = node_changed(conf, base + [ifname, 'member', 'interface'], key_mangling=('-', '_'))
+ tmp = node_changed(conf, base + [ifname, 'member', 'interface'])
if tmp:
if 'member' in bridge:
bridge['member'].update({'interface_remove' : tmp })
else:
bridge.update({'member' : {'interface_remove' : tmp }})
if dict_search('member.interface', bridge) is not None:
for interface in list(bridge['member']['interface']):
# Check if member interface is already member of another bridge
tmp = is_member(conf, interface, 'bridge')
if tmp and bridge['ifname'] not in tmp:
bridge['member']['interface'][interface].update({'is_bridge_member' : tmp})
# Check if member interface is already member of a bond
tmp = is_member(conf, interface, 'bonding')
if tmp: bridge['member']['interface'][interface].update({'is_bond_member' : tmp})
# Check if member interface is used as source-interface on another interface
tmp = is_source_interface(conf, interface)
if tmp: bridge['member']['interface'][interface].update({'is_source_interface' : tmp})
# Bridge members must not have an assigned address
tmp = has_address_configured(conf, interface)
if tmp: bridge['member']['interface'][interface].update({'has_address' : ''})
# Bridge members must not have a VRF attached
tmp = has_vrf_configured(conf, interface)
if tmp: bridge['member']['interface'][interface].update({'has_vrf' : ''})
# VLAN-aware bridge members must not have VLAN interface configuration
tmp = has_vlan_subinterface_configured(conf,interface)
if 'enable_vlan' in bridge and tmp:
bridge['member']['interface'][interface].update({'has_vlan' : ''})
# When using VXLAN member interfaces that are configured for Single
# VXLAN Device (SVD) we need to call the VXLAN conf-mode script to re-create
# VLAN to VNI mappings if required
if interface.startswith('vxlan'):
set_dependents('vxlan', conf, interface)
# delete empty dictionary keys - no need to run code paths if nothing is there to do
if 'member' in bridge:
if 'interface' in bridge['member'] and len(bridge['member']['interface']) == 0:
del bridge['member']['interface']
if len(bridge['member']) == 0:
del bridge['member']
return bridge
def verify(bridge):
if 'deleted' in bridge:
return None
verify_dhcpv6(bridge)
verify_vrf(bridge)
verify_mirror_redirect(bridge)
ifname = bridge['ifname']
if dict_search('member.interface', bridge):
for interface, interface_config in bridge['member']['interface'].items():
error_msg = f'Can not add interface "{interface}" to bridge, '
if interface == 'lo':
raise ConfigError('Loopback interface "lo" can not be added to a bridge')
if 'is_bridge_member' in interface_config:
tmp = next(iter(interface_config['is_bridge_member']))
raise ConfigError(error_msg + f'it is already a member of bridge "{tmp}"!')
if 'is_bond_member' in interface_config:
tmp = next(iter(interface_config['is_bond_member']))
raise ConfigError(error_msg + f'it is already a member of bond "{tmp}"!')
if 'is_source_interface' in interface_config:
tmp = interface_config['is_source_interface']
raise ConfigError(error_msg + f'it is the source-interface of "{tmp}"!')
if 'has_address' in interface_config:
raise ConfigError(error_msg + 'it has an address assigned!')
if 'has_vrf' in interface_config:
raise ConfigError(error_msg + 'it has a VRF assigned!')
if 'enable_vlan' in bridge:
if 'has_vlan' in interface_config:
raise ConfigError(error_msg + 'it has VLAN subinterface(s) assigned!')
if 'wlan' in interface:
raise ConfigError(error_msg + 'VLAN aware cannot be set!')
else:
for option in ['allowed_vlan', 'native_vlan']:
if option in interface_config:
raise ConfigError('Can not use VLAN options on non VLAN aware bridge')
if 'enable_vlan' in bridge:
if dict_search('vif.1', bridge):
raise ConfigError(f'VLAN 1 sub interface cannot be set for VLAN aware bridge {ifname}, and VLAN 1 is always the parent interface')
else:
if dict_search('vif', bridge):
raise ConfigError(f'You must first activate "enable-vlan" of {ifname} bridge to use "vif"')
return None
def generate(bridge):
return None
def apply(bridge):
br = BridgeIf(bridge['ifname'])
if 'deleted' in bridge:
# delete interface
br.remove()
else:
br.update(bridge)
for interface in dict_search('member.interface', bridge) or []:
if interface.startswith('vxlan'):
try:
call_dependents()
except ConfigError:
raise ConfigError('Error in updating VXLAN interface after changing bridge!')
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/netns.py b/src/conf_mode/netns.py
index 95ab83dbc..7cee33bc6 100755
--- a/src/conf_mode/netns.py
+++ b/src/conf_mode/netns.py
@@ -1,119 +1,118 @@
#!/usr/bin/env python3
#
# Copyright (C) 2021 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 tempfile import NamedTemporaryFile
from vyos.config import Config
from vyos.configdict import node_changed
from vyos.ifconfig import Interface
from vyos.utils.process import call
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()
def netns_interfaces(c, match):
"""
get NETNS bound interfaces
"""
matched = []
old_level = c.get_level()
c.set_level(['interfaces'])
section = c.get_config_dict([], get_first_key=True)
for type in section:
interfaces = section[type]
for name in interfaces:
interface = interfaces[name]
if 'netns' in interface:
v = interface.get('netns', '')
if v == match:
matched.append(name)
c.set_level(old_level)
return matched
def get_config(config=None):
if config:
conf = config
else:
conf = Config()
base = ['netns']
netns = conf.get_config_dict(base, get_first_key=True,
no_tag_node_value_mangle=True)
# determine which NETNS has been removed
for name in node_changed(conf, base + ['name']):
if 'netns_remove' not in netns:
netns.update({'netns_remove' : {}})
netns['netns_remove'][name] = {}
# get NETNS bound interfaces
interfaces = netns_interfaces(conf, name)
if interfaces: netns['netns_remove'][name]['interface'] = interfaces
return netns
def verify(netns):
# ensure NETNS is not assigned to any interface
if 'netns_remove' in netns:
for name, config in netns['netns_remove'].items():
if 'interface' in config:
- raise ConfigError(f'Can not remove NETNS "{name}", it still has '\
- f'member interfaces!')
+ raise ConfigError(f'Can not remove network namespace "{name}", it '\
+ f'still has member interfaces!')
if 'name' in netns:
for name, config in netns['name'].items():
# no tests (yet)
pass
return None
-
def generate(netns):
if not netns:
return None
return None
def apply(netns):
for tmp in (dict_search('netns_remove', netns) or []):
if os.path.isfile(f'/run/netns/{tmp}'):
call(f'ip netns del {tmp}')
if 'name' in netns:
for name, config in netns['name'].items():
if not os.path.isfile(f'/run/netns/{name}'):
call(f'ip netns add {name}')
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_nhrp.py b/src/conf_mode/protocols_nhrp.py
index 5ec0bc9e5..c339c6391 100755
--- a/src/conf_mode/protocols_nhrp.py
+++ b/src/conf_mode/protocols_nhrp.py
@@ -1,115 +1,115 @@
#!/usr/bin/env python3
#
# Copyright (C) 2021-2023 VyOS maintainers and contributors
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License version 2 or later as
# published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import os
from vyos.config import Config
from vyos.configdict import node_changed
from vyos.template import render
from vyos.utils.process import process_named_running
from vyos.utils.process import run
from vyos import ConfigError
from vyos import airbag
airbag.enable()
opennhrp_conf = '/run/opennhrp/opennhrp.conf'
nhrp_nftables_conf = '/run/nftables_nhrp.conf'
def get_config(config=None):
if config:
conf = config
else:
conf = Config()
base = ['protocols', 'nhrp']
nhrp = conf.get_config_dict(base, key_mangling=('-', '_'),
get_first_key=True, no_tag_node_value_mangle=True)
- nhrp['del_tunnels'] = node_changed(conf, base + ['tunnel'], key_mangling=('-', '_'))
+ nhrp['del_tunnels'] = node_changed(conf, base + ['tunnel'])
if not conf.exists(base):
return nhrp
nhrp['if_tunnel'] = conf.get_config_dict(['interfaces', 'tunnel'], key_mangling=('-', '_'),
get_first_key=True, no_tag_node_value_mangle=True)
nhrp['profile_map'] = {}
profile = conf.get_config_dict(['vpn', 'ipsec', 'profile'], key_mangling=('-', '_'),
get_first_key=True, no_tag_node_value_mangle=True)
for name, profile_conf in profile.items():
if 'bind' in profile_conf and 'tunnel' in profile_conf['bind']:
interfaces = profile_conf['bind']['tunnel']
if isinstance(interfaces, str):
interfaces = [interfaces]
for interface in interfaces:
nhrp['profile_map'][interface] = name
return nhrp
def verify(nhrp):
if 'tunnel' in nhrp:
for name, nhrp_conf in nhrp['tunnel'].items():
if not nhrp['if_tunnel'] or name not in nhrp['if_tunnel']:
raise ConfigError(f'Tunnel interface "{name}" does not exist')
tunnel_conf = nhrp['if_tunnel'][name]
if 'encapsulation' not in tunnel_conf or tunnel_conf['encapsulation'] != 'gre':
raise ConfigError(f'Tunnel "{name}" is not an mGRE tunnel')
if 'remote' in tunnel_conf:
raise ConfigError(f'Tunnel "{name}" cannot have a remote address defined')
if 'map' in nhrp_conf:
for map_name, map_conf in nhrp_conf['map'].items():
if 'nbma_address' not in map_conf:
raise ConfigError(f'nbma-address missing on map {map_name} on tunnel {name}')
if 'dynamic_map' in nhrp_conf:
for map_name, map_conf in nhrp_conf['dynamic_map'].items():
if 'nbma_domain_name' not in map_conf:
raise ConfigError(f'nbma-domain-name missing on dynamic-map {map_name} on tunnel {name}')
return None
def generate(nhrp):
if not os.path.exists(nhrp_nftables_conf):
nhrp['first_install'] = True
render(opennhrp_conf, 'nhrp/opennhrp.conf.j2', nhrp)
render(nhrp_nftables_conf, 'nhrp/nftables.conf.j2', nhrp)
return None
def apply(nhrp):
nft_rc = run(f'nft -f {nhrp_nftables_conf}')
if nft_rc != 0:
raise ConfigError('Failed to apply NHRP tunnel firewall rules')
action = 'restart' if nhrp and 'tunnel' in nhrp else 'stop'
service_rc = run(f'systemctl {action} opennhrp.service')
if service_rc != 0:
raise ConfigError(f'Failed to {action} the NHRP service')
return None
if __name__ == '__main__':
try:
c = get_config()
verify(c)
generate(c)
apply(c)
except ConfigError as e:
print(e)
exit(1)