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