3 from scapy.layers.ipsec import ESP
4 from scapy.layers.inet import UDP
6 from framework import VppTestRunner
7 from template_ipsec import IpsecTra46Tests, IpsecTun46Tests, TemplateIpsec, \
8 IpsecTcpTests, IpsecTun4Tests, IpsecTra4Tests, config_tra_params, \
9 IPsecIPv4Params, IPsecIPv6Params, \
10 IpsecTra4, IpsecTun4, IpsecTra6, IpsecTun6
11 from vpp_ipsec import VppIpsecSpd, VppIpsecSpdEntry, VppIpsecSA,\
13 from vpp_ip_route import VppIpRoute, VppRoutePath
14 from vpp_ip import DpoProto
15 from vpp_papi import VppEnum
20 class ConfigIpsecESP(TemplateIpsec):
22 tra4_encrypt_node_name = "esp4-encrypt"
23 tra4_decrypt_node_name = "esp4-decrypt"
24 tra6_encrypt_node_name = "esp6-encrypt"
25 tra6_decrypt_node_name = "esp6-decrypt"
26 tun4_encrypt_node_name = "esp4-encrypt"
27 tun4_decrypt_node_name = "esp4-decrypt"
28 tun6_encrypt_node_name = "esp6-encrypt"
29 tun6_decrypt_node_name = "esp6-decrypt"
33 super(ConfigIpsecESP, cls).setUpClass()
36 def tearDownClass(cls):
37 super(ConfigIpsecESP, cls).tearDownClass()
40 super(ConfigIpsecESP, self).setUp()
43 super(ConfigIpsecESP, self).tearDown()
45 def config_network(self, params):
47 self.tun_if = self.pg0
48 self.tra_if = self.pg2
49 self.logger.info(self.vapi.ppcli("show int addr"))
51 self.tra_spd = VppIpsecSpd(self, self.tra_spd_id)
52 self.tra_spd.add_vpp_config()
53 self.net_objs.append(self.tra_spd)
54 self.tun_spd = VppIpsecSpd(self, self.tun_spd_id)
55 self.tun_spd.add_vpp_config()
56 self.net_objs.append(self.tun_spd)
58 b = VppIpsecSpdItfBinding(self, self.tun_spd,
61 self.net_objs.append(b)
63 b = VppIpsecSpdItfBinding(self, self.tra_spd,
66 self.net_objs.append(b)
69 self.config_esp_tra(p)
70 config_tra_params(p, self.encryption_type)
72 self.config_esp_tun(p)
75 d = DpoProto.DPO_PROTO_IP6 if p.is_ipv6 else DpoProto.DPO_PROTO_IP4
76 r = VppIpRoute(self, p.remote_tun_if_host, p.addr_len,
77 [VppRoutePath(self.tun_if.remote_addr[p.addr_type],
81 self.net_objs.append(r)
83 self.logger.info(self.vapi.ppcli("show ipsec all"))
85 def unconfig_network(self):
86 for o in reversed(self.net_objs):
90 def config_esp_tun(self, params):
91 addr_type = params.addr_type
92 scapy_tun_sa_id = params.scapy_tun_sa_id
93 scapy_tun_spi = params.scapy_tun_spi
94 vpp_tun_sa_id = params.vpp_tun_sa_id
95 vpp_tun_spi = params.vpp_tun_spi
96 auth_algo_vpp_id = params.auth_algo_vpp_id
97 auth_key = params.auth_key
98 crypt_algo_vpp_id = params.crypt_algo_vpp_id
99 crypt_key = params.crypt_key
100 remote_tun_if_host = params.remote_tun_if_host
101 addr_any = params.addr_any
102 addr_bcast = params.addr_bcast
103 e = VppEnum.vl_api_ipsec_spd_action_t
108 params.tun_sa_in = VppIpsecSA(self, scapy_tun_sa_id, scapy_tun_spi,
109 auth_algo_vpp_id, auth_key,
110 crypt_algo_vpp_id, crypt_key,
111 self.vpp_esp_protocol,
112 self.tun_if.local_addr[addr_type],
113 self.tun_if.remote_addr[addr_type],
116 params.tun_sa_out = VppIpsecSA(self, vpp_tun_sa_id, vpp_tun_spi,
117 auth_algo_vpp_id, auth_key,
118 crypt_algo_vpp_id, crypt_key,
119 self.vpp_esp_protocol,
120 self.tun_if.remote_addr[addr_type],
121 self.tun_if.local_addr[addr_type],
124 objs.append(params.tun_sa_in)
125 objs.append(params.tun_sa_out)
127 params.spd_policy_in_any = VppIpsecSpdEntry(self, self.tun_spd,
129 addr_any, addr_bcast,
130 addr_any, addr_bcast,
132 params.spd_policy_out_any = VppIpsecSpdEntry(self, self.tun_spd,
134 addr_any, addr_bcast,
135 addr_any, addr_bcast,
138 objs.append(params.spd_policy_out_any)
139 objs.append(params.spd_policy_in_any)
141 objs.append(VppIpsecSpdEntry(self, self.tun_spd, vpp_tun_sa_id,
142 remote_tun_if_host, remote_tun_if_host,
143 self.pg1.remote_addr[addr_type],
144 self.pg1.remote_addr[addr_type],
147 policy=e.IPSEC_API_SPD_ACTION_PROTECT,
149 objs.append(VppIpsecSpdEntry(self, self.tun_spd, scapy_tun_sa_id,
150 self.pg1.remote_addr[addr_type],
151 self.pg1.remote_addr[addr_type],
152 remote_tun_if_host, remote_tun_if_host,
154 policy=e.IPSEC_API_SPD_ACTION_PROTECT,
156 objs.append(VppIpsecSpdEntry(self, self.tun_spd, vpp_tun_sa_id,
157 remote_tun_if_host, remote_tun_if_host,
158 self.pg0.local_addr[addr_type],
159 self.pg0.local_addr[addr_type],
162 policy=e.IPSEC_API_SPD_ACTION_PROTECT,
164 objs.append(VppIpsecSpdEntry(self, self.tun_spd, scapy_tun_sa_id,
165 self.pg0.local_addr[addr_type],
166 self.pg0.local_addr[addr_type],
167 remote_tun_if_host, remote_tun_if_host,
169 policy=e.IPSEC_API_SPD_ACTION_PROTECT,
173 self.net_objs = self.net_objs + objs
175 def config_esp_tra(self, params):
176 addr_type = params.addr_type
177 scapy_tra_sa_id = params.scapy_tra_sa_id
178 scapy_tra_spi = params.scapy_tra_spi
179 vpp_tra_sa_id = params.vpp_tra_sa_id
180 vpp_tra_spi = params.vpp_tra_spi
181 auth_algo_vpp_id = params.auth_algo_vpp_id
182 auth_key = params.auth_key
183 crypt_algo_vpp_id = params.crypt_algo_vpp_id
184 crypt_key = params.crypt_key
185 addr_any = params.addr_any
186 addr_bcast = params.addr_bcast
187 flags = (VppEnum.vl_api_ipsec_sad_flags_t.
188 IPSEC_API_SAD_FLAG_USE_ANTI_REPLAY)
189 e = VppEnum.vl_api_ipsec_spd_action_t
190 flags = params.flags | flags
194 params.tra_sa_in = VppIpsecSA(self, scapy_tra_sa_id, scapy_tra_spi,
195 auth_algo_vpp_id, auth_key,
196 crypt_algo_vpp_id, crypt_key,
197 self.vpp_esp_protocol,
200 params.tra_sa_out = VppIpsecSA(self, vpp_tra_sa_id, vpp_tra_spi,
201 auth_algo_vpp_id, auth_key,
202 crypt_algo_vpp_id, crypt_key,
203 self.vpp_esp_protocol,
206 objs.append(params.tra_sa_in)
207 objs.append(params.tra_sa_out)
209 objs.append(VppIpsecSpdEntry(self, self.tra_spd, vpp_tra_sa_id,
210 addr_any, addr_bcast,
211 addr_any, addr_bcast,
213 objs.append(VppIpsecSpdEntry(self, self.tra_spd, vpp_tra_sa_id,
214 addr_any, addr_bcast,
215 addr_any, addr_bcast,
218 objs.append(VppIpsecSpdEntry(self, self.tra_spd, vpp_tra_sa_id,
219 self.tra_if.local_addr[addr_type],
220 self.tra_if.local_addr[addr_type],
221 self.tra_if.remote_addr[addr_type],
222 self.tra_if.remote_addr[addr_type],
224 policy=e.IPSEC_API_SPD_ACTION_PROTECT,
226 objs.append(VppIpsecSpdEntry(self, self.tra_spd, scapy_tra_sa_id,
227 self.tra_if.local_addr[addr_type],
228 self.tra_if.local_addr[addr_type],
229 self.tra_if.remote_addr[addr_type],
230 self.tra_if.remote_addr[addr_type],
231 0, policy=e.IPSEC_API_SPD_ACTION_PROTECT,
235 self.net_objs = self.net_objs + objs
238 class TemplateIpsecEsp(ConfigIpsecESP):
240 Basic test for ipsec esp sanity - tunnel and transport modes.
242 Below 4 cases are covered as part of this test
243 1) ipsec esp v4 transport basic test - IPv4 Transport mode
244 scenario using HMAC-SHA1-96 integrity algo
245 2) ipsec esp v4 transport burst test
246 Above test for 257 pkts
247 3) ipsec esp 4o4 tunnel basic test - IPv4 Tunnel mode
248 scenario using HMAC-SHA1-96 integrity algo
249 4) ipsec esp 4o4 tunnel burst test
250 Above test for 257 pkts
255 |pg2| <-------> |VPP|
260 --- encrypt --- plain ---
261 |pg0| <------- |VPP| <------ |pg1|
264 --- decrypt --- plain ---
265 |pg0| -------> |VPP| ------> |pg1|
271 super(TemplateIpsecEsp, cls).setUpClass()
274 def tearDownClass(cls):
275 super(TemplateIpsecEsp, cls).tearDownClass()
278 super(TemplateIpsecEsp, self).setUp()
279 self.config_network(self.params.values())
282 self.unconfig_network()
283 super(TemplateIpsecEsp, self).tearDown()
286 class TestIpsecEsp1(TemplateIpsecEsp, IpsecTra46Tests, IpsecTun46Tests):
287 """ Ipsec ESP - TUN & TRA tests """
291 class TestIpsecEsp2(TemplateIpsecEsp, IpsecTcpTests):
292 """ Ipsec ESP - TCP tests """
296 class TemplateIpsecEspUdp(ConfigIpsecESP):
303 super(TemplateIpsecEspUdp, cls).setUpClass()
306 def tearDownClass(cls):
307 super(TemplateIpsecEspUdp, cls).tearDownClass()
310 super(TemplateIpsecEspUdp, self).setUp()
312 self.tun_if = self.pg0
313 self.tra_if = self.pg2
314 self.logger.info(self.vapi.ppcli("show int addr"))
317 p.flags = (VppEnum.vl_api_ipsec_sad_flags_t.
318 IPSEC_API_SAD_FLAG_UDP_ENCAP)
319 p.nat_header = UDP(sport=5454, dport=4500)
321 self.tra_spd = VppIpsecSpd(self, self.tra_spd_id)
322 self.tra_spd.add_vpp_config()
323 VppIpsecSpdItfBinding(self, self.tra_spd,
324 self.tra_if).add_vpp_config()
326 self.config_esp_tra(p)
327 config_tra_params(p, self.encryption_type)
329 self.tun_spd = VppIpsecSpd(self, self.tun_spd_id)
330 self.tun_spd.add_vpp_config()
331 VppIpsecSpdItfBinding(self, self.tun_spd,
332 self.tun_if).add_vpp_config()
334 self.config_esp_tun(p)
335 self.logger.info(self.vapi.ppcli("show ipsec all"))
337 d = DpoProto.DPO_PROTO_IP4
338 VppIpRoute(self, p.remote_tun_if_host, p.addr_len,
339 [VppRoutePath(self.tun_if.remote_addr[p.addr_type],
341 proto=d)]).add_vpp_config()
344 super(TemplateIpsecEspUdp, self).tearDown()
346 def show_commands_at_teardown(self):
347 self.logger.info(self.vapi.cli("show hardware"))
350 class TestIpsecEspUdp(TemplateIpsecEspUdp, IpsecTra4Tests):
351 """ Ipsec NAT-T ESP UDP tests """
355 class TestIpsecEspAll(ConfigIpsecESP,
356 IpsecTra4, IpsecTra6,
357 IpsecTun4, IpsecTun6):
358 """ Ipsec ESP all Algos """
361 super(TestIpsecEspAll, self).setUp()
364 super(TestIpsecEspAll, self).tearDown()
366 def test_crypto_algs(self):
367 """All engines AES-[CBC, GCM]-[128, 192, 256] 3DES-CBC w/ & w/o ESN"""
369 # foreach VPP crypto engine
370 engines = ["ia32", "ipsecmb", "openssl"]
372 # foreach crypto algorithm
373 algos = [{'vpp-crypto': (VppEnum.vl_api_ipsec_crypto_alg_t.
374 IPSEC_API_CRYPTO_ALG_AES_GCM_128),
375 'vpp-integ': (VppEnum.vl_api_ipsec_integ_alg_t.
376 IPSEC_API_INTEG_ALG_NONE),
377 'scapy-crypto': "AES-GCM",
378 'scapy-integ': "NULL",
379 'key': "JPjyOWBeVEQiMe7h",
381 {'vpp-crypto': (VppEnum.vl_api_ipsec_crypto_alg_t.
382 IPSEC_API_CRYPTO_ALG_AES_GCM_192),
383 'vpp-integ': (VppEnum.vl_api_ipsec_integ_alg_t.
384 IPSEC_API_INTEG_ALG_NONE),
385 'scapy-crypto': "AES-GCM",
386 'scapy-integ': "NULL",
387 'key': "JPjyOWBeVEQiMe7h01234567",
389 {'vpp-crypto': (VppEnum.vl_api_ipsec_crypto_alg_t.
390 IPSEC_API_CRYPTO_ALG_AES_GCM_256),
391 'vpp-integ': (VppEnum.vl_api_ipsec_integ_alg_t.
392 IPSEC_API_INTEG_ALG_NONE),
393 'scapy-crypto': "AES-GCM",
394 'scapy-integ': "NULL",
395 'key': "JPjyOWBeVEQiMe7h0123456787654321",
397 {'vpp-crypto': (VppEnum.vl_api_ipsec_crypto_alg_t.
398 IPSEC_API_CRYPTO_ALG_AES_CBC_128),
399 'vpp-integ': (VppEnum.vl_api_ipsec_integ_alg_t.
400 IPSEC_API_INTEG_ALG_SHA1_96),
401 'scapy-crypto': "AES-CBC",
402 'scapy-integ': "HMAC-SHA1-96",
404 'key': "JPjyOWBeVEQiMe7h"},
405 {'vpp-crypto': (VppEnum.vl_api_ipsec_crypto_alg_t.
406 IPSEC_API_CRYPTO_ALG_AES_CBC_192),
407 'vpp-integ': (VppEnum.vl_api_ipsec_integ_alg_t.
408 IPSEC_API_INTEG_ALG_SHA1_96),
409 'scapy-crypto': "AES-CBC",
410 'scapy-integ': "HMAC-SHA1-96",
412 'key': "JPjyOWBeVEQiMe7hJPjyOWBe"},
413 {'vpp-crypto': (VppEnum.vl_api_ipsec_crypto_alg_t.
414 IPSEC_API_CRYPTO_ALG_AES_CBC_256),
415 'vpp-integ': (VppEnum.vl_api_ipsec_integ_alg_t.
416 IPSEC_API_INTEG_ALG_SHA1_96),
417 'scapy-crypto': "AES-CBC",
418 'scapy-integ': "HMAC-SHA1-96",
420 'key': "JPjyOWBeVEQiMe7hJPjyOWBeVEQiMe7h"},
421 {'vpp-crypto': (VppEnum.vl_api_ipsec_crypto_alg_t.
422 IPSEC_API_CRYPTO_ALG_3DES_CBC),
423 'vpp-integ': (VppEnum.vl_api_ipsec_integ_alg_t.
424 IPSEC_API_INTEG_ALG_SHA1_96),
425 'scapy-crypto': "3DES",
426 'scapy-integ': "HMAC-SHA1-96",
428 'key': "JPjyOWBeVEQiMe7h00112233"}]
430 # with and without ESN
432 VppEnum.vl_api_ipsec_sad_flags_t.IPSEC_API_SAD_FLAG_USE_ESN]
435 # loop through the VPP engines
437 for engine in engines:
438 self.vapi.cli("set crypto handler all %s" % engine)
440 # loop through each of the algorithms
443 # with self.subTest(algo=algo['scapy']):
446 # setup up the config paramters
448 self.ipv4_params = IPsecIPv4Params()
449 self.ipv6_params = IPsecIPv6Params()
451 self.params = {self.ipv4_params.addr_type:
453 self.ipv6_params.addr_type:
456 for _, p in self.params.items():
457 p.auth_algo_vpp_id = algo['vpp-integ']
458 p.crypt_algo_vpp_id = algo['vpp-crypto']
459 p.crypt_algo = algo['scapy-crypto']
460 p.auth_algo = algo['scapy-integ']
461 p.crypt_key = algo['key']
462 p.salt = algo['salt']
463 p.flags = p.flags | flag
465 self.reporter.send_keep_alive(self)
468 # configure the SPDs. SAs, etc
470 self.config_network(self.params.values())
474 # An exhautsive 4o6, 6o4 is not necessary
477 self.verify_tra_basic6(count=NUM_PKTS)
478 self.verify_tra_basic4(count=NUM_PKTS)
479 self.verify_tun_66(self.params[socket.AF_INET6],
481 self.verify_tun_44(self.params[socket.AF_INET],
485 # remove the SPDs, SAs, etc
487 self.unconfig_network()
490 # reconfigure the network and SA to run the
493 self.config_network(self.params.values())
494 self.verify_tra_anti_replay()
495 self.unconfig_network()
498 if __name__ == '__main__':
499 unittest.main(testRunner=VppTestRunner)