openstack
/
deb-python-pykmip
Code Issues Proposed changes

Add MAC operation support for client

This commit is contained in:
Hao Shen 2017-01-31 09:06:24 -08:00
parent a2696b722f
commit 2d45f33d43
9 changed files with 423 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

59
kmip/demos/pie/mac.py Normal file
View File

@ -0,0 +1,59 @@
# Copyright (c) 2017 Pure Storage, Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import logging
import sys
import binascii
from kmip.core import enums
from kmip.demos import utils
from kmip.pie import client
if __name__ == '__main__':
logger = utils.build_console_logger(logging.INFO)
# Build and parse arguments
parser = utils.build_cli_parser(enums.Operation.MAC)
opts, args = parser.parse_args(sys.argv[1:])
config = opts.config
uid = opts.uuid
algorithm = opts.algorithm
data = (
b'\x00\x01\x02\x03\x04\x05\x06\x07\x08\x09\x0A\x0B\x0C\x0D\x0E'
b'\x0F')
# Exit early if the arguments are not specified
if uid is None:
logger.error('No UUID provided, exiting early from demo')
sys.exit()
if algorithm is None:
logger.error('No algorithm provided, exiting early from demo')
sys.exit()
algorithm = getattr(enums.CryptographicAlgorithm, algorithm, None)
# Build the client and connect to the server
with client.ProxyKmipClient(config=config) as client:
try:
uid, mac_data = client.mac(uid, algorithm, data)
logger.info("Successfully done MAC using key with ID: "
"{0}".format(uid))
logger.info("MACed data: {0}".format(
str(binascii.hexlify(mac_data))))
except Exception as e:
logger.error(e)

35
kmip/demos/utils.py
View File

@ -214,15 +214,32 @@ def build_cli_parser(operation=None):
"SECRET_DATA"))
elif operation is Operation.DISCOVER_VERSIONS:
parser.add_option(
"-v",
"--protocol-versions",
action="store",
type="str",
default=None,
dest="protocol_versions",
help=("Protocol versions supported by client. "
"ex. '1.1,1.2 1.3'"))
"-v",
"--protocol-versions",
action="store",
type="str",
default=None,
dest="protocol_versions",
help=("Protocol versions supported by client. "
"ex. '1.1,1.2 1.3'"))
elif operation is Operation.MAC:
parser.add_option(
"-i",
"--uuid",
action="store",
type="str",
default=None,
dest="uuid",
help="The unique ID of the managed object that is the key"
"to use for the MAC operation")
parser.add_option(
"-a",
"--algorithm",
action="store",
type="str",
default=None,
dest="algorithm",
help="Encryption algorithm for the secret (e.g., AES)")
return parser

14
kmip/pie/api.py
View File

