Add option to send oauth client credentials in body (#2509)

Co-authored-by: Jochen Ott <[email protected]>
Co-authored-by: Patryk Czajka <[email protected]>

Author: 3 people

DESCRIPTION.md

@@ -10,6 +10,7 @@ Source code is also available at: https://github.com/snowflakedb/snowflake-conne
 - v3.18.0(TBD)
   - Added the `workload_identity_impersonation_path` parameter to support service account impersonation for Workload Identity Federation on GCP and AWS workloads only
   - Fixed `get_results_from_sfqid` when using `DictCursor` and executing multiple statements at once
+  - Added the `oauth_credentials_in_body` parameter supporting an option to send the oauth client credentials in the request body
 
 - v3.17.3(September 02,2025)
   - Enhanced configuration file permission warning messages.

src/snowflake/connector/auth/oauth_credentials.py

@@ -27,6 +27,7 @@ def __init__(
         token_request_url: str,
         scope: str,
         connection: SnowflakeConnection | None = None,
+        credentials_in_body: bool = False,
         **kwargs,
     ) -> None:
         self._validate_client_credentials_present(client_id, client_secret, connection)
@@ -40,6 +41,7 @@ def __init__(
             **kwargs,
         )
         self._application = application
+        self._credentials_in_body = credentials_in_body
         self._origin: str | None = None
 
     def _get_oauth_type_id(self) -> str:
@@ -60,4 +62,7 @@ def _request_tokens(
             "grant_type": "client_credentials",
             "scope": self._scope,
         }
+        if self._credentials_in_body:
+            fields["client_id"] = self._client_id
+            fields["client_secret"] = self._client_secret
         return self._get_request_token_response(conn, fields)

src/snowflake/connector/connection.py

@@ -338,6 +338,11 @@ def _get_private_bytes_from_file(
         (type(None), str),
         # SNOW-1825621: OAUTH implementation
     ),
+    "oauth_credentials_in_body": (
+        False,
+        bool,
+        # SNOW-2300649: Option to send client credentials in body
+    ),
     "oauth_authorization_url": (
         "https://{host}:{port}/oauth/authorize",
         str,
@@ -1312,6 +1317,7 @@ def __open_connection(self):
                         host=self.host, port=self.port
                     ),
                     scope=self._oauth_scope,
+                    credentials_in_body=self._oauth_credentials_in_body,
                     connection=self,
                 )
             elif self._authenticator == USR_PWD_MFA_AUTHENTICATOR:

test/unit/test_auth_oauth_credentials.py

@@ -27,10 +27,15 @@ def test_auth_oauth_credentials_oauth_type():
 
 
 @pytest.mark.parametrize(
-    "authenticator", ["OAUTH_CLIENT_CREDENTIALS", "oauth_client_credentials"]
+    "authenticator, oauth_credentials_in_body",
+    [
+        ("OAUTH_CLIENT_CREDENTIALS", True),
+        ("oauth_client_credentials", False),
+        ("Oauth_Client_Credentials", None),
+    ],
 )
-def test_oauth_client_credentials_authenticator_is_case_insensitive(
-    monkeypatch, authenticator
+def test_oauth_client_credentials_parameters(
+    monkeypatch, authenticator, oauth_credentials_in_body
 ):
     """Test that OAuth client credentials authenticator is case insensitive."""
     import snowflake.connector
@@ -53,30 +58,59 @@ def mock_post_request(self, url, headers, json_body, **kwargs):
 
     # Mock the OAuth client credentials token request to avoid making HTTP requests
     def mock_get_request_token_response(self, connection, fields):
-        # Simulate successful token retrieval
+        # Return fields to verify they are set correctly in tests
         return (
-            "mock_access_token",
+            str(fields),
             None,
-        )  # Client credentials doesn't use refresh tokens
+        )
 
     monkeypatch.setattr(
         AuthByOauthCredentials,
         "_get_request_token_response",
         mock_get_request_token_response,
     )
 
+    oauth_credentials_in_body_arg = (
+        {"oauth_credentials_in_body": oauth_credentials_in_body}
+        if oauth_credentials_in_body is not None
+        else {}
+    )
     # Create connection with OAuth client credentials authenticator
     conn = snowflake.connector.connect(
         user="testuser",
         account="testaccount",
         authenticator=authenticator,
         oauth_client_id="test_client_id",
         oauth_client_secret="test_client_secret",
+        **oauth_credentials_in_body_arg,
     )
 
     # Verify that the auth_class is an instance of AuthByOauthCredentials
     assert isinstance(conn.auth_class, AuthByOauthCredentials)
 
+    # Verify that the credentials_in_body attribute is set correctly
+    expected_credentials_in_body = (
+        oauth_credentials_in_body if oauth_credentials_in_body is not None else False
+    )
+    assert conn.auth_class._credentials_in_body is expected_credentials_in_body
+
+    str_fields, _ = conn.auth_class._request_tokens(
+        conn=conn,
+        authenticator=authenticator,
+        account="<unused-acount>",
+        user="<unused-user>",
+        service_name=None,
+    )
+    credential_fields = (
+        ", 'client_id': 'test_client_id', 'client_secret': 'test_client_secret'"
+        if expected_credentials_in_body
+        else ""
+    )
+    assert (
+        str_fields
+        == "{'grant_type': 'client_credentials', 'scope': ''" + credential_fields + "}"
+    )
+
     conn.close()