From: Neale Ranns Date: Wed, 10 Apr 2019 12:39:10 +0000 (+0000) Subject: IPSEC: for each engine and algorithm tests X-Git-Tag: v19.04-rc1~6 X-Git-Url: https://gerrit.fd.io/r/gitweb?p=vpp.git;a=commitdiff_plain;h=4f33c80c5de96fccc15a91135099ee437f75f252 IPSEC: for each engine and algorithm tests refactor the IPSEC tests a bit so we can parameterise the setup. Change-Id: I777e5eb8f29ca1dce3dd273ebd05dae5846790af Signed-off-by: Neale Ranns --- diff --git a/src/vnet/ipsec/ipsec_sa.c b/src/vnet/ipsec/ipsec_sa.c index af37b2e49cc..da12560bb7d 100644 --- a/src/vnet/ipsec/ipsec_sa.c +++ b/src/vnet/ipsec/ipsec_sa.c @@ -268,7 +268,7 @@ ipsec_sa_del (u32 id) hash_unset (im->sa_index_by_sa_id, sa->id); err = ipsec_call_add_del_callbacks (im, sa, sa_index, 0); if (err) - return VNET_API_ERROR_SYSCALL_ERROR_1; + return VNET_API_ERROR_SYSCALL_ERROR_2; if (ipsec_sa_is_set_IS_TUNNEL (sa) && !ipsec_sa_is_set_IS_INBOUND (sa)) { diff --git a/test/template_ipsec.py b/test/template_ipsec.py index e6f1d248824..0c873071800 100644 --- a/test/template_ipsec.py +++ b/test/template_ipsec.py @@ -74,9 +74,9 @@ class IPsecIPv6Params(object): self.auth_key = 'C91KUR9GYMm5GfkEvNjX' self.crypt_algo_vpp_id = (VppEnum.vl_api_ipsec_crypto_alg_t. - IPSEC_API_CRYPTO_ALG_AES_CBC_256) + IPSEC_API_CRYPTO_ALG_AES_CBC_128) self.crypt_algo = 'AES-CBC' # scapy name - self.crypt_key = 'JPjyOWBeVEQiMe7hJPjyOWBeVEQiMe7h' + self.crypt_key = 'JPjyOWBeVEQiMe7h' self.flags = 0 self.nat_header = None @@ -106,8 +106,8 @@ def config_tun_params(p, encryption_type, tun_if): def config_tra_params(p, encryption_type): - use_esn = p.flags & (VppEnum.vl_api_ipsec_sad_flags_t. - IPSEC_API_SAD_FLAG_USE_ESN) + use_esn = bool(p.flags & (VppEnum.vl_api_ipsec_sad_flags_t. + IPSEC_API_SAD_FLAG_USE_ESN)) p.scapy_tra_sa = SecurityAssociation( encryption_type, spi=p.vpp_tra_spi, @@ -146,6 +146,8 @@ class TemplateIpsec(VppTestCase): |tun_if| -------> |VPP| ------> |pg1| ------ --- --- """ + tun_spd_id = 1 + tra_spd_id = 2 def ipsec_select_backend(self): """ empty method to be overloaded when necessary """ @@ -168,37 +170,40 @@ class TemplateIpsec(VppTestCase): self.params = {self.ipv4_params.addr_type: self.ipv4_params, self.ipv6_params.addr_type: self.ipv6_params} + def config_interfaces(self): + self.create_pg_interfaces(range(3)) + self.interfaces = list(self.pg_interfaces) + for i in self.interfaces: + i.admin_up() + i.config_ip4() + i.resolve_arp() + i.config_ip6() + i.resolve_ndp() + def setUp(self): super(TemplateIpsec, self).setUp() self.setup_params() - self.tun_spd_id = 1 - self.tra_spd_id = 2 - self.vpp_esp_protocol = (VppEnum.vl_api_ipsec_proto_t. IPSEC_API_PROTO_ESP) self.vpp_ah_protocol = (VppEnum.vl_api_ipsec_proto_t. IPSEC_API_PROTO_AH) - self.create_pg_interfaces(range(3)) - self.interfaces = list(self.pg_interfaces) - for i in self.interfaces: - i.admin_up() - i.config_ip4() - i.resolve_arp() - i.config_ip6() - i.resolve_ndp() + self.config_interfaces() self.ipsec_select_backend() - def tearDown(self): - super(TemplateIpsec, self).tearDown() - + def unconfig_interfaces(self): for i in self.interfaces: i.admin_down() i.unconfig_ip4() i.unconfig_ip6() + def tearDown(self): + super(TemplateIpsec, self).tearDown() + + self.unconfig_interfaces() + if not self.vpp_dead: self.vapi.cli("show hardware") @@ -229,9 +234,8 @@ class TemplateIpsec(VppTestCase): for i in range(count)] -class IpsecTcpTests(object): - def test_tcp_checksum(self): - """ verify checksum correctness for vpp generated packets """ +class IpsecTcp(object): + def verify_tcp_checksum(self): self.vapi.cli("test http server") p = self.params[socket.AF_INET] config_tun_params(p, self.encryption_type, self.tun_if) @@ -246,9 +250,15 @@ class IpsecTcpTests(object): self.assert_packet_checksums_valid(decrypted) -class IpsecTra4Tests(object): - def test_tra_anti_replay(self, count=1): - """ ipsec v4 transport anti-reply test """ +class IpsecTcpTests(IpsecTcp): + def test_tcp_checksum(self): + """ verify checksum correctness for vpp generated packets """ + self.verify_tcp_checksum() + + +class IpsecTra4(object): + """ verify methods for Transport v4 """ + def verify_tra_anti_replay(self, count=1): p = self.params[socket.AF_INET] use_esn = p.vpp_tra_sa.use_esn @@ -379,7 +389,7 @@ class IpsecTra4Tests(object): p.scapy_tra_sa.seq_num = 351 p.vpp_tra_sa.seq_num = 351 - def test_tra_basic(self, count=1): + def verify_tra_basic4(self, count=1): """ ipsec v4 transport basic test """ self.vapi.cli("clear errors") try: @@ -413,14 +423,25 @@ class IpsecTra4Tests(object): self.assert_packet_counter_equal(self.tra4_encrypt_node_name, count) self.assert_packet_counter_equal(self.tra4_decrypt_node_name, count) + +class IpsecTra4Tests(IpsecTra4): + """ UT test methods for Transport v4 """ + def test_tra_anti_replay(self): + """ ipsec v4 transport anti-reply test """ + self.verify_tra_anti_replay(count=1) + + def test_tra_basic(self, count=1): + """ ipsec v4 transport basic test """ + self.verify_tra_basic4(count=1) + def test_tra_burst(self): """ ipsec v4 transport burst test """ - self.test_tra_basic(count=257) + self.verify_tra_basic4(count=257) -class IpsecTra6Tests(object): - def test_tra_basic6(self, count=1): - """ ipsec v6 transport basic test """ +class IpsecTra6(object): + """ verify methods for Transport v6 """ + def verify_tra_basic6(self, count=1): self.vapi.cli("clear errors") try: p = self.params[socket.AF_INET6] @@ -452,17 +473,25 @@ class IpsecTra6Tests(object): self.assert_packet_counter_equal(self.tra6_encrypt_node_name, count) self.assert_packet_counter_equal(self.tra6_decrypt_node_name, count) + +class IpsecTra6Tests(IpsecTra6): + """ UT test methods for Transport v6 """ + def test_tra_basic6(self): + """ ipsec v6 transport basic test """ + self.verify_tra_basic6(count=1) + def test_tra_burst6(self): """ ipsec v6 transport burst test """ - self.test_tra_basic6(count=257) + self.verify_tra_basic6(count=257) class IpsecTra46Tests(IpsecTra4Tests, IpsecTra6Tests): + """ UT test methods for Transport v6 and v4""" pass class IpsecTun4(object): - + """ verify methods for Tunnel v4 """ def verify_counters(self, p, count): if (hasattr(p, "spd_policy_in_any")): pkts = p.spd_policy_in_any.get_stats()['packets'] @@ -576,7 +605,7 @@ class IpsecTun4(object): class IpsecTun4Tests(IpsecTun4): - + """ UT test methods for Tunnel v4 """ def test_tun_basic44(self): """ ipsec 4o4 tunnel basic test """ self.verify_tun_44(self.params[socket.AF_INET], count=1) @@ -587,7 +616,7 @@ class IpsecTun4Tests(IpsecTun4): class IpsecTun6(object): - + """ verify methods for Tunnel v6 """ def verify_counters(self, p, count): if (hasattr(p, "tun_sa_in")): pkts = p.tun_sa_in.get_stats()['packets'] @@ -681,6 +710,7 @@ class IpsecTun6(object): class IpsecTun6Tests(IpsecTun6): + """ UT test methods for Tunnel v6 """ def test_tun_basic66(self): """ ipsec 6o6 tunnel basic test """ @@ -692,6 +722,7 @@ class IpsecTun6Tests(IpsecTun6): class IpsecTun46Tests(IpsecTun4Tests, IpsecTun6Tests): + """ UT test methods for Tunnel v6 & v4 """ pass diff --git a/test/test_ipsec_ah.py b/test/test_ipsec_ah.py index 93e16468ca1..ec0c5586c2b 100644 --- a/test/test_ipsec_ah.py +++ b/test/test_ipsec_ah.py @@ -5,7 +5,8 @@ from scapy.layers.ipsec import AH from framework import VppTestRunner from template_ipsec import TemplateIpsec, IpsecTra46Tests, IpsecTun46Tests, \ - config_tun_params, config_tra_params, IPsecIPv4Params, IPsecIPv6Params + config_tun_params, config_tra_params, IPsecIPv4Params, IPsecIPv6Params, \ + IpsecTra4, IpsecTun4, IpsecTra6, IpsecTun6 from template_ipsec import IpsecTcpTests from vpp_ipsec import VppIpsecSA, VppIpsecSpd, VppIpsecSpdEntry,\ VppIpsecSpdItfBinding @@ -14,7 +15,7 @@ from vpp_ip import DpoProto from vpp_papi import VppEnum -class TemplateIpsecAh(TemplateIpsec): +class ConfigIpsecAH(TemplateIpsec): """ Basic test for IPSEC using AH transport and Tunnel mode @@ -34,51 +35,74 @@ class TemplateIpsecAh(TemplateIpsec): |pg0| -------> |VPP| ------> |pg1| --- --- --- """ + encryption_type = AH + net_objs = [] + tra4_encrypt_node_name = "ah4-encrypt" + tra4_decrypt_node_name = "ah4-decrypt" + tra6_encrypt_node_name = "ah6-encrypt" + tra6_decrypt_node_name = "ah6-decrypt" + tun4_encrypt_node_name = "ah4-encrypt" + tun4_decrypt_node_name = "ah4-decrypt" + tun6_encrypt_node_name = "ah6-encrypt" + tun6_decrypt_node_name = "ah6-decrypt" @classmethod def setUpClass(cls): - super(TemplateIpsecAh, cls).setUpClass() + super(ConfigIpsecAH, cls).setUpClass() @classmethod def tearDownClass(cls): - super(TemplateIpsecAh, cls).tearDownClass() + super(ConfigIpsecAH, cls).tearDownClass() def setUp(self): - super(TemplateIpsecAh, self).setUp() + super(ConfigIpsecAH, self).setUp() + + def tearDown(self): + super(ConfigIpsecAH, self).tearDown() - self.encryption_type = AH + def config_network(self, params): + self.net_objs = [] self.tun_if = self.pg0 self.tra_if = self.pg2 self.logger.info(self.vapi.ppcli("show int addr")) self.tra_spd = VppIpsecSpd(self, self.tra_spd_id) self.tra_spd.add_vpp_config() - VppIpsecSpdItfBinding(self, self.tra_spd, - self.tra_if).add_vpp_config() + self.net_objs.append(self.tra_spd) self.tun_spd = VppIpsecSpd(self, self.tun_spd_id) self.tun_spd.add_vpp_config() - VppIpsecSpdItfBinding(self, self.tun_spd, - self.tun_if).add_vpp_config() + self.net_objs.append(self.tun_spd) - for _, p in self.params.items(): + b = VppIpsecSpdItfBinding(self, self.tra_spd, + self.tra_if) + b.add_vpp_config() + self.net_objs.append(b) + + b = VppIpsecSpdItfBinding(self, self.tun_spd, + self.tun_if) + b.add_vpp_config() + self.net_objs.append(b) + + for p in params: self.config_ah_tra(p) config_tra_params(p, self.encryption_type) - self.logger.info(self.vapi.ppcli("show ipsec")) - for _, p in self.params.items(): + for p in params: self.config_ah_tun(p) - self.logger.info(self.vapi.ppcli("show ipsec")) - for _, p in self.params.items(): + for p in params: d = DpoProto.DPO_PROTO_IP6 if p.is_ipv6 else DpoProto.DPO_PROTO_IP4 - VppIpRoute(self, p.remote_tun_if_host, p.addr_len, - [VppRoutePath(self.tun_if.remote_addr[p.addr_type], - 0xffffffff, - proto=d)], - is_ip6=p.is_ipv6).add_vpp_config() - - def tearDown(self): - super(TemplateIpsecAh, self).tearDown() - if not self.vpp_dead: - self.vapi.cli("show hardware") + r = VppIpRoute(self, p.remote_tun_if_host, p.addr_len, + [VppRoutePath(self.tun_if.remote_addr[p.addr_type], + 0xffffffff, + proto=d)], + is_ip6=p.is_ipv6) + r.add_vpp_config() + self.net_objs.append(r) + self.logger.info(self.vapi.ppcli("show ipsec all")) + + def unconfig_network(self): + for o in reversed(self.net_objs): + o.remove_vpp_config() + self.net_objs = [] def config_ah_tun(self, params): addr_type = params.addr_type @@ -95,6 +119,7 @@ class TemplateIpsecAh(TemplateIpsec): addr_bcast = params.addr_bcast flags = params.flags e = VppEnum.vl_api_ipsec_spd_action_t + objs = [] params.tun_sa_in = VppIpsecSA(self, scapy_tun_sa_id, scapy_tun_spi, auth_algo_vpp_id, auth_key, @@ -103,7 +128,7 @@ class TemplateIpsecAh(TemplateIpsec): self.tun_if.local_addr[addr_type], self.tun_if.remote_addr[addr_type], flags=flags) - params.tun_sa_in.add_vpp_config() + params.tun_sa_out = VppIpsecSA(self, vpp_tun_sa_id, vpp_tun_spi, auth_algo_vpp_id, auth_key, crypt_algo_vpp_id, crypt_key, @@ -111,53 +136,62 @@ class TemplateIpsecAh(TemplateIpsec): self.tun_if.remote_addr[addr_type], self.tun_if.local_addr[addr_type], flags=flags) - params.tun_sa_out.add_vpp_config() + + objs.append(params.tun_sa_in) + objs.append(params.tun_sa_out) params.spd_policy_in_any = VppIpsecSpdEntry(self, self.tun_spd, vpp_tun_sa_id, addr_any, addr_bcast, addr_any, addr_bcast, socket.IPPROTO_AH) - params.spd_policy_in_any.add_vpp_config() params.spd_policy_out_any = VppIpsecSpdEntry(self, self.tun_spd, vpp_tun_sa_id, addr_any, addr_bcast, addr_any, addr_bcast, socket.IPPROTO_AH, is_outbound=0) - params.spd_policy_out_any.add_vpp_config() - - VppIpsecSpdEntry(self, self.tun_spd, vpp_tun_sa_id, - remote_tun_if_host, - remote_tun_if_host, - self.pg1.remote_addr[addr_type], - self.pg1.remote_addr[addr_type], - 0, priority=10, - policy=e.IPSEC_API_SPD_ACTION_PROTECT, - is_outbound=0).add_vpp_config() - VppIpsecSpdEntry(self, self.tun_spd, scapy_tun_sa_id, - self.pg1.remote_addr[addr_type], - self.pg1.remote_addr[addr_type], - remote_tun_if_host, - remote_tun_if_host, - 0, policy=e.IPSEC_API_SPD_ACTION_PROTECT, - priority=10).add_vpp_config() - - VppIpsecSpdEntry(self, self.tun_spd, vpp_tun_sa_id, - remote_tun_if_host, - remote_tun_if_host, - self.pg0.local_addr[addr_type], - self.pg0.local_addr[addr_type], - 0, priority=20, - policy=e.IPSEC_API_SPD_ACTION_PROTECT, - is_outbound=0).add_vpp_config() - VppIpsecSpdEntry(self, self.tun_spd, scapy_tun_sa_id, - self.pg0.local_addr[addr_type], - self.pg0.local_addr[addr_type], - remote_tun_if_host, - remote_tun_if_host, - 0, policy=e.IPSEC_API_SPD_ACTION_PROTECT, - priority=20).add_vpp_config() + + objs.append(params.spd_policy_out_any) + objs.append(params.spd_policy_in_any) + + e1 = VppIpsecSpdEntry(self, self.tun_spd, vpp_tun_sa_id, + remote_tun_if_host, + remote_tun_if_host, + self.pg1.remote_addr[addr_type], + self.pg1.remote_addr[addr_type], + 0, priority=10, + policy=e.IPSEC_API_SPD_ACTION_PROTECT, + is_outbound=0) + e2 = VppIpsecSpdEntry(self, self.tun_spd, scapy_tun_sa_id, + self.pg1.remote_addr[addr_type], + self.pg1.remote_addr[addr_type], + remote_tun_if_host, + remote_tun_if_host, + 0, policy=e.IPSEC_API_SPD_ACTION_PROTECT, + priority=10) + e3 = VppIpsecSpdEntry(self, self.tun_spd, vpp_tun_sa_id, + remote_tun_if_host, + remote_tun_if_host, + self.pg0.local_addr[addr_type], + self.pg0.local_addr[addr_type], + 0, priority=20, + policy=e.IPSEC_API_SPD_ACTION_PROTECT, + is_outbound=0) + e4 = VppIpsecSpdEntry(self, self.tun_spd, scapy_tun_sa_id, + self.pg0.local_addr[addr_type], + self.pg0.local_addr[addr_type], + remote_tun_if_host, + remote_tun_if_host, + 0, policy=e.IPSEC_API_SPD_ACTION_PROTECT, + priority=20) + + objs = objs + [e1, e2, e3, e4] + + for o in objs: + o.add_vpp_config() + + self.net_objs = self.net_objs + objs def config_ah_tra(self, params): addr_type = params.addr_type @@ -174,162 +208,168 @@ class TemplateIpsecAh(TemplateIpsec): flags = params.flags | (VppEnum.vl_api_ipsec_sad_flags_t. IPSEC_API_SAD_FLAG_USE_ANTI_REPLAY) e = VppEnum.vl_api_ipsec_spd_action_t + objs = [] params.tra_sa_in = VppIpsecSA(self, scapy_tra_sa_id, scapy_tra_spi, auth_algo_vpp_id, auth_key, crypt_algo_vpp_id, crypt_key, self.vpp_ah_protocol, flags=flags) - params.tra_sa_in.add_vpp_config() params.tra_sa_out = VppIpsecSA(self, vpp_tra_sa_id, vpp_tra_spi, auth_algo_vpp_id, auth_key, crypt_algo_vpp_id, crypt_key, self.vpp_ah_protocol, flags=flags) - params.tra_sa_out.add_vpp_config() - - VppIpsecSpdEntry(self, self.tra_spd, vpp_tra_sa_id, - addr_any, addr_bcast, - addr_any, addr_bcast, - socket.IPPROTO_AH).add_vpp_config() - VppIpsecSpdEntry(self, self.tra_spd, scapy_tra_sa_id, - addr_any, addr_bcast, - addr_any, addr_bcast, - socket.IPPROTO_AH, - is_outbound=0).add_vpp_config() - - VppIpsecSpdEntry(self, self.tra_spd, vpp_tra_sa_id, - self.tra_if.local_addr[addr_type], - self.tra_if.local_addr[addr_type], - self.tra_if.remote_addr[addr_type], - self.tra_if.remote_addr[addr_type], - 0, priority=10, - policy=e.IPSEC_API_SPD_ACTION_PROTECT, - is_outbound=0).add_vpp_config() - VppIpsecSpdEntry(self, self.tra_spd, scapy_tra_sa_id, - self.tra_if.local_addr[addr_type], - self.tra_if.local_addr[addr_type], - self.tra_if.remote_addr[addr_type], - self.tra_if.remote_addr[addr_type], - 0, policy=e.IPSEC_API_SPD_ACTION_PROTECT, - priority=10).add_vpp_config() - - -class TestIpsecAh1(TemplateIpsecAh, IpsecTcpTests): - """ Ipsec AH - TCP tests """ - pass + objs.append(params.tra_sa_in) + objs.append(params.tra_sa_out) + + objs.append(VppIpsecSpdEntry(self, self.tra_spd, vpp_tra_sa_id, + addr_any, addr_bcast, + addr_any, addr_bcast, + socket.IPPROTO_AH)) + objs.append(VppIpsecSpdEntry(self, self.tra_spd, scapy_tra_sa_id, + addr_any, addr_bcast, + addr_any, addr_bcast, + socket.IPPROTO_AH, + is_outbound=0)) + objs.append(VppIpsecSpdEntry(self, self.tra_spd, vpp_tra_sa_id, + self.tra_if.local_addr[addr_type], + self.tra_if.local_addr[addr_type], + self.tra_if.remote_addr[addr_type], + self.tra_if.remote_addr[addr_type], + 0, priority=10, + policy=e.IPSEC_API_SPD_ACTION_PROTECT, + is_outbound=0)) + objs.append(VppIpsecSpdEntry(self, self.tra_spd, scapy_tra_sa_id, + self.tra_if.local_addr[addr_type], + self.tra_if.local_addr[addr_type], + self.tra_if.remote_addr[addr_type], + self.tra_if.remote_addr[addr_type], + 0, policy=e.IPSEC_API_SPD_ACTION_PROTECT, + priority=10)) + + for o in objs: + o.add_vpp_config() + self.net_objs = self.net_objs + objs + + +class TemplateIpsecAh(ConfigIpsecAH): + """ + Basic test for IPSEC using AH transport and Tunnel mode -class TestIpsecAh2(TemplateIpsecAh, IpsecTra46Tests, IpsecTun46Tests): - """ Ipsec AH w/ SHA1 """ - tra4_encrypt_node_name = "ah4-encrypt" - tra4_decrypt_node_name = "ah4-decrypt" - tra6_encrypt_node_name = "ah6-encrypt" - tra6_decrypt_node_name = "ah6-decrypt" - tun4_encrypt_node_name = "ah4-encrypt" - tun4_decrypt_node_name = "ah4-decrypt" - tun6_encrypt_node_name = "ah6-encrypt" - tun6_decrypt_node_name = "ah6-decrypt" - - -class TestIpsecAh3(TemplateIpsecAh, IpsecTra46Tests, IpsecTun46Tests): - """ Ipsec AH w/ SHA1 & ESN """ - - tra4_encrypt_node_name = "ah4-encrypt" - tra4_decrypt_node_name = "ah4-decrypt" - tra6_encrypt_node_name = "ah6-encrypt" - tra6_decrypt_node_name = "ah6-decrypt" - tun4_encrypt_node_name = "ah4-encrypt" - tun4_decrypt_node_name = "ah4-decrypt" - tun6_encrypt_node_name = "ah6-encrypt" - tun6_decrypt_node_name = "ah6-decrypt" + TRANSPORT MODE: - def setup_params(self): - self.ipv4_params = IPsecIPv4Params() - self.ipv6_params = IPsecIPv6Params() - self.params = {self.ipv4_params.addr_type: self.ipv4_params, - self.ipv6_params.addr_type: self.ipv6_params} - for _, p in self.params.items(): - p.flags = (VppEnum.vl_api_ipsec_sad_flags_t. - IPSEC_API_SAD_FLAG_USE_ESN) + --- encrypt --- + |pg2| <-------> |VPP| + --- decrypt --- + TUNNEL MODE: -class TestIpsecAh4(TemplateIpsecAh, IpsecTra46Tests, IpsecTun46Tests): - """ Ipsec AH w/ SHA256 """ + --- encrypt --- plain --- + |pg0| <------- |VPP| <------ |pg1| + --- --- --- - tra4_encrypt_node_name = "ah4-encrypt" - tra4_decrypt_node_name = "ah4-decrypt" - tra6_encrypt_node_name = "ah6-encrypt" - tra6_decrypt_node_name = "ah6-decrypt" - tun4_encrypt_node_name = "ah4-encrypt" - tun4_decrypt_node_name = "ah4-decrypt" - tun6_encrypt_node_name = "ah6-encrypt" - tun6_decrypt_node_name = "ah6-decrypt" + --- decrypt --- plain --- + |pg0| -------> |VPP| ------> |pg1| + --- --- --- + """ + @classmethod + def setUpClass(cls): + super(TemplateIpsecAh, cls).setUpClass() - def setup_params(self): - self.ipv4_params = IPsecIPv4Params() - self.ipv6_params = IPsecIPv6Params() - self.ipv4_params.auth_algo_vpp_id = (VppEnum.vl_api_ipsec_integ_alg_t. - IPSEC_API_INTEG_ALG_SHA_256_128) - self.ipv4_params.auth_algo = 'SHA2-256-128' # scapy name - self.ipv6_params.auth_algo_vpp_id = (VppEnum.vl_api_ipsec_integ_alg_t. - IPSEC_API_INTEG_ALG_SHA_256_128) - self.ipv6_params.auth_algo = 'SHA2-256-128' # scapy name + @classmethod + def tearDownClass(cls): + super(TemplateIpsecAh, cls).tearDownClass() - self.params = {self.ipv4_params.addr_type: self.ipv4_params, - self.ipv6_params.addr_type: self.ipv6_params} + def setUp(self): + super(TemplateIpsecAh, self).setUp() + self.config_network(self.params.values()) + def tearDown(self): + self.unconfig_network() + super(TemplateIpsecAh, self).tearDown() -class TestIpsecAh5(TemplateIpsecAh, IpsecTra46Tests, IpsecTun46Tests): - """ Ipsec AH w/ SHA384 """ - tra4_encrypt_node_name = "ah4-encrypt" - tra4_decrypt_node_name = "ah4-decrypt" - tra6_encrypt_node_name = "ah6-encrypt" - tra6_decrypt_node_name = "ah6-decrypt" - tun4_encrypt_node_name = "ah4-encrypt" - tun4_decrypt_node_name = "ah4-decrypt" - tun6_encrypt_node_name = "ah6-encrypt" - tun6_decrypt_node_name = "ah6-decrypt" +class TestIpsecAh1(TemplateIpsecAh, IpsecTcpTests): + """ Ipsec AH - TCP tests """ + pass - def setup_params(self): - self.ipv4_params = IPsecIPv4Params() - self.ipv6_params = IPsecIPv6Params() - self.ipv4_params.auth_algo_vpp_id = (VppEnum.vl_api_ipsec_integ_alg_t. - IPSEC_API_INTEG_ALG_SHA_384_192) - self.ipv4_params.auth_algo = 'SHA2-384-192' # scapy name - self.ipv6_params.auth_algo_vpp_id = (VppEnum.vl_api_ipsec_integ_alg_t. - IPSEC_API_INTEG_ALG_SHA_384_192) - self.ipv6_params.auth_algo = 'SHA2-384-192' # scapy name - self.params = {self.ipv4_params.addr_type: self.ipv4_params, - self.ipv6_params.addr_type: self.ipv6_params} +class TestIpsecAh2(TemplateIpsecAh, IpsecTra46Tests, IpsecTun46Tests): + """ Ipsec AH w/ SHA1 """ + pass -class TestIpsecAh6(TemplateIpsecAh, IpsecTra46Tests, IpsecTun46Tests): - """ Ipsec AH w/ SHA512 """ +class TestIpsecAhAll(ConfigIpsecAH, + IpsecTra4, IpsecTra6, + IpsecTun4, IpsecTun6): + """ Ipsec AH all Algos """ - tra4_encrypt_node_name = "ah4-encrypt" - tra4_decrypt_node_name = "ah4-decrypt" - tra6_encrypt_node_name = "ah6-encrypt" - tra6_decrypt_node_name = "ah6-decrypt" - tun4_encrypt_node_name = "ah4-encrypt" - tun4_decrypt_node_name = "ah4-decrypt" - tun6_encrypt_node_name = "ah6-encrypt" - tun6_decrypt_node_name = "ah6-decrypt" + def setUp(self): + super(TestIpsecAhAll, self).setUp() - def setup_params(self): - self.ipv4_params = IPsecIPv4Params() - self.ipv6_params = IPsecIPv6Params() - self.ipv4_params.auth_algo_vpp_id = (VppEnum.vl_api_ipsec_integ_alg_t. - IPSEC_API_INTEG_ALG_SHA_512_256) - self.ipv4_params.auth_algo = 'SHA2-512-256' # scapy name - self.ipv6_params.auth_algo_vpp_id = (VppEnum.vl_api_ipsec_integ_alg_t. - IPSEC_API_INTEG_ALG_SHA_512_256) - self.ipv6_params.auth_algo = 'SHA2-512-256' # scapy name - - self.params = {self.ipv4_params.addr_type: self.ipv4_params, - self.ipv6_params.addr_type: self.ipv6_params} + def tearDown(self): + super(TestIpsecAhAll, self).tearDown() + + def test_integ_algs(self): + """SHA1_96, SHA256, SHA394, SHA512 w/ & w/o ESN""" + algos = [{'vpp': VppEnum.vl_api_ipsec_integ_alg_t. + IPSEC_API_INTEG_ALG_SHA1_96, + 'scapy': "HMAC-SHA1-96"}, + {'vpp': VppEnum.vl_api_ipsec_integ_alg_t. + IPSEC_API_INTEG_ALG_SHA_256_128, + 'scapy': "SHA2-256-128"}, + {'vpp': VppEnum.vl_api_ipsec_integ_alg_t. + IPSEC_API_INTEG_ALG_SHA_384_192, + 'scapy': "SHA2-384-192"}, + {'vpp': VppEnum.vl_api_ipsec_integ_alg_t. + IPSEC_API_INTEG_ALG_SHA_512_256, + 'scapy': "SHA2-512-256"}] + + flags = [0, (VppEnum.vl_api_ipsec_sad_flags_t. + IPSEC_API_SAD_FLAG_USE_ESN)] + + # + # loop through each of the algorithms + # + for algo in algos: + # with self.subTest(algo=algo['scapy']): + for flag in flags: + # + # setup up the config paramters + # + self.ipv4_params = IPsecIPv4Params() + self.ipv6_params = IPsecIPv6Params() + + self.params = {self.ipv4_params.addr_type: + self.ipv4_params, + self.ipv6_params.addr_type: + self.ipv6_params} + + for _, p in self.params.items(): + p.auth_algo_vpp_id = algo['vpp'] + p.auth_algo = algo['scapy'] + p.flags = p.flags | flag + + # + # configure the SPDs. SAs, etc + # + self.config_network(self.params.values()) + + # + # run some traffic. + # An exhautsive 4o6, 6o4 is not necessary for each algo + # + self.verify_tra_basic6(count=17) + self.verify_tra_basic4(count=17) + self.verify_tun_66(self.params[socket.AF_INET6], count=17) + self.verify_tun_44(self.params[socket.AF_INET], count=17) + + # + # remove the SPDs, SAs, etc + # + self.unconfig_network() if __name__ == '__main__': diff --git a/test/test_ipsec_esp.py b/test/test_ipsec_esp.py index c3a6b43996c..283914071ca 100644 --- a/test/test_ipsec_esp.py +++ b/test/test_ipsec_esp.py @@ -5,149 +5,227 @@ from scapy.layers.inet import UDP from framework import VppTestRunner from template_ipsec import IpsecTra46Tests, IpsecTun46Tests, TemplateIpsec, \ - IpsecTcpTests, IpsecTun4Tests, IpsecTra4Tests, config_tra_params + IpsecTcpTests, IpsecTun4Tests, IpsecTra4Tests, config_tra_params, \ + IPsecIPv4Params, IPsecIPv6Params, \ + IpsecTra4, IpsecTun4, IpsecTra6, IpsecTun6 from vpp_ipsec import VppIpsecSpd, VppIpsecSpdEntry, VppIpsecSA,\ - VppIpsecSpdItfBinding + VppIpsecSpdItfBinding from vpp_ip_route import VppIpRoute, VppRoutePath from vpp_ip import DpoProto from vpp_papi import VppEnum -def config_esp_tun(test, params): - addr_type = params.addr_type - scapy_tun_sa_id = params.scapy_tun_sa_id - scapy_tun_spi = params.scapy_tun_spi - vpp_tun_sa_id = params.vpp_tun_sa_id - vpp_tun_spi = params.vpp_tun_spi - auth_algo_vpp_id = params.auth_algo_vpp_id - auth_key = params.auth_key - crypt_algo_vpp_id = params.crypt_algo_vpp_id - crypt_key = params.crypt_key - remote_tun_if_host = params.remote_tun_if_host - addr_any = params.addr_any - addr_bcast = params.addr_bcast - e = VppEnum.vl_api_ipsec_spd_action_t - - params.tun_sa_in = VppIpsecSA(test, scapy_tun_sa_id, scapy_tun_spi, - auth_algo_vpp_id, auth_key, - crypt_algo_vpp_id, crypt_key, - test.vpp_esp_protocol, - test.tun_if.local_addr[addr_type], - test.tun_if.remote_addr[addr_type]) - params.tun_sa_in.add_vpp_config() - params.tun_sa_out = VppIpsecSA(test, vpp_tun_sa_id, vpp_tun_spi, - auth_algo_vpp_id, auth_key, - crypt_algo_vpp_id, crypt_key, - test.vpp_esp_protocol, - test.tun_if.remote_addr[addr_type], - test.tun_if.local_addr[addr_type]) - params.tun_sa_out.add_vpp_config() - - params.spd_policy_in_any = VppIpsecSpdEntry(test, test.tun_spd, - scapy_tun_sa_id, - addr_any, addr_bcast, - addr_any, addr_bcast, - socket.IPPROTO_ESP) - params.spd_policy_in_any.add_vpp_config() - params.spd_policy_out_any = VppIpsecSpdEntry(test, test.tun_spd, - scapy_tun_sa_id, - addr_any, addr_bcast, - addr_any, addr_bcast, - socket.IPPROTO_ESP, - is_outbound=0) - params.spd_policy_out_any.add_vpp_config() - - VppIpsecSpdEntry(test, test.tun_spd, vpp_tun_sa_id, - remote_tun_if_host, remote_tun_if_host, - test.pg1.remote_addr[addr_type], - test.pg1.remote_addr[addr_type], - 0, - priority=10, - policy=e.IPSEC_API_SPD_ACTION_PROTECT, - is_outbound=0).add_vpp_config() - VppIpsecSpdEntry(test, test.tun_spd, scapy_tun_sa_id, - test.pg1.remote_addr[addr_type], - test.pg1.remote_addr[addr_type], - remote_tun_if_host, remote_tun_if_host, - 0, - policy=e.IPSEC_API_SPD_ACTION_PROTECT, - priority=10).add_vpp_config() - - VppIpsecSpdEntry(test, test.tun_spd, vpp_tun_sa_id, - remote_tun_if_host, remote_tun_if_host, - test.pg0.local_addr[addr_type], - test.pg0.local_addr[addr_type], - 0, - priority=20, - policy=e.IPSEC_API_SPD_ACTION_PROTECT, - is_outbound=0).add_vpp_config() - VppIpsecSpdEntry(test, test.tun_spd, scapy_tun_sa_id, - test.pg0.local_addr[addr_type], - test.pg0.local_addr[addr_type], - remote_tun_if_host, remote_tun_if_host, - 0, - policy=e.IPSEC_API_SPD_ACTION_PROTECT, - priority=20).add_vpp_config() - - -def config_esp_tra(test, params): - addr_type = params.addr_type - scapy_tra_sa_id = params.scapy_tra_sa_id - scapy_tra_spi = params.scapy_tra_spi - vpp_tra_sa_id = params.vpp_tra_sa_id - vpp_tra_spi = params.vpp_tra_spi - auth_algo_vpp_id = params.auth_algo_vpp_id - auth_key = params.auth_key - crypt_algo_vpp_id = params.crypt_algo_vpp_id - crypt_key = params.crypt_key - addr_any = params.addr_any - addr_bcast = params.addr_bcast - flags = (VppEnum.vl_api_ipsec_sad_flags_t. - IPSEC_API_SAD_FLAG_USE_ANTI_REPLAY) - e = VppEnum.vl_api_ipsec_spd_action_t - flags = params.flags | flags - - params.tra_sa_in = VppIpsecSA(test, scapy_tra_sa_id, scapy_tra_spi, - auth_algo_vpp_id, auth_key, - crypt_algo_vpp_id, crypt_key, - test.vpp_esp_protocol, - flags=flags) - params.tra_sa_in.add_vpp_config() - params.tra_sa_out = VppIpsecSA(test, vpp_tra_sa_id, vpp_tra_spi, - auth_algo_vpp_id, auth_key, - crypt_algo_vpp_id, crypt_key, - test.vpp_esp_protocol, - flags=flags) - params.tra_sa_out.add_vpp_config() - - VppIpsecSpdEntry(test, test.tra_spd, vpp_tra_sa_id, - addr_any, addr_bcast, - addr_any, addr_bcast, - socket.IPPROTO_ESP).add_vpp_config() - VppIpsecSpdEntry(test, test.tra_spd, vpp_tra_sa_id, - addr_any, addr_bcast, - addr_any, addr_bcast, - socket.IPPROTO_ESP, - is_outbound=0).add_vpp_config() - - VppIpsecSpdEntry(test, test.tra_spd, vpp_tra_sa_id, - test.tra_if.local_addr[addr_type], - test.tra_if.local_addr[addr_type], - test.tra_if.remote_addr[addr_type], - test.tra_if.remote_addr[addr_type], - 0, priority=10, - policy=e.IPSEC_API_SPD_ACTION_PROTECT, - is_outbound=0).add_vpp_config() - VppIpsecSpdEntry(test, test.tra_spd, scapy_tra_sa_id, - test.tra_if.local_addr[addr_type], - test.tra_if.local_addr[addr_type], - test.tra_if.remote_addr[addr_type], - test.tra_if.remote_addr[addr_type], - 0, policy=e.IPSEC_API_SPD_ACTION_PROTECT, - priority=10).add_vpp_config() - - -class TemplateIpsecEsp(TemplateIpsec): +class ConfigIpsecESP(TemplateIpsec): + encryption_type = ESP + tra4_encrypt_node_name = "esp4-encrypt" + tra4_decrypt_node_name = "esp4-decrypt" + tra6_encrypt_node_name = "esp6-encrypt" + tra6_decrypt_node_name = "esp6-decrypt" + tun4_encrypt_node_name = "esp4-encrypt" + tun4_decrypt_node_name = "esp4-decrypt" + tun6_encrypt_node_name = "esp6-encrypt" + tun6_decrypt_node_name = "esp6-decrypt" + + @classmethod + def setUpClass(cls): + super(ConfigIpsecESP, cls).setUpClass() + + @classmethod + def tearDownClass(cls): + super(ConfigIpsecESP, cls).tearDownClass() + + def setUp(self): + super(ConfigIpsecESP, self).setUp() + + def tearDown(self): + super(ConfigIpsecESP, self).tearDown() + + def config_network(self, params): + self.net_objs = [] + self.tun_if = self.pg0 + self.tra_if = self.pg2 + self.logger.info(self.vapi.ppcli("show int addr")) + + self.tra_spd = VppIpsecSpd(self, self.tra_spd_id) + self.tra_spd.add_vpp_config() + self.net_objs.append(self.tra_spd) + self.tun_spd = VppIpsecSpd(self, self.tun_spd_id) + self.tun_spd.add_vpp_config() + self.net_objs.append(self.tun_spd) + + b = VppIpsecSpdItfBinding(self, self.tun_spd, + self.tun_if) + b.add_vpp_config() + self.net_objs.append(b) + + b = VppIpsecSpdItfBinding(self, self.tra_spd, + self.tra_if) + b.add_vpp_config() + self.net_objs.append(b) + + for p in params: + self.config_esp_tra(p) + config_tra_params(p, self.encryption_type) + for p in params: + self.config_esp_tun(p) + + for p in params: + d = DpoProto.DPO_PROTO_IP6 if p.is_ipv6 else DpoProto.DPO_PROTO_IP4 + r = VppIpRoute(self, p.remote_tun_if_host, p.addr_len, + [VppRoutePath(self.tun_if.remote_addr[p.addr_type], + 0xffffffff, + proto=d)], + is_ip6=p.is_ipv6) + r.add_vpp_config() + self.net_objs.append(r) + + self.logger.info(self.vapi.ppcli("show ipsec all")) + + def unconfig_network(self): + for o in reversed(self.net_objs): + o.remove_vpp_config() + self.net_objs = [] + + def config_esp_tun(self, params): + addr_type = params.addr_type + scapy_tun_sa_id = params.scapy_tun_sa_id + scapy_tun_spi = params.scapy_tun_spi + vpp_tun_sa_id = params.vpp_tun_sa_id + vpp_tun_spi = params.vpp_tun_spi + auth_algo_vpp_id = params.auth_algo_vpp_id + auth_key = params.auth_key + crypt_algo_vpp_id = params.crypt_algo_vpp_id + crypt_key = params.crypt_key + remote_tun_if_host = params.remote_tun_if_host + addr_any = params.addr_any + addr_bcast = params.addr_bcast + e = VppEnum.vl_api_ipsec_spd_action_t + objs = [] + + params.tun_sa_in = VppIpsecSA(self, scapy_tun_sa_id, scapy_tun_spi, + auth_algo_vpp_id, auth_key, + crypt_algo_vpp_id, crypt_key, + self.vpp_esp_protocol, + self.tun_if.local_addr[addr_type], + self.tun_if.remote_addr[addr_type]) + params.tun_sa_out = VppIpsecSA(self, vpp_tun_sa_id, vpp_tun_spi, + auth_algo_vpp_id, auth_key, + crypt_algo_vpp_id, crypt_key, + self.vpp_esp_protocol, + self.tun_if.remote_addr[addr_type], + self.tun_if.local_addr[addr_type]) + objs.append(params.tun_sa_in) + objs.append(params.tun_sa_out) + + params.spd_policy_in_any = VppIpsecSpdEntry(self, self.tun_spd, + scapy_tun_sa_id, + addr_any, addr_bcast, + addr_any, addr_bcast, + socket.IPPROTO_ESP) + params.spd_policy_out_any = VppIpsecSpdEntry(self, self.tun_spd, + scapy_tun_sa_id, + addr_any, addr_bcast, + addr_any, addr_bcast, + socket.IPPROTO_ESP, + is_outbound=0) + objs.append(params.spd_policy_out_any) + objs.append(params.spd_policy_in_any) + + objs.append(VppIpsecSpdEntry(self, self.tun_spd, vpp_tun_sa_id, + remote_tun_if_host, remote_tun_if_host, + self.pg1.remote_addr[addr_type], + self.pg1.remote_addr[addr_type], + 0, + priority=10, + policy=e.IPSEC_API_SPD_ACTION_PROTECT, + is_outbound=0)) + objs.append(VppIpsecSpdEntry(self, self.tun_spd, scapy_tun_sa_id, + self.pg1.remote_addr[addr_type], + self.pg1.remote_addr[addr_type], + remote_tun_if_host, remote_tun_if_host, + 0, + policy=e.IPSEC_API_SPD_ACTION_PROTECT, + priority=10)) + objs.append(VppIpsecSpdEntry(self, self.tun_spd, vpp_tun_sa_id, + remote_tun_if_host, remote_tun_if_host, + self.pg0.local_addr[addr_type], + self.pg0.local_addr[addr_type], + 0, + priority=20, + policy=e.IPSEC_API_SPD_ACTION_PROTECT, + is_outbound=0)) + objs.append(VppIpsecSpdEntry(self, self.tun_spd, scapy_tun_sa_id, + self.pg0.local_addr[addr_type], + self.pg0.local_addr[addr_type], + remote_tun_if_host, remote_tun_if_host, + 0, + policy=e.IPSEC_API_SPD_ACTION_PROTECT, + priority=20)) + for o in objs: + o.add_vpp_config() + self.net_objs = self.net_objs + objs + + def config_esp_tra(self, params): + addr_type = params.addr_type + scapy_tra_sa_id = params.scapy_tra_sa_id + scapy_tra_spi = params.scapy_tra_spi + vpp_tra_sa_id = params.vpp_tra_sa_id + vpp_tra_spi = params.vpp_tra_spi + auth_algo_vpp_id = params.auth_algo_vpp_id + auth_key = params.auth_key + crypt_algo_vpp_id = params.crypt_algo_vpp_id + crypt_key = params.crypt_key + addr_any = params.addr_any + addr_bcast = params.addr_bcast + flags = (VppEnum.vl_api_ipsec_sad_flags_t. + IPSEC_API_SAD_FLAG_USE_ANTI_REPLAY) + e = VppEnum.vl_api_ipsec_spd_action_t + flags = params.flags | flags + objs = [] + + params.tra_sa_in = VppIpsecSA(self, scapy_tra_sa_id, scapy_tra_spi, + auth_algo_vpp_id, auth_key, + crypt_algo_vpp_id, crypt_key, + self.vpp_esp_protocol, + flags=flags) + params.tra_sa_out = VppIpsecSA(self, vpp_tra_sa_id, vpp_tra_spi, + auth_algo_vpp_id, auth_key, + crypt_algo_vpp_id, crypt_key, + self.vpp_esp_protocol, + flags=flags) + objs.append(params.tra_sa_in) + objs.append(params.tra_sa_out) + + objs.append(VppIpsecSpdEntry(self, self.tra_spd, vpp_tra_sa_id, + addr_any, addr_bcast, + addr_any, addr_bcast, + socket.IPPROTO_ESP)) + objs.append(VppIpsecSpdEntry(self, self.tra_spd, vpp_tra_sa_id, + addr_any, addr_bcast, + addr_any, addr_bcast, + socket.IPPROTO_ESP, + is_outbound=0)) + objs.append(VppIpsecSpdEntry(self, self.tra_spd, vpp_tra_sa_id, + self.tra_if.local_addr[addr_type], + self.tra_if.local_addr[addr_type], + self.tra_if.remote_addr[addr_type], + self.tra_if.remote_addr[addr_type], + 0, priority=10, + policy=e.IPSEC_API_SPD_ACTION_PROTECT, + is_outbound=0)) + objs.append(VppIpsecSpdEntry(self, self.tra_spd, scapy_tra_sa_id, + self.tra_if.local_addr[addr_type], + self.tra_if.local_addr[addr_type], + self.tra_if.remote_addr[addr_type], + self.tra_if.remote_addr[addr_type], + 0, policy=e.IPSEC_API_SPD_ACTION_PROTECT, + priority=10)) + for o in objs: + o.add_vpp_config() + self.net_objs = self.net_objs + objs + + +class TemplateIpsecEsp(ConfigIpsecESP): """ Basic test for ipsec esp sanity - tunnel and transport modes. @@ -188,54 +266,16 @@ class TemplateIpsecEsp(TemplateIpsec): def setUp(self): super(TemplateIpsecEsp, self).setUp() - self.encryption_type = ESP - self.tun_if = self.pg0 - self.tra_if = self.pg2 - self.logger.info(self.vapi.ppcli("show int addr")) - - self.tra_spd = VppIpsecSpd(self, self.tra_spd_id) - self.tra_spd.add_vpp_config() - VppIpsecSpdItfBinding(self, self.tra_spd, - self.tra_if).add_vpp_config() - - for _, p in self.params.items(): - config_esp_tra(self, p) - config_tra_params(p, self.encryption_type) - self.logger.info(self.vapi.ppcli("show ipsec")) - - self.tun_spd = VppIpsecSpd(self, self.tun_spd_id) - self.tun_spd.add_vpp_config() - VppIpsecSpdItfBinding(self, self.tun_spd, - self.tun_if).add_vpp_config() - - for _, p in self.params.items(): - config_esp_tun(self, p) - self.logger.info(self.vapi.ppcli("show ipsec")) - - for _, p in self.params.items(): - d = DpoProto.DPO_PROTO_IP6 if p.is_ipv6 else DpoProto.DPO_PROTO_IP4 - VppIpRoute(self, p.remote_tun_if_host, p.addr_len, - [VppRoutePath(self.tun_if.remote_addr[p.addr_type], - 0xffffffff, - proto=d)], - is_ip6=p.is_ipv6).add_vpp_config() + self.config_network(self.params.values()) def tearDown(self): + self.unconfig_network() super(TemplateIpsecEsp, self).tearDown() - if not self.vpp_dead: - self.vapi.cli("show hardware") class TestIpsecEsp1(TemplateIpsecEsp, IpsecTra46Tests, IpsecTun46Tests): """ Ipsec ESP - TUN & TRA tests """ - tra4_encrypt_node_name = "esp4-encrypt" - tra4_decrypt_node_name = "esp4-decrypt" - tra6_encrypt_node_name = "esp6-encrypt" - tra6_decrypt_node_name = "esp6-decrypt" - tun4_encrypt_node_name = "esp4-encrypt" - tun4_decrypt_node_name = "esp4-decrypt" - tun6_encrypt_node_name = "esp6-encrypt" - tun6_decrypt_node_name = "esp6-decrypt" + pass class TestIpsecEsp2(TemplateIpsecEsp, IpsecTcpTests): @@ -243,7 +283,7 @@ class TestIpsecEsp2(TemplateIpsecEsp, IpsecTcpTests): pass -class TemplateIpsecEspUdp(TemplateIpsec): +class TemplateIpsecEspUdp(ConfigIpsecESP): """ UDP encapped ESP """ @@ -258,7 +298,7 @@ class TemplateIpsecEspUdp(TemplateIpsec): def setUp(self): super(TemplateIpsecEspUdp, self).setUp() - self.encryption_type = ESP + self.net_objs = [] self.tun_if = self.pg0 self.tra_if = self.pg2 self.logger.info(self.vapi.ppcli("show int addr")) @@ -273,7 +313,7 @@ class TemplateIpsecEspUdp(TemplateIpsec): VppIpsecSpdItfBinding(self, self.tra_spd, self.tra_if).add_vpp_config() - config_esp_tra(self, p) + self.config_esp_tra(p) config_tra_params(p, self.encryption_type) self.tun_spd = VppIpsecSpd(self, self.tun_spd_id) @@ -281,7 +321,7 @@ class TemplateIpsecEspUdp(TemplateIpsec): VppIpsecSpdItfBinding(self, self.tun_spd, self.tun_if).add_vpp_config() - config_esp_tun(self, p) + self.config_esp_tun(p) self.logger.info(self.vapi.ppcli("show ipsec")) d = DpoProto.DPO_PROTO_IP4 @@ -298,12 +338,93 @@ class TemplateIpsecEspUdp(TemplateIpsec): class TestIpsecEspUdp(TemplateIpsecEspUdp, IpsecTra4Tests, IpsecTun4Tests): """ Ipsec NAT-T ESP UDP tests """ - tra4_encrypt_node_name = "esp4-encrypt" - tra4_decrypt_node_name = "esp4-decrypt" - tun4_encrypt_node_name = "esp4-encrypt" - tun4_decrypt_node_name = "esp4-decrypt" pass +class TestIpsecEspAll(ConfigIpsecESP, + IpsecTra4, IpsecTra6, + IpsecTun4, IpsecTun6): + """ Ipsec ESP all Algos """ + + def setUp(self): + super(TestIpsecEspAll, self).setUp() + + def tearDown(self): + super(TestIpsecEspAll, self).tearDown() + + def test_crypto_algs(self): + """All engines AES-CBC-[128, 192, 256] w/o ESN""" + + # foreach VPP crypto engine + engines = ["ia32", "openssl"] + + # foreach crypto algorithm + algos = [{'vpp': VppEnum.vl_api_ipsec_crypto_alg_t. + IPSEC_API_CRYPTO_ALG_AES_CBC_128, + 'scapy': "AES-CBC", + 'key': "JPjyOWBeVEQiMe7h"}, + {'vpp': VppEnum.vl_api_ipsec_crypto_alg_t. + IPSEC_API_CRYPTO_ALG_AES_CBC_192, + 'scapy': "AES-CBC", + 'key': "JPjyOWBeVEQiMe7hJPjyOWBe"}, + {'vpp': VppEnum.vl_api_ipsec_crypto_alg_t. + IPSEC_API_CRYPTO_ALG_AES_CBC_256, + 'scapy': "AES-CBC", + 'key': "JPjyOWBeVEQiMe7hJPjyOWBeVEQiMe7h"}] + + # bug found in VPP needs fixing with flag + # (VppEnum.vl_api_ipsec_sad_flags_t.IPSEC_API_SAD_FLAG_USE_ESN) + flags = [0] + + # + # loop through the VPP engines + # + for engine in engines: + self.vapi.cli("set crypto engine all %s" % engine) + + # + # loop through each of the algorithms + # + for algo in algos: + # with self.subTest(algo=algo['scapy']): + for flag in flags: + # + # setup up the config paramters + # + self.ipv4_params = IPsecIPv4Params() + self.ipv6_params = IPsecIPv6Params() + + self.params = {self.ipv4_params.addr_type: + self.ipv4_params, + self.ipv6_params.addr_type: + self.ipv6_params} + + for _, p in self.params.items(): + p.crypt_algo_vpp_id = algo['vpp'] + p.crypt_algo = algo['scapy'] + p.crypt_key = algo['key'] + p.flags = p.flags | flag + + # + # configure the SPDs. SAs, etc + # + self.config_network(self.params.values()) + + # + # run some traffic. + # An exhautsive 4o6, 6o4 is not necessary + # for each algo + # + self.verify_tra_basic6(count=17) + self.verify_tra_basic4(count=17) + self.verify_tun_66(self.params[socket.AF_INET6], 17) + self.verify_tun_44(self.params[socket.AF_INET], 17) + + # + # remove the SPDs, SAs, etc + # + self.unconfig_network() + + if __name__ == '__main__': unittest.main(testRunner=VppTestRunner)