test/test_ipsec_tun_if_esp.py

import unittest
import socket
import copy

from scapy.layers.ipsec import SecurityAssociation, ESP
from scapy.layers.l2 import Ether, GRE, Dot1Q
from scapy.packet import Raw, bind_layers
from scapy.layers.inet import IP, UDP, ICMP
from scapy.layers.inet6 import IPv6, ICMPv6EchoRequest
from scapy.contrib.mpls import MPLS
from asfframework import VppTestRunner, tag_fixme_vpp_workers
from template_ipsec import (
    TemplateIpsec,
    IpsecTun4Tests,
    IpsecTun6Tests,
    IpsecTun4,
    IpsecTun6,
    IpsecTcpTests,
    mk_scapy_crypt_key,
    IpsecTun6HandoffTests,
    IpsecTun4HandoffTests,
    config_tun_params,
)
from vpp_gre_interface import VppGreInterface
from vpp_ipip_tun_interface import VppIpIpTunInterface
from vpp_ip_route import (
    VppIpRoute,
    VppRoutePath,
    DpoProto,
    VppMplsLabel,
    VppMplsTable,
    VppMplsRoute,
    FibPathProto,
)
from vpp_ipsec import VppIpsecSA, VppIpsecTunProtect, VppIpsecInterface
from vpp_l2 import VppBridgeDomain, VppBridgeDomainPort
from vpp_sub_interface import L2_VTR_OP, VppDot1QSubint
from vpp_teib import VppTeib
from util import ppp
from vpp_papi import VppEnum
from vpp_papi_provider import CliFailedCommandError
from vpp_acl import AclRule, VppAcl, VppAclInterface
from vpp_policer import PolicerAction, VppPolicer, Dir


def config_tun_params(p, encryption_type, tun_if, src=None, dst=None):
    ip_class_by_addr_type = {socket.AF_INET: IP, socket.AF_INET6: IPv6}
    esn_en = bool(
        p.flags & (VppEnum.vl_api_ipsec_sad_flags_t.IPSEC_API_SAD_FLAG_USE_ESN)
    )
    crypt_key = mk_scapy_crypt_key(p)
    if tun_if:
        p.tun_dst = tun_if.remote_ip
        p.tun_src = tun_if.local_ip
    else:
        p.tun_dst = dst
        p.tun_src = src

    if p.nat_header:
        is_default_port = p.nat_header.dport == 4500
    else:
        is_default_port = True

    if is_default_port:
        outbound_nat_header = p.nat_header
    else:
        outbound_nat_header = UDP(sport=p.nat_header.dport, dport=p.nat_header.sport)
        bind_layers(UDP, ESP, dport=p.nat_header.dport)

    p.scapy_tun_sa = SecurityAssociation(
        encryption_type,
        spi=p.scapy_tun_spi,
        crypt_algo=p.crypt_algo,
        crypt_key=crypt_key,
        auth_algo=p.auth_algo,
        auth_key=p.auth_key,
        tunnel_header=ip_class_by_addr_type[p.addr_type](src=p.tun_dst, dst=p.tun_src),
        nat_t_header=outbound_nat_header,
        esn_en=esn_en,
    )
    p.vpp_tun_sa = SecurityAssociation(
        encryption_type,
        spi=p.vpp_tun_spi,
        crypt_algo=p.crypt_algo,
        crypt_key=crypt_key,
        auth_algo=p.auth_algo,
        auth_key=p.auth_key,
        tunnel_header=ip_class_by_addr_type[p.addr_type](dst=p.tun_dst, src=p.tun_src),
        nat_t_header=p.nat_header,
        esn_en=esn_en,
    )


def config_tra_params(p, encryption_type, tun_if):
    ip_class_by_addr_type = {socket.AF_INET: IP, socket.AF_INET6: IPv6}
    esn_en = bool(
        p.flags & (VppEnum.vl_api_ipsec_sad_flags_t.IPSEC_API_SAD_FLAG_USE_ESN)
    )
    crypt_key = mk_scapy_crypt_key(p)
    p.tun_dst = tun_if.remote_ip
    p.tun_src = tun_if.local_ip

    if p.nat_header:
        is_default_port = p.nat_header.dport == 4500
    else:
        is_default_port = True

    if is_default_port:
        outbound_nat_header = p.nat_header
    else:
        outbound_nat_header = UDP(sport=p.nat_header.dport, dport=p.nat_header.sport)
        bind_layers(UDP, ESP, dport=p.nat_header.dport)

    p.scapy_tun_sa = SecurityAssociation(
        encryption_type,
        spi=p.scapy_tun_spi,
        crypt_algo=p.crypt_algo,
        crypt_key=crypt_key,
        auth_algo=p.auth_algo,
        auth_key=p.auth_key,
        esn_en=esn_en,
        nat_t_header=outbound_nat_header,
    )
    p.vpp_tun_sa = SecurityAssociation(
        encryption_type,
        spi=p.vpp_tun_spi,
        crypt_algo=p.crypt_algo,
        crypt_key=crypt_key,
        auth_algo=p.auth_algo,
        auth_key=p.auth_key,
        esn_en=esn_en,
        nat_t_header=p.nat_header,
    )


class TemplateIpsec4TunProtect(object):
    """IPsec IPv4 Tunnel protect"""

    encryption_type = ESP
    tun4_encrypt_node_name = "esp4-encrypt-tun"
    tun4_decrypt_node_name = ["esp4-decrypt-tun", "esp4-decrypt-tun-post"]
    tun4_input_node = "ipsec4-tun-input"

    def config_sa_tra(self, p):
        config_tun_params(p, self.encryption_type, p.tun_if)

        p.tun_sa_out = VppIpsecSA(
            self,
            p.vpp_tun_sa_id,
            p.vpp_tun_spi,
            p.auth_algo_vpp_id,
            p.auth_key,
            p.crypt_algo_vpp_id,
            p.crypt_key,
            self.vpp_esp_protocol,
            flags=p.flags,
        )
        p.tun_sa_out.add_vpp_config()

        p.tun_sa_in = VppIpsecSA(
            self,
            p.scapy_tun_sa_id,
            p.scapy_tun_spi,
            p.auth_algo_vpp_id,
            p.auth_key,
            p.crypt_algo_vpp_id,
            p.crypt_key,
            self.vpp_esp_protocol,
            flags=p.flags,
        )
        p.tun_sa_in.add_vpp_config()

    def config_sa_tun(self, p):
        config_tun_params(p, self.encryption_type, p.tun_if)

        p.tun_sa_out = VppIpsecSA(
            self,
            p.vpp_tun_sa_id,
            p.vpp_tun_spi,
            p.auth_algo_vpp_id,
            p.auth_key,
            p.crypt_algo_vpp_id,
            p.crypt_key,
            self.vpp_esp_protocol,
            self.tun_if.local_addr[p.addr_type],
            self.tun_if.remote_addr[p.addr_type],
            flags=p.flags,
        )
        p.tun_sa_out.add_vpp_config()

        p.tun_sa_in = VppIpsecSA(
            self,
            p.scapy_tun_sa_id,
            p.scapy_tun_spi,
            p.auth_algo_vpp_id,
            p.auth_key,
            p.crypt_algo_vpp_id,
            p.crypt_key,
            self.vpp_esp_protocol,
            self.tun_if.remote_addr[p.addr_type],
            self.tun_if.local_addr[p.addr_type],
            flags=p.flags,
        )
        p.tun_sa_in.add_vpp_config()

    def config_protect(self, p):
        p.tun_protect = VppIpsecTunProtect(self, p.tun_if, p.tun_sa_out, [p.tun_sa_in])
        p.tun_protect.add_vpp_config()

    def config_network(self, p):
        if hasattr(p, "tun_dst"):
            tun_dst = p.tun_dst
        else:
            tun_dst = self.pg0.remote_ip4
        p.tun_if = VppIpIpTunInterface(self, self.pg0, self.pg0.local_ip4, tun_dst)
        p.tun_if.add_vpp_config()
        p.tun_if.admin_up()
        p.tun_if.config_ip4()
        p.tun_if.config_ip6()

        p.route = VppIpRoute(
            self,
            p.remote_tun_if_host,
            32,
            [VppRoutePath(p.tun_if.remote_ip4, 0xFFFFFFFF)],
        )
        p.route.add_vpp_config()
        r = VppIpRoute(
            self,
            p.remote_tun_if_host6,
            128,
            [
                VppRoutePath(
                    p.tun_if.remote_ip6, 0xFFFFFFFF, proto=DpoProto.DPO_PROTO_IP6
                )
            ],
        )
        r.add_vpp_config()

    def unconfig_network(self, p):
        p.route.remove_vpp_config()
        p.tun_if.remove_vpp_config()

    def unconfig_protect(self, p):
        p.tun_protect.remove_vpp_config()

    def unconfig_sa(self, p):
        p.tun_sa_out.remove_vpp_config()
        p.tun_sa_in.remove_vpp_config()


class TemplateIpsec4TunIfEsp(TemplateIpsec4TunProtect, TemplateIpsec):
    """IPsec tunnel interface tests"""

    encryption_type = ESP

    @classmethod
    def setUpClass(cls):
        super(TemplateIpsec4TunIfEsp, cls).setUpClass()

    @classmethod
    def tearDownClass(cls):
        super(TemplateIpsec4TunIfEsp, cls).tearDownClass()

    def setUp(self):
        super(TemplateIpsec4TunIfEsp, self).setUp()

        self.tun_if = self.pg0

        p = self.ipv4_params

        self.config_network(p)
        self.config_sa_tra(p)
        self.config_protect(p)

    def tearDown(self):
        super(TemplateIpsec4TunIfEsp, self).tearDown()


class TemplateIpsec4TunIfEspUdp(TemplateIpsec4TunProtect, TemplateIpsec):
    """IPsec UDP tunnel interface tests"""

    tun4_encrypt_node_name = "esp4-encrypt-tun"
    tun4_decrypt_node_name = ["esp4-decrypt-tun", "esp4-decrypt-tun-post"]
    encryption_type = ESP

    @classmethod
    def setUpClass(cls):
        super(TemplateIpsec4TunIfEspUdp, cls).setUpClass()

    @classmethod
    def tearDownClass(cls):
        super(TemplateIpsec4TunIfEspUdp, cls).tearDownClass()

    def verify_encrypted(self, p, sa, rxs):
        for rx in rxs:
            try:
                # ensure the UDP ports are correct before we decrypt
                # which strips them
                self.assertTrue(rx.haslayer(UDP))
                self.assert_equal(rx[UDP].sport, p.nat_header.sport)
                self.assert_equal(rx[UDP].dport, p.nat_header.dport)

                pkt = sa.decrypt(rx[IP])
                if not pkt.haslayer(IP):
                    pkt = IP(pkt[Raw].load)

                self.assert_packet_checksums_valid(pkt)
                self.assert_equal(pkt[IP].dst, "1.1.1.1")
                self.assert_equal(pkt[IP].src, self.pg1.remote_ip4)
            except (IndexError, AssertionError):
                self.logger.debug(ppp("Unexpected packet:", rx))
                try:
                    self.logger.debug(ppp("Decrypted packet:", pkt))
                except:
                    pass
                raise

    def config_sa_tra(self, p):
        config_tun_params(p, self.encryption_type, p.tun_if)

        p.tun_sa_out = VppIpsecSA(
            self,
            p.vpp_tun_sa_id,
            p.vpp_tun_spi,
            p.auth_algo_vpp_id,
            p.auth_key,
            p.crypt_algo_vpp_id,
            p.crypt_key,
            self.vpp_esp_protocol,
            flags=p.flags,
            udp_src=p.nat_header.sport,
            udp_dst=p.nat_header.dport,
        )
        p.tun_sa_out.add_vpp_config()

        p.tun_sa_in = VppIpsecSA(
            self,
            p.scapy_tun_sa_id,
            p.scapy_tun_spi,
            p.auth_algo_vpp_id,
            p.auth_key,
            p.crypt_algo_vpp_id,
            p.crypt_key,
            self.vpp_esp_protocol,
            flags=p.flags
            | VppEnum.vl_api_ipsec_sad_flags_t.IPSEC_API_SAD_FLAG_IS_INBOUND,
            udp_src=p.nat_header.sport,
            udp_dst=p.nat_header.dport,
        )
        p.tun_sa_in.add_vpp_config()

    def setUp(self):
        super(TemplateIpsec4TunIfEspUdp, self).setUp()

        p = self.ipv4_params
        p.flags = VppEnum.vl_api_ipsec_sad_flags_t.IPSEC_API_SAD_FLAG_UDP_ENCAP
        p.nat_header = UDP(sport=5454, dport=4500)

        self.tun_if = self.pg0

        self.config_network(p)
        self.config_sa_tra(p)
        self.config_protect(p)

    def tearDown(self):
        super(TemplateIpsec4TunIfEspUdp, self).tearDown()


class TemplateIpsec4TunTfc:
    """IPsec IPv4 tunnel with TFC"""

    def gen_encrypt_pkts(self, p, sa, sw_intf, src, dst, count=1, payload_size=54):
        pkt = (
            IP(src=src, dst=dst, len=28 + payload_size)
            / ICMP()
            / Raw(b"X" * payload_size)
            / Padding(b"Y" * 100)
        )
        return [
            Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) / sa.encrypt(pkt)
            for i in range(count)
        ]

    def verify_decrypted(self, p, rxs):
        for rx in rxs:
            self.assert_equal(rx[IP].src, p.remote_tun_if_host)
            self.assert_equal(rx[IP].dst, self.pg1.remote_ip4)
            self.assert_equal(rx[IP].len, len(rx[IP]))
            self.assert_packet_checksums_valid(rx)


class TestIpsec4TunIfEsp1(TemplateIpsec4TunIfEsp, IpsecTun4Tests):
    """Ipsec ESP - TUN tests"""

    tun4_encrypt_node_name = "esp4-encrypt-tun"
    tun4_decrypt_node_name = ["esp4-decrypt-tun", "esp4-decrypt-tun-post"]

    def test_tun_basic64(self):
        """ipsec 6o4 tunnel basic test"""
        self.tun4_encrypt_node_name = "esp4-encrypt-tun"

        self.verify_tun_64(self.params[socket.AF_INET], count=1)

    def test_tun_burst64(self):
        """ipsec 6o4 tunnel basic test"""
        self.tun4_encrypt_node_name = "esp4-encrypt-tun"

        self.verify_tun_64(self.params[socket.AF_INET], count=257)

    def test_tun_basic_frag44(self):
        """ipsec 4o4 tunnel frag basic test"""
        self.tun4_encrypt_node_name = "esp4-encrypt-tun"

        p = self.ipv4_params

        self.vapi.sw_interface_set_mtu(p.tun_if.sw_if_index, [1500, 0, 0, 0])
        self.verify_tun_44(
            self.params[socket.AF_INET], count=1, payload_size=1800, n_rx=2
        )
        self.vapi.sw_interface_set_mtu(p.tun_if.sw_if_index, [9000, 0, 0, 0])


class TestIpsec4TunIfEspUdp(TemplateIpsec4TunIfEspUdp, IpsecTun4Tests):
    """Ipsec ESP UDP tests"""

    tun4_input_node = "ipsec4-tun-input"

    def setUp(self):
        super(TestIpsec4TunIfEspUdp, self).setUp()

    def test_keepalive(self):
        """IPSEC NAT Keepalive"""
        self.verify_keepalive(self.ipv4_params)


class TestIpsec4TunIfEspUdpGCM(TemplateIpsec4TunIfEspUdp, IpsecTun4Tests):
    """Ipsec ESP UDP GCM tests"""

    tun4_input_node = "ipsec4-tun-input"

    def setUp(self):
        super(TestIpsec4TunIfEspUdpGCM, self).setUp()
        p = self.ipv4_params
        p.auth_algo_vpp_id = VppEnum.vl_api_ipsec_integ_alg_t.IPSEC_API_INTEG_ALG_NONE
        p.crypt_algo_vpp_id = (
            VppEnum.vl_api_ipsec_crypto_alg_t.IPSEC_API_CRYPTO_ALG_AES_GCM_256
        )
        p.crypt_algo = "AES-GCM"
        p.auth_algo = "NULL"
        p.crypt_key = b"JPjyOWBeVEQiMe7hJPjyOWBeVEQiMe7h"
        p.salt = 0


class TestIpsec4TunIfEspUdpUpdate(TemplateIpsec4TunIfEspUdp, IpsecTun4Tests):
    """Ipsec ESP UDP update tests"""

    tun4_input_node = "ipsec4-tun-input"

    def setUp(self):
        super(TestIpsec4TunIfEspUdpUpdate, self).setUp()
        p = self.ipv4_params
        p.nat_header = UDP(sport=6565, dport=7676)
        config_tun_params(p, self.encryption_type, p.tun_if)
        p.tun_sa_in.update_vpp_config(
            udp_src=p.nat_header.dport, udp_dst=p.nat_header.sport
        )
        p.tun_sa_out.update_vpp_config(
            udp_src=p.nat_header.sport, udp_dst=p.nat_header.dport
        )


class TestIpsec4TunIfEsp2(TemplateIpsec4TunIfEsp, IpsecTcpTests):
    """Ipsec ESP - TCP tests"""

    pass


