fix(db_query): Disallow usage of certain functions in *_by (#18981) (#…

…19134)

* fix(db_query): Disallow blacklisted functions in (order|group)_by

Changes:
- allow only functions that are not blacklisted in *_by clause:
  currently just sleep
- perf improvemnts: lower, in, split, strip & other low hanging  micro optimizations

Handle the following use cases:
- upper/lower case function usages
- spaces between function name and brackets

* test(db_query): Add tests for *_by checks

(cherry picked from commit 6062d81)

Co-authored-by: gavin <gavin18d@gmail.com>

frappe/model/db_query.py

@@ -873,26 +873,37 @@ def set_order_by(self, args):
 				if hasattr(meta, "is_submittable") and meta.is_submittable:
 					args.order_by = "`tab{0}`.docstatus asc, {1}".format(self.doctype, args.order_by)
 
-	def validate_order_by_and_group_by(self, parameters):
+	def validate_order_by_and_group_by(self, parameters: str):
 		"""Check order by, group by so that atleast one column is selected and does not have subquery"""
 		if not parameters:
 			return
 
+		blacklisted_sql_functions = {
+			"sleep",
+		}
 		_lower = parameters.lower()
+
 		if "select" in _lower and "from" in _lower:
 			frappe.throw(_("Cannot use sub-query in order by"))
 
 		if ORDER_GROUP_PATTERN.match(_lower):
 			frappe.throw(_("Illegal SQL Query"))
 
 		for field in parameters.split(","):
-			if "." in field and field.strip().startswith("`tab"):
-				tbl = field.strip().split(".")[0]
+			field = field.strip()
+			function = field.split("(", 1)[0].rstrip().lower()
+			full_field_name = "." in field and field.startswith("`tab")
+
+			if full_field_name:
+				tbl = field.split(".", 1)[0]
 				if tbl not in self.tables:
 					if tbl.startswith("`"):
 						tbl = tbl[4:-1]
 					frappe.throw(_("Please select atleast 1 column from {0} to sort/group").format(tbl))
 
+			if function in blacklisted_sql_functions:
+				frappe.throw(_("Cannot use {0} in order/group by").format(field))
+
 	def add_limit(self):
 		if self.limit_page_length:
 			return "limit %s offset %s" % (self.limit_page_length, self.limit_start)

frappe/tests/test_db_query.py

@@ -414,6 +414,46 @@ def test_filter_sanitizer(self):
 		)
 		self.assertTrue("DefaultValue" in [d["name"] for d in out])
 
+	def test_order_by_group_by_sanitizer(self):
+		# order by with blacklisted function
+		with self.assertRaises(frappe.ValidationError):
+			DatabaseQuery("DocType").execute(
+				fields=["name"],
+				order_by="sleep (1) asc",
+			)
+
+		# group by with blacklisted function
+		with self.assertRaises(frappe.ValidationError):
+			DatabaseQuery("DocType").execute(
+				fields=["name"],
+				group_by="SLEEP(0)",
+			)
+
+		# sub query in order by
+		with self.assertRaises(frappe.ValidationError):
+			DatabaseQuery("DocType").execute(
+				fields=["name"],
+				order_by="(select rank from tabRankedDocTypes where tabRankedDocTypes.name = tabDocType.name) asc",
+			)
+
+		# validate allowed usage
+		DatabaseQuery("DocType").execute(
+			fields=["name"],
+			order_by="name asc",
+		)
+		DatabaseQuery("DocType").execute(
+			fields=["name"],
+			order_by="name asc",
+			group_by="name",
+		)
+
+		# check mariadb specific syntax
+		if frappe.db.db_type == "mariadb":
+			DatabaseQuery("DocType").execute(
+				fields=["name"],
+				order_by="timestamp(modified)",
+			)
+
 	def test_of_not_of_descendant_ancestors(self):
 		frappe.set_user("Administrator")
 		clear_user_permissions_for_doctype("Nested DocType")