X-Git-Url: https://gerrit.fd.io/r/gitweb?a=blobdiff_plain;f=test%2Fvpp_ipsec.py;h=268fe687876dee5bb2accc0cdfcd99bdb527fa7f;hb=7c0eb56f4;hp=8c6da77cba8fb5bbae6d6e9e5da1078f9931f038;hpb=311124e21b9b8ca11073d82408ffb81d98790969;p=vpp.git diff --git a/test/vpp_ipsec.py b/test/vpp_ipsec.py index 8c6da77cba8..268fe687876 100644 --- a/test/vpp_ipsec.py +++ b/test/vpp_ipsec.py @@ -1,5 +1,6 @@ -from vpp_object import * +from vpp_object import VppObject from ipaddress import ip_address +from vpp_papi import VppEnum try: text_type = unicode @@ -7,6 +8,10 @@ except NameError: text_type = str +def mk_counter(): + return {'packets': 0, 'bytes': 0} + + class VppIpsecSpd(VppObject): """ VPP SPD DB @@ -23,9 +28,6 @@ class VppIpsecSpd(VppObject): def remove_vpp_config(self): self.test.vapi.ipsec_spd_add_del(self.id, is_add=0) - def __str__(self): - return self.object_id() - def object_id(self): return "ipsec-spd-%d" % self.id @@ -40,7 +42,7 @@ class VppIpsecSpd(VppObject): class VppIpsecSpdItfBinding(VppObject): """ VPP SPD DB to interface binding - (i.e. this SPD is used on this interfce) + (i.e. this SPD is used on this interface) """ def __init__(self, test, spd, itf): @@ -58,9 +60,6 @@ class VppIpsecSpdItfBinding(VppObject): self.itf.sw_if_index, is_add=0) - def __str__(self): - return self.object_id() - def object_id(self): return "bind-%s-to-%s" % (self.spd.id, self.itf) @@ -82,7 +81,7 @@ class VppIpsecSpdEntry(VppObject): remote_start, remote_stop, proto, priority=100, - policy=0, + policy=None, is_outbound=1, remote_port_start=0, remote_port_stop=65535, @@ -98,7 +97,11 @@ class VppIpsecSpdEntry(VppObject): self.proto = proto self.is_outbound = is_outbound self.priority = priority - self.policy = policy + if not policy: + self.policy = (VppEnum.vl_api_ipsec_spd_action_t. + IPSEC_API_SPD_ACTION_BYPASS) + else: + self.policy = policy self.is_ipv6 = (0 if self.local_start.version == 4 else 1) self.local_port_start = local_port_start self.local_port_stop = local_port_stop @@ -106,13 +109,13 @@ class VppIpsecSpdEntry(VppObject): self.remote_port_stop = remote_port_stop def add_vpp_config(self): - self.test.vapi.ipsec_spd_add_del_entry( + rv = self.test.vapi.ipsec_spd_entry_add_del( self.spd.id, self.sa_id, - self.local_start.packed, - self.local_stop.packed, - self.remote_start.packed, - self.remote_stop.packed, + self.local_start, + self.local_stop, + self.remote_start, + self.remote_stop, protocol=self.proto, is_ipv6=self.is_ipv6, is_outbound=self.is_outbound, @@ -122,16 +125,17 @@ class VppIpsecSpdEntry(VppObject): local_port_stop=self.local_port_stop, remote_port_start=self.remote_port_start, remote_port_stop=self.remote_port_stop) + self.stat_index = rv.stat_index self.test.registry.register(self, self.test.logger) def remove_vpp_config(self): - self.test.vapi.ipsec_spd_add_del_entry( + self.test.vapi.ipsec_spd_entry_add_del( self.spd.id, self.sa_id, - self.local_start.packed, - self.local_stop.packed, - self.remote_start.packed, - self.remote_stop.packed, + self.local_start, + self.local_stop, + self.remote_start, + self.remote_stop, protocol=self.proto, is_ipv6=self.is_ipv6, is_outbound=self.is_outbound, @@ -143,9 +147,6 @@ class VppIpsecSpdEntry(VppObject): remote_port_stop=self.remote_port_stop, is_add=0) - def __str__(self): - return self.object_id() - def object_id(self): return "spd-entry-%d-%d-%d-%d-%d-%d" % (self.spd.id, self.priority, @@ -157,15 +158,26 @@ class VppIpsecSpdEntry(VppObject): def query_vpp_config(self): ss = self.test.vapi.ipsec_spd_dump(self.spd.id) for s in ss: - if s.sa_id == self.sa_id and \ - s.is_outbound == self.is_outbound and \ - s.priority == self.priority and \ - s.policy == self.policy and \ - s.is_ipv6 == self.is_ipv6 and \ - s.remote_start_port == self.remote_port_start: + if s.entry.sa_id == self.sa_id and \ + s.entry.is_outbound == self.is_outbound and \ + s.entry.priority == self.priority and \ + s.entry.policy == self.policy and \ + s.entry.remote_address_start == self.remote_start and \ + s.entry.remote_port_start == self.remote_port_start: return True return False + def get_stats(self, worker=None): + c = self.test.statistics.get_counter("/net/ipsec/policy") + 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 VppIpsecSA(VppObject): """ @@ -177,8 +189,8 @@ class VppIpsecSA(VppObject): crypto_alg, crypto_key, proto, tun_src=None, tun_dst=None, - use_anti_replay=0, - udp_encap=0): + flags=None, salt=0): + e = VppEnum.vl_api_ipsec_sad_flags_t self.test = test self.id = id self.spi = spi @@ -187,22 +199,24 @@ class VppIpsecSA(VppObject): self.crypto_alg = crypto_alg self.crypto_key = crypto_key self.proto = proto - self.is_tunnel = 0 - self.is_tunnel_v6 = 0 + self.salt = salt + self.tun_src = tun_src self.tun_dst = tun_dst + if not flags: + self.flags = e.IPSEC_API_SAD_FLAG_NONE + else: + self.flags = flags if (tun_src): self.tun_src = ip_address(text_type(tun_src)) - self.is_tunnel = 1 + self.flags = self.flags | e.IPSEC_API_SAD_FLAG_IS_TUNNEL if (self.tun_src.version == 6): - self.is_tunnel_v6 = 1 + self.flags = self.flags | e.IPSEC_API_SAD_FLAG_IS_TUNNEL_V6 if (tun_dst): self.tun_dst = ip_address(text_type(tun_dst)) - self.use_anti_replay = use_anti_replay - self.udp_encap = udp_encap def add_vpp_config(self): - self.test.vapi.ipsec_sad_add_del_entry( + r = self.test.vapi.ipsec_sad_entry_add_del( self.id, self.spi, self.integ_alg, @@ -210,16 +224,15 @@ class VppIpsecSA(VppObject): self.crypto_alg, self.crypto_key, self.proto, - (self.tun_src.packed if self.tun_src else []), - (self.tun_dst.packed if self.tun_dst else []), - is_tunnel=self.is_tunnel, - is_tunnel_ipv6=self.is_tunnel_v6, - use_anti_replay=self.use_anti_replay, - udp_encap=self.udp_encap) + (self.tun_src if self.tun_src else []), + (self.tun_dst if self.tun_dst else []), + flags=self.flags, + salt=self.salt) + self.stat_index = r.stat_index self.test.registry.register(self, self.test.logger) def remove_vpp_config(self): - self.test.vapi.ipsec_sad_add_del_entry( + self.test.vapi.ipsec_sad_entry_add_del( self.id, self.spi, self.integ_alg, @@ -227,23 +240,85 @@ class VppIpsecSA(VppObject): self.crypto_alg, self.crypto_key, self.proto, - (self.tun_src.packed if self.tun_src else []), - (self.tun_dst.packed if self.tun_dst else []), - is_tunnel=self.is_tunnel, - is_tunnel_ipv6=self.is_tunnel_v6, - use_anti_replay=self.use_anti_replay, - udp_encap=self.udp_encap, + (self.tun_src if self.tun_src else []), + (self.tun_dst if self.tun_dst else []), + flags=self.flags, is_add=0) - def __str__(self): - return self.object_id() - def object_id(self): return "ipsec-sa-%d" % self.id def query_vpp_config(self): bs = self.test.vapi.ipsec_sa_dump() for b in bs: - if b.sa_id == self.id: + if b.entry.sad_id == self.id: + 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