1 from vpp_object import VppObject
2 from ipaddress import ip_address
3 from vpp_papi import VppEnum
4 from vpp_interface import VppInterface
13 return {"packets": 0, "bytes": 0}
16 class VppIpsecSpd(VppObject):
21 def __init__(self, test, id):
25 def add_vpp_config(self):
26 self.test.vapi.ipsec_spd_add_del(self.id)
27 self.test.registry.register(self, self.test.logger)
29 def remove_vpp_config(self):
30 self.test.vapi.ipsec_spd_add_del(self.id, is_add=0)
33 return "ipsec-spd-%d" % self.id
35 def query_vpp_config(self):
36 spds = self.test.vapi.ipsec_spds_dump()
38 if spd.spd_id == self.id:
43 class VppIpsecSpdItfBinding(VppObject):
45 VPP SPD DB to interface binding
46 (i.e. this SPD is used on this interface)
49 def __init__(self, test, spd, itf):
54 def add_vpp_config(self):
55 self.test.vapi.ipsec_interface_add_del_spd(self.spd.id, self.itf.sw_if_index)
56 self.test.registry.register(self, self.test.logger)
58 def remove_vpp_config(self):
59 self.test.vapi.ipsec_interface_add_del_spd(
60 self.spd.id, self.itf.sw_if_index, is_add=0
64 return "bind-%s-to-%s" % (self.spd.id, self.itf)
66 def query_vpp_config(self):
67 bs = self.test.vapi.ipsec_spd_interface_dump()
69 if b.sw_if_index == self.itf.sw_if_index:
74 class VppIpsecSpdEntry(VppObject):
88 proto=socket.IPPROTO_RAW,
93 remote_port_stop=65535,
95 local_port_stop=65535,
100 self.local_start = ip_address(text_type(local_start))
101 self.local_stop = ip_address(text_type(local_stop))
102 self.remote_start = ip_address(text_type(remote_start))
103 self.remote_stop = ip_address(text_type(remote_stop))
105 self.is_outbound = is_outbound
106 self.priority = priority
108 self.policy = VppEnum.vl_api_ipsec_spd_action_t.IPSEC_API_SPD_ACTION_BYPASS
111 self.is_ipv6 = 0 if self.local_start.version == 4 else 1
112 self.local_port_start = local_port_start
113 self.local_port_stop = local_port_stop
114 self.remote_port_start = remote_port_start
115 self.remote_port_stop = remote_port_stop
117 def add_vpp_config(self):
118 rv = self.test.vapi.ipsec_spd_entry_add_del(
126 is_ipv6=self.is_ipv6,
127 is_outbound=self.is_outbound,
128 priority=self.priority,
130 local_port_start=self.local_port_start,
131 local_port_stop=self.local_port_stop,
132 remote_port_start=self.remote_port_start,
133 remote_port_stop=self.remote_port_stop,
135 self.stat_index = rv.stat_index
136 self.test.registry.register(self, self.test.logger)
139 def remove_vpp_config(self):
140 self.test.vapi.ipsec_spd_entry_add_del(
148 is_ipv6=self.is_ipv6,
149 is_outbound=self.is_outbound,
150 priority=self.priority,
152 local_port_start=self.local_port_start,
153 local_port_stop=self.local_port_stop,
154 remote_port_start=self.remote_port_start,
155 remote_port_stop=self.remote_port_stop,
160 return "spd-entry-%d-%d-%d-%d-%d-%d" % (
166 self.remote_port_start,
169 def query_vpp_config(self):
170 ss = self.test.vapi.ipsec_spd_dump(self.spd.id)
173 s.entry.sa_id == self.sa_id
174 and s.entry.is_outbound == self.is_outbound
175 and s.entry.priority == self.priority
176 and s.entry.policy == self.policy
177 and s.entry.remote_address_start == self.remote_start
178 and s.entry.remote_port_start == self.remote_port_start
183 def get_stats(self, worker=None):
184 c = self.test.statistics.get_counter("/net/ipsec/policy")
188 total["packets"] += t[self.stat_index]["packets"]
191 # +1 to skip main thread
192 return c[worker + 1][self.stat_index]
195 class VppIpsecSA(VppObject):
200 DEFAULT_UDP_PORT = 4500
222 e = VppEnum.vl_api_ipsec_sad_flags_t
226 self.integ_alg = integ_alg
227 self.integ_key = integ_key
228 self.crypto_alg = crypto_alg
229 self.crypto_key = crypto_key
234 self.tun_src = tun_src
235 self.tun_dst = tun_dst
237 self.flags = e.IPSEC_API_SAD_FLAG_NONE
241 self.tun_src = ip_address(text_type(tun_src))
242 self.flags = self.flags | e.IPSEC_API_SAD_FLAG_IS_TUNNEL
244 self.tun_dst = ip_address(text_type(tun_dst))
245 self.udp_src = udp_src
246 self.udp_dst = udp_dst
248 VppEnum.vl_api_tunnel_encap_decap_flags_t.TUNNEL_API_ENCAP_DECAP_FLAG_NONE
251 self.tun_flags = tun_flags
252 self.dscp = VppEnum.vl_api_ip_dscp_t.IP_API_DSCP_CS0
257 self.hop_limit = hop_limit
259 def tunnel_encode(self):
261 "src": (self.tun_src if self.tun_src else []),
262 "dst": (self.tun_dst if self.tun_dst else []),
263 "encap_decap_flags": self.tun_flags,
265 "hop_limit": self.hop_limit,
266 "table_id": self.table_id,
269 def add_vpp_config(self):
273 "integrity_algorithm": self.integ_alg,
275 "length": len(self.integ_key),
276 "data": self.integ_key,
278 "crypto_algorithm": self.crypto_alg,
280 "data": self.crypto_key,
281 "length": len(self.crypto_key),
283 "protocol": self.proto,
284 "tunnel": self.tunnel_encode(),
288 # don't explicitly send the defaults, let papi fill them in
290 entry["udp_src_port"] = self.udp_src
292 entry["udp_dst_port"] = self.udp_dst
293 r = self.test.vapi.ipsec_sad_entry_add(entry=entry)
294 self.stat_index = r.stat_index
295 self.test.registry.register(self, self.test.logger)
298 def update_vpp_config(
299 self, udp_src=None, udp_dst=None, is_tun=False, tun_src=None, tun_dst=None
303 self.tun_src = ip_address(text_type(tun_src))
305 self.tun_dst = ip_address(text_type(tun_dst))
307 self.udp_src = udp_src
309 self.udp_dst = udp_dst
310 self.test.vapi.ipsec_sad_entry_update(
313 tunnel=self.tunnel_encode(),
314 udp_src_port=udp_src,
315 udp_dst_port=udp_dst,
318 def remove_vpp_config(self):
319 self.test.vapi.ipsec_sad_entry_del(id=self.id)
322 return "ipsec-sa-%d" % self.id
324 def query_vpp_config(self):
325 e = VppEnum.vl_api_ipsec_sad_flags_t
327 bs = self.test.vapi.ipsec_sa_v3_dump()
329 if b.entry.sad_id == self.id:
330 # if udp encap is configured then the ports should match
331 # those configured or the default
332 if self.flags & e.IPSEC_API_SAD_FLAG_UDP_ENCAP:
333 if not b.entry.flags & e.IPSEC_API_SAD_FLAG_UDP_ENCAP:
336 if self.udp_src != b.entry.udp_src_port:
339 if self.DEFAULT_UDP_PORT != b.entry.udp_src_port:
342 if self.udp_dst != b.entry.udp_dst_port:
345 if self.DEFAULT_UDP_PORT != b.entry.udp_dst_port:
350 def get_stats(self, worker=None):
351 c = self.test.statistics.get_counter("/net/ipsec/sa")
355 total["packets"] += t[self.stat_index]["packets"]
358 # +1 to skip main thread
359 return c[worker + 1][self.stat_index]
361 def get_err(self, name, worker=None):
362 c = self.test.statistics.get_counter("/net/ipsec/sa/err/" + name)
366 total += t[self.stat_index]
369 # +1 to skip main thread
370 return c[worker + 1][self.stat_index]
373 class VppIpsecTunProtect(VppObject):
375 VPP IPSEC tunnel protection
378 def __init__(self, test, itf, sa_out, sas_in, nh=None):
383 self.sas_in.append(sa.id)
384 self.sa_out = sa_out.id
389 def update_vpp_config(self, sa_out, sas_in):
392 self.sas_in.append(sa.id)
393 self.sa_out = sa_out.id
394 self.test.vapi.ipsec_tunnel_protect_update(
396 "sw_if_index": self.itf._sw_if_index,
397 "n_sa_in": len(self.sas_in),
398 "sa_out": self.sa_out,
399 "sa_in": self.sas_in,
405 return "ipsec-tun-protect-%s-%s" % (self.itf, self.nh)
407 def add_vpp_config(self):
408 self.test.vapi.ipsec_tunnel_protect_update(
410 "sw_if_index": self.itf._sw_if_index,
411 "n_sa_in": len(self.sas_in),
412 "sa_out": self.sa_out,
413 "sa_in": self.sas_in,
417 self.test.registry.register(self, self.test.logger)
419 def remove_vpp_config(self):
420 self.test.vapi.ipsec_tunnel_protect_del(
421 sw_if_index=self.itf.sw_if_index, nh=self.nh
424 def query_vpp_config(self):
425 bs = self.test.vapi.ipsec_tunnel_protect_dump(sw_if_index=self.itf.sw_if_index)
427 if b.tun.sw_if_index == self.itf.sw_if_index and self.nh == str(b.tun.nh):
432 class VppIpsecInterface(VppInterface):
437 def __init__(self, test, mode=None, instance=0xFFFFFFFF):
438 super(VppIpsecInterface, self).__init__(test)
442 self.mode = VppEnum.vl_api_tunnel_mode_t.TUNNEL_API_MODE_P2P
443 self.instance = instance
445 def add_vpp_config(self):
446 r = self.test.vapi.ipsec_itf_create(
448 "user_instance": self.instance,
452 self.set_sw_if_index(r.sw_if_index)
453 self.test.registry.register(self, self.test.logger)
454 ts = self.test.vapi.ipsec_itf_dump(sw_if_index=self._sw_if_index)
455 self.instance = ts[0].itf.user_instance
458 def remove_vpp_config(self):
459 self.test.vapi.ipsec_itf_delete(sw_if_index=self._sw_if_index)
461 def query_vpp_config(self):
462 ts = self.test.vapi.ipsec_itf_dump(sw_if_index=0xFFFFFFFF)
464 if t.itf.sw_if_index == self._sw_if_index:
469 return self.object_id()
472 return "ipsec%d" % self.instance