test/test_ip6.py

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