diff options
Diffstat (limited to 'smoketest')
| -rwxr-xr-x | smoketest/scripts/cli/test_nat66.py | 68 | ||||
| -rwxr-xr-x | smoketest/scripts/cli/test_vpn_openconnect.py | 79 |
2 files changed, 99 insertions, 48 deletions
diff --git a/smoketest/scripts/cli/test_nat66.py b/smoketest/scripts/cli/test_nat66.py index aac6a30f9..4b5625569 100755 --- a/smoketest/scripts/cli/test_nat66.py +++ b/smoketest/scripts/cli/test_nat66.py @@ -42,6 +42,17 @@ class TestNAT66(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_source_nat66(self): source_prefix = 'fc00::/64' translation_prefix = 'fc01::/64' @@ -49,29 +60,23 @@ class TestNAT66(VyOSUnitTestSHIM.TestCase): self.cli_set(src_path + ['rule', '1', 'source', 'prefix', source_prefix]) self.cli_set(src_path + ['rule', '1', 'translation', 'address', translation_prefix]) - # check validate() - outbound-interface must be defined - self.cli_commit() + self.cli_set(src_path + ['rule', '2', 'outbound-interface', 'eth1']) + self.cli_set(src_path + ['rule', '2', 'source', 'prefix', source_prefix]) + self.cli_set(src_path + ['rule', '2', 'translation', 'address', 'masquerade']) - tmp = cmd('sudo nft -j list table ip6 nat') - data_json = jmespath.search('nftables[?rule].rule[?chain]', json.loads(tmp)) + self.cli_set(src_path + ['rule', '3', 'outbound-interface', 'eth1']) + self.cli_set(src_path + ['rule', '3', 'source', 'prefix', source_prefix]) + self.cli_set(src_path + ['rule', '3', 'exclude']) - for idx in range(0, len(data_json)): - data = data_json[idx] + self.cli_commit() - self.assertEqual(data['chain'], 'POSTROUTING') - self.assertEqual(data['family'], 'ip6') - self.assertEqual(data['table'], 'nat') + 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'] + ] - 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]) - translation_address = dict_search('snat.addr.prefix.addr', data['expr'][3]) - translation_mask = dict_search('snat.addr.prefix.len', data['expr'][3]) - - self.assertEqual(iface, 'eth1') - # check for translation address - self.assertEqual(f'{translation_address}/{translation_mask}', translation_prefix) - self.assertEqual(f'{address}/{mask}', source_prefix) + self.verify_nftables(nftables_search, 'ip6 nat') def test_source_nat66_address(self): source_prefix = 'fc00::/64' @@ -106,28 +111,25 @@ class TestNAT66(VyOSUnitTestSHIM.TestCase): def test_destination_nat66(self): destination_address = 'fc00::1' translation_address = 'fc01::1' + source_address = 'fc02::1' self.cli_set(dst_path + ['rule', '1', 'inbound-interface', 'eth1']) self.cli_set(dst_path + ['rule', '1', 'destination', 'address', destination_address]) self.cli_set(dst_path + ['rule', '1', 'translation', 'address', translation_address]) + self.cli_set(dst_path + ['rule', '2', 'inbound-interface', 'eth1']) + self.cli_set(dst_path + ['rule', '2', 'destination', 'address', destination_address]) + self.cli_set(dst_path + ['rule', '2', 'source', 'address', source_address]) + self.cli_set(dst_path + ['rule', '2', 'exclude']) + # 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'], 'PREROUTING') - self.assertEqual(data['family'], 'ip6') - self.assertEqual(data['table'], 'nat') - - iface = dict_search('match.right', data['expr'][0]) - dnat_addr = dict_search('dnat.addr', data['expr'][3]) + nftables_search = [ + ['iifname "eth1"', 'ip6 daddr fc00::1', 'dnat to fc01::1'], + ['iifname "eth1"', 'ip6 saddr fc02::1', 'ip6 daddr fc00::1', 'return'] + ] - self.assertEqual(dnat_addr, translation_address) - self.assertEqual(iface, 'eth1') + self.verify_nftables(nftables_search, 'ip6 nat') def test_destination_nat66_prefix(self): destination_prefix = 'fc00::/64' diff --git a/smoketest/scripts/cli/test_vpn_openconnect.py b/smoketest/scripts/cli/test_vpn_openconnect.py index bda279342..8572d6d66 100755 --- a/smoketest/scripts/cli/test_vpn_openconnect.py +++ b/smoketest/scripts/cli/test_vpn_openconnect.py @@ -1,6 +1,6 @@ #!/usr/bin/env python3 # -# 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 @@ -19,6 +19,7 @@ import unittest from base_vyostest_shim import VyOSUnitTestSHIM from vyos.util import process_named_running +from vyos.util import read_file OCSERV_CONF = '/run/ocserv/ocserv.conf' base_path = ['vpn', 'openconnect'] @@ -46,36 +47,84 @@ MIGHAgEAMBMGByqGSM49AgEGCCqGSM49AwEHBG0wawIBAQQgPLpD0Ohhoq0g4nhx u8/3jHMM7sDwL3aWzW/zp54/LhCWUoLMjDdDEEigK4fal4ZF9aA9F0Ww """ -class TestVpnOpenconnect(VyOSUnitTestSHIM.TestCase): +PROCESS_NAME = 'ocserv-main' +config_file = '/run/ocserv/ocserv.conf' +auth_file = '/run/ocserv/ocpasswd' +otp_file = '/run/ocserv/users.oath' + +class TestVPNOpenConnect(VyOSUnitTestSHIM.TestCase): + @classmethod + def setUpClass(cls): + super(TestVPNOpenConnect, cls).setUpClass() + + # ensure we can also run this test on a live system - so lets clean + # out the current configuration :) + cls.cli_delete(cls, base_path) + + cls.cli_set(cls, pki_path + ['ca', 'openconnect', 'certificate', cert_data.replace('\n','')]) + cls.cli_set(cls, pki_path + ['certificate', 'openconnect', 'certificate', cert_data.replace('\n','')]) + cls.cli_set(cls, pki_path + ['certificate', 'openconnect', 'private', 'key', key_data.replace('\n','')]) + + @classmethod + def tearDownClass(cls): + cls.cli_delete(cls, pki_path) + super(TestVPNOpenConnect, cls).tearDownClass() + def tearDown(self): - # Delete vpn openconnect configuration - self.cli_delete(pki_path) + self.assertTrue(process_named_running(PROCESS_NAME)) + self.cli_delete(base_path) self.cli_commit() - def test_vpn(self): + self.assertFalse(process_named_running(PROCESS_NAME)) + + def test_ocserv(self): user = 'vyos_user' password = 'vyos_pass' otp = '37500000026900000000200000000000' - - self.cli_delete(pki_path) - self.cli_delete(base_path) - - self.cli_set(pki_path + ['ca', 'openconnect', 'certificate', cert_data.replace('\n','')]) - self.cli_set(pki_path + ['certificate', 'openconnect', 'certificate', cert_data.replace('\n','')]) - self.cli_set(pki_path + ['certificate', 'openconnect', 'private', 'key', key_data.replace('\n','')]) + v4_subnet = '192.0.2.0/24' + v6_prefix = '2001:db8:1000::/64' + v6_len = '126' + name_server = ['1.2.3.4', '1.2.3.5', '2001:db8::1'] + split_dns = ['vyos.net', 'vyos.io'] self.cli_set(base_path + ['authentication', 'local-users', 'username', user, 'password', password]) self.cli_set(base_path + ['authentication', 'local-users', 'username', user, 'otp', 'key', otp]) self.cli_set(base_path + ['authentication', 'mode', 'local', 'password-otp']) - self.cli_set(base_path + ['network-settings', 'client-ip-settings', 'subnet', '192.0.2.0/24']) + + self.cli_set(base_path + ['network-settings', 'client-ip-settings', 'subnet', v4_subnet]) + self.cli_set(base_path + ['network-settings', 'client-ipv6-pool', 'prefix', v6_prefix]) + self.cli_set(base_path + ['network-settings', 'client-ipv6-pool', 'mask', v6_len]) + + for ns in name_server: + self.cli_set(base_path + ['network-settings', 'name-server', ns]) + for domain in split_dns: + self.cli_set(base_path + ['network-settings', 'split-dns', domain]) + self.cli_set(base_path + ['ssl', 'ca-certificate', 'openconnect']) self.cli_set(base_path + ['ssl', 'certificate', 'openconnect']) self.cli_commit() - # Check for running process - self.assertTrue(process_named_running('ocserv-main')) + # Verify configuration + daemon_config = read_file(config_file) + + # authentication mode local password-otp + self.assertIn(f'auth = "plain[passwd=/run/ocserv/ocpasswd,otp=/run/ocserv/users.oath]"', daemon_config) + self.assertIn(f'ipv4-network = {v4_subnet}', daemon_config) + self.assertIn(f'ipv6-network = {v6_prefix}', daemon_config) + self.assertIn(f'ipv6-subnet-prefix = {v6_len}', daemon_config) + + for ns in name_server: + self.assertIn(f'dns = {ns}', daemon_config) + for domain in split_dns: + self.assertIn(f'split-dns = {domain}', daemon_config) + + auth_config = read_file(auth_file) + self.assertIn(f'{user}:*:$', auth_config) + + otp_config = read_file(otp_file) + self.assertIn(f'HOTP/T30/6 {user} - {otp}', otp_config) if __name__ == '__main__': unittest.main(verbosity=2) |