SNOW-1820480 making OCSP validation code more resillient (#2107)

Author: sfc-gh-mkeller

DESCRIPTION.md

@@ -8,6 +8,11 @@ Source code is also available at: https://github.com/snowflakedb/snowflake-conne
 
 # Release Notes
 
+- v3.12.4(TBD)
+  - 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.
+
 - 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.

src/snowflake/connector/ocsp_asn1crypto.py

@@ -5,6 +5,7 @@
 
 from __future__ import annotations
 
+import typing
 from base64 import b64decode, b64encode
 from collections import OrderedDict
 from datetime import datetime, timezone
@@ -28,6 +29,9 @@
 from cryptography.hazmat.backends import default_backend
 from cryptography.hazmat.primitives import hashes, serialization
 from cryptography.hazmat.primitives.asymmetric import padding, utils
+from cryptography.hazmat.primitives.asymmetric.dsa import DSAPublicKey
+from cryptography.hazmat.primitives.asymmetric.ec import ECDSA, EllipticCurvePublicKey
+from cryptography.hazmat.primitives.asymmetric.rsa import RSAPublicKey
 from OpenSSL.SSL import Connection
 
 from snowflake.connector.errorcode import (
@@ -368,9 +372,21 @@ def verify_signature(self, signature_algorithm, signature, cert, data):
         hasher = hashes.Hash(chosen_hash, backend)
         hasher.update(data.dump())
         digest = hasher.finalize()
+        additional_kwargs: dict[str, typing.Any] = dict()
+        if isinstance(public_key, RSAPublicKey):
+            additional_kwargs["padding"] = padding.PKCS1v15()
+            additional_kwargs["algorithm"] = utils.Prehashed(chosen_hash)
+        elif isinstance(public_key, DSAPublicKey):
+            additional_kwargs["algorithm"] = utils.Prehashed(chosen_hash)
+        elif isinstance(public_key, EllipticCurvePublicKey):
+            additional_kwargs["signature_algorithm"] = ECDSA(
+                utils.Prehashed(chosen_hash)
+            )
         try:
             public_key.verify(
-                signature, digest, padding.PKCS1v15(), utils.Prehashed(chosen_hash)
+                signature,
+                digest,
+                **additional_kwargs,
             )
         except InvalidSignature:
             raise RevocationCheckError(msg="Failed to verify the signature")

test/integ/conftest.py

@@ -177,6 +177,7 @@ def init_test_schema(db_parameters) -> Generator[None, None, None]:
         database=ret["database"],
         account=ret["account"],
         protocol=ret["protocol"],
+        role=ret["role"],
     ) as con:
         con.cursor().execute(f"CREATE SCHEMA IF NOT EXISTS {TEST_SCHEMA}")
         yield