summaryrefslogtreecommitdiffstats
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.py174
1 files changed, 126 insertions, 48 deletions
diff --git a/src/plugins/ikev2/test/test_ikev2.py b/src/plugins/ikev2/test/test_ikev2.py
index 918b5cef624..77698d6a5ed 100644
--- a/src/plugins/ikev2/test/test_ikev2.py
+++ b/src/plugins/ikev2/test/test_ikev2.py
@@ -19,6 +19,9 @@ from vpp_papi import VppEnum
KEY_PAD = b"Key Pad for IKEv2"
+SALT_SIZE = 4
+GCM_ICV_SIZE = 16
+GCM_IV_SIZE = 8
# defined in rfc3526
@@ -65,19 +68,47 @@ class CryptoAlgo(object):
if self.cipher is not None:
self.bs = self.cipher.block_size // 8
- def encrypt(self, data, key):
- iv = os.urandom(self.bs)
- encryptor = Cipher(self.cipher(key), self.mode(iv),
- default_backend()).encryptor()
- return iv + encryptor.update(data) + encryptor.finalize()
-
- def decrypt(self, data, key, icv=None):
- iv = data[:self.bs]
- ct = data[self.bs:]
- decryptor = Cipher(algorithms.AES(key),
- modes.CBC(iv),
- default_backend()).decryptor()
- return decryptor.update(ct) + decryptor.finalize()
+ if self.name == 'AES-GCM-16ICV':
+ self.iv_len = GCM_IV_SIZE
+ else:
+ self.iv_len = self.bs
+
+ def encrypt(self, data, key, aad=None):
+ iv = os.urandom(self.iv_len)
+ if aad is None:
+ encryptor = Cipher(self.cipher(key), self.mode(iv),
+ default_backend()).encryptor()
+ return iv + encryptor.update(data) + encryptor.finalize()
+ else:
+ salt = key[-SALT_SIZE:]
+ nonce = salt + iv
+ encryptor = Cipher(self.cipher(key[:-SALT_SIZE]), self.mode(nonce),
+ default_backend()).encryptor()
+ encryptor.authenticate_additional_data(aad)
+ data = encryptor.update(data) + encryptor.finalize()
+ data += encryptor.tag[:GCM_ICV_SIZE]
+ return iv + data
+
+ def decrypt(self, data, key, aad=None, icv=None):
+ if aad is None:
+ iv = data[:self.iv_len]
+ ct = data[self.iv_len:]
+ decryptor = Cipher(algorithms.AES(key),
+ self.mode(iv),
+ default_backend()).decryptor()
+ return decryptor.update(ct) + decryptor.finalize()
+ else:
+ salt = key[-SALT_SIZE:]
+ nonce = salt + data[:GCM_IV_SIZE]
+ ct = data[GCM_IV_SIZE:]
+ key = key[:-SALT_SIZE]
+ decryptor = Cipher(algorithms.AES(key),
+ self.mode(nonce, icv, len(icv)),
+ default_backend()).decryptor()
+ decryptor.authenticate_additional_data(aad)
+ pt = decryptor.update(ct) + decryptor.finalize()
+ pad_len = pt[-1] + 1
+ return pt[:-pad_len]
def pad(self, data):
pad_len = (len(data) // self.bs + 1) * self.bs - len(data)
@@ -241,7 +272,7 @@ class IKEv2SA(object):
return r
def calc_prf(self, prf, key, data):
- h = self.ike_integ_alg.mac(key, prf, backend=default_backend())
+ h = self.ike_prf_alg.mac(key, prf, backend=default_backend())
h.update(data)
return h.finalize()
@@ -258,10 +289,16 @@ class IKEv2SA(object):
encr_key_len = self.ike_crypto_key_len
tr_prf_key_len = self.ike_prf_alg.key_len
integ_key_len = self.ike_integ_alg.key_len
+ if integ_key_len == 0:
+ salt_size = 4
+ else:
+ salt_size = 0
+
l = (prf_key_trunc +
integ_key_len * 2 +
encr_key_len * 2 +
- tr_prf_key_len * 2)
+ tr_prf_key_len * 2 +
+ salt_size * 2)
keymat = self.calc_prfplus(prf, self.skeyseed, s, l)
pos = 0
@@ -273,10 +310,10 @@ class IKEv2SA(object):
self.sk_ar = keymat[pos:pos+integ_key_len]
pos += integ_key_len
- self.sk_ei = keymat[pos:pos+encr_key_len]
- pos += encr_key_len
- self.sk_er = keymat[pos:pos+encr_key_len]
- pos += encr_key_len
+ self.sk_ei = keymat[pos:pos+encr_key_len + salt_size]
+ pos += encr_key_len + salt_size
+ self.sk_er = keymat[pos:pos+encr_key_len + salt_size]
+ pos += encr_key_len + salt_size
self.sk_pi = keymat[pos:pos+tr_prf_key_len]
pos += tr_prf_key_len
@@ -303,9 +340,9 @@ class IKEv2SA(object):
else:
raise TypeError('unknown auth method type!')
- def encrypt(self, data):
+ def encrypt(self, data, aad=None):
data = self.ike_crypto_alg.pad(data)
- return self.ike_crypto_alg.encrypt(data, self.my_cryptokey)
+ return self.ike_crypto_alg.encrypt(data, self.my_cryptokey, aad)
@property
def peer_authkey(self):
@@ -355,17 +392,23 @@ class IKEv2SA(object):
h.update(data)
return h.finalize()
- def decrypt(self, data):
- return self.ike_crypto_alg.decrypt(data, self.peer_cryptokey)
+ def decrypt(self, data, aad=None, icv=None):
+ return self.ike_crypto_alg.decrypt(data, self.peer_cryptokey, aad, icv)
def hmac_and_decrypt(self, ike):
ep = ike[ikev2.IKEv2_payload_Encrypted]
- self.verify_hmac(raw(ike))
- integ_trunc = self.ike_integ_alg.trunc_len
+ if self.ike_crypto == 'AES-GCM-16ICV':
+ aad_len = len(ikev2.IKEv2_payload_Encrypted()) + len(ikev2.IKEv2())
+ ct = ep.load[:-GCM_ICV_SIZE]
+ tag = ep.load[-GCM_ICV_SIZE:]
+ return self.decrypt(ct, raw(ike)[:aad_len], tag)
+ else:
+ self.verify_hmac(raw(ike))
+ integ_trunc = self.ike_integ_alg.trunc_len
- # remove ICV and decrypt payload
- ct = ep.load[:-integ_trunc]
- return self.decrypt(ct)
+ # remove ICV and decrypt payload
+ ct = ep.load[:-integ_trunc]
+ return self.decrypt(ct)
def generate_ts(self):
c = self.child_sas[0]
@@ -388,7 +431,7 @@ class IKEv2SA(object):
if integ not in AUTH_ALGOS:
raise TypeError('unsupported auth algo %r' % integ)
- self.ike_integ = integ
+ self.ike_integ = None if integ == 'NULL' else integ
self.ike_integ_alg = AUTH_ALGOS[integ]
if prf not in PRF_ALGOS:
@@ -411,7 +454,7 @@ class IKEv2SA(object):
self.esp_integ_alg = AUTH_ALGOS[integ]
def crypto_attr(self, key_len):
- if self.ike_crypto in ['AES-CBC', 'AES-GCM']:
+ if self.ike_crypto in ['AES-CBC', 'AES-GCM-16ICV']:
return (0x800e << 16 | key_len << 3, 12)
else:
raise Exception('unsupported attribute type')
@@ -542,24 +585,47 @@ class TemplateResponder(VppTestCase):
number_of_TSs=len(tsr),
traffic_selector=tsr) /
ikev2.IKEv2_payload_Notify(type='INITIAL_CONTACT'))
- encr = self.sa.encrypt(raw(plain))
-
- trunc_len = self.sa.ike_integ_alg.trunc_len
- plen = len(encr) + len(ikev2.IKEv2_payload_Encrypted()) + trunc_len
- tlen = plen + len(ikev2.IKEv2())
-
- sk_p = ikev2.IKEv2_payload_Encrypted(next_payload='IDi',
- length=plen, load=encr)
- sa_auth = (ikev2.IKEv2(init_SPI=self.sa.ispi, resp_SPI=self.sa.rspi,
- length=tlen, flags='Initiator', exch_type='IKE_AUTH', id=1))
- sa_auth /= sk_p
-
- integ_data = raw(sa_auth)
- hmac_data = self.sa.compute_hmac(self.sa.ike_integ_alg.mod(),
- self.sa.my_authkey, integ_data)
- sa_auth = sa_auth / Raw(hmac_data[:trunc_len])
- assert(len(sa_auth) == tlen)
+ if self.sa.ike_crypto == 'AES-GCM-16ICV':
+ data = self.sa.ike_crypto_alg.pad(raw(plain))
+ plen = len(data) + GCM_IV_SIZE + GCM_ICV_SIZE +\
+ len(ikev2.IKEv2_payload_Encrypted())
+ tlen = plen + len(ikev2.IKEv2())
+
+ # prepare aad data
+ sk_p = ikev2.IKEv2_payload_Encrypted(next_payload='IDi',
+ length=plen)
+ sa_auth = (ikev2.IKEv2(init_SPI=self.sa.ispi,
+ resp_SPI=self.sa.rspi, id=1,
+ length=tlen, flags='Initiator', exch_type='IKE_AUTH'))
+ sa_auth /= sk_p
+
+ encr = self.sa.encrypt(raw(plain), raw(sa_auth))
+ sk_p = ikev2.IKEv2_payload_Encrypted(next_payload='IDi',
+ length=plen, load=encr)
+ sa_auth = (ikev2.IKEv2(init_SPI=self.sa.ispi,
+ resp_SPI=self.sa.rspi, id=1,
+ length=tlen, flags='Initiator', exch_type='IKE_AUTH'))
+ sa_auth /= sk_p
+ else:
+ encr = self.sa.encrypt(raw(plain))
+ trunc_len = self.sa.ike_integ_alg.trunc_len
+ plen = len(encr) + len(ikev2.IKEv2_payload_Encrypted()) + trunc_len
+ tlen = plen + len(ikev2.IKEv2())
+
+ sk_p = ikev2.IKEv2_payload_Encrypted(next_payload='IDi',
+ length=plen, load=encr)
+ sa_auth = (ikev2.IKEv2(init_SPI=self.sa.ispi,
+ resp_SPI=self.sa.rspi, id=1,
+ length=tlen, flags='Initiator', exch_type='IKE_AUTH'))
+ sa_auth /= sk_p
+
+ integ_data = raw(sa_auth)
+ hmac_data = self.sa.compute_hmac(self.sa.ike_integ_alg.mod(),
+ self.sa.my_authkey, integ_data)
+ sa_auth = sa_auth / Raw(hmac_data[:trunc_len])
+
+ assert(len(sa_auth) == tlen)
packet = self.create_ike_msg(self.pg0, sa_auth, self.sa.sport,
self.sa.dport, self.sa.natt)
self.pg0.add_stream(packet)
@@ -590,8 +656,9 @@ class TemplateResponder(VppTestCase):
sa = ih[ikev2.IKEv2_payload_SA]
self.sa.r_nonce = ih[ikev2.IKEv2_payload_Nonce].load
self.sa.r_dh_data = ih[ikev2.IKEv2_payload_KE].load
- except AttributeError as e:
+ except IndexError as e:
self.logger.error("unexpected reply: SA/Nonce/KE payload found!")
+ self.logger.error(ih.show())
raise
self.sa.complete_dh_data()
self.sa.calc_keys()
@@ -783,5 +850,16 @@ class TestAES_CBC_128_SHA256_128_MODP3072_ESP_AES_GCM_16\
'ike-dh': '3072MODPgr'})
+class Test_IKE_AES_GCM_16_256(TemplateResponder, Ikev2Params):
+ """
+ IKE:AES_GCM_16_256
+ """
+ def config_tc(self):
+ self.config_params({
+ 'ike-crypto': ('AES-GCM-16ICV', 32),
+ 'ike-integ': 'NULL',
+ 'ike-dh': '2048MODPgr'})
+
+
if __name__ == '__main__':
unittest.main(testRunner=VppTestRunner)