Revize da92e92a

[Merge conflict] - reverted controller version from #8705

src/controllers/certificates_controller.py

@@ -3 +3 @@
 from itertools import chain
 from json import JSONDecodeError
 
-from flask import request, Response
+from flask import request
 from injector import inject
 
 from src.constants import CA_ID, \
@@ -22 +22 @@
 from src.utils.logger import Logger
 from src.utils.util import dict_to_string
 
-EXTENSIONS = "extensions"
-
 TREE_NODE_TYPE_COUNT = 3
 KEY_PEM = "key_pem"
 PASSWORD = "password"
 KEY = "key"
 FILTERING = "filtering"
-PAGE = "page"
-PER_PAGE = "per_page"
 ISSUER = "issuer"
 US = "usage"
 NOT_AFTER = "notAfter"
 NOT_BEFORE = "notBefore"
 COMMON_NAME = "CN"
 ID = "id"
-CA = "CA"
 USAGE = "usage"
 SUBJECT = "subject"
 VALIDITY_DAYS = "validityDays"
-TYPE = "type"
+CA = "CA"
 ISSUED_BY = "issuedby"
 STATUS = "status"
 REASON = "reason"
 REASON_UNDEFINED = "unspecified"
-NAME = "name"
-PASSWORD = "password"
 
 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."}
@@ -60 +53 @@
 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."}
-E_IDENTITY_NAME_NOT_SPECIFIED = {"success": False, "data": "Invalid request, missing identity name."}
-E_IDENTITY_PASSWORD_NOT_SPECIFIED = {"success": False, "data": "Invalid request, missing identity password."}
 
 
 class CertController:
-    USAGE_KEY_MAP = {'CA': CA_ID, 'SSL': SSL_ID, 'digitalSignature': SIGNATURE_ID, 'authentication': AUTHENTICATION_ID}
-    INVERSE_USAGE_KEY_MAP = {k: v for v, k in USAGE_KEY_MAP.items()}
-    FILTERING_TYPE_KEY_MAP = {'root': ROOT_CA_ID, 'inter': INTERMEDIATE_CA_ID, 'end': CERTIFICATE_ID}
-    # INVERSE_FILTERING_TYPE_KEY_MAP = {k: v for v, k in FILTERING_TYPE_KEY_MAP.items()}
-
+    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()}
 
     @inject
     def __init__(self, certificate_service: CertificateService, key_service: KeyService):
@@ -280 +268 @@
         Logger.info(f"\n\t{request.referrer}"
                     f"\n\t{request.method}   {request.path}   {request.scheme}")
 
-
-        # the filtering parameter can be read as URL argument or as a request body
-        if request.is_json:
-            data = request.get_json()
-        else:
-            data = {}
-
-        if FILTERING in request.args.keys():
-            try:
-                data[FILTERING] = json.loads(request.args[FILTERING])
-            except JSONDecodeError:
-                Logger.error(f"The request must be JSON-formatted.")
-                return E_NOT_JSON_FORMAT, C_BAD_REQUEST
-
-        if PAGE in request.args.keys():
-            try:
-                data[PAGE] = json.loads(request.args[PAGE])
-            except JSONDecodeError:
-                Logger.error(f"The request must be JSON-formatted.")
-                return E_NOT_JSON_FORMAT, C_BAD_REQUEST
-
-        if PER_PAGE in request.args.keys():
-            try:
-                data[PER_PAGE] = json.loads(request.args[PER_PAGE])
-            except JSONDecodeError:
-                Logger.error(f"The request must be JSON-formatted.")
-                return E_NOT_JSON_FORMAT, C_BAD_REQUEST
-
-        Logger.info(f"\n\tRequest body:"
-                    f"\n{dict_to_string(data)}")
-
-        target_types = {ROOT_CA_ID, INTERMEDIATE_CA_ID, CERTIFICATE_ID}
-        target_usages = {v for v in CertController.INVERSE_USAGE_KEY_MAP.keys()}
-        target_cn_substring = None
+        targets = {ROOT_CA_ID, INTERMEDIATE_CA_ID, CERTIFICATE_ID}                  # all targets
         issuer_id = -1
 
