test/test_ip6.py

   1 #!/usr/bin/env python
   2
   3 import unittest
   4 from socket import AF_INET6
   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 from vpp_ip_route import VppIpRoute, VppRoutePath, find_route, VppIpMRoute, \
  10     VppMRoutePath, MRouteItfFlags, MRouteEntryFlags, VppMplsIpBind, \
  11     VppMplsRoute
  12 from vpp_neighbor import find_nbr, VppNeighbor
  13
  14 from scapy.packet import Raw
  15 from scapy.layers.l2 import Ether, Dot1Q
  16 from scapy.layers.inet6 import IPv6, UDP, ICMPv6ND_NS, ICMPv6ND_RS, \
  17     ICMPv6ND_RA, ICMPv6NDOptSrcLLAddr, getmacbyip6, ICMPv6MRD_Solicitation, \
  18     ICMPv6NDOptMTU, ICMPv6NDOptSrcLLAddr, ICMPv6NDOptPrefixInfo, \
  19     ICMPv6ND_NA, ICMPv6NDOptDstLLAddr, ICMPv6DestUnreach, icmp6types
  20
  21 from util import ppp
  22 from scapy.utils6 import in6_getnsma, in6_getnsmac, in6_ptop, in6_islladdr, \
  23     in6_mactoifaceid, in6_ismaddr
  24 from scapy.utils import inet_pton, inet_ntop
  25 from scapy.contrib.mpls import MPLS
  26
  27
  28 def mk_ll_addr(mac):
  29     euid = in6_mactoifaceid(mac)
  30     addr = "fe80::" + euid
  31     return addr
  32
  33
  34 class TestIPv6ND(VppTestCase):
  35     def validate_ra(self, intf, rx, dst_ip=None):
  36         if not dst_ip:
  37             dst_ip = intf.remote_ip6
  38
  39         # unicasted packets must come to the unicast mac
  40         self.assertEqual(rx[Ether].dst, intf.remote_mac)
  41
  42         # and from the router's MAC
  43         self.assertEqual(rx[Ether].src, intf.local_mac)
  44
  45         # the rx'd RA should be addressed to the sender's source
  46         self.assertTrue(rx.haslayer(ICMPv6ND_RA))
  47         self.assertEqual(in6_ptop(rx[IPv6].dst),
  48                          in6_ptop(dst_ip))
  49
  50         # and come from the router's link local
  51         self.assertTrue(in6_islladdr(rx[IPv6].src))
  52         self.assertEqual(in6_ptop(rx[IPv6].src),
  53                          in6_ptop(mk_ll_addr(intf.local_mac)))
  54
  55     def validate_na(self, intf, rx, dst_ip=None, tgt_ip=None):
  56         if not dst_ip:
  57             dst_ip = intf.remote_ip6
  58         if not tgt_ip:
  59             dst_ip = intf.local_ip6
  60
  61         # unicasted packets must come to the unicast mac
  62         self.assertEqual(rx[Ether].dst, intf.remote_mac)
  63
  64         # and from the router's MAC
  65         self.assertEqual(rx[Ether].src, intf.local_mac)
  66
  67         # the rx'd NA should be addressed to the sender's source
  68         self.assertTrue(rx.haslayer(ICMPv6ND_NA))
  69         self.assertEqual(in6_ptop(rx[IPv6].dst),
  70                          in6_ptop(dst_ip))
  71
  72         # and come from the target address
  73         self.assertEqual(in6_ptop(rx[IPv6].src), in6_ptop(tgt_ip))
  74
  75         # Dest link-layer options should have the router's MAC
  76         dll = rx[ICMPv6NDOptDstLLAddr]
  77         self.assertEqual(dll.lladdr, intf.local_mac)
  78
  79     def validate_ns(self, intf, rx, tgt_ip):
  80         nsma = in6_getnsma(inet_pton(AF_INET6, tgt_ip))
  81         dst_ip = inet_ntop(AF_INET6, nsma)
  82
  83         # NS is broadcast
  84         self.assertEqual(rx[Ether].dst, "ff:ff:ff:ff:ff:ff")
  85
  86         # and from the router's MAC
  87         self.assertEqual(rx[Ether].src, intf.local_mac)
  88
  89         # the rx'd NS should be addressed to an mcast address
  90         # derived from the target address
  91         self.assertEqual(in6_ptop(rx[IPv6].dst), in6_ptop(dst_ip))
  92
  93         # expect the tgt IP in the NS header
  94         ns = rx[ICMPv6ND_NS]
  95         self.assertEqual(in6_ptop(ns.tgt), in6_ptop(tgt_ip))
  96
  97         # packet is from the router's local address
  98         self.assertEqual(in6_ptop(rx[IPv6].src), intf.local_ip6)
  99
 100         # Src link-layer options should have the router's MAC
 101         sll = rx[ICMPv6NDOptSrcLLAddr]
 102         self.assertEqual(sll.lladdr, intf.local_mac)
 103
 104     def send_and_expect_ra(self, intf, pkts, remark, dst_ip=None,
 105                            filter_out_fn=is_ipv6_misc):
 106         intf.add_stream(pkts)
 107         self.pg_enable_capture(self.pg_interfaces)
 108         self.pg_start()
 109         rx = intf.get_capture(1, filter_out_fn=filter_out_fn)
 110
 111         self.assertEqual(len(rx), 1)
 112         rx = rx[0]
 113         self.validate_ra(intf, rx, dst_ip)
 114
 115     def send_and_expect_na(self, intf, pkts, remark, dst_ip=None,
 116                            tgt_ip=None,
 117                            filter_out_fn=is_ipv6_misc):
 118         intf.add_stream(pkts)
 119         self.pg_enable_capture(self.pg_interfaces)
 120         self.pg_start()
 121         rx = intf.get_capture(1, filter_out_fn=filter_out_fn)
 122
 123         self.assertEqual(len(rx), 1)
 124         rx = rx[0]
 125         self.validate_na(intf, rx, dst_ip, tgt_ip)
 126
 127     def send_and_expect_ns(self, tx_intf, rx_intf, pkts, tgt_ip,
 128                            filter_out_fn=is_ipv6_misc):
 129         tx_intf.add_stream(pkts)
 130         self.pg_enable_capture(self.pg_interfaces)
 131         self.pg_start()
 132         rx = rx_intf.get_capture(1, filter_out_fn=filter_out_fn)
 133
 134         self.assertEqual(len(rx), 1)
 135         rx = rx[0]
 136         self.validate_ns(rx_intf, rx, tgt_ip)
 137
 138     def send_and_assert_no_replies(self, intf, pkts, remark):
 139         intf.add_stream(pkts)
 140         self.pg_enable_capture(self.pg_interfaces)
 141         self.pg_start()
 142         for i in self.pg_interfaces:
 143             i.get_capture(0)
 144             i.assert_nothing_captured(remark=remark)
 145
 146     def verify_ip(self, rx, smac, dmac, sip, dip):
 147         ether = rx[Ether]
 148         self.assertEqual(ether.dst, dmac)
 149         self.assertEqual(ether.src, smac)
 150
 151         ip = rx[IPv6]
 152         self.assertEqual(ip.src, sip)
 153         self.assertEqual(ip.dst, dip)
 154
 155
 156 class TestIPv6(TestIPv6ND):
 157     """ IPv6 Test Case """
 158
 159     @classmethod
 160     def setUpClass(cls):
 161         super(TestIPv6, cls).setUpClass()
 162
 163     def setUp(self):
 164         """
 165         Perform test setup before test case.
 166
 167         **Config:**
 168             - create 3 pg interfaces
 169                 - untagged pg0 interface
 170                 - Dot1Q subinterface on pg1
 171                 - Dot1AD subinterface on pg2
 172             - setup interfaces:
 173                 - put it into UP state
 174                 - set IPv6 addresses
 175                 - resolve neighbor address using NDP
 176             - configure 200 fib entries
 177
 178         :ivar list interfaces: pg interfaces and subinterfaces.
 179         :ivar dict flows: IPv4 packet flows in test.
 180         :ivar list pg_if_packet_sizes: packet sizes in test.
 181
 182         *TODO:* Create AD sub interface
 183         """
 184         super(TestIPv6, self).setUp()
 185
 186         # create 3 pg interfaces
 187         self.create_pg_interfaces(range(3))
 188
 189         # create 2 subinterfaces for p1 and pg2
 190         self.sub_interfaces = [
 191             VppDot1QSubint(self, self.pg1, 100),
 192             VppDot1QSubint(self, self.pg2, 200)
 193             # TODO: VppDot1ADSubint(self, self.pg2, 200, 300, 400)
 194         ]
 195
 196         # packet flows mapping pg0 -> pg1.sub, pg2.sub, etc.
 197         self.flows = dict()
 198         self.flows[self.pg0] = [self.pg1.sub_if, self.pg2.sub_if]
 199         self.flows[self.pg1.sub_if] = [self.pg0, self.pg2.sub_if]
 200         self.flows[self.pg2.sub_if] = [self.pg0, self.pg1.sub_if]
 201
 202         # packet sizes
 203         self.pg_if_packet_sizes = [64, 512, 1518, 9018]
 204         self.sub_if_packet_sizes = [64, 512, 1518 + 4, 9018 + 4]
 205
 206         self.interfaces = list(self.pg_interfaces)
 207         self.interfaces.extend(self.sub_interfaces)
 208
 209         # setup all interfaces
 210         for i in self.interfaces:
 211             i.admin_up()
 212             i.config_ip6()
 213             i.resolve_ndp()
 214
 215         # config 2M FIB entries
 216         self.config_fib_entries(200)
 217
 218     def tearDown(self):
 219         """Run standard test teardown and log ``show ip6 neighbors``."""
 220         for i in self.sub_interfaces:
 221             i.unconfig_ip6()
 222             i.ip6_disable()
 223             i.admin_down()
 224             i.remove_vpp_config()
 225
 226         super(TestIPv6, self).tearDown()
 227         if not self.vpp_dead:
 228             self.logger.info(self.vapi.cli("show ip6 neighbors"))
 229             # info(self.vapi.cli("show ip6 fib"))  # many entries
 230
 231     def config_fib_entries(self, count):
 232         """For each interface add to the FIB table *count* routes to
 233         "fd02::1/128" destination with interface's local address as next-hop
 234         address.
 235
 236         :param int count: Number of FIB entries.
 237
 238         - *TODO:* check if the next-hop address shouldn't be remote address
 239           instead of local address.
 240         """
 241         n_int = len(self.interfaces)
 242         percent = 0
 243         counter = 0.0
 244         dest_addr = inet_pton(AF_INET6, "fd02::1")
 245         dest_addr_len = 128
 246         for i in self.interfaces:
 247             next_hop_address = i.local_ip6n
 248             for j in range(count / n_int):
 249                 self.vapi.ip_add_del_route(
 250                     dest_addr, dest_addr_len, next_hop_address, is_ipv6=1)
 251                 counter += 1
 252                 if counter / count * 100 > percent:
 253                     self.logger.info("Configure %d FIB entries .. %d%% done" %
 254                                      (count, percent))
 255                     percent += 1
 256
 257     def create_stream(self, src_if, packet_sizes):
 258         """Create input packet stream for defined interface.
 259
 260         :param VppInterface src_if: Interface to create packet stream for.
 261         :param list packet_sizes: Required packet sizes.
 262         """
 263         pkts = []
 264         for i in range(0, 257):
 265             dst_if = self.flows[src_if][i % 2]
 266             info = self.create_packet_info(src_if, dst_if)
 267             payload = self.info_to_payload(info)
 268             p = (Ether(dst=src_if.local_mac, src=src_if.remote_mac) /
 269                  IPv6(src=src_if.remote_ip6, dst=dst_if.remote_ip6) /
 270                  UDP(sport=1234, dport=1234) /
 271                  Raw(payload))
 272             info.data = p.copy()
 273             if isinstance(src_if, VppSubInterface):
 274                 p = src_if.add_dot1_layer(p)
 275             size = packet_sizes[(i // 2) % len(packet_sizes)]
 276             self.extend_packet(p, size)
 277             pkts.append(p)
 278         return pkts
 279
 280     def verify_capture(self, dst_if, capture):
 281         """Verify captured input packet stream for defined interface.
 282
 283         :param VppInterface dst_if: Interface to verify captured packet stream
 284                                     for.
 285         :param list capture: Captured packet stream.
 286         """
 287         self.logger.info("Verifying capture on interface %s" % dst_if.name)
 288         last_info = dict()
 289         for i in self.interfaces:
 290             last_info[i.sw_if_index] = None
 291         is_sub_if = False
 292         dst_sw_if_index = dst_if.sw_if_index
 293         if hasattr(dst_if, 'parent'):
 294             is_sub_if = True
 295         for packet in capture:
 296             if is_sub_if:
 297                 # Check VLAN tags and Ethernet header
 298                 packet = dst_if.remove_dot1_layer(packet)
 299             self.assertTrue(Dot1Q not in packet)
 300             try:
 301                 ip = packet[IPv6]
 302                 udp = packet[UDP]
 303                 payload_info = self.payload_to_info(str(packet[Raw]))
 304                 packet_index = payload_info.index
 305                 self.assertEqual(payload_info.dst, dst_sw_if_index)
 306                 self.logger.debug(
 307                     "Got packet on port %s: src=%u (id=%u)" %
 308                     (dst_if.name, payload_info.src, packet_index))
 309                 next_info = self.get_next_packet_info_for_interface2(
 310                     payload_info.src, dst_sw_if_index,
 311                     last_info[payload_info.src])
 312                 last_info[payload_info.src] = next_info
 313                 self.assertTrue(next_info is not None)
 314                 self.assertEqual(packet_index, next_info.index)
 315                 saved_packet = next_info.data
 316                 # Check standard fields
 317                 self.assertEqual(ip.src, saved_packet[IPv6].src)
 318                 self.assertEqual(ip.dst, saved_packet[IPv6].dst)
 319                 self.assertEqual(udp.sport, saved_packet[UDP].sport)
 320                 self.assertEqual(udp.dport, saved_packet[UDP].dport)
 321             except:
 322                 self.logger.error(ppp("Unexpected or invalid packet:", packet))
 323                 raise
 324         for i in self.interfaces:
 325             remaining_packet = self.get_next_packet_info_for_interface2(
 326                 i.sw_if_index, dst_sw_if_index, last_info[i.sw_if_index])
 327             self.assertTrue(remaining_packet is None,
 328                             "Interface %s: Packet expected from interface %s "
 329                             "didn't arrive" % (dst_if.name, i.name))
 330
 331     def test_fib(self):
 332         """ IPv6 FIB test
 333
 334         Test scenario:
 335             - Create IPv6 stream for pg0 interface
 336             - Create IPv6 tagged streams for pg1's and pg2's subinterface.
 337             - Send and verify received packets on each interface.
 338         """
 339
 340         pkts = self.create_stream(self.pg0, self.pg_if_packet_sizes)
 341         self.pg0.add_stream(pkts)
 342
 343         for i in self.sub_interfaces:
 344             pkts = self.create_stream(i, self.sub_if_packet_sizes)
 345             i.parent.add_stream(pkts)
 346
 347         self.pg_enable_capture(self.pg_interfaces)
 348         self.pg_start()
 349
 350         pkts = self.pg0.get_capture()
 351         self.verify_capture(self.pg0, pkts)
 352
 353         for i in self.sub_interfaces:
 354             pkts = i.parent.get_capture()
 355             self.verify_capture(i, pkts)
 356
 357     def test_ns(self):
 358         """ IPv6 Neighbour Solicitation Exceptions
 359
 360         Test scenario:
 361            - Send an NS Sourced from an address not covered by the link sub-net
 362            - Send an NS to an mcast address the router has not joined
 363            - Send NS for a target address the router does not onn.
 364         """
 365
 366         #
 367         # An NS from a non link source address
 368         #
 369         nsma = in6_getnsma(inet_pton(AF_INET6, self.pg0.local_ip6))
 370         d = inet_ntop(AF_INET6, nsma)
 371
 372         p = (Ether(dst=in6_getnsmac(nsma)) /
 373              IPv6(dst=d, src="2002::2") /
 374              ICMPv6ND_NS(tgt=self.pg0.local_ip6) /
 375              ICMPv6NDOptSrcLLAddr(lladdr=self.pg0.remote_mac))
 376         pkts = [p]
 377
 378         self.send_and_assert_no_replies(
 379             self.pg0, pkts,
 380             "No response to NS source by address not on sub-net")
 381
 382         #
 383         # An NS for sent to a solicited mcast group the router is
 384         # not a member of FAILS
 385         #
 386         if 0:
 387             nsma = in6_getnsma(inet_pton(AF_INET6, "fd::ffff"))
 388             d = inet_ntop(AF_INET6, nsma)
 389
 390             p = (Ether(dst=in6_getnsmac(nsma)) /
 391                  IPv6(dst=d, src=self.pg0.remote_ip6) /
 392                  ICMPv6ND_NS(tgt=self.pg0.local_ip6) /
 393                  ICMPv6NDOptSrcLLAddr(lladdr=self.pg0.remote_mac))
 394             pkts = [p]
 395
 396             self.send_and_assert_no_replies(
 397                 self.pg0, pkts,
 398                 "No response to NS sent to unjoined mcast address")
 399
 400         #
 401         # An NS whose target address is one the router does not own
 402         #
 403         nsma = in6_getnsma(inet_pton(AF_INET6, self.pg0.local_ip6))
 404         d = inet_ntop(AF_INET6, nsma)
 405
 406         p = (Ether(dst=in6_getnsmac(nsma)) /
 407              IPv6(dst=d, src=self.pg0.remote_ip6) /
 408              ICMPv6ND_NS(tgt="fd::ffff") /
 409              ICMPv6NDOptSrcLLAddr(lladdr=self.pg0.remote_mac))
 410         pkts = [p]
 411
 412         self.send_and_assert_no_replies(self.pg0, pkts,
 413                                         "No response to NS for unknown target")
 414
 415         #
 416         # A neighbor entry that has no associated FIB-entry
 417         #
 418         self.pg0.generate_remote_hosts(4)
 419         nd_entry = VppNeighbor(self,
 420                                self.pg0.sw_if_index,
 421                                self.pg0.remote_hosts[2].mac,
 422                                self.pg0.remote_hosts[2].ip6,
 423                                af=AF_INET6,
 424                                is_no_fib_entry=1)
 425         nd_entry.add_vpp_config()
 426
 427         #
 428         # check we have the neighbor, but no route
 429         #
 430         self.assertTrue(find_nbr(self,
 431                                  self.pg0.sw_if_index,
 432                                  self.pg0._remote_hosts[2].ip6,
 433                                  inet=AF_INET6))
 434         self.assertFalse(find_route(self,
 435                                     self.pg0._remote_hosts[2].ip6,
 436                                     128,
 437                                     inet=AF_INET6))
 438
 439         #
 440         # send an NS from a link local address to the interface's global
 441         # address
 442         #
 443         p = (Ether(dst=in6_getnsmac(nsma), src=self.pg0.remote_mac) /
 444              IPv6(dst=d, src=self.pg0._remote_hosts[2].ip6_ll) /
 445              ICMPv6ND_NS(tgt=self.pg0.local_ip6) /
 446              ICMPv6NDOptSrcLLAddr(lladdr=self.pg0.remote_mac))
 447
 448         self.send_and_expect_na(self.pg0, p,
 449                                 "NS from link-local",
 450                                 dst_ip=self.pg0._remote_hosts[2].ip6_ll,
 451                                 tgt_ip=self.pg0.local_ip6)
 452
 453         #
 454         # we should have learned an ND entry for the peer's link-local
 455         # but not inserted a route to it in the FIB
 456         #
 457         self.assertTrue(find_nbr(self,
 458                                  self.pg0.sw_if_index,
 459                                  self.pg0._remote_hosts[2].ip6_ll,
 460                                  inet=AF_INET6))
 461         self.assertFalse(find_route(self,
 462                                     self.pg0._remote_hosts[2].ip6_ll,
 463                                     128,
 464                                     inet=AF_INET6))
 465
 466         #
 467         # An NS to the router's own Link-local
 468         #
 469         p = (Ether(dst=in6_getnsmac(nsma), src=self.pg0.remote_mac) /
 470              IPv6(dst=d, src=self.pg0._remote_hosts[3].ip6_ll) /
 471              ICMPv6ND_NS(tgt=self.pg0.local_ip6_ll) /
 472              ICMPv6NDOptSrcLLAddr(lladdr=self.pg0.remote_mac))
 473
 474         self.send_and_expect_na(self.pg0, p,
 475                                 "NS to/from link-local",
 476                                 dst_ip=self.pg0._remote_hosts[3].ip6_ll,
 477                                 tgt_ip=self.pg0.local_ip6_ll)
 478
 479         #
 480         # we should have learned an ND entry for the peer's link-local
 481         # but not inserted a route to it in the FIB
 482         #
 483         self.assertTrue(find_nbr(self,
 484                                  self.pg0.sw_if_index,
 485                                  self.pg0._remote_hosts[3].ip6_ll,
 486                                  inet=AF_INET6))
 487         self.assertFalse(find_route(self,
 488                                     self.pg0._remote_hosts[3].ip6_ll,
 489                                     128,
 490                                     inet=AF_INET6))
 491
 492     def test_ns_duplicates(self):
 493         """ ARP Duplicates"""
 494
 495         #
 496         # Generate some hosts on the LAN
 497         #
 498         self.pg1.generate_remote_hosts(3)
 499
 500         #
 501         # Add host 1 on pg1 and pg2
 502         #
 503         ns_pg1 = VppNeighbor(self,
 504                              self.pg1.sw_if_index,
 505                              self.pg1.remote_hosts[1].mac,
 506                              self.pg1.remote_hosts[1].ip6,
 507                              af=AF_INET6)
 508         ns_pg1.add_vpp_config()
 509         ns_pg2 = VppNeighbor(self,
 510                              self.pg2.sw_if_index,
 511                              self.pg2.remote_mac,
 512                              self.pg1.remote_hosts[1].ip6,
 513                              af=AF_INET6)
 514         ns_pg2.add_vpp_config()
 515
 516         #
 517         # IP packet destined for pg1 remote host arrives on pg1 again.
 518         #
 519         p = (Ether(dst=self.pg0.local_mac,
 520                    src=self.pg0.remote_mac) /
 521              IPv6(src=self.pg0.remote_ip6,
 522                   dst=self.pg1.remote_hosts[1].ip6) /
 523              UDP(sport=1234, dport=1234) /
 524              Raw())
 525
 526         self.pg0.add_stream(p)
 527         self.pg_enable_capture(self.pg_interfaces)
 528         self.pg_start()
 529
 530         rx1 = self.pg1.get_capture(1)
 531
 532         self.verify_ip(rx1[0],
 533                        self.pg1.local_mac,
 534                        self.pg1.remote_hosts[1].mac,
 535                        self.pg0.remote_ip6,
 536                        self.pg1.remote_hosts[1].ip6)
 537
 538         #
 539         # remove the duplicate on pg1
 540         # packet stream shoud generate ARPs out of pg1
 541         #
 542         ns_pg1.remove_vpp_config()
 543
 544         self.send_and_expect_ns(self.pg0, self.pg1,
 545                                 p, self.pg1.remote_hosts[1].ip6)
 546
 547         #
 548         # Add it back
 549         #
 550         ns_pg1.add_vpp_config()
 551
 552         self.pg0.add_stream(p)
 553         self.pg_enable_capture(self.pg_interfaces)
 554         self.pg_start()
 555
 556         rx1 = self.pg1.get_capture(1)
 557
 558         self.verify_ip(rx1[0],
 559                        self.pg1.local_mac,
 560                        self.pg1.remote_hosts[1].mac,
 561                        self.pg0.remote_ip6,
 562                        self.pg1.remote_hosts[1].ip6)
 563
 564     def validate_ra(self, intf, rx, dst_ip=None, mtu=9000, pi_opt=None):
 565         if not dst_ip:
 566             dst_ip = intf.remote_ip6
 567
 568         # unicasted packets must come to the unicast mac
 569         self.assertEqual(rx[Ether].dst, intf.remote_mac)
 570
 571         # and from the router's MAC
 572         self.assertEqual(rx[Ether].src, intf.local_mac)
 573
 574         # the rx'd RA should be addressed to the sender's source
 575         self.assertTrue(rx.haslayer(ICMPv6ND_RA))
 576         self.assertEqual(in6_ptop(rx[IPv6].dst),
 577                          in6_ptop(dst_ip))
 578
 579         # and come from the router's link local
 580         self.assertTrue(in6_islladdr(rx[IPv6].src))
 581         self.assertEqual(in6_ptop(rx[IPv6].src),
 582                          in6_ptop(mk_ll_addr(intf.local_mac)))
 583
 584         # it should contain the links MTU
 585         ra = rx[ICMPv6ND_RA]
 586         self.assertEqual(ra[ICMPv6NDOptMTU].mtu, mtu)
 587
 588         # it should contain the source's link layer address option
 589         sll = ra[ICMPv6NDOptSrcLLAddr]
 590         self.assertEqual(sll.lladdr, intf.local_mac)
 591
 592         if not pi_opt:
 593             # the RA should not contain prefix information
 594             self.assertFalse(ra.haslayer(ICMPv6NDOptPrefixInfo))
 595         else:
 596             raos = rx.getlayer(ICMPv6NDOptPrefixInfo, 1)
 597
 598             # the options are nested in the scapy packet in way that i cannot
 599             # decipher how to decode. this 1st layer of option always returns
 600             # nested classes, so a direct obj1=obj2 comparison always fails.
 601             # however, the getlayer(.., 2) does give one instnace.
 602             # so we cheat here and construct a new opt instnace for comparison
 603             rd = ICMPv6NDOptPrefixInfo(prefixlen=raos.prefixlen,
 604                                        prefix=raos.prefix,
 605                                        L=raos.L,
 606                                        A=raos.A)
 607             if type(pi_opt) is list:
 608                 for ii in range(len(pi_opt)):
 609                     self.assertEqual(pi_opt[ii], rd)
 610                     rd = rx.getlayer(ICMPv6NDOptPrefixInfo, ii+2)
 611             else:
 612                 self.assertEqual(pi_opt, raos)
 613
 614     def send_and_expect_ra(self, intf, pkts, remark, dst_ip=None,
 615                            filter_out_fn=is_ipv6_misc,
 616                            opt=None):
 617         intf.add_stream(pkts)
 618         self.pg_enable_capture(self.pg_interfaces)
 619         self.pg_start()
 620         rx = intf.get_capture(1, filter_out_fn=filter_out_fn)
 621
 622         self.assertEqual(len(rx), 1)
 623         rx = rx[0]
 624         self.validate_ra(intf, rx, dst_ip, pi_opt=opt)
 625
 626     def test_rs(self):
 627         """ IPv6 Router Solicitation Exceptions
 628
 629         Test scenario:
 630         """
 631
 632         #
 633         # Before we begin change the IPv6 RA responses to use the unicast
 634         # address - that way we will not confuse them with the periodic
 635         # RAs which go to the mcast address
 636         # Sit and wait for the first periodic RA.
 637         #
 638         # TODO
 639         #
 640         self.pg0.ip6_ra_config(send_unicast=1)
 641
 642         #
 643         # An RS from a link source address
 644         #  - expect an RA in return
 645         #
 646         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
 647              IPv6(dst=self.pg0.local_ip6, src=self.pg0.remote_ip6) /
 648              ICMPv6ND_RS())
 649         pkts = [p]
 650         self.send_and_expect_ra(self.pg0, pkts, "Genuine RS")
 651
 652         #
 653         # For the next RS sent the RA should be rate limited
 654         #
 655         self.send_and_assert_no_replies(self.pg0, pkts, "RA rate limited")
 656
 657         #
 658         # When we reconfiure the IPv6 RA config, we reset the RA rate limiting,
 659         # so we need to do this before each test below so as not to drop
 660         # packets for rate limiting reasons. Test this works here.
 661         #
 662         self.pg0.ip6_ra_config(send_unicast=1)
 663         self.send_and_expect_ra(self.pg0, pkts, "Rate limit reset RS")
 664
 665         #
 666         # An RS sent from a non-link local source
 667         #
 668         self.pg0.ip6_ra_config(send_unicast=1)
 669         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
 670              IPv6(dst=self.pg0.local_ip6, src="2002::ffff") /
 671              ICMPv6ND_RS())
 672         pkts = [p]
 673         self.send_and_assert_no_replies(self.pg0, pkts,
 674                                         "RS from non-link source")
 675
 676         #
 677         # Source an RS from a link local address
 678         #
 679         self.pg0.ip6_ra_config(send_unicast=1)
 680         ll = mk_ll_addr(self.pg0.remote_mac)
 681         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
 682              IPv6(dst=self.pg0.local_ip6, src=ll) /
 683              ICMPv6ND_RS())
 684         pkts = [p]
 685         self.send_and_expect_ra(self.pg0, pkts,
 686                                 "RS sourced from link-local",
 687                                 dst_ip=ll)
 688
 689         #
 690         # Send the RS multicast
 691         #
 692         self.pg0.ip6_ra_config(send_unicast=1)
 693         dmac = in6_getnsmac(inet_pton(AF_INET6, "ff02::2"))
 694         ll = mk_ll_addr(self.pg0.remote_mac)
 695         p = (Ether(dst=dmac, src=self.pg0.remote_mac) /
 696              IPv6(dst="ff02::2", src=ll) /
 697              ICMPv6ND_RS())
 698         pkts = [p]
 699         self.send_and_expect_ra(self.pg0, pkts,
 700                                 "RS sourced from link-local",
 701                                 dst_ip=ll)
 702
 703         #
 704         # Source from the unspecified address ::. This happens when the RS
 705         # is sent before the host has a configured address/sub-net,
 706         # i.e. auto-config. Since the sender has no IP address, the reply
 707         # comes back mcast - so the capture needs to not filter this.
 708         # If we happen to pick up the periodic RA at this point then so be it,
 709         # it's not an error.
 710         #
 711         self.pg0.ip6_ra_config(send_unicast=1, suppress=1)
 712         p = (Ether(dst=dmac, src=self.pg0.remote_mac) /
 713              IPv6(dst="ff02::2", src="::") /
 714              ICMPv6ND_RS())
 715         pkts = [p]
 716         self.send_and_expect_ra(self.pg0, pkts,
 717                                 "RS sourced from unspecified",
 718                                 dst_ip="ff02::1",
 719                                 filter_out_fn=None)
 720
 721         #
 722         # Configure The RA to announce the links prefix
 723         #
 724         self.pg0.ip6_ra_prefix(self.pg0.local_ip6n,
 725                                self.pg0.local_ip6_prefix_len)
 726
 727         #
 728         # RAs should now contain the prefix information option
 729         #
 730         opt = ICMPv6NDOptPrefixInfo(prefixlen=self.pg0.local_ip6_prefix_len,
 731                                     prefix=self.pg0.local_ip6,
 732                                     L=1,
 733                                     A=1)
 734
 735         self.pg0.ip6_ra_config(send_unicast=1)
 736         ll = mk_ll_addr(self.pg0.remote_mac)
 737         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
 738              IPv6(dst=self.pg0.local_ip6, src=ll) /
 739              ICMPv6ND_RS())
 740         self.send_and_expect_ra(self.pg0, p,
 741                                 "RA with prefix-info",
 742                                 dst_ip=ll,
 743                                 opt=opt)
 744
 745         #
 746         # Change the prefix info to not off-link
 747         #  L-flag is clear
 748         #
 749         self.pg0.ip6_ra_prefix(self.pg0.local_ip6n,
 750                                self.pg0.local_ip6_prefix_len,
 751                                off_link=1)
 752
 753         opt = ICMPv6NDOptPrefixInfo(prefixlen=self.pg0.local_ip6_prefix_len,
 754                                     prefix=self.pg0.local_ip6,
 755                                     L=0,
 756                                     A=1)
 757
 758         self.pg0.ip6_ra_config(send_unicast=1)
 759         self.send_and_expect_ra(self.pg0, p,
 760                                 "RA with Prefix info with L-flag=0",
 761                                 dst_ip=ll,
 762                                 opt=opt)
 763
 764         #
 765         # Change the prefix info to not off-link, no-autoconfig
 766         #  L and A flag are clear in the advert
 767         #
 768         self.pg0.ip6_ra_prefix(self.pg0.local_ip6n,
 769                                self.pg0.local_ip6_prefix_len,
 770                                off_link=1,
 771                                no_autoconfig=1)
 772
 773         opt = ICMPv6NDOptPrefixInfo(prefixlen=self.pg0.local_ip6_prefix_len,
 774                                     prefix=self.pg0.local_ip6,
 775                                     L=0,
 776                                     A=0)
 777
 778         self.pg0.ip6_ra_config(send_unicast=1)
 779         self.send_and_expect_ra(self.pg0, p,
 780                                 "RA with Prefix info with A & L-flag=0",
 781                                 dst_ip=ll,
 782                                 opt=opt)
 783
 784         #
 785         # Change the flag settings back to the defaults
 786         #  L and A flag are set in the advert
 787         #
 788         self.pg0.ip6_ra_prefix(self.pg0.local_ip6n,
 789                                self.pg0.local_ip6_prefix_len)
 790
 791         opt = ICMPv6NDOptPrefixInfo(prefixlen=self.pg0.local_ip6_prefix_len,
 792                                     prefix=self.pg0.local_ip6,
 793                                     L=1,
 794                                     A=1)
 795
 796         self.pg0.ip6_ra_config(send_unicast=1)
 797         self.send_and_expect_ra(self.pg0, p,
 798                                 "RA with Prefix info",
 799                                 dst_ip=ll,
 800                                 opt=opt)
 801
 802         #
 803         # Change the prefix info to not off-link, no-autoconfig
 804         #  L and A flag are clear in the advert
 805         #
 806         self.pg0.ip6_ra_prefix(self.pg0.local_ip6n,
 807                                self.pg0.local_ip6_prefix_len,
 808                                off_link=1,
 809                                no_autoconfig=1)
 810
 811         opt = ICMPv6NDOptPrefixInfo(prefixlen=self.pg0.local_ip6_prefix_len,
 812                                     prefix=self.pg0.local_ip6,
 813                                     L=0,
 814                                     A=0)
 815
 816         self.pg0.ip6_ra_config(send_unicast=1)
 817         self.send_and_expect_ra(self.pg0, p,
 818                                 "RA with Prefix info with A & L-flag=0",
 819                                 dst_ip=ll,
 820                                 opt=opt)
 821
 822         #
 823         # Use the reset to defults option to revert to defaults
 824         #  L and A flag are clear in the advert
 825         #
 826         self.pg0.ip6_ra_prefix(self.pg0.local_ip6n,
 827                                self.pg0.local_ip6_prefix_len,
 828                                use_default=1)
 829
 830         opt = ICMPv6NDOptPrefixInfo(prefixlen=self.pg0.local_ip6_prefix_len,
 831                                     prefix=self.pg0.local_ip6,
 832                                     L=1,
 833                                     A=1)
 834
 835         self.pg0.ip6_ra_config(send_unicast=1)
 836         self.send_and_expect_ra(self.pg0, p,
 837                                 "RA with Prefix reverted to defaults",
 838                                 dst_ip=ll,
 839                                 opt=opt)
 840
 841         #
 842         # Advertise Another prefix. With no L-flag/A-flag
 843         #
 844         self.pg0.ip6_ra_prefix(self.pg1.local_ip6n,
 845                                self.pg1.local_ip6_prefix_len,
 846                                off_link=1,
 847                                no_autoconfig=1)
 848
 849         opt = [ICMPv6NDOptPrefixInfo(prefixlen=self.pg0.local_ip6_prefix_len,
 850                                      prefix=self.pg0.local_ip6,
 851                                      L=1,
 852                                      A=1),
 853                ICMPv6NDOptPrefixInfo(prefixlen=self.pg1.local_ip6_prefix_len,
 854                                      prefix=self.pg1.local_ip6,
 855                                      L=0,
 856                                      A=0)]
 857
 858         self.pg0.ip6_ra_config(send_unicast=1)
 859         ll = mk_ll_addr(self.pg0.remote_mac)
 860         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
 861              IPv6(dst=self.pg0.local_ip6, src=ll) /
 862              ICMPv6ND_RS())
 863         self.send_and_expect_ra(self.pg0, p,
 864                                 "RA with multiple Prefix infos",
 865                                 dst_ip=ll,
 866                                 opt=opt)
 867
 868         #
 869         # Remove the first refix-info - expect the second is still in the
 870         # advert
 871         #
 872         self.pg0.ip6_ra_prefix(self.pg0.local_ip6n,
 873                                self.pg0.local_ip6_prefix_len,
 874                                is_no=1)
 875
 876         opt = ICMPv6NDOptPrefixInfo(prefixlen=self.pg1.local_ip6_prefix_len,
 877                                     prefix=self.pg1.local_ip6,
 878                                     L=0,
 879                                     A=0)
 880
 881         self.pg0.ip6_ra_config(send_unicast=1)
 882         self.send_and_expect_ra(self.pg0, p,
 883                                 "RA with Prefix reverted to defaults",
 884                                 dst_ip=ll,
 885                                 opt=opt)
 886
 887         #
 888         # Remove the second prefix-info - expect no prefix-info i nthe adverts
 889         #
 890         self.pg0.ip6_ra_prefix(self.pg1.local_ip6n,
 891                                self.pg1.local_ip6_prefix_len,
 892                                is_no=1)
 893
 894         self.pg0.ip6_ra_config(send_unicast=1)
 895         self.send_and_expect_ra(self.pg0, p,
 896                                 "RA with Prefix reverted to defaults",
 897                                 dst_ip=ll)
 898
 899         #
 900         # Reset the periodic advertisements back to default values
 901         #
 902         self.pg0.ip6_ra_config(no=1, suppress=1, send_unicast=0)
 903
 904
 905 class IPv6NDProxyTest(TestIPv6ND):
 906     """ IPv6 ND ProxyTest Case """
 907
 908     def setUp(self):
 909         super(IPv6NDProxyTest, self).setUp()
 910
 911         # create 3 pg interfaces
 912         self.create_pg_interfaces(range(3))
 913
 914         # pg0 is the master interface, with the configured subnet
 915         self.pg0.admin_up()
 916         self.pg0.config_ip6()
 917         self.pg0.resolve_ndp()
 918
 919         self.pg1.ip6_enable()
 920         self.pg2.ip6_enable()
 921
 922     def tearDown(self):
 923         super(IPv6NDProxyTest, self).tearDown()
 924
 925     def test_nd_proxy(self):
 926         """ IPv6 Proxy ND """
 927
 928         #
 929         # Generate some hosts in the subnet that we are proxying
 930         #
 931         self.pg0.generate_remote_hosts(8)
 932
 933         nsma = in6_getnsma(inet_pton(AF_INET6, self.pg0.local_ip6))
 934         d = inet_ntop(AF_INET6, nsma)
 935
 936         #
 937         # Send an NS for one of those remote hosts on one of the proxy links
 938         # expect no response since it's from an address that is not
 939         # on the link that has the prefix configured
 940         #
 941         ns_pg1 = (Ether(dst=in6_getnsmac(nsma), src=self.pg1.remote_mac) /
 942                   IPv6(dst=d, src=self.pg0._remote_hosts[2].ip6) /
 943                   ICMPv6ND_NS(tgt=self.pg0.local_ip6) /
 944                   ICMPv6NDOptSrcLLAddr(lladdr=self.pg0._remote_hosts[2].mac))
 945
 946         self.send_and_assert_no_replies(self.pg1, ns_pg1, "Off link NS")
 947
 948         #
 949         # Add proxy support for the host
 950         #
 951         self.vapi.ip6_nd_proxy(
 952             inet_pton(AF_INET6, self.pg0._remote_hosts[2].ip6),
 953             self.pg1.sw_if_index)
 954
 955         #
 956         # try that NS again. this time we expect an NA back
 957         #
 958         self.send_and_expect_na(self.pg1, ns_pg1,
 959                                 "NS to proxy entry",
 960                                 dst_ip=self.pg0._remote_hosts[2].ip6,
 961                                 tgt_ip=self.pg0.local_ip6)
 962
 963         #
 964         # ... and that we have an entry in the ND cache
 965         #
 966         self.assertTrue(find_nbr(self,
 967                                  self.pg1.sw_if_index,
 968                                  self.pg0._remote_hosts[2].ip6,
 969                                  inet=AF_INET6))
 970
 971         #
 972         # ... and we can route traffic to it
 973         #
 974         t = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
 975              IPv6(dst=self.pg0._remote_hosts[2].ip6,
 976                   src=self.pg0.remote_ip6) /
 977              UDP(sport=10000, dport=20000) /
 978              Raw('\xa5' * 100))
 979
 980         self.pg0.add_stream(t)
 981         self.pg_enable_capture(self.pg_interfaces)
 982         self.pg_start()
 983         rx = self.pg1.get_capture(1)
 984         rx = rx[0]
 985
 986         self.assertEqual(rx[Ether].dst, self.pg0._remote_hosts[2].mac)
 987         self.assertEqual(rx[Ether].src, self.pg1.local_mac)
 988
 989         self.assertEqual(rx[IPv6].src, t[IPv6].src)
 990         self.assertEqual(rx[IPv6].dst, t[IPv6].dst)
 991
 992         #
 993         # Test we proxy for the host on the main interface
 994         #
 995         ns_pg0 = (Ether(dst=in6_getnsmac(nsma), src=self.pg0.remote_mac) /
 996                   IPv6(dst=d, src=self.pg0.remote_ip6) /
 997                   ICMPv6ND_NS(tgt=self.pg0._remote_hosts[2].ip6) /
 998                   ICMPv6NDOptSrcLLAddr(lladdr=self.pg0.remote_mac))
 999
