aswi2021jmsd-gitlab4/src/controllers/certificates_controller.py @ da0fc952

import json
from datetime import datetime
from itertools import chain
from json import JSONDecodeError

from flask import request, Response
from injector import inject

from src.constants import CA_ID, \
    SSL_ID, SIGNATURE_ID, AUTHENTICATION_ID, \
    DATETIME_FORMAT, ROOT_CA_ID, INTERMEDIATE_CA_ID, CERTIFICATE_ID  # TODO DATABASE_FILE - not the Controller's
from src.controllers.return_codes import *
from src.exceptions.database_exception import DatabaseException
from src.exceptions.unknown_exception import UnknownException
from src.model.subject import Subject
from src.services.certificate_service import CertificateService, RevocationReasonInvalidException, \
    CertificateStatusInvalidException, CertificateNotFoundException, CertificateAlreadyRevokedException, \
    CertificateCannotBeSetToValid
#  responsibility.
from src.services.cryptography import CryptographyException
from src.services.key_service import KeyService
from src.utils.logger import Logger
from src.utils.util import dict_to_string

EXTENSIONS = "extensions"

TREE_NODE_TYPE_COUNT = 3
KEY_PEM = "key_pem"
PASSWORD = "password"
KEY = "key"
FILTERING = "filtering"
PAGE = "page"
PER_PAGE = "per_page"
ISSUER = "issuer"
US = "usage"
NOT_AFTER = "notAfter"
NOT_BEFORE = "notBefore"
COMMON_NAME = "CN"
ID = "id"
CA = "CA"
USAGE = "usage"
SUBJECT = "subject"
VALIDITY_DAYS = "validityDays"
TYPE = "type"
ISSUED_BY = "issuedby"
STATUS = "status"
REASON = "reason"
REASON_UNDEFINED = "unspecified"
NAME = "name"
PASSWORD = "password"

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

