在 Python 服务中替换不可信 pickle 反序列化：从缓存与队列注入到 JSON 载荷迁移

把缓存和消息队列当作“内部可信边界”，是 Python 服务中常见的安全假设：数据来自 Redis、数据库或任务队列，似乎就可以直接 pickle.loads。但只要攻击者能影响某条写入路径、队列消息或缓存内容，读取端就可能在反序列化时执行其中嵌入的操作。迁移时的核心目标不是换一种序列化语法，而是让数据只表达经过校验的业务字段，不携带可执行对象。

风险如何进入服务

pickle 不只是把数据还原成字典、列表等值。反序列化过程可以按载荷指示查找并调用对象，因此处理不可信 pickle 可能导致任意代码执行。风险取决于数据是否可被影响，不取决于存储介质是不是“内部的”。

常见数据流包括：

• API 或管理功能写入缓存，另一个服务从缓存读取并反序列化。
• 多个应用共享队列，其中一个生产者可以提交其他消费者会处理的消息。
• 缓存或队列权限配置过宽，或某个上游服务被攻陷，攻击者因而能写入载荷。
• 服务把数据库记录、文件内容等外部数据读出后，当作可信 pickle 处理。

这不意味着所有缓存和队列都已暴露。审计时应验证谁能写入、谁会读取，以及数据是否经过可信校验，而不是仅凭组件部署在内网就排除风险。

最小复现：反序列化触发命令执行

以下代码使用无害的 echo 命令演示执行时机。它适用于本地隔离测试，不应在生产环境运行：

import os
import pickle


class Demo:
    def __reduce__(self):
        return (os.system, ("echo PICKLE_EXECUTED",))


payload = pickle.dumps(Demo())
pickle.loads(payload)  # 反序列化时执行载荷指定的调用

__reduce__ 返回的不是要还原的普通字段，而是一个可调用对象及其参数。pickle.loads 处理载荷时会调用它；示例中，调用最终交给了 os.system。实际攻击者不必在服务端提交一个 Python 类实例：只要能构造相应 pickle 字节并让服务处理，风险就可能成立。

典型的脆弱调用点可能藏在缓存或任务处理封装中：

value = pickle.loads(cache.get(key))

关键问题不在这行代码是否使用了缓存，而在 cache.get(key) 返回的数据能否被不可信方影响。改用 pickle.Unpickler 或包装反序列化函数，也不会自动消除这个问题。

审计调用点与数据来源

先在仓库中搜索反序列化入口，再沿数据流核实来源。可以从这些关键词开始：

pickle.loads
pickle.load
pickle.Unpickler
dill.loads
cloudpickle.loads

也要检查项目自建封装和依赖库中的反序列化路径。仅搜索调用点不够：需要追踪数据从哪里写入、谁有写权限、是否存在共享凭证，以及哪些消费者会读取。还要核实任务重试、死信队列、缓存回填和历史数据回放等路径，因为它们可能继续处理旧载荷。

静态扫描可以帮助定位可疑调用，但不能单独证明安全或漏洞成立。对每个命中点记录：

• 来源：载荷来自哪个生产者、存储或接口。
• 写入权限：哪些身份或服务可以创建、修改对应数据。
• 读取行为：是否直接反序列化，读取失败时是否会走其他解析路径。
• 影响范围：哪个进程、账号和运行环境会处理载荷。
• 验证结果：在隔离环境中使用无害测试载荷，确认是否能触发预期调用。

不要把真实生产数据复制到测试环境后直接反序列化。检查工具也应与执行路径区分开；任何无法确认安全的旧载荷，都不应通过“试着加载一下”来检查。

将载荷迁移为 JSON 或结构化消息

JSON 不会像 pickle 那样把任意 Python 对象恢复为可执行对象，但它也不会自动验证业务含义。迁移时应同时约束格式、字段、类型和大小。

例如，将任务消息限制为明确字段，并在解析后做校验：

import json


def parse_task(raw: bytes) -> dict:
    if len(raw) > 64 * 1024:
        raise ValueError("消息过大")

    data = json.loads(raw)

    if not isinstance(data, dict):
        raise ValueError("消息必须是对象")
    if set(data) != {"task_id", "operation"}:
        raise ValueError("消息字段不符合约定")
    if not isinstance(data["task_id"], str):
        raise ValueError("task_id 必须是字符串")
    if data["operation"] not in {"refresh", "rebuild"}:
        raise ValueError("不支持的 operation")

    return data

示例展示的是边界校验思路，不是完整的业务验证方案。实际服务还应限制嵌套深度和字段长度，并验证身份、权限、状态转换及操作参数。不同任务类型可以使用不同结构或显式版本字段，避免一个宽泛字典被多个消费者以不同方式解释。

迁移步骤可以按以下顺序推进：

1. 盘点格式和消费者：整理每类载荷的字段、生产者、消费者、保留期和重试路径；先明确哪些旧数据必须继续处理。
2. 定义消息契约：约定格式版本、必填字段、可选字段、类型、允许值和兼容规则。优先使用简单的 JSON 值或项目已有的结构化消息格式。
3. 实现解析与校验：拒绝未知操作、错误类型、缺失字段和超限载荷；校验失败时记录必要的诊断信息并隔离消息，不要自动退回到 pickle 解析。
4. 更新生产者和消费者：先部署能识别新格式的消费者，再逐步让生产者改写新格式。若消费者可独立升级，应确认旧生产者产生的消息仍能被明确处理。
5. 处置历史载荷：在隔离的迁移程序中处理确有必要保留的数据，优先从可信业务源重建消息。不要为了省事，让在线服务长期保留不可信 pickle 的兼容入口。
6. 回归并观察：覆盖正常消息、字段缺失、错误类型、未知版本、过大载荷、重复投递、重试及死信路径；监控解析失败率、积压和业务结果。

兼容、灰度与回滚

滚动发布时，先部署能够处理新格式的消费者，再切换生产者。若需要短暂的双格式过渡，必须明确每种格式的来源与处理期限，并将旧格式限制在可信、受控的迁移路径中。不要采用“先尝试 JSON，失败就 pickle.loads”的兜底写法：攻击者可以利用格式识别或解析失败，把数据引向危险分支。

可以在消息中加入明确的版本标识，并按版本分派到对应解析器。版本字段本身也要校验；未知版本应隔离或拒绝，不能猜测格式。灰度期间，逐步切换生产者或流量比例，同时观察消费者错误、任务积压、重试与死信数量，并抽查业务结果是否与旧流程一致。

回滚应优先回滚生产者发布或流量开关，继续让已部署的消费者处理新格式；如需回退消费者，必须确认它仍能安全处理队列中已写入的新消息。若系统只能靠重新启用不可信 pickle 才能恢复服务，说明回滚方案本身会重新打开风险。应预先设计暂停生产、隔离消息、恢复业务数据或切换到兼容消费者的方案，而不是在故障时临时放宽反序列化边界。

把 pickle 的载荷加签也不能把它变成安全格式：签名只能在正确验证后帮助确认来源，不能改变 pickle 反序列化具有执行能力这一事实。长期修复应是缩小写入权限、移除不可信 pickle 读取、明确消息契约，并用输入校验和回归测试持续验证新边界。