-        unfiltered = True
-
-        if PER_PAGE in data:
-            unfiltered = False
-            page = data.get(PAGE, 0)
-            per_page = data[PER_PAGE]
-        else:
-            page = None
-            per_page = None
-
-        if FILTERING in data:                                                   # if the 'filtering' field exists
-            unfiltered = False
-            if isinstance(data[FILTERING], dict):                               # and it is also a 'dict'
-
-                # noinspection DuplicatedCode
-                if TYPE in data[FILTERING]:                                     # containing 'type'
-                    if isinstance(data[FILTERING][TYPE], list):                 # which is a 'list',
-                                                                                # map every field to id
-                        try:
-                            target_types = {CertController.FILTERING_TYPE_KEY_MAP[v] for v in data[FILTERING][TYPE]}
-                        except KeyError as e:
-                            Logger.error(f"Invalid request, wrong parameters '{FILTERING}.{TYPE}' - '{e}'.")
-                            return E_WRONG_PARAMETERS, C_BAD_REQUEST
-                    else:
-                        Logger.error(f"Invalid request, wrong parameters '{FILTERING}.{TYPE}'.")
-                        return E_WRONG_PARAMETERS, C_BAD_REQUEST
-
-                # noinspection DuplicatedCode
-                if USAGE in data[FILTERING]:                                    # containing 'usage'
-                    if isinstance(data[FILTERING][USAGE], list):                # which is a 'list',
-                                                                                # map every field to id
-                        try:
-                            target_usages = {CertController.USAGE_KEY_MAP[v] for v in data[FILTERING][USAGE]}
-                        except KeyError as e:
-                            Logger.error(f"Invalid request, wrong parameters '{FILTERING}.{USAGE}' - '{e}'.")
-                            return E_WRONG_PARAMETERS, C_BAD_REQUEST
-                    else:
-                        Logger.error(f"Invalid request, wrong parameters '{FILTERING}.{USAGE}'.")
-                        return E_WRONG_PARAMETERS, C_BAD_REQUEST
-
-                if COMMON_NAME in data[FILTERING]:                              # containing 'CN'
-                    if isinstance(data[FILTERING][COMMON_NAME], str):           # which is a 'str'
-                        target_cn_substring = data[FILTERING][COMMON_NAME]
-                    else:
-                        Logger.error(f"Invalid request, wrong parameters '{FILTERING}.{COMMON_NAME}'.")
-                        return E_WRONG_PARAMETERS, C_BAD_REQUEST
-
-                if ISSUED_BY in data[FILTERING]:                                # containing 'issuedby'
-                    if isinstance(data[FILTERING][ISSUED_BY], int):             # which is an 'int'
-                        issuer_id = data[FILTERING][ISSUED_BY]                  # then get its children only
-
+        # the filtering parameter can be read as URL argument or as a request body
+        if request.is_json or FILTERING in request.args.keys():                     # if the request carries JSON data
+            if request.is_json:
+                data = request.get_json()                                           # get it
             else:
-                Logger.error(f"Invalid request, wrong parameters '{FILTERING}'.")
-                return E_WRONG_PARAMETERS, C_BAD_REQUEST
+                try:
+                    data = {FILTERING: json.loads(request.args[FILTERING])}
+                except JSONDecodeError:
+                    Logger.error(f"The request must be JSON-formatted.")
+                    return E_NOT_JSON_FORMAT, C_BAD_REQUEST
 
-        if unfiltered:                                                      # if not filtering
-            certs = self.certificate_service.get_certificates()
-        elif issuer_id >= 0:                                                # if filtering by an issuer
-            try:
-                                                                            # get his children, filtered
-                certs = self.certificate_service.get_certificates_issued_by_filter(
-                    issuer_id=issuer_id,
-                    target_types=target_types,
-                    target_usages=target_usages,
-                    target_cn_substring=target_cn_substring,
-                    page=page,
-                    per_page=per_page
-                )
-            except CertificateNotFoundException:                            # if id does not exist
-                Logger.error(f"No such certificate found 'ID = {issuer_id}'.")
-                return E_NO_CERTIFICATES_FOUND, C_NOT_FOUND                 # throw
+            Logger.info(f"\n\tRequest body:"
+                        f"\n{dict_to_string(data)}")
+
+            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:
+                            Logger.error(f"Invalid request, wrong parameters '{FILTERING}.{CA}'.")
+                            return E_WRONG_PARAMETERS, C_BAD_REQUEST
+                    if ISSUED_BY in data[FILTERING]:                                # containing 'issuedby'
+                        if isinstance(data[FILTERING][ISSUED_BY], int):             # which is an 'int'
+                            issuer_id = data[FILTERING][ISSUED_BY]                  # then get its children only
+                else:
+                    Logger.error(f"Invalid request, wrong parameters '{FILTERING}'.")
+                    return E_WRONG_PARAMETERS, C_BAD_REQUEST
+            if issuer_id >= 0:                                                      # if filtering by an issuer
+                try:
+                    children = self.certificate_service.get_certificates_issued_by(issuer_id)  # get his children
+                except CertificateNotFoundException:                                # if id does not exist
+                    Logger.error(f"No such certificate found 'ID = {issuer_id}'.")
+                    return E_NO_CERTIFICATES_FOUND, C_NOT_FOUND                     # throw
+
+                certs = [child for child in children if child.type_id in targets]
+
+            elif len(targets) == TREE_NODE_TYPE_COUNT:                              # = 3 -> root node,
+                                                                                    # intermediate node,
+                                                                                    # or leaf node
+
+                                                                                    # if filtering did not change the
+                                                                                    # targets,
+                certs = self.certificate_service.get_certificates()                 # fetch everything
+            else:                                                                   # otherwise fetch targets only
+                certs = list(chain(*(self.certificate_service.get_certificates(target) for target in targets)))
         else:
