aswi2021jmsd-gitlab4/src/services/crl/crl_service.py @ 7f9b2c58

from datetime import datetime

from injector import inject

from src.dao.certificate_repository import CertificateRepository
from src.dao.private_key_repository import PrivateKeyRepository
from src.exceptions.certificate_not_found_exception import CertificateNotFoundException
from src.exceptions.private_key_not_found_exception import PrivateKeyNotFoundException
from src.services.crl.ca_index_file_line_generator import create_index_file_revoked_line, create_index_file_valid_line
from src.services.cryptography import CryptographyService
from src.utils.temporary_file import TemporaryFile


class CrlService:
    @inject
    def __init__(self,
                 certificate_repository: CertificateRepository,
                 key_repository: PrivateKeyRepository,
                 cryptography_service: CryptographyService
                 ):
        self.key_repository = key_repository
        self.certificate_repository = certificate_repository
        self.cryptography_service = cryptography_service

    def create_revoked_index(self, ca_id) -> str:
        """
        Queries the certificate repository and looks for all certificates revoked by the certificate authority given
        by the passed ID. Found certificates are then put into a string representing the CA's database index file.
        
        :param ca_id: ID of the CA whose revoked certificates should be put into the index file
        :return: a str representing the content of a CA index file
        """
        # get issuing certificate
        certificate = self.certificate_repository.read(ca_id)
        if certificate is None:
            raise CertificateNotFoundException(ca_id)

        # get subject and notAfter of the issuer
        subject, _, not_after = self.cryptography_service.parse_cert_pem(certificate.pem_data)

        index_lines = [create_index_file_valid_line(certificate, subject, not_after)]
        # iterate over revoked certificates of the CA given by an ID
        for certificate in self.certificate_repository.get_all_revoked_by(ca_id):
            # extract the complete subject information and not_after date field
            subject, _, not_after = self.cryptography_service.parse_cert_pem(certificate.pem_data)

            line = create_index_file_revoked_line(certificate,
                                                  subject,
                                                  # parse revocation date from unix timestamp to struct_time
                                                  datetime.utcfromtimestamp(int(certificate.revocation_date)).timetuple(),
                                                  not_after)

            # append it to the list of lines
            index_lines.append(line)

        # join all lines with a new line
        return "\n".join(index_lines)

    def create_index(self, ca_id) -> str:
        """
        Queries the certificate repository and looks for all certificates issued by the certificate authority given
        by the passed ID. Found certificates are then put into a string representing the CA's database index file.

        :param ca_id: ID of the CA whose issued (=child) certificates should be put into the index file
        :return: a str representing the content of a CA index file
        """
        # get issuing certificate
        certificate = self.certificate_repository.read(ca_id)
        if certificate is None:
            raise CertificateNotFoundException(ca_id)

        # get subject and notAfter of the issuer
        subject, _, not_after = self.cryptography_service.parse_cert_pem(certificate.pem_data)

        index_lines = [create_index_file_valid_line(certificate, subject, not_after)]
        # iterate over revoked certificates of the CA given by an ID
        for certificate in self.certificate_repository.get_all_issued_by(ca_id):
            # extract the complete subject information and not_after date field
            subject, _, not_after = self.cryptography_service.parse_cert_pem(certificate.pem_data)
            if len(certificate.revocation_reason) > 0:
                line = create_index_file_revoked_line(certificate,
                                                      subject,
                                                      # parse revocation date from unix timestamp to struct_time
                                                      datetime.utcfromtimestamp(
                                                          int(certificate.revocation_date)).timetuple(),
                                                      not_after)
            else:
                line = create_index_file_valid_line(certificate, subject, not_after)

            # append it to the list of lines
            index_lines.append(line)

        # join all lines with a new line
        return "\n".join(index_lines)

    def generate_crl_response(self, ca_id: int) -> str:
        """
        Generate a CRL for the given certificate authority
        that contains all revoked certificates

        :param ca_id: ID of a CA whose CRL shall be generated
        :return: CRL in PEM format
        """
        # get cert and check if the requested CA exists and if not throw an exception
        cert = self.certificate_repository.read(ca_id)
        if cert is None:
            raise CertificateNotFoundException(ca_id)

        # get key and check if it exists
        key = self.key_repository.read(cert.private_key_id)
        if key is None:
            raise PrivateKeyNotFoundException(ca_id)

        # Create an index file and call cryptography service to generate CRL
        with TemporaryFile("crl.index", f"{self.create_revoked_index(ca_id)}\n") as index_path:
            crl_content = self.cryptography_service.generate_crl(cert, key, index_path)

        return crl_content

    def generate_ocsp_response(self, ca_id: int, der_ocsp_request: bytes):
        # get cert and check if the requested CA exists and if not throw an exception
        cert = self.certificate_repository.read(ca_id)
        if cert is None:
            raise CertificateNotFoundException(ca_id)

        # get key and check if it exists
        key = self.key_repository.read(cert.private_key_id)
        if key is None:
            raise PrivateKeyNotFoundException(ca_id)

        # Create an index file and call cryptography service to generate the OCSP response
        with TemporaryFile("crl.index", f"{self.create_index(ca_id)}\n") as index_path:
            ocsp_content = self.cryptography_service.generate_ocsp(cert, key, index_path, der_ocsp_request)

        return ocsp_content