Revize 6425fa36

Re #8670 - Modified deletion for historical storage of certificates

SQLite_database.sql

@@ -1 +1 @@
 /* ---------------------------------------------------- */
 /*  Generated by Enterprise Architect Version 13.5 		*/
-/*  Created On : 11-dub-2021 0:19:51 				*/
+/*  Created On : 27-dub-2021 11:31:54 				*/
 /*  DBMS       : SQLite 								*/
 /* ---------------------------------------------------- */
 
 /* Drop Tables */
 
-DROP TABLE IF EXISTS 'PrivateKeys'
+DROP TABLE IF EXISTS 'Certificates'
 ;
 
 DROP TABLE IF EXISTS 'CertificateTypes'
 ;
 
-DROP TABLE IF EXISTS 'UsageTypes'
-;
-
-DROP TABLE IF EXISTS 'Certificates'
-;
-
 DROP TABLE IF EXISTS 'CertificateUsages'
 ;
 
-/* Create Tables with Primary and Foreign Keys, Check and Unique Constraints */
-
-CREATE TABLE 'PrivateKeys'
-(
-	'id' INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT,
-	'private_key' TEXT NOT NULL,
-	'password' TEXT NULL
-)
+DROP TABLE IF EXISTS 'PrivateKeys'
 ;
 
-CREATE TABLE 'CertificateTypes'
-(
-	'id' INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT,
-	'certificate_type' TEXT NOT NULL
-)
+DROP TABLE IF EXISTS 'UsageTypes'
 ;
 
-CREATE TABLE 'UsageTypes'
-(
-	'id' INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT,
-	'usage_type' TEXT NOT NULL
-)
-;
+/* Create Tables with Primary and Foreign Keys, Check and Unique Constraints */
 
 CREATE TABLE 'Certificates'
 (
@@ -54 +32 @@
 	'pem_data' TEXT NOT NULL,
 	'revocation_date' TEXT NULL,
 	'revocation_reason' TEXT NULL,
+	'deletion_date' TEXT NULL,
 	'private_key_id' INTEGER NOT NULL,
 	'certificate_type_id' INTEGER NOT NULL,
 	'parent_certificate_id' INTEGER NOT NULL,
@@ -63 +42 @@
 )
 ;
 
+CREATE TABLE 'CertificateTypes'
+(
+	'id' INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT,
+	'certificate_type' TEXT NOT NULL
+)
+;
+
 CREATE TABLE 'CertificateUsages'
 (
 	'id' INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT,
@@ -72 +58 @@
 	CONSTRAINT 'FK_UsageTypes' FOREIGN KEY ('usage_type_id') REFERENCES 'UsageTypes' ('id') ON DELETE No Action ON UPDATE No Action
 )
 ;
+
+CREATE TABLE 'PrivateKeys'
+(
+	'id' INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT,
+	'private_key' TEXT NOT NULL,
+	'password' TEXT NULL
+)
+;
+
+CREATE TABLE 'UsageTypes'
+(
+	'id' INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT,
+	'usage_type' TEXT NOT NULL
+)
+;

src/constants.py

@@ -37 +37 @@
 COL_PEM_DATA = "pem_data"
 COL_REVOCATION_DATE = "revocation_date"
 COL_REVOCATION_REASON = "revocation_reason"
+COL_DELETION_DATE = "deletion_date"
 COL_PRIVATE_KEY_ID = "private_key_id"
 COL_TYPE_ID = "certificate_type_id"
 COL_PARENT_ID = "parent_certificate_id"

src/dao/certificate_repository.py

@@ +1 @@
+import time
 from sqlite3 import Connection, Error
 from typing import Dict, List
 
@@ -81 +82 @@
 
         try:
             sql = (f"SELECT * FROM {TAB_CERTIFICATES} "
-                   f"WHERE {COL_ID} = ?")
+                   f"WHERE {COL_ID} = ? AND ({COL_DELETION_DATE} IS NULL OR {COL_DELETION_DATE} = '')")
             values = [certificate_id]
             self.cursor.execute(sql, values)
             certificate_row = self.cursor.fetchone()
