Projekt

Obecné

Profil

Stáhnout (33.3 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
            extensions = ""
154
            if EXTENSIONS in body:
155
                extensions = body[EXTENSIONS]
156

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

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

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

    
175
                if issuer_key is None:                                          # if it does not
176
                    Logger.error(f"Internal server error (corrupted database).")
177
                    self.key_service.delete_key(key.private_key_id)             # free
178
                    return E_CORRUPTED_DATABASE, C_INTERNAL_SERVER_ERROR        # and throw
179

    
180
                f = self.certificate_service.create_ca if CA_ID in usages_dict and usages_dict[CA_ID] else \
181
                    self.certificate_service.create_end_cert
182

    
183
                # noinspection PyArgumentList
184
                cert = f(                                                       # create inter CA or end cert
185
                    key,                                                        # according to whether 'CA' is among
186
                    subject,                                                    # the usages' fields
187
                    issuer,
188
                    issuer_key,
189
                    usages=usages_dict,
190
                    days=body[VALIDITY_DAYS],
191
                    extensions=extensions
192
                )
193

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

    
208
    def get_certificate_by_id(self, id):
209
        """get certificate by ID
210

    
211
        Get certificate in PEM format by ID
212

    
213
        :param id: ID of a certificate to be queried
214
        :type id: dict | bytes
215

    
216
        :rtype: PemResponse
217
        """
218

    
219
        Logger.info(f"\n\t{request.referrer}"
220
                    f"\n\t{request.method}   {request.path}   {request.scheme}"
221
                    f"\n\tCertificate ID = {id}")
222
        try:
223
            v = int(id)
224
        except ValueError:
225
            Logger.error(f"Invalid request, wrong parameters 'id'[{id}].")
226
            return E_WRONG_PARAMETERS, C_BAD_REQUEST
227

    
228
        cert = self.certificate_service.get_certificate(v)
229

    
230
        if cert is None:
231
            Logger.error(f"No such certificate found 'ID = {v}'.")
232
            return E_NO_CERTIFICATES_FOUND, C_NO_DATA
233
        else:
234
            return {"success": True, "data": cert.pem_data}, C_SUCCESS
235

    
236
    def get_certificate_details_by_id(self, id):
237
        """get certificate's details by ID
238

    
239
        Get certificate details by ID
240

    
241
        :param id: ID of a certificate whose details are to be queried
242
        :type id: dict | bytes
243

    
244
        :rtype: CertificateResponse
245
        """
246

    
247
        Logger.info(f"\n\t{request.referrer}"
248
                    f"\n\t{request.method}   {request.path}   {request.scheme}"
249
                    f"\n\tCertificate ID = {id}")
250

    
251
        try:
252
            v = int(id)
253
        except ValueError:
254
            Logger.error(f"Invalid request, wrong parameters 'id'[{id}].")
255
            return E_WRONG_PARAMETERS, C_BAD_REQUEST
256

    
257
        cert = self.certificate_service.get_certificate(v)
258

    
259
        if cert is None:
260
            Logger.error(f"No such certificate found 'ID = {v}'.")
261
            return E_NO_CERTIFICATES_FOUND, C_NOT_FOUND
262

    
263
        data = self.cert_to_dict_full(cert)
264
        if data is None:
265
            return E_CORRUPTED_DATABASE, C_INTERNAL_SERVER_ERROR
266

    
267
        try:
268
            state = self.certificate_service.get_certificate_state(v)
269
            data["status"] = state
270
        except CertificateNotFoundException:
271
            Logger.error(f"No such certificate found 'ID = {id}'.")
272

    
273
        return {"success": True, "data": data}, C_SUCCESS
274

    
275
    def get_certificate_list(self):
276
        """get list of certificates
277

    
278
        Lists certificates based on provided filtering options
279

    
280
        :param filtering: Filter certificate type to be queried
281
        :type filtering: dict | bytes
282

    
283
        :rtype: CertificateListResponse
284
        """
285

    
286
        Logger.info(f"\n\t{request.referrer}"
287
                    f"\n\t{request.method}   {request.path}   {request.scheme}")
288

    
289

    
290
        # the filtering parameter can be read as URL argument or as a request body
291
        if request.is_json:
292
            data = request.get_json()
293
        else:
294
            data = {}
295

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

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

    
310
        if PER_PAGE in request.args.keys():
311
            try:
312
                data[PER_PAGE] = json.loads(request.args[PER_PAGE])
313
            except JSONDecodeError:
314
                Logger.error(f"The request must be JSON-formatted.")
315
                return E_NOT_JSON_FORMAT, C_BAD_REQUEST
316

    
317
        Logger.info(f"\n\tRequest body:"
318
                    f"\n{dict_to_string(data)}")
319

    
320
        target_types = {ROOT_CA_ID, INTERMEDIATE_CA_ID, CERTIFICATE_ID}
321
        target_usages = {v for v in CertController.INVERSE_USAGE_KEY_MAP.keys()}
322
        target_cn_substring = None
323
        issuer_id = -1
324

    
325
        unfiltered = True
326

    
327
        if PER_PAGE in data:
328
            unfiltered = False
329
            page = data.get(PAGE, 0)
330
            per_page = data[PER_PAGE]
331
        else:
332
            page = None
333
            per_page = None
334

    
335
        if FILTERING in data:                                                   # if the 'filtering' field exists
336
            unfiltered = False
337
            if isinstance(data[FILTERING], dict):                               # and it is also a 'dict'
338

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

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

    
365
                if COMMON_NAME in data[FILTERING]:                              # containing 'CN'
366
                    if isinstance(data[FILTERING][COMMON_NAME], str):           # which is a 'str'
367
                        target_cn_substring = data[FILTERING][COMMON_NAME]
368
                    else:
369
                        Logger.error(f"Invalid request, wrong parameters '{FILTERING}.{COMMON_NAME}'.")
370
                        return E_WRONG_PARAMETERS, C_BAD_REQUEST
371

    
372
                if ISSUED_BY in data[FILTERING]:                                # containing 'issuedby'
373
                    if isinstance(data[FILTERING][ISSUED_BY], int):             # which is an 'int'
374
                        issuer_id = data[FILTERING][ISSUED_BY]                  # then get its children only
375

    
376
            else:
377
                Logger.error(f"Invalid request, wrong parameters '{FILTERING}'.")
378
                return E_WRONG_PARAMETERS, C_BAD_REQUEST
379

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

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

    
424
    def get_certificate_root_by_id(self, id):
425
        """get certificate's root of trust chain by ID
426

    
427
        Get certificate's root of trust chain in PEM format by ID
428

    
429
        :param id: ID of a child certificate whose root is to be queried
430
        :type id: dict | bytes
431

    
432
        :rtype: PemResponse
433
        """
434

    
435
        Logger.info(f"\n\t{request.referrer}"
436
                    f"\n\t{request.method}   {request.path}   {request.scheme}"
437
                    f"\n\tCertificate ID = {id}")
438

    
439
        try:
440
            v = int(id)
441
        except ValueError:
442
            Logger.error(f"Invalid request, wrong parameters 'id'[{id}].")
443
            return E_WRONG_PARAMETERS, C_BAD_REQUEST
444

    
445
        cert = self.certificate_service.get_certificate(v)
446

    
447
        if cert is None:
448
            Logger.error(f"No such certificate found 'ID = {v}'.")
449
            return E_NO_CERTIFICATES_FOUND, C_NO_DATA
450

    
451
        trust_chain = self.certificate_service.get_chain_of_trust(cert.parent_id, exclude_root=False)
452
        if trust_chain is None or len(trust_chain) == 0:
453
            Logger.error(f"No such certificate found (empty list).")
454
            return E_NO_CERTIFICATES_FOUND, C_NO_DATA
455

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

    
458
    def get_certificate_trust_chain_by_id(self, id):
459
        """get certificate's trust chain by ID (including root certificate)
460

    
461
        Get certificate trust chain in PEM format by ID
462

    
463
        :param id: ID of a child certificate whose chain is to be queried
464
        :type id: dict | bytes
465

    
466
        :rtype: PemResponse
467
        """
468

    
469
        Logger.info(f"\n\t{request.referrer}"
470
                    f"\n\t{request.method}   {request.path}   {request.scheme}"
471
                    f"\n\tCertificate ID = {id}")
472

    
473
        try:
474
            v = int(id)
475
        except ValueError:
476
            Logger.error(f"Invalid request, wrong parameters 'id'[{id}].")
477
            return E_WRONG_PARAMETERS, C_BAD_REQUEST
478

    
479
        cert = self.certificate_service.get_certificate(v)
480

    
481
        if cert is None:
482
            Logger.error(f"No such certificate found 'ID = {v}'.")
483
            return E_NO_CERTIFICATES_FOUND, C_NO_DATA
484

    
485
        if cert.parent_id is None:
486
            Logger.error(f"Parent ID is empty in certificate 'ID = {v}'.")
487
            return E_NO_CERTIFICATES_FOUND, C_NO_DATA
488

    
489
        trust_chain = self.certificate_service.get_chain_of_trust(cert.parent_id, exclude_root=False)
490

    
491
        ret = []
492
        for intermediate in trust_chain:
493
            ret.append(intermediate.pem_data)
494

    
495
        return {"success": True, "data": "".join(ret)}, C_SUCCESS
496

    
497
    def set_certificate_status(self, id):
498
        """
499
        Revoke a certificate given by ID
500
            - revocation request may contain revocation reason
501
            - revocation reason is verified based on the possible predefined values
502
            - if revocation reason is not specified 'undefined' value is used
503
        :param id: Identifier of the certificate to be revoked
504
        :type id: int
505

    
506
        :rtype: SuccessResponse | ErrorResponse (see OpenAPI definition)
507
        """
508

    
509
        Logger.info(f"\n\t{request.referrer}"
510
                    f"\n\t{request.method}   {request.path}   {request.scheme}"
511
                    f"\n\tCertificate ID = {id}")
512

    
513
        required_keys = {STATUS}  # required keys
514

    
515
        # check if the request contains a JSON body
516
        if request.is_json:
517
            request_body = request.get_json()
518

    
519
            Logger.info(f"\n\tRequest body:"
520
                        f"\n{dict_to_string(request_body)}")
521

    
522
            # try to parse certificate identifier -> if it is not int return error 400
523
            try:
524
                identifier = int(id)
525
            except ValueError:
526
                Logger.error(f"Invalid request, wrong parameters 'id'[{id}].")
527
                return E_WRONG_PARAMETERS, C_BAD_REQUEST
528

    
529
            # verify that all required keys are present
530
            if not all(k in request_body for k in required_keys):
531
                Logger.error(f"Invalid request, missing parameters.")
532
                return E_MISSING_PARAMETERS, C_BAD_REQUEST
533

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

    
559
    def cert_to_dict_partial(self, c):
560
        """
561
        Dictionarizes a certificate directly fetched from the database. Contains partial information.
562
        :param c: target cert
563
        :return: certificate dict (compliant with some parts of the REST API)
564
        """
565

    
566
        # TODO check log
567
        Logger.debug(f"Function launched.")
568

    
569
        c_issuer = self.certificate_service.get_certificate(c.parent_id)
570
        if c_issuer is None:
571
            return None
572

    
573
        return {
574
            ID: c.certificate_id,
575
            COMMON_NAME: c.common_name,
576
            NOT_BEFORE: datetime.strptime(c.valid_from, DATETIME_FORMAT).date(),
577
            NOT_AFTER: datetime.strptime(c.valid_to, DATETIME_FORMAT).date(),
578
            USAGE: {CertController.INVERSE_USAGE_KEY_MAP[k]: v for k, v in c.usages.items()},
579
            ISSUER: {
580
                ID: c_issuer.certificate_id,
581
                COMMON_NAME: c_issuer.common_name
582
            }
583
        }
584

    
585
    def cert_to_dict_full(self, c):
586
        """
587
        Dictionarizes a certificate directly fetched from the database, but adds subject info.
588
        Contains full information.
589
        :param c: target cert
590
        :return: certificate dict (compliant with some parts of the REST API)
591
        """
592

    
593
        Logger.info(f"Function launched.")
594

    
595
        subj = self.certificate_service.get_subject_from_certificate(c)
596
        c_issuer = self.certificate_service.get_certificate(c.parent_id)
597
        if c_issuer is None:
598
            return None
599

    
600
        return {
601
            SUBJECT: subj.to_dict(),
602
            NOT_BEFORE: datetime.strptime(c.valid_from, DATETIME_FORMAT).date(),
603
            NOT_AFTER: datetime.strptime(c.valid_to, DATETIME_FORMAT).date(),
604
            USAGE: {CertController.INVERSE_USAGE_KEY_MAP[k]: v for k, v in c.usages.items()},
605
            CA: c_issuer.certificate_id
606
        }
607

    
608
    def get_private_key_of_a_certificate(self, id):
609
        """
610
        Get a private key used to sign a certificate in PEM format specified by certificate's ID
611

    
612
        :param id: ID of a certificate whose private key is to be queried
613
        :type id: dict | bytes
614

    
615
        :rtype: PemResponse
616
        """
617

    
618
        Logger.info(f"\n\t{request.referrer}"
619
                    f"\n\t{request.method}   {request.path}   {request.scheme}"
620
                    f"\n\tCertificate ID = {id}")
621

    
622
        # try to parse the supplied ID
623
        try:
624
            v = int(id)
625
        except ValueError:
626
            Logger.error(f"Invalid request, wrong parameters 'id'[{id}].")
627
            return E_WRONG_PARAMETERS, C_BAD_REQUEST
628

    
629
        # find a certificate using the given ID
630
        cert = self.certificate_service.get_certificate(v)
631

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

    
644
    def get_public_key_of_a_certificate(self, id):
645
        """
646
        Get a public key of a certificate in PEM format specified by certificate's ID
647

    
648
        :param id: ID of a certificate whose public key is to be queried
649
        :type id: dict | bytes
650

    
651
        :rtype: PemResponse
652
        """
653

    
654
        Logger.info(f"\n\t{request.referrer}"
655
                    f"\n\t{request.method}   {request.path}   {request.scheme}"
656
                    f"\n\tCertificate ID = {id}")
657

    
658
        # try to parse the supplied ID
659
        try:
660
            v = int(id)
661
        except ValueError:
662
            Logger.error(f"Invalid request, wrong parameters 'id'[{id}].")
663
            return E_WRONG_PARAMETERS, C_BAD_REQUEST
664

    
665
        # find a certificate using the given ID
666
        cert = self.certificate_service.get_certificate(v)
667

    
668
        if cert is None:
669
            Logger.error(f"No such certificate found 'ID = {v}'.")
670
            return E_NO_CERTIFICATES_FOUND, C_NOT_FOUND
671
        else:
672
            return {"success": True, "data": self.certificate_service.get_public_key_from_certificate(cert)}, C_SUCCESS
673

    
674
    def delete_certificate(self, id):
675
        """
676
        Deletes a certificate identified by ID, including its corresponding subtree (all descendants).
677
        :param id: target certificate ID
678
        :rtype: DeleteResponse
679
        """
680

    
681
        Logger.info(f"\n\t{request.referrer}"
682
                    f"\n\t{request.method}   {request.path}   {request.scheme}"
683
                    f"\n\tCertificate ID = {id}")
684

    
685
        try:
686
            v = int(id)
687
        except ValueError:
688
            Logger.error(f"Invalid request, wrong parameters 'id'[{id}].")
689
            return E_WRONG_PARAMETERS, C_BAD_REQUEST
690

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

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

    
705
    def generate_certificate_pkcs_identity(self, id):
706
        """
707
        Generates a PKCS12 identity (including the chain of trust) of the certificate given by the specified ID.
708
        Response is of application/x-pkcs12 type.
709

    
710
        :param id: ID of a certificate whose PKCS12 identity should be generated
711
        :type id: int
712

    
713
        :rtype: Response
714
        """
715

    
716
        Logger.info(f"\n\t{request.referrer}"
717
                    f"\n\t{request.method}   {request.path}   {request.scheme}"
718
                    f"\n\tCertificate ID = {id}")
719

    
720
        # try to parse the supplied ID
721
        try:
722
            v = int(id)
723
        except ValueError:
724
            Logger.error(f"Invalid request, wrong parameters 'id'[{id}] (expected integer).")
725
            return E_WRONG_PARAMETERS, C_BAD_REQUEST
726

    
727
        # find a certificate using the given ID
728
        cert = self.certificate_service.get_certificate(v)
729

    
730
        if request.is_json:                                                         # accept JSON only
731
            body = request.get_json()
732

    
733
            # check whether the request is well formed meaning that it contains all required fields
734
            if NAME not in body.keys():
735
                return E_IDENTITY_NAME_NOT_SPECIFIED, C_BAD_REQUEST
736

    
737
            if PASSWORD not in body.keys():
738
                return E_IDENTITY_PASSWORD_NOT_SPECIFIED, C_BAD_REQUEST
739

    
740
            # parse required fields from the request
741
            identity_name = body[NAME]
742
            identity_password = body[PASSWORD]
743

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