/src/dao/certificate_repository.py - Komentovat - Aplikace pro správu X.509 certifikátů (Yoso Czech s.r.o.) - JMSD - Redmine

| Větev: | Tag: | Revize:

aswi2021jmsd-gitlab4/src/dao/certificate_repository.py @ cf247eaa

-fa36
+import time
-            cf247eaa
+from sqlite3 import Connection, Error, DatabaseError, IntegrityError, ProgrammingError, OperationalError, \
     NotSupportedError
 from typing import Dict, List, Set
-aefe
+            David Friesecký
-            d65b022d
+from src.exceptions.database_exception import DatabaseException
-d2add74
+from injector import inject
-            f8b23532
+from src.constants import *
-d2add74
+from src.model.certificate import Certificate
-e31b492
+from src.utils.logger import Logger
-            e9e55282
+            David Friesecký
-            b3c80ccb
+INTEGRITY_ERROR_MSG = "Database relational integrity corrupted."
 PROGRAMMING_ERROR_MSG = "Exception raised for programming errors (etc. SQL statement)."
 OPERATIONAL_ERROR_MSG = "Exception raised for errors that are related to the database’s operation."
 NOT_SUPPORTED_ERROR_MSG = "Method or database API was used which is not supported by the database"
 DATABASE_ERROR_MSG = "Unknown exception that are related to the database."
 ERROR_MSG = "Unknown exception."
-            e9e55282
+            David Friesecký
-            f8b23532
+class CertificateRepository:
-            25053504
+            David Friesecký
-d2add74
+    @inject
     def __init__(self, connection: Connection):
-            a0602bad
+        """
-            f8b23532
+        Constructor of the CertificateRepository object
-            a0602bad
+            David Friesecký
-            f8b23532
+        :param connection: Instance of the Connection object
-            a0602bad
+        """
-            f8b23532
+        self.connection = connection
-d2add74
+        self.cursor = connection.cursor()
-            e9e55282
+            David Friesecký
-f5
+    def create(self, certificate: Certificate):
-            a0602bad
+        """
-            f8b23532
+        Creates a certificate.
         For root certificate (CA) the parent certificate id is modified to the same id (id == parent_id).
-            a0602bad
+            David Friesecký
-            f8b23532
+        :param certificate: Instance of the Certificate object
-            a0602bad
+            David Friesecký
-            f8b23532
+        :return: the result of whether the creation was successful
-            a0602bad
+        """
-            b3c80ccb
+        Logger.debug("Function launched.")
-            f8b23532
+        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
-            f3125948
+            # TODO assure that this is correct
             if certificate.type_id == ROOT_CA_ID:
-            f8b23532
+                certificate.parent_id = last_id
-d06df
+                self.update(last_id, certificate)
-            f8b23532
+            else:
-            fa72c969
+                for usage_id, usage_value in certificate.usages.items():
-            f8b23532
+                    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()
-            b3c80ccb
+        except IntegrityError:
             Logger.error(INTEGRITY_ERROR_MSG)
             raise DatabaseException(INTEGRITY_ERROR_MSG)
         except ProgrammingError:
             Logger.error(PROGRAMMING_ERROR_MSG)
             raise DatabaseException(PROGRAMMING_ERROR_MSG)
         except OperationalError:
             Logger.error(OPERATIONAL_ERROR_MSG)
             raise DatabaseException(OPERATIONAL_ERROR_MSG)
         except NotSupportedError:
             Logger.error(NOT_SUPPORTED_ERROR_MSG)
             raise DatabaseException(NOT_SUPPORTED_ERROR_MSG)
         except DatabaseError:
             Logger.error(DATABASE_ERROR_MSG)
             raise DatabaseException(DATABASE_ERROR_MSG)
         except Error:
             Logger.error(ERROR_MSG)
             raise DatabaseException(ERROR_MSG)
-            f8b23532
+            David Friesecký
-f5
+        return last_id
-            f8b23532
+            David Friesecký
-f5
+    def read(self, certificate_id: int):
-            f8b23532
+        """
         Reads (selects) a certificate.
-            e9e55282
+            David Friesecký
-            f8b23532
+        :param certificate_id: ID of specific certificate
-            e9e55282
+            David Friesecký
-            f8b23532
+        :return: instance of the Certificate object
         """
