use unified interface for generating OCSP keys

mathiasertl · mathiasertl · commit d4f4ac29cefa · 2026-01-21T18:19:29.000+01:00
diff --git a/ca/ca/settings.py b/ca/ca/settings.py
@@ -112,7 +112,7 @@
 
 # Make this unique, and don't share it with anybody.
 SECRET_KEY = ""
-SECRET_KEY_FILE = None
+SECRET_KEY_FILE = ""
 
 MIDDLEWARE = [
     "django.middleware.security.SecurityMiddleware",
diff --git a/ca/ca/settings_utils.py b/ca/ca/settings_utils.py
@@ -180,12 +180,13 @@ class ProjectSettingsModelMixin:
         "This setting has no effect if you define the ``LOGGING`` setting."
     )
     STORAGES: dict[str, dict[str, Any]] = Field(default_factory=dict)
-    SECRET_KEY_FILE: str | None = Field(
-        default=None,
+    SECRET_KEY_FILE: str = Field(
+        default="",
         description="A path to a file that stores Django's `SECRET_KEY "
         "<https://docs.djangoproject.com/en/dev/ref/settings/#std:setting-SECRET_KEY>`_. The setting is only "
         "used if no ``SECRET_KEY`` is defined.",
         examples=["/var/lib/django-ca/certs/ca/shared/secret_key"],
+        json_schema_extra={"default_explanation": "Not set."},
     )
     USE_TZ: bool
 
diff --git a/ca/django_ca/celery/messages.py b/ca/django_ca/celery/messages.py
@@ -46,6 +46,7 @@ class UseCertificateAuthoritiesTaskArgs(CeleryMessageModel):
 
     @model_validator(mode="after")
     def validate_exclude(self) -> Self:
+        """Validator to make sure that not both `serials` and `exclude` is set."""
         if self.serials and self.exclude:
             raise ValueError("Message cannot contain both serials and excluded serials.")
         return self
diff --git a/ca/django_ca/management/base.py b/ca/django_ca/management/base.py
@@ -14,11 +14,8 @@
 """Command subclasses and argparse helpers for django-ca."""
 
 import abc
-import argparse
-import typing
-from collections.abc import Callable
 from datetime import UTC, datetime, timedelta
-from typing import Any
+from typing import TYPE_CHECKING, Any, ClassVar
 
 from cryptography import x509
 from cryptography.x509.oid import AuthorityInformationAccessOID
@@ -44,7 +41,8 @@
     ConfigurableExtensionType,
 )
 
-if typing.TYPE_CHECKING:
+if TYPE_CHECKING:
+    from celery.local import Proxy  # celery.local is defined in our stubs
     from django_stubs_ext import StrOrPromise
 
 
