| import unittest |
| import struct |
| import errno |
| import json |
| import os |
| |
| from pyverbs.providers.mlx5.mlx5dv_mkey import Mlx5Mkey, Mlx5MrInterleaved, \ |
| Mlx5MkeyConfAttr, Mlx5SigCrc, Mlx5SigBlockDomain, Mlx5SigBlockAttr |
| from pyverbs.providers.mlx5.mlx5dv_crypto import Mlx5CryptoLoginAttr, Mlx5DEK, \ |
| Mlx5DEKInitAttr, Mlx5CryptoAttr, Mlx5CryptoLogin, Mlx5CryptoExtLoginAttr |
| from pyverbs.providers.mlx5.mlx5dv import Mlx5Context, Mlx5DVContextAttr, \ |
| Mlx5DVQPInitAttr, Mlx5QP |
| from pyverbs.pyverbs_error import PyverbsRDMAError, PyverbsUserError |
| from tests.mlx5_base import Mlx5RDMATestCase, Mlx5PyverbsAPITestCase |
| import pyverbs.providers.mlx5.mlx5_enums as dve |
| from pyverbs.wr import SGE, SendWR, RecvWR |
| from pyverbs.qp import QPInitAttrEx, QPCap |
| from tests.base import RCResources |
| from pyverbs.pd import PD |
| import pyverbs.enums as e |
| import tests.utils as u |
| |
| DEK_OPAQUE = b'dek' |
| |
| """ |
| Crytpo operation requires specific input from the user, e.g. the wrapped credential that the |
| device is configured with. The input should be provided in JSON format in a file in this path: |
| "/tmp/mlx5_crypto_test.txt". |
| User can also set this environment variable with his file path: MLX5_CRYPTO_TEST_INFO |
| |
| This doc describes the input option with examples to the format: |
| Mandatory: |
| Wrapped credential: The credential that was configured in the device wrapped. |
| Wrapped key: Wrapped key for the DEK creation. The key should be encrypted using the KEK |
| (Key Encrypted Key). |
| Optional: |
| Wrapped 256 bits key: Wrapped key for the DEK creation when the key size of 256 bits is required. |
| If not provided, that test will only use the key in size of 128 bits. |
| Encrypted data for 512 of 'c': If a user wants to have data validation, he needs to provide |
| expected encrypted data for a plain text of 512 bytes of the |
| character 'c'. If not provided, data validation will be skipped. |
| |
| Example of content of such file: |
| [{"credential": [8704278040424473809, 4403447855848063568, 13892768337045135232, |
| 5942481448427925932, 171338997253969038, 5703425261028721211], |
| "wrapped_key": [contains 5 integers of 64bits each], |
| "plaintext_key": [contains 8 integers of 32bits each], |
| "plaintext_256_bits_key": [contains 16 integers of 32bits each], |
| "encrypted_data_for_512_c": [contains 64 integers of 64bits each], |
| "wrapped_256_bits_key": [contains 9 integers of 64bits each]}] |
| """ |
| |
| |
| def check_crypto_caps(dev_name, is_wrapped_dek_mode, multi_block_support=False): |
| """ |
| Check that this device support crypto actions. |
| :param dev_name: The device name. |
| :param is_wrapped_dek_mode: True when wrapped_dek and False when plaintext. |
| :param multi_block_support: If True, check for multi-block support. |
| """ |
| mlx5dv_attr = Mlx5DVContextAttr() |
| ctx = Mlx5Context(mlx5dv_attr, name=dev_name) |
| crypto_caps = ctx.query_mlx5_device().crypto_caps |
| failed_selftests = crypto_caps['failed_selftests'] |
| if failed_selftests: |
| raise unittest.SkipTest(f'The device crypto selftest failed ({failed_selftests})') |
| single_block_cap = dve.MLX5DV_CRYPTO_ENGINES_CAP_AES_XTS_SINGLE_BLOCK \ |
| & crypto_caps['crypto_engines'] |
| multi_block_cap = dve.MLX5DV_CRYPTO_ENGINES_CAP_AES_XTS_MULTI_BLOCK \ |
| & crypto_caps['crypto_engines'] |
| if not single_block_cap and not multi_block_cap: |
| raise unittest.SkipTest('The device crypto engines does not support AES') |
| elif multi_block_support and not multi_block_cap: |
| raise unittest.SkipTest('The device crypto engines does not support multi blocks') |
| elif multi_block_cap: |
| assert single_block_cap, \ |
| 'The device crypto engines do not support single block but support multi blocks' |
| dev_wrapped_import_method = crypto_caps['wrapped_import_method'] |
| if is_wrapped_dek_mode: |
| if not dve.MLX5DV_CRYPTO_WRAPPED_IMPORT_METHOD_CAP_AES_XTS & dev_wrapped_import_method or \ |
| not dve.MLX5DV_CRYPTO_CAPS_WRAPPED_CRYPTO_OPERATIONAL & crypto_caps['flags']: |
| raise unittest.SkipTest('The device does not support wrapped DEK') |
| elif dve.MLX5DV_CRYPTO_WRAPPED_IMPORT_METHOD_CAP_AES_XTS & dev_wrapped_import_method and \ |
| dve.MLX5DV_CRYPTO_CAPS_WRAPPED_CRYPTO_OPERATIONAL & crypto_caps['flags']: |
| raise unittest.SkipTest('The device does not support plaintext DEK') |
| |
| |
| def require_crypto_login_details(instance): |
| """ |
| Parse the crypto login session details from this file: |
| '/tmp/mlx5_crypto_test.txt' |
| If the file doesn't exists or the content is not in Json format, skip the test. |
| :param instance: The test instance. |
| """ |
| crypto_file = '/tmp/mlx5_crypto_test.txt' |
| if 'MLX5_CRYPTO_TEST_INFO' in os.environ: |
| crypto_file = os.environ['MLX5_CRYPTO_TEST_INFO'] |
| try: |
| with open(crypto_file, 'r') as f: |
| test_details = f.read() |
| setattr(instance, 'crypto_details', test_details) |
| json.loads(test_details) |
| instance.crypto_details = json.loads(test_details)[0] |
| except json.JSONDecodeError: |
| raise unittest.SkipTest(f'The crypto data in {crypto_file} must be in Json format') |
| except FileNotFoundError: |
| raise unittest.SkipTest(f'Crypto login details must be supplied in {crypto_file}') |
| |
| |
| def requires_crypto_support(is_wrapped_dek_mode, multi_block_support=False): |
| """ |
| :param is_wrapped_dek_mode: True when wapped_dek and False when plaintext. |
| :param multi_block_support: If True, check for multi-block support. |
| """ |
| def outer(func): |
| def inner(instance): |
| require_crypto_login_details(instance) |
| check_crypto_caps(instance.dev_name, is_wrapped_dek_mode, multi_block_support) |
| return func(instance) |
| return inner |
| return outer |
| |
| |
| class Mlx5CryptoResources(RCResources): |
| def __init__(self, dev_name, ib_port, gid_index, dv_send_ops_flags=0, |
| mkey_create_flags=dve.MLX5DV_MKEY_INIT_ATTR_FLAGS_INDIRECT, |
| msg_size=1024): |
| self.dv_send_ops_flags = dv_send_ops_flags |
| self.mkey_create_flags = mkey_create_flags |
| self.max_inline_data = 512 |
| self.send_ops_flags = e.IBV_QP_EX_WITH_SEND |
| super().__init__(dev_name, ib_port, gid_index, msg_size=msg_size) |
| self.create_mkeys() |
| |
| def create_mkeys(self): |
| try: |
| self.wire_enc_mkey = Mlx5Mkey(self.pd, self.mkey_create_flags, max_entries=1) |
| self.mem_enc_mkey = Mlx5Mkey(self.pd, self.mkey_create_flags, max_entries=1) |
| except PyverbsRDMAError as ex: |
| if ex.error_code == errno.EOPNOTSUPP: |
| raise unittest.SkipTest('Create Mkey is not supported') |
| raise ex |
| |
| def create_qp_cap(self): |
| return QPCap(max_send_wr=self.num_msgs, max_recv_wr=self.num_msgs, |
| max_inline_data=self.max_inline_data) |
| |
| def create_qp_init_attr(self): |
| comp_mask = e.IBV_QP_INIT_ATTR_PD | e.IBV_QP_INIT_ATTR_SEND_OPS_FLAGS |
| return QPInitAttrEx(cap=self.create_qp_cap(), pd=self.pd, scq=self.cq, |
| rcq=self.cq, qp_type=e.IBV_QPT_RC, |
| send_ops_flags=self.send_ops_flags, |
| comp_mask=comp_mask) |
| |
| def create_qps(self): |
| try: |
| qp_init_attr = self.create_qp_init_attr() |
| comp_mask = dve.MLX5DV_QP_INIT_ATTR_MASK_SEND_OPS_FLAGS |
| attr = Mlx5DVQPInitAttr(comp_mask=comp_mask, |
| send_ops_flags=self.dv_send_ops_flags) |
| qp = Mlx5QP(self.ctx, qp_init_attr, attr) |
| self.qps.append(qp) |
| self.qps_num.append(qp.qp_num) |
| self.psns.append(0) |
| except PyverbsRDMAError as ex: |
| if ex.error_code == errno.EOPNOTSUPP: |
| raise unittest.SkipTest('Create Mlx5DV QP is not supported') |
| raise ex |
| |
| |
| class Mlx5CryptoAPITest(Mlx5PyverbsAPITestCase): |
| def __init__(self, methodName='runTest'): |
| super().__init__(methodName) |
| self.crypto_details = None |
| |
| def verify_create_dek_out_of_login_session(self): |
| """ |
| Verify that create DEK out of crypto login session is not permited. |
| """ |
| with self.assertRaises(PyverbsRDMAError) as ex: |
| Mlx5DEK(self.ctx, self.dek_init_attr) |
| self.assertEqual(ex.exception.error_code, errno.EINVAL) |
| |
| def verify_login_state(self, expected_state): |
| """ |
| Query the session login state and verify that it's as expected. |
| """ |
| state = Mlx5Context.query_login_state(self.ctx) |
| self.assertEqual(state, expected_state) |
| |
| def verify_login_twice(self): |
| """ |
| Verify that when there is already a login session alive the second login |
| fails. |
| """ |
| with self.assertRaises(PyverbsRDMAError) as ex: |
| Mlx5Context.crypto_login(self.ctx, self.login_attr) |
| self.assertEqual(ex.exception.error_code, errno.EEXIST) |
| |
| def verify_dek_opaque(self): |
| """ |
| Query the DEK and verify that its opaque is as expected. |
| """ |
| dek_attr = self.dek.query() |
| self.assertEqual(dek_attr.opaque, DEK_OPAQUE) |
| |
| @requires_crypto_support(is_wrapped_dek_mode=True) |
| def test_mlx5_dek_management(self): |
| """ |
| Test crypto login and DEK management APIs. |
| The test checks also that invalid actions are not permited, e.g, create |
| DEK not in login session on wrapped DEK mode. |
| """ |
| try: |
| self.pd = PD(self.ctx) |
| cred_bytes = struct.pack('!6Q', *self.crypto_details['credential']) |
| key = struct.pack('!5Q', *self.crypto_details['wrapped_key']) |
| self.dek_init_attr = \ |
| Mlx5DEKInitAttr(self.pd, key=key, |
| key_size=dve.MLX5DV_CRYPTO_KEY_SIZE_128, |
| key_purpose=dve.MLX5DV_CRYPTO_KEY_PURPOSE_AES_XTS, |
| opaque=DEK_OPAQUE) |
| self.verify_create_dek_out_of_login_session() |
| self.verify_login_state(dve.MLX5DV_CRYPTO_LOGIN_STATE_NO_LOGIN) |
| |
| # Login to crypto session |
| self.login_attr = Mlx5CryptoLoginAttr(cred_bytes) |
| Mlx5Context.crypto_login(self.ctx, self.login_attr) |
| self.verify_login_state(dve.MLX5DV_CRYPTO_LOGIN_STATE_VALID) |
| self.verify_login_twice() |
| self.dek = Mlx5DEK(self.ctx, self.dek_init_attr) |
| self.verify_dek_opaque() |
| self.dek.close() |
| |
| # Logout from crypto session |
| Mlx5Context.crypto_logout(self.ctx) |
| self.verify_login_state(dve.MLX5DV_CRYPTO_LOGIN_STATE_NO_LOGIN) |
| except PyverbsRDMAError as ex: |
| print(ex) |
| if ex.error_code == errno.EOPNOTSUPP: |
| raise unittest.SkipTest('Create crypto elements is not supported') |
| raise ex |
| |
| |
| class Mlx5CryptoTrafficTest(Mlx5RDMATestCase): |
| """ |
| Test the mlx5 cryto APIs. |
| """ |
| def setUp(self): |
| super().setUp() |
| self.iters = 10 |
| self.crypto_details = None |
| self.validate_data = False |
| self.is_multi_block = False |
| self.msg_size = 1024 |
| self.key_size = dve.MLX5DV_CRYPTO_KEY_SIZE_128 |
| |
| def create_client_dek(self): |
| """ |
| Create DEK using the client resources. |
| """ |
| cred_bytes = struct.pack('!6Q', *self.crypto_details['credential']) |
| log_attr = Mlx5CryptoLoginAttr(cred_bytes) |
| Mlx5Context.crypto_login(self.client.ctx, log_attr) |
| self._create_wrapped_dek() |
| |
| def create_client_wrapped_dek_login_obj(self): |
| """ |
| Create CryptoExtLoginAttr and crypto login object |
| """ |
| cred_bytes = struct.pack('!6Q', *self.crypto_details['credential']) |
| log_attr = Mlx5CryptoExtLoginAttr(cred_bytes, 48) |
| crypto_login_obj = Mlx5CryptoLogin(self.client.ctx, log_attr) |
| comp_mask=dve.MLX5DV_DEK_INIT_ATTR_CRYPTO_LOGIN |
| self._create_wrapped_dek(comp_mask, crypto_login_obj) |
| |
| def _create_wrapped_dek(self, comp_mask=0, crypto_login_obj=None): |
| """ |
| Create wrapped DEK using the client resources. |
| """ |
| key = struct.pack('!5Q', *self.crypto_details['wrapped_key']) |
| if self.key_size == dve.MLX5DV_CRYPTO_KEY_SIZE_256: |
| key = struct.pack('!9Q', *self.crypto_details['wrapped_256_bits_key']) |
| |
| if comp_mask == dve.MLX5DV_DEK_INIT_ATTR_CRYPTO_LOGIN: |
| self.dek_attr = Mlx5DEKInitAttr(self.client.pd, key=key, |
| key_size=self.key_size, |
| key_purpose=dve.MLX5DV_CRYPTO_KEY_PURPOSE_AES_XTS, |
| comp_mask=comp_mask, crypto_login=crypto_login_obj) |
| else: |
| self.dek_attr = Mlx5DEKInitAttr(self.client.pd, key=key, |
| key_size=self.key_size, |
| key_purpose=dve.MLX5DV_CRYPTO_KEY_PURPOSE_AES_XTS) |
| |
| self.dek = Mlx5DEK(self.client.ctx, self.dek_attr) |
| |
| def create_client_plaintext_dek(self): |
| """ |
| Create DEK using the client resources. |
| """ |
| key = struct.pack('8I',*self.crypto_details['plaintext_key']) |
| if self.key_size == dve.MLX5DV_CRYPTO_KEY_SIZE_256: |
| key = struct.pack('16I', *self.crypto_details['plaintext_256_bits_key']) |
| self.dek_attr = Mlx5DEKInitAttr(self.client.pd,key=key, |
| key_size=self.key_size, |
| comp_mask=dve.MLX5DV_DEK_INIT_ATTR_CRYPTO_LOGIN, |
| crypto_login=None) |
| self.dek = Mlx5DEK(self.client.ctx, self.dek_attr) |
| |
| def reg_client_mkey(self, signature=False): |
| """ |
| Configure an mkey with crypto attributes. |
| :param signature: True if signature configuration requested. |
| """ |
| num_of_configuration = 4 if signature else 3 |
| for mkey in [self.client.wire_enc_mkey, self.client.mem_enc_mkey]: |
| self.client.qp.wr_start() |
| self.client.qp.wr_flags = e.IBV_SEND_SIGNALED | e.IBV_SEND_INLINE |
| offset = 0 if mkey == self.client.wire_enc_mkey else self.client.msg_size/2 |
| sge = SGE(self.client.mr.buf + offset, self.client.msg_size/2, self.client.mr.lkey) |
| self.client.qp.wr_mkey_configure(mkey, num_of_configuration, Mlx5MkeyConfAttr()) |
| self.client.qp.wr_set_mkey_access_flags(e.IBV_ACCESS_LOCAL_WRITE) |
| self.client.qp.wr_set_mkey_layout_list([sge]) |
| if signature: |
| self.configure_mkey_signature() |
| initial_tweak = struct.pack('!2Q', int(0), int(0)) |
| encrypt_on_tx = mkey == self.client.wire_enc_mkey |
| sign_crypto_order = dve.MLX5DV_SIGNATURE_CRYPTO_ORDER_SIGNATURE_BEFORE_CRYPTO_ON_TX |
| crypto_attr = Mlx5CryptoAttr(crypto_standard=dve.MLX5DV_CRYPTO_STANDARD_AES_XTS, |
| encrypt_on_tx=encrypt_on_tx, |
| signature_crypto_order=sign_crypto_order, |
| data_unit_size=dve.MLX5DV_BLOCK_SIZE_512, |
| dek=self.dek, initial_tweak=initial_tweak) |
| self.client.qp.wr_set_mkey_crypto(crypto_attr) |
| self.client.qp.wr_complete() |
| u.poll_cq(self.client.cq) |
| |
| def configure_mkey_signature(self): |
| """ |
| Configure an mkey with signature attributes. |
| """ |
| sig_crc = Mlx5SigCrc(crc_type=dve.MLX5DV_SIG_CRC_TYPE_CRC32, seed=0xFFFFFFFF) |
| sig_block_domain = Mlx5SigBlockDomain(sig_type=dve.MLX5DV_SIG_TYPE_CRC, crc=sig_crc, |
| block_size=dve.MLX5DV_BLOCK_SIZE_512) |
| sig_attr = Mlx5SigBlockAttr(wire=sig_block_domain, |
| check_mask=dve.MLX5DV_SIG_MASK_CRC32) |
| self.client.qp.wr_set_mkey_sig_block(sig_attr) |
| |
| def get_send_wr(self, player, wire_encryption): |
| mkey = player.wire_enc_mkey if wire_encryption else player.mem_enc_mkey |
| sge = SGE(0, player.msg_size/2, mkey.lkey) |
| return SendWR(opcode=e.IBV_WR_SEND, num_sge=1, sg=[sge]) |
| |
| def get_recv_wr(self, player, wire_encryption): |
| offset = 0 if wire_encryption else player.msg_size/2 |
| sge = SGE(player.mr.buf + offset, player.msg_size/2, player.mr.lkey) |
| return RecvWR(sg=[sge], num_sge=1) |
| |
| def prepare_validate_data(self): |
| data_size = int(self.msg_size / 2) |
| self.client.mr.write('c' * data_size, data_size) |
| if self.is_multi_block: |
| encrypted_data = struct.pack('!128Q', *self.crypto_details['encrypted_data_for_1024_c']) |
| else: |
| encrypted_data = struct.pack('!64Q', *self.crypto_details['encrypted_data_for_512_c']) |
| self.client.mr.write(encrypted_data, data_size, offset=data_size) |
| |
| def validate_crypto_data(self): |
| """ |
| Validate the server MR data. Verify that the encryption/decryption works well. |
| """ |
| data_size= int(self.msg_size / 2) |
| send_msg = self.client.mr.read(self.msg_size, 0) |
| recv_msg = self.server.mr.read(self.msg_size, 0) |
| self.assertEqual(send_msg[0:data_size], recv_msg[data_size:self.msg_size]) |
| self.assertEqual(send_msg[data_size:self.msg_size], recv_msg[0:data_size]) |
| |
| def init_data_validation(self): |
| if 'encrypted_data_for_512_c' in self.crypto_details and not self.is_multi_block: |
| self.validate_data = True |
| elif 'encrypted_data_for_1024_c' in self.crypto_details and self.is_multi_block: |
| self.validate_data = True |
| |
| def traffic(self): |
| """ |
| Perform RC traffic using the configured mkeys. |
| """ |
| if self.validate_data: |
| self.prepare_validate_data() |
| for _ in range(self.iters): |
| self.server.qp.post_recv(self.get_recv_wr(self.server, wire_encryption=True)) |
| self.server.qp.post_recv(self.get_recv_wr(self.server, wire_encryption=False)) |
| self.client.qp.post_send(self.get_send_wr(self.client, wire_encryption=True)) |
| self.client.qp.post_send(self.get_send_wr(self.client, wire_encryption=False)) |
| u.poll_cq(self.client.cq, count=2) |
| u.poll_cq(self.server.cq, count=2) |
| if self.validate_data: |
| self.validate_crypto_data() |
| |
| def run_crypto_dek_test(self, create_dek_func): |
| """ |
| Creates player and DEK, using the give function, for different test |
| cases, and runs traffic. |
| :param create_dek_func: Function that creates a DEK. |
| """ |
| self.init_data_validation() |
| mkey_flags = dve.MLX5DV_MKEY_INIT_ATTR_FLAGS_CRYPTO | \ |
| dve.MLX5DV_MKEY_INIT_ATTR_FLAGS_INDIRECT |
| self.create_players(Mlx5CryptoResources, |
| dv_send_ops_flags=dve.MLX5DV_QP_EX_WITH_MKEY_CONFIGURE, |
| mkey_create_flags=mkey_flags, msg_size=self.msg_size) |
| create_dek_func() |
| self.reg_client_mkey() |
| self.traffic() |
| |
| @requires_crypto_support(is_wrapped_dek_mode=True) |
| def test_mlx5_crypto_mkey_old_api(self): |
| """ |
| Create Mkeys, register a memory layout using the mkeys, configure |
| crypto attributes on it and then run traffic. |
| """ |
| self.run_crypto_dek_test(self.create_client_dek) |
| |
| @requires_crypto_support(is_wrapped_dek_mode=True) |
| def test_mlx5_crypto_wrapped_dek(self): |
| """ |
| Create Mkeys, register a memory layout using the mkeys, |
| configure crypto attributes on it using login object API and then run |
| traffic. |
| Use wrapped DEK with new API. |
| """ |
| self.run_crypto_dek_test(self.create_client_wrapped_dek_login_obj) |
| |
| @requires_crypto_support(is_wrapped_dek_mode=False) |
| def test_mlx5_crypto_plaintext_dek(self): |
| """ |
| Create Mkeys, register a memory layout using the mkeys, |
| configure crypto attributes on it using login object API and then run |
| traffic. |
| Use plaintext DEK. |
| """ |
| self.run_crypto_dek_test(self.create_client_plaintext_dek) |
| |
| @requires_crypto_support(is_wrapped_dek_mode=True) |
| def test_mlx5_crypto_signature_mkey(self): |
| """ |
| Create Mkeys, register a memory layout using this mkey, configure |
| crypto and signature attributes on it and then perform traffic using |
| this mkey. |
| """ |
| if 'wrapped_256_bits_key' in self.crypto_details: |
| self.key_size = dve.MLX5DV_CRYPTO_KEY_SIZE_256 |
| mkey_flags = dve.MLX5DV_MKEY_INIT_ATTR_FLAGS_CRYPTO | \ |
| dve.MLX5DV_MKEY_INIT_ATTR_FLAGS_INDIRECT | \ |
| dve.MLX5DV_MKEY_INIT_ATTR_FLAGS_BLOCK_SIGNATURE |
| self.create_players(Mlx5CryptoResources, |
| dv_send_ops_flags=dve.MLX5DV_QP_EX_WITH_MKEY_CONFIGURE, |
| mkey_create_flags=mkey_flags) |
| self.create_client_dek() |
| self.reg_client_mkey(signature=True) |
| self.traffic() |
| |
| @requires_crypto_support(is_wrapped_dek_mode=False, multi_block_support=True) |
| def test_mlx5_plaintext_dek_multi_block(self): |
| self.is_multi_block = True |
| self.msg_size = 2048 |
| self.run_crypto_dek_test(self.create_client_plaintext_dek) |
| |
| @requires_crypto_support(is_wrapped_dek_mode=True, multi_block_support=True) |
| def test_mlx5_wrapped_dek_multi_block(self): |
| self.is_multi_block = True |
| self.msg_size = 2048 |
| self.run_crypto_dek_test(self.create_client_wrapped_dek_login_obj) |
