1
|
from typing import Dict
|
2
|
from sqlite3 import Connection, Cursor, Error
|
3
|
|
4
|
from src.db_objects.certificate import Certificate
|
5
|
from src.constants import *
|
6
|
|
7
|
|
8
|
class CertificateRepository:
|
9
|
|
10
|
def __init__(self, connection: Connection, cursor: Cursor):
|
11
|
"""
|
12
|
Constructor of the CertificateRepository object
|
13
|
|
14
|
:param connection: Instance of the Connection object
|
15
|
:param cursor: Instance of the Cursor object
|
16
|
"""
|
17
|
|
18
|
self.connection = connection
|
19
|
self.cursor = cursor
|
20
|
|
21
|
def create(self, certificate: Certificate) -> bool:
|
22
|
"""
|
23
|
Creates a certificate.
|
24
|
For root certificate (CA) the parent certificate id is modified to the same id (id == parent_id).
|
25
|
|
26
|
:param certificate: Instance of the Certificate object
|
27
|
|
28
|
:return: the result of whether the creation was successful
|
29
|
"""
|
30
|
|
31
|
try:
|
32
|
sql = (f"INSERT INTO {TAB_CERTIFICATES} "
|
33
|
f"({COL_COMMON_NAME},"
|
34
|
f"{COL_VALID_FROM},"
|
35
|
f"{COL_VALID_TO},"
|
36
|
f"{COL_PEM_DATA},"
|
37
|
f"{COL_PRIVATE_KEY_ID},"
|
38
|
f"{COL_TYPE_ID},"
|
39
|
f"{COL_PARENT_ID})"
|
40
|
f"VALUES(?,?,?,?,?,?,?)")
|
41
|
values = [certificate.common_name,
|
42
|
certificate.valid_from,
|
43
|
certificate.valid_to,
|
44
|
certificate.pem_data,
|
45
|
certificate.private_key_id,
|
46
|
certificate.type_id,
|
47
|
certificate.parent_id]
|
48
|
self.cursor.execute(sql, values)
|
49
|
self.connection.commit()
|
50
|
|
51
|
last_id: int = self.cursor.lastrowid
|
52
|
|
53
|
if certificate.usages[ROOT_CA_ID - 1]:
|
54
|
certificate.parent_id = last_id
|
55
|
return self.update(last_id, certificate)
|
56
|
else:
|
57
|
for usage_id, usage_value in certificate.usages:
|
58
|
if usage_value:
|
59
|
sql = (f"INSERT INTO {TAB_CERTIFICATE_USAGES} "
|
60
|
f"({COL_CERTIFICATE_ID},"
|
61
|
f"{COL_USAGE_TYPE_ID}) "
|
62
|
f"VALUES (?,?)")
|
63
|
values = [last_id, usage_id]
|
64
|
self.cursor.execute(sql, values)
|
65
|
self.connection.commit()
|
66
|
except Error as e:
|
67
|
print(e)
|
68
|
return False
|
69
|
|
70
|
return True
|
71
|
|
72
|
def read(self, certificate_id: int) -> Certificate:
|
73
|
"""
|
74
|
Reads (selects) a certificate.
|
75
|
|
76
|
:param certificate_id: ID of specific certificate
|
77
|
|
78
|
:return: instance of the Certificate object
|
79
|
"""
|
80
|
|
81
|
try:
|
82
|
sql = (f"SELECT * FROM {TAB_CERTIFICATES} "
|
83
|
f"WHERE {COL_ID} = ?")
|
84
|
values = [certificate_id]
|
85
|
self.cursor.execute(sql, values)
|
86
|
certificate_row = self.cursor.fetchall()
|
87
|
|
88
|
sql = (f"SELECT * FROM {TAB_CERTIFICATE_USAGES} "
|
89
|
f"WHERE {COL_CERTIFICATE_ID} = ?")
|
90
|
self.cursor.execute(sql, values)
|
91
|
usage_rows = self.cursor.fetchall()
|
92
|
|
93
|
usage_dict: Dict[int, bool] = {}
|
94
|
for usage_row in usage_rows:
|
95
|
usage_dict[usage_row[2]] = True
|
96
|
|
97
|
certificate: Certificate = Certificate(certificate_row[0],
|
98
|
certificate_row[1],
|
99
|
certificate_row[2],
|
100
|
certificate_row[3],
|
101
|
certificate_row[4],
|
102
|
certificate_row[5],
|
103
|
certificate_row[6],
|
104
|
certificate_row[7],
|
105
|
usage_dict)
|
106
|
except Error as e:
|
107
|
print(e)
|
108
|
return None
|
109
|
|
110
|
return certificate
|
111
|
|
112
|
def update(self, certificate_id: int, certificate: Certificate) -> bool:
|
113
|
"""
|
114
|
Updates a certificate.
|
115
|
If the parameter of certificate (Certificate object) is not to be changed,
|
116
|
the same value must be specified.
|
117
|
|
118
|
:param certificate_id: ID of specific certificate
|
119
|
:param certificate: Instance of the Certificate object
|
120
|
|
121
|
:return: the result of whether the updation was successful
|
122
|
"""
|
123
|
|
124
|
try:
|
125
|
sql = (f"UPDATE {TAB_CERTIFICATES} "
|
126
|
f"SET {COL_COMMON_NAME} = ?, "
|
127
|
f"{COL_VALID_FROM} = ?, "
|
128
|
f"{COL_VALID_TO} = ?, "
|
129
|
f"{COL_PEM_DATA} = ?, "
|
130
|
f"{COL_PRIVATE_KEY_ID} = ?, "
|
131
|
f"{COL_TYPE_ID} = ?, "
|
132
|
f"{COL_PARENT_ID} = ? "
|
133
|
f"WHERE {COL_ID} = ?")
|
134
|
values = [certificate.common_name,
|
135
|
certificate.valid_from,
|
136
|
certificate.valid_to,
|
137
|
certificate.pem_data,
|
138
|
certificate.private_key_id,
|
139
|
certificate.type_id,
|
140
|
certificate.parent_id]
|
141
|
self.cursor.execute(sql, values)
|
142
|
self.connection.commit()
|
143
|
|
144
|
sql = (f"DELETE FROM {TAB_CERTIFICATE_USAGES} "
|
145
|
f"WHERE {COL_CERTIFICATE_ID} = ?")
|
146
|
values = [certificate_id]
|
147
|
self.cursor.execute(sql, values)
|
148
|
self.connection.commit()
|
149
|
|
150
|
for usage_id, usage_value in certificate.usages:
|
151
|
if usage_value:
|
152
|
sql = (f"INSERT INTO {TAB_CERTIFICATE_USAGES} "
|
153
|
f"({COL_CERTIFICATE_ID},"
|
154
|
f"{COL_USAGE_TYPE_ID}) "
|
155
|
f"VALUES (?,?)")
|
156
|
values = [certificate_id, usage_id]
|
157
|
self.cursor.execute(sql, values)
|
158
|
self.connection.commit()
|
159
|
except Error as e:
|
160
|
print(e)
|
161
|
return False
|
162
|
|
163
|
return True
|
164
|
|
165
|
def delete(self, certificate_id: int) -> bool:
|
166
|
"""
|
167
|
Deletes a certificate
|
168
|
|
169
|
:param certificate_id: ID of specific certificate
|
170
|
|
171
|
:return: the result of whether the deletion was successful
|
172
|
"""
|
173
|
|
174
|
try:
|
175
|
sql = (f"DELETE FROM {TAB_CERTIFICATES} "
|
176
|
f"WHERE {COL_ID} = ?")
|
177
|
values = [certificate_id]
|
178
|
self.cursor.execute(sql, values)
|
179
|
self.connection.commit()
|
180
|
except Error as e:
|
181
|
print(e)
|
182
|
return False
|
183
|
|
184
|
return True
|