diff options
Diffstat (limited to 'src')
| -rwxr-xr-x | src/conf_mode/interfaces_tunnel.py | 12 | ||||
| -rwxr-xr-x | src/conf_mode/protocols_nhrp.py | 118 | ||||
| -rwxr-xr-x | src/conf_mode/vpn_ipsec.py | 27 | ||||
| -rw-r--r-- | src/migration-scripts/nhrp/0-to-1 | 128 | ||||
| -rwxr-xr-x | src/op_mode/dhcp.py | 2 | ||||
| -rwxr-xr-x | src/op_mode/ipsec.py | 23 | ||||
| -rwxr-xr-x | src/op_mode/nhrp.py | 101 | ||||
| -rwxr-xr-x | src/op_mode/vtysh_wrapper.sh | 2 | ||||
| -rw-r--r-- | src/tests/test_configd_inspect.py | 210 |
9 files changed, 356 insertions, 267 deletions
diff --git a/src/conf_mode/interfaces_tunnel.py b/src/conf_mode/interfaces_tunnel.py index 98ef98d12..ee1436e49 100755 --- a/src/conf_mode/interfaces_tunnel.py +++ b/src/conf_mode/interfaces_tunnel.py @@ -1,6 +1,6 @@ #!/usr/bin/env python3 # -# Copyright (C) 2018-2024 yOS maintainers and contributors +# Copyright (C) 2018-2025 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 @@ -13,9 +13,8 @@ # # You should have received a copy of the GNU General Public License # along with this program. If not, see <http://www.gnu.org/licenses/>. - from sys import exit - +import ipaddress from vyos.config import Config from vyos.configdict import get_interface_dict from vyos.configdict import is_node_changed @@ -89,6 +88,13 @@ def verify(tunnel): raise ConfigError('Tunnel used for NHRP, it can not be deleted!') return None + if 'nhrp' in tunnel: + if 'address' in tunnel: + address_list = dict_search('address', tunnel) + for tunip in address_list: + if ipaddress.ip_network(tunip, strict=False).prefixlen != 32: + raise ConfigError( + 'Tunnel is used for NHRP, Netmask should be /32!') verify_tunnel(tunnel) diff --git a/src/conf_mode/protocols_nhrp.py b/src/conf_mode/protocols_nhrp.py index 0bd68b7d8..ac92c9d99 100755 --- a/src/conf_mode/protocols_nhrp.py +++ b/src/conf_mode/protocols_nhrp.py @@ -1,6 +1,6 @@ #!/usr/bin/env python3 # -# Copyright (C) 2021-2024 VyOS maintainers and contributors +# Copyright (C) 2021-2025 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 @@ -14,95 +14,112 @@ # 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 sys import argv +import ipaddress from vyos.config import Config -from vyos.configdict import node_changed from vyos.template import render +from vyos.configverify import has_frr_protocol_in_dict from vyos.utils.process import run +from vyos.utils.dict import dict_search from vyos import ConfigError from vyos import airbag +from vyos.frrender import FRRender +from vyos.frrender import get_frrender_dict +from vyos.utils.process import is_systemd_service_running + airbag.enable() -opennhrp_conf = '/run/opennhrp/opennhrp.conf' +nflog_redirect = 1 +nflog_multicast = 2 nhrp_nftables_conf = '/run/nftables_nhrp.conf' + def get_config(config=None): if config: conf = config else: conf = Config() - base = ['protocols', 'nhrp'] - - nhrp = conf.get_config_dict(base, key_mangling=('-', '_'), - get_first_key=True, no_tag_node_value_mangle=True) - nhrp['del_tunnels'] = node_changed(conf, base + ['tunnel']) - - if not conf.exists(base): - return nhrp - nhrp['if_tunnel'] = conf.get_config_dict(['interfaces', 'tunnel'], key_mangling=('-', '_'), - get_first_key=True, no_tag_node_value_mangle=True) + return get_frrender_dict(conf, argv) - nhrp['profile_map'] = {} - profile = conf.get_config_dict(['vpn', 'ipsec', 'profile'], key_mangling=('-', '_'), - get_first_key=True, no_tag_node_value_mangle=True) - for name, profile_conf in profile.items(): - if 'bind' in profile_conf and 'tunnel' in profile_conf['bind']: - interfaces = profile_conf['bind']['tunnel'] - if isinstance(interfaces, str): - interfaces = [interfaces] - for interface in interfaces: - nhrp['profile_map'][interface] = name - - return nhrp - -def verify(nhrp): - if 'tunnel' in nhrp: - for name, nhrp_conf in nhrp['tunnel'].items(): - if not nhrp['if_tunnel'] or name not in nhrp['if_tunnel']: +def verify(config_dict): + if not config_dict or 'deleted' in config_dict: + return None + if 'tunnel' in config_dict: + for name, nhrp_conf in config_dict['tunnel'].items(): + if not config_dict['if_tunnel'] or name not in config_dict['if_tunnel']: raise ConfigError(f'Tunnel interface "{name}" does not exist') - tunnel_conf = nhrp['if_tunnel'][name] + tunnel_conf = config_dict['if_tunnel'][name] + if 'address' in tunnel_conf: + address_list = dict_search('address', tunnel_conf) + for tunip in address_list: + if ipaddress.ip_network(tunip, + strict=False).prefixlen != 32: + raise ConfigError( + f'Tunnel {name} is used for NHRP, Netmask should be /32!') if 'encapsulation' not in tunnel_conf or tunnel_conf['encapsulation'] != 'gre': raise ConfigError(f'Tunnel "{name}" is not an mGRE tunnel') + if 'network_id' not in nhrp_conf: + raise ConfigError(f'network-id is not specified in tunnel "{name}"') + if 'remote' in tunnel_conf: raise ConfigError(f'Tunnel "{name}" cannot have a remote address defined') - if 'map' in nhrp_conf: - for map_name, map_conf in nhrp_conf['map'].items(): - if 'nbma_address' not in map_conf: + map_tunnelip = dict_search('map.tunnel_ip', nhrp_conf) + if map_tunnelip: + for map_name, map_conf in map_tunnelip.items(): + if 'nbma' not in map_conf: raise ConfigError(f'nbma-address missing on map {map_name} on tunnel {name}') - if 'dynamic_map' in nhrp_conf: - for map_name, map_conf in nhrp_conf['dynamic_map'].items(): - if 'nbma_domain_name' not in map_conf: - raise ConfigError(f'nbma-domain-name missing on dynamic-map {map_name} on tunnel {name}') + nhs_tunnelip = dict_search('nhs.tunnel_ip', nhrp_conf) + nbma_list = [] + if nhs_tunnelip: + for nhs_name, nhs_conf in nhs_tunnelip.items(): + if 'nbma' not in nhs_conf: + raise ConfigError(f'nbma-address missing on map nhs {nhs_name} on tunnel {name}') + if nhs_name != 'dynamic': + if len(list(dict_search('nbma', nhs_conf))) > 1: + raise ConfigError( + f'Static nhs tunnel-ip {nhs_name} cannot contain multiple nbma-addresses') + for nbma_ip in dict_search('nbma', nhs_conf): + if nbma_ip not in nbma_list: + nbma_list.append(nbma_ip) + else: + raise ConfigError( + f'Nbma address {nbma_ip} cannot be maped to several tunnel-ip') return None -def generate(nhrp): - if not os.path.exists(nhrp_nftables_conf): - nhrp['first_install'] = True - render(opennhrp_conf, 'nhrp/opennhrp.conf.j2', nhrp) - render(nhrp_nftables_conf, 'nhrp/nftables.conf.j2', nhrp) +def generate(config_dict): + if not has_frr_protocol_in_dict(config_dict, 'nhrp'): + return None + + if 'deleted' in config_dict['nhrp']: + return None + render(nhrp_nftables_conf, 'frr/nhrpd_nftables.conf.j2', config_dict['nhrp']) + + if config_dict and not is_systemd_service_running('vyos-configd.service'): + FRRender().generate(config_dict) return None -def apply(nhrp): + +def apply(config_dict): + nft_rc = run(f'nft --file {nhrp_nftables_conf}') if nft_rc != 0: raise ConfigError('Failed to apply NHRP tunnel firewall rules') - action = 'restart' if nhrp and 'tunnel' in nhrp else 'stop' - service_rc = run(f'systemctl {action} opennhrp.service') - if service_rc != 0: - raise ConfigError(f'Failed to {action} the NHRP service') - + if config_dict and not is_systemd_service_running('vyos-configd.service'): + FRRender().apply() return None + if __name__ == '__main__': try: c = get_config() @@ -112,3 +129,4 @@ if __name__ == '__main__': except ConfigError as e: print(e) exit(1) + diff --git a/src/conf_mode/vpn_ipsec.py b/src/conf_mode/vpn_ipsec.py index e22b7550c..25604d2a2 100755 --- a/src/conf_mode/vpn_ipsec.py +++ b/src/conf_mode/vpn_ipsec.py @@ -1,6 +1,6 @@ #!/usr/bin/env python3 # -# Copyright (C) 2021-2024 VyOS maintainers and contributors +# Copyright (C) 2021-2025 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 @@ -86,8 +86,6 @@ def get_config(config=None): 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=('-', '_'), @@ -95,6 +93,14 @@ def get_config(config=None): get_first_key=True, with_pki=True) + ipsec['nhrp_exists'] = conf.exists(['protocols', 'nhrp', 'tunnel']) + if ipsec['nhrp_exists']: + set_dependents('nhrp', conf) + + if not conf.exists(base): + ipsec.update({'deleted' : ''}) + return ipsec + # We have to cleanup the default dict, as default values could # enable features which are not explicitly enabled on the # CLI. E.g. dead-peer-detection defaults should not be injected @@ -115,7 +121,6 @@ def get_config(config=None): 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']) if ipsec['nhrp_exists']: set_dependents('nhrp', conf) @@ -196,8 +201,8 @@ def verify_pki_rsa(pki, rsa_conf): return True def verify(ipsec): - if not ipsec: - return None + if not ipsec or 'deleted' in ipsec: + return if 'authentication' in ipsec: if 'psk' in ipsec['authentication']: @@ -624,7 +629,7 @@ def generate_pki_files_rsa(pki, rsa_conf): def generate(ipsec): cleanup_pki_files() - if not ipsec: + if not ipsec or 'deleted' in 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) @@ -721,15 +726,12 @@ def generate(ipsec): def apply(ipsec): systemd_service = 'strongswan.service' - if not ipsec: + if not ipsec or 'deleted' in ipsec: call(f'systemctl stop {systemd_service}') - if vti_updown_db_exists(): remove_vti_updown_db() - else: call(f'systemctl reload-or-restart {systemd_service}') - if ipsec['enabled_vti_interfaces']: with open_vti_updown_db_for_create_or_update() as db: db.removeAllOtherInterfaces(ipsec['enabled_vti_interfaces']) @@ -737,7 +739,7 @@ def apply(ipsec): db.commit(lambda interface: ipsec['vti_interface_dicts'][interface]) elif vti_updown_db_exists(): remove_vti_updown_db() - + if ipsec: if ipsec.get('nhrp_exists', False): try: call_dependents() @@ -746,7 +748,6 @@ def apply(ipsec): # ConfigError("ConfigError('Interface ethN requires an IP address!')") pass - if __name__ == '__main__': try: ipsec = get_config() diff --git a/src/migration-scripts/nhrp/0-to-1 b/src/migration-scripts/nhrp/0-to-1 new file mode 100644 index 000000000..249010def --- /dev/null +++ b/src/migration-scripts/nhrp/0-to-1 @@ -0,0 +1,128 @@ +# Copyright 2025 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/>. + +# Migration from Opennhrp to FRR NHRP +import ipaddress + +from vyos.configtree import ConfigTree + +base = ['protocols', 'nhrp', 'tunnel'] +interface_base = ['interfaces', 'tunnel'] + +def migrate(config: ConfigTree) -> None: + if not config.exists(base): + return + + for tunnel_name in config.list_nodes(base): + ## Cisco Authentication migration + if config.exists(base + [tunnel_name,'cisco-authentication']): + auth = config.return_value(base + [tunnel_name,'cisco-authentication']) + config.delete(base + [tunnel_name,'cisco-authentication']) + config.set(base + [tunnel_name,'authentication'], value=auth) + ## Delete Dynamic-map to fqdn + if config.exists(base + [tunnel_name,'dynamic-map']): + config.delete(base + [tunnel_name,'dynamic-map']) + ## Holdtime migration + if config.exists(base + [tunnel_name,'holding-time']): + holdtime = config.return_value(base + [tunnel_name,'holding-time']) + config.delete(base + [tunnel_name,'holding-time']) + config.set(base + [tunnel_name,'holdtime'], value=holdtime) + ## Add network-id + config.set(base + [tunnel_name, 'network-id'], value='1') + ## Map and nhs migration + nhs_tunnelip_list = [] + nhs_nbmaip_list = [] + is_nhs = False + if config.exists(base + [tunnel_name,'map']): + is_map = False + for tunnel_ip in config.list_nodes(base + [tunnel_name, 'map']): + tunnel_ip_path = base + [tunnel_name, 'map', tunnel_ip] + tunnel_ip = tunnel_ip.split('/')[0] + if config.exists(tunnel_ip_path + ['cisco']): + config.delete(tunnel_ip_path + ['cisco']) + if config.exists(tunnel_ip_path + ['nbma-address']): + nbma = config.return_value(tunnel_ip_path + ['nbma-address']) + if config.exists (tunnel_ip_path + ['register']): + config.delete(tunnel_ip_path + ['register']) + config.delete(tunnel_ip_path + ['nbma-address']) + config.set(base + [tunnel_name, 'nhs', 'tunnel-ip', tunnel_ip, 'nbma'], value=nbma) + is_nhs = True + if tunnel_ip not in nhs_tunnelip_list: + nhs_tunnelip_list.append(tunnel_ip) + if nbma not in nhs_nbmaip_list: + nhs_nbmaip_list.append(nbma) + else: + config.delete(tunnel_ip_path + ['nbma-address']) + config.set(base + [tunnel_name, 'map_test', 'tunnel-ip', tunnel_ip, 'nbma'], value=nbma) + is_map = True + config.delete(base + [tunnel_name,'map']) + + if is_nhs: + config.set_tag(base + [tunnel_name, 'nhs', 'tunnel-ip']) + + if is_map: + config.copy(base + [tunnel_name, 'map_test'], base + [tunnel_name, 'map']) + config.delete(base + [tunnel_name, 'map_test']) + config.set_tag(base + [tunnel_name, 'map', 'tunnel-ip']) + + # + # Change netmask to /32 on tunnel interface + # If nhs is alone, add static route tunnel network to nhs + # + if config.exists(interface_base + [tunnel_name, 'address']): + tunnel_ip_list = [] + for tunnel_ip in config.return_values( + interface_base + [tunnel_name, 'address']): + tunnel_ip_ch = tunnel_ip.split('/')[0]+'/32' + if tunnel_ip_ch not in tunnel_ip_list: + tunnel_ip_list.append(tunnel_ip_ch) + for nhs in nhs_tunnelip_list: + config.set(['protocols', 'static', 'route', str(ipaddress.ip_network(tunnel_ip, strict=False)), 'next-hop', nhs, 'distance'], value='250') + if nhs_tunnelip_list: + if not config.is_tag(['protocols', 'static', 'route']): + config.set_tag(['protocols', 'static', 'route']) + if not config.is_tag(['protocols', 'static', 'route', str(ipaddress.ip_network(tunnel_ip, strict=False)), 'next-hop']): + config.set_tag(['protocols', 'static', 'route', str(ipaddress.ip_network(tunnel_ip, strict=False)), 'next-hop']) + + config.delete(interface_base + [tunnel_name, 'address']) + for tunnel_ip in tunnel_ip_list: + config.set( + interface_base + [tunnel_name, 'address'], value=tunnel_ip, replace=False) + + ## Map multicast migration + if config.exists(base + [tunnel_name, 'multicast']): + multicast_map = config.return_value( + base + [tunnel_name, 'multicast']) + if multicast_map == 'nhs': + config.delete(base + [tunnel_name, 'multicast']) + for nbma in nhs_nbmaip_list: + config.set(base + [tunnel_name, 'multicast'], value=nbma, + replace=False) + + ## Delete non-cahching + if config.exists(base + [tunnel_name, 'non-caching']): + config.delete(base + [tunnel_name, 'non-caching']) + ## Delete shortcut-destination + if config.exists(base + [tunnel_name, 'shortcut-destination']): + if not config.exists(base + [tunnel_name, 'shortcut']): + config.set(base + [tunnel_name, 'shortcut']) + config.delete(base + [tunnel_name, 'shortcut-destination']) + ## Delete shortcut-target + if config.exists(base + [tunnel_name, 'shortcut-target']): + if not config.exists(base + [tunnel_name, 'shortcut']): + config.set(base + [tunnel_name, 'shortcut']) + config.delete(base + [tunnel_name, 'shortcut-target']) + ## Set registration-no-unique + config.set(base + [tunnel_name, 'registration-no-unique'])
\ No newline at end of file diff --git a/src/op_mode/dhcp.py b/src/op_mode/dhcp.py index 20f54df25..45de86cab 100755 --- a/src/op_mode/dhcp.py +++ b/src/op_mode/dhcp.py @@ -113,7 +113,7 @@ def _get_raw_server_leases(family='inet', pool=None, sorted=None, state=[], orig data_lease['origin'] = 'local' # TODO: Determine remote in HA data_lease['hostname'] = lease.get('hostname', '-') # remove trailing dot to ensure consistency for `vyos-hostsd-client` - if data_lease['hostname'][-1] == '.': + if data_lease['hostname'] and data_lease['hostname'][-1] == '.': data_lease['hostname'] = data_lease['hostname'][:-1] if family == 'inet': diff --git a/src/op_mode/ipsec.py b/src/op_mode/ipsec.py index 02ba126b4..1ab50b105 100755 --- a/src/op_mode/ipsec.py +++ b/src/op_mode/ipsec.py @@ -1,6 +1,6 @@ #!/usr/bin/env python3 # -# Copyright (C) 2022-2024 VyOS maintainers and contributors +# Copyright (C) 2022-2025 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 @@ -700,15 +700,6 @@ def reset_profile_dst(profile: str, tunnel: str, nbma_dst: str): ] ) ) - # initiate IKE SAs - for ike in sa_nbma_list: - if ike_sa_name in ike: - vyos.ipsec.vici_initiate( - ike_sa_name, - 'dmvpn', - ike[ike_sa_name]['local-host'], - ike[ike_sa_name]['remote-host'], - ) print( f'Profile {profile} tunnel {tunnel} remote-host {nbma_dst} reset result: success' ) @@ -732,18 +723,6 @@ def reset_profile_all(profile: str, tunnel: str): ) # terminate IKE SAs vyos.ipsec.terminate_vici_by_name(ike_sa_name, None) - # initiate IKE SAs - for ike in sa_list: - if ike_sa_name in ike: - vyos.ipsec.vici_initiate( - ike_sa_name, - 'dmvpn', - ike[ike_sa_name]['local-host'], - ike[ike_sa_name]['remote-host'], - ) - print( - f'Profile {profile} tunnel {tunnel} remote-host {ike[ike_sa_name]["remote-host"]} reset result: success' - ) print(f'Profile {profile} tunnel {tunnel} reset result: success') except vyos.ipsec.ViciInitiateError as err: raise vyos.opmode.UnconfiguredSubsystem(err) diff --git a/src/op_mode/nhrp.py b/src/op_mode/nhrp.py deleted file mode 100755 index e66f33079..000000000 --- a/src/op_mode/nhrp.py +++ /dev/null @@ -1,101 +0,0 @@ -#!/usr/bin/env python3 -# -# Copyright (C) 2023 VyOS maintainers and contributors -# -# This program is free software; you can redistribute it and/or modify -# it under the terms of the GNU General Public License version 2 or later as -# published by the Free Software Foundation. -# -# This program is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU General Public License for more details. -# -# You should have received a copy of the GNU General Public License -# along with this program. If not, see <http://www.gnu.org/licenses/>. - -import sys -import tabulate -import vyos.opmode - -from vyos.utils.process import cmd -from vyos.utils.process import process_named_running -from vyos.utils.dict import colon_separated_to_dict - - -def _get_formatted_output(output_dict: dict) -> str: - """ - Create formatted table for CLI output - :param output_dict: dictionary for API - :type output_dict: dict - :return: tabulate string - :rtype: str - """ - print(f"Status: {output_dict['Status']}") - output: str = tabulate.tabulate(output_dict['routes'], headers='keys', - numalign="left") - return output - - -def _get_formatted_dict(output_string: str) -> dict: - """ - Format string returned from CMD to API list - :param output_string: String received by CMD - :type output_string: str - :return: dictionary for API - :rtype: dict - """ - formatted_dict: dict = { - 'Status': '', - 'routes': [] - } - output_list: list = output_string.split('\n\n') - for list_a in output_list: - output_dict = colon_separated_to_dict(list_a, True) - if 'Status' in output_dict: - formatted_dict['Status'] = output_dict['Status'] - else: - formatted_dict['routes'].append(output_dict) - return formatted_dict - - -def show_interface(raw: bool): - """ - Command 'show nhrp interface' - :param raw: if API - :type raw: bool - """ - if not process_named_running('opennhrp'): - raise vyos.opmode.UnconfiguredSubsystem('OpenNHRP is not running.') - interface_string: str = cmd('sudo opennhrpctl interface show') - interface_dict: dict = _get_formatted_dict(interface_string) - if raw: - return interface_dict - else: - return _get_formatted_output(interface_dict) - - -def show_tunnel(raw: bool): - """ - Command 'show nhrp tunnel' - :param raw: if API - :type raw: bool - """ - if not process_named_running('opennhrp'): - raise vyos.opmode.UnconfiguredSubsystem('OpenNHRP is not running.') - tunnel_string: str = cmd('sudo opennhrpctl show') - tunnel_dict: list = _get_formatted_dict(tunnel_string) - if raw: - return tunnel_dict - else: - return _get_formatted_output(tunnel_dict) - - -if __name__ == '__main__': - try: - res = vyos.opmode.run(sys.modules[__name__]) - if res: - print(res) - except (ValueError, vyos.opmode.Error) as e: - print(e) - sys.exit(1) diff --git a/src/op_mode/vtysh_wrapper.sh b/src/op_mode/vtysh_wrapper.sh index 25d09ce77..bc472f7bb 100755 --- a/src/op_mode/vtysh_wrapper.sh +++ b/src/op_mode/vtysh_wrapper.sh @@ -2,5 +2,5 @@ declare -a tmp # FRR uses ospf6 where we use ospfv3, and we use reset over clear for BGP, # thus alter the commands -tmp=$(echo $@ | sed -e "s/ospfv3/ospf6/" | sed -e "s/^reset bgp/clear bgp/" | sed -e "s/^reset ip bgp/clear ip bgp/") +tmp=$(echo $@ | sed -e "s/ospfv3/ospf6/" | sed -e "s/^reset bgp/clear bgp/" | sed -e "s/^reset ip bgp/clear ip bgp/"| sed -e "s/^reset ip nhrp/clear ip nhrp/") vtysh -c "$tmp" diff --git a/src/tests/test_configd_inspect.py b/src/tests/test_configd_inspect.py index ccd631893..a0470221d 100644 --- a/src/tests/test_configd_inspect.py +++ b/src/tests/test_configd_inspect.py @@ -1,4 +1,4 @@ -# Copyright (C) 2020-2024 VyOS maintainers and contributors +# Copyright (C) 2020-2025 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 @@ -12,93 +12,151 @@ # 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 ast import json -import warnings -import importlib.util -from inspect import signature -from inspect import getsource -from functools import wraps from unittest import TestCase INC_FILE = 'data/configd-include.json' CONF_DIR = 'src/conf_mode' -f_list = ['get_config', 'verify', 'generate', 'apply'] - -def import_script(s): - path = os.path.join(CONF_DIR, s) - name = os.path.splitext(s)[0].replace('-', '_') - spec = importlib.util.spec_from_file_location(name, path) - module = importlib.util.module_from_spec(spec) - spec.loader.exec_module(module) - return module - -# importing conf_mode scripts imports jinja2 with deprecation warning -def ignore_deprecation_warning(f): - @wraps(f) - def decorated_function(*args, **kwargs): - with warnings.catch_warnings(): - warnings.simplefilter("ignore") - f(*args, **kwargs) - return decorated_function +funcs = ['get_config', 'verify', 'generate', 'apply'] + + +class FunctionSig(ast.NodeVisitor): + def __init__(self): + self.func_sig_len = dict.fromkeys(funcs, None) + self.get_config_default_values = [] + + def visit_FunctionDef(self, node): + func_name = node.name + if func_name in funcs: + self.func_sig_len[func_name] = len(node.args.args) + + if func_name == 'get_config': + for default in node.args.defaults: + if isinstance(default, ast.Constant): + self.get_config_default_values.append(default.value) + + self.generic_visit(node) + + def get_sig_lengths(self): + return self.func_sig_len + + def get_config_default(self): + return self.get_config_default_values[0] + + +class LegacyCall(ast.NodeVisitor): + def __init__(self): + self.legacy_func_count = 0 + + def visit_Constant(self, node): + value = node.value + if isinstance(value, str): + if 'my_set' in value or 'my_delete' in value: + self.legacy_func_count += 1 + + self.generic_visit(node) + + def get_legacy_func_count(self): + return self.legacy_func_count + + +class ConfigInstance(ast.NodeVisitor): + def __init__(self): + self.count = 0 + + def visit_Call(self, node): + if isinstance(node.func, ast.Name): + name = node.func.id + if name == 'Config': + self.count += 1 + self.generic_visit(node) + + def get_count(self): + return self.count + + +class FunctionConfigInstance(ast.NodeVisitor): + def __init__(self): + self.func_config_instance = dict.fromkeys(funcs, 0) + + def visit_FunctionDef(self, node): + func_name = node.name + if func_name in funcs: + config_instance = ConfigInstance() + config_instance.visit(node) + self.func_config_instance[func_name] = config_instance.get_count() + self.generic_visit(node) + + def get_func_config_instance(self): + return self.func_config_instance + class TestConfigdInspect(TestCase): def setUp(self): + self.ast_list = [] + with open(INC_FILE) as f: self.inc_list = json.load(f) - @ignore_deprecation_warning - def test_signatures(self): for s in self.inc_list: - m = import_script(s) - for i in f_list: - f = getattr(m, i, None) - self.assertIsNotNone(f, f"'{s}': missing function '{i}'") - sig = signature(f) - par = sig.parameters - l = len(par) - self.assertEqual(l, 1, - f"'{s}': '{i}' incorrect signature") - if i == 'get_config': - for p in par.values(): - self.assertTrue(p.default is None, - f"'{s}': '{i}' incorrect signature") - - @ignore_deprecation_warning - def test_function_instance(self): - for s in self.inc_list: - m = import_script(s) - for i in f_list: - f = getattr(m, i, None) - if not f: - continue - str_f = getsource(f) - # Regex not XXXConfig() T3108 - n = len(re.findall(r'[^a-zA-Z]Config\(\)', str_f)) - if i == 'get_config': - self.assertEqual(n, 1, - f"'{s}': '{i}' no instance of Config") - if i != 'get_config': - self.assertEqual(n, 0, - f"'{s}': '{i}' instance of Config") - - @ignore_deprecation_warning - def test_file_instance(self): - for s in self.inc_list: - m = import_script(s) - str_m = getsource(m) - # Regex not XXXConfig T3108 - n = len(re.findall(r'[^a-zA-Z]Config\(\)', str_m)) - self.assertEqual(n, 1, - f"'{s}' more than one instance of Config") - - @ignore_deprecation_warning + s_path = f'{CONF_DIR}/{s}' + with open(s_path) as f: + s_str = f.read() + s_tree = ast.parse(s_str) + self.ast_list.append((s, s_tree)) + + def test_signatures(self): + for s, t in self.ast_list: + visitor = FunctionSig() + visitor.visit(t) + sig_lens = visitor.get_sig_lengths() + + for f in funcs: + self.assertIsNotNone(sig_lens[f], f"'{s}': '{f}' missing") + self.assertEqual(sig_lens[f], 1, f"'{s}': '{f}' incorrect signature") + + self.assertEqual( + visitor.get_config_default(), + None, + f"'{s}': 'get_config' incorrect signature", + ) + + def test_file_config_instance(self): + for s, t in self.ast_list: + visitor = ConfigInstance() + visitor.visit(t) + count = visitor.get_count() + + self.assertEqual(count, 1, f"'{s}' more than one instance of Config") + + def test_function_config_instance(self): + for s, t in self.ast_list: + visitor = FunctionConfigInstance() + visitor.visit(t) + func_config_instance = visitor.get_func_config_instance() + + for f in funcs: + if f == 'get_config': + self.assertTrue( + func_config_instance[f] > 0, + f"'{s}': '{f}' no instance of Config", + ) + self.assertTrue( + func_config_instance[f] < 2, + f"'{s}': '{f}' more than one instance of Config", + ) + else: + self.assertEqual( + func_config_instance[f], 0, f"'{s}': '{f}' instance of Config" + ) + def test_config_modification(self): - for s in self.inc_list: - m = import_script(s) - str_m = getsource(m) - n = str_m.count('my_set') - self.assertEqual(n, 0, f"'{s}' modifies config") + for s, t in self.ast_list: + visitor = LegacyCall() + visitor.visit(t) + legacy_func_count = visitor.get_legacy_func_count() + + self.assertEqual(legacy_func_count, 0, f"'{s}' modifies config") |