fix(db_query): Space resilient sanitization (backport #18996) (#19044)

* fix(db_query): Space resilient matching

(cherry picked from commit 575d32e)

# Conflicts:
#	frappe/model/db_query.py

* test: Add more tests for illegal subquery and fn usage

(cherry picked from commit 1f91324)

* fix: Move function check inside subquery

(cherry picked from commit 1a5e5f5)

# Conflicts:
#	frappe/model/db_query.py

* fix: Strip white spaces on lower cased field value

Co-authored-by: Ankush Menat <ankushmenat@gmail.com>
(cherry picked from commit 35827af)

# Conflicts:
#	frappe/model/db_query.py

* fix: Resolve conflicts in #19044

Co-authored-by: Gavin D'souza <gavin18d@gmail.com>

frappe/model/db_query.py

@@ -347,14 +347,21 @@ def _is_query(field):
 				_raise_exception()
 
 		for field in self.fields:
-			if SUB_QUERY_PATTERN.match(field):
-				if any(f"({keyword}" in field.lower() for keyword in blacklisted_keywords):
-					_raise_exception()
+			lower_field = field.lower().strip()
 
-				if any("{0}(".format(keyword) in field.lower() for keyword in blacklisted_functions):
-					_raise_exception()
+			if SUB_QUERY_PATTERN.match(field):
+				if lower_field[0] == "(":
+					subquery_token = lower_field[1:].lstrip().split(" ", 1)[0]
+					if subquery_token in blacklisted_keywords:
+						_raise_exception()
+
+				function = lower_field.split("(", 1)[0].rstrip()
+				if function in blacklisted_functions:
+					frappe.throw(
+						_("Use of function {0} in field is restricted").format(function), exc=frappe.DataError
+					)
 
-				if "@" in field.lower():
+				if "@" in lower_field:
 					# prevent access to global variables
 					_raise_exception()
 
@@ -370,7 +377,7 @@ def _is_query(field):
 				if STRICT_FIELD_PATTERN.match(field):
 					frappe.throw(_("Illegal SQL Query"))
 
-				if STRICT_UNION_PATTERN.match(field.lower()):
+				if STRICT_UNION_PATTERN.match(lower_field):
 					frappe.throw(_("Illegal SQL Query"))
 
 	def extract_tables(self):

frappe/tests/test_db_query.py

@@ -282,6 +282,43 @@ def test_query_fields_sanitizer(self):
 			)
 			self.assertTrue("date_diff" in data[0])
 
+		with self.assertRaises(frappe.DataError):
+			DatabaseQuery("DocType").execute(
+				fields=["name", "issingle", "if (issingle=1, (select name from tabUser), count(name))"],
+				limit_start=0,
+				limit_page_length=1,
+			)
+
+		with self.assertRaises(frappe.DataError):
+			DatabaseQuery("DocType").execute(
+				fields=["name", "issingle", "if(issingle=1, (select name from tabUser), count(name))"],
+				limit_start=0,
+				limit_page_length=1,
+			)
+
+		with self.assertRaises(frappe.DataError):
+			DatabaseQuery("DocType").execute(
+				fields=[
+					"name",
+					"issingle",
+					"( select name from `tabUser` where `tabDocType`.owner = `tabUser`.name )",
+				],
+				limit_start=0,
+				limit_page_length=1,
+				ignore_permissions=True,
+			)
+
+		with self.assertRaises(frappe.DataError):
+			DatabaseQuery("DocType").execute(
+				fields=[
+					"name",
+					"issingle",
+					"(select name from `tabUser` where `tabDocType`.owner = `tabUser`.name )",
+				],
+				limit_start=0,
+				limit_page_length=1,
+			)
+
 	def test_nested_permission(self):
 		frappe.set_user("Administrator")
 		create_nested_doctype()