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 TemplateIpsec4TunProtect(object):
763 """ IPsec IPv4 Tunnel protect """
765 encryption_type = ESP
766 tun4_encrypt_node_name = "esp4-encrypt-tun"
767 tun4_decrypt_node_name = "esp4-decrypt-tun"
768 tun4_input_node = "ipsec4-tun-input"
770 def config_sa_tra(self, p):
771 config_tun_params(p, self.encryption_type, p.tun_if)
773 p.tun_sa_out = VppIpsecSA(self, p.scapy_tun_sa_id, p.scapy_tun_spi,
774 p.auth_algo_vpp_id, p.auth_key,
775 p.crypt_algo_vpp_id, p.crypt_key,
776 self.vpp_esp_protocol,
778 p.tun_sa_out.add_vpp_config()
780 p.tun_sa_in = VppIpsecSA(self, p.vpp_tun_sa_id, p.vpp_tun_spi,
781 p.auth_algo_vpp_id, p.auth_key,
782 p.crypt_algo_vpp_id, p.crypt_key,
783 self.vpp_esp_protocol,
785 p.tun_sa_in.add_vpp_config()
787 def config_sa_tun(self, p):
788 config_tun_params(p, self.encryption_type, p.tun_if)
790 p.tun_sa_out = VppIpsecSA(self, p.scapy_tun_sa_id, p.scapy_tun_spi,
791 p.auth_algo_vpp_id, p.auth_key,
792 p.crypt_algo_vpp_id, p.crypt_key,
793 self.vpp_esp_protocol,
794 self.tun_if.remote_addr[p.addr_type],
795 self.tun_if.local_addr[p.addr_type],
797 p.tun_sa_out.add_vpp_config()
799 p.tun_sa_in = VppIpsecSA(self, p.vpp_tun_sa_id, p.vpp_tun_spi,
800 p.auth_algo_vpp_id, p.auth_key,
801 p.crypt_algo_vpp_id, p.crypt_key,
802 self.vpp_esp_protocol,
803 self.tun_if.remote_addr[p.addr_type],
804 self.tun_if.local_addr[p.addr_type],
806 p.tun_sa_in.add_vpp_config()
808 def config_protect(self, p):
809 p.tun_protect = VppIpsecTunProtect(self,
813 p.tun_protect.add_vpp_config()
815 def config_network(self, p):
816 p.tun_if = VppIpIpTunInterface(self, self.pg0,
819 p.tun_if.add_vpp_config()
821 p.tun_if.config_ip4()
822 p.tun_if.config_ip6()
824 p.route = VppIpRoute(self, p.remote_tun_if_host, 32,
825 [VppRoutePath(p.tun_if.remote_ip4,
827 p.route.add_vpp_config()
828 r = VppIpRoute(self, p.remote_tun_if_host6, 128,
829 [VppRoutePath(p.tun_if.remote_ip6,
831 proto=DpoProto.DPO_PROTO_IP6)])
834 def unconfig_network(self, p):
835 p.route.remove_vpp_config()
836 p.tun_if.remove_vpp_config()
838 def unconfig_protect(self, p):
839 p.tun_protect.remove_vpp_config()
841 def unconfig_sa(self, p):
842 p.tun_sa_out.remove_vpp_config()
843 p.tun_sa_in.remove_vpp_config()
846 class TestIpsec4TunProtect(TemplateIpsec,
847 TemplateIpsec4TunProtect,
849 """ IPsec IPv4 Tunnel protect - transport mode"""
852 super(TestIpsec4TunProtect, self).setUp()
854 self.tun_if = self.pg0
857 super(TestIpsec4TunProtect, self).tearDown()
859 def test_tun_44(self):
860 """IPSEC tunnel protect"""
864 self.config_network(p)
865 self.config_sa_tra(p)
866 self.config_protect(p)
868 self.verify_tun_44(p, count=127)
869 c = p.tun_if.get_rx_stats()
870 self.assertEqual(c['packets'], 127)
871 c = p.tun_if.get_tx_stats()
872 self.assertEqual(c['packets'], 127)
874 self.vapi.cli("clear ipsec sa")
875 self.verify_tun_64(p, count=127)
876 c = p.tun_if.get_rx_stats()
877 self.assertEqual(c['packets'], 254)
878 c = p.tun_if.get_tx_stats()
879 self.assertEqual(c['packets'], 254)
881 # rekey - create new SAs and update the tunnel protection
883 np.crypt_key = b'X' + p.crypt_key[1:]
884 np.scapy_tun_spi += 100
885 np.scapy_tun_sa_id += 1
886 np.vpp_tun_spi += 100
887 np.vpp_tun_sa_id += 1
888 np.tun_if.local_spi = p.vpp_tun_spi
889 np.tun_if.remote_spi = p.scapy_tun_spi
891 self.config_sa_tra(np)
892 self.config_protect(np)
895 self.verify_tun_44(np, count=127)
896 c = p.tun_if.get_rx_stats()
897 self.assertEqual(c['packets'], 381)
898 c = p.tun_if.get_tx_stats()
899 self.assertEqual(c['packets'], 381)
902 self.unconfig_protect(np)
904 self.unconfig_network(p)
907 class TestIpsec4TunProtectUdp(TemplateIpsec,
908 TemplateIpsec4TunProtect,
910 """ IPsec IPv4 Tunnel protect - transport mode"""
913 super(TestIpsec4TunProtectUdp, self).setUp()
915 self.tun_if = self.pg0
918 p.flags = (VppEnum.vl_api_ipsec_sad_flags_t.
919 IPSEC_API_SAD_FLAG_UDP_ENCAP)
920 p.nat_header = UDP(sport=5454, dport=4500)
921 self.config_network(p)
922 self.config_sa_tra(p)
923 self.config_protect(p)
927 self.unconfig_protect(p)
929 self.unconfig_network(p)
930 super(TestIpsec4TunProtectUdp, self).tearDown()
932 def test_tun_44(self):
933 """IPSEC UDP tunnel protect"""
937 self.verify_tun_44(p, count=127)
938 c = p.tun_if.get_rx_stats()
939 self.assertEqual(c['packets'], 127)
940 c = p.tun_if.get_tx_stats()
941 self.assertEqual(c['packets'], 127)
943 def test_keepalive(self):
944 """ IPSEC NAT Keepalive """
945 self.verify_keepalive(self.ipv4_params)
948 class TestIpsec4TunProtectTun(TemplateIpsec,
949 TemplateIpsec4TunProtect,
951 """ IPsec IPv4 Tunnel protect - tunnel mode"""
953 encryption_type = ESP
954 tun4_encrypt_node_name = "esp4-encrypt-tun"
955 tun4_decrypt_node_name = "esp4-decrypt-tun"
958 super(TestIpsec4TunProtectTun, self).setUp()
960 self.tun_if = self.pg0
963 super(TestIpsec4TunProtectTun, self).tearDown()
965 def gen_encrypt_pkts(self, sa, sw_intf, src, dst, count=1,
967 return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
968 sa.encrypt(IP(src=sw_intf.remote_ip4,
969 dst=sw_intf.local_ip4) /
970 IP(src=src, dst=dst) /
971 UDP(sport=1144, dport=2233) /
972 Raw(b'X' * payload_size))
973 for i in range(count)]
975 def gen_pkts(self, sw_intf, src, dst, count=1,
977 return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
978 IP(src=src, dst=dst) /
979 UDP(sport=1144, dport=2233) /
980 Raw(b'X' * payload_size)
981 for i in range(count)]
983 def verify_decrypted(self, p, rxs):
985 self.assert_equal(rx[IP].dst, self.pg1.remote_ip4)
986 self.assert_equal(rx[IP].src, p.remote_tun_if_host)
987 self.assert_packet_checksums_valid(rx)
989 def verify_encrypted(self, p, sa, rxs):
992 pkt = sa.decrypt(rx[IP])
993 if not pkt.haslayer(IP):
994 pkt = IP(pkt[Raw].load)
995 self.assert_packet_checksums_valid(pkt)
996 self.assert_equal(pkt[IP].dst, self.pg0.remote_ip4)
997 self.assert_equal(pkt[IP].src, self.pg0.local_ip4)
998 inner = pkt[IP].payload
999 self.assertEqual(inner[IP][IP].dst, p.remote_tun_if_host)
1001 except (IndexError, AssertionError):
1002 self.logger.debug(ppp("Unexpected packet:", rx))
1004 self.logger.debug(ppp("Decrypted packet:", pkt))
1009 def test_tun_44(self):
1010 """IPSEC tunnel protect """
1012 p = self.ipv4_params
1014 self.config_network(p)
1015 self.config_sa_tun(p)
1016 self.config_protect(p)
1018 self.verify_tun_44(p, count=127)
1020 c = p.tun_if.get_rx_stats()
1021 self.assertEqual(c['packets'], 127)
1022 c = p.tun_if.get_tx_stats()
1023 self.assertEqual(c['packets'], 127)
1025 # rekey - create new SAs and update the tunnel protection
1027 np.crypt_key = b'X' + p.crypt_key[1:]
1028 np.scapy_tun_spi += 100
1029 np.scapy_tun_sa_id += 1
1030 np.vpp_tun_spi += 100
1031 np.vpp_tun_sa_id += 1
1032 np.tun_if.local_spi = p.vpp_tun_spi
1033 np.tun_if.remote_spi = p.scapy_tun_spi
1035 self.config_sa_tun(np)
1036 self.config_protect(np)
1039 self.verify_tun_44(np, count=127)
1040 c = p.tun_if.get_rx_stats()
1041 self.assertEqual(c['packets'], 254)
1042 c = p.tun_if.get_tx_stats()
1043 self.assertEqual(c['packets'], 254)
1046 self.unconfig_protect(np)
1047 self.unconfig_sa(np)
1048 self.unconfig_network(p)
1051 class TemplateIpsec6TunProtect(object):
1052 """ IPsec IPv6 Tunnel protect """
1054 def config_sa_tra(self, p):
1055 config_tun_params(p, self.encryption_type, p.tun_if)
1057 p.tun_sa_out = VppIpsecSA(self, p.scapy_tun_sa_id, p.scapy_tun_spi,
1058 p.auth_algo_vpp_id, p.auth_key,
1059 p.crypt_algo_vpp_id, p.crypt_key,
1060 self.vpp_esp_protocol)
1061 p.tun_sa_out.add_vpp_config()
1063 p.tun_sa_in = VppIpsecSA(self, p.vpp_tun_sa_id, p.vpp_tun_spi,
1064 p.auth_algo_vpp_id, p.auth_key,
1065 p.crypt_algo_vpp_id, p.crypt_key,
1066 self.vpp_esp_protocol)
1067 p.tun_sa_in.add_vpp_config()
1069 def config_sa_tun(self, p):
1070 config_tun_params(p, self.encryption_type, p.tun_if)
1072 p.tun_sa_out = VppIpsecSA(self, p.scapy_tun_sa_id, p.scapy_tun_spi,
1073 p.auth_algo_vpp_id, p.auth_key,
1074 p.crypt_algo_vpp_id, p.crypt_key,
1075 self.vpp_esp_protocol,
1076 self.tun_if.remote_addr[p.addr_type],
1077 self.tun_if.local_addr[p.addr_type])
1078 p.tun_sa_out.add_vpp_config()
1080 p.tun_sa_in = VppIpsecSA(self, p.vpp_tun_sa_id, p.vpp_tun_spi,
1081 p.auth_algo_vpp_id, p.auth_key,
1082 p.crypt_algo_vpp_id, p.crypt_key,
1083 self.vpp_esp_protocol,
1084 self.tun_if.remote_addr[p.addr_type],
1085 self.tun_if.local_addr[p.addr_type])
1086 p.tun_sa_in.add_vpp_config()
1088 def config_protect(self, p):
1089 p.tun_protect = VppIpsecTunProtect(self,
1093 p.tun_protect.add_vpp_config()
1095 def config_network(self, p):
1096 p.tun_if = VppIpIpTunInterface(self, self.pg0,
1098 self.pg0.remote_ip6)
1099 p.tun_if.add_vpp_config()
1101 p.tun_if.config_ip6()
1102 p.tun_if.config_ip4()
1104 p.route = VppIpRoute(self, p.remote_tun_if_host, 128,
1105 [VppRoutePath(p.tun_if.remote_ip6,
1107 proto=DpoProto.DPO_PROTO_IP6)])
1108 p.route.add_vpp_config()
1109 r = VppIpRoute(self, p.remote_tun_if_host4, 32,
1110 [VppRoutePath(p.tun_if.remote_ip4,
1114 def unconfig_network(self, p):
1115 p.route.remove_vpp_config()
1116 p.tun_if.remove_vpp_config()
1118 def unconfig_protect(self, p):
1119 p.tun_protect.remove_vpp_config()
1121 def unconfig_sa(self, p):
1122 p.tun_sa_out.remove_vpp_config()
1123 p.tun_sa_in.remove_vpp_config()
1126 class TestIpsec6TunProtect(TemplateIpsec,
1127 TemplateIpsec6TunProtect,
1129 """ IPsec IPv6 Tunnel protect - transport mode"""
1131 encryption_type = ESP
1132 tun6_encrypt_node_name = "esp6-encrypt-tun"
1133 tun6_decrypt_node_name = "esp6-decrypt-tun"
1136 super(TestIpsec6TunProtect, self).setUp()
1138 self.tun_if = self.pg0
1141 super(TestIpsec6TunProtect, self).tearDown()
1143 def test_tun_66(self):
1144 """IPSEC tunnel protect"""
1146 p = self.ipv6_params
1148 self.config_network(p)
1149 self.config_sa_tra(p)
1150 self.config_protect(p)
1152 self.verify_tun_66(p, count=127)
1153 c = p.tun_if.get_rx_stats()
1154 self.assertEqual(c['packets'], 127)
1155 c = p.tun_if.get_tx_stats()
1156 self.assertEqual(c['packets'], 127)
1158 # rekey - create new SAs and update the tunnel protection
1160 np.crypt_key = b'X' + p.crypt_key[1:]
1161 np.scapy_tun_spi += 100
1162 np.scapy_tun_sa_id += 1
1163 np.vpp_tun_spi += 100
1164 np.vpp_tun_sa_id += 1
1165 np.tun_if.local_spi = p.vpp_tun_spi
1166 np.tun_if.remote_spi = p.scapy_tun_spi
1168 self.config_sa_tra(np)
1169 self.config_protect(np)
1172 self.verify_tun_66(np, count=127)
1173 c = p.tun_if.get_rx_stats()
1174 self.assertEqual(c['packets'], 254)
1175 c = p.tun_if.get_tx_stats()
1176 self.assertEqual(c['packets'], 254)
1179 # 1) add two input SAs [old, new]
1180 # 2) swap output SA to [new]
1181 # 3) use only [new] input SA
1183 np3.crypt_key = b'Z' + p.crypt_key[1:]
1184 np3.scapy_tun_spi += 100
1185 np3.scapy_tun_sa_id += 1
1186 np3.vpp_tun_spi += 100
1187 np3.vpp_tun_sa_id += 1
1188 np3.tun_if.local_spi = p.vpp_tun_spi
1189 np3.tun_if.remote_spi = p.scapy_tun_spi
1191 self.config_sa_tra(np3)
1194 p.tun_protect.update_vpp_config(np.tun_sa_out,
1195 [np.tun_sa_in, np3.tun_sa_in])
1196 self.verify_tun_66(np, np, count=127)
1197 self.verify_tun_66(np3, np, count=127)
1200 p.tun_protect.update_vpp_config(np3.tun_sa_out,
1201 [np.tun_sa_in, np3.tun_sa_in])
1202 self.verify_tun_66(np, np3, count=127)
1203 self.verify_tun_66(np3, np3, count=127)
1206 p.tun_protect.update_vpp_config(np3.tun_sa_out,
1208 self.verify_tun_66(np3, np3, count=127)
1209 self.verify_drop_tun_66(np, count=127)
1211 c = p.tun_if.get_rx_stats()
1212 self.assertEqual(c['packets'], 127*7)
1213 c = p.tun_if.get_tx_stats()
1214 self.assertEqual(c['packets'], 127*7)
1215 self.unconfig_sa(np)
1218 self.unconfig_protect(np3)
1219 self.unconfig_sa(np3)
1220 self.unconfig_network(p)
1222 def test_tun_46(self):
1223 """IPSEC tunnel protect"""
1225 p = self.ipv6_params
1227 self.config_network(p)
1228 self.config_sa_tra(p)
1229 self.config_protect(p)
1231 self.verify_tun_46(p, count=127)
1232 c = p.tun_if.get_rx_stats()
1233 self.assertEqual(c['packets'], 127)
1234 c = p.tun_if.get_tx_stats()
1235 self.assertEqual(c['packets'], 127)
1238 self.unconfig_protect(p)
1240 self.unconfig_network(p)
1243 class TestIpsec6TunProtectTun(TemplateIpsec,
1244 TemplateIpsec6TunProtect,
1246 """ IPsec IPv6 Tunnel protect - tunnel mode"""
1248 encryption_type = ESP
1249 tun6_encrypt_node_name = "esp6-encrypt-tun"
1250 tun6_decrypt_node_name = "esp6-decrypt-tun"
1253 super(TestIpsec6TunProtectTun, self).setUp()
1255 self.tun_if = self.pg0
1258 super(TestIpsec6TunProtectTun, self).tearDown()
1260 def gen_encrypt_pkts6(self, sa, sw_intf, src, dst, count=1,
1262 return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
1263 sa.encrypt(IPv6(src=sw_intf.remote_ip6,
1264 dst=sw_intf.local_ip6) /
1265 IPv6(src=src, dst=dst) /
1266 UDP(sport=1166, dport=2233) /
1267 Raw(b'X' * payload_size))
1268 for i in range(count)]
1270 def gen_pkts6(self, sw_intf, src, dst, count=1,
1272 return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
1273 IPv6(src=src, dst=dst) /
1274 UDP(sport=1166, dport=2233) /
1275 Raw(b'X' * payload_size)
1276 for i in range(count)]
1278 def verify_decrypted6(self, p, rxs):
1280 self.assert_equal(rx[IPv6].dst, self.pg1.remote_ip6)
1281 self.assert_equal(rx[IPv6].src, p.remote_tun_if_host)
1282 self.assert_packet_checksums_valid(rx)
1284 def verify_encrypted6(self, p, sa, rxs):
1287 pkt = sa.decrypt(rx[IPv6])
1288 if not pkt.haslayer(IPv6):
1289 pkt = IPv6(pkt[Raw].load)
1290 self.assert_packet_checksums_valid(pkt)
1291 self.assert_equal(pkt[IPv6].dst, self.pg0.remote_ip6)
1292 self.assert_equal(pkt[IPv6].src, self.pg0.local_ip6)
1293 inner = pkt[IPv6].payload
1294 self.assertEqual(inner[IPv6][IPv6].dst, p.remote_tun_if_host)
1296 except (IndexError, AssertionError):
1297 self.logger.debug(ppp("Unexpected packet:", rx))
1299 self.logger.debug(ppp("Decrypted packet:", pkt))
1304 def test_tun_66(self):
1305 """IPSEC tunnel protect """
1307 p = self.ipv6_params
1309 self.config_network(p)
1310 self.config_sa_tun(p)
1311 self.config_protect(p)
1313 self.verify_tun_66(p, count=127)
1315 c = p.tun_if.get_rx_stats()
1316 self.assertEqual(c['packets'], 127)
1317 c = p.tun_if.get_tx_stats()
1318 self.assertEqual(c['packets'], 127)
1320 # rekey - create new SAs and update the tunnel protection
1322 np.crypt_key = b'X' + p.crypt_key[1:]
1323 np.scapy_tun_spi += 100
1324 np.scapy_tun_sa_id += 1
1325 np.vpp_tun_spi += 100
1326 np.vpp_tun_sa_id += 1
1327 np.tun_if.local_spi = p.vpp_tun_spi
1328 np.tun_if.remote_spi = p.scapy_tun_spi
1330 self.config_sa_tun(np)
1331 self.config_protect(np)
1334 self.verify_tun_66(np, count=127)
1335 c = p.tun_if.get_rx_stats()
1336 self.assertEqual(c['packets'], 254)
1337 c = p.tun_if.get_tx_stats()
1338 self.assertEqual(c['packets'], 254)
1341 self.unconfig_protect(np)
1342 self.unconfig_sa(np)
1343 self.unconfig_network(p)
1346 if __name__ == '__main__':
1347 unittest.main(testRunner=VppTestRunner)