summaryrefslogtreecommitdiffstats
path: root/test/test_ikev2.py
diff options
context:
space:
mode:
Diffstat (limited to 'test/test_ikev2.py')
-rw-r--r--test/test_ikev2.py1698
1 files changed, 948 insertions, 750 deletions
diff --git a/test/test_ikev2.py b/test/test_ikev2.py
index 58a7ec3cd9a..5b699dd0d8d 100644
--- a/test/test_ikev2.py
+++ b/test/test_ikev2.py
@@ -38,7 +38,9 @@ GCM_IV_SIZE = 8
# defined in rfc3526
# tuple structure is (p, g, key_len)
DH = {
- '2048MODPgr': (long_converter("""
+ "2048MODPgr": (
+ long_converter(
+ """
FFFFFFFF FFFFFFFF C90FDAA2 2168C234 C4C6628B 80DC1CD1
29024E08 8A67CC74 020BBEA6 3B139B22 514A0879 8E3404DD
EF9519B3 CD3A431B 302B0A6D F25F1437 4FE1356D 6D51C245
@@ -49,9 +51,14 @@ DH = {
670C354E 4ABC9804 F1746C08 CA18217C 32905E46 2E36CE3B
E39E772C 180E8603 9B2783A2 EC07A28F B5C55DF0 6F4C52C9
DE2BCBF6 95581718 3995497C EA956AE5 15D22618 98FA0510
- 15728E5A 8AACAA68 FFFFFFFF FFFFFFFF"""), 2, 256),
-
- '3072MODPgr': (long_converter("""
+ 15728E5A 8AACAA68 FFFFFFFF FFFFFFFF"""
+ ),
+ 2,
+ 256,
+ ),
+ "3072MODPgr": (
+ long_converter(
+ """
FFFFFFFF FFFFFFFF C90FDAA2 2168C234 C4C6628B 80DC1CD1
29024E08 8A67CC74 020BBEA6 3B139B22 514A0879 8E3404DD
EF9519B3 CD3A431B 302B0A6D F25F1437 4FE1356D 6D51C245
@@ -67,7 +74,11 @@ DH = {
ABF5AE8C DB0933D7 1E8C94E0 4A25619D CEE3D226 1AD2EE6B
F12FFA06 D98A0864 D8760273 3EC86A64 521F2B18 177B200C
BBE11757 7A615D6C 770988C0 BAD946E2 08E24FA0 74E5AB31
- 43DB5BFC E0FD108E 4B82D120 A93AD2CA FFFFFFFF FFFFFFFF"""), 2, 384)
+ 43DB5BFC E0FD108E 4B82D120 A93AD2CA FFFFFFFF FFFFFFFF"""
+ ),
+ 2,
+ 384,
+ ),
}
@@ -79,7 +90,7 @@ class CryptoAlgo(object):
if self.cipher is not None:
self.bs = self.cipher.block_size // 8
- if self.name == 'AES-GCM-16ICV':
+ if self.name == "AES-GCM-16ICV":
self.iv_len = GCM_IV_SIZE
else:
self.iv_len = self.bs
@@ -87,14 +98,16 @@ class CryptoAlgo(object):
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()
+ 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 = 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]
@@ -102,26 +115,26 @@ class CryptoAlgo(object):
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()
+ 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 = Cipher(
+ algorithms.AES(key), self.mode(nonce, icv, len(icv)), default_backend()
+ ).decryptor()
decryptor.authenticate_additional_data(aad)
return decryptor.update(ct) + decryptor.finalize()
def pad(self, data):
pad_len = (len(data) // self.bs + 1) * self.bs - len(data)
- data = data + b'\x00' * (pad_len - 1)
+ data = data + b"\x00" * (pad_len - 1)
return data + bytes([pad_len - 1])
@@ -135,36 +148,34 @@ class AuthAlgo(object):
CRYPTO_ALGOS = {
- 'NULL': CryptoAlgo('NULL', cipher=None, mode=None),
- 'AES-CBC': CryptoAlgo('AES-CBC', cipher=algorithms.AES, mode=modes.CBC),
- 'AES-GCM-16ICV': CryptoAlgo('AES-GCM-16ICV', cipher=algorithms.AES,
- mode=modes.GCM),
+ "NULL": CryptoAlgo("NULL", cipher=None, mode=None),
+ "AES-CBC": CryptoAlgo("AES-CBC", cipher=algorithms.AES, mode=modes.CBC),
+ "AES-GCM-16ICV": CryptoAlgo("AES-GCM-16ICV", cipher=algorithms.AES, mode=modes.GCM),
}
AUTH_ALGOS = {
- 'NULL': AuthAlgo('NULL', mac=None, mod=None, key_len=0, trunc_len=0),
- 'HMAC-SHA1-96': AuthAlgo('HMAC-SHA1-96', hmac.HMAC, hashes.SHA1, 20, 12),
- 'SHA2-256-128': AuthAlgo('SHA2-256-128', hmac.HMAC, hashes.SHA256, 32, 16),
- 'SHA2-384-192': AuthAlgo('SHA2-384-192', hmac.HMAC, hashes.SHA256, 48, 24),
- 'SHA2-512-256': AuthAlgo('SHA2-512-256', hmac.HMAC, hashes.SHA256, 64, 32),
+ "NULL": AuthAlgo("NULL", mac=None, mod=None, key_len=0, trunc_len=0),
+ "HMAC-SHA1-96": AuthAlgo("HMAC-SHA1-96", hmac.HMAC, hashes.SHA1, 20, 12),
+ "SHA2-256-128": AuthAlgo("SHA2-256-128", hmac.HMAC, hashes.SHA256, 32, 16),
+ "SHA2-384-192": AuthAlgo("SHA2-384-192", hmac.HMAC, hashes.SHA256, 48, 24),
+ "SHA2-512-256": AuthAlgo("SHA2-512-256", hmac.HMAC, hashes.SHA256, 64, 32),
}
PRF_ALGOS = {
- 'NULL': AuthAlgo('NULL', mac=None, mod=None, key_len=0, trunc_len=0),
- 'PRF_HMAC_SHA2_256': AuthAlgo('PRF_HMAC_SHA2_256', hmac.HMAC,
- hashes.SHA256, 32),
+ "NULL": AuthAlgo("NULL", mac=None, mod=None, key_len=0, trunc_len=0),
+ "PRF_HMAC_SHA2_256": AuthAlgo("PRF_HMAC_SHA2_256", hmac.HMAC, hashes.SHA256, 32),
}
CRYPTO_IDS = {
- 12: 'AES-CBC',
- 20: 'AES-GCM-16ICV',
+ 12: "AES-CBC",
+ 20: "AES-GCM-16ICV",
}
INTEG_IDS = {
- 2: 'HMAC-SHA1-96',
- 12: 'SHA2-256-128',
- 13: 'SHA2-384-192',
- 14: 'SHA2-512-256',
+ 2: "HMAC-SHA1-96",
+ 12: "SHA2-256-128",
+ 13: "SHA2-384-192",
+ 14: "SHA2-512-256",
}
@@ -182,11 +193,24 @@ class IKEv2ChildSA(object):
class IKEv2SA(object):
- def __init__(self, test, is_initiator=True, i_id=None, r_id=None,
- spi=b'\x01\x02\x03\x04\x05\x06\x07\x08', id_type='fqdn',
- nonce=None, auth_data=None, local_ts=None, remote_ts=None,
- auth_method='shared-key', priv_key=None, i_natt=False,
- r_natt=False, udp_encap=False):
+ def __init__(
+ self,
+ test,
+ is_initiator=True,
+ i_id=None,
+ r_id=None,
+ spi=b"\x01\x02\x03\x04\x05\x06\x07\x08",
+ id_type="fqdn",
+ nonce=None,
+ auth_data=None,
+ local_ts=None,
+ remote_ts=None,
+ auth_method="shared-key",
+ priv_key=None,
+ i_natt=False,
+ r_natt=False,
+ udp_encap=False,
+ ):
self.udp_encap = udp_encap
self.i_natt = i_natt
self.r_natt = r_natt
@@ -211,15 +235,14 @@ class IKEv2SA(object):
self.id_type = id_type
self.auth_method = auth_method
if self.is_initiator:
- self.rspi = 8 * b'\x00'
+ self.rspi = 8 * b"\x00"
self.ispi = spi
self.i_nonce = nonce
else:
self.rspi = spi
- self.ispi = 8 * b'\x00'
+ self.ispi = 8 * b"\x00"
self.r_nonce = nonce
- self.child_sas = [IKEv2ChildSA(local_ts, remote_ts,
- self.is_initiator)]
+ self.child_sas = [IKEv2ChildSA(local_ts, remote_ts, self.is_initiator)]
def new_msg_id(self):
self.msg_id += 1
@@ -245,13 +268,14 @@ class IKEv2SA(object):
priv = self.dh_private_key
peer = self.peer_dh_pub_key
p, g, l = self.ike_group
- return pow(int.from_bytes(peer, 'big'),
- int.from_bytes(priv, 'big'), p).to_bytes(l, 'big')
+ return pow(
+ int.from_bytes(peer, "big"), int.from_bytes(priv, "big"), p
+ ).to_bytes(l, "big")
def generate_dh_data(self):
# generate DH keys
if self.ike_dh not in DH:
- raise NotImplementedError('%s not in DH group' % self.ike_dh)
+ raise NotImplementedError("%s not in DH group" % self.ike_dh)
if self.dh_params is None:
dhg = DH[self.ike_dh]
@@ -261,13 +285,13 @@ class IKEv2SA(object):
priv = self.dh_params.generate_private_key()
pub = priv.public_key()
x = priv.private_numbers().x
- self.dh_private_key = x.to_bytes(priv.key_size // 8, 'big')
+ self.dh_private_key = x.to_bytes(priv.key_size // 8, "big")
y = pub.public_numbers().y
if self.is_initiator:
- self.i_dh_data = y.to_bytes(pub.key_size // 8, 'big')
+ self.i_dh_data = y.to_bytes(pub.key_size // 8, "big")
else:
- self.r_dh_data = y.to_bytes(pub.key_size // 8, 'big')
+ self.r_dh_data = y.to_bytes(pub.key_size // 8, "big")
def complete_dh_data(self):
self.dh_shared_secret = self.compute_secret()
@@ -281,41 +305,39 @@ class IKEv2SA(object):
integ_key_len = self.esp_integ_alg.key_len
salt_len = 0 if integ_key_len else 4
- l = (integ_key_len * 2 +
- encr_key_len * 2 +
- salt_len * 2)
+ l = integ_key_len * 2 + encr_key_len * 2 + salt_len * 2
keymat = self.calc_prfplus(prf, self.sk_d, s, l)
pos = 0
- c.sk_ei = keymat[pos:pos+encr_key_len]
+ c.sk_ei = keymat[pos : pos + encr_key_len]
pos += encr_key_len
if integ_key_len:
- c.sk_ai = keymat[pos:pos+integ_key_len]
+ c.sk_ai = keymat[pos : pos + integ_key_len]
pos += integ_key_len
else:
- c.salt_ei = keymat[pos:pos+salt_len]
+ c.salt_ei = keymat[pos : pos + salt_len]
pos += salt_len
- c.sk_er = keymat[pos:pos+encr_key_len]
+ c.sk_er = keymat[pos : pos + encr_key_len]
pos += encr_key_len
if integ_key_len:
- c.sk_ar = keymat[pos:pos+integ_key_len]
+ c.sk_ar = keymat[pos : pos + integ_key_len]
pos += integ_key_len
else:
- c.salt_er = keymat[pos:pos+salt_len]
+ c.salt_er = keymat[pos : pos + salt_len]
pos += salt_len
def calc_prfplus(self, prf, key, seed, length):
- r = b''
+ r = b""
t = None
x = 1
while len(r) < length and x < 255:
if t is not None:
s = t
else:
- s = b''
+ s = b""
s = s + seed + bytes([x])
t = self.calc_prf(prf, key, s)
r = r + t
@@ -348,30 +370,32 @@ class IKEv2SA(object):
else:
salt_size = 0
- l = (prf_key_trunc +
- integ_key_len * 2 +
- encr_key_len * 2 +
- tr_prf_key_len * 2 +
- salt_size * 2)
+ l = (
+ prf_key_trunc
+ + integ_key_len * 2
+ + encr_key_len * 2
+ + tr_prf_key_len * 2
+ + salt_size * 2
+ )
keymat = self.calc_prfplus(prf, self.skeyseed, s, l)
pos = 0
- self.sk_d = keymat[:pos+prf_key_trunc]
+ self.sk_d = keymat[: pos + prf_key_trunc]
pos += prf_key_trunc
- self.sk_ai = keymat[pos:pos+integ_key_len]
+ self.sk_ai = keymat[pos : pos + integ_key_len]
pos += integ_key_len
- self.sk_ar = keymat[pos:pos+integ_key_len]
+ self.sk_ar = keymat[pos : pos + integ_key_len]
pos += integ_key_len
- self.sk_ei = keymat[pos:pos+encr_key_len + salt_size]
+ 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]
+ 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]
+ self.sk_pi = keymat[pos : pos + tr_prf_key_len]
pos += tr_prf_key_len
- self.sk_pr = keymat[pos:pos+tr_prf_key_len]
+ self.sk_pr = keymat[pos : pos + tr_prf_key_len]
def generate_authmsg(self, prf, packet):
if self.is_initiator:
@@ -393,14 +417,15 @@ class IKEv2SA(object):
else:
packet = self.init_resp_packet
authmsg = self.generate_authmsg(prf, raw(packet))
- if self.auth_method == 'shared-key':
+ if self.auth_method == "shared-key":
psk = self.calc_prf(prf, self.auth_data, KEY_PAD)
self.auth_data = self.calc_prf(prf, psk, authmsg)
- elif self.auth_method == 'rsa-sig':
- self.auth_data = self.priv_key.sign(authmsg, padding.PKCS1v15(),
- hashes.SHA1())
+ elif self.auth_method == "rsa-sig":
+ self.auth_data = self.priv_key.sign(
+ authmsg, padding.PKCS1v15(), hashes.SHA1()
+ )
else:
- raise TypeError('unknown auth method type!')
+ raise TypeError("unknown auth method type!")
def encrypt(self, data, aad=None):
data = self.ike_crypto_alg.pad(data)
@@ -431,7 +456,7 @@ class IKEv2SA(object):
return self.sk_ei
def concat(self, alg, key_len):
- return alg + '-' + str(key_len * 8)
+ return alg + "-" + str(key_len * 8)
@property
def vpp_ike_cypto_alg(self):
@@ -445,8 +470,9 @@ class IKEv2SA(object):
integ_trunc = self.ike_integ_alg.trunc_len
exp_hmac = ikemsg[-integ_trunc:]
data = ikemsg[:-integ_trunc]
- computed_hmac = self.compute_hmac(self.ike_integ_alg.mod(),
- self.peer_authkey, data)
+ computed_hmac = self.compute_hmac(
+ self.ike_integ_alg.mod(), self.peer_authkey, data
+ )
self.test.assertEqual(computed_hmac[:integ_trunc], exp_hmac)
def compute_hmac(self, integ, key, data):
@@ -459,7 +485,7 @@ class IKEv2SA(object):
def hmac_and_decrypt(self, ike):
ep = ike[ikev2.IKEv2_payload_Encrypted]
- if self.ike_crypto == 'AES-GCM-16ICV':
+ 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:]
@@ -473,26 +499,26 @@ class IKEv2SA(object):
plain = self.decrypt(ct)
# remove padding
pad_len = plain[-1]
- return plain[:-pad_len - 1]
+ return plain[: -pad_len - 1]
def build_ts_addr(self, ts, version):
- return {'starting_address_v' + version: ts['start_addr'],
- 'ending_address_v' + version: ts['end_addr']}
+ return {
+ "starting_address_v" + version: ts["start_addr"],
+ "ending_address_v" + version: ts["end_addr"],
+ }
def generate_ts(self, is_ip4):
c = self.child_sas[0]
- ts_data = {'IP_protocol_ID': 0,
- 'start_port': 0,
- 'end_port': 0xffff}
+ ts_data = {"IP_protocol_ID": 0, "start_port": 0, "end_port": 0xFFFF}
if is_ip4:
- ts_data.update(self.build_ts_addr(c.local_ts, '4'))
+ ts_data.update(self.build_ts_addr(c.local_ts, "4"))
ts1 = ikev2.IPv4TrafficSelector(**ts_data)
- ts_data.update(self.build_ts_addr(c.remote_ts, '4'))
+ ts_data.update(self.build_ts_addr(c.remote_ts, "4"))
ts2 = ikev2.IPv4TrafficSelector(**ts_data)
else:
- ts_data.update(self.build_ts_addr(c.local_ts, '6'))
+ ts_data.update(self.build_ts_addr(c.local_ts, "6"))
ts1 = ikev2.IPv6TrafficSelector(**ts_data)
- ts_data.update(self.build_ts_addr(c.remote_ts, '6'))
+ ts_data.update(self.build_ts_addr(c.remote_ts, "6"))
ts2 = ikev2.IPv6TrafficSelector(**ts_data)
if self.is_initiator:
@@ -501,18 +527,18 @@ class IKEv2SA(object):
def set_ike_props(self, crypto, crypto_key_len, integ, prf, dh):
if crypto not in CRYPTO_ALGOS:
- raise TypeError('unsupported encryption algo %r' % crypto)
+ raise TypeError("unsupported encryption algo %r" % crypto)
self.ike_crypto = crypto
self.ike_crypto_alg = CRYPTO_ALGOS[crypto]
self.ike_crypto_key_len = crypto_key_len
if integ not in AUTH_ALGOS:
- raise TypeError('unsupported auth algo %r' % integ)
- self.ike_integ = None if integ == 'NULL' else integ
+ raise TypeError("unsupported auth algo %r" % integ)
+ self.ike_integ = None if integ == "NULL" else integ
self.ike_integ_alg = AUTH_ALGOS[integ]
if prf not in PRF_ALGOS:
- raise TypeError('unsupported prf algo %r' % prf)
+ raise TypeError("unsupported prf algo %r" % prf)
self.ike_prf = prf
self.ike_prf_alg = PRF_ALGOS[prf]
self.ike_dh = dh
@@ -521,20 +547,20 @@ class IKEv2SA(object):
def set_esp_props(self, crypto, crypto_key_len, integ):
self.esp_crypto_key_len = crypto_key_len
if crypto not in CRYPTO_ALGOS:
- raise TypeError('unsupported encryption algo %r' % crypto)
+ raise TypeError("unsupported encryption algo %r" % crypto)
self.esp_crypto = crypto
self.esp_crypto_alg = CRYPTO_ALGOS[crypto]
if integ not in AUTH_ALGOS:
- raise TypeError('unsupported auth algo %r' % integ)
- self.esp_integ = None if integ == 'NULL' else integ
+ raise TypeError("unsupported auth algo %r" % integ)
+ self.esp_integ = None if integ == "NULL" else integ
self.esp_integ_alg = AUTH_ALGOS[integ]
def crypto_attr(self, key_len):
- if self.ike_crypto in ['AES-CBC', 'AES-GCM-16ICV']:
- return (0x800e << 16 | key_len << 3, 12)
+ if self.ike_crypto in ["AES-CBC", "AES-GCM-16ICV"]:
+ return (0x800E << 16 | key_len << 3, 12)
else:
- raise Exception('unsupported attribute type')
+ raise Exception("unsupported attribute type")
def ike_crypto_attr(self):
return self.crypto_attr(self.ike_crypto_key_len)
@@ -545,19 +571,20 @@ class IKEv2SA(object):
def compute_nat_sha1(self, ip, port, rspi=None):
if rspi is None:
rspi = self.rspi
- data = self.ispi + rspi + ip + (port).to_bytes(2, 'big')
+ data = self.ispi + rspi + ip + (port).to_bytes(2, "big")
digest = hashes.Hash(hashes.SHA1(), backend=default_backend())
digest.update(data)
return digest.finalize()
class IkePeer(VppTestCase):
- """ common class for initiator and responder """
+ """common class for initiator and responder"""
@classmethod
def setUpClass(cls):
import scapy.contrib.ikev2 as _ikev2
- globals()['ikev2'] = _ikev2
+
+ globals()["ikev2"] = _ikev2
super(IkePeer, cls).setUpClass()
cls.create_pg_interfaces(range(2))
for i in cls.pg_interfaces:
@@ -591,36 +618,46 @@ class IkePeer(VppTestCase):
self.assertIsNotNone(self.p.query_vpp_config())
if self.sa.is_initiator:
self.sa.generate_dh_data()
- self.vapi.cli('ikev2 set logging level 4')
- self.vapi.cli('event-lo clear')
+ self.vapi.cli("ikev2 set logging level 4")
+ self.vapi.cli("event-lo clear")
- def assert_counter(self, count, name, version='ip4'):
- node_name = '/err/ikev2-%s/' % version + name
+ def assert_counter(self, count, name, version="ip4"):
+ node_name = "/err/ikev2-%s/" % version + name
self.assertEqual(count, self.statistics.get_err_counter(node_name))
def create_rekey_request(self):
sa, first_payload = self.generate_auth_payload(is_rekey=True)
header = ikev2.IKEv2(
- init_SPI=self.sa.ispi,
- resp_SPI=self.sa.rspi, id=self.sa.new_msg_id(),
- flags='Initiator', exch_type='CREATE_CHILD_SA')
+ init_SPI=self.sa.ispi,
+ resp_SPI=self.sa.rspi,
+ id=self.sa.new_msg_id(),
+ flags="Initiator",
+ exch_type="CREATE_CHILD_SA",
+ )
ike_msg = self.encrypt_ike_msg(header, sa, first_payload)
- return self.create_packet(self.pg0, ike_msg, self.sa.sport,
- self.sa.dport, self.sa.natt, self.ip6)
+ return self.create_packet(
+ self.pg0, ike_msg, self.sa.sport, self.sa.dport, self.sa.natt, self.ip6
+ )
def create_empty_request(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)
- return self.create_packet(self.pg0, msg, self.sa.sport,
- self.sa.dport, self.sa.natt, self.ip6)
-
- def create_packet(self, src_if, msg, sport=500, dport=500, natt=False,
- use_ip6=False):
+ 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)
+ return self.create_packet(
+ self.pg0, msg, self.sa.sport, self.sa.dport, self.sa.natt, self.ip6
+ )
+
+ def create_packet(
+ self, src_if, msg, sport=500, dport=500, natt=False, use_ip6=False
+ ):
if use_ip6:
src_ip = src_if.remote_ip6
dst_ip = src_if.local_ip6
@@ -629,12 +666,14 @@ class IkePeer(VppTestCase):
src_ip = src_if.remote_ip4
dst_ip = src_if.local_ip4
ip_layer = IP
- res = (Ether(dst=src_if.local_mac, src=src_if.remote_mac) /
- ip_layer(src=src_ip, dst=dst_ip) /
- UDP(sport=sport, dport=dport))
+ res = (
+ Ether(dst=src_if.local_mac, src=src_if.remote_mac)
+ / ip_layer(src=src_ip, dst=dst_ip)
+ / UDP(sport=sport, dport=dport)
+ )
if natt:
# insert non ESP marker
- res = res / Raw(b'\x00' * 4)
+ res = res / Raw(b"\x00" * 4)
return res / msg
def verify_udp(self, udp):
@@ -651,7 +690,7 @@ class IkePeer(VppTestCase):
esp = packet[ESP]
ih = self.verify_and_remove_non_esp_marker(esp)
self.assertEqual(ih.version, 0x20)
- self.assertNotIn('Version', ih.flags)
+ self.assertNotIn("Version", ih.flags)
return ih
def verify_and_remove_non_esp_marker(self, packet):
@@ -659,26 +698,32 @@ class IkePeer(VppTestCase):
# if we are in nat traversal mode check for non esp marker
# and remove it
data = raw(packet)
- self.assertEqual(data[:4], b'\x00' * 4)
+ self.assertEqual(data[:4], b"\x00" * 4)
return ikev2.IKEv2(data[4:])
else:
return packet
def encrypt_ike_msg(self, header, plain, first_payload):
- if self.sa.ike_crypto == 'AES-GCM-16ICV':
+ 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())
+ 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=first_payload,
- length=plen)
+ sk_p = ikev2.IKEv2_payload_Encrypted(
+ next_payload=first_payload, length=plen
+ )
header.length = tlen
res = header / sk_p
encr = self.sa.encrypt(raw(plain), raw(res))
- sk_p = ikev2.IKEv2_payload_Encrypted(next_payload=first_payload,
- length=plen, load=encr)
+ sk_p = ikev2.IKEv2_payload_Encrypted(
+ next_payload=first_payload, length=plen, load=encr
+ )
res = header / sk_p
else:
encr = self.sa.encrypt(raw(plain))
@@ -686,16 +731,18 @@ class IkePeer(VppTestCase):
plen = len(encr) + len(ikev2.IKEv2_payload_Encrypted()) + trunc_len
tlen = plen + len(ikev2.IKEv2())
- sk_p = ikev2.IKEv2_payload_Encrypted(next_payload=first_payload,
- length=plen, load=encr)
+ sk_p = ikev2.IKEv2_payload_Encrypted(
+ next_payload=first_payload, length=plen, load=encr
+ )
header.length = tlen
res = header / sk_p
integ_data = raw(res)
- hmac_data = self.sa.compute_hmac(self.sa.ike_integ_alg.mod(),
- self.sa.my_authkey, integ_data)
+ hmac_data = self.sa.compute_hmac(
+ self.sa.ike_integ_alg.mod(), self.sa.my_authkey, integ_data
+ )
res = res / Raw(hmac_data[:trunc_len])
- assert(len(res) == tlen)
+ assert len(res) == tlen
return res
def verify_udp_encap(self, ipsec_sa):
@@ -747,37 +794,37 @@ class IkePeer(VppTestCase):
# verify crypto keys
self.assertEqual(sa0.crypto_key.length, len(c.sk_er))
self.assertEqual(sa1.crypto_key.length, len(c.sk_ei))
- self.assertEqual(sa0.crypto_key.data[:len(c.sk_er)], c.sk_er)
- self.assertEqual(sa1.crypto_key.data[:len(c.sk_ei)], c.sk_ei)
+ self.assertEqual(sa0.crypto_key.data[: len(c.sk_er)], c.sk_er)
+ self.assertEqual(sa1.crypto_key.data[: len(c.sk_ei)], c.sk_ei)
# verify integ keys
if vpp_integ_alg:
self.assertEqual(sa0.integrity_key.length, len(c.sk_ar))
self.assertEqual(sa1.integrity_key.length, len(c.sk_ai))
- self.assertEqual(sa0.integrity_key.data[:len(c.sk_ar)], c.sk_ar)
- self.assertEqual(sa1.integrity_key.data[:len(c.sk_ai)], c.sk_ai)
+ self.assertEqual(sa0.integrity_key.data[: len(c.sk_ar)], c.sk_ar)
+ self.assertEqual(sa1.integrity_key.data[: len(c.sk_ai)], c.sk_ai)
else:
- self.assertEqual(sa0.salt.to_bytes(4, 'little'), c.salt_er)
- self.assertEqual(sa1.salt.to_bytes(4, 'little'), c.salt_ei)
+ self.assertEqual(sa0.salt.to_bytes(4, "little"), c.salt_er)
+ self.assertEqual(sa1.salt.to_bytes(4, "little"), c.salt_ei)
def verify_keymat(self, api_keys, keys, name):
km = getattr(keys, name)
api_km = getattr(api_keys, name)
- api_km_len = getattr(api_keys, name + '_len')
+ api_km_len = getattr(api_keys, name + "_len")
self.assertEqual(len(km), api_km_len)
self.assertEqual(km, api_km[:api_km_len])
def verify_id(self, api_id, exp_id):
self.assertEqual(api_id.type, IDType.value(exp_id.type))
self.assertEqual(api_id.data_len, exp_id.data_len)
- self.assertEqual(bytes(api_id.data, 'ascii'), exp_id.type)
+ self.assertEqual(bytes(api_id.data, "ascii"), exp_id.type)
def verify_ike_sas(self):
r = self.vapi.ikev2_sa_dump()
self.assertEqual(len(r), 1)
sa = r[0].sa
- self.assertEqual(self.sa.ispi, (sa.ispi).to_bytes(8, 'big'))
- self.assertEqual(self.sa.rspi, (sa.rspi).to_bytes(8, 'big'))
+ self.assertEqual(self.sa.ispi, (sa.ispi).to_bytes(8, "big"))
+ self.assertEqual(self.sa.rspi, (sa.rspi).to_bytes(8, "big"))
if self.ip6:
if self.sa.is_initiator:
self.assertEqual(sa.iaddr, IPv6Address(self.pg0.remote_ip6))
@@ -792,55 +839,53 @@ class IkePeer(VppTestCase):
else:
self.assertEqual(sa.iaddr, IPv4Address(self.pg0.local_ip4))
self.assertEqual(sa.raddr, IPv4Address(self.pg0.remote_ip4))
- self.verify_keymat(sa.keys, self.sa, 'sk_d')
- self.verify_keymat(sa.keys, self.sa, 'sk_ai')
- self.verify_keymat(sa.keys, self.sa, 'sk_ar')
- self.verify_keymat(sa.keys, self.sa, 'sk_ei')
- self.verify_keymat(sa.keys, self.sa, 'sk_er')
- self.verify_keymat(sa.keys, self.sa, 'sk_pi')
- self.verify_keymat(sa.keys, self.sa, 'sk_pr')
+ self.verify_keymat(sa.keys, self.sa, "sk_d")
+ self.verify_keymat(sa.keys, self.sa, "sk_ai")
+ self.verify_keymat(sa.keys, self.sa, "sk_ar")
+ self.verify_keymat(sa.keys, self.sa, "sk_ei")
+ self.verify_keymat(sa.keys, self.sa, "sk_er")
+ self.verify_keymat(sa.keys, self.sa, "sk_pi")
+ self.verify_keymat(sa.keys, self.sa, "sk_pr")
self.assertEqual(sa.i_id.type, self.sa.id_type)
self.assertEqual(sa.r_id.type, self.sa.id_type)
self.assertEqual(sa.i_id.data_len, len(self.sa.i_id))
self.assertEqual(sa.r_id.data_len, len(self.idr))
- self.assertEqual(bytes(sa.i_id.data, 'ascii'), self.sa.i_id)
- self.assertEqual(bytes(sa.r_id.data, 'ascii'), self.idr)
+ self.assertEqual(bytes(sa.i_id.data, "ascii"), self.sa.i_id)
+ self.assertEqual(bytes(sa.r_id.data, "ascii"), self.idr)
r = self.vapi.ikev2_child_sa_dump(sa_index=sa.sa_index)
self.assertEqual(len(r), 1)
csa = r[0].child_sa
self.assertEqual(csa.sa_index, sa.sa_index)
c = self.sa.child_sas[0]
- if hasattr(c, 'sk_ai'):
- self.verify_keymat(csa.keys, c, 'sk_ai')
- self.verify_keymat(csa.keys, c, 'sk_ar')
- self.verify_keymat(csa.keys, c, 'sk_ei')
- self.verify_keymat(csa.keys, c, 'sk_er')
- self.assertEqual(csa.i_spi.to_bytes(4, 'big'), c.ispi)
- self.assertEqual(csa.r_spi.to_bytes(4, 'big'), c.rspi)
+ if hasattr(c, "sk_ai"):
+ self.verify_keymat(csa.keys, c, "sk_ai")
+ self.verify_keymat(csa.keys, c, "sk_ar")
+ self.verify_keymat(csa.keys, c, "sk_ei")
+ self.verify_keymat(csa.keys, c, "sk_er")
+ self.assertEqual(csa.i_spi.to_bytes(4, "big"), c.ispi)
+ self.assertEqual(csa.r_spi.to_bytes(4, "big"), c.rspi)
tsi, tsr = self.sa.generate_ts(self.p.ts_is_ip4)
tsi = tsi[0]
tsr = tsr[0]
r = self.vapi.ikev2_traffic_selector_dump(
- is_initiator=True, sa_index=sa.sa_index,
- child_sa_index=csa.child_sa_index)
+ is_initiator=True, sa_index=sa.sa_index, child_sa_index=csa.child_sa_index
+ )
self.assertEqual(len(r), 1)
ts = r[0].ts
self.verify_ts(r[0].ts, tsi[0], True)
r = self.vapi.ikev2_traffic_selector_dump(
- is_initiator=False, sa_index=sa.sa_index,
- child_sa_index=csa.child_sa_index)
+ is_initiator=False, sa_index=sa.sa_index, child_sa_index=csa.child_sa_index
+ )
self.assertEqual(len(r), 1)
self.verify_ts(r[0].ts, tsr[0], False)
- n = self.vapi.ikev2_nonce_get(is_initiator=True,
- sa_index=sa.sa_index)
+ n = self.vapi.ikev2_nonce_get(is_initiator=True, sa_index=sa.sa_index)
self.verify_nonce(n, self.sa.i_nonce)
- n = self.vapi.ikev2_nonce_get(is_initiator=False,
- sa_index=sa.sa_index)
+ n = self.vapi.ikev2_nonce_get(is_initiator=False, sa_index=sa.sa_index)
self.verify_nonce(n, self.sa.r_nonce)
def verify_nonce(self, api_nonce, nonce):
@@ -854,61 +899,65 @@ class IkePeer(VppTestCase):
self.assertFalse(api_ts.is_local)
if self.p.ts_is_ip4:
- self.assertEqual(api_ts.start_addr,
- IPv4Address(ts.starting_address_v4))
- self.assertEqual(api_ts.end_addr,
- IPv4Address(ts.ending_address_v4))
+ self.assertEqual(api_ts.start_addr, IPv4Address(ts.starting_address_v4))
+ self.assertEqual(api_ts.end_addr, IPv4Address(ts.ending_address_v4))
else:
- self.assertEqual(api_ts.start_addr,
- IPv6Address(ts.starting_address_v6))
- self.assertEqual(api_ts.end_addr,
- IPv6Address(ts.ending_address_v6))
+ self.assertEqual(api_ts.start_addr, IPv6Address(ts.starting_address_v6))
+ self.assertEqual(api_ts.end_addr, IPv6Address(ts.ending_address_v6))
self.assertEqual(api_ts.start_port, ts.start_port)
self.assertEqual(api_ts.end_port, ts.end_port)
self.assertEqual(api_ts.protocol_id, ts.IP_protocol_ID)
class TemplateInitiator(IkePeer):
- """ initiator test template """
+ """initiator test template"""
def initiate_del_sa_from_initiator(self):
- ispi = int.from_bytes(self.sa.ispi, 'little')
+ ispi = int.from_bytes(self.sa.ispi, "little")
self.pg0.enable_capture()
self.pg_start()
self.vapi.ikev2_initiate_del_ike_sa(ispi=ispi)
capture = self.pg0.get_capture(1)
ih = self.get_ike_header(capture[0])
- self.assertNotIn('Response', ih.flags)
- self.assertIn('Initiator', ih.flags)
+ self.assertNotIn("Response", ih.flags)
+ self.assertIn("Initiator", ih.flags)
self.assertEqual(ih.init_SPI, self.sa.ispi)
self.assertEqual(ih.resp_SPI, self.sa.rspi)
plain = self.sa.hmac_and_decrypt(ih)
d = ikev2.IKEv2_payload_Delete(plain)
self.assertEqual(d.proto, 1) # proto=IKEv2
- header = ikev2.IKEv2(init_SPI=self.sa.ispi, resp_SPI=self.sa.rspi,
- flags='Response', exch_type='INFORMATIONAL',
- id=ih.id, next_payload='Encrypted')
- resp = self.encrypt_ike_msg(header, b'', None)
+ header = ikev2.IKEv2(
+ init_SPI=self.sa.ispi,
+ resp_SPI=self.sa.rspi,
+ flags="Response",
+ exch_type="INFORMATIONAL",
+ id=ih.id,
+ next_payload="Encrypted",
+ )
+ resp = self.encrypt_ike_msg(header, b"", None)
self.send_and_assert_no_replies(self.pg0, resp)
def verify_del_sa(self, packet):
ih = self.get_ike_header(packet)
self.assertEqual(ih.id, self.sa.msg_id)
self.assertEqual(ih.exch_type, 37) # exchange informational
- self.assertIn('Response', ih.flags)
- self.assertIn('Initiator', ih.flags)
+ self.assertIn("Response", ih.flags)
+ self.assertIn("Initiator", ih.flags)
plain = self.sa.hmac_and_decrypt(ih)
- self.assertEqual(plain, b'')
+ self.assertEqual(plain, b"")
def initiate_del_sa_from_responder(self):
- header = ikev2.IKEv2(init_SPI=self.sa.ispi, resp_SPI=self.sa.rspi,
- exch_type='INFORMATIONAL',
- id=self.sa.new_msg_id())
- del_sa = ikev2.IKEv2_payload_Delete(proto='IKEv2')
- ike_msg = self.encrypt_ike_msg(header, del_sa, 'Delete')
- packet = self.create_packet(self.pg0, ike_msg,
- self.sa.sport, self.sa.dport,
- self.sa.natt, self.ip6)
+ header = ikev2.IKEv2(
+ init_SPI=self.sa.ispi,
+ resp_SPI=self.sa.rspi,
+ exch_type="INFORMATIONAL",
+ id=self.sa.new_msg_id(),
+ )
+ del_sa = ikev2.IKEv2_payload_Delete(proto="IKEv2")
+ ike_msg = self.encrypt_ike_msg(header, del_sa, "Delete")
+ packet = self.create_packet(
+ self.pg0, ike_msg, self.sa.sport, self.sa.dport, self.sa.natt, self.ip6
+ )
self.pg0.add_stream(packet)
self.pg0.enable_capture()
self.pg_start()
@@ -935,26 +984,28 @@ class TemplateInitiator(IkePeer):
s = self.find_notify_payload(packet, 16388)
self.assertIsNotNone(s)
src_sha = self.sa.compute_nat_sha1(
- inet_pton(socket.AF_INET, iph.src), udp.sport, b'\x00' * 8)
+ inet_pton(socket.AF_INET, iph.src), udp.sport, b"\x00" * 8
+ )
self.assertEqual(s.load, src_sha)
# NAT_DETECTION_DESTINATION_IP
s = self.find_notify_payload(packet, 16389)
self.assertIsNotNone(s)
dst_sha = self.sa.compute_nat_sha1(
- inet_pton(socket.AF_INET, iph.dst), udp.dport, b'\x00' * 8)
+ inet_pton(socket.AF_INET, iph.dst), udp.dport, b"\x00" * 8
+ )
self.assertEqual(s.load, dst_sha)
def verify_sa_init_request(self, packet):
udp = packet[UDP]
self.sa.dport = udp.sport
ih = packet[ikev2.IKEv2]
- self.assertNotEqual(ih.init_SPI, 8 * b'\x00')
+ self.assertNotEqual(ih.init_SPI, 8 * b"\x00")
self.assertEqual(ih.exch_type, 34) # SA_INIT
self.sa.ispi = ih.init_SPI
- self.assertEqual(ih.resp_SPI, 8 * b'\x00')
- self.assertIn('Initiator', ih.flags)
- self.assertNotIn('Response', ih.flags)
+ self.assertEqual(ih.resp_SPI, 8 * b"\x00")
+ self.assertIn("Initiator", ih.flags)
+ self.assertNotIn("Response", ih.flags)
self.sa.i_nonce = ih[ikev2.IKEv2_payload_Nonce].load
self.sa.i_dh_data = ih[ikev2.IKEv2_payload_KE].load
@@ -962,20 +1013,23 @@ class TemplateInitiator(IkePeer):
self.assertEqual(prop.proto, 1) # proto = ikev2
self.assertEqual(prop.proposal, 1)
self.assertEqual(prop.trans[0].transform_type, 1) # encryption
- self.assertEqual(prop.trans[0].transform_id,
- self.p.ike_transforms['crypto_alg'])
+ self.assertEqual(
+ prop.trans[0].transform_id, self.p.ike_transforms["crypto_alg"]
+ )
self.assertEqual(prop.trans[1].transform_type, 2) # prf
self.assertEqual(prop.trans[1].transform_id, 5) # "hmac-sha2-256"
self.assertEqual(prop.trans[2].transform_type, 4) # dh
- self.assertEqual(prop.trans[2].transform_id,
- self.p.ike_transforms['dh_group'])
+ self.assertEqual(prop.trans[2].transform_id, self.p.ike_transforms["dh_group"])
self.verify_nat_detection(packet)
self.sa.set_ike_props(
- crypto='AES-GCM-16ICV', crypto_key_len=32,
- integ='NULL', prf='PRF_HMAC_SHA2_256', dh='3072MODPgr')
- self.sa.set_esp_props(crypto='AES-CBC', crypto_key_len=32,
- integ='SHA2-256-128')
+ crypto="AES-GCM-16ICV",
+ crypto_key_len=32,
+ integ="NULL",
+ prf="PRF_HMAC_SHA2_256",
+ dh="3072MODPgr",
+ )
+ self.sa.set_esp_props(crypto="AES-CBC", crypto_key_len=32, integ="SHA2-256-128")
self.sa.generate_dh_data()
self.sa.complete_dh_data()
self.sa.calc_keys()
@@ -995,8 +1049,8 @@ class TemplateInitiator(IkePeer):
self.assertEqual(ih.resp_SPI, self.sa.rspi)
self.assertEqual(ih.init_SPI, self.sa.ispi)
self.assertEqual(ih.exch_type, 35) # IKE_AUTH
- self.assertIn('Initiator', ih.flags)
- self.assertNotIn('Response', ih.flags)
+ self.assertIn("Initiator", ih.flags)
+ self.assertNotIn("Response", ih.flags)
udp = packet[UDP]
self.verify_udp(udp)
@@ -1013,48 +1067,67 @@ class TemplateInitiator(IkePeer):
prop = idi[ikev2.IKEv2_payload_Proposal]
c = self.sa.child_sas[0]
c.ispi = prop.SPI
- self.update_esp_transforms(
- prop[ikev2.IKEv2_payload_Transform], self.sa)
+ self.update_esp_transforms(prop[ikev2.IKEv2_payload_Transform], self.sa)
def send_init_response(self):
tr_attr = self.sa.ike_crypto_attr()
- trans = (ikev2.IKEv2_payload_Transform(transform_type='Encryption',
- transform_id=self.sa.ike_crypto, length=tr_attr[1],
- key_length=tr_attr[0]) /
- ikev2.IKEv2_payload_Transform(transform_type='Integrity',
- transform_id=self.sa.ike_integ) /
- ikev2.IKEv2_payload_Transform(transform_type='PRF',
- transform_id=self.sa.ike_prf_alg.name) /
- ikev2.IKEv2_payload_Transform(transform_type='GroupDesc',
- transform_id=self.sa.ike_dh))
- props = (ikev2.IKEv2_payload_Proposal(proposal=1, proto='IKEv2',
- trans_nb=4, trans=trans))
+ trans = (
+ ikev2.IKEv2_payload_Transform(
+ transform_type="Encryption",
+ transform_id=self.sa.ike_crypto,
+ length=tr_attr[1],
+ key_length=tr_attr[0],
+ )
+ / ikev2.IKEv2_payload_Transform(
+ transform_type="Integrity", transform_id=self.sa.ike_integ
+ )
+ / ikev2.IKEv2_payload_Transform(
+ transform_type="PRF", transform_id=self.sa.ike_prf_alg.name
+ )
+ / ikev2.IKEv2_payload_Transform(
+ transform_type="GroupDesc", transform_id=self.sa.ike_dh
+ )
+ )
+ props = ikev2.IKEv2_payload_Proposal(
+ proposal=1, proto="IKEv2", trans_nb=4, trans=trans
+ )
src_address = inet_pton(socket.AF_INET, self.pg0.remote_ip4)
if self.sa.natt:
- dst_address = b'\x0a\x0a\x0a\x0a'
+ dst_address = b"\x0a\x0a\x0a\x0a"
else:
dst_address = inet_pton(socket.AF_INET, self.pg0.local_ip4)
src_nat = self.sa.compute_nat_sha1(src_address, self.sa.sport)
dst_nat = self.sa.compute_nat_sha1(dst_address, self.sa.dport)
self.sa.init_resp_packet = (
- ikev2.IKEv2(init_SPI=self.sa.ispi, resp_SPI=self.sa.rspi,
- exch_type='IKE_SA_INIT', flags='Response') /
- ikev2.IKEv2_payload_SA(next_payload='KE', prop=props) /
- ikev2.IKEv2_payload_KE(next_payload='Nonce',
- group=self.sa.ike_dh,
- load=self.sa.my_dh_pub_key) /
- ikev2.IKEv2_payload_Nonce(load=self.sa.r_nonce,
- next_payload='Notify') /
- ikev2.IKEv2_payload_Notify(
- type='NAT_DETECTION_SOURCE_IP', load=src_nat,
- next_payload='Notify') / ikev2.IKEv2_payload_Notify(
- type='NAT_DETECTION_DESTINATION_IP', load=dst_nat))
-
- ike_msg = self.create_packet(self.pg0, self.sa.init_resp_packet,
- self.sa.sport, self.sa.dport,
- False, self.ip6)
+ ikev2.IKEv2(
+ init_SPI=self.sa.ispi,
+ resp_SPI=self.sa.rspi,
+ exch_type="IKE_SA_INIT",
+ flags="Response",
+ )
+ / ikev2.IKEv2_payload_SA(next_payload="KE", prop=props)
+ / ikev2.IKEv2_payload_KE(
+ next_payload="Nonce", group=self.sa.ike_dh, load=self.sa.my_dh_pub_key
+ )
+ / ikev2.IKEv2_payload_Nonce(load=self.sa.r_nonce, next_payload="Notify")
+ / ikev2.IKEv2_payload_Notify(
+ type="NAT_DETECTION_SOURCE_IP", load=src_nat, next_payload="Notify"
+ )
+ / ikev2.IKEv2_payload_Notify(
+ type="NAT_DETECTION_DESTINATION_IP", load=dst_nat
+ )
+ )
+
+ ike_msg = self.create_packet(
+ self.pg0,
+ self.sa.init_resp_packet,
+ self.sa.sport,
+ self.sa.dport,
+ False,
+ self.ip6,
+ )
self.pg_send(self.pg0, ike_msg)
capture = self.pg0.get_capture(1)
self.verify_sa_auth_req(capture[0])
@@ -1070,47 +1143,64 @@ class TemplateInitiator(IkePeer):
def send_auth_response(self):
tr_attr = self.sa.esp_crypto_attr()
- trans = (ikev2.IKEv2_payload_Transform(transform_type='Encryption',
- transform_id=self.sa.esp_crypto, length=tr_attr[1],
- key_length=tr_attr[0]) /
- ikev2.IKEv2_payload_Transform(transform_type='Integrity',
- transform_id=self.sa.esp_integ) /
- ikev2.IKEv2_payload_Transform(
- transform_type='Extended Sequence Number',
- transform_id='No ESN') /
- ikev2.IKEv2_payload_Transform(
- transform_type='Extended Sequence Number',
- transform_id='ESN'))
+ trans = (
+ ikev2.IKEv2_payload_Transform(
+ transform_type="Encryption",
+ transform_id=self.sa.esp_crypto,
+ length=tr_attr[1],
+ key_length=tr_attr[0],
+ )
+ / ikev2.IKEv2_payload_Transform(
+ transform_type="Integrity", transform_id=self.sa.esp_integ
+ )
+ / ikev2.IKEv2_payload_Transform(
+ transform_type="Extended Sequence Number", transform_id="No ESN"
+ )
+ / ikev2.IKEv2_payload_Transform(
+ transform_type="Extended Sequence Number", transform_id="ESN"
+ )
+ )
c = self.sa.child_sas[0]
- props = (ikev2.IKEv2_payload_Proposal(proposal=1, proto='ESP',
- SPIsize=4, SPI=c.rspi, trans_nb=4, trans=trans))
+ props = ikev2.IKEv2_payload_Proposal(
+ proposal=1, proto="ESP", SPIsize=4, SPI=c.rspi, trans_nb=4, trans=trans
+ )
tsi, tsr = self.sa.generate_ts(self.p.ts_is_ip4)
- plain = (ikev2.IKEv2_payload_IDi(next_payload='IDr',
- IDtype=self.sa.id_type, load=self.sa.i_id) /
- ikev2.IKEv2_payload_IDr(next_payload='AUTH',
- IDtype=self.sa.id_type, load=self.sa.r_id) /
- ikev2.IKEv2_payload_AUTH(next_payload='SA',
- auth_type=AuthMethod.value(self.sa.auth_method),
- load=self.sa.auth_data) /
- ikev2.IKEv2_payload_SA(next_payload='TSi', prop=props) /
- ikev2.IKEv2_payload_TSi(next_payload='TSr',
- number_of_TSs=len(tsi),
- traffic_selector=tsi) /
- ikev2.IKEv2_payload_TSr(next_payload='Notify',
- number_of_TSs=len(tsr),
- traffic_selector=tsr) /
- ikev2.IKEv2_payload_Notify(type='INITIAL_CONTACT'))
+ plain = (
+ ikev2.IKEv2_payload_IDi(
+ next_payload="IDr", IDtype=self.sa.id_type, load=self.sa.i_id
+ )
+ / ikev2.IKEv2_payload_IDr(
+ next_payload="AUTH", IDtype=self.sa.id_type, load=self.sa.r_id
+ )
+ / ikev2.IKEv2_payload_AUTH(
+ next_payload="SA",
+ auth_type=AuthMethod.value(self.sa.auth_method),
+ load=self.sa.auth_data,
+ )
+ / ikev2.IKEv2_payload_SA(next_payload="TSi", prop=props)
+ / ikev2.IKEv2_payload_TSi(
+ next_payload="TSr", number_of_TSs=len(tsi), traffic_selector=tsi
+ )
+ / ikev2.IKEv2_payload_TSr(
+ next_payload="Notify", number_of_TSs=len(tsr), traffic_selector=tsr
+ )
+ / ikev2.IKEv2_payload_Notify(type="INITIAL_CONTACT")
+ )
header = ikev2.IKEv2(
- init_SPI=self.sa.ispi,
- resp_SPI=self.sa.rspi, id=self.sa.new_msg_id(),
- flags='Response', exch_type='IKE_AUTH')
-
- ike_msg = self.encrypt_ike_msg(header, plain, 'IDi')
- packet = self.create_packet(self.pg0, ike_msg, self.sa.sport,
- self.sa.dport, self.sa.natt, self.ip6)
+ init_SPI=self.sa.ispi,
+ resp_SPI=self.sa.rspi,
+ id=self.sa.new_msg_id(),
+ flags="Response",
+ exch_type="IKE_AUTH",
+ )
+
+ ike_msg = self.encrypt_ike_msg(header, plain, "IDi")
+ packet = self.create_packet(
+ self.pg0, ike_msg, self.sa.sport, self.sa.dport, self.sa.natt, self.ip6
+ )
self.pg_send(self.pg0, packet)
def test_initiator(self):
@@ -1122,51 +1212,58 @@ class TemplateInitiator(IkePeer):
class TemplateResponder(IkePeer):
- """ responder test template """
+ """responder test template"""
def initiate_del_sa_from_responder(self):
self.pg0.enable_capture()
self.pg_start()
- self.vapi.ikev2_initiate_del_ike_sa(
- ispi=int.from_bytes(self.sa.ispi, 'little'))
+ self.vapi.ikev2_initiate_del_ike_sa(ispi=int.from_bytes(self.sa.ispi, "little"))
capture = self.pg0.get_capture(1)
ih = self.get_ike_header(capture[0])
- self.assertNotIn('Response', ih.flags)
- self.assertNotIn('Initiator', ih.flags)
+ self.assertNotIn("Response", ih.flags)
+ self.assertNotIn("Initiator", ih.flags)
self.assertEqual(ih.exch_type, 37) # INFORMATIONAL
plain = self.sa.hmac_and_decrypt(ih)
d = ikev2.IKEv2_payload_Delete(plain)
self.assertEqual(d.proto, 1) # proto=IKEv2
self.assertEqual(ih.init_SPI, self.sa.ispi)
self.assertEqual(ih.resp_SPI, self.sa.rspi)
- header = ikev2.IKEv2(init_SPI=self.sa.ispi, resp_SPI=self.sa.rspi,
- flags='Initiator+Response',
- exch_type='INFORMATIONAL',
- id=ih.id, next_payload='Encrypted')
- resp = self.encrypt_ike_msg(header, b'', None)
+ header = ikev2.IKEv2(
+ init_SPI=self.sa.ispi,
+ resp_SPI=self.sa.rspi,
+ flags="Initiator+Response",
+ exch_type="INFORMATIONAL",
+ id=ih.id,
+ next_payload="Encrypted",
+ )
+ resp = self.encrypt_ike_msg(header, b"", None)
self.send_and_assert_no_replies(self.pg0, resp)
def verify_del_sa(self, packet):
ih = self.get_ike_header(packet)
self.assertEqual(ih.id, self.sa.msg_id)
self.assertEqual(ih.exch_type, 37) # exchange informational
- self.assertIn('Response', ih.flags)
- self.assertNotIn('Initiator', ih.flags)
+ self.assertIn("Response", ih.flags)
+ self.assertNotIn("Initiator", ih.flags)
self.assertEqual(ih.next_payload, 46) # Encrypted
self.assertEqual(ih.init_SPI, self.sa.ispi)
self.assertEqual(ih.resp_SPI, self.sa.rspi)
plain = self.sa.hmac_and_decrypt(ih)
- self.assertEqual(plain, b'')
+ self.assertEqual(plain, b"")
def initiate_del_sa_from_initiator(self):
- header = ikev2.IKEv2(init_SPI=self.sa.ispi, resp_SPI=self.sa.rspi,
- flags='Initiator', exch_type='INFORMATIONAL',
- id=self.sa.new_msg_id())
- del_sa = ikev2.IKEv2_payload_Delete(proto='IKEv2')
- ike_msg = self.encrypt_ike_msg(header, del_sa, 'Delete')
- packet = self.create_packet(self.pg0, ike_msg,
- self.sa.sport, self.sa.dport,
- self.sa.natt, self.ip6)
+ header = ikev2.IKEv2(
+ init_SPI=self.sa.ispi,
+ resp_SPI=self.sa.rspi,
+ flags="Initiator",
+ exch_type="INFORMATIONAL",
+ id=self.sa.new_msg_id(),
+ )
+ del_sa = ikev2.IKEv2_payload_Delete(proto="IKEv2")
+ ike_msg = self.encrypt_ike_msg(header, del_sa, "Delete")
+ packet = self.create_packet(
+ self.pg0, ike_msg, self.sa.sport, self.sa.dport, self.sa.natt, self.ip6
+ )
self.pg0.add_stream(packet)
self.pg0.enable_capture()
self.pg_start()
@@ -1175,56 +1272,72 @@ class TemplateResponder(IkePeer):
def send_sa_init_req(self):
tr_attr = self.sa.ike_crypto_attr()
- trans = (ikev2.IKEv2_payload_Transform(transform_type='Encryption',
- transform_id=self.sa.ike_crypto, length=tr_attr[1],
- key_length=tr_attr[0]) /
- ikev2.IKEv2_payload_Transform(transform_type='Integrity',
- transform_id=self.sa.ike_integ) /
- ikev2.IKEv2_payload_Transform(transform_type='PRF',
- transform_id=self.sa.ike_prf_alg.name) /
- ikev2.IKEv2_payload_Transform(transform_type='GroupDesc',
- transform_id=self.sa.ike_dh))
-
- props = (ikev2.IKEv2_payload_Proposal(proposal=1, proto='IKEv2',
- trans_nb=4, trans=trans))
-
- next_payload = None if self.ip6 else 'Notify'
+ trans = (
+ ikev2.IKEv2_payload_Transform(
+ transform_type="Encryption",
+ transform_id=self.sa.ike_crypto,
+ length=tr_attr[1],
+ key_length=tr_attr[0],
+ )
+ / ikev2.IKEv2_payload_Transform(
+ transform_type="Integrity", transform_id=self.sa.ike_integ
+ )
+ / ikev2.IKEv2_payload_Transform(
+ transform_type="PRF", transform_id=self.sa.ike_prf_alg.name
+ )
+ / ikev2.IKEv2_payload_Transform(
+ transform_type="GroupDesc", transform_id=self.sa.ike_dh
+ )
+ )
+
+ props = ikev2.IKEv2_payload_Proposal(
+ proposal=1, proto="IKEv2", trans_nb=4, trans=trans
+ )
+
+ next_payload = None if self.ip6 else "Notify"
self.sa.init_req_packet = (
- ikev2.IKEv2(init_SPI=self.sa.ispi,
- flags='Initiator', exch_type='IKE_SA_INIT') /
- ikev2.IKEv2_payload_SA(next_payload='KE', prop=props) /
- ikev2.IKEv2_payload_KE(next_payload='Nonce',
- group=self.sa.ike_dh,
- load=self.sa.my_dh_pub_key) /
- ikev2.IKEv2_payload_Nonce(next_payload=next_payload,
- load=self.sa.i_nonce))
+ ikev2.IKEv2(
+ init_SPI=self.sa.ispi, flags="Initiator", exch_type="IKE_SA_INIT"
+ )
+ / ikev2.IKEv2_payload_SA(next_payload="KE", prop=props)
+ / ikev2.IKEv2_payload_KE(
+ next_payload="Nonce", group=self.sa.ike_dh, load=self.sa.my_dh_pub_key
+ )
+ / ikev2.IKEv2_payload_Nonce(next_payload=next_payload, load=self.sa.i_nonce)
+ )
if not self.ip6:
if self.sa.i_natt:
- src_address = b'\x0a\x0a\x0a\x01'
+ src_address = b"\x0a\x0a\x0a\x01"
else:
src_address = inet_pton(socket.AF_INET, self.pg0.remote_ip4)
if self.sa.r_natt:
- dst_address = b'\x0a\x0a\x0a\x0a'
+ dst_address = b"\x0a\x0a\x0a\x0a"
else:
dst_address = inet_pton(socket.AF_INET, self.pg0.local_ip4)
src_nat = self.sa.compute_nat_sha1(src_address, self.sa.sport)
dst_nat = self.sa.compute_nat_sha1(dst_address, self.sa.dport)
nat_src_detection = ikev2.IKEv2_payload_Notify(
- type='NAT_DETECTION_SOURCE_IP', load=src_nat,
- next_payload='Notify')
+ type="NAT_DETECTION_SOURCE_IP", load=src_nat, next_payload="Notify"
+ )
nat_dst_detection = ikev2.IKEv2_payload_Notify(
- type='NAT_DETECTION_DESTINATION_IP', load=dst_nat)
- self.sa.init_req_packet = (self.sa.init_req_packet /
- nat_src_detection /
- nat_dst_detection)
-
- ike_msg = self.create_packet(self.pg0, self.sa.init_req_packet,
- self.sa.sport, self.sa.dport,
- self.sa.natt, self.ip6)
+ type="NAT_DETECTION_DESTINATION_IP", load=dst_nat
+ )
+ self.sa.init_req_packet = (
+ self.sa.init_req_packet / nat_src_detection / nat_dst_detection
+ )
+
+ ike_msg = self.create_packet(
+ self.pg0,
+ self.sa.init_req_packet,
+ self.sa.sport,
+ self.sa.dport,
+ self.sa.natt,
+ self.ip6,
+ )
self.pg0.add_stream(ike_msg)
self.pg0.enable_capture()
self.pg_start()
@@ -1233,65 +1346,83 @@ class TemplateResponder(IkePeer):
def generate_auth_payload(self, last_payload=None, is_rekey=False):
tr_attr = self.sa.esp_crypto_attr()
- last_payload = last_payload or 'Notify'
- trans = (ikev2.IKEv2_payload_Transform(transform_type='Encryption',
- transform_id=self.sa.esp_crypto, length=tr_attr[1],
- key_length=tr_attr[0]) /
- ikev2.IKEv2_payload_Transform(transform_type='Integrity',
- transform_id=self.sa.esp_integ) /
- ikev2.IKEv2_payload_Transform(
- transform_type='Extended Sequence Number',
- transform_id='No ESN') /
- ikev2.IKEv2_payload_Transform(
- transform_type='Extended Sequence Number',
- transform_id='ESN'))
+ last_payload = last_payload or "Notify"
+ trans = (
+ ikev2.IKEv2_payload_Transform(
+ transform_type="Encryption",
+ transform_id=self.sa.esp_crypto,
+ length=tr_attr[1],
+ key_length=tr_attr[0],
+ )
+ / ikev2.IKEv2_payload_Transform(
+ transform_type="Integrity", transform_id=self.sa.esp_integ
+ )
+ / ikev2.IKEv2_payload_Transform(
+ transform_type="Extended Sequence Number", transform_id="No ESN"
+ )
+ / ikev2.IKEv2_payload_Transform(
+ transform_type="Extended Sequence Number", transform_id="ESN"
+ )
+ )
c = self.sa.child_sas[0]
- props = (ikev2.IKEv2_payload_Proposal(proposal=1, proto='ESP',
- SPIsize=4, SPI=c.ispi, trans_nb=4, trans=trans))
+ props = ikev2.IKEv2_payload_Proposal(
+ proposal=1, proto="ESP", SPIsize=4, SPI=c.ispi, trans_nb=4, trans=trans
+ )
tsi, tsr = self.sa.generate_ts(self.p.ts_is_ip4)
- plain = (ikev2.IKEv2_payload_AUTH(next_payload='SA',
- auth_type=AuthMethod.value(self.sa.auth_method),
- load=self.sa.auth_data) /
- ikev2.IKEv2_payload_SA(next_payload='TSi', prop=props) /
- ikev2.IKEv2_payload_TSi(next_payload='TSr',
- number_of_TSs=len(tsi), traffic_selector=tsi) /
- ikev2.IKEv2_payload_TSr(next_payload=last_payload,
- number_of_TSs=len(tsr), traffic_selector=tsr))
+ plain = (
+ ikev2.IKEv2_payload_AUTH(
+ next_payload="SA",
+ auth_type=AuthMethod.value(self.sa.auth_method),
+ load=self.sa.auth_data,
+ )
+ / ikev2.IKEv2_payload_SA(next_payload="TSi", prop=props)
+ / ikev2.IKEv2_payload_TSi(
+ next_payload="TSr", number_of_TSs=len(tsi), traffic_selector=tsi
+ )
+ / ikev2.IKEv2_payload_TSr(
+ next_payload=last_payload, number_of_TSs=len(tsr), traffic_selector=tsr
+ )
+ )
if is_rekey:
- first_payload = 'Nonce'
- plain = (ikev2.IKEv2_payload_Nonce(load=self.sa.i_nonce,
- next_payload='SA') / plain /
- ikev2.IKEv2_payload_Notify(type='REKEY_SA',
- proto='ESP', SPI=c.ispi))
+ first_payload = "Nonce"
+ plain = (
+ ikev2.IKEv2_payload_Nonce(load=self.sa.i_nonce, next_payload="SA")
+ / plain
+ / ikev2.IKEv2_payload_Notify(type="REKEY_SA", proto="ESP", SPI=c.ispi)
+ )
else:
- first_payload = 'IDi'
+ first_payload = "IDi"
if self.no_idr_auth:
- ids = ikev2.IKEv2_payload_IDi(next_payload='AUTH',
- IDtype=self.sa.id_type,
- load=self.sa.i_id)
+ ids = ikev2.IKEv2_payload_IDi(
+ next_payload="AUTH", IDtype=self.sa.id_type, load=self.sa.i_id
+ )
else:
- ids = (ikev2.IKEv2_payload_IDi(next_payload='IDr',
- IDtype=self.sa.id_type, load=self.sa.i_id) /
- ikev2.IKEv2_payload_IDr(next_payload='AUTH',
- IDtype=self.sa.id_type, load=self.sa.r_id))
+ ids = ikev2.IKEv2_payload_IDi(
+ next_payload="IDr", IDtype=self.sa.id_type, load=self.sa.i_id
+ ) / ikev2.IKEv2_payload_IDr(
+ next_payload="AUTH", IDtype=self.sa.id_type, load=self.sa.r_id
+ )
plain = ids / plain
return plain, first_payload
def send_sa_auth(self):
- plain, first_payload = self.generate_auth_payload(
- last_payload='Notify')
- plain = plain / ikev2.IKEv2_payload_Notify(type='INITIAL_CONTACT')
+ plain, first_payload = self.generate_auth_payload(last_payload="Notify")
+ plain = plain / ikev2.IKEv2_payload_Notify(type="INITIAL_CONTACT")
header = ikev2.IKEv2(
- init_SPI=self.sa.ispi,
- resp_SPI=self.sa.rspi, id=self.sa.new_msg_id(),
- flags='Initiator', exch_type='IKE_AUTH')
+ init_SPI=self.sa.ispi,
+ resp_SPI=self.sa.rspi,
+ id=self.sa.new_msg_id(),
+ flags="Initiator",
+ exch_type="IKE_AUTH",
+ )
ike_msg = self.encrypt_ike_msg(header, plain, first_payload)
- packet = self.create_packet(self.pg0, ike_msg, self.sa.sport,
- self.sa.dport, self.sa.natt, self.ip6)
+ packet = self.create_packet(
+ self.pg0, ike_msg, self.sa.sport, self.sa.dport, self.sa.natt, self.ip6
+ )
self.pg0.add_stream(packet)
self.pg0.enable_capture()
self.pg_start()
@@ -1303,7 +1434,7 @@ class TemplateResponder(IkePeer):
self.assertEqual(ih.id, self.sa.msg_id)
self.assertEqual(ih.exch_type, 34)
- self.assertIn('Response', ih.flags)
+ self.assertIn("Response", ih.flags)
self.assertEqual(ih.init_SPI, self.sa.ispi)
self.assertNotEqual(ih.resp_SPI, 0)
self.sa.rspi = ih.resp_SPI
@@ -1331,12 +1462,12 @@ class TemplateResponder(IkePeer):
self.sa.child_sas[0].rspi = prop.SPI
self.sa.calc_child_keys()
- IKE_NODE_SUFFIX = 'ip4'
+ IKE_NODE_SUFFIX = "ip4"
def verify_counters(self):
- self.assert_counter(2, 'processed', self.IKE_NODE_SUFFIX)
- self.assert_counter(1, 'init_sa_req', self.IKE_NODE_SUFFIX)
- self.assert_counter(1, 'ike_auth_req', self.IKE_NODE_SUFFIX)
+ self.assert_counter(2, "processed", self.IKE_NODE_SUFFIX)
+ self.assert_counter(1, "init_sa_req", self.IKE_NODE_SUFFIX)
+ self.assert_counter(1, "ike_auth_req", self.IKE_NODE_SUFFIX)
r = self.vapi.ikev2_sa_dump()
s = r[0].sa.stats
@@ -1356,59 +1487,60 @@ class Ikev2Params(object):
ec = VppEnum.vl_api_ipsec_crypto_alg_t
ei = VppEnum.vl_api_ipsec_integ_alg_t
self.vpp_enums = {
- 'AES-CBC-128': ec.IPSEC_API_CRYPTO_ALG_AES_CBC_128,
- 'AES-CBC-192': ec.IPSEC_API_CRYPTO_ALG_AES_CBC_192,
- 'AES-CBC-256': ec.IPSEC_API_CRYPTO_ALG_AES_CBC_256,
- 'AES-GCM-16ICV-128': ec.IPSEC_API_CRYPTO_ALG_AES_GCM_128,
- 'AES-GCM-16ICV-192': ec.IPSEC_API_CRYPTO_ALG_AES_GCM_192,
- 'AES-GCM-16ICV-256': ec.IPSEC_API_CRYPTO_ALG_AES_GCM_256,
-
- 'HMAC-SHA1-96': ei.IPSEC_API_INTEG_ALG_SHA1_96,
- 'SHA2-256-128': ei.IPSEC_API_INTEG_ALG_SHA_256_128,
- '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']
+ "AES-CBC-128": ec.IPSEC_API_CRYPTO_ALG_AES_CBC_128,
+ "AES-CBC-192": ec.IPSEC_API_CRYPTO_ALG_AES_CBC_192,
+ "AES-CBC-256": ec.IPSEC_API_CRYPTO_ALG_AES_CBC_256,
+ "AES-GCM-16ICV-128": ec.IPSEC_API_CRYPTO_ALG_AES_GCM_128,
+ "AES-GCM-16ICV-192": ec.IPSEC_API_CRYPTO_ALG_AES_GCM_192,
+ "AES-GCM-16ICV-256": ec.IPSEC_API_CRYPTO_ALG_AES_GCM_256,
+ "HMAC-SHA1-96": ei.IPSEC_API_INTEG_ALG_SHA1_96,
+ "SHA2-256-128": ei.IPSEC_API_INTEG_ALG_SHA_256_128,
+ "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']
- i_natt = False if 'i_natt' not in params else params['i_natt']
- r_natt = False if 'r_natt' not in params else params['r_natt']
- self.p = Profile(self, 'pr1')
- self.ip6 = False if 'ip6' not in params else params['ip6']
-
- if 'auth' in params and params['auth'] == 'rsa-sig':
- auth_method = 'rsa-sig'
+ 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"]
+ )
+ i_natt = False if "i_natt" not in params else params["i_natt"]
+ r_natt = False if "r_natt" not in params else params["r_natt"]
+ self.p = Profile(self, "pr1")
+ self.ip6 = False if "ip6" not in params else params["ip6"]
+
+ if "auth" in params and params["auth"] == "rsa-sig":
+ auth_method = "rsa-sig"
work_dir = f"{config.vpp_ws_dir}/src/plugins/ikev2/test/certs/"
- self.vapi.ikev2_set_local_key(
- key_file=work_dir + params['server-key'])
-
- client_file = work_dir + params['client-cert']
- server_pem = open(work_dir + params['server-cert']).read()
- client_priv = open(work_dir + params['client-key']).read()
- client_priv = load_pem_private_key(str.encode(client_priv), None,
- default_backend())
+ self.vapi.ikev2_set_local_key(key_file=work_dir + params["server-key"])
+
+ client_file = work_dir + params["client-cert"]
+ server_pem = open(work_dir + params["server-cert"]).read()
+ client_priv = open(work_dir + params["client-key"]).read()
+ client_priv = load_pem_private_key(
+ str.encode(client_priv), None, default_backend()
+ )
self.peer_cert = x509.load_pem_x509_certificate(
- str.encode(server_pem),
- default_backend())
- self.p.add_auth(method='rsa-sig', data=str.encode(client_file))
+ str.encode(server_pem), default_backend()
+ )
+ self.p.add_auth(method="rsa-sig", data=str.encode(client_file))
auth_data = None
else:
- auth_data = b'$3cr3tpa$$w0rd'
- self.p.add_auth(method='shared-key', data=auth_data)
- auth_method = 'shared-key'
+ auth_data = b"$3cr3tpa$$w0rd"
+ self.p.add_auth(method="shared-key", data=auth_data)
+ auth_method = "shared-key"
client_priv = None
- is_init = True if 'is_initiator' not in params else\
- params['is_initiator']
- self.no_idr_auth = params.get('no_idr_in_auth', False)
+ is_init = True if "is_initiator" not in params else params["is_initiator"]
+ self.no_idr_auth = params.get("no_idr_in_auth", False)
- idr = {'id_type': 'fqdn', 'data': b'vpp.home'}
- idi = {'id_type': 'fqdn', 'data': b'roadwarrior.example.com'}
- r_id = self.idr = idr['data']
- i_id = self.idi = idi['data']
+ idr = {"id_type": "fqdn", "data": b"vpp.home"}
+ idi = {"id_type": "fqdn", "data": b"roadwarrior.example.com"}
+ r_id = self.idr = idr["data"]
+ i_id = self.idi = idi["data"]
if is_init:
# scapy is initiator, VPP is responder
self.p.add_local_id(**idr)
@@ -1421,70 +1553,90 @@ class Ikev2Params(object):
if not self.no_idr_auth:
self.p.add_remote_id(**idr)
- loc_ts = {'start_addr': '10.10.10.0', 'end_addr': '10.10.10.255'} if\
- 'loc_ts' not in params else params['loc_ts']
- rem_ts = {'start_addr': '10.0.0.0', 'end_addr': '10.0.0.255'} if\
- 'rem_ts' not in params else params['rem_ts']
+ loc_ts = (
+ {"start_addr": "10.10.10.0", "end_addr": "10.10.10.255"}
+ if "loc_ts" not in params
+ else params["loc_ts"]
+ )
+ rem_ts = (
+ {"start_addr": "10.0.0.0", "end_addr": "10.0.0.255"}
+ if "rem_ts" not in params
+ else params["rem_ts"]
+ )
self.p.add_local_ts(**loc_ts)
self.p.add_remote_ts(**rem_ts)
- if 'responder' in params:
- self.p.add_responder(params['responder'])
- if 'ike_transforms' in params:
- self.p.add_ike_transforms(params['ike_transforms'])
- if 'esp_transforms' in params:
- self.p.add_esp_transforms(params['esp_transforms'])
-
- udp_encap = False if 'udp_encap' not in params else\
- params['udp_encap']
+ if "responder" in params:
+ self.p.add_responder(params["responder"])
+ if "ike_transforms" in params:
+ self.p.add_ike_transforms(params["ike_transforms"])
+ if "esp_transforms" in params:
+ self.p.add_esp_transforms(params["esp_transforms"])
+
+ udp_encap = False if "udp_encap" not in params else params["udp_encap"]
if udp_encap:
self.p.set_udp_encap(True)
- if 'responder_hostname' in params:
- hn = params['responder_hostname']
+ if "responder_hostname" in params:
+ hn = params["responder_hostname"]
self.p.add_responder_hostname(hn)
# configure static dns record
self.vapi.dns_name_server_add_del(
- is_ip6=0, is_add=1,
- server_address=IPv4Address(u'8.8.8.8').packed)
+ is_ip6=0, is_add=1, server_address=IPv4Address("8.8.8.8").packed
+ )
self.vapi.dns_enable_disable(enable=1)
- cmd = "dns cache add {} {}".format(hn['hostname'],
- self.pg0.remote_ip4)
+ cmd = "dns cache add {} {}".format(hn["hostname"], self.pg0.remote_ip4)
self.vapi.cli(cmd)
- self.sa = IKEv2SA(self, i_id=i_id, r_id=r_id,
- is_initiator=is_init,
- id_type=self.p.local_id['id_type'],
- i_natt=i_natt, r_natt=r_natt,
- priv_key=client_priv, auth_method=auth_method,
- nonce=params.get('nonce'),
- auth_data=auth_data, udp_encap=udp_encap,
- local_ts=self.p.remote_ts, remote_ts=self.p.local_ts)
+ self.sa = IKEv2SA(
+ self,
+ i_id=i_id,
+ r_id=r_id,
+ is_initiator=is_init,
+ id_type=self.p.local_id["id_type"],
+ i_natt=i_natt,
+ r_natt=r_natt,
+ priv_key=client_priv,
+ auth_method=auth_method,
+ nonce=params.get("nonce"),
+ auth_data=auth_data,
+ udp_encap=udp_encap,
+ local_ts=self.p.remote_ts,
+ remote_ts=self.p.local_ts,
+ )
if is_init:
- ike_crypto = ('AES-CBC', 32) if 'ike-crypto' not in params else\
- params['ike-crypto']
- ike_integ = 'HMAC-SHA1-96' if 'ike-integ' not in params else\
- params['ike-integ']
- ike_dh = '2048MODPgr' if 'ike-dh' not in params else\
- params['ike-dh']
-
- esp_crypto = ('AES-CBC', 32) if 'esp-crypto' not in params else\
- params['esp-crypto']
- esp_integ = 'HMAC-SHA1-96' if 'esp-integ' not in params else\
- params['esp-integ']
+ ike_crypto = (
+ ("AES-CBC", 32) if "ike-crypto" not in params else params["ike-crypto"]
+ )
+ ike_integ = (
+ "HMAC-SHA1-96" if "ike-integ" not in params else params["ike-integ"]
+ )
+ ike_dh = "2048MODPgr" if "ike-dh" not in params else params["ike-dh"]
+
+ esp_crypto = (
+ ("AES-CBC", 32) if "esp-crypto" not in params else params["esp-crypto"]
+ )
+ esp_integ = (
+ "HMAC-SHA1-96" if "esp-integ" not in params else params["esp-integ"]
+ )
self.sa.set_ike_props(
- crypto=ike_crypto[0], crypto_key_len=ike_crypto[1],
- integ=ike_integ, prf='PRF_HMAC_SHA2_256', dh=ike_dh)
+ crypto=ike_crypto[0],
+ crypto_key_len=ike_crypto[1],
+ integ=ike_integ,
+ prf="PRF_HMAC_SHA2_256",
+ dh=ike_dh,
+ )
self.sa.set_esp_props(
- crypto=esp_crypto[0], crypto_key_len=esp_crypto[1],
- integ=esp_integ)
+ crypto=esp_crypto[0], crypto_key_len=esp_crypto[1], integ=esp_integ
+ )
class TestApi(VppTestCase):
- """ Test IKEV2 API """
+ """Test IKEV2 API"""
+
@classmethod
def setUpClass(cls):
super(TestApi, cls).setUpClass()
@@ -1501,241 +1653,249 @@ class TestApi(VppTestCase):
self.assertEqual(len(r), 0)
def configure_profile(self, cfg):
- p = Profile(self, cfg['name'])
- p.add_local_id(id_type=cfg['loc_id'][0], data=cfg['loc_id'][1])
- p.add_remote_id(id_type=cfg['rem_id'][0], data=cfg['rem_id'][1])
- p.add_local_ts(**cfg['loc_ts'])
- p.add_remote_ts(**cfg['rem_ts'])
- p.add_responder(cfg['responder'])
- p.add_ike_transforms(cfg['ike_ts'])
- p.add_esp_transforms(cfg['esp_ts'])
- p.add_auth(**cfg['auth'])
- p.set_udp_encap(cfg['udp_encap'])
- p.set_ipsec_over_udp_port(cfg['ipsec_over_udp_port'])
- if 'lifetime_data' in cfg:
- p.set_lifetime_data(cfg['lifetime_data'])
- if 'tun_itf' in cfg:
- p.set_tunnel_interface(cfg['tun_itf'])
- if 'natt_disabled' in cfg and cfg['natt_disabled']:
+ p = Profile(self, cfg["name"])
+ p.add_local_id(id_type=cfg["loc_id"][0], data=cfg["loc_id"][1])
+ p.add_remote_id(id_type=cfg["rem_id"][0], data=cfg["rem_id"][1])
+ p.add_local_ts(**cfg["loc_ts"])
+ p.add_remote_ts(**cfg["rem_ts"])
+ p.add_responder(cfg["responder"])
+ p.add_ike_transforms(cfg["ike_ts"])
+ p.add_esp_transforms(cfg["esp_ts"])
+ p.add_auth(**cfg["auth"])
+ p.set_udp_encap(cfg["udp_encap"])
+ p.set_ipsec_over_udp_port(cfg["ipsec_over_udp_port"])
+ if "lifetime_data" in cfg:
+ p.set_lifetime_data(cfg["lifetime_data"])
+ if "tun_itf" in cfg:
+ p.set_tunnel_interface(cfg["tun_itf"])
+ if "natt_disabled" in cfg and cfg["natt_disabled"]:
p.disable_natt()
p.add_vpp_config()
return p
def test_profile_api(self):
- """ test profile dump API """
+ """test profile dump API"""
loc_ts4 = {
- 'proto': 8,
- 'start_port': 1,
- 'end_port': 19,
- 'start_addr': '3.3.3.2',
- 'end_addr': '3.3.3.3',
- }
+ "proto": 8,
+ "start_port": 1,
+ "end_port": 19,
+ "start_addr": "3.3.3.2",
+ "end_addr": "3.3.3.3",
+ }
rem_ts4 = {
- 'proto': 9,
- 'start_port': 10,
- 'end_port': 119,
- 'start_addr': '4.5.76.80',
- 'end_addr': '2.3.4.6',
- }
+ "proto": 9,
+ "start_port": 10,
+ "end_port": 119,
+ "start_addr": "4.5.76.80",
+ "end_addr": "2.3.4.6",
+ }
loc_ts6 = {
- 'proto': 8,
- 'start_port': 1,
- 'end_port': 19,
- 'start_addr': 'ab::1',
- 'end_addr': 'ab::4',
- }
+ "proto": 8,
+ "start_port": 1,
+ "end_port": 19,
+ "start_addr": "ab::1",
+ "end_addr": "ab::4",
+ }
rem_ts6 = {
- 'proto': 9,
- 'start_port': 10,
- 'end_port': 119,
- 'start_addr': 'cd::12',
- 'end_addr': 'cd::13',
- }
+ "proto": 9,
+ "start_port": 10,
+ "end_port": 119,
+ "start_addr": "cd::12",
+ "end_addr": "cd::13",
+ }
conf = {
- 'p1': {
- 'name': 'p1',
- 'natt_disabled': True,
- 'loc_id': ('fqdn', b'vpp.home'),
- 'rem_id': ('fqdn', b'roadwarrior.example.com'),
- 'loc_ts': loc_ts4,
- 'rem_ts': rem_ts4,
- 'responder': {'sw_if_index': 0, 'addr': '5.6.7.8'},
- 'ike_ts': {
- 'crypto_alg': 20,
- 'crypto_key_size': 32,
- 'integ_alg': 0,
- 'dh_group': 1},
- 'esp_ts': {
- 'crypto_alg': 13,
- 'crypto_key_size': 24,
- 'integ_alg': 2},
- 'auth': {'method': 'shared-key', 'data': b'sharedkeydata'},
- 'udp_encap': True,
- 'ipsec_over_udp_port': 4501,
- 'lifetime_data': {
- 'lifetime': 123,
- 'lifetime_maxdata': 20192,
- 'lifetime_jitter': 9,
- 'handover': 132},
+ "p1": {
+ "name": "p1",
+ "natt_disabled": True,
+ "loc_id": ("fqdn", b"vpp.home"),
+ "rem_id": ("fqdn", b"roadwarrior.example.com"),
+ "loc_ts": loc_ts4,
+ "rem_ts": rem_ts4,
+ "responder": {"sw_if_index": 0, "addr": "5.6.7.8"},
+ "ike_ts": {
+ "crypto_alg": 20,
+ "crypto_key_size": 32,
+ "integ_alg": 0,
+ "dh_group": 1,
+ },
+ "esp_ts": {"crypto_alg": 13, "crypto_key_size": 24, "integ_alg": 2},
+ "auth": {"method": "shared-key", "data": b"sharedkeydata"},
+ "udp_encap": True,
+ "ipsec_over_udp_port": 4501,
+ "lifetime_data": {
+ "lifetime": 123,
+ "lifetime_maxdata": 20192,
+ "lifetime_jitter": 9,
+ "handover": 132,
+ },
+ },
+ "p2": {
+ "name": "p2",
+ "loc_id": ("ip4-addr", b"192.168.2.1"),
+ "rem_id": ("ip6-addr", b"abcd::1"),
+ "loc_ts": loc_ts6,
+ "rem_ts": rem_ts6,
+ "responder": {"sw_if_index": 4, "addr": "def::10"},
+ "ike_ts": {
+ "crypto_alg": 12,
+ "crypto_key_size": 16,
+ "integ_alg": 3,
+ "dh_group": 3,
+ },
+ "esp_ts": {"crypto_alg": 9, "crypto_key_size": 24, "integ_alg": 4},
+ "auth": {"method": "shared-key", "data": b"sharedkeydata"},
+ "udp_encap": False,
+ "ipsec_over_udp_port": 4600,
+ "tun_itf": 0,
},
- 'p2': {
- 'name': 'p2',
- 'loc_id': ('ip4-addr', b'192.168.2.1'),
- 'rem_id': ('ip6-addr', b'abcd::1'),
- 'loc_ts': loc_ts6,
- 'rem_ts': rem_ts6,
- 'responder': {'sw_if_index': 4, 'addr': 'def::10'},
- 'ike_ts': {
- 'crypto_alg': 12,
- 'crypto_key_size': 16,
- 'integ_alg': 3,
- 'dh_group': 3},
- 'esp_ts': {
- 'crypto_alg': 9,
- 'crypto_key_size': 24,
- 'integ_alg': 4},
- 'auth': {'method': 'shared-key', 'data': b'sharedkeydata'},
- 'udp_encap': False,
- 'ipsec_over_udp_port': 4600,
- 'tun_itf': 0}
}
- self.p1 = self.configure_profile(conf['p1'])
- self.p2 = self.configure_profile(conf['p2'])
+ self.p1 = self.configure_profile(conf["p1"])
+ self.p2 = self.configure_profile(conf["p2"])
r = self.vapi.ikev2_profile_dump()
self.assertEqual(len(r), 2)
- self.verify_profile(r[0].profile, conf['p1'])
- self.verify_profile(r[1].profile, conf['p2'])
+ self.verify_profile(r[0].profile, conf["p1"])
+ self.verify_profile(r[1].profile, conf["p2"])
def verify_id(self, api_id, cfg_id):
self.assertEqual(api_id.type, IDType.value(cfg_id[0]))
- self.assertEqual(bytes(api_id.data, 'ascii'), cfg_id[1])
+ self.assertEqual(bytes(api_id.data, "ascii"), cfg_id[1])
def verify_ts(self, api_ts, cfg_ts):
- self.assertEqual(api_ts.protocol_id, cfg_ts['proto'])
- self.assertEqual(api_ts.start_port, cfg_ts['start_port'])
- self.assertEqual(api_ts.end_port, cfg_ts['end_port'])
- self.assertEqual(api_ts.start_addr,
- ip_address(text_type(cfg_ts['start_addr'])))
- self.assertEqual(api_ts.end_addr,
- ip_address(text_type(cfg_ts['end_addr'])))
+ self.assertEqual(api_ts.protocol_id, cfg_ts["proto"])
+ self.assertEqual(api_ts.start_port, cfg_ts["start_port"])
+ self.assertEqual(api_ts.end_port, cfg_ts["end_port"])
+ self.assertEqual(api_ts.start_addr, ip_address(text_type(cfg_ts["start_addr"])))
+ self.assertEqual(api_ts.end_addr, ip_address(text_type(cfg_ts["end_addr"])))
def verify_responder(self, api_r, cfg_r):
- self.assertEqual(api_r.sw_if_index, cfg_r['sw_if_index'])
- self.assertEqual(api_r.addr, ip_address(cfg_r['addr']))
+ self.assertEqual(api_r.sw_if_index, cfg_r["sw_if_index"])
+ self.assertEqual(api_r.addr, ip_address(cfg_r["addr"]))
def verify_transforms(self, api_ts, cfg_ts):
- self.assertEqual(api_ts.crypto_alg, cfg_ts['crypto_alg'])
- self.assertEqual(api_ts.crypto_key_size, cfg_ts['crypto_key_size'])
- self.assertEqual(api_ts.integ_alg, cfg_ts['integ_alg'])
+ self.assertEqual(api_ts.crypto_alg, cfg_ts["crypto_alg"])
+ self.assertEqual(api_ts.crypto_key_size, cfg_ts["crypto_key_size"])
+ self.assertEqual(api_ts.integ_alg, cfg_ts["integ_alg"])
def verify_ike_transforms(self, api_ts, cfg_ts):
self.verify_transforms(api_ts, cfg_ts)
- self.assertEqual(api_ts.dh_group, cfg_ts['dh_group'])
+ self.assertEqual(api_ts.dh_group, cfg_ts["dh_group"])
def verify_esp_transforms(self, api_ts, cfg_ts):
self.verify_transforms(api_ts, cfg_ts)
def verify_auth(self, api_auth, cfg_auth):
- self.assertEqual(api_auth.method, AuthMethod.value(cfg_auth['method']))
- self.assertEqual(api_auth.data, cfg_auth['data'])
- self.assertEqual(api_auth.data_len, len(cfg_auth['data']))
+ self.assertEqual(api_auth.method, AuthMethod.value(cfg_auth["method"]))
+ self.assertEqual(api_auth.data, cfg_auth["data"])
+ self.assertEqual(api_auth.data_len, len(cfg_auth["data"]))
def verify_lifetime_data(self, p, ld):
- self.assertEqual(p.lifetime, ld['lifetime'])
- self.assertEqual(p.lifetime_maxdata, ld['lifetime_maxdata'])
- self.assertEqual(p.lifetime_jitter, ld['lifetime_jitter'])
- self.assertEqual(p.handover, ld['handover'])
+ self.assertEqual(p.lifetime, ld["lifetime"])
+ self.assertEqual(p.lifetime_maxdata, ld["lifetime_maxdata"])
+ self.assertEqual(p.lifetime_jitter, ld["lifetime_jitter"])
+ self.assertEqual(p.handover, ld["handover"])
def verify_profile(self, ap, cp):
- self.assertEqual(ap.name, cp['name'])
- self.assertEqual(ap.udp_encap, cp['udp_encap'])
- self.verify_id(ap.loc_id, cp['loc_id'])
- self.verify_id(ap.rem_id, cp['rem_id'])
- self.verify_ts(ap.loc_ts, cp['loc_ts'])
- self.verify_ts(ap.rem_ts, cp['rem_ts'])
- self.verify_responder(ap.responder, cp['responder'])
- self.verify_ike_transforms(ap.ike_ts, cp['ike_ts'])
- self.verify_esp_transforms(ap.esp_ts, cp['esp_ts'])
- self.verify_auth(ap.auth, cp['auth'])
- natt_dis = False if 'natt_disabled' not in cp else cp['natt_disabled']
+ self.assertEqual(ap.name, cp["name"])
+ self.assertEqual(ap.udp_encap, cp["udp_encap"])
+ self.verify_id(ap.loc_id, cp["loc_id"])
+ self.verify_id(ap.rem_id, cp["rem_id"])
+ self.verify_ts(ap.loc_ts, cp["loc_ts"])
+ self.verify_ts(ap.rem_ts, cp["rem_ts"])
+ self.verify_responder(ap.responder, cp["responder"])
+ self.verify_ike_transforms(ap.ike_ts, cp["ike_ts"])
+ self.verify_esp_transforms(ap.esp_ts, cp["esp_ts"])
+ self.verify_auth(ap.auth, cp["auth"])
+ natt_dis = False if "natt_disabled" not in cp else cp["natt_disabled"]
self.assertTrue(natt_dis == ap.natt_disabled)
- if 'lifetime_data' in cp:
- self.verify_lifetime_data(ap, cp['lifetime_data'])
- self.assertEqual(ap.ipsec_over_udp_port, cp['ipsec_over_udp_port'])
- if 'tun_itf' in cp:
- self.assertEqual(ap.tun_itf, cp['tun_itf'])
+ if "lifetime_data" in cp:
+ self.verify_lifetime_data(ap, cp["lifetime_data"])
+ self.assertEqual(ap.ipsec_over_udp_port, cp["ipsec_over_udp_port"])
+ if "tun_itf" in cp:
+ self.assertEqual(ap.tun_itf, cp["tun_itf"])
else:
- self.assertEqual(ap.tun_itf, 0xffffffff)
+ self.assertEqual(ap.tun_itf, 0xFFFFFFFF)
@tag_fixme_vpp_workers
class TestResponderBehindNAT(TemplateResponder, Ikev2Params):
- """ test responder - responder behind NAT """
+ """test responder - responder behind NAT"""
- IKE_NODE_SUFFIX = 'ip4-natt'
+ IKE_NODE_SUFFIX = "ip4-natt"
def config_tc(self):
- self.config_params({'r_natt': True})
+ self.config_params({"r_natt": True})
@tag_fixme_vpp_workers
class TestInitiatorNATT(TemplateInitiator, Ikev2Params):
- """ test ikev2 initiator - NAT traversal (intitiator behind NAT) """
+ """test ikev2 initiator - NAT traversal (intitiator behind NAT)"""
def config_tc(self):
- self.config_params({
- 'i_natt': True,
- 'is_initiator': False, # seen from test case perspective
- # thus vpp is initiator
- 'responder': {'sw_if_index': self.pg0.sw_if_index,
- 'addr': self.pg0.remote_ip4},
- 'ike-crypto': ('AES-GCM-16ICV', 32),
- 'ike-integ': 'NULL',
- 'ike-dh': '3072MODPgr',
- 'ike_transforms': {
- 'crypto_alg': 20, # "aes-gcm-16"
- 'crypto_key_size': 256,
- 'dh_group': 15, # "modp-3072"
- },
- 'esp_transforms': {
- 'crypto_alg': 12, # "aes-cbc"
- 'crypto_key_size': 256,
- # "hmac-sha2-256-128"
- 'integ_alg': 12}})
+ self.config_params(
+ {
+ "i_natt": True,
+ "is_initiator": False, # seen from test case perspective
+ # thus vpp is initiator
+ "responder": {
+ "sw_if_index": self.pg0.sw_if_index,
+ "addr": self.pg0.remote_ip4,
+ },
+ "ike-crypto": ("AES-GCM-16ICV", 32),
+ "ike-integ": "NULL",
+ "ike-dh": "3072MODPgr",
+ "ike_transforms": {
+ "crypto_alg": 20, # "aes-gcm-16"
+ "crypto_key_size": 256,
+ "dh_group": 15, # "modp-3072"
+ },
+ "esp_transforms": {
+ "crypto_alg": 12, # "aes-cbc"
+ "crypto_key_size": 256,
+ # "hmac-sha2-256-128"
+ "integ_alg": 12,
+ },
+ }
+ )
@tag_fixme_vpp_workers
class TestInitiatorPsk(TemplateInitiator, Ikev2Params):
- """ test ikev2 initiator - pre shared key auth """
+ """test ikev2 initiator - pre shared key auth"""
def config_tc(self):
- self.config_params({
- 'is_initiator': False, # seen from test case perspective
- # thus vpp is initiator
- 'ike-crypto': ('AES-GCM-16ICV', 32),
- 'ike-integ': 'NULL',
- 'ike-dh': '3072MODPgr',
- 'ike_transforms': {
- 'crypto_alg': 20, # "aes-gcm-16"
- 'crypto_key_size': 256,
- 'dh_group': 15, # "modp-3072"
- },
- 'esp_transforms': {
- 'crypto_alg': 12, # "aes-cbc"
- 'crypto_key_size': 256,
- # "hmac-sha2-256-128"
- 'integ_alg': 12},
- 'responder_hostname': {'hostname': 'vpp.responder.org',
- 'sw_if_index': self.pg0.sw_if_index}})
+ self.config_params(
+ {
+ "is_initiator": False, # seen from test case perspective
+ # thus vpp is initiator
+ "ike-crypto": ("AES-GCM-16ICV", 32),
+ "ike-integ": "NULL",
+ "ike-dh": "3072MODPgr",
+ "ike_transforms": {
+ "crypto_alg": 20, # "aes-gcm-16"
+ "crypto_key_size": 256,
+ "dh_group": 15, # "modp-3072"
+ },
+ "esp_transforms": {
+ "crypto_alg": 12, # "aes-cbc"
+ "crypto_key_size": 256,
+ # "hmac-sha2-256-128"
+ "integ_alg": 12,
+ },
+ "responder_hostname": {
+ "hostname": "vpp.responder.org",
+ "sw_if_index": self.pg0.sw_if_index,
+ },
+ }
+ )
@tag_fixme_vpp_workers
class TestInitiatorRequestWindowSize(TestInitiatorPsk):
- """ test initiator - request window size (1) """
+ """test initiator - request window size (1)"""
def rekey_respond(self, req, update_child_sa_data):
ih = self.get_ike_header(req)
@@ -1749,19 +1909,25 @@ class TestInitiatorRequestWindowSize(TestInitiatorPsk):
self.sa.child_sas[0].rspi = prop.SPI
self.sa.calc_child_keys()
- header = ikev2.IKEv2(init_SPI=self.sa.ispi, resp_SPI=self.sa.rspi,
- flags='Response', exch_type=36,
- id=ih.id, next_payload='Encrypted')
- resp = self.encrypt_ike_msg(header, sa, 'SA')
- packet = self.create_packet(self.pg0, resp, self.sa.sport,
- self.sa.dport, self.sa.natt, self.ip6)
+ header = ikev2.IKEv2(
+ init_SPI=self.sa.ispi,
+ resp_SPI=self.sa.rspi,
+ flags="Response",
+ exch_type=36,
+ id=ih.id,
+ next_payload="Encrypted",
+ )
+ resp = self.encrypt_ike_msg(header, sa, "SA")
+ packet = self.create_packet(
+ self.pg0, resp, self.sa.sport, self.sa.dport, self.sa.natt, self.ip6
+ )
self.send_and_assert_no_replies(self.pg0, packet)
def test_initiator(self):
super(TestInitiatorRequestWindowSize, self).test_initiator()
self.pg0.enable_capture()
self.pg_start()
- ispi = int.from_bytes(self.sa.child_sas[0].ispi, 'little')
+ ispi = int.from_bytes(self.sa.child_sas[0].ispi, "little")
self.vapi.ikev2_initiate_rekey_child_sa(ispi=ispi)
self.vapi.ikev2_initiate_rekey_child_sa(ispi=ispi)
capture = self.pg0.get_capture(2)
@@ -1777,18 +1943,18 @@ class TestInitiatorRequestWindowSize(TestInitiatorPsk):
@tag_fixme_vpp_workers
class TestInitiatorRekey(TestInitiatorPsk):
- """ test ikev2 initiator - rekey """
+ """test ikev2 initiator - rekey"""
def rekey_from_initiator(self):
- ispi = int.from_bytes(self.sa.child_sas[0].ispi, 'little')
+ ispi = int.from_bytes(self.sa.child_sas[0].ispi, "little")
self.pg0.enable_capture()
self.pg_start()
self.vapi.ikev2_initiate_rekey_child_sa(ispi=ispi)
capture = self.pg0.get_capture(1)
ih = self.get_ike_header(capture[0])
self.assertEqual(ih.exch_type, 36) # CHILD_SA
- self.assertNotIn('Response', ih.flags)
- self.assertIn('Initiator', ih.flags)
+ self.assertNotIn("Response", ih.flags)
+ self.assertIn("Initiator", ih.flags)
plain = self.sa.hmac_and_decrypt(ih)
sa = ikev2.IKEv2_payload_SA(plain)
prop = sa[ikev2.IKEv2_payload_Proposal]
@@ -1798,12 +1964,18 @@ class TestInitiatorRekey(TestInitiatorPsk):
self.sa.child_sas[0].ispi = prop.SPI
self.sa.child_sas[0].rspi = prop.SPI
self.sa.calc_child_keys()
- header = ikev2.IKEv2(init_SPI=self.sa.ispi, resp_SPI=self.sa.rspi,
- flags='Response', exch_type=36,
- id=ih.id, next_payload='Encrypted')
- resp = self.encrypt_ike_msg(header, sa, 'SA')
- packet = self.create_packet(self.pg0, resp, self.sa.sport,
- self.sa.dport, self.sa.natt, self.ip6)
+ header = ikev2.IKEv2(
+ init_SPI=self.sa.ispi,
+ resp_SPI=self.sa.rspi,
+ flags="Response",
+ exch_type=36,
+ id=ih.id,
+ next_payload="Encrypted",
+ )
+ resp = self.encrypt_ike_msg(header, sa, "SA")
+ packet = self.create_packet(
+ self.pg0, resp, self.sa.sport, self.sa.dport, self.sa.natt, self.ip6
+ )
self.send_and_assert_no_replies(self.pg0, packet)
def test_initiator(self):
@@ -1815,45 +1987,51 @@ class TestInitiatorRekey(TestInitiatorPsk):
@tag_fixme_vpp_workers
class TestInitiatorDelSAFromResponder(TemplateInitiator, Ikev2Params):
- """ test ikev2 initiator - delete IKE SA from responder """
+ """test ikev2 initiator - delete IKE SA from responder"""
def config_tc(self):
- self.config_params({
- 'del_sa_from_responder': True,
- 'is_initiator': False, # seen from test case perspective
- # thus vpp is initiator
- 'responder': {'sw_if_index': self.pg0.sw_if_index,
- 'addr': self.pg0.remote_ip4},
- 'ike-crypto': ('AES-GCM-16ICV', 32),
- 'ike-integ': 'NULL',
- 'ike-dh': '3072MODPgr',
- 'ike_transforms': {
- 'crypto_alg': 20, # "aes-gcm-16"
- 'crypto_key_size': 256,
- 'dh_group': 15, # "modp-3072"
- },
- 'esp_transforms': {
- 'crypto_alg': 12, # "aes-cbc"
- 'crypto_key_size': 256,
- # "hmac-sha2-256-128"
- 'integ_alg': 12},
- 'no_idr_in_auth': True})
+ self.config_params(
+ {
+ "del_sa_from_responder": True,
+ "is_initiator": False, # seen from test case perspective
+ # thus vpp is initiator
+ "responder": {
+ "sw_if_index": self.pg0.sw_if_index,
+ "addr": self.pg0.remote_ip4,
+ },
+ "ike-crypto": ("AES-GCM-16ICV", 32),
+ "ike-integ": "NULL",
+ "ike-dh": "3072MODPgr",
+ "ike_transforms": {
+ "crypto_alg": 20, # "aes-gcm-16"
+ "crypto_key_size": 256,
+ "dh_group": 15, # "modp-3072"
+ },
+ "esp_transforms": {
+ "crypto_alg": 12, # "aes-cbc"
+ "crypto_key_size": 256,
+ # "hmac-sha2-256-128"
+ "integ_alg": 12,
+ },
+ "no_idr_in_auth": True,
+ }
+ )
@tag_fixme_vpp_workers
class TestResponderInitBehindNATT(TemplateResponder, Ikev2Params):
- """ test ikev2 responder - initiator behind NAT """
+ """test ikev2 responder - initiator behind NAT"""
- IKE_NODE_SUFFIX = 'ip4-natt'
+ IKE_NODE_SUFFIX = "ip4-natt"
def config_tc(self):
- self.config_params(
- {'i_natt': True})
+ self.config_params({"i_natt": True})
@tag_fixme_vpp_workers
class TestResponderPsk(TemplateResponder, Ikev2Params):
- """ test ikev2 responder - pre shared key auth """
+ """test ikev2 responder - pre shared key auth"""
+
def config_tc(self):
self.config_params()
@@ -1863,8 +2041,9 @@ class TestResponderDpd(TestResponderPsk):
"""
Dead peer detection test
"""
+
def config_tc(self):
- self.config_params({'dpd_disabled': False})
+ self.config_params({"dpd_disabled": False})
def tearDown(self):
pass
@@ -1879,7 +2058,7 @@ class TestResponderDpd(TestResponderPsk):
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'')
+ self.assertEqual(plain, b"")
# wait for SA expiration
time.sleep(3)
ike_sas = self.vapi.ikev2_sa_dump()
@@ -1890,7 +2069,7 @@ class TestResponderDpd(TestResponderPsk):
@tag_fixme_vpp_workers
class TestResponderRekey(TestResponderPsk):
- """ test ikev2 responder - rekey """
+ """test ikev2 responder - rekey"""
def rekey_from_initiator(self):
packet = self.create_rekey_request()
@@ -1912,18 +2091,19 @@ class TestResponderRekey(TestResponderPsk):
self.sa.calc_child_keys()
self.verify_ike_sas()
self.verify_ipsec_sas(is_rekey=True)
- self.assert_counter(1, 'rekey_req', 'ip4')
+ self.assert_counter(1, "rekey_req", "ip4")
r = self.vapi.ikev2_sa_dump()
self.assertEqual(r[0].sa.stats.n_rekey_req, 1)
class TestResponderVrf(TestResponderPsk, Ikev2Params):
- """ test ikev2 responder - non-default table id """
+ """test ikev2 responder - non-default table id"""
@classmethod
def setUpClass(cls):
import scapy.contrib.ikev2 as _ikev2
- globals()['ikev2'] = _ikev2
+
+ globals()["ikev2"] = _ikev2
super(IkePeer, cls).setUpClass()
cls.create_pg_interfaces(range(1))
cls.vapi.cli("ip table add 1")
@@ -1936,7 +2116,7 @@ class TestResponderVrf(TestResponderPsk, Ikev2Params):
i.resolve_ndp()
def config_tc(self):
- self.config_params({'dpd_disabled': False})
+ self.config_params({"dpd_disabled": False})
def test_responder(self):
self.vapi.ikev2_profile_set_liveness(period=2, max_retries=1)
@@ -1947,53 +2127,67 @@ class TestResponderVrf(TestResponderPsk, Ikev2Params):
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'')
+ self.assertEqual(plain, b"")
@tag_fixme_vpp_workers
class TestResponderRsaSign(TemplateResponder, Ikev2Params):
- """ test ikev2 responder - cert based auth """
+ """test ikev2 responder - cert based auth"""
+
def config_tc(self):
- self.config_params({
- 'udp_encap': True,
- 'auth': 'rsa-sig',
- 'server-key': 'server-key.pem',
- 'client-key': 'client-key.pem',
- 'client-cert': 'client-cert.pem',
- 'server-cert': 'server-cert.pem'})
+ self.config_params(
+ {
+ "udp_encap": True,
+ "auth": "rsa-sig",
+ "server-key": "server-key.pem",
+ "client-key": "client-key.pem",
+ "client-cert": "client-cert.pem",
+ "server-cert": "server-cert.pem",
+ }
+ )
@tag_fixme_vpp_workers
-class Test_IKE_AES_CBC_128_SHA256_128_MODP2048_ESP_AES_CBC_192_SHA_384_192\
- (TemplateResponder, Ikev2Params):
+class Test_IKE_AES_CBC_128_SHA256_128_MODP2048_ESP_AES_CBC_192_SHA_384_192(
+ TemplateResponder, Ikev2Params
+):
"""
IKE:AES_CBC_128_SHA256_128,DH=modp2048 ESP:AES_CBC_192_SHA_384_192
"""
+
def config_tc(self):
- self.config_params({
- 'ike-crypto': ('AES-CBC', 16),
- 'ike-integ': 'SHA2-256-128',
- 'esp-crypto': ('AES-CBC', 24),
- 'esp-integ': 'SHA2-384-192',
- 'ike-dh': '2048MODPgr',
- 'nonce': os.urandom(256),
- 'no_idr_in_auth': True})
+ self.config_params(
+ {
+ "ike-crypto": ("AES-CBC", 16),
+ "ike-integ": "SHA2-256-128",
+ "esp-crypto": ("AES-CBC", 24),
+ "esp-integ": "SHA2-384-192",
+ "ike-dh": "2048MODPgr",
+ "nonce": os.urandom(256),
+ "no_idr_in_auth": True,
+ }
+ )
@tag_fixme_vpp_workers
-class TestAES_CBC_128_SHA256_128_MODP3072_ESP_AES_GCM_16\
- (TemplateResponder, Ikev2Params):
+class TestAES_CBC_128_SHA256_128_MODP3072_ESP_AES_GCM_16(
+ TemplateResponder, Ikev2Params
+):
"""
IKE:AES_CBC_128_SHA256_128,DH=modp3072 ESP:AES_GCM_16
"""
+
def config_tc(self):
- self.config_params({
- 'ike-crypto': ('AES-CBC', 32),
- 'ike-integ': 'SHA2-256-128',
- 'esp-crypto': ('AES-GCM-16ICV', 32),
- 'esp-integ': 'NULL',
- 'ike-dh': '3072MODPgr'})
+ self.config_params(
+ {
+ "ike-crypto": ("AES-CBC", 32),
+ "ike-integ": "SHA2-256-128",
+ "esp-crypto": ("AES-GCM-16ICV", 32),
+ "esp-integ": "NULL",
+ "ike-dh": "3072MODPgr",
+ }
+ )
@tag_fixme_vpp_workers
@@ -2002,20 +2196,21 @@ class Test_IKE_AES_GCM_16_256(TemplateResponder, Ikev2Params):
IKE:AES_GCM_16_256
"""
- IKE_NODE_SUFFIX = 'ip6'
+ IKE_NODE_SUFFIX = "ip6"
def config_tc(self):
- self.config_params({
- 'del_sa_from_responder': True,
- 'ip6': True,
- 'natt': True,
- 'ike-crypto': ('AES-GCM-16ICV', 32),
- 'ike-integ': 'NULL',
- 'ike-dh': '2048MODPgr',
- 'loc_ts': {'start_addr': 'ab:cd::0',
- 'end_addr': 'ab:cd::10'},
- 'rem_ts': {'start_addr': '11::0',
- 'end_addr': '11::100'}})
+ self.config_params(
+ {
+ "del_sa_from_responder": True,
+ "ip6": True,
+ "natt": True,
+ "ike-crypto": ("AES-GCM-16ICV", 32),
+ "ike-integ": "NULL",
+ "ike-dh": "2048MODPgr",
+ "loc_ts": {"start_addr": "ab:cd::0", "end_addr": "ab:cd::10"},
+ "rem_ts": {"start_addr": "11::0", "end_addr": "11::100"},
+ }
+ )
@tag_fixme_vpp_workers
@@ -2033,8 +2228,8 @@ class TestInitiatorKeepaliveMsg(TestInitiatorPsk):
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'')
- self.assert_counter(1, 'keepalive', 'ip4')
+ self.assertEqual(plain, b"")
+ self.assert_counter(1, "keepalive", "ip4")
r = self.vapi.ikev2_sa_dump()
self.assertEqual(1, r[0].sa.stats.n_keepalives)
@@ -2044,7 +2239,7 @@ class TestInitiatorKeepaliveMsg(TestInitiatorPsk):
class TestMalformedMessages(TemplateResponder, Ikev2Params):
- """ malformed packet test """
+ """malformed packet test"""
def tearDown(self):
pass
@@ -2053,23 +2248,26 @@ class TestMalformedMessages(TemplateResponder, Ikev2Params):
self.config_params()
def create_ike_init_msg(self, length=None, payload=None):
- msg = ikev2.IKEv2(length=length, init_SPI='\x11' * 8,
- flags='Initiator', exch_type='IKE_SA_INIT')
+ msg = ikev2.IKEv2(
+ length=length,
+ init_SPI="\x11" * 8,
+ flags="Initiator",
+ exch_type="IKE_SA_INIT",
+ )
if payload is not None:
msg /= payload
- return self.create_packet(self.pg0, msg, self.sa.sport,
- self.sa.dport)
+ return self.create_packet(self.pg0, msg, self.sa.sport, self.sa.dport)
def verify_bad_packet_length(self):
- ike_msg = self.create_ike_init_msg(length=0xdead)
+ ike_msg = self.create_ike_init_msg(length=0xDEAD)
self.send_and_assert_no_replies(self.pg0, ike_msg * self.pkt_count)
- self.assert_counter(self.pkt_count, 'bad_length')
+ self.assert_counter(self.pkt_count, "bad_length")
def verify_bad_sa_payload_length(self):
- p = ikev2.IKEv2_payload_SA(length=0xdead)
+ p = ikev2.IKEv2_payload_SA(length=0xDEAD)
ike_msg = self.create_ike_init_msg(payload=p)
self.send_and_assert_no_replies(self.pg0, ike_msg * self.pkt_count)
- self.assert_counter(self.pkt_count, 'malformed_packet')
+ self.assert_counter(self.pkt_count, "malformed_packet")
def test_responder(self):
self.pkt_count = 254
@@ -2077,5 +2275,5 @@ class TestMalformedMessages(TemplateResponder, Ikev2Params):
self.verify_bad_sa_payload_length()
-if __name__ == '__main__':
+if __name__ == "__main__":
unittest.main(testRunner=VppTestRunner)