ligang/vpp
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
vpp/test/test_ipsec_tun_if_esp.py
Dave Wallace cf9356d642 tests: update scapy to version 2.4.5 
- Required for Ubuntu 24.04 LTS jobs
- temporarily disable TestIpsecEsp1 and
  TestIpsecAhAll tests until a patch can
  be added to fix them

Type: test

Change-Id: I1ae7b170117182c3252629bbbb770775e2c496c9
Signed-off-by: Benoît Ganne <bganne@cisco.com>
Signed-off-by: Dave Wallace <dwallacelf@gmail.com>
2024-08-07 09:15:17 +00:00

3907 lines
122 KiB
Python
Raw Permalink Blame History

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, Padding
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
from config import config
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, count=127
)
self.send_and_assert_no_replies(self.pg1, tx)
self.unconfig_protect(p)
self.unconfig_sa(p)
self.unconfig_network(p)
def test_tun_44_async(self):
"""IPSec SA with NULL algos using async crypto"""
p = self.ipv4_params
self.vapi.ipsec_set_async_mode(async_enable=True)
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, count=127
)
self.send_and_assert_no_replies(self.pg1, tx)
self.unconfig_protect(p)
self.unconfig_sa(p)
self.unconfig_network(p)
self.vapi.ipsec_set_async_mode(async_enable=False)
@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)
@unittest.skipIf(
"gre" in config.excluded_plugins, "Exclude tests depending on GRE plugin"
)
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()
@unittest.skipIf(
"gre" in config.excluded_plugins, "Exclude tests depending on GRE plugin"
)
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()
@unittest.skipIf(
"gre" in config.excluded_plugins, "Exclude tests depending on GRE plugin"
)
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()
@unittest.skipIf(
"gre" in config.excluded_plugins, "Exclude tests depending on GRE plugin"
)
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()
@unittest.skipIf(
"gre" in config.excluded_plugins, "Exclude tests depending on GRE plugin"
)
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()
@unittest.skipIf(
"gre" in config.excluded_plugins, "Exclude tests depending on GRE plugin"
)
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)
@unittest.skipIf(
"gre" in config.excluded_plugins, "Exclude tests depending on GRE plugin"
)
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()
@unittest.skipIf(
"gre" in config.excluded_plugins, "Exclude tests depending on GRE plugin"
)
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)
@unittest.skipIf(
"gre" in config.excluded_plugins, "Exclude tests depending on GRE plugin"
)
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
@unittest.skipIf(
"acl" in config.excluded_plugins, "Exclude tests depending on ACL plugin"
)
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"""
@unittest.skipIf(
"acl" in config.excluded_plugins, "Exclude tests depending on ACL plugin"
)
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)
Reference in New Issue View Git Blame Copy Permalink