6 from framework import VppTestCase, VppTestRunner
7 from vpp_ip_route import VppIpRoute, VppRoutePath
8 from vpp_pppoe_interface import VppPppoeInterface, VppPppoe6Interface
9 from vpp_papi_provider import L2_VTR_OP
11 from scapy.packet import Raw
12 from scapy.layers.l2 import Ether
13 from scapy.layers.ppp import PPPoE, PPPoED, PPP
14 from scapy.layers.inet import IP, UDP
15 from scapy.layers.inet6 import IPv6
16 from scapy.volatile import RandMAC, RandIP
18 from util import ppp, ppc, mactobinary
22 class TestPPPoE(VppTestCase):
23 """ PPPoE Test Case """
27 super(TestPPPoE, cls).setUpClass()
30 cls.dst_ip = "100.1.1.100"
31 cls.dst_ipn = socket.inet_pton(socket.AF_INET, cls.dst_ip)
34 super(TestPPPoE, self).setUp()
36 # create 2 pg interfaces
37 self.create_pg_interfaces(range(3))
39 for i in self.pg_interfaces:
45 super(TestPPPoE, self).tearDown()
47 self.logger.info(self.vapi.cli("show int"))
48 self.logger.info(self.vapi.cli("show pppoe fib"))
49 self.logger.info(self.vapi.cli("show pppoe session"))
50 self.logger.info(self.vapi.cli("show ip fib"))
51 self.logger.info(self.vapi.cli("show trace"))
53 for i in self.pg_interfaces:
57 def create_stream_pppoe_discovery(self, src_if, dst_if,
60 for i in range(count):
61 # create packet info stored in the test case instance
62 info = self.create_packet_info(src_if, dst_if)
63 # convert the info into packet payload
64 payload = self.info_to_payload(info)
65 # create the packet itself
66 p = (Ether(dst=src_if.local_mac, src=client_mac) /
69 # store a copy of the packet in the packet info
71 # append the packet to the list
74 # return the created packet list
77 def create_stream_pppoe_lcp(self, src_if, dst_if,
78 client_mac, session_id, count=1):
80 for i in range(count):
81 # create packet info stored in the test case instance
82 info = self.create_packet_info(src_if, dst_if)
83 # convert the info into packet payload
84 payload = self.info_to_payload(info)
85 # create the packet itself
86 p = (Ether(dst=src_if.local_mac, src=client_mac) /
87 PPPoE(sessionid=session_id) /
90 # store a copy of the packet in the packet info
92 # append the packet to the list
95 # return the created packet list
98 def create_stream_pppoe_ip4(self, src_if, dst_if,
99 client_mac, session_id, client_ip, count=1):
101 for i in range(count):
102 # create packet info stored in the test case instance
103 info = self.create_packet_info(src_if, dst_if)
104 # convert the info into packet payload
105 payload = self.info_to_payload(info)
106 # create the packet itself
107 p = (Ether(dst=src_if.local_mac, src=client_mac) /
108 PPPoE(sessionid=session_id) /
110 IP(src=client_ip, dst=self.dst_ip) /
112 # store a copy of the packet in the packet info
114 # append the packet to the list
117 # return the created packet list
120 def create_stream_ip4(self, src_if, dst_if, client_ip, dst_ip, count=1):
122 for i in range(count):
123 # create packet info stored in the test case instance
124 info = self.create_packet_info(src_if, dst_if)
125 # convert the info into packet payload
126 payload = self.info_to_payload(info)
127 # create the packet itself
128 p = (Ether(dst=src_if.local_mac, src=src_if.remote_mac) /
129 IP(src=dst_ip, dst=client_ip) /
131 # store a copy of the packet in the packet info
133 # append the packet to the list
136 # return the created packet list
139 def verify_decapped_pppoe(self, src_if, capture, sent):
140 self.assertEqual(len(capture), len(sent))
142 for i in range(len(capture)):
150 self.assertEqual(rx_ip.src, tx_ip.src)
151 self.assertEqual(rx_ip.dst, tx_ip.dst)
154 self.logger.error(ppp("Rx:", rx))
155 self.logger.error(ppp("Tx:", tx))
158 def verify_encaped_pppoe(self, src_if, capture, sent, session_id):
160 self.assertEqual(len(capture), len(sent))
162 for i in range(len(capture)):
170 self.assertEqual(rx_ip.src, tx_ip.src)
171 self.assertEqual(rx_ip.dst, tx_ip.dst)
175 self.assertEqual(rx_pppoe.sessionid, session_id)
178 self.logger.error(ppp("Rx:", rx))
179 self.logger.error(ppp("Tx:", tx))
182 def test_PPPoE_Decap(self):
183 """ PPPoE Decap Test """
185 self.vapi.cli("clear trace")
188 # Add a route that resolves the server's destination
190 route_sever_dst = VppIpRoute(self, "100.1.1.100", 32,
191 [VppRoutePath(self.pg1.remote_ip4,
192 self.pg1.sw_if_index)])
193 route_sever_dst.add_vpp_config()
195 # Send PPPoE Discovery
196 tx0 = self.create_stream_pppoe_discovery(self.pg0, self.pg1,
198 self.pg0.add_stream(tx0)
202 tx1 = self.create_stream_pppoe_lcp(self.pg0, self.pg1,
205 self.pg0.add_stream(tx1)
208 # Create PPPoE session
209 pppoe_if = VppPppoeInterface(self,
213 pppoe_if.add_vpp_config()
216 # Send tunneled packets that match the created tunnel and
217 # are decapped and forwarded
219 tx2 = self.create_stream_pppoe_ip4(self.pg0, self.pg1,
223 self.pg0.add_stream(tx2)
225 self.pg_enable_capture(self.pg_interfaces)
228 rx2 = self.pg1.get_capture(len(tx2))
229 self.verify_decapped_pppoe(self.pg0, rx2, tx2)
231 self.logger.info(self.vapi.cli("show pppoe fib"))
232 self.logger.info(self.vapi.cli("show pppoe session"))
233 self.logger.info(self.vapi.cli("show ip fib"))
239 # Delete PPPoE session
240 pppoe_if.remove_vpp_config()
242 # Delete a route that resolves the server's destination
243 route_sever_dst.remove_vpp_config()
245 def test_PPPoE_Encap(self):
246 """ PPPoE Encap Test """
248 self.vapi.cli("clear trace")
251 # Add a route that resolves the server's destination
253 route_sever_dst = VppIpRoute(self, "100.1.1.100", 32,
254 [VppRoutePath(self.pg1.remote_ip4,
255 self.pg1.sw_if_index)])
256 route_sever_dst.add_vpp_config()
258 # Send PPPoE Discovery
259 tx0 = self.create_stream_pppoe_discovery(self.pg0, self.pg1,
261 self.pg0.add_stream(tx0)
265 tx1 = self.create_stream_pppoe_lcp(self.pg0, self.pg1,
268 self.pg0.add_stream(tx1)
271 # Create PPPoE session
272 pppoe_if = VppPppoeInterface(self,
276 pppoe_if.add_vpp_config()
279 # Send a packet stream that is routed into the session
280 # - packets are PPPoE encapped
282 self.vapi.cli("clear trace")
283 tx2 = self.create_stream_ip4(self.pg1, self.pg0,
284 self.pg0.remote_ip4, self.dst_ip)
285 self.pg1.add_stream(tx2)
287 self.pg_enable_capture(self.pg_interfaces)
290 rx2 = self.pg0.get_capture(len(tx2))
291 self.verify_encaped_pppoe(self.pg1, rx2, tx2, self.session_id)
293 self.logger.info(self.vapi.cli("show pppoe fib"))
294 self.logger.info(self.vapi.cli("show pppoe session"))
295 self.logger.info(self.vapi.cli("show ip fib"))
301 # Delete PPPoE session
302 pppoe_if.remove_vpp_config()
304 # Delete a route that resolves the server's destination
305 route_sever_dst.remove_vpp_config()
307 def test_PPPoE_Add_Twice(self):
308 """ PPPoE Add Same Session Twice Test """
310 self.vapi.cli("clear trace")
313 # Add a route that resolves the server's destination
315 route_sever_dst = VppIpRoute(self, "100.1.1.100", 32,
316 [VppRoutePath(self.pg1.remote_ip4,
317 self.pg1.sw_if_index)])
318 route_sever_dst.add_vpp_config()
320 # Send PPPoE Discovery
321 tx0 = self.create_stream_pppoe_discovery(self.pg0, self.pg1,
323 self.pg0.add_stream(tx0)
327 tx1 = self.create_stream_pppoe_lcp(self.pg0, self.pg1,
330 self.pg0.add_stream(tx1)
333 # Create PPPoE session
334 pppoe_if = VppPppoeInterface(self,
338 pppoe_if.add_vpp_config()
341 # The double create (create the same session twice) should fail,
342 # and we should still be able to use the original
345 gre_if.add_vpp_config()
349 self.fail("Double GRE tunnel add does not fail")
355 # Delete PPPoE session
356 pppoe_if.remove_vpp_config()
358 # Delete a route that resolves the server's destination
359 route_sever_dst.remove_vpp_config()
361 def test_PPPoE_Del_Twice(self):
362 """ PPPoE Delete Same Session Twice Test """
364 self.vapi.cli("clear trace")
367 # Add a route that resolves the server's destination
369 route_sever_dst = VppIpRoute(self, "100.1.1.100", 32,
370 [VppRoutePath(self.pg1.remote_ip4,
371 self.pg1.sw_if_index)])
372 route_sever_dst.add_vpp_config()
374 # Send PPPoE Discovery
375 tx0 = self.create_stream_pppoe_discovery(self.pg0, self.pg1,
377 self.pg0.add_stream(tx0)
381 tx1 = self.create_stream_pppoe_lcp(self.pg0, self.pg1,
384 self.pg0.add_stream(tx1)
387 # Create PPPoE session
388 pppoe_if = VppPppoeInterface(self,
392 pppoe_if.add_vpp_config()
394 # Delete PPPoE session
395 pppoe_if.remove_vpp_config()
398 # The double del (del the same session twice) should fail,
399 # and we should still be able to use the original
402 gre_if.remove_vpp_config()
406 self.fail("Double GRE tunnel del does not fail")
412 # Delete a route that resolves the server's destination
413 route_sever_dst.remove_vpp_config()
415 def test_PPPoE_Decap_Multiple(self):
416 """ PPPoE Decap Multiple Sessions Test """
418 self.vapi.cli("clear trace")
421 # Add a route that resolves the server's destination
423 route_sever_dst = VppIpRoute(self, "100.1.1.100", 32,
424 [VppRoutePath(self.pg1.remote_ip4,
425 self.pg1.sw_if_index)])
426 route_sever_dst.add_vpp_config()
428 # Send PPPoE Discovery 1
429 tx0 = self.create_stream_pppoe_discovery(self.pg0, self.pg1,
431 self.pg0.add_stream(tx0)
434 # Send PPPoE PPP LCP 1
435 tx1 = self.create_stream_pppoe_lcp(self.pg0, self.pg1,
438 self.pg0.add_stream(tx1)
441 # Create PPPoE session 1
442 pppoe_if1 = VppPppoeInterface(self,
446 pppoe_if1.add_vpp_config()
448 # Send PPPoE Discovery 2
449 tx3 = self.create_stream_pppoe_discovery(self.pg2, self.pg1,
451 self.pg2.add_stream(tx3)
454 # Send PPPoE PPP LCP 2
455 tx4 = self.create_stream_pppoe_lcp(self.pg2, self.pg1,
458 self.pg2.add_stream(tx4)
461 # Create PPPoE session 2
462 pppoe_if2 = VppPppoeInterface(self,
466 pppoe_if2.add_vpp_config()
469 # Send tunneled packets that match the created tunnel and
470 # are decapped and forwarded
472 tx2 = self.create_stream_pppoe_ip4(self.pg0, self.pg1,
476 self.pg0.add_stream(tx2)
478 self.pg_enable_capture(self.pg_interfaces)
481 rx2 = self.pg1.get_capture(len(tx2))
482 self.verify_decapped_pppoe(self.pg0, rx2, tx2)
484 tx5 = self.create_stream_pppoe_ip4(self.pg2, self.pg1,
488 self.pg2.add_stream(tx5)
490 self.pg_enable_capture(self.pg_interfaces)
493 rx5 = self.pg1.get_capture(len(tx5))
494 self.verify_decapped_pppoe(self.pg2, rx5, tx5)
496 self.logger.info(self.vapi.cli("show pppoe fib"))
497 self.logger.info(self.vapi.cli("show pppoe session"))
498 self.logger.info(self.vapi.cli("show ip fib"))
504 # Delete PPPoE session
505 pppoe_if1.remove_vpp_config()
506 pppoe_if2.remove_vpp_config()
508 # Delete a route that resolves the server's destination
509 route_sever_dst.remove_vpp_config()
511 def test_PPPoE_Encap_Multiple(self):
512 """ PPPoE Encap Multiple Sessions Test """
514 self.vapi.cli("clear trace")
517 # Add a route that resolves the server's destination
519 route_sever_dst = VppIpRoute(self, "100.1.1.100", 32,
520 [VppRoutePath(self.pg1.remote_ip4,
521 self.pg1.sw_if_index)])
522 route_sever_dst.add_vpp_config()
524 # Send PPPoE Discovery 1
525 tx0 = self.create_stream_pppoe_discovery(self.pg0, self.pg1,
527 self.pg0.add_stream(tx0)
530 # Send PPPoE PPP LCP 1
531 tx1 = self.create_stream_pppoe_lcp(self.pg0, self.pg1,
534 self.pg0.add_stream(tx1)
537 # Create PPPoE session 1
538 pppoe_if1 = VppPppoeInterface(self,
542 pppoe_if1.add_vpp_config()
544 # Send PPPoE Discovery 2
545 tx3 = self.create_stream_pppoe_discovery(self.pg2, self.pg1,
547 self.pg2.add_stream(tx3)
550 # Send PPPoE PPP LCP 2
551 tx4 = self.create_stream_pppoe_lcp(self.pg2, self.pg1,
554 self.pg2.add_stream(tx4)
557 # Create PPPoE session 2
558 pppoe_if2 = VppPppoeInterface(self,
562 pppoe_if2.add_vpp_config()
565 # Send a packet stream that is routed into the session
566 # - packets are PPPoE encapped
568 self.vapi.cli("clear trace")
569 tx2 = self.create_stream_ip4(self.pg1, self.pg0,
570 self.pg0.remote_ip4, self.dst_ip)
571 self.pg1.add_stream(tx2)
573 self.pg_enable_capture(self.pg_interfaces)
576 rx2 = self.pg0.get_capture(len(tx2))
577 self.verify_encaped_pppoe(self.pg1, rx2, tx2, self.session_id)
579 tx5 = self.create_stream_ip4(self.pg1, self.pg2,
580 self.pg2.remote_ip4, self.dst_ip)
581 self.pg1.add_stream(tx5)
583 self.pg_enable_capture(self.pg_interfaces)
586 rx5 = self.pg2.get_capture(len(tx5))
587 self.verify_encaped_pppoe(self.pg1, rx5, tx5, self.session_id + 1)
589 self.logger.info(self.vapi.cli("show pppoe fib"))
590 self.logger.info(self.vapi.cli("show pppoe session"))
591 self.logger.info(self.vapi.cli("show ip fib"))
597 # Delete PPPoE session
598 pppoe_if1.remove_vpp_config()
599 pppoe_if2.remove_vpp_config()
601 # Delete a route that resolves the server's destination
602 route_sever_dst.remove_vpp_config()
604 if __name__ == '__main__':
605 unittest.main(testRunner=VppTestRunner)