WIP Add request and response payloads for the DeriveKey operation

This commit is contained in:
Peter Hamilton 2017-06-22 13:42:23 -04:00
parent ed22fb15f9
commit b34105001c
2 changed files with 835 additions and 0 deletions
kmip
core/messages/payloads
tests/unit/core/messages/payloads

@ -0,0 +1,519 @@
# Copyright (c) 2017 The Johns Hopkins University/Applied Physics Laboratory
# 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 six
from kmip.core import attributes
from kmip.core import enums
from kmip.core import objects
from kmip.core import primitives
from kmip.core import utils
class DeriveKeyRequestPayload(primitives.Struct):
"""
A request payload for the DeriveKey operation.
Attributes:
object_type: The type of the object that should be derived.
unique_identifiers: A set of unique IDs of managed objects to be used
with the derivation process.
derivation_method: The method that should be used to derive the new
cryptographic object.
derivation_parameters: A collection of settings relevant for the
derivation method.
template_attribute: A collection of attributes that should be set on
the newly derived cryptographic object.
"""
def __init__(self,
object_type=None,
unique_identifiers=None,
derivation_method=None,
derivation_parameters=None,
template_attribute=None):
"""
Construct a DeriveKey request payload struct.
Args:
object_type (ObjectType): An enumeration specifying the type of
the object to derive. Optional, defaults to None. Required
for encoding and decoding.
unique_identifiers (list): A list of strings representing the IDs
of managed objects (e.g., symmetric keys) to be used for
derivation. Optional, defaults to None. At least one value is
required for encoding and decoding.
derivation_method (DerivationMethod): An enumeration specifying
the type of derivation function to use (e.g., PBKDF2).
Optional, defaults to None. Required for encoding and
decoding.
derivation_parameters (DerivationParameters): A structure
containing cryptographic settings relevant for the derivation
method. Optional, defaults to None. Required for encoding and
decoding.
template_attribute (TemplateAttribute): A structure containing a
set of attributes (e.g., cryptographic algorithm,
cryptographic length) that should be set on the newly derived
cryptographic object. Optional, defaults to None. Required
for encoding and decoding.
"""
super(DeriveKeyRequestPayload, self).__init__(
enums.Tags.REQUEST_PAYLOAD
)
self._object_type = None
self._unique_identifiers = None
self._derivation_method = None
self._derivation_parameters = None
self._template_attribute = None
self.object_type = object_type
self.unique_identifiers = unique_identifiers
self.derivation_method = derivation_method
self.derivation_parameters = derivation_parameters
self.template_attribute = template_attribute
@property
def object_type(self):
if self._object_type:
return self._object_type.value
else:
return None
@object_type.setter
def object_type(self, value):
if value is None:
self._object_type = None
elif isinstance(value, enums.ObjectType):
self._object_type = primitives.Enumeration(
enums.ObjectType,
value=value,
tag=enums.Tags.OBJECT_TYPE
)
else:
raise TypeError("object type must be an ObjectType enumeration")
@property
def unique_identifiers(self):
if self._unique_identifiers:
unique_identifiers = []
for i in self._unique_identifiers:
unique_identifiers.append(i.value)
return unique_identifiers
else:
return None
@unique_identifiers.setter
def unique_identifiers(self, value):
if value is None:
self._unique_identifiers = None
elif isinstance(value, list):
unique_identifiers = []
for i in value:
if isinstance(i, six.string_types):
unique_identifiers.append(
primitives.TextString(
value=i,
tag=enums.Tags.UNIQUE_IDENTIFIER
)
)
else:
raise TypeError(
"unique identifiers must be a list of strings"
)
self._unique_identifiers = unique_identifiers
else:
raise TypeError("unique identifiers must be a list of strings")
@property
def derivation_method(self):
if self._derivation_method:
return self._derivation_method.value
else:
return None
@derivation_method.setter
def derivation_method(self, value):
if value is None:
self._derivation_method = None
elif isinstance(value, enums.DerivationMethod):
self._derivation_method = primitives.Enumeration(
enums.DerivationMethod,
value=value,
tag=enums.Tags.DERIVATION_METHOD
)
else:
raise TypeError(
"derivation method must be a DerivationMethod enumeration"
)
@property
def derivation_parameters(self):
if self._derivation_parameters:
return self._derivation_parameters
else:
return None
@derivation_parameters.setter
def derivation_parameters(self, value):
if value is None:
self._derivation_parameters = None
elif isinstance(value, attributes.DerivationParameters):
self._derivation_parameters = value
else:
raise TypeError(
"derivation parameters must be a DerivationParameters struct"
)
@property
def template_attribute(self):
if self._template_attribute:
return self._template_attribute
else:
return None
@template_attribute.setter
def template_attribute(self, value):
if value is None:
self._template_attribute = None
elif isinstance(value, objects.TemplateAttribute):
self._template_attribute = value
else:
raise TypeError(
"template attribute must be a TemplateAttribute struct"
)
def read(self, input_stream):
"""
Read the data encoding the DeriveKey request payload and decode it
into its constituent parts.
Args:
input_stream (stream): A data stream containing encoded object
data, supporting a read method; usually a BytearrayStream
object.
Raises:
ValueError: Raised if the data attribute is missing from the
encoded payload.
"""
super(DeriveKeyRequestPayload, self).read(input_stream)
local_stream = utils.BytearrayStream(input_stream.read(self.length))
if self.is_tag_next(enums.Tags.OBJECT_TYPE, local_stream):
self._object_type = primitives.Enumeration(
enums.ObjectType,
tag=enums.Tags.OBJECT_TYPE
)
self._object_type.read(local_stream)
else:
raise ValueError(
"invalid payload missing object type"
)
unique_identifiers = []
while self.is_tag_next(enums.Tags.UNIQUE_IDENTIFIER, local_stream):
unique_identifier = primitives.TextString(
tag=enums.Tags.UNIQUE_IDENTIFIER
)
unique_identifier.read(local_stream)
unique_identifiers.append(unique_identifier)
if not unique_identifiers:
raise ValueError("invalid payload missing unique identifiers")
else:
self._unique_identifiers = unique_identifiers
if self.is_tag_next(enums.Tags.DERIVATION_METHOD, local_stream):
self._derivation_method = primitives.Enumeration(
enums.DerivationMethod,
tag=enums.Tags.DERIVATION_METHOD
)
self._derivation_method.read(local_stream)
else:
raise ValueError(
"invalid payload missing derivation method"
)
if self.is_tag_next(enums.Tags.DERIVATION_PARAMETERS, local_stream):
self._derivation_parameters = attributes.DerivationParameters()
self._derivation_parameters.read(local_stream)
else:
raise ValueError(
"invalid payload missing derivation parameters"
)
if self.is_tag_next(enums.Tags.TEMPLATE_ATTRIBUTE, local_stream):
self._template_attribute = objects.TemplateAttribute()
self._template_attribute.read(local_stream)
else:
raise ValueError(
"invalid payload missing template attribute"
)
self.is_oversized(local_stream)
def write(self, output_stream):
"""
Write the data encoding the DeriveKey request payload to a stream.
Args:
output_stream (stream): A data stream in which to encode object
data, supporting a write method; usually a BytearrayStream
object.
Raises:
ValueError: Raised if the data attribute is not defined.
"""
local_stream = utils.BytearrayStream()
if self._object_type:
self._object_type.write(local_stream)
else:
raise ValueError("invalid payload missing object type")
if self._unique_identifiers:
for unique_identifier in self._unique_identifiers:
unique_identifier.write(local_stream)
else:
raise ValueError("invalid payload missing unique identifiers")
if self._derivation_method:
self._derivation_method.write(local_stream)
else:
raise ValueError("invalid payload missing derivation method")
if self._derivation_parameters:
self._derivation_parameters.write(local_stream)
else:
raise ValueError("invalid payload missing derivation parameters")
if self._template_attribute:
self._template_attribute.write(local_stream)
else:
raise ValueError("invalid payload missing template attributes")
self.length = local_stream.length()
super(DeriveKeyRequestPayload, self).write(output_stream)
output_stream.write(local_stream.buffer)
def __eq__(self, other):
if isinstance(other, DeriveKeyRequestPayload):
if self.object_type != other.object_type:
return False
elif self.unique_identifiers != other.unique_identifiers:
return False
elif self.derivation_method != other.derivation_method:
return False
elif self.derivation_parameters != other.derivation_parameters:
return False
elif self.template_attribute != other.template_attribute:
return False
else:
return True
else:
return NotImplemented
def __ne__(self, other):
if isinstance(other, DeriveKeyRequestPayload):
return not (self == other)
else:
return NotImplemented
def __repr__(self):
args = ", ".join([
"object_type={0}".format(self.object_type),
"unique_identifiers={0}".format(self.unique_identifiers),
"derivation_method={0}".format(self.derivation_method),
"derivation_parameters={0}".format(
repr(self.derivation_parameters)
),
"template_attribute={0}".format(repr(self.template_attribute))
])
return "DeriveKeyRequestPayload({0})".format(args)
def __str__(self):
return str({
'object_type': self.object_type,
'unique_identifiers': self.unique_identifiers,
'derivation_method': self.derivation_method,
'derivation_parameters': self.derivation_parameters,
'template_attribute': self.template_attribute
})
class DeriveKeyResponsePayload(primitives.Struct):
"""
A response payload for the DeriveKey operation.
Attributes:
unique_identifier: The unique ID of the newly derived cryptographic
object.
template_attribute: A collection of attributes that were implicitly
set by the server on the newly derived cryptographic object.
"""
def __init__(self,
unique_identifier=None,
template_attribute=None):
"""
Construct a DeriveKey response payload struct.
Args:
unique_identifier (string): A string representing the ID of the
newly derived managed object. Optional, defaults to None. At
least one value is required for encoding and decoding.
template_attribute (TemplateAttribute): A structure containing a
set of attributes (e.g., cryptographic algorithm,
cryptographic length) implicitly set by the server on the
newly derived cryptographic object. Optional, defaults to
None.
"""
super(DeriveKeyResponsePayload, self).__init__(
enums.Tags.RESPONSE_PAYLOAD
)
self._unique_identifier = None
self._template_attribute = None
self.unique_identifier = unique_identifier
self.template_attribute = template_attribute
@property
def unique_identifier(self):
if self._unique_identifier:
return self._unique_identifier.value
else:
return None
@unique_identifier.setter
def unique_identifier(self, value):
if value is None:
self._unique_identifier = None
elif isinstance(value, six.string_types):
self._unique_identifier = primitives.TextString(
value=value,
tag=enums.Tags.UNIQUE_IDENTIFIER
)
else:
raise TypeError("unique identifier must be a string")
@property
def template_attribute(self):
if self._template_attribute:
return self._template_attribute
else:
return None
@template_attribute.setter
def template_attribute(self, value):
if value is None:
self._template_attribute = None
elif isinstance(value, objects.TemplateAttribute):
self._template_attribute = value
else:
raise TypeError(
"template attribute must be a TemplateAttribute struct"
)
def read(self, input_stream):
"""
Read the data encoding the DeriveKey response payload and decode it
into its constituent parts.
Args:
input_stream (stream): A data stream containing encoded object
data, supporting a read method; usually a BytearrayStream
object.
Raises:
ValueError: Raised if the data attribute is missing from the
encoded payload.
"""
super(DeriveKeyResponsePayload, self).read(input_stream)
local_stream = utils.BytearrayStream(input_stream.read(self.length))
if self.is_tag_next(enums.Tags.UNIQUE_IDENTIFIER, local_stream):
self._unique_identifier = primitives.TextString(
tag=enums.Tags.UNIQUE_IDENTIFIER
)
self._unique_identifier.read(local_stream)
else:
raise ValueError(
"invalid payload missing the unique identifier attribute"
)
if self.is_tag_next(enums.Tags.TEMPLATE_ATTRIBUTE, local_stream):
self._template_attribute = objects.TemplateAttribute()
self._template_attribute.read(local_stream)
self.is_oversized(local_stream)
def write(self, output_stream):
"""
Write the data encoding the DeriveKey response payload to a stream.
Args:
output_stream (stream): A data stream in which to encode object
data, supporting a write method; usually a BytearrayStream
object.
Raises:
ValueError: Raised if the data attribute is not defined.
"""
local_stream = utils.BytearrayStream()
if self._unique_identifier:
self._unique_identifier.write(local_stream)
else:
raise ValueError(
"invalid payload missing the unique identifier attribute"
)
if self._template_attribute:
self._template_attribute.write(local_stream)
self.length = local_stream.length()
super(DeriveKeyResponsePayload, self).write(output_stream)
output_stream.write(local_stream.buffer)
def __eq__(self, other):
if isinstance(other, DeriveKeyResponsePayload):
if self.unique_identifier != other.unique_identifier:
return False
elif self.template_attribute != other.template_attribute:
return False
else:
return True
else:
return NotImplemented
def __ne__(self, other):
if isinstance(other, DeriveKeyResponsePayload):
return not (self == other)
else:
return NotImplemented
def __repr__(self):
args = ", ".join([
"unique_identifier={0}".format(self.unique_identifier),
"template_attribute={0}".format(repr(self.template_attribute))
])
return "DeriveKeyResponsePayload({0})".format(args)
def __str__(self):
return str({
'unique_identifier': self.unique_identifier,
'template_attribute': self.template_attribute
})

