IPSEC-AH: anti-replay testing

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

4 files changed, 111 insertions, 30 deletions

diff --git a/test/template_ipsec.py b/test/template_ipsec.py
index 71485c58239..961f63aa59c 100644
--- a/test/template_ipsec.py
+++ b/test/template_ipsec.py
@@ -158,20 +158,20 @@ class TemplateIpsec(VppTestCase):
                 src=self.tun_if.local_addr[params.addr_type]))
         return vpp_tun_sa, scapy_tun_sa
 
-    def configure_sa_tra(self, 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)
-        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)
-        return vpp_tra_sa, scapy_tra_sa
+    @classmethod
+    def configure_sa_tra(cls, params):
+        params.scapy_tra_sa = SecurityAssociation(cls.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(cls.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)
 
 
 class IpsecTcpTests(object):
@@ -192,24 +192,96 @@ class IpsecTcpTests(object):
 
 
 class IpsecTraTests(object):
+    def test_tra_anti_replay(self, count=1):
+        """ ipsec v4 transport anti-reply test """
+        p = self.params[socket.AF_INET]
+
+        # fire in a packet with seq number 1
+        pkt = (Ether(src=self.tra_if.remote_mac,
+                     dst=self.tra_if.local_mac) /
+               p.scapy_tra_sa.encrypt(IP(src=self.tra_if.remote_ip4,
+                                         dst=self.tra_if.local_ip4) /
+                                      ICMP(),
+                                      seq_num=1))
+        recv_pkts = self.send_and_expect(self.tra_if, [pkt], self.tra_if)
+
+        # now move the window over to 235
+        pkt = (Ether(src=self.tra_if.remote_mac,
+                     dst=self.tra_if.local_mac) /
+               p.scapy_tra_sa.encrypt(IP(src=self.tra_if.remote_ip4,
+                                         dst=self.tra_if.local_ip4) /
+                                      ICMP(),
+                                      seq_num=235))
+        recv_pkts = self.send_and_expect(self.tra_if, [pkt], self.tra_if)
+
+        # the window size is 64 packets
+        # in window are still accepted
+        pkt = (Ether(src=self.tra_if.remote_mac,
+                     dst=self.tra_if.local_mac) /
+               p.scapy_tra_sa.encrypt(IP(src=self.tra_if.remote_ip4,
+                                         dst=self.tra_if.local_ip4) /
+                                      ICMP(),
+                                      seq_num=172))
+        recv_pkts = self.send_and_expect(self.tra_if, [pkt], self.tra_if)
+
+        # out of window are dropped
+        pkt = (Ether(src=self.tra_if.remote_mac,
+                     dst=self.tra_if.local_mac) /
+               p.scapy_tra_sa.encrypt(IP(src=self.tra_if.remote_ip4,
+                                         dst=self.tra_if.local_ip4) /
+                                      ICMP(),
+                                      seq_num=17))
+        self.send_and_assert_no_replies(self.tra_if, pkt * 17)
+
+        err = self.statistics.get_counter(
+            '/err/%s/SA replayed packet' % self.tra4_decrypt_node_name)
+        self.assertEqual(err, 17)
+
+        # a packet that does not decrypt does not move the window forward
+        bogus_sa = SecurityAssociation(self.encryption_type,
+                                       p.vpp_tra_spi)
+        pkt = (Ether(src=self.tra_if.remote_mac,
+                     dst=self.tra_if.local_mac) /
+               bogus_sa.encrypt(IP(src=self.tra_if.remote_ip4,
+                                   dst=self.tra_if.local_ip4) /
+                                ICMP(),
+                                seq_num=350))
+        self.send_and_assert_no_replies(self.tra_if, pkt * 17)
+
+        err = self.statistics.get_counter(
+            '/err/%s/Integrity check failed' % self.tra4_decrypt_node_name)
+        self.assertEqual(err, 17)
+
+        # which we can determine since this packet is still in the window
+        pkt = (Ether(src=self.tra_if.remote_mac,
+                     dst=self.tra_if.local_mac) /
+               p.scapy_tra_sa.encrypt(IP(src=self.tra_if.remote_ip4,
+                                         dst=self.tra_if.local_ip4) /
+                                      ICMP(),
+                                      seq_num=234))
+        recv_pkts = self.send_and_expect(self.tra_if, [pkt], self.tra_if)
+
+        # move the security-associations seq number on to the last we used
+        p.scapy_tra_sa.seq_num = 351
+        p.vpp_tra_sa.seq_num = 351
+
     def test_tra_basic(self, count=1):
         """ ipsec v4 transport basic test """
         self.vapi.cli("clear errors")
         try:
             p = self.params[socket.AF_INET]
-            vpp_tra_sa, scapy_tra_sa = self.configure_sa_tra(p)
-            send_pkts = self.gen_encrypt_pkts(scapy_tra_sa, self.tra_if,
+            send_pkts = self.gen_encrypt_pkts(p.scapy_tra_sa, self.tra_if,
                                               src=self.tra_if.remote_ip4,
                                               dst=self.tra_if.local_ip4,
                                               count=count)
             recv_pkts = self.send_and_expect(self.tra_if, send_pkts,
                                              self.tra_if)
-            for p in recv_pkts:
+            for rx in recv_pkts:
                 try:
-                    decrypted = vpp_tra_sa.decrypt(p[IP])
+                    decrypted = p.vpp_tra_sa.decrypt(rx[IP])
                     self.assert_packet_checksums_valid(decrypted)
                 except:
-                    self.logger.debug(ppp("Unexpected packet:", p))
+                    self.logger.debug(ppp("Unexpected packet:", rx))
                     raise
         finally:
             self.logger.info(self.vapi.ppcli("show error"))
@@ -227,19 +299,18 @@ class IpsecTraTests(object):
         self.vapi.cli("clear errors")
         try:
             p = self.params[socket.AF_INET6]
-            vpp_tra_sa, scapy_tra_sa = self.configure_sa_tra(p)
-            send_pkts = self.gen_encrypt_pkts6(scapy_tra_sa, self.tra_if,
+            send_pkts = self.gen_encrypt_pkts6(p.scapy_tra_sa, self.tra_if,
                                                src=self.tra_if.remote_ip6,
                                                dst=self.tra_if.local_ip6,
                                                count=count)
             recv_pkts = self.send_and_expect(self.tra_if, send_pkts,
                                              self.tra_if)
-            for p in recv_pkts:
+            for rx in recv_pkts:
                 try:
-                    decrypted = vpp_tra_sa.decrypt(p[IPv6])
+                    decrypted = p.vpp_tra_sa.decrypt(rx[IPv6])
                     self.assert_packet_checksums_valid(decrypted)
                 except:
-                    self.logger.debug(ppp("Unexpected packet:", p))
+                    self.logger.debug(ppp("Unexpected packet:", rx))
                     raise
         finally:
             self.logger.info(self.vapi.ppcli("show error"))
diff --git a/test/test_ipsec_ah.py b/test/test_ipsec_ah.py
index e832bfa2a27..928cd53c1f1 100644
--- a/test/test_ipsec_ah.py
+++ b/test/test_ipsec_ah.py
@@ -45,6 +45,7 @@ class TemplateIpsecAh(TemplateIpsec):
                                              cls.tra_if.sw_if_index)
         for _, p in cls.params.items():
             cls.config_ah_tra(p)
+            cls.configure_sa_tra(p)
         cls.logger.info(cls.vapi.ppcli("show ipsec"))
         for _, p in cls.params.items():
             cls.config_ah_tun(p)
@@ -134,12 +135,14 @@ class TemplateIpsecAh(TemplateIpsec):
                                          auth_algo_vpp_id, auth_key,
                                          crypt_algo_vpp_id, crypt_key,
                                          cls.vpp_ah_protocol, is_tunnel=0,
-                                         is_tunnel_ipv6=0)
+                                         is_tunnel_ipv6=0,
+                                         use_anti_replay=1)
         cls.vapi.ipsec_sad_add_del_entry(vpp_tra_sa_id, vpp_tra_spi,
                                          auth_algo_vpp_id, auth_key,
                                          crypt_algo_vpp_id, crypt_key,
                                          cls.vpp_ah_protocol, is_tunnel=0,
-                                         is_tunnel_ipv6=0)
+                                         is_tunnel_ipv6=0,
+                                         use_anti_replay=1)
         l_startaddr = r_startaddr = socket.inet_pton(addr_type, addr_any)
         l_stopaddr = r_stopaddr = socket.inet_pton(addr_type, addr_bcast)
         cls.vapi.ipsec_spd_add_del_entry(cls.tra_spd_id, vpp_tra_sa_id,
diff --git a/test/test_ipsec_esp.py b/test/test_ipsec_esp.py
index ed9d0d9d4ce..d22f965a31b 100644
--- a/test/test_ipsec_esp.py
+++ b/test/test_ipsec_esp.py
@@ -51,6 +51,7 @@ class TemplateIpsecEsp(TemplateIpsec):
                                              cls.tra_if.sw_if_index)
         for _, p in cls.params.items():
             cls.config_esp_tra(p)