1000         self.send_and_expect_na(self.pg0, ns_pg0,
1001                                 "NS to proxy entry on main",
1002                                 tgt_ip=self.pg0._remote_hosts[2].ip6,
1003                                 dst_ip=self.pg0.remote_ip6)
1004
1005         #
1006         # Setup and resolve proxy for another host on another interface
1007         #
1008         ns_pg2 = (Ether(dst=in6_getnsmac(nsma), src=self.pg2.remote_mac) /
1009                   IPv6(dst=d, src=self.pg0._remote_hosts[3].ip6) /
1010                   ICMPv6ND_NS(tgt=self.pg0.local_ip6) /
1011                   ICMPv6NDOptSrcLLAddr(lladdr=self.pg0._remote_hosts[2].mac))
1012
1013         self.vapi.ip6_nd_proxy(
1014             inet_pton(AF_INET6, self.pg0._remote_hosts[3].ip6),
1015             self.pg2.sw_if_index)
1016
1017         self.send_and_expect_na(self.pg2, ns_pg2,
1018                                 "NS to proxy entry other interface",
1019                                 dst_ip=self.pg0._remote_hosts[3].ip6,
1020                                 tgt_ip=self.pg0.local_ip6)
1021
1022         self.assertTrue(find_nbr(self,
1023                                  self.pg2.sw_if_index,
1024                                  self.pg0._remote_hosts[3].ip6,
1025                                  inet=AF_INET6))
1026
1027         #
1028         # hosts can communicate. pg2->pg1
1029         #
1030         t2 = (Ether(dst=self.pg2.local_mac,
1031                     src=self.pg0.remote_hosts[3].mac) /
1032               IPv6(dst=self.pg0._remote_hosts[2].ip6,
1033                    src=self.pg0._remote_hosts[3].ip6) /
1034               UDP(sport=10000, dport=20000) /
1035               Raw('\xa5' * 100))
1036
1037         self.pg2.add_stream(t2)
1038         self.pg_enable_capture(self.pg_interfaces)
1039         self.pg_start()
1040         rx = self.pg1.get_capture(1)
1041         rx = rx[0]
1042
1043         self.assertEqual(rx[Ether].dst, self.pg0._remote_hosts[2].mac)
1044         self.assertEqual(rx[Ether].src, self.pg1.local_mac)
1045
1046         self.assertEqual(rx[IPv6].src, t2[IPv6].src)
1047         self.assertEqual(rx[IPv6].dst, t2[IPv6].dst)
1048
1049         #
1050         # remove the proxy configs
1051         #
1052         self.vapi.ip6_nd_proxy(
1053             inet_pton(AF_INET6, self.pg0._remote_hosts[2].ip6),
1054             self.pg1.sw_if_index,
1055             is_del=1)
1056         self.vapi.ip6_nd_proxy(
1057             inet_pton(AF_INET6, self.pg0._remote_hosts[3].ip6),
1058             self.pg2.sw_if_index,
1059             is_del=1)
1060
1061         self.assertFalse(find_nbr(self,
1062                                   self.pg2.sw_if_index,
1063                                   self.pg0._remote_hosts[3].ip6,
1064                                   inet=AF_INET6))
1065         self.assertFalse(find_nbr(self,
1066                                   self.pg1.sw_if_index,
1067                                   self.pg0._remote_hosts[2].ip6,
1068                                   inet=AF_INET6))
1069
1070         #
1071         # no longer proxy-ing...
1072         #
1073         self.send_and_assert_no_replies(self.pg0, ns_pg0, "Proxy unconfigured")
1074         self.send_and_assert_no_replies(self.pg1, ns_pg1, "Proxy unconfigured")
1075         self.send_and_assert_no_replies(self.pg2, ns_pg2, "Proxy unconfigured")
1076
1077         #
1078         # no longer forwarding. traffic generates NS out of the glean/main
1079         # interface
1080         #
1081         self.pg2.add_stream(t2)
1082         self.pg_enable_capture(self.pg_interfaces)
1083         self.pg_start()
1084
1085         rx = self.pg0.get_capture(1)
1086
1087         self.assertTrue(rx[0].haslayer(ICMPv6ND_NS))
1088
1089
1090 class TestIPNull(VppTestCase):
1091     """ IPv6 routes via NULL """
1092
1093     def setUp(self):
1094         super(TestIPNull, self).setUp()
1095
1096         # create 2 pg interfaces
1097         self.create_pg_interfaces(range(1))
1098
1099         for i in self.pg_interfaces:
1100             i.admin_up()
1101             i.config_ip6()
1102             i.resolve_ndp()
1103
1104     def tearDown(self):
1105         super(TestIPNull, self).tearDown()
1106         for i in self.pg_interfaces:
1107             i.unconfig_ip6()
1108             i.admin_down()
1109
1110     def test_ip_null(self):
1111         """ IP NULL route """
1112
1113         p = (Ether(src=self.pg0.remote_mac,
1114                    dst=self.pg0.local_mac) /
1115              IPv6(src=self.pg0.remote_ip6, dst="2001::1") /
1116              UDP(sport=1234, dport=1234) /
1117              Raw('\xa5' * 100))
1118
1119         #
1120         # A route via IP NULL that will reply with ICMP unreachables
1121         #
1122         ip_unreach = VppIpRoute(self, "2001::", 64, [], is_unreach=1, is_ip6=1)
1123         ip_unreach.add_vpp_config()
1124
1125         self.pg0.add_stream(p)
1126         self.pg_enable_capture(self.pg_interfaces)
1127         self.pg_start()
1128
1129         rx = self.pg0.get_capture(1)
1130         rx = rx[0]
1131         icmp = rx[ICMPv6DestUnreach]
1132
1133         # 0 = "No route to destination"
1134         self.assertEqual(icmp.code, 0)
1135
1136         # ICMP is rate limited. pause a bit
1137         self.sleep(1)
1138
1139         #
1140         # A route via IP NULL that will reply with ICMP prohibited
1141         #
1142         ip_prohibit = VppIpRoute(self, "2001::1", 128, [],
1143                                  is_prohibit=1, is_ip6=1)
1144         ip_prohibit.add_vpp_config()
1145
1146         self.pg0.add_stream(p)
1147         self.pg_enable_capture(self.pg_interfaces)
1148         self.pg_start()
1149
1150         rx = self.pg0.get_capture(1)
1151         rx = rx[0]
1152         icmp = rx[ICMPv6DestUnreach]
1153
1154         # 1 = "Communication with destination administratively prohibited"
1155         self.assertEqual(icmp.code, 1)
1156
1157
1158 class TestIPDisabled(VppTestCase):
1159     """ IPv6 disabled """
1160
1161     def setUp(self):
1162         super(TestIPDisabled, self).setUp()
1163
1164         # create 2 pg interfaces
1165         self.create_pg_interfaces(range(2))
1166
1167         # PG0 is IP enalbed
1168         self.pg0.admin_up()
1169         self.pg0.config_ip6()
1170         self.pg0.resolve_ndp()
1171
1172         # PG 1 is not IP enabled
1173         self.pg1.admin_up()
1174
1175     def tearDown(self):
1176         super(TestIPDisabled, self).tearDown()
1177         for i in self.pg_interfaces:
1178             i.unconfig_ip4()
1179             i.admin_down()
1180
1181     def send_and_assert_no_replies(self, intf, pkts, remark):
1182         intf.add_stream(pkts)
1183         self.pg_enable_capture(self.pg_interfaces)
1184         self.pg_start()
1185         for i in self.pg_interfaces:
1186             i.get_capture(0)
1187             i.assert_nothing_captured(remark=remark)
1188
1189     def test_ip_disabled(self):
1190         """ IP Disabled """
1191
1192         #
1193         # An (S,G).
1194         # one accepting interface, pg0, 2 forwarding interfaces
1195         #
1196         route_ff_01 = VppIpMRoute(
1197             self,
1198             "::",
1199             "ffef::1", 128,
1200             MRouteEntryFlags.MFIB_ENTRY_FLAG_NONE,
1201             [VppMRoutePath(self.pg1.sw_if_index,
1202                            MRouteItfFlags.MFIB_ITF_FLAG_ACCEPT),
1203              VppMRoutePath(self.pg0.sw_if_index,
1204                            MRouteItfFlags.MFIB_ITF_FLAG_FORWARD)],
1205             is_ip6=1)
1206         route_ff_01.add_vpp_config()
1207
1208         pu = (Ether(src=self.pg1.remote_mac,
1209                     dst=self.pg1.local_mac) /
1210               IPv6(src="2001::1", dst=self.pg0.remote_ip6) /
1211               UDP(sport=1234, dport=1234) /
1212               Raw('\xa5' * 100))
1213         pm = (Ether(src=self.pg1.remote_mac,
1214                     dst=self.pg1.local_mac) /
1215               IPv6(src="2001::1", dst="ffef::1") /
1216               UDP(sport=1234, dport=1234) /
1217               Raw('\xa5' * 100))
1218
1219         #
1220         # PG1 does not forward IP traffic
1221         #
1222         self.send_and_assert_no_replies(self.pg1, pu, "IPv6 disabled")
1223         self.send_and_assert_no_replies(self.pg1, pm, "IPv6 disabled")
1224
1225         #
1226         # IP enable PG1
1227         #
1228         self.pg1.config_ip6()
1229
1230         #
1231         # Now we get packets through
1232         #
1233         self.pg1.add_stream(pu)
1234         self.pg_enable_capture(self.pg_interfaces)
1235         self.pg_start()
1236         rx = self.pg0.get_capture(1)
1237
1238         self.pg1.add_stream(pm)
1239         self.pg_enable_capture(self.pg_interfaces)
1240         self.pg_start()
1241         rx = self.pg0.get_capture(1)
1242
1243         #
1244         # Disable PG1
1245         #
1246         self.pg1.unconfig_ip6()
1247
1248         #
1249         # PG1 does not forward IP traffic
1250         #
1251         self.send_and_assert_no_replies(self.pg1, pu, "IPv6 disabled")
1252         self.send_and_assert_no_replies(self.pg1, pm, "IPv6 disabled")
1253
1254
1255 class TestIP6LoadBalance(VppTestCase):
1256     """ IPv6 Load-Balancing """
1257
1258     def setUp(self):
1259         super(TestIP6LoadBalance, self).setUp()
1260
1261         self.create_pg_interfaces(range(5))
1262
1263         for i in self.pg_interfaces:
1264             i.admin_up()
1265             i.config_ip6()
1266             i.resolve_ndp()
1267             i.enable_mpls()
1268
1269     def tearDown(self):
1270         super(TestIP6LoadBalance, self).tearDown()
1271         for i in self.pg_interfaces:
1272             i.unconfig_ip6()
1273             i.admin_down()
1274             i.disable_mpls()
1275
1276     def send_and_expect_load_balancing(self, input, pkts, outputs):
1277         input.add_stream(pkts)
1278         self.pg_enable_capture(self.pg_interfaces)
1279         self.pg_start()
1280         for oo in outputs:
1281             rx = oo._get_capture(1)
1282             self.assertNotEqual(0, len(rx))
1283
1284     def send_and_expect_one_itf(self, input, pkts, itf):
1285         input.add_stream(pkts)
1286         self.pg_enable_capture(self.pg_interfaces)
1287         self.pg_start()
1288         rx = itf.get_capture(len(pkts))
1289
1290     def test_ip6_load_balance(self):
1291         """ IPv6 Load-Balancing """
1292
1293         #
1294         # An array of packets that differ only in the destination port
1295         #  - IP only
1296         #  - MPLS EOS
1297         #  - MPLS non-EOS
1298         #  - MPLS non-EOS with an entropy label
1299         #
1300         port_ip_pkts = []
1301         port_mpls_pkts = []
1302         port_mpls_neos_pkts = []
1303         port_ent_pkts = []
1304
1305         #
1306         # An array of packets that differ only in the source address
1307         #
1308         src_ip_pkts = []
1309         src_mpls_pkts = []
1310
1311         for ii in range(65):
1312             port_ip_hdr = (IPv6(dst="3000::1", src="3000:1::1") /
1313                            UDP(sport=1234, dport=1234 + ii) /
1314                            Raw('\xa5' * 100))
1315             port_ip_pkts.append((Ether(src=self.pg0.remote_mac,
1316                                        dst=self.pg0.local_mac) /
1317                                  port_ip_hdr))
1318             port_mpls_pkts.append((Ether(src=self.pg0.remote_mac,
1319                                          dst=self.pg0.local_mac) /
1320                                    MPLS(label=66, ttl=2) /
1321                                    port_ip_hdr))
1322             port_mpls_neos_pkts.append((Ether(src=self.pg0.remote_mac,
1323                                               dst=self.pg0.local_mac) /
1324                                         MPLS(label=67, ttl=2) /
1325                                         MPLS(label=77, ttl=2) /
1326                                         port_ip_hdr))
1327             port_ent_pkts.append((Ether(src=self.pg0.remote_mac,
1328                                         dst=self.pg0.local_mac) /
1329                                   MPLS(label=67, ttl=2) /
1330                                   MPLS(label=14, ttl=2) /
1331                                   MPLS(label=999, ttl=2) /
1332                                   port_ip_hdr))
1333             src_ip_hdr = (IPv6(dst="3000::1", src="3000:1::%d" % ii) /
1334                           UDP(sport=1234, dport=1234) /
1335                           Raw('\xa5' * 100))
1336             src_ip_pkts.append((Ether(src=self.pg0.remote_mac,
1337                                       dst=self.pg0.local_mac) /
1338                                 src_ip_hdr))
1339             src_mpls_pkts.append((Ether(src=self.pg0.remote_mac,
1340                                         dst=self.pg0.local_mac) /
1341                                   MPLS(label=66, ttl=2) /
1342                                   src_ip_hdr))
1343
1344         #
1345         # A route for the IP pacekts
1346         #
1347         route_3000_1 = VppIpRoute(self, "3000::1", 128,
1348                                   [VppRoutePath(self.pg1.remote_ip6,
1349                                                 self.pg1.sw_if_index,
1350                                                 is_ip6=1),
1351                                    VppRoutePath(self.pg2.remote_ip6,
1352                                                 self.pg2.sw_if_index,
1353                                                 is_ip6=1)],
1354                                   is_ip6=1)
1355         route_3000_1.add_vpp_config()
1356
1357         #
1358         # a local-label for the EOS packets
1359         #
1360         binding = VppMplsIpBind(self, 66, "3000::1", 128, is_ip6=1)
1361         binding.add_vpp_config()
1362
1363         #
1364         # An MPLS route for the non-EOS packets
1365         #
1366         route_67 = VppMplsRoute(self, 67, 0,
1367                                 [VppRoutePath(self.pg1.remote_ip6,
1368                                               self.pg1.sw_if_index,
1369                                               labels=[67],
1370                                               is_ip6=1),
1371                                  VppRoutePath(self.pg2.remote_ip6,
1372                                               self.pg2.sw_if_index,
1373                                               labels=[67],
1374                                               is_ip6=1)])
1375         route_67.add_vpp_config()
1376
1377         #
1378         # inject the packet on pg0 - expect load-balancing across the 2 paths
1379         #  - since the default hash config is to use IP src,dst and port
1380         #    src,dst
1381         # We are not going to ensure equal amounts of packets across each link,
1382         # since the hash algorithm is statistical and therefore this can never
1383         # be guaranteed. But wuth 64 different packets we do expect some
1384         # balancing. So instead just ensure there is traffic on each link.
1385         #
1386         self.send_and_expect_load_balancing(self.pg0, port_ip_pkts,
1387                                             [self.pg1, self.pg2])
1388         self.send_and_expect_load_balancing(self.pg0, src_ip_pkts,
1389                                             [self.pg1, self.pg2])
1390         self.send_and_expect_load_balancing(self.pg0, port_mpls_pkts,
1391                                             [self.pg1, self.pg2])
1392         self.send_and_expect_load_balancing(self.pg0, src_mpls_pkts,
1393                                             [self.pg1, self.pg2])
1394         self.send_and_expect_load_balancing(self.pg0, port_mpls_neos_pkts,
1395                                             [self.pg1, self.pg2])
1396
1397         #
1398         # The packets with Entropy label in should not load-balance,
1399         # since the Entorpy value is fixed.
1400         #
1401         self.send_and_expect_one_itf(self.pg0, port_ent_pkts, self.pg1)
1402
1403         #
1404         # change the flow hash config so it's only IP src,dst
1405         #  - now only the stream with differing source address will
1406         #    load-balance
1407         #
1408         self.vapi.set_ip_flow_hash(0, is_ip6=1, src=1, dst=1, sport=0, dport=0)
1409
1410         self.send_and_expect_load_balancing(self.pg0, src_ip_pkts,
1411                                             [self.pg1, self.pg2])
1412         self.send_and_expect_load_balancing(self.pg0, src_mpls_pkts,
1413                                             [self.pg1, self.pg2])
1414         self.send_and_expect_one_itf(self.pg0, port_ip_pkts, self.pg2)
1415
1416         #
1417         # change the flow hash config back to defaults
1418         #
1419         self.vapi.set_ip_flow_hash(0, is_ip6=1, src=1, dst=1, sport=1, dport=1)
1420
1421         #
1422         # Recursive prefixes
1423         #  - testing that 2 stages of load-balancing occurs and there is no
1424         #    polarisation (i.e. only 2 of 4 paths are used)
1425         #
1426         port_pkts = []
1427         src_pkts = []
1428
1429         for ii in range(257):
1430             port_pkts.append((Ether(src=self.pg0.remote_mac,
1431                                     dst=self.pg0.local_mac) /
1432                               IPv6(dst="4000::1", src="4000:1::1") /
1433                               UDP(sport=1234, dport=1234 + ii) /
1434                               Raw('\xa5' * 100)))
1435             src_pkts.append((Ether(src=self.pg0.remote_mac,
1436                                    dst=self.pg0.local_mac) /
1437                              IPv6(dst="4000::1", src="4000:1::%d" % ii) /
1438                              UDP(sport=1234, dport=1234) /
1439                              Raw('\xa5' * 100)))
1440
1441         route_3000_2 = VppIpRoute(self, "3000::2", 128,
1442                                   [VppRoutePath(self.pg3.remote_ip6,
1443                                                 self.pg3.sw_if_index,
1444                                                 is_ip6=1),
1445                                    VppRoutePath(self.pg4.remote_ip6,
1446                                                 self.pg4.sw_if_index,
1447                                                 is_ip6=1)],
1448                                   is_ip6=1)
1449         route_3000_2.add_vpp_config()
1450
1451         route_4000_1 = VppIpRoute(self, "4000::1", 128,
1452                                   [VppRoutePath("3000::1",
1453                                                 0xffffffff,
1454                                                 is_ip6=1),
1455                                    VppRoutePath("3000::2",
1456                                                 0xffffffff,
1457                                                 is_ip6=1)],
1458                                   is_ip6=1)
1459         route_4000_1.add_vpp_config()
1460
1461         #
1462         # inject the packet on pg0 - expect load-balancing across all 4 paths
1463         #
1464         self.vapi.cli("clear trace")
1465         self.send_and_expect_load_balancing(self.pg0, port_pkts,
1466                                             [self.pg1, self.pg2,
1467                                              self.pg3, self.pg4])
1468         self.send_and_expect_load_balancing(self.pg0, src_pkts,
1469                                             [self.pg1, self.pg2,
1470                                              self.pg3, self.pg4])
1471
1472
1473 if __name__ == '__main__':
1474     unittest.main(testRunner=VppTestRunner)