test/test_pipe.py

#!/usr/bin/env python3
from socket import AF_INET, AF_INET6, inet_pton
import unittest
from ipaddress import IPv4Network

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

from framework import VppTestCase, VppTestRunner
from vpp_interface import VppInterface
from vpp_ip_route import VppIpTable, VppIpRoute, VppRoutePath
from vpp_acl import AclRule, VppAcl, VppAclInterface

NUM_PKTS = 67


class VppPipe(VppInterface):
    """
    VPP Pipe
    """

    @property
    def east(self):
        return self.result.pipe_sw_if_index[1]

    @property
    def west(self):
        return self.result.pipe_sw_if_index[0]

    def __init__(self, test, instance=0xFFFFFFFF):
        super(VppPipe, self).__init__(test)
        self._test = test
        self.instance = instance

    def add_vpp_config(self):
        self.result = self._test.vapi.pipe_create(
            0 if self.instance == 0xFFFFFFFF else 1, self.instance
        )
        self.set_sw_if_index(self.result.sw_if_index)

    def remove_vpp_config(self):
        self._test.vapi.pipe_delete(self.result.sw_if_index)

    def object_id(self):
        return "pipe-%d" % (self._sw_if_index)

    def query_vpp_config(self):
        pipes = self._test.vapi.pipe_dump()
        for p in pipes:
            if p.sw_if_index == self.result.sw_if_index:
                return True
        return False

    def set_unnumbered(self, ip_sw_if_index, is_east, is_add=True):
        if is_east:
            res = self._test.vapi.sw_interface_set_unnumbered(
                ip_sw_if_index, self.east, is_add
            )
        else:
            res = self._test.vapi.sw_interface_set_unnumbered(
                ip_sw_if_index, self.west, is_add
            )