-            e9e55282
+            David Friesecký
-            b3c80ccb
+        Logger.debug("Function launched.")
-            f8b23532
+        try:
             sql = (f"SELECT * FROM {TAB_CERTIFICATES} "
-fa36
+                   f"WHERE {COL_ID} = ? AND ({COL_DELETION_DATE} IS NULL OR {COL_DELETION_DATE} = '')")
-            f8b23532
+            values = [certificate_id]
             self.cursor.execute(sql, values)
-            a7411982
+            certificate_row = self.cursor.fetchone()
-            e9e55282
+            David Friesecký
-f4a5f24
+            if certificate_row is None:
                 return None
-            f8b23532
+            sql = (f"SELECT * FROM {TAB_CERTIFICATE_USAGES} "
                    f"WHERE {COL_CERTIFICATE_ID} = ?")
             self.cursor.execute(sql, values)
             usage_rows = self.cursor.fetchall()
-aefe
+            David Friesecký
             usage_dict: Dict[int, bool] = {}
             for usage_row in usage_rows:
-            f8b23532
+                usage_dict[usage_row[2]] = True
             certificate: Certificate = Certificate(certificate_row[0],
                                                    certificate_row[1],
                                                    certificate_row[2],
                                                    certificate_row[3],
                                                    certificate_row[4],
-            58051326
+                                                   certificate_row[8],
                                                    certificate_row[9],
-fa36
+                                                   certificate_row[10],
-            58051326
+                                                   usage_dict,
                                                    certificate_row[5],
                                                    certificate_row[6])
-            b3c80ccb
+        except IntegrityError:
             Logger.error(INTEGRITY_ERROR_MSG)
             raise DatabaseException(INTEGRITY_ERROR_MSG)
         except ProgrammingError:
             Logger.error(PROGRAMMING_ERROR_MSG)
             raise DatabaseException(PROGRAMMING_ERROR_MSG)
         except OperationalError:
             Logger.error(OPERATIONAL_ERROR_MSG)
             raise DatabaseException(OPERATIONAL_ERROR_MSG)
         except NotSupportedError:
             Logger.error(NOT_SUPPORTED_ERROR_MSG)
             raise DatabaseException(NOT_SUPPORTED_ERROR_MSG)
         except DatabaseError:
             Logger.error(DATABASE_ERROR_MSG)
             raise DatabaseException(DATABASE_ERROR_MSG)
         except Error:
             Logger.error(ERROR_MSG)
             raise DatabaseException(ERROR_MSG)
-            f8b23532
+            David Friesecký
-            d65b022d
+        return certificate
-            f8b23532
+            David Friesecký
-            58051326
+    def read_all(self, filter_type: int = None):
-e22e20c
+        """
         Reads (selects) all certificates (with type).
         :param filter_type: ID of certificate type from CertificateTypes table
         :return: list of certificates
         """
-            b3c80ccb
+        Logger.debug("Function launched.")
-e22e20c
+        try:
             sql_extension = ""
             values = []
             if filter_type is not None:
-fa36
+                sql_extension = (f" AND {COL_TYPE_ID} = ("
-f64f062
+                                 f"SELECT {COL_ID} FROM {TAB_CERTIFICATE_TYPES} WHERE {COL_ID} = ?)")
-e22e20c
+                values = [filter_type]
-fa36
+            sql = (f"SELECT * FROM {TAB_CERTIFICATES} "
                    f"WHERE ({COL_DELETION_DATE} IS NULL OR {COL_DELETION_DATE} = ''){sql_extension}")
-e22e20c
+            self.cursor.execute(sql, values)
             certificate_rows = self.cursor.fetchall()
             certificates: List[Certificate] = []
             for certificate_row in certificate_rows:
                 sql = (f"SELECT * FROM {TAB_CERTIFICATE_USAGES} "
                        f"WHERE {COL_CERTIFICATE_ID} = ?")
                 values = [certificate_row[0]]
                 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
                 certificates.append(Certificate(certificate_row[0],
                                                 certificate_row[1],
                                                 certificate_row[2],
                                                 certificate_row[3],
                                                 certificate_row[4],
-            58051326
+                                                certificate_row[8],
                                                 certificate_row[9],
-fa36
+                                                certificate_row[10],
-            58051326
+                                                usage_dict,
                                                 certificate_row[5],
                                                 certificate_row[6]))
