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 r = VppIpRoute(self, p.remote_tun_if_host, 32,
51 [VppRoutePath(p.tun_if.remote_ip4,
54 r = VppIpRoute(self, p.remote_tun_if_host6, 128,
55 [VppRoutePath(p.tun_if.remote_ip6,
57 proto=DpoProto.DPO_PROTO_IP6)])
62 self.vapi.cli("show hardware")
63 super(TemplateIpsec4TunIfEsp, self).tearDown()
66 class TestIpsec4TunIfEsp1(TemplateIpsec4TunIfEsp, IpsecTun4Tests):
67 """ Ipsec ESP - TUN tests """
68 tun4_encrypt_node_name = "esp4-encrypt-tun"
69 tun4_decrypt_node_name = "esp4-decrypt"
71 def test_tun_basic64(self):
72 """ ipsec 6o4 tunnel basic test """
73 self.tun4_encrypt_node_name = "esp6-encrypt-tun"
75 self.verify_tun_64(self.params[socket.AF_INET], count=1)
77 def test_tun_burst64(self):
78 """ ipsec 6o4 tunnel basic test """
79 self.tun4_encrypt_node_name = "esp6-encrypt-tun"
81 self.verify_tun_64(self.params[socket.AF_INET], count=257)
83 def test_tun_basic_frag44(self):
84 """ ipsec 4o4 tunnel frag basic test """
85 self.tun4_encrypt_node_name = "esp4-encrypt-tun"
89 self.vapi.sw_interface_set_mtu(p.tun_if.sw_if_index,
91 self.verify_tun_44(self.params[socket.AF_INET],
92 count=1, payload_size=1800, n_rx=2)
93 self.vapi.sw_interface_set_mtu(p.tun_if.sw_if_index,
97 class TestIpsec4TunIfEsp2(TemplateIpsec4TunIfEsp, IpsecTcpTests):
98 """ Ipsec ESP - TCP tests """
102 class TemplateIpsec6TunIfEsp(TemplateIpsec):
103 """ IPsec tunnel interface tests """
105 encryption_type = ESP
108 super(TemplateIpsec6TunIfEsp, self).setUp()
110 self.tun_if = self.pg0
113 tun_if = VppIpsecTunInterface(self, self.pg0, p.vpp_tun_spi,
114 p.scapy_tun_spi, p.crypt_algo_vpp_id,
115 p.crypt_key, p.crypt_key,
116 p.auth_algo_vpp_id, p.auth_key,
117 p.auth_key, is_ip6=True)
118 tun_if.add_vpp_config()
123 r = VppIpRoute(self, p.remote_tun_if_host, 128,
124 [VppRoutePath(tun_if.remote_ip6,
126 proto=DpoProto.DPO_PROTO_IP6)])
128 r = VppIpRoute(self, p.remote_tun_if_host4, 32,
129 [VppRoutePath(tun_if.remote_ip4,
134 if not self.vpp_dead:
135 self.vapi.cli("show hardware")
136 super(TemplateIpsec6TunIfEsp, self).tearDown()
139 class TestIpsec6TunIfEsp1(TemplateIpsec6TunIfEsp, IpsecTun6Tests):
140 """ Ipsec ESP - TUN tests """
141 tun6_encrypt_node_name = "esp6-encrypt-tun"
142 tun6_decrypt_node_name = "esp6-decrypt"
144 def test_tun_basic46(self):
145 """ ipsec 4o6 tunnel basic test """
146 self.tun6_encrypt_node_name = "esp4-encrypt-tun"
147 self.verify_tun_46(self.params[socket.AF_INET6], count=1)
149 def test_tun_burst46(self):
150 """ ipsec 4o6 tunnel burst test """
151 self.tun6_encrypt_node_name = "esp4-encrypt-tun"
152 self.verify_tun_46(self.params[socket.AF_INET6], count=257)
155 class TestIpsec4MultiTunIfEsp(TemplateIpsec, IpsecTun4):
156 """ IPsec IPv4 Multi Tunnel interface """
158 encryption_type = ESP
159 tun4_encrypt_node_name = "esp4-encrypt-tun"
160 tun4_decrypt_node_name = "esp4-decrypt"
163 super(TestIpsec4MultiTunIfEsp, self).setUp()
165 self.tun_if = self.pg0
167 self.multi_params = []
170 p = copy.copy(self.ipv4_params)
172 p.remote_tun_if_host = "1.1.1.%d" % (ii + 1)
173 p.scapy_tun_sa_id = p.scapy_tun_sa_id + ii
174 p.scapy_tun_spi = p.scapy_tun_spi + ii
175 p.vpp_tun_sa_id = p.vpp_tun_sa_id + ii
176 p.vpp_tun_spi = p.vpp_tun_spi + ii
178 p.scapy_tra_sa_id = p.scapy_tra_sa_id + ii
179 p.scapy_tra_spi = p.scapy_tra_spi + ii
180 p.vpp_tra_sa_id = p.vpp_tra_sa_id + ii
181 p.vpp_tra_spi = p.vpp_tra_spi + ii
183 config_tun_params(p, self.encryption_type, self.tun_if)
184 self.multi_params.append(p)
186 p.tun_if = VppIpsecTunInterface(self, self.pg0, p.vpp_tun_spi,
189 p.crypt_key, p.crypt_key,
190 p.auth_algo_vpp_id, p.auth_key,
192 p.tun_if.add_vpp_config()
194 p.tun_if.config_ip4()
196 VppIpRoute(self, p.remote_tun_if_host, 32,
197 [VppRoutePath(p.tun_if.remote_ip4,
198 0xffffffff)]).add_vpp_config()
201 if not self.vpp_dead:
202 self.vapi.cli("show hardware")
203 super(TestIpsec4MultiTunIfEsp, self).tearDown()
205 def test_tun_44(self):
206 """Multiple IPSEC tunnel interfaces """
207 for p in self.multi_params:
208 self.verify_tun_44(p, count=127)
209 c = p.tun_if.get_rx_stats()
210 self.assertEqual(c['packets'], 127)
211 c = p.tun_if.get_tx_stats()
212 self.assertEqual(c['packets'], 127)
215 class TestIpsec4TunIfEspAll(TemplateIpsec, IpsecTun4):
216 """ IPsec IPv4 Tunnel interface all Algos """
218 encryption_type = ESP
219 tun4_encrypt_node_name = "esp4-encrypt-tun"
220 tun4_decrypt_node_name = "esp4-decrypt"
222 def config_network(self, p):
223 config_tun_params(p, self.encryption_type, self.tun_if)
225 p.tun_if = VppIpsecTunInterface(self, self.pg0, p.vpp_tun_spi,
228 p.crypt_key, p.crypt_key,
229 p.auth_algo_vpp_id, p.auth_key,
232 p.tun_if.add_vpp_config()
234 p.tun_if.config_ip4()
235 self.logger.info(self.vapi.cli("sh ipsec sa 0"))
236 self.logger.info(self.vapi.cli("sh ipsec sa 1"))
238 p.route = VppIpRoute(self, p.remote_tun_if_host, 32,
239 [VppRoutePath(p.tun_if.remote_ip4,
241 p.route.add_vpp_config()
243 def unconfig_network(self, p):
244 p.tun_if.unconfig_ip4()
245 p.tun_if.remove_vpp_config()
246 p.route.remove_vpp_config()
249 super(TestIpsec4TunIfEspAll, self).setUp()
251 self.tun_if = self.pg0
254 super(TestIpsec4TunIfEspAll, self).tearDown()
258 # change the key and the SPI
260 p.crypt_key = 'X' + p.crypt_key[1:]
262 p.scapy_tun_sa_id += 1
265 p.tun_if.local_spi = p.vpp_tun_spi
266 p.tun_if.remote_spi = p.scapy_tun_spi
268 config_tun_params(p, self.encryption_type, self.tun_if)
270 p.tun_sa_in = VppIpsecSA(self,
277 self.vpp_esp_protocol,
278 self.tun_if.local_addr[p.addr_type],
279 self.tun_if.remote_addr[p.addr_type],
282 p.tun_sa_out = VppIpsecSA(self,
289 self.vpp_esp_protocol,
290 self.tun_if.remote_addr[p.addr_type],
291 self.tun_if.local_addr[p.addr_type],
294 p.tun_sa_in.add_vpp_config()
295 p.tun_sa_out.add_vpp_config()
297 self.vapi.ipsec_tunnel_if_set_sa(sw_if_index=p.tun_if.sw_if_index,
298 sa_id=p.tun_sa_in.id,
300 self.vapi.ipsec_tunnel_if_set_sa(sw_if_index=p.tun_if.sw_if_index,
301 sa_id=p.tun_sa_out.id,
303 self.logger.info(self.vapi.cli("sh ipsec sa"))
305 def test_tun_44(self):
306 """IPSEC tunnel all algos """
308 # foreach VPP crypto engine
309 engines = ["ia32", "ipsecmb", "openssl"]
311 # foreach crypto algorithm
312 algos = [{'vpp-crypto': (VppEnum.vl_api_ipsec_crypto_alg_t.
313 IPSEC_API_CRYPTO_ALG_AES_GCM_128),
314 'vpp-integ': (VppEnum.vl_api_ipsec_integ_alg_t.
315 IPSEC_API_INTEG_ALG_NONE),
316 'scapy-crypto': "AES-GCM",
317 'scapy-integ': "NULL",
318 'key': "JPjyOWBeVEQiMe7h",
320 {'vpp-crypto': (VppEnum.vl_api_ipsec_crypto_alg_t.
321 IPSEC_API_CRYPTO_ALG_AES_GCM_192),
322 'vpp-integ': (VppEnum.vl_api_ipsec_integ_alg_t.
323 IPSEC_API_INTEG_ALG_NONE),
324 'scapy-crypto': "AES-GCM",
325 'scapy-integ': "NULL",
326 'key': "JPjyOWBeVEQiMe7hJPjyOWBe",
328 {'vpp-crypto': (VppEnum.vl_api_ipsec_crypto_alg_t.
329 IPSEC_API_CRYPTO_ALG_AES_GCM_256),
330 'vpp-integ': (VppEnum.vl_api_ipsec_integ_alg_t.
331 IPSEC_API_INTEG_ALG_NONE),
332 'scapy-crypto': "AES-GCM",
333 'scapy-integ': "NULL",
334 'key': "JPjyOWBeVEQiMe7hJPjyOWBeVEQiMe7h",
336 {'vpp-crypto': (VppEnum.vl_api_ipsec_crypto_alg_t.
337 IPSEC_API_CRYPTO_ALG_AES_CBC_128),
338 'vpp-integ': (VppEnum.vl_api_ipsec_integ_alg_t.
339 IPSEC_API_INTEG_ALG_SHA1_96),
340 'scapy-crypto': "AES-CBC",
341 'scapy-integ': "HMAC-SHA1-96",
343 'key': "JPjyOWBeVEQiMe7h"},
344 {'vpp-crypto': (VppEnum.vl_api_ipsec_crypto_alg_t.
345 IPSEC_API_CRYPTO_ALG_AES_CBC_192),
346 'vpp-integ': (VppEnum.vl_api_ipsec_integ_alg_t.
347 IPSEC_API_INTEG_ALG_SHA1_96),
348 'scapy-crypto': "AES-CBC",
349 'scapy-integ': "HMAC-SHA1-96",
351 'key': "JPjyOWBeVEQiMe7hJPjyOWBe"},
352 {'vpp-crypto': (VppEnum.vl_api_ipsec_crypto_alg_t.
353 IPSEC_API_CRYPTO_ALG_AES_CBC_256),
354 'vpp-integ': (VppEnum.vl_api_ipsec_integ_alg_t.
355 IPSEC_API_INTEG_ALG_SHA1_96),
356 'scapy-crypto': "AES-CBC",
357 'scapy-integ': "HMAC-SHA1-96",
359 'key': "JPjyOWBeVEQiMe7hJPjyOWBeVEQiMe7h"}]
361 for engine in engines:
362 self.vapi.cli("set crypto handler all %s" % engine)
365 # loop through each of the algorithms
368 # with self.subTest(algo=algo['scapy']):
370 p = copy.copy(self.ipv4_params)
371 p.auth_algo_vpp_id = algo['vpp-integ']
372 p.crypt_algo_vpp_id = algo['vpp-crypto']
373 p.crypt_algo = algo['scapy-crypto']
374 p.auth_algo = algo['scapy-integ']
375 p.crypt_key = algo['key']
376 p.salt = algo['salt']
378 self.config_network(p)
380 self.verify_tun_44(p, count=127)
381 c = p.tun_if.get_rx_stats()
382 self.assertEqual(c['packets'], 127)
383 c = p.tun_if.get_tx_stats()
384 self.assertEqual(c['packets'], 127)
390 self.verify_tun_44(p, count=127)
392 self.unconfig_network(p)
393 p.tun_sa_out.remove_vpp_config()
394 p.tun_sa_in.remove_vpp_config()
397 class TestIpsec6MultiTunIfEsp(TemplateIpsec, IpsecTun6):
398 """ IPsec IPv6 Multi Tunnel interface """
400 encryption_type = ESP
401 tun6_encrypt_node_name = "esp6-encrypt-tun"
402 tun6_decrypt_node_name = "esp6-decrypt"
405 super(TestIpsec6MultiTunIfEsp, self).setUp()
407 self.tun_if = self.pg0
409 self.multi_params = []
412 p = copy.copy(self.ipv6_params)
414 p.remote_tun_if_host = "1111::%d" % (ii + 1)
415 p.scapy_tun_sa_id = p.scapy_tun_sa_id + ii
416 p.scapy_tun_spi = p.scapy_tun_spi + ii
417 p.vpp_tun_sa_id = p.vpp_tun_sa_id + ii
418 p.vpp_tun_spi = p.vpp_tun_spi + ii
420 p.scapy_tra_sa_id = p.scapy_tra_sa_id + ii
421 p.scapy_tra_spi = p.scapy_tra_spi + ii
422 p.vpp_tra_sa_id = p.vpp_tra_sa_id + ii
423 p.vpp_tra_spi = p.vpp_tra_spi + ii
425 config_tun_params(p, self.encryption_type, self.tun_if)
426 self.multi_params.append(p)
428 p.tun_if = VppIpsecTunInterface(self, self.pg0, p.vpp_tun_spi,
431 p.crypt_key, p.crypt_key,
432 p.auth_algo_vpp_id, p.auth_key,
433 p.auth_key, is_ip6=True)
434 p.tun_if.add_vpp_config()
436 p.tun_if.config_ip6()
438 r = VppIpRoute(self, p.remote_tun_if_host, 128,
439 [VppRoutePath(p.tun_if.remote_ip6,
441 proto=DpoProto.DPO_PROTO_IP6)])
445 if not self.vpp_dead:
446 self.vapi.cli("show hardware")
447 super(TestIpsec6MultiTunIfEsp, self).tearDown()
449 def test_tun_66(self):
450 """Multiple IPSEC tunnel interfaces """
451 for p in self.multi_params:
452 self.verify_tun_66(p, count=127)
453 c = p.tun_if.get_rx_stats()
454 self.assertEqual(c['packets'], 127)
455 c = p.tun_if.get_tx_stats()
456 self.assertEqual(c['packets'], 127)
459 class TemplateIpsecGRETunIfEsp(TemplateIpsec):
460 """ IPsec GRE tunnel interface tests """
462 encryption_type = ESP
463 omac = "00:11:22:33:44:55"
465 def gen_encrypt_pkts(self, sa, sw_intf, src, dst, count=1,
467 return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
468 sa.encrypt(IP(src=self.pg0.remote_ip4,
469 dst=self.pg0.local_ip4) /
471 Ether(dst=self.omac) /
472 IP(src="1.1.1.1", dst="1.1.1.2") /
473 UDP(sport=1144, dport=2233) /
474 Raw('X' * payload_size))
475 for i in range(count)]
477 def gen_pkts(self, sw_intf, src, dst, count=1,
479 return [Ether(dst=self.omac) /
480 IP(src="1.1.1.1", dst="1.1.1.2") /
481 UDP(sport=1144, dport=2233) /
482 Raw('X' * payload_size)
483 for i in range(count)]
485 def verify_decrypted(self, p, rxs):
487 self.assert_equal(rx[Ether].dst, self.omac)
488 self.assert_equal(rx[IP].dst, "1.1.1.2")
490 def verify_encrypted(self, p, sa, rxs):
493 pkt = sa.decrypt(rx[IP])
494 if not pkt.haslayer(IP):
495 pkt = IP(pkt[Raw].load)
496 self.assert_packet_checksums_valid(pkt)
497 self.assert_equal(pkt[IP].dst, self.pg0.remote_ip4)
498 self.assert_equal(pkt[IP].src, self.pg0.local_ip4)
499 self.assertTrue(pkt.haslayer(GRE))
501 self.assertEqual(e[Ether].dst, self.omac)
502 self.assertEqual(e[IP].dst, "1.1.1.2")
503 except (IndexError, AssertionError):
504 self.logger.debug(ppp("Unexpected packet:", rx))
506 self.logger.debug(ppp("Decrypted packet:", pkt))
512 super(TemplateIpsecGRETunIfEsp, self).setUp()
514 self.tun_if = self.pg0
518 bd1 = VppBridgeDomain(self, 1)
521 p.tun_sa_out = VppIpsecSA(self, p.scapy_tun_sa_id, p.scapy_tun_spi,
522 p.auth_algo_vpp_id, p.auth_key,
523 p.crypt_algo_vpp_id, p.crypt_key,
524 self.vpp_esp_protocol,
527 p.tun_sa_out.add_vpp_config()
529 p.tun_sa_in = VppIpsecSA(self, p.vpp_tun_sa_id, p.vpp_tun_spi,
530 p.auth_algo_vpp_id, p.auth_key,
531 p.crypt_algo_vpp_id, p.crypt_key,
532 self.vpp_esp_protocol,
535 p.tun_sa_in.add_vpp_config()
537 self.tun = VppIpsecGRETunInterface(self, self.pg0,
541 self.tun.add_vpp_config()
543 self.tun.config_ip4()
545 VppIpRoute(self, p.remote_tun_if_host, 32,
546 [VppRoutePath(self.tun.remote_ip4,
547 0xffffffff)]).add_vpp_config()
548 VppBridgeDomainPort(self, bd1, self.tun).add_vpp_config()
549 VppBridgeDomainPort(self, bd1, self.pg1).add_vpp_config()
552 if not self.vpp_dead:
553 self.vapi.cli("show hardware")
554 self.tun.unconfig_ip4()
555 super(TemplateIpsecGRETunIfEsp, self).tearDown()
558 @unittest.skipIf(is_skip_aarch64_set and is_platform_aarch64,
559 "test doesn't work on aarch64")
560 class TestIpsecGRETunIfEsp1(TemplateIpsecGRETunIfEsp, IpsecTun4Tests):
561 """ Ipsec GRE ESP - TUN tests """
562 tun4_encrypt_node_name = "esp4-encrypt"
563 tun4_decrypt_node_name = "esp4-decrypt"
565 if __name__ == '__main__':
566 unittest.main(testRunner=VppTestRunner)