2
0
mirror of https://github.com/openkmip/pykmip synced 2025-12-24 04:04:20 +00:00

Add UniqueIdentifier 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 Unique
Identifier. 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 Unique Identifier filtering. Simply use the
"--unique-identifier" flag to specify a Unique Identifier string
value for the Locate script to filter on.
This commit is contained in:
Peter Hamilton
2019-07-31 15:37:25 -04:00
committed by Peter Hamilton
parent 8441bb4302
commit 9e95d14e08
7 changed files with 218 additions and 0 deletions

View File

@@ -5065,6 +5065,119 @@ class TestKmipEngine(testtools.TestCase):
)
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.