Snow-2923722: add private key passphrase kwargs to connection init (#2736)

Co-authored-by: yassun7010 <yassun7010@outlook.com>

Author: sfc-gh-fpawlowski

DESCRIPTION.md

@@ -16,6 +16,7 @@ Source code is also available at: https://github.com/snowflakedb/snowflake-conne
   - Added `SnowflakeCursor.stats` property to expose granular DML statistics (rows inserted, deleted, updated, and duplicates) for operations like CTAS where `rowcount` is insufficient.
   - Added support for injecting SPCS service identifier token (`SPCS_TOKEN`) into login requests when present in SPCS containers.
   - Introduced shared library([source code](https://github.com/snowflakedb/universal-driver/tree/main/sf_mini_core)) for extended telemetry to identify and prepare testing platform for native rust extensions.
+  - Added `private_key_passphrase` connection parameter for key pair authentication with encrypted private keys.
 
 - v4.1.1(December 12,2025)
   - Relaxed pandas dependency requirements for Python below 3.12.

src/snowflake/connector/aio/_connection.py

@@ -343,6 +343,7 @@ async def __open_connection(self):
 
             elif self._authenticator == KEY_PAIR_AUTHENTICATOR:
                 private_key = self._private_key
+                private_key_passphrase = self._private_key_passphrase
 
                 if self._private_key_file:
                     private_key = _get_private_bytes_from_file(
@@ -352,6 +353,7 @@ async def __open_connection(self):
 
                 self.auth_class = AuthByKeyPair(
                     private_key=private_key,
+                    private_key_passphrase=private_key_passphrase,
                     timeout=self.login_timeout,
                     backoff_generator=self._backoff_generator,
                 )

src/snowflake/connector/auth/_auth.py

@@ -633,8 +633,8 @@ def get_token_from_private_key(
     from . import AuthByKeyPair
 
     auth_instance = AuthByKeyPair(
-        private_key,
-        DAY_IN_SECONDS,
+        private_key=private_key,
+        lifetime_in_seconds=DAY_IN_SECONDS,
     )  # token valid for 24 hours
     return auth_instance.prepare(account=account, user=user)
 

src/snowflake/connector/auth/keypair.py

@@ -40,6 +40,7 @@ class AuthByKeyPair(AuthByPlugin):
     def __init__(
         self,
         private_key: bytes | str | RSAPrivateKey,
+        private_key_passphrase: bytes | None = None,
         lifetime_in_seconds: int = LIFETIME,
         **kwargs,
     ) -> None:
@@ -72,6 +73,7 @@ def __init__(
         )
 
         self._private_key: bytes | str | RSAPrivateKey | None = private_key
+        self._private_key_passphrase: bytes | None = private_key_passphrase
         self._jwt_token = ""
         self._jwt_token_exp = 0
         self._lifetime = timedelta(
@@ -116,13 +118,14 @@ def prepare(
             try:
                 private_key = load_der_private_key(
                     data=self._private_key,
-                    password=None,
+                    password=self._private_key_passphrase,
                     backend=default_backend(),
                 )
             except Exception as e:
                 raise ProgrammingError(
                     msg=f"Failed to load private key: {e}\nPlease provide a valid "
-                    "unencrypted rsa private key in DER format as bytes object",
+                    "rsa private key in DER format as bytes object. If the key is "
+                    "encrypted, provide the passphrase via private_key_passphrase",
                     errno=ER_INVALID_PRIVATE_KEY,
                 )
 

src/snowflake/connector/connection.py

@@ -232,6 +232,7 @@ def _get_private_bytes_from_file(
     "passcode_in_password": (False, bool),  # Snowflake MFA
     "passcode": (None, (type(None), str)),  # Snowflake MFA
     "private_key": (None, (type(None), bytes, str, RSAPrivateKey)),
+    "private_key_passphrase": (None, (type(None), bytes)),
     "private_key_file": (None, (type(None), str)),
     "private_key_file_pwd": (None, (type(None), str, bytes)),
     "token": (None, (type(None), str)),  # OAuth/JWT/PAT/OIDC Token
@@ -1472,6 +1473,7 @@ def __open_connection(self):
 
             elif self._authenticator == KEY_PAIR_AUTHENTICATOR:
                 private_key = self._private_key
+                private_key_passphrase = self._private_key_passphrase
 
                 if self._private_key_file:
                     private_key = _get_private_bytes_from_file(
@@ -1481,6 +1483,7 @@ def __open_connection(self):
 
                 self.auth_class = AuthByKeyPair(
                     private_key=private_key,
+                    private_key_passphrase=private_key_passphrase,
                     timeout=self.login_timeout,
                     backoff_generator=self._backoff_generator,
                 )

test/unit/test_auth_keypair.py

@@ -64,6 +64,86 @@ def test_auth_keypair(authenticator):
     assert rest.master_token == "MASTER_TOKEN"
 
 
+def test_auth_keypair_with_passphrase():
+    """Simple Key Pair test with passphrase."""
+
+    passphrase = b"test"
+    private_key_der, public_key_der_encoded = generate_key_pair(
+        2048,
+        passphrase=passphrase,
+    )
+    application = "testapplication"
+    account = "testaccount"
+    user = "testuser"
+    auth_instance = AuthByKeyPair(
+        private_key=private_key_der,
+        private_key_passphrase=passphrase,
+    )
+    auth_instance._retry_ctx.set_start_time()
+    auth_instance.handle_timeout(
+        authenticator="SNOWFLAKE_JWT",
+        service_name=None,
+        account=account,
+        user=user,
+        password=None,
+    )
+
+    # success test case
+    rest = _init_rest(application, _create_mock_auth_keypair_rest_response())
+    auth = Auth(rest)
+    auth.authenticate(auth_instance, account, user)
+    assert not rest._connection.errorhandler.called  # not error
+    assert rest.token == "TOKEN"
+    assert rest.master_token == "MASTER_TOKEN"
+
+
+def test_auth_keypair_encrypted_without_passphrase():
+    """Test that encrypted key without passphrase raises error with helpful message."""
+    from snowflake.connector.errors import ProgrammingError
+
+    passphrase = b"test"
+    private_key_der, _ = generate_key_pair(
+        2048,
+        passphrase=passphrase,
+    )
+    account = "testaccount"
+    user = "testuser"
+
+    # Create auth instance without providing passphrase for encrypted key
+    auth_instance = AuthByKeyPair(private_key=private_key_der)
+
+    with raises(ProgrammingError) as ex:
+        auth_instance.prepare(account=account, user=user)
+
+    # Verify the error message mentions the passphrase option
+    assert "private_key_passphrase" in str(ex.value)
+
+
+def test_auth_keypair_wrong_passphrase():
+    """Test that wrong passphrase raises error."""
+    from snowflake.connector.errors import ProgrammingError
+
+    passphrase = b"correct_passphrase"
+    private_key_der, _ = generate_key_pair(
+        2048,
+        passphrase=passphrase,
+    )
+    account = "testaccount"
+    user = "testuser"
+
+    # Create auth instance with wrong passphrase
+    auth_instance = AuthByKeyPair(
+        private_key=private_key_der,
+        private_key_passphrase=b"wrong_passphrase",
+    )
+
+    with raises(ProgrammingError) as ex:
+        auth_instance.prepare(account=account, user=user)
+
+    # Verify the error mentions the private key loading failure
+    assert "Failed to load private key" in str(ex.value)
+
+
 def test_auth_prepare_body_does_not_overwrite_client_environment_fields():
     private_key_der, _ = generate_key_pair(2048)
     auth_class = AuthByKeyPair(private_key=private_key_der)
@@ -169,15 +249,19 @@ def _init_rest(application, post_requset):
     return rest
 
 
-def generate_key_pair(key_length):
+def generate_key_pair(key_length: int, *, passphrase: bytes | None = None):
     private_key = rsa.generate_private_key(
         backend=default_backend(), public_exponent=65537, key_size=key_length
     )
 
     private_key_der = private_key.private_bytes(
         encoding=serialization.Encoding.DER,
         format=serialization.PrivateFormat.PKCS8,
-        encryption_algorithm=serialization.NoEncryption(),
+        encryption_algorithm=(
+            serialization.BestAvailableEncryption(passphrase)
+            if passphrase
+            else serialization.NoEncryption()
+        ),
     )
 
     public_key_pem = (