test/test_ip6.py

   1 #!/usr/bin/env python
   2
   3 import unittest
   4 import socket
   5
   6 from framework import VppTestCase, VppTestRunner
   7 from vpp_sub_interface import VppSubInterface, VppDot1QSubint
   8 from vpp_pg_interface import is_ipv6_misc
   9
  10 from scapy.packet import Raw
  11 from scapy.layers.l2 import Ether, Dot1Q
  12 from scapy.layers.inet6 import IPv6, UDP, ICMPv6ND_NS, ICMPv6ND_RS, \
  13     ICMPv6ND_RA, ICMPv6NDOptSrcLLAddr, getmacbyip6, ICMPv6MRD_Solicitation
  14 from util import ppp
  15 from scapy.utils6 import in6_getnsma, in6_getnsmac, in6_ptop, in6_islladdr, \
  16     in6_mactoifaceid, in6_ismaddr
  17 from scapy.utils import inet_pton, inet_ntop
  18
  19
  20 def mk_ll_addr(mac):
  21     euid = in6_mactoifaceid(mac)
  22     addr = "fe80::" + euid
  23     return addr
  24
  25
  26 class TestIPv6(VppTestCase):
  27     """ IPv6 Test Case """
  28
  29     @classmethod
  30     def setUpClass(cls):
  31         super(TestIPv6, cls).setUpClass()
  32
  33     def setUp(self):
  34         """
  35         Perform test setup before test case.
  36
  37         **Config:**
  38             - create 3 pg interfaces
  39                 - untagged pg0 interface
  40                 - Dot1Q subinterface on pg1
  41                 - Dot1AD subinterface on pg2
  42             - setup interfaces:
  43                 - put it into UP state
  44                 - set IPv6 addresses
  45                 - resolve neighbor address using NDP
  46             - configure 200 fib entries
  47
  48         :ivar list interfaces: pg interfaces and subinterfaces.
  49         :ivar dict flows: IPv4 packet flows in test.
  50         :ivar list pg_if_packet_sizes: packet sizes in test.
  51
  52         *TODO:* Create AD sub interface
  53         """
  54         super(TestIPv6, self).setUp()
  55
  56         # create 3 pg interfaces
  57         self.create_pg_interfaces(range(3))
  58
  59         # create 2 subinterfaces for p1 and pg2
  60         self.sub_interfaces = [
  61             VppDot1QSubint(self, self.pg1, 100),
  62             VppDot1QSubint(self, self.pg2, 200)
  63             # TODO: VppDot1ADSubint(self, self.pg2, 200, 300, 400)
  64         ]
  65
  66         # packet flows mapping pg0 -> pg1.sub, pg2.sub, etc.
  67         self.flows = dict()
  68         self.flows[self.pg0] = [self.pg1.sub_if, self.pg2.sub_if]
  69         self.flows[self.pg1.sub_if] = [self.pg0, self.pg2.sub_if]
  70         self.flows[self.pg2.sub_if] = [self.pg0, self.pg1.sub_if]
  71
  72         # packet sizes
  73         self.pg_if_packet_sizes = [64, 512, 1518, 9018]
  74         self.sub_if_packet_sizes = [64, 512, 1518 + 4, 9018 + 4]
  75
  76         self.interfaces = list(self.pg_interfaces)
  77         self.interfaces.extend(self.sub_interfaces)
  78
  79         # setup all interfaces
  80         for i in self.interfaces:
  81             i.admin_up()
  82             i.config_ip6()
  83             i.resolve_ndp()
  84
  85         # config 2M FIB entries
  86         self.config_fib_entries(200)
  87
  88     def tearDown(self):
  89         """Run standard test teardown and log ``show ip6 neighbors``."""
  90         for i in self.sub_interfaces:
  91             i.unconfig_ip6()
  92             i.ip6_disable()
  93             i.admin_down()
  94             i.remove_vpp_config()
  95
  96         super(TestIPv6, self).tearDown()
  97         if not self.vpp_dead:
  98             self.logger.info(self.vapi.cli("show ip6 neighbors"))
  99             # info(self.vapi.cli("show ip6 fib"))  # many entries
 100
 101     def config_fib_entries(self, count):
 102         """For each interface add to the FIB table *count* routes to
 103         "fd02::1/128" destination with interface's local address as next-hop
 104         address.
 105
 106         :param int count: Number of FIB entries.
 107
 108         - *TODO:* check if the next-hop address shouldn't be remote address
 109           instead of local address.
 110         """
 111         n_int = len(self.interfaces)
 112         percent = 0
 113         counter = 0.0
 114         dest_addr = socket.inet_pton(socket.AF_INET6, "fd02::1")
 115         dest_addr_len = 128
 116         for i in self.interfaces:
 117             next_hop_address = i.local_ip6n
 118             for j in range(count / n_int):
 119                 self.vapi.ip_add_del_route(
 120                     dest_addr, dest_addr_len, next_hop_address, is_ipv6=1)
 121                 counter += 1
 122                 if counter / count * 100 > percent:
 123                     self.logger.info("Configure %d FIB entries .. %d%% done" %
 124                                      (count, percent))
 125                     percent += 1
 126
 127     def create_stream(self, src_if, packet_sizes):
 128         """Create input packet stream for defined interface.
 129
 130         :param VppInterface src_if: Interface to create packet stream for.
 131         :param list packet_sizes: Required packet sizes.
 132         """
 133         pkts = []
 134         for i in range(0, 257):
 135             dst_if = self.flows[src_if][i % 2]
 136             info = self.create_packet_info(src_if, dst_if)
 137             payload = self.info_to_payload(info)
 138             p = (Ether(dst=src_if.local_mac, src=src_if.remote_mac) /
 139                  IPv6(src=src_if.remote_ip6, dst=dst_if.remote_ip6) /
 140                  UDP(sport=1234, dport=1234) /
 141                  Raw(payload))
 142             info.data = p.copy()
 143             if isinstance(src_if, VppSubInterface):
 144                 p = src_if.add_dot1_layer(p)
 145             size = packet_sizes[(i // 2) % len(packet_sizes)]
 146             self.extend_packet(p, size)
 147             pkts.append(p)
 148         return pkts
 149
 150     def verify_capture(self, dst_if, capture):
 151         """Verify captured input packet stream for defined interface.
 152
 153         :param VppInterface dst_if: Interface to verify captured packet stream
 154                                     for.
 155         :param list capture: Captured packet stream.
 156         """
 157         self.logger.info("Verifying capture on interface %s" % dst_if.name)
 158         last_info = dict()
 159         for i in self.interfaces:
 160             last_info[i.sw_if_index] = None
 161         is_sub_if = False
 162         dst_sw_if_index = dst_if.sw_if_index
 163         if hasattr(dst_if, 'parent'):
 164             is_sub_if = True
 165         for packet in capture:
 166             if is_sub_if:
 167                 # Check VLAN tags and Ethernet header
 168                 packet = dst_if.remove_dot1_layer(packet)
 169             self.assertTrue(Dot1Q not in packet)
 170             try:
 171                 ip = packet[IPv6]
 172                 udp = packet[UDP]
 173                 payload_info = self.payload_to_info(str(packet[Raw]))
 174                 packet_index = payload_info.index
 175                 self.assertEqual(payload_info.dst, dst_sw_if_index)
 176                 self.logger.debug(
 177                     "Got packet on port %s: src=%u (id=%u)" %
 178                     (dst_if.name, payload_info.src, packet_index))
 179                 next_info = self.get_next_packet_info_for_interface2(
 180                     payload_info.src, dst_sw_if_index,
 181                     last_info[payload_info.src])
 182                 last_info[payload_info.src] = next_info
 183                 self.assertTrue(next_info is not None)
 184                 self.assertEqual(packet_index, next_info.index)
 185                 saved_packet = next_info.data
 186                 # Check standard fields
 187                 self.assertEqual(ip.src, saved_packet[IPv6].src)
 188                 self.assertEqual(ip.dst, saved_packet[IPv6].dst)
 189                 self.assertEqual(udp.sport, saved_packet[UDP].sport)
 190                 self.assertEqual(udp.dport, saved_packet[UDP].dport)
 191             except:
 192                 self.logger.error(ppp("Unexpected or invalid packet:", packet))
 193                 raise
 194         for i in self.interfaces:
 195             remaining_packet = self.get_next_packet_info_for_interface2(
 196                 i.sw_if_index, dst_sw_if_index, last_info[i.sw_if_index])
 197             self.assertTrue(remaining_packet is None,
 198                             "Interface %s: Packet expected from interface %s "
 199                             "didn't arrive" % (dst_if.name, i.name))
 200
 201     def test_fib(self):
 202         """ IPv6 FIB test
 203
 204         Test scenario:
 205             - Create IPv6 stream for pg0 interface
 206             - Create IPv6 tagged streams for pg1's and pg2's subinterface.
 207             - Send and verify received packets on each interface.
 208         """
 209
 210         pkts = self.create_stream(self.pg0, self.pg_if_packet_sizes)
 211         self.pg0.add_stream(pkts)
 212
 213         for i in self.sub_interfaces:
 214             pkts = self.create_stream(i, self.sub_if_packet_sizes)
 215             i.parent.add_stream(pkts)
 216
 217         self.pg_enable_capture(self.pg_interfaces)
 218         self.pg_start()
 219
 220         pkts = self.pg0.get_capture()
 221         self.verify_capture(self.pg0, pkts)
 222
 223         for i in self.sub_interfaces:
 224             pkts = i.parent.get_capture()
 225             self.verify_capture(i, pkts)
 226
 227     def send_and_assert_no_replies(self, intf, pkts, remark):
 228         intf.add_stream(pkts)
 229         self.pg_enable_capture(self.pg_interfaces)
 230         self.pg_start()
 231         intf.assert_nothing_captured(remark=remark)
 232
 233     def test_ns(self):
 234         """ IPv6 Neighbour Solicitation Exceptions
 235
 236         Test scenario:
 237            - Send an NS Sourced from an address not covered by the link sub-net
 238            - Send an NS to an mcast address the router has not joined
 239            - Send NS for a target address the router does not onn.
 240         """
 241
 242         #
 243         # An NS from a non link source address
 244         #
 245         nsma = in6_getnsma(inet_pton(socket.AF_INET6, self.pg0.local_ip6))
 246         d = inet_ntop(socket.AF_INET6, nsma)
 247
 248         p = (Ether(dst=in6_getnsmac(nsma)) /
 249              IPv6(dst=d, src="2002::2") /
 250              ICMPv6ND_NS(tgt=self.pg0.local_ip6) /
 251              ICMPv6NDOptSrcLLAddr(lladdr=self.pg0.remote_mac))
 252         pkts = [p]
 253
 254         self.send_and_assert_no_replies(
 255             self.pg0, pkts,
 256             "No response to NS source by address not on sub-net")
 257
 258         #
 259         # An NS for sent to a solicited mcast group the router is
 260         # not a member of FAILS
 261         #
 262         if 0:
 263             nsma = in6_getnsma(inet_pton(socket.AF_INET6, "fd::ffff"))
 264             d = inet_ntop(socket.AF_INET6, nsma)
 265
 266             p = (Ether(dst=in6_getnsmac(nsma)) /
 267                  IPv6(dst=d, src=self.pg0.remote_ip6) /
 268                  ICMPv6ND_NS(tgt=self.pg0.local_ip6) /
 269                  ICMPv6NDOptSrcLLAddr(lladdr=self.pg0.remote_mac))
 270             pkts = [p]
 271
 272             self.send_and_assert_no_replies(
 273                 self.pg0, pkts,
 274                 "No response to NS sent to unjoined mcast address")
 275
 276         #
 277         # An NS whose target address is one the router does not own
 278         #
 279         nsma = in6_getnsma(inet_pton(socket.AF_INET6, self.pg0.local_ip6))
 280         d = inet_ntop(socket.AF_INET6, nsma)
 281
 282         p = (Ether(dst=in6_getnsmac(nsma)) /
 283              IPv6(dst=d, src=self.pg0.remote_ip6) /
 284              ICMPv6ND_NS(tgt="fd::ffff") /
 285              ICMPv6NDOptSrcLLAddr(lladdr=self.pg0.remote_mac))
 286         pkts = [p]
 287
 288         self.send_and_assert_no_replies(self.pg0, pkts,
 289                                         "No response to NS for unknown target")
 290
 291     def validate_ra(self, intf, rx, dst_ip=None):
 292         if not dst_ip:
 293             dst_ip = intf.remote_ip6
 294
 295         # unicasted packets must come to the unicast mac
 296         self.assertEqual(rx[Ether].dst, intf.remote_mac)
 297
 298         # and from the router's MAC
 299         self.assertEqual(rx[Ether].src, intf.local_mac)
 300
 301         # the rx'd RA should be addressed to the sender's source
 302         self.assertTrue(rx.haslayer(ICMPv6ND_RA))
 303         self.assertEqual(in6_ptop(rx[IPv6].dst),
 304                          in6_ptop(dst_ip))
 305
 306         # and come from the router's link local
 307         self.assertTrue(in6_islladdr(rx[IPv6].src))
 308         self.assertEqual(in6_ptop(rx[IPv6].src),
 309                          in6_ptop(mk_ll_addr(intf.local_mac)))
 310
 311     def send_and_expect_ra(self, intf, pkts, remark, dst_ip=None,
 312                            filter_out_fn=is_ipv6_misc):
 313         intf.add_stream(pkts)
 314         self.pg0.add_stream(pkts)
 315         self.pg_enable_capture(self.pg_interfaces)
 316         self.pg_start()
 317         rx = intf.get_capture(1, filter_out_fn=filter_out_fn)
 318
 319         self.assertEqual(len(rx), 1)
 320         rx = rx[0]
 321         self.validate_ra(intf, rx, dst_ip)
 322
 323     def test_rs(self):
 324         """ IPv6 Router Solicitation Exceptions
 325
 326         Test scenario:
 327         """
 328
 329         #
 330         # Before we begin change the IPv6 RA responses to use the unicast
 331         # address - that way we will not confuse them with the periodic
 332         # RAs which go to the mcast address
 333         # Sit and wait for the first periodic RA.
 334         #
 335         # TODO
 336         #
 337         self.pg0.ip6_ra_config(send_unicast=1)
 338
 339         #
 340         # An RS from a link source address
 341         #  - expect an RA in return
 342         #
 343         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
 344              IPv6(dst=self.pg0.local_ip6, src=self.pg0.remote_ip6) /
 345              ICMPv6ND_RS())
 346         pkts = [p]
 347         self.send_and_expect_ra(self.pg0, pkts, "Genuine RS")
 348
 349         #
 350         # For the next RS sent the RA should be rate limited
 351         #
 352         self.send_and_assert_no_replies(self.pg0, pkts, "RA rate limited")
 353
 354         #
 355         # When we reconfiure the IPv6 RA config, we reset the RA rate limiting,
 356         # so we need to do this before each test below so as not to drop
 357         # packets for rate limiting reasons. Test this works here.
 358         #
 359         self.pg0.ip6_ra_config(send_unicast=1)
 360         self.send_and_expect_ra(self.pg0, pkts, "Rate limit reset RS")
 361
 362         #
 363         # An RS sent from a non-link local source
 364         #
 365         self.pg0.ip6_ra_config(send_unicast=1)
 366         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
 367              IPv6(dst=self.pg0.local_ip6, src="2002::ffff") /
 368              ICMPv6ND_RS())
 369         pkts = [p]
 370         self.send_and_assert_no_replies(self.pg0, pkts,
 371                                         "RS from non-link source")
 372
 373         #
 374         # Source an RS from a link local address
 375         #
 376         self.pg0.ip6_ra_config(send_unicast=1)
 377         ll = mk_ll_addr(self.pg0.remote_mac)
 378         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
 379              IPv6(dst=self.pg0.local_ip6, src=ll) /
 380              ICMPv6ND_RS())
 381         pkts = [p]
 382         self.send_and_expect_ra(self.pg0, pkts,
 383                                 "RS sourced from link-local",
 384                                 dst_ip=ll)
 385
 386         #
 387         # Send the RS multicast
 388         #
 389         self.pg0.ip6_ra_config(send_unicast=1)
 390         dmac = in6_getnsmac(inet_pton(socket.AF_INET6, "ff02::2"))
 391         ll = mk_ll_addr(self.pg0.remote_mac)
 392         p = (Ether(dst=dmac, src=self.pg0.remote_mac) /
 393              IPv6(dst="ff02::2", src=ll) /
 394              ICMPv6ND_RS())
 395         pkts = [p]
 396         self.send_and_expect_ra(self.pg0, pkts,
 397                                 "RS sourced from link-local",
 398                                 dst_ip=ll)
 399
 400         #
 401         # Source from the unspecified address ::. This happens when the RS
 402         # is sent before the host has a configured address/sub-net,
 403         # i.e. auto-config. Since the sender has no IP address, the reply
 404         # comes back mcast - so the capture needs to not filter this.
 405         # If we happen to pick up the periodic RA at this point then so be it,
 406         # it's not an error.
 407         #
 408         self.pg0.ip6_ra_config(send_unicast=1, suppress=1)
 409         p = (Ether(dst=dmac, src=self.pg0.remote_mac) /
 410              IPv6(dst="ff02::2", src="::") /
 411              ICMPv6ND_RS())
 412         pkts = [p]
 413         self.send_and_expect_ra(self.pg0, pkts,
 414                                 "RS sourced from unspecified",
 415                                 dst_ip="ff02::1",
 416                                 filter_out_fn=None)
 417
 418         #
 419         # Reset the periodic advertisements back to default values
 420         #
 421         self.pg0.ip6_ra_config(no=1, suppress=1, send_unicast=0)
 422
 423 if __name__ == '__main__':
 424     unittest.main(testRunner=VppTestRunner)