-            certs = self.certificate_service.get_certificates_filter(
-                target_types=target_types,
-                target_usages=target_usages,
-                target_cn_substring=target_cn_substring,
-                page=page,
-                per_page=per_page
-            )
+            certs = self.certificate_service.get_certificates()                     # if no params, fetch everything
 
         if certs is None:
             Logger.error(f"Internal server error (unknown origin).")
@@ -402 +330 @@
         elif len(certs) == 0:
             # TODO check log level
             Logger.warning(f"No such certificate found (empty list).")
-            return {"success": True, "data": []}, C_SUCCESS
+            return E_NO_CERTIFICATES_FOUND, C_NO_DATA
         else:
             ret = []
             for c in certs:
@@ -569 +497 @@
             COMMON_NAME: c.common_name,
             NOT_BEFORE: datetime.strptime(c.valid_from, DATETIME_FORMAT).date(),
             NOT_AFTER: datetime.strptime(c.valid_to, DATETIME_FORMAT).date(),
-            USAGE: {CertController.INVERSE_USAGE_KEY_MAP[k]: v for k, v in c.usages.items()},
+            USAGE: {CertController.INVERSE_KEY_MAP[k]: v for k, v in c.usages.items()},
             ISSUER: {
                 ID: c_issuer.certificate_id,
                 COMMON_NAME: c_issuer.common_name
@@ -595 +523 @@
             SUBJECT: subj.to_dict(),
             NOT_BEFORE: datetime.strptime(c.valid_from, DATETIME_FORMAT).date(),
             NOT_AFTER: datetime.strptime(c.valid_to, DATETIME_FORMAT).date(),
-            USAGE: {CertController.INVERSE_USAGE_KEY_MAP[k]: v for k, v in c.usages.items()},
+            USAGE: {CertController.INVERSE_KEY_MAP[k]: v for k, v in c.usages.items()},
             CA: c_issuer.certificate_id
         }
 
@@ -694 +622 @@
             Logger.error(f"Internal server error (unknown origin).")
             return E_GENERAL_ERROR, C_INTERNAL_SERVER_ERROR
 
-        return {"success": True, "data": "The certificate and its descendants have been successfully deleted."}
-
-    def generate_certificate_pkcs_identity(self, id):
-        """
-        Generates a PKCS12 identity (including the chain of trust) of the certificate given by the specified ID.
-        Response is of application/x-pkcs12 type.
-
-        :param id: ID of a certificate whose PKCS12 identity should be generated
-        :type id: int
-
-        :rtype: Response
-        """
-
-        Logger.info(f"\n\t{request.referrer}"
-                    f"\n\t{request.method}   {request.path}   {request.scheme}"
-                    f"\n\tCertificate ID = {id}")
-
-        # try to parse the supplied ID
-        try:
-            v = int(id)
-        except ValueError:
-            Logger.error(f"Invalid request, wrong parameters 'id'[{id}] (expected integer).")
-            return E_WRONG_PARAMETERS, C_BAD_REQUEST
-
-        # find a certificate using the given ID
-        cert = self.certificate_service.get_certificate(v)
-
-        if request.is_json:                                                         # accept JSON only
-            body = request.get_json()
-
-            # check whether the request is well formed meaning that it contains all required fields
-            if NAME not in body.keys():
-                return E_IDENTITY_NAME_NOT_SPECIFIED, C_BAD_REQUEST
-
-            if PASSWORD not in body.keys():
-                return E_IDENTITY_PASSWORD_NOT_SPECIFIED, C_BAD_REQUEST
-
-            # parse required fields from the request
-            identity_name = body[NAME]
-            identity_password = body[PASSWORD]
-
-            # check whether a certificated specified by the given ID exists
-            if cert is None:
-                Logger.error(f"No such certificate found 'ID = {v}'.")
-                return E_NO_CERTIFICATES_FOUND, C_NOT_FOUND
-            else:
-                # try to load it's private key
-                key = self.key_service.get_key(cert.private_key_id)
-                if key is None:
-                    Logger.error(
-                        f"The private key 'ID = {cert.private_key_id}'of the certificate 'ID = {cert.certificate_id}' does not exist.")
-                    return E_NO_CERTIFICATES_FOUND, C_INTERNAL_SERVER_ERROR
-                else:
-                    # generate PKCS12 identity
-                    identity_byte_array = self.certificate_service.generate_pkcs_identity(cert, key,
-                                                                                          identity_name,
-                                                                                          identity_password)
-                    return Response(identity_byte_array, mimetype='application/x-pkcs12')
+        return {"success": True, "data": "The certificate and its descendants have been successfully deleted."}