Merge branch 'main' into SNOW-1763594-add-TupleCursor-and-type-DictCu…

…rsor

.github/CODEOWNERS

@@ -1 +1 @@
-@snowflakedb/snowpark-python-api
+@sfc-gh-mkeller

.github/workflows/jira_issue.yml

@@ -37,7 +37,7 @@ jobs:
           summary: '${{ github.event.issue.title }}'
           description: |
             ${{ github.event.issue.body }} \\ \\ _Created from GitHub Action_ for ${{ github.event.issue.html_url }}
-          fields: '{"customfield_11401":{"id":"14586"},"assignee":{"id":"61027a237ab143006ecfb9a2"},"components":[{"id":"16413"}]}'
+          fields: '{"customfield_11401":{"id":"14723"},"assignee":{"id":"712020:e1f41916-da57-4fe8-b317-116d5229aa51"},"components":[{"id":"16413"}], "labels": ["oss"], "priority": {"id": "10001"} }'
 
       - name: Update GitHub Issue
         uses: ./jira/gajira-issue-update

DESCRIPTION.md

@@ -8,6 +8,20 @@ Source code is also available at: https://github.com/snowflakedb/snowflake-conne
 
 # Release Notes
 
+- v3.12.5(TBD)
+  - Added a feature to limit the sizes of IO-bound ThreadPoolExecutors during PUT and GET commands.
+  - Adding support for the new PAT authentication method.
+  - Updated README.md to include instructions on how to verify package signatures using `cosign`.
+  - Updated the log level for cursor's chunk rowcount from INFO to DEBUG.
+  - Added a feature to verify if the connection is still good enough to send queries over.
+  - Added support for base64-encoded DER private key strings in the `private_key` authentication type.
+
+- v3.12.4(December 3,2024)
+  - Fixed a bug where multipart uploads to Azure would be missing their MD5 hashes.
+  - Fixed a bug where OpenTelemetry header injection would sometimes cause Exceptions to be thrown.
+  - Fixed a bug where OCSP checks would throw TypeError and make mainly GCP blob storage unreachable.
+  - Bumped pyOpenSSL dependency from >=16.2.0,<25.0.0 to >=22.0.0,<25.0.0.
+
 - v3.12.3(October 25,2024)
   - Improved the error message for SSL-related issues to provide clearer guidance when an SSL error occurs.
   - Improved error message for SQL execution cancellations caused by timeout.

README.md

