do not always renew CRLs, similar to OCSP keys

Author: mathiasertl

ca/ca/settings.py

@@ -194,7 +194,7 @@
 CELERY_BEAT_SCHEDULE = {
     "generate-crls": {
         "task": "django_ca.tasks.generate_crls",
-        "schedule": 86100,
+        "schedule": 3600,
     },
     "generate-ocsp-keys": {
         # Attempt to regenerate OCSP responder certificates every hour. Certificates are only regenerated if

ca/django_ca/conf.py

@@ -39,6 +39,7 @@
     AnnotatedEllipticCurveName,
     AnnotatedSignatureHashAlgorithmName,
     CertificateRevocationListReasonCode,
+    PositiveTimedelta,
     PowerOfTwoInt,
     Serial,
     UniqueElementsTuple,
@@ -56,8 +57,7 @@
 #   https://github.com/pydantic/pydantic/issues/10459
 # TimedeltaAsDays = Annotated[timedelta, BeforeValidator(timedelta_as_number_parser("days"))]
 DayValidator = BeforeValidator(timedelta_as_number_parser("days"))
-PositiveTimedelta = Annotated[timedelta, Ge(timedelta(days=1))]
-AcmeCertValidity = Annotated[PositiveTimedelta, Le(timedelta(days=365)), DayValidator]
+AcmeCertValidity = Annotated[PositiveTimedelta, Ge(timedelta(days=1)), Le(timedelta(days=365)), DayValidator]
 
 _KT = TypeVar("_KT")
 _KV = TypeVar("_KV")
@@ -205,14 +205,16 @@ def validate_scope(self) -> Self:
 class CertificateRevocationListProfileOverride(CertificateRevocationListBaseModel):
     """Model for overriding fields of a CRL Profile."""
 
-    expires: timedelta | None = None
+    expires: Annotated[PositiveTimedelta, Ge(timedelta(seconds=600))] | None = None
+    renewal: Annotated[PositiveTimedelta, Ge(timedelta(seconds=300))] | None = None
     skip: bool = False
 
 
 class CertificateRevocationListProfile(CertificateRevocationListBaseModel):
     """Model for profiles for CRL generation."""
 
-    expires: timedelta = timedelta(days=1)
+    expires: Annotated[PositiveTimedelta, Ge(timedelta(seconds=600))] = timedelta(days=1)
+    renewal: Annotated[PositiveTimedelta, Ge(timedelta(seconds=300))] = timedelta(hours=12)
     OVERRIDES: dict[Serial, CertificateRevocationListProfileOverride] = Field(default_factory=dict)
 
     @model_validator(mode="after")
@@ -279,7 +281,7 @@ class SettingsModel(BaseModel):
     CA_DEFAULT_ELLIPTIC_CURVE: AnnotatedEllipticCurveName = Field(
         default="secp256r1", description="The default elliptic curve for EC based CAs."
     )
-    CA_DEFAULT_EXPIRES: Annotated[PositiveTimedelta, DayValidator] = Field(
+    CA_DEFAULT_EXPIRES: Annotated[PositiveTimedelta, Ge(timedelta(days=1)), DayValidator] = Field(
         default=timedelta(days=100),
         description="The default validity time for a new certificate.",
         examples=[timedelta(days=45), 45],

ca/django_ca/managers.py

@@ -736,6 +736,8 @@ def scope(
             only_some_reasons: frozenset[x509.ReasonFlags] | None = None,
         ) -> "CertificateRevocationListQuerySet": ...
 
+        def valid(self, now: datetime | None = None) -> "CertificateRevocationListQuerySet": ...
+
     def _add_issuing_distribution_point_extension(
         self,
         builder: x509.CertificateRevocationListBuilder,

ca/django_ca/models.py

@@ -212,7 +212,7 @@ def __str__(self) -> str:
         return self.mail
 
     @classmethod
-    def from_addr(cls, addr: str) -> "Self":
+    def from_addr(cls, addr: str) -> Self:
         """Class constructor that creates an instance from an email address."""
         name = ""
         match = re.match(r"(.*?)\s*<(.*)>", addr)
@@ -648,7 +648,7 @@ def key_type(self) -> ParsableKeyType:
     def cache_crls(self, key_backend_options: BaseModel) -> None:  # pylint: disable=C0116
         self.generate_crls(key_backend_options)
 
-    def generate_crls(self, key_backend_options: BaseModel) -> None:
+    def generate_crls(self, key_backend_options: BaseModel, force: bool = False) -> None:
         """Generate certificate revocation lists (CRLs) for this certificate authority.
 
         .. versionchanged:: 3.0.0
@@ -659,9 +659,17 @@ def generate_crls(self, key_backend_options: BaseModel) -> None:
         .. versionchanged:: 1.25.0
 
             Support for passing a custom hash algorithm to this function was removed.
+
+        Parameters
+        ----------
+        key_backend_options : BaseModel
+            Options required for using the private key of the certificate authority.
+        force : bool, optional
+            Set to ``True`` to force regeneration of CRLs. By default, keys are only regenerated if they
+            expire within the renewal period of the respective :ref:settings-ca-crl-profiles`.
         """
-        for crl_profile in model_settings.CA_CRL_PROFILES.values():
-            now = datetime.now(tz=UTC)
+        for name, crl_profile in model_settings.CA_CRL_PROFILES.items():
+            now = timezone.now()
 
             # If there is an override for the current CA, create a new profile model with values updated from
             # the override.
@@ -673,6 +681,19 @@ def generate_crls(self, key_backend_options: BaseModel) -> None:
                 config.update(crl_profile_override.model_dump(exclude_unset=True))
                 crl_profile = CertificateRevocationListProfile.model_validate(config)
 
+            if force is False:
+                crl_qs = CertificateRevocationList.objects.scope(
+                    serial=self.serial,
+                    only_contains_ca_certs=crl_profile.only_contains_ca_certs,
+                    only_contains_user_certs=crl_profile.only_contains_user_certs,
+                    only_contains_attribute_certs=crl_profile.only_contains_attribute_certs,
+                    only_some_reasons=crl_profile.only_some_reasons,
+                )
+                current_crl = crl_qs.valid(now=now).newest()
+                if current_crl is not None and current_crl.next_update > now + crl_profile.renewal:
+                    log.info("%s: %s CRL profile is not yet scheduled for renewal.", self.serial, name)
+                    continue
+
             crl = CertificateRevocationList.objects.create_certificate_revocation_list(
                 ca=self,
                 key_backend_options=key_backend_options,

ca/django_ca/pydantic/type_aliases.py

@@ -17,7 +17,7 @@
 from datetime import timedelta
 from typing import Annotated, Any, TypeVar
 
-from annotated_types import Ge
+from annotated_types import Gt
 from pydantic import AfterValidator, AwareDatetime, BeforeValidator, Field, PlainSerializer
 
 from cryptography import x509
@@ -134,7 +134,7 @@
 """A datetime that is timezone-aware and in the future."""
 
 DayValidator = BeforeValidator(timedelta_as_number_parser("days"))
-PositiveTimedelta = Annotated[timedelta, Ge(timedelta(seconds=0))]
+PositiveTimedelta = Annotated[timedelta, Gt(timedelta(seconds=0))]
 
 UniqueTupleTypeVar = TypeVar("UniqueTupleTypeVar", bound=tuple[Hashable, ...])
 UniqueElementsTuple = Annotated[UniqueTupleTypeVar, AfterValidator(unique_validator)]

ca/django_ca/querysets.py

@@ -290,6 +290,11 @@ def scope(
             only_contains_attribute_certs=only_contains_attribute_certs,
         ).reasons(only_some_reasons)
 
+    def valid(self, now: datetime | None = None) -> "CertificateRevocationListQuerySet":
+        if now is None:  # pragma: no cover
+            now = timezone.now()
+        return self.exclude(next_update__lt=now).exclude(last_update__gt=now)
+
 
 class AcmeAccountQuerySet(AcmeAccountQuerySetBase):
     """QuerySet for :py:class:`~django_ca.models.AcmeAccount`."""

ca/django_ca/tests/base/assertions.py

@@ -31,6 +31,7 @@
 from cryptography.x509.oid import ExtensionOID
 from OpenSSL.crypto import FILETYPE_PEM, X509Store, X509StoreContext, load_certificate
 
+from django.core.cache import cache
 from django.core.exceptions import ImproperlyConfigured, ValidationError
 from django.core.management import CommandError
 
@@ -40,15 +41,17 @@
 from django_ca.constants import ReasonFlags
 from django_ca.deprecation import RemovedInDjangoCA310Warning, RemovedInDjangoCA320Warning
 from django_ca.key_backends.storages.models import StoragesUsePrivateKeyOptions
-from django_ca.models import Certificate, CertificateAuthority, X509CertMixin
+from django_ca.models import Certificate, CertificateAuthority, CertificateRevocationList, X509CertMixin
 from django_ca.signals import post_create_ca, post_issue_cert, post_sign_cert, pre_create_ca, pre_sign_cert
 from django_ca.tests.base.mocks import mock_signal
 from django_ca.tests.base.utils import (
     authority_information_access,
     basic_constraints,
     cmd_e2e,
+    crl_cache_key,
     crl_distribution_points,
     distribution_point,
+    get_idp,
     uri,
 )
 
@@ -278,6 +281,27 @@ def ext_sorter(ext: x509.Extension[x509.ExtensionType]) -> str:
             assert not list(entry.extensions)
 
 
+def assert_crls(ca: CertificateAuthority, number: int = 0) -> None:
+    """Test the CRLs for the given certificate authority."""
+    algorithm = ca.algorithm
+    for kwargs in [{"only_contains_ca_certs": True}, {"only_contains_user_certs": True}]:
+        der_key = crl_cache_key(ca.serial, **kwargs)
+        pem_key = crl_cache_key(ca.serial, Encoding.PEM, **kwargs)
+        idp = get_idp(full_name=None, **kwargs)
+
+        # Fetch and test CRLs from the cache
+        der_crl = cache.get(der_key)
+        pem_crl = cache.get(pem_key)
+        assert_crl(der_crl, crl_number=number, idp=idp, encoding=Encoding.DER, signer=ca, algorithm=algorithm)
+        assert_crl(pem_crl, crl_number=number, idp=idp, encoding=Encoding.PEM, signer=ca, algorithm=algorithm)
+
+        # Fetch from the database and verify
+        db_crl = CertificateRevocationList.objects.scope(serial=ca.serial, **kwargs).newest()
+        assert db_crl is not None
+        assert db_crl.data == der_crl
+        assert db_crl.number == number
+
+
 def assert_e2e_error(
     cmd: typing.Sequence[str],
     stdout: str | re.Pattern[str] = "",

ca/django_ca/tests/models/test_certificate_authority.py

@@ -40,6 +40,7 @@
 
 import pytest
 from freezegun import freeze_time
+from freezegun.api import FrozenDateTimeFactory
 from pytest_django.fixtures import SettingsWrapper
 
 from django_ca.conf import model_settings
@@ -50,6 +51,7 @@
 from django_ca.tests.base.assertions import (
     assert_certificate,
     assert_crl,
+    assert_crls,
     assert_removed_in_310,
     assert_sign_cert_signals,
 )
@@ -124,112 +126,48 @@ def test_root(root: CertificateAuthority, child: CertificateAuthority) -> None:
 @pytest.mark.freeze_time(TIMESTAMPS["everything_valid"])
 def test_generate_crls(settings: SettingsWrapper, usable_ca: CertificateAuthority) -> None:
     """Test generation of CRLs."""
-    ca_private_key_options = StoragesUsePrivateKeyOptions(password=CERT_DATA[usable_ca.name].get("password"))
-    der_user_key = crl_cache_key(usable_ca.serial, only_contains_user_certs=True)
-    pem_user_key = crl_cache_key(usable_ca.serial, Encoding.PEM, only_contains_user_certs=True)
-    der_ca_key = crl_cache_key(usable_ca.serial, only_contains_ca_certs=True)
-    pem_ca_key = crl_cache_key(usable_ca.serial, Encoding.PEM, only_contains_ca_certs=True)
-    user_idp = get_idp(full_name=None, only_contains_user_certs=True)
-    ca_idp = get_idp(full_name=None, only_contains_ca_certs=True)
+    key_backend_options = StoragesUsePrivateKeyOptions(password=CERT_DATA[usable_ca.name].get("password"))
 
-    assert cache.get(der_ca_key) is None
-    assert cache.get(pem_ca_key) is None
-    assert cache.get(der_user_key) is None
-    assert cache.get(pem_user_key) is None
+    usable_ca.generate_crls(key_backend_options)
+    assert_crls(usable_ca)
 
-    usable_ca.generate_crls(ca_private_key_options)
 
-    der_user_crl = cache.get(der_user_key)
-    pem_user_crl = cache.get(pem_user_key)
-    assert_crl(
-        der_user_crl,
-        idp=user_idp,
-        encoding=Encoding.DER,
-        signer=usable_ca,
-        algorithm=usable_ca.algorithm,
-    )
-    assert_crl(
-        pem_user_crl,
-        idp=user_idp,
-        encoding=Encoding.PEM,
-        signer=usable_ca,
-        algorithm=usable_ca.algorithm,
-    )
+@pytest.mark.usefixtures("clear_cache")
+@pytest.mark.freeze_time
+def test_generate_crls_with_regeneration(
+    settings: SettingsWrapper, freezer: FrozenDateTimeFactory, usable_root: CertificateAuthority
+) -> None:
+    """Test generation of CRLs."""
+    print(TIMESTAMPS["everything_valid"])
+    key_backend_options = StoragesUsePrivateKeyOptions()
 
-    der_ca_crl = cache.get(der_ca_key)
-    pem_ca_crl = cache.get(pem_ca_key)
-    assert_crl(
-        der_ca_crl,
-        idp=ca_idp,
-        encoding=Encoding.DER,
-        signer=usable_ca,
-        algorithm=usable_ca.algorithm,
-    )
-    assert_crl(
-        pem_ca_crl,
-        idp=ca_idp,
-        encoding=Encoding.PEM,
-        signer=usable_ca,
-        algorithm=usable_ca.algorithm,
-    )
+    # Generate CRLs and test contents (just to be sure)
+    usable_root.generate_crls(key_backend_options)
+    assert_crls(usable_root)
 
-    # cache again - which will force triggering a new computation
-    usable_ca.generate_crls(ca_private_key_options)
-
-    # Get CRLs from cache - we have a new CRLNumber
-    der_user_crl = cache.get(der_user_key)
-    pem_user_crl = cache.get(pem_user_key)
-    assert_crl(
-        der_user_crl,
-        idp=user_idp,
-        crl_number=1,
-        encoding=Encoding.DER,
-        signer=usable_ca,
-        algorithm=usable_ca.algorithm,
-    )
-    assert_crl(
-        pem_user_crl,
-        idp=user_idp,
-        crl_number=1,
-        encoding=Encoding.PEM,
-        signer=usable_ca,
-        algorithm=usable_ca.algorithm,
-    )
+    # Generate again - should not do anything, b/c values are cached
+    usable_root.generate_crls(key_backend_options)
+    assert_crls(usable_root)
 
-    der_ca_crl = cache.get(der_ca_key)
-    pem_ca_crl = cache.get(pem_ca_key)
-    assert_crl(
-        der_ca_crl,
-        idp=ca_idp,
-        crl_number=1,
-        encoding=Encoding.DER,
-        signer=usable_ca,
-        algorithm=usable_ca.algorithm,
-    )
-    assert_crl(
-        pem_ca_crl,
-        idp=ca_idp,
-        crl_number=1,
-        encoding=Encoding.PEM,
-        signer=usable_ca,
-        algorithm=usable_ca.algorithm,
-    )
+    # move time ahead so that we regenerate
+    freezer.tick(timedelta(hours=12))
 
-    # clear caches and skip generation
-    cache.clear()
-    crl_profiles = {
-        k: v.model_dump(exclude={"encodings", "scope"}) for k, v in model_settings.CA_CRL_PROFILES.items()
-    }
-    crl_profiles["ca"]["OVERRIDES"][usable_ca.serial] = {"skip": True}
-    crl_profiles["user"]["OVERRIDES"][usable_ca.serial] = {"skip": True}
+    # generate again - values should be cached and no new CRLs are created
+    usable_root.generate_crls(key_backend_options)
+    assert_crls(usable_root, number=1)
 
-    settings.CA_CRL_PROFILES = crl_profiles
-    usable_ca.generate_crls(ca_private_key_options)
 
-    assert cache.get(der_ca_key) is None
-    assert cache.get(pem_ca_key) is None
-    assert cache.get(der_user_key) is None
-    assert cache.get(pem_user_key) is None
+@pytest.mark.usefixtures("clear_cache")
+@pytest.mark.freeze_time(TIMESTAMPS["everything_valid"])
+def test_generate_crls_with_force(settings: SettingsWrapper, usable_root: CertificateAuthority) -> None:
+    """Test the force option."""
+    # Initially generate CRLs
+    usable_root.generate_crls(KEY_BACKEND_OPTIONS)
+    assert_crls(usable_root)
+
+    # Generate CRLs again, forcing regeneration
+    usable_root.generate_crls(KEY_BACKEND_OPTIONS, force=True)
+    assert_crls(usable_root, number=1)
 
 
 @pytest.mark.usefixtures("clear_cache")
@@ -280,6 +218,29 @@ def test_generate_crls_with_overrides(settings: SettingsWrapper, usable_root: Ce
     assert pem_user_crl.next_update_utc == TIMESTAMPS["everything_valid"] + timedelta(days=3)
 
 
+@pytest.mark.usefixtures("clear_cache")
+@pytest.mark.freeze_time(TIMESTAMPS["everything_valid"])
+def test_generate_crls_with_skip(settings: SettingsWrapper, usable_root: CertificateAuthority) -> None:
+    """Test the skip option in CRL profiles."""
+    crl_profiles = {
+        k: v.model_dump(exclude={"encodings", "scope"}) for k, v in model_settings.CA_CRL_PROFILES.items()
+    }
+    crl_profiles["ca"]["OVERRIDES"][usable_root.serial] = {"skip": True}
+    crl_profiles["user"]["OVERRIDES"][usable_root.serial] = {"skip": True}
+
+    settings.CA_CRL_PROFILES = crl_profiles
+    usable_root.generate_crls(StoragesUsePrivateKeyOptions())
+
+    der_user_key = crl_cache_key(usable_root.serial, only_contains_user_certs=True)
+    pem_user_key = crl_cache_key(usable_root.serial, Encoding.PEM, only_contains_user_certs=True)
+    der_ca_key = crl_cache_key(usable_root.serial, only_contains_ca_certs=True)
+    pem_ca_key = crl_cache_key(usable_root.serial, Encoding.PEM, only_contains_ca_certs=True)
+    assert cache.get(der_ca_key) is None
+    assert cache.get(pem_ca_key) is None
+    assert cache.get(der_user_key) is None
+    assert cache.get(pem_user_key) is None
+
+
 @pytest.mark.usefixtures("clear_cache")
 @pytest.mark.freeze_time(TIMESTAMPS["everything_valid"])
 def test_deprecated_cache_crls(usable_root: CertificateAuthority) -> None: