5 from scapy.layers.ipsec import SecurityAssociation, ESP
6 from scapy.layers.l2 import Ether, Raw, GRE
7 from scapy.layers.inet import IP, UDP
8 from scapy.layers.inet6 import IPv6
9 from framework import VppTestRunner
10 from template_ipsec import TemplateIpsec, IpsecTun4Tests, IpsecTun6Tests, \
11 IpsecTun4, IpsecTun6, IpsecTcpTests, mk_scapy_crypt_key
12 from vpp_ipsec_tun_interface import VppIpsecTunInterface
13 from vpp_gre_interface import VppGreInterface
14 from vpp_ipip_tun_interface import VppIpIpTunInterface
15 from vpp_ip_route import VppIpRoute, VppRoutePath, DpoProto
16 from vpp_ipsec import VppIpsecSA, VppIpsecTunProtect
17 from vpp_l2 import VppBridgeDomain, VppBridgeDomainPort
19 from vpp_papi import VppEnum
22 def config_tun_params(p, encryption_type, tun_if):
23 ip_class_by_addr_type = {socket.AF_INET: IP, socket.AF_INET6: IPv6}
24 use_esn = bool(p.flags & (VppEnum.vl_api_ipsec_sad_flags_t.
25 IPSEC_API_SAD_FLAG_USE_ESN))
26 crypt_key = mk_scapy_crypt_key(p)
27 p.scapy_tun_sa = SecurityAssociation(
28 encryption_type, spi=p.vpp_tun_spi,
29 crypt_algo=p.crypt_algo,
31 auth_algo=p.auth_algo, auth_key=p.auth_key,
32 tunnel_header=ip_class_by_addr_type[p.addr_type](
35 nat_t_header=p.nat_header,
37 p.vpp_tun_sa = SecurityAssociation(
38 encryption_type, spi=p.scapy_tun_spi,
39 crypt_algo=p.crypt_algo,
41 auth_algo=p.auth_algo, auth_key=p.auth_key,
42 tunnel_header=ip_class_by_addr_type[p.addr_type](
45 nat_t_header=p.nat_header,
49 class TemplateIpsec4TunIfEsp(TemplateIpsec):
50 """ IPsec tunnel interface tests """
56 super(TemplateIpsec4TunIfEsp, cls).setUpClass()
59 def tearDownClass(cls):
60 super(TemplateIpsec4TunIfEsp, cls).tearDownClass()
63 super(TemplateIpsec4TunIfEsp, self).setUp()
65 self.tun_if = self.pg0
69 p.tun_if = VppIpsecTunInterface(self, self.pg0, p.vpp_tun_spi,
70 p.scapy_tun_spi, p.crypt_algo_vpp_id,
71 p.crypt_key, p.crypt_key,
72 p.auth_algo_vpp_id, p.auth_key,
74 p.tun_if.add_vpp_config()
78 config_tun_params(p, self.encryption_type, p.tun_if)
80 r = VppIpRoute(self, p.remote_tun_if_host, 32,
81 [VppRoutePath(p.tun_if.remote_ip4,
84 r = VppIpRoute(self, p.remote_tun_if_host6, 128,
85 [VppRoutePath(p.tun_if.remote_ip6,
87 proto=DpoProto.DPO_PROTO_IP6)])
91 super(TemplateIpsec4TunIfEsp, self).tearDown()
94 class TemplateIpsec4TunIfEspUdp(TemplateIpsec):
95 """ IPsec UDP tunnel interface tests """
97 tun4_encrypt_node_name = "esp4-encrypt-tun"
98 tun4_decrypt_node_name = "esp4-decrypt-tun"
103 super(TemplateIpsec4TunIfEspUdp, cls).setUpClass()
106 def tearDownClass(cls):
107 super(TemplateIpsec4TunIfEspUdp, cls).tearDownClass()
110 super(TemplateIpsec4TunIfEspUdp, self).setUp()
112 self.tun_if = self.pg0
115 p.flags = (VppEnum.vl_api_ipsec_sad_flags_t.
116 IPSEC_API_SAD_FLAG_UDP_ENCAP)
117 p.nat_header = UDP(sport=5454, dport=4500)
119 p.tun_if = VppIpsecTunInterface(self, self.pg0, p.vpp_tun_spi,
120 p.scapy_tun_spi, p.crypt_algo_vpp_id,
121 p.crypt_key, p.crypt_key,
122 p.auth_algo_vpp_id, p.auth_key,
123 p.auth_key, udp_encap=True)
124 p.tun_if.add_vpp_config()
126 p.tun_if.config_ip4()
127 p.tun_if.config_ip6()
128 config_tun_params(p, self.encryption_type, p.tun_if)
130 r = VppIpRoute(self, p.remote_tun_if_host, 32,
131 [VppRoutePath(p.tun_if.remote_ip4,
134 r = VppIpRoute(self, p.remote_tun_if_host6, 128,
135 [VppRoutePath(p.tun_if.remote_ip6,
137 proto=DpoProto.DPO_PROTO_IP6)])
141 super(TemplateIpsec4TunIfEspUdp, self).tearDown()
144 class TestIpsec4TunIfEsp1(TemplateIpsec4TunIfEsp, IpsecTun4Tests):
145 """ Ipsec ESP - TUN tests """
146 tun4_encrypt_node_name = "esp4-encrypt-tun"
147 tun4_decrypt_node_name = "esp4-decrypt-tun"
149 def test_tun_basic64(self):
150 """ ipsec 6o4 tunnel basic test """
151 self.tun4_encrypt_node_name = "esp4-encrypt-tun"
153 self.verify_tun_64(self.params[socket.AF_INET], count=1)
155 def test_tun_burst64(self):
156 """ ipsec 6o4 tunnel basic test """
157 self.tun4_encrypt_node_name = "esp4-encrypt-tun"
159 self.verify_tun_64(self.params[socket.AF_INET], count=257)
161 def test_tun_basic_frag44(self):
162 """ ipsec 4o4 tunnel frag basic test """
163 self.tun4_encrypt_node_name = "esp4-encrypt-tun"
167 self.vapi.sw_interface_set_mtu(p.tun_if.sw_if_index,
169 self.verify_tun_44(self.params[socket.AF_INET],
170 count=1, payload_size=1800, n_rx=2)
171 self.vapi.sw_interface_set_mtu(p.tun_if.sw_if_index,
175 class TestIpsec4TunIfEspUdp(TemplateIpsec4TunIfEspUdp, IpsecTun4Tests):
176 """ Ipsec ESP UDP tests """
178 tun4_input_node = "ipsec4-tun-input"
180 def test_keepalive(self):
181 """ IPSEC NAT Keepalive """
182 self.verify_keepalive(self.ipv4_params)
185 class TestIpsec4TunIfEsp2(TemplateIpsec4TunIfEsp, IpsecTcpTests):
186 """ Ipsec ESP - TCP tests """
190 class TemplateIpsec6TunIfEsp(TemplateIpsec):
191 """ IPsec tunnel interface tests """
193 encryption_type = ESP
196 super(TemplateIpsec6TunIfEsp, self).setUp()
198 self.tun_if = self.pg0
201 p.tun_if = VppIpsecTunInterface(self, self.pg0, p.vpp_tun_spi,
202 p.scapy_tun_spi, p.crypt_algo_vpp_id,
203 p.crypt_key, p.crypt_key,
204 p.auth_algo_vpp_id, p.auth_key,
205 p.auth_key, is_ip6=True)
206 p.tun_if.add_vpp_config()
208 p.tun_if.config_ip6()
209 p.tun_if.config_ip4()
210 config_tun_params(p, self.encryption_type, p.tun_if)
212 r = VppIpRoute(self, p.remote_tun_if_host, 128,
213 [VppRoutePath(p.tun_if.remote_ip6,
215 proto=DpoProto.DPO_PROTO_IP6)])
217 r = VppIpRoute(self, p.remote_tun_if_host4, 32,
218 [VppRoutePath(p.tun_if.remote_ip4,
223 super(TemplateIpsec6TunIfEsp, self).tearDown()
226 class TestIpsec6TunIfEsp1(TemplateIpsec6TunIfEsp, IpsecTun6Tests):
227 """ Ipsec ESP - TUN tests """
228 tun6_encrypt_node_name = "esp6-encrypt-tun"
229 tun6_decrypt_node_name = "esp6-decrypt-tun"
231 def test_tun_basic46(self):
232 """ ipsec 4o6 tunnel basic test """
233 self.tun6_encrypt_node_name = "esp6-encrypt-tun"
234 self.verify_tun_46(self.params[socket.AF_INET6], count=1)
236 def test_tun_burst46(self):
237 """ ipsec 4o6 tunnel burst test """
238 self.tun6_encrypt_node_name = "esp6-encrypt-tun"
239 self.verify_tun_46(self.params[socket.AF_INET6], count=257)
242 class TestIpsec4MultiTunIfEsp(TemplateIpsec, IpsecTun4):
243 """ IPsec IPv4 Multi Tunnel interface """
245 encryption_type = ESP
246 tun4_encrypt_node_name = "esp4-encrypt-tun"
247 tun4_decrypt_node_name = "esp4-decrypt-tun"
250 super(TestIpsec4MultiTunIfEsp, self).setUp()
252 self.tun_if = self.pg0
254 self.multi_params = []
255 self.pg0.generate_remote_hosts(10)
256 self.pg0.configure_ipv4_neighbors()
259 p = copy.copy(self.ipv4_params)
261 p.remote_tun_if_host = "1.1.1.%d" % (ii + 1)
262 p.scapy_tun_sa_id = p.scapy_tun_sa_id + ii
263 p.scapy_tun_spi = p.scapy_tun_spi + ii
264 p.vpp_tun_sa_id = p.vpp_tun_sa_id + ii
265 p.vpp_tun_spi = p.vpp_tun_spi + ii
267 p.scapy_tra_sa_id = p.scapy_tra_sa_id + ii
268 p.scapy_tra_spi = p.scapy_tra_spi + ii
269 p.vpp_tra_sa_id = p.vpp_tra_sa_id + ii
270 p.vpp_tra_spi = p.vpp_tra_spi + ii
272 p.tun_if = VppIpsecTunInterface(self, self.pg0, p.vpp_tun_spi,
275 p.crypt_key, p.crypt_key,
276 p.auth_algo_vpp_id, p.auth_key,
278 dst=self.pg0.remote_hosts[ii].ip4)
279 p.tun_if.add_vpp_config()
281 p.tun_if.config_ip4()
282 config_tun_params(p, self.encryption_type, p.tun_if)
283 self.multi_params.append(p)
285 VppIpRoute(self, p.remote_tun_if_host, 32,
286 [VppRoutePath(p.tun_if.remote_ip4,
287 0xffffffff)]).add_vpp_config()
290 super(TestIpsec4MultiTunIfEsp, self).tearDown()
292 def test_tun_44(self):
293 """Multiple IPSEC tunnel interfaces """
294 for p in self.multi_params:
295 self.verify_tun_44(p, count=127)
296 c = p.tun_if.get_rx_stats()
297 self.assertEqual(c['packets'], 127)
298 c = p.tun_if.get_tx_stats()
299 self.assertEqual(c['packets'], 127)
302 class TestIpsec4TunIfEspAll(TemplateIpsec, IpsecTun4):
303 """ IPsec IPv4 Tunnel interface all Algos """
305 encryption_type = ESP
306 tun4_encrypt_node_name = "esp4-encrypt-tun"
307 tun4_decrypt_node_name = "esp4-decrypt-tun"
309 def config_network(self, p):
311 p.tun_if = VppIpsecTunInterface(self, self.pg0, p.vpp_tun_spi,
314 p.crypt_key, p.crypt_key,
315 p.auth_algo_vpp_id, p.auth_key,
318 p.tun_if.add_vpp_config()
320 p.tun_if.config_ip4()
321 config_tun_params(p, self.encryption_type, p.tun_if)
322 self.logger.info(self.vapi.cli("sh ipsec sa 0"))
323 self.logger.info(self.vapi.cli("sh ipsec sa 1"))
325 p.route = VppIpRoute(self, p.remote_tun_if_host, 32,
326 [VppRoutePath(p.tun_if.remote_ip4,
328 p.route.add_vpp_config()
330 def unconfig_network(self, p):
331 p.tun_if.unconfig_ip4()
332 p.tun_if.remove_vpp_config()
333 p.route.remove_vpp_config()
336 super(TestIpsec4TunIfEspAll, self).setUp()
338 self.tun_if = self.pg0
341 super(TestIpsec4TunIfEspAll, self).tearDown()
345 # change the key and the SPI
347 p.crypt_key = b'X' + p.crypt_key[1:]
349 p.scapy_tun_sa_id += 1
352 p.tun_if.local_spi = p.vpp_tun_spi
353 p.tun_if.remote_spi = p.scapy_tun_spi
355 config_tun_params(p, self.encryption_type, p.tun_if)
357 p.tun_sa_in = VppIpsecSA(self,
364 self.vpp_esp_protocol,
367 p.tun_sa_out = VppIpsecSA(self,
374 self.vpp_esp_protocol,
377 p.tun_sa_in.add_vpp_config()
378 p.tun_sa_out.add_vpp_config()
380 self.vapi.ipsec_tunnel_if_set_sa(sw_if_index=p.tun_if.sw_if_index,
381 sa_id=p.tun_sa_in.id,
383 self.vapi.ipsec_tunnel_if_set_sa(sw_if_index=p.tun_if.sw_if_index,
384 sa_id=p.tun_sa_out.id,
386 self.logger.info(self.vapi.cli("sh ipsec sa"))
388 def test_tun_44(self):
389 """IPSEC tunnel all algos """
391 # foreach VPP crypto engine
392 engines = ["ia32", "ipsecmb", "openssl"]
394 # foreach crypto algorithm
395 algos = [{'vpp-crypto': (VppEnum.vl_api_ipsec_crypto_alg_t.
396 IPSEC_API_CRYPTO_ALG_AES_GCM_128),
397 'vpp-integ': (VppEnum.vl_api_ipsec_integ_alg_t.
398 IPSEC_API_INTEG_ALG_NONE),
399 'scapy-crypto': "AES-GCM",
400 'scapy-integ': "NULL",
401 'key': b"JPjyOWBeVEQiMe7h",
403 {'vpp-crypto': (VppEnum.vl_api_ipsec_crypto_alg_t.
404 IPSEC_API_CRYPTO_ALG_AES_GCM_192),
405 'vpp-integ': (VppEnum.vl_api_ipsec_integ_alg_t.
406 IPSEC_API_INTEG_ALG_NONE),
407 'scapy-crypto': "AES-GCM",
408 'scapy-integ': "NULL",
409 'key': b"JPjyOWBeVEQiMe7hJPjyOWBe",
411 {'vpp-crypto': (VppEnum.vl_api_ipsec_crypto_alg_t.
412 IPSEC_API_CRYPTO_ALG_AES_GCM_256),
413 'vpp-integ': (VppEnum.vl_api_ipsec_integ_alg_t.
414 IPSEC_API_INTEG_ALG_NONE),
415 'scapy-crypto': "AES-GCM",
416 'scapy-integ': "NULL",
417 'key': b"JPjyOWBeVEQiMe7hJPjyOWBeVEQiMe7h",
419 {'vpp-crypto': (VppEnum.vl_api_ipsec_crypto_alg_t.
420 IPSEC_API_CRYPTO_ALG_AES_CBC_128),
421 'vpp-integ': (VppEnum.vl_api_ipsec_integ_alg_t.
422 IPSEC_API_INTEG_ALG_SHA1_96),
423 'scapy-crypto': "AES-CBC",
424 'scapy-integ': "HMAC-SHA1-96",
426 'key': b"JPjyOWBeVEQiMe7h"},
427 {'vpp-crypto': (VppEnum.vl_api_ipsec_crypto_alg_t.
428 IPSEC_API_CRYPTO_ALG_AES_CBC_192),
429 'vpp-integ': (VppEnum.vl_api_ipsec_integ_alg_t.
430 IPSEC_API_INTEG_ALG_SHA1_96),
431 'scapy-crypto': "AES-CBC",
432 'scapy-integ': "HMAC-SHA1-96",
434 'key': b"JPjyOWBeVEQiMe7hJPjyOWBe"},
435 {'vpp-crypto': (VppEnum.vl_api_ipsec_crypto_alg_t.
436 IPSEC_API_CRYPTO_ALG_AES_CBC_256),
437 'vpp-integ': (VppEnum.vl_api_ipsec_integ_alg_t.
438 IPSEC_API_INTEG_ALG_SHA1_96),
439 'scapy-crypto': "AES-CBC",
440 'scapy-integ': "HMAC-SHA1-96",
442 'key': b"JPjyOWBeVEQiMe7hJPjyOWBeVEQiMe7h"},
443 {'vpp-crypto': (VppEnum.vl_api_ipsec_crypto_alg_t.
444 IPSEC_API_CRYPTO_ALG_NONE),
445 'vpp-integ': (VppEnum.vl_api_ipsec_integ_alg_t.
446 IPSEC_API_INTEG_ALG_SHA1_96),
447 'scapy-crypto': "NULL",
448 'scapy-integ': "HMAC-SHA1-96",
450 'key': b"JPjyOWBeVEQiMe7hJPjyOWBeVEQiMe7h"}]
452 for engine in engines:
453 self.vapi.cli("set crypto handler all %s" % engine)
456 # loop through each of the algorithms
459 # with self.subTest(algo=algo['scapy']):
461 p = copy.copy(self.ipv4_params)
462 p.auth_algo_vpp_id = algo['vpp-integ']
463 p.crypt_algo_vpp_id = algo['vpp-crypto']
464 p.crypt_algo = algo['scapy-crypto']
465 p.auth_algo = algo['scapy-integ']
466 p.crypt_key = algo['key']
467 p.salt = algo['salt']
469 self.config_network(p)
471 self.verify_tun_44(p, count=127)
472 c = p.tun_if.get_rx_stats()
473 self.assertEqual(c['packets'], 127)
474 c = p.tun_if.get_tx_stats()
475 self.assertEqual(c['packets'], 127)
481 self.verify_tun_44(p, count=127)
483 self.unconfig_network(p)
484 p.tun_sa_out.remove_vpp_config()
485 p.tun_sa_in.remove_vpp_config()
488 class TestIpsec6MultiTunIfEsp(TemplateIpsec, IpsecTun6):
489 """ IPsec IPv6 Multi Tunnel interface """
491 encryption_type = ESP
492 tun6_encrypt_node_name = "esp6-encrypt-tun"
493 tun6_decrypt_node_name = "esp6-decrypt-tun"
496 super(TestIpsec6MultiTunIfEsp, self).setUp()
498 self.tun_if = self.pg0
500 self.multi_params = []
501 self.pg0.generate_remote_hosts(10)
502 self.pg0.configure_ipv6_neighbors()
505 p = copy.copy(self.ipv6_params)
507 p.remote_tun_if_host = "1111::%d" % (ii + 1)
508 p.scapy_tun_sa_id = p.scapy_tun_sa_id + ii
509 p.scapy_tun_spi = p.scapy_tun_spi + ii
510 p.vpp_tun_sa_id = p.vpp_tun_sa_id + ii
511 p.vpp_tun_spi = p.vpp_tun_spi + ii
513 p.scapy_tra_sa_id = p.scapy_tra_sa_id + ii
514 p.scapy_tra_spi = p.scapy_tra_spi + ii
515 p.vpp_tra_sa_id = p.vpp_tra_sa_id + ii
516 p.vpp_tra_spi = p.vpp_tra_spi + ii
518 p.tun_if = VppIpsecTunInterface(self, self.pg0, p.vpp_tun_spi,
521 p.crypt_key, p.crypt_key,
522 p.auth_algo_vpp_id, p.auth_key,
523 p.auth_key, is_ip6=True,
524 dst=self.pg0.remote_hosts[ii].ip6)
525 p.tun_if.add_vpp_config()
527 p.tun_if.config_ip6()
528 config_tun_params(p, self.encryption_type, p.tun_if)
529 self.multi_params.append(p)
531 r = VppIpRoute(self, p.remote_tun_if_host, 128,
532 [VppRoutePath(p.tun_if.remote_ip6,
534 proto=DpoProto.DPO_PROTO_IP6)])
538 super(TestIpsec6MultiTunIfEsp, self).tearDown()
540 def test_tun_66(self):
541 """Multiple IPSEC tunnel interfaces """
542 for p in self.multi_params:
543 self.verify_tun_66(p, count=127)
544 c = p.tun_if.get_rx_stats()
545 self.assertEqual(c['packets'], 127)
546 c = p.tun_if.get_tx_stats()
547 self.assertEqual(c['packets'], 127)
550 class TestIpsecGreTebIfEsp(TemplateIpsec,
552 """ Ipsec GRE TEB ESP - TUN tests """
553 tun4_encrypt_node_name = "esp4-encrypt-tun"
554 tun4_decrypt_node_name = "esp4-decrypt-tun"
555 encryption_type = ESP
556 omac = "00:11:22:33:44:55"
558 def gen_encrypt_pkts(self, sa, sw_intf, src, dst, count=1,
560 return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
561 sa.encrypt(IP(src=self.pg0.remote_ip4,
562 dst=self.pg0.local_ip4) /
564 Ether(dst=self.omac) /
565 IP(src="1.1.1.1", dst="1.1.1.2") /
566 UDP(sport=1144, dport=2233) /
567 Raw(b'X' * payload_size))
568 for i in range(count)]
570 def gen_pkts(self, sw_intf, src, dst, count=1,
572 return [Ether(dst=self.omac) /
573 IP(src="1.1.1.1", dst="1.1.1.2") /
574 UDP(sport=1144, dport=2233) /
575 Raw(b'X' * payload_size)
576 for i in range(count)]
578 def verify_decrypted(self, p, rxs):
580 self.assert_equal(rx[Ether].dst, self.omac)
581 self.assert_equal(rx[IP].dst, "1.1.1.2")
583 def verify_encrypted(self, p, sa, rxs):
586 pkt = sa.decrypt(rx[IP])
587 if not pkt.haslayer(IP):
588 pkt = IP(pkt[Raw].load)
589 self.assert_packet_checksums_valid(pkt)
590 self.assert_equal(pkt[IP].dst, self.pg0.remote_ip4)
591 self.assert_equal(pkt[IP].src, self.pg0.local_ip4)
592 self.assertTrue(pkt.haslayer(GRE))
594 self.assertEqual(e[Ether].dst, self.omac)
595 self.assertEqual(e[IP].dst, "1.1.1.2")
596 except (IndexError, AssertionError):
597 self.logger.debug(ppp("Unexpected packet:", rx))
599 self.logger.debug(ppp("Decrypted packet:", pkt))
605 super(TestIpsecGreTebIfEsp, self).setUp()
607 self.tun_if = self.pg0
611 bd1 = VppBridgeDomain(self, 1)
614 p.tun_sa_out = VppIpsecSA(self, p.scapy_tun_sa_id, p.scapy_tun_spi,
615 p.auth_algo_vpp_id, p.auth_key,
616 p.crypt_algo_vpp_id, p.crypt_key,
617 self.vpp_esp_protocol,
620 p.tun_sa_out.add_vpp_config()
622 p.tun_sa_in = VppIpsecSA(self, p.vpp_tun_sa_id, p.vpp_tun_spi,
623 p.auth_algo_vpp_id, p.auth_key,
624 p.crypt_algo_vpp_id, p.crypt_key,
625 self.vpp_esp_protocol,
628 p.tun_sa_in.add_vpp_config()
630 p.tun_if = VppGreInterface(self,
633 type=(VppEnum.vl_api_gre_tunnel_type_t.
634 GRE_API_TUNNEL_TYPE_TEB))
635 p.tun_if.add_vpp_config()
637 p.tun_protect = VppIpsecTunProtect(self,
642 p.tun_protect.add_vpp_config()
645 p.tun_if.config_ip4()
646 config_tun_params(p, self.encryption_type, p.tun_if)
648 VppBridgeDomainPort(self, bd1, p.tun_if).add_vpp_config()
649 VppBridgeDomainPort(self, bd1, self.pg1).add_vpp_config()
651 self.vapi.cli("clear ipsec sa")
655 p.tun_if.unconfig_ip4()
656 super(TestIpsecGreTebIfEsp, self).tearDown()
659 class TestIpsecGreIfEsp(TemplateIpsec,
661 """ Ipsec GRE ESP - TUN tests """
662 tun4_encrypt_node_name = "esp4-encrypt-tun"
663 tun4_decrypt_node_name = "esp4-decrypt-tun"
664 encryption_type = ESP
666 def gen_encrypt_pkts(self, sa, sw_intf, src, dst, count=1,
668 return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
669 sa.encrypt(IP(src=self.pg0.remote_ip4,
670 dst=self.pg0.local_ip4) /
672 IP(src=self.pg1.local_ip4,
673 dst=self.pg1.remote_ip4) /
674 UDP(sport=1144, dport=2233) /
675 Raw(b'X' * payload_size))
676 for i in range(count)]
678 def gen_pkts(self, sw_intf, src, dst, count=1,
680 return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
681 IP(src="1.1.1.1", dst="1.1.1.2") /
682 UDP(sport=1144, dport=2233) /
683 Raw(b'X' * payload_size)
684 for i in range(count)]
686 def verify_decrypted(self, p, rxs):
688 self.assert_equal(rx[Ether].dst, self.pg1.remote_mac)
689 self.assert_equal(rx[IP].dst, self.pg1.remote_ip4)
691 def verify_encrypted(self, p, sa, rxs):
694 pkt = sa.decrypt(rx[IP])
695 if not pkt.haslayer(IP):
696 pkt = IP(pkt[Raw].load)
697 self.assert_packet_checksums_valid(pkt)
698 self.assert_equal(pkt[IP].dst, self.pg0.remote_ip4)
699 self.assert_equal(pkt[IP].src, self.pg0.local_ip4)
700 self.assertTrue(pkt.haslayer(GRE))
702 self.assertEqual(e[IP].dst, "1.1.1.2")
703 except (IndexError, AssertionError):
704 self.logger.debug(ppp("Unexpected packet:", rx))
706 self.logger.debug(ppp("Decrypted packet:", pkt))
712 super(TestIpsecGreIfEsp, self).setUp()
714 self.tun_if = self.pg0
718 bd1 = VppBridgeDomain(self, 1)
721 p.tun_sa_out = VppIpsecSA(self, p.scapy_tun_sa_id, p.scapy_tun_spi,
722 p.auth_algo_vpp_id, p.auth_key,
723 p.crypt_algo_vpp_id, p.crypt_key,
724 self.vpp_esp_protocol,
727 p.tun_sa_out.add_vpp_config()
729 p.tun_sa_in = VppIpsecSA(self, p.vpp_tun_sa_id, p.vpp_tun_spi,
730 p.auth_algo_vpp_id, p.auth_key,
731 p.crypt_algo_vpp_id, p.crypt_key,
732 self.vpp_esp_protocol,
735 p.tun_sa_in.add_vpp_config()
737 p.tun_if = VppGreInterface(self,
740 p.tun_if.add_vpp_config()
742 p.tun_protect = VppIpsecTunProtect(self,
746 p.tun_protect.add_vpp_config()
749 p.tun_if.config_ip4()
750 config_tun_params(p, self.encryption_type, p.tun_if)
752 VppIpRoute(self, "1.1.1.2", 32,
753 [VppRoutePath(p.tun_if.remote_ip4,
754 0xffffffff)]).add_vpp_config()
758 p.tun_if.unconfig_ip4()
759 super(TestIpsecGreIfEsp, self).tearDown()
762 class TestIpsecGreIfEspTra(TemplateIpsec,
764 """ Ipsec GRE ESP - TRA tests """
765 tun4_encrypt_node_name = "esp4-encrypt-tun"
766 tun4_decrypt_node_name = "esp4-decrypt-tun"
767 encryption_type = ESP
769 def gen_encrypt_pkts(self, sa, sw_intf, src, dst, count=1,
771 return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
772 sa.encrypt(IP(src=self.pg0.remote_ip4,
773 dst=self.pg0.local_ip4) /
775 IP(src=self.pg1.local_ip4,
776 dst=self.pg1.remote_ip4) /
777 UDP(sport=1144, dport=2233) /
778 Raw(b'X' * payload_size))
779 for i in range(count)]
781 def gen_pkts(self, sw_intf, src, dst, count=1,
783 return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
784 IP(src="1.1.1.1", dst="1.1.1.2") /
785 UDP(sport=1144, dport=2233) /
786 Raw(b'X' * payload_size)
787 for i in range(count)]
789 def verify_decrypted(self, p, rxs):
791 self.assert_equal(rx[Ether].dst, self.pg1.remote_mac)
792 self.assert_equal(rx[IP].dst, self.pg1.remote_ip4)
794 def verify_encrypted(self, p, sa, rxs):
797 pkt = sa.decrypt(rx[IP])
798 if not pkt.haslayer(IP):
799 pkt = IP(pkt[Raw].load)
800 self.assert_packet_checksums_valid(pkt)
801 self.assertTrue(pkt.haslayer(GRE))
803 self.assertEqual(e[IP].dst, "1.1.1.2")
804 except (IndexError, AssertionError):
805 self.logger.debug(ppp("Unexpected packet:", rx))
807 self.logger.debug(ppp("Decrypted packet:", pkt))
813 super(TestIpsecGreIfEspTra, self).setUp()
815 self.tun_if = self.pg0
819 bd1 = VppBridgeDomain(self, 1)
822 p.tun_sa_out = VppIpsecSA(self, p.scapy_tun_sa_id, p.scapy_tun_spi,
823 p.auth_algo_vpp_id, p.auth_key,
824 p.crypt_algo_vpp_id, p.crypt_key,
825 self.vpp_esp_protocol)
826 p.tun_sa_out.add_vpp_config()
828 p.tun_sa_in = VppIpsecSA(self, p.vpp_tun_sa_id, p.vpp_tun_spi,
829 p.auth_algo_vpp_id, p.auth_key,
830 p.crypt_algo_vpp_id, p.crypt_key,
831 self.vpp_esp_protocol)
832 p.tun_sa_in.add_vpp_config()
834 p.tun_if = VppGreInterface(self,
837 p.tun_if.add_vpp_config()
839 p.tun_protect = VppIpsecTunProtect(self,
843 p.tun_protect.add_vpp_config()
846 p.tun_if.config_ip4()
847 config_tun_params(p, self.encryption_type, p.tun_if)
849 VppIpRoute(self, "1.1.1.2", 32,
850 [VppRoutePath(p.tun_if.remote_ip4,
851 0xffffffff)]).add_vpp_config()
855 p.tun_if.unconfig_ip4()
856 super(TestIpsecGreIfEspTra, self).tearDown()
859 class TemplateIpsec4TunProtect(object):
860 """ IPsec IPv4 Tunnel protect """
862 encryption_type = ESP
863 tun4_encrypt_node_name = "esp4-encrypt-tun"
864 tun4_decrypt_node_name = "esp4-decrypt-tun"
865 tun4_input_node = "ipsec4-tun-input"
867 def config_sa_tra(self, p):
868 config_tun_params(p, self.encryption_type, p.tun_if)
870 p.tun_sa_out = VppIpsecSA(self, p.scapy_tun_sa_id, p.scapy_tun_spi,
871 p.auth_algo_vpp_id, p.auth_key,
872 p.crypt_algo_vpp_id, p.crypt_key,
873 self.vpp_esp_protocol,
875 p.tun_sa_out.add_vpp_config()
877 p.tun_sa_in = VppIpsecSA(self, p.vpp_tun_sa_id, p.vpp_tun_spi,
878 p.auth_algo_vpp_id, p.auth_key,
879 p.crypt_algo_vpp_id, p.crypt_key,
880 self.vpp_esp_protocol,
882 p.tun_sa_in.add_vpp_config()
884 def config_sa_tun(self, p):
885 config_tun_params(p, self.encryption_type, p.tun_if)
887 p.tun_sa_out = VppIpsecSA(self, p.scapy_tun_sa_id, p.scapy_tun_spi,
888 p.auth_algo_vpp_id, p.auth_key,
889 p.crypt_algo_vpp_id, p.crypt_key,
890 self.vpp_esp_protocol,
891 self.tun_if.remote_addr[p.addr_type],
892 self.tun_if.local_addr[p.addr_type],
894 p.tun_sa_out.add_vpp_config()
896 p.tun_sa_in = VppIpsecSA(self, p.vpp_tun_sa_id, p.vpp_tun_spi,
897 p.auth_algo_vpp_id, p.auth_key,
898 p.crypt_algo_vpp_id, p.crypt_key,
899 self.vpp_esp_protocol,
900 self.tun_if.remote_addr[p.addr_type],
901 self.tun_if.local_addr[p.addr_type],
903 p.tun_sa_in.add_vpp_config()
905 def config_protect(self, p):
906 p.tun_protect = VppIpsecTunProtect(self,
910 p.tun_protect.add_vpp_config()
912 def config_network(self, p):
913 p.tun_if = VppIpIpTunInterface(self, self.pg0,
916 p.tun_if.add_vpp_config()
918 p.tun_if.config_ip4()
919 p.tun_if.config_ip6()
921 p.route = VppIpRoute(self, p.remote_tun_if_host, 32,
922 [VppRoutePath(p.tun_if.remote_ip4,
924 p.route.add_vpp_config()
925 r = VppIpRoute(self, p.remote_tun_if_host6, 128,
926 [VppRoutePath(p.tun_if.remote_ip6,
928 proto=DpoProto.DPO_PROTO_IP6)])
931 def unconfig_network(self, p):
932 p.route.remove_vpp_config()
933 p.tun_if.remove_vpp_config()
935 def unconfig_protect(self, p):
936 p.tun_protect.remove_vpp_config()
938 def unconfig_sa(self, p):
939 p.tun_sa_out.remove_vpp_config()
940 p.tun_sa_in.remove_vpp_config()
943 class TestIpsec4TunProtect(TemplateIpsec,
944 TemplateIpsec4TunProtect,
946 """ IPsec IPv4 Tunnel protect - transport mode"""
949 super(TestIpsec4TunProtect, self).setUp()
951 self.tun_if = self.pg0
954 super(TestIpsec4TunProtect, self).tearDown()
956 def test_tun_44(self):
957 """IPSEC tunnel protect"""
961 self.config_network(p)
962 self.config_sa_tra(p)
963 self.config_protect(p)
965 self.verify_tun_44(p, count=127)
966 c = p.tun_if.get_rx_stats()
967 self.assertEqual(c['packets'], 127)
968 c = p.tun_if.get_tx_stats()
969 self.assertEqual(c['packets'], 127)
971 self.vapi.cli("clear ipsec sa")
972 self.verify_tun_64(p, count=127)
973 c = p.tun_if.get_rx_stats()
974 self.assertEqual(c['packets'], 254)
975 c = p.tun_if.get_tx_stats()
976 self.assertEqual(c['packets'], 254)
978 # rekey - create new SAs and update the tunnel protection
980 np.crypt_key = b'X' + p.crypt_key[1:]
981 np.scapy_tun_spi += 100
982 np.scapy_tun_sa_id += 1
983 np.vpp_tun_spi += 100
984 np.vpp_tun_sa_id += 1
985 np.tun_if.local_spi = p.vpp_tun_spi
986 np.tun_if.remote_spi = p.scapy_tun_spi
988 self.config_sa_tra(np)
989 self.config_protect(np)
992 self.verify_tun_44(np, count=127)
993 c = p.tun_if.get_rx_stats()
994 self.assertEqual(c['packets'], 381)
995 c = p.tun_if.get_tx_stats()
996 self.assertEqual(c['packets'], 381)
999 self.unconfig_protect(np)
1000 self.unconfig_sa(np)
1001 self.unconfig_network(p)
1004 class TestIpsec4TunProtectUdp(TemplateIpsec,
1005 TemplateIpsec4TunProtect,
1007 """ IPsec IPv4 Tunnel protect - transport mode"""
1010 super(TestIpsec4TunProtectUdp, self).setUp()
1012 self.tun_if = self.pg0
1014 p = self.ipv4_params
1015 p.flags = (VppEnum.vl_api_ipsec_sad_flags_t.
1016 IPSEC_API_SAD_FLAG_UDP_ENCAP)
1017 p.nat_header = UDP(sport=5454, dport=4500)
1018 self.config_network(p)
1019 self.config_sa_tra(p)
1020 self.config_protect(p)
1023 p = self.ipv4_params
1024 self.unconfig_protect(p)
1026 self.unconfig_network(p)
1027 super(TestIpsec4TunProtectUdp, self).tearDown()
1029 def test_tun_44(self):
1030 """IPSEC UDP tunnel protect"""
1032 p = self.ipv4_params
1034 self.verify_tun_44(p, count=127)
1035 c = p.tun_if.get_rx_stats()
1036 self.assertEqual(c['packets'], 127)
1037 c = p.tun_if.get_tx_stats()
1038 self.assertEqual(c['packets'], 127)
1040 def test_keepalive(self):
1041 """ IPSEC NAT Keepalive """
1042 self.verify_keepalive(self.ipv4_params)
1045 class TestIpsec4TunProtectTun(TemplateIpsec,
1046 TemplateIpsec4TunProtect,
1048 """ IPsec IPv4 Tunnel protect - tunnel mode"""
1050 encryption_type = ESP
1051 tun4_encrypt_node_name = "esp4-encrypt-tun"
1052 tun4_decrypt_node_name = "esp4-decrypt-tun"
1055 super(TestIpsec4TunProtectTun, self).setUp()
1057 self.tun_if = self.pg0
1060 super(TestIpsec4TunProtectTun, self).tearDown()
1062 def gen_encrypt_pkts(self, sa, sw_intf, src, dst, count=1,
1064 return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
1065 sa.encrypt(IP(src=sw_intf.remote_ip4,
1066 dst=sw_intf.local_ip4) /
1067 IP(src=src, dst=dst) /
1068 UDP(sport=1144, dport=2233) /
1069 Raw(b'X' * payload_size))
1070 for i in range(count)]
1072 def gen_pkts(self, sw_intf, src, dst, count=1,
1074 return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
1075 IP(src=src, dst=dst) /
1076 UDP(sport=1144, dport=2233) /
1077 Raw(b'X' * payload_size)
1078 for i in range(count)]
1080 def verify_decrypted(self, p, rxs):
1082 self.assert_equal(rx[IP].dst, self.pg1.remote_ip4)
1083 self.assert_equal(rx[IP].src, p.remote_tun_if_host)
1084 self.assert_packet_checksums_valid(rx)
1086 def verify_encrypted(self, p, sa, rxs):
1089 pkt = sa.decrypt(rx[IP])
1090 if not pkt.haslayer(IP):
1091 pkt = IP(pkt[Raw].load)
1092 self.assert_packet_checksums_valid(pkt)
1093 self.assert_equal(pkt[IP].dst, self.pg0.remote_ip4)
1094 self.assert_equal(pkt[IP].src, self.pg0.local_ip4)
1095 inner = pkt[IP].payload
1096 self.assertEqual(inner[IP][IP].dst, p.remote_tun_if_host)
1098 except (IndexError, AssertionError):
1099 self.logger.debug(ppp("Unexpected packet:", rx))
1101 self.logger.debug(ppp("Decrypted packet:", pkt))
1106 def test_tun_44(self):
1107 """IPSEC tunnel protect """
1109 p = self.ipv4_params
1111 self.config_network(p)
1112 self.config_sa_tun(p)
1113 self.config_protect(p)
1115 self.verify_tun_44(p, count=127)
1117 c = p.tun_if.get_rx_stats()
1118 self.assertEqual(c['packets'], 127)
1119 c = p.tun_if.get_tx_stats()
1120 self.assertEqual(c['packets'], 127)
1122 # rekey - create new SAs and update the tunnel protection
1124 np.crypt_key = b'X' + p.crypt_key[1:]
1125 np.scapy_tun_spi += 100
1126 np.scapy_tun_sa_id += 1
1127 np.vpp_tun_spi += 100
1128 np.vpp_tun_sa_id += 1
1129 np.tun_if.local_spi = p.vpp_tun_spi
1130 np.tun_if.remote_spi = p.scapy_tun_spi
1132 self.config_sa_tun(np)
1133 self.config_protect(np)
1136 self.verify_tun_44(np, count=127)
1137 c = p.tun_if.get_rx_stats()
1138 self.assertEqual(c['packets'], 254)
1139 c = p.tun_if.get_tx_stats()
1140 self.assertEqual(c['packets'], 254)
1143 self.unconfig_protect(np)
1144 self.unconfig_sa(np)
1145 self.unconfig_network(p)
1148 class TemplateIpsec6TunProtect(object):
1149 """ IPsec IPv6 Tunnel protect """
1151 def config_sa_tra(self, p):
1152 config_tun_params(p, self.encryption_type, p.tun_if)
1154 p.tun_sa_out = VppIpsecSA(self, p.scapy_tun_sa_id, p.scapy_tun_spi,
1155 p.auth_algo_vpp_id, p.auth_key,
1156 p.crypt_algo_vpp_id, p.crypt_key,
1157 self.vpp_esp_protocol)
1158 p.tun_sa_out.add_vpp_config()
1160 p.tun_sa_in = VppIpsecSA(self, p.vpp_tun_sa_id, p.vpp_tun_spi,
1161 p.auth_algo_vpp_id, p.auth_key,
1162 p.crypt_algo_vpp_id, p.crypt_key,
1163 self.vpp_esp_protocol)
1164 p.tun_sa_in.add_vpp_config()
1166 def config_sa_tun(self, p):
1167 config_tun_params(p, self.encryption_type, p.tun_if)
1169 p.tun_sa_out = VppIpsecSA(self, p.scapy_tun_sa_id, p.scapy_tun_spi,
1170 p.auth_algo_vpp_id, p.auth_key,
1171 p.crypt_algo_vpp_id, p.crypt_key,
1172 self.vpp_esp_protocol,
1173 self.tun_if.remote_addr[p.addr_type],
1174 self.tun_if.local_addr[p.addr_type])
1175 p.tun_sa_out.add_vpp_config()
1177 p.tun_sa_in = VppIpsecSA(self, p.vpp_tun_sa_id, p.vpp_tun_spi,
1178 p.auth_algo_vpp_id, p.auth_key,
1179 p.crypt_algo_vpp_id, p.crypt_key,
1180 self.vpp_esp_protocol,
1181 self.tun_if.remote_addr[p.addr_type],
1182 self.tun_if.local_addr[p.addr_type])
1183 p.tun_sa_in.add_vpp_config()
1185 def config_protect(self, p):
1186 p.tun_protect = VppIpsecTunProtect(self,
1190 p.tun_protect.add_vpp_config()
1192 def config_network(self, p):
1193 p.tun_if = VppIpIpTunInterface(self, self.pg0,
1195 self.pg0.remote_ip6)
1196 p.tun_if.add_vpp_config()
1198 p.tun_if.config_ip6()
1199 p.tun_if.config_ip4()
1201 p.route = VppIpRoute(self, p.remote_tun_if_host, 128,
1202 [VppRoutePath(p.tun_if.remote_ip6,
1204 proto=DpoProto.DPO_PROTO_IP6)])
1205 p.route.add_vpp_config()
1206 r = VppIpRoute(self, p.remote_tun_if_host4, 32,
1207 [VppRoutePath(p.tun_if.remote_ip4,
1211 def unconfig_network(self, p):
1212 p.route.remove_vpp_config()
1213 p.tun_if.remove_vpp_config()
1215 def unconfig_protect(self, p):
1216 p.tun_protect.remove_vpp_config()
1218 def unconfig_sa(self, p):
1219 p.tun_sa_out.remove_vpp_config()
1220 p.tun_sa_in.remove_vpp_config()
1223 class TestIpsec6TunProtect(TemplateIpsec,
1224 TemplateIpsec6TunProtect,
1226 """ IPsec IPv6 Tunnel protect - transport mode"""
1228 encryption_type = ESP
1229 tun6_encrypt_node_name = "esp6-encrypt-tun"
1230 tun6_decrypt_node_name = "esp6-decrypt-tun"
1233 super(TestIpsec6TunProtect, self).setUp()
1235 self.tun_if = self.pg0
1238 super(TestIpsec6TunProtect, self).tearDown()
1240 def test_tun_66(self):
1241 """IPSEC tunnel protect"""
1243 p = self.ipv6_params
1245 self.config_network(p)
1246 self.config_sa_tra(p)
1247 self.config_protect(p)
1249 self.verify_tun_66(p, count=127)
1250 c = p.tun_if.get_rx_stats()
1251 self.assertEqual(c['packets'], 127)
1252 c = p.tun_if.get_tx_stats()
1253 self.assertEqual(c['packets'], 127)
1255 # rekey - create new SAs and update the tunnel protection
1257 np.crypt_key = b'X' + p.crypt_key[1:]
1258 np.scapy_tun_spi += 100
1259 np.scapy_tun_sa_id += 1
1260 np.vpp_tun_spi += 100
1261 np.vpp_tun_sa_id += 1
1262 np.tun_if.local_spi = p.vpp_tun_spi
1263 np.tun_if.remote_spi = p.scapy_tun_spi
1265 self.config_sa_tra(np)
1266 self.config_protect(np)
1269 self.verify_tun_66(np, count=127)
1270 c = p.tun_if.get_rx_stats()
1271 self.assertEqual(c['packets'], 254)
1272 c = p.tun_if.get_tx_stats()
1273 self.assertEqual(c['packets'], 254)
1276 # 1) add two input SAs [old, new]
1277 # 2) swap output SA to [new]
1278 # 3) use only [new] input SA
1280 np3.crypt_key = b'Z' + p.crypt_key[1:]
1281 np3.scapy_tun_spi += 100
1282 np3.scapy_tun_sa_id += 1
1283 np3.vpp_tun_spi += 100
1284 np3.vpp_tun_sa_id += 1
1285 np3.tun_if.local_spi = p.vpp_tun_spi
1286 np3.tun_if.remote_spi = p.scapy_tun_spi
1288 self.config_sa_tra(np3)
1291 p.tun_protect.update_vpp_config(np.tun_sa_out,
1292 [np.tun_sa_in, np3.tun_sa_in])
1293 self.verify_tun_66(np, np, count=127)
1294 self.verify_tun_66(np3, np, count=127)
1297 p.tun_protect.update_vpp_config(np3.tun_sa_out,
1298 [np.tun_sa_in, np3.tun_sa_in])
1299 self.verify_tun_66(np, np3, count=127)
1300 self.verify_tun_66(np3, np3, count=127)
1303 p.tun_protect.update_vpp_config(np3.tun_sa_out,
1305 self.verify_tun_66(np3, np3, count=127)
1306 self.verify_drop_tun_66(np, count=127)
1308 c = p.tun_if.get_rx_stats()
1309 self.assertEqual(c['packets'], 127*7)
1310 c = p.tun_if.get_tx_stats()
1311 self.assertEqual(c['packets'], 127*7)
1312 self.unconfig_sa(np)
1315 self.unconfig_protect(np3)
1316 self.unconfig_sa(np3)
1317 self.unconfig_network(p)
1319 def test_tun_46(self):
1320 """IPSEC tunnel protect"""
1322 p = self.ipv6_params
1324 self.config_network(p)
1325 self.config_sa_tra(p)
1326 self.config_protect(p)
1328 self.verify_tun_46(p, count=127)
1329 c = p.tun_if.get_rx_stats()
1330 self.assertEqual(c['packets'], 127)
1331 c = p.tun_if.get_tx_stats()
1332 self.assertEqual(c['packets'], 127)
1335 self.unconfig_protect(p)
1337 self.unconfig_network(p)
1340 class TestIpsec6TunProtectTun(TemplateIpsec,
1341 TemplateIpsec6TunProtect,
1343 """ IPsec IPv6 Tunnel protect - tunnel mode"""
1345 encryption_type = ESP
1346 tun6_encrypt_node_name = "esp6-encrypt-tun"
1347 tun6_decrypt_node_name = "esp6-decrypt-tun"
1350 super(TestIpsec6TunProtectTun, self).setUp()
1352 self.tun_if = self.pg0
1355 super(TestIpsec6TunProtectTun, self).tearDown()
1357 def gen_encrypt_pkts6(self, sa, sw_intf, src, dst, count=1,
1359 return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
1360 sa.encrypt(IPv6(src=sw_intf.remote_ip6,
1361 dst=sw_intf.local_ip6) /
1362 IPv6(src=src, dst=dst) /
1363 UDP(sport=1166, dport=2233) /
1364 Raw(b'X' * payload_size))
1365 for i in range(count)]
1367 def gen_pkts6(self, sw_intf, src, dst, count=1,
1369 return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
1370 IPv6(src=src, dst=dst) /
1371 UDP(sport=1166, dport=2233) /
1372 Raw(b'X' * payload_size)
1373 for i in range(count)]
1375 def verify_decrypted6(self, p, rxs):
1377 self.assert_equal(rx[IPv6].dst, self.pg1.remote_ip6)
1378 self.assert_equal(rx[IPv6].src, p.remote_tun_if_host)
1379 self.assert_packet_checksums_valid(rx)
1381 def verify_encrypted6(self, p, sa, rxs):
1384 pkt = sa.decrypt(rx[IPv6])
1385 if not pkt.haslayer(IPv6):
1386 pkt = IPv6(pkt[Raw].load)
1387 self.assert_packet_checksums_valid(pkt)
1388 self.assert_equal(pkt[IPv6].dst, self.pg0.remote_ip6)
1389 self.assert_equal(pkt[IPv6].src, self.pg0.local_ip6)
1390 inner = pkt[IPv6].payload
1391 self.assertEqual(inner[IPv6][IPv6].dst, p.remote_tun_if_host)
1393 except (IndexError, AssertionError):
1394 self.logger.debug(ppp("Unexpected packet:", rx))
1396 self.logger.debug(ppp("Decrypted packet:", pkt))
1401 def test_tun_66(self):
1402 """IPSEC tunnel protect """
1404 p = self.ipv6_params
1406 self.config_network(p)
1407 self.config_sa_tun(p)
1408 self.config_protect(p)
1410 self.verify_tun_66(p, count=127)
1412 c = p.tun_if.get_rx_stats()
1413 self.assertEqual(c['packets'], 127)
1414 c = p.tun_if.get_tx_stats()
1415 self.assertEqual(c['packets'], 127)
1417 # rekey - create new SAs and update the tunnel protection
1419 np.crypt_key = b'X' + p.crypt_key[1:]
1420 np.scapy_tun_spi += 100
1421 np.scapy_tun_sa_id += 1
1422 np.vpp_tun_spi += 100
1423 np.vpp_tun_sa_id += 1
1424 np.tun_if.local_spi = p.vpp_tun_spi
1425 np.tun_if.remote_spi = p.scapy_tun_spi
1427 self.config_sa_tun(np)
1428 self.config_protect(np)
1431 self.verify_tun_66(np, count=127)
1432 c = p.tun_if.get_rx_stats()
1433 self.assertEqual(c['packets'], 254)
1434 c = p.tun_if.get_tx_stats()
1435 self.assertEqual(c['packets'], 254)
1438 self.unconfig_protect(np)
1439 self.unconfig_sa(np)
1440 self.unconfig_network(p)
1443 if __name__ == '__main__':
1444 unittest.main(testRunner=VppTestRunner)