4 from socket import AF_INET6
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_ip_route import VppIpRoute, VppRoutePath, find_route, VppIpMRoute, \
10 VppMRoutePath, MRouteItfFlags, MRouteEntryFlags
11 from vpp_neighbor import find_nbr, VppNeighbor
13 from scapy.packet import Raw
14 from scapy.layers.l2 import Ether, Dot1Q
15 from scapy.layers.inet6 import IPv6, UDP, ICMPv6ND_NS, ICMPv6ND_RS, \
16 ICMPv6ND_RA, ICMPv6NDOptSrcLLAddr, getmacbyip6, ICMPv6MRD_Solicitation, \
17 ICMPv6NDOptMTU, ICMPv6NDOptSrcLLAddr, ICMPv6NDOptPrefixInfo, \
18 ICMPv6ND_NA, ICMPv6NDOptDstLLAddr, ICMPv6DestUnreach, icmp6types
21 from scapy.utils6 import in6_getnsma, in6_getnsmac, in6_ptop, in6_islladdr, \
22 in6_mactoifaceid, in6_ismaddr
23 from scapy.utils import inet_pton, inet_ntop
27 euid = in6_mactoifaceid(mac)
28 addr = "fe80::" + euid
32 class TestIPv6ND(VppTestCase):
33 def validate_ra(self, intf, rx, dst_ip=None):
35 dst_ip = intf.remote_ip6
37 # unicasted packets must come to the unicast mac
38 self.assertEqual(rx[Ether].dst, intf.remote_mac)
40 # and from the router's MAC
41 self.assertEqual(rx[Ether].src, intf.local_mac)
43 # the rx'd RA should be addressed to the sender's source
44 self.assertTrue(rx.haslayer(ICMPv6ND_RA))
45 self.assertEqual(in6_ptop(rx[IPv6].dst),
48 # and come from the router's link local
49 self.assertTrue(in6_islladdr(rx[IPv6].src))
50 self.assertEqual(in6_ptop(rx[IPv6].src),
51 in6_ptop(mk_ll_addr(intf.local_mac)))
53 def validate_na(self, intf, rx, dst_ip=None, tgt_ip=None):
55 dst_ip = intf.remote_ip6
57 dst_ip = intf.local_ip6
59 # unicasted packets must come to the unicast mac
60 self.assertEqual(rx[Ether].dst, intf.remote_mac)
62 # and from the router's MAC
63 self.assertEqual(rx[Ether].src, intf.local_mac)
65 # the rx'd NA should be addressed to the sender's source
66 self.assertTrue(rx.haslayer(ICMPv6ND_NA))
67 self.assertEqual(in6_ptop(rx[IPv6].dst),
70 # and come from the target address
71 self.assertEqual(in6_ptop(rx[IPv6].src), in6_ptop(tgt_ip))
73 # Dest link-layer options should have the router's MAC
74 dll = rx[ICMPv6NDOptDstLLAddr]
75 self.assertEqual(dll.lladdr, intf.local_mac)
77 def send_and_expect_ra(self, intf, pkts, remark, dst_ip=None,
78 filter_out_fn=is_ipv6_misc):
80 self.pg_enable_capture(self.pg_interfaces)
82 rx = intf.get_capture(1, filter_out_fn=filter_out_fn)
84 self.assertEqual(len(rx), 1)
86 self.validate_ra(intf, rx, dst_ip)
88 def send_and_expect_na(self, intf, pkts, remark, dst_ip=None,
90 filter_out_fn=is_ipv6_misc):
92 self.pg_enable_capture(self.pg_interfaces)
94 rx = intf.get_capture(1, filter_out_fn=filter_out_fn)
96 self.assertEqual(len(rx), 1)
98 self.validate_na(intf, rx, dst_ip, tgt_ip)
100 def send_and_assert_no_replies(self, intf, pkts, remark):
101 intf.add_stream(pkts)
102 self.pg_enable_capture(self.pg_interfaces)
104 for i in self.pg_interfaces:
106 i.assert_nothing_captured(remark=remark)
109 class TestIPv6(TestIPv6ND):
110 """ IPv6 Test Case """
114 super(TestIPv6, cls).setUpClass()
118 Perform test setup before test case.
121 - create 3 pg interfaces
122 - untagged pg0 interface
123 - Dot1Q subinterface on pg1
124 - Dot1AD subinterface on pg2
126 - put it into UP state
128 - resolve neighbor address using NDP
129 - configure 200 fib entries
131 :ivar list interfaces: pg interfaces and subinterfaces.
132 :ivar dict flows: IPv4 packet flows in test.
133 :ivar list pg_if_packet_sizes: packet sizes in test.
135 *TODO:* Create AD sub interface
137 super(TestIPv6, self).setUp()
139 # create 3 pg interfaces
140 self.create_pg_interfaces(range(3))
142 # create 2 subinterfaces for p1 and pg2
143 self.sub_interfaces = [
144 VppDot1QSubint(self, self.pg1, 100),
145 VppDot1QSubint(self, self.pg2, 200)
146 # TODO: VppDot1ADSubint(self, self.pg2, 200, 300, 400)
149 # packet flows mapping pg0 -> pg1.sub, pg2.sub, etc.
151 self.flows[self.pg0] = [self.pg1.sub_if, self.pg2.sub_if]
152 self.flows[self.pg1.sub_if] = [self.pg0, self.pg2.sub_if]
153 self.flows[self.pg2.sub_if] = [self.pg0, self.pg1.sub_if]
156 self.pg_if_packet_sizes = [64, 512, 1518, 9018]
157 self.sub_if_packet_sizes = [64, 512, 1518 + 4, 9018 + 4]
159 self.interfaces = list(self.pg_interfaces)
160 self.interfaces.extend(self.sub_interfaces)
162 # setup all interfaces
163 for i in self.interfaces:
168 # config 2M FIB entries
169 self.config_fib_entries(200)
172 """Run standard test teardown and log ``show ip6 neighbors``."""
173 for i in self.sub_interfaces:
177 i.remove_vpp_config()
179 super(TestIPv6, self).tearDown()
180 if not self.vpp_dead:
181 self.logger.info(self.vapi.cli("show ip6 neighbors"))
182 # info(self.vapi.cli("show ip6 fib")) # many entries
184 def config_fib_entries(self, count):
185 """For each interface add to the FIB table *count* routes to
186 "fd02::1/128" destination with interface's local address as next-hop
189 :param int count: Number of FIB entries.
191 - *TODO:* check if the next-hop address shouldn't be remote address
192 instead of local address.
194 n_int = len(self.interfaces)
197 dest_addr = inet_pton(AF_INET6, "fd02::1")
199 for i in self.interfaces:
200 next_hop_address = i.local_ip6n
201 for j in range(count / n_int):
202 self.vapi.ip_add_del_route(
203 dest_addr, dest_addr_len, next_hop_address, is_ipv6=1)
205 if counter / count * 100 > percent:
206 self.logger.info("Configure %d FIB entries .. %d%% done" %
210 def create_stream(self, src_if, packet_sizes):
211 """Create input packet stream for defined interface.
213 :param VppInterface src_if: Interface to create packet stream for.
214 :param list packet_sizes: Required packet sizes.
217 for i in range(0, 257):
218 dst_if = self.flows[src_if][i % 2]
219 info = self.create_packet_info(src_if, dst_if)
220 payload = self.info_to_payload(info)
221 p = (Ether(dst=src_if.local_mac, src=src_if.remote_mac) /
222 IPv6(src=src_if.remote_ip6, dst=dst_if.remote_ip6) /
223 UDP(sport=1234, dport=1234) /
226 if isinstance(src_if, VppSubInterface):
227 p = src_if.add_dot1_layer(p)
228 size = packet_sizes[(i // 2) % len(packet_sizes)]
229 self.extend_packet(p, size)
233 def verify_capture(self, dst_if, capture):
234 """Verify captured input packet stream for defined interface.
236 :param VppInterface dst_if: Interface to verify captured packet stream
238 :param list capture: Captured packet stream.
240 self.logger.info("Verifying capture on interface %s" % dst_if.name)
242 for i in self.interfaces:
243 last_info[i.sw_if_index] = None
245 dst_sw_if_index = dst_if.sw_if_index
246 if hasattr(dst_if, 'parent'):
248 for packet in capture:
250 # Check VLAN tags and Ethernet header
251 packet = dst_if.remove_dot1_layer(packet)
252 self.assertTrue(Dot1Q not in packet)
256 payload_info = self.payload_to_info(str(packet[Raw]))
257 packet_index = payload_info.index
258 self.assertEqual(payload_info.dst, dst_sw_if_index)
260 "Got packet on port %s: src=%u (id=%u)" %
261 (dst_if.name, payload_info.src, packet_index))
262 next_info = self.get_next_packet_info_for_interface2(
263 payload_info.src, dst_sw_if_index,
264 last_info[payload_info.src])
265 last_info[payload_info.src] = next_info
266 self.assertTrue(next_info is not None)
267 self.assertEqual(packet_index, next_info.index)
268 saved_packet = next_info.data
269 # Check standard fields
270 self.assertEqual(ip.src, saved_packet[IPv6].src)
271 self.assertEqual(ip.dst, saved_packet[IPv6].dst)
272 self.assertEqual(udp.sport, saved_packet[UDP].sport)
273 self.assertEqual(udp.dport, saved_packet[UDP].dport)
275 self.logger.error(ppp("Unexpected or invalid packet:", packet))
277 for i in self.interfaces:
278 remaining_packet = self.get_next_packet_info_for_interface2(
279 i.sw_if_index, dst_sw_if_index, last_info[i.sw_if_index])
280 self.assertTrue(remaining_packet is None,
281 "Interface %s: Packet expected from interface %s "
282 "didn't arrive" % (dst_if.name, i.name))
288 - Create IPv6 stream for pg0 interface
289 - Create IPv6 tagged streams for pg1's and pg2's subinterface.
290 - Send and verify received packets on each interface.
293 pkts = self.create_stream(self.pg0, self.pg_if_packet_sizes)
294 self.pg0.add_stream(pkts)
296 for i in self.sub_interfaces:
297 pkts = self.create_stream(i, self.sub_if_packet_sizes)
298 i.parent.add_stream(pkts)
300 self.pg_enable_capture(self.pg_interfaces)
303 pkts = self.pg0.get_capture()
304 self.verify_capture(self.pg0, pkts)
306 for i in self.sub_interfaces:
307 pkts = i.parent.get_capture()
308 self.verify_capture(i, pkts)
311 """ IPv6 Neighbour Solicitation Exceptions
314 - Send an NS Sourced from an address not covered by the link sub-net
315 - Send an NS to an mcast address the router has not joined
316 - Send NS for a target address the router does not onn.
320 # An NS from a non link source address
322 nsma = in6_getnsma(inet_pton(AF_INET6, self.pg0.local_ip6))
323 d = inet_ntop(AF_INET6, nsma)
325 p = (Ether(dst=in6_getnsmac(nsma)) /
326 IPv6(dst=d, src="2002::2") /
327 ICMPv6ND_NS(tgt=self.pg0.local_ip6) /
328 ICMPv6NDOptSrcLLAddr(lladdr=self.pg0.remote_mac))
331 self.send_and_assert_no_replies(
333 "No response to NS source by address not on sub-net")
336 # An NS for sent to a solicited mcast group the router is
337 # not a member of FAILS
340 nsma = in6_getnsma(inet_pton(AF_INET6, "fd::ffff"))
341 d = inet_ntop(AF_INET6, nsma)
343 p = (Ether(dst=in6_getnsmac(nsma)) /
344 IPv6(dst=d, src=self.pg0.remote_ip6) /
345 ICMPv6ND_NS(tgt=self.pg0.local_ip6) /
346 ICMPv6NDOptSrcLLAddr(lladdr=self.pg0.remote_mac))
349 self.send_and_assert_no_replies(
351 "No response to NS sent to unjoined mcast address")
354 # An NS whose target address is one the router does not own
356 nsma = in6_getnsma(inet_pton(AF_INET6, self.pg0.local_ip6))
357 d = inet_ntop(AF_INET6, nsma)
359 p = (Ether(dst=in6_getnsmac(nsma)) /
360 IPv6(dst=d, src=self.pg0.remote_ip6) /
361 ICMPv6ND_NS(tgt="fd::ffff") /
362 ICMPv6NDOptSrcLLAddr(lladdr=self.pg0.remote_mac))
365 self.send_and_assert_no_replies(self.pg0, pkts,
366 "No response to NS for unknown target")
369 # A neighbor entry that has no associated FIB-entry
371 self.pg0.generate_remote_hosts(4)
372 nd_entry = VppNeighbor(self,
373 self.pg0.sw_if_index,
374 self.pg0.remote_hosts[2].mac,
375 self.pg0.remote_hosts[2].ip6,
378 nd_entry.add_vpp_config()
381 # check we have the neighbor, but no route
383 self.assertTrue(find_nbr(self,
384 self.pg0.sw_if_index,
385 self.pg0._remote_hosts[2].ip6,
387 self.assertFalse(find_route(self,
388 self.pg0._remote_hosts[2].ip6,
393 # send an NS from a link local address to the interface's global
396 p = (Ether(dst=in6_getnsmac(nsma), src=self.pg0.remote_mac) /
397 IPv6(dst=d, src=self.pg0._remote_hosts[2].ip6_ll) /
398 ICMPv6ND_NS(tgt=self.pg0.local_ip6) /
399 ICMPv6NDOptSrcLLAddr(lladdr=self.pg0.remote_mac))
401 self.send_and_expect_na(self.pg0, p,
402 "NS from link-local",
403 dst_ip=self.pg0._remote_hosts[2].ip6_ll,
404 tgt_ip=self.pg0.local_ip6)
407 # we should have learned an ND entry for the peer's link-local
408 # but not inserted a route to it in the FIB
410 self.assertTrue(find_nbr(self,
411 self.pg0.sw_if_index,
412 self.pg0._remote_hosts[2].ip6_ll,
414 self.assertFalse(find_route(self,
415 self.pg0._remote_hosts[2].ip6_ll,
420 # An NS to the router's own Link-local
422 p = (Ether(dst=in6_getnsmac(nsma), src=self.pg0.remote_mac) /
423 IPv6(dst=d, src=self.pg0._remote_hosts[3].ip6_ll) /
424 ICMPv6ND_NS(tgt=self.pg0.local_ip6_ll) /
425 ICMPv6NDOptSrcLLAddr(lladdr=self.pg0.remote_mac))
427 self.send_and_expect_na(self.pg0, p,
428 "NS to/from link-local",
429 dst_ip=self.pg0._remote_hosts[3].ip6_ll,
430 tgt_ip=self.pg0.local_ip6_ll)
433 # we should have learned an ND entry for the peer's link-local
434 # but not inserted a route to it in the FIB
436 self.assertTrue(find_nbr(self,
437 self.pg0.sw_if_index,
438 self.pg0._remote_hosts[3].ip6_ll,
440 self.assertFalse(find_route(self,
441 self.pg0._remote_hosts[3].ip6_ll,
445 def validate_ra(self, intf, rx, dst_ip=None, mtu=9000, pi_opt=None):
447 dst_ip = intf.remote_ip6
449 # unicasted packets must come to the unicast mac
450 self.assertEqual(rx[Ether].dst, intf.remote_mac)
452 # and from the router's MAC
453 self.assertEqual(rx[Ether].src, intf.local_mac)
455 # the rx'd RA should be addressed to the sender's source
456 self.assertTrue(rx.haslayer(ICMPv6ND_RA))
457 self.assertEqual(in6_ptop(rx[IPv6].dst),
460 # and come from the router's link local
461 self.assertTrue(in6_islladdr(rx[IPv6].src))
462 self.assertEqual(in6_ptop(rx[IPv6].src),
463 in6_ptop(mk_ll_addr(intf.local_mac)))
465 # it should contain the links MTU
467 self.assertEqual(ra[ICMPv6NDOptMTU].mtu, mtu)
469 # it should contain the source's link layer address option
470 sll = ra[ICMPv6NDOptSrcLLAddr]
471 self.assertEqual(sll.lladdr, intf.local_mac)
474 # the RA should not contain prefix information
475 self.assertFalse(ra.haslayer(ICMPv6NDOptPrefixInfo))
477 raos = rx.getlayer(ICMPv6NDOptPrefixInfo, 1)
479 # the options are nested in the scapy packet in way that i cannot
480 # decipher how to decode. this 1st layer of option always returns
481 # nested classes, so a direct obj1=obj2 comparison always fails.
482 # however, the getlayer(.., 2) does give one instnace.
483 # so we cheat here and construct a new opt instnace for comparison
484 rd = ICMPv6NDOptPrefixInfo(prefixlen=raos.prefixlen,
488 if type(pi_opt) is list:
489 for ii in range(len(pi_opt)):
490 self.assertEqual(pi_opt[ii], rd)
491 rd = rx.getlayer(ICMPv6NDOptPrefixInfo, ii+2)
493 self.assertEqual(pi_opt, raos)
495 def send_and_expect_ra(self, intf, pkts, remark, dst_ip=None,
496 filter_out_fn=is_ipv6_misc,
498 intf.add_stream(pkts)
499 self.pg_enable_capture(self.pg_interfaces)
501 rx = intf.get_capture(1, filter_out_fn=filter_out_fn)
503 self.assertEqual(len(rx), 1)
505 self.validate_ra(intf, rx, dst_ip, pi_opt=opt)
508 """ IPv6 Router Solicitation Exceptions
514 # Before we begin change the IPv6 RA responses to use the unicast
515 # address - that way we will not confuse them with the periodic
516 # RAs which go to the mcast address
517 # Sit and wait for the first periodic RA.
521 self.pg0.ip6_ra_config(send_unicast=1)
524 # An RS from a link source address
525 # - expect an RA in return
527 p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
528 IPv6(dst=self.pg0.local_ip6, src=self.pg0.remote_ip6) /
531 self.send_and_expect_ra(self.pg0, pkts, "Genuine RS")
534 # For the next RS sent the RA should be rate limited
536 self.send_and_assert_no_replies(self.pg0, pkts, "RA rate limited")
539 # When we reconfiure the IPv6 RA config, we reset the RA rate limiting,
540 # so we need to do this before each test below so as not to drop
541 # packets for rate limiting reasons. Test this works here.
543 self.pg0.ip6_ra_config(send_unicast=1)
544 self.send_and_expect_ra(self.pg0, pkts, "Rate limit reset RS")
547 # An RS sent from a non-link local source
549 self.pg0.ip6_ra_config(send_unicast=1)
550 p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
551 IPv6(dst=self.pg0.local_ip6, src="2002::ffff") /
554 self.send_and_assert_no_replies(self.pg0, pkts,
555 "RS from non-link source")
558 # Source an RS from a link local address
560 self.pg0.ip6_ra_config(send_unicast=1)
561 ll = mk_ll_addr(self.pg0.remote_mac)
562 p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
563 IPv6(dst=self.pg0.local_ip6, src=ll) /
566 self.send_and_expect_ra(self.pg0, pkts,
567 "RS sourced from link-local",
571 # Send the RS multicast
573 self.pg0.ip6_ra_config(send_unicast=1)
574 dmac = in6_getnsmac(inet_pton(AF_INET6, "ff02::2"))
575 ll = mk_ll_addr(self.pg0.remote_mac)
576 p = (Ether(dst=dmac, src=self.pg0.remote_mac) /
577 IPv6(dst="ff02::2", src=ll) /
580 self.send_and_expect_ra(self.pg0, pkts,
581 "RS sourced from link-local",
585 # Source from the unspecified address ::. This happens when the RS
586 # is sent before the host has a configured address/sub-net,
587 # i.e. auto-config. Since the sender has no IP address, the reply
588 # comes back mcast - so the capture needs to not filter this.
589 # If we happen to pick up the periodic RA at this point then so be it,
592 self.pg0.ip6_ra_config(send_unicast=1, suppress=1)
593 p = (Ether(dst=dmac, src=self.pg0.remote_mac) /
594 IPv6(dst="ff02::2", src="::") /
597 self.send_and_expect_ra(self.pg0, pkts,
598 "RS sourced from unspecified",
603 # Configure The RA to announce the links prefix
605 self.pg0.ip6_ra_prefix(self.pg0.local_ip6n,
606 self.pg0.local_ip6_prefix_len)
609 # RAs should now contain the prefix information option
611 opt = ICMPv6NDOptPrefixInfo(prefixlen=self.pg0.local_ip6_prefix_len,
612 prefix=self.pg0.local_ip6,
616 self.pg0.ip6_ra_config(send_unicast=1)
617 ll = mk_ll_addr(self.pg0.remote_mac)
618 p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
619 IPv6(dst=self.pg0.local_ip6, src=ll) /
621 self.send_and_expect_ra(self.pg0, p,
622 "RA with prefix-info",
627 # Change the prefix info to not off-link
630 self.pg0.ip6_ra_prefix(self.pg0.local_ip6n,
631 self.pg0.local_ip6_prefix_len,
634 opt = ICMPv6NDOptPrefixInfo(prefixlen=self.pg0.local_ip6_prefix_len,
635 prefix=self.pg0.local_ip6,
639 self.pg0.ip6_ra_config(send_unicast=1)
640 self.send_and_expect_ra(self.pg0, p,
641 "RA with Prefix info with L-flag=0",
646 # Change the prefix info to not off-link, no-autoconfig
647 # L and A flag are clear in the advert
649 self.pg0.ip6_ra_prefix(self.pg0.local_ip6n,
650 self.pg0.local_ip6_prefix_len,
654 opt = ICMPv6NDOptPrefixInfo(prefixlen=self.pg0.local_ip6_prefix_len,
655 prefix=self.pg0.local_ip6,
659 self.pg0.ip6_ra_config(send_unicast=1)
660 self.send_and_expect_ra(self.pg0, p,
661 "RA with Prefix info with A & L-flag=0",
666 # Change the flag settings back to the defaults
667 # L and A flag are set in the advert
669 self.pg0.ip6_ra_prefix(self.pg0.local_ip6n,
670 self.pg0.local_ip6_prefix_len)
672 opt = ICMPv6NDOptPrefixInfo(prefixlen=self.pg0.local_ip6_prefix_len,
673 prefix=self.pg0.local_ip6,
677 self.pg0.ip6_ra_config(send_unicast=1)
678 self.send_and_expect_ra(self.pg0, p,
679 "RA with Prefix info",
684 # Change the prefix info to not off-link, no-autoconfig
685 # L and A flag are clear in the advert
687 self.pg0.ip6_ra_prefix(self.pg0.local_ip6n,
688 self.pg0.local_ip6_prefix_len,
692 opt = ICMPv6NDOptPrefixInfo(prefixlen=self.pg0.local_ip6_prefix_len,
693 prefix=self.pg0.local_ip6,
697 self.pg0.ip6_ra_config(send_unicast=1)
698 self.send_and_expect_ra(self.pg0, p,
699 "RA with Prefix info with A & L-flag=0",
704 # Use the reset to defults option to revert to defaults
705 # L and A flag are clear in the advert
707 self.pg0.ip6_ra_prefix(self.pg0.local_ip6n,
708 self.pg0.local_ip6_prefix_len,
711 opt = ICMPv6NDOptPrefixInfo(prefixlen=self.pg0.local_ip6_prefix_len,
712 prefix=self.pg0.local_ip6,
716 self.pg0.ip6_ra_config(send_unicast=1)
717 self.send_and_expect_ra(self.pg0, p,
718 "RA with Prefix reverted to defaults",
723 # Advertise Another prefix. With no L-flag/A-flag
725 self.pg0.ip6_ra_prefix(self.pg1.local_ip6n,
726 self.pg1.local_ip6_prefix_len,
730 opt = [ICMPv6NDOptPrefixInfo(prefixlen=self.pg0.local_ip6_prefix_len,
731 prefix=self.pg0.local_ip6,
734 ICMPv6NDOptPrefixInfo(prefixlen=self.pg1.local_ip6_prefix_len,
735 prefix=self.pg1.local_ip6,
739 self.pg0.ip6_ra_config(send_unicast=1)
740 ll = mk_ll_addr(self.pg0.remote_mac)
741 p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
742 IPv6(dst=self.pg0.local_ip6, src=ll) /
744 self.send_and_expect_ra(self.pg0, p,
745 "RA with multiple Prefix infos",
750 # Remove the first refix-info - expect the second is still in the
753 self.pg0.ip6_ra_prefix(self.pg0.local_ip6n,
754 self.pg0.local_ip6_prefix_len,
757 opt = ICMPv6NDOptPrefixInfo(prefixlen=self.pg1.local_ip6_prefix_len,
758 prefix=self.pg1.local_ip6,
762 self.pg0.ip6_ra_config(send_unicast=1)
763 self.send_and_expect_ra(self.pg0, p,
764 "RA with Prefix reverted to defaults",
769 # Remove the second prefix-info - expect no prefix-info i nthe adverts
771 self.pg0.ip6_ra_prefix(self.pg1.local_ip6n,
772 self.pg1.local_ip6_prefix_len,
775 self.pg0.ip6_ra_config(send_unicast=1)
776 self.send_and_expect_ra(self.pg0, p,
777 "RA with Prefix reverted to defaults",
781 # Reset the periodic advertisements back to default values
783 self.pg0.ip6_ra_config(no=1, suppress=1, send_unicast=0)
786 class IPv6NDProxyTest(TestIPv6ND):
787 """ IPv6 ND ProxyTest Case """
790 super(IPv6NDProxyTest, self).setUp()
792 # create 3 pg interfaces
793 self.create_pg_interfaces(range(3))
795 # pg0 is the master interface, with the configured subnet
797 self.pg0.config_ip6()
798 self.pg0.resolve_ndp()
800 self.pg1.ip6_enable()
801 self.pg2.ip6_enable()
804 super(IPv6NDProxyTest, self).tearDown()
806 def test_nd_proxy(self):
807 """ IPv6 Proxy ND """
810 # Generate some hosts in the subnet that we are proxying
812 self.pg0.generate_remote_hosts(8)
814 nsma = in6_getnsma(inet_pton(AF_INET6, self.pg0.local_ip6))
815 d = inet_ntop(AF_INET6, nsma)
818 # Send an NS for one of those remote hosts on one of the proxy links
819 # expect no response since it's from an address that is not
820 # on the link that has the prefix configured
822 ns_pg1 = (Ether(dst=in6_getnsmac(nsma), src=self.pg1.remote_mac) /
823 IPv6(dst=d, src=self.pg0._remote_hosts[2].ip6) /
824 ICMPv6ND_NS(tgt=self.pg0.local_ip6) /
825 ICMPv6NDOptSrcLLAddr(lladdr=self.pg0._remote_hosts[2].mac))
827 self.send_and_assert_no_replies(self.pg1, ns_pg1, "Off link NS")
830 # Add proxy support for the host
832 self.vapi.ip6_nd_proxy(
833 inet_pton(AF_INET6, self.pg0._remote_hosts[2].ip6),
834 self.pg1.sw_if_index)
837 # try that NS again. this time we expect an NA back
839 self.send_and_expect_na(self.pg1, ns_pg1,
841 dst_ip=self.pg0._remote_hosts[2].ip6,
842 tgt_ip=self.pg0.local_ip6)
845 # ... and that we have an entry in the ND cache
847 self.assertTrue(find_nbr(self,
848 self.pg1.sw_if_index,
849 self.pg0._remote_hosts[2].ip6,
853 # ... and we can route traffic to it
855 t = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
856 IPv6(dst=self.pg0._remote_hosts[2].ip6,
857 src=self.pg0.remote_ip6) /
858 UDP(sport=10000, dport=20000) /
861 self.pg0.add_stream(t)
862 self.pg_enable_capture(self.pg_interfaces)
864 rx = self.pg1.get_capture(1)
867 self.assertEqual(rx[Ether].dst, self.pg0._remote_hosts[2].mac)
868 self.assertEqual(rx[Ether].src, self.pg1.local_mac)
870 self.assertEqual(rx[IPv6].src, t[IPv6].src)
871 self.assertEqual(rx[IPv6].dst, t[IPv6].dst)
874 # Test we proxy for the host on the main interface
876 ns_pg0 = (Ether(dst=in6_getnsmac(nsma), src=self.pg0.remote_mac) /
877 IPv6(dst=d, src=self.pg0.remote_ip6) /
878 ICMPv6ND_NS(tgt=self.pg0._remote_hosts[2].ip6) /
879 ICMPv6NDOptSrcLLAddr(lladdr=self.pg0.remote_mac))
881 self.send_and_expect_na(self.pg0, ns_pg0,
882 "NS to proxy entry on main",
883 tgt_ip=self.pg0._remote_hosts[2].ip6,
884 dst_ip=self.pg0.remote_ip6)
887 # Setup and resolve proxy for another host on another interface
889 ns_pg2 = (Ether(dst=in6_getnsmac(nsma), src=self.pg2.remote_mac) /
890 IPv6(dst=d, src=self.pg0._remote_hosts[3].ip6) /
891 ICMPv6ND_NS(tgt=self.pg0.local_ip6) /
892 ICMPv6NDOptSrcLLAddr(lladdr=self.pg0._remote_hosts[2].mac))
894 self.vapi.ip6_nd_proxy(
895 inet_pton(AF_INET6, self.pg0._remote_hosts[3].ip6),
896 self.pg2.sw_if_index)
898 self.send_and_expect_na(self.pg2, ns_pg2,
899 "NS to proxy entry other interface",
900 dst_ip=self.pg0._remote_hosts[3].ip6,
901 tgt_ip=self.pg0.local_ip6)
903 self.assertTrue(find_nbr(self,
904 self.pg2.sw_if_index,
905 self.pg0._remote_hosts[3].ip6,
909 # hosts can communicate. pg2->pg1
911 t2 = (Ether(dst=self.pg2.local_mac,
912 src=self.pg0.remote_hosts[3].mac) /
913 IPv6(dst=self.pg0._remote_hosts[2].ip6,
914 src=self.pg0._remote_hosts[3].ip6) /
915 UDP(sport=10000, dport=20000) /
918 self.pg2.add_stream(t2)
919 self.pg_enable_capture(self.pg_interfaces)
921 rx = self.pg1.get_capture(1)
924 self.assertEqual(rx[Ether].dst, self.pg0._remote_hosts[2].mac)
925 self.assertEqual(rx[Ether].src, self.pg1.local_mac)
927 self.assertEqual(rx[IPv6].src, t2[IPv6].src)
928 self.assertEqual(rx[IPv6].dst, t2[IPv6].dst)
931 # remove the proxy configs
933 self.vapi.ip6_nd_proxy(
934 inet_pton(AF_INET6, self.pg0._remote_hosts[2].ip6),
935 self.pg1.sw_if_index,
937 self.vapi.ip6_nd_proxy(
938 inet_pton(AF_INET6, self.pg0._remote_hosts[3].ip6),
939 self.pg2.sw_if_index,
942 self.assertFalse(find_nbr(self,
943 self.pg2.sw_if_index,
944 self.pg0._remote_hosts[3].ip6,
946 self.assertFalse(find_nbr(self,
947 self.pg1.sw_if_index,
948 self.pg0._remote_hosts[2].ip6,
952 # no longer proxy-ing...
954 self.send_and_assert_no_replies(self.pg0, ns_pg0, "Proxy unconfigured")
955 self.send_and_assert_no_replies(self.pg1, ns_pg1, "Proxy unconfigured")
956 self.send_and_assert_no_replies(self.pg2, ns_pg2, "Proxy unconfigured")
959 # no longer forwarding. traffic generates NS out of the glean/main
962 self.pg2.add_stream(t2)
963 self.pg_enable_capture(self.pg_interfaces)
966 rx = self.pg0.get_capture(1)
968 self.assertTrue(rx[0].haslayer(ICMPv6ND_NS))
971 class TestIPNull(VppTestCase):
972 """ IPv6 routes via NULL """
975 super(TestIPNull, self).setUp()
977 # create 2 pg interfaces
978 self.create_pg_interfaces(range(1))
980 for i in self.pg_interfaces:
986 super(TestIPNull, self).tearDown()
987 for i in self.pg_interfaces:
991 def test_ip_null(self):
992 """ IP NULL route """
994 p = (Ether(src=self.pg0.remote_mac,
995 dst=self.pg0.local_mac) /
996 IPv6(src=self.pg0.remote_ip6, dst="2001::1") /
997 UDP(sport=1234, dport=1234) /
1001 # A route via IP NULL that will reply with ICMP unreachables
1003 ip_unreach = VppIpRoute(self, "2001::", 64, [], is_unreach=1, is_ip6=1)
1004 ip_unreach.add_vpp_config()
1006 self.pg0.add_stream(p)
1007 self.pg_enable_capture(self.pg_interfaces)
1010 rx = self.pg0.get_capture(1)
1012 icmp = rx[ICMPv6DestUnreach]
1014 # 0 = "No route to destination"
1015 self.assertEqual(icmp.code, 0)
1017 # ICMP is rate limited. pause a bit
1021 # A route via IP NULL that will reply with ICMP prohibited
1023 ip_prohibit = VppIpRoute(self, "2001::1", 128, [],
1024 is_prohibit=1, is_ip6=1)
1025 ip_prohibit.add_vpp_config()
1027 self.pg0.add_stream(p)
1028 self.pg_enable_capture(self.pg_interfaces)
1031 rx = self.pg0.get_capture(1)
1033 icmp = rx[ICMPv6DestUnreach]
1035 # 1 = "Communication with destination administratively prohibited"
1036 self.assertEqual(icmp.code, 1)
1039 class TestIPDisabled(VppTestCase):
1040 """ IPv6 disabled """
1043 super(TestIPDisabled, self).setUp()
1045 # create 2 pg interfaces
1046 self.create_pg_interfaces(range(2))
1050 self.pg0.config_ip6()
1051 self.pg0.resolve_ndp()
1053 # PG 1 is not IP enabled
1057 super(TestIPDisabled, self).tearDown()
1058 for i in self.pg_interfaces:
1062 def send_and_assert_no_replies(self, intf, pkts, remark):
1063 intf.add_stream(pkts)
1064 self.pg_enable_capture(self.pg_interfaces)
1066 for i in self.pg_interfaces:
1068 i.assert_nothing_captured(remark=remark)
1070 def test_ip_disabled(self):
1075 # one accepting interface, pg0, 2 forwarding interfaces
1077 route_ff_01 = VppIpMRoute(
1081 MRouteEntryFlags.MFIB_ENTRY_FLAG_NONE,
1082 [VppMRoutePath(self.pg1.sw_if_index,
1083 MRouteItfFlags.MFIB_ITF_FLAG_ACCEPT),
1084 VppMRoutePath(self.pg0.sw_if_index,
1085 MRouteItfFlags.MFIB_ITF_FLAG_FORWARD)],
1087 route_ff_01.add_vpp_config()
1089 pu = (Ether(src=self.pg1.remote_mac,
1090 dst=self.pg1.local_mac) /
1091 IPv6(src="2001::1", dst=self.pg0.remote_ip6) /
1092 UDP(sport=1234, dport=1234) /
1094 pm = (Ether(src=self.pg1.remote_mac,
1095 dst=self.pg1.local_mac) /
1096 IPv6(src="2001::1", dst="ffef::1") /
1097 UDP(sport=1234, dport=1234) /
1101 # PG1 does not forward IP traffic
1103 self.send_and_assert_no_replies(self.pg1, pu, "IPv6 disabled")
1104 self.send_and_assert_no_replies(self.pg1, pm, "IPv6 disabled")
1109 self.pg1.config_ip6()
1112 # Now we get packets through
1114 self.pg1.add_stream(pu)
1115 self.pg_enable_capture(self.pg_interfaces)
1117 rx = self.pg0.get_capture(1)
1119 self.pg1.add_stream(pm)
1120 self.pg_enable_capture(self.pg_interfaces)
1122 rx = self.pg0.get_capture(1)
1127 self.pg1.unconfig_ip6()
1130 # PG1 does not forward IP traffic
1132 self.send_and_assert_no_replies(self.pg1, pu, "IPv6 disabled")
1133 self.send_and_assert_no_replies(self.pg1, pm, "IPv6 disabled")
1136 class TestIP6LoadBalance(VppTestCase):
1137 """ IPv6 Load-Balancing """
1140 super(TestIP6LoadBalance, self).setUp()
1142 self.create_pg_interfaces(range(5))
1144 for i in self.pg_interfaces:
1150 super(TestIP6LoadBalance, self).tearDown()
1151 for i in self.pg_interfaces:
1155 def send_and_expect_load_balancing(self, input, pkts, outputs):
1156 input.add_stream(pkts)
1157 self.pg_enable_capture(self.pg_interfaces)
1160 rx = oo._get_capture(1)
1161 self.assertNotEqual(0, len(rx))
1163 def test_ip6_load_balance(self):
1164 """ IPv6 Load-Balancing """
1167 # An array of packets that differ only in the destination port
1172 # An array of packets that differ only in the source address
1176 for ii in range(65):
1177 port_pkts.append((Ether(src=self.pg0.remote_mac,
1178 dst=self.pg0.local_mac) /
1179 IPv6(dst="3000::1", src="3000:1::1") /
1180 UDP(sport=1234, dport=1234 + ii) /
1182 src_pkts.append((Ether(src=self.pg0.remote_mac,
1183 dst=self.pg0.local_mac) /
1184 IPv6(dst="3000::1", src="3000:1::%d" % ii) /
1185 UDP(sport=1234, dport=1234) /
1188 route_3000_1 = VppIpRoute(self, "3000::1", 128,
1189 [VppRoutePath(self.pg1.remote_ip6,
1190 self.pg1.sw_if_index,
1192 VppRoutePath(self.pg2.remote_ip6,
1193 self.pg2.sw_if_index,
1196 route_3000_1.add_vpp_config()
1199 # inject the packet on pg0 - expect load-balancing across the 2 paths
1200 # - since the default hash config is to use IP src,dst and port
1202 # We are not going to ensure equal amounts of packets across each link,
1203 # since the hash algorithm is statistical and therefore this can never
1204 # be guaranteed. But wuth 64 different packets we do expect some
1205 # balancing. So instead just ensure there is traffic on each link.
1207 self.send_and_expect_load_balancing(self.pg0, port_pkts,
1208 [self.pg1, self.pg2])
1209 self.send_and_expect_load_balancing(self.pg0, src_pkts,
1210 [self.pg1, self.pg2])
1213 # change the flow hash config so it's only IP src,dst
1214 # - now only the stream with differing source address will
1217 self.vapi.set_ip_flow_hash(0, is_ip6=1, src=1, dst=1, sport=0, dport=0)
1219 self.send_and_expect_load_balancing(self.pg0, src_pkts,
1220 [self.pg1, self.pg2])
1222 self.pg0.add_stream(port_pkts)
1223 self.pg_enable_capture(self.pg_interfaces)
1226 rx = self.pg2.get_capture(len(port_pkts))
1229 # change the flow hash config back to defaults
1231 self.vapi.set_ip_flow_hash(0, is_ip6=1, src=1, dst=1, sport=1, dport=1)
1234 # Recursive prefixes
1235 # - testing that 2 stages of load-balancing occurs and there is no
1236 # polarisation (i.e. only 2 of 4 paths are used)
1241 for ii in range(257):
1242 port_pkts.append((Ether(src=self.pg0.remote_mac,
1243 dst=self.pg0.local_mac) /
1244 IPv6(dst="4000::1", src="4000:1::1") /
1245 UDP(sport=1234, dport=1234 + ii) /
1247 src_pkts.append((Ether(src=self.pg0.remote_mac,
1248 dst=self.pg0.local_mac) /
1249 IPv6(dst="4000::1", src="4000:1::%d" % ii) /
1250 UDP(sport=1234, dport=1234) /
1253 route_3000_2 = VppIpRoute(self, "3000::2", 128,
1254 [VppRoutePath(self.pg3.remote_ip6,
1255 self.pg3.sw_if_index,
1257 VppRoutePath(self.pg4.remote_ip6,
1258 self.pg4.sw_if_index,
1261 route_3000_2.add_vpp_config()
1263 route_4000_1 = VppIpRoute(self, "4000::1", 128,
1264 [VppRoutePath("3000::1",
1267 VppRoutePath("3000::2",
1271 route_4000_1.add_vpp_config()
1274 # inject the packet on pg0 - expect load-balancing across all 4 paths
1276 self.vapi.cli("clear trace")
1277 self.send_and_expect_load_balancing(self.pg0, port_pkts,
1278 [self.pg1, self.pg2,
1279 self.pg3, self.pg4])
1280 self.send_and_expect_load_balancing(self.pg0, src_pkts,
1281 [self.pg1, self.pg2,
1282 self.pg3, self.pg4])
1285 if __name__ == '__main__':
1286 unittest.main(testRunner=VppTestRunner)