Class Certificate

celery/security/certificate.py:31–73

X.509 certificate.

class Certificate:
    """X.509 certificate."""

    def __init__(self, cert: str) -> None:
        with reraise_errors(
            'Invalid certificate: {0!r}', errors=(ValueError,)
        ):
            self._cert = load_pem_x509_certificate(
                ensure_bytes(cert), backend=default_backend())

            if not isinstance(self._cert.public_key(), rsa.RSAPublicKey):
                raise ValueError("Non-RSA certificates are not supported.")

    def has_expired(self) -> bool:
        """Check if the certificate has expired."""
        return datetime.datetime.now(datetime.timezone.utc) >= self._cert.not_valid_after_utc

    def get_pubkey(self) -> (
        DSAPublicKey | EllipticCurvePublicKey | Ed448PublicKey | Ed25519PublicKey | RSAPublicKey
    ):
        return self._cert.public_key()

    def get_serial_number(self) -> int:
        """Return the serial number in the certificate."""
        return self._cert.serial_number

    def get_issuer(self) -> str:
        """Return issuer (CA) as a string."""
        return ' '.join(x.value for x in self._cert.issuer)

    def get_id(self) -> str:
        """Serial number/issuer pair uniquely identifies a certificate."""
        return f'{self.get_issuer()} {self.get_serial_number()}'

    def verify(self, data: bytes, signature: bytes, digest: HashAlgorithm | Prehashed) -> None:
        """Verify signature for string containing data."""
        with reraise_errors('Bad signature: {0!r}'):

            pad = padding.PSS(
                mgf=padding.MGF1(digest),
                salt_length=padding.PSS.MAX_LENGTH)

            self.get_pubkey().verify(signature, ensure_bytes(data), pad, digest)


class CertStore: