def encrypt_dynamodb_item()

in src/dynamodb_encryption_sdk/encrypted/item.py [0:0]

• 44 lines of code
• 10 McCabe index (conditional complexity)

def encrypt_dynamodb_item(item, crypto_config):
    # type: (dynamodb_types.ITEM, CryptoConfig) -> dynamodb_types.ITEM
    """Encrypt a DynamoDB item.

    >>> from dynamodb_encryption_sdk.encrypted.item import encrypt_dynamodb_item
    >>> plaintext_item = {
    ...     'some': {'S': 'data'},
    ...     'more': {'N': '5'}
    ... }
    >>> encrypted_item = encrypt_dynamodb_item(
    ...     item=plaintext_item,
    ...     crypto_config=my_crypto_config
    ... )

    .. note::

        This handles DynamoDB-formatted items and is for use with the boto3 DynamoDB client.

    :param dict item: Plaintext DynamoDB item
    :param CryptoConfig crypto_config: Cryptographic configuration
    :returns: Encrypted and signed DynamoDB item
    :rtype: dict
    """
    if crypto_config.attribute_actions.take_no_actions:
        # If we explicitly have been told not to do anything to this item, just copy it.
        return item.copy()

    for reserved_name in ReservedAttributes:
        if reserved_name.value in item:
            raise EncryptionError(
                'Reserved attribute name "{}" is not allowed in plaintext item.'.format(reserved_name.value)
            )

    encryption_materials = crypto_config.encryption_materials()

    inner_material_description = encryption_materials.material_description.copy()
    try:
        encryption_materials.encryption_key
    except AttributeError:
        if crypto_config.attribute_actions.contains_action(CryptoAction.ENCRYPT_AND_SIGN):
            raise EncryptionError(
                "Attribute actions ask for some attributes to be encrypted but no encryption key is available"
            )

        encrypted_item = item.copy()
    else:
        # Add the attribute encryption mode to the inner material description
        encryption_mode = MaterialDescriptionValues.CBC_PKCS5_ATTRIBUTE_ENCRYPTION.value
        inner_material_description[MaterialDescriptionKeys.ATTRIBUTE_ENCRYPTION_MODE.value] = encryption_mode

        algorithm_descriptor = encryption_materials.encryption_key.algorithm + encryption_mode

        encrypted_item = {}
        for name, attribute in item.items():
            if crypto_config.attribute_actions.action(name) is CryptoAction.ENCRYPT_AND_SIGN:
                encrypted_item[name] = encrypt_attribute(
                    attribute_name=name,
                    attribute=attribute,
                    encryption_key=encryption_materials.encryption_key,
                    algorithm=algorithm_descriptor,
                )
            else:
                encrypted_item[name] = attribute.copy()

    signature_attribute = sign_item(encrypted_item, encryption_materials.signing_key, crypto_config)
    encrypted_item[ReservedAttributes.SIGNATURE.value] = signature_attribute

    try:
        # Add the signing key algorithm identifier to the inner material description if provided
        inner_material_description[
            MaterialDescriptionKeys.SIGNING_KEY_ALGORITHM.value
        ] = encryption_materials.signing_key.signing_algorithm()
    except NotImplementedError:
        # Not all signing keys will provide this value
        pass

    material_description_attribute = serialize_material_description(inner_material_description)
    encrypted_item[ReservedAttributes.MATERIAL_DESCRIPTION.value] = material_description_attribute

    return encrypted_item