checking if the downloaded CRL is more recent

Author: sfc-gh-pczajka

src/snowflake/connector/crl.py

@@ -616,6 +616,51 @@ def _fetch_crl_from_url(self, crl_url: str) -> bytes | None:
             logger.exception("Failed to download CRL from %s", crl_url)
             return None
 
+    def _get_crl_last_update(
+        self, crl: x509.CertificateRevocationList
+    ) -> datetime | None:
+        """
+        Get the last_update timestamp from a CRL.
+
+        Args:
+            crl: The CRL to extract the timestamp from
+
+        Returns:
+            The last_update timestamp, or None if not available
+        """
+        try:
+            return crl.last_update_utc
+        except AttributeError:
+            return getattr(crl, "last_update", None)
+
+    def _is_crl_more_recent(
+        self,
+        new_crl: x509.CertificateRevocationList,
+        cached_crl: x509.CertificateRevocationList,
+    ) -> bool:
+        """
+        Check if a newly downloaded CRL is more recent than a cached CRL.
+
+        Args:
+            new_crl: The newly downloaded CRL
+            cached_crl: The cached CRL
+
+        Returns:
+            True if new_crl is more recent (has a later last_update), False otherwise
+        """
+        new_last_update = self._get_crl_last_update(new_crl)
+        cached_last_update = self._get_crl_last_update(cached_crl)
+
+        if new_last_update is None:
+            logger.warning("New CRL has no last_update timestamp")
+            return False
+
+        if cached_last_update is None:
+            logger.warning("Cached CRL has no last_update timestamp")
+            return True
+
+        return new_last_update > cached_last_update
+
     def _download_crl(
         self, crl_url: str
     ) -> tuple[x509.CertificateRevocationList | None, datetime | None]:
@@ -652,9 +697,27 @@ def _check_certificate_against_crl_url(
             or cached_crl.is_crl_expired_by(now)
             or cached_crl.is_evicted_by(now, self._cache_validity_time)
         ):
+            logger.debug("Cached CRL is None/expired/evicted, downloading new CRL")
             crl, ts = self._download_crl(crl_url)
-            if crl and ts:
-                self._put_crl_to_cache(crl_url, crl, ts)
+            if crl is not None and ts is not None:
+                # Only cache the downloaded CRL if it's more recent than the cached one
+                is_more_recent = cached_crl is None or self._is_crl_more_recent(
+                    crl, cached_crl.crl
+                )
+                logger.debug(
+                    "Is downloaded CRL more recent? cached_crl is None=%s, is_more_recent=%s",
+                    cached_crl is None,
+                    is_more_recent,
+                )
+                if is_more_recent:
+                    self._put_crl_to_cache(crl_url, crl, ts)
+                    logger.debug("Cached newly downloaded CRL for %s", crl_url)
+                else:
+                    logger.info(
+                        "Downloaded CRL for %s is not more recent than cached version, keeping cached CRL",
+                        crl_url,
+                    )
+                    crl = cached_crl.crl
         else:
             crl = cached_crl.crl
 

test/unit/test_crl.py

@@ -24,7 +24,12 @@
     CRLValidationResult,
     CRLValidator,
 )
-from snowflake.connector.crl_cache import CRLCacheEntry, CRLCacheManager
+from snowflake.connector.crl_cache import (
+    CRLCacheEntry,
+    CRLCacheManager,
+    CRLInMemoryCache,
+    NoopCRLCache,
+)
 from snowflake.connector.session_manager import SessionManager
 
 
@@ -166,6 +171,39 @@ def generate_valid_crl(self) -> bytes:
             )
             return crl.public_bytes(serialization.Encoding.DER)
 
+        def create_crl_with_timestamp(
+            self, last_update: datetime
+        ) -> x509.CertificateRevocationList:
+            """
+            Create a CRL with a specific last_update timestamp.
+
+            Args:
+                last_update: The last_update timestamp for the CRL
+
+            Returns:
+                A CRL object with the specified timestamp
+            """
+            builder = x509.CertificateRevocationListBuilder()
+            builder = builder.issuer_name(self.ca_certificate.subject)
+            builder = builder.last_update(last_update)
+            builder = builder.next_update(
+                datetime.now(timezone.utc) + timedelta(days=1)
+            )
+
+            # Add any revoked certificates
+            for serial_number in self.revoked_serial_numbers:
+                revoked_cert = (
+                    x509.RevokedCertificateBuilder()
+                    .serial_number(serial_number)
+                    .revocation_date(datetime.now(timezone.utc))
+                    .build()
+                )
+                builder = builder.add_revoked_certificate(revoked_cert)
+
+            return builder.sign(
+                self.ca_private_key, hashes.SHA256(), backend=default_backend()
+            )
+
         def generate_expired_crl(self) -> bytes:
             """Generate an expired CRL"""
             builder = x509.CertificateRevocationListBuilder()