@ -90,3 +90,17 @@ class KmipClient:
uid (string): The unique ID of the managed object to destroy.
"""
pass
@abc.abstractmethod
def mac(self, uid, algorithm, data):
"""
Get the message authentication code for data.
Args:
uid (string): The unique ID of the managed object that is the key
to use for the MAC operation.
algorithm (CryptographicAlgorithm): An enumeration defining the
algorithm to use to generate the MAC.
data (string): The data to be MACed.
"""
pass

55
kmip/pie/client.py
View File

@ -21,6 +21,9 @@ from kmip.core import objects as cobjects
from kmip.core.factories import attributes
from kmip.core.attributes import CryptographicParameters, \
CryptographicAlgorithm
from kmip.pie import api
from kmip.pie import exceptions
from kmip.pie import factory
@ -476,6 +479,58 @@ class ProxyKmipClient(api.KmipClient):
message = result.result_message.value
raise exceptions.KmipOperationFailure(status, reason, message)
def mac(self, uid, algorithm, data):
"""
Get the message authentication code for data.
Args:
uid (string): The unique ID of the managed object that is the key
to use for the MAC operation.
algorithm (CryptographicAlgorithm): An enumeration defining the
algorithm to use to generate the MAC.
data (string): The data to be MACed.
Returns:
string: The unique ID of the managed object that is the key
to use for the MAC operation.
string: The data MACed
Raises:
ClientConnectionNotOpen: if the client connection is unusable
KmipOperationFailure: if the operation result is a failure
TypeError: if the input arguments are invalid
"""
# Check inputs
if not isinstance(uid, six.string_types):
raise TypeError("uid must be a string")
if not isinstance(algorithm, enums.CryptographicAlgorithm):
raise TypeError(
"algorithm must be a CryptographicAlgorithm enumeration")
if not isinstance(data, six.binary_type):
raise TypeError(
"data must be bytes")
# Verify that operations can be given at this time
if not self._is_open:
raise exceptions.ClientConnectionNotOpen()
parameters_attribute = CryptographicParameters(
cryptographic_algorithm=CryptographicAlgorithm(algorithm))
# Create the symmetric key and handle the results
result = self.proxy.mac(uid, parameters_attribute, data)
status = result.result_status.value
if status == enums.ResultStatus.SUCCESS:
uid = result.uuid.value
mac_data = result.mac_data.value
return uid, mac_data
else:
reason = result.result_reason.value
message = result.result_message.value
raise exceptions.KmipOperationFailure(status, reason, message)
def _build_key_attributes(self, algorithm, length):
# Build a list of core key attributes.
algorithm_attribute = self.attribute_factory.create_attribute(

49
kmip/services/kmip_client.py
View File

@ -27,6 +27,7 @@ from kmip.services.results import QueryResult
from kmip.services.results import RegisterResult
from kmip.services.results import RekeyKeyPairResult
from kmip.services.results import RevokeResult
from kmip.services.results import MACResult
from kmip.core import attributes as attr
@ -60,6 +61,7 @@ from kmip.core.messages.payloads import query
from kmip.core.messages.payloads import rekey_key_pair
from kmip.core.messages.payloads import register
from kmip.core.messages.payloads import revoke
from kmip.core.messages.payloads import mac
from kmip.services.server.kmip_protocol import KMIPProtocol
@ -428,6 +430,14 @@ class KMIPProxy(KMIP):
results = self._process_batch_items(response)
return results[0]
def mac(self, unique_identifier=None, cryptographic_parameters=None,
data=None, credential=None):
return self._mac(
unique_identifier=unique_identifier,
cryptographic_parameters=cryptographic_parameters,
data=data,
credential=credential)
def _create(self,
object_type=None,
template_attribute=None,
@ -919,6 +929,43 @@ class KMIPProxy(KMIP):
uuids)
return result
def _mac(self,
unique_identifier=None,
cryptographic_parameters=None,
data=None,
credential=None):
operation = Operation(OperationEnum.MAC)
req_pl = mac.MACRequestPayload(
unique_identifier=attr.UniqueIdentifier(unique_identifier),
cryptographic_parameters=cryptographic_parameters,
data=objects.Data(data))
batch_item = messages.RequestBatchItem(operation=operation,
request_payload=req_pl)
message = self._build_request_message(credential, [batch_item])
self._send_message(message)
message = messages.ResponseMessage()
data = self._receive_message()
message.read(data)
batch_items = message.batch_items
batch_item = batch_items[0]
payload = batch_item.response_payload
if payload is None:
payload_unique_identifier = None
payload_mac_data = None
else:
payload_unique_identifier = payload.unique_identifier
payload_mac_data = payload.mac_data
result = MACResult(batch_item.result_status,
batch_item.result_reason,
batch_item.result_message,
payload_unique_identifier,
payload_mac_data)
return result
# TODO (peter-hamilton) Augment to handle device credentials
def _build_credential(self):
if (self.username is None) and (self.password is None):
@ -937,7 +984,7 @@ class KMIPProxy(KMIP):
return credential
def _build_request_message(self, credential, batch_items):
protocol_version = ProtocolVersion.create(1, 1)
protocol_version = ProtocolVersion.create(1, 2)
if credential is None:
credential = self._build_credential()

17
kmip/services/results.py
View File

@ -295,3 +295,20 @@ class RevokeResult(OperationResult):
super(RevokeResult, self).__init__(
result_status, result_reason, result_message)
self.unique_identifier = unique_identifier
class MACResult(OperationResult):
def __init__(self,
result_status,
result_reason=None,
result_message=None,
uuid=None,
mac_data=None):
super(MACResult, self).__init__(
result_status,
result_reason,
result_message
)
self.uuid = uuid
self.mac_data = mac_data

10
kmip/tests/unit/pie/test_api.py
View File

@ -44,6 +44,9 @@ class DummyKmipClient(api.KmipClient):
def destroy(self, uid):
super(DummyKmipClient, self).destroy(uid)
def mac(self, uid, algorithm, data):
super(DummyKmipClient, self).mac(uid, algorithm, data)
class TestKmipClient(testtools.TestCase):
"""
@ -106,3 +109,10 @@ class TestKmipClient(testtools.TestCase):
"""
dummy = DummyKmipClient()
dummy.destroy('uid')
def test_mac(self):
"""
Test that the mac method can be called without error.
"""
dummy = DummyKmipClient()
dummy.mac('uid', 'algorithm', 'data')

102
kmip/tests/unit/pie/test_client.py
View File

@ -1056,3 +1056,105 @@ class TestProxyKmipClient(testtools.TestCase):
self.assertIsInstance(opn.attribute_value, attr.OperationPolicyName)
self.assertEqual(opn.attribute_name.value, 'Operation Policy Name')
self.assertEqual(opn.attribute_value.value, 'test')
@mock.patch('kmip.pie.client.KMIPProxy',
mock.MagicMock(spec_set=KMIPProxy))
def test_mac(self):
"""
Test the MAC client with proper input.
"""
uuid = 'aaaaaaaa-1111-2222-3333-ffffffffffff'
algorithm = enums.CryptographicAlgorithm.HMAC_SHA256
data = (b'\x00\x01\x02\x03\x04')
result = results.MACResult(
contents.ResultStatus(enums.ResultStatus.SUCCESS),
uuid=attr.UniqueIdentifier(uuid),
mac_data=obj.MACData(data))
with ProxyKmipClient() as client:
client.proxy.mac.return_value = result
uid, mac_data = client.mac(uuid, algorithm, data)
self.assertEqual(uid, uuid)
self.assertEqual(mac_data, data)
@mock.patch('kmip.pie.client.KMIPProxy',
mock.MagicMock(spec_set=KMIPProxy))
def test_mac_on_invalid_inputs(self):
"""
Test that a TypeError exception is raised when wrong type
of arguments are given to mac operation.
"""
uuid = 'aaaaaaaa-1111-2222-3333-ffffffffffff'
uuid_invalid = int(123)
algorithm = enums.CryptographicAlgorithm.HMAC_SHA256
algorithm_invalid = enums.CryptographicUsageMask.MAC_GENERATE
data = (b'\x00\x01\x02\x03\x04')
data_invalid = int(123)
result = results.MACResult(
contents.ResultStatus(enums.ResultStatus.SUCCESS),
uuid=attr.UniqueIdentifier(uuid),
mac_data=obj.MACData(data))
args = [uuid_invalid, algorithm, data]
with ProxyKmipClient() as client:
client.proxy.mac.return_value = result
self.assertRaises(TypeError, client.mac, *args)
args = [uuid, algorithm_invalid, data]
with ProxyKmipClient() as client:
client.proxy.mac.return_value = result
self.assertRaises(TypeError, client.mac, *args)
args = [uuid, algorithm, data_invalid]
with ProxyKmipClient() as client:
client.proxy.mac.return_value = result
self.assertRaises(TypeError, client.mac, *args)
@mock.patch('kmip.pie.client.KMIPProxy',
mock.MagicMock(spec_set=KMIPProxy))
def test_mac_on_operation_failure(self):
"""
Test that a KmipOperationFailure exception is raised when the
backend fails to generate MAC.
"""
uuid = 'aaaaaaaa-1111-2222-3333-ffffffffffff'
algorithm = enums.CryptographicAlgorithm.HMAC_SHA256
data = (b'\x00\x01\x02\x03\x04')
status = enums.ResultStatus.OPERATION_FAILED
reason = enums.ResultReason.GENERAL_FAILURE
message = "Test failure message"
result = results.OperationResult(
contents.ResultStatus(status),
contents.ResultReason(reason),
contents.ResultMessage(message))
error_msg = str(KmipOperationFailure(status, reason, message))
client = ProxyKmipClient()
client.open()
client.proxy.mac.return_value = result
args = [uuid, algorithm, data]
self.assertRaisesRegexp(
KmipOperationFailure, error_msg, client.mac, *args)
@mock.patch('kmip.pie.client.KMIPProxy',
mock.MagicMock(spec_set=KMIPProxy))
def test_mac_on_closed(self):
"""
Test that a ClientConnectionNotOpen exception is raised when trying
to do mac on an unopened client connection.
"""
client = ProxyKmipClient()
uuid = 'aaaaaaaa-1111-2222-3333-ffffffffffff'
algorithm = enums.CryptographicAlgorithm.HMAC_SHA256
data = (b'\x00\x01\x02\x03\x04')
args = [uuid, algorithm, data]
self.assertRaises(
ClientConnectionNotOpen, client.mac, *args)

92
kmip/tests/unit/services/test_kmip_client.py
View File

@ -16,6 +16,9 @@
from testtools import TestCase
from kmip.core.attributes import PrivateKeyUniqueIdentifier
from kmip.core.attributes import CryptographicParameters, \
CryptographicAlgorithm
from kmip.core.enums import AuthenticationSuite
from kmip.core.enums import ConformanceClause
@ -24,6 +27,8 @@ from kmip.core.enums import ResultStatus as ResultStatusEnum
from kmip.core.enums import ResultReason as ResultReasonEnum
from kmip.core.enums import Operation as OperationEnum
from kmip.core.enums import QueryFunction as QueryFunctionEnum
from kmip.core.enums import CryptographicAlgorithm as \
CryptographicAlgorithmEnum
from kmip.core.factories.attributes import AttributeFactory
from kmip.core.factories.credentials import CredentialFactory
@ -714,6 +719,93 @@ class TestKMIPClient(TestCase):
self.client._create_socket(sock)
self.assertEqual(ssl.SSLSocket, type(self.client.socket))
@mock.patch('kmip.services.kmip_client.KMIPProxy._send_message',
mock.MagicMock())
@mock.patch('kmip.services.kmip_client.KMIPProxy._receive_message',
mock.MagicMock())
def test_mac(self):
from kmip.core.utils import BytearrayStream
request_expected = (
b'\x42\x00\x78\x01\x00\x00\x00\xa0\x42\x00\x77\x01\x00\x00\x00\x38'
b'\x42\x00\x69\x01\x00\x00\x00\x20\x42\x00\x6a\x02\x00\x00\x00\x04'
b'\x00\x00\x00\x01\x00\x00\x00\x00\x42\x00\x6b\x02\x00\x00\x00\x04'
b'\x00\x00\x00\x02\x00\x00\x00\x00\x42\x00\x0d\x02\x00\x00\x00\x04'
b'\x00\x00\x00\x01\x00\x00\x00\x00\x42\x00\x0f\x01\x00\x00\x00\x58'
b'\x42\x00\x5c\x05\x00\x00\x00\x04\x00\x00\x00\x23\x00\x00\x00\x00'
b'\x42\x00\x79\x01\x00\x00\x00\x40\x42\x00\x94\x07\x00\x00\x00\x01'
b'\x31\x00\x00\x00\x00\x00\x00\x00\x42\x00\x2b\x01\x00\x00\x00\x10'
b'\x42\x00\x28\x05\x00\x00\x00\x04\x00\x00\x00\x0b\x00\x00\x00\x00'
b'\x42\x00\xc2\x08\x00\x00\x00\x10\x00\x01\x02\x03\x04\x05\x06\x07'
b'\x08\x09\x0a\x0b\x0c\x0d\x0e\x0f')
response = (
b'\x42\x00\x7b\x01\x00\x00\x00\xd8\x42\x00\x7a\x01\x00\x00\x00\x48'
b'\x42\x00\x69\x01\x00\x00\x00\x20\x42\x00\x6a\x02\x00\x00\x00\x04'
b'\x00\x00\x00\x01\x00\x00\x00\x00\x42\x00\x6b\x02\x00\x00\x00\x04'
b'\x00\x00\x00\x02\x00\x00\x00\x00\x42\x00\x92\x09\x00\x00\x00\x08'
b'\x00\x00\x00\x00\x58\x8a\x3f\x23\x42\x00\x0d\x02\x00\x00\x00\x04'
b'\x00\x00\x00\x01\x00\x00\x00\x00\x42\x00\x0f\x01\x00\x00\x00\x80'
b'\x42\x00\x5c\x05\x00\x00\x00\x04\x00\x00\x00\x23\x00\x00\x00\x00'
b'\x42\x00\x7f\x05\x00\x00\x00\x04\x00\x00\x00\x00\x00\x00\x00\x00'
b'\x42\x00\x7c\x01\x00\x00\x00\x58\x42\x00\x94\x07\x00\x00\x00\x01'
b'\x31\x00\x00\x00\x00\x00\x00\x00\x42\x00\xc6\x08\x00\x00\x00\x40'
b'\x99\x8b\x55\x59\x90\x9b\x85\x87\x5b\x90\x63\x13\x12\xbb\x32\x9f'
b'\x6a\xc4\xed\x97\x6e\xac\x99\xe5\x21\x53\xc4\x19\x28\xf2\x2a\x5b'
b'\xef\x79\xa4\xbe\x05\x3b\x31\x49\x19\xe0\x75\x23\xb9\xbe\xc8\x23'
b'\x35\x60\x7e\x49\xba\xa9\x7e\xe0\x9e\x6b\x3d\x55\xf4\x51\xff\x7c'
)
response_no_payload = (
b'\x42\x00\x7b\x01\x00\x00\x00\x78\x42\x00\x7a\x01\x00\x00\x00\x48'
b'\x42\x00\x69\x01\x00\x00\x00\x20\x42\x00\x6a\x02\x00\x00\x00\x04'
b'\x00\x00\x00\x01\x00\x00\x00\x00\x42\x00\x6b\x02\x00\x00\x00\x04'
b'\x00\x00\x00\x02\x00\x00\x00\x00\x42\x00\x92\x09\x00\x00\x00\x08'
b'\x00\x00\x00\x00\x58\x8a\x3f\x23\x42\x00\x0d\x02\x00\x00\x00\x04'
b'\x00\x00\x00\x01\x00\x00\x00\x00\x42\x00\x0f\x01\x00\x00\x00\x80'
b'\x42\x00\x5c\x05\x00\x00\x00\x04\x00\x00\x00\x23\x00\x00\x00\x00'
b'\x42\x00\x7f\x05\x00\x00\x00\x04\x00\x00\x00\x00\x00\x00\x00\x00'
)
data = (b'\x00\x01\x02\x03\x04\x05\x06\x07\x08\x09\x0A\x0B'
b'\x0C\x0D\x0E\x0F')
mdata = (b'\x99\x8b\x55\x59\x90\x9b\x85\x87\x5b\x90\x63\x13'
b'\x12\xbb\x32\x9f'
b'\x6a\xc4\xed\x97\x6e\xac\x99\xe5\x21\x53\xc4\x19'
b'\x28\xf2\x2a\x5b'
b'\xef\x79\xa4\xbe\x05\x3b\x31\x49\x19\xe0\x75\x23'
b'\xb9\xbe\xc8\x23'
b'\x35\x60\x7e\x49\xba\xa9\x7e\xe0\x9e\x6b\x3d\x55'
b'\xf4\x51\xff\x7c')
def verify_request(message):
stream = BytearrayStream()
message.write(stream)
self.assertEqual(stream.buffer, request_expected)
uuid = '1'
cryptographic_parameters = CryptographicParameters(
cryptographic_algorithm=CryptographicAlgorithm(
CryptographicAlgorithmEnum.HMAC_SHA512)
)
self.client._send_message.side_effect = verify_request
self.client._receive_message.return_value = BytearrayStream(response)
result = self.client.mac(uuid, cryptographic_parameters,
data)
self.assertEqual(result.uuid.value, uuid)
self.assertEqual(result.mac_data.value, mdata)
self.client._receive_message.return_value = \
BytearrayStream(response_no_payload)
result = self.client.mac(uuid, cryptographic_parameters,
data)
self.assertEqual(result.uuid, None)
self.assertEqual(result.mac_data, None)
class TestClientProfileInformation(TestCase):
"""