aswi2021jmsd-gitlab/src/services/certificate_service.py @ 2a90f4fd

from src.constants import ROOT_CA_ID, INTERMEDIATE_CA_ID, CA_ID, CERTIFICATE_ID
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"


class CertificateService:

    def __init__(self, cryptography_service: CryptographyService, certificate_repository: CertificateRepository):
        self.cryptography_service = cryptography_service
        self.certificate_repository = certificate_repository

    # 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):
        if usages is None:
            usages = {}

        # 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)
        # specify CA usage
        usages[CA_ID] = True

        # wrap into Certificate class
        certificate = self.__create_wrapper(cert_pem, key.private_key_id, subject.common_name, 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, common_name, usages, parent_id, cert_type):
        # 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)

        # create a certificate wrapper
        certificate = Certificate(-1, 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):
        if usages is None:
            usages = {}

        extensions = extensions + "\n" + CA_EXTENSIONS
        # 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, subject.common_name, 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):
        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, subject.common_name, 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:
        return self.certificate_repository.read(unique_id)

    def get_certificates(self, cert_type=None) -> Certificate:
        return self.certificate_repository.read_all(cert_type)