5 from scapy.layers.inet import IP, ICMP, TCP, UDP
6 from scapy.layers.ipsec import SecurityAssociation, ESP
7 from scapy.layers.l2 import Ether
8 from scapy.packet import raw, Raw
9 from scapy.layers.inet6 import (
18 from framework import VppTestCase, VppTestRunner
19 from util import ppp, reassemble4, fragment_rfc791, fragment_rfc8200
20 from vpp_papi import VppEnum
22 from vpp_ipsec import VppIpsecSpd, VppIpsecSpdEntry, VppIpsecSpdItfBinding
23 from ipaddress import ip_address
28 class IPsecIPv4Params:
30 addr_type = socket.AF_INET
32 addr_bcast = "255.255.255.255"
37 self.remote_tun_if_host = "1.1.1.1"
38 self.remote_tun_if_host6 = "1111::1"
40 self.scapy_tun_sa_id = 100
41 self.scapy_tun_spi = 1000
42 self.vpp_tun_sa_id = 200
43 self.vpp_tun_spi = 2000
45 self.scapy_tra_sa_id = 300
46 self.scapy_tra_spi = 3000
47 self.vpp_tra_sa_id = 400
48 self.vpp_tra_spi = 4000
50 self.outer_hop_limit = 64
51 self.inner_hop_limit = 255
52 self.outer_flow_label = 0
53 self.inner_flow_label = 0x12345
55 self.auth_algo_vpp_id = (
56 VppEnum.vl_api_ipsec_integ_alg_t.IPSEC_API_INTEG_ALG_SHA1_96
58 self.auth_algo = "HMAC-SHA1-96" # scapy name
59 self.auth_key = b"C91KUR9GYMm5GfkEvNjX"
61 self.crypt_algo_vpp_id = (
62 VppEnum.vl_api_ipsec_crypto_alg_t.IPSEC_API_CRYPTO_ALG_AES_CBC_128
64 self.crypt_algo = "AES-CBC" # scapy name
65 self.crypt_key = b"JPjyOWBeVEQiMe7h"
68 self.nat_header = None
70 VppEnum.vl_api_tunnel_encap_decap_flags_t.TUNNEL_API_ENCAP_DECAP_FLAG_NONE
73 self.async_mode = False
76 class IPsecIPv6Params:
78 addr_type = socket.AF_INET6
80 addr_bcast = "ffff:ffff:ffff:ffff:ffff:ffff:ffff:ffff"
85 self.remote_tun_if_host = "1111:1111:1111:1111:1111:1111:1111:1111"
86 self.remote_tun_if_host4 = "1.1.1.1"
88 self.scapy_tun_sa_id = 500
89 self.scapy_tun_spi = 3001
90 self.vpp_tun_sa_id = 600
91 self.vpp_tun_spi = 3000
93 self.scapy_tra_sa_id = 700
94 self.scapy_tra_spi = 4001
95 self.vpp_tra_sa_id = 800
96 self.vpp_tra_spi = 4000
98 self.outer_hop_limit = 64
99 self.inner_hop_limit = 255
100 self.outer_flow_label = 0
101 self.inner_flow_label = 0x12345
103 self.auth_algo_vpp_id = (
104 VppEnum.vl_api_ipsec_integ_alg_t.IPSEC_API_INTEG_ALG_SHA1_96
106 self.auth_algo = "HMAC-SHA1-96" # scapy name
107 self.auth_key = b"C91KUR9GYMm5GfkEvNjX"
109 self.crypt_algo_vpp_id = (
110 VppEnum.vl_api_ipsec_crypto_alg_t.IPSEC_API_CRYPTO_ALG_AES_CBC_128
112 self.crypt_algo = "AES-CBC" # scapy name
113 self.crypt_key = b"JPjyOWBeVEQiMe7h"
116 self.nat_header = None
118 VppEnum.vl_api_tunnel_encap_decap_flags_t.TUNNEL_API_ENCAP_DECAP_FLAG_NONE
121 self.async_mode = False
124 def mk_scapy_crypt_key(p):
125 if p.crypt_algo in ("AES-GCM", "AES-CTR"):
126 return p.crypt_key + struct.pack("!I", p.salt)
131 def config_tun_params(p, encryption_type, tun_if):
132 ip_class_by_addr_type = {socket.AF_INET: IP, socket.AF_INET6: IPv6}
134 p.flags & (VppEnum.vl_api_ipsec_sad_flags_t.IPSEC_API_SAD_FLAG_USE_ESN)
136 p.tun_dst = tun_if.remote_addr[p.addr_type]
137 p.tun_src = tun_if.local_addr[p.addr_type]
138 crypt_key = mk_scapy_crypt_key(p)
139 p.scapy_tun_sa = SecurityAssociation(
142 crypt_algo=p.crypt_algo,
144 auth_algo=p.auth_algo,
146 tunnel_header=ip_class_by_addr_type[p.addr_type](src=p.tun_dst, dst=p.tun_src),
147 nat_t_header=p.nat_header,
150 p.vpp_tun_sa = SecurityAssociation(
153 crypt_algo=p.crypt_algo,
155 auth_algo=p.auth_algo,
157 tunnel_header=ip_class_by_addr_type[p.addr_type](dst=p.tun_dst, src=p.tun_src),
158 nat_t_header=p.nat_header,
163 def config_tra_params(p, encryption_type):
165 p.flags & (VppEnum.vl_api_ipsec_sad_flags_t.IPSEC_API_SAD_FLAG_USE_ESN)
167 crypt_key = mk_scapy_crypt_key(p)
168 p.scapy_tra_sa = SecurityAssociation(
171 crypt_algo=p.crypt_algo,
173 auth_algo=p.auth_algo,
175 nat_t_header=p.nat_header,
178 p.vpp_tra_sa = SecurityAssociation(
181 crypt_algo=p.crypt_algo,
183 auth_algo=p.auth_algo,
185 nat_t_header=p.nat_header,
190 class TemplateIpsec(VppTestCase):
195 |tra_if| <-------> |VPP|
200 ------ encrypt --- plain ---
201 |tun_if| <------- |VPP| <------ |pg1|
204 ------ decrypt --- plain ---
205 |tun_if| -------> |VPP| ------> |pg1|
212 def ipsec_select_backend(self):
213 """empty method to be overloaded when necessary"""
218 super(TemplateIpsec, cls).setUpClass()
221 def tearDownClass(cls):
222 super(TemplateIpsec, cls).tearDownClass()
224 def setup_params(self):
225 if not hasattr(self, "ipv4_params"):
226 self.ipv4_params = IPsecIPv4Params()
227 if not hasattr(self, "ipv6_params"):
228 self.ipv6_params = IPsecIPv6Params()
230 self.ipv4_params.addr_type: self.ipv4_params,
231 self.ipv6_params.addr_type: self.ipv6_params,
234 def config_interfaces(self):
235 self.create_pg_interfaces(range(3))
236 self.interfaces = list(self.pg_interfaces)
237 for i in self.interfaces:
245 super(TemplateIpsec, self).setUp()
249 self.vpp_esp_protocol = VppEnum.vl_api_ipsec_proto_t.IPSEC_API_PROTO_ESP
250 self.vpp_ah_protocol = VppEnum.vl_api_ipsec_proto_t.IPSEC_API_PROTO_AH
252 self.config_interfaces()
254 self.ipsec_select_backend()
256 def unconfig_interfaces(self):
257 for i in self.interfaces:
263 super(TemplateIpsec, self).tearDown()
265 self.unconfig_interfaces()
267 def show_commands_at_teardown(self):
268 self.logger.info(self.vapi.cli("show hardware"))
270 def gen_encrypt_pkts(self, p, sa, sw_intf, src, dst, count=1, payload_size=54):
272 Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac)
273 / sa.encrypt(IP(src=src, dst=dst) / ICMP() / Raw(b"X" * payload_size))
274 for i in range(count)
277 def gen_encrypt_pkts6(self, p, sa, sw_intf, src, dst, count=1, payload_size=54):
279 Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac)
281 IPv6(src=src, dst=dst, hlim=p.inner_hop_limit, fl=p.inner_flow_label)
282 / ICMPv6EchoRequest(id=0, seq=1, data="X" * payload_size)
284 for i in range(count)
287 def gen_pkts(self, sw_intf, src, dst, count=1, payload_size=54):
289 Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac)
290 / IP(src=src, dst=dst)
292 / Raw(b"X" * payload_size)
293 for i in range(count)
296 def gen_pkts6(self, p, sw_intf, src, dst, count=1, payload_size=54):
298 Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac)
299 / IPv6(src=src, dst=dst, hlim=p.inner_hop_limit, fl=p.inner_flow_label)
300 / ICMPv6EchoRequest(id=0, seq=1, data="X" * payload_size)
301 for i in range(count)
305 class IpsecTcp(object):
306 def verify_tcp_checksum(self):
307 # start http cli server listener on http://0.0.0.0:80
308 self.vapi.cli("http cli server")
309 p = self.params[socket.AF_INET]
311 src=self.tun_if.remote_mac, dst=self.tun_if.local_mac
312 ) / p.scapy_tun_sa.encrypt(
313 IP(src=p.remote_tun_if_host, dst=self.tun_if.local_ip4)
314 / TCP(flags="S", dport=80)
316 self.logger.debug(ppp("Sending packet:", send))
317 recv = self.send_and_expect(self.tun_if, [send], self.tun_if)
319 decrypted = p.vpp_tun_sa.decrypt(recv[IP])
320 self.assert_packet_checksums_valid(decrypted)
323 class IpsecTcpTests(IpsecTcp):
324 def test_tcp_checksum(self):
325 """verify checksum correctness for vpp generated packets"""
326 self.verify_tcp_checksum()
329 class IpsecTra4(object):
330 """verify methods for Transport v4"""
332 def get_replay_counts(self, p):
333 replay_node_name = "/err/%s/replay" % self.tra4_decrypt_node_name[0]
334 count = self.statistics.get_err_counter(replay_node_name)
337 replay_post_node_name = (
338 "/err/%s/replay" % self.tra4_decrypt_node_name[p.async_mode]
340 count += self.statistics.get_err_counter(replay_post_node_name)
344 def get_hash_failed_counts(self, p):
345 if ESP == self.encryption_type and p.crypt_algo == "AES-GCM":
346 hash_failed_node_name = (
347 "/err/%s/decryption_failed" % self.tra4_decrypt_node_name[p.async_mode]
350 hash_failed_node_name = (
351 "/err/%s/integ_error" % self.tra4_decrypt_node_name[p.async_mode]
353 count = self.statistics.get_err_counter(hash_failed_node_name)
356 count += self.statistics.get_err_counter("/err/crypto-dispatch/bad-hmac")
360 def verify_hi_seq_num(self):
361 p = self.params[socket.AF_INET]
362 saf = VppEnum.vl_api_ipsec_sad_flags_t
363 esn_on = p.vpp_tra_sa.esn_en
364 ar_on = p.flags & saf.IPSEC_API_SAD_FLAG_USE_ANTI_REPLAY
366 seq_cycle_node_name = "/err/%s/seq_cycled" % self.tra4_encrypt_node_name
367 replay_count = self.get_replay_counts(p)
368 hash_failed_count = self.get_hash_failed_counts(p)
369 seq_cycle_count = self.statistics.get_err_counter(seq_cycle_node_name)
371 # a few packets so we get the rx seq number above the window size and
372 # thus can simulate a wrap with an out of window packet
375 Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
376 / p.scapy_tra_sa.encrypt(
377 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
381 for seq in range(63, 80)
383 recv_pkts = self.send_and_expect(self.tra_if, pkts, self.tra_if)
385 # these 4 packets will all choose seq-num 0 to decrpyt since none
386 # are out of window when first checked. however, once #200 has
387 # decrypted it will move the window to 200 and has #81 is out of
388 # window. this packet should be dropped.
391 Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
392 / p.scapy_tra_sa.encrypt(
393 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
398 Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
399 / p.scapy_tra_sa.encrypt(
400 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
405 Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
406 / p.scapy_tra_sa.encrypt(
407 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
412 Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
413 / p.scapy_tra_sa.encrypt(
414 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
420 # if anti-replay is off then we won't drop #81
421 n_rx = 3 if ar_on else 4
422 self.send_and_expect(self.tra_if, pkts, self.tra_if, n_rx=n_rx)
423 # this packet is one before the wrap
426 Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
427 / p.scapy_tra_sa.encrypt(
428 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
433 recv_pkts = self.send_and_expect(self.tra_if, pkts, self.tra_if)
435 # a replayed packet, then an out of window, then a legit
436 # tests that a early failure on the batch doesn't affect subsequent packets.
439 Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
440 / p.scapy_tra_sa.encrypt(
441 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
446 Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
447 / p.scapy_tra_sa.encrypt(
448 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
453 Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
454 / p.scapy_tra_sa.encrypt(
455 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
460 n_rx = 1 if ar_on else 3
461 recv_pkts = self.send_and_expect(self.tra_if, pkts, self.tra_if, n_rx=n_rx)
463 # move the window over half way to a wrap
466 Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
467 / p.scapy_tra_sa.encrypt(
468 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
473 recv_pkts = self.send_and_expect(self.tra_if, pkts, self.tra_if)
475 # anti-replay will drop old packets, no anti-replay will not
478 Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
479 / p.scapy_tra_sa.encrypt(
480 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
487 self.send_and_assert_no_replies(self.tra_if, pkts)
489 recv_pkts = self.send_and_expect(self.tra_if, pkts, self.tra_if)
493 # validate wrapping the ESN
496 # wrap scapy's TX SA SN
497 p.scapy_tra_sa.seq_num = 0x100000005
499 # send a packet that wraps the window for both AR and no AR
502 Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
503 / p.scapy_tra_sa.encrypt(
504 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4)
511 rxs = self.send_and_expect(self.tra_if, pkts, self.tra_if)
513 decrypted = p.vpp_tra_sa.decrypt(rx[0][IP])
515 # move the window forward to half way to the next wrap
518 Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
519 / p.scapy_tra_sa.encrypt(
520 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4)
527 rxs = self.send_and_expect(self.tra_if, pkts, self.tra_if)
529 # a packet less than 2^30 from the current position is:
530 # - AR: out of window and dropped
534 Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
535 / p.scapy_tra_sa.encrypt(
536 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4)
544 self.send_and_assert_no_replies(self.tra_if, pkts)
546 self.send_and_expect(self.tra_if, pkts, self.tra_if)
548 # a packet more than 2^30 from the current position is:
549 # - AR: out of window and dropped
550 # - non-AR: considered a wrap, but since it's not a wrap
551 # it won't decrpyt and so will be dropped
554 Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
555 / p.scapy_tra_sa.encrypt(
556 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4)
563 self.send_and_assert_no_replies(self.tra_if, pkts)
565 # a packet less than 2^30 from the current position and is a
566 # wrap; (the seq is currently at 0x180000005).
567 # - AR: out of window so considered a wrap, so accepted
568 # - non-AR: not considered a wrap, so won't decrypt
569 p.scapy_tra_sa.seq_num = 0x260000005
572 Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
573 / p.scapy_tra_sa.encrypt(
574 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4)
581 self.send_and_expect(self.tra_if, pkts, self.tra_if)
583 self.send_and_assert_no_replies(self.tra_if, pkts)
586 # window positions are different now for AR/non-AR
587 # move non-AR forward
590 # a packet more than 2^30 from the current position and is a
591 # wrap; (the seq is currently at 0x180000005).
593 # - non-AR: not considered a wrap, so won't decrypt
597 Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
598 / p.scapy_tra_sa.encrypt(
599 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4)
605 Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
606 / p.scapy_tra_sa.encrypt(
607 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4)
613 self.send_and_expect(self.tra_if, pkts, self.tra_if)
617 Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
618 / p.scapy_tra_sa.encrypt(
619 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4)
625 self.send_and_expect(self.tra_if, pkts, self.tra_if)
627 def verify_tra_anti_replay(self):
628 p = self.params[socket.AF_INET]
629 esn_en = p.vpp_tra_sa.esn_en
631 seq_cycle_node_name = "/err/%s/seq_cycled" % self.tra4_encrypt_node_name
632 replay_count = self.get_replay_counts(p)
633 hash_failed_count = self.get_hash_failed_counts(p)
634 seq_cycle_count = self.statistics.get_err_counter(seq_cycle_node_name)
635 hash_err = "integ_error"
637 if ESP == self.encryption_type:
638 undersize_node_name = "/err/%s/runt" % self.tra4_decrypt_node_name[0]
639 undersize_count = self.statistics.get_err_counter(undersize_node_name)
640 # For AES-GCM an error in the hash is reported as a decryption failure
641 if p.crypt_algo == "AES-GCM":
642 hash_err = "decryption_failed"
643 # In async mode, we don't report errors in the hash.
648 # send packets with seq numbers 1->34
649 # this means the window size is still in Case B (see RFC4303
652 # for reasons i haven't investigated Scapy won't create a packet with
657 Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
658 / p.scapy_tra_sa.encrypt(
659 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
663 for seq in range(1, 34)
665 recv_pkts = self.send_and_expect(self.tra_if, pkts, self.tra_if)
667 # replayed packets are dropped
668 self.send_and_assert_no_replies(self.tra_if, pkts, timeout=0.2)
669 replay_count += len(pkts)
670 self.assertEqual(self.get_replay_counts(p), replay_count)
671 err = p.tra_sa_in.get_err("replay")
672 self.assertEqual(err, replay_count)
675 # now send a batch of packets all with the same sequence number
676 # the first packet in the batch is legitimate, the rest bogus
678 self.vapi.cli("clear error")
679 self.vapi.cli("clear node counters")
681 src=self.tra_if.remote_mac, dst=self.tra_if.local_mac
682 ) / p.scapy_tra_sa.encrypt(
683 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
686 recv_pkts = self.send_and_expect(self.tra_if, pkts * 8, self.tra_if, n_rx=1)
688 self.assertEqual(self.get_replay_counts(p), replay_count)
689 err = p.tra_sa_in.get_err("replay")
690 self.assertEqual(err, replay_count)
693 # now move the window over to 257 (more than one byte) and into Case A
695 self.vapi.cli("clear error")
697 src=self.tra_if.remote_mac, dst=self.tra_if.local_mac
698 ) / p.scapy_tra_sa.encrypt(
699 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
702 recv_pkts = self.send_and_expect(self.tra_if, [pkt], self.tra_if)
704 # replayed packets are dropped
705 self.send_and_assert_no_replies(self.tra_if, pkt * 3, timeout=0.2)
707 self.assertEqual(self.get_replay_counts(p), replay_count)
708 err = p.tra_sa_in.get_err("replay")
709 self.assertEqual(err, replay_count)
711 # the window size is 64 packets
712 # in window are still accepted
714 src=self.tra_if.remote_mac, dst=self.tra_if.local_mac
715 ) / p.scapy_tra_sa.encrypt(
716 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
719 recv_pkts = self.send_and_expect(self.tra_if, [pkt], self.tra_if)
721 # a packet that does not decrypt does not move the window forward
722 bogus_sa = SecurityAssociation(
723 self.encryption_type,
725 crypt_algo=p.crypt_algo,
726 crypt_key=mk_scapy_crypt_key(p)[::-1],
727 auth_algo=p.auth_algo,
728 auth_key=p.auth_key[::-1],
731 src=self.tra_if.remote_mac, dst=self.tra_if.local_mac
732 ) / bogus_sa.encrypt(
733 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
736 self.send_and_assert_no_replies(self.tra_if, pkt * 17, timeout=0.2)
738 hash_failed_count += 17
739 self.assertEqual(self.get_hash_failed_counts(p), hash_failed_count)
741 err = p.tra_sa_in.get_err(hash_err)
742 self.assertEqual(err, hash_failed_count)
744 # a malformed 'runt' packet
745 # created by a mis-constructed SA
746 if ESP == self.encryption_type and p.crypt_algo != "NULL":
747 bogus_sa = SecurityAssociation(self.encryption_type, p.scapy_tra_spi)
749 src=self.tra_if.remote_mac, dst=self.tra_if.local_mac
750 ) / bogus_sa.encrypt(
751 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
754 self.send_and_assert_no_replies(self.tra_if, pkt * 17, timeout=0.2)
756 undersize_count += 17
757 self.assert_error_counter_equal(undersize_node_name, undersize_count)
758 err = p.tra_sa_in.get_err("runt")
759 self.assertEqual(err, undersize_count)
761 # which we can determine since this packet is still in the window
763 src=self.tra_if.remote_mac, dst=self.tra_if.local_mac
764 ) / p.scapy_tra_sa.encrypt(
765 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
768 self.send_and_expect(self.tra_if, [pkt], self.tra_if)
771 # out of window are dropped
772 # this is Case B. So VPP will consider this to be a high seq num wrap
773 # and so the decrypt attempt will fail
776 src=self.tra_if.remote_mac, dst=self.tra_if.local_mac
777 ) / p.scapy_tra_sa.encrypt(
778 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
781 self.send_and_assert_no_replies(self.tra_if, pkt * 17, timeout=0.2)
784 # an out of window error with ESN looks like a high sequence
785 # wrap. but since it isn't then the verify will fail.
786 hash_failed_count += 17
787 self.assertEqual(self.get_hash_failed_counts(p), hash_failed_count)
789 err = p.tra_sa_in.get_err(hash_err)
790 self.assertEqual(err, hash_failed_count)
794 self.assertEqual(self.get_replay_counts(p), replay_count)
795 err = p.tra_sa_in.get_err("replay")
796 self.assertEqual(err, replay_count)
798 # valid packet moves the window over to 258
800 src=self.tra_if.remote_mac, dst=self.tra_if.local_mac
801 ) / p.scapy_tra_sa.encrypt(
802 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
805 rx = self.send_and_expect(self.tra_if, [pkt], self.tra_if)
806 decrypted = p.vpp_tra_sa.decrypt(rx[0][IP])
809 # move VPP's SA TX seq-num to just before the seq-number wrap.
810 # then fire in a packet that VPP should drop on TX because it
811 # causes the TX seq number to wrap; unless we're using extened sequence
814 self.vapi.cli("test ipsec sa %d seq 0xffffffff" % p.vpp_tra_sa_id)
815 self.logger.info(self.vapi.ppcli("show ipsec sa 0"))
816 self.logger.info(self.vapi.ppcli("show ipsec sa 1"))
820 Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
821 / p.scapy_tra_sa.encrypt(
822 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
826 for seq in range(259, 280)
830 rxs = self.send_and_expect(self.tra_if, pkts, self.tra_if)
833 # in order for scapy to decrypt its SA's high order number needs
836 p.vpp_tra_sa.seq_num = 0x100000000
838 decrypted = p.vpp_tra_sa.decrypt(rx[0][IP])
841 # wrap scapy's TX high sequence number. VPP is in case B, so it
842 # will consider this a high seq wrap also.
843 # The low seq num we set it to will place VPP's RX window in Case A
845 p.scapy_tra_sa.seq_num = 0x100000005
847 src=self.tra_if.remote_mac, dst=self.tra_if.local_mac
848 ) / p.scapy_tra_sa.encrypt(
849 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
852 rx = self.send_and_expect(self.tra_if, [pkt], self.tra_if)
854 decrypted = p.vpp_tra_sa.decrypt(rx[0][IP])
857 # A packet that has seq num between (2^32-64) and 5 is within
860 p.scapy_tra_sa.seq_num = 0xFFFFFFFD
862 src=self.tra_if.remote_mac, dst=self.tra_if.local_mac
863 ) / p.scapy_tra_sa.encrypt(
864 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
867 rx = self.send_and_expect(self.tra_if, [pkt], self.tra_if)
868 decrypted = p.vpp_tra_sa.decrypt(rx[0][IP])
871 # While in case A we cannot wrap the high sequence number again
872 # because VPP will consider this packet to be one that moves the
876 src=self.tra_if.remote_mac, dst=self.tra_if.local_mac
877 ) / p.scapy_tra_sa.encrypt(
878 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
881 self.send_and_assert_no_replies(
882 self.tra_if, [pkt], self.tra_if, timeout=0.2
885 hash_failed_count += 1
886 self.assertEqual(self.get_hash_failed_counts(p), hash_failed_count)
888 err = p.tra_sa_in.get_err(hash_err)
889 self.assertEqual(err, hash_failed_count)
892 # but if we move the window forward to case B, then we can wrap
895 p.scapy_tra_sa.seq_num = 0x100000555
897 src=self.tra_if.remote_mac, dst=self.tra_if.local_mac
898 ) / p.scapy_tra_sa.encrypt(
899 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
902 rx = self.send_and_expect(self.tra_if, [pkt], self.tra_if)
903 decrypted = p.vpp_tra_sa.decrypt(rx[0][IP])
905 p.scapy_tra_sa.seq_num = 0x200000444
907 src=self.tra_if.remote_mac, dst=self.tra_if.local_mac
908 ) / p.scapy_tra_sa.encrypt(
909 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
912 rx = self.send_and_expect(self.tra_if, [pkt], self.tra_if)
913 decrypted = p.vpp_tra_sa.decrypt(rx[0][IP])
917 # without ESN TX sequence numbers can't wrap and packets are
918 # dropped from here on out.
920 self.send_and_assert_no_replies(self.tra_if, pkts, timeout=0.2)
921 seq_cycle_count += len(pkts)
922 self.assert_error_counter_equal(seq_cycle_node_name, seq_cycle_count)
923 err = p.tra_sa_out.get_err("seq_cycled")
924 self.assertEqual(err, seq_cycle_count)
926 # move the security-associations seq number on to the last we used
927 self.vapi.cli("test ipsec sa %d seq 0x15f" % p.scapy_tra_sa_id)
928 p.scapy_tra_sa.seq_num = 351
929 p.vpp_tra_sa.seq_num = 351
931 def verify_tra_lost(self):
932 p = self.params[socket.AF_INET]
933 esn_en = p.vpp_tra_sa.esn_en
936 # send packets with seq numbers 1->34
937 # this means the window size is still in Case B (see RFC4303
940 # for reasons i haven't investigated Scapy won't create a packet with
945 Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
946 / p.scapy_tra_sa.encrypt(
947 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
951 for seq in range(1, 3)
953 self.send_and_expect(self.tra_if, pkts, self.tra_if)
955 self.assertEqual(p.tra_sa_in.get_err("lost"), 0)
957 # skip a sequence number
960 Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
961 / p.scapy_tra_sa.encrypt(
962 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
966 for seq in range(4, 6)
968 self.send_and_expect(self.tra_if, pkts, self.tra_if)
970 self.assertEqual(p.tra_sa_in.get_err("lost"), 0)
972 # the lost packet are counted untill we get up past the first
973 # sizeof(replay_window) packets
976 Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
977 / p.scapy_tra_sa.encrypt(
978 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
982 for seq in range(6, 100)
984 self.send_and_expect(self.tra_if, pkts, self.tra_if)
986 self.assertEqual(p.tra_sa_in.get_err("lost"), 1)
988 # lost of holes in the sequence
991 Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
992 / p.scapy_tra_sa.encrypt(
993 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
997 for seq in range(100, 200, 2)
999 self.send_and_expect(self.tra_if, pkts, self.tra_if, n_rx=50)
1003 Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
1004 / p.scapy_tra_sa.encrypt(
1005 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
1009 for seq in range(200, 300)
1011 self.send_and_expect(self.tra_if, pkts, self.tra_if)
1013 self.assertEqual(p.tra_sa_in.get_err("lost"), 51)
1015 # a big hole in the seq number space
1018 Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
1019 / p.scapy_tra_sa.encrypt(
1020 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
1024 for seq in range(400, 500)
1026 self.send_and_expect(self.tra_if, pkts, self.tra_if)
1028 self.assertEqual(p.tra_sa_in.get_err("lost"), 151)
1030 def verify_tra_basic4(self, count=1, payload_size=54):
1031 """ipsec v4 transport basic test"""
1032 self.vapi.cli("clear errors")
1033 self.vapi.cli("clear ipsec sa")
1035 p = self.params[socket.AF_INET]
1036 send_pkts = self.gen_encrypt_pkts(
1040 src=self.tra_if.remote_ip4,
1041 dst=self.tra_if.local_ip4,
1043 payload_size=payload_size,
1045 recv_pkts = self.send_and_expect(self.tra_if, send_pkts, self.tra_if)
1046 for rx in recv_pkts:
1047 self.assertEqual(len(rx) - len(Ether()), rx[IP].len)
1048 self.assert_packet_checksums_valid(rx)
1050 decrypted = p.vpp_tra_sa.decrypt(rx[IP])
1051 self.assert_packet_checksums_valid(decrypted)
1053 self.logger.debug(ppp("Unexpected packet:", rx))
1056 self.logger.info(self.vapi.ppcli("show error"))
1057 self.logger.info(self.vapi.ppcli("show ipsec all"))
1059 pkts = p.tra_sa_in.get_stats()["packets"]
1061 pkts, count, "incorrect SA in counts: expected %d != %d" % (count, pkts)
1063 pkts = p.tra_sa_out.get_stats()["packets"]
1065 pkts, count, "incorrect SA out counts: expected %d != %d" % (count, pkts)
1067 self.assertEqual(p.tra_sa_out.get_err("lost"), 0)
1068 self.assertEqual(p.tra_sa_in.get_err("lost"), 0)
1070 self.assert_packet_counter_equal(self.tra4_encrypt_node_name, count)
1071 self.assert_packet_counter_equal(self.tra4_decrypt_node_name[0], count)
1074 class IpsecTra4Tests(IpsecTra4):
1075 """UT test methods for Transport v4"""
1077 def test_tra_anti_replay(self):
1078 """ipsec v4 transport anti-replay test"""
1079 self.verify_tra_anti_replay()
1081 def test_tra_lost(self):
1082 """ipsec v4 transport lost packet test"""
1083 self.verify_tra_lost()
1085 def test_tra_basic(self, count=1):
1086 """ipsec v4 transport basic test"""
1087 self.verify_tra_basic4(count=1)
1089 def test_tra_burst(self):
1090 """ipsec v4 transport burst test"""
1091 self.verify_tra_basic4(count=257)
1094 class IpsecTra6(object):
1095 """verify methods for Transport v6"""
1097 def verify_tra_basic6(self, count=1, payload_size=54):
1098 self.vapi.cli("clear errors")
1099 self.vapi.cli("clear ipsec sa")
1101 p = self.params[socket.AF_INET6]
1102 send_pkts = self.gen_encrypt_pkts6(
1106 src=self.tra_if.remote_ip6,
1107 dst=self.tra_if.local_ip6,
1109 payload_size=payload_size,
1111 recv_pkts = self.send_and_expect(self.tra_if, send_pkts, self.tra_if)
1112 for rx in recv_pkts:
1113 self.assertEqual(len(rx) - len(Ether()) - len(IPv6()), rx[IPv6].plen)
1115 decrypted = p.vpp_tra_sa.decrypt(rx[IPv6])
1116 self.assert_packet_checksums_valid(decrypted)
1118 self.logger.debug(ppp("Unexpected packet:", rx))
1121 self.logger.info(self.vapi.ppcli("show error"))
1122 self.logger.info(self.vapi.ppcli("show ipsec all"))
1124 pkts = p.tra_sa_in.get_stats()["packets"]
1126 pkts, count, "incorrect SA in counts: expected %d != %d" % (count, pkts)
1128 pkts = p.tra_sa_out.get_stats()["packets"]
1130 pkts, count, "incorrect SA out counts: expected %d != %d" % (count, pkts)
1132 self.assert_packet_counter_equal(self.tra6_encrypt_node_name, count)
1133 self.assert_packet_counter_equal(self.tra6_decrypt_node_name[0], count)
1135 def gen_encrypt_pkts_ext_hdrs6(
1136 self, sa, sw_intf, src, dst, count=1, payload_size=54
1139 Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac)
1141 IPv6(src=src, dst=dst)
1142 / ICMPv6EchoRequest(id=0, seq=1, data="X" * payload_size)
1144 for i in range(count)
1147 def gen_pkts_ext_hdrs6(self, sw_intf, src, dst, count=1, payload_size=54):
1149 Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac)
1150 / IPv6(src=src, dst=dst)
1151 / IPv6ExtHdrHopByHop()
1152 / IPv6ExtHdrFragment(id=2, offset=200)
1153 / Raw(b"\xff" * 200)
1154 for i in range(count)
1157 def verify_tra_encrypted6(self, p, sa, rxs):
1160 self.assert_packet_checksums_valid(rx)
1162 decrypt_pkt = p.vpp_tra_sa.decrypt(rx[IPv6])
1163 decrypted.append(decrypt_pkt)
1164 self.assert_equal(decrypt_pkt.src, self.tra_if.local_ip6)
1165 self.assert_equal(decrypt_pkt.dst, self.tra_if.remote_ip6)
1167 self.logger.debug(ppp("Unexpected packet:", rx))
1169 self.logger.debug(ppp("Decrypted packet:", decrypt_pkt))
1175 def verify_tra_66_ext_hdrs(self, p):
1179 # check we can decrypt with options
1181 tx = self.gen_encrypt_pkts_ext_hdrs6(
1184 src=self.tra_if.remote_ip6,
1185 dst=self.tra_if.local_ip6,
1188 self.send_and_expect(self.tra_if, tx, self.tra_if)
1191 # injecting a packet from ourselves to be routed of box is a hack
1192 # but it matches an outbout policy, alors je ne regrette rien
1195 # one extension before ESP
1197 Ether(src=self.pg2.remote_mac, dst=self.pg2.local_mac)
1198 / IPv6(src=self.tra_if.local_ip6, dst=self.tra_if.remote_ip6)
1199 / IPv6ExtHdrFragment(id=2, offset=200)
1200 / Raw(b"\xff" * 200)
1203 rxs = self.send_and_expect(self.pg2, [tx], self.tra_if)
1204 dcs = self.verify_tra_encrypted6(p, p.vpp_tra_sa, rxs)
1207 # for reasons i'm not going to investigate scapy does not
1208 # created the correct headers after decrypt. but reparsing
1209 # the ipv6 packet fixes it
1210 dc = IPv6(raw(dc[IPv6]))
1211 self.assert_equal(dc[IPv6ExtHdrFragment].id, 2)
1213 # two extensions before ESP
1215 Ether(src=self.pg2.remote_mac, dst=self.pg2.local_mac)
1216 / IPv6(src=self.tra_if.local_ip6, dst=self.tra_if.remote_ip6)
1217 / IPv6ExtHdrHopByHop()
1218 / IPv6ExtHdrFragment(id=2, offset=200)
1219 / Raw(b"\xff" * 200)
1222 rxs = self.send_and_expect(self.pg2, [tx], self.tra_if)
1223 dcs = self.verify_tra_encrypted6(p, p.vpp_tra_sa, rxs)
1226 dc = IPv6(raw(dc[IPv6]))
1227 self.assertTrue(dc[IPv6ExtHdrHopByHop])
1228 self.assert_equal(dc[IPv6ExtHdrFragment].id, 2)
1230 # two extensions before ESP, one after
1232 Ether(src=self.pg2.remote_mac, dst=self.pg2.local_mac)
1233 / IPv6(src=self.tra_if.local_ip6, dst=self.tra_if.remote_ip6)
1234 / IPv6ExtHdrHopByHop()
1235 / IPv6ExtHdrFragment(id=2, offset=200)
1236 / IPv6ExtHdrDestOpt()
1237 / Raw(b"\xff" * 200)
1240 rxs = self.send_and_expect(self.pg2, [tx], self.tra_if)
1241 dcs = self.verify_tra_encrypted6(p, p.vpp_tra_sa, rxs)
1244 dc = IPv6(raw(dc[IPv6]))
1245 self.assertTrue(dc[IPv6ExtHdrDestOpt])
1246 self.assertTrue(dc[IPv6ExtHdrHopByHop])
1247 self.assert_equal(dc[IPv6ExtHdrFragment].id, 2)
1250 class IpsecTra6Tests(IpsecTra6):
1251 """UT test methods for Transport v6"""
1253 def test_tra_basic6(self):
1254 """ipsec v6 transport basic test"""
1255 self.verify_tra_basic6(count=1)
1257 def test_tra_burst6(self):
1258 """ipsec v6 transport burst test"""
1259 self.verify_tra_basic6(count=257)
1262 class IpsecTra6ExtTests(IpsecTra6):
1263 def test_tra_ext_hdrs_66(self):
1264 """ipsec 6o6 tra extension headers test"""
1265 self.verify_tra_66_ext_hdrs(self.params[socket.AF_INET6])
1268 class IpsecTra46Tests(IpsecTra4Tests, IpsecTra6Tests):
1269 """UT test methods for Transport v6 and v4"""
1274 class IpsecTun4(object):
1275 """verify methods for Tunnel v4"""
1277 def verify_counters4(self, p, count, n_frags=None, worker=None):
1280 if hasattr(p, "spd_policy_in_any"):
1281 pkts = p.spd_policy_in_any.get_stats(worker)["packets"]
1285 "incorrect SPD any policy: expected %d != %d" % (count, pkts),
1288 if hasattr(p, "tun_sa_in"):
1289 pkts = p.tun_sa_in.get_stats(worker)["packets"]
1291 pkts, count, "incorrect SA in counts: expected %d != %d" % (count, pkts)
1293 pkts = p.tun_sa_out.get_stats(worker)["packets"]
1297 "incorrect SA out counts: expected %d != %d" % (count, pkts),
1300 self.assert_packet_counter_equal(self.tun4_encrypt_node_name, n_frags)
1301 self.assert_packet_counter_equal(self.tun4_decrypt_node_name[0], count)
1303 def verify_decrypted(self, p, rxs):
1305 self.assert_equal(rx[IP].src, p.remote_tun_if_host)
1306 self.assert_equal(rx[IP].dst, self.pg1.remote_ip4)
1307 self.assert_packet_checksums_valid(rx)
1309 def verify_esp_padding(self, sa, esp_payload, decrypt_pkt):
1310 align = sa.crypt_algo.block_size
1313 exp_len = (len(decrypt_pkt) + 2 + (align - 1)) & ~(align - 1)
1314 exp_len += sa.crypt_algo.iv_size
1315 exp_len += sa.crypt_algo.icv_size or sa.auth_algo.icv_size
1316 self.assertEqual(exp_len, len(esp_payload))
1318 def verify_encrypted(self, p, sa, rxs):
1322 self.assertEqual(rx[UDP].dport, p.nat_header.dport)
1323 self.assert_packet_checksums_valid(rx)
1324 self.assertEqual(len(rx) - len(Ether()), rx[IP].len)
1327 decrypt_pkt = p.vpp_tun_sa.decrypt(rx_ip)
1328 if not decrypt_pkt.haslayer(IP):
1329 decrypt_pkt = IP(decrypt_pkt[Raw].load)
1330 if rx_ip.proto == socket.IPPROTO_ESP:
1331 self.verify_esp_padding(sa, rx_ip[ESP].data, decrypt_pkt)
1332 decrypt_pkts.append(decrypt_pkt)
1333 self.assert_equal(decrypt_pkt.src, self.pg1.remote_ip4)
1334 self.assert_equal(decrypt_pkt.dst, p.remote_tun_if_host)
1336 self.logger.debug(ppp("Unexpected packet:", rx))
1338 self.logger.debug(ppp("Decrypted packet:", decrypt_pkt))
1342 pkts = reassemble4(decrypt_pkts)
1344 self.assert_packet_checksums_valid(pkt)
1346 def verify_tun_44(self, p, count=1, payload_size=64, n_rx=None):
1347 self.vapi.cli("clear errors")
1348 self.vapi.cli("clear ipsec counters")
1349 self.vapi.cli("clear ipsec sa")
1353 send_pkts = self.gen_encrypt_pkts(
1357 src=p.remote_tun_if_host,
1358 dst=self.pg1.remote_ip4,
1360 payload_size=payload_size,
1362 recv_pkts = self.send_and_expect(self.tun_if, send_pkts, self.pg1)
1363 self.verify_decrypted(p, recv_pkts)
1365 send_pkts = self.gen_pkts(
1367 src=self.pg1.remote_ip4,
1368 dst=p.remote_tun_if_host,
1370 payload_size=payload_size,
1372 recv_pkts = self.send_and_expect(self.pg1, send_pkts, self.tun_if, n_rx)
1373 self.verify_encrypted(p, p.vpp_tun_sa, recv_pkts)
1375 for rx in recv_pkts:
1376 self.assertEqual(rx[IP].src, p.tun_src)
1377 self.assertEqual(rx[IP].dst, p.tun_dst)
1380 self.logger.info(self.vapi.ppcli("show error"))
1381 self.logger.info(self.vapi.ppcli("show ipsec all"))
1383 self.logger.info(self.vapi.ppcli("show ipsec sa 0"))
1384 self.logger.info(self.vapi.ppcli("show ipsec sa 4"))
1385 self.verify_counters4(p, count, n_rx)
1387 def verify_tun_dropped_44(self, p, count=1, payload_size=64, n_rx=None):
1388 self.vapi.cli("clear errors")
1392 send_pkts = self.gen_encrypt_pkts(
1396 src=p.remote_tun_if_host,
1397 dst=self.pg1.remote_ip4,
1400 self.send_and_assert_no_replies(self.tun_if, send_pkts)
1402 send_pkts = self.gen_pkts(
1404 src=self.pg1.remote_ip4,
1405 dst=p.remote_tun_if_host,
1407 payload_size=payload_size,
1409 self.send_and_assert_no_replies(self.pg1, send_pkts)
1412 self.logger.info(self.vapi.ppcli("show error"))
1413 self.logger.info(self.vapi.ppcli("show ipsec all"))
1415 def verify_tun_reass_44(self, p):
1416 self.vapi.cli("clear errors")
1417 self.vapi.ip_reassembly_enable_disable(
1418 sw_if_index=self.tun_if.sw_if_index, enable_ip4=True
1422 send_pkts = self.gen_encrypt_pkts(
1426 src=p.remote_tun_if_host,
1427 dst=self.pg1.remote_ip4,
1431 send_pkts = fragment_rfc791(send_pkts[0], 1400)
1432 recv_pkts = self.send_and_expect(self.tun_if, send_pkts, self.pg1, n_rx=1)
1433 self.verify_decrypted(p, recv_pkts)
1435 send_pkts = self.gen_pkts(
1436 self.pg1, src=self.pg1.remote_ip4, dst=p.remote_tun_if_host, count=1
1438 recv_pkts = self.send_and_expect(self.pg1, send_pkts, self.tun_if)
1439 self.verify_encrypted(p, p.vpp_tun_sa, recv_pkts)
1442 self.logger.info(self.vapi.ppcli("show error"))
1443 self.logger.info(self.vapi.ppcli("show ipsec all"))
1445 self.verify_counters4(p, 1, 1)
1446 self.vapi.ip_reassembly_enable_disable(
1447 sw_if_index=self.tun_if.sw_if_index, enable_ip4=False
1450 def verify_tun_64(self, p, count=1):
1451 self.vapi.cli("clear errors")
1452 self.vapi.cli("clear ipsec sa")
1454 send_pkts = self.gen_encrypt_pkts6(
1458 src=p.remote_tun_if_host6,
1459 dst=self.pg1.remote_ip6,
1462 recv_pkts = self.send_and_expect(self.tun_if, send_pkts, self.pg1)
1463 for recv_pkt in recv_pkts:
1464 self.assert_equal(recv_pkt[IPv6].src, p.remote_tun_if_host6)
1465 self.assert_equal(recv_pkt[IPv6].dst, self.pg1.remote_ip6)
1466 self.assert_packet_checksums_valid(recv_pkt)
1467 send_pkts = self.gen_pkts6(
1470 src=self.pg1.remote_ip6,
1471 dst=p.remote_tun_if_host6,
1474 recv_pkts = self.send_and_expect(self.pg1, send_pkts, self.tun_if)
1475 for recv_pkt in recv_pkts:
1477 decrypt_pkt = p.vpp_tun_sa.decrypt(recv_pkt[IP])
1478 if not decrypt_pkt.haslayer(IPv6):
1479 decrypt_pkt = IPv6(decrypt_pkt[Raw].load)
1480 self.assert_equal(decrypt_pkt.src, self.pg1.remote_ip6)
1481 self.assert_equal(decrypt_pkt.dst, p.remote_tun_if_host6)
1482 self.assert_packet_checksums_valid(decrypt_pkt)
1484 self.logger.error(ppp("Unexpected packet:", recv_pkt))
1486 self.logger.debug(ppp("Decrypted packet:", decrypt_pkt))
1491 self.logger.info(self.vapi.ppcli("show error"))
1492 self.logger.info(self.vapi.ppcli("show ipsec all"))
1494 self.verify_counters4(p, count)
1496 def verify_keepalive(self, p):
1497 # the sizeof Raw is calculated to pad to the minimum ehternet
1498 # frame size of 64 btyes
1500 Ether(src=self.tun_if.remote_mac, dst=self.tun_if.local_mac)
1501 / IP(src=p.remote_tun_if_host, dst=self.tun_if.local_ip4)
1502 / UDP(sport=333, dport=4500)
1506 self.send_and_assert_no_replies(self.tun_if, pkt * 31)
1507 self.assert_error_counter_equal(
1508 "/err/%s/nat_keepalive" % self.tun4_input_node, 31
1512 Ether(src=self.tun_if.remote_mac, dst=self.tun_if.local_mac)
1513 / IP(src=p.remote_tun_if_host, dst=self.tun_if.local_ip4)
1514 / UDP(sport=333, dport=4500)
1517 self.send_and_assert_no_replies(self.tun_if, pkt * 31)
1518 self.assert_error_counter_equal("/err/%s/too_short" % self.tun4_input_node, 31)
1521 Ether(src=self.tun_if.remote_mac, dst=self.tun_if.local_mac)
1522 / IP(src=p.remote_tun_if_host, dst=self.tun_if.local_ip4)
1523 / UDP(sport=333, dport=4500)
1527 self.send_and_assert_no_replies(self.tun_if, pkt * 31)
1528 self.assert_error_counter_equal("/err/%s/too_short" % self.tun4_input_node, 62)
1531 class IpsecTun4Tests(IpsecTun4):
1532 """UT test methods for Tunnel v4"""
1534 def test_tun_basic44(self):
1535 """ipsec 4o4 tunnel basic test"""
1536 self.verify_tun_44(self.params[socket.AF_INET], count=1)
1537 self.tun_if.admin_down()
1538 self.tun_if.resolve_arp()
1539 self.tun_if.admin_up()
1540 self.verify_tun_44(self.params[socket.AF_INET], count=1)
1542 def test_tun_reass_basic44(self):
1543 """ipsec 4o4 tunnel basic reassembly test"""
1544 self.verify_tun_reass_44(self.params[socket.AF_INET])
1546 def test_tun_burst44(self):
1547 """ipsec 4o4 tunnel burst test"""
1548 self.verify_tun_44(self.params[socket.AF_INET], count=127)
1551 class IpsecTun6(object):
1552 """verify methods for Tunnel v6"""
1554 def verify_counters6(self, p_in, p_out, count, worker=None):
1555 if hasattr(p_in, "tun_sa_in"):
1556 pkts = p_in.tun_sa_in.get_stats(worker)["packets"]
1558 pkts, count, "incorrect SA in counts: expected %d != %d" % (count, pkts)
1560 if hasattr(p_out, "tun_sa_out"):
1561 pkts = p_out.tun_sa_out.get_stats(worker)["packets"]
1565 "incorrect SA out counts: expected %d != %d" % (count, pkts),
1567 self.assert_packet_counter_equal(self.tun6_encrypt_node_name, count)
1568 self.assert_packet_counter_equal(self.tun6_decrypt_node_name[0], count)
1570 def verify_decrypted6(self, p, rxs):
1572 self.assert_equal(rx[IPv6].src, p.remote_tun_if_host)
1573 self.assert_equal(rx[IPv6].dst, self.pg1.remote_ip6)
1574 self.assert_packet_checksums_valid(rx)
1576 def verify_encrypted6(self, p, sa, rxs):
1578 self.assert_packet_checksums_valid(rx)
1579 self.assertEqual(len(rx) - len(Ether()) - len(IPv6()), rx[IPv6].plen)
1580 self.assert_equal(rx[IPv6].hlim, p.outer_hop_limit)
1581 if p.outer_flow_label:
1582 self.assert_equal(rx[IPv6].fl, p.outer_flow_label)
1584 decrypt_pkt = p.vpp_tun_sa.decrypt(rx[IPv6])
1585 if not decrypt_pkt.haslayer(IPv6):
1586 decrypt_pkt = IPv6(decrypt_pkt[Raw].load)
1587 self.assert_packet_checksums_valid(decrypt_pkt)
1588 self.assert_equal(decrypt_pkt.src, self.pg1.remote_ip6)
1589 self.assert_equal(decrypt_pkt.dst, p.remote_tun_if_host)
1590 self.assert_equal(decrypt_pkt.hlim, p.inner_hop_limit - 1)
1591 self.assert_equal(decrypt_pkt.fl, p.inner_flow_label)
1593 self.logger.debug(ppp("Unexpected packet:", rx))
1595 self.logger.debug(ppp("Decrypted packet:", decrypt_pkt))
1600 def verify_drop_tun_tx_66(self, p_in, count=1, payload_size=64):
1601 self.vapi.cli("clear errors")
1602 self.vapi.cli("clear ipsec sa")
1604 send_pkts = self.gen_pkts6(
1607 src=self.pg1.remote_ip6,
1608 dst=p_in.remote_tun_if_host,
1610 payload_size=payload_size,
1612 self.send_and_assert_no_replies(self.tun_if, send_pkts)
1613 self.logger.info(self.vapi.cli("sh punt stats"))
1615 def verify_drop_tun_rx_66(self, p_in, count=1, payload_size=64):
1616 self.vapi.cli("clear errors")
1617 self.vapi.cli("clear ipsec sa")
1619 send_pkts = self.gen_encrypt_pkts6(
1623 src=p_in.remote_tun_if_host,
1624 dst=self.pg1.remote_ip6,
1627 self.send_and_assert_no_replies(self.tun_if, send_pkts)
1629 def verify_drop_tun_66(self, p_in, count=1, payload_size=64):
1630 self.verify_drop_tun_tx_66(p_in, count=count, payload_size=payload_size)
1631 self.verify_drop_tun_rx_66(p_in, count=count, payload_size=payload_size)
1633 def verify_tun_66(self, p_in, p_out=None, count=1, payload_size=64):
1634 self.vapi.cli("clear errors")
1635 self.vapi.cli("clear ipsec sa")
1639 send_pkts = self.gen_encrypt_pkts6(
1643 src=p_in.remote_tun_if_host,
1644 dst=self.pg1.remote_ip6,
1646 payload_size=payload_size,
1648 recv_pkts = self.send_and_expect(self.tun_if, send_pkts, self.pg1)
1649 self.verify_decrypted6(p_in, recv_pkts)
1651 send_pkts = self.gen_pkts6(
1654 src=self.pg1.remote_ip6,
1655 dst=p_out.remote_tun_if_host,
1657 payload_size=payload_size,
1659 recv_pkts = self.send_and_expect(self.pg1, send_pkts, self.tun_if)
1660 self.verify_encrypted6(p_out, p_out.vpp_tun_sa, recv_pkts)
1662 for rx in recv_pkts:
1663 self.assertEqual(rx[IPv6].src, p_out.tun_src)
1664 self.assertEqual(rx[IPv6].dst, p_out.tun_dst)
1667 self.logger.info(self.vapi.ppcli("show error"))
1668 self.logger.info(self.vapi.ppcli("show ipsec all"))
1669 self.verify_counters6(p_in, p_out, count)
1671 def verify_tun_reass_66(self, p):
1672 self.vapi.cli("clear errors")
1673 self.vapi.ip_reassembly_enable_disable(
1674 sw_if_index=self.tun_if.sw_if_index, enable_ip6=True
1678 send_pkts = self.gen_encrypt_pkts6(
1682 src=p.remote_tun_if_host,
1683 dst=self.pg1.remote_ip6,
1687 send_pkts = fragment_rfc8200(send_pkts[0], 1, 1400, self.logger)
1688 recv_pkts = self.send_and_expect(self.tun_if, send_pkts, self.pg1, n_rx=1)
1689 self.verify_decrypted6(p, recv_pkts)
1691 send_pkts = self.gen_pkts6(
1694 src=self.pg1.remote_ip6,
1695 dst=p.remote_tun_if_host,
1699 recv_pkts = self.send_and_expect(self.pg1, send_pkts, self.tun_if)
1700 self.verify_encrypted6(p, p.vpp_tun_sa, recv_pkts)
1702 self.logger.info(self.vapi.ppcli("show error"))
1703 self.logger.info(self.vapi.ppcli("show ipsec all"))
1704 self.verify_counters6(p, p, 1)
1705 self.vapi.ip_reassembly_enable_disable(
1706 sw_if_index=self.tun_if.sw_if_index, enable_ip6=False
1709 def verify_tun_46(self, p, count=1):
1710 """ipsec 4o6 tunnel basic test"""
1711 self.vapi.cli("clear errors")
1712 self.vapi.cli("clear ipsec sa")
1714 send_pkts = self.gen_encrypt_pkts(
1718 src=p.remote_tun_if_host4,
1719 dst=self.pg1.remote_ip4,
1722 recv_pkts = self.send_and_expect(self.tun_if, send_pkts, self.pg1)
1723 for recv_pkt in recv_pkts:
1724 self.assert_equal(recv_pkt[IP].src, p.remote_tun_if_host4)
1725 self.assert_equal(recv_pkt[IP].dst, self.pg1.remote_ip4)
1726 self.assert_packet_checksums_valid(recv_pkt)
1727 send_pkts = self.gen_pkts(
1729 src=self.pg1.remote_ip4,
1730 dst=p.remote_tun_if_host4,
1733 recv_pkts = self.send_and_expect(self.pg1, send_pkts, self.tun_if)
1734 for recv_pkt in recv_pkts:
1736 decrypt_pkt = p.vpp_tun_sa.decrypt(recv_pkt[IPv6])
1737 if not decrypt_pkt.haslayer(IP):
1738 decrypt_pkt = IP(decrypt_pkt[Raw].load)
1739 self.assert_equal(decrypt_pkt.src, self.pg1.remote_ip4)
1740 self.assert_equal(decrypt_pkt.dst, p.remote_tun_if_host4)
1741 self.assert_packet_checksums_valid(decrypt_pkt)
1743 self.logger.debug(ppp("Unexpected packet:", recv_pkt))
1745 self.logger.debug(ppp("Decrypted packet:", decrypt_pkt))
1750 self.logger.info(self.vapi.ppcli("show error"))
1751 self.logger.info(self.vapi.ppcli("show ipsec all"))
1752 self.verify_counters6(p, p, count)
1754 def verify_keepalive(self, p):
1755 # the sizeof Raw is calculated to pad to the minimum ehternet
1756 # frame size of 64 btyes
1758 Ether(src=self.tun_if.remote_mac, dst=self.tun_if.local_mac)
1759 / IPv6(src=p.remote_tun_if_host, dst=self.tun_if.local_ip6)
1760 / UDP(sport=333, dport=4500)
1764 self.send_and_assert_no_replies(self.tun_if, pkt * 31)
1765 self.assert_error_counter_equal(
1766 "/err/%s/nat_keepalive" % self.tun6_input_node, 31
1770 Ether(src=self.tun_if.remote_mac, dst=self.tun_if.local_mac)
1771 / IPv6(src=p.remote_tun_if_host, dst=self.tun_if.local_ip6)
1772 / UDP(sport=333, dport=4500)
1775 self.send_and_assert_no_replies(self.tun_if, pkt * 31)
1776 self.assert_error_counter_equal("/err/%s/too_short" % self.tun6_input_node, 31)
1779 Ether(src=self.tun_if.remote_mac, dst=self.tun_if.local_mac)
1780 / IPv6(src=p.remote_tun_if_host, dst=self.tun_if.local_ip6)
1781 / UDP(sport=333, dport=4500)
1785 self.send_and_assert_no_replies(self.tun_if, pkt * 31)
1786 self.assert_error_counter_equal("/err/%s/too_short" % self.tun6_input_node, 62)
1789 class IpsecTun6Tests(IpsecTun6):
1790 """UT test methods for Tunnel v6"""
1792 def test_tun_basic66(self):
1793 """ipsec 6o6 tunnel basic test"""
1794 self.verify_tun_66(self.params[socket.AF_INET6], count=1)
1796 def test_tun_reass_basic66(self):
1797 """ipsec 6o6 tunnel basic reassembly test"""
1798 self.verify_tun_reass_66(self.params[socket.AF_INET6])
1800 def test_tun_burst66(self):
1801 """ipsec 6o6 tunnel burst test"""
1802 self.verify_tun_66(self.params[socket.AF_INET6], count=257)
1805 class IpsecTun6HandoffTests(IpsecTun6):
1806 """UT test methods for Tunnel v6 with multiple workers"""
1808 vpp_worker_count = 2
1810 def test_tun_handoff_66(self):
1811 """ipsec 6o6 tunnel worker hand-off test"""
1812 self.vapi.cli("clear errors")
1813 self.vapi.cli("clear ipsec sa")
1816 p = self.params[socket.AF_INET6]
1818 # inject alternately on worker 0 and 1. all counts on the SA
1819 # should be against worker 0
1820 for worker in [0, 1, 0, 1]:
1821 send_pkts = self.gen_encrypt_pkts6(
1825 src=p.remote_tun_if_host,
1826 dst=self.pg1.remote_ip6,
1829 recv_pkts = self.send_and_expect(
1830 self.tun_if, send_pkts, self.pg1, worker=worker
1832 self.verify_decrypted6(p, recv_pkts)
1834 send_pkts = self.gen_pkts6(
1837 src=self.pg1.remote_ip6,
1838 dst=p.remote_tun_if_host,
1841 recv_pkts = self.send_and_expect(
1842 self.pg1, send_pkts, self.tun_if, worker=worker
1844 self.verify_encrypted6(p, p.vpp_tun_sa, recv_pkts)
1846 # all counts against the first worker that was used
1847 self.verify_counters6(p, p, 4 * N_PKTS, worker=0)
1850 class IpsecTun4HandoffTests(IpsecTun4):
1851 """UT test methods for Tunnel v4 with multiple workers"""
1853 vpp_worker_count = 2
1855 def test_tun_handooff_44(self):
1856 """ipsec 4o4 tunnel worker hand-off test"""
1857 self.vapi.cli("clear errors")
1858 self.vapi.cli("clear ipsec sa")
1861 p = self.params[socket.AF_INET]
1863 # inject alternately on worker 0 and 1. all counts on the SA
1864 # should be against worker 0
1865 for worker in [0, 1, 0, 1]:
1866 send_pkts = self.gen_encrypt_pkts(
1870 src=p.remote_tun_if_host,
1871 dst=self.pg1.remote_ip4,
1874 recv_pkts = self.send_and_expect(
1875 self.tun_if, send_pkts, self.pg1, worker=worker
1877 self.verify_decrypted(p, recv_pkts)
1879 send_pkts = self.gen_pkts(
1881 src=self.pg1.remote_ip4,
1882 dst=p.remote_tun_if_host,
1885 recv_pkts = self.send_and_expect(
1886 self.pg1, send_pkts, self.tun_if, worker=worker
1888 self.verify_encrypted(p, p.vpp_tun_sa, recv_pkts)
1890 # all counts against the first worker that was used
1891 self.verify_counters4(p, 4 * N_PKTS, worker=0)
1894 class IpsecTun46Tests(IpsecTun4Tests, IpsecTun6Tests):
1895 """UT test methods for Tunnel v6 & v4"""
1900 class IPSecIPv4Fwd(VppTestCase):
1901 """Test IPSec by capturing and verifying IPv4 forwarded pkts"""
1904 def setUpConstants(cls):
1905 super(IPSecIPv4Fwd, cls).setUpConstants()
1908 super(IPSecIPv4Fwd, self).setUp()
1909 # store SPD objects so we can remove configs on tear down
1911 self.spd_policies = []
1914 # remove SPD policies
1915 for obj in self.spd_policies:
1916 obj.remove_vpp_config()
1917 self.spd_policies = []
1918 # remove SPD items (interface bindings first, then SPD)
1919 for obj in reversed(self.spd_objs):
1920 obj.remove_vpp_config()
1922 # close down pg intfs
1923 for pg in self.pg_interfaces:
1926 super(IPSecIPv4Fwd, self).tearDown()
1928 def create_interfaces(self, num_ifs=2):
1929 # create interfaces pg0 ... pg<num_ifs>
1930 self.create_pg_interfaces(range(num_ifs))
1931 for pg in self.pg_interfaces:
1932 # put the interface up
1934 # configure IPv4 address on the interface
1936 # resolve ARP, so that we know VPP MAC
1938 self.logger.info(self.vapi.ppcli("show int addr"))
1940 def spd_create_and_intf_add(self, spd_id, pg_list):
1941 spd = VppIpsecSpd(self, spd_id)
1942 spd.add_vpp_config()
1943 self.spd_objs.append(spd)
1945 spdItf = VppIpsecSpdItfBinding(self, spd, pg)
1946 spdItf.add_vpp_config()
1947 self.spd_objs.append(spdItf)
1949 def get_policy(self, policy_type):
1950 e = VppEnum.vl_api_ipsec_spd_action_t
1951 if policy_type == "protect":
1952 return e.IPSEC_API_SPD_ACTION_PROTECT
1953 elif policy_type == "bypass":
1954 return e.IPSEC_API_SPD_ACTION_BYPASS
1955 elif policy_type == "discard":
1956 return e.IPSEC_API_SPD_ACTION_DISCARD
1958 raise Exception("Invalid policy type: %s", policy_type)
1960 def spd_add_rem_policy(
1972 local_ip_start=ip_address("0.0.0.0"),
1973 local_ip_stop=ip_address("255.255.255.255"),
1974 remote_ip_start=ip_address("0.0.0.0"),
1975 remote_ip_stop=ip_address("255.255.255.255"),
1976 remote_port_start=0,
1977 remote_port_stop=65535,
1979 local_port_stop=65535,
1981 spd = VppIpsecSpd(self, spd_id)
1984 src_range_low = ip_address("0.0.0.0")
1985 src_range_high = ip_address("255.255.255.255")
1986 dst_range_low = ip_address("0.0.0.0")
1987 dst_range_high = ip_address("255.255.255.255")
1990 src_range_low = local_ip_start
1991 src_range_high = local_ip_stop
1992 dst_range_low = remote_ip_start
1993 dst_range_high = remote_ip_stop
1996 src_range_low = src_if.remote_ip4
1997 src_range_high = src_if.remote_ip4
1998 dst_range_low = dst_if.remote_ip4
1999 dst_range_high = dst_if.remote_ip4
2001 spdEntry = VppIpsecSpdEntry(
2011 policy=self.get_policy(policy_type),
2013 remote_port_start=remote_port_start,
2014 remote_port_stop=remote_port_stop,
2015 local_port_start=local_port_start,
2016 local_port_stop=local_port_stop,
2020 spdEntry.add_vpp_config()
2021 self.spd_policies.append(spdEntry)
2023 spdEntry.remove_vpp_config()
2024 self.spd_policies.remove(spdEntry)
2025 self.logger.info(self.vapi.ppcli("show ipsec all"))
2028 def create_stream(self, src_if, dst_if, pkt_count, src_prt=1234, dst_prt=5678):
2030 for i in range(pkt_count):
2031 # create packet info stored in the test case instance
2032 info = self.create_packet_info(src_if, dst_if)
2033 # convert the info into packet payload
2034 payload = self.info_to_payload(info)
2035 # create the packet itself
2037 Ether(dst=src_if.local_mac, src=src_if.remote_mac)
2038 / IP(src=src_if.remote_ip4, dst=dst_if.remote_ip4)
2039 / UDP(sport=src_prt, dport=dst_prt)
2042 # store a copy of the packet in the packet info
2043 info.data = p.copy()
2044 # append the packet to the list
2046 # return the created packet list
2049 def verify_capture(self, src_if, dst_if, capture):
2051 for packet in capture:
2055 # convert the payload to packet info object
2056 payload_info = self.payload_to_info(packet)
2057 # make sure the indexes match
2059 payload_info.src, src_if.sw_if_index, "source sw_if_index"
2062 payload_info.dst, dst_if.sw_if_index, "destination sw_if_index"
2064 packet_info = self.get_next_packet_info_for_interface2(
2065 src_if.sw_if_index, dst_if.sw_if_index, packet_info
2067 # make sure we didn't run out of saved packets
2068 self.assertIsNotNone(packet_info)
2070 payload_info.index, packet_info.index, "packet info index"
2072 saved_packet = packet_info.data # fetch the saved packet
2073 # assert the values match
2074 self.assert_equal(ip.src, saved_packet[IP].src, "IP source address")
2075 # ... more assertions here
2076 self.assert_equal(udp.sport, saved_packet[UDP].sport, "UDP source port")
2077 except Exception as e:
2078 self.logger.error(ppp("Unexpected or invalid packet:", packet))
2080 remaining_packet = self.get_next_packet_info_for_interface2(
2081 src_if.sw_if_index, dst_if.sw_if_index, packet_info
2085 "Interface %s: Packet expected from interface "
2086 "%s didn't arrive" % (dst_if.name, src_if.name),
2089 def verify_policy_match(self, pkt_count, spdEntry):
2090 self.logger.info("XXXX %s %s", str(spdEntry), str(spdEntry.get_stats()))
2091 matched_pkts = spdEntry.get_stats().get("packets")
2092 self.logger.info("Policy %s matched: %d pkts", str(spdEntry), matched_pkts)
2093 self.assert_equal(pkt_count, matched_pkts)
2096 class SpdFlowCacheTemplate(IPSecIPv4Fwd):
2098 def setUpConstants(cls):
2099 super(SpdFlowCacheTemplate, cls).setUpConstants()
2100 # Override this method with required cmdline parameters e.g.
2101 # cls.vpp_cmdline.extend(["ipsec", "{",
2102 # "ipv4-outbound-spd-flow-cache on",
2104 # cls.logger.info("VPP modified cmdline is %s" % " "
2105 # .join(cls.vpp_cmdline))
2108 super(SpdFlowCacheTemplate, self).setUp()
2111 super(SpdFlowCacheTemplate, self).tearDown()
2113 def get_spd_flow_cache_entries(self, outbound):
2114 """'show ipsec spd' output:
2115 ipv4-inbound-spd-flow-cache-entries: 0
2116 ipv4-outbound-spd-flow-cache-entries: 0
2118 show_ipsec_reply = self.vapi.cli("show ipsec spd")
2119 # match the relevant section of 'show ipsec spd' output
2121 regex_match = re.search(
2122 "ipv4-outbound-spd-flow-cache-entries: (.*)",
2127 regex_match = re.search(
2128 "ipv4-inbound-spd-flow-cache-entries: (.*)", show_ipsec_reply, re.DOTALL
2130 if regex_match is None:
2132 "Unable to find spd flow cache entries \
2133 in 'show ipsec spd' CLI output - regex failed to match"
2137 num_entries = int(regex_match.group(1))
2140 "Unable to get spd flow cache entries \
2141 from 'show ipsec spd' string: %s",
2142 regex_match.group(0),
2144 self.logger.info("%s", regex_match.group(0))
2147 def verify_num_outbound_flow_cache_entries(self, expected_elements):
2149 self.get_spd_flow_cache_entries(outbound=True), expected_elements
2152 def verify_num_inbound_flow_cache_entries(self, expected_elements):
2154 self.get_spd_flow_cache_entries(outbound=False), expected_elements
2157 def crc32_supported(self):
2158 # lscpu is part of util-linux package, available on all Linux Distros
2159 stream = os.popen("lscpu")
2160 cpu_info = stream.read()
2161 # feature/flag "crc32" on Aarch64 and "sse4_2" on x86
2162 # see vppinfra/crc32.h
2163 if "crc32" or "sse4_2" in cpu_info:
2164 self.logger.info("\ncrc32 supported:\n" + cpu_info)
2167 self.logger.info("\ncrc32 NOT supported:\n" + cpu_info)
2171 class IPSecIPv6Fwd(VppTestCase):
2172 """Test IPSec by capturing and verifying IPv6 forwarded pkts"""
2175 def setUpConstants(cls):
2176 super(IPSecIPv6Fwd, cls).setUpConstants()
2179 super(IPSecIPv6Fwd, self).setUp()
2180 # store SPD objects so we can remove configs on tear down
2182 self.spd_policies = []
2185 # remove SPD policies
2186 for obj in self.spd_policies:
2187 obj.remove_vpp_config()
2188 self.spd_policies = []
2189 # remove SPD items (interface bindings first, then SPD)
2190 for obj in reversed(self.spd_objs):
2191 obj.remove_vpp_config()
2193 # close down pg intfs
2194 for pg in self.pg_interfaces:
2197 super(IPSecIPv6Fwd, self).tearDown()
2199 def create_interfaces(self, num_ifs=2):
2200 # create interfaces pg0 ... pg<num_ifs>
2201 self.create_pg_interfaces(range(num_ifs))
2202 for pg in self.pg_interfaces:
2203 # put the interface up
2205 # configure IPv6 address on the interface
2208 self.logger.info(self.vapi.ppcli("show int addr"))
2210 def spd_create_and_intf_add(self, spd_id, pg_list):
2211 spd = VppIpsecSpd(self, spd_id)
2212 spd.add_vpp_config()
2213 self.spd_objs.append(spd)
2215 spdItf = VppIpsecSpdItfBinding(self, spd, pg)
2216 spdItf.add_vpp_config()
2217 self.spd_objs.append(spdItf)
2219 def get_policy(self, policy_type):
2220 e = VppEnum.vl_api_ipsec_spd_action_t
2221 if policy_type == "protect":
2222 return e.IPSEC_API_SPD_ACTION_PROTECT
2223 elif policy_type == "bypass":
2224 return e.IPSEC_API_SPD_ACTION_BYPASS
2225 elif policy_type == "discard":
2226 return e.IPSEC_API_SPD_ACTION_DISCARD
2228 raise Exception("Invalid policy type: %s", policy_type)
2230 def spd_add_rem_policy(
2242 local_ip_start=ip_address("0::0"),
2243 local_ip_stop=ip_address("ffff:ffff:ffff:ffff:ffff:ffff:ffff:ffff"),
2244 remote_ip_start=ip_address("0::0"),
2245 remote_ip_stop=ip_address("ffff:ffff:ffff:ffff:ffff:ffff:ffff:ffff"),
2246 remote_port_start=0,
2247 remote_port_stop=65535,
2249 local_port_stop=65535,
2251 spd = VppIpsecSpd(self, spd_id)
2254 src_range_low = ip_address("0::0")
2255 src_range_high = ip_address("ffff:ffff:ffff:ffff:ffff:ffff:ffff:ffff")
2256 dst_range_low = ip_address("0::0")
2257 dst_range_high = ip_address("ffff:ffff:ffff:ffff:ffff:ffff:ffff:ffff")
2260 src_range_low = local_ip_start
2261 src_range_high = local_ip_stop
2262 dst_range_low = remote_ip_start
2263 dst_range_high = remote_ip_stop
2266 src_range_low = src_if.remote_ip6
2267 src_range_high = src_if.remote_ip6
2268 dst_range_low = dst_if.remote_ip6
2269 dst_range_high = dst_if.remote_ip6
2271 spdEntry = VppIpsecSpdEntry(
2281 policy=self.get_policy(policy_type),
2283 remote_port_start=remote_port_start,
2284 remote_port_stop=remote_port_stop,
2285 local_port_start=local_port_start,
2286 local_port_stop=local_port_stop,
2290 spdEntry.add_vpp_config()
2291 self.spd_policies.append(spdEntry)
2293 spdEntry.remove_vpp_config()
2294 self.spd_policies.remove(spdEntry)
2295 self.logger.info(self.vapi.ppcli("show ipsec all"))
2298 def create_stream(self, src_if, dst_if, pkt_count, src_prt=1234, dst_prt=5678):
2300 for i in range(pkt_count):
2301 # create packet info stored in the test case instance
2302 info = self.create_packet_info(src_if, dst_if)
2303 # convert the info into packet payload
2304 payload = self.info_to_payload(info)
2305 # create the packet itself
2307 Ether(dst=src_if.local_mac, src=src_if.remote_mac)
2308 / IPv6(src=src_if.remote_ip6, dst=dst_if.remote_ip6)
2309 / UDP(sport=src_prt, dport=dst_prt)
2312 # store a copy of the packet in the packet info
2313 info.data = p.copy()
2314 # append the packet to the list
2316 # return the created packet list
2319 def verify_capture(self, src_if, dst_if, capture):
2321 for packet in capture:
2325 # convert the payload to packet info object
2326 payload_info = self.payload_to_info(packet)
2327 # make sure the indexes match
2329 payload_info.src, src_if.sw_if_index, "source sw_if_index"
2332 payload_info.dst, dst_if.sw_if_index, "destination sw_if_index"
2334 packet_info = self.get_next_packet_info_for_interface2(
2335 src_if.sw_if_index, dst_if.sw_if_index, packet_info
2337 # make sure we didn't run out of saved packets
2338 self.assertIsNotNone(packet_info)
2340 payload_info.index, packet_info.index, "packet info index"
2342 saved_packet = packet_info.data # fetch the saved packet
2343 # assert the values match
2344 self.assert_equal(ip.src, saved_packet[IPv6].src, "IP source address")
2345 # ... more assertions here
2346 self.assert_equal(udp.sport, saved_packet[UDP].sport, "UDP source port")
2347 except Exception as e:
2348 self.logger.error(ppp("Unexpected or invalid packet:", packet))
2350 remaining_packet = self.get_next_packet_info_for_interface2(
2351 src_if.sw_if_index, dst_if.sw_if_index, packet_info
2355 "Interface %s: Packet expected from interface "
2356 "%s didn't arrive" % (dst_if.name, src_if.name),
2359 def verify_policy_match(self, pkt_count, spdEntry):
2360 self.logger.info("XXXX %s %s", str(spdEntry), str(spdEntry.get_stats()))
2361 matched_pkts = spdEntry.get_stats().get("packets")
2362 self.logger.info("Policy %s matched: %d pkts", str(spdEntry), matched_pkts)
2363 self.assert_equal(pkt_count, matched_pkts)
2366 if __name__ == "__main__":
2367 unittest.main(testRunner=VppTestRunner)