#!/usr/bin/env python3 import unittest import binascii from socket import AF_INET6 from framework import VppTestCase, VppTestRunner from vpp_ip_route import VppIpRoute, VppRoutePath, FibPathProto, VppIpTable from vpp_srv6 import SRv6LocalSIDBehaviors, VppSRv6LocalSID, VppSRv6Policy, \ SRv6PolicyType, VppSRv6Steering, SRv6PolicySteeringTypes import scapy.compat from scapy.packet import Raw from scapy.layers.l2 import Ether, Dot1Q from scapy.layers.inet6 import IPv6, UDP, IPv6ExtHdrSegmentRouting from scapy.layers.inet import IP, UDP from util import ppp class TestSRv6(VppTestCase): """ SRv6 Test Case """ @classmethod def setUpClass(cls): super(TestSRv6, cls).setUpClass() @classmethod def tearDownClass(cls): super(TestSRv6, cls).tearDownClass() def setUp(self): """ Perform test setup before each test case. """ super(TestSRv6, self).setUp() # packet sizes, inclusive L2 overhead self.pg_packet_sizes = [64, 512, 1518, 9018] # reset packet_infos self.reset_packet_infos() def tearDown(self): """ Clean up test setup after each test case. """ self.teardown_interfaces() super(TestSRv6, self).tearDown() def configure_interface(self, interface, ipv6=False, ipv4=False, ipv6_table_id=0, ipv4_table_id=0): """ Configure interface. :param ipv6: configure IPv6 on interface :param ipv4: configure IPv4 on interface :param ipv6_table_id: FIB table_id for IPv6 :param ipv4_table_id: FIB table_id for IPv4 """ self.logger.debug("Configuring interface %s" % (interface.name)) if ipv6: self.logger.debug("Configuring IPv6") interface.set_table_ip6(ipv6_table_id) interface.config_ip6() interface.resolve_ndp(timeout=5) if ipv4: self.logger.debug("Configuring IPv4") interface.set_table_ip4(ipv4_table_id) interface.config_ip4() interface.resolve_arp() interface.admin_up() def setup_interfaces(self, ipv6=[], ipv4=[], ipv6_table_id=[], ipv4_table_id=[]): """ Create and configure interfaces. :param ipv6: list of interface IPv6 capabilities :param ipv4: list of interface IPv4 capabilities :param ipv6_table_id: list of intf IPv6 FIB table_ids :param ipv4_table_id: list of intf IPv4 FIB table_ids :returns: List of created interfaces. """ # how many interfaces? if len(ipv6): count = len(ipv6) else: count = len(ipv4) self.logger.debug("Creating and configuring %d interfaces" % (count)) # fill up ipv6 and ipv4 lists if needed # not enabled (False) is the default if len(ipv6) < count: ipv6 += (count - len(ipv6)) * [False] if len(ipv4) < count: ipv4 += (count - len(ipv4)) * [False] # fill up table_id lists if needed # table_id 0 (global) is the default if len(ipv6_table_id) < count: ipv6_table_id += (count - len(ipv6_table_id)) * [0] if len(ipv4_table_id) < count: ipv4_table_id += (count - len(ipv4_table_id)) * [0] # create 'count' pg interfaces self.create_pg_interfaces(range(count)) # setup all interfaces for i in range(count): intf = self.pg_interfaces[i] self.configure_interface(intf, ipv6[i], ipv4[i], ipv6_table_id[i], ipv4_table_id[i]) if any(ipv6): self.logger.debug(self.vapi.cli("show ip6 neighbors")) if any(ipv4): self.logger.debug(self.vapi.cli("show ip4 neighbors")) self.logger.debug(self.vapi.cli("show interface")) self.logger.debug(self.vapi.cli("show hardware")) return self.pg_interfaces def teardown_interfaces(self): """ Unconfigure and bring down interface. """ self.logger.debug("Tearing down interfaces") # tear down all interfaces # AFAIK they cannot be deleted for i in self.pg_interfaces: self.logger.debug("Tear down interface %s" % (i.name)) i.admin_down() i.unconfig() i.set_table_ip4(0) i.set_table_ip6(0) @unittest.skipUnless(0, "PC to fix") def test_SRv6_T_Encaps(self): """ Test SRv6 Transit.Encaps behavior for IPv6. """ # send traffic to one destination interface # source and destination are IPv6 only self.setup_interfaces(ipv6=[True, True]) # configure FIB entries route = VppIpRoute(self, "a4::", 64, [VppRoutePath(self.pg1.remote_ip6, self.pg1.sw_if_index)]) route.add_vpp_config() # configure encaps IPv6 source address # needs to be done before SR Policy config # TODO: API? self.vapi.cli("set sr encaps source addr a3::") bsid = 'a3::9999:1' # configure SRv6 Policy # Note: segment list order: first -> last sr_policy = VppSRv6Policy( self, bsid=bsid, is_encap=1, sr_type=SRv6PolicyType.SR_POLICY_TYPE_DEFAULT, weight=1, fib_table=0, segments=['a4::', 'a5::', 'a6::c7'], source='a3::') sr_policy.add_vpp_config() self.sr_policy = sr_policy # log the sr policies self.logger.info(self.vapi.cli("show sr policies")) # steer IPv6 traffic to a7::/64 into SRv6 Policy # use the bsid of the above self.sr_policy pol_steering = VppSRv6Steering( self, bsid=self.sr_policy.bsid, prefix="a7::", mask_width=64, traffic_type=SRv6PolicySteeringTypes.SR_STEER_IPV6, sr_policy_index=0, table_id=0, sw_if_index=0) pol_steering.add_vpp_config() # log the sr steering policies self.logger.info(self.vapi.cli("show sr steering policies")) # create packets count = len(self.pg_packet_sizes) dst_inner = 'a7::1234' pkts = [] # create IPv6 packets without SRH packet_header = self.create_packet_header_IPv6(dst_inner) # create traffic stream pg0->pg1 pkts.extend(self.create_stream(self.pg0, self.pg1, packet_header, self.pg_packet_sizes, count)) # create IPv6 packets with SRH # packets with segments-left 1, active segment a7:: packet_header = self.create_packet_header_IPv6_SRH( sidlist=['a8::', 'a7::', 'a6::'], segleft=1) # create traffic stream pg0->pg1 pkts.extend(self.create_stream(self.pg0, self.pg1, packet_header, self.pg_packet_sizes, count)) # create IPv6 packets with SRH and IPv6 # packets with segments-left 1, active segment a7:: packet_header = self.create_packet_header_IPv6_SRH_IPv6( dst_inner, sidlist=['a8::', 'a7::', 'a6::'], segleft=1) # create traffic stream pg0->pg1 pkts.extend(self.create_stream(self.pg0, self.pg1, packet_header, self.pg_packet_sizes, count)) # send packets and verify received packets self.send_and_verify_pkts(self.pg0, pkts, self.pg1, self.compare_rx_tx_packet_T_Encaps) # log the localsid counters self.logger.info(self.vapi.cli("show sr localsid")) # remove SR steering pol_steering.remove_vpp_config() self.logger.info(self.vapi.cli("show sr steering policies")) # remove SR Policies self.sr_policy.remove_vpp_config() self.logger.info(self.vapi.cli("show sr policies")) # remove FIB entries # done by tearDown # cleanup interfaces self.teardown_interfaces() @unittest.skipUnless(0, "PC to fix") def test_SRv6_T_Insert(self): """ Test SRv6 Transit.Insert behavior (IPv6 only). """ # send traffic to one destination interface # source and destination are IPv6 only self.setup_interfaces(ipv6=[True, True]) # configure FIB entries route = VppIpRoute(self, "a4::", 64, [VppRoutePath(self.pg1.remote_ip6, self.pg1.sw_if_index)]) route.add_vpp_config() # configure encaps IPv6 source address # needs to be done before SR Policy config # TODO: API? self.vapi.cli("set sr encaps source addr a3::") bsid = 'a3::9999:1' # configure SRv6 Policy # Note: segment list order: first -> last sr_policy = VppSRv6Policy( self, bsid=bsid, is_encap=0, sr_type=SRv6PolicyType.SR_POLICY_TYPE_DEFAULT, weight=1, fib_table=0, segments=['a4::', 'a5::', 'a6::c7'], source='a3::') sr_policy.add_vpp_config() self.sr_policy = sr_policy # log the sr policies self.logger.info(self.vapi.cli("show sr policies")) # steer IPv6 traffic to a7::/64 into SRv6 Policy # use the bsid of the above self.sr_policy pol_steering = VppSRv6Steering( self, bsid=self.sr_policy.bsid, prefix="a7::", mask_width=64, traffic_type=SRv6PolicySteeringTypes.SR_STEER_IPV6, sr_policy_index=0, table_id=0, sw_if_index=0) pol_steering.add_vpp_config() # log the sr steering policies self.logger.info(self.vapi.cli("show sr steering policies")) # create packets count = len(self.pg_packet_sizes) dst_inner = 'a7::1234' pkts = [] # create IPv6 packets without SRH packet_header = self.create_packet_header_IPv6(dst_inner) # create traffic stream pg0->pg1 pkts.extend(self.create_stream(self.pg0, self.pg1, packet_header, self.pg_packet_sizes, count)) # create IPv6 packets with SRH # packets with segments-left 1, active segment a7:: packet_header = self.create_packet_header_IPv6_SRH( sidlist=['a8::', 'a7::', 'a6::'], segleft=1) # create traffic stream pg0->pg1 pkts.extend(self.create_stream(self.pg0, self.pg1, packet_header, self.pg_packet_sizes, count)) # send packets and verify received packets self.send_and_verify_pkts(self.pg0, pkts, self.pg1, self.compare_rx_tx_packet_T_Insert) # log the localsid counters self.logger.info(self.vapi.cli("show sr localsid")) # remove SR steering pol_steering.remove_vpp_config() self.logger.info(self.vapi.cli("show sr steering policies")) # remove SR Policies self.sr_policy.remove_vpp_config() self.logger.info(self.vapi.cli("show sr policies")) # remove FIB entries # done by tearDown # cleanup interfaces self.teardown_interfaces() @unittest.skipUnless(0, "PC to fix") def test_SRv6_T_Encaps_IPv4(self): """ Test SRv6 Transit.Encaps behavior for IPv4. """ # send traffic to one destination interface # source interface is IPv4 only # destination interface is IPv6 only self.setup_interfaces(ipv6=[False, True], ipv4=[True, False]) # configure FIB entries route = VppIpRoute(self, "a4::", 64, [VppRoutePath(self.pg1.remote_ip6, self.pg1.sw_if_index)]) route.add_vpp_config() # configure encaps IPv6 source address # needs to be done before SR Policy config # TODO: API? self.vapi.cli("set sr encaps source addr a3::") bsid = 'a3::9999:1' # configure SRv6 Policy # Note: segment list order: first -> last sr_policy = VppSRv6Policy( self, bsid=bsid, is_encap=1, sr_type=SRv6PolicyType.SR_POLICY_TYPE_DEFAULT, weight=1, fib_table=0, segments=['a4::', 'a5::', 'a6::c7'], source='a3::') sr_policy.add_vpp_config() self.sr_policy = sr_policy # log the sr policies self.logger.info(self.vapi.cli("show sr policies")) # steer IPv4 traffic to 7.1.1.0/24 into SRv6 Policy # use the bsid of the above self.sr_policy pol_steering = VppSRv6Steering( self, bsid=self.sr_policy.bsid, prefix="7.1.1.0", mask_width=24, traffic_type=SRv6PolicySteeringTypes.SR_STEER_IPV4, sr_policy_index=0, table_id=0, sw_if_index=0) pol_steering.add_vpp_config() # log the sr steering policies self.logger.info(self.vapi.cli("show sr steering policies")) # create packets count = len(self.pg_packet_sizes) dst_inner = '7.1.1.123' pkts = [] # create IPv4 packets packet_header = self.create_packet_header_IPv4(dst_inner) # create traffic stream pg0->pg1 pkts.extend(self.create_stream(self.pg0, self.pg1, packet_header, self.pg_packet_sizes, count)) # send packets and verify received packets self.send_and_verify_pkts(self.pg0, pkts, self.pg1, self.compare_rx_tx_packet_T_Encaps_IPv4) # log the localsid counters self.logger.info(self.vapi.cli("show sr localsid")) # remove SR steering pol_steering.remove_vpp_config() self.logger.info(self.vapi.cli("show sr steering policies")) # remove SR Policies self.sr_policy.remove_vpp_config() self.logger.info(self.vapi.cli("show sr policies")) # remove FIB entries # done by tearDown # cleanup interfaces self.teardown_interfaces() @unittest.skip("VPP crashes after running this test") def test_SRv6_T_Encaps_L2(self): """ Test SRv6 Transit.Encaps behavior for L2. """ # send traffic to one destination interface # source interface is IPv4 only TODO? # destination interface is IPv6 only self.setup_interfaces(ipv6=[False, True], ipv4=[False, False]) # configure FIB entries route = VppIpRoute(self, "a4::", 64, [VppRoutePath(self.pg1.remote_ip6, self.pg1.sw_if_index)]) route.add_vpp_config() # configure encaps IPv6 source address # needs to be done before SR Policy config # TODO: API? self.vapi.cli("set sr encaps source addr a3::") bsid = 'a3::9999:1' # configure SRv6 Policy # Note: segment list order: first -> last sr_policy = VppSRv6Policy( self, bsid=bsid, is_encap=1, sr_type=SRv6PolicyType.SR_POLICY_TYPE_DEFAULT, weight=1, fib_table=0, segments=['a4::', 'a5::', 'a6::c7'], source='a3::') sr_policy.add_vpp_config() self.sr_policy = sr_policy # log the sr policies self.logger.info(self.vapi.cli("show sr policies")) # steer L2 traffic into SRv6 Policy # use the bsid of the above self.sr_policy pol_steering = VppSRv6Steering( self, bsid=self.sr_policy.bsid, prefix="::", mask_width=0, traffic_type=SRv6PolicySteeringTypes.SR_STEER_L2, sr_policy_index=0, table_id=0, sw_if_index=self.pg0.sw_if_index) pol_steering.add_vpp_config() # log the sr steering policies self.logger.info(self.vapi.cli("show sr steering policies")) # create packets count = len(self.pg_packet_sizes) pkts = [] # create L2 packets without dot1q header packet_header = self.create_packet_header_L2() # create traffic stream pg0->pg1 pkts.extend(self.create_stream(self.pg0, self.pg1, packet_header, self.pg_packet_sizes, count)) # create L2 packets with dot1q header packet_header = self.create_packet_header_L2(vlan=123) # create traffic stream pg0->pg1 pkts.extend(self.create_stream(self.pg0, self.pg1, packet_header, self.pg_packet_sizes, count)) # send packets and verify received packets self.send_and_verify_pkts(self.pg0, pkts, self.pg1, self.compare_rx_tx_packet_T_Encaps_L2) # log the localsid counters self.logger.info(self.vapi.cli("show sr localsid")) # remove SR steering pol_steering.remove_vpp_config() self.logger.info(self.vapi.cli("show sr steering policies")) # remove SR Policies self.sr_policy.remove_vpp_config() self.logger.info(self.vapi.cli("show sr policies")) # remove FIB entries # done by tearDown # cleanup interfaces self.teardown_interfaces() def test_SRv6_End(self): """ Test SRv6 End (without PSP) behavior. """ # send traffic to one destination interface # source and destination interfaces are IPv6 only self.setup_interfaces(ipv6=[True, True]) # configure FIB entries route = VppIpRoute(self, "a4::", 64, [VppRoutePath(self.pg1.remote_ip6, self.pg1.sw_if_index)]) route.add_vpp_config() # configure SRv6 localSID End without PSP behavior localsid = VppSRv6LocalSID( self, localsid='A3::0', behavior=SRv6LocalSIDBehaviors.SR_BEHAVIOR_END, nh_addr=0, end_psp=0, sw_if_index=0, vlan_index=0, fib_table=0) localsid.add_vpp_config() # log the localsids self.logger.debug(self.vapi.cli("show sr localsid")) # create IPv6 packets with SRH (SL=2, SL=1, SL=0) # send one packet per SL value per packet size # SL=0 packet with localSID End with USP needs 2nd SRH count = len(self.pg_packet_sizes) dst_inner = 'a4::1234' pkts = [] # packets with segments-left 2, active segment a3:: packet_header = self.create_packet_header_IPv6_SRH_IPv6( dst_inner, sidlist=['a5::', 'a4::', 'a3::'], segleft=2) # create traffic stream pg0->pg1 pkts.extend(self.create_stream(self.pg0, self.pg1, packet_header, self.pg_packet_sizes, count)) # packets with segments-left 1, active segment a3:: packet_header = self.create_packet_header_IPv6_SRH_IPv6( dst_inner, sidlist=['a4::', 'a3::', 'a2::'], segleft=1) # add to traffic stream pg0->pg1 pkts.extend(self.create_stream(self.pg0, self.pg1, packet_header, self.pg_packet_sizes, count)) # TODO: test behavior with SL=0 packet (needs 2*SRH?) expected_count = len(pkts) # packets without SRH (should not crash) packet_header = self.create_packet_header_IPv6('a3::') # create traffic stream pg0->pg1 pkts.extend(self.create_stream(self.pg0, self.pg1, packet_header, self.pg_packet_sizes, count)) # send packets and verify received packets self.send_and_verify_pkts(self.pg0, pkts, self.pg1, self.compare_rx_tx_packet_End, expected_count=expected_count) # log the localsid counters self.logger.info(self.vapi.cli("show sr localsid")) # remove SRv6 localSIDs localsid.remove_vpp_config() # remove FIB entries # done by tearDown # cleanup interfaces self.teardown_interfaces() def test_SRv6_End_with_PSP(self): """ Test SRv6 End with PSP behavior. """ # send traffic to one destination interface # source and destination interfaces are IPv6 only self.setup_interfaces(ipv6=[True, True]) # configure FIB entries route = VppIpRoute(self, "a4::", 64, [VppRoutePath(self.pg1.remote_ip6, self.pg1.sw_if_index)]) route.add_vpp_config() # configure SRv6 localSID End with PSP behavior localsid = VppSRv6LocalSID( self, localsid='A3::0', behavior=SRv6LocalSIDBehaviors.SR_BEHAVIOR_END, nh_addr=0, end_psp=1, sw_if_index=0, vlan_index=0, fib_table=0) localsid.add_vpp_config() # log the localsids self.logger.debug(self.vapi.cli("show sr localsid")) # create IPv6 packets with SRH (SL=2, SL=1) # send one packet per SL value per packet size # SL=0 packet with localSID End with PSP is dropped count = len(self.pg_packet_sizes) dst_inner = 'a4::1234' pkts = [] # packets with segments-left 2, active segment a3:: packet_header = self.create_packet_header_IPv6_SRH_IPv6( dst_inner, sidlist=['a5::', 'a4::', 'a3::'], segleft=2) # create traffic stream pg0->pg1 pkts.extend(self.create_stream(self.pg0, self.pg1, packet_header, self.pg_packet_sizes, count)) # packets with segments-left 1, active segment a3:: packet_header = self.create_packet_header_IPv6_SRH_IPv6( dst_inner, sidlist=['a4::', 'a3::', 'a2::'], segleft=1) # add to traffic stream pg0->pg1 pkts.extend(self.create_stream(self.pg0, self.pg1, packet_header, self.pg_packet_sizes, count)) # send packets and verify received packets self.send_and_verify_pkts(self.pg0, pkts, self.pg1, self.compare_rx_tx_packet_End_PSP) # log the localsid counters self.logger.info(self.vapi.cli("show sr localsid")) # remove SRv6 localSIDs localsid.remove_vpp_config() # remove FIB entries # done by tearDown # cleanup interfaces self.teardown_interfaces() def test_SRv6_End_X(self): """ Test SRv6 End.X (without PSP) behavior. """ # create three interfaces (1 source, 2 destinations) # source and destination interfaces are IPv6 only self.setup_interfaces(ipv6=[True, True, True]) # configure FIB entries # a4::/64 via pg1 and pg2 route = VppIpRoute(self, "a4::", 64, [VppRoutePath(self.pg1.remote_ip6, self.pg1.sw_if_index), VppRoutePath(self.pg2.remote_ip6, self.pg2.sw_if_index)]) route.add_vpp_config() self.logger.debug(self.vapi.cli("show ip6 fib")) # configure SRv6 localSID End.X without PSP behavior # End.X points to interface pg1 localsid = VppSRv6LocalSID( self, localsid='A3::C4', behavior=SRv6LocalSIDBehaviors.SR_BEHAVIOR_X, nh_addr=self.pg1.remote_ip6, end_psp=0, sw_if_index=self.pg1.sw_if_index, vlan_index=0, fib_table=0) localsid.add_vpp_config() # log the localsids self.logger.debug(self.vapi.cli("show sr localsid")) # create IPv6 packets with SRH (SL=2, SL=1) # send one packet per SL value per packet size # SL=0 packet with localSID End with PSP is dropped count = len(self.pg_packet_sizes) dst_inner = 'a4::1234' pkts = [] # packets with segments-left 2, active segment a3::c4 packet_header = self.create_packet_header_IPv6_SRH_IPv6( dst_inner, sidlist=['a5::', 'a4::', 'a3::c4'], segleft=2) # create traffic stream pg0->pg1 pkts.extend(self.create_stream(self.pg0, self.pg1, packet_header, self.pg_packet_sizes, count)) # packets with segments-left 1, active segment a3::c4 packet_header = self.create_packet_header_IPv6_SRH_IPv6( dst_inner, sidlist=['a4::', 'a3::c4', 'a2::'], segleft=1) # add to traffic stream pg0->pg1 pkts.extend(self.create_stream(self.pg0, self.pg1, packet_header, self.pg_packet_sizes, count)) # send packets and verify received packets # using same comparison function as End (no PSP) self.send_and_verify_pkts(self.pg0, pkts, self.pg1, self.compare_rx_tx_packet_End) # assert nothing was received on the other interface (pg2) self.pg2.assert_nothing_captured("mis-directed packet(s)") # log the localsid counters self.logger.info(self.vapi.cli("show sr localsid")) # remove SRv6 localSIDs localsid.remove_vpp_config() # remove FIB entries # done by tearDown # cleanup interfaces self.teardown_interfaces() def test_SRv6_End_X_with_PSP(self): """ Test SRv6 End.X with PSP behavior. """ # create three interfaces (1 source, 2 destinations) # source and destination interfaces are IPv6 only self.setup_interfaces(ipv6=[True, True, True]) # configure FIB entries # a4::/64 via pg1 and pg2 route = VppIpRoute(self, "a4::", 64, [VppRoutePath( self.pg1.remote_ip6, self.pg1.sw_if_index), VppRoutePath(self.pg2.remote_ip6, self.pg2.sw_if_index)]) route.add_vpp_config() # configure SRv6 localSID End with PSP behavior localsid = VppSRv6LocalSID( self, localsid='A3::C4', behavior=SRv6LocalSIDBehaviors.SR_BEHAVIOR_X, nh_addr=self.pg1.remote_ip6, end_psp=1, sw_if_index=self.pg1.sw_if_index, vlan_index=0, fib_table=0) localsid.add_vpp_config() # log the localsids self.logger.debug(self.vapi.cli("show sr localsid")) # create IPv6 packets with SRH (SL=2, SL=1) # send one packet per SL value per packet size # SL=0 packet with localSID End with PSP is dropped count = len(self.pg_packet_sizes) dst_inner = 'a4::1234' pkts = [] # packets with segments-left 2, active segment a3:: packet_header = self.create_packet_header_IPv6_SRH_IPv6( dst_inner, sidlist=['a5::', 'a4::', 'a3::c4'], segleft=2) # create traffic stream pg0->pg1 pkts.extend(self.create_stream(self.pg0, self.pg1, packet_header, self.pg_packet_sizes, count)) # packets with segments-left 1, active segment a3:: packet_header = self.create_packet_header_IPv6_SRH_IPv6( dst_inner, sidlist=['a4::', 'a3::c4', 'a2::'], segleft=1) # add to traffic stream pg0->pg1 pkts.extend(self.create_stream(self.pg0, self.pg1, packet_header, self.pg_packet_sizes, count)) # send packets and verify received packets # using same comparison function as End with PSP self.send_and_verify_pkts(self.pg0, pkts, self.pg1, self.compare_rx_tx_packet_End_PSP) # assert nothing was received on the other interface (pg2) self.pg2.assert_nothing_captured("mis-directed packet(s)") # log the localsid counters self.logger.info(self.vapi.cli("show sr localsid")) # remove SRv6 localSIDs localsid.remove_vpp_config() # remove FIB entries # done by tearDown # cleanup interfaces self.teardown_interfaces() def test_SRv6_End_DX6(self): """ Test SRv6 End.DX6 behavior. """ # send traffic to one destination interface # source and destination interfaces are IPv6 only self.setup_interfaces(ipv6=[True, True]) # configure SRv6 localSID End.DX6 behavior localsid = VppSRv6LocalSID( self, localsid='A3::C4', behavior=SRv6LocalSIDBehaviors.SR_BEHAVIOR_DX6, nh_addr=self.pg1.remote_ip6, end_psp=0, sw_if_index=self.pg1.sw_if_index, vlan_index=0, fib_table=0) localsid.add_vpp_config() # log the localsids self.logger.debug(self.vapi.cli("show sr localsid")) # create IPv6 packets with SRH (SL=0) # send one packet per packet size count = len(self.pg_packet_sizes) dst_inner = 'a4::1234' # inner header destination address pkts = [] # packets with SRH, segments-left 0, active segment a3::c4 packet_header = self.create_packet_header_IPv6_SRH_IPv6( dst_inner, sidlist=['a3::c4', 'a2::', 'a1::'], segleft=0) # add to traffic stream pg0->pg1 pkts.extend(self.create_stream(self.pg0, self.pg1, packet_header, self.pg_packet_sizes, count)) # packets without SRH, IPv6 in IPv6 # outer IPv6 dest addr is the localsid End.DX6 packet_header = self.create_packet_header_IPv6_IPv6( dst_inner, dst_outer='a3::c4') # add to traffic stream pg0->pg1 pkts.extend(self.create_stream(self.pg0, self.pg1, packet_header, self.pg_packet_sizes, count)) # send packets and verify received packets self.send_and_verify_pkts(self.pg0, pkts, self.pg1, self.compare_rx_tx_packet_End_DX6) # log the localsid counters self.logger.info(self.vapi.cli("show sr localsid")) # remove SRv6 localSIDs localsid.remove_vpp_config() # cleanup interfaces self.teardown_interfaces() def test_SRv6_End_DT6(self): """ Test SRv6 End.DT6 behavior. """ # create three interfaces (1 source, 2 destinations) # all interfaces are IPv6 only # source interface in global FIB (0) # destination interfaces in global and vrf vrf_1 = 1 ipt = VppIpTable(self, vrf_1, is_ip6=True) ipt.add_vpp_config() self.setup_interfaces(ipv6=[True, True, True], ipv6_table_id=[0, 0, vrf_1]) # configure FIB entries # a4::/64 is reachable # via pg1 in table 0 (global) # and via pg2 in table vrf_1 route0 = VppIpRoute(self, "a4::", 64, [VppRoutePath(self.pg1.remote_ip6, self.pg1.sw_if_index, nh_table_id=0)], table_id=0) route0.add_vpp_config() route1 = VppIpRoute(self, "a4::", 64, [VppRoutePath(self.pg2.remote_ip6, self.pg2.sw_if_index, nh_table_id=vrf_1)], table_id=vrf_1) route1.add_vpp_config() self.logger.debug(self.vapi.cli("show ip6 fib")) # configure SRv6 localSID End.DT6 behavior # Note: # fib_table: where the localsid is installed # sw_if_index: in T-variants of localsid this is the vrf table_id localsid = VppSRv6LocalSID( self, localsid='A3::C4', behavior=SRv6LocalSIDBehaviors.SR_BEHAVIOR_DT6, nh_addr=0, end_psp=0, sw_if_index=vrf_1, vlan_index=0, fib_table=0) localsid.add_vpp_config() # log the localsids self.logger.debug(self.vapi.cli("show sr localsid")) # create IPv6 packets with SRH (SL=0) # send one packet per packet size count = len(self.pg_packet_sizes) dst_inner = 'a4::1234' # inner header destination address pkts = [] # packets with SRH, segments-left 0, active segment a3::c4 packet_header = self.create_packet_header_IPv6_SRH_IPv6( dst_inner, sidlist=['a3::c4', 'a2::', 'a1::'], segleft=0) # add to traffic stream pg0->pg1 pkts.extend(self.create_stream(self.pg0, self.pg2, packet_header, self.pg_packet_sizes, count)) # packets without SRH, IPv6 in IPv6 # outer IPv6 dest addr is the localsid End.DT6 packet_header = self.create_packet_header_IPv6_IPv6( dst_inner, dst_outer='a3::c4') # add to traffic stream pg0->pg1 pkts.extend(self.create_stream(self.pg0, self.pg2, packet_header, self.pg_packet_sizes, count)) # send packets and verify received packets # using same comparison function as End.DX6 self.send_and_verify_pkts(self.pg0, pkts, self.pg2, self.compare_rx_tx_packet_End_DX6) # assert nothing was received on the other interface (pg2) self.pg1.assert_nothing_captured("mis-directed packet(s)") # log the localsid counters self.logger.info(self.vapi.cli("show sr localsid")) # remove SRv6 localSIDs localsid.remove_vpp_config() # remove FIB entries # done by tearDown # cleanup interfaces self.teardown_interfaces() def test_SRv6_End_DX4(self): """ Test SRv6 End.DX4 behavior. """ # send traffic to one destination interface # source interface is IPv6 only # destination interface is IPv4 only self.setup_interfaces(ipv6=[True, False], ipv4=[False, True]) # configure SRv6 localSID End.DX4 behavior localsid = VppSRv6LocalSID( self, localsid='A3::C4', behavior=SRv6LocalSIDBehaviors.SR_BEHAVIOR_DX4, nh_addr=self.pg1.remote_ip4, end_psp=0, sw_if_index=self.pg1.sw_if_index, vlan_index=0, fib_table=0) localsid.add_vpp_config() # log the localsids self.logger.debug(self.vapi.cli("show sr localsid")) # send one packet per packet size count = len(self.pg_packet_sizes) dst_inner = '4.1.1.123' # inner header destination address pkts = [] # packets with SRH, segments-left 0, active segment a3::c4 packet_header = self.create_packet_header_IPv6_SRH_IPv4( dst_inner, sidlist=['a3::c4', 'a2::', 'a1::'], segleft=0) # add to traffic stream pg0->pg1 pkts.extend(self.create_stream(self.pg0, self.pg1, packet_header, self.pg_packet_sizes, count)) # packets without SRH, IPv4 in IPv6 # outer IPv6 dest addr is the localsid End.DX4 packet_header = self.create_packet_header_IPv6_IPv4( dst_inner, dst_outer='a3::c4') # add to traffic stream pg0->pg1 pkts.extend(self.create_stream(self.pg0, self.pg1, packet_header, self.pg_packet_sizes, count)) # send packets and verify received packets self.send_and_verify_pkts(self.pg0, pkts, self.pg1, self.compare_rx_tx_packet_End_DX4) # log the localsid counters self.logger.info(self.vapi.cli("show sr localsid")) # remove SRv6 localSIDs localsid.remove_vpp_config() # cleanup interfaces self.teardown_interfaces() def test_SRv6_End_DT4(self): """ Test SRv6 End.DT4 behavior. """ # create three interfaces (1 source, 2 destinations) # source interface is IPv6-only # destination interfaces are IPv4 only # source interface in global FIB (0) # destination interfaces in global and vrf vrf_1 = 1 ipt = VppIpTable(self, vrf_1) ipt.add_vpp_config() self.setup_interfaces(ipv6=[True, False, False], ipv4=[False, True, True], ipv6_table_id=[0, 0, 0], ipv4_table_id=[0, 0, vrf_1]) # configure FIB entries # 4.1.1.0/24 is reachable # via pg1 in table 0 (global) # and via pg2 in table vrf_1 route0 = VppIpRoute(self, "4.1.1.0", 24, [VppRoutePath(self.pg1.remote_ip4, self.pg1.sw_if_index, nh_table_id=0)], table_id=0) route0.add_vpp_config() route1 = VppIpRoute(self, "4.1.1.0", 24, [VppRoutePath(self.pg2.remote_ip4, self.pg2.sw_if_index, nh_table_id=vrf_1)], table_id=vrf_1) route1.add_vpp_config() self.logger.debug(self.vapi.cli("show ip fib")) # configure SRv6 localSID End.DT6 behavior # Note: # fib_table: where the localsid is installed # sw_if_index: in T-variants of localsid: vrf table_id localsid = VppSRv6LocalSID( self, localsid='A3::C4', behavior=SRv6LocalSIDBehaviors.SR_BEHAVIOR_DT4, nh_addr=0, end_psp=0, sw_if_index=vrf_1, vlan_index=0, fib_table=0) localsid.add_vpp_config() # log the localsids self.logger.debug(self.vapi.cli("show sr localsid")) # create IPv6 packets with SRH (SL=0) # send one packet per packet size count = len(self.pg_packet_sizes) dst_inner = '4.1.1.123' # inner header destination address pkts = [] # packets with SRH, segments-left 0, active segment a3::c4 packet_header = self.create_packet_header_IPv6_SRH_IPv4( dst_inner, sidlist=['a3::c4', 'a2::', 'a1::'], segleft=0) # add to traffic stream pg0->pg1 pkts.extend(self.create_stream(self.pg0, self.pg2, packet_header, self.pg_packet_sizes, count)) # packets without SRH, IPv6 in IPv6 # outer IPv6 dest addr is the localsid End.DX4 packet_header = self.create_packet_header_IPv6_IPv4( dst_inner, dst_outer='a3::c4') # add to traffic stream pg0->pg1 pkts.extend(self.create_stream(self.pg0, self.pg2, packet_header, self.pg_packet_sizes, count)) # send packets and verify received packets # using same comparison function as End.DX4 self.send_and_verify_pkts(self.pg0, pkts, self.pg2, self.compare_rx_tx_packet_End_DX4) # assert nothing was received on the other interface (pg2) self.pg1.assert_nothing_captured("mis-directed packet(s)") # log the localsid counters self.logger.info(self.vapi.cli("show sr localsid")) # remove SRv6 localSIDs localsid.remove_vpp_config() # remove FIB entries # done by tearDown # cleanup interfaces self.teardown_interfaces() def test_SRv6_End_DX2(self): """ Test SRv6 End.DX2 behavior. """ # send traffic to one destination interface # source interface is IPv6 only self.setup_interfaces(ipv6=[True, False], ipv4=[False, False]) # configure SRv6 localSID End.DX2 behavior localsid = VppSRv6LocalSID( self, localsid='A3::C4', behavior=SRv6LocalSIDBehaviors.SR_BEHAVIOR_DX2, nh_addr=0, end_psp=0, sw_if_index=self.pg1.sw_if_index, vlan_index=0, fib_table=0) localsid.add_vpp_config() # log the localsids self.logger.debug(self.vapi.cli("show sr localsid")) # send one packet per packet size count = len(self.pg_packet_sizes) pkts = [] # packets with SRH, segments-left 0, active segment a3::c4 # L2 has no dot1q header packet_header = self.create_packet_header_IPv6_SRH_L2( sidlist=['a3::c4', 'a2::', 'a1::'], segleft=0, vlan=0) # add to traffic stream pg0->pg1 pkts.extend(self.create_stream(self.pg0, self.pg1, packet_header, self.pg_packet_sizes, count)) # packets with SRH, segments-left 0, active segment a3::c4 # L2 has dot1q header packet_header = self.create_packet_header_IPv6_SRH_L2( sidlist=['a3::c4', 'a2::', 'a1::'], segleft=0, vlan=123) # add to traffic stream pg0->pg1 pkts.extend(self.create_stream(self.pg0, self.pg1, packet_header, self.pg_packet_sizes, count)) # packets without SRH, L2 in IPv6 # outer IPv6 dest addr is the localsid End.DX2 # L2 has no dot1q header packet_header = self.create_packet_header_IPv6_L2( dst_outer='a3::c4', vlan=0) # add to traffic stream pg0->pg1 pkts.extend(self.create_stream(self.pg0, self.pg1, packet_header, self.pg_packet_sizes, count)) # packets without SRH, L2 in IPv6 # outer IPv6 dest addr is the localsid End.DX2 # L2 has dot1q header packet_header = self.create_packet_header_IPv6_L2( dst_outer='a3::c4', vlan=123) # add to traffic stream pg0->pg1 pkts.extend(self.create_stream(self.pg0, self.pg1, packet_header, self.pg_packet_sizes, count)) # send packets and verify received packets self.send_and_verify_pkts(self.pg0, pkts, self.pg1, self.compare_rx_tx_packet_End_DX2) # log the localsid counters self.logger.info(self.vapi.cli("show sr localsid")) # remove SRv6 localSIDs localsid.remove_vpp_config() # cleanup interfaces self.teardown_interfaces() @unittest.skipUnless(0, "PC to fix") def test_SRv6_T_Insert_Classifier(self): """ Test SRv6 Transit.Insert behavior (IPv6 only). steer packets using the classifier """ # send traffic to one destination interface # source and destination are IPv6 only self.setup_interfaces(ipv6=[False, False, False, True, True]) # configure FIB entries route = VppIpRoute(self, "a4::", 64, [VppRoutePath( self.pg4.remote_ip6, self.pg4.sw_if_index)]) route.add_vpp_config() # configure encaps IPv6 source address # needs to be done before SR Policy config # TODO: API? self.vapi.cli("set sr encaps source addr a3::") bsid = 'a3::9999:1' # configure SRv6 Policy # Note: segment list order: first -> last sr_policy = VppSRv6Policy( self, bsid=bsid, is_encap=0, sr_type=SRv6PolicyType.SR_POLICY_TYPE_DEFAULT, weight=1, fib_table=0, segments=['a4::', 'a5::', 'a6::c7'], source='a3::') sr_policy.add_vpp_config() self.sr_policy = sr_policy # log the sr policies self.logger.info(self.vapi.cli("show sr policies")) # add classify table # mask on dst ip address prefix a7::/8 mask = '{!s:0<16}'.format('ff') r = self.vapi.classify_add_del_table( 1, binascii.unhexlify(mask), match_n_vectors=(len(mask) - 1) // 32 + 1, current_data_flag=1, skip_n_vectors=2) # data offset self.assertIsNotNone(r, 'No response msg for add_del_table') table_index = r.new_table_index # add the source routing node as a ip6 inacl netxt node r = self.vapi.add_node_next('ip6-inacl', 'sr-pl-rewrite-insert') inacl_next_node_index = r.node_index match = '{!s:0<16}'.format('a7') r = self.vapi.classify_add_del_session( 1, table_index, binascii.unhexlify(match), hit_next_index=inacl_next_node_index, action=3, metadata=0) # sr policy index self.assertIsNotNone(r, 'No response msg for add_del_session') # log the classify table used in the steering policy self.logger.info(self.vapi.cli("show classify table")) r } 
#!/usr/bin/env python
import random
import socket
import unittest
import time
import re

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

