ipsec: new api for sa ips and ports updates

Useful to update the tunnel paramaters and udp ports (NAT-T) of an SA
without having to rekey. Could be done by deleting and re-adding the
SA but it would not preserve the anti-replay window if there is one.
Use case: a nat update/reboot between the 2 endpoints of the tunnel.

Type: feature
Change-Id: Icf5c0aac218603e8aa9a008ed6f614e4a6db59a0
Signed-off-by: Arthur de Kerhor <arthurdekerhor@gmail.com>

3 files changed, 61 insertions, 6 deletions

diff --git a/test/template_ipsec.py b/test/template_ipsec.py
index 9d9ea3a86d3..d00216c7308 100644
--- a/test/template_ipsec.py
+++ b/test/template_ipsec.py
@@ -1291,7 +1291,7 @@ class IpsecTun4(object):
         decrypt_pkts = []
         for rx in rxs:
             if p.nat_header:
-                self.assertEqual(rx[UDP].dport, 4500)
+                self.assertEqual(rx[UDP].dport, p.nat_header.dport)
             self.assert_packet_checksums_valid(rx)
             self.assertEqual(len(rx) - len(Ether()), rx[IP].len)
             try:
diff --git a/test/test_ipsec_tun_if_esp.py b/test/test_ipsec_tun_if_esp.py
index 61a66d40a4e..fe05f98e6e6 100644
--- a/test/test_ipsec_tun_if_esp.py
+++ b/test/test_ipsec_tun_if_esp.py
@@ -300,7 +300,7 @@ class TemplateIpsec4TunIfEspUdp(TemplateIpsec4TunProtect, TemplateIpsec):
                 # which strips them
                 self.assertTrue(rx.haslayer(UDP))
                 self.assert_equal(rx[UDP].sport, p.nat_header.sport)
-                self.assert_equal(rx[UDP].dport, 4500)
+                self.assert_equal(rx[UDP].dport, p.nat_header.dport)
 
                 pkt = sa.decrypt(rx[IP])
                 if not pkt.haslayer(IP):
@@ -344,7 +344,8 @@ class TemplateIpsec4TunIfEspUdp(TemplateIpsec4TunProtect, TemplateIpsec):
             p.crypt_algo_vpp_id,
             p.crypt_key,
             self.vpp_esp_protocol,
-            flags=p.flags,
+            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,
         )
@@ -429,6 +430,24 @@ class TestIpsec4TunIfEspUdpGCM(TemplateIpsec4TunIfEspUdp, IpsecTun4Tests):
         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"""
 
@@ -583,7 +602,7 @@ class TemplateIpsec6TunIfEspUdp(TemplateIpsec6TunProtect, TemplateIpsec):
                 # which strips them
                 self.assertTrue(rx.haslayer(UDP))
                 self.assert_equal(rx[UDP].sport, p.nat_header.sport)
-                self.assert_equal(rx[UDP].dport, 4500)
+                self.assert_equal(rx[UDP].dport, p.nat_header.dport)
 
                 pkt = sa.decrypt(rx[IP])
                 if not pkt.haslayer(IP):
@@ -629,7 +648,8 @@ class TemplateIpsec6TunIfEspUdp(TemplateIpsec6TunProtect, TemplateIpsec):
             p.crypt_algo_vpp_id,
             p.crypt_key,
             self.vpp_esp_protocol,
-            flags=p.flags,
+            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,
         )
@@ -2957,7 +2977,8 @@ class TemplateIpsecItf4(object):
             self.vpp_esp_protocol,
             dst,
             src,
-            flags=p.flags,
+            flags=p.flags
+            | VppEnum.vl_api_ipsec_sad_flags_t.IPSEC_API_SAD_FLAG_IS_INBOUND,
         )
         p.tun_sa_in.add_vpp_config()
 
@@ -3063,6 +3084,20 @@ class TestIpsecItf4(TemplateIpsec, TemplateIpsecItf4, IpsecTun4):
 
         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
diff --git a/test/vpp_ipsec.py b/test/vpp_ipsec.py
index eb0209fc57a..f50d491c396 100644
--- a/test/vpp_ipsec.py
+++ b/test/vpp_ipsec.py
@@ -295,6 +295,26 @@ class VppIpsecSA(VppObject):
         self.test.registry.register(self, self.test.logger)
         return self
 
+    def update_vpp_config(
+        self, udp_src=None, udp_dst=None, is_tun=False, tun_src=None, tun_dst=None
+    ):
+        if is_tun:
+            if tun_src:
+                self.tun_src = ip_address(text_type(tun_src))
+            if tun_dst:
+                self.tun_dst = ip_address(text_type(tun_dst))
+        if udp_src:
+            self.udp_src = udp_src
+        if udp_dst:
+            self.udp_dst = udp_dst
+        self.test.vapi.ipsec_sad_entry_update(
+            sad_id=self.id,
+            is_tun=is_tun,
+            tunnel=self.tunnel_encode(),
+            udp_src_port=udp_src,
+            udp_dst_port=udp_dst,
+        )
+
     def remove_vpp_config(self):
         self.test.vapi.ipsec_sad_entry_del(id=self.id)