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 IPv6, ICMPv6EchoRequest, IPv6ExtHdrHopByHop, \
10 IPv6ExtHdrFragment, IPv6ExtHdrDestOpt
13 from framework import VppTestCase, VppTestRunner
14 from util import ppp, reassemble4, fragment_rfc791, fragment_rfc8200
15 from vpp_papi import VppEnum
18 class IPsecIPv4Params:
20 addr_type = socket.AF_INET
22 addr_bcast = "255.255.255.255"
27 self.remote_tun_if_host = '1.1.1.1'
28 self.remote_tun_if_host6 = '1111::1'
30 self.scapy_tun_sa_id = 100
31 self.scapy_tun_spi = 1000
32 self.vpp_tun_sa_id = 200
33 self.vpp_tun_spi = 2000
35 self.scapy_tra_sa_id = 300
36 self.scapy_tra_spi = 3000
37 self.vpp_tra_sa_id = 400
38 self.vpp_tra_spi = 4000
40 self.auth_algo_vpp_id = (VppEnum.vl_api_ipsec_integ_alg_t.
41 IPSEC_API_INTEG_ALG_SHA1_96)
42 self.auth_algo = 'HMAC-SHA1-96' # scapy name
43 self.auth_key = b'C91KUR9GYMm5GfkEvNjX'
45 self.crypt_algo_vpp_id = (VppEnum.vl_api_ipsec_crypto_alg_t.
46 IPSEC_API_CRYPTO_ALG_AES_CBC_128)
47 self.crypt_algo = 'AES-CBC' # scapy name
48 self.crypt_key = b'JPjyOWBeVEQiMe7h'
51 self.nat_header = None
52 self.tun_flags = (VppEnum.vl_api_tunnel_encap_decap_flags_t.
53 TUNNEL_API_ENCAP_DECAP_FLAG_NONE)
57 class IPsecIPv6Params:
59 addr_type = socket.AF_INET6
61 addr_bcast = "ffff:ffff:ffff:ffff:ffff:ffff:ffff:ffff"
66 self.remote_tun_if_host = '1111:1111:1111:1111:1111:1111:1111:1111'
67 self.remote_tun_if_host4 = '1.1.1.1'
69 self.scapy_tun_sa_id = 500
70 self.scapy_tun_spi = 3001
71 self.vpp_tun_sa_id = 600
72 self.vpp_tun_spi = 3000
74 self.scapy_tra_sa_id = 700
75 self.scapy_tra_spi = 4001
76 self.vpp_tra_sa_id = 800
77 self.vpp_tra_spi = 4000
79 self.auth_algo_vpp_id = (VppEnum.vl_api_ipsec_integ_alg_t.
80 IPSEC_API_INTEG_ALG_SHA1_96)
81 self.auth_algo = 'HMAC-SHA1-96' # scapy name
82 self.auth_key = b'C91KUR9GYMm5GfkEvNjX'
84 self.crypt_algo_vpp_id = (VppEnum.vl_api_ipsec_crypto_alg_t.
85 IPSEC_API_CRYPTO_ALG_AES_CBC_128)
86 self.crypt_algo = 'AES-CBC' # scapy name
87 self.crypt_key = b'JPjyOWBeVEQiMe7h'
90 self.nat_header = None
91 self.tun_flags = (VppEnum.vl_api_tunnel_encap_decap_flags_t.
92 TUNNEL_API_ENCAP_DECAP_FLAG_NONE)
96 def mk_scapy_crypt_key(p):
97 if p.crypt_algo == "AES-GCM":
98 return p.crypt_key + struct.pack("!I", p.salt)
103 def config_tun_params(p, encryption_type, tun_if):
104 ip_class_by_addr_type = {socket.AF_INET: IP, socket.AF_INET6: IPv6}
105 esn_en = bool(p.flags & (VppEnum.vl_api_ipsec_sad_flags_t.
106 IPSEC_API_SAD_FLAG_USE_ESN))
107 p.tun_dst = tun_if.remote_addr[p.addr_type]
108 p.tun_src = tun_if.local_addr[p.addr_type]
109 crypt_key = mk_scapy_crypt_key(p)
110 p.scapy_tun_sa = SecurityAssociation(
111 encryption_type, spi=p.vpp_tun_spi,
112 crypt_algo=p.crypt_algo,
114 auth_algo=p.auth_algo, auth_key=p.auth_key,
115 tunnel_header=ip_class_by_addr_type[p.addr_type](
118 nat_t_header=p.nat_header,
120 p.vpp_tun_sa = SecurityAssociation(
121 encryption_type, spi=p.scapy_tun_spi,
122 crypt_algo=p.crypt_algo,
124 auth_algo=p.auth_algo, auth_key=p.auth_key,
125 tunnel_header=ip_class_by_addr_type[p.addr_type](
128 nat_t_header=p.nat_header,
132 def config_tra_params(p, encryption_type):
133 esn_en = bool(p.flags & (VppEnum.vl_api_ipsec_sad_flags_t.
134 IPSEC_API_SAD_FLAG_USE_ESN))
135 crypt_key = mk_scapy_crypt_key(p)
136 p.scapy_tra_sa = SecurityAssociation(
139 crypt_algo=p.crypt_algo,
141 auth_algo=p.auth_algo,
143 nat_t_header=p.nat_header,
145 p.vpp_tra_sa = SecurityAssociation(
148 crypt_algo=p.crypt_algo,
150 auth_algo=p.auth_algo,
152 nat_t_header=p.nat_header,
156 class TemplateIpsec(VppTestCase):
161 |tra_if| <-------> |VPP|
166 ------ encrypt --- plain ---
167 |tun_if| <------- |VPP| <------ |pg1|
170 ------ decrypt --- plain ---
171 |tun_if| -------> |VPP| ------> |pg1|
177 def ipsec_select_backend(self):
178 """ empty method to be overloaded when necessary """
183 super(TemplateIpsec, cls).setUpClass()
186 def tearDownClass(cls):
187 super(TemplateIpsec, cls).tearDownClass()
189 def setup_params(self):
190 if not hasattr(self, 'ipv4_params'):
191 self.ipv4_params = IPsecIPv4Params()
192 if not hasattr(self, 'ipv6_params'):
193 self.ipv6_params = IPsecIPv6Params()
194 self.params = {self.ipv4_params.addr_type: self.ipv4_params,
195 self.ipv6_params.addr_type: self.ipv6_params}
197 def config_interfaces(self):
198 self.create_pg_interfaces(range(3))
199 self.interfaces = list(self.pg_interfaces)
200 for i in self.interfaces:
208 super(TemplateIpsec, self).setUp()
212 self.vpp_esp_protocol = (VppEnum.vl_api_ipsec_proto_t.
214 self.vpp_ah_protocol = (VppEnum.vl_api_ipsec_proto_t.
217 self.config_interfaces()
219 self.ipsec_select_backend()
221 def unconfig_interfaces(self):
222 for i in self.interfaces:
228 super(TemplateIpsec, self).tearDown()
230 self.unconfig_interfaces()
232 def show_commands_at_teardown(self):
233 self.logger.info(self.vapi.cli("show hardware"))
235 def gen_encrypt_pkts(self, p, sa, sw_intf, src, dst, count=1,
237 return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
238 sa.encrypt(IP(src=src, dst=dst) /
239 ICMP() / Raw(b'X' * payload_size))
240 for i in range(count)]
242 def gen_encrypt_pkts6(self, p, sa, sw_intf, src, dst, count=1,
244 return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
245 sa.encrypt(IPv6(src=src, dst=dst) /
246 ICMPv6EchoRequest(id=0, seq=1,
247 data='X' * payload_size))
248 for i in range(count)]
250 def gen_pkts(self, sw_intf, src, dst, count=1, payload_size=54):
251 return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
252 IP(src=src, dst=dst) / ICMP() / Raw(b'X' * payload_size)
253 for i in range(count)]
255 def gen_pkts6(self, sw_intf, src, dst, count=1, payload_size=54):
256 return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
257 IPv6(src=src, dst=dst) /
258 ICMPv6EchoRequest(id=0, seq=1, data='X' * payload_size)
259 for i in range(count)]
262 class IpsecTcp(object):
263 def verify_tcp_checksum(self):
264 self.vapi.cli("test http server")
265 p = self.params[socket.AF_INET]
266 send = (Ether(src=self.tun_if.remote_mac, dst=self.tun_if.local_mac) /
267 p.scapy_tun_sa.encrypt(IP(src=p.remote_tun_if_host,
268 dst=self.tun_if.local_ip4) /
269 TCP(flags='S', dport=80)))
270 self.logger.debug(ppp("Sending packet:", send))
271 recv = self.send_and_expect(self.tun_if, [send], self.tun_if)
273 decrypted = p.vpp_tun_sa.decrypt(recv[IP])
274 self.assert_packet_checksums_valid(decrypted)
277 class IpsecTcpTests(IpsecTcp):
278 def test_tcp_checksum(self):
279 """ verify checksum correctness for vpp generated packets """
280 self.verify_tcp_checksum()
283 class IpsecTra4(object):
284 """ verify methods for Transport v4 """
285 def verify_tra_anti_replay(self):
286 p = self.params[socket.AF_INET]
287 esn_en = p.vpp_tra_sa.esn_en
289 seq_cycle_node_name = ('/err/%s/sequence number cycled' %
290 self.tra4_encrypt_node_name)
291 replay_node_name = ('/err/%s/SA replayed packet' %
292 self.tra4_decrypt_node_name)
293 if ESP == self.encryption_type and p.crypt_algo == "AES-GCM":
294 hash_failed_node_name = ('/err/%s/ESP decryption failed' %
295 self.tra4_decrypt_node_name)
297 hash_failed_node_name = ('/err/%s/Integrity check failed' %
298 self.tra4_decrypt_node_name)
299 replay_count = self.statistics.get_err_counter(replay_node_name)
300 hash_failed_count = self.statistics.get_err_counter(
301 hash_failed_node_name)
302 seq_cycle_count = self.statistics.get_err_counter(seq_cycle_node_name)
304 if ESP == self.encryption_type:
305 undersize_node_name = ('/err/%s/undersized packet' %
306 self.tra4_decrypt_node_name)
307 undersize_count = self.statistics.get_err_counter(
311 # send packets with seq numbers 1->34
312 # this means the window size is still in Case B (see RFC4303
315 # for reasons i haven't investigated Scapy won't create a packet with
318 pkts = [(Ether(src=self.tra_if.remote_mac,
319 dst=self.tra_if.local_mac) /
320 p.scapy_tra_sa.encrypt(IP(src=self.tra_if.remote_ip4,
321 dst=self.tra_if.local_ip4) /
324 for seq in range(1, 34)]
325 recv_pkts = self.send_and_expect(self.tra_if, pkts, self.tra_if)
327 # replayed packets are dropped
328 self.send_and_assert_no_replies(self.tra_if, pkts)
329 replay_count += len(pkts)
330 self.assert_error_counter_equal(replay_node_name, replay_count)
333 # now send a batch of packets all with the same sequence number
334 # the first packet in the batch is legitimate, the rest bogus
336 pkts = (Ether(src=self.tra_if.remote_mac,
337 dst=self.tra_if.local_mac) /
338 p.scapy_tra_sa.encrypt(IP(src=self.tra_if.remote_ip4,
339 dst=self.tra_if.local_ip4) /
342 recv_pkts = self.send_and_expect(self.tra_if, pkts * 8,
345 self.assert_error_counter_equal(replay_node_name, replay_count)
348 # now move the window over to 257 (more than one byte) and into Case A
350 pkt = (Ether(src=self.tra_if.remote_mac,
351 dst=self.tra_if.local_mac) /
352 p.scapy_tra_sa.encrypt(IP(src=self.tra_if.remote_ip4,
353 dst=self.tra_if.local_ip4) /
356 recv_pkts = self.send_and_expect(self.tra_if, [pkt], self.tra_if)
358 # replayed packets are dropped
359 self.send_and_assert_no_replies(self.tra_if, pkt * 3)
361 self.assert_error_counter_equal(replay_node_name, replay_count)
363 # the window size is 64 packets
364 # in window are still accepted
365 pkt = (Ether(src=self.tra_if.remote_mac,
366 dst=self.tra_if.local_mac) /
367 p.scapy_tra_sa.encrypt(IP(src=self.tra_if.remote_ip4,
368 dst=self.tra_if.local_ip4) /
371 recv_pkts = self.send_and_expect(self.tra_if, [pkt], self.tra_if)
373 # a packet that does not decrypt does not move the window forward
374 bogus_sa = SecurityAssociation(self.encryption_type,
376 crypt_algo=p.crypt_algo,
377 crypt_key=mk_scapy_crypt_key(p)[::-1],
378 auth_algo=p.auth_algo,
379 auth_key=p.auth_key[::-1])
380 pkt = (Ether(src=self.tra_if.remote_mac,
381 dst=self.tra_if.local_mac) /
382 bogus_sa.encrypt(IP(src=self.tra_if.remote_ip4,
383 dst=self.tra_if.local_ip4) /
386 self.send_and_assert_no_replies(self.tra_if, pkt * 17)
388 hash_failed_count += 17
389 self.assert_error_counter_equal(hash_failed_node_name,
392 # a malformed 'runt' packet
393 # created by a mis-constructed SA
394 if (ESP == self.encryption_type and p.crypt_algo != "NULL"):
395 bogus_sa = SecurityAssociation(self.encryption_type,
397 pkt = (Ether(src=self.tra_if.remote_mac,
398 dst=self.tra_if.local_mac) /
399 bogus_sa.encrypt(IP(src=self.tra_if.remote_ip4,
400 dst=self.tra_if.local_ip4) /
403 self.send_and_assert_no_replies(self.tra_if, pkt * 17)
405 undersize_count += 17
406 self.assert_error_counter_equal(undersize_node_name,
409 # which we can determine since this packet is still in the window
410 pkt = (Ether(src=self.tra_if.remote_mac,
411 dst=self.tra_if.local_mac) /
412 p.scapy_tra_sa.encrypt(IP(src=self.tra_if.remote_ip4,
413 dst=self.tra_if.local_ip4) /
416 self.send_and_expect(self.tra_if, [pkt], self.tra_if)
419 # out of window are dropped
420 # this is Case B. So VPP will consider this to be a high seq num wrap
421 # and so the decrypt attempt will fail
423 pkt = (Ether(src=self.tra_if.remote_mac,
424 dst=self.tra_if.local_mac) /
425 p.scapy_tra_sa.encrypt(IP(src=self.tra_if.remote_ip4,
426 dst=self.tra_if.local_ip4) /
429 self.send_and_assert_no_replies(self.tra_if, pkt * 17)
432 # an out of window error with ESN looks like a high sequence
433 # wrap. but since it isn't then the verify will fail.
434 hash_failed_count += 17
435 self.assert_error_counter_equal(hash_failed_node_name,
440 self.assert_error_counter_equal(replay_node_name,
443 # valid packet moves the window over to 258
444 pkt = (Ether(src=self.tra_if.remote_mac,
445 dst=self.tra_if.local_mac) /
446 p.scapy_tra_sa.encrypt(IP(src=self.tra_if.remote_ip4,
447 dst=self.tra_if.local_ip4) /
450 rx = self.send_and_expect(self.tra_if, [pkt], self.tra_if)
451 decrypted = p.vpp_tra_sa.decrypt(rx[0][IP])
454 # move VPP's SA TX seq-num to just before the seq-number wrap.
455 # then fire in a packet that VPP should drop on TX because it
456 # causes the TX seq number to wrap; unless we're using extened sequence
459 self.vapi.cli("test ipsec sa %d seq 0xffffffff" % p.scapy_tra_sa_id)
460 self.logger.info(self.vapi.ppcli("show ipsec sa 0"))
461 self.logger.info(self.vapi.ppcli("show ipsec sa 1"))
463 pkts = [(Ether(src=self.tra_if.remote_mac,
464 dst=self.tra_if.local_mac) /
465 p.scapy_tra_sa.encrypt(IP(src=self.tra_if.remote_ip4,
466 dst=self.tra_if.local_ip4) /
469 for seq in range(259, 280)]
472 rxs = self.send_and_expect(self.tra_if, pkts, self.tra_if)
475 # in order for scapy to decrypt its SA's high order number needs
478 p.vpp_tra_sa.seq_num = 0x100000000
480 decrypted = p.vpp_tra_sa.decrypt(rx[0][IP])
483 # wrap scapy's TX high sequence number. VPP is in case B, so it
484 # will consider this a high seq wrap also.
485 # The low seq num we set it to will place VPP's RX window in Case A
487 p.scapy_tra_sa.seq_num = 0x100000005
488 pkt = (Ether(src=self.tra_if.remote_mac,
489 dst=self.tra_if.local_mac) /
490 p.scapy_tra_sa.encrypt(IP(src=self.tra_if.remote_ip4,
491 dst=self.tra_if.local_ip4) /
493 seq_num=0x100000005))
494 rx = self.send_and_expect(self.tra_if, [pkt], self.tra_if)
495 decrypted = p.vpp_tra_sa.decrypt(rx[0][IP])
498 # A packet that has seq num between (2^32-64) and 5 is within
501 p.scapy_tra_sa.seq_num = 0xfffffffd
502 pkt = (Ether(src=self.tra_if.remote_mac,
503 dst=self.tra_if.local_mac) /
504 p.scapy_tra_sa.encrypt(IP(src=self.tra_if.remote_ip4,
505 dst=self.tra_if.local_ip4) /
508 rx = self.send_and_expect(self.tra_if, [pkt], self.tra_if)
509 decrypted = p.vpp_tra_sa.decrypt(rx[0][IP])
512 # While in case A we cannot wrap the high sequence number again
513 # becuase VPP will consider this packet to be one that moves the
516 pkt = (Ether(src=self.tra_if.remote_mac,
517 dst=self.tra_if.local_mac) /
518 p.scapy_tra_sa.encrypt(IP(src=self.tra_if.remote_ip4,
519 dst=self.tra_if.local_ip4) /
521 seq_num=0x200000999))
522 self.send_and_assert_no_replies(self.tra_if, [pkt], self.tra_if)
524 hash_failed_count += 1
525 self.assert_error_counter_equal(hash_failed_node_name,
529 # but if we move the wondow forward to case B, then we can wrap
532 p.scapy_tra_sa.seq_num = 0x100000555
533 pkt = (Ether(src=self.tra_if.remote_mac,
534 dst=self.tra_if.local_mac) /
535 p.scapy_tra_sa.encrypt(IP(src=self.tra_if.remote_ip4,
536 dst=self.tra_if.local_ip4) /
538 seq_num=0x100000555))
539 rx = self.send_and_expect(self.tra_if, [pkt], self.tra_if)
540 decrypted = p.vpp_tra_sa.decrypt(rx[0][IP])
542 p.scapy_tra_sa.seq_num = 0x200000444
543 pkt = (Ether(src=self.tra_if.remote_mac,
544 dst=self.tra_if.local_mac) /
545 p.scapy_tra_sa.encrypt(IP(src=self.tra_if.remote_ip4,
546 dst=self.tra_if.local_ip4) /
548 seq_num=0x200000444))
549 rx = self.send_and_expect(self.tra_if, [pkt], self.tra_if)
550 decrypted = p.vpp_tra_sa.decrypt(rx[0][IP])
554 # without ESN TX sequence numbers can't wrap and packets are
555 # dropped from here on out.
557 self.send_and_assert_no_replies(self.tra_if, pkts)
558 seq_cycle_count += len(pkts)
559 self.assert_error_counter_equal(seq_cycle_node_name,
562 # move the security-associations seq number on to the last we used
563 self.vapi.cli("test ipsec sa %d seq 0x15f" % p.scapy_tra_sa_id)
564 p.scapy_tra_sa.seq_num = 351
565 p.vpp_tra_sa.seq_num = 351
567 def verify_tra_basic4(self, count=1, payload_size=54):
568 """ ipsec v4 transport basic test """
569 self.vapi.cli("clear errors")
570 self.vapi.cli("clear ipsec sa")
572 p = self.params[socket.AF_INET]
573 send_pkts = self.gen_encrypt_pkts(p, p.scapy_tra_sa, self.tra_if,
574 src=self.tra_if.remote_ip4,
575 dst=self.tra_if.local_ip4,
577 payload_size=payload_size)
578 recv_pkts = self.send_and_expect(self.tra_if, send_pkts,
581 self.assertEqual(len(rx) - len(Ether()), rx[IP].len)
582 self.assert_packet_checksums_valid(rx)
584 decrypted = p.vpp_tra_sa.decrypt(rx[IP])
585 self.assert_packet_checksums_valid(decrypted)
587 self.logger.debug(ppp("Unexpected packet:", rx))
590 self.logger.info(self.vapi.ppcli("show error"))
591 self.logger.info(self.vapi.ppcli("show ipsec all"))
593 pkts = p.tra_sa_in.get_stats()['packets']
594 self.assertEqual(pkts, count,
595 "incorrect SA in counts: expected %d != %d" %
597 pkts = p.tra_sa_out.get_stats()['packets']
598 self.assertEqual(pkts, count,
599 "incorrect SA out counts: expected %d != %d" %
602 self.assert_packet_counter_equal(self.tra4_encrypt_node_name, count)
603 self.assert_packet_counter_equal(self.tra4_decrypt_node_name, count)
606 class IpsecTra4Tests(IpsecTra4):
607 """ UT test methods for Transport v4 """
608 def test_tra_anti_replay(self):
609 """ ipsec v4 transport anti-replay test """
610 self.verify_tra_anti_replay()
612 def test_tra_basic(self, count=1):
613 """ ipsec v4 transport basic test """
614 self.verify_tra_basic4(count=1)
616 def test_tra_burst(self):
617 """ ipsec v4 transport burst test """
618 self.verify_tra_basic4(count=257)
621 class IpsecTra6(object):
622 """ verify methods for Transport v6 """
623 def verify_tra_basic6(self, count=1, payload_size=54):
624 self.vapi.cli("clear errors")
625 self.vapi.cli("clear ipsec sa")
627 p = self.params[socket.AF_INET6]
628 send_pkts = self.gen_encrypt_pkts6(p, p.scapy_tra_sa, self.tra_if,
629 src=self.tra_if.remote_ip6,
630 dst=self.tra_if.local_ip6,
632 payload_size=payload_size)
633 recv_pkts = self.send_and_expect(self.tra_if, send_pkts,
636 self.assertEqual(len(rx) - len(Ether()) - len(IPv6()),
639 decrypted = p.vpp_tra_sa.decrypt(rx[IPv6])
640 self.assert_packet_checksums_valid(decrypted)
642 self.logger.debug(ppp("Unexpected packet:", rx))
645 self.logger.info(self.vapi.ppcli("show error"))
646 self.logger.info(self.vapi.ppcli("show ipsec all"))
648 pkts = p.tra_sa_in.get_stats()['packets']
649 self.assertEqual(pkts, count,
650 "incorrect SA in counts: expected %d != %d" %
652 pkts = p.tra_sa_out.get_stats()['packets']
653 self.assertEqual(pkts, count,
654 "incorrect SA out counts: expected %d != %d" %
656 self.assert_packet_counter_equal(self.tra6_encrypt_node_name, count)
657 self.assert_packet_counter_equal(self.tra6_decrypt_node_name, count)
659 def gen_encrypt_pkts_ext_hdrs6(self, sa, sw_intf, src, dst, count=1,
661 return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
662 sa.encrypt(IPv6(src=src, dst=dst) /
663 ICMPv6EchoRequest(id=0, seq=1,
664 data='X' * payload_size))
665 for i in range(count)]
667 def gen_pkts_ext_hdrs6(self, sw_intf, src, dst, count=1, payload_size=54):
668 return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
669 IPv6(src=src, dst=dst) /
670 IPv6ExtHdrHopByHop() /
671 IPv6ExtHdrFragment(id=2, offset=200) /
673 for i in range(count)]
675 def verify_tra_encrypted6(self, p, sa, rxs):
678 self.assert_packet_checksums_valid(rx)
680 decrypt_pkt = p.vpp_tra_sa.decrypt(rx[IPv6])
681 decrypted.append(decrypt_pkt)
682 self.assert_equal(decrypt_pkt.src, self.tra_if.local_ip6)
683 self.assert_equal(decrypt_pkt.dst, self.tra_if.remote_ip6)
685 self.logger.debug(ppp("Unexpected packet:", rx))
687 self.logger.debug(ppp("Decrypted packet:", decrypt_pkt))
693 def verify_tra_66_ext_hdrs(self, p):
697 # check we can decrypt with options
699 tx = self.gen_encrypt_pkts_ext_hdrs6(p.scapy_tra_sa, self.tra_if,
700 src=self.tra_if.remote_ip6,
701 dst=self.tra_if.local_ip6,
703 self.send_and_expect(self.tra_if, tx, self.tra_if)
706 # injecting a packet from ourselves to be routed of box is a hack
707 # but it matches an outbout policy, alors je ne regrette rien
710 # one extension before ESP
711 tx = (Ether(src=self.pg2.remote_mac, dst=self.pg2.local_mac) /
712 IPv6(src=self.tra_if.local_ip6,
713 dst=self.tra_if.remote_ip6) /
714 IPv6ExtHdrFragment(id=2, offset=200) /
717 rxs = self.send_and_expect(self.pg2, [tx], self.tra_if)
718 dcs = self.verify_tra_encrypted6(p, p.vpp_tra_sa, rxs)
721 # for reasons i'm not going to investigate scapy does not
722 # created the correct headers after decrypt. but reparsing
723 # the ipv6 packet fixes it
724 dc = IPv6(raw(dc[IPv6]))
725 self.assert_equal(dc[IPv6ExtHdrFragment].id, 2)
727 # two extensions before ESP
728 tx = (Ether(src=self.pg2.remote_mac, dst=self.pg2.local_mac) /
729 IPv6(src=self.tra_if.local_ip6,
730 dst=self.tra_if.remote_ip6) /
731 IPv6ExtHdrHopByHop() /
732 IPv6ExtHdrFragment(id=2, offset=200) /
735 rxs = self.send_and_expect(self.pg2, [tx], self.tra_if)
736 dcs = self.verify_tra_encrypted6(p, p.vpp_tra_sa, rxs)
739 dc = IPv6(raw(dc[IPv6]))
740 self.assertTrue(dc[IPv6ExtHdrHopByHop])
741 self.assert_equal(dc[IPv6ExtHdrFragment].id, 2)
743 # two extensions before ESP, one after
744 tx = (Ether(src=self.pg2.remote_mac, dst=self.pg2.local_mac) /
745 IPv6(src=self.tra_if.local_ip6,
746 dst=self.tra_if.remote_ip6) /
747 IPv6ExtHdrHopByHop() /
748 IPv6ExtHdrFragment(id=2, offset=200) /
749 IPv6ExtHdrDestOpt() /
752 rxs = self.send_and_expect(self.pg2, [tx], self.tra_if)
753 dcs = self.verify_tra_encrypted6(p, p.vpp_tra_sa, rxs)
756 dc = IPv6(raw(dc[IPv6]))
757 self.assertTrue(dc[IPv6ExtHdrDestOpt])
758 self.assertTrue(dc[IPv6ExtHdrHopByHop])
759 self.assert_equal(dc[IPv6ExtHdrFragment].id, 2)
762 class IpsecTra6Tests(IpsecTra6):
763 """ UT test methods for Transport v6 """
764 def test_tra_basic6(self):
765 """ ipsec v6 transport basic test """
766 self.verify_tra_basic6(count=1)
768 def test_tra_burst6(self):
769 """ ipsec v6 transport burst test """
770 self.verify_tra_basic6(count=257)
773 class IpsecTra6ExtTests(IpsecTra6):
774 def test_tra_ext_hdrs_66(self):
775 """ ipsec 6o6 tra extension headers test """
776 self.verify_tra_66_ext_hdrs(self.params[socket.AF_INET6])
779 class IpsecTra46Tests(IpsecTra4Tests, IpsecTra6Tests):
780 """ UT test methods for Transport v6 and v4"""
784 class IpsecTun4(object):
785 """ verify methods for Tunnel v4 """
786 def verify_counters4(self, p, count, n_frags=None, worker=None):
789 if (hasattr(p, "spd_policy_in_any")):
790 pkts = p.spd_policy_in_any.get_stats(worker)['packets']
791 self.assertEqual(pkts, count,
792 "incorrect SPD any policy: expected %d != %d" %
795 if (hasattr(p, "tun_sa_in")):
796 pkts = p.tun_sa_in.get_stats(worker)['packets']
797 self.assertEqual(pkts, count,
798 "incorrect SA in counts: expected %d != %d" %
800 pkts = p.tun_sa_out.get_stats(worker)['packets']
801 self.assertEqual(pkts, n_frags,
802 "incorrect SA out counts: expected %d != %d" %
805 self.assert_packet_counter_equal(self.tun4_encrypt_node_name, n_frags)
806 self.assert_packet_counter_equal(self.tun4_decrypt_node_name, count)
808 def verify_decrypted(self, p, rxs):
810 self.assert_equal(rx[IP].src, p.remote_tun_if_host)
811 self.assert_equal(rx[IP].dst, self.pg1.remote_ip4)
812 self.assert_packet_checksums_valid(rx)
814 def verify_esp_padding(self, sa, esp_payload, decrypt_pkt):
815 align = sa.crypt_algo.block_size
818 exp_len = (len(decrypt_pkt) + 2 + (align - 1)) & ~(align - 1)
819 exp_len += sa.crypt_algo.iv_size
820 exp_len += sa.crypt_algo.icv_size or sa.auth_algo.icv_size
821 self.assertEqual(exp_len, len(esp_payload))
823 def verify_encrypted(self, p, sa, rxs):
827 self.assertEqual(rx[UDP].dport, 4500)
828 self.assert_packet_checksums_valid(rx)
829 self.assertEqual(len(rx) - len(Ether()), rx[IP].len)
832 decrypt_pkt = p.vpp_tun_sa.decrypt(rx_ip)
833 if not decrypt_pkt.haslayer(IP):
834 decrypt_pkt = IP(decrypt_pkt[Raw].load)
835 if rx_ip.proto == socket.IPPROTO_ESP:
836 self.verify_esp_padding(sa, rx_ip[ESP].data, decrypt_pkt)
837 decrypt_pkts.append(decrypt_pkt)
838 self.assert_equal(decrypt_pkt.src, self.pg1.remote_ip4)
839 self.assert_equal(decrypt_pkt.dst, p.remote_tun_if_host)
841 self.logger.debug(ppp("Unexpected packet:", rx))
843 self.logger.debug(ppp("Decrypted packet:", decrypt_pkt))
847 pkts = reassemble4(decrypt_pkts)
849 self.assert_packet_checksums_valid(pkt)
851 def verify_tun_44(self, p, count=1, payload_size=64, n_rx=None):
852 self.vapi.cli("clear errors")
853 self.vapi.cli("clear ipsec counters")
854 self.vapi.cli("clear ipsec sa")
858 send_pkts = self.gen_encrypt_pkts(p, p.scapy_tun_sa, self.tun_if,
859 src=p.remote_tun_if_host,
860 dst=self.pg1.remote_ip4,
862 payload_size=payload_size)
863 recv_pkts = self.send_and_expect(self.tun_if, send_pkts, self.pg1)
864 self.verify_decrypted(p, recv_pkts)
866 send_pkts = self.gen_pkts(self.pg1, src=self.pg1.remote_ip4,
867 dst=p.remote_tun_if_host, count=count,
868 payload_size=payload_size)
869 recv_pkts = self.send_and_expect(self.pg1, send_pkts,
871 self.verify_encrypted(p, p.vpp_tun_sa, recv_pkts)
874 self.assertEqual(rx[IP].src, p.tun_src)
875 self.assertEqual(rx[IP].dst, p.tun_dst)
878 self.logger.info(self.vapi.ppcli("show error"))
879 self.logger.info(self.vapi.ppcli("show ipsec all"))
881 self.logger.info(self.vapi.ppcli("show ipsec sa 0"))
882 self.logger.info(self.vapi.ppcli("show ipsec sa 4"))
883 self.verify_counters4(p, count, n_rx)
885 def verify_tun_dropped_44(self, p, count=1, payload_size=64, n_rx=None):
886 self.vapi.cli("clear errors")
890 send_pkts = self.gen_encrypt_pkts(p, p.scapy_tun_sa, self.tun_if,
891 src=p.remote_tun_if_host,
892 dst=self.pg1.remote_ip4,
894 self.send_and_assert_no_replies(self.tun_if, send_pkts)
896 send_pkts = self.gen_pkts(self.pg1, src=self.pg1.remote_ip4,
897 dst=p.remote_tun_if_host, count=count,
898 payload_size=payload_size)
899 self.send_and_assert_no_replies(self.pg1, send_pkts)
902 self.logger.info(self.vapi.ppcli("show error"))
903 self.logger.info(self.vapi.ppcli("show ipsec all"))
905 def verify_tun_reass_44(self, p):
906 self.vapi.cli("clear errors")
907 self.vapi.ip_reassembly_enable_disable(
908 sw_if_index=self.tun_if.sw_if_index, enable_ip4=True)
911 send_pkts = self.gen_encrypt_pkts(p, p.scapy_tun_sa, self.tun_if,
912 src=p.remote_tun_if_host,
913 dst=self.pg1.remote_ip4,
916 send_pkts = fragment_rfc791(send_pkts[0], 1400)
917 recv_pkts = self.send_and_expect(self.tun_if, send_pkts,
919 self.verify_decrypted(p, recv_pkts)
921 send_pkts = self.gen_pkts(self.pg1, src=self.pg1.remote_ip4,
922 dst=p.remote_tun_if_host, count=1)
923 recv_pkts = self.send_and_expect(self.pg1, send_pkts,
925 self.verify_encrypted(p, p.vpp_tun_sa, recv_pkts)
928 self.logger.info(self.vapi.ppcli("show error"))
929 self.logger.info(self.vapi.ppcli("show ipsec all"))
931 self.verify_counters4(p, 1, 1)
932 self.vapi.ip_reassembly_enable_disable(
933 sw_if_index=self.tun_if.sw_if_index, enable_ip4=False)
935 def verify_tun_64(self, p, count=1):
936 self.vapi.cli("clear errors")
937 self.vapi.cli("clear ipsec sa")
939 send_pkts = self.gen_encrypt_pkts6(p, p.scapy_tun_sa, self.tun_if,
940 src=p.remote_tun_if_host6,
941 dst=self.pg1.remote_ip6,
943 recv_pkts = self.send_and_expect(self.tun_if, send_pkts, self.pg1)
944 for recv_pkt in recv_pkts:
945 self.assert_equal(recv_pkt[IPv6].src, p.remote_tun_if_host6)
946 self.assert_equal(recv_pkt[IPv6].dst, self.pg1.remote_ip6)
947 self.assert_packet_checksums_valid(recv_pkt)
948 send_pkts = self.gen_pkts6(self.pg1, src=self.pg1.remote_ip6,
949 dst=p.remote_tun_if_host6, count=count)
950 recv_pkts = self.send_and_expect(self.pg1, send_pkts, self.tun_if)
951 for recv_pkt in recv_pkts:
953 decrypt_pkt = p.vpp_tun_sa.decrypt(recv_pkt[IP])
954 if not decrypt_pkt.haslayer(IPv6):
955 decrypt_pkt = IPv6(decrypt_pkt[Raw].load)
956 self.assert_equal(decrypt_pkt.src, self.pg1.remote_ip6)
957 self.assert_equal(decrypt_pkt.dst, p.remote_tun_if_host6)
958 self.assert_packet_checksums_valid(decrypt_pkt)
960 self.logger.error(ppp("Unexpected packet:", recv_pkt))
963 ppp("Decrypted packet:", decrypt_pkt))
968 self.logger.info(self.vapi.ppcli("show error"))
969 self.logger.info(self.vapi.ppcli("show ipsec all"))
971 self.verify_counters4(p, count)
973 def verify_keepalive(self, p):
974 pkt = (Ether(src=self.tun_if.remote_mac, dst=self.tun_if.local_mac) /
975 IP(src=p.remote_tun_if_host, dst=self.tun_if.local_ip4) /
976 UDP(sport=333, dport=4500) /
978 self.send_and_assert_no_replies(self.tun_if, pkt*31)
979 self.assert_error_counter_equal(
980 '/err/%s/NAT Keepalive' % self.tun4_input_node, 31)
982 pkt = (Ether(src=self.tun_if.remote_mac, dst=self.tun_if.local_mac) /
983 IP(src=p.remote_tun_if_host, dst=self.tun_if.local_ip4) /
984 UDP(sport=333, dport=4500) /
986 self.send_and_assert_no_replies(self.tun_if, pkt*31)
987 self.assert_error_counter_equal(
988 '/err/%s/Too Short' % self.tun4_input_node, 31)
991 class IpsecTun4Tests(IpsecTun4):
992 """ UT test methods for Tunnel v4 """
993 def test_tun_basic44(self):
994 """ ipsec 4o4 tunnel basic test """
995 self.verify_tun_44(self.params[socket.AF_INET], count=1)
996 self.tun_if.admin_down()
997 self.tun_if.resolve_arp()
998 self.tun_if.admin_up()
999 self.verify_tun_44(self.params[socket.AF_INET], count=1)
1001 def test_tun_reass_basic44(self):
1002 """ ipsec 4o4 tunnel basic reassembly test """
1003 self.verify_tun_reass_44(self.params[socket.AF_INET])
1005 def test_tun_burst44(self):
1006 """ ipsec 4o4 tunnel burst test """
1007 self.verify_tun_44(self.params[socket.AF_INET], count=127)
1010 class IpsecTun6(object):
1011 """ verify methods for Tunnel v6 """
1012 def verify_counters6(self, p_in, p_out, count, worker=None):
1013 if (hasattr(p_in, "tun_sa_in")):
1014 pkts = p_in.tun_sa_in.get_stats(worker)['packets']
1015 self.assertEqual(pkts, count,
1016 "incorrect SA in counts: expected %d != %d" %
1018 if (hasattr(p_out, "tun_sa_out")):
1019 pkts = p_out.tun_sa_out.get_stats(worker)['packets']
1020 self.assertEqual(pkts, count,
1021 "incorrect SA out counts: expected %d != %d" %
1023 self.assert_packet_counter_equal(self.tun6_encrypt_node_name, count)
1024 self.assert_packet_counter_equal(self.tun6_decrypt_node_name, count)
1026 def verify_decrypted6(self, p, rxs):
1028 self.assert_equal(rx[IPv6].src, p.remote_tun_if_host)
1029 self.assert_equal(rx[IPv6].dst, self.pg1.remote_ip6)
1030 self.assert_packet_checksums_valid(rx)
1032 def verify_encrypted6(self, p, sa, rxs):
1034 self.assert_packet_checksums_valid(rx)
1035 self.assertEqual(len(rx) - len(Ether()) - len(IPv6()),
1038 decrypt_pkt = p.vpp_tun_sa.decrypt(rx[IPv6])
1039 if not decrypt_pkt.haslayer(IPv6):
1040 decrypt_pkt = IPv6(decrypt_pkt[Raw].load)
1041 self.assert_packet_checksums_valid(decrypt_pkt)
1042 self.assert_equal(decrypt_pkt.src, self.pg1.remote_ip6)
1043 self.assert_equal(decrypt_pkt.dst, p.remote_tun_if_host)
1045 self.logger.debug(ppp("Unexpected packet:", rx))
1047 self.logger.debug(ppp("Decrypted packet:", decrypt_pkt))
1052 def verify_drop_tun_66(self, p_in, count=1, payload_size=64):
1053 self.vapi.cli("clear errors")
1054 self.vapi.cli("clear ipsec sa")
1056 send_pkts = self.gen_encrypt_pkts6(p_in, p_in.scapy_tun_sa,
1058 src=p_in.remote_tun_if_host,
1059 dst=self.pg1.remote_ip6,
1061 self.send_and_assert_no_replies(self.tun_if, send_pkts)
1062 self.logger.info(self.vapi.cli("sh punt stats"))
1064 def verify_tun_66(self, p_in, p_out=None, count=1, payload_size=64):
1065 self.vapi.cli("clear errors")
1066 self.vapi.cli("clear ipsec sa")
1070 send_pkts = self.gen_encrypt_pkts6(p_in, p_in.scapy_tun_sa,
1072 src=p_in.remote_tun_if_host,
1073 dst=self.pg1.remote_ip6,
1075 payload_size=payload_size)
1076 recv_pkts = self.send_and_expect(self.tun_if, send_pkts, self.pg1)
1077 self.verify_decrypted6(p_in, recv_pkts)
1079 send_pkts = self.gen_pkts6(self.pg1, src=self.pg1.remote_ip6,
1080 dst=p_out.remote_tun_if_host,
1082 payload_size=payload_size)
1083 recv_pkts = self.send_and_expect(self.pg1, send_pkts, self.tun_if)
1084 self.verify_encrypted6(p_out, p_out.vpp_tun_sa, recv_pkts)
1086 for rx in recv_pkts:
1087 self.assertEqual(rx[IPv6].src, p_out.tun_src)
1088 self.assertEqual(rx[IPv6].dst, p_out.tun_dst)
1091 self.logger.info(self.vapi.ppcli("show error"))
1092 self.logger.info(self.vapi.ppcli("show ipsec all"))
1093 self.verify_counters6(p_in, p_out, count)
1095 def verify_tun_reass_66(self, p):
1096 self.vapi.cli("clear errors")
1097 self.vapi.ip_reassembly_enable_disable(
1098 sw_if_index=self.tun_if.sw_if_index, enable_ip6=True)
1101 send_pkts = self.gen_encrypt_pkts6(p, p.scapy_tun_sa, self.tun_if,
1102 src=p.remote_tun_if_host,
1103 dst=self.pg1.remote_ip6,
1106 send_pkts = fragment_rfc8200(send_pkts[0], 1, 1400, self.logger)
1107 recv_pkts = self.send_and_expect(self.tun_if, send_pkts,
1109 self.verify_decrypted6(p, recv_pkts)
1111 send_pkts = self.gen_pkts6(self.pg1, src=self.pg1.remote_ip6,
1112 dst=p.remote_tun_if_host,
1115 recv_pkts = self.send_and_expect(self.pg1, send_pkts,
1117 self.verify_encrypted6(p, p.vpp_tun_sa, recv_pkts)
1119 self.logger.info(self.vapi.ppcli("show error"))
1120 self.logger.info(self.vapi.ppcli("show ipsec all"))
1121 self.verify_counters6(p, p, 1)
1122 self.vapi.ip_reassembly_enable_disable(
1123 sw_if_index=self.tun_if.sw_if_index, enable_ip6=False)
1125 def verify_tun_46(self, p, count=1):
1126 """ ipsec 4o6 tunnel basic test """
1127 self.vapi.cli("clear errors")
1128 self.vapi.cli("clear ipsec sa")
1130 send_pkts = self.gen_encrypt_pkts(p, p.scapy_tun_sa, self.tun_if,
1131 src=p.remote_tun_if_host4,
1132 dst=self.pg1.remote_ip4,
1134 recv_pkts = self.send_and_expect(self.tun_if, send_pkts, self.pg1)
1135 for recv_pkt in recv_pkts:
1136 self.assert_equal(recv_pkt[IP].src, p.remote_tun_if_host4)
1137 self.assert_equal(recv_pkt[IP].dst, self.pg1.remote_ip4)
1138 self.assert_packet_checksums_valid(recv_pkt)
1139 send_pkts = self.gen_pkts(self.pg1, src=self.pg1.remote_ip4,
1140 dst=p.remote_tun_if_host4,
1142 recv_pkts = self.send_and_expect(self.pg1, send_pkts, self.tun_if)
1143 for recv_pkt in recv_pkts:
1145 decrypt_pkt = p.vpp_tun_sa.decrypt(recv_pkt[IPv6])
1146 if not decrypt_pkt.haslayer(IP):
1147 decrypt_pkt = IP(decrypt_pkt[Raw].load)
1148 self.assert_equal(decrypt_pkt.src, self.pg1.remote_ip4)
1149 self.assert_equal(decrypt_pkt.dst, p.remote_tun_if_host4)
1150 self.assert_packet_checksums_valid(decrypt_pkt)
1152 self.logger.debug(ppp("Unexpected packet:", recv_pkt))
1154 self.logger.debug(ppp("Decrypted packet:",
1160 self.logger.info(self.vapi.ppcli("show error"))
1161 self.logger.info(self.vapi.ppcli("show ipsec all"))
1162 self.verify_counters6(p, p, count)
1165 class IpsecTun6Tests(IpsecTun6):
1166 """ UT test methods for Tunnel v6 """
1168 def test_tun_basic66(self):
1169 """ ipsec 6o6 tunnel basic test """
1170 self.verify_tun_66(self.params[socket.AF_INET6], count=1)
1172 def test_tun_reass_basic66(self):
1173 """ ipsec 6o6 tunnel basic reassembly test """
1174 self.verify_tun_reass_66(self.params[socket.AF_INET6])
1176 def test_tun_burst66(self):
1177 """ ipsec 6o6 tunnel burst test """
1178 self.verify_tun_66(self.params[socket.AF_INET6], count=257)
1181 class IpsecTun6HandoffTests(IpsecTun6):
1182 """ UT test methods for Tunnel v6 with multiple workers """
1183 worker_config = "workers 2"
1185 def test_tun_handoff_66(self):
1186 """ ipsec 6o6 tunnel worker hand-off test """
1188 p = self.params[socket.AF_INET6]
1190 # inject alternately on worker 0 and 1. all counts on the SA
1191 # should be against worker 0
1192 for worker in [0, 1, 0, 1]:
1193 send_pkts = self.gen_encrypt_pkts6(p, p.scapy_tun_sa, self.tun_if,
1194 src=p.remote_tun_if_host,
1195 dst=self.pg1.remote_ip6,
1197 recv_pkts = self.send_and_expect(self.tun_if, send_pkts,
1198 self.pg1, worker=worker)
1199 self.verify_decrypted6(p, recv_pkts)
1201 send_pkts = self.gen_pkts6(self.pg1, src=self.pg1.remote_ip6,
1202 dst=p.remote_tun_if_host,
1204 recv_pkts = self.send_and_expect(self.pg1, send_pkts,
1205 self.tun_if, worker=worker)
1206 self.verify_encrypted6(p, p.vpp_tun_sa, recv_pkts)
1208 # all counts against the first worker that was used
1209 self.verify_counters6(p, p, 4*N_PKTS, worker=0)
1212 class IpsecTun4HandoffTests(IpsecTun4):
1213 """ UT test methods for Tunnel v4 with multiple workers """
1214 worker_config = "workers 2"
1216 def test_tun_handooff_44(self):
1217 """ ipsec 4o4 tunnel worker hand-off test """
1219 p = self.params[socket.AF_INET]
1221 # inject alternately on worker 0 and 1. all counts on the SA
1222 # should be against worker 0
1223 for worker in [0, 1, 0, 1]:
1224 send_pkts = self.gen_encrypt_pkts(p, p.scapy_tun_sa, self.tun_if,
1225 src=p.remote_tun_if_host,
1226 dst=self.pg1.remote_ip4,
1228 recv_pkts = self.send_and_expect(self.tun_if, send_pkts,
1229 self.pg1, worker=worker)
1230 self.verify_decrypted(p, recv_pkts)
1232 send_pkts = self.gen_pkts(self.pg1, src=self.pg1.remote_ip4,
1233 dst=p.remote_tun_if_host,
1235 recv_pkts = self.send_and_expect(self.pg1, send_pkts,
1236 self.tun_if, worker=worker)
1237 self.verify_encrypted(p, p.vpp_tun_sa, recv_pkts)
1239 # all counts against the first worker that was used
1240 self.verify_counters4(p, 4*N_PKTS, worker=0)
1243 class IpsecTun46Tests(IpsecTun4Tests, IpsecTun6Tests):
1244 """ UT test methods for Tunnel v6 & v4 """
1248 if __name__ == '__main__':
1249 unittest.main(testRunner=VppTestRunner)