--- /dev/null
+from random import randint
+from socket import AF_INET, AF_INET6
+from scapy.all import *
+from scapy.packet import *
+from scapy.fields import *
+from lisp import *
+from framework import *
+from vpp_object import *
+
+
+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_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('\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_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_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_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 {}!'.format(eid))
+
+ def __str__(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 {}!'.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):
+ return self.test.vapi.lisp_eid_table_dump(
+ eid_set=1, prefix_length=self._eid.prefix_length,
+ vni=self._vni, eid_type=self._eid.eid_type, eid=str(self._eid))
+
+ def query_vpp_config(self):
+ mapping = self.get_lisp_mapping_dump_entry()
+ return mapping
+
+
+class VppLocalMapping(VppLispMapping):
+ """ LISP Local mapping """
+ def __init__(self, test, eid, ls_name, vni=0, priority=1, weight=1,
+ key_id=LispKeyIdType.NONE, key=''):
+ super(VppLocalMapping, self).__init__(test, eid, vni, priority, weight)
+ self._ls_name = ls_name
+ self._key_id = key_id
+ self._key = key
+
+ @property
+ def ls_name(self):
+ return self._ls_name
+
+ @property
+ def key_id(self):
+ return self._key_id
+
+ @property
+ def key(self):
+ return self._key
+
+ def add_vpp_config(self):
+ self.test.vapi.lisp_local_mapping(
+ ls_name=self._ls_name, eid_type=self._eid.eid_type,
+ eid=str(self._eid), prefix_len=self._eid.prefix_length,
+ vni=self._vni, key_id=self._key_id, key=self._key)
+ self._test.registry.register(self, self.test.logger)
+
+ def remove_vpp_config(self):
+ self.test.vapi.lisp_local_mapping(
+ ls_name=self._ls_name, eid_type=self._eid.eid_type,
+ eid=str(self._eid), prefix_len=self._eid.prefix_length,
+ vni=self._vni, is_add=0)
+
+ def object_id(self):
+ return 'lisp-eid-local-mapping-%s[%d]' % (self._eid, self._vni)
+
+
+class VppRemoteMapping(VppLispMapping):
+
+ def __init__(self, test, eid, rlocs=None, vni=0, priority=1, weight=1):
+ super(VppRemoteMapping, self).__init__(test, eid, vni, priority,
+ weight)
+ self._rlocs = rlocs
+
+ @property
+ def rlocs(self):
+ return self._rlocs
+
+ def add_vpp_config(self):
+ self.test.vapi.lisp_remote_mapping(
+ rlocs=self._rlocs, eid_type=self._eid.eid_type,
+ eid=str(self._eid), eid_prefix_len=self._eid.prefix_length,
+ vni=self._vni, rlocs_num=len(self._rlocs))
+ self._test.registry.register(self, self.test.logger)
+
+ def remove_vpp_config(self):
+ self.test.vapi.lisp_remote_mapping(
+ eid_type=self._eid.eid_type, eid=str(self._eid),
+ eid_prefix_len=self._eid.prefix_length, vni=self._vni,
+ is_add=0, rlocs_num=0)
+
+ def object_id(self):
+ return 'lisp-eid-remote-mapping-%s[%d]' % (self._eid, self._vni)
+
+
+class VppLispAdjacency(VppObject):
+ """ Represents LISP adjacency in VPP """
+
+ def __init__(self, test, leid, reid, vni=0):
+ self._leid = LispEID(leid)
+ self._reid = LispEID(reid)
+ if self._leid.eid_type != self._reid.eid_type:
+ raise Exception('remote and local EID are different types!')
+ self._vni = vni
+ self._test = test
+
+ @property
+ def test(self):
+ return self._test
+
+ @property
+ def leid(self):
+ return self._leid
+
+ @property
+ def reid(self):
+ return self._reid
+
+ @property
+ def vni(self):
+ return self._vni
+
+ def add_vpp_config(self):
+ self.test.vapi.lisp_adjacency(
+ leid=str(self._leid),
+ reid=str(self._reid), eid_type=self._leid.eid_type,
+ leid_len=self._leid.prefix_length,
+ reid_len=self._reid.prefix_length, vni=self._vni)
+ self._test.registry.register(self, self.test.logger)
+
+ def eid_equal(self, eid, eid_type, eid_data, prefix_len):
+ if eid.eid_type != eid_type:
+ return False
+
+ if eid_type == LispEIDType.IP4 or eid_type == LispEIDType.IP6:
+ if eid.prefix_length != prefix_len:
+ return False
+
+ if str(eid) != eid_data[0:eid.data_length]:
+ return False
+
+ return True
+
+ def query_vpp_config(self):
+ res = self.test.vapi.lisp_adjacencies_get(vni=self._vni)
+ for adj in res.adjacencies:
+ if self.eid_equal(self._leid, adj.eid_type, adj.leid,
+ adj.leid_prefix_len) and \
+ self.eid_equal(self._reid, adj.eid_type, adj.reid,
+ adj.reid_prefix_len):
+ return True
+ return False
+
+ def remove_vpp_config(self):
+ self.test.vapi.lisp_adjacency(
+ leid=str(self._leid),
+ reid=str(self._reid), eid_type=self._leid.eid_type,
+ leid_len=self._leid.prefix_length,
+ reid_len=self._reid.prefix_length, vni=self._vni, is_add=0)
+
+ def object_id(self):
+ return 'lisp-adjacency-%s-%s[%d]' % (self._leid, self._reid, self._vni)
--- /dev/null
+#!/usr/bin/env python
+import unittest
+
+from scapy.packet import Raw
+from scapy.layers.inet import IP, UDP, Ether
+from py_lispnetworking.lisp import LISP_GPE_Header
+
+from util import ppp, ForeignAddressFactory
+from framework import VppTestCase, VppTestRunner
+from lisp import *
+
+
+class Driver(object):
+
+ config_order = ['locator-sets',
+ 'locators',
+ 'local-mappings',
+ 'remote-mappings',
+ 'adjacencies']
+
+ """ Basic class for data driven testing """
+ def __init__(self, test, test_cases):
+ self._test_cases = test_cases
+ self._test = test
+
+ @property
+ def test_cases(self):
+ return self._test_cases
+
+ @property
+ def test(self):
+ return self._test
+
+ def create_packet(self, src_if, dst_if, deid, payload=''):
+ """
+ Create IPv4 packet
+
+ param: src_if
+ param: dst_if
+ """
+ packet = (Ether(dst=src_if.local_mac, src=src_if.remote_mac) /
+ IP(src=src_if.remote_ip4, dst=deid) /
+ Raw(payload))
+ return packet
+
+ @abstractmethod
+ def run(self):
+ """ testing procedure """
+ pass
+
+
+class SimpleDriver(Driver):
+ """ Implements simple test procedure """
+ def __init__(self, test, test_cases):
+ super(SimpleDriver, self).__init__(test, test_cases)
+
+ def verify_capture(self, src_loc, dst_loc, capture):
+ """
+ Verify captured packet
+
+ :param src_loc: source locator address
+ :param dst_loc: destination locator address
+ :param capture: list of captured packets
+ """
+ self.test.assertEqual(len(capture), 1, "Unexpected number of "
+ "packets! Expected 1 but {} received"
+ .format(len(capture)))
+ packet = capture[0]
+ try:
+ ip_hdr = packet[IP]
+ # assert the values match
+ self.test.assertEqual(ip_hdr.src, src_loc, "IP source address")
+ self.test.assertEqual(ip_hdr.dst, dst_loc,
+ "IP destination address")
+ gpe_hdr = packet[LISP_GPE_Header]
+ self.test.assertEqual(gpe_hdr.next_proto, 1,
+ "next_proto is not ipv4!")
+ ih = gpe_hdr[IP]
+ self.test.assertEqual(ih.src, self.test.pg0.remote_ip4,
+ "unexpected source EID!")
+ self.test.assertEqual(ih.dst, self.test.deid_ip4,
+ "unexpected dest EID!")
+ except:
+ self.test.logger.error(ppp("Unexpected or invalid packet:",
+ packet))
+ raise
+
+ def configure_tc(self, tc):
+ for config_item in self.config_order:
+ for vpp_object in tc[config_item]:
+ vpp_object.add_vpp_config()
+
+ def run(self, dest):
+ """ Send traffic for each test case and verify that it
+ is encapsulated """
+ for tc in enumerate(self.test_cases):
+ self.test.logger.info('Running {}'.format(tc[1]['name']))
+ self.configure_tc(tc[1])
+
+ print self.test.vapi.cli("sh lisp loc")
+ print self.test.vapi.cli("sh lisp eid")
+ print self.test.vapi.cli("sh lisp adj vni 0")
+ print self.test.vapi.cli("sh lisp gpe entry")
+
+ packet = self.create_packet(self.test.pg0, self.test.pg1, dest,
+ 'data')
+ self.test.pg0.add_stream(packet)
+ self.test.pg0.enable_capture()
+ self.test.pg1.enable_capture()
+ self.test.pg_start()
+ capture = self.test.pg1.get_capture(1)
+ self.verify_capture(self.test.pg1.local_ip4,
+ self.test.pg1.remote_ip4, capture)
+ self.test.pg0.assert_nothing_captured()
+
+
+class TestLisp(VppTestCase):
+ """ Basic LISP test """
+
+ @classmethod
+ def setUpClass(cls):
+ super(TestLisp, cls).setUpClass()
+ cls.faf = ForeignAddressFactory()
+ cls.create_pg_interfaces(range(2)) # create pg0 and pg1
+ for i in cls.pg_interfaces:
+ i.admin_up() # put the interface upsrc_if
+ i.config_ip4() # configure IPv4 address on the interface
+ i.resolve_arp() # resolve ARP, so that we know VPP MAC
+
+ def setUp(self):
+ super(TestLisp, self).setUp()
+ self.vapi.lisp_enable_disable(is_enabled=1)
+
+ def test_lisp_basic_encap(self):
+ """Test case for basic encapsulation"""
+
+ self.deid_ip4_net = self.faf.net
+ self.deid_ip4 = self.faf.get_ip4()
+ self.seid_ip4 = '{}/{}'.format(self.pg0.local_ip4, 32)
+ self.rloc_ip4 = self.pg1.remote_ip4n
+
+ test_cases = [
+ {
+ 'name': 'basic ip4 over ip4',
+ 'locator-sets': [VppLispLocatorSet(self, 'ls-4o4')],
+ 'locators': [
+ VppLispLocator(self, self.pg1.sw_if_index, 'ls-4o4')
+ ],
+ 'local-mappings': [
+ VppLocalMapping(self, self.seid_ip4, 'ls-4o4')
+ ],
+ 'remote-mappings': [
+ VppRemoteMapping(self, self.deid_ip4_net,
+ [{
+ "is_ip4": 1,
+ "priority": 1,
+ "weight": 1,
+ "addr": self.rloc_ip4
+ }])
+ ],
+ 'adjacencies': [
+ VppLispAdjacency(self, self.seid_ip4, self.deid_ip4_net)
+ ]
+ }
+ ]
+ self.test_driver = SimpleDriver(self, test_cases)
+ self.test_driver.run(self.deid_ip4)
+
+
+if __name__ == '__main__':
+ unittest.main(testRunner=VppTestRunner)
import os
+import socket
import fnmatch
import time
from hook import Hook
def ip_mfib_dump(self):
return self.api(self.papi.ip_mfib_dump, {})
+
+ def lisp_enable_disable(self, is_enabled):
+ return self.api(
+ self.papi.lisp_enable_disable,
+ {
+ 'is_en': is_enabled,
+ })
+
+ def lisp_locator_set(self,
+ ls_name,
+ is_add=1):
+ return self.api(
+ self.papi.lisp_add_del_locator_set,
+ {
+ 'is_add': is_add,
+ 'locator_set_name': ls_name
+ })
+
+ def lisp_locator_set_dump(self):
+ return self.api(self.papi.lisp_locator_set_dump, {})
+
+ def lisp_locator(self,
+ ls_name,
+ sw_if_index,
+ priority=1,
+ weight=1,
+ is_add=1):
+ return self.api(
+ self.papi.lisp_add_del_locator,
+ {
+ 'is_add': is_add,
+ 'locator_set_name': ls_name,
+ 'sw_if_index': sw_if_index,
+ 'priority': priority,
+ 'weight': weight
+ })
+
+ def lisp_locator_dump(self, is_index_set, ls_name=None, ls_index=0):
+ return self.api(
+ self.papi.lisp_locator_dump,
+ {
+ 'is_index_set': is_index_set,
+ 'ls_name': ls_name,
+ 'ls_index': ls_index,
+ })
+
+ def lisp_local_mapping(self,
+ ls_name,
+ eid_type,
+ eid,
+ prefix_len,
+ vni=0,
+ key_id=0,
+ key="",
+ is_add=1):
+ return self.api(
+ self.papi.lisp_add_del_local_eid,
+ {
+ 'locator_set_name': ls_name,
+ 'is_add': is_add,
+ 'eid_type': eid_type,
+ 'eid': eid,
+ 'prefix_len': prefix_len,
+ 'vni': vni,
+ 'key_id': key_id,
+ 'key': key
+ })
+
+ def lisp_eid_table_dump(self,
+ eid_set=0,
+ prefix_length=0,
+ vni=0,
+ eid_type=0,
+ eid=None,
+ filter_opt=0):
+ return self.api(
+ self.papi.lisp_eid_table_dump,
+ {
+ 'eid_set': eid_set,
+ 'prefix_length': prefix_length,
+ 'vni': vni,
+ 'eid_type': eid_type,
+ 'eid': eid,
+ 'filter': filter_opt,
+ })
+
+ def lisp_remote_mapping(self,
+ eid_type,
+ eid,
+ eid_prefix_len=0,
+ vni=0,
+ rlocs=None,
+ rlocs_num=0,
+ is_src_dst=0,
+ is_add=1):
+ return self.api(
+ self.papi.lisp_add_del_remote_mapping,
+ {
+ 'is_add': is_add,
+ 'eid_type': eid_type,
+ 'eid': eid,
+ 'eid_len': eid_prefix_len,
+ 'rloc_num': rlocs_num,
+ 'rlocs': rlocs,
+ 'vni': vni,
+ 'is_src_dst': is_src_dst,
+ })
+
+ def lisp_adjacency(self,
+ leid,
+ reid,
+ leid_len,
+ reid_len,
+ eid_type,
+ is_add=1,
+ vni=0):
+ return self.api(
+ self.papi.lisp_add_del_adjacency,
+ {
+ 'is_add': is_add,
+ 'vni': vni,
+ 'eid_type': eid_type,
+ 'leid': leid,
+ 'reid': reid,
+ 'leid_len': leid_len,
+ 'reid_len': reid_len,
+ })
+
+ def lisp_adjacencies_get(self, vni=0):
+ return self.api(
+ self.papi.lisp_adjacencies_get,
+ {
+ 'vni': vni
+ })
Code Review / vpp.git / commitdiff