-            b3c80ccb
+        except IntegrityError:
             Logger.error(INTEGRITY_ERROR_MSG)
             raise DatabaseException(INTEGRITY_ERROR_MSG)
         except ProgrammingError:
             Logger.error(PROGRAMMING_ERROR_MSG)
             raise DatabaseException(PROGRAMMING_ERROR_MSG)
         except OperationalError:
             Logger.error(OPERATIONAL_ERROR_MSG)
             raise DatabaseException(OPERATIONAL_ERROR_MSG)
         except NotSupportedError:
             Logger.error(NOT_SUPPORTED_ERROR_MSG)
             raise DatabaseException(NOT_SUPPORTED_ERROR_MSG)
         except DatabaseError:
             Logger.error(DATABASE_ERROR_MSG)
             raise DatabaseException(DATABASE_ERROR_MSG)
         except Error:
             Logger.error(ERROR_MSG)
             raise DatabaseException(ERROR_MSG)
-e22e20c
+            David Friesecký
-f3af523
+        return certificates
-e22e20c
+            David Friesecký
-            cf247eaa
+    def read_all_filter(self, target_types: Set[int], target_usages: Set[int], target_cn_substring: str, page: int,
                         per_page: int):
         """
         Reads (selects) all certificates according to a specific filtering and pagination options.
         :param target_types: certificate types (filter)
         :param target_usages: certificate usages (filter)
         :param target_cn_substring: certificate CN substring (filter)
         :param page: target page
         :param per_page: target page size
         :return: list of certificates
         """
         Logger.debug("Function launched.")
         try:
             values = []
             values += list(target_types)
             values += list(target_usages)
             sql = (
                 f"SELECT a.{COL_ID}, {COL_COMMON_NAME}, {COL_VALID_FROM}, {COL_VALID_TO}, {COL_PEM_DATA}, "
                 f"{COL_REVOCATION_DATE}, {COL_REVOCATION_REASON}, {COL_PRIVATE_KEY_ID}, {COL_TYPE_ID}, {COL_PARENT_ID} "
                 f"FROM {TAB_CERTIFICATES} a "
                 f"WHERE (a.{COL_DELETION_DATE} IS NULL OR a.{COL_DELETION_DATE} = '') "
                 f"AND a.{COL_TYPE_ID} IN ({','.join('?' * len(target_types))}) "
                 f"AND (SELECT COUNT(*) FROM {TAB_CERTIFICATE_USAGES} "
                 f"WHERE {COL_CERTIFICATE_ID} = a.{COL_ID} "
                 f"AND {COL_USAGE_TYPE_ID} IN ({','.join('?' * len(target_usages))}) "
                 f") > 0 "
+            )
             if target_cn_substring is not None:
                 sql += f" AND a.{COL_COMMON_NAME} LIKE '%' || ? || '%'"
                 values += [target_cn_substring]
             if page is not None and per_page is not None:
                 sql += f" LIMIT {per_page} OFFSET {page * per_page}"
             self.cursor.execute(sql, values)
             certificate_rows = self.cursor.fetchall()
             certificates: List[Certificate] = []
             for certificate_row in certificate_rows:
                 sql = (f"SELECT {COL_USAGE_TYPE_ID} FROM {TAB_CERTIFICATE_USAGES} "
                        f"WHERE {COL_CERTIFICATE_ID} = ?")
                 types = [certificate_row[0]]
                 self.cursor.execute(sql, types)
                 usage_rows = self.cursor.fetchall()
                 usage_dict: Dict[int, bool] = {}
                 for usage_row in usage_rows:
                     usage_dict[usage_row[0]] = True
                 certificates.append(Certificate(certificate_row[0],
                                                 certificate_row[1],
                                                 certificate_row[2],
                                                 certificate_row[3],
                                                 certificate_row[4],
                                                 certificate_row[7],
                                                 certificate_row[8],
                                                 certificate_row[9],
                                                 usage_dict,
                                                 certificate_row[5],
                                                 certificate_row[6]))
         except IntegrityError:
             Logger.error(INTEGRITY_ERROR_MSG)
             raise DatabaseException(INTEGRITY_ERROR_MSG)
         except ProgrammingError:
             Logger.error(PROGRAMMING_ERROR_MSG)
             raise DatabaseException(PROGRAMMING_ERROR_MSG)
         except OperationalError:
             Logger.error(OPERATIONAL_ERROR_MSG)
             raise DatabaseException(OPERATIONAL_ERROR_MSG)
         except NotSupportedError:
             Logger.error(NOT_SUPPORTED_ERROR_MSG)
             raise DatabaseException(NOT_SUPPORTED_ERROR_MSG)
         except DatabaseError:
             Logger.error(DATABASE_ERROR_MSG)
             raise DatabaseException(DATABASE_ERROR_MSG)
         except Error:
             Logger.error(ERROR_MSG)
             raise DatabaseException(ERROR_MSG)
         return certificates
-            58051326
+    def update(self, certificate_id: int, certificate: Certificate):
-            a0602bad
+        """
-            f8b23532
+        Updates a certificate.
         If the parameter of certificate (Certificate object) is not to be changed,
         the same value must be specified.
-            a0602bad
+            David Friesecký
         :param certificate_id: ID of specific certificate
-            f8b23532
+        :param certificate: Instance of the Certificate object
-            a0602bad
+            David Friesecký
         :return: the result of whether the updation was successful
         """
-            b3c80ccb
+        Logger.debug("Function launched.")
-            f8b23532
+        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,
-f5
+                      certificate.parent_id,
                       certificate_id]
-fa36
+            David Friesecký
-            f8b23532
+            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()
-            c36d3299
+            check_updated = self.cursor.rowcount > 0
-            f3125948
+            # iterate over usage pairs
             for usage_id, usage_value in certificate.usages.items():
-            f8b23532
+                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()
-            b3c80ccb
+        except IntegrityError:
             Logger.error(INTEGRITY_ERROR_MSG)
             raise DatabaseException(INTEGRITY_ERROR_MSG)
         except ProgrammingError:
             Logger.error(PROGRAMMING_ERROR_MSG)
             raise DatabaseException(PROGRAMMING_ERROR_MSG)
         except OperationalError:
             Logger.error(OPERATIONAL_ERROR_MSG)
             raise DatabaseException(OPERATIONAL_ERROR_MSG)
         except NotSupportedError:
             Logger.error(NOT_SUPPORTED_ERROR_MSG)
             raise DatabaseException(NOT_SUPPORTED_ERROR_MSG)
         except DatabaseError:
             Logger.error(DATABASE_ERROR_MSG)
             raise DatabaseException(DATABASE_ERROR_MSG)
         except Error:
             Logger.error(ERROR_MSG)
             raise DatabaseException(ERROR_MSG)
-            f8b23532
+            David Friesecký
-            c36d3299
+        return check_updated
-            e9e55282
+            David Friesecký
-            58051326
+    def delete(self, certificate_id: int):
-            a0602bad
+        """
         Deletes a certificate
         :param certificate_id: ID of specific certificate
         :return: the result of whether the deletion was successful
         """
-            b3c80ccb
+        Logger.debug("Function launched.")
-            f8b23532
+        try:
-fa36
+            sql = (f"UPDATE {TAB_CERTIFICATES} "
                    f"SET {COL_DELETION_DATE} = ? "
                    f"WHERE {COL_ID} = ? AND ({COL_DELETION_DATE} IS NULL OR {COL_DELETION_DATE} = '')")
             values = [int(time.time()),
                       certificate_id]
-            f8b23532
+            self.cursor.execute(sql, values)
             self.connection.commit()
-            b3c80ccb
+        except IntegrityError:
             Logger.error(INTEGRITY_ERROR_MSG)
             raise DatabaseException(INTEGRITY_ERROR_MSG)
         except ProgrammingError:
             Logger.error(PROGRAMMING_ERROR_MSG)
             raise DatabaseException(PROGRAMMING_ERROR_MSG)
         except OperationalError:
             Logger.error(OPERATIONAL_ERROR_MSG)
             raise DatabaseException(OPERATIONAL_ERROR_MSG)
         except NotSupportedError:
             Logger.error(NOT_SUPPORTED_ERROR_MSG)
             raise DatabaseException(NOT_SUPPORTED_ERROR_MSG)
         except DatabaseError:
             Logger.error(DATABASE_ERROR_MSG)
             raise DatabaseException(DATABASE_ERROR_MSG)
         except Error:
             Logger.error(ERROR_MSG)
             raise DatabaseException(ERROR_MSG)
