#!/usr/bin/env python3 """ACL plugin Test Case HLD: """ import unittest import random 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 framework import tag_fixme_vpp_workers from util import Host, ppp from ipaddress import IPv4Network, IPv6Network from vpp_lo_interface import VppLoInterface from vpp_acl import AclRule, VppAcl, VppAclInterface, VppEtypeWhitelist from vpp_ip import INVALID_INDEX @tag_fixme_vpp_workers class TestACLplugin(VppTestCase): """ACL plugin 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(TestACLplugin, 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_v2( bd_id=cls.bd_id, uu_flood=1, learn=1, flood=1, forward=1, is_add=1 ) for pg_if in cls.pg_interfaces: cls.vapi.sw_interface_set_l2_bridge( rx_sw_if_index=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() count = 16 start = 0 n_int = len(cls.pg_interfaces) macs_per_if = count // n_int i = -1 for pg_if in cls.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 = cls.hosts_by_pg_idx[pg_if.sw_if_index] for j in range(int(start_nr), int(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) except Exception: super(TestACLplugin, cls).tearDownClass() raise @classmethod def tearDownClass(cls): super(TestACLplugin, cls).tearDownClass() def setUp(self): super(TestACLplugin, self).setUp() self.reset_packet_infos() def tearDown(self): """ Show various debug prints after each test. """ super(TestACLplugin, self).tearDown() def show_commands_at_teardown(self): cli = "show vlib graph l2-input-feat-arc" self.logger.info(self.vapi.ppcli(cli)) cli = "show vlib graph l2-input-feat-arc-end" self.logger.info(self.vapi.ppcli(cli)) cli = "show vlib graph l2-output-feat-arc" self.logger.info(self.vapi.ppcli(cli)) cli = "show vlib graph l2-output-feat-arc-end" self.logger.info(self.vapi.ppcli(cli)) self.logger.info(self.vapi.ppcli("show l2fib verbose")) self.logger.info(self.vapi.ppcli("show acl-plugin acl")) self.logger.info(self.vapi.ppcli("show acl-plugin interface")) self.logger.info(self.vapi.ppcli("show acl-plugin tables")) self.logger.info(self.vapi.ppcli("show bridge-domain %s detail" % self.bd_id)) def create_rule( self, ip=0, permit_deny=0, ports=PORTS_ALL, proto=-1, s_prefix=0, s_ip=0, d_prefix=0, d_ip=0, ): if ip: src_prefix = IPv6Network((s_ip, s_prefix)) dst_prefix = IPv6Network((d_ip, d_prefix)) else: src_prefix = IPv4Network((s_ip, s_prefix)) dst_prefix = IPv4Network((d_ip, d_prefix)) return AclRule( is_permit=permit_deny, ports=ports, proto=proto, src_prefix=src_prefix, dst_prefix=dst_prefix, ) def apply_rules(self, rules, tag=None): acl = VppAcl(self, rules, tag=tag) acl.add_vpp_config() self.logger.info("Dumped ACL: " + str(acl.dump())) # Apply a ACL on the interface as inbound for i in self.pg_interfaces: acl_if = VppAclInterface( self, sw_if_index=i.sw_if_index, n_input=1, acls=[acl] ) acl_if.add_vpp_config() return acl.acl_index def apply_rules_to(self, rules, tag=None, sw_if_index=INVALID_INDEX): acl = VppAcl(self, rules, tag=tag) acl.add_vpp_config() self.logger.info("Dumped ACL: " + str(acl.dump())) # Apply a ACL on the interface as inbound acl_if = VppAclInterface(self, sw_if_index=sw_if_index, n_input=1, acls=[acl]) return acl.acl_index def etype_whitelist(self, whitelist, n_input, add=True): # Apply whitelists on all the interfaces if add: self._wl = [] for i in self.pg_interfaces: self._wl.append( VppEtypeWhitelist( self, sw_if_index=i.sw_if_index, whitelist=whitelist, n_input=n_input, ).add_vpp_config() ) else: if hasattr(self, "_wl"): for wl in self._wl: wl.remove_vpp_config() 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[int(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(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 = 
import socket
import unittest

from scapy.layers.inet import IP, ICMP
from scapy.layers.l2 import Ether
from scapy.layers.ipsec import SecurityAssociation, ESP

from framework import VppTestCase, VppTestRunner


