test/test_ip6.py

   1 #!/usr/bin/env python
   2
   3 import unittest
   4 from socket import AF_INET6
   5
   6 from framework import VppTestCase, VppTestRunner
   7 from vpp_sub_interface import VppSubInterface, VppDot1QSubint
   8 from vpp_pg_interface import is_ipv6_misc
   9 from vpp_neighbor import find_nbr
  10 from vpp_ip_route import VppIpRoute, VppRoutePath
  11
  12 from scapy.packet import Raw
  13 from scapy.layers.l2 import Ether, Dot1Q
  14 from scapy.layers.inet6 import IPv6, UDP, ICMPv6ND_NS, ICMPv6ND_RS, \
  15     ICMPv6ND_RA, ICMPv6NDOptSrcLLAddr, getmacbyip6, ICMPv6MRD_Solicitation, \
  16     ICMPv6NDOptMTU, ICMPv6NDOptSrcLLAddr, ICMPv6NDOptPrefixInfo, \
  17     ICMPv6ND_NA, ICMPv6NDOptDstLLAddr, ICMPv6DestUnreach, icmp6types
  18
  19 from util import ppp
  20 from scapy.utils6 import in6_getnsma, in6_getnsmac, in6_ptop, in6_islladdr, \
  21     in6_mactoifaceid, in6_ismaddr
  22 from scapy.utils import inet_pton, inet_ntop
  23
  24
  25 def mk_ll_addr(mac):
  26     euid = in6_mactoifaceid(mac)
  27     addr = "fe80::" + euid
  28     return addr
  29
  30
  31 class TestIPv6ND(VppTestCase):
  32     def validate_ra(self, intf, rx, dst_ip=None):
  33         if not dst_ip:
  34             dst_ip = intf.remote_ip6
  35
  36         # unicasted packets must come to the unicast mac
  37         self.assertEqual(rx[Ether].dst, intf.remote_mac)
  38
  39         # and from the router's MAC
  40         self.assertEqual(rx[Ether].src, intf.local_mac)
  41
  42         # the rx'd RA should be addressed to the sender's source
  43         self.assertTrue(rx.haslayer(ICMPv6ND_RA))
  44         self.assertEqual(in6_ptop(rx[IPv6].dst),
  45                          in6_ptop(dst_ip))
  46
  47         # and come from the router's link local
  48         self.assertTrue(in6_islladdr(rx[IPv6].src))
  49         self.assertEqual(in6_ptop(rx[IPv6].src),
  50                          in6_ptop(mk_ll_addr(intf.local_mac)))
  51
  52     def validate_na(self, intf, rx, dst_ip=None, tgt_ip=None):
  53         if not dst_ip:
  54             dst_ip = intf.remote_ip6
  55         if not tgt_ip:
  56             dst_ip = intf.local_ip6
  57
  58         # unicasted packets must come to the unicast mac
  59         self.assertEqual(rx[Ether].dst, intf.remote_mac)
  60
  61         # and from the router's MAC
  62         self.assertEqual(rx[Ether].src, intf.local_mac)
  63
  64         # the rx'd NA should be addressed to the sender's source
  65         self.assertTrue(rx.haslayer(ICMPv6ND_NA))
  66         self.assertEqual(in6_ptop(rx[IPv6].dst),
  67                          in6_ptop(dst_ip))
  68
  69         # and come from the target address
  70         self.assertEqual(in6_ptop(rx[IPv6].src), in6_ptop(tgt_ip))
  71
  72         # Dest link-layer options should have the router's MAC
  73         dll = rx[ICMPv6NDOptDstLLAddr]
  74         self.assertEqual(dll.lladdr, intf.local_mac)
  75
  76     def send_and_expect_ra(self, intf, pkts, remark, dst_ip=None,
  77                            filter_out_fn=is_ipv6_misc):
  78         intf.add_stream(pkts)
  79         self.pg0.add_stream(pkts)
  80         self.pg_enable_capture(self.pg_interfaces)
  81         self.pg_start()
  82         rx = intf.get_capture(1, filter_out_fn=filter_out_fn)
  83
  84         self.assertEqual(len(rx), 1)
  85         rx = rx[0]
  86         self.validate_ra(intf, rx, dst_ip)
  87
  88     def send_and_assert_no_replies(self, intf, pkts, remark):
  89         intf.add_stream(pkts)
  90         self.pg_enable_capture(self.pg_interfaces)
  91         self.pg_start()
  92         intf.assert_nothing_captured(remark=remark)
  93
  94
  95 class TestIPv6(TestIPv6ND):
  96     """ IPv6 Test Case """
  97
  98     @classmethod
  99     def setUpClass(cls):
 100         super(TestIPv6, cls).setUpClass()
 101
 102     def setUp(self):
 103         """
 104         Perform test setup before test case.
 105
 106         **Config:**
 107             - create 3 pg interfaces
 108                 - untagged pg0 interface
 109                 - Dot1Q subinterface on pg1
 110                 - Dot1AD subinterface on pg2
 111             - setup interfaces:
 112                 - put it into UP state
 113                 - set IPv6 addresses
 114                 - resolve neighbor address using NDP
 115             - configure 200 fib entries
 116
 117         :ivar list interfaces: pg interfaces and subinterfaces.
 118         :ivar dict flows: IPv4 packet flows in test.
 119         :ivar list pg_if_packet_sizes: packet sizes in test.
 120
 121         *TODO:* Create AD sub interface
 122         """
 123         super(TestIPv6, self).setUp()
 124
 125         # create 3 pg interfaces
 126         self.create_pg_interfaces(range(3))
 127
 128         # create 2 subinterfaces for p1 and pg2
 129         self.sub_interfaces = [
 130             VppDot1QSubint(self, self.pg1, 100),
 131             VppDot1QSubint(self, self.pg2, 200)
 132             # TODO: VppDot1ADSubint(self, self.pg2, 200, 300, 400)
 133         ]
 134
 135         # packet flows mapping pg0 -> pg1.sub, pg2.sub, etc.
 136         self.flows = dict()
 137         self.flows[self.pg0] = [self.pg1.sub_if, self.pg2.sub_if]
 138         self.flows[self.pg1.sub_if] = [self.pg0, self.pg2.sub_if]
 139         self.flows[self.pg2.sub_if] = [self.pg0, self.pg1.sub_if]
 140
 141         # packet sizes
 142         self.pg_if_packet_sizes = [64, 512, 1518, 9018]
 143         self.sub_if_packet_sizes = [64, 512, 1518 + 4, 9018 + 4]
 144
 145         self.interfaces = list(self.pg_interfaces)
 146         self.interfaces.extend(self.sub_interfaces)
 147
 148         # setup all interfaces
 149         for i in self.interfaces:
 150             i.admin_up()
 151             i.config_ip6()
 152             i.resolve_ndp()
 153
 154         # config 2M FIB entries
 155         self.config_fib_entries(200)
 156
 157     def tearDown(self):
 158         """Run standard test teardown and log ``show ip6 neighbors``."""
 159         for i in self.sub_interfaces:
 160             i.unconfig_ip6()
 161             i.ip6_disable()
 162             i.admin_down()
 163             i.remove_vpp_config()
 164
 165         super(TestIPv6, self).tearDown()
 166         if not self.vpp_dead:
 167             self.logger.info(self.vapi.cli("show ip6 neighbors"))
 168             # info(self.vapi.cli("show ip6 fib"))  # many entries
 169
 170     def config_fib_entries(self, count):
 171         """For each interface add to the FIB table *count* routes to
 172         "fd02::1/128" destination with interface's local address as next-hop
 173         address.
 174
 175         :param int count: Number of FIB entries.
 176
 177         - *TODO:* check if the next-hop address shouldn't be remote address
 178           instead of local address.
 179         """
 180         n_int = len(self.interfaces)
 181         percent = 0
 182         counter = 0.0
 183         dest_addr = inet_pton(AF_INET6, "fd02::1")
 184         dest_addr_len = 128
 185         for i in self.interfaces:
 186             next_hop_address = i.local_ip6n
 187             for j in range(count / n_int):
 188                 self.vapi.ip_add_del_route(
 189                     dest_addr, dest_addr_len, next_hop_address, is_ipv6=1)
 190                 counter += 1
 191                 if counter / count * 100 > percent:
 192                     self.logger.info("Configure %d FIB entries .. %d%% done" %
 193                                      (count, percent))
 194                     percent += 1
 195
 196     def create_stream(self, src_if, packet_sizes):
 197         """Create input packet stream for defined interface.
 198
 199         :param VppInterface src_if: Interface to create packet stream for.
 200         :param list packet_sizes: Required packet sizes.
 201         """
 202         pkts = []
 203         for i in range(0, 257):
 204             dst_if = self.flows[src_if][i % 2]
 205             info = self.create_packet_info(src_if, dst_if)
 206             payload = self.info_to_payload(info)
 207             p = (Ether(dst=src_if.local_mac, src=src_if.remote_mac) /
 208                  IPv6(src=src_if.remote_ip6, dst=dst_if.remote_ip6) /
 209                  UDP(sport=1234, dport=1234) /
 210                  Raw(payload))
 211             info.data = p.copy()
 212             if isinstance(src_if, VppSubInterface):
 213                 p = src_if.add_dot1_layer(p)
 214             size = packet_sizes[(i // 2) % len(packet_sizes)]
 215             self.extend_packet(p, size)
 216             pkts.append(p)
 217         return pkts
 218
 219     def verify_capture(self, dst_if, capture):
 220         """Verify captured input packet stream for defined interface.
 221
 222         :param VppInterface dst_if: Interface to verify captured packet stream
 223                                     for.
 224         :param list capture: Captured packet stream.
 225         """
 226         self.logger.info("Verifying capture on interface %s" % dst_if.name)
 227         last_info = dict()
 228         for i in self.interfaces:
 229             last_info[i.sw_if_index] = None
 230         is_sub_if = False
 231         dst_sw_if_index = dst_if.sw_if_index
 232         if hasattr(dst_if, 'parent'):
 233             is_sub_if = True
 234         for packet in capture:
 235             if is_sub_if:
 236                 # Check VLAN tags and Ethernet header
 237                 packet = dst_if.remove_dot1_layer(packet)
 238             self.assertTrue(Dot1Q not in packet)
 239             try:
 240                 ip = packet[IPv6]
 241                 udp = packet[UDP]
 242                 payload_info = self.payload_to_info(str(packet[Raw]))
 243                 packet_index = payload_info.index
 244                 self.assertEqual(payload_info.dst, dst_sw_if_index)
 245                 self.logger.debug(
 246                     "Got packet on port %s: src=%u (id=%u)" %
 247                     (dst_if.name, payload_info.src, packet_index))
 248                 next_info = self.get_next_packet_info_for_interface2(
 249                     payload_info.src, dst_sw_if_index,
 250                     last_info[payload_info.src])
 251                 last_info[payload_info.src] = next_info
 252                 self.assertTrue(next_info is not None)
 253                 self.assertEqual(packet_index, next_info.index)
 254                 saved_packet = next_info.data
 255                 # Check standard fields
 256                 self.assertEqual(ip.src, saved_packet[IPv6].src)
 257                 self.assertEqual(ip.dst, saved_packet[IPv6].dst)
 258                 self.assertEqual(udp.sport, saved_packet[UDP].sport)
 259                 self.assertEqual(udp.dport, saved_packet[UDP].dport)
 260             except:
 261                 self.logger.error(ppp("Unexpected or invalid packet:", packet))
 262                 raise
 263         for i in self.interfaces:
 264             remaining_packet = self.get_next_packet_info_for_interface2(
 265                 i.sw_if_index, dst_sw_if_index, last_info[i.sw_if_index])
 266             self.assertTrue(remaining_packet is None,
 267                             "Interface %s: Packet expected from interface %s "
 268                             "didn't arrive" % (dst_if.name, i.name))
 269
 270     def test_fib(self):
 271         """ IPv6 FIB test
 272
 273         Test scenario:
 274             - Create IPv6 stream for pg0 interface
 275             - Create IPv6 tagged streams for pg1's and pg2's subinterface.
 276             - Send and verify received packets on each interface.
 277         """
 278
 279         pkts = self.create_stream(self.pg0, self.pg_if_packet_sizes)
 280         self.pg0.add_stream(pkts)
 281
 282         for i in self.sub_interfaces:
 283             pkts = self.create_stream(i, self.sub_if_packet_sizes)
 284             i.parent.add_stream(pkts)
 285
 286         self.pg_enable_capture(self.pg_interfaces)
 287         self.pg_start()
 288
 289         pkts = self.pg0.get_capture()
 290         self.verify_capture(self.pg0, pkts)
 291
 292         for i in self.sub_interfaces:
 293             pkts = i.parent.get_capture()
 294             self.verify_capture(i, pkts)
 295
 296     def test_ns(self):
 297         """ IPv6 Neighbour Solicitation Exceptions
 298
 299         Test scenario:
 300            - Send an NS Sourced from an address not covered by the link sub-net
 301            - Send an NS to an mcast address the router has not joined
 302            - Send NS for a target address the router does not onn.
 303         """
 304
 305         #
 306         # An NS from a non link source address
 307         #
 308         nsma = in6_getnsma(inet_pton(AF_INET6, self.pg0.local_ip6))
 309         d = inet_ntop(AF_INET6, nsma)
 310
 311         p = (Ether(dst=in6_getnsmac(nsma)) /
 312              IPv6(dst=d, src="2002::2") /
 313              ICMPv6ND_NS(tgt=self.pg0.local_ip6) /
 314              ICMPv6NDOptSrcLLAddr(lladdr=self.pg0.remote_mac))
 315         pkts = [p]
 316
 317         self.send_and_assert_no_replies(
 318             self.pg0, pkts,
 319             "No response to NS source by address not on sub-net")
 320
 321         #
 322         # An NS for sent to a solicited mcast group the router is
 323         # not a member of FAILS
 324         #
 325         if 0:
 326             nsma = in6_getnsma(inet_pton(AF_INET6, "fd::ffff"))
 327             d = inet_ntop(AF_INET6, nsma)
 328
 329             p = (Ether(dst=in6_getnsmac(nsma)) /
 330                  IPv6(dst=d, src=self.pg0.remote_ip6) /
 331                  ICMPv6ND_NS(tgt=self.pg0.local_ip6) /
 332                  ICMPv6NDOptSrcLLAddr(lladdr=self.pg0.remote_mac))
 333             pkts = [p]
 334
 335             self.send_and_assert_no_replies(
 336                 self.pg0, pkts,
 337                 "No response to NS sent to unjoined mcast address")
 338
 339         #
 340         # An NS whose target address is one the router does not own
 341         #
 342         nsma = in6_getnsma(inet_pton(AF_INET6, self.pg0.local_ip6))
 343         d = inet_ntop(AF_INET6, nsma)
 344
 345         p = (Ether(dst=in6_getnsmac(nsma)) /
 346              IPv6(dst=d, src=self.pg0.remote_ip6) /
 347              ICMPv6ND_NS(tgt="fd::ffff") /
 348              ICMPv6NDOptSrcLLAddr(lladdr=self.pg0.remote_mac))
 349         pkts = [p]
 350
 351         self.send_and_assert_no_replies(self.pg0, pkts,
 352                                         "No response to NS for unknown target")
 353
 354     def validate_ra(self, intf, rx, dst_ip=None, mtu=9000, pi_opt=None):
 355         if not dst_ip:
 356             dst_ip = intf.remote_ip6
 357
 358         # unicasted packets must come to the unicast mac
 359         self.assertEqual(rx[Ether].dst, intf.remote_mac)
 360
 361         # and from the router's MAC
 362         self.assertEqual(rx[Ether].src, intf.local_mac)
 363
 364         # the rx'd RA should be addressed to the sender's source
 365         self.assertTrue(rx.haslayer(ICMPv6ND_RA))
 366         self.assertEqual(in6_ptop(rx[IPv6].dst),
 367                          in6_ptop(dst_ip))
 368
 369         # and come from the router's link local
 370         self.assertTrue(in6_islladdr(rx[IPv6].src))
 371         self.assertEqual(in6_ptop(rx[IPv6].src),
 372                          in6_ptop(mk_ll_addr(intf.local_mac)))
 373
 374         # it should contain the links MTU
 375         ra = rx[ICMPv6ND_RA]
 376         self.assertEqual(ra[ICMPv6NDOptMTU].mtu, mtu)
 377
 378         # it should contain the source's link layer address option
 379         sll = ra[ICMPv6NDOptSrcLLAddr]
 380         self.assertEqual(sll.lladdr, intf.local_mac)
 381
 382         if not pi_opt:
 383             # the RA should not contain prefix information
 384             self.assertFalse(ra.haslayer(ICMPv6NDOptPrefixInfo))
 385         else:
 386             raos = rx.getlayer(ICMPv6NDOptPrefixInfo, 1)
 387
 388             # the options are nested in the scapy packet in way that i cannot
 389             # decipher how to decode. this 1st layer of option always returns
 390             # nested classes, so a direct obj1=obj2 comparison always fails.
 391             # however, the getlayer(.., 2) does give one instnace.
 392             # so we cheat here and construct a new opt instnace for comparison
 393             rd = ICMPv6NDOptPrefixInfo(prefixlen=raos.prefixlen,
 394                                        prefix=raos.prefix,
 395                                        L=raos.L,
 396                                        A=raos.A)
 397             if type(pi_opt) is list:
 398                 for ii in range(len(pi_opt)):
 399                     self.assertEqual(pi_opt[ii], rd)
 400                     rd = rx.getlayer(ICMPv6NDOptPrefixInfo, ii+2)
 401             else:
 402                 self.assertEqual(pi_opt, raos)
 403
 404     def send_and_expect_ra(self, intf, pkts, remark, dst_ip=None,
 405                            filter_out_fn=is_ipv6_misc,
 406                            opt=None):
 407         intf.add_stream(pkts)
 408         self.pg_enable_capture(self.pg_interfaces)
 409         self.pg_start()
 410         rx = intf.get_capture(1, filter_out_fn=filter_out_fn)
 411
 412         self.assertEqual(len(rx), 1)
 413         rx = rx[0]
 414         self.validate_ra(intf, rx, dst_ip, pi_opt=opt)
 415
 416     def test_rs(self):
 417         """ IPv6 Router Solicitation Exceptions
 418
 419         Test scenario:
 420         """
 421
 422         #
 423         # Before we begin change the IPv6 RA responses to use the unicast
 424         # address - that way we will not confuse them with the periodic
 425         # RAs which go to the mcast address
 426         # Sit and wait for the first periodic RA.
 427         #
 428         # TODO
 429         #
 430         self.pg0.ip6_ra_config(send_unicast=1)
 431
 432         #
 433         # An RS from a link source address
 434         #  - expect an RA in return
 435         #
 436         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
 437              IPv6(dst=self.pg0.local_ip6, src=self.pg0.remote_ip6) /
 438              ICMPv6ND_RS())
 439         pkts = [p]
 440         self.send_and_expect_ra(self.pg0, pkts, "Genuine RS")
 441
 442         #
 443         # For the next RS sent the RA should be rate limited
 444         #
 445         self.send_and_assert_no_replies(self.pg0, pkts, "RA rate limited")
 446
 447         #
 448         # When we reconfiure the IPv6 RA config, we reset the RA rate limiting,
 449         # so we need to do this before each test below so as not to drop
 450         # packets for rate limiting reasons. Test this works here.
 451         #
 452         self.pg0.ip6_ra_config(send_unicast=1)
 453         self.send_and_expect_ra(self.pg0, pkts, "Rate limit reset RS")
 454
 455         #
 456         # An RS sent from a non-link local source
 457         #
 458         self.pg0.ip6_ra_config(send_unicast=1)
 459         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
 460              IPv6(dst=self.pg0.local_ip6, src="2002::ffff") /
 461              ICMPv6ND_RS())
 462         pkts = [p]
 463         self.send_and_assert_no_replies(self.pg0, pkts,
 464                                         "RS from non-link source")
 465
 466         #
 467         # Source an RS from a link local address
 468         #
 469         self.pg0.ip6_ra_config(send_unicast=1)
 470         ll = mk_ll_addr(self.pg0.remote_mac)
 471         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
 472              IPv6(dst=self.pg0.local_ip6, src=ll) /
 473              ICMPv6ND_RS())
 474         pkts = [p]
 475         self.send_and_expect_ra(self.pg0, pkts,
 476                                 "RS sourced from link-local",
 477                                 dst_ip=ll)
 478
 479         #
 480         # Send the RS multicast
 481         #
 482         self.pg0.ip6_ra_config(send_unicast=1)
 483         dmac = in6_getnsmac(inet_pton(AF_INET6, "ff02::2"))
 484         ll = mk_ll_addr(self.pg0.remote_mac)
 485         p = (Ether(dst=dmac, src=self.pg0.remote_mac) /
 486              IPv6(dst="ff02::2", src=ll) /
 487              ICMPv6ND_RS())
 488         pkts = [p]
 489         self.send_and_expect_ra(self.pg0, pkts,
 490                                 "RS sourced from link-local",
 491                                 dst_ip=ll)
 492
 493         #
 494         # Source from the unspecified address ::. This happens when the RS
 495         # is sent before the host has a configured address/sub-net,
 496         # i.e. auto-config. Since the sender has no IP address, the reply
 497         # comes back mcast - so the capture needs to not filter this.
 498         # If we happen to pick up the periodic RA at this point then so be it,
 499         # it's not an error.
 500         #
 501         self.pg0.ip6_ra_config(send_unicast=1, suppress=1)
 502         p = (Ether(dst=dmac, src=self.pg0.remote_mac) /
 503              IPv6(dst="ff02::2", src="::") /
 504              ICMPv6ND_RS())
 505         pkts = [p]
 506         self.send_and_expect_ra(self.pg0, pkts,
 507                                 "RS sourced from unspecified",
 508                                 dst_ip="ff02::1",
 509                                 filter_out_fn=None)
 510
 511         #
 512         # Configure The RA to announce the links prefix
 513         #
 514         self.pg0.ip6_ra_prefix(self.pg0.local_ip6n,
 515                                self.pg0.local_ip6_prefix_len)
 516
 517         #
 518         # RAs should now contain the prefix information option
 519         #
 520         opt = ICMPv6NDOptPrefixInfo(prefixlen=self.pg0.local_ip6_prefix_len,
 521                                     prefix=self.pg0.local_ip6,
 522                                     L=1,
 523                                     A=1)
 524
 525         self.pg0.ip6_ra_config(send_unicast=1)
 526         ll = mk_ll_addr(self.pg0.remote_mac)
 527         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
 528              IPv6(dst=self.pg0.local_ip6, src=ll) /
 529              ICMPv6ND_RS())
 530         self.send_and_expect_ra(self.pg0, p,
 531                                 "RA with prefix-info",
 532                                 dst_ip=ll,
 533                                 opt=opt)
 534
 535         #
 536         # Change the prefix info to not off-link
 537         #  L-flag is clear
 538         #
 539         self.pg0.ip6_ra_prefix(self.pg0.local_ip6n,
 540                                self.pg0.local_ip6_prefix_len,
 541                                off_link=1)
 542
 543         opt = ICMPv6NDOptPrefixInfo(prefixlen=self.pg0.local_ip6_prefix_len,
 544                                     prefix=self.pg0.local_ip6,
 545                                     L=0,
 546                                     A=1)
 547
 548         self.pg0.ip6_ra_config(send_unicast=1)
 549         self.send_and_expect_ra(self.pg0, p,
 550                                 "RA with Prefix info with L-flag=0",
 551                                 dst_ip=ll,
 552                                 opt=opt)
 553
 554         #
 555         # Change the prefix info to not off-link, no-autoconfig
 556         #  L and A flag are clear in the advert
 557         #
 558         self.pg0.ip6_ra_prefix(self.pg0.local_ip6n,
 559                                self.pg0.local_ip6_prefix_len,
 560                                off_link=1,
 561                                no_autoconfig=1)
 562
 563         opt = ICMPv6NDOptPrefixInfo(prefixlen=self.pg0.local_ip6_prefix_len,
 564                                     prefix=self.pg0.local_ip6,
 565                                     L=0,
 566                                     A=0)
 567
 568         self.pg0.ip6_ra_config(send_unicast=1)
 569         self.send_and_expect_ra(self.pg0, p,
 570                                 "RA with Prefix info with A & L-flag=0",
 571                                 dst_ip=ll,
 572                                 opt=opt)
 573
 574         #
 575         # Change the flag settings back to the defaults
 576         #  L and A flag are set in the advert
 577         #
 578         self.pg0.ip6_ra_prefix(self.pg0.local_ip6n,
 579                                self.pg0.local_ip6_prefix_len)
 580
 581         opt = ICMPv6NDOptPrefixInfo(prefixlen=self.pg0.local_ip6_prefix_len,
 582                                     prefix=self.pg0.local_ip6,
 583                                     L=1,
 584                                     A=1)
 585
 586         self.pg0.ip6_ra_config(send_unicast=1)
 587         self.send_and_expect_ra(self.pg0, p,
 588                                 "RA with Prefix info",
 589                                 dst_ip=ll,
 590                                 opt=opt)
 591
 592         #
 593         # Change the prefix info to not off-link, no-autoconfig
 594         #  L and A flag are clear in the advert
 595         #
 596         self.pg0.ip6_ra_prefix(self.pg0.local_ip6n,
 597                                self.pg0.local_ip6_prefix_len,
 598                                off_link=1,
 599                                no_autoconfig=1)
 600
 601         opt = ICMPv6NDOptPrefixInfo(prefixlen=self.pg0.local_ip6_prefix_len,
 602                                     prefix=self.pg0.local_ip6,
 603                                     L=0,
 604                                     A=0)
 605
 606         self.pg0.ip6_ra_config(send_unicast=1)
 607         self.send_and_expect_ra(self.pg0, p,
 608                                 "RA with Prefix info with A & L-flag=0",
 609                                 dst_ip=ll,
 610                                 opt=opt)
 611
 612         #
 613         # Use the reset to defults option to revert to defaults
 614         #  L and A flag are clear in the advert
 615         #
 616         self.pg0.ip6_ra_prefix(self.pg0.local_ip6n,
 617                                self.pg0.local_ip6_prefix_len,
 618                                use_default=1)
 619
 620         opt = ICMPv6NDOptPrefixInfo(prefixlen=self.pg0.local_ip6_prefix_len,
 621                                     prefix=self.pg0.local_ip6,
 622                                     L=1,
 623                                     A=1)
 624
 625         self.pg0.ip6_ra_config(send_unicast=1)
 626         self.send_and_expect_ra(self.pg0, p,
 627                                 "RA with Prefix reverted to defaults",
 628                                 dst_ip=ll,
 629                                 opt=opt)
 630
 631         #
 632         # Advertise Another prefix. With no L-flag/A-flag
 633         #
 634         self.pg0.ip6_ra_prefix(self.pg1.local_ip6n,
 635                                self.pg1.local_ip6_prefix_len,
 636                                off_link=1,
 637                                no_autoconfig=1)
 638
 639         opt = [ICMPv6NDOptPrefixInfo(prefixlen=self.pg0.local_ip6_prefix_len,
 640                                      prefix=self.pg0.local_ip6,
 641                                      L=1,
 642                                      A=1),
 643                ICMPv6NDOptPrefixInfo(prefixlen=self.pg1.local_ip6_prefix_len,
 644                                      prefix=self.pg1.local_ip6,
 645                                      L=0,
 646                                      A=0)]
 647
 648         self.pg0.ip6_ra_config(send_unicast=1)
 649         ll = mk_ll_addr(self.pg0.remote_mac)
 650         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
 651              IPv6(dst=self.pg0.local_ip6, src=ll) /
 652              ICMPv6ND_RS())
 653         self.send_and_expect_ra(self.pg0, p,
 654                                 "RA with multiple Prefix infos",
 655                                 dst_ip=ll,
 656                                 opt=opt)
 657
 658         #
 659         # Remove the first refix-info - expect the second is still in the
 660         # advert
 661         #
 662         self.pg0.ip6_ra_prefix(self.pg0.local_ip6n,
 663                                self.pg0.local_ip6_prefix_len,
 664                                is_no=1)
 665
 666         opt = ICMPv6NDOptPrefixInfo(prefixlen=self.pg1.local_ip6_prefix_len,
 667                                     prefix=self.pg1.local_ip6,
 668                                     L=0,
 669                                     A=0)
 670
 671         self.pg0.ip6_ra_config(send_unicast=1)
 672         self.send_and_expect_ra(self.pg0, p,
 673                                 "RA with Prefix reverted to defaults",
 674                                 dst_ip=ll,
 675                                 opt=opt)
 676
 677         #
 678         # Remove the second prefix-info - expect no prefix-info i nthe adverts
 679         #
 680         self.pg0.ip6_ra_prefix(self.pg1.local_ip6n,
 681                                self.pg1.local_ip6_prefix_len,
 682                                is_no=1)
 683
 684         self.pg0.ip6_ra_config(send_unicast=1)
 685         self.send_and_expect_ra(self.pg0, p,
 686                                 "RA with Prefix reverted to defaults",
 687                                 dst_ip=ll)
 688
 689         #
 690         # Reset the periodic advertisements back to default values
 691         #
 692         self.pg0.ip6_ra_config(no=1, suppress=1, send_unicast=0)
 693
 694
 695 class IPv6NDProxyTest(TestIPv6ND):
 696     """ IPv6 ND ProxyTest Case """
 697
 698     def setUp(self):
 699         super(IPv6NDProxyTest, self).setUp()
 700
 701         # create 3 pg interfaces
 702         self.create_pg_interfaces(range(3))
 703
 704         # pg0 is the master interface, with the configured subnet
 705         self.pg0.admin_up()
 706         self.pg0.config_ip6()
 707         self.pg0.resolve_ndp()
 708
 709         self.pg1.ip6_enable()
 710         self.pg2.ip6_enable()
 711
 712     def tearDown(self):
 713         super(IPv6NDProxyTest, self).tearDown()
 714
 715     def test_nd_proxy(self):
 716         """ IPv6 Proxy ND """
 717
 718         #
 719         # Generate some hosts in the subnet that we are proxying
 720         #
 721         self.pg0.generate_remote_hosts(8)
 722
 723         nsma = in6_getnsma(inet_pton(AF_INET6, self.pg0.local_ip6))
 724         d = inet_ntop(AF_INET6, nsma)
 725
 726         #
 727         # Send an NS for one of those remote hosts on one of the proxy links
 728         # expect no response since it's from an address that is not
 729         # on the link that has the prefix configured
 730         #
 731         ns_pg1 = (Ether(dst=in6_getnsmac(nsma), src=self.pg1.remote_mac) /
 732                   IPv6(dst=d, src=self.pg0._remote_hosts[2].ip6) /
 733                   ICMPv6ND_NS(tgt=self.pg0.local_ip6) /
 734                   ICMPv6NDOptSrcLLAddr(lladdr=self.pg0._remote_hosts[2].mac))
 735
 736         self.send_and_assert_no_replies(self.pg1, ns_pg1, "Off link NS")
 737
 738         #
 739         # Add proxy support for the host
 740         #
 741         self.vapi.ip6_nd_proxy(
 742             inet_pton(AF_INET6, self.pg0._remote_hosts[2].ip6),
 743             self.pg1.sw_if_index)
 744
 745         #
 746         # try that NS again. this time we expect an NA back
 747         #
 748         self.pg1.add_stream(ns_pg1)
 749         self.pg_enable_capture(self.pg_interfaces)
 750         self.pg_start()
 751         rx = self.pg1.get_capture(1)
 752
 753         self.validate_na(self.pg1, rx[0],
 754                          dst_ip=self.pg0._remote_hosts[2].ip6,
 755                          tgt_ip=self.pg0.local_ip6)
 756
 757         #
 758         # ... and that we have an entry in the ND cache
 759         #
 760         self.assertTrue(find_nbr(self,
 761                                  self.pg1.sw_if_index,
 762                                  self.pg0._remote_hosts[2].ip6,
 763                                  inet=AF_INET6))
 764
 765         #
 766         # ... and we can route traffic to it
 767         #
 768         t = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
 769              IPv6(dst=self.pg0._remote_hosts[2].ip6,
 770                   src=self.pg0.remote_ip6) /
 771              UDP(sport=10000, dport=20000) /
 772              Raw('\xa5' * 100))
 773
 774         self.pg0.add_stream(t)
 775         self.pg_enable_capture(self.pg_interfaces)
 776         self.pg_start()
 777         rx = self.pg1.get_capture(1)
 778         rx = rx[0]
 779
 780         self.assertEqual(rx[Ether].dst, self.pg0._remote_hosts[2].mac)
 781         self.assertEqual(rx[Ether].src, self.pg1.local_mac)
 782
 783         self.assertEqual(rx[IPv6].src, t[IPv6].src)
 784         self.assertEqual(rx[IPv6].dst, t[IPv6].dst)
 785
 786         #
 787         # Test we proxy for the host on the main interface
 788         #
 789         ns_pg0 = (Ether(dst=in6_getnsmac(nsma), src=self.pg0.remote_mac) /
 790                   IPv6(dst=d, src=self.pg0.remote_ip6) /
 791                   ICMPv6ND_NS(tgt=self.pg0._remote_hosts[2].ip6) /
 792                   ICMPv6NDOptSrcLLAddr(lladdr=self.pg0.remote_mac))
 793
 794         self.pg0.add_stream(ns_pg0)
 795         self.pg_enable_capture(self.pg_interfaces)
 796         self.pg_start()
 797         rx = self.pg0.get_capture(1)
 798
 799         self.validate_na(self.pg0, rx[0],
 800                          tgt_ip=self.pg0._remote_hosts[2].ip6,
 801                          dst_ip=self.pg0.remote_ip6)
 802
 803         #
 804         # Setup and resolve proxy for another host on another interface
 805         #
 806         ns_pg2 = (Ether(dst=in6_getnsmac(nsma), src=self.pg2.remote_mac) /
 807                   IPv6(dst=d, src=self.pg0._remote_hosts[3].ip6) /
 808                   ICMPv6ND_NS(tgt=self.pg0.local_ip6) /
 809                   ICMPv6NDOptSrcLLAddr(lladdr=self.pg0._remote_hosts[2].mac))
 810
 811         self.vapi.ip6_nd_proxy(
 812             inet_pton(AF_INET6, self.pg0._remote_hosts[3].ip6),
 813             self.pg2.sw_if_index)
 814
 815         self.pg2.add_stream(ns_pg2)
 816         self.pg_enable_capture(self.pg_interfaces)
 817         self.pg_start()
 818         rx = self.pg2.get_capture(1)
 819
 820         self.validate_na(self.pg2, rx[0],
 821                          dst_ip=self.pg0._remote_hosts[3].ip6,
 822                          tgt_ip=self.pg0.local_ip6)
 823
 824         self.assertTrue(find_nbr(self,
 825                                  self.pg2.sw_if_index,
 826                                  self.pg0._remote_hosts[3].ip6,
 827                                  inet=AF_INET6))
 828
 829         #
 830         # hosts can communicate. pg2->pg1
 831         #
 832         t2 = (Ether(dst=self.pg2.local_mac,
 833                     src=self.pg0.remote_hosts[3].mac) /
 834               IPv6(dst=self.pg0._remote_hosts[2].ip6,
 835                    src=self.pg0._remote_hosts[3].ip6) /
 836               UDP(sport=10000, dport=20000) /
 837               Raw('\xa5' * 100))
 838
 839         self.pg2.add_stream(t2)
 840         self.pg_enable_capture(self.pg_interfaces)
 841         self.pg_start()
 842         rx = self.pg1.get_capture(1)
 843         rx = rx[0]
 844
 845         self.assertEqual(rx[Ether].dst, self.pg0._remote_hosts[2].mac)
 846         self.assertEqual(rx[Ether].src, self.pg1.local_mac)
 847
 848         self.assertEqual(rx[IPv6].src, t2[IPv6].src)
 849         self.assertEqual(rx[IPv6].dst, t2[IPv6].dst)
 850
 851         #
 852         # remove the proxy configs
 853         #
 854         self.vapi.ip6_nd_proxy(
 855             inet_pton(AF_INET6, self.pg0._remote_hosts[2].ip6),
 856             self.pg1.sw_if_index,
 857             is_del=1)
 858         self.vapi.ip6_nd_proxy(
 859             inet_pton(AF_INET6, self.pg0._remote_hosts[3].ip6),
 860             self.pg2.sw_if_index,
 861             is_del=1)
 862
 863         self.assertFalse(find_nbr(self,
 864                                   self.pg2.sw_if_index,
 865                                   self.pg0._remote_hosts[3].ip6,
 866                                   inet=AF_INET6))
 867         self.assertFalse(find_nbr(self,
 868                                   self.pg1.sw_if_index,
 869                                   self.pg0._remote_hosts[2].ip6,
 870                                   inet=AF_INET6))
 871
 872         #
 873         # no longer proxy-ing...
 874         #
 875         self.send_and_assert_no_replies(self.pg0, ns_pg0, "Proxy unconfigured")
 876         self.send_and_assert_no_replies(self.pg1, ns_pg1, "Proxy unconfigured")
 877         self.send_and_assert_no_replies(self.pg2, ns_pg2, "Proxy unconfigured")
 878
 879         #
 880         # no longer forwarding. traffic generates NS out of the glean/main
 881         # interface
 882         #
 883         self.pg2.add_stream(t2)
 884         self.pg_enable_capture(self.pg_interfaces)
 885         self.pg_start()
 886
 887         rx = self.pg0.get_capture(1)
 888
 889         self.assertTrue(rx[0].haslayer(ICMPv6ND_NS))
 890
 891
 892 class TestIPNull(VppTestCase):
 893     """ IPv6 routes via NULL """
 894
 895     def setUp(self):
 896         super(TestIPNull, self).setUp()
 897
 898         # create 2 pg interfaces
 899         self.create_pg_interfaces(range(1))
 900
 901         for i in self.pg_interfaces:
 902             i.admin_up()
 903             i.config_ip6()
 904             i.resolve_ndp()
 905
 906     def tearDown(self):
 907         super(TestIPNull, self).tearDown()
 908         for i in self.pg_interfaces:
 909             i.unconfig_ip6()
 910             i.admin_down()
 911
 912     def test_ip_null(self):
 913         """ IP NULL route """
 914
 915         p = (Ether(src=self.pg0.remote_mac,
 916                    dst=self.pg0.local_mac) /
 917              IPv6(src=self.pg0.remote_ip6, dst="2001::1") /
 918              UDP(sport=1234, dport=1234) /
 919              Raw('\xa5' * 100))
 920
 921         #
 922         # A route via IP NULL that will reply with ICMP unreachables
 923         #
 924         ip_unreach = VppIpRoute(self, "2001::", 64, [], is_unreach=1, is_ip6=1)
 925         ip_unreach.add_vpp_config()
 926
 927         self.pg0.add_stream(p)
 928         self.pg_enable_capture(self.pg_interfaces)
 929         self.pg_start()
 930
 931         rx = self.pg0.get_capture(1)
 932         rx = rx[0]
 933         icmp = rx[ICMPv6DestUnreach]
 934
 935         # 0 = "No route to destination"
 936         self.assertEqual(icmp.code, 0)
 937
 938         # ICMP is rate limited. pause a bit
 939         self.sleep(1)
 940
 941         #
 942         # A route via IP NULL that will reply with ICMP prohibited
 943         #
 944         ip_prohibit = VppIpRoute(self, "2001::1", 128, [],
 945                                  is_prohibit=1, is_ip6=1)
 946         ip_prohibit.add_vpp_config()
 947
 948         self.pg0.add_stream(p)
 949         self.pg_enable_capture(self.pg_interfaces)
 950         self.pg_start()
 951
 952         rx = self.pg0.get_capture(1)
 953         rx = rx[0]
 954         icmp = rx[ICMPv6DestUnreach]
 955
 956         # 1 = "Communication with destination administratively prohibited"
 957         self.assertEqual(icmp.code, 1)
 958
 959
 960 if __name__ == '__main__':
 961     unittest.main(testRunner=VppTestRunner)