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
18 class ConfigIpsecESP(TemplateIpsec):
20 tra4_encrypt_node_name = "esp4-encrypt"
21 tra4_decrypt_node_name = "esp4-decrypt"
22 tra6_encrypt_node_name = "esp6-encrypt"
23 tra6_decrypt_node_name = "esp6-decrypt"
24 tun4_encrypt_node_name = "esp4-encrypt"
25 tun4_decrypt_node_name = "esp4-decrypt"
26 tun6_encrypt_node_name = "esp6-encrypt"
27 tun6_decrypt_node_name = "esp6-decrypt"
31 super(ConfigIpsecESP, cls).setUpClass()
34 def tearDownClass(cls):
35 super(ConfigIpsecESP, cls).tearDownClass()
38 super(ConfigIpsecESP, self).setUp()
41 super(ConfigIpsecESP, self).tearDown()
43 def config_network(self, params):
45 self.tun_if = self.pg0
46 self.tra_if = self.pg2
47 self.logger.info(self.vapi.ppcli("show int addr"))
49 self.tra_spd = VppIpsecSpd(self, self.tra_spd_id)
50 self.tra_spd.add_vpp_config()
51 self.net_objs.append(self.tra_spd)
52 self.tun_spd = VppIpsecSpd(self, self.tun_spd_id)
53 self.tun_spd.add_vpp_config()
54 self.net_objs.append(self.tun_spd)
56 b = VppIpsecSpdItfBinding(self, self.tun_spd,
59 self.net_objs.append(b)
61 b = VppIpsecSpdItfBinding(self, self.tra_spd,
64 self.net_objs.append(b)
67 self.config_esp_tra(p)
68 config_tra_params(p, self.encryption_type)
70 self.config_esp_tun(p)
73 d = DpoProto.DPO_PROTO_IP6 if p.is_ipv6 else DpoProto.DPO_PROTO_IP4
74 r = VppIpRoute(self, p.remote_tun_if_host, p.addr_len,
75 [VppRoutePath(self.tun_if.remote_addr[p.addr_type],
80 self.net_objs.append(r)
82 self.logger.info(self.vapi.ppcli("show ipsec all"))
84 def unconfig_network(self):
85 for o in reversed(self.net_objs):
89 def config_esp_tun(self, params):
90 addr_type = params.addr_type
91 scapy_tun_sa_id = params.scapy_tun_sa_id
92 scapy_tun_spi = params.scapy_tun_spi
93 vpp_tun_sa_id = params.vpp_tun_sa_id
94 vpp_tun_spi = params.vpp_tun_spi
95 auth_algo_vpp_id = params.auth_algo_vpp_id
96 auth_key = params.auth_key
97 crypt_algo_vpp_id = params.crypt_algo_vpp_id
98 crypt_key = params.crypt_key
99 remote_tun_if_host = params.remote_tun_if_host
100 addr_any = params.addr_any
101 addr_bcast = params.addr_bcast
102 e = VppEnum.vl_api_ipsec_spd_action_t
105 params.tun_sa_in = VppIpsecSA(self, scapy_tun_sa_id, scapy_tun_spi,
106 auth_algo_vpp_id, auth_key,
107 crypt_algo_vpp_id, crypt_key,
108 self.vpp_esp_protocol,
109 self.tun_if.local_addr[addr_type],
110 self.tun_if.remote_addr[addr_type])
111 params.tun_sa_out = VppIpsecSA(self, vpp_tun_sa_id, vpp_tun_spi,
112 auth_algo_vpp_id, auth_key,
113 crypt_algo_vpp_id, crypt_key,
114 self.vpp_esp_protocol,
115 self.tun_if.remote_addr[addr_type],
116 self.tun_if.local_addr[addr_type])
117 objs.append(params.tun_sa_in)
118 objs.append(params.tun_sa_out)
120 params.spd_policy_in_any = VppIpsecSpdEntry(self, self.tun_spd,
122 addr_any, addr_bcast,
123 addr_any, addr_bcast,
125 params.spd_policy_out_any = VppIpsecSpdEntry(self, self.tun_spd,
127 addr_any, addr_bcast,
128 addr_any, addr_bcast,
131 objs.append(params.spd_policy_out_any)
132 objs.append(params.spd_policy_in_any)
134 objs.append(VppIpsecSpdEntry(self, self.tun_spd, vpp_tun_sa_id,
135 remote_tun_if_host, remote_tun_if_host,
136 self.pg1.remote_addr[addr_type],
137 self.pg1.remote_addr[addr_type],
140 policy=e.IPSEC_API_SPD_ACTION_PROTECT,
142 objs.append(VppIpsecSpdEntry(self, self.tun_spd, scapy_tun_sa_id,
143 self.pg1.remote_addr[addr_type],
144 self.pg1.remote_addr[addr_type],
145 remote_tun_if_host, remote_tun_if_host,
147 policy=e.IPSEC_API_SPD_ACTION_PROTECT,
149 objs.append(VppIpsecSpdEntry(self, self.tun_spd, vpp_tun_sa_id,
150 remote_tun_if_host, remote_tun_if_host,
151 self.pg0.local_addr[addr_type],
152 self.pg0.local_addr[addr_type],
155 policy=e.IPSEC_API_SPD_ACTION_PROTECT,
157 objs.append(VppIpsecSpdEntry(self, self.tun_spd, scapy_tun_sa_id,
158 self.pg0.local_addr[addr_type],
159 self.pg0.local_addr[addr_type],
160 remote_tun_if_host, remote_tun_if_host,
162 policy=e.IPSEC_API_SPD_ACTION_PROTECT,
166 self.net_objs = self.net_objs + objs
168 def config_esp_tra(self, params):
169 addr_type = params.addr_type
170 scapy_tra_sa_id = params.scapy_tra_sa_id
171 scapy_tra_spi = params.scapy_tra_spi
172 vpp_tra_sa_id = params.vpp_tra_sa_id
173 vpp_tra_spi = params.vpp_tra_spi
174 auth_algo_vpp_id = params.auth_algo_vpp_id
175 auth_key = params.auth_key
176 crypt_algo_vpp_id = params.crypt_algo_vpp_id
177 crypt_key = params.crypt_key
178 addr_any = params.addr_any
179 addr_bcast = params.addr_bcast
180 flags = (VppEnum.vl_api_ipsec_sad_flags_t.
181 IPSEC_API_SAD_FLAG_USE_ANTI_REPLAY)
182 e = VppEnum.vl_api_ipsec_spd_action_t
183 flags = params.flags | flags
186 params.tra_sa_in = VppIpsecSA(self, scapy_tra_sa_id, scapy_tra_spi,
187 auth_algo_vpp_id, auth_key,
188 crypt_algo_vpp_id, crypt_key,
189 self.vpp_esp_protocol,
191 params.tra_sa_out = VppIpsecSA(self, vpp_tra_sa_id, vpp_tra_spi,
192 auth_algo_vpp_id, auth_key,
193 crypt_algo_vpp_id, crypt_key,
194 self.vpp_esp_protocol,
196 objs.append(params.tra_sa_in)
197 objs.append(params.tra_sa_out)
199 objs.append(VppIpsecSpdEntry(self, self.tra_spd, vpp_tra_sa_id,
200 addr_any, addr_bcast,
201 addr_any, addr_bcast,
203 objs.append(VppIpsecSpdEntry(self, self.tra_spd, vpp_tra_sa_id,
204 addr_any, addr_bcast,
205 addr_any, addr_bcast,
208 objs.append(VppIpsecSpdEntry(self, self.tra_spd, vpp_tra_sa_id,
209 self.tra_if.local_addr[addr_type],
210 self.tra_if.local_addr[addr_type],
211 self.tra_if.remote_addr[addr_type],
212 self.tra_if.remote_addr[addr_type],
214 policy=e.IPSEC_API_SPD_ACTION_PROTECT,
216 objs.append(VppIpsecSpdEntry(self, self.tra_spd, scapy_tra_sa_id,
217 self.tra_if.local_addr[addr_type],
218 self.tra_if.local_addr[addr_type],
219 self.tra_if.remote_addr[addr_type],
220 self.tra_if.remote_addr[addr_type],
221 0, policy=e.IPSEC_API_SPD_ACTION_PROTECT,
225 self.net_objs = self.net_objs + objs
228 class TemplateIpsecEsp(ConfigIpsecESP):
230 Basic test for ipsec esp sanity - tunnel and transport modes.
232 Below 4 cases are covered as part of this test
233 1) ipsec esp v4 transport basic test - IPv4 Transport mode
234 scenario using HMAC-SHA1-96 integrity algo
235 2) ipsec esp v4 transport burst test
236 Above test for 257 pkts
237 3) ipsec esp 4o4 tunnel basic test - IPv4 Tunnel mode
238 scenario using HMAC-SHA1-96 integrity algo
239 4) ipsec esp 4o4 tunnel burst test
240 Above test for 257 pkts
245 |pg2| <-------> |VPP|
250 --- encrypt --- plain ---
251 |pg0| <------- |VPP| <------ |pg1|
254 --- decrypt --- plain ---
255 |pg0| -------> |VPP| ------> |pg1|
261 super(TemplateIpsecEsp, cls).setUpClass()
264 def tearDownClass(cls):
265 super(TemplateIpsecEsp, cls).tearDownClass()
268 super(TemplateIpsecEsp, self).setUp()
269 self.config_network(self.params.values())
272 self.unconfig_network()
273 super(TemplateIpsecEsp, self).tearDown()
276 class TestIpsecEsp1(TemplateIpsecEsp, IpsecTra46Tests, IpsecTun46Tests):
277 """ Ipsec ESP - TUN & TRA tests """
281 class TestIpsecEsp2(TemplateIpsecEsp, IpsecTcpTests):
282 """ Ipsec ESP - TCP tests """
286 class TemplateIpsecEspUdp(ConfigIpsecESP):
293 super(TemplateIpsecEspUdp, cls).setUpClass()
296 def tearDownClass(cls):
297 super(TemplateIpsecEspUdp, cls).tearDownClass()
300 super(TemplateIpsecEspUdp, self).setUp()
302 self.tun_if = self.pg0
303 self.tra_if = self.pg2
304 self.logger.info(self.vapi.ppcli("show int addr"))
307 p.flags = (VppEnum.vl_api_ipsec_sad_flags_t.
308 IPSEC_API_SAD_FLAG_UDP_ENCAP)
309 p.nat_header = UDP(sport=5454, dport=4500)
311 self.tra_spd = VppIpsecSpd(self, self.tra_spd_id)
312 self.tra_spd.add_vpp_config()
313 VppIpsecSpdItfBinding(self, self.tra_spd,
314 self.tra_if).add_vpp_config()
316 self.config_esp_tra(p)
317 config_tra_params(p, self.encryption_type)
319 self.tun_spd = VppIpsecSpd(self, self.tun_spd_id)
320 self.tun_spd.add_vpp_config()
321 VppIpsecSpdItfBinding(self, self.tun_spd,
322 self.tun_if).add_vpp_config()
324 self.config_esp_tun(p)
325 self.logger.info(self.vapi.ppcli("show ipsec all"))
327 d = DpoProto.DPO_PROTO_IP4
328 VppIpRoute(self, p.remote_tun_if_host, p.addr_len,
329 [VppRoutePath(self.tun_if.remote_addr[p.addr_type],
331 proto=d)]).add_vpp_config()
334 super(TemplateIpsecEspUdp, self).tearDown()
335 if not self.vpp_dead:
336 self.vapi.cli("show hardware")
339 class TestIpsecEspUdp(TemplateIpsecEspUdp, IpsecTra4Tests, IpsecTun4Tests):
340 """ Ipsec NAT-T ESP UDP tests """
344 class TestIpsecEspAll(ConfigIpsecESP,
345 IpsecTra4, IpsecTra6,
346 IpsecTun4, IpsecTun6):
347 """ Ipsec ESP all Algos """
350 super(TestIpsecEspAll, self).setUp()
353 super(TestIpsecEspAll, self).tearDown()
355 def test_crypto_algs(self):
356 """All engines AES-CBC-[128, 192, 256] w/o ESN"""
358 # foreach VPP crypto engine
359 engines = ["ia32", "ipsecmb", "openssl"]
361 # foreach crypto algorithm
362 algos = [{'vpp': VppEnum.vl_api_ipsec_crypto_alg_t.
363 IPSEC_API_CRYPTO_ALG_AES_CBC_128,
365 'key': "JPjyOWBeVEQiMe7h"},
366 {'vpp': VppEnum.vl_api_ipsec_crypto_alg_t.
367 IPSEC_API_CRYPTO_ALG_AES_CBC_192,
369 'key': "JPjyOWBeVEQiMe7hJPjyOWBe"},
370 {'vpp': VppEnum.vl_api_ipsec_crypto_alg_t.
371 IPSEC_API_CRYPTO_ALG_AES_CBC_256,
373 'key': "JPjyOWBeVEQiMe7hJPjyOWBeVEQiMe7h"}]
375 # bug found in VPP needs fixing with flag
376 # (VppEnum.vl_api_ipsec_sad_flags_t.IPSEC_API_SAD_FLAG_USE_ESN)
380 # loop through the VPP engines
382 for engine in engines:
383 self.vapi.cli("set crypto engine all %s" % engine)
386 # loop through each of the algorithms
389 # with self.subTest(algo=algo['scapy']):
392 # setup up the config paramters
394 self.ipv4_params = IPsecIPv4Params()
395 self.ipv6_params = IPsecIPv6Params()
397 self.params = {self.ipv4_params.addr_type:
399 self.ipv6_params.addr_type:
402 for _, p in self.params.items():
403 p.crypt_algo_vpp_id = algo['vpp']
404 p.crypt_algo = algo['scapy']
405 p.crypt_key = algo['key']
406 p.flags = p.flags | flag
409 # configure the SPDs. SAs, etc
411 self.config_network(self.params.values())
415 # An exhautsive 4o6, 6o4 is not necessary
418 self.verify_tra_basic6(count=17)
419 self.verify_tra_basic4(count=17)
420 self.verify_tun_66(self.params[socket.AF_INET6], 17)
421 self.verify_tun_44(self.params[socket.AF_INET], 17)
424 # remove the SPDs, SAs, etc
426 self.unconfig_network()
429 if __name__ == '__main__':
430 unittest.main(testRunner=VppTestRunner)