X-Git-Url: https://gerrit.fd.io/r/gitweb?a=blobdiff_plain;f=test%2Ftest_ip6.py;h=684eff5546ec13ed22ab7aa8820c894a11310072;hb=62fe07c8e30a6ac718fe65528592fe6964d753fa;hp=700b3344a9e538ae8d69a9f0ed465a58ddf6f2a2;hpb=71275e3d1ed4b7a536b7ec8d13995743beccde6b;p=vpp.git diff --git a/test/test_ip6.py b/test/test_ip6.py index 700b3344a9e..684eff5546e 100644 --- a/test/test_ip6.py +++ b/test/test_ip6.py @@ -8,15 +8,16 @@ from vpp_sub_interface import VppSubInterface, VppDot1QSubint from vpp_pg_interface import is_ipv6_misc from vpp_ip_route import VppIpRoute, VppRoutePath, find_route, VppIpMRoute, \ VppMRoutePath, MRouteItfFlags, MRouteEntryFlags, VppMplsIpBind, \ - VppMplsRoute + VppMplsRoute, DpoProto, VppMplsTable from vpp_neighbor import find_nbr, VppNeighbor from scapy.packet import Raw from scapy.layers.l2 import Ether, Dot1Q -from scapy.layers.inet6 import IPv6, UDP, ICMPv6ND_NS, ICMPv6ND_RS, \ +from scapy.layers.inet6 import IPv6, UDP, TCP, ICMPv6ND_NS, ICMPv6ND_RS, \ ICMPv6ND_RA, ICMPv6NDOptSrcLLAddr, getmacbyip6, ICMPv6MRD_Solicitation, \ ICMPv6NDOptMTU, ICMPv6NDOptSrcLLAddr, ICMPv6NDOptPrefixInfo, \ - ICMPv6ND_NA, ICMPv6NDOptDstLLAddr, ICMPv6DestUnreach, icmp6types + ICMPv6ND_NA, ICMPv6NDOptDstLLAddr, ICMPv6DestUnreach, icmp6types, \ + ICMPv6TimeExceeded from util import ppp from scapy.utils6 import in6_getnsma, in6_getnsmac, in6_ptop, in6_islladdr, \ @@ -76,6 +77,31 @@ class TestIPv6ND(VppTestCase): dll = rx[ICMPv6NDOptDstLLAddr] self.assertEqual(dll.lladdr, intf.local_mac) + def validate_ns(self, intf, rx, tgt_ip): + nsma = in6_getnsma(inet_pton(AF_INET6, tgt_ip)) + dst_ip = inet_ntop(AF_INET6, nsma) + + # NS is broadcast + self.assertEqual(rx[Ether].dst, "ff:ff:ff:ff:ff:ff") + + # and from the router's MAC + self.assertEqual(rx[Ether].src, intf.local_mac) + + # the rx'd NS should be addressed to an mcast address + # derived from the target address + self.assertEqual(in6_ptop(rx[IPv6].dst), in6_ptop(dst_ip)) + + # expect the tgt IP in the NS header + ns = rx[ICMPv6ND_NS] + self.assertEqual(in6_ptop(ns.tgt), in6_ptop(tgt_ip)) + + # packet is from the router's local address + self.assertEqual(in6_ptop(rx[IPv6].src), intf.local_ip6) + + # Src link-layer options should have the router's MAC + sll = rx[ICMPv6NDOptSrcLLAddr] + self.assertEqual(sll.lladdr, intf.local_mac) + def send_and_expect_ra(self, intf, pkts, remark, dst_ip=None, filter_out_fn=is_ipv6_misc): intf.add_stream(pkts) @@ -99,6 +125,17 @@ class TestIPv6ND(VppTestCase): rx = rx[0] self.validate_na(intf, rx, dst_ip, tgt_ip) + def send_and_expect_ns(self, tx_intf, rx_intf, pkts, tgt_ip, + filter_out_fn=is_ipv6_misc): + tx_intf.add_stream(pkts) + self.pg_enable_capture(self.pg_interfaces) + self.pg_start() + rx = rx_intf.get_capture(1, filter_out_fn=filter_out_fn) + + self.assertEqual(len(rx), 1) + rx = rx[0] + self.validate_ns(rx_intf, rx, tgt_ip) + def send_and_assert_no_replies(self, intf, pkts, remark): intf.add_stream(pkts) self.pg_enable_capture(self.pg_interfaces) @@ -107,6 +144,15 @@ class TestIPv6ND(VppTestCase): i.get_capture(0) i.assert_nothing_captured(remark=remark) + def verify_ip(self, rx, smac, dmac, sip, dip): + ether = rx[Ether] + self.assertEqual(ether.dst, dmac) + self.assertEqual(ether.src, smac) + + ip = rx[IPv6] + self.assertEqual(ip.src, sip) + self.assertEqual(ip.dst, dip) + class TestIPv6(TestIPv6ND): """ IPv6 Test Case """ @@ -444,6 +490,78 @@ class TestIPv6(TestIPv6ND): 128, inet=AF_INET6)) + def test_ns_duplicates(self): + """ ND Duplicates""" + + # + # Generate some hosts on the LAN + # + self.pg1.generate_remote_hosts(3) + + # + # Add host 1 on pg1 and pg2 + # + ns_pg1 = VppNeighbor(self, + self.pg1.sw_if_index, + self.pg1.remote_hosts[1].mac, + self.pg1.remote_hosts[1].ip6, + af=AF_INET6) + ns_pg1.add_vpp_config() + ns_pg2 = VppNeighbor(self, + self.pg2.sw_if_index, + self.pg2.remote_mac, + self.pg1.remote_hosts[1].ip6, + af=AF_INET6) + ns_pg2.add_vpp_config() + + # + # IP packet destined for pg1 remote host arrives on pg1 again. + # + p = (Ether(dst=self.pg0.local_mac, + src=self.pg0.remote_mac) / + IPv6(src=self.pg0.remote_ip6, + dst=self.pg1.remote_hosts[1].ip6) / + UDP(sport=1234, dport=1234) / + Raw()) + + self.pg0.add_stream(p) + self.pg_enable_capture(self.pg_interfaces) + self.pg_start() + + rx1 = self.pg1.get_capture(1) + + self.verify_ip(rx1[0], + self.pg1.local_mac, + self.pg1.remote_hosts[1].mac, + self.pg0.remote_ip6, + self.pg1.remote_hosts[1].ip6) + + # + # remove the duplicate on pg1 + # packet stream shoud generate NSs out of pg1 + # + ns_pg1.remove_vpp_config() + + self.send_and_expect_ns(self.pg0, self.pg1, + p, self.pg1.remote_hosts[1].ip6) + + # + # Add it back + # + ns_pg1.add_vpp_config() + + self.pg0.add_stream(p) + self.pg_enable_capture(self.pg_interfaces) + self.pg_start() + + rx1 = self.pg1.get_capture(1) + + self.verify_ip(rx1[0], + self.pg1.local_mac, + self.pg1.remote_hosts[1].mac, + self.pg0.remote_ip6, + self.pg1.remote_hosts[1].ip6) + def validate_ra(self, intf, rx, dst_ip=None, mtu=9000, pi_opt=None): if not dst_ip: dst_ip = intf.remote_ip6 @@ -1143,6 +1261,9 @@ class TestIP6LoadBalance(VppTestCase): self.create_pg_interfaces(range(5)) + mpls_tbl = VppMplsTable(self, 0) + mpls_tbl.add_vpp_config() + for i in self.pg_interfaces: i.admin_up() i.config_ip6() @@ -1150,13 +1271,14 @@ class TestIP6LoadBalance(VppTestCase): i.enable_mpls() def tearDown(self): - super(TestIP6LoadBalance, self).tearDown() for i in self.pg_interfaces: i.unconfig_ip6() i.admin_down() i.disable_mpls() + super(TestIP6LoadBalance, self).tearDown() def send_and_expect_load_balancing(self, input, pkts, outputs): + self.vapi.cli("clear trace") input.add_stream(pkts) self.pg_enable_capture(self.pg_interfaces) self.pg_start() @@ -1165,6 +1287,7 @@ class TestIP6LoadBalance(VppTestCase): self.assertNotEqual(0, len(rx)) def send_and_expect_one_itf(self, input, pkts, itf): + self.vapi.cli("clear trace") input.add_stream(pkts) self.pg_enable_capture(self.pg_interfaces) self.pg_start() @@ -1230,10 +1353,10 @@ class TestIP6LoadBalance(VppTestCase): route_3000_1 = VppIpRoute(self, "3000::1", 128, [VppRoutePath(self.pg1.remote_ip6, self.pg1.sw_if_index, - is_ip6=1), + proto=DpoProto.DPO_PROTO_IP6), VppRoutePath(self.pg2.remote_ip6, self.pg2.sw_if_index, - is_ip6=1)], + proto=DpoProto.DPO_PROTO_IP6)], is_ip6=1) route_3000_1.add_vpp_config() @@ -1250,11 +1373,11 @@ class TestIP6LoadBalance(VppTestCase): [VppRoutePath(self.pg1.remote_ip6, self.pg1.sw_if_index, labels=[67], - is_ip6=1), + proto=DpoProto.DPO_PROTO_IP6), VppRoutePath(self.pg2.remote_ip6, self.pg2.sw_if_index, labels=[67], - is_ip6=1)]) + proto=DpoProto.DPO_PROTO_IP6)]) route_67.add_vpp_config() # @@ -1324,20 +1447,20 @@ class TestIP6LoadBalance(VppTestCase): route_3000_2 = VppIpRoute(self, "3000::2", 128, [VppRoutePath(self.pg3.remote_ip6, self.pg3.sw_if_index, - is_ip6=1), + proto=DpoProto.DPO_PROTO_IP6), VppRoutePath(self.pg4.remote_ip6, self.pg4.sw_if_index, - is_ip6=1)], + proto=DpoProto.DPO_PROTO_IP6)], is_ip6=1) route_3000_2.add_vpp_config() route_4000_1 = VppIpRoute(self, "4000::1", 128, [VppRoutePath("3000::1", 0xffffffff, - is_ip6=1), + proto=DpoProto.DPO_PROTO_IP6), VppRoutePath("3000::2", 0xffffffff, - is_ip6=1)], + proto=DpoProto.DPO_PROTO_IP6)], is_ip6=1) route_4000_1.add_vpp_config() @@ -1352,6 +1475,224 @@ class TestIP6LoadBalance(VppTestCase): [self.pg1, self.pg2, self.pg3, self.pg4]) + # + # Recursive prefixes + # - testing that 2 stages of load-balancing no choices + # + port_pkts = [] + + for ii in range(257): + port_pkts.append((Ether(src=self.pg0.remote_mac, + dst=self.pg0.local_mac) / + IPv6(dst="6000::1", src="6000:1::1") / + UDP(sport=1234, dport=1234 + ii) / + Raw('\xa5' * 100))) + + route_5000_2 = VppIpRoute(self, "5000::2", 128, + [VppRoutePath(self.pg3.remote_ip6, + self.pg3.sw_if_index, + proto=DpoProto.DPO_PROTO_IP6)], + is_ip6=1) + route_5000_2.add_vpp_config() + + route_6000_1 = VppIpRoute(self, "6000::1", 128, + [VppRoutePath("5000::2", + 0xffffffff, + proto=DpoProto.DPO_PROTO_IP6)], + is_ip6=1) + route_6000_1.add_vpp_config() + + # + # inject the packet on pg0 - expect load-balancing across all 4 paths + # + self.vapi.cli("clear trace") + self.send_and_expect_one_itf(self.pg0, port_pkts, self.pg3) + + +class TestIP6Punt(VppTestCase): + """ IPv6 Punt Police/Redirect """ + + def setUp(self): + super(TestIP6Punt, 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(TestIP6Punt, self).tearDown() + for i in self.pg_interfaces: + i.unconfig_ip6() + i.admin_down() + + def send_and_expect(self, input, pkts, output): + input.add_stream(pkts) + self.pg_enable_capture(self.pg_interfaces) + self.pg_start() + rx = output.get_capture(len(pkts)) + return rx + + def send_and_assert_no_replies(self, intf, pkts, remark): + self.vapi.cli("clear trace") + intf.add_stream(pkts) + self.pg_enable_capture(self.pg_interfaces) + self.pg_start() + for i in self.pg_interfaces: + i.get_capture(0) + i.assert_nothing_captured(remark=remark) + + 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) / + TCP(sport=1234, dport=1234) / + Raw('\xa5' * 100)) + + pkts = p * 1025 + + # + # Configure a punt redirect via pg1. + # + nh_addr = inet_pton(AF_INET6, + self.pg1.remote_ip6) + self.vapi.ip_punt_redirect(self.pg0.sw_if_index, + self.pg1.sw_if_index, + nh_addr, + is_ip6=1) + + self.send_and_expect(self.pg0, pkts, self.pg1) + + # + # add a policer + # + policer = self.vapi.policer_add_del("ip6-punt", 400, 0, 10, 0, + rate_type=1) + 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 recieved should be greater than 0, + # but not equal to the number sent, since some were policed + # + rx = self.pg1._get_capture(1) + self.assertTrue(len(rx) > 0) + self.assertTrue(len(rx) < len(pkts)) + + # + # remove the poilcer. back to full rx + # + self.vapi.ip_punt_police(policer.policer_index, is_add=0, is_ip6=1) + self.vapi.policer_add_del("ip6-punt", 400, 0, 10, 0, + rate_type=1, is_add=0) + 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, + is_ip6=1) + 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, + is_ip6=1) + 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, + is_ip6=1) + + +class TestIP6Input(VppTestCase): + """ IPv6 Input Exceptions """ + + 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 send_and_expect(self, input, pkts, output): + self.vapi.cli("clear trace") + input.add_stream(pkts) + self.pg_enable_capture(self.pg_interfaces) + self.pg_start() + rx = output.get_capture(len(pkts)) + return rx + + def send_and_assert_no_replies(self, intf, pkts, remark): + self.vapi.cli("clear trace") + intf.add_stream(pkts) + self.pg_enable_capture(self.pg_interfaces) + self.pg_start() + for i in self.pg_interfaces: + i.get_capture(0) + i.assert_nothing_captured(remark=remark) + + def test_ip_input(self): + """ IP6 Input Exceptions """ + + # + # bad version - this is dropped + # + p_version = (Ether(src=self.pg0.remote_mac, + dst=self.pg0.local_mac) / + IPv6(src=self.pg0.remote_ip6, + dst=self.pg1.remote_ip6, + version=3) / + UDP(sport=1234, dport=1234) / + Raw('\xa5' * 100)) + + self.send_and_assert_no_replies(self.pg0, p_version * 65, + "funky version") + + # + # hop limit - IMCP 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) / + UDP(sport=1234, dport=1234) / + Raw('\xa5' * 100)) + + rx = self.send_and_expect(self.pg0, p_version * 65, self.pg0) + rx = rx[0] + icmp = rx[ICMPv6TimeExceeded] + self.assertEqual(icmp.type, 3) + # 0: "hop limit exceeded in transit", + self.assertEqual(icmp.code, 0) + if __name__ == '__main__': unittest.main(testRunner=VppTestRunner)