Revize 88a68c29

Re #8476 - Added as a part of the swagger-generated server stub. Imported required objects, SQLite3 thread issue hack (performance impact negligible, violates encapsulation), implemented `create_certificate`.

swagger_server/controllers/certificates_controller.py

@@ +1 @@
+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 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():
+    _ = 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!'