Revize 58051326

Re #8578 - Implemented new functions
- set_certificate_revoked, clear_certificate_revocation, get_all_revoked_by, get_all_issued_by
- update returns of all functions

src/dao/certificate_repository.py

@@ -34 +34 @@
                    f"{COL_VALID_FROM},"
                    f"{COL_VALID_TO},"
                    f"{COL_PEM_DATA},"
+                   f"{COL_REVOCATION_DATE},"
+                   f"{COL_REVOCATION_REASON}"
                    f"{COL_PRIVATE_KEY_ID},"
                    f"{COL_TYPE_ID},"
                    f"{COL_PARENT_ID})"
@@ -42 +44 @@
                       certificate.valid_from,
                       certificate.valid_to,
                       certificate.pem_data,
+                      certificate.revocation_date,
+                      certificate.revocation_reason,
                       certificate.private_key_id,
                       certificate.type_id,
                       certificate.parent_id]
@@ -65 +69 @@
                         self.cursor.execute(sql, values)
                         self.connection.commit()
         except Error as e:
-            print(e)
-            return None
+            return e
 
         return last_id
 
@@ -103 +106 @@
                                                    certificate_row[2],
                                                    certificate_row[3],
                                                    certificate_row[4],
-                                                   certificate_row[5],
-                                                   certificate_row[6],
                                                    certificate_row[7],
-                                                   usage_dict)
+                                                   certificate_row[8],
+                                                   certificate_row[9],
+                                                   usage_dict,
+                                                   certificate_row[5],
+                                                   certificate_row[6])
         except Error as e:
-            print(e)
-            return None
+            return e
 
         if len(certificate_row) > 0:
             return certificate
         else:
             return None
 
-    def read_all(self, filter_type: int = None) -> List[Certificate]:
+    def read_all(self, filter_type: int = None):
         """
         Reads (selects) all certificates (with type).
 
@@ -154 +158 @@
                                                 certificate_row[2],
                                                 certificate_row[3],
                                                 certificate_row[4],
-                                                certificate_row[5],
-                                                certificate_row[6],
                                                 certificate_row[7],
-                                                usage_dict))
+                                                certificate_row[8],
+                                                certificate_row[9],
+                                                usage_dict,
+                                                certificate_row[5],
+                                                certificate_row[6]))
         except Error as e:
-            print(e)
-            return None
+            return e
 
-        return certificates
+        if len(certificates) > 0:
+            return certificates
+        else:
+            return None
 
-    def update(self, certificate_id: int, certificate: Certificate) -> bool:
+    def update(self, certificate_id: int, certificate: Certificate):
         """
         Updates a certificate.
         If the parameter of certificate (Certificate object) is not to be changed,
@@ -182 +190 @@
                    f"{COL_VALID_FROM} = ?, "
                    f"{COL_VALID_TO} = ?, "
                    f"{COL_PEM_DATA} = ?, "
+                   f"{COL_REVOCATION_DATE} = ?, "
+                   f"{COL_REVOCATION_REASON} = ?, "
                    f"{COL_PRIVATE_KEY_ID} = ?, "
                    f"{COL_TYPE_ID} = ?, "
                    f"{COL_PARENT_ID} = ? "
@@ -190 +200 @@
                       certificate.valid_from,
                       certificate.valid_to,
                       certificate.pem_data,
+                      certificate.revocation_date,
+                      certificate.revocation_reason,
                       certificate.private_key_id,
                       certificate.type_id,
                       certificate.parent_id,
@@ -214 +226 @@
                     self.cursor.execute(sql, values)
                     self.connection.commit()
         except Error as e:
-            print(e)
-            return False
+            return e
 
         return True
 
-    def delete(self, certificate_id: int) -> bool:
+    def delete(self, certificate_id: int):
         """
         Deletes a certificate
 
@@ -235 +246 @@
             self.cursor.execute(sql, values)
             self.connection.commit()
         except Error as e:
-            print(e)
-            return False
+            return e
 
         return self.cursor.rowcount > 0
+
+    def set_certificate_revoked(
+            self, certificate_id: int, revocation_date: str, revocation_reason: str = "unspecified"):
+        """
+        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
+
+        :return: the result of whether the revocation was successful
+        """
+
+        try:
+            sql = (f"UPDATE {TAB_CERTIFICATES} "
+                   f"SET {COL_REVOCATION_DATE} = ?, "
+                   f"{COL_REVOCATION_REASON} = ?, "
+                   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()
+        except Error as e:
+            return e
+
+        return True
+
+    def clear_certificate_revocation(self, certificate_id: int):
+        """
+        Clear revocation of a certificate
+
+        :param certificate_id: ID of specific certificate
+
+        :return: the result of whether the clear revocation was successful
+        """
+
+        try:
+            sql = (f"UPDATE {TAB_CERTIFICATES} "
+                   f"SET {COL_REVOCATION_DATE} = '', "
+                   f"{COL_REVOCATION_REASON} = '', "
+                   f"WHERE {COL_ID} = ?")
+            values = [certificate_id]
+            self.cursor.execute(sql, values)
+            self.connection.commit()
+        except Error as e:
+            return e
+
+        return True
+
+    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:
+            # list of the certificates
+            # None if the list is empty
+            # sqlite3.Error if an exception is thrown
+        """
+
+        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[7],
+                                                certificate_row[8],
+                                                certificate_row[9],
+                                                usage_dict,
+                                                certificate_row[5],
+                                                certificate_row[6]))
+        except Error as e:
+            return e
+
+        if len(certificates) > 0:
+            return certificates
+        else:
+            return None
+
+    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:
+            # list of the certificates
+            # None if the list is empty
+            # sqlite3.Error if an exception is thrown
+        """
+
+        try:
+            sql = (f"SELECT * FROM {TAB_CERTIFICATES} "
+                   f"WHERE {COL_PARENT_ID} = ?")
+            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[7],
+                                                certificate_row[8],
+                                                certificate_row[9],
+                                                usage_dict,
+                                                certificate_row[5],
+                                                certificate_row[6]))
+        except Error as e:
+            return e
+
+        if len(certificates) > 0:
+            return certificates
+        else:
+            return None