TEST: IPSEC NAT-T with UDP header
Change-Id: I5ef8b3f4be40a7a0b0f1cb90dc0e15a4711e8664
Signed-off-by: Neale Ranns <nranns@cisco.com>
diff --git a/src/vnet/ipsec/ipsec.api b/src/vnet/ipsec/ipsec.api
index 1928372..778bd69 100644
--- a/src/vnet/ipsec/ipsec.api
+++ b/src/vnet/ipsec/ipsec.api
@@ -235,6 +235,7 @@
/* come-on Ole please fix this */
IPSEC_API_SAD_COMBO_12 = 12,
+ IPSEC_API_SAD_COMBO_18 = 18,
IPSEC_API_SAD_COMBO_20 = 20,
};
diff --git a/test/template_ipsec.py b/test/template_ipsec.py
index 53b6cec..68f1183 100644
--- a/test/template_ipsec.py
+++ b/test/template_ipsec.py
@@ -1,7 +1,7 @@
import unittest
import socket
-from scapy.layers.inet import IP, ICMP, TCP
+from scapy.layers.inet import IP, ICMP, TCP, UDP
from scapy.layers.ipsec import SecurityAssociation
from scapy.layers.l2 import Ether, Raw
from scapy.layers.inet6 import IPv6, ICMPv6EchoRequest
@@ -41,6 +41,8 @@
IPSEC_API_CRYPTO_ALG_AES_CBC_128)
self.crypt_algo = 'AES-CBC' # scapy name
self.crypt_key = 'JPjyOWBeVEQiMe7h'
+ self.flags = 0
+ self.nat_header = None
class IPsecIPv6Params(object):
@@ -73,6 +75,8 @@
IPSEC_API_CRYPTO_ALG_AES_CBC_256)
self.crypt_algo = 'AES-CBC' # scapy name
self.crypt_key = 'JPjyOWBeVEQiMe7hJPjyOWBeVEQiMe7h'
+ self.flags = 0
+ self.nat_header = None
class TemplateIpsec(VppTestCase):
@@ -168,29 +172,35 @@
auth_algo=params.auth_algo, auth_key=params.auth_key,
tunnel_header=ip_class_by_addr_type[params.addr_type](
src=self.tun_if.remote_addr[params.addr_type],
- dst=self.tun_if.local_addr[params.addr_type]))
+ dst=self.tun_if.local_addr[params.addr_type]),
+ nat_t_header=params.nat_header)
vpp_tun_sa = SecurityAssociation(
self.encryption_type, spi=params.scapy_tun_spi,
crypt_algo=params.crypt_algo, crypt_key=params.crypt_key,
auth_algo=params.auth_algo, auth_key=params.auth_key,
tunnel_header=ip_class_by_addr_type[params.addr_type](
dst=self.tun_if.remote_addr[params.addr_type],
- src=self.tun_if.local_addr[params.addr_type]))
+ src=self.tun_if.local_addr[params.addr_type]),
+ nat_t_header=params.nat_header)
return vpp_tun_sa, scapy_tun_sa
def configure_sa_tra(self, params):
- params.scapy_tra_sa = SecurityAssociation(self.encryption_type,
- spi=params.vpp_tra_spi,
- crypt_algo=params.crypt_algo,
- crypt_key=params.crypt_key,
- auth_algo=params.auth_algo,
- auth_key=params.auth_key)
- params.vpp_tra_sa = SecurityAssociation(self.encryption_type,
- spi=params.scapy_tra_spi,
- crypt_algo=params.crypt_algo,
- crypt_key=params.crypt_key,
- auth_algo=params.auth_algo,
- auth_key=params.auth_key)
+ params.scapy_tra_sa = SecurityAssociation(
+ self.encryption_type,
+ spi=params.vpp_tra_spi,
+ crypt_algo=params.crypt_algo,
+ crypt_key=params.crypt_key,
+ auth_algo=params.auth_algo,
+ auth_key=params.auth_key,
+ nat_t_header=params.nat_header)
+ params.vpp_tra_sa = SecurityAssociation(
+ self.encryption_type,
+ spi=params.scapy_tra_spi,
+ crypt_algo=params.crypt_algo,
+ crypt_key=params.crypt_key,
+ auth_algo=params.auth_algo,
+ auth_key=params.auth_key,
+ nat_t_header=params.nat_header)
class IpsecTcpTests(object):
@@ -210,7 +220,7 @@
self.assert_packet_checksums_valid(decrypted)
-class IpsecTraTests(object):
+class IpsecTra4Tests(object):
def test_tra_anti_replay(self, count=1):
""" ipsec v4 transport anti-reply test """
p = self.params[socket.AF_INET]
@@ -320,6 +330,8 @@
""" ipsec v4 transport burst test """
self.test_tra_basic(count=257)
+
+class IpsecTra6Tests(object):
def test_tra_basic6(self, count=1):
""" ipsec v6 transport basic test """
self.vapi.cli("clear errors")
@@ -358,6 +370,10 @@
self.test_tra_basic6(count=257)
+class IpsecTra46Tests(IpsecTra4Tests, IpsecTra6Tests):
+ pass
+
+
class IpsecTun4Tests(object):
def test_tun_basic44(self, count=1):
""" ipsec 4o4 tunnel basic test """
@@ -477,7 +493,7 @@
self.test_tun_basic66(count=257)
-class IpsecTunTests(IpsecTun4Tests, IpsecTun6Tests):
+class IpsecTun46Tests(IpsecTun4Tests, IpsecTun6Tests):
pass
diff --git a/test/test_ipsec_ah.py b/test/test_ipsec_ah.py
index f99bb85..7498f51 100644
--- a/test/test_ipsec_ah.py
+++ b/test/test_ipsec_ah.py
@@ -4,7 +4,7 @@
from scapy.layers.ipsec import AH
from framework import VppTestRunner
-from template_ipsec import TemplateIpsec, IpsecTraTests, IpsecTunTests
+from template_ipsec import TemplateIpsec, IpsecTra46Tests, IpsecTun46Tests
from template_ipsec import IpsecTcpTests
from vpp_ipsec import VppIpsecSA, VppIpsecSpd, VppIpsecSpdEntry,\
VppIpsecSpdItfBinding
@@ -203,7 +203,7 @@
priority=10).add_vpp_config()
-class TestIpsecAh1(TemplateIpsecAh, IpsecTraTests, IpsecTunTests):
+class TestIpsecAh1(TemplateIpsecAh, IpsecTra46Tests, IpsecTun46Tests):
""" Ipsec AH - TUN & TRA tests """
tra4_encrypt_node_name = "ah4-encrypt"
tra4_decrypt_node_name = "ah4-decrypt"
diff --git a/test/test_ipsec_esp.py b/test/test_ipsec_esp.py
index 7a05f0d..09b7240 100644
--- a/test/test_ipsec_esp.py
+++ b/test/test_ipsec_esp.py
@@ -1,10 +1,11 @@
import socket
import unittest
from scapy.layers.ipsec import ESP
+from scapy.layers.inet import UDP
from framework import VppTestRunner
-from template_ipsec import IpsecTraTests, IpsecTunTests
-from template_ipsec import TemplateIpsec, IpsecTcpTests
+from template_ipsec import IpsecTra46Tests, IpsecTun46Tests, TemplateIpsec, \
+ IpsecTcpTests, IpsecTun4Tests, IpsecTra4Tests
from vpp_ipsec import VppIpsecSpd, VppIpsecSpdEntry, VppIpsecSA,\
VppIpsecSpdItfBinding
from vpp_ip_route import VppIpRoute, VppRoutePath
@@ -12,6 +13,140 @@
from vpp_papi import VppEnum
+def config_esp_tun(test, params):
+ addr_type = params.addr_type
+ scapy_tun_sa_id = params.scapy_tun_sa_id
+ scapy_tun_spi = params.scapy_tun_spi
+ vpp_tun_sa_id = params.vpp_tun_sa_id
+ vpp_tun_spi = params.vpp_tun_spi
+ auth_algo_vpp_id = params.auth_algo_vpp_id
+ auth_key = params.auth_key
+ crypt_algo_vpp_id = params.crypt_algo_vpp_id
+ crypt_key = params.crypt_key
+ remote_tun_if_host = params.remote_tun_if_host
+ addr_any = params.addr_any
+ addr_bcast = params.addr_bcast
+ e = VppEnum.vl_api_ipsec_spd_action_t
+
+ params.tun_sa_in = VppIpsecSA(test, scapy_tun_sa_id, scapy_tun_spi,
+ auth_algo_vpp_id, auth_key,
+ crypt_algo_vpp_id, crypt_key,
+ test.vpp_esp_protocol,
+ test.tun_if.local_addr[addr_type],
+ test.tun_if.remote_addr[addr_type])
+ params.tun_sa_in.add_vpp_config()
+ params.tun_sa_out = VppIpsecSA(test, vpp_tun_sa_id, vpp_tun_spi,
+ auth_algo_vpp_id, auth_key,
+ crypt_algo_vpp_id, crypt_key,
+ test.vpp_esp_protocol,
+ test.tun_if.remote_addr[addr_type],
+ test.tun_if.local_addr[addr_type])
+ params.tun_sa_out.add_vpp_config()
+
+ params.spd_policy_in_any = VppIpsecSpdEntry(test, test.tun_spd,
+ scapy_tun_sa_id,
+ addr_any, addr_bcast,
+ addr_any, addr_bcast,
+ socket.IPPROTO_ESP)
+ params.spd_policy_in_any.add_vpp_config()
+ params.spd_policy_out_any = VppIpsecSpdEntry(test, test.tun_spd,
+ scapy_tun_sa_id,
+ addr_any, addr_bcast,
+ addr_any, addr_bcast,
+ socket.IPPROTO_ESP,
+ is_outbound=0)
+ params.spd_policy_out_any.add_vpp_config()
+
+ VppIpsecSpdEntry(test, test.tun_spd, vpp_tun_sa_id,
+ remote_tun_if_host, remote_tun_if_host,
+ test.pg1.remote_addr[addr_type],
+ test.pg1.remote_addr[addr_type],
+ 0,
+ priority=10,
+ policy=e.IPSEC_API_SPD_ACTION_PROTECT,
+ is_outbound=0).add_vpp_config()
+ VppIpsecSpdEntry(test, test.tun_spd, scapy_tun_sa_id,
+ test.pg1.remote_addr[addr_type],
+ test.pg1.remote_addr[addr_type],
+ remote_tun_if_host, remote_tun_if_host,
+ 0,
+ policy=e.IPSEC_API_SPD_ACTION_PROTECT,
+ priority=10).add_vpp_config()
+
+ VppIpsecSpdEntry(test, test.tun_spd, vpp_tun_sa_id,
+ remote_tun_if_host, remote_tun_if_host,
+ test.pg0.local_addr[addr_type],
+ test.pg0.local_addr[addr_type],
+ 0,
+ priority=20,
+ policy=e.IPSEC_API_SPD_ACTION_PROTECT,
+ is_outbound=0).add_vpp_config()
+ VppIpsecSpdEntry(test, test.tun_spd, scapy_tun_sa_id,
+ test.pg0.local_addr[addr_type],
+ test.pg0.local_addr[addr_type],
+ remote_tun_if_host, remote_tun_if_host,
+ 0,
+ policy=e.IPSEC_API_SPD_ACTION_PROTECT,
+ priority=20).add_vpp_config()
+
+
+def config_esp_tra(test, params):
+ addr_type = params.addr_type
+ scapy_tra_sa_id = params.scapy_tra_sa_id
+ scapy_tra_spi = params.scapy_tra_spi
+ vpp_tra_sa_id = params.vpp_tra_sa_id
+ vpp_tra_spi = params.vpp_tra_spi
+ auth_algo_vpp_id = params.auth_algo_vpp_id
+ auth_key = params.auth_key
+ crypt_algo_vpp_id = params.crypt_algo_vpp_id
+ crypt_key = params.crypt_key
+ addr_any = params.addr_any
+ addr_bcast = params.addr_bcast
+ flags = (VppEnum.vl_api_ipsec_sad_flags_t.
+ IPSEC_API_SAD_FLAG_USE_ANTI_REPLAY)
+ e = VppEnum.vl_api_ipsec_spd_action_t
+ flags = params.flags | flags
+
+ params.tra_sa_in = VppIpsecSA(test, scapy_tra_sa_id, scapy_tra_spi,
+ auth_algo_vpp_id, auth_key,
+ crypt_algo_vpp_id, crypt_key,
+ test.vpp_esp_protocol,
+ flags=flags)
+ params.tra_sa_in.add_vpp_config()
+ params.tra_sa_out = VppIpsecSA(test, vpp_tra_sa_id, vpp_tra_spi,
+ auth_algo_vpp_id, auth_key,
+ crypt_algo_vpp_id, crypt_key,
+ test.vpp_esp_protocol,
+ flags=flags)
+ params.tra_sa_out.add_vpp_config()
+
+ VppIpsecSpdEntry(test, test.tra_spd, vpp_tra_sa_id,
+ addr_any, addr_bcast,
+ addr_any, addr_bcast,
+ socket.IPPROTO_ESP).add_vpp_config()
+ VppIpsecSpdEntry(test, test.tra_spd, vpp_tra_sa_id,
+ addr_any, addr_bcast,
+ addr_any, addr_bcast,
+ socket.IPPROTO_ESP,
+ is_outbound=0).add_vpp_config()
+
+ VppIpsecSpdEntry(test, test.tra_spd, vpp_tra_sa_id,
+ test.tra_if.local_addr[addr_type],
+ test.tra_if.local_addr[addr_type],
+ test.tra_if.remote_addr[addr_type],
+ test.tra_if.remote_addr[addr_type],
+ 0, priority=10,
+ policy=e.IPSEC_API_SPD_ACTION_PROTECT,
+ is_outbound=0).add_vpp_config()
+ VppIpsecSpdEntry(test, test.tra_spd, scapy_tra_sa_id,
+ test.tra_if.local_addr[addr_type],
+ test.tra_if.local_addr[addr_type],
+ test.tra_if.remote_addr[addr_type],
+ test.tra_if.remote_addr[addr_type],
+ 0, policy=e.IPSEC_API_SPD_ACTION_PROTECT,
+ priority=10).add_vpp_config()
+
+
class TemplateIpsecEsp(TemplateIpsec):
"""
Basic test for ipsec esp sanity - tunnel and transport modes.
@@ -42,6 +177,8 @@
|pg0| -------> |VPP| ------> |pg1|
--- --- ---
"""
+ config_esp_tun = config_esp_tun
+ config_esp_tra = config_esp_tra
def setUp(self):
super(TemplateIpsecEsp, self).setUp()
@@ -82,139 +219,8 @@
if not self.vpp_dead:
self.vapi.cli("show hardware")
- def config_esp_tun(self, params):
- addr_type = params.addr_type
- scapy_tun_sa_id = params.scapy_tun_sa_id
- scapy_tun_spi = params.scapy_tun_spi
- vpp_tun_sa_id = params.vpp_tun_sa_id
- vpp_tun_spi = params.vpp_tun_spi
- auth_algo_vpp_id = params.auth_algo_vpp_id
- auth_key = params.auth_key
- crypt_algo_vpp_id = params.crypt_algo_vpp_id
- crypt_key = params.crypt_key
- remote_tun_if_host = params.remote_tun_if_host
- addr_any = params.addr_any
- addr_bcast = params.addr_bcast
- e = VppEnum.vl_api_ipsec_spd_action_t
- params.tun_sa_in = VppIpsecSA(self, scapy_tun_sa_id, scapy_tun_spi,
- auth_algo_vpp_id, auth_key,
- crypt_algo_vpp_id, crypt_key,
- self.vpp_esp_protocol,
- self.tun_if.local_addr[addr_type],
- self.tun_if.remote_addr[addr_type])
- params.tun_sa_in.add_vpp_config()
- params.tun_sa_out = VppIpsecSA(self, vpp_tun_sa_id, vpp_tun_spi,
- auth_algo_vpp_id, auth_key,
- crypt_algo_vpp_id, crypt_key,
- self.vpp_esp_protocol,
- self.tun_if.remote_addr[addr_type],
- self.tun_if.local_addr[addr_type])
- params.tun_sa_out.add_vpp_config()
-
- params.spd_policy_in_any = VppIpsecSpdEntry(self, self.tun_spd,
- scapy_tun_sa_id,
- addr_any, addr_bcast,
- addr_any, addr_bcast,
- socket.IPPROTO_ESP)
- params.spd_policy_in_any.add_vpp_config()
- params.spd_policy_out_any = VppIpsecSpdEntry(self, self.tun_spd,
- scapy_tun_sa_id,
- addr_any, addr_bcast,
- addr_any, addr_bcast,
- socket.IPPROTO_ESP,
- is_outbound=0)
- params.spd_policy_out_any.add_vpp_config()
-
- VppIpsecSpdEntry(self, self.tun_spd, vpp_tun_sa_id,
- remote_tun_if_host, remote_tun_if_host,
- self.pg1.remote_addr[addr_type],
- self.pg1.remote_addr[addr_type],
- 0,
- priority=10,
- policy=e.IPSEC_API_SPD_ACTION_PROTECT,
- is_outbound=0).add_vpp_config()
- VppIpsecSpdEntry(self, self.tun_spd, scapy_tun_sa_id,
- self.pg1.remote_addr[addr_type],
- self.pg1.remote_addr[addr_type],
- remote_tun_if_host, remote_tun_if_host,
- 0,
- policy=e.IPSEC_API_SPD_ACTION_PROTECT,
- priority=10).add_vpp_config()
-
- VppIpsecSpdEntry(self, self.tun_spd, vpp_tun_sa_id,
- remote_tun_if_host, remote_tun_if_host,
- self.pg0.local_addr[addr_type],
- self.pg0.local_addr[addr_type],
- 0,
- priority=20,
- policy=e.IPSEC_API_SPD_ACTION_PROTECT,
- is_outbound=0).add_vpp_config()
- VppIpsecSpdEntry(self, self.tun_spd, scapy_tun_sa_id,
- self.pg0.local_addr[addr_type],
- self.pg0.local_addr[addr_type],
- remote_tun_if_host, remote_tun_if_host,
- 0,
- policy=e.IPSEC_API_SPD_ACTION_PROTECT,
- priority=20).add_vpp_config()
-
- def config_esp_tra(self, params):
- addr_type = params.addr_type
- scapy_tra_sa_id = params.scapy_tra_sa_id
- scapy_tra_spi = params.scapy_tra_spi
- vpp_tra_sa_id = params.vpp_tra_sa_id
- vpp_tra_spi = params.vpp_tra_spi
- auth_algo_vpp_id = params.auth_algo_vpp_id
- auth_key = params.auth_key
- crypt_algo_vpp_id = params.crypt_algo_vpp_id
- crypt_key = params.crypt_key
- addr_any = params.addr_any
- addr_bcast = params.addr_bcast
- flags = (VppEnum.vl_api_ipsec_sad_flags_t.
- IPSEC_API_SAD_FLAG_USE_ANTI_REPLAY)
- e = VppEnum.vl_api_ipsec_spd_action_t
-
- params.tra_sa_in = VppIpsecSA(self, scapy_tra_sa_id, scapy_tra_spi,
- auth_algo_vpp_id, auth_key,
- crypt_algo_vpp_id, crypt_key,
- self.vpp_esp_protocol,
- flags=flags)
- params.tra_sa_in.add_vpp_config()
- params.tra_sa_out = VppIpsecSA(self, vpp_tra_sa_id, vpp_tra_spi,
- auth_algo_vpp_id, auth_key,
- crypt_algo_vpp_id, crypt_key,
- self.vpp_esp_protocol,
- flags=flags)
- params.tra_sa_out.add_vpp_config()
-
- VppIpsecSpdEntry(self, self.tra_spd, vpp_tra_sa_id,
- addr_any, addr_bcast,
- addr_any, addr_bcast,
- socket.IPPROTO_ESP).add_vpp_config()
- VppIpsecSpdEntry(self, self.tra_spd, vpp_tra_sa_id,
- addr_any, addr_bcast,
- addr_any, addr_bcast,
- socket.IPPROTO_ESP,
- is_outbound=0).add_vpp_config()
-
- VppIpsecSpdEntry(self, self.tra_spd, vpp_tra_sa_id,
- self.tra_if.local_addr[addr_type],
- self.tra_if.local_addr[addr_type],
- self.tra_if.remote_addr[addr_type],
- self.tra_if.remote_addr[addr_type],
- 0, priority=10,
- policy=e.IPSEC_API_SPD_ACTION_PROTECT,
- is_outbound=0).add_vpp_config()
- VppIpsecSpdEntry(self, self.tra_spd, scapy_tra_sa_id,
- self.tra_if.local_addr[addr_type],
- self.tra_if.local_addr[addr_type],
- self.tra_if.remote_addr[addr_type],
- self.tra_if.remote_addr[addr_type],
- 0, policy=e.IPSEC_API_SPD_ACTION_PROTECT,
- priority=10).add_vpp_config()
-
-
-class TestIpsecEsp1(TemplateIpsecEsp, IpsecTraTests, IpsecTunTests):
+class TestIpsecEsp1(TemplateIpsecEsp, IpsecTra46Tests, IpsecTun46Tests):
""" Ipsec ESP - TUN & TRA tests """
tra4_encrypt_node_name = "esp4-encrypt"
tra4_decrypt_node_name = "esp4-decrypt"
@@ -231,5 +237,61 @@
pass
+class TemplateIpsecEspUdp(TemplateIpsec):
+ """
+ UDP encapped ESP
+ """
+ config_esp_tun = config_esp_tun
+ config_esp_tra = config_esp_tra
+
+ def setUp(self):
+ super(TemplateIpsecEspUdp, self).setUp()
+ self.encryption_type = ESP
+ self.tun_if = self.pg0
+ self.tra_if = self.pg2
+ self.logger.info(self.vapi.ppcli("show int addr"))
+
+ 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.tra_spd = VppIpsecSpd(self, self.tra_spd_id)
+ self.tra_spd.add_vpp_config()
+ VppIpsecSpdItfBinding(self, self.tra_spd,
+ self.tra_if).add_vpp_config()
+
+ self.config_esp_tra(p)
+ self.configure_sa_tra(p)
+
+ self.tun_spd = VppIpsecSpd(self, self.tun_spd_id)
+ self.tun_spd.add_vpp_config()
+ VppIpsecSpdItfBinding(self, self.tun_spd,
+ self.tun_if).add_vpp_config()
+
+ self.config_esp_tun(p)
+ self.logger.info(self.vapi.ppcli("show ipsec"))
+
+ d = DpoProto.DPO_PROTO_IP4
+ VppIpRoute(self, p.remote_tun_if_host, p.addr_len,
+ [VppRoutePath(self.tun_if.remote_addr[p.addr_type],
+ 0xffffffff,
+ proto=d)]).add_vpp_config()
+
+ def tearDown(self):
+ super(TemplateIpsecEspUdp, self).tearDown()
+ if not self.vpp_dead:
+ self.vapi.cli("show hardware")
+
+
+class TestIpsecEspUdp(TemplateIpsecEspUdp, IpsecTra4Tests, IpsecTun4Tests):
+ """ Ipsec NAT-T ESP UDP tests """
+ tra4_encrypt_node_name = "esp4-encrypt"
+ tra4_decrypt_node_name = "esp4-decrypt"
+ tun4_encrypt_node_name = "esp4-encrypt"
+ tun4_decrypt_node_name = "esp4-decrypt"
+ pass
+
+
if __name__ == '__main__':
unittest.main(testRunner=VppTestRunner)