TEST: IPSEC NAT-T with UDP header

Change-Id: I5ef8b3f4be40a7a0b0f1cb90dc0e15a4711e8664
Signed-off-by: Neale Ranns <nranns@cisco.com>

1 files changed, 197 insertions, 135 deletions

diff --git a/test/test_ipsec_esp.py b/test/test_ipsec_esp.py
index 7a05f0d2183..09b72401541 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_ip import DpoProto
 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 @@ class TemplateIpsecEsp(TemplateIpsec):
     |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 @@ class TemplateIpsecEsp(TemplateIpsec):
         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 @@ class TestIpsecEsp2(TemplateIpsecEsp, IpsecTcpTests):
     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)