Revize 1fa243ca

Re #8569 Added application initialization and running dependency injection

app.py

@@ -1 @@
-from flask import Flask, redirect
 import os
+
+from flask import Flask, redirect
+from injector import Injector
+
+from src.config import configuration
+from src.config.connection_provider import ConnectionProvider
 from src.controllers.certificates_controller import CertController
 
-app = Flask(__name__)
+
+class X509ManagementApp(Flask):
+
+    def run(self, host=None, port=None, debug=None, load_dotenv=True, **options):
+        with self.app_context():
+            initialize_app()
+        super(X509ManagementApp, self).run(host=host, port=port, debug=debug, load_dotenv=load_dotenv, **options)
+
+
+app = X509ManagementApp(__name__)
+certificate_controller: CertController = None
 
 
 @app.route('/')
@@ -12 +27 @@
 
 @app.route('/api/certificates', methods=["POST"])
 def create_certificate():
-    return CertController.create_certificate()
+    return certificate_controller.create_certificate()
 
 
 @app.route('/api/certificates', methods=["GET"])
 def get_cert_list():
-    return CertController.get_certificate_list()
+    return certificate_controller.get_certificate_list()
 
 
 @app.route('/api/certificates/<id>', methods=["GET"])
 def get_cert(id):
-    return CertController.get_certificate_by_id(id)
+    return certificate_controller.get_certificate_by_id(id)
 
 
 @app.route('/api/certificates/<id>/details', methods=["GET"])
-def get_cert_detes(id):
-    return CertController.get_certificate_details_by_id(id)
+def get_cert_details(id):
+    return certificate_controller.get_certificate_details_by_id(id)
 
 
 @app.route('/api/certificates/<id>/root', methods=["GET"])
 def get_cert_root(id):
-    return CertController.get_certificate_root_by_id(id)
+    return certificate_controller.get_certificate_root_by_id(id)
 
 
 @app.route('/api/certificates/<id>/chain', methods=["GET"])
 def get_cert_chain(id):
-    return CertController.get_certificate_trust_chain_by_id(id)
+    return certificate_controller.get_certificate_trust_chain_by_id(id)
+
+
+def initialize_app():
+    global certificate_controller
+
+    injector = Injector([configuration.configure_default, ConnectionProvider])
+    certificate_controller = injector.get(CertController)
 
 
 if __name__ == '__main__':
-    host = "0.0.0.0"
-    port = 5000
+    app_host = "0.0.0.0"
+    app_port = 5000
 
     # TODO better load this from config.py
     if "FLASK_HOST" in os.environ:
-        host = os.environ["FLASK_HOST"]
+        app_host = os.environ["FLASK_HOST"]
 
     if "FLASK_PORT" in os.environ:
-        host = os.environ["FLASK_PORT"]
+        app_host = os.environ["FLASK_PORT"]
 
-    app.run(host=host, port=port)
+    app.run(host=app_host, port=app_port)

src/controllers/certificates_controller.py

@@ -4 +4 @@
 from json import JSONDecodeError
 
 from flask import request
-from src.dao.private_key_repository import PrivateKeyRepository
+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.model.subject import Subject
 from src.services.certificate_service import CertificateService
-from src.dao.certificate_repository import CertificateRepository        # TODO not the Controller's responsibility.
-from src.services.cryptography import CryptographyService               # TODO not the Controller's responsibility.
-from sqlite3 import Connection                                          # TODO not the Controller's responsibility.
-from src.constants import 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.
+#  responsibility.
 from src.services.key_service import KeyService
 
 TREE_NODE_TYPE_COUNT = 3
@@ -44 +42 @@
 C_INTERNAL_SERVER_ERROR = 500
 C_SUCCESS = 200
 
-__ = CryptographyService()                                              # TODO not the Controller's responsibility.
-CERTIFICATE_SERVICE = CertificateService(__, CertificateRepository(None, None))
-KEY_SERVICE = KeyService(__, PrivateKeyRepository(None, None))          # TODO not the Controller's responsibility.
-
 
 class CertController:
     KEY_MAP = {'CA': CA_ID, 'SSL': SSL_ID, 'digitalSignature': SIGNATURE_ID, 'authentication': AUTHENTICATION_ID}
     INVERSE_KEY_MAP = {k: v for v, k in KEY_MAP.items()}
 
