Projekt

Obecné

Profil

Stáhnout (28.6 KB) Statistiky
| Větev: | Tag: | Revize:
1
import json
2
from datetime import datetime
3
from itertools import chain
4
from json import JSONDecodeError
5

    
6
from flask import request
7
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
from src.controllers.return_codes import *
13
from src.exceptions.database_exception import DatabaseException
14
from src.exceptions.unknown_exception import UnknownException
15
from src.model.subject import Subject
16
from src.services.certificate_service import CertificateService, RevocationReasonInvalidException, \
17
    CertificateStatusInvalidException, CertificateNotFoundException, CertificateAlreadyRevokedException, \
18
    CertificateCannotBeSetToValid
19
#  responsibility.
20
from src.services.key_service import KeyService
21
from src.utils.logger import Logger
22
from src.utils.util import dict_to_string
23

    
24
TREE_NODE_TYPE_COUNT = 3
25

    
26
FILTERING = "filtering"
27
PAGE = "page"
28
PER_PAGE = "per_page"
29
ISSUER = "issuer"
30
US = "usage"
31
NOT_AFTER = "notAfter"
32
NOT_BEFORE = "notBefore"
33
COMMON_NAME = "CN"
34
ID = "id"
35
USAGE = "usage"
36
SUBJECT = "subject"
37
VALIDITY_DAYS = "validityDays"
38
TYPE = "type"
39
ISSUED_BY = "issuedby"
40
STATUS = "status"
41
REASON = "reason"
42
REASON_UNDEFINED = "unspecified"
43

    
44
E_NO_ISSUER_FOUND = {"success": False, "data": "No certificate authority with such unique ID exists."}
45
E_NO_CERTIFICATES_FOUND = {"success": False, "data": "No such certificate found."}
46
E_NO_CERTIFICATE_ALREADY_REVOKED = {"success": False, "data": "Certificate is already revoked."}
47
E_NO_CERT_PRIVATE_KEY_FOUND = {"success": False,
48
                               "data": "Internal server error (certificate's private key cannot be found)."}
49
E_NOT_JSON_FORMAT = {"success": False, "data": "The request must be JSON-formatted."}
50
E_CORRUPTED_DATABASE = {"success": False, "data": "Internal server error (corrupted database)."}
51
E_GENERAL_ERROR = {"success": False, "data": "Internal server error (unknown origin)."}
52
E_MISSING_PARAMETERS = {"success": False, "data": "Invalid request, missing parameters."}
53
E_WRONG_PARAMETERS = {"success": False, "data": "Invalid request, wrong parameters."}
54

    
55

    
56
class CertController:
57
    USAGE_KEY_MAP = {'CA': CA_ID, 'SSL': SSL_ID, 'digitalSignature': SIGNATURE_ID, 'authentication': AUTHENTICATION_ID}
58
    INVERSE_USAGE_KEY_MAP = {k: v for v, k in USAGE_KEY_MAP.items()}
59
    FILTERING_TYPE_KEY_MAP = {'root': ROOT_CA_ID, 'inter': INTERMEDIATE_CA_ID, 'end': CERTIFICATE_ID}
60
    # INVERSE_FILTERING_TYPE_KEY_MAP = {k: v for v, k in FILTERING_TYPE_KEY_MAP.items()}
61

    
62

    
63
    @inject
64
    def __init__(self, certificate_service: CertificateService, key_service: KeyService):
65
        self.certificate_service = certificate_service
66
        self.key_service = key_service
67

    
68
    def create_certificate(self):
69
        """create new certificate
70

    
71
        Create a new certificate based on given information
72

    
73
        :param body: Certificate data to be created
74
        :type body: dict | bytes
75

    
76
        :rtype: CreatedResponse
77
        """
78

    
79
        Logger.info(f"\n\t{request.referrer}"
80
                    f"\n\t{request.method}   {request.path}   {request.scheme}")
81

    
82
        required_keys = {SUBJECT, USAGE, VALIDITY_DAYS}                             # required fields of the POST req
83

    
84
        if request.is_json:                                                         # accept JSON only
85
            body = request.get_json()
86

    
87
            Logger.info(f"\n\tRequest body:"
88
                        f"\n{dict_to_string(body)}")
89

    
90
            if not all(k in body for k in required_keys):                           # verify that all keys are present
91
                Logger.error(f"Invalid request, missing parameters")
