test/test_classifier.py

   1 #!/usr/bin/env python
   2
   3 import binascii
   4 import socket
   5 import unittest
   6
   7 from framework import VppTestCase, VppTestRunner
   8
   9 from scapy.packet import Raw
  10 from scapy.layers.l2 import Ether
  11 from scapy.layers.inet import IP, UDP, TCP
  12 from util import ppp
  13 from vpp_ip_route import VppIpRoute, VppRoutePath
  14 from vpp_ip import INVALID_INDEX
  15
  16
  17 class TestClassifier(VppTestCase):
  18     """ Classifier Test Case """
  19
  20     @classmethod
  21     def setUpClass(cls):
  22         """
  23         Perform standard class setup (defined by class method setUpClass in
  24         class VppTestCase) before running the test case, set test case related
  25         variables and configure VPP.
  26         """
  27         super(TestClassifier, cls).setUpClass()
  28         cls.acl_active_table = ''
  29
  30     @classmethod
  31     def tearDownClass(cls):
  32         super(TestClassifier, cls).tearDownClass()
  33
  34     def setUp(self):
  35         """
  36         Perform test setup before test case.
  37
  38         **Config:**
  39             - create 4 pg interfaces
  40                 - untagged pg0/pg1/pg2 interface
  41                     pg0 -------> pg1 (IP ACL)
  42                            \
  43                             ---> pg2 (MAC ACL))
  44                              \
  45                               -> pg3 (PBR)
  46             - setup interfaces:
  47                 - put it into UP state
  48                 - set IPv4 addresses
  49                 - resolve neighbor address using ARP
  50
  51         :ivar list interfaces: pg interfaces.
  52         :ivar list pg_if_packet_sizes: packet sizes in test.
  53         :ivar dict acl_tbl_idx: ACL table index.
  54         :ivar int pbr_vrfid: VRF id for PBR test.
  55         """
  56         self.reset_packet_infos()
  57         super(TestClassifier, self).setUp()
  58
  59         # create 4 pg interfaces
  60         self.create_pg_interfaces(range(4))
  61
  62         # packet sizes to test
  63         self.pg_if_packet_sizes = [64, 9018]
  64
  65         self.interfaces = list(self.pg_interfaces)
  66
  67         # ACL & PBR vars
  68         self.acl_tbl_idx = {}
  69         self.pbr_vrfid = 200
  70
  71         # setup all interfaces
  72         for intf in self.interfaces:
  73             intf.admin_up()
  74             intf.config_ip4()
  75             intf.resolve_arp()
  76
  77     def tearDown(self):
  78         """Run standard test teardown and acl related log."""
  79         if not self.vpp_dead:
  80             if self.acl_active_table == 'ip_out':
  81                 self.output_acl_set_interface(
  82                     self.pg0, self.acl_tbl_idx.get(self.acl_active_table), 0)
  83                 self.acl_active_table = ''
  84             elif self.acl_active_table != '':
  85                 self.input_acl_set_interface(
  86                     self.pg0, self.acl_tbl_idx.get(self.acl_active_table), 0)
  87                 self.acl_active_table = ''
  88             for intf in self.interfaces:
  89                 intf.unconfig_ip4()
  90                 intf.admin_down()
  91
  92         super(TestClassifier, self).tearDown()
  93
  94     def show_commands_at_teardown(self):
  95         self.logger.info(self.vapi.ppcli("show inacl type ip4"))
  96         self.logger.info(self.vapi.ppcli("show outacl type ip4"))
  97         self.logger.info(self.vapi.cli("show classify table verbose"))
  98         self.logger.info(self.vapi.cli("show ip fib"))
  99
 100     def create_stream(self, src_if, dst_if, packet_sizes,
 101                       proto_l=UDP(sport=1234, dport=5678)):
 102         """Create input packet stream for defined interfaces.
 103
 104         :param VppInterface src_if: Source Interface for packet stream.
 105         :param VppInterface dst_if: Destination Interface for packet stream.
 106         :param list packet_sizes: packet size to test.
 107         :param Scapy proto_l: Required IP protocol. Default protocol is UDP.
 108         """
 109         pkts = []
 110
 111         for size in packet_sizes:
 112             info = self.create_packet_info(src_if, dst_if)
 113             payload = self.info_to_payload(info)
 114             p = (Ether(dst=src_if.local_mac, src=src_if.remote_mac) /
 115                  IP(src=src_if.remote_ip4, dst=dst_if.remote_ip4) /
 116                  proto_l /
 117                  Raw(payload))
 118             info.data = p.copy()
 119             self.extend_packet(p, size)
 120             pkts.append(p)
 121         return pkts
 122
 123     def verify_capture(self, dst_if, capture, proto_l=UDP):
 124         """Verify captured input packet stream for defined interface.
 125
 126         :param VppInterface dst_if: Interface to verify captured packet stream.
 127         :param list capture: Captured packet stream.
 128         :param Scapy proto_l: Required IP protocol. Default protocol is UDP.
 129         """
 130         self.logger.info("Verifying capture on interface %s" % dst_if.name)
 131         last_info = dict()
 132         for i in self.interfaces:
 133             last_info[i.sw_if_index] = None
 134         dst_sw_if_index = dst_if.sw_if_index
 135         for packet in capture:
 136             try:
 137                 ip_received = packet[IP]
 138                 proto_received = packet[proto_l]
 139                 payload_info = self.payload_to_info(packet[Raw])
 140                 packet_index = payload_info.index
 141                 self.assertEqual(payload_info.dst, dst_sw_if_index)
 142                 self.logger.debug(
 143                     "Got packet on port %s: src=%u (id=%u)" %
 144                     (dst_if.name, payload_info.src, packet_index))
 145                 next_info = self.get_next_packet_info_for_interface2(
 146                     payload_info.src, dst_sw_if_index,
 147                     last_info[payload_info.src])
 148                 last_info[payload_info.src] = next_info
 149                 self.assertTrue(next_info is not None)
 150                 self.assertEqual(packet_index, next_info.index)
 151                 saved_packet = next_info.data
 152                 ip_saved = saved_packet[IP]
 153                 proto_saved = saved_packet[proto_l]
 154                 # Check standard fields
 155                 self.assertEqual(ip_received.src, ip_saved.src)
 156                 self.assertEqual(ip_received.dst, ip_saved.dst)
 157                 self.assertEqual(proto_received.sport, proto_saved.sport)
 158                 self.assertEqual(proto_received.dport, proto_saved.dport)
 159             except:
 160                 self.logger.error(ppp("Unexpected or invalid packet:", packet))
 161                 raise
 162         for i in self.interfaces:
 163             remaining_packet = self.get_next_packet_info_for_interface2(
 164                 i.sw_if_index, dst_sw_if_index, last_info[i.sw_if_index])
 165             self.assertTrue(remaining_packet is None,
 166                             "Interface %s: Packet expected from interface %s "
 167                             "didn't arrive" % (dst_if.name, i.name))
 168
 169     def verify_vrf(self, vrf_id):
 170         """
 171         Check if the FIB table / VRF ID is configured.
 172
 173         :param int vrf_id: The FIB table / VRF ID to be verified.
 174         :return: 1 if the FIB table / VRF ID is configured, otherwise return 0.
 175         """
 176         ip_fib_dump = self.vapi.ip_route_dump(vrf_id, False)
 177         vrf_count = len(ip_fib_dump)
 178
 179         if vrf_count == 0:
 180             self.logger.info("IPv4 VRF ID %d is not configured" % vrf_id)
 181             return 0
 182         else:
 183             self.logger.info("IPv4 VRF ID %d is configured" % vrf_id)
 184             return 1
 185
 186     @staticmethod
 187     def build_ip_mask(proto='', src_ip='', dst_ip='',
 188                       src_port='', dst_port=''):
 189         """Build IP ACL mask data with hexstring format.
 190
 191         :param str proto: protocol number <0-ff>
 192         :param str src_ip: source ip address <0-ffffffff>
 193         :param str dst_ip: destination ip address <0-ffffffff>
 194         :param str src_port: source port number <0-ffff>
 195         :param str dst_port: destination port number <0-ffff>
 196         """
 197
 198         return ('{!s:0>20}{!s:0>12}{!s:0>8}{!s:0>4}{!s:0>4}'.format(
 199             proto, src_ip, dst_ip, src_port, dst_port)).rstrip('0')
 200
 201     @staticmethod
 202     def build_ip_match(proto=0, src_ip='', dst_ip='',
 203                        src_port=0, dst_port=0):
 204         """Build IP ACL match data with hexstring format.
 205
 206         :param int proto: protocol number with valid option "x"
 207         :param str src_ip: source ip address with format of "x.x.x.x"
 208         :param str dst_ip: destination ip address with format of "x.x.x.x"
 209         :param int src_port: source port number "x"
 210         :param int dst_port: destination port number "x"
 211         """
 212         if src_ip:
 213             src_ip = binascii.hexlify(socket.inet_aton(src_ip))
 214         if dst_ip:
 215             dst_ip = binascii.hexlify(socket.inet_aton(dst_ip))
 216
 217         return ('{!s:0>20}{!s:0>12}{!s:0>8}{!s:0>4}{!s:0>4}'.format(
 218             hex(proto)[2:], src_ip, dst_ip, hex(src_port)[2:],
 219             hex(dst_port)[2:])).rstrip('0')
 220
 221     @staticmethod
 222     def build_mac_mask(dst_mac='', src_mac='', ether_type=''):
 223         """Build MAC ACL mask data with hexstring format.
 224
 225         :param str dst_mac: source MAC address <0-ffffffffffff>
 226         :param str src_mac: destination MAC address <0-ffffffffffff>
 227         :param str ether_type: ethernet type <0-ffff>
 228         """
 229
 230         return ('{!s:0>12}{!s:0>12}{!s:0>4}'.format(
 231             dst_mac, src_mac, ether_type)).rstrip('0')
 232
 233     @staticmethod
 234     def build_mac_match(dst_mac='', src_mac='', ether_type=''):
 235         """Build MAC ACL match data with hexstring format.
 236
 237         :param str dst_mac: source MAC address <x:x:x:x:x:x>
 238         :param str src_mac: destination MAC address <x:x:x:x:x:x>
 239         :param str ether_type: ethernet type <0-ffff>
 240         """
 241         if dst_mac:
 242             dst_mac = dst_mac.replace(':', '')
 243         if src_mac:
 244             src_mac = src_mac.replace(':', '')
 245
 246         return ('{!s:0>12}{!s:0>12}{!s:0>4}'.format(
 247             dst_mac, src_mac, ether_type)).rstrip('0')
 248
 249     def create_classify_table(self, key, mask, data_offset=0):
 250         """Create Classify Table
 251
 252         :param str key: key for classify table (ex, ACL name).
 253         :param str mask: mask value for interested traffic.
 254         :param int data_offset:
 255         """
 256         r = self.vapi.classify_add_del_table(
 257             is_add=1,
 258             mask=binascii.unhexlify(mask),
 259             match_n_vectors=(len(mask) - 1) // 32 + 1,
 260             miss_next_index=0,
 261             current_data_flag=1,
 262             current_data_offset=data_offset)
 263         self.assertIsNotNone(r, 'No response msg for add_del_table')
 264         self.acl_tbl_idx[key] = r.new_table_index
 265
 266     def create_classify_session(self, table_index, match, pbr_option=0,
 267                                 vrfid=0, is_add=1):
 268         """Create Classify Session
 269
 270         :param int table_index: table index to identify classify table.
 271         :param str match: matched value for interested traffic.
 272         :param int pbr_option: enable/disable PBR feature.
 273         :param int vrfid: VRF id.
 274         :param int is_add: option to configure classify session.
 275             - create(1) or delete(0)
 276         """
 277         r = self.vapi.classify_add_del_session(
 278             is_add,
 279             table_index,
 280             binascii.unhexlify(match),
 281             opaque_index=0,
 282             action=pbr_option,
 283             metadata=vrfid)
 284         self.assertIsNotNone(r, 'No response msg for add_del_session')
 285
 286     def input_acl_set_interface(self, intf, table_index, is_add=1):
 287         """Configure Input ACL interface
 288
 289         :param VppInterface intf: Interface to apply Input ACL feature.
 290         :param int table_index: table index to identify classify table.
 291         :param int is_add: option to configure classify session.
 292             - enable(1) or disable(0)
 293         """
 294         r = self.vapi.input_acl_set_interface(
 295             is_add,
 296             intf.sw_if_index,
 297             ip4_table_index=table_index)
 298         self.assertIsNotNone(r, 'No response msg for acl_set_interface')
 299
 300     def output_acl_set_interface(self, intf, table_index, is_add=1):
 301         """Configure Output ACL interface
 302
 303         :param VppInterface intf: Interface to apply Output ACL feature.
 304         :param int table_index: table index to identify classify table.
 305         :param int is_add: option to configure classify session.
 306             - enable(1) or disable(0)
 307         """
 308         r = self.vapi.output_acl_set_interface(
 309             is_add,
 310             intf.sw_if_index,
 311             ip4_table_index=table_index)
 312         self.assertIsNotNone(r, 'No response msg for acl_set_interface')
 313
 314
 315 # Tests split to different test case classes because of issue reported in
 316 # ticket VPP-1336
 317 class TestClassifierIP(TestClassifier):
 318     """ Classifier IP Test Case """
 319
 320     @classmethod
 321     def setUpClass(cls):
 322         super(TestClassifierIP, cls).setUpClass()
 323
 324     @classmethod
 325     def tearDownClass(cls):
 326         super(TestClassifierIP, cls).tearDownClass()
 327
 328     def test_iacl_src_ip(self):
 329         """ Source IP iACL test
 330
 331         Test scenario for basic IP ACL with source IP
 332             - Create IPv4 stream for pg0 -> pg1 interface.
 333             - Create iACL with source IP address.
 334             - Send and verify received packets on pg1 interface.
 335         """
 336
 337         # Basic iACL testing with source IP
 338         pkts = self.create_stream(self.pg0, self.pg1, self.pg_if_packet_sizes)
 339         self.pg0.add_stream(pkts)
 340
 341         key = 'ip_src'
 342         self.create_classify_table(key, self.build_ip_mask(src_ip='ffffffff'))
 343         self.create_classify_session(
 344             self.acl_tbl_idx.get(key),
 345             self.build_ip_match(src_ip=self.pg0.remote_ip4))
 346         self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get(key))
 347         self.acl_active_table = key
 348
 349         self.pg_enable_capture(self.pg_interfaces)
 350         self.pg_start()
 351
 352         pkts = self.pg1.get_capture(len(pkts))
 353         self.verify_capture(self.pg1, pkts)
 354         self.pg0.assert_nothing_captured(remark="packets forwarded")
 355         self.pg2.assert_nothing_captured(remark="packets forwarded")
 356         self.pg3.assert_nothing_captured(remark="packets forwarded")
 357
 358     def test_iacl_dst_ip(self):
 359         """ Destination IP iACL test
 360
 361         Test scenario for basic IP ACL with destination IP
 362             - Create IPv4 stream for pg0 -> pg1 interface.
 363             - Create iACL with destination IP address.
 364             - Send and verify received packets on pg1 interface.
 365         """
 366
 367         # Basic iACL testing with destination IP
 368         pkts = self.create_stream(self.pg0, self.pg1, self.pg_if_packet_sizes)
 369         self.pg0.add_stream(pkts)
 370
 371         key = 'ip_dst'
 372         self.create_classify_table(key, self.build_ip_mask(dst_ip='ffffffff'))
 373         self.create_classify_session(
 374             self.acl_tbl_idx.get(key),
 375             self.build_ip_match(dst_ip=self.pg1.remote_ip4))
 376         self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get(key))
 377         self.acl_active_table = key
 378
 379         self.pg_enable_capture(self.pg_interfaces)
 380         self.pg_start()
 381
 382         pkts = self.pg1.get_capture(len(pkts))
 383         self.verify_capture(self.pg1, pkts)
 384         self.pg0.assert_nothing_captured(remark="packets forwarded")
 385         self.pg2.assert_nothing_captured(remark="packets forwarded")
 386         self.pg3.assert_nothing_captured(remark="packets forwarded")
 387
 388     def test_iacl_src_dst_ip(self):
 389         """ Source and destination IP iACL test
 390
 391         Test scenario for basic IP ACL with source and destination IP
 392             - Create IPv4 stream for pg0 -> pg1 interface.
 393             - Create iACL with source and destination IP addresses.
 394             - Send and verify received packets on pg1 interface.
 395         """
 396
 397         # Basic iACL testing with source and destination IP
 398         pkts = self.create_stream(self.pg0, self.pg1, self.pg_if_packet_sizes)
 399         self.pg0.add_stream(pkts)
 400
 401         key = 'ip'
 402         self.create_classify_table(
 403             key, self.build_ip_mask(src_ip='ffffffff', dst_ip='ffffffff'))
 404         self.create_classify_session(
 405             self.acl_tbl_idx.get(key),
 406             self.build_ip_match(src_ip=self.pg0.remote_ip4,
 407                                 dst_ip=self.pg1.remote_ip4))
 408         self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get(key))
 409         self.acl_active_table = key
 410
 411         self.pg_enable_capture(self.pg_interfaces)
 412         self.pg_start()
 413
 414         pkts = self.pg1.get_capture(len(pkts))
 415         self.verify_capture(self.pg1, pkts)
 416         self.pg0.assert_nothing_captured(remark="packets forwarded")
 417         self.pg2.assert_nothing_captured(remark="packets forwarded")
 418         self.pg3.assert_nothing_captured(remark="packets forwarded")
 419
 420
 421 class TestClassifierUDP(TestClassifier):
 422     """ Classifier UDP proto Test Case """
 423
 424     @classmethod
 425     def setUpClass(cls):
 426         super(TestClassifierUDP, cls).setUpClass()
 427
 428     @classmethod
 429     def tearDownClass(cls):
 430         super(TestClassifierUDP, cls).tearDownClass()
 431
 432     def test_iacl_proto_udp(self):
 433         """ UDP protocol iACL test
 434
 435         Test scenario for basic protocol ACL with UDP protocol
 436             - Create IPv4 stream for pg0 -> pg1 interface.
 437             - Create iACL with UDP IP protocol.
 438             - Send and verify received packets on pg1 interface.
 439         """
 440
 441         # Basic iACL testing with UDP protocol
 442         pkts = self.create_stream(self.pg0, self.pg1, self.pg_if_packet_sizes)
 443         self.pg0.add_stream(pkts)
 444
 445         key = 'proto_udp'
 446         self.create_classify_table(key, self.build_ip_mask(proto='ff'))
 447         self.create_classify_session(
 448             self.acl_tbl_idx.get(key),
 449             self.build_ip_match(proto=socket.IPPROTO_UDP))
 450         self.input_acl_set_interface(
 451             self.pg0, self.acl_tbl_idx.get(key))
 452         self.acl_active_table = key
 453
 454         self.pg_enable_capture(self.pg_interfaces)
 455         self.pg_start()
 456
 457         pkts = self.pg1.get_capture(len(pkts))
 458         self.verify_capture(self.pg1, pkts)
 459         self.pg0.assert_nothing_captured(remark="packets forwarded")
 460         self.pg2.assert_nothing_captured(remark="packets forwarded")
 461         self.pg3.assert_nothing_captured(remark="packets forwarded")
 462
 463     def test_iacl_proto_udp_sport(self):
 464         """ UDP source port iACL test
 465
 466         Test scenario for basic protocol ACL with UDP and sport
 467             - Create IPv4 stream for pg0 -> pg1 interface.
 468             - Create iACL with UDP IP protocol and defined sport.
 469             - Send and verify received packets on pg1 interface.
 470         """
 471
 472         # Basic iACL testing with UDP and sport
 473         sport = 38
 474         pkts = self.create_stream(self.pg0, self.pg1, self.pg_if_packet_sizes,
 475                                   UDP(sport=sport, dport=5678))
 476         self.pg0.add_stream(pkts)
 477
 478         key = 'proto_udp_sport'
 479         self.create_classify_table(
 480             key, self.build_ip_mask(proto='ff', src_port='ffff'))
 481         self.create_classify_session(
 482             self.acl_tbl_idx.get(key),
 483             self.build_ip_match(proto=socket.IPPROTO_UDP, src_port=sport))
 484         self.input_acl_set_interface(
 485             self.pg0, self.acl_tbl_idx.get(key))
 486         self.acl_active_table = key
 487
 488         self.pg_enable_capture(self.pg_interfaces)
 489         self.pg_start()
 490
 491         pkts = self.pg1.get_capture(len(pkts))
 492         self.verify_capture(self.pg1, pkts)
 493         self.pg0.assert_nothing_captured(remark="packets forwarded")
 494         self.pg2.assert_nothing_captured(remark="packets forwarded")
 495         self.pg3.assert_nothing_captured(remark="packets forwarded")
 496
 497     def test_iacl_proto_udp_dport(self):
 498         """ UDP destination port iACL test
 499
 500         Test scenario for basic protocol ACL with UDP and dport
 501             - Create IPv4 stream for pg0 -> pg1 interface.
 502             - Create iACL with UDP IP protocol and defined dport.
 503             - Send and verify received packets on pg1 interface.
 504         """
 505
 506         # Basic iACL testing with UDP and dport
 507         dport = 427
 508         pkts = self.create_stream(self.pg0, self.pg1, self.pg_if_packet_sizes,
 509                                   UDP(sport=1234, dport=dport))
 510         self.pg0.add_stream(pkts)
 511
 512         key = 'proto_udp_dport'
 513         self.create_classify_table(
 514             key, self.build_ip_mask(proto='ff', dst_port='ffff'))
 515         self.create_classify_session(
 516             self.acl_tbl_idx.get(key),
 517             self.build_ip_match(proto=socket.IPPROTO_UDP, dst_port=dport))
 518         self.input_acl_set_interface(
 519             self.pg0, self.acl_tbl_idx.get(key))
 520         self.acl_active_table = key
 521
 522         self.pg_enable_capture(self.pg_interfaces)
 523         self.pg_start()
 524
 525         pkts = self.pg1.get_capture(len(pkts))
 526         self.verify_capture(self.pg1, pkts)
 527         self.pg0.assert_nothing_captured(remark="packets forwarded")
 528         self.pg2.assert_nothing_captured(remark="packets forwarded")
 529         self.pg3.assert_nothing_captured(remark="packets forwarded")
 530
 531     def test_iacl_proto_udp_sport_dport(self):
 532         """ UDP source and destination ports iACL test
 533
 534         Test scenario for basic protocol ACL with UDP and sport and dport
 535             - Create IPv4 stream for pg0 -> pg1 interface.
 536             - Create iACL with UDP IP protocol and defined sport and dport.
 537             - Send and verify received packets on pg1 interface.
 538         """
 539
 540         # Basic iACL testing with UDP and sport and dport
 541         sport = 13720
 542         dport = 9080
 543         pkts = self.create_stream(self.pg0, self.pg1, self.pg_if_packet_sizes,
 544                                   UDP(sport=sport, dport=dport))
 545         self.pg0.add_stream(pkts)
 546
 547         key = 'proto_udp_ports'
 548         self.create_classify_table(
 549             key,
 550             self.build_ip_mask(proto='ff', src_port='ffff', dst_port='ffff'))
 551         self.create_classify_session(
 552             self.acl_tbl_idx.get(key),
 553             self.build_ip_match(proto=socket.IPPROTO_UDP, src_port=sport,
 554                                 dst_port=dport))
 555         self.input_acl_set_interface(
 556             self.pg0, self.acl_tbl_idx.get(key))
 557         self.acl_active_table = key
 558
 559         self.pg_enable_capture(self.pg_interfaces)
 560         self.pg_start()
 561
 562         pkts = self.pg1.get_capture(len(pkts))
 563         self.verify_capture(self.pg1, pkts)
 564         self.pg0.assert_nothing_captured(remark="packets forwarded")
 565         self.pg2.assert_nothing_captured(remark="packets forwarded")
 566         self.pg3.assert_nothing_captured(remark="packets forwarded")
 567
 568
 569 class TestClassifierTCP(TestClassifier):
 570     """ Classifier TCP proto Test Case """
 571
 572     @classmethod
 573     def setUpClass(cls):
 574         super(TestClassifierTCP, cls).setUpClass()
 575
 576     @classmethod
 577     def tearDownClass(cls):
 578         super(TestClassifierTCP, cls).tearDownClass()
 579
 580     def test_iacl_proto_tcp(self):
 581         """ TCP protocol iACL test
 582
 583         Test scenario for basic protocol ACL with TCP protocol
 584             - Create IPv4 stream for pg0 -> pg1 interface.
 585             - Create iACL with TCP IP protocol.
 586             - Send and verify received packets on pg1 interface.
 587         """
 588
 589         # Basic iACL testing with TCP protocol
 590         pkts = self.create_stream(self.pg0, self.pg1, self.pg_if_packet_sizes,
 591                                   TCP(sport=1234, dport=5678))
 592         self.pg0.add_stream(pkts)
 593
 594         key = 'proto_tcp'
 595         self.create_classify_table(key, self.build_ip_mask(proto='ff'))
 596         self.create_classify_session(
 597             self.acl_tbl_idx.get(key),
 598             self.build_ip_match(proto=socket.IPPROTO_TCP))
 599         self.input_acl_set_interface(
 600             self.pg0, self.acl_tbl_idx.get(key))
 601         self.acl_active_table = key
 602
 603         self.pg_enable_capture(self.pg_interfaces)
 604         self.pg_start()
 605
 606         pkts = self.pg1.get_capture(len(pkts))
 607         self.verify_capture(self.pg1, pkts, TCP)
 608         self.pg0.assert_nothing_captured(remark="packets forwarded")
 609         self.pg2.assert_nothing_captured(remark="packets forwarded")
 610         self.pg3.assert_nothing_captured(remark="packets forwarded")
 611
 612     def test_iacl_proto_tcp_sport(self):
 613         """ TCP source port iACL test
 614
 615         Test scenario for basic protocol ACL with TCP and sport
 616             - Create IPv4 stream for pg0 -> pg1 interface.
 617             - Create iACL with TCP IP protocol and defined sport.
 618             - Send and verify received packets on pg1 interface.
 619         """
 620
 621         # Basic iACL testing with TCP and sport
 622         sport = 38
 623         pkts = self.create_stream(self.pg0, self.pg1, self.pg_if_packet_sizes,
 624                                   TCP(sport=sport, dport=5678))
 625         self.pg0.add_stream(pkts)
 626
 627         key = 'proto_tcp_sport'
 628         self.create_classify_table(
 629             key, self.build_ip_mask(proto='ff', src_port='ffff'))
 630         self.create_classify_session(
 631             self.acl_tbl_idx.get(key),
 632             self.build_ip_match(proto=socket.IPPROTO_TCP, src_port=sport))
 633         self.input_acl_set_interface(
 634             self.pg0, self.acl_tbl_idx.get(key))
 635         self.acl_active_table = key
 636
 637         self.pg_enable_capture(self.pg_interfaces)
 638         self.pg_start()
 639
 640         pkts = self.pg1.get_capture(len(pkts))
 641         self.verify_capture(self.pg1, pkts, TCP)
 642         self.pg0.assert_nothing_captured(remark="packets forwarded")
 643         self.pg2.assert_nothing_captured(remark="packets forwarded")
 644         self.pg3.assert_nothing_captured(remark="packets forwarded")
 645
 646     def test_iacl_proto_tcp_dport(self):
 647         """ TCP destination port iACL test
 648
 649         Test scenario for basic protocol ACL with TCP and dport
 650             - Create IPv4 stream for pg0 -> pg1 interface.
 651             - Create iACL with TCP IP protocol and defined dport.
 652             - Send and verify received packets on pg1 interface.
 653         """
 654
 655         # Basic iACL testing with TCP and dport
 656         dport = 427
 657         pkts = self.create_stream(self.pg0, self.pg1, self.pg_if_packet_sizes,
 658                                   TCP(sport=1234, dport=dport))
 659         self.pg0.add_stream(pkts)
 660
 661         key = 'proto_tcp_sport'
 662         self.create_classify_table(
 663             key, self.build_ip_mask(proto='ff', dst_port='ffff'))
 664         self.create_classify_session(
 665             self.acl_tbl_idx.get(key),
 666             self.build_ip_match(proto=socket.IPPROTO_TCP, dst_port=dport))
 667         self.input_acl_set_interface(
 668             self.pg0, self.acl_tbl_idx.get(key))
 669         self.acl_active_table = key
 670
 671         self.pg_enable_capture(self.pg_interfaces)
 672         self.pg_start()
 673
 674         pkts = self.pg1.get_capture(len(pkts))
 675         self.verify_capture(self.pg1, pkts, TCP)
 676         self.pg0.assert_nothing_captured(remark="packets forwarded")
 677         self.pg2.assert_nothing_captured(remark="packets forwarded")
 678         self.pg3.assert_nothing_captured(remark="packets forwarded")
 679
 680     def test_iacl_proto_tcp_sport_dport(self):
 681         """ TCP source and destination ports iACL test
 682
 683         Test scenario for basic protocol ACL with TCP and sport and dport
 684             - Create IPv4 stream for pg0 -> pg1 interface.
 685             - Create iACL with TCP IP protocol and defined sport and dport.
 686             - Send and verify received packets on pg1 interface.
 687         """
 688
 689         # Basic iACL testing with TCP and sport and dport
 690         sport = 13720
 691         dport = 9080
 692         pkts = self.create_stream(self.pg0, self.pg1, self.pg_if_packet_sizes,
 693                                   TCP(sport=sport, dport=dport))
 694         self.pg0.add_stream(pkts)
 695
 696         key = 'proto_tcp_ports'
 697         self.create_classify_table(
 698             key,
 699             self.build_ip_mask(proto='ff', src_port='ffff', dst_port='ffff'))
 700         self.create_classify_session(
 701             self.acl_tbl_idx.get(key),
 702             self.build_ip_match(proto=socket.IPPROTO_TCP, src_port=sport,
 703                                 dst_port=dport))
 704         self.input_acl_set_interface(
 705             self.pg0, self.acl_tbl_idx.get(key))
 706         self.acl_active_table = key
 707
 708         self.pg_enable_capture(self.pg_interfaces)
 709         self.pg_start()
 710
 711         pkts = self.pg1.get_capture(len(pkts))
 712         self.verify_capture(self.pg1, pkts, TCP)
 713         self.pg0.assert_nothing_captured(remark="packets forwarded")
 714         self.pg2.assert_nothing_captured(remark="packets forwarded")
 715         self.pg3.assert_nothing_captured(remark="packets forwarded")
 716
 717
 718 class TestClassifierIPOut(TestClassifier):
 719     """ Classifier output IP Test Case """
 720
 721     @classmethod
 722     def setUpClass(cls):
 723         super(TestClassifierIPOut, cls).setUpClass()
 724
 725     @classmethod
 726     def tearDownClass(cls):
 727         super(TestClassifierIPOut, cls).tearDownClass()
 728
 729     def test_acl_ip_out(self):
 730         """ Output IP ACL test
 731
 732         Test scenario for basic IP ACL with source IP
 733             - Create IPv4 stream for pg1 -> pg0 interface.
 734             - Create ACL with source IP address.
 735             - Send and verify received packets on pg0 interface.
 736         """
 737
 738         # Basic oACL testing with source IP
 739         pkts = self.create_stream(self.pg1, self.pg0, self.pg_if_packet_sizes)
 740         self.pg1.add_stream(pkts)
 741
 742         key = 'ip_out'
 743         self.create_classify_table(
 744             key, self.build_ip_mask(src_ip='ffffffff'), data_offset=0)
 745         self.create_classify_session(
 746             self.acl_tbl_idx.get(key),
 747             self.build_ip_match(src_ip=self.pg1.remote_ip4))
 748         self.output_acl_set_interface(self.pg0, self.acl_tbl_idx.get(key))
 749         self.acl_active_table = key
 750
 751         self.pg_enable_capture(self.pg_interfaces)
 752         self.pg_start()
 753
 754         pkts = self.pg0.get_capture(len(pkts))
 755         self.verify_capture(self.pg0, pkts)
 756         self.pg1.assert_nothing_captured(remark="packets forwarded")
 757         self.pg2.assert_nothing_captured(remark="packets forwarded")
 758         self.pg3.assert_nothing_captured(remark="packets forwarded")
 759
 760
 761 class TestClassifierMAC(TestClassifier):
 762     """ Classifier MAC Test Case """
 763
 764     @classmethod
 765     def setUpClass(cls):
 766         super(TestClassifierMAC, cls).setUpClass()
 767
 768     @classmethod
 769     def tearDownClass(cls):
 770         super(TestClassifierMAC, cls).tearDownClass()
 771
 772     def test_acl_mac(self):
 773         """ MAC ACL test
 774
 775         Test scenario for basic MAC ACL with source MAC
 776             - Create IPv4 stream for pg0 -> pg2 interface.
 777             - Create ACL with source MAC address.
 778             - Send and verify received packets on pg2 interface.
 779         """
 780
 781         # Basic iACL testing with source MAC
 782         pkts = self.create_stream(self.pg0, self.pg2, self.pg_if_packet_sizes)
 783         self.pg0.add_stream(pkts)
 784
 785         key = 'mac'
 786         self.create_classify_table(
 787             key, self.build_mac_mask(src_mac='ffffffffffff'), data_offset=-14)
 788         self.create_classify_session(
 789             self.acl_tbl_idx.get(key),
 790             self.build_mac_match(src_mac=self.pg0.remote_mac))
 791         self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get(key))
 792         self.acl_active_table = key
 793
 794         self.pg_enable_capture(self.pg_interfaces)
 795         self.pg_start()
 796
 797         pkts = self.pg2.get_capture(len(pkts))
 798         self.verify_capture(self.pg2, pkts)
 799         self.pg0.assert_nothing_captured(remark="packets forwarded")
 800         self.pg1.assert_nothing_captured(remark="packets forwarded")
 801         self.pg3.assert_nothing_captured(remark="packets forwarded")
 802
 803
 804 class TestClassifierPBR(TestClassifier):
 805     """ Classifier PBR Test Case """
 806
 807     @classmethod
 808     def setUpClass(cls):
 809         super(TestClassifierPBR, cls).setUpClass()
 810
 811     @classmethod
 812     def tearDownClass(cls):
 813         super(TestClassifierPBR, cls).tearDownClass()
 814
 815     def test_acl_pbr(self):
 816         """ IP PBR test
 817
 818         Test scenario for PBR with source IP
 819             - Create IPv4 stream for pg0 -> pg3 interface.
 820             - Configure PBR fib entry for packet forwarding.
 821             - Send and verify received packets on pg3 interface.
 822         """
 823
 824         # PBR testing with source IP
 825         pkts = self.create_stream(self.pg0, self.pg3, self.pg_if_packet_sizes)
 826         self.pg0.add_stream(pkts)
 827
 828         key = 'pbr'
 829         self.create_classify_table(key, self.build_ip_mask(src_ip='ffffffff'))
 830         pbr_option = 1
 831         # this will create the VRF/table in which we will insert the route
 832         self.create_classify_session(
 833             self.acl_tbl_idx.get(key),
 834             self.build_ip_match(src_ip=self.pg0.remote_ip4),
 835             pbr_option, self.pbr_vrfid)
 836         self.assertTrue(self.verify_vrf(self.pbr_vrfid))
 837         r = VppIpRoute(self, self.pg3.local_ip4, 24,
 838                        [VppRoutePath(self.pg3.remote_ip4,
 839                                      INVALID_INDEX)],
 840                        table_id=self.pbr_vrfid)
 841         r.add_vpp_config()
 842
 843         self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get(key))
 844
 845         self.pg_enable_capture(self.pg_interfaces)
 846         self.pg_start()
 847
 848         pkts = self.pg3.get_capture(len(pkts))
 849         self.verify_capture(self.pg3, pkts)
 850         self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get(key), 0)
 851         self.pg0.assert_nothing_captured(remark="packets forwarded")
 852         self.pg1.assert_nothing_captured(remark="packets forwarded")
 853         self.pg2.assert_nothing_captured(remark="packets forwarded")
 854
 855         # remove the classify session and the route
 856         r.remove_vpp_config()
 857         self.create_classify_session(
 858             self.acl_tbl_idx.get(key),
 859             self.build_ip_match(src_ip=self.pg0.remote_ip4),
 860             pbr_option, self.pbr_vrfid, is_add=0)
 861
 862         # and the table should be gone.
 863         self.assertFalse(self.verify_vrf(self.pbr_vrfid))
 864
 865 if __name__ == '__main__':
 866     unittest.main(testRunner=VppTestRunner)