class TemplateIpsec6TunProtect(object):
    """IPsec IPv6 Tunnel protect"""

    def config_sa_tra(self, p):
        config_tun_params(p, self.encryption_type, p.tun_if)

        p.tun_sa_out = VppIpsecSA(
            self,
            p.vpp_tun_sa_id,
            p.vpp_tun_spi,
            p.auth_algo_vpp_id,
            p.auth_key,
            p.crypt_algo_vpp_id,
            p.crypt_key,
            self.vpp_esp_protocol,
        )
        p.tun_sa_out.add_vpp_config()

        p.tun_sa_in = VppIpsecSA(
            self,
            p.scapy_tun_sa_id,
            p.scapy_tun_spi,
            p.auth_algo_vpp_id,
            p.auth_key,
            p.crypt_algo_vpp_id,
            p.crypt_key,
            self.vpp_esp_protocol,
        )
        p.tun_sa_in.add_vpp_config()

    def config_sa_tun(self, p):
        config_tun_params(p, self.encryption_type, p.tun_if)

        p.tun_sa_out = VppIpsecSA(
            self,
            p.vpp_tun_sa_id,
            p.vpp_tun_spi,
            p.auth_algo_vpp_id,
            p.auth_key,
            p.crypt_algo_vpp_id,
            p.crypt_key,
            self.vpp_esp_protocol,
            self.tun_if.local_addr[p.addr_type],
            self.tun_if.remote_addr[p.addr_type],
        )
        p.tun_sa_out.add_vpp_config()

        p.tun_sa_in = VppIpsecSA(
            self,
            p.scapy_tun_sa_id,
            p.scapy_tun_spi,
            p.auth_algo_vpp_id,
            p.auth_key,
            p.crypt_algo_vpp_id,
            p.crypt_key,
            self.vpp_esp_protocol,
            self.tun_if.remote_addr[p.addr_type],
            self.tun_if.local_addr[p.addr_type],
        )
        p.tun_sa_in.add_vpp_config()

    def config_protect(self, p):
        p.tun_protect = VppIpsecTunProtect(self, p.tun_if, p.tun_sa_out, [p.tun_sa_in])
        p.tun_protect.add_vpp_config()

    def config_network(self, p):
        if hasattr(p, "tun_dst"):
            tun_dst = p.tun_dst
        else:
            tun_dst = self.pg0.remote_ip6
        p.tun_if = VppIpIpTunInterface(self, self.pg0, self.pg0.local_ip6, tun_dst)
        p.tun_if.add_vpp_config()
        p.tun_if.admin_up()
        p.tun_if.config_ip6()
        p.tun_if.config_ip4()

        p.route = VppIpRoute(
            self,
            p.remote_tun_if_host,
            128,
            [
                VppRoutePath(
                    p.tun_if.remote_ip6, 0xFFFFFFFF, proto=DpoProto.DPO_PROTO_IP6
                )
            ],
        )
        p.route.add_vpp_config()
        r = VppIpRoute(
            self,
            p.remote_tun_if_host4,
            32,
            [VppRoutePath(p.tun_if.remote_ip4, 0xFFFFFFFF)],
        )
        r.add_vpp_config()

    def unconfig_network(self, p):
        p.route.remove_vpp_config()
        p.tun_if.remove_vpp_config()

    def unconfig_protect(self, p):
        p.tun_protect.remove_vpp_config()

    def unconfig_sa(self, p):
        p.tun_sa_out.remove_vpp_config()
        p.tun_sa_in.remove_vpp_config()


class TemplateIpsec6TunIfEsp(TemplateIpsec6TunProtect, TemplateIpsec):
    """IPsec tunnel interface tests"""

    encryption_type = ESP

    def setUp(self):
        super(TemplateIpsec6TunIfEsp, self).setUp()

        self.tun_if = self.pg0

        p = self.ipv6_params
        self.config_network(p)
        self.config_sa_tra(p)
        self.config_protect(p)

    def tearDown(self):
        super(TemplateIpsec6TunIfEsp, self).tearDown()


class TemplateIpsec6TunIfEspUdp(TemplateIpsec6TunProtect, TemplateIpsec):
    """IPsec6 UDP tunnel interface tests"""

    tun4_encrypt_node_name = "esp6-encrypt-tun"
    tun4_decrypt_node_name = ["esp6-decrypt-tun", "esp6-decrypt-tun-post"]
    encryption_type = ESP

    @classmethod
    def setUpClass(cls):
        super(TemplateIpsec6TunIfEspUdp, cls).setUpClass()

    @classmethod
    def tearDownClass(cls):
        super(TemplateIpsec6TunIfEspUdp, cls).tearDownClass()

    def verify_encrypted(self, p, sa, rxs):
        for rx in rxs:
            try:
                # ensure the UDP ports are correct before we decrypt
                # which strips them
                self.assertTrue(rx.haslayer(UDP))
                self.assert_equal(rx[UDP].sport, p.nat_header.sport)
                self.assert_equal(rx[UDP].dport, p.nat_header.dport)

                pkt = sa.decrypt(rx[IP])
                if not pkt.haslayer(IP):
                    pkt = IP(pkt[Raw].load)

                self.assert_packet_checksums_valid(pkt)
                self.assert_equal(
                    pkt[IP].dst, "1111:1111:1111:1111:1111:1111:1111:1111"
                )
                self.assert_equal(pkt[IP].src, self.pg1.remote_ip6)
            except (IndexError, AssertionError):
                self.logger.debug(ppp("Unexpected packet:", rx))
                try:
                    self.logger.debug(ppp("Decrypted packet:", pkt))
                except:
                    pass
                raise

    def config_sa_tra(self, p):
        config_tun_params(p, self.encryption_type, p.tun_if)

        p.tun_sa_out = VppIpsecSA(
            self,
            p.vpp_tun_sa_id,
            p.vpp_tun_spi,
            p.auth_algo_vpp_id,
            p.auth_key,
            p.crypt_algo_vpp_id,
            p.crypt_key,
            self.vpp_esp_protocol,
            flags=p.flags,
            udp_src=p.nat_header.sport,
            udp_dst=p.nat_header.dport,
        )
        p.tun_sa_out.add_vpp_config()

        p.tun_sa_in = VppIpsecSA(
            self,
            p.scapy_tun_sa_id,
            p.scapy_tun_spi,
            p.auth_algo_vpp_id,
            p.auth_key,
            p.crypt_algo_vpp_id,
            p.crypt_key,
            self.vpp_esp_protocol,
            flags=p.flags
            | VppEnum.vl_api_ipsec_sad_flags_t.IPSEC_API_SAD_FLAG_IS_INBOUND,
            udp_src=p.nat_header.sport,
            udp_dst=p.nat_header.dport,
        )
        p.tun_sa_in.add_vpp_config()

    def setUp(self):
        super(TemplateIpsec6TunIfEspUdp, self).setUp()

        p = self.ipv6_params
        p.flags = VppEnum.vl_api_ipsec_sad_flags_t.IPSEC_API_SAD_FLAG_UDP_ENCAP
        p.nat_header = UDP(sport=5454, dport=4500)

        self.tun_if = self.pg0

        self.config_network(p)
        self.config_sa_tra(p)
        self.config_protect(p)

    def tearDown(self):
        super(TemplateIpsec6TunIfEspUdp, self).tearDown()


class TemplateIpsec6TunTfc:
    """IPsec IPv6 tunnel with TFC"""

    def gen_encrypt_pkts6(self, p, sa, sw_intf, src, dst, count=1, payload_size=54):
        return [
            Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac)
            / sa.encrypt(
                IPv6(src=src, dst=dst, hlim=p.inner_hop_limit, fl=p.inner_flow_label)
                / ICMPv6EchoRequest(id=0, seq=1, data="X" * payload_size)
                / Padding(b"Y" * 100)
            )
            for i in range(count)
        ]

    def verify_decrypted6(self, p, rxs):
        for rx in rxs:
            self.assert_equal(rx[IPv6].src, p.remote_tun_if_host)
            self.assert_equal(rx[IPv6].dst, self.pg1.remote_ip6)
            self.assert_equal(rx[IPv6].plen, len(rx[IPv6].payload))
            self.assert_packet_checksums_valid(rx)


class TestIpsec6TunIfEspUdp(TemplateIpsec6TunIfEspUdp, IpsecTun6Tests):
    """Ipsec ESP 6 UDP tests"""

    tun6_input_node = "ipsec6-tun-input"
    tun6_encrypt_node_name = "esp6-encrypt-tun"
    tun6_decrypt_node_name = ["esp6-decrypt-tun", "esp6-decrypt-tun-post"]

    def setUp(self):
        super(TestIpsec6TunIfEspUdp, self).setUp()

    def test_keepalive(self):
        """IPSEC6 NAT Keepalive"""
        self.verify_keepalive(self.ipv6_params)


class TestIpsec6TunIfEspUdpGCM(TemplateIpsec6TunIfEspUdp, IpsecTun6Tests):
    """Ipsec ESP 6 UDP GCM tests"""

    tun6_input_node = "ipsec6-tun-input"
    tun6_encrypt_node_name = "esp6-encrypt-tun"
    tun6_decrypt_node_name = ["esp6-decrypt-tun", "esp6-decrypt-tun-post"]

    def setUp(self):
        super(TestIpsec6TunIfEspUdpGCM, self).setUp()
        p = self.ipv6_params
        p.auth_algo_vpp_id = VppEnum.vl_api_ipsec_integ_alg_t.IPSEC_API_INTEG_ALG_NONE
        p.crypt_algo_vpp_id = (
            VppEnum.vl_api_ipsec_crypto_alg_t.IPSEC_API_CRYPTO_ALG_AES_GCM_256
        )
        p.crypt_algo = "AES-GCM"
        p.auth_algo = "NULL"
        p.crypt_key = b"JPjyOWBeVEQiMe7hJPjyOWBeVEQiMe7h"
        p.salt = 0


class TestIpsec6TunIfEsp1(TemplateIpsec6TunIfEsp, IpsecTun6Tests):
    """Ipsec ESP - TUN tests"""

    tun6_encrypt_node_name = "esp6-encrypt-tun"
    tun6_decrypt_node_name = ["esp6-decrypt-tun", "esp6-decrypt-tun-post"]

    def test_tun_basic46(self):
        """ipsec 4o6 tunnel basic test"""
        self.tun6_encrypt_node_name = "esp6-encrypt-tun"
        self.verify_tun_46(self.params[socket.AF_INET6], count=1)

    def test_tun_burst46(self):
        """ipsec 4o6 tunnel burst test"""
        self.tun6_encrypt_node_name = "esp6-encrypt-tun"
        self.verify_tun_46(self.params[socket.AF_INET6], count=257)


class TestIpsec6TunIfEspHandoff(TemplateIpsec6TunIfEsp, IpsecTun6HandoffTests):
    """Ipsec ESP 6 Handoff tests"""

    tun6_encrypt_node_name = "esp6-encrypt-tun"
    tun6_decrypt_node_name = ["esp6-decrypt-tun", "esp6-decrypt-tun-post"]

    def test_tun_handoff_66_police(self):
        """ESP 6o6 tunnel with policer worker hand-off test"""
        self.vapi.cli("clear errors")
        self.vapi.cli("clear ipsec sa")

        N_PKTS = 15
        p = self.params[socket.AF_INET6]

        action_tx = PolicerAction(
            VppEnum.vl_api_sse2_qos_action_type_t.SSE2_QOS_ACTION_API_TRANSMIT, 0
        )
        policer = VppPolicer(
            self,
            "pol1",
            80,
            0,
            1000,
            0,
            conform_action=action_tx,
            exceed_action=action_tx,
            violate_action=action_tx,
        )
        policer.add_vpp_config()

        # Start policing on tun
        policer.apply_vpp_config(p.tun_if.sw_if_index, Dir.RX, True)

        for pol_bind in [1, 0]:
            policer.bind_vpp_config(pol_bind, True)

            # inject alternately on worker 0 and 1.
            for worker in [0, 1, 0, 1]:
                send_pkts = self.gen_encrypt_pkts6(
                    p,
                    p.scapy_tun_sa,
                    self.tun_if,
                    src=p.remote_tun_if_host,
                    dst=self.pg1.remote_ip6,
                    count=N_PKTS,
                )
                recv_pkts = self.send_and_expect(
                    self.tun_if, send_pkts, self.pg1, worker=worker
                )
                self.verify_decrypted6(p, recv_pkts)
                self.logger.debug(self.vapi.cli("show trace max 100"))

            stats = policer.get_stats()
            stats0 = policer.get_stats(worker=0)
            stats1 = policer.get_stats(worker=1)

            if pol_bind == 1:
                # First pass: Worker 1, should have done all the policing
                self.assertEqual(stats, stats1)

                # Worker 0, should have handed everything off
                self.assertEqual(stats0["conform_packets"], 0)
                self.assertEqual(stats0["exceed_packets"], 0)
                self.assertEqual(stats0["violate_packets"], 0)
            else:
                # Second pass: both workers should have policed equal amounts
                self.assertGreater(stats1["conform_packets"], 0)
                self.assertEqual(stats1["exceed_packets"], 0)
                self.assertGreater(stats1["violate_packets"], 0)

                self.assertGreater(stats0["conform_packets"], 0)
                self.assertEqual(stats0["exceed_packets"], 0)
                self.assertGreater(stats0["violate_packets"], 0)

                self.assertEqual(
                    stats0["conform_packets"] + stats0["violate_packets"],
                    stats1["conform_packets"] + stats1["violate_packets"],
                )

        policer.apply_vpp_config(p.tun_if.sw_if_index, Dir.RX, False)
        policer.remove_vpp_config()


class TestIpsec4TunIfEspHandoff(TemplateIpsec4TunIfEsp, IpsecTun4HandoffTests):
    """Ipsec ESP 4 Handoff tests"""

    tun4_encrypt_node_name = "esp4-encrypt-tun"
    tun4_decrypt_node_name = ["esp4-decrypt-tun", "esp4-decrypt-tun-post"]

    def test_tun_handoff_44_police(self):
        """ESP 4o4 tunnel with policer worker hand-off test"""
        self.vapi.cli("clear errors")
        self.vapi.cli("clear ipsec sa")

        N_PKTS = 15
        p = self.params[socket.AF_INET]

        action_tx = PolicerAction(
            VppEnum.vl_api_sse2_qos_action_type_t.SSE2_QOS_ACTION_API_TRANSMIT, 0
        )
        policer = VppPolicer(
            self,
            "pol1",
            80,
            0,
            1000,
            0,
            conform_action=action_tx,
            exceed_action=action_tx,
            violate_action=action_tx,
        )
        policer.add_vpp_config()

        # Start policing on tun
        policer.apply_vpp_config(p.tun_if.sw_if_index, Dir.RX, True)

        for pol_bind in [1, 0]:
            policer.bind_vpp_config(pol_bind, True)

            # inject alternately on worker 0 and 1.
            for worker in [0, 1, 0, 1]:
                send_pkts = self.gen_encrypt_pkts(
                    p,
                    p.scapy_tun_sa,
                    self.tun_if,
                    src=p.remote_tun_if_host,
                    dst=self.pg1.remote_ip4,
                    count=N_PKTS,
                )
                recv_pkts = self.send_and_expect(
                    self.tun_if, send_pkts, self.pg1, worker=worker
                )
                self.verify_decrypted(p, recv_pkts)
                self.logger.debug(self.vapi.cli("show trace max 100"))

            stats = policer.get_stats()
            stats0 = policer.get_stats(worker=0)
            stats1 = policer.get_stats(worker=1)

            if pol_bind == 1:
                # First pass: Worker 1, should have done all the policing
                self.assertEqual(stats, stats1)

                # Worker 0, should have handed everything off
                self.assertEqual(stats0["conform_packets"], 0)
                self.assertEqual(stats0["exceed_packets"], 0)
                self.assertEqual(stats0["violate_packets"], 0)
            else:
                # Second pass: both workers should have policed equal amounts
                self.assertGreater(stats1["conform_packets"], 0)
                self.assertEqual(stats1["exceed_packets"], 0)
                self.assertGreater(stats1["violate_packets"], 0)

                self.assertGreater(stats0["conform_packets"], 0)
                self.assertEqual(stats0["exceed_packets"], 0)
                self.assertGreater(stats0["violate_packets"], 0)

                self.assertEqual(
                    stats0["conform_packets"] + stats0["violate_packets"],
                    stats1["conform_packets"] + stats1["violate_packets"],
                )

        policer.apply_vpp_config(p.tun_if.sw_if_index, Dir.RX, False)
        policer.remove_vpp_config()


