Projekt

Obecné

Profil

Stáhnout (32 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

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

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

    
63

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

    
70

    
71
    @inject
72
    def __init__(self, certificate_service: CertificateService, key_service: KeyService):
73
        self.certificate_service = certificate_service
74
        self.key_service = key_service
75

    
76
    def create_certificate(self):
77
        """create new certificate
78

    
79
        Create a new certificate based on given information
80

    
81
        :param body: Certificate data to be created
82
        :type body: dict | bytes
83

    
84
        :rtype: CreatedResponse
85
        """
86

    
87
        Logger.info(f"\n\t{request.referrer}"
88
                    f"\n\t{request.method}   {request.path}   {request.scheme}")
89

    
90
        required_keys = {SUBJECT, USAGE, VALIDITY_DAYS}                             # required fields of the POST req
91

    
92
        if request.is_json:                                                         # accept JSON only
93
            body = request.get_json()
94

    
95
            Logger.info(f"\n\tRequest body:"
96
                        f"\n{dict_to_string(body)}")
97

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

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

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

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

    
112
            usages_dict = {}
113

    
114
            if not isinstance(body[USAGE], dict):                                   # type checking
115
                Logger.error(f"Invalid request, wrong parameter '{USAGE}'.")
116
                return E_WRONG_PARAMETERS, C_BAD_REQUEST
117

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

    
124
            key = self.key_service.create_new_key()                                 # TODO pass key
125

    
126
            extensions = ""
127
            if EXTENSIONS in body:
128
                extensions = body[EXTENSIONS]
129

    
130
            try:
131
                if CA not in body or body[CA] is None:                              # if issuer omitted (legal) or none
132
                    cert = self.certificate_service.create_root_ca(                 # create a root CA
133
                        key,
134
                        subject,
135
                        usages=usages_dict,                                         # TODO ignoring usages -> discussion
136
                        days=body[VALIDITY_DAYS],
137
                        extensions=extensions
138
                    )
139
                else:
140
                    issuer = self.certificate_service.get_certificate(body[CA])     # get base issuer info
141

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

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

    
149
                    if issuer_key is None:                                          # if it does not
150
                        Logger.error(f"Internal server error (corrupted database).")
151
                        self.key_service.delete_key(key.private_key_id)             # free
152
                        return E_CORRUPTED_DATABASE, C_INTERNAL_SERVER_ERROR        # and throw
153

    
154
                    f = self.certificate_service.create_ca if CA_ID in usages_dict and usages_dict[CA_ID] else \
155
                        self.certificate_service.create_end_cert
156

    
157
                    # noinspection PyArgumentList
158
                    cert = f(                                                       # create inter CA or end cert
159
                        key,                                                        # according to whether 'CA' is among
160
                        subject,                                                    # the usages' fields
161
                        issuer,
162
                        issuer_key,
163
                        usages=usages_dict,
164
                        days=body[VALIDITY_DAYS],
165
                        extensions=extensions
166
                    )
167
            except CryptographyException as e:
168
                return E_WRONG_PARAMETERS, C_BAD_REQUEST
169

    
170
            if cert is not None:
171
                return {"success": True,
172
                        "data": cert.certificate_id}, C_CREATED_SUCCESSFULLY
173
            else:                                                                   # if this fails, then
174
                Logger.error(f"Internal error: The certificate could not have been created.")
175
                self.key_service.delete_key(key.private_key_id)                          # free
176
                return {"success": False,                                           # and wonder what the cause is,
177
                        "data": "Internal error: The certificate could not have been created."}, C_BAD_REQUEST
178
                                                                                    # as obj/None carries only one bit
179
                                                                                    # of error information
180
        else:
181
            Logger.error(f"The request must be JSON-formatted.")
182
            return E_NOT_JSON_FORMAT, C_BAD_REQUEST                                 # throw in case of non-JSON format
183

    
184
    def get_certificate_by_id(self, id):
185
        """get certificate by ID
186

    
187
        Get certificate in PEM format by ID
188

    
189
        :param id: ID of a certificate to be queried
190
        :type id: dict | bytes
191

    
192
        :rtype: PemResponse
193
        """
194

    
195
        Logger.info(f"\n\t{request.referrer}"
196
                    f"\n\t{request.method}   {request.path}   {request.scheme}"
197
                    f"\n\tCertificate ID = {id}")
198
        try:
199
            v = int(id)
200
        except ValueError:
201
            Logger.error(f"Invalid request, wrong parameters 'id'[{id}].")
202
            return E_WRONG_PARAMETERS, C_BAD_REQUEST
203

    
204
        cert = self.certificate_service.get_certificate(v)
205

    
206
        if cert is None:
207
            Logger.error(f"No such certificate found 'ID = {v}'.")
208
            return E_NO_CERTIFICATES_FOUND, C_NO_DATA
209
        else:
210
            return {"success": True, "data": cert.pem_data}, C_SUCCESS
211

    
212
    def get_certificate_details_by_id(self, id):
213
        """get certificate's details by ID
214

    
215
        Get certificate details by ID
216

    
217
        :param id: ID of a certificate whose details are to be queried
218
        :type id: dict | bytes
219

    
220
        :rtype: CertificateResponse
221
        """
222

    
223
        Logger.info(f"\n\t{request.referrer}"
224
                    f"\n\t{request.method}   {request.path}   {request.scheme}"
225
                    f"\n\tCertificate ID = {id}")
226

    
227
        try:
228
            v = int(id)
229
        except ValueError:
230
            Logger.error(f"Invalid request, wrong parameters 'id'[{id}].")
231
            return E_WRONG_PARAMETERS, C_BAD_REQUEST
232

    
233
        cert = self.certificate_service.get_certificate(v)
234

    
235
        if cert is None:
236
            Logger.error(f"No such certificate found 'ID = {v}'.")
237
            return E_NO_CERTIFICATES_FOUND, C_NOT_FOUND
238

    
239
        data = self.cert_to_dict_full(cert)
240
        if data is None:
241
            return E_CORRUPTED_DATABASE, C_INTERNAL_SERVER_ERROR
242

    
243
        try:
244
            state = self.certificate_service.get_certificate_state(v)
245
            data["status"] = state
246
        except CertificateNotFoundException:
247
            Logger.error(f"No such certificate found 'ID = {id}'.")
248

    
249
        return {"success": True, "data": data}, C_SUCCESS
250

    
251
    def get_certificate_list(self):
252
        """get list of certificates
253

    
254
        Lists certificates based on provided filtering options
255

    
256
        :param filtering: Filter certificate type to be queried
257
        :type filtering: dict | bytes
258

    
259
        :rtype: CertificateListResponse
260
        """
261

    
262
        Logger.info(f"\n\t{request.referrer}"
263
                    f"\n\t{request.method}   {request.path}   {request.scheme}")
264

    
265

    
266
        # the filtering parameter can be read as URL argument or as a request body
267
        if request.is_json:
268
            data = request.get_json()
269
        else:
270
            data = {}
271

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

    
279
        if PAGE in request.args.keys():
280
            try:
281
                data[PAGE] = json.loads(request.args[PAGE])
282
            except JSONDecodeError:
283
                Logger.error(f"The request must be JSON-formatted.")
284
                return E_NOT_JSON_FORMAT, C_BAD_REQUEST
285

    
286
        if PER_PAGE in request.args.keys():
287
            try:
288
                data[PER_PAGE] = json.loads(request.args[PER_PAGE])
289
            except JSONDecodeError:
290
                Logger.error(f"The request must be JSON-formatted.")
291
                return E_NOT_JSON_FORMAT, C_BAD_REQUEST
292

    
293
        Logger.info(f"\n\tRequest body:"
294
                    f"\n{dict_to_string(data)}")
295

    
296
        target_types = {ROOT_CA_ID, INTERMEDIATE_CA_ID, CERTIFICATE_ID}
297
        target_usages = {v for v in CertController.INVERSE_USAGE_KEY_MAP.keys()}
298
        target_cn_substring = None
299
        issuer_id = -1
300

    
301
        unfiltered = True
302

    
303
        if PER_PAGE in data:
304
            unfiltered = False
305
            page = data.get(PAGE, 0)
306
            per_page = data[PER_PAGE]
307
        else:
308
            page = None
309
            per_page = None
310

    
311
        if FILTERING in data:                                                   # if the 'filtering' field exists
312
            unfiltered = False
313
            if isinstance(data[FILTERING], dict):                               # and it is also a 'dict'
314

    
315
                # noinspection DuplicatedCode
316
                if TYPE in data[FILTERING]:                                     # containing 'type'
317
                    if isinstance(data[FILTERING][TYPE], list):                 # which is a 'list',
318
                                                                                # map every field to id
319
                        try:
320
                            target_types = {CertController.FILTERING_TYPE_KEY_MAP[v] for v in data[FILTERING][TYPE]}
321
                        except KeyError as e:
322
                            Logger.error(f"Invalid request, wrong parameters '{FILTERING}.{TYPE}' - '{e}'.")
323
                            return E_WRONG_PARAMETERS, C_BAD_REQUEST
324
                    else:
325
                        Logger.error(f"Invalid request, wrong parameters '{FILTERING}.{TYPE}'.")
326
                        return E_WRONG_PARAMETERS, C_BAD_REQUEST
327

    
328
                # noinspection DuplicatedCode
329
                if USAGE in data[FILTERING]:                                    # containing 'usage'
330
                    if isinstance(data[FILTERING][USAGE], list):                # which is a 'list',
331
                                                                                # map every field to id
332
                        try:
333
                            target_usages = {CertController.USAGE_KEY_MAP[v] for v in data[FILTERING][USAGE]}
334
                        except KeyError as e:
335
                            Logger.error(f"Invalid request, wrong parameters '{FILTERING}.{USAGE}' - '{e}'.")
336
                            return E_WRONG_PARAMETERS, C_BAD_REQUEST
337
                    else:
338
                        Logger.error(f"Invalid request, wrong parameters '{FILTERING}.{USAGE}'.")
339
                        return E_WRONG_PARAMETERS, C_BAD_REQUEST
340

    
341
                if COMMON_NAME in data[FILTERING]:                              # containing 'CN'
342
                    if isinstance(data[FILTERING][COMMON_NAME], str):           # which is a 'str'
343
                        target_cn_substring = data[FILTERING][COMMON_NAME]
344
                    else:
345
                        Logger.error(f"Invalid request, wrong parameters '{FILTERING}.{COMMON_NAME}'.")
346
                        return E_WRONG_PARAMETERS, C_BAD_REQUEST
347

    
348
                if ISSUED_BY in data[FILTERING]:                                # containing 'issuedby'
349
                    if isinstance(data[FILTERING][ISSUED_BY], int):             # which is an 'int'
350
                        issuer_id = data[FILTERING][ISSUED_BY]                  # then get its children only
351

    
352
            else:
353
                Logger.error(f"Invalid request, wrong parameters '{FILTERING}'.")
354
                return E_WRONG_PARAMETERS, C_BAD_REQUEST
355

    
356
        if unfiltered:                                                      # if not filtering
357
            certs = self.certificate_service.get_certificates()
358
        elif issuer_id >= 0:                                                # if filtering by an issuer
359
            try:
360
                                                                            # get his children, filtered
361
                certs = self.certificate_service.get_certificates_issued_by_filter(
362
                    issuer_id=issuer_id,
363
                    target_types=target_types,
364
                    target_usages=target_usages,
365
                    target_cn_substring=target_cn_substring,
366
                    page=page,
367
                    per_page=per_page
368
                )
369
            except CertificateNotFoundException:                            # if id does not exist
370
                Logger.error(f"No such certificate found 'ID = {issuer_id}'.")
371
                return E_NO_CERTIFICATES_FOUND, C_NOT_FOUND                 # throw
372
        else:
373
            certs = self.certificate_service.get_certificates_filter(
374
                target_types=target_types,
375
                target_usages=target_usages,
376
                target_cn_substring=target_cn_substring,
377
                page=page,
378
                per_page=per_page
379
            )
380

    
381
        if certs is None:
382
            Logger.error(f"Internal server error (unknown origin).")
383
            return E_GENERAL_ERROR, C_INTERNAL_SERVER_ERROR
384
        elif len(certs) == 0:
385
            # TODO check log level
386
            Logger.warning(f"No such certificate found (empty list).")
387
            return {"success": True, "data": []}, C_SUCCESS
388
        else:
389
            ret = []
390
            for c in certs:
391
                data = self.cert_to_dict_partial(c)
392
                if data is None:
393
                    Logger.error(f"Internal server error (corrupted database).")
394
                    return E_CORRUPTED_DATABASE, C_INTERNAL_SERVER_ERROR
395
                ret.append(
396
                    data
397
                )
398
            return {"success": True, "data": ret}, C_SUCCESS
399

    
400
    def get_certificate_root_by_id(self, id):
401
        """get certificate's root of trust chain by ID
402

    
403
        Get certificate's root of trust chain in PEM format by ID
404

    
405
        :param id: ID of a child certificate whose root is to be queried
406
        :type id: dict | bytes
407

    
408
        :rtype: PemResponse
409
        """
410

    
411
        Logger.info(f"\n\t{request.referrer}"
412
                    f"\n\t{request.method}   {request.path}   {request.scheme}"
413
                    f"\n\tCertificate ID = {id}")
414

    
415
        try:
416
            v = int(id)
417
        except ValueError:
418
            Logger.error(f"Invalid request, wrong parameters 'id'[{id}].")
419
            return E_WRONG_PARAMETERS, C_BAD_REQUEST
420

    
421
        cert = self.certificate_service.get_certificate(v)
422

    
423
        if cert is None:
424
            Logger.error(f"No such certificate found 'ID = {v}'.")
425
            return E_NO_CERTIFICATES_FOUND, C_NO_DATA
426

    
427
        trust_chain = self.certificate_service.get_chain_of_trust(cert.parent_id, exclude_root=False)
428
        if trust_chain is None or len(trust_chain) == 0:
429
            Logger.error(f"No such certificate found (empty list).")
430
            return E_NO_CERTIFICATES_FOUND, C_NO_DATA
431

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

    
434
    def get_certificate_trust_chain_by_id(self, id):
435
        """get certificate's trust chain by ID (including root certificate)
436

    
437
        Get certificate trust chain in PEM format by ID
438

    
439
        :param id: ID of a child certificate whose chain is to be queried
440
        :type id: dict | bytes
441

    
442
        :rtype: PemResponse
443
        """
444

    
445
        Logger.info(f"\n\t{request.referrer}"
446
                    f"\n\t{request.method}   {request.path}   {request.scheme}"
447
                    f"\n\tCertificate ID = {id}")
448

    
449
        try:
450
            v = int(id)
451
        except ValueError:
452
            Logger.error(f"Invalid request, wrong parameters 'id'[{id}].")
453
            return E_WRONG_PARAMETERS, C_BAD_REQUEST
454

    
455
        cert = self.certificate_service.get_certificate(v)
456

    
457
        if cert is None:
458
            Logger.error(f"No such certificate found 'ID = {v}'.")
459
            return E_NO_CERTIFICATES_FOUND, C_NO_DATA
460

    
461
        if cert.parent_id is None:
462
            Logger.error(f"Parent ID is empty in certificate 'ID = {v}'.")
463
            return E_NO_CERTIFICATES_FOUND, C_NO_DATA
464

    
465
        trust_chain = self.certificate_service.get_chain_of_trust(cert.parent_id, exclude_root=False)
466

    
467
        ret = []
468
        for intermediate in trust_chain:
469
            ret.append(intermediate.pem_data)
470

    
471
        return {"success": True, "data": "".join(ret)}, C_SUCCESS
472

    
473
    def set_certificate_status(self, id):
474
        """
475
        Revoke a certificate given by ID
476
            - revocation request may contain revocation reason
477
            - revocation reason is verified based on the possible predefined values
478
            - if revocation reason is not specified 'undefined' value is used
479
        :param id: Identifier of the certificate to be revoked
480
        :type id: int
481

    
482
        :rtype: SuccessResponse | ErrorResponse (see OpenAPI definition)
483
        """
484

    
485
        Logger.info(f"\n\t{request.referrer}"
486
                    f"\n\t{request.method}   {request.path}   {request.scheme}"
487
                    f"\n\tCertificate ID = {id}")
488

    
489
        required_keys = {STATUS}  # required keys
490

    
491
        # check if the request contains a JSON body
492
        if request.is_json:
493
            request_body = request.get_json()
494

    
495
            Logger.info(f"\n\tRequest body:"
496
                        f"\n{dict_to_string(request_body)}")
497

    
498
            # try to parse certificate identifier -> if it is not int return error 400
499
            try:
500
                identifier = int(id)
501
            except ValueError:
502
                Logger.error(f"Invalid request, wrong parameters 'id'[{id}].")
503
                return E_WRONG_PARAMETERS, C_BAD_REQUEST
504

    
505
            # verify that all required keys are present
506
            if not all(k in request_body for k in required_keys):
507
                Logger.error(f"Invalid request, missing parameters.")
508
                return E_MISSING_PARAMETERS, C_BAD_REQUEST
509

    
510
            # get status and reason from the request
511
            status = request_body[STATUS]
512
            reason = request_body.get(REASON, REASON_UNDEFINED)
513
            try:
514
                # set certificate status using certificate_service
515
                self.certificate_service.set_certificate_revocation_status(identifier, status, reason)
516
            except (RevocationReasonInvalidException, CertificateStatusInvalidException):
517
                # these exceptions are thrown in case invalid status or revocation reason is passed to the controller
518
                Logger.error(f"Invalid request, wrong parameters.")
519
                return E_WRONG_PARAMETERS, C_BAD_REQUEST
520
            except CertificateAlreadyRevokedException:
521
                Logger.error(f"Certificate is already revoked 'ID = {identifier}'.")
522
                return E_NO_CERTIFICATE_ALREADY_REVOKED, C_BAD_REQUEST
523
            except CertificateNotFoundException:
524
                Logger.error(f"No such certificate found 'ID = {identifier}'.")
525
                return E_NO_CERTIFICATES_FOUND, C_NOT_FOUND
526
            except CertificateCannotBeSetToValid as e:
527
                return {"success": False, "data": str(e)}, C_BAD_REQUEST
528
            return {"success": True,
529
                    "data": "Certificate status updated successfully."}, C_SUCCESS
530
        # throw an error in case the request does not contain a json body
531
        else:
532
            Logger.error(f"The request must be JSON-formatted.")
533
            return E_NOT_JSON_FORMAT, C_BAD_REQUEST
534

    
535
    def cert_to_dict_partial(self, c):
536
        """
537
        Dictionarizes a certificate directly fetched from the database. Contains partial information.
538
        :param c: target cert
539
        :return: certificate dict (compliant with some parts of the REST API)
540
        """
541

    
542
        # TODO check log
543
        Logger.debug(f"Function launched.")
544

    
545
        c_issuer = self.certificate_service.get_certificate(c.parent_id)
546
        if c_issuer is None:
547
            return None
548

    
549
        return {
550
            ID: c.certificate_id,
551
            COMMON_NAME: c.common_name,
552
            NOT_BEFORE: datetime.strptime(c.valid_from, DATETIME_FORMAT).date(),
553
            NOT_AFTER: datetime.strptime(c.valid_to, DATETIME_FORMAT).date(),
554
            USAGE: {CertController.INVERSE_USAGE_KEY_MAP[k]: v for k, v in c.usages.items()},
555
            ISSUER: {
556
                ID: c_issuer.certificate_id,
557
                COMMON_NAME: c_issuer.common_name
558
            }
559
        }
560

    
561
    def cert_to_dict_full(self, c):
562
        """
563
        Dictionarizes a certificate directly fetched from the database, but adds subject info.
564
        Contains full information.
565
        :param c: target cert
566
        :return: certificate dict (compliant with some parts of the REST API)
567
        """
568

    
569
        Logger.info(f"Function launched.")
570

    
571
        subj = self.certificate_service.get_subject_from_certificate(c)
572
        c_issuer = self.certificate_service.get_certificate(c.parent_id)
573
        if c_issuer is None:
574
            return None
575

    
576
        return {
577
            SUBJECT: subj.to_dict(),
578
            NOT_BEFORE: datetime.strptime(c.valid_from, DATETIME_FORMAT).date(),
579
            NOT_AFTER: datetime.strptime(c.valid_to, DATETIME_FORMAT).date(),
580
            USAGE: {CertController.INVERSE_USAGE_KEY_MAP[k]: v for k, v in c.usages.items()},
581
            CA: c_issuer.certificate_id
582
        }
583

    
584
    def get_private_key_of_a_certificate(self, id):
585
        """
586
        Get a private key used to sign a certificate in PEM format specified by certificate's ID
587

    
588
        :param id: ID of a certificate whose private key is to be queried
589
        :type id: dict | bytes
590

    
591
        :rtype: PemResponse
592
        """
593

    
594
        Logger.info(f"\n\t{request.referrer}"
595
                    f"\n\t{request.method}   {request.path}   {request.scheme}"
596
                    f"\n\tCertificate ID = {id}")
597

    
598
        # try to parse the supplied ID
599
        try:
600
            v = int(id)
601
        except ValueError:
602
            Logger.error(f"Invalid request, wrong parameters 'id'[{id}].")
603
            return E_WRONG_PARAMETERS, C_BAD_REQUEST
604

    
605
        # find a certificate using the given ID
606
        cert = self.certificate_service.get_certificate(v)
607

    
608
        if cert is None:
609
            Logger.error(f"No such certificate found 'ID = {v}'.")
610
            return E_NO_CERTIFICATES_FOUND, C_NOT_FOUND
611
        else:
612
            # certificate exists, fetch it's private key
613
            private_key = self.key_service.get_key(cert.private_key_id)
614
            if cert is None:
615
                Logger.error(f"Internal server error (certificate's private key cannot be found).")
616
                return E_NO_CERT_PRIVATE_KEY_FOUND, C_INTERNAL_SERVER_ERROR
617
            else:
618
                return {"success": True, "data": private_key.private_key}, C_SUCCESS
619

    
620
    def get_public_key_of_a_certificate(self, id):
621
        """
622
        Get a public key of a certificate in PEM format specified by certificate's ID
623

    
624
        :param id: ID of a certificate whose public key is to be queried
625
        :type id: dict | bytes
626

    
627
        :rtype: PemResponse
628
        """
629

    
630
        Logger.info(f"\n\t{request.referrer}"
631
                    f"\n\t{request.method}   {request.path}   {request.scheme}"
632
                    f"\n\tCertificate ID = {id}")
633

    
634
        # try to parse the supplied ID
635
        try:
636
            v = int(id)
637
        except ValueError:
638
            Logger.error(f"Invalid request, wrong parameters 'id'[{id}].")
639
            return E_WRONG_PARAMETERS, C_BAD_REQUEST
640

    
641
        # find a certificate using the given ID
642
        cert = self.certificate_service.get_certificate(v)
643

    
644
        if cert is None:
645
            Logger.error(f"No such certificate found 'ID = {v}'.")
646
            return E_NO_CERTIFICATES_FOUND, C_NOT_FOUND
647
        else:
648
            return {"success": True, "data": self.certificate_service.get_public_key_from_certificate(cert)}, C_SUCCESS
649

    
650
    def delete_certificate(self, id):
651
        """
652
        Deletes a certificate identified by ID, including its corresponding subtree (all descendants).
653
        :param id: target certificate ID
654
        :rtype: DeleteResponse
655
        """
656

    
657
        Logger.info(f"\n\t{request.referrer}"
658
                    f"\n\t{request.method}   {request.path}   {request.scheme}"
659
                    f"\n\tCertificate ID = {id}")
660

    
661
        try:
662
            v = int(id)
663
        except ValueError:
664
            Logger.error(f"Invalid request, wrong parameters 'id'[{id}].")
665
            return E_WRONG_PARAMETERS, C_BAD_REQUEST
666

    
667
        try:
668
            self.certificate_service.delete_certificate(v)
669
        except CertificateNotFoundException:
670
            Logger.error(f"No such certificate found 'ID = {v}'.")
671
            return E_NO_CERTIFICATES_FOUND, C_NOT_FOUND
672
        except DatabaseException:
673
            Logger.error(f"Internal server error (corrupted database).")
674
            return E_CORRUPTED_DATABASE, C_INTERNAL_SERVER_ERROR
675
        except CertificateStatusInvalidException or RevocationReasonInvalidException or UnknownException:
676
            Logger.error(f"Internal server error (unknown origin).")
677
            return E_GENERAL_ERROR, C_INTERNAL_SERVER_ERROR
678

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

    
681
    def generate_certificate_pkcs_identity(self, id):
682
        """
683
        Generates a PKCS12 identity (including the chain of trust) of the certificate given by the specified ID.
684
        Response is of application/x-pkcs12 type.
685

    
686
        :param id: ID of a certificate whose PKCS12 identity should be generated
687
        :type id: int
688

    
689
        :rtype: Response
690
        """
691

    
692
        Logger.info(f"\n\t{request.referrer}"
693
                    f"\n\t{request.method}   {request.path}   {request.scheme}"
694
                    f"\n\tCertificate ID = {id}")
695

    
696
        # try to parse the supplied ID
697
        try:
698
            v = int(id)
699
        except ValueError:
700
            Logger.error(f"Invalid request, wrong parameters 'id'[{id}] (expected integer).")
701
            return E_WRONG_PARAMETERS, C_BAD_REQUEST
702

    
703
        # find a certificate using the given ID
704
        cert = self.certificate_service.get_certificate(v)
705

    
706
        if request.is_json:                                                         # accept JSON only
707
            body = request.get_json()
708

    
709
            # check whether the request is well formed meaning that it contains all required fields
710
            if NAME not in body.keys():
711
                return E_IDENTITY_NAME_NOT_SPECIFIED, C_BAD_REQUEST
712

    
713
            if PASSWORD not in body.keys():
714
                return E_IDENTITY_PASSWORD_NOT_SPECIFIED, C_BAD_REQUEST
715

    
716
            # parse required fields from the request
717
            identity_name = body[NAME]
718
            identity_password = body[PASSWORD]
719

    
720
            # check whether a certificated specified by the given ID exists
721
            if cert is None:
722
                Logger.error(f"No such certificate found 'ID = {v}'.")
723
                return E_NO_CERTIFICATES_FOUND, C_NOT_FOUND
724
            else:
725
                # try to load it's private key
726
                key = self.key_service.get_key(cert.private_key_id)
727
                if key is None:
728
                    Logger.error(
729
                        f"The private key 'ID = {cert.private_key_id}'of the certificate 'ID = {cert.certificate_id}' does not exist.")
730
                    return E_NO_CERTIFICATES_FOUND, C_INTERNAL_SERVER_ERROR
731
                else:
732
                    # generate PKCS12 identity
733
                    identity_byte_array = self.certificate_service.generate_pkcs_identity(cert, key,
734
                                                                                          identity_name,
735
                                                                                          identity_password)
736
                    return Response(identity_byte_array, mimetype='application/x-pkcs12')
(2-2/4)