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/SA replayed packet" % self.tra4_decrypt_node_name[0]
334 count = self.statistics.get_err_counter(replay_node_name)
337 replay_post_node_name = (
338 "/err/%s/SA replayed packet" % 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/ESP decryption failed"
348 % self.tra4_decrypt_node_name[p.async_mode]
351 hash_failed_node_name = (
352 "/err/%s/Integrity check failed"
353 % self.tra4_decrypt_node_name[p.async_mode]
355 count = self.statistics.get_err_counter(hash_failed_node_name)
358 count += self.statistics.get_err_counter("/err/crypto-dispatch/bad-hmac")
362 def verify_hi_seq_num(self):
363 p = self.params[socket.AF_INET]
364 saf = VppEnum.vl_api_ipsec_sad_flags_t
365 esn_on = p.vpp_tra_sa.esn_en
366 ar_on = p.flags & saf.IPSEC_API_SAD_FLAG_USE_ANTI_REPLAY
368 seq_cycle_node_name = (
369 "/err/%s/sequence number cycled (packet dropped)"
370 % self.tra4_encrypt_node_name
372 replay_count = self.get_replay_counts(p)
373 hash_failed_count = self.get_hash_failed_counts(p)
374 seq_cycle_count = self.statistics.get_err_counter(seq_cycle_node_name)
376 # a few packets so we get the rx seq number above the window size and
377 # thus can simulate a wrap with an out of window packet
380 Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
381 / p.scapy_tra_sa.encrypt(
382 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
386 for seq in range(63, 80)
388 recv_pkts = self.send_and_expect(self.tra_if, pkts, self.tra_if)
390 # these 4 packets will all choose seq-num 0 to decrpyt since none
391 # are out of window when first checked. however, once #200 has
392 # decrypted it will move the window to 200 and has #81 is out of
393 # window. this packet should be dropped.
396 Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
397 / p.scapy_tra_sa.encrypt(
398 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
403 Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
404 / p.scapy_tra_sa.encrypt(
405 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
410 Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
411 / p.scapy_tra_sa.encrypt(
412 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
417 Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
418 / p.scapy_tra_sa.encrypt(
419 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
425 # if anti-replay is off then we won't drop #81
426 n_rx = 3 if ar_on else 4
427 self.send_and_expect(self.tra_if, pkts, self.tra_if, n_rx=n_rx)
428 # this packet is one before the wrap
431 Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
432 / p.scapy_tra_sa.encrypt(
433 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
438 recv_pkts = self.send_and_expect(self.tra_if, pkts, self.tra_if)
440 # move the window over half way to a wrap
443 Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
444 / p.scapy_tra_sa.encrypt(
445 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
450 recv_pkts = self.send_and_expect(self.tra_if, pkts, self.tra_if)
452 # anti-replay will drop old packets, no anti-replay will not
455 Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
456 / p.scapy_tra_sa.encrypt(
457 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
464 self.send_and_assert_no_replies(self.tra_if, pkts)
466 recv_pkts = self.send_and_expect(self.tra_if, pkts, self.tra_if)
470 # validate wrapping the ESN
473 # wrap scapy's TX SA SN
474 p.scapy_tra_sa.seq_num = 0x100000005
476 # send a packet that wraps the window for both AR and no AR
479 Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
480 / p.scapy_tra_sa.encrypt(
481 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4)
488 rxs = self.send_and_expect(self.tra_if, pkts, self.tra_if)
490 decrypted = p.vpp_tra_sa.decrypt(rx[0][IP])
492 # move the window forward to half way to the next wrap
495 Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
496 / p.scapy_tra_sa.encrypt(
497 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4)
504 rxs = self.send_and_expect(self.tra_if, pkts, self.tra_if)
506 # a packet less than 2^30 from the current position is:
507 # - AR: out of window and dropped
511 Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
512 / p.scapy_tra_sa.encrypt(
513 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4)
521 self.send_and_assert_no_replies(self.tra_if, pkts)
523 self.send_and_expect(self.tra_if, pkts, self.tra_if)
525 # a packet more than 2^30 from the current position is:
526 # - AR: out of window and dropped
527 # - non-AR: considered a wrap, but since it's not a wrap
528 # it won't decrpyt and so will be dropped
531 Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
532 / p.scapy_tra_sa.encrypt(
533 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4)
540 self.send_and_assert_no_replies(self.tra_if, pkts)
542 # a packet less than 2^30 from the current position and is a
543 # wrap; (the seq is currently at 0x180000005).
544 # - AR: out of window so considered a wrap, so accepted
545 # - non-AR: not considered a wrap, so won't decrypt
546 p.scapy_tra_sa.seq_num = 0x260000005
549 Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
550 / p.scapy_tra_sa.encrypt(
551 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4)
558 self.send_and_expect(self.tra_if, pkts, self.tra_if)
560 self.send_and_assert_no_replies(self.tra_if, pkts)
563 # window positions are different now for AR/non-AR
564 # move non-AR forward
567 # a packet more than 2^30 from the current position and is a
568 # wrap; (the seq is currently at 0x180000005).
570 # - non-AR: not considered a wrap, so won't decrypt
574 Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
575 / p.scapy_tra_sa.encrypt(
576 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4)
582 Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
583 / p.scapy_tra_sa.encrypt(
584 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4)
590 self.send_and_expect(self.tra_if, pkts, self.tra_if)
594 Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
595 / p.scapy_tra_sa.encrypt(
596 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4)
602 self.send_and_expect(self.tra_if, pkts, self.tra_if)
604 def verify_tra_anti_replay(self):
605 p = self.params[socket.AF_INET]
606 esn_en = p.vpp_tra_sa.esn_en
608 seq_cycle_node_name = (
609 "/err/%s/sequence number cycled (packet dropped)"
610 % self.tra4_encrypt_node_name
612 replay_count = self.get_replay_counts(p)
613 hash_failed_count = self.get_hash_failed_counts(p)
614 seq_cycle_count = self.statistics.get_err_counter(seq_cycle_node_name)
616 if ESP == self.encryption_type:
617 undersize_node_name = (
618 "/err/%s/undersized packet" % self.tra4_decrypt_node_name[0]
620 undersize_count = self.statistics.get_err_counter(undersize_node_name)
623 # send packets with seq numbers 1->34
624 # this means the window size is still in Case B (see RFC4303
627 # for reasons i haven't investigated Scapy won't create a packet with
632 Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
633 / p.scapy_tra_sa.encrypt(
634 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
638 for seq in range(1, 34)
640 recv_pkts = self.send_and_expect(self.tra_if, pkts, self.tra_if)
642 # replayed packets are dropped
643 self.send_and_assert_no_replies(self.tra_if, pkts, timeout=0.2)
644 replay_count += len(pkts)
645 self.assertEqual(self.get_replay_counts(p), replay_count)
648 # now send a batch of packets all with the same sequence number
649 # the first packet in the batch is legitimate, the rest bogus
651 self.vapi.cli("clear error")
652 self.vapi.cli("clear node counters")
654 src=self.tra_if.remote_mac, dst=self.tra_if.local_mac
655 ) / p.scapy_tra_sa.encrypt(
656 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
659 recv_pkts = self.send_and_expect(self.tra_if, pkts * 8, self.tra_if, n_rx=1)
661 self.assertEqual(self.get_replay_counts(p), replay_count)
664 # now move the window over to 257 (more than one byte) and into Case A
666 self.vapi.cli("clear error")
668 src=self.tra_if.remote_mac, dst=self.tra_if.local_mac
669 ) / p.scapy_tra_sa.encrypt(
670 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
673 recv_pkts = self.send_and_expect(self.tra_if, [pkt], self.tra_if)
675 # replayed packets are dropped
676 self.send_and_assert_no_replies(self.tra_if, pkt * 3, timeout=0.2)
678 self.assertEqual(self.get_replay_counts(p), replay_count)
680 # the window size is 64 packets
681 # in window are still accepted
683 src=self.tra_if.remote_mac, dst=self.tra_if.local_mac
684 ) / p.scapy_tra_sa.encrypt(
685 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
688 recv_pkts = self.send_and_expect(self.tra_if, [pkt], self.tra_if)
690 # a packet that does not decrypt does not move the window forward
691 bogus_sa = SecurityAssociation(
692 self.encryption_type,
694 crypt_algo=p.crypt_algo,
695 crypt_key=mk_scapy_crypt_key(p)[::-1],
696 auth_algo=p.auth_algo,
697 auth_key=p.auth_key[::-1],
700 src=self.tra_if.remote_mac, dst=self.tra_if.local_mac
701 ) / bogus_sa.encrypt(
702 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
705 self.send_and_assert_no_replies(self.tra_if, pkt * 17, timeout=0.2)
707 hash_failed_count += 17
708 self.assertEqual(self.get_hash_failed_counts(p), hash_failed_count)
710 # a malformed 'runt' packet
711 # created by a mis-constructed SA
712 if ESP == self.encryption_type and p.crypt_algo != "NULL":
713 bogus_sa = SecurityAssociation(self.encryption_type, p.vpp_tra_spi)
715 src=self.tra_if.remote_mac, dst=self.tra_if.local_mac
716 ) / bogus_sa.encrypt(
717 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
720 self.send_and_assert_no_replies(self.tra_if, pkt * 17, timeout=0.2)
722 undersize_count += 17
723 self.assert_error_counter_equal(undersize_node_name, undersize_count)
725 # which we can determine since this packet is still in the window
727 src=self.tra_if.remote_mac, dst=self.tra_if.local_mac
728 ) / p.scapy_tra_sa.encrypt(
729 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
732 self.send_and_expect(self.tra_if, [pkt], self.tra_if)
735 # out of window are dropped
736 # this is Case B. So VPP will consider this to be a high seq num wrap
737 # and so the decrypt attempt will fail
740 src=self.tra_if.remote_mac, dst=self.tra_if.local_mac
741 ) / p.scapy_tra_sa.encrypt(
742 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
745 self.send_and_assert_no_replies(self.tra_if, pkt * 17, timeout=0.2)
748 # an out of window error with ESN looks like a high sequence
749 # wrap. but since it isn't then the verify will fail.
750 hash_failed_count += 17
751 self.assertEqual(self.get_hash_failed_counts(p), hash_failed_count)
755 self.assertEqual(self.get_replay_counts(p), replay_count)
757 # valid packet moves the window over to 258
759 src=self.tra_if.remote_mac, dst=self.tra_if.local_mac
760 ) / p.scapy_tra_sa.encrypt(
761 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
764 rx = self.send_and_expect(self.tra_if, [pkt], self.tra_if)
765 decrypted = p.vpp_tra_sa.decrypt(rx[0][IP])
768 # move VPP's SA TX seq-num to just before the seq-number wrap.
769 # then fire in a packet that VPP should drop on TX because it
770 # causes the TX seq number to wrap; unless we're using extened sequence
773 self.vapi.cli("test ipsec sa %d seq 0xffffffff" % p.scapy_tra_sa_id)
774 self.logger.info(self.vapi.ppcli("show ipsec sa 0"))
775 self.logger.info(self.vapi.ppcli("show ipsec sa 1"))
779 Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
780 / p.scapy_tra_sa.encrypt(
781 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
785 for seq in range(259, 280)
789 rxs = self.send_and_expect(self.tra_if, pkts, self.tra_if)
792 # in order for scapy to decrypt its SA's high order number needs
795 p.vpp_tra_sa.seq_num = 0x100000000
797 decrypted = p.vpp_tra_sa.decrypt(rx[0][IP])
800 # wrap scapy's TX high sequence number. VPP is in case B, so it
801 # will consider this a high seq wrap also.
802 # The low seq num we set it to will place VPP's RX window in Case A
804 p.scapy_tra_sa.seq_num = 0x100000005
806 src=self.tra_if.remote_mac, dst=self.tra_if.local_mac
807 ) / p.scapy_tra_sa.encrypt(
808 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
811 rx = self.send_and_expect(self.tra_if, [pkt], self.tra_if)
813 decrypted = p.vpp_tra_sa.decrypt(rx[0][IP])
816 # A packet that has seq num between (2^32-64) and 5 is within
819 p.scapy_tra_sa.seq_num = 0xFFFFFFFD
821 src=self.tra_if.remote_mac, dst=self.tra_if.local_mac
822 ) / p.scapy_tra_sa.encrypt(
823 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
826 rx = self.send_and_expect(self.tra_if, [pkt], self.tra_if)
827 decrypted = p.vpp_tra_sa.decrypt(rx[0][IP])
830 # While in case A we cannot wrap the high sequence number again
831 # because VPP will consider this packet to be one that moves the
835 src=self.tra_if.remote_mac, dst=self.tra_if.local_mac
836 ) / p.scapy_tra_sa.encrypt(
837 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
840 self.send_and_assert_no_replies(
841 self.tra_if, [pkt], self.tra_if, timeout=0.2
844 hash_failed_count += 1
845 self.assertEqual(self.get_hash_failed_counts(p), hash_failed_count)
848 # but if we move the window forward to case B, then we can wrap
851 p.scapy_tra_sa.seq_num = 0x100000555
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 rx = self.send_and_expect(self.tra_if, [pkt], self.tra_if)
859 decrypted = p.vpp_tra_sa.decrypt(rx[0][IP])
861 p.scapy_tra_sa.seq_num = 0x200000444
863 src=self.tra_if.remote_mac, dst=self.tra_if.local_mac
864 ) / p.scapy_tra_sa.encrypt(
865 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
868 rx = self.send_and_expect(self.tra_if, [pkt], self.tra_if)
869 decrypted = p.vpp_tra_sa.decrypt(rx[0][IP])
873 # without ESN TX sequence numbers can't wrap and packets are
874 # dropped from here on out.
876 self.send_and_assert_no_replies(self.tra_if, pkts, timeout=0.2)
877 seq_cycle_count += len(pkts)
878 self.assert_error_counter_equal(seq_cycle_node_name, seq_cycle_count)
880 # move the security-associations seq number on to the last we used
881 self.vapi.cli("test ipsec sa %d seq 0x15f" % p.scapy_tra_sa_id)
882 p.scapy_tra_sa.seq_num = 351
883 p.vpp_tra_sa.seq_num = 351
885 def verify_tra_lost(self):
886 p = self.params[socket.AF_INET]
887 esn_en = p.vpp_tra_sa.esn_en
890 # send packets with seq numbers 1->34
891 # this means the window size is still in Case B (see RFC4303
894 # for reasons i haven't investigated Scapy won't create a packet with
899 Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
900 / p.scapy_tra_sa.encrypt(
901 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
905 for seq in range(1, 3)
907 self.send_and_expect(self.tra_if, pkts, self.tra_if)
909 self.assertEqual(p.tra_sa_out.get_lost(), 0)
911 # skip a sequence number
914 Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
915 / p.scapy_tra_sa.encrypt(
916 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
920 for seq in range(4, 6)
922 self.send_and_expect(self.tra_if, pkts, self.tra_if)
924 self.assertEqual(p.tra_sa_out.get_lost(), 0)
926 # the lost packet are counted untill we get up past the first
927 # sizeof(replay_window) packets
930 Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
931 / p.scapy_tra_sa.encrypt(
932 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
936 for seq in range(6, 100)
938 self.send_and_expect(self.tra_if, pkts, self.tra_if)
940 self.assertEqual(p.tra_sa_out.get_lost(), 1)
942 # lost of holes in the sequence
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(100, 200, 2)
953 self.send_and_expect(self.tra_if, pkts, self.tra_if, n_rx=50)
957 Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
958 / p.scapy_tra_sa.encrypt(
959 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
963 for seq in range(200, 300)
965 self.send_and_expect(self.tra_if, pkts, self.tra_if)
967 self.assertEqual(p.tra_sa_out.get_lost(), 51)
969 # a big hole in the seq number space
972 Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
973 / p.scapy_tra_sa.encrypt(
974 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
978 for seq in range(400, 500)
980 self.send_and_expect(self.tra_if, pkts, self.tra_if)
982 self.assertEqual(p.tra_sa_out.get_lost(), 151)
984 def verify_tra_basic4(self, count=1, payload_size=54):
985 """ipsec v4 transport basic test"""
986 self.vapi.cli("clear errors")
987 self.vapi.cli("clear ipsec sa")
989 p = self.params[socket.AF_INET]
990 send_pkts = self.gen_encrypt_pkts(
994 src=self.tra_if.remote_ip4,
995 dst=self.tra_if.local_ip4,
997 payload_size=payload_size,
999 recv_pkts = self.send_and_expect(self.tra_if, send_pkts, self.tra_if)
1000 for rx in recv_pkts:
1001 self.assertEqual(len(rx) - len(Ether()), rx[IP].len)
1002 self.assert_packet_checksums_valid(rx)
1004 decrypted = p.vpp_tra_sa.decrypt(rx[IP])
1005 self.assert_packet_checksums_valid(decrypted)
1007 self.logger.debug(ppp("Unexpected packet:", rx))
1010 self.logger.info(self.vapi.ppcli("show error"))
1011 self.logger.info(self.vapi.ppcli("show ipsec all"))
1013 pkts = p.tra_sa_in.get_stats()["packets"]
1015 pkts, count, "incorrect SA in counts: expected %d != %d" % (count, pkts)
1017 pkts = p.tra_sa_out.get_stats()["packets"]
1019 pkts, count, "incorrect SA out counts: expected %d != %d" % (count, pkts)
1021 self.assertEqual(p.tra_sa_out.get_lost(), 0)
1022 self.assertEqual(p.tra_sa_in.get_lost(), 0)
1024 self.assert_packet_counter_equal(self.tra4_encrypt_node_name, count)
1025 self.assert_packet_counter_equal(self.tra4_decrypt_node_name[0], count)
1028 class IpsecTra4Tests(IpsecTra4):
1029 """UT test methods for Transport v4"""
1031 def test_tra_anti_replay(self):
1032 """ipsec v4 transport anti-replay test"""
1033 self.verify_tra_anti_replay()
1035 def test_tra_lost(self):
1036 """ipsec v4 transport lost packet test"""
1037 self.verify_tra_lost()
1039 def test_tra_basic(self, count=1):
1040 """ipsec v4 transport basic test"""
1041 self.verify_tra_basic4(count=1)
1043 def test_tra_burst(self):
1044 """ipsec v4 transport burst test"""
1045 self.verify_tra_basic4(count=257)
1048 class IpsecTra6(object):
1049 """verify methods for Transport v6"""
1051 def verify_tra_basic6(self, count=1, payload_size=54):
1052 self.vapi.cli("clear errors")
1053 self.vapi.cli("clear ipsec sa")
1055 p = self.params[socket.AF_INET6]
1056 send_pkts = self.gen_encrypt_pkts6(
1060 src=self.tra_if.remote_ip6,
1061 dst=self.tra_if.local_ip6,
1063 payload_size=payload_size,
1065 recv_pkts = self.send_and_expect(self.tra_if, send_pkts, self.tra_if)
1066 for rx in recv_pkts:
1067 self.assertEqual(len(rx) - len(Ether()) - len(IPv6()), rx[IPv6].plen)
1069 decrypted = p.vpp_tra_sa.decrypt(rx[IPv6])
1070 self.assert_packet_checksums_valid(decrypted)
1072 self.logger.debug(ppp("Unexpected packet:", rx))
1075 self.logger.info(self.vapi.ppcli("show error"))
1076 self.logger.info(self.vapi.ppcli("show ipsec all"))
1078 pkts = p.tra_sa_in.get_stats()["packets"]
1080 pkts, count, "incorrect SA in counts: expected %d != %d" % (count, pkts)
1082 pkts = p.tra_sa_out.get_stats()["packets"]
1084 pkts, count, "incorrect SA out counts: expected %d != %d" % (count, pkts)
1086 self.assert_packet_counter_equal(self.tra6_encrypt_node_name, count)
1087 self.assert_packet_counter_equal(self.tra6_decrypt_node_name[0], count)
1089 def gen_encrypt_pkts_ext_hdrs6(
1090 self, sa, sw_intf, src, dst, count=1, payload_size=54
1093 Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac)
1095 IPv6(src=src, dst=dst)
1096 / ICMPv6EchoRequest(id=0, seq=1, data="X" * payload_size)
1098 for i in range(count)
1101 def gen_pkts_ext_hdrs6(self, sw_intf, src, dst, count=1, payload_size=54):
1103 Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac)
1104 / IPv6(src=src, dst=dst)
1105 / IPv6ExtHdrHopByHop()
1106 / IPv6ExtHdrFragment(id=2, offset=200)
1107 / Raw(b"\xff" * 200)
1108 for i in range(count)
1111 def verify_tra_encrypted6(self, p, sa, rxs):
1114 self.assert_packet_checksums_valid(rx)
1116 decrypt_pkt = p.vpp_tra_sa.decrypt(rx[IPv6])
1117 decrypted.append(decrypt_pkt)
1118 self.assert_equal(decrypt_pkt.src, self.tra_if.local_ip6)
1119 self.assert_equal(decrypt_pkt.dst, self.tra_if.remote_ip6)
1121 self.logger.debug(ppp("Unexpected packet:", rx))
1123 self.logger.debug(ppp("Decrypted packet:", decrypt_pkt))
1129 def verify_tra_66_ext_hdrs(self, p):
1133 # check we can decrypt with options
1135 tx = self.gen_encrypt_pkts_ext_hdrs6(
1138 src=self.tra_if.remote_ip6,
1139 dst=self.tra_if.local_ip6,
1142 self.send_and_expect(self.tra_if, tx, self.tra_if)
1145 # injecting a packet from ourselves to be routed of box is a hack
1146 # but it matches an outbout policy, alors je ne regrette rien
1149 # one extension before ESP
1151 Ether(src=self.pg2.remote_mac, dst=self.pg2.local_mac)
1152 / IPv6(src=self.tra_if.local_ip6, dst=self.tra_if.remote_ip6)
1153 / IPv6ExtHdrFragment(id=2, offset=200)
1154 / Raw(b"\xff" * 200)
1157 rxs = self.send_and_expect(self.pg2, [tx], self.tra_if)
1158 dcs = self.verify_tra_encrypted6(p, p.vpp_tra_sa, rxs)
1161 # for reasons i'm not going to investigate scapy does not
1162 # created the correct headers after decrypt. but reparsing
1163 # the ipv6 packet fixes it
1164 dc = IPv6(raw(dc[IPv6]))
1165 self.assert_equal(dc[IPv6ExtHdrFragment].id, 2)
1167 # two extensions 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 / IPv6ExtHdrHopByHop()
1172 / IPv6ExtHdrFragment(id=2, offset=200)
1173 / Raw(b"\xff" * 200)
1176 rxs = self.send_and_expect(self.pg2, [tx], self.tra_if)
1177 dcs = self.verify_tra_encrypted6(p, p.vpp_tra_sa, rxs)
1180 dc = IPv6(raw(dc[IPv6]))
1181 self.assertTrue(dc[IPv6ExtHdrHopByHop])
1182 self.assert_equal(dc[IPv6ExtHdrFragment].id, 2)
1184 # two extensions before ESP, one after
1186 Ether(src=self.pg2.remote_mac, dst=self.pg2.local_mac)
1187 / IPv6(src=self.tra_if.local_ip6, dst=self.tra_if.remote_ip6)
1188 / IPv6ExtHdrHopByHop()
1189 / IPv6ExtHdrFragment(id=2, offset=200)
1190 / IPv6ExtHdrDestOpt()
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[IPv6ExtHdrDestOpt])
1200 self.assertTrue(dc[IPv6ExtHdrHopByHop])
1201 self.assert_equal(dc[IPv6ExtHdrFragment].id, 2)
1204 class IpsecTra6Tests(IpsecTra6):
1205 """UT test methods for Transport v6"""
1207 def test_tra_basic6(self):
1208 """ipsec v6 transport basic test"""
1209 self.verify_tra_basic6(count=1)
1211 def test_tra_burst6(self):
1212 """ipsec v6 transport burst test"""
1213 self.verify_tra_basic6(count=257)
1216 class IpsecTra6ExtTests(IpsecTra6):
1217 def test_tra_ext_hdrs_66(self):
1218 """ipsec 6o6 tra extension headers test"""
1219 self.verify_tra_66_ext_hdrs(self.params[socket.AF_INET6])
1222 class IpsecTra46Tests(IpsecTra4Tests, IpsecTra6Tests):
1223 """UT test methods for Transport v6 and v4"""
1228 class IpsecTun4(object):
1229 """verify methods for Tunnel v4"""
1231 def verify_counters4(self, p, count, n_frags=None, worker=None):
1234 if hasattr(p, "spd_policy_in_any"):
1235 pkts = p.spd_policy_in_any.get_stats(worker)["packets"]
1239 "incorrect SPD any policy: expected %d != %d" % (count, pkts),
1242 if hasattr(p, "tun_sa_in"):
1243 pkts = p.tun_sa_in.get_stats(worker)["packets"]
1245 pkts, count, "incorrect SA in counts: expected %d != %d" % (count, pkts)
1247 pkts = p.tun_sa_out.get_stats(worker)["packets"]
1251 "incorrect SA out counts: expected %d != %d" % (count, pkts),
1254 self.assert_packet_counter_equal(self.tun4_encrypt_node_name, n_frags)
1255 self.assert_packet_counter_equal(self.tun4_decrypt_node_name[0], count)
1257 def verify_decrypted(self, p, rxs):
1259 self.assert_equal(rx[IP].src, p.remote_tun_if_host)
1260 self.assert_equal(rx[IP].dst, self.pg1.remote_ip4)
1261 self.assert_packet_checksums_valid(rx)
1263 def verify_esp_padding(self, sa, esp_payload, decrypt_pkt):
1264 align = sa.crypt_algo.block_size
1267 exp_len = (len(decrypt_pkt) + 2 + (align - 1)) & ~(align - 1)
1268 exp_len += sa.crypt_algo.iv_size
1269 exp_len += sa.crypt_algo.icv_size or sa.auth_algo.icv_size
1270 self.assertEqual(exp_len, len(esp_payload))
1272 def verify_encrypted(self, p, sa, rxs):
1276 self.assertEqual(rx[UDP].dport, 4500)
1277 self.assert_packet_checksums_valid(rx)
1278 self.assertEqual(len(rx) - len(Ether()), rx[IP].len)
1281 decrypt_pkt = p.vpp_tun_sa.decrypt(rx_ip)
1282 if not decrypt_pkt.haslayer(IP):
1283 decrypt_pkt = IP(decrypt_pkt[Raw].load)
1284 if rx_ip.proto == socket.IPPROTO_ESP:
1285 self.verify_esp_padding(sa, rx_ip[ESP].data, decrypt_pkt)
1286 decrypt_pkts.append(decrypt_pkt)
1287 self.assert_equal(decrypt_pkt.src, self.pg1.remote_ip4)
1288 self.assert_equal(decrypt_pkt.dst, p.remote_tun_if_host)
1290 self.logger.debug(ppp("Unexpected packet:", rx))
1292 self.logger.debug(ppp("Decrypted packet:", decrypt_pkt))
1296 pkts = reassemble4(decrypt_pkts)
1298 self.assert_packet_checksums_valid(pkt)
1300 def verify_tun_44(self, p, count=1, payload_size=64, n_rx=None):
1301 self.vapi.cli("clear errors")
1302 self.vapi.cli("clear ipsec counters")
1303 self.vapi.cli("clear ipsec sa")
1307 send_pkts = self.gen_encrypt_pkts(
1311 src=p.remote_tun_if_host,
1312 dst=self.pg1.remote_ip4,
1314 payload_size=payload_size,
1316 recv_pkts = self.send_and_expect(self.tun_if, send_pkts, self.pg1)
1317 self.verify_decrypted(p, recv_pkts)
1319 send_pkts = self.gen_pkts(
1321 src=self.pg1.remote_ip4,
1322 dst=p.remote_tun_if_host,
1324 payload_size=payload_size,
1326 recv_pkts = self.send_and_expect(self.pg1, send_pkts, self.tun_if, n_rx)
1327 self.verify_encrypted(p, p.vpp_tun_sa, recv_pkts)
1329 for rx in recv_pkts:
1330 self.assertEqual(rx[IP].src, p.tun_src)
1331 self.assertEqual(rx[IP].dst, p.tun_dst)
1334 self.logger.info(self.vapi.ppcli("show error"))
1335 self.logger.info(self.vapi.ppcli("show ipsec all"))
1337 self.logger.info(self.vapi.ppcli("show ipsec sa 0"))
1338 self.logger.info(self.vapi.ppcli("show ipsec sa 4"))
1339 self.verify_counters4(p, count, n_rx)
1341 def verify_tun_dropped_44(self, p, count=1, payload_size=64, n_rx=None):
1342 self.vapi.cli("clear errors")
1346 send_pkts = self.gen_encrypt_pkts(
1350 src=p.remote_tun_if_host,
1351 dst=self.pg1.remote_ip4,
1354 self.send_and_assert_no_replies(self.tun_if, send_pkts)
1356 send_pkts = self.gen_pkts(
1358 src=self.pg1.remote_ip4,
1359 dst=p.remote_tun_if_host,
1361 payload_size=payload_size,
1363 self.send_and_assert_no_replies(self.pg1, send_pkts)
1366 self.logger.info(self.vapi.ppcli("show error"))
1367 self.logger.info(self.vapi.ppcli("show ipsec all"))
1369 def verify_tun_reass_44(self, p):
1370 self.vapi.cli("clear errors")
1371 self.vapi.ip_reassembly_enable_disable(
1372 sw_if_index=self.tun_if.sw_if_index, enable_ip4=True
1376 send_pkts = self.gen_encrypt_pkts(
1380 src=p.remote_tun_if_host,
1381 dst=self.pg1.remote_ip4,
1385 send_pkts = fragment_rfc791(send_pkts[0], 1400)
1386 recv_pkts = self.send_and_expect(self.tun_if, send_pkts, self.pg1, n_rx=1)
1387 self.verify_decrypted(p, recv_pkts)
1389 send_pkts = self.gen_pkts(
1390 self.pg1, src=self.pg1.remote_ip4, dst=p.remote_tun_if_host, count=1
1392 recv_pkts = self.send_and_expect(self.pg1, send_pkts, self.tun_if)
1393 self.verify_encrypted(p, p.vpp_tun_sa, recv_pkts)
1396 self.logger.info(self.vapi.ppcli("show error"))
1397 self.logger.info(self.vapi.ppcli("show ipsec all"))
1399 self.verify_counters4(p, 1, 1)
1400 self.vapi.ip_reassembly_enable_disable(
1401 sw_if_index=self.tun_if.sw_if_index, enable_ip4=False
1404 def verify_tun_64(self, p, count=1):
1405 self.vapi.cli("clear errors")
1406 self.vapi.cli("clear ipsec sa")
1408 send_pkts = self.gen_encrypt_pkts6(
1412 src=p.remote_tun_if_host6,
1413 dst=self.pg1.remote_ip6,
1416 recv_pkts = self.send_and_expect(self.tun_if, send_pkts, self.pg1)
1417 for recv_pkt in recv_pkts:
1418 self.assert_equal(recv_pkt[IPv6].src, p.remote_tun_if_host6)
1419 self.assert_equal(recv_pkt[IPv6].dst, self.pg1.remote_ip6)
1420 self.assert_packet_checksums_valid(recv_pkt)
1421 send_pkts = self.gen_pkts6(
1424 src=self.pg1.remote_ip6,
1425 dst=p.remote_tun_if_host6,
1428 recv_pkts = self.send_and_expect(self.pg1, send_pkts, self.tun_if)
1429 for recv_pkt in recv_pkts:
1431 decrypt_pkt = p.vpp_tun_sa.decrypt(recv_pkt[IP])
1432 if not decrypt_pkt.haslayer(IPv6):
1433 decrypt_pkt = IPv6(decrypt_pkt[Raw].load)
1434 self.assert_equal(decrypt_pkt.src, self.pg1.remote_ip6)
1435 self.assert_equal(decrypt_pkt.dst, p.remote_tun_if_host6)
1436 self.assert_packet_checksums_valid(decrypt_pkt)
1438 self.logger.error(ppp("Unexpected packet:", recv_pkt))
1440 self.logger.debug(ppp("Decrypted packet:", decrypt_pkt))
1445 self.logger.info(self.vapi.ppcli("show error"))
1446 self.logger.info(self.vapi.ppcli("show ipsec all"))
1448 self.verify_counters4(p, count)
1450 def verify_keepalive(self, p):
1451 # the sizeof Raw is calculated to pad to the minimum ehternet
1452 # frame size of 64 btyes
1454 Ether(src=self.tun_if.remote_mac, dst=self.tun_if.local_mac)
1455 / IP(src=p.remote_tun_if_host, dst=self.tun_if.local_ip4)
1456 / UDP(sport=333, dport=4500)
1460 self.send_and_assert_no_replies(self.tun_if, pkt * 31)
1461 self.assert_error_counter_equal(
1462 "/err/%s/NAT Keepalive" % self.tun4_input_node, 31
1466 Ether(src=self.tun_if.remote_mac, dst=self.tun_if.local_mac)
1467 / IP(src=p.remote_tun_if_host, dst=self.tun_if.local_ip4)
1468 / UDP(sport=333, dport=4500)
1471 self.send_and_assert_no_replies(self.tun_if, pkt * 31)
1472 self.assert_error_counter_equal("/err/%s/Too Short" % self.tun4_input_node, 31)
1475 Ether(src=self.tun_if.remote_mac, dst=self.tun_if.local_mac)
1476 / IP(src=p.remote_tun_if_host, dst=self.tun_if.local_ip4)
1477 / UDP(sport=333, dport=4500)
1481 self.send_and_assert_no_replies(self.tun_if, pkt * 31)
1482 self.assert_error_counter_equal("/err/%s/Too Short" % self.tun4_input_node, 62)
1485 class IpsecTun4Tests(IpsecTun4):
1486 """UT test methods for Tunnel v4"""
1488 def test_tun_basic44(self):
1489 """ipsec 4o4 tunnel basic test"""
1490 self.verify_tun_44(self.params[socket.AF_INET], count=1)
1491 self.tun_if.admin_down()
1492 self.tun_if.resolve_arp()
1493 self.tun_if.admin_up()
1494 self.verify_tun_44(self.params[socket.AF_INET], count=1)
1496 def test_tun_reass_basic44(self):
1497 """ipsec 4o4 tunnel basic reassembly test"""
1498 self.verify_tun_reass_44(self.params[socket.AF_INET])
1500 def test_tun_burst44(self):
1501 """ipsec 4o4 tunnel burst test"""
1502 self.verify_tun_44(self.params[socket.AF_INET], count=127)
1505 class IpsecTun6(object):
1506 """verify methods for Tunnel v6"""
1508 def verify_counters6(self, p_in, p_out, count, worker=None):
1509 if hasattr(p_in, "tun_sa_in"):
1510 pkts = p_in.tun_sa_in.get_stats(worker)["packets"]
1512 pkts, count, "incorrect SA in counts: expected %d != %d" % (count, pkts)
1514 if hasattr(p_out, "tun_sa_out"):
1515 pkts = p_out.tun_sa_out.get_stats(worker)["packets"]
1519 "incorrect SA out counts: expected %d != %d" % (count, pkts),
1521 self.assert_packet_counter_equal(self.tun6_encrypt_node_name, count)
1522 self.assert_packet_counter_equal(self.tun6_decrypt_node_name[0], count)
1524 def verify_decrypted6(self, p, rxs):
1526 self.assert_equal(rx[IPv6].src, p.remote_tun_if_host)
1527 self.assert_equal(rx[IPv6].dst, self.pg1.remote_ip6)
1528 self.assert_packet_checksums_valid(rx)
1530 def verify_encrypted6(self, p, sa, rxs):
1532 self.assert_packet_checksums_valid(rx)
1533 self.assertEqual(len(rx) - len(Ether()) - len(IPv6()), rx[IPv6].plen)
1534 self.assert_equal(rx[IPv6].hlim, p.outer_hop_limit)
1535 if p.outer_flow_label:
1536 self.assert_equal(rx[IPv6].fl, p.outer_flow_label)
1538 decrypt_pkt = p.vpp_tun_sa.decrypt(rx[IPv6])
1539 if not decrypt_pkt.haslayer(IPv6):
1540 decrypt_pkt = IPv6(decrypt_pkt[Raw].load)
1541 self.assert_packet_checksums_valid(decrypt_pkt)
1542 self.assert_equal(decrypt_pkt.src, self.pg1.remote_ip6)
1543 self.assert_equal(decrypt_pkt.dst, p.remote_tun_if_host)
1544 self.assert_equal(decrypt_pkt.hlim, p.inner_hop_limit - 1)
1545 self.assert_equal(decrypt_pkt.fl, p.inner_flow_label)
1547 self.logger.debug(ppp("Unexpected packet:", rx))
1549 self.logger.debug(ppp("Decrypted packet:", decrypt_pkt))
1554 def verify_drop_tun_tx_66(self, p_in, count=1, payload_size=64):
1555 self.vapi.cli("clear errors")
1556 self.vapi.cli("clear ipsec sa")
1558 send_pkts = self.gen_pkts6(
1561 src=self.pg1.remote_ip6,
1562 dst=p_in.remote_tun_if_host,
1564 payload_size=payload_size,
1566 self.send_and_assert_no_replies(self.tun_if, send_pkts)
1567 self.logger.info(self.vapi.cli("sh punt stats"))
1569 def verify_drop_tun_rx_66(self, p_in, count=1, payload_size=64):
1570 self.vapi.cli("clear errors")
1571 self.vapi.cli("clear ipsec sa")
1573 send_pkts = self.gen_encrypt_pkts6(
1577 src=p_in.remote_tun_if_host,
1578 dst=self.pg1.remote_ip6,
1581 self.send_and_assert_no_replies(self.tun_if, send_pkts)
1583 def verify_drop_tun_66(self, p_in, count=1, payload_size=64):
1584 self.verify_drop_tun_tx_66(p_in, count=count, payload_size=payload_size)
1585 self.verify_drop_tun_rx_66(p_in, count=count, payload_size=payload_size)
1587 def verify_tun_66(self, p_in, p_out=None, count=1, payload_size=64):
1588 self.vapi.cli("clear errors")
1589 self.vapi.cli("clear ipsec sa")
1593 send_pkts = self.gen_encrypt_pkts6(
1597 src=p_in.remote_tun_if_host,
1598 dst=self.pg1.remote_ip6,
1600 payload_size=payload_size,
1602 recv_pkts = self.send_and_expect(self.tun_if, send_pkts, self.pg1)
1603 self.verify_decrypted6(p_in, recv_pkts)
1605 send_pkts = self.gen_pkts6(
1608 src=self.pg1.remote_ip6,
1609 dst=p_out.remote_tun_if_host,
1611 payload_size=payload_size,
1613 recv_pkts = self.send_and_expect(self.pg1, send_pkts, self.tun_if)
1614 self.verify_encrypted6(p_out, p_out.vpp_tun_sa, recv_pkts)
1616 for rx in recv_pkts:
1617 self.assertEqual(rx[IPv6].src, p_out.tun_src)
1618 self.assertEqual(rx[IPv6].dst, p_out.tun_dst)
1621 self.logger.info(self.vapi.ppcli("show error"))
1622 self.logger.info(self.vapi.ppcli("show ipsec all"))
1623 self.verify_counters6(p_in, p_out, count)
1625 def verify_tun_reass_66(self, p):
1626 self.vapi.cli("clear errors")
1627 self.vapi.ip_reassembly_enable_disable(
1628 sw_if_index=self.tun_if.sw_if_index, enable_ip6=True
1632 send_pkts = self.gen_encrypt_pkts6(
1636 src=p.remote_tun_if_host,
1637 dst=self.pg1.remote_ip6,
1641 send_pkts = fragment_rfc8200(send_pkts[0], 1, 1400, self.logger)
1642 recv_pkts = self.send_and_expect(self.tun_if, send_pkts, self.pg1, n_rx=1)
1643 self.verify_decrypted6(p, recv_pkts)
1645 send_pkts = self.gen_pkts6(
1648 src=self.pg1.remote_ip6,
1649 dst=p.remote_tun_if_host,
1653 recv_pkts = self.send_and_expect(self.pg1, send_pkts, self.tun_if)
1654 self.verify_encrypted6(p, p.vpp_tun_sa, recv_pkts)
1656 self.logger.info(self.vapi.ppcli("show error"))
1657 self.logger.info(self.vapi.ppcli("show ipsec all"))
1658 self.verify_counters6(p, p, 1)
1659 self.vapi.ip_reassembly_enable_disable(
1660 sw_if_index=self.tun_if.sw_if_index, enable_ip6=False
1663 def verify_tun_46(self, p, count=1):
1664 """ipsec 4o6 tunnel basic test"""
1665 self.vapi.cli("clear errors")
1666 self.vapi.cli("clear ipsec sa")
1668 send_pkts = self.gen_encrypt_pkts(
1672 src=p.remote_tun_if_host4,
1673 dst=self.pg1.remote_ip4,
1676 recv_pkts = self.send_and_expect(self.tun_if, send_pkts, self.pg1)
1677 for recv_pkt in recv_pkts:
1678 self.assert_equal(recv_pkt[IP].src, p.remote_tun_if_host4)
1679 self.assert_equal(recv_pkt[IP].dst, self.pg1.remote_ip4)
1680 self.assert_packet_checksums_valid(recv_pkt)
1681 send_pkts = self.gen_pkts(
1683 src=self.pg1.remote_ip4,
1684 dst=p.remote_tun_if_host4,
1687 recv_pkts = self.send_and_expect(self.pg1, send_pkts, self.tun_if)
1688 for recv_pkt in recv_pkts:
1690 decrypt_pkt = p.vpp_tun_sa.decrypt(recv_pkt[IPv6])
1691 if not decrypt_pkt.haslayer(IP):
1692 decrypt_pkt = IP(decrypt_pkt[Raw].load)
1693 self.assert_equal(decrypt_pkt.src, self.pg1.remote_ip4)
1694 self.assert_equal(decrypt_pkt.dst, p.remote_tun_if_host4)
1695 self.assert_packet_checksums_valid(decrypt_pkt)
1697 self.logger.debug(ppp("Unexpected packet:", recv_pkt))
1699 self.logger.debug(ppp("Decrypted packet:", decrypt_pkt))
1704 self.logger.info(self.vapi.ppcli("show error"))
1705 self.logger.info(self.vapi.ppcli("show ipsec all"))
1706 self.verify_counters6(p, p, count)
1709 class IpsecTun6Tests(IpsecTun6):
1710 """UT test methods for Tunnel v6"""
1712 def test_tun_basic66(self):
1713 """ipsec 6o6 tunnel basic test"""
1714 self.verify_tun_66(self.params[socket.AF_INET6], count=1)
1716 def test_tun_reass_basic66(self):
1717 """ipsec 6o6 tunnel basic reassembly test"""
1718 self.verify_tun_reass_66(self.params[socket.AF_INET6])
1720 def test_tun_burst66(self):
1721 """ipsec 6o6 tunnel burst test"""
1722 self.verify_tun_66(self.params[socket.AF_INET6], count=257)
1725 class IpsecTun6HandoffTests(IpsecTun6):
1726 """UT test methods for Tunnel v6 with multiple workers"""
1728 vpp_worker_count = 2
1730 def test_tun_handoff_66(self):
1731 """ipsec 6o6 tunnel worker hand-off test"""
1732 self.vapi.cli("clear errors")
1733 self.vapi.cli("clear ipsec sa")
1736 p = self.params[socket.AF_INET6]
1738 # inject alternately on worker 0 and 1. all counts on the SA
1739 # should be against worker 0
1740 for worker in [0, 1, 0, 1]:
1741 send_pkts = self.gen_encrypt_pkts6(
1745 src=p.remote_tun_if_host,
1746 dst=self.pg1.remote_ip6,
1749 recv_pkts = self.send_and_expect(
1750 self.tun_if, send_pkts, self.pg1, worker=worker
1752 self.verify_decrypted6(p, recv_pkts)
1754 send_pkts = self.gen_pkts6(
1757 src=self.pg1.remote_ip6,
1758 dst=p.remote_tun_if_host,
1761 recv_pkts = self.send_and_expect(
1762 self.pg1, send_pkts, self.tun_if, worker=worker
1764 self.verify_encrypted6(p, p.vpp_tun_sa, recv_pkts)
1766 # all counts against the first worker that was used
1767 self.verify_counters6(p, p, 4 * N_PKTS, worker=0)
1770 class IpsecTun4HandoffTests(IpsecTun4):
1771 """UT test methods for Tunnel v4 with multiple workers"""
1773 vpp_worker_count = 2
1775 def test_tun_handooff_44(self):
1776 """ipsec 4o4 tunnel worker hand-off test"""
1777 self.vapi.cli("clear errors")
1778 self.vapi.cli("clear ipsec sa")
1781 p = self.params[socket.AF_INET]
1783 # inject alternately on worker 0 and 1. all counts on the SA
1784 # should be against worker 0
1785 for worker in [0, 1, 0, 1]:
1786 send_pkts = self.gen_encrypt_pkts(
1790 src=p.remote_tun_if_host,
1791 dst=self.pg1.remote_ip4,
1794 recv_pkts = self.send_and_expect(
1795 self.tun_if, send_pkts, self.pg1, worker=worker
1797 self.verify_decrypted(p, recv_pkts)
1799 send_pkts = self.gen_pkts(
1801 src=self.pg1.remote_ip4,
1802 dst=p.remote_tun_if_host,
1805 recv_pkts = self.send_and_expect(
1806 self.pg1, send_pkts, self.tun_if, worker=worker
1808 self.verify_encrypted(p, p.vpp_tun_sa, recv_pkts)
1810 # all counts against the first worker that was used
1811 self.verify_counters4(p, 4 * N_PKTS, worker=0)
1814 class IpsecTun46Tests(IpsecTun4Tests, IpsecTun6Tests):
1815 """UT test methods for Tunnel v6 & v4"""
1820 class IPSecIPv4Fwd(VppTestCase):
1821 """Test IPSec by capturing and verifying IPv4 forwarded pkts"""
1824 def setUpConstants(cls):
1825 super(IPSecIPv4Fwd, cls).setUpConstants()
1828 super(IPSecIPv4Fwd, self).setUp()
1829 # store SPD objects so we can remove configs on tear down
1831 self.spd_policies = []
1834 # remove SPD policies
1835 for obj in self.spd_policies:
1836 obj.remove_vpp_config()
1837 self.spd_policies = []
1838 # remove SPD items (interface bindings first, then SPD)
1839 for obj in reversed(self.spd_objs):
1840 obj.remove_vpp_config()
1842 # close down pg intfs
1843 for pg in self.pg_interfaces:
1846 super(IPSecIPv4Fwd, self).tearDown()
1848 def create_interfaces(self, num_ifs=2):
1849 # create interfaces pg0 ... pg<num_ifs>
1850 self.create_pg_interfaces(range(num_ifs))
1851 for pg in self.pg_interfaces:
1852 # put the interface up
1854 # configure IPv4 address on the interface
1856 # resolve ARP, so that we know VPP MAC
1858 self.logger.info(self.vapi.ppcli("show int addr"))
1860 def spd_create_and_intf_add(self, spd_id, pg_list):
1861 spd = VppIpsecSpd(self, spd_id)
1862 spd.add_vpp_config()
1863 self.spd_objs.append(spd)
1865 spdItf = VppIpsecSpdItfBinding(self, spd, pg)
1866 spdItf.add_vpp_config()
1867 self.spd_objs.append(spdItf)
1869 def get_policy(self, policy_type):
1870 e = VppEnum.vl_api_ipsec_spd_action_t
1871 if policy_type == "protect":
1872 return e.IPSEC_API_SPD_ACTION_PROTECT
1873 elif policy_type == "bypass":
1874 return e.IPSEC_API_SPD_ACTION_BYPASS
1875 elif policy_type == "discard":
1876 return e.IPSEC_API_SPD_ACTION_DISCARD
1878 raise Exception("Invalid policy type: %s", policy_type)
1880 def spd_add_rem_policy(
1892 spd = VppIpsecSpd(self, spd_id)
1895 src_range_low = ip_address("0.0.0.0")
1896 src_range_high = ip_address("255.255.255.255")
1897 dst_range_low = ip_address("0.0.0.0")
1898 dst_range_high = ip_address("255.255.255.255")
1900 src_range_low = src_if.remote_ip4
1901 src_range_high = src_if.remote_ip4
1902 dst_range_low = dst_if.remote_ip4
1903 dst_range_high = dst_if.remote_ip4
1905 spdEntry = VppIpsecSpdEntry(
1915 policy=self.get_policy(policy_type),
1920 spdEntry.add_vpp_config()
1921 self.spd_policies.append(spdEntry)
1923 spdEntry.remove_vpp_config()
1924 self.spd_policies.remove(spdEntry)
1925 self.logger.info(self.vapi.ppcli("show ipsec all"))
1928 def create_stream(self, src_if, dst_if, pkt_count, src_prt=1234, dst_prt=5678):
1930 for i in range(pkt_count):
1931 # create packet info stored in the test case instance
1932 info = self.create_packet_info(src_if, dst_if)
1933 # convert the info into packet payload
1934 payload = self.info_to_payload(info)
1935 # create the packet itself
1937 Ether(dst=src_if.local_mac, src=src_if.remote_mac)
1938 / IP(src=src_if.remote_ip4, dst=dst_if.remote_ip4)
1939 / UDP(sport=src_prt, dport=dst_prt)
1942 # store a copy of the packet in the packet info
1943 info.data = p.copy()
1944 # append the packet to the list
1946 # return the created packet list
1949 def verify_capture(self, src_if, dst_if, capture):
1951 for packet in capture:
1955 # convert the payload to packet info object
1956 payload_info = self.payload_to_info(packet)
1957 # make sure the indexes match
1959 payload_info.src, src_if.sw_if_index, "source sw_if_index"
1962 payload_info.dst, dst_if.sw_if_index, "destination sw_if_index"
1964 packet_info = self.get_next_packet_info_for_interface2(
1965 src_if.sw_if_index, dst_if.sw_if_index, packet_info
1967 # make sure we didn't run out of saved packets
1968 self.assertIsNotNone(packet_info)
1970 payload_info.index, packet_info.index, "packet info index"
1972 saved_packet = packet_info.data # fetch the saved packet
1973 # assert the values match
1974 self.assert_equal(ip.src, saved_packet[IP].src, "IP source address")
1975 # ... more assertions here
1976 self.assert_equal(udp.sport, saved_packet[UDP].sport, "UDP source port")
1977 except Exception as e:
1978 self.logger.error(ppp("Unexpected or invalid packet:", packet))
1980 remaining_packet = self.get_next_packet_info_for_interface2(
1981 src_if.sw_if_index, dst_if.sw_if_index, packet_info
1985 "Interface %s: Packet expected from interface "
1986 "%s didn't arrive" % (dst_if.name, src_if.name),
1989 def verify_policy_match(self, pkt_count, spdEntry):
1990 self.logger.info("XXXX %s %s", str(spdEntry), str(spdEntry.get_stats()))
1991 matched_pkts = spdEntry.get_stats().get("packets")
1992 self.logger.info("Policy %s matched: %d pkts", str(spdEntry), matched_pkts)
1993 self.assert_equal(pkt_count, matched_pkts)
1996 class SpdFlowCacheTemplate(IPSecIPv4Fwd):
1998 def setUpConstants(cls):
1999 super(SpdFlowCacheTemplate, cls).setUpConstants()
2000 # Override this method with required cmdline parameters e.g.
2001 # cls.vpp_cmdline.extend(["ipsec", "{",
2002 # "ipv4-outbound-spd-flow-cache on",
2004 # cls.logger.info("VPP modified cmdline is %s" % " "
2005 # .join(cls.vpp_cmdline))
2008 super(SpdFlowCacheTemplate, self).setUp()
2011 super(SpdFlowCacheTemplate, self).tearDown()
2013 def get_spd_flow_cache_entries(self, outbound):
2014 """'show ipsec spd' output:
2015 ipv4-inbound-spd-flow-cache-entries: 0
2016 ipv4-outbound-spd-flow-cache-entries: 0
2018 show_ipsec_reply = self.vapi.cli("show ipsec spd")
2019 # match the relevant section of 'show ipsec spd' output
2021 regex_match = re.search(
2022 "ipv4-outbound-spd-flow-cache-entries: (.*)",
2027 regex_match = re.search(
2028 "ipv4-inbound-spd-flow-cache-entries: (.*)", show_ipsec_reply, re.DOTALL
2030 if regex_match is None:
2032 "Unable to find spd flow cache entries \
2033 in 'show ipsec spd' CLI output - regex failed to match"
2037 num_entries = int(regex_match.group(1))
2040 "Unable to get spd flow cache entries \
2041 from 'show ipsec spd' string: %s",
2042 regex_match.group(0),
2044 self.logger.info("%s", regex_match.group(0))
2047 def verify_num_outbound_flow_cache_entries(self, expected_elements):
2049 self.get_spd_flow_cache_entries(outbound=True), expected_elements
2052 def verify_num_inbound_flow_cache_entries(self, expected_elements):
2054 self.get_spd_flow_cache_entries(outbound=False), expected_elements
2057 def crc32_supported(self):
2058 # lscpu is part of util-linux package, available on all Linux Distros
2059 stream = os.popen("lscpu")
2060 cpu_info = stream.read()
2061 # feature/flag "crc32" on Aarch64 and "sse4_2" on x86
2062 # see vppinfra/crc32.h
2063 if "crc32" or "sse4_2" in cpu_info:
2064 self.logger.info("\ncrc32 supported:\n" + cpu_info)
2067 self.logger.info("\ncrc32 NOT supported:\n" + cpu_info)
2071 if __name__ == "__main__":
2072 unittest.main(testRunner=VppTestRunner)