92
                return E_MISSING_PARAMETERS, C_BAD_REQUEST
93

    
94
            if not isinstance(body[VALIDITY_DAYS], int):                            # type checking
95
                Logger.error(f"Invalid request, wrong parameter '{VALIDITY_DAYS}'.")
96
                return E_WRONG_PARAMETERS, C_BAD_REQUEST
97

    
98
            subject = Subject.from_dict(body[SUBJECT])                              # generate Subject from passed dict
99

    
100
            if subject is None:                                                     # if the format is incorrect
101
                Logger.error(f"Invalid request, wrong parameter '{SUBJECT}'.")
102
                return E_WRONG_PARAMETERS, C_BAD_REQUEST
103

    
104
            usages_dict = {}
105

    
106
            if not isinstance(body[USAGE], dict):                                   # type checking
107
                Logger.error(f"Invalid request, wrong parameter '{USAGE}'.")
108
                return E_WRONG_PARAMETERS, C_BAD_REQUEST
109

    
110
            for k, v in body[USAGE].items():                                        # for each usage
111
                if k not in CertController.USAGE_KEY_MAP:                                 # check that it is a valid usage
112
                    Logger.error(f"Invalid request, wrong parameter '{USAGE}'[{k}].")
113
                    return E_WRONG_PARAMETERS, C_BAD_REQUEST                        # and throw if it is not
114
                usages_dict[CertController.USAGE_KEY_MAP[k]] = v                          # otherwise translate key and set
115

    
116
            key = self.key_service.create_new_key()                                      # TODO pass key
117

    
118
            if TYPE not in body or body[TYPE] is None:                                  # if issuer omitted (legal) or none
119
                cert = self.certificate_service.create_root_ca(                          # create a root CA
120
                    key,
121
                    subject,
122
                    usages=usages_dict,                                             # TODO ignoring usages -> discussion
123
                    days=body[VALIDITY_DAYS]
124
                )
125
            else:
126
                issuer = self.certificate_service.get_certificate(body[TYPE])              # get base issuer info
127

    
128
                if issuer is None:                                                  # if such issuer does not exist
129
                    Logger.error(f"No certificate authority with such unique ID exists 'ID = {key.private_key_id}'.")
130
                    self.key_service.delete_key(key.private_key_id)                      # free
131
                    return E_NO_ISSUER_FOUND, C_BAD_REQUEST                         # and throw
132

    
133
                issuer_key = self.key_service.get_key(issuer.private_key_id)             # get issuer's key, which must exist
134

    
135
                if issuer_key is None:                                              # if it does not
136
                    Logger.error(f"Internal server error (corrupted database).")
137
                    self.key_service.delete_key(key.private_key_id)                      # free
138
                    return E_CORRUPTED_DATABASE, C_INTERNAL_SERVER_ERROR            # and throw
139

    
140
                f = self.certificate_service.create_ca if CA_ID in usages_dict and usages_dict[CA_ID] else \
141
                    self.certificate_service.create_end_cert
142

    
143
                # noinspection PyTypeChecker
144
                cert = f(                                                           # create inter CA or end cert
145
                    key,                                                            # according to whether 'CA' is among
146
                    subject,                                                        # the usages' fields
147
                    issuer,
148
                    issuer_key,
149
                    usages=usages_dict,
150
                    days=body[VALIDITY_DAYS]
151
                )
152

    
153
            if cert is not None:
154
                return {"success": True,
155
                        "data": cert.certificate_id}, C_CREATED_SUCCESSFULLY
156
            else:                                                                   # if this fails, then
157
                Logger.error(f"Internal error: The certificate could not have been created.")
158
                self.key_service.delete_key(key.private_key_id)                          # free
159
                return {"success": False,                                           # and wonder what the cause is,
160
                        "data": "Internal error: The certificate could not have been created."}, C_BAD_REQUEST
161
                                                                                    # as obj/None carries only one bit
162
                                                                                    # of error information
163
        else:
164
            Logger.error(f"The request must be JSON-formatted.")
165
            return E_NOT_JSON_FORMAT, C_BAD_REQUEST                                 # throw in case of non-JSON format
166

    
167
    def get_certificate_by_id(self, id):
168
        """get certificate by ID
169

    
170
        Get certificate in PEM format by ID
171

    
172
        :param id: ID of a certificate to be queried
173
        :type id: dict | bytes
174

    
175
        :rtype: PemResponse
176
        """
