4 from scapy.layers.ipsec import AH
5 from scapy.layers.inet import IP, UDP
6 from scapy.layers.inet6 import IPv6
7 from scapy.layers.l2 import Ether
8 from scapy.packet import Raw
10 from framework import VppTestRunner
11 from template_ipsec import TemplateIpsec, IpsecTra46Tests, IpsecTun46Tests, \
12 config_tun_params, config_tra_params, IPsecIPv4Params, IPsecIPv6Params, \
13 IpsecTra4, IpsecTun4, IpsecTra6, IpsecTun6, \
14 IpsecTun6HandoffTests, IpsecTun4HandoffTests
15 from template_ipsec import IpsecTcpTests
16 from vpp_ipsec import VppIpsecSA, VppIpsecSpd, VppIpsecSpdEntry,\
18 from vpp_ip_route import VppIpRoute, VppRoutePath
19 from vpp_ip import DpoProto
20 from vpp_papi import VppEnum
23 class ConfigIpsecAH(TemplateIpsec):
25 Basic test for IPSEC using AH transport and Tunnel mode
35 --- encrypt --- plain ---
36 |pg0| <------- |VPP| <------ |pg1|
39 --- decrypt --- plain ---
40 |pg0| -------> |VPP| ------> |pg1|
46 tra4_encrypt_node_name = "ah4-encrypt"
47 tra4_decrypt_node_name = ["ah4-decrypt", "ah4-decrypt"]
48 tra6_encrypt_node_name = "ah6-encrypt"
49 tra6_decrypt_node_name = ["ah6-decrypt", "ah6-decrypt"]
50 tun4_encrypt_node_name = "ah4-encrypt"
51 tun4_decrypt_node_name = ["ah4-decrypt", "ah4-decrypt"]
52 tun6_encrypt_node_name = "ah6-encrypt"
53 tun6_decrypt_node_name = ["ah6-decrypt", "ah6-decrypt"]
57 super(ConfigIpsecAH, cls).setUpClass()
60 def tearDownClass(cls):
61 super(ConfigIpsecAH, cls).tearDownClass()
64 super(ConfigIpsecAH, self).setUp()
67 super(ConfigIpsecAH, self).tearDown()
69 def config_network(self, params):
71 self.tun_if = self.pg0
72 self.tra_if = self.pg2
73 self.logger.info(self.vapi.ppcli("show int addr"))
75 self.tra_spd = VppIpsecSpd(self, self.tra_spd_id)
76 self.tra_spd.add_vpp_config()
77 self.net_objs.append(self.tra_spd)
78 self.tun_spd = VppIpsecSpd(self, self.tun_spd_id)
79 self.tun_spd.add_vpp_config()
80 self.net_objs.append(self.tun_spd)
82 b = VppIpsecSpdItfBinding(self, self.tra_spd,
85 self.net_objs.append(b)
87 b = VppIpsecSpdItfBinding(self, self.tun_spd,
90 self.net_objs.append(b)
94 config_tra_params(p, self.encryption_type)
97 config_tun_params(p, self.encryption_type, self.tun_if)
99 d = DpoProto.DPO_PROTO_IP6 if p.is_ipv6 else DpoProto.DPO_PROTO_IP4
100 r = VppIpRoute(self, p.remote_tun_if_host, p.addr_len,
101 [VppRoutePath(self.tun_if.remote_addr[p.addr_type],
105 self.net_objs.append(r)
106 self.logger.info(self.vapi.ppcli("show ipsec all"))
108 def unconfig_network(self):
109 for o in reversed(self.net_objs):
110 o.remove_vpp_config()
113 def config_ah_tun(self, params):
114 addr_type = params.addr_type
115 scapy_tun_sa_id = params.scapy_tun_sa_id
116 scapy_tun_spi = params.scapy_tun_spi
117 vpp_tun_sa_id = params.vpp_tun_sa_id
118 vpp_tun_spi = params.vpp_tun_spi
119 auth_algo_vpp_id = params.auth_algo_vpp_id
120 auth_key = params.auth_key
121 crypt_algo_vpp_id = params.crypt_algo_vpp_id
122 crypt_key = params.crypt_key
123 remote_tun_if_host = params.remote_tun_if_host
124 addr_any = params.addr_any
125 addr_bcast = params.addr_bcast
127 tun_flags = params.tun_flags
128 e = VppEnum.vl_api_ipsec_spd_action_t
130 params.outer_hop_limit = 253
131 params.outer_flow_label = 0x12345
133 params.tun_sa_in = VppIpsecSA(self, scapy_tun_sa_id, scapy_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.local_addr[addr_type],
138 self.tun_if.remote_addr[addr_type],
143 params.tun_sa_out = VppIpsecSA(self, vpp_tun_sa_id, vpp_tun_spi,
144 auth_algo_vpp_id, auth_key,
145 crypt_algo_vpp_id, crypt_key,
146 self.vpp_ah_protocol,
147 self.tun_if.remote_addr[addr_type],
148 self.tun_if.local_addr[addr_type],
153 objs.append(params.tun_sa_in)
154 objs.append(params.tun_sa_out)
156 params.spd_policy_in_any = VppIpsecSpdEntry(self, self.tun_spd,
158 addr_any, addr_bcast,
159 addr_any, addr_bcast,
161 params.spd_policy_out_any = VppIpsecSpdEntry(self, self.tun_spd,
163 addr_any, addr_bcast,
164 addr_any, addr_bcast,
168 objs.append(params.spd_policy_out_any)
169 objs.append(params.spd_policy_in_any)
171 e1 = VppIpsecSpdEntry(self, self.tun_spd, vpp_tun_sa_id,
174 self.pg1.remote_addr[addr_type],
175 self.pg1.remote_addr[addr_type],
177 policy=e.IPSEC_API_SPD_ACTION_PROTECT,
179 e2 = VppIpsecSpdEntry(self, self.tun_spd, scapy_tun_sa_id,
180 self.pg1.remote_addr[addr_type],
181 self.pg1.remote_addr[addr_type],
184 0, policy=e.IPSEC_API_SPD_ACTION_PROTECT,
186 e3 = VppIpsecSpdEntry(self, self.tun_spd, vpp_tun_sa_id,
189 self.pg0.local_addr[addr_type],
190 self.pg0.local_addr[addr_type],
192 policy=e.IPSEC_API_SPD_ACTION_PROTECT,
194 e4 = VppIpsecSpdEntry(self, self.tun_spd, scapy_tun_sa_id,
195 self.pg0.local_addr[addr_type],
196 self.pg0.local_addr[addr_type],
199 0, policy=e.IPSEC_API_SPD_ACTION_PROTECT,
202 objs = objs + [e1, e2, e3, e4]
207 self.net_objs = self.net_objs + objs
209 def config_ah_tra(self, params):
210 addr_type = params.addr_type
211 scapy_tra_sa_id = params.scapy_tra_sa_id
212 scapy_tra_spi = params.scapy_tra_spi
213 vpp_tra_sa_id = params.vpp_tra_sa_id
214 vpp_tra_spi = params.vpp_tra_spi
215 auth_algo_vpp_id = params.auth_algo_vpp_id
216 auth_key = params.auth_key
217 crypt_algo_vpp_id = params.crypt_algo_vpp_id
218 crypt_key = params.crypt_key
219 addr_any = params.addr_any
220 addr_bcast = params.addr_bcast
221 flags = params.flags | (VppEnum.vl_api_ipsec_sad_flags_t.
222 IPSEC_API_SAD_FLAG_USE_ANTI_REPLAY)
223 e = VppEnum.vl_api_ipsec_spd_action_t
226 params.tra_sa_in = VppIpsecSA(self, scapy_tra_sa_id, scapy_tra_spi,
227 auth_algo_vpp_id, auth_key,
228 crypt_algo_vpp_id, crypt_key,
229 self.vpp_ah_protocol,
231 params.tra_sa_out = VppIpsecSA(self, vpp_tra_sa_id, vpp_tra_spi,
232 auth_algo_vpp_id, auth_key,
233 crypt_algo_vpp_id, crypt_key,
234 self.vpp_ah_protocol,
237 objs.append(params.tra_sa_in)
238 objs.append(params.tra_sa_out)
240 objs.append(VppIpsecSpdEntry(self, self.tra_spd, vpp_tra_sa_id,
241 addr_any, addr_bcast,
242 addr_any, addr_bcast,
244 objs.append(VppIpsecSpdEntry(self, self.tra_spd, scapy_tra_sa_id,
245 addr_any, addr_bcast,
246 addr_any, addr_bcast,
249 objs.append(VppIpsecSpdEntry(self, self.tra_spd, vpp_tra_sa_id,
250 self.tra_if.local_addr[addr_type],
251 self.tra_if.local_addr[addr_type],
252 self.tra_if.remote_addr[addr_type],
253 self.tra_if.remote_addr[addr_type],
255 policy=e.IPSEC_API_SPD_ACTION_PROTECT,
257 objs.append(VppIpsecSpdEntry(self, self.tra_spd, scapy_tra_sa_id,
258 self.tra_if.local_addr[addr_type],
259 self.tra_if.local_addr[addr_type],
260 self.tra_if.remote_addr[addr_type],
261 self.tra_if.remote_addr[addr_type],
262 0, policy=e.IPSEC_API_SPD_ACTION_PROTECT,
267 self.net_objs = self.net_objs + objs
270 class TemplateIpsecAh(ConfigIpsecAH):
272 Basic test for IPSEC using AH transport and Tunnel mode
277 |pg2| <-------> |VPP|
282 --- encrypt --- plain ---
283 |pg0| <------- |VPP| <------ |pg1|
286 --- decrypt --- plain ---
287 |pg0| -------> |VPP| ------> |pg1|
293 super(TemplateIpsecAh, cls).setUpClass()
296 def tearDownClass(cls):
297 super(TemplateIpsecAh, cls).tearDownClass()
300 super(TemplateIpsecAh, self).setUp()
301 self.config_network(self.params.values())
304 self.unconfig_network()
305 super(TemplateIpsecAh, self).tearDown()
308 class TestIpsecAh1(TemplateIpsecAh, IpsecTcpTests):
309 """ Ipsec AH - TCP tests """
313 class TestIpsecAh2(TemplateIpsecAh, IpsecTra46Tests, IpsecTun46Tests):
314 """ Ipsec AH w/ SHA1 """
318 class TestIpsecAhTun(TemplateIpsecAh, IpsecTun46Tests):
319 """ Ipsec AH - TUN encap tests """
322 self.ipv4_params = IPsecIPv4Params()
323 self.ipv6_params = IPsecIPv6Params()
325 c = (VppEnum.vl_api_tunnel_encap_decap_flags_t.
326 TUNNEL_API_ENCAP_DECAP_FLAG_ENCAP_COPY_DSCP)
327 c1 = c | (VppEnum.vl_api_tunnel_encap_decap_flags_t.
328 TUNNEL_API_ENCAP_DECAP_FLAG_ENCAP_COPY_ECN)
330 self.ipv4_params.tun_flags = c
331 self.ipv6_params.tun_flags = c1
333 super(TestIpsecAhTun, self).setUp()
335 def gen_pkts(self, sw_intf, src, dst, count=1, payload_size=54):
336 # set the DSCP + ECN - flags are set to copy only DSCP
337 return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
338 IP(src=src, dst=dst, tos=5) /
339 UDP(sport=4444, dport=4444) /
340 Raw(b'X' * payload_size)
341 for i in range(count)]
343 def gen_pkts6(self, p, sw_intf, src, dst, count=1, payload_size=54):
344 # set the DSCP + ECN - flags are set to copy both
345 return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
346 IPv6(src=src, dst=dst, tc=5) /
347 UDP(sport=4444, dport=4444) /
348 Raw(b'X' * payload_size)
349 for i in range(count)]
351 def verify_encrypted(self, p, sa, rxs):
352 # just check that only the DSCP is copied
354 self.assertEqual(rx[IP].tos, 4)
356 def verify_encrypted6(self, p, sa, rxs):
357 # just check that the DSCP & ECN are copied
359 self.assertEqual(rx[IPv6].tc, 5)
362 class TestIpsecAhTun2(TemplateIpsecAh, IpsecTun46Tests):
363 """ Ipsec AH - TUN encap tests """
366 self.ipv4_params = IPsecIPv4Params()
367 self.ipv6_params = IPsecIPv6Params()
369 self.ipv4_params.dscp = 3
370 self.ipv6_params.dscp = 4
372 super(TestIpsecAhTun2, self).setUp()
374 def gen_pkts(self, sw_intf, src, dst, count=1, payload_size=54):
375 # set the DSCP + ECN - flags are set to copy only DSCP
376 return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
377 IP(src=src, dst=dst, tos=0) /
378 UDP(sport=4444, dport=4444) /
379 Raw(b'X' * payload_size)
380 for i in range(count)]
382 def gen_pkts6(self, p, sw_intf, src, dst, count=1, payload_size=54):
383 # set the DSCP + ECN - flags are set to copy both
384 return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
385 IPv6(src=src, dst=dst, tc=0) /
386 UDP(sport=4444, dport=4444) /
387 Raw(b'X' * payload_size)
388 for i in range(count)]
390 def verify_encrypted(self, p, sa, rxs):
391 # just check that only the DSCP is copied
393 self.assertEqual(rx[IP].tos, 0xc)
395 def verify_encrypted6(self, p, sa, rxs):
396 # just check that the DSCP & ECN are copied
398 self.assertEqual(rx[IPv6].tc, 0x10)
401 class TestIpsecAhHandoff(TemplateIpsecAh,
402 IpsecTun6HandoffTests,
403 IpsecTun4HandoffTests):
404 """ Ipsec AH Handoff """
408 class TestIpsecAhAll(ConfigIpsecAH,
409 IpsecTra4, IpsecTra6,
410 IpsecTun4, IpsecTun6):
411 """ Ipsec AH all Algos """
414 super(TestIpsecAhAll, self).setUp()
417 super(TestIpsecAhAll, self).tearDown()
419 def test_integ_algs(self):
420 """All Engines SHA[1_96, 256, 384, 512] w/ & w/o ESN"""
421 # foreach VPP crypto engine
422 engines = ["ia32", "ipsecmb", "openssl"]
424 algos = [{'vpp': VppEnum.vl_api_ipsec_integ_alg_t.
425 IPSEC_API_INTEG_ALG_SHA1_96,
426 'scapy': "HMAC-SHA1-96"},
427 {'vpp': VppEnum.vl_api_ipsec_integ_alg_t.
428 IPSEC_API_INTEG_ALG_SHA_256_128,
429 'scapy': "SHA2-256-128"},
430 {'vpp': VppEnum.vl_api_ipsec_integ_alg_t.
431 IPSEC_API_INTEG_ALG_SHA_384_192,
432 'scapy': "SHA2-384-192"},
433 {'vpp': VppEnum.vl_api_ipsec_integ_alg_t.
434 IPSEC_API_INTEG_ALG_SHA_512_256,
435 'scapy': "SHA2-512-256"}]
437 flags = [0, (VppEnum.vl_api_ipsec_sad_flags_t.
438 IPSEC_API_SAD_FLAG_USE_ESN)]
441 # loop through the VPP engines
443 for engine in engines:
444 self.vapi.cli("set crypto handler all %s" % engine)
446 # loop through each of the algorithms
449 # with self.subTest(algo=algo['scapy']):
452 # setup up the config paramters
454 self.ipv4_params = IPsecIPv4Params()
455 self.ipv6_params = IPsecIPv6Params()
457 self.params = {self.ipv4_params.addr_type:
459 self.ipv6_params.addr_type:
462 for _, p in self.params.items():
463 p.auth_algo_vpp_id = algo['vpp']
464 p.auth_algo = algo['scapy']
465 p.flags = p.flags | flag
468 # configure the SPDs. SAs, etc
470 self.config_network(self.params.values())
474 # An exhautsive 4o6, 6o4 is not necessary for each algo
476 self.verify_tra_basic6(count=17)
477 self.verify_tra_basic4(count=17)
478 self.verify_tun_66(self.params[socket.AF_INET6], count=17)
479 self.verify_tun_44(self.params[socket.AF_INET], count=17)
482 # remove the SPDs, SAs, etc
484 self.unconfig_network()
487 if __name__ == '__main__':
488 unittest.main(testRunner=VppTestRunner)