Function cryptography_symmetric_encrypt

st2common/st2common/util/crypto.py:222–290

Encrypt the provided plaintext using AES encryption. NOTE 1: This function return a string which is fully compatible with Keyczar.Encrypt() method. NOTE 2: This function is loosely based on keyczar AESKey.Encrypt() (Apache 2.0 license). The final encrypted string value consists o

(encrypt_key, plaintext)

def cryptography_symmetric_encrypt(encrypt_key, plaintext):
    """
    Encrypt the provided plaintext using AES encryption.

    NOTE 1: This function return a string which is fully compatible with Keyczar.Encrypt() method.

    NOTE 2: This function is loosely based on keyczar AESKey.Encrypt() (Apache 2.0 license).

    The final encrypted string value consists of:

    [message bytes][HMAC signature bytes for the message] where message consists of
    [keyczar header plaintext][IV bytes][ciphertext bytes]

    NOTE: Header itself is unused, but it's added so the format is compatible with keyczar format.

    """
    if not isinstance(encrypt_key, AESKey):
        raise TypeError(
            "Encrypted key needs to be an AESkey class instance"
            f" (was {type(encrypt_key)})."
        )
    if not isinstance(plaintext, (six.text_type, six.string_types, six.binary_type)):
        raise TypeError(
            "Plaintext needs to either be a string/unicode or bytes"
            f" (was {type(plaintext)})."
        )

    aes_key_bytes = encrypt_key.aes_key_bytes
    hmac_key_bytes = encrypt_key.hmac_key_bytes

    if not isinstance(aes_key_bytes, six.binary_type):
        raise TypeError(f"AESKey is not bytes (it is {type(aes_key_bytes)}).")
    if not isinstance(hmac_key_bytes, six.binary_type):
        raise TypeError(f"HMACKey is not bytes (it is {type(hmac_key_bytes)}).")

    if isinstance(plaintext, (six.text_type, six.string_types)):
        # Convert data to bytes
        data = plaintext.encode("utf-8")
    else:
        data = plaintext

    # Pad data
    data = pkcs5_pad(data)

    # Generate IV
    iv_bytes = os.urandom(KEYCZAR_AES_BLOCK_SIZE)

    backend = default_backend()
    cipher = Cipher(algorithms.AES(aes_key_bytes), modes.CBC(iv_bytes), backend=backend)
    encryptor = cipher.encryptor()

    # NOTE: We don't care about actual Keyczar header value, we only care about the length (5
    # bytes) so we simply add 5 0's
    header_bytes = b"00000"

    ciphertext_bytes = encryptor.update(data) + encryptor.finalize()
    msg_bytes = header_bytes + iv_bytes + ciphertext_bytes