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