diff options
Diffstat (limited to 'smoketest/scripts/cli')
| -rw-r--r-- | smoketest/scripts/cli/base_accel_ppp_test.py | 28 | ||||
| -rwxr-xr-x | smoketest/scripts/cli/test_firewall.py | 310 | ||||
| -rwxr-xr-x | smoketest/scripts/cli/test_interfaces_ethernet.py | 38 | ||||
| -rwxr-xr-x | smoketest/scripts/cli/test_load_balancning_wan.py | 1 | ||||
| -rwxr-xr-x | smoketest/scripts/cli/test_nat.py | 141 | ||||
| -rwxr-xr-x | smoketest/scripts/cli/test_nat66.py | 67 | ||||
| -rwxr-xr-x | smoketest/scripts/cli/test_policy_route.py | 82 | ||||
| -rwxr-xr-x | smoketest/scripts/cli/test_protocols_bgp.py | 2 | ||||
| -rwxr-xr-x | smoketest/scripts/cli/test_protocols_isis.py | 47 | ||||
| -rwxr-xr-x | smoketest/scripts/cli/test_protocols_nhrp.py | 104 | ||||
| -rwxr-xr-x | smoketest/scripts/cli/test_service_ipoe-server.py | 91 | ||||
| -rwxr-xr-x | smoketest/scripts/cli/test_service_pppoe-server.py | 52 | ||||
| -rwxr-xr-x | smoketest/scripts/cli/test_system_conntrack.py | 2 | ||||
| -rwxr-xr-x | smoketest/scripts/cli/test_vpn_ipsec.py | 50 | ||||
| -rwxr-xr-x | smoketest/scripts/cli/test_vpn_sstp.py | 54 | ||||
| -rwxr-xr-x | smoketest/scripts/cli/test_zone_policy.py | 69 |
16 files changed, 750 insertions, 388 deletions
diff --git a/smoketest/scripts/cli/base_accel_ppp_test.py b/smoketest/scripts/cli/base_accel_ppp_test.py index b2acb03cc..471bdaffb 100644 --- a/smoketest/scripts/cli/base_accel_ppp_test.py +++ b/smoketest/scripts/cli/base_accel_ppp_test.py @@ -1,4 +1,4 @@ -# Copyright (C) 2020 VyOS maintainers and contributors +# Copyright (C) 2020-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 @@ -27,6 +27,17 @@ from vyos.util import process_named_running class BasicAccelPPPTest: class TestCase(VyOSUnitTestSHIM.TestCase): + + @classmethod + def setUpClass(cls): + cls._process_name = 'accel-pppd' + + super(BasicAccelPPPTest.TestCase, cls).setUpClass() + + # ensure we can also run this test on a live system - so lets clean + # out the current configuration :) + cls.cli_delete(cls, cls._base_path) + def setUp(self): self._gateway = '192.0.2.1' # ensure we can also run this test on a live system - so lets clean @@ -34,9 +45,15 @@ class BasicAccelPPPTest: self.cli_delete(self._base_path) def tearDown(self): + # Check for running process + self.assertTrue(process_named_running(self._process_name)) + self.cli_delete(self._base_path) self.cli_commit() + # Check for running process + self.assertFalse(process_named_running(self._process_name)) + def set(self, path): self.cli_set(self._base_path + path) @@ -113,9 +130,6 @@ class BasicAccelPPPTest: tmp = re.findall(regex, tmp) self.assertTrue(tmp) - # Check for running process - self.assertTrue(process_named_running(self._process_name)) - # Check local-users default value(s) self.delete(['authentication', 'local-users', 'username', user, 'static-ip']) # commit changes @@ -127,9 +141,6 @@ class BasicAccelPPPTest: tmp = re.findall(regex, tmp) self.assertTrue(tmp) - # Check for running process - self.assertTrue(process_named_running(self._process_name)) - def test_accel_radius_authentication(self): # Test configuration of RADIUS authentication for PPPoE server self.basic_config() @@ -186,9 +197,6 @@ class BasicAccelPPPTest: self.assertEqual(f'req-limit=0', server[4]) self.assertEqual(f'fail-time=0', server[5]) - # Check for running process - self.assertTrue(process_named_running(self._process_name)) - # # Disable Radius Accounting # diff --git a/smoketest/scripts/cli/test_firewall.py b/smoketest/scripts/cli/test_firewall.py index 684a07681..4b2cf9864 100755 --- a/smoketest/scripts/cli/test_firewall.py +++ b/smoketest/scripts/cli/test_firewall.py @@ -45,15 +45,11 @@ class TestFirewall(VyOSUnitTestSHIM.TestCase): # out the current configuration :) cls.cli_delete(cls, ['firewall']) - cls.cli_set(cls, ['interfaces', 'ethernet', 'eth0', 'address', '172.16.10.1/24']) - @classmethod def tearDownClass(cls): - cls.cli_delete(cls, ['interfaces', 'ethernet', 'eth0', 'address', '172.16.10.1/24']) super(TestFirewall, cls).tearDownClass() def tearDown(self): - self.cli_delete(['interfaces', 'ethernet', 'eth0', 'firewall']) self.cli_delete(['firewall']) self.cli_commit() @@ -67,7 +63,7 @@ class TestFirewall(VyOSUnitTestSHIM.TestCase): ['chain NAME_smoketest'] ] - self.verify_nftables(nftables_search, 'ip filter', inverse=True) + self.verify_nftables(nftables_search, 'ip vyos_filter', inverse=True) def verify_nftables(self, nftables_search, table, inverse=False, args=''): nftables_output = cmd(f'sudo nft {args} list table {table}') @@ -95,8 +91,9 @@ class TestFirewall(VyOSUnitTestSHIM.TestCase): ['ip saddr @GEOIP_CC_smoketest_1', 'drop'], ['ip saddr != @GEOIP_CC_smoketest_2', 'return'] ] + # -t prevents 1000+ GeoIP elements being returned - self.verify_nftables(nftables_search, 'ip filter', args='-t') + self.verify_nftables(nftables_search, 'ip vyos_filter', args='-t') def test_groups(self): hostmap_path = ['system', 'static-host-mapping', 'host-name'] @@ -114,6 +111,7 @@ class TestFirewall(VyOSUnitTestSHIM.TestCase): self.cli_set(['firewall', 'group', 'port-group', 'smoketest_port', 'port', '123']) self.cli_set(['firewall', 'group', 'domain-group', 'smoketest_domain', 'address', 'example.com']) self.cli_set(['firewall', 'group', 'domain-group', 'smoketest_domain', 'address', 'example.org']) + self.cli_set(['firewall', 'name', 'smoketest', 'rule', '1', 'action', 'accept']) self.cli_set(['firewall', 'name', 'smoketest', 'rule', '1', 'source', 'group', 'network-group', 'smoketest_network']) self.cli_set(['firewall', 'name', 'smoketest', 'rule', '1', 'destination', 'address', '172.16.10.10']) @@ -124,7 +122,7 @@ class TestFirewall(VyOSUnitTestSHIM.TestCase): self.cli_set(['firewall', 'name', 'smoketest', 'rule', '3', 'action', 'accept']) self.cli_set(['firewall', 'name', 'smoketest', 'rule', '3', 'source', 'group', 'domain-group', 'smoketest_domain']) - self.cli_set(['interfaces', 'ethernet', 'eth0', 'firewall', 'in', 'name', 'smoketest']) + self.cli_set(['firewall', 'interface', 'eth0', 'in', 'name', 'smoketest']) self.cli_commit() nftables_search = [ @@ -139,7 +137,7 @@ class TestFirewall(VyOSUnitTestSHIM.TestCase): ['192.0.2.10, 192.0.2.11 }'], ['ip saddr @D_smoketest_domain', 'return'] ] - self.verify_nftables(nftables_search, 'ip filter') + self.verify_nftables(nftables_search, 'ip vyos_filter') self.cli_delete(['system', 'static-host-mapping']) self.cli_commit() @@ -156,7 +154,7 @@ class TestFirewall(VyOSUnitTestSHIM.TestCase): self.cli_set(['firewall', 'name', 'smoketest', 'rule', '1', 'destination', 'group', 'port-group', 'smoketest_port1']) self.cli_set(['firewall', 'name', 'smoketest', 'rule', '1', 'protocol', 'tcp_udp']) - self.cli_set(['interfaces', 'ethernet', 'eth0', 'firewall', 'in', 'name', 'smoketest']) + self.cli_set(['firewall', 'interface', 'eth0', 'in', 'name', 'smoketest']) self.cli_commit() @@ -174,83 +172,182 @@ class TestFirewall(VyOSUnitTestSHIM.TestCase): ['elements = { 53, 123 }'] ] - self.verify_nftables(nftables_search, 'ip filter') + self.verify_nftables(nftables_search, 'ip vyos_filter') - def test_basic_rules(self): + def test_ipv4_basic_rules(self): + name = 'smoketest' + interface = 'eth0' mss_range = '501-1460' - self.cli_set(['firewall', 'name', 'smoketest', 'default-action', 'drop']) - self.cli_set(['firewall', 'name', 'smoketest', 'enable-default-log']) - self.cli_set(['firewall', 'name', 'smoketest', 'rule', '1', 'action', 'accept']) - self.cli_set(['firewall', 'name', 'smoketest', 'rule', '1', 'source', 'address', '172.16.20.10']) - self.cli_set(['firewall', 'name', 'smoketest', 'rule', '1', 'destination', 'address', '172.16.10.10']) - self.cli_set(['firewall', 'name', 'smoketest', 'rule', '1', 'log', 'enable']) - self.cli_set(['firewall', 'name', 'smoketest', 'rule', '1', 'log-level', 'debug']) - self.cli_set(['firewall', 'name', 'smoketest', 'rule', '1', 'ttl', 'eq', '15']) - self.cli_set(['firewall', 'name', 'smoketest', 'rule', '2', 'action', 'reject']) - self.cli_set(['firewall', 'name', 'smoketest', 'rule', '2', 'protocol', 'tcp']) - self.cli_set(['firewall', 'name', 'smoketest', 'rule', '2', 'destination', 'port', '8888']) - self.cli_set(['firewall', 'name', 'smoketest', 'rule', '2', 'log', 'enable']) - self.cli_set(['firewall', 'name', 'smoketest', 'rule', '2', 'log-level', 'err']) - self.cli_set(['firewall', 'name', 'smoketest', 'rule', '2', 'tcp', 'flags', 'syn']) - self.cli_set(['firewall', 'name', 'smoketest', 'rule', '2', 'tcp', 'flags', 'not', 'ack']) - self.cli_set(['firewall', 'name', 'smoketest', 'rule', '2', 'ttl', 'gt', '102']) - self.cli_set(['firewall', 'name', 'smoketest', 'rule', '3', 'action', 'accept']) - self.cli_set(['firewall', 'name', 'smoketest', 'rule', '3', 'protocol', 'tcp']) - self.cli_set(['firewall', 'name', 'smoketest', 'rule', '3', 'destination', 'port', '22']) - self.cli_set(['firewall', 'name', 'smoketest', 'rule', '3', 'limit', 'rate', '5/minute']) - self.cli_set(['firewall', 'name', 'smoketest', 'rule', '3', 'log', 'disable']) - self.cli_set(['firewall', 'name', 'smoketest', 'rule', '4', 'action', 'drop']) - self.cli_set(['firewall', 'name', 'smoketest', 'rule', '4', 'protocol', 'tcp']) - self.cli_set(['firewall', 'name', 'smoketest', 'rule', '4', 'destination', 'port', '22']) - self.cli_set(['firewall', 'name', 'smoketest', 'rule', '4', 'recent', 'count', '10']) - self.cli_set(['firewall', 'name', 'smoketest', 'rule', '4', 'recent', 'time', 'minute']) - self.cli_set(['firewall', 'name', 'smoketest', 'rule', '5', 'action', 'accept']) - self.cli_set(['firewall', 'name', 'smoketest', 'rule', '5', 'protocol', 'tcp']) - self.cli_set(['firewall', 'name', 'smoketest', 'rule', '5', 'tcp', 'flags', 'syn']) - self.cli_set(['firewall', 'name', 'smoketest', 'rule', '5', 'tcp', 'mss', mss_range]) - - self.cli_set(['interfaces', 'ethernet', 'eth0', 'firewall', 'in', 'name', 'smoketest']) + + self.cli_set(['firewall', 'name', name, 'default-action', 'drop']) + self.cli_set(['firewall', 'name', name, 'enable-default-log']) + self.cli_set(['firewall', 'name', name, 'rule', '1', 'action', 'accept']) + self.cli_set(['firewall', 'name', name, 'rule', '1', 'source', 'address', '172.16.20.10']) + self.cli_set(['firewall', 'name', name, 'rule', '1', 'destination', 'address', '172.16.10.10']) + self.cli_set(['firewall', 'name', name, 'rule', '1', 'log', 'enable']) + self.cli_set(['firewall', 'name', name, 'rule', '1', 'log-level', 'debug']) + self.cli_set(['firewall', 'name', name, 'rule', '1', 'ttl', 'eq', '15']) + self.cli_set(['firewall', 'name', name, 'rule', '2', 'action', 'reject']) + self.cli_set(['firewall', 'name', name, 'rule', '2', 'protocol', 'tcp']) + self.cli_set(['firewall', 'name', name, 'rule', '2', 'destination', 'port', '8888']) + self.cli_set(['firewall', 'name', name, 'rule', '2', 'log', 'enable']) + self.cli_set(['firewall', 'name', name, 'rule', '2', 'log-level', 'err']) + self.cli_set(['firewall', 'name', name, 'rule', '2', 'tcp', 'flags', 'syn']) + self.cli_set(['firewall', 'name', name, 'rule', '2', 'tcp', 'flags', 'not', 'ack']) + self.cli_set(['firewall', 'name', name, 'rule', '2', 'ttl', 'gt', '102']) + self.cli_set(['firewall', 'name', name, 'rule', '3', 'action', 'accept']) + self.cli_set(['firewall', 'name', name, 'rule', '3', 'protocol', 'tcp']) + self.cli_set(['firewall', 'name', name, 'rule', '3', 'destination', 'port', '22']) + self.cli_set(['firewall', 'name', name, 'rule', '3', 'limit', 'rate', '5/minute']) + self.cli_set(['firewall', 'name', name, 'rule', '3', 'log', 'disable']) + self.cli_set(['firewall', 'name', name, 'rule', '4', 'action', 'drop']) + self.cli_set(['firewall', 'name', name, 'rule', '4', 'protocol', 'tcp']) + self.cli_set(['firewall', 'name', name, 'rule', '4', 'destination', 'port', '22']) + self.cli_set(['firewall', 'name', name, 'rule', '4', 'recent', 'count', '10']) + self.cli_set(['firewall', 'name', name, 'rule', '4', 'recent', 'time', 'minute']) + self.cli_set(['firewall', 'name', name, 'rule', '5', 'action', 'accept']) + self.cli_set(['firewall', 'name', name, 'rule', '5', 'protocol', 'tcp']) + self.cli_set(['firewall', 'name', name, 'rule', '5', 'tcp', 'flags', 'syn']) + self.cli_set(['firewall', 'name', name, 'rule', '5', 'tcp', 'mss', mss_range]) + self.cli_set(['firewall', 'name', name, 'rule', '6', 'action', 'return']) + self.cli_set(['firewall', 'name', name, 'rule', '6', 'protocol', 'gre']) + + self.cli_set(['firewall', 'interface', interface, 'in', 'name', name]) self.cli_commit() nftables_search = [ - ['iifname "eth0"', 'jump NAME_smoketest'], - ['saddr 172.16.20.10', 'daddr 172.16.10.10', 'log prefix "[smoketest-1-A]" level debug', 'ip ttl 15','return'], - ['tcp flags & (syn | ack) == syn', 'tcp dport { 8888 }', 'log prefix "[smoketest-2-R]" level err', 'ip ttl > 102', 'reject'], - ['tcp dport { 22 }', 'limit rate 5/minute', 'return'], + [f'iifname "{interface}"', f'jump NAME_{name}'], + ['saddr 172.16.20.10', 'daddr 172.16.10.10', 'log prefix "[smoketest-1-A]" level debug', 'ip ttl 15', 'return'], + ['tcp flags syn / syn,ack', 'tcp dport 8888', 'log prefix "[smoketest-2-R]" level err', 'ip ttl > 102', 'reject'], + ['tcp dport 22', 'limit rate 5/minute', 'return'], ['log prefix "[smoketest-default-D]"','smoketest default-action', 'drop'], - ['tcp dport { 22 }', 'add @RECENT_smoketest_4 { ip saddr limit rate over 10/minute burst 10 packets }', 'drop'], - [f'tcp flags & syn == syn tcp option maxseg size {mss_range}'] + ['tcp dport 22', 'add @RECENT_smoketest_4 { ip saddr limit rate over 10/minute burst 10 packets }', 'drop'], + ['tcp flags & syn == syn', f'tcp option maxseg size {mss_range}'], + ['meta l4proto gre', 'return'] + ] + + self.verify_nftables(nftables_search, 'ip vyos_filter') + + def test_ipv4_advanced(self): + name = 'smoketest-adv' + name2 = 'smoketest-adv2' + interface = 'eth0' + + self.cli_set(['firewall', 'name', name, 'default-action', 'drop']) + self.cli_set(['firewall', 'name', name, 'enable-default-log']) + + self.cli_set(['firewall', 'name', name, 'rule', '6', 'action', 'accept']) + self.cli_set(['firewall', 'name', name, 'rule', '6', 'packet-length', '64']) + self.cli_set(['firewall', 'name', name, 'rule', '6', 'packet-length', '512']) + self.cli_set(['firewall', 'name', name, 'rule', '6', 'packet-length', '1024']) + self.cli_set(['firewall', 'name', name, 'rule', '6', 'dscp', '17']) + self.cli_set(['firewall', 'name', name, 'rule', '6', 'dscp', '52']) + + self.cli_set(['firewall', 'name', name, 'rule', '7', 'action', 'accept']) + self.cli_set(['firewall', 'name', name, 'rule', '7', 'packet-length', '1-30000']) + self.cli_set(['firewall', 'name', name, 'rule', '7', 'packet-length-exclude', '60000-65535']) + self.cli_set(['firewall', 'name', name, 'rule', '7', 'dscp', '3-11']) + self.cli_set(['firewall', 'name', name, 'rule', '7', 'dscp-exclude', '21-25']) + + self.cli_set(['firewall', 'name', name2, 'default-action', 'jump']) + self.cli_set(['firewall', 'name', name2, 'default-jump-target', name]) + self.cli_set(['firewall', 'name', name2, 'enable-default-log']) + self.cli_set(['firewall', 'name', name2, 'rule', '1', 'source', 'address', '198.51.100.1']) + self.cli_set(['firewall', 'name', name2, 'rule', '1', 'action', 'jump']) + self.cli_set(['firewall', 'name', name2, 'rule', '1', 'jump-target', name]) + + self.cli_set(['firewall', 'interface', interface, 'in', 'name', name]) + + self.cli_commit() + + nftables_search = [ + [f'iifname "{interface}"', f'jump NAME_{name}'], + ['ip length { 64, 512, 1024 }', 'ip dscp { 0x11, 0x34 }', 'return'], + ['ip length 1-30000', 'ip length != 60000-65535', 'ip dscp 0x03-0x0b', 'ip dscp != 0x15-0x19', 'return'], + [f'log prefix "[{name}-default-D]"', 'drop'], + ['ip saddr 198.51.100.1', f'jump NAME_{name}'], + [f'log prefix "[{name2}-default-J]"', f'jump NAME_{name}'] ] - self.verify_nftables(nftables_search, 'ip filter') + self.verify_nftables(nftables_search, 'ip vyos_filter') - def test_basic_rules_ipv6(self): - self.cli_set(['firewall', 'ipv6-name', 'v6-smoketest', 'default-action', 'drop']) - self.cli_set(['firewall', 'ipv6-name', 'v6-smoketest', 'enable-default-log']) + def test_ipv6_basic_rules(self): + name = 'v6-smoketest' + interface = 'eth0' - self.cli_set(['firewall', 'ipv6-name', 'v6-smoketest', 'rule', '1', 'action', 'accept']) - self.cli_set(['firewall', 'ipv6-name', 'v6-smoketest', 'rule', '1', 'source', 'address', '2002::1']) - self.cli_set(['firewall', 'ipv6-name', 'v6-smoketest', 'rule', '1', 'destination', 'address', '2002::1:1']) - self.cli_set(['firewall', 'ipv6-name', 'v6-smoketest', 'rule', '1', 'log', 'enable']) - self.cli_set(['firewall', 'ipv6-name', 'v6-smoketest', 'rule', '1', 'log-level', 'crit']) + self.cli_set(['firewall', 'ipv6-name', name, 'default-action', 'drop']) + self.cli_set(['firewall', 'ipv6-name', name, 'enable-default-log']) - self.cli_set(['firewall', 'ipv6-name', 'v6-smoketest', 'rule', '2', 'action', 'reject']) - self.cli_set(['firewall', 'ipv6-name', 'v6-smoketest', 'rule', '2', 'protocol', 'tcp_udp']) - self.cli_set(['firewall', 'ipv6-name', 'v6-smoketest', 'rule', '2', 'destination', 'port', '8888']) + self.cli_set(['firewall', 'ipv6-name', name, 'rule', '1', 'action', 'accept']) + self.cli_set(['firewall', 'ipv6-name', name, 'rule', '1', 'source', 'address', '2002::1']) + self.cli_set(['firewall', 'ipv6-name', name, 'rule', '1', 'destination', 'address', '2002::1:1']) + self.cli_set(['firewall', 'ipv6-name', name, 'rule', '1', 'log', 'enable']) + self.cli_set(['firewall', 'ipv6-name', name, 'rule', '1', 'log-level', 'crit']) - self.cli_set(['interfaces', 'ethernet', 'eth0', 'firewall', 'in', 'ipv6-name', 'v6-smoketest']) + self.cli_set(['firewall', 'ipv6-name', name, 'rule', '2', 'action', 'reject']) + self.cli_set(['firewall', 'ipv6-name', name, 'rule', '2', 'protocol', 'tcp_udp']) + self.cli_set(['firewall', 'ipv6-name', name, 'rule', '2', 'destination', 'port', '8888']) + + self.cli_set(['firewall', 'ipv6-name', name, 'rule', '3', 'action', 'return']) + self.cli_set(['firewall', 'ipv6-name', name, 'rule', '3', 'protocol', 'gre']) + + self.cli_set(['firewall', 'interface', interface, 'in', 'ipv6-name', name]) self.cli_commit() nftables_search = [ - ['iifname "eth0"', 'jump NAME6_v6-smoketest'], + [f'iifname "{interface}"', f'jump NAME6_{name}'], ['saddr 2002::1', 'daddr 2002::1:1', 'log prefix "[v6-smoketest-1-A]" level crit', 'return'], - ['meta l4proto { tcp, udp }', 'th dport { 8888 }', 'reject'], - ['smoketest default-action', 'log prefix "[v6-smoketest-default-D]"', 'drop'] + ['meta l4proto { tcp, udp }', 'th dport 8888', 'reject'], + ['smoketest default-action', f'log prefix "[{name}-default-D]"', 'drop'], + ['meta l4proto gre', 'return'] + ] + + self.verify_nftables(nftables_search, 'ip6 vyos_filter') + + def test_ipv6_advanced(self): + name = 'v6-smoketest-adv' + name2 = 'v6-smoketest-adv2' + interface = 'eth0' + + self.cli_set(['firewall', 'ipv6-name', name, 'default-action', 'drop']) + self.cli_set(['firewall', 'ipv6-name', name, 'enable-default-log']) + + self.cli_set(['firewall', 'ipv6-name', name, 'rule', '3', 'action', 'accept']) + self.cli_set(['firewall', 'ipv6-name', name, 'rule', '3', 'packet-length', '65']) + self.cli_set(['firewall', 'ipv6-name', name, 'rule', '3', 'packet-length', '513']) + self.cli_set(['firewall', 'ipv6-name', name, 'rule', '3', 'packet-length', '1025']) + self.cli_set(['firewall', 'ipv6-name', name, 'rule', '3', 'dscp', '18']) + self.cli_set(['firewall', 'ipv6-name', name, 'rule', '3', 'dscp', '53']) + + self.cli_set(['firewall', 'ipv6-name', name, 'rule', '4', 'action', 'accept']) + self.cli_set(['firewall', 'ipv6-name', name, 'rule', '4', 'packet-length', '1-1999']) + self.cli_set(['firewall', 'ipv6-name', name, 'rule', '4', 'packet-length-exclude', '60000-65535']) + self.cli_set(['firewall', 'ipv6-name', name, 'rule', '4', 'dscp', '4-14']) + self.cli_set(['firewall', 'ipv6-name', name, 'rule', '4', 'dscp-exclude', '31-35']) + + self.cli_set(['firewall', 'ipv6-name', name2, 'default-action', 'jump']) + self.cli_set(['firewall', 'ipv6-name', name2, 'default-jump-target', name]) + self.cli_set(['firewall', 'ipv6-name', name2, 'enable-default-log']) + self.cli_set(['firewall', 'ipv6-name', name2, 'rule', '1', 'source', 'address', '2001:db8::/64']) + self.cli_set(['firewall', 'ipv6-name', name2, 'rule', '1', 'action', 'jump']) + self.cli_set(['firewall', 'ipv6-name', name2, 'rule', '1', 'jump-target', name]) + + self.cli_set(['firewall', 'interface', interface, 'in', 'ipv6-name', name]) + + self.cli_commit() + + nftables_search = [ + [f'iifname "{interface}"', f'jump NAME6_{name}'], + ['ip6 length { 65, 513, 1025 }', 'ip6 dscp { af21, 0x35 }', 'return'], + ['ip6 length 1-1999', 'ip6 length != 60000-65535', 'ip6 dscp 0x04-0x0e', 'ip6 dscp != 0x1f-0x23', 'return'], + [f'log prefix "[{name}-default-D]"', 'drop'], + ['ip6 saddr 2001:db8::/64', f'jump NAME6_{name}'], + [f'log prefix "[{name2}-default-J]"', f'jump NAME6_{name}'] ] - self.verify_nftables(nftables_search, 'ip6 filter') + self.verify_nftables(nftables_search, 'ip6 vyos_filter') def test_state_policy(self): self.cli_set(['firewall', 'state-policy', 'established', 'action', 'accept']) @@ -260,45 +357,48 @@ class TestFirewall(VyOSUnitTestSHIM.TestCase): self.cli_commit() chains = { - 'ip filter': ['VYOS_FW_FORWARD', 'VYOS_FW_OUTPUT', 'VYOS_FW_LOCAL'], - 'ip6 filter': ['VYOS_FW6_FORWARD', 'VYOS_FW6_OUTPUT', 'VYOS_FW6_LOCAL'] + 'ip vyos_filter': ['VYOS_FW_FORWARD', 'VYOS_FW_OUTPUT', 'VYOS_FW_LOCAL'], + 'ip6 vyos_filter': ['VYOS_FW6_FORWARD', 'VYOS_FW6_OUTPUT', 'VYOS_FW6_LOCAL'] } - for table in ['ip filter', 'ip6 filter']: + for table in ['ip vyos_filter', 'ip6 vyos_filter']: for chain in chains[table]: nftables_output = cmd(f'sudo nft list chain {table} {chain}') self.assertTrue('jump VYOS_STATE_POLICY' in nftables_output) - def test_state_and_status_rules(self): - self.cli_set(['firewall', 'name', 'smoketest', 'default-action', 'drop']) - self.cli_set(['firewall', 'name', 'smoketest', 'rule', '1', 'action', 'accept']) - self.cli_set(['firewall', 'name', 'smoketest', 'rule', '1', 'state', 'established', 'enable']) - self.cli_set(['firewall', 'name', 'smoketest', 'rule', '1', 'state', 'related', 'enable']) - self.cli_set(['firewall', 'name', 'smoketest', 'rule', '2', 'action', 'reject']) - self.cli_set(['firewall', 'name', 'smoketest', 'rule', '2', 'state', 'invalid', 'enable']) - self.cli_set(['firewall', 'name', 'smoketest', 'rule', '3', 'action', 'accept']) - self.cli_set(['firewall', 'name', 'smoketest', 'rule', '3', 'state', 'new', 'enable']) + def test_ipv4_state_and_status_rules(self): + name = 'smoketest-state' + interface = 'eth0' + + self.cli_set(['firewall', 'name', name, 'default-action', 'drop']) + self.cli_set(['firewall', 'name', name, 'rule', '1', 'action', 'accept']) + self.cli_set(['firewall', 'name', name, 'rule', '1', 'state', 'established', 'enable']) + self.cli_set(['firewall', 'name', name, 'rule', '1', 'state', 'related', 'enable']) + self.cli_set(['firewall', 'name', name, 'rule', '2', 'action', 'reject']) + self.cli_set(['firewall', 'name', name, 'rule', '2', 'state', 'invalid', 'enable']) + self.cli_set(['firewall', 'name', name, 'rule', '3', 'action', 'accept']) + self.cli_set(['firewall', 'name', name, 'rule', '3', 'state', 'new', 'enable']) - self.cli_set(['firewall', 'name', 'smoketest', 'rule', '3', 'connection-status', 'nat', 'destination']) - self.cli_set(['firewall', 'name', 'smoketest', 'rule', '4', 'action', 'accept']) - self.cli_set(['firewall', 'name', 'smoketest', 'rule', '4', 'state', 'new', 'enable']) - self.cli_set(['firewall', 'name', 'smoketest', 'rule', '4', 'state', 'established', 'enable']) - self.cli_set(['firewall', 'name', 'smoketest', 'rule', '4', 'connection-status', 'nat', 'source']) + self.cli_set(['firewall', 'name', name, 'rule', '3', 'connection-status', 'nat', 'destination']) + self.cli_set(['firewall', 'name', name, 'rule', '4', 'action', 'accept']) + self.cli_set(['firewall', 'name', name, 'rule', '4', 'state', 'new', 'enable']) + self.cli_set(['firewall', 'name', name, 'rule', '4', 'state', 'established', 'enable']) + self.cli_set(['firewall', 'name', name, 'rule', '4', 'connection-status', 'nat', 'source']) - self.cli_set(['interfaces', 'ethernet', 'eth0', 'firewall', 'in', 'name', 'smoketest']) + self.cli_set(['firewall', 'interface', interface, 'in', 'name', name]) self.cli_commit() nftables_search = [ - ['iifname "eth0"', 'jump NAME_smoketest'], + [f'iifname "{interface}"', f'jump NAME_{name}'], ['ct state { established, related }', 'return'], - ['ct state { invalid }', 'reject'], - ['ct state { new }', 'ct status { dnat }', 'return'], - ['ct state { established, new }', 'ct status { snat }', 'return'], - ['smoketest default-action', 'drop'] + ['ct state invalid', 'reject'], + ['ct state new', 'ct status dnat', 'return'], + ['ct state { established, new }', 'ct status snat', 'return'], + ['drop', f'comment "{name} default-action drop"'] ] - self.verify_nftables(nftables_search, 'ip filter') + self.verify_nftables(nftables_search, 'ip vyos_filter') def test_sysfs(self): for name, conf in sysfs_config.items(): @@ -317,5 +417,35 @@ class TestFirewall(VyOSUnitTestSHIM.TestCase): with open(path, 'r') as f: self.assertNotEqual(f.read().strip(), conf['default'], msg=path) + def test_zone_basic(self): + self.cli_set(['firewall', 'name', 'smoketest', 'default-action', 'drop']) + self.cli_set(['firewall', 'zone', 'smoketest-eth0', 'interface', 'eth0']) + self.cli_set(['firewall', 'zone', 'smoketest-eth0', 'from', 'smoketest-local', 'firewall', 'name', 'smoketest']) + self.cli_set(['firewall', 'zone', 'smoketest-local', 'local-zone']) + self.cli_set(['firewall', 'zone', 'smoketest-local', 'from', 'smoketest-eth0', 'firewall', 'name', 'smoketest']) + + self.cli_commit() + + nftables_search = [ + ['chain VZONE_smoketest-eth0'], + ['chain VZONE_smoketest-local_IN'], + ['chain VZONE_smoketest-local_OUT'], + ['oifname "eth0"', 'jump VZONE_smoketest-eth0'], + ['jump VZONE_smoketest-local_IN'], + ['jump VZONE_smoketest-local_OUT'], + ['iifname "eth0"', 'jump NAME_smoketest'], + ['oifname "eth0"', 'jump NAME_smoketest'] + ] + + nftables_output = cmd('sudo nft list table ip vyos_filter') + + for search in nftables_search: + matched = False + for line in nftables_output.split("\n"): + if all(item in line for item in search): + matched = True + break + self.assertTrue(matched) + if __name__ == '__main__': unittest.main(verbosity=2) diff --git a/smoketest/scripts/cli/test_interfaces_ethernet.py b/smoketest/scripts/cli/test_interfaces_ethernet.py index 05d2ae5f5..5049bd5b0 100755 --- a/smoketest/scripts/cli/test_interfaces_ethernet.py +++ b/smoketest/scripts/cli/test_interfaces_ethernet.py @@ -17,6 +17,7 @@ import os import re import unittest +from glob import glob from netifaces import AF_INET from netifaces import AF_INET6 @@ -185,6 +186,43 @@ class EthernetInterfaceTest(BasicInterfaceTest.TestCase): self.assertEqual(f'{cpus:x}', f'{rps_cpus:x}') + def test_offloading_rfs(self): + global_rfs_flow = 32768 + rfs_flow = global_rfs_flow + + for interface in self._interfaces: + self.cli_set(self._base_path + [interface, 'offload', 'rfs']) + + self.cli_commit() + + for interface in self._interfaces: + queues = len(glob(f'/sys/class/net/{interface}/queues/rx-*')) + rfs_flow = int(global_rfs_flow/queues) + for i in range(0, queues): + tmp = read_file(f'/sys/class/net/{interface}/queues/rx-{i}/rps_flow_cnt') + self.assertEqual(int(tmp), rfs_flow) + + tmp = read_file(f'/proc/sys/net/core/rps_sock_flow_entries') + self.assertEqual(int(tmp), global_rfs_flow) + + + # delete configuration of RFS and check all values returned to default "0" + for interface in self._interfaces: + self.cli_delete(self._base_path + [interface, 'offload', 'rfs']) + + self.cli_commit() + + for interface in self._interfaces: + queues = len(glob(f'/sys/class/net/{interface}/queues/rx-*')) + rfs_flow = int(global_rfs_flow/queues) + for i in range(0, queues): + tmp = read_file(f'/sys/class/net/{interface}/queues/rx-{i}/rps_flow_cnt') + self.assertEqual(int(tmp), 0) + + tmp = read_file(f'/proc/sys/net/core/rps_sock_flow_entries') + self.assertEqual(int(tmp), 0) + + def test_non_existing_interface(self): unknonw_interface = self._base_path + ['eth667'] self.cli_set(unknonw_interface) diff --git a/smoketest/scripts/cli/test_load_balancning_wan.py b/smoketest/scripts/cli/test_load_balancning_wan.py index 303dece86..23020b9b1 100755 --- a/smoketest/scripts/cli/test_load_balancning_wan.py +++ b/smoketest/scripts/cli/test_load_balancning_wan.py @@ -177,6 +177,7 @@ class TestLoadBalancingWan(VyOSUnitTestSHIM.TestCase): }""" nat_vyos_pre_snat_hook = """table ip nat { chain VYOS_PRE_SNAT_HOOK { + type nat hook postrouting priority srcnat - 1; policy accept; counter jump WANLOADBALANCE return } diff --git a/smoketest/scripts/cli/test_nat.py b/smoketest/scripts/cli/test_nat.py index 408facfb3..f824838c0 100755 --- a/smoketest/scripts/cli/test_nat.py +++ b/smoketest/scripts/cli/test_nat.py @@ -26,6 +26,7 @@ from vyos.util import dict_search base_path = ['nat'] src_path = base_path + ['source'] dst_path = base_path + ['destination'] +static_path = base_path + ['static'] class TestNAT(VyOSUnitTestSHIM.TestCase): @classmethod @@ -40,10 +41,24 @@ class TestNAT(VyOSUnitTestSHIM.TestCase): self.cli_delete(base_path) self.cli_commit() + def verify_nftables(self, nftables_search, table, inverse=False, args=''): + nftables_output = cmd(f'sudo nft {args} list table {table}') + + for search in nftables_search: + matched = False + for line in nftables_output.split("\n"): + if all(item in line for item in search): + matched = True + break + self.assertTrue(not matched if inverse else matched, msg=search) + def test_snat(self): rules = ['100', '110', '120', '130', '200', '210', '220', '230'] outbound_iface_100 = 'eth0' outbound_iface_200 = 'eth1' + + nftables_search = ['jump VYOS_PRE_SNAT_HOOK'] + for rule in rules: network = f'192.168.{rule}.0/24' # depending of rule order we check either for source address for NAT @@ -52,51 +67,16 @@ class TestNAT(VyOSUnitTestSHIM.TestCase): self.cli_set(src_path + ['rule', rule, 'source', 'address', network]) self.cli_set(src_path + ['rule', rule, 'outbound-interface', outbound_iface_100]) self.cli_set(src_path + ['rule', rule, 'translation', 'address', 'masquerade']) + nftables_search.append([f'saddr {network}', f'oifname "{outbound_iface_100}"', 'masquerade']) else: self.cli_set(src_path + ['rule', rule, 'destination', 'address', network]) self.cli_set(src_path + ['rule', rule, 'outbound-interface', outbound_iface_200]) self.cli_set(src_path + ['rule', rule, 'exclude']) + nftables_search.append([f'daddr {network}', f'oifname "{outbound_iface_200}"', 'return']) self.cli_commit() - tmp = cmd('sudo nft -j list chain ip nat POSTROUTING') - data_json = jmespath.search('nftables[?rule].rule[?chain]', json.loads(tmp)) - - for idx in range(0, len(data_json)): - data = data_json[idx] - if idx == 0: - self.assertEqual(data['chain'], 'POSTROUTING') - self.assertEqual(data['family'], 'ip') - self.assertEqual(data['table'], 'nat') - - jump_target = dict_search('jump.target', data['expr'][1]) - self.assertEqual(jump_target,'VYOS_PRE_SNAT_HOOK') - else: - rule = str(rules[idx - 1]) - network = f'192.168.{rule}.0/24' - - self.assertEqual(data['chain'], 'POSTROUTING') - self.assertEqual(data['comment'], f'SRC-NAT-{rule}') - self.assertEqual(data['family'], 'ip') - self.assertEqual(data['table'], 'nat') - - iface = dict_search('match.right', data['expr'][0]) - direction = dict_search('match.left.payload.field', data['expr'][1]) - address = dict_search('match.right.prefix.addr', data['expr'][1]) - mask = dict_search('match.right.prefix.len', data['expr'][1]) - - if int(rule) < 200: - self.assertEqual(direction, 'saddr') - self.assertEqual(iface, outbound_iface_100) - # check for masquerade keyword - self.assertIn('masquerade', data['expr'][3]) - else: - self.assertEqual(direction, 'daddr') - self.assertEqual(iface, outbound_iface_200) - # check for return keyword due to 'exclude' - self.assertIn('return', data['expr'][3]) - - self.assertEqual(f'{address}/{mask}', network) + self.verify_nftables(nftables_search, 'ip vyos_nat') def test_dnat(self): rules = ['100', '110', '120', '130', '200', '210', '220', '230'] @@ -105,56 +85,29 @@ class TestNAT(VyOSUnitTestSHIM.TestCase): inbound_proto_100 = 'udp' inbound_proto_200 = 'tcp' + nftables_search = ['jump VYOS_PRE_DNAT_HOOK'] + for rule in rules: port = f'10{rule}' self.cli_set(dst_path + ['rule', rule, 'source', 'port', port]) self.cli_set(dst_path + ['rule', rule, 'translation', 'address', '192.0.2.1']) self.cli_set(dst_path + ['rule', rule, 'translation', 'port', port]) + rule_search = [f'dnat to 192.0.2.1:{port}'] if int(rule) < 200: self.cli_set(dst_path + ['rule', rule, 'protocol', inbound_proto_100]) self.cli_set(dst_path + ['rule', rule, 'inbound-interface', inbound_iface_100]) + rule_search.append(f'{inbound_proto_100} sport {port}') + rule_search.append(f'iifname "{inbound_iface_100}"') else: self.cli_set(dst_path + ['rule', rule, 'protocol', inbound_proto_200]) self.cli_set(dst_path + ['rule', rule, 'inbound-interface', inbound_iface_200]) + rule_search.append(f'iifname "{inbound_iface_200}"') - self.cli_commit() - - tmp = cmd('sudo nft -j list chain ip nat PREROUTING') - data_json = jmespath.search('nftables[?rule].rule[?chain]', json.loads(tmp)) - - for idx in range(0, len(data_json)): - data = data_json[idx] - if idx == 0: - self.assertEqual(data['chain'], 'PREROUTING') - self.assertEqual(data['family'], 'ip') - self.assertEqual(data['table'], 'nat') + nftables_search.append(rule_search) - jump_target = dict_search('jump.target', data['expr'][1]) - self.assertEqual(jump_target,'VYOS_PRE_DNAT_HOOK') - else: + self.cli_commit() - rule = str(rules[idx - 1]) - port = int(f'10{rule}') - - self.assertEqual(data['chain'], 'PREROUTING') - self.assertEqual(data['comment'].split()[0], f'DST-NAT-{rule}') - self.assertEqual(data['family'], 'ip') - self.assertEqual(data['table'], 'nat') - - iface = dict_search('match.right', data['expr'][0]) - direction = dict_search('match.left.payload.field', data['expr'][1]) - protocol = dict_search('match.left.payload.protocol', data['expr'][1]) - dnat_addr = dict_search('dnat.addr', data['expr'][3]) - dnat_port = dict_search('dnat.port', data['expr'][3]) - - self.assertEqual(direction, 'sport') - self.assertEqual(dnat_addr, '192.0.2.1') - self.assertEqual(dnat_port, port) - if int(rule) < 200: - self.assertEqual(iface, inbound_iface_100) - self.assertEqual(protocol, inbound_proto_100) - else: - self.assertEqual(iface, inbound_iface_200) + self.verify_nftables(nftables_search, 'ip vyos_nat') def test_snat_required_translation_address(self): # T2813: Ensure translation address is specified @@ -193,8 +146,48 @@ class TestNAT(VyOSUnitTestSHIM.TestCase): # without any rule self.cli_set(src_path) self.cli_set(dst_path) + self.cli_set(static_path) + self.cli_commit() + + def test_dnat_without_translation_address(self): + self.cli_set(dst_path + ['rule', '1', 'inbound-interface', 'eth1']) + self.cli_set(dst_path + ['rule', '1', 'destination', 'port', '443']) + self.cli_set(dst_path + ['rule', '1', 'protocol', 'tcp']) + self.cli_set(dst_path + ['rule', '1', 'translation', 'port', '443']) + + self.cli_commit() + + nftables_search = [ + ['iifname "eth1"', 'tcp dport 443', 'dnat to :443'] + ] + + self.verify_nftables(nftables_search, 'ip vyos_nat') + + def test_static_nat(self): + dst_addr_1 = '10.0.1.1' + translate_addr_1 = '192.168.1.1' + dst_addr_2 = '203.0.113.0/24' + translate_addr_2 = '192.0.2.0/24' + ifname = 'eth0' + + self.cli_set(static_path + ['rule', '10', 'destination', 'address', dst_addr_1]) + self.cli_set(static_path + ['rule', '10', 'inbound-interface', ifname]) + self.cli_set(static_path + ['rule', '10', 'translation', 'address', translate_addr_1]) + + self.cli_set(static_path + ['rule', '20', 'destination', 'address', dst_addr_2]) + self.cli_set(static_path + ['rule', '20', 'inbound-interface', ifname]) + self.cli_set(static_path + ['rule', '20', 'translation', 'address', translate_addr_2]) + self.cli_commit() + nftables_search = [ + [f'iifname "{ifname}"', f'ip daddr {dst_addr_1}', f'dnat to {translate_addr_1}'], + [f'oifname "{ifname}"', f'ip saddr {translate_addr_1}', f'snat to {dst_addr_1}'], + [f'iifname "{ifname}"', f'dnat ip prefix to ip daddr map {{ {dst_addr_2} : {translate_addr_2} }}'], + [f'oifname "{ifname}"', f'snat ip prefix to ip saddr map {{ {translate_addr_2} : {dst_addr_2} }}'] + ] + + self.verify_nftables(nftables_search, 'ip vyos_static_nat') if __name__ == '__main__': unittest.main(verbosity=2) diff --git a/smoketest/scripts/cli/test_nat66.py b/smoketest/scripts/cli/test_nat66.py index c5db066db..6cf7ca0a1 100755 --- a/smoketest/scripts/cli/test_nat66.py +++ b/smoketest/scripts/cli/test_nat66.py @@ -71,12 +71,12 @@ class TestNAT66(VyOSUnitTestSHIM.TestCase): self.cli_commit() nftables_search = [ - ['oifname "eth1"', 'ip6 saddr fc00::/64', 'snat prefix to fc01::/64'], - ['oifname "eth1"', 'ip6 saddr fc00::/64', 'masquerade'], - ['oifname "eth1"', 'ip6 saddr fc00::/64', 'return'] + ['oifname "eth1"', f'ip6 saddr {source_prefix}', f'snat prefix to {translation_prefix}'], + ['oifname "eth1"', f'ip6 saddr {source_prefix}', 'masquerade'], + ['oifname "eth1"', f'ip6 saddr {source_prefix}', 'return'] ] - self.verify_nftables(nftables_search, 'ip6 nat') + self.verify_nftables(nftables_search, 'ip6 vyos_nat') def test_source_nat66_address(self): source_prefix = 'fc00::/64' @@ -88,25 +88,11 @@ class TestNAT66(VyOSUnitTestSHIM.TestCase): # check validate() - outbound-interface must be defined self.cli_commit() - tmp = cmd('sudo nft -j list table ip6 nat') - data_json = jmespath.search('nftables[?rule].rule[?chain]', json.loads(tmp)) - - for idx in range(0, len(data_json)): - data = data_json[idx] - - self.assertEqual(data['chain'], 'POSTROUTING') - self.assertEqual(data['family'], 'ip6') - self.assertEqual(data['table'], 'nat') - - iface = dict_search('match.right', data['expr'][0]) - address = dict_search('match.right.prefix.addr', data['expr'][2]) - mask = dict_search('match.right.prefix.len', data['expr'][2]) - snat_address = dict_search('snat.addr', data['expr'][3]) + nftables_search = [ + ['oifname "eth1"', f'ip6 saddr {source_prefix}', f'snat to {translation_address}'] + ] - self.assertEqual(iface, 'eth1') - # check for translation address - self.assertEqual(snat_address, translation_address) - self.assertEqual(f'{address}/{mask}', source_prefix) + self.verify_nftables(nftables_search, 'ip6 vyos_nat') def test_destination_nat66(self): destination_address = 'fc00::1' @@ -129,7 +115,7 @@ class TestNAT66(VyOSUnitTestSHIM.TestCase): ['iifname "eth1"', 'ip6 saddr fc02::1', 'ip6 daddr fc00::1', 'return'] ] - self.verify_nftables(nftables_search, 'ip6 nat') + self.verify_nftables(nftables_search, 'ip6 vyos_nat') def test_destination_nat66_protocol(self): translation_address = '2001:db8:1111::1' @@ -150,10 +136,10 @@ class TestNAT66(VyOSUnitTestSHIM.TestCase): self.cli_commit() nftables_search = [ - ['iifname "eth1"', 'tcp dport { 4545 } ip6 saddr 2001:db8:2222::/64 tcp sport { 8080 } dnat to 2001:db8:1111::1:5555'] + ['iifname "eth1"', 'tcp dport 4545', 'ip6 saddr 2001:db8:2222::/64', 'tcp sport 8080', 'dnat to 2001:db8:1111::1:5555'] ] - self.verify_nftables(nftables_search, 'ip6 nat') + self.verify_nftables(nftables_search, 'ip6 vyos_nat') def test_destination_nat66_prefix(self): destination_prefix = 'fc00::/64' @@ -165,22 +151,25 @@ class TestNAT66(VyOSUnitTestSHIM.TestCase): # check validate() - outbound-interface must be defined self.cli_commit() - tmp = cmd('sudo nft -j list table ip6 nat') - data_json = jmespath.search('nftables[?rule].rule[?chain]', json.loads(tmp)) + nftables_search = [ + ['iifname "eth1"', f'ip6 daddr {destination_prefix}', f'dnat prefix to {translation_prefix}'] + ] + + self.verify_nftables(nftables_search, 'ip6 vyos_nat') - for idx in range(0, len(data_json)): - data = data_json[idx] + def test_destination_nat66_without_translation_address(self): + self.cli_set(dst_path + ['rule', '1', 'inbound-interface', 'eth1']) + self.cli_set(dst_path + ['rule', '1', 'destination', 'port', '443']) + self.cli_set(dst_path + ['rule', '1', 'protocol', 'tcp']) + self.cli_set(dst_path + ['rule', '1', 'translation', 'port', '443']) - self.assertEqual(data['chain'], 'PREROUTING') - self.assertEqual(data['family'], 'ip6') - self.assertEqual(data['table'], 'nat') + self.cli_commit() - iface = dict_search('match.right', data['expr'][0]) - translation_address = dict_search('dnat.addr.prefix.addr', data['expr'][3]) - translation_mask = dict_search('dnat.addr.prefix.len', data['expr'][3]) + nftables_search = [ + ['iifname "eth1"', 'tcp dport 443', 'dnat to :443'] + ] - self.assertEqual(f'{translation_address}/{translation_mask}', translation_prefix) - self.assertEqual(iface, 'eth1') + self.verify_nftables(nftables_search, 'ip6 vyos_nat') def test_source_nat66_required_translation_prefix(self): # T2813: Ensure translation address is specified @@ -219,10 +208,10 @@ class TestNAT66(VyOSUnitTestSHIM.TestCase): self.cli_commit() nftables_search = [ - ['oifname "eth1"', 'ip6 saddr 2001:db8:2222::/64 tcp dport { 9999 } tcp sport { 8080 } snat to 2001:db8:1111::1:80'] + ['oifname "eth1"', 'ip6 saddr 2001:db8:2222::/64', 'tcp dport 9999', 'tcp sport 8080', 'snat to 2001:db8:1111::1:80'] ] - self.verify_nftables(nftables_search, 'ip6 nat') + self.verify_nftables(nftables_search, 'ip6 vyos_nat') def test_nat66_no_rules(self): # T3206: deleting all rules but keep the direction 'destination' or diff --git a/smoketest/scripts/cli/test_policy_route.py b/smoketest/scripts/cli/test_policy_route.py index 534cfb082..046e385bb 100755 --- a/smoketest/scripts/cli/test_policy_route.py +++ b/smoketest/scripts/cli/test_policy_route.py @@ -53,7 +53,7 @@ class TestPolicyRoute(VyOSUnitTestSHIM.TestCase): ['chain VYOS_PBR_smoketest'] ] - self.verify_nftables(nftables_search, 'ip filter', inverse=True) + self.verify_nftables(nftables_search, 'ip mangle', inverse=True) def verify_nftables(self, nftables_search, table, inverse=False): nftables_output = cmd(f'sudo nft list table {table}') @@ -127,7 +127,7 @@ class TestPolicyRoute(VyOSUnitTestSHIM.TestCase): nftables_search = [ [f'iifname "{interface}"', 'jump VYOS_PBR_smoketest'], - ['tcp flags & (syn | ack) == syn', 'tcp dport { 8888 }', 'meta mark set ' + mark_hex] + ['tcp flags syn / syn,ack', 'tcp dport 8888', 'meta mark set ' + mark_hex] ] self.verify_nftables(nftables_search, 'ip mangle') @@ -136,7 +136,7 @@ class TestPolicyRoute(VyOSUnitTestSHIM.TestCase): nftables6_search = [ [f'iifname "{interface}"', 'jump VYOS_PBR6_smoketest'], - ['meta l4proto { tcp, udp }', 'th dport { 8888 }', 'meta mark set ' + mark_hex] + ['meta l4proto { tcp, udp }', 'th dport 8888', 'meta mark set ' + mark_hex] ] self.verify_nftables(nftables6_search, 'ip6 mangle') @@ -158,5 +158,81 @@ class TestPolicyRoute(VyOSUnitTestSHIM.TestCase): self.assertTrue(matched) + def test_pbr_matching_criteria(self): + self.cli_set(['policy', 'route', 'smoketest', 'rule', '1', 'protocol', 'udp']) + self.cli_set(['policy', 'route', 'smoketest', 'rule', '1', 'action', 'drop']) + self.cli_set(['policy', 'route', 'smoketest', 'rule', '2', 'protocol', 'tcp']) + self.cli_set(['policy', 'route', 'smoketest', 'rule', '2', 'tcp', 'flags', 'syn']) + self.cli_set(['policy', 'route', 'smoketest', 'rule', '2', 'tcp', 'flags', 'not', 'ack']) + self.cli_set(['policy', 'route', 'smoketest', 'rule', '2', 'set', 'table', table_id]) + self.cli_set(['policy', 'route', 'smoketest', 'rule', '3', 'source', 'address', '198.51.100.0/24']) + self.cli_set(['policy', 'route', 'smoketest', 'rule', '3', 'protocol', 'tcp']) + self.cli_set(['policy', 'route', 'smoketest', 'rule', '3', 'destination', 'port', '22']) + self.cli_set(['policy', 'route', 'smoketest', 'rule', '3', 'state', 'new', 'enable']) + self.cli_set(['policy', 'route', 'smoketest', 'rule', '3', 'ttl', 'gt', '2']) + self.cli_set(['policy', 'route', 'smoketest', 'rule', '3', 'set', 'table', table_id]) + self.cli_set(['policy', 'route', 'smoketest', 'rule', '4', 'protocol', 'icmp']) + self.cli_set(['policy', 'route', 'smoketest', 'rule', '4', 'icmp', 'type-name', 'echo-request']) + self.cli_set(['policy', 'route', 'smoketest', 'rule', '4', 'packet-length', '128']) + self.cli_set(['policy', 'route', 'smoketest', 'rule', '4', 'packet-length', '1024-2048']) + self.cli_set(['policy', 'route', 'smoketest', 'rule', '4', 'log', 'enable']) + self.cli_set(['policy', 'route', 'smoketest', 'rule', '4', 'set', 'table', table_id]) + self.cli_set(['policy', 'route', 'smoketest', 'rule', '5', 'dscp', '41']) + self.cli_set(['policy', 'route', 'smoketest', 'rule', '5', 'dscp', '57-59']) + self.cli_set(['policy', 'route', 'smoketest', 'rule', '5', 'set', 'table', table_id]) + + self.cli_set(['policy', 'route6', 'smoketest6', 'rule', '1', 'protocol', 'udp']) + self.cli_set(['policy', 'route6', 'smoketest6', 'rule', '1', 'action', 'drop']) + self.cli_set(['policy', 'route6', 'smoketest6', 'rule', '2', 'protocol', 'tcp']) + self.cli_set(['policy', 'route6', 'smoketest6', 'rule', '2', 'tcp', 'flags', 'syn']) + self.cli_set(['policy', 'route6', 'smoketest6', 'rule', '2', 'tcp', 'flags', 'not', 'ack']) + self.cli_set(['policy', 'route6', 'smoketest6', 'rule', '2', 'set', 'table', table_id]) + self.cli_set(['policy', 'route6', 'smoketest6', 'rule', '3', 'source', 'address', '2001:db8::0/64']) + self.cli_set(['policy', 'route6', 'smoketest6', 'rule', '3', 'protocol', 'tcp']) + self.cli_set(['policy', 'route6', 'smoketest6', 'rule', '3', 'destination', 'port', '22']) + self.cli_set(['policy', 'route6', 'smoketest6', 'rule', '3', 'state', 'new', 'enable']) + self.cli_set(['policy', 'route6', 'smoketest6', 'rule', '3', 'hop-limit', 'gt', '2']) + self.cli_set(['policy', 'route6', 'smoketest6', 'rule', '3', 'set', 'table', table_id]) + self.cli_set(['policy', 'route6', 'smoketest6', 'rule', '4', 'protocol', 'icmpv6']) + self.cli_set(['policy', 'route6', 'smoketest6', 'rule', '4', 'icmpv6', 'type', 'echo-request']) + self.cli_set(['policy', 'route6', 'smoketest6', 'rule', '4', 'packet-length-exclude', '128']) + self.cli_set(['policy', 'route6', 'smoketest6', 'rule', '4', 'packet-length-exclude', '1024-2048']) + self.cli_set(['policy', 'route6', 'smoketest6', 'rule', '4', 'log', 'enable']) + self.cli_set(['policy', 'route6', 'smoketest6', 'rule', '4', 'set', 'table', table_id]) + self.cli_set(['policy', 'route6', 'smoketest6', 'rule', '5', 'dscp-exclude', '61']) + self.cli_set(['policy', 'route6', 'smoketest6', 'rule', '5', 'dscp-exclude', '14-19']) + self.cli_set(['policy', 'route6', 'smoketest6', 'rule', '5', 'set', 'table', table_id]) + + self.cli_set(['interfaces', 'ethernet', interface, 'policy', 'route', 'smoketest']) + self.cli_set(['interfaces', 'ethernet', interface, 'policy', 'route6', 'smoketest6']) + + self.cli_commit() + + mark_hex = "{0:#010x}".format(table_mark_offset - int(table_id)) + + # IPv4 + nftables_search = [ + [f'iifname "{interface}"', 'jump VYOS_PBR_smoketest'], + ['meta l4proto udp', 'drop'], + ['tcp flags syn / syn,ack', 'meta mark set ' + mark_hex], + ['ct state new', 'tcp dport 22', 'ip saddr 198.51.100.0/24', 'ip ttl > 2', 'meta mark set ' + mark_hex], + ['meta l4proto icmp', 'log prefix "[smoketest-4-A]"', 'icmp type echo-request', 'ip length { 128, 1024-2048 }', 'meta mark set ' + mark_hex], + ['ip dscp { 0x29, 0x39-0x3b }', 'meta mark set ' + mark_hex] + ] + + self.verify_nftables(nftables_search, 'ip mangle') + + # IPv6 + nftables6_search = [ + [f'iifname "{interface}"', 'jump VYOS_PBR6_smoketest'], + ['meta l4proto udp', 'drop'], + ['tcp flags syn / syn,ack', 'meta mark set ' + mark_hex], + ['ct state new', 'tcp dport 22', 'ip6 saddr 2001:db8::/64', 'ip6 hoplimit > 2', 'meta mark set ' + mark_hex], + ['meta l4proto ipv6-icmp', 'log prefix "[smoketest6-4-A]"', 'icmpv6 type echo-request', 'ip6 length != { 128, 1024-2048 }', 'meta mark set ' + mark_hex], + ['ip6 dscp != { 0x0e-0x13, 0x3d }', 'meta mark set ' + mark_hex] + ] + + self.verify_nftables(nftables6_search, 'ip6 mangle') + if __name__ == '__main__': unittest.main(verbosity=2) diff --git a/smoketest/scripts/cli/test_protocols_bgp.py b/smoketest/scripts/cli/test_protocols_bgp.py index 6196ffe60..d2dad8c1a 100755 --- a/smoketest/scripts/cli/test_protocols_bgp.py +++ b/smoketest/scripts/cli/test_protocols_bgp.py @@ -287,6 +287,7 @@ class TestProtocolsBGP(VyOSUnitTestSHIM.TestCase): self.cli_set(base_path + ['parameters', 'bestpath', 'as-path', 'multipath-relax']) self.cli_set(base_path + ['parameters', 'bestpath', 'bandwidth', 'default-weight-for-missing']) self.cli_set(base_path + ['parameters', 'bestpath', 'compare-routerid']) + self.cli_set(base_path + ['parameters', 'bestpath', 'peer-type', 'multipath-relax']) self.cli_set(base_path + ['parameters', 'conditional-advertisement', 'timer', cond_adv_timer]) self.cli_set(base_path + ['parameters', 'fast-convergence']) @@ -318,6 +319,7 @@ class TestProtocolsBGP(VyOSUnitTestSHIM.TestCase): self.assertIn(f' bgp bestpath as-path multipath-relax', frrconfig) self.assertIn(f' bgp bestpath bandwidth default-weight-for-missing', frrconfig) self.assertIn(f' bgp bestpath compare-routerid', frrconfig) + self.assertIn(f' bgp bestpath peer-type multipath-relax', frrconfig) self.assertIn(f' bgp minimum-holdtime {min_hold_time}', frrconfig) self.assertIn(f' bgp reject-as-sets', frrconfig) self.assertIn(f' bgp shutdown', frrconfig) diff --git a/smoketest/scripts/cli/test_protocols_isis.py b/smoketest/scripts/cli/test_protocols_isis.py index ee4be0b37..e4bb9e1f8 100755 --- a/smoketest/scripts/cli/test_protocols_isis.py +++ b/smoketest/scripts/cli/test_protocols_isis.py @@ -262,5 +262,52 @@ class TestProtocolsISIS(VyOSUnitTestSHIM.TestCase): self.assertIn(f' isis bfd', tmp) self.assertIn(f' isis bfd profile {bfd_profile}', tmp) + def test_isis_07_segment_routing_configuration(self): + global_block_low = "1000" + global_block_high = "1999" + local_block_low = "2000" + local_block_high = "2999" + interface = 'lo' + maximum_stack_size = '5' + prefix_one = '192.168.0.1/32' + prefix_two = '192.168.0.2/32' + prefix_three = '192.168.0.3/32' + prefix_four = '192.168.0.4/32' + prefix_one_value = '1' + prefix_two_value = '2' + prefix_three_value = '60000' + prefix_four_value = '65000' + + self.cli_set(base_path + ['net', net]) + self.cli_set(base_path + ['interface', interface]) + self.cli_set(base_path + ['segment-routing', 'enable']) + self.cli_set(base_path + ['segment-routing', 'maximum-label-depth', maximum_stack_size]) + self.cli_set(base_path + ['segment-routing', 'global-block', 'low-label-value', global_block_low]) + self.cli_set(base_path + ['segment-routing', 'global-block', 'high-label-value', global_block_high]) + self.cli_set(base_path + ['segment-routing', 'local-block', 'low-label-value', local_block_low]) + self.cli_set(base_path + ['segment-routing', 'local-block', 'high-label-value', local_block_high]) + self.cli_set(base_path + ['segment-routing', 'prefix', prefix_one, 'index', 'value', prefix_one_value]) + self.cli_set(base_path + ['segment-routing', 'prefix', prefix_one, 'index', 'explicit-null']) + self.cli_set(base_path + ['segment-routing', 'prefix', prefix_two, 'index', 'value', prefix_two_value]) + self.cli_set(base_path + ['segment-routing', 'prefix', prefix_two, 'index', 'no-php-flag']) + self.cli_set(base_path + ['segment-routing', 'prefix', prefix_three, 'absolute', 'value', prefix_three_value]) + self.cli_set(base_path + ['segment-routing', 'prefix', prefix_three, 'absolute', 'explicit-null']) + self.cli_set(base_path + ['segment-routing', 'prefix', prefix_four, 'absolute', 'value', prefix_four_value]) + self.cli_set(base_path + ['segment-routing', 'prefix', prefix_four, 'absolute', 'no-php-flag']) + + # Commit all changes + self.cli_commit() + + # Verify all changes + tmp = self.getFRRconfig(f'router isis {domain}', daemon='isisd') + self.assertIn(f' net {net}', tmp) + self.assertIn(f' segment-routing on', tmp) + self.assertIn(f' segment-routing global-block {global_block_low} {global_block_high} local-block {local_block_low} {local_block_high}', tmp) + self.assertIn(f' segment-routing node-msd {maximum_stack_size}', tmp) + self.assertIn(f' segment-routing prefix {prefix_one} index {prefix_one_value} explicit-null', tmp) + self.assertIn(f' segment-routing prefix {prefix_two} index {prefix_two_value} no-php-flag', tmp) + self.assertIn(f' segment-routing prefix {prefix_three} absolute {prefix_three_value} explicit-null', tmp) + self.assertIn(f' segment-routing prefix {prefix_four} absolute {prefix_four_value} no-php-flag', tmp) + if __name__ == '__main__': unittest.main(verbosity=2) diff --git a/smoketest/scripts/cli/test_protocols_nhrp.py b/smoketest/scripts/cli/test_protocols_nhrp.py index 40b19fec7..59252875b 100755 --- a/smoketest/scripts/cli/test_protocols_nhrp.py +++ b/smoketest/scripts/cli/test_protocols_nhrp.py @@ -26,65 +26,79 @@ nhrp_path = ['protocols', 'nhrp'] vpn_path = ['vpn', 'ipsec'] class TestProtocolsNHRP(VyOSUnitTestSHIM.TestCase): + @classmethod + def setUpClass(cls): + super(TestProtocolsNHRP, cls).setUpClass() + + # ensure we can also run this test on a live system - so lets clean + # out the current configuration :) + cls.cli_delete(cls, nhrp_path) + cls.cli_delete(cls, tunnel_path) + def tearDown(self): self.cli_delete(nhrp_path) self.cli_delete(tunnel_path) self.cli_commit() def test_config(self): - self.cli_delete(nhrp_path) - self.cli_delete(tunnel_path) + tunnel_if = "tun100" + tunnel_source = "192.0.2.1" + tunnel_encapsulation = "gre" + esp_group = "ESP-HUB" + ike_group = "IKE-HUB" + nhrp_secret = "vyos123" + nhrp_profile = "NHRPVPN" + ipsec_secret = "secret" # Tunnel - self.cli_set(tunnel_path + ["tun100", "address", "172.16.253.134/29"]) - self.cli_set(tunnel_path + ["tun100", "encapsulation", "gre"]) - self.cli_set(tunnel_path + ["tun100", "source-address", "192.0.2.1"]) - self.cli_set(tunnel_path + ["tun100", "multicast", "enable"]) - self.cli_set(tunnel_path + ["tun100", "parameters", "ip", "key", "1"]) + self.cli_set(tunnel_path + [tunnel_if, "address", "172.16.253.134/29"]) + self.cli_set(tunnel_path + [tunnel_if, "encapsulation", tunnel_encapsulation]) + self.cli_set(tunnel_path + [tunnel_if, "source-address", tunnel_source]) + self.cli_set(tunnel_path + [tunnel_if, "multicast", "enable"]) + self.cli_set(tunnel_path + [tunnel_if, "parameters", "ip", "key", "1"]) # NHRP - self.cli_set(nhrp_path + ["tunnel", "tun100", "cisco-authentication", "secret"]) - self.cli_set(nhrp_path + ["tunnel", "tun100", "holding-time", "300"]) - self.cli_set(nhrp_path + ["tunnel", "tun100", "multicast", "dynamic"]) - self.cli_set(nhrp_path + ["tunnel", "tun100", "redirect"]) - self.cli_set(nhrp_path + ["tunnel", "tun100", "shortcut"]) + self.cli_set(nhrp_path + ["tunnel", tunnel_if, "cisco-authentication", nhrp_secret]) + self.cli_set(nhrp_path + ["tunnel", tunnel_if, "holding-time", "300"]) + self.cli_set(nhrp_path + ["tunnel", tunnel_if, "multicast", "dynamic"]) + self.cli_set(nhrp_path + ["tunnel", tunnel_if, "redirect"]) + self.cli_set(nhrp_path + ["tunnel", tunnel_if, "shortcut"]) # IKE/ESP Groups - self.cli_set(vpn_path + ["esp-group", "ESP-HUB", "compression", "disable"]) - self.cli_set(vpn_path + ["esp-group", "ESP-HUB", "lifetime", "1800"]) - self.cli_set(vpn_path + ["esp-group", "ESP-HUB", "mode", "transport"]) - self.cli_set(vpn_path + ["esp-group", "ESP-HUB", "pfs", "dh-group2"]) - self.cli_set(vpn_path + ["esp-group", "ESP-HUB", "proposal", "1", "encryption", "aes256"]) - self.cli_set(vpn_path + ["esp-group", "ESP-HUB", "proposal", "1", "hash", "sha1"]) - self.cli_set(vpn_path + ["esp-group", "ESP-HUB", "proposal", "2", "encryption", "3des"]) - self.cli_set(vpn_path + ["esp-group", "ESP-HUB", "proposal", "2", "hash", "md5"]) - self.cli_set(vpn_path + ["ike-group", "IKE-HUB", "ikev2-reauth", "no"]) - self.cli_set(vpn_path + ["ike-group", "IKE-HUB", "key-exchange", "ikev1"]) - self.cli_set(vpn_path + ["ike-group", "IKE-HUB", "lifetime", "3600"]) - self.cli_set(vpn_path + ["ike-group", "IKE-HUB", "proposal", "1", "dh-group", "2"]) - self.cli_set(vpn_path + ["ike-group", "IKE-HUB", "proposal", "1", "encryption", "aes256"]) - self.cli_set(vpn_path + ["ike-group", "IKE-HUB", "proposal", "1", "hash", "sha1"]) - self.cli_set(vpn_path + ["ike-group", "IKE-HUB", "proposal", "2", "dh-group", "2"]) - self.cli_set(vpn_path + ["ike-group", "IKE-HUB", "proposal", "2", "encryption", "aes128"]) - self.cli_set(vpn_path + ["ike-group", "IKE-HUB", "proposal", "2", "hash", "sha1"]) + self.cli_set(vpn_path + ["esp-group", esp_group, "lifetime", "1800"]) + self.cli_set(vpn_path + ["esp-group", esp_group, "mode", "transport"]) + self.cli_set(vpn_path + ["esp-group", esp_group, "pfs", "dh-group2"]) + self.cli_set(vpn_path + ["esp-group", esp_group, "proposal", "1", "encryption", "aes256"]) + self.cli_set(vpn_path + ["esp-group", esp_group, "proposal", "1", "hash", "sha1"]) + self.cli_set(vpn_path + ["esp-group", esp_group, "proposal", "2", "encryption", "3des"]) + self.cli_set(vpn_path + ["esp-group", esp_group, "proposal", "2", "hash", "md5"]) + + self.cli_set(vpn_path + ["ike-group", ike_group, "key-exchange", "ikev1"]) + self.cli_set(vpn_path + ["ike-group", ike_group, "lifetime", "3600"]) + self.cli_set(vpn_path + ["ike-group", ike_group, "proposal", "1", "dh-group", "2"]) + self.cli_set(vpn_path + ["ike-group", ike_group, "proposal", "1", "encryption", "aes256"]) + self.cli_set(vpn_path + ["ike-group", ike_group, "proposal", "1", "hash", "sha1"]) + self.cli_set(vpn_path + ["ike-group", ike_group, "proposal", "2", "dh-group", "2"]) + self.cli_set(vpn_path + ["ike-group", ike_group, "proposal", "2", "encryption", "aes128"]) + self.cli_set(vpn_path + ["ike-group", ike_group, "proposal", "2", "hash", "sha1"]) # Profile - Not doing full DMVPN checks here, just want to verify the profile name in the output self.cli_set(vpn_path + ["interface", "eth0"]) - self.cli_set(vpn_path + ["profile", "NHRPVPN", "authentication", "mode", "pre-shared-secret"]) - self.cli_set(vpn_path + ["profile", "NHRPVPN", "authentication", "pre-shared-secret", "secret"]) - self.cli_set(vpn_path + ["profile", "NHRPVPN", "bind", "tunnel", "tun100"]) - self.cli_set(vpn_path + ["profile", "NHRPVPN", "esp-group", "ESP-HUB"]) - self.cli_set(vpn_path + ["profile", "NHRPVPN", "ike-group", "IKE-HUB"]) + self.cli_set(vpn_path + ["profile", nhrp_profile, "authentication", "mode", "pre-shared-secret"]) + self.cli_set(vpn_path + ["profile", nhrp_profile, "authentication", "pre-shared-secret", ipsec_secret]) + self.cli_set(vpn_path + ["profile", nhrp_profile, "bind", "tunnel", tunnel_if]) + self.cli_set(vpn_path + ["profile", nhrp_profile, "esp-group", esp_group]) + self.cli_set(vpn_path + ["profile", nhrp_profile, "ike-group", ike_group]) self.cli_commit() opennhrp_lines = [ - 'interface tun100 #hub NHRPVPN', - 'cisco-authentication secret', - 'holding-time 300', - 'shortcut', - 'multicast dynamic', - 'redirect' + f'interface {tunnel_if} #hub {nhrp_profile}', + f'cisco-authentication {nhrp_secret}', + f'holding-time 300', + f'shortcut', + f'multicast dynamic', + f'redirect' ] tmp_opennhrp_conf = read_file('/run/opennhrp/opennhrp.conf') @@ -93,13 +107,13 @@ class TestProtocolsNHRP(VyOSUnitTestSHIM.TestCase): self.assertIn(line, tmp_opennhrp_conf) firewall_matches = [ - 'ip protocol gre', - 'ip saddr 192.0.2.1', - 'ip daddr 224.0.0.0/4', - 'comment "VYOS_NHRP_tun100"' + f'ip protocol {tunnel_encapsulation}', + f'ip saddr {tunnel_source}', + f'ip daddr 224.0.0.0/4', + f'comment "VYOS_NHRP_{tunnel_if}"' ] - self.assertTrue(find_nftables_rule('ip filter', 'VYOS_FW_OUTPUT', firewall_matches) is not None) + self.assertTrue(find_nftables_rule('ip vyos_nhrp_filter', 'VYOS_NHRP_OUTPUT', firewall_matches) is not None) self.assertTrue(process_named_running('opennhrp')) if __name__ == '__main__': diff --git a/smoketest/scripts/cli/test_service_ipoe-server.py b/smoketest/scripts/cli/test_service_ipoe-server.py new file mode 100755 index 000000000..bdab35834 --- /dev/null +++ b/smoketest/scripts/cli/test_service_ipoe-server.py @@ -0,0 +1,91 @@ +#!/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 re +import unittest + +from base_accel_ppp_test import BasicAccelPPPTest +from vyos.configsession import ConfigSessionError +from vyos.util import cmd + +from configparser import ConfigParser + +ac_name = 'ACN' +interface = 'eth0' + +class TestServiceIPoEServer(BasicAccelPPPTest.TestCase): + @classmethod + def setUpClass(cls): + cls._base_path = ['service', 'ipoe-server'] + cls._config_file = '/run/accel-pppd/ipoe.conf' + cls._chap_secrets = '/run/accel-pppd/ipoe.chap-secrets' + + # call base-classes classmethod + super(TestServiceIPoEServer, cls).setUpClass() + + def verify(self, conf): + super().verify(conf) + + # Validate configuration values + accel_modules = list(conf['modules'].keys()) + self.assertIn('log_syslog', accel_modules) + self.assertIn('ipoe', accel_modules) + self.assertIn('shaper', accel_modules) + self.assertIn('ipv6pool', accel_modules) + self.assertIn('ipv6_nd', accel_modules) + self.assertIn('ipv6_dhcp', accel_modules) + self.assertIn('ippool', accel_modules) + + def basic_config(self): + self.set(['interface', interface, 'client-subnet', '192.168.0.0/24']) + + def test_accel_local_authentication(self): + mac_address = '08:00:27:2f:d8:06' + self.set(['authentication', 'interface', interface, 'mac', mac_address]) + self.set(['authentication', 'mode', 'local']) + + # No IPoE interface configured + with self.assertRaises(ConfigSessionError): + self.cli_commit() + + # Test configuration of local authentication for PPPoE server + self.basic_config() + + # commit changes + self.cli_commit() + + # Validate configuration values + conf = ConfigParser(allow_no_value=True, delimiters='=') + conf.read(self._config_file) + + # check proper path to chap-secrets file + self.assertEqual(conf['chap-secrets']['chap-secrets'], self._chap_secrets) + + accel_modules = list(conf['modules'].keys()) + self.assertIn('chap-secrets', accel_modules) + + # basic verification + self.verify(conf) + + # check local users + tmp = cmd(f'sudo cat {self._chap_secrets}') + regex = f'{interface}\s+\*\s+{mac_address}\s+\*' + tmp = re.findall(regex, tmp) + self.assertTrue(tmp) + +if __name__ == '__main__': + unittest.main(verbosity=2) + diff --git a/smoketest/scripts/cli/test_service_pppoe-server.py b/smoketest/scripts/cli/test_service_pppoe-server.py index 51cc098ef..7546c2e3d 100755 --- a/smoketest/scripts/cli/test_service_pppoe-server.py +++ b/smoketest/scripts/cli/test_service_pppoe-server.py @@ -1,6 +1,6 @@ #!/usr/bin/env python3 # -# Copyright (C) 020 VyOS maintainers and contributors +# 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 @@ -14,27 +14,27 @@ # 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 unittest from base_accel_ppp_test import BasicAccelPPPTest from configparser import ConfigParser -from vyos.configsession import ConfigSessionError -from vyos.util import process_named_running +from vyos.util import read_file +from vyos.template import range_to_regex local_if = ['interfaces', 'dummy', 'dum667'] ac_name = 'ACN' interface = 'eth0' class TestServicePPPoEServer(BasicAccelPPPTest.TestCase): - def setUp(self): - self._base_path = ['service', 'pppoe-server'] - self._process_name = 'accel-pppd' - self._config_file = '/run/accel-pppd/pppoe.conf' - self._chap_secrets = '/run/accel-pppd/pppoe.chap-secrets' + @classmethod + def setUpClass(cls): + cls._base_path = ['service', 'pppoe-server'] + cls._config_file = '/run/accel-pppd/pppoe.conf' + cls._chap_secrets = '/run/accel-pppd/pppoe.chap-secrets' - super().setUp() + # call base-classes classmethod + super(TestServicePPPoEServer, cls).setUpClass() def tearDown(self): self.cli_delete(local_if) @@ -120,8 +120,6 @@ class TestServicePPPoEServer(BasicAccelPPPTest.TestCase): # check interface-cache self.assertEqual(conf['ppp']['unit-cache'], interface_cache) - # Check for running process - self.assertTrue(process_named_running(self._process_name)) def test_pppoe_server_authentication_protocols(self): # Test configuration of local authentication for PPPoE server @@ -139,8 +137,6 @@ class TestServicePPPoEServer(BasicAccelPPPTest.TestCase): self.assertEqual(conf['modules']['auth_mschap_v2'], None) - # Check for running process - self.assertTrue(process_named_running(self._process_name)) def test_pppoe_server_client_ip_pool(self): # Test configuration of IPv6 client pools @@ -168,9 +164,6 @@ class TestServicePPPoEServer(BasicAccelPPPTest.TestCase): self.assertEqual(conf['ip-pool'][start_stop], None) self.assertEqual(conf['ip-pool']['gw-ip-address'], self._gateway) - # Check for running process - self.assertTrue(process_named_running(self._process_name)) - def test_pppoe_server_client_ipv6_pool(self): # Test configuration of IPv6 client pools @@ -211,9 +204,6 @@ class TestServicePPPoEServer(BasicAccelPPPTest.TestCase): self.assertEqual(conf['ipv6-pool'][client_prefix], None) self.assertEqual(conf['ipv6-pool']['delegate'], f'{delegate_prefix},{delegate_mask}') - # Check for running process - self.assertTrue(process_named_running(self._process_name)) - def test_accel_radius_authentication(self): radius_called_sid = 'ifname:mac' @@ -234,5 +224,27 @@ class TestServicePPPoEServer(BasicAccelPPPTest.TestCase): self.assertEqual(conf['radius']['acct-interim-jitter'], radius_acct_interim_jitter) + def test_pppoe_server_vlan(self): + + vlans = ['100', '200', '300-310'] + + # Test configuration of local authentication for PPPoE server + self.basic_config() + + for vlan in vlans: + self.set(['interface', interface, 'vlan', vlan]) + + # commit changes + self.cli_commit() + + # Validate configuration values + config = read_file(self._config_file) + for vlan in vlans: + tmp = range_to_regex(vlan) + self.assertIn(f'interface=re:^{interface}\.{tmp}$', config) + + tmp = ','.join(vlans) + self.assertIn(f'vlan-mon={interface},{tmp}', config) + if __name__ == '__main__': unittest.main(verbosity=2) diff --git a/smoketest/scripts/cli/test_system_conntrack.py b/smoketest/scripts/cli/test_system_conntrack.py index 95c2a6c55..fd16146b1 100755 --- a/smoketest/scripts/cli/test_system_conntrack.py +++ b/smoketest/scripts/cli/test_system_conntrack.py @@ -59,7 +59,7 @@ class TestSystemConntrack(VyOSUnitTestSHIM.TestCase): }, 'net.netfilter.nf_conntrack_tcp_max_retrans' :{ 'cli' : ['tcp', 'max-retrans'], - 'test_value' : '1024', + 'test_value' : '128', 'default_value' : '3', }, 'net.netfilter.nf_conntrack_icmp_timeout' :{ diff --git a/smoketest/scripts/cli/test_vpn_ipsec.py b/smoketest/scripts/cli/test_vpn_ipsec.py index 8a6514d57..bd242104f 100755 --- a/smoketest/scripts/cli/test_vpn_ipsec.py +++ b/smoketest/scripts/cli/test_vpn_ipsec.py @@ -1,6 +1,6 @@ #!/usr/bin/env python3 # -# Copyright (C) 2021 VyOS maintainers and contributors +# Copyright (C) 2021-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 @@ -33,6 +33,7 @@ dhcp_waiting_file = '/tmp/ipsec_dhcp_waiting' swanctl_file = '/etc/swanctl/swanctl.conf' peer_ip = '203.0.113.45' +connection_name = 'main-branch' interface = 'eth1' vif = '100' esp_group = 'MyESPGroup' @@ -150,7 +151,7 @@ class TestVPNIPsec(VyOSUnitTestSHIM.TestCase): self.cli_set(ethernet_path + [interface, 'vif', vif, 'address', 'dhcp']) # Use VLAN to avoid getting IP from qemu dhcp server # Site to site - peer_base_path = base_path + ['site-to-site', 'peer', peer_ip] + peer_base_path = base_path + ['site-to-site', 'peer', connection_name] self.cli_set(peer_base_path + ['authentication', 'mode', 'pre-shared-secret']) self.cli_set(peer_base_path + ['authentication', 'pre-shared-secret', secret]) self.cli_set(peer_base_path + ['ike-group', ike_group]) @@ -173,7 +174,7 @@ class TestVPNIPsec(VyOSUnitTestSHIM.TestCase): priority = '20' life_bytes = '100000' life_packets = '2000000' - peer_base_path = base_path + ['site-to-site', 'peer', peer_ip] + peer_base_path = base_path + ['site-to-site', 'peer', connection_name] self.cli_set(base_path + ['esp-group', esp_group, 'life-bytes', life_bytes]) self.cli_set(base_path + ['esp-group', esp_group, 'life-packets', life_packets]) @@ -183,6 +184,7 @@ class TestVPNIPsec(VyOSUnitTestSHIM.TestCase): self.cli_set(peer_base_path + ['ike-group', ike_group]) self.cli_set(peer_base_path + ['default-esp-group', esp_group]) self.cli_set(peer_base_path + ['local-address', local_address]) + self.cli_set(peer_base_path + ['remote-address', peer_ip]) self.cli_set(peer_base_path + ['tunnel', '1', 'protocol', 'tcp']) self.cli_set(peer_base_path + ['tunnel', '1', 'local', 'prefix', '172.16.10.0/24']) self.cli_set(peer_base_path + ['tunnel', '1', 'local', 'prefix', '172.16.11.0/24']) @@ -211,11 +213,11 @@ class TestVPNIPsec(VyOSUnitTestSHIM.TestCase): f'local_addrs = {local_address} # dhcp:no', f'remote_addrs = {peer_ip}', f'mode = tunnel', - f'peer_{peer_ip.replace(".","-")}_tunnel_1', + f'{connection_name}-tunnel-1', f'local_ts = 172.16.10.0/24[tcp/443],172.16.11.0/24[tcp/443]', f'remote_ts = 172.17.10.0/24[tcp/443],172.17.11.0/24[tcp/443]', f'mode = tunnel', - f'peer_{peer_ip.replace(".","-")}_tunnel_2', + f'{connection_name}-tunnel-2', f'local_ts = 10.1.0.0/16', f'remote_ts = 10.2.0.0/16', f'priority = {priority}', @@ -226,7 +228,7 @@ class TestVPNIPsec(VyOSUnitTestSHIM.TestCase): swanctl_secrets_lines = [ f'id-local = {local_address} # dhcp:no', - f'id-remote = {peer_ip}', + f'id-remote_{peer_ip.replace(".","-")} = {peer_ip}', f'secret = "{secret}"' ] for line in swanctl_secrets_lines: @@ -236,18 +238,24 @@ class TestVPNIPsec(VyOSUnitTestSHIM.TestCase): def test_03_site_to_site_vti(self): local_address = '192.0.2.10' vti = 'vti10' + # IKE + self.cli_set(base_path + ['ike-group', ike_group, 'key-exchange', 'ikev2']) + self.cli_set(base_path + ['ike-group', ike_group, 'disable-mobike']) + # ESP + self.cli_set(base_path + ['esp-group', esp_group, 'compression']) # VTI interface self.cli_set(vti_path + [vti, 'address', '10.1.1.1/24']) - self.cli_set(base_path + ['ike-group', ike_group, 'key-exchange', 'ikev2']) # Site to site - peer_base_path = base_path + ['site-to-site', 'peer', peer_ip] + peer_base_path = base_path + ['site-to-site', 'peer', connection_name] self.cli_set(peer_base_path + ['authentication', 'mode', 'pre-shared-secret']) self.cli_set(peer_base_path + ['authentication', 'pre-shared-secret', secret]) self.cli_set(peer_base_path + ['connection-type', 'none']) + self.cli_set(peer_base_path + ['force-udp-encapsulation']) self.cli_set(peer_base_path + ['ike-group', ike_group]) self.cli_set(peer_base_path + ['default-esp-group', esp_group]) self.cli_set(peer_base_path + ['local-address', local_address]) + self.cli_set(peer_base_path + ['remote-address', peer_ip]) self.cli_set(peer_base_path + ['tunnel', '1', 'local', 'prefix', '172.16.10.0/24']) self.cli_set(peer_base_path + ['tunnel', '1', 'local', 'prefix', '172.16.11.0/24']) self.cli_set(peer_base_path + ['tunnel', '1', 'remote', 'prefix', '172.17.10.0/24']) @@ -269,10 +277,12 @@ class TestVPNIPsec(VyOSUnitTestSHIM.TestCase): f'proposals = aes128-sha1-modp1024', f'esp_proposals = aes128-sha1-modp1024', f'local_addrs = {local_address} # dhcp:no', + f'mobike = no', f'remote_addrs = {peer_ip}', f'mode = tunnel', f'local_ts = 172.16.10.0/24,172.16.11.0/24', f'remote_ts = 172.17.10.0/24,172.17.11.0/24', + f'ipcomp = yes', f'start_action = none', f'if_id_in = {if_id}', # will be 11 for vti10 - shifted by one f'if_id_out = {if_id}', @@ -283,7 +293,7 @@ class TestVPNIPsec(VyOSUnitTestSHIM.TestCase): swanctl_secrets_lines = [ f'id-local = {local_address} # dhcp:no', - f'id-remote = {peer_ip}', + f'id-remote_{peer_ip.replace(".","-")} = {peer_ip}', f'secret = "{secret}"' ] for line in swanctl_secrets_lines: @@ -311,7 +321,6 @@ class TestVPNIPsec(VyOSUnitTestSHIM.TestCase): self.cli_set(nhrp_path + ['tunnel', tunnel_if, 'shortcut']) # IKE/ESP Groups - self.cli_set(base_path + ['esp-group', esp_group, 'compression', 'disable']) self.cli_set(base_path + ['esp-group', esp_group, 'lifetime', esp_lifetime]) self.cli_set(base_path + ['esp-group', esp_group, 'mode', 'transport']) self.cli_set(base_path + ['esp-group', esp_group, 'pfs', 'dh-group2']) @@ -320,7 +329,6 @@ class TestVPNIPsec(VyOSUnitTestSHIM.TestCase): self.cli_set(base_path + ['esp-group', esp_group, 'proposal', '3', 'encryption', '3des']) self.cli_set(base_path + ['esp-group', esp_group, 'proposal', '3', 'hash', 'md5']) - self.cli_set(base_path + ['ike-group', ike_group, 'ikev2-reauth', 'no']) self.cli_set(base_path + ['ike-group', ike_group, 'key-exchange', 'ikev1']) self.cli_set(base_path + ['ike-group', ike_group, 'lifetime', ike_lifetime]) self.cli_set(base_path + ['ike-group', ike_group, 'proposal', '2', 'dh-group', '2']) @@ -366,10 +374,11 @@ class TestVPNIPsec(VyOSUnitTestSHIM.TestCase): self.cli_set(vti_path + [vti, 'address', '192.168.0.1/31']) peer_ip = '172.18.254.202' + connection_name = 'office' local_address = '172.18.254.201' - peer_base_path = base_path + ['site-to-site', 'peer', peer_ip] + peer_base_path = base_path + ['site-to-site', 'peer', connection_name] - self.cli_set(peer_base_path + ['authentication', 'id', peer_name]) + self.cli_set(peer_base_path + ['authentication', 'local-id', peer_name]) self.cli_set(peer_base_path + ['authentication', 'mode', 'x509']) self.cli_set(peer_base_path + ['authentication', 'remote-id', 'peer2']) self.cli_set(peer_base_path + ['authentication', 'x509', 'ca-certificate', ca_name]) @@ -378,6 +387,7 @@ class TestVPNIPsec(VyOSUnitTestSHIM.TestCase): self.cli_set(peer_base_path + ['ike-group', ike_group]) self.cli_set(peer_base_path + ['ikev2-reauth', 'inherit']) self.cli_set(peer_base_path + ['local-address', local_address]) + self.cli_set(peer_base_path + ['remote-address', peer_ip]) self.cli_set(peer_base_path + ['vti', 'bind', vti]) self.cli_set(peer_base_path + ['vti', 'esp-group', esp_group]) @@ -391,7 +401,7 @@ class TestVPNIPsec(VyOSUnitTestSHIM.TestCase): # to also support a vti0 interface if_id = str(int(if_id) +1) swanctl_lines = [ - f'peer_{tmp}', + f'{connection_name}', f'version = 0', # key-exchange not set - defaulting to 0 for ikev1 and ikev2 f'send_cert = always', f'mobike = yes', @@ -416,7 +426,7 @@ class TestVPNIPsec(VyOSUnitTestSHIM.TestCase): self.assertIn(line, swanctl_conf) swanctl_secrets_lines = [ - f'peer_{tmp}', + f'{connection_name}', f'file = {peer_name}.pem', ] for line in swanctl_secrets_lines: @@ -430,7 +440,7 @@ class TestVPNIPsec(VyOSUnitTestSHIM.TestCase): local_address = '192.0.2.5' local_id = 'vyos-r1' remote_id = 'vyos-r2' - peer_base_path = base_path + ['site-to-site', 'peer', peer_ip] + peer_base_path = base_path + ['site-to-site', 'peer', connection_name] self.cli_set(tunnel_path + ['tun1', 'encapsulation', 'gre']) self.cli_set(tunnel_path + ['tun1', 'source-address', local_address]) @@ -438,10 +448,9 @@ class TestVPNIPsec(VyOSUnitTestSHIM.TestCase): self.cli_set(base_path + ['interface', interface]) self.cli_set(base_path + ['options', 'flexvpn']) self.cli_set(base_path + ['options', 'interface', 'tun1']) - self.cli_set(base_path + ['ike-group', ike_group, 'ikev2-reauth', 'no']) self.cli_set(base_path + ['ike-group', ike_group, 'key-exchange', 'ikev2']) - self.cli_set(peer_base_path + ['authentication', 'id', local_id]) + self.cli_set(peer_base_path + ['authentication', 'local-id', local_id]) self.cli_set(peer_base_path + ['authentication', 'mode', 'pre-shared-secret']) self.cli_set(peer_base_path + ['authentication', 'pre-shared-secret', secret]) self.cli_set(peer_base_path + ['authentication', 'remote-id', remote_id]) @@ -449,6 +458,7 @@ class TestVPNIPsec(VyOSUnitTestSHIM.TestCase): self.cli_set(peer_base_path + ['ike-group', ike_group]) self.cli_set(peer_base_path + ['default-esp-group', esp_group]) self.cli_set(peer_base_path + ['local-address', local_address]) + self.cli_set(peer_base_path + ['remote-address', peer_ip]) self.cli_set(peer_base_path + ['tunnel', '1', 'protocol', 'gre']) self.cli_set(peer_base_path + ['virtual-address', '203.0.113.55']) @@ -464,7 +474,7 @@ class TestVPNIPsec(VyOSUnitTestSHIM.TestCase): f'life_time = 3600s', # default value f'local_addrs = {local_address} # dhcp:no', f'remote_addrs = {peer_ip}', - f'peer_{peer_ip.replace(".","-")}_tunnel_1', + f'{connection_name}-tunnel-1', f'mode = tunnel', ] @@ -473,7 +483,7 @@ class TestVPNIPsec(VyOSUnitTestSHIM.TestCase): swanctl_secrets_lines = [ f'id-local = {local_address} # dhcp:no', - f'id-remote = {peer_ip}', + f'id-remote_{peer_ip.replace(".","-")} = {peer_ip}', f'id-localid = {local_id}', f'id-remoteid = {remote_id}', f'secret = "{secret}"', diff --git a/smoketest/scripts/cli/test_vpn_sstp.py b/smoketest/scripts/cli/test_vpn_sstp.py index f58920b5b..434e3aa05 100755 --- a/smoketest/scripts/cli/test_vpn_sstp.py +++ b/smoketest/scripts/cli/test_vpn_sstp.py @@ -19,29 +19,49 @@ import unittest from base_accel_ppp_test import BasicAccelPPPTest from vyos.util import read_file - pki_path = ['pki'] -cert_data = 'MIICFDCCAbugAwIBAgIUfMbIsB/ozMXijYgUYG80T1ry+mcwCgYIKoZIzj0EAwIwWTELMAkGA1UEBhMCR0IxEzARBgNVBAgMClNvbWUtU3RhdGUxEjAQBgNVBAcMCVNvbWUtQ2l0eTENMAsGA1UECgwEVnlPUzESMBAGA1UEAwwJVnlPUyBUZXN0MB4XDTIxMDcyMDEyNDUxMloXDTI2MDcxOTEyNDUxMlowWTELMAkGA1UEBhMCR0IxEzARBgNVBAgMClNvbWUtU3RhdGUxEjAQBgNVBAcMCVNvbWUtQ2l0eTENMAsGA1UECgwEVnlPUzESMBAGA1UEAwwJVnlPUyBUZXN0MFkwEwYHKoZIzj0CAQYIKoZIzj0DAQcDQgAE01HrLcNttqq4/PtoMua8rMWEkOdBu7vP94xzDO7A8C92ls1v86eePy4QllKCzIw3QxBIoCuH2peGRfWgPRdFsKNhMF8wDwYDVR0TAQH/BAUwAwEB/zAOBgNVHQ8BAf8EBAMCAYYwHQYDVR0lBBYwFAYIKwYBBQUHAwIGCCsGAQUFBwMBMB0GA1UdDgQWBBSu+JnU5ZC4mkuEpqg2+Mk4K79oeDAKBggqhkjOPQQDAgNHADBEAiBEFdzQ/Bc3LftzngrY605UhA6UprHhAogKgROv7iR4QgIgEFUxTtW3xXJcnUPWhhUFhyZoqfn8dE93+dm/LDnp7C0=' -key_data = 'MIGHAgEAMBMGByqGSM49AgEGCCqGSM49AwEHBG0wawIBAQQgPLpD0Ohhoq0g4nhx2KMIuze7ucKUt/lBEB2wc03IxXyhRANCAATTUestw222qrj8+2gy5rysxYSQ50G7u8/3jHMM7sDwL3aWzW/zp54/LhCWUoLMjDdDEEigK4fal4ZF9aA9F0Ww' + +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 +""" class TestVPNSSTPServer(BasicAccelPPPTest.TestCase): - def setUp(self): - self._base_path = ['vpn', 'sstp'] - self._process_name = 'accel-pppd' - self._config_file = '/run/accel-pppd/sstp.conf' - self._chap_secrets = '/run/accel-pppd/sstp.chap-secrets' - super().setUp() + @classmethod + def setUpClass(cls): + cls._base_path = ['vpn', 'sstp'] + cls._config_file = '/run/accel-pppd/sstp.conf' + cls._chap_secrets = '/run/accel-pppd/sstp.chap-secrets' - def tearDown(self): - self.cli_delete(pki_path) - super().tearDown() + # call base-classes classmethod + super(TestVPNSSTPServer, cls).setUpClass() - def basic_config(self): - self.cli_delete(pki_path) - self.cli_set(pki_path + ['ca', 'sstp', 'certificate', cert_data]) - self.cli_set(pki_path + ['certificate', 'sstp', 'certificate', cert_data]) - self.cli_set(pki_path + ['certificate', 'sstp', 'private', 'key', key_data]) + cls.cli_set(cls, pki_path + ['ca', 'sstp', 'certificate', cert_data.replace('\n','')]) + cls.cli_set(cls, pki_path + ['certificate', 'sstp', 'certificate', cert_data.replace('\n','')]) + cls.cli_set(cls, pki_path + ['certificate', 'sstp', 'private', 'key', key_data.replace('\n','')]) + @classmethod + def tearDownClass(cls): + cls.cli_delete(cls, pki_path) + + super(TestVPNSSTPServer, cls).tearDownClass() + + def basic_config(self): # SSL is mandatory self.set(['ssl', 'ca-certificate', 'sstp']) self.set(['ssl', 'certificate', 'sstp']) diff --git a/smoketest/scripts/cli/test_zone_policy.py b/smoketest/scripts/cli/test_zone_policy.py deleted file mode 100755 index 2c580e2f1..000000000 --- a/smoketest/scripts/cli/test_zone_policy.py +++ /dev/null @@ -1,69 +0,0 @@ -#!/usr/bin/env python3 -# -# Copyright (C) 2021-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 unittest - -from base_vyostest_shim import VyOSUnitTestSHIM - -from vyos.util import cmd - -class TestZonePolicy(VyOSUnitTestSHIM.TestCase): - @classmethod - def setUpClass(cls): - super(TestZonePolicy, cls).setUpClass() - cls.cli_set(cls, ['firewall', 'name', 'smoketest', 'default-action', 'drop']) - - @classmethod - def tearDownClass(cls): - cls.cli_delete(cls, ['firewall']) - super(TestZonePolicy, cls).tearDownClass() - - def tearDown(self): - self.cli_delete(['zone-policy']) - self.cli_commit() - - def test_basic_zone(self): - self.cli_set(['zone-policy', 'zone', 'smoketest-eth0', 'interface', 'eth0']) - self.cli_set(['zone-policy', 'zone', 'smoketest-eth0', 'from', 'smoketest-local', 'firewall', 'name', 'smoketest']) - self.cli_set(['zone-policy', 'zone', 'smoketest-local', 'local-zone']) - self.cli_set(['zone-policy', 'zone', 'smoketest-local', 'from', 'smoketest-eth0', 'firewall', 'name', 'smoketest']) - - self.cli_commit() - - nftables_search = [ - ['chain VZONE_smoketest-eth0'], - ['chain VZONE_smoketest-local_IN'], - ['chain VZONE_smoketest-local_OUT'], - ['oifname { "eth0" }', 'jump VZONE_smoketest-eth0'], - ['jump VZONE_smoketest-local_IN'], - ['jump VZONE_smoketest-local_OUT'], - ['iifname { "eth0" }', 'jump NAME_smoketest'], - ['oifname { "eth0" }', 'jump NAME_smoketest'] - ] - - nftables_output = cmd('sudo nft list table ip filter') - - for search in nftables_search: - matched = False - for line in nftables_output.split("\n"): - if all(item in line for item in search): - matched = True - break - self.assertTrue(matched) - - -if __name__ == '__main__': - unittest.main(verbosity=2) |