Add signature verification

Author: sfc-gh-pczajka

src/snowflake/connector/crl.py

@@ -12,7 +12,7 @@
 from cryptography import x509
 from cryptography.hazmat._oid import ExtensionOID
 from cryptography.hazmat.backends import default_backend
-from cryptography.hazmat.primitives import serialization
+from cryptography.hazmat.primitives import hashes, serialization
 from cryptography.hazmat.primitives.asymmetric import ec, padding, rsa
 from OpenSSL.SSL import Connection as SSLConnection
 
@@ -335,18 +335,35 @@ def traverse_chain(cert: x509.Certificate) -> CRLValidationResult | None:
                 # found a trusted certificate
                 logger.debug("Found trusted certificate: %s", cert.subject)
                 return CRLValidationResult.UNREVOKED
+
             if cert.issuer in self._trusted_ca:
                 # issuer is trusted by OS
                 logger.debug("Found certificate with trusted issuer: %s", cert.issuer)
-                return self._validate_certificate_with_cache(
+                if self._verify_certificate_signature(
                     cert, self._trusted_ca[cert.issuer]
-                )
+                ):
+                    return self._validate_certificate_with_cache(
+                        cert, self._trusted_ca[cert.issuer]
+                    )
+                else:
+                    logger.debug(
+                        "Certificate signature verification failed: for %s, looking for other paths",
+                        cert,
+                    )
+
             if cert.issuer in is_being_visited:
                 # cycle detected - invalid path
                 return None
 
             valid_results: list[tuple[CRLValidationResult, x509.Certificate]] = []
             for ca_cert in subject_certificates[cert.issuer]:
+                if not self._verify_certificate_signature(cert, ca_cert):
+                    logger.debug(
+                        "Certificate signature verification failed for %s, looking for other paths",
+                        cert,
+                    )
+                    continue
+
                 is_being_visited.add(cert.issuer)
                 ca_result = traverse_chain(ca_cert)
                 is_being_visited.remove(cert.issuer)
