基于PyCryptodome的RSA+AES客户端-服务端加解密实现的合理性咨询

首先，作为加密新手，你选择的RSA加密会话密钥 + AES-EAX加密业务数据的混合加密方案是非常标准且安全的思路，先给你点个赞！下面针对你的问题逐一解答：

问题1：发送加密后的会话密钥、nonce、tag是否合适？ #

完全合适，这甚至是这类混合加密方案的标准操作：

• 加密后的会话密钥（enc_session_key）：只有持有对应RSA私钥的接收方能解密得到AES会话密钥，从根上保证了会话密钥的安全性；
• nonce：AES-EAX模式需要唯一的初始化向量，用于确保相同明文+密钥每次加密得到不同密文，必须随密文一起传输；
• tag：EAX模式的认证标签，用于解密时校验密文的完整性与真实性，防止数据被篡改，也是必须传输的核心参数。

所以你传输这几个字段的做法是完全正确的。

问题2：有没有遗漏的关键要点？ #

你的核心加密逻辑是没问题的，但有几个需要注意的安全和健壮性细节：

• Pickle序列化的严重安全风险
  Pickle是极不安全的序列化方式，反序列化不可信数据时，攻击者可以构造恶意Pickle包触发任意代码执行。如果你的服务端/客户端接收了被篡改的数据包，用pickle.loads()处理会直接导致系统被入侵。
  建议替换为更安全的方案：比如用JSON配合Base64编码（将二进制字段转为字符串）、MsgPack（原生支持二进制），或者自定义二进制格式打包字段（类似你处理Socket传输长度的方式）。

• 缺失异常处理逻辑
  当前代码没有处理任何异常场景：比如RSA解密时密文被篡改会抛出异常、AES tag校验失败（数据被篡改）会抛出异常、Socket传输中断、Pickle序列化/反序列化失败等。建议在核心函数中添加try-except块，捕获并处理这些异常，避免程序直接崩溃，同时可以在异常时做重试或错误提示。

• 数据类型一致性检查
  要确保传入encrypt_data的data是bytes类型，如果是字符串需要先编码（比如data.encode('utf-8')），否则AES.encrypt_and_digest会抛出类型错误。可以在encrypt_data开头添加类型检查，或者在调用时统一处理数据格式。

• RSA私钥的安全存储
  decrypt_data中直接从文件读取私钥，要确保私钥文件的权限设置严格（比如Linux下设为600，仅拥有者可读写），绝对不能将私钥硬编码或明文存储在代码仓库中。

• 接收端逻辑的一致性
  你只给出了发送端的send_data，要确保接收端逻辑完全对应：先读取8字节的长度（用struct.unpack("<Q", ...)解析），再读取对应长度的数据，最后反序列化并解密。如果接收端逻辑错误，会导致数据解析失败。

格式化后的完整代码 #

import struct
import pickle
import os
from Crypto.PublicKey import RSA
from Crypto.Cipher import AES, PKCS1_OAEP
from Crypto.Random import get_random_bytes

def send_data(socket, key, data):
    enc_session_key, nonce, tag, ciphertext = encrypt_data(key, data)
    
    encapsulated_data = [enc_session_key, nonce, tag, ciphertext]
    
    data = pickle.dumps(encapsulated_data)
    
    stream = bytes(data)
    stream_length = len(stream)
    socket.sendall(struct.pack("<Q", stream_length))
    socket.sendall(stream)

def encrypt_data(key, data):
    recipient_key = key
    session_key = get_random_bytes(16)

    # Encrypt the session key with the public RSA key
    cipher_rsa = PKCS1_OAEP.new(recipient_key)
    enc_session_key = cipher_rsa.encrypt(session_key)

    # Encrypt the data with the AES session key
    cipher_aes = AES.new(session_key, AES.MODE_EAX)
    ciphertext, tag = cipher_aes.encrypt_and_digest(data)

    return enc_session_key, cipher_aes.nonce, tag, ciphertext

def decrypt_data(enc_session_key, nonce, tag, ciphertext):
    PRIVATE_PATH = "private_key.pem"  # 假设的私钥文件路径
    if not os.path.exists(PRIVATE_PATH):
        return None

    private_key = RSA.import_key(open(PRIVATE_PATH).read())

    # Decrypt the session key with the private RSA key
    cipher_rsa = PKCS1_OAEP.new(private_key)
    session_key = cipher_rsa.decrypt(enc_session_key)

    # Decrypt the data with the AES session key
    cipher_aes = AES.new(session_key, AES.MODE_EAX, nonce)
    data = cipher_aes.decrypt_and_verify(ciphertext, tag)

    return data

备注：内容来源于stack exchange，提问作者Solitude