177

    
178
        Logger.info(f"\n\t{request.referrer}"
179
                    f"\n\t{request.method}   {request.path}   {request.scheme}"
180
                    f"\n\tCertificate ID = {id}")
181
        try:
182
            v = int(id)
183
        except ValueError:
184
            Logger.error(f"Invalid request, wrong parameters 'id'[{id}].")
185
            return E_WRONG_PARAMETERS, C_BAD_REQUEST
186

    
187
        cert = self.certificate_service.get_certificate(v)
188

    
189
        if cert is None:
190
            Logger.error(f"No such certificate found 'ID = {v}'.")
191
            return E_NO_CERTIFICATES_FOUND, C_NO_DATA
192
        else:
193
            return {"success": True, "data": cert.pem_data}, C_SUCCESS
194

    
195
    def get_certificate_details_by_id(self, id):
196
        """get certificate's details by ID
197

    
198
        Get certificate details by ID
199

    
200
        :param id: ID of a certificate whose details are to be queried
201
        :type id: dict | bytes
202

    
203
        :rtype: CertificateResponse
204
        """
205

    
206
        Logger.info(f"\n\t{request.referrer}"
207
                    f"\n\t{request.method}   {request.path}   {request.scheme}"
208
                    f"\n\tCertificate ID = {id}")
209

    
210
        try:
211
            v = int(id)
212
        except ValueError:
213
            Logger.error(f"Invalid request, wrong parameters 'id'[{id}].")
214
            return E_WRONG_PARAMETERS, C_BAD_REQUEST
215

    
216
        cert = self.certificate_service.get_certificate(v)
217

    
218
        if cert is None:
219
            Logger.error(f"No such certificate found 'ID = {v}'.")
220
            return E_NO_CERTIFICATES_FOUND, C_NOT_FOUND
221

    
222
        data = self.cert_to_dict_full(cert)
223
        if data is None:
224
            return E_CORRUPTED_DATABASE, C_INTERNAL_SERVER_ERROR
225

    
226
        try:
227
            state = self.certificate_service.get_certificate_state(v)
228
            data["status"] = state
229
        except CertificateNotFoundException:
230
            Logger.error(f"No such certificate found 'ID = {id}'.")
231

    
232
        return {"success": True, "data": data}, C_SUCCESS
233

    
234
    def get_certificate_list(self):
235
        """get list of certificates
236

    
237
        Lists certificates based on provided filtering options
238

    
239
        :param filtering: Filter certificate type to be queried
240
        :type filtering: dict | bytes
241

    
242
        :rtype: CertificateListResponse
243
        """
244

    
245
        Logger.info(f"\n\t{request.referrer}"
246
                    f"\n\t{request.method}   {request.path}   {request.scheme}")
247

    
248

    
249
        # the filtering parameter can be read as URL argument or as a request body
250
        if request.is_json:
251
            data = request.get_json()
252
        else:
253
            data = {}
254

    
255
        if FILTERING in request.args.keys():
256
            try:
257
                data[FILTERING] = json.loads(request.args[FILTERING])
258
            except JSONDecodeError:
259
                Logger.error(f"The request must be JSON-formatted.")
260
                return E_NOT_JSON_FORMAT, C_BAD_REQUEST
261

    
262
        if PAGE in request.args.keys():
263
            try:
264
                data[PAGE] = json.loads(request.args[PAGE])
265
            except JSONDecodeError:
266
                Logger.error(f"The request must be JSON-formatted.")
267
                return E_NOT_JSON_FORMAT, C_BAD_REQUEST
268

    
269
        if PER_PAGE in request.args.keys():
270
            try:
271
                data[PER_PAGE] = json.loads(request.args[PER_PAGE])
272
            except JSONDecodeError:
273
                Logger.error(f"The request must be JSON-formatted.")
274
                return E_NOT_JSON_FORMAT, C_BAD_REQUEST
275

    
276
        Logger.info(f"\n\tRequest body:"
277
                    f"\n{dict_to_string(data)}")
278

    
279
        target_types = {ROOT_CA_ID, INTERMEDIATE_CA_ID, CERTIFICATE_ID}
280
        target_usages = {v for v in CertController.INVERSE_USAGE_KEY_MAP.keys()}
