UnicodeDecodeError: кодек "utf-8" не может декодировать байты в позиции 65534-65535: неожиданный конец данных

Ответ 1

Основная проблема с вашим кодом заключается в том, что вы используете строки. AES работает с двоичными данными, и если бы вы использовали PyCryptodome, этот код повышал бы TypeError:

Object type <class 'str'> cannot be passed to C code

Pycrypto принимает строки, но кодирует их в байты внутри, поэтому не имеет смысла декодировать ваши байты в строку, потому что они будут закодированы обратно в байты. Кроме того, он кодирует ASCII (тестируется с PyCrypto v2.6.1, Python v2.7), и поэтому этот код, например:

encryptor.encrypt(u'ψ' * 16)

поднимет UnicodeEncodeError:

File "C:\Python27\lib\site-packages\Crypto\Cipher\blockalgo.py", line 244, in encrypt
    return self._cipher.encrypt(plaintext)
UnicodeEncodeError: 'ascii' codec can't encode characters in position 0-15

Вы должны всегда использовать байты при шифровании или расшифровке данных. Затем вы можете декодировать текст в строку, если это текст.

Следующий вопрос - ваш метод заполнения. Он создает строку, поэтому вы получаете TypeError при попытке применить его к открытому тексту, который должен быть байтами. Вы можете исправить это, если вы набиваете байты,

chunk += b' ' * (16 - len(chunk) % 16)

но было бы лучше использовать дополнение PKCS7 (в настоящее время вы используете нулевое дополнение, но вместо пробела вместо нулевого байта).

PyCryptodome предоставляет функции заполнения, но, похоже, вы используете PyCrypto. В этом случае вы можете реализовать дополнение PKCS7 или, еще лучше, скопировать функции заполнения PyCryptodome.

try:
    from Crypto.Util.Padding import pad, unpad
except ImportError:
    from Crypto.Util.py3compat import bchr, bord

    def pad(data_to_pad, block_size):
        padding_len = block_size-len(data_to_pad)%block_size
        padding = bchr(padding_len)*padding_len
        return data_to_pad + padding

    def unpad(padded_data, block_size):
        pdata_len = len(padded_data)
        if pdata_len % block_size:
            raise ValueError("Input data is not padded")
        padding_len = bord(padded_data[-1])
        if padding_len<1 or padding_len>min(block_size, pdata_len):
            raise ValueError("Padding is incorrect.")
        if padded_data[-padding_len:]!=bchr(padding_len)*padding_len:
            raise ValueError("PKCS#7 padding is incorrect.")
        return padded_data[:-padding_len]

Функции pad и unpad были скопированы из Crypto.Util.Padding и изменены, чтобы использовать только дополнение PKCS7. Обратите внимание, что при использовании дополнения PKCS7 важно проложить последний кусок, даже если его размер кратен размеру блока, иначе вы не сможете правильно отладить.

Применяя эти изменения к функции encrypt_file,

def encrypt_file(key, in_filename, out_filename=None, chunksize=64*1024):
    if not out_filename:
        out_filename = in_filename + '.enc'
    iv = os.urandom(16)
    encryptor = AES.new(key, AES.MODE_CBC, iv)
    filesize = os.path.getsize(in_filename)
    with open(in_filename, 'rb') as infile:
        with open(out_filename, 'wb') as outfile:
            outfile.write(struct.pack('<Q', filesize))
            outfile.write(iv)
            pos = 0
            while pos < filesize:
                chunk = infile.read(chunksize)
                pos += len(chunk)
                if pos == filesize:
                    chunk = pad(chunk, AES.block_size)
                outfile.write(encryptor.encrypt(chunk))

и соответствующую функцию decrypt_file,

