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