import socket from vpp_object import VppObject class VppLispLocatorSet(VppObject): """ Represents LISP locator set in VPP """ def __init__(self, test, ls_name): self._test = test self._ls_name = ls_name @property def test(self): return self._test @property def ls_name(self): return self._ls_name def add_vpp_config(self): self.test.vapi.lisp_add_del_locator_set(ls_name=self._ls_name) self._test.registry.register(self, self.test.logger) def get_lisp_locator_sets_dump_entry(self): result = self.test.vapi.lisp_locator_set_dump() for ls in result: if ls.ls_name.strip(b'\x00') == self._ls_name: return ls return None def query_vpp_config(self): return self.get_lisp_locator_sets_dump_entry() is not None def remove_vpp_config(self): self.test.vapi.lisp_add_del_locator_set(ls_name=self._ls_name, is_add=0) def object_id(self): return 'lisp-locator-set-%s' % self._ls_name class VppLispLocator(VppObject): """ Represents LISP locator in VPP """ def __init__(self, test, sw_if_index, ls_name, priority=1, weight=1): self._test = test self._sw_if_index = sw_if_index self._ls_name = ls_name self._priority = priority self._weight = weight @property def test(self): """ Test which created this locator """ return self._test @property def ls_name(self): """ Locator set name """ return self._ls_name @property def sw_if_index(self): return self._sw_if_index @property def priority(self): return self._priority @property def weight(self): return self._weight def add_vpp_config(self): self.test.vapi.lisp_add_del_locator(ls_name=self._ls_name, sw_if_index=self._sw_if_index, priority=self._priority, weight=self._weight) self._test.registry.register(self, self.test.logger) def get_lisp_locator_dump_entry(self): locators = self.test.vapi.lisp_locator_dump( is_index_set=0, ls_name=self._ls_name) for locator in locators: if locator.sw_if_index == self._sw_if_index: return locator return None def query_vpp_config(self): locator = self.get_lisp_locator_dump_entry() return locator is not None def remove_vpp_config(self): self.test.vapi.lisp_add_del_locator( ls_name=self._ls_name, sw_if_index=self._sw_if_index, priority=self._priority, weight=self._weight, is_add=0) self._test.registry.register(self, self.test.logger) def object_id(self): return 'lisp-locator-%s-%d' % (self._ls_name, self._sw_if_index) class LispEIDType(object): IP4 = 0 IP6 = 1 MAC = 2 class LispKeyIdType(object): NONE = 0 SHA1 = 1 SHA256 = 2 class LispEID(object): """ Lisp endpoint identifier """ def __init__(self, eid): self.eid = eid # find out whether EID is ip4 prefix, ip6 prefix or MAC if self.eid.find("/") != -1: if self.eid.find(":") == -1: self.eid_type = LispEIDType.IP4 self.data_length = 4 else: self.eid_type = LispEIDType.IP6 self.data_length = 16 self.eid_address = self.eid.split("/")[0] self.prefix_length = int(self.eid.split("/")[1]) elif self.eid.count(":") == 5: # MAC address self.eid_type = LispEIDType.MAC self.eid_address = self.eid self.prefix_length = 0 self.data_length = 6 else: raise Exception('Unsupported EID format {!s}!'.format(eid)) @property def packed(self): if self.eid_type == LispEIDType.IP4: return socket.inet_pton(socket.AF_INET, self.eid_address) elif self.eid_type == LispEIDType.IP6: return socket.inet_pton(socket.AF_INET6, self.eid_address) elif self.eid_type == LispEIDType.MAC: return Exception('Unimplemented') raise Exception('Unknown EID type {!s}!'.format(self.eid_type)) class VppLispMapping(VppObject): """ Represents common features for remote and local LISP mapping in VPP """ def __init__(self, test, eid, vni=0, priority=1, weight=1): self._eid = LispEID(eid) self._test = test self._priority = priority self._weight = weight self._vni = vni @property def test(self): return self._test @property def vni(self): return self._vni @property def eid(self): return self._eid @property def priority(self): return self._priority @property def weight(self): return self._weight def get_lisp_mapping_dump_entry(self):