from framework import VppTestCase, VppTestRunner, running_extended_tests
from vpp_object import VppObject
from vpp_pg_interface import CaptureTimeoutError
from util import ppp
from ipfix import IPFIX, Set, Template, Data, IPFIXDecoder
from vpp_ip_route import VppIpRoute, VppRoutePath


class VppCFLOW(VppObject):
    """CFLOW object for IPFIX exporter and Flowprobe feature"""

    def __init__(self, test, intf='pg2', active=0, passive=0, timeout=100,
                 mtu=1024, datapath='l2', layer='l2 l3 l4'):
        self._test = test
        self._intf = intf
        self._active = active
        if passive == 0 or passive < active:
            self._passive = active+1
        else:
            self._passive = passive
        self._datapath = datapath           # l2 ip4 ip6
        self._collect = layer               # l2 l3 l4
        self._timeout = timeout
        self._mtu = mtu
        self._configured = False

    def add_vpp_config(self):
        self.enable_exporter()
        self._test.vapi.ppcli("flowprobe params record %s active %s "
                              "passive %s" % (self._collect, self._active,
                                              self._passive))
        self.enable_flowprobe_feature()
        self._test.vapi.cli("ipfix flush")
        self._configured = True

    def remove_vpp_config(self):
        self.disable_exporter()
        self.disable_flowprobe_feature()
        self._test.vapi.cli("ipfix flush")
        self._configured = False

    def enable_exporter(self):
        self._test.vapi.set_ipfix_exporter(
            collector_address=self._test.pg0.remote_ip4n,
            src_address=self._test.pg0.local_ip4n,
            path_mtu=self._mtu,
            template_interval=self._timeout)

    def enable_flowprobe_feature(self):
        self._test.vapi.ppcli("flowprobe feature add-del %s %s" %
                              (self._intf, self._datapath))

    def disable_exporter(self):
        self._test.vapi.cli("set ipfix exporter collector 0.0.0.0")

    def disable_flowprobe_feature(self):
        self._test.vapi.cli("flowprobe feature add-del %s %s disable" %
                            (self._intf, self._datapath))

    def object_id(self):
        return "ipfix-collector-%s" % (self._src, self.dst)

    def query_vpp_config(self):
        return self._configured

    def verify_templates(self, decoder=None, timeout=1, count=3):
        templates = []
        p = self._test.wait_for_cflow_packet(self._test.collector, 2, timeout)
        self._test.assertTrue(p.haslayer(IPFIX))
        if decoder is not None and p.haslayer(Template):
            templates.append(p[Template].templateID)
            decoder.add_template(p.getlayer(Template))
        if count > 1:
            p = self._test.wait_for_cflow_packet(self._test.collector, 2)
            self._test.assertTrue(p.haslayer(IPFIX))
            if decoder is not None and p.haslayer(Template):
                templates.append(p[Template].templateID)
                decoder.add_template(p.getlayer(Template))
        if count > 2:
            p = self._test.wait_for_cflow_packet(self._test.collector, 2)
            self._test.assertTrue(p.haslayer(IPFIX))
            if decoder is not None and p.haslayer(Template):
                templates.append(p[Template].templateID)
                decoder.add_template(p.getlayer(Template))
        return templates


