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,
57 self.test.registry.register(self, self.test.logger)
59 def remove_vpp_config(self):
60 self.test.vapi.ipsec_interface_add_del_spd(self.spd.id,
65 return "bind-%s-to-%s" % (self.spd.id, self.itf)
67 def query_vpp_config(self):
68 bs = self.test.vapi.ipsec_spd_interface_dump()
70 if b.sw_if_index == self.itf.sw_if_index:
75 class VppIpsecSpdEntry(VppObject):
80 def __init__(self, test, spd, sa_id,
81 local_start, local_stop,
82 remote_start, remote_stop,
88 remote_port_stop=65535,
90 local_port_stop=65535):
94 self.local_start = ip_address(text_type(local_start))
95 self.local_stop = ip_address(text_type(local_stop))
96 self.remote_start = ip_address(text_type(remote_start))
97 self.remote_stop = ip_address(text_type(remote_stop))
99 self.is_outbound = is_outbound
100 self.priority = priority
102 self.policy = (VppEnum.vl_api_ipsec_spd_action_t.
103 IPSEC_API_SPD_ACTION_BYPASS)
106 self.is_ipv6 = (0 if self.local_start.version == 4 else 1)
107 self.local_port_start = local_port_start
108 self.local_port_stop = local_port_stop
109 self.remote_port_start = remote_port_start
110 self.remote_port_stop = remote_port_stop
112 def add_vpp_config(self):
113 rv = self.test.vapi.ipsec_spd_entry_add_del(
121 is_ipv6=self.is_ipv6,
122 is_outbound=self.is_outbound,
123 priority=self.priority,
125 local_port_start=self.local_port_start,
126 local_port_stop=self.local_port_stop,
127 remote_port_start=self.remote_port_start,
128 remote_port_stop=self.remote_port_stop)
129 self.stat_index = rv.stat_index
130 self.test.registry.register(self, self.test.logger)
132 def remove_vpp_config(self):
133 self.test.vapi.ipsec_spd_entry_add_del(
141 is_ipv6=self.is_ipv6,
142 is_outbound=self.is_outbound,
143 priority=self.priority,
145 local_port_start=self.local_port_start,
146 local_port_stop=self.local_port_stop,
147 remote_port_start=self.remote_port_start,
148 remote_port_stop=self.remote_port_stop,
152 return "spd-entry-%d-%d-%d-%d-%d-%d" % (self.spd.id,
157 self.remote_port_start)
159 def query_vpp_config(self):
160 ss = self.test.vapi.ipsec_spd_dump(self.spd.id)
162 if s.entry.sa_id == self.sa_id and \
163 s.entry.is_outbound == self.is_outbound and \
164 s.entry.priority == self.priority and \
165 s.entry.policy == self.policy and \
166 s.entry.remote_address_start == self.remote_start and \
167 s.entry.remote_port_start == self.remote_port_start:
171 def get_stats(self, worker=None):
172 c = self.test.statistics.get_counter("/net/ipsec/policy")
176 total['packets'] += t[self.stat_index]['packets']
179 # +1 to skip main thread
180 return c[worker+1][self.stat_index]
183 class VppIpsecSA(VppObject):
188 DEFAULT_UDP_PORT = 4500
190 def __init__(self, test, id, spi,
191 integ_alg, integ_key,
192 crypto_alg, crypto_key,
194 tun_src=None, tun_dst=None,
195 flags=None, salt=0, tun_flags=None,
197 udp_src=None, udp_dst=None, hop_limit=None):
198 e = VppEnum.vl_api_ipsec_sad_flags_t
202 self.integ_alg = integ_alg
203 self.integ_key = integ_key
204 self.crypto_alg = crypto_alg
205 self.crypto_key = crypto_key
210 self.tun_src = tun_src
211 self.tun_dst = tun_dst
213 self.flags = e.IPSEC_API_SAD_FLAG_NONE
217 self.tun_src = ip_address(text_type(tun_src))
218 self.flags = self.flags | e.IPSEC_API_SAD_FLAG_IS_TUNNEL
219 if (self.tun_src.version == 6):
220 self.flags = self.flags | e.IPSEC_API_SAD_FLAG_IS_TUNNEL_V6
222 self.tun_dst = ip_address(text_type(tun_dst))
223 self.udp_src = udp_src
224 self.udp_dst = udp_dst
225 self.tun_flags = (VppEnum.vl_api_tunnel_encap_decap_flags_t.
226 TUNNEL_API_ENCAP_DECAP_FLAG_NONE)
228 self.tun_flags = tun_flags
229 self.dscp = VppEnum.vl_api_ip_dscp_t.IP_API_DSCP_CS0
234 self.hop_limit = hop_limit
236 def tunnel_encode(self):
237 return {'src': (self.tun_src if self.tun_src else []),
238 'dst': (self.tun_dst if self.tun_dst else []),
239 'encap_decap_flags': self.tun_flags,
241 'hop_limit': self.hop_limit,
242 'table_id': self.table_id
245 def add_vpp_config(self):
249 'integrity_algorithm': self.integ_alg,
251 'length': len(self.integ_key),
252 'data': self.integ_key,
254 'crypto_algorithm': self.crypto_alg,
256 'data': self.crypto_key,
257 'length': len(self.crypto_key),
259 'protocol': self.proto,
260 'tunnel': self.tunnel_encode(),
264 # don't explicitly send the defaults, let papi fill them in
266 entry['udp_src_port'] = self.udp_src
268 entry['udp_dst_port'] = self.udp_dst
269 r = self.test.vapi.ipsec_sad_entry_add_del_v3(is_add=1, entry=entry)
270 self.stat_index = r.stat_index
271 self.test.registry.register(self, self.test.logger)
274 def remove_vpp_config(self):
275 r = self.test.vapi.ipsec_sad_entry_add_del_v3(
280 'integrity_algorithm': self.integ_alg,
282 'length': len(self.integ_key),
283 'data': self.integ_key,
285 'crypto_algorithm': self.crypto_alg,
287 'data': self.crypto_key,
288 'length': len(self.crypto_key),
290 'protocol': self.proto,
291 'tunnel': self.tunnel_encode(),
296 return "ipsec-sa-%d" % self.id
298 def query_vpp_config(self):
299 e = VppEnum.vl_api_ipsec_sad_flags_t
301 bs = self.test.vapi.ipsec_sa_v3_dump()
303 if b.entry.sad_id == self.id:
304 # if udp encap is configured then the ports should match
305 # those configured or the default
306 if (self.flags & e.IPSEC_API_SAD_FLAG_UDP_ENCAP):
307 if not b.entry.flags & e.IPSEC_API_SAD_FLAG_UDP_ENCAP:
310 if self.udp_src != b.entry.udp_src_port:
313 if self.DEFAULT_UDP_PORT != b.entry.udp_src_port:
316 if self.udp_dst != b.entry.udp_dst_port:
319 if self.DEFAULT_UDP_PORT != b.entry.udp_dst_port:
324 def get_stats(self, worker=None):
325 c = self.test.statistics.get_counter("/net/ipsec/sa")
329 total['packets'] += t[self.stat_index]['packets']
332 # +1 to skip main thread
333 return c[worker+1][self.stat_index]
336 class VppIpsecTunProtect(VppObject):
338 VPP IPSEC tunnel protection
341 def __init__(self, test, itf, sa_out, sas_in, nh=None):
346 self.sas_in.append(sa.id)
347 self.sa_out = sa_out.id
352 def update_vpp_config(self, sa_out, sas_in):
355 self.sas_in.append(sa.id)
356 self.sa_out = sa_out.id
357 self.test.vapi.ipsec_tunnel_protect_update(
359 'sw_if_index': self.itf._sw_if_index,
360 'n_sa_in': len(self.sas_in),
361 'sa_out': self.sa_out,
362 'sa_in': self.sas_in,
366 return "ipsec-tun-protect-%s-%s" % (self.itf, self.nh)
368 def add_vpp_config(self):
369 self.test.vapi.ipsec_tunnel_protect_update(
371 'sw_if_index': self.itf._sw_if_index,
372 'n_sa_in': len(self.sas_in),
373 'sa_out': self.sa_out,
374 'sa_in': self.sas_in,
376 self.test.registry.register(self, self.test.logger)
378 def remove_vpp_config(self):
379 self.test.vapi.ipsec_tunnel_protect_del(
380 sw_if_index=self.itf.sw_if_index,
383 def query_vpp_config(self):
384 bs = self.test.vapi.ipsec_tunnel_protect_dump(
385 sw_if_index=self.itf.sw_if_index)
387 if b.tun.sw_if_index == self.itf.sw_if_index and \
388 self.nh == str(b.tun.nh):
393 class VppIpsecInterface(VppInterface):
398 def __init__(self, test, mode=None, instance=0xffffffff):
399 super(VppIpsecInterface, self).__init__(test)
403 self.mode = (VppEnum.vl_api_tunnel_mode_t.
405 self.instance = instance
407 def add_vpp_config(self):
408 r = self.test.vapi.ipsec_itf_create(itf={
409 'user_instance': self.instance,
412 self.set_sw_if_index(r.sw_if_index)
413 self.test.registry.register(self, self.test.logger)
416 def remove_vpp_config(self):
417 self.test.vapi.ipsec_itf_delete(sw_if_index=self._sw_if_index)
419 def query_vpp_config(self):
420 ts = self.test.vapi.ipsec_itf_dump(sw_if_index=0xffffffff)
422 if t.tunnel.sw_if_index == self._sw_if_index:
427 return self.object_id()
430 return "ipsec-%d" % self._sw_if_index