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 IpsecTun6HandoffTests, IpsecTun4HandoffTests
11 from template_ipsec import IpsecTcpTests
12 from vpp_ipsec import VppIpsecSA, VppIpsecSpd, VppIpsecSpdEntry,\
14 from vpp_ip_route import VppIpRoute, VppRoutePath
15 from vpp_ip import DpoProto
16 from vpp_papi import VppEnum
19 class ConfigIpsecAH(TemplateIpsec):
21 Basic test for IPSEC using AH transport and Tunnel mode
31 --- encrypt --- plain ---
32 |pg0| <------- |VPP| <------ |pg1|
35 --- decrypt --- plain ---
36 |pg0| -------> |VPP| ------> |pg1|
41 tra4_encrypt_node_name = "ah4-encrypt"
42 tra4_decrypt_node_name = "ah4-decrypt"
43 tra6_encrypt_node_name = "ah6-encrypt"
44 tra6_decrypt_node_name = "ah6-decrypt"
45 tun4_encrypt_node_name = "ah4-encrypt"
46 tun4_decrypt_node_name = "ah4-decrypt"
47 tun6_encrypt_node_name = "ah6-encrypt"
48 tun6_decrypt_node_name = "ah6-decrypt"
52 super(ConfigIpsecAH, cls).setUpClass()
55 def tearDownClass(cls):
56 super(ConfigIpsecAH, cls).tearDownClass()
59 super(ConfigIpsecAH, self).setUp()
62 super(ConfigIpsecAH, self).tearDown()
64 def config_network(self, params):
66 self.tun_if = self.pg0
67 self.tra_if = self.pg2
68 self.logger.info(self.vapi.ppcli("show int addr"))
70 self.tra_spd = VppIpsecSpd(self, self.tra_spd_id)
71 self.tra_spd.add_vpp_config()
72 self.net_objs.append(self.tra_spd)
73 self.tun_spd = VppIpsecSpd(self, self.tun_spd_id)
74 self.tun_spd.add_vpp_config()
75 self.net_objs.append(self.tun_spd)
77 b = VppIpsecSpdItfBinding(self, self.tra_spd,
80 self.net_objs.append(b)
82 b = VppIpsecSpdItfBinding(self, self.tun_spd,
85 self.net_objs.append(b)
89 config_tra_params(p, self.encryption_type)
92 config_tun_params(p, self.encryption_type, self.tun_if)
94 d = DpoProto.DPO_PROTO_IP6 if p.is_ipv6 else DpoProto.DPO_PROTO_IP4
95 r = VppIpRoute(self, p.remote_tun_if_host, p.addr_len,
96 [VppRoutePath(self.tun_if.remote_addr[p.addr_type],
100 self.net_objs.append(r)
101 self.logger.info(self.vapi.ppcli("show ipsec all"))
103 def unconfig_network(self):
104 for o in reversed(self.net_objs):
105 o.remove_vpp_config()
108 def config_ah_tun(self, params):
109 addr_type = params.addr_type
110 scapy_tun_sa_id = params.scapy_tun_sa_id
111 scapy_tun_spi = params.scapy_tun_spi
112 vpp_tun_sa_id = params.vpp_tun_sa_id
113 vpp_tun_spi = params.vpp_tun_spi
114 auth_algo_vpp_id = params.auth_algo_vpp_id
115 auth_key = params.auth_key
116 crypt_algo_vpp_id = params.crypt_algo_vpp_id
117 crypt_key = params.crypt_key
118 remote_tun_if_host = params.remote_tun_if_host
119 addr_any = params.addr_any
120 addr_bcast = params.addr_bcast
122 e = VppEnum.vl_api_ipsec_spd_action_t
125 params.tun_sa_in = VppIpsecSA(self, scapy_tun_sa_id, scapy_tun_spi,
126 auth_algo_vpp_id, auth_key,
127 crypt_algo_vpp_id, crypt_key,
128 self.vpp_ah_protocol,
129 self.tun_if.local_addr[addr_type],
130 self.tun_if.remote_addr[addr_type],
133 params.tun_sa_out = VppIpsecSA(self, vpp_tun_sa_id, vpp_tun_spi,
134 auth_algo_vpp_id, auth_key,
135 crypt_algo_vpp_id, crypt_key,
136 self.vpp_ah_protocol,
137 self.tun_if.remote_addr[addr_type],
138 self.tun_if.local_addr[addr_type],
141 objs.append(params.tun_sa_in)
142 objs.append(params.tun_sa_out)
144 params.spd_policy_in_any = VppIpsecSpdEntry(self, self.tun_spd,
146 addr_any, addr_bcast,
147 addr_any, addr_bcast,
149 params.spd_policy_out_any = VppIpsecSpdEntry(self, self.tun_spd,
151 addr_any, addr_bcast,
152 addr_any, addr_bcast,
156 objs.append(params.spd_policy_out_any)
157 objs.append(params.spd_policy_in_any)
159 e1 = VppIpsecSpdEntry(self, self.tun_spd, vpp_tun_sa_id,
162 self.pg1.remote_addr[addr_type],
163 self.pg1.remote_addr[addr_type],
165 policy=e.IPSEC_API_SPD_ACTION_PROTECT,
167 e2 = VppIpsecSpdEntry(self, self.tun_spd, scapy_tun_sa_id,
168 self.pg1.remote_addr[addr_type],
169 self.pg1.remote_addr[addr_type],
172 0, policy=e.IPSEC_API_SPD_ACTION_PROTECT,
174 e3 = VppIpsecSpdEntry(self, self.tun_spd, vpp_tun_sa_id,
177 self.pg0.local_addr[addr_type],
178 self.pg0.local_addr[addr_type],
180 policy=e.IPSEC_API_SPD_ACTION_PROTECT,
182 e4 = VppIpsecSpdEntry(self, self.tun_spd, scapy_tun_sa_id,
183 self.pg0.local_addr[addr_type],
184 self.pg0.local_addr[addr_type],
187 0, policy=e.IPSEC_API_SPD_ACTION_PROTECT,
190 objs = objs + [e1, e2, e3, e4]
195 self.net_objs = self.net_objs + objs
197 def config_ah_tra(self, params):
198 addr_type = params.addr_type
199 scapy_tra_sa_id = params.scapy_tra_sa_id
200 scapy_tra_spi = params.scapy_tra_spi
201 vpp_tra_sa_id = params.vpp_tra_sa_id
202 vpp_tra_spi = params.vpp_tra_spi
203 auth_algo_vpp_id = params.auth_algo_vpp_id
204 auth_key = params.auth_key
205 crypt_algo_vpp_id = params.crypt_algo_vpp_id
206 crypt_key = params.crypt_key
207 addr_any = params.addr_any
208 addr_bcast = params.addr_bcast
209 flags = params.flags | (VppEnum.vl_api_ipsec_sad_flags_t.
210 IPSEC_API_SAD_FLAG_USE_ANTI_REPLAY)
211 e = VppEnum.vl_api_ipsec_spd_action_t
214 params.tra_sa_in = VppIpsecSA(self, scapy_tra_sa_id, scapy_tra_spi,
215 auth_algo_vpp_id, auth_key,
216 crypt_algo_vpp_id, crypt_key,
217 self.vpp_ah_protocol,
219 params.tra_sa_out = VppIpsecSA(self, vpp_tra_sa_id, vpp_tra_spi,
220 auth_algo_vpp_id, auth_key,
221 crypt_algo_vpp_id, crypt_key,
222 self.vpp_ah_protocol,
225 objs.append(params.tra_sa_in)
226 objs.append(params.tra_sa_out)
228 objs.append(VppIpsecSpdEntry(self, self.tra_spd, vpp_tra_sa_id,
229 addr_any, addr_bcast,
230 addr_any, addr_bcast,
232 objs.append(VppIpsecSpdEntry(self, self.tra_spd, scapy_tra_sa_id,
233 addr_any, addr_bcast,
234 addr_any, addr_bcast,
237 objs.append(VppIpsecSpdEntry(self, self.tra_spd, vpp_tra_sa_id,
238 self.tra_if.local_addr[addr_type],
239 self.tra_if.local_addr[addr_type],
240 self.tra_if.remote_addr[addr_type],
241 self.tra_if.remote_addr[addr_type],
243 policy=e.IPSEC_API_SPD_ACTION_PROTECT,
245 objs.append(VppIpsecSpdEntry(self, self.tra_spd, scapy_tra_sa_id,
246 self.tra_if.local_addr[addr_type],
247 self.tra_if.local_addr[addr_type],
248 self.tra_if.remote_addr[addr_type],
249 self.tra_if.remote_addr[addr_type],
250 0, policy=e.IPSEC_API_SPD_ACTION_PROTECT,
255 self.net_objs = self.net_objs + objs
258 class TemplateIpsecAh(ConfigIpsecAH):
260 Basic test for IPSEC using AH transport and Tunnel mode
265 |pg2| <-------> |VPP|
270 --- encrypt --- plain ---
271 |pg0| <------- |VPP| <------ |pg1|
274 --- decrypt --- plain ---
275 |pg0| -------> |VPP| ------> |pg1|
280 super(TemplateIpsecAh, cls).setUpClass()
283 def tearDownClass(cls):
284 super(TemplateIpsecAh, cls).tearDownClass()
287 super(TemplateIpsecAh, self).setUp()
288 self.config_network(self.params.values())
291 self.unconfig_network()
292 super(TemplateIpsecAh, self).tearDown()
295 class TestIpsecAh1(TemplateIpsecAh, IpsecTcpTests):
296 """ Ipsec AH - TCP tests """
300 class TestIpsecAh2(TemplateIpsecAh, IpsecTra46Tests, IpsecTun46Tests):
301 """ Ipsec AH w/ SHA1 """
305 class TestIpsecAhHandoff(TemplateIpsecAh,
306 IpsecTun6HandoffTests,
307 IpsecTun4HandoffTests):
308 """ Ipsec AH Handoff """
312 class TestIpsecAhAll(ConfigIpsecAH,
313 IpsecTra4, IpsecTra6,
314 IpsecTun4, IpsecTun6):
315 """ Ipsec AH all Algos """
318 super(TestIpsecAhAll, self).setUp()
321 super(TestIpsecAhAll, self).tearDown()
323 def test_integ_algs(self):
324 """All Engines SHA[1_96, 256, 384, 512] w/ & w/o ESN"""
325 # foreach VPP crypto engine
326 engines = ["ia32", "ipsecmb", "openssl"]
328 algos = [{'vpp': VppEnum.vl_api_ipsec_integ_alg_t.
329 IPSEC_API_INTEG_ALG_SHA1_96,
330 'scapy': "HMAC-SHA1-96"},
331 {'vpp': VppEnum.vl_api_ipsec_integ_alg_t.
332 IPSEC_API_INTEG_ALG_SHA_256_128,
333 'scapy': "SHA2-256-128"},
334 {'vpp': VppEnum.vl_api_ipsec_integ_alg_t.
335 IPSEC_API_INTEG_ALG_SHA_384_192,
336 'scapy': "SHA2-384-192"},
337 {'vpp': VppEnum.vl_api_ipsec_integ_alg_t.
338 IPSEC_API_INTEG_ALG_SHA_512_256,
339 'scapy': "SHA2-512-256"}]
341 flags = [0, (VppEnum.vl_api_ipsec_sad_flags_t.
342 IPSEC_API_SAD_FLAG_USE_ESN)]
345 # loop through the VPP engines
347 for engine in engines:
348 self.vapi.cli("set crypto handler all %s" % engine)
350 # loop through each of the algorithms
353 # with self.subTest(algo=algo['scapy']):
356 # setup up the config paramters
358 self.ipv4_params = IPsecIPv4Params()
359 self.ipv6_params = IPsecIPv6Params()
361 self.params = {self.ipv4_params.addr_type:
363 self.ipv6_params.addr_type:
366 for _, p in self.params.items():
367 p.auth_algo_vpp_id = algo['vpp']
368 p.auth_algo = algo['scapy']
369 p.flags = p.flags | flag
372 # configure the SPDs. SAs, etc
374 self.config_network(self.params.values())
378 # An exhautsive 4o6, 6o4 is not necessary for each algo
380 self.verify_tra_basic6(count=17)
381 self.verify_tra_basic4(count=17)
382 self.verify_tun_66(self.params[socket.AF_INET6], count=17)
383 self.verify_tun_44(self.params[socket.AF_INET], count=17)
386 # remove the SPDs, SAs, etc
388 self.unconfig_network()
391 if __name__ == '__main__':
392 unittest.main(testRunner=VppTestRunner)