引言：加密不等于安全

很多开发者认为，只要对消息内容做了端到端加密，通信就是安全的。但现实远比这复杂。

让我们看一个场景：假设你在使用一个加密聊天软件，每条消息发送时，网络上会出现一个约 1500 字节的数据包，每隔 5 秒准时出现一次。即使消息内容被 AES-256-GCM 加密得固若金汤，一个被动观察者依然能从中推断出大量信息：

• 你正在与某人通信（通信模式暴露了“有对话”）
• 你的在线时段（定时出现的心跳暴露了“活跃状态”）
• 你的行为模式（静默期 vs 活跃期的切换）

这种攻击方式被称为流量分析（Traffic Analysis）。它不攻击密码算法，而是攻击通信的元数据——包的大小、发送频率、连接模式。

流量分析的核心洞察：元数据本身就是信息。你不需要读懂信的内容，只要知道有人在寄信、寄给谁、多频繁，就已经获得了有价值的情报。

KaleidoTalk 在设计之初就把这个问题纳入了考虑。本文将详细介绍我们实现的流量混淆（Cover Traffic）方案，它通过固定包长和随机心跳两种机制，让外部观察者无法从网络流量特征中推断用户行为。

本文中展示的代码均来自 KaleidoTalk 项目，完整源码可在 GitHub 查看。KaleidoTalk 是一个开源的端到端加密聊天系统，采用 GPL v3 许可。

二、流量混淆的设计目标

在设计方案之前，我们先明确要解决什么问题：

基于这些威胁，我们设定了三个核心目标：

目标 1：隐藏真实消息长度

无论发送的是 10 字节的 "Hello" 还是 10KB 的文件，外部观察者看到的包大小应该是一致的。

目标 2：隐藏通信模式

发送消息时和空闲时的流量特征应该无法区分。外部观察者不应该能判断“现在有人在聊天”还是“用户只是挂着”。

目标 3：隐藏消息边界

多包消息的分片边界不应该暴露。外部观察者不应该能区分“这是一个大消息的一部分”和“这是多个小消息”。

下面我们逐一讲解实现方案。

三、固定包长协议：让每个包看起来都一样

3.1 协议设计

最直接的做法是固定每个数据包的大小。在 KaleidoTalk 中，我们将每个应用层数据包固定为2048 字节。

# 来自 padding.py PACKET_SIZE = 2048 # 固定包大小 HEADER_SIZE = 7 # 头部大小（type + length + seq + total） MAX_PAYLOAD = PACKET_SIZE - HEADER_SIZE # 最大有效载荷 # 包类型 TYPE_PADDING = 0x00 # 纯填充包（心跳） TYPE_DATA = 0x01 # 完整数据包 TYPE_FRAGMENT_FIRST = 0x02 # 分片：第一片 TYPE_FRAGMENT_MID = 0x03 # 分片：中间片 TYPE_FRAGMENT_LAST = 0x04 # 分片：最后一片

每个包的前 7 个字节是头部，包含四个字段：

[0:1] type - 1字节，标识包类型 [1:3] length - 2字节，有效载荷长度（大端序） [3:5] seq - 2字节，分片序号 [5:7] total - 2字节，总分片数 [7:N] payload - 实际数据 [N:END] padding - 随机填充字节

# 来自 padding.py def build_packet(data: bytes, packet_type: int = TYPE_DATA, frag_seq: int = 0, frag_total: int = 0) -> bytes: if len(data) > MAX_PAYLOAD: raise ValueError(f"Data {len(data)} bytes exceeds single packet limit {MAX_PAYLOAD}") header = struct.pack('>BHHH', packet_type, len(data), frag_seq, frag_total) payload = header + data # 用随机字节填充到固定长度 padding_len = PACKET_SIZE - len(payload) padding = os.urandom(padding_len) if padding_len > 0 else b'' return payload + padding

设计要点：

1. 头部紧贴载荷：length字段让接收方知道从哪里开始读取有效数据，剩余部分全部丢弃。
2. 随机填充：使用os.urandom()生成不可预测的填充字节，避免填充内容本身成为指纹。
3. 协议无关性：固定包长协议是传输层之上的封装，底层可以是 TCP、TLS 或任何可靠流式传输。

3.2 大消息的分片与重组

当消息超过MAX_PAYLOAD（2048 - 7 = 2041 字节）时，需要分片传输。分片机制同样遵循固定包长原则——每个分片仍然是 2048 字节的完整包。

