UnicodeDecodeError: 'utf-8'编解码器无法解码数据的第65534-65535个位置的字节：数据意外结束

Question

我想使用简单的AES加密算法来加密文件，以下是我的Python3源代码。

import os, random, struct
from Crypto.Cipher import AES

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)
            while True:
                chunk = infile.read(chunksize)
                if len(chunk) == 0:
                    break
                elif len(chunk) % 16 != 0:
                    chunk += ' ' * (16 - len(chunk) % 16)
                outfile.write(encryptor.encrypt(chunk.decode('UTF-8','strict')))

一些文件可以正常工作，但某些文件会遇到如下错误信息：

encrypt_file("qwertyqwertyqwer",'/tmp/test1' , out_filename=None, chunksize=64*1024)

没有错误信息，正常工作。

encrypt_file("qwertyqwertyqwer",'/tmp/test2' , out_filename=None, chunksize=64*1024)

Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "<stdin>", line 17, in encrypt_file
UnicodeDecodeError: 'utf-8' codec can't decode bytes in position 65534-65535: unexpected end of data

如何修复我的encrypt_file函数？

按照t.m.adam所说的方法来修复。

outfile.write(encryptor.encrypt(chunk.decode('UTF-8','strict')))

as

outfile.write(encryptor.encrypt(chunk))

尝试使用一些文件。

encrypt_file("qwertyqwertyqwer",'/tmp/test' , out_filename=None, chunksize=64*1024)
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "<stdin>", line 16, in encrypt_file
TypeError: can't concat bytes to str

- user7988893

---

为什么要解码数据？.encrypt()可以处理字节，实际上，在Python3中，如果您传递字符串，它会引发异常。 - t.m.adam

类型错误：无法将字节串连接到字符串。 - user7988893

你能给我整个Python加密文件的函数吗？ - user7988893

Answer 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。

- t.m.adam

Answer 2

除了被接受的答案之外，我认为展示多个简单AES加密实现对于读者/新学习者可能很有用：

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.")

- Sean Pianka

---

+1，不错的课程，但我担心这个问题的标题不会吸引到对Python加密感兴趣的读者。此外，这对于较大的文件将无法工作，您需要像OP一样将它们“切成块”。 - t.m.adam

因为你无法将几个GB存储到一个字符串中，这会引发MemoryError错误。 - t.m.adam

是的，只要块大小是块密码模式（仅适用于CBC和ECB）的倍数，分块就可以正常工作。最后一个块可以是任何大小，将被填充。 - t.m.adam

如果您需要一些提示，SHA256不适合作为密钥派生函数，最好使用hashlib.pbkdf2_hmac与一个长的、随机的盐（至少8个字节）和高数量的迭代（至少10,000次）。此外，您可能希望对密文进行HMAC处理，因为在某些情况下，CBC可能容易受到填充预言攻击。我编写了一个Python类来解决所有这些问题，如果您想要，可以在这里找到它。目前它只使用基于密码的密钥，但我很快会更新它。 - t.m.adam

很高兴你喜欢它！我几年前开始了这个项目，当时我需要将Python加密代码移植到PHP。从那时起，我改进了实现并添加了更多的编程语言。最大的问题是.NET，因为它的KDF选项有限，并且不支持任何认证加密算法，所以我不得不使用Encrypt-then-MAC。欢迎批评/贡献。 - t.m.adam