281
        target_cn_substring = None
282
        issuer_id = -1
283
        if PER_PAGE in data:
284
            page = data.get(PAGE, 0)
285
            per_page = data[PER_PAGE]
286
        else:
287
            page = None
288
            per_page = None
289

    
290
        if FILTERING in data:                                                   # if the 'filtering' field exists
291
            if isinstance(data[FILTERING], dict):                               # and it is also a 'dict'
292

    
293
                # noinspection DuplicatedCode
294
                if TYPE in data[FILTERING]:                                     # containing 'type'
295
                    if isinstance(data[FILTERING][TYPE], list):                 # which is a 'list',
296
                                                                                # map every field to id
297
                        try:
298
                            target_types = {CertController.FILTERING_TYPE_KEY_MAP[v] for v in data[FILTERING][TYPE]}
299
                        except KeyError as e:
300
                            Logger.error(f"Invalid request, wrong parameters '{FILTERING}.{TYPE}' - '{e}'.")
301
                            return E_WRONG_PARAMETERS, C_BAD_REQUEST
302
                    else:
303
                        Logger.error(f"Invalid request, wrong parameters '{FILTERING}.{TYPE}'.")
304
                        return E_WRONG_PARAMETERS, C_BAD_REQUEST
305

    
306
                # noinspection DuplicatedCode
307
                if USAGE in data[FILTERING]:                                    # containing 'usage'
308
                    if isinstance(data[FILTERING][USAGE], list):                # which is a 'list',
309
                                                                                # map every field to id
310
                        try:
311
                            target_usages = {CertController.USAGE_KEY_MAP[v] for v in data[FILTERING][USAGE]}
312
                        except KeyError as e:
313
                            Logger.error(f"Invalid request, wrong parameters '{FILTERING}.{TYPE}' - '{e}'.")
314
                            return E_WRONG_PARAMETERS, C_BAD_REQUEST
315
                    else:
316
                        Logger.error(f"Invalid request, wrong parameters '{FILTERING}.{USAGE}'.")
317
                        return E_WRONG_PARAMETERS, C_BAD_REQUEST
318

    
319
                if COMMON_NAME in data[FILTERING]:                              # containing 'CN'
320
                    if isinstance(data[FILTERING][COMMON_NAME], str):           # which is a 'str'
321
                        target_cn_substring = data[FILTERING][COMMON_NAME]
322
                    else:
323
                        Logger.error(f"Invalid request, wrong parameters '{FILTERING}.{COMMON_NAME}'.")
324
                        return E_WRONG_PARAMETERS, C_BAD_REQUEST
325

    
326
                if ISSUED_BY in data[FILTERING]:                                # containing 'issuedby'
327
                    if isinstance(data[FILTERING][ISSUED_BY], int):             # which is an 'int'
328
                        issuer_id = data[FILTERING][ISSUED_BY]                  # then get its children only
329

    
330
            else:
331
                Logger.error(f"Invalid request, wrong parameters '{FILTERING}'.")
332
                return E_WRONG_PARAMETERS, C_BAD_REQUEST
333

    
334
            if issuer_id >= 0:                                                  # if filtering by an issuer
335
                try:
336
                                                                                # get his children, filtered
337
                    certs = self.certificate_service.get_certificates_issued_by_filter(
338
                        issuer_id=issuer_id,
339
                        target_types=target_types,
340
                        target_usages=target_usages,
341
                        target_cn_substring=target_cn_substring,
342
                        page=page,
343
                        per_page=per_page
344
                    )
345
                except CertificateNotFoundException:                            # if id does not exist
346
                    Logger.error(f"No such certificate found 'ID = {issuer_id}'.")
347
                    return E_NO_CERTIFICATES_FOUND, C_NOT_FOUND                 # throw
348
            else:
349
                certs = self.certificate_service.get_certificates_filter(
350
                    target_types=target_types,
351
                    target_usages=target_usages,
352
                    target_cn_substring=target_cn_substring,
353
                    page=page,
354
                    per_page=per_page
355
                )
356
        else:
357
            certs = self.certificate_service.get_certificates()
358

    
359
        if certs is None:
360
            Logger.error(f"Internal server error (unknown origin).")
361
            return E_GENERAL_ERROR, C_INTERNAL_SERVER_ERROR
362
        elif len(certs) == 0:
363
            # TODO check log level
