aswi2021jmsd-gitlab/src/controllers/certificates_controller.py @ eb2ea1ff

from datetime import datetime
from itertools import chain
from flask import request
from src.dao.private_key_repository import PrivateKeyRepository
from src.model.subject import Subject
from src.services.certificate_service import CertificateService
from src.dao.certificate_repository import CertificateRepository  # TODO not the Controller's responsibility. 1
from src.services.cryptography import CryptographyService  # TODO not the Controller's responsibility. 2
from sqlite3 import Connection  # TODO not the Controller's responsibility. 3
from src.constants import DICT_USAGES, CA_ID, \
    DATABASE_FILE_LOCATION, SSL_ID, SIGNATURE_ID, AUTHENTICATION_ID, \
    DATETIME_FORMAT, ROOT_CA_ID, INTERMEDIATE_CA_ID, CERTIFICATE_ID  # TODO DATABASE_FILE - not the Controller's
#  responsibility. 4
from src.services.key_service import KeyService

# _ = Connection("../" + DATABASE_FILE)                                 # TODO not the Controller's responsibility. 5
# cursor = _.cursor()                                                   # TODO responsibility of the
                                                                        #  CertificateRepository. It makes no sense to
                                                                        #  supply a different cursor than precisely
                                                                        #  the one corresponding to the connection.
                                                                        #  The cursor can always be generated from the
                                                                        #  connection instance.
FILTERING = "filtering"
ISSUER = "issuer"
US = "usage"
NOT_AFTER = "notAfter"
NOT_BEFORE = "notBefore"
COMMON_NAME = "CN"
ID = "id"
E_NO_ISSUER_FOUND = {"success": False,
                           "data": "No certificate authority with such unique ID exists."}
E_NO_CERTIFICATES_FOUND = {"success": False, "data": "No certificates 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."}
USAGE = "usage"
SUBJECT = "subject"
VALIDITY_DAYS = "validityDays"
CA = "CA"

__ = CryptographyService()  # TODO not the Controller's responsibility. 6
CERTIFICATE_SERVICE = CertificateService(__, CertificateRepository(None, None))
# TODO open for discussion. Expected:
#  CS = CertificateService.get_instance()
#  or something like that.

KEY_SERVICE = KeyService(__, PrivateKeyRepository(None, None))  # TODO as above