def decrypt_file(key, in_filename, out_filename=None, chunksize=64*1024):
    if not out_filename:
        out_filename = in_filename + '.dec'
    with open(in_filename, 'rb') as infile:
        filesize = struct.unpack('<Q', infile.read(8))[0]
        iv = infile.read(16)
        encryptor = AES.new(key, AES.MODE_CBC, iv)
        with open(out_filename, 'wb') as outfile:
            encrypted_filesize = os.path.getsize(in_filename)
            pos = 8 + 16 # the filesize and IV.
            while pos < encrypted_filesize:
                chunk = infile.read(chunksize)
                pos += len(chunk)
                chunk = encryptor.decrypt(chunk)
                if pos == encrypted_filesize:
                    chunk = unpad(chunk, AES.block_size)
                outfile.write(chunk)

Этот код совместим с Python2/Python3, и он должен работать либо с PyCryptodome, либо с PyCrypto.

Однако, если вы используете PyCrypto, я рекомендую обновить PyCryptodome. PyCryptodome - это вилка PyCrypto, и она предоставляет один и тот же API (поэтому вам не придется слишком сильно менять свой код), а также некоторые дополнительные функции: дополнения, алгоритмы аутентифицированного шифрования, KDF и т.д. С другой стороны, PyCrypto не а также некоторые версии страдают от переполнения буфера на основе кучи: CVE-2013-7459.

Ответ 2

В дополнение к принятому ответу, я считаю, что показ нескольких реализаций может быть полезен для читателей/новых учеников:

import os
import sys
import pickle
import base64
import hashlib
import errno

from Crypto import Random
from Crypto.Cipher import AES

DEFAULT_STORAGE_DIR = os.path.join(os.path.dirname(__file__), '.ncrypt')

def create_dir(dir_name):
    """ Safely create a new directory. """
    try:
        os.makedirs(dir_name)
        return dir_name
    except OSError as e:
        if e.errno != errno.EEXIST:
            raise OSError('Unable to create directory.')


class AESCipher(object):
    DEFAULT_CIPHER_PICKLE_FNAME = "cipher.pkl"

    def __init__(self, key):
        self.bs = 32  # block size
        self.key = hashlib.sha256(key.encode()).digest()

    def encrypt(self, raw):
        raw = self._pad(raw)
        iv = Random.new().read(AES. block_size)
        cipher = AES.new(self.key, AES.MODE_CBC, iv)
        return base64.b64encode(iv + cipher.encrypt(raw))

    def decrypt(self, enc):
        enc = base64.b64decode(enc)
        iv = enc[:AES.block_size]
        cipher = AES.new(self.key, AES.MODE_CBC, iv)
        return self._unpad(cipher.decrypt(enc[AES.block_size:])).decode('utf-8')

    def _pad(self, s):
        return s + (self.bs - len(s) % self.bs) * chr(self.bs - len(s) % self.bs)

    @staticmethod
    def _unpad(s):
        return s[:-ord(s[len(s)-1:])]

И иллюстрирующие примеры использования выше:

while True:
    option = input('\n'.join(["="*80,
                              "| Select an operation:",
                              "| 1) E : Encrypt",
                              "| 2) D : Decrypt",
                              "| 3) H : Help",
                              "| 4) G : Generate new cipher",
                              "| 5) Q : Quit",
                              "="*80,
                              "> "])).lower()
    print()

    if option == 'e' or option == 1:
        plaintext = input('Enter plaintext to encrypt: ')
        print("Encrypted: {}".format(cipher.encrypt(plaintext).decode("utf-8")))

    elif option == 'd' or option == 2:
        ciphertext = input('Enter ciphertext to decrypt: ')
        print("Decrypted: {}".format(cipher.decrypt(ciphertext.encode("utf-8"))))

    elif option == 'h' or option == 3:
        print("Help:\n\tE: Encrypt plaintext\n\tD: Decrypt ciphertext.")

    elif option == 'g' or option == 4:
        if input("Are you sure? [yes/no]: ").lower() in ["yes", "y"]:
            cipher = AESCipher(key=input('Enter cipher password: '))

            with open(pickle_fname, 'wb') as f:
                pickle.dump(cipher, f)
            print("Generated new cipher.")

    elif option == 'q' or option == 5:
        raise EOFError
    else:
        print("Unknown operation.")