diff --git a/python/vyos/config.py b/python/vyos/config.py index 0ca41718f..ca7b035e5 100644 --- a/python/vyos/config.py +++ b/python/vyos/config.py @@ -1,573 +1,583 @@ # Copyright 2017, 2019-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/>. """ A library for reading VyOS running config data. This library is used internally by all config scripts of VyOS, but its API should be considered stable and safe to use in user scripts. Note that this module will not work outside VyOS. Node taxonomy ############# There are multiple types of config tree nodes in VyOS, each requires its own set of operations. *Leaf nodes* (such as "address" in interfaces) can have values, but cannot -have children. +have children. Leaf nodes can have one value, multiple values, or no values at all. For example, "system host-name" is a single-value leaf node, "system name-server" is a multi-value leaf node (commonly abbreviated "multi node"), and "system ip disable-forwarding" is a valueless leaf node. Non-leaf nodes cannot have values, but they can have child nodes. They are divided into two classes depending on whether the names of their children are fixed or not. For example, under "system", the names of all valid child nodes are predefined ("login", "name-server" etc.). To the contrary, children of the "system task-scheduler task" node can have arbitrary names. Such nodes are called *tag nodes*. This terminology is confusing but we keep using it for lack of a better word. No one remembers if the "tag" in "task Foo" is "task" or "Foo", but the distinction is irrelevant in practice. Configuration modes ################### VyOS has two distinct modes: operational mode and configuration mode. When a user logins, the CLI is in the operational mode. In this mode, only the running (effective) config is accessible for reading. When a user enters the "configure" command, a configuration session is setup. Every config session has its *proposed* (or *session*) config built on top of the current running config. When changes are commited, if commit succeeds, the proposed config is merged into the running config. In configuration mode, "base" functions like `exists`, `return_value` return values from the session config, while functions prefixed "effective" return values from the running config. In operational mode, all functions return values from the running config. """ import re import json from copy import deepcopy from typing import Union import vyos.configtree from vyos.xml_ref import multi_to_list from vyos.xml_ref import from_source from vyos.xml_ref import ext_dict_merge from vyos.xml_ref import relative_defaults from vyos.utils.dict import get_sub_dict from vyos.utils.dict import mangle_dict_keys from vyos.configsource import ConfigSource from vyos.configsource import ConfigSourceSession class ConfigDict(dict): _from_defaults = {} _dict_kwargs = {} def from_defaults(self, path: list[str]) -> bool: return from_source(self._from_defaults, path) @property def kwargs(self) -> dict: return self._dict_kwargs def config_dict_merge(src: dict, dest: Union[dict, ConfigDict]) -> ConfigDict: if not isinstance(dest, ConfigDict): dest = ConfigDict(dest) return ext_dict_merge(src, dest) class Config(object): """ The class of config access objects. Internally, in the current implementation, this object is *almost* stateless, the only state it keeps is relative *config path* for convenient access to config subtrees. """ def __init__(self, session_env=None, config_source=None): if config_source is None: self._config_source = ConfigSourceSession(session_env) else: if not isinstance(config_source, ConfigSource): raise TypeError("config_source not of type ConfigSource") self._config_source = config_source self._level = [] self._dict_cache = {} (self._running_config, self._session_config) = self._config_source.get_configtree_tuple() def get_config_tree(self, effective=False): if effective: return self._running_config return self._session_config def _make_path(self, path): # Backwards-compatibility stuff: original implementation used string paths # libvyosconfig paths are lists, but since node names cannot contain whitespace, # splitting at whitespace is reasonably safe. # It may cause problems with exists() when it's used for checking values, # since values may contain whitespace. if isinstance(path, str): path = re.split(r'\s+', path) elif isinstance(path, list): pass else: raise TypeError("Path must be a whitespace-separated string or a list") return (self._level + path) def set_level(self, path): """ Set the *edit level*, that is, a relative config tree path. Once set, all operations will be relative to this path, for example, after ``set_level("system")``, calling ``exists("name-server")`` is equivalent to calling ``exists("system name-server"`` without ``set_level``. Args: path (str|list): relative config path """ # Make sure there's always a space between default path (level) # and path supplied as method argument # XXX: for small strings in-place concatenation is not a problem if isinstance(path, str): if path: self._level = re.split(r'\s+', path) else: self._level = [] elif isinstance(path, list): self._level = path.copy() else: raise TypeError("Level path must be either a whitespace-separated string or a list") def get_level(self): """ Gets the current edit level. Returns: str: current edit level """ return(self._level.copy()) def exists(self, path): """ Checks if a node or value with given path exists in the proposed config. Args: path (str): Configuration tree path Returns: True if node or value exists in the proposed config, False otherwise Note: This function should not be used outside of configuration sessions. In operational mode scripts, use ``exists_effective``. """ if self._session_config is None: return False # Assume the path is a node path first if self._session_config.exists(self._make_path(path)): return True else: # If that check fails, it may mean the path has a value at the end. # libvyosconfig exists() works only for _nodes_, not _values_ # libvyattacfg also worked for values, so we emulate that case here if isinstance(path, str): path = re.split(r'\s+', path) path_without_value = path[:-1] try: # return_values() is safe to use with single-value nodes, # it simply returns a single-item list in that case. values = self._session_config.return_values(self._make_path(path_without_value)) # If we got this far, the node does exist and has values, # so we need to check if it has the value in question among its values. return (path[-1] in values) except vyos.configtree.ConfigTreeError: # Even the parent node doesn't exist at all return False def session_changed(self): """ Returns: True if the config session has uncommited changes, False otherwise. """ return self._config_source.session_changed() def in_session(self): """ Returns: True if called from a configuration session, False otherwise. """ return self._config_source.in_session() def show_config(self, path=[], default=None, effective=False): """ Args: path (str list): Configuration tree path, or empty default (str): Default value to return Returns: str: working configuration """ return self._config_source.show_config(path, default, effective) def get_cached_root_dict(self, effective=False): cached = self._dict_cache.get(effective, {}) if cached: return cached if effective: config = self._running_config else: config = self._session_config if config: config_dict = json.loads(config.to_json()) else: config_dict = {} self._dict_cache[effective] = config_dict return config_dict def verify_mangling(self, key_mangling): if not (isinstance(key_mangling, tuple) and \ (len(key_mangling) == 2) and \ isinstance(key_mangling[0], str) and \ isinstance(key_mangling[1], str)): raise ValueError("key_mangling must be a tuple of two strings") def get_config_dict(self, path=[], effective=False, key_mangling=None, get_first_key=False, no_multi_convert=False, no_tag_node_value_mangle=False, - with_defaults=False, with_recursive_defaults=False): + with_defaults=False, + with_recursive_defaults=False, + with_pki=False): """ Args: path (str list): Configuration tree path, can be empty effective=False: effective or session config key_mangling=None: mangle dict keys according to regex and replacement get_first_key=False: if k = path[:-1], return sub-dict d[k] instead of {k: d[k]} no_multi_convert=False: if convert, return single value of multi node as list Returns: a dict representation of the config under path """ kwargs = locals().copy() del kwargs['self'] del kwargs['no_multi_convert'] del kwargs['with_defaults'] del kwargs['with_recursive_defaults'] + del kwargs['with_pki'] lpath = self._make_path(path) root_dict = self.get_cached_root_dict(effective) conf_dict = get_sub_dict(root_dict, lpath, get_first_key=get_first_key) rpath = lpath if get_first_key else lpath[:-1] if not no_multi_convert: conf_dict = multi_to_list(rpath, conf_dict) if key_mangling is not None: self.verify_mangling(key_mangling) conf_dict = mangle_dict_keys(conf_dict, key_mangling[0], key_mangling[1], abs_path=rpath, no_tag_node_value_mangle=no_tag_node_value_mangle) if with_defaults or with_recursive_defaults: defaults = self.get_config_defaults(**kwargs, recursive=with_recursive_defaults) conf_dict = config_dict_merge(defaults, conf_dict) else: conf_dict = ConfigDict(conf_dict) + if with_pki and conf_dict: + pki_dict = self.get_config_dict(['pki'], key_mangling=('-', '_'), + no_tag_node_value_mangle=True, + get_first_key=True) + if pki_dict: + conf_dict['pki'] = pki_dict + # save optional args for a call to get_config_defaults setattr(conf_dict, '_dict_kwargs', kwargs) return conf_dict def get_config_defaults(self, path=[], effective=False, key_mangling=None, no_tag_node_value_mangle=False, get_first_key=False, recursive=False) -> dict: lpath = self._make_path(path) root_dict = self.get_cached_root_dict(effective) conf_dict = get_sub_dict(root_dict, lpath, get_first_key) defaults = relative_defaults(lpath, conf_dict, get_first_key=get_first_key, recursive=recursive) rpath = lpath if get_first_key else lpath[:-1] if key_mangling is not None: self.verify_mangling(key_mangling) defaults = mangle_dict_keys(defaults, key_mangling[0], key_mangling[1], abs_path=rpath, no_tag_node_value_mangle=no_tag_node_value_mangle) return defaults def merge_defaults(self, config_dict: ConfigDict, recursive=False): if not isinstance(config_dict, ConfigDict): raise TypeError('argument is not of type ConfigDict') if not config_dict.kwargs: raise ValueError('argument missing metadata') args = config_dict.kwargs d = self.get_config_defaults(**args, recursive=recursive) config_dict = config_dict_merge(d, config_dict) return config_dict def is_multi(self, path): """ Args: path (str): Configuration tree path Returns: True if a node can have multiple values, False otherwise. Note: It also returns False if node doesn't exist. """ self._config_source.set_level(self.get_level) return self._config_source.is_multi(path) def is_tag(self, path): """ Args: path (str): Configuration tree path Returns: True if a node is a tag node, False otherwise. Note: It also returns False if node doesn't exist. """ self._config_source.set_level(self.get_level) return self._config_source.is_tag(path) def is_leaf(self, path): """ Args: path (str): Configuration tree path Returns: True if a node is a leaf node, False otherwise. Note: It also returns False if node doesn't exist. """ self._config_source.set_level(self.get_level) return self._config_source.is_leaf(path) def return_value(self, path, default=None): """ Retrieve a value of single-value leaf node in the running or proposed config Args: path (str): Configuration tree path default (str): Default value to return if node does not exist Returns: str: Node value, if it has any None: if node is valueless *or* if it doesn't exist Note: Due to the issue with treatment of valueless nodes by this function, valueless nodes should be checked with ``exists`` instead. This function cannot be used outside a configuration session. In operational mode scripts, use ``return_effective_value``. """ if self._session_config: try: value = self._session_config.return_value(self._make_path(path)) except vyos.configtree.ConfigTreeError: value = None else: value = None if not value: return(default) else: return(value) def return_values(self, path, default=[]): """ Retrieve all values of a multi-value leaf node in the running or proposed config Args: path (str): Configuration tree path Returns: str list: Node values, if it has any []: if node does not exist Note: This function cannot be used outside a configuration session. In operational mode scripts, use ``return_effective_values``. """ if self._session_config: try: values = self._session_config.return_values(self._make_path(path)) except vyos.configtree.ConfigTreeError: values = [] else: values = [] if not values: return(default.copy()) else: return(values) def list_nodes(self, path, default=[]): """ Retrieve names of all children of a tag node in the running or proposed config Args: path (str): Configuration tree path Returns: string list: child node names """ if self._session_config: try: nodes = self._session_config.list_nodes(self._make_path(path)) except vyos.configtree.ConfigTreeError: nodes = [] else: nodes = [] if not nodes: return(default.copy()) else: return(nodes) def exists_effective(self, path): """ Checks if a node or value exists in the running (effective) config. Args: path (str): Configuration tree path Returns: True if node exists in the running config, False otherwise Note: This function is safe to use in operational mode. In configuration mode, it ignores uncommited changes. """ if self._running_config is None: return False # Assume the path is a node path first if self._running_config.exists(self._make_path(path)): return True else: # If that check fails, it may mean the path has a value at the end. # libvyosconfig exists() works only for _nodes_, not _values_ # libvyattacfg also worked for values, so we emulate that case here if isinstance(path, str): path = re.split(r'\s+', path) path_without_value = path[:-1] try: # return_values() is safe to use with single-value nodes, # it simply returns a single-item list in that case. values = self._running_config.return_values(self._make_path(path_without_value)) # If we got this far, the node does exist and has values, # so we need to check if it has the value in question among its values. return (path[-1] in values) except vyos.configtree.ConfigTreeError: # Even the parent node doesn't exist at all return False def return_effective_value(self, path, default=None): """ Retrieve a values of a single-value leaf node in a running (effective) config Args: path (str): Configuration tree path default (str): Default value to return if node does not exist Returns: str: Node value """ if self._running_config: try: value = self._running_config.return_value(self._make_path(path)) except vyos.configtree.ConfigTreeError: value = None else: value = None if not value: return(default) else: return(value) def return_effective_values(self, path, default=[]): """ Retrieve all values of a multi-value node in a running (effective) config Args: path (str): Configuration tree path Returns: str list: A list of values """ if self._running_config: try: values = self._running_config.return_values(self._make_path(path)) except vyos.configtree.ConfigTreeError: values = [] else: values = [] if not values: return(default.copy()) else: return(values) def list_effective_nodes(self, path, default=[]): """ Retrieve names of all children of a tag node in the running config Args: path (str): Configuration tree path Returns: str list: child node names """ if self._running_config: try: nodes = self._running_config.list_nodes(self._make_path(path)) except vyos.configtree.ConfigTreeError: nodes = [] else: nodes = [] if not nodes: return(default.copy()) else: return(nodes) diff --git a/python/vyos/configdict.py b/python/vyos/configdict.py index 089d9d3d5..4111d7271 100644 --- a/python/vyos/configdict.py +++ b/python/vyos/configdict.py @@ -1,666 +1,668 @@ # 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, 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 """ 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): +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_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): +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.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) + with_recursive_defaults=True, + with_pki=with_pki) # 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/smoketest/scripts/cli/test_service_https.py b/smoketest/scripts/cli/test_service_https.py index 703e3e8c4..280932fd7 100755 --- a/smoketest/scripts/cli/test_service_https.py +++ b/smoketest/scripts/cli/test_service_https.py @@ -1,449 +1,448 @@ #!/usr/bin/env python3 # # Copyright (C) 2019-2023 VyOS maintainers and contributors # # This program is free software; you can redistribute it and/or modify # it under the terms of the GNU General Public License version 2 or later as # published by the Free Software Foundation. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program. If not, see <http://www.gnu.org/licenses/>. import unittest import json from requests import request from urllib3.exceptions import InsecureRequestWarning from base_vyostest_shim import VyOSUnitTestSHIM from base_vyostest_shim import ignore_warning from vyos.utils.file import read_file from vyos.utils.process import call from vyos.utils.process import process_named_running from vyos.configsession import ConfigSessionError base_path = ['service', 'https'] pki_base = ['pki'] cert_data = """ MIICFDCCAbugAwIBAgIUfMbIsB/ozMXijYgUYG80T1ry+mcwCgYIKoZIzj0EAwIw WTELMAkGA1UEBhMCR0IxEzARBgNVBAgMClNvbWUtU3RhdGUxEjAQBgNVBAcMCVNv bWUtQ2l0eTENMAsGA1UECgwEVnlPUzESMBAGA1UEAwwJVnlPUyBUZXN0MB4XDTIx MDcyMDEyNDUxMloXDTI2MDcxOTEyNDUxMlowWTELMAkGA1UEBhMCR0IxEzARBgNV BAgMClNvbWUtU3RhdGUxEjAQBgNVBAcMCVNvbWUtQ2l0eTENMAsGA1UECgwEVnlP UzESMBAGA1UEAwwJVnlPUyBUZXN0MFkwEwYHKoZIzj0CAQYIKoZIzj0DAQcDQgAE 01HrLcNttqq4/PtoMua8rMWEkOdBu7vP94xzDO7A8C92ls1v86eePy4QllKCzIw3 QxBIoCuH2peGRfWgPRdFsKNhMF8wDwYDVR0TAQH/BAUwAwEB/zAOBgNVHQ8BAf8E BAMCAYYwHQYDVR0lBBYwFAYIKwYBBQUHAwIGCCsGAQUFBwMBMB0GA1UdDgQWBBSu +JnU5ZC4mkuEpqg2+Mk4K79oeDAKBggqhkjOPQQDAgNHADBEAiBEFdzQ/Bc3Lftz ngrY605UhA6UprHhAogKgROv7iR4QgIgEFUxTtW3xXJcnUPWhhUFhyZoqfn8dE93 +dm/LDnp7C0= """ key_data = """ MIGHAgEAMBMGByqGSM49AgEGCCqGSM49AwEHBG0wawIBAQQgPLpD0Ohhoq0g4nhx 2KMIuze7ucKUt/lBEB2wc03IxXyhRANCAATTUestw222qrj8+2gy5rysxYSQ50G7 u8/3jHMM7sDwL3aWzW/zp54/LhCWUoLMjDdDEEigK4fal4ZF9aA9F0Ww """ # to test load config via HTTP URL nginx_conf_smoketest = """ server { listen 8000; server_name localhost; root /tmp; index index.html; location / { try_files $uri $uri/ =404; autoindex on; } } """ PROCESS_NAME = 'nginx' class TestHTTPSService(VyOSUnitTestSHIM.TestCase): @classmethod def setUpClass(cls): super(TestHTTPSService, cls).setUpClass() # ensure we can also run this test on a live system - so lets clean # out the current configuration :) cls.cli_delete(cls, base_path) cls.cli_delete(cls, pki_base) def tearDown(self): self.cli_delete(base_path) self.cli_delete(pki_base) self.cli_commit() # Check for stopped process self.assertFalse(process_named_running(PROCESS_NAME)) def test_server_block(self): vhost_id = 'example' address = '0.0.0.0' port = '8443' name = 'example.org' test_path = base_path + ['virtual-host', vhost_id] self.cli_set(test_path + ['listen-address', address]) self.cli_set(test_path + ['port', port]) self.cli_set(test_path + ['server-name', name]) self.cli_commit() nginx_config = read_file('/etc/nginx/sites-enabled/default') self.assertIn(f'listen {address}:{port} ssl;', nginx_config) self.assertIn(f'ssl_protocols TLSv1.2 TLSv1.3;', nginx_config) self.assertTrue(process_named_running(PROCESS_NAME)) def test_certificate(self): self.cli_set(pki_base + ['certificate', 'test_https', 'certificate', cert_data.replace('\n','')]) self.cli_set(pki_base + ['certificate', 'test_https', 'private', 'key', key_data.replace('\n','')]) self.cli_set(base_path + ['certificates', 'certificate', 'test_https']) self.cli_commit() self.assertTrue(process_named_running(PROCESS_NAME)) def test_api_missing_keys(self): self.cli_set(base_path + ['api']) self.assertRaises(ConfigSessionError, self.cli_commit) def test_api_incomplete_key(self): self.cli_set(base_path + ['api', 'keys', 'id', 'key-01']) self.assertRaises(ConfigSessionError, self.cli_commit) @ignore_warning(InsecureRequestWarning) def test_api_auth(self): vhost_id = 'example' address = '127.0.0.1' port = '443' # default value name = 'localhost' key = 'MySuperSecretVyOS' self.cli_set(base_path + ['api', 'keys', 'id', 'key-01', 'key', key]) test_path = base_path + ['virtual-host', vhost_id] self.cli_set(test_path + ['listen-address', address]) self.cli_set(test_path + ['server-name', name]) self.cli_commit() nginx_config = read_file('/etc/nginx/sites-enabled/default') self.assertIn(f'listen {address}:{port} ssl;', nginx_config) self.assertIn(f'ssl_protocols TLSv1.2 TLSv1.3;', nginx_config) url = f'https://{address}/retrieve' payload = {'data': '{"op": "showConfig", "path": []}', 'key': f'{key}'} headers = {} r = request('POST', url, verify=False, headers=headers, data=payload) # Must get HTTP code 200 on success self.assertEqual(r.status_code, 200) payload_invalid = {'data': '{"op": "showConfig", "path": []}', 'key': 'invalid'} r = request('POST', url, verify=False, headers=headers, data=payload_invalid) # Must get HTTP code 401 on invalid key (Unauthorized) self.assertEqual(r.status_code, 401) payload_no_key = {'data': '{"op": "showConfig", "path": []}'} r = request('POST', url, verify=False, headers=headers, data=payload_no_key) # Must get HTTP code 401 on missing key (Unauthorized) self.assertEqual(r.status_code, 401) # Check path config payload = {'data': '{"op": "showConfig", "path": ["system", "login"]}', 'key': f'{key}'} r = request('POST', url, verify=False, headers=headers, data=payload) response = r.json() vyos_user_exists = 'vyos' in response.get('data', {}).get('user', {}) self.assertTrue(vyos_user_exists, "The 'vyos' user does not exist in the response.") # GraphQL auth test: a missing key will return status code 400, as # 'key' is a non-nullable field in the schema; an incorrect key is # caught by the resolver, and returns success 'False', so one must # check the return value. self.cli_set(base_path + ['api', 'graphql']) self.cli_commit() graphql_url = f'https://{address}/graphql' query_valid_key = f""" {{ SystemStatus (data: {{key: "{key}"}}) {{ success errors data {{ result }} }} }} """ r = request('POST', graphql_url, verify=False, headers=headers, json={'query': query_valid_key}) success = r.json()['data']['SystemStatus']['success'] self.assertTrue(success) query_invalid_key = """ { SystemStatus (data: {key: "invalid"}) { success errors data { result } } } """ r = request('POST', graphql_url, verify=False, headers=headers, json={'query': query_invalid_key}) success = r.json()['data']['SystemStatus']['success'] self.assertFalse(success) query_no_key = """ { SystemStatus (data: {}) { success errors data { result } } } """ r = request('POST', graphql_url, verify=False, headers=headers, json={'query': query_no_key}) success = r.json()['data']['SystemStatus']['success'] self.assertFalse(success) # GraphQL token authentication test: request token; pass in header # of query. self.cli_set(base_path + ['api', 'graphql', 'authentication', 'type', 'token']) self.cli_commit() mutation = """ mutation { AuthToken (data: {username: "vyos", password: "vyos"}) { success errors data { result } } } """ r = request('POST', graphql_url, verify=False, headers=headers, json={'query': mutation}) token = r.json()['data']['AuthToken']['data']['result']['token'] headers = {'Authorization': f'Bearer {token}'} query = """ { ShowVersion (data: {}) { success errors op_mode_error { name message vyos_code } data { result } } } """ r = request('POST', graphql_url, verify=False, headers=headers, json={'query': query}) success = r.json()['data']['ShowVersion']['success'] self.assertTrue(success) @ignore_warning(InsecureRequestWarning) def test_api_add_delete(self): address = '127.0.0.1' key = 'VyOS-key' url = f'https://{address}/retrieve' payload = {'data': '{"op": "showConfig", "path": []}', 'key': f'{key}'} headers = {} self.cli_set(base_path) self.cli_commit() r = request('POST', url, verify=False, headers=headers, data=payload) # api not configured; expect 503 self.assertEqual(r.status_code, 503) self.cli_set(base_path + ['api', 'keys', 'id', 'key-01', 'key', key]) self.cli_commit() r = request('POST', url, verify=False, headers=headers, data=payload) # api configured; expect 200 self.assertEqual(r.status_code, 200) self.cli_delete(base_path + ['api']) self.cli_commit() r = request('POST', url, verify=False, headers=headers, data=payload) # api deleted; expect 503 self.assertEqual(r.status_code, 503) @ignore_warning(InsecureRequestWarning) def test_api_show(self): address = '127.0.0.1' key = 'VyOS-key' url = f'https://{address}/show' headers = {} self.cli_set(base_path + ['api', 'keys', 'id', 'key-01', 'key', key]) self.cli_commit() payload = { 'data': '{"op": "show", "path": ["system", "image"]}', 'key': f'{key}', } r = request('POST', url, verify=False, headers=headers, data=payload) self.assertEqual(r.status_code, 200) @ignore_warning(InsecureRequestWarning) def test_api_generate(self): address = '127.0.0.1' key = 'VyOS-key' url = f'https://{address}/generate' headers = {} self.cli_set(base_path + ['api', 'keys', 'id', 'key-01', 'key', key]) self.cli_commit() payload = { 'data': '{"op": "generate", "path": ["macsec", "mka", "cak", "gcm-aes-256"]}', 'key': f'{key}', } r = request('POST', url, verify=False, headers=headers, data=payload) self.assertEqual(r.status_code, 200) @ignore_warning(InsecureRequestWarning) def test_api_configure(self): address = '127.0.0.1' key = 'VyOS-key' url = f'https://{address}/configure' headers = {} conf_interface = 'dum0' conf_address = '192.0.2.44/32' self.cli_set(base_path + ['api', 'keys', 'id', 'key-01', 'key', key]) self.cli_commit() payload_path = [ "interfaces", "dummy", f"{conf_interface}", "address", f"{conf_address}", ] payload = {'data': json.dumps({"op": "set", "path": payload_path}), 'key': key} r = request('POST', url, verify=False, headers=headers, data=payload) self.assertEqual(r.status_code, 200) @ignore_warning(InsecureRequestWarning) def test_api_config_file(self): address = '127.0.0.1' key = 'VyOS-key' url = f'https://{address}/config-file' headers = {} self.cli_set(base_path + ['api', 'keys', 'id', 'key-01', 'key', key]) self.cli_commit() payload = { 'data': '{"op": "save"}', 'key': f'{key}', } r = request('POST', url, verify=False, headers=headers, data=payload) self.assertEqual(r.status_code, 200) @ignore_warning(InsecureRequestWarning) def test_api_reset(self): address = '127.0.0.1' key = 'VyOS-key' url = f'https://{address}/reset' headers = {} self.cli_set(base_path + ['api', 'keys', 'id', 'key-01', 'key', key]) self.cli_commit() payload = { 'data': '{"op": "reset", "path": ["ip", "arp", "table"]}', 'key': f'{key}', } r = request('POST', url, verify=False, headers=headers, data=payload) self.assertEqual(r.status_code, 200) @ignore_warning(InsecureRequestWarning) def test_api_config_file_load_http(self): - """Test load config from HTTP URL - """ + # Test load config from HTTP URL address = '127.0.0.1' key = 'VyOS-key' url = f'https://{address}/config-file' url_config = f'https://{address}/configure' headers = {} tmp_file = 'tmp-config.boot' nginx_tmp_site = '/etc/nginx/sites-enabled/smoketest' self.cli_set(base_path + ['api', 'keys', 'id', 'key-01', 'key', key]) self.cli_commit() # load config via HTTP requires nginx config call(f'sudo touch {nginx_tmp_site}') call(f'sudo chown vyos:vyattacfg {nginx_tmp_site}') call(f'sudo chmod +w {nginx_tmp_site}') with open(nginx_tmp_site, 'w') as f: f.write(nginx_conf_smoketest) call('sudo nginx -s reload') # save config payload = { 'data': '{"op": "save", "file": "/tmp/tmp-config.boot"}', 'key': f'{key}', } r = request('POST', url, verify=False, headers=headers, data=payload) self.assertEqual(r.status_code, 200) # change config payload = { 'data': '{"op": "set", "path": ["interfaces", "dummy", "dum1", "address", "192.0.2.31/32"]}', 'key': f'{key}', } r = request('POST', url_config, verify=False, headers=headers, data=payload) self.assertEqual(r.status_code, 200) # load config from URL payload = { 'data': '{"op": "load", "file": "http://localhost:8000/tmp-config.boot"}', 'key': f'{key}', } r = request('POST', url, verify=False, headers=headers, data=payload) self.assertEqual(r.status_code, 200) # cleanup tmp nginx conf call(f'sudo rm -rf {nginx_tmp_site}') call('sudo nginx -s reload') if __name__ == '__main__': unittest.main(verbosity=5) diff --git a/src/conf_mode/interfaces_ethernet.py b/src/conf_mode/interfaces_ethernet.py index 7374a29f7..2c0f846c3 100755 --- a/src/conf_mode/interfaces_ethernet.py +++ b/src/conf_mode/interfaces_ethernet.py @@ -1,400 +1,391 @@ #!/usr/bin/env python3 # -# Copyright (C) 2019-2021 VyOS maintainers and contributors +# 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/>. import os import pprint from glob import glob from sys import exit from vyos.base import Warning from vyos.config import Config from vyos.configdict import get_interface_dict from vyos.configdict import is_node_changed from vyos.configverify import verify_address from vyos.configverify import verify_dhcpv6 from vyos.configverify import verify_eapol from vyos.configverify import verify_interface_exists from vyos.configverify import verify_mirror_redirect from vyos.configverify import verify_mtu from vyos.configverify import verify_mtu_ipv6 from vyos.configverify import verify_vlan_config from vyos.configverify import verify_vrf from vyos.configverify import verify_bond_bridge_member from vyos.ethtool import Ethtool from vyos.ifconfig import EthernetIf from vyos.ifconfig import BondIf from vyos.pki import find_chain from vyos.pki import encode_certificate from vyos.pki import load_certificate from vyos.pki import wrap_private_key from vyos.template import render from vyos.utils.process import call from vyos.utils.dict import dict_search from vyos.utils.dict import dict_to_paths_values from vyos.utils.dict import dict_set from vyos.utils.dict import dict_delete from vyos.utils.file import write_file from vyos import ConfigError from vyos import airbag airbag.enable() # XXX: wpa_supplicant works on the source interface cfg_dir = '/run/wpa_supplicant' wpa_suppl_conf = '/run/wpa_supplicant/{ifname}.conf' def update_bond_options(conf: Config, eth_conf: dict) -> list: """ Return list of blocked options if interface is a bond member :param conf: Config object :type conf: Config :param eth_conf: Ethernet config dictionary :type eth_conf: dict :return: List of blocked options :rtype: list """ blocked_list = [] bond_name = list(eth_conf['is_bond_member'].keys())[0] config_without_defaults = conf.get_config_dict( ['interfaces', 'ethernet', eth_conf['ifname']], key_mangling=('-', '_'), get_first_key=True, no_tag_node_value_mangle=True, with_defaults=False, with_recursive_defaults=False) config_with_defaults = conf.get_config_dict( ['interfaces', 'ethernet', eth_conf['ifname']], key_mangling=('-', '_'), get_first_key=True, no_tag_node_value_mangle=True, with_defaults=True, with_recursive_defaults=True) bond_config_with_defaults = conf.get_config_dict( ['interfaces', 'bonding', bond_name], key_mangling=('-', '_'), get_first_key=True, no_tag_node_value_mangle=True, with_defaults=True, with_recursive_defaults=True) eth_dict_paths = dict_to_paths_values(config_without_defaults) eth_path_base = ['interfaces', 'ethernet', eth_conf['ifname']] #if option is configured under ethernet section for option_path, option_value in eth_dict_paths.items(): bond_option_value = dict_search(option_path, bond_config_with_defaults) #If option is allowed for changing then continue if option_path in EthernetIf.get_bond_member_allowed_options(): continue # if option is inherited from bond then set valued from bond interface if option_path in BondIf.get_inherit_bond_options(): # If option equals to bond option then do nothing if option_value == bond_option_value: continue else: # if ethernet has option and bond interface has # then copy it from bond if bond_option_value is not None: if is_node_changed(conf, eth_path_base + option_path.split('.')): Warning( f'Cannot apply "{option_path.replace(".", " ")}" to "{option_value}".' \ f' Interface "{eth_conf["ifname"]}" is a bond member.' \ f' Option is inherited from bond "{bond_name}"') dict_set(option_path, bond_option_value, eth_conf) continue # if ethernet has option and bond interface does not have # then delete it form dict and do not apply it else: if is_node_changed(conf, eth_path_base + option_path.split('.')): Warning( f'Cannot apply "{option_path.replace(".", " ")}".' \ f' Interface "{eth_conf["ifname"]}" is a bond member.' \ f' Option is inherited from bond "{bond_name}"') dict_delete(option_path, eth_conf) blocked_list.append(option_path) # if inherited option is not configured under ethernet section but configured under bond section for option_path in BondIf.get_inherit_bond_options(): bond_option_value = dict_search(option_path, bond_config_with_defaults) if bond_option_value is not None: if option_path not in eth_dict_paths: if is_node_changed(conf, eth_path_base + option_path.split('.')): Warning( f'Cannot apply "{option_path.replace(".", " ")}" to "{dict_search(option_path, config_with_defaults)}".' \ f' Interface "{eth_conf["ifname"]}" is a bond member. ' \ f'Option is inherited from bond "{bond_name}"') dict_set(option_path, bond_option_value, eth_conf) eth_conf['bond_blocked_changes'] = blocked_list return None 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() - # This must be called prior to get_interface_dict(), as this function will - # alter the config level (config.set_level()) - pki = conf.get_config_dict(['pki'], key_mangling=('-', '_'), - get_first_key=True, no_tag_node_value_mangle=True) - base = ['interfaces', 'ethernet'] - ifname, ethernet = get_interface_dict(conf, base) + ifname, ethernet = get_interface_dict(conf, base, with_pki=True) + if 'is_bond_member' in ethernet: update_bond_options(conf, ethernet) - if 'deleted' not in ethernet: - if pki: ethernet['pki'] = pki - tmp = is_node_changed(conf, base + [ifname, 'speed']) if tmp: ethernet.update({'speed_duplex_changed': {}}) tmp = is_node_changed(conf, base + [ifname, 'duplex']) if tmp: ethernet.update({'speed_duplex_changed': {}}) return ethernet - - def verify_speed_duplex(ethernet: dict, ethtool: Ethtool): """ Verify speed and duplex :param ethernet: dictionary which is received from get_interface_dict :type ethernet: dict :param ethtool: Ethernet object :type ethtool: Ethtool """ if ((ethernet['speed'] == 'auto' and ethernet['duplex'] != 'auto') or (ethernet['speed'] != 'auto' and ethernet['duplex'] == 'auto')): raise ConfigError( 'Speed/Duplex missmatch. Must be both auto or manually configured') if ethernet['speed'] != 'auto' and ethernet['duplex'] != 'auto': # We need to verify if the requested speed and duplex setting is # supported by the underlaying NIC. speed = ethernet['speed'] duplex = ethernet['duplex'] if not ethtool.check_speed_duplex(speed, duplex): raise ConfigError( f'Adapter does not support changing speed ' \ f'and duplex settings to: {speed}/{duplex}!') def verify_flow_control(ethernet: dict, ethtool: Ethtool): """ Verify flow control :param ethernet: dictionary which is received from get_interface_dict :type ethernet: dict :param ethtool: Ethernet object :type ethtool: Ethtool """ if 'disable_flow_control' in ethernet: if not ethtool.check_flow_control(): raise ConfigError( 'Adapter does not support changing flow-control settings!') def verify_ring_buffer(ethernet: dict, ethtool: Ethtool): """ Verify ring buffer :param ethernet: dictionary which is received from get_interface_dict :type ethernet: dict :param ethtool: Ethernet object :type ethtool: Ethtool """ if 'ring_buffer' in ethernet: max_rx = ethtool.get_ring_buffer_max('rx') if not max_rx: raise ConfigError( 'Driver does not support RX ring-buffer configuration!') max_tx = ethtool.get_ring_buffer_max('tx') if not max_tx: raise ConfigError( 'Driver does not support TX ring-buffer configuration!') rx = dict_search('ring_buffer.rx', ethernet) if rx and int(rx) > int(max_rx): raise ConfigError(f'Driver only supports a maximum RX ring-buffer ' \ f'size of "{max_rx}" bytes!') tx = dict_search('ring_buffer.tx', ethernet) if tx and int(tx) > int(max_tx): raise ConfigError(f'Driver only supports a maximum TX ring-buffer ' \ f'size of "{max_tx}" bytes!') def verify_offload(ethernet: dict, ethtool: Ethtool): """ Verify offloading capabilities :param ethernet: dictionary which is received from get_interface_dict :type ethernet: dict :param ethtool: Ethernet object :type ethtool: Ethtool """ if dict_search('offload.rps', ethernet) != None: if not os.path.exists(f'/sys/class/net/{ethernet["ifname"]}/queues/rx-0/rps_cpus'): raise ConfigError('Interface does not suport RPS!') driver = ethtool.get_driver_name() # T3342 - Xen driver requires special treatment if driver == 'vif': if int(ethernet['mtu']) > 1500 and dict_search('offload.sg', ethernet) == None: raise ConfigError('Xen netback drivers requires scatter-gatter offloading '\ 'for MTU size larger then 1500 bytes') def verify_allowedbond_changes(ethernet: dict): """ Verify changed options if interface is in bonding :param ethernet: dictionary which is received from get_interface_dict :type ethernet: dict """ if 'bond_blocked_changes' in ethernet: for option in ethernet['bond_blocked_changes']: raise ConfigError(f'Cannot configure "{option.replace(".", " ")}"' \ f' on interface "{ethernet["ifname"]}".' \ f' Interface is a bond member') def verify(ethernet): if 'deleted' in ethernet: return None if 'is_bond_member' in ethernet: verify_bond_member(ethernet) else: verify_ethernet(ethernet) def verify_bond_member(ethernet): """ Verification function for ethernet interface which is in bonding :param ethernet: dictionary which is received from get_interface_dict :type ethernet: dict """ ifname = ethernet['ifname'] verify_interface_exists(ifname) verify_eapol(ethernet) verify_mirror_redirect(ethernet) ethtool = Ethtool(ifname) verify_speed_duplex(ethernet, ethtool) verify_flow_control(ethernet, ethtool) verify_ring_buffer(ethernet, ethtool) verify_offload(ethernet, ethtool) verify_allowedbond_changes(ethernet) def verify_ethernet(ethernet): """ Verification function for simple ethernet interface :param ethernet: dictionary which is received from get_interface_dict :type ethernet: dict """ ifname = ethernet['ifname'] verify_interface_exists(ifname) verify_mtu(ethernet) verify_mtu_ipv6(ethernet) verify_dhcpv6(ethernet) verify_address(ethernet) verify_vrf(ethernet) verify_bond_bridge_member(ethernet) verify_eapol(ethernet) verify_mirror_redirect(ethernet) ethtool = Ethtool(ifname) # No need to check speed and duplex keys as both have default values. verify_speed_duplex(ethernet, ethtool) verify_flow_control(ethernet, ethtool) verify_ring_buffer(ethernet, ethtool) verify_offload(ethernet, ethtool) # use common function to verify VLAN configuration verify_vlan_config(ethernet) return None def generate(ethernet): # render real configuration file once wpa_supplicant_conf = wpa_suppl_conf.format(**ethernet) if 'deleted' in ethernet: # delete configuration on interface removal if os.path.isfile(wpa_supplicant_conf): os.unlink(wpa_supplicant_conf) return None if 'eapol' in ethernet: ifname = ethernet['ifname'] render(wpa_supplicant_conf, 'ethernet/wpa_supplicant.conf.j2', ethernet) cert_file_path = os.path.join(cfg_dir, f'{ifname}_cert.pem') cert_key_path = os.path.join(cfg_dir, f'{ifname}_cert.key') cert_name = ethernet['eapol']['certificate'] pki_cert = ethernet['pki']['certificate'][cert_name] loaded_pki_cert = load_certificate(pki_cert['certificate']) loaded_ca_certs = {load_certificate(c['certificate']) for c in ethernet['pki']['ca'].values()} if 'ca' in ethernet['pki'] else {} cert_full_chain = find_chain(loaded_pki_cert, loaded_ca_certs) write_file(cert_file_path, '\n'.join(encode_certificate(c) for c in cert_full_chain)) write_file(cert_key_path, wrap_private_key(pki_cert['private']['key'])) if 'ca_certificate' in ethernet['eapol']: ca_cert_file_path = os.path.join(cfg_dir, f'{ifname}_ca.pem') ca_chains = [] for ca_cert_name in ethernet['eapol']['ca_certificate']: pki_ca_cert = ethernet['pki']['ca'][ca_cert_name] loaded_ca_cert = load_certificate(pki_ca_cert['certificate']) ca_full_chain = find_chain(loaded_ca_cert, loaded_ca_certs) ca_chains.append( '\n'.join(encode_certificate(c) for c in ca_full_chain)) write_file(ca_cert_file_path, '\n'.join(ca_chains)) return None def apply(ethernet): ifname = ethernet['ifname'] # take care about EAPoL supplicant daemon eapol_action='stop' e = EthernetIf(ifname) if 'deleted' in ethernet: # delete interface e.remove() else: e.update(ethernet) if 'eapol' in ethernet: eapol_action='reload-or-restart' call(f'systemctl {eapol_action} wpa_supplicant-wired@{ifname}') 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_openvpn.py b/src/conf_mode/interfaces_openvpn.py index bdeb44837..45569dd21 100755 --- a/src/conf_mode/interfaces_openvpn.py +++ b/src/conf_mode/interfaces_openvpn.py @@ -1,732 +1,728 @@ #!/usr/bin/env python3 # -# Copyright (C) 2019-2023 VyOS maintainers and contributors +# 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/>. import os import re import tempfile from cryptography.hazmat.primitives.asymmetric import ec from glob import glob from sys import exit from ipaddress import IPv4Address from ipaddress import IPv4Network from ipaddress import IPv6Address from ipaddress import IPv6Network from ipaddress import summarize_address_range from netifaces import interfaces from secrets import SystemRandom from shutil import rmtree from vyos.base import DeprecationWarning from vyos.config import Config from vyos.configdict import get_interface_dict from vyos.configdict import is_node_changed from vyos.configverify import verify_vrf from vyos.configverify import verify_bridge_delete from vyos.configverify import verify_mirror_redirect from vyos.configverify import verify_bond_bridge_member from vyos.ifconfig import VTunIf from vyos.pki import load_dh_parameters from vyos.pki import load_private_key from vyos.pki import sort_ca_chain from vyos.pki import verify_ca_chain from vyos.pki import wrap_certificate from vyos.pki import wrap_crl from vyos.pki import wrap_dh_parameters from vyos.pki import wrap_openvpn_key from vyos.pki import wrap_private_key from vyos.template import render from vyos.template import is_ipv4 from vyos.template import is_ipv6 from vyos.utils.dict import dict_search from vyos.utils.dict import dict_search_args from vyos.utils.list import is_list_equal from vyos.utils.file import makedir from vyos.utils.file import read_file from vyos.utils.file import write_file from vyos.utils.kernel import check_kmod from vyos.utils.kernel import unload_kmod from vyos.utils.process import call from vyos.utils.permission import chown from vyos.utils.process import cmd from vyos.utils.network import is_addr_assigned from vyos import ConfigError from vyos import airbag airbag.enable() user = 'openvpn' group = 'openvpn' cfg_dir = '/run/openvpn' cfg_file = '/run/openvpn/{ifname}.conf' otp_path = '/config/auth/openvpn' otp_file = '/config/auth/openvpn/{ifname}-otp-secrets' secret_chars = list('ABCDEFGHIJKLMNOPQRSTUVWXYZ234567') service_file = '/run/systemd/system/openvpn@{ifname}.service.d/20-override.conf' 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', 'openvpn'] - ifname, openvpn = get_interface_dict(conf, base) + ifname, openvpn = get_interface_dict(conf, base, with_pki=True) openvpn['auth_user_pass_file'] = '/run/openvpn/{ifname}.pw'.format(**openvpn) if 'deleted' in openvpn: return openvpn - openvpn['pki'] = conf.get_config_dict(['pki'], key_mangling=('-', '_'), - get_first_key=True, - no_tag_node_value_mangle=True) - if is_node_changed(conf, base + [ifname, 'openvpn-option']): openvpn.update({'restart_required': {}}) if is_node_changed(conf, base + [ifname, 'enable-dco']): openvpn.update({'restart_required': {}}) # We have to get the dict using 'get_config_dict' instead of 'get_interface_dict' # as 'get_interface_dict' merges the defaults in, so we can not check for defaults in there. tmp = conf.get_config_dict(base + [openvpn['ifname']], get_first_key=True) # We have to cleanup the config dict, as default values could enable features # which are not explicitly enabled on the CLI. Example: server mfa totp # originate comes with defaults, which will enable the # totp plugin, even when not set via CLI so we # need to check this first and drop those keys if dict_search('server.mfa.totp', tmp) == None: del openvpn['server']['mfa'] # OpenVPN Data-Channel-Offload (DCO) is a Kernel module. If loaded it applies to all # OpenVPN interfaces. Check if DCO is used by any other interface instance. tmp = conf.get_config_dict(base, key_mangling=('-', '_'), get_first_key=True) for interface, interface_config in tmp.items(): # If one interface has DCO configured, enable it. No need to further check # all other OpenVPN interfaces. We must use a dedicated key to indicate # the Kernel module must be loaded or not. The per interface "offload.dco" # key is required per OpenVPN interface instance. if dict_search('offload.dco', interface_config) != None: openvpn['module_load_dco'] = {} break return openvpn def is_ec_private_key(pki, cert_name): if not pki or 'certificate' not in pki: return False if cert_name not in pki['certificate']: return False pki_cert = pki['certificate'][cert_name] if 'private' not in pki_cert or 'key' not in pki_cert['private']: return False key = load_private_key(pki_cert['private']['key']) return isinstance(key, ec.EllipticCurvePrivateKey) def verify_pki(openvpn): pki = openvpn['pki'] interface = openvpn['ifname'] mode = openvpn['mode'] shared_secret_key = dict_search_args(openvpn, 'shared_secret_key') tls = dict_search_args(openvpn, 'tls') if not bool(shared_secret_key) ^ bool(tls): # xor check if only one is set raise ConfigError('Must specify only one of "shared-secret-key" and "tls"') if mode in ['server', 'client'] and not tls: raise ConfigError('Must specify "tls" for server and client modes') if not pki: raise ConfigError('PKI is not configured') if shared_secret_key: if not dict_search_args(pki, 'openvpn', 'shared_secret'): raise ConfigError('There are no openvpn shared-secrets in PKI configuration') if shared_secret_key not in pki['openvpn']['shared_secret']: raise ConfigError(f'Invalid shared-secret on openvpn interface {interface}') # If PSK settings are correct, warn about its deprecation - DeprecationWarning("OpenVPN shared-secret support will be removed in future VyOS versions.\n\ - Please migrate your site-to-site tunnels to TLS.\n\ - You can use self-signed certificates with peer fingerprint verification, consult the documentation for details.") + DeprecationWarning('OpenVPN shared-secret support will be removed in future '\ + 'VyOS versions. Please migrate your site-to-site tunnels to '\ + 'TLS. You can use self-signed certificates with peer fingerprint '\ + 'verification, consult the documentation for details.') if tls: if (mode in ['server', 'client']) and ('ca_certificate' not in tls): raise ConfigError(f'Must specify "tls ca-certificate" on openvpn interface {interface},\ it is required in server and client modes') else: if ('ca_certificate' not in tls) and ('peer_fingerprint' not in tls): raise ConfigError('Either "tls ca-certificate" or "tls peer-fingerprint" is required\ on openvpn interface {interface} in site-to-site mode') if 'ca_certificate' in tls: for ca_name in tls['ca_certificate']: if ca_name not in pki['ca']: raise ConfigError(f'Invalid CA certificate on openvpn interface {interface}') if len(tls['ca_certificate']) > 1: sorted_chain = sort_ca_chain(tls['ca_certificate'], pki['ca']) if not verify_ca_chain(sorted_chain, pki['ca']): raise ConfigError(f'CA certificates are not a valid chain') if mode != 'client' and 'auth_key' not in tls: if 'certificate' not in tls: raise ConfigError(f'Missing "tls certificate" on openvpn interface {interface}') if 'certificate' in tls: if tls['certificate'] not in pki['certificate']: raise ConfigError(f'Invalid certificate on openvpn interface {interface}') if dict_search_args(pki, 'certificate', tls['certificate'], 'private', 'password_protected') is not None: raise ConfigError(f'Cannot use encrypted private key on openvpn interface {interface}') if 'dh_params' in tls: pki_dh = pki['dh'][tls['dh_params']] dh_params = load_dh_parameters(pki_dh['parameters']) dh_numbers = dh_params.parameter_numbers() dh_bits = dh_numbers.p.bit_length() if dh_bits < 2048: raise ConfigError(f'Minimum DH key-size is 2048 bits') if 'auth_key' in tls or 'crypt_key' in tls: if not dict_search_args(pki, 'openvpn', 'shared_secret'): raise ConfigError('There are no openvpn shared-secrets in PKI configuration') if 'auth_key' in tls: if tls['auth_key'] not in pki['openvpn']['shared_secret']: raise ConfigError(f'Invalid auth-key on openvpn interface {interface}') if 'crypt_key' in tls: if tls['crypt_key'] not in pki['openvpn']['shared_secret']: raise ConfigError(f'Invalid crypt-key on openvpn interface {interface}') def verify(openvpn): if 'deleted' in openvpn: # remove totp secrets file if totp is not configured if os.path.isfile(otp_file.format(**openvpn)): os.remove(otp_file.format(**openvpn)) verify_bridge_delete(openvpn) return None if 'mode' not in openvpn: raise ConfigError('Must specify OpenVPN operation mode!') # # OpenVPN client mode - VERIFY # if openvpn['mode'] == 'client': if 'local_port' in openvpn: raise ConfigError('Cannot specify "local-port" in client mode') if 'local_host' in openvpn: raise ConfigError('Cannot specify "local-host" in client mode') if 'remote_host' not in openvpn: raise ConfigError('Must specify "remote-host" in client mode') if openvpn['protocol'] == 'tcp-passive': raise ConfigError('Protocol "tcp-passive" is not valid in client mode') if dict_search('tls.dh_params', openvpn): raise ConfigError('Cannot specify "tls dh-params" in client mode') # # OpenVPN site-to-site - VERIFY # elif openvpn['mode'] == 'site-to-site': if 'local_address' not in openvpn and 'is_bridge_member' not in openvpn: raise ConfigError('Must specify "local-address" or add interface to bridge') if 'local_address' in openvpn: if len([addr for addr in openvpn['local_address'] if is_ipv4(addr)]) > 1: raise ConfigError('Only one IPv4 local-address can be specified') if len([addr for addr in openvpn['local_address'] if is_ipv6(addr)]) > 1: raise ConfigError('Only one IPv6 local-address can be specified') if openvpn['device_type'] == 'tun': if 'remote_address' not in openvpn: raise ConfigError('Must specify "remote-address"') if 'remote_address' in openvpn: if len([addr for addr in openvpn['remote_address'] if is_ipv4(addr)]) > 1: raise ConfigError('Only one IPv4 remote-address can be specified') if len([addr for addr in openvpn['remote_address'] if is_ipv6(addr)]) > 1: raise ConfigError('Only one IPv6 remote-address can be specified') if not 'local_address' in openvpn: raise ConfigError('"remote-address" requires "local-address"') v4loAddr = [addr for addr in openvpn['local_address'] if is_ipv4(addr)] v4remAddr = [addr for addr in openvpn['remote_address'] if is_ipv4(addr)] if v4loAddr and not v4remAddr: raise ConfigError('IPv4 "local-address" requires IPv4 "remote-address"') elif v4remAddr and not v4loAddr: raise ConfigError('IPv4 "remote-address" requires IPv4 "local-address"') v6remAddr = [addr for addr in openvpn['remote_address'] if is_ipv6(addr)] v6loAddr = [addr for addr in openvpn['local_address'] if is_ipv6(addr)] if v6loAddr and not v6remAddr: raise ConfigError('IPv6 "local-address" requires IPv6 "remote-address"') elif v6remAddr and not v6loAddr: raise ConfigError('IPv6 "remote-address" requires IPv6 "local-address"') if is_list_equal(v4loAddr, v4remAddr) or is_list_equal(v6loAddr, v6remAddr): raise ConfigError('"local-address" and "remote-address" cannot be the same') if dict_search('local_host', openvpn) in dict_search('local_address', openvpn): raise ConfigError('"local-address" cannot be the same as "local-host"') if dict_search('remote_host', openvpn) in dict_search('remote_address', openvpn): raise ConfigError('"remote-address" and "remote-host" can not be the same') if openvpn['device_type'] == 'tap' and 'local_address' in openvpn: # we can only have one local_address, this is ensured above v4addr = None for laddr in openvpn['local_address']: if is_ipv4(laddr): v4addr = laddr break if v4addr in openvpn['local_address'] and 'subnet_mask' not in openvpn['local_address'][v4addr]: raise ConfigError('Must specify IPv4 "subnet-mask" for local-address') if dict_search('encryption.ncp_ciphers', openvpn): raise ConfigError('NCP ciphers can only be used in client or server mode') else: # checks for client-server or site-to-site bridged if 'local_address' in openvpn or 'remote_address' in openvpn: raise ConfigError('Cannot specify "local-address" or "remote-address" ' \ 'in client/server or bridge mode') # # OpenVPN server mode - VERIFY # if openvpn['mode'] == 'server': if openvpn['protocol'] == 'tcp-active': raise ConfigError('Protocol "tcp-active" is not valid in server mode') if dict_search('authentication.username', openvpn) or dict_search('authentication.password', openvpn): raise ConfigError('Cannot specify "authentication" in server mode') if 'remote_port' in openvpn: raise ConfigError('Cannot specify "remote-port" in server mode') if 'remote_host' in openvpn: raise ConfigError('Cannot specify "remote-host" in server mode') tmp = dict_search('server.subnet', openvpn) if tmp: v4_subnets = len([subnet for subnet in tmp if is_ipv4(subnet)]) v6_subnets = len([subnet for subnet in tmp if is_ipv6(subnet)]) if v4_subnets > 1: raise ConfigError('Cannot specify more than 1 IPv4 server subnet') if v6_subnets > 1: raise ConfigError('Cannot specify more than 1 IPv6 server subnet') for subnet in tmp: if is_ipv4(subnet): subnet = IPv4Network(subnet) if openvpn['device_type'] == 'tun' and subnet.prefixlen > 29: raise ConfigError('Server subnets smaller than /29 with device type "tun" are not supported') elif openvpn['device_type'] == 'tap' and subnet.prefixlen > 30: raise ConfigError('Server subnets smaller than /30 with device type "tap" are not supported') for client in (dict_search('client', openvpn) or []): if client['ip'] and not IPv4Address(client['ip'][0]) in subnet: raise ConfigError(f'Client "{client["name"]}" IP {client["ip"][0]} not in server subnet {subnet}') else: if 'is_bridge_member' not in openvpn: raise ConfigError('Must specify "server subnet" or add interface to bridge in server mode') if hasattr(dict_search('server.client', openvpn), '__iter__'): for client_k, client_v in dict_search('server.client', openvpn).items(): if (client_v.get('ip') and len(client_v['ip']) > 1) or (client_v.get('ipv6_ip') and len(client_v['ipv6_ip']) > 1): raise ConfigError(f'Server client "{client_k}": cannot specify more than 1 IPv4 and 1 IPv6 IP') if dict_search('server.client_ip_pool', openvpn): if not (dict_search('server.client_ip_pool.start', openvpn) and dict_search('server.client_ip_pool.stop', openvpn)): raise ConfigError('Server client-ip-pool requires both start and stop addresses') else: v4PoolStart = IPv4Address(dict_search('server.client_ip_pool.start', openvpn)) v4PoolStop = IPv4Address(dict_search('server.client_ip_pool.stop', openvpn)) if v4PoolStart > v4PoolStop: raise ConfigError(f'Server client-ip-pool start address {v4PoolStart} is larger than stop address {v4PoolStop}') v4PoolSize = int(v4PoolStop) - int(v4PoolStart) if v4PoolSize >= 65536: raise ConfigError(f'Server client-ip-pool is too large [{v4PoolStart} -> {v4PoolStop} = {v4PoolSize}], maximum is 65536 addresses.') v4PoolNets = list(summarize_address_range(v4PoolStart, v4PoolStop)) for client in (dict_search('client', openvpn) or []): if client['ip']: for v4PoolNet in v4PoolNets: if IPv4Address(client['ip'][0]) in v4PoolNet: print(f'Warning: Client "{client["name"]}" IP {client["ip"][0]} is in server IP pool, it is not reserved for this client.') # configuring a client_ip_pool will set 'server ... nopool' which is currently incompatible with 'server-ipv6' (probably to be fixed upstream) for subnet in (dict_search('server.subnet', openvpn) or []): if is_ipv6(subnet): raise ConfigError(f'Setting client-ip-pool is incompatible having an IPv6 server subnet.') for subnet in (dict_search('server.subnet', openvpn) or []): if is_ipv6(subnet): tmp = dict_search('client_ipv6_pool.base', openvpn) if tmp: if not dict_search('server.client_ip_pool', openvpn): raise ConfigError('IPv6 server pool requires an IPv4 server pool') if int(tmp.split('/')[1]) >= 112: raise ConfigError('IPv6 server pool must be larger than /112') # # todo - weird logic # v6PoolStart = IPv6Address(tmp) v6PoolStop = IPv6Network((v6PoolStart, openvpn['server_ipv6_pool_prefixlen']), strict=False)[-1] # don't remove the parentheses, it's a 2-tuple v6PoolSize = int(v6PoolStop) - int(v6PoolStart) if int(openvpn['server_ipv6_pool_prefixlen']) > 96 else 65536 if v6PoolSize < v4PoolSize: raise ConfigError(f'IPv6 server pool must be at least as large as the IPv4 pool (current sizes: IPv6={v6PoolSize} IPv4={v4PoolSize})') v6PoolNets = list(summarize_address_range(v6PoolStart, v6PoolStop)) for client in (dict_search('client', openvpn) or []): if client['ipv6_ip']: for v6PoolNet in v6PoolNets: if IPv6Address(client['ipv6_ip'][0]) in v6PoolNet: print(f'Warning: Client "{client["name"]}" IP {client["ipv6_ip"][0]} is in server IP pool, it is not reserved for this client.') # add mfa users to the file the mfa plugin uses if dict_search('server.mfa.totp', openvpn): user_data = '' if not os.path.isfile(otp_file.format(**openvpn)): write_file(otp_file.format(**openvpn), user_data, user=user, group=group, mode=0o644) ovpn_users = read_file(otp_file.format(**openvpn)) for client in (dict_search('server.client', openvpn) or []): exists = None for ovpn_user in ovpn_users.split('\n'): if re.search('^' + client + ' ', ovpn_user): user_data += f'{ovpn_user}\n' exists = 'true' if not exists: random = SystemRandom() totp_secret = ''.join(random.choice(secret_chars) for _ in range(16)) user_data += f'{client} otp totp:sha1:base32:{totp_secret}::xxx *\n' write_file(otp_file.format(**openvpn), user_data, user=user, group=group, mode=0o644) else: # checks for both client and site-to-site go here if dict_search('server.reject_unconfigured_clients', openvpn): raise ConfigError('Option reject-unconfigured-clients only supported in server mode') if 'replace_default_route' in openvpn and 'remote_host' not in openvpn: raise ConfigError('Cannot set "replace-default-route" without "remote-host"') # # OpenVPN common verification section # not depending on any operation mode # # verify specified IP address is present on any interface on this system if 'local_host' in openvpn: if not is_addr_assigned(openvpn['local_host']): print('local-host IP address "{local_host}" not assigned' \ ' to any interface'.format(**openvpn)) # TCP active if openvpn['protocol'] == 'tcp-active': if 'local_port' in openvpn: raise ConfigError('Cannot specify "local-port" with "tcp-active"') if 'remote_host' not in openvpn: raise ConfigError('Must specify "remote-host" with "tcp-active"') # # TLS/encryption # if 'shared_secret_key' in openvpn: if dict_search('encryption.cipher', openvpn) in ['aes128gcm', 'aes192gcm', 'aes256gcm']: raise ConfigError('GCM encryption with shared-secret-key not supported') if 'tls' in openvpn: if {'auth_key', 'crypt_key'} <= set(openvpn['tls']): raise ConfigError('TLS auth and crypt keys are mutually exclusive') tmp = dict_search('tls.role', openvpn) if tmp: if openvpn['mode'] in ['client', 'server']: if not dict_search('tls.auth_key', openvpn): raise ConfigError('Cannot specify "tls role" in client-server mode') if tmp == 'active': if openvpn['protocol'] == 'tcp-passive': raise ConfigError('Cannot specify "tcp-passive" when "tls role" is "active"') if dict_search('tls.dh_params', openvpn): raise ConfigError('Cannot specify "tls dh-params" when "tls role" is "active"') elif tmp == 'passive': if openvpn['protocol'] == 'tcp-active': raise ConfigError('Cannot specify "tcp-active" when "tls role" is "passive"') if 'certificate' in openvpn['tls'] and is_ec_private_key(openvpn['pki'], openvpn['tls']['certificate']): if 'dh_params' in openvpn['tls']: print('Warning: using dh-params and EC keys simultaneously will ' \ 'lead to DH ciphers being used instead of ECDH') if dict_search('encryption.cipher', openvpn) == 'none': print('Warning: "encryption none" was specified!') print('No encryption will be performed and data is transmitted in ' \ 'plain text over the network!') verify_pki(openvpn) # # Auth user/pass # if (dict_search('authentication.username', openvpn) and not dict_search('authentication.password', openvpn)): raise ConfigError('Password for authentication is missing') if (dict_search('authentication.password', openvpn) and not dict_search('authentication.username', openvpn)): raise ConfigError('Username for authentication is missing') verify_vrf(openvpn) verify_bond_bridge_member(openvpn) verify_mirror_redirect(openvpn) return None def generate_pki_files(openvpn): pki = openvpn['pki'] if not pki: return None interface = openvpn['ifname'] shared_secret_key = dict_search_args(openvpn, 'shared_secret_key') tls = dict_search_args(openvpn, 'tls') if shared_secret_key: pki_key = pki['openvpn']['shared_secret'][shared_secret_key] key_path = os.path.join(cfg_dir, f'{interface}_shared.key') write_file(key_path, wrap_openvpn_key(pki_key['key']), user=user, group=group) if tls: if 'ca_certificate' in tls: cert_path = os.path.join(cfg_dir, f'{interface}_ca.pem') crl_path = os.path.join(cfg_dir, f'{interface}_crl.pem') if os.path.exists(cert_path): os.unlink(cert_path) if os.path.exists(crl_path): os.unlink(crl_path) for cert_name in sort_ca_chain(tls['ca_certificate'], pki['ca']): pki_ca = pki['ca'][cert_name] if 'certificate' in pki_ca: write_file(cert_path, wrap_certificate(pki_ca['certificate']) + "\n", user=user, group=group, mode=0o600, append=True) if 'crl' in pki_ca: for crl in pki_ca['crl']: write_file(crl_path, wrap_crl(crl) + "\n", user=user, group=group, mode=0o600, append=True) openvpn['tls']['crl'] = True if 'certificate' in tls: cert_name = tls['certificate'] pki_cert = pki['certificate'][cert_name] if 'certificate' in pki_cert: cert_path = os.path.join(cfg_dir, f'{interface}_cert.pem') write_file(cert_path, wrap_certificate(pki_cert['certificate']), user=user, group=group, mode=0o600) if 'private' in pki_cert and 'key' in pki_cert['private']: key_path = os.path.join(cfg_dir, f'{interface}_cert.key') write_file(key_path, wrap_private_key(pki_cert['private']['key']), user=user, group=group, mode=0o600) openvpn['tls']['private_key'] = True if 'dh_params' in tls: dh_name = tls['dh_params'] pki_dh = pki['dh'][dh_name] if 'parameters' in pki_dh: dh_path = os.path.join(cfg_dir, f'{interface}_dh.pem') write_file(dh_path, wrap_dh_parameters(pki_dh['parameters']), user=user, group=group, mode=0o600) if 'auth_key' in tls: key_name = tls['auth_key'] pki_key = pki['openvpn']['shared_secret'][key_name] if 'key' in pki_key: key_path = os.path.join(cfg_dir, f'{interface}_auth.key') write_file(key_path, wrap_openvpn_key(pki_key['key']), user=user, group=group, mode=0o600) if 'crypt_key' in tls: key_name = tls['crypt_key'] pki_key = pki['openvpn']['shared_secret'][key_name] if 'key' in pki_key: key_path = os.path.join(cfg_dir, f'{interface}_crypt.key') write_file(key_path, wrap_openvpn_key(pki_key['key']), user=user, group=group, mode=0o600) def generate(openvpn): interface = openvpn['ifname'] directory = os.path.dirname(cfg_file.format(**openvpn)) openvpn['plugin_dir'] = '/usr/lib/openvpn' # create base config directory on demand makedir(directory, user, group) # enforce proper permissions on /run/openvpn chown(directory, user, group) # we can't know in advance which clients have been removed, # thus all client configs will be removed and re-added on demand ccd_dir = os.path.join(directory, 'ccd', interface) if os.path.isdir(ccd_dir): rmtree(ccd_dir, ignore_errors=True) # Remove systemd directories with overrides service_dir = os.path.dirname(service_file.format(**openvpn)) if os.path.isdir(service_dir): rmtree(service_dir, ignore_errors=True) if 'deleted' in openvpn or 'disable' in openvpn: return None # create client config directory on demand makedir(ccd_dir, user, group) # Fix file permissons for keys generate_pki_files(openvpn) # Generate User/Password authentication file if 'authentication' in openvpn: render(openvpn['auth_user_pass_file'], 'openvpn/auth.pw.j2', openvpn, user=user, group=group, permission=0o600) else: # delete old auth file if present if os.path.isfile(openvpn['auth_user_pass_file']): os.remove(openvpn['auth_user_pass_file']) # Generate client specific configuration server_client = dict_search_args(openvpn, 'server', 'client') if server_client: for client, client_config in server_client.items(): client_file = os.path.join(ccd_dir, client) # Our client need's to know its subnet mask ... client_config['server_subnet'] = dict_search('server.subnet', openvpn) render(client_file, 'openvpn/client.conf.j2', client_config, user=user, group=group) # we need to support quoting of raw parameters from OpenVPN CLI # see https://vyos.dev/T1632 render(cfg_file.format(**openvpn), 'openvpn/server.conf.j2', openvpn, formater=lambda _: _.replace(""", '"'), user=user, group=group) # Render 20-override.conf for OpenVPN service render(service_file.format(**openvpn), 'openvpn/service-override.conf.j2', openvpn, formater=lambda _: _.replace(""", '"'), user=user, group=group) # Reload systemd services config to apply an override call(f'systemctl daemon-reload') return None def apply(openvpn): interface = openvpn['ifname'] # Do some cleanup when OpenVPN is disabled/deleted if 'deleted' in openvpn or 'disable' in openvpn: call(f'systemctl stop openvpn@{interface}.service') for cleanup_file in glob(f'/run/openvpn/{interface}.*'): if os.path.isfile(cleanup_file): os.unlink(cleanup_file) if interface in interfaces(): VTunIf(interface).remove() # dynamically load/unload DCO Kernel extension if requested dco_module = 'ovpn_dco_v2' if 'module_load_dco' in openvpn: check_kmod(dco_module) else: unload_kmod(dco_module) # Now bail out early if interface is disabled or got deleted if 'deleted' in openvpn or 'disable' in openvpn: return None # verify specified IP address is present on any interface on this system # Allow to bind service to nonlocal address, if it virtaual-vrrp address # or if address will be assign later if 'local_host' in openvpn: if not is_addr_assigned(openvpn['local_host']): cmd('sysctl -w net.ipv4.ip_nonlocal_bind=1') # No matching OpenVPN process running - maybe it got killed or none # existed - nevertheless, spawn new OpenVPN process action = 'reload-or-restart' if 'restart_required' in openvpn: action = 'restart' call(f'systemctl {action} openvpn@{interface}.service') o = VTunIf(**openvpn) o.update(openvpn) 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_sstpc.py b/src/conf_mode/interfaces_sstpc.py index b588910dc..b9d7a74fb 100755 --- a/src/conf_mode/interfaces_sstpc.py +++ b/src/conf_mode/interfaces_sstpc.py @@ -1,145 +1,141 @@ #!/usr/bin/env python3 # # Copyright (C) 2022 VyOS maintainers and contributors # # This program is free software; you can redistribute it and/or modify # it under the terms of the GNU General Public License version 2 or later as # published by the Free Software Foundation. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program. If not, see <http://www.gnu.org/licenses/>. import os from sys import exit from vyos.config import Config from vyos.configdict import get_interface_dict from vyos.configdict import is_node_changed from vyos.configverify import verify_authentication from vyos.configverify import verify_vrf from vyos.ifconfig import SSTPCIf from vyos.pki import encode_certificate from vyos.pki import find_chain from vyos.pki import load_certificate from vyos.template import render from vyos.utils.process import call from vyos.utils.dict import dict_search from vyos.utils.process import is_systemd_service_running from vyos.utils.file import write_file 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', 'sstpc'] - ifname, sstpc = get_interface_dict(conf, base) + ifname, sstpc = get_interface_dict(conf, base, with_pki=True) # We should only terminate the SSTP client session if critical parameters # change. All parameters that can be changed on-the-fly (like interface # description) should not lead to a reconnect! for options in ['authentication', 'no_peer_dns', 'no_default_route', 'server', 'ssl']: if is_node_changed(conf, base + [ifname, options]): sstpc.update({'shutdown_required': {}}) # bail out early - no need to further process other nodes break - # Load PKI certificates for later processing - sstpc['pki'] = conf.get_config_dict(['pki'], key_mangling=('-', '_'), - get_first_key=True, - no_tag_node_value_mangle=True) return sstpc def verify(sstpc): if 'deleted' in sstpc: return None verify_authentication(sstpc) verify_vrf(sstpc) if not dict_search('server', sstpc): raise ConfigError('Remote SSTP server must be specified!') if not dict_search('ssl.ca_certificate', sstpc): raise ConfigError('Missing mandatory CA certificate!') return None def generate(sstpc): ifname = sstpc['ifname'] config_sstpc = f'/etc/ppp/peers/{ifname}' sstpc['ca_file_path'] = f'/run/sstpc/{ifname}_ca-cert.pem' if 'deleted' in sstpc: for file in [sstpc['ca_file_path'], config_sstpc]: if os.path.exists(file): os.unlink(file) return None ca_name = sstpc['ssl']['ca_certificate'] pki_ca_cert = sstpc['pki']['ca'][ca_name] loaded_ca_cert = load_certificate(pki_ca_cert['certificate']) loaded_ca_certs = {load_certificate(c['certificate']) for c in sstpc['pki']['ca'].values()} if 'ca' in sstpc['pki'] else {} ca_full_chain = find_chain(loaded_ca_cert, loaded_ca_certs) write_file(sstpc['ca_file_path'], '\n'.join(encode_certificate(c) for c in ca_full_chain)) render(config_sstpc, 'sstp-client/peer.j2', sstpc, permission=0o640) return None def apply(sstpc): ifname = sstpc['ifname'] if 'deleted' in sstpc or 'disable' in sstpc: if os.path.isdir(f'/sys/class/net/{ifname}'): p = SSTPCIf(ifname) p.remove() call(f'systemctl stop ppp@{ifname}.service') return None # reconnect should only be necessary when specific options change, # like server, authentication ... (see get_config() for details) if ((not is_systemd_service_running(f'ppp@{ifname}.service')) or 'shutdown_required' in sstpc): # cleanup system (e.g. FRR routes first) if os.path.isdir(f'/sys/class/net/{ifname}'): p = SSTPCIf(ifname) p.remove() call(f'systemctl restart ppp@{ifname}.service') # When interface comes "live" a hook is called: # /etc/ppp/ip-up.d/96-vyos-sstpc-callback # which triggers SSTPCIf.update() else: if os.path.isdir(f'/sys/class/net/{ifname}'): p = SSTPCIf(ifname) p.update(sstpc) 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/load-balancing_reverse-proxy.py b/src/conf_mode/load-balancing_reverse-proxy.py index 333ebc66c..7338fe573 100755 --- a/src/conf_mode/load-balancing_reverse-proxy.py +++ b/src/conf_mode/load-balancing_reverse-proxy.py @@ -1,169 +1,166 @@ #!/usr/bin/env python3 # -# Copyright (C) 2023 VyOS maintainers and contributors +# 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/>. import os from sys import exit from shutil import rmtree from vyos.config import Config from vyos.utils.process import call from vyos.utils.network import check_port_availability from vyos.utils.network import is_listen_port_bind_service from vyos.pki import wrap_certificate from vyos.pki import wrap_private_key from vyos.template import render from vyos import ConfigError from vyos import airbag airbag.enable() load_balancing_dir = '/run/haproxy' load_balancing_conf_file = f'{load_balancing_dir}/haproxy.cfg' systemd_service = 'haproxy.service' systemd_override = r'/run/systemd/system/haproxy.service.d/10-override.conf' def get_config(config=None): if config: conf = config else: conf = Config() base = ['load-balancing', 'reverse-proxy'] + if not conf.exists(base): + return None lb = conf.get_config_dict(base, get_first_key=True, key_mangling=('-', '_'), - no_tag_node_value_mangle=True) - - if lb: - lb['pki'] = conf.get_config_dict(['pki'], key_mangling=('-', '_'), - get_first_key=True, no_tag_node_value_mangle=True) - - if lb: - lb = conf.merge_defaults(lb, recursive=True) + no_tag_node_value_mangle=True, + with_recursive_defaults=True, + with_pki=True) return lb def verify(lb): if not lb: return None if 'backend' not in lb or 'service' not in lb: raise ConfigError(f'"service" and "backend" must be configured!') for front, front_config in lb['service'].items(): if 'port' not in front_config: raise ConfigError(f'"{front} service port" must be configured!') # Check if bind address:port are used by another service tmp_address = front_config.get('address', '0.0.0.0') tmp_port = front_config['port'] if check_port_availability(tmp_address, int(tmp_port), 'tcp') is not True and \ not is_listen_port_bind_service(int(tmp_port), 'haproxy'): raise ConfigError(f'"TCP" port "{tmp_port}" is used by another service') for back, back_config in lb['backend'].items(): if 'server' not in back_config: raise ConfigError(f'"{back} server" must be configured!') for bk_server, bk_server_conf in back_config['server'].items(): if 'address' not in bk_server_conf or 'port' not in bk_server_conf: raise ConfigError(f'"backend {back} server {bk_server} address and port" must be configured!') if {'send_proxy', 'send_proxy_v2'} <= set(bk_server_conf): raise ConfigError(f'Cannot use both "send-proxy" and "send-proxy-v2" for server "{bk_server}"') def generate(lb): if not lb: # Delete /run/haproxy/haproxy.cfg config_files = [load_balancing_conf_file, systemd_override] for file in config_files: if os.path.isfile(file): os.unlink(file) # Delete old directories if os.path.isdir(load_balancing_dir): rmtree(load_balancing_dir, ignore_errors=True) return None # Create load-balance dir if not os.path.isdir(load_balancing_dir): os.mkdir(load_balancing_dir) # SSL Certificates for frontend for front, front_config in lb['service'].items(): if 'ssl' in front_config: if 'certificate' in front_config['ssl']: cert_names = front_config['ssl']['certificate'] for cert_name in cert_names: pki_cert = lb['pki']['certificate'][cert_name] cert_file_path = os.path.join(load_balancing_dir, f'{cert_name}.pem') cert_key_path = os.path.join(load_balancing_dir, f'{cert_name}.pem.key') with open(cert_file_path, 'w') as f: f.write(wrap_certificate(pki_cert['certificate'])) if 'private' in pki_cert and 'key' in pki_cert['private']: with open(cert_key_path, 'w') as f: f.write(wrap_private_key(pki_cert['private']['key'])) if 'ca_certificate' in front_config['ssl']: ca_name = front_config['ssl']['ca_certificate'] pki_ca_cert = lb['pki']['ca'][ca_name] ca_cert_file_path = os.path.join(load_balancing_dir, f'{ca_name}.pem') with open(ca_cert_file_path, 'w') as f: f.write(wrap_certificate(pki_ca_cert['certificate'])) # SSL Certificates for backend for back, back_config in lb['backend'].items(): if 'ssl' in back_config: if 'ca_certificate' in back_config['ssl']: ca_name = back_config['ssl']['ca_certificate'] pki_ca_cert = lb['pki']['ca'][ca_name] ca_cert_file_path = os.path.join(load_balancing_dir, f'{ca_name}.pem') with open(ca_cert_file_path, 'w') as f: f.write(wrap_certificate(pki_ca_cert['certificate'])) render(load_balancing_conf_file, 'load-balancing/haproxy.cfg.j2', lb) render(systemd_override, 'load-balancing/override_haproxy.conf.j2', lb) return None def apply(lb): call('systemctl daemon-reload') if not lb: call(f'systemctl stop {systemd_service}') else: call(f'systemctl reload-or-restart {systemd_service}') 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/service_https.py b/src/conf_mode/service_https.py index 3dc5dfc01..cb40acc9f 100755 --- a/src/conf_mode/service_https.py +++ b/src/conf_mode/service_https.py @@ -1,335 +1,330 @@ #!/usr/bin/env python3 # -# Copyright (C) 2019-2023 VyOS maintainers and contributors +# 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/>. import os import sys import json from copy import deepcopy from time import sleep import vyos.defaults import vyos.certbot_util from vyos.base import Warning from vyos.config import Config from vyos.configdiff import get_config_diff from vyos.configverify import verify_vrf from vyos import ConfigError from vyos.pki import wrap_certificate from vyos.pki import wrap_private_key from vyos.template import render from vyos.utils.process import call from vyos.utils.process import is_systemd_service_running from vyos.utils.process import is_systemd_service_active from vyos.utils.network import check_port_availability from vyos.utils.network import is_listen_port_bind_service from vyos.utils.file import write_file from vyos import airbag airbag.enable() config_file = '/etc/nginx/sites-available/default' systemd_override = r'/run/systemd/system/nginx.service.d/override.conf' cert_dir = '/etc/ssl/certs' key_dir = '/etc/ssl/private' certbot_dir = vyos.defaults.directories['certbot'] api_config_state = '/run/http-api-state' systemd_service = '/run/systemd/system/vyos-http-api.service' # https config needs to coordinate several subsystems: api, certbot, # self-signed certificate, as well as the virtual hosts defined within the # https config definition itself. Consequently, one needs a general dict, # encompassing the https and other configs, and a list of such virtual hosts # (server blocks in nginx terminology) to pass to the jinja2 template. default_server_block = { 'id' : '', 'address' : '*', 'port' : '443', 'name' : ['_'], 'api' : False, 'vyos_cert' : {}, 'certbot' : False } def get_config(config=None): if config: conf = config else: conf = Config() base = ['service', 'https'] if not conf.exists(base): return None diff = get_config_diff(conf) - https = conf.get_config_dict(base, get_first_key=True) - - if https: - https['pki'] = conf.get_config_dict(['pki'], key_mangling=('-', '_'), - no_tag_node_value_mangle=True, - get_first_key=True) + https = conf.get_config_dict(base, get_first_key=True, with_pki=True) https['children_changed'] = diff.node_changed_children(base) https['api_add_or_delete'] = diff.node_changed_presence(base + ['api']) if 'api' not in https: return https http_api = conf.get_config_dict(base + ['api'], key_mangling=('-', '_'), no_tag_node_value_mangle=True, get_first_key=True, with_recursive_defaults=True) if http_api.from_defaults(['graphql']): del http_api['graphql'] # Do we run inside a VRF context? vrf_path = ['service', 'https', 'vrf'] if conf.exists(vrf_path): http_api['vrf'] = conf.return_value(vrf_path) https['api'] = http_api return https def verify(https): from vyos.utils.dict import dict_search if https is None: return None if 'certificates' in https: certificates = https['certificates'] if 'certificate' in certificates: if not https['pki']: - raise ConfigError("PKI is not configured") + raise ConfigError('PKI is not configured') cert_name = certificates['certificate'] if cert_name not in https['pki']['certificate']: raise ConfigError("Invalid certificate on https configuration") pki_cert = https['pki']['certificate'][cert_name] if 'certificate' not in pki_cert: raise ConfigError("Missing certificate on https configuration") if 'private' not in pki_cert or 'key' not in pki_cert['private']: raise ConfigError("Missing certificate private key on https configuration") if 'certbot' in https['certificates']: vhost_names = [] for _, vh_conf in https.get('virtual-host', {}).items(): vhost_names += vh_conf.get('server-name', []) domains = https['certificates']['certbot'].get('domain-name', []) domains_found = [domain for domain in domains if domain in vhost_names] if not domains_found: raise ConfigError("At least one 'virtual-host <id> server-name' " "matching the 'certbot domain-name' is required.") server_block_list = [] # organize by vhosts vhost_dict = https.get('virtual-host', {}) if not vhost_dict: # no specified virtual hosts (server blocks); use default server_block_list.append(default_server_block) else: for vhost in list(vhost_dict): server_block = deepcopy(default_server_block) data = vhost_dict.get(vhost, {}) server_block['address'] = data.get('listen-address', '*') server_block['port'] = data.get('port', '443') server_block_list.append(server_block) for entry in server_block_list: _address = entry.get('address') _address = '0.0.0.0' if _address == '*' else _address _port = entry.get('port') proto = 'tcp' if check_port_availability(_address, int(_port), proto) is not True and \ not is_listen_port_bind_service(int(_port), 'nginx'): raise ConfigError(f'"{proto}" port "{_port}" is used by another service') verify_vrf(https) # Verify API server settings, if present if 'api' in https: keys = dict_search('api.keys.id', https) gql_auth_type = dict_search('api.graphql.authentication.type', https) # If "api graphql" is not defined and `gql_auth_type` is None, # there's certainly no JWT auth option, and keys are required jwt_auth = (gql_auth_type == "token") # Check for incomplete key configurations in every case valid_keys_exist = False if keys: for k in keys: if 'key' not in keys[k]: raise ConfigError(f'Missing HTTPS API key string for key id "{k}"') else: valid_keys_exist = True # If only key-based methods are enabled, # fail the commit if no valid key configurations are found if (not valid_keys_exist) and (not jwt_auth): raise ConfigError('At least one HTTPS API key is required unless GraphQL token authentication is enabled') if (not valid_keys_exist) and jwt_auth: Warning(f'API keys are not configured: the classic (non-GraphQL) API will be unavailable.') return None def generate(https): if https is None: return None if 'api' not in https: if os.path.exists(systemd_service): os.unlink(systemd_service) else: render(systemd_service, 'https/vyos-http-api.service.j2', https['api']) with open(api_config_state, 'w') as f: json.dump(https['api'], f, indent=2) server_block_list = [] # organize by vhosts vhost_dict = https.get('virtual-host', {}) if not vhost_dict: # no specified virtual hosts (server blocks); use default server_block_list.append(default_server_block) else: for vhost in list(vhost_dict): server_block = deepcopy(default_server_block) server_block['id'] = vhost data = vhost_dict.get(vhost, {}) server_block['address'] = data.get('listen-address', '*') server_block['port'] = data.get('port', '443') name = data.get('server-name', ['_']) server_block['name'] = name allow_client = data.get('allow-client', {}) server_block['allow_client'] = allow_client.get('address', []) server_block_list.append(server_block) # get certificate data cert_dict = https.get('certificates', {}) if 'certificate' in cert_dict: cert_name = cert_dict['certificate'] pki_cert = https['pki']['certificate'][cert_name] cert_path = os.path.join(cert_dir, f'{cert_name}.pem') key_path = os.path.join(key_dir, f'{cert_name}.pem') server_cert = str(wrap_certificate(pki_cert['certificate'])) if 'ca-certificate' in cert_dict: ca_cert = cert_dict['ca-certificate'] server_cert += '\n' + str(wrap_certificate(https['pki']['ca'][ca_cert]['certificate'])) write_file(cert_path, server_cert) write_file(key_path, wrap_private_key(pki_cert['private']['key'])) vyos_cert_data = { 'crt': cert_path, 'key': key_path } for block in server_block_list: block['vyos_cert'] = vyos_cert_data # letsencrypt certificate using certbot certbot = False cert_domains = cert_dict.get('certbot', {}).get('domain-name', []) if cert_domains: certbot = True for domain in cert_domains: sub_list = vyos.certbot_util.choose_server_block(server_block_list, domain) if sub_list: for sb in sub_list: sb['certbot'] = True sb['certbot_dir'] = certbot_dir # certbot organizes certificates by first domain sb['certbot_domain_dir'] = cert_domains[0] if 'api' in list(https): vhost_list = https.get('api-restrict', {}).get('virtual-host', []) if not vhost_list: for block in server_block_list: block['api'] = True else: for block in server_block_list: if block['id'] in vhost_list: block['api'] = True data = { 'server_block_list': server_block_list, 'certbot': certbot } render(config_file, 'https/nginx.default.j2', data) render(systemd_override, 'https/override.conf.j2', https) return None def apply(https): # Reload systemd manager configuration call('systemctl daemon-reload') http_api_service_name = 'vyos-http-api.service' https_service_name = 'nginx.service' if https is None: if is_systemd_service_active(f'{http_api_service_name}'): call(f'systemctl stop {http_api_service_name}') call(f'systemctl stop {https_service_name}') return if 'api' in https['children_changed']: if 'api' in https: if is_systemd_service_running(f'{http_api_service_name}'): call(f'systemctl reload {http_api_service_name}') else: call(f'systemctl restart {http_api_service_name}') # Let uvicorn settle before (possibly) restarting nginx sleep(1) else: if is_systemd_service_active(f'{http_api_service_name}'): call(f'systemctl stop {http_api_service_name}') if (not is_systemd_service_running(f'{https_service_name}') or https['api_add_or_delete'] or set(https['children_changed']) - set(['api'])): call(f'systemctl restart {https_service_name}') if __name__ == '__main__': try: c = get_config() verify(c) generate(c) apply(c) except ConfigError as e: print(e) sys.exit(1) diff --git a/src/conf_mode/vpn_ipsec.py b/src/conf_mode/vpn_ipsec.py index 9e9385ddb..7fd32c230 100755 --- a/src/conf_mode/vpn_ipsec.py +++ b/src/conf_mode/vpn_ipsec.py @@ -1,596 +1,594 @@ #!/usr/bin/env python3 # -# Copyright (C) 2021-2023 VyOS maintainers and contributors +# 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 ipaddress import os import re import jmespath from sys import exit from time import sleep from time import time from vyos.base import Warning from vyos.config import Config from vyos.configdict import leaf_node_changed from vyos.configverify import verify_interface_exists from vyos.defaults import directories from vyos.ifconfig import Interface from vyos.pki import encode_certificate from vyos.pki import encode_public_key from vyos.pki import find_chain from vyos.pki import load_certificate from vyos.pki import load_private_key from vyos.pki import wrap_certificate from vyos.pki import wrap_crl from vyos.pki import wrap_public_key from vyos.pki import wrap_private_key from vyos.template import ip_from_cidr from vyos.template import is_ipv4 from vyos.template import is_ipv6 from vyos.template import render from vyos.utils.network import is_ipv6_link_local from vyos.utils.dict import dict_search from vyos.utils.dict import dict_search_args from vyos.utils.process import call from vyos.utils.process import run from vyos import ConfigError from vyos import airbag airbag.enable() dhcp_wait_attempts = 2 dhcp_wait_sleep = 1 swanctl_dir = '/etc/swanctl' charon_conf = '/etc/strongswan.d/charon.conf' charon_dhcp_conf = '/etc/strongswan.d/charon/dhcp.conf' charon_radius_conf = '/etc/strongswan.d/charon/eap-radius.conf' interface_conf = '/etc/strongswan.d/interfaces_use.conf' swanctl_conf = f'{swanctl_dir}/swanctl.conf' default_install_routes = 'yes' vici_socket = '/var/run/charon.vici' CERT_PATH = f'{swanctl_dir}/x509/' PUBKEY_PATH = f'{swanctl_dir}/pubkey/' KEY_PATH = f'{swanctl_dir}/private/' CA_PATH = f'{swanctl_dir}/x509ca/' CRL_PATH = f'{swanctl_dir}/x509crl/' DHCP_HOOK_IFLIST = '/tmp/ipsec_dhcp_waiting' def get_config(config=None): if config: conf = config else: conf = Config() base = ['vpn', 'ipsec'] l2tp_base = ['vpn', 'l2tp', 'remote-access', 'ipsec-settings'] if not conf.exists(base): return None # retrieve common dictionary keys ipsec = conf.get_config_dict(base, key_mangling=('-', '_'), no_tag_node_value_mangle=True, get_first_key=True, - with_recursive_defaults=True) + with_recursive_defaults=True, + with_pki=True) ipsec['dhcp_no_address'] = {} ipsec['install_routes'] = 'no' if conf.exists(base + ["options", "disable-route-autoinstall"]) else default_install_routes ipsec['interface_change'] = leaf_node_changed(conf, base + ['interface']) ipsec['nhrp_exists'] = conf.exists(['protocols', 'nhrp', 'tunnel']) - ipsec['pki'] = conf.get_config_dict(['pki'], key_mangling=('-', '_'), - no_tag_node_value_mangle=True, - get_first_key=True) tmp = conf.get_config_dict(l2tp_base, key_mangling=('-', '_'), no_tag_node_value_mangle=True, get_first_key=True) if tmp: ipsec['l2tp'] = conf.merge_defaults(tmp, recursive=True) ipsec['l2tp_outside_address'] = conf.return_value(['vpn', 'l2tp', 'remote-access', 'outside-address']) ipsec['l2tp_ike_default'] = 'aes256-sha1-modp1024,3des-sha1-modp1024' ipsec['l2tp_esp_default'] = 'aes256-sha1,3des-sha1' return ipsec def get_dhcp_address(iface): addresses = Interface(iface).get_addr() if not addresses: return None for address in addresses: if not is_ipv6_link_local(address): return ip_from_cidr(address) return None def verify_pki_x509(pki, x509_conf): if not pki or 'ca' not in pki or 'certificate' not in pki: raise ConfigError(f'PKI is not configured') ca_cert_name = x509_conf['ca_certificate'] cert_name = x509_conf['certificate'] if not dict_search_args(pki, 'ca', ca_cert_name, 'certificate'): raise ConfigError(f'Missing CA certificate on specified PKI CA certificate "{ca_cert_name}"') if not dict_search_args(pki, 'certificate', cert_name, 'certificate'): raise ConfigError(f'Missing certificate on specified PKI certificate "{cert_name}"') if not dict_search_args(pki, 'certificate', cert_name, 'private', 'key'): raise ConfigError(f'Missing private key on specified PKI certificate "{cert_name}"') return True def verify_pki_rsa(pki, rsa_conf): if not pki or 'key_pair' not in pki: raise ConfigError(f'PKI is not configured') local_key = rsa_conf['local_key'] remote_key = rsa_conf['remote_key'] if not dict_search_args(pki, 'key_pair', local_key, 'private', 'key'): raise ConfigError(f'Missing private key on specified local-key "{local_key}"') if not dict_search_args(pki, 'key_pair', remote_key, 'public', 'key'): raise ConfigError(f'Missing public key on specified remote-key "{remote_key}"') return True def verify(ipsec): if not ipsec: return None if 'authentication' in ipsec: if 'psk' in ipsec['authentication']: for psk, psk_config in ipsec['authentication']['psk'].items(): if 'id' not in psk_config or 'secret' not in psk_config: raise ConfigError(f'Authentication psk "{psk}" missing "id" or "secret"') if 'interfaces' in ipsec : for ifname in ipsec['interface']: verify_interface_exists(ifname) if 'l2tp' in ipsec: if 'esp_group' in ipsec['l2tp']: if 'esp_group' not in ipsec or ipsec['l2tp']['esp_group'] not in ipsec['esp_group']: raise ConfigError(f"Invalid esp-group on L2TP remote-access config") if 'ike_group' in ipsec['l2tp']: if 'ike_group' not in ipsec or ipsec['l2tp']['ike_group'] not in ipsec['ike_group']: raise ConfigError(f"Invalid ike-group on L2TP remote-access config") if 'authentication' not in ipsec['l2tp']: raise ConfigError(f'Missing authentication settings on L2TP remote-access config') if 'mode' not in ipsec['l2tp']['authentication']: raise ConfigError(f'Missing authentication mode on L2TP remote-access config') if not ipsec['l2tp_outside_address']: raise ConfigError(f'Missing outside-address on L2TP remote-access config') if ipsec['l2tp']['authentication']['mode'] == 'pre-shared-secret': if 'pre_shared_secret' not in ipsec['l2tp']['authentication']: raise ConfigError(f'Missing pre shared secret on L2TP remote-access config') if ipsec['l2tp']['authentication']['mode'] == 'x509': if 'x509' not in ipsec['l2tp']['authentication']: raise ConfigError(f'Missing x509 settings on L2TP remote-access config') x509 = ipsec['l2tp']['authentication']['x509'] if 'ca_certificate' not in x509 or 'certificate' not in x509: raise ConfigError(f'Missing x509 certificates on L2TP remote-access config') verify_pki_x509(ipsec['pki'], x509) if 'profile' in ipsec: for profile, profile_conf in ipsec['profile'].items(): if 'esp_group' in profile_conf: if 'esp_group' not in ipsec or profile_conf['esp_group'] not in ipsec['esp_group']: raise ConfigError(f"Invalid esp-group on {profile} profile") else: raise ConfigError(f"Missing esp-group on {profile} profile") if 'ike_group' in profile_conf: if 'ike_group' not in ipsec or profile_conf['ike_group'] not in ipsec['ike_group']: raise ConfigError(f"Invalid ike-group on {profile} profile") else: raise ConfigError(f"Missing ike-group on {profile} profile") if 'authentication' not in profile_conf: raise ConfigError(f"Missing authentication on {profile} profile") if 'remote_access' in ipsec: if 'connection' in ipsec['remote_access']: for name, ra_conf in ipsec['remote_access']['connection'].items(): if 'esp_group' in ra_conf: if 'esp_group' not in ipsec or ra_conf['esp_group'] not in ipsec['esp_group']: raise ConfigError(f"Invalid esp-group on {name} remote-access config") else: raise ConfigError(f"Missing esp-group on {name} remote-access config") if 'ike_group' in ra_conf: if 'ike_group' not in ipsec or ra_conf['ike_group'] not in ipsec['ike_group']: raise ConfigError(f"Invalid ike-group on {name} remote-access config") ike = ra_conf['ike_group'] if dict_search(f'ike_group.{ike}.key_exchange', ipsec) != 'ikev2': raise ConfigError('IPsec remote-access connections requires IKEv2!') else: raise ConfigError(f"Missing ike-group on {name} remote-access config") if 'authentication' not in ra_conf: raise ConfigError(f"Missing authentication on {name} remote-access config") if ra_conf['authentication']['server_mode'] == 'x509': if 'x509' not in ra_conf['authentication']: raise ConfigError(f"Missing x509 settings on {name} remote-access config") x509 = ra_conf['authentication']['x509'] if 'ca_certificate' not in x509 or 'certificate' not in x509: raise ConfigError(f"Missing x509 certificates on {name} remote-access config") verify_pki_x509(ipsec['pki'], x509) elif ra_conf['authentication']['server_mode'] == 'pre-shared-secret': if 'pre_shared_secret' not in ra_conf['authentication']: raise ConfigError(f"Missing pre-shared-key on {name} remote-access config") if 'client_mode' not in ra_conf['authentication']: raise ConfigError('Client authentication method is required!') if dict_search('authentication.client_mode', ra_conf) == 'eap-radius': if dict_search('remote_access.radius.server', ipsec) == None: raise ConfigError('RADIUS authentication requires at least one server') if 'pool' in ra_conf: if {'dhcp', 'radius'} <= set(ra_conf['pool']): raise ConfigError(f'Can not use both DHCP and RADIUS for address allocation '\ f'at the same time for "{name}"!') if 'dhcp' in ra_conf['pool'] and len(ra_conf['pool']) > 1: raise ConfigError(f'Can not use DHCP and a predefined address pool for "{name}"!') if 'radius' in ra_conf['pool'] and len(ra_conf['pool']) > 1: raise ConfigError(f'Can not use RADIUS and a predefined address pool for "{name}"!') for pool in ra_conf['pool']: if pool == 'dhcp': if dict_search('remote_access.dhcp.server', ipsec) == None: raise ConfigError('IPsec DHCP server is not configured!') elif pool == 'radius': if dict_search('remote_access.radius.server', ipsec) == None: raise ConfigError('IPsec RADIUS server is not configured!') if dict_search('authentication.client_mode', ra_conf) != 'eap-radius': raise ConfigError('RADIUS IP pool requires eap-radius client authentication!') elif 'pool' not in ipsec['remote_access'] or pool not in ipsec['remote_access']['pool']: raise ConfigError(f'Requested pool "{pool}" does not exist!') if 'pool' in ipsec['remote_access']: for pool, pool_config in ipsec['remote_access']['pool'].items(): if 'prefix' not in pool_config: raise ConfigError(f'Missing madatory prefix option for pool "{pool}"!') if 'name_server' in pool_config: if len(pool_config['name_server']) > 2: raise ConfigError(f'Only two name-servers are supported for remote-access pool "{pool}"!') for ns in pool_config['name_server']: v4_addr_and_ns = is_ipv4(ns) and not is_ipv4(pool_config['prefix']) v6_addr_and_ns = is_ipv6(ns) and not is_ipv6(pool_config['prefix']) if v4_addr_and_ns or v6_addr_and_ns: raise ConfigError('Must use both IPv4 or IPv6 addresses for pool prefix and name-server adresses!') if 'exclude' in pool_config: for exclude in pool_config['exclude']: v4_addr_and_exclude = is_ipv4(exclude) and not is_ipv4(pool_config['prefix']) v6_addr_and_exclude = is_ipv6(exclude) and not is_ipv6(pool_config['prefix']) if v4_addr_and_exclude or v6_addr_and_exclude: raise ConfigError('Must use both IPv4 or IPv6 addresses for pool prefix and exclude prefixes!') if 'radius' in ipsec['remote_access'] and 'server' in ipsec['remote_access']['radius']: for server, server_config in ipsec['remote_access']['radius']['server'].items(): if 'key' not in server_config: raise ConfigError(f'Missing RADIUS secret key for server "{server}"') if 'site_to_site' in ipsec and 'peer' in ipsec['site_to_site']: for peer, peer_conf in ipsec['site_to_site']['peer'].items(): has_default_esp = False # Peer name it is swanctl connection name and shouldn't contain dots or colons, T4118 if bool(re.search(':|\.', peer)): raise ConfigError(f'Incorrect peer name "{peer}" ' f'Peer name can contain alpha-numeric letters, hyphen and underscore') if 'remote_address' not in peer_conf: print(f'You should set correct remote-address "peer {peer} remote-address x.x.x.x"\n') if 'default_esp_group' in peer_conf: has_default_esp = True if 'esp_group' not in ipsec or peer_conf['default_esp_group'] not in ipsec['esp_group']: raise ConfigError(f"Invalid esp-group on site-to-site peer {peer}") if 'ike_group' in peer_conf: if 'ike_group' not in ipsec or peer_conf['ike_group'] not in ipsec['ike_group']: raise ConfigError(f"Invalid ike-group on site-to-site peer {peer}") else: raise ConfigError(f"Missing ike-group on site-to-site peer {peer}") if 'authentication' not in peer_conf or 'mode' not in peer_conf['authentication']: raise ConfigError(f"Missing authentication on site-to-site peer {peer}") if {'id', 'use_x509_id'} <= set(peer_conf['authentication']): raise ConfigError(f"Manually set peer id and use-x509-id are mutually exclusive!") if peer_conf['authentication']['mode'] == 'x509': if 'x509' not in peer_conf['authentication']: raise ConfigError(f"Missing x509 settings on site-to-site peer {peer}") x509 = peer_conf['authentication']['x509'] if 'ca_certificate' not in x509 or 'certificate' not in x509: raise ConfigError(f"Missing x509 certificates on site-to-site peer {peer}") verify_pki_x509(ipsec['pki'], x509) elif peer_conf['authentication']['mode'] == 'rsa': if 'rsa' not in peer_conf['authentication']: raise ConfigError(f"Missing RSA settings on site-to-site peer {peer}") rsa = peer_conf['authentication']['rsa'] if 'local_key' not in rsa: raise ConfigError(f"Missing RSA local-key on site-to-site peer {peer}") if 'remote_key' not in rsa: raise ConfigError(f"Missing RSA remote-key on site-to-site peer {peer}") verify_pki_rsa(ipsec['pki'], rsa) if 'local_address' not in peer_conf and 'dhcp_interface' not in peer_conf: raise ConfigError(f"Missing local-address or dhcp-interface on site-to-site peer {peer}") if 'dhcp_interface' in peer_conf: dhcp_interface = peer_conf['dhcp_interface'] verify_interface_exists(dhcp_interface) dhcp_base = directories['isc_dhclient_dir'] if not os.path.exists(f'{dhcp_base}/dhclient_{dhcp_interface}.conf'): raise ConfigError(f"Invalid dhcp-interface on site-to-site peer {peer}") address = get_dhcp_address(dhcp_interface) count = 0 while not address and count < dhcp_wait_attempts: address = get_dhcp_address(dhcp_interface) count += 1 sleep(dhcp_wait_sleep) if not address: ipsec['dhcp_no_address'][peer] = dhcp_interface print(f"Failed to get address from dhcp-interface on site-to-site peer {peer} -- skipped") continue if 'vti' in peer_conf: if 'local_address' in peer_conf and 'dhcp_interface' in peer_conf: raise ConfigError(f"A single local-address or dhcp-interface is required when using VTI on site-to-site peer {peer}") if dict_search('options.disable_route_autoinstall', ipsec) == None: Warning('It\'s recommended to use ipsec vti with the next command\n[set vpn ipsec option disable-route-autoinstall]') if 'bind' in peer_conf['vti']: vti_interface = peer_conf['vti']['bind'] if not os.path.exists(f'/sys/class/net/{vti_interface}'): raise ConfigError(f'VTI interface {vti_interface} for site-to-site peer {peer} does not exist!') if 'vti' not in peer_conf and 'tunnel' not in peer_conf: raise ConfigError(f"No VTI or tunnel specified on site-to-site peer {peer}") if 'tunnel' in peer_conf: for tunnel, tunnel_conf in peer_conf['tunnel'].items(): if 'esp_group' not in tunnel_conf and not has_default_esp: raise ConfigError(f"Missing esp-group on tunnel {tunnel} for site-to-site peer {peer}") esp_group_name = tunnel_conf['esp_group'] if 'esp_group' in tunnel_conf else peer_conf['default_esp_group'] if esp_group_name not in ipsec['esp_group']: raise ConfigError(f"Invalid esp-group on tunnel {tunnel} for site-to-site peer {peer}") esp_group = ipsec['esp_group'][esp_group_name] if 'mode' in esp_group and esp_group['mode'] == 'transport': if 'protocol' in tunnel_conf and ((peer in ['any', '0.0.0.0']) or ('local_address' not in peer_conf or peer_conf['local_address'] in ['any', '0.0.0.0'])): raise ConfigError(f"Fixed local-address or peer required when a protocol is defined with ESP transport mode on tunnel {tunnel} for site-to-site peer {peer}") if ('local' in tunnel_conf and 'prefix' in tunnel_conf['local']) or ('remote' in tunnel_conf and 'prefix' in tunnel_conf['remote']): raise ConfigError(f"Local/remote prefix cannot be used with ESP transport mode on tunnel {tunnel} for site-to-site peer {peer}") def cleanup_pki_files(): for path in [CERT_PATH, CA_PATH, CRL_PATH, KEY_PATH, PUBKEY_PATH]: if not os.path.exists(path): continue for file in os.listdir(path): file_path = os.path.join(path, file) if os.path.isfile(file_path): os.unlink(file_path) def generate_pki_files_x509(pki, x509_conf): ca_cert_name = x509_conf['ca_certificate'] ca_cert_data = dict_search_args(pki, 'ca', ca_cert_name, 'certificate') ca_cert_crls = dict_search_args(pki, 'ca', ca_cert_name, 'crl') or [] ca_index = 1 crl_index = 1 ca_cert = load_certificate(ca_cert_data) pki_ca_certs = [load_certificate(ca['certificate']) for ca in pki['ca'].values()] ca_cert_chain = find_chain(ca_cert, pki_ca_certs) cert_name = x509_conf['certificate'] cert_data = dict_search_args(pki, 'certificate', cert_name, 'certificate') key_data = dict_search_args(pki, 'certificate', cert_name, 'private', 'key') protected = 'passphrase' in x509_conf for ca_cert_obj in ca_cert_chain: with open(os.path.join(CA_PATH, f'{ca_cert_name}_{ca_index}.pem'), 'w') as f: f.write(encode_certificate(ca_cert_obj)) ca_index += 1 for crl in ca_cert_crls: with open(os.path.join(CRL_PATH, f'{ca_cert_name}_{crl_index}.pem'), 'w') as f: f.write(wrap_crl(crl)) crl_index += 1 with open(os.path.join(CERT_PATH, f'{cert_name}.pem'), 'w') as f: f.write(wrap_certificate(cert_data)) with open(os.path.join(KEY_PATH, f'x509_{cert_name}.pem'), 'w') as f: f.write(wrap_private_key(key_data, protected)) def generate_pki_files_rsa(pki, rsa_conf): local_key_name = rsa_conf['local_key'] local_key_data = dict_search_args(pki, 'key_pair', local_key_name, 'private', 'key') protected = 'passphrase' in rsa_conf remote_key_name = rsa_conf['remote_key'] remote_key_data = dict_search_args(pki, 'key_pair', remote_key_name, 'public', 'key') local_key = load_private_key(local_key_data, rsa_conf['passphrase'] if protected else None) with open(os.path.join(KEY_PATH, f'rsa_{local_key_name}.pem'), 'w') as f: f.write(wrap_private_key(local_key_data, protected)) with open(os.path.join(PUBKEY_PATH, f'{local_key_name}.pem'), 'w') as f: f.write(encode_public_key(local_key.public_key())) with open(os.path.join(PUBKEY_PATH, f'{remote_key_name}.pem'), 'w') as f: f.write(wrap_public_key(remote_key_data)) def generate(ipsec): cleanup_pki_files() if not ipsec: for config_file in [charon_dhcp_conf, charon_radius_conf, interface_conf, swanctl_conf]: if os.path.isfile(config_file): os.unlink(config_file) render(charon_conf, 'ipsec/charon.j2', {'install_routes': default_install_routes}) return if ipsec['dhcp_no_address']: with open(DHCP_HOOK_IFLIST, 'w') as f: f.write(" ".join(ipsec['dhcp_no_address'].values())) elif os.path.exists(DHCP_HOOK_IFLIST): os.unlink(DHCP_HOOK_IFLIST) for path in [swanctl_dir, CERT_PATH, CA_PATH, CRL_PATH, PUBKEY_PATH]: if not os.path.exists(path): os.mkdir(path, mode=0o755) if not os.path.exists(KEY_PATH): os.mkdir(KEY_PATH, mode=0o700) if 'l2tp' in ipsec: if 'authentication' in ipsec['l2tp'] and 'x509' in ipsec['l2tp']['authentication']: generate_pki_files_x509(ipsec['pki'], ipsec['l2tp']['authentication']['x509']) if 'remote_access' in ipsec and 'connection' in ipsec['remote_access']: for rw, rw_conf in ipsec['remote_access']['connection'].items(): if 'authentication' in rw_conf and 'x509' in rw_conf['authentication']: generate_pki_files_x509(ipsec['pki'], rw_conf['authentication']['x509']) if 'site_to_site' in ipsec and 'peer' in ipsec['site_to_site']: for peer, peer_conf in ipsec['site_to_site']['peer'].items(): if peer in ipsec['dhcp_no_address']: continue if peer_conf['authentication']['mode'] == 'x509': generate_pki_files_x509(ipsec['pki'], peer_conf['authentication']['x509']) elif peer_conf['authentication']['mode'] == 'rsa': generate_pki_files_rsa(ipsec['pki'], peer_conf['authentication']['rsa']) local_ip = '' if 'local_address' in peer_conf: local_ip = peer_conf['local_address'] elif 'dhcp_interface' in peer_conf: local_ip = get_dhcp_address(peer_conf['dhcp_interface']) ipsec['site_to_site']['peer'][peer]['local_address'] = local_ip if 'tunnel' in peer_conf: for tunnel, tunnel_conf in peer_conf['tunnel'].items(): local_prefixes = dict_search_args(tunnel_conf, 'local', 'prefix') remote_prefixes = dict_search_args(tunnel_conf, 'remote', 'prefix') if not local_prefixes or not remote_prefixes: continue passthrough = None for local_prefix in local_prefixes: for remote_prefix in remote_prefixes: local_net = ipaddress.ip_network(local_prefix) remote_net = ipaddress.ip_network(remote_prefix) if local_net.overlaps(remote_net): if passthrough is None: passthrough = [] passthrough.append(local_prefix) ipsec['site_to_site']['peer'][peer]['tunnel'][tunnel]['passthrough'] = passthrough # auth psk <tag> dhcp-interface <xxx> if jmespath.search('authentication.psk.*.dhcp_interface', ipsec): for psk, psk_config in ipsec['authentication']['psk'].items(): if 'dhcp_interface' in psk_config: for iface in psk_config['dhcp_interface']: id = get_dhcp_address(iface) if id: ipsec['authentication']['psk'][psk]['id'].append(id) render(charon_conf, 'ipsec/charon.j2', ipsec) render(charon_dhcp_conf, 'ipsec/charon/dhcp.conf.j2', ipsec) render(charon_radius_conf, 'ipsec/charon/eap-radius.conf.j2', ipsec) render(interface_conf, 'ipsec/interfaces_use.conf.j2', ipsec) render(swanctl_conf, 'ipsec/swanctl.conf.j2', ipsec) def resync_nhrp(ipsec): if ipsec and not ipsec['nhrp_exists']: return tmp = run('/usr/libexec/vyos/conf_mode/protocols_nhrp.py') if tmp > 0: print('ERROR: failed to reapply NHRP settings!') def apply(ipsec): systemd_service = 'strongswan.service' if not ipsec: call(f'systemctl stop {systemd_service}') else: call(f'systemctl reload-or-restart {systemd_service}') resync_nhrp(ipsec) if __name__ == '__main__': try: ipsec = get_config() verify(ipsec) generate(ipsec) apply(ipsec) except ConfigError as e: print(e) exit(1) diff --git a/src/conf_mode/vpn_openconnect.py b/src/conf_mode/vpn_openconnect.py index a039172c4..421ac6997 100755 --- a/src/conf_mode/vpn_openconnect.py +++ b/src/conf_mode/vpn_openconnect.py @@ -1,296 +1,292 @@ #!/usr/bin/env python3 # # Copyright (C) 2018-2023 VyOS maintainers and contributors # # This program is free software; you can redistribute it and/or modify # it under the terms of the GNU General Public License version 2 or later as # published by the Free Software Foundation. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program. If not, see <http://www.gnu.org/licenses/>. import os from sys import exit from vyos.base import Warning from vyos.config import Config from vyos.pki import wrap_certificate from vyos.pki import wrap_private_key from vyos.template import render from vyos.utils.process import call from vyos.utils.network import check_port_availability from vyos.utils.process import is_systemd_service_running from vyos.utils.network import is_listen_port_bind_service from vyos.utils.dict import dict_search from vyos import ConfigError from passlib.hash import sha512_crypt from time import sleep from vyos import airbag airbag.enable() cfg_dir = '/run/ocserv' ocserv_conf = cfg_dir + '/ocserv.conf' ocserv_passwd = cfg_dir + '/ocpasswd' ocserv_otp_usr = cfg_dir + '/users.oath' radius_cfg = cfg_dir + '/radiusclient.conf' radius_servers = cfg_dir + '/radius_servers' # Generate hash from user cleartext password def get_hash(password): return sha512_crypt.hash(password) def get_config(config=None): if config: conf = config else: conf = Config() base = ['vpn', 'openconnect'] if not conf.exists(base): return None ocserv = conf.get_config_dict(base, key_mangling=('-', '_'), get_first_key=True, - with_recursive_defaults=True) - - if ocserv: - ocserv['pki'] = conf.get_config_dict(['pki'], key_mangling=('-', '_'), - no_tag_node_value_mangle=True, - get_first_key=True) + with_recursive_defaults=True, + with_pki=True) return ocserv def verify(ocserv): if ocserv is None: return None # Check if listen-ports not binded other services # It can be only listen by 'ocserv-main' for proto, port in ocserv.get('listen_ports').items(): if check_port_availability(ocserv['listen_address'], int(port), proto) is not True and \ not is_listen_port_bind_service(int(port), 'ocserv-main'): raise ConfigError(f'"{proto}" port "{port}" is used by another service') # Check accounting if "accounting" in ocserv: if "mode" in ocserv["accounting"] and "radius" in ocserv["accounting"]["mode"]: if not origin["accounting"]['radius']['server']: raise ConfigError('Openconnect accounting mode radius requires at least one RADIUS server') if "authentication" not in ocserv or "mode" not in ocserv["authentication"]: raise ConfigError('Accounting depends on OpenConnect authentication configuration') elif "radius" not in ocserv["authentication"]["mode"]: raise ConfigError('RADIUS accounting must be used with RADIUS authentication') # Check authentication if "authentication" in ocserv: if "mode" in ocserv["authentication"]: if ("local" in ocserv["authentication"]["mode"] and "radius" in ocserv["authentication"]["mode"]): raise ConfigError('OpenConnect authentication modes are mutually-exclusive, remove either local or radius from your configuration') if "radius" in ocserv["authentication"]["mode"]: if not ocserv["authentication"]['radius']['server']: raise ConfigError('Openconnect authentication mode radius requires at least one RADIUS server') if "local" in ocserv["authentication"]["mode"]: if not ocserv["authentication"]["local_users"]: raise ConfigError('openconnect mode local required at least one user') if not ocserv["authentication"]["local_users"]["username"]: raise ConfigError('openconnect mode local required at least one user') else: # For OTP mode: verify that each local user has an OTP key if "otp" in ocserv["authentication"]["mode"]["local"]: users_wo_key = [] for user, user_config in ocserv["authentication"]["local_users"]["username"].items(): # User has no OTP key defined if dict_search('otp.key', user_config) == None: users_wo_key.append(user) if users_wo_key: raise ConfigError(f'OTP enabled, but no OTP key is configured for these users:\n{users_wo_key}') # For password (and default) mode: verify that each local user has password if "password" in ocserv["authentication"]["mode"]["local"] or "otp" not in ocserv["authentication"]["mode"]["local"]: users_wo_pswd = [] for user in ocserv["authentication"]["local_users"]["username"]: if not "password" in ocserv["authentication"]["local_users"]["username"][user]: users_wo_pswd.append(user) if users_wo_pswd: raise ConfigError(f'password required for users:\n{users_wo_pswd}') # Validate that if identity-based-config is configured all child config nodes are set if 'identity_based_config' in ocserv["authentication"]: if 'disabled' not in ocserv["authentication"]["identity_based_config"]: Warning("Identity based configuration files is a 3rd party addition. Use at your own risk, this might break the ocserv daemon!") if 'mode' not in ocserv["authentication"]["identity_based_config"]: raise ConfigError('OpenConnect radius identity-based-config enabled but mode not selected') elif 'group' in ocserv["authentication"]["identity_based_config"]["mode"] and "radius" not in ocserv["authentication"]["mode"]: raise ConfigError('OpenConnect config-per-group must be used with radius authentication') if 'directory' not in ocserv["authentication"]["identity_based_config"]: raise ConfigError('OpenConnect identity-based-config enabled but directory not set') if 'default_config' not in ocserv["authentication"]["identity_based_config"]: raise ConfigError('OpenConnect identity-based-config enabled but default-config not set') else: raise ConfigError('openconnect authentication mode required') else: raise ConfigError('openconnect authentication credentials required') # Check ssl if 'ssl' not in ocserv: raise ConfigError('openconnect ssl required') if not ocserv['pki'] or 'certificate' not in ocserv['pki']: raise ConfigError('PKI not configured') ssl = ocserv['ssl'] if 'certificate' not in ssl: raise ConfigError('openconnect ssl certificate required') cert_name = ssl['certificate'] if cert_name not in ocserv['pki']['certificate']: raise ConfigError('Invalid openconnect ssl certificate') cert = ocserv['pki']['certificate'][cert_name] if 'certificate' not in cert: raise ConfigError('Missing certificate in PKI') if 'private' not in cert or 'key' not in cert['private']: raise ConfigError('Missing private key in PKI') if 'ca_certificate' in ssl: if 'ca' not in ocserv['pki']: raise ConfigError('PKI not configured') if ssl['ca_certificate'] not in ocserv['pki']['ca']: raise ConfigError('Invalid openconnect ssl CA certificate') # Check network settings if "network_settings" in ocserv: if "push_route" in ocserv["network_settings"]: # Replace default route if "0.0.0.0/0" in ocserv["network_settings"]["push_route"]: ocserv["network_settings"]["push_route"].remove("0.0.0.0/0") ocserv["network_settings"]["push_route"].append("default") else: ocserv["network_settings"]["push_route"] = ["default"] else: raise ConfigError('openconnect network settings required') def generate(ocserv): if not ocserv: return None if "radius" in ocserv["authentication"]["mode"]: if dict_search(ocserv, 'accounting.mode.radius'): # Render radius client configuration render(radius_cfg, 'ocserv/radius_conf.j2', ocserv) merged_servers = ocserv["accounting"]["radius"]["server"] | ocserv["authentication"]["radius"]["server"] # Render radius servers # Merge the accounting and authentication servers into a single dictionary render(radius_servers, 'ocserv/radius_servers.j2', {'server': merged_servers}) else: # Render radius client configuration render(radius_cfg, 'ocserv/radius_conf.j2', ocserv) # Render radius servers render(radius_servers, 'ocserv/radius_servers.j2', ocserv["authentication"]["radius"]) elif "local" in ocserv["authentication"]["mode"]: # if mode "OTP", generate OTP users file parameters if "otp" in ocserv["authentication"]["mode"]["local"]: if "local_users" in ocserv["authentication"]: for user in ocserv["authentication"]["local_users"]["username"]: # OTP token type from CLI parameters: otp_interval = str(ocserv["authentication"]["local_users"]["username"][user]["otp"].get("interval")) token_type = ocserv["authentication"]["local_users"]["username"][user]["otp"].get("token_type") otp_length = str(ocserv["authentication"]["local_users"]["username"][user]["otp"].get("otp_length")) if token_type == "hotp-time": otp_type = "HOTP/T" + otp_interval elif token_type == "hotp-event": otp_type = "HOTP/E" else: otp_type = "HOTP/T" + otp_interval ocserv["authentication"]["local_users"]["username"][user]["otp"]["token_tmpl"] = otp_type + "/" + otp_length # if there is a password, generate hash if "password" in ocserv["authentication"]["mode"]["local"] or not "otp" in ocserv["authentication"]["mode"]["local"]: if "local_users" in ocserv["authentication"]: for user in ocserv["authentication"]["local_users"]["username"]: ocserv["authentication"]["local_users"]["username"][user]["hash"] = get_hash(ocserv["authentication"]["local_users"]["username"][user]["password"]) if "password-otp" in ocserv["authentication"]["mode"]["local"]: # Render local users ocpasswd render(ocserv_passwd, 'ocserv/ocserv_passwd.j2', ocserv["authentication"]["local_users"]) # Render local users OTP keys render(ocserv_otp_usr, 'ocserv/ocserv_otp_usr.j2', ocserv["authentication"]["local_users"]) elif "password" in ocserv["authentication"]["mode"]["local"]: # Render local users ocpasswd render(ocserv_passwd, 'ocserv/ocserv_passwd.j2', ocserv["authentication"]["local_users"]) elif "otp" in ocserv["authentication"]["mode"]["local"]: # Render local users OTP keys render(ocserv_otp_usr, 'ocserv/ocserv_otp_usr.j2', ocserv["authentication"]["local_users"]) else: # Render local users ocpasswd render(ocserv_passwd, 'ocserv/ocserv_passwd.j2', ocserv["authentication"]["local_users"]) else: if "local_users" in ocserv["authentication"]: for user in ocserv["authentication"]["local_users"]["username"]: ocserv["authentication"]["local_users"]["username"][user]["hash"] = get_hash(ocserv["authentication"]["local_users"]["username"][user]["password"]) # Render local users render(ocserv_passwd, 'ocserv/ocserv_passwd.j2', ocserv["authentication"]["local_users"]) if "ssl" in ocserv: cert_file_path = os.path.join(cfg_dir, 'cert.pem') cert_key_path = os.path.join(cfg_dir, 'cert.key') ca_cert_file_path = os.path.join(cfg_dir, 'ca.pem') if 'certificate' in ocserv['ssl']: cert_name = ocserv['ssl']['certificate'] pki_cert = ocserv['pki']['certificate'][cert_name] with open(cert_file_path, 'w') as f: f.write(wrap_certificate(pki_cert['certificate'])) if 'private' in pki_cert and 'key' in pki_cert['private']: with open(cert_key_path, 'w') as f: f.write(wrap_private_key(pki_cert['private']['key'])) if 'ca_certificate' in ocserv['ssl']: ca_name = ocserv['ssl']['ca_certificate'] pki_ca_cert = ocserv['pki']['ca'][ca_name] with open(ca_cert_file_path, 'w') as f: f.write(wrap_certificate(pki_ca_cert['certificate'])) # Render config render(ocserv_conf, 'ocserv/ocserv_config.j2', ocserv) def apply(ocserv): if not ocserv: call('systemctl stop ocserv.service') for file in [ocserv_conf, ocserv_passwd, ocserv_otp_usr]: if os.path.exists(file): os.unlink(file) else: call('systemctl reload-or-restart ocserv.service') counter = 0 while True: # exit early when service runs if is_systemd_service_running("ocserv.service"): break sleep(0.250) if counter > 5: raise ConfigError('openconnect failed to start, check the logs for details') break counter += 1 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/vpn_sstp.py b/src/conf_mode/vpn_sstp.py index ac053cc76..6bf9307e1 100755 --- a/src/conf_mode/vpn_sstp.py +++ b/src/conf_mode/vpn_sstp.py @@ -1,173 +1,170 @@ #!/usr/bin/env python3 # -# Copyright (C) 2018-2023 VyOS maintainers and contributors +# Copyright (C) 2018-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 vyos.config import Config from vyos.configdict import get_accel_dict from vyos.configdict import dict_merge from vyos.pki import wrap_certificate from vyos.pki import wrap_private_key from vyos.template import render from vyos.utils.process import call from vyos.utils.network import check_port_availability from vyos.utils.dict import dict_search from vyos.accel_ppp_util import verify_accel_ppp_base_service from vyos.accel_ppp_util import verify_accel_ppp_ip_pool from vyos.accel_ppp_util import get_pools_in_order from vyos.utils.network import is_listen_port_bind_service from vyos.utils.file import write_file from vyos import ConfigError from vyos import airbag airbag.enable() cfg_dir = '/run/accel-pppd' sstp_conf = '/run/accel-pppd/sstp.conf' sstp_chap_secrets = '/run/accel-pppd/sstp.chap-secrets' cert_file_path = os.path.join(cfg_dir, 'sstp-cert.pem') cert_key_path = os.path.join(cfg_dir, 'sstp-cert.key') ca_cert_file_path = os.path.join(cfg_dir, 'sstp-ca.pem') def get_config(config=None): if config: conf = config else: conf = Config() base = ['vpn', 'sstp'] if not conf.exists(base): return None # retrieve common dictionary keys - sstp = get_accel_dict(conf, base, sstp_chap_secrets) + sstp = get_accel_dict(conf, base, sstp_chap_secrets, with_pki=True) if dict_search('client_ip_pool', sstp): # Multiple named pools require ordered values T5099 sstp['ordered_named_pools'] = get_pools_in_order(dict_search('client_ip_pool', sstp)) - if sstp: - sstp['pki'] = conf.get_config_dict(['pki'], key_mangling=('-', '_'), - get_first_key=True, - no_tag_node_value_mangle=True) + sstp['server_type'] = 'sstp' return sstp def verify(sstp): if not sstp: return None port = sstp.get('port') proto = 'tcp' if check_port_availability('0.0.0.0', int(port), proto) is not True and \ not is_listen_port_bind_service(int(port), 'accel-pppd'): raise ConfigError(f'"{proto}" port "{port}" is used by another service') verify_accel_ppp_base_service(sstp) if 'client_ip_pool' not in sstp and 'client_ipv6_pool' not in sstp: raise ConfigError('Client IP subnet required') verify_accel_ppp_ip_pool(sstp) # # SSL certificate checks # if not sstp['pki']: raise ConfigError('PKI is not configured') if 'ssl' not in sstp: raise ConfigError('SSL missing on SSTP config') ssl = sstp['ssl'] # CA if 'ca_certificate' not in ssl: raise ConfigError('SSL CA certificate missing on SSTP config') ca_name = ssl['ca_certificate'] if ca_name not in sstp['pki']['ca']: raise ConfigError('Invalid CA certificate on SSTP config') if 'certificate' not in sstp['pki']['ca'][ca_name]: raise ConfigError('Missing certificate data for CA certificate on SSTP config') # Certificate if 'certificate' not in ssl: raise ConfigError('SSL certificate missing on SSTP config') cert_name = ssl['certificate'] if cert_name not in sstp['pki']['certificate']: raise ConfigError('Invalid certificate on SSTP config') pki_cert = sstp['pki']['certificate'][cert_name] if 'certificate' not in pki_cert: raise ConfigError('Missing certificate data for certificate on SSTP config') if 'private' not in pki_cert or 'key' not in pki_cert['private']: raise ConfigError('Missing private key for certificate on SSTP config') if 'password_protected' in pki_cert['private']: raise ConfigError('Encrypted private key is not supported on SSTP config') def generate(sstp): if not sstp: return None # accel-cmd reload doesn't work so any change results in a restart of the daemon render(sstp_conf, 'accel-ppp/sstp.config.j2', sstp) cert_name = sstp['ssl']['certificate'] pki_cert = sstp['pki']['certificate'][cert_name] ca_cert_name = sstp['ssl']['ca_certificate'] pki_ca = sstp['pki']['ca'][ca_cert_name] write_file(cert_file_path, wrap_certificate(pki_cert['certificate'])) write_file(cert_key_path, wrap_private_key(pki_cert['private']['key'])) write_file(ca_cert_file_path, wrap_certificate(pki_ca['certificate'])) if dict_search('authentication.mode', sstp) == 'local': render(sstp_chap_secrets, 'accel-ppp/chap-secrets.config_dict.j2', sstp, permission=0o640) else: if os.path.exists(sstp_chap_secrets): os.unlink(sstp_chap_secrets) return sstp def apply(sstp): if not sstp: call('systemctl stop accel-ppp@sstp.service') for file in [sstp_chap_secrets, sstp_conf]: if os.path.exists(file): os.unlink(file) return None call('systemctl restart accel-ppp@sstp.service') if __name__ == '__main__': try: c = get_config() verify(c) generate(c) apply(c) except ConfigError as e: print(e) exit(1)