5 from scapy.layers.ipsec import ESP
6 from scapy.layers.l2 import Ether, Raw, GRE
7 from scapy.layers.inet import IP, UDP
8 from framework import VppTestRunner, is_skip_aarch64_set, is_platform_aarch64
9 from template_ipsec import TemplateIpsec, IpsecTun4Tests, IpsecTun6Tests, \
10 IpsecTun4, IpsecTun6, IpsecTcpTests, config_tun_params
11 from vpp_ipsec_tun_interface import VppIpsecTunInterface, \
12 VppIpsecGRETunInterface
13 from vpp_ip_route import VppIpRoute, VppRoutePath, DpoProto
14 from vpp_ipsec import VppIpsecSA
15 from vpp_l2 import VppBridgeDomain, VppBridgeDomainPort
17 from vpp_papi import VppEnum
20 class TemplateIpsec4TunIfEsp(TemplateIpsec):
21 """ IPsec tunnel interface tests """
27 super(TemplateIpsec4TunIfEsp, cls).setUpClass()
30 def tearDownClass(cls):
31 super(TemplateIpsec4TunIfEsp, cls).tearDownClass()
34 super(TemplateIpsec4TunIfEsp, self).setUp()
36 self.tun_if = self.pg0
40 p.tun_if = VppIpsecTunInterface(self, self.pg0, p.vpp_tun_spi,
41 p.scapy_tun_spi, p.crypt_algo_vpp_id,
42 p.crypt_key, p.crypt_key,
43 p.auth_algo_vpp_id, p.auth_key,
45 p.tun_if.add_vpp_config()
50 VppIpRoute(self, p.remote_tun_if_host, 32,
51 [VppRoutePath(p.tun_if.remote_ip4,
52 0xffffffff)]).add_vpp_config()
53 VppIpRoute(self, p.remote_tun_if_host6, 128,
54 [VppRoutePath(p.tun_if.remote_ip6,
56 proto=DpoProto.DPO_PROTO_IP6)],
57 is_ip6=1).add_vpp_config()
61 self.vapi.cli("show hardware")
62 super(TemplateIpsec4TunIfEsp, self).tearDown()
65 class TestIpsec4TunIfEsp1(TemplateIpsec4TunIfEsp, IpsecTun4Tests):
66 """ Ipsec ESP - TUN tests """
67 tun4_encrypt_node_name = "esp4-encrypt-tun"
68 tun4_decrypt_node_name = "esp4-decrypt"
70 def test_tun_basic64(self):
71 """ ipsec 6o4 tunnel basic test """
72 self.tun4_encrypt_node_name = "esp6-encrypt-tun"
74 self.verify_tun_64(self.params[socket.AF_INET], count=1)
76 def test_tun_burst64(self):
77 """ ipsec 6o4 tunnel basic test """
78 self.tun4_encrypt_node_name = "esp6-encrypt-tun"
80 self.verify_tun_64(self.params[socket.AF_INET], count=257)
82 def test_tun_basic_frag44(self):
83 """ ipsec 4o4 tunnel frag basic test """
84 self.tun4_encrypt_node_name = "esp4-encrypt-tun"
88 self.vapi.sw_interface_set_mtu(p.tun_if.sw_if_index,
90 self.verify_tun_44(self.params[socket.AF_INET],
91 count=1, payload_size=1800, n_rx=2)
92 self.vapi.sw_interface_set_mtu(p.tun_if.sw_if_index,
96 class TestIpsec4TunIfEsp2(TemplateIpsec4TunIfEsp, IpsecTcpTests):
97 """ Ipsec ESP - TCP tests """
101 class TemplateIpsec6TunIfEsp(TemplateIpsec):
102 """ IPsec tunnel interface tests """
104 encryption_type = ESP
107 super(TemplateIpsec6TunIfEsp, self).setUp()
109 self.tun_if = self.pg0
112 tun_if = VppIpsecTunInterface(self, self.pg0, p.vpp_tun_spi,
113 p.scapy_tun_spi, p.crypt_algo_vpp_id,
114 p.crypt_key, p.crypt_key,
115 p.auth_algo_vpp_id, p.auth_key,
116 p.auth_key, is_ip6=True)
117 tun_if.add_vpp_config()
122 VppIpRoute(self, p.remote_tun_if_host, 128,
123 [VppRoutePath(tun_if.remote_ip6,
125 proto=DpoProto.DPO_PROTO_IP6)],
126 is_ip6=1).add_vpp_config()
127 VppIpRoute(self, p.remote_tun_if_host4, 32,
128 [VppRoutePath(tun_if.remote_ip4,
129 0xffffffff)]).add_vpp_config()
132 if not self.vpp_dead:
133 self.vapi.cli("show hardware")
134 super(TemplateIpsec6TunIfEsp, self).tearDown()
137 class TestIpsec6TunIfEsp1(TemplateIpsec6TunIfEsp, IpsecTun6Tests):
138 """ Ipsec ESP - TUN tests """
139 tun6_encrypt_node_name = "esp6-encrypt-tun"
140 tun6_decrypt_node_name = "esp6-decrypt"
142 def test_tun_basic46(self):
143 """ ipsec 4o6 tunnel basic test """
144 self.tun6_encrypt_node_name = "esp4-encrypt-tun"
145 self.verify_tun_46(self.params[socket.AF_INET6], count=1)
147 def test_tun_burst46(self):
148 """ ipsec 4o6 tunnel burst test """
149 self.tun6_encrypt_node_name = "esp4-encrypt-tun"
150 self.verify_tun_46(self.params[socket.AF_INET6], count=257)
153 class TestIpsec4MultiTunIfEsp(TemplateIpsec, IpsecTun4):
154 """ IPsec IPv4 Multi Tunnel interface """
156 encryption_type = ESP
157 tun4_encrypt_node_name = "esp4-encrypt-tun"
158 tun4_decrypt_node_name = "esp4-decrypt"
161 super(TestIpsec4MultiTunIfEsp, self).setUp()
163 self.tun_if = self.pg0
165 self.multi_params = []
168 p = copy.copy(self.ipv4_params)
170 p.remote_tun_if_host = "1.1.1.%d" % (ii + 1)
171 p.scapy_tun_sa_id = p.scapy_tun_sa_id + ii
172 p.scapy_tun_spi = p.scapy_tun_spi + ii
173 p.vpp_tun_sa_id = p.vpp_tun_sa_id + ii
174 p.vpp_tun_spi = p.vpp_tun_spi + ii
176 p.scapy_tra_sa_id = p.scapy_tra_sa_id + ii
177 p.scapy_tra_spi = p.scapy_tra_spi + ii
178 p.vpp_tra_sa_id = p.vpp_tra_sa_id + ii
179 p.vpp_tra_spi = p.vpp_tra_spi + ii
181 config_tun_params(p, self.encryption_type, self.tun_if)
182 self.multi_params.append(p)
184 p.tun_if = VppIpsecTunInterface(self, self.pg0, p.vpp_tun_spi,
187 p.crypt_key, p.crypt_key,
188 p.auth_algo_vpp_id, p.auth_key,
190 p.tun_if.add_vpp_config()
192 p.tun_if.config_ip4()
194 VppIpRoute(self, p.remote_tun_if_host, 32,
195 [VppRoutePath(p.tun_if.remote_ip4,
196 0xffffffff)]).add_vpp_config()
199 if not self.vpp_dead:
200 self.vapi.cli("show hardware")
201 super(TestIpsec4MultiTunIfEsp, self).tearDown()
203 def test_tun_44(self):
204 """Multiple IPSEC tunnel interfaces """
205 for p in self.multi_params:
206 self.verify_tun_44(p, count=127)
207 c = p.tun_if.get_rx_stats()
208 self.assertEqual(c['packets'], 127)
209 c = p.tun_if.get_tx_stats()
210 self.assertEqual(c['packets'], 127)
213 class TestIpsec4TunIfEspAll(TemplateIpsec, IpsecTun4):
214 """ IPsec IPv4 Tunnel interface all Algos """
216 encryption_type = ESP
217 tun4_encrypt_node_name = "esp4-encrypt-tun"
218 tun4_decrypt_node_name = "esp4-decrypt"
220 def config_network(self, p):
221 config_tun_params(p, self.encryption_type, self.tun_if)
223 p.tun_if = VppIpsecTunInterface(self, self.pg0, p.vpp_tun_spi,
226 p.crypt_key, p.crypt_key,
227 p.auth_algo_vpp_id, p.auth_key,
230 p.tun_if.add_vpp_config()
232 p.tun_if.config_ip4()
233 self.logger.info(self.vapi.cli("sh ipsec sa 0"))
234 self.logger.info(self.vapi.cli("sh ipsec sa 1"))
236 p.route = VppIpRoute(self, p.remote_tun_if_host, 32,
237 [VppRoutePath(p.tun_if.remote_ip4,
239 p.route.add_vpp_config()
241 def unconfig_network(self, p):
242 p.tun_if.unconfig_ip4()
243 p.tun_if.remove_vpp_config()
244 p.route.remove_vpp_config()
247 super(TestIpsec4TunIfEspAll, self).setUp()
249 self.tun_if = self.pg0
252 super(TestIpsec4TunIfEspAll, self).tearDown()
256 # change the key and the SPI
258 p.crypt_key = 'X' + p.crypt_key[1:]
260 p.scapy_tun_sa_id += 1
263 p.tun_if.local_spi = p.vpp_tun_spi
264 p.tun_if.remote_spi = p.scapy_tun_spi
266 config_tun_params(p, self.encryption_type, self.tun_if)
268 p.tun_sa_in = VppIpsecSA(self,
275 self.vpp_esp_protocol,
276 self.tun_if.local_addr[p.addr_type],
277 self.tun_if.remote_addr[p.addr_type],
280 p.tun_sa_out = VppIpsecSA(self,
287 self.vpp_esp_protocol,
288 self.tun_if.remote_addr[p.addr_type],
289 self.tun_if.local_addr[p.addr_type],
292 p.tun_sa_in.add_vpp_config()
293 p.tun_sa_out.add_vpp_config()
295 self.vapi.ipsec_tunnel_if_set_sa(sw_if_index=p.tun_if.sw_if_index,
296 sa_id=p.tun_sa_in.id,
298 self.vapi.ipsec_tunnel_if_set_sa(sw_if_index=p.tun_if.sw_if_index,
299 sa_id=p.tun_sa_out.id,
301 self.logger.info(self.vapi.cli("sh ipsec sa"))
303 def test_tun_44(self):
304 """IPSEC tunnel all algos """
306 # foreach VPP crypto engine
307 engines = ["ia32", "ipsecmb", "openssl"]
309 # foreach crypto algorithm
310 algos = [{'vpp-crypto': (VppEnum.vl_api_ipsec_crypto_alg_t.
311 IPSEC_API_CRYPTO_ALG_AES_GCM_128),
312 'vpp-integ': (VppEnum.vl_api_ipsec_integ_alg_t.
313 IPSEC_API_INTEG_ALG_NONE),
314 'scapy-crypto': "AES-GCM",
315 'scapy-integ': "NULL",
316 'key': "JPjyOWBeVEQiMe7h",
318 {'vpp-crypto': (VppEnum.vl_api_ipsec_crypto_alg_t.
319 IPSEC_API_CRYPTO_ALG_AES_GCM_192),
320 'vpp-integ': (VppEnum.vl_api_ipsec_integ_alg_t.
321 IPSEC_API_INTEG_ALG_NONE),
322 'scapy-crypto': "AES-GCM",
323 'scapy-integ': "NULL",
324 'key': "JPjyOWBeVEQiMe7hJPjyOWBe",
326 {'vpp-crypto': (VppEnum.vl_api_ipsec_crypto_alg_t.
327 IPSEC_API_CRYPTO_ALG_AES_GCM_256),
328 'vpp-integ': (VppEnum.vl_api_ipsec_integ_alg_t.
329 IPSEC_API_INTEG_ALG_NONE),
330 'scapy-crypto': "AES-GCM",
331 'scapy-integ': "NULL",
332 'key': "JPjyOWBeVEQiMe7hJPjyOWBeVEQiMe7h",
334 {'vpp-crypto': (VppEnum.vl_api_ipsec_crypto_alg_t.
335 IPSEC_API_CRYPTO_ALG_AES_CBC_128),
336 'vpp-integ': (VppEnum.vl_api_ipsec_integ_alg_t.
337 IPSEC_API_INTEG_ALG_SHA1_96),
338 'scapy-crypto': "AES-CBC",
339 'scapy-integ': "HMAC-SHA1-96",
341 'key': "JPjyOWBeVEQiMe7h"},
342 {'vpp-crypto': (VppEnum.vl_api_ipsec_crypto_alg_t.
343 IPSEC_API_CRYPTO_ALG_AES_CBC_192),
344 'vpp-integ': (VppEnum.vl_api_ipsec_integ_alg_t.
345 IPSEC_API_INTEG_ALG_SHA1_96),
346 'scapy-crypto': "AES-CBC",
347 'scapy-integ': "HMAC-SHA1-96",
349 'key': "JPjyOWBeVEQiMe7hJPjyOWBe"},
350 {'vpp-crypto': (VppEnum.vl_api_ipsec_crypto_alg_t.
351 IPSEC_API_CRYPTO_ALG_AES_CBC_256),
352 'vpp-integ': (VppEnum.vl_api_ipsec_integ_alg_t.
353 IPSEC_API_INTEG_ALG_SHA1_96),
354 'scapy-crypto': "AES-CBC",
355 'scapy-integ': "HMAC-SHA1-96",
357 'key': "JPjyOWBeVEQiMe7hJPjyOWBeVEQiMe7h"}]
359 for engine in engines:
360 self.vapi.cli("set crypto handler all %s" % engine)
363 # loop through each of the algorithms
366 # with self.subTest(algo=algo['scapy']):
368 p = copy.copy(self.ipv4_params)
369 p.auth_algo_vpp_id = algo['vpp-integ']
370 p.crypt_algo_vpp_id = algo['vpp-crypto']
371 p.crypt_algo = algo['scapy-crypto']
372 p.auth_algo = algo['scapy-integ']
373 p.crypt_key = algo['key']
374 p.salt = algo['salt']
376 self.config_network(p)
378 self.verify_tun_44(p, count=127)
379 c = p.tun_if.get_rx_stats()
380 self.assertEqual(c['packets'], 127)
381 c = p.tun_if.get_tx_stats()
382 self.assertEqual(c['packets'], 127)
388 self.verify_tun_44(p, count=127)
390 self.unconfig_network(p)
391 p.tun_sa_out.remove_vpp_config()
392 p.tun_sa_in.remove_vpp_config()
395 class TestIpsec6MultiTunIfEsp(TemplateIpsec, IpsecTun6):
396 """ IPsec IPv6 Multi Tunnel interface """
398 encryption_type = ESP
399 tun6_encrypt_node_name = "esp6-encrypt-tun"
400 tun6_decrypt_node_name = "esp6-decrypt"
403 super(TestIpsec6MultiTunIfEsp, self).setUp()
405 self.tun_if = self.pg0
407 self.multi_params = []
410 p = copy.copy(self.ipv6_params)
412 p.remote_tun_if_host = "1111::%d" % (ii + 1)
413 p.scapy_tun_sa_id = p.scapy_tun_sa_id + ii
414 p.scapy_tun_spi = p.scapy_tun_spi + ii
415 p.vpp_tun_sa_id = p.vpp_tun_sa_id + ii
416 p.vpp_tun_spi = p.vpp_tun_spi + ii
418 p.scapy_tra_sa_id = p.scapy_tra_sa_id + ii
419 p.scapy_tra_spi = p.scapy_tra_spi + ii
420 p.vpp_tra_sa_id = p.vpp_tra_sa_id + ii
421 p.vpp_tra_spi = p.vpp_tra_spi + ii
423 config_tun_params(p, self.encryption_type, self.tun_if)
424 self.multi_params.append(p)
426 p.tun_if = VppIpsecTunInterface(self, self.pg0, p.vpp_tun_spi,
429 p.crypt_key, p.crypt_key,
430 p.auth_algo_vpp_id, p.auth_key,
431 p.auth_key, is_ip6=True)
432 p.tun_if.add_vpp_config()
434 p.tun_if.config_ip6()
436 VppIpRoute(self, p.remote_tun_if_host, 128,
437 [VppRoutePath(p.tun_if.remote_ip6,
439 proto=DpoProto.DPO_PROTO_IP6)],
440 is_ip6=1).add_vpp_config()
443 if not self.vpp_dead:
444 self.vapi.cli("show hardware")
445 super(TestIpsec6MultiTunIfEsp, self).tearDown()
447 def test_tun_66(self):
448 """Multiple IPSEC tunnel interfaces """
449 for p in self.multi_params:
450 self.verify_tun_66(p, count=127)
451 c = p.tun_if.get_rx_stats()
452 self.assertEqual(c['packets'], 127)
453 c = p.tun_if.get_tx_stats()
454 self.assertEqual(c['packets'], 127)
457 class TemplateIpsecGRETunIfEsp(TemplateIpsec):
458 """ IPsec GRE tunnel interface tests """
460 encryption_type = ESP
461 omac = "00:11:22:33:44:55"
463 def gen_encrypt_pkts(self, sa, sw_intf, src, dst, count=1,
465 return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
466 sa.encrypt(IP(src=self.pg0.remote_ip4,
467 dst=self.pg0.local_ip4) /
469 Ether(dst=self.omac) /
470 IP(src="1.1.1.1", dst="1.1.1.2") /
471 UDP(sport=1144, dport=2233) /
472 Raw('X' * payload_size))
473 for i in range(count)]
475 def gen_pkts(self, sw_intf, src, dst, count=1,
477 return [Ether(dst=self.omac) /
478 IP(src="1.1.1.1", dst="1.1.1.2") /
479 UDP(sport=1144, dport=2233) /
480 Raw('X' * payload_size)
481 for i in range(count)]
483 def verify_decrypted(self, p, rxs):
485 self.assert_equal(rx[Ether].dst, self.omac)
486 self.assert_equal(rx[IP].dst, "1.1.1.2")
488 def verify_encrypted(self, p, sa, rxs):
491 pkt = sa.decrypt(rx[IP])
492 if not pkt.haslayer(IP):
493 pkt = IP(pkt[Raw].load)
494 self.assert_packet_checksums_valid(pkt)
495 self.assert_equal(pkt[IP].dst, self.pg0.remote_ip4)
496 self.assert_equal(pkt[IP].src, self.pg0.local_ip4)
497 self.assertTrue(pkt.haslayer(GRE))
499 self.assertEqual(e[Ether].dst, self.omac)
500 self.assertEqual(e[IP].dst, "1.1.1.2")
501 except (IndexError, AssertionError):
502 self.logger.debug(ppp("Unexpected packet:", rx))
504 self.logger.debug(ppp("Decrypted packet:", pkt))
510 super(TemplateIpsecGRETunIfEsp, self).setUp()
512 self.tun_if = self.pg0
516 bd1 = VppBridgeDomain(self, 1)
519 p.tun_sa_out = VppIpsecSA(self, p.scapy_tun_sa_id, p.scapy_tun_spi,
520 p.auth_algo_vpp_id, p.auth_key,
521 p.crypt_algo_vpp_id, p.crypt_key,
522 self.vpp_esp_protocol,
525 p.tun_sa_out.add_vpp_config()
527 p.tun_sa_in = VppIpsecSA(self, p.vpp_tun_sa_id, p.vpp_tun_spi,
528 p.auth_algo_vpp_id, p.auth_key,
529 p.crypt_algo_vpp_id, p.crypt_key,
530 self.vpp_esp_protocol,
533 p.tun_sa_in.add_vpp_config()
535 self.tun = VppIpsecGRETunInterface(self, self.pg0,
539 self.tun.add_vpp_config()
541 self.tun.config_ip4()
543 VppIpRoute(self, p.remote_tun_if_host, 32,
544 [VppRoutePath(self.tun.remote_ip4,
545 0xffffffff)]).add_vpp_config()
546 VppBridgeDomainPort(self, bd1, self.tun).add_vpp_config()
547 VppBridgeDomainPort(self, bd1, self.pg1).add_vpp_config()
550 if not self.vpp_dead:
551 self.vapi.cli("show hardware")
552 self.tun.unconfig_ip4()
553 super(TemplateIpsecGRETunIfEsp, self).tearDown()
556 @unittest.skipIf(is_skip_aarch64_set and is_platform_aarch64,
557 "test doesn't work on aarch64")
558 class TestIpsecGRETunIfEsp1(TemplateIpsecGRETunIfEsp, IpsecTun4Tests):
559 """ Ipsec GRE ESP - TUN tests """
560 tun4_encrypt_node_name = "esp4-encrypt"
561 tun4_decrypt_node_name = "esp4-decrypt"
563 if __name__ == '__main__':
564 unittest.main(testRunner=VppTestRunner)