+class TestIP6Punt(VppTestCase):
+ """ IPv6 Punt Police/Redirect """
+
+ @classmethod
+ def setUpClass(cls):
+ super(TestIP6Punt, cls).setUpClass()
+
+ @classmethod
+ def tearDownClass(cls):
+ super(TestIP6Punt, cls).tearDownClass()
+
+ def setUp(self):
+ super(TestIP6Punt, self).setUp()
+
+ self.create_pg_interfaces(range(4))
+
+ for i in self.pg_interfaces:
+ i.admin_up()
+ i.config_ip6()
+ i.resolve_ndp()
+
+ def tearDown(self):
+ super(TestIP6Punt, self).tearDown()
+ for i in self.pg_interfaces:
+ i.unconfig_ip6()
+ i.admin_down()
+
+ def test_ip_punt(self):
+ """ IP6 punt police and redirect """
+
+ p = (Ether(src=self.pg0.remote_mac,
+ dst=self.pg0.local_mac) /
+ IPv6(src=self.pg0.remote_ip6,
+ dst=self.pg0.local_ip6) /
+ inet6.TCP(sport=1234, dport=1234) /
+ Raw(b'\xa5' * 100))
+
+ pkts = p * 1025
+
+ #
+ # Configure a punt redirect via pg1.
+ #
+ nh_addr = self.pg1.remote_ip6
+ self.vapi.ip_punt_redirect(self.pg0.sw_if_index,
+ self.pg1.sw_if_index,
+ nh_addr)
+
+ self.send_and_expect(self.pg0, pkts, self.pg1)
+
+ #
+ # add a policer
+ #
+ policer = VppPolicer(self, "ip6-punt", 400, 0, 10, 0, rate_type=1)
+ policer.add_vpp_config()
+ self.vapi.ip_punt_police(policer.policer_index, is_ip6=1)
+
+ self.vapi.cli("clear trace")
+ self.pg0.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+
+ #
+ # the number of packet received should be greater than 0,
+ # but not equal to the number sent, since some were policed
+ #
+ rx = self.pg1._get_capture(1)
+ self.assertGreater(len(rx), 0)
+ self.assertLess(len(rx), len(pkts))
+
+ #
+ # remove the policer. back to full rx
+ #
+ self.vapi.ip_punt_police(policer.policer_index, is_add=0, is_ip6=1)
+ policer.remove_vpp_config()
+ self.send_and_expect(self.pg0, pkts, self.pg1)
+
+ #
+ # remove the redirect. expect full drop.
+ #
+ self.vapi.ip_punt_redirect(self.pg0.sw_if_index,
+ self.pg1.sw_if_index,
+ nh_addr,
+ is_add=0)
+ self.send_and_assert_no_replies(self.pg0, pkts,
+ "IP no punt config")
+
+ #
+ # Add a redirect that is not input port selective
+ #
+ self.vapi.ip_punt_redirect(0xffffffff,
+ self.pg1.sw_if_index,
+ nh_addr)
+ self.send_and_expect(self.pg0, pkts, self.pg1)
+
+ self.vapi.ip_punt_redirect(0xffffffff,
+ self.pg1.sw_if_index,
+ nh_addr,
+ is_add=0)
+
+ def test_ip_punt_dump(self):
+ """ IP6 punt redirect dump"""
+
+ #
+ # Configure a punt redirects
+ #
+ nh_addr = self.pg3.remote_ip6
+ self.vapi.ip_punt_redirect(self.pg0.sw_if_index,
+ self.pg3.sw_if_index,
+ nh_addr)
+ self.vapi.ip_punt_redirect(self.pg1.sw_if_index,
+ self.pg3.sw_if_index,
+ nh_addr)
+ self.vapi.ip_punt_redirect(self.pg2.sw_if_index,
+ self.pg3.sw_if_index,
+ '0::0')
+
+ #
+ # Dump pg0 punt redirects
+ #
+ punts = self.vapi.ip_punt_redirect_dump(self.pg0.sw_if_index,
+ is_ipv6=1)
+ for p in punts:
+ self.assertEqual(p.punt.rx_sw_if_index, self.pg0.sw_if_index)
+
+ #
+ # Dump punt redirects for all interfaces
+ #
+ punts = self.vapi.ip_punt_redirect_dump(0xffffffff, is_ipv6=1)
+ self.assertEqual(len(punts), 3)
+ for p in punts:
+ self.assertEqual(p.punt.tx_sw_if_index, self.pg3.sw_if_index)
+ self.assertNotEqual(punts[1].punt.nh, self.pg3.remote_ip6)
+ self.assertEqual(str(punts[2].punt.nh), '::')
+
+
+class TestIPDeag(VppTestCase):
+ """ IPv6 Deaggregate Routes """
+
+ @classmethod
+ def setUpClass(cls):
+ super(TestIPDeag, cls).setUpClass()
+
+ @classmethod
+ def tearDownClass(cls):
+ super(TestIPDeag, cls).tearDownClass()
+
+ def setUp(self):
+ super(TestIPDeag, self).setUp()
+
+ self.create_pg_interfaces(range(3))
+
+ for i in self.pg_interfaces:
+ i.admin_up()
+ i.config_ip6()
+ i.resolve_ndp()
+
+ def tearDown(self):
+ super(TestIPDeag, self).tearDown()
+ for i in self.pg_interfaces:
+ i.unconfig_ip6()
+ i.admin_down()
+
+ def test_ip_deag(self):
+ """ IP Deag Routes """
+
+ #
+ # Create a table to be used for:
+ # 1 - another destination address lookup
+ # 2 - a source address lookup
+ #
+ table_dst = VppIpTable(self, 1, is_ip6=1)
+ table_src = VppIpTable(self, 2, is_ip6=1)
+ table_dst.add_vpp_config()
+ table_src.add_vpp_config()
+
+ #
+ # Add a route in the default table to point to a deag/
+ # second lookup in each of these tables
+ #
+ route_to_dst = VppIpRoute(self, "1::1", 128,
+ [VppRoutePath("::",
+ 0xffffffff,
+ nh_table_id=1)])
+ route_to_src = VppIpRoute(
+ self, "1::2", 128,
+ [VppRoutePath("::",
+ 0xffffffff,
+ nh_table_id=2,
+ type=FibPathType.FIB_PATH_TYPE_SOURCE_LOOKUP)])
+
+ route_to_dst.add_vpp_config()
+ route_to_src.add_vpp_config()
+
+ #
+ # packets to these destination are dropped, since they'll
+ # hit the respective default routes in the second table
+ #
+ p_dst = (Ether(src=self.pg0.remote_mac,
+ dst=self.pg0.local_mac) /
+ IPv6(src="5::5", dst="1::1") /
+ inet6.TCP(sport=1234, dport=1234) /
+ Raw(b'\xa5' * 100))
+ p_src = (Ether(src=self.pg0.remote_mac,
+ dst=self.pg0.local_mac) /
+ IPv6(src="2::2", dst="1::2") /
+ inet6.TCP(sport=1234, dport=1234) /
+ Raw(b'\xa5' * 100))
+ pkts_dst = p_dst * 257
+ pkts_src = p_src * 257
+
+ self.send_and_assert_no_replies(self.pg0, pkts_dst,
+ "IP in dst table")
+ self.send_and_assert_no_replies(self.pg0, pkts_src,
+ "IP in src table")
+
+ #
+ # add a route in the dst table to forward via pg1
+ #
+ route_in_dst = VppIpRoute(self, "1::1", 128,
+ [VppRoutePath(self.pg1.remote_ip6,
+ self.pg1.sw_if_index)],
+ table_id=1)
+ route_in_dst.add_vpp_config()
+
+ self.send_and_expect(self.pg0, pkts_dst, self.pg1)
+
+ #
+ # add a route in the src table to forward via pg2
+ #
+ route_in_src = VppIpRoute(self, "2::2", 128,
+ [VppRoutePath(self.pg2.remote_ip6,
+ self.pg2.sw_if_index)],
+ table_id=2)
+ route_in_src.add_vpp_config()
+ self.send_and_expect(self.pg0, pkts_src, self.pg2)
+
+ #
+ # loop in the lookup DP
+ #
+ route_loop = VppIpRoute(self, "3::3", 128,
+ [VppRoutePath("::",
+ 0xffffffff)])
+ route_loop.add_vpp_config()
+
+ p_l = (Ether(src=self.pg0.remote_mac,
+ dst=self.pg0.local_mac) /
+ IPv6(src="3::4", dst="3::3") /
+ inet6.TCP(sport=1234, dport=1234) /
+ Raw(b'\xa5' * 100))
+
+ self.send_and_assert_no_replies(self.pg0, p_l * 257,
+ "IP lookup loop")
+
+
+class TestIP6Input(VppTestCase):
+ """ IPv6 Input Exception Test Cases """
+
+ @classmethod
+ def setUpClass(cls):
+ super(TestIP6Input, cls).setUpClass()
+
+ @classmethod
+ def tearDownClass(cls):
+ super(TestIP6Input, cls).tearDownClass()
+
+ def setUp(self):
+ super(TestIP6Input, self).setUp()
+
+ self.create_pg_interfaces(range(2))
+
+ for i in self.pg_interfaces:
+ i.admin_up()
+ i.config_ip6()
+ i.resolve_ndp()
+
+ def tearDown(self):
+ super(TestIP6Input, self).tearDown()
+ for i in self.pg_interfaces:
+ i.unconfig_ip6()
+ i.admin_down()
+
+ def test_ip_input_icmp_reply(self):
+ """ IP6 Input Exception - Return ICMP (3,0) """
+ #
+ # hop limit - ICMP replies
+ #
+ p_version = (Ether(src=self.pg0.remote_mac,
+ dst=self.pg0.local_mac) /
+ IPv6(src=self.pg0.remote_ip6,
+ dst=self.pg1.remote_ip6,
+ hlim=1) /
+ inet6.UDP(sport=1234, dport=1234) /
+ Raw(b'\xa5' * 100))
+
+ rx = self.send_and_expect(self.pg0, p_version * NUM_PKTS, self.pg0)
+ rx = rx[0]
+ icmp = rx[ICMPv6TimeExceeded]
+
+ # 0: "hop limit exceeded in transit",
+ self.assertEqual((icmp.type, icmp.code), (3, 0))
+
+ icmpv6_data = '\x0a' * 18
+ all_0s = "::"
+ all_1s = "FFFF:FFFF:FFFF:FFFF:FFFF:FFFF:FFFF:FFFF"
+
+ @parameterized.expand([
+ # Name, src, dst, l4proto, msg, timeout
+ ("src='iface', dst='iface'", None, None,
+ inet6.UDP(sport=1234, dport=1234), "funky version", None),
+ ("src='All 0's', dst='iface'", all_0s, None,
+ ICMPv6EchoRequest(id=0xb, seq=5, data=icmpv6_data), None, 0.1),
+ ("src='iface', dst='All 0's'", None, all_0s,
+ ICMPv6EchoRequest(id=0xb, seq=5, data=icmpv6_data), None, 0.1),
+ ("src='All 1's', dst='iface'", all_1s, None,
+ ICMPv6EchoRequest(id=0xb, seq=5, data=icmpv6_data), None, 0.1),
+ ("src='iface', dst='All 1's'", None, all_1s,
+ ICMPv6EchoRequest(id=0xb, seq=5, data=icmpv6_data), None, 0.1),
+ ("src='All 1's', dst='All 1's'", all_1s, all_1s,
+ ICMPv6EchoRequest(id=0xb, seq=5, data=icmpv6_data), None, 0.1),
+
+ ])
+ def test_ip_input_no_replies(self, name, src, dst, l4, msg, timeout):
+
+ self._testMethodDoc = 'IPv6 Input Exception - %s' % name
+
+ p_version = (Ether(src=self.pg0.remote_mac,
+ dst=self.pg0.local_mac) /
+ IPv6(src=src or self.pg0.remote_ip6,
+ dst=dst or self.pg1.remote_ip6,
+ version=3) /
+ l4 /
+ Raw(b'\xa5' * 100))
+
+ self.send_and_assert_no_replies(self.pg0, p_version * NUM_PKTS,
+ remark=msg or "",
+ timeout=timeout)
+
+ def test_hop_by_hop(self):
+ """ Hop-by-hop header test """
+
+ p = (Ether(src=self.pg0.remote_mac,
+ dst=self.pg0.local_mac) /
+ IPv6(src=self.pg0.remote_ip6, dst=self.pg0.local_ip6) /
+ IPv6ExtHdrHopByHop() /
+ inet6.UDP(sport=1234, dport=1234) /
+ Raw(b'\xa5' * 100))
+
+ self.pg0.add_stream(p)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+
+
+class TestIPReplace(VppTestCase):
+ """ IPv6 Table Replace """
+
+ @classmethod
+ def setUpClass(cls):
+ super(TestIPReplace, cls).setUpClass()
+
+ @classmethod
+ def tearDownClass(cls):
+ super(TestIPReplace, cls).tearDownClass()
+
+ def setUp(self):
+ super(TestIPReplace, self).setUp()
+
+ self.create_pg_interfaces(range(4))
+
+ table_id = 1
+ self.tables = []
+
+ for i in self.pg_interfaces:
+ i.admin_up()
+ i.config_ip6()
+ i.resolve_arp()
+ i.generate_remote_hosts(2)
+ self.tables.append(VppIpTable(self, table_id,
+ True).add_vpp_config())
+ table_id += 1
+
+ def tearDown(self):
+ super(TestIPReplace, self).tearDown()
+ for i in self.pg_interfaces:
+ i.admin_down()
+ i.unconfig_ip4()
+
+ def test_replace(self):
+ """ IP Table Replace """
+
+ N_ROUTES = 20
+ links = [self.pg0, self.pg1, self.pg2, self.pg3]
+ routes = [[], [], [], []]
+
+ # the sizes of 'empty' tables
+ for t in self.tables:
+ self.assertEqual(len(t.dump()), 2)
+ self.assertEqual(len(t.mdump()), 5)
+
+ # load up the tables with some routes
+ for ii, t in enumerate(self.tables):
+ for jj in range(1, N_ROUTES):
+ uni = VppIpRoute(
+ self, "2001::%d" % jj if jj != 0 else "2001::", 128,
+ [VppRoutePath(links[ii].remote_hosts[0].ip6,
+ links[ii].sw_if_index),
+ VppRoutePath(links[ii].remote_hosts[1].ip6,
+ links[ii].sw_if_index)],
+ table_id=t.table_id).add_vpp_config()
+ multi = VppIpMRoute(
+ self, "::",
+ "ff:2001::%d" % jj, 128,
+ MRouteEntryFlags.MFIB_ENTRY_FLAG_NONE,
+ [VppMRoutePath(self.pg0.sw_if_index,
+ MRouteItfFlags.MFIB_ITF_FLAG_ACCEPT,
+ proto=FibPathProto.FIB_PATH_NH_PROTO_IP6),
+ VppMRoutePath(self.pg1.sw_if_index,
+ MRouteItfFlags.MFIB_ITF_FLAG_FORWARD,
+ proto=FibPathProto.FIB_PATH_NH_PROTO_IP6),
+ VppMRoutePath(self.pg2.sw_if_index,
+ MRouteItfFlags.MFIB_ITF_FLAG_FORWARD,
+ proto=FibPathProto.FIB_PATH_NH_PROTO_IP6),
+ VppMRoutePath(self.pg3.sw_if_index,
+ MRouteItfFlags.MFIB_ITF_FLAG_FORWARD,
+ proto=FibPathProto.FIB_PATH_NH_PROTO_IP6)],
+ table_id=t.table_id).add_vpp_config()
+ routes[ii].append({'uni': uni,
+ 'multi': multi})
+
+ #
+ # replace the tables a few times
+ #
+ for kk in range(3):
+ # replace each table
+ for t in self.tables:
+ t.replace_begin()
+
+ # all the routes are still there
+ for ii, t in enumerate(self.tables):
+ dump = t.dump()
+ mdump = t.mdump()
+ for r in routes[ii]:
+ self.assertTrue(find_route_in_dump(dump, r['uni'], t))
+ self.assertTrue(find_mroute_in_dump(mdump, r['multi'], t))
+
+ # redownload the even numbered routes
+ for ii, t in enumerate(self.tables):
+ for jj in range(0, N_ROUTES, 2):
+ routes[ii][jj]['uni'].add_vpp_config()
+ routes[ii][jj]['multi'].add_vpp_config()
+
+ # signal each table converged
+ for t in self.tables:
+ t.replace_end()
+
+ # we should find the even routes, but not the odd
+ for ii, t in enumerate(self.tables):
+ dump = t.dump()
+ mdump = t.mdump()
+ for jj in range(0, N_ROUTES, 2):
+ self.assertTrue(find_route_in_dump(
+ dump, routes[ii][jj]['uni'], t))
+ self.assertTrue(find_mroute_in_dump(
+ mdump, routes[ii][jj]['multi'], t))
+ for jj in range(1, N_ROUTES - 1, 2):
+ self.assertFalse(find_route_in_dump(
+ dump, routes[ii][jj]['uni'], t))
+ self.assertFalse(find_mroute_in_dump(
+ mdump, routes[ii][jj]['multi'], t))
+
+ # reload all the routes
+ for ii, t in enumerate(self.tables):
+ for r in routes[ii]:
+ r['uni'].add_vpp_config()
+ r['multi'].add_vpp_config()
+
+ # all the routes are still there
+ for ii, t in enumerate(self.tables):
+ dump = t.dump()
+ mdump = t.mdump()
+ for r in routes[ii]:
+ self.assertTrue(find_route_in_dump(dump, r['uni'], t))
+ self.assertTrue(find_mroute_in_dump(mdump, r['multi'], t))
+
+ #
+ # finally flush the tables for good measure
+ #
+ for t in self.tables:
+ t.flush()
+ self.assertEqual(len(t.dump()), 2)
+ self.assertEqual(len(t.mdump()), 5)
+
+
+class TestIP6Replace(VppTestCase):
+ """ IPv4 Interface Address Replace """
+
+ @classmethod
+ def setUpClass(cls):
+ super(TestIP6Replace, cls).setUpClass()
+
+ @classmethod
+ def tearDownClass(cls):
+ super(TestIP6Replace, cls).tearDownClass()
+
+ def setUp(self):
+ super(TestIP6Replace, self).setUp()
+
+ self.create_pg_interfaces(range(4))
+
+ for i in self.pg_interfaces:
+ i.admin_up()
+
+ def tearDown(self):
+ super(TestIP6Replace, self).tearDown()
+ for i in self.pg_interfaces:
+ i.admin_down()
+
+ def get_n_pfxs(self, intf):
+ return len(self.vapi.ip_address_dump(intf.sw_if_index, True))
+
+ def test_replace(self):
+ """ IP interface address replace """
+
+ intf_pfxs = [[], [], [], []]
+
+ # add prefixes to each of the interfaces
+ for i in range(len(self.pg_interfaces)):
+ intf = self.pg_interfaces[i]
+
+ # 2001:16:x::1/64
+ addr = "2001:16:%d::1" % intf.sw_if_index
+ a = VppIpInterfaceAddress(self, intf, addr, 64).add_vpp_config()
+ intf_pfxs[i].append(a)
+
+ # 2001:16:x::2/64 - a different address in the same subnet as above
+ addr = "2001:16:%d::2" % intf.sw_if_index
+ a = VppIpInterfaceAddress(self, intf, addr, 64).add_vpp_config()
+ intf_pfxs[i].append(a)
+
+ # 2001:15:x::2/64 - a different address and subnet
+ addr = "2001:15:%d::2" % intf.sw_if_index
+ a = VppIpInterfaceAddress(self, intf, addr, 64).add_vpp_config()
+ intf_pfxs[i].append(a)
+
+ # a dump should n_address in it
+ for intf in self.pg_interfaces:
+ self.assertEqual(self.get_n_pfxs(intf), 3)
+
+ #
+ # remove all the address thru a replace
+ #
+ self.vapi.sw_interface_address_replace_begin()
+ self.vapi.sw_interface_address_replace_end()
+ for intf in self.pg_interfaces:
+ self.assertEqual(self.get_n_pfxs(intf), 0)
+
+ #
+ # add all the interface addresses back
+ #
+ for p in intf_pfxs:
+ for v in p:
+ v.add_vpp_config()
+ for intf in self.pg_interfaces:
+ self.assertEqual(self.get_n_pfxs(intf), 3)
+
+ #
+ # replace again, but this time update/re-add the address on the first
+ # two interfaces
+ #
+ self.vapi.sw_interface_address_replace_begin()
+
+ for p in intf_pfxs[:2]:
+ for v in p:
+ v.add_vpp_config()
+
+ self.vapi.sw_interface_address_replace_end()
+
+ # on the first two the address still exist,
+ # on the other two they do not
+ for intf in self.pg_interfaces[:2]:
+ self.assertEqual(self.get_n_pfxs(intf), 3)
+ for p in intf_pfxs[:2]:
+ for v in p:
+ self.assertTrue(v.query_vpp_config())
+ for intf in self.pg_interfaces[2:]:
+ self.assertEqual(self.get_n_pfxs(intf), 0)
+
+ #
+ # add all the interface addresses back on the last two
+ #
+ for p in intf_pfxs[2:]:
+ for v in p:
+ v.add_vpp_config()
+ for intf in self.pg_interfaces:
+ self.assertEqual(self.get_n_pfxs(intf), 3)
+
+ #
+ # replace again, this time add different prefixes on all the interfaces
+ #
+ self.vapi.sw_interface_address_replace_begin()
+
+ pfxs = []
+ for intf in self.pg_interfaces:
+ # 2001:18:x::1/64
+ addr = "2001:18:%d::1" % intf.sw_if_index
+ pfxs.append(VppIpInterfaceAddress(self, intf, addr,
+ 64).add_vpp_config())
+
+ self.vapi.sw_interface_address_replace_end()
+
+ # only .18 should exist on each interface
+ for intf in self.pg_interfaces:
+ self.assertEqual(self.get_n_pfxs(intf), 1)
+ for pfx in pfxs:
+ self.assertTrue(pfx.query_vpp_config())
+
+ #
+ # remove everything
+ #
+ self.vapi.sw_interface_address_replace_begin()
+ self.vapi.sw_interface_address_replace_end()
+ for intf in self.pg_interfaces:
+ self.assertEqual(self.get_n_pfxs(intf), 0)
+
+ #
+ # add prefixes to each interface. post-begin add the prefix from
+ # interface X onto interface Y. this would normally be an error
+ # since it would generate a 'duplicate address' warning. but in
+ # this case, since what is newly downloaded is sane, it's ok
+ #
+ for intf in self.pg_interfaces:
+ # 2001:18:x::1/64
+ addr = "2001:18:%d::1" % intf.sw_if_index
+ VppIpInterfaceAddress(self, intf, addr, 64).add_vpp_config()
+
+ self.vapi.sw_interface_address_replace_begin()
+
+ pfxs = []
+ for intf in self.pg_interfaces:
+ # 2001:18:x::1/64
+ addr = "2001:18:%d::1" % (intf.sw_if_index + 1)
+ pfxs.append(VppIpInterfaceAddress(self, intf,
+ addr, 64).add_vpp_config())
+
+ self.vapi.sw_interface_address_replace_end()
+
+ self.logger.info(self.vapi.cli("sh int addr"))
+
+ for intf in self.pg_interfaces:
+ self.assertEqual(self.get_n_pfxs(intf), 1)
+ for pfx in pfxs:
+ self.assertTrue(pfx.query_vpp_config())
+
+