E_UNHANDLED_CRYPTOGRAPHY_ERROR = {"success": False, "data": "An unknown error has happened in the cryptography library."}


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


    @inject
    def __init__(self, certificate_service: CertificateService, key_service: KeyService):
        self.certificate_service = certificate_service
        self.key_service = key_service

    def create_certificate(self):
        """create new certificate

        Create a new certificate based on given information

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

        :rtype: CreatedResponse
        """

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

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

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

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

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

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

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

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

            usages_dict = {}

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

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

            if KEY in body:
                if isinstance(body[KEY], dict):
                    if PASSWORD in body[KEY]:
                        passphrase = body[KEY][PASSWORD]
                        if KEY_PEM in body[KEY]:
                            key_pem = body[KEY][KEY_PEM]
                            if not self.key_service.verify_key(key_pem, passphrase=passphrase):
                                Logger.error(f"Passphrase specified but invalid.")
                                return E_WRONG_PASSWORD, C_BAD_REQUEST
                            key = self.key_service.wrap_custom_key(key_pem, passphrase=passphrase)
                        else:
                            key = self.key_service.create_new_key(passphrase)
                    else:
                        if KEY_PEM in body[KEY]:
                            key_pem = body[KEY][KEY_PEM]
                            if not self.key_service.verify_key(key_pem, passphrase=None):
                                Logger.error("Passphrase ommited but required.")
                                return E_WRONG_PASSWORD, C_BAD_REQUEST
                            key = self.key_service.wrap_custom_key(key_pem, passphrase=None)
                        else:
                            return E_WRONG_PARAMETERS, C_BAD_REQUEST                # if "key" exists but is empty
                else:
                    return E_WRONG_PARAMETERS, C_BAD_REQUEST
            else:
                key = self.key_service.create_new_key()                             # if "key" does not exist

            extensions = ""
            if EXTENSIONS in body:
                extensions = body[EXTENSIONS]
            try:
                if CA not in body or body[CA] is None:                                  # if issuer omitted (legal) or none
                    cert = self.certificate_service.create_root_ca(                     # create a root CA
                        key,
                        subject,
                        usages=usages_dict,                                             # TODO ignoring usages -> discussion
                        days=body[VALIDITY_DAYS],
                        extensions=extensions
                    )
                else:
                    issuer = self.certificate_service.get_certificate(body[CA])         # get base issuer info

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

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

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

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

                    # noinspection PyArgumentList
                    cert = f(                                                       # create inter CA or end cert
                        key,                                                        # according to whether 'CA' is among
                        subject,                                                    # the usages' fields
                        issuer,
                        issuer_key,
                        usages=usages_dict,
                        days=body[VALIDITY_DAYS],
                        extensions=extensions
                    )

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

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

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

        Get certificate in PEM format by ID

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

        :rtype: PemResponse
        """

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

        cert = self.certificate_service.get_certificate(v)

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

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

        Get certificate details by ID

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

        :rtype: CertificateResponse
        """

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

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

        cert = self.certificate_service.get_certificate(v)

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

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

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

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

    def get_certificate_list(self):
        """get list of certificates

        Lists certificates based on provided filtering options

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

        :rtype: CertificateListResponse
        """

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


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

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

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

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

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

        target_types = {ROOT_CA_ID, INTERMEDIATE_CA_ID, CERTIFICATE_ID}
        target_usages = None
        target_cn_substring = None
        issuer_id = -1

        unfiltered = True

        if PER_PAGE in data:
            unfiltered = False
            page = data.get(PAGE, 0)
            per_page = data[PER_PAGE]
        else:
            page = None
            per_page = None

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

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

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

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

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

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

        if unfiltered:                                                      # if not filtering
            certs = self.certificate_service.get_certificates()
        elif issuer_id >= 0:                                                # if filtering by an issuer
            try:
                                                                            # get his children, filtered
                certs = self.certificate_service.get_certificates_issued_by_filter(
                    issuer_id=issuer_id,
                    target_types=target_types,
                    target_usages=target_usages,
                    target_cn_substring=target_cn_substring,
                    page=page,
                    per_page=per_page
                )
            except CertificateNotFoundException:                            # if id does not exist
                Logger.error(f"No such certificate found 'ID = {issuer_id}'.")
                return E_NO_CERTIFICATES_FOUND, C_NOT_FOUND                 # throw
        else:
            certs = self.certificate_service.get_certificates_filter(
                target_types=target_types,
                target_usages=target_usages,
                target_cn_substring=target_cn_substring,
                page=page,
                per_page=per_page
            )

        if certs is None:
            Logger.error(f"Internal server error (unknown origin).")
            return E_GENERAL_ERROR, C_INTERNAL_SERVER_ERROR
        elif len(certs) == 0:
            # TODO check log level
            Logger.warning(f"No such certificate found (empty list).")
            return {"success": True, "data": []}, C_SUCCESS
        else:
            ret = []
            for c in certs:
                data = self.cert_to_dict_partial(c)
                if data is None:
                    Logger.error(f"Internal server error (corrupted database).")
                    return E_CORRUPTED_DATABASE, C_INTERNAL_SERVER_ERROR
                ret.append(
                    data
                )
            return {"success": True, "data": ret}, C_SUCCESS

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

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

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

        :rtype: PemResponse
        """

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

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

        cert = self.certificate_service.get_certificate(v)

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

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

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

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

        Get certificate trust chain in PEM format by ID

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

        :rtype: PemResponse
        """

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

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

        cert = self.certificate_service.get_certificate(v)

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

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

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

        ret = []
        for intermediate in trust_chain:
            ret.append(intermediate.pem_data)

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

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

        :rtype: SuccessResponse | ErrorResponse (see OpenAPI definition)
        """

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

        required_keys = {STATUS}  # required keys

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

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

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

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

            # get status and reason from the request
            status = request_body[STATUS]
            reason = request_body.get(REASON, REASON_UNDEFINED)
            try:
                # set certificate status using certificate_service
                self.certificate_service.set_certificate_revocation_status(identifier, status, reason)
            except (RevocationReasonInvalidException, CertificateStatusInvalidException):
                # these exceptions are thrown in case invalid status or revocation reason is passed to the controller
                Logger.error(f"Invalid request, wrong parameters.")
                return E_WRONG_PARAMETERS, C_BAD_REQUEST
            except CertificateAlreadyRevokedException:
                Logger.error(f"Certificate is already revoked 'ID = {identifier}'.")
                return E_NO_CERTIFICATE_ALREADY_REVOKED, C_BAD_REQUEST
            except CertificateNotFoundException:
                Logger.error(f"No such certificate found 'ID = {identifier}'.")
                return E_NO_CERTIFICATES_FOUND, C_NOT_FOUND
            except CertificateCannotBeSetToValid as e:
                return {"success": False, "data": str(e)}, C_BAD_REQUEST
            return {"success": True,
                    "data": "Certificate status updated successfully."}, C_SUCCESS
        # throw an error in case the request does not contain a json body
        else:
            Logger.error(f"The request must be JSON-formatted.")
            return E_NOT_JSON_FORMAT, C_BAD_REQUEST

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

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

        c_issuer = self.certificate_service.get_certificate(c.parent_id)
        if c_issuer is None:
            return None

        return {
            ID: c.certificate_id,
            COMMON_NAME: c.common_name,
            NOT_BEFORE: datetime.strptime(c.valid_from, DATETIME_FORMAT).date(),
            NOT_AFTER: datetime.strptime(c.valid_to, DATETIME_FORMAT).date(),
            USAGE: {CertController.INVERSE_USAGE_KEY_MAP[k]: v for k, v in c.usages.items()},
            ISSUER: {
                ID: c_issuer.certificate_id,
                COMMON_NAME: c_issuer.common_name
            }
        }

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

        Logger.info(f"Function launched.")

        subj = self.certificate_service.get_subject_from_certificate(c)
        c_issuer = self.certificate_service.get_certificate(c.parent_id)
        if c_issuer is None:
            return None

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

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

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

        :rtype: PemResponse
        """

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

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

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

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

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

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

        :rtype: PemResponse
        """

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

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

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

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

    def delete_certificate(self, id):
        """
        Deletes a certificate identified by ID, including its corresponding subtree (all descendants).
        :param id: target certificate ID
        :rtype: DeleteResponse
        """

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

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

        try:
            self.certificate_service.delete_certificate(v)
        except CertificateNotFoundException:
            Logger.error(f"No such certificate found 'ID = {v}'.")
            return E_NO_CERTIFICATES_FOUND, C_NOT_FOUND
        except DatabaseException:
            Logger.error(f"Internal server error (corrupted database).")
            return E_CORRUPTED_DATABASE, C_INTERNAL_SERVER_ERROR
        except CertificateStatusInvalidException or RevocationReasonInvalidException or UnknownException:
            Logger.error(f"Internal server error (unknown origin).")
            return E_GENERAL_ERROR, C_INTERNAL_SERVER_ERROR

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

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

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

        :rtype: Response
        """

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

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

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

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

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

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

            # parse required fields from the request
            identity_name = body[NAME]
            identity_password = body[PASSWORD]

            # check whether a certificated specified by the given ID exists
            if cert is None:
                Logger.error(f"No such certificate found 'ID = {v}'.")
                return E_NO_CERTIFICATES_FOUND, C_NOT_FOUND
            else:
                # try to load it's private key
                key = self.key_service.get_key(cert.private_key_id)
                if key is None:
                    Logger.error(
                        f"The private key 'ID = {cert.private_key_id}'of the certificate 'ID = {cert.certificate_id}' does not exist.")
                    return E_NO_CERTIFICATES_FOUND, C_INTERNAL_SERVER_ERROR
                else:
                    # generate PKCS12 identity
                    identity_byte_array = self.certificate_service.generate_pkcs_identity(cert, key,
                                                                                          identity_name,
                                                                                          identity_password)
                    return Response(identity_byte_array, mimetype='application/x-pkcs12')

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