Add CryptographicLength filtering support for the Locate operation

kmip/demos/pie/locate.py

@@ -37,6 +37,7 @@
     state = opts.state
     object_type = opts.object_type
     cryptographic_algorithm = opts.cryptographic_algorithm
+    cryptographic_length = opts.cryptographic_length
 
     attribute_factory = AttributeFactory()
 
@@ -120,6 +121,21 @@
                 )
             )
             sys.exit(-5)
+    if cryptographic_length:
+        if cryptographic_length > 0:
+            attributes.append(
+                attribute_factory.create_attribute(
+                    enums.AttributeType.CRYPTOGRAPHIC_LENGTH,
+                    cryptographic_length
+                )
+            )
+        else:
+            logger.error(
+                "Invalid cryptographic length provided: {}".format(
+                    opts.cryptographic_length
+                )
+            )
+            sys.exit(-6)
 
     # Build the client and connect to the server
     with client.ProxyKmipClient(

kmip/demos/units/locate.py

@@ -40,6 +40,7 @@
     state = opts.state
     object_type = opts.object_type
     cryptographic_algorithm = opts.cryptographic_algorithm
+    cryptographic_length = opts.cryptographic_length
 
     attribute_factory = AttributeFactory()
     credential_factory = CredentialFactory()
@@ -146,6 +147,22 @@
             )
             client.close()
             sys.exit(-5)
+    if cryptographic_length:
+        if cryptographic_length > 0:
+            attributes.append(
+                attribute_factory.create_attribute(
+                    enums.AttributeType.CRYPTOGRAPHIC_LENGTH,
+                    cryptographic_length
+                )
+            )
+        else:
+            logger.error(
+                "Invalid cryptographic length provided: {}".format(
+                    opts.cryptographic_length
+                )
+            )
+            client.close()
+            sys.exit(-6)
 
     result = client.locate(attributes=attributes, credential=credential)
     client.close()

kmip/demos/utils.py

@@ -279,6 +279,14 @@ def build_cli_parser(operation=None):
             dest="cryptographic_algorithm",
             help="The cryptographic algorithm of the secret (e.g., AES, RSA)"
         )
+        parser.add_option(
+            "--cryptographic-length",
+            action="store",
+            type="int",
+            default=None,
+            dest="cryptographic_length",
+            help="The cryptographic length of the secret (e.g., 128, 2048)"
+        )
     elif operation is Operation.REGISTER:
         parser.add_option(
             "-f",

kmip/services/server/engine.py

@@ -1701,6 +1701,20 @@ def _process_locate(self, payload):
                             )
                             add_object = False
                             break
+                    elif name == "Cryptographic Length":
+                        value = value.value
+                        if value != attribute:
+                            self._logger.debug(
+                                "Failed match: "
+                                "the specified cryptographic length ({}) "
+                                "does not match the object's cryptographic "
+                                "length ({}).".format(
+                                    value,
+                                    attribute
+                                )
+                            )
+                            add_object = False
+                            break
                     elif name == enums.AttributeType.INITIAL_DATE.value:
                         initial_date["value"] = attribute
                         self._track_date_attributes(

kmip/tests/integration/services/test_integration.py

@@ -1398,6 +1398,29 @@ def test_symmetric_key_create_getattributes_locate_destroy(self):
         self.assertEqual(ResultStatus.SUCCESS, result.result_status.value)
         self.assertEqual(0, len(result.uuids))
 
+        # Test locating each key by its cryptographic length.
+        result = self.client.locate(
+            attributes=[
+                self.attr_factory.create_attribute(
+                    enums.AttributeType.CRYPTOGRAPHIC_LENGTH,
+                    128
+                )
+            ]
+        )
+        self.assertEqual(2, len(result.uuids))
+        self.assertIn(uid_a, result.uuids)
+        self.assertIn(uid_b, result.uuids)
+
+        result = self.client.locate(
+            attributes=[
+                self.attr_factory.create_attribute(
+                    enums.AttributeType.CRYPTOGRAPHIC_LENGTH,
+                    2048
+                )
+            ]
+        )
+        self.assertEqual(0, len(result.uuids))
+
         # Clean up keys
         result = self.client.destroy(uid_a)
         self.assertEqual(ResultStatus.SUCCESS, result.result_status.value)

kmip/tests/integration/services/test_proxykmipclient.py

@@ -1043,6 +1043,39 @@ def test_create_getattributes_locate_destroy(self):
         )
         self.assertEqual(0, len(result))
 
