CVE-2023-34233: Improper Neutralization of Special Elements used in a Command ('Command Injection') (CWE-77)
value: The Snowflake Connector for Python provides an interface for developing Python applications that can connect to Snowflake and perform all standard operations. Versions prior to 3.0.2 are vulnerable to command injection via single sign-on(SSO) browser URL authentication. In order to exploit the potential for command injection, an attacker would need to be successful in (1) establishing a malicious resource and (2) redirecting users to utilize the resource. The attacker could set up a malicious, publicly accessible server which responds to the SSO URL with an attack payload. If the attacker then tricked a user into visiting the maliciously crafted connection URL, the user’s local machine would render the malicious payload, leading to a remote code execution. This attack scenario can be mitigated through URL whitelisting as well as common anti-phishing resources. Version 3.0.2 contains a patch for this issue.
CWE description: The product constructs all or part of a command using externally-influenced input from an upstream component, but it does not neutralize or incorrectly neutralizes special elements that could modify the intended command when it is sent to a downstream component.
CVSS v3 base score: 8.8
Published: 2023-06-08T21:15Z
Repository: https://github.com/snowflakedb/snowflake-connector-python
Language: Python
Files changed: .pre-commit-config.yaml, DESCRIPTION.md, setup.py, src/snowflake/connector/auth/webbrowser.py, src/snowflake/connector/snow_logging.py, src/snowflake/connector/url_util.py, test/unit/test_auth_webbrowser.py, test/unit/test_url_util.py
Commit date: 2023-03-23 00:16:55 +0000
Commit message: SNOW-761004 Added URL Validator and URL escaping of strings (#1480)


Diff: diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml
index 25169e2..a4b1e24 100644
--- a/.pre-commit-config.yaml
+++ b/.pre-commit-config.yaml
@@ -91,10 +91,11 @@ repos:
                 | snow_logging
                 | sqlstate
                 | ssd_internal_keys
                 | test_util
                 | util_text
+                | url_util
                 | version
             ).py$
         additional_dependencies:
             - types-requests
             - types-urllib3
diff --git a/DESCRIPTION.md b/DESCRIPTION.md
index 511b0ca..05287f3 100644
--- a/DESCRIPTION.md
+++ b/DESCRIPTION.md
@@ -10,10 +10,11 @@ Source code is also available at: https://github.com/snowflakedb/snowflake-conne
 
 - v3.0.2(Unreleased)
 
   - 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)
 
   - Improved the robustness of OCSP response caching to handle errors in cases of serialization and deserialization.
   - Updated async_executes method's doc-string.
diff --git a/setup.py b/setup.py
index 0933604..93ff83e 100644
--- a/setup.py
+++ b/setup.py
@@ -47,15 +47,17 @@ try:
     from Cython.Build import cythonize
     from Cython.Distutils import build_ext
 
     _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(
         [
             Extension(
                 name="snowflake.connector.arrow_iterator",
@@ -64,11 +66,10 @@ if _ABLE_TO_COMPILE_EXTENSIONS:
         ],
         compile_time_env=dict(ARROW_LESS_THAN_8=pyarrow_version < (8,)),
     )
 
     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
         arrow_libs_to_copy = {
             "linux": [
diff --git a/src/snowflake/connector/auth/webbrowser.py b/src/snowflake/connector/auth/webbrowser.py
index 527b10e..3c1b271 100644
--- a/src/snowflake/connector/auth/webbrowser.py
+++ b/src/snowflake/connector/auth/webbrowser.py
@@ -21,19 +21,21 @@ from ..constants import (
     HTTP_HEADER_SERVICE_NAME,
     HTTP_HEADER_USER_AGENT,
 )
 from ..errorcode import (
     ER_IDP_CONNECTION_ERROR,
+    ER_INVALID_VALUE,
     ER_NO_HOSTNAME_FOUND,
     ER_UNABLE_TO_OPEN_BROWSER,
 )
 from ..errors import OperationalError
 from ..network import (
     CONTENT_TYPE_APPLICATION_JSON,
     EXTERNAL_BROWSER_AUTHENTICATOR,
     PYTHON_CONNECTOR_USER_AGENT,
 )
+from ..url_util import is_valid_url
 from . import Auth
 from .by_plugin import AuthByPlugin, AuthType
 
 if TYPE_CHECKING:
     from .. import SnowflakeConnection
@@ -129,22 +131,33 @@ class AuthByWebBrowser(AuthByPlugin):
                 else:
                     raise ex
             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):
                 print(
                     "We were unable to open a browser window for you, "
diff --git a/src/snowflake/connector/snow_logging.py b/src/snowflake/connector/snow_logging.py
index 665c2d1..2e639f2 100644
--- a/src/snowflake/connector/snow_logging.py
+++ b/src/snowflake/connector/snow_logging.py
@@ -65,11 +65,11 @@ class SnowLogger(logging.LoggerAdapter):
         **kwargs: Any,
     ) -> None:
         warnings.warn(
             "The 'warn' method is deprecated, " "use 'warning' instead",
             DeprecationWarning,
-            2,
+            stacklevel=2,
         )
         self.warning(msg, path_name, func_name, *args, **kwargs)
 
     def error(  # type: ignore[override]
         self,
diff --git a/src/snowflake/connector/url_util.py b/src/snowflake/connector/url_util.py
new file mode 100644
index 0000000..d202502
--- /dev/null
+++ b/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="")
diff --git a/test/unit/test_auth_webbrowser.py b/test/unit/test_auth_webbrowser.py
index 39c9b94..6791a01 100644
--- a/test/unit/test_auth_webbrowser.py
+++ b/test/unit/test_auth_webbrowser.py
@@ -30,10 +30,11 @@ ACCOUNT = "testaccount"
 USER = "testuser"
 PASSWORD = "testpassword"
 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):
     _ = application
     _ = port
@@ -316,5 +317,41 @@ def test_idtoken_reauth():
                 SnowflakeConnection(
                     user="user",
                     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
diff --git a/test/unit/test_url_util.py b/test/unit/test_url_util.py
new file mode 100644
index 0000000..8ed8d52
--- /dev/null
+++ b/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) == ""

Vulnerable code:     warnings.warn("Cannot compile native C code, because of a missing build dependency")


            logger.debug("step 1: query GS to obtain SSO url")
            sso_url = self._get_sso_url(
                conn, authenticator, service_name, account, callback_port, user
            )

            2,
Fixed code:                 | url_util
  - Validate SSO URL before opening it in the browser for External browser authenticator.
    warnings.warn(
        "Cannot compile native C code, because of a missing build dependency",
        stacklevel=1,
    )
    ER_INVALID_VALUE,
from ..url_util import is_valid_url
            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

            stacklevel=2,
#
# 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="")
INVALID_SSO_URL = "this is an invalid URL"


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
#
# 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) == ""