test/test_acl_plugin_conns.py

#!/usr/bin/env python
""" ACL plugin extended stateful tests """

import unittest
from framework import VppTestCase, VppTestRunner, running_extended_tests
from scapy.layers.l2 import Ether
from scapy.packet import Raw
from scapy.layers.inet import IP, UDP
from scapy.packet import Packet
from socket import inet_pton, AF_INET, AF_INET6
from scapy.layers.inet6 import IPv6, ICMPv6Unknown, ICMPv6EchoRequest
from scapy.layers.inet6 import ICMPv6EchoReply, IPv6ExtHdrRouting
from scapy.layers.inet6 import IPv6ExtHdrFragment
from pprint import pprint
from random import randint


def to_acl_rule(self, is_permit, wildcard_sport=False):
    p = self
    rule_family = AF_INET6 if p.haslayer(IPv6) else AF_INET
    rule_prefix_len = 128 if p.haslayer(IPv6) else 32
    rule_l3_layer = IPv6 if p.haslayer(IPv6) else IP
    rule_l4_sport = p.sport
    rule_l4_dport = p.dport
    if p.haslayer(IPv6):
        rule_l4_proto = p[IPv6].nh
    else:
        rule_l4_proto = p[IP].proto

    if wildcard_sport:
        rule_l4_sport_first = 0
        rule_l4_sport_last = 65535
    else:
        rule_l4_sport_first = rule_l4_sport
        rule_l4_sport_last = rule_l4_sport

    new_rule = {
          'is_permit': is_permit,
          'is_ipv6': p.haslayer(IPv6),
          'src_ip_addr': inet_pton(rule_family,
                                   p[rule_l3_layer].src),
          'src_ip_prefix_len': rule_prefix_len,
          'dst_ip_addr': inet_pton(rule_family,
                                   p[rule_l3_layer].dst),
          'dst_ip_prefix_len': rule_prefix_len,
          'srcport_or_icmptype_first': rule_l4_sport_first,
          'srcport_or_icmptype_last': rule_l4_sport_last,
          'dstport_or_icmpcode_first': rule_l4_dport,
          'dstport_or_icmpcode_last': rule_l4_dport,
          'proto': rule_l4_proto,
         }
    return new_rule

Packet.to_acl_rule = to_acl_rule


class IterateWithSleep():
    def __init__(self, testcase, n_iters, description, sleep_sec):
        self.curr = 0
        self.testcase = testcase
        self.n_iters = n_iters
        self.sleep_sec = sleep_sec
        self.description = description

    def __iter__(self):
        for x in range(0, self.n_iters):
            yield x
            self.testcase.sleep(self.sleep_sec)


class Conn():
    def __init__(self, testcase, if1, if2, af, l4proto, port1, port2):
        self.testcase = testcase
        self.ifs = [None, None]
        self.ifs[0] = if1
        self.ifs[1] = if2
        self.address_family = af
        self.l4proto = l4proto
        self.ports = [None, None]
        self.ports[0] = port1
        self.ports[1] = port2
        self

    def pkt(self, side):
        is_ip6 = 1 if self.address_family == AF_INET6 else 0
        s0 = side
        s1 = 1-side
        src_if = self.ifs[s0]
        dst_if = self.ifs[s1]
        layer_3 = [IP(src=src_if.remote_ip4, dst=dst_if.remote_ip4),
                   IPv6(src=src_if.remote_ip6, dst=dst_if.remote_ip6)]
        payload = "x"
        p = (Ether(dst=src_if.local_mac, src=src_if.remote_mac) /
             layer_3[is_ip6] /
             self.l4proto(sport=self.ports[s0], dport=self.ports[s1]) /
             Raw(payload))
        return p

    def apply_acls(self, reflect_side, acl_side):
        pkts = []
        pkts.append(self.pkt(0))
        pkts.append(self.pkt(1))
        pkt = pkts[reflect_side]

        r = []
        r.append(pkt.to_acl_rule(2, wildcard_sport=True))
        r.append(self.wildcard_rule(0))
        res = self.testcase.api_acl_add_replace(0xffffffff, r)
        self.testcase.assert_equal(res.retval, 0, "error adding ACL")
        reflect_acl_index = res.acl_index

        r = []
        r.append(self.wildcard_rule(0))
        res = self.testcase.api_acl_add_replace(0xffffffff, r)
        self.testcase.assert_equal(res.retval, 0, "error adding deny ACL")
        deny_acl_index = res.acl_index

        if reflect_side == acl_side:
            self.testcase.api_acl_interface_set_acl_list(
                   self.ifs[acl_side].sw_if_index, 2, 1,
                   [reflect_acl_index,
                    deny_acl_index])
            self.testcase.api_acl_interface_set_acl_list(
                   self.ifs[1-acl_side].sw_if_index, 0, 0, [])
        else:
            self.testcase.api_acl_interface_set_acl_list(
                   self.ifs[acl_side].sw_if_index, 2, 1,
                   [deny_acl_index,
                    reflect_acl_index])
            self.testcase.api_acl_interface_set_acl_list(
                   self.ifs[1-acl_side].sw_if_index, 0, 0, [])

    def wildcard_rule(self, is_permit):
        any_addr = ["0.0.0.0", "::"]
        rule_family = self.address_family
        is_ip6 = 1 if rule_family == AF_INET6 else 0
        new_rule = {
              'is_permit': is_permit,
              'is_ipv6': is_ip6,
              'src_ip_addr': inet_pton(rule_family, any_addr[is_ip6]),
              'src_ip_prefix_len': 0,
              'dst_ip_addr': inet_pton(rule_family, any_addr[is_ip6]),
              'dst_ip_prefix_len': 0,
              'srcport_or_icmptype_first': 0,
              'srcport_or_icmptype_last': 65535,
              'dstport_or_icmpcode_first': 0,
              'dstport_or_icmpcode_last': 65535,
              'proto': 0,
             }
        return new_rule

    def send(self, side):
        self.ifs[side].add_stream(self.pkt(side))
        self.ifs[1-side].enable_capture()
        self.testcase.pg_start()

    def recv(self, side):
        p = self.ifs[side].wait_for_packet(1)
        return p

    def send_through(self, side):
        self.send(side)
        p = self.recv(1-side)
        return p

    def send_pingpong(self, side):
        p1 = self.send_through(side)
        p2 = self.send_through(1-side)
        return [p1, p2]


