#!/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)