6 from parameterized import parameterized
8 import scapy.layers.inet6 as inet6
9 from scapy.contrib.mpls import MPLS
10 from scapy.layers.inet6 import IPv6, ICMPv6ND_NS, ICMPv6ND_RS, \
11 ICMPv6ND_RA, ICMPv6NDOptMTU, ICMPv6NDOptSrcLLAddr, ICMPv6NDOptPrefixInfo, \
12 ICMPv6ND_NA, ICMPv6NDOptDstLLAddr, ICMPv6DestUnreach, icmp6types, \
13 ICMPv6TimeExceeded, ICMPv6EchoRequest, ICMPv6EchoReply, IPv6ExtHdrHopByHop
14 from scapy.layers.l2 import Ether, Dot1Q
15 from scapy.packet import Raw
16 from scapy.utils import inet_pton, inet_ntop
17 from scapy.utils6 import in6_getnsma, in6_getnsmac, in6_ptop, in6_islladdr, \
21 from framework import VppTestCase, VppTestRunner
22 from util import ppp, ip6_normalize, mk_ll_addr
23 from vpp_ip import DpoProto
24 from vpp_ip_route import VppIpRoute, VppRoutePath, find_route, VppIpMRoute, \
25 VppMRoutePath, MRouteItfFlags, MRouteEntryFlags, VppMplsIpBind, \
26 VppMplsRoute, VppMplsTable, VppIpTable, FibPathType, FibPathProto, \
27 VppIpInterfaceAddress, find_route_in_dump, find_mroute_in_dump
28 from vpp_neighbor import find_nbr, VppNeighbor
29 from vpp_pg_interface import is_ipv6_misc
30 from vpp_sub_interface import VppSubInterface, VppDot1QSubint
31 from ipaddress import IPv6Network, IPv6Address
33 AF_INET6 = socket.AF_INET6
43 class TestIPv6ND(VppTestCase):
44 def validate_ra(self, intf, rx, dst_ip=None):
46 dst_ip = intf.remote_ip6
48 # unicasted packets must come to the unicast mac
49 self.assertEqual(rx[Ether].dst, intf.remote_mac)
51 # and from the router's MAC
52 self.assertEqual(rx[Ether].src, intf.local_mac)
54 # the rx'd RA should be addressed to the sender's source
55 self.assertTrue(rx.haslayer(ICMPv6ND_RA))
56 self.assertEqual(in6_ptop(rx[IPv6].dst),
59 # and come from the router's link local
60 self.assertTrue(in6_islladdr(rx[IPv6].src))
61 self.assertEqual(in6_ptop(rx[IPv6].src),
62 in6_ptop(mk_ll_addr(intf.local_mac)))
64 def validate_na(self, intf, rx, dst_ip=None, tgt_ip=None):
66 dst_ip = intf.remote_ip6
68 dst_ip = intf.local_ip6
70 # unicasted packets must come to the unicast mac
71 self.assertEqual(rx[Ether].dst, intf.remote_mac)
73 # and from the router's MAC
74 self.assertEqual(rx[Ether].src, intf.local_mac)
76 # the rx'd NA should be addressed to the sender's source
77 self.assertTrue(rx.haslayer(ICMPv6ND_NA))
78 self.assertEqual(in6_ptop(rx[IPv6].dst),
81 # and come from the target address
83 in6_ptop(rx[IPv6].src), in6_ptop(tgt_ip))
85 # Dest link-layer options should have the router's MAC
86 dll = rx[ICMPv6NDOptDstLLAddr]
87 self.assertEqual(dll.lladdr, intf.local_mac)
89 def validate_ns(self, intf, rx, tgt_ip):
90 nsma = in6_getnsma(inet_pton(AF_INET6, tgt_ip))
91 dst_ip = inet_ntop(AF_INET6, nsma)
94 self.assertEqual(rx[Ether].dst, in6_getnsmac(nsma))
96 # and from the router's MAC
97 self.assertEqual(rx[Ether].src, intf.local_mac)
99 # the rx'd NS should be addressed to an mcast address
100 # derived from the target address
102 in6_ptop(rx[IPv6].dst), in6_ptop(dst_ip))
104 # expect the tgt IP in the NS header
106 self.assertEqual(in6_ptop(ns.tgt), in6_ptop(tgt_ip))
108 # packet is from the router's local address
110 in6_ptop(rx[IPv6].src), intf.local_ip6)
112 # Src link-layer options should have the router's MAC
113 sll = rx[ICMPv6NDOptSrcLLAddr]
114 self.assertEqual(sll.lladdr, intf.local_mac)
116 def send_and_expect_ra(self, intf, pkts, remark, dst_ip=None,
117 filter_out_fn=is_ipv6_misc):
118 intf.add_stream(pkts)
119 self.pg_enable_capture(self.pg_interfaces)
121 rx = intf.get_capture(1, filter_out_fn=filter_out_fn)
123 self.assertEqual(len(rx), 1)
125 self.validate_ra(intf, rx, dst_ip)
127 def send_and_expect_na(self, intf, pkts, remark, dst_ip=None,
129 filter_out_fn=is_ipv6_misc):
130 intf.add_stream(pkts)
131 self.pg_enable_capture(self.pg_interfaces)
133 rx = intf.get_capture(1, filter_out_fn=filter_out_fn)
135 self.assertEqual(len(rx), 1)
137 self.validate_na(intf, rx, dst_ip, tgt_ip)
139 def send_and_expect_ns(self, tx_intf, rx_intf, pkts, tgt_ip,
140 filter_out_fn=is_ipv6_misc):
141 tx_intf.add_stream(pkts)
142 self.pg_enable_capture(self.pg_interfaces)
144 rx = rx_intf.get_capture(1, filter_out_fn=filter_out_fn)
146 self.assertEqual(len(rx), 1)
148 self.validate_ns(rx_intf, rx, tgt_ip)
150 def verify_ip(self, rx, smac, dmac, sip, dip):
152 self.assertEqual(ether.dst, dmac)
153 self.assertEqual(ether.src, smac)
156 self.assertEqual(ip.src, sip)
157 self.assertEqual(ip.dst, dip)
160 class TestIPv6(TestIPv6ND):
161 """ IPv6 Test Case """
165 super(TestIPv6, cls).setUpClass()
168 def tearDownClass(cls):
169 super(TestIPv6, cls).tearDownClass()
173 Perform test setup before test case.
176 - create 3 pg interfaces
177 - untagged pg0 interface
178 - Dot1Q subinterface on pg1
179 - Dot1AD subinterface on pg2
181 - put it into UP state
183 - resolve neighbor address using NDP
184 - configure 200 fib entries
186 :ivar list interfaces: pg interfaces and subinterfaces.
187 :ivar dict flows: IPv4 packet flows in test.
189 *TODO:* Create AD sub interface
191 super(TestIPv6, self).setUp()
193 # create 3 pg interfaces
194 self.create_pg_interfaces(range(3))
196 # create 2 subinterfaces for p1 and pg2
197 self.sub_interfaces = [
198 VppDot1QSubint(self, self.pg1, 100),
199 VppDot1QSubint(self, self.pg2, 200)
200 # TODO: VppDot1ADSubint(self, self.pg2, 200, 300, 400)
203 # packet flows mapping pg0 -> pg1.sub, pg2.sub, etc.
205 self.flows[self.pg0] = [self.pg1.sub_if, self.pg2.sub_if]
206 self.flows[self.pg1.sub_if] = [self.pg0, self.pg2.sub_if]
207 self.flows[self.pg2.sub_if] = [self.pg0, self.pg1.sub_if]
210 self.pg_if_packet_sizes = [64, 1500, 9020]
212 self.interfaces = list(self.pg_interfaces)
213 self.interfaces.extend(self.sub_interfaces)
215 # setup all interfaces
216 for i in self.interfaces:
222 """Run standard test teardown and log ``show ip6 neighbors``."""
223 for i in self.interfaces:
227 for i in self.sub_interfaces:
228 i.remove_vpp_config()
230 super(TestIPv6, self).tearDown()
231 if not self.vpp_dead:
232 self.logger.info(self.vapi.cli("show ip6 neighbors"))
233 # info(self.vapi.cli("show ip6 fib")) # many entries
235 def modify_packet(self, src_if, packet_size, pkt):
236 """Add load, set destination IP and extend packet to required packet
237 size for defined interface.
239 :param VppInterface src_if: Interface to create packet for.
240 :param int packet_size: Required packet size.
241 :param Scapy pkt: Packet to be modified.
243 dst_if_idx = int(packet_size / 10 % 2)
244 dst_if = self.flows[src_if][dst_if_idx]
245 info = self.create_packet_info(src_if, dst_if)
246 payload = self.info_to_payload(info)
247 p = pkt / Raw(payload)
248 p[IPv6].dst = dst_if.remote_ip6
250 if isinstance(src_if, VppSubInterface):
251 p = src_if.add_dot1_layer(p)
252 self.extend_packet(p, packet_size)
256 def create_stream(self, src_if):
257 """Create input packet stream for defined interface.
259 :param VppInterface src_if: Interface to create packet stream for.
261 hdr_ext = 4 if isinstance(src_if, VppSubInterface) else 0
262 pkt_tmpl = (Ether(dst=src_if.local_mac, src=src_if.remote_mac) /
263 IPv6(src=src_if.remote_ip6) /
264 inet6.UDP(sport=1234, dport=1234))
266 pkts = [self.modify_packet(src_if, i, pkt_tmpl)
267 for i in moves.range(self.pg_if_packet_sizes[0],
268 self.pg_if_packet_sizes[1], 10)]
269 pkts_b = [self.modify_packet(src_if, i, pkt_tmpl)
270 for i in moves.range(self.pg_if_packet_sizes[1] + hdr_ext,
271 self.pg_if_packet_sizes[2] + hdr_ext,
277 def verify_capture(self, dst_if, capture):
278 """Verify captured input packet stream for defined interface.
280 :param VppInterface dst_if: Interface to verify captured packet stream
282 :param list capture: Captured packet stream.
284 self.logger.info("Verifying capture on interface %s" % dst_if.name)
286 for i in self.interfaces:
287 last_info[i.sw_if_index] = None
289 dst_sw_if_index = dst_if.sw_if_index
290 if hasattr(dst_if, 'parent'):
292 for packet in capture:
294 # Check VLAN tags and Ethernet header
295 packet = dst_if.remove_dot1_layer(packet)
296 self.assertTrue(Dot1Q not in packet)
299 udp = packet[inet6.UDP]
300 payload_info = self.payload_to_info(packet[Raw])
301 packet_index = payload_info.index
302 self.assertEqual(payload_info.dst, dst_sw_if_index)
304 "Got packet on port %s: src=%u (id=%u)" %
305 (dst_if.name, payload_info.src, packet_index))
306 next_info = self.get_next_packet_info_for_interface2(
307 payload_info.src, dst_sw_if_index,
308 last_info[payload_info.src])
309 last_info[payload_info.src] = next_info
310 self.assertTrue(next_info is not None)
311 self.assertEqual(packet_index, next_info.index)
312 saved_packet = next_info.data
313 # Check standard fields
315 ip.src, saved_packet[IPv6].src)
317 ip.dst, saved_packet[IPv6].dst)
319 udp.sport, saved_packet[inet6.UDP].sport)
321 udp.dport, saved_packet[inet6.UDP].dport)
323 self.logger.error(ppp("Unexpected or invalid packet:", packet))
325 for i in self.interfaces:
326 remaining_packet = self.get_next_packet_info_for_interface2(
327 i.sw_if_index, dst_sw_if_index, last_info[i.sw_if_index])
328 self.assertTrue(remaining_packet is None,
329 "Interface %s: Packet expected from interface %s "
330 "didn't arrive" % (dst_if.name, i.name))
332 def test_next_header_anomaly(self):
333 """ IPv6 next header anomaly test
336 - ipv6 next header field = Fragment Header (44)
337 - next header is ICMPv6 Echo Request
338 - wait for reassembly
340 pkt = (Ether(src=self.pg0.local_mac, dst=self.pg0.remote_mac) /
341 IPv6(src=self.pg0.remote_ip6, dst=self.pg0.local_ip6, nh=44) /
344 self.pg0.add_stream(pkt)
347 # wait for reassembly
354 - Create IPv6 stream for pg0 interface
355 - Create IPv6 tagged streams for pg1's and pg2's subinterface.
356 - Send and verify received packets on each interface.
359 pkts = self.create_stream(self.pg0)
360 self.pg0.add_stream(pkts)
362 for i in self.sub_interfaces:
363 pkts = self.create_stream(i)
364 i.parent.add_stream(pkts)
366 self.pg_enable_capture(self.pg_interfaces)
369 pkts = self.pg0.get_capture()
370 self.verify_capture(self.pg0, pkts)
372 for i in self.sub_interfaces:
373 pkts = i.parent.get_capture()
374 self.verify_capture(i, pkts)
377 """ IPv6 Neighbour Solicitation Exceptions
380 - Send an NS Sourced from an address not covered by the link sub-net
381 - Send an NS to an mcast address the router has not joined
382 - Send NS for a target address the router does not onn.
386 # An NS from a non link source address
388 nsma = in6_getnsma(inet_pton(AF_INET6, self.pg0.local_ip6))
389 d = inet_ntop(AF_INET6, nsma)
391 p = (Ether(dst=in6_getnsmac(nsma)) /
392 IPv6(dst=d, src="2002::2") /
393 ICMPv6ND_NS(tgt=self.pg0.local_ip6) /
394 ICMPv6NDOptSrcLLAddr(
395 lladdr=self.pg0.remote_mac))
398 self.send_and_assert_no_replies(
400 "No response to NS source by address not on sub-net")
403 # An NS for sent to a solicited mcast group the router is
404 # not a member of FAILS
407 nsma = in6_getnsma(inet_pton(AF_INET6, "fd::ffff"))
408 d = inet_ntop(AF_INET6, nsma)
410 p = (Ether(dst=in6_getnsmac(nsma)) /
411 IPv6(dst=d, src=self.pg0.remote_ip6) /
412 ICMPv6ND_NS(tgt=self.pg0.local_ip6) /
413 ICMPv6NDOptSrcLLAddr(
414 lladdr=self.pg0.remote_mac))
417 self.send_and_assert_no_replies(
419 "No response to NS sent to unjoined mcast address")
422 # An NS whose target address is one the router does not own
424 nsma = in6_getnsma(inet_pton(AF_INET6, self.pg0.local_ip6))
425 d = inet_ntop(AF_INET6, nsma)
427 p = (Ether(dst=in6_getnsmac(nsma)) /
428 IPv6(dst=d, src=self.pg0.remote_ip6) /
429 ICMPv6ND_NS(tgt="fd::ffff") /
430 ICMPv6NDOptSrcLLAddr(
431 lladdr=self.pg0.remote_mac))
434 self.send_and_assert_no_replies(self.pg0, pkts,
435 "No response to NS for unknown target")
438 # A neighbor entry that has no associated FIB-entry
440 self.pg0.generate_remote_hosts(4)
441 nd_entry = VppNeighbor(self,
442 self.pg0.sw_if_index,
443 self.pg0.remote_hosts[2].mac,
444 self.pg0.remote_hosts[2].ip6,
446 nd_entry.add_vpp_config()
449 # check we have the neighbor, but no route
451 self.assertTrue(find_nbr(self,
452 self.pg0.sw_if_index,
453 self.pg0._remote_hosts[2].ip6))
454 self.assertFalse(find_route(self,
455 self.pg0._remote_hosts[2].ip6,
459 # send an NS from a link local address to the interface's global
462 p = (Ether(dst=in6_getnsmac(nsma), src=self.pg0.remote_mac) /
464 dst=d, src=self.pg0._remote_hosts[2].ip6_ll) /
465 ICMPv6ND_NS(tgt=self.pg0.local_ip6) /
466 ICMPv6NDOptSrcLLAddr(
467 lladdr=self.pg0.remote_mac))
469 self.send_and_expect_na(self.pg0, p,
470 "NS from link-local",
471 dst_ip=self.pg0._remote_hosts[2].ip6_ll,
472 tgt_ip=self.pg0.local_ip6)
475 # we should have learned an ND entry for the peer's link-local
476 # but not inserted a route to it in the FIB
478 self.assertTrue(find_nbr(self,
479 self.pg0.sw_if_index,
480 self.pg0._remote_hosts[2].ip6_ll))
481 self.assertFalse(find_route(self,
482 self.pg0._remote_hosts[2].ip6_ll,
486 # An NS to the router's own Link-local
488 p = (Ether(dst=in6_getnsmac(nsma), src=self.pg0.remote_mac) /
490 dst=d, src=self.pg0._remote_hosts[3].ip6_ll) /
491 ICMPv6ND_NS(tgt=self.pg0.local_ip6_ll) /
492 ICMPv6NDOptSrcLLAddr(
493 lladdr=self.pg0.remote_mac))
495 self.send_and_expect_na(self.pg0, p,
496 "NS to/from link-local",
497 dst_ip=self.pg0._remote_hosts[3].ip6_ll,
498 tgt_ip=self.pg0.local_ip6_ll)
501 # we should have learned an ND entry for the peer's link-local
502 # but not inserted a route to it in the FIB
504 self.assertTrue(find_nbr(self,
505 self.pg0.sw_if_index,
506 self.pg0._remote_hosts[3].ip6_ll))
507 self.assertFalse(find_route(self,
508 self.pg0._remote_hosts[3].ip6_ll,
511 def test_ns_duplicates(self):
515 # Generate some hosts on the LAN
517 self.pg1.generate_remote_hosts(3)
520 # Add host 1 on pg1 and pg2
522 ns_pg1 = VppNeighbor(self,
523 self.pg1.sw_if_index,
524 self.pg1.remote_hosts[1].mac,
525 self.pg1.remote_hosts[1].ip6)
526 ns_pg1.add_vpp_config()
527 ns_pg2 = VppNeighbor(self,
528 self.pg2.sw_if_index,
530 self.pg1.remote_hosts[1].ip6)
531 ns_pg2.add_vpp_config()
534 # IP packet destined for pg1 remote host arrives on pg1 again.
536 p = (Ether(dst=self.pg0.local_mac,
537 src=self.pg0.remote_mac) /
538 IPv6(src=self.pg0.remote_ip6,
539 dst=self.pg1.remote_hosts[1].ip6) /
540 inet6.UDP(sport=1234, dport=1234) /
543 self.pg0.add_stream(p)
544 self.pg_enable_capture(self.pg_interfaces)
547 rx1 = self.pg1.get_capture(1)
549 self.verify_ip(rx1[0],
551 self.pg1.remote_hosts[1].mac,
553 self.pg1.remote_hosts[1].ip6)
556 # remove the duplicate on pg1
557 # packet stream should generate NSs out of pg1
559 ns_pg1.remove_vpp_config()
561 self.send_and_expect_ns(self.pg0, self.pg1,
562 p, self.pg1.remote_hosts[1].ip6)
567 ns_pg1.add_vpp_config()
569 self.pg0.add_stream(p)
570 self.pg_enable_capture(self.pg_interfaces)
573 rx1 = self.pg1.get_capture(1)
575 self.verify_ip(rx1[0],
577 self.pg1.remote_hosts[1].mac,
579 self.pg1.remote_hosts[1].ip6)
581 def validate_ra(self, intf, rx, dst_ip=None, mtu=9000, pi_opt=None):
583 dst_ip = intf.remote_ip6
585 # unicasted packets must come to the unicast mac
586 self.assertEqual(rx[Ether].dst, intf.remote_mac)
588 # and from the router's MAC
589 self.assertEqual(rx[Ether].src, intf.local_mac)
591 # the rx'd RA should be addressed to the sender's source
592 self.assertTrue(rx.haslayer(ICMPv6ND_RA))
593 self.assertEqual(in6_ptop(rx[IPv6].dst),
596 # and come from the router's link local
597 self.assertTrue(in6_islladdr(rx[IPv6].src))
598 self.assertEqual(in6_ptop(rx[IPv6].src),
599 in6_ptop(mk_ll_addr(intf.local_mac)))
601 # it should contain the links MTU
603 self.assertEqual(ra[ICMPv6NDOptMTU].mtu, mtu)
605 # it should contain the source's link layer address option
606 sll = ra[ICMPv6NDOptSrcLLAddr]
607 self.assertEqual(sll.lladdr, intf.local_mac)
610 # the RA should not contain prefix information
611 self.assertFalse(ra.haslayer(
612 ICMPv6NDOptPrefixInfo))
614 raos = rx.getlayer(ICMPv6NDOptPrefixInfo, 1)
616 # the options are nested in the scapy packet in way that i cannot
617 # decipher how to decode. this 1st layer of option always returns
618 # nested classes, so a direct obj1=obj2 comparison always fails.
619 # however, the getlayer(.., 2) does give one instance.
620 # so we cheat here and construct a new opt instance for comparison
621 rd = ICMPv6NDOptPrefixInfo(
622 prefixlen=raos.prefixlen,
626 if type(pi_opt) is list:
627 for ii in range(len(pi_opt)):
628 self.assertEqual(pi_opt[ii], rd)
630 ICMPv6NDOptPrefixInfo, ii + 2)
632 self.assertEqual(pi_opt, raos, 'Expected: %s, received: %s'
633 % (pi_opt.show(dump=True),
634 raos.show(dump=True)))
636 def send_and_expect_ra(self, intf, pkts, remark, dst_ip=None,
637 filter_out_fn=is_ipv6_misc,
639 intf.add_stream(pkts)
640 self.pg_enable_capture(self.pg_interfaces)
642 rx = intf.get_capture(1, filter_out_fn=filter_out_fn)
644 self.assertEqual(len(rx), 1)
646 self.validate_ra(intf, rx, dst_ip, pi_opt=opt)
649 """ IPv6 Router Solicitation Exceptions
655 # Before we begin change the IPv6 RA responses to use the unicast
656 # address - that way we will not confuse them with the periodic
657 # RAs which go to the mcast address
658 # Sit and wait for the first periodic RA.
662 self.pg0.ip6_ra_config(send_unicast=1)
665 # An RS from a link source address
666 # - expect an RA in return
668 p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
670 dst=self.pg0.local_ip6, src=self.pg0.remote_ip6) /
673 self.send_and_expect_ra(self.pg0, pkts, "Genuine RS")
676 # For the next RS sent the RA should be rate limited
678 self.send_and_assert_no_replies(self.pg0, pkts, "RA rate limited")
681 # When we reconfigure the IPv6 RA config,
682 # we reset the RA rate limiting,
683 # so we need to do this before each test below so as not to drop
684 # packets for rate limiting reasons. Test this works here.
686 self.pg0.ip6_ra_config(send_unicast=1)
687 self.send_and_expect_ra(self.pg0, pkts, "Rate limit reset RS")
690 # An RS sent from a non-link local source
692 self.pg0.ip6_ra_config(send_unicast=1)
693 p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
694 IPv6(dst=self.pg0.local_ip6,
698 self.send_and_assert_no_replies(self.pg0, pkts,
699 "RS from non-link source")
702 # Source an RS from a link local address
704 self.pg0.ip6_ra_config(send_unicast=1)
705 ll = mk_ll_addr(self.pg0.remote_mac)
706 p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
707 IPv6(dst=self.pg0.local_ip6, src=ll) /
710 self.send_and_expect_ra(self.pg0, pkts,
711 "RS sourced from link-local",
715 # Send the RS multicast
717 self.pg0.ip6_ra_config(send_unicast=1)
718 dmac = in6_getnsmac(inet_pton(AF_INET6, "ff02::2"))
719 ll = mk_ll_addr(self.pg0.remote_mac)
720 p = (Ether(dst=dmac, src=self.pg0.remote_mac) /
721 IPv6(dst="ff02::2", src=ll) /
724 self.send_and_expect_ra(self.pg0, pkts,
725 "RS sourced from link-local",
729 # Source from the unspecified address ::. This happens when the RS
730 # is sent before the host has a configured address/sub-net,
731 # i.e. auto-config. Since the sender has no IP address, the reply
732 # comes back mcast - so the capture needs to not filter this.
733 # If we happen to pick up the periodic RA at this point then so be it,
736 self.pg0.ip6_ra_config(send_unicast=1, suppress=1)
737 p = (Ether(dst=dmac, src=self.pg0.remote_mac) /
738 IPv6(dst="ff02::2", src="::") /
741 self.send_and_expect_ra(self.pg0, pkts,
742 "RS sourced from unspecified",
747 # Configure The RA to announce the links prefix
749 self.pg0.ip6_ra_prefix('%s/%s' % (self.pg0.local_ip6,
750 self.pg0.local_ip6_prefix_len))
753 # RAs should now contain the prefix information option
755 opt = ICMPv6NDOptPrefixInfo(
756 prefixlen=self.pg0.local_ip6_prefix_len,
757 prefix=self.pg0.local_ip6,
761 self.pg0.ip6_ra_config(send_unicast=1)
762 ll = mk_ll_addr(self.pg0.remote_mac)
763 p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
764 IPv6(dst=self.pg0.local_ip6, src=ll) /
766 self.send_and_expect_ra(self.pg0, p,
767 "RA with prefix-info",
772 # Change the prefix info to not off-link
775 self.pg0.ip6_ra_prefix('%s/%s' % (self.pg0.local_ip6,
776 self.pg0.local_ip6_prefix_len),
779 opt = ICMPv6NDOptPrefixInfo(
780 prefixlen=self.pg0.local_ip6_prefix_len,
781 prefix=self.pg0.local_ip6,
785 self.pg0.ip6_ra_config(send_unicast=1)
786 self.send_and_expect_ra(self.pg0, p,
787 "RA with Prefix info with L-flag=0",
792 # Change the prefix info to not off-link, no-autoconfig
793 # L and A flag are clear in the advert
795 self.pg0.ip6_ra_prefix('%s/%s' % (self.pg0.local_ip6,
796 self.pg0.local_ip6_prefix_len),
800 opt = ICMPv6NDOptPrefixInfo(
801 prefixlen=self.pg0.local_ip6_prefix_len,
802 prefix=self.pg0.local_ip6,
806 self.pg0.ip6_ra_config(send_unicast=1)
807 self.send_and_expect_ra(self.pg0, p,
808 "RA with Prefix info with A & L-flag=0",
813 # Change the flag settings back to the defaults
814 # L and A flag are set in the advert
816 self.pg0.ip6_ra_prefix('%s/%s' % (self.pg0.local_ip6,
817 self.pg0.local_ip6_prefix_len))
819 opt = ICMPv6NDOptPrefixInfo(
820 prefixlen=self.pg0.local_ip6_prefix_len,
821 prefix=self.pg0.local_ip6,
825 self.pg0.ip6_ra_config(send_unicast=1)
826 self.send_and_expect_ra(self.pg0, p,
827 "RA with Prefix info",
832 # Change the prefix info to not off-link, no-autoconfig
833 # L and A flag are clear in the advert
835 self.pg0.ip6_ra_prefix('%s/%s' % (self.pg0.local_ip6,
836 self.pg0.local_ip6_prefix_len),
840 opt = ICMPv6NDOptPrefixInfo(
841 prefixlen=self.pg0.local_ip6_prefix_len,
842 prefix=self.pg0.local_ip6,
846 self.pg0.ip6_ra_config(send_unicast=1)
847 self.send_and_expect_ra(self.pg0, p,
848 "RA with Prefix info with A & L-flag=0",
853 # Use the reset to defaults option to revert to defaults
854 # L and A flag are clear in the advert
856 self.pg0.ip6_ra_prefix('%s/%s' % (self.pg0.local_ip6,
857 self.pg0.local_ip6_prefix_len),
860 opt = ICMPv6NDOptPrefixInfo(
861 prefixlen=self.pg0.local_ip6_prefix_len,
862 prefix=self.pg0.local_ip6,
866 self.pg0.ip6_ra_config(send_unicast=1)
867 self.send_and_expect_ra(self.pg0, p,
868 "RA with Prefix reverted to defaults",
873 # Advertise Another prefix. With no L-flag/A-flag
875 self.pg0.ip6_ra_prefix('%s/%s' % (self.pg1.local_ip6,
876 self.pg1.local_ip6_prefix_len),
880 opt = [ICMPv6NDOptPrefixInfo(
881 prefixlen=self.pg0.local_ip6_prefix_len,
882 prefix=self.pg0.local_ip6,
885 ICMPv6NDOptPrefixInfo(
886 prefixlen=self.pg1.local_ip6_prefix_len,
887 prefix=self.pg1.local_ip6,
891 self.pg0.ip6_ra_config(send_unicast=1)
892 ll = mk_ll_addr(self.pg0.remote_mac)
893 p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
894 IPv6(dst=self.pg0.local_ip6, src=ll) /
896 self.send_and_expect_ra(self.pg0, p,
897 "RA with multiple Prefix infos",
902 # Remove the first prefix-info - expect the second is still in the
905 self.pg0.ip6_ra_prefix('%s/%s' % (self.pg0.local_ip6,
906 self.pg0.local_ip6_prefix_len),
909 opt = ICMPv6NDOptPrefixInfo(
910 prefixlen=self.pg1.local_ip6_prefix_len,
911 prefix=self.pg1.local_ip6,
915 self.pg0.ip6_ra_config(send_unicast=1)
916 self.send_and_expect_ra(self.pg0, p,
917 "RA with Prefix reverted to defaults",
922 # Remove the second prefix-info - expect no prefix-info in the adverts
924 self.pg0.ip6_ra_prefix('%s/%s' % (self.pg1.local_ip6,
925 self.pg1.local_ip6_prefix_len),
928 self.pg0.ip6_ra_config(send_unicast=1)
929 self.send_and_expect_ra(self.pg0, p,
930 "RA with Prefix reverted to defaults",
934 # Reset the periodic advertisements back to default values
936 self.pg0.ip6_ra_config(no=1, suppress=1, send_unicast=0)
939 class TestIPv6IfAddrRoute(VppTestCase):
940 """ IPv6 Interface Addr Route Test Case """
944 super(TestIPv6IfAddrRoute, cls).setUpClass()
947 def tearDownClass(cls):
948 super(TestIPv6IfAddrRoute, cls).tearDownClass()
951 super(TestIPv6IfAddrRoute, self).setUp()
953 # create 1 pg interface
954 self.create_pg_interfaces(range(1))
956 for i in self.pg_interfaces:
962 super(TestIPv6IfAddrRoute, self).tearDown()
963 for i in self.pg_interfaces:
967 def test_ipv6_ifaddrs_same_prefix(self):
968 """ IPv6 Interface Addresses Same Prefix test
972 - Verify no route in FIB for prefix 2001:10::/64
973 - Configure IPv4 address 2001:10::10/64 on an interface
974 - Verify route in FIB for prefix 2001:10::/64
975 - Configure IPv4 address 2001:10::20/64 on an interface
976 - Delete 2001:10::10/64 from interface
977 - Verify route in FIB for prefix 2001:10::/64
978 - Delete 2001:10::20/64 from interface
979 - Verify no route in FIB for prefix 2001:10::/64
982 addr1 = "2001:10::10"
983 addr2 = "2001:10::20"
985 if_addr1 = VppIpInterfaceAddress(self, self.pg0, addr1, 64)
986 if_addr2 = VppIpInterfaceAddress(self, self.pg0, addr2, 64)
987 self.assertFalse(if_addr1.query_vpp_config())
988 self.assertFalse(find_route(self, addr1, 128))
989 self.assertFalse(find_route(self, addr2, 128))
991 # configure first address, verify route present
992 if_addr1.add_vpp_config()
993 self.assertTrue(if_addr1.query_vpp_config())
994 self.assertTrue(find_route(self, addr1, 128))
995 self.assertFalse(find_route(self, addr2, 128))
997 # configure second address, delete first, verify route not removed
998 if_addr2.add_vpp_config()
999 if_addr1.remove_vpp_config()
1000 self.assertFalse(if_addr1.query_vpp_config())
1001 self.assertTrue(if_addr2.query_vpp_config())
1002 self.assertFalse(find_route(self, addr1, 128))
1003 self.assertTrue(find_route(self, addr2, 128))
1005 # delete second address, verify route removed
1006 if_addr2.remove_vpp_config()
1007 self.assertFalse(if_addr1.query_vpp_config())
1008 self.assertFalse(find_route(self, addr1, 128))
1009 self.assertFalse(find_route(self, addr2, 128))
1012 class TestICMPv6Echo(VppTestCase):
1013 """ ICMPv6 Echo Test Case """
1016 def setUpClass(cls):
1017 super(TestICMPv6Echo, cls).setUpClass()
1020 def tearDownClass(cls):
1021 super(TestICMPv6Echo, cls).tearDownClass()
1024 super(TestICMPv6Echo, self).setUp()
1026 # create 1 pg interface
1027 self.create_pg_interfaces(range(1))
1029 for i in self.pg_interfaces:
1035 super(TestICMPv6Echo, self).tearDown()
1036 for i in self.pg_interfaces:
1041 def test_icmpv6_echo(self):
1042 """ VPP replies to ICMPv6 Echo Request
1046 - Receive ICMPv6 Echo Request message on pg0 interface.
1047 - Check outgoing ICMPv6 Echo Reply message on pg0 interface.
1052 icmpv6_data = b'\x0a' * 18
1053 p_echo_request = (Ether(src=self.pg0.remote_mac,
1054 dst=self.pg0.local_mac) /
1055 IPv6(src=self.pg0.remote_ip6,
1056 dst=self.pg0.local_ip6) /
1062 self.pg0.add_stream(p_echo_request)
1063 self.pg_enable_capture(self.pg_interfaces)
1066 rx = self.pg0.get_capture(1)
1070 icmpv6 = rx[ICMPv6EchoReply]
1072 self.assertEqual(ether.src, self.pg0.local_mac)
1073 self.assertEqual(ether.dst, self.pg0.remote_mac)
1075 self.assertEqual(ipv6.src, self.pg0.local_ip6)
1076 self.assertEqual(ipv6.dst, self.pg0.remote_ip6)
1079 icmp6types[icmpv6.type], "Echo Reply")
1080 self.assertEqual(icmpv6.id, icmpv6_id)
1081 self.assertEqual(icmpv6.seq, icmpv6_seq)
1082 self.assertEqual(icmpv6.data, icmpv6_data)
1085 class TestIPv6RD(TestIPv6ND):
1086 """ IPv6 Router Discovery Test Case """
1089 def setUpClass(cls):
1090 super(TestIPv6RD, cls).setUpClass()
1093 def tearDownClass(cls):
1094 super(TestIPv6RD, cls).tearDownClass()
1097 super(TestIPv6RD, self).setUp()
1099 # create 2 pg interfaces
1100 self.create_pg_interfaces(range(2))
1102 self.interfaces = list(self.pg_interfaces)
1104 # setup all interfaces
1105 for i in self.interfaces:
1110 for i in self.interfaces:
1113 super(TestIPv6RD, self).tearDown()
1115 def test_rd_send_router_solicitation(self):
1116 """ Verify router solicitation packets """
1119 self.pg_enable_capture(self.pg_interfaces)
1121 self.vapi.ip6nd_send_router_solicitation(self.pg1.sw_if_index,
1123 rx_list = self.pg1.get_capture(count, timeout=3)
1124 self.assertEqual(len(rx_list), count)
1125 for packet in rx_list:
1126 self.assertEqual(packet.haslayer(IPv6), 1)
1127 self.assertEqual(packet[IPv6].haslayer(
1129 dst = ip6_normalize(packet[IPv6].dst)
1130 dst2 = ip6_normalize("ff02::2")
1131 self.assert_equal(dst, dst2)
1132 src = ip6_normalize(packet[IPv6].src)
1133 src2 = ip6_normalize(self.pg1.local_ip6_ll)
1134 self.assert_equal(src, src2)
1136 bool(packet[ICMPv6ND_RS].haslayer(
1137 ICMPv6NDOptSrcLLAddr)))
1139 packet[ICMPv6NDOptSrcLLAddr].lladdr,
1142 def verify_prefix_info(self, reported_prefix, prefix_option):
1143 prefix = IPv6Network(
1144 text_type(prefix_option.getfieldval("prefix") +
1146 text_type(prefix_option.getfieldval("prefixlen"))),
1148 self.assert_equal(reported_prefix.prefix.network_address,
1149 prefix.network_address)
1150 L = prefix_option.getfieldval("L")
1151 A = prefix_option.getfieldval("A")
1152 option_flags = (L << 7) | (A << 6)
1153 self.assert_equal(reported_prefix.flags, option_flags)
1154 self.assert_equal(reported_prefix.valid_time,
1155 prefix_option.getfieldval("validlifetime"))
1156 self.assert_equal(reported_prefix.preferred_time,
1157 prefix_option.getfieldval("preferredlifetime"))
1159 def test_rd_receive_router_advertisement(self):
1160 """ Verify events triggered by received RA packets """
1162 self.vapi.want_ip6_ra_events()
1164 prefix_info_1 = ICMPv6NDOptPrefixInfo(
1168 preferredlifetime=500,
1173 prefix_info_2 = ICMPv6NDOptPrefixInfo(
1177 preferredlifetime=1000,
1182 p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
1183 IPv6(dst=self.pg1.local_ip6_ll,
1184 src=mk_ll_addr(self.pg1.remote_mac)) /
1188 self.pg1.add_stream([p])
1191 ev = self.vapi.wait_for_event(10, "ip6_ra_event")
1193 self.assert_equal(ev.current_hop_limit, 0)
1194 self.assert_equal(ev.flags, 8)
1195 self.assert_equal(ev.router_lifetime_in_sec, 1800)
1196 self.assert_equal(ev.neighbor_reachable_time_in_msec, 0)
1198 ev.time_in_msec_between_retransmitted_neighbor_solicitations, 0)
1200 self.assert_equal(ev.n_prefixes, 2)
1202 self.verify_prefix_info(ev.prefixes[0], prefix_info_1)
1203 self.verify_prefix_info(ev.prefixes[1], prefix_info_2)
1206 class TestIPv6RDControlPlane(TestIPv6ND):
1207 """ IPv6 Router Discovery Control Plane Test Case """
1210 def setUpClass(cls):
1211 super(TestIPv6RDControlPlane, cls).setUpClass()
1214 def tearDownClass(cls):
1215 super(TestIPv6RDControlPlane, cls).tearDownClass()
1218 super(TestIPv6RDControlPlane, self).setUp()
1220 # create 1 pg interface
1221 self.create_pg_interfaces(range(1))
1223 self.interfaces = list(self.pg_interfaces)
1225 # setup all interfaces
1226 for i in self.interfaces:
1231 super(TestIPv6RDControlPlane, self).tearDown()
1234 def create_ra_packet(pg, routerlifetime=None):
1235 src_ip = pg.remote_ip6_ll
1236 dst_ip = pg.local_ip6
1237 if routerlifetime is not None:
1238 ra = ICMPv6ND_RA(routerlifetime=routerlifetime)
1241 p = (Ether(dst=pg.local_mac, src=pg.remote_mac) /
1242 IPv6(dst=dst_ip, src=src_ip) / ra)
1246 def get_default_routes(fib):
1249 if entry.route.prefix.prefixlen == 0:
1250 for path in entry.route.paths:
1251 if path.sw_if_index != 0xFFFFFFFF:
1253 defaut_route['sw_if_index'] = path.sw_if_index
1254 defaut_route['next_hop'] = path.nh.address.ip6
1255 list.append(defaut_route)
1259 def get_interface_addresses(fib, pg):
1262 if entry.route.prefix.prefixlen == 128:
1263 path = entry.route.paths[0]
1264 if path.sw_if_index == pg.sw_if_index:
1265 list.append(str(entry.route.prefix.network_address))
1269 """ Test handling of SLAAC addresses and default routes """
1271 fib = self.vapi.ip_route_dump(0, True)
1272 default_routes = self.get_default_routes(fib)
1273 initial_addresses = set(self.get_interface_addresses(fib, self.pg0))
1274 self.assertEqual(default_routes, [])
1275 router_address = IPv6Address(text_type(self.pg0.remote_ip6_ll))
1277 self.vapi.ip6_nd_address_autoconfig(self.pg0.sw_if_index, 1, 1)
1282 packet = (self.create_ra_packet(
1283 self.pg0) / ICMPv6NDOptPrefixInfo(
1287 preferredlifetime=2,
1290 ) / ICMPv6NDOptPrefixInfo(
1294 preferredlifetime=1000,
1298 self.pg0.add_stream([packet])
1301 self.sleep_on_vpp_time(0.1)
1303 fib = self.vapi.ip_route_dump(0, True)
1305 # check FIB for new address
1306 addresses = set(self.get_interface_addresses(fib, self.pg0))
1307 new_addresses = addresses.difference(initial_addresses)
1308 self.assertEqual(len(new_addresses), 1)
1309 prefix = IPv6Network(text_type("%s/%d" % (list(new_addresses)[0], 20)),
1311 self.assertEqual(prefix, IPv6Network(text_type('1::/20')))
1313 # check FIB for new default route
1314 default_routes = self.get_default_routes(fib)
1315 self.assertEqual(len(default_routes), 1)
1316 dr = default_routes[0]
1317 self.assertEqual(dr['sw_if_index'], self.pg0.sw_if_index)
1318 self.assertEqual(dr['next_hop'], router_address)
1320 # send RA to delete default route
1321 packet = self.create_ra_packet(self.pg0, routerlifetime=0)
1322 self.pg0.add_stream([packet])
1325 self.sleep_on_vpp_time(0.1)
1327 # check that default route is deleted
1328 fib = self.vapi.ip_route_dump(0, True)
1329 default_routes = self.get_default_routes(fib)
1330 self.assertEqual(len(default_routes), 0)
1332 self.sleep_on_vpp_time(0.1)
1335 packet = self.create_ra_packet(self.pg0)
1336 self.pg0.add_stream([packet])
1339 self.sleep_on_vpp_time(0.1)
1341 # check FIB for new default route
1342 fib = self.vapi.ip_route_dump(0, True)
1343 default_routes = self.get_default_routes(fib)
1344 self.assertEqual(len(default_routes), 1)
1345 dr = default_routes[0]
1346 self.assertEqual(dr['sw_if_index'], self.pg0.sw_if_index)
1347 self.assertEqual(dr['next_hop'], router_address)
1349 # send RA, updating router lifetime to 1s
1350 packet = self.create_ra_packet(self.pg0, 1)
1351 self.pg0.add_stream([packet])
1354 self.sleep_on_vpp_time(0.1)
1356 # check that default route still exists
1357 fib = self.vapi.ip_route_dump(0, True)
1358 default_routes = self.get_default_routes(fib)
1359 self.assertEqual(len(default_routes), 1)
1360 dr = default_routes[0]
1361 self.assertEqual(dr['sw_if_index'], self.pg0.sw_if_index)
1362 self.assertEqual(dr['next_hop'], router_address)
1364 self.sleep_on_vpp_time(1)
1366 # check that default route is deleted
1367 fib = self.vapi.ip_route_dump(0, True)
1368 default_routes = self.get_default_routes(fib)
1369 self.assertEqual(len(default_routes), 0)
1371 # check FIB still contains the SLAAC address
1372 addresses = set(self.get_interface_addresses(fib, self.pg0))
1373 new_addresses = addresses.difference(initial_addresses)
1375 self.assertEqual(len(new_addresses), 1)
1376 prefix = IPv6Network(text_type("%s/%d" % (list(new_addresses)[0], 20)),
1378 self.assertEqual(prefix, IPv6Network(text_type('1::/20')))
1380 self.sleep_on_vpp_time(1)
1382 # check that SLAAC address is deleted
1383 fib = self.vapi.ip_route_dump(0, True)
1384 addresses = set(self.get_interface_addresses(fib, self.pg0))
1385 new_addresses = addresses.difference(initial_addresses)
1386 self.assertEqual(len(new_addresses), 0)
1389 class IPv6NDProxyTest(TestIPv6ND):
1390 """ IPv6 ND ProxyTest Case """
1393 def setUpClass(cls):
1394 super(IPv6NDProxyTest, cls).setUpClass()
1397 def tearDownClass(cls):
1398 super(IPv6NDProxyTest, cls).tearDownClass()
1401 super(IPv6NDProxyTest, self).setUp()
1403 # create 3 pg interfaces
1404 self.create_pg_interfaces(range(3))
1406 # pg0 is the master interface, with the configured subnet
1408 self.pg0.config_ip6()
1409 self.pg0.resolve_ndp()
1411 self.pg1.ip6_enable()
1412 self.pg2.ip6_enable()
1415 super(IPv6NDProxyTest, self).tearDown()
1417 def test_nd_proxy(self):
1418 """ IPv6 Proxy ND """
1421 # Generate some hosts in the subnet that we are proxying
1423 self.pg0.generate_remote_hosts(8)
1425 nsma = in6_getnsma(inet_pton(AF_INET6, self.pg0.local_ip6))
1426 d = inet_ntop(AF_INET6, nsma)
1429 # Send an NS for one of those remote hosts on one of the proxy links
1430 # expect no response since it's from an address that is not
1431 # on the link that has the prefix configured
1433 ns_pg1 = (Ether(dst=in6_getnsmac(nsma), src=self.pg1.remote_mac) /
1435 src=self.pg0._remote_hosts[2].ip6) /
1436 ICMPv6ND_NS(tgt=self.pg0.local_ip6) /
1437 ICMPv6NDOptSrcLLAddr(
1438 lladdr=self.pg0._remote_hosts[2].mac))
1440 self.send_and_assert_no_replies(self.pg1, ns_pg1, "Off link NS")
1443 # Add proxy support for the host
1445 self.vapi.ip6nd_proxy_add_del(
1446 ip=inet_pton(AF_INET6, self.pg0._remote_hosts[2].ip6),
1447 sw_if_index=self.pg1.sw_if_index)
1450 # try that NS again. this time we expect an NA back
1452 self.send_and_expect_na(self.pg1, ns_pg1,
1453 "NS to proxy entry",
1454 dst_ip=self.pg0._remote_hosts[2].ip6,
1455 tgt_ip=self.pg0.local_ip6)
1458 # ... and that we have an entry in the ND cache
1460 self.assertTrue(find_nbr(self,
1461 self.pg1.sw_if_index,
1462 self.pg0._remote_hosts[2].ip6))
1465 # ... and we can route traffic to it
1467 t = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
1468 IPv6(dst=self.pg0._remote_hosts[2].ip6,
1469 src=self.pg0.remote_ip6) /
1470 inet6.UDP(sport=10000, dport=20000) /
1473 self.pg0.add_stream(t)
1474 self.pg_enable_capture(self.pg_interfaces)
1476 rx = self.pg1.get_capture(1)
1479 self.assertEqual(rx[Ether].dst, self.pg0._remote_hosts[2].mac)
1480 self.assertEqual(rx[Ether].src, self.pg1.local_mac)
1482 self.assertEqual(rx[IPv6].src,
1484 self.assertEqual(rx[IPv6].dst,
1488 # Test we proxy for the host on the main interface
1490 ns_pg0 = (Ether(dst=in6_getnsmac(nsma), src=self.pg0.remote_mac) /
1491 IPv6(dst=d, src=self.pg0.remote_ip6) /
1493 tgt=self.pg0._remote_hosts[2].ip6) /
1494 ICMPv6NDOptSrcLLAddr(
1495 lladdr=self.pg0.remote_mac))
1497 self.send_and_expect_na(self.pg0, ns_pg0,
1498 "NS to proxy entry on main",
1499 tgt_ip=self.pg0._remote_hosts[2].ip6,
1500 dst_ip=self.pg0.remote_ip6)
1503 # Setup and resolve proxy for another host on another interface
1505 ns_pg2 = (Ether(dst=in6_getnsmac(nsma), src=self.pg2.remote_mac) /
1507 src=self.pg0._remote_hosts[3].ip6) /
1508 ICMPv6ND_NS(tgt=self.pg0.local_ip6) /
1509 ICMPv6NDOptSrcLLAddr(
1510 lladdr=self.pg0._remote_hosts[2].mac))
1512 self.vapi.ip6nd_proxy_add_del(
1513 ip=inet_pton(AF_INET6, self.pg0._remote_hosts[3].ip6),
1514 sw_if_index=self.pg2.sw_if_index)
1516 self.send_and_expect_na(self.pg2, ns_pg2,
1517 "NS to proxy entry other interface",
1518 dst_ip=self.pg0._remote_hosts[3].ip6,
1519 tgt_ip=self.pg0.local_ip6)
1521 self.assertTrue(find_nbr(self,
1522 self.pg2.sw_if_index,
1523 self.pg0._remote_hosts[3].ip6))
1526 # hosts can communicate. pg2->pg1
1528 t2 = (Ether(dst=self.pg2.local_mac,
1529 src=self.pg0.remote_hosts[3].mac) /
1530 IPv6(dst=self.pg0._remote_hosts[2].ip6,
1531 src=self.pg0._remote_hosts[3].ip6) /
1532 inet6.UDP(sport=10000, dport=20000) /
1535 self.pg2.add_stream(t2)
1536 self.pg_enable_capture(self.pg_interfaces)
1538 rx = self.pg1.get_capture(1)
1541 self.assertEqual(rx[Ether].dst, self.pg0._remote_hosts[2].mac)
1542 self.assertEqual(rx[Ether].src, self.pg1.local_mac)
1544 self.assertEqual(rx[IPv6].src,
1546 self.assertEqual(rx[IPv6].dst,
1550 # remove the proxy configs
1552 self.vapi.ip6nd_proxy_add_del(
1553 ip=inet_pton(AF_INET6, self.pg0._remote_hosts[2].ip6),
1554 sw_if_index=self.pg1.sw_if_index, is_del=1)
1555 self.vapi.ip6nd_proxy_add_del(
1556 ip=inet_pton(AF_INET6, self.pg0._remote_hosts[3].ip6),
1557 sw_if_index=self.pg2.sw_if_index, is_del=1)
1559 self.assertFalse(find_nbr(self,
1560 self.pg2.sw_if_index,
1561 self.pg0._remote_hosts[3].ip6))
1562 self.assertFalse(find_nbr(self,
1563 self.pg1.sw_if_index,
1564 self.pg0._remote_hosts[2].ip6))
1567 # no longer proxy-ing...
1569 self.send_and_assert_no_replies(self.pg0, ns_pg0, "Proxy unconfigured")
1570 self.send_and_assert_no_replies(self.pg1, ns_pg1, "Proxy unconfigured")
1571 self.send_and_assert_no_replies(self.pg2, ns_pg2, "Proxy unconfigured")
1574 # no longer forwarding. traffic generates NS out of the glean/main
1577 self.pg2.add_stream(t2)
1578 self.pg_enable_capture(self.pg_interfaces)
1581 rx = self.pg0.get_capture(1)
1583 self.assertTrue(rx[0].haslayer(ICMPv6ND_NS))
1586 class TestIPNull(VppTestCase):
1587 """ IPv6 routes via NULL """
1590 def setUpClass(cls):
1591 super(TestIPNull, cls).setUpClass()
1594 def tearDownClass(cls):
1595 super(TestIPNull, cls).tearDownClass()
1598 super(TestIPNull, self).setUp()
1600 # create 2 pg interfaces
1601 self.create_pg_interfaces(range(1))
1603 for i in self.pg_interfaces:
1609 super(TestIPNull, self).tearDown()
1610 for i in self.pg_interfaces:
1614 def test_ip_null(self):
1615 """ IP NULL route """
1617 p = (Ether(src=self.pg0.remote_mac,
1618 dst=self.pg0.local_mac) /
1619 IPv6(src=self.pg0.remote_ip6, dst="2001::1") /
1620 inet6.UDP(sport=1234, dport=1234) /
1624 # A route via IP NULL that will reply with ICMP unreachables
1626 ip_unreach = VppIpRoute(
1628 [VppRoutePath("::", 0xffffffff,
1629 type=FibPathType.FIB_PATH_TYPE_ICMP_UNREACH)])
1630 ip_unreach.add_vpp_config()
1632 self.pg0.add_stream(p)
1633 self.pg_enable_capture(self.pg_interfaces)
1636 rx = self.pg0.get_capture(1)
1638 icmp = rx[ICMPv6DestUnreach]
1640 # 0 = "No route to destination"
1641 self.assertEqual(icmp.code, 0)
1643 # ICMP is rate limited. pause a bit
1647 # A route via IP NULL that will reply with ICMP prohibited
1649 ip_prohibit = VppIpRoute(
1650 self, "2001::1", 128,
1651 [VppRoutePath("::", 0xffffffff,
1652 type=FibPathType.FIB_PATH_TYPE_ICMP_PROHIBIT)])
1653 ip_prohibit.add_vpp_config()
1655 self.pg0.add_stream(p)
1656 self.pg_enable_capture(self.pg_interfaces)
1659 rx = self.pg0.get_capture(1)
1661 icmp = rx[ICMPv6DestUnreach]
1663 # 1 = "Communication with destination administratively prohibited"
1664 self.assertEqual(icmp.code, 1)
1667 class TestIPDisabled(VppTestCase):
1668 """ IPv6 disabled """
1671 def setUpClass(cls):
1672 super(TestIPDisabled, cls).setUpClass()
1675 def tearDownClass(cls):
1676 super(TestIPDisabled, cls).tearDownClass()
1679 super(TestIPDisabled, self).setUp()
1681 # create 2 pg interfaces
1682 self.create_pg_interfaces(range(2))
1686 self.pg0.config_ip6()
1687 self.pg0.resolve_ndp()
1689 # PG 1 is not IP enabled
1693 super(TestIPDisabled, self).tearDown()
1694 for i in self.pg_interfaces:
1698 def test_ip_disabled(self):
1703 # one accepting interface, pg0, 2 forwarding interfaces
1705 route_ff_01 = VppIpMRoute(
1709 MRouteEntryFlags.MFIB_ENTRY_FLAG_NONE,
1710 [VppMRoutePath(self.pg1.sw_if_index,
1711 MRouteItfFlags.MFIB_ITF_FLAG_ACCEPT),
1712 VppMRoutePath(self.pg0.sw_if_index,
1713 MRouteItfFlags.MFIB_ITF_FLAG_FORWARD)])
1714 route_ff_01.add_vpp_config()
1716 pu = (Ether(src=self.pg1.remote_mac,
1717 dst=self.pg1.local_mac) /
1718 IPv6(src="2001::1", dst=self.pg0.remote_ip6) /
1719 inet6.UDP(sport=1234, dport=1234) /
1721 pm = (Ether(src=self.pg1.remote_mac,
1722 dst=self.pg1.local_mac) /
1723 IPv6(src="2001::1", dst="ffef::1") /
1724 inet6.UDP(sport=1234, dport=1234) /
1728 # PG1 does not forward IP traffic
1730 self.send_and_assert_no_replies(self.pg1, pu, "IPv6 disabled")
1731 self.send_and_assert_no_replies(self.pg1, pm, "IPv6 disabled")
1736 self.pg1.config_ip6()
1739 # Now we get packets through
1741 self.pg1.add_stream(pu)
1742 self.pg_enable_capture(self.pg_interfaces)
1744 rx = self.pg0.get_capture(1)
1746 self.pg1.add_stream(pm)
1747 self.pg_enable_capture(self.pg_interfaces)
1749 rx = self.pg0.get_capture(1)
1754 self.pg1.unconfig_ip6()
1757 # PG1 does not forward IP traffic
1759 self.send_and_assert_no_replies(self.pg1, pu, "IPv6 disabled")
1760 self.send_and_assert_no_replies(self.pg1, pm, "IPv6 disabled")
1763 class TestIP6LoadBalance(VppTestCase):
1764 """ IPv6 Load-Balancing """
1767 def setUpClass(cls):
1768 super(TestIP6LoadBalance, cls).setUpClass()
1771 def tearDownClass(cls):
1772 super(TestIP6LoadBalance, cls).tearDownClass()
1775 super(TestIP6LoadBalance, self).setUp()
1777 self.create_pg_interfaces(range(5))
1779 mpls_tbl = VppMplsTable(self, 0)
1780 mpls_tbl.add_vpp_config()
1782 for i in self.pg_interfaces:
1789 for i in self.pg_interfaces:
1793 super(TestIP6LoadBalance, self).tearDown()
1795 def pg_send(self, input, pkts):
1796 self.vapi.cli("clear trace")
1797 input.add_stream(pkts)
1798 self.pg_enable_capture(self.pg_interfaces)
1801 def send_and_expect_load_balancing(self, input, pkts, outputs):
1802 self.pg_send(input, pkts)
1804 rx = oo._get_capture(1)
1805 self.assertNotEqual(0, len(rx))
1807 def send_and_expect_one_itf(self, input, pkts, itf):
1808 self.pg_send(input, pkts)
1809 rx = itf.get_capture(len(pkts))
1811 def test_ip6_load_balance(self):
1812 """ IPv6 Load-Balancing """
1815 # An array of packets that differ only in the destination port
1819 # - MPLS non-EOS with an entropy label
1823 port_mpls_neos_pkts = []
1827 # An array of packets that differ only in the source address
1832 for ii in range(NUM_PKTS):
1834 IPv6(dst="3000::1", src="3000:1::1") /
1835 inet6.UDP(sport=1234, dport=1234 + ii) /
1837 port_ip_pkts.append((Ether(src=self.pg0.remote_mac,
1838 dst=self.pg0.local_mac) /
1840 port_mpls_pkts.append((Ether(src=self.pg0.remote_mac,
1841 dst=self.pg0.local_mac) /
1842 MPLS(label=66, ttl=2) /
1844 port_mpls_neos_pkts.append((Ether(src=self.pg0.remote_mac,
1845 dst=self.pg0.local_mac) /
1846 MPLS(label=67, ttl=2) /
1847 MPLS(label=77, ttl=2) /
1849 port_ent_pkts.append((Ether(src=self.pg0.remote_mac,
1850 dst=self.pg0.local_mac) /
1851 MPLS(label=67, ttl=2) /
1852 MPLS(label=14, ttl=2) /
1853 MPLS(label=999, ttl=2) /
1856 IPv6(dst="3000::1", src="3000:1::%d" % ii) /
1857 inet6.UDP(sport=1234, dport=1234) /
1859 src_ip_pkts.append((Ether(src=self.pg0.remote_mac,
1860 dst=self.pg0.local_mac) /
1862 src_mpls_pkts.append((Ether(src=self.pg0.remote_mac,
1863 dst=self.pg0.local_mac) /
1864 MPLS(label=66, ttl=2) /
1868 # A route for the IP packets
1870 route_3000_1 = VppIpRoute(self, "3000::1", 128,
1871 [VppRoutePath(self.pg1.remote_ip6,
1872 self.pg1.sw_if_index),
1873 VppRoutePath(self.pg2.remote_ip6,
1874 self.pg2.sw_if_index)])
1875 route_3000_1.add_vpp_config()
1878 # a local-label for the EOS packets
1880 binding = VppMplsIpBind(self, 66, "3000::1", 128, is_ip6=1)
1881 binding.add_vpp_config()
1884 # An MPLS route for the non-EOS packets
1886 route_67 = VppMplsRoute(self, 67, 0,
1887 [VppRoutePath(self.pg1.remote_ip6,
1888 self.pg1.sw_if_index,
1890 VppRoutePath(self.pg2.remote_ip6,
1891 self.pg2.sw_if_index,
1893 route_67.add_vpp_config()
1896 # inject the packet on pg0 - expect load-balancing across the 2 paths
1897 # - since the default hash config is to use IP src,dst and port
1899 # We are not going to ensure equal amounts of packets across each link,
1900 # since the hash algorithm is statistical and therefore this can never
1901 # be guaranteed. But with 64 different packets we do expect some
1902 # balancing. So instead just ensure there is traffic on each link.
1904 self.send_and_expect_load_balancing(self.pg0, port_ip_pkts,
1905 [self.pg1, self.pg2])
1906 self.send_and_expect_load_balancing(self.pg0, src_ip_pkts,
1907 [self.pg1, self.pg2])
1908 self.send_and_expect_load_balancing(self.pg0, port_mpls_pkts,
1909 [self.pg1, self.pg2])
1910 self.send_and_expect_load_balancing(self.pg0, src_mpls_pkts,
1911 [self.pg1, self.pg2])
1912 self.send_and_expect_load_balancing(self.pg0, port_mpls_neos_pkts,
1913 [self.pg1, self.pg2])
1916 # The packets with Entropy label in should not load-balance,
1917 # since the Entropy value is fixed.
1919 self.send_and_expect_one_itf(self.pg0, port_ent_pkts, self.pg1)
1922 # change the flow hash config so it's only IP src,dst
1923 # - now only the stream with differing source address will
1926 self.vapi.set_ip_flow_hash(vrf_id=0, src=1, dst=1, sport=0, dport=0,
1929 self.send_and_expect_load_balancing(self.pg0, src_ip_pkts,
1930 [self.pg1, self.pg2])
1931 self.send_and_expect_load_balancing(self.pg0, src_mpls_pkts,
1932 [self.pg1, self.pg2])
1933 self.send_and_expect_one_itf(self.pg0, port_ip_pkts, self.pg2)
1936 # change the flow hash config back to defaults
1938 self.vapi.set_ip_flow_hash(vrf_id=0, src=1, dst=1, sport=1, dport=1,
1942 # Recursive prefixes
1943 # - testing that 2 stages of load-balancing occurs and there is no
1944 # polarisation (i.e. only 2 of 4 paths are used)
1949 for ii in range(257):
1950 port_pkts.append((Ether(src=self.pg0.remote_mac,
1951 dst=self.pg0.local_mac) /
1954 inet6.UDP(sport=1234,
1956 Raw(b'\xa5' * 100)))
1957 src_pkts.append((Ether(src=self.pg0.remote_mac,
1958 dst=self.pg0.local_mac) /
1960 src="4000:1::%d" % ii) /
1961 inet6.UDP(sport=1234, dport=1234) /
1962 Raw(b'\xa5' * 100)))
1964 route_3000_2 = VppIpRoute(self, "3000::2", 128,
1965 [VppRoutePath(self.pg3.remote_ip6,
1966 self.pg3.sw_if_index),
1967 VppRoutePath(self.pg4.remote_ip6,
1968 self.pg4.sw_if_index)])
1969 route_3000_2.add_vpp_config()
1971 route_4000_1 = VppIpRoute(self, "4000::1", 128,
1972 [VppRoutePath("3000::1",
1974 VppRoutePath("3000::2",
1976 route_4000_1.add_vpp_config()
1979 # inject the packet on pg0 - expect load-balancing across all 4 paths
1981 self.vapi.cli("clear trace")
1982 self.send_and_expect_load_balancing(self.pg0, port_pkts,
1983 [self.pg1, self.pg2,
1984 self.pg3, self.pg4])
1985 self.send_and_expect_load_balancing(self.pg0, src_pkts,
1986 [self.pg1, self.pg2,
1987 self.pg3, self.pg4])
1990 # Recursive prefixes
1991 # - testing that 2 stages of load-balancing no choices
1995 for ii in range(257):
1996 port_pkts.append((Ether(src=self.pg0.remote_mac,
1997 dst=self.pg0.local_mac) /
2000 inet6.UDP(sport=1234,
2002 Raw(b'\xa5' * 100)))
2004 route_5000_2 = VppIpRoute(self, "5000::2", 128,
2005 [VppRoutePath(self.pg3.remote_ip6,
2006 self.pg3.sw_if_index)])
2007 route_5000_2.add_vpp_config()
2009 route_6000_1 = VppIpRoute(self, "6000::1", 128,
2010 [VppRoutePath("5000::2",
2012 route_6000_1.add_vpp_config()
2015 # inject the packet on pg0 - expect load-balancing across all 4 paths
2017 self.vapi.cli("clear trace")
2018 self.send_and_expect_one_itf(self.pg0, port_pkts, self.pg3)
2021 class TestIP6Punt(VppTestCase):
2022 """ IPv6 Punt Police/Redirect """
2025 def setUpClass(cls):
2026 super(TestIP6Punt, cls).setUpClass()
2029 def tearDownClass(cls):
2030 super(TestIP6Punt, cls).tearDownClass()
2033 super(TestIP6Punt, self).setUp()
2035 self.create_pg_interfaces(range(4))
2037 for i in self.pg_interfaces:
2043 super(TestIP6Punt, self).tearDown()
2044 for i in self.pg_interfaces:
2048 def test_ip_punt(self):
2049 """ IP6 punt police and redirect """
2051 p = (Ether(src=self.pg0.remote_mac,
2052 dst=self.pg0.local_mac) /
2053 IPv6(src=self.pg0.remote_ip6,
2054 dst=self.pg0.local_ip6) /
2055 inet6.TCP(sport=1234, dport=1234) /
2061 # Configure a punt redirect via pg1.
2063 nh_addr = self.pg1.remote_ip6
2064 self.vapi.ip_punt_redirect(self.pg0.sw_if_index,
2065 self.pg1.sw_if_index,
2068 self.send_and_expect(self.pg0, pkts, self.pg1)
2073 policer = self.vapi.policer_add_del(b"ip6-punt", 400, 0, 10, 0,
2075 self.vapi.ip_punt_police(policer.policer_index, is_ip6=1)
2077 self.vapi.cli("clear trace")
2078 self.pg0.add_stream(pkts)
2079 self.pg_enable_capture(self.pg_interfaces)
2083 # the number of packet received should be greater than 0,
2084 # but not equal to the number sent, since some were policed
2086 rx = self.pg1._get_capture(1)
2087 self.assertGreater(len(rx), 0)
2088 self.assertLess(len(rx), len(pkts))
2091 # remove the policer. back to full rx
2093 self.vapi.ip_punt_police(policer.policer_index, is_add=0, is_ip6=1)
2094 self.vapi.policer_add_del(b"ip6-punt", 400, 0, 10, 0,
2095 rate_type=1, is_add=0)
2096 self.send_and_expect(self.pg0, pkts, self.pg1)
2099 # remove the redirect. expect full drop.
2101 self.vapi.ip_punt_redirect(self.pg0.sw_if_index,
2102 self.pg1.sw_if_index,
2105 self.send_and_assert_no_replies(self.pg0, pkts,
2106 "IP no punt config")
2109 # Add a redirect that is not input port selective
2111 self.vapi.ip_punt_redirect(0xffffffff,
2112 self.pg1.sw_if_index,
2114 self.send_and_expect(self.pg0, pkts, self.pg1)
2116 self.vapi.ip_punt_redirect(0xffffffff,
2117 self.pg1.sw_if_index,
2121 def test_ip_punt_dump(self):
2122 """ IP6 punt redirect dump"""
2125 # Configure a punt redirects
2127 nh_addr = self.pg3.remote_ip6
2128 self.vapi.ip_punt_redirect(self.pg0.sw_if_index,
2129 self.pg3.sw_if_index,
2131 self.vapi.ip_punt_redirect(self.pg1.sw_if_index,
2132 self.pg3.sw_if_index,
2134 self.vapi.ip_punt_redirect(self.pg2.sw_if_index,
2135 self.pg3.sw_if_index,
2139 # Dump pg0 punt redirects
2141 punts = self.vapi.ip_punt_redirect_dump(self.pg0.sw_if_index,
2144 self.assertEqual(p.punt.rx_sw_if_index, self.pg0.sw_if_index)
2147 # Dump punt redirects for all interfaces
2149 punts = self.vapi.ip_punt_redirect_dump(0xffffffff, is_ipv6=1)
2150 self.assertEqual(len(punts), 3)
2152 self.assertEqual(p.punt.tx_sw_if_index, self.pg3.sw_if_index)
2153 self.assertNotEqual(punts[1].punt.nh, self.pg3.remote_ip6)
2154 self.assertEqual(str(punts[2].punt.nh), '::')
2157 class TestIPDeag(VppTestCase):
2158 """ IPv6 Deaggregate Routes """
2161 def setUpClass(cls):
2162 super(TestIPDeag, cls).setUpClass()
2165 def tearDownClass(cls):
2166 super(TestIPDeag, cls).tearDownClass()
2169 super(TestIPDeag, self).setUp()
2171 self.create_pg_interfaces(range(3))
2173 for i in self.pg_interfaces:
2179 super(TestIPDeag, self).tearDown()
2180 for i in self.pg_interfaces:
2184 def test_ip_deag(self):
2185 """ IP Deag Routes """
2188 # Create a table to be used for:
2189 # 1 - another destination address lookup
2190 # 2 - a source address lookup
2192 table_dst = VppIpTable(self, 1, is_ip6=1)
2193 table_src = VppIpTable(self, 2, is_ip6=1)
2194 table_dst.add_vpp_config()
2195 table_src.add_vpp_config()
2198 # Add a route in the default table to point to a deag/
2199 # second lookup in each of these tables
2201 route_to_dst = VppIpRoute(self, "1::1", 128,
2205 route_to_src = VppIpRoute(
2210 type=FibPathType.FIB_PATH_TYPE_SOURCE_LOOKUP)])
2212 route_to_dst.add_vpp_config()
2213 route_to_src.add_vpp_config()
2216 # packets to these destination are dropped, since they'll
2217 # hit the respective default routes in the second table
2219 p_dst = (Ether(src=self.pg0.remote_mac,
2220 dst=self.pg0.local_mac) /
2221 IPv6(src="5::5", dst="1::1") /
2222 inet6.TCP(sport=1234, dport=1234) /
2224 p_src = (Ether(src=self.pg0.remote_mac,
2225 dst=self.pg0.local_mac) /
2226 IPv6(src="2::2", dst="1::2") /
2227 inet6.TCP(sport=1234, dport=1234) /
2229 pkts_dst = p_dst * 257
2230 pkts_src = p_src * 257
2232 self.send_and_assert_no_replies(self.pg0, pkts_dst,
2234 self.send_and_assert_no_replies(self.pg0, pkts_src,
2238 # add a route in the dst table to forward via pg1
2240 route_in_dst = VppIpRoute(self, "1::1", 128,
2241 [VppRoutePath(self.pg1.remote_ip6,
2242 self.pg1.sw_if_index)],
2244 route_in_dst.add_vpp_config()
2246 self.send_and_expect(self.pg0, pkts_dst, self.pg1)
2249 # add a route in the src table to forward via pg2
2251 route_in_src = VppIpRoute(self, "2::2", 128,
2252 [VppRoutePath(self.pg2.remote_ip6,
2253 self.pg2.sw_if_index)],
2255 route_in_src.add_vpp_config()
2256 self.send_and_expect(self.pg0, pkts_src, self.pg2)
2259 # loop in the lookup DP
2261 route_loop = VppIpRoute(self, "3::3", 128,
2264 route_loop.add_vpp_config()
2266 p_l = (Ether(src=self.pg0.remote_mac,
2267 dst=self.pg0.local_mac) /
2268 IPv6(src="3::4", dst="3::3") /
2269 inet6.TCP(sport=1234, dport=1234) /
2272 self.send_and_assert_no_replies(self.pg0, p_l * 257,
2276 class TestIP6Input(VppTestCase):
2277 """ IPv6 Input Exception Test Cases """
2280 def setUpClass(cls):
2281 super(TestIP6Input, cls).setUpClass()
2284 def tearDownClass(cls):
2285 super(TestIP6Input, cls).tearDownClass()
2288 super(TestIP6Input, self).setUp()
2290 self.create_pg_interfaces(range(2))
2292 for i in self.pg_interfaces:
2298 super(TestIP6Input, self).tearDown()
2299 for i in self.pg_interfaces:
2303 def test_ip_input_icmp_reply(self):
2304 """ IP6 Input Exception - Return ICMP (3,0) """
2306 # hop limit - ICMP replies
2308 p_version = (Ether(src=self.pg0.remote_mac,
2309 dst=self.pg0.local_mac) /
2310 IPv6(src=self.pg0.remote_ip6,
2311 dst=self.pg1.remote_ip6,
2313 inet6.UDP(sport=1234, dport=1234) /
2316 rx = self.send_and_expect(self.pg0, p_version * NUM_PKTS, self.pg0)
2318 icmp = rx[ICMPv6TimeExceeded]
2320 # 0: "hop limit exceeded in transit",
2321 self.assertEqual((icmp.type, icmp.code), (3, 0))
2323 icmpv6_data = '\x0a' * 18
2325 all_1s = "FFFF:FFFF:FFFF:FFFF:FFFF:FFFF:FFFF:FFFF"
2327 @parameterized.expand([
2328 # Name, src, dst, l4proto, msg, timeout
2329 ("src='iface', dst='iface'", None, None,
2330 inet6.UDP(sport=1234, dport=1234), "funky version", None),
2331 ("src='All 0's', dst='iface'", all_0s, None,
2332 ICMPv6EchoRequest(id=0xb, seq=5, data=icmpv6_data), None, 0.1),
2333 ("src='iface', dst='All 0's'", None, all_0s,
2334 ICMPv6EchoRequest(id=0xb, seq=5, data=icmpv6_data), None, 0.1),
2335 ("src='All 1's', dst='iface'", all_1s, None,
2336 ICMPv6EchoRequest(id=0xb, seq=5, data=icmpv6_data), None, 0.1),
2337 ("src='iface', dst='All 1's'", None, all_1s,
2338 ICMPv6EchoRequest(id=0xb, seq=5, data=icmpv6_data), None, 0.1),
2339 ("src='All 1's', dst='All 1's'", all_1s, all_1s,
2340 ICMPv6EchoRequest(id=0xb, seq=5, data=icmpv6_data), None, 0.1),
2343 def test_ip_input_no_replies(self, name, src, dst, l4, msg, timeout):
2345 self._testMethodDoc = 'IPv6 Input Exception - %s' % name
2347 p_version = (Ether(src=self.pg0.remote_mac,
2348 dst=self.pg0.local_mac) /
2349 IPv6(src=src or self.pg0.remote_ip6,
2350 dst=dst or self.pg1.remote_ip6,
2355 self.send_and_assert_no_replies(self.pg0, p_version * NUM_PKTS,
2359 def test_hop_by_hop(self):
2360 """ Hop-by-hop header test """
2362 p = (Ether(src=self.pg0.remote_mac,
2363 dst=self.pg0.local_mac) /
2364 IPv6(src=self.pg0.remote_ip6, dst=self.pg0.local_ip6) /
2365 IPv6ExtHdrHopByHop() /
2366 inet6.UDP(sport=1234, dport=1234) /
2369 self.pg0.add_stream(p)
2370 self.pg_enable_capture(self.pg_interfaces)
2374 class TestIPReplace(VppTestCase):
2375 """ IPv6 Table Replace """
2378 def setUpClass(cls):
2379 super(TestIPReplace, cls).setUpClass()
2382 def tearDownClass(cls):
2383 super(TestIPReplace, cls).tearDownClass()
2386 super(TestIPReplace, self).setUp()
2388 self.create_pg_interfaces(range(4))
2393 for i in self.pg_interfaces:
2397 i.generate_remote_hosts(2)
2398 self.tables.append(VppIpTable(self, table_id,
2399 True).add_vpp_config())
2403 super(TestIPReplace, self).tearDown()
2404 for i in self.pg_interfaces:
2408 def test_replace(self):
2409 """ IP Table Replace """
2412 links = [self.pg0, self.pg1, self.pg2, self.pg3]
2413 routes = [[], [], [], []]
2415 # the sizes of 'empty' tables
2416 for t in self.tables:
2417 self.assertEqual(len(t.dump()), 2)
2418 self.assertEqual(len(t.mdump()), 5)
2420 # load up the tables with some routes
2421 for ii, t in enumerate(self.tables):
2422 for jj in range(1, N_ROUTES):
2424 self, "2001::%d" % jj if jj != 0 else "2001::", 128,
2425 [VppRoutePath(links[ii].remote_hosts[0].ip6,
2426 links[ii].sw_if_index),
2427 VppRoutePath(links[ii].remote_hosts[1].ip6,
2428 links[ii].sw_if_index)],
2429 table_id=t.table_id).add_vpp_config()
2430 multi = VppIpMRoute(
2432 "ff:2001::%d" % jj, 128,
2433 MRouteEntryFlags.MFIB_ENTRY_FLAG_NONE,
2434 [VppMRoutePath(self.pg0.sw_if_index,
2435 MRouteItfFlags.MFIB_ITF_FLAG_ACCEPT,
2436 proto=FibPathProto.FIB_PATH_NH_PROTO_IP6),
2437 VppMRoutePath(self.pg1.sw_if_index,
2438 MRouteItfFlags.MFIB_ITF_FLAG_FORWARD,
2439 proto=FibPathProto.FIB_PATH_NH_PROTO_IP6),
2440 VppMRoutePath(self.pg2.sw_if_index,
2441 MRouteItfFlags.MFIB_ITF_FLAG_FORWARD,
2442 proto=FibPathProto.FIB_PATH_NH_PROTO_IP6),
2443 VppMRoutePath(self.pg3.sw_if_index,
2444 MRouteItfFlags.MFIB_ITF_FLAG_FORWARD,
2445 proto=FibPathProto.FIB_PATH_NH_PROTO_IP6)],
2446 table_id=t.table_id).add_vpp_config()
2447 routes[ii].append({'uni': uni,
2451 # replace the tables a few times
2454 # replace each table
2455 for t in self.tables:
2458 # all the routes are still there
2459 for ii, t in enumerate(self.tables):
2462 for r in routes[ii]:
2463 self.assertTrue(find_route_in_dump(dump, r['uni'], t))
2464 self.assertTrue(find_mroute_in_dump(mdump, r['multi'], t))
2466 # redownload the even numbered routes
2467 for ii, t in enumerate(self.tables):
2468 for jj in range(0, N_ROUTES, 2):
2469 routes[ii][jj]['uni'].add_vpp_config()
2470 routes[ii][jj]['multi'].add_vpp_config()
2472 # signal each table converged
2473 for t in self.tables:
2476 # we should find the even routes, but not the odd
2477 for ii, t in enumerate(self.tables):
2480 for jj in range(0, N_ROUTES, 2):
2481 self.assertTrue(find_route_in_dump(
2482 dump, routes[ii][jj]['uni'], t))
2483 self.assertTrue(find_mroute_in_dump(
2484 mdump, routes[ii][jj]['multi'], t))
2485 for jj in range(1, N_ROUTES - 1, 2):
2486 self.assertFalse(find_route_in_dump(
2487 dump, routes[ii][jj]['uni'], t))
2488 self.assertFalse(find_mroute_in_dump(
2489 mdump, routes[ii][jj]['multi'], t))
2491 # reload all the routes
2492 for ii, t in enumerate(self.tables):
2493 for r in routes[ii]:
2494 r['uni'].add_vpp_config()
2495 r['multi'].add_vpp_config()
2497 # all the routes are still there
2498 for ii, t in enumerate(self.tables):
2501 for r in routes[ii]:
2502 self.assertTrue(find_route_in_dump(dump, r['uni'], t))
2503 self.assertTrue(find_mroute_in_dump(mdump, r['multi'], t))
2506 # finally flush the tables for good measure
2508 for t in self.tables:
2510 self.assertEqual(len(t.dump()), 2)
2511 self.assertEqual(len(t.mdump()), 5)
2514 if __name__ == '__main__':
2515 unittest.main(testRunner=VppTestRunner)