Projekt

Obecné

Profil

Stáhnout (33.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
    CertificateCannotBeSetToValid
19
#  responsibility.
20
from src.services.cryptography import CryptographyException
21
from src.services.key_service import KeyService
22
from src.utils.logger import Logger
23
from src.utils.util import dict_to_string
24

    
25
EXTENSIONS = "extensions"
26

    
27
TREE_NODE_TYPE_COUNT = 3
28
KEY_PEM = "key_pem"
29
PASSWORD = "password"
30
KEY = "key"
31
FILTERING = "filtering"
32
PAGE = "page"
33
PER_PAGE = "per_page"
34
ISSUER = "issuer"
35
US = "usage"
36
NOT_AFTER = "notAfter"
37
NOT_BEFORE = "notBefore"
38
COMMON_NAME = "CN"
39
ID = "id"
40
CA = "CA"
41
USAGE = "usage"
42
SUBJECT = "subject"
43
VALIDITY_DAYS = "validityDays"
44
TYPE = "type"
45
ISSUED_BY = "issuedby"
46
STATUS = "status"
47
REASON = "reason"
48
REASON_UNDEFINED = "unspecified"
49
NAME = "name"
50
PASSWORD = "password"
51

    
52
E_NO_ISSUER_FOUND = {"success": False, "data": "No certificate authority with such unique ID exists."}
53
E_NO_CERTIFICATES_FOUND = {"success": False, "data": "No such certificate found."}
54
E_NO_CERTIFICATE_ALREADY_REVOKED = {"success": False, "data": "Certificate is already revoked."}
55
E_NO_CERT_PRIVATE_KEY_FOUND = {"success": False,
56
                               "data": "Internal server error (certificate's private key cannot be found)."}
57
E_NOT_JSON_FORMAT = {"success": False, "data": "The request must be JSON-formatted."}
58
E_CORRUPTED_DATABASE = {"success": False, "data": "Internal server error (corrupted database)."}
59
E_GENERAL_ERROR = {"success": False, "data": "Internal server error (unknown origin)."}
60
E_MISSING_PARAMETERS = {"success": False, "data": "Invalid request, missing parameters."}
61
E_WRONG_PARAMETERS = {"success": False, "data": "Invalid request, wrong parameters."}
62
E_WRONG_PASSWORD = {"success": False, "data": "The provided passphrase does not match the provided key."}
63
E_IDENTITY_NAME_NOT_SPECIFIED = {"success": False, "data": "Invalid request, missing identity name."}
64
E_IDENTITY_PASSWORD_NOT_SPECIFIED = {"success": False, "data": "Invalid request, missing identity password."}
65

    
66

    
67
class CertController:
68
    USAGE_KEY_MAP = {'CA': CA_ID, 'SSL': SSL_ID, 'digitalSignature': SIGNATURE_ID, 'authentication': AUTHENTICATION_ID}
69
    INVERSE_USAGE_KEY_MAP = {k: v for v, k in USAGE_KEY_MAP.items()}
70
    FILTERING_TYPE_KEY_MAP = {'root': ROOT_CA_ID, 'inter': INTERMEDIATE_CA_ID, 'end': CERTIFICATE_ID}
71
    # INVERSE_FILTERING_TYPE_KEY_MAP = {k: v for v, k in FILTERING_TYPE_KEY_MAP.items()}
72

    
73

    
74
    @inject
75
    def __init__(self, certificate_service: CertificateService, key_service: KeyService):
76
        self.certificate_service = certificate_service
77
        self.key_service = key_service
78

    
79
    def create_certificate(self):
80
        """create new certificate
81

    
82
        Create a new certificate based on given information
83

    
84
        :param body: Certificate data to be created
85
        :type body: dict | bytes
86

    
87
        :rtype: CreatedResponse
88
        """
89

    
90
        Logger.info(f"\n\t{request.referrer}"
91
                    f"\n\t{request.method}   {request.path}   {request.scheme}")
92

    
93
        required_keys = {SUBJECT, USAGE, VALIDITY_DAYS}                             # required fields of the POST req
94

    
95
        if request.is_json:                                                         # accept JSON only
96
            body = request.get_json()
97

    
98
            Logger.info(f"\n\tRequest body:"
99
                        f"\n{dict_to_string(body)}")
100

    
101
            if not all(k in body for k in required_keys):                           # verify that all keys are present
102
                Logger.error(f"Invalid request, missing parameters")
103
                return E_MISSING_PARAMETERS, C_BAD_REQUEST
104

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

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

    
111
            if subject is None:                                                     # if the format is incorrect
112
                Logger.error(f"Invalid request, wrong parameter '{SUBJECT}'.")
113
                return E_WRONG_PARAMETERS, C_BAD_REQUEST
114

    
115
            usages_dict = {}
116

    
117
            if USAGE not in body or not isinstance(body[USAGE], list):              # type checking
118
                Logger.error(f"Invalid request, wrong parameter '{USAGE}'.")
119
                return E_WRONG_PARAMETERS, C_BAD_REQUEST
120

    
121
            for v in body[USAGE]:                                                   # for each usage
122
                if v not in CertController.USAGE_KEY_MAP:                           # check that it is a valid usage
123
                    Logger.error(f"Invalid request, wrong parameter '{USAGE}'[{v}].")
124
                    return E_WRONG_PARAMETERS, C_BAD_REQUEST                        # and throw if it is not
125
                usages_dict[CertController.USAGE_KEY_MAP[v]] = True                 # otherwise translate key and set
126

    
127
            if KEY in body:
128
                if isinstance(body[KEY], dict):
129
                    if PASSWORD in body[KEY]:
130
                        passphrase = body[KEY][PASSWORD]
131
                        if KEY_PEM in body[KEY]:
132
                            key_pem = body[KEY][KEY_PEM]
133
                            if not self.key_service.verify_key(key_pem, passphrase=passphrase):
134
                                Logger.error(f"Passphrase specified but invalid.")
135
                                return E_WRONG_PASSWORD, C_BAD_REQUEST
136
                            key = self.key_service.wrap_custom_key(key_pem, passphrase=passphrase)
137
                        else:
138
                            key = self.key_service.create_new_key(passphrase)
139
                    else:
140
                        if KEY_PEM in body[KEY]:
141
                            key_pem = body[KEY][KEY_PEM]
142
                            if not self.key_service.verify_key(key_pem, passphrase=None):
143
                                Logger.error("Passphrase ommited but required.")
144
                                return E_WRONG_PASSWORD, C_BAD_REQUEST
145
                            key = self.key_service.wrap_custom_key(key_pem, passphrase=None)
146
                        else:
147
                            return E_WRONG_PARAMETERS, C_BAD_REQUEST                # if "key" exists but is empty
148
                else:
149
                    return E_WRONG_PARAMETERS, C_BAD_REQUEST
150
            else:
151
                key = self.key_service.create_new_key()                             # if "key" does not exist
152

    
153
            if CA not in body or body[CA] is None:                                  # if issuer omitted (legal) or none
154
                cert = self.certificate_service.create_root_ca(                     # create a root CA
155
                    key,
156
                    subject,
157
                    usages=usages_dict,                                             # TODO ignoring usages -> discussion
158
                    days=body[VALIDITY_DAYS]
159
                )
160
            else:
161
                issuer = self.certificate_service.get_certificate(body[CA])         # get base issuer info
162

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

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

    
170
                if issuer_key is None:                                              # if it does not
171
                    Logger.error(f"Internal server error (corrupted database).")
172
                    self.key_service.delete_key(key.private_key_id)                 # free
173
                    return E_CORRUPTED_DATABASE, C_INTERNAL_SERVER_ERROR            # and throw
174

    
175
                f = self.certificate_service.create_ca if CA_ID in usages_dict and usages_dict[CA_ID] else \
176
                    self.certificate_service.create_end_cert
177

    
178
                # noinspection PyTypeChecker
179
                cert = f(                                                           # create inter CA or end cert
180
                    key,                                                            # according to whether 'CA' is among
181
                    subject,                                                        # the usages' fields
182
                    issuer,
183
                    issuer_key,
184
                    usages=usages_dict,
185
                    days=body[VALIDITY_DAYS]
186
                )
187

    
188
            if cert is not None:
189
                return {"success": True,
190
                        "data": cert.certificate_id}, C_CREATED_SUCCESSFULLY
191
            else:                                                                   # if this fails, then
192
                Logger.error(f"Internal error: The certificate could not have been created.")
193
                self.key_service.delete_key(key.private_key_id)                     # free
194
                return {"success": False,                                           # and wonder what the cause is,
195
                        "data": "Internal error: The certificate could not have been created."}, C_BAD_REQUEST
196
                                                                                    # as obj/None carries only one bit
197
                                                                                    # of error information
198
        else:
199
            Logger.error(f"The request must be JSON-formatted.")
200
            return E_NOT_JSON_FORMAT, C_BAD_REQUEST                                 # throw in case of non-JSON format
201

    
202
    def get_certificate_by_id(self, id):
203
        """get certificate by ID
204

    
205
        Get certificate in PEM format by ID
206

    
207
        :param id: ID of a certificate to be queried
208
        :type id: dict | bytes
209

    
210
        :rtype: PemResponse
211
        """
212

    
213
        Logger.info(f"\n\t{request.referrer}"
214
                    f"\n\t{request.method}   {request.path}   {request.scheme}"
215
                    f"\n\tCertificate ID = {id}")
216
        try:
217
            v = int(id)
218
        except ValueError:
219
            Logger.error(f"Invalid request, wrong parameters 'id'[{id}].")
220
            return E_WRONG_PARAMETERS, C_BAD_REQUEST
221

    
222
        cert = self.certificate_service.get_certificate(v)
223

    
224
        if cert is None:
225
            Logger.error(f"No such certificate found 'ID = {v}'.")
226
            return E_NO_CERTIFICATES_FOUND, C_NO_DATA
227
        else:
228
            return {"success": True, "data": cert.pem_data}, C_SUCCESS
229

    
230
    def get_certificate_details_by_id(self, id):
231
        """get certificate's details by ID
232

    
233
        Get certificate details by ID
234

    
235
        :param id: ID of a certificate whose details are to be queried
236
        :type id: dict | bytes
237

    
238
        :rtype: CertificateResponse
239
        """
240

    
241
        Logger.info(f"\n\t{request.referrer}"
242
                    f"\n\t{request.method}   {request.path}   {request.scheme}"
243
                    f"\n\tCertificate ID = {id}")
244

    
245
        try:
246
            v = int(id)
247
        except ValueError:
248
            Logger.error(f"Invalid request, wrong parameters 'id'[{id}].")
249
            return E_WRONG_PARAMETERS, C_BAD_REQUEST
250

    
251
        cert = self.certificate_service.get_certificate(v)
252

    
253
        if cert is None:
254
            Logger.error(f"No such certificate found 'ID = {v}'.")
255
            return E_NO_CERTIFICATES_FOUND, C_NOT_FOUND
256

    
257
        data = self.cert_to_dict_full(cert)
258
        if data is None:
259
            return E_CORRUPTED_DATABASE, C_INTERNAL_SERVER_ERROR
260

    
261
        try:
262
            state = self.certificate_service.get_certificate_state(v)
263
            data["status"] = state
264
        except CertificateNotFoundException:
265
            Logger.error(f"No such certificate found 'ID = {id}'.")
266

    
267
        return {"success": True, "data": data}, C_SUCCESS
268

    
269
    def get_certificate_list(self):
270
        """get list of certificates
271

    
272
        Lists certificates based on provided filtering options
273

    
274
        :param filtering: Filter certificate type to be queried
275
        :type filtering: dict | bytes
276

    
277
        :rtype: CertificateListResponse
278
        """
279

    
280
        Logger.info(f"\n\t{request.referrer}"
281
                    f"\n\t{request.method}   {request.path}   {request.scheme}")
282

    
283

    
284
        # the filtering parameter can be read as URL argument or as a request body
285
        if request.is_json:
286
            data = request.get_json()
287
        else:
288
            data = {}
289

    
290
        if FILTERING in request.args.keys():
291
            try:
292
                data[FILTERING] = json.loads(request.args[FILTERING])
293
            except JSONDecodeError:
294
                Logger.error(f"The request must be JSON-formatted.")
295
                return E_NOT_JSON_FORMAT, C_BAD_REQUEST
296

    
297
        if PAGE in request.args.keys():
298
            try:
299
                data[PAGE] = json.loads(request.args[PAGE])
300
            except JSONDecodeError:
301
                Logger.error(f"The request must be JSON-formatted.")
302
                return E_NOT_JSON_FORMAT, C_BAD_REQUEST
303

    
304
        if PER_PAGE in request.args.keys():
305
            try:
306
                data[PER_PAGE] = json.loads(request.args[PER_PAGE])
307
            except JSONDecodeError:
308
                Logger.error(f"The request must be JSON-formatted.")
309
                return E_NOT_JSON_FORMAT, C_BAD_REQUEST
310

    
311
        Logger.info(f"\n\tRequest body:"
312
                    f"\n{dict_to_string(data)}")
313

    
314
        target_types = {ROOT_CA_ID, INTERMEDIATE_CA_ID, CERTIFICATE_ID}
315
        target_usages = {v for v in CertController.INVERSE_USAGE_KEY_MAP.keys()}
316
        target_cn_substring = None
317
        issuer_id = -1
318

    
319
        unfiltered = True
320

    
321
        if PER_PAGE in data:
322
            unfiltered = False
323
            page = data.get(PAGE, 0)
324
            per_page = data[PER_PAGE]
325
        else:
326
            page = None
327
            per_page = None
328

    
329
        if FILTERING in data:                                                   # if the 'filtering' field exists
330
            unfiltered = False
331
            if isinstance(data[FILTERING], dict):                               # and it is also a 'dict'
332

    
333
                # noinspection DuplicatedCode
334
                if TYPE in data[FILTERING]:                                     # containing 'type'
335
                    if isinstance(data[FILTERING][TYPE], list):                 # which is a 'list',
336
                                                                                # map every field to id
337
                        try:
338
                            target_types = {CertController.FILTERING_TYPE_KEY_MAP[v] for v in data[FILTERING][TYPE]}
339
                        except KeyError as e:
340
                            Logger.error(f"Invalid request, wrong parameters '{FILTERING}.{TYPE}' - '{e}'.")
341
                            return E_WRONG_PARAMETERS, C_BAD_REQUEST
342
                    else:
343
                        Logger.error(f"Invalid request, wrong parameters '{FILTERING}.{TYPE}'.")
344
                        return E_WRONG_PARAMETERS, C_BAD_REQUEST
345

    
346
                # noinspection DuplicatedCode
347
                if USAGE in data[FILTERING]:                                    # containing 'usage'
348
                    if isinstance(data[FILTERING][USAGE], list):                # which is a 'list',
349
                                                                                # map every field to id
350
                        try:
351
                            target_usages = {CertController.USAGE_KEY_MAP[v] for v in data[FILTERING][USAGE]}
352
                        except KeyError as e:
353
                            Logger.error(f"Invalid request, wrong parameters '{FILTERING}.{USAGE}' - '{e}'.")
354
                            return E_WRONG_PARAMETERS, C_BAD_REQUEST
355
                    else:
356
                        Logger.error(f"Invalid request, wrong parameters '{FILTERING}.{USAGE}'.")
357
                        return E_WRONG_PARAMETERS, C_BAD_REQUEST
358

    
359
                if COMMON_NAME in data[FILTERING]:                              # containing 'CN'
360
                    if isinstance(data[FILTERING][COMMON_NAME], str):           # which is a 'str'
361
                        target_cn_substring = data[FILTERING][COMMON_NAME]
362
                    else:
363
                        Logger.error(f"Invalid request, wrong parameters '{FILTERING}.{COMMON_NAME}'.")
364
                        return E_WRONG_PARAMETERS, C_BAD_REQUEST
365

    
366
                if ISSUED_BY in data[FILTERING]:                                # containing 'issuedby'
367
                    if isinstance(data[FILTERING][ISSUED_BY], int):             # which is an 'int'
368
                        issuer_id = data[FILTERING][ISSUED_BY]                  # then get its children only
369

    
370
            else:
371
                Logger.error(f"Invalid request, wrong parameters '{FILTERING}'.")
372
                return E_WRONG_PARAMETERS, C_BAD_REQUEST
373

    
374
        if unfiltered:                                                      # if not filtering
375
            certs = self.certificate_service.get_certificates()
376
        elif issuer_id >= 0:                                                # if filtering by an issuer
377
            try:
378
                                                                            # get his children, filtered
379
                certs = self.certificate_service.get_certificates_issued_by_filter(
380
                    issuer_id=issuer_id,
381
                    target_types=target_types,
382
                    target_usages=target_usages,
383
                    target_cn_substring=target_cn_substring,
384
                    page=page,
385
                    per_page=per_page
386
                )
387
            except CertificateNotFoundException:                            # if id does not exist
388
                Logger.error(f"No such certificate found 'ID = {issuer_id}'.")
389
                return E_NO_CERTIFICATES_FOUND, C_NOT_FOUND                 # throw
390
        else:
391
            certs = self.certificate_service.get_certificates_filter(
392
                target_types=target_types,
393
                target_usages=target_usages,
394
                target_cn_substring=target_cn_substring,
395
                page=page,
396
                per_page=per_page
397
            )
398

    
399
        if certs is None:
400
            Logger.error(f"Internal server error (unknown origin).")
401
            return E_GENERAL_ERROR, C_INTERNAL_SERVER_ERROR
402
        elif len(certs) == 0:
403
            # TODO check log level
404
            Logger.warning(f"No such certificate found (empty list).")
405
            return {"success": True, "data": []}, C_SUCCESS
406
        else:
407
            ret = []
408
            for c in certs:
409
                data = self.cert_to_dict_partial(c)
410
                if data is None:
411
                    Logger.error(f"Internal server error (corrupted database).")
412
                    return E_CORRUPTED_DATABASE, C_INTERNAL_SERVER_ERROR
413
                ret.append(
414
                    data
415
                )
416
            return {"success": True, "data": ret}, C_SUCCESS
417

    
418
    def get_certificate_root_by_id(self, id):
419
        """get certificate's root of trust chain by ID
420

    
421
        Get certificate's root of trust chain in PEM format by ID
422

    
423
        :param id: ID of a child certificate whose root is to be queried
424
        :type id: dict | bytes
425

    
426
        :rtype: PemResponse
427
        """
428

    
429
        Logger.info(f"\n\t{request.referrer}"
430
                    f"\n\t{request.method}   {request.path}   {request.scheme}"
431
                    f"\n\tCertificate ID = {id}")
432

    
433
        try:
434
            v = int(id)
435
        except ValueError:
436
            Logger.error(f"Invalid request, wrong parameters 'id'[{id}].")
437
            return E_WRONG_PARAMETERS, C_BAD_REQUEST
438

    
439
        cert = self.certificate_service.get_certificate(v)
440

    
441
        if cert is None:
442
            Logger.error(f"No such certificate found 'ID = {v}'.")
443
            return E_NO_CERTIFICATES_FOUND, C_NO_DATA
444

    
445
        trust_chain = self.certificate_service.get_chain_of_trust(cert.parent_id, exclude_root=False)
446
        if trust_chain is None or len(trust_chain) == 0:
447
            Logger.error(f"No such certificate found (empty list).")
448
            return E_NO_CERTIFICATES_FOUND, C_NO_DATA
449

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

    
452
    def get_certificate_trust_chain_by_id(self, id):
453
        """get certificate's trust chain by ID (including root certificate)
454

    
455
        Get certificate trust chain in PEM format by ID
456

    
457
        :param id: ID of a child certificate whose chain is to be queried
458
        :type id: dict | bytes
459

    
460
        :rtype: PemResponse
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
        try:
468
            v = int(id)
469
        except ValueError:
470
            Logger.error(f"Invalid request, wrong parameters 'id'[{id}].")
471
            return E_WRONG_PARAMETERS, C_BAD_REQUEST
472

    
473
        cert = self.certificate_service.get_certificate(v)
474

    
475
        if cert is None:
476
            Logger.error(f"No such certificate found 'ID = {v}'.")
477
            return E_NO_CERTIFICATES_FOUND, C_NO_DATA
478

    
479
        if cert.parent_id is None:
480
            Logger.error(f"Parent ID is empty in certificate 'ID = {v}'.")
481
            return E_NO_CERTIFICATES_FOUND, C_NO_DATA
482

    
483
        trust_chain = self.certificate_service.get_chain_of_trust(cert.parent_id, exclude_root=False)
484

    
485
        ret = []
486
        for intermediate in trust_chain:
487
            ret.append(intermediate.pem_data)
488

    
489
        return {"success": True, "data": "".join(ret)}, C_SUCCESS
490

    
491
    def set_certificate_status(self, id):
492
        """
493
        Revoke a certificate given by ID
494
            - revocation request may contain revocation reason
495
            - revocation reason is verified based on the possible predefined values
496
            - if revocation reason is not specified 'undefined' value is used
497
        :param id: Identifier of the certificate to be revoked
498
        :type id: int
499

    
500
        :rtype: SuccessResponse | ErrorResponse (see OpenAPI definition)
501
        """
502

    
503
        Logger.info(f"\n\t{request.referrer}"
504
                    f"\n\t{request.method}   {request.path}   {request.scheme}"
505
                    f"\n\tCertificate ID = {id}")
506

    
507
        required_keys = {STATUS}  # required keys
508

    
509
        # check if the request contains a JSON body
510
        if request.is_json:
511
            request_body = request.get_json()
512

    
513
            Logger.info(f"\n\tRequest body:"
514
                        f"\n{dict_to_string(request_body)}")
515

    
516
            # try to parse certificate identifier -> if it is not int return error 400
517
            try:
518
                identifier = int(id)
519
            except ValueError:
520
                Logger.error(f"Invalid request, wrong parameters 'id'[{id}].")
521
                return E_WRONG_PARAMETERS, C_BAD_REQUEST
522

    
523
            # verify that all required keys are present
524
            if not all(k in request_body for k in required_keys):
525
                Logger.error(f"Invalid request, missing parameters.")
526
                return E_MISSING_PARAMETERS, C_BAD_REQUEST
527

    
528
            # get status and reason from the request
529
            status = request_body[STATUS]
530
            reason = request_body.get(REASON, REASON_UNDEFINED)
531
            try:
532
                # set certificate status using certificate_service
533
                self.certificate_service.set_certificate_revocation_status(identifier, status, reason)
534
            except (RevocationReasonInvalidException, CertificateStatusInvalidException):
535
                # these exceptions are thrown in case invalid status or revocation reason is passed to the controller
536
                Logger.error(f"Invalid request, wrong parameters.")
537
                return E_WRONG_PARAMETERS, C_BAD_REQUEST
538
            except CertificateAlreadyRevokedException:
539
                Logger.error(f"Certificate is already revoked 'ID = {identifier}'.")
540
                return E_NO_CERTIFICATE_ALREADY_REVOKED, C_BAD_REQUEST
541
            except CertificateNotFoundException:
542
                Logger.error(f"No such certificate found 'ID = {identifier}'.")
543
                return E_NO_CERTIFICATES_FOUND, C_NOT_FOUND
544
            except CertificateCannotBeSetToValid as e:
545
                return {"success": False, "data": str(e)}, C_BAD_REQUEST
546
            return {"success": True,
547
                    "data": "Certificate status updated successfully."}, C_SUCCESS
548
        # throw an error in case the request does not contain a json body
549
        else:
550
            Logger.error(f"The request must be JSON-formatted.")
551
            return E_NOT_JSON_FORMAT, C_BAD_REQUEST
552

    
553
    def cert_to_dict_partial(self, c):
554
        """
555
        Dictionarizes a certificate directly fetched from the database. Contains partial information.
556
        :param c: target cert
557
        :return: certificate dict (compliant with some parts of the REST API)
558
        """
559

    
560
        # TODO check log
561
        Logger.debug(f"Function launched.")
562

    
563
        c_issuer = self.certificate_service.get_certificate(c.parent_id)
564
        if c_issuer is None:
565
            return None
566

    
567
        return {
568
            ID: c.certificate_id,
569
            COMMON_NAME: c.common_name,
570
            NOT_BEFORE: datetime.strptime(c.valid_from, DATETIME_FORMAT).date(),
571
            NOT_AFTER: datetime.strptime(c.valid_to, DATETIME_FORMAT).date(),
572
            USAGE: {CertController.INVERSE_USAGE_KEY_MAP[k]: v for k, v in c.usages.items()},
573
            ISSUER: {
574
                ID: c_issuer.certificate_id,
575
                COMMON_NAME: c_issuer.common_name
576
            }
577
        }
578

    
579
    def cert_to_dict_full(self, c):
580
        """
581
        Dictionarizes a certificate directly fetched from the database, but adds subject info.
582
        Contains full information.
583
        :param c: target cert
584
        :return: certificate dict (compliant with some parts of the REST API)
585
        """
586

    
587
        Logger.info(f"Function launched.")
588

    
589
        subj = self.certificate_service.get_subject_from_certificate(c)
590
        c_issuer = self.certificate_service.get_certificate(c.parent_id)
591
        if c_issuer is None:
592
            return None
593

    
594
        return {
595
            SUBJECT: subj.to_dict(),
596
            NOT_BEFORE: datetime.strptime(c.valid_from, DATETIME_FORMAT).date(),
597
            NOT_AFTER: datetime.strptime(c.valid_to, DATETIME_FORMAT).date(),
598
            USAGE: {CertController.INVERSE_USAGE_KEY_MAP[k]: v for k, v in c.usages.items()},
599
            CA: c_issuer.certificate_id
600
        }
601

    
602
    def get_private_key_of_a_certificate(self, id):
603
        """
604
        Get a private key used to sign a certificate in PEM format specified by certificate's ID
605

    
606
        :param id: ID of a certificate whose private key is to be queried
607
        :type id: dict | bytes
608

    
609
        :rtype: PemResponse
610
        """
611

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

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

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

    
626
        if cert is None:
627
            Logger.error(f"No such certificate found 'ID = {v}'.")
628
            return E_NO_CERTIFICATES_FOUND, C_NOT_FOUND
629
        else:
630
            # certificate exists, fetch it's private key
631
            private_key = self.key_service.get_key(cert.private_key_id)
632
            if cert is None:
633
                Logger.error(f"Internal server error (certificate's private key cannot be found).")
634
                return E_NO_CERT_PRIVATE_KEY_FOUND, C_INTERNAL_SERVER_ERROR
635
            else:
636
                return {"success": True, "data": private_key.private_key}, C_SUCCESS
637

    
638
    def get_public_key_of_a_certificate(self, id):
639
        """
640
        Get a public key of a certificate in PEM format specified by certificate's ID
641

    
642
        :param id: ID of a certificate whose public key is to be queried
643
        :type id: dict | bytes
644

    
645
        :rtype: PemResponse
646
        """
647

    
648
        Logger.info(f"\n\t{request.referrer}"
649
                    f"\n\t{request.method}   {request.path}   {request.scheme}"
650
                    f"\n\tCertificate ID = {id}")
651

    
652
        # try to parse the supplied ID
653
        try:
654
            v = int(id)
655
        except ValueError:
656
            Logger.error(f"Invalid request, wrong parameters 'id'[{id}].")
657
            return E_WRONG_PARAMETERS, C_BAD_REQUEST
658

    
659
        # find a certificate using the given ID
660
        cert = self.certificate_service.get_certificate(v)
661

    
662
        if cert is None:
663
            Logger.error(f"No such certificate found 'ID = {v}'.")
664
            return E_NO_CERTIFICATES_FOUND, C_NOT_FOUND
665
        else:
666
            return {"success": True, "data": self.certificate_service.get_public_key_from_certificate(cert)}, C_SUCCESS
667

    
668
    def delete_certificate(self, id):
669
        """
670
        Deletes a certificate identified by ID, including its corresponding subtree (all descendants).
671
        :param id: target certificate ID
672
        :rtype: DeleteResponse
673
        """
674

    
675
        Logger.info(f"\n\t{request.referrer}"
676
                    f"\n\t{request.method}   {request.path}   {request.scheme}"
677
                    f"\n\tCertificate ID = {id}")
678

    
679
        try:
680
            v = int(id)
681
        except ValueError:
682
            Logger.error(f"Invalid request, wrong parameters 'id'[{id}].")
683
            return E_WRONG_PARAMETERS, C_BAD_REQUEST
684

    
685
        try:
686
            self.certificate_service.delete_certificate(v)
687
        except CertificateNotFoundException:
688
            Logger.error(f"No such certificate found 'ID = {v}'.")
689
            return E_NO_CERTIFICATES_FOUND, C_NOT_FOUND
690
        except DatabaseException:
691
            Logger.error(f"Internal server error (corrupted database).")
692
            return E_CORRUPTED_DATABASE, C_INTERNAL_SERVER_ERROR
693
        except CertificateStatusInvalidException or RevocationReasonInvalidException or UnknownException:
694
            Logger.error(f"Internal server error (unknown origin).")
695
            return E_GENERAL_ERROR, C_INTERNAL_SERVER_ERROR
696

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

    
699
    def generate_certificate_pkcs_identity(self, id):
700
        """
701
        Generates a PKCS12 identity (including the chain of trust) of the certificate given by the specified ID.
702
        Response is of application/x-pkcs12 type.
703

    
704
        :param id: ID of a certificate whose PKCS12 identity should be generated
705
        :type id: int
706

    
707
        :rtype: Response
708
        """
709

    
710
        Logger.info(f"\n\t{request.referrer}"
711
                    f"\n\t{request.method}   {request.path}   {request.scheme}"
712
                    f"\n\tCertificate ID = {id}")
713

    
714
        # try to parse the supplied ID
715
        try:
716
            v = int(id)
717
        except ValueError:
718
            Logger.error(f"Invalid request, wrong parameters 'id'[{id}] (expected integer).")
719
            return E_WRONG_PARAMETERS, C_BAD_REQUEST
720

    
721
        # find a certificate using the given ID
722
        cert = self.certificate_service.get_certificate(v)
723

    
724
        if request.is_json:                                                         # accept JSON only
725
            body = request.get_json()
726

    
727
            # check whether the request is well formed meaning that it contains all required fields
728
            if NAME not in body.keys():
729
                return E_IDENTITY_NAME_NOT_SPECIFIED, C_BAD_REQUEST
730

    
731
            if PASSWORD not in body.keys():
732
                return E_IDENTITY_PASSWORD_NOT_SPECIFIED, C_BAD_REQUEST
733

    
734
            # parse required fields from the request
735
            identity_name = body[NAME]
736
            identity_password = body[PASSWORD]
737

    
738
            # check whether a certificated specified by the given ID exists
739
            if cert is None:
740
                Logger.error(f"No such certificate found 'ID = {v}'.")
741
                return E_NO_CERTIFICATES_FOUND, C_NOT_FOUND
742
            else:
743
                # try to load it's private key
744
                key = self.key_service.get_key(cert.private_key_id)
745
                if key is None:
746
                    Logger.error(
747
                        f"The private key 'ID = {cert.private_key_id}'of the certificate 'ID = {cert.certificate_id}' does not exist.")
748
                    return E_NO_CERTIFICATES_FOUND, C_INTERNAL_SERVER_ERROR
749
                else:
750
                    # generate PKCS12 identity
751
                    identity_byte_array = self.certificate_service.generate_pkcs_identity(cert, key,
752
                                                                                          identity_name,
753
                                                                                          identity_password)
754
                    return Response(identity_byte_array, mimetype='application/x-pkcs12')
(2-2/4)