refactor: Add optional URL scheme validation

frappe/event_streaming/doctype/event_producer/event_producer.py

@@ -55,7 +55,8 @@ class EventProducer(Document):
 			self.reload()
 
 	def check_url(self):
-		frappe.utils.validate_url(self.producer_url, throw=True)
+		valid_url_schemes = ("http", "https", "ftp", "ftps")
+		frappe.utils.validate_url(self.producer_url, throw=True, valid_schemes=valid_url_schemes)
 
 		# remove '/' from the end of the url like http://test_site.com/
 		# to prevent mismatch in get_url() results

frappe/oauth.py

@@ -599,9 +599,10 @@ def get_client_scopes(client_id):
 def get_userinfo(user):
 	picture = None
 	frappe_server_url = get_server_url()
+	valid_url_schemes = ("http", "https", "ftp", "ftps")
 
 	if user.user_image:
-		if frappe.utils.validate_url(user.user_image):
+		if frappe.utils.validate_url(user.user_image, valid_schemes=valid_url_schemes):
 			picture = user.user_image
 		else:
 			picture = frappe_server_url + "/" + user.user_image

frappe/tests/test_utils.py

@@ -138,7 +138,7 @@ class TestValidationUtils(unittest.TestCase):
 
 		# Valid URLs
 		self.assertTrue(validate_url('https://google.com'))
-		self.assertTrue(validate_url('https://frappe.io', throw=True))
+		self.assertTrue(validate_url('http://frappe.io', throw=True))
 		
 		# Invalid URLs without throw
 		self.assertFalse(validate_url('google.io'))
@@ -147,6 +147,18 @@ class TestValidationUtils(unittest.TestCase):
 		# Invalid URL with throw
 		self.assertRaises(frappe.ValidationError, validate_url, 'frappe', throw=True)
 
+		# Scheme validation
+		self.assertFalse(validate_url('https://google.com', valid_schemes='http'))
+		self.assertTrue(validate_url('ftp://frappe.cloud', valid_schemes=['https', 'ftp']))
+		self.assertFalse(validate_url('bolo://frappe.io', valid_schemes=("http", "https", "ftp", "ftps")))
+		self.assertRaises(
+			frappe.ValidationError, 
+			validate_url, 
+			'bitcoin://joker.edu', 
+			valid_schemes='https',
+			throw=True
+		)
+
 	def test_valid_email(self):
 		# Edge cases
 		self.assertFalse(validate_email_address(''))

frappe/utils/__init__.py

@@ -155,21 +155,32 @@ def split_emails(txt):
 
 	return email_list
 
-def validate_url(txt, throw=False):
-	try:
-		url = urlparse(txt).netloc
-		if not url:
-			raise frappe.ValidationError
-		else:
-			return True
+def validate_url(txt, throw=False, valid_schemes=None):
+	"""
+		Tests wether the `txt` is a valid URL
 
-	except Exception:
-		if throw:
-			frappe.throw(
-				frappe._("'{0}' is not a valid URL").format(frappe.bold(txt))
-			)
-	
-	return False
+		Parameters:
+			throw (`bool`): throws a validationError if URL is not valid
+			valid_schemes (`str` or `list`): if provided checks the given URL's scheme against this 
+
+		Returns:
+			bool: if `txt` represents a valid URL
+	"""
+	url = urlparse(txt)
+	is_valid = bool(url.netloc)
+
+	# Handle scheme validation
+	if isinstance(valid_schemes, str):
+		is_valid = is_valid and (url.scheme == valid_schemes)
+	elif isinstance(valid_schemes, (list, tuple, set)):
+		is_valid = is_valid and (url.scheme in valid_schemes)
+
+	if not is_valid and throw:
+		frappe.throw(
+			frappe._("'{0}' is not a valid URL").format(frappe.bold(txt))
+		)
+
+	return is_valid
 
 def random_string(length):
 	"""generate a random string"""