@tag_fixme_vpp_workers
class TestIpsec4MultiTunIfEsp(TemplateIpsec4TunProtect, TemplateIpsec, IpsecTun4):
    """IPsec IPv4 Multi Tunnel interface"""

    encryption_type = ESP
    tun4_encrypt_node_name = "esp4-encrypt-tun"
    tun4_decrypt_node_name = ["esp4-decrypt-tun", "esp4-decrypt-tun-post"]

    def setUp(self):
        super(TestIpsec4MultiTunIfEsp, self).setUp()

        self.tun_if = self.pg0

        self.multi_params = []
        self.pg0.generate_remote_hosts(10)
        self.pg0.configure_ipv4_neighbors()

        for ii in range(10):
            p = copy.copy(self.ipv4_params)

            p.remote_tun_if_host = "1.1.1.%d" % (ii + 1)
            p.scapy_tun_sa_id = p.scapy_tun_sa_id + ii
            p.scapy_tun_spi = p.scapy_tun_spi + ii
            p.vpp_tun_sa_id = p.vpp_tun_sa_id + ii
            p.vpp_tun_spi = p.vpp_tun_spi + ii

            p.scapy_tra_sa_id = p.scapy_tra_sa_id + ii
            p.scapy_tra_spi = p.scapy_tra_spi + ii
            p.vpp_tra_sa_id = p.vpp_tra_sa_id + ii
            p.vpp_tra_spi = p.vpp_tra_spi + ii
            p.tun_dst = self.pg0.remote_hosts[ii].ip4

            self.multi_params.append(p)
            self.config_network(p)
            self.config_sa_tra(p)
            self.config_protect(p)

    def tearDown(self):
        super(TestIpsec4MultiTunIfEsp, self).tearDown()

    def test_tun_44(self):
        """Multiple IPSEC tunnel interfaces"""
        for p in self.multi_params:
            self.verify_tun_44(p, count=127)
            self.assertEqual(p.tun_if.get_rx_stats(), 127)
            self.assertEqual(p.tun_if.get_tx_stats(), 127)

    def test_tun_rr_44(self):
        """Round-robin packets acrros multiple interface"""
        tx = []
        for p in self.multi_params:
            tx = tx + self.gen_encrypt_pkts(
                p,
                p.scapy_tun_sa,
                self.tun_if,
                src=p.remote_tun_if_host,
                dst=self.pg1.remote_ip4,
            )
        rxs = self.send_and_expect(self.tun_if, tx, self.pg1)

        for rx, p in zip(rxs, self.multi_params):
            self.verify_decrypted(p, [rx])

        tx = []
        for p in self.multi_params:
            tx = tx + self.gen_pkts(
                self.pg1, src=self.pg1.remote_ip4, dst=p.remote_tun_if_host
            )
        rxs = self.send_and_expect(self.pg1, tx, self.tun_if)

        for rx, p in zip(rxs, self.multi_params):
            self.verify_encrypted(p, p.vpp_tun_sa, [rx])


class TestIpsec4TunIfEspAll(TemplateIpsec4TunProtect, TemplateIpsec, IpsecTun4):
    """IPsec IPv4 Tunnel interface all Algos"""

    encryption_type = ESP
    tun4_encrypt_node_name = "esp4-encrypt-tun"
    tun4_decrypt_node_name = ["esp4-decrypt-tun", "esp4-decrypt-tun-post"]

    def setUp(self):
        super(TestIpsec4TunIfEspAll, self).setUp()

        self.tun_if = self.pg0
        p = self.ipv4_params

        self.config_network(p)
        self.config_sa_tra(p)
        self.config_protect(p)

    def tearDown(self):
        p = self.ipv4_params
        self.unconfig_protect(p)
        self.unconfig_network(p)
        self.unconfig_sa(p)

        super(TestIpsec4TunIfEspAll, self).tearDown()

    def rekey(self, p):
        #
        # change the key and the SPI
        #
        np = copy.copy(p)
        p.crypt_key = b"X" + p.crypt_key[1:]
        p.scapy_tun_spi += 1
        p.scapy_tun_sa_id += 1
        p.vpp_tun_spi += 1
        p.vpp_tun_sa_id += 1
        p.tun_if.local_spi = p.vpp_tun_spi
        p.tun_if.remote_spi = p.scapy_tun_spi

        config_tun_params(p, self.encryption_type, p.tun_if)

        p.tun_sa_out = VppIpsecSA(
            self,
            p.vpp_tun_sa_id,
            p.vpp_tun_spi,
            p.auth_algo_vpp_id,
            p.auth_key,
            p.crypt_algo_vpp_id,
            p.crypt_key,
            self.vpp_esp_protocol,
            flags=p.flags,
            salt=p.salt,
        )
        p.tun_sa_in = VppIpsecSA(
            self,
            p.scapy_tun_sa_id,
            p.scapy_tun_spi,
            p.auth_algo_vpp_id,
            p.auth_key,
            p.crypt_algo_vpp_id,
            p.crypt_key,
            self.vpp_esp_protocol,
            flags=p.flags,
            salt=p.salt,
        )
        p.tun_sa_in.add_vpp_config()
        p.tun_sa_out.add_vpp_config()

        self.config_protect(p)
        np.tun_sa_out.remove_vpp_config()
        np.tun_sa_in.remove_vpp_config()
        self.logger.info(self.vapi.cli("sh ipsec sa"))

    def test_tun_44(self):
        """IPSEC tunnel all algos"""

        # foreach VPP crypto engine
        engines = ["ia32", "ipsecmb", "openssl"]

        # foreach crypto algorithm
        algos = [
            {
                "vpp-crypto": (
                    VppEnum.vl_api_ipsec_crypto_alg_t.IPSEC_API_CRYPTO_ALG_AES_GCM_128
                ),
                "vpp-integ": (
                    VppEnum.vl_api_ipsec_integ_alg_t.IPSEC_API_INTEG_ALG_NONE
                ),
                "scapy-crypto": "AES-GCM",
                "scapy-integ": "NULL",
                "key": b"JPjyOWBeVEQiMe7h",
                "salt": 3333,
            },
            {
                "vpp-crypto": (
                    VppEnum.vl_api_ipsec_crypto_alg_t.IPSEC_API_CRYPTO_ALG_AES_GCM_192
                ),
                "vpp-integ": (
                    VppEnum.vl_api_ipsec_integ_alg_t.IPSEC_API_INTEG_ALG_NONE
                ),
                "scapy-crypto": "AES-GCM",
                "scapy-integ": "NULL",
                "key": b"JPjyOWBeVEQiMe7hJPjyOWBe",
                "salt": 0,
            },
            {
                "vpp-crypto": (
                    VppEnum.vl_api_ipsec_crypto_alg_t.IPSEC_API_CRYPTO_ALG_AES_GCM_256
                ),
                "vpp-integ": (
                    VppEnum.vl_api_ipsec_integ_alg_t.IPSEC_API_INTEG_ALG_NONE
                ),
                "scapy-crypto": "AES-GCM",
                "scapy-integ": "NULL",
                "key": b"JPjyOWBeVEQiMe7hJPjyOWBeVEQiMe7h",
                "salt": 9999,
            },
            {
                "vpp-crypto": (
                    VppEnum.vl_api_ipsec_crypto_alg_t.IPSEC_API_CRYPTO_ALG_AES_CBC_128
                ),
                "vpp-integ": (
                    VppEnum.vl_api_ipsec_integ_alg_t.IPSEC_API_INTEG_ALG_SHA1_96
                ),
                "scapy-crypto": "AES-CBC",
                "scapy-integ": "HMAC-SHA1-96",
                "salt": 0,
                "key": b"JPjyOWBeVEQiMe7h",
            },
            {
                "vpp-crypto": (
                    VppEnum.vl_api_ipsec_crypto_alg_t.IPSEC_API_CRYPTO_ALG_AES_CBC_192
                ),
                "vpp-integ": (
                    VppEnum.vl_api_ipsec_integ_alg_t.IPSEC_API_INTEG_ALG_SHA_512_256
                ),
                "scapy-crypto": "AES-CBC",
                "scapy-integ": "SHA2-512-256",
                "salt": 0,
                "key": b"JPjyOWBeVEQiMe7hJPjyOWBe",
            },
            {
                "vpp-crypto": (
                    VppEnum.vl_api_ipsec_crypto_alg_t.IPSEC_API_CRYPTO_ALG_AES_CBC_256
                ),
                "vpp-integ": (
                    VppEnum.vl_api_ipsec_integ_alg_t.IPSEC_API_INTEG_ALG_SHA_256_128
                ),
                "scapy-crypto": "AES-CBC",
                "scapy-integ": "SHA2-256-128",
                "salt": 0,
                "key": b"JPjyOWBeVEQiMe7hJPjyOWBeVEQiMe7h",
            },
            {
                "vpp-crypto": (
                    VppEnum.vl_api_ipsec_crypto_alg_t.IPSEC_API_CRYPTO_ALG_NONE
                ),
                "vpp-integ": (
                    VppEnum.vl_api_ipsec_integ_alg_t.IPSEC_API_INTEG_ALG_SHA1_96
                ),
                "scapy-crypto": "NULL",
                "scapy-integ": "HMAC-SHA1-96",
                "salt": 0,
                "key": b"JPjyOWBeVEQiMe7hJPjyOWBeVEQiMe7h",
            },
        ]

        for engine in engines:
            self.vapi.cli("set crypto handler all %s" % engine)

            #
            # loop through each of the algorithms
            #
            for algo in algos:
                # with self.subTest(algo=algo['scapy']):

                p = self.ipv4_params
                p.auth_algo_vpp_id = algo["vpp-integ"]
                p.crypt_algo_vpp_id = algo["vpp-crypto"]
                p.crypt_algo = algo["scapy-crypto"]
                p.auth_algo = algo["scapy-integ"]
                p.crypt_key = algo["key"]
                p.salt = algo["salt"]

                #
                # rekey the tunnel
                #
                self.rekey(p)
                self.verify_tun_44(p, count=127)


class TestIpsec4TunIfEspNoAlgo(TemplateIpsec4TunProtect, TemplateIpsec, IpsecTun4):
    """IPsec IPv4 Tunnel interface no Algos"""

    encryption_type = ESP
    tun4_encrypt_node_name = "esp4-encrypt-tun"
    tun4_decrypt_node_name = ["esp4-decrypt-tun", "esp4-decrypt-tun-post"]

    def setUp(self):
        super(TestIpsec4TunIfEspNoAlgo, self).setUp()

        self.tun_if = self.pg0
        p = self.ipv4_params
        p.auth_algo_vpp_id = VppEnum.vl_api_ipsec_integ_alg_t.IPSEC_API_INTEG_ALG_NONE
        p.auth_algo = "NULL"
        p.auth_key = []

        p.crypt_algo_vpp_id = (
            VppEnum.vl_api_ipsec_crypto_alg_t.IPSEC_API_CRYPTO_ALG_NONE
        )
        p.crypt_algo = "NULL"
        p.crypt_key = []

    def tearDown(self):
        super(TestIpsec4TunIfEspNoAlgo, self).tearDown()

    def test_tun_44(self):
        """IPSec SA with NULL algos"""
        p = self.ipv4_params

        self.config_network(p)
        self.config_sa_tra(p)
        self.config_protect(p)

        tx = self.gen_pkts(self.pg1, src=self.pg1.remote_ip4, dst=p.remote_tun_if_host)
        self.send_and_assert_no_replies(self.pg1, tx)

        self.unconfig_protect(p)
        self.unconfig_sa(p)
        self.unconfig_network(p)


@tag_fixme_vpp_workers
class TestIpsec6MultiTunIfEsp(TemplateIpsec6TunProtect, TemplateIpsec, IpsecTun6):
    """IPsec IPv6 Multi Tunnel interface"""

    encryption_type = ESP
    tun6_encrypt_node_name = "esp6-encrypt-tun"
    tun6_decrypt_node_name = ["esp6-decrypt-tun", "esp6-decrypt-tun-post"]

    def setUp(self):
        super(TestIpsec6MultiTunIfEsp, self).setUp()

        self.tun_if = self.pg0

        self.multi_params = []
        self.pg0.generate_remote_hosts(10)
        self.pg0.configure_ipv6_neighbors()

        for ii in range(10):
            p = copy.copy(self.ipv6_params)

            p.remote_tun_if_host = "1111::%d" % (ii + 1)
            p.scapy_tun_sa_id = p.scapy_tun_sa_id + ii
            p.scapy_tun_spi = p.scapy_tun_spi + ii
            p.vpp_tun_sa_id = p.vpp_tun_sa_id + ii
            p.vpp_tun_spi = p.vpp_tun_spi + ii

            p.scapy_tra_sa_id = p.scapy_tra_sa_id + ii
            p.scapy_tra_spi = p.scapy_tra_spi + ii
            p.vpp_tra_sa_id = p.vpp_tra_sa_id + ii
            p.vpp_tra_spi = p.vpp_tra_spi + ii
            p.tun_dst = self.pg0.remote_hosts[ii].ip6

            self.multi_params.append(p)
            self.config_network(p)
            self.config_sa_tra(p)
            self.config_protect(p)

    def tearDown(self):
        super(TestIpsec6MultiTunIfEsp, self).tearDown()

    def test_tun_66(self):
        """Multiple IPSEC tunnel interfaces"""
        for p in self.multi_params:
            self.verify_tun_66(p, count=127)
            self.assertEqual(p.tun_if.get_rx_stats(), 127)
            self.assertEqual(p.tun_if.get_tx_stats(), 127)


class TestIpsecGreTebIfEsp(TemplateIpsec, IpsecTun4Tests):
    """Ipsec GRE TEB ESP - TUN tests"""

    tun4_encrypt_node_name = "esp4-encrypt-tun"
    tun4_decrypt_node_name = ["esp4-decrypt-tun", "esp4-decrypt-tun-post"]
    encryption_type = ESP
    omac = "00:11:22:33:44:55"

    def gen_encrypt_pkts(self, p, sa, sw_intf, src, dst, count=1, payload_size=100):
        return [
            Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac)
            / sa.encrypt(
                IP(src=self.pg0.remote_ip4, dst=self.pg0.local_ip4)
                / GRE()
                / Ether(dst=self.omac)
                / IP(src="1.1.1.1", dst="1.1.1.2")
                / UDP(sport=1144, dport=2233)
                / Raw(b"X" * payload_size)
            )
            for i in range(count)
        ]

    def gen_pkts(self, sw_intf, src, dst, count=1, payload_size=100):
        return [
            Ether(dst=self.omac)
            / IP(src="1.1.1.1", dst="1.1.1.2")
            / UDP(sport=1144, dport=2233)
            / Raw(b"X" * payload_size)
            for i in range(count)
        ]

    def verify_decrypted(self, p, rxs):
        for rx in rxs:
            self.assert_equal(rx[Ether].dst, self.omac)
            self.assert_equal(rx[IP].dst, "1.1.1.2")

    def verify_encrypted(self, p, sa, rxs):
        for rx in rxs:
            try:
                pkt = sa.decrypt(rx[IP])
                if not pkt.haslayer(IP):
                    pkt = IP(pkt[Raw].load)
                self.assert_packet_checksums_valid(pkt)
                self.assert_equal(pkt[IP].dst, self.pg0.remote_ip4)
                self.assert_equal(pkt[IP].src, self.pg0.local_ip4)
                self.assertTrue(pkt.haslayer(GRE))
                e = pkt[Ether]
                self.assertEqual(e[Ether].dst, self.omac)
                self.assertEqual(e[IP].dst, "1.1.1.2")
            except (IndexError, AssertionError):
                self.logger.debug(ppp("Unexpected packet:", rx))
                try:
                    self.logger.debug(ppp("Decrypted packet:", pkt))
                except:
                    pass
                raise

    def setUp(self):
        super(TestIpsecGreTebIfEsp, self).setUp()

        self.tun_if = self.pg0

        p = self.ipv4_params

        bd1 = VppBridgeDomain(self, 1)
        bd1.add_vpp_config()

        p.tun_sa_out = VppIpsecSA(
            self,
            p.vpp_tun_sa_id,
            p.vpp_tun_spi,
            p.auth_algo_vpp_id,
            p.auth_key,
            p.crypt_algo_vpp_id,
            p.crypt_key,
            self.vpp_esp_protocol,
            self.pg0.local_ip4,
            self.pg0.remote_ip4,
        )
        p.tun_sa_out.add_vpp_config()

        p.tun_sa_in = VppIpsecSA(
            self,
            p.scapy_tun_sa_id,
            p.scapy_tun_spi,
            p.auth_algo_vpp_id,
            p.auth_key,
            p.crypt_algo_vpp_id,
            p.crypt_key,
            self.vpp_esp_protocol,
            self.pg0.remote_ip4,
            self.pg0.local_ip4,
        )
        p.tun_sa_in.add_vpp_config()

        p.tun_if = VppGreInterface(
            self,
            self.pg0.local_ip4,
            self.pg0.remote_ip4,
            type=(VppEnum.vl_api_gre_tunnel_type_t.GRE_API_TUNNEL_TYPE_TEB),
        )
        p.tun_if.add_vpp_config()

        p.tun_protect = VppIpsecTunProtect(self, p.tun_if, p.tun_sa_out, [p.tun_sa_in])

        p.tun_protect.add_vpp_config()

        p.tun_if.admin_up()
        p.tun_if.config_ip4()
        config_tun_params(p, self.encryption_type, p.tun_if)

        VppBridgeDomainPort(self, bd1, p.tun_if).add_vpp_config()
        VppBridgeDomainPort(self, bd1, self.pg1).add_vpp_config()

        self.vapi.cli("clear ipsec sa")
        self.vapi.cli("sh adj")
        self.vapi.cli("sh ipsec tun")

    def tearDown(self):
        p = self.ipv4_params
        p.tun_if.unconfig_ip4()
        super(TestIpsecGreTebIfEsp, self).tearDown()