+        # Test locating each key by its cryptographic length.
+        result = self.client.locate(
+            attributes=[
+                self.attribute_factory.create_attribute(
+                    enums.AttributeType.CRYPTOGRAPHIC_LENGTH,
+                    128
+                )
+            ]
+        )
+        self.assertEqual(1, len(result))
+        self.assertIn(b_id, result)
+
+        result = self.client.locate(
+            attributes=[
+                self.attribute_factory.create_attribute(
+                    enums.AttributeType.CRYPTOGRAPHIC_LENGTH,
+                    256
+                )
+            ]
+        )
+        self.assertEqual(1, len(result))
+        self.assertIn(a_id, result)
+
+        result = self.client.locate(
+            attributes=[
+                self.attribute_factory.create_attribute(
+                    enums.AttributeType.CRYPTOGRAPHIC_LENGTH,
+                    2048
+                )
+            ]
+        )
+        self.assertEqual(0, len(result))
+
         # Clean up the keys
         self.client.destroy(a_id)
         self.client.destroy(b_id)

kmip/tests/unit/services/server/test_engine.py

@@ -4977,6 +4977,94 @@ def test_locate_with_cryptographic_algorithm(self):
         )
         self.assertEqual(0, len(response_payload.unique_identifiers))
 
+    def test_locate_with_cryptographic_length(self):
+        """
+        Test the Locate operation when the 'Cryptographic Length' attribute
+        is given.
+        """
+        e = engine.KmipEngine()
+        e._data_store = self.engine
+        e._data_store_session_factory = self.session_factory
+        e._data_session = e._data_store_session_factory()
+        e._is_allowed_by_operation_policy = mock.Mock(return_value=True)
+        e._logger = mock.MagicMock()
+
+        key = (
+            b'\x01\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'
+        )
+
+        obj_a = pie_objects.SymmetricKey(
+            enums.CryptographicAlgorithm.AES,
+            128,
+            key,
+            name='name1'
+        )
+        obj_b = pie_objects.SecretData(
+            key,
+            enums.SecretDataType.PASSWORD
+        )
+
+        e._data_session.add(obj_a)
+        e._data_session.add(obj_b)
+        e._data_session.commit()
+        e._data_session = e._data_store_session_factory()
+
+        id_a = str(obj_a.unique_identifier)
+
+        attribute_factory = factory.AttributeFactory()
+
+        # Locate the symmetric key object based on its cryptographic length.
+        attrs = [
+            attribute_factory.create_attribute(
+                enums.AttributeType.CRYPTOGRAPHIC_LENGTH,
+                128
+            )
+        ]
+        payload = payloads.LocateRequestPayload(attributes=attrs)
+        e._logger.reset_mock()
+        response_payload = e._process_locate(payload)
+        e._data_session.commit()
+        e._data_session = e._data_store_session_factory()
+
+        e._logger.info.assert_any_call("Processing operation: Locate")
+        e._logger.debug.assert_any_call(
+            "Locate filter matched object: {}".format(id_a)
+        )
+        e._logger.debug.assert_any_call(
+            "Failed match: "
+            "the specified attribute (Cryptographic Length) is not "
+            "applicable for the object's object type (SECRET_DATA)."
+        )
+        self.assertEqual(1, len(response_payload.unique_identifiers))
+        self.assertIn(id_a, response_payload.unique_identifiers)
+
+        # Try to locate a non-existent object based on its cryptographic
+        # algorithm.
+        attrs = [
+            attribute_factory.create_attribute(
+                enums.AttributeType.CRYPTOGRAPHIC_LENGTH,
+                2048
+            )
+        ]
+        payload = payloads.LocateRequestPayload(attributes=attrs)
+        e._logger.reset_mock()
+        response_payload = e._process_locate(payload)
+        e._data_session.commit()
+        e._data_session = e._data_store_session_factory()
+
+        e._logger.info.assert_any_call("Processing operation: Locate")
+        e._logger.debug.assert_any_call(
+            "Failed match: "
+            "the specified cryptographic length (2048) does not match "
+            "the object's cryptographic length (128)."
+        )
+        e._logger.debug.assert_any_call(
+            "Failed match: "
+            "the specified attribute (Cryptographic Length) is not "
+            "applicable for the object's object type (SECRET_DATA)."
+        )
+        self.assertEqual(0, len(response_payload.unique_identifiers))
+
     def test_get(self):
         """
         Test that a Get request can be processed correctly.