364
            Logger.warning(f"No such certificate found (empty list).")
365
            return E_NO_CERTIFICATES_FOUND, C_NO_DATA
366
        else:
367
            ret = []
368
            for c in certs:
369
                data = self.cert_to_dict_partial(c)
370
                if data is None:
371
                    Logger.error(f"Internal server error (corrupted database).")
372
                    return E_CORRUPTED_DATABASE, C_INTERNAL_SERVER_ERROR
373
                ret.append(
374
                    data
375
                )
376
            return {"success": True, "data": ret}, C_SUCCESS
377

    
378
    def get_certificate_root_by_id(self, id):
379
        """get certificate's root of trust chain by ID
380

    
381
        Get certificate's root of trust chain in PEM format by ID
382

    
383
        :param id: ID of a child certificate whose root is to be queried
384
        :type id: dict | bytes
385

    
386
        :rtype: PemResponse
387
        """
388

    
389
        Logger.info(f"\n\t{request.referrer}"
390
                    f"\n\t{request.method}   {request.path}   {request.scheme}"
391
                    f"\n\tCertificate ID = {id}")
392

    
393
        try:
394
            v = int(id)
395
        except ValueError:
396
            Logger.error(f"Invalid request, wrong parameters 'id'[{id}].")
397
            return E_WRONG_PARAMETERS, C_BAD_REQUEST
398

    
399
        cert = self.certificate_service.get_certificate(v)
400

    
401
        if cert is None:
402
            Logger.error(f"No such certificate found 'ID = {v}'.")
403
            return E_NO_CERTIFICATES_FOUND, C_NO_DATA
404

    
405
        trust_chain = self.certificate_service.get_chain_of_trust(cert.parent_id, exclude_root=False)
406
        if trust_chain is None or len(trust_chain) == 0:
407
            Logger.error(f"No such certificate found (empty list).")
408
            return E_NO_CERTIFICATES_FOUND, C_NO_DATA
409

    
410
        return {"success": True, "data": trust_chain[-1].pem_data}, C_SUCCESS
411

    
412
    def get_certificate_trust_chain_by_id(self, id):
413
        """get certificate's trust chain by ID (including root certificate)
414

    
415
        Get certificate trust chain in PEM format by ID
416

    
417
        :param id: ID of a child certificate whose chain is to be queried
418
        :type id: dict | bytes
419

    
420
        :rtype: PemResponse
421
        """
422

    
423
        Logger.info(f"\n\t{request.referrer}"
424
                    f"\n\t{request.method}   {request.path}   {request.scheme}"
425
                    f"\n\tCertificate ID = {id}")
426

    
427
        try:
428
            v = int(id)
429
        except ValueError:
430
            Logger.error(f"Invalid request, wrong parameters 'id'[{id}].")
431
            return E_WRONG_PARAMETERS, C_BAD_REQUEST
432

    
433
        cert = self.certificate_service.get_certificate(v)
434

    
435
        if cert is None:
436
            Logger.error(f"No such certificate found 'ID = {v}'.")
437
            return E_NO_CERTIFICATES_FOUND, C_NO_DATA
438

    
439
        if cert.parent_id is None:
440
            Logger.error(f"Parent ID is empty in certificate 'ID = {v}'.")
441
            return E_NO_CERTIFICATES_FOUND, C_NO_DATA
442

    
443
        trust_chain = self.certificate_service.get_chain_of_trust(cert.parent_id, exclude_root=False)
444

    
445
        ret = []
446
        for intermediate in trust_chain:
447
            ret.append(intermediate.pem_data)
448

    
449
        return {"success": True, "data": "".join(ret)}, C_SUCCESS
450

    
451
    def set_certificate_status(self, id):
452
        """
453
        Revoke a certificate given by ID
454
            - revocation request may contain revocation reason
455
            - revocation reason is verified based on the possible predefined values
456
            - if revocation reason is not specified 'undefined' value is used
457
        :param id: Identifier of the certificate to be revoked
458
        :type id: int
459

    
460
        :rtype: SuccessResponse | ErrorResponse (see OpenAPI definition)
461
        """
462

    
463
        Logger.info(f"\n\t{request.referrer}"
464
                    f"\n\t{request.method}   {request.path}   {request.scheme}"
465
                    f"\n\tCertificate ID = {id}")
466

    
467
        required_keys = {STATUS}  # required keys
468

    
469
        # check if the request contains a JSON body
470
        if request.is_json:
471
            request_body = request.get_json()
472

    
473
            Logger.info(f"\n\tRequest body:"
474
                        f"\n{dict_to_string(request_body)}")
475

    
476
            # try to parse certificate identifier -> if it is not int return error 400
477
            try:
478
                identifier = int(id)
479
            except ValueError:
480
                Logger.error(f"Invalid request, wrong parameters 'id'[{id}].")
481
                return E_WRONG_PARAMETERS, C_BAD_REQUEST
482

    
483
            # verify that all required keys are present
484
            if not all(k in request_body for k in required_keys):
485
                Logger.error(f"Invalid request, missing parameters.")
486
                return E_MISSING_PARAMETERS, C_BAD_REQUEST
487

    
488
            # get status and reason from the request
489
            status = request_body[STATUS]
490
            reason = request_body.get(REASON, REASON_UNDEFINED)
491
            try:
492
                # set certificate status using certificate_service
493
                self.certificate_service.set_certificate_revocation_status(identifier, status, reason)
494
            except (RevocationReasonInvalidException, CertificateStatusInvalidException):
495
                # these exceptions are thrown in case invalid status or revocation reason is passed to the controller
496
                Logger.error(f"Invalid request, wrong parameters.")
497
                return E_WRONG_PARAMETERS, C_BAD_REQUEST
498
            except CertificateAlreadyRevokedException:
499
                Logger.error(f"Certificate is already revoked 'ID = {identifier}'.")
500
                return E_NO_CERTIFICATE_ALREADY_REVOKED, C_BAD_REQUEST
501
            except CertificateNotFoundException:
502
                Logger.error(f"No such certificate found 'ID = {identifier}'.")
503
                return E_NO_CERTIFICATES_FOUND, C_NOT_FOUND
504
            except CertificateCannotBeSetToValid as e:
505
                return {"success": False, "data": str(e)}, C_BAD_REQUEST
506
            return {"success": True,
507
                    "data": "Certificate status updated successfully."}, C_SUCCESS
508
        # throw an error in case the request does not contain a json body
509
        else:
510
            Logger.error(f"The request must be JSON-formatted.")
511
            return E_NOT_JSON_FORMAT, C_BAD_REQUEST
512

    
513
    def cert_to_dict_partial(self, c):
514
        """
515
        Dictionarizes a certificate directly fetched from the database. Contains partial information.
516
        :param c: target cert
517
        :return: certificate dict (compliant with some parts of the REST API)
518
        """
519

    
520
        # TODO check log
521
        Logger.debug(f"Function launched.")
522

    
523
        c_issuer = self.certificate_service.get_certificate(c.parent_id)
524
        if c_issuer is None:
525
            return None
526

    
527
        return {
528
            ID: c.certificate_id,
529
            COMMON_NAME: c.common_name,
530
            NOT_BEFORE: datetime.strptime(c.valid_from, DATETIME_FORMAT).date(),
531
            NOT_AFTER: datetime.strptime(c.valid_to, DATETIME_FORMAT).date(),
532
            USAGE: {CertController.INVERSE_USAGE_KEY_MAP[k]: v for k, v in c.usages.items()},
533
            ISSUER: {
534
                ID: c_issuer.certificate_id,
535
                COMMON_NAME: c_issuer.common_name
536
            }
537
        }
538

    
539
    def cert_to_dict_full(self, c):
540
        """
541
        Dictionarizes a certificate directly fetched from the database, but adds subject info.
542
        Contains full information.
543
        :param c: target cert
544
        :return: certificate dict (compliant with some parts of the REST API)
545
        """
546

    
547
        Logger.info(f"Function launched.")
548

    
549
        subj = self.certificate_service.get_subject_from_certificate(c)
550
        c_issuer = self.certificate_service.get_certificate(c.parent_id)
551
        if c_issuer is None:
552
            return None
553

    
554
        return {
555
            SUBJECT: subj.to_dict(),
556
            NOT_BEFORE: datetime.strptime(c.valid_from, DATETIME_FORMAT).date(),
557
            NOT_AFTER: datetime.strptime(c.valid_to, DATETIME_FORMAT).date(),
558
            USAGE: {CertController.INVERSE_USAGE_KEY_MAP[k]: v for k, v in c.usages.items()},
559
            TYPE: c_issuer.certificate_id
560
        }
561

    
562
    def get_private_key_of_a_certificate(self, id):
563
        """
564
        Get a private key used to sign a certificate in PEM format specified by certificate's ID
565

    
566
        :param id: ID of a certificate whose private key is to be queried
567
        :type id: dict | bytes
568

    
569
        :rtype: PemResponse
570
        """
571

    
572
        Logger.info(f"\n\t{request.referrer}"
573
                    f"\n\t{request.method}   {request.path}   {request.scheme}"
574
                    f"\n\tCertificate ID = {id}")
575

    
576
        # try to parse the supplied ID
577
        try:
578
            v = int(id)
579
        except ValueError:
580
            Logger.error(f"Invalid request, wrong parameters 'id'[{id}].")
581
            return E_WRONG_PARAMETERS, C_BAD_REQUEST
582

    
583
        # find a certificate using the given ID
584
        cert = self.certificate_service.get_certificate(v)
585

    
586
        if cert is None:
587
            Logger.error(f"No such certificate found 'ID = {v}'.")
588
            return E_NO_CERTIFICATES_FOUND, C_NOT_FOUND
589
        else:
590
            # certificate exists, fetch it's private key
591
            private_key = self.key_service.get_key(cert.private_key_id)
592
            if cert is None:
593
                Logger.error(f"Internal server error (certificate's private key cannot be found).")
594
                return E_NO_CERT_PRIVATE_KEY_FOUND, C_INTERNAL_SERVER_ERROR
595
            else:
596
                return {"success": True, "data": private_key.private_key}, C_SUCCESS
597

    
598
    def get_public_key_of_a_certificate(self, id):
599
        """
600
        Get a public key of a certificate in PEM format specified by certificate's ID
601

    
602
        :param id: ID of a certificate whose public key is to be queried
603
        :type id: dict | bytes
604

    
605
        :rtype: PemResponse
606
        """
607

    
608
        Logger.info(f"\n\t{request.referrer}"
609
                    f"\n\t{request.method}   {request.path}   {request.scheme}"
610
                    f"\n\tCertificate ID = {id}")
611

    
612
        # try to parse the supplied ID
613
        try:
614
            v = int(id)
615
        except ValueError:
616
            Logger.error(f"Invalid request, wrong parameters 'id'[{id}].")
617
            return E_WRONG_PARAMETERS, C_BAD_REQUEST
618

    
619
        # find a certificate using the given ID
620
        cert = self.certificate_service.get_certificate(v)
621

    
622
        if cert is None:
623
            Logger.error(f"No such certificate found 'ID = {v}'.")
624
            return E_NO_CERTIFICATES_FOUND, C_NOT_FOUND
625
        else:
626
            return {"success": True, "data": self.certificate_service.get_public_key_from_certificate(cert)}, C_SUCCESS
627

    
628
    def delete_certificate(self, id):
629
        """
630
        Deletes a certificate identified by ID, including its corresponding subtree (all descendants).
631
        :param id: target certificate ID
632
        :rtype: DeleteResponse
633
        """
634

    
635
        Logger.info(f"\n\t{request.referrer}"
636
                    f"\n\t{request.method}   {request.path}   {request.scheme}"
637
                    f"\n\tCertificate ID = {id}")
638

    
639
        try:
640
            v = int(id)
641
        except ValueError:
642
            Logger.error(f"Invalid request, wrong parameters 'id'[{id}].")
643
            return E_WRONG_PARAMETERS, C_BAD_REQUEST
644

    
645
        try:
646
            self.certificate_service.delete_certificate(v)
647
        except CertificateNotFoundException:
648
            Logger.error(f"No such certificate found 'ID = {v}'.")
649
            return E_NO_CERTIFICATES_FOUND, C_NOT_FOUND
650
        except DatabaseException:
651
            Logger.error(f"Internal server error (corrupted database).")
652
            return E_CORRUPTED_DATABASE, C_INTERNAL_SERVER_ERROR
653
        except CertificateStatusInvalidException or RevocationReasonInvalidException or UnknownException:
654
            Logger.error(f"Internal server error (unknown origin).")
655
            return E_GENERAL_ERROR, C_INTERNAL_SERVER_ERROR
656

    
657
        return {"success": True, "data": "The certificate and its descendants have been successfully deleted."}
(2-2/4)