Read private key from env

requirements.txt

@@ -1,4 +1,5 @@
 Jinja2~=3.0
 PyYAML~=6.0
 snowflake-connector-python>=2.8,<4.0
+cryptography~=45.0.5
 structlog~=24.1.0

schemachange/config/DeployConfig.py

@@ -9,6 +9,7 @@
 from schemachange.config.utils import (
     get_snowflake_identifier_string,
     get_snowflake_password,
+    get_snowflake_private_key
 )
 
 
@@ -93,4 +94,8 @@ def get_session_kwargs(self) -> dict:
         if snowflake_password is not None and snowflake_password:
             session_kwargs["password"] = snowflake_password
 
+        snowflake_private_key = get_snowflake_private_key()
+        if snowflake_private_key:
+            session_kwargs["private_key"] = snowflake_private_key
+
         return {k: v for k, v in session_kwargs.items() if v is not None}

schemachange/config/utils.py

@@ -10,8 +10,11 @@
 import jinja2.ext
 import structlog
 import yaml
-from schemachange.JinjaEnvVar import JinjaEnvVar
 import warnings
+from schemachange.JinjaEnvVar import JinjaEnvVar
+from cryptography.hazmat.backends import default_backend
+from cryptography.hazmat.primitives import serialization
+from cryptography.hazmat.primitives.asymmetric.types import PrivateKeyTypes
 
 logger = structlog.getLogger(__name__)
 
@@ -161,3 +164,15 @@ def get_snowflake_password() -> str | None:
         return snowsql_pwd
     else:
         return None
+
+def get_snowflake_private_key() -> str | PrivateKeyTypes:
+    snowflake_private_key = os.getenv("SNOWFLAKE_PRIVATE_KEY")
+
+    if snowflake_private_key is not None and snowflake_private_key:
+        return serialization.load_pem_private_key(
+                snowflake_private_key.encode('utf-8'),
+                password=None,
+                backend=default_backend()
+            )
+
+    return None

schemachange/session/SnowflakeSession.py

@@ -63,6 +63,7 @@ def __init__(
             "schema": schema,  # TODO: Remove when connections.toml is enforced
             "role": role,  # TODO: Remove when connections.toml is enforced
             "warehouse": warehouse,  # TODO: Remove when connections.toml is enforced
+            "private_key": kwargs.get("private_key"),
             "private_key_file": kwargs.get(
                 "private_key_path"
             ),  # TODO: Remove when connections.toml is enforced