class MethodHolder(VppTestCase):
    """ Flow-per-packet plugin: test L2, IP4, IP6 reporting """

    # Test variables
    debug_print = False
    max_number_of_packets = 10
    pkts = []

    @classmethod
    def setUpClass(cls):
        """
        Perform standard class setup (defined by class method setUpClass in
        class VppTestCase) before running the test case, set test case related
        variables and configure VPP.
        """
        super(MethodHolder, cls).setUpClass()
        try:
            # Create pg interfaces
            cls.create_pg_interfaces(range(9))

            # Packet sizes
            cls.pg_if_packet_sizes = [64, 512, 1518, 9018]

            # Create BD with MAC learning and unknown unicast flooding disabled
            # and put interfaces to this BD
            cls.vapi.bridge_domain_add_del(bd_id=1, uu_flood=1, learn=1)
            cls.vapi.sw_interface_set_l2_bridge(cls.pg1._sw_if_index, bd_id=1)
            cls.vapi.sw_interface_set_l2_bridge(cls.pg2._sw_if_index, bd_id=1)

            # Set up all interfaces
            for i in cls.pg_interfaces:
                i.admin_up()

            cls.pg0.config_ip4()
            cls.pg0.configure_ipv4_neighbors()
            cls.collector = cls.pg0

            cls.pg1.config_ip4()
            cls.pg1.resolve_arp()
            cls.pg2.config_ip4()
            cls.pg2.resolve_arp()
            cls.pg3.config_ip4()
            cls.pg3.resolve_arp()
            cls.pg4.config_ip4()
            cls.pg4.resolve_arp()
            cls.pg7.config_ip4()
            cls.pg8.config_ip4()
            cls.pg8.configure_ipv4_neighbors()

            cls.pg5.config_ip6()
            cls.pg5.resolve_ndp()
            cls.pg5.disable_ipv6_ra()
            cls.pg6.config_ip6()
            cls.pg6.resolve_ndp()
            cls.pg6.disable_ipv6_ra()
        except Exception:
            super(MethodHolder, cls).tearDownClass()
            raise

    def create_stream(self, src_if=None, dst_if=None, packets=None,
                      size=None, ip_ver='v4'):
        """Create a packet stream to tickle the plugin

        :param VppInterface src_if: Source interface for packet stream
        :param VppInterface src_if: Dst interface for packet stream
        """
        if src_if is None:
            src_if = self.pg1
        if dst_if is None:
            dst_if = self.pg2
        self.pkts = []
        if packets is None:
            packets = random.randint(1, self.max_number_of_packets)
        pkt_size = size
        for p in range(0, packets):
            if size is None:
                pkt_size = random.choice(self.pg_if_packet_sizes)
            info = self.create_packet_info(src_if, dst_if)
            payload = self.info_to_payload(info)
            p = Ether(src=src_if.remote_mac, dst=src_if.local_mac)
            if ip_ver == 'v4':
                p /= IP(src=src_if.remote_ip4, dst=dst_if.remote_ip4)
            else:
                p /= IPv6(src=src_if.remote_ip6, dst=dst_if.remote_ip6)
            p /= UDP(sport=1234, dport=4321)
            p /= Raw(payload)
            info.data = p.copy()
            self.extend_packet(p, pkt_size)
            self.pkts.append(p)

    def verify_cflow_data(self, decoder, capture, cflow):
        octets = 0
        packets = 0
        for p in capture:
            octets += p[IP].len
            packets += 1
        if cflow.haslayer(Data):
            data = decoder.decode_data_set(cflow.getlayer(Set))
            for record in data:
                self.assertEqual(int(record[1].encode('hex'), 16), octets)
                self.assertEqual(int(record[2].encode('hex'), 16), packets)

    def send_packets(self, src_if=None, dst_if=None):
        if src_if