@unittest.skipUnless(running_extended_tests(), "part of extended tests")
class ACLPluginConnTestCase(VppTestCase):
    """ ACL plugin connection-oriented extended testcases """

    @classmethod
    def setUpClass(self):
        super(ACLPluginConnTestCase, self).setUpClass()
        # create pg0 and pg1
        self.create_pg_interfaces(range(2))
        for i in self.pg_interfaces:
            i.admin_up()
            i.config_ip4()
            i.config_ip6()
            i.resolve_arp()
            i.resolve_ndp()

    def tearDown(self):
        """Run standard test teardown and log various show commands
        """
        super(ACLPluginConnTestCase, self).tearDown()
        if not self.vpp_dead:
            self.logger.info(self.vapi.cli("show ip arp"))
            self.logger.info(self.vapi.cli("show ip6 neighbors"))
            self.logger.info(self.vapi.cli("show acl-plugin sessions"))
            self.logger.info(self.vapi.cli("show acl-plugin acl"))
            self.logger.info(self.vapi.cli("show acl-plugin interface"))
            self.logger.info(self.vapi.cli("show acl-plugin tables"))

    def api_acl_add_replace(self, acl_index, r, count=-1, tag="",
                            expected_retval=0):
        """Add/replace an ACL

        :param int acl_index: ACL index to replace, 4294967295 to create new.
        :param acl_rule r: ACL rules array.
        :param str tag: symbolic tag (description) for this ACL.
        :param int count: number of rules.
        """
        if (count < 0):
            count = len(r)
        return self.vapi.api(self.vapi.papi.acl_add_replace,
                             {'acl_index': acl_index,
                              'r': r,
                              'count': count,
                              'tag': tag
                              }, expected_retval=expected_retval)

    def api_acl_interface_set_acl_list(self, sw_if_index, count, n_input, acls,
                                       expected_retval=0):
        return self.vapi.api(self.vapi.papi.acl_interface_set_acl_list,
                             {'sw_if_index': sw_if_index,
                              'count': count,
                              'n_input': n_input,
                              'acls': acls
                              }, expected_retval=expected_retval)

    def api_acl_dump(self, acl_index, expected_retval=0):
        return self.vapi.api(self.vapi.papi.acl_dump,
                             {'acl_index': acl_index},
                             expected_retval=expected_retval)

    def run_basic_conn_test(self, af, acl_side):
        """ Basic conn timeout test """
        conn1 = Conn(self, self.pg0, self.pg1, af, UDP, 42001, 4242)
        conn1.apply_acls(0, acl_side)
        conn1.send_through(0)
        # the return packets should pass
        conn1.send_through(1)
        # send some packets on conn1, ensure it doesn't go away
        for i in IterateWithSleep(self, 20, "Keep conn active", 0.3):
            conn1.send_through(1)
        # allow the conn to time out
        for i in IterateWithSleep(self, 30, "Wait for timeout", 0.1):
            pass
        # now try to send a packet on the reflected side
        try:
            p2 = conn1.send_through(1).command()
        except:
            # If we asserted while waiting, it's good.
            # the conn should have timed out.
            p2 = None
        self.assert_equal(p2, None, "packet on long-idle conn")

    def run_active_conn_test(self, af, acl_side):
        """ Idle connection behind active connection test """
        base = 10000 + 1000*acl_side
        conn1 = Conn(self, self.pg0, self.pg1, af, UDP, base + 1, 2323)
        conn2 = Conn(self, self.pg0, self.pg1, af, UDP, base + 2, 2323)
        conn3 = Conn(self, self.pg0, self.pg1, af, UDP, base + 3, 2323)
        conn1.apply_acls(0, acl_side)
        conn1.send(0)
        conn1.recv(1)
        # create and check that the conn2/3 work
        self.sleep(0.1)
        conn2.send_pingpong(0)
        self.sleep(0.1)
        conn3.send_pingpong(0)
        # send some packets on conn1, keep conn2/3 idle
        for i in IterateWithSleep(self, 20, "Keep conn active", 0.2):
            conn1.send_through(1)
        try:
            p2 = conn2.send_through(1).command()
        except:
            # If we asserted while waiting, it's good.
            # the conn should have timed out.
            p2 = None
        # We should have not received the packet on a long-idle
        # connection, because it should have timed out
        # If it didn't - it is a problem
        self.assert_equal(p2, None, "packet on long-idle conn")

    def run_clear_conn_test(self, af, acl_side):
        """ Clear the connections via CLI """
        conn1 = Conn(self, self.pg0, self.pg1, af, UDP, 42001, 4242)
        conn1.apply_acls(0, acl_side)
        conn1.send_through(0)
        # the return packets should pass
        conn1.send_through(1)
        # send some packets on conn1, ensure it doesn't go away
        for i in IterateWithSleep(self, 20, "Keep conn active", 0.3):
            conn1.send_through(1)
        # clear all connections
        self.vapi.ppcli("clear acl-plugin sessions")
        # now try to send a packet on the reflected side
        try:
            p2 = conn1.send_through(1).command()
        except:
            # If we asserted while waiting, it's good.
            # the conn should have timed out.
            p2 = None
        self.assert_equal(p2, None, "packet on supposedly deleted conn")

    def test_0000_conn_prepare_test(self):
        """ Prepare the settings """
        self.vapi.ppcli("set acl-plugin session timeout udp idle 1")

    def test_0001_basic_conn_test(self):
        """ IPv4: Basic conn timeout test reflect on ingress """
        self.run_basic_conn_test(AF_INET, 0)

    def test_0002_basic_conn_test(self):
        """ IPv4: Basic conn timeout test reflect on egress """
        self.run_basic_conn_test(AF_INET, 1)

    def test_0005_clear_conn_test(self):
        """ IPv4: reflect egress, clear conn """
        self.run_clear_conn_test(AF_INET, 1)

    def test_0006_clear_conn_test(self):
        """ IPv4: reflect ingress, clear conn """
        self.run_clear_conn_test(AF_INET, 0)

    def test_0011_active_conn_test(self):
        """ IPv4: Idle conn behind active conn, reflect on ingress """
        self.run_active_conn_test(AF_INET, 0)

    def test_0012_active_conn_test(self):
        """ IPv4: Idle conn behind active conn, reflect on egress """
        self.run_active_conn_test(AF_INET, 1)

    def test_1001_basic_conn_test(self):
        """ IPv6: Basic conn timeout test reflect on ingress """
        self.run_basic_conn_test(AF_INET6, 0)

    def test_1002_basic_conn_test(self):
        """ IPv6: Basic conn timeout test reflect on egress """
        self.run_basic_conn_test(AF_INET6, 1)

    def test_1005_clear_conn_test(self):
        """ IPv6: reflect egress, clear conn """
        self.run_clear_conn_test(AF_INET6, 1)

    def test_1006_clear_conn_test(self):
        """ IPv6: reflect ingress, clear conn """
        self.run_clear_conn_test(AF_INET6, 0)

    def test_1011_active_conn_test(self):
        """ IPv6: Idle conn behind active conn, reflect on ingress """
        self.run_active_conn_test(AF_INET6, 0)

    def test_1012_active_conn_test(self):
        """ IPv6: Idle conn behind active conn, reflect on egress """
        self.run_active_conn_test(AF_INET6, 1)