-    @staticmethod
-    def setup():
-        """
-        SQLite3 thread issue hack.
-        :return:
-        """
-        _ = Connection("db/database_sqlite.db")
-        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():
+    @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
@@ -76 +62 @@
 
         :rtype: CreatedResponse
         """
-        CertController.setup()                                                      # TODO remove after issue fixed
-
         required_keys = {SUBJECT, USAGE, VALIDITY_DAYS}                             # required fields of the POST req
 
         if request.is_json:                                                         # accept JSON only
@@ -103 +87 @@
                     return E_WRONG_PARAMETERS, C_BAD_REQUEST                        # and throw if it is not
                 usages_dict[CertController.KEY_MAP[k]] = v                          # otherwise translate key and set
 
-            key = KEY_SERVICE.create_new_key()                                      # TODO pass key
+            key = self.key_service.create_new_key()                                      # TODO pass key
 
             if CA not in body or body[CA] is None:                                  # if issuer omitted (legal) or none
-                cert = CERTIFICATE_SERVICE.create_root_ca(                          # create a root CA
+                cert = self.certificate_service.create_root_ca(                          # create a root CA
                     key,
                     subject,
                     usages=usages_dict,                                             # TODO ignoring usages -> discussion
                     days=body[VALIDITY_DAYS]
                 )
             else:
-                issuer = CERTIFICATE_SERVICE.get_certificate(body[CA])              # get base issuer info
+                issuer = self.certificate_service.get_certificate(body[CA])              # get base issuer info
 
                 if issuer is None:                                                  # if such issuer does not exist
-                    KEY_SERVICE.delete_key(key.private_key_id)                      # free
+                    self.key_service.delete_key(key.private_key_id)                      # free
                     return E_NO_ISSUER_FOUND, C_BAD_REQUEST                         # and throw
 
-                issuer_key = KEY_SERVICE.get_key(issuer.private_key_id)             # get issuer's key, which must exist
+                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
-                    KEY_SERVICE.delete_key(key.private_key_id)                      # free
+                    self.key_service.delete_key(key.private_key_id)                      # free
                     return E_CORRUPTED_DATABASE, C_INTERNAL_SERVER_ERROR            # and throw
 
-                f = CERTIFICATE_SERVICE.create_ca if CA_ID in usages_dict and usages_dict[CA_ID] else \
-                    CERTIFICATE_SERVICE.create_end_cert
+                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 PyTypeChecker
                 cert = f(                                                           # create inter CA or end cert
@@ -142 +126 @@
                 return {"success": True,
                         "data": cert.certificate_id}, C_CREATED_SUCCESSFULLY
             else:                                                                   # if this fails, then
-                KEY_SERVICE.delete_key(key.private_key_id)                          # free
+                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
@@ -150 +134 @@
         else:
             return E_NOT_JSON_FORMAT, C_BAD_REQUEST                                 # throw in case of non-JSON format
 
-    @staticmethod
-    def get_certificate_by_id(id):
+    def get_certificate_by_id(self, id):
         """get certificate by ID
 
         Get certificate in PEM format by ID
@@ -161 +144 @@
 
         :rtype: PemResponse
         """
-        CertController.setup()                                                      # TODO remove after issue fixed
-
         try:
             v = int(id)
         except ValueError:
             return E_WRONG_PARAMETERS, C_BAD_REQUEST
 
-        cert = CERTIFICATE_SERVICE.get_certificate(v)
+        cert = self.certificate_service.get_certificate(v)
 
         if cert is None:
             return E_NO_CERTIFICATES_FOUND, C_NO_DATA                               
         else:
             return {"success": True, "data": cert.pem_data}, C_SUCCESS
 
-    @staticmethod
-    def get_certificate_details_by_id(id):
+    def get_certificate_details_by_id(self, id):
         """get certificate's details by ID
 
         Get certificate details by ID
@@ -186 +166 @@
 
         :rtype: CertificateResponse
         """
-        CertController.setup()                                                      # TODO remove after issue fixed
-
         try:
             v = int(id)
         except ValueError:
             return E_WRONG_PARAMETERS, C_BAD_REQUEST
 
-        cert = CERTIFICATE_SERVICE.get_certificate(v)
+        cert = self.certificate_service.get_certificate(v)
 
         if cert is None:
             return E_NO_CERTIFICATES_FOUND, C_NO_DATA                               
         else:
-            data = CertController.cert_to_dict_full(cert)
+            data = self.cert_to_dict_full(cert)
             if data is None:
                 return E_CORRUPTED_DATABASE, C_INTERNAL_SERVER_ERROR
             return {"success": True, "data": data}, C_SUCCESS
 
-    @staticmethod
-    def get_certificate_list():
+    def get_certificate_list(self):
         """get list of certificates
 
         Lists certificates based on provided filtering options
