mirror of https://github.com/OpenKMIP/PyKMIP.git
Update the Create payloads to support KMIP 2.0
This change updates the Create payloads to support KMIP 2.0 features, including swapping out TemplateAttributes for the new Attributes structure in the request payload and removing all attribute-related encodings from the response payload. Unit tests have been added to cover these changes.
This commit is contained in:
Peter Hamilton
Peter Hamilton
parent
b8ca94b518
commit
004ff92ffb
|
|
@ -126,17 +126,37 @@ class CreateRequestPayload(primitives.Struct):
|
|||
"type."
|
||||
)
|
||||
|
||||
if self.is_tag_next(enums.Tags.TEMPLATE_ATTRIBUTE, local_buffer):
|
||||
self._template_attribute = objects.TemplateAttribute()
|
||||
self._template_attribute.read(
|
||||
local_buffer,
|
||||
kmip_version=kmip_version
|
||||
)
|
||||
if kmip_version < enums.KMIPVersion.KMIP_2_0:
|
||||
if self.is_tag_next(enums.Tags.TEMPLATE_ATTRIBUTE, local_buffer):
|
||||
self._template_attribute = objects.TemplateAttribute()
|
||||
self._template_attribute.read(
|
||||
local_buffer,
|
||||
kmip_version=kmip_version
|
||||
)
|
||||
else:
|
||||
raise exceptions.InvalidKmipEncoding(
|
||||
"The Create request payload encoding is missing the "
|
||||
"template attribute."
|
||||
)
|
||||
else:
|
||||
raise exceptions.InvalidKmipEncoding(
|
||||
"The Create request payload encoding is missing the template "
|
||||
"attribute."
|
||||
)
|
||||
# NOTE (ph) For now, leave attributes natively in TemplateAttribute
|
||||
# form and just convert to the KMIP 2.0 Attributes form as needed
|
||||
# for encoding/decoding purposes. Changing the payload to require
|
||||
# the new Attributes structure will trigger a bunch of second-order
|
||||
# effects across the client and server codebases that is beyond
|
||||
# the scope of updating the Create payloads to support KMIP 2.0.
|
||||
if self.is_tag_next(enums.Tags.ATTRIBUTES, local_buffer):
|
||||
attributes = objects.Attributes()
|
||||
attributes.read(local_buffer, kmip_version=kmip_version)
|
||||
value = objects.convert_attributes_to_template_attribute(
|
||||
attributes
|
||||
)
|
||||
self._template_attribute = value
|
||||
else:
|
||||
raise exceptions.InvalidKmipEncoding(
|
||||
"The Create request payload encoding is missing the "
|
||||
"attributes structure."
|
||||
)
|
||||
|
||||
self.is_oversized(local_buffer)
|
||||
|
||||
|
|
@ -164,16 +184,34 @@ class CreateRequestPayload(primitives.Struct):
|
|||
"The Create request payload is missing the object type field."
|
||||
)
|
||||
|
||||
if self._template_attribute:
|
||||
self._template_attribute.write(
|
||||
local_buffer,
|
||||
kmip_version=kmip_version
|
||||
)
|
||||
if kmip_version < enums.KMIPVersion.KMIP_2_0:
|
||||
if self._template_attribute:
|
||||
self._template_attribute.write(
|
||||
local_buffer,
|
||||
kmip_version=kmip_version
|
||||
)
|
||||
else:
|
||||
raise exceptions.InvalidField(
|
||||
"The Create request payload is missing the template "
|
||||
"attribute field."
|
||||
)
|
||||
else:
|
||||
raise exceptions.InvalidField(
|
||||
"The Create request payload is missing the template attribute "
|
||||
"field."
|
||||
)
|
||||
# NOTE (ph) For now, leave attributes natively in TemplateAttribute
|
||||
# form and just convert to the KMIP 2.0 Attributes form as needed
|
||||
# for encoding/decoding purposes. Changing the payload to require
|
||||
# the new Attributes structure will trigger a bunch of second-order
|
||||
# effects across the client and server codebases that is beyond
|
||||
# the scope of updating the Create payloads to support KMIP 2.0.
|
||||
if self._template_attribute:
|
||||
attributes = objects.convert_template_attribute_to_attributes(
|
||||
self._template_attribute
|
||||
)
|
||||
attributes.write(local_buffer, kmip_version=kmip_version)
|
||||
else:
|
||||
raise exceptions.InvalidField(
|
||||
"The Create request payload is missing the template "
|
||||
"attribute field."
|
||||
)
|
||||
|
||||
self.length = local_buffer.length()
|
||||
super(CreateRequestPayload, self).write(
|
||||
|
|
@ -360,12 +398,13 @@ class CreateResponsePayload(primitives.Struct):
|
|||
"identifier."
|
||||
)
|
||||
|
||||
if self.is_tag_next(enums.Tags.TEMPLATE_ATTRIBUTE, local_buffer):
|
||||
self._template_attribute = objects.TemplateAttribute()
|
||||
self._template_attribute.read(
|
||||
local_buffer,
|
||||
kmip_version=kmip_version
|
||||
)
|
||||
if kmip_version < enums.KMIPVersion.KMIP_2_0:
|
||||
if self.is_tag_next(enums.Tags.TEMPLATE_ATTRIBUTE, local_buffer):
|
||||
self._template_attribute = objects.TemplateAttribute()
|
||||
self._template_attribute.read(
|
||||
local_buffer,
|
||||
kmip_version=kmip_version
|
||||
)
|
||||
|
||||
self.is_oversized(local_buffer)
|
||||
|
||||
|
|
@ -404,11 +443,12 @@ class CreateResponsePayload(primitives.Struct):
|
|||
"field."
|
||||
)
|
||||
|
||||
if self._template_attribute:
|
||||
self._template_attribute.write(
|
||||
local_buffer,
|
||||
kmip_version=kmip_version
|
||||
)
|
||||
if kmip_version < enums.KMIPVersion.KMIP_2_0:
|
||||
if self._template_attribute:
|
||||
self._template_attribute.write(
|
||||
local_buffer,
|
||||
kmip_version=kmip_version
|
||||
)
|
||||
|
||||
self.length = local_buffer.length()
|
||||
super(CreateResponsePayload, self).write(
|
||||
|
|
|
|||
|
|
@ -66,6 +66,25 @@ class TestCreateRequestPayload(testtools.TestCase):
|
|||
b'\x42\x00\x0B\x02\x00\x00\x00\x04\x00\x00\x00\x0C\x00\x00\x00\x00'
|
||||
)
|
||||
|
||||
# Encoding obtained from the KMIP 1.1 testing document,
|
||||
# Section 3.1.1, and manually converted into KMIP 2.0 format.
|
||||
#
|
||||
# This encoding matches the following set of values:
|
||||
# Request Payload
|
||||
# Object Type - Symmetric Key
|
||||
# Attributes
|
||||
# Cryptographic Algorithm - AES
|
||||
# Cryptographic Length - 128
|
||||
# Cryptographic Usage Mask - Encrypt | Decrypt
|
||||
self.full_encoding_with_attributes = utils.BytearrayStream(
|
||||
b'\x42\x00\x79\x01\x00\x00\x00\x48'
|
||||
b'\x42\x00\x57\x05\x00\x00\x00\x04\x00\x00\x00\x02\x00\x00\x00\x00'
|
||||
b'\x42\x01\x25\x01\x00\x00\x00\x30'
|
||||
b'\x42\x00\x28\x05\x00\x00\x00\x04\x00\x00\x00\x03\x00\x00\x00\x00'
|
||||
b'\x42\x00\x2A\x02\x00\x00\x00\x04\x00\x00\x00\x80\x00\x00\x00\x00'
|
||||
b'\x42\x00\x2C\x02\x00\x00\x00\x04\x00\x00\x00\x0C\x00\x00\x00\x00'
|
||||
)
|
||||
|
||||
# Encoding obtained from the KMIP 1.1 testing document,
|
||||
# Section 3.1.1.
|
||||
#
|
||||
|
|
@ -219,6 +238,64 @@ class TestCreateRequestPayload(testtools.TestCase):
|
|||
payload.template_attribute
|
||||
)
|
||||
|
||||
def test_read_kmip_2_0(self):
|
||||
"""
|
||||
Test that a Create request payload can be read from a data stream
|
||||
encoded with the KMIP 2.0 format.
|
||||
"""
|
||||
payload = payloads.CreateRequestPayload()
|
||||
|
||||
self.assertEqual(None, payload.object_type)
|
||||
self.assertEqual(None, payload.template_attribute)
|
||||
|
||||
payload.read(
|
||||
self.full_encoding_with_attributes,
|
||||
kmip_version=enums.KMIPVersion.KMIP_2_0
|
||||
)
|
||||
|
||||
self.assertEqual(
|
||||
enums.ObjectType.SYMMETRIC_KEY,
|
||||
payload.object_type
|
||||
)
|
||||
self.assertEqual(
|
||||
objects.TemplateAttribute(
|
||||
attributes=[
|
||||
objects.Attribute(
|
||||
attribute_name=objects.Attribute.AttributeName(
|
||||
'Cryptographic Algorithm'
|
||||
),
|
||||
attribute_value=primitives.Enumeration(
|
||||
enums.CryptographicAlgorithm,
|
||||
value=enums.CryptographicAlgorithm.AES,
|
||||
tag=enums.Tags.CRYPTOGRAPHIC_ALGORITHM
|
||||
)
|
||||
),
|
||||
objects.Attribute(
|
||||
attribute_name=objects.Attribute.AttributeName(
|
||||
'Cryptographic Length'
|
||||
),
|
||||
attribute_value=primitives.Integer(
|
||||
value=128,
|
||||
tag=enums.Tags.CRYPTOGRAPHIC_LENGTH
|
||||
)
|
||||
),
|
||||
objects.Attribute(
|
||||
attribute_name=objects.Attribute.AttributeName(
|
||||
'Cryptographic Usage Mask'
|
||||
),
|
||||
attribute_value=primitives.Integer(
|
||||
value=(
|
||||
enums.CryptographicUsageMask.ENCRYPT.value |
|
||||
enums.CryptographicUsageMask.DECRYPT.value
|
||||
),
|
||||
tag=enums.Tags.CRYPTOGRAPHIC_USAGE_MASK
|
||||
)
|
||||
)
|
||||
]
|
||||
),
|
||||
payload.template_attribute
|
||||
)
|
||||
|
||||
def test_read_missing_object_type(self):
|
||||
"""
|
||||
Test that an InvalidKmipEncoding error is raised during the decoding
|
||||
|
|
@ -258,6 +335,28 @@ class TestCreateRequestPayload(testtools.TestCase):
|
|||
*args
|
||||
)
|
||||
|
||||
def test_read_missing_attributes(self):
|
||||
"""
|
||||
Test that an InvalidKmipEncoding error is raised during the decoding
|
||||
of a Create request payload when the attributes structure is missing
|
||||
from the encoding.
|
||||
"""
|
||||
payload = payloads.CreateRequestPayload()
|
||||
|
||||
self.assertIsNone(payload.object_type)
|
||||
self.assertIsNone(payload.template_attribute)
|
||||
|
||||
args = (self.no_template_attribute_encoding, )
|
||||
kwargs = {"kmip_version": enums.KMIPVersion.KMIP_2_0}
|
||||
self.assertRaisesRegex(
|
||||
exceptions.InvalidKmipEncoding,
|
||||
"The Create request payload encoding is missing the attributes "
|
||||
"structure.",
|
||||
payload.read,
|
||||
*args,
|
||||
**kwargs
|
||||
)
|
||||
|
||||
def test_write(self):
|
||||
"""
|
||||
Test that a Create request payload can be written to a data stream.
|
||||
|
|
@ -307,6 +406,56 @@ class TestCreateRequestPayload(testtools.TestCase):
|
|||
self.assertEqual(len(self.full_encoding), len(stream))
|
||||
self.assertEqual(str(self.full_encoding), str(stream))
|
||||
|
||||
def test_write_kmip_2_0(self):
|
||||
"""
|
||||
Test that a Create request payload can be written to a data stream
|
||||
encoded with the KMIP 2.0 format.
|
||||
"""
|
||||
payload = payloads.CreateRequestPayload(
|
||||
object_type=enums.ObjectType.SYMMETRIC_KEY,
|
||||
template_attribute=objects.TemplateAttribute(
|
||||
attributes=[
|
||||
objects.Attribute(
|
||||
attribute_name=objects.Attribute.AttributeName(
|
||||
'Cryptographic Algorithm'
|
||||
),
|
||||
attribute_value=primitives.Enumeration(
|
||||
enums.CryptographicAlgorithm,
|
||||
value=enums.CryptographicAlgorithm.AES,
|
||||
tag=enums.Tags.CRYPTOGRAPHIC_ALGORITHM
|
||||
)
|
||||
),
|
||||
objects.Attribute(
|
||||
attribute_name=objects.Attribute.AttributeName(
|
||||
'Cryptographic Length'
|
||||
),
|
||||
attribute_value=primitives.Integer(
|
||||
value=128,
|
||||
tag=enums.Tags.CRYPTOGRAPHIC_LENGTH
|
||||
)
|
||||
),
|
||||
objects.Attribute(
|
||||
attribute_name=objects.Attribute.AttributeName(
|
||||
'Cryptographic Usage Mask'
|
||||
),
|
||||
attribute_value=primitives.Integer(
|
||||
value=(
|
||||
enums.CryptographicUsageMask.ENCRYPT.value |
|
||||
enums.CryptographicUsageMask.DECRYPT.value
|
||||
),
|
||||
tag=enums.Tags.CRYPTOGRAPHIC_USAGE_MASK
|
||||
)
|
||||
)
|
||||
]
|
||||
)
|
||||
)
|
||||
|
||||
stream = utils.BytearrayStream()
|
||||
payload.write(stream, kmip_version=enums.KMIPVersion.KMIP_2_0)
|
||||
|
||||
self.assertEqual(len(self.full_encoding_with_attributes), len(stream))
|
||||
self.assertEqual(str(self.full_encoding_with_attributes), str(stream))
|
||||
|
||||
def test_write_missing_object_type(self):
|
||||
"""
|
||||
Test that an InvalidField error is raised during the encoding of a
|
||||
|
|
@ -367,6 +516,28 @@ class TestCreateRequestPayload(testtools.TestCase):
|
|||
*args
|
||||
)
|
||||
|
||||
def test_write_missing_attributes(self):
|
||||
"""
|
||||
Test that an InvalidField error is raised during the encoding of a
|
||||
Create request payload when the payload is missing the template
|
||||
attribute.
|
||||
"""
|
||||
payload = payloads.CreateRequestPayload(
|
||||
object_type=enums.ObjectType.SYMMETRIC_KEY
|
||||
)
|
||||
|
||||
stream = utils.BytearrayStream()
|
||||
args = (stream, )
|
||||
kwargs = {"kmip_version": enums.KMIPVersion.KMIP_2_0}
|
||||
self.assertRaisesRegex(
|
||||
exceptions.InvalidField,
|
||||
"The Create request payload is missing the template attribute "
|
||||
"field.",
|
||||
payload.write,
|
||||
*args,
|
||||
**kwargs
|
||||
)
|
||||
|
||||
def test_repr(self):
|
||||
"""
|
||||
Test that repr can be applied to a Create request payload structure.
|
||||
|
|
@ -961,6 +1132,32 @@ class TestCreateResponsePayload(testtools.TestCase):
|
|||
payload.template_attribute
|
||||
)
|
||||
|
||||
def test_read_kmip_2_0(self):
|
||||
"""
|
||||
Test that a Create response payload can be read from a data stream
|
||||
encoded with the KMIP 2.0 format.
|
||||
"""
|
||||
payload = payloads.CreateResponsePayload()
|
||||
|
||||
self.assertIsNone(payload.object_type)
|
||||
self.assertIsNone(payload.unique_identifier)
|
||||
self.assertIsNone(payload.template_attribute)
|
||||
|
||||
payload.read(
|
||||
self.no_template_attribute_encoding,
|
||||
kmip_version=enums.KMIPVersion.KMIP_2_0
|
||||
)
|
||||
|
||||
self.assertEqual(
|
||||
enums.ObjectType.SYMMETRIC_KEY,
|
||||
payload.object_type
|
||||
)
|
||||
self.assertEqual(
|
||||
'fb4b5b9c-6188-4c63-8142-fe9c328129fc',
|
||||
payload.unique_identifier
|
||||
)
|
||||
self.assertIsNone(payload.template_attribute)
|
||||
|
||||
def test_read_missing_object_type(self):
|
||||
"""
|
||||
Test that an InvalidKmipEncoding error is raised during the decoding
|
||||
|
|
@ -1054,6 +1251,36 @@ class TestCreateResponsePayload(testtools.TestCase):
|
|||
self.assertEqual(len(self.full_encoding), len(stream))
|
||||
self.assertEqual(str(self.full_encoding), str(stream))
|
||||
|
||||
def test_write_kmip_2_0(self):
|
||||
"""
|
||||
Test that a Create response payload can be written to a data stream
|
||||
encoded with the KMIP 2.0 format.
|
||||
"""
|
||||
payload = payloads.CreateResponsePayload(
|
||||
object_type=enums.ObjectType.SYMMETRIC_KEY,
|
||||
unique_identifier="fb4b5b9c-6188-4c63-8142-fe9c328129fc",
|
||||
template_attribute=objects.TemplateAttribute(
|
||||
attributes=[
|
||||
objects.Attribute(
|
||||
attribute_name=objects.Attribute.AttributeName(
|
||||
"State"
|
||||
),
|
||||
attribute_value=primitives.Enumeration(
|
||||
enums.State,
|
||||
value=enums.State.PRE_ACTIVE,
|
||||
tag=enums.Tags.STATE
|
||||
)
|
||||
)
|
||||
]
|
||||
)
|
||||
)
|
||||
|
||||
stream = utils.BytearrayStream()
|
||||
payload.write(stream, kmip_version=enums.KMIPVersion.KMIP_2_0)
|
||||
|
||||
self.assertEqual(len(self.no_template_attribute_encoding), len(stream))
|
||||
self.assertEqual(str(self.no_template_attribute_encoding), str(stream))
|
||||
|
||||
def test_write_missing_object_type(self):
|
||||
"""
|
||||
Test that an InvalidField error is raised during the encoding of a
|
||||
|
|
|
|||
Loading…
Reference in New Issue