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