class TestIpsecGreTebVlanIfEsp(TemplateIpsec, IpsecTun4Tests):
    """Ipsec GRE TEB ESP - TUN tests"""

    tun4_encrypt_node_name = "esp4-encrypt-tun"
    tun4_decrypt_node_name = ["esp4-decrypt-tun", "esp4-decrypt-tun-post"]
    encryption_type = ESP
    omac = "00:11:22:33:44:55"

    def gen_encrypt_pkts(self, p, sa, sw_intf, src, dst, count=1, payload_size=100):
        return [
            Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac)
            / sa.encrypt(
                IP(src=self.pg0.remote_ip4, dst=self.pg0.local_ip4)
                / GRE()
                / Ether(dst=self.omac)
                / IP(src="1.1.1.1", dst="1.1.1.2")
                / UDP(sport=1144, dport=2233)
                / Raw(b"X" * payload_size)
            )
            for i in range(count)
        ]

    def gen_pkts(self, sw_intf, src, dst, count=1, payload_size=100):
        return [
            Ether(dst=self.omac)
            / Dot1Q(vlan=11)
            / IP(src="1.1.1.1", dst="1.1.1.2")
            / UDP(sport=1144, dport=2233)
            / Raw(b"X" * payload_size)
            for i in range(count)
        ]

    def verify_decrypted(self, p, rxs):
        for rx in rxs:
            self.assert_equal(rx[Ether].dst, self.omac)
            self.assert_equal(rx[Dot1Q].vlan, 11)
            self.assert_equal(rx[IP].dst, "1.1.1.2")

    def verify_encrypted(self, p, sa, rxs):
        for rx in rxs:
            try:
                pkt = sa.decrypt(rx[IP])
                if not pkt.haslayer(IP):
                    pkt = IP(pkt[Raw].load)
                self.assert_packet_checksums_valid(pkt)
                self.assert_equal(pkt[IP].dst, self.pg0.remote_ip4)
                self.assert_equal(pkt[IP].src, self.pg0.local_ip4)
                self.assertTrue(pkt.haslayer(GRE))
                e = pkt[Ether]
                self.assertEqual(e[Ether].dst, self.omac)
                self.assertFalse(e.haslayer(Dot1Q))
                self.assertEqual(e[IP].dst, "1.1.1.2")
            except (IndexError, AssertionError):
                self.logger.debug(ppp("Unexpected packet:", rx))
                try:
                    self.logger.debug(ppp("Decrypted packet:", pkt))
                except:
                    pass
                raise

    def setUp(self):
        super(TestIpsecGreTebVlanIfEsp, self).setUp()

        self.tun_if = self.pg0

        p = self.ipv4_params

        bd1 = VppBridgeDomain(self, 1)
        bd1.add_vpp_config()

        self.pg1_11 = VppDot1QSubint(self, self.pg1, 11)
        self.vapi.l2_interface_vlan_tag_rewrite(
            sw_if_index=self.pg1_11.sw_if_index,
            vtr_op=L2_VTR_OP.L2_POP_1,
            push_dot1q=11,
        )
        self.pg1_11.admin_up()

        p.tun_sa_out = VppIpsecSA(
            self,
            p.vpp_tun_sa_id,
            p.vpp_tun_spi,
            p.auth_algo_vpp_id,
            p.auth_key,
            p.crypt_algo_vpp_id,
            p.crypt_key,
            self.vpp_esp_protocol,
            self.pg0.local_ip4,
            self.pg0.remote_ip4,
        )
        p.tun_sa_out.add_vpp_config()

        p.tun_sa_in = VppIpsecSA(
            self,
            p.scapy_tun_sa_id,
            p.scapy_tun_spi,
            p.auth_algo_vpp_id,
            p.auth_key,
            p.crypt_algo_vpp_id,
            p.crypt_key,
            self.vpp_esp_protocol,
            self.pg0.remote_ip4,
            self.pg0.local_ip4,
        )
        p.tun_sa_in.add_vpp_config()

        p.tun_if = VppGreInterface(
            self,
            self.pg0.local_ip4,
            self.pg0.remote_ip4,
            type=(VppEnum.vl_api_gre_tunnel_type_t.GRE_API_TUNNEL_TYPE_TEB),
        )
        p.tun_if.add_vpp_config()

        p.tun_protect = VppIpsecTunProtect(self, p.tun_if, p.tun_sa_out, [p.tun_sa_in])

        p.tun_protect.add_vpp_config()

        p.tun_if.admin_up()
        p.tun_if.config_ip4()
        config_tun_params(p, self.encryption_type, p.tun_if)

        VppBridgeDomainPort(self, bd1, p.tun_if).add_vpp_config()
        VppBridgeDomainPort(self, bd1, self.pg1_11).add_vpp_config()

        self.vapi.cli("clear ipsec sa")

    def tearDown(self):
        p = self.ipv4_params
        p.tun_if.unconfig_ip4()
        super(TestIpsecGreTebVlanIfEsp, self).tearDown()
        self.pg1_11.admin_down()
        self.pg1_11.remove_vpp_config()


class TestIpsecGreTebIfEspTra(TemplateIpsec, IpsecTun4Tests):
    """Ipsec GRE TEB ESP - Tra tests"""

    tun4_encrypt_node_name = "esp4-encrypt-tun"
    tun4_decrypt_node_name = ["esp4-decrypt-tun", "esp4-decrypt-tun-post"]
    encryption_type = ESP
    omac = "00:11:22:33:44:55"

    def gen_encrypt_pkts(self, p, sa, sw_intf, src, dst, count=1, payload_size=100):
        return [
            Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac)
            / sa.encrypt(
                IP(src=self.pg0.remote_ip4, dst=self.pg0.local_ip4)
                / GRE()
                / Ether(dst=self.omac)
                / IP(src="1.1.1.1", dst="1.1.1.2")
                / UDP(sport=1144, dport=2233)
                / Raw(b"X" * payload_size)
            )
            for i in range(count)
        ]

    def gen_pkts(self, sw_intf, src, dst, count=1, payload_size=100):
        return [
            Ether(dst=self.omac)
            / IP(src="1.1.1.1", dst="1.1.1.2")
            / UDP(sport=1144, dport=2233)
            / Raw(b"X" * payload_size)
            for i in range(count)
        ]

    def verify_decrypted(self, p, rxs):
        for rx in rxs:
            self.assert_equal(rx[Ether].dst, self.omac)
            self.assert_equal(rx[IP].dst, "1.1.1.2")

    def verify_encrypted(self, p, sa, rxs):
        for rx in rxs:
            try:
                pkt = sa.decrypt(rx[IP])
                if not pkt.haslayer(IP):
                    pkt = IP(pkt[Raw].load)
                self.assert_packet_checksums_valid(pkt)
                self.assert_equal(pkt[IP].dst, self.pg0.remote_ip4)
                self.assert_equal(pkt[IP].src, self.pg0.local_ip4)
                self.assertTrue(pkt.haslayer(GRE))
                e = pkt[Ether]
                self.assertEqual(e[Ether].dst, self.omac)
                self.assertEqual(e[IP].dst, "1.1.1.2")
            except (IndexError, AssertionError):
                self.logger.debug(ppp("Unexpected packet:", rx))
                try:
                    self.logger.debug(ppp("Decrypted packet:", pkt))
                except:
                    pass
                raise

    def setUp(self):
        super(TestIpsecGreTebIfEspTra, self).setUp()

        self.tun_if = self.pg0

        p = self.ipv4_params

        bd1 = VppBridgeDomain(self, 1)
        bd1.add_vpp_config()

        p.tun_sa_out = VppIpsecSA(
            self,
            p.vpp_tun_sa_id,
            p.vpp_tun_spi,
            p.auth_algo_vpp_id,
            p.auth_key,
            p.crypt_algo_vpp_id,
            p.crypt_key,
            self.vpp_esp_protocol,
        )
        p.tun_sa_out.add_vpp_config()

        p.tun_sa_in = VppIpsecSA(
            self,
            p.scapy_tun_sa_id,
            p.scapy_tun_spi,
            p.auth_algo_vpp_id,
            p.auth_key,
            p.crypt_algo_vpp_id,
            p.crypt_key,
            self.vpp_esp_protocol,
        )
        p.tun_sa_in.add_vpp_config()

        p.tun_if = VppGreInterface(
            self,
            self.pg0.local_ip4,
            self.pg0.remote_ip4,
            type=(VppEnum.vl_api_gre_tunnel_type_t.GRE_API_TUNNEL_TYPE_TEB),
        )
        p.tun_if.add_vpp_config()

        p.tun_protect = VppIpsecTunProtect(self, p.tun_if, p.tun_sa_out, [p.tun_sa_in])

        p.tun_protect.add_vpp_config()

        p.tun_if.admin_up()
        p.tun_if.config_ip4()
        config_tra_params(p, self.encryption_type, p.tun_if)

        VppBridgeDomainPort(self, bd1, p.tun_if).add_vpp_config()
        VppBridgeDomainPort(self, bd1, self.pg1).add_vpp_config()

        self.vapi.cli("clear ipsec sa")

    def tearDown(self):
        p = self.ipv4_params
        p.tun_if.unconfig_ip4()
        super(TestIpsecGreTebIfEspTra, self).tearDown()


class TestIpsecGreTebUdpIfEspTra(TemplateIpsec, IpsecTun4Tests):
    """Ipsec GRE TEB UDP ESP - Tra tests"""

    tun4_encrypt_node_name = "esp4-encrypt-tun"
    tun4_decrypt_node_name = ["esp4-decrypt-tun", "esp4-decrypt-tun-post"]
    encryption_type = ESP
    omac = "00:11:22:33:44:55"

    def gen_encrypt_pkts(self, p, sa, sw_intf, src, dst, count=1, payload_size=100):
        return [
            Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac)
            / sa.encrypt(
                IP(src=self.pg0.remote_ip4, dst=self.pg0.local_ip4)
                / GRE()
                / Ether(dst=self.omac)
                / IP(src="1.1.1.1", dst="1.1.1.2")
                / UDP(sport=1144, dport=2233)
                / Raw(b"X" * payload_size)
            )
            for i in range(count)
        ]

    def gen_pkts(self, sw_intf, src, dst, count=1, payload_size=100):
        return [
            Ether(dst=self.omac)
            / IP(src="1.1.1.1", dst="1.1.1.2")
            / UDP(sport=1144, dport=2233)
            / Raw(b"X" * payload_size)
            for i in range(count)
        ]

    def verify_decrypted(self, p, rxs):
        for rx in rxs:
            self.assert_equal(rx[Ether].dst, self.omac)
            self.assert_equal(rx[IP].dst, "1.1.1.2")

    def verify_encrypted(self, p, sa, rxs):
        for rx in rxs:
            self.assertTrue(rx.haslayer(UDP))
            self.assertEqual(rx[UDP].dport, 4545)
            self.assertEqual(rx[UDP].sport, 5454)
            try:
                pkt = sa.decrypt(rx[IP])
                if not pkt.haslayer(IP):
                    pkt = IP(pkt[Raw].load)
                self.assert_packet_checksums_valid(pkt)
                self.assert_equal(pkt[IP].dst, self.pg0.remote_ip4)
                self.assert_equal(pkt[IP].src, self.pg0.local_ip4)
                self.assertTrue(pkt.haslayer(GRE))
                e = pkt[Ether]
                self.assertEqual(e[Ether].dst, self.omac)
                self.assertEqual(e[IP].dst, "1.1.1.2")
            except (IndexError, AssertionError):
                self.logger.debug(ppp("Unexpected packet:", rx))
                try:
                    self.logger.debug(ppp("Decrypted packet:", pkt))
                except:
                    pass
                raise

    def setUp(self):
        super(TestIpsecGreTebUdpIfEspTra, self).setUp()

        self.tun_if = self.pg0

        p = self.ipv4_params
        p = self.ipv4_params
        p.flags = VppEnum.vl_api_ipsec_sad_flags_t.IPSEC_API_SAD_FLAG_UDP_ENCAP
        p.nat_header = UDP(sport=5454, dport=4545)

        bd1 = VppBridgeDomain(self, 1)
        bd1.add_vpp_config()

        p.tun_sa_out = VppIpsecSA(
            self,
            p.vpp_tun_sa_id,
            p.vpp_tun_spi,
            p.auth_algo_vpp_id,
            p.auth_key,
            p.crypt_algo_vpp_id,
            p.crypt_key,
            self.vpp_esp_protocol,
            flags=p.flags,
            udp_src=5454,
            udp_dst=4545,
        )
        p.tun_sa_out.add_vpp_config()

        p.tun_sa_in = VppIpsecSA(
            self,
            p.scapy_tun_sa_id,
            p.scapy_tun_spi,
            p.auth_algo_vpp_id,
            p.auth_key,
            p.crypt_algo_vpp_id,
            p.crypt_key,
            self.vpp_esp_protocol,
            flags=(
                p.flags | VppEnum.vl_api_ipsec_sad_flags_t.IPSEC_API_SAD_FLAG_IS_INBOUND
            ),
            udp_src=4545,
            udp_dst=5454,
        )
        p.tun_sa_in.add_vpp_config()

        p.tun_if = VppGreInterface(
            self,
            self.pg0.local_ip4,
            self.pg0.remote_ip4,
            type=(VppEnum.vl_api_gre_tunnel_type_t.GRE_API_TUNNEL_TYPE_TEB),
        )
        p.tun_if.add_vpp_config()

        p.tun_protect = VppIpsecTunProtect(self, p.tun_if, p.tun_sa_out, [p.tun_sa_in])

        p.tun_protect.add_vpp_config()

        p.tun_if.admin_up()
        p.tun_if.config_ip4()
        config_tra_params(p, self.encryption_type, p.tun_if)

        VppBridgeDomainPort(self, bd1, p.tun_if).add_vpp_config()
        VppBridgeDomainPort(self, bd1, self.pg1).add_vpp_config()

        self.vapi.cli("clear ipsec sa")
        self.logger.info(self.vapi.cli("sh ipsec sa 0"))

    def tearDown(self):
        p = self.ipv4_params
        p.tun_if.unconfig_ip4()
        super(TestIpsecGreTebUdpIfEspTra, self).tearDown()


class TestIpsecGreIfEsp(TemplateIpsec, IpsecTun4Tests):
    """Ipsec GRE ESP - TUN tests"""

    tun4_encrypt_node_name = "esp4-encrypt-tun"
    tun4_decrypt_node_name = ["esp4-decrypt-tun", "esp4-decrypt-tun-post"]
    encryption_type = ESP

    def gen_encrypt_pkts(self, p, sa, sw_intf, src, dst, count=1, payload_size=100):
        return [
            Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac)
            / sa.encrypt(
                IP(src=self.pg0.remote_ip4, dst=self.pg0.local_ip4)
                / GRE()
                / IP(src=self.pg1.local_ip4, dst=self.pg1.remote_ip4)
                / UDP(sport=1144, dport=2233)
                / Raw(b"X" * payload_size)
            )
            for i in range(count)
        ]

    def gen_pkts(self, sw_intf, src, dst, count=1, payload_size=100):
        return [
            Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac)
            / IP(src="1.1.1.1", dst="1.1.1.2")
            / UDP(sport=1144, dport=2233)
            / Raw(b"X" * payload_size)
            for i in range(count)
        ]

    def verify_decrypted(self, p, rxs):
        for rx in rxs:
            self.assert_equal(rx[Ether].dst, self.pg1.remote_mac)
            self.assert_equal(rx[IP].dst, self.pg1.remote_ip4)

    def verify_encrypted(self, p, sa, rxs):
        for rx in rxs:
            try:
                pkt = sa.decrypt(rx[IP])
                if not pkt.haslayer(IP):
                    pkt = IP(pkt[Raw].load)
                self.assert_packet_checksums_valid(pkt)
                self.assert_equal(pkt[IP].dst, self.pg0.remote_ip4)
                self.assert_equal(pkt[IP].src, self.pg0.local_ip4)
                self.assertTrue(pkt.haslayer(GRE))
                e = pkt[GRE]
                self.assertEqual(e[IP].dst, "1.1.1.2")
            except (IndexError, AssertionError):
                self.logger.debug(ppp("Unexpected packet:", rx))
                try:
                    self.logger.debug(ppp("Decrypted packet:", pkt))
                except:
                    pass
                raise

    def setUp(self):
        super(TestIpsecGreIfEsp, self).setUp()

        self.tun_if = self.pg0

        p = self.ipv4_params

        bd1 = VppBridgeDomain(self, 1)
        bd1.add_vpp_config()

        p.tun_sa_out = VppIpsecSA(
            self,
            p.vpp_tun_sa_id,
            p.vpp_tun_spi,
            p.auth_algo_vpp_id,
            p.auth_key,
            p.crypt_algo_vpp_id,
            p.crypt_key,
            self.vpp_esp_protocol,
            self.pg0.local_ip4,
            self.pg0.remote_ip4,
        )
        p.tun_sa_out.add_vpp_config()

        p.tun_sa_in = VppIpsecSA(
            self,
            p.scapy_tun_sa_id,
            p.scapy_tun_spi,
            p.auth_algo_vpp_id,
            p.auth_key,
            p.crypt_algo_vpp_id,
            p.crypt_key,
            self.vpp_esp_protocol,
            self.pg0.remote_ip4,
            self.pg0.local_ip4,
        )
        p.tun_sa_in.add_vpp_config()

        p.tun_if = VppGreInterface(self, self.pg0.local_ip4, self.pg0.remote_ip4)
        p.tun_if.add_vpp_config()

        p.tun_protect = VppIpsecTunProtect(self, p.tun_if, p.tun_sa_out, [p.tun_sa_in])
        p.tun_protect.add_vpp_config()

        p.tun_if.admin_up()
        p.tun_if.config_ip4()
        config_tun_params(p, self.encryption_type, p.tun_if)

        VppIpRoute(
            self, "1.1.1.2", 32, [VppRoutePath(p.tun_if.remote_ip4, 0xFFFFFFFF)]
        ).add_vpp_config()

    def tearDown(self):
        p = self.ipv4_params
        p.tun_if.unconfig_ip4()
        super(TestIpsecGreIfEsp, self).tearDown()


