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(object):
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 = 1001
32 self.vpp_tun_sa_id = 200
33 self.vpp_tun_spi = 1000
35 self.scapy_tra_sa_id = 300
36 self.scapy_tra_spi = 2001
37 self.vpp_tra_sa_id = 400
38 self.vpp_tra_spi = 2000
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
54 class IPsecIPv6Params(object):
56 addr_type = socket.AF_INET6
58 addr_bcast = "ffff:ffff:ffff:ffff:ffff:ffff:ffff:ffff"
63 self.remote_tun_if_host = '1111:1111:1111:1111:1111:1111:1111:1111'
64 self.remote_tun_if_host4 = '1.1.1.1'
66 self.scapy_tun_sa_id = 500
67 self.scapy_tun_spi = 3001
68 self.vpp_tun_sa_id = 600
69 self.vpp_tun_spi = 3000
71 self.scapy_tra_sa_id = 700
72 self.scapy_tra_spi = 4001
73 self.vpp_tra_sa_id = 800
74 self.vpp_tra_spi = 4000
76 self.auth_algo_vpp_id = (VppEnum.vl_api_ipsec_integ_alg_t.
77 IPSEC_API_INTEG_ALG_SHA1_96)
78 self.auth_algo = 'HMAC-SHA1-96' # scapy name
79 self.auth_key = b'C91KUR9GYMm5GfkEvNjX'
81 self.crypt_algo_vpp_id = (VppEnum.vl_api_ipsec_crypto_alg_t.
82 IPSEC_API_CRYPTO_ALG_AES_CBC_128)
83 self.crypt_algo = 'AES-CBC' # scapy name
84 self.crypt_key = b'JPjyOWBeVEQiMe7h'
87 self.nat_header = None
90 def mk_scapy_crypt_key(p):
91 if p.crypt_algo == "AES-GCM":
92 return p.crypt_key + struct.pack("!I", p.salt)
97 def config_tun_params(p, encryption_type, tun_if):
98 ip_class_by_addr_type = {socket.AF_INET: IP, socket.AF_INET6: IPv6}
99 esn_en = bool(p.flags & (VppEnum.vl_api_ipsec_sad_flags_t.
100 IPSEC_API_SAD_FLAG_USE_ESN))
101 p.tun_dst = tun_if.remote_addr[p.addr_type]
102 p.tun_src = tun_if.local_addr[p.addr_type]
103 crypt_key = mk_scapy_crypt_key(p)
104 p.scapy_tun_sa = SecurityAssociation(
105 encryption_type, spi=p.vpp_tun_spi,
106 crypt_algo=p.crypt_algo,
108 auth_algo=p.auth_algo, auth_key=p.auth_key,
109 tunnel_header=ip_class_by_addr_type[p.addr_type](
112 nat_t_header=p.nat_header,
114 p.vpp_tun_sa = SecurityAssociation(
115 encryption_type, spi=p.scapy_tun_spi,
116 crypt_algo=p.crypt_algo,
118 auth_algo=p.auth_algo, auth_key=p.auth_key,
119 tunnel_header=ip_class_by_addr_type[p.addr_type](
122 nat_t_header=p.nat_header,
126 def config_tra_params(p, encryption_type):
127 esn_en = bool(p.flags & (VppEnum.vl_api_ipsec_sad_flags_t.
128 IPSEC_API_SAD_FLAG_USE_ESN))
129 crypt_key = mk_scapy_crypt_key(p)
130 p.scapy_tra_sa = SecurityAssociation(
133 crypt_algo=p.crypt_algo,
135 auth_algo=p.auth_algo,
137 nat_t_header=p.nat_header,
139 p.vpp_tra_sa = SecurityAssociation(
142 crypt_algo=p.crypt_algo,
144 auth_algo=p.auth_algo,
146 nat_t_header=p.nat_header,
150 class TemplateIpsec(VppTestCase):
155 |tra_if| <-------> |VPP|
160 ------ encrypt --- plain ---
161 |tun_if| <------- |VPP| <------ |pg1|
164 ------ decrypt --- plain ---
165 |tun_if| -------> |VPP| ------> |pg1|
171 def ipsec_select_backend(self):
172 """ empty method to be overloaded when necessary """
177 super(TemplateIpsec, cls).setUpClass()
180 def tearDownClass(cls):
181 super(TemplateIpsec, cls).tearDownClass()
183 def setup_params(self):
184 self.ipv4_params = IPsecIPv4Params()
185 self.ipv6_params = IPsecIPv6Params()
186 self.params = {self.ipv4_params.addr_type: self.ipv4_params,
187 self.ipv6_params.addr_type: self.ipv6_params}
189 def config_interfaces(self):
190 self.create_pg_interfaces(range(3))
191 self.interfaces = list(self.pg_interfaces)
192 for i in self.interfaces:
200 super(TemplateIpsec, self).setUp()
204 self.vpp_esp_protocol = (VppEnum.vl_api_ipsec_proto_t.
206 self.vpp_ah_protocol = (VppEnum.vl_api_ipsec_proto_t.
209 self.config_interfaces()
211 self.ipsec_select_backend()
213 def unconfig_interfaces(self):
214 for i in self.interfaces:
220 super(TemplateIpsec, self).tearDown()
222 self.unconfig_interfaces()
224 def show_commands_at_teardown(self):
225 self.logger.info(self.vapi.cli("show hardware"))
227 def gen_encrypt_pkts(self, p, sa, sw_intf, src, dst, count=1,
229 return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
230 sa.encrypt(IP(src=src, dst=dst) /
231 ICMP() / Raw(b'X' * payload_size))
232 for i in range(count)]
234 def gen_encrypt_pkts6(self, p, sa, sw_intf, src, dst, count=1,
236 return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
237 sa.encrypt(IPv6(src=src, dst=dst) /
238 ICMPv6EchoRequest(id=0, seq=1,
239 data='X' * payload_size))
240 for i in range(count)]
242 def gen_pkts(self, sw_intf, src, dst, count=1, payload_size=54):
243 return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
244 IP(src=src, dst=dst) / ICMP() / Raw(b'X' * payload_size)
245 for i in range(count)]
247 def gen_pkts6(self, sw_intf, src, dst, count=1, payload_size=54):
248 return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
249 IPv6(src=src, dst=dst) /
250 ICMPv6EchoRequest(id=0, seq=1, data='X' * payload_size)
251 for i in range(count)]
254 class IpsecTcp(object):
255 def verify_tcp_checksum(self):
256 self.vapi.cli("test http server")
257 p = self.params[socket.AF_INET]
258 send = (Ether(src=self.tun_if.remote_mac, dst=self.tun_if.local_mac) /
259 p.scapy_tun_sa.encrypt(IP(src=p.remote_tun_if_host,
260 dst=self.tun_if.local_ip4) /
261 TCP(flags='S', dport=80)))
262 self.logger.debug(ppp("Sending packet:", send))
263 recv = self.send_and_expect(self.tun_if, [send], self.tun_if)
265 decrypted = p.vpp_tun_sa.decrypt(recv[IP])
266 self.assert_packet_checksums_valid(decrypted)
269 class IpsecTcpTests(IpsecTcp):
270 def test_tcp_checksum(self):
271 """ verify checksum correctness for vpp generated packets """
272 self.verify_tcp_checksum()
275 class IpsecTra4(object):
276 """ verify methods for Transport v4 """
277 def verify_tra_anti_replay(self):
278 p = self.params[socket.AF_INET]
279 esn_en = p.vpp_tra_sa.esn_en
281 seq_cycle_node_name = ('/err/%s/sequence number cycled' %
282 self.tra4_encrypt_node_name)
283 replay_node_name = ('/err/%s/SA replayed packet' %
284 self.tra4_decrypt_node_name)
285 if ESP == self.encryption_type and p.crypt_algo == "AES-GCM":
286 hash_failed_node_name = ('/err/%s/ESP decryption failed' %
287 self.tra4_decrypt_node_name)
289 hash_failed_node_name = ('/err/%s/Integrity check failed' %
290 self.tra4_decrypt_node_name)
291 replay_count = self.statistics.get_err_counter(replay_node_name)
292 hash_failed_count = self.statistics.get_err_counter(
293 hash_failed_node_name)
294 seq_cycle_count = self.statistics.get_err_counter(seq_cycle_node_name)
296 if ESP == self.encryption_type:
297 undersize_node_name = ('/err/%s/undersized packet' %
298 self.tra4_decrypt_node_name)
299 undersize_count = self.statistics.get_err_counter(
303 # send packets with seq numbers 1->34
304 # this means the window size is still in Case B (see RFC4303
307 # for reasons i haven't investigated Scapy won't create a packet with
310 pkts = [(Ether(src=self.tra_if.remote_mac,
311 dst=self.tra_if.local_mac) /
312 p.scapy_tra_sa.encrypt(IP(src=self.tra_if.remote_ip4,
313 dst=self.tra_if.local_ip4) /
316 for seq in range(1, 34)]
317 recv_pkts = self.send_and_expect(self.tra_if, pkts, self.tra_if)
319 # replayed packets are dropped
320 self.send_and_assert_no_replies(self.tra_if, pkts)
321 replay_count += len(pkts)
322 self.assert_error_counter_equal(replay_node_name, replay_count)
325 # now send a batch of packets all with the same sequence number
326 # the first packet in the batch is legitimate, the rest bogus
328 pkts = (Ether(src=self.tra_if.remote_mac,
329 dst=self.tra_if.local_mac) /
330 p.scapy_tra_sa.encrypt(IP(src=self.tra_if.remote_ip4,
331 dst=self.tra_if.local_ip4) /
334 recv_pkts = self.send_and_expect(self.tra_if, pkts * 8,
337 self.assert_error_counter_equal(replay_node_name, replay_count)
340 # now move the window over to 257 (more than one byte) and into Case A
342 pkt = (Ether(src=self.tra_if.remote_mac,
343 dst=self.tra_if.local_mac) /
344 p.scapy_tra_sa.encrypt(IP(src=self.tra_if.remote_ip4,
345 dst=self.tra_if.local_ip4) /
348 recv_pkts = self.send_and_expect(self.tra_if, [pkt], self.tra_if)
350 # replayed packets are dropped
351 self.send_and_assert_no_replies(self.tra_if, pkt * 3)
353 self.assert_error_counter_equal(replay_node_name, replay_count)
355 # the window size is 64 packets
356 # in window are still accepted
357 pkt = (Ether(src=self.tra_if.remote_mac,
358 dst=self.tra_if.local_mac) /
359 p.scapy_tra_sa.encrypt(IP(src=self.tra_if.remote_ip4,
360 dst=self.tra_if.local_ip4) /
363 recv_pkts = self.send_and_expect(self.tra_if, [pkt], self.tra_if)
365 # a packet that does not decrypt does not move the window forward
366 bogus_sa = SecurityAssociation(self.encryption_type,
368 crypt_algo=p.crypt_algo,
369 crypt_key=mk_scapy_crypt_key(p)[::-1],
370 auth_algo=p.auth_algo,
371 auth_key=p.auth_key[::-1])
372 pkt = (Ether(src=self.tra_if.remote_mac,
373 dst=self.tra_if.local_mac) /
374 bogus_sa.encrypt(IP(src=self.tra_if.remote_ip4,
375 dst=self.tra_if.local_ip4) /
378 self.send_and_assert_no_replies(self.tra_if, pkt * 17)
380 hash_failed_count += 17
381 self.assert_error_counter_equal(hash_failed_node_name,
384 # a malformed 'runt' packet
385 # created by a mis-constructed SA
386 if (ESP == self.encryption_type and p.crypt_algo != "NULL"):
387 bogus_sa = SecurityAssociation(self.encryption_type,
389 pkt = (Ether(src=self.tra_if.remote_mac,
390 dst=self.tra_if.local_mac) /
391 bogus_sa.encrypt(IP(src=self.tra_if.remote_ip4,
392 dst=self.tra_if.local_ip4) /
395 self.send_and_assert_no_replies(self.tra_if, pkt * 17)
397 undersize_count += 17
398 self.assert_error_counter_equal(undersize_node_name,
401 # which we can determine since this packet is still in the window
402 pkt = (Ether(src=self.tra_if.remote_mac,
403 dst=self.tra_if.local_mac) /
404 p.scapy_tra_sa.encrypt(IP(src=self.tra_if.remote_ip4,
405 dst=self.tra_if.local_ip4) /
408 self.send_and_expect(self.tra_if, [pkt], self.tra_if)
411 # out of window are dropped
412 # this is Case B. So VPP will consider this to be a high seq num wrap
413 # and so the decrypt attempt will fail
415 pkt = (Ether(src=self.tra_if.remote_mac,
416 dst=self.tra_if.local_mac) /
417 p.scapy_tra_sa.encrypt(IP(src=self.tra_if.remote_ip4,
418 dst=self.tra_if.local_ip4) /
421 self.send_and_assert_no_replies(self.tra_if, pkt * 17)
424 # an out of window error with ESN looks like a high sequence
425 # wrap. but since it isn't then the verify will fail.
426 hash_failed_count += 17
427 self.assert_error_counter_equal(hash_failed_node_name,
432 self.assert_error_counter_equal(replay_node_name,
435 # valid packet moves the window over to 258
436 pkt = (Ether(src=self.tra_if.remote_mac,
437 dst=self.tra_if.local_mac) /
438 p.scapy_tra_sa.encrypt(IP(src=self.tra_if.remote_ip4,
439 dst=self.tra_if.local_ip4) /
442 rx = self.send_and_expect(self.tra_if, [pkt], self.tra_if)
443 decrypted = p.vpp_tra_sa.decrypt(rx[0][IP])
446 # move VPP's SA TX seq-num to just before the seq-number wrap.
447 # then fire in a packet that VPP should drop on TX because it
448 # causes the TX seq number to wrap; unless we're using extened sequence
451 self.vapi.cli("test ipsec sa %d seq 0xffffffff" % p.scapy_tra_sa_id)
452 self.logger.info(self.vapi.ppcli("show ipsec sa 0"))
453 self.logger.info(self.vapi.ppcli("show ipsec sa 1"))
455 pkts = [(Ether(src=self.tra_if.remote_mac,
456 dst=self.tra_if.local_mac) /
457 p.scapy_tra_sa.encrypt(IP(src=self.tra_if.remote_ip4,
458 dst=self.tra_if.local_ip4) /
461 for seq in range(259, 280)]
464 rxs = self.send_and_expect(self.tra_if, pkts, self.tra_if)
467 # in order for scapy to decrypt its SA's high order number needs
470 p.vpp_tra_sa.seq_num = 0x100000000
472 decrypted = p.vpp_tra_sa.decrypt(rx[0][IP])
475 # wrap scapy's TX high sequence number. VPP is in case B, so it
476 # will consider this a high seq wrap also.
477 # The low seq num we set it to will place VPP's RX window in Case A
479 p.scapy_tra_sa.seq_num = 0x100000005
480 pkt = (Ether(src=self.tra_if.remote_mac,
481 dst=self.tra_if.local_mac) /
482 p.scapy_tra_sa.encrypt(IP(src=self.tra_if.remote_ip4,
483 dst=self.tra_if.local_ip4) /
485 seq_num=0x100000005))
486 rx = self.send_and_expect(self.tra_if, [pkt], self.tra_if)
487 decrypted = p.vpp_tra_sa.decrypt(rx[0][IP])
490 # A packet that has seq num between (2^32-64) and 5 is within
493 p.scapy_tra_sa.seq_num = 0xfffffffd
494 pkt = (Ether(src=self.tra_if.remote_mac,
495 dst=self.tra_if.local_mac) /
496 p.scapy_tra_sa.encrypt(IP(src=self.tra_if.remote_ip4,
497 dst=self.tra_if.local_ip4) /
500 rx = self.send_and_expect(self.tra_if, [pkt], self.tra_if)
501 decrypted = p.vpp_tra_sa.decrypt(rx[0][IP])
504 # While in case A we cannot wrap the high sequence number again
505 # becuase VPP will consider this packet to be one that moves the
508 pkt = (Ether(src=self.tra_if.remote_mac,
509 dst=self.tra_if.local_mac) /
510 p.scapy_tra_sa.encrypt(IP(src=self.tra_if.remote_ip4,
511 dst=self.tra_if.local_ip4) /
513 seq_num=0x200000999))
514 self.send_and_assert_no_replies(self.tra_if, [pkt], self.tra_if)
516 hash_failed_count += 1
517 self.assert_error_counter_equal(hash_failed_node_name,
521 # but if we move the wondow forward to case B, then we can wrap
524 p.scapy_tra_sa.seq_num = 0x100000555
525 pkt = (Ether(src=self.tra_if.remote_mac,
526 dst=self.tra_if.local_mac) /
527 p.scapy_tra_sa.encrypt(IP(src=self.tra_if.remote_ip4,
528 dst=self.tra_if.local_ip4) /
530 seq_num=0x100000555))
531 rx = self.send_and_expect(self.tra_if, [pkt], self.tra_if)
532 decrypted = p.vpp_tra_sa.decrypt(rx[0][IP])
534 p.scapy_tra_sa.seq_num = 0x200000444
535 pkt = (Ether(src=self.tra_if.remote_mac,
536 dst=self.tra_if.local_mac) /
537 p.scapy_tra_sa.encrypt(IP(src=self.tra_if.remote_ip4,
538 dst=self.tra_if.local_ip4) /
540 seq_num=0x200000444))
541 rx = self.send_and_expect(self.tra_if, [pkt], self.tra_if)
542 decrypted = p.vpp_tra_sa.decrypt(rx[0][IP])
546 # without ESN TX sequence numbers can't wrap and packets are
547 # dropped from here on out.
549 self.send_and_assert_no_replies(self.tra_if, pkts)
550 seq_cycle_count += len(pkts)
551 self.assert_error_counter_equal(seq_cycle_node_name,
554 # move the security-associations seq number on to the last we used
555 self.vapi.cli("test ipsec sa %d seq 0x15f" % p.scapy_tra_sa_id)
556 p.scapy_tra_sa.seq_num = 351
557 p.vpp_tra_sa.seq_num = 351
559 def verify_tra_basic4(self, count=1, payload_size=54):
560 """ ipsec v4 transport basic test """
561 self.vapi.cli("clear errors")
562 self.vapi.cli("clear ipsec sa")
564 p = self.params[socket.AF_INET]
565 send_pkts = self.gen_encrypt_pkts(p, p.scapy_tra_sa, self.tra_if,
566 src=self.tra_if.remote_ip4,
567 dst=self.tra_if.local_ip4,
569 payload_size=payload_size)
570 recv_pkts = self.send_and_expect(self.tra_if, send_pkts,
573 self.assertEqual(len(rx) - len(Ether()), rx[IP].len)
574 self.assert_packet_checksums_valid(rx)
576 decrypted = p.vpp_tra_sa.decrypt(rx[IP])
577 self.assert_packet_checksums_valid(decrypted)
579 self.logger.debug(ppp("Unexpected packet:", rx))
582 self.logger.info(self.vapi.ppcli("show error"))
583 self.logger.info(self.vapi.ppcli("show ipsec all"))
585 pkts = p.tra_sa_in.get_stats()['packets']
586 self.assertEqual(pkts, count,
587 "incorrect SA in counts: expected %d != %d" %
589 pkts = p.tra_sa_out.get_stats()['packets']
590 self.assertEqual(pkts, count,
591 "incorrect SA out counts: expected %d != %d" %
594 self.assert_packet_counter_equal(self.tra4_encrypt_node_name, count)
595 self.assert_packet_counter_equal(self.tra4_decrypt_node_name, count)
598 class IpsecTra4Tests(IpsecTra4):
599 """ UT test methods for Transport v4 """
600 def test_tra_anti_replay(self):
601 """ ipsec v4 transport anti-replay test """
602 self.verify_tra_anti_replay()
604 def test_tra_basic(self, count=1):
605 """ ipsec v4 transport basic test """
606 self.verify_tra_basic4(count=1)
608 def test_tra_burst(self):
609 """ ipsec v4 transport burst test """
610 self.verify_tra_basic4(count=257)
613 class IpsecTra6(object):
614 """ verify methods for Transport v6 """
615 def verify_tra_basic6(self, count=1, payload_size=54):
616 self.vapi.cli("clear errors")
617 self.vapi.cli("clear ipsec sa")
619 p = self.params[socket.AF_INET6]
620 send_pkts = self.gen_encrypt_pkts6(p, p.scapy_tra_sa, self.tra_if,
621 src=self.tra_if.remote_ip6,
622 dst=self.tra_if.local_ip6,
624 payload_size=payload_size)
625 recv_pkts = self.send_and_expect(self.tra_if, send_pkts,
628 self.assertEqual(len(rx) - len(Ether()) - len(IPv6()),
631 decrypted = p.vpp_tra_sa.decrypt(rx[IPv6])
632 self.assert_packet_checksums_valid(decrypted)
634 self.logger.debug(ppp("Unexpected packet:", rx))
637 self.logger.info(self.vapi.ppcli("show error"))
638 self.logger.info(self.vapi.ppcli("show ipsec all"))
640 pkts = p.tra_sa_in.get_stats()['packets']
641 self.assertEqual(pkts, count,
642 "incorrect SA in counts: expected %d != %d" %
644 pkts = p.tra_sa_out.get_stats()['packets']
645 self.assertEqual(pkts, count,
646 "incorrect SA out counts: expected %d != %d" %
648 self.assert_packet_counter_equal(self.tra6_encrypt_node_name, count)
649 self.assert_packet_counter_equal(self.tra6_decrypt_node_name, count)
651 def gen_encrypt_pkts_ext_hdrs6(self, sa, sw_intf, src, dst, count=1,
653 return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
654 sa.encrypt(IPv6(src=src, dst=dst) /
655 ICMPv6EchoRequest(id=0, seq=1,
656 data='X' * payload_size))
657 for i in range(count)]
659 def gen_pkts_ext_hdrs6(self, sw_intf, src, dst, count=1, payload_size=54):
660 return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
661 IPv6(src=src, dst=dst) /
662 IPv6ExtHdrHopByHop() /
663 IPv6ExtHdrFragment(id=2, offset=200) /
665 for i in range(count)]
667 def verify_tra_encrypted6(self, p, sa, rxs):
670 self.assert_packet_checksums_valid(rx)
672 decrypt_pkt = p.vpp_tra_sa.decrypt(rx[IPv6])
673 decrypted.append(decrypt_pkt)
674 self.assert_equal(decrypt_pkt.src, self.tra_if.local_ip6)
675 self.assert_equal(decrypt_pkt.dst, self.tra_if.remote_ip6)
677 self.logger.debug(ppp("Unexpected packet:", rx))
679 self.logger.debug(ppp("Decrypted packet:", decrypt_pkt))
685 def verify_tra_66_ext_hdrs(self, p):
689 # check we can decrypt with options
691 tx = self.gen_encrypt_pkts_ext_hdrs6(p.scapy_tra_sa, self.tra_if,
692 src=self.tra_if.remote_ip6,
693 dst=self.tra_if.local_ip6,
695 self.send_and_expect(self.tra_if, tx, self.tra_if)
698 # injecting a packet from ourselves to be routed of box is a hack
699 # but it matches an outbout policy, alors je ne regrette rien
702 # one extension before ESP
703 tx = (Ether(src=self.pg2.remote_mac, dst=self.pg2.local_mac) /
704 IPv6(src=self.tra_if.local_ip6,
705 dst=self.tra_if.remote_ip6) /
706 IPv6ExtHdrFragment(id=2, offset=200) /
709 rxs = self.send_and_expect(self.pg2, [tx], self.tra_if)
710 dcs = self.verify_tra_encrypted6(p, p.vpp_tra_sa, rxs)
713 # for reasons i'm not going to investigate scapy does not
714 # created the correct headers after decrypt. but reparsing
715 # the ipv6 packet fixes it
716 dc = IPv6(raw(dc[IPv6]))
717 self.assert_equal(dc[IPv6ExtHdrFragment].id, 2)
719 # two extensions before ESP
720 tx = (Ether(src=self.pg2.remote_mac, dst=self.pg2.local_mac) /
721 IPv6(src=self.tra_if.local_ip6,
722 dst=self.tra_if.remote_ip6) /
723 IPv6ExtHdrHopByHop() /
724 IPv6ExtHdrFragment(id=2, offset=200) /
727 rxs = self.send_and_expect(self.pg2, [tx], self.tra_if)
728 dcs = self.verify_tra_encrypted6(p, p.vpp_tra_sa, rxs)
731 dc = IPv6(raw(dc[IPv6]))
732 self.assertTrue(dc[IPv6ExtHdrHopByHop])
733 self.assert_equal(dc[IPv6ExtHdrFragment].id, 2)
735 # two extensions before ESP, one after
736 tx = (Ether(src=self.pg2.remote_mac, dst=self.pg2.local_mac) /
737 IPv6(src=self.tra_if.local_ip6,
738 dst=self.tra_if.remote_ip6) /
739 IPv6ExtHdrHopByHop() /
740 IPv6ExtHdrFragment(id=2, offset=200) /
741 IPv6ExtHdrDestOpt() /
744 rxs = self.send_and_expect(self.pg2, [tx], self.tra_if)
745 dcs = self.verify_tra_encrypted6(p, p.vpp_tra_sa, rxs)
748 dc = IPv6(raw(dc[IPv6]))
749 self.assertTrue(dc[IPv6ExtHdrDestOpt])
750 self.assertTrue(dc[IPv6ExtHdrHopByHop])
751 self.assert_equal(dc[IPv6ExtHdrFragment].id, 2)
754 class IpsecTra6Tests(IpsecTra6):
755 """ UT test methods for Transport v6 """
756 def test_tra_basic6(self):
757 """ ipsec v6 transport basic test """
758 self.verify_tra_basic6(count=1)
760 def test_tra_burst6(self):
761 """ ipsec v6 transport burst test """
762 self.verify_tra_basic6(count=257)
765 class IpsecTra6ExtTests(IpsecTra6):
766 def test_tra_ext_hdrs_66(self):
767 """ ipsec 6o6 tra extension headers test """
768 self.verify_tra_66_ext_hdrs(self.params[socket.AF_INET6])
771 class IpsecTra46Tests(IpsecTra4Tests, IpsecTra6Tests):
772 """ UT test methods for Transport v6 and v4"""
776 class IpsecTun4(object):
777 """ verify methods for Tunnel v4 """
778 def verify_counters4(self, p, count, n_frags=None, worker=None):
781 if (hasattr(p, "spd_policy_in_any")):
782 pkts = p.spd_policy_in_any.get_stats(worker)['packets']
783 self.assertEqual(pkts, count,
784 "incorrect SPD any policy: expected %d != %d" %
787 if (hasattr(p, "tun_sa_in")):
788 pkts = p.tun_sa_in.get_stats(worker)['packets']
789 self.assertEqual(pkts, count,
790 "incorrect SA in counts: expected %d != %d" %
792 pkts = p.tun_sa_out.get_stats(worker)['packets']
793 self.assertEqual(pkts, count,
794 "incorrect SA out counts: expected %d != %d" %
797 self.assert_packet_counter_equal(self.tun4_encrypt_node_name, n_frags)
798 self.assert_packet_counter_equal(self.tun4_decrypt_node_name, count)
800 def verify_decrypted(self, p, rxs):
802 self.assert_equal(rx[IP].src, p.remote_tun_if_host)
803 self.assert_equal(rx[IP].dst, self.pg1.remote_ip4)
804 self.assert_packet_checksums_valid(rx)
806 def verify_esp_padding(self, sa, esp_payload, decrypt_pkt):
807 align = sa.crypt_algo.block_size
810 exp_len = (len(decrypt_pkt) + 2 + (align - 1)) & ~(align - 1)
811 exp_len += sa.crypt_algo.iv_size
812 exp_len += sa.crypt_algo.icv_size or sa.auth_algo.icv_size
813 self.assertEqual(exp_len, len(esp_payload))
815 def verify_encrypted(self, p, sa, rxs):
819 self.assertEqual(rx[UDP].dport, 4500)
820 self.assert_packet_checksums_valid(rx)
821 self.assertEqual(len(rx) - len(Ether()), rx[IP].len)
824 decrypt_pkt = p.vpp_tun_sa.decrypt(rx_ip)
825 if not decrypt_pkt.haslayer(IP):
826 decrypt_pkt = IP(decrypt_pkt[Raw].load)
827 if rx_ip.proto == socket.IPPROTO_ESP:
828 self.verify_esp_padding(sa, rx_ip[ESP].data, decrypt_pkt)
829 decrypt_pkts.append(decrypt_pkt)
830 self.assert_equal(decrypt_pkt.src, self.pg1.remote_ip4)
831 self.assert_equal(decrypt_pkt.dst, p.remote_tun_if_host)
833 self.logger.debug(ppp("Unexpected packet:", rx))
835 self.logger.debug(ppp("Decrypted packet:", decrypt_pkt))
839 pkts = reassemble4(decrypt_pkts)
841 self.assert_packet_checksums_valid(pkt)
843 def verify_tun_44(self, p, count=1, payload_size=64, n_rx=None):
844 self.vapi.cli("clear errors")
845 self.vapi.cli("clear ipsec counters")
846 self.vapi.cli("clear ipsec sa")
850 send_pkts = self.gen_encrypt_pkts(p, p.scapy_tun_sa, self.tun_if,
851 src=p.remote_tun_if_host,
852 dst=self.pg1.remote_ip4,
854 payload_size=payload_size)
855 recv_pkts = self.send_and_expect(self.tun_if, send_pkts, self.pg1)
856 self.verify_decrypted(p, recv_pkts)
858 send_pkts = self.gen_pkts(self.pg1, src=self.pg1.remote_ip4,
859 dst=p.remote_tun_if_host, count=count,
860 payload_size=payload_size)
861 recv_pkts = self.send_and_expect(self.pg1, send_pkts,
863 self.verify_encrypted(p, p.vpp_tun_sa, recv_pkts)
866 self.assertEqual(rx[IP].src, p.tun_src)
867 self.assertEqual(rx[IP].dst, p.tun_dst)
870 self.logger.info(self.vapi.ppcli("show error"))
871 self.logger.info(self.vapi.ppcli("show ipsec all"))
873 self.logger.info(self.vapi.ppcli("show ipsec sa 0"))
874 self.logger.info(self.vapi.ppcli("show ipsec sa 4"))
875 self.verify_counters4(p, count, n_rx)
877 def verify_tun_dropped_44(self, p, count=1, payload_size=64, n_rx=None):
878 self.vapi.cli("clear errors")
882 send_pkts = self.gen_encrypt_pkts(p, p.scapy_tun_sa, self.tun_if,
883 src=p.remote_tun_if_host,
884 dst=self.pg1.remote_ip4,
886 self.send_and_assert_no_replies(self.tun_if, send_pkts)
888 send_pkts = self.gen_pkts(self.pg1, src=self.pg1.remote_ip4,
889 dst=p.remote_tun_if_host, count=count,
890 payload_size=payload_size)
891 self.send_and_assert_no_replies(self.pg1, send_pkts)
894 self.logger.info(self.vapi.ppcli("show error"))
895 self.logger.info(self.vapi.ppcli("show ipsec all"))
897 def verify_tun_reass_44(self, p):
898 self.vapi.cli("clear errors")
899 self.vapi.ip_reassembly_enable_disable(
900 sw_if_index=self.tun_if.sw_if_index, enable_ip4=True)
903 send_pkts = self.gen_encrypt_pkts(p, p.scapy_tun_sa, self.tun_if,
904 src=p.remote_tun_if_host,
905 dst=self.pg1.remote_ip4,
908 send_pkts = fragment_rfc791(send_pkts[0], 1400)
909 recv_pkts = self.send_and_expect(self.tun_if, send_pkts,
911 self.verify_decrypted(p, recv_pkts)
913 send_pkts = self.gen_pkts(self.pg1, src=self.pg1.remote_ip4,
914 dst=p.remote_tun_if_host, count=1)
915 recv_pkts = self.send_and_expect(self.pg1, send_pkts,
917 self.verify_encrypted(p, p.vpp_tun_sa, recv_pkts)
920 self.logger.info(self.vapi.ppcli("show error"))
921 self.logger.info(self.vapi.ppcli("show ipsec all"))
923 self.verify_counters4(p, 1, 1)
924 self.vapi.ip_reassembly_enable_disable(
925 sw_if_index=self.tun_if.sw_if_index, enable_ip4=False)
927 def verify_tun_64(self, p, count=1):
928 self.vapi.cli("clear errors")
929 self.vapi.cli("clear ipsec sa")
931 send_pkts = self.gen_encrypt_pkts6(p, p.scapy_tun_sa, self.tun_if,
932 src=p.remote_tun_if_host6,
933 dst=self.pg1.remote_ip6,
935 recv_pkts = self.send_and_expect(self.tun_if, send_pkts, self.pg1)
936 for recv_pkt in recv_pkts:
937 self.assert_equal(recv_pkt[IPv6].src, p.remote_tun_if_host6)
938 self.assert_equal(recv_pkt[IPv6].dst, self.pg1.remote_ip6)
939 self.assert_packet_checksums_valid(recv_pkt)
940 send_pkts = self.gen_pkts6(self.pg1, src=self.pg1.remote_ip6,
941 dst=p.remote_tun_if_host6, count=count)
942 recv_pkts = self.send_and_expect(self.pg1, send_pkts, self.tun_if)
943 for recv_pkt in recv_pkts:
945 decrypt_pkt = p.vpp_tun_sa.decrypt(recv_pkt[IP])
946 if not decrypt_pkt.haslayer(IPv6):
947 decrypt_pkt = IPv6(decrypt_pkt[Raw].load)
948 self.assert_equal(decrypt_pkt.src, self.pg1.remote_ip6)
949 self.assert_equal(decrypt_pkt.dst, p.remote_tun_if_host6)
950 self.assert_packet_checksums_valid(decrypt_pkt)
952 self.logger.error(ppp("Unexpected packet:", recv_pkt))
955 ppp("Decrypted packet:", decrypt_pkt))
960 self.logger.info(self.vapi.ppcli("show error"))
961 self.logger.info(self.vapi.ppcli("show ipsec all"))
963 self.verify_counters4(p, count)
965 def verify_keepalive(self, p):
966 pkt = (Ether(src=self.tun_if.remote_mac, dst=self.tun_if.local_mac) /
967 IP(src=p.remote_tun_if_host, dst=self.tun_if.local_ip4) /
968 UDP(sport=333, dport=4500) /
970 self.send_and_assert_no_replies(self.tun_if, pkt*31)
971 self.assert_error_counter_equal(
972 '/err/%s/NAT Keepalive' % self.tun4_input_node, 31)
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/Too Short' % self.tun4_input_node, 31)
983 class IpsecTun4Tests(IpsecTun4):
984 """ UT test methods for Tunnel v4 """
985 def test_tun_basic44(self):
986 """ ipsec 4o4 tunnel basic test """
987 self.verify_tun_44(self.params[socket.AF_INET], count=1)
988 self.tun_if.admin_down()
989 self.tun_if.resolve_arp()
990 self.tun_if.admin_up()
991 self.verify_tun_44(self.params[socket.AF_INET], count=1)
993 def test_tun_reass_basic44(self):
994 """ ipsec 4o4 tunnel basic reassembly test """
995 self.verify_tun_reass_44(self.params[socket.AF_INET])
997 def test_tun_burst44(self):
998 """ ipsec 4o4 tunnel burst test """
999 self.verify_tun_44(self.params[socket.AF_INET], count=127)
1002 class IpsecTun6(object):
1003 """ verify methods for Tunnel v6 """
1004 def verify_counters6(self, p_in, p_out, count, worker=None):
1005 if (hasattr(p_in, "tun_sa_in")):
1006 pkts = p_in.tun_sa_in.get_stats(worker)['packets']
1007 self.assertEqual(pkts, count,
1008 "incorrect SA in counts: expected %d != %d" %
1010 if (hasattr(p_out, "tun_sa_out")):
1011 pkts = p_out.tun_sa_out.get_stats(worker)['packets']
1012 self.assertEqual(pkts, count,
1013 "incorrect SA out counts: expected %d != %d" %
1015 self.assert_packet_counter_equal(self.tun6_encrypt_node_name, count)
1016 self.assert_packet_counter_equal(self.tun6_decrypt_node_name, count)
1018 def verify_decrypted6(self, p, rxs):
1020 self.assert_equal(rx[IPv6].src, p.remote_tun_if_host)
1021 self.assert_equal(rx[IPv6].dst, self.pg1.remote_ip6)
1022 self.assert_packet_checksums_valid(rx)
1024 def verify_encrypted6(self, p, sa, rxs):
1026 self.assert_packet_checksums_valid(rx)
1027 self.assertEqual(len(rx) - len(Ether()) - len(IPv6()),
1030 decrypt_pkt = p.vpp_tun_sa.decrypt(rx[IPv6])
1031 if not decrypt_pkt.haslayer(IPv6):
1032 decrypt_pkt = IPv6(decrypt_pkt[Raw].load)
1033 self.assert_packet_checksums_valid(decrypt_pkt)
1034 self.assert_equal(decrypt_pkt.src, self.pg1.remote_ip6)
1035 self.assert_equal(decrypt_pkt.dst, p.remote_tun_if_host)
1037 self.logger.debug(ppp("Unexpected packet:", rx))
1039 self.logger.debug(ppp("Decrypted packet:", decrypt_pkt))
1044 def verify_drop_tun_66(self, p_in, count=1, payload_size=64):
1045 self.vapi.cli("clear errors")
1046 self.vapi.cli("clear ipsec sa")
1048 send_pkts = self.gen_encrypt_pkts6(p_in, p_in.scapy_tun_sa,
1050 src=p_in.remote_tun_if_host,
1051 dst=self.pg1.remote_ip6,
1053 self.send_and_assert_no_replies(self.tun_if, send_pkts)
1054 self.logger.info(self.vapi.cli("sh punt stats"))
1056 def verify_tun_66(self, p_in, p_out=None, count=1, payload_size=64):
1057 self.vapi.cli("clear errors")
1058 self.vapi.cli("clear ipsec sa")
1062 send_pkts = self.gen_encrypt_pkts6(p_in, p_in.scapy_tun_sa,
1064 src=p_in.remote_tun_if_host,
1065 dst=self.pg1.remote_ip6,
1067 payload_size=payload_size)
1068 recv_pkts = self.send_and_expect(self.tun_if, send_pkts, self.pg1)
1069 self.verify_decrypted6(p_in, recv_pkts)
1071 send_pkts = self.gen_pkts6(self.pg1, src=self.pg1.remote_ip6,
1072 dst=p_out.remote_tun_if_host,
1074 payload_size=payload_size)
1075 recv_pkts = self.send_and_expect(self.pg1, send_pkts, self.tun_if)
1076 self.verify_encrypted6(p_out, p_out.vpp_tun_sa, recv_pkts)
1078 for rx in recv_pkts:
1079 self.assertEqual(rx[IPv6].src, p_out.tun_src)
1080 self.assertEqual(rx[IPv6].dst, p_out.tun_dst)
1083 self.logger.info(self.vapi.ppcli("show error"))
1084 self.logger.info(self.vapi.ppcli("show ipsec all"))
1085 self.verify_counters6(p_in, p_out, count)
1087 def verify_tun_reass_66(self, p):
1088 self.vapi.cli("clear errors")
1089 self.vapi.ip_reassembly_enable_disable(
1090 sw_if_index=self.tun_if.sw_if_index, enable_ip6=True)
1093 send_pkts = self.gen_encrypt_pkts6(p, p.scapy_tun_sa, self.tun_if,
1094 src=p.remote_tun_if_host,
1095 dst=self.pg1.remote_ip6,
1098 send_pkts = fragment_rfc8200(send_pkts[0], 1, 1400, self.logger)
1099 recv_pkts = self.send_and_expect(self.tun_if, send_pkts,
1101 self.verify_decrypted6(p, recv_pkts)
1103 send_pkts = self.gen_pkts6(self.pg1, src=self.pg1.remote_ip6,
1104 dst=p.remote_tun_if_host,
1107 recv_pkts = self.send_and_expect(self.pg1, send_pkts,
1109 self.verify_encrypted6(p, p.vpp_tun_sa, recv_pkts)
1111 self.logger.info(self.vapi.ppcli("show error"))
1112 self.logger.info(self.vapi.ppcli("show ipsec all"))
1113 self.verify_counters6(p, p, 1)
1114 self.vapi.ip_reassembly_enable_disable(
1115 sw_if_index=self.tun_if.sw_if_index, enable_ip6=False)
1117 def verify_tun_46(self, p, count=1):
1118 """ ipsec 4o6 tunnel basic test """
1119 self.vapi.cli("clear errors")
1120 self.vapi.cli("clear ipsec sa")
1122 send_pkts = self.gen_encrypt_pkts(p, p.scapy_tun_sa, self.tun_if,
1123 src=p.remote_tun_if_host4,
1124 dst=self.pg1.remote_ip4,
1126 recv_pkts = self.send_and_expect(self.tun_if, send_pkts, self.pg1)
1127 for recv_pkt in recv_pkts:
1128 self.assert_equal(recv_pkt[IP].src, p.remote_tun_if_host4)
1129 self.assert_equal(recv_pkt[IP].dst, self.pg1.remote_ip4)
1130 self.assert_packet_checksums_valid(recv_pkt)
1131 send_pkts = self.gen_pkts(self.pg1, src=self.pg1.remote_ip4,
1132 dst=p.remote_tun_if_host4,
1134 recv_pkts = self.send_and_expect(self.pg1, send_pkts, self.tun_if)
1135 for recv_pkt in recv_pkts:
1137 decrypt_pkt = p.vpp_tun_sa.decrypt(recv_pkt[IPv6])
1138 if not decrypt_pkt.haslayer(IP):
1139 decrypt_pkt = IP(decrypt_pkt[Raw].load)
1140 self.assert_equal(decrypt_pkt.src, self.pg1.remote_ip4)
1141 self.assert_equal(decrypt_pkt.dst, p.remote_tun_if_host4)
1142 self.assert_packet_checksums_valid(decrypt_pkt)
1144 self.logger.debug(ppp("Unexpected packet:", recv_pkt))
1146 self.logger.debug(ppp("Decrypted packet:",
1152 self.logger.info(self.vapi.ppcli("show error"))
1153 self.logger.info(self.vapi.ppcli("show ipsec all"))
1154 self.verify_counters6(p, p, count)
1157 class IpsecTun6Tests(IpsecTun6):
1158 """ UT test methods for Tunnel v6 """
1160 def test_tun_basic66(self):
1161 """ ipsec 6o6 tunnel basic test """
1162 self.verify_tun_66(self.params[socket.AF_INET6], count=1)
1164 def test_tun_reass_basic66(self):
1165 """ ipsec 6o6 tunnel basic reassembly test """
1166 self.verify_tun_reass_66(self.params[socket.AF_INET6])
1168 def test_tun_burst66(self):
1169 """ ipsec 6o6 tunnel burst test """
1170 self.verify_tun_66(self.params[socket.AF_INET6], count=257)
1173 class IpsecTun6HandoffTests(IpsecTun6):
1174 """ UT test methods for Tunnel v6 with multiple workers """
1175 worker_config = "workers 2"
1177 def test_tun_handoff_66(self):
1178 """ ipsec 6o6 tunnel worker hand-off test """
1180 p = self.params[socket.AF_INET6]
1182 # inject alternately on worker 0 and 1. all counts on the SA
1183 # should be against worker 0
1184 for worker in [0, 1, 0, 1]:
1185 send_pkts = self.gen_encrypt_pkts6(p, p.scapy_tun_sa, self.tun_if,
1186 src=p.remote_tun_if_host,
1187 dst=self.pg1.remote_ip6,
1189 recv_pkts = self.send_and_expect(self.tun_if, send_pkts,
1190 self.pg1, worker=worker)
1191 self.verify_decrypted6(p, recv_pkts)
1193 send_pkts = self.gen_pkts6(self.pg1, src=self.pg1.remote_ip6,
1194 dst=p.remote_tun_if_host,
1196 recv_pkts = self.send_and_expect(self.pg1, send_pkts,
1197 self.tun_if, worker=worker)
1198 self.verify_encrypted6(p, p.vpp_tun_sa, recv_pkts)
1200 # all counts against the first worker that was used
1201 self.verify_counters6(p, p, 4*N_PKTS, worker=0)
1204 class IpsecTun4HandoffTests(IpsecTun4):
1205 """ UT test methods for Tunnel v4 with multiple workers """
1206 worker_config = "workers 2"
1208 def test_tun_handooff_44(self):
1209 """ ipsec 4o4 tunnel worker hand-off test """
1211 p = self.params[socket.AF_INET]
1213 # inject alternately on worker 0 and 1. all counts on the SA
1214 # should be against worker 0
1215 for worker in [0, 1, 0, 1]:
1216 send_pkts = self.gen_encrypt_pkts(p, p.scapy_tun_sa, self.tun_if,
1217 src=p.remote_tun_if_host,
1218 dst=self.pg1.remote_ip4,
1220 recv_pkts = self.send_and_expect(self.tun_if, send_pkts,
1221 self.pg1, worker=worker)
1222 self.verify_decrypted(p, recv_pkts)
1224 send_pkts = self.gen_pkts(self.pg1, src=self.pg1.remote_ip4,
1225 dst=p.remote_tun_if_host,
1227 recv_pkts = self.send_and_expect(self.pg1, send_pkts,
1228 self.tun_if, worker=worker)
1229 self.verify_encrypted(p, p.vpp_tun_sa, recv_pkts)
1231 # all counts against the first worker that was used
1232 self.verify_counters4(p, 4*N_PKTS, worker=0)
1235 class IpsecTun46Tests(IpsecTun4Tests, IpsecTun6Tests):
1236 """ UT test methods for Tunnel v6 & v4 """
1240 if __name__ == '__main__':
1241 unittest.main(testRunner=VppTestRunner)