4 from scapy.layers.ipsec import AH
6 from framework import VppTestRunner
7 from template_ipsec import TemplateIpsec, IpsecTra46Tests, IpsecTun46Tests, \
8 config_tun_params, config_tra_params, IPsecIPv4Params, IPsecIPv6Params, \
9 IpsecTra4, IpsecTun4, IpsecTra6, IpsecTun6
10 from template_ipsec import IpsecTcpTests
11 from vpp_ipsec import VppIpsecSA, VppIpsecSpd, VppIpsecSpdEntry,\
13 from vpp_ip_route import VppIpRoute, VppRoutePath
14 from vpp_ip import DpoProto
15 from vpp_papi import VppEnum
18 class ConfigIpsecAH(TemplateIpsec):
20 Basic test for IPSEC using AH transport and Tunnel mode
30 --- encrypt --- plain ---
31 |pg0| <------- |VPP| <------ |pg1|
34 --- decrypt --- plain ---
35 |pg0| -------> |VPP| ------> |pg1|
40 tra4_encrypt_node_name = "ah4-encrypt"
41 tra4_decrypt_node_name = "ah4-decrypt"
42 tra6_encrypt_node_name = "ah6-encrypt"
43 tra6_decrypt_node_name = "ah6-decrypt"
44 tun4_encrypt_node_name = "ah4-encrypt"
45 tun4_decrypt_node_name = "ah4-decrypt"
46 tun6_encrypt_node_name = "ah6-encrypt"
47 tun6_decrypt_node_name = "ah6-decrypt"
51 super(ConfigIpsecAH, cls).setUpClass()
54 def tearDownClass(cls):
55 super(ConfigIpsecAH, cls).tearDownClass()
58 super(ConfigIpsecAH, self).setUp()
61 super(ConfigIpsecAH, self).tearDown()
63 def config_network(self, params):
65 self.tun_if = self.pg0
66 self.tra_if = self.pg2
67 self.logger.info(self.vapi.ppcli("show int addr"))
69 self.tra_spd = VppIpsecSpd(self, self.tra_spd_id)
70 self.tra_spd.add_vpp_config()
71 self.net_objs.append(self.tra_spd)
72 self.tun_spd = VppIpsecSpd(self, self.tun_spd_id)
73 self.tun_spd.add_vpp_config()
74 self.net_objs.append(self.tun_spd)
76 b = VppIpsecSpdItfBinding(self, self.tra_spd,
79 self.net_objs.append(b)
81 b = VppIpsecSpdItfBinding(self, self.tun_spd,
84 self.net_objs.append(b)
88 config_tra_params(p, self.encryption_type)
92 d = DpoProto.DPO_PROTO_IP6 if p.is_ipv6 else DpoProto.DPO_PROTO_IP4
93 r = VppIpRoute(self, p.remote_tun_if_host, p.addr_len,
94 [VppRoutePath(self.tun_if.remote_addr[p.addr_type],
99 self.net_objs.append(r)
100 self.logger.info(self.vapi.ppcli("show ipsec all"))
102 def unconfig_network(self):
103 for o in reversed(self.net_objs):
104 o.remove_vpp_config()
107 def config_ah_tun(self, params):
108 addr_type = params.addr_type
109 scapy_tun_sa_id = params.scapy_tun_sa_id
110 scapy_tun_spi = params.scapy_tun_spi
111 vpp_tun_sa_id = params.vpp_tun_sa_id
112 vpp_tun_spi = params.vpp_tun_spi
113 auth_algo_vpp_id = params.auth_algo_vpp_id
114 auth_key = params.auth_key
115 crypt_algo_vpp_id = params.crypt_algo_vpp_id
116 crypt_key = params.crypt_key
117 remote_tun_if_host = params.remote_tun_if_host
118 addr_any = params.addr_any
119 addr_bcast = params.addr_bcast
121 e = VppEnum.vl_api_ipsec_spd_action_t
124 params.tun_sa_in = VppIpsecSA(self, scapy_tun_sa_id, scapy_tun_spi,
125 auth_algo_vpp_id, auth_key,
126 crypt_algo_vpp_id, crypt_key,
127 self.vpp_ah_protocol,
128 self.tun_if.local_addr[addr_type],
129 self.tun_if.remote_addr[addr_type],
132 params.tun_sa_out = VppIpsecSA(self, vpp_tun_sa_id, vpp_tun_spi,
133 auth_algo_vpp_id, auth_key,
134 crypt_algo_vpp_id, crypt_key,
135 self.vpp_ah_protocol,
136 self.tun_if.remote_addr[addr_type],
137 self.tun_if.local_addr[addr_type],
140 objs.append(params.tun_sa_in)
141 objs.append(params.tun_sa_out)
143 params.spd_policy_in_any = VppIpsecSpdEntry(self, self.tun_spd,
145 addr_any, addr_bcast,
146 addr_any, addr_bcast,
148 params.spd_policy_out_any = VppIpsecSpdEntry(self, self.tun_spd,
150 addr_any, addr_bcast,
151 addr_any, addr_bcast,
155 objs.append(params.spd_policy_out_any)
156 objs.append(params.spd_policy_in_any)
158 e1 = VppIpsecSpdEntry(self, self.tun_spd, vpp_tun_sa_id,
161 self.pg1.remote_addr[addr_type],
162 self.pg1.remote_addr[addr_type],
164 policy=e.IPSEC_API_SPD_ACTION_PROTECT,
166 e2 = VppIpsecSpdEntry(self, self.tun_spd, scapy_tun_sa_id,
167 self.pg1.remote_addr[addr_type],
168 self.pg1.remote_addr[addr_type],
171 0, policy=e.IPSEC_API_SPD_ACTION_PROTECT,
173 e3 = VppIpsecSpdEntry(self, self.tun_spd, vpp_tun_sa_id,
176 self.pg0.local_addr[addr_type],
177 self.pg0.local_addr[addr_type],
179 policy=e.IPSEC_API_SPD_ACTION_PROTECT,
181 e4 = VppIpsecSpdEntry(self, self.tun_spd, scapy_tun_sa_id,
182 self.pg0.local_addr[addr_type],
183 self.pg0.local_addr[addr_type],
186 0, policy=e.IPSEC_API_SPD_ACTION_PROTECT,
189 objs = objs + [e1, e2, e3, e4]
194 self.net_objs = self.net_objs + objs
196 def config_ah_tra(self, params):
197 addr_type = params.addr_type
198 scapy_tra_sa_id = params.scapy_tra_sa_id
199 scapy_tra_spi = params.scapy_tra_spi
200 vpp_tra_sa_id = params.vpp_tra_sa_id
201 vpp_tra_spi = params.vpp_tra_spi
202 auth_algo_vpp_id = params.auth_algo_vpp_id
203 auth_key = params.auth_key
204 crypt_algo_vpp_id = params.crypt_algo_vpp_id
205 crypt_key = params.crypt_key
206 addr_any = params.addr_any
207 addr_bcast = params.addr_bcast
208 flags = params.flags | (VppEnum.vl_api_ipsec_sad_flags_t.
209 IPSEC_API_SAD_FLAG_USE_ANTI_REPLAY)
210 e = VppEnum.vl_api_ipsec_spd_action_t
213 params.tra_sa_in = VppIpsecSA(self, scapy_tra_sa_id, scapy_tra_spi,
214 auth_algo_vpp_id, auth_key,
215 crypt_algo_vpp_id, crypt_key,
216 self.vpp_ah_protocol,
218 params.tra_sa_out = VppIpsecSA(self, vpp_tra_sa_id, vpp_tra_spi,
219 auth_algo_vpp_id, auth_key,
220 crypt_algo_vpp_id, crypt_key,
221 self.vpp_ah_protocol,
224 objs.append(params.tra_sa_in)
225 objs.append(params.tra_sa_out)
227 objs.append(VppIpsecSpdEntry(self, self.tra_spd, vpp_tra_sa_id,
228 addr_any, addr_bcast,
229 addr_any, addr_bcast,
231 objs.append(VppIpsecSpdEntry(self, self.tra_spd, scapy_tra_sa_id,
232 addr_any, addr_bcast,
233 addr_any, addr_bcast,
236 objs.append(VppIpsecSpdEntry(self, self.tra_spd, vpp_tra_sa_id,
237 self.tra_if.local_addr[addr_type],
238 self.tra_if.local_addr[addr_type],
239 self.tra_if.remote_addr[addr_type],
240 self.tra_if.remote_addr[addr_type],
242 policy=e.IPSEC_API_SPD_ACTION_PROTECT,
244 objs.append(VppIpsecSpdEntry(self, self.tra_spd, scapy_tra_sa_id,
245 self.tra_if.local_addr[addr_type],
246 self.tra_if.local_addr[addr_type],
247 self.tra_if.remote_addr[addr_type],
248 self.tra_if.remote_addr[addr_type],
249 0, policy=e.IPSEC_API_SPD_ACTION_PROTECT,
254 self.net_objs = self.net_objs + objs
257 class TemplateIpsecAh(ConfigIpsecAH):
259 Basic test for IPSEC using AH transport and Tunnel mode
264 |pg2| <-------> |VPP|
269 --- encrypt --- plain ---
270 |pg0| <------- |VPP| <------ |pg1|
273 --- decrypt --- plain ---
274 |pg0| -------> |VPP| ------> |pg1|
279 super(TemplateIpsecAh, cls).setUpClass()
282 def tearDownClass(cls):
283 super(TemplateIpsecAh, cls).tearDownClass()
286 super(TemplateIpsecAh, self).setUp()
287 self.config_network(self.params.values())
290 self.unconfig_network()
291 super(TemplateIpsecAh, self).tearDown()
294 class TestIpsecAh1(TemplateIpsecAh, IpsecTcpTests):
295 """ Ipsec AH - TCP tests """
299 class TestIpsecAh2(TemplateIpsecAh, IpsecTra46Tests, IpsecTun46Tests):
300 """ Ipsec AH w/ SHA1 """
304 class TestIpsecAhAll(ConfigIpsecAH,
305 IpsecTra4, IpsecTra6,
306 IpsecTun4, IpsecTun6):
307 """ Ipsec AH all Algos """
310 super(TestIpsecAhAll, self).setUp()
313 super(TestIpsecAhAll, self).tearDown()
315 def test_integ_algs(self):
316 """All Engines SHA[1_96, 256, 384, 512] w/ & w/o ESN"""
317 # foreach VPP crypto engine
318 engines = ["ia32", "ipsecmb", "openssl"]
320 algos = [{'vpp': VppEnum.vl_api_ipsec_integ_alg_t.
321 IPSEC_API_INTEG_ALG_SHA1_96,
322 'scapy': "HMAC-SHA1-96"},
323 {'vpp': VppEnum.vl_api_ipsec_integ_alg_t.
324 IPSEC_API_INTEG_ALG_SHA_256_128,
325 'scapy': "SHA2-256-128"},
326 {'vpp': VppEnum.vl_api_ipsec_integ_alg_t.
327 IPSEC_API_INTEG_ALG_SHA_384_192,
328 'scapy': "SHA2-384-192"},
329 {'vpp': VppEnum.vl_api_ipsec_integ_alg_t.
330 IPSEC_API_INTEG_ALG_SHA_512_256,
331 'scapy': "SHA2-512-256"}]
333 flags = [0, (VppEnum.vl_api_ipsec_sad_flags_t.
334 IPSEC_API_SAD_FLAG_USE_ESN)]
337 # loop through the VPP engines
339 for engine in engines:
340 self.vapi.cli("set crypto handler all %s" % engine)
342 # loop through each of the algorithms
345 # with self.subTest(algo=algo['scapy']):
348 # setup up the config paramters
350 self.ipv4_params = IPsecIPv4Params()
351 self.ipv6_params = IPsecIPv6Params()
353 self.params = {self.ipv4_params.addr_type:
355 self.ipv6_params.addr_type:
358 for _, p in self.params.items():
359 p.auth_algo_vpp_id = algo['vpp']
360 p.auth_algo = algo['scapy']
361 p.flags = p.flags | flag
364 # configure the SPDs. SAs, etc
366 self.config_network(self.params.values())
370 # An exhautsive 4o6, 6o4 is not necessary for each algo
372 self.verify_tra_basic6(count=17)
373 self.verify_tra_basic4(count=17)
374 self.verify_tun_66(self.params[socket.AF_INET6], count=17)
375 self.verify_tun_44(self.params[socket.AF_INET], count=17)
378 # remove the SPDs, SAs, etc
380 self.unconfig_network()
383 if __name__ == '__main__':
384 unittest.main(testRunner=VppTestRunner)