X-Git-Url: https://gerrit.fd.io/r/gitweb?a=blobdiff_plain;f=test%2Ftemplate_ipsec.py;h=7888a6788abfdcdc29179cabb8ee0d48148e1e41;hb=17dcec0b9;hp=bf13d71f631de77e09e90c8b74a08a1d3c52dbe0;hpb=611864f4bddf787aff3323f162da589b1b26529c;p=vpp.git diff --git a/test/template_ipsec.py b/test/template_ipsec.py index bf13d71f631..7888a6788ab 100644 --- a/test/template_ipsec.py +++ b/test/template_ipsec.py @@ -8,60 +8,71 @@ from scapy.layers.inet6 import IPv6, ICMPv6EchoRequest from framework import VppTestCase, VppTestRunner from util import ppp +from vpp_papi import VppEnum class IPsecIPv4Params(object): + addr_type = socket.AF_INET addr_any = "0.0.0.0" addr_bcast = "255.255.255.255" addr_len = 32 is_ipv6 = 0 - remote_tun_if_host = '1.1.1.1' - scapy_tun_sa_id = 10 - scapy_tun_spi = 1001 - vpp_tun_sa_id = 20 - vpp_tun_spi = 1000 + def __init__(self): + self.remote_tun_if_host = '1.1.1.1' + + self.scapy_tun_sa_id = 10 + self.scapy_tun_spi = 1001 + self.vpp_tun_sa_id = 20 + self.vpp_tun_spi = 1000 - scapy_tra_sa_id = 30 - scapy_tra_spi = 2001 - vpp_tra_sa_id = 40 - vpp_tra_spi = 2000 + self.scapy_tra_sa_id = 30 + self.scapy_tra_spi = 2001 + self.vpp_tra_sa_id = 40 + self.vpp_tra_spi = 2000 - auth_algo_vpp_id = 2 # internal VPP enum value for SHA1_96 - auth_algo = 'HMAC-SHA1-96' # scapy name - auth_key = 'C91KUR9GYMm5GfkEvNjX' + self.auth_algo_vpp_id = (VppEnum.vl_api_ipsec_integ_alg_t. + IPSEC_API_INTEG_ALG_SHA1_96) + self.auth_algo = 'HMAC-SHA1-96' # scapy name + self.auth_key = 'C91KUR9GYMm5GfkEvNjX' - crypt_algo_vpp_id = 1 # internal VPP enum value for AES_CBC_128 - crypt_algo = 'AES-CBC' # scapy name - crypt_key = 'JPjyOWBeVEQiMe7h' + self.crypt_algo_vpp_id = (VppEnum.vl_api_ipsec_crypto_alg_t. + IPSEC_API_CRYPTO_ALG_AES_CBC_128) + self.crypt_algo = 'AES-CBC' # scapy name + self.crypt_key = 'JPjyOWBeVEQiMe7h' class IPsecIPv6Params(object): + addr_type = socket.AF_INET6 addr_any = "0::0" addr_bcast = "ffff:ffff:ffff:ffff:ffff:ffff:ffff:ffff" addr_len = 128 is_ipv6 = 1 - remote_tun_if_host = '1111:1111:1111:1111:1111:1111:1111:1111' - scapy_tun_sa_id = 50 - scapy_tun_spi = 3001 - vpp_tun_sa_id = 60 - vpp_tun_spi = 3000 + def __init__(self): + self.remote_tun_if_host = '1111:1111:1111:1111:1111:1111:1111:1111' + + self.scapy_tun_sa_id = 50 + self.scapy_tun_spi = 3001 + self.vpp_tun_sa_id = 60 + self.vpp_tun_spi = 3000 - scapy_tra_sa_id = 70 - scapy_tra_spi = 4001 - vpp_tra_sa_id = 80 - vpp_tra_spi = 4000 + self.scapy_tra_sa_id = 70 + self.scapy_tra_spi = 4001 + self.vpp_tra_sa_id = 80 + self.vpp_tra_spi = 4000 - auth_algo_vpp_id = 4 # internal VPP enum value for SHA_256_128 - auth_algo = 'SHA2-256-128' # scapy name - auth_key = 'C91KUR9GYMm5GfkEvNjX' + self.auth_algo_vpp_id = (VppEnum.vl_api_ipsec_integ_alg_t. + IPSEC_API_INTEG_ALG_SHA_256_128) + self.auth_algo = 'SHA2-256-128' # scapy name + self.auth_key = 'C91KUR9GYMm5GfkEvNjX' - crypt_algo_vpp_id = 3 # internal VPP enum value for AES_CBC_256 - crypt_algo = 'AES-CBC' # scapy name - crypt_key = 'JPjyOWBeVEQiMe7hJPjyOWBeVEQiMe7h' + self.crypt_algo_vpp_id = (VppEnum.vl_api_ipsec_crypto_alg_t. + IPSEC_API_CRYPTO_ALG_AES_CBC_256) + self.crypt_algo = 'AES-CBC' # scapy name + self.crypt_key = 'JPjyOWBeVEQiMe7hJPjyOWBeVEQiMe7h' class TemplateIpsec(VppTestCase): @@ -82,33 +93,48 @@ class TemplateIpsec(VppTestCase): |tun_if| -------> |VPP| ------> |pg1| ------ --- --- """ - ipv4_params = IPsecIPv4Params() - ipv6_params = IPsecIPv6Params() - params = {ipv4_params.addr_type: ipv4_params, - ipv6_params.addr_type: ipv6_params} - payload = "XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX" + def ipsec_select_backend(self): + """ empty method to be overloaded when necessary """ + pass - tun_spd_id = 1 - tra_spd_id = 2 + def setUp(self): + super(TemplateIpsec, self).setUp() - vpp_esp_protocol = 1 - vpp_ah_protocol = 0 + 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} - @classmethod - def setUpClass(cls): - super(TemplateIpsec, cls).setUpClass() - cls.create_pg_interfaces(range(3)) - cls.interfaces = list(cls.pg_interfaces) - for i in cls.interfaces: + self.payload = "XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX"\ + "XXXXXXXXXXXXXXXXXXXXX" + + 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.ipsec_select_backend() def tearDown(self): super(TemplateIpsec, self).tearDown() + + for i in self.interfaces: + i.admin_down() + i.unconfig_ip4() + i.unconfig_ip6() + if not self.vpp_dead: self.vapi.cli("show hardware") @@ -153,19 +179,18 @@ class TemplateIpsec(VppTestCase): return vpp_tun_sa, scapy_tun_sa def configure_sa_tra(self, params): - scapy_tra_sa = SecurityAssociation(self.encryption_type, - spi=params.vpp_tra_spi, - crypt_algo=params.crypt_algo, - crypt_key=params.crypt_key, - auth_algo=params.auth_algo, - auth_key=params.auth_key) - vpp_tra_sa = SecurityAssociation(self.encryption_type, - spi=params.scapy_tra_spi, - crypt_algo=params.crypt_algo, - crypt_key=params.crypt_key, - auth_algo=params.auth_algo, - auth_key=params.auth_key) - return vpp_tra_sa, scapy_tra_sa + params.scapy_tra_sa = SecurityAssociation(self.encryption_type, + spi=params.vpp_tra_spi, + crypt_algo=params.crypt_algo, + crypt_key=params.crypt_key, + auth_algo=params.auth_algo, + auth_key=params.auth_key) + params.vpp_tra_sa = SecurityAssociation(self.encryption_type, + spi=params.scapy_tra_spi, + crypt_algo=params.crypt_algo, + crypt_key=params.crypt_key, + auth_algo=params.auth_algo, + auth_key=params.auth_key) class IpsecTcpTests(object): @@ -186,54 +211,131 @@ class IpsecTcpTests(object): class IpsecTraTests(object): + def test_tra_anti_replay(self, count=1): + """ ipsec v4 transport anti-reply test """ + p = self.params[socket.AF_INET] + + # fire in a packet with seq number 1 + pkt = (Ether(src=self.tra_if.remote_mac, + dst=self.tra_if.local_mac) / + p.scapy_tra_sa.encrypt(IP(src=self.tra_if.remote_ip4, + dst=self.tra_if.local_ip4) / + ICMP(), + seq_num=1)) + recv_pkts = self.send_and_expect(self.tra_if, [pkt], self.tra_if) + + # now move the window over to 235 + pkt = (Ether(src=self.tra_if.remote_mac, + dst=self.tra_if.local_mac) / + p.scapy_tra_sa.encrypt(IP(src=self.tra_if.remote_ip4, + dst=self.tra_if.local_ip4) / + ICMP(), + seq_num=235)) + recv_pkts = self.send_and_expect(self.tra_if, [pkt], self.tra_if) + + # the window size is 64 packets + # in window are still accepted + pkt = (Ether(src=self.tra_if.remote_mac, + dst=self.tra_if.local_mac) / + p.scapy_tra_sa.encrypt(IP(src=self.tra_if.remote_ip4, + dst=self.tra_if.local_ip4) / + ICMP(), + seq_num=172)) + recv_pkts = self.send_and_expect(self.tra_if, [pkt], self.tra_if) + + # out of window are dropped + pkt = (Ether(src=self.tra_if.remote_mac, + dst=self.tra_if.local_mac) / + p.scapy_tra_sa.encrypt(IP(src=self.tra_if.remote_ip4, + dst=self.tra_if.local_ip4) / + ICMP(), + seq_num=17)) + self.send_and_assert_no_replies(self.tra_if, pkt * 17) + + self.assert_packet_counter_equal( + '/err/%s/SA replayed packet' % self.tra4_decrypt_node_name, 17) + + # a packet that does not decrypt does not move the window forward + bogus_sa = SecurityAssociation(self.encryption_type, + p.vpp_tra_spi) + pkt = (Ether(src=self.tra_if.remote_mac, + dst=self.tra_if.local_mac) / + bogus_sa.encrypt(IP(src=self.tra_if.remote_ip4, + dst=self.tra_if.local_ip4) / + ICMP(), + seq_num=350)) + self.send_and_assert_no_replies(self.tra_if, pkt * 17) + + self.assert_packet_counter_equal( + '/err/%s/Integrity check failed' % self.tra4_decrypt_node_name, 17) + + # which we can determine since this packet is still in the window + pkt = (Ether(src=self.tra_if.remote_mac, + dst=self.tra_if.local_mac) / + p.scapy_tra_sa.encrypt(IP(src=self.tra_if.remote_ip4, + dst=self.tra_if.local_ip4) / + ICMP(), + seq_num=234)) + self.send_and_expect(self.tra_if, [pkt], self.tra_if) + + # move the security-associations seq number on to the last we used + p.scapy_tra_sa.seq_num = 351 + p.vpp_tra_sa.seq_num = 351 + def test_tra_basic(self, count=1): """ ipsec v4 transport basic test """ + self.vapi.cli("clear errors") try: p = self.params[socket.AF_INET] - vpp_tra_sa, scapy_tra_sa = self.configure_sa_tra(p) - send_pkts = self.gen_encrypt_pkts(scapy_tra_sa, self.tra_if, + send_pkts = self.gen_encrypt_pkts(p.scapy_tra_sa, self.tra_if, src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4, count=count) recv_pkts = self.send_and_expect(self.tra_if, send_pkts, self.tra_if) - for p in recv_pkts: + for rx in recv_pkts: try: - decrypted = vpp_tra_sa.decrypt(p[IP]) + decrypted = p.vpp_tra_sa.decrypt(rx[IP]) self.assert_packet_checksums_valid(decrypted) except: - self.logger.debug(ppp("Unexpected packet:", p)) + self.logger.debug(ppp("Unexpected packet:", rx)) raise finally: self.logger.info(self.vapi.ppcli("show error")) self.logger.info(self.vapi.ppcli("show ipsec")) + self.assert_packet_counter_equal(self.tra4_encrypt_node_name, count) + self.assert_packet_counter_equal(self.tra4_decrypt_node_name, count) + def test_tra_burst(self): """ ipsec v4 transport burst test """ self.test_tra_basic(count=257) def test_tra_basic6(self, count=1): """ ipsec v6 transport basic test """ + self.vapi.cli("clear errors") try: p = self.params[socket.AF_INET6] - vpp_tra_sa, scapy_tra_sa = self.configure_sa_tra(p) - send_pkts = self.gen_encrypt_pkts6(scapy_tra_sa, self.tra_if, + send_pkts = self.gen_encrypt_pkts6(p.scapy_tra_sa, self.tra_if, src=self.tra_if.remote_ip6, dst=self.tra_if.local_ip6, count=count) recv_pkts = self.send_and_expect(self.tra_if, send_pkts, self.tra_if) - for p in recv_pkts: + for rx in recv_pkts: try: - decrypted = vpp_tra_sa.decrypt(p[IPv6]) + decrypted = p.vpp_tra_sa.decrypt(rx[IPv6]) self.assert_packet_checksums_valid(decrypted) except: - self.logger.debug(ppp("Unexpected packet:", p)) + self.logger.debug(ppp("Unexpected packet:", rx)) raise finally: self.logger.info(self.vapi.ppcli("show error")) self.logger.info(self.vapi.ppcli("show ipsec")) + self.assert_packet_counter_equal(self.tra6_encrypt_node_name, count) + self.assert_packet_counter_equal(self.tra6_decrypt_node_name, count) + def test_tra_burst6(self): """ ipsec v6 transport burst test """ self.test_tra_basic6(count=257) @@ -242,6 +344,7 @@ class IpsecTraTests(object): class IpsecTun4Tests(object): def test_tun_basic44(self, count=1): """ ipsec 4o4 tunnel basic test """ + self.vapi.cli("clear errors") try: p = self.params[socket.AF_INET] vpp_tun_sa, scapy_tun_sa = self.configure_sa_tun(p) @@ -277,6 +380,9 @@ class IpsecTun4Tests(object): self.logger.info(self.vapi.ppcli("show error")) self.logger.info(self.vapi.ppcli("show ipsec")) + self.assert_packet_counter_equal(self.tun4_encrypt_node_name, count) + self.assert_packet_counter_equal(self.tun4_decrypt_node_name, count) + def test_tun_burst44(self): """ ipsec 4o4 tunnel burst test """ self.test_tun_basic44(count=257) @@ -285,6 +391,7 @@ class IpsecTun4Tests(object): class IpsecTun6Tests(object): def test_tun_basic66(self, count=1): """ ipsec 6o6 tunnel basic test """ + self.vapi.cli("clear errors") try: p = self.params[socket.AF_INET6] vpp_tun_sa, scapy_tun_sa = self.configure_sa_tun(p) @@ -321,6 +428,9 @@ class IpsecTun6Tests(object): self.logger.info(self.vapi.ppcli("show error")) self.logger.info(self.vapi.ppcli("show ipsec")) + self.assert_packet_counter_equal(self.tun6_encrypt_node_name, count) + self.assert_packet_counter_equal(self.tun6_decrypt_node_name, count) + def test_tun_burst66(self): """ ipsec 6o6 tunnel burst test """ self.test_tun_basic66(count=257)