class TestIpsecGreIfEspTra(TemplateIpsec, IpsecTun4Tests):
    """Ipsec GRE ESP - TRA tests"""

    tun4_encrypt_node_name = "esp4-encrypt-tun"
    tun4_decrypt_node_name = ["esp4-decrypt-tun", "esp4-decrypt-tun-post"]
    encryption_type = ESP

    def gen_encrypt_pkts(self, p, sa, sw_intf, src, dst, count=1, payload_size=100):
        return [
            Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac)
            / sa.encrypt(
                IP(src=self.pg0.remote_ip4, dst=self.pg0.local_ip4)
                / GRE()
                / IP(src=self.pg1.local_ip4, dst=self.pg1.remote_ip4)
                / UDP(sport=1144, dport=2233)
                / Raw(b"X" * payload_size)
            )
            for i in range(count)
        ]

    def gen_encrypt_non_ip_pkts(self, sa, sw_intf, src, dst, count=1, payload_size=100):
        return [
            Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac)
            / sa.encrypt(
                IP(src=self.pg0.remote_ip4, dst=self.pg0.local_ip4)
                / GRE()
                / UDP(sport=1144, dport=2233)
                / Raw(b"X" * payload_size)
            )
            for i in range(count)
        ]

    def gen_pkts(self, sw_intf, src, dst, count=1, payload_size=100):
        return [
            Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac)
            / IP(src="1.1.1.1", dst="1.1.1.2")
            / UDP(sport=1144, dport=2233)
            / Raw(b"X" * payload_size)
            for i in range(count)
        ]

    def verify_decrypted(self, p, rxs):
        for rx in rxs:
            self.assert_equal(rx[Ether].dst, self.pg1.remote_mac)
            self.assert_equal(rx[IP].dst, self.pg1.remote_ip4)

    def verify_encrypted(self, p, sa, rxs):
        for rx in rxs:
            try:
                pkt = sa.decrypt(rx[IP])
                if not pkt.haslayer(IP):
                    pkt = IP(pkt[Raw].load)
                self.assert_packet_checksums_valid(pkt)
                self.assertTrue(pkt.haslayer(GRE))
                e = pkt[GRE]
                self.assertEqual(e[IP].dst, "1.1.1.2")
            except (IndexError, AssertionError):
                self.logger.debug(ppp("Unexpected packet:", rx))
                try:
                    self.logger.debug(ppp("Decrypted packet:", pkt))
                except:
                    pass
                raise

    def setUp(self):
        super(TestIpsecGreIfEspTra, self).setUp()

        self.tun_if = self.pg0

        p = self.ipv4_params

        p.tun_sa_out = VppIpsecSA(
            self,
            p.vpp_tun_sa_id,
            p.vpp_tun_spi,
            p.auth_algo_vpp_id,
            p.auth_key,
            p.crypt_algo_vpp_id,
            p.crypt_key,
            self.vpp_esp_protocol,
        )
        p.tun_sa_out.add_vpp_config()

        p.tun_sa_in = VppIpsecSA(
            self,
            p.scapy_tun_sa_id,
            p.scapy_tun_spi,
            p.auth_algo_vpp_id,
            p.auth_key,
            p.crypt_algo_vpp_id,
            p.crypt_key,
            self.vpp_esp_protocol,
        )
        p.tun_sa_in.add_vpp_config()

        p.tun_if = VppGreInterface(self, self.pg0.local_ip4, self.pg0.remote_ip4)
        p.tun_if.add_vpp_config()

        p.tun_protect = VppIpsecTunProtect(self, p.tun_if, p.tun_sa_out, [p.tun_sa_in])
        p.tun_protect.add_vpp_config()

        p.tun_if.admin_up()
        p.tun_if.config_ip4()
        config_tra_params(p, self.encryption_type, p.tun_if)

        VppIpRoute(
            self, "1.1.1.2", 32, [VppRoutePath(p.tun_if.remote_ip4, 0xFFFFFFFF)]
        ).add_vpp_config()

    def tearDown(self):
        p = self.ipv4_params
        p.tun_if.unconfig_ip4()
        super(TestIpsecGreIfEspTra, self).tearDown()

    def test_gre_non_ip(self):
        p = self.ipv4_params
        tx = self.gen_encrypt_non_ip_pkts(
            p.scapy_tun_sa,
            self.tun_if,
            src=p.remote_tun_if_host,
            dst=self.pg1.remote_ip6,
        )
        self.send_and_assert_no_replies(self.tun_if, tx)
        node_name = "/err/%s/unsup_payload" % self.tun4_decrypt_node_name[0]
        self.assertEqual(1, self.statistics.get_err_counter(node_name))
        err = p.tun_sa_in.get_err("unsup_payload")
        self.assertEqual(err, 1)


class TestIpsecGre6IfEspTra(TemplateIpsec, IpsecTun6Tests):
    """Ipsec GRE ESP - TRA tests"""

    tun6_encrypt_node_name = "esp6-encrypt-tun"
    tun6_decrypt_node_name = ["esp6-decrypt-tun", "esp6-decrypt-tun-post"]
    encryption_type = ESP

    def gen_encrypt_pkts6(self, p, sa, sw_intf, src, dst, count=1, payload_size=100):
        return [
            Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac)
            / sa.encrypt(
                IPv6(src=self.pg0.remote_ip6, dst=self.pg0.local_ip6)
                / GRE()
                / IPv6(src=self.pg1.local_ip6, dst=self.pg1.remote_ip6)
                / UDP(sport=1144, dport=2233)
                / Raw(b"X" * payload_size)
            )
            for i in range(count)
        ]

    def gen_pkts6(self, p, sw_intf, src, dst, count=1, payload_size=100):
        return [
            Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac)
            / IPv6(src="1::1", dst="1::2")
            / UDP(sport=1144, dport=2233)
            / Raw(b"X" * payload_size)
            for i in range(count)
        ]

    def verify_decrypted6(self, p, rxs):
        for rx in rxs:
            self.assert_equal(rx[Ether].dst, self.pg1.remote_mac)
            self.assert_equal(rx[IPv6].dst, self.pg1.remote_ip6)

    def verify_encrypted6(self, p, sa, rxs):
        for rx in rxs:
            try:
                pkt = sa.decrypt(rx[IPv6])
                if not pkt.haslayer(IPv6):
                    pkt = IPv6(pkt[Raw].load)
                self.assert_packet_checksums_valid(pkt)
                self.assertTrue(pkt.haslayer(GRE))
                e = pkt[GRE]
                self.assertEqual(e[IPv6].dst, "1::2")
            except (IndexError, AssertionError):
                self.logger.debug(ppp("Unexpected packet:", rx))
                try:
                    self.logger.debug(ppp("Decrypted packet:", pkt))
                except:
                    pass
                raise

    def setUp(self):
        super(TestIpsecGre6IfEspTra, self).setUp()

        self.tun_if = self.pg0

        p = self.ipv6_params

        bd1 = VppBridgeDomain(self, 1)
        bd1.add_vpp_config()

        p.tun_sa_out = VppIpsecSA(
            self,
            p.vpp_tun_sa_id,
            p.vpp_tun_spi,
            p.auth_algo_vpp_id,
            p.auth_key,
            p.crypt_algo_vpp_id,
            p.crypt_key,
            self.vpp_esp_protocol,
        )
        p.tun_sa_out.add_vpp_config()

        p.tun_sa_in = VppIpsecSA(
            self,
            p.scapy_tun_sa_id,
            p.scapy_tun_spi,
            p.auth_algo_vpp_id,
            p.auth_key,
            p.crypt_algo_vpp_id,
            p.crypt_key,
            self.vpp_esp_protocol,
        )
        p.tun_sa_in.add_vpp_config()

        p.tun_if = VppGreInterface(self, self.pg0.local_ip6, self.pg0.remote_ip6)
        p.tun_if.add_vpp_config()

        p.tun_protect = VppIpsecTunProtect(self, p.tun_if, p.tun_sa_out, [p.tun_sa_in])
        p.tun_protect.add_vpp_config()

        p.tun_if.admin_up()
        p.tun_if.config_ip6()
        config_tra_params(p, self.encryption_type, p.tun_if)

        r = VppIpRoute(
            self,
            "1::2",
            128,
            [
                VppRoutePath(
                    p.tun_if.remote_ip6, 0xFFFFFFFF, proto=DpoProto.DPO_PROTO_IP6
                )
            ],
        )
        r.add_vpp_config()

    def tearDown(self):
        p = self.ipv6_params
        p.tun_if.unconfig_ip6()
        super(TestIpsecGre6IfEspTra, self).tearDown()


class TestIpsecMGreIfEspTra4(TemplateIpsec, IpsecTun4):
    """Ipsec mGRE ESP v4 TRA tests"""

    tun4_encrypt_node_name = "esp4-encrypt-tun"
    tun4_decrypt_node_name = ["esp4-decrypt-tun", "esp4-decrypt-tun-post"]
    encryption_type = ESP

    def gen_encrypt_pkts(self, p, sa, sw_intf, src, dst, count=1, payload_size=100):
        return [
            Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac)
            / sa.encrypt(
                IP(src=p.tun_dst, dst=self.pg0.local_ip4)
                / GRE()
                / IP(src=self.pg1.local_ip4, dst=self.pg1.remote_ip4)
                / UDP(sport=1144, dport=2233)
                / Raw(b"X" * payload_size)
            )
            for i in range(count)
        ]

    def gen_pkts(self, sw_intf, src, dst, count=1, payload_size=100):
        return [
            Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac)
            / IP(src="1.1.1.1", dst=dst)
            / UDP(sport=1144, dport=2233)
            / Raw(b"X" * payload_size)
            for i in range(count)
        ]

    def verify_decrypted(self, p, rxs):
        for rx in rxs:
            self.assert_equal(rx[Ether].dst, self.pg1.remote_mac)
            self.assert_equal(rx[IP].dst, self.pg1.remote_ip4)

    def verify_encrypted(self, p, sa, rxs):
        for rx in rxs:
            try:
                pkt = sa.decrypt(rx[IP])
                if not pkt.haslayer(IP):
                    pkt = IP(pkt[Raw].load)
                self.assert_packet_checksums_valid(pkt)
                self.assertTrue(pkt.haslayer(GRE))
                e = pkt[GRE]
                self.assertEqual(e[IP].dst, p.remote_tun_if_host)
            except (IndexError, AssertionError):
                self.logger.debug(ppp("Unexpected packet:", rx))
                try:
                    self.logger.debug(ppp("Decrypted packet:", pkt))
                except:
                    pass
                raise

    def setUp(self):
        super(TestIpsecMGreIfEspTra4, self).setUp()

        N_NHS = 16
        self.tun_if = self.pg0
        p = self.ipv4_params
        p.tun_if = VppGreInterface(
            self,
            self.pg0.local_ip4,
            "0.0.0.0",
            mode=(VppEnum.vl_api_tunnel_mode_t.TUNNEL_API_MODE_MP),
        )
        p.tun_if.add_vpp_config()
        p.tun_if.admin_up()
        p.tun_if.config_ip4()
        p.tun_if.generate_remote_hosts(N_NHS)
        self.pg0.generate_remote_hosts(N_NHS)
        self.pg0.configure_ipv4_neighbors()

        # setup some SAs for several next-hops on the interface
        self.multi_params = []

        for ii in range(N_NHS):
            p = copy.copy(self.ipv4_params)

            p.remote_tun_if_host = "1.1.1.%d" % (ii + 1)
            p.scapy_tun_sa_id = p.scapy_tun_sa_id + ii
            p.scapy_tun_spi = p.scapy_tun_spi + ii
            p.vpp_tun_sa_id = p.vpp_tun_sa_id + ii
            p.vpp_tun_spi = p.vpp_tun_spi + ii

            p.scapy_tra_sa_id = p.scapy_tra_sa_id + ii
            p.scapy_tra_spi = p.scapy_tra_spi + ii
            p.vpp_tra_sa_id = p.vpp_tra_sa_id + ii
            p.vpp_tra_spi = p.vpp_tra_spi + ii
            p.tun_sa_out = VppIpsecSA(
                self,
                p.vpp_tun_sa_id,
                p.vpp_tun_spi,
                p.auth_algo_vpp_id,
                p.auth_key,
                p.crypt_algo_vpp_id,
                p.crypt_key,
                self.vpp_esp_protocol,
            )
            p.tun_sa_out.add_vpp_config()

            p.tun_sa_in = VppIpsecSA(
                self,
                p.scapy_tun_sa_id,
                p.scapy_tun_spi,
                p.auth_algo_vpp_id,
                p.auth_key,
                p.crypt_algo_vpp_id,
                p.crypt_key,
                self.vpp_esp_protocol,
            )
            p.tun_sa_in.add_vpp_config()

            p.tun_protect = VppIpsecTunProtect(
                self,
                p.tun_if,
                p.tun_sa_out,
                [p.tun_sa_in],
                nh=p.tun_if.remote_hosts[ii].ip4,
            )
            p.tun_protect.add_vpp_config()
            config_tra_params(p, self.encryption_type, p.tun_if)
            self.multi_params.append(p)

            VppIpRoute(
                self,
                p.remote_tun_if_host,
                32,
                [VppRoutePath(p.tun_if.remote_hosts[ii].ip4, p.tun_if.sw_if_index)],
            ).add_vpp_config()

            # in this v4 variant add the teibs after the protect
            p.teib = VppTeib(
                self,
                p.tun_if,
                p.tun_if.remote_hosts[ii].ip4,
                self.pg0.remote_hosts[ii].ip4,
            ).add_vpp_config()
            p.tun_dst = self.pg0.remote_hosts[ii].ip4
        self.logger.info(self.vapi.cli("sh ipsec protect-hash"))

    def tearDown(self):
        p = self.ipv4_params
        p.tun_if.unconfig_ip4()
        super(TestIpsecMGreIfEspTra4, self).tearDown()

    def test_tun_44(self):
        """mGRE IPSEC 44"""
        N_PKTS = 63
        for p in self.multi_params:
            self.verify_tun_44(p, count=N_PKTS)
            p.teib.remove_vpp_config()
            self.verify_tun_dropped_44(p, count=N_PKTS)
            p.teib.add_vpp_config()
            self.verify_tun_44(p, count=N_PKTS)


