X-Git-Url: https://gerrit.fd.io/r/gitweb?p=vpp.git;a=blobdiff_plain;f=test%2Ftest_ip6.py;h=77ae63bfa3966c3cf62dbf5db3716bd691bb99ee;hp=6c44d79a18f4aa30af7580e95b46f24e0434fc06;hb=8feeaff56;hpb=0bcad32b3870f9998fa1393418081cdda685272f diff --git a/test/test_ip6.py b/test/test_ip6.py index 6c44d79a18f..77ae63bfa39 100644 --- a/test/test_ip6.py +++ b/test/test_ip6.py @@ -3,6 +3,8 @@ import socket import unittest +from parameterized import parameterized +import scapy.compat import scapy.layers.inet6 as inet6 from scapy.contrib.mpls import MPLS from scapy.layers.inet6 import IPv6, ICMPv6ND_NS, ICMPv6ND_RS, \ @@ -17,7 +19,7 @@ from scapy.utils6 import in6_getnsma, in6_getnsmac, in6_ptop, in6_islladdr, \ from six import moves from framework import VppTestCase, VppTestRunner -from util import ppp, ip6_normalize +from util import ppp, ip6_normalize, mk_ll_addr from vpp_ip import DpoProto from vpp_ip_route import VppIpRoute, VppRoutePath, find_route, VppIpMRoute, \ VppMRoutePath, MRouteItfFlags, MRouteEntryFlags, VppMplsIpBind, \ @@ -25,14 +27,14 @@ from vpp_ip_route import VppIpRoute, VppRoutePath, find_route, VppIpMRoute, \ from vpp_neighbor import find_nbr, VppNeighbor from vpp_pg_interface import is_ipv6_misc from vpp_sub_interface import VppSubInterface, VppDot1QSubint +from ipaddress import IPv6Network, IPv4Network AF_INET6 = socket.AF_INET6 - -def mk_ll_addr(mac): - euid = in6_mactoifaceid(mac) - addr = "fe80::" + euid - return addr +try: + text_type = unicode +except NameError: + text_type = str class TestIPv6ND(VppTestCase): @@ -244,8 +246,10 @@ class TestIPv6(TestIPv6ND): for i in self.interfaces: next_hop_address = i.local_ip6n for j in range(count / n_int): - self.vapi.ip_add_del_route( - dest_addr, dest_addr_len, next_hop_address, is_ipv6=1) + self.vapi.ip_add_del_route(dst_address=dest_addr, + dst_address_length=dest_addr_len, + next_hop_address=next_hop_address, + is_ipv6=1) counter += 1 if counter / count * 100 > percent: self.logger.info("Configure %d FIB entries .. %d%% done" % @@ -317,7 +321,7 @@ class TestIPv6(TestIPv6ND): try: ip = packet[IPv6] udp = packet[inet6.UDP] - payload_info = self.payload_to_info(str(packet[Raw])) + payload_info = self.payload_to_info(packet[Raw]) packet_index = payload_info.index self.assertEqual(payload_info.dst, dst_sw_if_index) self.logger.debug( @@ -444,7 +448,6 @@ class TestIPv6(TestIPv6ND): self.pg0.sw_if_index, self.pg0.remote_hosts[2].mac, self.pg0.remote_hosts[2].ip6, - af=AF_INET6, is_no_fib_entry=1) nd_entry.add_vpp_config() @@ -453,8 +456,7 @@ class TestIPv6(TestIPv6ND): # self.assertTrue(find_nbr(self, self.pg0.sw_if_index, - self.pg0._remote_hosts[2].ip6, - inet=AF_INET6)) + self.pg0._remote_hosts[2].ip6)) self.assertFalse(find_route(self, self.pg0._remote_hosts[2].ip6, 128, @@ -482,8 +484,7 @@ class TestIPv6(TestIPv6ND): # self.assertTrue(find_nbr(self, self.pg0.sw_if_index, - self.pg0._remote_hosts[2].ip6_ll, - inet=AF_INET6)) + self.pg0._remote_hosts[2].ip6_ll)) self.assertFalse(find_route(self, self.pg0._remote_hosts[2].ip6_ll, 128, @@ -510,8 +511,7 @@ class TestIPv6(TestIPv6ND): # self.assertTrue(find_nbr(self, self.pg0.sw_if_index, - self.pg0._remote_hosts[3].ip6_ll, - inet=AF_INET6)) + self.pg0._remote_hosts[3].ip6_ll)) self.assertFalse(find_route(self, self.pg0._remote_hosts[3].ip6_ll, 128, @@ -531,14 +531,12 @@ class TestIPv6(TestIPv6ND): ns_pg1 = VppNeighbor(self, self.pg1.sw_if_index, self.pg1.remote_hosts[1].mac, - self.pg1.remote_hosts[1].ip6, - af=AF_INET6) + self.pg1.remote_hosts[1].ip6) 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) + self.pg1.remote_hosts[1].ip6) ns_pg2.add_vpp_config() # @@ -565,7 +563,7 @@ class TestIPv6(TestIPv6ND): # # remove the duplicate on pg1 - # packet stream shoud generate NSs out of pg1 + # packet stream should generate NSs out of pg1 # ns_pg1.remove_vpp_config() @@ -628,7 +626,7 @@ class TestIPv6(TestIPv6ND): # decipher how to decode. this 1st layer of option always returns # nested classes, so a direct obj1=obj2 comparison always fails. # however, the getlayer(.., 2) does give one instnace. - # so we cheat here and construct a new opt instnace for comparison + # so we cheat here and construct a new opt instance for comparison rd = ICMPv6NDOptPrefixInfo( prefixlen=raos.prefixlen, prefix=raos.prefix, @@ -687,7 +685,8 @@ class TestIPv6(TestIPv6ND): self.send_and_assert_no_replies(self.pg0, pkts, "RA rate limited") # - # When we reconfiure the IPv6 RA config, we reset the RA rate limiting, + # When we reconfigure the IPv6 RA config, + # we reset the RA rate limiting, # so we need to do this before each test below so as not to drop # packets for rate limiting reasons. Test this works here. # @@ -754,7 +753,7 @@ class TestIPv6(TestIPv6ND): # # Configure The RA to announce the links prefix # - self.pg0.ip6_ra_prefix(self.pg0.local_ip6n, + self.pg0.ip6_ra_prefix(self.pg0.local_ip6, self.pg0.local_ip6_prefix_len) # @@ -780,7 +779,7 @@ class TestIPv6(TestIPv6ND): # Change the prefix info to not off-link # L-flag is clear # - self.pg0.ip6_ra_prefix(self.pg0.local_ip6n, + self.pg0.ip6_ra_prefix(self.pg0.local_ip6, self.pg0.local_ip6_prefix_len, off_link=1) @@ -800,7 +799,7 @@ class TestIPv6(TestIPv6ND): # Change the prefix info to not off-link, no-autoconfig # L and A flag are clear in the advert # - self.pg0.ip6_ra_prefix(self.pg0.local_ip6n, + self.pg0.ip6_ra_prefix(self.pg0.local_ip6, self.pg0.local_ip6_prefix_len, off_link=1, no_autoconfig=1) @@ -821,7 +820,7 @@ class TestIPv6(TestIPv6ND): # Change the flag settings back to the defaults # L and A flag are set in the advert # - self.pg0.ip6_ra_prefix(self.pg0.local_ip6n, + self.pg0.ip6_ra_prefix(self.pg0.local_ip6, self.pg0.local_ip6_prefix_len) opt = ICMPv6NDOptPrefixInfo( @@ -840,7 +839,7 @@ class TestIPv6(TestIPv6ND): # Change the prefix info to not off-link, no-autoconfig # L and A flag are clear in the advert # - self.pg0.ip6_ra_prefix(self.pg0.local_ip6n, + self.pg0.ip6_ra_prefix(self.pg0.local_ip6, self.pg0.local_ip6_prefix_len, off_link=1, no_autoconfig=1) @@ -858,10 +857,10 @@ class TestIPv6(TestIPv6ND): opt=opt) # - # Use the reset to defults option to revert to defaults + # Use the reset to defaults option to revert to defaults # L and A flag are clear in the advert # - self.pg0.ip6_ra_prefix(self.pg0.local_ip6n, + self.pg0.ip6_ra_prefix(self.pg0.local_ip6, self.pg0.local_ip6_prefix_len, use_default=1) @@ -880,7 +879,7 @@ class TestIPv6(TestIPv6ND): # # Advertise Another prefix. With no L-flag/A-flag # - self.pg0.ip6_ra_prefix(self.pg1.local_ip6n, + self.pg0.ip6_ra_prefix(self.pg1.local_ip6, self.pg1.local_ip6_prefix_len, off_link=1, no_autoconfig=1) @@ -907,10 +906,10 @@ class TestIPv6(TestIPv6ND): opt=opt) # - # Remove the first refix-info - expect the second is still in the + # Remove the first prefix-info - expect the second is still in the # advert # - self.pg0.ip6_ra_prefix(self.pg0.local_ip6n, + self.pg0.ip6_ra_prefix(self.pg0.local_ip6, self.pg0.local_ip6_prefix_len, is_no=1) @@ -927,9 +926,9 @@ class TestIPv6(TestIPv6ND): opt=opt) # - # Remove the second prefix-info - expect no prefix-info i nthe adverts + # Remove the second prefix-info - expect no prefix-info in the adverts # - self.pg0.ip6_ra_prefix(self.pg1.local_ip6n, + self.pg0.ip6_ra_prefix(self.pg1.local_ip6, self.pg1.local_ip6_prefix_len, is_no=1) @@ -976,7 +975,7 @@ class TestICMPv6Echo(VppTestCase): icmpv6_id = 0xb icmpv6_seq = 5 - icmpv6_data = '\x0a' * 18 + icmpv6_data = b'\x0a' * 18 p_echo_request = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) / IPv6(src=self.pg0.remote_ip6, @@ -1063,11 +1062,13 @@ class TestIPv6RD(TestIPv6ND): self.pg1.local_mac) def verify_prefix_info(self, reported_prefix, prefix_option): - prefix = socket.inet_pton(socket.AF_INET6, - prefix_option.getfieldval("prefix")) - self.assert_equal(reported_prefix.dst_address, prefix) - self.assert_equal(reported_prefix.dst_address_length, - prefix_option.getfieldval("prefixlen")) + prefix = IPv6Network( + text_type(prefix_option.getfieldval("prefix") + + "/" + + text_type(prefix_option.getfieldval("prefixlen"))), + strict=False) + self.assert_equal(reported_prefix.prefix.network_address, + prefix.network_address) L = prefix_option.getfieldval("L") A = prefix_option.getfieldval("A") option_flags = (L << 7) | (A << 6) @@ -1166,10 +1167,10 @@ class TestIPv6RDControlPlane(TestIPv6ND): if entry.address_length == 0: for path in entry.path: if path.sw_if_index != 0xFFFFFFFF: - defaut_route = {} - defaut_route['sw_if_index'] = path.sw_if_index - defaut_route['next_hop'] = path.next_hop - list.append(defaut_route) + default_route = {} + default_route['sw_if_index'] = path.sw_if_index + default_route['next_hop'] = path.next_hop + list.append(default_route) return list @staticmethod @@ -1348,9 +1349,9 @@ class IPv6NDProxyTest(TestIPv6ND): # # Add proxy support for the host # - self.vapi.ip6_nd_proxy( - inet_pton(AF_INET6, self.pg0._remote_hosts[2].ip6), - self.pg1.sw_if_index) + self.vapi.ip6nd_proxy_add_del( + ip=inet_pton(AF_INET6, self.pg0._remote_hosts[2].ip6), + sw_if_index=self.pg1.sw_if_index) # # try that NS again. this time we expect an NA back @@ -1365,8 +1366,7 @@ class IPv6NDProxyTest(TestIPv6ND): # self.assertTrue(find_nbr(self, self.pg1.sw_if_index, - self.pg0._remote_hosts[2].ip6, - inet=AF_INET6)) + self.pg0._remote_hosts[2].ip6)) # # ... and we can route traffic to it @@ -1416,9 +1416,9 @@ class IPv6NDProxyTest(TestIPv6ND): ICMPv6NDOptSrcLLAddr( lladdr=self.pg0._remote_hosts[2].mac)) - self.vapi.ip6_nd_proxy( - inet_pton(AF_INET6, self.pg0._remote_hosts[3].ip6), - self.pg2.sw_if_index) + self.vapi.ip6nd_proxy_add_del( + ip=inet_pton(AF_INET6, self.pg0._remote_hosts[3].ip6), + sw_if_index=self.pg2.sw_if_index) self.send_and_expect_na(self.pg2, ns_pg2, "NS to proxy entry other interface", @@ -1427,8 +1427,7 @@ class IPv6NDProxyTest(TestIPv6ND): self.assertTrue(find_nbr(self, self.pg2.sw_if_index, - self.pg0._remote_hosts[3].ip6, - inet=AF_INET6)) + self.pg0._remote_hosts[3].ip6)) # # hosts can communicate. pg2->pg1 @@ -1457,23 +1456,19 @@ class IPv6NDProxyTest(TestIPv6ND): # # remove the proxy configs # - self.vapi.ip6_nd_proxy( - inet_pton(AF_INET6, self.pg0._remote_hosts[2].ip6), - self.pg1.sw_if_index, - is_del=1) - self.vapi.ip6_nd_proxy( - inet_pton(AF_INET6, self.pg0._remote_hosts[3].ip6), - self.pg2.sw_if_index, - is_del=1) + self.vapi.ip6nd_proxy_add_del( + ip=inet_pton(AF_INET6, self.pg0._remote_hosts[2].ip6), + sw_if_index=self.pg1.sw_if_index, is_del=1) + self.vapi.ip6nd_proxy_add_del( + ip=inet_pton(AF_INET6, self.pg0._remote_hosts[3].ip6), + sw_if_index=self.pg2.sw_if_index, is_del=1) self.assertFalse(find_nbr(self, self.pg2.sw_if_index, - self.pg0._remote_hosts[3].ip6, - inet=AF_INET6)) + self.pg0._remote_hosts[3].ip6)) self.assertFalse(find_nbr(self, self.pg1.sw_if_index, - self.pg0._remote_hosts[2].ip6, - inet=AF_INET6)) + self.pg0._remote_hosts[2].ip6)) # # no longer proxy-ing... @@ -1572,7 +1567,7 @@ class TestIPDisabled(VppTestCase): # create 2 pg interfaces self.create_pg_interfaces(range(2)) - # PG0 is IP enalbed + # PG0 is IP enabled self.pg0.admin_up() self.pg0.config_ip6() self.pg0.resolve_ndp() @@ -1676,20 +1671,20 @@ class TestIP6LoadBalance(VppTestCase): i.disable_mpls() super(TestIP6LoadBalance, self).tearDown() - def send_and_expect_load_balancing(self, input, pkts, outputs): + def pg_send(self, input, pkts): self.vapi.cli("clear trace") input.add_stream(pkts) self.pg_enable_capture(self.pg_interfaces) self.pg_start() + + def send_and_expect_load_balancing(self, input, pkts, outputs): + self.pg_send(input, pkts) for oo in outputs: rx = oo._get_capture(1) 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() + self.pg_send(input, pkts) rx = itf.get_capture(len(pkts)) def test_ip6_load_balance(self): @@ -1749,7 +1744,7 @@ class TestIP6LoadBalance(VppTestCase): src_ip_hdr)) # - # A route for the IP pacekts + # A route for the IP packets # route_3000_1 = VppIpRoute(self, "3000::1", 128, [VppRoutePath(self.pg1.remote_ip6, @@ -1787,7 +1782,7 @@ class TestIP6LoadBalance(VppTestCase): # src,dst # We are not going to ensure equal amounts of packets across each link, # since the hash algorithm is statistical and therefore this can never - # be guaranteed. But wuth 64 different packets we do expect some + # be guaranteed. But with 64 different packets we do expect some # balancing. So instead just ensure there is traffic on each link. # self.send_and_expect_load_balancing(self.pg0, port_ip_pkts, @@ -1803,7 +1798,7 @@ class TestIP6LoadBalance(VppTestCase): # # The packets with Entropy label in should not load-balance, - # since the Entorpy value is fixed. + # since the Entropy value is fixed. # self.send_and_expect_one_itf(self.pg0, port_ent_pkts, self.pg1) @@ -1812,7 +1807,8 @@ class TestIP6LoadBalance(VppTestCase): # - now only the stream with differing source address will # load-balance # - self.vapi.set_ip_flow_hash(0, is_ip6=1, src=1, dst=1, sport=0, dport=0) + self.vapi.set_ip_flow_hash(vrf_id=0, src=1, dst=1, sport=0, dport=0, + is_ipv6=1) self.send_and_expect_load_balancing(self.pg0, src_ip_pkts, [self.pg1, self.pg2]) @@ -1823,7 +1819,8 @@ class TestIP6LoadBalance(VppTestCase): # # change the flow hash config back to defaults # - self.vapi.set_ip_flow_hash(0, is_ip6=1, src=1, dst=1, sport=1, dport=1) + self.vapi.set_ip_flow_hash(vrf_id=0, src=1, dst=1, sport=1, dport=1, + is_ipv6=1) # # Recursive prefixes @@ -1959,7 +1956,7 @@ class TestIP6Punt(VppTestCase): # # add a policer # - policer = self.vapi.policer_add_del("ip6-punt", 400, 0, 10, 0, + policer = self.vapi.policer_add_del(b"ip6-punt", 400, 0, 10, 0, rate_type=1) self.vapi.ip_punt_police(policer.policer_index, is_ip6=1) @@ -1969,7 +1966,7 @@ class TestIP6Punt(VppTestCase): self.pg_start() # - # the number of packet recieved should be greater than 0, + # 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) @@ -1977,10 +1974,10 @@ class TestIP6Punt(VppTestCase): self.assertLess(len(rx), len(pkts)) # - # remove the poilcer. back to full rx + # remove the policer. 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, + self.vapi.policer_add_del(b"ip6-punt", 400, 0, 10, 0, rate_type=1, is_add=0) self.send_and_expect(self.pg0, pkts, self.pg1) @@ -2163,7 +2160,7 @@ class TestIPDeag(VppTestCase): class TestIP6Input(VppTestCase): - """ IPv6 Input Exceptions """ + """ IPv6 Input Exception Test Cases """ def setUp(self): super(TestIP6Input, self).setUp() @@ -2181,25 +2178,10 @@ class TestIP6Input(VppTestCase): i.unconfig_ip6() i.admin_down() - 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) / - inet6.UDP(sport=1234, dport=1234) / - Raw('\xa5' * 100)) - - self.send_and_assert_no_replies(self.pg0, p_version * 65, - "funky version") - + def test_ip_input_icmp_reply(self): + """ IP6 Input Exception - Return ICMP (3,0) """ # - # hop limit - IMCP replies + # hop limit - ICMP replies # p_version = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) / @@ -2212,9 +2194,45 @@ class TestIP6Input(VppTestCase): 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) + 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('\xa5' * 100)) + + self.send_and_assert_no_replies(self.pg0, p_version * 65, + remark=msg or "", + timeout=timeout) if __name__ == '__main__':