@@ -103 +104 @@
                                                    certificate_row[2],
                                                    certificate_row[3],
                                                    certificate_row[4],
-                                                   certificate_row[7],
                                                    certificate_row[8],
                                                    certificate_row[9],
+                                                   certificate_row[10],
                                                    usage_dict,
                                                    certificate_row[5],
                                                    certificate_row[6])
@@ -127 +128 @@
             sql_extension = ""
             values = []
             if filter_type is not None:
-                sql_extension = (f" WHERE {COL_TYPE_ID} = ("
+                sql_extension = (f" AND {COL_TYPE_ID} = ("
                                  f"SELECT {COL_ID} FROM {TAB_CERTIFICATE_TYPES} WHERE {COL_ID} = ?)")
                 values = [filter_type]
 
-            sql = f"SELECT * FROM {TAB_CERTIFICATES}{sql_extension}"
+            sql = (f"SELECT * FROM {TAB_CERTIFICATES} "
+                   f"WHERE ({COL_DELETION_DATE} IS NULL OR {COL_DELETION_DATE} = ''){sql_extension}")
             self.cursor.execute(sql, values)
             certificate_rows = self.cursor.fetchall()
 
@@ -152 +154 @@
                                                 certificate_row[2],
                                                 certificate_row[3],
                                                 certificate_row[4],
-                                                certificate_row[7],
                                                 certificate_row[8],
                                                 certificate_row[9],
+                                                certificate_row[10],
                                                 usage_dict,
                                                 certificate_row[5],
                                                 certificate_row[6]))
@@ -193 +195 @@
                       certificate.type_id,
                       certificate.parent_id,
                       certificate_id]
+
+            print(f"Parent: {certificate.parent_id}")
+
             self.cursor.execute(sql, values)
             self.connection.commit()
 
@@ -227 +232 @@
         """
 
         try:
-            sql = (f"DELETE FROM {TAB_CERTIFICATES} "
-                   f"WHERE {COL_ID} = ?")
-            values = [certificate_id]
+            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]
             self.cursor.execute(sql, values)
             self.connection.commit()
         except Error as e:
@@ -331 +338 @@
                                                 certificate_row[2],
                                                 certificate_row[3],
                                                 certificate_row[4],
-                                                certificate_row[7],
                                                 certificate_row[8],
                                                 certificate_row[9],
+                                                certificate_row[10],
                                                 usage_dict,
                                                 certificate_row[5],
                                                 certificate_row[6]))
@@ -356 +363 @@
 
         try:
             sql = (f"SELECT * FROM {TAB_CERTIFICATES} "
-                   f"WHERE {COL_PARENT_ID} = ? AND {COL_ID} != ?")
+                   f"WHERE {COL_PARENT_ID} = ? AND {COL_ID} != ? AND "
+                   f"({COL_DELETION_DATE} IS NULL OR {COL_DELETION_DATE} = '')")
             values = [certificate_id, certificate_id]
             self.cursor.execute(sql, values)
             certificate_rows = self.cursor.fetchall()
@@ -378 +386 @@
                                                 certificate_row[2],
                                                 certificate_row[3],
                                                 certificate_row[4],
-                                                certificate_row[7],
                                                 certificate_row[8],
                                                 certificate_row[9],
+                                                certificate_row[10],
                                                 usage_dict,
                                                 certificate_row[5],
                                                 certificate_row[6]))

src/db/init_queries.py

@@ -1 +1 @@
 SCHEMA_SQL = """
 /* ---------------------------------------------------- */
 /*  Generated by Enterprise Architect Version 13.5 		*/
-/*  Created On : 01-dub-2021 15:16:53 				*/
+/*  Created On : 27-dub-2021 11:31:54 				*/
 /*  DBMS       : SQLite 								*/
 /* ---------------------------------------------------- */
 
 /* Drop Tables */
 
