From 6dcae13c5b78296e7978d2e4ced2fdcf04b04ff8 Mon Sep 17 00:00:00 2001 From: Peter Hamilton Date: Wed, 31 Jul 2019 12:07:28 -0400 Subject: [PATCH] Add CryptographicLength filtering support for the Locate operation This change updates Locate operation support in the PyKMIP server, allowing users to filter objects based on the object's Cryptographic Length. If an object's type does not support the Cryptographic Length attribute, the object is not a match. 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 Cryptographic Length filtering. Simply use the "--cryptographic-length" flag to specify a Cryptographic Length integer value for the Locate script to filter on. --- kmip/demos/pie/locate.py | 16 ++++ kmip/demos/units/locate.py | 17 ++++ kmip/demos/utils.py | 8 ++ kmip/services/server/engine.py | 14 +++ .../integration/services/test_integration.py | 23 +++++ .../services/test_proxykmipclient.py | 33 +++++++ .../tests/unit/services/server/test_engine.py | 88 +++++++++++++++++++ 7 files changed, 199 insertions(+) diff --git a/kmip/demos/pie/locate.py b/kmip/demos/pie/locate.py index 691804c..2ef5682 100644 --- a/kmip/demos/pie/locate.py +++ b/kmip/demos/pie/locate.py @@ -37,6 +37,7 @@ if __name__ == '__main__': 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 @@ if __name__ == '__main__': ) ) 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( diff --git a/kmip/demos/units/locate.py b/kmip/demos/units/locate.py index 5ee2f22..f8eeb71 100644 --- a/kmip/demos/units/locate.py +++ b/kmip/demos/units/locate.py @@ -40,6 +40,7 @@ if __name__ == '__main__': 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 @@ if __name__ == '__main__': ) 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() diff --git a/kmip/demos/utils.py b/kmip/demos/utils.py index 2b6c48e..8b7593a 100644 --- a/kmip/demos/utils.py +++ b/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", diff --git a/kmip/services/server/engine.py b/kmip/services/server/engine.py index 0b1a32c..e838c46 100644 --- a/kmip/services/server/engine.py +++ b/kmip/services/server/engine.py @@ -1701,6 +1701,20 @@ class KmipEngine(object): ) 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( diff --git a/kmip/tests/integration/services/test_integration.py b/kmip/tests/integration/services/test_integration.py index c650339..e5f4d27 100644 --- a/kmip/tests/integration/services/test_integration.py +++ b/kmip/tests/integration/services/test_integration.py @@ -1398,6 +1398,29 @@ class TestIntegration(testtools.TestCase): 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) diff --git a/kmip/tests/integration/services/test_proxykmipclient.py b/kmip/tests/integration/services/test_proxykmipclient.py index 9449f0d..aa6904e 100644 --- a/kmip/tests/integration/services/test_proxykmipclient.py +++ b/kmip/tests/integration/services/test_proxykmipclient.py @@ -1043,6 +1043,39 @@ class TestProxyKmipClientIntegration(testtools.TestCase): ) 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) diff --git a/kmip/tests/unit/services/server/test_engine.py b/kmip/tests/unit/services/server/test_engine.py index 0667e23..6c392e1 100644 --- a/kmip/tests/unit/services/server/test_engine.py +++ b/kmip/tests/unit/services/server/test_engine.py @@ -4977,6 +4977,94 @@ class TestKmipEngine(testtools.TestCase): ) 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.