Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
aio_opened = True
###
if isinstance(data, bytes):
yield data
# Test if data is a file obj first, since it's considered Iterable too
elif isinstance(data, io.BufferedIOBase):
while True:
chunk = data.read(4096)
if not chunk:
return
yield chunk
elif isinstance(data, AsyncBufferedReader):
while True:
chunk = await data.read(4096)
if not chunk:
break
yield chunk
if aio_opened:
await data.close()
elif isinstance(data, Iterable):
for chunk in data: # type: ignore
yield chunk
elif isinstance(data, AsyncIterable):
async for chunk in data:
yield chunk
def _(file, *, loop=None, executor=None):
return AsyncBufferedReader(file, loop=loop, executor=executor)
from aiofiles.threadpool.binary import AsyncBufferedReader
from Crypto import Random # nosec
from Crypto.Cipher import AES # nosec
from Crypto.Hash import SHA256 # nosec
from Crypto.Util import Counter # nosec
from .attachments import _get_decryption_info_dict
AsyncDataT = Union[
str,
Path,
bytes,
Iterable[bytes],
AsyncIterable[bytes],
io.BufferedIOBase,
AsyncBufferedReader,
]
_EncryptedReturnT = AsyncGenerator[Union[bytes, Dict[str, Any]], None]
async def async_encrypt_attachment(data: AsyncDataT) -> _EncryptedReturnT:
"""Async generator to encrypt data in order to send it as an encrypted
attachment.
This function lazily encrypts and yields data, thus it can be used to
encrypt large files without fully loading them into memory if an iterable
or async iterable of bytes is passed as data.
Args:
data (str/Path/bytes/Iterable[bytes]/AsyncIterable[bytes]/
io.BufferedIOBase/AsyncBufferedReader): The data to encrypt.