-            f8b23532
+            David Friesecký
-            45744020
+        return self.cursor.rowcount > 0
-            58051326
+            David Friesecký
     def set_certificate_revoked(
-b049f43
+            self, certificate_id: int, revocation_date: str, revocation_reason: str = REV_REASON_UNSPECIFIED):
-            58051326
+        """
         Revoke a certificate
         :param certificate_id: ID of specific certificate
         :param revocation_date: Date, when the certificate is revoked
         :param revocation_reason: Reason of the revocation
-b049f43
+        :return:
             the result of whether the revocation was successful OR
             sqlite3.Error if an exception is thrown
-            58051326
+        """
-            b3c80ccb
+        Logger.debug("Function launched.")
-            58051326
+        try:
-b049f43
+            if revocation_date != "" and revocation_reason == "":
                 revocation_reason = REV_REASON_UNSPECIFIED
             elif revocation_date == "":
                 return False
-            58051326
+            sql = (f"UPDATE {TAB_CERTIFICATES} "
                    f"SET {COL_REVOCATION_DATE} = ?, "
-b049f43
+                   f"{COL_REVOCATION_REASON} = ? "
-            58051326
+                   f"WHERE {COL_ID} = ? AND ({COL_REVOCATION_DATE} IS NULL OR {COL_REVOCATION_DATE} = '')")
             values = [revocation_date,
                       revocation_reason,
                       certificate_id]
             self.cursor.execute(sql, values)
             self.connection.commit()
-            b3c80ccb
+        except IntegrityError:
             Logger.error(INTEGRITY_ERROR_MSG)
             raise DatabaseException(INTEGRITY_ERROR_MSG)
         except ProgrammingError:
             Logger.error(PROGRAMMING_ERROR_MSG)
             raise DatabaseException(PROGRAMMING_ERROR_MSG)
         except OperationalError:
             Logger.error(OPERATIONAL_ERROR_MSG)
             raise DatabaseException(OPERATIONAL_ERROR_MSG)
         except NotSupportedError:
             Logger.error(NOT_SUPPORTED_ERROR_MSG)
             raise DatabaseException(NOT_SUPPORTED_ERROR_MSG)
         except DatabaseError:
             Logger.error(DATABASE_ERROR_MSG)
             raise DatabaseException(DATABASE_ERROR_MSG)
         except Error:
             Logger.error(ERROR_MSG)
             raise DatabaseException(ERROR_MSG)
-b049f43
+            David Friesecký
-            d65b022d
+        return self.cursor.rowcount > 0
-            58051326
+            David Friesecký
     def clear_certificate_revocation(self, certificate_id: int):
         """
         Clear revocation of a certificate
         :param certificate_id: ID of specific certificate
-b049f43
+        :return:
             the result of whether the clear revocation was successful OR
             sqlite3.Error if an exception is thrown
-            58051326
+        """
-            b3c80ccb
+        Logger.debug("Function launched.")
-            58051326
+        try:
             sql = (f"UPDATE {TAB_CERTIFICATES} "
                    f"SET {COL_REVOCATION_DATE} = '', "
-b049f43
+                   f"{COL_REVOCATION_REASON} = '' "
-            58051326
+                   f"WHERE {COL_ID} = ?")
             values = [certificate_id]
             self.cursor.execute(sql, values)
             self.connection.commit()
-            b3c80ccb
+        except IntegrityError:
             Logger.error(INTEGRITY_ERROR_MSG)
             raise DatabaseException(INTEGRITY_ERROR_MSG)
         except ProgrammingError:
             Logger.error(PROGRAMMING_ERROR_MSG)
             raise DatabaseException(PROGRAMMING_ERROR_MSG)
         except OperationalError:
             Logger.error(OPERATIONAL_ERROR_MSG)
             raise DatabaseException(OPERATIONAL_ERROR_MSG)
         except NotSupportedError:
             Logger.error(NOT_SUPPORTED_ERROR_MSG)
             raise DatabaseException(NOT_SUPPORTED_ERROR_MSG)
         except DatabaseError:
             Logger.error(DATABASE_ERROR_MSG)
             raise DatabaseException(DATABASE_ERROR_MSG)
         except Error:
             Logger.error(ERROR_MSG)
             raise DatabaseException(ERROR_MSG)
-            f8b23532
+            David Friesecký
-            45744020
+        return self.cursor.rowcount > 0
-            58051326
+            David Friesecký
     def get_all_revoked_by(self, certificate_id: int):
         """
         Get list of the revoked certificates that are direct descendants of the certificate with the ID
         :param certificate_id: ID of specific certificate
         :return:
-b049f43
+            list of the certificates OR
             None if the list is empty OR
             sqlite3.Error if an exception is thrown
-            58051326
+        """
-            b3c80ccb
+        Logger.debug("Function launched.")
-            58051326
+        try:
             sql = (f"SELECT * FROM {TAB_CERTIFICATES} "
                    f"WHERE {COL_PARENT_ID} = ? AND {COL_REVOCATION_DATE} IS NOT NULL AND {COL_REVOCATION_DATE} != ''")
             values = [certificate_id]
             self.cursor.execute(sql, values)
             certificate_rows = self.cursor.fetchall()
             certificates: List[Certificate] = []
             for certificate_row in certificate_rows:
                 sql = (f"SELECT * FROM {TAB_CERTIFICATE_USAGES} "
                        f"WHERE {COL_CERTIFICATE_ID} = ?")
                 values = [certificate_row[0]]
                 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
                 certificates.append(Certificate(certificate_row[0],
                                                 certificate_row[1],
                                                 certificate_row[2],
                                                 certificate_row[3],
                                                 certificate_row[4],
                                                 certificate_row[8],
                                                 certificate_row[9],
-fa36
+                                                certificate_row[10],
-            58051326
+                                                usage_dict,
                                                 certificate_row[5],
                                                 certificate_row[6]))
-            b3c80ccb
+        except IntegrityError:
             Logger.error(INTEGRITY_ERROR_MSG)
             raise DatabaseException(INTEGRITY_ERROR_MSG)
         except ProgrammingError:
             Logger.error(PROGRAMMING_ERROR_MSG)
             raise DatabaseException(PROGRAMMING_ERROR_MSG)
         except OperationalError:
             Logger.error(OPERATIONAL_ERROR_MSG)
             raise DatabaseException(OPERATIONAL_ERROR_MSG)
         except NotSupportedError:
             Logger.error(NOT_SUPPORTED_ERROR_MSG)
             raise DatabaseException(NOT_SUPPORTED_ERROR_MSG)
         except DatabaseError:
             Logger.error(DATABASE_ERROR_MSG)
             raise DatabaseException(DATABASE_ERROR_MSG)
         except Error:
             Logger.error(ERROR_MSG)
             raise DatabaseException(ERROR_MSG)
-            58051326
+            David Friesecký
-            d65b022d
+        return certificates
-            58051326
+            David Friesecký
     def get_all_issued_by(self, certificate_id: int):
         """
         Get list of the certificates that are direct descendants of the certificate with the ID
         :param certificate_id: ID of specific certificate
         :return:
-b049f43
+            list of the certificates OR
             None if the list is empty OR
             sqlite3.Error if an exception is thrown
-            58051326
+        """
-            b3c80ccb
+        Logger.debug("Function launched.")
-            58051326
+        try:
             sql = (f"SELECT * FROM {TAB_CERTIFICATES} "
-fa36
+                   f"WHERE {COL_PARENT_ID} = ? AND {COL_ID} != ? AND "
                    f"({COL_DELETION_DATE} IS NULL OR {COL_DELETION_DATE} = '')")
-b049f43
+            values = [certificate_id, certificate_id]
-            58051326
+            self.cursor.execute(sql, values)
             certificate_rows = self.cursor.fetchall()
             certificates: List[Certificate] = []
             for certificate_row in certificate_rows:
                 sql = (f"SELECT * FROM {TAB_CERTIFICATE_USAGES} "
                        f"WHERE {COL_CERTIFICATE_ID} = ?")
                 values = [certificate_row[0]]
                 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
                 certificates.append(Certificate(certificate_row[0],
                                                 certificate_row[1],
                                                 certificate_row[2],
                                                 certificate_row[3],
                                                 certificate_row[4],
                                                 certificate_row[8],
                                                 certificate_row[9],
-fa36
+                                                certificate_row[10],
-            58051326
+                                                usage_dict,
                                                 certificate_row[5],
                                                 certificate_row[6]))
-            cf247eaa
+        except IntegrityError:
             Logger.error(INTEGRITY_ERROR_MSG)
             raise DatabaseException(INTEGRITY_ERROR_MSG)
         except ProgrammingError:
             Logger.error(PROGRAMMING_ERROR_MSG)
             raise DatabaseException(PROGRAMMING_ERROR_MSG)
         except OperationalError:
             Logger.error(OPERATIONAL_ERROR_MSG)
             raise DatabaseException(OPERATIONAL_ERROR_MSG)
         except NotSupportedError:
             Logger.error(NOT_SUPPORTED_ERROR_MSG)
             raise DatabaseException(NOT_SUPPORTED_ERROR_MSG)
         except DatabaseError:
             Logger.error(DATABASE_ERROR_MSG)
             raise DatabaseException(DATABASE_ERROR_MSG)
         except Error:
             Logger.error(ERROR_MSG)
             raise DatabaseException(ERROR_MSG)
         return certificates
     def get_all_issued_by_filter(self, issuer_id, target_types, target_usages, target_cn_substring, page, per_page):
         """
         Get list of the certificates that are direct descendants of the certificate with the ID
         :param target_types: certificate types (filter)
         :param target_usages: certificate usages (filter)
         :param target_cn_substring: certificate CN substring (filter)
         :param page: target page
         :param per_page: target page size
         :param issuer_id: ID of specific certificate
         :return:
             list of the certificates OR
             None if the list is empty
         """
         Logger.debug("Function launched.")
         try:
             values = [issuer_id, issuer_id]
             values += list(target_types)
             values += list(target_usages)
             sql = (
                 f"SELECT a.{COL_ID}, {COL_COMMON_NAME}, {COL_VALID_FROM}, {COL_VALID_TO}, {COL_PEM_DATA}, "
                 f"{COL_REVOCATION_DATE}, {COL_REVOCATION_REASON}, {COL_PRIVATE_KEY_ID}, {COL_TYPE_ID}, {COL_PARENT_ID} "
                 f"FROM {TAB_CERTIFICATES} a "
                 f"WHERE (a.{COL_DELETION_DATE} IS NULL OR a.{COL_DELETION_DATE} = '') "
                 f"AND a.{COL_PARENT_ID} = ? AND a.{COL_ID} != ? "
                 f"AND a.{COL_TYPE_ID} IN ({','.join('?' * len(target_types))}) "
                 f"AND (SELECT COUNT(*) FROM {TAB_CERTIFICATE_USAGES} "
                 f"WHERE {COL_CERTIFICATE_ID} = a.{COL_ID} "
                 f"AND {COL_USAGE_TYPE_ID} IN ({','.join('?' * len(target_usages))}) "
                 f") > 0 "
+            )
             if target_cn_substring is not None:
                 sql += f" AND a.{COL_COMMON_NAME} LIKE '%' || ? || '%'"
                 values += [target_cn_substring]
             if page is not None and per_page is not None:
                 sql += f" LIMIT {per_page} OFFSET {page * per_page}"
             self.cursor.execute(sql, values)
             certificate_rows = self.cursor.fetchall()
             certificates: List[Certificate] = []
             for certificate_row in certificate_rows:
                 sql = (f"SELECT * FROM {TAB_CERTIFICATE_USAGES} "
                        f"WHERE {COL_CERTIFICATE_ID} = ?")
                 values = [certificate_row[0]]
                 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
                 certificates.append(Certificate(certificate_row[0],
                                                 certificate_row[1],
                                                 certificate_row[2],
                                                 certificate_row[3],
                                                 certificate_row[4],
                                                 certificate_row[7],
                                                 certificate_row[8],
                                                 certificate_row[9],
                                                 usage_dict,
                                                 certificate_row[5],
                                                 certificate_row[6]))
