1 |
ed55c677
|
Jan Pašek
|
import json
|
2 |
5b57121e
|
Captain_Trojan
|
from datetime import datetime
|
3 |
|
|
from itertools import chain
|
4 |
6422796d
|
Stanislav Král
|
from json import JSONDecodeError
|
5 |
|
|
|
6 |
5b57121e
|
Captain_Trojan
|
from flask import request
|
7 |
1fa243ca
|
Jan Pašek
|
from injector import inject
|
8 |
|
|
|
9 |
|
|
from src.constants import CA_ID, \
|
10 |
|
|
SSL_ID, SIGNATURE_ID, AUTHENTICATION_ID, \
|
11 |
|
|
DATETIME_FORMAT, ROOT_CA_ID, INTERMEDIATE_CA_ID, CERTIFICATE_ID # TODO DATABASE_FILE - not the Controller's
|
12 |
2cecaf70
|
Jan Pašek
|
from src.exceptions.database_exception import DatabaseException
|
13 |
5b57121e
|
Captain_Trojan
|
from src.model.subject import Subject
|
14 |
2cecaf70
|
Jan Pašek
|
from src.services.certificate_service import CertificateService, RevocationReasonInvalidException, \
|
15 |
|
|
CertificateStatusInvalidException
|
16 |
1fa243ca
|
Jan Pašek
|
# responsibility.
|
17 |
5b57121e
|
Captain_Trojan
|
from src.services.key_service import KeyService
|
18 |
|
|
|
19 |
9247d70a
|
Captain_Trojan
|
TREE_NODE_TYPE_COUNT = 3
|
20 |
|
|
|
21 |
5b57121e
|
Captain_Trojan
|
FILTERING = "filtering"
|
22 |
|
|
ISSUER = "issuer"
|
23 |
|
|
US = "usage"
|
24 |
|
|
NOT_AFTER = "notAfter"
|
25 |
|
|
NOT_BEFORE = "notBefore"
|
26 |
|
|
COMMON_NAME = "CN"
|
27 |
|
|
ID = "id"
|
28 |
9247d70a
|
Captain_Trojan
|
USAGE = "usage"
|
29 |
|
|
SUBJECT = "subject"
|
30 |
|
|
VALIDITY_DAYS = "validityDays"
|
31 |
|
|
CA = "CA"
|
32 |
2cecaf70
|
Jan Pašek
|
STATUS = "status"
|
33 |
|
|
REASON = "reason"
|
34 |
|
|
REASON_UNDEFINED = "undefined"
|
35 |
9247d70a
|
Captain_Trojan
|
|
36 |
|
|
E_NO_ISSUER_FOUND = {"success": False, "data": "No certificate authority with such unique ID exists."}
|
37 |
d53c2fdc
|
Captain_Trojan
|
E_NO_CERTIFICATES_FOUND = {"success": False, "data": "No such certificate found."}
|
38 |
cfda1725
|
Stanislav Král
|
E_NO_CERT_PRIVATE_KEY_FOUND = {"success": False,
|
39 |
|
|
"data": "Internal server error (certificate's private key cannot be found)."}
|
40 |
5b57121e
|
Captain_Trojan
|
E_NOT_JSON_FORMAT = {"success": False, "data": "The request must be JSON-formatted."}
|
41 |
|
|
E_CORRUPTED_DATABASE = {"success": False, "data": "Internal server error (corrupted database)."}
|
42 |
|
|
E_GENERAL_ERROR = {"success": False, "data": "Internal server error (unknown origin)."}
|
43 |
|
|
E_MISSING_PARAMETERS = {"success": False, "data": "Invalid request, missing parameters."}
|
44 |
|
|
E_WRONG_PARAMETERS = {"success": False, "data": "Invalid request, wrong parameters."}
|
45 |
|
|
|
46 |
9247d70a
|
Captain_Trojan
|
C_CREATED_SUCCESSFULLY = 201
|
47 |
|
|
C_BAD_REQUEST = 400
|
48 |
cfda1725
|
Stanislav Král
|
C_NOT_FOUND = 404
|
49 |
2cecaf70
|
Jan Pašek
|
C_NO_DATA = 205 # TODO related to 204 issue # TODO related to 204 issue
|
50 |
9247d70a
|
Captain_Trojan
|
C_INTERNAL_SERVER_ERROR = 500
|
51 |
|
|
C_SUCCESS = 200
|
52 |
5b57121e
|
Captain_Trojan
|
|
53 |
|
|
|
54 |
|
|
class CertController:
|
55 |
5b6d9513
|
Captain_Trojan
|
KEY_MAP = {'CA': CA_ID, 'SSL': SSL_ID, 'digitalSignature': SIGNATURE_ID, 'authentication': AUTHENTICATION_ID}
|
56 |
|
|
INVERSE_KEY_MAP = {k: v for v, k in KEY_MAP.items()}
|
57 |
5b57121e
|
Captain_Trojan
|
|
58 |
1fa243ca
|
Jan Pašek
|
@inject
|
59 |
|
|
def __init__(self, certificate_service: CertificateService, key_service: KeyService):
|
60 |
|
|
self.certificate_service = certificate_service
|
61 |
|
|
self.key_service = key_service
|
62 |
|
|
|
63 |
|
|
def create_certificate(self):
|
64 |
5b57121e
|
Captain_Trojan
|
"""create new certificate
|
65 |
|
|
|
66 |
9247d70a
|
Captain_Trojan
|
Create a new certificate based on given information
|
67 |
5b57121e
|
Captain_Trojan
|
|
68 |
|
|
:param body: Certificate data to be created
|
69 |
|
|
:type body: dict | bytes
|
70 |
|
|
|
71 |
|
|
:rtype: CreatedResponse
|
72 |
|
|
"""
|
73 |
9247d70a
|
Captain_Trojan
|
required_keys = {SUBJECT, USAGE, VALIDITY_DAYS} # required fields of the POST req
|
74 |
5b57121e
|
Captain_Trojan
|
|
75 |
9247d70a
|
Captain_Trojan
|
if request.is_json: # accept JSON only
|
76 |
5b57121e
|
Captain_Trojan
|
body = request.get_json()
|
77 |
9247d70a
|
Captain_Trojan
|
if not all(k in body for k in required_keys): # verify that all keys are present
|
78 |
|
|
return E_MISSING_PARAMETERS, C_BAD_REQUEST
|
79 |
5b57121e
|
Captain_Trojan
|
|
80 |
9247d70a
|
Captain_Trojan
|
if not isinstance(body[VALIDITY_DAYS], int): # type checking
|
81 |
|
|
return E_WRONG_PARAMETERS, C_BAD_REQUEST
|
82 |
5b57121e
|
Captain_Trojan
|
|
83 |
9247d70a
|
Captain_Trojan
|
subject = Subject.from_dict(body[SUBJECT]) # generate Subject from passed dict
|
84 |
5b57121e
|
Captain_Trojan
|
|
85 |
9247d70a
|
Captain_Trojan
|
if subject is None: # if the format is incorrect
|
86 |
|
|
return E_WRONG_PARAMETERS, C_BAD_REQUEST
|
87 |
5b57121e
|
Captain_Trojan
|
|
88 |
5b6d9513
|
Captain_Trojan
|
usages_dict = {}
|
89 |
5b57121e
|
Captain_Trojan
|
|
90 |
9247d70a
|
Captain_Trojan
|
if not isinstance(body[USAGE], dict): # type checking
|
91 |
|
|
return E_WRONG_PARAMETERS, C_BAD_REQUEST
|
92 |
5b57121e
|
Captain_Trojan
|
|
93 |
9247d70a
|
Captain_Trojan
|
for k, v in body[USAGE].items(): # for each usage
|
94 |
|
|
if k not in CertController.KEY_MAP: # check that it is a valid usage
|
95 |
|
|
return E_WRONG_PARAMETERS, C_BAD_REQUEST # and throw if it is not
|
96 |
|
|
usages_dict[CertController.KEY_MAP[k]] = v # otherwise translate key and set
|
97 |
5b57121e
|
Captain_Trojan
|
|
98 |
1fa243ca
|
Jan Pašek
|
key = self.key_service.create_new_key() # TODO pass key
|
99 |
5b57121e
|
Captain_Trojan
|
|
100 |
493022a0
|
Jan Pašek
|
if CA not in body or body[CA] is None: # if issuer omitted (legal) or none
|
101 |
1fa243ca
|
Jan Pašek
|
cert = self.certificate_service.create_root_ca( # create a root CA
|
102 |
5b57121e
|
Captain_Trojan
|
key,
|
103 |
|
|
subject,
|
104 |
9247d70a
|
Captain_Trojan
|
usages=usages_dict, # TODO ignoring usages -> discussion
|
105 |
5b57121e
|
Captain_Trojan
|
days=body[VALIDITY_DAYS]
|
106 |
|
|
)
|
107 |
|
|
else:
|
108 |
1fa243ca
|
Jan Pašek
|
issuer = self.certificate_service.get_certificate(body[CA]) # get base issuer info
|
109 |
5b57121e
|
Captain_Trojan
|
|
110 |
9247d70a
|
Captain_Trojan
|
if issuer is None: # if such issuer does not exist
|
111 |
1fa243ca
|
Jan Pašek
|
self.key_service.delete_key(key.private_key_id) # free
|
112 |
9247d70a
|
Captain_Trojan
|
return E_NO_ISSUER_FOUND, C_BAD_REQUEST # and throw
|
113 |
5b57121e
|
Captain_Trojan
|
|
114 |
1fa243ca
|
Jan Pašek
|
issuer_key = self.key_service.get_key(issuer.private_key_id) # get issuer's key, which must exist
|
115 |
5b57121e
|
Captain_Trojan
|
|
116 |
9247d70a
|
Captain_Trojan
|
if issuer_key is None: # if it does not
|
117 |
1fa243ca
|
Jan Pašek
|
self.key_service.delete_key(key.private_key_id) # free
|
118 |
9247d70a
|
Captain_Trojan
|
return E_CORRUPTED_DATABASE, C_INTERNAL_SERVER_ERROR # and throw
|
119 |
5b57121e
|
Captain_Trojan
|
|
120 |
1fa243ca
|
Jan Pašek
|
f = self.certificate_service.create_ca if CA_ID in usages_dict and usages_dict[CA_ID] else \
|
121 |
|
|
self.certificate_service.create_end_cert
|
122 |
5b57121e
|
Captain_Trojan
|
|
123 |
9247d70a
|
Captain_Trojan
|
# noinspection PyTypeChecker
|
124 |
|
|
cert = f( # create inter CA or end cert
|
125 |
|
|
key, # according to whether 'CA' is among
|
126 |
|
|
subject, # the usages' fields
|
127 |
5b57121e
|
Captain_Trojan
|
issuer,
|
128 |
|
|
issuer_key,
|
129 |
|
|
usages=usages_dict,
|
130 |
|
|
days=body[VALIDITY_DAYS]
|
131 |
|
|
)
|
132 |
|
|
|
133 |
|
|
if cert is not None:
|
134 |
|
|
return {"success": True,
|
135 |
9247d70a
|
Captain_Trojan
|
"data": cert.certificate_id}, C_CREATED_SUCCESSFULLY
|
136 |
|
|
else: # if this fails, then
|
137 |
1fa243ca
|
Jan Pašek
|
self.key_service.delete_key(key.private_key_id) # free
|
138 |
9247d70a
|
Captain_Trojan
|
return {"success": False, # and wonder what the cause is,
|
139 |
|
|
"data": "Internal error: The certificate could not have been created."}, C_BAD_REQUEST
|
140 |
|
|
# as obj/None carries only one bit
|
141 |
|
|
# of error information
|
142 |
5b57121e
|
Captain_Trojan
|
else:
|
143 |
9247d70a
|
Captain_Trojan
|
return E_NOT_JSON_FORMAT, C_BAD_REQUEST # throw in case of non-JSON format
|
144 |
5b57121e
|
Captain_Trojan
|
|
145 |
1fa243ca
|
Jan Pašek
|
def get_certificate_by_id(self, id):
|
146 |
5b57121e
|
Captain_Trojan
|
"""get certificate by ID
|
147 |
|
|
|
148 |
9247d70a
|
Captain_Trojan
|
Get certificate in PEM format by ID
|
149 |
5b57121e
|
Captain_Trojan
|
|
150 |
|
|
:param id: ID of a certificate to be queried
|
151 |
|
|
:type id: dict | bytes
|
152 |
|
|
|
153 |
|
|
:rtype: PemResponse
|
154 |
|
|
"""
|
155 |
fb987403
|
Captain_Trojan
|
try:
|
156 |
|
|
v = int(id)
|
157 |
|
|
except ValueError:
|
158 |
9247d70a
|
Captain_Trojan
|
return E_WRONG_PARAMETERS, C_BAD_REQUEST
|
159 |
fb987403
|
Captain_Trojan
|
|
160 |
1fa243ca
|
Jan Pašek
|
cert = self.certificate_service.get_certificate(v)
|
161 |
fb987403
|
Captain_Trojan
|
|
162 |
|
|
if cert is None:
|
163 |
9247d70a
|
Captain_Trojan
|
return E_NO_CERTIFICATES_FOUND, C_NO_DATA
|
164 |
fb987403
|
Captain_Trojan
|
else:
|
165 |
9247d70a
|
Captain_Trojan
|
return {"success": True, "data": cert.pem_data}, C_SUCCESS
|
166 |
5b57121e
|
Captain_Trojan
|
|
167 |
1fa243ca
|
Jan Pašek
|
def get_certificate_details_by_id(self, id):
|
168 |
5b57121e
|
Captain_Trojan
|
"""get certificate's details by ID
|
169 |
|
|
|
170 |
9247d70a
|
Captain_Trojan
|
Get certificate details by ID
|
171 |
5b57121e
|
Captain_Trojan
|
|
172 |
|
|
:param id: ID of a certificate whose details are to be queried
|
173 |
|
|
:type id: dict | bytes
|
174 |
|
|
|
175 |
|
|
:rtype: CertificateResponse
|
176 |
|
|
"""
|
177 |
5b6d9513
|
Captain_Trojan
|
try:
|
178 |
|
|
v = int(id)
|
179 |
|
|
except ValueError:
|
180 |
9247d70a
|
Captain_Trojan
|
return E_WRONG_PARAMETERS, C_BAD_REQUEST
|
181 |
5b6d9513
|
Captain_Trojan
|
|
182 |
1fa243ca
|
Jan Pašek
|
cert = self.certificate_service.get_certificate(v)
|
183 |
5b6d9513
|
Captain_Trojan
|
|
184 |
|
|
if cert is None:
|
185 |
9247d70a
|
Captain_Trojan
|
return E_NO_CERTIFICATES_FOUND, C_NO_DATA
|
186 |
5b6d9513
|
Captain_Trojan
|
else:
|
187 |
1fa243ca
|
Jan Pašek
|
data = self.cert_to_dict_full(cert)
|
188 |
5b6d9513
|
Captain_Trojan
|
if data is None:
|
189 |
9247d70a
|
Captain_Trojan
|
return E_CORRUPTED_DATABASE, C_INTERNAL_SERVER_ERROR
|
190 |
|
|
return {"success": True, "data": data}, C_SUCCESS
|
191 |
5b57121e
|
Captain_Trojan
|
|
192 |
1fa243ca
|
Jan Pašek
|
def get_certificate_list(self):
|
193 |
5b57121e
|
Captain_Trojan
|
"""get list of certificates
|
194 |
|
|
|
195 |
9247d70a
|
Captain_Trojan
|
Lists certificates based on provided filtering options
|
196 |
5b57121e
|
Captain_Trojan
|
|
197 |
|
|
:param filtering: Filter certificate type to be queried
|
198 |
|
|
:type filtering: dict | bytes
|
199 |
|
|
|
200 |
|
|
:rtype: CertificateListResponse
|
201 |
|
|
"""
|
202 |
9247d70a
|
Captain_Trojan
|
targets = {ROOT_CA_ID, INTERMEDIATE_CA_ID, CERTIFICATE_ID} # all targets
|
203 |
ed55c677
|
Jan Pašek
|
|
204 |
|
|
# the filtering parameter can be read as URL argument or as a request body
|
205 |
|
|
if request.is_json or "filtering" in request.args.keys(): # if the request carries JSON data
|
206 |
|
|
if request.is_json:
|
207 |
|
|
data = request.get_json() # get it
|
208 |
|
|
else:
|
209 |
6422796d
|
Stanislav Král
|
try:
|
210 |
|
|
data = {"filtering": json.loads(request.args["filtering"])}
|
211 |
|
|
except JSONDecodeError:
|
212 |
|
|
return E_NOT_JSON_FORMAT, C_BAD_REQUEST
|
213 |
|
|
|
214 |
9247d70a
|
Captain_Trojan
|
if FILTERING in data: # if the 'filtering' field exists
|
215 |
|
|
if isinstance(data[FILTERING], dict): # and it is also a 'dict'
|
216 |
|
|
if CA in data[FILTERING]: # containing 'CA'
|
217 |
|
|
if isinstance(data[FILTERING][CA], bool): # which is a 'bool',
|
218 |
|
|
if data[FILTERING][CA]: # then filter according to 'CA'.
|
219 |
5b57121e
|
Captain_Trojan
|
targets.remove(CERTIFICATE_ID)
|
220 |
|
|
else:
|
221 |
|
|
targets.remove(ROOT_CA_ID)
|
222 |
|
|
targets.remove(INTERMEDIATE_CA_ID)
|
223 |
|
|
else:
|
224 |
9247d70a
|
Captain_Trojan
|
return E_WRONG_PARAMETERS, C_BAD_REQUEST
|
225 |
5b57121e
|
Captain_Trojan
|
else:
|
226 |
9247d70a
|
Captain_Trojan
|
return E_WRONG_PARAMETERS, C_BAD_REQUEST
|
227 |
5b57121e
|
Captain_Trojan
|
|
228 |
9247d70a
|
Captain_Trojan
|
if len(targets) == TREE_NODE_TYPE_COUNT: # = 3 -> root node,
|
229 |
|
|
# intermediate node,
|
230 |
|
|
# or leaf node
|
231 |
|
|
|
232 |
|
|
# if filtering did not change the
|
233 |
|
|
# targets,
|
234 |
1fa243ca
|
Jan Pašek
|
certs = self.certificate_service.get_certificates() # fetch everything
|
235 |
9247d70a
|
Captain_Trojan
|
else: # otherwise fetch targets only
|
236 |
1fa243ca
|
Jan Pašek
|
certs = list(chain(*(self.certificate_service.get_certificates(target) for target in targets)))
|
237 |
5b57121e
|
Captain_Trojan
|
|
238 |
|
|
if certs is None:
|
239 |
9247d70a
|
Captain_Trojan
|
return E_GENERAL_ERROR, C_INTERNAL_SERVER_ERROR
|
240 |
5b57121e
|
Captain_Trojan
|
elif len(certs) == 0:
|
241 |
9247d70a
|
Captain_Trojan
|
return E_NO_CERTIFICATES_FOUND, C_NO_DATA
|
242 |
5b57121e
|
Captain_Trojan
|
else:
|
243 |
|
|
ret = []
|
244 |
|
|
for c in certs:
|
245 |
1fa243ca
|
Jan Pašek
|
data = self.cert_to_dict_partial(c)
|
246 |
5b6d9513
|
Captain_Trojan
|
if data is None:
|
247 |
9247d70a
|
Captain_Trojan
|
return E_CORRUPTED_DATABASE, C_INTERNAL_SERVER_ERROR
|
248 |
5b57121e
|
Captain_Trojan
|
ret.append(
|
249 |
5b6d9513
|
Captain_Trojan
|
data
|
250 |
5b57121e
|
Captain_Trojan
|
)
|
251 |
9247d70a
|
Captain_Trojan
|
return {"success": True, "data": ret}, C_SUCCESS
|
252 |
5b57121e
|
Captain_Trojan
|
|
253 |
1fa243ca
|
Jan Pašek
|
def get_certificate_root_by_id(self, id):
|
254 |
5b57121e
|
Captain_Trojan
|
"""get certificate's root of trust chain by ID
|
255 |
|
|
|
256 |
9247d70a
|
Captain_Trojan
|
Get certificate's root of trust chain in PEM format by ID
|
257 |
5b57121e
|
Captain_Trojan
|
|
258 |
|
|
:param id: ID of a child certificate whose root is to be queried
|
259 |
|
|
:type id: dict | bytes
|
260 |
|
|
|
261 |
|
|
:rtype: PemResponse
|
262 |
|
|
"""
|
263 |
d53c2fdc
|
Captain_Trojan
|
try:
|
264 |
|
|
v = int(id)
|
265 |
|
|
except ValueError:
|
266 |
9247d70a
|
Captain_Trojan
|
return E_WRONG_PARAMETERS, C_BAD_REQUEST
|
267 |
d53c2fdc
|
Captain_Trojan
|
|
268 |
1fa243ca
|
Jan Pašek
|
cert = self.certificate_service.get_certificate(v)
|
269 |
d53c2fdc
|
Captain_Trojan
|
|
270 |
|
|
if cert is None:
|
271 |
11a90594
|
Jan Pašek
|
return E_NO_CERTIFICATES_FOUND, C_NO_DATA
|
272 |
d53c2fdc
|
Captain_Trojan
|
|
273 |
1fa243ca
|
Jan Pašek
|
trust_chain = self.certificate_service.get_chain_of_trust(cert.parent_id, exclude_root=False)
|
274 |
|
|
if trust_chain is None or len(trust_chain) == 0:
|
275 |
11a90594
|
Jan Pašek
|
return E_NO_CERTIFICATES_FOUND, C_NO_DATA
|
276 |
d53c2fdc
|
Captain_Trojan
|
|
277 |
11a90594
|
Jan Pašek
|
return {"success": True, "data": trust_chain[-1].pem_data}, C_SUCCESS
|
278 |
5b57121e
|
Captain_Trojan
|
|
279 |
1fa243ca
|
Jan Pašek
|
def get_certificate_trust_chain_by_id(self, id):
|
280 |
5b57121e
|
Captain_Trojan
|
"""get certificate's trust chain by ID
|
281 |
|
|
|
282 |
9247d70a
|
Captain_Trojan
|
Get certificate trust chain in PEM format by ID
|
283 |
5b57121e
|
Captain_Trojan
|
|
284 |
|
|
:param id: ID of a child certificate whose chain is to be queried
|
285 |
|
|
:type id: dict | bytes
|
286 |
|
|
|
287 |
|
|
:rtype: PemResponse
|
288 |
|
|
"""
|
289 |
|
|
|
290 |
aa740737
|
Captain_Trojan
|
try:
|
291 |
|
|
v = int(id)
|
292 |
|
|
except ValueError:
|
293 |
9247d70a
|
Captain_Trojan
|
return E_WRONG_PARAMETERS, C_BAD_REQUEST
|
294 |
aa740737
|
Captain_Trojan
|
|
295 |
1fa243ca
|
Jan Pašek
|
cert = self.certificate_service.get_certificate(v)
|
296 |
aa740737
|
Captain_Trojan
|
|
297 |
|
|
if cert is None:
|
298 |
9247d70a
|
Captain_Trojan
|
return E_NO_CERTIFICATES_FOUND, C_NO_DATA
|
299 |
aa740737
|
Captain_Trojan
|
|
300 |
11a90594
|
Jan Pašek
|
if cert.parent_id is None:
|
301 |
|
|
return E_NO_CERTIFICATES_FOUND, C_NO_DATA
|
302 |
aa740737
|
Captain_Trojan
|
|
303 |
1fa243ca
|
Jan Pašek
|
trust_chain = self.certificate_service.get_chain_of_trust(cert.parent_id)
|
304 |
11a90594
|
Jan Pašek
|
|
305 |
|
|
ret = []
|
306 |
|
|
for intermediate in trust_chain:
|
307 |
|
|
ret.append(intermediate.pem_data)
|
308 |
5b6d9513
|
Captain_Trojan
|
|
309 |
9247d70a
|
Captain_Trojan
|
return {"success": True, "data": "".join(ret)}, C_SUCCESS
|
310 |
5b6d9513
|
Captain_Trojan
|
|
311 |
2cecaf70
|
Jan Pašek
|
def set_certificate_status(self, id):
|
312 |
|
|
"""
|
313 |
|
|
Revoke a certificate given by ID
|
314 |
|
|
- revocation request may contain revocation reason
|
315 |
|
|
- revocation reason is verified based on the possible predefined values
|
316 |
|
|
- if revocation reason is not specified 'undefined' value is used
|
317 |
|
|
:param id: Identifier of the certificate to be revoked
|
318 |
|
|
:type id: int
|
319 |
|
|
|
320 |
|
|
:rtype: SuccessResponse | ErrorResponse (see OpenAPI definition)
|
321 |
|
|
"""
|
322 |
|
|
required_keys = {STATUS} # required keys
|
323 |
|
|
|
324 |
|
|
# try to parse certificate identifier -> if it is not int return error 400
|
325 |
|
|
try:
|
326 |
|
|
identifier = int(id)
|
327 |
|
|
except ValueError:
|
328 |
|
|
return E_WRONG_PARAMETERS, C_BAD_REQUEST
|
329 |
|
|
|
330 |
|
|
# check if the request contains a JSON body
|
331 |
|
|
if request.is_json:
|
332 |
|
|
request_body = request.get_json()
|
333 |
|
|
# verify that all required keys are present
|
334 |
|
|
if not all(k in request_body for k in required_keys):
|
335 |
|
|
return E_MISSING_PARAMETERS, C_BAD_REQUEST
|
336 |
|
|
|
337 |
|
|
# get status and reason from the request
|
338 |
|
|
status = request_body[STATUS]
|
339 |
|
|
reason = request_body.get(REASON, REASON_UNDEFINED)
|
340 |
|
|
try:
|
341 |
|
|
# set certificate status using certificate_service
|
342 |
|
|
self.certificate_service.set_certificate_revocation_status(status, reason)
|
343 |
|
|
except (RevocationReasonInvalidException, CertificateStatusInvalidException):
|
344 |
|
|
# these exceptions are thrown in case invalid status or revocation reason is passed to the controller
|
345 |
|
|
return E_WRONG_PARAMETERS, C_BAD_REQUEST
|
346 |
|
|
except DatabaseException:
|
347 |
|
|
return E_WRONG_PARAMETERS, C_BAD_REQUEST
|
348 |
|
|
return {"success": True,
|
349 |
|
|
"data": "Certificate status updated successfully."}, C_CREATED_SUCCESSFULLY
|
350 |
|
|
# throw an error in case the request does not contain a json body
|
351 |
|
|
else:
|
352 |
|
|
return E_NOT_JSON_FORMAT, C_BAD_REQUEST
|
353 |
|
|
|
354 |
1fa243ca
|
Jan Pašek
|
def cert_to_dict_partial(self, c):
|
355 |
9247d70a
|
Captain_Trojan
|
"""
|
356 |
|
|
Dictionarizes a certificate directly fetched from the database. Contains partial information.
|
357 |
|
|
:param c: target cert
|
358 |
|
|
:return: certificate dict (compliant with some parts of the REST API)
|
359 |
|
|
"""
|
360 |
1fa243ca
|
Jan Pašek
|
c_issuer = self.certificate_service.get_certificate(c.parent_id)
|
361 |
5b6d9513
|
Captain_Trojan
|
if c_issuer is None:
|
362 |
|
|
return None
|
363 |
|
|
|
364 |
|
|
return {
|
365 |
|
|
ID: c.certificate_id,
|
366 |
|
|
COMMON_NAME: c.common_name,
|
367 |
|
|
NOT_BEFORE: datetime.strptime(c.valid_from, DATETIME_FORMAT).date(),
|
368 |
|
|
NOT_AFTER: datetime.strptime(c.valid_to, DATETIME_FORMAT).date(),
|
369 |
|
|
USAGE: {CertController.INVERSE_KEY_MAP[k]: v for k, v in c.usages.items()},
|
370 |
|
|
ISSUER: {
|
371 |
|
|
ID: c_issuer.certificate_id,
|
372 |
|
|
COMMON_NAME: c_issuer.common_name
|
373 |
|
|
}
|
374 |
|
|
}
|
375 |
|
|
|
376 |
1fa243ca
|
Jan Pašek
|
def cert_to_dict_full(self, c):
|
377 |
9247d70a
|
Captain_Trojan
|
"""
|
378 |
|
|
Dictionarizes a certificate directly fetched from the database, but adds subject info.
|
379 |
|
|
Contains full information.
|
380 |
|
|
:param c: target cert
|
381 |
|
|
:return: certificate dict (compliant with some parts of the REST API)
|
382 |
|
|
"""
|
383 |
1fa243ca
|
Jan Pašek
|
subj = self.certificate_service.get_subject_from_certificate(c)
|
384 |
|
|
c_issuer = self.certificate_service.get_certificate(c.parent_id)
|
385 |
5b6d9513
|
Captain_Trojan
|
if c_issuer is None:
|
386 |
|
|
return None
|
387 |
|
|
|
388 |
|
|
return {
|
389 |
|
|
SUBJECT: subj.to_dict(),
|
390 |
|
|
NOT_BEFORE: datetime.strptime(c.valid_from, DATETIME_FORMAT).date(),
|
391 |
|
|
NOT_AFTER: datetime.strptime(c.valid_to, DATETIME_FORMAT).date(),
|
392 |
|
|
USAGE: {CertController.INVERSE_KEY_MAP[k]: v for k, v in c.usages.items()},
|
393 |
|
|
CA: c_issuer.certificate_id
|
394 |
|
|
}
|
395 |
cfda1725
|
Stanislav Král
|
|
396 |
ce8b9aaf
|
Stanislav Král
|
def get_private_key_of_a_certificate(self, id):
|
397 |
cfda1725
|
Stanislav Král
|
"""
|
398 |
|
|
Get a private key used to sign a certificate in PEM format specified by certificate's ID
|
399 |
|
|
|
400 |
ce8b9aaf
|
Stanislav Král
|
:param id: ID of a certificate whose private key is to be queried
|
401 |
|
|
:type id: dict | bytes
|
402 |
|
|
|
403 |
|
|
:rtype: PemResponse
|
404 |
|
|
"""
|
405 |
|
|
|
406 |
|
|
# try to parse the supplied ID
|
407 |
|
|
try:
|
408 |
|
|
v = int(id)
|
409 |
|
|
except ValueError:
|
410 |
|
|
return E_WRONG_PARAMETERS, C_BAD_REQUEST
|
411 |
|
|
|
412 |
eade7427
|
Stanislav Král
|
# find a certificate using the given ID
|
413 |
ce8b9aaf
|
Stanislav Král
|
cert = self.certificate_service.get_certificate(v)
|
414 |
|
|
|
415 |
|
|
if cert is None:
|
416 |
|
|
return E_NO_CERTIFICATES_FOUND, C_NOT_FOUND
|
417 |
|
|
else:
|
418 |
|
|
# certificate exists, fetch it's private key
|
419 |
|
|
private_key = self.key_service.get_key(cert.private_key_id)
|
420 |
|
|
if cert is None:
|
421 |
|
|
return E_NO_CERT_PRIVATE_KEY_FOUND, C_INTERNAL_SERVER_ERROR
|
422 |
|
|
else:
|
423 |
|
|
return {"success": True, "data": private_key.private_key}, C_SUCCESS
|
424 |
|
|
|
425 |
|
|
def get_public_key_of_a_certificate(self, id):
|
426 |
|
|
"""
|
427 |
|
|
Get a public key of a certificate in PEM format specified by certificate's ID
|
428 |
|
|
|
429 |
cfda1725
|
Stanislav Král
|
:param id: ID of a certificate whose public key is to be queried
|
430 |
|
|
:type id: dict | bytes
|
431 |
|
|
|
432 |
|
|
:rtype: PemResponse
|
433 |
|
|
"""
|
434 |
|
|
|
435 |
|
|
# try to parse the supplied ID
|
436 |
|
|
try:
|
437 |
|
|
v = int(id)
|
438 |
|
|
except ValueError:
|
439 |
|
|
return E_WRONG_PARAMETERS, C_BAD_REQUEST
|
440 |
|
|
|
441 |
eade7427
|
Stanislav Král
|
# find a certificate using the given ID
|
442 |
cfda1725
|
Stanislav Král
|
cert = self.certificate_service.get_certificate(v)
|
443 |
|
|
|
444 |
|
|
if cert is None:
|
445 |
|
|
return E_NO_CERTIFICATES_FOUND, C_NOT_FOUND
|
446 |
|
|
else:
|
447 |
d3bfacfc
|
Stanislav Král
|
return {"success": True, "data": self.certificate_service.get_public_key_from_certificate(cert)}, C_SUCCESS
|