class CertController:

    @staticmethod
    def setup():
        """
        SQLite3 thread issue hack.
        :return:
        """
        _ = Connection(DATABASE_FILE_LOCATION.shortest_relative_path())
        CERTIFICATE_SERVICE.certificate_repository.connection = _
        CERTIFICATE_SERVICE.certificate_repository.cursor = _.cursor()
        KEY_SERVICE.private_key_repository.connection = _
        KEY_SERVICE.private_key_repository.cursor = _.cursor()

    @staticmethod
    def create_certificate():  # noqa: E501
        """create new certificate

        Create a new certificate based on given information # noqa: E501

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

        :rtype: CreatedResponse
        """
        CertController.setup()  # TODO remove after issue fixed

        key_map = {'CA': CA_ID, 'SSL': SSL_ID, 'digitalSignature': SIGNATURE_ID, 'authentication': AUTHENTICATION_ID}
        required_keys = {SUBJECT, USAGE, VALIDITY_DAYS}

        if request.is_json:
            body = request.get_json()
            if not all(k in body for k in required_keys):
                return E_MISSING_PARAMETERS, 400

            if not isinstance(body[VALIDITY_DAYS], int):
                return E_WRONG_PARAMETERS, 400

            subject = Subject.from_dict(body[SUBJECT])

            if subject is None:
                return E_WRONG_PARAMETERS, 400

            usages_dict = DICT_USAGES.copy()

            if not isinstance(body[USAGE], dict):
                return E_WRONG_PARAMETERS, 400

            for k, v in body[USAGE].items():
                if k not in key_map:
                    return E_WRONG_PARAMETERS, 400
                usages_dict[key_map[k]] = v

            key = KEY_SERVICE.create_new_key()  # TODO pass key

            if CA not in body:
                cert = CERTIFICATE_SERVICE.create_root_ca(
                    key,
                    subject,
                    usages=usages_dict,
                    days=body[VALIDITY_DAYS]
                )
            else:
                issuer = CERTIFICATE_SERVICE.get_certificate(body[CA])

                if issuer is None:
                    return E_NO_ISSUER_FOUND, 400

                issuer_key = KEY_SERVICE.get_key(issuer.private_key_id)

                if issuer_key is None:
                    return E_CORRUPTED_DATABASE, 500

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

                cert = f(
                    key,
                    subject,
                    issuer,
                    issuer_key,
                    usages=usages_dict,
                    days=body[VALIDITY_DAYS]
                )

            if cert is not None:
                return {"success": True,
                        "data": cert.certificate_id}, 201
            else:
                return {"success": False,
                        "data": "Internal error: The certificate could not have been created."}, 400
        else:
            return E_NOT_JSON_FORMAT, 400

    @staticmethod
    def get_certificate_by_id(id):  # noqa: E501
        """get certificate by ID

        Get certificate in PEM format by ID # noqa: E501

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

        :rtype: PemResponse
        """
        CertController.setup()  # TODO remove after issue fixed

        try:
            v = int(id)
        except ValueError:
            return E_WRONG_PARAMETERS, 400

        cert = CERTIFICATE_SERVICE.get_certificate(v)

        if cert is None:
            return E_NO_CERTIFICATES_FOUND, 205                 # TODO related to 204 issue
        else:
            return {"success": True, "data": cert.pem_data}

    @staticmethod
    def get_certificate_details_by_id(id):  # noqa: E501
        """get certificate's details by ID

        Get certificate details by ID # noqa: E501

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

        :rtype: CertificateResponse
        """
        CertController.setup()  # TODO remove after issue fixed

        if connexion.request.is_json:
            id = IdParameter.from_dict(connexion.request.get_json())  # noqa: E501
        return 'do some magic!'

    @staticmethod
    def get_certificate_list():  # noqa: E501
        """get list of certificates

        Lists certificates based on provided filtering options # noqa: E501

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

        :rtype: CertificateListResponse
        """
        CertController.setup()  # TODO remove after issue fixed

        key_map = {CA_ID: 'CA', SSL_ID: 'SSL', SIGNATURE_ID: 'digitalSignature', AUTHENTICATION_ID: 'authentication'}

        targets = {ROOT_CA_ID, INTERMEDIATE_CA_ID, CERTIFICATE_ID}
        if request.is_json:
            data = request.get_json()
            if FILTERING in data:
                if isinstance(data[FILTERING], dict):
                    if CA in data[FILTERING]:
                        if isinstance(data[FILTERING][CA], bool):
                            if data[FILTERING][CA]:
                                targets.remove(CERTIFICATE_ID)
                            else:
                                targets.remove(ROOT_CA_ID)
                                targets.remove(INTERMEDIATE_CA_ID)
                        else:
                            return E_WRONG_PARAMETERS, 400
                else:
                    return E_WRONG_PARAMETERS, 400

        if len(targets) == 3:
            certs = CERTIFICATE_SERVICE.get_certificates()
        else:
            certs = list(chain(*(CERTIFICATE_SERVICE.get_certificates(target) for target in targets)))

        if certs is None:
            return E_GENERAL_ERROR, 500
        elif len(certs) == 0:
            return E_NO_CERTIFICATES_FOUND, 205         # TODO related to 204 issue
        else:
            ret = []
            for c in certs:
                c_issuer = CERTIFICATE_SERVICE.get_certificate(c.parent_id)
                if c_issuer is None:
                    return E_CORRUPTED_DATABASE, 500

                ret.append(
                    {
                        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: {key_map[k]: v for k, v in c.usages.items()},
                        ISSUER: {
                            ID: c_issuer.certificate_id,
                            COMMON_NAME: c_issuer.common_name
                        }
                    }
                )
            return {"success": True, "data": ret}



    @staticmethod
    def get_certificate_root_by_id(id):  # noqa: E501
        """get certificate's root of trust chain by ID

        Get certificate's root of trust chain in PEM format by ID # noqa: E501

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

        :rtype: PemResponse
        """
        CertController.setup()  # TODO remove after issue fixed

        if connexion.request.is_json:
            id = IdParameter.from_dict(connexion.request.get_json())  # noqa: E501
        return 'do some magic!'

    @staticmethod
    def get_certificate_trust_chain_by_id(id):  # noqa: E501
        """get certificate's trust chain by ID

        Get certificate trust chain in PEM format by ID # noqa: E501

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

        :rtype: PemResponse
        """
        CertController.setup()  # TODO remove after issue fixed

        if connexion.request.is_json:
            id = IdParameter.from_dict(connexion.request.get_json())  # noqa: E501
        return 'do some magic!'