class TestPipe(VppTestCase):
    """Pipes"""

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

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

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

        self.create_pg_interfaces(range(4))

        for i in self.pg_interfaces:
            i.admin_up()

    def tearDown(self):
        for i in self.pg_interfaces:
            i.admin_down()

        super(TestPipe, self).tearDown()

    def test_pipe(self):
        """Pipes"""

        pipes = [VppPipe(self), VppPipe(self, 10)]

        for p in pipes:
            p.add_vpp_config()
            p.admin_up()

        #
        # L2 cross-connect pipe0 east with pg0 and west with pg1
        #
        self.vapi.sw_interface_set_l2_xconnect(
            self.pg0.sw_if_index, pipes[0].east, enable=1
        )
        self.vapi.sw_interface_set_l2_xconnect(
            pipes[0].east, self.pg0.sw_if_index, enable=1
        )
        self.vapi.sw_interface_set_l2_xconnect(
            self.pg1.sw_if_index, pipes[0].west, enable=1
        )
        self.vapi.sw_interface_set_l2_xconnect(
            pipes[0].west, self.pg1.sw_if_index, enable=1
        )

        # test bi-directional L2 flow pg0<->pg1
        p = (
            Ether(src=self.pg0.remote_mac, dst=self.pg1.remote_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 * NUM_PKTS, self.pg1)
        self.send_and_expect(self.pg1, p * NUM_PKTS, self.pg0)

        #
        # Attach ACL to ensure features are run on the pipe
        #
        rule_1 = AclRule(
            is_permit=0,
            proto=17,
            src_prefix=IPv4Network("1.1.1.1/32"),
            dst_prefix=IPv4Network("1.1.1.2/32"),
            ports=1234,
        )
        acl = VppAcl(self, rules=[rule_1])
        acl.add_vpp_config()

        # Apply the ACL on the pipe on output
        acl_if_e = VppAclInterface(
            self, sw_if_index=pipes[0].east, n_input=0, acls=[acl]
        )
        acl_if_e.add_vpp_config()

        self.send_and_assert_no_replies(self.pg0, p * NUM_PKTS)
        self.send_and_expect(self.pg1, p * NUM_PKTS, self.pg0)

        # remove from output and apply on input
        acl_if_e.remove_vpp_config()
        acl_if_w = VppAclInterface(
            self, sw_if_index=pipes[0].west, n_input=1, acls=[acl]
        )
        acl_if_w.add_vpp_config()

        self.send_and_assert_no_replies(self.pg0, p * NUM_PKTS)
        self.send_and_expect(self.pg1, p * NUM_PKTS, self.pg0)

        acl_if_w.remove_vpp_config()
        self.send_and_expect(self.pg0, p * NUM_PKTS, self.pg1)
        self.send_and_expect(self.pg1, p * NUM_PKTS, self.pg0)

        #
        # L3 routes in two separate tables so a pipe can be used to L3
        # x-connect
        #
        tables = []
        tables.append(VppIpTable(self, 1))
        tables.append(VppIpTable(self, 2))

        for t in tables:
            t.add_vpp_config()

        self.pg2.set_table_ip4(1)
        self.pg2.config_ip4()
        self.pg2.resolve_arp()
        self.pg3.set_table_ip4(2)
        self.pg3.config_ip4()
        self.pg3.resolve_arp()

        routes = []
        routes.append(
            VppIpRoute(
                self,
                "1.1.1.1",
                32,
                [VppRoutePath(self.pg3.remote_ip4, self.pg3.sw_if_index)],
                table_id=2,
            )
        )
        routes.append(
            VppIpRoute(
                self,
                "1.1.1.1",
                32,
                [VppRoutePath("0.0.0.0", pipes[1].east)],
                table_id=1,
            )
        )
        routes.append(
            VppIpRoute(
                self,
                "1.1.1.2",
                32,
                [VppRoutePath("0.0.0.0", pipes[1].west)],
                table_id=2,
            )
        )
        routes.append(
            VppIpRoute(
                self,
                "1.1.1.2",
                32,
                [VppRoutePath(self.pg2.remote_ip4, self.pg2.sw_if_index)],
                table_id=1,
            )
        )

        for r in routes:
            r.add_vpp_config()

        p_east = (
            Ether(src=self.pg2.remote_mac, dst=self.pg2.local_mac)
            / IP(src="1.1.1.2", dst="1.1.1.1")
            / UDP(sport=1234, dport=1234)
            / Raw(b"\xa5" * 100)
        )

        # bind the pipe ends to the correct tables
        self.vapi.sw_interface_set_table(pipes[1].west, 0, 2)
        self.vapi.sw_interface_set_table(pipes[1].east, 0, 1)

        # IP is not enabled on the pipes at this point
        self.send_and_assert_no_replies(self.pg2, p_east * NUM_PKTS)

        # IP enable the Pipes by making them unnumbered
        pipes[1].set_unnumbered(self.pg2.sw_if_index, True)
        pipes[1].set_unnumbered(self.pg3.sw_if_index, False)

        self.send_and_expect(self.pg2, p_east * NUM_PKTS, self.pg3)

        # and the return path
        p_west = (
            Ether(src=self.pg3.remote_mac, dst=self.pg3.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.pg3, p_west * NUM_PKTS, self.pg2)

        #
        # Use ACLs to test features run on the Pipes
        #
        acl_if_e1 = VppAclInterface(
            self, sw_if_index=pipes[1].east, n_input=0, acls=[acl]
        )
        acl_if_e1.add_vpp_config()
        self.send_and_assert_no_replies(self.pg2, p_east * NUM_PKTS)
        self.send_and_expect(self.pg3, p_west * NUM_PKTS, self.pg2)

        # remove from output and apply on input
        acl_if_e1.remove_vpp_config()
        acl_if_w1 = VppAclInterface(
            self, sw_if_index=pipes[1].west, n_input=1, acls=[acl]
        )
        acl_if_w1.add_vpp_config()
        self.send_and_assert_no_replies(self.pg2, p_east * NUM_PKTS)
        self.send_and_expect(self.pg3, p_west * NUM_PKTS, self.pg2)
        acl_if_w1.remove_vpp_config()

        self.send_and_expect(self.pg2, p_east * NUM_PKTS, self.pg3)
        self.send_and_expect(self.pg3, p_west * NUM_PKTS, self.pg2)

        # cleanup (so the tables delete)
        self.pg2.unconfig_ip4()
        self.pg2.set_table_ip4(0)
        self.pg3.unconfig_ip4()
        self.pg3.set_table_ip4(0)
        self.vapi.sw_interface_set_table(pipes[1].west, 0, 0)
        self.vapi.sw_interface_set_table(pipes[1].east, 0, 0)


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