Revize cf247eaa

Re #8702 - Added filtering methods to CertRepository.

src/dao/certificate_repository.py

@@ -1 +1 @@
 import time
-from sqlite3 import Connection, Error, DatabaseError, IntegrityError, ProgrammingError, OperationalError, NotSupportedError
-from typing import Dict, List
+from sqlite3 import Connection, Error, DatabaseError, IntegrityError, ProgrammingError, OperationalError, \
+    NotSupportedError
+from typing import Dict, List, Set
 
 from src.exceptions.database_exception import DatabaseException
 from injector import inject
@@ -227 +228 @@
 
         return certificates
 
+    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
+
     def update(self, certificate_id: int, certificate: Certificate):
         """
         Updates a certificate.
@@ -567 +654 @@
 
         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]))
+        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_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