3 from vpp_object import VppObject
4 from ipaddress import ip_address
5 from vpp_papi import VppEnum
6 from vpp_interface import VppInterface
15 return {"packets": 0, "bytes": 0}
18 class VppIpsecSpd(VppObject):
23 def __init__(self, test, id):
27 def add_vpp_config(self):
28 self.test.vapi.ipsec_spd_add_del(self.id)
29 self.test.registry.register(self, self.test.logger)
31 def remove_vpp_config(self):
32 self.test.vapi.ipsec_spd_add_del(self.id, is_add=0)
35 return "ipsec-spd-%d" % self.id
37 def query_vpp_config(self):
38 spds = self.test.vapi.ipsec_spds_dump()
40 if spd.spd_id == self.id:
45 class VppIpsecSpdItfBinding(VppObject):
47 VPP SPD DB to interface binding
48 (i.e. this SPD is used on this interface)
51 def __init__(self, test, spd, itf):
56 def add_vpp_config(self):
57 self.test.vapi.ipsec_interface_add_del_spd(self.spd.id, self.itf.sw_if_index)
58 self.test.registry.register(self, self.test.logger)
60 def remove_vpp_config(self):
61 self.test.vapi.ipsec_interface_add_del_spd(
62 self.spd.id, self.itf.sw_if_index, is_add=0
66 return "bind-%s-to-%s" % (self.spd.id, self.itf)
68 def query_vpp_config(self):
69 bs = self.test.vapi.ipsec_spd_interface_dump()
71 if b.sw_if_index == self.itf.sw_if_index:
76 class VppIpsecSpdEntry(VppObject):
90 proto=socket.IPPROTO_RAW,
95 remote_port_stop=65535,
97 local_port_stop=65535,
102 self.local_start = ip_address(text_type(local_start))
103 self.local_stop = ip_address(text_type(local_stop))
104 self.remote_start = ip_address(text_type(remote_start))
105 self.remote_stop = ip_address(text_type(remote_stop))
107 self.is_outbound = is_outbound
108 self.priority = priority
110 self.policy = VppEnum.vl_api_ipsec_spd_action_t.IPSEC_API_SPD_ACTION_BYPASS
113 self.is_ipv6 = 0 if self.local_start.version == 4 else 1
114 self.local_port_start = local_port_start
115 self.local_port_stop = local_port_stop
116 self.remote_port_start = remote_port_start
117 self.remote_port_stop = remote_port_stop
119 def add_vpp_config(self):
120 rv = self.test.vapi.ipsec_spd_entry_add_del(
128 is_ipv6=self.is_ipv6,
129 is_outbound=self.is_outbound,
130 priority=self.priority,
132 local_port_start=self.local_port_start,
133 local_port_stop=self.local_port_stop,
134 remote_port_start=self.remote_port_start,
135 remote_port_stop=self.remote_port_stop,
137 self.stat_index = rv.stat_index
138 self.test.registry.register(self, self.test.logger)
141 def remove_vpp_config(self):
142 self.test.vapi.ipsec_spd_entry_add_del(
150 is_ipv6=self.is_ipv6,
151 is_outbound=self.is_outbound,
152 priority=self.priority,
154 local_port_start=self.local_port_start,
155 local_port_stop=self.local_port_stop,
156 remote_port_start=self.remote_port_start,
157 remote_port_stop=self.remote_port_stop,
162 return "spd-entry-%d-%d-%d-%d-%d-%d" % (
168 self.remote_port_start,
171 def query_vpp_config(self):
172 ss = self.test.vapi.ipsec_spd_dump(self.spd.id)
175 s.entry.sa_id == self.sa_id
176 and s.entry.is_outbound == self.is_outbound
177 and s.entry.priority == self.priority
178 and s.entry.policy == self.policy
179 and s.entry.remote_address_start == self.remote_start
180 and s.entry.remote_port_start == self.remote_port_start
185 def get_stats(self, worker=None):
186 c = self.test.statistics.get_counter("/net/ipsec/policy")
190 total["packets"] += t[self.stat_index]["packets"]
193 # +1 to skip main thread
194 return c[worker + 1][self.stat_index]
197 class VppIpsecSA(VppObject):
202 DEFAULT_UDP_PORT = 4500
223 anti_replay_window_size=0,
225 e = VppEnum.vl_api_ipsec_sad_flags_t
229 self.integ_alg = integ_alg
230 self.integ_key = integ_key
231 self.crypto_alg = crypto_alg
232 self.crypto_key = crypto_key
235 self.anti_replay_window_size = anti_replay_window_size
238 self.tun_src = tun_src
239 self.tun_dst = tun_dst
241 self.flags = e.IPSEC_API_SAD_FLAG_NONE
245 self.tun_src = ip_address(text_type(tun_src))
246 self.flags = self.flags | e.IPSEC_API_SAD_FLAG_IS_TUNNEL
248 self.tun_dst = ip_address(text_type(tun_dst))
249 self.udp_src = udp_src
250 self.udp_dst = udp_dst
252 VppEnum.vl_api_tunnel_encap_decap_flags_t.TUNNEL_API_ENCAP_DECAP_FLAG_NONE
255 self.tun_flags = tun_flags
256 self.dscp = VppEnum.vl_api_ip_dscp_t.IP_API_DSCP_CS0
261 self.hop_limit = hop_limit
263 def tunnel_encode(self):
265 "src": (self.tun_src if self.tun_src else []),
266 "dst": (self.tun_dst if self.tun_dst else []),
267 "encap_decap_flags": self.tun_flags,
269 "hop_limit": self.hop_limit,
270 "table_id": self.table_id,
273 def add_vpp_config(self):
277 "integrity_algorithm": self.integ_alg,
279 "length": len(self.integ_key),
280 "data": self.integ_key,
282 "crypto_algorithm": self.crypto_alg,
284 "data": self.crypto_key,
285 "length": len(self.crypto_key),
287 "protocol": self.proto,
288 "tunnel": self.tunnel_encode(),
291 "anti_replay_window_size": self.anti_replay_window_size,
293 # don't explicitly send the defaults, let papi fill them in
295 entry["udp_src_port"] = self.udp_src
297 entry["udp_dst_port"] = self.udp_dst
298 r = self.test.vapi.ipsec_sad_entry_add_v2(entry=entry)
299 self.stat_index = r.stat_index
300 self.test.registry.register(self, self.test.logger)
303 def update_vpp_config(
304 self, udp_src=None, udp_dst=None, is_tun=False, tun_src=None, tun_dst=None
308 self.tun_src = ip_address(text_type(tun_src))
310 self.tun_dst = ip_address(text_type(tun_dst))
312 self.udp_src = udp_src
314 self.udp_dst = udp_dst
315 self.test.vapi.ipsec_sad_entry_update(
318 tunnel=self.tunnel_encode(),
319 udp_src_port=udp_src,
320 udp_dst_port=udp_dst,
323 def remove_vpp_config(self):
324 self.test.vapi.ipsec_sad_entry_del(id=self.id)
327 return "ipsec-sa-%d" % self.id
329 def query_vpp_config(self):
330 e = VppEnum.vl_api_ipsec_sad_flags_t
332 bs = self.test.vapi.ipsec_sa_v5_dump()
334 if b.entry.sad_id == self.id:
335 # if udp encap is configured then the ports should match
336 # those configured or the default
337 if self.flags & e.IPSEC_API_SAD_FLAG_UDP_ENCAP:
338 if not b.entry.flags & e.IPSEC_API_SAD_FLAG_UDP_ENCAP:
341 if self.udp_src != b.entry.udp_src_port:
344 if self.DEFAULT_UDP_PORT != b.entry.udp_src_port:
347 if self.udp_dst != b.entry.udp_dst_port:
350 if self.DEFAULT_UDP_PORT != b.entry.udp_dst_port:
355 def get_stats(self, worker=None):
356 c = self.test.statistics.get_counter("/net/ipsec/sa")
360 total["packets"] += t[self.stat_index]["packets"]
363 # +1 to skip main thread
364 return c[worker + 1][self.stat_index]
366 def get_err(self, name, worker=None):
367 c = self.test.statistics.get_counter("/net/ipsec/sa/err/" + name)
371 total += t[self.stat_index]
374 # +1 to skip main thread
375 return c[worker + 1][self.stat_index]
378 class VppIpsecTunProtect(VppObject):
380 VPP IPSEC tunnel protection
383 def __init__(self, test, itf, sa_out, sas_in, nh=None):
388 self.sas_in.append(sa.id)
389 self.sa_out = sa_out.id
394 def update_vpp_config(self, sa_out, sas_in):
397 self.sas_in.append(sa.id)
398 self.sa_out = sa_out.id
399 self.test.vapi.ipsec_tunnel_protect_update(
401 "sw_if_index": self.itf._sw_if_index,
402 "n_sa_in": len(self.sas_in),
403 "sa_out": self.sa_out,
404 "sa_in": self.sas_in,
410 return "ipsec-tun-protect-%s-%s" % (self.itf, self.nh)
412 def add_vpp_config(self):
413 self.test.vapi.ipsec_tunnel_protect_update(
415 "sw_if_index": self.itf._sw_if_index,
416 "n_sa_in": len(self.sas_in),
417 "sa_out": self.sa_out,
418 "sa_in": self.sas_in,
422 self.test.registry.register(self, self.test.logger)
424 def remove_vpp_config(self):
425 self.test.vapi.ipsec_tunnel_protect_del(
426 sw_if_index=self.itf.sw_if_index, nh=self.nh
429 def query_vpp_config(self):
430 bs = self.test.vapi.ipsec_tunnel_protect_dump(sw_if_index=self.itf.sw_if_index)
432 if b.tun.sw_if_index == self.itf.sw_if_index and self.nh == str(b.tun.nh):
437 class VppIpsecInterface(VppInterface):
442 def __init__(self, test, mode=None, instance=0xFFFFFFFF):
443 super(VppIpsecInterface, self).__init__(test)
447 self.mode = VppEnum.vl_api_tunnel_mode_t.TUNNEL_API_MODE_P2P
448 self.instance = instance
450 def add_vpp_config(self):
451 r = self.test.vapi.ipsec_itf_create(
453 "user_instance": self.instance,
457 self.set_sw_if_index(r.sw_if_index)
458 self.test.registry.register(self, self.test.logger)
459 ts = self.test.vapi.ipsec_itf_dump(sw_if_index=self._sw_if_index)
460 self.instance = ts[0].itf.user_instance
463 def remove_vpp_config(self):
464 self.test.vapi.ipsec_itf_delete(sw_if_index=self._sw_if_index)
466 def query_vpp_config(self):
467 ts = self.test.vapi.ipsec_itf_dump(sw_if_index=0xFFFFFFFF)
469 if t.itf.sw_if_index == self._sw_if_index:
474 return self.object_id()
477 return "ipsec%d" % self.instance