aswi2021jmsd-gitlab/src/dao/certificate_repository.py @ f8b23532

from typing import Dict
from sqlite3 import Connection, Cursor, Error

from src.db_objects.certificate import Certificate
from src.constants import *


class CertificateRepository:

    def __init__(self, connection: Connection, cursor: Cursor):
        """
        Constructor of the CertificateRepository object

        :param connection: Instance of the Connection object
        :param cursor: Instance of the Cursor object
        """

        self.connection = connection
        self.cursor = cursor

    def create(self, certificate: Certificate) -> bool:
        """
        Creates a certificate.
        For root certificate (CA) the parent certificate id is modified to the same id (id == parent_id).

        :param certificate: Instance of the Certificate object

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

        try:
            sql = (f"INSERT INTO {TAB_CERTIFICATES} "
                   f"({COL_COMMON_NAME},"
                   f"{COL_VALID_FROM},"
                   f"{COL_VALID_TO},"
                   f"{COL_PEM_DATA},"
                   f"{COL_PRIVATE_KEY_ID},"
                   f"{COL_TYPE_ID},"
                   f"{COL_PARENT_ID})"
                   f"VALUES(?,?,?,?,?,?,?)")
            values = [certificate.common_name,
                      certificate.valid_from,
                      certificate.valid_to,
                      certificate.pem_data,
                      certificate.private_key_id,
                      certificate.type_id,
                      certificate.parent_id]
            self.cursor.execute(sql, values)
            self.connection.commit()

            last_id: int = self.cursor.lastrowid

            if certificate.usages[ROOT_CA_ID - 1]:
                certificate.parent_id = last_id
                return self.update(last_id, certificate)
            else:
                for usage_id, usage_value in certificate.usages:
                    if usage_value:
                        sql = (f"INSERT INTO {TAB_CERTIFICATE_USAGES} "
                               f"({COL_CERTIFICATE_ID},"
                               f"{COL_USAGE_TYPE_ID}) "
                               f"VALUES (?,?)")
                        values = [last_id, usage_id]
                        self.cursor.execute(sql, values)
                        self.connection.commit()
        except Error as e:
            print(e)
            return False

        return True

    def read(self, certificate_id: int) -> Certificate:
        """
        Reads (selects) a certificate.

        :param certificate_id: ID of specific certificate

        :return: instance of the Certificate object
        """

        try:
            sql = (f"SELECT * FROM {TAB_CERTIFICATES} "
                   f"WHERE {COL_ID} = ?")
            values = [certificate_id]
            self.cursor.execute(sql, values)
            certificate_row = self.cursor.fetchall()

            sql = (f"SELECT * FROM {TAB_CERTIFICATE_USAGES} "
                   f"WHERE {COL_CERTIFICATE_ID} = ?")
            self.cursor.execute(sql, values)
            usage_rows = self.cursor.fetchall()

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

            certificate: Certificate = 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)
        except Error as e:
            print(e)
            return None

        return certificate

    def update(self, certificate_id: int, certificate: Certificate) -> bool:
        """
        Updates a certificate.
        If the parameter of certificate (Certificate object) is not to be changed,
        the same value must be specified.

        :param certificate_id: ID of specific certificate
        :param certificate: Instance of the Certificate object

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

        try:
            sql = (f"UPDATE {TAB_CERTIFICATES} "
                   f"SET {COL_COMMON_NAME} = ?, "
                   f"{COL_VALID_FROM} = ?, "
                   f"{COL_VALID_TO} = ?, "
                   f"{COL_PEM_DATA} = ?, "
                   f"{COL_PRIVATE_KEY_ID} = ?, "
                   f"{COL_TYPE_ID} = ?, "
                   f"{COL_PARENT_ID} = ? "
                   f"WHERE {COL_ID} = ?")
            values = [certificate.common_name,
                      certificate.valid_from,
                      certificate.valid_to,
                      certificate.pem_data,
                      certificate.private_key_id,
                      certificate.type_id,
                      certificate.parent_id]
            self.cursor.execute(sql, values)
            self.connection.commit()

            sql = (f"DELETE FROM {TAB_CERTIFICATE_USAGES} "
                   f"WHERE {COL_CERTIFICATE_ID} = ?")
            values = [certificate_id]
            self.cursor.execute(sql, values)
            self.connection.commit()

            for usage_id, usage_value in certificate.usages:
                if usage_value:
                    sql = (f"INSERT INTO {TAB_CERTIFICATE_USAGES} "
                           f"({COL_CERTIFICATE_ID},"
                           f"{COL_USAGE_TYPE_ID}) "
                           f"VALUES (?,?)")
                    values = [certificate_id, usage_id]
                    self.cursor.execute(sql, values)
                    self.connection.commit()
        except Error as e:
            print(e)
            return False

        return True

    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
        """

        try:
            sql = (f"DELETE FROM {TAB_CERTIFICATES} "
                   f"WHERE {COL_ID} = ?")
            values = [certificate_id]
            self.cursor.execute(sql, values)
            self.connection.commit()
        except Error as e:
            print(e)
            return False

        return True