(CkPython) Streaming Encryption

Encrypt and decrypt using a stream.

Chilkat Python Downloads install with pip pip3 install chilkat or download... Python Module for Windows, MacOS, Linux, Alpine Linux, Solaris

import chilkat

# This example assumes the Chilkat API to have been previously unlocked.
# See Global Unlock Sample for sample code.

crypt = chilkat.CkCrypt2()

# Setup encryption using the chacha20 algorithm...
crypt.put_CryptAlgorithm("chacha20")
crypt.put_KeyLength(256)
crypt.put_EncodingMode("hex")
ivHex = "000000000000000000000002"
crypt.SetEncodedIV(ivHex,"hex")
crypt.put_InitialCount(42)
keyHex = "1c9240a5eb55d38af333888604f6b5f0473917c1402b80099dca5cbc207075c0"
crypt.SetEncodedKey(keyHex,"hex")

plainText = "The quick brown fox jumped over the lazy dog.\r\n"

stream = chilkat.CkStream()

# We'll save the encrypted output in eStrings to demonstrate streaming decryption next.
eStrings = chilkat.CkStringArray()

# Start a background task that will encrypt a stream.
# task is a CkTask
task = crypt.EncryptStreamAsync(stream)
success = task.Run()

# Write plainText to the stream, and read chacha20 encrypted text..

for i in range(1,11):
    # Note: An encryption algorithm's block size will cause buffering,
    # and therefore not every loop iteration will produce output.
    success = stream.WriteString(plainText)
    if (stream.get_DataAvailable() == True):
        cipherText = stream.readBytesENC("hex")
        print(cipherText)
        success = eStrings.Append(cipherText)

# Tell the background task that the stream has ended.
success = stream.WriteClose()

# Let's make sure the background task finished.
# It should already be the case that the task is finished.
while (task.get_Finished() != True) :
    task.SleepMs(20)

# Get any remaining data available from the stream.
if (stream.get_DataAvailable() == True):
    cipherText = stream.readBytesENC("hex")
    print(cipherText)
    success = eStrings.Append(cipherText)

if (task.get_TaskSuccess() != True):
    print("async encryption failed:")
    print(task.resultErrorText())
    success = False

print("-- encrypt finished --")

# Now decrypt to return the original.

# Reset the stream object so it can be used again.
stream.Reset()

# Start a background task that will decrypt a stream.
# task is a CkTask
task = crypt.DecryptStreamAsync(stream)
success = task.Run()

n = eStrings.get_Count()
for i in range(0,n):
    success = stream.WriteBytesENC(eStrings.getString(i),"hex")
    if (stream.get_DataAvailable() == True):
        plainText = stream.readString()
        print(plainText)

# Tell the background task that the stream has ended.
success = stream.WriteClose()

# Let's make sure the background task finished.
# It should already be the case that the task is finished.
while (task.get_Finished() != True) :
    task.SleepMs(20)

# Get any remaining data available from the stream.
if (stream.get_DataAvailable() == True):
    plainText = stream.readString()
    print(plainText)

if (task.get_TaskSuccess() != True):
    print("async decryption failed:")
    print(task.resultErrorText())
    success = False

print("-- decrypt finished --")