aboutsummaryrefslogtreecommitdiffstats
path: root/src/plugins/ikev2/test/test_ikev2.py
diff options
context:
space:
mode:
Diffstat (limited to 'src/plugins/ikev2/test/test_ikev2.py')
-rw-r--r--src/plugins/ikev2/test/test_ikev2.py63
1 files changed, 62 insertions, 1 deletions
diff --git a/src/plugins/ikev2/test/test_ikev2.py b/src/plugins/ikev2/test/test_ikev2.py
index 61dd53e7988..453a47e04c8 100644
--- a/src/plugins/ikev2/test/test_ikev2.py
+++ b/src/plugins/ikev2/test/test_ikev2.py
@@ -1,4 +1,5 @@
import os
+import time
from socket import inet_pton
from cryptography import x509
from cryptography.hazmat.backends import default_backend
@@ -584,7 +585,6 @@ class IkePeer(VppTestCase):
self.sa.generate_dh_data()
self.vapi.cli('ikev2 set logging level 4')
self.vapi.cli('event-lo clear')
- self.vapi.cli('ikev2 dpd disable')
def create_packet(self, src_if, msg, sport=500, dport=500, natt=False,
use_ip6=False):
@@ -1289,6 +1289,10 @@ class Ikev2Params(object):
'SHA2-384-192': ei.IPSEC_API_INTEG_ALG_SHA_384_192,
'SHA2-512-256': ei.IPSEC_API_INTEG_ALG_SHA_512_256}
+ dpd_disabled = True if 'dpd_disabled' not in params else\
+ params['dpd_disabled']
+ if dpd_disabled:
+ self.vapi.cli('ikev2 dpd disable')
self.del_sa_from_responder = False if 'del_sa_from_responder'\
not in params else params['del_sa_from_responder']
is_natt = 'natt' in params and params['natt'] or False
@@ -1663,6 +1667,35 @@ class TestResponderPsk(TemplateResponder, Ikev2Params):
self.config_params()
+class TestResponderDpd(TestResponderPsk):
+ """
+ Dead peer detection test
+ """
+ def config_tc(self):
+ self.config_params({'dpd_disabled': False})
+
+ def tearDown(self):
+ pass
+
+ def test_responder(self):
+ self.vapi.ikev2_profile_set_liveness(period=2, max_retries=1)
+ super(TestResponderDpd, self).test_responder()
+ self.pg0.enable_capture()
+ self.pg_start()
+ # capture empty request but don't reply
+ capture = self.pg0.get_capture(expected_count=1, timeout=5)
+ ih = self.get_ike_header(capture[0])
+ self.assertEqual(ih.exch_type, 37) # INFORMATIONAL
+ plain = self.sa.hmac_and_decrypt(ih)
+ self.assertEqual(plain, b'')
+ # wait for SA expiration
+ time.sleep(3)
+ ike_sas = self.vapi.ikev2_sa_dump()
+ self.assertEqual(len(ike_sas), 0)
+ ipsec_sas = self.vapi.ipsec_sa_dump()
+ self.assertEqual(len(ipsec_sas), 0)
+
+
class TestResponderRekey(TestResponderPsk):
""" test ikev2 responder - rekey """
@@ -1755,6 +1788,34 @@ class Test_IKE_AES_GCM_16_256(TemplateResponder, Ikev2Params):
'end_addr': '11::100'}})
+class TestInitiatorKeepaliveMsg(TestInitiatorPsk):
+ """
+ Test for keep alive messages
+ """
+
+ def send_empty_req_from_responder(self):
+ header = ikev2.IKEv2(init_SPI=self.sa.ispi, resp_SPI=self.sa.rspi,
+ id=self.sa.new_msg_id(), flags='Initiator',
+ exch_type='INFORMATIONAL',
+ next_payload='Encrypted')
+
+ msg = self.encrypt_ike_msg(header, b'', None)
+ packet = self.create_packet(self.pg0, msg, self.sa.sport,
+ self.sa.dport, self.sa.natt, self.ip6)
+ self.pg0.add_stream(packet)
+ self.pg0.enable_capture()
+ self.pg_start()
+ capture = self.pg0.get_capture(1)
+ ih = self.get_ike_header(capture[0])
+ self.assertEqual(ih.id, self.sa.msg_id)
+ plain = self.sa.hmac_and_decrypt(ih)
+ self.assertEqual(plain, b'')
+
+ def test_initiator(self):
+ super(TestInitiatorKeepaliveMsg, self).test_initiator()
+ self.send_empty_req_from_responder()
+
+
class TestMalformedMessages(TemplateResponder, Ikev2Params):
""" malformed packet test """