classifier-based ACL: testcases for L2 ACLs + fix the enabling of outbound L2 ACL

There was no test coverage for the L2 ACL (other than indirect by
means of ACL plugin tests), so the enabling of the outbound ACL
got fumbled throughout the revisions of the refactoring.

Fix both issues - the error and the lack of test coverage for L2 ACL.

Change-Id: Ib7f42780ef84b4a4f70bd88d7319aeeda866cf06
Signed-off-by: Andrew Yourtchenko <ayourtch@gmail.com>

1 files changed, 633 insertions, 0 deletions

diff --git a/test/test_classify_l2_acl.py b/test/test_classify_l2_acl.py
new file mode 100644
index 00000000000..c5660a5701c
--- /dev/null
+++ b/test/test_classify_l2_acl.py
@@ -0,0 +1,633 @@
+#!/usr/bin/env python
+""" Classifier-based L2 ACL Test Case HLD:
+"""
+
+import unittest
+import random
+import binascii
+import socket
+
+
+from scapy.packet import Raw
+from scapy.layers.l2 import Ether
+from scapy.layers.inet import IP, TCP, UDP, ICMP
+from scapy.layers.inet6 import IPv6, ICMPv6EchoRequest
+from scapy.layers.inet6 import IPv6ExtHdrFragment
+from framework import VppTestCase, VppTestRunner
+from util import Host, ppp
+
+
+class TestClassifyAcl(VppTestCase):
+    """ Classifier-based L2 input and output ACL Test Case """
+
+    # traffic types
+    IP = 0
+    ICMP = 1
+
+    # IP version
+    IPRANDOM = -1
+    IPV4 = 0
+    IPV6 = 1
+
+    # rule types
+    DENY = 0
+    PERMIT = 1
+
+    # supported protocols
+    proto = [[6, 17], [1, 58]]
+    proto_map = {1: 'ICMP', 58: 'ICMPv6EchoRequest', 6: 'TCP', 17: 'UDP'}
+    ICMPv4 = 0
+    ICMPv6 = 1
+    TCP = 0
+    UDP = 1
+    PROTO_ALL = 0
+
+    # port ranges
+    PORTS_ALL = -1
+    PORTS_RANGE = 0
+    PORTS_RANGE_2 = 1
+    udp_sport_from = 10
+    udp_sport_to = udp_sport_from + 5
+    udp_dport_from = 20000
+    udp_dport_to = udp_dport_from + 5000
+    tcp_sport_from = 30
+    tcp_sport_to = tcp_sport_from + 5
+    tcp_dport_from = 40000
+    tcp_dport_to = tcp_dport_from + 5000
+
+    udp_sport_from_2 = 90
+    udp_sport_to_2 = udp_sport_from_2 + 5
+    udp_dport_from_2 = 30000
+    udp_dport_to_2 = udp_dport_from_2 + 5000
+    tcp_sport_from_2 = 130
+    tcp_sport_to_2 = tcp_sport_from_2 + 5
+    tcp_dport_from_2 = 20000
+    tcp_dport_to_2 = tcp_dport_from_2 + 5000
+
+    icmp4_type = 8  # echo request
+    icmp4_code = 3
+    icmp6_type = 128  # echo request
+    icmp6_code = 3
+
+    icmp4_type_2 = 8
+    icmp4_code_from_2 = 5
+    icmp4_code_to_2 = 20
+    icmp6_type_2 = 128
+    icmp6_code_from_2 = 8
+    icmp6_code_to_2 = 42
+
+    # Test variables
+    bd_id = 1
+
+    @classmethod
+    def setUpClass(cls):
+        """
+        Perform standard class setup (defined by class method setUpClass in
+        class VppTestCase) before running the test case, set test case related
+        variables and configure VPP.
+        """
+        super(TestClassifyAcl, cls).setUpClass()
+
+        try:
+            # Create 2 pg interfaces
+            cls.create_pg_interfaces(range(2))
+
+            # Packet flows mapping pg0 -> pg1, pg2 etc.
+            cls.flows = dict()
+            cls.flows[cls.pg0] = [cls.pg1]
+
+            # Packet sizes
+            cls.pg_if_packet_sizes = [64, 512, 1518, 9018]
+
+            # Create BD with MAC learning and unknown unicast flooding disabled
+            # and put interfaces to this BD
+            cls.vapi.bridge_domain_add_del(bd_id=cls.bd_id, uu_flood=1,
+                                           learn=1)
+            for pg_if in cls.pg_interfaces:
+                cls.vapi.sw_interface_set_l2_bridge(pg_if.sw_if_index,
+                                                    bd_id=cls.bd_id)
+
+            # Set up all interfaces
+            for i in cls.pg_interfaces:
+                i.admin_up()
+
+            # Mapping between packet-generator index and lists of test hosts
+            cls.hosts_by_pg_idx = dict()
+            for pg_if in cls.pg_interfaces:
+                cls.hosts_by_pg_idx[pg_if.sw_if_index] = []
+
+            # Create list of deleted hosts
+            cls.deleted_hosts_by_pg_idx = dict()
+            for pg_if in cls.pg_interfaces:
+                cls.deleted_hosts_by_pg_idx[pg_if.sw_if_index] = []
+
+            # warm-up the mac address tables
+            # self.warmup_test()
+
+        except Exception:
+            super(TestClassifyAcl, cls).tearDownClass()
+            raise
+
+    def setUp(self):
+        super(TestClassifyAcl, self).setUp()
+
+        self.acl_tbl_idx = {}
+        self.reset_packet_infos()
+
+    def tearDown(self):
+        """
+        Show various debug prints after each test.
+        """
+        super(TestClassifyAcl, self).tearDown()
+        if not self.vpp_dead:
+            self.logger.info(self.vapi.ppcli("show inacl type l2"))
+            self.logger.info(self.vapi.ppcli("show outacl type l2"))
+            self.logger.info(self.vapi.ppcli("show classify tables verbose"))
+            self.logger.info(self.vapi.ppcli("show bridge-domain %s detail"
+                                             % self.bd_id))
+
+    @staticmethod
+    def build_ip_mask(proto='', src_ip='', dst_ip='',
+                      src_port='', dst_port=''):
+        """Build IP ACL mask data with hexstring format
+
+        :param str proto: protocol number <0-ff>
+        :param str src_ip: source ip address <0-ffffffff>
+        :param str dst_ip: destination ip address <0-ffffffff>
+        :param str src_port: source port number <0-ffff>
+        :param str dst_port: destination port number <0-ffff>
+        """
+
+        return ('{:0>20}{:0>12}{:0>8}{:0>12}{:0>4}'.format(
+            proto, src_ip, dst_ip, src_port, dst_port)).rstrip('0')
+
+    @staticmethod
+    def build_ip_match(proto='', src_ip='', dst_ip='',
+                       src_port='', dst_port=''):
+        """Build IP ACL match data with hexstring format
+
+        :param str proto: protocol number with valid option "<0-ff>"
+        :param str src_ip: source ip address with format of "x.x.x.x"
+        :param str dst_ip: destination ip address with format of "x.x.x.x"
+        :param str src_port: source port number <0-ffff>
+        :param str dst_port: destination port number <0-ffff>
+        """
+        if src_ip:
+            src_ip = socket.inet_aton(src_ip).encode('hex')
+        if dst_ip:
+            dst_ip = socket.inet_aton(dst_ip).encode('hex')
+
+        return ('{:0>20}{:0>12}{:0>8}{:0>12}{:0>4}'.format(
+            proto, src_ip, dst_ip, src_port, dst_port)).rstrip('0')
+
+    @staticmethod
+    def build_mac_mask(dst_mac='', src_mac='', ether_type=''):
+        """Build MAC ACL mask data with hexstring format
+
+        :param str dst_mac: source MAC address <0-ffffffffffff>
+        :param str src_mac: destination MAC address <0-ffffffffffff>
+        :param str ether_type: ethernet type <0-ffff>
+        """
+
+        return ('{:0>12}{:0>12}{:0>4}'.format(dst_mac, src_mac,
+                                              ether_type)).rstrip('0')
+
+    @staticmethod
+    def build_mac_match(dst_mac='', src_mac='', ether_type=''):
+        """Build MAC ACL match data with hexstring format
+
+        :param str dst_mac: source MAC address <x:x:x:x:x:x>
+        :param str src_mac: destination MAC address <x:x:x:x:x:x>
+        :param str ether_type: ethernet type <0-ffff>
+        """
+        if dst_mac:
+            dst_mac = dst_mac.replace(':', '')
+        if src_mac:
+            src_mac = src_mac.replace(':', '')
+
+        return ('{:0>12}{:0>12}{:0>4}'.format(dst_mac, src_mac,
+                                              ether_type)).rstrip('0')
+
+    def create_classify_table(self, key, mask, data_offset=0, is_add=1):
+        """Create Classify Table
+
+        :param str key: key for classify table (ex, ACL name).
+        :param str mask: mask value for interested traffic.
+        :param int match_n_vectors:
+        :param int is_add: option to configure classify table.
+            - create(1) or delete(0)
+        """
+        r = self.vapi.classify_add_del_table(
+            is_add,
+            binascii.unhexlify(mask),
+            match_n_vectors=(len(mask) - 1) // 32 + 1,
+            miss_next_index=0,
+            current_data_flag=1,
+            current_data_offset=data_offset)
+        self.assertIsNotNone(r, msg='No response msg for add_del_table')
+        self.acl_tbl_idx[key] = r.new_table_index
+
+    def create_classify_session(self, intf, table_index, match,
+                                hit_next_index=0xffffffff, is_add=1):
+        """Create Classify Session
+
+        :param VppInterface intf: Interface to apply classify session.
+        :param int table_index: table index to identify classify table.
+        :param str match: matched value for interested traffic.
+        :param int pbr_action: enable/disable PBR feature.
+        :param int vrfid: VRF id.
+        :param int is_add: option to configure classify session.
+            - create(1) or delete(0)
+        """
+        r = self.vapi.classify_add_del_session(
+            is_add,
+            table_index,
+            binascii.unhexlify(match),
+            hit_next_index=hit_next_index)
+        self.assertIsNotNone(r, msg='No response msg for add_del_session')
+
+    def input_acl_set_interface(self, intf, table_index, is_add=1):
+        """Configure Input ACL interface
+
+        :param VppInterface intf: Interface to apply Input ACL feature.
+        :param int table_index: table index to identify classify table.
+        :param int is_add: option to configure classify session.
+            - enable(1) or disable(0)
+        """
+        r = self.vapi.input_acl_set_interface(
+            is_add,
+            intf.sw_if_index,
+            l2_table_index=table_index)
+        self.assertIsNotNone(r, msg='No response msg for acl_set_interface')
+
+    def output_acl_set_interface(self, intf, table_index, is_add=1):
+        """Configure Output ACL interface
+
+        :param VppInterface intf: Interface to apply Output ACL feature.
+        :param int table_index: table index to identify classify table.
+        :param int is_add: option to configure classify session.
+            - enable(1) or disable(0)
+        """
+        r = self.vapi.output_acl_set_interface(
+            is_add,
+            intf.sw_if_index,
+            l2_table_index=table_index)
+        self.assertIsNotNone(r, msg='No response msg for acl_set_interface')
+
+    def create_hosts(self, count, start=0):
+        """
+        Create required number of host MAC addresses and distribute them among
+        interfaces. Create host IPv4 address for every host MAC address.
+
+        :param int count: Number of hosts to create MAC/IPv4 addresses for.
+        :param int start: Number to start numbering from.
+        """
+        n_int = len(self.pg_interfaces)
+        macs_per_if = count / n_int
+        i = -1
+        for pg_if in self.pg_interfaces:
+            i += 1
+            start_nr = macs_per_if * i + start
+            end_nr = count + start if i == (n_int - 1) \
+                else macs_per_if * (i + 1) + start
+            hosts = self.hosts_by_pg_idx[pg_if.sw_if_index]
+            for j in range(start_nr, end_nr):
+                host = Host(
+                    "00:00:00:ff:%02x:%02x" % (pg_if.sw_if_index, j),
+                    "172.17.1%02x.%u" % (pg_if.sw_if_index, j),
+                    "2017:dead:%02x::%u" % (pg_if.sw_if_index, j))
+                hosts.append(host)
+
+    def create_upper_layer(self, packet_index, proto, ports=0):
+        p = self.proto_map[proto]
+        if p == 'UDP':
+            if ports == 0:
+                return UDP(sport=random.randint(self.udp_sport_from,
+                                                self.udp_sport_to),
+                           dport=random.randint(self.udp_dport_from,
+                                                self.udp_dport_to))
+            else:
+                return UDP(sport=ports, dport=ports)
+        elif p == 'TCP':
+            if ports == 0:
+                return TCP(sport=random.randint(self.tcp_sport_from,
+                                                self.tcp_sport_to),
+                           dport=random.randint(self.tcp_dport_from,
+                                                self.tcp_dport_to))
+            else:
+                return TCP(sport=ports, dport=ports)
+        return ''
+
+    def create_stream(self, src_if, packet_sizes, traffic_type=0, ipv6=0,
+                      proto=-1, ports=0, fragments=False,
+                      pkt_raw=True, etype=-1):
+        """
+        Create input packet stream for defined interface using hosts or
+        deleted_hosts list.
+
+        :param object src_if: Interface to create packet stream for.
+        :param list packet_sizes: List of required packet sizes.
+        :param traffic_type: 1: ICMP packet, 2: IPv6 with EH, 0: otherwise.
+        :return: Stream of packets.
+        """
+        pkts = []
+        if self.flows.__contains__(src_if):
+            src_hosts = self.hosts_by_pg_idx[src_if.sw_if_index]
+            for dst_if in self.flows[src_if]:
+                dst_hosts = self.hosts_by_pg_idx[dst_if.sw_if_index]
+                n_int = len(dst_hosts) * len(src_hosts)
+                for i in range(0, n_int):
+                    dst_host = dst_hosts[i / len(src_hosts)]
+                    src_host = src_hosts[i % len(src_hosts)]
+                    pkt_info = self.create_packet_info(src_if, dst_if)
+                    if ipv6 == 1:
+                        pkt_info.ip = 1
+                    elif ipv6 == 0:
+                        pkt_info.ip = 0
+                    else:
+                        pkt_info.ip = random.choice([0, 1])
+                    if proto == -1:
+                        pkt_info.proto = random.choice(self.proto[self.IP])
+                    else:
+                        pkt_info.proto = proto
+                    payload = self.info_to_payload(pkt_info)
+                    p = Ether(dst=dst_host.mac, src=src_host.mac)
+                    if etype > 0:
+                        p = Ether(dst=dst_host.mac,
+                                  src=src_host.mac,
+                                  type=etype)
+                    if pkt_info.ip:
+                        p /= IPv6(dst=dst_host.ip6, src=src_host.ip6)
+                        if fragments:
+                            p /= IPv6ExtHdrFragment(offset=64, m=1)
+                    else:
+                        if fragments:
+                            p /= IP(src=src_host.ip4, dst=dst_host.ip4,
+                                    flags=1, frag=64)
+                        else:
+                            p /= IP(src=src_host.ip4, dst=dst_host.ip4)
+                    if traffic_type == self.ICMP:
+                        if pkt_info.ip:
+                            p /= ICMPv6EchoRequest(type=self.icmp6_type,
+                                                   code=self.icmp6_code)
+                        else:
+                            p /= ICMP(type=self.icmp4_type,
+                                      code=self.icmp4_code)
+                    else:
+                        p /= self.create_upper_layer(i, pkt_info.proto, ports)
+                    if pkt_raw:
+                        p /= Raw(payload)
+                        pkt_info.data = p.copy()
+                    if pkt_raw:
+                        size = random.choice(packet_sizes)
+                        self.extend_packet(p, size)
+                    pkts.append(p)
+        return pkts
+
+    def verify_capture(self, pg_if, capture,
+                       traffic_type=0, ip_type=0, etype=-1):
+        """
+        Verify captured input packet stream for defined interface.
+
+        :param object pg_if: Interface to verify captured packet stream for.
+        :param list capture: Captured packet stream.
+        :param traffic_type: 1: ICMP packet, 2: IPv6 with EH, 0: otherwise.
+        """
+        last_info = dict()
+        for i in self.pg_interfaces:
+            last_info[i.sw_if_index] = None
+        dst_sw_if_index = pg_if.sw_if_index
+        for packet in capture:
+            if etype > 0:
+                if packet[Ether].type != etype:
+                    self.logger.error(ppp("Unexpected ethertype in packet:",
+                                          packet))
+                else:
+                    continue
+            try:
+                # Raw data for ICMPv6 are stored in ICMPv6EchoRequest.data
+                if traffic_type == self.ICMP and ip_type == self.IPV6:
+                    payload_info = self.payload_to_info(
+                        packet[ICMPv6EchoRequest].data)
+                    payload = packet[ICMPv6EchoRequest]
+                else:
+                    payload_info = self.payload_to_info(str(packet[Raw]))
+                    payload = packet[self.proto_map[payload_info.proto]]
+            except:
+                self.logger.error(ppp("Unexpected or invalid packet "
+                                      "(outside network):", packet))
+                raise
+
+            if ip_type != 0:
+                self.assertEqual(payload_info.ip, ip_type)
+            if traffic_type == self.ICMP:
+                try:
+                    if payload_info.ip == 0:
+                        self.assertEqual(payload.type, self.icmp4_type)
+                        self.assertEqual(payload.code, self.icmp4_code)
+                    else:
+                        self.assertEqual(payload.type, self.icmp6_type)
+                        self.assertEqual(payload.code, self.icmp6_code)
+                except:
+                    self.logger.error(ppp("Unexpected or invalid packet "
+                                          "(outside network):", packet))
+                    raise
+            else:
+                try:
+                    ip_version = IPv6 if payload_info.ip == 1 else IP
+
+                    ip = packet[ip_version]
+                    packet_index = payload_info.index
+
+                    self.assertEqual(payload_info.dst, dst_sw_if_index)
+                    self.logger.debug("Got packet on port %s: src=%u (id=%u)" %
+                                      (pg_if.name, payload_info.src,
+                                       packet_index))
+                    next_info = self.get_next_packet_info_for_interface2(
+                        payload_info.src, dst_sw_if_index,
+                        last_info[payload_info.src])
+                    last_info[payload_info.src] = next_info
+                    self.assertTrue(next_info is not None)
+                    self.assertEqual(packet_index, next_info.index)
+                    saved_packet = next_info.data
+                    # Check standard fields
+                    self.assertEqual(ip.src, saved_packet[ip_version].src)
+                    self.assertEqual(ip.dst, saved_packet[ip_version].dst)
+                    p = self.proto_map[payload_info.proto]
+                    if p == 'TCP':
+                        tcp = packet[TCP]
+                        self.assertEqual(tcp.sport, saved_packet[
+                            TCP].sport)
+                        self.assertEqual(tcp.dport, saved_packet[
+                            TCP].dport)
+                    elif p == 'UDP':
+                        udp = packet[UDP]
+                        self.assertEqual(udp.sport, saved_packet[
+                            UDP].sport)
+                        self.assertEqual(udp.dport, saved_packet[
+                            UDP].dport)
+                except:
+                    self.logger.error(ppp("Unexpected or invalid packet:",
+                                          packet))
+                    raise
+        for i in self.pg_interfaces:
+            remaining_packet = self.get_next_packet_info_for_interface2(
+                i, dst_sw_if_index, last_info[i.sw_if_index])
+            self.assertTrue(
+                remaining_packet is None,
+                "Port %u: Packet expected from source %u didn't arrive" %
+                (dst_sw_if_index, i.sw_if_index))
+
+    def run_traffic_no_check(self):
+        # Test
+        # Create incoming packet streams for packet-generator interfaces
+        for i in self.pg_interfaces:
+            if self.flows.__contains__(i):
+                pkts = self.create_stream(i, self.pg_if_packet_sizes)
+                if len(pkts) > 0:
+                    i.add_stream(pkts)
+
+        # Enable packet capture and start packet sending
+        self.pg_enable_capture(self.pg_interfaces)
+        self.pg_start()
+
+    def run_verify_test(self, traffic_type=0, ip_type=0, proto=-1, ports=0,
+                        frags=False, pkt_raw=True, etype=-1):
+        # Test
+        # Create incoming packet streams for packet-generator interfaces
+        pkts_cnt = 0
+        for i in self.pg_interfaces:
+            if self.flows.__contains__(i):
+                pkts = self.create_stream(i, self.pg_if_packet_sizes,
+                                          traffic_type, ip_type, proto, ports,
+                                          frags, pkt_raw, etype)
+                if len(pkts) > 0:
+                    i.add_stream(pkts)
+                    pkts_cnt += len(pkts)
+
+        # Enable packet capture and start packet sendingself.IPV
+        self.pg_enable_capture(self.pg_interfaces)
+        self.pg_start()
+
+        # Verify
+        # Verify outgoing packet streams per packet-generator interface
+        for src_if in self.pg_interfaces:
+            if self.flows.__contains__(src_if):
+                for dst_if in self.flows[src_if]:
+                    capture = dst_if.get_capture(pkts_cnt)
+                    self.logger.info("Verifying capture on interface %s" %
+                                     dst_if.name)
+                    self.verify_capture(dst_if, capture,
+                                        traffic_type, ip_type, etype)
+
+    def run_verify_negat_test(self, traffic_type=0, ip_type=0, proto=-1,
+                              ports=0, frags=False, etype=-1):
+        # Test
+        self.reset_packet_infos()
+        for i in self.pg_interfaces:
+            if self.flows.__contains__(i):
+                pkts = self.create_stream(i, self.pg_if_packet_sizes,
+                                          traffic_type, ip_type, proto, ports,
+                                          frags, True, etype)
+                if len(pkts) > 0:
+                    i.add_stream(pkts)
+
+        # Enable packet capture and start packet sending
+        self.pg_enable_capture(self.pg_interfaces)
+        self.pg_start()
+
+        # Verify
+        # Verify outgoing packet streams per packet-generator interface
+        for src_if in self.pg_interfaces:
+            if self.flows.__contains__(src_if):
+                for dst_if in self.flows[src_if]:
+                    self.logger.info("Verifying capture on interface %s" %
+                                     dst_if.name)
+                    capture = dst_if.get_capture(0)
+                    self.assertEqual(len(capture), 0)
+
+    def build_classify_table(self, hit_next_index=0xffffffff):
+        # Basic ACL testing with source MAC
+        a_mask = self.build_mac_mask(src_mac='ffffffffffff')
+        self.create_classify_table('ip', a_mask)
+        for host in self.hosts_by_pg_idx[self.pg0.sw_if_index]:
+            self.create_classify_session(
+                self.pg0, self.acl_tbl_idx.get('ip'),
+                self.build_mac_match(src_mac=host.mac),
+                hit_next_index=hit_next_index)
+
+    def test_0000_warmup_test(self):
+        """ Learn the MAC addresses
+        """
+        self.create_hosts(2)
+        self.run_traffic_no_check()
+
+    def test_0010_inacl_permit(self):
+        """ Input  L2 ACL test - permit
+
+        Test scenario for basic IP ACL with source IP
+            - Create IPv4 stream for pg0 -> pg1 interface.
+            - Create ACL with source MAC address.
+            - Send and verify received packets on pg1 interface.
+        """
+        self.build_classify_table()
+        self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get('ip'))
+        self.run_verify_test(self.IP, self.IPV4, -1)
+        self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get('ip'), 0)
+
+    def test_0015_inacl_deny(self):
+        """ Input  L2 ACL test - deny
+
+        Test scenario for basic IP ACL with source IP
+            - Create IPv4 stream for pg0 -> pg1 interface.
+            - Create ACL with source MAC address.
+            - Send and verify no received packets on pg1 interface.
+        """
+        self.build_classify_table(hit_next_index=0)
+        self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get('ip'))
+        self.run_verify_negat_test(self.IP, self.IPV4, -1)
+        self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get('ip'), 0)
+
+    def test_0020_outacl_permit(self):
+        """ Output L2 ACL test - permit
+
+        Test scenario for basic IP ACL with source IP
+            - Create IPv4 stream for pg0 -> pg1 interface.
+            - Create ACL with source MAC address.
+            - Send and verify received packets on pg1 interface.
+        """
+        self.build_classify_table()
+        self.output_acl_set_interface(self.pg1, self.acl_tbl_idx.get('ip'))
+        self.run_verify_test(self.IP, self.IPV4, -1)
+        self.output_acl_set_interface(self.pg1, self.acl_tbl_idx.get('ip'), 0)
+
+    def test_0025_outacl_deny(self):
+        """ Output L2 ACL test - deny
+
+        Test scenario for basic IP ACL with source IP
+            - Create IPv4 stream for pg0 -> pg1 interface.
+            - Create ACL with source MAC address.
+            - Send and verify no received packets on pg1 interface.
+        """
+        self.build_classify_table(hit_next_index=0)
+        self.output_acl_set_interface(self.pg1, self.acl_tbl_idx.get('ip'))
+        self.run_verify_negat_test(self.IP, self.IPV4, -1)
+        self.output_acl_set_interface(self.pg1, self.acl_tbl_idx.get('ip'), 0)
+
+    def test_0030_inoutacl_permit(self):
+        """ Input+Output L2 ACL test - permit
+
+        Test scenario for basic IP ACL with source IP
+            - Create IPv4 stream for pg0 -> pg1 interface.
+            - Create ACLs with source MAC address.
+            - Send and verify received packets on pg1 interface.
+        """
+        self.build_classify_table()
+        self.output_acl_set_interface(self.pg1, self.acl_tbl_idx.get('ip'))
+        self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get('ip'))
+        self.run_verify_test(self.IP, self.IPV4, -1)
+        self.output_acl_set_interface(self.pg1, self.acl_tbl_idx.get('ip'), 0)
+        self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get('ip'), 0)
+
+if __name__ == '__main__':
+    unittest.main(testRunner=VppTestRunner)