aswi2021jmsd-gitlab/swagger_server/controllers/certificates_controller.py @ 1e07432d

from datetime import datetime

import connexion
import six

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  # TODO DATABASE_FILE - not the Controller's
                                                                    #  responsibility. 4
from src.services.key_service import KeyService
from swagger_server.models import CertificateRequest, CertificateListItem, CAUsage, IssuerListItem

from swagger_server.models.certificate import Certificate  # noqa: E501
from swagger_server.models.certificate_list_response import CertificateListResponse  # noqa: E501
from swagger_server.models.certificate_response import CertificateResponse  # noqa: E501
from swagger_server.models.created_response import CreatedResponse  # noqa: E501
from swagger_server.models.error_response import ErrorResponse  # noqa: E501
from swagger_server.models.filtering import Filtering  # noqa: E501
from swagger_server.models.id_parameter import IdParameter  # noqa: E501
from swagger_server.models.pem_response import PemResponse  # noqa: E501
from swagger_server import util


# _ = 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.
GENERAL_ERROR = "An error occured during processing of the request."
CORRUPTED_DATABASE = "Internal server error (corrupted database)."

__ = 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



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()


def create_certificate(body=None):  # 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
    """
    setup()                                                             # TODO remove after issue fixed

    if connexion.request.is_json:
        body = CertificateRequest.from_dict(connexion.request.get_json())  # noqa: E501
        # if body.subject is None or body.usage is None or body.validity_days is None:
        #     return 400

        key = KEY_SERVICE.create_new_key()                              # TODO pass key
        subject = Subject(
            common_name=body.subject.cn,
            country=body.subject.c,
            locality=body.subject.l,
            state=body.subject.st,
            organization=body.subject.o,
            organization_unit=body.subject.ou,
            email_address=body.subject.email_address
        )
        usages_dict = DICT_USAGES.copy()

        if body.ca is None:
            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 ErrorResponse(
                    success=False,
                    data="No certificate authority with such unique ID exists."
                ), 400

            issuer_key = KEY_SERVICE.get_key(issuer.private_key_id)

            if issuer_key is None:
                return ErrorResponse(
                    success=False,
                    data=CORRUPTED_DATABASE
                ), 400

            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 CreatedResponse(
                success=True,
                data=cert.certificate_id
            ), 201
        else:
            return ErrorResponse(
                success=False,
                data="The certificate could not have been created."
            ), 400
    else:
        return ErrorResponse(success=False, data="The request must be JSON-formatted."), 400


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
    """
    if connexion.request.is_json:
        id = IdParameter.from_dict(connexion.request.get_json())  # noqa: E501
    return 'do some magic!'


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
    """
    if connexion.request.is_json:
        id = IdParameter.from_dict(connexion.request.get_json())  # noqa: E501
    return 'do some magic!'


def get_certificate_list(filtering=None):  # 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
    """
    setup()                                                             # TODO remove after issue fixed

    key_map = {CA_ID: 'ca', SSL_ID: 'ssl', SIGNATURE_ID: 'digital_signature', AUTHENTICATION_ID: 'authentication'}

    if len(connexion.request.data) == 0:
        certs = CERTIFICATE_SERVICE.get_certificates()
        if certs is None:
            return ErrorResponse(success=False, data=GENERAL_ERROR), 500
        elif len(certs) == 0:
            return ErrorResponse(success=False, data="No certificates found."), 204
        else:
            ret = []
            for c in certs:
                c_issuer = CERTIFICATE_SERVICE.get_certificate(c.parent_id)
                if c_issuer is None:
                    return ErrorResponse(success=False, data=CORRUPTED_DATABASE)

                ret.append(
                    CertificateListItem(
                        id=c.certificate_id,
                        cn=c.common_name,
                        not_before=datetime.strptime(c.valid_from, DATETIME_FORMAT).date(),
                        not_after=datetime.strptime(c.valid_to, DATETIME_FORMAT).date(),
                        usage=CAUsage(**{key_map[k]: v for k, v in c.usages.items()}),
                        issuer=IssuerListItem(
                            id=c_issuer.certificate_id,
                            cn=c_issuer.common_name
                        )
                    )
                )
            return CertificateListResponse(success=True, data=ret)
    else:
        # TODO fix filtering issue (somehow)
        return ErrorResponse(success=False, data="The request must be JSON-formatted."), 400


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
    """
    if connexion.request.is_json:
        id = IdParameter.from_dict(connexion.request.get_json())  # noqa: E501
    return 'do some magic!'


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
    """
    if connexion.request.is_json:
        id = IdParameter.from_dict(connexion.request.get_json())  # noqa: E501
    return 'do some magic!'