Projekt

Obecné

Profil

Stáhnout (34.2 KB) Statistiky
| Větev: | Tag: | Revize:
1
import json
2
from datetime import datetime
3
from itertools import chain
4
from json import JSONDecodeError
5

    
6
from flask import request, Response
7
from injector import inject
8

    
9
from src.constants import CA_ID, \
10
    SSL_ID, SIGNATURE_ID, AUTHENTICATION_ID, \
11
    DATETIME_FORMAT, ROOT_CA_ID, INTERMEDIATE_CA_ID, CERTIFICATE_ID  # TODO DATABASE_FILE - not the Controller's
12
from src.controllers.return_codes import *
13
from src.exceptions.database_exception import DatabaseException
14
from src.exceptions.unknown_exception import UnknownException
15
from src.model.subject import Subject
16
from src.services.certificate_service import CertificateService, RevocationReasonInvalidException, \
17
    CertificateStatusInvalidException, CertificateNotFoundException, CertificateAlreadyRevokedException, \
18
    CertificateCannotBeSetToValid, 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

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

    
67

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

    
74

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

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

    
83
        Create a new certificate based on given information
84

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

    
88
        :rtype: CreatedResponse
89
        """
90

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

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

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

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

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

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

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

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

    
116
            usages_dict = {}
117

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

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

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

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

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

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

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

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

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

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

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

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

    
223
        Get certificate in PEM format by ID
224

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

    
228
        :rtype: PemResponse
229
        """
230

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

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

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

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

    
251
        Get certificate details by ID
252

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

    
256
        :rtype: CertificateResponse
257
        """
258

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

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

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

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

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

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

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

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

    
290
        Lists certificates based on provided filtering options
291

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

    
295
        :rtype: CertificateListResponse
296
        """
297

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

    
301

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

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

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

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

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

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

    
337
        unfiltered = True
338

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

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

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

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

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

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

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

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

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

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

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

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

    
444
        :rtype: PemResponse
445
        """
446

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

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

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

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

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

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

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

    
473
        Get certificate trust chain in PEM format by ID
474

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

    
478
        :rtype: PemResponse
479
        """
480

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

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

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

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

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

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

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

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

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

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

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

    
525
        required_keys = {STATUS}  # required keys
526

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

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

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

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

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

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

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

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

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

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

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

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

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

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

    
624
        :param id: ID of a certificate whose private 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
            # certificate exists, fetch it's private key
649
            private_key = self.key_service.get_key(cert.private_key_id)
650
            if cert is None:
651
                Logger.error(f"Internal server error (certificate's private key cannot be found).")
652
                return E_NO_CERT_PRIVATE_KEY_FOUND, C_INTERNAL_SERVER_ERROR
653
            else:
654
                return {"success": True, "data": private_key.private_key}, C_SUCCESS
655

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

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

    
663
        :rtype: PemResponse
664
        """
665

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

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

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

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

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

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

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

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

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

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

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

    
725
        :rtype: Response
726
        """
727

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

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

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

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

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

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

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

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