Django JSONField SQL注入漏洞复现 & 原理分析_django 注入

作者：小蓝xlanll | 2024-03-20 01:13:46

踩

django 注入

关于这个漏洞前几天看了很多的文章，其实大部分payload是一样的，都是如何去构造，或者去命令执行复现一下，我一开始也是这样去做的，但是这个漏洞是怎么形成的，对我来说可能一知半解，或者说完全不了解，在后面学习的过程中感觉吃力又去补习了别的知识，比如Django的两个基类，ArrayField、JSONField，Json.objects.filter()和QuerySet相关的知识，包括ORM注入等等

1/漏洞原理

PostgreSQL、SQLite3、MySQL、Oracle，在大部分情况下，以上四种数据库都能与Django框架配合工作，Django在对PostgreSQL提供了强大的功能同时，在成本、特性、速度和稳定性方面都做的比较平衡，当然官方也建议该框架配合Postgresql一起使用

相比较MySQL，PostgreSQL支持多种高级数据类型，比如array，MySQL只支持标准类型。同时PostgreSQL支持P地址数据类型、常量、函数调用、JSON和其他NoSQL功能，这让这个关系型数据库也拥有了一些NoSQL的特点，MySQL支持JSON不过不支持其他的NoSQL功能，当然PostgreSQL不只是一个关系型数据库，还支持非关系数据类型JSON

JSONField
ArrayField
HStoreField

在Django的model.py中定义JSONField：

max_length为最大长度类型，detail存储了可查询的信息

比如，detail中存储了如下的一些信息

如果我想查询关于ganyu的所有文章呢？

就可以使用Django的QuerySet（Collection为上方定义的类，QuerySet支持部分链式调用，如all接口就可以用于查询所有数据，detail__author中detail是一个JSONField，而下划线后的内容代表着JSON中的键名，而不再是常规QuerySet表示的“外键” )来实现

sganyu = Collection.objects.filter(detail__author='ganyu').all()

那么，如果查询内容含有python的tags的文章呢？构造一个包含python的查询集

sganyu = Collection.objects.filter(detail__tags__contains='python').all()

如何进行查找的呢？在Django中有两个基类，分别是Lookup（用于查找字符串）和Transform（用于转换字段），如以下的例子

参考连接：Lookup API reference | Django documentation | Django (djangoproject.com)

sganyu = Collection.objects.filter(detail__tags__contains='python').all()

__tags对应的是Transform，__contains对应的是'python'

__tags对应Transform，表如何去寻找关联的字段，就比如Collection.objects.filter(detail__author='ganyu').all()，就可以表示在author连接的用户表中找到ganyu
__contains对应'python'，表与后面的值进行对比

以上可以用sql语句理解为where 'users' lookup Transform 'value'

也就是select * from xxx where users.username = 'value'

到这里，JSONField用的KeyTransformFactory类返回KeyTransform对象，transform和lookup需要一个名为as_sql的方法用来生成SQL语句，如下


class KeyTransformFactory:
    def __init__(self, key_name):
        self.key_name = key_name
 
    def __call__(self, *args, **kwargs):
        return KeyTransform(self.key_name, *args, **kwargs)
 
class KeyTransform(Transform):
    operator = '->'
    nested_operator = '#>'
 
    def __init__(self, key_name, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.key_name = key_name
 
    def as_sql(self, compiler, connection):
        key_transforms = [self.key_name]
        previous = self.lhs
        while isinstance(previous, KeyTransform):
            key_transforms.insert(0, previous.key_name)
            previous = previous.lhs
        lhs, params = compiler.compile(previous)
        if len(key_transforms) > 1:
            return "(%s %s %%s)" % (lhs, self.nested_operator), [key_transforms] + params
        try:
            int(self.key_name)
        except ValueError:
            lookup = "'%s'" % self.key_name
        else:
            lookup = "%s" % self.key_name
        return "(%s %s %s)" % (lhs, self.operator, lookup), params