diff options
| -rwxr-xr-x | smoketest/scripts/cli/test_protocols_bgp.py | 24 | ||||
| -rwxr-xr-x | src/conf_mode/policy.py | 61 |
2 files changed, 70 insertions, 15 deletions
diff --git a/smoketest/scripts/cli/test_protocols_bgp.py b/smoketest/scripts/cli/test_protocols_bgp.py index 8ed0f7228..08697eebf 100755 --- a/smoketest/scripts/cli/test_protocols_bgp.py +++ b/smoketest/scripts/cli/test_protocols_bgp.py @@ -143,13 +143,7 @@ class TestProtocolsBGP(VyOSUnitTestSHIM.TestCase): self.cli_set(base_path + ['local-as', ASN]) def tearDown(self): - self.cli_delete(['policy', 'route-map', route_map_in]) - self.cli_delete(['policy', 'route-map', route_map_out]) - self.cli_delete(['policy', 'prefix-list', prefix_list_in]) - self.cli_delete(['policy', 'prefix-list', prefix_list_out]) - self.cli_delete(['policy', 'prefix-list6', prefix_list_in6]) - self.cli_delete(['policy', 'prefix-list6', prefix_list_out6]) - + self.cli_delete(['policy']) self.cli_delete(base_path) self.cli_commit() @@ -578,7 +572,7 @@ class TestProtocolsBGP(VyOSUnitTestSHIM.TestCase): verify_families = ['ipv4 unicast', 'ipv6 unicast','ipv4 multicast', 'ipv6 multicast'] flowspec_families = ['address-family ipv4 flowspec', 'address-family ipv6 flowspec'] flowspec_int = 'lo' - + # Per family distance support for family in distance_families: self.cli_set(base_path + ['address-family', family, 'distance', 'external', distance_external]) @@ -590,16 +584,16 @@ class TestProtocolsBGP(VyOSUnitTestSHIM.TestCase): if 'ipv6' in family: self.cli_set(base_path + ['address-family', family, 'distance', 'prefix', distance_v6_prefix, 'distance', distance_prefix_value]) - + # IPv4 flowspec interface check self.cli_set(base_path + ['address-family', 'ipv4-flowspec', 'local-install', 'interface', flowspec_int]) - + # IPv6 flowspec interface check self.cli_set(base_path + ['address-family', 'ipv6-flowspec', 'local-install', 'interface', flowspec_int]) - + # Commit changes self.cli_commit() - + # Verify FRR distances configuration frrconfig = self.getFRRconfig(f'router bgp {ASN}') self.assertIn(f'router bgp {ASN}', frrconfig) @@ -610,12 +604,12 @@ class TestProtocolsBGP(VyOSUnitTestSHIM.TestCase): self.assertIn(f'distance {distance_prefix_value} {distance_v4_prefix}', frrconfig) if 'ipv6' in family: self.assertIn(f'distance {distance_prefix_value} {distance_v6_prefix}', frrconfig) - + # Verify FRR flowspec configuration for family in flowspec_families: self.assertIn(f'{family}', frrconfig) self.assertIn(f'local-install {flowspec_int}', frrconfig) - + def test_bgp_10_vrf_simple(self): router_id = '127.0.0.3' vrfs = ['red', 'green', 'blue'] @@ -640,6 +634,6 @@ class TestProtocolsBGP(VyOSUnitTestSHIM.TestCase): self.assertIn(f'router bgp {ASN} vrf {vrf}', frrconfig) self.assertIn(f' bgp router-id {router_id}', frrconfig) - + if __name__ == '__main__': unittest.main(verbosity=2)
\ No newline at end of file diff --git a/src/conf_mode/policy.py b/src/conf_mode/policy.py index f0348fe06..74f23948c 100755 --- a/src/conf_mode/policy.py +++ b/src/conf_mode/policy.py @@ -25,6 +25,27 @@ from vyos import frr from vyos import airbag airbag.enable() +def routing_policy_find(key, dictionary): + # Recursively traverse a dictionary and extract the value assigned to + # a given key as generator object. This is made for routing policies, + # thus also import/export is checked + for k, v in dictionary.items(): + if k == key: + if isinstance(v, dict): + for a, b in v.items(): + if a in ['import', 'export']: + yield b + else: + yield v + elif isinstance(v, dict): + for result in routing_policy_find(key, v): + yield result + elif isinstance(v, list): + for d in v: + if isinstance(d, dict): + for result in routing_policy_find(key, d): + yield result + def get_config(config=None): if config: conf = config @@ -35,6 +56,15 @@ def get_config(config=None): policy = conf.get_config_dict(base, key_mangling=('-', '_'), get_first_key=True, no_tag_node_value_mangle=True) + # We also need some additional information from the config, prefix-lists + # and route-maps for instance. They will be used in verify(). + # + # XXX: one MUST always call this without the key_mangling() option! See + # vyos.configverify.verify_common_route_maps() for more information. + tmp = conf.get_config_dict(['protocols'], key_mangling=('-', '_'), + no_tag_node_value_mangle=True) + # Merge policy dict into "regular" config dict + policy = dict_merge(tmp, policy) return policy def verify(policy): @@ -102,6 +132,37 @@ def verify(policy): if tmp and tmp not in policy.get('large_community_list', []): raise ConfigError(f'large-community-list {tmp} does not exist!') + # Specified prefix-list must exist + tmp = dict_search('match.ip.address.prefix_list', rule_config) + if tmp and tmp not in policy.get('prefix_list', []): + raise ConfigError(f'prefix-list {tmp} does not exist!') + + # Specified prefix-list must exist + tmp = dict_search('match.ipv6.address.prefix_list', rule_config) + if tmp and tmp not in policy.get('prefix_list6', []): + raise ConfigError(f'prefix-list6 {tmp} does not exist!') + + # When routing protocols are active some use prefix-lists, route-maps etc. + # to apply the systems routing policy to the learned or redistributed routes. + # When the "routing policy" changes and policies, route-maps etc. are deleted, + # it is our responsibility to verify that the policy can not be deleted if it + # is used by any routing protocol + if 'protocols' in policy: + for policy_type in ['access_list', 'access_list6', 'as_path_list', 'community_list', + 'extcommunity_list', 'large_community_list', 'prefix_list', 'route_map']: + if policy_type in policy: + for policy_name in list(set(routing_policy_find(policy_type, policy['protocols']))): + found = False + if policy_name in policy[policy_type]: + found = True + # BGP uses prefix-list for selecting both an IPv4 or IPv6 AFI related + # list - we need to go the extra mile here and check both prefix-lists + if policy_type == 'prefix_list' and 'prefix_list6' in policy and policy_name in policy['prefix_list6']: + found = True + if not found: + tmp = policy_type.replace('_','-') + raise ConfigError(f'Can not delete {tmp} "{name}", still in use!') + return None def generate(policy): |