@@ -381,7 +379,7 @@ class BaseSignCertCommand(UsePrivateKeyMixin, BaseSignCommand, metaclass=abc.ABC
     """Base class for commands signing certificates (sign_cert, resign_cert)."""
 
     add_extensions_help = ""  # concrete classes should set this
-    subject_help: typing.ClassVar[str]  # concrete classes should set this
+    subject_help: ClassVar[str]  # concrete classes should set this
 
     def add_base_args(self, parser: CommandParser, no_default_ca: bool = False) -> ArgumentGroup:
         """Add common arguments for signing certificates."""
@@ -584,10 +582,10 @@ class GenerateCommandBase(UsePrivateKeyMixin, BaseCommand):
     """Base class for commands to generate CRLs and OCSP keys."""
 
     what: str
-    single_task: Callable[..., Any]
-    multiple_task: Callable[..., Any]
+    single_task: "Proxy[[UseCertificateAuthorityTaskArgs], Any]"
+    multiple_task: "Proxy[[UseCertificateAuthoritiesTaskArgs], Any]"
 
-    def add_arguments(self, parser: argparse.ArgumentParser) -> None:
+    def add_arguments(self, parser: CommandParser) -> None:
         parser.add_argument(
             "serials",
             metavar="SERIAL",
@@ -614,14 +612,23 @@ def handle(self, serials: list[str], exclude: list[str], force: bool, **options:
             raise CommandError("Cannot name serials and exclude list at the same time.")
 
         if len(serials) == 1:
-            ca, key_backend_options = self.get_key_backend_options(serials[0], options)
-
-            data = UseCertificateAuthorityTaskArgs(
-                serial=ca.serial,
-                force=force,
-                key_backend_options=key_backend_options.model_dump(mode="json", exclude_unset=True),
-            )
-            run_task(self.single_task, data)
+            try:
+                ca, key_backend_options = self.get_key_backend_options(serials[0], options)
+
+                single_data = UseCertificateAuthorityTaskArgs(
+                    serial=ca.serial,
+                    force=force,
+                    key_backend_options=key_backend_options.model_dump(mode="json", exclude_unset=True),
+                )
+
+                run_task(self.single_task, single_data)
+            except Exception as ex:
+                # Exceptions only happen if Celery is not used or the connection to the broker fails
+                raise CommandError(ex) from ex
         else:
-            data = UseCertificateAuthoritiesTaskArgs(serials=serials, exclude=exclude, force=force)
-            run_task(self.multiple_task, data)
+            multiple_data = UseCertificateAuthoritiesTaskArgs(serials=serials, exclude=exclude, force=force)
+            try:
+                run_task(self.multiple_task, multiple_data)
+            except Exception as ex:  # pragma: no cover  # does not usually happen
+                # Exceptions only happen if Celery is not used or the connection to the broker fails
+                raise CommandError(ex) from ex
diff --git a/ca/django_ca/management/commands/cache_crls.py b/ca/django_ca/management/commands/cache_crls.py
@@ -26,11 +26,11 @@
 class Command(GenerateCrlsCommand):  # noqa: D101
     help = "(Deprecated) Cache CRLs. Use generate_crls instead."
 
-    def handle(self, serials: list[str], **options: Any) -> None:
+    def handle(self, serials: list[str], exclude: list[str], force: bool, **options: Any) -> None:
         self.stderr.write(
             self.style.WARNING(
                 "Warning: This command is deprecated. Please use generate_crls instead. "
                 "This alias will be removed in django_ca~=3.2.0."
             )
         )
-        super().handle(serials, **options)
+        super().handle(serials, exclude=exclude, force=force, **options)
diff --git a/ca/django_ca/management/commands/generate_ocsp_keys.py b/ca/django_ca/management/commands/generate_ocsp_keys.py
@@ -16,81 +16,14 @@
 .. seealso:: https://docs.djangoproject.com/en/dev/howto/custom-management-commands/
 """
 
-from collections.abc import Iterable
-from typing import Any
+from django_ca.management.base import GenerateCommandBase
+from django_ca.tasks import generate_ocsp_key, generate_ocsp_keys
 
-from pydantic import ValidationError
 
-from django.core.management.base import CommandError, CommandParser
-
-from django_ca.celery import run_task
-from django_ca.celery.messages import UseCertificateAuthorityTaskArgs
-from django_ca.conf import model_settings
-from django_ca.management.base import BaseCommand
-from django_ca.management.mixins import UsePrivateKeyMixin
-from django_ca.models import CertificateAuthority
-from django_ca.tasks import generate_ocsp_key
-from django_ca.utils import add_colons
-
-
-class Command(UsePrivateKeyMixin, BaseCommand):
+class Command(GenerateCommandBase):
     """Implement the :command:`manage.py generate_ocsp_keys` command."""
 
     help = "Generate OCSP keys."
-
-    def add_arguments(self, parser: CommandParser) -> None:
-        parser.add_argument(
-            "serials",
-            metavar="SERIAL",
-            nargs="*",
-            help="Generate OCSP keys only for the given CA. If omitted, generate keys for all CAs.",
-        )
-
-        parser.add_argument(
-            "--force",
-            action="store_true",
-            default=False,
-            help="Force regeneration of OCSP responder certificates.",
-        )
-        parser.add_argument("--quiet", action="store_true", default=False, help="Do not output warnings.")
-        self.add_use_private_key_arguments(parser)
-
-    def handle(self, serials: Iterable[str], quiet: bool, force: bool, **options: Any) -> None:
-        if not serials:
-            serials = CertificateAuthority.objects.all().order_by("serial").values_list("serial", flat=True)
-
-        if "ocsp" not in model_settings.CA_PROFILES:
-            raise CommandError("ocsp: Undefined profile.")
-
-        errors = 0
-        for serial in serials:
-            serial = serial.replace(":", "").strip().upper()
-            hr_serial = add_colons(serial)
-            try:
-                ca: CertificateAuthority = CertificateAuthority.objects.get(serial=serial)
-            except CertificateAuthority.DoesNotExist:
-                self.stderr.write(self.style.ERROR(f"{hr_serial}: Unknown CA."))
-                continue
-
-            try:
-                key_backend_options = ca.key_backend.get_use_private_key_options(ca, options)
-            except ValidationError as ex:
-                self.validation_error_to_command_error(ex)
-            except Exception as ex:  # pragma: no cover  # pylint: disable=broad-exception-caught
-                if quiet is False:
-                    self.stderr.write(self.style.WARNING(f"{hr_serial}: {ex}"))
-                continue
-
-            try:
-                message = UseCertificateAuthorityTaskArgs(
-                    serial=serial,
-                    key_backend_options=key_backend_options.model_dump(mode="json", exclude_unset=True),
-                    force=force,
-                )
-                run_task(generate_ocsp_key, message)
-            except Exception as ex:  # pylint: disable=broad-exception-caught
-                self.stderr.write(f"{serial}: {ex}")
-                errors += 1
-
-        if errors:
-            raise CommandError(f"Generation of {errors} OCSP key(s) failed.")
+    what = "OCSP keys"
+    single_task = generate_ocsp_key
+    multiple_task = generate_ocsp_keys
diff --git a/ca/django_ca/management/mixins.py b/ca/django_ca/management/mixins.py
@@ -674,9 +674,10 @@ def get_signing_options(
 
         return key_backend_options, algorithm
 
-    def get_key_backend_options(
+    def get_key_backend_options(  # type: ignore[return]  # b/c of validation_error_to_command_error
         self, serial: str, options: dict[str, Any]
     ) -> tuple[CertificateAuthority, BaseModel]:
+        """Get CA and key backend options for the given input data."""
         serial = sanitize_serial(serial)
         hr_serial = add_colons(serial)
         try:
@@ -687,6 +688,6 @@ def get_key_backend_options(
         try:
             return ca, ca.key_backend.get_use_private_key_options(ca, options)
         except ValidationError as ex:
-            self.validation_error_to_command_error(ex)
-        except Exception as ex:  # pragma: no cover  # pylint: disable=broad-exception-caught
+            self.validation_error_to_command_error(ex)  # type: ignore[attr-defined]
+        except Exception as ex:  # pragma: no cover
             raise CommandError(str(ex)) from ex
diff --git a/ca/django_ca/tasks.py b/ca/django_ca/tasks.py
@@ -126,7 +126,7 @@ def generate_crls(data: UseCertificateAuthoritiesTaskArgs | None = None) -> None
             # NOTE: When using Celery, an exception will only be raised here if task.delay() itself raises an
             # exception, e.g. if the connection to the broker fails. Without celery, exceptions in
             # `generate_crl()` are raised here directly.
-            log.exception("Error caching CRL for %s", serial)
+            log.exception("Error generating CRL for %s", serial)
 
 
 @shared_task(base=DjangoCaTask)
diff --git a/ca/django_ca/tests/commands/test_generate_crls.py b/ca/django_ca/tests/commands/test_generate_crls.py
@@ -95,4 +95,4 @@ def test_deprecated_cmd_name(usable_root: CertificateAuthority) -> None:
 def test_with_serial_and_exclude() -> None:
     """Test passing both an explicit serial and an exclusion (makes no sense)."""
     with assert_command_error(r"^Cannot name serials and exclude list at the same time\.$"):
-        cmd("generate_crls", serials=["abc"], exclude=["def"])
+        cmd("generate_crls", ["abc"], exclude=["def"])
diff --git a/ca/django_ca/tests/commands/test_generate_ocsp_keys.py b/ca/django_ca/tests/commands/test_generate_ocsp_keys.py
@@ -13,7 +13,6 @@
 
 """Test the generate_ocsp_keys management command."""
 
-import io
 import typing
 from collections.abc import Iterable
 from datetime import UTC, datetime, timedelta
@@ -143,32 +142,28 @@ def test_with_celery(settings: SettingsWrapper, usable_root: CertificateAuthorit
     assert_no_key(usable_root.serial)
 
 
-def test_without_serial(
-    settings: SettingsWrapper, root: CertificateAuthority, ec: CertificateAuthority
-) -> None:
+def test_without_serial(settings: SettingsWrapper) -> None:
     """Test for all CAs."""
     settings.CA_USE_CELERY = True
-    kwargs = {"key_backend_options": {"password": None}, "force": False}
     with mock_celery_task(
-        "django_ca.tasks.generate_ocsp_key",
-        (mock.call(tuple(), {"data": {"serial": root.serial, **kwargs}})),
-        (mock.call(tuple(), {"data": {"serial": ec.serial, **kwargs}})),
+        "django_ca.tasks.generate_ocsp_keys",
+        (mock.call(tuple(), {"data": {"serials": [], "exclude": [], "force": False}})),
     ):
         cmd("generate_ocsp_keys")
 
 
 @pytest.mark.django_db
-def test_wrong_serial() -> None:
-    """Try passing an unknown CA."""
-    generate_ocsp_keys("ZZZZZ", stderr="0Z:ZZ:ZZ: Unknown CA.\n", no_color=True)
+def test_unknown_serial() -> None:
+    """Try passing an invalid CA."""
+    with assert_command_error(r"^0A:BC: Unknown CA\.$"):
+        generate_ocsp_keys("abc")
 
 
-def test_no_ocsp_profile(settings: SettingsWrapper, root: CertificateAuthority) -> None:
-    """Try when there is no OCSP profile."""
-    settings.CA_PROFILES = {"ocsp": None}
-    with assert_command_error(r"^ocsp: Undefined profile\.$"):
-        generate_ocsp_keys(root.serial)
-    assert_no_key(root.serial)
+@pytest.mark.django_db
+def test_invalid_serial() -> None:
+    """Try passing an invalid CA."""
+    with assert_command_error(r"^ZZZZZ: Serial has invalid characters$"):
+        generate_ocsp_keys("ZZZZZ")
 
 
 def test_deprecated_command_name(usable_root: CertificateAuthority) -> None:
@@ -185,10 +180,8 @@ def test_deprecated_command_name(usable_root: CertificateAuthority) -> None:
 @pytest.mark.usefixtures("tmpcadir")
 def test_without_private_key(root: CertificateAuthority) -> None:
     """Try regenerating the OCSP key when no CA private key is available."""
-    stderr_buffer = io.StringIO()
-    with assert_command_error(r"Generation of 1 OCSP key\(s\) failed\."):
-        cmd("generate_ocsp_keys", root.serial, stderr=stderr_buffer)
-    assert "No such file or directory" in stderr_buffer.getvalue()
+    with assert_command_error(r"No such file or directory"):
+        cmd("generate_ocsp_keys", root.serial)
 
 
 def test_model_validation_error(root: CertificateAuthority) -> None:
diff --git a/ca/django_ca/tests/tasks/test_generate_crls.py b/ca/django_ca/tests/tasks/test_generate_crls.py
@@ -81,7 +81,7 @@ def test_with_exception_child_task(caplog: LogCaptureFixture) -> None:
         caplog.at_level(logging.INFO),
     ):
         generate_crls()
-    assert "Error caching CRL" in caplog.text
+    assert "Error generating CRL" in caplog.text
 
 
 def test_deprecated_task_name(usable_root: CertificateAuthority) -> None:
diff --git a/ca/django_ca/tests/tasks/test_messages.py b/ca/django_ca/tests/tasks/test_messages.py
@@ -0,0 +1,24 @@
+# This file is part of django-ca (https://github.com/mathiasertl/django-ca).
+#
+# django-ca is free software: you can redistribute it and/or modify it under the terms of the GNU General
+# Public License as published by the Free Software Foundation, either version 3 of the License, or (at your
+# option) any later version.
+#
+# django-ca is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the
+# implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
+# for more details.
+#
+# You should have received a copy of the GNU General Public License along with django-ca. If not, see
+# <http://www.gnu.org/licenses/>.
+
+"""Test celery message validators."""
+
+import pytest
+
+from django_ca.celery.messages import UseCertificateAuthoritiesTaskArgs
+
+
+def test_serial_and_exclude() -> None:
+    """Test passing a serial and an exclude."""
+    with pytest.raises(ValueError, match=r"Message cannot contain both serials and excluded serials\."):
+        UseCertificateAuthoritiesTaskArgs(serials=["abc"], exclude=["def"])
diff --git a/docs/source/changelog/TBR_2.6.0.rst b/docs/source/changelog/TBR_2.6.0.rst
@@ -55,8 +55,12 @@ No changes yet.
 OCSP and CRLs
 *************
 
-* CRLs are now only regenerated if they expire within a renewal timeframe, similar to OCSP keys. This enables
+* CRLs are now only regenerated if they expire within a renewal interval, similar to OCSP keys. This enables
   running the task to generate CRLs much more frequently and only re-generating CRLs when required.
+* Celery tasks and management commands now accept key backend options when acting on a single CA and an
+  `exclude` parameter to exclude CAs. This allows you to exclude CAs that require special key backend options
+  (e.g. a password) from the default periodic task, and adding an additional, dedicated periodic task with key
+  backend options for that CA.
 
 ********
 REST API
@@ -73,6 +77,11 @@ Command-line
   will be removed in ``django-ca~=3.2.0``.
 * The `regenerate_ocsp_keys` command was renamed to :command:`manage.py generate_ocsp_keys` for consistency.
   The old name will be removed in ``django-ca~=3.2.0``.
+* :command:`manage.py generate_crls` now allows passing key backend options if exactly one CA is specified.
+* :command:`manage.py generate_crls` and :command:`manage.py generate_ocsp_keys` now use a unified interface
+  and have the same arguments.
+* :command:`manage.py generate_crls` and :command:`manage.py generate_ocsp_keys` now allow forcing generation
+  of CRLs/OCSP keys (even if not due for renewal) and excluding CAs from renewal.
 
 ***************
 Admin interface
@@ -90,6 +99,11 @@ Celery tasks
   name will be removed in ``django-ca~=3.1.0``.
 * **BACKWARDS INCOMPATIBLE:** Arguments to Celery tasks are now passed as Pydantic models, greatly improving
   type safety. This will require you to change your code if you call the task directly.
+* :py:func:`~django_ca.tasks.generate_crls` and :py:func:`~django_ca.tasks.generate_ocsp_keys` now support
+  the `force` parameter to force generating CRLs/OCSP keys, even if not due for renewal.
+* :py:func:`~django_ca.tasks.generate_crls` and :py:func:`~django_ca.tasks.generate_ocsp_keys` now support
+  the `exclude` parameter to exclude CAs from generating CRLs/OCSP keys.
+
 
 **********
 Python API
diff --git a/pyproject.toml b/pyproject.toml
@@ -428,7 +428,7 @@ env = [
 DJANGO_SETTINGS_MODULE = "ca.test_settings"
 addopts = [
     "--cov",
-    "--cov-report=html:docs/build/coverage/",
+    "--cov-report=html",
     "--cov-fail-under=100",
     "--random-order",
 ]

Original file line number	Diff line number	Diff line change
`@@ -26,11 +26,11 @@`
`26`	`26`	`class Command(GenerateCrlsCommand): # noqa: D101`
`27`	`27`	`help = "(Deprecated) Cache CRLs. Use generate_crls instead."`
`28`	`28`
`29`		`- def handle(self, serials: list[str], **options: Any) -> None:`
	`29`	`+ def handle(self, serials: list[str], exclude: list[str], force: bool, **options: Any) -> None:`
`30`	`30`	`self.stderr.write(`
`31`	`31`	`self.style.WARNING(`
`32`	`32`	`"Warning: This command is deprecated. Please use generate_crls instead. "`
`33`	`33`	`"This alias will be removed in django_ca~=3.2.0."`
`34`	`34`	`)`
`35`	`35`	`)`
`36`		`- super().handle(serials, **options)`
	`36`	`+ super().handle(serials, exclude=exclude, force=force, **options)`