Revize 9247d70a

Re #8476 - Heavy refactoring and added comments.

src/controllers/certificates_controller.py

@@ -4 +4 @@
 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, \
+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. 4
+    DATETIME_FORMAT, ROOT_CA_ID, INTERMEDIATE_CA_ID, CERTIFICATE_ID     # TODO DATABASE_FILE - not the Controller's
+                                                                        #  responsibility.
 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.
+TREE_NODE_TYPE_COUNT = 3
+
 FILTERING = "filtering"
 ISSUER = "issuer"
 US = "usage"
@@ -27 +22 @@
 NOT_BEFORE = "notBefore"
 COMMON_NAME = "CN"
 ID = "id"
-E_NO_ISSUER_FOUND = {"success": False,
-                           "data": "No certificate authority with such unique ID exists."}
+USAGE = "usage"
+SUBJECT = "subject"
+VALIDITY_DAYS = "validityDays"
+CA = "CA"
+
+E_NO_ISSUER_FOUND = {"success": False, "data": "No certificate authority with such unique ID exists."}
 E_NO_CERTIFICATES_FOUND = {"success": False, "data": "No such certificate 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.
+C_CREATED_SUCCESSFULLY = 201
+C_BAD_REQUEST = 400
+C_NO_DATA = 205                                                         # TODO related to 204 issue
+C_INTERNAL_SERVER_ERROR = 500
+C_SUCCESS = 200
 
-KEY_SERVICE = KeyService(__, PrivateKeyRepository(None, None))  # TODO as above
+__ = 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:
@@ -66 +63 @@
         KEY_SERVICE.private_key_repository.cursor = _.cursor()
 
     @staticmethod
-    def create_certificate():  # noqa: E501
+    def create_certificate():
         """create new certificate
 
-        Create a new certificate based on given information # noqa: E501
+        Create a new certificate based on given information
 
         :param body: Certificate data to be created
         :type body: dict | bytes
 
         :rtype: CreatedResponse
         """
-        CertController.setup()  # TODO remove after issue fixed
+        CertController.setup()                                                      # TODO remove after issue fixed
 
-        required_keys = {SUBJECT, USAGE, VALIDITY_DAYS}
+        required_keys = {SUBJECT, USAGE, VALIDITY_DAYS}                             # required fields of the POST req
 
-        if request.is_json:
+        if request.is_json:                                                         # accept JSON only
             body = request.get_json()
-            if not all(k in body for k in required_keys):
-                return E_MISSING_PARAMETERS, 400
+            if not all(k in body for k in required_keys):                           # verify that all keys are present
+                return E_MISSING_PARAMETERS, C_BAD_REQUEST
 
-            if not isinstance(body[VALIDITY_DAYS], int):
-                return E_WRONG_PARAMETERS, 400
+            if not isinstance(body[VALIDITY_DAYS], int):                            # type checking
+                return E_WRONG_PARAMETERS, C_BAD_REQUEST
 
-            subject = Subject.from_dict(body[SUBJECT])
+            subject = Subject.from_dict(body[SUBJECT])                              # generate Subject from passed dict
 
-            if subject is None:
-                return E_WRONG_PARAMETERS, 400
+            if subject is None:                                                     # if the format is incorrect
+                return E_WRONG_PARAMETERS, C_BAD_REQUEST
 
             usages_dict = {}
 
-            if not isinstance(body[USAGE], dict):
-                return E_WRONG_PARAMETERS, 400
+            if not isinstance(body[USAGE], dict):                                   # type checking
+                return E_WRONG_PARAMETERS, C_BAD_REQUEST
 
-            for k, v in body[USAGE].items():
-                if k not in CertController.KEY_MAP:
-                    return E_WRONG_PARAMETERS, 400
-                usages_dict[CertController.KEY_MAP[k]] = v
+            for k, v in body[USAGE].items():                                        # for each usage
+                if k not in CertController.KEY_MAP:                                 # check that it is a valid usage
+                    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 = KEY_SERVICE.create_new_key()                                      # TODO pass key
 
-            if CA not in body:
-                cert = CERTIFICATE_SERVICE.create_root_ca(
+            if CA not in body:                                                      # if issuer omitted (legal)
+                cert = CERTIFICATE_SERVICE.create_root_ca(                          # create a root CA
                     key,
                     subject,
-                    usages=usages_dict,
+                    usages=usages_dict,                                             # TODO ignoring usages -> discussion
                     days=body[VALIDITY_DAYS]
                 )
             else:
-                issuer = CERTIFICATE_SERVICE.get_certificate(body[CA])
+                issuer = CERTIFICATE_SERVICE.get_certificate(body[CA])              # get base issuer info
 
-                if issuer is None:
-                    KEY_SERVICE.delete_key(key.private_key_id)
-                    return E_NO_ISSUER_FOUND, 400
+                if issuer is None:                                                  # if such issuer does not exist
+                    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)
+                issuer_key = KEY_SERVICE.get_key(issuer.private_key_id)             # get issuer's key, which must exist
 
-                if issuer_key is None:
-                    KEY_SERVICE.delete_key(key.private_key_id)
-                    return E_CORRUPTED_DATABASE, 500
+                if issuer_key is None:                                              # if it does not
+                    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
 
-                cert = f(
-                    key,
-                    subject,
+                # noinspection PyTypeChecker
+                cert = f(                                                           # create inter CA or end cert
+                    key,                                                            # according to whether 'CA' is among
+                    subject,                                                        # the usages' fields
                     issuer,
                     issuer_key,
                     usages=usages_dict,
@@ -139 +137 @@
 
             if cert is not None:
                 return {"success": True,
-                        "data": cert.certificate_id}, 201
-            else:
-                KEY_SERVICE.delete_key(key.private_key_id)
-                return {"success": False,
-                        "data": "Internal error: The certificate could not have been created."}, 400
+                        "data": cert.certificate_id}, C_CREATED_SUCCESSFULLY
+            else:                                                                   # if this fails, then
+                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
+                                                                                    # of error information
         else:
-            return E_NOT_JSON_FORMAT, 400
+            return E_NOT_JSON_FORMAT, C_BAD_REQUEST                                 # throw in case of non-JSON format
 
     @staticmethod
-    def get_certificate_by_id(id):  # noqa: E501
+    def get_certificate_by_id(id):
         """get certificate by ID
 
-        Get certificate in PEM format by ID # noqa: E501
+        Get certificate in PEM format by ID
 
         :param id: ID of a certificate to be queried
         :type id: dict | bytes
 
         :rtype: PemResponse
         """
-        CertController.setup()  # TODO remove after issue fixed
+        CertController.setup()                                                      # TODO remove after issue fixed
 
         try:
             v = int(id)
         except ValueError:
-            return E_WRONG_PARAMETERS, 400
+            return E_WRONG_PARAMETERS, C_BAD_REQUEST
 
         cert = CERTIFICATE_SERVICE.get_certificate(v)
 
         if cert is None:
-            return E_NO_CERTIFICATES_FOUND, 205                 # TODO related to 204 issue
+            return E_NO_CERTIFICATES_FOUND, C_NO_DATA                               
         else:
-            return {"success": True, "data": cert.pem_data}
+            return {"success": True, "data": cert.pem_data}, C_SUCCESS
 
     @staticmethod
-    def get_certificate_details_by_id(id):  # noqa: E501
+    def get_certificate_details_by_id(id):
         """get certificate's details by ID
 
-        Get certificate details by ID # noqa: E501
+        Get certificate details by ID
 
         :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
+        CertController.setup()                                                      # TODO remove after issue fixed
 
         try:
             v = int(id)
         except ValueError:
-            return E_WRONG_PARAMETERS, 400
+            return E_WRONG_PARAMETERS, C_BAD_REQUEST
 
         cert = CERTIFICATE_SERVICE.get_certificate(v)
 
         if cert is None:
-            return E_NO_CERTIFICATES_FOUND, 205  # TODO related to 204 issue
+            return E_NO_CERTIFICATES_FOUND, C_NO_DATA                               
         else:
             data = CertController.cert_to_dict_full(cert)
             if data is None:
-                return E_CORRUPTED_DATABASE, 500
-            return {"success": True, "data": data}
+                return E_CORRUPTED_DATABASE, C_INTERNAL_SERVER_ERROR
+            return {"success": True, "data": data}, C_SUCCESS
 
     @staticmethod
-    def get_certificate_list():  # noqa: E501
+    def get_certificate_list():
         """get list of certificates
 
-        Lists certificates based on provided filtering options # noqa: E501
+        Lists certificates based on provided filtering options
 
         :param filtering: Filter certificate type to be queried
         :type filtering: dict | bytes
 
         :rtype: CertificateListResponse
         """
-        CertController.setup()  # TODO remove after issue fixed
-
-        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]:
+        CertController.setup()                                                      # TODO remove after issue fixed
+
+        targets = {ROOT_CA_ID, INTERMEDIATE_CA_ID, CERTIFICATE_ID}                  # all targets
+        if request.is_json:                                                         # if the request carries JSON data
+            data = request.get_json()                                               # get it
+            if FILTERING in data:                                                   # if the 'filtering' field exists
+                if isinstance(data[FILTERING], dict):                               # and it is also a 'dict'
+                    if CA in data[FILTERING]:                                       # containing 'CA'
+                        if isinstance(data[FILTERING][CA], bool):                   # which is a 'bool',
+                            if data[FILTERING][CA]:                                 # then filter according to 'CA'.
                                 targets.remove(CERTIFICATE_ID)
                             else:
                                 targets.remove(ROOT_CA_ID)
                                 targets.remove(INTERMEDIATE_CA_ID)
                         else:
-                            return E_WRONG_PARAMETERS, 400
+                            return E_WRONG_PARAMETERS, C_BAD_REQUEST
                 else:
-                    return E_WRONG_PARAMETERS, 400
+                    return E_WRONG_PARAMETERS, C_BAD_REQUEST
 
-        if len(targets) == 3:                               # potentially risky
-            certs = CERTIFICATE_SERVICE.get_certificates()
-        else:
+        if len(targets) == TREE_NODE_TYPE_COUNT:                                    # = 3 -> root node,
+                                                                                    # intermediate node,
+                                                                                    # or leaf node
+
+                                                                                    # if filtering did not change the
+                                                                                    # targets,
+            certs = CERTIFICATE_SERVICE.get_certificates()                          # fetch everything
+        else:                                                                       # otherwise fetch targets only
             certs = list(chain(*(CERTIFICATE_SERVICE.get_certificates(target) for target in targets)))
 
         if certs is None:
-            return E_GENERAL_ERROR, 500
+            return E_GENERAL_ERROR, C_INTERNAL_SERVER_ERROR
         elif len(certs) == 0:
-            return E_NO_CERTIFICATES_FOUND, 205         # TODO related to 204 issue
+            return E_NO_CERTIFICATES_FOUND, C_NO_DATA                               
         else:
             ret = []
             for c in certs:
                 data = CertController.cert_to_dict_partial(c)
                 if data is None:
-                    return E_CORRUPTED_DATABASE, 500
+                    return E_CORRUPTED_DATABASE, C_INTERNAL_SERVER_ERROR
                 ret.append(
                     data
                 )
-            return {"success": True, "data": ret}
+            return {"success": True, "data": ret}, C_SUCCESS
 
     @staticmethod
-    def get_certificate_root_by_id(id):  # noqa: E501
+    def get_certificate_root_by_id(id):
         """get certificate's root of trust chain by ID
 
-        Get certificate's root of trust chain in PEM format by ID # noqa: E501
+        Get certificate's root of trust chain in PEM format by ID
 
         :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
+        CertController.setup()                                                      # TODO remove after issue fixed
 
         try:
             v = int(id)
         except ValueError:
-            return E_WRONG_PARAMETERS, 400
+            return E_WRONG_PARAMETERS, C_BAD_REQUEST
 
         cert = CERTIFICATE_SERVICE.get_certificate(v)
 
         if cert is None:
-            return E_NO_CERTIFICATES_FOUND, 205  # TODO related to 204 issue
+            return E_NO_CERTIFICATES_FOUND, C_NO_DATA                               
 
         while cert.parent_id != cert.certificate_id:
             cert = CERTIFICATE_SERVICE.get_certificate(cert.parent_id)
             if cert is None:
-                return E_CORRUPTED_DATABASE, 500
+                return E_CORRUPTED_DATABASE, C_INTERNAL_SERVER_ERROR
 
-        return {"success": True, "data": cert.pem_data}
+        return {"success": True, "data": cert.pem_data}, C_SUCCESS
 
     @staticmethod
-    def get_certificate_trust_chain_by_id(id):  # noqa: E501
+    def get_certificate_trust_chain_by_id(id):
         """get certificate's trust chain by ID
 
-        Get certificate trust chain in PEM format by ID # noqa: E501
+        Get certificate trust chain in PEM format by ID
 
         :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
+        CertController.setup()                                                      # TODO remove after issue fixed
 
         try:
             v = int(id)
         except ValueError:
-            return E_WRONG_PARAMETERS, 400
+            return E_WRONG_PARAMETERS, C_BAD_REQUEST
 
         cert = CERTIFICATE_SERVICE.get_certificate(v)
 
         if cert is None:
-            return E_NO_CERTIFICATES_FOUND, 205  # TODO related to 204 issue
+            return E_NO_CERTIFICATES_FOUND, C_NO_DATA                               
 
         ret = []
 
         while True:
             cert = CERTIFICATE_SERVICE.get_certificate(cert.parent_id)
             if cert is None:
-                return E_CORRUPTED_DATABASE, 500
+                return E_CORRUPTED_DATABASE, C_INTERNAL_SERVER_ERROR
             elif cert.parent_id == cert.certificate_id:
                 break
             ret.append(cert.pem_data)
 
-        return {"success": True, "data": "".join(ret)}
+        return {"success": True, "data": "".join(ret)}, C_SUCCESS
 
     @staticmethod
     def cert_to_dict_partial(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)
         if c_issuer is None:
             return None
@@ -335 +345 @@
 
     @staticmethod
     def cert_to_dict_full(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)
         if c_issuer is None: