aswi2021jmsd-gitlab4/src/impl/certificate_repository_impl.py @ a0602bad

from typing import List, Dict

from ..db_objects.certificate import Certificate
from ..dao.repository import IRepository
from db_manager import DBManager
from ..constants import *


class CertificateRepositoryImpl(IRepository):

    def create(self, common_name: str, valid_from: str, valid_to: str, pem_data: str,
               private_key_id: int, type_id: int, parent_id: int, usages: Dict[int, bool]) -> bool:
        """
        Creates a certificate

        :param common_name: private key of the certificate
        :param valid_from: string of date (time) of validation after (from)
        :param valid_to: string of date (time) of validation before (to)
        :param pem_data: data of certificate in a PEM format
        :param private_key_id: ID from PrivateKeys table for specific private key
        :param type_id: ID from CertificateTypes table (ROOT_CA_ID=1, INTERMEDIATE_CA_ID=2, CERTIFICATE_ID=3)
        :param parent_id: ID from Certificates table for specific parent certificate
            for root CA use e.g. -1 (ID will be changed after created certificate)
        :param usages: dictionary of usages IDs which certificate use
            Dictionary[Integer, Boolean]
                - Key = ID (CA_ID=1, SSL_ID=2, SIGNATURE_ID=3, AUTHENTICATION_ID=4)
                - Value = used=True/unused=False

        :return: the result of whether the creation was successful
        """

        sql = (f"INSERT INTO {TAB_CERTIFICATES} ({COL_COMMON_NAME},{COL_VALID_FROM},{COL_VALID_TO},{COL_PEM_DATA},"
               f"{COL_PRIVATE_KEY_ID},{COL_TYPE_ID},{COL_PARENT_ID})"
               f"VALUES(?,?,?,?,?,?,?)")
        result = DBManager.create(sql, [common_name, valid_from, valid_to, pem_data,
                                        private_key_id, type_id, parent_id], usages)
        return result

    def read(self, certificate_id: int = None):
        """
        Reads (selects) a certificate or certificates

        :param certificate_id: ID of specific certificate

        :return: instance of Certificate class if ID was used otherwise list of this instances
        """

        sql_extend = ""

        if certificate_id is not None:
            sql_extend = f" WHERE {COL_ID} = ?"

        sql = f"SELECT * FROM {TAB_CERTIFICATES}{sql_extend}"
        certificate_rows = DBManager.read(sql, certificate_id)

        certificates: list = []

        for certificate_row in certificate_rows:
            sql = f"SELECT * FROM {TAB_CERTIFICATE_USAGES} WHERE {COL_CERTIFICATE_ID} = ?"
            usage_rows = DBManager.read(sql, certificate_row[0])

            usage_dict: Dict[int, bool] = {}
            for usage_row in usage_rows:
                usage_dict[usage_row[1]] = True

            certificates.append(Certificate(certificate_row[0],
                                            certificate_row[1],
                                            certificate_row[2],
                                            certificate_row[3],
                                            certificate_row[4],
                                            certificate_row[5],
                                            certificate_row[6],
                                            certificate_row[7],
                                            usage_dict))

        if certificate_id is not None:
            return certificates[0]

        return certificates

    def update(self, certificate_id: int, common_name: str = None, valid_from: str = None, valid_to: str = None,
               pem_data: str = None, private_key_id: int = None, type_id: int = None, parent_id: int = None,
               usages: Dict[int, bool] = None) -> bool:
        """
        Updates a certificate

        :param certificate_id: ID of specific certificate
        :param common_name: private key of the certificate
        :param valid_from: string of date (time) of validation after (from)
        :param valid_to: string of date (time) of validation before (to)
        :param pem_data: data of certificate in a PEM format
        :param private_key_id: ID from PrivateKeys table for specific private key
        :param type_id: ID from CertificateTypes table (ROOT_CA_ID=1, INTERMEDIATE_CA_ID=2, CERTIFICATE_ID=3)
        :param parent_id: ID from Certificates table for specific parent certificate
            for root CA use e.g. -1 (ID will be changed after created certificate)
        :param usages: dictionary of usages IDs which certificate use
            Dictionary[Integer, Boolean]
                - Key = ID (CA_ID=1, SSL_ID=2, SIGNATURE_ID=3, AUTHENTICATION_ID=4)
                - Value = used=True/unused=False

        :return: the result of whether the updation was successful
        """

        updated_list = []
        values = []
        if common_name is not None:
            updated_list.append(f"{COL_COMMON_NAME} = ?")
            values.append(common_name)
        if valid_from is not None:
            updated_list.append(f"{COL_VALID_FROM} = ?")
            values.append(valid_from)
        if valid_to is not None:
            updated_list.append(f"{COL_VALID_TO} = ?")
            values.append(valid_to)
        if pem_data is not None:
            updated_list.append(f"{COL_PEM_DATA} = ?")
            values.append(pem_data)
        if private_key_id is not None:
            updated_list.append(f"{COL_PRIVATE_KEY_ID} = ?")
            values.append(private_key_id)
        if type_id is not None:
            updated_list.append(f"{COL_TYPE_ID} = ?")
            values.append(type_id)
        if parent_id is not None:
            updated_list.append(f"{COL_PARENT_ID} = ?")
            values.append(parent_id)

        updated_str = ", ".join(updated_list)
        sql = f"UPDATE {TAB_CERTIFICATES} SET {updated_str} WHERE {COL_ID} = ?"
        result = DBManager.update(sql, certificate_id, values, usages)
        return result

    def delete(self, certificate_id: int) -> bool:
        """
        Deletes a certificate

        :param certificate_id: ID of specific certificate

        :return: the result of whether the deletion was successful
        """

        sql = f"DELETE FROM {TAB_CERTIFICATES} WHERE {COL_ID} = ?"
        result = DBManager.delete(sql, certificate_id, True)
        return result