Add UniqueIdentifier filtering support for the Locate operation

kmip/demos/pie/locate.py

@@ -38,6 +38,7 @@
     object_type = opts.object_type
     cryptographic_algorithm = opts.cryptographic_algorithm
     cryptographic_length = opts.cryptographic_length
+    unique_identifier = opts.unique_identifier
 
     attribute_factory = AttributeFactory()
 
@@ -136,6 +137,13 @@
                 )
             )
             sys.exit(-6)
+    if unique_identifier:
+        attributes.append(
+            attribute_factory.create_attribute(
+                enums.AttributeType.UNIQUE_IDENTIFIER,
+                unique_identifier
+            )
+        )
 
     # Build the client and connect to the server
     with client.ProxyKmipClient(

kmip/demos/units/locate.py

@@ -41,6 +41,7 @@
     object_type = opts.object_type
     cryptographic_algorithm = opts.cryptographic_algorithm
     cryptographic_length = opts.cryptographic_length
+    unique_identifier = opts.unique_identifier
 
     attribute_factory = AttributeFactory()
     credential_factory = CredentialFactory()
@@ -163,6 +164,13 @@
             )
             client.close()
             sys.exit(-6)
+    if unique_identifier:
+        attributes.append(
+            attribute_factory.create_attribute(
+                enums.AttributeType.UNIQUE_IDENTIFIER,
+                unique_identifier
+            )
+        )
 
     result = client.locate(attributes=attributes, credential=credential)
     client.close()

kmip/demos/utils.py

@@ -287,6 +287,15 @@ def build_cli_parser(operation=None):
             dest="cryptographic_length",
             help="The cryptographic length of the secret (e.g., 128, 2048)"
         )
+        parser.add_option(
+            "-i",
+            "--unique-identifier",
+            action="store",
+            type="str",
+            default=None,
+            dest="unique_identifier",
+            help="The unique identifier of the secret (e.g., 1, 2, 3)"
+        )
     elif operation is Operation.REGISTER:
         parser.add_option(
             "-f",

kmip/services/server/engine.py

@@ -1715,6 +1715,20 @@ def _process_locate(self, payload):
                             )
                             add_object = False
                             break
+                    elif name == "Unique Identifier":
+                        value = value.value
+                        if value != attribute:
+                            self._logger.debug(
+                                "Failed match: "
+                                "the specified unique identifier ({}) "
+                                "does not match the object's unique "
+                                "identifier ({}).".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

@@ -1421,6 +1421,39 @@ def test_symmetric_key_create_getattributes_locate_destroy(self):
         )
         self.assertEqual(0, len(result.uuids))
 
+        # Test locating each key by its unique identifier.
+        result = self.client.locate(
+            attributes=[
+                self.attr_factory.create_attribute(
+                    enums.AttributeType.UNIQUE_IDENTIFIER,
+                    uid_a
+                )
+            ]
+        )
+        self.assertEqual(1, len(result.uuids))
+        self.assertIn(uid_a, result.uuids)
+
+        result = self.client.locate(
+            attributes=[
+                self.attr_factory.create_attribute(
+                    enums.AttributeType.UNIQUE_IDENTIFIER,
+                    uid_b
+                )
+            ]
+        )
+        self.assertEqual(1, len(result.uuids))
+        self.assertIn(uid_b, result.uuids)
+
+        result = self.client.locate(
+            attributes=[
+                self.attr_factory.create_attribute(
+                    enums.AttributeType.UNIQUE_IDENTIFIER,
+                    "unknown"
+                )
+            ]
+        )
+        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

@@ -1076,6 +1076,39 @@ def test_create_getattributes_locate_destroy(self):
         )
         self.assertEqual(0, len(result))
 
+        # Test locating each key by its unique identifier.
+        result = self.client.locate(
+            attributes=[
+                self.attribute_factory.create_attribute(
+                    enums.AttributeType.UNIQUE_IDENTIFIER,
+                    a_id
+                )
+            ]
+        )
+        self.assertEqual(1, len(result))
+        self.assertIn(a_id, result)
+
+        result = self.client.locate(
+            attributes=[
+                self.attribute_factory.create_attribute(
+                    enums.AttributeType.UNIQUE_IDENTIFIER,
+                    b_id
+                )
+            ]
+        )
+        self.assertEqual(1, len(result))
+        self.assertIn(b_id, result)
+
+        result = self.client.locate(
+            attributes=[
+                self.attribute_factory.create_attribute(
+                    enums.AttributeType.UNIQUE_IDENTIFIER,
+                    "unknown"
+                )
+            ]
+        )
+        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

@@ -5065,6 +5065,119 @@ def test_locate_with_cryptographic_length(self):
         )
         self.assertEqual(0, len(response_payload.unique_identifiers))
 
+    def test_locate_with_unique_identifier(self):
+        """
+        Test the Locate operation when the 'Unique Identifier' 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)
+        id_b = str(obj_b.unique_identifier)
+
+        attribute_factory = factory.AttributeFactory()
+
+        # Locate the symmetric key object based on its unique identifier.
+        attrs = [
+            attribute_factory.create_attribute(
+                enums.AttributeType.UNIQUE_IDENTIFIER,
+                id_a
+            )
+        ]
+        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 unique identifier ({}) does not match "
+            "the object's unique identifier ({}).".format(id_a, id_b)
+        )
+        self.assertEqual(1, len(response_payload.unique_identifiers))
+        self.assertIn(id_a, response_payload.unique_identifiers)
+
+        attrs = [
+            attribute_factory.create_attribute(
+                enums.AttributeType.UNIQUE_IDENTIFIER,
+                id_b
+            )
+        ]
+        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 unique identifier ({}) does not match "
+            "the object's unique identifier ({}).".format(id_b, id_a)
+        )
+        e._logger.debug.assert_any_call(
+            "Locate filter matched object: {}".format(id_b)
+        )
+        self.assertEqual(1, len(response_payload.unique_identifiers))
+        self.assertIn(id_b, response_payload.unique_identifiers)
+
+        # Try to locate a non-existent object based on its cryptographic
+        # algorithm.
+        attrs = [
+            attribute_factory.create_attribute(
+                enums.AttributeType.UNIQUE_IDENTIFIER,
+                "unknown"
+            )
+        ]
+        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 unique identifier ({}) does not match "
+            "the object's unique identifier ({}).".format("unknown", id_a)
+        )
+        e._logger.debug.assert_any_call(
+            "Failed match: "
+            "the specified unique identifier ({}) does not match "
+            "the object's unique identifier ({}).".format("unknown", id_b)
+        )
+        self.assertEqual(0, len(response_payload.unique_identifiers))
+
     def test_get(self):
         """
         Test that a Get request can be processed correctly.