@ -0,0 +1,316 @@
# Copyright (c) 2017 The Johns Hopkins University/Applied Physics Laboratory
# 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 testtools
from kmip.core import attributes
from kmip.core import enums
from kmip.core import objects
from kmip.core import utils
from kmip.core.messages.payloads import derive_key
class TestDeriveKeyRequestPayload(testtools.TestCase):
"""
Test suite for the DeriveKey request payload.
"""
def setUp(self):
super(TestDeriveKeyRequestPayload, self).setUp()
# Encoding obtained in part from the KMIP 1.1 testing document. The
# rest of the encoding is a manual construction, since DeriveKey is
# not specifically detailed by the testing document.
#
# This encoding matches the following set of values:
# Object Type - SymmetricKey
# Unique Identifiers
# fb4b5b9c-6188-4c63-8142-fe9c328129fc
# 5c9b81ef-4ee5-42cd-ba2d-c002fdd0c7b3
# 1703250b-4d40-4de2-93a0-c494a1d4ae40
# Derivation Method - HMAC
# Derivation Parameters
# Cryptographic Parameters
# Hashing Algorithm - SHA-256
# Initialization Vector - 0x39487432492834A3
# Derivation Data - 0xFAD98B6ACA6D87DD
# Template Attribute
# Attribute
# Attribute Name - Cryptographic Algorithm
# Attribute Value - AES
# Attribute
# Attribute Name - Cryptographic Length
# Attribute Value - 128
self.full_encoding = utils.BytearrayStream(
b'\x42\x00\x79\x01\x00\x00\x01\x68'
b'\x42\x00\x57\x05\x00\x00\x00\x04\x00\x00\x00\x02\x00\x00\x00\x00'
b'\x42\x00\x94\x07\x00\x00\x00\x24\x66\x62\x34\x62\x35\x62\x39\x63'
b'\x2D\x36\x31\x38\x38\x2D\x34\x63\x36\x33\x2D\x38\x31\x34\x32\x2D'
b'\x66\x65\x39\x63\x33\x32\x38\x31\x32\x39\x66\x63\x00\x00\x00\x00'
b'\x42\x00\x94\x07\x00\x00\x00\x24\x35\x63\x39\x62\x38\x31\x65\x66'
b'\x2D\x34\x65\x65\x35\x2D\x34\x32\x63\x64\x2D\x62\x61\x32\x64\x2D'
b'\x63\x30\x30\x32\x66\x64\x64\x30\x63\x37\x62\x33\x00\x00\x00\x00'
b'\x42\x00\x94\x07\x00\x00\x00\x24\x31\x37\x30\x33\x32\x35\x30\x62'
b'\x2D\x34\x64\x34\x30\x2D\x34\x64\x65\x32\x2D\x39\x33\x61\x30\x2D'
b'\x63\x34\x39\x34\x61\x31\x64\x34\x61\x65\x34\x30\x00\x00\x00\x00'
b'\x42\x00\x31\x05\x00\x00\x00\x04\x00\x00\x00\x02\x00\x00\x00\x00'
b'\x42\x00\x32\x01\x00\x00\x00\x38'
b'\x42\x00\x2B\x01\x00\x00\x00\x10'
b'\x42\x00\x38\x05\x00\x00\x00\x04\x00\x00\x00\x06\x00\x00\x00\x00'
b'\x42\x00\x3A\x08\x00\x00\x00\x08\x39\x48\x74\x32\x49\x28\x34\xA3'
b'\x42\x00\x30\x08\x00\x00\x00\x08\xFA\xD9\x8B\x6A\xCA\x6D\x87\xDD'
b'\x42\x00\x91\x01\x00\x00\x00\x70'
b'\x42\x00\x08\x01\x00\x00\x00\x30'
b'\x42\x00\x0A\x07\x00\x00\x00\x17\x43\x72\x79\x70\x74\x6F\x67\x72'
b'\x61\x70\x68\x69\x63\x20\x41\x6C\x67\x6F\x72\x69\x74\x68\x6D\x00'
b'\x42\x00\x0B\x05\x00\x00\x00\x04\x00\x00\x00\x03\x00\x00\x00\x00'
b'\x42\x00\x08\x01\x00\x00\x00\x30'
b'\x42\x00\x0A\x07\x00\x00\x00\x14\x43\x72\x79\x70\x74\x6F\x67\x72'
b'\x61\x70\x68\x69\x63\x20\x4C\x65\x6E\x67\x74\x68\x00\x00\x00\x00'
b'\x42\x00\x0B\x02\x00\x00\x00\x04\x00\x00\x00\x80\x00\x00\x00\x00'
)
def tearDown(self):
super(TestDeriveKeyRequestPayload, self).tearDown()
def test_init(self):
"""
Test that a DeriveKey request payload can be constructed with no
arguments.
"""
payload = derive_key.DeriveKeyRequestPayload()
self.assertEqual(None, payload.object_type)
self.assertEqual(None, payload.unique_identifiers)
self.assertEqual(None, payload.derivation_method)
self.assertEqual(None, payload.derivation_parameters)
self.assertEqual(None, payload.template_attribute)
def test_init_with_args(self):
"""
Test that a DeriveKey request payload can be constructed with valid
values
"""
payload = derive_key.DeriveKeyRequestPayload(
object_type=enums.ObjectType.SYMMETRIC_KEY,
unique_identifiers=['00000000-1111-2222-3333-444444444444'],
derivation_method=enums.DerivationMethod.HASH,
derivation_parameters=attributes.DerivationParameters(),
template_attribute=objects.TemplateAttribute()
)
self.assertEqual(
enums.ObjectType.SYMMETRIC_KEY,
payload.object_type
)
self.assertEqual(
['00000000-1111-2222-3333-444444444444'],
payload.unique_identifiers
)
self.assertEqual(
enums.DerivationMethod.HASH,
payload.derivation_method
)
self.assertEqual(
attributes.DerivationParameters(),
payload.derivation_parameters
)
self.assertEqual(
objects.TemplateAttribute(),
payload.template_attribute
)
def test_invalid_object_type(self):
"""
Test that a TypeError is raised when an invalid value is used to set
the object type of a DeriveKey request payload.
"""
payload = derive_key.DeriveKeyRequestPayload()
args = (payload, 'object_type', 'invalid')
self.assertRaisesRegexp(
TypeError,
"object type must be an ObjectType enumeration",
setattr,
*args
)
def test_invalid_unique_identifiers(self):
"""
Test that a TypeError is raised when invalid values are used to set
the unique identifiers of a DeriveKey request payload.
"""
payload = derive_key.DeriveKeyRequestPayload()
args = (payload, 'unique_identifiers', 'invalid')
self.assertRaisesRegexp(
TypeError,
"unique identifiers must be a list of strings",
setattr,
*args
)
args = (payload, 'unique_identifiers', [0])
self.assertRaisesRegexp(
TypeError,
"unique identifiers must be a list of strings",
setattr,
*args
)
args = (payload, 'unique_identifiers', ['valid', 'valid', 0])
self.assertRaisesRegexp(
TypeError,
"unique identifiers must be a list of strings",
setattr,
*args
)
def test_invalid_derivation_method(self):
"""
Test that a TypeError is raised when an invalid value is used to set
the derivation method of a DeriveKey request payload.
"""
payload = derive_key.DeriveKeyRequestPayload()
args = (payload, 'derivation_method', 'invalid')
self.assertRaisesRegexp(
TypeError,
"derivation method must be a DerivationMethod enumeration",
setattr,
*args
)
def test_invalid_derivation_parameters(self):
"""
Test that a TypeError is raised when an invalid value is used to set
the derivation parameters of a DeriveKey request payload.
"""
payload = derive_key.DeriveKeyRequestPayload()
args = (payload, 'derivation_parameters', 'invalid')
self.assertRaisesRegexp(
TypeError,
"derivation parameters must be a DerivationParameters struct",
setattr,
*args
)
def test_invalid_template_attribute(self):
"""
Test that a TypeError is raised when an invalid value is used to set
the template attribute of a DeriveKey request payload.
"""
payload = derive_key.DeriveKeyRequestPayload()
args = (payload, 'template_attribute', 'invalid')
self.assertRaisesRegexp(
TypeError,
"template attribute must be a TemplateAttribute struct",
setattr,
*args
)
def test_read(self):
"""
Test that a DeriveKey request payload can be read from a data stream.
"""
payload = derive_key.DeriveKeyRequestPayload()
self.assertEqual(None, payload.object_type)
self.assertEqual(None, payload.unique_identifiers)
self.assertEqual(None, payload.derivation_method)
self.assertEqual(None, payload.derivation_parameters)
self.assertEqual(None, payload.template_attribute)
payload.read(self.full_encoding)
def test_read_missing_object_type(self):
self.skip('')
def test_read_missing_unique_identifiers(self):
self.skip('')
def test_read_missing_derivation_method(self):
self.skip('')
def test_read_missing_derivation_parameters(self):
self.skip('')
def test_read_missing_template_attribute(self):
self.skip('')
def test_write(self):
self.skip('')
def test_write_missing_object_type(self):
self.skip('')
def test_write_missing_unique_identifiers(self):
self.skip('')
def test_write_missing_derivation_method(self):
self.skip('')
def test_write_missing_derivation_parameters(self):
self.skip('')
def test_write_missing_template_attribute(self):
self.skip('')
def test_equal_on_equal(self):
self.skip('')
def test_equal_on_not_equal_object_type(self):
self.skip('')
def test_equal_on_not_equal_unique_identifiers(self):
self.skip('')
def test_equal_on_not_equal_derivation_method(self):
self.skip('')
def test_equal_on_not_equal_derivation_parameters(self):
self.skip('')
def test_equal_on_not_equal_template_attribute(self):
self.skip('')
def test_equal_on_type_mismatch(self):
self.skip('')
def test_not_equal_on_equal(self):
self.skip('')
def test_not_equal_on_not_equal_object_type(self):
self.skip('')
def test_not_equal_on_not_equal_unique_identifiers(self):
self.skip('')
def test_not_equal_on_not_equal_derivation_method(self):
self.skip('')
def test_not_equal_on_not_equal_derivation_parameters(self):
self.skip('')
def test_not_equal_on_not_equal_template_attribute(self):
self.skip('')
def test_not_equal_on_type_mismatch(self):
self.skip('')
def test_repr(self):
self.skip('')
def test_str(self):
self.skip('')