Projekt

Obecné

Profil

Stáhnout (29.2 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, Response
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
#  responsibility.
19
from src.services.key_service import KeyService
20
from src.utils.logger import Logger
21
from src.utils.util import dict_to_string
22

    
23
TREE_NODE_TYPE_COUNT = 3
24

    
25
FILTERING = "filtering"
26
ISSUER = "issuer"
27
US = "usage"
28
NOT_AFTER = "notAfter"
29
NOT_BEFORE = "notBefore"
30
COMMON_NAME = "CN"
31
ID = "id"
32
USAGE = "usage"
33
SUBJECT = "subject"
34
VALIDITY_DAYS = "validityDays"
35
CA = "CA"
36
ISSUED_BY = "issuedby"
37
STATUS = "status"
38
REASON = "reason"
39
REASON_UNDEFINED = "unspecified"
40
NAME = "name"
41
PASSWORD = "password"
42

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

    
56

    
57
class CertController:
58
    KEY_MAP = {'CA': CA_ID, 'SSL': SSL_ID, 'digitalSignature': SIGNATURE_ID, 'authentication': AUTHENTICATION_ID}
59
    INVERSE_KEY_MAP = {k: v for v, k in KEY_MAP.items()}
60

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

    
66
    def create_certificate(self):
67
        """create new certificate
68

    
69
        Create a new certificate based on given information
70

    
71
        :param body: Certificate data to be created
72
        :type body: dict | bytes
73

    
74
        :rtype: CreatedResponse
75
        """
76

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

    
80
        required_keys = {SUBJECT, USAGE, VALIDITY_DAYS}                             # required fields of the POST req
81

    
82
        if request.is_json:                                                         # accept JSON only
83
            body = request.get_json()
84

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

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

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

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

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

    
102
            usages_dict = {}
103

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

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

    
114
            key = self.key_service.create_new_key()                                      # TODO pass key
115

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

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

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

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

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

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

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

    
165
    def get_certificate_by_id(self, id):
166
        """get certificate by ID
167

    
168
        Get certificate in PEM format by ID
169

    
170
        :param id: ID of a certificate to be queried
171
        :type id: dict | bytes
172

    
173
        :rtype: PemResponse
174
        """
175

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

    
185
        cert = self.certificate_service.get_certificate(v)
186

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

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

    
196
        Get certificate details by ID
197

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

    
201
        :rtype: CertificateResponse
202
        """
203

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

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

    
214
        cert = self.certificate_service.get_certificate(v)
215

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

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

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

    
230
        return {"success": True, "data": data}, C_SUCCESS
231

    
232
    def get_certificate_list(self):
233
        """get list of certificates
234

    
235
        Lists certificates based on provided filtering options
236

    
237
        :param filtering: Filter certificate type to be queried
238
        :type filtering: dict | bytes
239

    
240
        :rtype: CertificateListResponse
241
        """
242

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

    
246
        targets = {ROOT_CA_ID, INTERMEDIATE_CA_ID, CERTIFICATE_ID}                  # all targets
247
        issuer_id = -1
248

    
249
        # the filtering parameter can be read as URL argument or as a request body
250
        if request.is_json or FILTERING in request.args.keys():                     # if the request carries JSON data
251
            if request.is_json:
252
                data = request.get_json()                                           # get it
253
            else:
254
                try:
255
                    data = {FILTERING: json.loads(request.args[FILTERING])}
256
                except JSONDecodeError:
257
                    Logger.error(f"The request must be JSON-formatted.")
258
                    return E_NOT_JSON_FORMAT, C_BAD_REQUEST
259

    
260
            Logger.info(f"\n\tRequest body:"
261
                        f"\n{dict_to_string(data)}")
262

    
263
            if FILTERING in data:                                                   # if the 'filtering' field exists
264
                if isinstance(data[FILTERING], dict):                               # and it is also a 'dict'
265
                    if CA in data[FILTERING]:                                       # containing 'CA'
266
                        if isinstance(data[FILTERING][CA], bool):                   # which is a 'bool',
267
                            if data[FILTERING][CA]:                                 # then filter according to 'CA'.
268
                                targets.remove(CERTIFICATE_ID)
269
                            else:
270
                                targets.remove(ROOT_CA_ID)
271
                                targets.remove(INTERMEDIATE_CA_ID)
272
                        else:
273
                            Logger.error(f"Invalid request, wrong parameters '{FILTERING}.{CA}'.")
274
                            return E_WRONG_PARAMETERS, C_BAD_REQUEST
275
                    if ISSUED_BY in data[FILTERING]:                                # containing 'issuedby'
276
                        if isinstance(data[FILTERING][ISSUED_BY], int):             # which is an 'int'
277
                            issuer_id = data[FILTERING][ISSUED_BY]                  # then get its children only
278
                else:
279
                    Logger.error(f"Invalid request, wrong parameters '{FILTERING}'.")
280
                    return E_WRONG_PARAMETERS, C_BAD_REQUEST
281
            if issuer_id >= 0:                                                      # if filtering by an issuer
282
                try:
283
                    children = self.certificate_service.get_certificates_issued_by(issuer_id)  # get his children
284
                except CertificateNotFoundException:                                # if id does not exist
285
                    Logger.error(f"No such certificate found 'ID = {issuer_id}'.")
286
                    return E_NO_CERTIFICATES_FOUND, C_NOT_FOUND                     # throw
287

    
288
                certs = [child for child in children if child.type_id in targets]
289

    
290
            elif len(targets) == TREE_NODE_TYPE_COUNT:                              # = 3 -> root node,
291
                                                                                    # intermediate node,
292
                                                                                    # or leaf node
293

    
294
                                                                                    # if filtering did not change the
295
                                                                                    # targets,
296
                certs = self.certificate_service.get_certificates()                 # fetch everything
297
            else:                                                                   # otherwise fetch targets only
298
                certs = list(chain(*(self.certificate_service.get_certificates(target) for target in targets)))
299
        else:
300
            certs = self.certificate_service.get_certificates()                     # if no params, fetch everything
301

    
302
        if certs is None:
303
            Logger.error(f"Internal server error (unknown origin).")
304
            return E_GENERAL_ERROR, C_INTERNAL_SERVER_ERROR
305
        elif len(certs) == 0:
306
            # TODO check log level
307
            Logger.warning(f"No such certificate found (empty list).")
308
            return E_NO_CERTIFICATES_FOUND, C_NO_DATA
309
        else:
310
            ret = []
311
            for c in certs:
312
                data = self.cert_to_dict_partial(c)
313
                if data is None:
314
                    Logger.error(f"Internal server error (corrupted database).")
315
                    return E_CORRUPTED_DATABASE, C_INTERNAL_SERVER_ERROR
316
                ret.append(
317
                    data
318
                )
319
            return {"success": True, "data": ret}, C_SUCCESS
320

    
321
    def get_certificate_root_by_id(self, id):
322
        """get certificate's root of trust chain by ID
323

    
324
        Get certificate's root of trust chain in PEM format by ID
325

    
326
        :param id: ID of a child certificate whose root is to be queried
327
        :type id: dict | bytes
328

    
329
        :rtype: PemResponse
330
        """
331

    
332
        Logger.info(f"\n\t{request.referrer}"
333
                    f"\n\t{request.method}   {request.path}   {request.scheme}"
334
                    f"\n\tCertificate ID = {id}")
335

    
336
        try:
337
            v = int(id)
338
        except ValueError:
339
            Logger.error(f"Invalid request, wrong parameters 'id'[{id}].")
340
            return E_WRONG_PARAMETERS, C_BAD_REQUEST
341

    
342
        cert = self.certificate_service.get_certificate(v)
343

    
344
        if cert is None:
345
            Logger.error(f"No such certificate found 'ID = {v}'.")
346
            return E_NO_CERTIFICATES_FOUND, C_NO_DATA
347

    
348
        trust_chain = self.certificate_service.get_chain_of_trust(cert.parent_id, exclude_root=False)
349
        if trust_chain is None or len(trust_chain) == 0:
350
            Logger.error(f"No such certificate found (empty list).")
351
            return E_NO_CERTIFICATES_FOUND, C_NO_DATA
352

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

    
355
    def get_certificate_trust_chain_by_id(self, id):
356
        """get certificate's trust chain by ID (including root certificate)
357

    
358
        Get certificate trust chain in PEM format by ID
359

    
360
        :param id: ID of a child certificate whose chain is to be queried
361
        :type id: dict | bytes
362

    
363
        :rtype: PemResponse
364
        """
365

    
366
        Logger.info(f"\n\t{request.referrer}"
367
                    f"\n\t{request.method}   {request.path}   {request.scheme}"
368
                    f"\n\tCertificate ID = {id}")
369

    
370
        try:
371
            v = int(id)
372
        except ValueError:
373
            Logger.error(f"Invalid request, wrong parameters 'id'[{id}].")
374
            return E_WRONG_PARAMETERS, C_BAD_REQUEST
375

    
376
        cert = self.certificate_service.get_certificate(v)
377

    
378
        if cert is None:
379
            Logger.error(f"No such certificate found 'ID = {v}'.")
380
            return E_NO_CERTIFICATES_FOUND, C_NO_DATA
381

    
382
        if cert.parent_id is None:
383
            Logger.error(f"Parent ID is empty in certificate 'ID = {v}'.")
384
            return E_NO_CERTIFICATES_FOUND, C_NO_DATA
385

    
386
        trust_chain = self.certificate_service.get_chain_of_trust(cert.parent_id, exclude_root=False)
387

    
388
        ret = []
389
        for intermediate in trust_chain:
390
            ret.append(intermediate.pem_data)
391

    
392
        return {"success": True, "data": "".join(ret)}, C_SUCCESS
393

    
394
    def set_certificate_status(self, id):
395
        """
396
        Revoke a certificate given by ID
397
            - revocation request may contain revocation reason
398
            - revocation reason is verified based on the possible predefined values
399
            - if revocation reason is not specified 'undefined' value is used
400
        :param id: Identifier of the certificate to be revoked
401
        :type id: int
402

    
403
        :rtype: SuccessResponse | ErrorResponse (see OpenAPI definition)
404
        """
405

    
406
        Logger.info(f"\n\t{request.referrer}"
407
                    f"\n\t{request.method}   {request.path}   {request.scheme}"
408
                    f"\n\tCertificate ID = {id}")
409

    
410
        required_keys = {STATUS}  # required keys
411

    
412
        # check if the request contains a JSON body
413
        if request.is_json:
414
            request_body = request.get_json()
415

    
416
            Logger.info(f"\n\tRequest body:"
417
                        f"\n{dict_to_string(request_body)}")
418

    
419
            # try to parse certificate identifier -> if it is not int return error 400
420
            try:
421
                identifier = int(id)
422
            except ValueError:
423
                Logger.error(f"Invalid request, wrong parameters 'id'[{id}].")
424
                return E_WRONG_PARAMETERS, C_BAD_REQUEST
425

    
426
            # verify that all required keys are present
427
            if not all(k in request_body for k in required_keys):
428
                Logger.error(f"Invalid request, missing parameters.")
429
                return E_MISSING_PARAMETERS, C_BAD_REQUEST
430

    
431
            # get status and reason from the request
432
            status = request_body[STATUS]
433
            reason = request_body.get(REASON, REASON_UNDEFINED)
434
            try:
435
                # set certificate status using certificate_service
436
                self.certificate_service.set_certificate_revocation_status(identifier, status, reason)
437
            except (RevocationReasonInvalidException, CertificateStatusInvalidException):
438
                # these exceptions are thrown in case invalid status or revocation reason is passed to the controller
439
                Logger.error(f"Invalid request, wrong parameters.")
440
                return E_WRONG_PARAMETERS, C_BAD_REQUEST
441
            except CertificateAlreadyRevokedException:
442
                Logger.error(f"Certificate is already revoked 'ID = {identifier}'.")
443
                return E_NO_CERTIFICATE_ALREADY_REVOKED, C_BAD_REQUEST
444
            except CertificateNotFoundException:
445
                Logger.error(f"No such certificate found 'ID = {identifier}'.")
446
                return E_NO_CERTIFICATES_FOUND, C_NOT_FOUND
447
            return {"success": True,
448
                    "data": "Certificate status updated successfully."}, C_SUCCESS
449
        # throw an error in case the request does not contain a json body
450
        else:
451
            Logger.error(f"The request must be JSON-formatted.")
452
            return E_NOT_JSON_FORMAT, C_BAD_REQUEST
453

    
454
    def cert_to_dict_partial(self, c):
455
        """
456
        Dictionarizes a certificate directly fetched from the database. Contains partial information.
457
        :param c: target cert
458
        :return: certificate dict (compliant with some parts of the REST API)
459
        """
460

    
461
        # TODO check log
462
        Logger.debug(f"Function launched.")
463

    
464
        c_issuer = self.certificate_service.get_certificate(c.parent_id)
465
        if c_issuer is None:
466
            return None
467

    
468
        return {
469
            ID: c.certificate_id,
470
            COMMON_NAME: c.common_name,
471
            NOT_BEFORE: datetime.strptime(c.valid_from, DATETIME_FORMAT).date(),
472
            NOT_AFTER: datetime.strptime(c.valid_to, DATETIME_FORMAT).date(),
473
            USAGE: {CertController.INVERSE_KEY_MAP[k]: v for k, v in c.usages.items()},
474
            ISSUER: {
475
                ID: c_issuer.certificate_id,
476
                COMMON_NAME: c_issuer.common_name
477
            }
478
        }
479

    
480
    def cert_to_dict_full(self, c):
481
        """
482
        Dictionarizes a certificate directly fetched from the database, but adds subject info.
483
        Contains full information.
484
        :param c: target cert
485
        :return: certificate dict (compliant with some parts of the REST API)
486
        """
487

    
488
        Logger.info(f"Function launched.")
489

    
490
        subj = self.certificate_service.get_subject_from_certificate(c)
491
        c_issuer = self.certificate_service.get_certificate(c.parent_id)
492
        if c_issuer is None:
493
            return None
494

    
495
        return {
496
            SUBJECT: subj.to_dict(),
497
            NOT_BEFORE: datetime.strptime(c.valid_from, DATETIME_FORMAT).date(),
498
            NOT_AFTER: datetime.strptime(c.valid_to, DATETIME_FORMAT).date(),
499
            USAGE: {CertController.INVERSE_KEY_MAP[k]: v for k, v in c.usages.items()},
500
            CA: c_issuer.certificate_id
501
        }
502

    
503
    def get_private_key_of_a_certificate(self, id):
504
        """
505
        Get a private key used to sign a certificate in PEM format specified by certificate's ID
506

    
507
        :param id: ID of a certificate whose private key is to be queried
508
        :type id: dict | bytes
509

    
510
        :rtype: PemResponse
511
        """
512

    
513
        Logger.info(f"\n\t{request.referrer}"
514
                    f"\n\t{request.method}   {request.path}   {request.scheme}"
515
                    f"\n\tCertificate ID = {id}")
516

    
517
        # try to parse the supplied ID
518
        try:
519
            v = int(id)
520
        except ValueError:
521
            Logger.error(f"Invalid request, wrong parameters 'id'[{id}].")
522
            return E_WRONG_PARAMETERS, C_BAD_REQUEST
523

    
524
        # find a certificate using the given ID
525
        cert = self.certificate_service.get_certificate(v)
526

    
527
        if cert is None:
528
            Logger.error(f"No such certificate found 'ID = {v}'.")
529
            return E_NO_CERTIFICATES_FOUND, C_NOT_FOUND
530
        else:
531
            # certificate exists, fetch it's private key
532
            private_key = self.key_service.get_key(cert.private_key_id)
533
            if cert is None:
534
                Logger.error(f"Internal server error (certificate's private key cannot be found).")
535
                return E_NO_CERT_PRIVATE_KEY_FOUND, C_INTERNAL_SERVER_ERROR
536
            else:
537
                return {"success": True, "data": private_key.private_key}, C_SUCCESS
538

    
539
    def get_public_key_of_a_certificate(self, id):
540
        """
541
        Get a public key of a certificate in PEM format specified by certificate's ID
542

    
543
        :param id: ID of a certificate whose public key is to be queried
544
        :type id: dict | bytes
545

    
546
        :rtype: PemResponse
547
        """
548

    
549
        Logger.info(f"\n\t{request.referrer}"
550
                    f"\n\t{request.method}   {request.path}   {request.scheme}"
551
                    f"\n\tCertificate ID = {id}")
552

    
553
        # try to parse the supplied ID
554
        try:
555
            v = int(id)
556
        except ValueError:
557
            Logger.error(f"Invalid request, wrong parameters 'id'[{id}].")
558
            return E_WRONG_PARAMETERS, C_BAD_REQUEST
559

    
560
        # find a certificate using the given ID
561
        cert = self.certificate_service.get_certificate(v)
562

    
563
        if cert is None:
564
            Logger.error(f"No such certificate found 'ID = {v}'.")
565
            return E_NO_CERTIFICATES_FOUND, C_NOT_FOUND
566
        else:
567
            return {"success": True, "data": self.certificate_service.get_public_key_from_certificate(cert)}, C_SUCCESS
568

    
569
    def delete_certificate(self, id):
570
        """
571
        Deletes a certificate identified by ID, including its corresponding subtree (all descendants).
572
        :param id: target certificate ID
573
        :rtype: DeleteResponse
574
        """
575

    
576
        Logger.info(f"\n\t{request.referrer}"
577
                    f"\n\t{request.method}   {request.path}   {request.scheme}"
578
                    f"\n\tCertificate ID = {id}")
579

    
580
        try:
581
            v = int(id)
582
        except ValueError:
583
            Logger.error(f"Invalid request, wrong parameters 'id'[{id}].")
584
            return E_WRONG_PARAMETERS, C_BAD_REQUEST
585

    
586
        try:
587
            self.certificate_service.delete_certificate(v)
588
        except CertificateNotFoundException:
589
            Logger.error(f"No such certificate found 'ID = {v}'.")
590
            return E_NO_CERTIFICATES_FOUND, C_NOT_FOUND
591
        except DatabaseException:
592
            Logger.error(f"Internal server error (corrupted database).")
593
            return E_CORRUPTED_DATABASE, C_INTERNAL_SERVER_ERROR
594
        except CertificateStatusInvalidException or RevocationReasonInvalidException or UnknownException:
595
            Logger.error(f"Internal server error (unknown origin).")
596
            return E_GENERAL_ERROR, C_INTERNAL_SERVER_ERROR
597

    
598
        return {"success": True, "data": "The certificate and its descendants have been successfully deleted."}
599

    
600
    def generate_certificate_pkcs_identity(self, id):
601
        """
602
        Generates a PKCS12 identity (including the chain of trust) of the certificate given by the specified ID.
603
        Response is of application/x-pkcs12 type.
604

    
605
        :param id: ID of a certificate whose PKCS12 identity should be generated
606
        :type id: int
607

    
608
        :rtype: Response
609
        """
610

    
611
        Logger.info(f"\n\t{request.referrer}"
612
                    f"\n\t{request.method}   {request.path}   {request.scheme}"
613
                    f"\n\tCertificate ID = {id}")
614

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

    
622
        # find a certificate using the given ID
623
        cert = self.certificate_service.get_certificate(v)
624

    
625
        if request.is_json:                                                         # accept JSON only
626
            body = request.get_json()
627

    
628
            # check whether the request is well formed meaning that it contains all required fields
629
            if NAME not in body.keys():
630
                return E_IDENTITY_NAME_NOT_SPECIFIED, C_BAD_REQUEST
631

    
632
            if PASSWORD not in body.keys():
633
                return E_IDENTITY_PASSWORD_NOT_SPECIFIED, C_BAD_REQUEST
634

    
635
            # parse required fields from the request
636
            identity_name = body[NAME]
637
            identity_password = body[PASSWORD]
638

    
639
            # check whether a certificated specified by the given ID exists
640
            if cert is None:
641
                Logger.error(f"No such certificate found 'ID = {v}'.")
642
                return E_NO_CERTIFICATES_FOUND, C_NOT_FOUND
643
            else:
644
                # try to load it's private key
645
                key = self.key_service.get_key(cert.private_key_id)
646
                if key is None:
647
                    Logger.error(
648
                        f"The private key 'ID = {cert.private_key_id}'of the certificate 'ID = {cert.certificate_id}' does not exist.")
649
                    return E_NO_CERTIFICATES_FOUND, C_INTERNAL_SERVER_ERROR
650
                else:
651
                    # generate PKCS12 identity
652
                    identity_byte_array = self.certificate_service.generate_pkcs_identity(cert.certificate_id, key,
653
                                                                                          identity_name,
654
                                                                                          identity_password)
655
                    return Response(identity_byte_array, mimetype='application/x-pkcs12')
(2-2/4)