# 来自 padding.py def fragment_data(data: bytes) -> list: if len(data) <= MAX_PAYLOAD: return [build_packet(data, TYPE_DATA)] total = (len(data) + MAX_PAYLOAD - 1) // MAX_PAYLOAD fragments = [] for i in range(total): start = i * MAX_PAYLOAD end = min(start + MAX_PAYLOAD, len(data)) chunk = data[start:end] if i == 0: ptype = TYPE_FRAGMENT_FIRST elif i == total - 1: ptype = TYPE_FRAGMENT_LAST else: ptype = TYPE_FRAGMENT_MID fragments.append(build_packet(chunk, ptype, frag_seq=i, frag_total=total)) return fragments

接收方使用FragmentReassembler类进行重组：

# 来自 padding.py class FragmentReassembler: def __init__(self): self._buffers = {} def feed(self, packet_type: int, data: bytes, frag_seq: int, frag_total: int): if packet_type == TYPE_DATA: return data # 无需重组 if frag_total <= 0 or frag_total > 1000: return None # 防御异常值 key = frag_total # 简化的标识方式 if key not in self._buffers: self._buffers[key] = {'total': frag_total, 'chunks': {}, 'timer': time.time()} buf = self._buffers[key] buf['chunks'][frag_seq] = data # 超时清理（30秒） if time.time() - buf['timer'] > 30: del self._buffers[key] return None # 检查是否收齐 if len(buf['chunks']) == buf['total']: result = b''.join(buf['chunks'][i] for i in range(buf['total'])) del self._buffers[key] return result return None

为什么分片机制对流量混淆很重要？

如果一个大消息的分片和多个小消息的独立包在外观上有区别，攻击者就能通过分析包序列来推断“这是一个大文件”还是“多个短消息”。我们通过以下方式消除这种区分能力：

1. 每个分片都是 2048 字节：和独立包在外观上完全一致
2. 分片类型在头部标记：只有解析头部才能区分，而头部是加密的
3. 超时清理机制：防止半成品分片在内存中堆积，也防止攻击者利用分片超时来探测

3.3 封装与解封装

PaddedSender和PaddedReceiver负责在应用层和传输层之间转换：

# 来自 padding.py class PaddedSender: @staticmethod def send(sock, data: bytes): packets = fragment_data(data) for pkt in packets: sock.sendall(pkt) class PaddedReceiver: def recv(self, sock) -> bytes: while True: # 从缓冲区提取完整包 while len(self._recv_buf) >= PACKET_SIZE: raw = self._recv_buf[:PACKET_SIZE] self._recv_buf = self._recv_buf[PACKET_SIZE:] ptype, data, frag_seq, frag_total = parse_packet(raw) if ptype == TYPE_PADDING: continue # 心跳包，跳过 result = self._reassembler.feed(ptype, data, frag_seq, frag_total) if result is not None: return result # 需要更多数据 chunk = sock.recv(PACKET_SIZE * 4) if not chunk: raise ConnectionError("Connection closed") self._recv_buf += chunk

四、随机心跳：让空闲状态看起来像在通信

固定包长解决了“包大小”的混淆，但还有一个问题：如果用户不说话，网络上就没有包。这种“静默期”本身就是一种强特征。

解决方案是在空闲时持续发送随机间隔的心跳包。外部观察者看到的是持续、稳定的流量，无法区分“用户正在聊天”和“用户在挂机”。

4.1 随机间隔算法

心跳间隔不能是固定值——固定的 5 秒间隔会让攻击者轻易识别出“这是心跳”。我们使用带随机抖动的间隔：

# 来自 padding.py BASE_INTERVAL = 5.0 # 基础间隔（秒） JITTER_RATIO = 1.0 / 3.0 # 抖动范围 def next_interval(): jitter = BASE_INTERVAL * JITTER_RATIO return BASE_INTERVAL + random.uniform(-jitter, jitter)

实际间隔在 3.33 秒到 6.67 秒之间均匀分布。为什么选择这个范围？

4.2 客户端心跳

# 来自 chat_client.py def _heartbeat_loop(self): while not self._heartbeat_stop.is_set(): self._heartbeat_stop.wait(next_interval()) if self._heartbeat_stop.is_set(): break try: if self.sock: with self._send_lock: self.sock.sendall(build_padding_packet()) except Exception: break