+ self.send_and_expect_only(self.pg0, port_pkts, self.pg3)
+
+
+class IP6PuntSetup(object):
+ """Setup for IPv6 Punt Police/Redirect"""
+
+ def punt_setup(self):
+ self.create_pg_interfaces(range(4))
+
+ for i in self.pg_interfaces:
+ i.admin_up()
+ i.config_ip6()
+ i.resolve_ndp()
+
+ self.pkt = (
+ 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)
+ )
+
+ def punt_teardown(self):
+ for i in self.pg_interfaces:
+ i.unconfig_ip6()
+ i.admin_down()
+
+
+class TestIP6Punt(IP6PuntSetup, VppTestCase):
+ """IPv6 Punt Police/Redirect"""
+
+ def setUp(self):
+ super(TestIP6Punt, self).setUp()
+ super(TestIP6Punt, self).punt_setup()
+
+ def tearDown(self):
+ super(TestIP6Punt, self).punt_teardown()
+ super(TestIP6Punt, self).tearDown()
+
+ def test_ip_punt(self):
+ """IP6 punt police and redirect"""
+
+ pkts = self.pkt * 1025
+
+ #
+ # Configure a punt redirect via pg1.
+ #
+ nh_addr = self.pg1.remote_ip6
+ ip_punt_redirect = VppIpPuntRedirect(
+ self, self.pg0.sw_if_index, self.pg1.sw_if_index, nh_addr
+ )
+ ip_punt_redirect.add_vpp_config()
+
+ 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()
+ ip_punt_policer = VppIpPuntPolicer(self, policer.policer_index, is_ip6=True)
+ ip_punt_policer.add_vpp_config()
+
+ 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)
+ stats = policer.get_stats()
+
+ # Single rate policer - expect conform, violate but no exceed
+ self.assertGreater(stats["conform_packets"], 0)
+ self.assertEqual(stats["exceed_packets"], 0)
+ self.assertGreater(stats["violate_packets"], 0)
+
+ self.assertGreater(len(rx), 0)
+ self.assertLess(len(rx), len(pkts))
+
+ #
+ # remove the policer. back to full rx
+ #
+ ip_punt_policer.remove_vpp_config()
+ policer.remove_vpp_config()
+ self.send_and_expect(self.pg0, pkts, self.pg1)
+
+ #
+ # remove the redirect. expect full drop.
+ #
+ ip_punt_redirect.remove_vpp_config()
+ self.send_and_assert_no_replies(self.pg0, pkts, "IP no punt config")
+
+ #
+ # Add a redirect that is not input port selective
+ #
+ ip_punt_redirect = VppIpPuntRedirect(
+ self, 0xFFFFFFFF, self.pg1.sw_if_index, nh_addr
+ )
+ ip_punt_redirect.add_vpp_config()
+ self.send_and_expect(self.pg0, pkts, self.pg1)
+ ip_punt_redirect.remove_vpp_config()
+
+ def test_ip_punt_dump(self):
+ """IP6 punt redirect dump"""
+
+ #
+ # Configure a punt redirects
+ #
+ nh_address = self.pg3.remote_ip6
+ ipr_03 = VppIpPuntRedirect(
+ self, self.pg0.sw_if_index, self.pg3.sw_if_index, nh_address
+ )
+ ipr_13 = VppIpPuntRedirect(
+ self, self.pg1.sw_if_index, self.pg3.sw_if_index, nh_address
+ )
+ ipr_23 = VppIpPuntRedirect(
+ self, self.pg2.sw_if_index, self.pg3.sw_if_index, "0::0"
+ )
+ ipr_03.add_vpp_config()
+ ipr_13.add_vpp_config()
+ ipr_23.add_vpp_config()
+
+ #
+ # Dump pg0 punt redirects
+ #
+ self.assertTrue(ipr_03.query_vpp_config())
+ self.assertTrue(ipr_13.query_vpp_config())
+ self.assertTrue(ipr_23.query_vpp_config())
+
+ #
+ # 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 TestIP6PuntHandoff(IP6PuntSetup, VppTestCase):
+ """IPv6 Punt Police/Redirect"""
+
+ vpp_worker_count = 2
+
+ def setUp(self):
+ super(TestIP6PuntHandoff, self).setUp()
+ super(TestIP6PuntHandoff, self).punt_setup()
+
+ def tearDown(self):
+ super(TestIP6PuntHandoff, self).punt_teardown()
+ super(TestIP6PuntHandoff, self).tearDown()
+
+ def test_ip_punt(self):
+ """IP6 punt policer thread handoff"""
+ pkts = self.pkt * NUM_PKTS
+
+ #
+ # Configure a punt redirect via pg1.
+ #
+ nh_addr = self.pg1.remote_ip6
+ ip_punt_redirect = VppIpPuntRedirect(
+ self, self.pg0.sw_if_index, self.pg1.sw_if_index, nh_addr
+ )
+ ip_punt_redirect.add_vpp_config()
+
+ action_tx = PolicerAction(
+ VppEnum.vl_api_sse2_qos_action_type_t.SSE2_QOS_ACTION_API_TRANSMIT, 0
+ )
+ #
+ # This policer drops no packets, we are just
+ # testing that they get to the right thread.
+ #
+ policer = VppPolicer(
+ self,
+ "ip6-punt",
+ 400,
+ 0,
+ 10,
+ 0,
+ 1,
+ 0,
+ 0,
+ False,
+ action_tx,
+ action_tx,
+ action_tx,
+ )
+ policer.add_vpp_config()
+ ip_punt_policer = VppIpPuntPolicer(self, policer.policer_index, is_ip6=True)
+ ip_punt_policer.add_vpp_config()
+
+ for worker in [0, 1]:
+ self.send_and_expect(self.pg0, pkts, self.pg1, worker=worker)
+ if worker == 0:
+ self.logger.debug(self.vapi.cli("show trace max 100"))
+
+ # Combined stats, all threads
+ stats = policer.get_stats()
+
+ # Single rate policer - expect conform, violate but no exceed
+ self.assertGreater(stats["conform_packets"], 0)
+ self.assertEqual(stats["exceed_packets"], 0)
+ self.assertGreater(stats["violate_packets"], 0)
+
+ # Worker 0, should have done all the policing
+ stats0 = policer.get_stats(worker=0)
+ self.assertEqual(stats, stats0)
+
+ # Worker 1, should have handed everything off
+ stats1 = policer.get_stats(worker=1)
+ self.assertEqual(stats1["conform_packets"], 0)
+ self.assertEqual(stats1["exceed_packets"], 0)
+ self.assertEqual(stats1["violate_packets"], 0)
+
+ # Bind the policer to worker 1 and repeat
+ policer.bind_vpp_config(1, True)
+ for worker in [0, 1]:
+ self.send_and_expect(self.pg0, pkts, self.pg1, worker=worker)
+ self.logger.debug(self.vapi.cli("show trace max 100"))
+
+ # The 2 workers should now have policed the same amount
+ stats = policer.get_stats()
+ stats0 = policer.get_stats(worker=0)
+ stats1 = policer.get_stats(worker=1)
+
+ self.assertGreater(stats0["conform_packets"], 0)
+ self.assertEqual(stats0["exceed_packets"], 0)
+ self.assertGreater(stats0["violate_packets"], 0)
+
+ self.assertGreater(stats1["conform_packets"], 0)
+ self.assertEqual(stats1["exceed_packets"], 0)
+ self.assertGreater(stats1["violate_packets"], 0)
+
+ self.assertEqual(
+ stats0["conform_packets"] + stats1["conform_packets"],
+ stats["conform_packets"],
+ )
+
+ self.assertEqual(
+ stats0["violate_packets"] + stats1["violate_packets"],
+ stats["violate_packets"],
+ )
+
+ # Unbind the policer and repeat
+ policer.bind_vpp_config(1, False)
+ for worker in [0, 1]:
+ self.send_and_expect(self.pg0, pkts, self.pg1, worker=worker)
+ self.logger.debug(self.vapi.cli("show trace max 100"))
+
+ # The policer should auto-bind to worker 0 when packets arrive
+ stats = policer.get_stats()
+ stats0new = policer.get_stats(worker=0)
+ stats1new = policer.get_stats(worker=1)
+
+ self.assertGreater(stats0new["conform_packets"], stats0["conform_packets"])
+ self.assertEqual(stats0new["exceed_packets"], 0)
+ self.assertGreater(stats0new["violate_packets"], stats0["violate_packets"])
+
+ self.assertEqual(stats1, stats1new)
+
+ #
+ # Clean up
+ #
+ ip_punt_policer.remove_vpp_config()
+ policer.remove_vpp_config()
+ ip_punt_redirect.remove_vpp_config()
+
+
+class TestIP6Deag(VppTestCase):
+ """IPv6 Deaggregate Routes"""
+
+ @classmethod
+ def setUpClass(cls):
+ super(TestIP6Deag, cls).setUpClass()
+
+ @classmethod
+ def tearDownClass(cls):
+ super(TestIP6Deag, cls).tearDownClass()
+
+ def setUp(self):
+ super(TestIP6Deag, 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(TestIP6Deag, 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)
+ )
+
+ rxs = self.send_and_expect_some(self.pg0, p_version * NUM_PKTS, self.pg0)
+
+ for rx in rxs:
+ 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 TestIP6Replace(VppTestCase):
+ """IPv6 Table 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))
+
+ table_id = 1
+ self.tables = []
+
+ for i in self.pg_interfaces:
+ i.admin_up()
+ i.config_ip6()
+ i.generate_remote_hosts(2)
+ self.tables.append(VppIpTable(self, table_id, True).add_vpp_config())
+ table_id += 1
+
+ def tearDown(self):
+ super(TestIP6Replace, self).tearDown()
+ for i in self.pg_interfaces:
+ i.admin_down()
+ i.unconfig_ip6()
+
+ def test_replace(self):
+ """IP Table Replace"""
+
+ MRouteItfFlags = VppEnum.vl_api_mfib_itf_flags_t
+ MRouteEntryFlags = VppEnum.vl_api_mfib_entry_flags_t
+ 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_API_ENTRY_FLAG_NONE,
+ [
+ VppMRoutePath(
+ self.pg0.sw_if_index,
+ MRouteItfFlags.MFIB_API_ITF_FLAG_ACCEPT,
+ proto=FibPathProto.FIB_PATH_NH_PROTO_IP6,
+ ),
+ VppMRoutePath(
+ self.pg1.sw_if_index,
+ MRouteItfFlags.MFIB_API_ITF_FLAG_FORWARD,
+ proto=FibPathProto.FIB_PATH_NH_PROTO_IP6,
+ ),
+ VppMRoutePath(
+ self.pg2.sw_if_index,
+ MRouteItfFlags.MFIB_API_ITF_FLAG_FORWARD,
+ proto=FibPathProto.FIB_PATH_NH_PROTO_IP6,
+ ),
+ VppMRoutePath(
+ self.pg3.sw_if_index,
+ MRouteItfFlags.MFIB_API_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 TestIP6AddrReplace(VppTestCase):
+ """IPv6 Interface Address Replace"""
+
+ @classmethod
+ def setUpClass(cls):
+ super(TestIP6AddrReplace, cls).setUpClass()
+
+ @classmethod
+ def tearDownClass(cls):
+ super(TestIP6AddrReplace, cls).tearDownClass()
+
+ def setUp(self):
+ super(TestIP6AddrReplace, self).setUp()
+
+ self.create_pg_interfaces(range(4))
+
+ for i in self.pg_interfaces:
+ i.admin_up()
+
+ def tearDown(self):
+ super(TestIP6AddrReplace, 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())
+
+
+class TestIP6LinkLocal(VppTestCase):
+ """IPv6 Link Local"""
+
+ @classmethod
+ def setUpClass(cls):
+ super(TestIP6LinkLocal, cls).setUpClass()
+
+ @classmethod
+ def tearDownClass(cls):
+ super(TestIP6LinkLocal, cls).tearDownClass()
+
+ def setUp(self):
+ super(TestIP6LinkLocal, self).setUp()
+
+ self.create_pg_interfaces(range(2))
+
+ for i in self.pg_interfaces:
+ i.admin_up()
+
+ def tearDown(self):
+ super(TestIP6LinkLocal, self).tearDown()
+ for i in self.pg_interfaces:
+ i.admin_down()
+
+ def test_ip6_ll(self):
+ """IPv6 Link Local"""
+
+ #
+ # two APIs to add a link local address.
+ # 1 - just like any other prefix
+ # 2 - with the special set LL API
+ #
+
+ #
+ # First with the API to set a 'normal' prefix
+ #
+ ll1 = "fe80:1::1"
+ ll2 = "fe80:2::2"
+ ll3 = "fe80:3::3"
+
+ VppNeighbor(
+ self, self.pg0.sw_if_index, self.pg0.remote_mac, ll2
+ ).add_vpp_config()
+
+ VppIpInterfaceAddress(self, self.pg0, ll1, 128).add_vpp_config()
+
+ #
+ # should be able to ping the ll
+ #
+ p_echo_request_1 = (
+ Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
+ / IPv6(src=ll2, dst=ll1)
+ / ICMPv6EchoRequest()
+ )
+
+ self.send_and_expect(self.pg0, [p_echo_request_1], self.pg0)
+
+ #
+ # change the link-local on pg0
+ #
+ v_ll3 = VppIpInterfaceAddress(self, self.pg0, ll3, 128).add_vpp_config()
+
+ p_echo_request_3 = (
+ Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
+ / IPv6(src=ll2, dst=ll3)
+ / ICMPv6EchoRequest()
+ )
+
+ self.send_and_expect(self.pg0, [p_echo_request_3], self.pg0)
+
+ #
+ # set a normal v6 prefix on the link
+ #
+ self.pg0.config_ip6()
+
+ self.send_and_expect(self.pg0, [p_echo_request_3], self.pg0)
+
+ # the link-local cannot be removed
+ with self.vapi.assert_negative_api_retval():
+ v_ll3.remove_vpp_config()
+
+ #
+ # Use the specific link-local API on pg1
+ #
+ VppIp6LinkLocalAddress(self, self.pg1, ll1).add_vpp_config()
+ self.send_and_expect(self.pg1, [p_echo_request_1], self.pg1)
+
+ VppIp6LinkLocalAddress(self, self.pg1, ll3).add_vpp_config()
+ self.send_and_expect(self.pg1, [p_echo_request_3], self.pg1)
+
+ def test_ip6_ll_p2p(self):
+ """IPv6 Link Local P2P (GRE)"""
+
+ self.pg0.config_ip4()
+ self.pg0.resolve_arp()
+ gre_if = VppGreInterface(
+ self, self.pg0.local_ip4, self.pg0.remote_ip4
+ ).add_vpp_config()
+ gre_if.admin_up()
+
+ ll1 = "fe80:1::1"
+ ll2 = "fe80:2::2"
+
+ VppIpInterfaceAddress(self, gre_if, ll1, 128).add_vpp_config()
+
+ self.logger.info(self.vapi.cli("sh ip6-ll gre0 fe80:2::2"))
+
+ p_echo_request_1 = (
+ Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
+ / IP(src=self.pg0.remote_ip4, dst=self.pg0.local_ip4)
+ / GRE()
+ / IPv6(src=ll2, dst=ll1)
+ / ICMPv6EchoRequest()
+ )
+ self.send_and_expect(self.pg0, [p_echo_request_1], self.pg0)
+
+ self.pg0.unconfig_ip4()
+ gre_if.remove_vpp_config()
+
+ def test_ip6_ll_p2mp(self):
+ """IPv6 Link Local P2MP (GRE)"""
+
+ self.pg0.config_ip4()
+ self.pg0.resolve_arp()
+
+ gre_if = VppGreInterface(
+ self,
+ self.pg0.local_ip4,
+ "0.0.0.0",
+ mode=(VppEnum.vl_api_tunnel_mode_t.TUNNEL_API_MODE_MP),
+ ).add_vpp_config()
+ gre_if.admin_up()
+
+ ll1 = "fe80:1::1"
+ ll2 = "fe80:2::2"
+
+ VppIpInterfaceAddress(self, gre_if, ll1, 128).add_vpp_config()
+
+ p_echo_request_1 = (
+ Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
+ / IP(src=self.pg0.remote_ip4, dst=self.pg0.local_ip4)
+ / GRE()
+ / IPv6(src=ll2, dst=ll1)
+ / ICMPv6EchoRequest()
+ )
+
+ # no route back at this point
+ self.send_and_assert_no_replies(self.pg0, [p_echo_request_1])
+
+ # add teib entry for the peer
+ teib = VppTeib(self, gre_if, ll2, self.pg0.remote_ip4)
+ teib.add_vpp_config()
+
+ self.logger.info(self.vapi.cli("sh ip6-ll gre0 %s" % ll2))
+ self.send_and_expect(self.pg0, [p_echo_request_1], self.pg0)
+
+ # teardown
+ self.pg0.unconfig_ip4()
+
+
+class TestIPv6PathMTU(VppTestCase):
+ """IPv6 Path MTU"""
+
+ def setUp(self):
+ super(TestIPv6PathMTU, self).setUp()
+
+ self.create_pg_interfaces(range(2))
+
+ # setup all interfaces
+ for i in self.pg_interfaces:
+ i.admin_up()
+ i.config_ip6()
+ i.resolve_ndp()
+
+ def tearDown(self):
+ super(TestIPv6PathMTU, self).tearDown()
+ for i in self.pg_interfaces:
+ i.unconfig_ip6()
+ i.admin_down()
+
+ def test_path_mtu_local(self):
+ """Path MTU for attached neighbour"""
+
+ self.vapi.cli("set log class ip level debug")
+ #
+ # The goal here is not test that fragmentation works correctly,
+ # that's done elsewhere, the intent is to ensure that the Path MTU
+ # settings are honoured.
+ #
+
+ #
+ # IPv6 will only frag locally generated packets, so use tunnelled
+ # packets post encap
+ #
+ tun = VppIpIpTunInterface(
+ self, self.pg1, self.pg1.local_ip6, self.pg1.remote_ip6
+ )
+ tun.add_vpp_config()
+ tun.admin_up()
+ tun.config_ip6()
+
+ # set the interface MTU to a reasonable value
+ self.vapi.sw_interface_set_mtu(self.pg1.sw_if_index, [2800, 0, 0, 0])
+
+ p_6k = (
+ Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
+ / IPv6(src=self.pg0.remote_ip6, dst=tun.remote_ip6)
+ / UDP(sport=1234, dport=5678)
+ / Raw(b"0xa" * 2000)
+ )
+ p_2k = (
+ Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
+ / IPv6(src=self.pg0.remote_ip6, dst=tun.remote_ip6)
+ / UDP(sport=1234, dport=5678)
+ / Raw(b"0xa" * 1000)
+ )
+ p_1k = (
+ Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
+ / IPv6(src=self.pg0.remote_ip6, dst=tun.remote_ip6)
+ / UDP(sport=1234, dport=5678)
+ / Raw(b"0xa" * 600)
+ )
+
+ nbr = VppNeighbor(
+ self, self.pg1.sw_if_index, self.pg1.remote_mac, self.pg1.remote_ip6
+ ).add_vpp_config()
+
+ # this is now the interface MTU frags
+ self.send_and_expect(self.pg0, [p_6k], self.pg1, n_rx=4)
+ self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=2)
+ self.send_and_expect(self.pg0, [p_1k], self.pg1)
+
+ # drop the path MTU for this neighbour to below the interface MTU
+ # expect more frags
+ pmtu = VppIpPathMtu(self, self.pg1.remote_ip6, 1300).add_vpp_config()
+
+ # print/format the adj delegate and trackers
+ self.logger.info(self.vapi.cli("sh ip pmtu"))
+ self.logger.info(self.vapi.cli("sh adj 7"))
+
+ self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=3)
+ self.send_and_expect(self.pg0, [p_1k], self.pg1, n_rx=2)
+
+ # increase the path MTU to more than the interface
+ # expect to use the interface MTU
+ pmtu.modify(8192)
+
+ self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=2)
+ self.send_and_expect(self.pg0, [p_1k], self.pg1)
+
+ # go back to an MTU from the path
+ pmtu.modify(1300)
+
+ self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=3)
+ self.send_and_expect(self.pg0, [p_1k], self.pg1, n_rx=2)
+
+ # raise the interface's MTU
+ # should still use that of the path
+ self.vapi.sw_interface_set_mtu(self.pg1.sw_if_index, [2000, 0, 0, 0])
+ self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=3)
+ self.send_and_expect(self.pg0, [p_1k], self.pg1, n_rx=2)
+
+ # set path high and interface low
+ pmtu.modify(2000)
+ self.vapi.sw_interface_set_mtu(self.pg1.sw_if_index, [1300, 0, 0, 0])
+ self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=3)
+ self.send_and_expect(self.pg0, [p_1k], self.pg1, n_rx=2)
+
+ # remove the path MTU
+ self.vapi.sw_interface_set_mtu(self.pg1.sw_if_index, [2800, 0, 0, 0])
+ pmtu.modify(0)
+
+ self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=2)
+ self.send_and_expect(self.pg0, [p_1k], self.pg1)
+
+ def test_path_mtu_remote(self):
+ """Path MTU for remote neighbour"""
+
+ self.vapi.cli("set log class ip level debug")
+ #
+ # The goal here is not test that fragmentation works correctly,
+ # that's done elsewhere, the intent is to ensure that the Path MTU
+ # settings are honoured.
+ #
+ tun_dst = "2001::1"
+
+ route = VppIpRoute(
+ self, tun_dst, 64, [VppRoutePath(self.pg1.remote_ip6, self.pg1.sw_if_index)]
+ ).add_vpp_config()
+
+ #
+ # IPv6 will only frag locally generated packets, so use tunnelled
+ # packets post encap
+ #
+ tun = VppIpIpTunInterface(self, self.pg1, self.pg1.local_ip6, tun_dst)
+ tun.add_vpp_config()
+ tun.admin_up()
+ tun.config_ip6()
+
+ # set the interface MTU to a reasonable value
+ self.vapi.sw_interface_set_mtu(self.pg1.sw_if_index, [2800, 0, 0, 0])
+
+ p_2k = (
+ Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
+ / IPv6(src=self.pg0.remote_ip6, dst=tun.remote_ip6)
+ / UDP(sport=1234, dport=5678)
+ / Raw(b"0xa" * 1000)
+ )
+ p_1k = (
+ Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
+ / IPv6(src=self.pg0.remote_ip6, dst=tun.remote_ip6)
+ / UDP(sport=1234, dport=5678)
+ / Raw(b"0xa" * 600)
+ )
+
+ nbr = VppNeighbor(
+ self, self.pg1.sw_if_index, self.pg1.remote_mac, self.pg1.remote_ip6
+ ).add_vpp_config()
+
+ # this is now the interface MTU frags
+ self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=2)
+ self.send_and_expect(self.pg0, [p_1k], self.pg1)
+
+ # drop the path MTU for this neighbour to below the interface MTU
+ # expect more frags
+ pmtu = VppIpPathMtu(self, tun_dst, 1300).add_vpp_config()
+
+ # print/format the fib entry/dpo
+ self.logger.info(self.vapi.cli("sh ip6 fib 2001::1"))
+
+ self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=3)
+ self.send_and_expect(self.pg0, [p_1k], self.pg1, n_rx=2)
+
+ # increase the path MTU to more than the interface
+ # expect to use the interface MTU
+ pmtu.modify(8192)
+
+ self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=2)
+ self.send_and_expect(self.pg0, [p_1k], self.pg1)
+
+ # go back to an MTU from the path
+ pmtu.modify(1300)
+
+ self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=3)
+ self.send_and_expect(self.pg0, [p_1k], self.pg1, n_rx=2)
+
+ # raise the interface's MTU
+ # should still use that of the path
+ self.vapi.sw_interface_set_mtu(self.pg1.sw_if_index, [2000, 0, 0, 0])
+ self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=3)
+ self.send_and_expect(self.pg0, [p_1k], self.pg1, n_rx=2)
+
+ # turn the tun_dst into an attached neighbour
+ route.modify([VppRoutePath("::", self.pg1.sw_if_index)])
+ nbr2 = VppNeighbor(
+ self, self.pg1.sw_if_index, self.pg1.remote_mac, tun_dst
+ ).add_vpp_config()
+
+ self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=3)
+ self.send_and_expect(self.pg0, [p_1k], self.pg1, n_rx=2)
+
+ # add back to not attached
+ nbr2.remove_vpp_config()
+ route.modify([VppRoutePath(self.pg1.remote_ip6, self.pg1.sw_if_index)])
+
+ # set path high and interface low
+ pmtu.modify(2000)
+ self.vapi.sw_interface_set_mtu(self.pg1.sw_if_index, [1300, 0, 0, 0])
+ self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=3)
+ self.send_and_expect(self.pg0, [p_1k], self.pg1, n_rx=2)
+
+ # remove the path MTU
+ self.vapi.sw_interface_set_mtu(self.pg1.sw_if_index, [2800, 0, 0, 0])
+ pmtu.remove_vpp_config()
+ self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=2)
+ self.send_and_expect(self.pg0, [p_1k], self.pg1)
+
+
+class TestIPFibSource(VppTestCase):
+ """IPv6 Table FibSource"""
+
+ @classmethod
+ def setUpClass(cls):
+ super(TestIPFibSource, cls).setUpClass()
+
+ @classmethod
+ def tearDownClass(cls):
+ super(TestIPFibSource, cls).tearDownClass()
+
+ def setUp(self):
+ super(TestIPFibSource, self).setUp()
+
+ self.create_pg_interfaces(range(2))
+
+ for i in self.pg_interfaces:
+ i.admin_up()
+ i.config_ip6()
+ i.resolve_arp()
+ i.generate_remote_hosts(2)
+ i.configure_ipv6_neighbors()
+
+ def tearDown(self):
+ super(TestIPFibSource, self).tearDown()
+ for i in self.pg_interfaces:
+ i.admin_down()
+ i.unconfig_ip4()
+
+ def test_fib_source(self):
+ """IP Table FibSource"""
+
+ routes = self.vapi.ip_route_v2_dump(0, True)
+
+ # 2 interfaces (4 routes) + 2 specials + 4 neighbours = 10 routes
+ self.assertEqual(len(routes), 10)
+
+ # dump all the sources in the FIB
+ sources = self.vapi.fib_source_dump()
+ for source in sources:
+ if source.src.name == "API":
+ api_source = source.src
+ if source.src.name == "interface":
+ intf_source = source.src
+ if source.src.name == "adjacency":
+ adj_source = source.src
+ if source.src.name == "special":
+ special_source = source.src
+ if source.src.name == "default-route":
+ dr_source = source.src
+
+ # dump the individual route types
+ routes = self.vapi.ip_route_v2_dump(0, True, src=adj_source.id)
+ self.assertEqual(len(routes), 4)
+ routes = self.vapi.ip_route_v2_dump(0, True, src=intf_source.id)
+ self.assertEqual(len(routes), 4)
+ routes = self.vapi.ip_route_v2_dump(0, True, src=special_source.id)
+ self.assertEqual(len(routes), 1)
+ routes = self.vapi.ip_route_v2_dump(0, True, src=dr_source.id)
+ self.assertEqual(len(routes), 1)
+
+ # add a new soure that'a better than the API
+ self.vapi.fib_source_add(
+ src={"name": "bgp", "priority": api_source.priority - 1}
+ )
+
+ # dump all the sources to check our new one is there
+ sources = self.vapi.fib_source_dump()
+
+ for source in sources:
+ if source.src.name == "bgp":
+ bgp_source = source.src
+
+ self.assertTrue(bgp_source)
+ self.assertEqual(bgp_source.priority, api_source.priority - 1)
+
+ # add a route with the default API source
+ r1 = VppIpRouteV2(
+ self,
+ "2001::1",
+ 128,
+ [VppRoutePath(self.pg0.remote_ip6, self.pg0.sw_if_index)],
+ ).add_vpp_config()
+
+ r2 = VppIpRouteV2(
+ self,
+ "2001::1",
+ 128,
+ [VppRoutePath(self.pg1.remote_ip6, self.pg1.sw_if_index)],
+ src=bgp_source.id,
+ ).add_vpp_config()
+
+ # ensure the BGP source takes priority
+ p = (
+ Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
+ / IPv6(src=self.pg0.remote_ip6, dst="2001::1")
+ / inet6.UDP(sport=1234, dport=1234)
+ / Raw(b"\xa5" * 100)
+ )
+
+ self.send_and_expect(self.pg0, [p], self.pg1)
+
+ r2.remove_vpp_config()
+ r1.remove_vpp_config()
+
+ self.assertFalse(find_route(self, "2001::1", 128))
+
+
+class TestIPxAF(VppTestCase):
+ """IP cross AF"""
+
+ @classmethod
+ def setUpClass(cls):
+ super(TestIPxAF, cls).setUpClass()
+
+ @classmethod
+ def tearDownClass(cls):
+ super(TestIPxAF, cls).tearDownClass()
+
+ def setUp(self):
+ super(TestIPxAF, self).setUp()
+
+ self.create_pg_interfaces(range(2))
+
+ for i in self.pg_interfaces:
+ i.admin_up()
+ i.config_ip6()
+ i.config_ip4()
+ i.resolve_arp()
+ i.resolve_ndp()
+
+ def tearDown(self):
+ super(TestIPxAF, self).tearDown()
+ for i in self.pg_interfaces:
+ i.admin_down()
+ i.unconfig_ip4()
+ i.unconfig_ip6()
+
+ def test_x_af(self):
+ """Cross AF routing"""
+
+ N_PKTS = 63
+ # a v4 route via a v6 attached next-hop
+ VppIpRoute(
+ self,
+ "1.1.1.1",
+ 32,
+ [VppRoutePath(self.pg1.remote_ip6, self.pg1.sw_if_index)],
+ ).add_vpp_config()
+
+ p = (
+ Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
+ / IP(src=self.pg0.remote_ip4, dst="1.1.1.1")
+ / UDP(sport=1234, dport=1234)
+ / Raw(b"\xa5" * 100)
+ )
+ rxs = self.send_and_expect(self.pg0, p * N_PKTS, self.pg1)
+
+ for rx in rxs:
+ self.assertEqual(rx[IP].dst, "1.1.1.1")
+
+ # a v6 route via a v4 attached next-hop
+ VppIpRoute(
+ self,
+ "2001::1",
+ 128,
+ [VppRoutePath(self.pg1.remote_ip4, self.pg1.sw_if_index)],
+ ).add_vpp_config()
+
+ p = (
+ Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
+ / IPv6(src=self.pg0.remote_ip6, dst="2001::1")
+ / UDP(sport=1234, dport=1234)
+ / Raw(b"\xa5" * 100)
+ )
+ rxs = self.send_and_expect(self.pg0, p * N_PKTS, self.pg1)
+
+ for rx in rxs:
+ self.assertEqual(rx[IPv6].dst, "2001::1")
+
+ # a recursive v4 route via a v6 next-hop (from above)
+ VppIpRoute(
+ self, "2.2.2.2", 32, [VppRoutePath("2001::1", 0xFFFFFFFF)]
+ ).add_vpp_config()
+
+ p = (
+ Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
+ / IP(src=self.pg0.remote_ip4, dst="2.2.2.2")
+ / UDP(sport=1234, dport=1234)
+ / Raw(b"\xa5" * 100)
+ )
+ rxs = self.send_and_expect(self.pg0, p * N_PKTS, self.pg1)
+
+ # a recursive v4 route via a v6 next-hop
+ VppIpRoute(
+ self, "2.2.2.3", 32, [VppRoutePath(self.pg1.remote_ip6, 0xFFFFFFFF)]
+ ).add_vpp_config()
+
+ p = (
+ Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
+ / IP(src=self.pg0.remote_ip4, dst="2.2.2.3")
+ / UDP(sport=1234, dport=1234)
+ / Raw(b"\xa5" * 100)
+ )
+ rxs = self.send_and_expect(self.pg0, p * N_PKTS, self.pg1)
+
+ # a recursive v6 route via a v4 next-hop
+ VppIpRoute(
+ self, "3001::1", 128, [VppRoutePath(self.pg1.remote_ip4, 0xFFFFFFFF)]
+ ).add_vpp_config()
+
+ p = (
+ Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
+ / IPv6(src=self.pg0.remote_ip6, dst="3001::1")
+ / UDP(sport=1234, dport=1234)
+ / Raw(b"\xa5" * 100)
+ )
+ rxs = self.send_and_expect(self.pg0, p * N_PKTS, self.pg1)
+
+ for rx in rxs:
+ self.assertEqual(rx[IPv6].dst, "3001::1")
+
+ VppIpRoute(
+ self, "3001::2", 128, [VppRoutePath("1.1.1.1", 0xFFFFFFFF)]
+ ).add_vpp_config()
+
+ p = (
+ Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
+ / IPv6(src=self.pg0.remote_ip6, dst="3001::2")
+ / UDP(sport=1234, dport=1234)
+ / Raw(b"\xa5" * 100)
+ )
+ rxs = self.send_and_expect(self.pg0, p * N_PKTS, self.pg1)
+
+ for rx in rxs:
+ self.assertEqual(rx[IPv6].dst, "3001::2")
+
+
+class TestIPv6Punt(VppTestCase):
+ """IPv6 Punt Police/Redirect"""
+
+ def setUp(self):
+ super(TestIPv6Punt, 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(TestIPv6Punt, self).tearDown()
+ for i in self.pg_interfaces:
+ i.unconfig_ip6()
+ i.admin_down()
+
+ def test_ip6_punt(self):
+ """IPv6 punt police and redirect"""
+
+ # use UDP packet that have a port we need to explicitly
+ # register to get punted.
+ pt_l4 = VppEnum.vl_api_punt_type_t.PUNT_API_TYPE_L4
+ af_ip6 = VppEnum.vl_api_address_family_t.ADDRESS_IP6
+ udp_proto = VppEnum.vl_api_ip_proto_t.IP_API_PROTO_UDP
+ punt_udp = {
+ "type": pt_l4,
+ "punt": {
+ "l4": {
+ "af": af_ip6,
+ "protocol": udp_proto,
+ "port": 7654,
+ }
+ },
+ }
+
+ self.vapi.set_punt(is_add=1, punt=punt_udp)
+
+ pkts = (
+ Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
+ / IPv6(src=self.pg0.remote_ip6, dst=self.pg0.local_ip6)
+ / UDP(sport=1234, dport=7654)
+ / Raw(b"\xa5" * 100)
+ ) * 1025
+
+ #
+ # Configure a punt redirect via pg1.
+ #
+ nh_addr = self.pg1.remote_ip6
+ ip_punt_redirect = VppIpPuntRedirect(
+ self, self.pg0.sw_if_index, self.pg1.sw_if_index, nh_addr
+ )
+ ip_punt_redirect.add_vpp_config()
+
+ 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()
+ ip_punt_policer = VppIpPuntPolicer(self, policer.policer_index, is_ip6=True)
+ ip_punt_policer.add_vpp_config()
+
+ 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)
+
+ stats = policer.get_stats()
+
+ # Single rate policer - expect conform, violate but no exceed
+ self.assertGreater(stats["conform_packets"], 0)
+ self.assertEqual(stats["exceed_packets"], 0)
+ self.assertGreater(stats["violate_packets"], 0)
+
+ self.assertGreater(len(rx), 0)
+ self.assertLess(len(rx), len(pkts))
+
+ #
+ # remove the policer. back to full rx
+ #
+ ip_punt_policer.remove_vpp_config()
+ policer.remove_vpp_config()
+ self.send_and_expect(self.pg0, pkts, self.pg1)
+
+ #
+ # remove the redirect. expect full drop.
+ #
+ ip_punt_redirect.remove_vpp_config()
+ self.send_and_assert_no_replies(self.pg0, pkts, "IP no punt config")
+
+ #
+ # Add a redirect that is not input port selective
+ #
+ ip_punt_redirect = VppIpPuntRedirect(
+ self, 0xFFFFFFFF, self.pg1.sw_if_index, nh_addr
+ )
+ ip_punt_redirect.add_vpp_config()
+ self.send_and_expect(self.pg0, pkts, self.pg1)
+ ip_punt_redirect.remove_vpp_config()
+
+ def test_ip6_punt_dump(self):
+ """IPv6 punt redirect dump"""
+
+ #
+ # Configure a punt redirects
+ #
+ nh_address = self.pg3.remote_ip6
+ ipr_03 = VppIpPuntRedirect(
+ self, self.pg0.sw_if_index, self.pg3.sw_if_index, nh_address
+ )
+ ipr_13 = VppIpPuntRedirect(
+ self, self.pg1.sw_if_index, self.pg3.sw_if_index, nh_address
+ )
+ ipr_23 = VppIpPuntRedirect(
+ self, self.pg2.sw_if_index, self.pg3.sw_if_index, "::"
+ )
+ ipr_03.add_vpp_config()
+ ipr_13.add_vpp_config()
+ ipr_23.add_vpp_config()
+
+ #
+ # Dump pg0 punt redirects
+ #
+ self.assertTrue(ipr_03.query_vpp_config())
+ self.assertTrue(ipr_13.query_vpp_config())
+ self.assertTrue(ipr_23.query_vpp_config())
+
+ #
+ # Dump punt redirects for all interfaces
+ #
+ punts = self.vapi.ip_punt_redirect_dump(sw_if_index=0xFFFFFFFF, is_ipv6=True)
+ 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), "::")