aswi2021jmsd-gitlab4/src/services/certificate_service.py @ 9c704fb1

from typing import List

from injector import inject

from src.config.configuration import Configuration
from src.constants import ROOT_CA_ID, INTERMEDIATE_CA_ID, CA_ID, CERTIFICATE_ID, CERTIFICATE_STATES, \
    CERTIFICATE_REVOCATION_REASONS
from src.dao.certificate_repository import CertificateRepository
from src.model.certificate import Certificate
from src.model.private_key import PrivateKey
from src.model.subject import Subject
from src.services.cryptography import CryptographyService

import time

DATE_FORMAT = "%d.%m.%Y %H:%M:%S"
CA_EXTENSIONS = "basicConstraints=critical,CA:TRUE"
CRL_EXTENSION = "crlDistributionPoints=URI:"
OCSP_EXTENSION = "authorityInfoAccess=OCSP;URI:"
STATUS_REVOKED = "revoked"
STATUS_VALID = "valid"


class CertificateService:

    @inject
    def __init__(self, cryptography_service: CryptographyService,
                 certificate_repository: CertificateRepository,
                 configuration: Configuration):
        self.cryptography_service = cryptography_service
        self.certificate_repository = certificate_repository
        self.configuration = configuration

    # TODO usages present in method parameters but not in class diagram
    def create_root_ca(self, key: PrivateKey, subject: Subject, extensions: str = "", config: str = "",
                       usages=None, days=30):
        """
        Creates a root CA certificate based on the given parameters.
        :param key: Private key to be used when generating the certificate
        :param subject: Subject to be used put into the certificate
        :param config: String containing the configuration to be used
        :param extensions: Name of the section in the configuration representing extensions
        :param usages: A dictionary containing usages of the certificate to be generated (see constants.py)
        :param days: Number of days for which the generated cert. will be considered valid
        :return: An instance of Certificate class representing the generated root CA cert
        """
        if usages is None:
            usages = {}

        cert_id = self.certificate_repository.get_next_id()
        extensions = extensions + "\n" + CRL_EXTENSION + " " + self.__get_crl_endpoint(cert_id)
        extensions = extensions + "\n" + OCSP_EXTENSION + " " + self.__get_ocsp_endpoint(cert_id)

        # create a new self signed  certificate
        cert_pem = self.cryptography_service.create_sscrt(subject, key.private_key, key_pass=key.password,
                                                          extensions=extensions, config=config, days=days)
        # specify CA usage
        usages[CA_ID] = True

        # wrap into Certificate class
        certificate = self.__create_wrapper(cert_pem, key.private_key_id, usages, 0,
                                            ROOT_CA_ID)

        # store the wrapper into the repository
        created_id = self.certificate_repository.create(certificate)

        # assign the generated ID to the inserted certificate
        certificate.certificate_id = created_id

        return certificate

    def __create_wrapper(self, cert_pem, private_key_id, usages, parent_id, cert_type):
        """
        Wraps the given parameters using the Certificate class. Uses CryptographyService to find out the notBefore and
        notAfter fields.
        :param cert_pem: PEM of the cert. to be wrapped
        :param private_key_id: ID of the private key used to create the given certificate
        :param usages: A dictionary containing usages of the generated certificate generated (see constants.py)
        :param parent_id: ID of the CA that issued this certificate
        :param cert_type: Type of this certificate (see constants.py)
        :return: An instance of the Certificate class wrapping the values passed  via method parameters
        """
        # parse the generated pem for subject and notBefore/notAfter fields
        # TODO this could be improved in the future in such way that calling openssl is not required to parse the dates
        subj, not_before, not_after = self.cryptography_service.parse_cert_pem(cert_pem)
        # format the parsed date
        not_before_formatted = time.strftime(DATE_FORMAT, not_before)
        not_after_formatted = time.strftime(DATE_FORMAT, not_after)

        # create a certificate wrapper
        certificate = Certificate(-1, subj.common_name, not_before_formatted, not_after_formatted, cert_pem,
                                  private_key_id, cert_type, parent_id, usages)

        return certificate

    # TODO config parameter present in class diagram but not here (unused)
    def create_ca(self, subject_key: PrivateKey, subject: Subject, issuer_cert: Certificate, issuer_key: PrivateKey,
                  extensions: str = "", days: int = 30, usages=None):
        """
        Creates an intermediate CA certificate issued by the given parent CA.
        :param subject_key: Private key to be used when generating the certificate
        :param subject: Subject to be used put into the certificate
        :param issuer_cert: Issuer certificate that will sign the CSR required to create an intermediate CA
        :param issuer_key: PK used to generate the issuer certificate
        :param extensions: Extensions to be used when generating the certificate
        :param usages: A dictionary containing usages of the certificate to be generated (see constants.py)
        :param days: Number of days for which the generated cert. will be considered valid
        :return: An instance of Certificate class representing the generated intermediate CA cert
        """
        if usages is None:
            usages = {}

        extensions = extensions + "\n" + CA_EXTENSIONS
        # Add CRL and OCSP distribution point to certificate extensions
        cert_id = self.certificate_repository.get_next_id()
        extensions = extensions + "\n" + CRL_EXTENSION + " " + self.__get_crl_endpoint(cert_id)
        extensions = extensions + "\n" + OCSP_EXTENSION + " " + self.__get_ocsp_endpoint(cert_id)

        # TODO implement AIA URI via extensions
        cert_pem = self.cryptography_service.create_crt(subject, subject_key.private_key, issuer_cert.pem_data,
                                                        issuer_key.private_key,
                                                        subject_key_pass=subject_key.password,
                                                        issuer_key_pass=issuer_key.password, extensions=extensions,
                                                        days=days)

        # specify CA usage
        usages[CA_ID] = True

        # wrap into Certificate class
        self.__create_wrapper(cert_pem, subject_key.private_key_id, usages,
                              issuer_cert.certificate_id, INTERMEDIATE_CA_ID)

        # parse the generated pem for subject and notBefore/notAfter fields
        subj, not_before, not_after = self.cryptography_service.parse_cert_pem(cert_pem)

        # format the parsed date
        not_before_formatted = time.strftime(DATE_FORMAT, not_before)
        not_after_formatted = time.strftime(DATE_FORMAT, not_after)

        # specify CA usage
        usages[CA_ID] = True

        # create a certificate wrapper
        certificate = Certificate(-1, subject.common_name, not_before_formatted, not_after_formatted, cert_pem,
                                  subject_key.private_key_id, INTERMEDIATE_CA_ID, issuer_cert.certificate_id, usages)

        # store the wrapper into the repository
        created_id = self.certificate_repository.create(certificate)

        # assign the generated ID to the inserted certificate
        certificate.certificate_id = created_id

        return certificate

    def create_end_cert(self, subject_key: PrivateKey, subject: Subject, issuer_cert: Certificate,
                        issuer_key: PrivateKey,
                        extensions: str = "", days: int = 30, usages=None):
        """
        Creates an end certificate issued by the given parent CA.
        :param subject_key: Private key to be used when generating the certificate
        :param subject: Subject to be used put into the certificate
        :param issuer_cert: Issuer certificate that will sign the CSR required to create an intermediate CA
        :param issuer_key: PK used to generate the issuer certificate
        :param extensions: Extensions to be used when generating the certificate
        :param usages: A dictionary containing usages of the certificate to be generated (see constants.py)
        :param days: Number of days for which the generated cert. will be considered valid
        :return: An instance of Certificate class representing the generated cert
        """
        if usages is None:
            usages = {}

        # generate a new certificate
        cert_pem = self.cryptography_service.create_crt(subject, subject_key.private_key, issuer_cert.pem_data,
                                                        issuer_key.private_key,
                                                        subject_key_pass=subject_key.password,
                                                        issuer_key_pass=issuer_key.password, extensions=extensions,
                                                        days=days)

        # wrap the generated certificate using Certificate class
        certificate = self.__create_wrapper(cert_pem, subject_key.private_key_id, usages,
                                            issuer_cert.certificate_id, CERTIFICATE_ID)

        created_id = self.certificate_repository.create(certificate)

        certificate.certificate_id = created_id

        return certificate

    def get_certificate(self, unique_id: int) -> Certificate:
        """
        Tries to fetch a certificate from the certificate repository using a given id.
        :param unique_id: ID of the certificate to be fetched
        :return: Instance of the Certificate class containing a certificate with the given id or `None` if such
        certificate is not found
        """
        return self.certificate_repository.read(unique_id)

    def get_certificates(self, cert_type=None) -> List[Certificate]:
        """
        Tries to fetch a list of all certificates from the certificate repository. Using the `cert_type` parameter only
        certificates of the given type can be returned.
        :param cert_type: Type of certificates to be returned
        :return: List of instances of the Certificate class representing all certificates present in the certificate
        repository. An empty list is returned when no certificates are found.
        """
        return self.certificate_repository.read_all(cert_type)

    def get_chain_of_trust(self, from_id: int, to_id: int = -1, exclude_root=True) -> List[Certificate]:
        """
        Traverses the certificate hierarchy tree upwards till a certificate with the `to_id` ID is found or till a
        root CA certificate is found. Root certificates are excluded from the chain by default.
        :param from_id: ID of the first certificate to be included in the chain of trust
        :param to_id: ID of the last certificate to be included in the chain of trust
        :param exclude_root: a flag indicating whether root CA certificate should be excluded
        :return: a list of certificates representing the chain of trust starting with the certificate given by `from_id`
        ID
        """
        # read the first certificate of the chain
        start_cert = self.certificate_repository.read(from_id)

        # if no cert is found or the current cert is root CA and root CAs should be excluded, then return an empty list
        if start_cert is None or (start_cert.type_id == ROOT_CA_ID and exclude_root):
            return []

        current_cert = start_cert
        chain_of_trust = [current_cert]

        # TODO could possibly be simplified
        if start_cert.type_id == ROOT_CA_ID:
            # the first cert found is a root ca
            return chain_of_trust

        while True:
            parent_cert = self.certificate_repository.read(current_cert.parent_id)

            # check whether parent certificate exists
            if parent_cert is None:
                break

            # check whether the found certificate is a root certificate
            if parent_cert.type_id == ROOT_CA_ID:
                if not exclude_root:
                    # append the found root cert only if root certificates should not be excluded from the CoT
                    chain_of_trust.append(parent_cert)
                break

            # append the certificate
            chain_of_trust.append(parent_cert)

            # stop iterating over certificates if the id of the found certificate matches `to_id` method parameter
            if parent_cert.certificate_id == to_id:
                break

            current_cert = parent_cert

        return chain_of_trust

    def delete_certificate(self, unique_id) -> bool:
        """
        Deletes a certificate

        :param unique_id: ID of specific certificate

        :return: `True` when the deletion was successful. `False` in other case
        """
        # TODO delete children?
        return self.certificate_repository.delete(unique_id)

    def set_certificate_revocation_status(self, id, status, reason="unspecified"):
        """
        Set certificate status to 'valid' or 'revoked'.
        If the new status is revoked a reason can be provided -> default is unspecified
        :param reason: reason for revocation
        :param id: identifier of the certificate whose status is to be changed
        :param status: new status of the certificate
        """
        if status not in CERTIFICATE_STATES:
            raise CertificateStatusInvalidException(status)
        if reason not in CERTIFICATE_REVOCATION_REASONS:
            raise RevocationReasonInvalidException(reason)

        updated = False
        if status == STATUS_VALID:
            updated = self.certificate_repository.clear_certificate_revocation(id)
        elif status == STATUS_REVOKED:
            revocation_timestamp = int(time.time())
            updated = self.certificate_repository.set_certificate_revoked(id, str(revocation_timestamp), reason)

        if not updated:
            raise CertificateNotFoundException(id)

    def get_subject_from_certificate(self, certificate: Certificate) -> Subject:
        """
        Get Subject distinguished name from a Certificate
        :param certificate: certificate instance whose Subject shall be parsed
        :return: instance of Subject class containing resulting distinguished name
        """
        (subject, _, _) = self.cryptography_service.parse_cert_pem(certificate.pem_data)
        return subject

    def get_public_key_from_certificate(self, certificate: Certificate):
        """
        Extracts a public key from the given certificate
        :param certificate: an instance of the Certificate class containing the certificate from which a public key
        should be extracted.
        :return: a string containing the extracted public key in PEM format
        """
        return self.cryptography_service.extract_public_key_from_certificate(certificate.pem_data)

    def __get_crl_endpoint(self, ca_identifier: int) -> str:
        """
        Get URL address of CRL distribution endpoint based on
        issuer's ID

        :param ca_identifier: ID of issuing authority
        :return: CRL endpoint for the given CA
        """
        return self.configuration.base_server_url + "/api/crl/" + str(ca_identifier)

    def __get_ocsp_endpoint(self, ca_identifier: int) -> str:
        """
        Get URL address of OCSP distribution endpoint based on
        issuer's ID

        :param ca_identifier: ID of issuing authority
        :return: OCSP endpoint for the given CA
        """
        return self.configuration.base_server_url + "/api/ocsp/" + str(ca_identifier)


class RevocationReasonInvalidException(Exception):
    """
    Exception that denotes that the caller was trying to revoke
    a certificate using an invalid revocation reason
    """

    def __init__(self, reason):
        self.reason = reason

    def __str__(self):
        return f"Revocation reason '{self.reason}' is not valid."


class CertificateStatusInvalidException(Exception):
    """
    Exception that denotes that the caller was trying to set
    a certificate to an invalid state
    """

    def __init__(self, status):
        self.status = status

    def __str__(self):
        return f"Certificate status '{self.status}' is not valid."


class CertificateNotFoundException(Exception):
    """
    Exception that denotes that the caller was trying to set
    a certificate to an invalid state
    """

    def __init__(self, id):
        self.id = id

    def __str__(self):
        return f"Certificate id '{self.id}' does not exist."