Source code for django_ca.celery.messages

# 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/>.

"""Messages for Celery tasks."""

from typing import Self

from pydantic import AwareDatetime, Field, model_validator

from django_ca.celery import CeleryMessageModel
from django_ca.conf import model_settings
from django_ca.constants import SIGNATURE_HASH_ALGORITHM_TYPES
from django_ca.pydantic import NameModel
from django_ca.pydantic.extensions import ConfigurableExtensionModel
from django_ca.pydantic.type_aliases import CSRType, Serial
from django_ca.typehints import JSON, SignatureHashAlgorithm, SignatureHashAlgorithmName

KeyBackendOptions = dict[str, JSON]


[docs] class UseCertificateAuthorityTaskArgs(CeleryMessageModel): """Parameters for using a single certificate authority.""" serial: Serial force: bool = False key_backend_options: KeyBackendOptions = Field(default_factory=dict)
[docs] class UseCertificateAuthoritiesTaskArgs(CeleryMessageModel): """Parameters for using multiple certificate authorities.""" serials: tuple[Serial, ...] = Field(default_factory=tuple) exclude: tuple[Serial, ...] = Field(default_factory=tuple) force: bool = False key_backend_options: dict[str, KeyBackendOptions] = Field(default_factory=dict)
[docs] @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
[docs] class ApiSignCertificateTaskArgs(CeleryMessageModel): """Parameters for ``django_ca.tasks.api_sign_certificate``.""" order_pk: int csr: CSRType subject: NameModel algorithm: SignatureHashAlgorithmName | None = None not_after: AwareDatetime | None = None extensions: list[ConfigurableExtensionModel] = Field(default_factory=list) profile: str = model_settings.CA_DEFAULT_PROFILE autogenerated: bool = False key_backend_options: KeyBackendOptions = Field(default_factory=dict)
[docs] def get_algorithm(self) -> SignatureHashAlgorithm | None: """Get algorithm class if set.""" if self.algorithm is not None: return SIGNATURE_HASH_ALGORITHM_TYPES[self.algorithm]() return None