Projekt

Obecné

Profil

Stáhnout (34.9 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
E_INVALID_EXTENSIONS = {"success": False, "data": "Error occurred while creating a certificate. "
66
                                                  "It may be caused by wrong format of extensions."}
67

    
68
E_UNHANDLED_CRYPTOGRAPHY_ERROR = {"success": False, "data": "An unknown error has happened in the cryptography library."}
69
E_UNHANDLED_DATABASE_ERROR = {"success": False, "data": "An unknown database error has happened."}
70
E_UNHANDLED_ERROR = {"success": False, "data": "An unknown error has happened."}
71

    
72

    
73
class CertController:
74
    USAGE_KEY_MAP = {'CA': CA_ID, 'SSL': SSL_ID, 'digitalSignature': SIGNATURE_ID, 'authentication': AUTHENTICATION_ID}
75
    INVERSE_USAGE_KEY_MAP = {k: v for v, k in USAGE_KEY_MAP.items()}
76
    FILTERING_TYPE_KEY_MAP = {'root': ROOT_CA_ID, 'inter': INTERMEDIATE_CA_ID, 'end': CERTIFICATE_ID}
77
    # INVERSE_FILTERING_TYPE_KEY_MAP = {k: v for v, k in FILTERING_TYPE_KEY_MAP.items()}
78

    
79

    
80
    @inject
81
    def __init__(self, certificate_service: CertificateService, key_service: KeyService):
82
        self.certificate_service = certificate_service
83
        self.key_service = key_service
84

    
85
    def create_certificate(self):
86
        """create new certificate
87

    
88
        Create a new certificate based on given information
89

    
90
        :param body: Certificate data to be created
91
        :type body: dict | bytes
92

    
93
        :rtype: CreatedResponse
94
        """
95

    
96
        Logger.info(f"\n\t{request.referrer}"
97
                    f"\n\t{request.method}   {request.path}   {request.scheme}")
98

    
99
        required_keys = {SUBJECT, USAGE, VALIDITY_DAYS}                             # required fields of the POST req
100

    
101
        if request.is_json:                                                         # accept JSON only
102
            body = request.get_json()
103

    
104
            Logger.info(f"\n\tRequest body:"
105
                        f"\n{dict_to_string(body)}")
106

    
107
            if not all(k in body for k in required_keys):                           # verify that all keys are present
108
                Logger.error(f"Invalid request, missing parameters")
109
                return E_MISSING_PARAMETERS, C_BAD_REQUEST
110

    
111
            if not isinstance(body[VALIDITY_DAYS], int):                            # type checking
112
                Logger.error(f"Invalid request, wrong parameter '{VALIDITY_DAYS}'.")
113
                return E_WRONG_PARAMETERS, C_BAD_REQUEST
114

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

    
117
            if subject is None:                                                     # if the format is incorrect
118
                Logger.error(f"Invalid request, wrong parameter '{SUBJECT}'.")
119
                return E_WRONG_PARAMETERS, C_BAD_REQUEST
120

    
121
            usages_dict = {}
122

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

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

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

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

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

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

    
181
                    if issuer_key is None:                                          # if it does not
182
                        Logger.error(f"Internal server error (corrupted database).")
183
                        self.key_service.delete_key(key.private_key_id)             # free
184
                        return E_CORRUPTED_DATABASE, C_INTERNAL_SERVER_ERROR        # and throw
185

    
186
                    f = self.certificate_service.create_ca if CA_ID in usages_dict and usages_dict[CA_ID] else \
187
                        self.certificate_service.create_end_cert
188

    
189
                    # noinspection PyArgumentList
190
                    cert = f(                                                       # create inter CA or end cert
191
                        key,                                                        # according to whether 'CA' is among
192
                        subject,                                                    # the usages' fields
193
                        issuer,
194
                        issuer_key,
195
                        usages=usages_dict,
196
                        days=body[VALIDITY_DAYS],
197
                        extensions=extensions
198
                    )
199

    
200
            # if extensions are specified and CryptoException occurs, the problem is probably in the
201
            # extensions format - otherwise error 500 is expected
202
            except CryptographyException as e:
203
                if len(extensions) > 0:
204
                    return E_INVALID_EXTENSIONS, C_BAD_REQUEST
205
                else:
206
                    raise CryptographyException(e.executable, e.args, e.message)
207

    
208
            if cert is not None:
209
                return {"success": True,
210
                        "data": cert.certificate_id}, C_CREATED_SUCCESSFULLY
211
            else:                                                                   # if this fails, then
212
                Logger.error(f"Internal error: The certificate could not have been created.")
213
                self.key_service.delete_key(key.private_key_id)                     # free
214
                return {"success": False,                                           # and wonder what the cause is,
215
                        "data": "Internal error: The certificate could not have been created."}, C_BAD_REQUEST
216
                                                                                    # as obj/None carries only one bit
217
                                                                                    # of error information
218
        else:
219
            Logger.error(f"The request must be JSON-formatted.")
220
            return E_NOT_JSON_FORMAT, C_BAD_REQUEST                                 # throw in case of non-JSON format
221

    
222
    def get_certificate_by_id(self, id):
223
        """get certificate by ID
224

    
225
        Get certificate in PEM format by ID
226

    
227
        :param id: ID of a certificate to be queried
228
        :type id: dict | bytes
229

    
230
        :rtype: PemResponse
231
        """
232

    
233
        Logger.info(f"\n\t{request.referrer}"
234
                    f"\n\t{request.method}   {request.path}   {request.scheme}"
235
                    f"\n\tCertificate ID = {id}")
236
        try:
237
            v = int(id)
238
        except ValueError:
239
            Logger.error(f"Invalid request, wrong parameters 'id'[{id}].")
240
            return E_WRONG_PARAMETERS, C_BAD_REQUEST
241

    
242
        cert = self.certificate_service.get_certificate(v)
243

    
244
        if cert is None:
245
            Logger.error(f"No such certificate found 'ID = {v}'.")
246
            return E_NO_CERTIFICATES_FOUND, C_NO_DATA
247
        else:
248
            return {"success": True, "data": cert.pem_data}, C_SUCCESS
249

    
250
    def get_certificate_details_by_id(self, id):
251
        """get certificate's details by ID
252

    
253
        Get certificate details by ID
254

    
255
        :param id: ID of a certificate whose details are to be queried
256
        :type id: dict | bytes
257

    
258
        :rtype: CertificateResponse
259
        """
260

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

    
265
        try:
266
            v = int(id)
267
        except ValueError:
268
            Logger.error(f"Invalid request, wrong parameters 'id'[{id}].")
269
            return E_WRONG_PARAMETERS, C_BAD_REQUEST
270

    
271
        cert = self.certificate_service.get_certificate(v)
272

    
273
        if cert is None:
274
            Logger.error(f"No such certificate found 'ID = {v}'.")
275
            return E_NO_CERTIFICATES_FOUND, C_NOT_FOUND
276

    
277
        data = self.cert_to_dict_full(cert)
278
        if data is None:
279
            return E_CORRUPTED_DATABASE, C_INTERNAL_SERVER_ERROR
280

    
281
        try:
282
            state = self.certificate_service.get_certificate_state(v)
283
            data["status"] = state
284
        except CertificateNotFoundException:
285
            Logger.error(f"No such certificate found 'ID = {id}'.")
286

    
287
        return {"success": True, "data": data}, C_SUCCESS
288

    
289
    def get_certificate_list(self):
290
        """get list of certificates
291

    
292
        Lists certificates based on provided filtering options
293

    
294
        :param filtering: Filter certificate type to be queried
295
        :type filtering: dict | bytes
296

    
297
        :rtype: CertificateListResponse
298
        """
299

    
300
        Logger.info(f"\n\t{request.referrer}"
301
                    f"\n\t{request.method}   {request.path}   {request.scheme}")
302

    
303

    
304
        # the filtering parameter can be read as URL argument or as a request body
305
        if request.is_json:
306
            data = request.get_json()
307
        else:
308
            data = {}
309

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

    
317
        if PAGE in request.args.keys():
318
            try:
319
                data[PAGE] = json.loads(request.args[PAGE])
320
            except JSONDecodeError:
321
                Logger.error(f"The request must be JSON-formatted.")
322
                return E_NOT_JSON_FORMAT, C_BAD_REQUEST
323

    
324
        if PER_PAGE in request.args.keys():
325
            try:
326
                data[PER_PAGE] = json.loads(request.args[PER_PAGE])
327
            except JSONDecodeError:
328
                Logger.error(f"The request must be JSON-formatted.")
329
                return E_NOT_JSON_FORMAT, C_BAD_REQUEST
330

    
331
        Logger.info(f"\n\tRequest body:"
332
                    f"\n{dict_to_string(data)}")
333

    
334
        target_types = {ROOT_CA_ID, INTERMEDIATE_CA_ID, CERTIFICATE_ID}
335
        target_usages = None
336
        target_cn_substring = None
337
        issuer_id = -1
338

    
339
        unfiltered = True
340

    
341
        if PER_PAGE in data:
342
            unfiltered = False
343
            page = data.get(PAGE, 0)
344
            per_page = data[PER_PAGE]
345
        else:
346
            page = None
347
            per_page = None
348

    
349
        if FILTERING in data:                                                   # if the 'filtering' field exists
350
            unfiltered = False
351
            if isinstance(data[FILTERING], dict):                               # and it is also a 'dict'
352

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

    
366
                # noinspection DuplicatedCode
367
                if USAGE in data[FILTERING]:                                    # containing 'usage'
368
                    if isinstance(data[FILTERING][USAGE], list):                # which is a 'list',
369
                                                                                # map every field to id
370
                        try:
371
                            target_usages = {CertController.USAGE_KEY_MAP[v] for v in data[FILTERING][USAGE]}
372
                        except KeyError as e:
373
                            Logger.error(f"Invalid request, wrong parameters '{FILTERING}.{USAGE}' - '{e}'.")
374
                            return E_WRONG_PARAMETERS, C_BAD_REQUEST
375
                    else:
376
                        Logger.error(f"Invalid request, wrong parameters '{FILTERING}.{USAGE}'.")
377
                        return E_WRONG_PARAMETERS, C_BAD_REQUEST
378

    
379
                if COMMON_NAME in data[FILTERING]:                              # containing 'CN'
380
                    if isinstance(data[FILTERING][COMMON_NAME], str):           # which is a 'str'
381
                        target_cn_substring = data[FILTERING][COMMON_NAME]
382
                    else:
383
                        Logger.error(f"Invalid request, wrong parameters '{FILTERING}.{COMMON_NAME}'.")
384
                        return E_WRONG_PARAMETERS, C_BAD_REQUEST
385

    
386
                if ISSUED_BY in data[FILTERING]:                                # containing 'issuedby'
387
                    if isinstance(data[FILTERING][ISSUED_BY], int):             # which is an 'int'
388
                        issuer_id = data[FILTERING][ISSUED_BY]                  # then get its children only
389

    
390
            else:
391
                Logger.error(f"Invalid request, wrong parameters '{FILTERING}'.")
392
                return E_WRONG_PARAMETERS, C_BAD_REQUEST
393

    
394
        if unfiltered:                                                      # if not filtering
395
            certs = self.certificate_service.get_certificates()
396
        elif issuer_id >= 0:                                                # if filtering by an issuer
397
            try:
398
                                                                            # get his children, filtered
399
                certs = self.certificate_service.get_certificates_issued_by_filter(
400
                    issuer_id=issuer_id,
401
                    target_types=target_types,
402
                    target_usages=target_usages,
403
                    target_cn_substring=target_cn_substring,
404
                    page=page,
405
                    per_page=per_page
406
                )
407
            except CertificateNotFoundException:                            # if id does not exist
408
                Logger.error(f"No such certificate found 'ID = {issuer_id}'.")
409
                return E_NO_CERTIFICATES_FOUND, C_NOT_FOUND                 # throw
410
        else:
411
            certs = self.certificate_service.get_certificates_filter(
412
                target_types=target_types,
413
                target_usages=target_usages,
414
                target_cn_substring=target_cn_substring,
415
                page=page,
416
                per_page=per_page
417
            )
418

    
419
        if certs is None:
420
            Logger.error(f"Internal server error (unknown origin).")
421
            return E_GENERAL_ERROR, C_INTERNAL_SERVER_ERROR
422
        elif len(certs) == 0:
423
            # TODO check log level
424
            Logger.warning(f"No such certificate found (empty list).")
425
            return {"success": True, "data": []}, C_SUCCESS
426
        else:
427
            ret = []
428
            for c in certs:
429
                data = self.cert_to_dict_partial(c)
430
                if data is None:
431
                    Logger.error(f"Internal server error (corrupted database).")
432
                    return E_CORRUPTED_DATABASE, C_INTERNAL_SERVER_ERROR
433
                ret.append(
434
                    data
435
                )
436
            return {"success": True, "data": ret}, C_SUCCESS
437

    
438
    def get_certificate_root_by_id(self, id):
439
        """get certificate's root of trust chain by ID
440

    
441
        Get certificate's root of trust chain in PEM format by ID
442

    
443
        :param id: ID of a child certificate whose root is to be queried
444
        :type id: dict | bytes
445

    
446
        :rtype: PemResponse
447
        """
448

    
449
        Logger.info(f"\n\t{request.referrer}"
450
                    f"\n\t{request.method}   {request.path}   {request.scheme}"
451
                    f"\n\tCertificate ID = {id}")
452

    
453
        try:
454
            v = int(id)
455
        except ValueError:
456
            Logger.error(f"Invalid request, wrong parameters 'id'[{id}].")
457
            return E_WRONG_PARAMETERS, C_BAD_REQUEST
458

    
459
        cert = self.certificate_service.get_certificate(v)
460

    
461
        if cert is None:
462
            Logger.error(f"No such certificate found '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
        if trust_chain is None or len(trust_chain) == 0:
467
            Logger.error(f"No such certificate found (empty list).")
468
            return E_NO_CERTIFICATES_FOUND, C_NO_DATA
469

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

    
472
    def get_certificate_trust_chain_by_id(self, id):
473
        """get certificate's trust chain by ID (including root certificate)
474

    
475
        Get certificate trust chain in PEM format by ID
476

    
477
        :param id: ID of a child certificate whose chain is to be queried
478
        :type id: dict | bytes
479

    
480
        :rtype: PemResponse
481
        """
482

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

    
487
        try:
488
            v = int(id)
489
        except ValueError:
490
            Logger.error(f"Invalid request, wrong parameters 'id'[{id}].")
491
            return E_WRONG_PARAMETERS, C_BAD_REQUEST
492

    
493
        cert = self.certificate_service.get_certificate(v)
494

    
495
        if cert is None:
496
            Logger.error(f"No such certificate found 'ID = {v}'.")
497
            return E_NO_CERTIFICATES_FOUND, C_NO_DATA
498

    
499
        if cert.parent_id is None:
500
            Logger.error(f"Parent ID is empty in certificate 'ID = {v}'.")
501
            return E_NO_CERTIFICATES_FOUND, C_NO_DATA
502

    
503
        trust_chain = self.certificate_service.get_chain_of_trust(cert.parent_id, exclude_root=False)
504

    
505
        ret = []
506
        for intermediate in trust_chain:
507
            ret.append(intermediate.pem_data)
508

    
509
        return {"success": True, "data": "".join(ret)}, C_SUCCESS
510

    
511
    def set_certificate_status(self, id):
512
        """
513
        Revoke a certificate given by ID
514
            - revocation request may contain revocation reason
515
            - revocation reason is verified based on the possible predefined values
516
            - if revocation reason is not specified 'undefined' value is used
517
        :param id: Identifier of the certificate to be revoked
518
        :type id: int
519

    
520
        :rtype: SuccessResponse | ErrorResponse (see OpenAPI definition)
521
        """
522

    
523
        Logger.info(f"\n\t{request.referrer}"
524
                    f"\n\t{request.method}   {request.path}   {request.scheme}"
525
                    f"\n\tCertificate ID = {id}")
526

    
527
        required_keys = {STATUS}  # required keys
528

    
529
        # check if the request contains a JSON body
530
        if request.is_json:
531
            request_body = request.get_json()
532

    
533
            Logger.info(f"\n\tRequest body:"
534
                        f"\n{dict_to_string(request_body)}")
535

    
536
            # try to parse certificate identifier -> if it is not int return error 400
537
            try:
538
                identifier = int(id)
539
            except ValueError:
540
                Logger.error(f"Invalid request, wrong parameters 'id'[{id}].")
541
                return E_WRONG_PARAMETERS, C_BAD_REQUEST
542

    
543
            # verify that all required keys are present
544
            if not all(k in request_body for k in required_keys):
545
                Logger.error(f"Invalid request, missing parameters.")
546
                return E_MISSING_PARAMETERS, C_BAD_REQUEST
547

    
548
            # get status and reason from the request
549
            status = request_body[STATUS]
550
            reason = request_body.get(REASON, REASON_UNDEFINED)
551
            try:
552
                # set certificate status using certificate_service
553
                self.certificate_service.set_certificate_revocation_status(identifier, status, reason)
554
            except (RevocationReasonInvalidException, CertificateStatusInvalidException):
555
                # these exceptions are thrown in case invalid status or revocation reason is passed to the controller
556
                Logger.error(f"Invalid request, wrong parameters.")
557
                return E_WRONG_PARAMETERS, C_BAD_REQUEST
558
            except CertificateAlreadyRevokedException:
559
                Logger.error(f"Certificate is already revoked 'ID = {identifier}'.")
560
                return E_NO_CERTIFICATE_ALREADY_REVOKED, C_BAD_REQUEST
561
            except CertificateNotFoundException:
562
                Logger.error(f"No such certificate found 'ID = {identifier}'.")
563
                return E_NO_CERTIFICATES_FOUND, C_NOT_FOUND
564
            except CertificateCannotBeSetToValid as e:
565
                return {"success": False, "data": str(e)}, C_BAD_REQUEST
566
            return {"success": True,
567
                    "data": "Certificate status updated successfully."}, C_SUCCESS
568
        # throw an error in case the request does not contain a json body
569
        else:
570
            Logger.error(f"The request must be JSON-formatted.")
571
            return E_NOT_JSON_FORMAT, C_BAD_REQUEST
572

    
573
    def cert_to_dict_partial(self, c):
574
        """
575
        Dictionarizes a certificate directly fetched from the database. Contains partial information.
576
        :param c: target cert
577
        :return: certificate dict (compliant with some parts of the REST API)
578
        """
579

    
580
        # TODO check log
581
        Logger.debug(f"Function launched.")
582

    
583
        c_issuer = self.certificate_service.get_certificate(c.parent_id)
584
        if c_issuer is None:
585
            return None
586

    
587
        return {
588
            ID: c.certificate_id,
589
            COMMON_NAME: c.common_name,
590
            NOT_BEFORE: datetime.strptime(c.valid_from, DATETIME_FORMAT).date(),
591
            NOT_AFTER: datetime.strptime(c.valid_to, DATETIME_FORMAT).date(),
592
            USAGE: {CertController.INVERSE_USAGE_KEY_MAP[k]: v for k, v in c.usages.items()},
593
            ISSUER: {
594
                ID: c_issuer.certificate_id,
595
                COMMON_NAME: c_issuer.common_name
596
            }
597
        }
598

    
599
    def cert_to_dict_full(self, c):
600
        """
601
        Dictionarizes a certificate directly fetched from the database, but adds subject info.
602
        Contains full information.
603
        :param c: target cert
604
        :return: certificate dict (compliant with some parts of the REST API)
605
        """
606

    
607
        Logger.info(f"Function launched.")
608

    
609
        subj = self.certificate_service.get_subject_from_certificate(c)
610
        c_issuer = self.certificate_service.get_certificate(c.parent_id)
611
        if c_issuer is None:
612
            return None
613

    
614
        return {
615
            SUBJECT: subj.to_dict(),
616
            NOT_BEFORE: datetime.strptime(c.valid_from, DATETIME_FORMAT).date(),
617
            NOT_AFTER: datetime.strptime(c.valid_to, DATETIME_FORMAT).date(),
618
            USAGE: {CertController.INVERSE_USAGE_KEY_MAP[k]: v for k, v in c.usages.items()},
619
            CA: c_issuer.certificate_id
620
        }
621

    
622
    def get_private_key_of_a_certificate(self, id):
623
        """
624
        Get a private key used to sign a certificate in PEM format specified by certificate's ID
625

    
626
        :param id: ID of a certificate whose private key is to be queried
627
        :type id: dict | bytes
628

    
629
        :rtype: PemResponse
630
        """
631

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

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

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

    
646
        if cert is None:
647
            Logger.error(f"No such certificate found 'ID = {v}'.")
648
            return E_NO_CERTIFICATES_FOUND, C_NOT_FOUND
649
        else:
650
            # certificate exists, fetch it's private key
651
            private_key = self.key_service.get_key(cert.private_key_id)
652
            if cert is None:
653
                Logger.error(f"Internal server error (certificate's private key cannot be found).")
654
                return E_NO_CERT_PRIVATE_KEY_FOUND, C_INTERNAL_SERVER_ERROR
655
            else:
656
                return {"success": True, "data": private_key.private_key}, C_SUCCESS
657

    
658
    def get_public_key_of_a_certificate(self, id):
659
        """
660
        Get a public key of a certificate in PEM format specified by certificate's ID
661

    
662
        :param id: ID of a certificate whose public key is to be queried
663
        :type id: dict | bytes
664

    
665
        :rtype: PemResponse
666
        """
667

    
668
        Logger.info(f"\n\t{request.referrer}"
669
                    f"\n\t{request.method}   {request.path}   {request.scheme}"
670
                    f"\n\tCertificate ID = {id}")
671

    
672
        # try to parse the supplied ID
673
        try:
674
            v = int(id)
675
        except ValueError:
676
            Logger.error(f"Invalid request, wrong parameters 'id'[{id}].")
677
            return E_WRONG_PARAMETERS, C_BAD_REQUEST
678

    
679
        # find a certificate using the given ID
680
        cert = self.certificate_service.get_certificate(v)
681

    
682
        if cert is None:
683
            Logger.error(f"No such certificate found 'ID = {v}'.")
684
            return E_NO_CERTIFICATES_FOUND, C_NOT_FOUND
685
        else:
686
            return {"success": True, "data": self.certificate_service.get_public_key_from_certificate(cert)}, C_SUCCESS
687

    
688
    def delete_certificate(self, id):
689
        """
690
        Deletes a certificate identified by ID, including its corresponding subtree (all descendants).
691
        :param id: target certificate ID
692
        :rtype: DeleteResponse
693
        """
694

    
695
        Logger.info(f"\n\t{request.referrer}"
696
                    f"\n\t{request.method}   {request.path}   {request.scheme}"
697
                    f"\n\tCertificate ID = {id}")
698

    
699
        try:
700
            v = int(id)
701
        except ValueError:
702
            Logger.error(f"Invalid request, wrong parameters 'id'[{id}].")
703
            return E_WRONG_PARAMETERS, C_BAD_REQUEST
704

    
705
        try:
706
            self.certificate_service.delete_certificate(v)
707
        except CertificateNotFoundException:
708
            Logger.error(f"No such certificate found 'ID = {v}'.")
709
            return E_NO_CERTIFICATES_FOUND, C_NOT_FOUND
710
        except DatabaseException:
711
            Logger.error(f"Internal server error (corrupted database).")
712
            return E_CORRUPTED_DATABASE, C_INTERNAL_SERVER_ERROR
713
        except CertificateStatusInvalidException or RevocationReasonInvalidException or UnknownException:
714
            Logger.error(f"Internal server error (unknown origin).")
715
            return E_GENERAL_ERROR, C_INTERNAL_SERVER_ERROR
716

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

    
719
    def generate_certificate_pkcs_identity(self, id):
720
        """
721
        Generates a PKCS12 identity (including the chain of trust) of the certificate given by the specified ID.
722
        Response is of application/x-pkcs12 type.
723

    
724
        :param id: ID of a certificate whose PKCS12 identity should be generated
725
        :type id: int
726

    
727
        :rtype: Response
728
        """
729

    
730
        Logger.info(f"\n\t{request.referrer}"
731
                    f"\n\t{request.method}   {request.path}   {request.scheme}"
732
                    f"\n\tCertificate ID = {id}")
733

    
734
        # try to parse the supplied ID
735
        try:
736
            v = int(id)
737
        except ValueError:
738
            Logger.error(f"Invalid request, wrong parameters 'id'[{id}] (expected integer).")
739
            return E_WRONG_PARAMETERS, C_BAD_REQUEST
740

    
741
        # find a certificate using the given ID
742
        cert = self.certificate_service.get_certificate(v)
743

    
744
        if request.is_json:                                                         # accept JSON only
745
            body = request.get_json()
746

    
747
            # check whether the request is well formed meaning that it contains all required fields
748
            if NAME not in body.keys():
749
                return E_IDENTITY_NAME_NOT_SPECIFIED, C_BAD_REQUEST
750

    
751
            if PASSWORD not in body.keys():
752
                return E_IDENTITY_PASSWORD_NOT_SPECIFIED, C_BAD_REQUEST
753

    
754
            # parse required fields from the request
755
            identity_name = body[NAME]
756
            identity_password = body[PASSWORD]
757

    
758
            # check whether a certificated specified by the given ID exists
759
            if cert is None:
760
                Logger.error(f"No such certificate found 'ID = {v}'.")
761
                return E_NO_CERTIFICATES_FOUND, C_NOT_FOUND
762
            else:
763
                # try to load it's private key
764
                key = self.key_service.get_key(cert.private_key_id)
765
                if key is None:
766
                    Logger.error(
767
                        f"The private key 'ID = {cert.private_key_id}'of the certificate 'ID = {cert.certificate_id}' does not exist.")
768
                    return E_NO_CERTIFICATES_FOUND, C_INTERNAL_SERVER_ERROR
769
                else:
770
                    # generate PKCS12 identity
771
                    identity_byte_array = self.certificate_service.generate_pkcs_identity(cert, key,
772
                                                                                          identity_name,
773
                                                                                          identity_password)
774
                    return Response(identity_byte_array, mimetype='application/x-pkcs12')
775

    
776
    @staticmethod
777
    def handle_cryptography_error(e):
778
        Logger.error(f"An unhandled CryptographyException has been raised: {str(e)}")
779
        return E_UNHANDLED_CRYPTOGRAPHY_ERROR, C_INTERNAL_SERVER_ERROR
780

    
781
    @staticmethod
782
    def handle_database_error(e):
783
        Logger.error(f"An unhandled DatabaseException has been raised: {str(e)}")
784
        return E_UNHANDLED_DATABASE_ERROR, C_INTERNAL_SERVER_ERROR
785

    
786
    @staticmethod
787
    def handle_generic_exception(e):
788
        Logger.error(f"An unknown Exception ({type(e)}) has been raised: {str(e)}")
789
        return E_UNHANDLED_ERROR, C_INTERNAL_SERVER_ERROR
(2-2/4)