@@ -567,52 +584,94 @@ def _check_certificate_against_crl_url(
         # Check if certificate is revoked
         return self._check_certificate_against_crl(cert, crl)
 
+    def _verify_certificate_signature(
+        self, cert: x509.Certificate, ca_cert: x509.Certificate
+    ) -> bool:
+        """Verify certificate signature with CA's public key"""
+        logger.debug(
+            "Verifying certificate signature of %s against %s public key", cert, ca_cert
+        )
+        return self._verify_signature(
+            public_key=ca_cert.public_key(),
+            signature=cert.signature,
+            data=cert.tbs_certificate_bytes,
+            hash_algorithm=cert.signature_hash_algorithm,
+            signature_type="certificate signature",
+        )
+
     def _verify_crl_signature(
         self, crl: x509.CertificateRevocationList, ca_cert: x509.Certificate
     ) -> bool:
         """Verify CRL signature with CA's public key"""
-        try:
-            # Get the signature algorithm from the CRL
-            signature_algorithm = crl.signature_algorithm_oid
-            hash_algorithm = crl.signature_hash_algorithm
+        # Get the signature algorithm from the CRL
+        signature_algorithm = crl.signature_algorithm_oid
+        hash_algorithm = crl.signature_hash_algorithm
+
+        logger.debug(
+            "Verifying CRL signature with algorithm: %s, hash: %s",
+            signature_algorithm,
+            hash_algorithm,
+        )
 
-            logger.debug(
-                "Verifying CRL signature with algorithm: %s, hash: %s",
-                signature_algorithm,
-                hash_algorithm,
-            )
+        result = self._verify_signature(
+            public_key=ca_cert.public_key(),
+            signature=crl.signature,
+            data=crl.tbs_certlist_bytes,
+            hash_algorithm=hash_algorithm,
+            signature_type="CRL signature",
+        )
 
-            # Determine the appropriate padding based on the signature algorithm
-            public_key = ca_cert.public_key()
+        if result:
+            logger.debug("CRL signature verification successful")
+        return result
 
+    def _verify_signature(
+        self,
+        public_key: rsa.RSAPublicKey | ec.EllipticCurvePublicKey | Any,
+        signature: bytes,
+        data: bytes,
+        hash_algorithm: hashes.HashAlgorithm,
+        signature_type: str = "signature",
+    ) -> bool:
+        """Verify a signature using a public key
+
+        Args:
+            public_key: The public key to use for verification
+            signature: The signature to verify
+            data: The data that was signed
+            hash_algorithm: The hash algorithm used in the signature
+            signature_type: Type of signature being verified (for logging)
+
+        Returns:
+            bool: True if verification succeeds, False otherwise
+        """
+        try:
             # Handle different key types with appropriate signature verification
             if isinstance(public_key, rsa.RSAPublicKey):
                 # For RSA signatures, we need to use PKCS1v15 padding
                 public_key.verify(
-                    crl.signature,
-                    crl.tbs_certlist_bytes,
+                    signature,
+                    data,
                     padding.PKCS1v15(),
                     hash_algorithm,
                 )
             elif isinstance(public_key, ec.EllipticCurvePublicKey):
                 # For EC signatures, use ECDSA algorithm
                 public_key.verify(
-                    crl.signature,
-                    crl.tbs_certlist_bytes,
+                    signature,
+                    data,
                     ec.ECDSA(hash_algorithm),
                 )
             else:
                 # For other key types (DSA, etc.), try without padding
                 public_key.verify(
-                    crl.signature,
-                    crl.tbs_certlist_bytes,
+                    signature,
+                    data,
                     hash_algorithm,
                 )
-
-            logger.debug("CRL signature verification successful")
             return True
         except Exception as e:
-            logger.warning("CRL signature verification failed: %s", e)
+            logger.warning("%s verification failed: %s", signature_type.capitalize(), e)
             return False
 
     def _check_certificate_against_crl(

test/unit/test_crl.py

@@ -1930,6 +1930,146 @@ def test_is_short_lived_certificate(cert_gen, issue_date, validity_days, expecte
     assert CRLValidator._is_short_lived_certificate(cert) == expected
 
 
+def test_validate_certificate_signatures(cert_gen, session_manager):
+    """Test that certificate validation fails with ERROR when signed by wrong key"""
+    # Create a certificate signed by the test CA
+    valid_cert = cert_gen.create_certificate_with_crl_distribution_points(
+        "CN=Test Server", []
+    )
+
+    # Create a different CA key pair
+    different_ca_key = rsa.generate_private_key(
+        public_exponent=65537, key_size=2048, backend=default_backend()
+    )
+    different_cert = (
+        x509.CertificateBuilder()
+        .subject_name(valid_cert.subject)
+        .issuer_name(cert_gen.ca_certificate.subject)
+        .public_key(cert_gen.ca_private_key.public_key())
+        .serial_number(x509.random_serial_number())
+        .not_valid_before(datetime.now(timezone.utc))
+        .not_valid_after(datetime.now(timezone.utc) + timedelta(days=365))
+        .add_extension(
+            x509.BasicConstraints(ca=True, path_length=None),
+            critical=True,
+        )
+        .sign(different_ca_key, hashes.SHA256(), backend=default_backend())
+    )
+    short_lived_different_cert = (
+        x509.CertificateBuilder()
+        .subject_name(valid_cert.subject)
+        .issuer_name(cert_gen.ca_certificate.subject)
+        .public_key(different_ca_key.public_key())
+        .serial_number(x509.random_serial_number())
+        .not_valid_before(datetime.now(timezone.utc))
+        .not_valid_after(datetime.now(timezone.utc) + timedelta(days=3))
+        .add_extension(
+            x509.BasicConstraints(ca=True, path_length=None),
+            critical=True,
+        )
+        .sign(different_ca_key, hashes.SHA256(), backend=default_backend())
+    )
+
+    validator = CRLValidator(
+        session_manager,
+        cert_revocation_check_mode=CertRevocationCheckMode.ENABLED,
+        allow_certificates_without_crl_url=True,
+        trusted_certificates=[cert_gen.ca_certificate],
+    )
+
+    # wrong signature - no path found = ERROR
+    assert (
+        validator._validate_single_chain([different_cert]) == CRLValidationResult.ERROR
+    )
+    # wrong signature - short-lived - no path found = ERROR
+    assert (
+        validator._validate_single_chain([short_lived_different_cert])
+        == CRLValidationResult.ERROR
+    )
+    # wrong signature does not stop from searching of new path
+    assert (
+        validator._validate_single_chain(
+            [different_cert, short_lived_different_cert, valid_cert]
+        )
+        == CRLValidationResult.UNREVOKED
+    )
+
+
+def test_validate_certificate_signatures_in_chain(cert_gen, session_manager):
+    """Test that certificate validation fails with ERROR when signed by wrong key"""
+    # Create a certificate chain signed by the test CA: leaf -> A -> B -> CA
+    # mingle with A -> B
+    chain = cert_gen.create_cross_signed_chain()
+
+    valid_cert = chain.BsignA
+
+    # Create a different CA key pair
+    different_key = rsa.generate_private_key(
+        public_exponent=65537, key_size=2048, backend=default_backend()
+    )
+    different_cert = (
+        x509.CertificateBuilder()
+        .subject_name(valid_cert.subject)
+        .issuer_name(cert_gen.ca_certificate.subject)
+        .public_key(cert_gen.ca_private_key.public_key())
+        .serial_number(x509.random_serial_number())
+        .not_valid_before(datetime.now(timezone.utc))
+        .not_valid_after(datetime.now(timezone.utc) + timedelta(days=365))
+        .add_extension(
+            x509.BasicConstraints(ca=True, path_length=None),
+            critical=True,
+        )
+        .sign(different_key, hashes.SHA256(), backend=default_backend())
+    )
+    short_lived_different_cert = (
+        x509.CertificateBuilder()
+        .subject_name(valid_cert.subject)
+        .issuer_name(cert_gen.ca_certificate.subject)
+        .public_key(different_key.public_key())
+        .serial_number(x509.random_serial_number())
+        .not_valid_before(datetime.now(timezone.utc))
+        .not_valid_after(datetime.now(timezone.utc) + timedelta(days=3))
+        .add_extension(
+            x509.BasicConstraints(ca=True, path_length=None),
+            critical=True,
+        )
+        .sign(different_key, hashes.SHA256(), backend=default_backend())
+    )
+
+    validator = CRLValidator(
+        session_manager,
+        allow_certificates_without_crl_url=True,
+        cert_revocation_check_mode=CertRevocationCheckMode.ENABLED,
+        trusted_certificates=[cert_gen.ca_certificate],
+    )
+
+    # wrong signature - no path found = ERROR
+    assert (
+        validator._validate_single_chain([chain.leafA, different_cert, chain.rootB])
+        == CRLValidationResult.ERROR
+    )
+    # wrong signature - short-lived - no path found = ERROR
+    assert (
+        validator._validate_single_chain(
+            [chain.leafA, short_lived_different_cert, chain.rootB]
+        )
+        == CRLValidationResult.ERROR
+    )
+    # wrong signature does not stop from searching of new path
+    assert (
+        validator._validate_single_chain(
+            [
+                chain.leafA,
+                different_cert,
+                short_lived_different_cert,
+                valid_cert,
+                chain.rootB,
+            ]
+        )
+        == CRLValidationResult.UNREVOKED
+    )
+
+
 def test_is_certificate_trusted_by_os(cert_gen):
     """Test OS certificate trust validation."""
     # Create a test certificate chain