class TestIpsecMGreIfEspTra6(TemplateIpsec, IpsecTun6):
    """Ipsec mGRE ESP v6 TRA tests"""

    tun6_encrypt_node_name = "esp6-encrypt-tun"
    tun6_decrypt_node_name = ["esp6-decrypt-tun", "esp6-decrypt-tun-post"]
    encryption_type = ESP

    def gen_encrypt_pkts6(self, p, sa, sw_intf, src, dst, count=1, payload_size=100):
        return [
            Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac)
            / sa.encrypt(
                IPv6(src=p.tun_dst, dst=self.pg0.local_ip6)
                / GRE()
                / IPv6(src=self.pg1.local_ip6, dst=self.pg1.remote_ip6)
                / UDP(sport=1144, dport=2233)
                / Raw(b"X" * payload_size)
            )
            for i in range(count)
        ]

    def gen_pkts6(self, p, sw_intf, src, dst, count=1, payload_size=100):
        return [
            Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac)
            / IPv6(src="1::1", dst=dst)
            / UDP(sport=1144, dport=2233)
            / Raw(b"X" * payload_size)
            for i in range(count)
        ]

    def verify_decrypted6(self, p, rxs):
        for rx in rxs:
            self.assert_equal(rx[Ether].dst, self.pg1.remote_mac)
            self.assert_equal(rx[IPv6].dst, self.pg1.remote_ip6)

    def verify_encrypted6(self, p, sa, rxs):
        for rx in rxs:
            try:
                pkt = sa.decrypt(rx[IPv6])
                if not pkt.haslayer(IPv6):
                    pkt = IPv6(pkt[Raw].load)
                self.assert_packet_checksums_valid(pkt)
                self.assertTrue(pkt.haslayer(GRE))
                e = pkt[GRE]
                self.assertEqual(e[IPv6].dst, p.remote_tun_if_host)
            except (IndexError, AssertionError):
                self.logger.debug(ppp("Unexpected packet:", rx))
                try:
                    self.logger.debug(ppp("Decrypted packet:", pkt))
                except:
                    pass
                raise

    def setUp(self):
        super(TestIpsecMGreIfEspTra6, self).setUp()

        self.vapi.cli("set logging class ipsec level debug")

        N_NHS = 16
        self.tun_if = self.pg0
        p = self.ipv6_params
        p.tun_if = VppGreInterface(
            self,
            self.pg0.local_ip6,
            "::",
            mode=(VppEnum.vl_api_tunnel_mode_t.TUNNEL_API_MODE_MP),
        )
        p.tun_if.add_vpp_config()
        p.tun_if.admin_up()
        p.tun_if.config_ip6()
        p.tun_if.generate_remote_hosts(N_NHS)
        self.pg0.generate_remote_hosts(N_NHS)
        self.pg0.configure_ipv6_neighbors()

        # setup some SAs for several next-hops on the interface
        self.multi_params = []

        for ii in range(N_NHS):
            p = copy.copy(self.ipv6_params)

            p.remote_tun_if_host = "1::%d" % (ii + 1)
            p.scapy_tun_sa_id = p.scapy_tun_sa_id + ii
            p.scapy_tun_spi = p.scapy_tun_spi + ii
            p.vpp_tun_sa_id = p.vpp_tun_sa_id + ii
            p.vpp_tun_spi = p.vpp_tun_spi + ii

            p.scapy_tra_sa_id = p.scapy_tra_sa_id + ii
            p.scapy_tra_spi = p.scapy_tra_spi + ii
            p.vpp_tra_sa_id = p.vpp_tra_sa_id + ii
            p.vpp_tra_spi = p.vpp_tra_spi + ii
            p.tun_sa_out = VppIpsecSA(
                self,
                p.vpp_tun_sa_id,
                p.vpp_tun_spi,
                p.auth_algo_vpp_id,
                p.auth_key,
                p.crypt_algo_vpp_id,
                p.crypt_key,
                self.vpp_esp_protocol,
            )
            p.tun_sa_out.add_vpp_config()

            p.tun_sa_in = VppIpsecSA(
                self,
                p.scapy_tun_sa_id,
                p.scapy_tun_spi,
                p.auth_algo_vpp_id,
                p.auth_key,
                p.crypt_algo_vpp_id,
                p.crypt_key,
                self.vpp_esp_protocol,
            )
            p.tun_sa_in.add_vpp_config()

            # in this v6 variant add the teibs first then the protection
            p.tun_dst = self.pg0.remote_hosts[ii].ip6
            VppTeib(
                self, p.tun_if, p.tun_if.remote_hosts[ii].ip6, p.tun_dst
            ).add_vpp_config()

            p.tun_protect = VppIpsecTunProtect(
                self,
                p.tun_if,
                p.tun_sa_out,
                [p.tun_sa_in],
                nh=p.tun_if.remote_hosts[ii].ip6,
            )
            p.tun_protect.add_vpp_config()
            config_tra_params(p, self.encryption_type, p.tun_if)
            self.multi_params.append(p)

            VppIpRoute(
                self,
                p.remote_tun_if_host,
                128,
                [VppRoutePath(p.tun_if.remote_hosts[ii].ip6, p.tun_if.sw_if_index)],
            ).add_vpp_config()
            p.tun_dst = self.pg0.remote_hosts[ii].ip6

        self.logger.info(self.vapi.cli("sh log"))
        self.logger.info(self.vapi.cli("sh ipsec protect-hash"))
        self.logger.info(self.vapi.cli("sh adj 41"))

    def tearDown(self):
        p = self.ipv6_params
        p.tun_if.unconfig_ip6()
        super(TestIpsecMGreIfEspTra6, self).tearDown()

    def test_tun_66(self):
        """mGRE IPSec 66"""
        for p in self.multi_params:
            self.verify_tun_66(p, count=63)


@tag_fixme_vpp_workers
class TestIpsec4TunProtect(TemplateIpsec, TemplateIpsec4TunProtect, IpsecTun4):
    """IPsec IPv4 Tunnel protect - transport mode"""

    def setUp(self):
        super(TestIpsec4TunProtect, self).setUp()

        self.tun_if = self.pg0

    def tearDown(self):
        super(TestIpsec4TunProtect, self).tearDown()

    def test_tun_44(self):
        """IPSEC tunnel protect"""

        p = self.ipv4_params

        self.config_network(p)
        self.config_sa_tra(p)
        self.config_protect(p)

        self.verify_tun_44(p, count=127)
        self.assertEqual(p.tun_if.get_rx_stats(), 127)
        self.assertEqual(p.tun_if.get_tx_stats(), 127)

        self.vapi.cli("clear ipsec sa")
        self.verify_tun_64(p, count=127)
        self.assertEqual(p.tun_if.get_rx_stats(), 254)
        self.assertEqual(p.tun_if.get_tx_stats(), 254)

        # rekey - create new SAs and update the tunnel protection
        np = copy.copy(p)
        np.crypt_key = b"X" + p.crypt_key[1:]
        np.scapy_tun_spi += 100
        np.scapy_tun_sa_id += 1
        np.vpp_tun_spi += 100
        np.vpp_tun_sa_id += 1
        np.tun_if.local_spi = p.vpp_tun_spi
        np.tun_if.remote_spi = p.scapy_tun_spi

        self.config_sa_tra(np)
        self.config_protect(np)
        self.unconfig_sa(p)

        self.verify_tun_44(np, count=127)
        self.assertEqual(p.tun_if.get_rx_stats(), 381)
        self.assertEqual(p.tun_if.get_tx_stats(), 381)

        # teardown
        self.unconfig_protect(np)
        self.unconfig_sa(np)
        self.unconfig_network(p)


@tag_fixme_vpp_workers
class TestIpsec4TunProtectTfc(TemplateIpsec4TunTfc, TestIpsec4TunProtect):
    """IPsec IPv4 Tunnel protect with TFC - transport mode"""


@tag_fixme_vpp_workers
class TestIpsec4TunProtectUdp(TemplateIpsec, TemplateIpsec4TunProtect, IpsecTun4):
    """IPsec IPv4 UDP Tunnel protect - transport mode"""

    def setUp(self):
        super(TestIpsec4TunProtectUdp, self).setUp()

        self.tun_if = self.pg0

        p = self.ipv4_params
        p.flags = VppEnum.vl_api_ipsec_sad_flags_t.IPSEC_API_SAD_FLAG_UDP_ENCAP
        p.nat_header = UDP(sport=4500, dport=4500)
        self.config_network(p)
        self.config_sa_tra(p)
        self.config_protect(p)

    def tearDown(self):
        p = self.ipv4_params
        self.unconfig_protect(p)
        self.unconfig_sa(p)
        self.unconfig_network(p)
        super(TestIpsec4TunProtectUdp, self).tearDown()

    def verify_encrypted(self, p, sa, rxs):
        # ensure encrypted packets are recieved with the default UDP ports
        for rx in rxs:
            self.assertEqual(rx[UDP].sport, 4500)
            self.assertEqual(rx[UDP].dport, 4500)
        super(TestIpsec4TunProtectUdp, self).verify_encrypted(p, sa, rxs)

    def test_tun_44(self):
        """IPSEC UDP tunnel protect"""

        p = self.ipv4_params

        self.verify_tun_44(p, count=127)
        self.assertEqual(p.tun_if.get_rx_stats(), 127)
        self.assertEqual(p.tun_if.get_tx_stats(), 127)

    def test_keepalive(self):
        """IPSEC NAT Keepalive"""
        self.verify_keepalive(self.ipv4_params)


@tag_fixme_vpp_workers
class TestIpsec4TunProtectUdpTfc(TemplateIpsec4TunTfc, TestIpsec4TunProtectUdp):
    """IPsec IPv4 UDP Tunnel protect with TFC - transport mode"""


@tag_fixme_vpp_workers
class TestIpsec4TunProtectTun(TemplateIpsec, TemplateIpsec4TunProtect, IpsecTun4):
    """IPsec IPv4 Tunnel protect - tunnel mode"""

    encryption_type = ESP
    tun4_encrypt_node_name = "esp4-encrypt-tun"
    tun4_decrypt_node_name = ["esp4-decrypt-tun", "esp4-decrypt-tun-post"]

    def setUp(self):
        super(TestIpsec4TunProtectTun, self).setUp()

        self.tun_if = self.pg0

    def tearDown(self):
        super(TestIpsec4TunProtectTun, self).tearDown()

    def gen_encrypt_pkts(self, p, sa, sw_intf, src, dst, count=1, payload_size=100):
        return [
            Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac)
            / sa.encrypt(
                IP(src=sw_intf.remote_ip4, dst=sw_intf.local_ip4)
                / IP(src=src, dst=dst)
                / UDP(sport=1144, dport=2233)
                / Raw(b"X" * payload_size)
            )
            for i in range(count)
        ]

    def gen_pkts(self, sw_intf, src, dst, count=1, payload_size=100):
        return [
            Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac)
            / IP(src=src, dst=dst)
            / UDP(sport=1144, dport=2233)
            / Raw(b"X" * payload_size)
            for i in range(count)
        ]

    def verify_decrypted(self, p, rxs):
        for rx in rxs:
            self.assert_equal(rx[IP].dst, self.pg1.remote_ip4)
            self.assert_equal(rx[IP].src, p.remote_tun_if_host)
            self.assert_packet_checksums_valid(rx)

    def verify_encrypted(self, p, sa, rxs):
        for rx in rxs:
            try:
                pkt = sa.decrypt(rx[IP])
                if not pkt.haslayer(IP):
                    pkt = IP(pkt[Raw].load)
                self.assert_packet_checksums_valid(pkt)
                self.assert_equal(pkt[IP].dst, self.pg0.remote_ip4)
                self.assert_equal(pkt[IP].src, self.pg0.local_ip4)
                inner = pkt[IP].payload
                self.assertEqual(inner[IP][IP].dst, p.remote_tun_if_host)

            except (IndexError, AssertionError):
                self.logger.debug(ppp("Unexpected packet:", rx))
                try:
                    self.logger.debug(ppp("Decrypted packet:", pkt))
                except:
                    pass
                raise

    def test_tun_44(self):
        """IPSEC tunnel protect"""

        p = self.ipv4_params

        self.config_network(p)
        self.config_sa_tun(p)
        self.config_protect(p)

        # also add an output features on the tunnel and physical interface
        # so we test they still work
        r_all = AclRule(True, src_prefix="0.0.0.0/0", dst_prefix="0.0.0.0/0", proto=0)
        a = VppAcl(self, [r_all]).add_vpp_config()

        VppAclInterface(self, self.pg0.sw_if_index, [a]).add_vpp_config()
        VppAclInterface(self, p.tun_if.sw_if_index, [a]).add_vpp_config()

        self.verify_tun_44(p, count=127)

        self.assertEqual(p.tun_if.get_rx_stats(), 127)
        self.assertEqual(p.tun_if.get_tx_stats(), 127)

        # rekey - create new SAs and update the tunnel protection
        np = copy.copy(p)
        np.crypt_key = b"X" + p.crypt_key[1:]
        np.scapy_tun_spi += 100
        np.scapy_tun_sa_id += 1
        np.vpp_tun_spi += 100
        np.vpp_tun_sa_id += 1
        np.tun_if.local_spi = p.vpp_tun_spi
        np.tun_if.remote_spi = p.scapy_tun_spi

        self.config_sa_tun(np)
        self.config_protect(np)
        self.unconfig_sa(p)

        self.verify_tun_44(np, count=127)
        self.assertEqual(p.tun_if.get_rx_stats(), 254)
        self.assertEqual(p.tun_if.get_tx_stats(), 254)

        # teardown
        self.unconfig_protect(np)
        self.unconfig_sa(np)
        self.unconfig_network(p)


class TestIpsec4TunProtectTunDrop(TemplateIpsec, TemplateIpsec4TunProtect, IpsecTun4):
    """IPsec IPv4 Tunnel protect - tunnel mode - drop"""

    encryption_type = ESP
    tun4_encrypt_node_name = "esp4-encrypt-tun"
    tun4_decrypt_node_name = ["esp4-decrypt-tun", "esp4-decrypt-tun-post"]

    def setUp(self):
        super(TestIpsec4TunProtectTunDrop, self).setUp()

        self.tun_if = self.pg0

    def tearDown(self):
        super(TestIpsec4TunProtectTunDrop, self).tearDown()

    def gen_encrypt_pkts(self, p, sa, sw_intf, src, dst, count=1, payload_size=100):
        return [
            Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac)
            / sa.encrypt(
                IP(src=sw_intf.remote_ip4, dst="5.5.5.5")
                / IP(src=src, dst=dst)
                / UDP(sport=1144, dport=2233)
                / Raw(b"X" * payload_size)
            )
            for i in range(count)
        ]

    def test_tun_drop_44(self):
        """IPSEC tunnel protect bogus tunnel header"""

        p = self.ipv4_params

        self.config_network(p)
        self.config_sa_tun(p)
        self.config_protect(p)

        tx = self.gen_encrypt_pkts(
            p,
            p.scapy_tun_sa,
            self.tun_if,
            src=p.remote_tun_if_host,
            dst=self.pg1.remote_ip4,
            count=63,
        )
        self.send_and_assert_no_replies(self.tun_if, tx)

        # teardown
        self.unconfig_protect(p)
        self.unconfig_sa(p)
        self.unconfig_network(p)


@tag_fixme_vpp_workers
class TestIpsec6TunProtect(TemplateIpsec, TemplateIpsec6TunProtect, IpsecTun6):
    """IPsec IPv6 Tunnel protect - transport mode"""

    encryption_type = ESP
    tun6_encrypt_node_name = "esp6-encrypt-tun"
    tun6_decrypt_node_name = ["esp6-decrypt-tun", "esp6-decrypt-tun-post"]

    def setUp(self):
        super(TestIpsec6TunProtect, self).setUp()

        self.tun_if = self.pg0

    def tearDown(self):
        super(TestIpsec6TunProtect, self).tearDown()

    def test_tun_66(self):
        """IPSEC tunnel protect 6o6"""

        p = self.ipv6_params

        self.config_network(p)
        self.config_sa_tra(p)
        self.config_protect(p)

        self.verify_tun_66(p, count=127)
        self.assertEqual(p.tun_if.get_rx_stats(), 127)
        self.assertEqual(p.tun_if.get_tx_stats(), 127)

        # rekey - create new SAs and update the tunnel protection
        np = copy.copy(p)
        np.crypt_key = b"X" + p.crypt_key[1:]
        np.scapy_tun_spi += 100
        np.scapy_tun_sa_id += 1
        np.vpp_tun_spi += 100
        np.vpp_tun_sa_id += 1
        np.tun_if.local_spi = p.vpp_tun_spi
        np.tun_if.remote_spi = p.scapy_tun_spi

        self.config_sa_tra(np)
        self.config_protect(np)
        self.unconfig_sa(p)

        self.verify_tun_66(np, count=127)
        self.assertEqual(p.tun_if.get_rx_stats(), 254)
        self.assertEqual(p.tun_if.get_tx_stats(), 254)

        # bounce the interface state
        p.tun_if.admin_down()
        self.verify_drop_tun_66(np, count=127)
        node = "/err/ipsec6-tun-input/disabled"
        self.assertEqual(127, self.statistics.get_err_counter(node))
        p.tun_if.admin_up()
        self.verify_tun_66(np, count=127)

        # 3 phase rekey
        #  1) add two input SAs [old, new]
        #  2) swap output SA to [new]
        #  3) use only [new] input SA
        np3 = copy.copy(np)
        np3.crypt_key = b"Z" + p.crypt_key[1:]
        np3.scapy_tun_spi += 100
        np3.scapy_tun_sa_id += 1
        np3.vpp_tun_spi += 100
        np3.vpp_tun_sa_id += 1
        np3.tun_if.local_spi = p.vpp_tun_spi
        np3.tun_if.remote_spi = p.scapy_tun_spi

        self.config_sa_tra(np3)

        # step 1;
        p.tun_protect.update_vpp_config(np.tun_sa_out, [np.tun_sa_in, np3.tun_sa_in])
        self.verify_tun_66(np, np, count=127)
        self.verify_tun_66(np3, np, count=127)

        # step 2;
        p.tun_protect.update_vpp_config(np3.tun_sa_out, [np.tun_sa_in, np3.tun_sa_in])
        self.verify_tun_66(np, np3, count=127)
        self.verify_tun_66(np3, np3, count=127)

        # step 1;
        p.tun_protect.update_vpp_config(np3.tun_sa_out, [np3.tun_sa_in])
        self.verify_tun_66(np3, np3, count=127)
        self.verify_drop_tun_rx_66(np, count=127)

        self.assertEqual(p.tun_if.get_rx_stats(), 127 * 9)
        self.assertEqual(p.tun_if.get_tx_stats(), 127 * 8)
        self.unconfig_sa(np)

        # teardown
        self.unconfig_protect(np3)
        self.unconfig_sa(np3)
        self.unconfig_network(p)

    def test_tun_46(self):
        """IPSEC tunnel protect 4o6"""

        p = self.ipv6_params

        self.config_network(p)
        self.config_sa_tra(p)
        self.config_protect(p)

        self.verify_tun_46(p, count=127)
        self.assertEqual(p.tun_if.get_rx_stats(), 127)
        self.assertEqual(p.tun_if.get_tx_stats(), 127)

        # teardown
        self.unconfig_protect(p)
        self.unconfig_sa(p)
        self.unconfig_network(p)


@tag_fixme_vpp_workers
class TestIpsec6TunProtectTfc(TemplateIpsec6TunTfc, TestIpsec6TunProtect):
    """IPsec IPv6 Tunnel protect with TFC - transport mode"""


