Revize b8f5752f

Re #8705 - Adjusted CertController to handle all combinations of PEM/pass supply. Added calls to a not-yet-existing `verify_key(...)` method of PK service.

src/controllers/certificates_controller.py

@@ -17 +17 @@
     CertificateStatusInvalidException, CertificateNotFoundException, CertificateAlreadyRevokedException, \
     CertificateCannotBeSetToValid
 #  responsibility.
+from src.services.cryptography import CryptographyException
 from src.services.key_service import KeyService
 from src.utils.logger import Logger
 from src.utils.util import dict_to_string
 
 TREE_NODE_TYPE_COUNT = 3
-
+KEY_PEM = "key_pem"
+PASSWORD = "password"
+KEY = "key"
 FILTERING = "filtering"
 ISSUER = "issuer"
 US = "usage"
@@ -49 +52 @@
 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."}
+E_WRONG_PASSWORD = {"success": False, "data": "The provided passphrase does not match the provided key."}
 
 
 class CertController:
@@ -98 +102 @@
 
             usages_dict = {}
 
-            if not isinstance(body[USAGE], dict):                                   # type checking
+            if USAGE not in body or not isinstance(body[USAGE], list):              # type checking
                 Logger.error(f"Invalid request, wrong parameter '{USAGE}'.")
                 return E_WRONG_PARAMETERS, C_BAD_REQUEST
 
-            for k, v in body[USAGE].items():                                        # for each usage
-                if k not in CertController.KEY_MAP:                                 # check that it is a valid usage
-                    Logger.error(f"Invalid request, wrong parameter '{USAGE}'[{k}].")
+            for v in body[USAGE]:                                                   # for each usage
+                if v not in CertController.KEY_MAP:                                 # check that it is a valid usage
+                    Logger.error(f"Invalid request, wrong parameter '{USAGE}'[{v}].")
                     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 = self.key_service.create_new_key()                                      # TODO pass key
+                usages_dict[CertController.KEY_MAP[v]] = True                       # otherwise translate key and set
+
+            if KEY in body:
+                if isinstance(body[KEY], dict):
+                    if PASSWORD in body[KEY]:
+                        passphrase = body[KEY][PASSWORD]
+                        if KEY_PEM in body[KEY]:
+                            key_pem = body[KEY][KEY_PEM]
+                            if not self.key_service.verify_key(key_pem, passphrase=passphrase):
+                                Logger.error(f"Passphrase specified but invalid.")
+                                return E_WRONG_PASSWORD, C_BAD_REQUEST
+                            key = self.key_service.wrap_custom_key(key_pem, passphrase=passphrase)
+                        else:
+                            key = self.key_service.create_new_key(passphrase)
+                    else:
+                        if KEY_PEM in body[KEY]:
+                            key_pem = body[KEY][KEY_PEM]
+                            if not self.key_service.verify_key(key_pem, passphrase=None):
+                                Logger.error("Passphrase ommited but required.")
+                                return E_WRONG_PASSWORD, C_BAD_REQUEST
+                            key = self.key_service.wrap_custom_key(key_pem, passphrase=None)
+                        else:
+                            key = self.key_service.create_new_key()                 # if "key" exists but is empty
+                else:
+                    return E_WRONG_PARAMETERS, C_BAD_REQUEST
+            else:
+                key = self.key_service.create_new_key()                             # if "key" does not exist
+                                                                                    # TODO Honza: line 134 / 138, should
+                                                                                    #   they both be allowed? I say one
+                                                                                    #   of those branches should return
+                                                                                    #   wrong params bad request (mby).
 
             if CA not in body or body[CA] is None:                                  # if issuer omitted (legal) or none
-                cert = self.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 = self.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
                     Logger.error(f"No certificate authority with such unique ID exists 'ID = {key.private_key_id}'.")
-                    self.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 = self.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
                     Logger.error(f"Internal server error (corrupted database).")
-                    self.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 = 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
-                    key,                                                            # according to whether 'CA' is among
-                    subject,                                                        # the usages' fields
-                    issuer,
-                    issuer_key,
-                    usages=usages_dict,
-                    days=body[VALIDITY_DAYS]
-                )
+                try:
+                    # 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,
+                        days=body[VALIDITY_DAYS]
+                    )
+                except CryptographyException as e:
+                    return {"success": False, "data": e.message}, C_BAD_REQUEST
 
             if cert is not None:
                 return {"success": True,
                         "data": cert.certificate_id}, C_CREATED_SUCCESSFULLY
             else:                                                                   # if this fails, then
                 Logger.error(f"Internal error: The certificate could not have been created.")
-                self.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