-DROP TABLE IF EXISTS 'PrivateKeys'
+DROP TABLE IF EXISTS 'Certificates'
 ;
 
 DROP TABLE IF EXISTS 'CertificateTypes'
 ;
 
-DROP TABLE IF EXISTS 'UsageTypes'
-;
-
-DROP TABLE IF EXISTS 'Certificates'
-;
-
 DROP TABLE IF EXISTS 'CertificateUsages'
 ;
 
-/* Create Tables with Primary and Foreign Keys, Check and Unique Constraints */
-
-CREATE TABLE 'PrivateKeys'
-(
-	'id' INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT,
-	'private_key' TEXT NOT NULL,
-	'password' TEXT NULL
-)
+DROP TABLE IF EXISTS 'PrivateKeys'
 ;
 
-CREATE TABLE 'CertificateTypes'
-(
-	'id' INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT,
-	'certificate_type' TEXT NOT NULL
-)
+DROP TABLE IF EXISTS 'UsageTypes'
 ;
 
-CREATE TABLE 'UsageTypes'
-(
-	'id' INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT,
-	'usage_type' TEXT NOT NULL
-)
-;
+/* Create Tables with Primary and Foreign Keys, Check and Unique Constraints */
 
 CREATE TABLE 'Certificates'
 (
@@ -55 +33 @@
 	'pem_data' TEXT NOT NULL,
 	'revocation_date' TEXT NULL,
 	'revocation_reason' TEXT NULL,
+	'deletion_date' TEXT NULL,
 	'private_key_id' INTEGER NOT NULL,
 	'certificate_type_id' INTEGER NOT NULL,
 	'parent_certificate_id' INTEGER NOT NULL,
@@ -64 +43 @@
 )
 ;
 
+CREATE TABLE 'CertificateTypes'
+(
+	'id' INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT,
+	'certificate_type' TEXT NOT NULL
+)
+;
+
 CREATE TABLE 'CertificateUsages'
 (
 	'id' INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT,
@@ -74 +60 @@
 )
 ;
 
+CREATE TABLE 'PrivateKeys'
+(
+	'id' INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT,
+	'private_key' TEXT NOT NULL,
+	'password' TEXT NULL
+)
+;
+
+CREATE TABLE 'UsageTypes'
+(
+	'id' INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT,
+	'usage_type' TEXT NOT NULL
+)
+;
+
 """
 
 DEFAULT_VALUES_SQL = """

tests/integration_tests/dao/certificate_repository_test.py

@@ -138 +138 @@
     assert len(certificates) == 2
     assert certificates[0].certificate_id == child_1_id and certificates[1].certificate_id == child_3_id
 
+    certificate_repository.delete(certificate_id=sec_tree_child_1)
+
+    certificates = certificate_repository.get_all_revoked_by(certificate_id=certificate_id)
+
+    assert len(certificates) == 2
+    assert certificates[0].certificate_id == child_1_id and certificates[1].certificate_id == child_3_id
 
 def test_get_all_issued_by(certificate_repository, private_key_repository_unique):
     private_key = PrivateKey(private_key_id=-1, private_key="test_pk", password="test_password")
@@ -191 +197 @@
     certificates = certificate_repository.get_all_issued_by(certificate_id=certificate_id)
 
     assert len(certificates) == 3
-    assert certificates[0].certificate_id == child_1_id and \
-           certificates[1].certificate_id == child_2_id and \
-           certificates[2].certificate_id == child_3_id
+    assert certificates[0].certificate_id == child_1_id
+    assert certificates[1].certificate_id == child_2_id
+    assert certificates[2].certificate_id == child_3_id
+
+    certificate_repository.delete(certificate_id=child_1_id)
+
+    certificates = certificate_repository.get_all_issued_by(certificate_id=certificate_id)
+
+    assert len(certificates) == 2
+    assert certificates[0].certificate_id == child_2_id
+    assert certificates[1].certificate_id == child_3_id
+