@@ -214 +191 @@
 
         :rtype: CertificateListResponse
         """
-        CertController.setup()                                                      # TODO remove after issue fixed
-
         targets = {ROOT_CA_ID, INTERMEDIATE_CA_ID, CERTIFICATE_ID}                  # all targets
 
         # the filtering parameter can be read as URL argument or as a request body
@@ -248 +223 @@
 
                                                                                     # if filtering did not change the
                                                                                     # targets,
-            certs = CERTIFICATE_SERVICE.get_certificates()                          # fetch everything
+            certs = self.certificate_service.get_certificates()                          # fetch everything
         else:                                                                       # otherwise fetch targets only
-            certs = list(chain(*(CERTIFICATE_SERVICE.get_certificates(target) for target in targets)))
+            certs = list(chain(*(self.certificate_service.get_certificates(target) for target in targets)))
 
         if certs is None:
             return E_GENERAL_ERROR, C_INTERNAL_SERVER_ERROR
@@ -259 +234 @@
         else:
             ret = []
             for c in certs:
-                data = CertController.cert_to_dict_partial(c)
+                data = self.cert_to_dict_partial(c)
                 if data is None:
                     return E_CORRUPTED_DATABASE, C_INTERNAL_SERVER_ERROR
                 ret.append(
@@ -267 +242 @@
                 )
             return {"success": True, "data": ret}, C_SUCCESS
 
-    @staticmethod
-    def get_certificate_root_by_id(id):
+    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
@@ -278 +252 @@
 
         :rtype: PemResponse
         """
-        CertController.setup()                                                      # TODO remove after issue fixed
-
         try:
             v = int(id)
         except ValueError:
             return E_WRONG_PARAMETERS, C_BAD_REQUEST
 
-        cert = CERTIFICATE_SERVICE.get_certificate(v)
+        cert = self.certificate_service.get_certificate(v)
 
         if cert is None:
             return E_NO_CERTIFICATES_FOUND, C_NO_DATA
 
-        trust_chain = CERTIFICATE_SERVICE.get_chain_of_trust(cert.parent_id, exclude_root=False)
-        if trust_chain is None or len(trust_chain) is 0:
+        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:
             return E_NO_CERTIFICATES_FOUND, C_NO_DATA
 
         return {"success": True, "data": trust_chain[-1].pem_data}, C_SUCCESS
 
-    @staticmethod
-    def get_certificate_trust_chain_by_id(id):
+    def get_certificate_trust_chain_by_id(self, id):
         """get certificate's trust chain by ID
 
         Get certificate trust chain in PEM format by ID
@@ -307 +278 @@
 
         :rtype: PemResponse
         """
-        CertController.setup()                                                      # TODO remove after issue fixed
 
         try:
             v = int(id)
         except ValueError:
             return E_WRONG_PARAMETERS, C_BAD_REQUEST
 
-        cert = CERTIFICATE_SERVICE.get_certificate(v)
+        cert = self.certificate_service.get_certificate(v)
 
         if cert is None:
             return E_NO_CERTIFICATES_FOUND, C_NO_DATA                               
@@ -322 +292 @@
         if cert.parent_id is None:
             return E_NO_CERTIFICATES_FOUND, C_NO_DATA
 
-        trust_chain = CERTIFICATE_SERVICE.get_chain_of_trust(cert.parent_id)
+        trust_chain = self.certificate_service.get_chain_of_trust(cert.parent_id)
 
         ret = []
         for intermediate in trust_chain:
@@ -330 +300 @@
 
         return {"success": True, "data": "".join(ret)}, C_SUCCESS
 
-    @staticmethod
-    def cert_to_dict_partial(c):
+    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)
         """
-        c_issuer = CERTIFICATE_SERVICE.get_certificate(c.parent_id)
+        c_issuer = self.certificate_service.get_certificate(c.parent_id)
         if c_issuer is None:
             return None
 
@@ -353 +322 @@
             }
         }
 
-    @staticmethod
-    def cert_to_dict_full(c):
+    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)
         """
-        subj = CERTIFICATE_SERVICE.get_subject_from_certificate(c)
-        c_issuer = CERTIFICATE_SERVICE.get_certificate(c.parent_id)
+        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