X-Git-Url: https://gerrit.fd.io/r/gitweb?a=blobdiff_plain;f=test%2Ftest_ip6.py;h=d1fb86d91e3a455a6261611776184dac2c7742b3;hb=23ff4ce21e71b697fb059ea82857413d993db9d0;hp=a9244bd3d60b1ba0a312a484d8f89d3ee2b55f37;hpb=a1f360647f1c89908294bf06667fba60f4369e8b;p=vpp.git diff --git a/test/test_ip6.py b/test/test_ip6.py index a9244bd3d60..d1fb86d91e3 100644 --- a/test/test_ip6.py +++ b/test/test_ip6.py @@ -7,13 +7,14 @@ import unittest from parameterized import parameterized import scapy.compat import scapy.layers.inet6 as inet6 +from scapy.layers.inet import UDP, IP from scapy.contrib.mpls import MPLS from scapy.layers.inet6 import IPv6, ICMPv6ND_NS, ICMPv6ND_RS, \ ICMPv6ND_RA, ICMPv6NDOptMTU, ICMPv6NDOptSrcLLAddr, ICMPv6NDOptPrefixInfo, \ ICMPv6ND_NA, ICMPv6NDOptDstLLAddr, ICMPv6DestUnreach, icmp6types, \ ICMPv6TimeExceeded, ICMPv6EchoRequest, ICMPv6EchoReply, \ IPv6ExtHdrHopByHop, ICMPv6MLReport2, ICMPv6MLDMultAddrRec -from scapy.layers.l2 import Ether, Dot1Q +from scapy.layers.l2 import Ether, Dot1Q, GRE from scapy.packet import Raw from scapy.utils6 import in6_getnsma, in6_getnsmac, in6_ptop, in6_islladdr, \ in6_mactoifaceid @@ -22,17 +23,20 @@ from six import moves from framework import VppTestCase, VppTestRunner, tag_run_solo from util import ppp, ip6_normalize, mk_ll_addr from vpp_papi import VppEnum -from vpp_ip import DpoProto, VppIpPuntPolicer, VppIpPuntRedirect +from vpp_ip import DpoProto, VppIpPuntPolicer, VppIpPuntRedirect, VppIpPathMtu from vpp_ip_route import VppIpRoute, VppRoutePath, find_route, VppIpMRoute, \ VppMRoutePath, VppMplsIpBind, \ VppMplsRoute, VppMplsTable, VppIpTable, FibPathType, FibPathProto, \ VppIpInterfaceAddress, find_route_in_dump, find_mroute_in_dump, \ - VppIp6LinkLocalAddress + VppIp6LinkLocalAddress, VppIpRouteV2 from vpp_neighbor import find_nbr, VppNeighbor +from vpp_ipip_tun_interface import VppIpIpTunInterface from vpp_pg_interface import is_ipv6_misc from vpp_sub_interface import VppSubInterface, VppDot1QSubint -from vpp_policer import VppPolicer +from vpp_policer import VppPolicer, PolicerAction from ipaddress import IPv6Network, IPv6Address +from vpp_gre_interface import VppGreInterface +from vpp_teib import VppTeib AF_INET6 = socket.AF_INET6 @@ -502,6 +506,18 @@ class TestIPv6(TestIPv6ND): dst_ip=self.pg0._remote_hosts[3].ip6_ll, tgt_ip=self.pg0.local_ip6_ll) + # + # do not respond to a NS for the peer's address + # + p = (Ether(dst=in6_getnsmac(nsma), src=self.pg0.remote_mac) / + IPv6(dst=d, + src=self.pg0._remote_hosts[3].ip6_ll) / + ICMPv6ND_NS(tgt=self.pg0._remote_hosts[3].ip6_ll) / + ICMPv6NDOptSrcLLAddr( + lladdr=self.pg0.remote_mac)) + + self.send_and_assert_no_replies(self.pg0, p) + # # we should have learned an ND entry for the peer's link-local # but not inserted a route to it in the FIB @@ -719,9 +735,23 @@ class TestIPv6(TestIPv6ND): "RS sourced from link-local", dst_ip=ll) + # + # Source an RS from a link local address + # Ensure suppress also applies to solicited RS + # + self.pg0.ip6_ra_config(send_unicast=1, suppress=1) + ll = mk_ll_addr(self.pg0.remote_mac) + p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) / + IPv6(dst=self.pg0.local_ip6, src=ll) / + ICMPv6ND_RS()) + pkts = [p] + self.send_and_assert_no_replies(self.pg0, pkts, + "Suppressed RS from link-local") + # # Send the RS multicast # + self.pg0.ip6_ra_config(no=1, suppress=1) # Reset suppress flag to zero self.pg0.ip6_ra_config(send_unicast=1) dmac = in6_getnsmac(inet_pton(AF_INET6, "ff02::2")) ll = mk_ll_addr(self.pg0.remote_mac) @@ -741,7 +771,7 @@ class TestIPv6(TestIPv6ND): # If we happen to pick up the periodic RA at this point then so be it, # it's not an error. # - self.pg0.ip6_ra_config(send_unicast=1, suppress=1) + self.pg0.ip6_ra_config(send_unicast=1, suppress=0) p = (Ether(dst=dmac, src=self.pg0.remote_mac) / IPv6(dst="ff02::2", src="::") / ICMPv6ND_RS()) @@ -2248,6 +2278,13 @@ class TestIP6Punt(IP6PuntSetup, VppTestCase): # 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)) @@ -2309,6 +2346,119 @@ class TestIP6Punt(IP6PuntSetup, VppTestCase): 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 TestIPDeag(VppTestCase): """ IPv6 Deaggregate Routes """ @@ -2868,6 +3018,11 @@ class TestIP6LinkLocal(VppTestCase): 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() # @@ -2915,6 +3070,676 @@ class TestIP6LinkLocal(VppTestCase): 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_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, 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), '::') + if __name__ == '__main__': unittest.main(testRunner=VppTestRunner)