import socket import scapy.compat from scapy.layers.inet import IP, UDP from scapy.layers.inet6 import IPv6 from scapy.layers.l2 import Ether, GRE from scapy.packet import Raw from scapy.data import IP_PROTOS from framework import VppTestCase from util import ppp from vpp_ip_route import VppIpRoute, VppRoutePath from vpp_ip import INVALID_INDEX """ TestLB is a subclass of VPPTestCase classes. TestLB class defines Load Balancer test cases for: - IP4 to GRE4 encap on per-port vip case - IP4 to GRE6 encap on per-port vip case - IP6 to GRE4 encap on per-port vip case - IP6 to GRE6 encap on per-port vip case - IP4 to L3DSR encap on vip case - IP4 to L3DSR encap on per-port vip case - IP4 to NAT4 encap on per-port vip case - IP6 to NAT6 encap on per-port vip case As stated in comments below, GRE has issues with IPv6. All test cases involving IPv6 are executed, but received packets are not parsed and checked. """ class TestLB(VppTestCase): """ Load Balancer Test Case """ @classmethod def setUpClass(cls): super(TestLB, cls).setUpClass() cls.ass = range(5) cls.packets = range(1) try: cls.create_pg_interfaces(range(2)) cls.interfaces = list(cls.pg_interfaces) for i in cls.interfaces: i.admin_up() i.config_ip4() i.config_ip6() i.disable_ipv6_ra() i.resolve_arp() i.resolve_ndp() dst4 = VppIpRoute(cls, "10.0.0.0", 24, [VppRoutePath(cls.pg1.remote_ip4, INVALID_INDEX)], register=False) dst4.add_vpp_config() dst6 = VppIpRoute(cls, "2002::", 16, [VppRoutePath(cls.pg1.remote_ip6, INVALID_INDEX)], register=False) dst6.add_vpp_config() cls.vapi.lb_conf(ip4_src_address="39.40.41.42", ip6_src_address="2004::1") except Exception: super(TestLB, cls).tearDownClass() raise @classmethod def tearDownClass(cls): super(TestLB, cls).tearDownClass() def tearDown(self): super(TestLB, self).tearDown() def show_commands_at_teardown(self): self.logger.info(self.vapi.cli("show lb vip verbose")) def getIPv4Flow(self, id): return (IP(dst="90.0.%u.%u" % (id / 255, id % 255), src="40.0.%u.%u" % (id / 255, id % 255)) / UDP(sport=10000 + id, dport=20000)) def getIPv6Flow(self, id): return (IPv6(dst="2001::%u" % (id), src="fd00:f00d:ffff::%u" % (id)) / UDP(sport=10000 + id, dport=20000)) def generatePackets(self, src_if, isv4): self.reset_packet_infos() pkts = [] for pktid in self.packets: info = self.create_packet_info(src_if, self.pg1) payload = self.info_to_payload(info) ip = self.getIPv4Flow(pktid) if isv4 else self.getIPv6Flow(pktid) packet = (Ether(dst=src_if.local_mac, src=src_if.remote_mac) / ip / Raw(payload)) self.extend_packet(packet, 128) info.data = packet.copy() pkts.append(packet) return pkts def checkInner(self, gre, isv4): IPver = IP if isv4 else IPv6 self.assertEqual(gre.proto, 0x0800 if isv4 else 0x86DD) self.assertEqual(gre.flags, 0) self.assertEqual(gre.version, 0) inner = IPver(scapy.compat.raw(gre.payload)) payload_info = self.payload_to_info(inner[Raw]) self.info = self.packet_infos[payload_info.index] self.assertEqual(payload_info.src, self.pg0.sw_if_index) self.assertEqual(scapy.compat.raw(inner), scapy.compat.raw(self.info.data[IPver])) def checkCapture(self, encap, isv4): self.pg0.assert_nothing_captured() out = self.pg1.get_capture(len(self.packets)) load = [0] * len(self.ass) self.info = None for p in out: try: asid = 0 gre = None if (encap == 'gre4'): ip = p[IP] asid = int(ip.dst.split(".")[3]) self.assertEqual(ip.version, 4) self.assertEqual(ip.flags, 0) self.assertEqual(ip.src, "39.40.41.42") self.assertEqual(ip.dst, "10.0.0.%u" % asid) self.assertEqual(ip.proto, 47) self.assertEqual(len(ip.options), 0) gre = p[GRE] self.checkInner(gre, isv4) elif (encap == 'gre6'): ip = p[IPv6] asid = ip.dst.split(":") asid = asid[len(asid) - 1] asid = 0 if asid == "" else int(asid) self.assertEqual(ip.version, 6) self.assertEqual(ip.tc, 0) self.assertEqual(ip.fl, 0) self.assertEqual(ip.src, "2004::1") self.assertEqual( socket.inet_pton(socket.AF_INET6, ip.dst), socket.inet_pton(socket.AF_INET6, "2002::%u" % asid) ) self.assertEqual(ip.nh, 47) # self.assertEqual(len(ip.options), 0) gre = GRE(scapy.compat.raw(p[IPv6].payload)) self.checkInner(gre, isv4) elif (encap == 'l3dsr'): ip = p[IP] asid = int(ip.dst.split(".")[3]) self.assertEqual(ip.version, 4) self.assertEqual(ip.flags, 0) self.assertEqual(ip.dst, "10.0.0.%u" % asid) self.assertEqual(ip.tos, 0x1c) self.assertEqual(len(
from vpp_interface import VppInterface


class VppMPLSTunnelInterface(VppInterface):
    """
    VPP MPLS Tunnel interface
    """

    def __init__(self, test, paths, is_multicast=0, is_l2=0):
        """ Create MPLS Tunnel interface """
        super(VppMPLSTunnelInterface, self).__init__(test)
        self.t_paths = paths
        self.is_multicast = is_multicast
        self.is_l2 = is_l2
        self.encoded_paths = []
        for path in self.t_paths:
            self.encoded_paths.append(path.encode())

    def add_vpp_config(self):
        reply = self.test.vapi.mpls_tunnel_add_del(
            0xffffffff,
            self.encoded_paths,
            is_multicast=self.is_multicast,
            l2_only=self.is_l2)
        self.set_sw_if_index(reply.sw_if_index)
        self.tunnel_index = reply.tunnel_index
        self._test.registry.register(self, self._test.logger