Projekt

Obecné

Profil

Stáhnout (34.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, InvalidSubjectAttribute
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

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

    
75

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

    
81
    def create_certificate(self):
82
        """create new certificate
83

    
84
        Create a new certificate based on given information
85

    
86
        :param body: Certificate data to be created
87
        :type body: dict | bytes
88

    
89
        :rtype: CreatedResponse
90
        """
91

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

    
95
        required_keys = {SUBJECT, USAGE, VALIDITY_DAYS}                             # required fields of the POST req
96

    
97
        if request.is_json:                                                         # accept JSON only
98
            body = request.get_json()
99

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

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

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

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

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

    
117
            usages_dict = {}
118

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

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

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

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

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

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

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

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

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

    
196
            # if extensions are specified and CryptoException occurs, the problem is probably in the
197
            # extensions format - otherwise error 500 is expected
198
            except CryptographyException as e:
199
                if len(extensions) > 0:
200
                    return E_INVALID_EXTENSIONS, C_BAD_REQUEST
201
                else:
202
                    raise CryptographyException(e.executable, e.args, e.message)
203
            except InvalidSubjectAttribute as e:
204
                Logger.warning(str(e))
205
                return {"success": False, "data": str(e)}, C_BAD_REQUEST
206

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

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

    
224
        Get certificate in PEM format by ID
225

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

    
229
        :rtype: PemResponse
230
        """
231

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

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

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

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

    
252
        Get certificate details by ID
253

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

    
257
        :rtype: CertificateResponse
258
        """
259

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

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

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

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

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

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

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

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

    
291
        Lists certificates based on provided filtering options
292

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

    
296
        :rtype: CertificateListResponse
297
        """
298

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

    
302

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

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

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

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

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

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

    
338
        unfiltered = True
339

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

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

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

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

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

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

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

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

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

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

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

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

    
445
        :rtype: PemResponse
446
        """
447

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

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

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

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

    
464
        trust_chain = self.certificate_service.get_chain_of_trust(cert.parent_id, exclude_root=False)
465
        if trust_chain is None or len(trust_chain) == 0:
466
            Logger.error(f"No such certificate found (empty list).")
467
            return E_NO_CERTIFICATES_FOUND, C_NO_DATA
468

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

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

    
474
        Get certificate trust chain in PEM format by ID
475

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

    
479
        :rtype: PemResponse
480
        """
481

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

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

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

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

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

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

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

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

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

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

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

    
526
        required_keys = {STATUS}  # required keys
527

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

    
628
        :rtype: PemResponse
629
        """
630

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

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

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

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

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

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

    
664
        :rtype: PemResponse
665
        """
666

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

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

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

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

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

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

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

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

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

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

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

    
726
        :rtype: Response
727
        """
728

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

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

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

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

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

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

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

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