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.layers.inet import UDP, IP
  11 from scapy.contrib.mpls import MPLS
  12 from scapy.layers.inet6 import IPv6, ICMPv6ND_NS, ICMPv6ND_RS, \
  13     ICMPv6ND_RA, ICMPv6NDOptMTU, ICMPv6NDOptSrcLLAddr, ICMPv6NDOptPrefixInfo, \
  14     ICMPv6ND_NA, ICMPv6NDOptDstLLAddr, ICMPv6DestUnreach, icmp6types, \
  15     ICMPv6TimeExceeded, ICMPv6EchoRequest, ICMPv6EchoReply, \
  16     IPv6ExtHdrHopByHop, ICMPv6MLReport2, ICMPv6MLDMultAddrRec
  17 from scapy.layers.l2 import Ether, Dot1Q, GRE
  18 from scapy.packet import Raw
  19 from scapy.utils6 import in6_getnsma, in6_getnsmac, in6_ptop, in6_islladdr, \
  20     in6_mactoifaceid
  21 from six import moves
  22
  23 from framework import VppTestCase, VppTestRunner, tag_run_solo
  24 from util import ppp, ip6_normalize, mk_ll_addr
  25 from vpp_papi import VppEnum
  26 from vpp_ip import DpoProto, VppIpPuntPolicer, VppIpPuntRedirect, VppIpPathMtu
  27 from vpp_ip_route import VppIpRoute, VppRoutePath, find_route, VppIpMRoute, \
  28     VppMRoutePath, VppMplsIpBind, \
  29     VppMplsRoute, VppMplsTable, VppIpTable, FibPathType, FibPathProto, \
  30     VppIpInterfaceAddress, find_route_in_dump, find_mroute_in_dump, \
  31     VppIp6LinkLocalAddress, VppIpRouteV2
  32 from vpp_neighbor import find_nbr, VppNeighbor
  33 from vpp_ipip_tun_interface import VppIpIpTunInterface
  34 from vpp_pg_interface import is_ipv6_misc
  35 from vpp_sub_interface import VppSubInterface, VppDot1QSubint
  36 from vpp_policer import VppPolicer, PolicerAction
  37 from ipaddress import IPv6Network, IPv6Address
  38 from vpp_gre_interface import VppGreInterface
  39 from vpp_teib import VppTeib
  40
  41 AF_INET6 = socket.AF_INET6
  42
  43 try:
  44     text_type = unicode
  45 except NameError:
  46     text_type = str
  47
  48 NUM_PKTS = 67
  49
  50
  51 class TestIPv6ND(VppTestCase):
  52     def validate_ra(self, intf, rx, dst_ip=None):
  53         if not dst_ip:
  54             dst_ip = intf.remote_ip6
  55
  56         # unicasted packets must come to the unicast mac
  57         self.assertEqual(rx[Ether].dst, intf.remote_mac)
  58
  59         # and from the router's MAC
  60         self.assertEqual(rx[Ether].src, intf.local_mac)
  61
  62         # the rx'd RA should be addressed to the sender's source
  63         self.assertTrue(rx.haslayer(ICMPv6ND_RA))
  64         self.assertEqual(in6_ptop(rx[IPv6].dst),
  65                          in6_ptop(dst_ip))
  66
  67         # and come from the router's link local
  68         self.assertTrue(in6_islladdr(rx[IPv6].src))
  69         self.assertEqual(in6_ptop(rx[IPv6].src),
  70                          in6_ptop(mk_ll_addr(intf.local_mac)))
  71
  72     def validate_na(self, intf, rx, dst_ip=None, tgt_ip=None):
  73         if not dst_ip:
  74             dst_ip = intf.remote_ip6
  75         if not tgt_ip:
  76             dst_ip = intf.local_ip6
  77
  78         # unicasted packets must come to the unicast mac
  79         self.assertEqual(rx[Ether].dst, intf.remote_mac)
  80
  81         # and from the router's MAC
  82         self.assertEqual(rx[Ether].src, intf.local_mac)
  83
  84         # the rx'd NA should be addressed to the sender's source
  85         self.assertTrue(rx.haslayer(ICMPv6ND_NA))
  86         self.assertEqual(in6_ptop(rx[IPv6].dst),
  87                          in6_ptop(dst_ip))
  88
  89         # and come from the target address
  90         self.assertEqual(
  91             in6_ptop(rx[IPv6].src), in6_ptop(tgt_ip))
  92
  93         # Dest link-layer options should have the router's MAC
  94         dll = rx[ICMPv6NDOptDstLLAddr]
  95         self.assertEqual(dll.lladdr, intf.local_mac)
  96
  97     def validate_ns(self, intf, rx, tgt_ip):
  98         nsma = in6_getnsma(inet_pton(AF_INET6, tgt_ip))
  99         dst_ip = inet_ntop(AF_INET6, nsma)
 100
 101         # NS is broadcast
 102         self.assertEqual(rx[Ether].dst, in6_getnsmac(nsma))
 103
 104         # and from the router's MAC
 105         self.assertEqual(rx[Ether].src, intf.local_mac)
 106
 107         # the rx'd NS should be addressed to an mcast address
 108         # derived from the target address
 109         self.assertEqual(
 110             in6_ptop(rx[IPv6].dst), in6_ptop(dst_ip))
 111
 112         # expect the tgt IP in the NS header
 113         ns = rx[ICMPv6ND_NS]
 114         self.assertEqual(in6_ptop(ns.tgt), in6_ptop(tgt_ip))
 115
 116         # packet is from the router's local address
 117         self.assertEqual(
 118             in6_ptop(rx[IPv6].src), intf.local_ip6)
 119
 120         # Src link-layer options should have the router's MAC
 121         sll = rx[ICMPv6NDOptSrcLLAddr]
 122         self.assertEqual(sll.lladdr, intf.local_mac)
 123
 124     def send_and_expect_ra(self, intf, pkts, remark, dst_ip=None,
 125                            filter_out_fn=is_ipv6_misc):
 126         intf.add_stream(pkts)
 127         self.pg_enable_capture(self.pg_interfaces)
 128         self.pg_start()
 129         rx = intf.get_capture(1, filter_out_fn=filter_out_fn)
 130
 131         self.assertEqual(len(rx), 1)
 132         rx = rx[0]
 133         self.validate_ra(intf, rx, dst_ip)
 134
 135     def send_and_expect_na(self, intf, pkts, remark, dst_ip=None,
 136                            tgt_ip=None,
 137                            filter_out_fn=is_ipv6_misc):
 138         intf.add_stream(pkts)
 139         self.pg_enable_capture(self.pg_interfaces)
 140         self.pg_start()
 141         rx = intf.get_capture(1, filter_out_fn=filter_out_fn)
 142
 143         self.assertEqual(len(rx), 1)
 144         rx = rx[0]
 145         self.validate_na(intf, rx, dst_ip, tgt_ip)
 146
 147     def send_and_expect_ns(self, tx_intf, rx_intf, pkts, tgt_ip,
 148                            filter_out_fn=is_ipv6_misc):
 149         self.vapi.cli("clear trace")
 150         tx_intf.add_stream(pkts)
 151         self.pg_enable_capture(self.pg_interfaces)
 152         self.pg_start()
 153         rx = rx_intf.get_capture(1, filter_out_fn=filter_out_fn)
 154
 155         self.assertEqual(len(rx), 1)
 156         rx = rx[0]
 157         self.validate_ns(rx_intf, rx, tgt_ip)
 158
 159     def verify_ip(self, rx, smac, dmac, sip, dip):
 160         ether = rx[Ether]
 161         self.assertEqual(ether.dst, dmac)
 162         self.assertEqual(ether.src, smac)
 163
 164         ip = rx[IPv6]
 165         self.assertEqual(ip.src, sip)
 166         self.assertEqual(ip.dst, dip)
 167
 168
 169 @tag_run_solo
 170 class TestIPv6(TestIPv6ND):
 171     """ IPv6 Test Case """
 172
 173     @classmethod
 174     def setUpClass(cls):
 175         super(TestIPv6, cls).setUpClass()
 176
 177     @classmethod
 178     def tearDownClass(cls):
 179         super(TestIPv6, cls).tearDownClass()
 180
 181     def setUp(self):
 182         """
 183         Perform test setup before test case.
 184
 185         **Config:**
 186             - create 3 pg interfaces
 187                 - untagged pg0 interface
 188                 - Dot1Q subinterface on pg1
 189                 - Dot1AD subinterface on pg2
 190             - setup interfaces:
 191                 - put it into UP state
 192                 - set IPv6 addresses
 193                 - resolve neighbor address using NDP
 194             - configure 200 fib entries
 195
 196         :ivar list interfaces: pg interfaces and subinterfaces.
 197         :ivar dict flows: IPv4 packet flows in test.
 198
 199         *TODO:* Create AD sub interface
 200         """
 201         super(TestIPv6, self).setUp()
 202
 203         # create 3 pg interfaces
 204         self.create_pg_interfaces(range(3))
 205
 206         # create 2 subinterfaces for p1 and pg2
 207         self.sub_interfaces = [
 208             VppDot1QSubint(self, self.pg1, 100),
 209             VppDot1QSubint(self, self.pg2, 200)
 210             # TODO: VppDot1ADSubint(self, self.pg2, 200, 300, 400)
 211         ]
 212
 213         # packet flows mapping pg0 -> pg1.sub, pg2.sub, etc.
 214         self.flows = dict()
 215         self.flows[self.pg0] = [self.pg1.sub_if, self.pg2.sub_if]
 216         self.flows[self.pg1.sub_if] = [self.pg0, self.pg2.sub_if]
 217         self.flows[self.pg2.sub_if] = [self.pg0, self.pg1.sub_if]
 218
 219         # packet sizes
 220         self.pg_if_packet_sizes = [64, 1500, 9020]
 221
 222         self.interfaces = list(self.pg_interfaces)
 223         self.interfaces.extend(self.sub_interfaces)
 224
 225         # setup all interfaces
 226         for i in self.interfaces:
 227             i.admin_up()
 228             i.config_ip6()
 229             i.resolve_ndp()
 230
 231     def tearDown(self):
 232         """Run standard test teardown and log ``show ip6 neighbors``."""
 233         for i in self.interfaces:
 234             i.unconfig_ip6()
 235             i.admin_down()
 236         for i in self.sub_interfaces:
 237             i.remove_vpp_config()
 238
 239         super(TestIPv6, self).tearDown()
 240         if not self.vpp_dead:
 241             self.logger.info(self.vapi.cli("show ip6 neighbors"))
 242             # info(self.vapi.cli("show ip6 fib"))  # many entries
 243
 244     def modify_packet(self, src_if, packet_size, pkt):
 245         """Add load, set destination IP and extend packet to required packet
 246         size for defined interface.
 247
 248         :param VppInterface src_if: Interface to create packet for.
 249         :param int packet_size: Required packet size.
 250         :param Scapy pkt: Packet to be modified.
 251         """
 252         dst_if_idx = int(packet_size / 10 % 2)
 253         dst_if = self.flows[src_if][dst_if_idx]
 254         info = self.create_packet_info(src_if, dst_if)
 255         payload = self.info_to_payload(info)
 256         p = pkt / Raw(payload)
 257         p[IPv6].dst = dst_if.remote_ip6
 258         info.data = p.copy()
 259         if isinstance(src_if, VppSubInterface):
 260             p = src_if.add_dot1_layer(p)
 261         self.extend_packet(p, packet_size)
 262
 263         return p
 264
 265     def create_stream(self, src_if):
 266         """Create input packet stream for defined interface.
 267
 268         :param VppInterface src_if: Interface to create packet stream for.
 269         """
 270         hdr_ext = 4 if isinstance(src_if, VppSubInterface) else 0
 271         pkt_tmpl = (Ether(dst=src_if.local_mac, src=src_if.remote_mac) /
 272                     IPv6(src=src_if.remote_ip6) /
 273                     inet6.UDP(sport=1234, dport=1234))
 274
 275         pkts = [self.modify_packet(src_if, i, pkt_tmpl)
 276                 for i in moves.range(self.pg_if_packet_sizes[0],
 277                                      self.pg_if_packet_sizes[1], 10)]
 278         pkts_b = [self.modify_packet(src_if, i, pkt_tmpl)
 279                   for i in moves.range(self.pg_if_packet_sizes[1] + hdr_ext,
 280                                        self.pg_if_packet_sizes[2] + hdr_ext,
 281                                        50)]
 282         pkts.extend(pkts_b)
 283
 284         return pkts
 285
 286     def verify_capture(self, dst_if, capture):
 287         """Verify captured input packet stream for defined interface.
 288
 289         :param VppInterface dst_if: Interface to verify captured packet stream
 290                                     for.
 291         :param list capture: Captured packet stream.
 292         """
 293         self.logger.info("Verifying capture on interface %s" % dst_if.name)
 294         last_info = dict()
 295         for i in self.interfaces:
 296             last_info[i.sw_if_index] = None
 297         is_sub_if = False
 298         dst_sw_if_index = dst_if.sw_if_index
 299         if hasattr(dst_if, 'parent'):
 300             is_sub_if = True
 301         for packet in capture:
 302             if is_sub_if:
 303                 # Check VLAN tags and Ethernet header
 304                 packet = dst_if.remove_dot1_layer(packet)
 305             self.assertTrue(Dot1Q not in packet)
 306             try:
 307                 ip = packet[IPv6]
 308                 udp = packet[inet6.UDP]
 309                 payload_info = self.payload_to_info(packet[Raw])
 310                 packet_index = payload_info.index
 311                 self.assertEqual(payload_info.dst, dst_sw_if_index)
 312                 self.logger.debug(
 313                     "Got packet on port %s: src=%u (id=%u)" %
 314                     (dst_if.name, payload_info.src, packet_index))
 315                 next_info = self.get_next_packet_info_for_interface2(
 316                     payload_info.src, dst_sw_if_index,
 317                     last_info[payload_info.src])
 318                 last_info[payload_info.src] = next_info
 319                 self.assertTrue(next_info is not None)
 320                 self.assertEqual(packet_index, next_info.index)
 321                 saved_packet = next_info.data
 322                 # Check standard fields
 323                 self.assertEqual(
 324                     ip.src, saved_packet[IPv6].src)
 325                 self.assertEqual(
 326                     ip.dst, saved_packet[IPv6].dst)
 327                 self.assertEqual(
 328                     udp.sport, saved_packet[inet6.UDP].sport)
 329                 self.assertEqual(
 330                     udp.dport, saved_packet[inet6.UDP].dport)
 331             except:
 332                 self.logger.error(ppp("Unexpected or invalid packet:", packet))
 333                 raise
 334         for i in self.interfaces:
 335             remaining_packet = self.get_next_packet_info_for_interface2(
 336                 i.sw_if_index, dst_sw_if_index, last_info[i.sw_if_index])
 337             self.assertTrue(remaining_packet is None,
 338                             "Interface %s: Packet expected from interface %s "
 339                             "didn't arrive" % (dst_if.name, i.name))
 340
 341     def test_next_header_anomaly(self):
 342         """ IPv6 next header anomaly test
 343
 344         Test scenario:
 345             - ipv6 next header field = Fragment Header (44)
 346             - next header is ICMPv6 Echo Request
 347             - wait for reassembly
 348         """
 349         pkt = (Ether(src=self.pg0.local_mac, dst=self.pg0.remote_mac) /
 350                IPv6(src=self.pg0.remote_ip6, dst=self.pg0.local_ip6, nh=44) /
 351                ICMPv6EchoRequest())
 352
 353         self.pg0.add_stream(pkt)
 354         self.pg_start()
 355
 356         # wait for reassembly
 357         self.sleep(10)
 358
 359     def test_fib(self):
 360         """ IPv6 FIB test
 361
 362         Test scenario:
 363             - Create IPv6 stream for pg0 interface
 364             - Create IPv6 tagged streams for pg1's and pg2's subinterface.
 365             - Send and verify received packets on each interface.
 366         """
 367
 368         pkts = self.create_stream(self.pg0)
 369         self.pg0.add_stream(pkts)
 370
 371         for i in self.sub_interfaces:
 372             pkts = self.create_stream(i)
 373             i.parent.add_stream(pkts)
 374
 375         self.pg_enable_capture(self.pg_interfaces)
 376         self.pg_start()
 377
 378         pkts = self.pg0.get_capture()
 379         self.verify_capture(self.pg0, pkts)
 380
 381         for i in self.sub_interfaces:
 382             pkts = i.parent.get_capture()
 383             self.verify_capture(i, pkts)
 384
 385     def test_ns(self):
 386         """ IPv6 Neighbour Solicitation Exceptions
 387
 388         Test scenario:
 389            - Send an NS Sourced from an address not covered by the link sub-net
 390            - Send an NS to an mcast address the router has not joined
 391            - Send NS for a target address the router does not onn.
 392         """
 393
 394         #
 395         # An NS from a non link source address
 396         #
 397         nsma = in6_getnsma(inet_pton(AF_INET6, self.pg0.local_ip6))
 398         d = inet_ntop(AF_INET6, nsma)
 399
 400         p = (Ether(dst=in6_getnsmac(nsma)) /
 401              IPv6(dst=d, src="2002::2") /
 402              ICMPv6ND_NS(tgt=self.pg0.local_ip6) /
 403              ICMPv6NDOptSrcLLAddr(
 404                  lladdr=self.pg0.remote_mac))
 405         pkts = [p]
 406
 407         self.send_and_assert_no_replies(
 408             self.pg0, pkts,
 409             "No response to NS source by address not on sub-net")
 410
 411         #
 412         # An NS for sent to a solicited mcast group the router is
 413         # not a member of FAILS
 414         #
 415         if 0:
 416             nsma = in6_getnsma(inet_pton(AF_INET6, "fd::ffff"))
 417             d = inet_ntop(AF_INET6, nsma)
 418
 419             p = (Ether(dst=in6_getnsmac(nsma)) /
 420                  IPv6(dst=d, src=self.pg0.remote_ip6) /
 421                  ICMPv6ND_NS(tgt=self.pg0.local_ip6) /
 422                  ICMPv6NDOptSrcLLAddr(
 423                      lladdr=self.pg0.remote_mac))
 424             pkts = [p]
 425
 426             self.send_and_assert_no_replies(
 427                 self.pg0, pkts,
 428                 "No response to NS sent to unjoined mcast address")
 429
 430         #
 431         # An NS whose target address is one the router does not own
 432         #
 433         nsma = in6_getnsma(inet_pton(AF_INET6, self.pg0.local_ip6))
 434         d = inet_ntop(AF_INET6, nsma)
 435
 436         p = (Ether(dst=in6_getnsmac(nsma)) /
 437              IPv6(dst=d, src=self.pg0.remote_ip6) /
 438              ICMPv6ND_NS(tgt="fd::ffff") /
 439              ICMPv6NDOptSrcLLAddr(
 440                  lladdr=self.pg0.remote_mac))
 441         pkts = [p]
 442
 443         self.send_and_assert_no_replies(self.pg0, pkts,
 444                                         "No response to NS for unknown target")
 445
 446         #
 447         # A neighbor entry that has no associated FIB-entry
 448         #
 449         self.pg0.generate_remote_hosts(4)
 450         nd_entry = VppNeighbor(self,
 451                                self.pg0.sw_if_index,
 452                                self.pg0.remote_hosts[2].mac,
 453                                self.pg0.remote_hosts[2].ip6,
 454                                is_no_fib_entry=1)
 455         nd_entry.add_vpp_config()
 456
 457         #
 458         # check we have the neighbor, but no route
 459         #
 460         self.assertTrue(find_nbr(self,
 461                                  self.pg0.sw_if_index,
 462                                  self.pg0._remote_hosts[2].ip6))
 463         self.assertFalse(find_route(self,
 464                                     self.pg0._remote_hosts[2].ip6,
 465                                     128))
 466
 467         #
 468         # send an NS from a link local address to the interface's global
 469         # address
 470         #
 471         p = (Ether(dst=in6_getnsmac(nsma), src=self.pg0.remote_mac) /
 472              IPv6(
 473                  dst=d, src=self.pg0._remote_hosts[2].ip6_ll) /
 474              ICMPv6ND_NS(tgt=self.pg0.local_ip6) /
 475              ICMPv6NDOptSrcLLAddr(
 476                  lladdr=self.pg0.remote_mac))
 477
 478         self.send_and_expect_na(self.pg0, p,
 479                                 "NS from link-local",
 480                                 dst_ip=self.pg0._remote_hosts[2].ip6_ll,
 481                                 tgt_ip=self.pg0.local_ip6)
 482
 483         #
 484         # we should have learned an ND entry for the peer's link-local
 485         # but not inserted a route to it in the FIB
 486         #
 487         self.assertTrue(find_nbr(self,
 488                                  self.pg0.sw_if_index,
 489                                  self.pg0._remote_hosts[2].ip6_ll))
 490         self.assertFalse(find_route(self,
 491                                     self.pg0._remote_hosts[2].ip6_ll,
 492                                     128))
 493
 494         #
 495         # An NS to the router's own Link-local
 496         #
 497         p = (Ether(dst=in6_getnsmac(nsma), src=self.pg0.remote_mac) /
 498              IPv6(
 499                  dst=d, src=self.pg0._remote_hosts[3].ip6_ll) /
 500              ICMPv6ND_NS(tgt=self.pg0.local_ip6_ll) /
 501              ICMPv6NDOptSrcLLAddr(
 502                  lladdr=self.pg0.remote_mac))
 503
 504         self.send_and_expect_na(self.pg0, p,
 505                                 "NS to/from link-local",
 506                                 dst_ip=self.pg0._remote_hosts[3].ip6_ll,
 507                                 tgt_ip=self.pg0.local_ip6_ll)
 508
 509         #
 510         # do not respond to a NS for the peer's address
 511         #
 512         p = (Ether(dst=in6_getnsmac(nsma), src=self.pg0.remote_mac) /
 513              IPv6(dst=d,
 514                   src=self.pg0._remote_hosts[3].ip6_ll) /
 515              ICMPv6ND_NS(tgt=self.pg0._remote_hosts[3].ip6_ll) /
 516              ICMPv6NDOptSrcLLAddr(
 517                  lladdr=self.pg0.remote_mac))
 518
 519         self.send_and_assert_no_replies(self.pg0, p)
 520
 521         #
 522         # we should have learned an ND entry for the peer's link-local
 523         # but not inserted a route to it in the FIB
 524         #
 525         self.assertTrue(find_nbr(self,
 526                                  self.pg0.sw_if_index,
 527                                  self.pg0._remote_hosts[3].ip6_ll))
 528         self.assertFalse(find_route(self,
 529                                     self.pg0._remote_hosts[3].ip6_ll,
 530                                     128))
 531
 532     def test_ns_duplicates(self):
 533         """ ND Duplicates"""
 534
 535         #
 536         # Generate some hosts on the LAN
 537         #
 538         self.pg1.generate_remote_hosts(3)
 539
 540         #
 541         # Add host 1 on pg1 and pg2
 542         #
 543         ns_pg1 = VppNeighbor(self,
 544                              self.pg1.sw_if_index,
 545                              self.pg1.remote_hosts[1].mac,
 546                              self.pg1.remote_hosts[1].ip6)
 547         ns_pg1.add_vpp_config()
 548         ns_pg2 = VppNeighbor(self,
 549                              self.pg2.sw_if_index,
 550                              self.pg2.remote_mac,
 551                              self.pg1.remote_hosts[1].ip6)
 552         ns_pg2.add_vpp_config()
 553
 554         #
 555         # IP packet destined for pg1 remote host arrives on pg1 again.
 556         #
 557         p = (Ether(dst=self.pg0.local_mac,
 558                    src=self.pg0.remote_mac) /
 559              IPv6(src=self.pg0.remote_ip6,
 560                   dst=self.pg1.remote_hosts[1].ip6) /
 561              inet6.UDP(sport=1234, dport=1234) /
 562              Raw())
 563
 564         self.pg0.add_stream(p)
 565         self.pg_enable_capture(self.pg_interfaces)
 566         self.pg_start()
 567
 568         rx1 = self.pg1.get_capture(1)
 569
 570         self.verify_ip(rx1[0],
 571                        self.pg1.local_mac,
 572                        self.pg1.remote_hosts[1].mac,
 573                        self.pg0.remote_ip6,
 574                        self.pg1.remote_hosts[1].ip6)
 575
 576         #
 577         # remove the duplicate on pg1
 578         # packet stream should generate NSs out of pg1
 579         #
 580         ns_pg1.remove_vpp_config()
 581
 582         self.send_and_expect_ns(self.pg0, self.pg1,
 583                                 p, self.pg1.remote_hosts[1].ip6)
 584
 585         #
 586         # Add it back
 587         #
 588         ns_pg1.add_vpp_config()
 589
 590         self.pg0.add_stream(p)
 591         self.pg_enable_capture(self.pg_interfaces)
 592         self.pg_start()
 593
 594         rx1 = self.pg1.get_capture(1)
 595
 596         self.verify_ip(rx1[0],
 597                        self.pg1.local_mac,
 598                        self.pg1.remote_hosts[1].mac,
 599                        self.pg0.remote_ip6,
 600                        self.pg1.remote_hosts[1].ip6)
 601
 602     def validate_ra(self, intf, rx, dst_ip=None, src_ip=None,
 603                     mtu=9000, pi_opt=None):
 604         if not dst_ip:
 605             dst_ip = intf.remote_ip6
 606         if not src_ip:
 607             src_ip = mk_ll_addr(intf.local_mac)
 608
 609         # unicasted packets must come to the unicast mac
 610         self.assertEqual(rx[Ether].dst, intf.remote_mac)
 611
 612         # and from the router's MAC
 613         self.assertEqual(rx[Ether].src, intf.local_mac)
 614
 615         # the rx'd RA should be addressed to the sender's source
 616         self.assertTrue(rx.haslayer(ICMPv6ND_RA))
 617         self.assertEqual(in6_ptop(rx[IPv6].dst),
 618                          in6_ptop(dst_ip))
 619
 620         # and come from the router's link local
 621         self.assertTrue(in6_islladdr(rx[IPv6].src))
 622         self.assertEqual(in6_ptop(rx[IPv6].src), in6_ptop(src_ip))
 623
 624         # it should contain the links MTU
 625         ra = rx[ICMPv6ND_RA]
 626         self.assertEqual(ra[ICMPv6NDOptMTU].mtu, mtu)
 627
 628         # it should contain the source's link layer address option
 629         sll = ra[ICMPv6NDOptSrcLLAddr]
 630         self.assertEqual(sll.lladdr, intf.local_mac)
 631
 632         if not pi_opt:
 633             # the RA should not contain prefix information
 634             self.assertFalse(ra.haslayer(
 635                 ICMPv6NDOptPrefixInfo))
 636         else:
 637             raos = rx.getlayer(ICMPv6NDOptPrefixInfo, 1)
 638
 639             # the options are nested in the scapy packet in way that i cannot
 640             # decipher how to decode. this 1st layer of option always returns
 641             # nested classes, so a direct obj1=obj2 comparison always fails.
 642             # however, the getlayer(.., 2) does give one instance.
 643             # so we cheat here and construct a new opt instance for comparison
 644             rd = ICMPv6NDOptPrefixInfo(
 645                 prefixlen=raos.prefixlen,
 646                 prefix=raos.prefix,
 647                 L=raos.L,
 648                 A=raos.A)
 649             if type(pi_opt) is list:
 650                 for ii in range(len(pi_opt)):
 651                     self.assertEqual(pi_opt[ii], rd)
 652                     rd = rx.getlayer(
 653                         ICMPv6NDOptPrefixInfo, ii + 2)
 654             else:
 655                 self.assertEqual(pi_opt, raos, 'Expected: %s, received: %s'
 656                                  % (pi_opt.show(dump=True),
 657                                     raos.show(dump=True)))
 658
 659     def send_and_expect_ra(self, intf, pkts, remark, dst_ip=None,
 660                            filter_out_fn=is_ipv6_misc,
 661                            opt=None,
 662                            src_ip=None):
 663         self.vapi.cli("clear trace")
 664         intf.add_stream(pkts)
 665         self.pg_enable_capture(self.pg_interfaces)
 666         self.pg_start()
 667         rx = intf.get_capture(1, filter_out_fn=filter_out_fn)
 668
 669         self.assertEqual(len(rx), 1)
 670         rx = rx[0]
 671         self.validate_ra(intf, rx, dst_ip, src_ip=src_ip, pi_opt=opt)
 672
 673     def test_rs(self):
 674         """ IPv6 Router Solicitation Exceptions
 675
 676         Test scenario:
 677         """
 678
 679         #
 680         # Before we begin change the IPv6 RA responses to use the unicast
 681         # address - that way we will not confuse them with the periodic
 682         # RAs which go to the mcast address
 683         # Sit and wait for the first periodic RA.
 684         #
 685         # TODO
 686         #
 687         self.pg0.ip6_ra_config(send_unicast=1)
 688
 689         #
 690         # An RS from a link source address
 691         #  - expect an RA in return
 692         #
 693         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
 694              IPv6(dst=self.pg0.local_ip6, src=self.pg0.remote_ip6) /
 695              ICMPv6ND_RS())
 696         pkts = [p]
 697         self.send_and_expect_ra(self.pg0, pkts, "Genuine RS")
 698
 699         #
 700         # For the next RS sent the RA should be rate limited
 701         #
 702         self.send_and_assert_no_replies(self.pg0, pkts, "RA rate limited")
 703
 704         #
 705         # When we reconfigure the IPv6 RA config,
 706         # we reset the RA rate limiting,
 707         # so we need to do this before each test below so as not to drop
 708         # packets for rate limiting reasons. Test this works here.
 709         #
 710         self.pg0.ip6_ra_config(send_unicast=1)
 711         self.send_and_expect_ra(self.pg0, pkts, "Rate limit reset RS")
 712
 713         #
 714         # An RS sent from a non-link local source
 715         #
 716         self.pg0.ip6_ra_config(send_unicast=1)
 717         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
 718              IPv6(dst=self.pg0.local_ip6,
 719                   src="2002::ffff") /
 720              ICMPv6ND_RS())
 721         pkts = [p]
 722         self.send_and_assert_no_replies(self.pg0, pkts,
 723                                         "RS from non-link source")
 724
 725         #
 726         # Source an RS from a link local address
 727         #
 728         self.pg0.ip6_ra_config(send_unicast=1)
 729         ll = mk_ll_addr(self.pg0.remote_mac)
 730         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
 731              IPv6(dst=self.pg0.local_ip6, src=ll) /
 732              ICMPv6ND_RS())
 733         pkts = [p]
 734         self.send_and_expect_ra(self.pg0, pkts,
 735                                 "RS sourced from link-local",
 736                                 dst_ip=ll)
 737
 738         #
 739         # Source an RS from a link local address
 740         # Ensure suppress also applies to solicited RS
 741         #
 742         self.pg0.ip6_ra_config(send_unicast=1, suppress=1)
 743         ll = mk_ll_addr(self.pg0.remote_mac)
 744         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
 745              IPv6(dst=self.pg0.local_ip6, src=ll) /
 746              ICMPv6ND_RS())
 747         pkts = [p]
 748         self.send_and_assert_no_replies(self.pg0, pkts,
 749                                         "Suppressed RS from link-local")
 750
 751         #
 752         # Send the RS multicast
 753         #
 754         self.pg0.ip6_ra_config(no=1, suppress=1)  # Reset suppress flag to zero
 755         self.pg0.ip6_ra_config(send_unicast=1)
 756         dmac = in6_getnsmac(inet_pton(AF_INET6, "ff02::2"))
 757         ll = mk_ll_addr(self.pg0.remote_mac)
 758         p = (Ether(dst=dmac, src=self.pg0.remote_mac) /
 759              IPv6(dst="ff02::2", src=ll) /
 760              ICMPv6ND_RS())
 761         pkts = [p]
 762         self.send_and_expect_ra(self.pg0, pkts,
 763                                 "RS sourced from link-local",
 764                                 dst_ip=ll)
 765
 766         #
 767         # Source from the unspecified address ::. This happens when the RS
 768         # is sent before the host has a configured address/sub-net,
 769         # i.e. auto-config. Since the sender has no IP address, the reply
 770         # comes back mcast - so the capture needs to not filter this.
 771         # If we happen to pick up the periodic RA at this point then so be it,
 772         # it's not an error.
 773         #
 774         self.pg0.ip6_ra_config(send_unicast=1, suppress=0)
 775         p = (Ether(dst=dmac, src=self.pg0.remote_mac) /
 776              IPv6(dst="ff02::2", src="::") /
 777              ICMPv6ND_RS())
 778         pkts = [p]
 779         self.send_and_expect_ra(self.pg0, pkts,
 780                                 "RS sourced from unspecified",
 781                                 dst_ip="ff02::1",
 782                                 filter_out_fn=None)
 783
 784         #
 785         # Configure The RA to announce the links prefix
 786         #
 787         self.pg0.ip6_ra_prefix('%s/%s' % (self.pg0.local_ip6,
 788                                self.pg0.local_ip6_prefix_len))
 789
 790         #
 791         # RAs should now contain the prefix information option
 792         #
 793         opt = ICMPv6NDOptPrefixInfo(
 794             prefixlen=self.pg0.local_ip6_prefix_len,
 795             prefix=self.pg0.local_ip6,
 796             L=1,
 797             A=1)
 798
 799         self.pg0.ip6_ra_config(send_unicast=1)
 800         ll = mk_ll_addr(self.pg0.remote_mac)
 801         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
 802              IPv6(dst=self.pg0.local_ip6, src=ll) /
 803              ICMPv6ND_RS())
 804         self.send_and_expect_ra(self.pg0, p,
 805                                 "RA with prefix-info",
 806                                 dst_ip=ll,
 807                                 opt=opt)
 808
 809         #
 810         # Change the prefix info to not off-link
 811         #  L-flag is clear
 812         #
 813         self.pg0.ip6_ra_prefix('%s/%s' % (self.pg0.local_ip6,
 814                                self.pg0.local_ip6_prefix_len),
 815                                off_link=1)
 816
 817         opt = ICMPv6NDOptPrefixInfo(
 818             prefixlen=self.pg0.local_ip6_prefix_len,
 819             prefix=self.pg0.local_ip6,
 820             L=0,
 821             A=1)
 822
 823         self.pg0.ip6_ra_config(send_unicast=1)
 824         self.send_and_expect_ra(self.pg0, p,
 825                                 "RA with Prefix info with L-flag=0",
 826                                 dst_ip=ll,
 827                                 opt=opt)
 828
 829         #
 830         # Change the prefix info to not off-link, no-autoconfig
 831         #  L and A flag are clear in the advert
 832         #
 833         self.pg0.ip6_ra_prefix('%s/%s' % (self.pg0.local_ip6,
 834                                self.pg0.local_ip6_prefix_len),
 835                                off_link=1,
 836                                no_autoconfig=1)
 837
 838         opt = ICMPv6NDOptPrefixInfo(
 839             prefixlen=self.pg0.local_ip6_prefix_len,
 840             prefix=self.pg0.local_ip6,
 841             L=0,
 842             A=0)
 843
 844         self.pg0.ip6_ra_config(send_unicast=1)
 845         self.send_and_expect_ra(self.pg0, p,
 846                                 "RA with Prefix info with A & L-flag=0",
 847                                 dst_ip=ll,
 848                                 opt=opt)
 849
 850         #
 851         # Change the flag settings back to the defaults
 852         #  L and A flag are set in the advert
 853         #
 854         self.pg0.ip6_ra_prefix('%s/%s' % (self.pg0.local_ip6,
 855                                self.pg0.local_ip6_prefix_len))
 856
 857         opt = ICMPv6NDOptPrefixInfo(
 858             prefixlen=self.pg0.local_ip6_prefix_len,
 859             prefix=self.pg0.local_ip6,
 860             L=1,
 861             A=1)
 862
 863         self.pg0.ip6_ra_config(send_unicast=1)
 864         self.send_and_expect_ra(self.pg0, p,
 865                                 "RA with Prefix info",
 866                                 dst_ip=ll,
 867                                 opt=opt)
 868
 869         #
 870         # Change the prefix info to not off-link, no-autoconfig
 871         #  L and A flag are clear in the advert
 872         #
 873         self.pg0.ip6_ra_prefix('%s/%s' % (self.pg0.local_ip6,
 874                                self.pg0.local_ip6_prefix_len),
 875                                off_link=1,
 876                                no_autoconfig=1)
 877
 878         opt = ICMPv6NDOptPrefixInfo(
 879             prefixlen=self.pg0.local_ip6_prefix_len,
 880             prefix=self.pg0.local_ip6,
 881             L=0,
 882             A=0)
 883
 884         self.pg0.ip6_ra_config(send_unicast=1)
 885         self.send_and_expect_ra(self.pg0, p,
 886                                 "RA with Prefix info with A & L-flag=0",
 887                                 dst_ip=ll,
 888                                 opt=opt)
 889
 890         #
 891         # Use the reset to defaults option to revert to defaults
 892         #  L and A flag are clear in the advert
 893         #
 894         self.pg0.ip6_ra_prefix('%s/%s' % (self.pg0.local_ip6,
 895                                self.pg0.local_ip6_prefix_len),
 896                                use_default=1)
 897
 898         opt = ICMPv6NDOptPrefixInfo(
 899             prefixlen=self.pg0.local_ip6_prefix_len,
 900             prefix=self.pg0.local_ip6,
 901             L=1,
 902             A=1)
 903
 904         self.pg0.ip6_ra_config(send_unicast=1)
 905         self.send_and_expect_ra(self.pg0, p,
 906                                 "RA with Prefix reverted to defaults",
 907                                 dst_ip=ll,
 908                                 opt=opt)
 909
 910         #
 911         # Advertise Another prefix. With no L-flag/A-flag
 912         #
 913         self.pg0.ip6_ra_prefix('%s/%s' % (self.pg1.local_ip6,
 914                                self.pg1.local_ip6_prefix_len),
 915                                off_link=1,
 916                                no_autoconfig=1)
 917
 918         opt = [ICMPv6NDOptPrefixInfo(
 919             prefixlen=self.pg0.local_ip6_prefix_len,
 920             prefix=self.pg0.local_ip6,
 921             L=1,
 922             A=1),
 923             ICMPv6NDOptPrefixInfo(
 924                 prefixlen=self.pg1.local_ip6_prefix_len,
 925                 prefix=self.pg1.local_ip6,
 926                 L=0,
 927                 A=0)]
 928
 929         self.pg0.ip6_ra_config(send_unicast=1)
 930         ll = mk_ll_addr(self.pg0.remote_mac)
 931         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
 932              IPv6(dst=self.pg0.local_ip6, src=ll) /
 933              ICMPv6ND_RS())
 934         self.send_and_expect_ra(self.pg0, p,
 935                                 "RA with multiple Prefix infos",
 936                                 dst_ip=ll,
 937                                 opt=opt)
 938
 939         #
 940         # Remove the first prefix-info - expect the second is still in the
 941         # advert
 942         #
 943         self.pg0.ip6_ra_prefix('%s/%s' % (self.pg0.local_ip6,
 944                                self.pg0.local_ip6_prefix_len),
 945                                is_no=1)
 946
 947         opt = ICMPv6NDOptPrefixInfo(
 948             prefixlen=self.pg1.local_ip6_prefix_len,
 949             prefix=self.pg1.local_ip6,
 950             L=0,
 951             A=0)
 952
 953         self.pg0.ip6_ra_config(send_unicast=1)
 954         self.send_and_expect_ra(self.pg0, p,
 955                                 "RA with Prefix reverted to defaults",
 956                                 dst_ip=ll,
 957                                 opt=opt)
 958
 959         #
 960         # Remove the second prefix-info - expect no prefix-info in the adverts
 961         #
 962         self.pg0.ip6_ra_prefix('%s/%s' % (self.pg1.local_ip6,
 963                                self.pg1.local_ip6_prefix_len),
 964                                is_no=1)
 965
 966         #
 967         # change the link's link local, so we know that works too.
 968         #
 969         self.vapi.sw_interface_ip6_set_link_local_address(
 970             sw_if_index=self.pg0.sw_if_index,
 971             ip="fe80::88")
 972
 973         self.pg0.ip6_ra_config(send_unicast=1)
 974         self.send_and_expect_ra(self.pg0, p,
 975                                 "RA with Prefix reverted to defaults",
 976                                 dst_ip=ll,
 977                                 src_ip="fe80::88")
 978
 979         #
 980         # Reset the periodic advertisements back to default values
 981         #
 982         self.pg0.ip6_ra_config(no=1, suppress=1, send_unicast=0)
 983
 984     def test_mld(self):
 985         """ MLD Report """
 986         #
 987         # test one MLD is sent after applying an IPv6 Address on an interface
 988         #
 989         self.pg_enable_capture(self.pg_interfaces)
 990         self.pg_start()
 991
 992         subitf = VppDot1QSubint(self, self.pg1, 99)
 993
 994         subitf.admin_up()
 995         subitf.config_ip6()
 996
 997         rxs = self.pg1._get_capture(timeout=4, filter_out_fn=None)
 998
 999         #
1000         # hunt for the MLD on vlan 99
1001         #
1002         for rx in rxs:
1003             # make sure ipv6 packets with hop by hop options have
1004             # correct checksums
1005             self.assert_packet_checksums_valid(rx)
1006             if rx.haslayer(IPv6ExtHdrHopByHop) and \
1007                rx.haslayer(Dot1Q) and \
1008                rx[Dot1Q].vlan == 99:
1009                 mld = rx[ICMPv6MLReport2]
1010
1011         self.assertEqual(mld.records_number, 4)
1012
1013
1014 class TestIPv6RouteLookup(VppTestCase):
1015     """ IPv6 Route Lookup Test Case """
1016     routes = []
1017
1018     def route_lookup(self, prefix, exact):
1019         return self.vapi.api(self.vapi.papi.ip_route_lookup,
1020                              {
1021                                  'table_id': 0,
1022                                  'exact': exact,
1023                                  'prefix': prefix,
1024                              })
1025
1026     @classmethod
1027     def setUpClass(cls):
1028         super(TestIPv6RouteLookup, cls).setUpClass()
1029
1030     @classmethod
1031     def tearDownClass(cls):
1032         super(TestIPv6RouteLookup, cls).tearDownClass()
1033
1034     def setUp(self):
1035         super(TestIPv6RouteLookup, self).setUp()
1036
1037         drop_nh = VppRoutePath("::1", 0xffffffff,
1038                                type=FibPathType.FIB_PATH_TYPE_DROP)
1039
1040         # Add 3 routes
1041         r = VppIpRoute(self, "2001:1111::", 32, [drop_nh])
1042         r.add_vpp_config()
1043         self.routes.append(r)
1044
1045         r = VppIpRoute(self, "2001:1111:2222::", 48, [drop_nh])
1046         r.add_vpp_config()
1047         self.routes.append(r)
1048
1049         r = VppIpRoute(self, "2001:1111:2222::1", 128, [drop_nh])
1050         r.add_vpp_config()
1051         self.routes.append(r)
1052
1053     def tearDown(self):
1054         # Remove the routes we added
1055         for r in self.routes:
1056             r.remove_vpp_config()
1057
1058         super(TestIPv6RouteLookup, self).tearDown()
1059
1060     def test_exact_match(self):
1061         # Verify we find the host route
1062         prefix = "2001:1111:2222::1/128"
1063         result = self.route_lookup(prefix, True)
1064         assert (prefix == str(result.route.prefix))
1065
1066         # Verify we find a middle prefix route
1067         prefix = "2001:1111:2222::/48"
1068         result = self.route_lookup(prefix, True)
1069         assert (prefix == str(result.route.prefix))
1070
1071         # Verify we do not find an available LPM.
1072         with self.vapi.assert_negative_api_retval():
1073             self.route_lookup("2001::2/128", True)
1074
1075     def test_longest_prefix_match(self):
1076         # verify we find lpm
1077         lpm_prefix = "2001:1111:2222::/48"
1078         result = self.route_lookup("2001:1111:2222::2/128", False)
1079         assert (lpm_prefix == str(result.route.prefix))
1080
1081         # Verify we find the exact when not requested
1082         result = self.route_lookup(lpm_prefix, False)
1083         assert (lpm_prefix == str(result.route.prefix))
1084
1085         # Can't seem to delete the default route so no negative LPM test.
1086
1087
1088 class TestIPv6IfAddrRoute(VppTestCase):
1089     """ IPv6 Interface Addr Route Test Case """
1090
1091     @classmethod
1092     def setUpClass(cls):
1093         super(TestIPv6IfAddrRoute, cls).setUpClass()
1094
1095     @classmethod
1096     def tearDownClass(cls):
1097         super(TestIPv6IfAddrRoute, cls).tearDownClass()
1098
1099     def setUp(self):
1100         super(TestIPv6IfAddrRoute, self).setUp()
1101
1102         # create 1 pg interface
1103         self.create_pg_interfaces(range(1))
1104
1105         for i in self.pg_interfaces:
1106             i.admin_up()
1107             i.config_ip6()
1108             i.resolve_ndp()
1109
1110     def tearDown(self):
1111         super(TestIPv6IfAddrRoute, self).tearDown()
1112         for i in self.pg_interfaces:
1113             i.unconfig_ip6()
1114             i.admin_down()
1115
1116     def test_ipv6_ifaddrs_same_prefix(self):
1117         """ IPv6 Interface Addresses Same Prefix test
1118
1119         Test scenario:
1120
1121             - Verify no route in FIB for prefix 2001:10::/64
1122             - Configure IPv4 address 2001:10::10/64  on an interface
1123             - Verify route in FIB for prefix 2001:10::/64
1124             - Configure IPv4 address 2001:10::20/64 on an interface
1125             - Delete 2001:10::10/64 from interface
1126             - Verify route in FIB for prefix 2001:10::/64
1127             - Delete 2001:10::20/64 from interface
1128             - Verify no route in FIB for prefix 2001:10::/64
1129         """
1130
1131         addr1 = "2001:10::10"
1132         addr2 = "2001:10::20"
1133
1134         if_addr1 = VppIpInterfaceAddress(self, self.pg0, addr1, 64)
1135         if_addr2 = VppIpInterfaceAddress(self, self.pg0, addr2, 64)
1136         self.assertFalse(if_addr1.query_vpp_config())
1137         self.assertFalse(find_route(self, addr1, 128))
1138         self.assertFalse(find_route(self, addr2, 128))
1139
1140         # configure first address, verify route present
1141         if_addr1.add_vpp_config()
1142         self.assertTrue(if_addr1.query_vpp_config())
1143         self.assertTrue(find_route(self, addr1, 128))
1144         self.assertFalse(find_route(self, addr2, 128))
1145
1146         # configure second address, delete first, verify route not removed
1147         if_addr2.add_vpp_config()
1148         if_addr1.remove_vpp_config()
1149         self.assertFalse(if_addr1.query_vpp_config())
1150         self.assertTrue(if_addr2.query_vpp_config())
1151         self.assertFalse(find_route(self, addr1, 128))
1152         self.assertTrue(find_route(self, addr2, 128))
1153
1154         # delete second address, verify route removed
1155         if_addr2.remove_vpp_config()
1156         self.assertFalse(if_addr1.query_vpp_config())
1157         self.assertFalse(find_route(self, addr1, 128))
1158         self.assertFalse(find_route(self, addr2, 128))
1159
1160     def test_ipv6_ifaddr_del(self):
1161         """ Delete an interface address that does not exist """
1162
1163         loopbacks = self.create_loopback_interfaces(1)
1164         lo = self.lo_interfaces[0]
1165
1166         lo.config_ip6()
1167         lo.admin_up()
1168
1169         #
1170         # try and remove pg0's subnet from lo
1171         #
1172         with self.vapi.assert_negative_api_retval():
1173             self.vapi.sw_interface_add_del_address(
1174                 sw_if_index=lo.sw_if_index,
1175                 prefix=self.pg0.local_ip6_prefix,
1176                 is_add=0)
1177
1178
1179 class TestICMPv6Echo(VppTestCase):
1180     """ ICMPv6 Echo Test Case """
1181
1182     @classmethod
1183     def setUpClass(cls):
1184         super(TestICMPv6Echo, cls).setUpClass()
1185
1186     @classmethod
1187     def tearDownClass(cls):
1188         super(TestICMPv6Echo, cls).tearDownClass()
1189
1190     def setUp(self):
1191         super(TestICMPv6Echo, self).setUp()
1192
1193         # create 1 pg interface
1194         self.create_pg_interfaces(range(1))
1195
1196         for i in self.pg_interfaces:
1197             i.admin_up()
1198             i.config_ip6()
1199             i.resolve_ndp(link_layer=True)
1200             i.resolve_ndp()
1201
1202     def tearDown(self):
1203         super(TestICMPv6Echo, self).tearDown()
1204         for i in self.pg_interfaces:
1205             i.unconfig_ip6()
1206             i.admin_down()
1207
1208     def test_icmpv6_echo(self):
1209         """ VPP replies to ICMPv6 Echo Request
1210
1211         Test scenario:
1212
1213             - Receive ICMPv6 Echo Request message on pg0 interface.
1214             - Check outgoing ICMPv6 Echo Reply message on pg0 interface.
1215         """
1216
1217         # test both with global and local ipv6 addresses
1218         dsts = (self.pg0.local_ip6, self.pg0.local_ip6_ll)
1219         id = 0xb
1220         seq = 5
1221         data = b'\x0a' * 18
1222         p = list()
1223         for dst in dsts:
1224             p.append((Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
1225                       IPv6(src=self.pg0.remote_ip6, dst=dst) /
1226                       ICMPv6EchoRequest(id=id, seq=seq, data=data)))
1227
1228         self.pg0.add_stream(p)
1229         self.pg_enable_capture(self.pg_interfaces)
1230         self.pg_start()
1231         rxs = self.pg0.get_capture(len(dsts))
1232
1233         for rx, dst in zip(rxs, dsts):
1234             ether = rx[Ether]
1235             ipv6 = rx[IPv6]
1236             icmpv6 = rx[ICMPv6EchoReply]
1237             self.assertEqual(ether.src, self.pg0.local_mac)
1238             self.assertEqual(ether.dst, self.pg0.remote_mac)
1239             self.assertEqual(ipv6.src, dst)
1240             self.assertEqual(ipv6.dst, self.pg0.remote_ip6)
1241             self.assertEqual(icmp6types[icmpv6.type], "Echo Reply")
1242             self.assertEqual(icmpv6.id, id)
1243             self.assertEqual(icmpv6.seq, seq)
1244             self.assertEqual(icmpv6.data, data)
1245
1246
1247 class TestIPv6RD(TestIPv6ND):
1248     """ IPv6 Router Discovery Test Case """
1249
1250     @classmethod
1251     def setUpClass(cls):
1252         super(TestIPv6RD, cls).setUpClass()
1253
1254     @classmethod
1255     def tearDownClass(cls):
1256         super(TestIPv6RD, cls).tearDownClass()
1257
1258     def setUp(self):
1259         super(TestIPv6RD, self).setUp()
1260
1261         # create 2 pg interfaces
1262         self.create_pg_interfaces(range(2))
1263
1264         self.interfaces = list(self.pg_interfaces)
1265
1266         # setup all interfaces
1267         for i in self.interfaces:
1268             i.admin_up()
1269             i.config_ip6()
1270
1271     def tearDown(self):
1272         for i in self.interfaces:
1273             i.unconfig_ip6()
1274             i.admin_down()
1275         super(TestIPv6RD, self).tearDown()
1276
1277     def test_rd_send_router_solicitation(self):
1278         """ Verify router solicitation packets """
1279
1280         count = 2
1281         self.pg_enable_capture(self.pg_interfaces)
1282         self.pg_start()
1283         self.vapi.ip6nd_send_router_solicitation(self.pg1.sw_if_index,
1284                                                  mrc=count)
1285         rx_list = self.pg1.get_capture(count, timeout=3)
1286         self.assertEqual(len(rx_list), count)
1287         for packet in rx_list:
1288             self.assertEqual(packet.haslayer(IPv6), 1)
1289             self.assertEqual(packet[IPv6].haslayer(
1290                 ICMPv6ND_RS), 1)
1291             dst = ip6_normalize(packet[IPv6].dst)
1292             dst2 = ip6_normalize("ff02::2")
1293             self.assert_equal(dst, dst2)
1294             src = ip6_normalize(packet[IPv6].src)
1295             src2 = ip6_normalize(self.pg1.local_ip6_ll)
1296             self.assert_equal(src, src2)
1297             self.assertTrue(
1298                 bool(packet[ICMPv6ND_RS].haslayer(
1299                     ICMPv6NDOptSrcLLAddr)))
1300             self.assert_equal(
1301                 packet[ICMPv6NDOptSrcLLAddr].lladdr,
1302                 self.pg1.local_mac)
1303
1304     def verify_prefix_info(self, reported_prefix, prefix_option):
1305         prefix = IPv6Network(
1306             text_type(prefix_option.getfieldval("prefix") +
1307                       "/" +
1308                       text_type(prefix_option.getfieldval("prefixlen"))),
1309             strict=False)
1310         self.assert_equal(reported_prefix.prefix.network_address,
1311                           prefix.network_address)
1312         L = prefix_option.getfieldval("L")
1313         A = prefix_option.getfieldval("A")
1314         option_flags = (L << 7) | (A << 6)
1315         self.assert_equal(reported_prefix.flags, option_flags)
1316         self.assert_equal(reported_prefix.valid_time,
1317                           prefix_option.getfieldval("validlifetime"))
1318         self.assert_equal(reported_prefix.preferred_time,
1319                           prefix_option.getfieldval("preferredlifetime"))
1320
1321     def test_rd_receive_router_advertisement(self):
1322         """ Verify events triggered by received RA packets """
1323
1324         self.vapi.want_ip6_ra_events(enable=1)
1325
1326         prefix_info_1 = ICMPv6NDOptPrefixInfo(
1327             prefix="1::2",
1328             prefixlen=50,
1329             validlifetime=200,
1330             preferredlifetime=500,
1331             L=1,
1332             A=1,
1333         )
1334
1335         prefix_info_2 = ICMPv6NDOptPrefixInfo(
1336             prefix="7::4",
1337             prefixlen=20,
1338             validlifetime=70,
1339             preferredlifetime=1000,
1340             L=1,
1341             A=0,
1342         )
1343
1344         p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
1345              IPv6(dst=self.pg1.local_ip6_ll,
1346                   src=mk_ll_addr(self.pg1.remote_mac)) /
1347              ICMPv6ND_RA() /
1348              prefix_info_1 /
1349              prefix_info_2)
1350         self.pg1.add_stream([p])
1351         self.pg_start()
1352
1353         ev = self.vapi.wait_for_event(10, "ip6_ra_event")
1354
1355         self.assert_equal(ev.current_hop_limit, 0)
1356         self.assert_equal(ev.flags, 8)
1357         self.assert_equal(ev.router_lifetime_in_sec, 1800)
1358         self.assert_equal(ev.neighbor_reachable_time_in_msec, 0)
1359         self.assert_equal(
1360             ev.time_in_msec_between_retransmitted_neighbor_solicitations, 0)
1361
1362         self.assert_equal(ev.n_prefixes, 2)
1363
1364         self.verify_prefix_info(ev.prefixes[0], prefix_info_1)
1365         self.verify_prefix_info(ev.prefixes[1], prefix_info_2)
1366
1367
1368 class TestIPv6RDControlPlane(TestIPv6ND):
1369     """ IPv6 Router Discovery Control Plane Test Case """
1370
1371     @classmethod
1372     def setUpClass(cls):
1373         super(TestIPv6RDControlPlane, cls).setUpClass()
1374
1375     @classmethod
1376     def tearDownClass(cls):
1377         super(TestIPv6RDControlPlane, cls).tearDownClass()
1378
1379     def setUp(self):
1380         super(TestIPv6RDControlPlane, self).setUp()
1381
1382         # create 1 pg interface
1383         self.create_pg_interfaces(range(1))
1384
1385         self.interfaces = list(self.pg_interfaces)
1386
1387         # setup all interfaces
1388         for i in self.interfaces:
1389             i.admin_up()
1390             i.config_ip6()
1391
1392     def tearDown(self):
1393         super(TestIPv6RDControlPlane, self).tearDown()
1394
1395     @staticmethod
1396     def create_ra_packet(pg, routerlifetime=None):
1397         src_ip = pg.remote_ip6_ll
1398         dst_ip = pg.local_ip6
1399         if routerlifetime is not None:
1400             ra = ICMPv6ND_RA(routerlifetime=routerlifetime)
1401         else:
1402             ra = ICMPv6ND_RA()
1403         p = (Ether(dst=pg.local_mac, src=pg.remote_mac) /
1404              IPv6(dst=dst_ip, src=src_ip) / ra)
1405         return p
1406
1407     @staticmethod
1408     def get_default_routes(fib):
1409         list = []
1410         for entry in fib:
1411             if entry.route.prefix.prefixlen == 0:
1412                 for path in entry.route.paths:
1413                     if path.sw_if_index != 0xFFFFFFFF:
1414                         defaut_route = {}
1415                         defaut_route['sw_if_index'] = path.sw_if_index
1416                         defaut_route['next_hop'] = path.nh.address.ip6
1417                         list.append(defaut_route)
1418         return list
1419
1420     @staticmethod
1421     def get_interface_addresses(fib, pg):
1422         list = []
1423         for entry in fib:
1424             if entry.route.prefix.prefixlen == 128:
1425                 path = entry.route.paths[0]
1426                 if path.sw_if_index == pg.sw_if_index:
1427                     list.append(str(entry.route.prefix.network_address))
1428         return list
1429
1430     def wait_for_no_default_route(self, n_tries=50, s_time=1):
1431         while (n_tries):
1432             fib = self.vapi.ip_route_dump(0, True)
1433             default_routes = self.get_default_routes(fib)
1434             if 0 == len(default_routes):
1435                 return True
1436             n_tries = n_tries - 1
1437             self.sleep(s_time)
1438
1439         return False
1440
1441     def test_all(self):
1442         """ Test handling of SLAAC addresses and default routes """
1443
1444         fib = self.vapi.ip_route_dump(0, True)
1445         default_routes = self.get_default_routes(fib)
1446         initial_addresses = set(self.get_interface_addresses(fib, self.pg0))
1447         self.assertEqual(default_routes, [])
1448         router_address = IPv6Address(text_type(self.pg0.remote_ip6_ll))
1449
1450         self.vapi.ip6_nd_address_autoconfig(self.pg0.sw_if_index, 1, 1)
1451
1452         self.sleep(0.1)
1453
1454         # send RA
1455         packet = (self.create_ra_packet(
1456             self.pg0) / ICMPv6NDOptPrefixInfo(
1457             prefix="1::",
1458             prefixlen=64,
1459             validlifetime=2,
1460             preferredlifetime=2,
1461             L=1,
1462             A=1,
1463         ) / ICMPv6NDOptPrefixInfo(
1464             prefix="7::",
1465             prefixlen=20,
1466             validlifetime=1500,
1467             preferredlifetime=1000,
1468             L=1,
1469             A=0,
1470         ))
1471         self.pg0.add_stream([packet])
1472         self.pg_start()
1473
1474         self.sleep_on_vpp_time(0.1)
1475
1476         fib = self.vapi.ip_route_dump(0, True)
1477
1478         # check FIB for new address
1479         addresses = set(self.get_interface_addresses(fib, self.pg0))
1480         new_addresses = addresses.difference(initial_addresses)
1481         self.assertEqual(len(new_addresses), 1)
1482         prefix = IPv6Network(text_type("%s/%d" % (list(new_addresses)[0], 20)),
1483                              strict=False)
1484         self.assertEqual(prefix, IPv6Network(text_type('1::/20')))
1485
1486         # check FIB for new default route
1487         default_routes = self.get_default_routes(fib)
1488         self.assertEqual(len(default_routes), 1)
1489         dr = default_routes[0]
1490         self.assertEqual(dr['sw_if_index'], self.pg0.sw_if_index)
1491         self.assertEqual(dr['next_hop'], router_address)
1492
1493         # send RA to delete default route
1494         packet = self.create_ra_packet(self.pg0, routerlifetime=0)
1495         self.pg0.add_stream([packet])
1496         self.pg_start()
1497
1498         self.sleep_on_vpp_time(0.1)
1499
1500         # check that default route is deleted
1501         fib = self.vapi.ip_route_dump(0, True)
1502         default_routes = self.get_default_routes(fib)
1503         self.assertEqual(len(default_routes), 0)
1504
1505         self.sleep_on_vpp_time(0.1)
1506
1507         # send RA
1508         packet = self.create_ra_packet(self.pg0)
1509         self.pg0.add_stream([packet])
1510         self.pg_start()
1511
1512         self.sleep_on_vpp_time(0.1)
1513
1514         # check FIB for new default route
1515         fib = self.vapi.ip_route_dump(0, True)
1516         default_routes = self.get_default_routes(fib)
1517         self.assertEqual(len(default_routes), 1)
1518         dr = default_routes[0]
1519         self.assertEqual(dr['sw_if_index'], self.pg0.sw_if_index)
1520         self.assertEqual(dr['next_hop'], router_address)
1521
1522         # send RA, updating router lifetime to 1s
1523         packet = self.create_ra_packet(self.pg0, 1)
1524         self.pg0.add_stream([packet])
1525         self.pg_start()
1526
1527         self.sleep_on_vpp_time(0.1)
1528
1529         # check that default route still exists
1530         fib = self.vapi.ip_route_dump(0, True)
1531         default_routes = self.get_default_routes(fib)
1532         self.assertEqual(len(default_routes), 1)
1533         dr = default_routes[0]
1534         self.assertEqual(dr['sw_if_index'], self.pg0.sw_if_index)
1535         self.assertEqual(dr['next_hop'], router_address)
1536
1537         self.sleep_on_vpp_time(1)
1538
1539         # check that default route is deleted
1540         self.assertTrue(self.wait_for_no_default_route())
1541
1542         # check FIB still contains the SLAAC address
1543         addresses = set(self.get_interface_addresses(fib, self.pg0))
1544         new_addresses = addresses.difference(initial_addresses)
1545
1546         self.assertEqual(len(new_addresses), 1)
1547         prefix = IPv6Network(text_type("%s/%d" % (list(new_addresses)[0], 20)),
1548                              strict=False)
1549         self.assertEqual(prefix, IPv6Network(text_type('1::/20')))
1550
1551         self.sleep_on_vpp_time(1)
1552
1553         # check that SLAAC address is deleted
1554         fib = self.vapi.ip_route_dump(0, True)
1555         addresses = set(self.get_interface_addresses(fib, self.pg0))
1556         new_addresses = addresses.difference(initial_addresses)
1557         self.assertEqual(len(new_addresses), 0)
1558
1559
1560 class IPv6NDProxyTest(TestIPv6ND):
1561     """ IPv6 ND ProxyTest Case """
1562
1563     @classmethod
1564     def setUpClass(cls):
1565         super(IPv6NDProxyTest, cls).setUpClass()
1566
1567     @classmethod
1568     def tearDownClass(cls):
1569         super(IPv6NDProxyTest, cls).tearDownClass()
1570
1571     def setUp(self):
1572         super(IPv6NDProxyTest, self).setUp()
1573
1574         # create 3 pg interfaces
1575         self.create_pg_interfaces(range(3))
1576
1577         # pg0 is the master interface, with the configured subnet
1578         self.pg0.admin_up()
1579         self.pg0.config_ip6()
1580         self.pg0.resolve_ndp()
1581
1582         self.pg1.ip6_enable()
1583         self.pg2.ip6_enable()
1584
1585     def tearDown(self):
1586         super(IPv6NDProxyTest, self).tearDown()
1587
1588     def test_nd_proxy(self):
1589         """ IPv6 Proxy ND """
1590
1591         #
1592         # Generate some hosts in the subnet that we are proxying
1593         #
1594         self.pg0.generate_remote_hosts(8)
1595
1596         nsma = in6_getnsma(inet_pton(AF_INET6, self.pg0.local_ip6))
1597         d = inet_ntop(AF_INET6, nsma)
1598
1599         #
1600         # Send an NS for one of those remote hosts on one of the proxy links
1601         # expect no response since it's from an address that is not
1602         # on the link that has the prefix configured
1603         #
1604         ns_pg1 = (Ether(dst=in6_getnsmac(nsma), src=self.pg1.remote_mac) /
1605                   IPv6(dst=d,
1606                        src=self.pg0._remote_hosts[2].ip6) /
1607                   ICMPv6ND_NS(tgt=self.pg0.local_ip6) /
1608                   ICMPv6NDOptSrcLLAddr(
1609                       lladdr=self.pg0._remote_hosts[2].mac))
1610
1611         self.send_and_assert_no_replies(self.pg1, ns_pg1, "Off link NS")
1612
1613         #
1614         # Add proxy support for the host
1615         #
1616         self.vapi.ip6nd_proxy_add_del(
1617             is_add=1, ip=inet_pton(AF_INET6, self.pg0._remote_hosts[2].ip6),
1618             sw_if_index=self.pg1.sw_if_index)
1619
1620         #
1621         # try that NS again. this time we expect an NA back
1622         #
1623         self.send_and_expect_na(self.pg1, ns_pg1,
1624                                 "NS to proxy entry",
1625                                 dst_ip=self.pg0._remote_hosts[2].ip6,
1626                                 tgt_ip=self.pg0.local_ip6)
1627
1628         #
1629         # ... and that we have an entry in the ND cache
1630         #
1631         self.assertTrue(find_nbr(self,
1632                                  self.pg1.sw_if_index,
1633                                  self.pg0._remote_hosts[2].ip6))
1634
1635         #
1636         # ... and we can route traffic to it
1637         #
1638         t = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
1639              IPv6(dst=self.pg0._remote_hosts[2].ip6,
1640                   src=self.pg0.remote_ip6) /
1641              inet6.UDP(sport=10000, dport=20000) /
1642              Raw(b'\xa5' * 100))
1643
1644         self.pg0.add_stream(t)
1645         self.pg_enable_capture(self.pg_interfaces)
1646         self.pg_start()
1647         rx = self.pg1.get_capture(1)
1648         rx = rx[0]
1649
1650         self.assertEqual(rx[Ether].dst, self.pg0._remote_hosts[2].mac)
1651         self.assertEqual(rx[Ether].src, self.pg1.local_mac)
1652
1653         self.assertEqual(rx[IPv6].src,
1654                          t[IPv6].src)
1655         self.assertEqual(rx[IPv6].dst,
1656                          t[IPv6].dst)
1657
1658         #
1659         # Test we proxy for the host on the main interface
1660         #
1661         ns_pg0 = (Ether(dst=in6_getnsmac(nsma), src=self.pg0.remote_mac) /
1662                   IPv6(dst=d, src=self.pg0.remote_ip6) /
1663                   ICMPv6ND_NS(
1664                       tgt=self.pg0._remote_hosts[2].ip6) /
1665                   ICMPv6NDOptSrcLLAddr(
1666                       lladdr=self.pg0.remote_mac))
1667
1668         self.send_and_expect_na(self.pg0, ns_pg0,
1669                                 "NS to proxy entry on main",
1670                                 tgt_ip=self.pg0._remote_hosts[2].ip6,
1671                                 dst_ip=self.pg0.remote_ip6)
1672
1673         #
1674         # Setup and resolve proxy for another host on another interface
1675         #
1676         ns_pg2 = (Ether(dst=in6_getnsmac(nsma), src=self.pg2.remote_mac) /
1677                   IPv6(dst=d,
1678                        src=self.pg0._remote_hosts[3].ip6) /
1679                   ICMPv6ND_NS(tgt=self.pg0.local_ip6) /
1680                   ICMPv6NDOptSrcLLAddr(
1681                       lladdr=self.pg0._remote_hosts[2].mac))
1682
1683         self.vapi.ip6nd_proxy_add_del(
1684             is_add=1, ip=inet_pton(AF_INET6, self.pg0._remote_hosts[3].ip6),
1685             sw_if_index=self.pg2.sw_if_index)
1686
1687         self.send_and_expect_na(self.pg2, ns_pg2,
1688                                 "NS to proxy entry other interface",
1689                                 dst_ip=self.pg0._remote_hosts[3].ip6,
1690                                 tgt_ip=self.pg0.local_ip6)
1691
1692         self.assertTrue(find_nbr(self,
1693                                  self.pg2.sw_if_index,
1694                                  self.pg0._remote_hosts[3].ip6))
1695
1696         #
1697         # hosts can communicate. pg2->pg1
1698         #
1699         t2 = (Ether(dst=self.pg2.local_mac,
1700                     src=self.pg0.remote_hosts[3].mac) /
1701               IPv6(dst=self.pg0._remote_hosts[2].ip6,
1702                    src=self.pg0._remote_hosts[3].ip6) /
1703               inet6.UDP(sport=10000, dport=20000) /
1704               Raw(b'\xa5' * 100))
1705
1706         self.pg2.add_stream(t2)
1707         self.pg_enable_capture(self.pg_interfaces)
1708         self.pg_start()
1709         rx = self.pg1.get_capture(1)
1710         rx = rx[0]
1711
1712         self.assertEqual(rx[Ether].dst, self.pg0._remote_hosts[2].mac)
1713         self.assertEqual(rx[Ether].src, self.pg1.local_mac)
1714
1715         self.assertEqual(rx[IPv6].src,
1716                          t2[IPv6].src)
1717         self.assertEqual(rx[IPv6].dst,
1718                          t2[IPv6].dst)
1719
1720         #
1721         # remove the proxy configs
1722         #
1723         self.vapi.ip6nd_proxy_add_del(
1724             ip=inet_pton(AF_INET6, self.pg0._remote_hosts[2].ip6),
1725             sw_if_index=self.pg1.sw_if_index, is_add=0)
1726         self.vapi.ip6nd_proxy_add_del(
1727             ip=inet_pton(AF_INET6, self.pg0._remote_hosts[3].ip6),
1728             sw_if_index=self.pg2.sw_if_index, is_add=0)
1729
1730         self.assertFalse(find_nbr(self,
1731                                   self.pg2.sw_if_index,
1732                                   self.pg0._remote_hosts[3].ip6))
1733         self.assertFalse(find_nbr(self,
1734                                   self.pg1.sw_if_index,
1735                                   self.pg0._remote_hosts[2].ip6))
1736
1737         #
1738         # no longer proxy-ing...
1739         #
1740         self.send_and_assert_no_replies(self.pg0, ns_pg0, "Proxy unconfigured")
1741         self.send_and_assert_no_replies(self.pg1, ns_pg1, "Proxy unconfigured")
1742         self.send_and_assert_no_replies(self.pg2, ns_pg2, "Proxy unconfigured")
1743
1744         #
1745         # no longer forwarding. traffic generates NS out of the glean/main
1746         # interface
1747         #
1748         self.pg2.add_stream(t2)
1749         self.pg_enable_capture(self.pg_interfaces)
1750         self.pg_start()
1751
1752         rx = self.pg0.get_capture(1)
1753
1754         self.assertTrue(rx[0].haslayer(ICMPv6ND_NS))
1755
1756
1757 class TestIPNull(VppTestCase):
1758     """ IPv6 routes via NULL """
1759
1760     @classmethod
1761     def setUpClass(cls):
1762         super(TestIPNull, cls).setUpClass()
1763
1764     @classmethod
1765     def tearDownClass(cls):
1766         super(TestIPNull, cls).tearDownClass()
1767
1768     def setUp(self):
1769         super(TestIPNull, self).setUp()
1770
1771         # create 2 pg interfaces
1772         self.create_pg_interfaces(range(1))
1773
1774         for i in self.pg_interfaces:
1775             i.admin_up()
1776             i.config_ip6()
1777             i.resolve_ndp()
1778
1779     def tearDown(self):
1780         super(TestIPNull, self).tearDown()
1781         for i in self.pg_interfaces:
1782             i.unconfig_ip6()
1783             i.admin_down()
1784
1785     def test_ip_null(self):
1786         """ IP NULL route """
1787
1788         p = (Ether(src=self.pg0.remote_mac,
1789                    dst=self.pg0.local_mac) /
1790              IPv6(src=self.pg0.remote_ip6, dst="2001::1") /
1791              inet6.UDP(sport=1234, dport=1234) /
1792              Raw(b'\xa5' * 100))
1793
1794         #
1795         # A route via IP NULL that will reply with ICMP unreachables
1796         #
1797         ip_unreach = VppIpRoute(
1798             self, "2001::", 64,
1799             [VppRoutePath("::", 0xffffffff,
1800                           type=FibPathType.FIB_PATH_TYPE_ICMP_UNREACH)])
1801         ip_unreach.add_vpp_config()
1802
1803         self.pg0.add_stream(p)
1804         self.pg_enable_capture(self.pg_interfaces)
1805         self.pg_start()
1806
1807         rx = self.pg0.get_capture(1)
1808         rx = rx[0]
1809         icmp = rx[ICMPv6DestUnreach]
1810
1811         # 0 = "No route to destination"
1812         self.assertEqual(icmp.code, 0)
1813
1814         # ICMP is rate limited. pause a bit
1815         self.sleep(1)
1816
1817         #
1818         # A route via IP NULL that will reply with ICMP prohibited
1819         #
1820         ip_prohibit = VppIpRoute(
1821             self, "2001::1", 128,
1822             [VppRoutePath("::", 0xffffffff,
1823                           type=FibPathType.FIB_PATH_TYPE_ICMP_PROHIBIT)])
1824         ip_prohibit.add_vpp_config()
1825
1826         self.pg0.add_stream(p)
1827         self.pg_enable_capture(self.pg_interfaces)
1828         self.pg_start()
1829
1830         rx = self.pg0.get_capture(1)
1831         rx = rx[0]
1832         icmp = rx[ICMPv6DestUnreach]
1833
1834         # 1 = "Communication with destination administratively prohibited"
1835         self.assertEqual(icmp.code, 1)
1836
1837
1838 class TestIPDisabled(VppTestCase):
1839     """ IPv6 disabled """
1840
1841     @classmethod
1842     def setUpClass(cls):
1843         super(TestIPDisabled, cls).setUpClass()
1844
1845     @classmethod
1846     def tearDownClass(cls):
1847         super(TestIPDisabled, cls).tearDownClass()
1848
1849     def setUp(self):
1850         super(TestIPDisabled, self).setUp()
1851
1852         # create 2 pg interfaces
1853         self.create_pg_interfaces(range(2))
1854
1855         # PG0 is IP enabled
1856         self.pg0.admin_up()
1857         self.pg0.config_ip6()
1858         self.pg0.resolve_ndp()
1859
1860         # PG 1 is not IP enabled
1861         self.pg1.admin_up()
1862
1863     def tearDown(self):
1864         super(TestIPDisabled, self).tearDown()
1865         for i in self.pg_interfaces:
1866             i.unconfig_ip4()
1867             i.admin_down()
1868
1869     def test_ip_disabled(self):
1870         """ IP Disabled """
1871
1872         MRouteItfFlags = VppEnum.vl_api_mfib_itf_flags_t
1873         MRouteEntryFlags = VppEnum.vl_api_mfib_entry_flags_t
1874         #
1875         # An (S,G).
1876         # one accepting interface, pg0, 2 forwarding interfaces
1877         #
1878         route_ff_01 = VppIpMRoute(
1879             self,
1880             "::",
1881             "ffef::1", 128,
1882             MRouteEntryFlags.MFIB_API_ENTRY_FLAG_NONE,
1883             [VppMRoutePath(self.pg1.sw_if_index,
1884                            MRouteItfFlags.MFIB_API_ITF_FLAG_ACCEPT),
1885              VppMRoutePath(self.pg0.sw_if_index,
1886                            MRouteItfFlags.MFIB_API_ITF_FLAG_FORWARD)])
1887         route_ff_01.add_vpp_config()
1888
1889         pu = (Ether(src=self.pg1.remote_mac,
1890                     dst=self.pg1.local_mac) /
1891               IPv6(src="2001::1", dst=self.pg0.remote_ip6) /
1892               inet6.UDP(sport=1234, dport=1234) /
1893               Raw(b'\xa5' * 100))
1894         pm = (Ether(src=self.pg1.remote_mac,
1895                     dst=self.pg1.local_mac) /
1896               IPv6(src="2001::1", dst="ffef::1") /
1897               inet6.UDP(sport=1234, dport=1234) /
1898               Raw(b'\xa5' * 100))
1899
1900         #
1901         # PG1 does not forward IP traffic
1902         #
1903         self.send_and_assert_no_replies(self.pg1, pu, "IPv6 disabled")
1904         self.send_and_assert_no_replies(self.pg1, pm, "IPv6 disabled")
1905
1906         #
1907         # IP enable PG1
1908         #
1909         self.pg1.config_ip6()
1910
1911         #
1912         # Now we get packets through
1913         #
1914         self.pg1.add_stream(pu)
1915         self.pg_enable_capture(self.pg_interfaces)
1916         self.pg_start()
1917         rx = self.pg0.get_capture(1)
1918
1919         self.pg1.add_stream(pm)
1920         self.pg_enable_capture(self.pg_interfaces)
1921         self.pg_start()
1922         rx = self.pg0.get_capture(1)
1923
1924         #
1925         # Disable PG1
1926         #
1927         self.pg1.unconfig_ip6()
1928
1929         #
1930         # PG1 does not forward IP traffic
1931         #
1932         self.send_and_assert_no_replies(self.pg1, pu, "IPv6 disabled")
1933         self.send_and_assert_no_replies(self.pg1, pm, "IPv6 disabled")
1934
1935
1936 class TestIP6LoadBalance(VppTestCase):
1937     """ IPv6 Load-Balancing """
1938
1939     @classmethod
1940     def setUpClass(cls):
1941         super(TestIP6LoadBalance, cls).setUpClass()
1942
1943     @classmethod
1944     def tearDownClass(cls):
1945         super(TestIP6LoadBalance, cls).tearDownClass()
1946
1947     def setUp(self):
1948         super(TestIP6LoadBalance, self).setUp()
1949
1950         self.create_pg_interfaces(range(5))
1951
1952         mpls_tbl = VppMplsTable(self, 0)
1953         mpls_tbl.add_vpp_config()
1954
1955         for i in self.pg_interfaces:
1956             i.admin_up()
1957             i.config_ip6()
1958             i.resolve_ndp()
1959             i.enable_mpls()
1960
1961     def tearDown(self):
1962         for i in self.pg_interfaces:
1963             i.unconfig_ip6()
1964             i.admin_down()
1965             i.disable_mpls()
1966         super(TestIP6LoadBalance, self).tearDown()
1967
1968     def pg_send(self, input, pkts):
1969         self.vapi.cli("clear trace")
1970         input.add_stream(pkts)
1971         self.pg_enable_capture(self.pg_interfaces)
1972         self.pg_start()
1973
1974     def send_and_expect_load_balancing(self, input, pkts, outputs):
1975         self.pg_send(input, pkts)
1976         rxs = []
1977         for oo in outputs:
1978             rx = oo._get_capture(1)
1979             self.assertNotEqual(0, len(rx))
1980             rxs.append(rx)
1981         return rxs
1982
1983     def send_and_expect_one_itf(self, input, pkts, itf):
1984         self.pg_send(input, pkts)
1985         rx = itf.get_capture(len(pkts))
1986
1987     def test_ip6_load_balance(self):
1988         """ IPv6 Load-Balancing """
1989
1990         #
1991         # An array of packets that differ only in the destination port
1992         #  - IP only
1993         #  - MPLS EOS
1994         #  - MPLS non-EOS
1995         #  - MPLS non-EOS with an entropy label
1996         #
1997         port_ip_pkts = []
1998         port_mpls_pkts = []
1999         port_mpls_neos_pkts = []
2000         port_ent_pkts = []
2001
2002         #
2003         # An array of packets that differ only in the source address
2004         #
2005         src_ip_pkts = []
2006         src_mpls_pkts = []
2007
2008         for ii in range(NUM_PKTS):
2009             port_ip_hdr = (
2010                 IPv6(dst="3000::1", src="3000:1::1") /
2011                 inet6.UDP(sport=1234, dport=1234 + ii) /
2012                 Raw(b'\xa5' * 100))
2013             port_ip_pkts.append((Ether(src=self.pg0.remote_mac,
2014                                        dst=self.pg0.local_mac) /
2015                                  port_ip_hdr))
2016             port_mpls_pkts.append((Ether(src=self.pg0.remote_mac,
2017                                          dst=self.pg0.local_mac) /
2018                                    MPLS(label=66, ttl=2) /
2019                                    port_ip_hdr))
2020             port_mpls_neos_pkts.append((Ether(src=self.pg0.remote_mac,
2021                                               dst=self.pg0.local_mac) /
2022                                         MPLS(label=67, ttl=2) /
2023                                         MPLS(label=77, ttl=2) /
2024                                         port_ip_hdr))
2025             port_ent_pkts.append((Ether(src=self.pg0.remote_mac,
2026                                         dst=self.pg0.local_mac) /
2027                                   MPLS(label=67, ttl=2) /
2028                                   MPLS(label=14, ttl=2) /
2029                                   MPLS(label=999, ttl=2) /
2030                                   port_ip_hdr))
2031             src_ip_hdr = (
2032                 IPv6(dst="3000::1", src="3000:1::%d" % ii) /
2033                 inet6.UDP(sport=1234, dport=1234) /
2034                 Raw(b'\xa5' * 100))
2035             src_ip_pkts.append((Ether(src=self.pg0.remote_mac,
2036                                       dst=self.pg0.local_mac) /
2037                                 src_ip_hdr))
2038             src_mpls_pkts.append((Ether(src=self.pg0.remote_mac,
2039                                         dst=self.pg0.local_mac) /
2040                                   MPLS(label=66, ttl=2) /
2041                                   src_ip_hdr))
2042
2043         #
2044         # A route for the IP packets
2045         #
2046         route_3000_1 = VppIpRoute(self, "3000::1", 128,
2047                                   [VppRoutePath(self.pg1.remote_ip6,
2048                                                 self.pg1.sw_if_index),
2049                                    VppRoutePath(self.pg2.remote_ip6,
2050                                                 self.pg2.sw_if_index)])
2051         route_3000_1.add_vpp_config()
2052
2053         #
2054         # a local-label for the EOS packets
2055         #
2056         binding = VppMplsIpBind(self, 66, "3000::1", 128, is_ip6=1)
2057         binding.add_vpp_config()
2058
2059         #
2060         # An MPLS route for the non-EOS packets
2061         #
2062         route_67 = VppMplsRoute(self, 67, 0,
2063                                 [VppRoutePath(self.pg1.remote_ip6,
2064                                               self.pg1.sw_if_index,
2065                                               labels=[67]),
2066                                  VppRoutePath(self.pg2.remote_ip6,
2067                                               self.pg2.sw_if_index,
2068                                               labels=[67])])
2069         route_67.add_vpp_config()
2070
2071         #
2072         # inject the packet on pg0 - expect load-balancing across the 2 paths
2073         #  - since the default hash config is to use IP src,dst and port
2074         #    src,dst
2075         # We are not going to ensure equal amounts of packets across each link,
2076         # since the hash algorithm is statistical and therefore this can never
2077         # be guaranteed. But with 64 different packets we do expect some
2078         # balancing. So instead just ensure there is traffic on each link.
2079         #
2080         rx = self.send_and_expect_load_balancing(self.pg0, port_ip_pkts,
2081                                                  [self.pg1, self.pg2])
2082         n_ip_pg0 = len(rx[0])
2083         self.send_and_expect_load_balancing(self.pg0, src_ip_pkts,
2084                                             [self.pg1, self.pg2])
2085         self.send_and_expect_load_balancing(self.pg0, port_mpls_pkts,
2086                                             [self.pg1, self.pg2])
2087         self.send_and_expect_load_balancing(self.pg0, src_mpls_pkts,
2088                                             [self.pg1, self.pg2])
2089         rx = self.send_and_expect_load_balancing(self.pg0, port_mpls_neos_pkts,
2090                                                  [self.pg1, self.pg2])
2091         n_mpls_pg0 = len(rx[0])
2092
2093         #
2094         # change the router ID and expect the distribution changes
2095         #
2096         self.vapi.set_ip_flow_hash_router_id(router_id=0x11111111)
2097
2098         rx = self.send_and_expect_load_balancing(self.pg0, port_ip_pkts,
2099                                                  [self.pg1, self.pg2])
2100         self.assertNotEqual(n_ip_pg0, len(rx[0]))
2101
2102         rx = self.send_and_expect_load_balancing(self.pg0, src_mpls_pkts,
2103                                                  [self.pg1, self.pg2])
2104         self.assertNotEqual(n_mpls_pg0, len(rx[0]))
2105
2106         #
2107         # The packets with Entropy label in should not load-balance,
2108         # since the Entropy value is fixed.
2109         #
2110         self.send_and_expect_one_itf(self.pg0, port_ent_pkts, self.pg1)
2111
2112         #
2113         # change the flow hash config so it's only IP src,dst
2114         #  - now only the stream with differing source address will
2115         #    load-balance
2116         #
2117         self.vapi.set_ip_flow_hash(vrf_id=0, src=1, dst=1, proto=1,
2118                                    sport=0, dport=0, is_ipv6=1)
2119
2120         self.send_and_expect_load_balancing(self.pg0, src_ip_pkts,
2121                                             [self.pg1, self.pg2])
2122         self.send_and_expect_load_balancing(self.pg0, src_mpls_pkts,
2123                                             [self.pg1, self.pg2])
2124         self.send_and_expect_one_itf(self.pg0, port_ip_pkts, self.pg2)
2125
2126         #
2127         # change the flow hash config back to defaults
2128         #
2129         self.vapi.set_ip_flow_hash(vrf_id=0, src=1, dst=1, sport=1, dport=1,
2130                                    proto=1, is_ipv6=1)
2131
2132         #
2133         # Recursive prefixes
2134         #  - testing that 2 stages of load-balancing occurs and there is no
2135         #    polarisation (i.e. only 2 of 4 paths are used)
2136         #
2137         port_pkts = []
2138         src_pkts = []
2139
2140         for ii in range(257):
2141             port_pkts.append((Ether(src=self.pg0.remote_mac,
2142                                     dst=self.pg0.local_mac) /
2143                               IPv6(dst="4000::1",
2144                                    src="4000:1::1") /
2145                               inet6.UDP(sport=1234,
2146                                         dport=1234 + ii) /
2147                               Raw(b'\xa5' * 100)))
2148             src_pkts.append((Ether(src=self.pg0.remote_mac,
2149                                    dst=self.pg0.local_mac) /
2150                              IPv6(dst="4000::1",
2151                                   src="4000:1::%d" % ii) /
2152                              inet6.UDP(sport=1234, dport=1234) /
2153                              Raw(b'\xa5' * 100)))
2154
2155         route_3000_2 = VppIpRoute(self, "3000::2", 128,
2156                                   [VppRoutePath(self.pg3.remote_ip6,
2157                                                 self.pg3.sw_if_index),
2158                                    VppRoutePath(self.pg4.remote_ip6,
2159                                                 self.pg4.sw_if_index)])
2160         route_3000_2.add_vpp_config()
2161
2162         route_4000_1 = VppIpRoute(self, "4000::1", 128,
2163                                   [VppRoutePath("3000::1",
2164                                                 0xffffffff),
2165                                    VppRoutePath("3000::2",
2166                                                 0xffffffff)])
2167         route_4000_1.add_vpp_config()
2168
2169         #
2170         # inject the packet on pg0 - expect load-balancing across all 4 paths
2171         #
2172         self.vapi.cli("clear trace")
2173         self.send_and_expect_load_balancing(self.pg0, port_pkts,
2174                                             [self.pg1, self.pg2,
2175                                              self.pg3, self.pg4])
2176         self.send_and_expect_load_balancing(self.pg0, src_pkts,
2177                                             [self.pg1, self.pg2,
2178                                              self.pg3, self.pg4])
2179
2180         #
2181         # Recursive prefixes
2182         #  - testing that 2 stages of load-balancing no choices
2183         #
2184         port_pkts = []
2185
2186         for ii in range(257):
2187             port_pkts.append((Ether(src=self.pg0.remote_mac,
2188                                     dst=self.pg0.local_mac) /
2189                               IPv6(dst="6000::1",
2190                                    src="6000:1::1") /
2191                               inet6.UDP(sport=1234,
2192                                         dport=1234 + ii) /
2193                               Raw(b'\xa5' * 100)))
2194
2195         route_5000_2 = VppIpRoute(self, "5000::2", 128,
2196                                   [VppRoutePath(self.pg3.remote_ip6,
2197                                                 self.pg3.sw_if_index)])
2198         route_5000_2.add_vpp_config()
2199
2200         route_6000_1 = VppIpRoute(self, "6000::1", 128,
2201                                   [VppRoutePath("5000::2",
2202                                                 0xffffffff)])
2203         route_6000_1.add_vpp_config()
2204
2205         #
2206         # inject the packet on pg0 - expect load-balancing across all 4 paths
2207         #
2208         self.vapi.cli("clear trace")
2209         self.send_and_expect_one_itf(self.pg0, port_pkts, self.pg3)
2210
2211
2212 class IP6PuntSetup(object):
2213     """ Setup for IPv6 Punt Police/Redirect """
2214
2215     def punt_setup(self):
2216         self.create_pg_interfaces(range(4))
2217
2218         for i in self.pg_interfaces:
2219             i.admin_up()
2220             i.config_ip6()
2221             i.resolve_ndp()
2222
2223         self.pkt = (Ether(src=self.pg0.remote_mac,
2224                           dst=self.pg0.local_mac) /
2225                     IPv6(src=self.pg0.remote_ip6,
2226                          dst=self.pg0.local_ip6) /
2227                     inet6.TCP(sport=1234, dport=1234) /
2228                     Raw(b'\xa5' * 100))
2229
2230     def punt_teardown(self):
2231         for i in self.pg_interfaces:
2232             i.unconfig_ip6()
2233             i.admin_down()
2234
2235
2236 class TestIP6Punt(IP6PuntSetup, VppTestCase):
2237     """ IPv6 Punt Police/Redirect """
2238
2239     def setUp(self):
2240         super(TestIP6Punt, self).setUp()
2241         super(TestIP6Punt, self).punt_setup()
2242
2243     def tearDown(self):
2244         super(TestIP6Punt, self).punt_teardown()
2245         super(TestIP6Punt, self).tearDown()
2246
2247     def test_ip_punt(self):
2248         """ IP6 punt police and redirect """
2249
2250         pkts = self.pkt * 1025
2251
2252         #
2253         # Configure a punt redirect via pg1.
2254         #
2255         nh_addr = self.pg1.remote_ip6
2256         ip_punt_redirect = VppIpPuntRedirect(self, self.pg0.sw_if_index,
2257                                              self.pg1.sw_if_index, nh_addr)
2258         ip_punt_redirect.add_vpp_config()
2259
2260         self.send_and_expect(self.pg0, pkts, self.pg1)
2261
2262         #
2263         # add a policer
2264         #
2265         policer = VppPolicer(self, "ip6-punt", 400, 0, 10, 0, rate_type=1)
2266         policer.add_vpp_config()
2267         ip_punt_policer = VppIpPuntPolicer(self, policer.policer_index,
2268                                            is_ip6=True)
2269         ip_punt_policer.add_vpp_config()
2270
2271         self.vapi.cli("clear trace")
2272         self.pg0.add_stream(pkts)
2273         self.pg_enable_capture(self.pg_interfaces)
2274         self.pg_start()
2275
2276         #
2277         # the number of packet received should be greater than 0,
2278         # but not equal to the number sent, since some were policed
2279         #
2280         rx = self.pg1._get_capture(1)
2281         stats = policer.get_stats()
2282
2283         # Single rate policer - expect conform, violate but no exceed
2284         self.assertGreater(stats['conform_packets'], 0)
2285         self.assertEqual(stats['exceed_packets'], 0)
2286         self.assertGreater(stats['violate_packets'], 0)
2287
2288         self.assertGreater(len(rx), 0)
2289         self.assertLess(len(rx), len(pkts))
2290
2291         #
2292         # remove the policer. back to full rx
2293         #
2294         ip_punt_policer.remove_vpp_config()
2295         policer.remove_vpp_config()
2296         self.send_and_expect(self.pg0, pkts, self.pg1)
2297
2298         #
2299         # remove the redirect. expect full drop.
2300         #
2301         ip_punt_redirect.remove_vpp_config()
2302         self.send_and_assert_no_replies(self.pg0, pkts,
2303                                         "IP no punt config")
2304
2305         #
2306         # Add a redirect that is not input port selective
2307         #
2308         ip_punt_redirect = VppIpPuntRedirect(self, 0xffffffff,
2309                                              self.pg1.sw_if_index, nh_addr)
2310         ip_punt_redirect.add_vpp_config()
2311         self.send_and_expect(self.pg0, pkts, self.pg1)
2312         ip_punt_redirect.remove_vpp_config()
2313
2314     def test_ip_punt_dump(self):
2315         """ IP6 punt redirect dump"""
2316
2317         #
2318         # Configure a punt redirects
2319         #
2320         nh_address = self.pg3.remote_ip6
2321         ipr_03 = VppIpPuntRedirect(self, self.pg0.sw_if_index,
2322                                    self.pg3.sw_if_index, nh_address)
2323         ipr_13 = VppIpPuntRedirect(self, self.pg1.sw_if_index,
2324                                    self.pg3.sw_if_index, nh_address)
2325         ipr_23 = VppIpPuntRedirect(self, self.pg2.sw_if_index,
2326                                    self.pg3.sw_if_index, '0::0')
2327         ipr_03.add_vpp_config()
2328         ipr_13.add_vpp_config()
2329         ipr_23.add_vpp_config()
2330
2331         #
2332         # Dump pg0 punt redirects
2333         #
2334         self.assertTrue(ipr_03.query_vpp_config())
2335         self.assertTrue(ipr_13.query_vpp_config())
2336         self.assertTrue(ipr_23.query_vpp_config())
2337
2338         #
2339         # Dump punt redirects for all interfaces
2340         #
2341         punts = self.vapi.ip_punt_redirect_dump(0xffffffff, is_ipv6=1)
2342         self.assertEqual(len(punts), 3)
2343         for p in punts:
2344             self.assertEqual(p.punt.tx_sw_if_index, self.pg3.sw_if_index)
2345         self.assertNotEqual(punts[1].punt.nh, self.pg3.remote_ip6)
2346         self.assertEqual(str(punts[2].punt.nh), '::')
2347
2348
2349 class TestIP6PuntHandoff(IP6PuntSetup, VppTestCase):
2350     """ IPv6 Punt Police/Redirect """
2351     vpp_worker_count = 2
2352
2353     def setUp(self):
2354         super(TestIP6PuntHandoff, self).setUp()
2355         super(TestIP6PuntHandoff, self).punt_setup()
2356
2357     def tearDown(self):
2358         super(TestIP6PuntHandoff, self).punt_teardown()
2359         super(TestIP6PuntHandoff, self).tearDown()
2360
2361     def test_ip_punt(self):
2362         """ IP6 punt policer thread handoff """
2363         pkts = self.pkt * NUM_PKTS
2364
2365         #
2366         # Configure a punt redirect via pg1.
2367         #
2368         nh_addr = self.pg1.remote_ip6
2369         ip_punt_redirect = VppIpPuntRedirect(self, self.pg0.sw_if_index,
2370                                              self.pg1.sw_if_index, nh_addr)
2371         ip_punt_redirect.add_vpp_config()
2372
2373         action_tx = PolicerAction(
2374             VppEnum.vl_api_sse2_qos_action_type_t.SSE2_QOS_ACTION_API_TRANSMIT,
2375             0)
2376         #
2377         # This policer drops no packets, we are just
2378         # testing that they get to the right thread.
2379         #
2380         policer = VppPolicer(self, "ip6-punt", 400, 0, 10, 0, 1,
2381                              0, 0, False, action_tx, action_tx, action_tx)
2382         policer.add_vpp_config()
2383         ip_punt_policer = VppIpPuntPolicer(self, policer.policer_index,
2384                                            is_ip6=True)
2385         ip_punt_policer.add_vpp_config()
2386
2387         for worker in [0, 1]:
2388             self.send_and_expect(self.pg0, pkts, self.pg1, worker=worker)
2389             if worker == 0:
2390                 self.logger.debug(self.vapi.cli("show trace max 100"))
2391
2392         # Combined stats, all threads
2393         stats = policer.get_stats()
2394
2395         # Single rate policer - expect conform, violate but no exceed
2396         self.assertGreater(stats['conform_packets'], 0)
2397         self.assertEqual(stats['exceed_packets'], 0)
2398         self.assertGreater(stats['violate_packets'], 0)
2399
2400         # Worker 0, should have done all the policing
2401         stats0 = policer.get_stats(worker=0)
2402         self.assertEqual(stats, stats0)
2403
2404         # Worker 1, should have handed everything off
2405         stats1 = policer.get_stats(worker=1)
2406         self.assertEqual(stats1['conform_packets'], 0)
2407         self.assertEqual(stats1['exceed_packets'], 0)
2408         self.assertEqual(stats1['violate_packets'], 0)
2409
2410         # Bind the policer to worker 1 and repeat
2411         policer.bind_vpp_config(1, True)
2412         for worker in [0, 1]:
2413             self.send_and_expect(self.pg0, pkts, self.pg1, worker=worker)
2414             self.logger.debug(self.vapi.cli("show trace max 100"))
2415
2416         # The 2 workers should now have policed the same amount
2417         stats = policer.get_stats()
2418         stats0 = policer.get_stats(worker=0)
2419         stats1 = policer.get_stats(worker=1)
2420
2421         self.assertGreater(stats0['conform_packets'], 0)
2422         self.assertEqual(stats0['exceed_packets'], 0)
2423         self.assertGreater(stats0['violate_packets'], 0)
2424
2425         self.assertGreater(stats1['conform_packets'], 0)
2426         self.assertEqual(stats1['exceed_packets'], 0)
2427         self.assertGreater(stats1['violate_packets'], 0)
2428
2429         self.assertEqual(stats0['conform_packets'] + stats1['conform_packets'],
2430                          stats['conform_packets'])
2431
2432         self.assertEqual(stats0['violate_packets'] + stats1['violate_packets'],
2433                          stats['violate_packets'])
2434
2435         # Unbind the policer and repeat
2436         policer.bind_vpp_config(1, False)
2437         for worker in [0, 1]:
2438             self.send_and_expect(self.pg0, pkts, self.pg1, worker=worker)
2439             self.logger.debug(self.vapi.cli("show trace max 100"))
2440
2441         # The policer should auto-bind to worker 0 when packets arrive
2442         stats = policer.get_stats()
2443         stats0new = policer.get_stats(worker=0)
2444         stats1new = policer.get_stats(worker=1)
2445
2446         self.assertGreater(stats0new['conform_packets'],
2447                            stats0['conform_packets'])
2448         self.assertEqual(stats0new['exceed_packets'], 0)
2449         self.assertGreater(stats0new['violate_packets'],
2450                            stats0['violate_packets'])
2451
2452         self.assertEqual(stats1, stats1new)
2453
2454         #
2455         # Clean up
2456         #
2457         ip_punt_policer.remove_vpp_config()
2458         policer.remove_vpp_config()
2459         ip_punt_redirect.remove_vpp_config()
2460
2461
2462 class TestIPDeag(VppTestCase):
2463     """ IPv6 Deaggregate Routes """
2464
2465     @classmethod
2466     def setUpClass(cls):
2467         super(TestIPDeag, cls).setUpClass()
2468
2469     @classmethod
2470     def tearDownClass(cls):
2471         super(TestIPDeag, cls).tearDownClass()
2472
2473     def setUp(self):
2474         super(TestIPDeag, self).setUp()
2475
2476         self.create_pg_interfaces(range(3))
2477
2478         for i in self.pg_interfaces:
2479             i.admin_up()
2480             i.config_ip6()
2481             i.resolve_ndp()
2482
2483     def tearDown(self):
2484         super(TestIPDeag, self).tearDown()
2485         for i in self.pg_interfaces:
2486             i.unconfig_ip6()
2487             i.admin_down()
2488
2489     def test_ip_deag(self):
2490         """ IP Deag Routes """
2491
2492         #
2493         # Create a table to be used for:
2494         #  1 - another destination address lookup
2495         #  2 - a source address lookup
2496         #
2497         table_dst = VppIpTable(self, 1, is_ip6=1)
2498         table_src = VppIpTable(self, 2, is_ip6=1)
2499         table_dst.add_vpp_config()
2500         table_src.add_vpp_config()
2501
2502         #
2503         # Add a route in the default table to point to a deag/
2504         # second lookup in each of these tables
2505         #
2506         route_to_dst = VppIpRoute(self, "1::1", 128,
2507                                   [VppRoutePath("::",
2508                                                 0xffffffff,
2509                                                 nh_table_id=1)])
2510         route_to_src = VppIpRoute(
2511             self, "1::2", 128,
2512             [VppRoutePath("::",
2513                           0xffffffff,
2514                           nh_table_id=2,
2515                           type=FibPathType.FIB_PATH_TYPE_SOURCE_LOOKUP)])
2516
2517         route_to_dst.add_vpp_config()
2518         route_to_src.add_vpp_config()
2519
2520         #
2521         # packets to these destination are dropped, since they'll
2522         # hit the respective default routes in the second table
2523         #
2524         p_dst = (Ether(src=self.pg0.remote_mac,
2525                        dst=self.pg0.local_mac) /
2526                  IPv6(src="5::5", dst="1::1") /
2527                  inet6.TCP(sport=1234, dport=1234) /
2528                  Raw(b'\xa5' * 100))
2529         p_src = (Ether(src=self.pg0.remote_mac,
2530                        dst=self.pg0.local_mac) /
2531                  IPv6(src="2::2", dst="1::2") /
2532                  inet6.TCP(sport=1234, dport=1234) /
2533                  Raw(b'\xa5' * 100))
2534         pkts_dst = p_dst * 257
2535         pkts_src = p_src * 257
2536
2537         self.send_and_assert_no_replies(self.pg0, pkts_dst,
2538                                         "IP in dst table")
2539         self.send_and_assert_no_replies(self.pg0, pkts_src,
2540                                         "IP in src table")
2541
2542         #
2543         # add a route in the dst table to forward via pg1
2544         #
2545         route_in_dst = VppIpRoute(self, "1::1", 128,
2546                                   [VppRoutePath(self.pg1.remote_ip6,
2547                                                 self.pg1.sw_if_index)],
2548                                   table_id=1)
2549         route_in_dst.add_vpp_config()
2550
2551         self.send_and_expect(self.pg0, pkts_dst, self.pg1)
2552
2553         #
2554         # add a route in the src table to forward via pg2
2555         #
2556         route_in_src = VppIpRoute(self, "2::2", 128,
2557                                   [VppRoutePath(self.pg2.remote_ip6,
2558                                                 self.pg2.sw_if_index)],
2559                                   table_id=2)
2560         route_in_src.add_vpp_config()
2561         self.send_and_expect(self.pg0, pkts_src, self.pg2)
2562
2563         #
2564         # loop in the lookup DP
2565         #
2566         route_loop = VppIpRoute(self, "3::3", 128,
2567                                 [VppRoutePath("::",
2568                                               0xffffffff)])
2569         route_loop.add_vpp_config()
2570
2571         p_l = (Ether(src=self.pg0.remote_mac,
2572                      dst=self.pg0.local_mac) /
2573                IPv6(src="3::4", dst="3::3") /
2574                inet6.TCP(sport=1234, dport=1234) /
2575                Raw(b'\xa5' * 100))
2576
2577         self.send_and_assert_no_replies(self.pg0, p_l * 257,
2578                                         "IP lookup loop")
2579
2580
2581 class TestIP6Input(VppTestCase):
2582     """ IPv6 Input Exception Test Cases """
2583
2584     @classmethod
2585     def setUpClass(cls):
2586         super(TestIP6Input, cls).setUpClass()
2587
2588     @classmethod
2589     def tearDownClass(cls):
2590         super(TestIP6Input, cls).tearDownClass()
2591
2592     def setUp(self):
2593         super(TestIP6Input, self).setUp()
2594
2595         self.create_pg_interfaces(range(2))
2596
2597         for i in self.pg_interfaces:
2598             i.admin_up()
2599             i.config_ip6()
2600             i.resolve_ndp()
2601
2602     def tearDown(self):
2603         super(TestIP6Input, self).tearDown()
2604         for i in self.pg_interfaces:
2605             i.unconfig_ip6()
2606             i.admin_down()
2607
2608     def test_ip_input_icmp_reply(self):
2609         """ IP6 Input Exception - Return ICMP (3,0) """
2610         #
2611         # hop limit - ICMP replies
2612         #
2613         p_version = (Ether(src=self.pg0.remote_mac,
2614                            dst=self.pg0.local_mac) /
2615                      IPv6(src=self.pg0.remote_ip6,
2616                           dst=self.pg1.remote_ip6,
2617                           hlim=1) /
2618                      inet6.UDP(sport=1234, dport=1234) /
2619                      Raw(b'\xa5' * 100))
2620
2621         rx = self.send_and_expect(self.pg0, p_version * NUM_PKTS, self.pg0)
2622         rx = rx[0]
2623         icmp = rx[ICMPv6TimeExceeded]
2624
2625         # 0: "hop limit exceeded in transit",
2626         self.assertEqual((icmp.type, icmp.code), (3, 0))
2627
2628     icmpv6_data = '\x0a' * 18
2629     all_0s = "::"
2630     all_1s = "FFFF:FFFF:FFFF:FFFF:FFFF:FFFF:FFFF:FFFF"
2631
2632     @parameterized.expand([
2633         # Name, src, dst, l4proto, msg, timeout
2634         ("src='iface',   dst='iface'", None, None,
2635          inet6.UDP(sport=1234, dport=1234), "funky version", None),
2636         ("src='All 0's', dst='iface'", all_0s, None,
2637          ICMPv6EchoRequest(id=0xb, seq=5, data=icmpv6_data), None, 0.1),
2638         ("src='iface',   dst='All 0's'", None, all_0s,
2639          ICMPv6EchoRequest(id=0xb, seq=5, data=icmpv6_data), None, 0.1),
2640         ("src='All 1's', dst='iface'", all_1s, None,
2641          ICMPv6EchoRequest(id=0xb, seq=5, data=icmpv6_data), None, 0.1),
2642         ("src='iface',   dst='All 1's'", None, all_1s,
2643          ICMPv6EchoRequest(id=0xb, seq=5, data=icmpv6_data), None, 0.1),
2644         ("src='All 1's', dst='All 1's'", all_1s, all_1s,
2645          ICMPv6EchoRequest(id=0xb, seq=5, data=icmpv6_data), None, 0.1),
2646
2647     ])
2648     def test_ip_input_no_replies(self, name, src, dst, l4, msg, timeout):
2649
2650         self._testMethodDoc = 'IPv6 Input Exception - %s' % name
2651
2652         p_version = (Ether(src=self.pg0.remote_mac,
2653                            dst=self.pg0.local_mac) /
2654                      IPv6(src=src or self.pg0.remote_ip6,
2655                           dst=dst or self.pg1.remote_ip6,
2656                           version=3) /
2657                      l4 /
2658                      Raw(b'\xa5' * 100))
2659
2660         self.send_and_assert_no_replies(self.pg0, p_version * NUM_PKTS,
2661                                         remark=msg or "",
2662                                         timeout=timeout)
2663
2664     def test_hop_by_hop(self):
2665         """ Hop-by-hop header test """
2666
2667         p = (Ether(src=self.pg0.remote_mac,
2668                    dst=self.pg0.local_mac) /
2669              IPv6(src=self.pg0.remote_ip6, dst=self.pg0.local_ip6) /
2670              IPv6ExtHdrHopByHop() /
2671              inet6.UDP(sport=1234, dport=1234) /
2672              Raw(b'\xa5' * 100))
2673
2674         self.pg0.add_stream(p)
2675         self.pg_enable_capture(self.pg_interfaces)
2676         self.pg_start()
2677
2678
2679 class TestIPReplace(VppTestCase):
2680     """ IPv6 Table Replace """
2681
2682     @classmethod
2683     def setUpClass(cls):
2684         super(TestIPReplace, cls).setUpClass()
2685
2686     @classmethod
2687     def tearDownClass(cls):
2688         super(TestIPReplace, cls).tearDownClass()
2689
2690     def setUp(self):
2691         super(TestIPReplace, self).setUp()
2692
2693         self.create_pg_interfaces(range(4))
2694
2695         table_id = 1
2696         self.tables = []
2697
2698         for i in self.pg_interfaces:
2699             i.admin_up()
2700             i.config_ip6()
2701             i.generate_remote_hosts(2)
2702             self.tables.append(VppIpTable(self, table_id,
2703                                           True).add_vpp_config())
2704             table_id += 1
2705
2706     def tearDown(self):
2707         super(TestIPReplace, self).tearDown()
2708         for i in self.pg_interfaces:
2709             i.admin_down()
2710             i.unconfig_ip6()
2711
2712     def test_replace(self):
2713         """ IP Table Replace """
2714
2715         MRouteItfFlags = VppEnum.vl_api_mfib_itf_flags_t
2716         MRouteEntryFlags = VppEnum.vl_api_mfib_entry_flags_t
2717         N_ROUTES = 20
2718         links = [self.pg0, self.pg1, self.pg2, self.pg3]
2719         routes = [[], [], [], []]
2720
2721         # the sizes of 'empty' tables
2722         for t in self.tables:
2723             self.assertEqual(len(t.dump()), 2)
2724             self.assertEqual(len(t.mdump()), 5)
2725
2726         # load up the tables with some routes
2727         for ii, t in enumerate(self.tables):
2728             for jj in range(1, N_ROUTES):
2729                 uni = VppIpRoute(
2730                     self, "2001::%d" % jj if jj != 0 else "2001::", 128,
2731                     [VppRoutePath(links[ii].remote_hosts[0].ip6,
2732                                   links[ii].sw_if_index),
2733                      VppRoutePath(links[ii].remote_hosts[1].ip6,
2734                                   links[ii].sw_if_index)],
2735                     table_id=t.table_id).add_vpp_config()
2736                 multi = VppIpMRoute(
2737                     self, "::",
2738                     "ff:2001::%d" % jj, 128,
2739                     MRouteEntryFlags.MFIB_API_ENTRY_FLAG_NONE,
2740                     [VppMRoutePath(self.pg0.sw_if_index,
2741                                    MRouteItfFlags.MFIB_API_ITF_FLAG_ACCEPT,
2742                                    proto=FibPathProto.FIB_PATH_NH_PROTO_IP6),
2743                      VppMRoutePath(self.pg1.sw_if_index,
2744                                    MRouteItfFlags.MFIB_API_ITF_FLAG_FORWARD,
2745                                    proto=FibPathProto.FIB_PATH_NH_PROTO_IP6),
2746                      VppMRoutePath(self.pg2.sw_if_index,
2747                                    MRouteItfFlags.MFIB_API_ITF_FLAG_FORWARD,
2748                                    proto=FibPathProto.FIB_PATH_NH_PROTO_IP6),
2749                      VppMRoutePath(self.pg3.sw_if_index,
2750                                    MRouteItfFlags.MFIB_API_ITF_FLAG_FORWARD,
2751                                    proto=FibPathProto.FIB_PATH_NH_PROTO_IP6)],
2752                     table_id=t.table_id).add_vpp_config()
2753                 routes[ii].append({'uni': uni,
2754                                    'multi': multi})
2755
2756         #
2757         # replace the tables a few times
2758         #
2759         for kk in range(3):
2760             # replace each table
2761             for t in self.tables:
2762                 t.replace_begin()
2763
2764             # all the routes are still there
2765             for ii, t in enumerate(self.tables):
2766                 dump = t.dump()
2767                 mdump = t.mdump()
2768                 for r in routes[ii]:
2769                     self.assertTrue(find_route_in_dump(dump, r['uni'], t))
2770                     self.assertTrue(find_mroute_in_dump(mdump, r['multi'], t))
2771
2772             # redownload the even numbered routes
2773             for ii, t in enumerate(self.tables):
2774                 for jj in range(0, N_ROUTES, 2):
2775                     routes[ii][jj]['uni'].add_vpp_config()
2776                     routes[ii][jj]['multi'].add_vpp_config()
2777
2778             # signal each table converged
2779             for t in self.tables:
2780                 t.replace_end()
2781
2782             # we should find the even routes, but not the odd
2783             for ii, t in enumerate(self.tables):
2784                 dump = t.dump()
2785                 mdump = t.mdump()
2786                 for jj in range(0, N_ROUTES, 2):
2787                     self.assertTrue(find_route_in_dump(
2788                         dump, routes[ii][jj]['uni'], t))
2789                     self.assertTrue(find_mroute_in_dump(
2790                         mdump, routes[ii][jj]['multi'], t))
2791                 for jj in range(1, N_ROUTES - 1, 2):
2792                     self.assertFalse(find_route_in_dump(
2793                         dump, routes[ii][jj]['uni'], t))
2794                     self.assertFalse(find_mroute_in_dump(
2795                         mdump, routes[ii][jj]['multi'], t))
2796
2797             # reload all the routes
2798             for ii, t in enumerate(self.tables):
2799                 for r in routes[ii]:
2800                     r['uni'].add_vpp_config()
2801                     r['multi'].add_vpp_config()
2802
2803             # all the routes are still there
2804             for ii, t in enumerate(self.tables):
2805                 dump = t.dump()
2806                 mdump = t.mdump()
2807                 for r in routes[ii]:
2808                     self.assertTrue(find_route_in_dump(dump, r['uni'], t))
2809                     self.assertTrue(find_mroute_in_dump(mdump, r['multi'], t))
2810
2811         #
2812         # finally flush the tables for good measure
2813         #
2814         for t in self.tables:
2815             t.flush()
2816             self.assertEqual(len(t.dump()), 2)
2817             self.assertEqual(len(t.mdump()), 5)
2818
2819
2820 class TestIP6Replace(VppTestCase):
2821     """ IPv4 Interface Address Replace """
2822
2823     @classmethod
2824     def setUpClass(cls):
2825         super(TestIP6Replace, cls).setUpClass()
2826
2827     @classmethod
2828     def tearDownClass(cls):
2829         super(TestIP6Replace, cls).tearDownClass()
2830
2831     def setUp(self):
2832         super(TestIP6Replace, self).setUp()
2833
2834         self.create_pg_interfaces(range(4))
2835
2836         for i in self.pg_interfaces:
2837             i.admin_up()
2838
2839     def tearDown(self):
2840         super(TestIP6Replace, self).tearDown()
2841         for i in self.pg_interfaces:
2842             i.admin_down()
2843
2844     def get_n_pfxs(self, intf):
2845         return len(self.vapi.ip_address_dump(intf.sw_if_index, True))
2846
2847     def test_replace(self):
2848         """ IP interface address replace """
2849
2850         intf_pfxs = [[], [], [], []]
2851
2852         # add prefixes to each of the interfaces
2853         for i in range(len(self.pg_interfaces)):
2854             intf = self.pg_interfaces[i]
2855
2856             # 2001:16:x::1/64
2857             addr = "2001:16:%d::1" % intf.sw_if_index
2858             a = VppIpInterfaceAddress(self, intf, addr, 64).add_vpp_config()
2859             intf_pfxs[i].append(a)
2860
2861             # 2001:16:x::2/64 - a different address in the same subnet as above
2862             addr = "2001:16:%d::2" % intf.sw_if_index
2863             a = VppIpInterfaceAddress(self, intf, addr, 64).add_vpp_config()
2864             intf_pfxs[i].append(a)
2865
2866             # 2001:15:x::2/64 - a different address and subnet
2867             addr = "2001:15:%d::2" % intf.sw_if_index
2868             a = VppIpInterfaceAddress(self, intf, addr, 64).add_vpp_config()
2869             intf_pfxs[i].append(a)
2870
2871         # a dump should n_address in it
2872         for intf in self.pg_interfaces:
2873             self.assertEqual(self.get_n_pfxs(intf), 3)
2874
2875         #
2876         # remove all the address thru a replace
2877         #
2878         self.vapi.sw_interface_address_replace_begin()
2879         self.vapi.sw_interface_address_replace_end()
2880         for intf in self.pg_interfaces:
2881             self.assertEqual(self.get_n_pfxs(intf), 0)
2882
2883         #
2884         # add all the interface addresses back
2885         #
2886         for p in intf_pfxs:
2887             for v in p:
2888                 v.add_vpp_config()
2889         for intf in self.pg_interfaces:
2890             self.assertEqual(self.get_n_pfxs(intf), 3)
2891
2892         #
2893         # replace again, but this time update/re-add the address on the first
2894         # two interfaces
2895         #
2896         self.vapi.sw_interface_address_replace_begin()
2897
2898         for p in intf_pfxs[:2]:
2899             for v in p:
2900                 v.add_vpp_config()
2901
2902         self.vapi.sw_interface_address_replace_end()
2903
2904         # on the first two the address still exist,
2905         # on the other two they do not
2906         for intf in self.pg_interfaces[:2]:
2907             self.assertEqual(self.get_n_pfxs(intf), 3)
2908         for p in intf_pfxs[:2]:
2909             for v in p:
2910                 self.assertTrue(v.query_vpp_config())
2911         for intf in self.pg_interfaces[2:]:
2912             self.assertEqual(self.get_n_pfxs(intf), 0)
2913
2914         #
2915         # add all the interface addresses back on the last two
2916         #
2917         for p in intf_pfxs[2:]:
2918             for v in p:
2919                 v.add_vpp_config()
2920         for intf in self.pg_interfaces:
2921             self.assertEqual(self.get_n_pfxs(intf), 3)
2922
2923         #
2924         # replace again, this time add different prefixes on all the interfaces
2925         #
2926         self.vapi.sw_interface_address_replace_begin()
2927
2928         pfxs = []
2929         for intf in self.pg_interfaces:
2930             # 2001:18:x::1/64
2931             addr = "2001:18:%d::1" % intf.sw_if_index
2932             pfxs.append(VppIpInterfaceAddress(self, intf, addr,
2933                                               64).add_vpp_config())
2934
2935         self.vapi.sw_interface_address_replace_end()
2936
2937         # only .18 should exist on each interface
2938         for intf in self.pg_interfaces:
2939             self.assertEqual(self.get_n_pfxs(intf), 1)
2940         for pfx in pfxs:
2941             self.assertTrue(pfx.query_vpp_config())
2942
2943         #
2944         # remove everything
2945         #
2946         self.vapi.sw_interface_address_replace_begin()
2947         self.vapi.sw_interface_address_replace_end()
2948         for intf in self.pg_interfaces:
2949             self.assertEqual(self.get_n_pfxs(intf), 0)
2950
2951         #
2952         # add prefixes to each interface. post-begin add the prefix from
2953         # interface X onto interface Y. this would normally be an error
2954         # since it would generate a 'duplicate address' warning. but in
2955         # this case, since what is newly downloaded is sane, it's ok
2956         #
2957         for intf in self.pg_interfaces:
2958             # 2001:18:x::1/64
2959             addr = "2001:18:%d::1" % intf.sw_if_index
2960             VppIpInterfaceAddress(self, intf, addr, 64).add_vpp_config()
2961
2962         self.vapi.sw_interface_address_replace_begin()
2963
2964         pfxs = []
2965         for intf in self.pg_interfaces:
2966             # 2001:18:x::1/64
2967             addr = "2001:18:%d::1" % (intf.sw_if_index + 1)
2968             pfxs.append(VppIpInterfaceAddress(self, intf,
2969                                               addr, 64).add_vpp_config())
2970
2971         self.vapi.sw_interface_address_replace_end()
2972
2973         self.logger.info(self.vapi.cli("sh int addr"))
2974
2975         for intf in self.pg_interfaces:
2976             self.assertEqual(self.get_n_pfxs(intf), 1)
2977         for pfx in pfxs:
2978             self.assertTrue(pfx.query_vpp_config())
2979
2980
2981 class TestIP6LinkLocal(VppTestCase):
2982     """ IPv6 Link Local """
2983
2984     @classmethod
2985     def setUpClass(cls):
2986         super(TestIP6LinkLocal, cls).setUpClass()
2987
2988     @classmethod
2989     def tearDownClass(cls):
2990         super(TestIP6LinkLocal, cls).tearDownClass()
2991
2992     def setUp(self):
2993         super(TestIP6LinkLocal, self).setUp()
2994
2995         self.create_pg_interfaces(range(2))
2996
2997         for i in self.pg_interfaces:
2998             i.admin_up()
2999
3000     def tearDown(self):
3001         super(TestIP6LinkLocal, self).tearDown()
3002         for i in self.pg_interfaces:
3003             i.admin_down()
3004
3005     def test_ip6_ll(self):
3006         """ IPv6 Link Local """
3007
3008         #
3009         # two APIs to add a link local address.
3010         #   1 - just like any other prefix
3011         #   2 - with the special set LL API
3012         #
3013
3014         #
3015         # First with the API to set a 'normal' prefix
3016         #
3017         ll1 = "fe80:1::1"
3018         ll2 = "fe80:2::2"
3019         ll3 = "fe80:3::3"
3020
3021         VppNeighbor(self,
3022                     self.pg0.sw_if_index,
3023                     self.pg0.remote_mac,
3024                     ll2).add_vpp_config()
3025
3026         VppIpInterfaceAddress(self, self.pg0, ll1, 128).add_vpp_config()
3027
3028         #
3029         # should be able to ping the ll
3030         #
3031         p_echo_request_1 = (Ether(src=self.pg0.remote_mac,
3032                                   dst=self.pg0.local_mac) /
3033                             IPv6(src=ll2,
3034                                  dst=ll1) /
3035                             ICMPv6EchoRequest())
3036
3037         self.send_and_expect(self.pg0, [p_echo_request_1], self.pg0)
3038
3039         #
3040         # change the link-local on pg0
3041         #
3042         v_ll3 = VppIpInterfaceAddress(self, self.pg0,
3043                                       ll3, 128).add_vpp_config()
3044
3045         p_echo_request_3 = (Ether(src=self.pg0.remote_mac,
3046                                   dst=self.pg0.local_mac) /
3047                             IPv6(src=ll2,
3048                                  dst=ll3) /
3049                             ICMPv6EchoRequest())
3050
3051         self.send_and_expect(self.pg0, [p_echo_request_3], self.pg0)
3052
3053         #
3054         # set a normal v6 prefix on the link
3055         #
3056         self.pg0.config_ip6()
3057
3058         self.send_and_expect(self.pg0, [p_echo_request_3], self.pg0)
3059
3060         # the link-local cannot be removed
3061         with self.vapi.assert_negative_api_retval():
3062             v_ll3.remove_vpp_config()
3063
3064         #
3065         # Use the specific link-local API on pg1
3066         #
3067         VppIp6LinkLocalAddress(self, self.pg1, ll1).add_vpp_config()
3068         self.send_and_expect(self.pg1, [p_echo_request_1], self.pg1)
3069
3070         VppIp6LinkLocalAddress(self, self.pg1, ll3).add_vpp_config()
3071         self.send_and_expect(self.pg1, [p_echo_request_3], self.pg1)
3072
3073     def test_ip6_ll_p2p(self):
3074         """ IPv6 Link Local P2P (GRE)"""
3075
3076         self.pg0.config_ip4()
3077         self.pg0.resolve_arp()
3078         gre_if = VppGreInterface(self,
3079                                  self.pg0.local_ip4,
3080                                  self.pg0.remote_ip4).add_vpp_config()
3081         gre_if.admin_up()
3082
3083         ll1 = "fe80:1::1"
3084         ll2 = "fe80:2::2"
3085
3086         VppIpInterfaceAddress(self, gre_if, ll1, 128).add_vpp_config()
3087
3088         self.logger.info(self.vapi.cli("sh ip6-ll gre0 fe80:2::2"))
3089
3090         p_echo_request_1 = (Ether(src=self.pg0.remote_mac,
3091                                   dst=self.pg0.local_mac) /
3092                             IP(src=self.pg0.remote_ip4,
3093                                dst=self.pg0.local_ip4) /
3094                             GRE() /
3095                             IPv6(src=ll2, dst=ll1) /
3096                             ICMPv6EchoRequest())
3097         self.send_and_expect(self.pg0, [p_echo_request_1], self.pg0)
3098
3099         self.pg0.unconfig_ip4()
3100         gre_if.remove_vpp_config()
3101
3102     def test_ip6_ll_p2mp(self):
3103         """ IPv6 Link Local P2MP (GRE)"""
3104
3105         self.pg0.config_ip4()
3106         self.pg0.resolve_arp()
3107
3108         gre_if = VppGreInterface(
3109             self,
3110             self.pg0.local_ip4,
3111             "0.0.0.0",
3112             mode=(VppEnum.vl_api_tunnel_mode_t.
3113                   TUNNEL_API_MODE_MP)).add_vpp_config()
3114         gre_if.admin_up()
3115
3116         ll1 = "fe80:1::1"
3117         ll2 = "fe80:2::2"
3118
3119         VppIpInterfaceAddress(self, gre_if, ll1, 128).add_vpp_config()
3120
3121         p_echo_request_1 = (Ether(src=self.pg0.remote_mac,
3122                                   dst=self.pg0.local_mac) /
3123                             IP(src=self.pg0.remote_ip4,
3124                                dst=self.pg0.local_ip4) /
3125                             GRE() /
3126                             IPv6(src=ll2, dst=ll1) /
3127                             ICMPv6EchoRequest())
3128
3129         # no route back at this point
3130         self.send_and_assert_no_replies(self.pg0, [p_echo_request_1])
3131
3132         # add teib entry for the peer
3133         teib = VppTeib(self, gre_if, ll2, self.pg0.remote_ip4)
3134         teib.add_vpp_config()
3135
3136         self.logger.info(self.vapi.cli("sh ip6-ll gre0 %s" % ll2))
3137         self.send_and_expect(self.pg0, [p_echo_request_1], self.pg0)
3138
3139         # teardown
3140         self.pg0.unconfig_ip4()
3141
3142
3143 class TestIPv6PathMTU(VppTestCase):
3144     """ IPv6 Path MTU """
3145
3146     def setUp(self):
3147         super(TestIPv6PathMTU, self).setUp()
3148
3149         self.create_pg_interfaces(range(2))
3150
3151         # setup all interfaces
3152         for i in self.pg_interfaces:
3153             i.admin_up()
3154             i.config_ip6()
3155             i.resolve_ndp()
3156
3157     def tearDown(self):
3158         super(TestIPv6PathMTU, self).tearDown()
3159         for i in self.pg_interfaces:
3160             i.unconfig_ip6()
3161             i.admin_down()
3162
3163     def test_path_mtu_local(self):
3164         """ Path MTU for attached neighbour """
3165
3166         self.vapi.cli("set log class ip level debug")
3167         #
3168         # The goal here is not test that fragmentation works correctly,
3169         # that's done elsewhere, the intent is to ensure that the Path MTU
3170         # settings are honoured.
3171         #
3172
3173         #
3174         # IPv6 will only frag locally generated packets, so use tunnelled
3175         # packets post encap
3176         #
3177         tun = VppIpIpTunInterface(
3178             self,
3179             self.pg1,
3180             self.pg1.local_ip6,
3181             self.pg1.remote_ip6)
3182         tun.add_vpp_config()
3183         tun.admin_up()
3184         tun.config_ip6()
3185
3186         # set the interface MTU to a reasonable value
3187         self.vapi.sw_interface_set_mtu(self.pg1.sw_if_index,
3188                                        [2800, 0, 0, 0])
3189
3190         p_2k = (Ether(dst=self.pg0.local_mac,
3191                       src=self.pg0.remote_mac) /
3192                 IPv6(src=self.pg0.remote_ip6,
3193                      dst=tun.remote_ip6) /
3194                 UDP(sport=1234, dport=5678) /
3195                 Raw(b'0xa' * 1000))
3196         p_1k = (Ether(dst=self.pg0.local_mac,
3197                       src=self.pg0.remote_mac) /
3198                 IPv6(src=self.pg0.remote_ip6,
3199                      dst=tun.remote_ip6) /
3200                 UDP(sport=1234, dport=5678) /
3201                 Raw(b'0xa' * 600))
3202
3203         nbr = VppNeighbor(self,
3204                           self.pg1.sw_if_index,
3205                           self.pg1.remote_mac,
3206                           self.pg1.remote_ip6).add_vpp_config()
3207
3208         # this is now the interface MTU frags
3209         self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=2)
3210         self.send_and_expect(self.pg0, [p_1k], self.pg1)
3211
3212         # drop the path MTU for this neighbour to below the interface MTU
3213         # expect more frags
3214         pmtu = VppIpPathMtu(self, self.pg1.remote_ip6, 1300).add_vpp_config()
3215
3216         # print/format the adj delegate and trackers
3217         self.logger.info(self.vapi.cli("sh ip pmtu"))
3218         self.logger.info(self.vapi.cli("sh adj 7"))
3219
3220         self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=3)
3221         self.send_and_expect(self.pg0, [p_1k], self.pg1, n_rx=2)
3222
3223         # increase the path MTU to more than the interface
3224         # expect to use the interface MTU
3225         pmtu.modify(8192)
3226
3227         self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=2)
3228         self.send_and_expect(self.pg0, [p_1k], self.pg1)
3229
3230         # go back to an MTU from the path
3231         pmtu.modify(1300)
3232
3233         self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=3)
3234         self.send_and_expect(self.pg0, [p_1k], self.pg1, n_rx=2)
3235
3236         # raise the interface's MTU
3237         # should still use that of the path
3238         self.vapi.sw_interface_set_mtu(self.pg1.sw_if_index,
3239                                        [2000, 0, 0, 0])
3240         self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=3)
3241         self.send_and_expect(self.pg0, [p_1k], self.pg1, n_rx=2)
3242
3243         # set path high and interface low
3244         pmtu.modify(2000)
3245         self.vapi.sw_interface_set_mtu(self.pg1.sw_if_index,
3246                                        [1300, 0, 0, 0])
3247         self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=3)
3248         self.send_and_expect(self.pg0, [p_1k], self.pg1, n_rx=2)
3249
3250         # remove the path MTU
3251         self.vapi.sw_interface_set_mtu(self.pg1.sw_if_index,
3252                                        [2800, 0, 0, 0])
3253         pmtu.modify(0)
3254
3255         self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=2)
3256         self.send_and_expect(self.pg0, [p_1k], self.pg1)
3257
3258     def test_path_mtu_remote(self):
3259         """ Path MTU for remote neighbour """
3260
3261         self.vapi.cli("set log class ip level debug")
3262         #
3263         # The goal here is not test that fragmentation works correctly,
3264         # that's done elsewhere, the intent is to ensure that the Path MTU
3265         # settings are honoured.
3266         #
3267         tun_dst = "2001::1"
3268
3269         route = VppIpRoute(
3270             self, tun_dst, 64,
3271             [VppRoutePath(self.pg1.remote_ip6,
3272                           self.pg1.sw_if_index)]).add_vpp_config()
3273
3274         #
3275         # IPv6 will only frag locally generated packets, so use tunnelled
3276         # packets post encap
3277         #
3278         tun = VppIpIpTunInterface(
3279             self,
3280             self.pg1,
3281             self.pg1.local_ip6,
3282             tun_dst)
3283         tun.add_vpp_config()
3284         tun.admin_up()
3285         tun.config_ip6()
3286
3287         # set the interface MTU to a reasonable value
3288         self.vapi.sw_interface_set_mtu(self.pg1.sw_if_index,
3289                                        [2800, 0, 0, 0])
3290
3291         p_2k = (Ether(dst=self.pg0.local_mac,
3292                       src=self.pg0.remote_mac) /
3293                 IPv6(src=self.pg0.remote_ip6,
3294                      dst=tun.remote_ip6) /
3295                 UDP(sport=1234, dport=5678) /
3296                 Raw(b'0xa' * 1000))
3297         p_1k = (Ether(dst=self.pg0.local_mac,
3298                       src=self.pg0.remote_mac) /
3299                 IPv6(src=self.pg0.remote_ip6,
3300                      dst=tun.remote_ip6) /
3301                 UDP(sport=1234, dport=5678) /
3302                 Raw(b'0xa' * 600))
3303
3304         nbr = VppNeighbor(self,
3305                           self.pg1.sw_if_index,
3306                           self.pg1.remote_mac,
3307                           self.pg1.remote_ip6).add_vpp_config()
3308
3309         # this is now the interface MTU frags
3310         self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=2)
3311         self.send_and_expect(self.pg0, [p_1k], self.pg1)
3312
3313         # drop the path MTU for this neighbour to below the interface MTU
3314         # expect more frags
3315         pmtu = VppIpPathMtu(self, tun_dst, 1300).add_vpp_config()
3316
3317         # print/format the fib entry/dpo
3318         self.logger.info(self.vapi.cli("sh ip6 fib 2001::1"))
3319
3320         self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=3)
3321         self.send_and_expect(self.pg0, [p_1k], self.pg1, n_rx=2)
3322
3323         # increase the path MTU to more than the interface
3324         # expect to use the interface MTU
3325         pmtu.modify(8192)
3326
3327         self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=2)
3328         self.send_and_expect(self.pg0, [p_1k], self.pg1)
3329
3330         # go back to an MTU from the path
3331         pmtu.modify(1300)
3332
3333         self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=3)
3334         self.send_and_expect(self.pg0, [p_1k], self.pg1, n_rx=2)
3335
3336         # raise the interface's MTU
3337         # should still use that of the path
3338         self.vapi.sw_interface_set_mtu(self.pg1.sw_if_index,
3339                                        [2000, 0, 0, 0])
3340         self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=3)
3341         self.send_and_expect(self.pg0, [p_1k], self.pg1, n_rx=2)
3342
3343         # turn the tun_dst into an attached neighbour
3344         route.modify([VppRoutePath("::",
3345                                    self.pg1.sw_if_index)])
3346         nbr2 = VppNeighbor(self,
3347                            self.pg1.sw_if_index,
3348                            self.pg1.remote_mac,
3349                            tun_dst).add_vpp_config()
3350
3351         self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=3)
3352         self.send_and_expect(self.pg0, [p_1k], self.pg1, n_rx=2)
3353
3354         # add back to not attached
3355         nbr2.remove_vpp_config()
3356         route.modify([VppRoutePath(self.pg1.remote_ip6,
3357                                    self.pg1.sw_if_index)])
3358
3359         # set path high and interface low
3360         pmtu.modify(2000)
3361         self.vapi.sw_interface_set_mtu(self.pg1.sw_if_index,
3362                                        [1300, 0, 0, 0])
3363         self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=3)
3364         self.send_and_expect(self.pg0, [p_1k], self.pg1, n_rx=2)
3365
3366         # remove the path MTU
3367         self.vapi.sw_interface_set_mtu(self.pg1.sw_if_index,
3368                                        [2800, 0, 0, 0])
3369         pmtu.remove_vpp_config()
3370         self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=2)
3371         self.send_and_expect(self.pg0, [p_1k], self.pg1)
3372
3373
3374 class TestIPFibSource(VppTestCase):
3375     """ IPv6 Table FibSource """
3376
3377     @classmethod
3378     def setUpClass(cls):
3379         super(TestIPFibSource, cls).setUpClass()
3380
3381     @classmethod
3382     def tearDownClass(cls):
3383         super(TestIPFibSource, cls).tearDownClass()
3384
3385     def setUp(self):
3386         super(TestIPFibSource, self).setUp()
3387
3388         self.create_pg_interfaces(range(2))
3389
3390         for i in self.pg_interfaces:
3391             i.admin_up()
3392             i.config_ip6()
3393             i.resolve_arp()
3394             i.generate_remote_hosts(2)
3395             i.configure_ipv6_neighbors()
3396
3397     def tearDown(self):
3398         super(TestIPFibSource, self).tearDown()
3399         for i in self.pg_interfaces:
3400             i.admin_down()
3401             i.unconfig_ip4()
3402
3403     def test_fib_source(self):
3404         """ IP Table FibSource """
3405
3406         routes = self.vapi.ip_route_v2_dump(0, True)
3407
3408         # 2 interfaces (4 routes) + 2 specials + 4 neighbours = 10 routes
3409         self.assertEqual(len(routes), 10)
3410
3411         # dump all the sources in the FIB
3412         sources = self.vapi.fib_source_dump()
3413         for source in sources:
3414             if (source.src.name == "API"):
3415                 api_source = source.src
3416             if (source.src.name == "interface"):
3417                 intf_source = source.src
3418             if (source.src.name == "adjacency"):
3419                 adj_source = source.src
3420             if (source.src.name == "special"):
3421                 special_source = source.src
3422             if (source.src.name == "default-route"):
3423                 dr_source = source.src
3424
3425         # dump the individual route types
3426         routes = self.vapi.ip_route_v2_dump(0, True, src=adj_source.id)
3427         self.assertEqual(len(routes), 4)
3428         routes = self.vapi.ip_route_v2_dump(0, True, src=intf_source.id)
3429         self.assertEqual(len(routes), 4)
3430         routes = self.vapi.ip_route_v2_dump(0, True, src=special_source.id)
3431         self.assertEqual(len(routes), 1)
3432         routes = self.vapi.ip_route_v2_dump(0, True, src=dr_source.id)
3433         self.assertEqual(len(routes), 1)
3434
3435         # add a new soure that'a better than the API
3436         self.vapi.fib_source_add(src={'name': "bgp",
3437                                       "priority": api_source.priority - 1})
3438
3439         # dump all the sources to check our new one is there
3440         sources = self.vapi.fib_source_dump()
3441
3442         for source in sources:
3443             if (source.src.name == "bgp"):
3444                 bgp_source = source.src
3445
3446         self.assertTrue(bgp_source)
3447         self.assertEqual(bgp_source.priority,
3448                          api_source.priority - 1)
3449
3450         # add a route with the default API source
3451         r1 = VppIpRouteV2(
3452             self, "2001::1", 128,
3453             [VppRoutePath(self.pg0.remote_ip6,
3454                           self.pg0.sw_if_index)]).add_vpp_config()
3455
3456         r2 = VppIpRouteV2(self, "2001::1", 128,
3457                           [VppRoutePath(self.pg1.remote_ip6,
3458                                         self.pg1.sw_if_index)],
3459                           src=bgp_source.id).add_vpp_config()
3460
3461         # ensure the BGP source takes priority
3462         p = (Ether(src=self.pg0.remote_mac,
3463                    dst=self.pg0.local_mac) /
3464              IPv6(src=self.pg0.remote_ip6, dst="2001::1") /
3465              inet6.UDP(sport=1234, dport=1234) /
3466              Raw(b'\xa5' * 100))
3467
3468         self.send_and_expect(self.pg0, [p], self.pg1)
3469
3470         r2.remove_vpp_config()
3471         r1.remove_vpp_config()
3472
3473         self.assertFalse(find_route(self, "2001::1", 128))
3474
3475
3476 class TestIPxAF(VppTestCase):
3477     """ IP cross AF """
3478
3479     @classmethod
3480     def setUpClass(cls):
3481         super(TestIPxAF, cls).setUpClass()
3482
3483     @classmethod
3484     def tearDownClass(cls):
3485         super(TestIPxAF, cls).tearDownClass()
3486
3487     def setUp(self):
3488         super(TestIPxAF, self).setUp()
3489
3490         self.create_pg_interfaces(range(2))
3491
3492         for i in self.pg_interfaces:
3493             i.admin_up()
3494             i.config_ip6()
3495             i.config_ip4()
3496             i.resolve_arp()
3497             i.resolve_ndp()
3498
3499     def tearDown(self):
3500         super(TestIPxAF, self).tearDown()
3501         for i in self.pg_interfaces:
3502             i.admin_down()
3503             i.unconfig_ip4()
3504             i.unconfig_ip6()
3505
3506     def test_x_af(self):
3507         """ Cross AF routing """
3508
3509         N_PKTS = 63
3510         # a v4 route via a v6 attached next-hop
3511         VppIpRoute(
3512             self, "1.1.1.1", 32,
3513             [VppRoutePath(self.pg1.remote_ip6,
3514                           self.pg1.sw_if_index)]).add_vpp_config()
3515
3516         p = (Ether(src=self.pg0.remote_mac,
3517                    dst=self.pg0.local_mac) /
3518              IP(src=self.pg0.remote_ip4, dst="1.1.1.1") /
3519              UDP(sport=1234, dport=1234) /
3520              Raw(b'\xa5' * 100))
3521         rxs = self.send_and_expect(self.pg0, p * N_PKTS, self.pg1)
3522
3523         for rx in rxs:
3524             self.assertEqual(rx[IP].dst, "1.1.1.1")
3525
3526         # a v6 route via a v4 attached next-hop
3527         VppIpRoute(
3528             self, "2001::1", 128,
3529             [VppRoutePath(self.pg1.remote_ip4,
3530                           self.pg1.sw_if_index)]).add_vpp_config()
3531
3532         p = (Ether(src=self.pg0.remote_mac,
3533                    dst=self.pg0.local_mac) /
3534              IPv6(src=self.pg0.remote_ip6, dst="2001::1") /
3535              UDP(sport=1234, dport=1234) /
3536              Raw(b'\xa5' * 100))
3537         rxs = self.send_and_expect(self.pg0, p * N_PKTS, self.pg1)
3538
3539         for rx in rxs:
3540             self.assertEqual(rx[IPv6].dst, "2001::1")
3541
3542         # a recursive v4 route via a v6 next-hop (from above)
3543         VppIpRoute(
3544             self, "2.2.2.2", 32,
3545             [VppRoutePath("2001::1",
3546                           0xffffffff)]).add_vpp_config()
3547
3548         p = (Ether(src=self.pg0.remote_mac,
3549                    dst=self.pg0.local_mac) /
3550              IP(src=self.pg0.remote_ip4, dst="2.2.2.2") /
3551              UDP(sport=1234, dport=1234) /
3552              Raw(b'\xa5' * 100))
3553         rxs = self.send_and_expect(self.pg0, p * N_PKTS, self.pg1)
3554
3555         # a recursive v4 route via a v6 next-hop
3556         VppIpRoute(
3557             self, "2.2.2.3", 32,
3558             [VppRoutePath(self.pg1.remote_ip6,
3559                           0xffffffff)]).add_vpp_config()
3560
3561         p = (Ether(src=self.pg0.remote_mac,
3562                    dst=self.pg0.local_mac) /
3563              IP(src=self.pg0.remote_ip4, dst="2.2.2.3") /
3564              UDP(sport=1234, dport=1234) /
3565              Raw(b'\xa5' * 100))
3566         rxs = self.send_and_expect(self.pg0, p * N_PKTS, self.pg1)
3567
3568         # a recursive v6 route via a v4 next-hop
3569         VppIpRoute(
3570             self, "3001::1", 128,
3571             [VppRoutePath(self.pg1.remote_ip4,
3572                           0xffffffff)]).add_vpp_config()
3573
3574         p = (Ether(src=self.pg0.remote_mac,
3575                    dst=self.pg0.local_mac) /
3576              IPv6(src=self.pg0.remote_ip6, dst="3001::1") /
3577              UDP(sport=1234, dport=1234) /
3578              Raw(b'\xa5' * 100))
3579         rxs = self.send_and_expect(self.pg0, p * N_PKTS, self.pg1)
3580
3581         for rx in rxs:
3582             self.assertEqual(rx[IPv6].dst, "3001::1")
3583
3584         VppIpRoute(
3585             self, "3001::2", 128,
3586             [VppRoutePath("1.1.1.1",
3587                           0xffffffff)]).add_vpp_config()
3588
3589         p = (Ether(src=self.pg0.remote_mac,
3590                    dst=self.pg0.local_mac) /
3591              IPv6(src=self.pg0.remote_ip6, dst="3001::2") /
3592              UDP(sport=1234, dport=1234) /
3593              Raw(b'\xa5' * 100))
3594         rxs = self.send_and_expect(self.pg0, p * N_PKTS, self.pg1)
3595
3596         for rx in rxs:
3597             self.assertEqual(rx[IPv6].dst, "3001::2")
3598
3599
3600 if __name__ == '__main__':
3601     unittest.main(testRunner=VppTestRunner)