From 942d55cca91cd1d60a4c747cf1c0856596f62629 Mon Sep 17 00:00:00 2001 From: Peter Hamilton Date: Wed, 31 Jul 2019 17:01:39 -0400 Subject: [PATCH] Add OperationPolicyName filtering support for Locate This change updates Locate operation support in the PyKMIP server, allowing users to filter objects based on the object's Operation Policy Name. Unit tests and integration tests have been added to test and verify the correctness of this feature. Additionally, the Locate demo scripts have also been updated to support Operation Policy Name filtering. Simply use the "--operation-policy-name" flag to specify an Operation Policy Name string value for the Locate script to filter on. --- kmip/demos/pie/locate.py | 8 ++ kmip/demos/units/locate.py | 8 ++ kmip/demos/utils.py | 8 ++ kmip/services/server/engine.py | 14 +++ .../integration/services/test_integration.py | 23 ++++ .../services/test_proxykmipclient.py | 23 ++++ .../tests/unit/services/server/test_engine.py | 113 ++++++++++++++++++ 7 files changed, 197 insertions(+) diff --git a/kmip/demos/pie/locate.py b/kmip/demos/pie/locate.py index 5ea294a..10f3b50 100644 --- a/kmip/demos/pie/locate.py +++ b/kmip/demos/pie/locate.py @@ -39,6 +39,7 @@ if __name__ == '__main__': cryptographic_algorithm = opts.cryptographic_algorithm cryptographic_length = opts.cryptographic_length unique_identifier = opts.unique_identifier + operation_policy_name = opts.operation_policy_name attribute_factory = AttributeFactory() @@ -144,6 +145,13 @@ if __name__ == '__main__': unique_identifier ) ) + if operation_policy_name: + attributes.append( + attribute_factory.create_attribute( + enums.AttributeType.OPERATION_POLICY_NAME, + operation_policy_name + ) + ) # Build the client and connect to the server with client.ProxyKmipClient( diff --git a/kmip/demos/units/locate.py b/kmip/demos/units/locate.py index 08785ad..7f8c5ca 100644 --- a/kmip/demos/units/locate.py +++ b/kmip/demos/units/locate.py @@ -42,6 +42,7 @@ if __name__ == '__main__': cryptographic_algorithm = opts.cryptographic_algorithm cryptographic_length = opts.cryptographic_length unique_identifier = opts.unique_identifier + operation_policy_name = opts.operation_policy_name attribute_factory = AttributeFactory() credential_factory = CredentialFactory() @@ -171,6 +172,13 @@ if __name__ == '__main__': unique_identifier ) ) + if operation_policy_name: + attributes.append( + attribute_factory.create_attribute( + enums.AttributeType.OPERATION_POLICY_NAME, + operation_policy_name + ) + ) result = client.locate(attributes=attributes, credential=credential) client.close() diff --git a/kmip/demos/utils.py b/kmip/demos/utils.py index 80068ca..6cd6a8d 100644 --- a/kmip/demos/utils.py +++ b/kmip/demos/utils.py @@ -296,6 +296,14 @@ def build_cli_parser(operation=None): dest="unique_identifier", help="The unique identifier of the secret (e.g., 1, 2, 3)" ) + parser.add_option( + "--operation-policy-name", + action="store", + type="str", + default=None, + dest="operation_policy_name", + help="The operation policy name of the secret (e.g., default)" + ) elif operation is Operation.REGISTER: parser.add_option( "-f", diff --git a/kmip/services/server/engine.py b/kmip/services/server/engine.py index 068df21..98ea104 100644 --- a/kmip/services/server/engine.py +++ b/kmip/services/server/engine.py @@ -1729,6 +1729,20 @@ class KmipEngine(object): ) add_object = False break + elif name == "Operation Policy Name": + value = value.value + if value != attribute: + self._logger.debug( + "Failed match: " + "the specified operation policy name ({}) " + "does not match the object's operation policy " + "name ({}).".format( + value, + attribute + ) + ) + add_object = False + break elif name == enums.AttributeType.INITIAL_DATE.value: initial_date["value"] = attribute self._track_date_attributes( diff --git a/kmip/tests/integration/services/test_integration.py b/kmip/tests/integration/services/test_integration.py index f0686a0..631d2d3 100644 --- a/kmip/tests/integration/services/test_integration.py +++ b/kmip/tests/integration/services/test_integration.py @@ -1454,6 +1454,29 @@ class TestIntegration(testtools.TestCase): ) self.assertEqual(0, len(result.uuids)) + # Test locating each key by its operation policy name. + result = self.client.locate( + attributes=[ + self.attr_factory.create_attribute( + enums.AttributeType.OPERATION_POLICY_NAME, + "default" + ) + ] + ) + 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.OPERATION_POLICY_NAME, + "unknown" + ) + ] + ) + self.assertEqual(0, len(result.uuids)) + # Clean up keys result = self.client.destroy(uid_a) self.assertEqual(ResultStatus.SUCCESS, result.result_status.value) diff --git a/kmip/tests/integration/services/test_proxykmipclient.py b/kmip/tests/integration/services/test_proxykmipclient.py index 032236d..26e28b0 100644 --- a/kmip/tests/integration/services/test_proxykmipclient.py +++ b/kmip/tests/integration/services/test_proxykmipclient.py @@ -1109,6 +1109,29 @@ class TestProxyKmipClientIntegration(testtools.TestCase): ) self.assertEqual(0, len(result)) + # Test locating each key by its operation policy name. + result = self.client.locate( + attributes=[ + self.attribute_factory.create_attribute( + enums.AttributeType.OPERATION_POLICY_NAME, + "default" + ) + ] + ) + self.assertEqual(2, len(result)) + self.assertIn(a_id, result) + self.assertIn(b_id, result) + + result = self.client.locate( + attributes=[ + self.attribute_factory.create_attribute( + enums.AttributeType.OPERATION_POLICY_NAME, + "unknown" + ) + ] + ) + self.assertEqual(0, len(result)) + # Clean up the keys self.client.destroy(a_id) self.client.destroy(b_id) diff --git a/kmip/tests/unit/services/server/test_engine.py b/kmip/tests/unit/services/server/test_engine.py index 393c9fd..24d839d 100644 --- a/kmip/tests/unit/services/server/test_engine.py +++ b/kmip/tests/unit/services/server/test_engine.py @@ -5178,6 +5178,119 @@ class TestKmipEngine(testtools.TestCase): ) self.assertEqual(0, len(response_payload.unique_identifiers)) + def test_locate_with_operation_policy_name(self): + """ + Test the Locate operation when the 'Operation Policy Name' 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_a.operation_policy_name = "default" + obj_b = pie_objects.SecretData( + key, + enums.SecretDataType.PASSWORD + ) + obj_b.operation_policy_name = "custom" + + 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.OPERATION_POLICY_NAME, + "default" + ) + ] + 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 operation policy name (default) does not match " + "the object's operation policy name (custom)." + ) + self.assertEqual(1, len(response_payload.unique_identifiers)) + self.assertIn(id_a, response_payload.unique_identifiers) + + attrs = [ + attribute_factory.create_attribute( + enums.AttributeType.OPERATION_POLICY_NAME, + "custom" + ) + ] + 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_b) + ) + e._logger.debug.assert_any_call( + "Failed match: " + "the specified operation policy name (custom) does not match " + "the object's operation policy name (default)." + ) + self.assertEqual(1, len(response_payload.unique_identifiers)) + self.assertIn(id_b, response_payload.unique_identifiers) + + attrs = [ + attribute_factory.create_attribute( + enums.AttributeType.OPERATION_POLICY_NAME, + "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 operation policy name (unknown) does not match " + "the object's operation policy name (default)." + ) + e._logger.debug.assert_any_call( + "Failed match: " + "the specified operation policy name (unknown) does not match " + "the object's operation policy name (custom)." + ) + self.assertEqual(0, len(response_payload.unique_identifiers)) + def test_get(self): """ Test that a Get request can be processed correctly.