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)
636 if ESP == self.encryption_type:
637 undersize_node_name = "/err/%s/runt" % self.tra4_decrypt_node_name[0]
638 undersize_count = self.statistics.get_err_counter(undersize_node_name)
641 # send packets with seq numbers 1->34
642 # this means the window size is still in Case B (see RFC4303
645 # for reasons i haven't investigated Scapy won't create a packet with
650 Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
651 / p.scapy_tra_sa.encrypt(
652 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
656 for seq in range(1, 34)
658 recv_pkts = self.send_and_expect(self.tra_if, pkts, self.tra_if)
660 # replayed packets are dropped
661 self.send_and_assert_no_replies(self.tra_if, pkts, timeout=0.2)
662 replay_count += len(pkts)
663 self.assertEqual(self.get_replay_counts(p), replay_count)
666 # now send a batch of packets all with the same sequence number
667 # the first packet in the batch is legitimate, the rest bogus
669 self.vapi.cli("clear error")
670 self.vapi.cli("clear node counters")
672 src=self.tra_if.remote_mac, dst=self.tra_if.local_mac
673 ) / p.scapy_tra_sa.encrypt(
674 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
677 recv_pkts = self.send_and_expect(self.tra_if, pkts * 8, self.tra_if, n_rx=1)
679 self.assertEqual(self.get_replay_counts(p), replay_count)
682 # now move the window over to 257 (more than one byte) and into Case A
684 self.vapi.cli("clear error")
686 src=self.tra_if.remote_mac, dst=self.tra_if.local_mac
687 ) / p.scapy_tra_sa.encrypt(
688 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
691 recv_pkts = self.send_and_expect(self.tra_if, [pkt], self.tra_if)
693 # replayed packets are dropped
694 self.send_and_assert_no_replies(self.tra_if, pkt * 3, timeout=0.2)
696 self.assertEqual(self.get_replay_counts(p), replay_count)
698 # the window size is 64 packets
699 # in window are still accepted
701 src=self.tra_if.remote_mac, dst=self.tra_if.local_mac
702 ) / p.scapy_tra_sa.encrypt(
703 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
706 recv_pkts = self.send_and_expect(self.tra_if, [pkt], self.tra_if)
708 # a packet that does not decrypt does not move the window forward
709 bogus_sa = SecurityAssociation(
710 self.encryption_type,
712 crypt_algo=p.crypt_algo,
713 crypt_key=mk_scapy_crypt_key(p)[::-1],
714 auth_algo=p.auth_algo,
715 auth_key=p.auth_key[::-1],
718 src=self.tra_if.remote_mac, dst=self.tra_if.local_mac
719 ) / bogus_sa.encrypt(
720 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
723 self.send_and_assert_no_replies(self.tra_if, pkt * 17, timeout=0.2)
725 hash_failed_count += 17
726 self.assertEqual(self.get_hash_failed_counts(p), hash_failed_count)
728 # a malformed 'runt' packet
729 # created by a mis-constructed SA
730 if ESP == self.encryption_type and p.crypt_algo != "NULL":
731 bogus_sa = SecurityAssociation(self.encryption_type, p.vpp_tra_spi)
733 src=self.tra_if.remote_mac, dst=self.tra_if.local_mac
734 ) / bogus_sa.encrypt(
735 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
738 self.send_and_assert_no_replies(self.tra_if, pkt * 17, timeout=0.2)
740 undersize_count += 17
741 self.assert_error_counter_equal(undersize_node_name, undersize_count)
743 # which we can determine since this packet is still in the window
745 src=self.tra_if.remote_mac, dst=self.tra_if.local_mac
746 ) / p.scapy_tra_sa.encrypt(
747 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
750 self.send_and_expect(self.tra_if, [pkt], self.tra_if)
753 # out of window are dropped
754 # this is Case B. So VPP will consider this to be a high seq num wrap
755 # and so the decrypt attempt will fail
758 src=self.tra_if.remote_mac, dst=self.tra_if.local_mac
759 ) / p.scapy_tra_sa.encrypt(
760 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
763 self.send_and_assert_no_replies(self.tra_if, pkt * 17, timeout=0.2)
766 # an out of window error with ESN looks like a high sequence
767 # wrap. but since it isn't then the verify will fail.
768 hash_failed_count += 17
769 self.assertEqual(self.get_hash_failed_counts(p), hash_failed_count)
773 self.assertEqual(self.get_replay_counts(p), replay_count)
775 # valid packet moves the window over to 258
777 src=self.tra_if.remote_mac, dst=self.tra_if.local_mac
778 ) / p.scapy_tra_sa.encrypt(
779 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
782 rx = self.send_and_expect(self.tra_if, [pkt], self.tra_if)
783 decrypted = p.vpp_tra_sa.decrypt(rx[0][IP])
786 # move VPP's SA TX seq-num to just before the seq-number wrap.
787 # then fire in a packet that VPP should drop on TX because it
788 # causes the TX seq number to wrap; unless we're using extened sequence
791 self.vapi.cli("test ipsec sa %d seq 0xffffffff" % p.scapy_tra_sa_id)
792 self.logger.info(self.vapi.ppcli("show ipsec sa 0"))
793 self.logger.info(self.vapi.ppcli("show ipsec sa 1"))
797 Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
798 / p.scapy_tra_sa.encrypt(
799 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
803 for seq in range(259, 280)
807 rxs = self.send_and_expect(self.tra_if, pkts, self.tra_if)
810 # in order for scapy to decrypt its SA's high order number needs
813 p.vpp_tra_sa.seq_num = 0x100000000
815 decrypted = p.vpp_tra_sa.decrypt(rx[0][IP])
818 # wrap scapy's TX high sequence number. VPP is in case B, so it
819 # will consider this a high seq wrap also.
820 # The low seq num we set it to will place VPP's RX window in Case A
822 p.scapy_tra_sa.seq_num = 0x100000005
824 src=self.tra_if.remote_mac, dst=self.tra_if.local_mac
825 ) / p.scapy_tra_sa.encrypt(
826 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
829 rx = self.send_and_expect(self.tra_if, [pkt], self.tra_if)
831 decrypted = p.vpp_tra_sa.decrypt(rx[0][IP])
834 # A packet that has seq num between (2^32-64) and 5 is within
837 p.scapy_tra_sa.seq_num = 0xFFFFFFFD
839 src=self.tra_if.remote_mac, dst=self.tra_if.local_mac
840 ) / p.scapy_tra_sa.encrypt(
841 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
844 rx = self.send_and_expect(self.tra_if, [pkt], self.tra_if)
845 decrypted = p.vpp_tra_sa.decrypt(rx[0][IP])
848 # While in case A we cannot wrap the high sequence number again
849 # because VPP will consider this packet to be one that moves the
853 src=self.tra_if.remote_mac, dst=self.tra_if.local_mac
854 ) / p.scapy_tra_sa.encrypt(
855 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
858 self.send_and_assert_no_replies(
859 self.tra_if, [pkt], self.tra_if, timeout=0.2
862 hash_failed_count += 1
863 self.assertEqual(self.get_hash_failed_counts(p), hash_failed_count)
866 # but if we move the window forward to case B, then we can wrap
869 p.scapy_tra_sa.seq_num = 0x100000555
871 src=self.tra_if.remote_mac, dst=self.tra_if.local_mac
872 ) / p.scapy_tra_sa.encrypt(
873 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
876 rx = self.send_and_expect(self.tra_if, [pkt], self.tra_if)
877 decrypted = p.vpp_tra_sa.decrypt(rx[0][IP])
879 p.scapy_tra_sa.seq_num = 0x200000444
881 src=self.tra_if.remote_mac, dst=self.tra_if.local_mac
882 ) / p.scapy_tra_sa.encrypt(
883 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
886 rx = self.send_and_expect(self.tra_if, [pkt], self.tra_if)
887 decrypted = p.vpp_tra_sa.decrypt(rx[0][IP])
891 # without ESN TX sequence numbers can't wrap and packets are
892 # dropped from here on out.
894 self.send_and_assert_no_replies(self.tra_if, pkts, timeout=0.2)
895 seq_cycle_count += len(pkts)
896 self.assert_error_counter_equal(seq_cycle_node_name, seq_cycle_count)
898 # move the security-associations seq number on to the last we used
899 self.vapi.cli("test ipsec sa %d seq 0x15f" % p.scapy_tra_sa_id)
900 p.scapy_tra_sa.seq_num = 351
901 p.vpp_tra_sa.seq_num = 351
903 def verify_tra_lost(self):
904 p = self.params[socket.AF_INET]
905 esn_en = p.vpp_tra_sa.esn_en
908 # send packets with seq numbers 1->34
909 # this means the window size is still in Case B (see RFC4303
912 # for reasons i haven't investigated Scapy won't create a packet with
917 Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
918 / p.scapy_tra_sa.encrypt(
919 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
923 for seq in range(1, 3)
925 self.send_and_expect(self.tra_if, pkts, self.tra_if)
927 self.assertEqual(p.tra_sa_out.get_lost(), 0)
929 # skip a sequence number
932 Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
933 / p.scapy_tra_sa.encrypt(
934 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
938 for seq in range(4, 6)
940 self.send_and_expect(self.tra_if, pkts, self.tra_if)
942 self.assertEqual(p.tra_sa_out.get_lost(), 0)
944 # the lost packet are counted untill we get up past the first
945 # sizeof(replay_window) packets
948 Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
949 / p.scapy_tra_sa.encrypt(
950 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
954 for seq in range(6, 100)
956 self.send_and_expect(self.tra_if, pkts, self.tra_if)
958 self.assertEqual(p.tra_sa_out.get_lost(), 1)
960 # lost of holes in the sequence
963 Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
964 / p.scapy_tra_sa.encrypt(
965 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
969 for seq in range(100, 200, 2)
971 self.send_and_expect(self.tra_if, pkts, self.tra_if, n_rx=50)
975 Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
976 / p.scapy_tra_sa.encrypt(
977 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
981 for seq in range(200, 300)
983 self.send_and_expect(self.tra_if, pkts, self.tra_if)
985 self.assertEqual(p.tra_sa_out.get_lost(), 51)
987 # a big hole in the seq number space
990 Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
991 / p.scapy_tra_sa.encrypt(
992 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
996 for seq in range(400, 500)
998 self.send_and_expect(self.tra_if, pkts, self.tra_if)
1000 self.assertEqual(p.tra_sa_out.get_lost(), 151)
1002 def verify_tra_basic4(self, count=1, payload_size=54):
1003 """ipsec v4 transport basic test"""
1004 self.vapi.cli("clear errors")
1005 self.vapi.cli("clear ipsec sa")
1007 p = self.params[socket.AF_INET]
1008 send_pkts = self.gen_encrypt_pkts(
1012 src=self.tra_if.remote_ip4,
1013 dst=self.tra_if.local_ip4,
1015 payload_size=payload_size,
1017 recv_pkts = self.send_and_expect(self.tra_if, send_pkts, self.tra_if)
1018 for rx in recv_pkts:
1019 self.assertEqual(len(rx) - len(Ether()), rx[IP].len)
1020 self.assert_packet_checksums_valid(rx)
1022 decrypted = p.vpp_tra_sa.decrypt(rx[IP])
1023 self.assert_packet_checksums_valid(decrypted)
1025 self.logger.debug(ppp("Unexpected packet:", rx))
1028 self.logger.info(self.vapi.ppcli("show error"))
1029 self.logger.info(self.vapi.ppcli("show ipsec all"))
1031 pkts = p.tra_sa_in.get_stats()["packets"]
1033 pkts, count, "incorrect SA in counts: expected %d != %d" % (count, pkts)
1035 pkts = p.tra_sa_out.get_stats()["packets"]
1037 pkts, count, "incorrect SA out counts: expected %d != %d" % (count, pkts)
1039 self.assertEqual(p.tra_sa_out.get_lost(), 0)
1040 self.assertEqual(p.tra_sa_in.get_lost(), 0)
1042 self.assert_packet_counter_equal(self.tra4_encrypt_node_name, count)
1043 self.assert_packet_counter_equal(self.tra4_decrypt_node_name[0], count)
1046 class IpsecTra4Tests(IpsecTra4):
1047 """UT test methods for Transport v4"""
1049 def test_tra_anti_replay(self):
1050 """ipsec v4 transport anti-replay test"""
1051 self.verify_tra_anti_replay()
1053 def test_tra_lost(self):
1054 """ipsec v4 transport lost packet test"""
1055 self.verify_tra_lost()
1057 def test_tra_basic(self, count=1):
1058 """ipsec v4 transport basic test"""
1059 self.verify_tra_basic4(count=1)
1061 def test_tra_burst(self):
1062 """ipsec v4 transport burst test"""
1063 self.verify_tra_basic4(count=257)
1066 class IpsecTra6(object):
1067 """verify methods for Transport v6"""
1069 def verify_tra_basic6(self, count=1, payload_size=54):
1070 self.vapi.cli("clear errors")
1071 self.vapi.cli("clear ipsec sa")
1073 p = self.params[socket.AF_INET6]
1074 send_pkts = self.gen_encrypt_pkts6(
1078 src=self.tra_if.remote_ip6,
1079 dst=self.tra_if.local_ip6,
1081 payload_size=payload_size,
1083 recv_pkts = self.send_and_expect(self.tra_if, send_pkts, self.tra_if)
1084 for rx in recv_pkts:
1085 self.assertEqual(len(rx) - len(Ether()) - len(IPv6()), rx[IPv6].plen)
1087 decrypted = p.vpp_tra_sa.decrypt(rx[IPv6])
1088 self.assert_packet_checksums_valid(decrypted)
1090 self.logger.debug(ppp("Unexpected packet:", rx))
1093 self.logger.info(self.vapi.ppcli("show error"))
1094 self.logger.info(self.vapi.ppcli("show ipsec all"))
1096 pkts = p.tra_sa_in.get_stats()["packets"]
1098 pkts, count, "incorrect SA in counts: expected %d != %d" % (count, pkts)
1100 pkts = p.tra_sa_out.get_stats()["packets"]
1102 pkts, count, "incorrect SA out counts: expected %d != %d" % (count, pkts)
1104 self.assert_packet_counter_equal(self.tra6_encrypt_node_name, count)
1105 self.assert_packet_counter_equal(self.tra6_decrypt_node_name[0], count)
1107 def gen_encrypt_pkts_ext_hdrs6(
1108 self, sa, sw_intf, src, dst, count=1, payload_size=54
1111 Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac)
1113 IPv6(src=src, dst=dst)
1114 / ICMPv6EchoRequest(id=0, seq=1, data="X" * payload_size)
1116 for i in range(count)
1119 def gen_pkts_ext_hdrs6(self, sw_intf, src, dst, count=1, payload_size=54):
1121 Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac)
1122 / IPv6(src=src, dst=dst)
1123 / IPv6ExtHdrHopByHop()
1124 / IPv6ExtHdrFragment(id=2, offset=200)
1125 / Raw(b"\xff" * 200)
1126 for i in range(count)
1129 def verify_tra_encrypted6(self, p, sa, rxs):
1132 self.assert_packet_checksums_valid(rx)
1134 decrypt_pkt = p.vpp_tra_sa.decrypt(rx[IPv6])
1135 decrypted.append(decrypt_pkt)
1136 self.assert_equal(decrypt_pkt.src, self.tra_if.local_ip6)
1137 self.assert_equal(decrypt_pkt.dst, self.tra_if.remote_ip6)
1139 self.logger.debug(ppp("Unexpected packet:", rx))
1141 self.logger.debug(ppp("Decrypted packet:", decrypt_pkt))
1147 def verify_tra_66_ext_hdrs(self, p):
1151 # check we can decrypt with options
1153 tx = self.gen_encrypt_pkts_ext_hdrs6(
1156 src=self.tra_if.remote_ip6,
1157 dst=self.tra_if.local_ip6,
1160 self.send_and_expect(self.tra_if, tx, self.tra_if)
1163 # injecting a packet from ourselves to be routed of box is a hack
1164 # but it matches an outbout policy, alors je ne regrette rien
1167 # one extension before ESP
1169 Ether(src=self.pg2.remote_mac, dst=self.pg2.local_mac)
1170 / IPv6(src=self.tra_if.local_ip6, dst=self.tra_if.remote_ip6)
1171 / IPv6ExtHdrFragment(id=2, offset=200)
1172 / Raw(b"\xff" * 200)
1175 rxs = self.send_and_expect(self.pg2, [tx], self.tra_if)
1176 dcs = self.verify_tra_encrypted6(p, p.vpp_tra_sa, rxs)
1179 # for reasons i'm not going to investigate scapy does not
1180 # created the correct headers after decrypt. but reparsing
1181 # the ipv6 packet fixes it
1182 dc = IPv6(raw(dc[IPv6]))
1183 self.assert_equal(dc[IPv6ExtHdrFragment].id, 2)
1185 # two extensions before ESP
1187 Ether(src=self.pg2.remote_mac, dst=self.pg2.local_mac)
1188 / IPv6(src=self.tra_if.local_ip6, dst=self.tra_if.remote_ip6)
1189 / IPv6ExtHdrHopByHop()
1190 / IPv6ExtHdrFragment(id=2, offset=200)
1191 / Raw(b"\xff" * 200)
1194 rxs = self.send_and_expect(self.pg2, [tx], self.tra_if)
1195 dcs = self.verify_tra_encrypted6(p, p.vpp_tra_sa, rxs)
1198 dc = IPv6(raw(dc[IPv6]))
1199 self.assertTrue(dc[IPv6ExtHdrHopByHop])
1200 self.assert_equal(dc[IPv6ExtHdrFragment].id, 2)
1202 # two extensions before ESP, one after
1204 Ether(src=self.pg2.remote_mac, dst=self.pg2.local_mac)
1205 / IPv6(src=self.tra_if.local_ip6, dst=self.tra_if.remote_ip6)
1206 / IPv6ExtHdrHopByHop()
1207 / IPv6ExtHdrFragment(id=2, offset=200)
1208 / IPv6ExtHdrDestOpt()
1209 / Raw(b"\xff" * 200)
1212 rxs = self.send_and_expect(self.pg2, [tx], self.tra_if)
1213 dcs = self.verify_tra_encrypted6(p, p.vpp_tra_sa, rxs)
1216 dc = IPv6(raw(dc[IPv6]))
1217 self.assertTrue(dc[IPv6ExtHdrDestOpt])
1218 self.assertTrue(dc[IPv6ExtHdrHopByHop])
1219 self.assert_equal(dc[IPv6ExtHdrFragment].id, 2)
1222 class IpsecTra6Tests(IpsecTra6):
1223 """UT test methods for Transport v6"""
1225 def test_tra_basic6(self):
1226 """ipsec v6 transport basic test"""
1227 self.verify_tra_basic6(count=1)
1229 def test_tra_burst6(self):
1230 """ipsec v6 transport burst test"""
1231 self.verify_tra_basic6(count=257)
1234 class IpsecTra6ExtTests(IpsecTra6):
1235 def test_tra_ext_hdrs_66(self):
1236 """ipsec 6o6 tra extension headers test"""
1237 self.verify_tra_66_ext_hdrs(self.params[socket.AF_INET6])
1240 class IpsecTra46Tests(IpsecTra4Tests, IpsecTra6Tests):
1241 """UT test methods for Transport v6 and v4"""
1246 class IpsecTun4(object):
1247 """verify methods for Tunnel v4"""
1249 def verify_counters4(self, p, count, n_frags=None, worker=None):
1252 if hasattr(p, "spd_policy_in_any"):
1253 pkts = p.spd_policy_in_any.get_stats(worker)["packets"]
1257 "incorrect SPD any policy: expected %d != %d" % (count, pkts),
1260 if hasattr(p, "tun_sa_in"):
1261 pkts = p.tun_sa_in.get_stats(worker)["packets"]
1263 pkts, count, "incorrect SA in counts: expected %d != %d" % (count, pkts)
1265 pkts = p.tun_sa_out.get_stats(worker)["packets"]
1269 "incorrect SA out counts: expected %d != %d" % (count, pkts),
1272 self.assert_packet_counter_equal(self.tun4_encrypt_node_name, n_frags)
1273 self.assert_packet_counter_equal(self.tun4_decrypt_node_name[0], count)
1275 def verify_decrypted(self, p, rxs):
1277 self.assert_equal(rx[IP].src, p.remote_tun_if_host)
1278 self.assert_equal(rx[IP].dst, self.pg1.remote_ip4)
1279 self.assert_packet_checksums_valid(rx)
1281 def verify_esp_padding(self, sa, esp_payload, decrypt_pkt):
1282 align = sa.crypt_algo.block_size
1285 exp_len = (len(decrypt_pkt) + 2 + (align - 1)) & ~(align - 1)
1286 exp_len += sa.crypt_algo.iv_size
1287 exp_len += sa.crypt_algo.icv_size or sa.auth_algo.icv_size
1288 self.assertEqual(exp_len, len(esp_payload))
1290 def verify_encrypted(self, p, sa, rxs):
1294 self.assertEqual(rx[UDP].dport, 4500)
1295 self.assert_packet_checksums_valid(rx)
1296 self.assertEqual(len(rx) - len(Ether()), rx[IP].len)
1299 decrypt_pkt = p.vpp_tun_sa.decrypt(rx_ip)
1300 if not decrypt_pkt.haslayer(IP):
1301 decrypt_pkt = IP(decrypt_pkt[Raw].load)
1302 if rx_ip.proto == socket.IPPROTO_ESP:
1303 self.verify_esp_padding(sa, rx_ip[ESP].data, decrypt_pkt)
1304 decrypt_pkts.append(decrypt_pkt)
1305 self.assert_equal(decrypt_pkt.src, self.pg1.remote_ip4)
1306 self.assert_equal(decrypt_pkt.dst, p.remote_tun_if_host)
1308 self.logger.debug(ppp("Unexpected packet:", rx))
1310 self.logger.debug(ppp("Decrypted packet:", decrypt_pkt))
1314 pkts = reassemble4(decrypt_pkts)
1316 self.assert_packet_checksums_valid(pkt)
1318 def verify_tun_44(self, p, count=1, payload_size=64, n_rx=None):
1319 self.vapi.cli("clear errors")
1320 self.vapi.cli("clear ipsec counters")
1321 self.vapi.cli("clear ipsec sa")
1325 send_pkts = self.gen_encrypt_pkts(
1329 src=p.remote_tun_if_host,
1330 dst=self.pg1.remote_ip4,
1332 payload_size=payload_size,
1334 recv_pkts = self.send_and_expect(self.tun_if, send_pkts, self.pg1)
1335 self.verify_decrypted(p, recv_pkts)
1337 send_pkts = self.gen_pkts(
1339 src=self.pg1.remote_ip4,
1340 dst=p.remote_tun_if_host,
1342 payload_size=payload_size,
1344 recv_pkts = self.send_and_expect(self.pg1, send_pkts, self.tun_if, n_rx)
1345 self.verify_encrypted(p, p.vpp_tun_sa, recv_pkts)
1347 for rx in recv_pkts:
1348 self.assertEqual(rx[IP].src, p.tun_src)
1349 self.assertEqual(rx[IP].dst, p.tun_dst)
1352 self.logger.info(self.vapi.ppcli("show error"))
1353 self.logger.info(self.vapi.ppcli("show ipsec all"))
1355 self.logger.info(self.vapi.ppcli("show ipsec sa 0"))
1356 self.logger.info(self.vapi.ppcli("show ipsec sa 4"))
1357 self.verify_counters4(p, count, n_rx)
1359 def verify_tun_dropped_44(self, p, count=1, payload_size=64, n_rx=None):
1360 self.vapi.cli("clear errors")
1364 send_pkts = self.gen_encrypt_pkts(
1368 src=p.remote_tun_if_host,
1369 dst=self.pg1.remote_ip4,
1372 self.send_and_assert_no_replies(self.tun_if, send_pkts)
1374 send_pkts = self.gen_pkts(
1376 src=self.pg1.remote_ip4,
1377 dst=p.remote_tun_if_host,
1379 payload_size=payload_size,
1381 self.send_and_assert_no_replies(self.pg1, send_pkts)
1384 self.logger.info(self.vapi.ppcli("show error"))
1385 self.logger.info(self.vapi.ppcli("show ipsec all"))
1387 def verify_tun_reass_44(self, p):
1388 self.vapi.cli("clear errors")
1389 self.vapi.ip_reassembly_enable_disable(
1390 sw_if_index=self.tun_if.sw_if_index, enable_ip4=True
1394 send_pkts = self.gen_encrypt_pkts(
1398 src=p.remote_tun_if_host,
1399 dst=self.pg1.remote_ip4,
1403 send_pkts = fragment_rfc791(send_pkts[0], 1400)
1404 recv_pkts = self.send_and_expect(self.tun_if, send_pkts, self.pg1, n_rx=1)
1405 self.verify_decrypted(p, recv_pkts)
1407 send_pkts = self.gen_pkts(
1408 self.pg1, src=self.pg1.remote_ip4, dst=p.remote_tun_if_host, count=1
1410 recv_pkts = self.send_and_expect(self.pg1, send_pkts, self.tun_if)
1411 self.verify_encrypted(p, p.vpp_tun_sa, recv_pkts)
1414 self.logger.info(self.vapi.ppcli("show error"))
1415 self.logger.info(self.vapi.ppcli("show ipsec all"))
1417 self.verify_counters4(p, 1, 1)
1418 self.vapi.ip_reassembly_enable_disable(
1419 sw_if_index=self.tun_if.sw_if_index, enable_ip4=False
1422 def verify_tun_64(self, p, count=1):
1423 self.vapi.cli("clear errors")
1424 self.vapi.cli("clear ipsec sa")
1426 send_pkts = self.gen_encrypt_pkts6(
1430 src=p.remote_tun_if_host6,
1431 dst=self.pg1.remote_ip6,
1434 recv_pkts = self.send_and_expect(self.tun_if, send_pkts, self.pg1)
1435 for recv_pkt in recv_pkts:
1436 self.assert_equal(recv_pkt[IPv6].src, p.remote_tun_if_host6)
1437 self.assert_equal(recv_pkt[IPv6].dst, self.pg1.remote_ip6)
1438 self.assert_packet_checksums_valid(recv_pkt)
1439 send_pkts = self.gen_pkts6(
1442 src=self.pg1.remote_ip6,
1443 dst=p.remote_tun_if_host6,
1446 recv_pkts = self.send_and_expect(self.pg1, send_pkts, self.tun_if)
1447 for recv_pkt in recv_pkts:
1449 decrypt_pkt = p.vpp_tun_sa.decrypt(recv_pkt[IP])
1450 if not decrypt_pkt.haslayer(IPv6):
1451 decrypt_pkt = IPv6(decrypt_pkt[Raw].load)
1452 self.assert_equal(decrypt_pkt.src, self.pg1.remote_ip6)
1453 self.assert_equal(decrypt_pkt.dst, p.remote_tun_if_host6)
1454 self.assert_packet_checksums_valid(decrypt_pkt)
1456 self.logger.error(ppp("Unexpected packet:", recv_pkt))
1458 self.logger.debug(ppp("Decrypted packet:", decrypt_pkt))
1463 self.logger.info(self.vapi.ppcli("show error"))
1464 self.logger.info(self.vapi.ppcli("show ipsec all"))
1466 self.verify_counters4(p, count)
1468 def verify_keepalive(self, p):
1469 # the sizeof Raw is calculated to pad to the minimum ehternet
1470 # frame size of 64 btyes
1472 Ether(src=self.tun_if.remote_mac, dst=self.tun_if.local_mac)
1473 / IP(src=p.remote_tun_if_host, dst=self.tun_if.local_ip4)
1474 / UDP(sport=333, dport=4500)
1478 self.send_and_assert_no_replies(self.tun_if, pkt * 31)
1479 self.assert_error_counter_equal(
1480 "/err/%s/nat_keepalive" % self.tun4_input_node, 31
1484 Ether(src=self.tun_if.remote_mac, dst=self.tun_if.local_mac)
1485 / IP(src=p.remote_tun_if_host, dst=self.tun_if.local_ip4)
1486 / UDP(sport=333, dport=4500)
1489 self.send_and_assert_no_replies(self.tun_if, pkt * 31)
1490 self.assert_error_counter_equal("/err/%s/too_short" % self.tun4_input_node, 31)
1493 Ether(src=self.tun_if.remote_mac, dst=self.tun_if.local_mac)
1494 / IP(src=p.remote_tun_if_host, dst=self.tun_if.local_ip4)
1495 / UDP(sport=333, dport=4500)
1499 self.send_and_assert_no_replies(self.tun_if, pkt * 31)
1500 self.assert_error_counter_equal("/err/%s/too_short" % self.tun4_input_node, 62)
1503 class IpsecTun4Tests(IpsecTun4):
1504 """UT test methods for Tunnel v4"""
1506 def test_tun_basic44(self):
1507 """ipsec 4o4 tunnel basic test"""
1508 self.verify_tun_44(self.params[socket.AF_INET], count=1)
1509 self.tun_if.admin_down()
1510 self.tun_if.resolve_arp()
1511 self.tun_if.admin_up()
1512 self.verify_tun_44(self.params[socket.AF_INET], count=1)
1514 def test_tun_reass_basic44(self):
1515 """ipsec 4o4 tunnel basic reassembly test"""
1516 self.verify_tun_reass_44(self.params[socket.AF_INET])
1518 def test_tun_burst44(self):
1519 """ipsec 4o4 tunnel burst test"""
1520 self.verify_tun_44(self.params[socket.AF_INET], count=127)
1523 class IpsecTun6(object):
1524 """verify methods for Tunnel v6"""
1526 def verify_counters6(self, p_in, p_out, count, worker=None):
1527 if hasattr(p_in, "tun_sa_in"):
1528 pkts = p_in.tun_sa_in.get_stats(worker)["packets"]
1530 pkts, count, "incorrect SA in counts: expected %d != %d" % (count, pkts)
1532 if hasattr(p_out, "tun_sa_out"):
1533 pkts = p_out.tun_sa_out.get_stats(worker)["packets"]
1537 "incorrect SA out counts: expected %d != %d" % (count, pkts),
1539 self.assert_packet_counter_equal(self.tun6_encrypt_node_name, count)
1540 self.assert_packet_counter_equal(self.tun6_decrypt_node_name[0], count)
1542 def verify_decrypted6(self, p, rxs):
1544 self.assert_equal(rx[IPv6].src, p.remote_tun_if_host)
1545 self.assert_equal(rx[IPv6].dst, self.pg1.remote_ip6)
1546 self.assert_packet_checksums_valid(rx)
1548 def verify_encrypted6(self, p, sa, rxs):
1550 self.assert_packet_checksums_valid(rx)
1551 self.assertEqual(len(rx) - len(Ether()) - len(IPv6()), rx[IPv6].plen)
1552 self.assert_equal(rx[IPv6].hlim, p.outer_hop_limit)
1553 if p.outer_flow_label:
1554 self.assert_equal(rx[IPv6].fl, p.outer_flow_label)
1556 decrypt_pkt = p.vpp_tun_sa.decrypt(rx[IPv6])
1557 if not decrypt_pkt.haslayer(IPv6):
1558 decrypt_pkt = IPv6(decrypt_pkt[Raw].load)
1559 self.assert_packet_checksums_valid(decrypt_pkt)
1560 self.assert_equal(decrypt_pkt.src, self.pg1.remote_ip6)
1561 self.assert_equal(decrypt_pkt.dst, p.remote_tun_if_host)
1562 self.assert_equal(decrypt_pkt.hlim, p.inner_hop_limit - 1)
1563 self.assert_equal(decrypt_pkt.fl, p.inner_flow_label)
1565 self.logger.debug(ppp("Unexpected packet:", rx))
1567 self.logger.debug(ppp("Decrypted packet:", decrypt_pkt))
1572 def verify_drop_tun_tx_66(self, p_in, count=1, payload_size=64):
1573 self.vapi.cli("clear errors")
1574 self.vapi.cli("clear ipsec sa")
1576 send_pkts = self.gen_pkts6(
1579 src=self.pg1.remote_ip6,
1580 dst=p_in.remote_tun_if_host,
1582 payload_size=payload_size,
1584 self.send_and_assert_no_replies(self.tun_if, send_pkts)
1585 self.logger.info(self.vapi.cli("sh punt stats"))
1587 def verify_drop_tun_rx_66(self, p_in, count=1, payload_size=64):
1588 self.vapi.cli("clear errors")
1589 self.vapi.cli("clear ipsec sa")
1591 send_pkts = self.gen_encrypt_pkts6(
1595 src=p_in.remote_tun_if_host,
1596 dst=self.pg1.remote_ip6,
1599 self.send_and_assert_no_replies(self.tun_if, send_pkts)
1601 def verify_drop_tun_66(self, p_in, count=1, payload_size=64):
1602 self.verify_drop_tun_tx_66(p_in, count=count, payload_size=payload_size)
1603 self.verify_drop_tun_rx_66(p_in, count=count, payload_size=payload_size)
1605 def verify_tun_66(self, p_in, p_out=None, count=1, payload_size=64):
1606 self.vapi.cli("clear errors")
1607 self.vapi.cli("clear ipsec sa")
1611 send_pkts = self.gen_encrypt_pkts6(
1615 src=p_in.remote_tun_if_host,
1616 dst=self.pg1.remote_ip6,
1618 payload_size=payload_size,
1620 recv_pkts = self.send_and_expect(self.tun_if, send_pkts, self.pg1)
1621 self.verify_decrypted6(p_in, recv_pkts)
1623 send_pkts = self.gen_pkts6(
1626 src=self.pg1.remote_ip6,
1627 dst=p_out.remote_tun_if_host,
1629 payload_size=payload_size,
1631 recv_pkts = self.send_and_expect(self.pg1, send_pkts, self.tun_if)
1632 self.verify_encrypted6(p_out, p_out.vpp_tun_sa, recv_pkts)
1634 for rx in recv_pkts:
1635 self.assertEqual(rx[IPv6].src, p_out.tun_src)
1636 self.assertEqual(rx[IPv6].dst, p_out.tun_dst)
1639 self.logger.info(self.vapi.ppcli("show error"))
1640 self.logger.info(self.vapi.ppcli("show ipsec all"))
1641 self.verify_counters6(p_in, p_out, count)
1643 def verify_tun_reass_66(self, p):
1644 self.vapi.cli("clear errors")
1645 self.vapi.ip_reassembly_enable_disable(
1646 sw_if_index=self.tun_if.sw_if_index, enable_ip6=True
1650 send_pkts = self.gen_encrypt_pkts6(
1654 src=p.remote_tun_if_host,
1655 dst=self.pg1.remote_ip6,
1659 send_pkts = fragment_rfc8200(send_pkts[0], 1, 1400, self.logger)
1660 recv_pkts = self.send_and_expect(self.tun_if, send_pkts, self.pg1, n_rx=1)
1661 self.verify_decrypted6(p, recv_pkts)
1663 send_pkts = self.gen_pkts6(
1666 src=self.pg1.remote_ip6,
1667 dst=p.remote_tun_if_host,
1671 recv_pkts = self.send_and_expect(self.pg1, send_pkts, self.tun_if)
1672 self.verify_encrypted6(p, p.vpp_tun_sa, recv_pkts)
1674 self.logger.info(self.vapi.ppcli("show error"))
1675 self.logger.info(self.vapi.ppcli("show ipsec all"))
1676 self.verify_counters6(p, p, 1)
1677 self.vapi.ip_reassembly_enable_disable(
1678 sw_if_index=self.tun_if.sw_if_index, enable_ip6=False
1681 def verify_tun_46(self, p, count=1):
1682 """ipsec 4o6 tunnel basic test"""
1683 self.vapi.cli("clear errors")
1684 self.vapi.cli("clear ipsec sa")
1686 send_pkts = self.gen_encrypt_pkts(
1690 src=p.remote_tun_if_host4,
1691 dst=self.pg1.remote_ip4,
1694 recv_pkts = self.send_and_expect(self.tun_if, send_pkts, self.pg1)
1695 for recv_pkt in recv_pkts:
1696 self.assert_equal(recv_pkt[IP].src, p.remote_tun_if_host4)
1697 self.assert_equal(recv_pkt[IP].dst, self.pg1.remote_ip4)
1698 self.assert_packet_checksums_valid(recv_pkt)
1699 send_pkts = self.gen_pkts(
1701 src=self.pg1.remote_ip4,
1702 dst=p.remote_tun_if_host4,
1705 recv_pkts = self.send_and_expect(self.pg1, send_pkts, self.tun_if)
1706 for recv_pkt in recv_pkts:
1708 decrypt_pkt = p.vpp_tun_sa.decrypt(recv_pkt[IPv6])
1709 if not decrypt_pkt.haslayer(IP):
1710 decrypt_pkt = IP(decrypt_pkt[Raw].load)
1711 self.assert_equal(decrypt_pkt.src, self.pg1.remote_ip4)
1712 self.assert_equal(decrypt_pkt.dst, p.remote_tun_if_host4)
1713 self.assert_packet_checksums_valid(decrypt_pkt)
1715 self.logger.debug(ppp("Unexpected packet:", recv_pkt))
1717 self.logger.debug(ppp("Decrypted packet:", decrypt_pkt))
1722 self.logger.info(self.vapi.ppcli("show error"))
1723 self.logger.info(self.vapi.ppcli("show ipsec all"))
1724 self.verify_counters6(p, p, count)
1726 def verify_keepalive(self, p):
1727 # the sizeof Raw is calculated to pad to the minimum ehternet
1728 # frame size of 64 btyes
1730 Ether(src=self.tun_if.remote_mac, dst=self.tun_if.local_mac)
1731 / IPv6(src=p.remote_tun_if_host, dst=self.tun_if.local_ip6)
1732 / UDP(sport=333, dport=4500)
1736 self.send_and_assert_no_replies(self.tun_if, pkt * 31)
1737 self.assert_error_counter_equal(
1738 "/err/%s/nat_keepalive" % self.tun6_input_node, 31
1742 Ether(src=self.tun_if.remote_mac, dst=self.tun_if.local_mac)
1743 / IPv6(src=p.remote_tun_if_host, dst=self.tun_if.local_ip6)
1744 / UDP(sport=333, dport=4500)
1747 self.send_and_assert_no_replies(self.tun_if, pkt * 31)
1748 self.assert_error_counter_equal("/err/%s/too_short" % self.tun6_input_node, 31)
1751 Ether(src=self.tun_if.remote_mac, dst=self.tun_if.local_mac)
1752 / IPv6(src=p.remote_tun_if_host, dst=self.tun_if.local_ip6)
1753 / UDP(sport=333, dport=4500)
1757 self.send_and_assert_no_replies(self.tun_if, pkt * 31)
1758 self.assert_error_counter_equal("/err/%s/too_short" % self.tun6_input_node, 62)
1761 class IpsecTun6Tests(IpsecTun6):
1762 """UT test methods for Tunnel v6"""
1764 def test_tun_basic66(self):
1765 """ipsec 6o6 tunnel basic test"""
1766 self.verify_tun_66(self.params[socket.AF_INET6], count=1)
1768 def test_tun_reass_basic66(self):
1769 """ipsec 6o6 tunnel basic reassembly test"""
1770 self.verify_tun_reass_66(self.params[socket.AF_INET6])
1772 def test_tun_burst66(self):
1773 """ipsec 6o6 tunnel burst test"""
1774 self.verify_tun_66(self.params[socket.AF_INET6], count=257)
1777 class IpsecTun6HandoffTests(IpsecTun6):
1778 """UT test methods for Tunnel v6 with multiple workers"""
1780 vpp_worker_count = 2
1782 def test_tun_handoff_66(self):
1783 """ipsec 6o6 tunnel worker hand-off test"""
1784 self.vapi.cli("clear errors")
1785 self.vapi.cli("clear ipsec sa")
1788 p = self.params[socket.AF_INET6]
1790 # inject alternately on worker 0 and 1. all counts on the SA
1791 # should be against worker 0
1792 for worker in [0, 1, 0, 1]:
1793 send_pkts = self.gen_encrypt_pkts6(
1797 src=p.remote_tun_if_host,
1798 dst=self.pg1.remote_ip6,
1801 recv_pkts = self.send_and_expect(
1802 self.tun_if, send_pkts, self.pg1, worker=worker
1804 self.verify_decrypted6(p, recv_pkts)
1806 send_pkts = self.gen_pkts6(
1809 src=self.pg1.remote_ip6,
1810 dst=p.remote_tun_if_host,
1813 recv_pkts = self.send_and_expect(
1814 self.pg1, send_pkts, self.tun_if, worker=worker
1816 self.verify_encrypted6(p, p.vpp_tun_sa, recv_pkts)
1818 # all counts against the first worker that was used
1819 self.verify_counters6(p, p, 4 * N_PKTS, worker=0)
1822 class IpsecTun4HandoffTests(IpsecTun4):
1823 """UT test methods for Tunnel v4 with multiple workers"""
1825 vpp_worker_count = 2
1827 def test_tun_handooff_44(self):
1828 """ipsec 4o4 tunnel worker hand-off test"""
1829 self.vapi.cli("clear errors")
1830 self.vapi.cli("clear ipsec sa")
1833 p = self.params[socket.AF_INET]
1835 # inject alternately on worker 0 and 1. all counts on the SA
1836 # should be against worker 0
1837 for worker in [0, 1, 0, 1]:
1838 send_pkts = self.gen_encrypt_pkts(
1842 src=p.remote_tun_if_host,
1843 dst=self.pg1.remote_ip4,
1846 recv_pkts = self.send_and_expect(
1847 self.tun_if, send_pkts, self.pg1, worker=worker
1849 self.verify_decrypted(p, recv_pkts)
1851 send_pkts = self.gen_pkts(
1853 src=self.pg1.remote_ip4,
1854 dst=p.remote_tun_if_host,
1857 recv_pkts = self.send_and_expect(
1858 self.pg1, send_pkts, self.tun_if, worker=worker
1860 self.verify_encrypted(p, p.vpp_tun_sa, recv_pkts)
1862 # all counts against the first worker that was used
1863 self.verify_counters4(p, 4 * N_PKTS, worker=0)
1866 class IpsecTun46Tests(IpsecTun4Tests, IpsecTun6Tests):
1867 """UT test methods for Tunnel v6 & v4"""
1872 class IPSecIPv4Fwd(VppTestCase):
1873 """Test IPSec by capturing and verifying IPv4 forwarded pkts"""
1876 def setUpConstants(cls):
1877 super(IPSecIPv4Fwd, cls).setUpConstants()
1880 super(IPSecIPv4Fwd, self).setUp()
1881 # store SPD objects so we can remove configs on tear down
1883 self.spd_policies = []
1886 # remove SPD policies
1887 for obj in self.spd_policies:
1888 obj.remove_vpp_config()
1889 self.spd_policies = []
1890 # remove SPD items (interface bindings first, then SPD)
1891 for obj in reversed(self.spd_objs):
1892 obj.remove_vpp_config()
1894 # close down pg intfs
1895 for pg in self.pg_interfaces:
1898 super(IPSecIPv4Fwd, self).tearDown()
1900 def create_interfaces(self, num_ifs=2):
1901 # create interfaces pg0 ... pg<num_ifs>
1902 self.create_pg_interfaces(range(num_ifs))
1903 for pg in self.pg_interfaces:
1904 # put the interface up
1906 # configure IPv4 address on the interface
1908 # resolve ARP, so that we know VPP MAC
1910 self.logger.info(self.vapi.ppcli("show int addr"))
1912 def spd_create_and_intf_add(self, spd_id, pg_list):
1913 spd = VppIpsecSpd(self, spd_id)
1914 spd.add_vpp_config()
1915 self.spd_objs.append(spd)
1917 spdItf = VppIpsecSpdItfBinding(self, spd, pg)
1918 spdItf.add_vpp_config()
1919 self.spd_objs.append(spdItf)
1921 def get_policy(self, policy_type):
1922 e = VppEnum.vl_api_ipsec_spd_action_t
1923 if policy_type == "protect":
1924 return e.IPSEC_API_SPD_ACTION_PROTECT
1925 elif policy_type == "bypass":
1926 return e.IPSEC_API_SPD_ACTION_BYPASS
1927 elif policy_type == "discard":
1928 return e.IPSEC_API_SPD_ACTION_DISCARD
1930 raise Exception("Invalid policy type: %s", policy_type)
1932 def spd_add_rem_policy(
1944 local_ip_start=ip_address("0.0.0.0"),
1945 local_ip_stop=ip_address("255.255.255.255"),
1946 remote_ip_start=ip_address("0.0.0.0"),
1947 remote_ip_stop=ip_address("255.255.255.255"),
1948 remote_port_start=0,
1949 remote_port_stop=65535,
1951 local_port_stop=65535,
1953 spd = VppIpsecSpd(self, spd_id)
1956 src_range_low = ip_address("0.0.0.0")
1957 src_range_high = ip_address("255.255.255.255")
1958 dst_range_low = ip_address("0.0.0.0")
1959 dst_range_high = ip_address("255.255.255.255")
1962 src_range_low = local_ip_start
1963 src_range_high = local_ip_stop
1964 dst_range_low = remote_ip_start
1965 dst_range_high = remote_ip_stop
1968 src_range_low = src_if.remote_ip4
1969 src_range_high = src_if.remote_ip4
1970 dst_range_low = dst_if.remote_ip4
1971 dst_range_high = dst_if.remote_ip4
1973 spdEntry = VppIpsecSpdEntry(
1983 policy=self.get_policy(policy_type),
1985 remote_port_start=remote_port_start,
1986 remote_port_stop=remote_port_stop,
1987 local_port_start=local_port_start,
1988 local_port_stop=local_port_stop,
1992 spdEntry.add_vpp_config()
1993 self.spd_policies.append(spdEntry)
1995 spdEntry.remove_vpp_config()
1996 self.spd_policies.remove(spdEntry)
1997 self.logger.info(self.vapi.ppcli("show ipsec all"))
2000 def create_stream(self, src_if, dst_if, pkt_count, src_prt=1234, dst_prt=5678):
2002 for i in range(pkt_count):
2003 # create packet info stored in the test case instance
2004 info = self.create_packet_info(src_if, dst_if)
2005 # convert the info into packet payload
2006 payload = self.info_to_payload(info)
2007 # create the packet itself
2009 Ether(dst=src_if.local_mac, src=src_if.remote_mac)
2010 / IP(src=src_if.remote_ip4, dst=dst_if.remote_ip4)
2011 / UDP(sport=src_prt, dport=dst_prt)
2014 # store a copy of the packet in the packet info
2015 info.data = p.copy()
2016 # append the packet to the list
2018 # return the created packet list
2021 def verify_capture(self, src_if, dst_if, capture):
2023 for packet in capture:
2027 # convert the payload to packet info object
2028 payload_info = self.payload_to_info(packet)
2029 # make sure the indexes match
2031 payload_info.src, src_if.sw_if_index, "source sw_if_index"
2034 payload_info.dst, dst_if.sw_if_index, "destination sw_if_index"
2036 packet_info = self.get_next_packet_info_for_interface2(
2037 src_if.sw_if_index, dst_if.sw_if_index, packet_info
2039 # make sure we didn't run out of saved packets
2040 self.assertIsNotNone(packet_info)
2042 payload_info.index, packet_info.index, "packet info index"
2044 saved_packet = packet_info.data # fetch the saved packet
2045 # assert the values match
2046 self.assert_equal(ip.src, saved_packet[IP].src, "IP source address")
2047 # ... more assertions here
2048 self.assert_equal(udp.sport, saved_packet[UDP].sport, "UDP source port")
2049 except Exception as e:
2050 self.logger.error(ppp("Unexpected or invalid packet:", packet))
2052 remaining_packet = self.get_next_packet_info_for_interface2(
2053 src_if.sw_if_index, dst_if.sw_if_index, packet_info
2057 "Interface %s: Packet expected from interface "
2058 "%s didn't arrive" % (dst_if.name, src_if.name),
2061 def verify_policy_match(self, pkt_count, spdEntry):
2062 self.logger.info("XXXX %s %s", str(spdEntry), str(spdEntry.get_stats()))
2063 matched_pkts = spdEntry.get_stats().get("packets")
2064 self.logger.info("Policy %s matched: %d pkts", str(spdEntry), matched_pkts)
2065 self.assert_equal(pkt_count, matched_pkts)
2068 class SpdFlowCacheTemplate(IPSecIPv4Fwd):
2070 def setUpConstants(cls):
2071 super(SpdFlowCacheTemplate, cls).setUpConstants()
2072 # Override this method with required cmdline parameters e.g.
2073 # cls.vpp_cmdline.extend(["ipsec", "{",
2074 # "ipv4-outbound-spd-flow-cache on",
2076 # cls.logger.info("VPP modified cmdline is %s" % " "
2077 # .join(cls.vpp_cmdline))
2080 super(SpdFlowCacheTemplate, self).setUp()
2083 super(SpdFlowCacheTemplate, self).tearDown()
2085 def get_spd_flow_cache_entries(self, outbound):
2086 """'show ipsec spd' output:
2087 ipv4-inbound-spd-flow-cache-entries: 0
2088 ipv4-outbound-spd-flow-cache-entries: 0
2090 show_ipsec_reply = self.vapi.cli("show ipsec spd")
2091 # match the relevant section of 'show ipsec spd' output
2093 regex_match = re.search(
2094 "ipv4-outbound-spd-flow-cache-entries: (.*)",
2099 regex_match = re.search(
2100 "ipv4-inbound-spd-flow-cache-entries: (.*)", show_ipsec_reply, re.DOTALL
2102 if regex_match is None:
2104 "Unable to find spd flow cache entries \
2105 in 'show ipsec spd' CLI output - regex failed to match"
2109 num_entries = int(regex_match.group(1))
2112 "Unable to get spd flow cache entries \
2113 from 'show ipsec spd' string: %s",
2114 regex_match.group(0),
2116 self.logger.info("%s", regex_match.group(0))
2119 def verify_num_outbound_flow_cache_entries(self, expected_elements):
2121 self.get_spd_flow_cache_entries(outbound=True), expected_elements
2124 def verify_num_inbound_flow_cache_entries(self, expected_elements):
2126 self.get_spd_flow_cache_entries(outbound=False), expected_elements
2129 def crc32_supported(self):
2130 # lscpu is part of util-linux package, available on all Linux Distros
2131 stream = os.popen("lscpu")
2132 cpu_info = stream.read()
2133 # feature/flag "crc32" on Aarch64 and "sse4_2" on x86
2134 # see vppinfra/crc32.h
2135 if "crc32" or "sse4_2" in cpu_info:
2136 self.logger.info("\ncrc32 supported:\n" + cpu_info)
2139 self.logger.info("\ncrc32 NOT supported:\n" + cpu_info)
2143 class IPSecIPv6Fwd(VppTestCase):
2144 """Test IPSec by capturing and verifying IPv6 forwarded pkts"""
2147 def setUpConstants(cls):
2148 super(IPSecIPv6Fwd, cls).setUpConstants()
2151 super(IPSecIPv6Fwd, self).setUp()
2152 # store SPD objects so we can remove configs on tear down
2154 self.spd_policies = []
2157 # remove SPD policies
2158 for obj in self.spd_policies:
2159 obj.remove_vpp_config()
2160 self.spd_policies = []
2161 # remove SPD items (interface bindings first, then SPD)
2162 for obj in reversed(self.spd_objs):
2163 obj.remove_vpp_config()
2165 # close down pg intfs
2166 for pg in self.pg_interfaces:
2169 super(IPSecIPv6Fwd, self).tearDown()
2171 def create_interfaces(self, num_ifs=2):
2172 # create interfaces pg0 ... pg<num_ifs>
2173 self.create_pg_interfaces(range(num_ifs))
2174 for pg in self.pg_interfaces:
2175 # put the interface up
2177 # configure IPv6 address on the interface
2180 self.logger.info(self.vapi.ppcli("show int addr"))
2182 def spd_create_and_intf_add(self, spd_id, pg_list):
2183 spd = VppIpsecSpd(self, spd_id)
2184 spd.add_vpp_config()
2185 self.spd_objs.append(spd)
2187 spdItf = VppIpsecSpdItfBinding(self, spd, pg)
2188 spdItf.add_vpp_config()
2189 self.spd_objs.append(spdItf)
2191 def get_policy(self, policy_type):
2192 e = VppEnum.vl_api_ipsec_spd_action_t
2193 if policy_type == "protect":
2194 return e.IPSEC_API_SPD_ACTION_PROTECT
2195 elif policy_type == "bypass":
2196 return e.IPSEC_API_SPD_ACTION_BYPASS
2197 elif policy_type == "discard":
2198 return e.IPSEC_API_SPD_ACTION_DISCARD
2200 raise Exception("Invalid policy type: %s", policy_type)
2202 def spd_add_rem_policy(
2214 local_ip_start=ip_address("0::0"),
2215 local_ip_stop=ip_address("ffff:ffff:ffff:ffff:ffff:ffff:ffff:ffff"),
2216 remote_ip_start=ip_address("0::0"),
2217 remote_ip_stop=ip_address("ffff:ffff:ffff:ffff:ffff:ffff:ffff:ffff"),
2218 remote_port_start=0,
2219 remote_port_stop=65535,
2221 local_port_stop=65535,
2223 spd = VppIpsecSpd(self, spd_id)
2226 src_range_low = ip_address("0::0")
2227 src_range_high = ip_address("ffff:ffff:ffff:ffff:ffff:ffff:ffff:ffff")
2228 dst_range_low = ip_address("0::0")
2229 dst_range_high = ip_address("ffff:ffff:ffff:ffff:ffff:ffff:ffff:ffff")
2232 src_range_low = local_ip_start
2233 src_range_high = local_ip_stop
2234 dst_range_low = remote_ip_start
2235 dst_range_high = remote_ip_stop
2238 src_range_low = src_if.remote_ip6
2239 src_range_high = src_if.remote_ip6
2240 dst_range_low = dst_if.remote_ip6
2241 dst_range_high = dst_if.remote_ip6
2243 spdEntry = VppIpsecSpdEntry(
2253 policy=self.get_policy(policy_type),
2255 remote_port_start=remote_port_start,
2256 remote_port_stop=remote_port_stop,
2257 local_port_start=local_port_start,
2258 local_port_stop=local_port_stop,
2262 spdEntry.add_vpp_config()
2263 self.spd_policies.append(spdEntry)
2265 spdEntry.remove_vpp_config()
2266 self.spd_policies.remove(spdEntry)
2267 self.logger.info(self.vapi.ppcli("show ipsec all"))
2270 def create_stream(self, src_if, dst_if, pkt_count, src_prt=1234, dst_prt=5678):
2272 for i in range(pkt_count):
2273 # create packet info stored in the test case instance
2274 info = self.create_packet_info(src_if, dst_if)
2275 # convert the info into packet payload
2276 payload = self.info_to_payload(info)
2277 # create the packet itself
2279 Ether(dst=src_if.local_mac, src=src_if.remote_mac)
2280 / IPv6(src=src_if.remote_ip6, dst=dst_if.remote_ip6)
2281 / UDP(sport=src_prt, dport=dst_prt)
2284 # store a copy of the packet in the packet info
2285 info.data = p.copy()
2286 # append the packet to the list
2288 # return the created packet list
2291 def verify_capture(self, src_if, dst_if, capture):
2293 for packet in capture:
2297 # convert the payload to packet info object
2298 payload_info = self.payload_to_info(packet)
2299 # make sure the indexes match
2301 payload_info.src, src_if.sw_if_index, "source sw_if_index"
2304 payload_info.dst, dst_if.sw_if_index, "destination sw_if_index"
2306 packet_info = self.get_next_packet_info_for_interface2(
2307 src_if.sw_if_index, dst_if.sw_if_index, packet_info
2309 # make sure we didn't run out of saved packets
2310 self.assertIsNotNone(packet_info)
2312 payload_info.index, packet_info.index, "packet info index"
2314 saved_packet = packet_info.data # fetch the saved packet
2315 # assert the values match
2316 self.assert_equal(ip.src, saved_packet[IPv6].src, "IP source address")
2317 # ... more assertions here
2318 self.assert_equal(udp.sport, saved_packet[UDP].sport, "UDP source port")
2319 except Exception as e:
2320 self.logger.error(ppp("Unexpected or invalid packet:", packet))
2322 remaining_packet = self.get_next_packet_info_for_interface2(
2323 src_if.sw_if_index, dst_if.sw_if_index, packet_info
2327 "Interface %s: Packet expected from interface "
2328 "%s didn't arrive" % (dst_if.name, src_if.name),
2331 def verify_policy_match(self, pkt_count, spdEntry):
2332 self.logger.info("XXXX %s %s", str(spdEntry), str(spdEntry.get_stats()))
2333 matched_pkts = spdEntry.get_stats().get("packets")
2334 self.logger.info("Policy %s matched: %d pkts", str(spdEntry), matched_pkts)
2335 self.assert_equal(pkt_count, matched_pkts)
2338 if __name__ == "__main__":
2339 unittest.main(testRunner=VppTestRunner)