@tag_fixme_vpp_workers
class TestIpsec6TunProtectTun(TemplateIpsec, TemplateIpsec6TunProtect, IpsecTun6):
    """IPsec IPv6 Tunnel protect - tunnel mode"""

    encryption_type = ESP
    tun6_encrypt_node_name = "esp6-encrypt-tun"
    tun6_decrypt_node_name = ["esp6-decrypt-tun", "esp6-decrypt-tun-post"]

    def setUp(self):
        super(TestIpsec6TunProtectTun, self).setUp()

        self.tun_if = self.pg0

    def tearDown(self):
        super(TestIpsec6TunProtectTun, self).tearDown()

    def gen_encrypt_pkts6(self, p, sa, sw_intf, src, dst, count=1, payload_size=100):
        return [
            Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac)
            / sa.encrypt(
                IPv6(src=sw_intf.remote_ip6, dst=sw_intf.local_ip6)
                / IPv6(src=src, dst=dst)
                / UDP(sport=1166, dport=2233)
                / Raw(b"X" * payload_size)
            )
            for i in range(count)
        ]

    def gen_pkts6(self, p, sw_intf, src, dst, count=1, payload_size=100):
        return [
            Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac)
            / IPv6(src=src, dst=dst)
            / UDP(sport=1166, dport=2233)
            / Raw(b"X" * payload_size)
            for i in range(count)
        ]

    def verify_decrypted6(self, p, rxs):
        for rx in rxs:
            self.assert_equal(rx[IPv6].dst, self.pg1.remote_ip6)
            self.assert_equal(rx[IPv6].src, p.remote_tun_if_host)
            self.assert_packet_checksums_valid(rx)

    def verify_encrypted6(self, p, sa, rxs):
        for rx in rxs:
            try:
                pkt = sa.decrypt(rx[IPv6])
                if not pkt.haslayer(IPv6):
                    pkt = IPv6(pkt[Raw].load)
                self.assert_packet_checksums_valid(pkt)
                self.assert_equal(pkt[IPv6].dst, self.pg0.remote_ip6)
                self.assert_equal(pkt[IPv6].src, self.pg0.local_ip6)
                inner = pkt[IPv6].payload
                self.assertEqual(inner[IPv6][IPv6].dst, p.remote_tun_if_host)

            except (IndexError, AssertionError):
                self.logger.debug(ppp("Unexpected packet:", rx))
                try:
                    self.logger.debug(ppp("Decrypted packet:", pkt))
                except:
                    pass
                raise

    def test_tun_66(self):
        """IPSEC tunnel protect"""

        p = self.ipv6_params

        self.config_network(p)
        self.config_sa_tun(p)
        self.config_protect(p)

        self.verify_tun_66(p, count=127)

        self.assertEqual(p.tun_if.get_rx_stats(), 127)
        self.assertEqual(p.tun_if.get_tx_stats(), 127)

        # rekey - create new SAs and update the tunnel protection
        np = copy.copy(p)
        np.crypt_key = b"X" + p.crypt_key[1:]
        np.scapy_tun_spi += 100
        np.scapy_tun_sa_id += 1
        np.vpp_tun_spi += 100
        np.vpp_tun_sa_id += 1
        np.tun_if.local_spi = p.vpp_tun_spi
        np.tun_if.remote_spi = p.scapy_tun_spi

        self.config_sa_tun(np)
        self.config_protect(np)
        self.unconfig_sa(p)

        self.verify_tun_66(np, count=127)
        self.assertEqual(p.tun_if.get_rx_stats(), 254)
        self.assertEqual(p.tun_if.get_tx_stats(), 254)

        # teardown
        self.unconfig_protect(np)
        self.unconfig_sa(np)
        self.unconfig_network(p)


class TestIpsec6TunProtectTunDrop(TemplateIpsec, TemplateIpsec6TunProtect, IpsecTun6):
    """IPsec IPv6 Tunnel protect - tunnel mode - drop"""

    encryption_type = ESP
    tun6_encrypt_node_name = "esp6-encrypt-tun"
    tun6_decrypt_node_name = ["esp6-decrypt-tun", "esp6-decrypt-tun-post"]

    def setUp(self):
        super(TestIpsec6TunProtectTunDrop, self).setUp()

        self.tun_if = self.pg0

    def tearDown(self):
        super(TestIpsec6TunProtectTunDrop, self).tearDown()

    def gen_encrypt_pkts6(self, p, sa, sw_intf, src, dst, count=1, payload_size=100):
        # the IP destination of the revelaed packet does not match
        # that assigned to the tunnel
        return [
            Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac)
            / sa.encrypt(
                IPv6(src=sw_intf.remote_ip6, dst="5::5")
                / IPv6(src=src, dst=dst)
                / UDP(sport=1144, dport=2233)
                / Raw(b"X" * payload_size)
            )
            for i in range(count)
        ]

    def test_tun_drop_66(self):
        """IPSEC 6 tunnel protect bogus tunnel header"""

        p = self.ipv6_params

        self.config_network(p)
        self.config_sa_tun(p)
        self.config_protect(p)

        tx = self.gen_encrypt_pkts6(
            p,
            p.scapy_tun_sa,
            self.tun_if,
            src=p.remote_tun_if_host,
            dst=self.pg1.remote_ip6,
            count=63,
        )
        self.send_and_assert_no_replies(self.tun_if, tx)

        self.unconfig_protect(p)
        self.unconfig_sa(p)
        self.unconfig_network(p)


class TemplateIpsecItf4(object):
    """IPsec Interface IPv4"""

    encryption_type = ESP
    tun4_encrypt_node_name = "esp4-encrypt-tun"
    tun4_decrypt_node_name = ["esp4-decrypt-tun", "esp4-decrypt-tun-post"]
    tun4_input_node = "ipsec4-tun-input"

    def config_sa_tun(self, p, src, dst):
        config_tun_params(p, self.encryption_type, None, src, dst)

        p.tun_sa_out = VppIpsecSA(
            self,
            p.vpp_tun_sa_id,
            p.vpp_tun_spi,
            p.auth_algo_vpp_id,
            p.auth_key,
            p.crypt_algo_vpp_id,
            p.crypt_key,
            self.vpp_esp_protocol,
            src,
            dst,
            flags=p.flags,
        )
        p.tun_sa_out.add_vpp_config()

        p.tun_sa_in = VppIpsecSA(
            self,
            p.scapy_tun_sa_id,
            p.scapy_tun_spi,
            p.auth_algo_vpp_id,
            p.auth_key,
            p.crypt_algo_vpp_id,
            p.crypt_key,
            self.vpp_esp_protocol,
            dst,
            src,
            flags=p.flags
            | VppEnum.vl_api_ipsec_sad_flags_t.IPSEC_API_SAD_FLAG_IS_INBOUND,
        )
        p.tun_sa_in.add_vpp_config()

    def config_protect(self, p):
        p.tun_protect = VppIpsecTunProtect(self, p.tun_if, p.tun_sa_out, [p.tun_sa_in])
        p.tun_protect.add_vpp_config()

    def config_network(self, p, instance=0xFFFFFFFF):
        p.tun_if = VppIpsecInterface(self, instance=instance)

        p.tun_if.add_vpp_config()
        p.tun_if.admin_up()
        p.tun_if.config_ip4()
        p.tun_if.config_ip6()

        p.route = VppIpRoute(
            self,
            p.remote_tun_if_host,
            32,
            [VppRoutePath(p.tun_if.remote_ip4, 0xFFFFFFFF)],
        )
        p.route.add_vpp_config()
        r = VppIpRoute(
            self,
            p.remote_tun_if_host6,
            128,
            [
                VppRoutePath(
                    p.tun_if.remote_ip6, 0xFFFFFFFF, proto=DpoProto.DPO_PROTO_IP6
                )
            ],
        )
        r.add_vpp_config()

    def unconfig_network(self, p):
        p.route.remove_vpp_config()
        p.tun_if.remove_vpp_config()

    def unconfig_protect(self, p):
        p.tun_protect.remove_vpp_config()

    def unconfig_sa(self, p):
        p.tun_sa_out.remove_vpp_config()
        p.tun_sa_in.remove_vpp_config()


@tag_fixme_vpp_workers
class TestIpsecItf4(TemplateIpsec, TemplateIpsecItf4, IpsecTun4):
    """IPsec Interface IPv4"""

    def setUp(self):
        super(TestIpsecItf4, self).setUp()

        self.tun_if = self.pg0

    def tearDown(self):
        super(TestIpsecItf4, self).tearDown()

    def test_tun_instance_44(self):
        p = self.ipv4_params
        self.config_network(p, instance=3)

        with self.assertRaises(CliFailedCommandError):
            self.vapi.cli("show interface ipsec0")

        output = self.vapi.cli("show interface ipsec3")
        self.assertTrue("unknown" not in output)

        self.unconfig_network(p)

    def test_tun_44(self):
        """IPSEC interface IPv4"""

        n_pkts = 127
        p = self.ipv4_params

        self.config_network(p)
        config_tun_params(
            p, self.encryption_type, None, self.pg0.local_ip4, self.pg0.remote_ip4
        )
        self.verify_tun_dropped_44(p, count=n_pkts)
        self.config_sa_tun(p, self.pg0.local_ip4, self.pg0.remote_ip4)
        self.config_protect(p)

        self.verify_tun_44(p, count=n_pkts)
        self.assertEqual(p.tun_if.get_rx_stats(), n_pkts)
        self.assertEqual(p.tun_if.get_tx_stats(), n_pkts)

        p.tun_if.admin_down()
        self.verify_tun_dropped_44(p, count=n_pkts)
        p.tun_if.admin_up()
        self.verify_tun_44(p, count=n_pkts)

        self.assertEqual(p.tun_if.get_rx_stats(), 3 * n_pkts)
        self.assertEqual(p.tun_if.get_tx_stats(), 2 * n_pkts)

        # it's a v6 packet when its encrypted
        self.tun4_encrypt_node_name = "esp6-encrypt-tun"

        self.verify_tun_64(p, count=n_pkts)
        self.assertEqual(p.tun_if.get_rx_stats(), 4 * n_pkts)
        self.assertEqual(p.tun_if.get_tx_stats(), 3 * n_pkts)

        self.tun4_encrypt_node_name = "esp4-encrypt-tun"

        # update the SA tunnel
        config_tun_params(
            p, self.encryption_type, None, self.pg2.local_ip4, self.pg2.remote_ip4
        )
        p.tun_sa_in.update_vpp_config(
            is_tun=True, tun_src=self.pg2.remote_ip4, tun_dst=self.pg2.local_ip4
        )
        p.tun_sa_out.update_vpp_config(
            is_tun=True, tun_src=self.pg2.local_ip4, tun_dst=self.pg2.remote_ip4
        )
        self.verify_tun_44(p, count=n_pkts)
        self.assertEqual(p.tun_if.get_rx_stats(), 5 * n_pkts)
        self.assertEqual(p.tun_if.get_tx_stats(), 4 * n_pkts)

        self.vapi.cli("clear interfaces")

        # rekey - create new SAs and update the tunnel protection
        np = copy.copy(p)
        np.crypt_key = b"X" + p.crypt_key[1:]
        np.scapy_tun_spi += 100
        np.scapy_tun_sa_id += 1
        np.vpp_tun_spi += 100
        np.vpp_tun_sa_id += 1
        np.tun_if.local_spi = p.vpp_tun_spi
        np.tun_if.remote_spi = p.scapy_tun_spi

        self.config_sa_tun(np, self.pg0.local_ip4, self.pg0.remote_ip4)
        self.config_protect(np)
        self.unconfig_sa(p)

        self.verify_tun_44(np, count=n_pkts)
        self.assertEqual(p.tun_if.get_rx_stats(), n_pkts)
        self.assertEqual(p.tun_if.get_tx_stats(), n_pkts)

        # teardown
        self.unconfig_protect(np)
        self.unconfig_sa(np)
        self.unconfig_network(p)

    def test_tun_44_null(self):
        """IPSEC interface IPv4 NULL auth/crypto"""

        n_pkts = 127
        p = copy.copy(self.ipv4_params)

        p.auth_algo_vpp_id = VppEnum.vl_api_ipsec_integ_alg_t.IPSEC_API_INTEG_ALG_NONE
        p.crypt_algo_vpp_id = (
            VppEnum.vl_api_ipsec_crypto_alg_t.IPSEC_API_CRYPTO_ALG_NONE
        )
        p.crypt_algo = "NULL"
        p.auth_algo = "NULL"

        self.config_network(p)
        self.config_sa_tun(p, self.pg0.local_ip4, self.pg0.remote_ip4)
        self.config_protect(p)

        self.logger.info(self.vapi.cli("sh ipsec sa"))
        self.verify_tun_44(p, count=n_pkts)

        # teardown
        self.unconfig_protect(p)
        self.unconfig_sa(p)
        self.unconfig_network(p)

    def test_tun_44_police(self):
        """IPSEC interface IPv4 with input policer"""
        n_pkts = 127
        p = self.ipv4_params

        self.config_network(p)
        self.config_sa_tun(p, self.pg0.local_ip4, self.pg0.remote_ip4)
        self.config_protect(p)

        action_tx = PolicerAction(
            VppEnum.vl_api_sse2_qos_action_type_t.SSE2_QOS_ACTION_API_TRANSMIT, 0
        )
        policer = VppPolicer(
            self,
            "pol1",
            80,
            0,
            1000,
            0,
            conform_action=action_tx,
            exceed_action=action_tx,
            violate_action=action_tx,
        )
        policer.add_vpp_config()

        # Start policing on tun
        policer.apply_vpp_config(p.tun_if.sw_if_index, Dir.RX, True)

        self.verify_tun_44(p, count=n_pkts)
        self.assertEqual(p.tun_if.get_rx_stats(), n_pkts)
        self.assertEqual(p.tun_if.get_tx_stats(), n_pkts)

        stats = policer.get_stats()

        # Single rate, 2 colour policer - expect conform, violate but no exceed
        self.assertGreater(stats["conform_packets"], 0)
        self.assertEqual(stats["exceed_packets"], 0)
        self.assertGreater(stats["violate_packets"], 0)

        # Stop policing on tun
        policer.apply_vpp_config(p.tun_if.sw_if_index, Dir.RX, False)
        self.verify_tun_44(p, count=n_pkts)

        # No new policer stats
        statsnew = policer.get_stats()
        self.assertEqual(stats, statsnew)

        # teardown
        policer.remove_vpp_config()
        self.unconfig_protect(p)
        self.unconfig_sa(p)
        self.unconfig_network(p)


@tag_fixme_vpp_workers
class TestIpsecItf4Tfc(TemplateIpsec4TunTfc, TestIpsecItf4):
    """IPsec Interface IPv4 with TFC"""


class TestIpsecItf4MPLS(TemplateIpsec, TemplateIpsecItf4, IpsecTun4):
    """IPsec Interface MPLSoIPv4"""

    tun4_encrypt_node_name = "esp-mpls-encrypt-tun"

    def setUp(self):
        super(TestIpsecItf4MPLS, self).setUp()

        self.tun_if = self.pg0

    def tearDown(self):
        super(TestIpsecItf4MPLS, self).tearDown()

    def gen_encrypt_pkts(self, p, sa, sw_intf, src, dst, count=1, payload_size=100):
        return [
            Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac)
            / sa.encrypt(
                MPLS(label=44, ttl=3)
                / IP(src=src, dst=dst)
                / UDP(sport=1166, dport=2233)
                / Raw(b"X" * payload_size)
            )
            for i in range(count)
        ]

    def verify_encrypted(self, p, sa, rxs):
        for rx in rxs:
            try:
                pkt = sa.decrypt(rx[IP])
                if not pkt.haslayer(IP):
                    pkt = IP(pkt[Raw].load)
                self.assert_packet_checksums_valid(pkt)
                self.assert_equal(pkt[MPLS].label, 44)
                self.assert_equal(pkt[IP].dst, p.remote_tun_if_host)
            except (IndexError, AssertionError):
                self.logger.debug(ppp("Unexpected packet:", rx))
                try:
                    self.logger.debug(ppp("Decrypted packet:", pkt))
                except:
                    pass
                raise

    def test_tun_mpls_o_ip4(self):
        """IPSEC interface MPLS over IPv4"""

        n_pkts = 127
        p = self.ipv4_params
        f = FibPathProto

        tbl = VppMplsTable(self, 0)
        tbl.add_vpp_config()

        self.config_network(p)
        # deag MPLS routes from the tunnel
        r4 = VppMplsRoute(
            self, 44, 1, [VppRoutePath(self.pg1.remote_ip4, self.pg1.sw_if_index)]
        ).add_vpp_config()
        p.route.modify(
            [
                VppRoutePath(
                    p.tun_if.remote_ip4, p.tun_if.sw_if_index, labels=[VppMplsLabel(44)]
                )
            ]
        )
        p.tun_if.enable_mpls()

        self.config_sa_tun(p, self.pg0.local_ip4, self.pg0.remote_ip4)
        self.config_protect(p)

        self.verify_tun_44(p, count=n_pkts)

        # cleanup
        p.tun_if.disable_mpls()
        self.unconfig_protect(p)
        self.unconfig_sa(p)
        self.unconfig_network(p)


