Merge pull request #7891 from TomeCirun/7812-fix-url-validator

[7812]Fix URL validator does not support ":" for specifying ports
ckan · Nov 8, 2023 · f922ec9 · f922ec9
2 parents 10acca1 + d21a9b9
commit f922ec9
Show file tree

Hide file tree

Showing 2 changed files with 59 additions and 9 deletions.
diff --git a/ckan/logic/validators.py b/ckan/logic/validators.py
@@ -772,25 +772,37 @@ def tag_not_in_vocabulary(key: FlattenKey, tag_dict: FlattenDataDict,
         return
 
 
-def url_validator(key: FlattenKey, data: FlattenDataDict,
-                  errors: FlattenErrorDict, context: Context) -> Any:
-    ''' Checks that the provided value (if it is present) is a valid URL '''
-
+def url_validator(
+    key: FlattenKey,
+    data: FlattenDataDict,
+    errors: FlattenErrorDict,
+    context: Context,
+) -> Any:
+    """Checks that the provided value (if it is present) is a valid URL"""
     url = data.get(key, None)
     if not url:
         return
 
     try:
         pieces = urlparse(url)
-        if all([pieces.scheme, pieces.netloc]) and \
-           set(pieces.netloc) <= set(string.ascii_letters + string.digits + '-.') and \
-           pieces.scheme in ['http', 'https']:
-            return
+        if all([pieces.scheme, pieces.netloc]) and pieces.scheme in [
+            "http",
+            "https",
+        ]:
+            hostname, port = (
+                pieces.netloc.split(":")
+                if ":" in pieces.netloc
+                else (pieces.netloc, None)
+            )
+            if set(hostname) <= set(
+                string.ascii_letters + string.digits + "-."
+            ) and (port is None or port.isdigit()):
+                return
     except ValueError:
         # url is invalid
         pass
 
-    errors[key].append(_('Please provide a valid URL'))
+    errors[key].append(_("Please provide a valid URL"))
 
 
 def user_name_exists(user_name: str, context: Context) -> Any:

diff --git a/ckan/tests/logic/test_validators.py b/ckan/tests/logic/test_validators.py
@@ -899,3 +899,41 @@ def convert(tag_string):
     assert convert("") == []
     assert convert("trailing comma,") == ["trailing comma"]
     assert convert("trailing comma space, ") == ["trailing comma space"]
+
+
+def test_url_validator():
+    key = ("url",)
+    errors = {(key): []}
+
+    # Test with a valid URL without a port
+    url = {("url",): "https://example.com"}
+    validators.url_validator(key, url, errors, {})
+    assert errors == {('url',): []}
+
+    # Test with a valid URL with a port
+    url = {("url",): "https://example.com:8080"}
+    validators.url_validator(key, url, errors, {})
+    assert errors == {('url',): []}
+
+    # Test with an invalid URL (invalid characters in hostname)
+    url = {("url",): "https://exa$mple.com"}
+    validators.url_validator(key, url, errors, {})
+    assert errors[key] == ['Please provide a valid URL']
+    errors[key].clear()
+
+    # Test with an invalid URL (invalid port)
+    url = {("url",): "https://example.com:80a80"}
+    validators.url_validator(key, url, errors, {})
+    assert errors[key] == ['Please provide a valid URL']
+    errors[key].clear()
+
+    # Test with an invalid URL (no scheme)
+    url = {("url",): "example.com"}
+    validators.url_validator(key, url, errors, {})
+    assert errors[key] == ['Please provide a valid URL']
+    errors[key].clear()
+
+    # Test with an invalid URL (unsupported scheme)
+    url = {("url",): "ftp://example.com"}
+    validators.url_validator(key, url, errors, {})
+    assert errors[key] == ['Please provide a valid URL']