SNOW-761004 Added URL Validator and URL escaping of strings (#1480)

.pre-commit-config.yaml

@@ -93,6 +93,7 @@ repos:
                 | ssd_internal_keys
                 | test_util
                 | util_text
+                | url_util
                 | version
             ).py$
         additional_dependencies:

DESCRIPTION.md

@@ -12,6 +12,7 @@ Source code is also available at: https://github.com/snowflakedb/snowflake-conne
 
   - Fixed a memory leak in the logging module of the Cython extension.
   - Fixed a bug where the `put` command on AWS raised `AttributeError` when the file size was larger than 200M.
+  - Validate SSO URL before opening it in the browser for External browser authenticator.
 
 - v3.0.1(February 28, 2023)
 

setup.py

@@ -49,11 +49,13 @@
 
     _ABLE_TO_COMPILE_EXTENSIONS = True
 except ImportError:
-    warnings.warn("Cannot compile native C code, because of a missing build dependency")
+    warnings.warn(
+        "Cannot compile native C code, because of a missing build dependency",
+        stacklevel=1,
+    )
     _ABLE_TO_COMPILE_EXTENSIONS = False
 
 if _ABLE_TO_COMPILE_EXTENSIONS:
-
     pyarrow_version = tuple(int(x) for x in pyarrow.__version__.split("."))
     extensions = cythonize(
         [
@@ -66,7 +68,6 @@
     )
 
     class MyBuildExt(build_ext):
-
         # list of libraries that will be bundled with python connector,
         # this list should be carefully examined when pyarrow lib is
         # upgraded

src/snowflake/connector/auth/webbrowser.py

@@ -23,6 +23,7 @@
 )
 from ..errorcode import (
     ER_IDP_CONNECTION_ERROR,
+    ER_INVALID_VALUE,
     ER_NO_HOSTNAME_FOUND,
     ER_UNABLE_TO_OPEN_BROWSER,
 )
@@ -32,6 +33,7 @@
     EXTERNAL_BROWSER_AUTHENTICATOR,
     PYTHON_CONNECTOR_USER_AGENT,
 )
+from ..url_util import is_valid_url
 from . import Auth
 from .by_plugin import AuthByPlugin, AuthType
 
