test/test_ip6.py

   1 #!/usr/bin/env python
   2
   3 import unittest
   4 import socket
   5
   6 from framework import VppTestCase, VppTestRunner
   7 from util import ppp, ip6_normalize
   8 from vpp_sub_interface import VppSubInterface, VppDot1QSubint
   9 from vpp_pg_interface import is_ipv6_misc
  10 from vpp_ip import DpoProto
  11 from vpp_ip_route import VppIpRoute, VppRoutePath, find_route, VppIpMRoute, \
  12     VppMRoutePath, MRouteItfFlags, MRouteEntryFlags, VppMplsIpBind, \
  13     VppMplsRoute, VppMplsTable, VppIpTable, VppIpAddress
  14 from vpp_neighbor import find_nbr, VppNeighbor
  15
  16 from scapy.packet import Raw
  17 from scapy.layers.l2 import Ether, Dot1Q
  18 import scapy.layers.inet6 as inet6
  19 from scapy.layers.inet6 import IPv6, ICMPv6ND_NS, ICMPv6ND_RS, \
  20     ICMPv6ND_RA, getmacbyip6, ICMPv6MRD_Solicitation, \
  21     ICMPv6NDOptMTU, ICMPv6NDOptSrcLLAddr, ICMPv6NDOptPrefixInfo, \
  22     ICMPv6ND_NA, ICMPv6NDOptDstLLAddr, ICMPv6DestUnreach, icmp6types, \
  23     ICMPv6TimeExceeded, ICMPv6EchoRequest, ICMPv6EchoReply
  24 from scapy.utils6 import in6_getnsma, in6_getnsmac, in6_ptop, in6_islladdr, \
  25     in6_mactoifaceid, in6_ismaddr
  26 from scapy.utils import inet_pton, inet_ntop
  27 from scapy.contrib.mpls import MPLS
  28
  29 AF_INET6 = socket.AF_INET6
  30
  31
  32 def mk_ll_addr(mac):
  33     euid = in6_mactoifaceid(mac)
  34     addr = "fe80::" + euid
  35     return addr
  36
  37
  38 class TestIPv6ND(VppTestCase):
  39     def validate_ra(self, intf, rx, dst_ip=None):
  40         if not dst_ip:
  41             dst_ip = intf.remote_ip6
  42
  43         # unicasted packets must come to the unicast mac
  44         self.assertEqual(rx[Ether].dst, intf.remote_mac)
  45
  46         # and from the router's MAC
  47         self.assertEqual(rx[Ether].src, intf.local_mac)
  48
  49         # the rx'd RA should be addressed to the sender's source
  50         self.assertTrue(rx.haslayer(ICMPv6ND_RA))
  51         self.assertEqual(in6_ptop(rx[IPv6].dst),
  52                          in6_ptop(dst_ip))
  53
  54         # and come from the router's link local
  55         self.assertTrue(in6_islladdr(rx[IPv6].src))
  56         self.assertEqual(in6_ptop(rx[IPv6].src),
  57                          in6_ptop(mk_ll_addr(intf.local_mac)))
  58
  59     def validate_na(self, intf, rx, dst_ip=None, tgt_ip=None):
  60         if not dst_ip:
  61             dst_ip = intf.remote_ip6
  62         if not tgt_ip:
  63             dst_ip = intf.local_ip6
  64
  65         # unicasted packets must come to the unicast mac
  66         self.assertEqual(rx[Ether].dst, intf.remote_mac)
  67
  68         # and from the router's MAC
  69         self.assertEqual(rx[Ether].src, intf.local_mac)
  70
  71         # the rx'd NA should be addressed to the sender's source
  72         self.assertTrue(rx.haslayer(ICMPv6ND_NA))
  73         self.assertEqual(in6_ptop(rx[IPv6].dst),
  74                          in6_ptop(dst_ip))
  75
  76         # and come from the target address
  77         self.assertEqual(
  78             in6_ptop(rx[IPv6].src), in6_ptop(tgt_ip))
  79
  80         # Dest link-layer options should have the router's MAC
  81         dll = rx[ICMPv6NDOptDstLLAddr]
  82         self.assertEqual(dll.lladdr, intf.local_mac)
  83
  84     def validate_ns(self, intf, rx, tgt_ip):
  85         nsma = in6_getnsma(inet_pton(AF_INET6, tgt_ip))
  86         dst_ip = inet_ntop(AF_INET6, nsma)
  87
  88         # NS is broadcast
  89         self.assertEqual(rx[Ether].dst, in6_getnsmac(nsma))
  90
  91         # and from the router's MAC
  92         self.assertEqual(rx[Ether].src, intf.local_mac)
  93
  94         # the rx'd NS should be addressed to an mcast address
  95         # derived from the target address
  96         self.assertEqual(
  97             in6_ptop(rx[IPv6].dst), in6_ptop(dst_ip))
  98
  99         # expect the tgt IP in the NS header
 100         ns = rx[ICMPv6ND_NS]
 101         self.assertEqual(in6_ptop(ns.tgt), in6_ptop(tgt_ip))
 102
 103         # packet is from the router's local address
 104         self.assertEqual(
 105             in6_ptop(rx[IPv6].src), intf.local_ip6)
 106
 107         # Src link-layer options should have the router's MAC
 108         sll = rx[ICMPv6NDOptSrcLLAddr]
 109         self.assertEqual(sll.lladdr, intf.local_mac)
 110
 111     def send_and_expect_ra(self, intf, pkts, remark, dst_ip=None,
 112                            filter_out_fn=is_ipv6_misc):
 113         intf.add_stream(pkts)
 114         self.pg_enable_capture(self.pg_interfaces)
 115         self.pg_start()
 116         rx = intf.get_capture(1, filter_out_fn=filter_out_fn)
 117
 118         self.assertEqual(len(rx), 1)
 119         rx = rx[0]
 120         self.validate_ra(intf, rx, dst_ip)
 121
 122     def send_and_expect_na(self, intf, pkts, remark, dst_ip=None,
 123                            tgt_ip=None,
 124                            filter_out_fn=is_ipv6_misc):
 125         intf.add_stream(pkts)
 126         self.pg_enable_capture(self.pg_interfaces)
 127         self.pg_start()
 128         rx = intf.get_capture(1, filter_out_fn=filter_out_fn)
 129
 130         self.assertEqual(len(rx), 1)
 131         rx = rx[0]
 132         self.validate_na(intf, rx, dst_ip, tgt_ip)
 133
 134     def send_and_expect_ns(self, tx_intf, rx_intf, pkts, tgt_ip,
 135                            filter_out_fn=is_ipv6_misc):
 136         tx_intf.add_stream(pkts)
 137         self.pg_enable_capture(self.pg_interfaces)
 138         self.pg_start()
 139         rx = rx_intf.get_capture(1, filter_out_fn=filter_out_fn)
 140
 141         self.assertEqual(len(rx), 1)
 142         rx = rx[0]
 143         self.validate_ns(rx_intf, rx, tgt_ip)
 144
 145     def verify_ip(self, rx, smac, dmac, sip, dip):
 146         ether = rx[Ether]
 147         self.assertEqual(ether.dst, dmac)
 148         self.assertEqual(ether.src, smac)
 149
 150         ip = rx[IPv6]
 151         self.assertEqual(ip.src, sip)
 152         self.assertEqual(ip.dst, dip)
 153
 154
 155 class TestIPv6(TestIPv6ND):
 156     """ IPv6 Test Case """
 157
 158     @classmethod
 159     def setUpClass(cls):
 160         super(TestIPv6, cls).setUpClass()
 161
 162     def setUp(self):
 163         """
 164         Perform test setup before test case.
 165
 166         **Config:**
 167             - create 3 pg interfaces
 168                 - untagged pg0 interface
 169                 - Dot1Q subinterface on pg1
 170                 - Dot1AD subinterface on pg2
 171             - setup interfaces:
 172                 - put it into UP state
 173                 - set IPv6 addresses
 174                 - resolve neighbor address using NDP
 175             - configure 200 fib entries
 176
 177         :ivar list interfaces: pg interfaces and subinterfaces.
 178         :ivar dict flows: IPv4 packet flows in test.
 179
 180         *TODO:* Create AD sub interface
 181         """
 182         super(TestIPv6, self).setUp()
 183
 184         # create 3 pg interfaces
 185         self.create_pg_interfaces(range(3))
 186
 187         # create 2 subinterfaces for p1 and pg2
 188         self.sub_interfaces = [
 189             VppDot1QSubint(self, self.pg1, 100),
 190             VppDot1QSubint(self, self.pg2, 200)
 191             # TODO: VppDot1ADSubint(self, self.pg2, 200, 300, 400)
 192         ]
 193
 194         # packet flows mapping pg0 -> pg1.sub, pg2.sub, etc.
 195         self.flows = dict()
 196         self.flows[self.pg0] = [self.pg1.sub_if, self.pg2.sub_if]
 197         self.flows[self.pg1.sub_if] = [self.pg0, self.pg2.sub_if]
 198         self.flows[self.pg2.sub_if] = [self.pg0, self.pg1.sub_if]
 199
 200         # packet sizes
 201         self.pg_if_packet_sizes = [64, 1500, 9020]
 202
 203         self.interfaces = list(self.pg_interfaces)
 204         self.interfaces.extend(self.sub_interfaces)
 205
 206         # setup all interfaces
 207         for i in self.interfaces:
 208             i.admin_up()
 209             i.config_ip6()
 210             i.resolve_ndp()
 211
 212         # config 2M FIB entries
 213         self.config_fib_entries(200)
 214
 215     def tearDown(self):
 216         """Run standard test teardown and log ``show ip6 neighbors``."""
 217         for i in self.interfaces:
 218             i.unconfig_ip6()
 219             i.ip6_disable()
 220             i.admin_down()
 221         for i in self.sub_interfaces:
 222             i.remove_vpp_config()
 223
 224         super(TestIPv6, self).tearDown()
 225         if not self.vpp_dead:
 226             self.logger.info(self.vapi.cli("show ip6 neighbors"))
 227             # info(self.vapi.cli("show ip6 fib"))  # many entries
 228
 229     def config_fib_entries(self, count):
 230         """For each interface add to the FIB table *count* routes to
 231         "fd02::1/128" destination with interface's local address as next-hop
 232         address.
 233
 234         :param int count: Number of FIB entries.
 235
 236         - *TODO:* check if the next-hop address shouldn't be remote address
 237           instead of local address.
 238         """
 239         n_int = len(self.interfaces)
 240         percent = 0
 241         counter = 0.0
 242         dest_addr = inet_pton(AF_INET6, "fd02::1")
 243         dest_addr_len = 128
 244         for i in self.interfaces:
 245             next_hop_address = i.local_ip6n
 246             for j in range(count / n_int):
 247                 self.vapi.ip_add_del_route(
 248                     dest_addr, dest_addr_len, next_hop_address, is_ipv6=1)
 249                 counter += 1
 250                 if counter / count * 100 > percent:
 251                     self.logger.info("Configure %d FIB entries .. %d%% done" %
 252                                      (count, percent))
 253                     percent += 1
 254
 255     def modify_packet(self, src_if, packet_size, pkt):
 256         """Add load, set destination IP and extend packet to required packet
 257         size for defined interface.
 258
 259         :param VppInterface src_if: Interface to create packet for.
 260         :param int packet_size: Required packet size.
 261         :param Scapy pkt: Packet to be modified.
 262         """
 263         dst_if_idx = packet_size / 10 % 2
 264         dst_if = self.flows[src_if][dst_if_idx]
 265         info = self.create_packet_info(src_if, dst_if)
 266         payload = self.info_to_payload(info)
 267         p = pkt / Raw(payload)
 268         p[IPv6].dst = dst_if.remote_ip6
 269         info.data = p.copy()
 270         if isinstance(src_if, VppSubInterface):
 271             p = src_if.add_dot1_layer(p)
 272         self.extend_packet(p, packet_size)
 273
 274         return p
 275
 276     def create_stream(self, src_if):
 277         """Create input packet stream for defined interface.
 278
 279         :param VppInterface src_if: Interface to create packet stream for.
 280         """
 281         hdr_ext = 4 if isinstance(src_if, VppSubInterface) else 0
 282         pkt_tmpl = (Ether(dst=src_if.local_mac, src=src_if.remote_mac) /
 283                     IPv6(src=src_if.remote_ip6) /
 284                     inet6.UDP(sport=1234, dport=1234))
 285
 286         pkts = [self.modify_packet(src_if, i, pkt_tmpl)
 287                 for i in xrange(self.pg_if_packet_sizes[0],
 288                                 self.pg_if_packet_sizes[1], 10)]
 289         pkts_b = [self.modify_packet(src_if, i, pkt_tmpl)
 290                   for i in xrange(self.pg_if_packet_sizes[1] + hdr_ext,
 291                                   self.pg_if_packet_sizes[2] + hdr_ext, 50)]
 292         pkts.extend(pkts_b)
 293
 294         return pkts
 295
 296     def verify_capture(self, dst_if, capture):
 297         """Verify captured input packet stream for defined interface.
 298
 299         :param VppInterface dst_if: Interface to verify captured packet stream
 300                                     for.
 301         :param list capture: Captured packet stream.
 302         """
 303         self.logger.info("Verifying capture on interface %s" % dst_if.name)
 304         last_info = dict()
 305         for i in self.interfaces:
 306             last_info[i.sw_if_index] = None
 307         is_sub_if = False
 308         dst_sw_if_index = dst_if.sw_if_index
 309         if hasattr(dst_if, 'parent'):
 310             is_sub_if = True
 311         for packet in capture:
 312             if is_sub_if:
 313                 # Check VLAN tags and Ethernet header
 314                 packet = dst_if.remove_dot1_layer(packet)
 315             self.assertTrue(Dot1Q not in packet)
 316             try:
 317                 ip = packet[IPv6]
 318                 udp = packet[inet6.UDP]
 319                 payload_info = self.payload_to_info(str(packet[Raw]))
 320                 packet_index = payload_info.index
 321                 self.assertEqual(payload_info.dst, dst_sw_if_index)
 322                 self.logger.debug(
 323                     "Got packet on port %s: src=%u (id=%u)" %
 324                     (dst_if.name, payload_info.src, packet_index))
 325                 next_info = self.get_next_packet_info_for_interface2(
 326                     payload_info.src, dst_sw_if_index,
 327                     last_info[payload_info.src])
 328                 last_info[payload_info.src] = next_info
 329                 self.assertTrue(next_info is not None)
 330                 self.assertEqual(packet_index, next_info.index)
 331                 saved_packet = next_info.data
 332                 # Check standard fields
 333                 self.assertEqual(
 334                     ip.src, saved_packet[IPv6].src)
 335                 self.assertEqual(
 336                     ip.dst, saved_packet[IPv6].dst)
 337                 self.assertEqual(
 338                     udp.sport, saved_packet[inet6.UDP].sport)
 339                 self.assertEqual(
 340                     udp.dport, saved_packet[inet6.UDP].dport)
 341             except:
 342                 self.logger.error(ppp("Unexpected or invalid packet:", packet))
 343                 raise
 344         for i in self.interfaces:
 345             remaining_packet = self.get_next_packet_info_for_interface2(
 346                 i.sw_if_index, dst_sw_if_index, last_info[i.sw_if_index])
 347             self.assertTrue(remaining_packet is None,
 348                             "Interface %s: Packet expected from interface %s "
 349                             "didn't arrive" % (dst_if.name, i.name))
 350
 351     def test_fib(self):
 352         """ IPv6 FIB test
 353
 354         Test scenario:
 355             - Create IPv6 stream for pg0 interface
 356             - Create IPv6 tagged streams for pg1's and pg2's subinterface.
 357             - Send and verify received packets on each interface.
 358         """
 359
 360         pkts = self.create_stream(self.pg0)
 361         self.pg0.add_stream(pkts)
 362
 363         for i in self.sub_interfaces:
 364             pkts = self.create_stream(i)
 365             i.parent.add_stream(pkts)
 366
 367         self.pg_enable_capture(self.pg_interfaces)
 368         self.pg_start()
 369
 370         pkts = self.pg0.get_capture()
 371         self.verify_capture(self.pg0, pkts)
 372
 373         for i in self.sub_interfaces:
 374             pkts = i.parent.get_capture()
 375             self.verify_capture(i, pkts)
 376
 377     def test_ns(self):
 378         """ IPv6 Neighbour Solicitation Exceptions
 379
 380         Test scenario:
 381            - Send an NS Sourced from an address not covered by the link sub-net
 382            - Send an NS to an mcast address the router has not joined
 383            - Send NS for a target address the router does not onn.
 384         """
 385
 386         #
 387         # An NS from a non link source address
 388         #
 389         nsma = in6_getnsma(inet_pton(AF_INET6, self.pg0.local_ip6))
 390         d = inet_ntop(AF_INET6, nsma)
 391
 392         p = (Ether(dst=in6_getnsmac(nsma)) /
 393              IPv6(dst=d, src="2002::2") /
 394              ICMPv6ND_NS(tgt=self.pg0.local_ip6) /
 395              ICMPv6NDOptSrcLLAddr(
 396                  lladdr=self.pg0.remote_mac))
 397         pkts = [p]
 398
 399         self.send_and_assert_no_replies(
 400             self.pg0, pkts,
 401             "No response to NS source by address not on sub-net")
 402
 403         #
 404         # An NS for sent to a solicited mcast group the router is
 405         # not a member of FAILS
 406         #
 407         if 0:
 408             nsma = in6_getnsma(inet_pton(AF_INET6, "fd::ffff"))
 409             d = inet_ntop(AF_INET6, nsma)
 410
 411             p = (Ether(dst=in6_getnsmac(nsma)) /
 412                  IPv6(dst=d, src=self.pg0.remote_ip6) /
 413                  ICMPv6ND_NS(tgt=self.pg0.local_ip6) /
 414                  ICMPv6NDOptSrcLLAddr(
 415                      lladdr=self.pg0.remote_mac))
 416             pkts = [p]
 417
 418             self.send_and_assert_no_replies(
 419                 self.pg0, pkts,
 420                 "No response to NS sent to unjoined mcast address")
 421
 422         #
 423         # An NS whose target address is one the router does not own
 424         #
 425         nsma = in6_getnsma(inet_pton(AF_INET6, self.pg0.local_ip6))
 426         d = inet_ntop(AF_INET6, nsma)
 427
 428         p = (Ether(dst=in6_getnsmac(nsma)) /
 429              IPv6(dst=d, src=self.pg0.remote_ip6) /
 430              ICMPv6ND_NS(tgt="fd::ffff") /
 431              ICMPv6NDOptSrcLLAddr(
 432                  lladdr=self.pg0.remote_mac))
 433         pkts = [p]
 434
 435         self.send_and_assert_no_replies(self.pg0, pkts,
 436                                         "No response to NS for unknown target")
 437
 438         #
 439         # A neighbor entry that has no associated FIB-entry
 440         #
 441         self.pg0.generate_remote_hosts(4)
 442         nd_entry = VppNeighbor(self,
 443                                self.pg0.sw_if_index,
 444                                self.pg0.remote_hosts[2].mac,
 445                                self.pg0.remote_hosts[2].ip6,
 446                                af=AF_INET6,
 447                                is_no_fib_entry=1)
 448         nd_entry.add_vpp_config()
 449
 450         #
 451         # check we have the neighbor, but no route
 452         #
 453         self.assertTrue(find_nbr(self,
 454                                  self.pg0.sw_if_index,
 455                                  self.pg0._remote_hosts[2].ip6,
 456                                  inet=AF_INET6))
 457         self.assertFalse(find_route(self,
 458                                     self.pg0._remote_hosts[2].ip6,
 459                                     128,
 460                                     inet=AF_INET6))
 461
 462         #
 463         # send an NS from a link local address to the interface's global
 464         # address
 465         #
 466         p = (Ether(dst=in6_getnsmac(nsma), src=self.pg0.remote_mac) /
 467              IPv6(
 468                  dst=d, src=self.pg0._remote_hosts[2].ip6_ll) /
 469              ICMPv6ND_NS(tgt=self.pg0.local_ip6) /
 470              ICMPv6NDOptSrcLLAddr(
 471                  lladdr=self.pg0.remote_mac))
 472
 473         self.send_and_expect_na(self.pg0, p,
 474                                 "NS from link-local",
 475                                 dst_ip=self.pg0._remote_hosts[2].ip6_ll,
 476                                 tgt_ip=self.pg0.local_ip6)
 477
 478         #
 479         # we should have learned an ND entry for the peer's link-local
 480         # but not inserted a route to it in the FIB
 481         #
 482         self.assertTrue(find_nbr(self,
 483                                  self.pg0.sw_if_index,
 484                                  self.pg0._remote_hosts[2].ip6_ll,
 485                                  inet=AF_INET6))
 486         self.assertFalse(find_route(self,
 487                                     self.pg0._remote_hosts[2].ip6_ll,
 488                                     128,
 489                                     inet=AF_INET6))
 490
 491         #
 492         # An NS to the router's own Link-local
 493         #
 494         p = (Ether(dst=in6_getnsmac(nsma), src=self.pg0.remote_mac) /
 495              IPv6(
 496                  dst=d, src=self.pg0._remote_hosts[3].ip6_ll) /
 497              ICMPv6ND_NS(tgt=self.pg0.local_ip6_ll) /
 498              ICMPv6NDOptSrcLLAddr(
 499                  lladdr=self.pg0.remote_mac))
 500
 501         self.send_and_expect_na(self.pg0, p,
 502                                 "NS to/from link-local",
 503                                 dst_ip=self.pg0._remote_hosts[3].ip6_ll,
 504                                 tgt_ip=self.pg0.local_ip6_ll)
 505
 506         #
 507         # we should have learned an ND entry for the peer's link-local
 508         # but not inserted a route to it in the FIB
 509         #
 510         self.assertTrue(find_nbr(self,
 511                                  self.pg0.sw_if_index,
 512                                  self.pg0._remote_hosts[3].ip6_ll,
 513                                  inet=AF_INET6))
 514         self.assertFalse(find_route(self,
 515                                     self.pg0._remote_hosts[3].ip6_ll,
 516                                     128,
 517                                     inet=AF_INET6))
 518
 519     def test_ns_duplicates(self):
 520         """ ND Duplicates"""
 521
 522         #
 523         # Generate some hosts on the LAN
 524         #
 525         self.pg1.generate_remote_hosts(3)
 526
 527         #
 528         # Add host 1 on pg1 and pg2
 529         #
 530         ns_pg1 = VppNeighbor(self,
 531                              self.pg1.sw_if_index,
 532                              self.pg1.remote_hosts[1].mac,
 533                              self.pg1.remote_hosts[1].ip6,
 534                              af=AF_INET6)
 535         ns_pg1.add_vpp_config()
 536         ns_pg2 = VppNeighbor(self,
 537                              self.pg2.sw_if_index,
 538                              self.pg2.remote_mac,
 539                              self.pg1.remote_hosts[1].ip6,
 540                              af=AF_INET6)
 541         ns_pg2.add_vpp_config()
 542
 543         #
 544         # IP packet destined for pg1 remote host arrives on pg1 again.
 545         #
 546         p = (Ether(dst=self.pg0.local_mac,
 547                    src=self.pg0.remote_mac) /
 548              IPv6(src=self.pg0.remote_ip6,
 549                   dst=self.pg1.remote_hosts[1].ip6) /
 550              inet6.UDP(sport=1234, dport=1234) /
 551              Raw())
 552
 553         self.pg0.add_stream(p)
 554         self.pg_enable_capture(self.pg_interfaces)
 555         self.pg_start()
 556
 557         rx1 = self.pg1.get_capture(1)
 558
 559         self.verify_ip(rx1[0],
 560                        self.pg1.local_mac,
 561                        self.pg1.remote_hosts[1].mac,
 562                        self.pg0.remote_ip6,
 563                        self.pg1.remote_hosts[1].ip6)
 564
 565         #
 566         # remove the duplicate on pg1
 567         # packet stream shoud generate NSs out of pg1
 568         #
 569         ns_pg1.remove_vpp_config()
 570
 571         self.send_and_expect_ns(self.pg0, self.pg1,
 572                                 p, self.pg1.remote_hosts[1].ip6)
 573
 574         #
 575         # Add it back
 576         #
 577         ns_pg1.add_vpp_config()
 578
 579         self.pg0.add_stream(p)
 580         self.pg_enable_capture(self.pg_interfaces)
 581         self.pg_start()
 582
 583         rx1 = self.pg1.get_capture(1)
 584
 585         self.verify_ip(rx1[0],
 586                        self.pg1.local_mac,
 587                        self.pg1.remote_hosts[1].mac,
 588                        self.pg0.remote_ip6,
 589                        self.pg1.remote_hosts[1].ip6)
 590
 591     def validate_ra(self, intf, rx, dst_ip=None, mtu=9000, pi_opt=None):
 592         if not dst_ip:
 593             dst_ip = intf.remote_ip6
 594
 595         # unicasted packets must come to the unicast mac
 596         self.assertEqual(rx[Ether].dst, intf.remote_mac)
 597
 598         # and from the router's MAC
 599         self.assertEqual(rx[Ether].src, intf.local_mac)
 600
 601         # the rx'd RA should be addressed to the sender's source
 602         self.assertTrue(rx.haslayer(ICMPv6ND_RA))
 603         self.assertEqual(in6_ptop(rx[IPv6].dst),
 604                          in6_ptop(dst_ip))
 605
 606         # and come from the router's link local
 607         self.assertTrue(in6_islladdr(rx[IPv6].src))
 608         self.assertEqual(in6_ptop(rx[IPv6].src),
 609                          in6_ptop(mk_ll_addr(intf.local_mac)))
 610
 611         # it should contain the links MTU
 612         ra = rx[ICMPv6ND_RA]
 613         self.assertEqual(ra[ICMPv6NDOptMTU].mtu, mtu)
 614
 615         # it should contain the source's link layer address option
 616         sll = ra[ICMPv6NDOptSrcLLAddr]
 617         self.assertEqual(sll.lladdr, intf.local_mac)
 618
 619         if not pi_opt:
 620             # the RA should not contain prefix information
 621             self.assertFalse(ra.haslayer(
 622                 ICMPv6NDOptPrefixInfo))
 623         else:
 624             raos = rx.getlayer(ICMPv6NDOptPrefixInfo, 1)
 625
 626             # the options are nested in the scapy packet in way that i cannot
 627             # decipher how to decode. this 1st layer of option always returns
 628             # nested classes, so a direct obj1=obj2 comparison always fails.
 629             # however, the getlayer(.., 2) does give one instnace.
 630             # so we cheat here and construct a new opt instnace for comparison
 631             rd = ICMPv6NDOptPrefixInfo(
 632                 prefixlen=raos.prefixlen,
 633                 prefix=raos.prefix,
 634                 L=raos.L,
 635                 A=raos.A)
 636             if type(pi_opt) is list:
 637                 for ii in range(len(pi_opt)):
 638                     self.assertEqual(pi_opt[ii], rd)
 639                     rd = rx.getlayer(
 640                         ICMPv6NDOptPrefixInfo, ii + 2)
 641             else:
 642                 self.assertEqual(pi_opt, raos)
 643
 644     def send_and_expect_ra(self, intf, pkts, remark, dst_ip=None,
 645                            filter_out_fn=is_ipv6_misc,
 646                            opt=None):
 647         intf.add_stream(pkts)
 648         self.pg_enable_capture(self.pg_interfaces)
 649         self.pg_start()
 650         rx = intf.get_capture(1, filter_out_fn=filter_out_fn)
 651
 652         self.assertEqual(len(rx), 1)
 653         rx = rx[0]
 654         self.validate_ra(intf, rx, dst_ip, pi_opt=opt)
 655
 656     def test_rs(self):
 657         """ IPv6 Router Solicitation Exceptions
 658
 659         Test scenario:
 660         """
 661
 662         #
 663         # Before we begin change the IPv6 RA responses to use the unicast
 664         # address - that way we will not confuse them with the periodic
 665         # RAs which go to the mcast address
 666         # Sit and wait for the first periodic RA.
 667         #
 668         # TODO
 669         #
 670         self.pg0.ip6_ra_config(send_unicast=1)
 671
 672         #
 673         # An RS from a link source address
 674         #  - expect an RA in return
 675         #
 676         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
 677              IPv6(
 678                  dst=self.pg0.local_ip6, src=self.pg0.remote_ip6) /
 679              ICMPv6ND_RS())
 680         pkts = [p]
 681         self.send_and_expect_ra(self.pg0, pkts, "Genuine RS")
 682
 683         #
 684         # For the next RS sent the RA should be rate limited
 685         #
 686         self.send_and_assert_no_replies(self.pg0, pkts, "RA rate limited")
 687
 688         #
 689         # When we reconfiure the IPv6 RA config, we reset the RA rate limiting,
 690         # so we need to do this before each test below so as not to drop
 691         # packets for rate limiting reasons. Test this works here.
 692         #
 693         self.pg0.ip6_ra_config(send_unicast=1)
 694         self.send_and_expect_ra(self.pg0, pkts, "Rate limit reset RS")
 695
 696         #
 697         # An RS sent from a non-link local source
 698         #
 699         self.pg0.ip6_ra_config(send_unicast=1)
 700         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
 701              IPv6(dst=self.pg0.local_ip6,
 702                   src="2002::ffff") /
 703              ICMPv6ND_RS())
 704         pkts = [p]
 705         self.send_and_assert_no_replies(self.pg0, pkts,
 706                                         "RS from non-link source")
 707
 708         #
 709         # Source an RS from a link local address
 710         #
 711         self.pg0.ip6_ra_config(send_unicast=1)
 712         ll = mk_ll_addr(self.pg0.remote_mac)
 713         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
 714              IPv6(dst=self.pg0.local_ip6, src=ll) /
 715              ICMPv6ND_RS())
 716         pkts = [p]
 717         self.send_and_expect_ra(self.pg0, pkts,
 718                                 "RS sourced from link-local",
 719                                 dst_ip=ll)
 720
 721         #
 722         # Send the RS multicast
 723         #
 724         self.pg0.ip6_ra_config(send_unicast=1)
 725         dmac = in6_getnsmac(inet_pton(AF_INET6, "ff02::2"))
 726         ll = mk_ll_addr(self.pg0.remote_mac)
 727         p = (Ether(dst=dmac, src=self.pg0.remote_mac) /
 728              IPv6(dst="ff02::2", src=ll) /
 729              ICMPv6ND_RS())
 730         pkts = [p]
 731         self.send_and_expect_ra(self.pg0, pkts,
 732                                 "RS sourced from link-local",
 733                                 dst_ip=ll)
 734
 735         #
 736         # Source from the unspecified address ::. This happens when the RS
 737         # is sent before the host has a configured address/sub-net,
 738         # i.e. auto-config. Since the sender has no IP address, the reply
 739         # comes back mcast - so the capture needs to not filter this.
 740         # If we happen to pick up the periodic RA at this point then so be it,
 741         # it's not an error.
 742         #
 743         self.pg0.ip6_ra_config(send_unicast=1, suppress=1)
 744         p = (Ether(dst=dmac, src=self.pg0.remote_mac) /
 745              IPv6(dst="ff02::2", src="::") /
 746              ICMPv6ND_RS())
 747         pkts = [p]
 748         self.send_and_expect_ra(self.pg0, pkts,
 749                                 "RS sourced from unspecified",
 750                                 dst_ip="ff02::1",
 751                                 filter_out_fn=None)
 752
 753         #
 754         # Configure The RA to announce the links prefix
 755         #
 756         self.pg0.ip6_ra_prefix(self.pg0.local_ip6n,
 757                                self.pg0.local_ip6_prefix_len)
 758
 759         #
 760         # RAs should now contain the prefix information option
 761         #
 762         opt = ICMPv6NDOptPrefixInfo(
 763             prefixlen=self.pg0.local_ip6_prefix_len,
 764             prefix=self.pg0.local_ip6,
 765             L=1,
 766             A=1)
 767
 768         self.pg0.ip6_ra_config(send_unicast=1)
 769         ll = mk_ll_addr(self.pg0.remote_mac)
 770         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
 771              IPv6(dst=self.pg0.local_ip6, src=ll) /
 772              ICMPv6ND_RS())
 773         self.send_and_expect_ra(self.pg0, p,
 774                                 "RA with prefix-info",
 775                                 dst_ip=ll,
 776                                 opt=opt)
 777
 778         #
 779         # Change the prefix info to not off-link
 780         #  L-flag is clear
 781         #
 782         self.pg0.ip6_ra_prefix(self.pg0.local_ip6n,
 783                                self.pg0.local_ip6_prefix_len,
 784                                off_link=1)
 785
 786         opt = ICMPv6NDOptPrefixInfo(
 787             prefixlen=self.pg0.local_ip6_prefix_len,
 788             prefix=self.pg0.local_ip6,
 789             L=0,
 790             A=1)
 791
 792         self.pg0.ip6_ra_config(send_unicast=1)
 793         self.send_and_expect_ra(self.pg0, p,
 794                                 "RA with Prefix info with L-flag=0",
 795                                 dst_ip=ll,
 796                                 opt=opt)
 797
 798         #
 799         # Change the prefix info to not off-link, no-autoconfig
 800         #  L and A flag are clear in the advert
 801         #
 802         self.pg0.ip6_ra_prefix(self.pg0.local_ip6n,
 803                                self.pg0.local_ip6_prefix_len,
 804                                off_link=1,
 805                                no_autoconfig=1)
 806
 807         opt = ICMPv6NDOptPrefixInfo(
 808             prefixlen=self.pg0.local_ip6_prefix_len,
 809             prefix=self.pg0.local_ip6,
 810             L=0,
 811             A=0)
 812
 813         self.pg0.ip6_ra_config(send_unicast=1)
 814         self.send_and_expect_ra(self.pg0, p,
 815                                 "RA with Prefix info with A & L-flag=0",
 816                                 dst_ip=ll,
 817                                 opt=opt)
 818
 819         #
 820         # Change the flag settings back to the defaults
 821         #  L and A flag are set in the advert
 822         #
 823         self.pg0.ip6_ra_prefix(self.pg0.local_ip6n,
 824                                self.pg0.local_ip6_prefix_len)
 825
 826         opt = ICMPv6NDOptPrefixInfo(
 827             prefixlen=self.pg0.local_ip6_prefix_len,
 828             prefix=self.pg0.local_ip6,
 829             L=1,
 830             A=1)
 831
 832         self.pg0.ip6_ra_config(send_unicast=1)
 833         self.send_and_expect_ra(self.pg0, p,
 834                                 "RA with Prefix info",
 835                                 dst_ip=ll,
 836                                 opt=opt)
 837
 838         #
 839         # Change the prefix info to not off-link, no-autoconfig
 840         #  L and A flag are clear in the advert
 841         #
 842         self.pg0.ip6_ra_prefix(self.pg0.local_ip6n,
 843                                self.pg0.local_ip6_prefix_len,
 844                                off_link=1,
 845                                no_autoconfig=1)
 846
 847         opt = ICMPv6NDOptPrefixInfo(
 848             prefixlen=self.pg0.local_ip6_prefix_len,
 849             prefix=self.pg0.local_ip6,
 850             L=0,
 851             A=0)
 852
 853         self.pg0.ip6_ra_config(send_unicast=1)
 854         self.send_and_expect_ra(self.pg0, p,
 855                                 "RA with Prefix info with A & L-flag=0",
 856                                 dst_ip=ll,
 857                                 opt=opt)
 858
 859         #
 860         # Use the reset to defults option to revert to defaults
 861         #  L and A flag are clear in the advert
 862         #
 863         self.pg0.ip6_ra_prefix(self.pg0.local_ip6n,
 864                                self.pg0.local_ip6_prefix_len,
 865                                use_default=1)
 866
 867         opt = ICMPv6NDOptPrefixInfo(
 868             prefixlen=self.pg0.local_ip6_prefix_len,
 869             prefix=self.pg0.local_ip6,
 870             L=1,
 871             A=1)
 872
 873         self.pg0.ip6_ra_config(send_unicast=1)
 874         self.send_and_expect_ra(self.pg0, p,
 875                                 "RA with Prefix reverted to defaults",
 876                                 dst_ip=ll,
 877                                 opt=opt)
 878
 879         #
 880         # Advertise Another prefix. With no L-flag/A-flag
 881         #
 882         self.pg0.ip6_ra_prefix(self.pg1.local_ip6n,
 883                                self.pg1.local_ip6_prefix_len,
 884                                off_link=1,
 885                                no_autoconfig=1)
 886
 887         opt = [ICMPv6NDOptPrefixInfo(
 888             prefixlen=self.pg0.local_ip6_prefix_len,
 889             prefix=self.pg0.local_ip6,
 890             L=1,
 891             A=1),
 892             ICMPv6NDOptPrefixInfo(
 893                 prefixlen=self.pg1.local_ip6_prefix_len,
 894                 prefix=self.pg1.local_ip6,
 895                 L=0,
 896                 A=0)]
 897
 898         self.pg0.ip6_ra_config(send_unicast=1)
 899         ll = mk_ll_addr(self.pg0.remote_mac)
 900         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
 901              IPv6(dst=self.pg0.local_ip6, src=ll) /
 902              ICMPv6ND_RS())
 903         self.send_and_expect_ra(self.pg0, p,
 904                                 "RA with multiple Prefix infos",
 905                                 dst_ip=ll,
 906                                 opt=opt)
 907
 908         #
 909         # Remove the first refix-info - expect the second is still in the
 910         # advert
 911         #
 912         self.pg0.ip6_ra_prefix(self.pg0.local_ip6n,
 913                                self.pg0.local_ip6_prefix_len,
 914                                is_no=1)
 915
 916         opt = ICMPv6NDOptPrefixInfo(
 917             prefixlen=self.pg1.local_ip6_prefix_len,
 918             prefix=self.pg1.local_ip6,
 919             L=0,
 920             A=0)
 921
 922         self.pg0.ip6_ra_config(send_unicast=1)
 923         self.send_and_expect_ra(self.pg0, p,
 924                                 "RA with Prefix reverted to defaults",
 925                                 dst_ip=ll,
 926                                 opt=opt)
 927
 928         #
 929         # Remove the second prefix-info - expect no prefix-info i nthe adverts
 930         #
 931         self.pg0.ip6_ra_prefix(self.pg1.local_ip6n,
 932                                self.pg1.local_ip6_prefix_len,
 933                                is_no=1)
 934
 935         self.pg0.ip6_ra_config(send_unicast=1)
 936         self.send_and_expect_ra(self.pg0, p,
 937                                 "RA with Prefix reverted to defaults",
 938                                 dst_ip=ll)
 939
 940         #
 941         # Reset the periodic advertisements back to default values
 942         #
 943         self.pg0.ip6_ra_config(no=1, suppress=1, send_unicast=0)
 944
 945
 946 class TestICMPv6Echo(VppTestCase):
 947     """ ICMPv6 Echo Test Case """
 948
 949     def setUp(self):
 950         super(TestICMPv6Echo, self).setUp()
 951
 952         # create 1 pg interface
 953         self.create_pg_interfaces(range(1))
 954
 955         for i in self.pg_interfaces:
 956             i.admin_up()
 957             i.config_ip6()
 958             i.resolve_ndp()
 959
 960     def tearDown(self):
 961         super(TestICMPv6Echo, self).tearDown()
 962         for i in self.pg_interfaces:
 963             i.unconfig_ip6()
 964             i.ip6_disable()
 965             i.admin_down()
 966
 967     def test_icmpv6_echo(self):
 968         """ VPP replies to ICMPv6 Echo Request
 969
 970         Test scenario:
 971
 972             - Receive ICMPv6 Echo Request message on pg0 interface.
 973             - Check outgoing ICMPv6 Echo Reply message on pg0 interface.
 974         """
 975
 976         icmpv6_id = 0xb
 977         icmpv6_seq = 5
 978         icmpv6_data = '\x0a' * 18
 979         p_echo_request = (Ether(src=self.pg0.remote_mac,
 980                                 dst=self.pg0.local_mac) /
 981                           IPv6(src=self.pg0.remote_ip6,
 982                                dst=self.pg0.local_ip6) /
 983                           ICMPv6EchoRequest(
 984                               id=icmpv6_id,
 985                               seq=icmpv6_seq,
 986                               data=icmpv6_data))
 987
 988         self.pg0.add_stream(p_echo_request)
 989         self.pg_enable_capture(self.pg_interfaces)
 990         self.pg_start()
 991
 992         rx = self.pg0.get_capture(1)
 993         rx = rx[0]
 994         ether = rx[Ether]
 995         ipv6 = rx[IPv6]
 996         icmpv6 = rx[ICMPv6EchoReply]
 997
 998         self.assertEqual(ether.src, self.pg0.local_mac)
 999         self.assertEqual(ether.dst, self.pg0.remote_mac)
