6 from framework import VppTestCase, VppTestRunner
7 from vpp_ip_route import VppIpRoute, VppRoutePath
8 from vpp_pppoe_interface import VppPppoeInterface
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
17 from util import ppp, ppc
21 class TestPPPoE(VppTestCase):
22 """ PPPoE Test Case """
26 super(TestPPPoE, cls).setUpClass()
29 cls.dst_ip = "100.1.1.100"
30 cls.dst_ipn = socket.inet_pton(socket.AF_INET, cls.dst_ip)
33 super(TestPPPoE, self).setUp()
35 # create 2 pg interfaces
36 self.create_pg_interfaces(range(3))
38 for i in self.pg_interfaces:
44 super(TestPPPoE, self).tearDown()
46 self.logger.info(self.vapi.cli("show int"))
47 self.logger.info(self.vapi.cli("show pppoe fib"))
48 self.logger.info(self.vapi.cli("show pppoe session"))
49 self.logger.info(self.vapi.cli("show ip fib"))
50 self.logger.info(self.vapi.cli("show trace"))
52 for i in self.pg_interfaces:
56 def create_stream_pppoe_discovery(self, src_if, dst_if,
59 for i in range(count):
60 # create packet info stored in the test case instance
61 info = self.create_packet_info(src_if, dst_if)
62 # convert the info into packet payload
63 payload = self.info_to_payload(info)
64 # create the packet itself
65 p = (Ether(dst=src_if.local_mac, src=client_mac) /
68 # store a copy of the packet in the packet info
70 # append the packet to the list
73 # return the created packet list
76 def create_stream_pppoe_lcp(self, src_if, dst_if,
77 client_mac, session_id, count=1):
79 for i in range(count):
80 # create packet info stored in the test case instance
81 info = self.create_packet_info(src_if, dst_if)
82 # convert the info into packet payload
83 payload = self.info_to_payload(info)
84 # create the packet itself
85 p = (Ether(dst=src_if.local_mac, src=client_mac) /
86 PPPoE(sessionid=session_id) /
89 # store a copy of the packet in the packet info
91 # append the packet to the list
94 # return the created packet list
97 def create_stream_pppoe_ip4(self, src_if, dst_if,
98 client_mac, session_id, client_ip, count=1):
100 for i in range(count):
101 # create packet info stored in the test case instance
102 info = self.create_packet_info(src_if, dst_if)
103 # convert the info into packet payload
104 payload = self.info_to_payload(info)
105 # create the packet itself
106 p = (Ether(dst=src_if.local_mac, src=client_mac) /
107 PPPoE(sessionid=session_id) /
109 IP(src=client_ip, dst=self.dst_ip) /
111 # store a copy of the packet in the packet info
113 # append the packet to the list
116 # return the created packet list
119 def create_stream_ip4(self, src_if, dst_if, client_ip, dst_ip, count=1):
121 for i in range(count):
122 # create packet info stored in the test case instance
123 info = self.create_packet_info(src_if, dst_if)
124 # convert the info into packet payload
125 payload = self.info_to_payload(info)
126 # create the packet itself
127 p = (Ether(dst=src_if.local_mac, src=src_if.remote_mac) /
128 IP(src=dst_ip, dst=client_ip) /
130 # store a copy of the packet in the packet info
132 # append the packet to the list
135 # return the created packet list
138 def verify_decapped_pppoe(self, src_if, capture, sent):
139 self.assertEqual(len(capture), len(sent))
141 for i in range(len(capture)):
149 self.assertEqual(rx_ip.src, tx_ip.src)
150 self.assertEqual(rx_ip.dst, tx_ip.dst)
153 self.logger.error(ppp("Rx:", rx))
154 self.logger.error(ppp("Tx:", tx))
157 def verify_encaped_pppoe(self, src_if, capture, sent, session_id):
159 self.assertEqual(len(capture), len(sent))
161 for i in range(len(capture)):
169 self.assertEqual(rx_ip.src, tx_ip.src)
170 self.assertEqual(rx_ip.dst, tx_ip.dst)
174 self.assertEqual(rx_pppoe.sessionid, session_id)
177 self.logger.error(ppp("Rx:", rx))
178 self.logger.error(ppp("Tx:", tx))
181 def test_PPPoE_Decap(self):
182 """ PPPoE Decap Test """
184 self.vapi.cli("clear trace")
187 # Add a route that resolves the server's destination
189 route_sever_dst = VppIpRoute(self, "100.1.1.100", 32,
190 [VppRoutePath(self.pg1.remote_ip4,
191 self.pg1.sw_if_index)])
192 route_sever_dst.add_vpp_config()
194 # Send PPPoE Discovery
195 tx0 = self.create_stream_pppoe_discovery(self.pg0, self.pg1,
197 self.pg0.add_stream(tx0)
201 tx1 = self.create_stream_pppoe_lcp(self.pg0, self.pg1,
204 self.pg0.add_stream(tx1)
207 # Create PPPoE session
208 pppoe_if = VppPppoeInterface(self,
212 pppoe_if.add_vpp_config()
215 # Send tunneled packets that match the created tunnel and
216 # are decapped and forwarded
218 tx2 = self.create_stream_pppoe_ip4(self.pg0, self.pg1,
222 self.pg0.add_stream(tx2)
224 self.pg_enable_capture(self.pg_interfaces)
227 rx2 = self.pg1.get_capture(len(tx2))
228 self.verify_decapped_pppoe(self.pg0, rx2, tx2)
230 self.logger.info(self.vapi.cli("show pppoe fib"))
231 self.logger.info(self.vapi.cli("show pppoe session"))
232 self.logger.info(self.vapi.cli("show ip fib"))
238 # Delete PPPoE session
239 pppoe_if.remove_vpp_config()
241 # Delete a route that resolves the server's destination
242 route_sever_dst.remove_vpp_config()
244 def test_PPPoE_Encap(self):
245 """ PPPoE Encap Test """
247 self.vapi.cli("clear trace")
250 # Add a route that resolves the server's destination
252 route_sever_dst = VppIpRoute(self, "100.1.1.100", 32,
253 [VppRoutePath(self.pg1.remote_ip4,
254 self.pg1.sw_if_index)])
255 route_sever_dst.add_vpp_config()
257 # Send PPPoE Discovery
258 tx0 = self.create_stream_pppoe_discovery(self.pg0, self.pg1,
260 self.pg0.add_stream(tx0)
264 tx1 = self.create_stream_pppoe_lcp(self.pg0, self.pg1,
267 self.pg0.add_stream(tx1)
270 # Create PPPoE session
271 pppoe_if = VppPppoeInterface(self,
275 pppoe_if.add_vpp_config()
278 # Send a packet stream that is routed into the session
279 # - packets are PPPoE encapped
281 self.vapi.cli("clear trace")
282 tx2 = self.create_stream_ip4(self.pg1, self.pg0,
283 self.pg0.remote_ip4, self.dst_ip, 65)
284 self.pg1.add_stream(tx2)
286 self.pg_enable_capture(self.pg_interfaces)
289 rx2 = self.pg0.get_capture(len(tx2))
290 self.verify_encaped_pppoe(self.pg1, rx2, tx2, self.session_id)
292 self.logger.info(self.vapi.cli("show pppoe fib"))
293 self.logger.info(self.vapi.cli("show pppoe session"))
294 self.logger.info(self.vapi.cli("show ip fib"))
295 self.logger.info(self.vapi.cli("show adj"))
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 pppoe_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 pppoe_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)