+            cls.configure_sa_tra(p)
         cls.logger.info(cls.vapi.ppcli("show ipsec"))
         cls.vapi.ipsec_spd_add_del(cls.tun_spd_id)
         cls.vapi.ipsec_interface_add_del_spd(cls.tun_spd_id,
@@ -144,11 +145,13 @@ class TemplateIpsecEsp(TemplateIpsec):
         cls.vapi.ipsec_sad_add_del_entry(scapy_tra_sa_id, scapy_tra_spi,
                                          auth_algo_vpp_id, auth_key,
                                          crypt_algo_vpp_id, crypt_key,
-                                         cls.vpp_esp_protocol, is_tunnel=0)
+                                         cls.vpp_esp_protocol, is_tunnel=0,
+                                         use_anti_replay=1)
         cls.vapi.ipsec_sad_add_del_entry(vpp_tra_sa_id, vpp_tra_spi,
                                          auth_algo_vpp_id, auth_key,
                                          crypt_algo_vpp_id, crypt_key,
-                                         cls.vpp_esp_protocol, is_tunnel=0)
+                                         cls.vpp_esp_protocol, is_tunnel=0,
+                                         use_anti_replay=1)
         l_startaddr = r_startaddr = socket.inet_pton(addr_type, addr_any)
         l_stopaddr = r_stopaddr = socket.inet_pton(addr_type, addr_bcast)
         cls.vapi.ipsec_spd_add_del_entry(cls.tra_spd_id, vpp_tra_sa_id,
diff --git a/test/vpp_papi_provider.py b/test/vpp_papi_provider.py
index 370758764df..130f17868b3 100644
--- a/test/vpp_papi_provider.py
+++ b/test/vpp_papi_provider.py
@@ -3348,7 +3348,9 @@ class VppPapiProvider(object):
                                 is_tunnel=1,
                                 is_tunnel_ipv6=0,
                                 is_add=1,
-                                udp_encap=0):
+                                udp_encap=0,
+                                use_anti_replay=0,
+                                use_extended_sequence_number=0):
         """ IPSEC SA add/del
         :param sad_id: security association ID
         :param spi: security param index of the SA in decimal
@@ -3381,7 +3383,9 @@ class VppPapiProvider(object):
              'is_add': is_add,
              'is_tunnel': is_tunnel,
              'is_tunnel_ipv6': is_tunnel_ipv6,
-             'udp_encap': udp_encap})
+             'udp_encap': udp_encap,
+             'use_extended_sequence_number': use_extended_sequence_number,
+             'use_anti_replay': use_anti_replay})
 
     def ipsec_spd_add_del_entry(self,
                                 spd_id,