@@ -1585,6 +1623,7 @@ def test_crl_validator_check_certificate_against_crl_expired(
     # Mock expired CRL
     mock_crl = Mock(spec=x509.CertificateRevocationList)
     mock_crl.next_update_utc = datetime.now(timezone.utc) - timedelta(days=1)  # Expired
+    mock_crl.last_update_utc = datetime.now(timezone.utc) - timedelta(days=2)
     mock_crl.get_revoked_certificate_by_serial_number.return_value = None
     mock_crl.issuer = parent.subject
     # Mock extensions to raise ExtensionNotFound for IDP extension
@@ -2476,3 +2515,122 @@ def test_check_certificate_against_crl_url_with_idp_mismatch(
     )
 
     assert result == CRLValidationResult.ERROR
+
+
+@pytest.mark.parametrize("downloaded_crl_is_newer", [True, False])
+def test_crl_validator_freshness_validation(
+    cert_gen, session_manager, downloaded_crl_is_newer
+):
+    """Test that validator uses the most recent CRL based on last_update timestamp"""
+    chain = cert_gen.create_simple_chain()
+
+    # Create CRLs with different timestamps
+    older_last_update = datetime.now(timezone.utc) - timedelta(days=2)
+    newer_last_update = datetime.now(timezone.utc)
+    older_crl = cert_gen.create_crl_with_timestamp(older_last_update)
+    newer_crl = cert_gen.create_crl_with_timestamp(newer_last_update)
+
+    # Determine which CRL to cache and which to download
+    cached_crl = older_crl if downloaded_crl_is_newer else newer_crl
+    downloaded_crl = newer_crl if downloaded_crl_is_newer else older_crl
+
+    # Create cache manager and pre-populate with the cached CRL
+    memory_cache = CRLInMemoryCache(cache_validity_time=timedelta(hours=24))
+    memory_cache.put(
+        "http://test.com/crl",
+        CRLCacheEntry(
+            crl=cached_crl, download_time=older_last_update
+        ),  # use old date to enforce "download" logic
+    )
+    cache_manager = CRLCacheManager(memory_cache, NoopCRLCache())
+
+    validator = CRLValidator(
+        session_manager=session_manager,
+        trusted_certificates=[cert_gen.ca_certificate],
+        cache_manager=cache_manager,
+    )
+
+    # Mock _download_crl to return the downloaded CRL
+    download_timestamp = datetime.now(timezone.utc)
+    with mock_patch.object(
+        validator, "_download_crl", return_value=(downloaded_crl, download_timestamp)
+    ) as mock_download, mock_patch.object(
+        validator, "_verify_crl_signature", return_value=True
+    ):
+        validator._check_certificate_against_crl_url(
+            chain.leaf_cert, cert_gen.ca_certificate, "http://test.com/crl"
+        )
+        # self-check for debug - ensure download was called
+        mock_download.assert_called_once_with("http://test.com/crl")
+
+    # Verify the cached CRL is the newer one
+    cached_entry = memory_cache.get("http://test.com/crl")
+    assert cached_entry is not None
+    final_cached_last_update = validator._get_crl_last_update(cached_entry.crl)
+    # Compare with tolerance for microseconds (CRL might lose precision)
+    assert abs((final_cached_last_update - newer_last_update).total_seconds()) < 1
+
+
+def test_get_crl_last_update(cert_gen):
+    """Test helper method to extract last_update from CRL"""
+    # Create a CRL with a known last_update
+    builder = x509.CertificateRevocationListBuilder()
+    builder = builder.issuer_name(cert_gen.ca_certificate.subject)
+    expected_last_update = datetime.now(timezone.utc)
+    builder = builder.last_update(expected_last_update)
+    builder = builder.next_update(datetime.now(timezone.utc) + timedelta(days=1))
+    crl = builder.sign(
+        cert_gen.ca_private_key, hashes.SHA256(), backend=default_backend()
+    )
+
+    validator = CRLValidator(
+        session_manager=Mock(),
+        trusted_certificates=[cert_gen.ca_certificate],
+    )
+
+    # Extract last_update
+    last_update = validator._get_crl_last_update(crl)
+    assert last_update is not None
+    # Compare with some tolerance for microseconds
+    assert abs((last_update - expected_last_update).total_seconds()) < 1
+
+
+def test_is_crl_more_recent(cert_gen):
+    """Test comparison of CRL freshness by last_update timestamp"""
+    # Create two CRLs with different last_update times
+    older_last_update = datetime.now(timezone.utc) - timedelta(hours=2)
+    newer_last_update = datetime.now(timezone.utc)
+
+    # Build older CRL
+    older_builder = x509.CertificateRevocationListBuilder()
+    older_builder = older_builder.issuer_name(cert_gen.ca_certificate.subject)
+    older_builder = older_builder.last_update(older_last_update)
+    older_builder = older_builder.next_update(
+        datetime.now(timezone.utc) + timedelta(days=1)
+    )
+    older_crl = older_builder.sign(
+        cert_gen.ca_private_key, hashes.SHA256(), backend=default_backend()
+    )
+
+    # Build newer CRL
+    newer_builder = x509.CertificateRevocationListBuilder()
+    newer_builder = newer_builder.issuer_name(cert_gen.ca_certificate.subject)
+    newer_builder = newer_builder.last_update(newer_last_update)
+    newer_builder = newer_builder.next_update(
+        datetime.now(timezone.utc) + timedelta(days=1)
+    )
+    newer_crl = newer_builder.sign(
+        cert_gen.ca_private_key, hashes.SHA256(), backend=default_backend()
+    )
+
+    validator = CRLValidator(
+        session_manager=Mock(),
+        trusted_certificates=[cert_gen.ca_certificate],
+    )
+
+    # Test that newer is more recent than older
+    assert validator._is_crl_more_recent(newer_crl, older_crl) is True
+    # Test that older is not more recent than newer
+    assert validator._is_crl_more_recent(older_crl, newer_crl) is False
+    # Test that a CRL is not more recent than itself
+    assert validator._is_crl_more_recent(newer_crl, newer_crl) is False