tyler.durden
/
PyKMIP
mirror of https://github.com/OpenKMIP/PyKMIP.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

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.
This commit is contained in:
Peter Hamilton 2019-07-31 17:01:39 -04:00 committed by Peter Hamilton
parent 9e95d14e08
commit 942d55cca9
7 changed files with 197 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
kmip/demos/pie/locate.py
View File

@ -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(

8
kmip/demos/units/locate.py
View File

@ -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()

8
kmip/demos/utils.py
View File

@ -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",

14
kmip/services/server/engine.py
View File

@ -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(

23
kmip/tests/integration/services/test_integration.py
View File

@ -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)

23
kmip/tests/integration/services/test_proxykmipclient.py
View File

@ -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)

113
kmip/tests/unit/services/server/test_engine.py
View File

@ -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.