"""Module that contains the CrlModel."""
from __future__ import annotations
from typing import TYPE_CHECKING, Any, ClassVar, Never
from cryptography import x509
from django.core.exceptions import ValidationError
from django.db import models
from django.utils import timezone
from django.utils.translation import gettext_lazy as _
from trustpoint.logger import LoggerMixin
from util.db import CustomDeleteActionModel
if TYPE_CHECKING:
import datetime
from pki.models import CaModel
[docs]
class CrlModel(LoggerMixin, CustomDeleteActionModel):
"""Certificate Revocation List Model.
This model stores CRLs for CAs (both issuing and non-issuing).
Multiple CRLs can exist for a single CA to track CRL history.
"""
[docs]
ca = models.ForeignKey(
'pki.CaModel',
related_name='crls',
on_delete=models.CASCADE,
verbose_name=_('Certificate Authority'),
help_text=_('The CA that issued this CRL'),
null=True,
blank=True,
)
[docs]
crl_pem = models.TextField(
verbose_name=_('CRL in PEM format'),
help_text=_('The Certificate Revocation List in PEM format')
)
[docs]
crl_number = models.PositiveBigIntegerField(
verbose_name=_('CRL Number'),
help_text=_('The CRL number from the CRL extension'),
null=True,
blank=True,
)
[docs]
this_update = models.DateTimeField(
verbose_name=_('This Update'),
help_text=_('The thisUpdate field from the CRL')
)
[docs]
next_update = models.DateTimeField(
verbose_name=_('Next Update'),
null=True,
blank=True,
help_text=_('The nextUpdate field from the CRL')
)
[docs]
is_active = models.BooleanField(
_('Active'),
default=True,
help_text=_('Whether this is the current active CRL for the CA')
)
[docs]
created_at = models.DateTimeField(
verbose_name=_('Created'),
auto_now_add=True
)
[docs]
updated_at = models.DateTimeField(
verbose_name=_('Updated'),
auto_now=True
)
def __str__(self) -> str:
"""Returns a human-readable string representation.
Returns:
str: Human-readable string representation.
"""
if self.ca is not None:
if self.crl_number is not None:
return f'CRL #{self.crl_number} for {self.ca.unique_name}'
return f'CRL for {self.ca.unique_name} (no number)'
if self.crl_number is not None:
return f'CRL #{self.crl_number} (no associated CA)'
return 'CRL (no associated CA)'
def __repr__(self) -> str:
"""Returns a string representation of the instance."""
return f'CrlModel(id={self.pk}, ca={self.ca_id}, crl_number={self.crl_number})'
[docs]
def raise_invalid_signature_error(self) -> Never:
"""Raises a ValidationError indicating an invalid CRL signature."""
raise ValidationError(
_(
'The CRL signature is invalid. '
'This CRL was not signed by this CA.'
)
)
@classmethod
[docs]
def create_from_pem(
cls,
ca: CaModel | None,
crl_pem: str,
*,
set_active: bool = True,
next_update_delta: datetime.timedelta | None = None,
) -> CrlModel:
"""Creates a new CRL from PEM data.
Args:
ca: The CA that issued this CRL. Can be None for CRLs not associated with a CA.
crl_pem: The CRL in PEM format.
set_active: If True and ca is provided, deactivates other CRLs for this CA and sets this as active.
next_update_delta: Optional timedelta to override the CRL's nextUpdate field.
If provided, sets nextUpdate to thisUpdate + delta.
Returns:
CrlModel: The newly created CRL model.
Raises:
ValidationError: If the CRL is invalid or doesn't match the CA.
"""
try:
crl = x509.load_pem_x509_crl(crl_pem.encode())
except Exception as e:
raise ValidationError(_('Failed to parse the CRL. It may be corrupted or invalid.')) from e
if ca is not None:
if not ca.ca_certificate_model:
raise ValidationError(
_('The CA does not have a valid certificate model.')
)
ca_cert = ca.ca_certificate_model.get_certificate_serializer().as_crypto()
# Check if CRL issuer matches CA subject
if crl.issuer != ca_cert.subject:
raise ValidationError(
_('The CRL issuer does not match the CA subject.')
)
if not crl.is_signature_valid(ca_cert.public_key()): # type: ignore[arg-type]
raise ValidationError(
_(
'The CRL signature is invalid. '
'This CRL was not signed by this CA.'
)
)
crl_number = None
try:
crl_number_ext = crl.extensions.get_extension_for_class(x509.CRLNumber)
crl_number = crl_number_ext.value.crl_number
except x509.ExtensionNotFound:
pass
this_update = crl.last_update_utc
next_update = crl.next_update_utc
if next_update_delta is not None:
next_update = this_update + next_update_delta
crl_model = cls(
ca=ca,
crl_pem=crl_pem,
crl_number=crl_number,
this_update=this_update,
next_update=next_update,
is_active=set_active if ca is not None else False,
)
crl_model.save()
if set_active and ca is not None:
cls.objects.filter(ca=ca, is_active=True).exclude(pk=crl_model.pk).update(is_active=False)
return crl_model
[docs]
def get_crl_as_crypto(self) -> x509.CertificateRevocationList:
"""Returns the CRL as a cryptography CertificateRevocationList object.
Returns:
x509.CertificateRevocationList: The CRL.
Raises:
ValidationError: If the CRL cannot be parsed.
"""
try:
return x509.load_pem_x509_crl(self.crl_pem.encode())
except Exception as e:
ca_name = self.ca.unique_name if self.ca else 'no associated CA'
self.logger.exception('Failed to load CRL for CA %s', ca_name)
raise ValidationError(_('Failed to parse the stored CRL.')) from e
[docs]
def get_revoked_serial_numbers(self) -> set[int]:
"""Returns a set of revoked certificate serial numbers.
Returns:
set[int]: Set of revoked serial numbers.
"""
crl = self.get_crl_as_crypto()
return {revoked_cert.serial_number for revoked_cert in crl}
[docs]
def is_certificate_revoked(self, serial_number: int) -> bool:
"""Checks if a certificate with the given serial number is revoked.
Args:
serial_number: The certificate serial number to check.
Returns:
bool: True if the certificate is revoked, False otherwise.
"""
return serial_number in self.get_revoked_serial_numbers()
[docs]
def is_expired(self) -> bool:
"""Checks if this CRL has expired based on nextUpdate.
Returns:
bool: True if the CRL has expired, False otherwise.
"""
if self.next_update is None:
return False
return timezone.now() > self.next_update
@property
[docs]
def days_left(self) -> int:
"""Returns number of days from now until next_update. If expired or no next_update, returns 0."""
if self.next_update is None:
return 0
now = timezone.now()
if self.next_update > now:
return (self.next_update - now).days
return 0
[docs]
def get_validity_hours(self) -> float | None:
"""Returns the validity period in hours.
Returns:
float | None: The validity period in hours, or None if not set.
"""
if self.next_update is None:
return None
validity_period = self.next_update - self.this_update
return validity_period.total_seconds() / 3600
[docs]
def save(self, *args: Any, **kwargs: Any) -> None:
"""Override save to validate before saving."""
# Ensure only one active CRL per CA
if self.is_active:
CrlModel.objects.filter(ca=self.ca, is_active=True).exclude(pk=self.pk).update(is_active=False)
super().save(*args, **kwargs)
[docs]
def pre_delete(self) -> None:
"""Called before deleting the model."""
ca_name = self.ca.unique_name if self.ca else 'no associated CA'
self.logger.info('Deleting CRL for CA %s (CRL Number: %s)', ca_name, self.crl_number)