aswi2021jmsd-gitlab4/swagger_server/controllers/certificates_controller.py @ 08f22957

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, Cursor                              # TODO not the Controller's responsibility. 3
from src.constants import DATABASE_FILE, DICT_USAGES, CA_ID         # TODO DATABASE_FILE - not the Controller's
                                                                    #  responsibility. 4
from src.services.key_service import KeyService
from swagger_server.models import CertificateRequest

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.

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

KEY_SERVICE = KeyService(__, PrivateKeyRepository(_, _.cursor()))       # TODO as above


def setup():
    """
    SQLite3 thread issue hack.
    :return:
    """
    _ = Connection("../" + DATABASE_FILE)
    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()

    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="Internal server error (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 CertificateResponse(
                success=True,
                data=cert.certificate_id
            ), 201
        else:
            return ErrorResponse(
                success=False,
                data="The certificate could not have been created."
            ), 400
    else:
        return 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
    """
    if connexion.request.is_json:
        filtering = Filtering.from_dict(connexion.request.get_json())  # noqa: E501
    return 'do some magic! XD'


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!'