src/plugins/acl/test/test_classify_l2_acl.py

   1 #!/usr/bin/env python
   2 """ Classifier-based L2 ACL Test Case HLD:
   3 """
   4
   5 import unittest
   6 import random
   7 import binascii
   8 import socket
   9
  10
  11 from scapy.packet import Raw
  12 from scapy.data import ETH_P_IP
  13 from scapy.layers.l2 import Ether
  14 from scapy.layers.inet import IP, TCP, UDP, ICMP
  15 from scapy.layers.inet6 import IPv6, ICMPv6EchoRequest
  16 from scapy.layers.inet6 import IPv6ExtHdrFragment
  17 from framework import VppTestCase, VppTestRunner
  18 from util import Host, ppp
  19
  20
  21 class TestClassifyAcl(VppTestCase):
  22     """ Classifier-based L2 input and output ACL Test Case """
  23
  24     # traffic types
  25     IP = 0
  26     ICMP = 1
  27
  28     # IP version
  29     IPRANDOM = -1
  30     IPV4 = 0
  31     IPV6 = 1
  32
  33     # rule types
  34     DENY = 0
  35     PERMIT = 1
  36
  37     # supported protocols
  38     proto = [[6, 17], [1, 58]]
  39     proto_map = {1: 'ICMP', 58: 'ICMPv6EchoRequest', 6: 'TCP', 17: 'UDP'}
  40     ICMPv4 = 0
  41     ICMPv6 = 1
  42     TCP = 0
  43     UDP = 1
  44     PROTO_ALL = 0
  45
  46     # port ranges
  47     PORTS_ALL = -1
  48     PORTS_RANGE = 0
  49     PORTS_RANGE_2 = 1
  50     udp_sport_from = 10
  51     udp_sport_to = udp_sport_from + 5
  52     udp_dport_from = 20000
  53     udp_dport_to = udp_dport_from + 5000
  54     tcp_sport_from = 30
  55     tcp_sport_to = tcp_sport_from + 5
  56     tcp_dport_from = 40000
  57     tcp_dport_to = tcp_dport_from + 5000
  58
  59     udp_sport_from_2 = 90
  60     udp_sport_to_2 = udp_sport_from_2 + 5
  61     udp_dport_from_2 = 30000
  62     udp_dport_to_2 = udp_dport_from_2 + 5000
  63     tcp_sport_from_2 = 130
  64     tcp_sport_to_2 = tcp_sport_from_2 + 5
  65     tcp_dport_from_2 = 20000
  66     tcp_dport_to_2 = tcp_dport_from_2 + 5000
  67
  68     icmp4_type = 8  # echo request
  69     icmp4_code = 3
  70     icmp6_type = 128  # echo request
  71     icmp6_code = 3
  72
  73     icmp4_type_2 = 8
  74     icmp4_code_from_2 = 5
  75     icmp4_code_to_2 = 20
  76     icmp6_type_2 = 128
  77     icmp6_code_from_2 = 8
  78     icmp6_code_to_2 = 42
  79
  80     # Test variables
  81     bd_id = 1
  82
  83     @classmethod
  84     def setUpClass(cls):
  85         """
  86         Perform standard class setup (defined by class method setUpClass in
  87         class VppTestCase) before running the test case, set test case related
  88         variables and configure VPP.
  89         """
  90         super(TestClassifyAcl, cls).setUpClass()
  91
  92         try:
  93             # Create 2 pg interfaces
  94             cls.create_pg_interfaces(range(2))
  95
  96             # Packet flows mapping pg0 -> pg1, pg2 etc.
  97             cls.flows = dict()
  98             cls.flows[cls.pg0] = [cls.pg1]
  99
 100             # Packet sizes
 101             cls.pg_if_packet_sizes = [64, 512, 1518, 9018]
 102
 103             # Create BD with MAC learning and unknown unicast flooding disabled
 104             # and put interfaces to this BD
 105             cls.vapi.bridge_domain_add_del(bd_id=cls.bd_id, uu_flood=1,
 106                                            learn=1)
 107             for pg_if in cls.pg_interfaces:
 108                 cls.vapi.sw_interface_set_l2_bridge(
 109                     rx_sw_if_index=pg_if.sw_if_index, bd_id=cls.bd_id)
 110
 111             # Set up all interfaces
 112             for i in cls.pg_interfaces:
 113                 i.admin_up()
 114
 115             # Mapping between packet-generator index and lists of test hosts
 116             cls.hosts_by_pg_idx = dict()
 117             for pg_if in cls.pg_interfaces:
 118                 cls.hosts_by_pg_idx[pg_if.sw_if_index] = []
 119
 120             # Create list of deleted hosts
 121             cls.deleted_hosts_by_pg_idx = dict()
 122             for pg_if in cls.pg_interfaces:
 123                 cls.deleted_hosts_by_pg_idx[pg_if.sw_if_index] = []
 124
 125             # warm-up the mac address tables
 126             # self.warmup_test()
 127
 128             # Holder of the active classify table key
 129             cls.acl_active_table = ''
 130
 131         except Exception:
 132             super(TestClassifyAcl, cls).tearDownClass()
 133             raise
 134
 135     @classmethod
 136     def tearDownClass(cls):
 137         super(TestClassifyAcl, cls).tearDownClass()
 138
 139     def setUp(self):
 140         super(TestClassifyAcl, self).setUp()
 141
 142         self.acl_tbl_idx = {}
 143         self.reset_packet_infos()
 144
 145     def tearDown(self):
 146         """
 147         Show various debug prints after each test.
 148         """
 149         if not self.vpp_dead:
 150             if self.acl_active_table == 'mac_inout':
 151                 self.output_acl_set_interface(
 152                     self.pg1, self.acl_tbl_idx.get(self.acl_active_table), 0)
 153                 self.input_acl_set_interface(
 154                     self.pg0, self.acl_tbl_idx.get(self.acl_active_table), 0)
 155                 self.acl_active_table = ''
 156             elif self.acl_active_table == 'mac_out':
 157                 self.output_acl_set_interface(
 158                     self.pg1, self.acl_tbl_idx.get(self.acl_active_table), 0)
 159                 self.acl_active_table = ''
 160             elif self.acl_active_table == 'mac_in':
 161                 self.input_acl_set_interface(
 162                     self.pg0, self.acl_tbl_idx.get(self.acl_active_table), 0)
 163                 self.acl_active_table = ''
 164
 165         super(TestClassifyAcl, self).tearDown()
 166
 167     def show_commands_at_teardown(self):
 168         self.logger.info(self.vapi.ppcli("show inacl type l2"))
 169         self.logger.info(self.vapi.ppcli("show outacl type l2"))
 170         self.logger.info(self.vapi.ppcli("show classify tables verbose"))
 171         self.logger.info(self.vapi.ppcli("show bridge-domain %s detail"
 172                                          % self.bd_id))
 173
 174     @staticmethod
 175     def build_mac_mask(dst_mac='', src_mac='', ether_type=''):
 176         """Build MAC ACL mask data with hexstring format
 177
 178         :param str dst_mac: source MAC address <0-ffffffffffff>
 179         :param str src_mac: destination MAC address <0-ffffffffffff>
 180         :param str ether_type: ethernet type <0-ffff>
 181         """
 182
 183         return ('{!s:0>12}{!s:0>12}{!s:0>4}'.format(
 184             dst_mac, src_mac, ether_type)).rstrip('0')
 185
 186     @staticmethod
 187     def build_mac_match(dst_mac='', src_mac='', ether_type=''):
 188         """Build MAC ACL match data with hexstring format
 189
 190         :param str dst_mac: source MAC address <x:x:x:x:x:x>
 191         :param str src_mac: destination MAC address <x:x:x:x:x:x>
 192         :param str ether_type: ethernet type <0-ffff>
 193         """
 194         if dst_mac:
 195             dst_mac = dst_mac.replace(':', '')
 196         if src_mac:
 197             src_mac = src_mac.replace(':', '')
 198
 199         return ('{!s:0>12}{!s:0>12}{!s:0>4}'.format(
 200             dst_mac, src_mac, ether_type)).rstrip('0')
 201
 202     def create_classify_table(self, key, mask, data_offset=0, is_add=1):
 203         """Create Classify Table
 204
 205         :param str key: key for classify table (ex, ACL name).
 206         :param str mask: mask value for interested traffic.
 207         :param int match_n_vectors:
 208         :param int is_add: option to configure classify table.
 209             - create(1) or delete(0)
 210         """
 211         r = self.vapi.classify_add_del_table(
 212             is_add,
 213             binascii.unhexlify(mask),
 214             match_n_vectors=(len(mask) - 1) // 32 + 1,
 215             miss_next_index=0,
 216             current_data_flag=1,
 217             current_data_offset=data_offset)
 218         self.assertIsNotNone(r, 'No response msg for add_del_table')
 219         self.acl_tbl_idx[key] = r.new_table_index
 220
 221     def create_classify_session(self, intf, table_index, match,
 222                                 hit_next_index=0xffffffff, is_add=1):
 223         """Create Classify Session
 224
 225         :param VppInterface intf: Interface to apply classify session.
 226         :param int table_index: table index to identify classify table.
 227         :param str match: matched value for interested traffic.
 228         :param int pbr_action: enable/disable PBR feature.
 229         :param int vrfid: VRF id.
 230         :param int is_add: option to configure classify session.
 231             - create(1) or delete(0)
 232         """
 233         r = self.vapi.classify_add_del_session(
 234             is_add,
 235             table_index,
 236             binascii.unhexlify(match),
 237             hit_next_index=hit_next_index)
 238         self.assertIsNotNone(r, 'No response msg for add_del_session')
 239
 240     def input_acl_set_interface(self, intf, table_index, is_add=1):
 241         """Configure Input ACL interface
 242
 243         :param VppInterface intf: Interface to apply Input ACL feature.
 244         :param int table_index: table index to identify classify table.
 245         :param int is_add: option to configure classify session.
 246             - enable(1) or disable(0)
 247         """
 248         r = self.vapi.input_acl_set_interface(
 249             is_add,
 250             intf.sw_if_index,
 251             l2_table_index=table_index)
 252         self.assertIsNotNone(r, 'No response msg for acl_set_interface')
 253
 254     def output_acl_set_interface(self, intf, table_index, is_add=1):
 255         """Configure Output ACL interface
 256
 257         :param VppInterface intf: Interface to apply Output ACL feature.
 258         :param int table_index: table index to identify classify table.
 259         :param int is_add: option to configure classify session.
 260             - enable(1) or disable(0)
 261         """
 262         r = self.vapi.output_acl_set_interface(
 263             is_add,
 264             intf.sw_if_index,
 265             l2_table_index=table_index)
 266         self.assertIsNotNone(r, 'No response msg for acl_set_interface')
 267
 268     def create_hosts(self, count, start=0):
 269         """
 270         Create required number of host MAC addresses and distribute them among
 271         interfaces. Create host IPv4 address for every host MAC address.
 272
 273         :param int count: Number of hosts to create MAC/IPv4 addresses for.
 274         :param int start: Number to start numbering from.
 275         """
 276         n_int = len(self.pg_interfaces)
 277         macs_per_if = count // n_int
 278         i = -1
 279         for pg_if in self.pg_interfaces:
 280             i += 1
 281             start_nr = macs_per_if * i + start
 282             end_nr = count + start if i == (n_int - 1) \
 283                 else macs_per_if * (i + 1) + start
 284             hosts = self.hosts_by_pg_idx[pg_if.sw_if_index]
 285             for j in range(start_nr, end_nr):
 286                 host = Host(
 287                     "00:00:00:ff:%02x:%02x" % (pg_if.sw_if_index, j),
 288                     "172.17.1%02x.%u" % (pg_if.sw_if_index, j),
 289                     "2017:dead:%02x::%u" % (pg_if.sw_if_index, j))
 290                 hosts.append(host)
 291
 292     def create_upper_layer(self, packet_index, proto, ports=0):
 293         p = self.proto_map[proto]
 294         if p == 'UDP':
 295             if ports == 0:
 296                 return UDP(sport=random.randint(self.udp_sport_from,
 297                                                 self.udp_sport_to),
 298                            dport=random.randint(self.udp_dport_from,
 299                                                 self.udp_dport_to))
 300             else:
 301                 return UDP(sport=ports, dport=ports)
 302         elif p == 'TCP':
 303             if ports == 0:
 304                 return TCP(sport=random.randint(self.tcp_sport_from,
 305                                                 self.tcp_sport_to),
 306                            dport=random.randint(self.tcp_dport_from,
 307                                                 self.tcp_dport_to))
 308             else:
 309                 return TCP(sport=ports, dport=ports)
 310         return ''
 311
 312     def create_stream(self, src_if, packet_sizes, traffic_type=0, ipv6=0,
 313                       proto=-1, ports=0, fragments=False,
 314                       pkt_raw=True, etype=-1):
 315         """
 316         Create input packet stream for defined interface using hosts or
 317         deleted_hosts list.
 318
 319         :param object src_if: Interface to create packet stream for.
 320         :param list packet_sizes: List of required packet sizes.
 321         :param traffic_type: 1: ICMP packet, 2: IPv6 with EH, 0: otherwise.
 322         :return: Stream of packets.
 323         """
 324         pkts = []
 325         if self.flows.__contains__(src_if):
 326             src_hosts = self.hosts_by_pg_idx[src_if.sw_if_index]
 327             for dst_if in self.flows[src_if]:
 328                 dst_hosts = self.hosts_by_pg_idx[dst_if.sw_if_index]
 329                 n_int = len(dst_hosts) * len(src_hosts)
 330                 for i in range(0, n_int):
 331                     dst_host = dst_hosts[i // len(src_hosts)]
 332                     src_host = src_hosts[i % len(src_hosts)]
 333                     pkt_info = self.create_packet_info(src_if, dst_if)
 334                     if ipv6 == 1:
 335                         pkt_info.ip = 1
 336                     elif ipv6 == 0:
 337                         pkt_info.ip = 0
 338                     else:
 339                         pkt_info.ip = random.choice([0, 1])
 340                     if proto == -1:
 341                         pkt_info.proto = random.choice(self.proto[self.IP])
 342                     else:
 343                         pkt_info.proto = proto
 344                     payload = self.info_to_payload(pkt_info)
 345                     p = Ether(dst=dst_host.mac, src=src_host.mac)
 346                     if etype > 0:
 347                         p = Ether(dst=dst_host.mac,
 348                                   src=src_host.mac,
 349                                   type=etype)
 350                     if pkt_info.ip:
 351                         p /= IPv6(dst=dst_host.ip6, src=src_host.ip6)
 352                         if fragments:
 353                             p /= IPv6ExtHdrFragment(offset=64, m=1)
 354                     else:
 355                         if fragments:
 356                             p /= IP(src=src_host.ip4, dst=dst_host.ip4,
 357                                     flags=1, frag=64)
 358                         else:
 359                             p /= IP(src=src_host.ip4, dst=dst_host.ip4)
 360                     if traffic_type == self.ICMP:
 361                         if pkt_info.ip:
 362                             p /= ICMPv6EchoRequest(type=self.icmp6_type,
 363                                                    code=self.icmp6_code)
 364                         else:
 365                             p /= ICMP(type=self.icmp4_type,
 366                                       code=self.icmp4_code)
 367                     else:
 368                         p /= self.create_upper_layer(i, pkt_info.proto, ports)
 369                     if pkt_raw:
 370                         p /= Raw(payload)
 371                         pkt_info.data = p.copy()
 372                     if pkt_raw:
 373                         size = random.choice(packet_sizes)
 374                         self.extend_packet(p, size)
 375                     pkts.append(p)
 376         return pkts
 377
 378     def verify_capture(self, pg_if, capture,
 379                        traffic_type=0, ip_type=0, etype=-1):
 380         """
 381         Verify captured input packet stream for defined interface.
 382
 383         :param object pg_if: Interface to verify captured packet stream for.
 384         :param list capture: Captured packet stream.
 385         :param traffic_type: 1: ICMP packet, 2: IPv6 with EH, 0: otherwise.
 386         """
 387         last_info = dict()
 388         for i in self.pg_interfaces:
 389             last_info[i.sw_if_index] = None
 390         dst_sw_if_index = pg_if.sw_if_index
 391         for packet in capture:
 392             if etype > 0:
 393                 if packet[Ether].type != etype:
 394                     self.logger.error(ppp("Unexpected ethertype in packet:",
 395                                           packet))
 396                 else:
 397                     continue
 398             try:
 399                 # Raw data for ICMPv6 are stored in ICMPv6EchoRequest.data
 400                 if traffic_type == self.ICMP and ip_type == self.IPV6:
 401                     payload_info = self.payload_to_info(
 402                         packet[ICMPv6EchoRequest].data)
 403                     payload = packet[ICMPv6EchoRequest]
 404                 else:
 405                     payload_info = self.payload_to_info(packet[Raw])
 406                     payload = packet[self.proto_map[payload_info.proto]]
 407             except:
 408                 self.logger.error(ppp("Unexpected or invalid packet "
 409                                       "(outside network):", packet))
 410                 raise
 411
 412             if ip_type != 0:
 413                 self.assertEqual(payload_info.ip, ip_type)
 414             if traffic_type == self.ICMP:
 415                 try:
 416                     if payload_info.ip == 0:
 417                         self.assertEqual(payload.type, self.icmp4_type)
 418                         self.assertEqual(payload.code, self.icmp4_code)
 419                     else:
 420                         self.assertEqual(payload.type, self.icmp6_type)
 421                         self.assertEqual(payload.code, self.icmp6_code)
 422                 except:
 423                     self.logger.error(ppp("Unexpected or invalid packet "
 424                                           "(outside network):", packet))
 425                     raise
 426             else:
 427                 try:
 428                     ip_version = IPv6 if payload_info.ip == 1 else IP
 429
 430                     ip = packet[ip_version]
 431                     packet_index = payload_info.index
 432
 433                     self.assertEqual(payload_info.dst, dst_sw_if_index)
 434                     self.logger.debug("Got packet on port %s: src=%u (id=%u)" %
 435                                       (pg_if.name, payload_info.src,
 436                                        packet_index))
 437                     next_info = self.get_next_packet_info_for_interface2(
 438                         payload_info.src, dst_sw_if_index,
 439                         last_info[payload_info.src])
 440                     last_info[payload_info.src] = next_info
 441                     self.assertTrue(next_info is not None)
 442                     self.assertEqual(packet_index, next_info.index)
 443                     saved_packet = next_info.data
 444                     # Check standard fields
 445                     self.assertEqual(ip.src, saved_packet[ip_version].src)
 446                     self.assertEqual(ip.dst, saved_packet[ip_version].dst)
 447                     p = self.proto_map[payload_info.proto]
 448                     if p == 'TCP':
 449                         tcp = packet[TCP]
 450                         self.assertEqual(tcp.sport, saved_packet[
 451                             TCP].sport)
 452                         self.assertEqual(tcp.dport, saved_packet[
 453                             TCP].dport)
 454                     elif p == 'UDP':
 455                         udp = packet[UDP]
 456                         self.assertEqual(udp.sport, saved_packet[
 457                             UDP].sport)
 458                         self.assertEqual(udp.dport, saved_packet[
 459                             UDP].dport)
 460                 except:
 461                     self.logger.error(ppp("Unexpected or invalid packet:",
 462                                           packet))
 463                     raise
 464         for i in self.pg_interfaces:
 465             remaining_packet = self.get_next_packet_info_for_interface2(
 466                 i, dst_sw_if_index, last_info[i.sw_if_index])
 467             self.assertTrue(
 468                 remaining_packet is None,
 469                 "Port %u: Packet expected from source %u didn't arrive" %
 470                 (dst_sw_if_index, i.sw_if_index))
 471
 472     def run_traffic_no_check(self):
 473         # Test
 474         # Create incoming packet streams for packet-generator interfaces
 475         for i in self.pg_interfaces:
 476             if self.flows.__contains__(i):
 477                 pkts = self.create_stream(i, self.pg_if_packet_sizes)
 478                 if len(pkts) > 0:
 479                     i.add_stream(pkts)
 480
 481         # Enable packet capture and start packet sending
 482         self.pg_enable_capture(self.pg_interfaces)
 483         self.pg_start()
 484
 485     def run_verify_test(self, traffic_type=0, ip_type=0, proto=-1, ports=0,
 486                         frags=False, pkt_raw=True, etype=-1):
 487         # Test
 488         # Create incoming packet streams for packet-generator interfaces
 489         pkts_cnt = 0
 490         for i in self.pg_interfaces:
 491             if self.flows.__contains__(i):
 492                 pkts = self.create_stream(i, self.pg_if_packet_sizes,
 493                                           traffic_type, ip_type, proto, ports,
 494                                           frags, pkt_raw, etype)
 495                 if len(pkts) > 0:
 496                     i.add_stream(pkts)
 497                     pkts_cnt += len(pkts)
 498
 499         # Enable packet capture and start packet sendingself.IPV
 500         self.pg_enable_capture(self.pg_interfaces)
 501         self.pg_start()
 502
 503         # Verify
 504         # Verify outgoing packet streams per packet-generator interface
 505         for src_if in self.pg_interfaces:
 506             if self.flows.__contains__(src_if):
 507                 for dst_if in self.flows[src_if]:
 508                     capture = dst_if.get_capture(pkts_cnt)
 509                     self.logger.info("Verifying capture on interface %s" %
 510                                      dst_if.name)
 511                     self.verify_capture(dst_if, capture,
 512                                         traffic_type, ip_type, etype)
 513
 514     def run_verify_negat_test(self, traffic_type=0, ip_type=0, proto=-1,
 515                               ports=0, frags=False, etype=-1):
 516         # Test
 517         self.reset_packet_infos()
 518         for i in self.pg_interfaces:
 519             if self.flows.__contains__(i):
 520                 pkts = self.create_stream(i, self.pg_if_packet_sizes,
 521                                           traffic_type, ip_type, proto, ports,
 522                                           frags, True, etype)
 523                 if len(pkts) > 0:
 524                     i.add_stream(pkts)
 525
 526         # Enable packet capture and start packet sending
 527         self.pg_enable_capture(self.pg_interfaces)
 528         self.pg_start()
 529
 530         # Verify
 531         # Verify outgoing packet streams per packet-generator interface
 532         for src_if in self.pg_interfaces:
 533             if self.flows.__contains__(src_if):
 534                 for dst_if in self.flows[src_if]:
 535                     self.logger.info("Verifying capture on interface %s" %
 536                                      dst_if.name)
 537                     capture = dst_if.get_capture(0)
 538                     self.assertEqual(len(capture), 0)
 539
 540     def build_classify_table(self, src_mac='', dst_mac='', ether_type='',
 541                              etype='', key='mac', hit_next_index=0xffffffff):
 542         # Basic ACL testing
 543         a_mask = self.build_mac_mask(src_mac=src_mac, dst_mac=dst_mac,
 544                                      ether_type=ether_type)
 545         self.create_classify_table(key, a_mask)
 546         for host in self.hosts_by_pg_idx[self.pg0.sw_if_index]:
 547             s_mac = host.mac if src_mac else ''
 548             if dst_mac:
 549                 for dst_if in self.flows[self.pg0]:
 550                     for dst_host in self.hosts_by_pg_idx[dst_if.sw_if_index]:
 551                         self.create_classify_session(
 552                             self.pg0, self.acl_tbl_idx.get(key),
 553                             self.build_mac_match(src_mac=s_mac,
 554                                                  dst_mac=dst_host.mac,
 555                                                  ether_type=etype),
 556                             hit_next_index=hit_next_index)
 557             else:
 558                 self.create_classify_session(
 559                     self.pg0, self.acl_tbl_idx.get(key),
 560                     self.build_mac_match(src_mac=s_mac, dst_mac='',
 561                                          ether_type=etype),
 562                     hit_next_index=hit_next_index)
 563
 564     def test_0000_warmup_test(self):
 565         """ Learn the MAC addresses
 566         """
 567         self.create_hosts(2)
 568         self.run_traffic_no_check()
 569
 570     def test_0010_inacl_permit_src_mac(self):
 571         """ Input  L2 ACL test - permit source MAC
 572
 573         Test scenario for basic IP ACL with source IP
 574             - Create IPv4 stream for pg0 -> pg1 interface.
 575             - Create ACL with source MAC address.
 576             - Send and verify received packets on pg1 interface.
 577         """
 578         key = 'mac_in'
 579         self.build_classify_table(src_mac='ffffffffffff', key=key)
 580         self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get(key))
 581         self.acl_active_table = key
 582         self.run_verify_test(self.IP, self.IPV4, -1)
 583
 584     def test_0011_inacl_permit_dst_mac(self):
 585         """ Input  L2 ACL test - permit destination MAC
 586
 587         Test scenario for basic IP ACL with source IP
 588             - Create IPv4 stream for pg0 -> pg1 interface.
 589             - Create ACL with destination MAC address.
 590             - Send and verify received packets on pg1 interface.
 591         """
 592         key = 'mac_in'
 593         self.build_classify_table(dst_mac='ffffffffffff', key=key)
 594         self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get(key))
 595         self.acl_active_table = key
 596         self.run_verify_test(self.IP, self.IPV4, -1)
 597
 598     def test_0012_inacl_permit_src_dst_mac(self):
 599         """ Input  L2 ACL test - permit source and destination MAC
 600
 601         Test scenario for basic IP ACL with source IP
 602             - Create IPv4 stream for pg0 -> pg1 interface.
 603             - Create ACL with source and destination MAC addresses.
 604             - Send and verify received packets on pg1 interface.
 605         """
 606         key = 'mac_in'
 607         self.build_classify_table(
 608             src_mac='ffffffffffff', dst_mac='ffffffffffff', key=key)
 609         self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get(key))
 610         self.acl_active_table = key
 611         self.run_verify_test(self.IP, self.IPV4, -1)
 612
 613     def test_0013_inacl_permit_ether_type(self):
 614         """ Input  L2 ACL test - permit ether_type
 615
 616         Test scenario for basic IP ACL with source IP
 617             - Create IPv4 stream for pg0 -> pg1 interface.
 618             - Create ACL with destination MAC address.
 619             - Send and verify received packets on pg1 interface.
 620         """
 621         key = 'mac_in'
 622         self.build_classify_table(
 623             ether_type='ffff', etype=hex(ETH_P_IP)[2:], key=key)
 624         self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get(key))
 625         self.acl_active_table = key
 626         self.run_verify_test(self.IP, self.IPV4, -1)
 627
 628     def test_0015_inacl_deny(self):
 629         """ Input  L2 ACL test - deny
 630
 631         Test scenario for basic IP ACL with source IP
 632             - Create IPv4 stream for pg0 -> pg1 interface.
 633
 634             - Create ACL with source MAC address.
 635             - Send and verify no received packets on pg1 interface.
 636         """
 637         key = 'mac_in'
 638         self.build_classify_table(
 639             src_mac='ffffffffffff', hit_next_index=0, key=key)
 640         self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get(key))
 641         self.acl_active_table = key
 642         self.run_verify_negat_test(self.IP, self.IPV4, -1)
 643
 644     def test_0020_outacl_permit(self):
 645         """ Output L2 ACL test - permit
 646
 647         Test scenario for basic IP ACL with source IP
 648             - Create IPv4 stream for pg0 -> pg1 interface.
 649             - Create ACL with source MAC address.
 650             - Send and verify received packets on pg1 interface.
 651         """
 652         key = 'mac_out'
 653         self.build_classify_table(src_mac='ffffffffffff', key=key)
 654         self.output_acl_set_interface(self.pg1, self.acl_tbl_idx.get(key))
 655         self.acl_active_table = key
 656         self.run_verify_test(self.IP, self.IPV4, -1)
 657
 658     def test_0025_outacl_deny(self):
 659         """ Output L2 ACL test - deny
 660
 661         Test scenario for basic IP ACL with source IP
 662             - Create IPv4 stream for pg0 -> pg1 interface.
 663             - Create ACL with source MAC address.
 664             - Send and verify no received packets on pg1 interface.
 665         """
 666         key = 'mac_out'
 667         self.build_classify_table(
 668             src_mac='ffffffffffff', hit_next_index=0, key=key)
 669         self.output_acl_set_interface(self.pg1, self.acl_tbl_idx.get(key))
 670         self.acl_active_table = key
 671         self.run_verify_negat_test(self.IP, self.IPV4, -1)
 672
 673     def test_0030_inoutacl_permit(self):
 674         """ Input+Output L2 ACL test - permit
 675
 676         Test scenario for basic IP ACL with source IP
 677             - Create IPv4 stream for pg0 -> pg1 interface.
 678             - Create ACLs with source MAC address.
 679             - Send and verify received packets on pg1 interface.
 680         """
 681         key = 'mac_inout'
 682         self.build_classify_table(src_mac='ffffffffffff', key=key)
 683         self.output_acl_set_interface(self.pg1, self.acl_tbl_idx.get(key))
 684         self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get(key))
 685         self.acl_active_table = key
 686         self.run_verify_test(self.IP, self.IPV4, -1)
 687
 688 if __name__ == '__main__':
 689     unittest.main(testRunner=VppTestRunner)