test/test_ip6.py

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