7 from scapy.packet import Raw
8 from scapy.layers.l2 import Ether
9 from scapy.layers.inet import IP, UDP
10 from scapy.layers.inet6 import IPv6
12 from framework import VppTestCase, VppTestRunner
13 from vpp_sub_interface import VppP2PSubint
14 from vpp_ip import DpoProto
15 from vpp_ip_route import VppIpRoute, VppRoutePath
16 from vpp_papi import mac_pton
19 class P2PEthernetAPI(VppTestCase):
20 """P2P Ethernet tests"""
26 super(P2PEthernetAPI, cls).setUpClass()
28 # Create pg interfaces
29 cls.create_pg_interfaces(range(4))
31 # Set up all interfaces
32 for i in cls.pg_interfaces:
36 def tearDownClass(cls):
37 super(P2PEthernetAPI, cls).tearDownClass()
39 def create_p2p_ethernet(self, parent_if, sub_id, remote_mac):
40 p2p = VppP2PSubint(self, parent_if, sub_id, mac_pton(remote_mac))
41 self.p2p_sub_ifs.append(p2p)
43 def delete_p2p_ethernet(self, parent_if, remote_mac):
44 self.vapi.p2p_ethernet_del(parent_if.sw_if_index, mac_pton(remote_mac))
47 """delete/create p2p subif"""
48 self.logger.info("FFP_TEST_START_0000")
50 self.create_p2p_ethernet(self.pg0, 1, "de:ad:00:00:00:01")
51 self.create_p2p_ethernet(self.pg0, 2, "de:ad:00:00:00:02")
52 intfs = self.vapi.cli("show interface")
54 self.assertIn("pg0.1", intfs)
55 self.assertIn("pg0.2", intfs)
56 self.assertNotIn("pg0.5", intfs)
59 self.create_p2p_ethernet(self.pg0, 5, "de:ad:00:00:00:ff")
60 intfs = self.vapi.cli("show interface")
61 self.assertIn("pg0.5", intfs)
63 self.delete_p2p_ethernet(self.pg0, "de:ad:00:00:00:ff")
65 intfs = self.vapi.cli("show interface")
67 self.assertIn("pg0.1", intfs)
68 self.assertIn("pg0.2", intfs)
69 self.assertNotIn("pg0.5", intfs)
71 self.logger.info("FFP_TEST_FINISH_0000")
73 def test_p2p_subif_creation_1k(self):
74 """create 1k of p2p subifs"""
75 self.logger.info("FFP_TEST_START_0001")
79 mac = int("dead00000000", 16)
81 for i in range(1, clients + 1):
83 macs.append(":".join(re.findall("..", "{:02x}".format(mac + i))))
84 self.vapi.p2p_ethernet_add(
85 self.pg2.sw_if_index, mac_pton(macs[i - 1]), i
88 self.logger.info("Failed to create subif %d %s" % (i, macs[i - 1]))
91 intfs = self.vapi.cli("show interface").split("\n")
94 if intf.startswith("pg2."):
96 self.assertEqual(count, clients)
98 self.logger.info("FFP_TEST_FINISH_0001")
101 class P2PEthernetIPV6(VppTestCase):
102 """P2P Ethernet IPv6 tests"""
109 super(P2PEthernetIPV6, cls).setUpClass()
111 # Create pg interfaces
112 cls.create_pg_interfaces(range(3))
115 cls.pg_if_packet_sizes = [64, 512, 1518, 9018]
117 # Set up all interfaces
118 for i in cls.pg_interfaces:
121 cls.pg0.generate_remote_hosts(3)
122 cls.pg0.configure_ipv6_neighbors()
125 cls.pg1.generate_remote_hosts(3)
126 cls.pg1.configure_ipv6_neighbors()
127 cls.pg1.disable_ipv6_ra()
130 def tearDownClass(cls):
131 super(P2PEthernetIPV6, cls).tearDownClass()
134 super(P2PEthernetIPV6, self).setUp()
135 for p in self.packets:
136 self.packets.remove(p)
137 self.p2p_sub_ifs.append(
138 self.create_p2p_ethernet(self.pg0, 1, self.pg0._remote_hosts[0].mac)
140 self.p2p_sub_ifs.append(
141 self.create_p2p_ethernet(self.pg0, 2, self.pg0._remote_hosts[1].mac)
143 self.vapi.cli("trace add p2p-ethernet-input 50")
146 while len(self.p2p_sub_ifs):
147 p2p = self.p2p_sub_ifs.pop()
148 self.delete_p2p_ethernet(p2p)
150 super(P2PEthernetIPV6, self).tearDown()
152 def create_p2p_ethernet(self, parent_if, sub_id, remote_mac):
153 p2p = VppP2PSubint(self, parent_if, sub_id, mac_pton(remote_mac))
156 p2p.disable_ipv6_ra()
159 def delete_p2p_ethernet(self, p2p):
162 self.vapi.p2p_ethernet_del(p2p.parent.sw_if_index, p2p.p2p_remote_mac)
165 self, src_mac=None, dst_mac=None, src_ip=None, dst_ip=None, size=None
169 pkt_size = random.choice(self.pg_if_packet_sizes)
170 p = Ether(src=src_mac, dst=dst_mac)
171 p /= IPv6(src=src_ip, dst=dst_ip)
172 p /= UDP(sport=1234, dport=4321) / Raw(b"\xa5" * 20)
173 self.extend_packet(p, pkt_size)
176 def send_packets(self, src_if=None, dst_if=None, packets=None, count=None):
177 self.pg_enable_capture([dst_if])
179 packets = self.packets
180 src_if.add_stream(packets)
184 return dst_if.get_capture(count)
186 def test_no_p2p_subif(self):
187 """standard routing without p2p subinterfaces"""
188 self.logger.info("FFP_TEST_START_0001")
190 self.pg0.config_ip6()
191 route_8000 = VppIpRoute(
195 [VppRoutePath(self.pg0.remote_ip6, self.pg0.sw_if_index)],
197 route_8000.add_vpp_config()
201 Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac)
202 / IPv6(src="3001::1", dst="8000::100")
203 / UDP(sport=1234, dport=1234)
207 self.send_packets(self.pg1, self.pg0)
209 self.pg0.unconfig_ip6()
210 self.logger.info("FFP_TEST_FINISH_0001")
212 def test_ip6_rx_p2p_subif(self):
213 """receive ipv6 packet via p2p subinterface"""
214 self.logger.info("FFP_TEST_START_0002")
216 route_9001 = VppIpRoute(
220 [VppRoutePath(self.pg1.remote_ip6, self.pg1.sw_if_index)],
222 route_9001.add_vpp_config()
226 src_mac=self.pg0._remote_hosts[0].mac,
227 dst_mac=self.pg0.local_mac,
228 src_ip=self.p2p_sub_ifs[0].remote_ip6,
233 self.send_packets(self.pg0, self.pg1, self.packets)
234 self.assert_packet_counter_equal("p2p-ethernet-input", 1)
236 route_9001.remove_vpp_config()
237 self.logger.info("FFP_TEST_FINISH_0002")
239 def test_ip6_rx_p2p_subif_route(self):
240 """route rx ip6 packet not matching p2p subinterface"""
241 self.logger.info("FFP_TEST_START_0003")
243 self.pg0.config_ip6()
245 route_3 = VppIpRoute(
249 [VppRoutePath(self.pg1._remote_hosts[0].ip6, self.pg1.sw_if_index)],
251 route_3.add_vpp_config()
255 src_mac="02:03:00:00:ff:ff",
256 dst_mac=self.pg0.local_mac,
262 self.send_packets(self.pg0, self.pg1)
264 self.pg0.unconfig_ip6()
266 route_3.remove_vpp_config()
268 self.logger.info("FFP_TEST_FINISH_0003")
270 def test_ip6_rx_p2p_subif_drop(self):
271 """drop rx packet not matching p2p subinterface"""
272 self.logger.info("FFP_TEST_START_0004")
274 route_9001 = VppIpRoute(
278 [VppRoutePath(self.pg1._remote_hosts[0].ip6, self.pg1.sw_if_index)],
280 route_9001.add_vpp_config()
284 src_mac="02:03:00:00:ff:ff",
285 dst_mac=self.pg0.local_mac,
292 self.send_packets(self.pg0, self.pg1, count=0)
293 self.logger.info("FFP_TEST_FINISH_0004")
295 def test_ip6_tx_p2p_subif(self):
296 """send packet via p2p subinterface"""
297 self.logger.info("FFP_TEST_START_0005")
299 self.pg0.config_ip6()
301 route_8000 = VppIpRoute(
305 [VppRoutePath(self.pg0.remote_hosts[0].ip6, self.pg0.sw_if_index)],
307 route_8000.add_vpp_config()
308 route_8001 = VppIpRoute(
314 self.p2p_sub_ifs[0].remote_ip6, self.p2p_sub_ifs[0].sw_if_index
318 route_8001.add_vpp_config()
319 route_8002 = VppIpRoute(
325 self.p2p_sub_ifs[1].remote_ip6, self.p2p_sub_ifs[1].sw_if_index
329 route_8002.add_vpp_config()
331 for i in range(0, 3):
334 src_mac=self.pg1.remote_mac,
335 dst_mac=self.pg1.local_mac,
336 src_ip=self.pg1.remote_ip6,
337 dst_ip="800%d::100" % i,
341 self.send_packets(self.pg1, self.pg0, count=3)
343 route_8000.remove_vpp_config()
344 route_8001.remove_vpp_config()
345 route_8002.remove_vpp_config()
347 self.pg0.unconfig_ip6()
348 self.logger.info("FFP_TEST_FINISH_0005")
350 def test_ip6_tx_p2p_subif_drop(self):
351 """drop tx ip6 packet not matching p2p subinterface"""
352 self.logger.info("FFP_TEST_START_0006")
356 src_mac="02:03:00:00:ff:ff",
357 dst_mac=self.pg0.local_mac,
364 self.send_packets(self.pg0, self.pg1, count=0)
365 self.logger.info("FFP_TEST_FINISH_0006")
368 class P2PEthernetIPV4(VppTestCase):
369 """P2P Ethernet IPv4 tests"""
376 super(P2PEthernetIPV4, cls).setUpClass()
378 # Create pg interfaces
379 cls.create_pg_interfaces(range(3))
382 cls.pg_if_packet_sizes = [64, 512, 1518, 9018]
384 # Set up all interfaces
385 for i in cls.pg_interfaces:
389 cls.pg0.generate_remote_hosts(5)
390 cls.pg0.configure_ipv4_neighbors()
393 cls.pg1.generate_remote_hosts(5)
394 cls.pg1.configure_ipv4_neighbors()
397 def tearDownClass(cls):
398 super(P2PEthernetIPV4, cls).tearDownClass()
401 super(P2PEthernetIPV4, self).setUp()
402 for p in self.packets:
403 self.packets.remove(p)
404 self.p2p_sub_ifs.append(
405 self.create_p2p_ethernet(self.pg0, 1, self.pg0._remote_hosts[0].mac)
407 self.p2p_sub_ifs.append(
408 self.create_p2p_ethernet(self.pg0, 2, self.pg0._remote_hosts[1].mac)
410 self.vapi.cli("trace add p2p-ethernet-input 50")
413 while len(self.p2p_sub_ifs):
414 p2p = self.p2p_sub_ifs.pop()
415 self.delete_p2p_ethernet(p2p)
416 super(P2PEthernetIPV4, self).tearDown()
419 self, src_mac=None, dst_mac=None, src_ip=None, dst_ip=None, size=None
423 pkt_size = random.choice(self.pg_if_packet_sizes)
424 p = Ether(src=src_mac, dst=dst_mac)
425 p /= IP(src=src_ip, dst=dst_ip)
426 p /= UDP(sport=1234, dport=4321) / Raw(b"\xa5" * 20)
427 self.extend_packet(p, pkt_size)
430 def send_packets(self, src_if=None, dst_if=None, packets=None, count=None):
431 self.pg_enable_capture([dst_if])
433 packets = self.packets
434 src_if.add_stream(packets)
438 return dst_if.get_capture(count)
440 def create_p2p_ethernet(self, parent_if, sub_id, remote_mac):
441 p2p = VppP2PSubint(self, parent_if, sub_id, mac_pton(remote_mac))
446 def delete_p2p_ethernet(self, p2p):
449 self.vapi.p2p_ethernet_del(p2p.parent.sw_if_index, p2p.p2p_remote_mac)
451 def test_ip4_rx_p2p_subif(self):
452 """receive ipv4 packet via p2p subinterface"""
453 self.logger.info("FFP_TEST_START_0002")
455 route_9000 = VppIpRoute(
459 [VppRoutePath(self.pg1.remote_ip4, self.pg1.sw_if_index)],
461 route_9000.add_vpp_config()
465 src_mac=self.pg0._remote_hosts[0].mac,
466 dst_mac=self.pg0.local_mac,
467 src_ip=self.p2p_sub_ifs[0].remote_ip4,
472 self.send_packets(self.pg0, self.pg1, self.packets)
474 self.assert_packet_counter_equal("p2p-ethernet-input", 1)
476 route_9000.remove_vpp_config()
477 self.logger.info("FFP_TEST_FINISH_0002")
479 def test_ip4_rx_p2p_subif_route(self):
480 """route rx packet not matching p2p subinterface"""
481 self.logger.info("FFP_TEST_START_0003")
483 route_9001 = VppIpRoute(
487 [VppRoutePath(self.pg1.remote_ip4, self.pg1.sw_if_index)],
489 route_9001.add_vpp_config()
493 src_mac="02:01:00:00:ff:ff",
494 dst_mac=self.pg0.local_mac,
500 self.send_packets(self.pg0, self.pg1)
502 route_9001.remove_vpp_config()
504 self.logger.info("FFP_TEST_FINISH_0003")
506 def test_ip4_tx_p2p_subif(self):
507 """send ip4 packet via p2p subinterface"""
508 self.logger.info("FFP_TEST_START_0005")
510 route_9100 = VppIpRoute(
517 self.pg0.sw_if_index,
521 route_9100.add_vpp_config()
522 route_9200 = VppIpRoute(
528 self.p2p_sub_ifs[0].remote_ip4,
529 self.p2p_sub_ifs[0].sw_if_index,
533 route_9200.add_vpp_config()
534 route_9300 = VppIpRoute(
540 self.p2p_sub_ifs[1].remote_ip4, self.p2p_sub_ifs[1].sw_if_index
544 route_9300.add_vpp_config()
546 for i in range(0, 3):
549 src_mac=self.pg1.remote_mac,
550 dst_mac=self.pg1.local_mac,
551 src_ip=self.pg1.remote_ip4,
552 dst_ip="9.%d.0.100" % (i + 1),
556 self.send_packets(self.pg1, self.pg0)
558 # route_7000.remove_vpp_config()
559 route_9100.remove_vpp_config()
560 route_9200.remove_vpp_config()
561 route_9300.remove_vpp_config()
563 self.logger.info("FFP_TEST_FINISH_0005")
565 def test_ip4_tx_p2p_subif_drop(self):
566 """drop tx ip4 packet not matching p2p subinterface"""
567 self.logger.info("FFP_TEST_START_0006")
571 src_mac="02:01:00:00:ff:ff",
572 dst_mac=self.pg0.local_mac,
579 self.send_packets(self.pg0, self.pg1, count=0)
580 self.logger.info("FFP_TEST_FINISH_0006")
583 if __name__ == "__main__":
584 unittest.main(testRunner=VppTestRunner)