certificate.py

import logging

from cryptography import x509
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.x509.oid import ExtensionOID, NameOID
from pytz import utc

from django.shortcuts import get_object_or_404
from django.db import models
from django.core.exceptions import SuspiciousOperation
from django.contrib.auth import get_user_model
from rest_framework.exceptions import ValidationError
from api.utils import validate_label
from api.exceptions import ServiceUnavailable
from scheduler import KubeException
from .base import AuditedModel
from .domain import Domain

User = get_user_model()
logger = logging.getLogger(__name__)


def get_subj_alt_name(peer_cert):
    try:
        san_extension = peer_cert.extensions.get_extension_for_oid(
            ExtensionOID.SUBJECT_ALTERNATIVE_NAME
        )
    except x509.ExtensionNotFound:
        return []

    return san_extension.value.get_values_for_type(x509.DNSName)


def public_keys_match(cert_public_key, private_key):
    def _public_key_der(key):
        """Serialize any public key to DER SubjectPublicKeyInfo bytes."""
        return key.public_bytes(
            serialization.Encoding.DER,
            serialization.PublicFormat.SubjectPublicKeyInfo,
        )

    return _public_key_der(cert_public_key) == _public_key_der(
        private_key.public_key()
    )


def get_certificate_not_valid_after(certificate):
    if hasattr(certificate, 'not_valid_after_utc'):
        return certificate.not_valid_after_utc

    return certificate.not_valid_after.replace(tzinfo=utc)


def get_certificate_not_valid_before(certificate):
    if hasattr(certificate, 'not_valid_before_utc'):
        return certificate.not_valid_before_utc

    return certificate.not_valid_before.replace(tzinfo=utc)


def validate_certificate(value):
    try:
        certificate_bytes = value.encode('utf-8') if isinstance(value, str) else value
        return x509.load_pem_x509_certificate(certificate_bytes)
    except (TypeError, ValueError) as e:
        raise ValidationError('Could not load certificate: {}'.format(e))


def validate_private_key(value):
    try:
        private_key_bytes = value.encode('utf-8') if isinstance(value, str) else value
        return serialization.load_pem_private_key(private_key_bytes, password=None)
    except (TypeError, ValueError) as e:
        raise ValidationError('Could not load private key: {}'.format(e))


def validate_cert_pair(certificate, private_key):
    # Load and validate the certificate and private key
    try:
        cert = validate_certificate(certificate)
        pkey = validate_private_key(private_key)
    except ValidationError as e:
        # The certificate and key should already have been validated
        raise SuspiciousOperation(e)

    if not public_keys_match(cert.public_key(), pkey):
        raise ValidationError('Certificate and private key do not match!')

    # Return tuple if everything went ok
    return (cert, pkey)


class Certificate(AuditedModel):
    """
    Public and private key pair used to secure application traffic at the router.
    """
    app = models.ForeignKey('App', on_delete=models.CASCADE)
    name = models.CharField(max_length=253, validators=[validate_label])
    # there is no upper limit on the size of an x.509 certificate
    certificate = models.TextField(validators=[validate_certificate])
    key = models.TextField(validators=[validate_private_key])
    # X.509 certificates allow any string of information as the common name.
    common_name = models.TextField(editable=False, unique=False, null=True)
    # A list of DNS records if certificate has SubjectAltName
    san = models.JSONField(default=None, null=True)
    # SHA256 fingerprint
    fingerprint = models.CharField(max_length=96, editable=False)
    # Expires and Start time of cert
    expires = models.DateTimeField(editable=False)
    starts = models.DateTimeField(editable=False)
    issuer = models.TextField(editable=False)
    subject = models.TextField(editable=False)

    class Meta:
        ordering = ['name', 'common_name', 'expires']
        unique_together = ('app', 'name')

    @property
    def domains(self):
        domains = []
        for data in Domain.objects.filter(certificate=self).distinct().order_by('domain'):
            domains.append(data.domain)

        return domains

    @property
    def certname(self):
        return '%s-certificate' % self.name

    def __str__(self):
        return self.name

    def save(self, *args, **kwargs):
        # Validate the provided certificate and key pair and test for a mismatch
        certificate, _ = validate_cert_pair(self.certificate, self.key)

        if not self.common_name:
            common_names = certificate.subject.get_attributes_for_oid(NameOID.COMMON_NAME)
            self.common_name = common_names[0].value if common_names else None

        # Grab expire date of the certificate
        if not self.expires:
            self.expires = get_certificate_not_valid_after(certificate)

        # Grab the start date of the certificate
        if not self.starts:
            self.starts = get_certificate_not_valid_before(certificate)

        # process issuers - separate each key/value with a slash
        self.issuer = certificate.issuer.rfc4514_string()

        # process subject - separate each key/value with a slash
        self.subject = certificate.subject.rfc4514_string()

        # public fingerprint of certificate
        fingerprint = certificate.fingerprint(hashes.SHA256()).hex().upper()
        self.fingerprint = ':'.join(fingerprint[i:i + 2] for i in range(0, len(fingerprint), 2))

        # SubjectAltName from the certificate - return a list
        self.san = get_subj_alt_name(certificate)

        return super(Certificate, self).save(*args, **kwargs)

    def delete(self, *args, **kwargs):
        # Remove from k8s and domain object if there are any
        if self.domains:
            for domain in self.domains:
                kwargs['domain'] = domain
                self.detach(*args, **kwargs)
                del kwargs['domain']

        # Delete from DB
        return super(Certificate, self).delete(*args, **kwargs)

    def attach(self, *args, **kwargs):
        # add the certificate to the domain
        domain = get_object_or_404(Domain, domain=kwargs['domain'])
        # create in kubernetes
        self.attach_in_kubernetes(domain)
        domain.certificate = self
        domain.save()

    def attach_in_kubernetes(self, domain):
        """Creates the certificate as a kubernetes secret"""
        # only create if it exists - We raise an exception when a secret doesn't exist
        try:
            namespace = domain.app.id
            data = {
                'tls.crt': self.certificate,
                'tls.key': self.key
            }

            secret = self.scheduler.secret.get(namespace, self.certname).json()['data']
        except KubeException:
            self.scheduler.secret.create(namespace, self.certname, data)
        else:
            # update cert secret to the TLS Ingress format if required
            if secret != data:
                try:
                    self.scheduler.secret.update(namespace, self.certname, data)
                except KubeException as e:
                    msg = 'There was a problem updating the certificate secret ' \
                          '{} for {}'.format(self.certname, namespace)
                    raise ServiceUnavailable(msg) from e

    def detach(self, *args, **kwargs):
        # remove the certificate from the domain
        domain = get_object_or_404(Domain, domain=kwargs['domain'])
        domain.certificate = None
        domain.save()

        namespace = domain.app.id

        # only delete if it exists and if no other domains depend on secret
        if len(self.domains) == 0:
            try:
                # We raise an exception when a secret doesn't exist
                self.scheduler.secret.get(namespace, self.certname)
                self.scheduler.secret.delete(namespace, self.certname)
            except KubeException as e:
                raise ServiceUnavailable(
                    "Could not delete certificate secret {} for application {}".format(
                        self.certname, namespace)) from e