4 from scapy.layers.ipsec import ESP
5 from scapy.layers.inet import UDP
7 from framework import VppTestRunner
8 from template_ipsec import IpsecTra46Tests, IpsecTun46Tests, TemplateIpsec, \
9 IpsecTcpTests, IpsecTun4Tests, IpsecTra4Tests, config_tra_params, \
10 IPsecIPv4Params, IPsecIPv6Params, \
11 IpsecTra4, IpsecTun4, IpsecTra6, IpsecTun6
12 from vpp_ipsec import VppIpsecSpd, VppIpsecSpdEntry, VppIpsecSA,\
14 from vpp_ip_route import VppIpRoute, VppRoutePath
15 from vpp_ip import DpoProto
16 from vpp_papi import VppEnum
19 class ConfigIpsecESP(TemplateIpsec):
21 tra4_encrypt_node_name = "esp4-encrypt"
22 tra4_decrypt_node_name = "esp4-decrypt"
23 tra6_encrypt_node_name = "esp6-encrypt"
24 tra6_decrypt_node_name = "esp6-decrypt"
25 tun4_encrypt_node_name = "esp4-encrypt"
26 tun4_decrypt_node_name = "esp4-decrypt"
27 tun6_encrypt_node_name = "esp6-encrypt"
28 tun6_decrypt_node_name = "esp6-decrypt"
32 super(ConfigIpsecESP, cls).setUpClass()
35 def tearDownClass(cls):
36 super(ConfigIpsecESP, cls).tearDownClass()
39 super(ConfigIpsecESP, self).setUp()
42 super(ConfigIpsecESP, self).tearDown()
44 def config_network(self, params):
46 self.tun_if = self.pg0
47 self.tra_if = self.pg2
48 self.logger.info(self.vapi.ppcli("show int addr"))
50 self.tra_spd = VppIpsecSpd(self, self.tra_spd_id)
51 self.tra_spd.add_vpp_config()
52 self.net_objs.append(self.tra_spd)
53 self.tun_spd = VppIpsecSpd(self, self.tun_spd_id)
54 self.tun_spd.add_vpp_config()
55 self.net_objs.append(self.tun_spd)
57 b = VppIpsecSpdItfBinding(self, self.tun_spd,
60 self.net_objs.append(b)
62 b = VppIpsecSpdItfBinding(self, self.tra_spd,
65 self.net_objs.append(b)
68 self.config_esp_tra(p)
69 config_tra_params(p, self.encryption_type)
71 self.config_esp_tun(p)
74 d = DpoProto.DPO_PROTO_IP6 if p.is_ipv6 else DpoProto.DPO_PROTO_IP4
75 r = VppIpRoute(self, p.remote_tun_if_host, p.addr_len,
76 [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
107 params.tun_sa_in = VppIpsecSA(self, scapy_tun_sa_id, scapy_tun_spi,
108 auth_algo_vpp_id, auth_key,
109 crypt_algo_vpp_id, crypt_key,
110 self.vpp_esp_protocol,
111 self.tun_if.local_addr[addr_type],
112 self.tun_if.remote_addr[addr_type],
114 params.tun_sa_out = VppIpsecSA(self, vpp_tun_sa_id, vpp_tun_spi,
115 auth_algo_vpp_id, auth_key,
116 crypt_algo_vpp_id, crypt_key,
117 self.vpp_esp_protocol,
118 self.tun_if.remote_addr[addr_type],
119 self.tun_if.local_addr[addr_type],
121 objs.append(params.tun_sa_in)
122 objs.append(params.tun_sa_out)
124 params.spd_policy_in_any = VppIpsecSpdEntry(self, self.tun_spd,
126 addr_any, addr_bcast,
127 addr_any, addr_bcast,
129 params.spd_policy_out_any = VppIpsecSpdEntry(self, self.tun_spd,
131 addr_any, addr_bcast,
132 addr_any, addr_bcast,
135 objs.append(params.spd_policy_out_any)
136 objs.append(params.spd_policy_in_any)
138 objs.append(VppIpsecSpdEntry(self, self.tun_spd, vpp_tun_sa_id,
139 remote_tun_if_host, remote_tun_if_host,
140 self.pg1.remote_addr[addr_type],
141 self.pg1.remote_addr[addr_type],
144 policy=e.IPSEC_API_SPD_ACTION_PROTECT,
146 objs.append(VppIpsecSpdEntry(self, self.tun_spd, scapy_tun_sa_id,
147 self.pg1.remote_addr[addr_type],
148 self.pg1.remote_addr[addr_type],
149 remote_tun_if_host, remote_tun_if_host,
151 policy=e.IPSEC_API_SPD_ACTION_PROTECT,
153 objs.append(VppIpsecSpdEntry(self, self.tun_spd, vpp_tun_sa_id,
154 remote_tun_if_host, remote_tun_if_host,
155 self.pg0.local_addr[addr_type],
156 self.pg0.local_addr[addr_type],
159 policy=e.IPSEC_API_SPD_ACTION_PROTECT,
161 objs.append(VppIpsecSpdEntry(self, self.tun_spd, scapy_tun_sa_id,
162 self.pg0.local_addr[addr_type],
163 self.pg0.local_addr[addr_type],
164 remote_tun_if_host, remote_tun_if_host,
166 policy=e.IPSEC_API_SPD_ACTION_PROTECT,
170 self.net_objs = self.net_objs + objs
172 def config_esp_tra(self, params):
173 addr_type = params.addr_type
174 scapy_tra_sa_id = params.scapy_tra_sa_id
175 scapy_tra_spi = params.scapy_tra_spi
176 vpp_tra_sa_id = params.vpp_tra_sa_id
177 vpp_tra_spi = params.vpp_tra_spi
178 auth_algo_vpp_id = params.auth_algo_vpp_id
179 auth_key = params.auth_key
180 crypt_algo_vpp_id = params.crypt_algo_vpp_id
181 crypt_key = params.crypt_key
182 addr_any = params.addr_any
183 addr_bcast = params.addr_bcast
184 flags = (VppEnum.vl_api_ipsec_sad_flags_t.
185 IPSEC_API_SAD_FLAG_USE_ANTI_REPLAY)
186 e = VppEnum.vl_api_ipsec_spd_action_t
187 flags = params.flags | flags
190 params.tra_sa_in = VppIpsecSA(self, scapy_tra_sa_id, scapy_tra_spi,
191 auth_algo_vpp_id, auth_key,
192 crypt_algo_vpp_id, crypt_key,
193 self.vpp_esp_protocol,
195 params.tra_sa_out = VppIpsecSA(self, vpp_tra_sa_id, vpp_tra_spi,
196 auth_algo_vpp_id, auth_key,
197 crypt_algo_vpp_id, crypt_key,
198 self.vpp_esp_protocol,
200 objs.append(params.tra_sa_in)
201 objs.append(params.tra_sa_out)
203 objs.append(VppIpsecSpdEntry(self, self.tra_spd, vpp_tra_sa_id,
204 addr_any, addr_bcast,
205 addr_any, addr_bcast,
207 objs.append(VppIpsecSpdEntry(self, self.tra_spd, vpp_tra_sa_id,
208 addr_any, addr_bcast,
209 addr_any, addr_bcast,
212 objs.append(VppIpsecSpdEntry(self, self.tra_spd, vpp_tra_sa_id,
213 self.tra_if.local_addr[addr_type],
214 self.tra_if.local_addr[addr_type],
215 self.tra_if.remote_addr[addr_type],
216 self.tra_if.remote_addr[addr_type],
218 policy=e.IPSEC_API_SPD_ACTION_PROTECT,
220 objs.append(VppIpsecSpdEntry(self, self.tra_spd, scapy_tra_sa_id,
221 self.tra_if.local_addr[addr_type],
222 self.tra_if.local_addr[addr_type],
223 self.tra_if.remote_addr[addr_type],
224 self.tra_if.remote_addr[addr_type],
225 0, policy=e.IPSEC_API_SPD_ACTION_PROTECT,
229 self.net_objs = self.net_objs + objs
232 class TemplateIpsecEsp(ConfigIpsecESP):
234 Basic test for ipsec esp sanity - tunnel and transport modes.
236 Below 4 cases are covered as part of this test
237 1) ipsec esp v4 transport basic test - IPv4 Transport mode
238 scenario using HMAC-SHA1-96 integrity algo
239 2) ipsec esp v4 transport burst test
240 Above test for 257 pkts
241 3) ipsec esp 4o4 tunnel basic test - IPv4 Tunnel mode
242 scenario using HMAC-SHA1-96 integrity algo
243 4) ipsec esp 4o4 tunnel burst test
244 Above test for 257 pkts
249 |pg2| <-------> |VPP|
254 --- encrypt --- plain ---
255 |pg0| <------- |VPP| <------ |pg1|
258 --- decrypt --- plain ---
259 |pg0| -------> |VPP| ------> |pg1|
265 super(TemplateIpsecEsp, cls).setUpClass()
268 def tearDownClass(cls):
269 super(TemplateIpsecEsp, cls).tearDownClass()
272 super(TemplateIpsecEsp, self).setUp()
273 self.config_network(self.params.values())
276 self.unconfig_network()
277 super(TemplateIpsecEsp, self).tearDown()
280 class TestIpsecEsp1(TemplateIpsecEsp, IpsecTra46Tests, IpsecTun46Tests):
281 """ Ipsec ESP - TUN & TRA tests """
285 class TestIpsecEsp2(TemplateIpsecEsp, IpsecTcpTests):
286 """ Ipsec ESP - TCP tests """
290 class TemplateIpsecEspUdp(ConfigIpsecESP):
297 super(TemplateIpsecEspUdp, cls).setUpClass()
300 def tearDownClass(cls):
301 super(TemplateIpsecEspUdp, cls).tearDownClass()
304 super(TemplateIpsecEspUdp, self).setUp()
306 self.tun_if = self.pg0
307 self.tra_if = self.pg2
308 self.logger.info(self.vapi.ppcli("show int addr"))
311 p.flags = (VppEnum.vl_api_ipsec_sad_flags_t.
312 IPSEC_API_SAD_FLAG_UDP_ENCAP)
313 p.nat_header = UDP(sport=5454, dport=4500)
315 self.tra_spd = VppIpsecSpd(self, self.tra_spd_id)
316 self.tra_spd.add_vpp_config()
317 VppIpsecSpdItfBinding(self, self.tra_spd,
318 self.tra_if).add_vpp_config()
320 self.config_esp_tra(p)
321 config_tra_params(p, self.encryption_type)
323 self.tun_spd = VppIpsecSpd(self, self.tun_spd_id)
324 self.tun_spd.add_vpp_config()
325 VppIpsecSpdItfBinding(self, self.tun_spd,
326 self.tun_if).add_vpp_config()
328 self.config_esp_tun(p)
329 self.logger.info(self.vapi.ppcli("show ipsec all"))
331 d = DpoProto.DPO_PROTO_IP4
332 VppIpRoute(self, p.remote_tun_if_host, p.addr_len,
333 [VppRoutePath(self.tun_if.remote_addr[p.addr_type],
335 proto=d)]).add_vpp_config()
338 super(TemplateIpsecEspUdp, self).tearDown()
340 def show_commands_at_teardown(self):
341 self.logger.info(self.vapi.cli("show hardware"))
344 class TestIpsecEspUdp(TemplateIpsecEspUdp, IpsecTra4Tests):
345 """ Ipsec NAT-T ESP UDP tests """
349 class TestIpsecEspAll(ConfigIpsecESP,
350 IpsecTra4, IpsecTra6,
351 IpsecTun4, IpsecTun6):
352 """ Ipsec ESP all Algos """
355 super(TestIpsecEspAll, self).setUp()
358 super(TestIpsecEspAll, self).tearDown()
360 def test_crypto_algs(self):
361 """All engines AES-[CBC, GCM]-[128, 192, 256] w/ & w/o ESN"""
363 # foreach VPP crypto engine
364 engines = ["ia32", "ipsecmb", "openssl"]
366 # foreach crypto algorithm
367 algos = [{'vpp-crypto': (VppEnum.vl_api_ipsec_crypto_alg_t.
368 IPSEC_API_CRYPTO_ALG_AES_GCM_128),
369 'vpp-integ': (VppEnum.vl_api_ipsec_integ_alg_t.
370 IPSEC_API_INTEG_ALG_NONE),
371 'scapy-crypto': "AES-GCM",
372 'scapy-integ': "NULL",
373 'key': "JPjyOWBeVEQiMe7h",
374 'salt': struct.pack("!L", 0)},
375 {'vpp-crypto': (VppEnum.vl_api_ipsec_crypto_alg_t.
376 IPSEC_API_CRYPTO_ALG_AES_GCM_256),
377 'vpp-integ': (VppEnum.vl_api_ipsec_integ_alg_t.
378 IPSEC_API_INTEG_ALG_NONE),
379 'scapy-crypto': "AES-GCM",
380 'scapy-integ': "NULL",
381 'key': "JPjyOWBeVEQiMe7h0123456787654321",
382 'salt': struct.pack("!L", 0)},
383 {'vpp-crypto': (VppEnum.vl_api_ipsec_crypto_alg_t.
384 IPSEC_API_CRYPTO_ALG_AES_CBC_128),
385 'vpp-integ': (VppEnum.vl_api_ipsec_integ_alg_t.
386 IPSEC_API_INTEG_ALG_SHA1_96),
387 'scapy-crypto': "AES-CBC",
388 'scapy-integ': "HMAC-SHA1-96",
390 'key': "JPjyOWBeVEQiMe7h"},
391 {'vpp-crypto': (VppEnum.vl_api_ipsec_crypto_alg_t.
392 IPSEC_API_CRYPTO_ALG_AES_CBC_192),
393 'vpp-integ': (VppEnum.vl_api_ipsec_integ_alg_t.
394 IPSEC_API_INTEG_ALG_SHA1_96),
395 'scapy-crypto': "AES-CBC",
396 'scapy-integ': "HMAC-SHA1-96",
398 'key': "JPjyOWBeVEQiMe7hJPjyOWBe"},
399 {'vpp-crypto': (VppEnum.vl_api_ipsec_crypto_alg_t.
400 IPSEC_API_CRYPTO_ALG_AES_CBC_256),
401 'vpp-integ': (VppEnum.vl_api_ipsec_integ_alg_t.
402 IPSEC_API_INTEG_ALG_SHA1_96),
403 'scapy-crypto': "AES-CBC",
404 'scapy-integ': "HMAC-SHA1-96",
406 'key': "JPjyOWBeVEQiMe7hJPjyOWBeVEQiMe7h"}]
408 # with and without ESN
410 VppEnum.vl_api_ipsec_sad_flags_t.IPSEC_API_SAD_FLAG_USE_ESN]
413 # loop through the VPP engines
415 for engine in engines:
416 self.vapi.cli("set crypto handler all %s" % engine)
419 # loop through each of the algorithms
422 # with self.subTest(algo=algo['scapy']):
425 # setup up the config paramters
427 self.ipv4_params = IPsecIPv4Params()
428 self.ipv6_params = IPsecIPv6Params()
430 self.params = {self.ipv4_params.addr_type:
432 self.ipv6_params.addr_type:
435 for _, p in self.params.items():
436 p.auth_algo_vpp_id = algo['vpp-integ']
437 p.crypt_algo_vpp_id = algo['vpp-crypto']
438 p.crypt_algo = algo['scapy-crypto']
439 p.auth_algo = algo['scapy-integ']
440 p.crypt_key = algo['key']
441 p.crypt_salt = algo['salt']
442 p.flags = p.flags | flag
445 # configure the SPDs. SAs, etc
447 self.config_network(self.params.values())
451 # An exhautsive 4o6, 6o4 is not necessary
454 self.verify_tra_basic6(count=17)
455 self.verify_tra_basic4(count=17)
456 self.verify_tun_66(self.params[socket.AF_INET6], 17)
457 self.verify_tun_44(self.params[socket.AF_INET], 17)
460 # remove the SPDs, SAs, etc
462 self.unconfig_network()
465 if __name__ == '__main__':
466 unittest.main(testRunner=VppTestRunner)