From 93cae5cb2f9a4ef1514cf1a41f714fef08005200 Mon Sep 17 00:00:00 2001 From: Mariusz Felisiak Date: Fri, 1 Apr 2022 08:10:22 +0200 Subject: Fixed CVE-2022-28346 -- Protected QuerySet.annotate(), aggregate(), and extra() against SQL injection in column aliases. Thanks Splunk team: Preston Elder, Jacob Davis, Jacob Moore, Matt Hanson, David Briggs, and a security researcher: Danylo Dmytriiev (DDV_UA) for the report. --- django/db/models/sql/query.py | 14 ++++++++++++++ 1 file changed, 14 insertions(+) (limited to 'django/db/models/sql/query.py') diff --git a/django/db/models/sql/query.py b/django/db/models/sql/query.py index c07a4b342a..894aa7db4a 100644 --- a/django/db/models/sql/query.py +++ b/django/db/models/sql/query.py @@ -40,10 +40,15 @@ from django.db.models.sql.constants import INNER, LOUTER, ORDER_DIR, SINGLE from django.db.models.sql.datastructures import BaseTable, Empty, Join, MultiJoin from django.db.models.sql.where import AND, OR, ExtraWhere, NothingNode, WhereNode from django.utils.functional import cached_property +from django.utils.regex_helper import _lazy_re_compile from django.utils.tree import Node __all__ = ["Query", "RawQuery"] +# Quotation marks ('"`[]), whitespace characters, semicolons, or inline +# SQL comments are forbidden in column aliases. +FORBIDDEN_ALIAS_PATTERN = _lazy_re_compile(r"['`\"\]\[;\s]|--|/\*|\*/") + def get_field_names_from_opts(opts): if opts is None: @@ -1091,8 +1096,16 @@ class Query(BaseExpression): alias = seen[int_model] = join_info.joins[-1] return alias or seen[None] + def check_alias(self, alias): + if FORBIDDEN_ALIAS_PATTERN.search(alias): + raise ValueError( + "Column aliases cannot contain whitespace characters, quotation marks, " + "semicolons, or SQL comments." + ) + def add_annotation(self, annotation, alias, is_summary=False, select=True): """Add a single annotation expression to the Query.""" + self.check_alias(alias) annotation = annotation.resolve_expression( self, allow_joins=True, reuse=None, summarize=is_summary ) @@ -2269,6 +2282,7 @@ class Query(BaseExpression): else: param_iter = iter([]) for name, entry in select.items(): + self.check_alias(name) entry = str(entry) entry_params = [] pos = entry.find("%s") -- cgit v1.3