From 63fe87343b92f235d5f6bdededed47a846beb3cc Mon Sep 17 00:00:00 2001 From: Joshua Janicas Date: Thu, 25 Jun 2026 14:30:53 -0400 Subject: [PATCH 1/6] fix: Choose proper pem vs der format instead of always using pem --- target_snowflake/connector.py | 21 +++++++++++++++------ 1 file changed, 15 insertions(+), 6 deletions(-) diff --git a/target_snowflake/connector.py b/target_snowflake/connector.py index 7dd6985..4afba55 100644 --- a/target_snowflake/connector.py +++ b/target_snowflake/connector.py @@ -165,6 +165,11 @@ def get_private_key(self): raise FileNotFoundError(error_message) with key_path.open("rb") as key_file: key_content = key_file.read() + p_key = serialization.load_pem_private_key( + key_content, + password=encoded_passphrase, + backend=default_backend(), + ) else: private_key = self.config["private_key"] self.logger.debug("Reading private key from config") @@ -175,7 +180,11 @@ def get_private_key(self): stacklevel=2, ) self.logger.info("Private key is in PEM format") - key_content = private_key.encode() + p_key = serialization.load_pem_private_key( + private_key.encode(), + password=encoded_passphrase, + backend=default_backend(), + ) else: try: self.logger.debug("Private key is in base64 format") @@ -183,11 +192,11 @@ def get_private_key(self): except binascii.Error as e: error_message = f"Invalid private key format: {e}" raise ValueError(error_message) from e - p_key = serialization.load_pem_private_key( - key_content, - password=encoded_passphrase, - backend=default_backend(), - ) + p_key = serialization.load_der_private_key( + key_content, + password=encoded_passphrase, + backend=default_backend(), + ) return p_key.private_bytes( encoding=serialization.Encoding.DER, From e367b55fd039707dd2836053cf3831bdabbbbd8b Mon Sep 17 00:00:00 2001 From: Joshua Janicas Date: Thu, 25 Jun 2026 15:06:41 -0400 Subject: [PATCH 2/6] fix: Fallback to PEM if DER doesn't work --- target_snowflake/connector.py | 25 ++++++++++++++++++------- 1 file changed, 18 insertions(+), 7 deletions(-) diff --git a/target_snowflake/connector.py b/target_snowflake/connector.py index 4afba55..961a8d6 100644 --- a/target_snowflake/connector.py +++ b/target_snowflake/connector.py @@ -186,17 +186,28 @@ def get_private_key(self): backend=default_backend(), ) else: + # Decode the base64-encoded private key bytes. try: self.logger.debug("Private key is in base64 format") key_content = base64.b64decode(private_key) except binascii.Error as e: - error_message = f"Invalid private key format: {e}" - raise ValueError(error_message) from e - p_key = serialization.load_der_private_key( - key_content, - password=encoded_passphrase, - backend=default_backend(), - ) + raise ValueError(f"Invalid private key format: {e}") from e + + # Try DER first; fall back to PEM for backward compatibility. + self.logger.debug("Attempting serialization of private key as DER") + try: + p_key = serialization.load_der_private_key( + key_content, + password=encoded_passphrase, + backend=default_backend(), + ) + except ValueError: + self.logger.debug("DER deserialization failed; retrying as PEM") + p_key = serialization.load_pem_private_key( + key_content, + password=encoded_passphrase, + backend=default_backend(), + ) return p_key.private_bytes( encoding=serialization.Encoding.DER, From ab42759bd6709be617a53fc002cfd65ec490931d Mon Sep 17 00:00:00 2001 From: Joshua Janicas Date: Thu, 25 Jun 2026 15:20:22 -0400 Subject: [PATCH 3/6] Undo some stylistic changes --- target_snowflake/connector.py | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/target_snowflake/connector.py b/target_snowflake/connector.py index 961a8d6..7c81b7b 100644 --- a/target_snowflake/connector.py +++ b/target_snowflake/connector.py @@ -180,8 +180,9 @@ def get_private_key(self): stacklevel=2, ) self.logger.info("Private key is in PEM format") + key_content = private_key.encode() p_key = serialization.load_pem_private_key( - private_key.encode(), + key_content, password=encoded_passphrase, backend=default_backend(), ) @@ -191,7 +192,8 @@ def get_private_key(self): self.logger.debug("Private key is in base64 format") key_content = base64.b64decode(private_key) except binascii.Error as e: - raise ValueError(f"Invalid private key format: {e}") from e + error_message = f"Invalid private key format: {e}" + raise ValueError(error_message) from e # Try DER first; fall back to PEM for backward compatibility. self.logger.debug("Attempting serialization of private key as DER") @@ -202,7 +204,7 @@ def get_private_key(self): backend=default_backend(), ) except ValueError: - self.logger.debug("DER deserialization failed; retrying as PEM") + self.logger.info("DER deserialization failed; retrying as PEM") p_key = serialization.load_pem_private_key( key_content, password=encoded_passphrase, From 77d3b0ffecc4a79dc80696a6301eee48efb44eb8 Mon Sep 17 00:00:00 2001 From: Joshua Janicas Date: Thu, 25 Jun 2026 15:21:57 -0400 Subject: [PATCH 4/6] Remove unnecessary comment Make debug instead of info Remove comment --- target_snowflake/connector.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/target_snowflake/connector.py b/target_snowflake/connector.py index 7c81b7b..4a1cb91 100644 --- a/target_snowflake/connector.py +++ b/target_snowflake/connector.py @@ -187,7 +187,6 @@ def get_private_key(self): backend=default_backend(), ) else: - # Decode the base64-encoded private key bytes. try: self.logger.debug("Private key is in base64 format") key_content = base64.b64decode(private_key) @@ -195,7 +194,6 @@ def get_private_key(self): error_message = f"Invalid private key format: {e}" raise ValueError(error_message) from e - # Try DER first; fall back to PEM for backward compatibility. self.logger.debug("Attempting serialization of private key as DER") try: p_key = serialization.load_der_private_key( @@ -204,7 +202,7 @@ def get_private_key(self): backend=default_backend(), ) except ValueError: - self.logger.info("DER deserialization failed; retrying as PEM") + self.logger.debug("DER deserialization failed; retrying as PEM") p_key = serialization.load_pem_private_key( key_content, password=encoded_passphrase, From b34acfef223de32b0fd992ed2e6f4f9b0e6b592d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Edgar=20Ram=C3=ADrez=20Mondrag=C3=B3n?= Date: Tue, 14 Jul 2026 18:12:13 -0600 Subject: [PATCH 5/6] Refactor MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Edgar Ramírez Mondragón --- target_snowflake/connector.py | 89 ++++++++++++++++++----------------- 1 file changed, 45 insertions(+), 44 deletions(-) diff --git a/target_snowflake/connector.py b/target_snowflake/connector.py index 4a1cb91..858e13a 100644 --- a/target_snowflake/connector.py +++ b/target_snowflake/connector.py @@ -35,6 +35,7 @@ from collections.abc import Generator, Iterable, Sequence import sqlalchemy as sa + from cryptography.hazmat.primitives.asymmetric.types import PrivateKeyTypes from sqlalchemy.engine import Engine @@ -153,7 +154,7 @@ def _convert_type(sql_type): # noqa: ANN205, ANN001 return sql_type - def get_private_key(self): + def _get_private_key(self) -> PrivateKeyTypes: """Get private key from the right location.""" phrase = self.config.get("private_key_passphrase") encoded_passphrase = phrase.encode() if phrase else None @@ -163,53 +164,53 @@ def get_private_key(self): if not key_path.is_file(): error_message = f"Private key file not found: {key_path}" raise FileNotFoundError(error_message) - with key_path.open("rb") as key_file: - key_content = key_file.read() - p_key = serialization.load_pem_private_key( + + return serialization.load_pem_private_key( + key_path.read_bytes(), + password=encoded_passphrase, + backend=default_backend(), + ) + + private_key: str = self.config["private_key"] + self.logger.debug("Reading private key from config") + if "-----BEGIN " in private_key: + warn( + "Use base64 encoded private key instead of PEM format", + DeprecationWarning, + stacklevel=2, + ) + self.logger.info("Private key is in PEM format") + key_content = private_key.encode() + return serialization.load_pem_private_key( key_content, password=encoded_passphrase, backend=default_backend(), ) - else: - private_key = self.config["private_key"] - self.logger.debug("Reading private key from config") - if "-----BEGIN " in private_key: - warn( - "Use base64 encoded private key instead of PEM format", - DeprecationWarning, - stacklevel=2, - ) - self.logger.info("Private key is in PEM format") - key_content = private_key.encode() - p_key = serialization.load_pem_private_key( - key_content, - password=encoded_passphrase, - backend=default_backend(), - ) - else: - try: - self.logger.debug("Private key is in base64 format") - key_content = base64.b64decode(private_key) - except binascii.Error as e: - error_message = f"Invalid private key format: {e}" - raise ValueError(error_message) from e - - self.logger.debug("Attempting serialization of private key as DER") - try: - p_key = serialization.load_der_private_key( - key_content, - password=encoded_passphrase, - backend=default_backend(), - ) - except ValueError: - self.logger.debug("DER deserialization failed; retrying as PEM") - p_key = serialization.load_pem_private_key( - key_content, - password=encoded_passphrase, - backend=default_backend(), - ) - - return p_key.private_bytes( + + try: + self.logger.debug("Private key is in base64 format") + key_content = base64.b64decode(private_key) + except binascii.Error as e: + error_message = f"Invalid private key format: {e}" + raise ValueError(error_message) from e + + self.logger.debug("Attempting serialization of private key as DER") + try: + return serialization.load_der_private_key( + key_content, + password=encoded_passphrase, + backend=default_backend(), + ) + except ValueError: + self.logger.debug("DER deserialization failed; retrying as PEM") + return serialization.load_pem_private_key( + key_content, + password=encoded_passphrase, + backend=default_backend(), + ) + + def get_private_key(self): + return self._get_private_key().private_bytes( encoding=serialization.Encoding.DER, format=serialization.PrivateFormat.PKCS8, encryption_algorithm=serialization.NoEncryption(), From 0208e1ff69fd2ac094d69f2ef395c611a0341823 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Edgar=20Ram=C3=ADrez=20Mondrag=C3=B3n?= Date: Tue, 14 Jul 2026 18:17:15 -0600 Subject: [PATCH 6/6] Moar refactor MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Edgar Ramírez Mondragón --- target_snowflake/connector.py | 27 +++++++++++---------------- 1 file changed, 11 insertions(+), 16 deletions(-) diff --git a/target_snowflake/connector.py b/target_snowflake/connector.py index 858e13a..88312b2 100644 --- a/target_snowflake/connector.py +++ b/target_snowflake/connector.py @@ -154,10 +154,8 @@ def _convert_type(sql_type): # noqa: ANN205, ANN001 return sql_type - def _get_private_key(self) -> PrivateKeyTypes: + def _get_private_key_content(self) -> bytes: """Get private key from the right location.""" - phrase = self.config.get("private_key_passphrase") - encoded_passphrase = phrase.encode() if phrase else None if "private_key_path" in self.config: self.logger.debug("Reading private key from file: %s", self.config["private_key_path"]) key_path = Path(self.config["private_key_path"]) @@ -165,11 +163,7 @@ def _get_private_key(self) -> PrivateKeyTypes: error_message = f"Private key file not found: {key_path}" raise FileNotFoundError(error_message) - return serialization.load_pem_private_key( - key_path.read_bytes(), - password=encoded_passphrase, - backend=default_backend(), - ) + return key_path.read_bytes() private_key: str = self.config["private_key"] self.logger.debug("Reading private key from config") @@ -180,12 +174,7 @@ def _get_private_key(self) -> PrivateKeyTypes: stacklevel=2, ) self.logger.info("Private key is in PEM format") - key_content = private_key.encode() - return serialization.load_pem_private_key( - key_content, - password=encoded_passphrase, - backend=default_backend(), - ) + return private_key.encode() try: self.logger.debug("Private key is in base64 format") @@ -194,7 +183,13 @@ def _get_private_key(self) -> PrivateKeyTypes: error_message = f"Invalid private key format: {e}" raise ValueError(error_message) from e - self.logger.debug("Attempting serialization of private key as DER") + return key_content + + def _load_private_key(self) -> PrivateKeyTypes: + phrase = self.config.get("private_key_passphrase") + encoded_passphrase = phrase.encode() if phrase else None + key_content = self._get_private_key_content() + try: return serialization.load_der_private_key( key_content, @@ -210,7 +205,7 @@ def _get_private_key(self) -> PrivateKeyTypes: ) def get_private_key(self): - return self._get_private_key().private_bytes( + return self._load_private_key().private_bytes( encoding=serialization.Encoding.DER, format=serialization.PrivateFormat.PKCS8, encryption_algorithm=serialization.NoEncryption(),