6 from framework import VppTestCase, VppTestRunner
7 from vpp_sub_interface import VppSubInterface, VppDot1QSubint
8 from vpp_pg_interface import is_ipv6_misc
10 from scapy.packet import Raw
11 from scapy.layers.l2 import Ether, Dot1Q
12 from scapy.layers.inet6 import IPv6, UDP, ICMPv6ND_NS, ICMPv6ND_RS, \
13 ICMPv6ND_RA, ICMPv6NDOptSrcLLAddr, getmacbyip6, ICMPv6MRD_Solicitation
15 from scapy.utils6 import in6_getnsma, in6_getnsmac, in6_ptop, in6_islladdr, \
16 in6_mactoifaceid, in6_ismaddr
17 from scapy.utils import inet_pton, inet_ntop
21 euid = in6_mactoifaceid(mac)
22 addr = "fe80::" + euid
26 class TestIPv6(VppTestCase):
27 """ IPv6 Test Case """
31 super(TestIPv6, cls).setUpClass()
35 Perform test setup before test case.
38 - create 3 pg interfaces
39 - untagged pg0 interface
40 - Dot1Q subinterface on pg1
41 - Dot1AD subinterface on pg2
43 - put it into UP state
45 - resolve neighbor address using NDP
46 - configure 200 fib entries
48 :ivar list interfaces: pg interfaces and subinterfaces.
49 :ivar dict flows: IPv4 packet flows in test.
50 :ivar list pg_if_packet_sizes: packet sizes in test.
52 *TODO:* Create AD sub interface
54 super(TestIPv6, self).setUp()
56 # create 3 pg interfaces
57 self.create_pg_interfaces(range(3))
59 # create 2 subinterfaces for p1 and pg2
60 self.sub_interfaces = [
61 VppDot1QSubint(self, self.pg1, 100),
62 VppDot1QSubint(self, self.pg2, 200)
63 # TODO: VppDot1ADSubint(self, self.pg2, 200, 300, 400)
66 # packet flows mapping pg0 -> pg1.sub, pg2.sub, etc.
68 self.flows[self.pg0] = [self.pg1.sub_if, self.pg2.sub_if]
69 self.flows[self.pg1.sub_if] = [self.pg0, self.pg2.sub_if]
70 self.flows[self.pg2.sub_if] = [self.pg0, self.pg1.sub_if]
73 self.pg_if_packet_sizes = [64, 512, 1518, 9018]
74 self.sub_if_packet_sizes = [64, 512, 1518 + 4, 9018 + 4]
76 self.interfaces = list(self.pg_interfaces)
77 self.interfaces.extend(self.sub_interfaces)
79 # setup all interfaces
80 for i in self.interfaces:
85 # config 2M FIB entries
86 self.config_fib_entries(200)
89 """Run standard test teardown and log ``show ip6 neighbors``."""
90 for i in self.sub_interfaces:
96 super(TestIPv6, self).tearDown()
98 self.logger.info(self.vapi.cli("show ip6 neighbors"))
99 # info(self.vapi.cli("show ip6 fib")) # many entries
101 def config_fib_entries(self, count):
102 """For each interface add to the FIB table *count* routes to
103 "fd02::1/128" destination with interface's local address as next-hop
106 :param int count: Number of FIB entries.
108 - *TODO:* check if the next-hop address shouldn't be remote address
109 instead of local address.
111 n_int = len(self.interfaces)
114 dest_addr = socket.inet_pton(socket.AF_INET6, "fd02::1")
116 for i in self.interfaces:
117 next_hop_address = i.local_ip6n
118 for j in range(count / n_int):
119 self.vapi.ip_add_del_route(
120 dest_addr, dest_addr_len, next_hop_address, is_ipv6=1)
122 if counter / count * 100 > percent:
123 self.logger.info("Configure %d FIB entries .. %d%% done" %
127 def create_stream(self, src_if, packet_sizes):
128 """Create input packet stream for defined interface.
130 :param VppInterface src_if: Interface to create packet stream for.
131 :param list packet_sizes: Required packet sizes.
134 for i in range(0, 257):
135 dst_if = self.flows[src_if][i % 2]
136 info = self.create_packet_info(src_if, dst_if)
137 payload = self.info_to_payload(info)
138 p = (Ether(dst=src_if.local_mac, src=src_if.remote_mac) /
139 IPv6(src=src_if.remote_ip6, dst=dst_if.remote_ip6) /
140 UDP(sport=1234, dport=1234) /
143 if isinstance(src_if, VppSubInterface):
144 p = src_if.add_dot1_layer(p)
145 size = packet_sizes[(i // 2) % len(packet_sizes)]
146 self.extend_packet(p, size)
150 def verify_capture(self, dst_if, capture):
151 """Verify captured input packet stream for defined interface.
153 :param VppInterface dst_if: Interface to verify captured packet stream
155 :param list capture: Captured packet stream.
157 self.logger.info("Verifying capture on interface %s" % dst_if.name)
159 for i in self.interfaces:
160 last_info[i.sw_if_index] = None
162 dst_sw_if_index = dst_if.sw_if_index
163 if hasattr(dst_if, 'parent'):
165 for packet in capture:
167 # Check VLAN tags and Ethernet header
168 packet = dst_if.remove_dot1_layer(packet)
169 self.assertTrue(Dot1Q not in packet)
173 payload_info = self.payload_to_info(str(packet[Raw]))
174 packet_index = payload_info.index
175 self.assertEqual(payload_info.dst, dst_sw_if_index)
177 "Got packet on port %s: src=%u (id=%u)" %
178 (dst_if.name, payload_info.src, packet_index))
179 next_info = self.get_next_packet_info_for_interface2(
180 payload_info.src, dst_sw_if_index,
181 last_info[payload_info.src])
182 last_info[payload_info.src] = next_info
183 self.assertTrue(next_info is not None)
184 self.assertEqual(packet_index, next_info.index)
185 saved_packet = next_info.data
186 # Check standard fields
187 self.assertEqual(ip.src, saved_packet[IPv6].src)
188 self.assertEqual(ip.dst, saved_packet[IPv6].dst)
189 self.assertEqual(udp.sport, saved_packet[UDP].sport)
190 self.assertEqual(udp.dport, saved_packet[UDP].dport)
192 self.logger.error(ppp("Unexpected or invalid packet:", packet))
194 for i in self.interfaces:
195 remaining_packet = self.get_next_packet_info_for_interface2(
196 i.sw_if_index, dst_sw_if_index, last_info[i.sw_if_index])
197 self.assertTrue(remaining_packet is None,
198 "Interface %s: Packet expected from interface %s "
199 "didn't arrive" % (dst_if.name, i.name))
205 - Create IPv6 stream for pg0 interface
206 - Create IPv6 tagged streams for pg1's and pg2's subinterface.
207 - Send and verify received packets on each interface.
210 pkts = self.create_stream(self.pg0, self.pg_if_packet_sizes)
211 self.pg0.add_stream(pkts)
213 for i in self.sub_interfaces:
214 pkts = self.create_stream(i, self.sub_if_packet_sizes)
215 i.parent.add_stream(pkts)
217 self.pg_enable_capture(self.pg_interfaces)
220 pkts = self.pg0.get_capture()
221 self.verify_capture(self.pg0, pkts)
223 for i in self.sub_interfaces:
224 pkts = i.parent.get_capture()
225 self.verify_capture(i, pkts)
227 def send_and_assert_no_replies(self, intf, pkts, remark):
228 intf.add_stream(pkts)
229 self.pg_enable_capture(self.pg_interfaces)
231 intf.assert_nothing_captured(remark=remark)
234 """ IPv6 Neighbour Solicitation Exceptions
237 - Send an NS Sourced from an address not covered by the link sub-net
238 - Send an NS to an mcast address the router has not joined
239 - Send NS for a target address the router does not onn.
243 # An NS from a non link source address
245 nsma = in6_getnsma(inet_pton(socket.AF_INET6, self.pg0.local_ip6))
246 d = inet_ntop(socket.AF_INET6, nsma)
248 p = (Ether(dst=in6_getnsmac(nsma)) /
249 IPv6(dst=d, src="2002::2") /
250 ICMPv6ND_NS(tgt=self.pg0.local_ip6) /
251 ICMPv6NDOptSrcLLAddr(lladdr=self.pg0.remote_mac))
254 self.send_and_assert_no_replies(
256 "No response to NS source by address not on sub-net")
259 # An NS for sent to a solicited mcast group the router is
260 # not a member of FAILS
263 nsma = in6_getnsma(inet_pton(socket.AF_INET6, "fd::ffff"))
264 d = inet_ntop(socket.AF_INET6, nsma)
266 p = (Ether(dst=in6_getnsmac(nsma)) /
267 IPv6(dst=d, src=self.pg0.remote_ip6) /
268 ICMPv6ND_NS(tgt=self.pg0.local_ip6) /
269 ICMPv6NDOptSrcLLAddr(lladdr=self.pg0.remote_mac))
272 self.send_and_assert_no_replies(
274 "No response to NS sent to unjoined mcast address")
277 # An NS whose target address is one the router does not own
279 nsma = in6_getnsma(inet_pton(socket.AF_INET6, self.pg0.local_ip6))
280 d = inet_ntop(socket.AF_INET6, nsma)
282 p = (Ether(dst=in6_getnsmac(nsma)) /
283 IPv6(dst=d, src=self.pg0.remote_ip6) /
284 ICMPv6ND_NS(tgt="fd::ffff") /
285 ICMPv6NDOptSrcLLAddr(lladdr=self.pg0.remote_mac))
288 self.send_and_assert_no_replies(self.pg0, pkts,
289 "No response to NS for unknown target")
291 def validate_ra(self, intf, rx, dst_ip=None):
293 dst_ip = intf.remote_ip6
295 # unicasted packets must come to the unicast mac
296 self.assertEqual(rx[Ether].dst, intf.remote_mac)
298 # and from the router's MAC
299 self.assertEqual(rx[Ether].src, intf.local_mac)
301 # the rx'd RA should be addressed to the sender's source
302 self.assertTrue(rx.haslayer(ICMPv6ND_RA))
303 self.assertEqual(in6_ptop(rx[IPv6].dst),
306 # and come from the router's link local
307 self.assertTrue(in6_islladdr(rx[IPv6].src))
308 self.assertEqual(in6_ptop(rx[IPv6].src),
309 in6_ptop(mk_ll_addr(intf.local_mac)))
311 def send_and_expect_ra(self, intf, pkts, remark, dst_ip=None,
312 filter_out_fn=is_ipv6_misc):
313 intf.add_stream(pkts)
314 self.pg0.add_stream(pkts)
315 self.pg_enable_capture(self.pg_interfaces)
317 rx = intf.get_capture(1, filter_out_fn=filter_out_fn)
319 self.assertEqual(len(rx), 1)
321 self.validate_ra(intf, rx, dst_ip)
324 """ IPv6 Router Solicitation Exceptions
330 # Before we begin change the IPv6 RA responses to use the unicast
331 # address - that way we will not confuse them with the periodic
332 # RAs which go to the mcast address
333 # Sit and wait for the first periodic RA.
337 self.pg0.ip6_ra_config(send_unicast=1)
340 # An RS from a link source address
341 # - expect an RA in return
343 p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
344 IPv6(dst=self.pg0.local_ip6, src=self.pg0.remote_ip6) /
347 self.send_and_expect_ra(self.pg0, pkts, "Genuine RS")
350 # For the next RS sent the RA should be rate limited
352 self.send_and_assert_no_replies(self.pg0, pkts, "RA rate limited")
355 # When we reconfiure the IPv6 RA config, we reset the RA rate limiting,
356 # so we need to do this before each test below so as not to drop
357 # packets for rate limiting reasons. Test this works here.
359 self.pg0.ip6_ra_config(send_unicast=1)
360 self.send_and_expect_ra(self.pg0, pkts, "Rate limit reset RS")
363 # An RS sent from a non-link local source
365 self.pg0.ip6_ra_config(send_unicast=1)
366 p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
367 IPv6(dst=self.pg0.local_ip6, src="2002::ffff") /
370 self.send_and_assert_no_replies(self.pg0, pkts,
371 "RS from non-link source")
374 # Source an RS from a link local address
376 self.pg0.ip6_ra_config(send_unicast=1)
377 ll = mk_ll_addr(self.pg0.remote_mac)
378 p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
379 IPv6(dst=self.pg0.local_ip6, src=ll) /
382 self.send_and_expect_ra(self.pg0, pkts,
383 "RS sourced from link-local",
387 # Send the RS multicast
389 self.pg0.ip6_ra_config(send_unicast=1)
390 dmac = in6_getnsmac(inet_pton(socket.AF_INET6, "ff02::2"))
391 ll = mk_ll_addr(self.pg0.remote_mac)
392 p = (Ether(dst=dmac, src=self.pg0.remote_mac) /
393 IPv6(dst="ff02::2", src=ll) /
396 self.send_and_expect_ra(self.pg0, pkts,
397 "RS sourced from link-local",
401 # Source from the unspecified address ::. This happens when the RS
402 # is sent before the host has a configured address/sub-net,
403 # i.e. auto-config. Since the sender has no IP address, the reply
404 # comes back mcast - so the capture needs to not filter this.
405 # If we happen to pick up the periodic RA at this point then so be it,
408 self.pg0.ip6_ra_config(send_unicast=1, suppress=1)
409 p = (Ether(dst=dmac, src=self.pg0.remote_mac) /
410 IPv6(dst="ff02::2", src="::") /
413 self.send_and_expect_ra(self.pg0, pkts,
414 "RS sourced from unspecified",
419 # Reset the periodic advertisements back to default values
421 self.pg0.ip6_ra_config(no=1, suppress=1, send_unicast=0)
423 if __name__ == '__main__':
424 unittest.main(testRunner=VppTestRunner)