8 from scapy.contrib.mpls import MPLS
9 from scapy.layers.inet import IP, UDP, TCP, ICMP, icmptypes, icmpcodes
10 from scapy.layers.l2 import Ether, Dot1Q, ARP
11 from scapy.packet import Raw
14 from framework import VppTestCase, VppTestRunner
16 from vpp_ip import VppIpPrefix
17 from vpp_ip_route import VppIpRoute, VppRoutePath, VppIpMRoute, \
18 VppMRoutePath, MRouteItfFlags, MRouteEntryFlags, VppMplsIpBind, \
19 VppMplsTable, VppIpTable, FibPathType, find_route, \
21 from vpp_ip import VppIpAddress
22 from vpp_sub_interface import VppSubInterface, VppDot1QSubint, VppDot1ADSubint
23 from vpp_papi import VppEnum
28 class TestIPv4(VppTestCase):
29 """ IPv4 Test Case """
33 super(TestIPv4, cls).setUpClass()
36 def tearDownClass(cls):
37 super(TestIPv4, cls).tearDownClass()
41 Perform test setup before test case.
44 - create 3 pg interfaces
45 - untagged pg0 interface
46 - Dot1Q subinterface on pg1
47 - Dot1AD subinterface on pg2
49 - put it into UP state
51 - resolve neighbor address using ARP
52 - configure 200 fib entries
54 :ivar list interfaces: pg interfaces and subinterfaces.
55 :ivar dict flows: IPv4 packet flows in test.
57 super(TestIPv4, self).setUp()
59 # create 3 pg interfaces
60 self.create_pg_interfaces(range(3))
62 # create 2 subinterfaces for pg1 and pg2
63 self.sub_interfaces = [
64 VppDot1QSubint(self, self.pg1, 100),
65 VppDot1ADSubint(self, self.pg2, 200, 300, 400)]
67 # packet flows mapping pg0 -> pg1.sub, pg2.sub, etc.
69 self.flows[self.pg0] = [self.pg1.sub_if, self.pg2.sub_if]
70 self.flows[self.pg1.sub_if] = [self.pg0, self.pg2.sub_if]
71 self.flows[self.pg2.sub_if] = [self.pg0, self.pg1.sub_if]
74 self.pg_if_packet_sizes = [64, 1500, 9020]
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
88 """Run standard test teardown and log ``show ip arp``."""
89 super(TestIPv4, self).tearDown()
91 def show_commands_at_teardown(self):
92 self.logger.info(self.vapi.cli("show ip arp"))
93 # info(self.vapi.cli("show ip fib")) # many entries
95 def modify_packet(self, src_if, packet_size, pkt):
96 """Add load, set destination IP and extend packet to required packet
97 size for defined interface.
99 :param VppInterface src_if: Interface to create packet for.
100 :param int packet_size: Required packet size.
101 :param Scapy pkt: Packet to be modified.
103 dst_if_idx = packet_size / 10 % 2
104 dst_if = self.flows[src_if][dst_if_idx]
105 info = self.create_packet_info(src_if, dst_if)
106 payload = self.info_to_payload(info)
108 p[IP].dst = dst_if.remote_ip4
110 if isinstance(src_if, VppSubInterface):
111 p = src_if.add_dot1_layer(p)
112 self.extend_packet(p, packet_size)
116 def create_stream(self, src_if):
117 """Create input packet stream for defined interface.
119 :param VppInterface src_if: Interface to create packet stream for.
121 hdr_ext = 4 if isinstance(src_if, VppSubInterface) else 0
122 pkt_tmpl = (Ether(dst=src_if.local_mac, src=src_if.remote_mac) /
123 IP(src=src_if.remote_ip4) /
124 UDP(sport=1234, dport=1234))
126 pkts = [self.modify_packet(src_if, i, pkt_tmpl)
127 for i in moves.range(self.pg_if_packet_sizes[0],
128 self.pg_if_packet_sizes[1], 10)]
129 pkts_b = [self.modify_packet(src_if, i, pkt_tmpl)
130 for i in moves.range(self.pg_if_packet_sizes[1] + hdr_ext,
131 self.pg_if_packet_sizes[2] + hdr_ext,
137 def verify_capture(self, dst_if, capture):
138 """Verify captured input packet stream for defined interface.
140 :param VppInterface dst_if: Interface to verify captured packet stream
142 :param list capture: Captured packet stream.
144 self.logger.info("Verifying capture on interface %s" % dst_if.name)
146 for i in self.interfaces:
147 last_info[i.sw_if_index] = None
149 dst_sw_if_index = dst_if.sw_if_index
150 if hasattr(dst_if, 'parent'):
152 for packet in capture:
154 # Check VLAN tags and Ethernet header
155 packet = dst_if.remove_dot1_layer(packet)
156 self.assertTrue(Dot1Q not in packet)
160 payload_info = self.payload_to_info(packet[Raw])
161 packet_index = payload_info.index
162 self.assertEqual(payload_info.dst, dst_sw_if_index)
164 "Got packet on port %s: src=%u (id=%u)" %
165 (dst_if.name, payload_info.src, packet_index))
166 next_info = self.get_next_packet_info_for_interface2(
167 payload_info.src, dst_sw_if_index,
168 last_info[payload_info.src])
169 last_info[payload_info.src] = next_info
170 self.assertTrue(next_info is not None)
171 self.assertEqual(packet_index, next_info.index)
172 saved_packet = next_info.data
173 # Check standard fields
174 self.assertEqual(ip.src, saved_packet[IP].src)
175 self.assertEqual(ip.dst, saved_packet[IP].dst)
176 self.assertEqual(udp.sport, saved_packet[UDP].sport)
177 self.assertEqual(udp.dport, saved_packet[UDP].dport)
179 self.logger.error(ppp("Unexpected or invalid packet:", packet))
181 for i in self.interfaces:
182 remaining_packet = self.get_next_packet_info_for_interface2(
183 i.sw_if_index, dst_sw_if_index, last_info[i.sw_if_index])
184 self.assertTrue(remaining_packet is None,
185 "Interface %s: Packet expected from interface %s "
186 "didn't arrive" % (dst_if.name, i.name))
193 - Create IPv4 stream for pg0 interface
194 - Create IPv4 tagged streams for pg1's and pg2's sub-interface.
195 - Send and verify received packets on each interface.
198 pkts = self.create_stream(self.pg0)
199 self.pg0.add_stream(pkts)
201 for i in self.sub_interfaces:
202 pkts = self.create_stream(i)
203 i.parent.add_stream(pkts)
205 self.pg_enable_capture(self.pg_interfaces)
208 pkts = self.pg0.get_capture()
209 self.verify_capture(self.pg0, pkts)
211 for i in self.sub_interfaces:
212 pkts = i.parent.get_capture()
213 self.verify_capture(i, pkts)
216 class TestIPv4IfAddrRoute(VppTestCase):
217 """ IPv4 Interface Addr Route Test Case """
221 super(TestIPv4IfAddrRoute, cls).setUpClass()
224 def tearDownClass(cls):
225 super(TestIPv4IfAddrRoute, cls).tearDownClass()
228 super(TestIPv4IfAddrRoute, self).setUp()
230 # create 1 pg interface
231 self.create_pg_interfaces(range(1))
233 for i in self.pg_interfaces:
239 super(TestIPv4IfAddrRoute, self).tearDown()
240 for i in self.pg_interfaces:
244 def test_ipv4_ifaddrs_same_prefix(self):
245 """ IPv4 Interface Addresses Same Prefix test
249 - Verify no route in FIB for prefix 10.10.10.0/24
250 - Configure IPv4 address 10.10.10.10/24 on an interface
251 - Verify route in FIB for prefix 10.10.10.0/24
252 - Configure IPv4 address 10.10.10.20/24 on an interface
253 - Delete 10.10.10.10/24 from interface
254 - Verify route in FIB for prefix 10.10.10.0/24
255 - Delete 10.10.10.20/24 from interface
256 - Verify no route in FIB for prefix 10.10.10.0/24
259 # create two addresses, verify route not present
260 if_addr1 = VppIpInterfaceAddress(self, self.pg0,
261 VppIpAddress("10.10.10.10"), 24)
262 if_addr2 = VppIpInterfaceAddress(self, self.pg0,
263 VppIpAddress("10.10.10.20"), 24)
264 self.assertFalse(if_addr1.query_vpp_config()) # 10.10.10.0/24
265 self.assertFalse(find_route(self, "10.10.10.10", 32))
266 self.assertFalse(find_route(self, "10.10.10.20", 32))
267 self.assertFalse(find_route(self, "10.10.10.255", 32))
268 self.assertFalse(find_route(self, "10.10.10.0", 32))
270 # configure first address, verify route present
271 if_addr1.add_vpp_config()
272 self.assertTrue(if_addr1.query_vpp_config()) # 10.10.10.0/24
273 self.assertTrue(find_route(self, "10.10.10.10", 32))
274 self.assertFalse(find_route(self, "10.10.10.20", 32))
275 self.assertTrue(find_route(self, "10.10.10.255", 32))
276 self.assertTrue(find_route(self, "10.10.10.0", 32))
278 # configure second address, delete first, verify route not removed
279 if_addr2.add_vpp_config()
280 if_addr1.remove_vpp_config()
281 self.assertTrue(if_addr1.query_vpp_config()) # 10.10.10.0/24
282 self.assertFalse(find_route(self, "10.10.10.10", 32))
283 self.assertTrue(find_route(self, "10.10.10.20", 32))
284 self.assertTrue(find_route(self, "10.10.10.255", 32))
285 self.assertTrue(find_route(self, "10.10.10.0", 32))
287 # delete second address, verify route removed
288 if_addr2.remove_vpp_config()
289 self.assertFalse(if_addr1.query_vpp_config()) # 10.10.10.0/24
290 self.assertFalse(find_route(self, "10.10.10.10", 32))
291 self.assertFalse(find_route(self, "10.10.10.20", 32))
292 self.assertFalse(find_route(self, "10.10.10.255", 32))
293 self.assertFalse(find_route(self, "10.10.10.0", 32))
295 def test_ipv4_ifaddr_route(self):
296 """ IPv4 Interface Address Route test
301 - Configure IPv4 address on loopback
302 - Verify that address is not in the FIB
304 - Verify that address is in the FIB now
305 - Bring loopback down
306 - Verify that address is not in the FIB anymore
308 - Configure IPv4 address on loopback
309 - Verify that address is in the FIB now
312 # create a loopback and configure IPv4
313 loopbacks = self.create_loopback_interfaces(1)
314 lo_if = self.lo_interfaces[0]
316 lo_if.local_ip4_prefix_len = 32
319 # The intf was down when addr was added -> entry not in FIB
320 fib4_dump = self.vapi.ip_route_dump(0)
321 self.assertFalse(lo_if.is_ip4_entry_in_fib_dump(fib4_dump))
323 # When intf is brought up, entry is added
325 fib4_dump = self.vapi.ip_route_dump(0)
326 self.assertTrue(lo_if.is_ip4_entry_in_fib_dump(fib4_dump))
328 # When intf is brought down, entry is removed
330 fib4_dump = self.vapi.ip_route_dump(0)
331 self.assertFalse(lo_if.is_ip4_entry_in_fib_dump(fib4_dump))
333 # Remove addr, bring up interface, re-add -> entry in FIB
337 fib4_dump = self.vapi.ip_route_dump(0)
338 self.assertTrue(lo_if.is_ip4_entry_in_fib_dump(fib4_dump))
341 class TestICMPEcho(VppTestCase):
342 """ ICMP Echo Test Case """
346 super(TestICMPEcho, cls).setUpClass()
349 def tearDownClass(cls):
350 super(TestICMPEcho, cls).tearDownClass()
353 super(TestICMPEcho, self).setUp()
355 # create 1 pg interface
356 self.create_pg_interfaces(range(1))
358 for i in self.pg_interfaces:
364 super(TestICMPEcho, self).tearDown()
365 for i in self.pg_interfaces:
369 def test_icmp_echo(self):
370 """ VPP replies to ICMP Echo Request
374 - Receive ICMP Echo Request message on pg0 interface.
375 - Check outgoing ICMP Echo Reply message on pg0 interface.
380 icmp_load = '\x0a' * 18
381 p_echo_request = (Ether(src=self.pg0.remote_mac,
382 dst=self.pg0.local_mac) /
383 IP(src=self.pg0.remote_ip4, dst=self.pg0.local_ip4) /
384 ICMP(id=icmp_id, seq=icmp_seq) /
387 self.pg0.add_stream(p_echo_request)
388 self.pg_enable_capture(self.pg_interfaces)
391 rx = self.pg0.get_capture(1)
397 self.assertEqual(ether.src, self.pg0.local_mac)
398 self.assertEqual(ether.dst, self.pg0.remote_mac)
400 self.assertEqual(ipv4.src, self.pg0.local_ip4)
401 self.assertEqual(ipv4.dst, self.pg0.remote_ip4)
403 self.assertEqual(icmptypes[icmp.type], "echo-reply")
404 self.assertEqual(icmp.id, icmp_id)
405 self.assertEqual(icmp.seq, icmp_seq)
406 self.assertEqual(icmp[Raw].load, icmp_load)
409 class TestIPv4FibCrud(VppTestCase):
410 """ FIB - add/update/delete - ip4 routes
418 ..note:: Python API is too slow to add many routes, needs replacement.
421 def config_fib_many_to_one(self, start_dest_addr, next_hop_addr,
425 :param start_dest_addr:
426 :param next_hop_addr:
428 :return list: added ips with 32 prefix
431 for i in range(count):
432 r = VppIpRoute(self, start_dest_addr % (i + start), 32,
433 [VppRoutePath(next_hop_addr, 0xffffffff)])
438 def unconfig_fib_many_to_one(self, start_dest_addr, next_hop_addr,
442 for i in range(count):
443 r = VppIpRoute(self, start_dest_addr % (i + start), 32,
444 [VppRoutePath(next_hop_addr, 0xffffffff)])
445 r.remove_vpp_config()
449 def create_stream(self, src_if, dst_if, routes, count):
452 for _ in range(count):
453 dst_addr = random.choice(routes).prefix.address
454 info = self.create_packet_info(src_if, dst_if)
455 payload = self.info_to_payload(info)
456 p = (Ether(dst=src_if.local_mac, src=src_if.remote_mac) /
457 IP(src=src_if.remote_ip4, dst=dst_addr) /
458 UDP(sport=1234, dport=1234) /
461 self.extend_packet(p, random.choice(self.pg_if_packet_sizes))
466 def _find_ip_match(self, find_in, pkt):
468 if self.payload_to_info(p[Raw]) == \
469 self.payload_to_info(pkt[Raw]):
470 if p[IP].src != pkt[IP].src:
472 if p[IP].dst != pkt[IP].dst:
474 if p[UDP].sport != pkt[UDP].sport:
476 if p[UDP].dport != pkt[UDP].dport:
481 def verify_capture(self, dst_interface, received_pkts, expected_pkts):
482 self.assertEqual(len(received_pkts), len(expected_pkts))
483 to_verify = list(expected_pkts)
484 for p in received_pkts:
485 self.assertEqual(p.src, dst_interface.local_mac)
486 self.assertEqual(p.dst, dst_interface.remote_mac)
487 x = self._find_ip_match(to_verify, p)
489 self.assertListEqual(to_verify, [])
491 def verify_route_dump(self, routes):
493 self.assertTrue(find_route(self, r.prefix.address, r.prefix.len))
495 def verify_not_in_route_dump(self, routes):
497 self.assertFalse(find_route(self, r.prefix.address, r.prefix.len))
502 #. Create and initialize 3 pg interfaces.
503 #. initialize class attributes configured_routes and deleted_routes
504 to store information between tests.
506 super(TestIPv4FibCrud, cls).setUpClass()
509 # create 3 pg interfaces
510 cls.create_pg_interfaces(range(3))
512 cls.interfaces = list(cls.pg_interfaces)
514 # setup all interfaces
515 for i in cls.interfaces:
520 cls.configured_routes = []
521 cls.deleted_routes = []
522 cls.pg_if_packet_sizes = [64, 512, 1518, 9018]
525 super(TestIPv4FibCrud, cls).tearDownClass()
529 def tearDownClass(cls):
530 super(TestIPv4FibCrud, cls).tearDownClass()
533 super(TestIPv4FibCrud, self).setUp()
534 self.reset_packet_infos()
536 self.configured_routes = []
537 self.deleted_routes = []
539 def test_1_add_routes(self):
540 """ Add 1k routes """
542 # add 100 routes check with traffic script.
543 self.configured_routes.extend(self.config_fib_many_to_one(
544 "10.0.0.%d", self.pg0.remote_ip4, 100))
546 self.verify_route_dump(self.configured_routes)
548 self.stream_1 = self.create_stream(
549 self.pg1, self.pg0, self.configured_routes, 100)
550 self.stream_2 = self.create_stream(
551 self.pg2, self.pg0, self.configured_routes, 100)
552 self.pg1.add_stream(self.stream_1)
553 self.pg2.add_stream(self.stream_2)
555 self.pg_enable_capture(self.pg_interfaces)
558 pkts = self.pg0.get_capture(len(self.stream_1) + len(self.stream_2))
559 self.verify_capture(self.pg0, pkts, self.stream_1 + self.stream_2)
561 def test_2_del_routes(self):
562 """ Delete 100 routes
564 - delete 10 routes check with traffic script.
566 # config 1M FIB entries
567 self.configured_routes.extend(self.config_fib_many_to_one(
568 "10.0.0.%d", self.pg0.remote_ip4, 100))
569 self.deleted_routes.extend(self.unconfig_fib_many_to_one(
570 "10.0.0.%d", self.pg0.remote_ip4, 10, start=10))
571 for x in self.deleted_routes:
572 self.configured_routes.remove(x)
574 self.verify_route_dump(self.configured_routes)
576 self.stream_1 = self.create_stream(
577 self.pg1, self.pg0, self.configured_routes, 100)
578 self.stream_2 = self.create_stream(
579 self.pg2, self.pg0, self.configured_routes, 100)
580 self.stream_3 = self.create_stream(
581 self.pg1, self.pg0, self.deleted_routes, 100)
582 self.stream_4 = self.create_stream(
583 self.pg2, self.pg0, self.deleted_routes, 100)
584 self.pg1.add_stream(self.stream_1 + self.stream_3)
585 self.pg2.add_stream(self.stream_2 + self.stream_4)
586 self.pg_enable_capture(self.pg_interfaces)
589 pkts = self.pg0.get_capture(len(self.stream_1) + len(self.stream_2))
590 self.verify_capture(self.pg0, pkts, self.stream_1 + self.stream_2)
592 def test_3_add_new_routes(self):
595 - re-add 5 routes check with traffic script.
596 - add 100 routes check with traffic script.
598 # config 1M FIB entries
599 self.configured_routes.extend(self.config_fib_many_to_one(
600 "10.0.0.%d", self.pg0.remote_ip4, 100))
601 self.deleted_routes.extend(self.unconfig_fib_many_to_one(
602 "10.0.0.%d", self.pg0.remote_ip4, 10, start=10))
603 for x in self.deleted_routes:
604 self.configured_routes.remove(x)
606 tmp = self.config_fib_many_to_one(
607 "10.0.0.%d", self.pg0.remote_ip4, 5, start=10)
608 self.configured_routes.extend(tmp)
610 self.deleted_routes.remove(x)
612 self.configured_routes.extend(self.config_fib_many_to_one(
613 "10.0.1.%d", self.pg0.remote_ip4, 100))
615 self.verify_route_dump(self.configured_routes)
617 self.stream_1 = self.create_stream(
618 self.pg1, self.pg0, self.configured_routes, 300)
619 self.stream_2 = self.create_stream(
620 self.pg2, self.pg0, self.configured_routes, 300)
621 self.stream_3 = self.create_stream(
622 self.pg1, self.pg0, self.deleted_routes, 100)
623 self.stream_4 = self.create_stream(
624 self.pg2, self.pg0, self.deleted_routes, 100)
626 self.pg1.add_stream(self.stream_1 + self.stream_3)
627 self.pg2.add_stream(self.stream_2 + self.stream_4)
628 self.pg_enable_capture(self.pg_interfaces)
631 pkts = self.pg0.get_capture(len(self.stream_1) + len(self.stream_2))
632 self.verify_capture(self.pg0, pkts, self.stream_1 + self.stream_2)
634 # delete 5 routes check with traffic script.
635 # add 100 routes check with traffic script.
636 self.deleted_routes.extend(self.unconfig_fib_many_to_one(
637 "10.0.0.%d", self.pg0.remote_ip4, 15))
638 self.deleted_routes.extend(self.unconfig_fib_many_to_one(
639 "10.0.0.%d", self.pg0.remote_ip4, 85))
640 self.deleted_routes.extend(self.unconfig_fib_many_to_one(
641 "10.0.1.%d", self.pg0.remote_ip4, 100))
642 self.verify_not_in_route_dump(self.deleted_routes)
645 class TestIPNull(VppTestCase):
646 """ IPv4 routes via NULL """
650 super(TestIPNull, cls).setUpClass()
653 def tearDownClass(cls):
654 super(TestIPNull, cls).tearDownClass()
657 super(TestIPNull, self).setUp()
659 # create 2 pg interfaces
660 self.create_pg_interfaces(range(2))
662 for i in self.pg_interfaces:
668 super(TestIPNull, self).tearDown()
669 for i in self.pg_interfaces:
673 def test_ip_null(self):
674 """ IP NULL route """
677 # A route via IP NULL that will reply with ICMP unreachables
679 ip_unreach = VppIpRoute(
680 self, "10.0.0.1", 32,
681 [VppRoutePath("0.0.0.0",
683 type=FibPathType.FIB_PATH_TYPE_ICMP_UNREACH)])
684 ip_unreach.add_vpp_config()
686 p_unreach = (Ether(src=self.pg0.remote_mac,
687 dst=self.pg0.local_mac) /
688 IP(src=self.pg0.remote_ip4, dst="10.0.0.1") /
689 UDP(sport=1234, dport=1234) /
691 self.pg0.add_stream(p_unreach)
692 self.pg_enable_capture(self.pg_interfaces)
695 rx = self.pg0.get_capture(1)
699 self.assertEqual(icmptypes[icmp.type], "dest-unreach")
700 self.assertEqual(icmpcodes[icmp.type][icmp.code], "host-unreachable")
701 self.assertEqual(icmp.src, self.pg0.remote_ip4)
702 self.assertEqual(icmp.dst, "10.0.0.1")
705 # ICMP replies are rate limited. so sit and spin.
710 # A route via IP NULL that will reply with ICMP prohibited
712 ip_prohibit = VppIpRoute(
713 self, "10.0.0.2", 32,
714 [VppRoutePath("0.0.0.0",
716 type=FibPathType.FIB_PATH_TYPE_ICMP_PROHIBIT)])
717 ip_prohibit.add_vpp_config()
719 p_prohibit = (Ether(src=self.pg0.remote_mac,
720 dst=self.pg0.local_mac) /
721 IP(src=self.pg0.remote_ip4, dst="10.0.0.2") /
722 UDP(sport=1234, dport=1234) /
725 self.pg0.add_stream(p_prohibit)
726 self.pg_enable_capture(self.pg_interfaces)
729 rx = self.pg0.get_capture(1)
734 self.assertEqual(icmptypes[icmp.type], "dest-unreach")
735 self.assertEqual(icmpcodes[icmp.type][icmp.code], "host-prohibited")
736 self.assertEqual(icmp.src, self.pg0.remote_ip4)
737 self.assertEqual(icmp.dst, "10.0.0.2")
739 def test_ip_drop(self):
740 """ IP Drop Routes """
742 p = (Ether(src=self.pg0.remote_mac,
743 dst=self.pg0.local_mac) /
744 IP(src=self.pg0.remote_ip4, dst="1.1.1.1") /
745 UDP(sport=1234, dport=1234) /
748 r1 = VppIpRoute(self, "1.1.1.0", 24,
749 [VppRoutePath(self.pg1.remote_ip4,
750 self.pg1.sw_if_index)])
753 rx = self.send_and_expect(self.pg0, p * NUM_PKTS, self.pg1)
756 # insert a more specific as a drop
758 r2 = VppIpRoute(self, "1.1.1.1", 32,
759 [VppRoutePath("0.0.0.0",
761 type=FibPathType.FIB_PATH_TYPE_DROP)])
764 self.send_and_assert_no_replies(self.pg0, p * NUM_PKTS, "Drop Route")
765 r2.remove_vpp_config()
766 rx = self.send_and_expect(self.pg0, p * NUM_PKTS, self.pg1)
769 class TestIPDisabled(VppTestCase):
770 """ IPv4 disabled """
774 super(TestIPDisabled, cls).setUpClass()
777 def tearDownClass(cls):
778 super(TestIPDisabled, cls).tearDownClass()
781 super(TestIPDisabled, self).setUp()
783 # create 2 pg interfaces
784 self.create_pg_interfaces(range(2))
788 self.pg0.config_ip4()
789 self.pg0.resolve_arp()
791 # PG 1 is not IP enabled
795 super(TestIPDisabled, self).tearDown()
796 for i in self.pg_interfaces:
800 def test_ip_disabled(self):
805 # one accepting interface, pg0, 2 forwarding interfaces
807 route_232_1_1_1 = VppIpMRoute(
811 MRouteEntryFlags.MFIB_ENTRY_FLAG_NONE,
812 [VppMRoutePath(self.pg1.sw_if_index,
813 MRouteItfFlags.MFIB_ITF_FLAG_ACCEPT),
814 VppMRoutePath(self.pg0.sw_if_index,
815 MRouteItfFlags.MFIB_ITF_FLAG_FORWARD)])
816 route_232_1_1_1.add_vpp_config()
818 pu = (Ether(src=self.pg1.remote_mac,
819 dst=self.pg1.local_mac) /
820 IP(src="10.10.10.10", dst=self.pg0.remote_ip4) /
821 UDP(sport=1234, dport=1234) /
823 pm = (Ether(src=self.pg1.remote_mac,
824 dst=self.pg1.local_mac) /
825 IP(src="10.10.10.10", dst="232.1.1.1") /
826 UDP(sport=1234, dport=1234) /
830 # PG1 does not forward IP traffic
832 self.send_and_assert_no_replies(self.pg1, pu, "IP disabled")
833 self.send_and_assert_no_replies(self.pg1, pm, "IP disabled")
838 self.pg1.config_ip4()
841 # Now we get packets through
843 self.pg1.add_stream(pu)
844 self.pg_enable_capture(self.pg_interfaces)
846 rx = self.pg0.get_capture(1)
848 self.pg1.add_stream(pm)
849 self.pg_enable_capture(self.pg_interfaces)
851 rx = self.pg0.get_capture(1)
856 self.pg1.unconfig_ip4()
859 # PG1 does not forward IP traffic
861 self.send_and_assert_no_replies(self.pg1, pu, "IP disabled")
862 self.send_and_assert_no_replies(self.pg1, pm, "IP disabled")
865 class TestIPSubNets(VppTestCase):
870 super(TestIPSubNets, cls).setUpClass()
873 def tearDownClass(cls):
874 super(TestIPSubNets, cls).tearDownClass()
877 super(TestIPSubNets, self).setUp()
879 # create a 2 pg interfaces
880 self.create_pg_interfaces(range(2))
882 # pg0 we will use to experiment
885 # pg1 is setup normally
887 self.pg1.config_ip4()
888 self.pg1.resolve_arp()
891 super(TestIPSubNets, self).tearDown()
892 for i in self.pg_interfaces:
895 def test_ip_sub_nets(self):
899 # Configure a covering route to forward so we know
900 # when we are dropping
902 cover_route = VppIpRoute(self, "10.0.0.0", 8,
903 [VppRoutePath(self.pg1.remote_ip4,
904 self.pg1.sw_if_index)])
905 cover_route.add_vpp_config()
907 p = (Ether(src=self.pg1.remote_mac,
908 dst=self.pg1.local_mac) /
909 IP(dst="10.10.10.10", src=self.pg0.local_ip4) /
910 UDP(sport=1234, dport=1234) /
913 self.pg1.add_stream(p)
914 self.pg_enable_capture(self.pg_interfaces)
916 rx = self.pg1.get_capture(1)
919 # Configure some non-/24 subnets on an IP interface
921 ip_addr_n = socket.inet_pton(socket.AF_INET, "10.10.10.10")
923 self.vapi.sw_interface_add_del_address(
924 sw_if_index=self.pg0.sw_if_index,
925 prefix=VppIpPrefix("10.10.10.10", 16).encode())
927 pn = (Ether(src=self.pg1.remote_mac,
928 dst=self.pg1.local_mac) /
929 IP(dst="10.10.0.0", src=self.pg0.local_ip4) /
930 UDP(sport=1234, dport=1234) /
932 pb = (Ether(src=self.pg1.remote_mac,
933 dst=self.pg1.local_mac) /
934 IP(dst="10.10.255.255", src=self.pg0.local_ip4) /
935 UDP(sport=1234, dport=1234) /
938 self.send_and_assert_no_replies(self.pg1, pn, "IP Network address")
939 self.send_and_assert_no_replies(self.pg1, pb, "IP Broadcast address")
941 # remove the sub-net and we are forwarding via the cover again
942 self.vapi.sw_interface_add_del_address(
943 sw_if_index=self.pg0.sw_if_index,
944 prefix=VppIpPrefix("10.10.10.10", 16).encode(), is_add=0)
946 self.pg1.add_stream(pn)
947 self.pg_enable_capture(self.pg_interfaces)
949 rx = self.pg1.get_capture(1)
950 self.pg1.add_stream(pb)
951 self.pg_enable_capture(self.pg_interfaces)
953 rx = self.pg1.get_capture(1)
956 # A /31 is a special case where the 'other-side' is an attached host
957 # packets to that peer generate ARP requests
959 ip_addr_n = socket.inet_pton(socket.AF_INET, "10.10.10.10")
961 self.vapi.sw_interface_add_del_address(
962 sw_if_index=self.pg0.sw_if_index,
963 prefix=VppIpPrefix("10.10.10.10", 31).encode())
965 pn = (Ether(src=self.pg1.remote_mac,
966 dst=self.pg1.local_mac) /
967 IP(dst="10.10.10.11", src=self.pg0.local_ip4) /
968 UDP(sport=1234, dport=1234) /
971 self.pg1.add_stream(pn)
972 self.pg_enable_capture(self.pg_interfaces)
974 rx = self.pg0.get_capture(1)
977 # remove the sub-net and we are forwarding via the cover again
978 self.vapi.sw_interface_add_del_address(
979 sw_if_index=self.pg0.sw_if_index,
980 prefix=VppIpPrefix("10.10.10.10", 31).encode(), is_add=0)
982 self.pg1.add_stream(pn)
983 self.pg_enable_capture(self.pg_interfaces)
985 rx = self.pg1.get_capture(1)
988 class TestIPLoadBalance(VppTestCase):
989 """ IPv4 Load-Balancing """
993 super(TestIPLoadBalance, cls).setUpClass()
996 def tearDownClass(cls):
997 super(TestIPLoadBalance, cls).tearDownClass()
1000 super(TestIPLoadBalance, self).setUp()
1002 self.create_pg_interfaces(range(5))
1003 mpls_tbl = VppMplsTable(self, 0)
1004 mpls_tbl.add_vpp_config()
1006 for i in self.pg_interfaces:
1013 for i in self.pg_interfaces:
1017 super(TestIPLoadBalance, self).tearDown()
1019 def send_and_expect_load_balancing(self, input, pkts, outputs):
1020 input.add_stream(pkts)
1021 self.pg_enable_capture(self.pg_interfaces)
1025 rx = oo._get_capture(1)
1026 self.assertNotEqual(0, len(rx))
1031 def send_and_expect_one_itf(self, input, pkts, itf):
1032 input.add_stream(pkts)
1033 self.pg_enable_capture(self.pg_interfaces)
1035 rx = itf.get_capture(len(pkts))
1037 def test_ip_load_balance(self):
1038 """ IP Load-Balancing """
1041 # An array of packets that differ only in the destination port
1047 # An array of packets that differ only in the source address
1052 for ii in range(NUM_PKTS):
1053 port_ip_hdr = (IP(dst="10.0.0.1", src="20.0.0.1") /
1054 UDP(sport=1234, dport=1234 + ii) /
1056 port_ip_pkts.append((Ether(src=self.pg0.remote_mac,
1057 dst=self.pg0.local_mac) /
1059 port_mpls_pkts.append((Ether(src=self.pg0.remote_mac,
1060 dst=self.pg0.local_mac) /
1061 MPLS(label=66, ttl=2) /
1064 src_ip_hdr = (IP(dst="10.0.0.1", src="20.0.0.%d" % ii) /
1065 UDP(sport=1234, dport=1234) /
1067 src_ip_pkts.append((Ether(src=self.pg0.remote_mac,
1068 dst=self.pg0.local_mac) /
1070 src_mpls_pkts.append((Ether(src=self.pg0.remote_mac,
1071 dst=self.pg0.local_mac) /
1072 MPLS(label=66, ttl=2) /
1075 route_10_0_0_1 = VppIpRoute(self, "10.0.0.1", 32,
1076 [VppRoutePath(self.pg1.remote_ip4,
1077 self.pg1.sw_if_index),
1078 VppRoutePath(self.pg2.remote_ip4,
1079 self.pg2.sw_if_index)])
1080 route_10_0_0_1.add_vpp_config()
1082 binding = VppMplsIpBind(self, 66, "10.0.0.1", 32)
1083 binding.add_vpp_config()
1086 # inject the packet on pg0 - expect load-balancing across the 2 paths
1087 # - since the default hash config is to use IP src,dst and port
1089 # We are not going to ensure equal amounts of packets across each link,
1090 # since the hash algorithm is statistical and therefore this can never
1091 # be guaranteed. But with 64 different packets we do expect some
1092 # balancing. So instead just ensure there is traffic on each link.
1094 self.send_and_expect_load_balancing(self.pg0, port_ip_pkts,
1095 [self.pg1, self.pg2])
1096 self.send_and_expect_load_balancing(self.pg0, src_ip_pkts,
1097 [self.pg1, self.pg2])
1098 self.send_and_expect_load_balancing(self.pg0, port_mpls_pkts,
1099 [self.pg1, self.pg2])
1100 self.send_and_expect_load_balancing(self.pg0, src_mpls_pkts,
1101 [self.pg1, self.pg2])
1104 # change the flow hash config so it's only IP src,dst
1105 # - now only the stream with differing source address will
1108 self.vapi.set_ip_flow_hash(vrf_id=0, src=1, dst=1, sport=0, dport=0)
1110 self.send_and_expect_load_balancing(self.pg0, src_ip_pkts,
1111 [self.pg1, self.pg2])
1112 self.send_and_expect_load_balancing(self.pg0, src_mpls_pkts,
1113 [self.pg1, self.pg2])
1115 self.send_and_expect_one_itf(self.pg0, port_ip_pkts, self.pg2)
1118 # change the flow hash config back to defaults
1120 self.vapi.set_ip_flow_hash(vrf_id=0, src=1, dst=1, sport=1, dport=1)
1123 # Recursive prefixes
1124 # - testing that 2 stages of load-balancing occurs and there is no
1125 # polarisation (i.e. only 2 of 4 paths are used)
1130 for ii in range(257):
1131 port_pkts.append((Ether(src=self.pg0.remote_mac,
1132 dst=self.pg0.local_mac) /
1133 IP(dst="1.1.1.1", src="20.0.0.1") /
1134 UDP(sport=1234, dport=1234 + ii) /
1136 src_pkts.append((Ether(src=self.pg0.remote_mac,
1137 dst=self.pg0.local_mac) /
1138 IP(dst="1.1.1.1", src="20.0.0.%d" % ii) /
1139 UDP(sport=1234, dport=1234) /
1142 route_10_0_0_2 = VppIpRoute(self, "10.0.0.2", 32,
1143 [VppRoutePath(self.pg3.remote_ip4,
1144 self.pg3.sw_if_index),
1145 VppRoutePath(self.pg4.remote_ip4,
1146 self.pg4.sw_if_index)])
1147 route_10_0_0_2.add_vpp_config()
1149 route_1_1_1_1 = VppIpRoute(self, "1.1.1.1", 32,
1150 [VppRoutePath("10.0.0.2", 0xffffffff),
1151 VppRoutePath("10.0.0.1", 0xffffffff)])
1152 route_1_1_1_1.add_vpp_config()
1155 # inject the packet on pg0 - expect load-balancing across all 4 paths
1157 self.vapi.cli("clear trace")
1158 self.send_and_expect_load_balancing(self.pg0, port_pkts,
1159 [self.pg1, self.pg2,
1160 self.pg3, self.pg4])
1161 self.send_and_expect_load_balancing(self.pg0, src_pkts,
1162 [self.pg1, self.pg2,
1163 self.pg3, self.pg4])
1166 # bring down pg1 expect LB to adjust to use only those that are pu
1168 self.pg1.link_down()
1170 rx = self.send_and_expect_load_balancing(self.pg0, src_pkts,
1171 [self.pg2, self.pg3,
1173 self.assertEqual(len(src_pkts), len(rx))
1176 # bring down pg2 expect LB to adjust to use only those that are pu
1178 self.pg2.link_down()
1180 rx = self.send_and_expect_load_balancing(self.pg0, src_pkts,
1181 [self.pg3, self.pg4])
1182 self.assertEqual(len(src_pkts), len(rx))
1185 # bring the links back up - expect LB over all again
1190 rx = self.send_and_expect_load_balancing(self.pg0, src_pkts,
1191 [self.pg1, self.pg2,
1192 self.pg3, self.pg4])
1193 self.assertEqual(len(src_pkts), len(rx))
1196 # The same link-up/down but this time admin state
1198 self.pg1.admin_down()
1199 self.pg2.admin_down()
1200 rx = self.send_and_expect_load_balancing(self.pg0, src_pkts,
1201 [self.pg3, self.pg4])
1202 self.assertEqual(len(src_pkts), len(rx))
1205 self.pg1.resolve_arp()
1206 self.pg2.resolve_arp()
1207 rx = self.send_and_expect_load_balancing(self.pg0, src_pkts,
1208 [self.pg1, self.pg2,
1209 self.pg3, self.pg4])
1210 self.assertEqual(len(src_pkts), len(rx))
1213 # Recursive prefixes
1214 # - testing that 2 stages of load-balancing, no choices
1218 for ii in range(257):
1219 port_pkts.append((Ether(src=self.pg0.remote_mac,
1220 dst=self.pg0.local_mac) /
1221 IP(dst="1.1.1.2", src="20.0.0.2") /
1222 UDP(sport=1234, dport=1234 + ii) /
1225 route_10_0_0_3 = VppIpRoute(self, "10.0.0.3", 32,
1226 [VppRoutePath(self.pg3.remote_ip4,
1227 self.pg3.sw_if_index)])
1228 route_10_0_0_3.add_vpp_config()
1230 route_1_1_1_2 = VppIpRoute(self, "1.1.1.2", 32,
1231 [VppRoutePath("10.0.0.3", 0xffffffff)])
1232 route_1_1_1_2.add_vpp_config()
1235 # inject the packet on pg0 - rx only on via routes output interface
1237 self.vapi.cli("clear trace")
1238 self.send_and_expect_one_itf(self.pg0, port_pkts, self.pg3)
1241 # Add a LB route in the presence of a down link - expect no
1242 # packets over the down link
1244 self.pg3.link_down()
1246 route_10_0_0_3 = VppIpRoute(self, "10.0.0.3", 32,
1247 [VppRoutePath(self.pg3.remote_ip4,
1248 self.pg3.sw_if_index),
1249 VppRoutePath(self.pg4.remote_ip4,
1250 self.pg4.sw_if_index)])
1251 route_10_0_0_3.add_vpp_config()
1254 for ii in range(257):
1255 port_pkts.append(Ether(src=self.pg0.remote_mac,
1256 dst=self.pg0.local_mac) /
1257 IP(dst="10.0.0.3", src="20.0.0.2") /
1258 UDP(sport=1234, dport=1234 + ii) /
1261 self.send_and_expect_one_itf(self.pg0, port_pkts, self.pg4)
1263 # bring the link back up
1266 rx = self.send_and_expect_load_balancing(self.pg0, port_pkts,
1267 [self.pg3, self.pg4])
1268 self.assertEqual(len(src_pkts), len(rx))
1271 class TestIPVlan0(VppTestCase):
1275 def setUpClass(cls):
1276 super(TestIPVlan0, cls).setUpClass()
1279 def tearDownClass(cls):
1280 super(TestIPVlan0, cls).tearDownClass()
1283 super(TestIPVlan0, self).setUp()
1285 self.create_pg_interfaces(range(2))
1286 mpls_tbl = VppMplsTable(self, 0)
1287 mpls_tbl.add_vpp_config()
1289 for i in self.pg_interfaces:
1296 for i in self.pg_interfaces:
1300 super(TestIPVlan0, self).tearDown()
1302 def test_ip_vlan_0(self):
1305 pkts = (Ether(src=self.pg0.remote_mac,
1306 dst=self.pg0.local_mac) /
1308 IP(dst=self.pg1.remote_ip4,
1309 src=self.pg0.remote_ip4) /
1310 UDP(sport=1234, dport=1234) /
1311 Raw('\xa5' * 100)) * NUM_PKTS
1314 # Expect that packets sent on VLAN-0 are forwarded on the
1317 self.send_and_expect(self.pg0, pkts, self.pg1)
1320 class TestIPPunt(VppTestCase):
1321 """ IPv4 Punt Police/Redirect """
1324 def setUpClass(cls):
1325 super(TestIPPunt, cls).setUpClass()
1328 def tearDownClass(cls):
1329 super(TestIPPunt, cls).tearDownClass()
1332 super(TestIPPunt, self).setUp()
1334 self.create_pg_interfaces(range(4))
1336 for i in self.pg_interfaces:
1342 super(TestIPPunt, self).tearDown()
1343 for i in self.pg_interfaces:
1347 def test_ip_punt(self):
1348 """ IP punt police and redirect """
1350 # use UDP packet that have a port we need to explicitly
1351 # register to get punted.
1352 pt_l4 = VppEnum.vl_api_punt_type_t.PUNT_API_TYPE_L4
1353 af_ip4 = VppEnum.vl_api_address_family_t.ADDRESS_IP4
1354 udp_proto = VppEnum.vl_api_ip_proto_t.IP_API_PROTO_UDP
1360 'protocol': udp_proto,
1366 self.vapi.set_punt(is_add=1, punt=punt_udp)
1368 p = (Ether(src=self.pg0.remote_mac,
1369 dst=self.pg0.local_mac) /
1370 IP(src=self.pg0.remote_ip4, dst=self.pg0.local_ip4) /
1371 UDP(sport=1234, dport=1234) /
1377 # Configure a punt redirect via pg1.
1379 nh_addr = self.pg1.remote_ip4
1380 self.vapi.ip_punt_redirect(self.pg0.sw_if_index,
1381 self.pg1.sw_if_index,
1384 self.send_and_expect(self.pg0, pkts, self.pg1)
1389 policer = self.vapi.policer_add_del(b"ip4-punt", 400, 0, 10, 0,
1391 self.vapi.ip_punt_police(policer.policer_index)
1393 self.vapi.cli("clear trace")
1394 self.pg0.add_stream(pkts)
1395 self.pg_enable_capture(self.pg_interfaces)
1399 # the number of packet received should be greater than 0,
1400 # but not equal to the number sent, since some were policed
1402 rx = self.pg1._get_capture(1)
1403 self.assertGreater(len(rx), 0)
1404 self.assertLess(len(rx), len(pkts))
1407 # remove the policer. back to full rx
1409 self.vapi.ip_punt_police(policer.policer_index, is_add=0)
1410 self.vapi.policer_add_del(b"ip4-punt", 400, 0, 10, 0,
1411 rate_type=1, is_add=0)
1412 self.send_and_expect(self.pg0, pkts, self.pg1)
1415 # remove the redirect. expect full drop.
1417 self.vapi.ip_punt_redirect(self.pg0.sw_if_index,
1418 self.pg1.sw_if_index,
1421 self.send_and_assert_no_replies(self.pg0, pkts,
1422 "IP no punt config")
1425 # Add a redirect that is not input port selective
1427 self.vapi.ip_punt_redirect(0xffffffff,
1428 self.pg1.sw_if_index,
1430 self.send_and_expect(self.pg0, pkts, self.pg1)
1432 self.vapi.ip_punt_redirect(0xffffffff,
1433 self.pg1.sw_if_index,
1437 def test_ip_punt_dump(self):
1438 """ IP4 punt redirect dump"""
1441 # Configure a punt redirects
1443 nh_address = self.pg3.remote_ip4
1444 self.vapi.ip_punt_redirect(self.pg0.sw_if_index,
1445 self.pg3.sw_if_index,
1447 self.vapi.ip_punt_redirect(self.pg1.sw_if_index,
1448 self.pg3.sw_if_index,
1450 self.vapi.ip_punt_redirect(self.pg2.sw_if_index,
1451 self.pg3.sw_if_index,
1455 # Dump pg0 punt redirects
1457 punts = self.vapi.ip_punt_redirect_dump(self.pg0.sw_if_index)
1459 self.assertEqual(p.punt.rx_sw_if_index, self.pg0.sw_if_index)
1462 # Dump punt redirects for all interfaces
1464 punts = self.vapi.ip_punt_redirect_dump(0xffffffff)
1465 self.assertEqual(len(punts), 3)
1467 self.assertEqual(p.punt.tx_sw_if_index, self.pg3.sw_if_index)
1468 self.assertNotEqual(punts[1].punt.nh, self.pg3.remote_ip4)
1469 self.assertEqual(str(punts[2].punt.nh), '0.0.0.0')
1472 class TestIPDeag(VppTestCase):
1473 """ IPv4 Deaggregate Routes """
1476 def setUpClass(cls):
1477 super(TestIPDeag, cls).setUpClass()
1480 def tearDownClass(cls):
1481 super(TestIPDeag, cls).tearDownClass()
1484 super(TestIPDeag, self).setUp()
1486 self.create_pg_interfaces(range(3))
1488 for i in self.pg_interfaces:
1494 super(TestIPDeag, self).tearDown()
1495 for i in self.pg_interfaces:
1499 def test_ip_deag(self):
1500 """ IP Deag Routes """
1503 # Create a table to be used for:
1504 # 1 - another destination address lookup
1505 # 2 - a source address lookup
1507 table_dst = VppIpTable(self, 1)
1508 table_src = VppIpTable(self, 2)
1509 table_dst.add_vpp_config()
1510 table_src.add_vpp_config()
1513 # Add a route in the default table to point to a deag/
1514 # second lookup in each of these tables
1516 route_to_dst = VppIpRoute(self, "1.1.1.1", 32,
1517 [VppRoutePath("0.0.0.0",
1520 route_to_src = VppIpRoute(
1521 self, "1.1.1.2", 32,
1522 [VppRoutePath("0.0.0.0",
1525 type=FibPathType.FIB_PATH_TYPE_SOURCE_LOOKUP)])
1526 route_to_dst.add_vpp_config()
1527 route_to_src.add_vpp_config()
1530 # packets to these destination are dropped, since they'll
1531 # hit the respective default routes in the second table
1533 p_dst = (Ether(src=self.pg0.remote_mac,
1534 dst=self.pg0.local_mac) /
1535 IP(src="5.5.5.5", dst="1.1.1.1") /
1536 TCP(sport=1234, dport=1234) /
1538 p_src = (Ether(src=self.pg0.remote_mac,
1539 dst=self.pg0.local_mac) /
1540 IP(src="2.2.2.2", dst="1.1.1.2") /
1541 TCP(sport=1234, dport=1234) /
1543 pkts_dst = p_dst * 257
1544 pkts_src = p_src * 257
1546 self.send_and_assert_no_replies(self.pg0, pkts_dst,
1548 self.send_and_assert_no_replies(self.pg0, pkts_src,
1552 # add a route in the dst table to forward via pg1
1554 route_in_dst = VppIpRoute(self, "1.1.1.1", 32,
1555 [VppRoutePath(self.pg1.remote_ip4,
1556 self.pg1.sw_if_index)],
1558 route_in_dst.add_vpp_config()
1560 self.send_and_expect(self.pg0, pkts_dst, self.pg1)
1563 # add a route in the src table to forward via pg2
1565 route_in_src = VppIpRoute(self, "2.2.2.2", 32,
1566 [VppRoutePath(self.pg2.remote_ip4,
1567 self.pg2.sw_if_index)],
1569 route_in_src.add_vpp_config()
1570 self.send_and_expect(self.pg0, pkts_src, self.pg2)
1573 # loop in the lookup DP
1575 route_loop = VppIpRoute(self, "2.2.2.3", 32,
1576 [VppRoutePath("0.0.0.0",
1579 route_loop.add_vpp_config()
1581 p_l = (Ether(src=self.pg0.remote_mac,
1582 dst=self.pg0.local_mac) /
1583 IP(src="2.2.2.4", dst="2.2.2.3") /
1584 TCP(sport=1234, dport=1234) /
1587 self.send_and_assert_no_replies(self.pg0, p_l * 257,
1591 class TestIPInput(VppTestCase):
1592 """ IPv4 Input Exceptions """
1595 def setUpClass(cls):
1596 super(TestIPInput, cls).setUpClass()
1599 def tearDownClass(cls):
1600 super(TestIPInput, cls).tearDownClass()
1603 super(TestIPInput, self).setUp()
1605 self.create_pg_interfaces(range(2))
1607 for i in self.pg_interfaces:
1613 super(TestIPInput, self).tearDown()
1614 for i in self.pg_interfaces:
1618 def test_ip_input(self):
1619 """ IP Input Exceptions """
1621 # i can't find a way in scapy to construct an IP packet
1622 # with a length less than the IP header length
1625 # Packet too short - this is forwarded
1627 p_short = (Ether(src=self.pg0.remote_mac,
1628 dst=self.pg0.local_mac) /
1629 IP(src=self.pg0.remote_ip4,
1630 dst=self.pg1.remote_ip4,
1632 UDP(sport=1234, dport=1234) /
1635 rx = self.send_and_expect(self.pg0, p_short * NUM_PKTS, self.pg1)
1638 # Packet too long - this is dropped
1640 p_long = (Ether(src=self.pg0.remote_mac,
1641 dst=self.pg0.local_mac) /
1642 IP(src=self.pg0.remote_ip4,
1643 dst=self.pg1.remote_ip4,
1645 UDP(sport=1234, dport=1234) /
1648 rx = self.send_and_assert_no_replies(self.pg0, p_long * NUM_PKTS,
1652 # bad chksum - this is dropped
1654 p_chksum = (Ether(src=self.pg0.remote_mac,
1655 dst=self.pg0.local_mac) /
1656 IP(src=self.pg0.remote_ip4,
1657 dst=self.pg1.remote_ip4,
1659 UDP(sport=1234, dport=1234) /
1662 rx = self.send_and_assert_no_replies(self.pg0, p_chksum * NUM_PKTS,
1666 # bad version - this is dropped
1668 p_ver = (Ether(src=self.pg0.remote_mac,
1669 dst=self.pg0.local_mac) /
1670 IP(src=self.pg0.remote_ip4,
1671 dst=self.pg1.remote_ip4,
1673 UDP(sport=1234, dport=1234) /
1676 rx = self.send_and_assert_no_replies(self.pg0, p_ver * NUM_PKTS,
1680 # fragment offset 1 - this is dropped
1682 p_frag = (Ether(src=self.pg0.remote_mac,
1683 dst=self.pg0.local_mac) /
1684 IP(src=self.pg0.remote_ip4,
1685 dst=self.pg1.remote_ip4,
1687 UDP(sport=1234, dport=1234) /
1690 rx = self.send_and_assert_no_replies(self.pg0, p_frag * NUM_PKTS,
1694 # TTL expired packet
1696 p_ttl = (Ether(src=self.pg0.remote_mac,
1697 dst=self.pg0.local_mac) /
1698 IP(src=self.pg0.remote_ip4,
1699 dst=self.pg1.remote_ip4,
1701 UDP(sport=1234, dport=1234) /
1704 rx = self.send_and_expect(self.pg0, p_ttl * NUM_PKTS, self.pg0)
1709 self.assertEqual(icmptypes[icmp.type], "time-exceeded")
1710 self.assertEqual(icmpcodes[icmp.type][icmp.code],
1711 "ttl-zero-during-transit")
1712 self.assertEqual(icmp.src, self.pg0.remote_ip4)
1713 self.assertEqual(icmp.dst, self.pg1.remote_ip4)
1718 p_mtu = (Ether(src=self.pg0.remote_mac,
1719 dst=self.pg0.local_mac) /
1720 IP(src=self.pg0.remote_ip4,
1721 dst=self.pg1.remote_ip4,
1722 ttl=10, flags='DF') /
1723 UDP(sport=1234, dport=1234) /
1726 self.vapi.sw_interface_set_mtu(self.pg1.sw_if_index, [1500, 0, 0, 0])
1728 rx = self.send_and_expect(self.pg0, p_mtu * NUM_PKTS, self.pg0)
1732 self.assertEqual(icmptypes[icmp.type], "dest-unreach")
1733 self.assertEqual(icmpcodes[icmp.type][icmp.code],
1734 "fragmentation-needed")
1735 self.assertEqual(icmp.src, self.pg0.remote_ip4)
1736 self.assertEqual(icmp.dst, self.pg1.remote_ip4)
1738 self.vapi.sw_interface_set_mtu(self.pg1.sw_if_index, [2500, 0, 0, 0])
1739 rx = self.send_and_expect(self.pg0, p_mtu * NUM_PKTS, self.pg1)
1741 # Reset MTU for subsequent tests
1742 self.vapi.sw_interface_set_mtu(self.pg1.sw_if_index, [9000, 0, 0, 0])
1745 # source address 0.0.0.0 and 25.255.255.255 and for-us
1747 p_s0 = (Ether(src=self.pg0.remote_mac,
1748 dst=self.pg0.local_mac) /
1750 dst=self.pg0.local_ip4) /
1752 Raw(load='\x0a' * 18))
1753 rx = self.send_and_assert_no_replies(self.pg0, p_s0 * 17)
1755 p_s0 = (Ether(src=self.pg0.remote_mac,
1756 dst=self.pg0.local_mac) /
1757 IP(src="255.255.255.255",
1758 dst=self.pg0.local_ip4) /
1760 Raw(load='\x0a' * 18))
1761 rx = self.send_and_assert_no_replies(self.pg0, p_s0 * 17)
1764 class TestIPDirectedBroadcast(VppTestCase):
1765 """ IPv4 Directed Broadcast """
1768 def setUpClass(cls):
1769 super(TestIPDirectedBroadcast, cls).setUpClass()
1772 def tearDownClass(cls):
1773 super(TestIPDirectedBroadcast, cls).tearDownClass()
1776 super(TestIPDirectedBroadcast, self).setUp()
1778 self.create_pg_interfaces(range(2))
1780 for i in self.pg_interfaces:
1784 super(TestIPDirectedBroadcast, self).tearDown()
1785 for i in self.pg_interfaces:
1788 def test_ip_input(self):
1789 """ IP Directed Broadcast """
1792 # set the directed broadcast on pg0 first, then config IP4 addresses
1793 # for pg1 directed broadcast is always disabled
1794 self.vapi.sw_interface_set_ip_directed_broadcast(
1795 self.pg0.sw_if_index, 1)
1797 p0 = (Ether(src=self.pg1.remote_mac,
1798 dst=self.pg1.local_mac) /
1800 dst=self.pg0._local_ip4_bcast) /
1801 UDP(sport=1234, dport=1234) /
1803 p1 = (Ether(src=self.pg0.remote_mac,
1804 dst=self.pg0.local_mac) /
1806 dst=self.pg1._local_ip4_bcast) /
1807 UDP(sport=1234, dport=1234) /
1810 self.pg0.config_ip4()
1811 self.pg0.resolve_arp()
1812 self.pg1.config_ip4()
1813 self.pg1.resolve_arp()
1816 # test packet is L2 broadcast
1818 rx = self.send_and_expect(self.pg1, p0 * NUM_PKTS, self.pg0)
1819 self.assertTrue(rx[0][Ether].dst, "ff:ff:ff:ff:ff:ff")
1821 self.send_and_assert_no_replies(self.pg0, p1 * NUM_PKTS,
1822 "directed broadcast disabled")
1825 # toggle directed broadcast on pg0
1827 self.vapi.sw_interface_set_ip_directed_broadcast(
1828 self.pg0.sw_if_index, 0)
1829 self.send_and_assert_no_replies(self.pg1, p0 * NUM_PKTS,
1830 "directed broadcast disabled")
1832 self.vapi.sw_interface_set_ip_directed_broadcast(
1833 self.pg0.sw_if_index, 1)
1834 rx = self.send_and_expect(self.pg1, p0 * NUM_PKTS, self.pg0)
1836 self.pg0.unconfig_ip4()
1837 self.pg1.unconfig_ip4()
1840 class TestIPLPM(VppTestCase):
1841 """ IPv4 longest Prefix Match """
1844 def setUpClass(cls):
1845 super(TestIPLPM, cls).setUpClass()
1848 def tearDownClass(cls):
1849 super(TestIPLPM, cls).tearDownClass()
1852 super(TestIPLPM, self).setUp()
1854 self.create_pg_interfaces(range(4))
1856 for i in self.pg_interfaces:
1862 super(TestIPLPM, self).tearDown()
1863 for i in self.pg_interfaces:
1867 def test_ip_lpm(self):
1868 """ IP longest Prefix Match """
1870 s_24 = VppIpRoute(self, "10.1.2.0", 24,
1871 [VppRoutePath(self.pg1.remote_ip4,
1872 self.pg1.sw_if_index)])
1873 s_24.add_vpp_config()
1874 s_8 = VppIpRoute(self, "10.0.0.0", 8,
1875 [VppRoutePath(self.pg2.remote_ip4,
1876 self.pg2.sw_if_index)])
1877 s_8.add_vpp_config()
1879 p_8 = (Ether(src=self.pg0.remote_mac,
1880 dst=self.pg0.local_mac) /
1883 UDP(sport=1234, dport=1234) /
1885 p_24 = (Ether(src=self.pg0.remote_mac,
1886 dst=self.pg0.local_mac) /
1889 UDP(sport=1234, dport=1234) /
1892 self.logger.info(self.vapi.cli("sh ip fib mtrie"))
1893 rx = self.send_and_expect(self.pg0, p_8 * NUM_PKTS, self.pg2)
1894 rx = self.send_and_expect(self.pg0, p_24 * NUM_PKTS, self.pg1)
1897 class TestIPv4Frag(VppTestCase):
1898 """ IPv4 fragmentation """
1901 def setUpClass(cls):
1902 super(TestIPv4Frag, cls).setUpClass()
1904 cls.create_pg_interfaces([0, 1])
1905 cls.src_if = cls.pg0
1906 cls.dst_if = cls.pg1
1908 # setup all interfaces
1909 for i in cls.pg_interfaces:
1915 def tearDownClass(cls):
1916 super(TestIPv4Frag, cls).tearDownClass()
1918 def test_frag_large_packets(self):
1919 """ Fragmentation of large packets """
1921 p = (Ether(dst=self.src_if.local_mac, src=self.src_if.remote_mac) /
1922 IP(src=self.src_if.remote_ip4, dst=self.dst_if.remote_ip4) /
1923 UDP(sport=1234, dport=5678) / Raw())
1924 self.extend_packet(p, 6000, "abcde")
1925 saved_payload = p[Raw].load
1927 # Force fragmentation by setting MTU of output interface
1928 # lower than packet size
1929 self.vapi.sw_interface_set_mtu(self.dst_if.sw_if_index,
1932 self.pg_enable_capture()
1933 self.src_if.add_stream(p)
1936 # Expecting 3 fragments because size of created fragments currently
1937 # cannot be larger then VPP buffer size (which is 2048)
1938 packets = self.dst_if.get_capture(3)
1940 # Assume VPP sends the fragments in order
1943 payload_offset = p.frag * 8
1944 if payload_offset > 0:
1945 payload_offset -= 8 # UDP header is not in payload
1946 self.assert_equal(payload_offset, len(payload))
1947 payload += p[Raw].load
1948 self.assert_equal(payload, saved_payload, "payload")
1951 if __name__ == '__main__':
1952 unittest.main(testRunner=VppTestRunner)