in src/aws_encryption_sdk/__init__.py [0:0]
def stream(self, **kwargs):
"""Provides an :py:func:`open`-like interface to the streaming encryptor/decryptor classes.
.. warning::
Take care when decrypting framed messages with large frame length and large non-framed
messages. In order to protect the authenticity of the encrypted data, no plaintext
is returned until it has been authenticated. Because of this, potentially large amounts
of data may be read into memory. In the case of framed messages, the entire contents
of each frame are read into memory and authenticated before returning any plaintext.
In the case of non-framed messages, the entire message is read into memory and
authenticated before returning any plaintext. The authenticated plaintext is held in
memory until it is requested.
.. note::
Consequently, keep the above decrypting consideration in mind when encrypting messages
to ensure that issues are not encountered when decrypting those messages.
.. code:: python
>>> import boto3
>>> from aws_cryptographic_material_providers.mpl import AwsCryptographicMaterialProviders
>>> from aws_cryptographic_material_providers.mpl.config import MaterialProvidersConfig
>>> from aws_cryptographic_material_providers.mpl.models import CreateAwsKmsKeyringInput
>>> from aws_cryptographic_material_providers.mpl.references import IKeyring
>>> import aws_encryption_sdk
>>> client = aws_encryption_sdk.EncryptionSDKClient()
>>> mat_prov: AwsCryptographicMaterialProviders = AwsCryptographicMaterialProviders(
... config=MaterialProvidersConfig()
... )
>>> keyring_input: CreateAwsKmsKeyringInput = CreateAwsKmsKeyringInput(
... kms_key_id='arn:aws:kms:us-east-1:2222222222222:key/22222222-2222-2222-2222-222222222222',
... kms_client=boto3.client('kms', region_name="us-west-2")
... )
>>> kms_keyring: IKeyring = mat_prov.create_aws_kms_keyring(
... input=keyring_input
... )
>>> plaintext_filename = 'my-secret-data.dat'
>>> ciphertext_filename = 'my-encrypted-data.ct'
>>> with open(plaintext_filename, 'rb') as pt_file, open(ciphertext_filename, 'wb') as ct_file:
... with client.stream(
... mode='e',
... source=pt_file,
... keyring=kms_keyring
... ) as encryptor:
... for chunk in encryptor:
... ct_file.write(chunk)
>>> decrypted_filename = 'my-decrypted-data.dat'
>>> with open(ciphertext_filename, 'rb') as ct_file, open(decrypted_filename, 'wb') as pt_file:
... with client.stream(
... mode='d',
... source=ct_file,
... keyring=kms_keyring
... ) as decryptor:
... for chunk in decryptor:
... pt_file.write(chunk)
:param str mode: Type of streaming client to return (e/encrypt: encryptor, d/decrypt: decryptor)
:param **kwargs: All other parameters provided are passed to the appropriate Streaming client
:returns: Streaming Encryptor or Decryptor, as requested
:rtype: :class:`aws_encryption_sdk.streaming_client.StreamEncryptor`
or :class:`aws_encryption_sdk.streaming_client.StreamDecryptor`
:raises ValueError: if supplied with an unsupported mode value
"""
self._set_config_kwargs("stream", kwargs)
mode = kwargs.pop("mode")
_signature_policy_map = {
"e": SignaturePolicy.ALLOW_ENCRYPT_ALLOW_DECRYPT,
"encrypt": SignaturePolicy.ALLOW_ENCRYPT_ALLOW_DECRYPT,
"d": SignaturePolicy.ALLOW_ENCRYPT_ALLOW_DECRYPT,
"decrypt": SignaturePolicy.ALLOW_ENCRYPT_ALLOW_DECRYPT,
"decrypt-unsigned": SignaturePolicy.ALLOW_ENCRYPT_FORBID_DECRYPT,
}
kwargs["signature_policy"] = _signature_policy_map[mode.lower()]
_stream_map = {
"e": StreamEncryptor,
"encrypt": StreamEncryptor,
"d": StreamDecryptor,
"decrypt": StreamDecryptor,
"decrypt-unsigned": StreamDecryptor,
}
try:
return _stream_map[mode.lower()](**kwargs)
except KeyError:
raise ValueError("Unsupported mode: {}".format(mode))