AES-256加密消息发送后解密报错ValueError:输入长度需为16的倍数问题排查

核心问题根源 #

TCP是面向流的传输协议，不存在天然的消息边界。你调用sock.send()发送的完整加密数据，接收端调用sock.recv(4096)时可能只收到部分数据，或者一次收到多条消息的拼接数据，直接传入解密函数就会因为base64解码后的数据长度不符合AES块大小要求报错，这也是你调整recv参数时报错信息会变化的核心原因。

排查验证方法 #

• 在接收端解密前打印recvData的长度和原始内容，对比发送端sendData的长度，确认是否存在接收不完整的情况
• 测试发送短于100字节的小消息，大概率可以正常解密，发送超过1024字节的长消息必然报错，即可验证是流接收不完整的问题

解决方案 #

方案1：添加消息长度前缀（最可靠，推荐） #

发送端在发送加密数据前，先发送4字节的大端整数表示后续加密数据的长度，接收端先收4字节拿到长度，再收对应长度的完整数据再解密。
修改后的发送代码：

import struct
def send(sock):
    while True:
        try:
            aes = AESCipher(key)
            send_data = aes.encrypt(input())
            # 先打包4字节长度前缀
            len_prefix = struct.pack('>I', len(send_data))
            sock.sendall(len_prefix + send_data)
        except Exception as e:
            print(e)
            continue

修改后的接收代码：

import struct
def receive(sock):
    while True:
        try:
            # 先收4字节长度前缀
            len_prefix = b''
            while len(len_prefix) < 4:
                chunk = sock.recv(4 - len(len_prefix))
                if not chunk:
                    # 连接断开直接返回
                    return
                len_prefix += chunk
            data_len = struct.unpack('>I', len_prefix)[0]
            # 再收完整的加密数据
            recv_data = b''
            while len(recv_data) < data_len:
                chunk = sock.recv(data_len - len(recv_data))
                if not chunk:
                    return
                recv_data += chunk
            aes = AESCipher(key)
            print('%s :' % opp_id, aes.decrypt(recv_data))
        except DecryptionException as e:
            print(e)

方案2：使用特殊分隔符（适合小数据交互场景） #

因为加密后的数据是base64编码的，不会出现换行符，可以在发送的加密数据末尾加b'\n'作为分隔符，接收端按行读取完整数据再解密：
修改发送代码：

def send(sock):
    while True:
        try:
            aes = AESCipher(key)
            send_data = aes.encrypt(input()) + b'\n'
            sock.sendall(send_data)
        except:
            continue

修改接收代码，可直接用socket封装的文件对象按行读取：

def receive(sock):
    # 把socket封装成文件对象，方便按行读取
    sock_file = sock.makefile('rb')
    while True:
        try:
            recv_data = sock_file.readline()
            if not recv_data:
                return
            # 去掉末尾的换行符
            recv_data = recv_data.strip()
            aes = AESCipher(key)
            print('%s :' % opp_id, aes.decrypt(recv_data))
        except DecryptionException as e:
            print(e)

额外优化建议 #

• 不要每次加解密都实例化AESCipher，同一个密钥的场景下可以复用实例，提升性能
• 空的except会吃掉所有异常，不利于排查问题，建议指定要捕获的异常类型
• 发送数据用sock.sendall()而不是sock.send()，确保所有数据都被完整发送

内容的提问来源于stack exchange，提问作者qwerty