summaryrefslogtreecommitdiffstats
path: root/test
diff options
context:
space:
mode:
Diffstat (limited to 'test')
-rw-r--r--test/template_ipsec.py12
-rw-r--r--test/test_ipsec_ah.py96
-rw-r--r--test/test_ipsec_esp.py93
-rw-r--r--test/test_ipsec_tun_if_esp.py30
-rw-r--r--test/vpp_ipsec.py20
-rw-r--r--test/vpp_papi_provider.py4
6 files changed, 230 insertions, 25 deletions
diff --git a/test/template_ipsec.py b/test/template_ipsec.py
index 532a7a0a189..7bd0b9ec81f 100644
--- a/test/template_ipsec.py
+++ b/test/template_ipsec.py
@@ -49,6 +49,9 @@ class IPsecIPv4Params(object):
self.salt = 0
self.flags = 0
self.nat_header = None
+ self.tun_flags = (VppEnum.vl_api_tunnel_encap_decap_flags_t.
+ TUNNEL_API_ENCAP_DECAP_FLAG_NONE)
+ self.dscp = 0
class IPsecIPv6Params(object):
@@ -85,6 +88,9 @@ class IPsecIPv6Params(object):
self.salt = 0
self.flags = 0
self.nat_header = None
+ self.tun_flags = (VppEnum.vl_api_tunnel_encap_decap_flags_t.
+ TUNNEL_API_ENCAP_DECAP_FLAG_NONE)
+ self.dscp = 0
def mk_scapy_crypt_key(p):
@@ -181,8 +187,10 @@ class TemplateIpsec(VppTestCase):
super(TemplateIpsec, cls).tearDownClass()
def setup_params(self):
- self.ipv4_params = IPsecIPv4Params()
- self.ipv6_params = IPsecIPv6Params()
+ if not hasattr(self, 'ipv4_params'):
+ self.ipv4_params = IPsecIPv4Params()
+ if not hasattr(self, 'ipv6_params'):
+ self.ipv6_params = IPsecIPv6Params()
self.params = {self.ipv4_params.addr_type: self.ipv4_params,
self.ipv6_params.addr_type: self.ipv6_params}
diff --git a/test/test_ipsec_ah.py b/test/test_ipsec_ah.py
index bc539a2bd03..d44492ddd26 100644
--- a/test/test_ipsec_ah.py
+++ b/test/test_ipsec_ah.py
@@ -2,6 +2,10 @@ import socket
import unittest
from scapy.layers.ipsec import AH
+from scapy.layers.inet import IP, UDP
+from scapy.layers.inet6 import IPv6
+from scapy.layers.l2 import Ether
+from scapy.packet import Raw
from framework import VppTestRunner
from template_ipsec import TemplateIpsec, IpsecTra46Tests, IpsecTun46Tests, \
@@ -119,6 +123,7 @@ class ConfigIpsecAH(TemplateIpsec):
addr_any = params.addr_any
addr_bcast = params.addr_bcast
flags = params.flags
+ tun_flags = params.tun_flags
e = VppEnum.vl_api_ipsec_spd_action_t
objs = []
@@ -128,7 +133,9 @@ class ConfigIpsecAH(TemplateIpsec):
self.vpp_ah_protocol,
self.tun_if.local_addr[addr_type],
self.tun_if.remote_addr[addr_type],
- flags=flags)
+ tun_flags=tun_flags,
+ flags=flags,
+ dscp=params.dscp)
params.tun_sa_out = VppIpsecSA(self, vpp_tun_sa_id, vpp_tun_spi,
auth_algo_vpp_id, auth_key,
@@ -136,7 +143,9 @@ class ConfigIpsecAH(TemplateIpsec):
self.vpp_ah_protocol,
self.tun_if.remote_addr[addr_type],
self.tun_if.local_addr[addr_type],
- flags=flags)
+ tun_flags=tun_flags,
+ flags=flags,
+ dscp=params.dscp)
objs.append(params.tun_sa_in)
objs.append(params.tun_sa_out)
@@ -302,6 +311,89 @@ class TestIpsecAh2(TemplateIpsecAh, IpsecTra46Tests, IpsecTun46Tests):
pass
+class TestIpsecAhTun(TemplateIpsecAh, IpsecTun46Tests):
+ """ Ipsec AH - TUN encap tests """
+
+ def setUp(self):
+ self.ipv4_params = IPsecIPv4Params()
+ self.ipv6_params = IPsecIPv6Params()
+
+ c = (VppEnum.vl_api_tunnel_encap_decap_flags_t.
+ TUNNEL_API_ENCAP_DECAP_FLAG_ENCAP_COPY_DSCP)
+ c1 = c | (VppEnum.vl_api_tunnel_encap_decap_flags_t.
+ TUNNEL_API_ENCAP_DECAP_FLAG_ENCAP_COPY_ECN)
+
+ self.ipv4_params.tun_flags = c
+ self.ipv6_params.tun_flags = c1
+
+ super(TestIpsecAhTun, self).setUp()
+
+ def gen_pkts(self, sw_intf, src, dst, count=1, payload_size=54):
+ # set the DSCP + ECN - flags are set to copy only DSCP
+ return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
+ IP(src=src, dst=dst, tos=5) /
+ UDP(sport=4444, dport=4444) /
+ Raw(b'X' * payload_size)
+ for i in range(count)]
+
+ def gen_pkts6(self, sw_intf, src, dst, count=1, payload_size=54):
+ # set the DSCP + ECN - flags are set to copy both
+ return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
+ IPv6(src=src, dst=dst, tc=5) /
+ UDP(sport=4444, dport=4444) /
+ Raw(b'X' * payload_size)
+ for i in range(count)]
+
+ def verify_encrypted(self, p, sa, rxs):
+ # just check that only the DSCP is copied
+ for rx in rxs:
+ self.assertEqual(rx[IP].tos, 4)
+
+ def verify_encrypted6(self, p, sa, rxs):
+ # just check that the DSCP & ECN are copied
+ for rx in rxs:
+ self.assertEqual(rx[IPv6].tc, 5)
+
+
+class TestIpsecAhTun2(TemplateIpsecAh, IpsecTun46Tests):
+ """ Ipsec AH - TUN encap tests """
+
+ def setUp(self):
+ self.ipv4_params = IPsecIPv4Params()
+ self.ipv6_params = IPsecIPv6Params()
+
+ self.ipv4_params.dscp = 3
+ self.ipv6_params.dscp = 4
+
+ super(TestIpsecAhTun2, self).setUp()
+
+ def gen_pkts(self, sw_intf, src, dst, count=1, payload_size=54):
+ # set the DSCP + ECN - flags are set to copy only DSCP
+ return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
+ IP(src=src, dst=dst, tos=0) /
+ UDP(sport=4444, dport=4444) /
+ Raw(b'X' * payload_size)
+ for i in range(count)]
+
+ def gen_pkts6(self, sw_intf, src, dst, count=1, payload_size=54):
+ # set the DSCP + ECN - flags are set to copy both
+ return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
+ IPv6(src=src, dst=dst, tc=0) /
+ UDP(sport=4444, dport=4444) /
+ Raw(b'X' * payload_size)
+ for i in range(count)]
+
+ def verify_encrypted(self, p, sa, rxs):
+ # just check that only the DSCP is copied
+ for rx in rxs:
+ self.assertEqual(rx[IP].tos, 0xc)
+
+ def verify_encrypted6(self, p, sa, rxs):
+ # just check that the DSCP & ECN are copied
+ for rx in rxs:
+ self.assertEqual(rx[IPv6].tc, 0x10)
+
+
class TestIpsecAhHandoff(TemplateIpsecAh,
IpsecTun6HandoffTests,
IpsecTun4HandoffTests):
diff --git a/test/test_ipsec_esp.py b/test/test_ipsec_esp.py
index dbd21f1f5fe..78da401ea8a 100644
--- a/test/test_ipsec_esp.py
+++ b/test/test_ipsec_esp.py
@@ -2,6 +2,9 @@ import socket
import unittest
from scapy.layers.ipsec import ESP
from scapy.layers.inet import IP, ICMP, UDP
+from scapy.layers.inet6 import IPv6
+from scapy.layers.l2 import Ether
+from scapy.packet import Raw
from parameterized import parameterized
from framework import VppTestRunner
@@ -107,6 +110,7 @@ class ConfigIpsecESP(TemplateIpsec):
addr_bcast = params.addr_bcast
e = VppEnum.vl_api_ipsec_spd_action_t
flags = params.flags
+ tun_flags = params.tun_flags
salt = params.salt
objs = []
@@ -116,6 +120,8 @@ class ConfigIpsecESP(TemplateIpsec):
self.vpp_esp_protocol,
self.tun_if.local_addr[addr_type],
self.tun_if.remote_addr[addr_type],
+ tun_flags=tun_flags,
+ dscp=params.dscp,
flags=flags,
salt=salt)
params.tun_sa_out = VppIpsecSA(self, vpp_tun_sa_id, vpp_tun_spi,
@@ -124,6 +130,8 @@ class ConfigIpsecESP(TemplateIpsec):
self.vpp_esp_protocol,
self.tun_if.remote_addr[addr_type],
self.tun_if.local_addr[addr_type],
+ tun_flags=tun_flags,
+ dscp=params.dscp,
flags=flags,
salt=salt)
objs.append(params.tun_sa_in)
@@ -368,6 +376,91 @@ class TestIpsecEsp1(TemplateIpsecEsp, IpsecTra46Tests,
self.tun4_encrypt_node_name = old_name
+class TestIpsecEspTun(TemplateIpsecEsp, IpsecTun46Tests):
+ """ Ipsec ESP - TUN encap tests """
+
+ def setUp(self):
+ self.ipv4_params = IPsecIPv4Params()
+ self.ipv6_params = IPsecIPv6Params()
+
+ c = (VppEnum.vl_api_tunnel_encap_decap_flags_t.
+ TUNNEL_API_ENCAP_DECAP_FLAG_ENCAP_COPY_DSCP)
+ c1 = c | (VppEnum.vl_api_tunnel_encap_decap_flags_t.
+ TUNNEL_API_ENCAP_DECAP_FLAG_ENCAP_COPY_ECN)
+
+ self.ipv4_params.tun_flags = c
+ self.ipv6_params.tun_flags = c1
+
+ super(TestIpsecEspTun, self).setUp()
+
+ def gen_pkts(self, sw_intf, src, dst, count=1, payload_size=54):
+ # set the DSCP + ECN - flags are set to copy only DSCP
+ return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
+ IP(src=src, dst=dst, tos=5) /
+ UDP(sport=4444, dport=4444) /
+ Raw(b'X' * payload_size)
+ for i in range(count)]
+
+ def gen_pkts6(self, sw_intf, src, dst, count=1, payload_size=54):
+ # set the DSCP + ECN - flags are set to copy both
+ return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
+ IPv6(src=src, dst=dst, tc=5) /
+ UDP(sport=4444, dport=4444) /
+ Raw(b'X' * payload_size)
+ for i in range(count)]
+
+ def verify_encrypted(self, p, sa, rxs):
+ # just check that only the DSCP is copied
+ for rx in rxs:
+ self.assertEqual(rx[IP].tos, 4)
+
+ def verify_encrypted6(self, p, sa, rxs):
+ # just check that the DSCP & ECN are copied
+ for rx in rxs:
+ self.assertEqual(rx[IPv6].tc, 5)
+
+
+class TestIpsecEspTun2(TemplateIpsecEsp, IpsecTun46Tests):
+ """ Ipsec ESP - TUN DSCP tests """
+
+ def setUp(self):
+ self.ipv4_params = IPsecIPv4Params()
+ self.ipv6_params = IPsecIPv6Params()
+
+ self.ipv4_params.dscp = VppEnum.vl_api_ip_dscp_t.IP_API_DSCP_EF
+ self.ipv6_params.dscp = VppEnum.vl_api_ip_dscp_t.IP_API_DSCP_AF11
+
+ super(TestIpsecEspTun2, self).setUp()
+
+ def gen_pkts(self, sw_intf, src, dst, count=1, payload_size=54):
+ # set the DSCP + ECN - flags are set to copy only DSCP
+ return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
+ IP(src=src, dst=dst) /
+ UDP(sport=4444, dport=4444) /
+ Raw(b'X' * payload_size)
+ for i in range(count)]
+
+ def gen_pkts6(self, sw_intf, src, dst, count=1, payload_size=54):
+ # set the DSCP + ECN - flags are set to copy both
+ return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
+ IPv6(src=src, dst=dst) /
+ UDP(sport=4444, dport=4444) /
+ Raw(b'X' * payload_size)
+ for i in range(count)]
+
+ def verify_encrypted(self, p, sa, rxs):
+ # just check that only the DSCP is copied
+ for rx in rxs:
+ self.assertEqual(rx[IP].tos,
+ VppEnum.vl_api_ip_dscp_t.IP_API_DSCP_EF << 2)
+
+ def verify_encrypted6(self, p, sa, rxs):
+ # just check that the DSCP & ECN are copied
+ for rx in rxs:
+ self.assertEqual(rx[IPv6].tc,
+ VppEnum.vl_api_ip_dscp_t.IP_API_DSCP_AF11 << 2)
+
+
class TestIpsecEsp2(TemplateIpsecEsp, IpsecTcpTests):
""" Ipsec ESP - TCP tests """
pass
diff --git a/test/test_ipsec_tun_if_esp.py b/test/test_ipsec_tun_if_esp.py
index fb31f221aed..ba421694d60 100644
--- a/test/test_ipsec_tun_if_esp.py
+++ b/test/test_ipsec_tun_if_esp.py
@@ -2841,6 +2841,8 @@ class TestIpsecMIfEsp4(TemplateIpsec, IpsecTun4):
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)
pkt = sa.decrypt(rx[IP])
if not pkt.haslayer(IP):
pkt = IP(pkt[Raw].load)
@@ -2887,20 +2889,24 @@ class TestIpsecMIfEsp4(TemplateIpsec, IpsecTun4):
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.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.local_ip4,
- self.pg0.remote_hosts[ii].ip4)
+ p.tun_sa_out = 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.local_ip4,
+ self.pg0.remote_hosts[ii].ip4,
+ dscp=VppEnum.vl_api_ip_dscp_t.IP_API_DSCP_EF)
p.tun_sa_out.add_vpp_config()
- p.tun_sa_in = 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.remote_hosts[ii].ip4,
- self.pg0.local_ip4)
+ p.tun_sa_in = 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.remote_hosts[ii].ip4,
+ self.pg0.local_ip4,
+ dscp=VppEnum.vl_api_ip_dscp_t.IP_API_DSCP_EF)
p.tun_sa_in.add_vpp_config()
p.tun_protect = VppIpsecTunProtect(
diff --git a/test/vpp_ipsec.py b/test/vpp_ipsec.py
index 479870621e0..d7ed8564fc4 100644
--- a/test/vpp_ipsec.py
+++ b/test/vpp_ipsec.py
@@ -192,8 +192,9 @@ class VppIpsecSA(VppObject):
crypto_alg, crypto_key,
proto,
tun_src=None, tun_dst=None,
- flags=None, salt=0, udp_src=None,
- udp_dst=None):
+ flags=None, salt=0, tun_flags=None,
+ dscp=None,
+ udp_src=None, udp_dst=None):
e = VppEnum.vl_api_ipsec_sad_flags_t
self.test = test
self.id = id
@@ -220,6 +221,13 @@ class VppIpsecSA(VppObject):
self.tun_dst = ip_address(text_type(tun_dst))
self.udp_src = udp_src
self.udp_dst = udp_dst
+ self.tun_flags = (VppEnum.vl_api_tunnel_encap_decap_flags_t.
+ TUNNEL_API_ENCAP_DECAP_FLAG_NONE)
+ if tun_flags:
+ self.tun_flags = tun_flags
+ self.dscp = VppEnum.vl_api_ip_dscp_t.IP_API_DSCP_CS0
+ if dscp:
+ self.dscp = dscp
def add_vpp_config(self):
entry = {
@@ -238,6 +246,8 @@ class VppIpsecSA(VppObject):
'protocol': self.proto,
'tunnel_src': (self.tun_src if self.tun_src else []),
'tunnel_dst': (self.tun_dst if self.tun_dst else []),
+ 'tunnel_flags': self.tun_flags,
+ 'dscp': self.dscp,
'flags': self.flags,
'salt': self.salt
}
@@ -246,12 +256,12 @@ class VppIpsecSA(VppObject):
entry['udp_src_port'] = self.udp_src
if self.udp_dst:
entry['udp_dst_port'] = self.udp_dst
- r = self.test.vapi.ipsec_sad_entry_add_del(is_add=1, entry=entry)
+ r = self.test.vapi.ipsec_sad_entry_add_del_v2(is_add=1, entry=entry)
self.stat_index = r.stat_index
self.test.registry.register(self, self.test.logger)
def remove_vpp_config(self):
- r = self.test.vapi.ipsec_sad_entry_add_del(
+ r = self.test.vapi.ipsec_sad_entry_add_del_v2(
is_add=0,
entry={
'sad_id': self.id,
@@ -279,7 +289,7 @@ class VppIpsecSA(VppObject):
def query_vpp_config(self):
e = VppEnum.vl_api_ipsec_sad_flags_t
- bs = self.test.vapi.ipsec_sa_dump()
+ bs = self.test.vapi.ipsec_sa_v2_dump()
for b in bs:
if b.entry.sad_id == self.id:
# if udp encap is configured then the ports should match
diff --git a/test/vpp_papi_provider.py b/test/vpp_papi_provider.py
index ec2f222b540..6e5ebc25fb4 100644
--- a/test/vpp_papi_provider.py
+++ b/test/vpp_papi_provider.py
@@ -946,10 +946,6 @@ class VppPapiProvider(object):
{'spd_index': spd_index if spd_index else 0,
'spd_index_valid': 1 if spd_index else 0})
- def ipsec_sa_dump(self, sa_id=None):
- return self.api(self.papi.ipsec_sa_dump,
- {'sa_id': sa_id if sa_id else 0xffffffff})
-
def ipsec_spd_entry_add_del(self,
spd_id,
sa_id,