class TemplateIpsecItf6(object):
    """IPsec Interface IPv6"""

    encryption_type = ESP
    tun6_encrypt_node_name = "esp6-encrypt-tun"
    tun6_decrypt_node_name = ["esp6-decrypt-tun", "esp6-decrypt-tun-post"]
    tun6_input_node = "ipsec6-tun-input"

    def config_sa_tun(self, p, src, dst):
        config_tun_params(p, self.encryption_type, None, src, dst)

        if not hasattr(p, "tun_flags"):
            p.tun_flags = None
        if not hasattr(p, "hop_limit"):
            p.hop_limit = 255

        p.tun_sa_out = VppIpsecSA(
            self,
            p.vpp_tun_sa_id,
            p.vpp_tun_spi,
            p.auth_algo_vpp_id,
            p.auth_key,
            p.crypt_algo_vpp_id,
            p.crypt_key,
            self.vpp_esp_protocol,
            src,
            dst,
            flags=p.flags,
            tun_flags=p.tun_flags,
            hop_limit=p.hop_limit,
        )
        p.tun_sa_out.add_vpp_config()

        p.tun_sa_in = VppIpsecSA(
            self,
            p.scapy_tun_sa_id,
            p.scapy_tun_spi,
            p.auth_algo_vpp_id,
            p.auth_key,
            p.crypt_algo_vpp_id,
            p.crypt_key,
            self.vpp_esp_protocol,
            dst,
            src,
            flags=p.flags,
        )
        p.tun_sa_in.add_vpp_config()

    def config_protect(self, p):
        p.tun_protect = VppIpsecTunProtect(self, p.tun_if, p.tun_sa_out, [p.tun_sa_in])
        p.tun_protect.add_vpp_config()

    def config_network(self, p):
        p.tun_if = VppIpsecInterface(self)

        p.tun_if.add_vpp_config()
        p.tun_if.admin_up()
        p.tun_if.config_ip4()
        p.tun_if.config_ip6()

        r = VppIpRoute(
            self,
            p.remote_tun_if_host4,
            32,
            [VppRoutePath(p.tun_if.remote_ip4, 0xFFFFFFFF)],
        )
        r.add_vpp_config()

        p.route = VppIpRoute(
            self,
            p.remote_tun_if_host,
            128,
            [
                VppRoutePath(
                    p.tun_if.remote_ip6, 0xFFFFFFFF, proto=DpoProto.DPO_PROTO_IP6
                )
            ],
        )
        p.route.add_vpp_config()

    def unconfig_network(self, p):
        p.route.remove_vpp_config()
        p.tun_if.remove_vpp_config()

    def unconfig_protect(self, p):
        p.tun_protect.remove_vpp_config()

    def unconfig_sa(self, p):
        p.tun_sa_out.remove_vpp_config()
        p.tun_sa_in.remove_vpp_config()


@tag_fixme_vpp_workers
class TestIpsecItf6(TemplateIpsec, TemplateIpsecItf6, IpsecTun6):
    """IPsec Interface IPv6"""

    def setUp(self):
        super(TestIpsecItf6, self).setUp()

        self.tun_if = self.pg0

    def tearDown(self):
        super(TestIpsecItf6, self).tearDown()

    def test_tun_66(self):
        """IPSEC interface IPv6"""

        tf = VppEnum.vl_api_tunnel_encap_decap_flags_t
        n_pkts = 127
        p = self.ipv6_params
        p.inner_hop_limit = 24
        p.outer_hop_limit = 23
        p.outer_flow_label = 243224
        p.tun_flags = tf.TUNNEL_API_ENCAP_DECAP_FLAG_ENCAP_COPY_HOP_LIMIT

        self.config_network(p)
        config_tun_params(
            p, self.encryption_type, None, self.pg0.local_ip6, self.pg0.remote_ip6
        )
        self.verify_drop_tun_66(p, count=n_pkts)
        self.config_sa_tun(p, self.pg0.local_ip6, self.pg0.remote_ip6)
        self.config_protect(p)

        self.verify_tun_66(p, count=n_pkts)
        self.assertEqual(p.tun_if.get_rx_stats(), n_pkts)
        self.assertEqual(p.tun_if.get_tx_stats(), n_pkts)

        p.tun_if.admin_down()
        self.verify_drop_tun_66(p, count=n_pkts)
        p.tun_if.admin_up()
        self.verify_tun_66(p, count=n_pkts)

        self.assertEqual(p.tun_if.get_rx_stats(), 3 * n_pkts)
        self.assertEqual(p.tun_if.get_tx_stats(), 2 * n_pkts)

        # it's a v4 packet when its encrypted
        self.tun6_encrypt_node_name = "esp4-encrypt-tun"

        self.verify_tun_46(p, count=n_pkts)
        self.assertEqual(p.tun_if.get_rx_stats(), 4 * n_pkts)
        self.assertEqual(p.tun_if.get_tx_stats(), 3 * n_pkts)

        self.tun6_encrypt_node_name = "esp6-encrypt-tun"

        self.vapi.cli("clear interfaces")

        # rekey - create new SAs and update the tunnel protection
        np = copy.copy(p)
        np.crypt_key = b"X" + p.crypt_key[1:]
        np.scapy_tun_spi += 100
        np.scapy_tun_sa_id += 1
        np.vpp_tun_spi += 100
        np.vpp_tun_sa_id += 1
        np.tun_if.local_spi = p.vpp_tun_spi
        np.tun_if.remote_spi = p.scapy_tun_spi
        np.inner_hop_limit = 24
        np.outer_hop_limit = 128
        np.inner_flow_label = 0xABCDE
        np.outer_flow_label = 0xABCDE
        np.hop_limit = 128
        np.tun_flags = tf.TUNNEL_API_ENCAP_DECAP_FLAG_ENCAP_COPY_FLOW_LABEL

        self.config_sa_tun(np, self.pg0.local_ip6, self.pg0.remote_ip6)
        self.config_protect(np)
        self.unconfig_sa(p)

        self.verify_tun_66(np, count=n_pkts)
        self.assertEqual(p.tun_if.get_rx_stats(), n_pkts)
        self.assertEqual(p.tun_if.get_tx_stats(), n_pkts)

        # teardown
        self.unconfig_protect(np)
        self.unconfig_sa(np)
        self.unconfig_network(p)

    def test_tun_66_police(self):
        """IPSEC interface IPv6 with input policer"""
        tf = VppEnum.vl_api_tunnel_encap_decap_flags_t
        n_pkts = 127
        p = self.ipv6_params
        p.inner_hop_limit = 24
        p.outer_hop_limit = 23
        p.outer_flow_label = 243224
        p.tun_flags = tf.TUNNEL_API_ENCAP_DECAP_FLAG_ENCAP_COPY_HOP_LIMIT

        self.config_network(p)
        self.config_sa_tun(p, self.pg0.local_ip6, self.pg0.remote_ip6)
        self.config_protect(p)

        action_tx = PolicerAction(
            VppEnum.vl_api_sse2_qos_action_type_t.SSE2_QOS_ACTION_API_TRANSMIT, 0
        )
        policer = VppPolicer(
            self,
            "pol1",
            80,
            0,
            1000,
            0,
            conform_action=action_tx,
            exceed_action=action_tx,
            violate_action=action_tx,
        )
        policer.add_vpp_config()

        # Start policing on tun
        policer.apply_vpp_config(p.tun_if.sw_if_index, Dir.RX, True)

        self.verify_tun_66(p, count=n_pkts)
        self.assertEqual(p.tun_if.get_rx_stats(), n_pkts)
        self.assertEqual(p.tun_if.get_tx_stats(), n_pkts)

        stats = policer.get_stats()

        # Single rate, 2 colour policer - expect conform, violate but no exceed
        self.assertGreater(stats["conform_packets"], 0)
        self.assertEqual(stats["exceed_packets"], 0)
        self.assertGreater(stats["violate_packets"], 0)

        # Stop policing on tun
        policer.apply_vpp_config(p.tun_if.sw_if_index, Dir.RX, False)
        self.verify_tun_66(p, count=n_pkts)

        # No new policer stats
        statsnew = policer.get_stats()
        self.assertEqual(stats, statsnew)

        # teardown
        policer.remove_vpp_config()
        self.unconfig_protect(p)
        self.unconfig_sa(p)
        self.unconfig_network(p)


@tag_fixme_vpp_workers
class TestIpsecItf6Tfc(TemplateIpsec6TunTfc, TestIpsecItf6):
    """IPsec Interface IPv6 with TFC"""


class TestIpsecMIfEsp4(TemplateIpsec, IpsecTun4):
    """Ipsec P2MP ESP v4 tests"""

    tun4_encrypt_node_name = "esp4-encrypt-tun"
    tun4_decrypt_node_name = ["esp4-decrypt-tun", "esp4-decrypt-tun-post"]
    encryption_type = ESP

    def gen_encrypt_pkts(self, p, sa, sw_intf, src, dst, count=1, payload_size=100):
        return [
            Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac)
            / sa.encrypt(
                IP(src=self.pg1.local_ip4, dst=self.pg1.remote_ip4)
                / UDP(sport=1144, dport=2233)
                / Raw(b"X" * payload_size)
            )
            for i in range(count)
        ]

    def gen_pkts(self, sw_intf, src, dst, count=1, payload_size=100):
        return [
            Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac)
            / IP(src="1.1.1.1", dst=dst)
            / UDP(sport=1144, dport=2233)
            / Raw(b"X" * payload_size)
            for i in range(count)
        ]

    def verify_decrypted(self, p, rxs):
        for rx in rxs:
            self.assert_equal(rx[Ether].dst, self.pg1.remote_mac)
            self.assert_equal(rx[IP].dst, self.pg1.remote_ip4)

    def verify_encrypted(self, p, sa, rxs):
        for rx in rxs:
            try:
                self.assertEqual(
                    rx[IP].tos, VppEnum.vl_api_ip_dscp_t.IP_API_DSCP_EF << 2
                )
                self.assertEqual(rx[IP].ttl, p.hop_limit)
                pkt = sa.decrypt(rx[IP])
                if not pkt.haslayer(IP):
                    pkt = IP(pkt[Raw].load)
                self.assert_packet_checksums_valid(pkt)
                e = pkt[IP]
                self.assertEqual(e[IP].dst, p.remote_tun_if_host)
            except (IndexError, AssertionError):
                self.logger.debug(ppp("Unexpected packet:", rx))
                try:
                    self.logger.debug(ppp("Decrypted packet:", pkt))
                except:
                    pass
                raise

    def setUp(self):
        super(TestIpsecMIfEsp4, self).setUp()

        N_NHS = 16
        self.tun_if = self.pg0
        p = self.ipv4_params
        p.tun_if = VppIpsecInterface(
            self, mode=(VppEnum.vl_api_tunnel_mode_t.TUNNEL_API_MODE_MP)
        )
        p.tun_if.add_vpp_config()
        p.tun_if.admin_up()
        p.tun_if.config_ip4()
        p.tun_if.unconfig_ip4()
        p.tun_if.config_ip4()
        p.tun_if.generate_remote_hosts(N_NHS)
        self.pg0.generate_remote_hosts(N_NHS)
        self.pg0.configure_ipv4_neighbors()

        r_all = AclRule(True, src_prefix="0.0.0.0/0", dst_prefix="0.0.0.0/0", proto=0)
        a = VppAcl(self, [r_all]).add_vpp_config()

        VppAclInterface(self, self.pg0.sw_if_index, [a]).add_vpp_config()
        VppAclInterface(self, p.tun_if.sw_if_index, [a]).add_vpp_config()

        # setup some SAs for several next-hops on the interface
        self.multi_params = []

        for ii in range(N_NHS):
            p = copy.copy(self.ipv4_params)

            p.remote_tun_if_host = "1.1.1.%d" % (ii + 1)
            p.scapy_tun_sa_id = p.scapy_tun_sa_id + ii
            p.scapy_tun_spi = p.scapy_tun_spi + ii
            p.vpp_tun_sa_id = p.vpp_tun_sa_id + ii
            p.vpp_tun_spi = p.vpp_tun_spi + ii

            p.scapy_tra_sa_id = p.scapy_tra_sa_id + ii
            p.scapy_tra_spi = p.scapy_tra_spi + ii
            p.vpp_tra_sa_id = p.vpp_tra_sa_id + ii
            p.vpp_tra_spi = p.vpp_tra_spi + ii
            p.hop_limit = ii + 10
            p.tun_sa_out = VppIpsecSA(
                self,
                p.vpp_tun_sa_id,
                p.vpp_tun_spi,
                p.auth_algo_vpp_id,
                p.auth_key,
                p.crypt_algo_vpp_id,
                p.crypt_key,
                self.vpp_esp_protocol,
                self.pg0.local_ip4,
                self.pg0.remote_hosts[ii].ip4,
                dscp=VppEnum.vl_api_ip_dscp_t.IP_API_DSCP_EF,
                hop_limit=p.hop_limit,
            )
            p.tun_sa_out.add_vpp_config()

            p.tun_sa_in = VppIpsecSA(
                self,
                p.scapy_tun_sa_id,
                p.scapy_tun_spi,
                p.auth_algo_vpp_id,
                p.auth_key,
                p.crypt_algo_vpp_id,
                p.crypt_key,
                self.vpp_esp_protocol,
                self.pg0.remote_hosts[ii].ip4,
                self.pg0.local_ip4,
                dscp=VppEnum.vl_api_ip_dscp_t.IP_API_DSCP_EF,
                hop_limit=p.hop_limit,
            )
            p.tun_sa_in.add_vpp_config()

            p.tun_protect = VppIpsecTunProtect(
                self,
                p.tun_if,
                p.tun_sa_out,
                [p.tun_sa_in],
                nh=p.tun_if.remote_hosts[ii].ip4,
            )
            p.tun_protect.add_vpp_config()
            config_tun_params(
                p,
                self.encryption_type,
                None,
                self.pg0.local_ip4,
                self.pg0.remote_hosts[ii].ip4,
            )
            self.multi_params.append(p)

            p.via_tun_route = VppIpRoute(
                self,
                p.remote_tun_if_host,
                32,
                [VppRoutePath(p.tun_if.remote_hosts[ii].ip4, p.tun_if.sw_if_index)],
            ).add_vpp_config()

            p.tun_dst = self.pg0.remote_hosts[ii].ip4

    def tearDown(self):
        p = self.ipv4_params
        p.tun_if.unconfig_ip4()
        super(TestIpsecMIfEsp4, self).tearDown()

    def test_tun_44(self):
        """P2MP IPSEC 44"""
        N_PKTS = 63
        for p in self.multi_params:
            self.verify_tun_44(p, count=N_PKTS)

        # remove one tunnel protect, the rest should still work
        self.multi_params[0].tun_protect.remove_vpp_config()
        self.verify_tun_dropped_44(self.multi_params[0], count=N_PKTS)
        self.multi_params[0].via_tun_route.remove_vpp_config()
        self.verify_tun_dropped_44(self.multi_params[0], count=N_PKTS)

        for p in self.multi_params[1:]:
            self.verify_tun_44(p, count=N_PKTS)

        self.multi_params[0].tun_protect.add_vpp_config()
        self.multi_params[0].via_tun_route.add_vpp_config()

        for p in self.multi_params:
            self.verify_tun_44(p, count=N_PKTS)


class TestIpsecItf6MPLS(TemplateIpsec, TemplateIpsecItf6, IpsecTun6):
    """IPsec Interface MPLSoIPv6"""

    tun6_encrypt_node_name = "esp-mpls-encrypt-tun"

    def setUp(self):
        super(TestIpsecItf6MPLS, self).setUp()

        self.tun_if = self.pg0

    def tearDown(self):
        super(TestIpsecItf6MPLS, self).tearDown()

    def gen_encrypt_pkts6(self, p, sa, sw_intf, src, dst, count=1, payload_size=100):
        return [
            Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac)
            / sa.encrypt(
                MPLS(label=66, ttl=3)
                / IPv6(src=src, dst=dst)
                / UDP(sport=1166, dport=2233)
                / Raw(b"X" * payload_size)
            )
            for i in range(count)
        ]

    def verify_encrypted6(self, p, sa, rxs):
        for rx in rxs:
            try:
                pkt = sa.decrypt(rx[IPv6])
                if not pkt.haslayer(IPv6):
                    pkt = IP(pkt[Raw].load)
                self.assert_packet_checksums_valid(pkt)
                self.assert_equal(pkt[MPLS].label, 66)
                self.assert_equal(pkt[IPv6].dst, p.remote_tun_if_host)
            except (IndexError, AssertionError):
                self.logger.debug(ppp("Unexpected packet:", rx))
                try:
                    self.logger.debug(ppp("Decrypted packet:", pkt))
                except:
                    pass
                raise

    def test_tun_mpls_o_ip6(self):
        """IPSEC interface MPLS over IPv6"""

        n_pkts = 127
        p = self.ipv6_params
        f = FibPathProto

        tbl = VppMplsTable(self, 0)
        tbl.add_vpp_config()

        self.config_network(p)
        # deag MPLS routes from the tunnel
        r6 = VppMplsRoute(
            self,
            66,
            1,
            [VppRoutePath(self.pg1.remote_ip6, self.pg1.sw_if_index)],
            eos_proto=f.FIB_PATH_NH_PROTO_IP6,
        ).add_vpp_config()
        p.route.modify(
            [
                VppRoutePath(
                    p.tun_if.remote_ip6, p.tun_if.sw_if_index, labels=[VppMplsLabel(66)]
                )
            ]
        )
        p.tun_if.enable_mpls()

        self.config_sa_tun(p, self.pg0.local_ip6, self.pg0.remote_ip6)
        self.config_protect(p)

        self.verify_tun_66(p, count=n_pkts)

        # cleanup
        p.tun_if.disable_mpls()
        self.unconfig_protect(p)
        self.unconfig_sa(p)
        self.unconfig_network(p)


if __name__ == "__main__":
    unittest.main(testRunner=VppTestRunner)