Revize d0471de8

Re #8702 - Reformed CertController's `get_list_of_certs` method, improved filtering logic.

src/controllers/certificates_controller.py

@@ -24 +24 @@
 TREE_NODE_TYPE_COUNT = 3
 
 FILTERING = "filtering"
+PAGE = "page"
+PER_PAGE = "per_page"
 ISSUER = "issuer"
 US = "usage"
 NOT_AFTER = "notAfter"
@@ -33 +35 @@
 USAGE = "usage"
 SUBJECT = "subject"
 VALIDITY_DAYS = "validityDays"
-CA = "CA"
+TYPE = "type"
 ISSUED_BY = "issuedby"
 STATUS = "status"
 REASON = "reason"
@@ -52 +54 @@
 
 
 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()}
+    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()}
+
 
     @inject
     def __init__(self, certificate_service: CertificateService, key_service: KeyService):
@@ -103 +108 @@
                 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
+                if k not in CertController.USAGE_KEY_MAP:                                 # check that it is a valid usage
                     Logger.error(f"Invalid request, wrong parameter '{USAGE}'[{k}].")
                     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
+                usages_dict[CertController.USAGE_KEY_MAP[k]] = v                          # otherwise translate key and set
 
             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
+            if TYPE not in body or body[TYPE] is None:                                  # if issuer omitted (legal) or none
                 cert = self.certificate_service.create_root_ca(                          # create a root CA
                     key,
                     subject,
@@ -118 +123 @@
                     days=body[VALIDITY_DAYS]
                 )
             else:
-                issuer = self.certificate_service.get_certificate(body[CA])              # get base issuer info
+                issuer = self.certificate_service.get_certificate(body[TYPE])              # 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}'.")
@@ -240 +245 @@
         Logger.info(f"\n\t{request.referrer}"
                     f"\n\t{request.method}   {request.path}   {request.scheme}")
 
-        targets = {ROOT_CA_ID, INTERMEDIATE_CA_ID, CERTIFICATE_ID}                  # all targets
-        issuer_id = -1
 
         # 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:
-                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 request.is_json:
+            data = request.get_json()
+        else:
+            data = {}
 
-            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}'.")
+        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
+        issuer_id = -1
+        if PER_PAGE in data:
+            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
+            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
-                    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
+                    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}.{TYPE}' - '{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
 
-                certs = [child for child in children if child.type_id in targets]
+                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
 
-            elif len(targets) == TREE_NODE_TYPE_COUNT:                              # = 3 -> root node,
-                                                                                    # intermediate node,
-                                                                                    # or leaf node
+                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
 
-                                                                                    # 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:
+                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:
+                                                                                # 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
+            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
+                )
         else:
-            certs = self.certificate_service.get_certificates()                     # if no params, fetch everything
+            certs = self.certificate_service.get_certificates()
 
         if certs is None:
             Logger.error(f"Internal server error (unknown origin).")
@@ -469 +529 @@
             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_KEY_MAP[k]: v for k, v in c.usages.items()},
+            USAGE: {CertController.INVERSE_USAGE_KEY_MAP[k]: v for k, v in c.usages.items()},
             ISSUER: {
                 ID: c_issuer.certificate_id,
                 COMMON_NAME: c_issuer.common_name
@@ -495 +555 @@
             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_KEY_MAP[k]: v for k, v in c.usages.items()},
-            CA: c_issuer.certificate_id
+            USAGE: {CertController.INVERSE_USAGE_KEY_MAP[k]: v for k, v in c.usages.items()},
+            TYPE: c_issuer.certificate_id
         }
 
     def get_private_key_of_a_certificate(self, id):