test/test_ip6.py

   1 #!/usr/bin/env python3
   2
   3 import socket
   4 from socket import inet_pton, inet_ntop
   5 import unittest
   6
   7 from parameterized import parameterized
   8 import scapy.compat
   9 import scapy.layers.inet6 as inet6
  10 from scapy.contrib.mpls import MPLS
  11 from scapy.layers.inet6 import IPv6, ICMPv6ND_NS, ICMPv6ND_RS, \
  12     ICMPv6ND_RA, ICMPv6NDOptMTU, ICMPv6NDOptSrcLLAddr, ICMPv6NDOptPrefixInfo, \
  13     ICMPv6ND_NA, ICMPv6NDOptDstLLAddr, ICMPv6DestUnreach, icmp6types, \
  14     ICMPv6TimeExceeded, ICMPv6EchoRequest, ICMPv6EchoReply, \
  15     IPv6ExtHdrHopByHop, ICMPv6MLReport2, ICMPv6MLDMultAddrRec
  16 from scapy.layers.l2 import Ether, Dot1Q
  17 from scapy.packet import Raw
  18 from scapy.utils6 import in6_getnsma, in6_getnsmac, in6_ptop, in6_islladdr, \
  19     in6_mactoifaceid
  20 from six import moves
  21
  22 from framework import VppTestCase, VppTestRunner
  23 from util import ppp, ip6_normalize, mk_ll_addr
  24 from vpp_ip import DpoProto
  25 from vpp_ip_route import VppIpRoute, VppRoutePath, find_route, VppIpMRoute, \
  26     VppMRoutePath, MRouteItfFlags, MRouteEntryFlags, VppMplsIpBind, \
  27     VppMplsRoute, VppMplsTable, VppIpTable, FibPathType, FibPathProto, \
  28     VppIpInterfaceAddress, find_route_in_dump, find_mroute_in_dump, \
  29     VppIp6LinkLocalAddress
  30 from vpp_neighbor import find_nbr, VppNeighbor
  31 from vpp_pg_interface import is_ipv6_misc
  32 from vpp_sub_interface import VppSubInterface, VppDot1QSubint
  33 from vpp_policer import VppPolicer
  34 from ipaddress import IPv6Network, IPv6Address
  35
  36 AF_INET6 = socket.AF_INET6
  37
  38 try:
  39     text_type = unicode
  40 except NameError:
  41     text_type = str
  42
  43 NUM_PKTS = 67
  44
  45
  46 class TestIPv6ND(VppTestCase):
  47     def validate_ra(self, intf, rx, dst_ip=None):
  48         if not dst_ip:
  49             dst_ip = intf.remote_ip6
  50
  51         # unicasted packets must come to the unicast mac
  52         self.assertEqual(rx[Ether].dst, intf.remote_mac)
  53
  54         # and from the router's MAC
  55         self.assertEqual(rx[Ether].src, intf.local_mac)
  56
  57         # the rx'd RA should be addressed to the sender's source
  58         self.assertTrue(rx.haslayer(ICMPv6ND_RA))
  59         self.assertEqual(in6_ptop(rx[IPv6].dst),
  60                          in6_ptop(dst_ip))
  61
  62         # and come from the router's link local
  63         self.assertTrue(in6_islladdr(rx[IPv6].src))
  64         self.assertEqual(in6_ptop(rx[IPv6].src),
  65                          in6_ptop(mk_ll_addr(intf.local_mac)))
  66
  67     def validate_na(self, intf, rx, dst_ip=None, tgt_ip=None):
  68         if not dst_ip:
  69             dst_ip = intf.remote_ip6
  70         if not tgt_ip:
  71             dst_ip = intf.local_ip6
  72
  73         # unicasted packets must come to the unicast mac
  74         self.assertEqual(rx[Ether].dst, intf.remote_mac)
  75
  76         # and from the router's MAC
  77         self.assertEqual(rx[Ether].src, intf.local_mac)
  78
  79         # the rx'd NA should be addressed to the sender's source
  80         self.assertTrue(rx.haslayer(ICMPv6ND_NA))
  81         self.assertEqual(in6_ptop(rx[IPv6].dst),
  82                          in6_ptop(dst_ip))
  83
  84         # and come from the target address
  85         self.assertEqual(
  86             in6_ptop(rx[IPv6].src), in6_ptop(tgt_ip))
  87
  88         # Dest link-layer options should have the router's MAC
  89         dll = rx[ICMPv6NDOptDstLLAddr]
  90         self.assertEqual(dll.lladdr, intf.local_mac)
  91
  92     def validate_ns(self, intf, rx, tgt_ip):
  93         nsma = in6_getnsma(inet_pton(AF_INET6, tgt_ip))
  94         dst_ip = inet_ntop(AF_INET6, nsma)
  95
  96         # NS is broadcast
  97         self.assertEqual(rx[Ether].dst, in6_getnsmac(nsma))
  98
  99         # and from the router's MAC
 100         self.assertEqual(rx[Ether].src, intf.local_mac)
 101
 102         # the rx'd NS should be addressed to an mcast address
 103         # derived from the target address
 104         self.assertEqual(
 105             in6_ptop(rx[IPv6].dst), in6_ptop(dst_ip))
 106
 107         # expect the tgt IP in the NS header
 108         ns = rx[ICMPv6ND_NS]
 109         self.assertEqual(in6_ptop(ns.tgt), in6_ptop(tgt_ip))
 110
 111         # packet is from the router's local address
 112         self.assertEqual(
 113             in6_ptop(rx[IPv6].src), intf.local_ip6)
 114
 115         # Src link-layer options should have the router's MAC
 116         sll = rx[ICMPv6NDOptSrcLLAddr]
 117         self.assertEqual(sll.lladdr, intf.local_mac)
 118
 119     def send_and_expect_ra(self, intf, pkts, remark, dst_ip=None,
 120                            filter_out_fn=is_ipv6_misc):
 121         intf.add_stream(pkts)
 122         self.pg_enable_capture(self.pg_interfaces)
 123         self.pg_start()
 124         rx = intf.get_capture(1, filter_out_fn=filter_out_fn)
 125
 126         self.assertEqual(len(rx), 1)
 127         rx = rx[0]
 128         self.validate_ra(intf, rx, dst_ip)
 129
 130     def send_and_expect_na(self, intf, pkts, remark, dst_ip=None,
 131                            tgt_ip=None,
 132                            filter_out_fn=is_ipv6_misc):
 133         intf.add_stream(pkts)
 134         self.pg_enable_capture(self.pg_interfaces)
 135         self.pg_start()
 136         rx = intf.get_capture(1, filter_out_fn=filter_out_fn)
 137
 138         self.assertEqual(len(rx), 1)
 139         rx = rx[0]
 140         self.validate_na(intf, rx, dst_ip, tgt_ip)
 141
 142     def send_and_expect_ns(self, tx_intf, rx_intf, pkts, tgt_ip,
 143                            filter_out_fn=is_ipv6_misc):
 144         self.vapi.cli("clear trace")
 145         tx_intf.add_stream(pkts)
 146         self.pg_enable_capture(self.pg_interfaces)
 147         self.pg_start()
 148         rx = rx_intf.get_capture(1, filter_out_fn=filter_out_fn)
 149
 150         self.assertEqual(len(rx), 1)
 151         rx = rx[0]
 152         self.validate_ns(rx_intf, rx, tgt_ip)
 153
 154     def verify_ip(self, rx, smac, dmac, sip, dip):
 155         ether = rx[Ether]
 156         self.assertEqual(ether.dst, dmac)
 157         self.assertEqual(ether.src, smac)
 158
 159         ip = rx[IPv6]
 160         self.assertEqual(ip.src, sip)
 161         self.assertEqual(ip.dst, dip)
 162
 163
 164 class TestIPv6(TestIPv6ND):
 165     """ IPv6 Test Case """
 166
 167     @classmethod
 168     def setUpClass(cls):
 169         super(TestIPv6, cls).setUpClass()
 170
 171     @classmethod
 172     def tearDownClass(cls):
 173         super(TestIPv6, cls).tearDownClass()
 174
 175     def setUp(self):
 176         """
 177         Perform test setup before test case.
 178
 179         **Config:**
 180             - create 3 pg interfaces
 181                 - untagged pg0 interface
 182                 - Dot1Q subinterface on pg1
 183                 - Dot1AD subinterface on pg2
 184             - setup interfaces:
 185                 - put it into UP state
 186                 - set IPv6 addresses
 187                 - resolve neighbor address using NDP
 188             - configure 200 fib entries
 189
 190         :ivar list interfaces: pg interfaces and subinterfaces.
 191         :ivar dict flows: IPv4 packet flows in test.
 192
 193         *TODO:* Create AD sub interface
 194         """
 195         super(TestIPv6, self).setUp()
 196
 197         # create 3 pg interfaces
 198         self.create_pg_interfaces(range(3))
 199
 200         # create 2 subinterfaces for p1 and pg2
 201         self.sub_interfaces = [
 202             VppDot1QSubint(self, self.pg1, 100),
 203             VppDot1QSubint(self, self.pg2, 200)
 204             # TODO: VppDot1ADSubint(self, self.pg2, 200, 300, 400)
 205         ]
 206
 207         # packet flows mapping pg0 -> pg1.sub, pg2.sub, etc.
 208         self.flows = dict()
 209         self.flows[self.pg0] = [self.pg1.sub_if, self.pg2.sub_if]
 210         self.flows[self.pg1.sub_if] = [self.pg0, self.pg2.sub_if]
 211         self.flows[self.pg2.sub_if] = [self.pg0, self.pg1.sub_if]
 212
 213         # packet sizes
 214         self.pg_if_packet_sizes = [64, 1500, 9020]
 215
 216         self.interfaces = list(self.pg_interfaces)
 217         self.interfaces.extend(self.sub_interfaces)
 218
 219         # setup all interfaces
 220         for i in self.interfaces:
 221             i.admin_up()
 222             i.config_ip6()
 223             i.resolve_ndp()
 224
 225     def tearDown(self):
 226         """Run standard test teardown and log ``show ip6 neighbors``."""
 227         for i in self.interfaces:
 228             i.unconfig_ip6()
 229             i.admin_down()
 230         for i in self.sub_interfaces:
 231             i.remove_vpp_config()
 232
 233         super(TestIPv6, self).tearDown()
 234         if not self.vpp_dead:
 235             self.logger.info(self.vapi.cli("show ip6 neighbors"))
 236             # info(self.vapi.cli("show ip6 fib"))  # many entries
 237
 238     def modify_packet(self, src_if, packet_size, pkt):
 239         """Add load, set destination IP and extend packet to required packet
 240         size for defined interface.
 241
 242         :param VppInterface src_if: Interface to create packet for.
 243         :param int packet_size: Required packet size.
 244         :param Scapy pkt: Packet to be modified.
 245         """
 246         dst_if_idx = int(packet_size / 10 % 2)
 247         dst_if = self.flows[src_if][dst_if_idx]
 248         info = self.create_packet_info(src_if, dst_if)
 249         payload = self.info_to_payload(info)
 250         p = pkt / Raw(payload)
 251         p[IPv6].dst = dst_if.remote_ip6
 252         info.data = p.copy()
 253         if isinstance(src_if, VppSubInterface):
 254             p = src_if.add_dot1_layer(p)
 255         self.extend_packet(p, packet_size)
 256
 257         return p
 258
 259     def create_stream(self, src_if):
 260         """Create input packet stream for defined interface.
 261
 262         :param VppInterface src_if: Interface to create packet stream for.
 263         """
 264         hdr_ext = 4 if isinstance(src_if, VppSubInterface) else 0
 265         pkt_tmpl = (Ether(dst=src_if.local_mac, src=src_if.remote_mac) /
 266                     IPv6(src=src_if.remote_ip6) /
 267                     inet6.UDP(sport=1234, dport=1234))
 268
 269         pkts = [self.modify_packet(src_if, i, pkt_tmpl)
 270                 for i in moves.range(self.pg_if_packet_sizes[0],
 271                                      self.pg_if_packet_sizes[1], 10)]
 272         pkts_b = [self.modify_packet(src_if, i, pkt_tmpl)
 273                   for i in moves.range(self.pg_if_packet_sizes[1] + hdr_ext,
 274                                        self.pg_if_packet_sizes[2] + hdr_ext,
 275                                        50)]
 276         pkts.extend(pkts_b)
 277
 278         return pkts
 279
 280     def verify_capture(self, dst_if, capture):
 281         """Verify captured input packet stream for defined interface.
 282
 283         :param VppInterface dst_if: Interface to verify captured packet stream
 284                                     for.
 285         :param list capture: Captured packet stream.
 286         """
 287         self.logger.info("Verifying capture on interface %s" % dst_if.name)
 288         last_info = dict()
 289         for i in self.interfaces:
 290             last_info[i.sw_if_index] = None
 291         is_sub_if = False
 292         dst_sw_if_index = dst_if.sw_if_index
 293         if hasattr(dst_if, 'parent'):
 294             is_sub_if = True
 295         for packet in capture:
 296             if is_sub_if:
 297                 # Check VLAN tags and Ethernet header
 298                 packet = dst_if.remove_dot1_layer(packet)
 299             self.assertTrue(Dot1Q not in packet)
 300             try:
 301                 ip = packet[IPv6]
 302                 udp = packet[inet6.UDP]
 303                 payload_info = self.payload_to_info(packet[Raw])
 304                 packet_index = payload_info.index
 305                 self.assertEqual(payload_info.dst, dst_sw_if_index)
 306                 self.logger.debug(
 307                     "Got packet on port %s: src=%u (id=%u)" %
 308                     (dst_if.name, payload_info.src, packet_index))
 309                 next_info = self.get_next_packet_info_for_interface2(
 310                     payload_info.src, dst_sw_if_index,
 311                     last_info[payload_info.src])
 312                 last_info[payload_info.src] = next_info
 313                 self.assertTrue(next_info is not None)
 314                 self.assertEqual(packet_index, next_info.index)
 315                 saved_packet = next_info.data
 316                 # Check standard fields
 317                 self.assertEqual(
 318                     ip.src, saved_packet[IPv6].src)
 319                 self.assertEqual(
 320                     ip.dst, saved_packet[IPv6].dst)
 321                 self.assertEqual(
 322                     udp.sport, saved_packet[inet6.UDP].sport)
 323                 self.assertEqual(
 324                     udp.dport, saved_packet[inet6.UDP].dport)
 325             except:
 326                 self.logger.error(ppp("Unexpected or invalid packet:", packet))
 327                 raise
 328         for i in self.interfaces:
 329             remaining_packet = self.get_next_packet_info_for_interface2(
 330                 i.sw_if_index, dst_sw_if_index, last_info[i.sw_if_index])
 331             self.assertTrue(remaining_packet is None,
 332                             "Interface %s: Packet expected from interface %s "
 333                             "didn't arrive" % (dst_if.name, i.name))
 334
 335     def test_next_header_anomaly(self):
 336         """ IPv6 next header anomaly test
 337
 338         Test scenario:
 339             - ipv6 next header field = Fragment Header (44)
 340             - next header is ICMPv6 Echo Request
 341             - wait for reassembly
 342         """
 343         pkt = (Ether(src=self.pg0.local_mac, dst=self.pg0.remote_mac) /
 344                IPv6(src=self.pg0.remote_ip6, dst=self.pg0.local_ip6, nh=44) /
 345                ICMPv6EchoRequest())
 346
 347         self.pg0.add_stream(pkt)
 348         self.pg_start()
 349
 350         # wait for reassembly
 351         self.sleep(10)
 352
 353     def test_fib(self):
 354         """ IPv6 FIB test
 355
 356         Test scenario:
 357             - Create IPv6 stream for pg0 interface
 358             - Create IPv6 tagged streams for pg1's and pg2's subinterface.
 359             - Send and verify received packets on each interface.
 360         """
 361
 362         pkts = self.create_stream(self.pg0)
 363         self.pg0.add_stream(pkts)
 364
 365         for i in self.sub_interfaces:
 366             pkts = self.create_stream(i)
 367             i.parent.add_stream(pkts)
 368
 369         self.pg_enable_capture(self.pg_interfaces)
 370         self.pg_start()
 371
 372         pkts = self.pg0.get_capture()
 373         self.verify_capture(self.pg0, pkts)
 374
 375         for i in self.sub_interfaces:
 376             pkts = i.parent.get_capture()
 377             self.verify_capture(i, pkts)
 378
 379     def test_ns(self):
 380         """ IPv6 Neighbour Solicitation Exceptions
 381
 382         Test scenario:
 383            - Send an NS Sourced from an address not covered by the link sub-net
 384            - Send an NS to an mcast address the router has not joined
 385            - Send NS for a target address the router does not onn.
 386         """
 387
 388         #
 389         # An NS from a non link source address
 390         #
 391         nsma = in6_getnsma(inet_pton(AF_INET6, self.pg0.local_ip6))
 392         d = inet_ntop(AF_INET6, nsma)
 393
 394         p = (Ether(dst=in6_getnsmac(nsma)) /
 395              IPv6(dst=d, src="2002::2") /
 396              ICMPv6ND_NS(tgt=self.pg0.local_ip6) /
 397              ICMPv6NDOptSrcLLAddr(
 398                  lladdr=self.pg0.remote_mac))
 399         pkts = [p]
 400
 401         self.send_and_assert_no_replies(
 402             self.pg0, pkts,
 403             "No response to NS source by address not on sub-net")
 404
 405         #
 406         # An NS for sent to a solicited mcast group the router is
 407         # not a member of FAILS
 408         #
 409         if 0:
 410             nsma = in6_getnsma(inet_pton(AF_INET6, "fd::ffff"))
 411             d = inet_ntop(AF_INET6, nsma)
 412
 413             p = (Ether(dst=in6_getnsmac(nsma)) /
 414                  IPv6(dst=d, src=self.pg0.remote_ip6) /
 415                  ICMPv6ND_NS(tgt=self.pg0.local_ip6) /
 416                  ICMPv6NDOptSrcLLAddr(
 417                      lladdr=self.pg0.remote_mac))
 418             pkts = [p]
 419
 420             self.send_and_assert_no_replies(
 421                 self.pg0, pkts,
 422                 "No response to NS sent to unjoined mcast address")
 423
 424         #
 425         # An NS whose target address is one the router does not own
 426         #
 427         nsma = in6_getnsma(inet_pton(AF_INET6, self.pg0.local_ip6))
 428         d = inet_ntop(AF_INET6, nsma)
 429
 430         p = (Ether(dst=in6_getnsmac(nsma)) /
 431              IPv6(dst=d, src=self.pg0.remote_ip6) /
 432              ICMPv6ND_NS(tgt="fd::ffff") /
 433              ICMPv6NDOptSrcLLAddr(
 434                  lladdr=self.pg0.remote_mac))
 435         pkts = [p]
 436
 437         self.send_and_assert_no_replies(self.pg0, pkts,
 438                                         "No response to NS for unknown target")
 439
 440         #
 441         # A neighbor entry that has no associated FIB-entry
 442         #
 443         self.pg0.generate_remote_hosts(4)
 444         nd_entry = VppNeighbor(self,
 445                                self.pg0.sw_if_index,
 446                                self.pg0.remote_hosts[2].mac,
 447                                self.pg0.remote_hosts[2].ip6,
 448                                is_no_fib_entry=1)
 449         nd_entry.add_vpp_config()
 450
 451         #
 452         # check we have the neighbor, but no route
 453         #
 454         self.assertTrue(find_nbr(self,
 455                                  self.pg0.sw_if_index,
 456                                  self.pg0._remote_hosts[2].ip6))
 457         self.assertFalse(find_route(self,
 458                                     self.pg0._remote_hosts[2].ip6,
 459                                     128))
 460
 461         #
 462         # send an NS from a link local address to the interface's global
 463         # address
 464         #
 465         p = (Ether(dst=in6_getnsmac(nsma), src=self.pg0.remote_mac) /
 466              IPv6(
 467                  dst=d, src=self.pg0._remote_hosts[2].ip6_ll) /
 468              ICMPv6ND_NS(tgt=self.pg0.local_ip6) /
 469              ICMPv6NDOptSrcLLAddr(
 470                  lladdr=self.pg0.remote_mac))
 471
 472         self.send_and_expect_na(self.pg0, p,
 473                                 "NS from link-local",
 474                                 dst_ip=self.pg0._remote_hosts[2].ip6_ll,
 475                                 tgt_ip=self.pg0.local_ip6)
 476
 477         #
 478         # we should have learned an ND entry for the peer's link-local
 479         # but not inserted a route to it in the FIB
 480         #
 481         self.assertTrue(find_nbr(self,
 482                                  self.pg0.sw_if_index,
 483                                  self.pg0._remote_hosts[2].ip6_ll))
 484         self.assertFalse(find_route(self,
 485                                     self.pg0._remote_hosts[2].ip6_ll,
 486                                     128))
 487
 488         #
 489         # An NS to the router's own Link-local
 490         #
 491         p = (Ether(dst=in6_getnsmac(nsma), src=self.pg0.remote_mac) /
 492              IPv6(
 493                  dst=d, src=self.pg0._remote_hosts[3].ip6_ll) /
 494              ICMPv6ND_NS(tgt=self.pg0.local_ip6_ll) /
 495              ICMPv6NDOptSrcLLAddr(
 496                  lladdr=self.pg0.remote_mac))
 497
 498         self.send_and_expect_na(self.pg0, p,
 499                                 "NS to/from link-local",
 500                                 dst_ip=self.pg0._remote_hosts[3].ip6_ll,
 501                                 tgt_ip=self.pg0.local_ip6_ll)
 502
 503         #
 504         # we should have learned an ND entry for the peer's link-local
 505         # but not inserted a route to it in the FIB
 506         #
 507         self.assertTrue(find_nbr(self,
 508                                  self.pg0.sw_if_index,
 509                                  self.pg0._remote_hosts[3].ip6_ll))
 510         self.assertFalse(find_route(self,
 511                                     self.pg0._remote_hosts[3].ip6_ll,
 512                                     128))
 513
 514     def test_ns_duplicates(self):
 515         """ ND Duplicates"""
 516
 517         #
 518         # Generate some hosts on the LAN
 519         #
 520         self.pg1.generate_remote_hosts(3)
 521
 522         #
 523         # Add host 1 on pg1 and pg2
 524         #
 525         ns_pg1 = VppNeighbor(self,
 526                              self.pg1.sw_if_index,
 527                              self.pg1.remote_hosts[1].mac,
 528                              self.pg1.remote_hosts[1].ip6)
 529         ns_pg1.add_vpp_config()
 530         ns_pg2 = VppNeighbor(self,
 531                              self.pg2.sw_if_index,
 532                              self.pg2.remote_mac,
 533                              self.pg1.remote_hosts[1].ip6)
 534         ns_pg2.add_vpp_config()
 535
 536         #
 537         # IP packet destined for pg1 remote host arrives on pg1 again.
 538         #
 539         p = (Ether(dst=self.pg0.local_mac,
 540                    src=self.pg0.remote_mac) /
 541              IPv6(src=self.pg0.remote_ip6,
 542                   dst=self.pg1.remote_hosts[1].ip6) /
 543              inet6.UDP(sport=1234, dport=1234) /
 544              Raw())
 545
 546         self.pg0.add_stream(p)
 547         self.pg_enable_capture(self.pg_interfaces)
 548         self.pg_start()
 549
 550         rx1 = self.pg1.get_capture(1)
 551
 552         self.verify_ip(rx1[0],
 553                        self.pg1.local_mac,
 554                        self.pg1.remote_hosts[1].mac,
 555                        self.pg0.remote_ip6,
 556                        self.pg1.remote_hosts[1].ip6)
 557
 558         #
 559         # remove the duplicate on pg1
 560         # packet stream should generate NSs out of pg1
 561         #
 562         ns_pg1.remove_vpp_config()
 563
 564         self.send_and_expect_ns(self.pg0, self.pg1,
 565                                 p, self.pg1.remote_hosts[1].ip6)
 566
 567         #
 568         # Add it back
 569         #
 570         ns_pg1.add_vpp_config()
 571
 572         self.pg0.add_stream(p)
 573         self.pg_enable_capture(self.pg_interfaces)
 574         self.pg_start()
 575
 576         rx1 = self.pg1.get_capture(1)
 577
 578         self.verify_ip(rx1[0],
 579                        self.pg1.local_mac,
 580                        self.pg1.remote_hosts[1].mac,
 581                        self.pg0.remote_ip6,
 582                        self.pg1.remote_hosts[1].ip6)
 583
 584     def validate_ra(self, intf, rx, dst_ip=None, src_ip=None,
 585                     mtu=9000, pi_opt=None):
 586         if not dst_ip:
 587             dst_ip = intf.remote_ip6
 588         if not src_ip:
 589             src_ip = mk_ll_addr(intf.local_mac)
 590
 591         # unicasted packets must come to the unicast mac
 592         self.assertEqual(rx[Ether].dst, intf.remote_mac)
 593
 594         # and from the router's MAC
 595         self.assertEqual(rx[Ether].src, intf.local_mac)
 596
 597         # the rx'd RA should be addressed to the sender's source
 598         self.assertTrue(rx.haslayer(ICMPv6ND_RA))
 599         self.assertEqual(in6_ptop(rx[IPv6].dst),
 600                          in6_ptop(dst_ip))
 601
 602         # and come from the router's link local
 603         self.assertTrue(in6_islladdr(rx[IPv6].src))
 604         self.assertEqual(in6_ptop(rx[IPv6].src), in6_ptop(src_ip))
 605
 606         # it should contain the links MTU
 607         ra = rx[ICMPv6ND_RA]
 608         self.assertEqual(ra[ICMPv6NDOptMTU].mtu, mtu)
 609
 610         # it should contain the source's link layer address option
 611         sll = ra[ICMPv6NDOptSrcLLAddr]
 612         self.assertEqual(sll.lladdr, intf.local_mac)
 613
 614         if not pi_opt:
 615             # the RA should not contain prefix information
 616             self.assertFalse(ra.haslayer(
 617                 ICMPv6NDOptPrefixInfo))
 618         else:
 619             raos = rx.getlayer(ICMPv6NDOptPrefixInfo, 1)
 620
 621             # the options are nested in the scapy packet in way that i cannot
 622             # decipher how to decode. this 1st layer of option always returns
 623             # nested classes, so a direct obj1=obj2 comparison always fails.
 624             # however, the getlayer(.., 2) does give one instance.
 625             # so we cheat here and construct a new opt instance for comparison
 626             rd = ICMPv6NDOptPrefixInfo(
 627                 prefixlen=raos.prefixlen,
 628                 prefix=raos.prefix,
 629                 L=raos.L,
 630                 A=raos.A)
 631             if type(pi_opt) is list:
 632                 for ii in range(len(pi_opt)):
 633                     self.assertEqual(pi_opt[ii], rd)
 634                     rd = rx.getlayer(
 635                         ICMPv6NDOptPrefixInfo, ii + 2)
 636             else:
 637                 self.assertEqual(pi_opt, raos, 'Expected: %s, received: %s'
 638                                  % (pi_opt.show(dump=True),
 639                                     raos.show(dump=True)))
 640
 641     def send_and_expect_ra(self, intf, pkts, remark, dst_ip=None,
 642                            filter_out_fn=is_ipv6_misc,
 643                            opt=None,
 644                            src_ip=None):
 645         self.vapi.cli("clear trace")
 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, src_ip=src_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(dst=self.pg0.local_ip6, src=self.pg0.remote_ip6) /
 677              ICMPv6ND_RS())
 678         pkts = [p]
 679         self.send_and_expect_ra(self.pg0, pkts, "Genuine RS")
 680
 681         #
 682         # For the next RS sent the RA should be rate limited
 683         #
 684         self.send_and_assert_no_replies(self.pg0, pkts, "RA rate limited")
 685
 686         #
 687         # When we reconfigure the IPv6 RA config,
 688         # 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('%s/%s' % (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('%s/%s' % (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('%s/%s' % (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('%s/%s' % (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('%s/%s' % (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 defaults option to revert to defaults
 860         #  L and A flag are clear in the advert
 861         #
 862         self.pg0.ip6_ra_prefix('%s/%s' % (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('%s/%s' % (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 prefix-info - expect the second is still in the
 909         # advert
 910         #
 911         self.pg0.ip6_ra_prefix('%s/%s' % (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 in the adverts
 929         #
 930         self.pg0.ip6_ra_prefix('%s/%s' % (self.pg1.local_ip6,
 931                                self.pg1.local_ip6_prefix_len),
 932                                is_no=1)
 933
 934         #
 935         # change the link's link local, so we know that works too.
 936         #
 937         self.vapi.sw_interface_ip6_set_link_local_address(
 938             sw_if_index=self.pg0.sw_if_index,
 939             ip="fe80::88")
 940
 941         self.pg0.ip6_ra_config(send_unicast=1)
 942         self.send_and_expect_ra(self.pg0, p,
 943                                 "RA with Prefix reverted to defaults",
 944                                 dst_ip=ll,
 945                                 src_ip="fe80::88")
 946
 947         #
 948         # Reset the periodic advertisements back to default values
 949         #
 950         self.pg0.ip6_ra_config(no=1, suppress=1, send_unicast=0)
 951
 952     def test_mld(self):
 953         """ MLD Report """
 954         #
 955         # test one MLD is sent after applying an IPv6 Address on an interface
 956         #
 957         self.pg_enable_capture(self.pg_interfaces)
 958         self.pg_start()
 959
 960         subitf = VppDot1QSubint(self, self.pg1, 99)
 961
 962         subitf.admin_up()
 963         subitf.config_ip6()
 964
 965         rxs = self.pg1._get_capture(timeout=4, filter_out_fn=None)
 966
 967         #
 968         # hunt for the MLD on vlan 99
 969         #
 970         for rx in rxs:
 971             # make sure ipv6 packets with hop by hop options have
 972             # correct checksums
 973             self.assert_packet_checksums_valid(rx)
 974             if rx.haslayer(IPv6ExtHdrHopByHop) and \
 975                rx.haslayer(Dot1Q) and \
 976                rx[Dot1Q].vlan == 99:
 977                 mld = rx[ICMPv6MLReport2]
 978
 979         self.assertEqual(mld.records_number, 4)
 980
 981
 982 class TestIPv6IfAddrRoute(VppTestCase):
 983     """ IPv6 Interface Addr Route Test Case """
 984
 985     @classmethod
 986     def setUpClass(cls):
 987         super(TestIPv6IfAddrRoute, cls).setUpClass()
 988
 989     @classmethod
 990     def tearDownClass(cls):
 991         super(TestIPv6IfAddrRoute, cls).tearDownClass()
 992
 993     def setUp(self):
 994         super(TestIPv6IfAddrRoute, self).setUp()
 995
 996         # create 1 pg interface
 997         self.create_pg_interfaces(range(1))
 998
 999         for i in self.pg_interfaces:
1000             i.admin_up()
1001             i.config_ip6()
1002             i.resolve_ndp()
1003
1004     def tearDown(self):
1005         super(TestIPv6IfAddrRoute, self).tearDown()
1006         for i in self.pg_interfaces:
1007             i.unconfig_ip6()
1008             i.admin_down()
1009
1010     def test_ipv6_ifaddrs_same_prefix(self):
1011         """ IPv6 Interface Addresses Same Prefix test
1012
1013         Test scenario:
1014
1015             - Verify no route in FIB for prefix 2001:10::/64
1016             - Configure IPv4 address 2001:10::10/64  on an interface
1017             - Verify route in FIB for prefix 2001:10::/64
1018             - Configure IPv4 address 2001:10::20/64 on an interface
1019             - Delete 2001:10::10/64 from interface
1020             - Verify route in FIB for prefix 2001:10::/64
1021             - Delete 2001:10::20/64 from interface
1022             - Verify no route in FIB for prefix 2001:10::/64
1023         """
1024
1025         addr1 = "2001:10::10"
1026         addr2 = "2001:10::20"
1027
1028         if_addr1 = VppIpInterfaceAddress(self, self.pg0, addr1, 64)
1029         if_addr2 = VppIpInterfaceAddress(self, self.pg0, addr2, 64)
1030         self.assertFalse(if_addr1.query_vpp_config())
1031         self.assertFalse(find_route(self, addr1, 128))
1032         self.assertFalse(find_route(self, addr2, 128))
1033
1034         # configure first address, verify route present
1035         if_addr1.add_vpp_config()
1036         self.assertTrue(if_addr1.query_vpp_config())
1037         self.assertTrue(find_route(self, addr1, 128))
1038         self.assertFalse(find_route(self, addr2, 128))
1039
1040         # configure second address, delete first, verify route not removed
1041         if_addr2.add_vpp_config()
1042         if_addr1.remove_vpp_config()
1043         self.assertFalse(if_addr1.query_vpp_config())
1044         self.assertTrue(if_addr2.query_vpp_config())
1045         self.assertFalse(find_route(self, addr1, 128))
1046         self.assertTrue(find_route(self, addr2, 128))
1047
1048         # delete second address, verify route removed
1049         if_addr2.remove_vpp_config()
1050         self.assertFalse(if_addr1.query_vpp_config())
1051         self.assertFalse(find_route(self, addr1, 128))
1052         self.assertFalse(find_route(self, addr2, 128))
1053
1054
1055 class TestICMPv6Echo(VppTestCase):
1056     """ ICMPv6 Echo Test Case """
1057
1058     @classmethod
1059     def setUpClass(cls):
1060         super(TestICMPv6Echo, cls).setUpClass()
1061
1062     @classmethod
1063     def tearDownClass(cls):
1064         super(TestICMPv6Echo, cls).tearDownClass()
1065
1066     def setUp(self):
1067         super(TestICMPv6Echo, self).setUp()
1068
1069         # create 1 pg interface
1070         self.create_pg_interfaces(range(1))
1071
1072         for i in self.pg_interfaces:
1073             i.admin_up()
1074             i.config_ip6()
1075             i.resolve_ndp()
1076
1077     def tearDown(self):
1078         super(TestICMPv6Echo, self).tearDown()
1079         for i in self.pg_interfaces:
1080             i.unconfig_ip6()
1081             i.admin_down()
1082
1083     def test_icmpv6_echo(self):
1084         """ VPP replies to ICMPv6 Echo Request
1085
1086         Test scenario:
1087
1088             - Receive ICMPv6 Echo Request message on pg0 interface.
1089             - Check outgoing ICMPv6 Echo Reply message on pg0 interface.
1090         """
1091
1092         icmpv6_id = 0xb
1093         icmpv6_seq = 5
1094         icmpv6_data = b'\x0a' * 18
1095         p_echo_request = (Ether(src=self.pg0.remote_mac,
1096                                 dst=self.pg0.local_mac) /
1097                           IPv6(src=self.pg0.remote_ip6,
1098                                dst=self.pg0.local_ip6) /
1099                           ICMPv6EchoRequest(
1100                               id=icmpv6_id,
1101                               seq=icmpv6_seq,
1102                               data=icmpv6_data))
1103
1104         self.pg0.add_stream(p_echo_request)
1105         self.pg_enable_capture(self.pg_interfaces)
1106         self.pg_start()
1107
1108         rx = self.pg0.get_capture(1)
1109         rx = rx[0]
1110         ether = rx[Ether]
1111         ipv6 = rx[IPv6]
1112         icmpv6 = rx[ICMPv6EchoReply]
1113
1114         self.assertEqual(ether.src, self.pg0.local_mac)
1115         self.assertEqual(ether.dst, self.pg0.remote_mac)
1116
1117         self.assertEqual(ipv6.src, self.pg0.local_ip6)
1118         self.assertEqual(ipv6.dst, self.pg0.remote_ip6)
1119
1120         self.assertEqual(
1121             icmp6types[icmpv6.type], "Echo Reply")
1122         self.assertEqual(icmpv6.id, icmpv6_id)
1123         self.assertEqual(icmpv6.seq, icmpv6_seq)
1124         self.assertEqual(icmpv6.data, icmpv6_data)
1125
1126
1127 class TestIPv6RD(TestIPv6ND):
1128     """ IPv6 Router Discovery Test Case """
1129
1130     @classmethod
1131     def setUpClass(cls):
1132         super(TestIPv6RD, cls).setUpClass()
1133
1134     @classmethod
1135     def tearDownClass(cls):
1136         super(TestIPv6RD, cls).tearDownClass()
1137
1138     def setUp(self):
1139         super(TestIPv6RD, self).setUp()
1140
1141         # create 2 pg interfaces
1142         self.create_pg_interfaces(range(2))
1143
1144         self.interfaces = list(self.pg_interfaces)
1145
1146         # setup all interfaces
1147         for i in self.interfaces:
1148             i.admin_up()
1149             i.config_ip6()
1150
1151     def tearDown(self):
1152         for i in self.interfaces:
1153             i.unconfig_ip6()
1154             i.admin_down()
1155         super(TestIPv6RD, self).tearDown()
1156
1157     def test_rd_send_router_solicitation(self):
1158         """ Verify router solicitation packets """
1159
1160         count = 2
1161         self.pg_enable_capture(self.pg_interfaces)
1162         self.pg_start()
1163         self.vapi.ip6nd_send_router_solicitation(self.pg1.sw_if_index,
1164                                                  mrc=count)
1165         rx_list = self.pg1.get_capture(count, timeout=3)
1166         self.assertEqual(len(rx_list), count)
1167         for packet in rx_list:
1168             self.assertEqual(packet.haslayer(IPv6), 1)
1169             self.assertEqual(packet[IPv6].haslayer(
1170                 ICMPv6ND_RS), 1)
1171             dst = ip6_normalize(packet[IPv6].dst)
1172             dst2 = ip6_normalize("ff02::2")
1173             self.assert_equal(dst, dst2)
1174             src = ip6_normalize(packet[IPv6].src)
1175             src2 = ip6_normalize(self.pg1.local_ip6_ll)
1176             self.assert_equal(src, src2)
1177             self.assertTrue(
1178                 bool(packet[ICMPv6ND_RS].haslayer(
1179                     ICMPv6NDOptSrcLLAddr)))
1180             self.assert_equal(
1181                 packet[ICMPv6NDOptSrcLLAddr].lladdr,
1182                 self.pg1.local_mac)
1183
1184     def verify_prefix_info(self, reported_prefix, prefix_option):
1185         prefix = IPv6Network(
1186             text_type(prefix_option.getfieldval("prefix") +
1187                       "/" +
1188                       text_type(prefix_option.getfieldval("prefixlen"))),
1189             strict=False)
1190         self.assert_equal(reported_prefix.prefix.network_address,
1191                           prefix.network_address)
1192         L = prefix_option.getfieldval("L")
1193         A = prefix_option.getfieldval("A")
1194         option_flags = (L << 7) | (A << 6)
1195         self.assert_equal(reported_prefix.flags, option_flags)
1196         self.assert_equal(reported_prefix.valid_time,
1197                           prefix_option.getfieldval("validlifetime"))
1198         self.assert_equal(reported_prefix.preferred_time,
1199                           prefix_option.getfieldval("preferredlifetime"))
1200
1201     def test_rd_receive_router_advertisement(self):
1202         """ Verify events triggered by received RA packets """
1203
1204         self.vapi.want_ip6_ra_events(enable=1)
1205
1206         prefix_info_1 = ICMPv6NDOptPrefixInfo(
1207             prefix="1::2",
1208             prefixlen=50,
1209             validlifetime=200,
1210             preferredlifetime=500,
1211             L=1,
1212             A=1,
1213         )
1214
1215         prefix_info_2 = ICMPv6NDOptPrefixInfo(
1216             prefix="7::4",
1217             prefixlen=20,
1218             validlifetime=70,
1219             preferredlifetime=1000,
1220             L=1,
1221             A=0,
1222         )
1223
1224         p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
1225              IPv6(dst=self.pg1.local_ip6_ll,
1226                   src=mk_ll_addr(self.pg1.remote_mac)) /
1227              ICMPv6ND_RA() /
1228              prefix_info_1 /
1229              prefix_info_2)
1230         self.pg1.add_stream([p])
1231         self.pg_start()
1232
1233         ev = self.vapi.wait_for_event(10, "ip6_ra_event")
1234
1235         self.assert_equal(ev.current_hop_limit, 0)
1236         self.assert_equal(ev.flags, 8)
1237         self.assert_equal(ev.router_lifetime_in_sec, 1800)
1238         self.assert_equal(ev.neighbor_reachable_time_in_msec, 0)
1239         self.assert_equal(
1240             ev.time_in_msec_between_retransmitted_neighbor_solicitations, 0)
1241
1242         self.assert_equal(ev.n_prefixes, 2)
1243
1244         self.verify_prefix_info(ev.prefixes[0], prefix_info_1)
1245         self.verify_prefix_info(ev.prefixes[1], prefix_info_2)
1246
1247
1248 class TestIPv6RDControlPlane(TestIPv6ND):
1249     """ IPv6 Router Discovery Control Plane Test Case """
1250
1251     @classmethod
1252     def setUpClass(cls):
1253         super(TestIPv6RDControlPlane, cls).setUpClass()
1254
1255     @classmethod
1256     def tearDownClass(cls):
1257         super(TestIPv6RDControlPlane, cls).tearDownClass()
1258
1259     def setUp(self):
1260         super(TestIPv6RDControlPlane, self).setUp()
1261
1262         # create 1 pg interface
1263         self.create_pg_interfaces(range(1))
1264
1265         self.interfaces = list(self.pg_interfaces)
1266
1267         # setup all interfaces
1268         for i in self.interfaces:
1269             i.admin_up()
1270             i.config_ip6()
1271
1272     def tearDown(self):
1273         super(TestIPv6RDControlPlane, self).tearDown()
1274
1275     @staticmethod
1276     def create_ra_packet(pg, routerlifetime=None):
1277         src_ip = pg.remote_ip6_ll
1278         dst_ip = pg.local_ip6
1279         if routerlifetime is not None:
1280             ra = ICMPv6ND_RA(routerlifetime=routerlifetime)
1281         else:
1282             ra = ICMPv6ND_RA()
1283         p = (Ether(dst=pg.local_mac, src=pg.remote_mac) /
1284              IPv6(dst=dst_ip, src=src_ip) / ra)
1285         return p
1286
1287     @staticmethod
1288     def get_default_routes(fib):
1289         list = []
1290         for entry in fib:
1291             if entry.route.prefix.prefixlen == 0:
1292                 for path in entry.route.paths:
1293                     if path.sw_if_index != 0xFFFFFFFF:
1294                         defaut_route = {}
1295                         defaut_route['sw_if_index'] = path.sw_if_index
1296                         defaut_route['next_hop'] = path.nh.address.ip6
1297                         list.append(defaut_route)
1298         return list
1299
1300     @staticmethod
1301     def get_interface_addresses(fib, pg):
1302         list = []
1303         for entry in fib:
1304             if entry.route.prefix.prefixlen == 128:
1305                 path = entry.route.paths[0]
1306                 if path.sw_if_index == pg.sw_if_index:
1307                     list.append(str(entry.route.prefix.network_address))
1308         return list
1309
1310     def wait_for_no_default_route(self, n_tries=50, s_time=1):
1311         while (n_tries):
1312             fib = self.vapi.ip_route_dump(0, True)
1313             default_routes = self.get_default_routes(fib)
1314             if 0 == len(default_routes):
1315                 return True
1316             n_tries = n_tries - 1
1317             self.sleep(s_time)
1318
1319         return False
1320
1321     def test_all(self):
1322         """ Test handling of SLAAC addresses and default routes """
1323
1324         fib = self.vapi.ip_route_dump(0, True)
1325         default_routes = self.get_default_routes(fib)
1326         initial_addresses = set(self.get_interface_addresses(fib, self.pg0))
1327         self.assertEqual(default_routes, [])
1328         router_address = IPv6Address(text_type(self.pg0.remote_ip6_ll))
1329
1330         self.vapi.ip6_nd_address_autoconfig(self.pg0.sw_if_index, 1, 1)
1331
1332         self.sleep(0.1)
1333
1334         # send RA
1335         packet = (self.create_ra_packet(
1336             self.pg0) / ICMPv6NDOptPrefixInfo(
1337             prefix="1::",
1338             prefixlen=64,
1339             validlifetime=2,
1340             preferredlifetime=2,
1341             L=1,
1342             A=1,
1343         ) / ICMPv6NDOptPrefixInfo(
1344             prefix="7::",
1345             prefixlen=20,
1346             validlifetime=1500,
1347             preferredlifetime=1000,
1348             L=1,
1349             A=0,
1350         ))
1351         self.pg0.add_stream([packet])
1352         self.pg_start()
1353
1354         self.sleep_on_vpp_time(0.1)
1355
1356         fib = self.vapi.ip_route_dump(0, True)
1357
1358         # check FIB for new address
1359         addresses = set(self.get_interface_addresses(fib, self.pg0))
1360         new_addresses = addresses.difference(initial_addresses)
1361         self.assertEqual(len(new_addresses), 1)
1362         prefix = IPv6Network(text_type("%s/%d" % (list(new_addresses)[0], 20)),
1363                              strict=False)
1364         self.assertEqual(prefix, IPv6Network(text_type('1::/20')))
1365
1366         # check FIB for new default route
1367         default_routes = self.get_default_routes(fib)
1368         self.assertEqual(len(default_routes), 1)
1369         dr = default_routes[0]
1370         self.assertEqual(dr['sw_if_index'], self.pg0.sw_if_index)
1371         self.assertEqual(dr['next_hop'], router_address)
1372
1373         # send RA to delete default route
1374         packet = self.create_ra_packet(self.pg0, routerlifetime=0)
1375         self.pg0.add_stream([packet])
1376         self.pg_start()
1377
1378         self.sleep_on_vpp_time(0.1)
1379
1380         # check that default route is deleted
1381         fib = self.vapi.ip_route_dump(0, True)
1382         default_routes = self.get_default_routes(fib)
1383         self.assertEqual(len(default_routes), 0)
1384
1385         self.sleep_on_vpp_time(0.1)
1386
1387         # send RA
1388         packet = self.create_ra_packet(self.pg0)
1389         self.pg0.add_stream([packet])
1390         self.pg_start()
1391
1392         self.sleep_on_vpp_time(0.1)
1393
1394         # check FIB for new default route
1395         fib = self.vapi.ip_route_dump(0, True)
1396         default_routes = self.get_default_routes(fib)
1397         self.assertEqual(len(default_routes), 1)
1398         dr = default_routes[0]
1399         self.assertEqual(dr['sw_if_index'], self.pg0.sw_if_index)
1400         self.assertEqual(dr['next_hop'], router_address)
1401
1402         # send RA, updating router lifetime to 1s
1403         packet = self.create_ra_packet(self.pg0, 1)
1404         self.pg0.add_stream([packet])
1405         self.pg_start()
1406
1407         self.sleep_on_vpp_time(0.1)
1408
1409         # check that default route still exists
1410         fib = self.vapi.ip_route_dump(0, True)
1411         default_routes = self.get_default_routes(fib)
1412         self.assertEqual(len(default_routes), 1)
1413         dr = default_routes[0]
1414         self.assertEqual(dr['sw_if_index'], self.pg0.sw_if_index)
1415         self.assertEqual(dr['next_hop'], router_address)
1416
1417         self.sleep_on_vpp_time(1)
1418
1419         # check that default route is deleted
1420         self.assertTrue(self.wait_for_no_default_route())
1421
1422         # check FIB still contains the SLAAC address
1423         addresses = set(self.get_interface_addresses(fib, self.pg0))
1424         new_addresses = addresses.difference(initial_addresses)
1425
1426         self.assertEqual(len(new_addresses), 1)
1427         prefix = IPv6Network(text_type("%s/%d" % (list(new_addresses)[0], 20)),
1428                              strict=False)
1429         self.assertEqual(prefix, IPv6Network(text_type('1::/20')))
1430
1431         self.sleep_on_vpp_time(1)
1432
1433         # check that SLAAC address is deleted
1434         fib = self.vapi.ip_route_dump(0, True)
1435         addresses = set(self.get_interface_addresses(fib, self.pg0))
1436         new_addresses = addresses.difference(initial_addresses)
1437         self.assertEqual(len(new_addresses), 0)
1438
1439
1440 class IPv6NDProxyTest(TestIPv6ND):
1441     """ IPv6 ND ProxyTest Case """
1442
1443     @classmethod
1444     def setUpClass(cls):
1445         super(IPv6NDProxyTest, cls).setUpClass()
1446
1447     @classmethod
1448     def tearDownClass(cls):
1449         super(IPv6NDProxyTest, cls).tearDownClass()
1450
1451     def setUp(self):
1452         super(IPv6NDProxyTest, self).setUp()
1453
1454         # create 3 pg interfaces
1455         self.create_pg_interfaces(range(3))
1456
1457         # pg0 is the master interface, with the configured subnet
1458         self.pg0.admin_up()
1459         self.pg0.config_ip6()
1460         self.pg0.resolve_ndp()
1461
1462         self.pg1.ip6_enable()
1463         self.pg2.ip6_enable()
1464
1465     def tearDown(self):
1466         super(IPv6NDProxyTest, self).tearDown()
1467
1468     def test_nd_proxy(self):
1469         """ IPv6 Proxy ND """
1470
1471         #
1472         # Generate some hosts in the subnet that we are proxying
1473         #
1474         self.pg0.generate_remote_hosts(8)
1475
1476         nsma = in6_getnsma(inet_pton(AF_INET6, self.pg0.local_ip6))
1477         d = inet_ntop(AF_INET6, nsma)
1478
1479         #
1480         # Send an NS for one of those remote hosts on one of the proxy links
1481         # expect no response since it's from an address that is not
1482         # on the link that has the prefix configured
1483         #
1484         ns_pg1 = (Ether(dst=in6_getnsmac(nsma), src=self.pg1.remote_mac) /
1485                   IPv6(dst=d,
1486                        src=self.pg0._remote_hosts[2].ip6) /
1487                   ICMPv6ND_NS(tgt=self.pg0.local_ip6) /
1488                   ICMPv6NDOptSrcLLAddr(
1489                       lladdr=self.pg0._remote_hosts[2].mac))
1490
1491         self.send_and_assert_no_replies(self.pg1, ns_pg1, "Off link NS")
1492
1493         #
1494         # Add proxy support for the host
1495         #
1496         self.vapi.ip6nd_proxy_add_del(
1497             is_add=1, ip=inet_pton(AF_INET6, self.pg0._remote_hosts[2].ip6),
1498             sw_if_index=self.pg1.sw_if_index)
1499
1500         #
1501         # try that NS again. this time we expect an NA back
1502         #
1503         self.send_and_expect_na(self.pg1, ns_pg1,
1504                                 "NS to proxy entry",
1505                                 dst_ip=self.pg0._remote_hosts[2].ip6,
1506                                 tgt_ip=self.pg0.local_ip6)
1507
1508         #
1509         # ... and that we have an entry in the ND cache
1510         #
1511         self.assertTrue(find_nbr(self,
1512                                  self.pg1.sw_if_index,
1513                                  self.pg0._remote_hosts[2].ip6))
1514
1515         #
1516         # ... and we can route traffic to it
1517         #
1518         t = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
1519              IPv6(dst=self.pg0._remote_hosts[2].ip6,
1520                   src=self.pg0.remote_ip6) /
1521              inet6.UDP(sport=10000, dport=20000) /
1522              Raw(b'\xa5' * 100))
1523
1524         self.pg0.add_stream(t)
1525         self.pg_enable_capture(self.pg_interfaces)
1526         self.pg_start()
1527         rx = self.pg1.get_capture(1)
1528         rx = rx[0]
1529
1530         self.assertEqual(rx[Ether].dst, self.pg0._remote_hosts[2].mac)
1531         self.assertEqual(rx[Ether].src, self.pg1.local_mac)
1532
1533         self.assertEqual(rx[IPv6].src,
1534                          t[IPv6].src)
1535         self.assertEqual(rx[IPv6].dst,
1536                          t[IPv6].dst)
1537
1538         #
1539         # Test we proxy for the host on the main interface
1540         #
1541         ns_pg0 = (Ether(dst=in6_getnsmac(nsma), src=self.pg0.remote_mac) /
1542                   IPv6(dst=d, src=self.pg0.remote_ip6) /
1543                   ICMPv6ND_NS(
1544                       tgt=self.pg0._remote_hosts[2].ip6) /
1545                   ICMPv6NDOptSrcLLAddr(
1546                       lladdr=self.pg0.remote_mac))
1547
1548         self.send_and_expect_na(self.pg0, ns_pg0,
1549                                 "NS to proxy entry on main",
1550                                 tgt_ip=self.pg0._remote_hosts[2].ip6,
1551                                 dst_ip=self.pg0.remote_ip6)
1552
1553         #
1554         # Setup and resolve proxy for another host on another interface
1555         #
1556         ns_pg2 = (Ether(dst=in6_getnsmac(nsma), src=self.pg2.remote_mac) /
1557                   IPv6(dst=d,
1558                        src=self.pg0._remote_hosts[3].ip6) /
1559                   ICMPv6ND_NS(tgt=self.pg0.local_ip6) /
1560                   ICMPv6NDOptSrcLLAddr(
1561                       lladdr=self.pg0._remote_hosts[2].mac))
1562
1563         self.vapi.ip6nd_proxy_add_del(
1564             is_add=1, ip=inet_pton(AF_INET6, self.pg0._remote_hosts[3].ip6),
1565             sw_if_index=self.pg2.sw_if_index)
1566
1567         self.send_and_expect_na(self.pg2, ns_pg2,
1568                                 "NS to proxy entry other interface",
1569                                 dst_ip=self.pg0._remote_hosts[3].ip6,
1570                                 tgt_ip=self.pg0.local_ip6)
1571
1572         self.assertTrue(find_nbr(self,
1573                                  self.pg2.sw_if_index,
1574                                  self.pg0._remote_hosts[3].ip6))
1575
1576         #
1577         # hosts can communicate. pg2->pg1
1578         #
1579         t2 = (Ether(dst=self.pg2.local_mac,
1580                     src=self.pg0.remote_hosts[3].mac) /
1581               IPv6(dst=self.pg0._remote_hosts[2].ip6,
1582                    src=self.pg0._remote_hosts[3].ip6) /
1583               inet6.UDP(sport=10000, dport=20000) /
1584               Raw(b'\xa5' * 100))
1585
1586         self.pg2.add_stream(t2)
1587         self.pg_enable_capture(self.pg_interfaces)
1588         self.pg_start()
1589         rx = self.pg1.get_capture(1)
1590         rx = rx[0]
1591
1592         self.assertEqual(rx[Ether].dst, self.pg0._remote_hosts[2].mac)
1593         self.assertEqual(rx[Ether].src, self.pg1.local_mac)
1594
1595         self.assertEqual(rx[IPv6].src,
1596                          t2[IPv6].src)
1597         self.assertEqual(rx[IPv6].dst,
1598                          t2[IPv6].dst)
1599
1600         #
1601         # remove the proxy configs
1602         #
1603         self.vapi.ip6nd_proxy_add_del(
1604             ip=inet_pton(AF_INET6, self.pg0._remote_hosts[2].ip6),
1605             sw_if_index=self.pg1.sw_if_index, is_add=0)
1606         self.vapi.ip6nd_proxy_add_del(
1607             ip=inet_pton(AF_INET6, self.pg0._remote_hosts[3].ip6),
1608             sw_if_index=self.pg2.sw_if_index, is_add=0)
1609
1610         self.assertFalse(find_nbr(self,
1611                                   self.pg2.sw_if_index,
1612                                   self.pg0._remote_hosts[3].ip6))
1613         self.assertFalse(find_nbr(self,
1614                                   self.pg1.sw_if_index,
1615                                   self.pg0._remote_hosts[2].ip6))
1616
1617         #
1618         # no longer proxy-ing...
1619         #
1620         self.send_and_assert_no_replies(self.pg0, ns_pg0, "Proxy unconfigured")
1621         self.send_and_assert_no_replies(self.pg1, ns_pg1, "Proxy unconfigured")
1622         self.send_and_assert_no_replies(self.pg2, ns_pg2, "Proxy unconfigured")
1623
1624         #
1625         # no longer forwarding. traffic generates NS out of the glean/main
1626         # interface
1627         #
1628         self.pg2.add_stream(t2)
1629         self.pg_enable_capture(self.pg_interfaces)
1630         self.pg_start()
1631
1632         rx = self.pg0.get_capture(1)
1633
1634         self.assertTrue(rx[0].haslayer(ICMPv6ND_NS))
1635
1636
1637 class TestIPNull(VppTestCase):
1638     """ IPv6 routes via NULL """
1639
1640     @classmethod
1641     def setUpClass(cls):
1642         super(TestIPNull, cls).setUpClass()
1643
1644     @classmethod
1645     def tearDownClass(cls):
1646         super(TestIPNull, cls).tearDownClass()
1647
1648     def setUp(self):
1649         super(TestIPNull, self).setUp()
1650
1651         # create 2 pg interfaces
1652         self.create_pg_interfaces(range(1))
1653
1654         for i in self.pg_interfaces:
1655             i.admin_up()
1656             i.config_ip6()
1657             i.resolve_ndp()
1658
1659     def tearDown(self):
1660         super(TestIPNull, self).tearDown()
1661         for i in self.pg_interfaces:
1662             i.unconfig_ip6()
1663             i.admin_down()
1664
1665     def test_ip_null(self):
1666         """ IP NULL route """
1667
1668         p = (Ether(src=self.pg0.remote_mac,
1669                    dst=self.pg0.local_mac) /
1670              IPv6(src=self.pg0.remote_ip6, dst="2001::1") /
1671              inet6.UDP(sport=1234, dport=1234) /
1672              Raw(b'\xa5' * 100))
1673
1674         #
1675         # A route via IP NULL that will reply with ICMP unreachables
1676         #
1677         ip_unreach = VppIpRoute(
1678             self, "2001::", 64,
1679             [VppRoutePath("::", 0xffffffff,
1680                           type=FibPathType.FIB_PATH_TYPE_ICMP_UNREACH)])
1681         ip_unreach.add_vpp_config()
1682
1683         self.pg0.add_stream(p)
1684         self.pg_enable_capture(self.pg_interfaces)
1685         self.pg_start()
1686
1687         rx = self.pg0.get_capture(1)
1688         rx = rx[0]
1689         icmp = rx[ICMPv6DestUnreach]
1690
1691         # 0 = "No route to destination"
1692         self.assertEqual(icmp.code, 0)
1693
1694         # ICMP is rate limited. pause a bit
1695         self.sleep(1)
1696
1697         #
1698         # A route via IP NULL that will reply with ICMP prohibited
1699         #
1700         ip_prohibit = VppIpRoute(
1701             self, "2001::1", 128,
1702             [VppRoutePath("::", 0xffffffff,
1703                           type=FibPathType.FIB_PATH_TYPE_ICMP_PROHIBIT)])
1704         ip_prohibit.add_vpp_config()
1705
1706         self.pg0.add_stream(p)
1707         self.pg_enable_capture(self.pg_interfaces)
1708         self.pg_start()
1709
1710         rx = self.pg0.get_capture(1)
1711         rx = rx[0]
1712         icmp = rx[ICMPv6DestUnreach]
1713
1714         # 1 = "Communication with destination administratively prohibited"
1715         self.assertEqual(icmp.code, 1)
1716
1717
1718 class TestIPDisabled(VppTestCase):
1719     """ IPv6 disabled """
1720
1721     @classmethod
1722     def setUpClass(cls):
1723         super(TestIPDisabled, cls).setUpClass()
1724
1725     @classmethod
1726     def tearDownClass(cls):
1727         super(TestIPDisabled, cls).tearDownClass()
1728
1729     def setUp(self):
1730         super(TestIPDisabled, self).setUp()
1731
1732         # create 2 pg interfaces
1733         self.create_pg_interfaces(range(2))
1734
1735         # PG0 is IP enabled
1736         self.pg0.admin_up()
1737         self.pg0.config_ip6()
1738         self.pg0.resolve_ndp()
1739
1740         # PG 1 is not IP enabled
1741         self.pg1.admin_up()
1742
1743     def tearDown(self):
1744         super(TestIPDisabled, self).tearDown()
1745         for i in self.pg_interfaces:
1746             i.unconfig_ip4()
1747             i.admin_down()
1748
1749     def test_ip_disabled(self):
1750         """ IP Disabled """
1751
1752         #
1753         # An (S,G).
1754         # one accepting interface, pg0, 2 forwarding interfaces
1755         #
1756         route_ff_01 = VppIpMRoute(
1757             self,
1758             "::",
1759             "ffef::1", 128,
1760             MRouteEntryFlags.MFIB_ENTRY_FLAG_NONE,
1761             [VppMRoutePath(self.pg1.sw_if_index,
1762                            MRouteItfFlags.MFIB_ITF_FLAG_ACCEPT),
1763              VppMRoutePath(self.pg0.sw_if_index,
1764                            MRouteItfFlags.MFIB_ITF_FLAG_FORWARD)])
1765         route_ff_01.add_vpp_config()
1766
1767         pu = (Ether(src=self.pg1.remote_mac,
1768                     dst=self.pg1.local_mac) /
1769               IPv6(src="2001::1", dst=self.pg0.remote_ip6) /
1770               inet6.UDP(sport=1234, dport=1234) /
1771               Raw(b'\xa5' * 100))
1772         pm = (Ether(src=self.pg1.remote_mac,
1773                     dst=self.pg1.local_mac) /
1774               IPv6(src="2001::1", dst="ffef::1") /
1775               inet6.UDP(sport=1234, dport=1234) /
1776               Raw(b'\xa5' * 100))
1777
1778         #
1779         # PG1 does not forward IP traffic
1780         #
1781         self.send_and_assert_no_replies(self.pg1, pu, "IPv6 disabled")
1782         self.send_and_assert_no_replies(self.pg1, pm, "IPv6 disabled")
1783
1784         #
1785         # IP enable PG1
1786         #
1787         self.pg1.config_ip6()
1788
1789         #
1790         # Now we get packets through
1791         #
1792         self.pg1.add_stream(pu)
1793         self.pg_enable_capture(self.pg_interfaces)
1794         self.pg_start()
1795         rx = self.pg0.get_capture(1)
1796
1797         self.pg1.add_stream(pm)
1798         self.pg_enable_capture(self.pg_interfaces)
1799         self.pg_start()
1800         rx = self.pg0.get_capture(1)
1801
1802         #
1803         # Disable PG1
1804         #
1805         self.pg1.unconfig_ip6()
1806
1807         #
1808         # PG1 does not forward IP traffic
1809         #
1810         self.send_and_assert_no_replies(self.pg1, pu, "IPv6 disabled")
1811         self.send_and_assert_no_replies(self.pg1, pm, "IPv6 disabled")
1812
1813
1814 class TestIP6LoadBalance(VppTestCase):
1815     """ IPv6 Load-Balancing """
1816
1817     @classmethod
1818     def setUpClass(cls):
1819         super(TestIP6LoadBalance, cls).setUpClass()
1820
1821     @classmethod
1822     def tearDownClass(cls):
1823         super(TestIP6LoadBalance, cls).tearDownClass()
1824
1825     def setUp(self):
1826         super(TestIP6LoadBalance, self).setUp()
1827
1828         self.create_pg_interfaces(range(5))
1829
1830         mpls_tbl = VppMplsTable(self, 0)
1831         mpls_tbl.add_vpp_config()
1832
1833         for i in self.pg_interfaces:
1834             i.admin_up()
1835             i.config_ip6()
1836             i.resolve_ndp()
1837             i.enable_mpls()
1838
1839     def tearDown(self):
1840         for i in self.pg_interfaces:
1841             i.unconfig_ip6()
1842             i.admin_down()
1843             i.disable_mpls()
1844         super(TestIP6LoadBalance, self).tearDown()
1845
1846     def pg_send(self, input, pkts):
1847         self.vapi.cli("clear trace")
1848         input.add_stream(pkts)
1849         self.pg_enable_capture(self.pg_interfaces)
1850         self.pg_start()
1851
1852     def send_and_expect_load_balancing(self, input, pkts, outputs):
1853         self.pg_send(input, pkts)
1854         for oo in outputs:
1855             rx = oo._get_capture(1)
1856             self.assertNotEqual(0, len(rx))
1857
1858     def send_and_expect_one_itf(self, input, pkts, itf):
1859         self.pg_send(input, pkts)
1860         rx = itf.get_capture(len(pkts))
1861
1862     def test_ip6_load_balance(self):
1863         """ IPv6 Load-Balancing """
1864
1865         #
1866         # An array of packets that differ only in the destination port
1867         #  - IP only
1868         #  - MPLS EOS
1869         #  - MPLS non-EOS
1870         #  - MPLS non-EOS with an entropy label
1871         #
1872         port_ip_pkts = []
1873         port_mpls_pkts = []
1874         port_mpls_neos_pkts = []
1875         port_ent_pkts = []
1876
1877         #
1878         # An array of packets that differ only in the source address
1879         #
1880         src_ip_pkts = []
1881         src_mpls_pkts = []
1882
1883         for ii in range(NUM_PKTS):
1884             port_ip_hdr = (
1885                 IPv6(dst="3000::1", src="3000:1::1") /
1886                 inet6.UDP(sport=1234, dport=1234 + ii) /
1887                 Raw(b'\xa5' * 100))
1888             port_ip_pkts.append((Ether(src=self.pg0.remote_mac,
1889                                        dst=self.pg0.local_mac) /
1890                                  port_ip_hdr))
1891             port_mpls_pkts.append((Ether(src=self.pg0.remote_mac,
1892                                          dst=self.pg0.local_mac) /
1893                                    MPLS(label=66, ttl=2) /
1894                                    port_ip_hdr))
1895             port_mpls_neos_pkts.append((Ether(src=self.pg0.remote_mac,
1896                                               dst=self.pg0.local_mac) /
1897                                         MPLS(label=67, ttl=2) /
1898                                         MPLS(label=77, ttl=2) /
1899                                         port_ip_hdr))
1900             port_ent_pkts.append((Ether(src=self.pg0.remote_mac,
1901                                         dst=self.pg0.local_mac) /
1902                                   MPLS(label=67, ttl=2) /
1903                                   MPLS(label=14, ttl=2) /
1904                                   MPLS(label=999, ttl=2) /
1905                                   port_ip_hdr))
1906             src_ip_hdr = (
1907                 IPv6(dst="3000::1", src="3000:1::%d" % ii) /
1908                 inet6.UDP(sport=1234, dport=1234) /
1909                 Raw(b'\xa5' * 100))
1910             src_ip_pkts.append((Ether(src=self.pg0.remote_mac,
1911                                       dst=self.pg0.local_mac) /
1912                                 src_ip_hdr))
1913             src_mpls_pkts.append((Ether(src=self.pg0.remote_mac,
1914                                         dst=self.pg0.local_mac) /
1915                                   MPLS(label=66, ttl=2) /
1916                                   src_ip_hdr))
1917
1918         #
1919         # A route for the IP packets
1920         #
1921         route_3000_1 = VppIpRoute(self, "3000::1", 128,
1922                                   [VppRoutePath(self.pg1.remote_ip6,
1923                                                 self.pg1.sw_if_index),
1924                                    VppRoutePath(self.pg2.remote_ip6,
1925                                                 self.pg2.sw_if_index)])
1926         route_3000_1.add_vpp_config()
1927
1928         #
1929         # a local-label for the EOS packets
1930         #
1931         binding = VppMplsIpBind(self, 66, "3000::1", 128, is_ip6=1)
1932         binding.add_vpp_config()
1933
1934         #
1935         # An MPLS route for the non-EOS packets
1936         #
1937         route_67 = VppMplsRoute(self, 67, 0,
1938                                 [VppRoutePath(self.pg1.remote_ip6,
1939                                               self.pg1.sw_if_index,
1940                                               labels=[67]),
1941                                  VppRoutePath(self.pg2.remote_ip6,
1942                                               self.pg2.sw_if_index,
1943                                               labels=[67])])
1944         route_67.add_vpp_config()
1945
1946         #
1947         # inject the packet on pg0 - expect load-balancing across the 2 paths
1948         #  - since the default hash config is to use IP src,dst and port
1949         #    src,dst
1950         # We are not going to ensure equal amounts of packets across each link,
1951         # since the hash algorithm is statistical and therefore this can never
1952         # be guaranteed. But with 64 different packets we do expect some
1953         # balancing. So instead just ensure there is traffic on each link.
1954         #
1955         self.send_and_expect_load_balancing(self.pg0, port_ip_pkts,
1956                                             [self.pg1, self.pg2])
1957         self.send_and_expect_load_balancing(self.pg0, src_ip_pkts,
1958                                             [self.pg1, self.pg2])
1959         self.send_and_expect_load_balancing(self.pg0, port_mpls_pkts,
1960                                             [self.pg1, self.pg2])
1961         self.send_and_expect_load_balancing(self.pg0, src_mpls_pkts,
1962                                             [self.pg1, self.pg2])
1963         self.send_and_expect_load_balancing(self.pg0, port_mpls_neos_pkts,
1964                                             [self.pg1, self.pg2])
1965
1966         #
1967         # The packets with Entropy label in should not load-balance,
1968         # since the Entropy value is fixed.
1969         #
1970         self.send_and_expect_one_itf(self.pg0, port_ent_pkts, self.pg1)
1971
1972         #
1973         # change the flow hash config so it's only IP src,dst
1974         #  - now only the stream with differing source address will
1975         #    load-balance
1976         #
1977         self.vapi.set_ip_flow_hash(vrf_id=0, src=1, dst=1, sport=0, dport=0,
1978                                    is_ipv6=1)
1979
1980         self.send_and_expect_load_balancing(self.pg0, src_ip_pkts,
1981                                             [self.pg1, self.pg2])
1982         self.send_and_expect_load_balancing(self.pg0, src_mpls_pkts,
1983                                             [self.pg1, self.pg2])
1984         self.send_and_expect_one_itf(self.pg0, port_ip_pkts, self.pg2)
1985
1986         #
1987         # change the flow hash config back to defaults
1988         #
1989         self.vapi.set_ip_flow_hash(vrf_id=0, src=1, dst=1, sport=1, dport=1,
1990                                    is_ipv6=1)
1991
1992         #
1993         # Recursive prefixes
1994         #  - testing that 2 stages of load-balancing occurs and there is no
1995         #    polarisation (i.e. only 2 of 4 paths are used)
1996         #
1997         port_pkts = []
1998         src_pkts = []
1999
2000         for ii in range(257):
2001             port_pkts.append((Ether(src=self.pg0.remote_mac,
2002                                     dst=self.pg0.local_mac) /
2003                               IPv6(dst="4000::1",
2004                                    src="4000:1::1") /
2005                               inet6.UDP(sport=1234,
2006                                         dport=1234 + ii) /
2007                               Raw(b'\xa5' * 100)))
2008             src_pkts.append((Ether(src=self.pg0.remote_mac,
2009                                    dst=self.pg0.local_mac) /
2010                              IPv6(dst="4000::1",
2011                                   src="4000:1::%d" % ii) /
2012                              inet6.UDP(sport=1234, dport=1234) /
2013                              Raw(b'\xa5' * 100)))
2014
2015         route_3000_2 = VppIpRoute(self, "3000::2", 128,
2016                                   [VppRoutePath(self.pg3.remote_ip6,
2017                                                 self.pg3.sw_if_index),
2018                                    VppRoutePath(self.pg4.remote_ip6,
2019                                                 self.pg4.sw_if_index)])
2020         route_3000_2.add_vpp_config()
2021
2022         route_4000_1 = VppIpRoute(self, "4000::1", 128,
2023                                   [VppRoutePath("3000::1",
2024                                                 0xffffffff),
2025                                    VppRoutePath("3000::2",
2026                                                 0xffffffff)])
2027         route_4000_1.add_vpp_config()
2028
2029         #
2030         # inject the packet on pg0 - expect load-balancing across all 4 paths
2031         #
2032         self.vapi.cli("clear trace")
2033         self.send_and_expect_load_balancing(self.pg0, port_pkts,
2034                                             [self.pg1, self.pg2,
2035                                              self.pg3, self.pg4])
2036         self.send_and_expect_load_balancing(self.pg0, src_pkts,
2037                                             [self.pg1, self.pg2,
2038                                              self.pg3, self.pg4])
2039
2040         #
2041         # Recursive prefixes
2042         #  - testing that 2 stages of load-balancing no choices
2043         #
2044         port_pkts = []
2045
2046         for ii in range(257):
2047             port_pkts.append((Ether(src=self.pg0.remote_mac,
2048                                     dst=self.pg0.local_mac) /
2049                               IPv6(dst="6000::1",
2050                                    src="6000:1::1") /
2051                               inet6.UDP(sport=1234,
2052                                         dport=1234 + ii) /
2053                               Raw(b'\xa5' * 100)))
2054
2055         route_5000_2 = VppIpRoute(self, "5000::2", 128,
2056                                   [VppRoutePath(self.pg3.remote_ip6,
2057                                                 self.pg3.sw_if_index)])
2058         route_5000_2.add_vpp_config()
2059
2060         route_6000_1 = VppIpRoute(self, "6000::1", 128,
2061                                   [VppRoutePath("5000::2",
2062                                                 0xffffffff)])
2063         route_6000_1.add_vpp_config()
2064
2065         #
2066         # inject the packet on pg0 - expect load-balancing across all 4 paths
2067         #
2068         self.vapi.cli("clear trace")
2069         self.send_and_expect_one_itf(self.pg0, port_pkts, self.pg3)
2070
2071
2072 class TestIP6Punt(VppTestCase):
2073     """ IPv6 Punt Police/Redirect """
2074
2075     @classmethod
2076     def setUpClass(cls):
2077         super(TestIP6Punt, cls).setUpClass()
2078
2079     @classmethod
2080     def tearDownClass(cls):
2081         super(TestIP6Punt, cls).tearDownClass()
2082
2083     def setUp(self):
2084         super(TestIP6Punt, self).setUp()
2085
2086         self.create_pg_interfaces(range(4))
2087
2088         for i in self.pg_interfaces:
2089             i.admin_up()
2090             i.config_ip6()
2091             i.resolve_ndp()
2092
2093     def tearDown(self):
2094         super(TestIP6Punt, self).tearDown()
2095         for i in self.pg_interfaces:
2096             i.unconfig_ip6()
2097             i.admin_down()
2098
2099     def test_ip_punt(self):
2100         """ IP6 punt police and redirect """
2101
2102         p = (Ether(src=self.pg0.remote_mac,
2103                    dst=self.pg0.local_mac) /
2104              IPv6(src=self.pg0.remote_ip6,
2105                   dst=self.pg0.local_ip6) /
2106              inet6.TCP(sport=1234, dport=1234) /
2107              Raw(b'\xa5' * 100))
2108
2109         pkts = p * 1025
2110
2111         #
2112         # Configure a punt redirect via pg1.
2113         #
2114         nh_addr = self.pg1.remote_ip6
2115         self.vapi.ip_punt_redirect(self.pg0.sw_if_index,
2116                                    self.pg1.sw_if_index,
2117                                    nh_addr)
2118
2119         self.send_and_expect(self.pg0, pkts, self.pg1)
2120
2121         #
2122         # add a policer
2123         #
2124         policer = VppPolicer(self, "ip6-punt", 400, 0, 10, 0, rate_type=1)
2125         policer.add_vpp_config()
2126         self.vapi.ip_punt_police(policer.policer_index, is_ip6=1)
2127
2128         self.vapi.cli("clear trace")
2129         self.pg0.add_stream(pkts)
2130         self.pg_enable_capture(self.pg_interfaces)
2131         self.pg_start()
2132
2133         #
2134         # the number of packet received should be greater than 0,
2135         # but not equal to the number sent, since some were policed
2136         #
2137         rx = self.pg1._get_capture(1)
2138         self.assertGreater(len(rx), 0)
2139         self.assertLess(len(rx), len(pkts))
2140
2141         #
2142         # remove the policer. back to full rx
2143         #
2144         self.vapi.ip_punt_police(policer.policer_index, is_add=0, is_ip6=1)
2145         policer.remove_vpp_config()
2146         self.send_and_expect(self.pg0, pkts, self.pg1)
2147
2148         #
2149         # remove the redirect. expect full drop.
2150         #
2151         self.vapi.ip_punt_redirect(self.pg0.sw_if_index,
2152                                    self.pg1.sw_if_index,
2153                                    nh_addr,
2154                                    is_add=0)
2155         self.send_and_assert_no_replies(self.pg0, pkts,
2156                                         "IP no punt config")
2157
2158         #
2159         # Add a redirect that is not input port selective
2160         #
2161         self.vapi.ip_punt_redirect(0xffffffff,
2162                                    self.pg1.sw_if_index,
2163                                    nh_addr)
2164         self.send_and_expect(self.pg0, pkts, self.pg1)
2165
2166         self.vapi.ip_punt_redirect(0xffffffff,
2167                                    self.pg1.sw_if_index,
2168                                    nh_addr,
2169                                    is_add=0)
2170
2171     def test_ip_punt_dump(self):
2172         """ IP6 punt redirect dump"""
2173
2174         #
2175         # Configure a punt redirects
2176         #
2177         nh_addr = self.pg3.remote_ip6
2178         self.vapi.ip_punt_redirect(self.pg0.sw_if_index,
2179                                    self.pg3.sw_if_index,
2180                                    nh_addr)
2181         self.vapi.ip_punt_redirect(self.pg1.sw_if_index,
2182                                    self.pg3.sw_if_index,
2183                                    nh_addr)
2184         self.vapi.ip_punt_redirect(self.pg2.sw_if_index,
2185                                    self.pg3.sw_if_index,
2186                                    '0::0')
2187
2188         #
2189         # Dump pg0 punt redirects
2190         #
2191         punts = self.vapi.ip_punt_redirect_dump(self.pg0.sw_if_index,
2192                                                 is_ipv6=1)
2193         for p in punts:
2194             self.assertEqual(p.punt.rx_sw_if_index, self.pg0.sw_if_index)
2195
2196         #
2197         # Dump punt redirects for all interfaces
2198         #
2199         punts = self.vapi.ip_punt_redirect_dump(0xffffffff, is_ipv6=1)
2200         self.assertEqual(len(punts), 3)
2201         for p in punts:
2202             self.assertEqual(p.punt.tx_sw_if_index, self.pg3.sw_if_index)
2203         self.assertNotEqual(punts[1].punt.nh, self.pg3.remote_ip6)
2204         self.assertEqual(str(punts[2].punt.nh), '::')
2205
2206
2207 class TestIPDeag(VppTestCase):
2208     """ IPv6 Deaggregate Routes """
2209
2210     @classmethod
2211     def setUpClass(cls):
2212         super(TestIPDeag, cls).setUpClass()
2213
2214     @classmethod
2215     def tearDownClass(cls):
2216         super(TestIPDeag, cls).tearDownClass()
2217
2218     def setUp(self):
2219         super(TestIPDeag, self).setUp()
2220
2221         self.create_pg_interfaces(range(3))
2222
2223         for i in self.pg_interfaces:
2224             i.admin_up()
2225             i.config_ip6()
2226             i.resolve_ndp()
2227
2228     def tearDown(self):
2229         super(TestIPDeag, self).tearDown()
2230         for i in self.pg_interfaces:
2231             i.unconfig_ip6()
2232             i.admin_down()
2233
2234     def test_ip_deag(self):
2235         """ IP Deag Routes """
2236
2237         #
2238         # Create a table to be used for:
2239         #  1 - another destination address lookup
2240         #  2 - a source address lookup
2241         #
2242         table_dst = VppIpTable(self, 1, is_ip6=1)
2243         table_src = VppIpTable(self, 2, is_ip6=1)
2244         table_dst.add_vpp_config()
2245         table_src.add_vpp_config()
2246
2247         #
2248         # Add a route in the default table to point to a deag/
2249         # second lookup in each of these tables
2250         #
2251         route_to_dst = VppIpRoute(self, "1::1", 128,
2252                                   [VppRoutePath("::",
2253                                                 0xffffffff,
2254                                                 nh_table_id=1)])
2255         route_to_src = VppIpRoute(
2256             self, "1::2", 128,
2257             [VppRoutePath("::",
2258                           0xffffffff,
2259                           nh_table_id=2,
2260                           type=FibPathType.FIB_PATH_TYPE_SOURCE_LOOKUP)])
2261
2262         route_to_dst.add_vpp_config()
2263         route_to_src.add_vpp_config()
2264
2265         #
2266         # packets to these destination are dropped, since they'll
2267         # hit the respective default routes in the second table
2268         #
2269         p_dst = (Ether(src=self.pg0.remote_mac,
2270                        dst=self.pg0.local_mac) /
2271                  IPv6(src="5::5", dst="1::1") /
2272                  inet6.TCP(sport=1234, dport=1234) /
2273                  Raw(b'\xa5' * 100))
2274         p_src = (Ether(src=self.pg0.remote_mac,
2275                        dst=self.pg0.local_mac) /
2276                  IPv6(src="2::2", dst="1::2") /
2277                  inet6.TCP(sport=1234, dport=1234) /
2278                  Raw(b'\xa5' * 100))
2279         pkts_dst = p_dst * 257
2280         pkts_src = p_src * 257
2281
2282         self.send_and_assert_no_replies(self.pg0, pkts_dst,
2283                                         "IP in dst table")
2284         self.send_and_assert_no_replies(self.pg0, pkts_src,
2285                                         "IP in src table")
2286
2287         #
2288         # add a route in the dst table to forward via pg1
2289         #
2290         route_in_dst = VppIpRoute(self, "1::1", 128,
2291                                   [VppRoutePath(self.pg1.remote_ip6,
2292                                                 self.pg1.sw_if_index)],
2293                                   table_id=1)
2294         route_in_dst.add_vpp_config()
2295
2296         self.send_and_expect(self.pg0, pkts_dst, self.pg1)
2297
2298         #
2299         # add a route in the src table to forward via pg2
2300         #
2301         route_in_src = VppIpRoute(self, "2::2", 128,
2302                                   [VppRoutePath(self.pg2.remote_ip6,
2303                                                 self.pg2.sw_if_index)],
2304                                   table_id=2)
2305         route_in_src.add_vpp_config()
2306         self.send_and_expect(self.pg0, pkts_src, self.pg2)
2307
2308         #
2309         # loop in the lookup DP
2310         #
2311         route_loop = VppIpRoute(self, "3::3", 128,
2312                                 [VppRoutePath("::",
2313                                               0xffffffff)])
2314         route_loop.add_vpp_config()
2315
2316         p_l = (Ether(src=self.pg0.remote_mac,
2317                      dst=self.pg0.local_mac) /
2318                IPv6(src="3::4", dst="3::3") /
2319                inet6.TCP(sport=1234, dport=1234) /
2320                Raw(b'\xa5' * 100))
2321
2322         self.send_and_assert_no_replies(self.pg0, p_l * 257,
2323                                         "IP lookup loop")
2324
2325
2326 class TestIP6Input(VppTestCase):
2327     """ IPv6 Input Exception Test Cases """
2328
2329     @classmethod
2330     def setUpClass(cls):
2331         super(TestIP6Input, cls).setUpClass()
2332
2333     @classmethod
2334     def tearDownClass(cls):
2335         super(TestIP6Input, cls).tearDownClass()
2336
2337     def setUp(self):
2338         super(TestIP6Input, self).setUp()
2339
2340         self.create_pg_interfaces(range(2))
2341
2342         for i in self.pg_interfaces:
2343             i.admin_up()
2344             i.config_ip6()
2345             i.resolve_ndp()
2346
2347     def tearDown(self):
2348         super(TestIP6Input, self).tearDown()
2349         for i in self.pg_interfaces:
2350             i.unconfig_ip6()
2351             i.admin_down()
2352
2353     def test_ip_input_icmp_reply(self):
2354         """ IP6 Input Exception - Return ICMP (3,0) """
2355         #
2356         # hop limit - ICMP replies
2357         #
2358         p_version = (Ether(src=self.pg0.remote_mac,
2359                            dst=self.pg0.local_mac) /
2360                      IPv6(src=self.pg0.remote_ip6,
2361                           dst=self.pg1.remote_ip6,
2362                           hlim=1) /
2363                      inet6.UDP(sport=1234, dport=1234) /
2364                      Raw(b'\xa5' * 100))
2365
2366         rx = self.send_and_expect(self.pg0, p_version * NUM_PKTS, self.pg0)
2367         rx = rx[0]
2368         icmp = rx[ICMPv6TimeExceeded]
2369
2370         # 0: "hop limit exceeded in transit",
2371         self.assertEqual((icmp.type, icmp.code), (3, 0))
2372
2373     icmpv6_data = '\x0a' * 18
2374     all_0s = "::"
2375     all_1s = "FFFF:FFFF:FFFF:FFFF:FFFF:FFFF:FFFF:FFFF"
2376
2377     @parameterized.expand([
2378         # Name, src, dst, l4proto, msg, timeout
2379         ("src='iface',   dst='iface'", None, None,
2380          inet6.UDP(sport=1234, dport=1234), "funky version", None),
2381         ("src='All 0's', dst='iface'", all_0s, None,
2382          ICMPv6EchoRequest(id=0xb, seq=5, data=icmpv6_data), None, 0.1),
2383         ("src='iface',   dst='All 0's'", None, all_0s,
2384          ICMPv6EchoRequest(id=0xb, seq=5, data=icmpv6_data), None, 0.1),
2385         ("src='All 1's', dst='iface'", all_1s, None,
2386          ICMPv6EchoRequest(id=0xb, seq=5, data=icmpv6_data), None, 0.1),
2387         ("src='iface',   dst='All 1's'", None, all_1s,
2388          ICMPv6EchoRequest(id=0xb, seq=5, data=icmpv6_data), None, 0.1),
2389         ("src='All 1's', dst='All 1's'", all_1s, all_1s,
2390          ICMPv6EchoRequest(id=0xb, seq=5, data=icmpv6_data), None, 0.1),
2391
2392     ])
2393     def test_ip_input_no_replies(self, name, src, dst, l4, msg, timeout):
2394
2395         self._testMethodDoc = 'IPv6 Input Exception - %s' % name
2396
2397         p_version = (Ether(src=self.pg0.remote_mac,
2398                            dst=self.pg0.local_mac) /
2399                      IPv6(src=src or self.pg0.remote_ip6,
2400                           dst=dst or self.pg1.remote_ip6,
2401                           version=3) /
2402                      l4 /
2403                      Raw(b'\xa5' * 100))
2404
2405         self.send_and_assert_no_replies(self.pg0, p_version * NUM_PKTS,
2406                                         remark=msg or "",
2407                                         timeout=timeout)
2408
2409     def test_hop_by_hop(self):
2410         """ Hop-by-hop header test """
2411
2412         p = (Ether(src=self.pg0.remote_mac,
2413                    dst=self.pg0.local_mac) /
2414              IPv6(src=self.pg0.remote_ip6, dst=self.pg0.local_ip6) /
2415              IPv6ExtHdrHopByHop() /
2416              inet6.UDP(sport=1234, dport=1234) /
2417              Raw(b'\xa5' * 100))
2418
2419         self.pg0.add_stream(p)
2420         self.pg_enable_capture(self.pg_interfaces)
2421         self.pg_start()
2422
2423
2424 class TestIPReplace(VppTestCase):
2425     """ IPv6 Table Replace """
2426
2427     @classmethod
2428     def setUpClass(cls):
2429         super(TestIPReplace, cls).setUpClass()
2430
2431     @classmethod
2432     def tearDownClass(cls):
2433         super(TestIPReplace, cls).tearDownClass()
2434
2435     def setUp(self):
2436         super(TestIPReplace, self).setUp()
2437
2438         self.create_pg_interfaces(range(4))
2439
2440         table_id = 1
2441         self.tables = []
2442
2443         for i in self.pg_interfaces:
2444             i.admin_up()
2445             i.config_ip6()
2446             i.generate_remote_hosts(2)
2447             self.tables.append(VppIpTable(self, table_id,
2448                                           True).add_vpp_config())
2449             table_id += 1
2450
2451     def tearDown(self):
2452         super(TestIPReplace, self).tearDown()
2453         for i in self.pg_interfaces:
2454             i.admin_down()
2455             i.unconfig_ip6()
2456
2457     def test_replace(self):
2458         """ IP Table Replace """
2459
2460         N_ROUTES = 20
2461         links = [self.pg0, self.pg1, self.pg2, self.pg3]
2462         routes = [[], [], [], []]
2463
2464         # the sizes of 'empty' tables
2465         for t in self.tables:
2466             self.assertEqual(len(t.dump()), 2)
2467             self.assertEqual(len(t.mdump()), 5)
2468
2469         # load up the tables with some routes
2470         for ii, t in enumerate(self.tables):
2471             for jj in range(1, N_ROUTES):
2472                 uni = VppIpRoute(
2473                     self, "2001::%d" % jj if jj != 0 else "2001::", 128,
2474                     [VppRoutePath(links[ii].remote_hosts[0].ip6,
2475                                   links[ii].sw_if_index),
2476                      VppRoutePath(links[ii].remote_hosts[1].ip6,
2477                                   links[ii].sw_if_index)],
2478                     table_id=t.table_id).add_vpp_config()
2479                 multi = VppIpMRoute(
2480                     self, "::",
2481                     "ff:2001::%d" % jj, 128,
2482                     MRouteEntryFlags.MFIB_ENTRY_FLAG_NONE,
2483                     [VppMRoutePath(self.pg0.sw_if_index,
2484                                    MRouteItfFlags.MFIB_ITF_FLAG_ACCEPT,
2485                                    proto=FibPathProto.FIB_PATH_NH_PROTO_IP6),
2486                      VppMRoutePath(self.pg1.sw_if_index,
2487                                    MRouteItfFlags.MFIB_ITF_FLAG_FORWARD,
2488                                    proto=FibPathProto.FIB_PATH_NH_PROTO_IP6),
2489                      VppMRoutePath(self.pg2.sw_if_index,
2490                                    MRouteItfFlags.MFIB_ITF_FLAG_FORWARD,
2491                                    proto=FibPathProto.FIB_PATH_NH_PROTO_IP6),
2492                      VppMRoutePath(self.pg3.sw_if_index,
2493                                    MRouteItfFlags.MFIB_ITF_FLAG_FORWARD,
2494                                    proto=FibPathProto.FIB_PATH_NH_PROTO_IP6)],
2495                     table_id=t.table_id).add_vpp_config()
2496                 routes[ii].append({'uni': uni,
2497                                    'multi': multi})
2498
2499         #
2500         # replace the tables a few times
2501         #
2502         for kk in range(3):
2503             # replace each table
2504             for t in self.tables:
2505                 t.replace_begin()
2506
2507             # all the routes are still there
2508             for ii, t in enumerate(self.tables):
2509                 dump = t.dump()
2510                 mdump = t.mdump()
2511                 for r in routes[ii]:
2512                     self.assertTrue(find_route_in_dump(dump, r['uni'], t))
2513                     self.assertTrue(find_mroute_in_dump(mdump, r['multi'], t))
2514
2515             # redownload the even numbered routes
2516             for ii, t in enumerate(self.tables):
2517                 for jj in range(0, N_ROUTES, 2):
2518                     routes[ii][jj]['uni'].add_vpp_config()
2519                     routes[ii][jj]['multi'].add_vpp_config()
2520
2521             # signal each table converged
2522             for t in self.tables:
2523                 t.replace_end()
2524
2525             # we should find the even routes, but not the odd
2526             for ii, t in enumerate(self.tables):
2527                 dump = t.dump()
2528                 mdump = t.mdump()
2529                 for jj in range(0, N_ROUTES, 2):
2530                     self.assertTrue(find_route_in_dump(
2531                         dump, routes[ii][jj]['uni'], t))
2532                     self.assertTrue(find_mroute_in_dump(
2533                         mdump, routes[ii][jj]['multi'], t))
2534                 for jj in range(1, N_ROUTES - 1, 2):
2535                     self.assertFalse(find_route_in_dump(
2536                         dump, routes[ii][jj]['uni'], t))
2537                     self.assertFalse(find_mroute_in_dump(
2538                         mdump, routes[ii][jj]['multi'], t))
2539
2540             # reload all the routes
2541             for ii, t in enumerate(self.tables):
2542                 for r in routes[ii]:
2543                     r['uni'].add_vpp_config()
2544                     r['multi'].add_vpp_config()
2545
2546             # all the routes are still there
2547             for ii, t in enumerate(self.tables):
2548                 dump = t.dump()
2549                 mdump = t.mdump()
2550                 for r in routes[ii]:
2551                     self.assertTrue(find_route_in_dump(dump, r['uni'], t))
2552                     self.assertTrue(find_mroute_in_dump(mdump, r['multi'], t))
2553
2554         #
2555         # finally flush the tables for good measure
2556         #
2557         for t in self.tables:
2558             t.flush()
2559             self.assertEqual(len(t.dump()), 2)
2560             self.assertEqual(len(t.mdump()), 5)
2561
2562
2563 class TestIP6Replace(VppTestCase):
2564     """ IPv4 Interface Address Replace """
2565
2566     @classmethod
2567     def setUpClass(cls):
2568         super(TestIP6Replace, cls).setUpClass()
2569
2570     @classmethod
2571     def tearDownClass(cls):
2572         super(TestIP6Replace, cls).tearDownClass()
2573
2574     def setUp(self):
2575         super(TestIP6Replace, self).setUp()
2576
2577         self.create_pg_interfaces(range(4))
2578
2579         for i in self.pg_interfaces:
2580             i.admin_up()
2581
2582     def tearDown(self):
2583         super(TestIP6Replace, self).tearDown()
2584         for i in self.pg_interfaces:
2585             i.admin_down()
2586
2587     def get_n_pfxs(self, intf):
2588         return len(self.vapi.ip_address_dump(intf.sw_if_index, True))
2589
2590     def test_replace(self):
2591         """ IP interface address replace """
2592
2593         intf_pfxs = [[], [], [], []]
2594
2595         # add prefixes to each of the interfaces
2596         for i in range(len(self.pg_interfaces)):
2597             intf = self.pg_interfaces[i]
2598
2599             # 2001:16:x::1/64
2600             addr = "2001:16:%d::1" % intf.sw_if_index
2601             a = VppIpInterfaceAddress(self, intf, addr, 64).add_vpp_config()
2602             intf_pfxs[i].append(a)
2603
2604             # 2001:16:x::2/64 - a different address in the same subnet as above
2605             addr = "2001:16:%d::2" % intf.sw_if_index
2606             a = VppIpInterfaceAddress(self, intf, addr, 64).add_vpp_config()
2607             intf_pfxs[i].append(a)
2608
2609             # 2001:15:x::2/64 - a different address and subnet
2610             addr = "2001:15:%d::2" % intf.sw_if_index
2611             a = VppIpInterfaceAddress(self, intf, addr, 64).add_vpp_config()
2612             intf_pfxs[i].append(a)
2613
2614         # a dump should n_address in it
2615         for intf in self.pg_interfaces:
2616             self.assertEqual(self.get_n_pfxs(intf), 3)
2617
2618         #
2619         # remove all the address thru a replace
2620         #
2621         self.vapi.sw_interface_address_replace_begin()
2622         self.vapi.sw_interface_address_replace_end()
2623         for intf in self.pg_interfaces:
2624             self.assertEqual(self.get_n_pfxs(intf), 0)
2625
2626         #
2627         # add all the interface addresses back
2628         #
2629         for p in intf_pfxs:
2630             for v in p:
2631                 v.add_vpp_config()
2632         for intf in self.pg_interfaces:
2633             self.assertEqual(self.get_n_pfxs(intf), 3)
2634
2635         #
2636         # replace again, but this time update/re-add the address on the first
2637         # two interfaces
2638         #
2639         self.vapi.sw_interface_address_replace_begin()
2640
2641         for p in intf_pfxs[:2]:
2642             for v in p:
2643                 v.add_vpp_config()
2644
2645         self.vapi.sw_interface_address_replace_end()
2646
2647         # on the first two the address still exist,
2648         # on the other two they do not
2649         for intf in self.pg_interfaces[:2]:
2650             self.assertEqual(self.get_n_pfxs(intf), 3)
2651         for p in intf_pfxs[:2]:
2652             for v in p:
2653                 self.assertTrue(v.query_vpp_config())
2654         for intf in self.pg_interfaces[2:]:
2655             self.assertEqual(self.get_n_pfxs(intf), 0)
2656
2657         #
2658         # add all the interface addresses back on the last two
2659         #
2660         for p in intf_pfxs[2:]:
2661             for v in p:
2662                 v.add_vpp_config()
2663         for intf in self.pg_interfaces:
2664             self.assertEqual(self.get_n_pfxs(intf), 3)
2665
2666         #
2667         # replace again, this time add different prefixes on all the interfaces
2668         #
2669         self.vapi.sw_interface_address_replace_begin()
2670
2671         pfxs = []
2672         for intf in self.pg_interfaces:
2673             # 2001:18:x::1/64
2674             addr = "2001:18:%d::1" % intf.sw_if_index
2675             pfxs.append(VppIpInterfaceAddress(self, intf, addr,
2676                                               64).add_vpp_config())
2677
2678         self.vapi.sw_interface_address_replace_end()
2679
2680         # only .18 should exist on each interface
2681         for intf in self.pg_interfaces:
2682             self.assertEqual(self.get_n_pfxs(intf), 1)
2683         for pfx in pfxs:
2684             self.assertTrue(pfx.query_vpp_config())
2685
2686         #
2687         # remove everything
2688         #
2689         self.vapi.sw_interface_address_replace_begin()
2690         self.vapi.sw_interface_address_replace_end()
2691         for intf in self.pg_interfaces:
2692             self.assertEqual(self.get_n_pfxs(intf), 0)
2693
2694         #
2695         # add prefixes to each interface. post-begin add the prefix from
2696         # interface X onto interface Y. this would normally be an error
2697         # since it would generate a 'duplicate address' warning. but in
2698         # this case, since what is newly downloaded is sane, it's ok
2699         #
2700         for intf in self.pg_interfaces:
2701             # 2001:18:x::1/64
2702             addr = "2001:18:%d::1" % intf.sw_if_index
2703             VppIpInterfaceAddress(self, intf, addr, 64).add_vpp_config()
2704
2705         self.vapi.sw_interface_address_replace_begin()
2706
2707         pfxs = []
2708         for intf in self.pg_interfaces:
2709             # 2001:18:x::1/64
2710             addr = "2001:18:%d::1" % (intf.sw_if_index + 1)
2711             pfxs.append(VppIpInterfaceAddress(self, intf,
2712                                               addr, 64).add_vpp_config())
2713
2714         self.vapi.sw_interface_address_replace_end()
2715
2716         self.logger.info(self.vapi.cli("sh int addr"))
2717
2718         for intf in self.pg_interfaces:
2719             self.assertEqual(self.get_n_pfxs(intf), 1)
2720         for pfx in pfxs:
2721             self.assertTrue(pfx.query_vpp_config())
2722
2723
2724 class TestIP6LinkLocal(VppTestCase):
2725     """ IPv6 Link Local """
2726
2727     @classmethod
2728     def setUpClass(cls):
2729         super(TestIP6LinkLocal, cls).setUpClass()
2730
2731     @classmethod
2732     def tearDownClass(cls):
2733         super(TestIP6LinkLocal, cls).tearDownClass()
2734
2735     def setUp(self):
2736         super(TestIP6LinkLocal, self).setUp()
2737
2738         self.create_pg_interfaces(range(2))
2739
2740         for i in self.pg_interfaces:
2741             i.admin_up()
2742
2743     def tearDown(self):
2744         super(TestIP6LinkLocal, self).tearDown()
2745         for i in self.pg_interfaces:
2746             i.admin_down()
2747
2748     def test_ip6_ll(self):
2749         """ IPv6 Link Local """
2750
2751         #
2752         # two APIs to add a link local address.
2753         #   1 - just like any other prefix
2754         #   2 - with the special set LL API
2755         #
2756
2757         #
2758         # First with the API to set a 'normal' prefix
2759         #
2760         ll1 = "fe80:1::1"
2761         ll2 = "fe80:2::2"
2762         ll3 = "fe80:3::3"
2763
2764         VppIpInterfaceAddress(self, self.pg0, ll1, 128).add_vpp_config()
2765
2766         #
2767         # should be able to ping the ll
2768         #
2769         p_echo_request_1 = (Ether(src=self.pg0.remote_mac,
2770                                   dst=self.pg0.local_mac) /
2771                             IPv6(src=ll2,
2772                                  dst=ll1) /
2773                             ICMPv6EchoRequest())
2774
2775         self.send_and_expect(self.pg0, [p_echo_request_1], self.pg0)
2776
2777         #
2778         # change the link-local on pg0
2779         #
2780         v_ll3 = VppIpInterfaceAddress(self, self.pg0,
2781                                       ll3, 128).add_vpp_config()
2782
2783         p_echo_request_3 = (Ether(src=self.pg0.remote_mac,
2784                                   dst=self.pg0.local_mac) /
2785                             IPv6(src=ll2,
2786                                  dst=ll3) /
2787                             ICMPv6EchoRequest())
2788
2789         self.send_and_expect(self.pg0, [p_echo_request_3], self.pg0)
2790
2791         #
2792         # set a normal v6 prefix on the link
2793         #
2794         self.pg0.config_ip6()
2795
2796         self.send_and_expect(self.pg0, [p_echo_request_3], self.pg0)
2797
2798         # the link-local cannot be removed
2799         with self.vapi.assert_negative_api_retval():
2800             v_ll3.remove_vpp_config()
2801
2802         #
2803         # Use the specific link-local API on pg1
2804         #
2805         VppIp6LinkLocalAddress(self, self.pg1, ll1).add_vpp_config()
2806         self.send_and_expect(self.pg1, [p_echo_request_1], self.pg1)
2807
2808         VppIp6LinkLocalAddress(self, self.pg1, ll3).add_vpp_config()
2809         self.send_and_expect(self.pg1, [p_echo_request_3], self.pg1)
2810
2811
2812 if __name__ == '__main__':
2813     unittest.main(testRunner=VppTestRunner)