vpp - Vector Packet Processing

Age	Commit message (Expand)	Author	Files	Lines
2020-11-30	tests: Fix unversioned python shebang lines	Dave Wallace	1	-1/+1
2020-02-21	ipsec: IPSec protection for multi-point tunnel interfaces	Neale Ranns	1	-2/+3
2020-02-04	teib: Rename NHRP to TEIB	Neale Ranns	1	-0/+51
92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152
#!/usr/bin/env python3

from socket import inet_pton, inet_ntop, AF_INET, AF_INET6
import unittest

from framework import VppTestCase, VppTestRunner
from vpp_ip import DpoProto
from vpp_ip_route import VppIpRoute, VppRoutePath, VppMplsLabel, VppIpTable

from scapy.packet import Raw
from scapy.layers.l2 import Ether
from scapy.layers.inet import IP, UDP
from scapy.layers.inet6 import IPv6

from vpp_object import VppObject

NUM_PKTS = 67


def find_l3xc(test, sw_if_index, dump_sw_if_index=None):
    if not dump_sw_if_index:
        dump_sw_if_index = sw_if_index
    xcs = test.vapi.l3xc_dump(dump_sw_if_index)
    for xc in xcs:
        if sw_if_index == xc.l3xc.sw_if_index:
            return True
    return False


class VppL3xc(VppObject):

    def __init__(self,  test, intf, paths, is_ip6=False):
        self._test = test
        self.intf = intf
        self.is_ip6 = is_ip6
        self.paths = paths
        self.encoded_paths = []
        for path in self.paths:
            self.encoded_paths.append(path.encode())

    def add_vpp_config(self):
        self._test.vapi.l3xc_update(
            l3xc={
                'is_ip6': self.is_ip6,
                'sw_if_index': self.intf.sw_if_index,
                'n_paths': len(self.paths),
                'paths': self.encoded_paths
            })
        self._test.registry.register(self, self._test.logger)

    def remove_vpp_config(self):
        self._test.vapi.l3xc_del(
            is_ip6=self.is_ip6,
            sw_if_index=self.intf.sw_if_index)

    def query_vpp_config(self):
        return find_l3xc(self._test, self.intf.sw_if_index)

    def object_id(self):
        return ("l3xc-%d" % self.intf.sw_if_index)


class TestL3xc(VppTestCase):
    """ L3XC Test Case """

    @classmethod
    def setUpClass(cls):
        super(TestL3xc, cls).setUpClass()

    @classmethod
    def tearDownClass(cls):
        super(TestL3xc, cls).tearDownClass()

    def setUp(self):
        super(TestL3xc, self).setUp()

        self.create_pg_interfaces(range(6))

        for i in self.pg_interfaces:
            i.admin_up()
            i.config_ip4()
            i.resolve_arp()
            i.config_ip6()
            i.resolve_ndp()

    def tearDown(self):
        for i in self.pg_interfaces:
            i.unconfig_ip4()
            i.unconfig_ip6()
            i.admin_down()
        super(TestL3xc, self).tearDown()

    def send_and_expect_load_balancing(self, input, pkts, outputs):
        self.pg_send(input, pkts)
        rxs = []
        for oo in outputs:
            rx = oo._get_capture(1)
            self.assertNotEqual(0, len(rx))
            for r in rx:
                rxs.append(r)
        return rxs

    def test_l3xc4(self):
        """ IPv4 X-Connect """

        #
        # x-connect pg0 to pg1 and pg2 to pg3->5
        #
        l3xc_1 = VppL3xc(self, self.pg0,
                         [VppRoutePath(self.pg1.remote_ip4,
                                       self.pg1.sw_if_index)])
        l3xc_1.add_vpp_config()
        l3xc_2 = VppL3xc(self, self.pg2,
                         [VppRoutePath(self.pg3.remote_ip4,
                                       self.pg3.sw_if_index),
                          VppRoutePath(self.pg4.remote_ip4,
                                       self.pg4.sw_if_index),
                          VppRoutePath(self.pg5.remote_ip4,
                                       self.pg5.sw_if_index)])
        l3xc_2.add_vpp_config()

        self.assertTrue(find_l3xc(self, self.pg2.sw_if_index, 0xffffffff))

        self.logger.info(self.vapi.cli("sh l3xc"))

        #
        # fire in packets. If it's forwarded then the L3XC was successful,
        # since default routing will drop it
        #
        p_1 = (Ether(src=self.pg0.remote_mac,
                     dst=self.pg0.local_mac) /
               IP(src="1.1.1.1", dst="1.1.1.2") /
               UDP(sport=1234, dport=1234) /
               Raw(b'\xa5' * 100))
        # self.send_and_expect(self.pg0, p_1*NUM_PKTS, self.pg1)

        p_2 = []
        for ii in range(NUM_PKTS):
            p_2.append(Ether(src=self.pg0.remote_mac,
                             dst=self.pg0.local_mac) /
                       IP(src="1.1.1.1", dst="1.1.1.2") /
                       UDP(sport=1000 + ii, dport=1234) /
                       Raw(b'\xa5' * 100))
        self.send_and_expect_load_balancing(self.pg2, p_2,
                                            [self.pg3, self.pg4, self.pg5])

        l3xc_2.remove_vpp_config()
        self.send_and_assert_no_replies(self.pg2, p_2)


if __name__ == '__main__':
    unittest.main(testRunner=VppTestRunner)