test/test_pppoe.py

   1 #!/usr/bin/env python3
   2
   3 import socket
   4 import unittest
   5
   6 from scapy.packet import Raw
   7 from scapy.layers.l2 import Ether
   8 from scapy.layers.ppp import PPPoE, PPPoED, PPP
   9 from scapy.layers.inet import IP
  10
  11 from framework import VppTestCase, VppTestRunner
  12 from vpp_ip_route import VppIpRoute, VppRoutePath
  13 from vpp_pppoe_interface import VppPppoeInterface
  14 from util import ppp, ppc
  15
  16
  17 class TestPPPoE(VppTestCase):
  18     """PPPoE Test Case"""
  19
  20     @classmethod
  21     def setUpClass(cls):
  22         super(TestPPPoE, cls).setUpClass()
  23
  24         cls.session_id = 1
  25         cls.dst_ip = "100.1.1.100"
  26         cls.dst_ipn = socket.inet_pton(socket.AF_INET, cls.dst_ip)
  27
  28     @classmethod
  29     def tearDownClass(cls):
  30         super(TestPPPoE, cls).tearDownClass()
  31
  32     def setUp(self):
  33         super(TestPPPoE, self).setUp()
  34
  35         # create 2 pg interfaces
  36         self.create_pg_interfaces(range(3))
  37
  38         for i in self.pg_interfaces:
  39             i.admin_up()
  40             i.config_ip4()
  41             i.resolve_arp()
  42
  43     def tearDown(self):
  44         super(TestPPPoE, self).tearDown()
  45
  46         for i in self.pg_interfaces:
  47             i.unconfig_ip4()
  48             i.admin_down()
  49
  50     def show_commands_at_teardown(self):
  51         self.logger.info(self.vapi.cli("show int"))
  52         self.logger.info(self.vapi.cli("show pppoe fib"))
  53         self.logger.info(self.vapi.cli("show pppoe session"))
  54         self.logger.info(self.vapi.cli("show ip fib"))
  55         self.logger.info(self.vapi.cli("show trace"))
  56
  57     def create_stream_pppoe_discovery(self, src_if, dst_if, client_mac, count=1):
  58         packets = []
  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 = (
  66                 Ether(dst=src_if.local_mac, src=client_mac)
  67                 / PPPoED(sessionid=0)
  68                 / Raw(payload)
  69             )
  70             # store a copy of the packet in the packet info
  71             info.data = p.copy()
  72             # append the packet to the list
  73             packets.append(p)
  74
  75         # return the created packet list
  76         return packets
  77
  78     def create_stream_pppoe_lcp(self, src_if, dst_if, client_mac, session_id, count=1):
  79         packets = []
  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 = (
  87                 Ether(dst=src_if.local_mac, src=client_mac)
  88                 / PPPoE(sessionid=session_id)
  89                 / PPP(proto=0xC021)
  90                 / Raw(payload)
  91             )
  92             # store a copy of the packet in the packet info
  93             info.data = p.copy()
  94             # append the packet to the list
  95             packets.append(p)
  96
  97         # return the created packet list
  98         return packets
  99
 100     def create_stream_pppoe_ip4(
 101         self, src_if, dst_if, client_mac, session_id, client_ip, count=1
 102     ):
 103         packets = []
 104         for i in range(count):
 105             # create packet info stored in the test case instance
 106             info = self.create_packet_info(src_if, dst_if)
 107             # convert the info into packet payload
 108             payload = self.info_to_payload(info)
 109             # create the packet itself
 110             p = (
 111                 Ether(dst=src_if.local_mac, src=client_mac)
 112                 / PPPoE(sessionid=session_id)
 113                 / PPP(proto=0x0021)
 114                 / IP(src=client_ip, dst=self.dst_ip)
 115                 / Raw(payload)
 116             )
 117             # store a copy of the packet in the packet info
 118             info.data = p.copy()
 119             # append the packet to the list
 120             packets.append(p)
 121
 122         # return the created packet list
 123         return packets
 124
 125     def create_stream_ip4(self, src_if, dst_if, client_ip, dst_ip, count=1):
 126         pkts = []
 127         for i in range(count):
 128             # create packet info stored in the test case instance
 129             info = self.create_packet_info(src_if, dst_if)
 130             # convert the info into packet payload
 131             payload = self.info_to_payload(info)
 132             # create the packet itself
 133             p = (
 134                 Ether(dst=src_if.local_mac, src=src_if.remote_mac)
 135                 / IP(src=dst_ip, dst=client_ip)
 136                 / Raw(payload)
 137             )
 138             # store a copy of the packet in the packet info
 139             info.data = p.copy()
 140             # append the packet to the list
 141             pkts.append(p)
 142
 143         # return the created packet list
 144         return pkts
 145
 146     def verify_decapped_pppoe(self, src_if, capture, sent):
 147         self.assertEqual(len(capture), len(sent))
 148
 149         for i in range(len(capture)):
 150             try:
 151                 tx = sent[i]
 152                 rx = capture[i]
 153
 154                 tx_ip = tx[IP]
 155                 rx_ip = rx[IP]
 156
 157                 self.assertEqual(rx_ip.src, tx_ip.src)
 158                 self.assertEqual(rx_ip.dst, tx_ip.dst)
 159
 160             except:
 161                 self.logger.error(ppp("Rx:", rx))
 162                 self.logger.error(ppp("Tx:", tx))
 163                 raise
 164
 165     def verify_encaped_pppoe(self, src_if, capture, sent, session_id):
 166         self.assertEqual(len(capture), len(sent))
 167
 168         for i in range(len(capture)):
 169             try:
 170                 tx = sent[i]
 171                 rx = capture[i]
 172
 173                 tx_ip = tx[IP]
 174                 rx_ip = rx[IP]
 175
 176                 self.assertEqual(rx_ip.src, tx_ip.src)
 177                 self.assertEqual(rx_ip.dst, tx_ip.dst)
 178
 179                 rx_pppoe = rx[PPPoE]
 180
 181                 self.assertEqual(rx_pppoe.sessionid, session_id)
 182
 183             except:
 184                 self.logger.error(ppp("Rx:", rx))
 185                 self.logger.error(ppp("Tx:", tx))
 186                 raise
 187
 188     def test_PPPoE_Decap(self):
 189         """PPPoE Decap Test"""
 190
 191         self.vapi.cli("clear trace")
 192
 193         #
 194         # Add a route that resolves the server's destination
 195         #
 196         route_sever_dst = VppIpRoute(
 197             self,
 198             "100.1.1.100",
 199             32,
 200             [VppRoutePath(self.pg1.remote_ip4, self.pg1.sw_if_index)],
 201         )
 202         route_sever_dst.add_vpp_config()
 203
 204         # Send PPPoE Discovery
 205         tx0 = self.create_stream_pppoe_discovery(
 206             self.pg0, self.pg1, self.pg0.remote_mac
 207         )
 208         self.pg0.add_stream(tx0)
 209         self.pg_start()
 210
 211         # Send PPPoE PPP LCP
 212         tx1 = self.create_stream_pppoe_lcp(
 213             self.pg0, self.pg1, self.pg0.remote_mac, self.session_id
 214         )
 215         self.pg0.add_stream(tx1)
 216         self.pg_start()
 217
 218         # Create PPPoE session
 219         pppoe_if = VppPppoeInterface(
 220             self, self.pg0.remote_ip4, self.pg0.remote_mac, self.session_id
 221         )
 222         pppoe_if.add_vpp_config()
 223         pppoe_if.set_unnumbered(self.pg0.sw_if_index)
 224
 225         #
 226         # Send tunneled packets that match the created tunnel and
 227         # are decapped and forwarded
 228         #
 229         tx2 = self.create_stream_pppoe_ip4(
 230             self.pg0,
 231             self.pg1,
 232             self.pg0.remote_mac,
 233             self.session_id,
 234             self.pg0.remote_ip4,
 235         )
 236         self.pg0.add_stream(tx2)
 237
 238         self.pg_enable_capture(self.pg_interfaces)
 239         self.pg_start()
 240
 241         rx2 = self.pg1.get_capture(len(tx2))
 242         self.verify_decapped_pppoe(self.pg0, rx2, tx2)
 243
 244         self.logger.info(self.vapi.cli("show pppoe fib"))
 245         self.logger.info(self.vapi.cli("show pppoe session"))
 246         self.logger.info(self.vapi.cli("show ip fib"))
 247
 248         #
 249         # test case cleanup
 250         #
 251
 252         # Delete PPPoE session
 253         pppoe_if.remove_vpp_config()
 254
 255         # Delete a route that resolves the server's destination
 256         route_sever_dst.remove_vpp_config()
 257
 258     def test_PPPoE_Encap(self):
 259         """PPPoE Encap Test"""
 260
 261         self.vapi.cli("clear trace")
 262
 263         #
 264         # Add a route that resolves the server's destination
 265         #
 266         route_sever_dst = VppIpRoute(
 267             self,
 268             "100.1.1.100",
 269             32,
 270             [VppRoutePath(self.pg1.remote_ip4, self.pg1.sw_if_index)],
 271         )
 272         route_sever_dst.add_vpp_config()
 273
 274         # Send PPPoE Discovery
 275         tx0 = self.create_stream_pppoe_discovery(
 276             self.pg0, self.pg1, self.pg0.remote_mac
 277         )
 278         self.pg0.add_stream(tx0)
 279         self.pg_start()
 280
 281         # Send PPPoE PPP LCP
 282         tx1 = self.create_stream_pppoe_lcp(
 283             self.pg0, self.pg1, self.pg0.remote_mac, self.session_id
 284         )
 285         self.pg0.add_stream(tx1)
 286         self.pg_start()
 287
 288         # Create PPPoE session
 289         pppoe_if = VppPppoeInterface(
 290             self, self.pg0.remote_ip4, self.pg0.remote_mac, self.session_id
 291         )
 292         pppoe_if.add_vpp_config()
 293         pppoe_if.set_unnumbered(self.pg0.sw_if_index)
 294
 295         #
 296         # Send a packet stream that is routed into the session
 297         #  - packets are PPPoE encapped
 298         #
 299         self.vapi.cli("clear trace")
 300         tx2 = self.create_stream_ip4(
 301             self.pg1, self.pg0, self.pg0.remote_ip4, self.dst_ip, 65
 302         )
 303         self.pg1.add_stream(tx2)
 304
 305         self.pg_enable_capture(self.pg_interfaces)
 306         self.pg_start()
 307
 308         rx2 = self.pg0.get_capture(len(tx2))
 309         self.verify_encaped_pppoe(self.pg1, rx2, tx2, self.session_id)
 310
 311         self.logger.info(self.vapi.cli("show pppoe fib"))
 312         self.logger.info(self.vapi.cli("show pppoe session"))
 313         self.logger.info(self.vapi.cli("show ip fib"))
 314         self.logger.info(self.vapi.cli("show adj"))
 315
 316         #
 317         # test case cleanup
 318         #
 319
 320         # Delete PPPoE session
 321         pppoe_if.remove_vpp_config()
 322
 323         # Delete a route that resolves the server's destination
 324         route_sever_dst.remove_vpp_config()
 325
 326     def test_PPPoE_Add_Twice(self):
 327         """PPPoE Add Same Session Twice Test"""
 328
 329         self.vapi.cli("clear trace")
 330
 331         #
 332         # Add a route that resolves the server's destination
 333         #
 334         route_sever_dst = VppIpRoute(
 335             self,
 336             "100.1.1.100",
 337             32,
 338             [VppRoutePath(self.pg1.remote_ip4, self.pg1.sw_if_index)],
 339         )
 340         route_sever_dst.add_vpp_config()
 341
 342         # Send PPPoE Discovery
 343         tx0 = self.create_stream_pppoe_discovery(
 344             self.pg0, self.pg1, self.pg0.remote_mac
 345         )
 346         self.pg0.add_stream(tx0)
 347         self.pg_start()
 348
 349         # Send PPPoE PPP LCP
 350         tx1 = self.create_stream_pppoe_lcp(
 351             self.pg0, self.pg1, self.pg0.remote_mac, self.session_id
 352         )
 353         self.pg0.add_stream(tx1)
 354         self.pg_start()
 355
 356         # Create PPPoE session
 357         pppoe_if = VppPppoeInterface(
 358             self, self.pg0.remote_ip4, self.pg0.remote_mac, self.session_id
 359         )
 360         pppoe_if.add_vpp_config()
 361         pppoe_if.set_unnumbered(self.pg0.sw_if_index)
 362
 363         #
 364         # The double create (create the same session twice) should fail,
 365         # and we should still be able to use the original
 366         #
 367         try:
 368             pppoe_if.add_vpp_config()
 369         except Exception:
 370             pass
 371         else:
 372             self.fail("Double GRE tunnel add does not fail")
 373
 374         #
 375         # test case cleanup
 376         #
 377
 378         # Delete PPPoE session
 379         pppoe_if.remove_vpp_config()
 380
 381         # Delete a route that resolves the server's destination
 382         route_sever_dst.remove_vpp_config()
 383
 384     def test_PPPoE_Del_Twice(self):
 385         """PPPoE Delete Same Session Twice Test"""
 386
 387         self.vapi.cli("clear trace")
 388
 389         #
 390         # Add a route that resolves the server's destination
 391         #
 392         route_sever_dst = VppIpRoute(
 393             self,
 394             "100.1.1.100",
 395             32,
 396             [VppRoutePath(self.pg1.remote_ip4, self.pg1.sw_if_index)],
 397         )
 398         route_sever_dst.add_vpp_config()
 399
 400         # Send PPPoE Discovery
 401         tx0 = self.create_stream_pppoe_discovery(
 402             self.pg0, self.pg1, self.pg0.remote_mac
 403         )
 404         self.pg0.add_stream(tx0)
 405         self.pg_start()
 406
 407         # Send PPPoE PPP LCP
 408         tx1 = self.create_stream_pppoe_lcp(
 409             self.pg0, self.pg1, self.pg0.remote_mac, self.session_id
 410         )
 411         self.pg0.add_stream(tx1)
 412         self.pg_start()
 413
 414         # Create PPPoE session
 415         pppoe_if = VppPppoeInterface(
 416             self, self.pg0.remote_ip4, self.pg0.remote_mac, self.session_id
 417         )
 418         pppoe_if.add_vpp_config()
 419
 420         # Delete PPPoE session
 421         pppoe_if.remove_vpp_config()
 422
 423         #
 424         # The double del (del the same session twice) should fail,
 425         # and we should still be able to use the original
 426         #
 427         try:
 428             pppoe_if.remove_vpp_config()
 429         except Exception:
 430             pass
 431         else:
 432             self.fail("Double GRE tunnel del does not fail")
 433
 434         #
 435         # test case cleanup
 436         #
 437
 438         # Delete a route that resolves the server's destination
 439         route_sever_dst.remove_vpp_config()
 440
 441     def test_PPPoE_Decap_Multiple(self):
 442         """PPPoE Decap Multiple Sessions Test"""
 443
 444         self.vapi.cli("clear trace")
 445
 446         #
 447         # Add a route that resolves the server's destination
 448         #
 449         route_sever_dst = VppIpRoute(
 450             self,
 451             "100.1.1.100",
 452             32,
 453             [VppRoutePath(self.pg1.remote_ip4, self.pg1.sw_if_index)],
 454         )
 455         route_sever_dst.add_vpp_config()
 456
 457         # Send PPPoE Discovery 1
 458         tx0 = self.create_stream_pppoe_discovery(
 459             self.pg0, self.pg1, self.pg0.remote_mac
 460         )
 461         self.pg0.add_stream(tx0)
 462         self.pg_start()
 463
 464         # Send PPPoE PPP LCP 1
 465         tx1 = self.create_stream_pppoe_lcp(
 466             self.pg0, self.pg1, self.pg0.remote_mac, self.session_id
 467         )
 468         self.pg0.add_stream(tx1)
 469         self.pg_start()
 470
 471         # Create PPPoE session 1
 472         pppoe_if1 = VppPppoeInterface(
 473             self, self.pg0.remote_ip4, self.pg0.remote_mac, self.session_id
 474         )
 475         pppoe_if1.add_vpp_config()
 476         pppoe_if1.set_unnumbered(self.pg0.sw_if_index)
 477
 478         # Send PPPoE Discovery 2
 479         tx3 = self.create_stream_pppoe_discovery(
 480             self.pg2, self.pg1, self.pg2.remote_mac
 481         )
 482         self.pg2.add_stream(tx3)
 483         self.pg_start()
 484
 485         # Send PPPoE PPP LCP 2
 486         tx4 = self.create_stream_pppoe_lcp(
 487             self.pg2, self.pg1, self.pg2.remote_mac, self.session_id + 1
 488         )
 489         self.pg2.add_stream(tx4)
 490         self.pg_start()
 491
 492         # Create PPPoE session 2
 493         pppoe_if2 = VppPppoeInterface(
 494             self, self.pg2.remote_ip4, self.pg2.remote_mac, self.session_id + 1
 495         )
 496         pppoe_if2.add_vpp_config()
 497         pppoe_if2.set_unnumbered(self.pg0.sw_if_index)
 498
 499         #
 500         # Send tunneled packets that match the created tunnel and
 501         # are decapped and forwarded
 502         #
 503         tx2 = self.create_stream_pppoe_ip4(
 504             self.pg0,
 505             self.pg1,
 506             self.pg0.remote_mac,
 507             self.session_id,
 508             self.pg0.remote_ip4,
 509         )
 510         self.pg0.add_stream(tx2)
 511
 512         self.pg_enable_capture(self.pg_interfaces)
 513         self.pg_start()
 514
 515         rx2 = self.pg1.get_capture(len(tx2))
 516         self.verify_decapped_pppoe(self.pg0, rx2, tx2)
 517
 518         tx5 = self.create_stream_pppoe_ip4(
 519             self.pg2,
 520             self.pg1,
 521             self.pg2.remote_mac,
 522             self.session_id + 1,
 523             self.pg2.remote_ip4,
 524         )
 525         self.pg2.add_stream(tx5)
 526
 527         self.pg_enable_capture(self.pg_interfaces)
 528         self.pg_start()
 529
 530         rx5 = self.pg1.get_capture(len(tx5))
 531         self.verify_decapped_pppoe(self.pg2, rx5, tx5)
 532
 533         self.logger.info(self.vapi.cli("show pppoe fib"))
 534         self.logger.info(self.vapi.cli("show pppoe session"))
 535         self.logger.info(self.vapi.cli("show ip fib"))
 536
 537         #
 538         # test case cleanup
 539         #
 540
 541         # Delete PPPoE session
 542         pppoe_if1.remove_vpp_config()
 543         pppoe_if2.remove_vpp_config()
 544
 545         # Delete a route that resolves the server's destination
 546         route_sever_dst.remove_vpp_config()
 547
 548     def test_PPPoE_Encap_Multiple(self):
 549         """PPPoE Encap Multiple Sessions Test"""
 550
 551         self.vapi.cli("clear trace")
 552
 553         #
 554         # Add a route that resolves the server's destination
 555         #
 556         route_sever_dst = VppIpRoute(
 557             self,
 558             "100.1.1.100",
 559             32,
 560             [VppRoutePath(self.pg1.remote_ip4, self.pg1.sw_if_index)],
 561         )
 562         route_sever_dst.add_vpp_config()
 563
 564         # Send PPPoE Discovery 1
 565         tx0 = self.create_stream_pppoe_discovery(
 566             self.pg0, self.pg1, self.pg0.remote_mac
 567         )
 568         self.pg0.add_stream(tx0)
 569         self.pg_start()
 570
 571         # Send PPPoE PPP LCP 1
 572         tx1 = self.create_stream_pppoe_lcp(
 573             self.pg0, self.pg1, self.pg0.remote_mac, self.session_id
 574         )
 575         self.pg0.add_stream(tx1)
 576         self.pg_start()
 577
 578         # Create PPPoE session 1
 579         pppoe_if1 = VppPppoeInterface(
 580             self, self.pg0.remote_ip4, self.pg0.remote_mac, self.session_id
 581         )
 582         pppoe_if1.add_vpp_config()
 583
 584         # Send PPPoE Discovery 2
 585         tx3 = self.create_stream_pppoe_discovery(
 586             self.pg2, self.pg1, self.pg2.remote_mac
 587         )
 588         self.pg2.add_stream(tx3)
 589         self.pg_start()
 590
 591         # Send PPPoE PPP LCP 2
 592         tx4 = self.create_stream_pppoe_lcp(
 593             self.pg2, self.pg1, self.pg2.remote_mac, self.session_id + 1
 594         )
 595         self.pg2.add_stream(tx4)
 596         self.pg_start()
 597
 598         # Create PPPoE session 2
 599         pppoe_if2 = VppPppoeInterface(
 600             self, self.pg2.remote_ip4, self.pg2.remote_mac, self.session_id + 1
 601         )
 602         pppoe_if2.add_vpp_config()
 603
 604         #
 605         # Send a packet stream that is routed into the session
 606         #  - packets are PPPoE encapped
 607         #
 608         self.vapi.cli("clear trace")
 609         tx2 = self.create_stream_ip4(
 610             self.pg1, self.pg0, self.pg0.remote_ip4, self.dst_ip
 611         )
 612         self.pg1.add_stream(tx2)
 613
 614         self.pg_enable_capture(self.pg_interfaces)
 615         self.pg_start()
 616
 617         rx2 = self.pg0.get_capture(len(tx2))
 618         self.verify_encaped_pppoe(self.pg1, rx2, tx2, self.session_id)
 619
 620         tx5 = self.create_stream_ip4(
 621             self.pg1, self.pg2, self.pg2.remote_ip4, self.dst_ip
 622         )
 623         self.pg1.add_stream(tx5)
 624
 625         self.pg_enable_capture(self.pg_interfaces)
 626         self.pg_start()
 627
 628         rx5 = self.pg2.get_capture(len(tx5))
 629         self.verify_encaped_pppoe(self.pg1, rx5, tx5, self.session_id + 1)
 630
 631         self.logger.info(self.vapi.cli("show pppoe fib"))
 632         self.logger.info(self.vapi.cli("show pppoe session"))
 633         self.logger.info(self.vapi.cli("show ip fib"))
 634
 635         #
 636         # test case cleanup
 637         #
 638
 639         # Delete PPPoE session
 640         pppoe_if1.remove_vpp_config()
 641         pppoe_if2.remove_vpp_config()
 642
 643         # Delete a route that resolves the server's destination
 644         route_sever_dst.remove_vpp_config()
 645
 646
 647 if __name__ == "__main__":
 648     unittest.main(testRunner=VppTestRunner)