class TestIpsecEsp(VppTestCase):
    """
    Basic test for ipsec esp sanity - tunnel and transport modes.

    Below 4 cases are covered as part of this test
    1) ipsec esp v4 transport basic test  - IPv4 Transport mode
        scenario using HMAC-SHA1-96 intergrity algo
    2) ipsec esp v4 transport burst test
        Above test for 257 pkts
    3) ipsec esp 4o4 tunnel basic test    - IPv4 Tunnel mode
        scenario using HMAC-SHA1-96 intergrity algo
    4) ipsec esp 4o4 tunnel burst test
        Above test for 257 pkts

    TRANSPORT MODE:

     ---   encrypt   ---
    |pg2| <-------> |VPP|
     ---   decrypt   ---

    TUNNEL MODE:

     ---   encrypt   ---   plain   ---
    |pg0| <-------  |VPP| <------ |pg1|
     ---             ---           ---

     ---   decrypt   ---   plain   ---
    |pg0| ------->  |VPP| ------> |pg1|
     ---             ---           ---

    Note : IPv6 is not covered
    """

    remote_pg0_lb_addr = '1.1.1.1'
    remote_pg1_lb_addr = '2.2.2.2'

    @classmethod
    def setUpClass(cls):
        super(TestIpsecEsp, cls).setUpClass()
        try:
            cls.create_pg_interfaces(range(3))
            cls.interfaces = list(cls.pg_interfaces)
            for i in cls.interfaces:
                i.admin_up()
                i.config_ip4()
                i.resolve_arp()
            cls.logger.info(cls.vapi.ppcli("show int addr"))
            cls.configEspTra()
            cls.logger.info(cls.vapi.ppcli("show ipsec"))
            cls.configEspTun()
            cls.logger.info(cls.vapi.ppcli("show ipsec"))
        except Exception:
            super(TestIpsecEsp, cls).tearDownClass()
            raise

    @classmethod
    def configEspTun(cls):
        try:
            spd_id = 1
            remote_sa_id = 10
            local_sa_id = 20
            remote_tun_spi = 1001
            local_tun_spi = 1000
            src4 = socket.inet_pton(socket.AF_INET, cls.remote_pg0_lb_addr)
            cls.vapi.ip_add_del_route(src4, 32, cls.pg0.remote_ip4n)
            dst4 = socket.inet_pton(socket.AF_INET, cls.remote_pg1_lb_addr)
            cls.vapi.ip_add_del_route(dst4, 32, cls.pg1.remote_ip4n)
            cls.vapi.ipsec_sad_add_del_entry(
                remote_sa_id,
                remote_tun_spi,
                cls.pg0.local_ip4n,
                cls.pg0.remote_ip4n,
                integrity_key_length=20,
                crypto_key_length=16,
                protocol=1)
            cls.vapi.ipsec_sad_add_del_entry(
                local_sa_id,
                local_tun_spi,
                cls.pg0.remote_ip4n,
                cls.pg0.local_ip4n,
                integrity_key_length=20,
                crypto_key_length=16,
                protocol=1)
            cls.vapi.ipsec_spd_add_del(spd_id)
            cls.vapi.ipsec_interface_add_del_spd(spd_id, cls.pg0.sw_if_index)
            l_startaddr = r_startaddr = socket.inet_pton(
                socket.AF_INET, "0.0.0.0")
            l_stopaddr = r_stopaddr = socket.inet_pton(
                socket.AF_INET, "255.255.255.255")
            cls.vapi.ipsec_spd_add_del_entry(
                spd_id,
                l_startaddr,
                l_stopaddr,
                r_startaddr,
                r_stopaddr,
                protocol=socket.IPPROTO_ESP)
            cls.vapi.ipsec_spd_add_del_entry(
                spd_id,
                l_startaddr,
                l_stopaddr,
                r_startaddr,
                r_stopaddr,
                protocol=socket.IPPROTO_ESP,
                is_outbound=0)
            l_startaddr = l_stopaddr = socket.inet_pton(
                socket.AF_INET, cls.remote_pg0_lb_addr)
            r_startaddr = r_stopaddr = socket.inet_pton(
                socket.AF_INET, cls.remote_pg1_lb_addr)
            cls.vapi.ipsec_spd_add_del_entry(
                spd_id,
                l_startaddr,
                l_stopaddr,
                r_startaddr,
                r_stopaddr,
                priority=10,
                policy=3,
                is_outbound=0,
                sa_id=local_sa_id)
            cls.vapi.ipsec_spd_add_del_entry(
                spd_id,
                r_startaddr,
                r_stopaddr,
                l_startaddr,
                l_stopaddr,
                priority=10,
                policy=3,
                sa_id=remote_sa_id)
        except Exception:
            raise

    @classmethod
    def configEspTra(cls):
        try:
            spd_id = 2
            remote_sa_id = 30
            local_sa_id = 40
            remote_tra_spi = 2001
            local_tra_spi = 2000
            cls.vapi.ipsec_sad_add_del_entry(
                remote_sa_id,
                remote_tra_spi,
                integrity_key_length=20,
                crypto_key_length=16,
                protocol=1,
                is_tunnel=0)
            cls.vapi.ipsec_sad_add_del_entry(
                local_sa_id,
                local_tra_spi,
                integrity_key_length=20,
                crypto_key_length=16,
                protocol=1,
                is_tunnel=0)
            cls.vapi.ipsec_spd_add_del(spd_id)
            cls.vapi.ipsec_interface_add_del_spd(spd_id, cls.pg2.sw_if_index)
            l_startaddr = r_startaddr = socket.inet_pton(
                socket.AF_INET, "0.0.0.0")
            l_stopaddr = r_stopaddr = socket.inet_pton(
                socket.AF_INET, "255.255.255.255")
            cls.vapi.ipsec_spd_add_del_entry(
                spd_id,
                l_startaddr,
                l_stopaddr,
                r_startaddr,
                r_stopaddr,
                protocol=socket.IPPROTO_ESP)
            cls.vapi.ipsec_spd_add_del_entry(
                spd_id,
                l_startaddr,
                l_stopaddr,
                r_startaddr,
                r_stopaddr,
                protocol=socket.IPPROTO_ESP,
                is_outbound=0)
            l_startaddr = l_stopaddr = cls.pg2.local_ip4n
            r_startaddr = r_stopaddr = cls.pg2.remote_ip4n
            cls.vapi.ipsec_spd_add_del_entry(
                spd_id,
                l_startaddr,
                l_stopaddr,
                r_startaddr,
                r_stopaddr,
                priority=10,
                policy=3,
                is_outbound=0,
                sa_id=local_sa_id)
            cls.vapi.ipsec_spd_add_del_entry(
                spd_id,
                l_startaddr,
                l_stopaddr,
                r_startaddr,
                r_stopaddr,
                priority=10,
                policy=3,
                sa_id=remote_sa_id)
        except Exception:
            raise

    def configScapySA(self, is_tun=False):
        if is_tun:
            self.remote_tun_sa = SecurityAssociation(
                ESP,
                spi=0x000003e8,
                crypt_algo='AES-CBC',
                crypt_key='JPjyOWBeVEQiMe7h',
                auth_algo='HMAC-SHA1-96',
                auth_key='C91KUR9GYMm5GfkEvNjX',
                tunnel_header=IP(
                    src=self.pg0.remote_ip4,
                    dst=self.pg0.local_ip4))
            self.local_tun_sa = SecurityAssociation(
                ESP,
                spi=0x000003e9,
                crypt_algo='AES-CBC',
                crypt_key='JPjyOWBeVEQiMe7h',
                auth_algo='HMAC-SHA1-96',
                auth_key='C91KUR9GYMm5GfkEvNjX',
                tunnel_header=IP(
                    dst=self.pg0.remote_ip4,
                    src=self.pg0.local_ip4))
        else:
            self.remote_tra_sa = SecurityAssociation(
                ESP,
                spi=0x000007d0,
                crypt_algo='AES-CBC',
                crypt_key='JPjyOWBeVEQiMe7h',
                auth_algo='HMAC-SHA1-96',
                auth_key='C91KUR9GYMm5GfkEvNjX')
            self.local_tra_sa = SecurityAssociation(
                ESP,
                spi=0x000007d1,
                crypt_algo='AES-CBC',
                crypt_key='JPjyOWBeVEQiMe7h',
                auth_algo='HMAC-SHA1-96',
                auth_key='C91KUR9GYMm5GfkEvNjX')

    def tearDown(self):
        super(TestIpsecEsp, self).tearDown()
        if not self.vpp_dead:
            self.vapi.cli("show hardware")

    def send_and_expect(self, input, pkts, output, count=1):
        input.add_stream(pkts)
        self.pg_enable_capture(self.pg_interfaces)
        self.pg_start()
        rx =