Python实现Minecraft服务器加密数据包完整接收问题求助

Minecraft服务器加密数据包时会把**整个数据包（包括varint格式的长度字段）**都用AES-CFB8加密，解决这个问题的核心是利用CFB8模式的流式特性——解密是逐字节进行的，不需要等待完整块，所以可以通过「边接收加密数据→边解密→边解析varint」的方式获取数据包长度，再接收剩余内容。

具体实现步骤 #

• 维护一个解密后的字节缓冲区，存储已解密但尚未处理的明文数据
• 每次从socket接收一小段加密数据（比如1024字节），立即用AES-CFB8解密
• 将解密后的字节追加到缓冲区，尝试从缓冲区中解析varint格式的数据包长度
• 一旦解析出完整的长度值，就从缓冲区中取出对应长度的明文数据作为完整数据包，剩余缓冲区内容留作下一个数据包的处理

代码示例 #

首先初始化AES解密器（假设你已经有共享密钥和初始化向量IV）：

from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from cryptography.hazmat.backends import default_backend

# shared_secret为你的共享密钥，iv为加密阶段使用的初始化向量（解密时需一致）
decryptor = Cipher(algorithms.AES(shared_secret), modes.CFB8(iv), backend=default_backend()).decryptor()

修改接收和解析逻辑，加入解密与缓冲区处理：

class EncryptedPacketReader:
    def __init__(self, socket, decryptor):
        self.socket = socket
        self.decryptor = decryptor
        self.buffer = b""

    def _recv_and_decrypt(self, size=1024):
        # 接收加密数据并解密
        encrypted_data = self.socket.recv(size)
        if not encrypted_data:
            raise EOFError("Socket closed")
        return self.decryptor.update(encrypted_data)

    def _read_varint(self):
        # 从缓冲区解析varint，数据不足时自动补充
        value = 0
        position = 0
        while True:
            if position >= len(self.buffer):
                self.buffer += self._recv_and_decrypt()
            byte = self.buffer[position]
            position += 1
            value |= (byte & 0x7F) << (7 * (position - 1))
            if not (byte & 0x80):
                # 移除已解析的字节，更新缓冲区
                self.buffer = self.buffer[position:]
                return value

    def read_packet(self):
        # 解析数据包总长度
        packet_length = self._read_varint()
        # 确保缓冲区有足够的数据包内容
        while len(self.buffer) < packet_length:
            self.buffer += self._recv_and_decrypt()
        # 提取完整数据包内容
        packet_data = self.buffer[:packet_length]
        self.buffer = self.buffer[packet_length:]
        # 解析数据包ID
        id_value = 0
        pos = 0
        while True:
            byte = packet_data[pos]
            pos += 1
            id_value |= (byte & 0x7F) << (7 * (pos - 1))
            if not (byte & 0x80):
                break
        # 返回数据包长度、ID和剩余数据
        return (packet_length, id_value, packet_data[pos:])

使用方式 #

# 初始化加密数据包读取器
reader = EncryptedPacketReader(s, decryptor)
# 读取加密后的数据包
packet_length, packet_id, packet_content = reader.read_packet()
log("Packet Info: ", (packet_length, packet_id))

关键说明 #

• AES-CFB8的解密器是状态化的，必须持续使用同一个decryptor对象处理所有加密数据，不能每次解密都重新创建，否则会导致解密错误
• 缓冲区机制确保跨数据包的字节不会丢失，比如一次接收的解密数据可能包含多个数据包的部分内容
• 逐字节解析varint的逻辑和未加密时一致，仅数据源变为解密后的缓冲区

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