+ # if udp encap is configured then the ports should match
+ # those configured or the default
+ if (self.flags & e.IPSEC_API_SAD_FLAG_UDP_ENCAP):
+ if not b.entry.flags & e.IPSEC_API_SAD_FLAG_UDP_ENCAP:
+ return False
+ if self.udp_src:
+ if self.udp_src != b.entry.udp_src_port:
+ return False
+ else:
+ if self.DEFAULT_UDP_PORT != b.entry.udp_src_port:
+ return False
+ if self.udp_dst:
+ if self.udp_dst != b.entry.udp_dst_port:
+ return False
+ else:
+ if self.DEFAULT_UDP_PORT != b.entry.udp_dst_port:
+ return False
+ return True
+ return False
+
+ def get_stats(self, worker=None):
+ c = self.test.statistics.get_counter("/net/ipsec/sa")
+ if worker is None:
+ total = mk_counter()
+ for t in c:
+ total['packets'] += t[self.stat_index]['packets']
+ return total
+ else:
+ # +1 to skip main thread
+ return c[worker+1][self.stat_index]
+
+
+class VppIpsecTunProtect(VppObject):
+ """
+ VPP IPSEC tunnel protection
+ """
+
+ def __init__(self, test, itf, sa_out, sas_in, nh=None):
+ self.test = test
+ self.itf = itf
+ self.sas_in = []
+ for sa in sas_in:
+ self.sas_in.append(sa.id)
+ self.sa_out = sa_out.id
+ self.nh = nh
+ if not self.nh:
+ self.nh = "0.0.0.0"
+
+ def update_vpp_config(self, sa_out, sas_in):
+ self.sas_in = []
+ for sa in sas_in:
+ self.sas_in.append(sa.id)
+ self.sa_out = sa_out.id
+ self.test.vapi.ipsec_tunnel_protect_update(
+ tunnel={
+ 'sw_if_index': self.itf._sw_if_index,
+ 'n_sa_in': len(self.sas_in),
+ 'sa_out': self.sa_out,
+ 'sa_in': self.sas_in,
+ 'nh': self.nh})
+
+ def object_id(self):
+ return "ipsec-tun-protect-%s-%s" % (self.itf, self.nh)
+
+ def add_vpp_config(self):
+ self.test.vapi.ipsec_tunnel_protect_update(
+ tunnel={
+ 'sw_if_index': self.itf._sw_if_index,
+ 'n_sa_in': len(self.sas_in),
+ 'sa_out': self.sa_out,
+ 'sa_in': self.sas_in,
+ 'nh': self.nh})
+ self.test.registry.register(self, self.test.logger)
+
+ def remove_vpp_config(self):
+ self.test.vapi.ipsec_tunnel_protect_del(
+ sw_if_index=self.itf.sw_if_index,
+ nh=self.nh)
+
+ def query_vpp_config(self):
+ bs = self.test.vapi.ipsec_tunnel_protect_dump(
+ sw_if_index=self.itf.sw_if_index)
+ for b in bs:
+ if b.tun.sw_if_index == self.itf.sw_if_index and \
+ self.nh == str(b.tun.nh):
+ return True
+ return False
+
+
+class VppIpsecInterface(VppInterface):
+ """
+ VPP IPSec interface
+ """
+
+ def __init__(self, test, mode=None, instance=0xffffffff):
+ super(VppIpsecInterface, self).__init__(test)
+
+ self.mode = mode
+ if not self.mode:
+ self.mode = (VppEnum.vl_api_tunnel_mode_t.
+ TUNNEL_API_MODE_P2P)
+ self.instance = instance
+
+ def add_vpp_config(self):
+ r = self.test.vapi.ipsec_itf_create(itf={
+ 'user_instance': self.instance,
+ 'mode': self.mode,
+ })
+ self.set_sw_if_index(r.sw_if_index)
+ self.test.registry.register(self, self.test.logger)
+ ts = self.test.vapi.ipsec_itf_dump(sw_if_index=self._sw_if_index)
+ self.instance = ts[0].itf.user_instance
+ return self
+
+ def remove_vpp_config(self):
+ self.test.vapi.ipsec_itf_delete(sw_if_index=self._sw_if_index)
+
+ def query_vpp_config(self):
+ ts = self.test.vapi.ipsec_itf_dump(sw_if_index=0xffffffff)
+ for t in ts:
+ if t.itf.sw_if_index == self._sw_if_index: