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 test_ip6_load_balance(self):
1969         """ IPv6 Load-Balancing """
1970
1971         #
1972         # An array of packets that differ only in the destination port
1973         #  - IP only
1974         #  - MPLS EOS
1975         #  - MPLS non-EOS
1976         #  - MPLS non-EOS with an entropy label
1977         #
1978         port_ip_pkts = []
1979         port_mpls_pkts = []
1980         port_mpls_neos_pkts = []
1981         port_ent_pkts = []
1982
1983         #
1984         # An array of packets that differ only in the source address
1985         #
1986         src_ip_pkts = []
1987         src_mpls_pkts = []
1988
1989         for ii in range(NUM_PKTS):
1990             port_ip_hdr = (
1991                 IPv6(dst="3000::1", src="3000:1::1") /
1992                 inet6.UDP(sport=1234, dport=1234 + ii) /
1993                 Raw(b'\xa5' * 100))
1994             port_ip_pkts.append((Ether(src=self.pg0.remote_mac,
1995                                        dst=self.pg0.local_mac) /
1996                                  port_ip_hdr))
1997             port_mpls_pkts.append((Ether(src=self.pg0.remote_mac,
1998                                          dst=self.pg0.local_mac) /
1999                                    MPLS(label=66, ttl=2) /
2000                                    port_ip_hdr))
2001             port_mpls_neos_pkts.append((Ether(src=self.pg0.remote_mac,
2002                                               dst=self.pg0.local_mac) /
2003                                         MPLS(label=67, ttl=2) /
2004                                         MPLS(label=77, ttl=2) /
2005                                         port_ip_hdr))
2006             port_ent_pkts.append((Ether(src=self.pg0.remote_mac,
2007                                         dst=self.pg0.local_mac) /
2008                                   MPLS(label=67, ttl=2) /
2009                                   MPLS(label=14, ttl=2) /
2010                                   MPLS(label=999, ttl=2) /
2011                                   port_ip_hdr))
2012             src_ip_hdr = (
2013                 IPv6(dst="3000::1", src="3000:1::%d" % ii) /
2014                 inet6.UDP(sport=1234, dport=1234) /
2015                 Raw(b'\xa5' * 100))
2016             src_ip_pkts.append((Ether(src=self.pg0.remote_mac,
2017                                       dst=self.pg0.local_mac) /
2018                                 src_ip_hdr))
2019             src_mpls_pkts.append((Ether(src=self.pg0.remote_mac,
2020                                         dst=self.pg0.local_mac) /
2021                                   MPLS(label=66, ttl=2) /
2022                                   src_ip_hdr))
2023
2024         #
2025         # A route for the IP packets
2026         #
2027         route_3000_1 = VppIpRoute(self, "3000::1", 128,
2028                                   [VppRoutePath(self.pg1.remote_ip6,
2029                                                 self.pg1.sw_if_index),
2030                                    VppRoutePath(self.pg2.remote_ip6,
2031                                                 self.pg2.sw_if_index)])
2032         route_3000_1.add_vpp_config()
2033
2034         #
2035         # a local-label for the EOS packets
2036         #
2037         binding = VppMplsIpBind(self, 66, "3000::1", 128, is_ip6=1)
2038         binding.add_vpp_config()
2039
2040         #
2041         # An MPLS route for the non-EOS packets
2042         #
2043         route_67 = VppMplsRoute(self, 67, 0,
2044                                 [VppRoutePath(self.pg1.remote_ip6,
2045                                               self.pg1.sw_if_index,
2046                                               labels=[67]),
2047                                  VppRoutePath(self.pg2.remote_ip6,
2048                                               self.pg2.sw_if_index,
2049                                               labels=[67])])
2050         route_67.add_vpp_config()
2051
2052         #
2053         # inject the packet on pg0 - expect load-balancing across the 2 paths
2054         #  - since the default hash config is to use IP src,dst and port
2055         #    src,dst
2056         # We are not going to ensure equal amounts of packets across each link,
2057         # since the hash algorithm is statistical and therefore this can never
2058         # be guaranteed. But with 64 different packets we do expect some
2059         # balancing. So instead just ensure there is traffic on each link.
2060         #
2061         rx = self.send_and_expect_load_balancing(self.pg0, port_ip_pkts,
2062                                                  [self.pg1, self.pg2])
2063         n_ip_pg0 = len(rx[0])
2064         self.send_and_expect_load_balancing(self.pg0, src_ip_pkts,
2065                                             [self.pg1, self.pg2])
2066         self.send_and_expect_load_balancing(self.pg0, port_mpls_pkts,
2067                                             [self.pg1, self.pg2])
2068         self.send_and_expect_load_balancing(self.pg0, src_mpls_pkts,
2069                                             [self.pg1, self.pg2])
2070         rx = self.send_and_expect_load_balancing(self.pg0, port_mpls_neos_pkts,
2071                                                  [self.pg1, self.pg2])
2072         n_mpls_pg0 = len(rx[0])
2073
2074         #
2075         # change the router ID and expect the distribution changes
2076         #
2077         self.vapi.set_ip_flow_hash_router_id(router_id=0x11111111)
2078
2079         rx = self.send_and_expect_load_balancing(self.pg0, port_ip_pkts,
2080                                                  [self.pg1, self.pg2])
2081         self.assertNotEqual(n_ip_pg0, len(rx[0]))
2082
2083         rx = self.send_and_expect_load_balancing(self.pg0, src_mpls_pkts,
2084                                                  [self.pg1, self.pg2])
2085         self.assertNotEqual(n_mpls_pg0, len(rx[0]))
2086
2087         #
2088         # The packets with Entropy label in should not load-balance,
2089         # since the Entropy value is fixed.
2090         #
2091         self.send_and_expect_only(self.pg0, port_ent_pkts, self.pg1)
2092
2093         #
2094         # change the flow hash config so it's only IP src,dst
2095         #  - now only the stream with differing source address will
2096         #    load-balance
2097         #
2098         self.vapi.set_ip_flow_hash(vrf_id=0, src=1, dst=1, proto=1,
2099                                    sport=0, dport=0, is_ipv6=1)
2100
2101         self.send_and_expect_load_balancing(self.pg0, src_ip_pkts,
2102                                             [self.pg1, self.pg2])
2103         self.send_and_expect_load_balancing(self.pg0, src_mpls_pkts,
2104                                             [self.pg1, self.pg2])
2105         self.send_and_expect_only(self.pg0, port_ip_pkts, self.pg2)
2106
2107         #
2108         # change the flow hash config back to defaults
2109         #
2110         self.vapi.set_ip_flow_hash(vrf_id=0, src=1, dst=1, sport=1, dport=1,
2111                                    proto=1, is_ipv6=1)
2112
2113         #
2114         # Recursive prefixes
2115         #  - testing that 2 stages of load-balancing occurs and there is no
2116         #    polarisation (i.e. only 2 of 4 paths are used)
2117         #
2118         port_pkts = []
2119         src_pkts = []
2120
2121         for ii in range(257):
2122             port_pkts.append((Ether(src=self.pg0.remote_mac,
2123                                     dst=self.pg0.local_mac) /
2124                               IPv6(dst="4000::1",
2125                                    src="4000:1::1") /
2126                               inet6.UDP(sport=1234,
2127                                         dport=1234 + ii) /
2128                               Raw(b'\xa5' * 100)))
2129             src_pkts.append((Ether(src=self.pg0.remote_mac,
2130                                    dst=self.pg0.local_mac) /
2131                              IPv6(dst="4000::1",
2132                                   src="4000:1::%d" % ii) /
2133                              inet6.UDP(sport=1234, dport=1234) /
2134                              Raw(b'\xa5' * 100)))
2135
2136         route_3000_2 = VppIpRoute(self, "3000::2", 128,
2137                                   [VppRoutePath(self.pg3.remote_ip6,
2138                                                 self.pg3.sw_if_index),
2139                                    VppRoutePath(self.pg4.remote_ip6,
2140                                                 self.pg4.sw_if_index)])
2141         route_3000_2.add_vpp_config()
2142
2143         route_4000_1 = VppIpRoute(self, "4000::1", 128,
2144                                   [VppRoutePath("3000::1",
2145                                                 0xffffffff),
2146                                    VppRoutePath("3000::2",
2147                                                 0xffffffff)])
2148         route_4000_1.add_vpp_config()
2149
2150         #
2151         # inject the packet on pg0 - expect load-balancing across all 4 paths
2152         #
2153         self.vapi.cli("clear trace")
2154         self.send_and_expect_load_balancing(self.pg0, port_pkts,
2155                                             [self.pg1, self.pg2,
2156                                              self.pg3, self.pg4])
2157         self.send_and_expect_load_balancing(self.pg0, src_pkts,
2158                                             [self.pg1, self.pg2,
2159                                              self.pg3, self.pg4])
2160
2161         #
2162         # Recursive prefixes
2163         #  - testing that 2 stages of load-balancing no choices
2164         #
2165         port_pkts = []
2166
2167         for ii in range(257):
2168             port_pkts.append((Ether(src=self.pg0.remote_mac,
2169                                     dst=self.pg0.local_mac) /
2170                               IPv6(dst="6000::1",
2171                                    src="6000:1::1") /
2172                               inet6.UDP(sport=1234,
2173                                         dport=1234 + ii) /
2174                               Raw(b'\xa5' * 100)))
2175
2176         route_5000_2 = VppIpRoute(self, "5000::2", 128,
2177                                   [VppRoutePath(self.pg3.remote_ip6,
2178                                                 self.pg3.sw_if_index)])
2179         route_5000_2.add_vpp_config()
2180
2181         route_6000_1 = VppIpRoute(self, "6000::1", 128,
2182                                   [VppRoutePath("5000::2",
2183                                                 0xffffffff)])
2184         route_6000_1.add_vpp_config()
2185
2186         #
2187         # inject the packet on pg0 - expect load-balancing across all 4 paths
2188         #
2189         self.vapi.cli("clear trace")
2190         self.send_and_expect_only(self.pg0, port_pkts, self.pg3)
2191
2192
2193 class IP6PuntSetup(object):
2194     """ Setup for IPv6 Punt Police/Redirect """
2195
2196     def punt_setup(self):
2197         self.create_pg_interfaces(range(4))
2198
2199         for i in self.pg_interfaces:
2200             i.admin_up()
2201             i.config_ip6()
2202             i.resolve_ndp()
2203
2204         self.pkt = (Ether(src=self.pg0.remote_mac,
2205                           dst=self.pg0.local_mac) /
2206                     IPv6(src=self.pg0.remote_ip6,
2207                          dst=self.pg0.local_ip6) /
2208                     inet6.TCP(sport=1234, dport=1234) /
2209                     Raw(b'\xa5' * 100))
2210
2211     def punt_teardown(self):
2212         for i in self.pg_interfaces:
2213             i.unconfig_ip6()
2214             i.admin_down()
2215
2216
2217 class TestIP6Punt(IP6PuntSetup, VppTestCase):
2218     """ IPv6 Punt Police/Redirect """
2219
2220     def setUp(self):
2221         super(TestIP6Punt, self).setUp()
2222         super(TestIP6Punt, self).punt_setup()
2223
2224     def tearDown(self):
2225         super(TestIP6Punt, self).punt_teardown()
2226         super(TestIP6Punt, self).tearDown()
2227
2228     def test_ip_punt(self):
2229         """ IP6 punt police and redirect """
2230
2231         pkts = self.pkt * 1025
2232
2233         #
2234         # Configure a punt redirect via pg1.
2235         #
2236         nh_addr = self.pg1.remote_ip6
2237         ip_punt_redirect = VppIpPuntRedirect(self, self.pg0.sw_if_index,
2238                                              self.pg1.sw_if_index, nh_addr)
2239         ip_punt_redirect.add_vpp_config()
2240
2241         self.send_and_expect(self.pg0, pkts, self.pg1)
2242
2243         #
2244         # add a policer
2245         #
2246         policer = VppPolicer(self, "ip6-punt", 400, 0, 10, 0, rate_type=1)
2247         policer.add_vpp_config()
2248         ip_punt_policer = VppIpPuntPolicer(self, policer.policer_index,
2249                                            is_ip6=True)
2250         ip_punt_policer.add_vpp_config()
2251
2252         self.vapi.cli("clear trace")
2253         self.pg0.add_stream(pkts)
2254         self.pg_enable_capture(self.pg_interfaces)
2255         self.pg_start()
2256
2257         #
2258         # the number of packet received should be greater than 0,
2259         # but not equal to the number sent, since some were policed
2260         #
2261         rx = self.pg1._get_capture(1)
2262         stats = policer.get_stats()
2263
2264         # Single rate policer - expect conform, violate but no exceed
2265         self.assertGreater(stats['conform_packets'], 0)
2266         self.assertEqual(stats['exceed_packets'], 0)
2267         self.assertGreater(stats['violate_packets'], 0)
2268
2269         self.assertGreater(len(rx), 0)
2270         self.assertLess(len(rx), len(pkts))
2271
2272         #
2273         # remove the policer. back to full rx
2274         #
2275         ip_punt_policer.remove_vpp_config()
2276         policer.remove_vpp_config()
2277         self.send_and_expect(self.pg0, pkts, self.pg1)
2278
2279         #
2280         # remove the redirect. expect full drop.
2281         #
2282         ip_punt_redirect.remove_vpp_config()
2283         self.send_and_assert_no_replies(self.pg0, pkts,
2284                                         "IP no punt config")
2285
2286         #
2287         # Add a redirect that is not input port selective
2288         #
2289         ip_punt_redirect = VppIpPuntRedirect(self, 0xffffffff,
2290                                              self.pg1.sw_if_index, nh_addr)
2291         ip_punt_redirect.add_vpp_config()
2292         self.send_and_expect(self.pg0, pkts, self.pg1)
2293         ip_punt_redirect.remove_vpp_config()
2294
2295     def test_ip_punt_dump(self):
2296         """ IP6 punt redirect dump"""
2297
2298         #
2299         # Configure a punt redirects
2300         #
2301         nh_address = self.pg3.remote_ip6
2302         ipr_03 = VppIpPuntRedirect(self, self.pg0.sw_if_index,
2303                                    self.pg3.sw_if_index, nh_address)
2304         ipr_13 = VppIpPuntRedirect(self, self.pg1.sw_if_index,
2305                                    self.pg3.sw_if_index, nh_address)
2306         ipr_23 = VppIpPuntRedirect(self, self.pg2.sw_if_index,
2307                                    self.pg3.sw_if_index, '0::0')
2308         ipr_03.add_vpp_config()
2309         ipr_13.add_vpp_config()
2310         ipr_23.add_vpp_config()
2311
2312         #
2313         # Dump pg0 punt redirects
2314         #
2315         self.assertTrue(ipr_03.query_vpp_config())
2316         self.assertTrue(ipr_13.query_vpp_config())
2317         self.assertTrue(ipr_23.query_vpp_config())
2318
2319         #
2320         # Dump punt redirects for all interfaces
2321         #
2322         punts = self.vapi.ip_punt_redirect_dump(0xffffffff, is_ipv6=1)
2323         self.assertEqual(len(punts), 3)
2324         for p in punts:
2325             self.assertEqual(p.punt.tx_sw_if_index, self.pg3.sw_if_index)
2326         self.assertNotEqual(punts[1].punt.nh, self.pg3.remote_ip6)
2327         self.assertEqual(str(punts[2].punt.nh), '::')
2328
2329
2330 class TestIP6PuntHandoff(IP6PuntSetup, VppTestCase):
2331     """ IPv6 Punt Police/Redirect """
2332     vpp_worker_count = 2
2333
2334     def setUp(self):
2335         super(TestIP6PuntHandoff, self).setUp()
2336         super(TestIP6PuntHandoff, self).punt_setup()
2337
2338     def tearDown(self):
2339         super(TestIP6PuntHandoff, self).punt_teardown()
2340         super(TestIP6PuntHandoff, self).tearDown()
2341
2342     def test_ip_punt(self):
2343         """ IP6 punt policer thread handoff """
2344         pkts = self.pkt * NUM_PKTS
2345
2346         #
2347         # Configure a punt redirect via pg1.
2348         #
2349         nh_addr = self.pg1.remote_ip6
2350         ip_punt_redirect = VppIpPuntRedirect(self, self.pg0.sw_if_index,
2351                                              self.pg1.sw_if_index, nh_addr)
2352         ip_punt_redirect.add_vpp_config()
2353
2354         action_tx = PolicerAction(
2355             VppEnum.vl_api_sse2_qos_action_type_t.SSE2_QOS_ACTION_API_TRANSMIT,
2356             0)
2357         #
2358         # This policer drops no packets, we are just
2359         # testing that they get to the right thread.
2360         #
2361         policer = VppPolicer(self, "ip6-punt", 400, 0, 10, 0, 1,
2362                              0, 0, False, action_tx, action_tx, action_tx)
2363         policer.add_vpp_config()
2364         ip_punt_policer = VppIpPuntPolicer(self, policer.policer_index,
2365                                            is_ip6=True)
2366         ip_punt_policer.add_vpp_config()
2367
2368         for worker in [0, 1]:
2369             self.send_and_expect(self.pg0, pkts, self.pg1, worker=worker)
2370             if worker == 0:
2371                 self.logger.debug(self.vapi.cli("show trace max 100"))
2372
2373         # Combined stats, all threads
2374         stats = policer.get_stats()
2375
2376         # Single rate policer - expect conform, violate but no exceed
2377         self.assertGreater(stats['conform_packets'], 0)
2378         self.assertEqual(stats['exceed_packets'], 0)
2379         self.assertGreater(stats['violate_packets'], 0)
2380
2381         # Worker 0, should have done all the policing
2382         stats0 = policer.get_stats(worker=0)
2383         self.assertEqual(stats, stats0)
2384
2385         # Worker 1, should have handed everything off
2386         stats1 = policer.get_stats(worker=1)
2387         self.assertEqual(stats1['conform_packets'], 0)
2388         self.assertEqual(stats1['exceed_packets'], 0)
2389         self.assertEqual(stats1['violate_packets'], 0)
2390
2391         # Bind the policer to worker 1 and repeat
2392         policer.bind_vpp_config(1, True)
2393         for worker in [0, 1]:
2394             self.send_and_expect(self.pg0, pkts, self.pg1, worker=worker)
2395             self.logger.debug(self.vapi.cli("show trace max 100"))
2396
2397         # The 2 workers should now have policed the same amount
2398         stats = policer.get_stats()
2399         stats0 = policer.get_stats(worker=0)
2400         stats1 = policer.get_stats(worker=1)
2401
2402         self.assertGreater(stats0['conform_packets'], 0)
2403         self.assertEqual(stats0['exceed_packets'], 0)
2404         self.assertGreater(stats0['violate_packets'], 0)
2405
2406         self.assertGreater(stats1['conform_packets'], 0)
2407         self.assertEqual(stats1['exceed_packets'], 0)
2408         self.assertGreater(stats1['violate_packets'], 0)
2409
2410         self.assertEqual(stats0['conform_packets'] + stats1['conform_packets'],
2411                          stats['conform_packets'])
2412
2413         self.assertEqual(stats0['violate_packets'] + stats1['violate_packets'],
2414                          stats['violate_packets'])
2415
2416         # Unbind the policer and repeat
2417         policer.bind_vpp_config(1, False)
2418         for worker in [0, 1]:
2419             self.send_and_expect(self.pg0, pkts, self.pg1, worker=worker)
2420             self.logger.debug(self.vapi.cli("show trace max 100"))
2421
2422         # The policer should auto-bind to worker 0 when packets arrive
2423         stats = policer.get_stats()
2424         stats0new = policer.get_stats(worker=0)
2425         stats1new = policer.get_stats(worker=1)
2426
2427         self.assertGreater(stats0new['conform_packets'],
2428                            stats0['conform_packets'])
2429         self.assertEqual(stats0new['exceed_packets'], 0)
2430         self.assertGreater(stats0new['violate_packets'],
2431                            stats0['violate_packets'])
2432
2433         self.assertEqual(stats1, stats1new)
2434
2435         #
2436         # Clean up
2437         #
2438         ip_punt_policer.remove_vpp_config()
2439         policer.remove_vpp_config()
2440         ip_punt_redirect.remove_vpp_config()
2441
2442
2443 class TestIPDeag(VppTestCase):
2444     """ IPv6 Deaggregate Routes """
2445
2446     @classmethod
2447     def setUpClass(cls):
2448         super(TestIPDeag, cls).setUpClass()
2449
2450     @classmethod
2451     def tearDownClass(cls):
2452         super(TestIPDeag, cls).tearDownClass()
2453
2454     def setUp(self):
2455         super(TestIPDeag, self).setUp()
2456
2457         self.create_pg_interfaces(range(3))
2458
2459         for i in self.pg_interfaces:
2460             i.admin_up()
2461             i.config_ip6()
2462             i.resolve_ndp()
2463
2464     def tearDown(self):
2465         super(TestIPDeag, self).tearDown()
2466         for i in self.pg_interfaces:
2467             i.unconfig_ip6()
2468             i.admin_down()
2469
2470     def test_ip_deag(self):
2471         """ IP Deag Routes """
2472
2473         #
2474         # Create a table to be used for:
2475         #  1 - another destination address lookup
2476         #  2 - a source address lookup
2477         #
2478         table_dst = VppIpTable(self, 1, is_ip6=1)
2479         table_src = VppIpTable(self, 2, is_ip6=1)
2480         table_dst.add_vpp_config()
2481         table_src.add_vpp_config()
2482
2483         #
2484         # Add a route in the default table to point to a deag/
2485         # second lookup in each of these tables
2486         #
2487         route_to_dst = VppIpRoute(self, "1::1", 128,
2488                                   [VppRoutePath("::",
2489                                                 0xffffffff,
2490                                                 nh_table_id=1)])
2491         route_to_src = VppIpRoute(
2492             self, "1::2", 128,
2493             [VppRoutePath("::",
2494                           0xffffffff,
2495                           nh_table_id=2,
2496                           type=FibPathType.FIB_PATH_TYPE_SOURCE_LOOKUP)])
2497
2498         route_to_dst.add_vpp_config()
2499         route_to_src.add_vpp_config()
2500
2501         #
2502         # packets to these destination are dropped, since they'll
2503         # hit the respective default routes in the second table
2504         #
2505         p_dst = (Ether(src=self.pg0.remote_mac,
2506                        dst=self.pg0.local_mac) /
2507                  IPv6(src="5::5", dst="1::1") /
2508                  inet6.TCP(sport=1234, dport=1234) /
2509                  Raw(b'\xa5' * 100))
2510         p_src = (Ether(src=self.pg0.remote_mac,
2511                        dst=self.pg0.local_mac) /
2512                  IPv6(src="2::2", dst="1::2") /
2513                  inet6.TCP(sport=1234, dport=1234) /
2514                  Raw(b'\xa5' * 100))
2515         pkts_dst = p_dst * 257
2516         pkts_src = p_src * 257
2517
2518         self.send_and_assert_no_replies(self.pg0, pkts_dst,
2519                                         "IP in dst table")
2520         self.send_and_assert_no_replies(self.pg0, pkts_src,
2521                                         "IP in src table")
2522
2523         #
2524         # add a route in the dst table to forward via pg1
2525         #
2526         route_in_dst = VppIpRoute(self, "1::1", 128,
2527                                   [VppRoutePath(self.pg1.remote_ip6,
2528                                                 self.pg1.sw_if_index)],
2529                                   table_id=1)
2530         route_in_dst.add_vpp_config()
2531
2532         self.send_and_expect(self.pg0, pkts_dst, self.pg1)
2533
2534         #
2535         # add a route in the src table to forward via pg2
2536         #
2537         route_in_src = VppIpRoute(self, "2::2", 128,
2538                                   [VppRoutePath(self.pg2.remote_ip6,
2539                                                 self.pg2.sw_if_index)],
2540                                   table_id=2)
2541         route_in_src.add_vpp_config()
2542         self.send_and_expect(self.pg0, pkts_src, self.pg2)
2543
2544         #
2545         # loop in the lookup DP
2546         #
2547         route_loop = VppIpRoute(self, "3::3", 128,
2548                                 [VppRoutePath("::",
2549                                               0xffffffff)])
2550         route_loop.add_vpp_config()
2551
2552         p_l = (Ether(src=self.pg0.remote_mac,
2553                      dst=self.pg0.local_mac) /
2554                IPv6(src="3::4", dst="3::3") /
2555                inet6.TCP(sport=1234, dport=1234) /
2556                Raw(b'\xa5' * 100))
2557
2558         self.send_and_assert_no_replies(self.pg0, p_l * 257,
2559                                         "IP lookup loop")
2560
2561
2562 class TestIP6Input(VppTestCase):
2563     """ IPv6 Input Exception Test Cases """
2564
2565     @classmethod
2566     def setUpClass(cls):
2567         super(TestIP6Input, cls).setUpClass()
2568
2569     @classmethod
2570     def tearDownClass(cls):
2571         super(TestIP6Input, cls).tearDownClass()
2572
2573     def setUp(self):
2574         super(TestIP6Input, self).setUp()
2575
2576         self.create_pg_interfaces(range(2))
2577
2578         for i in self.pg_interfaces:
2579             i.admin_up()
2580             i.config_ip6()
2581             i.resolve_ndp()
2582
2583     def tearDown(self):
2584         super(TestIP6Input, self).tearDown()
2585         for i in self.pg_interfaces:
2586             i.unconfig_ip6()
2587             i.admin_down()
2588
2589     def test_ip_input_icmp_reply(self):
2590         """ IP6 Input Exception - Return ICMP (3,0) """
2591         #
2592         # hop limit - ICMP replies
2593         #
2594         p_version = (Ether(src=self.pg0.remote_mac,
2595                            dst=self.pg0.local_mac) /
2596                      IPv6(src=self.pg0.remote_ip6,
2597                           dst=self.pg1.remote_ip6,
2598                           hlim=1) /
2599                      inet6.UDP(sport=1234, dport=1234) /
2600                      Raw(b'\xa5' * 100))
2601
2602         rxs = self.send_and_expect_some(self.pg0,
2603                                         p_version * NUM_PKTS,
2604                                         self.pg0)
2605
2606         for rx in rxs:
2607             icmp = rx[ICMPv6TimeExceeded]
2608             # 0: "hop limit exceeded in transit",
2609             self.assertEqual((icmp.type, icmp.code), (3, 0))
2610
2611     icmpv6_data = '\x0a' * 18
2612     all_0s = "::"
2613     all_1s = "FFFF:FFFF:FFFF:FFFF:FFFF:FFFF:FFFF:FFFF"
2614
2615     @parameterized.expand([
2616         # Name, src, dst, l4proto, msg, timeout
2617         ("src='iface',   dst='iface'", None, None,
2618          inet6.UDP(sport=1234, dport=1234), "funky version", None),
2619         ("src='All 0's', dst='iface'", all_0s, None,
2620          ICMPv6EchoRequest(id=0xb, seq=5, data=icmpv6_data), None, 0.1),
2621         ("src='iface',   dst='All 0's'", None, all_0s,
2622          ICMPv6EchoRequest(id=0xb, seq=5, data=icmpv6_data), None, 0.1),
2623         ("src='All 1's', dst='iface'", all_1s, None,
2624          ICMPv6EchoRequest(id=0xb, seq=5, data=icmpv6_data), None, 0.1),
2625         ("src='iface',   dst='All 1's'", None, all_1s,
2626          ICMPv6EchoRequest(id=0xb, seq=5, data=icmpv6_data), None, 0.1),
2627         ("src='All 1's', dst='All 1's'", all_1s, all_1s,
2628          ICMPv6EchoRequest(id=0xb, seq=5, data=icmpv6_data), None, 0.1),
2629
2630     ])
2631     def test_ip_input_no_replies(self, name, src, dst, l4, msg, timeout):
2632
2633         self._testMethodDoc = 'IPv6 Input Exception - %s' % name
2634
2635         p_version = (Ether(src=self.pg0.remote_mac,
2636                            dst=self.pg0.local_mac) /
2637                      IPv6(src=src or self.pg0.remote_ip6,
2638                           dst=dst or self.pg1.remote_ip6,
2639                           version=3) /
2640                      l4 /
2641                      Raw(b'\xa5' * 100))
2642
2643         self.send_and_assert_no_replies(self.pg0, p_version * NUM_PKTS,
2644                                         remark=msg or "",
2645                                         timeout=timeout)
2646
2647     def test_hop_by_hop(self):
2648         """ Hop-by-hop header test """
2649
2650         p = (Ether(src=self.pg0.remote_mac,
2651                    dst=self.pg0.local_mac) /
2652              IPv6(src=self.pg0.remote_ip6, dst=self.pg0.local_ip6) /
2653              IPv6ExtHdrHopByHop() /
2654              inet6.UDP(sport=1234, dport=1234) /
2655              Raw(b'\xa5' * 100))
2656
2657         self.pg0.add_stream(p)
2658         self.pg_enable_capture(self.pg_interfaces)
2659         self.pg_start()
2660
2661
2662 class TestIPReplace(VppTestCase):
2663     """ IPv6 Table Replace """
2664
2665     @classmethod
2666     def setUpClass(cls):
2667         super(TestIPReplace, cls).setUpClass()
2668
2669     @classmethod
2670     def tearDownClass(cls):
2671         super(TestIPReplace, cls).tearDownClass()
2672
2673     def setUp(self):
2674         super(TestIPReplace, self).setUp()
2675
2676         self.create_pg_interfaces(range(4))
2677
2678         table_id = 1
2679         self.tables = []
2680
2681         for i in self.pg_interfaces:
2682             i.admin_up()
2683             i.config_ip6()
2684             i.generate_remote_hosts(2)
2685             self.tables.append(VppIpTable(self, table_id,
2686                                           True).add_vpp_config())
2687             table_id += 1
2688
2689     def tearDown(self):
2690         super(TestIPReplace, self).tearDown()
2691         for i in self.pg_interfaces:
2692             i.admin_down()
2693             i.unconfig_ip6()
2694
2695     def test_replace(self):
2696         """ IP Table Replace """
2697
2698         MRouteItfFlags = VppEnum.vl_api_mfib_itf_flags_t
2699         MRouteEntryFlags = VppEnum.vl_api_mfib_entry_flags_t
2700         N_ROUTES = 20
2701         links = [self.pg0, self.pg1, self.pg2, self.pg3]
2702         routes = [[], [], [], []]
2703
2704         # the sizes of 'empty' tables
2705         for t in self.tables:
2706             self.assertEqual(len(t.dump()), 2)
2707             self.assertEqual(len(t.mdump()), 5)
2708
2709         # load up the tables with some routes
2710         for ii, t in enumerate(self.tables):
2711             for jj in range(1, N_ROUTES):
2712                 uni = VppIpRoute(
2713                     self, "2001::%d" % jj if jj != 0 else "2001::", 128,
2714                     [VppRoutePath(links[ii].remote_hosts[0].ip6,
2715                                   links[ii].sw_if_index),
2716                      VppRoutePath(links[ii].remote_hosts[1].ip6,
2717                                   links[ii].sw_if_index)],
2718                     table_id=t.table_id).add_vpp_config()
2719                 multi = VppIpMRoute(
2720                     self, "::",
2721                     "ff:2001::%d" % jj, 128,
2722                     MRouteEntryFlags.MFIB_API_ENTRY_FLAG_NONE,
2723                     [VppMRoutePath(self.pg0.sw_if_index,
2724                                    MRouteItfFlags.MFIB_API_ITF_FLAG_ACCEPT,
2725                                    proto=FibPathProto.FIB_PATH_NH_PROTO_IP6),
2726                      VppMRoutePath(self.pg1.sw_if_index,
2727                                    MRouteItfFlags.MFIB_API_ITF_FLAG_FORWARD,
2728                                    proto=FibPathProto.FIB_PATH_NH_PROTO_IP6),
2729                      VppMRoutePath(self.pg2.sw_if_index,
2730                                    MRouteItfFlags.MFIB_API_ITF_FLAG_FORWARD,
2731                                    proto=FibPathProto.FIB_PATH_NH_PROTO_IP6),
2732                      VppMRoutePath(self.pg3.sw_if_index,
2733                                    MRouteItfFlags.MFIB_API_ITF_FLAG_FORWARD,
2734                                    proto=FibPathProto.FIB_PATH_NH_PROTO_IP6)],
2735                     table_id=t.table_id).add_vpp_config()
2736                 routes[ii].append({'uni': uni,
2737                                    'multi': multi})
2738
2739         #
2740         # replace the tables a few times
2741         #
2742         for kk in range(3):
2743             # replace each table
2744             for t in self.tables:
2745                 t.replace_begin()
2746
2747             # all the routes are still there
2748             for ii, t in enumerate(self.tables):
2749                 dump = t.dump()
2750                 mdump = t.mdump()
2751                 for r in routes[ii]:
2752                     self.assertTrue(find_route_in_dump(dump, r['uni'], t))
2753                     self.assertTrue(find_mroute_in_dump(mdump, r['multi'], t))
2754
2755             # redownload the even numbered routes
2756             for ii, t in enumerate(self.tables):
2757                 for jj in range(0, N_ROUTES, 2):
2758                     routes[ii][jj]['uni'].add_vpp_config()
2759                     routes[ii][jj]['multi'].add_vpp_config()
2760
2761             # signal each table converged
2762             for t in self.tables:
2763                 t.replace_end()
2764
2765             # we should find the even routes, but not the odd
2766             for ii, t in enumerate(self.tables):
2767                 dump = t.dump()
2768                 mdump = t.mdump()
2769                 for jj in range(0, N_ROUTES, 2):
2770                     self.assertTrue(find_route_in_dump(
2771                         dump, routes[ii][jj]['uni'], t))
2772                     self.assertTrue(find_mroute_in_dump(
2773                         mdump, routes[ii][jj]['multi'], t))
2774                 for jj in range(1, N_ROUTES - 1, 2):
2775                     self.assertFalse(find_route_in_dump(
2776                         dump, routes[ii][jj]['uni'], t))
2777                     self.assertFalse(find_mroute_in_dump(
2778                         mdump, routes[ii][jj]['multi'], t))
2779
2780             # reload all the routes
2781             for ii, t in enumerate(self.tables):
2782                 for r in routes[ii]:
2783                     r['uni'].add_vpp_config()
2784                     r['multi'].add_vpp_config()
2785
2786             # all the routes are still there
2787             for ii, t in enumerate(self.tables):
2788                 dump = t.dump()
2789                 mdump = t.mdump()
2790                 for r in routes[ii]:
2791                     self.assertTrue(find_route_in_dump(dump, r['uni'], t))
2792                     self.assertTrue(find_mroute_in_dump(mdump, r['multi'], t))
2793
2794         #
2795         # finally flush the tables for good measure
2796         #
2797         for t in self.tables:
2798             t.flush()
2799             self.assertEqual(len(t.dump()), 2)
2800             self.assertEqual(len(t.mdump()), 5)
2801
2802
2803 class TestIP6Replace(VppTestCase):
2804     """ IPv4 Interface Address Replace """
2805
2806     @classmethod
2807     def setUpClass(cls):
2808         super(TestIP6Replace, cls).setUpClass()
2809
2810     @classmethod
2811     def tearDownClass(cls):
2812         super(TestIP6Replace, cls).tearDownClass()
2813
2814     def setUp(self):
2815         super(TestIP6Replace, self).setUp()
2816
2817         self.create_pg_interfaces(range(4))
2818
2819         for i in self.pg_interfaces:
2820             i.admin_up()
2821
2822     def tearDown(self):
2823         super(TestIP6Replace, self).tearDown()
2824         for i in self.pg_interfaces:
2825             i.admin_down()
2826
2827     def get_n_pfxs(self, intf):
2828         return len(self.vapi.ip_address_dump(intf.sw_if_index, True))
2829
2830     def test_replace(self):
2831         """ IP interface address replace """
2832
2833         intf_pfxs = [[], [], [], []]
2834
2835         # add prefixes to each of the interfaces
2836         for i in range(len(self.pg_interfaces)):
2837             intf = self.pg_interfaces[i]
2838
2839             # 2001:16:x::1/64
2840             addr = "2001:16:%d::1" % intf.sw_if_index
2841             a = VppIpInterfaceAddress(self, intf, addr, 64).add_vpp_config()
2842             intf_pfxs[i].append(a)
2843
2844             # 2001:16:x::2/64 - a different address in the same subnet as above
2845             addr = "2001:16:%d::2" % intf.sw_if_index
2846             a = VppIpInterfaceAddress(self, intf, addr, 64).add_vpp_config()
2847             intf_pfxs[i].append(a)
2848
2849             # 2001:15:x::2/64 - a different address and subnet
2850             addr = "2001:15:%d::2" % intf.sw_if_index
2851             a = VppIpInterfaceAddress(self, intf, addr, 64).add_vpp_config()
2852             intf_pfxs[i].append(a)
2853
2854         # a dump should n_address in it
2855         for intf in self.pg_interfaces:
2856             self.assertEqual(self.get_n_pfxs(intf), 3)
2857
2858         #
2859         # remove all the address thru a replace
2860         #
2861         self.vapi.sw_interface_address_replace_begin()
2862         self.vapi.sw_interface_address_replace_end()
2863         for intf in self.pg_interfaces:
2864             self.assertEqual(self.get_n_pfxs(intf), 0)
2865
2866         #
2867         # add all the interface addresses back
2868         #
2869         for p in intf_pfxs:
2870             for v in p:
2871                 v.add_vpp_config()
2872         for intf in self.pg_interfaces:
2873             self.assertEqual(self.get_n_pfxs(intf), 3)
2874
2875         #
2876         # replace again, but this time update/re-add the address on the first
2877         # two interfaces
2878         #
2879         self.vapi.sw_interface_address_replace_begin()
2880
2881         for p in intf_pfxs[:2]:
2882             for v in p:
2883                 v.add_vpp_config()
2884
2885         self.vapi.sw_interface_address_replace_end()
2886
2887         # on the first two the address still exist,
2888         # on the other two they do not
2889         for intf in self.pg_interfaces[:2]:
2890             self.assertEqual(self.get_n_pfxs(intf), 3)
2891         for p in intf_pfxs[:2]:
2892             for v in p:
2893                 self.assertTrue(v.query_vpp_config())
2894         for intf in self.pg_interfaces[2:]:
2895             self.assertEqual(self.get_n_pfxs(intf), 0)
2896
2897         #
2898         # add all the interface addresses back on the last two
2899         #
2900         for p in intf_pfxs[2:]:
2901             for v in p:
2902                 v.add_vpp_config()
2903         for intf in self.pg_interfaces:
2904             self.assertEqual(self.get_n_pfxs(intf), 3)
2905
2906         #
2907         # replace again, this time add different prefixes on all the interfaces
2908         #
2909         self.vapi.sw_interface_address_replace_begin()
2910
2911         pfxs = []
2912         for intf in self.pg_interfaces:
2913             # 2001:18:x::1/64
2914             addr = "2001:18:%d::1" % intf.sw_if_index
2915             pfxs.append(VppIpInterfaceAddress(self, intf, addr,
2916                                               64).add_vpp_config())
2917
2918         self.vapi.sw_interface_address_replace_end()
2919
2920         # only .18 should exist on each interface
2921         for intf in self.pg_interfaces:
2922             self.assertEqual(self.get_n_pfxs(intf), 1)
2923         for pfx in pfxs:
2924             self.assertTrue(pfx.query_vpp_config())
2925
2926         #
2927         # remove everything
2928         #
2929         self.vapi.sw_interface_address_replace_begin()
2930         self.vapi.sw_interface_address_replace_end()
2931         for intf in self.pg_interfaces:
2932             self.assertEqual(self.get_n_pfxs(intf), 0)
2933
2934         #
2935         # add prefixes to each interface. post-begin add the prefix from
2936         # interface X onto interface Y. this would normally be an error
2937         # since it would generate a 'duplicate address' warning. but in
2938         # this case, since what is newly downloaded is sane, it's ok
2939         #
2940         for intf in self.pg_interfaces:
2941             # 2001:18:x::1/64
2942             addr = "2001:18:%d::1" % intf.sw_if_index
2943             VppIpInterfaceAddress(self, intf, addr, 64).add_vpp_config()
2944
2945         self.vapi.sw_interface_address_replace_begin()
2946
2947         pfxs = []
2948         for intf in self.pg_interfaces:
2949             # 2001:18:x::1/64
2950             addr = "2001:18:%d::1" % (intf.sw_if_index + 1)
2951             pfxs.append(VppIpInterfaceAddress(self, intf,
2952                                               addr, 64).add_vpp_config())
2953
2954         self.vapi.sw_interface_address_replace_end()
2955
2956         self.logger.info(self.vapi.cli("sh int addr"))
2957
2958         for intf in self.pg_interfaces:
2959             self.assertEqual(self.get_n_pfxs(intf), 1)
2960         for pfx in pfxs:
2961             self.assertTrue(pfx.query_vpp_config())
2962
2963
2964 class TestIP6LinkLocal(VppTestCase):
2965     """ IPv6 Link Local """
2966
2967     @classmethod
2968     def setUpClass(cls):
2969         super(TestIP6LinkLocal, cls).setUpClass()
2970
2971     @classmethod
2972     def tearDownClass(cls):
2973         super(TestIP6LinkLocal, cls).tearDownClass()
2974
2975     def setUp(self):
2976         super(TestIP6LinkLocal, self).setUp()
2977
2978         self.create_pg_interfaces(range(2))
2979
2980         for i in self.pg_interfaces:
2981             i.admin_up()
2982
2983     def tearDown(self):
2984         super(TestIP6LinkLocal, self).tearDown()
2985         for i in self.pg_interfaces:
2986             i.admin_down()
2987
2988     def test_ip6_ll(self):
2989         """ IPv6 Link Local """
2990
2991         #
2992         # two APIs to add a link local address.
2993         #   1 - just like any other prefix
2994         #   2 - with the special set LL API
2995         #
2996
2997         #
2998         # First with the API to set a 'normal' prefix
2999         #
3000         ll1 = "fe80:1::1"
3001         ll2 = "fe80:2::2"
3002         ll3 = "fe80:3::3"
3003
3004         VppNeighbor(self,
3005                     self.pg0.sw_if_index,
3006                     self.pg0.remote_mac,
3007                     ll2).add_vpp_config()
3008
3009         VppIpInterfaceAddress(self, self.pg0, ll1, 128).add_vpp_config()
3010
3011         #
3012         # should be able to ping the ll
3013         #
3014         p_echo_request_1 = (Ether(src=self.pg0.remote_mac,
3015                                   dst=self.pg0.local_mac) /
3016                             IPv6(src=ll2,
3017                                  dst=ll1) /
3018                             ICMPv6EchoRequest())
3019
3020         self.send_and_expect(self.pg0, [p_echo_request_1], self.pg0)
3021
3022         #
3023         # change the link-local on pg0
3024         #
3025         v_ll3 = VppIpInterfaceAddress(self, self.pg0,
3026                                       ll3, 128).add_vpp_config()
3027
3028         p_echo_request_3 = (Ether(src=self.pg0.remote_mac,
3029                                   dst=self.pg0.local_mac) /
3030                             IPv6(src=ll2,
3031                                  dst=ll3) /
3032                             ICMPv6EchoRequest())
3033
3034         self.send_and_expect(self.pg0, [p_echo_request_3], self.pg0)
3035
3036         #
3037         # set a normal v6 prefix on the link
3038         #
3039         self.pg0.config_ip6()
3040
3041         self.send_and_expect(self.pg0, [p_echo_request_3], self.pg0)
3042
3043         # the link-local cannot be removed
3044         with self.vapi.assert_negative_api_retval():
3045             v_ll3.remove_vpp_config()
3046
3047         #
3048         # Use the specific link-local API on pg1
3049         #
3050         VppIp6LinkLocalAddress(self, self.pg1, ll1).add_vpp_config()
3051         self.send_and_expect(self.pg1, [p_echo_request_1], self.pg1)
3052
3053         VppIp6LinkLocalAddress(self, self.pg1, ll3).add_vpp_config()
3054         self.send_and_expect(self.pg1, [p_echo_request_3], self.pg1)
3055
3056     def test_ip6_ll_p2p(self):
3057         """ IPv6 Link Local P2P (GRE)"""
3058
3059         self.pg0.config_ip4()
3060         self.pg0.resolve_arp()
3061         gre_if = VppGreInterface(self,
3062                                  self.pg0.local_ip4,
3063                                  self.pg0.remote_ip4).add_vpp_config()
3064         gre_if.admin_up()
3065
3066         ll1 = "fe80:1::1"
3067         ll2 = "fe80:2::2"
3068
3069         VppIpInterfaceAddress(self, gre_if, ll1, 128).add_vpp_config()
3070
3071         self.logger.info(self.vapi.cli("sh ip6-ll gre0 fe80:2::2"))
3072
3073         p_echo_request_1 = (Ether(src=self.pg0.remote_mac,
3074                                   dst=self.pg0.local_mac) /
3075                             IP(src=self.pg0.remote_ip4,
3076                                dst=self.pg0.local_ip4) /
3077                             GRE() /
3078                             IPv6(src=ll2, dst=ll1) /
3079                             ICMPv6EchoRequest())
3080         self.send_and_expect(self.pg0, [p_echo_request_1], self.pg0)
3081
3082         self.pg0.unconfig_ip4()
3083         gre_if.remove_vpp_config()
3084
3085     def test_ip6_ll_p2mp(self):
3086         """ IPv6 Link Local P2MP (GRE)"""
3087
3088         self.pg0.config_ip4()
3089         self.pg0.resolve_arp()
3090
3091         gre_if = VppGreInterface(
3092             self,
3093             self.pg0.local_ip4,
3094             "0.0.0.0",
3095             mode=(VppEnum.vl_api_tunnel_mode_t.
3096                   TUNNEL_API_MODE_MP)).add_vpp_config()
3097         gre_if.admin_up()
3098
3099         ll1 = "fe80:1::1"
3100         ll2 = "fe80:2::2"
3101
3102         VppIpInterfaceAddress(self, gre_if, ll1, 128).add_vpp_config()
3103
3104         p_echo_request_1 = (Ether(src=self.pg0.remote_mac,
3105                                   dst=self.pg0.local_mac) /
3106                             IP(src=self.pg0.remote_ip4,
3107                                dst=self.pg0.local_ip4) /
3108                             GRE() /
3109                             IPv6(src=ll2, dst=ll1) /
3110                             ICMPv6EchoRequest())
3111
3112         # no route back at this point
3113         self.send_and_assert_no_replies(self.pg0, [p_echo_request_1])
3114
3115         # add teib entry for the peer
3116         teib = VppTeib(self, gre_if, ll2, self.pg0.remote_ip4)
3117         teib.add_vpp_config()
3118
3119         self.logger.info(self.vapi.cli("sh ip6-ll gre0 %s" % ll2))
3120         self.send_and_expect(self.pg0, [p_echo_request_1], self.pg0)
3121
3122         # teardown
3123         self.pg0.unconfig_ip4()
3124
3125
3126 class TestIPv6PathMTU(VppTestCase):
3127     """ IPv6 Path MTU """
3128
3129     def setUp(self):
3130         super(TestIPv6PathMTU, self).setUp()
3131
3132         self.create_pg_interfaces(range(2))
3133
3134         # setup all interfaces
3135         for i in self.pg_interfaces:
3136             i.admin_up()
3137             i.config_ip6()
3138             i.resolve_ndp()
3139
3140     def tearDown(self):
3141         super(TestIPv6PathMTU, self).tearDown()
3142         for i in self.pg_interfaces:
3143             i.unconfig_ip6()
3144             i.admin_down()
3145
3146     def test_path_mtu_local(self):
3147         """ Path MTU for attached neighbour """
3148
3149         self.vapi.cli("set log class ip level debug")
3150         #
3151         # The goal here is not test that fragmentation works correctly,
3152         # that's done elsewhere, the intent is to ensure that the Path MTU
3153         # settings are honoured.
3154         #
3155
3156         #
3157         # IPv6 will only frag locally generated packets, so use tunnelled
3158         # packets post encap
3159         #
3160         tun = VppIpIpTunInterface(
3161             self,
3162             self.pg1,
3163             self.pg1.local_ip6,
3164             self.pg1.remote_ip6)
3165         tun.add_vpp_config()
3166         tun.admin_up()
3167         tun.config_ip6()
3168
3169         # set the interface MTU to a reasonable value
3170         self.vapi.sw_interface_set_mtu(self.pg1.sw_if_index,
3171                                        [2800, 0, 0, 0])
3172
3173         p_2k = (Ether(dst=self.pg0.local_mac,
3174                       src=self.pg0.remote_mac) /
3175                 IPv6(src=self.pg0.remote_ip6,
3176                      dst=tun.remote_ip6) /
3177                 UDP(sport=1234, dport=5678) /
3178                 Raw(b'0xa' * 1000))
3179         p_1k = (Ether(dst=self.pg0.local_mac,
3180                       src=self.pg0.remote_mac) /
3181                 IPv6(src=self.pg0.remote_ip6,
3182                      dst=tun.remote_ip6) /
3183                 UDP(sport=1234, dport=5678) /
3184                 Raw(b'0xa' * 600))
3185
3186         nbr = VppNeighbor(self,
3187                           self.pg1.sw_if_index,
3188                           self.pg1.remote_mac,
3189                           self.pg1.remote_ip6).add_vpp_config()
3190
3191         # this is now the interface MTU frags
3192         self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=2)
3193         self.send_and_expect(self.pg0, [p_1k], self.pg1)
3194
3195         # drop the path MTU for this neighbour to below the interface MTU
3196         # expect more frags
3197         pmtu = VppIpPathMtu(self, self.pg1.remote_ip6, 1300).add_vpp_config()
3198
3199         # print/format the adj delegate and trackers
3200         self.logger.info(self.vapi.cli("sh ip pmtu"))
3201         self.logger.info(self.vapi.cli("sh adj 7"))
3202
3203         self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=3)
3204         self.send_and_expect(self.pg0, [p_1k], self.pg1, n_rx=2)
3205
3206         # increase the path MTU to more than the interface
3207         # expect to use the interface MTU
3208         pmtu.modify(8192)
3209
3210         self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=2)
3211         self.send_and_expect(self.pg0, [p_1k], self.pg1)
3212
3213         # go back to an MTU from the path
3214         pmtu.modify(1300)
3215
3216         self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=3)
3217         self.send_and_expect(self.pg0, [p_1k], self.pg1, n_rx=2)
3218
3219         # raise the interface's MTU
3220         # should still use that of the path
3221         self.vapi.sw_interface_set_mtu(self.pg1.sw_if_index,
3222                                        [2000, 0, 0, 0])
3223         self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=3)
3224         self.send_and_expect(self.pg0, [p_1k], self.pg1, n_rx=2)
3225
3226         # set path high and interface low
3227         pmtu.modify(2000)
3228         self.vapi.sw_interface_set_mtu(self.pg1.sw_if_index,
3229                                        [1300, 0, 0, 0])
3230         self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=3)
3231         self.send_and_expect(self.pg0, [p_1k], self.pg1, n_rx=2)
3232
3233         # remove the path MTU
3234         self.vapi.sw_interface_set_mtu(self.pg1.sw_if_index,
3235                                        [2800, 0, 0, 0])
3236         pmtu.modify(0)
3237
3238         self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=2)
3239         self.send_and_expect(self.pg0, [p_1k], self.pg1)
3240
3241     def test_path_mtu_remote(self):
3242         """ Path MTU for remote neighbour """
3243
3244         self.vapi.cli("set log class ip level debug")
3245         #
3246         # The goal here is not test that fragmentation works correctly,
3247         # that's done elsewhere, the intent is to ensure that the Path MTU
3248         # settings are honoured.
3249         #
3250         tun_dst = "2001::1"
3251
3252         route = VppIpRoute(
3253             self, tun_dst, 64,
3254             [VppRoutePath(self.pg1.remote_ip6,
3255                           self.pg1.sw_if_index)]).add_vpp_config()
3256
3257         #
3258         # IPv6 will only frag locally generated packets, so use tunnelled
3259         # packets post encap
3260         #
3261         tun = VppIpIpTunInterface(
3262             self,
3263             self.pg1,
3264             self.pg1.local_ip6,
3265             tun_dst)
3266         tun.add_vpp_config()
3267         tun.admin_up()
3268         tun.config_ip6()
3269
3270         # set the interface MTU to a reasonable value
3271         self.vapi.sw_interface_set_mtu(self.pg1.sw_if_index,
3272                                        [2800, 0, 0, 0])
3273
3274         p_2k = (Ether(dst=self.pg0.local_mac,
3275                       src=self.pg0.remote_mac) /
3276                 IPv6(src=self.pg0.remote_ip6,
3277                      dst=tun.remote_ip6) /
3278                 UDP(sport=1234, dport=5678) /
3279                 Raw(b'0xa' * 1000))
3280         p_1k = (Ether(dst=self.pg0.local_mac,
3281                       src=self.pg0.remote_mac) /
3282                 IPv6(src=self.pg0.remote_ip6,
3283                      dst=tun.remote_ip6) /
3284                 UDP(sport=1234, dport=5678) /
3285                 Raw(b'0xa' * 600))
3286
3287         nbr = VppNeighbor(self,
3288                           self.pg1.sw_if_index,
3289                           self.pg1.remote_mac,
3290                           self.pg1.remote_ip6).add_vpp_config()
3291
3292         # this is now the interface MTU frags
3293         self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=2)
3294         self.send_and_expect(self.pg0, [p_1k], self.pg1)
3295
3296         # drop the path MTU for this neighbour to below the interface MTU
3297         # expect more frags
3298         pmtu = VppIpPathMtu(self, tun_dst, 1300).add_vpp_config()
3299
3300         # print/format the fib entry/dpo
3301         self.logger.info(self.vapi.cli("sh ip6 fib 2001::1"))
3302
3303         self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=3)
3304         self.send_and_expect(self.pg0, [p_1k], self.pg1, n_rx=2)
3305
3306         # increase the path MTU to more than the interface
3307         # expect to use the interface MTU
3308         pmtu.modify(8192)
3309
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         # go back to an MTU from the path
3314         pmtu.modify(1300)
3315
3316         self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=3)
3317         self.send_and_expect(self.pg0, [p_1k], self.pg1, n_rx=2)
3318
3319         # raise the interface's MTU
3320         # should still use that of the path
3321         self.vapi.sw_interface_set_mtu(self.pg1.sw_if_index,
3322                                        [2000, 0, 0, 0])
3323         self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=3)
3324         self.send_and_expect(self.pg0, [p_1k], self.pg1, n_rx=2)
3325
3326         # turn the tun_dst into an attached neighbour
3327         route.modify([VppRoutePath("::",
3328                                    self.pg1.sw_if_index)])
3329         nbr2 = VppNeighbor(self,
3330                            self.pg1.sw_if_index,
3331                            self.pg1.remote_mac,
3332                            tun_dst).add_vpp_config()
3333
3334         self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=3)
3335         self.send_and_expect(self.pg0, [p_1k], self.pg1, n_rx=2)
3336
3337         # add back to not attached
3338         nbr2.remove_vpp_config()
3339         route.modify([VppRoutePath(self.pg1.remote_ip6,
3340                                    self.pg1.sw_if_index)])
3341
3342         # set path high and interface low
3343         pmtu.modify(2000)
3344         self.vapi.sw_interface_set_mtu(self.pg1.sw_if_index,
3345                                        [1300, 0, 0, 0])
3346         self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=3)
3347         self.send_and_expect(self.pg0, [p_1k], self.pg1, n_rx=2)
3348
3349         # remove the path MTU
3350         self.vapi.sw_interface_set_mtu(self.pg1.sw_if_index,
3351                                        [2800, 0, 0, 0])
3352         pmtu.remove_vpp_config()
3353         self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=2)
3354         self.send_and_expect(self.pg0, [p_1k], self.pg1)
3355
3356
3357 class TestIPFibSource(VppTestCase):
3358     """ IPv6 Table FibSource """
3359
3360     @classmethod
3361     def setUpClass(cls):
3362         super(TestIPFibSource, cls).setUpClass()
3363
3364     @classmethod
3365     def tearDownClass(cls):
3366         super(TestIPFibSource, cls).tearDownClass()
3367
3368     def setUp(self):
3369         super(TestIPFibSource, self).setUp()
3370
3371         self.create_pg_interfaces(range(2))
3372
3373         for i in self.pg_interfaces:
3374             i.admin_up()
3375             i.config_ip6()
3376             i.resolve_arp()
3377             i.generate_remote_hosts(2)
3378             i.configure_ipv6_neighbors()
3379
3380     def tearDown(self):
3381         super(TestIPFibSource, self).tearDown()
3382         for i in self.pg_interfaces:
3383             i.admin_down()
3384             i.unconfig_ip4()
3385
3386     def test_fib_source(self):
3387         """ IP Table FibSource """
3388
3389         routes = self.vapi.ip_route_v2_dump(0, True)
3390
3391         # 2 interfaces (4 routes) + 2 specials + 4 neighbours = 10 routes
3392         self.assertEqual(len(routes), 10)
3393
3394         # dump all the sources in the FIB
3395         sources = self.vapi.fib_source_dump()
3396         for source in sources:
3397             if (source.src.name == "API"):
3398                 api_source = source.src
3399             if (source.src.name == "interface"):
3400                 intf_source = source.src
3401             if (source.src.name == "adjacency"):
3402                 adj_source = source.src
3403             if (source.src.name == "special"):
3404                 special_source = source.src
3405             if (source.src.name == "default-route"):
3406                 dr_source = source.src
3407
3408         # dump the individual route types
3409         routes = self.vapi.ip_route_v2_dump(0, True, src=adj_source.id)
3410         self.assertEqual(len(routes), 4)
3411         routes = self.vapi.ip_route_v2_dump(0, True, src=intf_source.id)
3412         self.assertEqual(len(routes), 4)
3413         routes = self.vapi.ip_route_v2_dump(0, True, src=special_source.id)
3414         self.assertEqual(len(routes), 1)
3415         routes = self.vapi.ip_route_v2_dump(0, True, src=dr_source.id)
3416         self.assertEqual(len(routes), 1)
3417
3418         # add a new soure that'a better than the API
3419         self.vapi.fib_source_add(src={'name': "bgp",
3420                                       "priority": api_source.priority - 1})
3421
3422         # dump all the sources to check our new one is there
3423         sources = self.vapi.fib_source_dump()
3424
3425         for source in sources:
3426             if (source.src.name == "bgp"):
3427                 bgp_source = source.src
3428
3429         self.assertTrue(bgp_source)
3430         self.assertEqual(bgp_source.priority,
3431                          api_source.priority - 1)
3432
3433         # add a route with the default API source
3434         r1 = VppIpRouteV2(
3435             self, "2001::1", 128,
3436             [VppRoutePath(self.pg0.remote_ip6,
3437                           self.pg0.sw_if_index)]).add_vpp_config()
3438
3439         r2 = VppIpRouteV2(self, "2001::1", 128,
3440                           [VppRoutePath(self.pg1.remote_ip6,
3441                                         self.pg1.sw_if_index)],
3442                           src=bgp_source.id).add_vpp_config()
3443
3444         # ensure the BGP source takes priority
3445         p = (Ether(src=self.pg0.remote_mac,
3446                    dst=self.pg0.local_mac) /
3447              IPv6(src=self.pg0.remote_ip6, dst="2001::1") /
3448              inet6.UDP(sport=1234, dport=1234) /
3449              Raw(b'\xa5' * 100))
3450
3451         self.send_and_expect(self.pg0, [p], self.pg1)
3452
3453         r2.remove_vpp_config()
3454         r1.remove_vpp_config()
3455
3456         self.assertFalse(find_route(self, "2001::1", 128))
3457
3458
3459 class TestIPxAF(VppTestCase):
3460     """ IP cross AF """
3461
3462     @classmethod
3463     def setUpClass(cls):
3464         super(TestIPxAF, cls).setUpClass()
3465
3466     @classmethod
3467     def tearDownClass(cls):
3468         super(TestIPxAF, cls).tearDownClass()
3469
3470     def setUp(self):
3471         super(TestIPxAF, self).setUp()
3472
3473         self.create_pg_interfaces(range(2))
3474
3475         for i in self.pg_interfaces:
3476             i.admin_up()
3477             i.config_ip6()
3478             i.config_ip4()
3479             i.resolve_arp()
3480             i.resolve_ndp()
3481
3482     def tearDown(self):
3483         super(TestIPxAF, self).tearDown()
3484         for i in self.pg_interfaces:
3485             i.admin_down()
3486             i.unconfig_ip4()
3487             i.unconfig_ip6()
3488
3489     def test_x_af(self):
3490         """ Cross AF routing """
3491
3492         N_PKTS = 63
3493         # a v4 route via a v6 attached next-hop
3494         VppIpRoute(
3495             self, "1.1.1.1", 32,
3496             [VppRoutePath(self.pg1.remote_ip6,
3497                           self.pg1.sw_if_index)]).add_vpp_config()
3498
3499         p = (Ether(src=self.pg0.remote_mac,
3500                    dst=self.pg0.local_mac) /
3501              IP(src=self.pg0.remote_ip4, dst="1.1.1.1") /
3502              UDP(sport=1234, dport=1234) /
3503              Raw(b'\xa5' * 100))
3504         rxs = self.send_and_expect(self.pg0, p * N_PKTS, self.pg1)
3505
3506         for rx in rxs:
3507             self.assertEqual(rx[IP].dst, "1.1.1.1")
3508
3509         # a v6 route via a v4 attached next-hop
3510         VppIpRoute(
3511             self, "2001::1", 128,
3512             [VppRoutePath(self.pg1.remote_ip4,
3513                           self.pg1.sw_if_index)]).add_vpp_config()
3514
3515         p = (Ether(src=self.pg0.remote_mac,
3516                    dst=self.pg0.local_mac) /
3517              IPv6(src=self.pg0.remote_ip6, dst="2001::1") /
3518              UDP(sport=1234, dport=1234) /
3519              Raw(b'\xa5' * 100))
3520         rxs = self.send_and_expect(self.pg0, p * N_PKTS, self.pg1)
3521
3522         for rx in rxs:
3523             self.assertEqual(rx[IPv6].dst, "2001::1")
3524
3525         # a recursive v4 route via a v6 next-hop (from above)
3526         VppIpRoute(
3527             self, "2.2.2.2", 32,
3528             [VppRoutePath("2001::1",
3529                           0xffffffff)]).add_vpp_config()
3530
3531         p = (Ether(src=self.pg0.remote_mac,
3532                    dst=self.pg0.local_mac) /
3533              IP(src=self.pg0.remote_ip4, dst="2.2.2.2") /
3534              UDP(sport=1234, dport=1234) /
3535              Raw(b'\xa5' * 100))
3536         rxs = self.send_and_expect(self.pg0, p * N_PKTS, self.pg1)
3537
3538         # a recursive v4 route via a v6 next-hop
3539         VppIpRoute(
3540             self, "2.2.2.3", 32,
3541             [VppRoutePath(self.pg1.remote_ip6,
3542                           0xffffffff)]).add_vpp_config()
3543
3544         p = (Ether(src=self.pg0.remote_mac,
3545                    dst=self.pg0.local_mac) /
3546              IP(src=self.pg0.remote_ip4, dst="2.2.2.3") /
3547              UDP(sport=1234, dport=1234) /
3548              Raw(b'\xa5' * 100))
3549         rxs = self.send_and_expect(self.pg0, p * N_PKTS, self.pg1)
3550
3551         # a recursive v6 route via a v4 next-hop
3552         VppIpRoute(
3553             self, "3001::1", 128,
3554             [VppRoutePath(self.pg1.remote_ip4,
3555                           0xffffffff)]).add_vpp_config()
3556
3557         p = (Ether(src=self.pg0.remote_mac,
3558                    dst=self.pg0.local_mac) /
3559              IPv6(src=self.pg0.remote_ip6, dst="3001::1") /
3560              UDP(sport=1234, dport=1234) /
3561              Raw(b'\xa5' * 100))
3562         rxs = self.send_and_expect(self.pg0, p * N_PKTS, self.pg1)
3563
3564         for rx in rxs:
3565             self.assertEqual(rx[IPv6].dst, "3001::1")
3566
3567         VppIpRoute(
3568             self, "3001::2", 128,
3569             [VppRoutePath("1.1.1.1",
3570                           0xffffffff)]).add_vpp_config()
3571
3572         p = (Ether(src=self.pg0.remote_mac,
3573                    dst=self.pg0.local_mac) /
3574              IPv6(src=self.pg0.remote_ip6, dst="3001::2") /
3575              UDP(sport=1234, dport=1234) /
3576              Raw(b'\xa5' * 100))
3577         rxs = self.send_and_expect(self.pg0, p * N_PKTS, self.pg1)
3578
3579         for rx in rxs:
3580             self.assertEqual(rx[IPv6].dst, "3001::2")
3581
3582
3583 class TestIPv6Punt(VppTestCase):
3584     """ IPv6 Punt Police/Redirect """
3585
3586     def setUp(self):
3587         super(TestIPv6Punt, self).setUp()
3588         self.create_pg_interfaces(range(4))
3589
3590         for i in self.pg_interfaces:
3591             i.admin_up()
3592             i.config_ip6()
3593             i.resolve_ndp()
3594
3595     def tearDown(self):
3596         super(TestIPv6Punt, self).tearDown()
3597         for i in self.pg_interfaces:
3598             i.unconfig_ip6()
3599             i.admin_down()
3600
3601     def test_ip6_punt(self):
3602         """ IPv6 punt police and redirect """
3603
3604         # use UDP packet that have a port we need to explicitly
3605         # register to get punted.
3606         pt_l4 = VppEnum.vl_api_punt_type_t.PUNT_API_TYPE_L4
3607         af_ip6 = VppEnum.vl_api_address_family_t.ADDRESS_IP6
3608         udp_proto = VppEnum.vl_api_ip_proto_t.IP_API_PROTO_UDP
3609         punt_udp = {
3610             'type': pt_l4,
3611             'punt': {
3612                 'l4': {
3613                     'af': af_ip6,
3614                     'protocol': udp_proto,
3615                     'port': 7654,
3616                 }
3617             }
3618         }
3619
3620         self.vapi.set_punt(is_add=1, punt=punt_udp)
3621
3622         pkts = (Ether(src=self.pg0.remote_mac,
3623                       dst=self.pg0.local_mac) /
3624                 IPv6(src=self.pg0.remote_ip6, dst=self.pg0.local_ip6) /
3625                 UDP(sport=1234, dport=7654) /
3626                 Raw(b'\xa5' * 100)) * 1025
3627
3628         #
3629         # Configure a punt redirect via pg1.
3630         #
3631         nh_addr = self.pg1.remote_ip6
3632         ip_punt_redirect = VppIpPuntRedirect(self, self.pg0.sw_if_index,
3633                                              self.pg1.sw_if_index, nh_addr)
3634         ip_punt_redirect.add_vpp_config()
3635
3636         self.send_and_expect(self.pg0, pkts, self.pg1)
3637
3638         #
3639         # add a policer
3640         #
3641         policer = VppPolicer(self, "ip6-punt", 400, 0, 10, 0, rate_type=1)
3642         policer.add_vpp_config()
3643         ip_punt_policer = VppIpPuntPolicer(self, policer.policer_index,
3644                                            is_ip6=True)
3645         ip_punt_policer.add_vpp_config()
3646
3647         self.vapi.cli("clear trace")
3648         self.pg0.add_stream(pkts)
3649         self.pg_enable_capture(self.pg_interfaces)
3650         self.pg_start()
3651
3652         #
3653         # the number of packet received should be greater than 0,
3654         # but not equal to the number sent, since some were policed
3655         #
3656         rx = self.pg1._get_capture(1)
3657
3658         stats = policer.get_stats()
3659
3660         # Single rate policer - expect conform, violate but no exceed
3661         self.assertGreater(stats['conform_packets'], 0)
3662         self.assertEqual(stats['exceed_packets'], 0)
3663         self.assertGreater(stats['violate_packets'], 0)
3664
3665         self.assertGreater(len(rx), 0)
3666         self.assertLess(len(rx), len(pkts))
3667
3668         #
3669         # remove the policer. back to full rx
3670         #
3671         ip_punt_policer.remove_vpp_config()
3672         policer.remove_vpp_config()
3673         self.send_and_expect(self.pg0, pkts, self.pg1)
3674
3675         #
3676         # remove the redirect. expect full drop.
3677         #
3678         ip_punt_redirect.remove_vpp_config()
3679         self.send_and_assert_no_replies(self.pg0, pkts,
3680                                         "IP no punt config")
3681
3682         #
3683         # Add a redirect that is not input port selective
3684         #
3685         ip_punt_redirect = VppIpPuntRedirect(self, 0xffffffff,
3686                                              self.pg1.sw_if_index, nh_addr)
3687         ip_punt_redirect.add_vpp_config()
3688         self.send_and_expect(self.pg0, pkts, self.pg1)
3689         ip_punt_redirect.remove_vpp_config()
3690
3691     def test_ip6_punt_dump(self):
3692         """ IPv6 punt redirect dump"""
3693
3694         #
3695         # Configure a punt redirects
3696         #
3697         nh_address = self.pg3.remote_ip6
3698         ipr_03 = VppIpPuntRedirect(self, self.pg0.sw_if_index,
3699                                    self.pg3.sw_if_index, nh_address)
3700         ipr_13 = VppIpPuntRedirect(self, self.pg1.sw_if_index,
3701                                    self.pg3.sw_if_index, nh_address)
3702         ipr_23 = VppIpPuntRedirect(self, self.pg2.sw_if_index,
3703                                    self.pg3.sw_if_index, "::")
3704         ipr_03.add_vpp_config()
3705         ipr_13.add_vpp_config()
3706         ipr_23.add_vpp_config()
3707
3708         #
3709         # Dump pg0 punt redirects
3710         #
3711         self.assertTrue(ipr_03.query_vpp_config())
3712         self.assertTrue(ipr_13.query_vpp_config())
3713         self.assertTrue(ipr_23.query_vpp_config())
3714
3715         #
3716         # Dump punt redirects for all interfaces
3717         #
3718         punts = self.vapi.ip_punt_redirect_dump(sw_if_index=0xffffffff,
3719                                                 is_ipv6=True)
3720         self.assertEqual(len(punts), 3)
3721         for p in punts:
3722             self.assertEqual(p.punt.tx_sw_if_index, self.pg3.sw_if_index)
3723         self.assertNotEqual(punts[1].punt.nh, self.pg3.remote_ip6)
3724         self.assertEqual(str(punts[2].punt.nh), '::')
3725
3726
3727 if __name__ == '__main__':
3728     unittest.main(testRunner=VppTestRunner)