Python客户端服务端字符串AES加解密失败问题求助

核心错误点分析 #

• 密钥与IV不共享：客户端A、B各自随机生成key和iv，两端加密参数完全不匹配，这是解密失败的根本原因。AES-CBC要求加密和解密必须使用相同的密钥和初始化向量（IV）。
• 重复使用加密/解密器实例：cryptography库的Encryptor和Decryptor是一次性实例，调用finalize()后就无法再使用，多次复用会导致解密数据混乱。
• 双重填充操作：发送时先手动添加空格填充，再调用PKCS7填充，导致数据被过度填充，解密后无法还原原始内容。
• 去填充未完成：接收时仅调用unpadder.update()，未执行unpadder.finalize()，无法完成完整的去填充流程。
• 编码混用：手动填充时混用ASCII和UTF-8编码，虽然ASCII是UTF-8子集，但易引发潜在编码问题。

修正后的代码 #

client.py

import threading
import socket
import os
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives import padding

backend = default_backend()
PORT = 7777
SERVER_IP = 'localhost'

def main():
    client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    try:
        client.connect((SERVER_IP, PORT))
    except:
        return print('\nNão foi possível se conectar ao servidor!\n')

    # 首次连接时，接收服务器发送的密钥和IV（测试场景，生产环境需用安全密钥交换）
    key = client.recv(32)
    iv = client.recv(16)
    print(f"已获取共享密钥和IV")

    username = input('Nome do usuário: ')
    print(f'\n{username} conectado a {SERVER_IP} na porta {PORT}!\n')

    thread1 = threading.Thread(target=receiveMessages, args=[client, key, iv])
    thread2 = threading.Thread(target=sendMessages, args=[client, username, key, iv])

    thread1.start()
    thread2.start()

    thread1.join()
    thread2.join()


def receiveMessages(client, key, iv):
    while True:
        try:
            msg_rcvd = client.recv(2048)
            if not msg_rcvd:
                break
            print(f'encrypted msg received: {msg_rcvd}')
            
            # 每次解密都创建新的解密器实例
            cipher = Cipher(algorithms.AES(key), modes.CBC(iv), backend=backend)
            decryptor = cipher.decryptor()
            dec_ct = decryptor.update(msg_rcvd) + decryptor.finalize()
            print(f'msg after decryption: {dec_ct}')

            # 完整执行去填充操作
            unpadder = padding.PKCS7(128).unpadder()
            msg_unpad = unpadder.update(dec_ct) + unpadder.finalize()
            print(f'unpadded msg: {msg_unpad.decode("utf-8")}')
            
        except Exception as e:
            print(f'\nNão foi possível permanecer conectado no servidor! Erro: {e}\n')
            print('Pressione <Enter> Para continuar...')
            client.close()
            break
            

def sendMessages(client, username, key, iv):
    while True:
        try:
            msg = input('\n')
            if not msg:
                continue
            # 仅使用PKCS7填充，去掉手动填充
            full_msg = f"{username}: {msg}".encode('utf-8')
            padder = padding.PKCS7(128).padder()
            msg_pad = padder.update(full_msg) + padder.finalize()
            print(f'padded msg: {msg_pad}')

            # 每次加密都创建新的加密器实例
            cipher = Cipher(algorithms.AES(key), modes.CBC(iv), backend=backend)
            encryptor = cipher.encryptor()
            ct = encryptor.update(msg_pad) + encryptor.finalize()
            
            print(f'msg after encryption: {ct}')
            client.send(ct)
                
        except Exception as e:
            print(f'error: {e}')
            return

if __name__ == "__main__":
    main()

server.py

import threading
import socket
import os

clients = []
PORT = 7777
SERVER_IP = 'localhost'
# 服务器生成共享密钥和IV，分发给所有连接的客户端（测试场景）
shared_key = os.urandom(32)
shared_iv = os.urandom(16)

def main():
    server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    try:
        server.bind((SERVER_IP, PORT))
        server.listen()
        print(f'Servidor ligado em {SERVER_IP} na porta {PORT}')
    except:
        return print('\nNão foi possível iniciar o servidor!\n')

    while True:
        client, addr = server.accept()
        clients.append(client)
        print(f'Novo cliente conectado: {addr}')
        # 发送共享密钥和IV给新客户端
        client.send(shared_key)
        client.send(shared_iv)
        thread = threading.Thread(target=messagesTreatment, args=[client])
        thread.start()

def messagesTreatment(client):
    while True:
        try:
            msg = client.recv(2048)
            if not msg:
                break
            broadcast(msg, client)
        except:
            clients.remove(client)
            print(f'Cliente desconectado, total clientes: {len(clients)}')
            break

def broadcast(msg, client):
    for clientItem in clients:
        if clientItem != client:
            try:
                clientItem.send(msg)
            except:
                clients.remove(clientItem)

if __name__ == "__main__":
    main()

修正说明 #

1. 共享密钥与IV：由服务器生成统一的密钥和IV，新客户端连接时自动分发，确保所有客户端使用相同的加密参数。
2. 单次使用加密/解密器：每次加密、解密操作都创建新的Encryptor/Decryptor实例，避免复用导致的状态混乱。
3. 单一填充方式：移除手动空格填充，仅使用PKCS7标准填充，符合AES块加密的要求。
4. 完整去填充流程：调用unpadder.update() + unpadder.finalize()完成去填充，确保正确还原原始数据。
5. 统一编码：全程使用UTF-8编码，避免编码混用问题。

内容的提问来源于stack exchange，提问作者Josué Jr