-            b3c80ccb
+        except IntegrityError:
             Logger.error(INTEGRITY_ERROR_MSG)
             raise DatabaseException(INTEGRITY_ERROR_MSG)
         except ProgrammingError:
             Logger.error(PROGRAMMING_ERROR_MSG)
             raise DatabaseException(PROGRAMMING_ERROR_MSG)
         except OperationalError:
             Logger.error(OPERATIONAL_ERROR_MSG)
             raise DatabaseException(OPERATIONAL_ERROR_MSG)
         except NotSupportedError:
             Logger.error(NOT_SUPPORTED_ERROR_MSG)
             raise DatabaseException(NOT_SUPPORTED_ERROR_MSG)
         except DatabaseError:
             Logger.error(DATABASE_ERROR_MSG)
             raise DatabaseException(DATABASE_ERROR_MSG)
         except Error:
             Logger.error(ERROR_MSG)
             raise DatabaseException(ERROR_MSG)
-            58051326
+            David Friesecký
-            d65b022d
+        return certificates
-f6d8b8
+            Jan Pašek
-            85003184
+    def get_all_descendants_of(self, certificate_id: int):
         """
         Get a list of all certificates C such that the certificate identified by "certificate_id" belongs to a trust chain
         between C and its root certificate authority (i.e. is an ancestor of C).
         :param certificate_id: target certificate ID
         :return: list of all descendants
         """
-            b3c80ccb
+        Logger.debug("Function launched.")
         try:
             def dfs(children_of, this, collection: list):
                 for child in children_of(this.certificate_id):
                     dfs(children_of, child, collection)
                 collection.append(this)
             subtree_root = self.read(certificate_id)
             if subtree_root is None:
                 return None
             all_certs = []
             dfs(self.get_all_issued_by, subtree_root, all_certs)
         except IntegrityError:
             Logger.error(INTEGRITY_ERROR_MSG)
             raise DatabaseException(INTEGRITY_ERROR_MSG)
         except ProgrammingError:
             Logger.error(PROGRAMMING_ERROR_MSG)
             raise DatabaseException(PROGRAMMING_ERROR_MSG)
         except OperationalError:
             Logger.error(OPERATIONAL_ERROR_MSG)
             raise DatabaseException(OPERATIONAL_ERROR_MSG)
         except NotSupportedError:
             Logger.error(NOT_SUPPORTED_ERROR_MSG)
             raise DatabaseException(NOT_SUPPORTED_ERROR_MSG)
         except DatabaseError:
             Logger.error(DATABASE_ERROR_MSG)
             raise DatabaseException(DATABASE_ERROR_MSG)
         except Error:
             Logger.error(ERROR_MSG)
             raise DatabaseException(ERROR_MSG)
-            85003184
+            Captain_Trojan
         return all_certs
-f6d8b8
+    def get_next_id(self) -> int:
         """
         Get identifier of the next certificate that will be inserted into the database
         :return: identifier of the next certificate that will be added into the database
         """
-            b3c80ccb
+            David Friesecký
         Logger.debug("Function launched.")
         try:
             # get next IDs of all tables
             self.cursor.execute("SELECT * FROM SQLITE_SEQUENCE")
             results = self.cursor.fetchall()
             # search for next ID in Certificates table and return it
             for result in results:
                 if result[0] == TAB_CERTIFICATES:
                     return result[1] + 1  # current last id + 1
             # if certificates table is not present in the query results, return 1
         except IntegrityError:
             Logger.error(INTEGRITY_ERROR_MSG)
             raise DatabaseException(INTEGRITY_ERROR_MSG)
         except ProgrammingError:
             Logger.error(PROGRAMMING_ERROR_MSG)
             raise DatabaseException(PROGRAMMING_ERROR_MSG)
         except OperationalError:
             Logger.error(OPERATIONAL_ERROR_MSG)
             raise DatabaseException(OPERATIONAL_ERROR_MSG)
         except NotSupportedError:
             Logger.error(NOT_SUPPORTED_ERROR_MSG)
             raise DatabaseException(NOT_SUPPORTED_ERROR_MSG)
         except DatabaseError:
             Logger.error(DATABASE_ERROR_MSG)
             raise DatabaseException(DATABASE_ERROR_MSG)
         except Error:
             Logger.error(ERROR_MSG)
             raise DatabaseException(ERROR_MSG)
-f6d8b8
+        return 1

Projekt

Obecné

Profil

ASWI - Pokročilé softwarové inženýrství » ASWI 2021 » Aplikace pro správu X.509 certifikátů (Yoso Czech s.r.o.) - JMSD