@@ -131,18 +133,29 @@ def prepare(
             socket_connection.listen(0)  # no backlog
             callback_port = socket_connection.getsockname()[1]
 
+            logger.debug("step 1: query GS to obtain SSO url")
+            sso_url = self._get_sso_url(
+                conn, authenticator, service_name, account, callback_port, user
+            )
+
+            logger.debug("Validate SSO URL")
+            if not is_valid_url(sso_url):
+                self._handle_failure(
+                    conn=conn,
+                    ret={
+                        "code": ER_INVALID_VALUE,
+                        "message": (f"The SSO URL provided {sso_url} is invalid"),
+                    },
+                )
+                return
+
             print(
                 "Initiating login request with your identity provider. A "
                 "browser window should have opened for you to complete the "
                 "login. If you can't see it, check existing browser windows, "
                 "or your OS settings. Press CTRL+C to abort and try again..."
             )
 
-            logger.debug("step 1: query GS to obtain SSO url")
-            sso_url = self._get_sso_url(
-                conn, authenticator, service_name, account, callback_port, user
-            )
-
             logger.debug("step 2: open a browser")
             print(f"Going to open: {sso_url} to authenticate...")
             if not self._webbrowser.open_new(sso_url):

src/snowflake/connector/snow_logging.py

@@ -67,7 +67,7 @@ def warn(  # type: ignore[override]
         warnings.warn(
             "The 'warn' method is deprecated, " "use 'warning' instead",
             DeprecationWarning,
-            2,
+            stacklevel=2,
         )
         self.warning(msg, path_name, func_name, *args, **kwargs)
 

src/snowflake/connector/url_util.py

@@ -0,0 +1,43 @@
+#
+# Copyright (c) 2012-2023 Snowflake Computing Inc. All rights reserved.
+#
+
+from __future__ import annotations
+
+import re
+import urllib.parse
+from logging import getLogger
+
+logger = getLogger(__name__)
+
+
+URL_VALIDATOR = re.compile(
+    "^http(s?)\\:\\/\\/[0-9a-zA-Z]([-.\\w]*[0-9a-zA-Z@:])*(:(0-9)*)*(\\/?)([a-zA-Z0-9\\-\\.\\?\\,\\&\\(\\)\\/\\\\\\+&%\\$#_=@]*)?$"
+)
+
+
+def is_valid_url(url: str) -> bool:
+    """Confirms if the provided URL is a valid HTTP/ HTTPs URL
+
+    Args:
+        url: the URL that needs to be validated
+
+    Returns:
+        true/ false depending on whether the URL is valid or not
+    """
+    return bool(URL_VALIDATOR.match(url))
+
+
+def url_encode_str(target: str | None) -> str:
+    """Converts a target string into escaped URL safe string
+
+    Args:
+        target: string to be URL encoded
+
+    Returns:
+        URL encoded string
+    """
+    if target is None:
+        logger.debug("The string to be URL encoded is None")
+        return ""
+    return urllib.parse.quote_plus(target, safe="")

test/unit/test_auth_webbrowser.py

@@ -32,6 +32,7 @@
 SERVICE_NAME = ""
 REF_PROOF_KEY = "MOCK_PROOF_KEY"
 REF_SSO_URL = "https://testsso.snowflake.net/sso"
+INVALID_SSO_URL = "this is an invalid URL"
 
 
 def mock_webserver(target_instance, application, port):
@@ -318,3 +319,39 @@ class StopExecuting(Exception):
                     account="account",
                     auth_class=auth_inst,
                 )
+
+
+def test_auth_webbrowser_invalid_sso(monkeypatch):
+    """Authentication by WebBrowser with failed to start web browser case."""
+    rest = _init_rest(INVALID_SSO_URL, REF_PROOF_KEY)
+
+    # mock webbrowser
+    mock_webbrowser = MagicMock()
+    mock_webbrowser.open_new.return_value = False
+
+    # mock socket
+    mock_socket_instance = MagicMock()
+    mock_socket_instance.getsockname.return_value = [None, 12345]
+
+    mock_socket_client = MagicMock()
+    mock_socket_client.recv.return_value = (
+        "\r\n".join(["GET /?token=MOCK_TOKEN HTTP/1.1", "User-Agent: snowflake-agent"])
+    ).encode("utf-8")
+    mock_socket_instance.accept.return_value = (mock_socket_client, None)
+    mock_socket = Mock(return_value=mock_socket_instance)
+
+    auth = AuthByWebBrowser(
+        application=APPLICATION,
+        webbrowser_pkg=mock_webbrowser,
+        socket_pkg=mock_socket,
+    )
+    auth.prepare(
+        conn=rest._connection,
+        authenticator=AUTHENTICATOR,
+        service_name=SERVICE_NAME,
+        account=ACCOUNT,
+        user=USER,
+        password=PASSWORD,
+    )
+    assert rest._connection.errorhandler.called  # an error
+    assert auth.assertion_content is None

test/unit/test_url_util.py

@@ -0,0 +1,26 @@
+#
+# Copyright (c) 2012-2023 Snowflake Computing Inc. All rights reserved.
+#
+
+try:
+    from snowflake.connector.url_util import is_valid_url, url_encode_str
+except ImportError:
+
+    def is_valid_url(s):
+        return False
+
+
+def test_url_validator():
+    assert is_valid_url("https://ssoTestURL.okta.com")
+    assert is_valid_url("https://ssoTestURL.okta.com:8080")
+    assert is_valid_url("https://ssoTestURL.okta.com/testpathvalue")
+
+    assert not is_valid_url("-a Calculator")
+    assert not is_valid_url("This is a random text")
+    assert not is_valid_url("file://TestForFile")
+
+
+def test_encoder():
+    assert url_encode_str("Hello @World") == "Hello+%40World"
+    assert url_encode_str("Test//String") == "Test%2F%2FString"
+    assert url_encode_str(None) == ""