@@ -81,3 +81,28 @@ conn = snowflake.connector.connect(
 )
 conn.telemetry_enabled = False
 ```
+
+## Verifying Package Signatures
+
+To ensure the authenticity and integrity of the Python package, follow the steps below to verify the package signature using `cosign`.
+
+**Steps to verify the signature:**
+- Install cosign:
+  - This example is using golang installation: [installing-cosign-with-go](https://edu.chainguard.dev/open-source/sigstore/cosign/how-to-install-cosign/#installing-cosign-with-go)
+- Download the file from the repository like pypi:
+  - https://pypi.org/project/snowflake-connector-python/#files
+- Download the signature files from the release tag, replace the version number with the version you are verifying:
+  - https://github.com/snowflakedb/snowflake-connector-python/releases/tag/v3.12.2
+- Verify signature:
+  ````bash
+  # replace the version number with the version you are verifying
+  ./cosign verify-blob snowflake_connector_python-3.12.2.tar.gz \
+  --key snowflake-connector-python-v3.12.2.pub \
+  --signature resources.linux.snowflake_connector_python-3.12.2.tar.gz.sig
+
+  Verified OK
+  ````
+
+## NOTE
+
+This library currently does not support GCP regional endpoints.  Please ensure that any workloads using through this library do not require support for regional endpoints on GCP.  If you have questions about this, please contact [Snowflake Support](https://community.snowflake.com/s/article/How-To-Submit-a-Support-Case-in-Snowflake-Lodge).

setup.cfg

@@ -46,7 +46,7 @@ install_requires =
     asn1crypto>0.24.0,<2.0.0
     cffi>=1.9,<2.0.0
     cryptography>=3.1.0
-    pyOpenSSL>=16.2.0,<25.0.0
+    pyOpenSSL>=22.0.0,<25.0.0
     pyjwt<3.0.0
     pytz
     requests<3.0.0

src/snowflake/connector/auth/__init__.py

@@ -11,6 +11,7 @@
 from .keypair import AuthByKeyPair
 from .oauth import AuthByOAuth
 from .okta import AuthByOkta
+from .pat import AuthByPAT
 from .usrpwdmfa import AuthByUsrPwdMfa
 from .webbrowser import AuthByWebBrowser
 
@@ -23,13 +24,15 @@
         AuthByUsrPwdMfa,
         AuthByWebBrowser,
         AuthByIdToken,
+        AuthByPAT,
     )
 )
 
 __all__ = [
     "AuthByPlugin",
     "AuthByDefault",
     "AuthByKeyPair",
+    "AuthByPAT",
     "AuthByOAuth",
     "AuthByOkta",
     "AuthByUsrPwdMfa",

src/snowflake/connector/auth/by_plugin.py

@@ -54,6 +54,7 @@ class AuthType(Enum):
     ID_TOKEN = "ID_TOKEN"
     USR_PWD_MFA = "USERNAME_PASSWORD_MFA"
     OKTA = "OKTA"
+    PAT = "PROGRAMMATIC_ACCESS_TOKEN'"
 
 
 class AuthByPlugin(ABC):

src/snowflake/connector/auth/keypair.py

@@ -43,7 +43,7 @@ class AuthByKeyPair(AuthByPlugin):
 
     def __init__(
         self,
-        private_key: bytes | RSAPrivateKey,
+        private_key: bytes | str | RSAPrivateKey,
         lifetime_in_seconds: int = LIFETIME,
         **kwargs,
     ) -> None:
@@ -75,7 +75,7 @@ def __init__(
             ).total_seconds()
         )
 
-        self._private_key: bytes | RSAPrivateKey | None = private_key
+        self._private_key: bytes | str | RSAPrivateKey | None = private_key
         self._jwt_token = ""
         self._jwt_token_exp = 0
         self._lifetime = timedelta(
@@ -105,6 +105,17 @@ def prepare(
 
         now = datetime.now(timezone.utc).replace(tzinfo=None)
 
+        if isinstance(self._private_key, str):
+            try:
+                self._private_key = base64.b64decode(self._private_key)
+            except Exception as e:
+                raise ProgrammingError(
+                    msg=f"Failed to decode private key: {e}\nPlease provide a valid "
+                    "unencrypted rsa private key in base64-encoded DER format as a "
+                    "str object",
+                    errno=ER_INVALID_PRIVATE_KEY,
+                )
+
         if isinstance(self._private_key, bytes):
             try:
                 private_key = load_der_private_key(

src/snowflake/connector/auth/oauth.py

@@ -22,7 +22,7 @@ def type_(self) -> AuthType:
         return AuthType.OAUTH
 
     @property
-    def assertion_content(self) -> str:
+    def assertion_content(self) -> str | None:
         """Returns the token."""
         return self._oauth_token
 

src/snowflake/connector/auth/pat.py

@@ -0,0 +1,43 @@
+#
+# Copyright (c) 2012-2023 Snowflake Computing Inc. All rights reserved.
+#
+
+from __future__ import annotations
+
+import typing
+
+from snowflake.connector.network import PROGRAMMATIC_ACCESS_TOKEN
+
+from .by_plugin import AuthByPlugin, AuthType
+
+
+class AuthByPAT(AuthByPlugin):
+
+    def __init__(self, pat_token: str, **kwargs) -> None:
+        super().__init__(**kwargs)
+        self._pat_token: str | None = pat_token
+
+    def type_(self) -> AuthType:
+        return AuthType.PAT
+
+    def reset_secrets(self) -> None:
+        self._pat_token = None
+
+    def update_body(self, body: dict[typing.Any, typing.Any]) -> None:
+        body["data"]["AUTHENTICATOR"] = PROGRAMMATIC_ACCESS_TOKEN
+        body["data"]["TOKEN"] = self._pat_token
+
+    def prepare(
+        self,
+        **kwargs: typing.Any,
+    ) -> None:
+        """Nothing to do here, token should be obtained outside the driver."""
+        pass
+
+    def reauthenticate(self, **kwargs: typing.Any) -> dict[str, bool]:
+        return {"success": False}
+
+    @property
+    def assertion_content(self) -> str | None:
+        """Returns the token."""
+        return self._pat_token

src/snowflake/connector/azure_storage_client.py

@@ -4,6 +4,7 @@
 
 from __future__ import annotations
 
+import base64
 import json
 import os
 import xml.etree.ElementTree as ET
@@ -17,6 +18,7 @@
 from .constants import FileHeader, ResultStatus
 from .encryption_util import EncryptionMetadata
 from .storage_client import SnowflakeStorageClient
+from .util_text import get_md5
 from .vendored import requests
 
 if TYPE_CHECKING:  # pragma: no cover
@@ -149,7 +151,7 @@ def get_file_header(self, filename: str) -> FileHeader | None:
                 )
             )
             return FileHeader(
-                digest=r.headers.get("x-ms-meta-sfcdigest"),
+                digest=r.headers.get(SFCDIGEST),
                 content_length=int(r.headers.get("Content-Length")),
                 encryption_metadata=encryption_metadata,
             )
@@ -236,7 +238,27 @@ def _complete_multipart_upload(self) -> None:
             part = ET.Element("Latest")
             part.text = block_id
             root.append(part)
-        headers = {"x-ms-blob-content-encoding": "utf-8"}
+        # SNOW-1778088: We need to calculate the MD5 sum of this file for Azure Blob storage
+        new_stream = not bool(self.meta.src_stream or self.meta.intermediate_stream)
+        fd = (
+            self.meta.src_stream
+            or self.meta.intermediate_stream
+            or open(self.meta.real_src_file_name, "rb")
+        )
+        try:
+            if not new_stream:
+                # Reset position in file
+                fd.seek(0)
+            file_content = fd.read()
+        finally:
+            if new_stream:
+                fd.close()
+        headers = {
+            "x-ms-blob-content-encoding": "utf-8",
+            "x-ms-blob-content-md5": base64.b64encode(get_md5(file_content)).decode(
+                "utf-8"
+            ),
+        }
         azure_metadata = self._prepare_file_metadata()
         headers.update(azure_metadata)
         retry_id = "COMPLETE"

src/snowflake/connector/connection.py

@@ -41,6 +41,7 @@
     AuthByKeyPair,
     AuthByOAuth,
     AuthByOkta,
+    AuthByPAT,
     AuthByPlugin,
     AuthByUsrPwdMfa,
     AuthByWebBrowser,
@@ -99,6 +100,7 @@
     EXTERNAL_BROWSER_AUTHENTICATOR,
     KEY_PAIR_AUTHENTICATOR,
     OAUTH_AUTHENTICATOR,
+    PROGRAMMATIC_ACCESS_TOKEN,
     REQUEST_ID,
     USR_PWD_MFA_AUTHENTICATOR,
     ReauthenticationRequest,
@@ -183,10 +185,14 @@ def _get_private_bytes_from_file(
     "backoff_policy": (DEFAULT_BACKOFF_POLICY, Callable),
     "passcode_in_password": (False, bool),  # Snowflake MFA
     "passcode": (None, (type(None), str)),  # Snowflake MFA
-    "private_key": (None, (type(None), bytes, RSAPrivateKey)),
+    "private_key": (None, (type(None), bytes, str, RSAPrivateKey)),
     "private_key_file": (None, (type(None), str)),
     "private_key_file_pwd": (None, (type(None), str, bytes)),
-    "token": (None, (type(None), str)),  # OAuth or JWT Token
+    "token": (None, (type(None), str)),  # OAuth/JWT/PAT Token
+    "token_file_path": (
+        None,
+        (type(None), str, bytes),
+    ),  # OAuth/JWT/PAT Token file path
     "authenticator": (DEFAULT_AUTHENTICATOR, (type(None), str)),
     "mfa_callback": (None, (type(None), Callable)),
     "password_callback": (None, (type(None), Callable)),
@@ -295,6 +301,10 @@ def _get_private_bytes_from_file(
         False,
         bool,
     ),  # disable saml url check in okta authentication
+    "iobound_tpe_limit": (
+        None,
+        (type(None), int),
+    ),  # SNOW-1817982: limit iobound TPE sizes when executing PUT/GET
 }
 
 APPLICATION_RE = re.compile(r"[\w\d_]+")
@@ -727,6 +737,10 @@ def auth_class(self, value: AuthByPlugin) -> None:
     def is_query_context_cache_disabled(self) -> bool:
         return self._disable_query_context_cache
 
+    @property
+    def iobound_tpe_limit(self) -> int | None:
+        return self._iobound_tpe_limit
+
     def connect(self, **kwargs) -> None:
         """Establishes connection to Snowflake."""
         logger.debug("connect")
@@ -1090,6 +1104,8 @@ def __open_connection(self):
                     timeout=self.login_timeout,
                     backoff_generator=self._backoff_generator,
                 )
+            elif self._authenticator == PROGRAMMATIC_ACCESS_TOKEN:
+                self.auth_class = AuthByPAT(self._token)
             else:
                 # okta URL, e.g., https://<account>.okta.com/
                 self.auth_class = AuthByOkta(
@@ -1238,11 +1254,12 @@ def __config(self, **kwargs):
             if (
                 self.auth_class is None
                 and self._authenticator
-                not in [
+                not in (
                     EXTERNAL_BROWSER_AUTHENTICATOR,
                     OAUTH_AUTHENTICATOR,
                     KEY_PAIR_AUTHENTICATOR,
-                ]
+                    PROGRAMMATIC_ACCESS_TOKEN,
+                )
                 and not self._password
             ):
                 Error.errorhandler_wrapper(
@@ -1661,7 +1678,7 @@ def _log_telemetry(self, telemetry_data) -> None:
             self._telemetry.try_add_log_to_batch(telemetry_data)
 
     def _add_heartbeat(self) -> None:
-        """Add an hourly heartbeat query in order to keep connection alive."""
+        """Add a periodic heartbeat query in order to keep connection alive."""
         if not self.heartbeat_thread:
             self._validate_client_session_keep_alive_heartbeat_frequency()
             heartbeat_wref = weakref.WeakMethod(self._heartbeat_tick)
@@ -1687,7 +1704,7 @@ def _cancel_heartbeat(self) -> None:
             logger.debug("stopped heartbeat")
 
     def _heartbeat_tick(self) -> None:
-        """Execute a hearbeat if connection isn't closed yet."""
+        """Execute a heartbeat if connection isn't closed yet."""
         if not self.is_closed():
             logger.debug("heartbeating!")
             self.rest._heartbeat()
@@ -1974,3 +1991,21 @@ def _log_telemetry_imported_packages(self) -> None:
                     connection=self,
                 )
             )
+
+    def is_valid(self) -> bool:
+        """This function tries to answer the question: Is this connection still good for sending queries?
+        Attempts to validate the connections both on the TCP/IP and Session levels."""
+        logger.debug("validating connection and session")
+        if self.is_closed():
+            logger.debug("connection is already closed and not valid")
+            return False
+
+        try:
+            logger.debug("trying to heartbeat into the session to validate")
+            hb_result = self.rest._heartbeat()
+            session_valid = hb_result.get("success")
+            logger.debug("session still valid? %s", session_valid)
+            return bool(session_valid)
+        except Exception as e:
+            logger.debug("session could not be validated due to exception: %s", e)
+            return False