diff --git a/smoketest/scripts/cli/test_policy.py b/smoketest/scripts/cli/test_policy.py
index 9f9c11fb5..d055762f4 100755
--- a/smoketest/scripts/cli/test_policy.py
+++ b/smoketest/scripts/cli/test_policy.py
@@ -1,1445 +1,1445 @@
#!/usr/bin/env python3
#
# Copyright (C) 2021 VyOS maintainers and contributors
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License version 2 or later as
# published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import unittest
from base_vyostest_shim import VyOSUnitTestSHIM
from vyos.configsession import ConfigSessionError
from vyos.util import cmd
base_path = ['policy']
class TestPolicy(VyOSUnitTestSHIM.TestCase):
def tearDown(self):
self.cli_delete(base_path)
self.cli_commit()
def test_access_list(self):
acls = {
'50' : {
'rule' : {
'5' : {
'action' : 'permit',
'source' : { 'any' : '' },
},
'10' : {
'action' : 'deny',
'source' : { 'host' : '1.2.3.4' },
},
},
},
'150' : {
'rule' : {
'5' : {
'action' : 'permit',
'source' : { 'any' : '' },
'destination' : { 'host' : '2.2.2.2' },
},
'10' : {
'action' : 'deny',
'source' : { 'any' : '' },
'destination' : { 'any' : '' },
},
},
},
'2000' : {
'rule' : {
'5' : {
'action' : 'permit',
'destination' : { 'any' : '' },
'source' : { 'network' : '10.0.0.0', 'inverse-mask' : '0.255.255.255' },
},
'10' : {
'action' : 'permit',
'destination' : { 'any' : '' },
'source' : { 'network' : '172.16.0.0', 'inverse-mask' : '0.15.255.255' },
},
'15' : {
'action' : 'permit',
'destination' : { 'any' : '' },
'source' : { 'network' : '192.168.0.0', 'inverse-mask' : '0.0.255.255' },
},
'20' : {
'action' : 'permit',
'destination' : { 'network' : '172.16.0.0', 'inverse-mask' : '0.15.255.255' },
'source' : { 'network' : '10.0.0.0', 'inverse-mask' : '0.255.255.255' },
},
'25' : {
'action' : 'deny',
'destination' : { 'network' : '192.168.0.0', 'inverse-mask' : '0.0.255.255' },
'source' : { 'network' : '172.16.0.0', 'inverse-mask' : '0.15.255.255' },
},
'30' : {
'action' : 'deny',
'destination' : { 'any' : '' },
'source' : { 'any' : '' },
},
},
},
}
for acl, acl_config in acls.items():
path = base_path + ['access-list', acl]
self.cli_set(path + ['description', f'VyOS-ACL-{acl}'])
if 'rule' not in acl_config:
continue
for rule, rule_config in acl_config['rule'].items():
self.cli_set(path + ['rule', rule, 'action', rule_config['action']])
for direction in ['source', 'destination']:
if direction in rule_config:
if 'any' in rule_config[direction]:
self.cli_set(path + ['rule', rule, direction, 'any'])
if 'host' in rule_config[direction]:
self.cli_set(path + ['rule', rule, direction, 'host', rule_config[direction]['host']])
if 'network' in rule_config[direction]:
self.cli_set(path + ['rule', rule, direction, 'network', rule_config[direction]['network']])
self.cli_set(path + ['rule', rule, direction, 'inverse-mask', rule_config[direction]['inverse-mask']])
self.cli_commit()
config = self.getFRRconfig('access-list', end='')
for acl, acl_config in acls.items():
for rule, rule_config in acl_config['rule'].items():
tmp = f'access-list {acl} seq {rule}'
if rule_config['action'] == 'permit':
tmp += ' permit'
else:
tmp += ' deny'
if {'source', 'destination'} <= set(rule_config):
tmp += ' ip'
for direction in ['source', 'destination']:
if direction in rule_config:
if 'any' in rule_config[direction]:
tmp += ' any'
if 'host' in rule_config[direction]:
# XXX: Some weird side rule from the old vyatta days
# possible to clean this up after the vyos-1x migration
if int(acl) in range(100, 200) or int(acl) in range(2000, 2700):
tmp += ' host'
tmp += ' ' + rule_config[direction]['host']
if 'network' in rule_config[direction]:
tmp += ' ' + rule_config[direction]['network'] + ' ' + rule_config[direction]['inverse-mask']
self.assertIn(tmp, config)
def test_access_list6(self):
acls = {
'50' : {
'rule' : {
'5' : {
'action' : 'permit',
'source' : { 'any' : '' },
},
'10' : {
'action' : 'deny',
'source' : { 'network' : '2001:db8:10::/48', 'exact-match' : '' },
},
'15' : {
'action' : 'deny',
'source' : { 'network' : '2001:db8:20::/48' },
},
},
},
'100' : {
'rule' : {
'5' : {
'action' : 'deny',
'source' : { 'network' : '2001:db8:10::/64', 'exact-match' : '' },
},
'10' : {
'action' : 'deny',
'source' : { 'network' : '2001:db8:20::/64', },
},
'15' : {
'action' : 'deny',
'source' : { 'network' : '2001:db8:30::/64', 'exact-match' : '' },
},
'20' : {
'action' : 'deny',
'source' : { 'network' : '2001:db8:40::/64', 'exact-match' : '' },
},
'25' : {
'action' : 'deny',
'source' : { 'any' : '' },
},
},
},
}
for acl, acl_config in acls.items():
path = base_path + ['access-list6', acl]
self.cli_set(path + ['description', f'VyOS-ACL-{acl}'])
if 'rule' not in acl_config:
continue
for rule, rule_config in acl_config['rule'].items():
self.cli_set(path + ['rule', rule, 'action', rule_config['action']])
for direction in ['source', 'destination']:
if direction in rule_config:
if 'any' in rule_config[direction]:
self.cli_set(path + ['rule', rule, direction, 'any'])
if 'network' in rule_config[direction]:
self.cli_set(path + ['rule', rule, direction, 'network', rule_config[direction]['network']])
if 'exact-match' in rule_config[direction]:
self.cli_set(path + ['rule', rule, direction, 'exact-match'])
self.cli_commit()
config = self.getFRRconfig('ipv6 access-list', end='')
for acl, acl_config in acls.items():
for rule, rule_config in acl_config['rule'].items():
tmp = f'ipv6 access-list {acl} seq {rule}'
if rule_config['action'] == 'permit':
tmp += ' permit'
else:
tmp += ' deny'
if {'source', 'destination'} <= set(rule_config):
tmp += ' ip'
for direction in ['source', 'destination']:
if direction in rule_config:
if 'any' in rule_config[direction]:
tmp += ' any'
if 'network' in rule_config[direction]:
tmp += ' ' + rule_config[direction]['network']
if 'exact-match' in rule_config[direction]:
tmp += ' exact-match'
self.assertIn(tmp, config)
def test_as_path_list(self):
test_data = {
'VyOS' : {
'rule' : {
'5' : {
'action' : 'permit',
'regex' : '^44501 64502$',
},
'10' : {
'action' : 'permit',
'regex' : '44501|44502|44503',
},
'15' : {
'action' : 'permit',
'regex' : '^44501_([0-9]+_)+',
},
},
},
'Customers' : {
'rule' : {
'5' : {
'action' : 'permit',
'regex' : '_10_',
},
'10' : {
'action' : 'permit',
'regex' : '_20_',
},
'15' : {
'action' : 'permit',
'regex' : '_30_',
},
'20' : {
'action' : 'deny',
'regex' : '_40_',
},
},
},
'bogons' : {
'rule' : {
'5' : {
'action' : 'permit',
'regex' : '_0_',
},
'10' : {
'action' : 'permit',
'regex' : '_23456_',
},
'15' : {
'action' : 'permit',
'regex' : '_6449[6-9]_|_65[0-4][0-9][0-9]_|_655[0-4][0-9]_|_6555[0-1]_',
},
'20' : {
'action' : 'permit',
'regex' : '_6555[2-9]_|_655[6-9][0-9]_|_65[6-9][0-9][0-9]_|_6[6-9][0-9][0-9][0-]_|_[7-9][0-9][0-9][0-9][0-9]_|_1[0-2][0-9][0-9][0-9][0-9]_|_130[0-9][0-9][0-9]_|_1310[0-6][0-9]_|_13107[01]_',
},
},
},
}
for as_path, as_path_config in test_data.items():
path = base_path + ['as-path-list', as_path]
self.cli_set(path + ['description', f'VyOS-ASPATH-{as_path}'])
if 'rule' not in as_path_config:
continue
for rule, rule_config in as_path_config['rule'].items():
if 'action' in rule_config:
self.cli_set(path + ['rule', rule, 'action', rule_config['action']])
if 'regex' in rule_config:
self.cli_set(path + ['rule', rule, 'regex', rule_config['regex']])
self.cli_commit()
config = self.getFRRconfig('bgp as-path access-list', end='')
for as_path, as_path_config in test_data.items():
if 'rule' not in as_path_config:
continue
for rule, rule_config in as_path_config['rule'].items():
tmp = f'bgp as-path access-list {as_path} seq {rule}'
if rule_config['action'] == 'permit':
tmp += ' permit'
else:
tmp += ' deny'
tmp += ' ' + rule_config['regex']
self.assertIn(tmp, config)
def test_community_list(self):
test_data = {
'100' : {
'rule' : {
'5' : {
'action' : 'permit',
'regex' : '.*',
},
},
},
'200' : {
'rule' : {
'5' : {
'action' : 'deny',
'regex' : '^1:201$',
},
'10' : {
'action' : 'deny',
'regex' : '1:101$',
},
'15' : {
'action' : 'deny',
'regex' : '^1:100$',
},
},
},
}
for comm_list, comm_list_config in test_data.items():
path = base_path + ['community-list', comm_list]
self.cli_set(path + ['description', f'VyOS-COMM-{comm_list}'])
if 'rule' not in comm_list_config:
continue
for rule, rule_config in comm_list_config['rule'].items():
if 'action' in rule_config:
self.cli_set(path + ['rule', rule, 'action', rule_config['action']])
if 'regex' in rule_config:
self.cli_set(path + ['rule', rule, 'regex', rule_config['regex']])
self.cli_commit()
config = self.getFRRconfig('bgp community-list', end='')
for comm_list, comm_list_config in test_data.items():
if 'rule' not in comm_list_config:
continue
for rule, rule_config in comm_list_config['rule'].items():
tmp = f'bgp community-list {comm_list} seq {rule}'
if rule_config['action'] == 'permit':
tmp += ' permit'
else:
tmp += ' deny'
tmp += ' ' + rule_config['regex']
self.assertIn(tmp, config)
def test_extended_community_list(self):
test_data = {
'foo' : {
'rule' : {
'5' : {
'action' : 'permit',
'regex' : '.*',
},
},
},
'200' : {
'rule' : {
'5' : {
'action' : 'deny',
'regex' : '^1:201$',
},
'10' : {
'action' : 'deny',
'regex' : '1:101$',
},
'15' : {
'action' : 'deny',
'regex' : '^1:100$',
},
},
},
}
for comm_list, comm_list_config in test_data.items():
path = base_path + ['extcommunity-list', comm_list]
self.cli_set(path + ['description', f'VyOS-EXTCOMM-{comm_list}'])
if 'rule' not in comm_list_config:
continue
for rule, rule_config in comm_list_config['rule'].items():
if 'action' in rule_config:
self.cli_set(path + ['rule', rule, 'action', rule_config['action']])
if 'regex' in rule_config:
self.cli_set(path + ['rule', rule, 'regex', rule_config['regex']])
self.cli_commit()
config = self.getFRRconfig('bgp extcommunity-list', end='')
for comm_list, comm_list_config in test_data.items():
if 'rule' not in comm_list_config:
continue
for rule, rule_config in comm_list_config['rule'].items():
# if the community is not a number but a name, the expanded
# keyword is used
expanded = ''
if not comm_list.isnumeric():
expanded = ' expanded'
tmp = f'bgp extcommunity-list{expanded} {comm_list} seq {rule}'
if rule_config['action'] == 'permit':
tmp += ' permit'
else:
tmp += ' deny'
tmp += ' ' + rule_config['regex']
self.assertIn(tmp, config)
def test_large_community_list(self):
test_data = {
'foo' : {
'rule' : {
'5' : {
'action' : 'permit',
'regex' : '667:123:100',
},
},
},
'bar' : {
'rule' : {
'5' : {
'action' : 'permit',
'regex' : '65000:120:10',
},
'10' : {
'action' : 'permit',
'regex' : '65000:120:20',
},
'15' : {
'action' : 'permit',
'regex' : '65000:120:30',
},
},
},
}
for comm_list, comm_list_config in test_data.items():
path = base_path + ['large-community-list', comm_list]
self.cli_set(path + ['description', f'VyOS-LARGECOMM-{comm_list}'])
if 'rule' not in comm_list_config:
continue
for rule, rule_config in comm_list_config['rule'].items():
if 'action' in rule_config:
self.cli_set(path + ['rule', rule, 'action', rule_config['action']])
if 'regex' in rule_config:
self.cli_set(path + ['rule', rule, 'regex', rule_config['regex']])
self.cli_commit()
config = self.getFRRconfig('bgp large-community-list', end='')
for comm_list, comm_list_config in test_data.items():
if 'rule' not in comm_list_config:
continue
for rule, rule_config in comm_list_config['rule'].items():
tmp = f'bgp large-community-list expanded {comm_list} seq {rule}'
if rule_config['action'] == 'permit':
tmp += ' permit'
else:
tmp += ' deny'
tmp += ' ' + rule_config['regex']
self.assertIn(tmp, config)
def test_prefix_list(self):
test_data = {
'foo' : {
'rule' : {
'5' : {
'action' : 'permit',
'prefix' : '10.0.0.0/8',
'ge' : '16',
'le' : '24',
},
'10' : {
'action' : 'deny',
'prefix' : '172.16.0.0/12',
'ge' : '16',
},
'15' : {
'action' : 'permit',
'prefix' : '192.168.0.0/16',
},
},
},
'bar' : {
'rule' : {
'5' : {
'action' : 'permit',
'prefix' : '10.0.10.0/24',
'ge' : '25',
'le' : '26',
},
'10' : {
'action' : 'deny',
'prefix' : '10.0.20.0/24',
'le' : '25',
},
'15' : {
'action' : 'permit',
'prefix' : '10.0.25.0/24',
},
},
},
}
for prefix_list, prefix_list_config in test_data.items():
path = base_path + ['prefix-list', prefix_list]
self.cli_set(path + ['description', f'VyOS-PFX-LIST-{prefix_list}'])
if 'rule' not in prefix_list_config:
continue
for rule, rule_config in prefix_list_config['rule'].items():
if 'action' in rule_config:
self.cli_set(path + ['rule', rule, 'action', rule_config['action']])
if 'prefix' in rule_config:
self.cli_set(path + ['rule', rule, 'prefix', rule_config['prefix']])
if 'ge' in rule_config:
self.cli_set(path + ['rule', rule, 'ge', rule_config['ge']])
if 'le' in rule_config:
self.cli_set(path + ['rule', rule, 'le', rule_config['le']])
self.cli_commit()
config = self.getFRRconfig('ip prefix-list', end='')
for prefix_list, prefix_list_config in test_data.items():
if 'rule' not in prefix_list_config:
continue
for rule, rule_config in prefix_list_config['rule'].items():
tmp = f'ip prefix-list {prefix_list} seq {rule}'
if rule_config['action'] == 'permit':
tmp += ' permit'
else:
tmp += ' deny'
tmp += ' ' + rule_config['prefix']
if 'ge' in rule_config:
tmp += ' ge ' + rule_config['ge']
if 'le' in rule_config:
tmp += ' le ' + rule_config['le']
self.assertIn(tmp, config)
def test_prefix_list6(self):
test_data = {
'foo' : {
'rule' : {
'5' : {
'action' : 'permit',
'prefix' : '2001:db8::/32',
'ge' : '40',
'le' : '48',
},
'10' : {
'action' : 'deny',
'prefix' : '2001:db8::/32',
'ge' : '48',
},
'15' : {
'action' : 'permit',
'prefix' : '2001:db8:1000::/64',
},
},
},
'bar' : {
'rule' : {
'5' : {
'action' : 'permit',
'prefix' : '2001:db8:100::/40',
'ge' : '48',
},
'10' : {
'action' : 'permit',
'prefix' : '2001:db8:200::/40',
'ge' : '48',
},
'15' : {
'action' : 'deny',
'prefix' : '2001:db8:300::/40',
'le' : '64',
},
},
},
}
for prefix_list, prefix_list_config in test_data.items():
path = base_path + ['prefix-list6', prefix_list]
self.cli_set(path + ['description', f'VyOS-PFX-LIST-{prefix_list}'])
if 'rule' not in prefix_list_config:
continue
for rule, rule_config in prefix_list_config['rule'].items():
if 'action' in rule_config:
self.cli_set(path + ['rule', rule, 'action', rule_config['action']])
if 'prefix' in rule_config:
self.cli_set(path + ['rule', rule, 'prefix', rule_config['prefix']])
if 'ge' in rule_config:
self.cli_set(path + ['rule', rule, 'ge', rule_config['ge']])
if 'le' in rule_config:
self.cli_set(path + ['rule', rule, 'le', rule_config['le']])
self.cli_commit()
config = self.getFRRconfig('ipv6 prefix-list', end='')
for prefix_list, prefix_list_config in test_data.items():
if 'rule' not in prefix_list_config:
continue
for rule, rule_config in prefix_list_config['rule'].items():
tmp = f'ipv6 prefix-list {prefix_list} seq {rule}'
if rule_config['action'] == 'permit':
tmp += ' permit'
else:
tmp += ' deny'
tmp += ' ' + rule_config['prefix']
if 'ge' in rule_config:
tmp += ' ge ' + rule_config['ge']
if 'le' in rule_config:
tmp += ' le ' + rule_config['le']
self.assertIn(tmp, config)
def test_route_map(self):
access_list = '50'
as_path_list = '100'
test_interface = 'eth0'
community_list = 'BGP-comm-0815'
# ext community name only allows alphanumeric characters and no hyphen :/
# maybe change this if possible in vyos-1x rewrite
extcommunity_list = 'BGPextcomm123'
large_community_list = 'bgp-large-community-123456'
prefix_list = 'foo-pfx-list'
ipv6_nexthop = 'fe80::1'
local_pref = '300'
metric = '50'
peer = '2.3.4.5'
tag = '6542'
goto = '25'
test_data = {
'foo-map-bar' : {
'rule' : {
'5' : {
'action' : 'permit',
'continue' : '20',
},
'10' : {
'action' : 'permit',
'call' : 'complicated-configuration',
},
},
},
'a-matching-rule-0815': {
'rule' : {
'5' : {
'action' : 'deny',
'match' : {
'as-path' : as_path_list,
'rpki-invalid': '',
'tag': tag,
},
},
'10' : {
'action' : 'permit',
'match' : {
'community' : community_list,
'interface' : test_interface,
'rpki-not-found': '',
},
},
'15' : {
'action' : 'permit',
'match' : {
'extcommunity' : extcommunity_list,
'rpki-valid': '',
},
'on-match' : {
'next' : '',
},
},
'20' : {
'action' : 'permit',
'match' : {
'ip-address-acl': access_list,
'ip-nexthop-acl': access_list,
'ip-route-source-acl': access_list,
'ipv6-address-acl': access_list,
'origin-incomplete' : '',
},
'on-match' : {
'goto' : goto,
},
},
'25' : {
'action' : 'permit',
'match' : {
'ip-address-pfx': prefix_list,
'ip-nexthop-pfx': prefix_list,
'ip-route-source-pfx': prefix_list,
'ipv6-address-pfx': prefix_list,
'origin-igp': '',
},
},
'30' : {
'action' : 'permit',
'match' : {
'ipv6-nexthop' : ipv6_nexthop,
'large-community' : large_community_list,
'local-pref' : local_pref,
'metric': metric,
'origin-egp': '',
'peer' : peer,
},
},
},
},
'complicated-configuration' : {
'rule' : {
'10' : {
'action' : 'deny',
'set' : {
'aggregator-as' : '1234567890',
'aggregator-ip' : '10.255.255.0',
'as-path-exclude' : '1234',
'as-path-prepend' : '1234567890 987654321',
'atomic-aggregate' : '',
'distance' : '110',
'extcommunity-bw' : '20000',
'extcommunity-rt' : '123:456',
'extcommunity-soo' : '456:789',
'ipv6-next-hop-global': '2001::1',
'ipv6-next-hop-local' : 'fe80::1',
'ip-next-hop' : '192.168.1.1',
'large-community' : '100:200:300',
'local-preference' : '500',
'metric' : '150',
'metric-type' : 'type-1',
'origin' : 'incomplete',
'originator-id' : '172.16.10.1',
'src' : '100.0.0.1',
'tag' : '65530',
'weight' : '2',
},
},
},
},
'bandwidth-configuration' : {
'rule' : {
'10' : {
'action' : 'deny',
'set' : {
'as-path-prepend' : '100 100',
'distance' : '200',
'extcommunity-bw' : 'num-multipaths',
},
},
},
},
'evpn-configuration' : {
'rule' : {
'10' : {
'action' : 'permit',
'match' : {
'evpn-default-route' : '',
'evpn-rd' : '100:300',
'evpn-route-type' : 'prefix',
'evpn-vni' : '1234',
},
},
},
},
}
self.cli_set(['policy', 'access-list', access_list, 'rule', '10', 'action', 'permit'])
self.cli_set(['policy', 'access-list', access_list, 'rule', '10', 'source', 'host', '1.1.1.1'])
self.cli_set(['policy', 'access-list6', access_list, 'rule', '10', 'action', 'permit'])
self.cli_set(['policy', 'access-list6', access_list, 'rule', '10', 'source', 'network', '2001:db8::/32'])
self.cli_set(['policy', 'as-path-list', as_path_list, 'rule', '10', 'action', 'permit'])
self.cli_set(['policy', 'as-path-list', as_path_list, 'rule', '10', 'regex', '64501 64502'])
self.cli_set(['policy', 'community-list', community_list, 'rule', '10', 'action', 'deny'])
self.cli_set(['policy', 'community-list', community_list, 'rule', '10', 'regex', '65432'])
self.cli_set(['policy', 'extcommunity-list', extcommunity_list, 'rule', '10', 'action', 'deny'])
self.cli_set(['policy', 'extcommunity-list', extcommunity_list, 'rule', '10', 'regex', '65000'])
self.cli_set(['policy', 'large-community-list', large_community_list, 'rule', '10', 'action', 'permit'])
self.cli_set(['policy', 'large-community-list', large_community_list, 'rule', '10', 'regex', '100:200:300'])
self.cli_set(['policy', 'prefix-list', prefix_list, 'rule', '10', 'action', 'permit'])
self.cli_set(['policy', 'prefix-list', prefix_list, 'rule', '10', 'prefix', '192.0.2.0/24'])
self.cli_set(['policy', 'prefix-list6', prefix_list, 'rule', '10', 'action', 'permit'])
self.cli_set(['policy', 'prefix-list6', prefix_list, 'rule', '10', 'prefix', '2001:db8::/32'])
for route_map, route_map_config in test_data.items():
path = base_path + ['route-map', route_map]
self.cli_set(path + ['description', f'VyOS ROUTE-MAP {route_map}'])
if 'rule' not in route_map_config:
continue
for rule, rule_config in route_map_config['rule'].items():
if 'action' in rule_config:
self.cli_set(path + ['rule', rule, 'action', rule_config['action']])
if 'call' in rule_config:
self.cli_set(path + ['rule', rule, 'call', rule_config['call']])
if 'continue' in rule_config:
self.cli_set(path + ['rule', rule, 'continue', rule_config['continue']])
if 'match' in rule_config:
if 'as-path' in rule_config['match']:
self.cli_set(path + ['rule', rule, 'match', 'as-path', rule_config['match']['as-path']])
if 'community' in rule_config['match']:
self.cli_set(path + ['rule', rule, 'match', 'community', 'community-list', rule_config['match']['community']])
self.cli_set(path + ['rule', rule, 'match', 'community', 'exact-match'])
if 'evpn-default-route' in rule_config['match']:
self.cli_set(path + ['rule', rule, 'match', 'evpn', 'default-route'])
if 'evpn-rd' in rule_config['match']:
self.cli_set(path + ['rule', rule, 'match', 'evpn', 'rd', rule_config['match']['evpn-rd']])
if 'evpn-route-type' in rule_config['match']:
self.cli_set(path + ['rule', rule, 'match', 'evpn', 'route-type', rule_config['match']['evpn-route-type']])
if 'evpn-vni' in rule_config['match']:
self.cli_set(path + ['rule', rule, 'match', 'evpn', 'vni', rule_config['match']['evpn-vni']])
if 'extcommunity' in rule_config['match']:
self.cli_set(path + ['rule', rule, 'match', 'extcommunity', rule_config['match']['extcommunity']])
if 'interface' in rule_config['match']:
self.cli_set(path + ['rule', rule, 'match', 'interface', rule_config['match']['interface']])
if 'ip-address-acl' in rule_config['match']:
self.cli_set(path + ['rule', rule, 'match', 'ip', 'address', 'access-list', rule_config['match']['ip-address-acl']])
if 'ip-address-pfx' in rule_config['match']:
self.cli_set(path + ['rule', rule, 'match', 'ip', 'address', 'prefix-list', rule_config['match']['ip-address-pfx']])
if 'ip-nexthop-acl' in rule_config['match']:
self.cli_set(path + ['rule', rule, 'match', 'ip', 'nexthop', 'access-list', rule_config['match']['ip-nexthop-acl']])
if 'ip-nexthop-pfx' in rule_config['match']:
self.cli_set(path + ['rule', rule, 'match', 'ip', 'nexthop', 'prefix-list', rule_config['match']['ip-nexthop-pfx']])
if 'ip-route-source-acl' in rule_config['match']:
self.cli_set(path + ['rule', rule, 'match', 'ip', 'route-source', 'access-list', rule_config['match']['ip-route-source-acl']])
if 'ip-route-source-pfx' in rule_config['match']:
self.cli_set(path + ['rule', rule, 'match', 'ip', 'route-source', 'prefix-list', rule_config['match']['ip-route-source-pfx']])
if 'ipv6-address-acl' in rule_config['match']:
self.cli_set(path + ['rule', rule, 'match', 'ipv6', 'address', 'access-list', rule_config['match']['ipv6-address-acl']])
if 'ipv6-address-pfx' in rule_config['match']:
self.cli_set(path + ['rule', rule, 'match', 'ipv6', 'address', 'prefix-list', rule_config['match']['ipv6-address-pfx']])
if 'ipv6-nexthop' in rule_config['match']:
self.cli_set(path + ['rule', rule, 'match', 'ipv6', 'nexthop', rule_config['match']['ipv6-nexthop']])
if 'large-community' in rule_config['match']:
self.cli_set(path + ['rule', rule, 'match', 'large-community', 'large-community-list', rule_config['match']['large-community']])
if 'local-pref' in rule_config['match']:
self.cli_set(path + ['rule', rule, 'match', 'local-preference', rule_config['match']['local-pref']])
if 'metric' in rule_config['match']:
self.cli_set(path + ['rule', rule, 'match', 'metric', rule_config['match']['metric']])
if 'origin-igp' in rule_config['match']:
self.cli_set(path + ['rule', rule, 'match', 'origin', 'igp'])
if 'origin-egp' in rule_config['match']:
self.cli_set(path + ['rule', rule, 'match', 'origin', 'egp'])
if 'origin-incomplete' in rule_config['match']:
self.cli_set(path + ['rule', rule, 'match', 'origin', 'incomplete'])
if 'peer' in rule_config['match']:
self.cli_set(path + ['rule', rule, 'match', 'peer', rule_config['match']['peer']])
if 'rpki-invalid' in rule_config['match']:
self.cli_set(path + ['rule', rule, 'match', 'rpki', 'invalid'])
if 'rpki-not-found' in rule_config['match']:
self.cli_set(path + ['rule', rule, 'match', 'rpki', 'notfound'])
if 'rpki-valid' in rule_config['match']:
self.cli_set(path + ['rule', rule, 'match', 'rpki', 'valid'])
if 'tag' in rule_config['match']:
self.cli_set(path + ['rule', rule, 'match', 'tag', rule_config['match']['tag']])
if 'on-match' in rule_config:
if 'goto' in rule_config['on-match']:
self.cli_set(path + ['rule', rule, 'on-match', 'goto', rule_config['on-match']['goto']])
if 'next' in rule_config['on-match']:
self.cli_set(path + ['rule', rule, 'on-match', 'next'])
if 'set' in rule_config:
if 'aggregator-as' in rule_config['set']:
self.cli_set(path + ['rule', rule, 'set', 'aggregator', 'as', rule_config['set']['aggregator-as']])
if 'aggregator-ip' in rule_config['set']:
self.cli_set(path + ['rule', rule, 'set', 'aggregator', 'ip', rule_config['set']['aggregator-ip']])
if 'as-path-exclude' in rule_config['set']:
self.cli_set(path + ['rule', rule, 'set', 'as-path-exclude', rule_config['set']['as-path-exclude']])
if 'as-path-prepend' in rule_config['set']:
self.cli_set(path + ['rule', rule, 'set', 'as-path-prepend', rule_config['set']['as-path-prepend']])
if 'atomic-aggregate' in rule_config['set']:
self.cli_set(path + ['rule', rule, 'set', 'atomic-aggregate'])
if 'distance' in rule_config['set']:
self.cli_set(path + ['rule', rule, 'set', 'distance', rule_config['set']['distance']])
if 'extcommunity-bw' in rule_config['set']:
self.cli_set(path + ['rule', rule, 'set', 'extcommunity', 'bandwidth', rule_config['set']['extcommunity-bw']])
if 'extcommunity-rt' in rule_config['set']:
self.cli_set(path + ['rule', rule, 'set', 'extcommunity', 'rt', rule_config['set']['extcommunity-rt']])
if 'extcommunity-soo' in rule_config['set']:
self.cli_set(path + ['rule', rule, 'set', 'extcommunity', 'soo', rule_config['set']['extcommunity-soo']])
if 'ipv6-next-hop-global' in rule_config['set']:
self.cli_set(path + ['rule', rule, 'set', 'ipv6-next-hop', 'global', rule_config['set']['ipv6-next-hop-global']])
if 'ipv6-next-hop-local' in rule_config['set']:
self.cli_set(path + ['rule', rule, 'set', 'ipv6-next-hop', 'local', rule_config['set']['ipv6-next-hop-local']])
if 'ip-next-hop' in rule_config['set']:
self.cli_set(path + ['rule', rule, 'set', 'ip-next-hop', rule_config['set']['ip-next-hop']])
if 'large-community' in rule_config['set']:
self.cli_set(path + ['rule', rule, 'set', 'large-community', rule_config['set']['large-community']])
if 'local-preference' in rule_config['set']:
self.cli_set(path + ['rule', rule, 'set', 'local-preference', rule_config['set']['local-preference']])
if 'metric' in rule_config['set']:
self.cli_set(path + ['rule', rule, 'set', 'metric', rule_config['set']['metric']])
if 'metric-type' in rule_config['set']:
self.cli_set(path + ['rule', rule, 'set', 'metric-type', rule_config['set']['metric-type']])
if 'origin' in rule_config['set']:
self.cli_set(path + ['rule', rule, 'set', 'origin', rule_config['set']['origin']])
if 'originator-id' in rule_config['set']:
self.cli_set(path + ['rule', rule, 'set', 'originator-id', rule_config['set']['originator-id']])
if 'src' in rule_config['set']:
self.cli_set(path + ['rule', rule, 'set', 'src', rule_config['set']['src']])
if 'tag' in rule_config['set']:
self.cli_set(path + ['rule', rule, 'set', 'tag', rule_config['set']['tag']])
if 'weight' in rule_config['set']:
self.cli_set(path + ['rule', rule, 'set', 'weight', rule_config['set']['weight']])
self.cli_commit()
for route_map, route_map_config in test_data.items():
if 'rule' not in route_map_config:
continue
for rule, rule_config in route_map_config['rule'].items():
name = f'route-map {route_map} {rule_config["action"]} {rule}'
config = self.getFRRconfig(name)
self.assertIn(name, config)
if 'call' in rule_config:
tmp = 'call ' + rule_config['call']
self.assertIn(tmp, config)
if 'continue' in rule_config:
tmp = 'on-match goto ' + rule_config['continue']
self.assertIn(tmp, config)
if 'match' in rule_config:
if 'as-path' in rule_config['match']:
tmp = 'match as-path ' + rule_config['match']['as-path']
self.assertIn(tmp, config)
if 'community' in rule_config['match']:
tmp = f'match community {rule_config["match"]["community"]} exact-match'
self.assertIn(tmp, config)
if 'evpn-default-route' in rule_config['match']:
tmp = f'match evpn default-route'
self.assertIn(tmp, config)
if 'evpn-rd' in rule_config['match']:
tmp = f'match evpn rd {rule_config["match"]["evpn-rd"]}'
self.assertIn(tmp, config)
if 'evpn-route-type' in rule_config['match']:
tmp = f'match evpn route-type {rule_config["match"]["evpn-route-type"]}'
self.assertIn(tmp, config)
if 'evpn-vni' in rule_config['match']:
tmp = f'match evpn vni {rule_config["match"]["evpn-vni"]}'
self.assertIn(tmp, config)
if 'extcommunity' in rule_config['match']:
tmp = f'match extcommunity {rule_config["match"]["extcommunity"]}'
self.assertIn(tmp, config)
if 'interface' in rule_config['match']:
tmp = f'match interface {rule_config["match"]["interface"]}'
self.assertIn(tmp, config)
if 'ip-address-acl' in rule_config['match']:
tmp = f'match ip address {rule_config["match"]["ip-address-acl"]}'
self.assertIn(tmp, config)
if 'ip-address-pfx' in rule_config['match']:
tmp = f'match ip address prefix-list {rule_config["match"]["ip-address-pfx"]}'
self.assertIn(tmp, config)
if 'ip-nexthop-acl' in rule_config['match']:
tmp = f'match ip next-hop {rule_config["match"]["ip-nexthop-acl"]}'
self.assertIn(tmp, config)
if 'ip-nexthop-pfx' in rule_config['match']:
tmp = f'match ip next-hop prefix-list {rule_config["match"]["ip-nexthop-pfx"]}'
self.assertIn(tmp, config)
if 'ip-route-source-acl' in rule_config['match']:
tmp = f'match ip route-source {rule_config["match"]["ip-route-source-acl"]}'
self.assertIn(tmp, config)
if 'ip-route-source-pfx' in rule_config['match']:
tmp = f'match ip route-source prefix-list {rule_config["match"]["ip-route-source-pfx"]}'
self.assertIn(tmp, config)
if 'ipv6-address-acl' in rule_config['match']:
tmp = f'match ipv6 address {rule_config["match"]["ipv6-address-acl"]}'
self.assertIn(tmp, config)
if 'ipv6-address-pfx' in rule_config['match']:
tmp = f'match ipv6 address prefix-list {rule_config["match"]["ipv6-address-pfx"]}'
self.assertIn(tmp, config)
if 'ipv6-nexthop' in rule_config['match']:
tmp = f'match ipv6 next-hop {rule_config["match"]["ipv6-nexthop"]}'
self.assertIn(tmp, config)
if 'large-community' in rule_config['match']:
tmp = f'match large-community {rule_config["match"]["large-community"]}'
self.assertIn(tmp, config)
if 'local-pref' in rule_config['match']:
tmp = f'match local-preference {rule_config["match"]["local-pref"]}'
self.assertIn(tmp, config)
if 'metric' in rule_config['match']:
tmp = f'match metric {rule_config["match"]["metric"]}'
self.assertIn(tmp, config)
if 'origin-igp' in rule_config['match']:
tmp = f'match origin igp'
self.assertIn(tmp, config)
if 'origin-egp' in rule_config['match']:
tmp = f'match origin egp'
self.assertIn(tmp, config)
if 'origin-incomplete' in rule_config['match']:
tmp = f'match origin incomplete'
self.assertIn(tmp, config)
if 'peer' in rule_config['match']:
tmp = f'match peer {rule_config["match"]["peer"]}'
self.assertIn(tmp, config)
if 'rpki-invalid' in rule_config['match']:
tmp = f'match rpki invalid'
self.assertIn(tmp, config)
if 'rpki-not-found' in rule_config['match']:
tmp = f'match rpki notfound'
self.assertIn(tmp, config)
if 'rpki-valid' in rule_config['match']:
tmp = f'match rpki valid'
self.assertIn(tmp, config)
if 'tag' in rule_config['match']:
tmp = f'match tag {rule_config["match"]["tag"]}'
self.assertIn(tmp, config)
if 'on-match' in rule_config:
if 'goto' in rule_config['on-match']:
tmp = f'on-match goto {rule_config["on-match"]["goto"]}'
self.assertIn(tmp, config)
if 'next' in rule_config['on-match']:
tmp = f'on-match next'
self.assertIn(tmp, config)
if 'set' in rule_config:
tmp = ' set '
if 'aggregator-as' in rule_config['set']:
tmp += 'aggregator as ' + rule_config['set']['aggregator-as']
elif 'aggregator-ip' in rule_config['set']:
tmp += ' ' + rule_config['set']['aggregator-ip']
elif 'as-path-exclude' in rule_config['set']:
tmp += 'as-path exclude ' + rule_config['set']['as-path-exclude']
elif 'as-path-prepend' in rule_config['set']:
tmp += 'as-path prepend ' + rule_config['set']['as-path-prepend']
elif 'atomic-aggregate' in rule_config['set']:
tmp += 'atomic-aggregate'
elif 'distance' in rule_config['set']:
tmp += 'distance ' + rule_config['set']['distance']
elif 'extcommunity-bw' in rule_config['set']:
tmp += 'extcommunity bandwidth' + rule_config['set']['extcommunity-bw']
elif 'extcommunity-rt' in rule_config['set']:
tmp += 'extcommunity rt' + rule_config['set']['extcommunity-rt']
elif 'extcommunity-soo' in rule_config['set']:
tmp += 'extcommunity rt' + rule_config['set']['extcommunity-soo']
elif 'ip-next-hop' in rule_config['set']:
tmp += 'ip next-hop ' + rule_config['set']['ip-next-hop']
elif 'ipv6-next-hop-global' in rule_config['set']:
tmp += 'ipv6 next-hop global ' + rule_config['set']['ipv6-next-hop-global']
elif 'ipv6-next-hop-local' in rule_config['set']:
tmp += 'ipv6 next-hop local ' + rule_config['set']['ipv6-next-hop-local']
elif 'large-community' in rule_config['set']:
tmp += 'large-community ' + rule_config['set']['large-community']
elif 'local-preference' in rule_config['set']:
tmp += 'local-preference ' + rule_config['set']['local-preference']
elif 'metric' in rule_config['set']:
tmp += 'metric ' + rule_config['set']['metric']
elif 'metric-type' in rule_config['set']:
tmp += 'metric-type ' + rule_config['set']['metric-type']
elif 'origin' in rule_config['set']:
tmp += 'origin ' + rule_config['set']['origin']
elif 'originator-id' in rule_config['set']:
tmp += 'originator-id ' + rule_config['set']['originator-id']
elif 'src' in rule_config['set']:
tmp += 'src ' + rule_config['set']['src']
elif 'tag' in rule_config['set']:
tmp += 'tag ' + rule_config['set']['tag']
elif 'weight' in rule_config['set']:
tmp += 'weight ' + rule_config['set']['weight']
self.assertIn(tmp, config)
# Test set table for some sources
def test_table_id(self):
path = base_path + ['local-route']
sources = ['203.0.113.1', '203.0.113.2']
rule = '50'
table = '23'
for src in sources:
self.cli_set(path + ['rule', rule, 'set', 'table', table])
self.cli_set(path + ['rule', rule, 'source', src])
self.cli_commit()
# Check generated configuration
# Expected values
original = """
50: from 203.0.113.1 lookup 23
50: from 203.0.113.2 lookup 23
"""
tmp = cmd('ip rule show prio 50')
self.assertEqual(sort_ip(tmp), sort_ip(original))
# Test set table for fwmark
def test_fwmark_table_id(self):
path = base_path + ['local-route']
fwmk = '24'
rule = '101'
table = '154'
self.cli_set(path + ['rule', rule, 'set', 'table', table])
self.cli_set(path + ['rule', rule, 'fwmark', fwmk])
self.cli_commit()
# Check generated configuration
# Expected values
original = """
101: from all fwmark 0x18 lookup 154
"""
tmp = cmd('ip rule show prio 101')
self.assertEqual(sort_ip(tmp), sort_ip(original))
# Test set table for destination
def test_destination_table_id(self):
path = base_path + ['local-route']
dst = '203.0.113.1'
rule = '102'
table = '154'
self.cli_set(path + ['rule', rule, 'set', 'table', table])
self.cli_set(path + ['rule', rule, 'destination', dst])
self.cli_commit()
# Check generated configuration
# Expected values
original = """
102: from all to 203.0.113.1 lookup 154
"""
tmp = cmd('ip rule show prio 102')
self.assertEqual(sort_ip(tmp), sort_ip(original))
# Test set table for sources with fwmark
def test_fwmark_sources_table_id(self):
path = base_path + ['local-route']
sources = ['203.0.113.11', '203.0.113.12']
fwmk = '23'
rule = '100'
table = '150'
for src in sources:
self.cli_set(path + ['rule', rule, 'set', 'table', table])
self.cli_set(path + ['rule', rule, 'source', src])
self.cli_set(path + ['rule', rule, 'fwmark', fwmk])
self.cli_commit()
# Check generated configuration
# Expected values
original = """
100: from 203.0.113.11 fwmark 0x17 lookup 150
100: from 203.0.113.12 fwmark 0x17 lookup 150
"""
tmp = cmd('ip rule show prio 100')
self.assertEqual(sort_ip(tmp), sort_ip(original))
# Test set table for sources and destinations with fwmark
def test_fwmark_sources_destination_table_id(self):
path = base_path + ['local-route']
sources = ['203.0.113.11', '203.0.113.12']
destinations = ['203.0.113.13', '203.0.113.15']
fwmk = '23'
rule = '103'
table = '150'
for src in sources:
for dst in destinations:
self.cli_set(path + ['rule', rule, 'set', 'table', table])
self.cli_set(path + ['rule', rule, 'source', src])
self.cli_set(path + ['rule', rule, 'destination', dst])
self.cli_set(path + ['rule', rule, 'fwmark', fwmk])
self.cli_commit()
# Check generated configuration
# Expected values
original = """
103: from 203.0.113.11 to 203.0.113.13 fwmark 0x17 lookup 150
103: from 203.0.113.11 to 203.0.113.15 fwmark 0x17 lookup 150
103: from 203.0.113.12 to 203.0.113.13 fwmark 0x17 lookup 150
103: from 203.0.113.12 to 203.0.113.15 fwmark 0x17 lookup 150
"""
tmp = cmd('ip rule show prio 103')
self.assertEqual(sort_ip(tmp), sort_ip(original))
# Test set table ipv6 for some sources ipv6
def test_ipv6_table_id(self):
path = base_path + ['local-route6']
sources = ['2001:db8:123::/48', '2001:db8:126::/48']
rule = '50'
table = '23'
for src in sources:
self.cli_set(path + ['rule', rule, 'set', 'table', table])
self.cli_set(path + ['rule', rule, 'source', src])
self.cli_commit()
# Check generated configuration
# Expected values
original = """
50: from 2001:db8:123::/48 lookup 23
50: from 2001:db8:126::/48 lookup 23
"""
tmp = cmd('ip -6 rule show prio 50')
self.assertEqual(sort_ip(tmp), sort_ip(original))
# Test set table for fwmark ipv6
def test_fwmark_ipv6_table_id(self):
path = base_path + ['local-route6']
fwmk = '24'
rule = '100'
table = '154'
self.cli_set(path + ['rule', rule, 'set', 'table', table])
self.cli_set(path + ['rule', rule, 'fwmark', fwmk])
self.cli_commit()
# Check generated configuration
# Expected values
original = """
100: from all fwmark 0x18 lookup 154
"""
tmp = cmd('ip -6 rule show prio 100')
self.assertEqual(sort_ip(tmp), sort_ip(original))
# Test set table for destination ipv6
def test_destination_ipv6_table_id(self):
path = base_path + ['local-route6']
dst = '2001:db8:1337::/126'
rule = '101'
table = '154'
self.cli_set(path + ['rule', rule, 'set', 'table', table])
self.cli_set(path + ['rule', rule, 'destination', dst])
self.cli_commit()
# Check generated configuration
# Expected values
original = """
101: from all to 2001:db8:1337::/126 lookup 154
"""
tmp = cmd('ip -6 rule show prio 101')
self.assertEqual(sort_ip(tmp), sort_ip(original))
# Test set table for sources with fwmark ipv6
def test_fwmark_sources_ipv6_table_id(self):
path = base_path + ['local-route6']
sources = ['2001:db8:1338::/126', '2001:db8:1339::/126']
fwmk = '23'
rule = '102'
table = '150'
for src in sources:
self.cli_set(path + ['rule', rule, 'set', 'table', table])
self.cli_set(path + ['rule', rule, 'source', src])
self.cli_set(path + ['rule', rule, 'fwmark', fwmk])
self.cli_commit()
# Check generated configuration
# Expected values
original = """
102: from 2001:db8:1338::/126 fwmark 0x17 lookup 150
102: from 2001:db8:1339::/126 fwmark 0x17 lookup 150
"""
tmp = cmd('ip -6 rule show prio 102')
self.assertEqual(sort_ip(tmp), sort_ip(original))
# Test set table for sources and destinations with fwmark ipv6
def test_fwmark_sources_destination_ipv6_table_id(self):
path = base_path + ['local-route6']
sources = ['2001:db8:1338::/126', '2001:db8:1339::/56']
destinations = ['2001:db8:13::/48', '2001:db8:16::/48']
fwmk = '23'
rule = '103'
table = '150'
for src in sources:
for dst in destinations:
self.cli_set(path + ['rule', rule, 'set', 'table', table])
self.cli_set(path + ['rule', rule, 'source', src])
self.cli_set(path + ['rule', rule, 'destination', dst])
self.cli_set(path + ['rule', rule, 'fwmark', fwmk])
self.cli_commit()
# Check generated configuration
# Expected values
original = """
103: from 2001:db8:1338::/126 to 2001:db8:13::/48 fwmark 0x17 lookup 150
103: from 2001:db8:1338::/126 to 2001:db8:16::/48 fwmark 0x17 lookup 150
103: from 2001:db8:1339::/56 to 2001:db8:13::/48 fwmark 0x17 lookup 150
103: from 2001:db8:1339::/56 to 2001:db8:16::/48 fwmark 0x17 lookup 150
"""
tmp = cmd('ip -6 rule show prio 103')
self.assertEqual(sort_ip(tmp), sort_ip(original))
# Test delete table for sources and destination with fwmark ipv4/ipv6
def test_delete_ipv4_ipv6_table_id(self):
path = base_path + ['local-route']
path_v6 = base_path + ['local-route6']
- sources = ['203.0.113.1/24', '203.0.114.5']
- destinations = ['203.0.112.1/24', '203.0.116.5']
+ sources = ['203.0.113.0/24', '203.0.114.5']
+ destinations = ['203.0.112.0/24', '203.0.116.5']
sources_v6 = ['2001:db8:1338::/126', '2001:db8:1339::/56']
destinations_v6 = ['2001:db8:13::/48', '2001:db8:16::/48']
fwmk = '23'
rule = '103'
table = '150'
for src in sources:
for dst in destinations:
self.cli_set(path + ['rule', rule, 'set', 'table', table])
self.cli_set(path + ['rule', rule, 'source', src])
self.cli_set(path + ['rule', rule, 'destination', dst])
self.cli_set(path + ['rule', rule, 'fwmark', fwmk])
for src in sources_v6:
for dst in destinations_v6:
self.cli_set(path_v6 + ['rule', rule, 'set', 'table', table])
self.cli_set(path_v6 + ['rule', rule, 'source', src])
self.cli_set(path_v6 + ['rule', rule, 'destination', dst])
self.cli_set(path_v6 + ['rule', rule, 'fwmark', fwmk])
self.cli_commit()
# Check generated configuration
# Expected values
original = """
103: from 203.0.113.1/24 to 203.0.112.1/24 fwmark 0x17 lookup 150
103: from 203.0.113.1/24 to 203.0.116.5 fwmark 0x17 lookup 150
103: from 203.0.114.5 to 203.0.112.1/24 fwmark 0x17 lookup 150
103: from 203.0.114.5 to 203.0.116.5 fwmark 0x17 lookup 150
"""
original_v6 = """
103: from 20016 to 2001:db8:13::/48 fwmark 0x17 lookup 150
103: from 2001:db8:1338::/126 to 2001:db8:16::/48 fwmark 0x17 lookup 150
103: from 2001:db8:1339::/56 to 2001:db8:13::/48 fwmark 0x17 lookup 150
103: from 2001:db8:1339::/56 to 2001:db8:16::/48 fwmark 0x17 lookup 150
"""
tmp = cmd('ip rule show prio 103')
tmp_v6 = cmd('ip -6 rule show prio 103')
self.assertEqual(sort_ip(tmp), sort_ip(original))
self.assertEqual(sort_ip(tmp_v6), sort_ip(original_v6))
self.cli_delete(path)
self.cli_delete(path_v6)
self.cli_commit()
tmp = cmd('ip rule show prio 103')
tmp_v6 = cmd('ip -6 rule show prio 103')
- original = ['']
- original_v6 = ['']
+ original = None
+ original_v6 = None
self.assertEqual(sort_ip(tmp), original)
self.assertEqual(sort_ip(tmp_v6), original_v6)
def sort_ip(output):
return output.splitlines().sort()
if __name__ == '__main__':
unittest.main(verbosity=2)
diff --git a/src/conf_mode/policy-local-route.py b/src/conf_mode/policy-local-route.py
index 2541603e2..6dabb37ae 100755
--- a/src/conf_mode/policy-local-route.py
+++ b/src/conf_mode/policy-local-route.py
@@ -1,164 +1,169 @@
#!/usr/bin/env python3
#
# Copyright (C) 2020-2021 VyOS maintainers and contributors
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License version 2 or later as
# published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import os
from sys import exit
from vyos.config import Config
from vyos.configdict import dict_merge
from vyos.configdict import node_changed
from vyos.configdict import leaf_node_changed
from vyos.template import render
from vyos.util import call
from vyos import ConfigError
from vyos import airbag
airbag.enable()
def get_config(config=None):
if config:
conf = config
else:
conf = Config()
base = ['policy']
pbr = conf.get_config_dict(base, key_mangling=('-', '_'), get_first_key=True)
for route in ['local_route', 'local_route6']:
dict_id = 'rule_remove' if route == 'local_route' else 'rule6_remove'
route_key = 'local-route' if route == 'local_route' else 'local-route6'
base_rule = base + [route_key, 'rule']
# delete policy local-route
dict = {}
tmp = node_changed(conf, base_rule, key_mangling=('-', '_'))
if tmp:
for rule in (tmp or []):
src = leaf_node_changed(conf, base_rule + [rule, 'source'])
fwmk = leaf_node_changed(conf, base_rule + [rule, 'fwmark'])
dst = leaf_node_changed(conf, base_rule + [rule, 'destination'])
rule_def = {}
if src:
rule_def = dict_merge({'source' : src}, rule_def)
if fwmk:
rule_def = dict_merge({'fwmark' : fwmk}, rule_def)
if dst:
rule_def = dict_merge({'destination' : dst}, rule_def)
dict = dict_merge({dict_id : {rule : rule_def}}, dict)
pbr.update(dict)
if not route in pbr:
continue
# delete policy local-route rule x source x.x.x.x
# delete policy local-route rule x fwmark x
# delete policy local-route rule x destination x.x.x.x
if 'rule' in pbr[route]:
for rule in pbr[route]['rule']:
src = leaf_node_changed(conf, base_rule + [rule, 'source'])
fwmk = leaf_node_changed(conf, base_rule + [rule, 'fwmark'])
dst = leaf_node_changed(conf, base_rule + [rule, 'destination'])
rule_def = {}
if src:
rule_def = dict_merge({'source' : src}, rule_def)
if fwmk:
rule_def = dict_merge({'fwmark' : fwmk}, rule_def)
if dst:
rule_def = dict_merge({'destination' : dst}, rule_def)
dict = dict_merge({dict_id : {rule : rule_def}}, dict)
pbr.update(dict)
return pbr
def verify(pbr):
# bail out early - looks like removal from running config
if not pbr:
return None
for route in ['local_route', 'local_route6']:
if not route in pbr:
continue
pbr_route = pbr[route]
if 'rule' in pbr_route:
for rule in pbr_route['rule']:
if 'source' not in pbr_route['rule'][rule] and 'destination' not in pbr_route['rule'][rule] and 'fwmark' not in pbr_route['rule'][rule]:
raise ConfigError('Source or destination address or fwmark is required!')
else:
if 'set' not in pbr_route['rule'][rule] or 'table' not in pbr_route['rule'][rule]['set']:
raise ConfigError('Table set is required!')
return None
def generate(pbr):
if not pbr:
return None
return None
def apply(pbr):
if not pbr:
return None
# Delete old rule if needed
for rule_rm in ['rule_remove', 'rule6_remove']:
if rule_rm in pbr:
v6 = " -6" if rule_rm == 'rule6_remove' else ""
for rule, rule_config in pbr[rule_rm].items():
- for src in (rule_config['source'] or ['']):
+ rule_config['source'] = rule_config['source'] if 'source' in rule_config else ['']
+ for src in rule_config['source']:
f_src = '' if src == '' else f' from {src} '
- for dst in (rule_config['destination'] or ['']):
+ rule_config['destination'] = rule_config['destination'] if 'destination' in rule_config else ['']
+ for dst in rule_config['destination']:
f_dst = '' if dst == '' else f' to {dst} '
- for fwmk in (rule_config['fwmark'] or ['']):
+ rule_config['fwmark'] = rule_config['fwmark'] if 'fwmark' in rule_config else ['']
+ for fwmk in rule_config['fwmark']:
f_fwmk = '' if fwmk == '' else f' fwmark {fwmk} '
call(f'ip{v6} rule del prio {rule} {f_src}{f_dst}{f_fwmk}')
# Generate new config
for route in ['local_route', 'local_route6']:
if not route in pbr:
continue
v6 = " -6" if route == 'local_route6' else ""
pbr_route = pbr[route]
if 'rule' in pbr_route:
for rule, rule_config in pbr_route['rule'].items():
table = rule_config['set']['table']
- for src in (rule_config['source'] or ['all']):
+ rule_config['source'] = rule_config['source'] if 'source' in rule_config else ['all']
+ for src in rule_config['source'] or ['all']:
f_src = '' if src == '' else f' from {src} '
- for dst in (rule_config['destination'] or ['all']):
+ rule_config['destination'] = rule_config['destination'] if 'destination' in rule_config else ['all']
+ for dst in rule_config['destination']:
f_dst = '' if dst == '' else f' to {dst} '
f_fwmk = ''
if 'fwmark' in rule_config:
fwmk = rule_config['fwmark']
f_fwmk = f' fwmark {fwmk} '
call(f'ip{v6} rule add prio {rule} {f_src}{f_dst}{f_fwmk} lookup {table}')
return None
if __name__ == '__main__':
try:
c = get_config()
verify(c)
generate(c)
apply(c)
except ConfigError as e:
print(e)
exit(1)