feat: Add validate_and_sanitize_search_inputs decorator (#11194)

* feat: Add validate_and_sanitize_search_inputs decorator

Co-authored-by: Nabin Hait <nabinhait@gmail.com>
Co-authored-by: Prssanna Desai <prssud@gmail.com>
Co-authored-by: Shivam Mishra <scmmishra@users.noreply.github.com>

* refactor: Move validate_and_sanitize_search_inputs to init

* refactor: Move validate_and_sanitize_search_inputs to search.py

* test: validate_and_sanitize_search_inputs decorator

* chore: Add wrapt module

* refactor: Use @wrapt to define validate_and_sanitize_search_inputs decorator

* test: Add a case to make sure frappe.call works as well

Co-authored-by: Nabin Hait <nabinhait@gmail.com>
Co-authored-by: Prssanna Desai <prssud@gmail.com>
Co-authored-by: Shivam Mishra <scmmishra@users.noreply.github.com>

frappe/__init__.py

@@ -1707,3 +1707,7 @@ def mock(type, size=1, locale='en'):
 
 	from frappe.chat.util import squashify
 	return squashify(results)
+
+def validate_and_sanitize_search_inputs(fn):
+	from frappe.desk.search import validate_and_sanitize_search_inputs as func
+	return func(fn)

frappe/desk/search.py

@@ -10,6 +10,7 @@ from frappe.handler import is_whitelisted
 from frappe import _
 from six import string_types
 import re
+import wrapt
 
 UNTRANSLATED_DOCTYPES = ["DocType", "Role"]
 
@@ -206,3 +207,15 @@ def scrub_custom_query(query, key, txt):
 	if '%s' in query:
 		query = query.replace('%s', ((txt or '') + '%'))
 	return query
+
+@wrapt.decorator
+def validate_and_sanitize_search_inputs(fn, instance, args, kwargs):
+	kwargs.update(dict(zip(fn.__code__.co_varnames, args)))
+	sanitize_searchfield(kwargs['searchfield'])
+	kwargs['start'] = cint(kwargs['start'])
+	kwargs['page_len'] = cint(kwargs['page_len'])
+
+	if kwargs['doctype'] and not frappe.db.exists('DocType', kwargs['doctype']):
+		return []
+
+	return fn(**kwargs)

frappe/tests/test_search.py

@@ -50,3 +50,31 @@ class TestSearch(unittest.TestCase):
 
 	def tearDown(self):
 		frappe.local.lang = 'en'
+
+	def test_validate_and_sanitize_search_inputs(self):
+
+		# should raise error if searchfield is injectable
+		self.assertRaises(frappe.DataError,
+			get_data, *('User', 'Random', 'select * from tabSessions) --', '1', '10', dict()))
+
+		# page_len and start should be converted to int
+		self.assertListEqual(get_data('User', 'Random', 'email', 'name or (select * from tabSessions)', '10', dict()),
+			['User', 'Random', 'email', 0, 10, {}])
+		self.assertListEqual(get_data('User', 'Random', 'email', page_len='2', start='10', filters=dict()),
+			['User', 'Random', 'email', 10, 2, {}])
+
+		# DocType can be passed as None which should be accepted
+		self.assertListEqual(get_data(None, 'Random', 'email', '2', '10', dict()),
+			[None, 'Random', 'email', 2, 10, {}])
+
+		# return empty string if passed doctype is invalid
+		self.assertListEqual(get_data("Random DocType", 'Random', 'email', '2', '10', dict()), [])
+
+		# should not fail if function is called via frappe.call with extra arguments
+		args = ("Random DocType", 'Random', 'email', '2', '10', dict())
+		kwargs = {'as_dict': False}
+		self.assertListEqual(frappe.call('frappe.tests.test_search.get_data', *args, **kwargs), [])
+
+@frappe.validate_and_sanitize_search_inputs
+def get_data(doctype, txt, searchfield, start, page_len, filters):
+	return [doctype, txt, searchfield, start, page_len, filters]

requirements.txt

@@ -69,4 +69,5 @@ Whoosh==2.7.4
 xlrd==1.2.0
 zxcvbn-python==4.4.24
 pycryptodome==3.9.8
-paytmchecksum==1.7.0
+paytmchecksum==1.7.0
+wrapt==1.10.11