1000
1001         self.assertEqual(ipv6.src, self.pg0.local_ip6)
1002         self.assertEqual(ipv6.dst, self.pg0.remote_ip6)
1003
1004         self.assertEqual(
1005             icmp6types[icmpv6.type], "Echo Reply")
1006         self.assertEqual(icmpv6.id, icmpv6_id)
1007         self.assertEqual(icmpv6.seq, icmpv6_seq)
1008         self.assertEqual(icmpv6.data, icmpv6_data)
1009
1010
1011 class TestIPv6RD(TestIPv6ND):
1012     """ IPv6 Router Discovery Test Case """
1013
1014     @classmethod
1015     def setUpClass(cls):
1016         super(TestIPv6RD, cls).setUpClass()
1017
1018     def setUp(self):
1019         super(TestIPv6RD, self).setUp()
1020
1021         # create 2 pg interfaces
1022         self.create_pg_interfaces(range(2))
1023
1024         self.interfaces = list(self.pg_interfaces)
1025
1026         # setup all interfaces
1027         for i in self.interfaces:
1028             i.admin_up()
1029             i.config_ip6()
1030
1031     def tearDown(self):
1032         for i in self.interfaces:
1033             i.unconfig_ip6()
1034             i.admin_down()
1035         super(TestIPv6RD, self).tearDown()
1036
1037     def test_rd_send_router_solicitation(self):
1038         """ Verify router solicitation packets """
1039
1040         count = 2
1041         self.pg_enable_capture(self.pg_interfaces)
1042         self.pg_start()
1043         self.vapi.ip6nd_send_router_solicitation(self.pg1.sw_if_index,
1044                                                  mrc=count)
1045         rx_list = self.pg1.get_capture(count, timeout=3)
1046         self.assertEqual(len(rx_list), count)
1047         for packet in rx_list:
1048             self.assertEqual(packet.haslayer(IPv6), 1)
1049             self.assertEqual(packet[IPv6].haslayer(
1050                 ICMPv6ND_RS), 1)
1051             dst = ip6_normalize(packet[IPv6].dst)
1052             dst2 = ip6_normalize("ff02::2")
1053             self.assert_equal(dst, dst2)
1054             src = ip6_normalize(packet[IPv6].src)
1055             src2 = ip6_normalize(self.pg1.local_ip6_ll)
1056             self.assert_equal(src, src2)
1057             self.assertTrue(
1058                 bool(packet[ICMPv6ND_RS].haslayer(
1059                     ICMPv6NDOptSrcLLAddr)))
1060             self.assert_equal(
1061                 packet[ICMPv6NDOptSrcLLAddr].lladdr,
1062                 self.pg1.local_mac)
1063
1064     def verify_prefix_info(self, reported_prefix, prefix_option):
1065         prefix = socket.inet_pton(socket.AF_INET6,
1066                                   prefix_option.getfieldval("prefix"))
1067         self.assert_equal(reported_prefix.dst_address, prefix)
1068         self.assert_equal(reported_prefix.dst_address_length,
1069                           prefix_option.getfieldval("prefixlen"))
1070         L = prefix_option.getfieldval("L")
1071         A = prefix_option.getfieldval("A")
1072         option_flags = (L << 7) | (A << 6)
1073         self.assert_equal(reported_prefix.flags, option_flags)
1074         self.assert_equal(reported_prefix.valid_time,
1075                           prefix_option.getfieldval("validlifetime"))
1076         self.assert_equal(reported_prefix.preferred_time,
1077                           prefix_option.getfieldval("preferredlifetime"))
1078
1079     def test_rd_receive_router_advertisement(self):
1080         """ Verify events triggered by received RA packets """
1081
1082         self.vapi.want_ip6_ra_events()
1083
1084         prefix_info_1 = ICMPv6NDOptPrefixInfo(
1085             prefix="1::2",
1086             prefixlen=50,
1087             validlifetime=200,
1088             preferredlifetime=500,
1089             L=1,
1090             A=1,
1091         )
1092
1093         prefix_info_2 = ICMPv6NDOptPrefixInfo(
1094             prefix="7::4",
1095             prefixlen=20,
1096             validlifetime=70,
1097             preferredlifetime=1000,
1098             L=1,
1099             A=0,
1100         )
1101
1102         p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
1103              IPv6(dst=self.pg1.local_ip6_ll,
1104                   src=mk_ll_addr(self.pg1.remote_mac)) /
1105              ICMPv6ND_RA() /
1106              prefix_info_1 /
1107              prefix_info_2)
1108         self.pg1.add_stream([p])
1109         self.pg_start()
1110
1111         ev = self.vapi.wait_for_event(10, "ip6_ra_event")
1112
1113         self.assert_equal(ev.current_hop_limit, 0)
1114         self.assert_equal(ev.flags, 8)
1115         self.assert_equal(ev.router_lifetime_in_sec, 1800)
1116         self.assert_equal(ev.neighbor_reachable_time_in_msec, 0)
1117         self.assert_equal(
1118             ev.time_in_msec_between_retransmitted_neighbor_solicitations, 0)
1119
1120         self.assert_equal(ev.n_prefixes, 2)
1121
1122         self.verify_prefix_info(ev.prefixes[0], prefix_info_1)
1123         self.verify_prefix_info(ev.prefixes[1], prefix_info_2)
1124
1125
1126 class TestIPv6RDControlPlane(TestIPv6ND):
1127     """ IPv6 Router Discovery Control Plane Test Case """
1128
1129     @classmethod
1130     def setUpClass(cls):
1131         super(TestIPv6RDControlPlane, cls).setUpClass()
1132
1133     def setUp(self):
1134         super(TestIPv6RDControlPlane, self).setUp()
1135
1136         # create 1 pg interface
1137         self.create_pg_interfaces(range(1))
1138
1139         self.interfaces = list(self.pg_interfaces)
1140
1141         # setup all interfaces
1142         for i in self.interfaces:
1143             i.admin_up()
1144             i.config_ip6()
1145
1146     def tearDown(self):
1147         super(TestIPv6RDControlPlane, self).tearDown()
1148
1149     @staticmethod
1150     def create_ra_packet(pg, routerlifetime=None):
1151         src_ip = pg.remote_ip6_ll
1152         dst_ip = pg.local_ip6
1153         if routerlifetime is not None:
1154             ra = ICMPv6ND_RA(routerlifetime=routerlifetime)
1155         else:
1156             ra = ICMPv6ND_RA()
1157         p = (Ether(dst=pg.local_mac, src=pg.remote_mac) /
1158              IPv6(dst=dst_ip, src=src_ip) / ra)
1159         return p
1160
1161     @staticmethod
1162     def get_default_routes(fib):
1163         list = []
1164         for entry in fib:
1165             if entry.address_length == 0:
1166                 for path in entry.path:
1167                     if path.sw_if_index != 0xFFFFFFFF:
1168                         defaut_route = {}
1169                         defaut_route['sw_if_index'] = path.sw_if_index
1170                         defaut_route['next_hop'] = path.next_hop
1171                         list.append(defaut_route)
1172         return list
1173
1174     @staticmethod
1175     def get_interface_addresses(fib, pg):
1176         list = []
1177         for entry in fib:
1178             if entry.address_length == 128:
1179                 path = entry.path[0]
1180                 if path.sw_if_index == pg.sw_if_index:
1181                     list.append(entry.address)
1182         return list
1183
1184     def test_all(self):
1185         """ Test handling of SLAAC addresses and default routes """
1186
1187         fib = self.vapi.ip6_fib_dump()
1188         default_routes = self.get_default_routes(fib)
1189         initial_addresses = set(self.get_interface_addresses(fib, self.pg0))
1190         self.assertEqual(default_routes, [])
1191         router_address = self.pg0.remote_ip6n_ll
1192
1193         self.vapi.ip6_nd_address_autoconfig(self.pg0.sw_if_index, 1, 1)
1194
1195         self.sleep(0.1)
1196
1197         # send RA
1198         packet = (self.create_ra_packet(
1199             self.pg0) / ICMPv6NDOptPrefixInfo(
1200             prefix="1::",
1201             prefixlen=64,
1202             validlifetime=2,
1203             preferredlifetime=2,
1204             L=1,
1205             A=1,
1206         ) / ICMPv6NDOptPrefixInfo(
1207             prefix="7::",
1208             prefixlen=20,
1209             validlifetime=1500,
1210             preferredlifetime=1000,
1211             L=1,
1212             A=0,
1213         ))
1214         self.pg0.add_stream([packet])
1215         self.pg_start()
1216
1217         self.sleep(0.1)
1218
1219         fib = self.vapi.ip6_fib_dump()
1220
1221         # check FIB for new address
1222         addresses = set(self.get_interface_addresses(fib, self.pg0))
1223         new_addresses = addresses.difference(initial_addresses)
1224         self.assertEqual(len(new_addresses), 1)
1225         prefix = list(new_addresses)[0][:8] + '\0\0\0\0\0\0\0\0'
1226         self.assertEqual(inet_ntop(AF_INET6, prefix), '1::')
1227
1228         # check FIB for new default route
1229         default_routes = self.get_default_routes(fib)
1230         self.assertEqual(len(default_routes), 1)
1231         dr = default_routes[0]
1232         self.assertEqual(dr['sw_if_index'], self.pg0.sw_if_index)
1233         self.assertEqual(dr['next_hop'], router_address)
1234
1235         # send RA to delete default route
1236         packet = self.create_ra_packet(self.pg0, routerlifetime=0)
1237         self.pg0.add_stream([packet])
1238         self.pg_start()
1239
1240         self.sleep(0.1)
1241
1242         # check that default route is deleted
1243         fib = self.vapi.ip6_fib_dump()
1244         default_routes = self.get_default_routes(fib)
1245         self.assertEqual(len(default_routes), 0)
1246
1247         self.sleep(0.1)
1248
1249         # send RA
1250         packet = self.create_ra_packet(self.pg0)
1251         self.pg0.add_stream([packet])
1252         self.pg_start()
1253
1254         self.sleep(0.1)
1255
1256         # check FIB for new default route
1257         fib = self.vapi.ip6_fib_dump()
1258         default_routes = self.get_default_routes(fib)
1259         self.assertEqual(len(default_routes), 1)
1260         dr = default_routes[0]
1261         self.assertEqual(dr['sw_if_index'], self.pg0.sw_if_index)
1262         self.assertEqual(dr['next_hop'], router_address)
1263
1264         # send RA, updating router lifetime to 1s
1265         packet = self.create_ra_packet(self.pg0, 1)
1266         self.pg0.add_stream([packet])
1267         self.pg_start()
1268
1269         self.sleep(0.1)
1270
1271         # check that default route still exists
1272         fib = self.vapi.ip6_fib_dump()
1273         default_routes = self.get_default_routes(fib)
1274         self.assertEqual(len(default_routes), 1)
1275         dr = default_routes[0]
1276         self.assertEqual(dr['sw_if_index'], self.pg0.sw_if_index)
1277         self.assertEqual(dr['next_hop'], router_address)
1278
1279         self.sleep(1)
1280
1281         # check that default route is deleted
1282         fib = self.vapi.ip6_fib_dump()
1283         default_routes = self.get_default_routes(fib)
1284         self.assertEqual(len(default_routes), 0)
1285
1286         # check FIB still contains the SLAAC address
1287         addresses = set(self.get_interface_addresses(fib, self.pg0))
1288         new_addresses = addresses.difference(initial_addresses)
1289         self.assertEqual(len(new_addresses), 1)
1290         prefix = list(new_addresses)[0][:8] + '\0\0\0\0\0\0\0\0'
1291         self.assertEqual(inet_ntop(AF_INET6, prefix), '1::')
1292
1293         self.sleep(1)
1294
1295         # check that SLAAC address is deleted
1296         fib = self.vapi.ip6_fib_dump()
1297         addresses = set(self.get_interface_addresses(fib, self.pg0))
1298         new_addresses = addresses.difference(initial_addresses)
1299         self.assertEqual(len(new_addresses), 0)
1300
1301
1302 class IPv6NDProxyTest(TestIPv6ND):
1303     """ IPv6 ND ProxyTest Case """
1304
1305     def setUp(self):
1306         super(IPv6NDProxyTest, self).setUp()
1307
1308         # create 3 pg interfaces
1309         self.create_pg_interfaces(range(3))
1310
1311         # pg0 is the master interface, with the configured subnet
1312         self.pg0.admin_up()
1313         self.pg0.config_ip6()
1314         self.pg0.resolve_ndp()
1315
1316         self.pg1.ip6_enable()
1317         self.pg2.ip6_enable()
1318
1319     def tearDown(self):
1320         super(IPv6NDProxyTest, self).tearDown()
1321
1322     def test_nd_proxy(self):
1323         """ IPv6 Proxy ND """
1324
1325         #
1326         # Generate some hosts in the subnet that we are proxying
1327         #
1328         self.pg0.generate_remote_hosts(8)
1329
1330         nsma = in6_getnsma(inet_pton(AF_INET6, self.pg0.local_ip6))
1331         d = inet_ntop(AF_INET6, nsma)
1332
1333         #
1334         # Send an NS for one of those remote hosts on one of the proxy links
1335         # expect no response since it's from an address that is not
1336         # on the link that has the prefix configured
1337         #
1338         ns_pg1 = (Ether(dst=in6_getnsmac(nsma), src=self.pg1.remote_mac) /
1339                   IPv6(dst=d,
1340                        src=self.pg0._remote_hosts[2].ip6) /
1341                   ICMPv6ND_NS(tgt=self.pg0.local_ip6) /
1342                   ICMPv6NDOptSrcLLAddr(
1343                       lladdr=self.pg0._remote_hosts[2].mac))
1344
1345         self.send_and_assert_no_replies(self.pg1, ns_pg1, "Off link NS")
1346
1347         #
1348         # Add proxy support for the host
1349         #
1350         self.vapi.ip6_nd_proxy(
1351             inet_pton(AF_INET6, self.pg0._remote_hosts[2].ip6),
1352             self.pg1.sw_if_index)
1353
1354         #
1355         # try that NS again. this time we expect an NA back
1356         #
1357         self.send_and_expect_na(self.pg1, ns_pg1,
1358                                 "NS to proxy entry",
1359                                 dst_ip=self.pg0._remote_hosts[2].ip6,
1360                                 tgt_ip=self.pg0.local_ip6)
1361
1362         #
1363         # ... and that we have an entry in the ND cache
1364         #
1365         self.assertTrue(find_nbr(self,
1366                                  self.pg1.sw_if_index,
1367                                  self.pg0._remote_hosts[2].ip6,
1368                                  inet=AF_INET6))
1369
1370         #
1371         # ... and we can route traffic to it
1372         #
1373         t = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
1374              IPv6(dst=self.pg0._remote_hosts[2].ip6,
1375                   src=self.pg0.remote_ip6) /
1376              inet6.UDP(sport=10000, dport=20000) /
1377              Raw('\xa5' * 100))
1378
1379         self.pg0.add_stream(t)
1380         self.pg_enable_capture(self.pg_interfaces)
1381         self.pg_start()
1382         rx = self.pg1.get_capture(1)
1383         rx = rx[0]
1384
1385         self.assertEqual(rx[Ether].dst, self.pg0._remote_hosts[2].mac)
1386         self.assertEqual(rx[Ether].src, self.pg1.local_mac)
1387
1388         self.assertEqual(rx[IPv6].src,
1389                          t[IPv6].src)
1390         self.assertEqual(rx[IPv6].dst,
1391                          t[IPv6].dst)
1392
1393         #
1394         # Test we proxy for the host on the main interface
1395         #
1396         ns_pg0 = (Ether(dst=in6_getnsmac(nsma), src=self.pg0.remote_mac) /
1397                   IPv6(dst=d, src=self.pg0.remote_ip6) /
1398                   ICMPv6ND_NS(
1399                       tgt=self.pg0._remote_hosts[2].ip6) /
1400                   ICMPv6NDOptSrcLLAddr(
1401                       lladdr=self.pg0.remote_mac))
1402
1403         self.send_and_expect_na(self.pg0, ns_pg0,
1404                                 "NS to proxy entry on main",
1405                                 tgt_ip=self.pg0._remote_hosts[2].ip6,
1406                                 dst_ip=self.pg0.remote_ip6)
1407
1408         #
1409         # Setup and resolve proxy for another host on another interface
1410         #
1411         ns_pg2 = (Ether(dst=in6_getnsmac(nsma), src=self.pg2.remote_mac) /
1412                   IPv6(dst=d,
1413                        src=self.pg0._remote_hosts[3].ip6) /
1414                   ICMPv6ND_NS(tgt=self.pg0.local_ip6) /
1415                   ICMPv6NDOptSrcLLAddr(
1416                       lladdr=self.pg0._remote_hosts[2].mac))
1417
1418         self.vapi.ip6_nd_proxy(
1419             inet_pton(AF_INET6, self.pg0._remote_hosts[3].ip6),
1420             self.pg2.sw_if_index)
1421
1422         self.send_and_expect_na(self.pg2, ns_pg2,
1423                                 "NS to proxy entry other interface",
1424                                 dst_ip=self.pg0._remote_hosts[3].ip6,
1425                                 tgt_ip=self.pg0.local_ip6)
1426
1427         self.assertTrue(find_nbr(self,
1428                                  self.pg2.sw_if_index,
1429                                  self.pg0._remote_hosts[3].ip6,
1430                                  inet=AF_INET6))
1431
1432         #
1433         # hosts can communicate. pg2->pg1
1434         #
1435         t2 = (Ether(dst=self.pg2.local_mac,
1436                     src=self.pg0.remote_hosts[3].mac) /
1437               IPv6(dst=self.pg0._remote_hosts[2].ip6,
1438                    src=self.pg0._remote_hosts[3].ip6) /
1439               inet6.UDP(sport=10000, dport=20000) /
1440               Raw('\xa5' * 100))
1441
1442         self.pg2.add_stream(t2)
1443         self.pg_enable_capture(self.pg_interfaces)
1444         self.pg_start()
1445         rx = self.pg1.get_capture(1)
1446         rx = rx[0]
1447
1448         self.assertEqual(rx[Ether].dst, self.pg0._remote_hosts[2].mac)
1449         self.assertEqual(rx[Ether].src, self.pg1.local_mac)
1450
1451         self.assertEqual(rx[IPv6].src,
1452                          t2[IPv6].src)
1453         self.assertEqual(rx[IPv6].dst,
1454                          t2[IPv6].dst)
1455
1456         #
1457         # remove the proxy configs
1458         #
1459         self.vapi.ip6_nd_proxy(
1460             inet_pton(AF_INET6, self.pg0._remote_hosts[2].ip6),
1461             self.pg1.sw_if_index,
1462             is_del=1)
1463         self.vapi.ip6_nd_proxy(
1464             inet_pton(AF_INET6, self.pg0._remote_hosts[3].ip6),
1465             self.pg2.sw_if_index,
1466             is_del=1)
1467
1468         self.assertFalse(find_nbr(self,
1469                                   self.pg2.sw_if_index,
1470                                   self.pg0._remote_hosts[3].ip6,
1471                                   inet=AF_INET6))
1472         self.assertFalse(find_nbr(self,
1473                                   self.pg1.sw_if_index,
1474                                   self.pg0._remote_hosts[2].ip6,
1475                                   inet=AF_INET6))
1476
1477         #
1478         # no longer proxy-ing...
1479         #
1480         self.send_and_assert_no_replies(self.pg0, ns_pg0, "Proxy unconfigured")
1481         self.send_and_assert_no_replies(self.pg1, ns_pg1, "Proxy unconfigured")
1482         self.send_and_assert_no_replies(self.pg2, ns_pg2, "Proxy unconfigured")
1483
1484         #
1485         # no longer forwarding. traffic generates NS out of the glean/main
1486         # interface
1487         #
1488         self.pg2.add_stream(t2)
1489         self.pg_enable_capture(self.pg_interfaces)
1490         self.pg_start()
1491
1492         rx = self.pg0.get_capture(1)
1493
1494         self.assertTrue(rx[0].haslayer(ICMPv6ND_NS))
1495
1496
1497 class TestIPNull(VppTestCase):
1498     """ IPv6 routes via NULL """
1499
1500     def setUp(self):
1501         super(TestIPNull, self).setUp()
1502
1503         # create 2 pg interfaces
1504         self.create_pg_interfaces(range(1))
1505
1506         for i in self.pg_interfaces:
1507             i.admin_up()
1508             i.config_ip6()
1509             i.resolve_ndp()
1510
1511     def tearDown(self):
1512         super(TestIPNull, self).tearDown()
1513         for i in self.pg_interfaces:
1514             i.unconfig_ip6()
1515             i.admin_down()
1516
1517     def test_ip_null(self):
1518         """ IP NULL route """
1519
1520         p = (Ether(src=self.pg0.remote_mac,
1521                    dst=self.pg0.local_mac) /
1522              IPv6(src=self.pg0.remote_ip6, dst="2001::1") /
1523              inet6.UDP(sport=1234, dport=1234) /
1524              Raw('\xa5' * 100))
1525
1526         #
1527         # A route via IP NULL that will reply with ICMP unreachables
1528         #
1529         ip_unreach = VppIpRoute(self, "2001::", 64, [], is_unreach=1, is_ip6=1)
1530         ip_unreach.add_vpp_config()
1531
1532         self.pg0.add_stream(p)
1533         self.pg_enable_capture(self.pg_interfaces)
1534         self.pg_start()
1535
1536         rx = self.pg0.get_capture(1)
1537         rx = rx[0]
1538         icmp = rx[ICMPv6DestUnreach]
1539
1540         # 0 = "No route to destination"
1541         self.assertEqual(icmp.code, 0)
1542
1543         # ICMP is rate limited. pause a bit
1544         self.sleep(1)
1545
1546         #
1547         # A route via IP NULL that will reply with ICMP prohibited
1548         #
1549         ip_prohibit = VppIpRoute(self, "2001::1", 128, [],
1550                                  is_prohibit=1, is_ip6=1)
1551         ip_prohibit.add_vpp_config()
1552
1553         self.pg0.add_stream(p)
1554         self.pg_enable_capture(self.pg_interfaces)
1555         self.pg_start()
1556
1557         rx = self.pg0.get_capture(1)
1558         rx = rx[0]
1559         icmp = rx[ICMPv6DestUnreach]
1560
1561         # 1 = "Communication with destination administratively prohibited"
1562         self.assertEqual(icmp.code, 1)
1563
1564
1565 class TestIPDisabled(VppTestCase):
1566     """ IPv6 disabled """
1567
1568     def setUp(self):
1569         super(TestIPDisabled, self).setUp()
1570
1571         # create 2 pg interfaces
1572         self.create_pg_interfaces(range(2))
1573
1574         # PG0 is IP enalbed
1575         self.pg0.admin_up()
1576         self.pg0.config_ip6()
1577         self.pg0.resolve_ndp()
1578
1579         # PG 1 is not IP enabled
1580         self.pg1.admin_up()
1581
1582     def tearDown(self):
1583         super(TestIPDisabled, self).tearDown()
1584         for i in self.pg_interfaces:
1585             i.unconfig_ip4()
1586             i.admin_down()
1587
1588     def test_ip_disabled(self):
1589         """ IP Disabled """
1590
1591         #
1592         # An (S,G).
1593         # one accepting interface, pg0, 2 forwarding interfaces
1594         #
1595         route_ff_01 = VppIpMRoute(
1596             self,
1597             "::",
1598             "ffef::1", 128,
1599             MRouteEntryFlags.MFIB_ENTRY_FLAG_NONE,
1600             [VppMRoutePath(self.pg1.sw_if_index,
1601                            MRouteItfFlags.MFIB_ITF_FLAG_ACCEPT),
1602              VppMRoutePath(self.pg0.sw_if_index,
1603                            MRouteItfFlags.MFIB_ITF_FLAG_FORWARD)],
1604             is_ip6=1)
1605         route_ff_01.add_vpp_config()
1606
1607         pu = (Ether(src=self.pg1.remote_mac,
1608                     dst=self.pg1.local_mac) /
1609               IPv6(src="2001::1", dst=self.pg0.remote_ip6) /
1610               inet6.UDP(sport=1234, dport=1234) /
1611               Raw('\xa5' * 100))
1612         pm = (Ether(src=self.pg1.remote_mac,
1613                     dst=self.pg1.local_mac) /
1614               IPv6(src="2001::1", dst="ffef::1") /
1615               inet6.UDP(sport=1234, dport=1234) /
1616               Raw('\xa5' * 100))
1617
1618         #
1619         # PG1 does not forward IP traffic
1620         #
1621         self.send_and_assert_no_replies(self.pg1, pu, "IPv6 disabled")
1622         self.send_and_assert_no_replies(self.pg1, pm, "IPv6 disabled")
1623
1624         #
1625         # IP enable PG1
1626         #
1627         self.pg1.config_ip6()
1628
1629         #
1630         # Now we get packets through
1631         #
1632         self.pg1.add_stream(pu)
1633         self.pg_enable_capture(self.pg_interfaces)
1634         self.pg_start()
1635         rx = self.pg0.get_capture(1)
1636
1637         self.pg1.add_stream(pm)
1638         self.pg_enable_capture(self.pg_interfaces)
1639         self.pg_start()
1640         rx = self.pg0.get_capture(1)
1641
1642         #
1643         # Disable PG1
1644         #
1645         self.pg1.unconfig_ip6()
1646
1647         #
1648         # PG1 does not forward IP traffic
1649         #
1650         self.send_and_assert_no_replies(self.pg1, pu, "IPv6 disabled")
1651         self.send_and_assert_no_replies(self.pg1, pm, "IPv6 disabled")
1652
1653
1654 class TestIP6LoadBalance(VppTestCase):
1655     """ IPv6 Load-Balancing """
1656
1657     def setUp(self):
1658         super(TestIP6LoadBalance, self).setUp()
1659
1660         self.create_pg_interfaces(range(5))
1661
1662         mpls_tbl = VppMplsTable(self, 0)
1663         mpls_tbl.add_vpp_config()
1664
1665         for i in self.pg_interfaces:
1666             i.admin_up()
1667             i.config_ip6()
1668             i.resolve_ndp()
1669             i.enable_mpls()
1670
1671     def tearDown(self):
1672         for i in self.pg_interfaces:
1673             i.unconfig_ip6()
1674             i.admin_down()
1675             i.disable_mpls()
1676         super(TestIP6LoadBalance, self).tearDown()
1677
1678     def send_and_expect_load_balancing(self, input, pkts, outputs):
1679         self.vapi.cli("clear trace")
1680         input.add_stream(pkts)
1681         self.pg_enable_capture(self.pg_interfaces)
1682         self.pg_start()
1683         for oo in outputs:
1684             rx = oo._get_capture(1)
1685             self.assertNotEqual(0, len(rx))
1686
1687     def send_and_expect_one_itf(self, input, pkts, itf):
1688         self.vapi.cli("clear trace")
1689         input.add_stream(pkts)
1690         self.pg_enable_capture(self.pg_interfaces)
1691         self.pg_start()
1692         rx = itf.get_capture(len(pkts))
1693
1694     def test_ip6_load_balance(self):
1695         """ IPv6 Load-Balancing """
1696
1697         #
1698         # An array of packets that differ only in the destination port
1699         #  - IP only
1700         #  - MPLS EOS
1701         #  - MPLS non-EOS
1702         #  - MPLS non-EOS with an entropy label
1703         #
1704         port_ip_pkts = []
1705         port_mpls_pkts = []
1706         port_mpls_neos_pkts = []
1707         port_ent_pkts = []
1708
1709         #
1710         # An array of packets that differ only in the source address
1711         #
1712         src_ip_pkts = []
1713         src_mpls_pkts = []
1714
1715         for ii in range(65):
1716             port_ip_hdr = (
1717                 IPv6(dst="3000::1", src="3000:1::1") /
1718                 inet6.UDP(sport=1234, dport=1234 + ii) /
1719                 Raw('\xa5' * 100))
1720             port_ip_pkts.append((Ether(src=self.pg0.remote_mac,
1721                                        dst=self.pg0.local_mac) /
1722                                  port_ip_hdr))
1723             port_mpls_pkts.append((Ether(src=self.pg0.remote_mac,
1724                                          dst=self.pg0.local_mac) /
1725                                    MPLS(label=66, ttl=2) /
1726                                    port_ip_hdr))
1727             port_mpls_neos_pkts.append((Ether(src=self.pg0.remote_mac,
1728                                               dst=self.pg0.local_mac) /
1729                                         MPLS(label=67, ttl=2) /
1730                                         MPLS(label=77, ttl=2) /
1731                                         port_ip_hdr))
1732             port_ent_pkts.append((Ether(src=self.pg0.remote_mac,
1733                                         dst=self.pg0.local_mac) /
1734                                   MPLS(label=67, ttl=2) /
1735                                   MPLS(label=14, ttl=2) /
1736                                   MPLS(label=999, ttl=2) /
1737                                   port_ip_hdr))
1738             src_ip_hdr = (
1739                 IPv6(dst="3000::1", src="3000:1::%d" % ii) /
1740                 inet6.UDP(sport=1234, dport=1234) /
1741                 Raw('\xa5' * 100))
1742             src_ip_pkts.append((Ether(src=self.pg0.remote_mac,
1743                                       dst=self.pg0.local_mac) /
1744                                 src_ip_hdr))
1745             src_mpls_pkts.append((Ether(src=self.pg0.remote_mac,
1746                                         dst=self.pg0.local_mac) /
1747                                   MPLS(label=66, ttl=2) /
1748                                   src_ip_hdr))
1749
1750         #
1751         # A route for the IP pacekts
1752         #
1753         route_3000_1 = VppIpRoute(self, "3000::1", 128,
1754                                   [VppRoutePath(self.pg1.remote_ip6,
1755                                                 self.pg1.sw_if_index,
1756                                                 proto=DpoProto.DPO_PROTO_IP6),
1757                                    VppRoutePath(self.pg2.remote_ip6,
1758                                                 self.pg2.sw_if_index,
1759                                                 proto=DpoProto.DPO_PROTO_IP6)],
1760                                   is_ip6=1)
1761         route_3000_1.add_vpp_config()
1762
1763         #
1764         # a local-label for the EOS packets
1765         #
1766         binding = VppMplsIpBind(self, 66, "3000::1", 128, is_ip6=1)
1767         binding.add_vpp_config()
1768
1769         #
1770         # An MPLS route for the non-EOS packets
1771         #
1772         route_67 = VppMplsRoute(self, 67, 0,
1773                                 [VppRoutePath(self.pg1.remote_ip6,
1774                                               self.pg1.sw_if_index,
1775                                               labels=[67],
1776                                               proto=DpoProto.DPO_PROTO_IP6),
1777                                  VppRoutePath(self.pg2.remote_ip6,
1778                                               self.pg2.sw_if_index,
1779                                               labels=[67],
1780                                               proto=DpoProto.DPO_PROTO_IP6)])
1781         route_67.add_vpp_config()
1782
1783         #
1784         # inject the packet on pg0 - expect load-balancing across the 2 paths
1785         #  - since the default hash config is to use IP src,dst and port
1786         #    src,dst
1787         # We are not going to ensure equal amounts of packets across each link,
1788         # since the hash algorithm is statistical and therefore this can never
1789         # be guaranteed. But wuth 64 different packets we do expect some
1790         # balancing. So instead just ensure there is traffic on each link.
1791         #
1792         self.send_and_expect_load_balancing(self.pg0, port_ip_pkts,
1793                                             [self.pg1, self.pg2])
1794         self.send_and_expect_load_balancing(self.pg0, src_ip_pkts,
1795                                             [self.pg1, self.pg2])
1796         self.send_and_expect_load_balancing(self.pg0, port_mpls_pkts,
1797                                             [self.pg1, self.pg2])
1798         self.send_and_expect_load_balancing(self.pg0, src_mpls_pkts,
1799                                             [self.pg1, self.pg2])
1800         self.send_and_expect_load_balancing(self.pg0, port_mpls_neos_pkts,
1801                                             [self.pg1, self.pg2])
1802
1803         #
1804         # The packets with Entropy label in should not load-balance,
1805         # since the Entorpy value is fixed.
1806         #
1807         self.send_and_expect_one_itf(self.pg0, port_ent_pkts, self.pg1)
1808
1809         #
1810         # change the flow hash config so it's only IP src,dst
1811         #  - now only the stream with differing source address will
1812         #    load-balance
1813         #
1814         self.vapi.set_ip_flow_hash(0, is_ip6=1, src=1, dst=1, sport=0, dport=0)
1815
1816         self.send_and_expect_load_balancing(self.pg0, src_ip_pkts,
1817                                             [self.pg1, self.pg2])
1818         self.send_and_expect_load_balancing(self.pg0, src_mpls_pkts,
1819                                             [self.pg1, self.pg2])
1820         self.send_and_expect_one_itf(self.pg0, port_ip_pkts, self.pg2)
1821
1822         #
1823         # change the flow hash config back to defaults
1824         #
1825         self.vapi.set_ip_flow_hash(0, is_ip6=1, src=1, dst=1, sport=1, dport=1)
1826
1827         #
1828         # Recursive prefixes
1829         #  - testing that 2 stages of load-balancing occurs and there is no
1830         #    polarisation (i.e. only 2 of 4 paths are used)
1831         #
1832         port_pkts = []
1833         src_pkts = []
1834
1835         for ii in range(257):
1836             port_pkts.append((Ether(src=self.pg0.remote_mac,
1837                                     dst=self.pg0.local_mac) /
1838                               IPv6(dst="4000::1",
1839                                    src="4000:1::1") /
1840                               inet6.UDP(sport=1234,
1841                                         dport=1234 + ii) /
1842                               Raw('\xa5' * 100)))
1843             src_pkts.append((Ether(src=self.pg0.remote_mac,
1844                                    dst=self.pg0.local_mac) /
1845                              IPv6(dst="4000::1",
1846                                   src="4000:1::%d" % ii) /
1847                              inet6.UDP(sport=1234, dport=1234) /
1848                              Raw('\xa5' * 100)))
1849
1850         route_3000_2 = VppIpRoute(self, "3000::2", 128,
1851                                   [VppRoutePath(self.pg3.remote_ip6,
1852                                                 self.pg3.sw_if_index,
1853                                                 proto=DpoProto.DPO_PROTO_IP6),
1854                                    VppRoutePath(self.pg4.remote_ip6,
1855                                                 self.pg4.sw_if_index,
1856                                                 proto=DpoProto.DPO_PROTO_IP6)],
1857                                   is_ip6=1)
1858         route_3000_2.add_vpp_config()
1859
1860         route_4000_1 = VppIpRoute(self, "4000::1", 128,
1861                                   [VppRoutePath("3000::1",
1862                                                 0xffffffff,
1863                                                 proto=DpoProto.DPO_PROTO_IP6),
1864                                    VppRoutePath("3000::2",
1865                                                 0xffffffff,
1866                                                 proto=DpoProto.DPO_PROTO_IP6)],
1867                                   is_ip6=1)
1868         route_4000_1.add_vpp_config()
1869
1870         #
1871         # inject the packet on pg0 - expect load-balancing across all 4 paths
1872         #
1873         self.vapi.cli("clear trace")
1874         self.send_and_expect_load_balancing(self.pg0, port_pkts,
1875                                             [self.pg1, self.pg2,
1876                                              self.pg3, self.pg4])
1877         self.send_and_expect_load_balancing(self.pg0, src_pkts,
1878                                             [self.pg1, self.pg2,
1879                                              self.pg3, self.pg4])
1880
1881         #
1882         # Recursive prefixes
1883         #  - testing that 2 stages of load-balancing no choices
1884         #
1885         port_pkts = []
1886
1887         for ii in range(257):
1888             port_pkts.append((Ether(src=self.pg0.remote_mac,
1889                                     dst=self.pg0.local_mac) /
1890                               IPv6(dst="6000::1",
1891                                    src="6000:1::1") /
1892                               inet6.UDP(sport=1234,
1893                                         dport=1234 + ii) /
1894                               Raw('\xa5' * 100)))
1895
1896         route_5000_2 = VppIpRoute(self, "5000::2", 128,
1897                                   [VppRoutePath(self.pg3.remote_ip6,
1898                                                 self.pg3.sw_if_index,
1899                                                 proto=DpoProto.DPO_PROTO_IP6)],
1900                                   is_ip6=1)
1901         route_5000_2.add_vpp_config()
1902
1903         route_6000_1 = VppIpRoute(self, "6000::1", 128,
1904                                   [VppRoutePath("5000::2",
1905                                                 0xffffffff,
1906                                                 proto=DpoProto.DPO_PROTO_IP6)],
1907                                   is_ip6=1)
1908         route_6000_1.add_vpp_config()
1909
1910         #
1911         # inject the packet on pg0 - expect load-balancing across all 4 paths
1912         #
1913         self.vapi.cli("clear trace")
1914         self.send_and_expect_one_itf(self.pg0, port_pkts, self.pg3)
1915
1916
1917 class TestIP6Punt(VppTestCase):
1918     """ IPv6 Punt Police/Redirect """
1919
1920     def setUp(self):
1921         super(TestIP6Punt, self).setUp()
1922
1923         self.create_pg_interfaces(range(4))
1924
1925         for i in self.pg_interfaces:
1926             i.admin_up()
1927             i.config_ip6()
1928             i.resolve_ndp()
1929
1930     def tearDown(self):
1931         super(TestIP6Punt, self).tearDown()
1932         for i in self.pg_interfaces:
1933             i.unconfig_ip6()
1934             i.admin_down()
1935
1936     def test_ip_punt(self):
1937         """ IP6 punt police and redirect """
1938
1939         p = (Ether(src=self.pg0.remote_mac,
1940                    dst=self.pg0.local_mac) /
1941              IPv6(src=self.pg0.remote_ip6,
1942                   dst=self.pg0.local_ip6) /
1943              inet6.TCP(sport=1234, dport=1234) /
1944              Raw('\xa5' * 100))
1945
1946         pkts = p * 1025
1947
1948         #
1949         # Configure a punt redirect via pg1.
1950         #
1951         nh_addr = VppIpAddress(self.pg1.remote_ip6).encode()
1952         self.vapi.ip_punt_redirect(self.pg0.sw_if_index,
1953                                    self.pg1.sw_if_index,
1954                                    nh_addr)
1955
1956         self.send_and_expect(self.pg0, pkts, self.pg1)
1957
1958         #
1959         # add a policer
1960         #
1961         policer = self.vapi.policer_add_del("ip6-punt", 400, 0, 10, 0,
1962                                             rate_type=1)
1963         self.vapi.ip_punt_police(policer.policer_index, is_ip6=1)
1964
1965         self.vapi.cli("clear trace")
1966         self.pg0.add_stream(pkts)
1967         self.pg_enable_capture(self.pg_interfaces)
1968         self.pg_start()
1969
1970         #
1971         # the number of packet recieved should be greater than 0,
1972         # but not equal to the number sent, since some were policed
1973         #
1974         rx = self.pg1._get_capture(1)
1975         self.assertGreater(len(rx), 0)
1976         self.assertLess(len(rx), len(pkts))
1977
1978         #
1979         # remove the poilcer. back to full rx
1980         #
1981         self.vapi.ip_punt_police(policer.policer_index, is_add=0, is_ip6=1)
1982         self.vapi.policer_add_del("ip6-punt", 400, 0, 10, 0,
1983                                   rate_type=1, is_add=0)
1984         self.send_and_expect(self.pg0, pkts, self.pg1)
1985
1986         #
1987         # remove the redirect. expect full drop.
1988         #
1989         self.vapi.ip_punt_redirect(self.pg0.sw_if_index,
1990                                    self.pg1.sw_if_index,
1991                                    nh_addr,
1992                                    is_add=0)
1993         self.send_and_assert_no_replies(self.pg0, pkts,
1994                                         "IP no punt config")
1995
1996         #
1997         # Add a redirect that is not input port selective
1998         #
1999         self.vapi.ip_punt_redirect(0xffffffff,
2000                                    self.pg1.sw_if_index,
2001                                    nh_addr)
2002         self.send_and_expect(self.pg0, pkts, self.pg1)
2003
2004         self.vapi.ip_punt_redirect(0xffffffff,
2005                                    self.pg1.sw_if_index,
2006                                    nh_addr,
2007                                    is_add=0)
2008
2009     def test_ip_punt_dump(self):
2010         """ IP6 punt redirect dump"""
2011
2012         #
2013         # Configure a punt redirects
2014         #
2015         nh_addr = VppIpAddress(self.pg3.remote_ip6).encode()
2016         self.vapi.ip_punt_redirect(self.pg0.sw_if_index,
2017                                    self.pg3.sw_if_index,
2018                                    nh_addr)
2019         self.vapi.ip_punt_redirect(self.pg1.sw_if_index,
2020                                    self.pg3.sw_if_index,
2021                                    nh_addr)
2022         self.vapi.ip_punt_redirect(self.pg2.sw_if_index,
2023                                    self.pg3.sw_if_index,
2024                                    VppIpAddress('0::0').encode())
2025
2026         #
2027         # Dump pg0 punt redirects
2028         #
2029         punts = self.vapi.ip_punt_redirect_dump(self.pg0.sw_if_index,
2030                                                 is_ipv6=1)
2031         for p in punts:
2032             self.assertEqual(p.punt.rx_sw_if_index, self.pg0.sw_if_index)
2033
2034         #
2035         # Dump punt redirects for all interfaces
2036         #
2037         punts = self.vapi.ip_punt_redirect_dump(0xffffffff, is_ipv6=1)
2038         self.assertEqual(len(punts), 3)
2039         for p in punts:
2040             self.assertEqual(p.punt.tx_sw_if_index, self.pg3.sw_if_index)
2041         self.assertNotEqual(punts[1].punt.nh.un.ip6, self.pg3.remote_ip6)
2042         self.assertEqual(punts[2].punt.nh.un.ip6.address, '\x00'*16)
2043
2044
2045 class TestIPDeag(VppTestCase):
2046     """ IPv6 Deaggregate Routes """
2047
2048     def setUp(self):
2049         super(TestIPDeag, self).setUp()
2050
2051         self.create_pg_interfaces(range(3))
2052
2053         for i in self.pg_interfaces:
2054             i.admin_up()
2055             i.config_ip6()
2056             i.resolve_ndp()
2057
2058     def tearDown(self):
2059         super(TestIPDeag, self).tearDown()
2060         for i in self.pg_interfaces:
2061             i.unconfig_ip6()
2062             i.admin_down()
2063
2064     def test_ip_deag(self):
2065         """ IP Deag Routes """
2066
2067         #
2068         # Create a table to be used for:
2069         #  1 - another destination address lookup
2070         #  2 - a source address lookup
2071         #
2072         table_dst = VppIpTable(self, 1, is_ip6=1)
2073         table_src = VppIpTable(self, 2, is_ip6=1)
2074         table_dst.add_vpp_config()
2075         table_src.add_vpp_config()
2076
2077         #
2078         # Add a route in the default table to point to a deag/
2079         # second lookup in each of these tables
2080         #
2081         route_to_dst = VppIpRoute(self, "1::1", 128,
2082                                   [VppRoutePath("::",
2083                                                 0xffffffff,
2084                                                 nh_table_id=1,
2085                                                 proto=DpoProto.DPO_PROTO_IP6)],
2086                                   is_ip6=1)
2087         route_to_src = VppIpRoute(self, "1::2", 128,
2088                                   [VppRoutePath("::",
2089                                                 0xffffffff,
2090                                                 nh_table_id=2,
2091                                                 is_source_lookup=1,
2092                                                 proto=DpoProto.DPO_PROTO_IP6)],
2093                                   is_ip6=1)
2094         route_to_dst.add_vpp_config()
2095         route_to_src.add_vpp_config()
2096
2097         #
2098         # packets to these destination are dropped, since they'll
2099         # hit the respective default routes in the second table
2100         #
2101         p_dst = (Ether(src=self.pg0.remote_mac,
2102                        dst=self.pg0.local_mac) /
2103                  IPv6(src="5::5", dst="1::1") /
2104                  inet6.TCP(sport=1234, dport=1234) /
2105                  Raw('\xa5' * 100))
2106         p_src = (Ether(src=self.pg0.remote_mac,
2107                        dst=self.pg0.local_mac) /
2108                  IPv6(src="2::2", dst="1::2") /
2109                  inet6.TCP(sport=1234, dport=1234) /
2110                  Raw('\xa5' * 100))
2111         pkts_dst = p_dst * 257
2112         pkts_src = p_src * 257
2113
2114         self.send_and_assert_no_replies(self.pg0, pkts_dst,
2115                                         "IP in dst table")
2116         self.send_and_assert_no_replies(self.pg0, pkts_src,
2117                                         "IP in src table")
2118
2119         #
2120         # add a route in the dst table to forward via pg1
2121         #
2122         route_in_dst = VppIpRoute(self, "1::1", 128,
2123                                   [VppRoutePath(self.pg1.remote_ip6,
2124                                                 self.pg1.sw_if_index,
2125                                                 proto=DpoProto.DPO_PROTO_IP6)],
2126                                   is_ip6=1,
2127                                   table_id=1)
2128         route_in_dst.add_vpp_config()
2129
2130         self.send_and_expect(self.pg0, pkts_dst, self.pg1)
2131
2132         #
2133         # add a route in the src table to forward via pg2
2134         #
2135         route_in_src = VppIpRoute(self, "2::2", 128,
2136                                   [VppRoutePath(self.pg2.remote_ip6,
2137                                                 self.pg2.sw_if_index,
2138                                                 proto=DpoProto.DPO_PROTO_IP6)],
2139                                   is_ip6=1,
2140                                   table_id=2)
2141         route_in_src.add_vpp_config()
2142         self.send_and_expect(self.pg0, pkts_src, self.pg2)
2143
2144         #
2145         # loop in the lookup DP
2146         #
2147         route_loop = VppIpRoute(self, "3::3", 128,
2148                                 [VppRoutePath("::",
2149                                               0xffffffff,
2150                                               proto=DpoProto.DPO_PROTO_IP6)],
2151                                 is_ip6=1)
2152         route_loop.add_vpp_config()
2153
2154         p_l = (Ether(src=self.pg0.remote_mac,
2155                      dst=self.pg0.local_mac) /
2156                IPv6(src="3::4", dst="3::3") /
2157                inet6.TCP(sport=1234, dport=1234) /
2158                Raw('\xa5' * 100))
2159
2160         self.send_and_assert_no_replies(self.pg0, p_l * 257,
2161                                         "IP lookup loop")
2162
2163
2164 class TestIP6Input(VppTestCase):
2165     """ IPv6 Input Exceptions """
2166
2167     def setUp(self):
2168         super(TestIP6Input, self).setUp()
2169
2170         self.create_pg_interfaces(range(2))
2171
2172         for i in self.pg_interfaces:
2173             i.admin_up()
2174             i.config_ip6()
2175             i.resolve_ndp()
2176
2177     def tearDown(self):
2178         super(TestIP6Input, self).tearDown()
2179         for i in self.pg_interfaces:
2180             i.unconfig_ip6()
2181             i.admin_down()
2182
2183     def test_ip_input(self):
2184         """ IP6 Input Exceptions """
2185
2186         #
2187         # bad version - this is dropped
2188         #
2189         p_version = (Ether(src=self.pg0.remote_mac,
2190                            dst=self.pg0.local_mac) /
2191                      IPv6(src=self.pg0.remote_ip6,
2192                           dst=self.pg1.remote_ip6,
2193                           version=3) /
2194                      inet6.UDP(sport=1234, dport=1234) /
2195                      Raw('\xa5' * 100))
2196
2197         self.send_and_assert_no_replies(self.pg0, p_version * 65,
2198                                         "funky version")
2199
2200         #
2201         # hop limit - IMCP replies
2202         #
2203         p_version = (Ether(src=self.pg0.remote_mac,
2204                            dst=self.pg0.local_mac) /
2205                      IPv6(src=self.pg0.remote_ip6,
2206                           dst=self.pg1.remote_ip6,
2207                           hlim=1) /
2208                      inet6.UDP(sport=1234, dport=1234) /
2209                      Raw('\xa5' * 100))
2210
2211         rx = self.send_and_expect(self.pg0, p_version * 65, self.pg0)
2212         rx = rx[0]
2213         icmp = rx[ICMPv6TimeExceeded]
2214         self.assertEqual(icmp.type, 3)
2215         # 0: "hop limit exceeded in transit",
2216         self.assertEqual(icmp.code, 0)
2217
2218
2219 if __name__ == '__main__':
2220     unittest.main(testRunner=VppTestRunner)