Add the DefaultsInformation structure

This change adds the DefaultsInformation structure, a KMIP 2.0
addition that is used to specify default attribute values for
KMIP managed objects. A unit test suite is included to cover the
new structure.
This commit is contained in:
Peter Hamilton 2019-04-11 16:28:29 -04:00 committed by Peter Hamilton
parent dae68f7e46
commit 01b724b4e2
2 changed files with 741 additions and 0 deletions

View File

@ -3840,3 +3840,165 @@ class ObjectDefaults(primitives.Struct):
return not (self == other)
else:
return NotImplemented
class DefaultsInformation(primitives.Struct):
"""
"""
def __init__(self, object_defaults=None):
"""
"""
super(DefaultsInformation, self).__init__(
tag=enums.Tags.DEFAULTS_INFORMATION
)
self._object_defaults = None
self.object_defaults = object_defaults
@property
def object_defaults(self):
return self._object_defaults
@object_defaults.setter
def object_defaults(self, value):
if value is None:
self._object_defaults = None
elif isinstance(value, list):
object_defaults = []
for v in value:
if not isinstance(v, ObjectDefaults):
raise TypeError(
"Object defaults must be a list of ObjectDefaults "
"structures."
)
else:
object_defaults.append(v)
self._object_defaults = object_defaults
else:
raise TypeError(
"Object defaults must be a list of ObjectDefaults structures."
)
def read(self, input_buffer, kmip_version=enums.KMIPVersion.KMIP_2_0):
"""
Read the data encoding the DefaultsInformation structure and decode it
into its constituent parts.
Args:
input_buffer (stream): A data stream containing encoded object
data, supporting a read method; usually a BytearrayStream
object.
kmip_version (KMIPVersion): An enumeration defining the KMIP
version with which the object will be decoded. Optional,
defaults to KMIP 2.0.
Raises:
InvalidKmipEncoding: Raised if the object defaults are missing
from the encoding.
VersionNotSupported: Raised when a KMIP version is provided that
does not support the DefaultsInformation structure.
"""
if kmip_version < enums.KMIPVersion.KMIP_2_0:
raise exceptions.VersionNotSupported(
"KMIP {} does not support the DefaultsInformation "
"object.".format(
kmip_version.value
)
)
super(DefaultsInformation, self).read(
input_buffer,
kmip_version=kmip_version
)
local_buffer = utils.BytearrayStream(input_buffer.read(self.length))
object_defaults = []
while self.is_tag_next(enums.Tags.OBJECT_DEFAULTS, local_buffer):
object_default = ObjectDefaults()
object_default.read(local_buffer, kmip_version=kmip_version)
object_defaults.append(object_default)
if len(object_defaults) == 0:
raise exceptions.InvalidKmipEncoding(
"The DefaultsInformation encoding is missing the object "
"defaults structure."
)
else:
self._object_defaults = object_defaults
self.is_oversized(local_buffer)
def write(self, output_buffer, kmip_version=enums.KMIPVersion.KMIP_2_0):
"""
Write the DefaultsInformation structure encoding to the data stream.
Args:
output_buffer (stream): A data stream in which to encode
Attributes structure data, supporting a write method.
kmip_version (enum): A KMIPVersion enumeration defining the KMIP
version with which the object will be encoded. Optional,
defaults to KMIP 2.0.
Raises:
InvalidField: Raised if the object defaults field is not defined.
VersionNotSupported: Raised when a KMIP version is provided that
does not support the DefaultsInformation structure.
"""
if kmip_version < enums.KMIPVersion.KMIP_2_0:
raise exceptions.VersionNotSupported(
"KMIP {} does not support the DefaultsInformation "
"object.".format(
kmip_version.value
)
)
local_buffer = BytearrayStream()
if self._object_defaults:
for object_default in self._object_defaults:
object_default.write(local_buffer, kmip_version=kmip_version)
else:
raise exceptions.InvalidField(
"The DefaultsInformation structure is missing the object "
"defaults field."
)
self.length = local_buffer.length()
super(DefaultsInformation, self).write(
output_buffer,
kmip_version=kmip_version
)
output_buffer.write(local_buffer.buffer)
def __repr__(self):
d = "object_defaults={}".format(
'{}'.format(
repr(self.object_defaults)
) if self.object_defaults else None
)
return "DefaultsInformation({})".format(d)
def __str__(self):
d = '"object_defaults": {}'.format(
"[{}]".format(
", ".join([str(x) for x in self.object_defaults])
) if self.object_defaults else None
)
return '{' + d + '}'
def __eq__(self, other):
if isinstance(other, DefaultsInformation):
if self.object_defaults == other.object_defaults:
return True
else:
return False
else:
return NotImplemented
def __ne__(self, other):
if isinstance(other, DefaultsInformation):
return not (self == other)
else:
return NotImplemented

View File

@ -5143,3 +5143,582 @@ class TestObjectDefaults(testtools.TestCase):
self.assertTrue(a != b)
self.assertTrue(b != a)
class TestDefaultsInformation(testtools.TestCase):
def setUp(self):
super(TestDefaultsInformation, self).setUp()
# This encoding matches the following set of values:
#
# DefaultsInformation
# ObjectDefaults
# Object Type - Symmetric Key
# Attributes
# Cryptographic Algorithm - AES
# Cryptographic Length - 128
# Cryptographic Usage Mask - Encrypt | Decrypt
self.full_encoding = utils.BytearrayStream(
b'\x42\x01\x52\x01\x00\x00\x00\x50'
b'\x42\x01\x53\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'
)
# This encoding matches the following set of values:
#
# DefaultsInformation
self.no_object_defaults_encoding = utils.BytearrayStream(
b'\x42\x01\x52\x01\x00\x00\x00\x00'
)
def tearDown(self):
super(TestDefaultsInformation, self).tearDown()
def test_invalid_object_defaults(self):
"""
Test that a TypeError is raised when an invalid value is used to set
the object defaults of an DefaultsInformation structure.
"""
kwargs = {"object_defaults": [0]}
self.assertRaisesRegex(
TypeError,
"Object defaults must be a list of ObjectDefaults structures.",
objects.DefaultsInformation,
**kwargs
)
args = (objects.DefaultsInformation(), "object_defaults", 0)
self.assertRaisesRegex(
TypeError,
"Object defaults must be a list of ObjectDefaults structures.",
setattr,
*args
)
def test_invalid_object_defaults_list(self):
"""
Test that a TypeError is raised when an invalid object defaults list
is used with the DefaultsInformation structure.
"""
kwargs = {"object_defaults": "invalid"}
self.assertRaisesRegex(
TypeError,
"Object defaults must be a list of ObjectDefaults structures.",
objects.DefaultsInformation,
**kwargs
)
args = (
objects.DefaultsInformation(),
"object_defaults",
"invalid"
)
self.assertRaisesRegex(
TypeError,
"Object defaults must be a list of ObjectDefaults structures.",
setattr,
*args
)
def test_read(self):
"""
Test that a DefaultsInformation structure can be correctly read in
from a data stream.
"""
defaults_information = objects.DefaultsInformation()
self.assertIsNone(defaults_information.object_defaults)
defaults_information.read(
self.full_encoding,
kmip_version=enums.KMIPVersion.KMIP_2_0
)
expected = [
objects.ObjectDefaults(
object_type=enums.ObjectType.SYMMETRIC_KEY,
attributes=objects.Attributes(
attributes=[
primitives.Enumeration(
enums.CryptographicAlgorithm,
value=enums.CryptographicAlgorithm.AES,
tag=enums.Tags.CRYPTOGRAPHIC_ALGORITHM
),
primitives.Integer(
value=128,
tag=enums.Tags.CRYPTOGRAPHIC_LENGTH
),
primitives.Integer(
value=(
enums.CryptographicUsageMask.ENCRYPT.value |
enums.CryptographicUsageMask.DECRYPT.value
),
tag=enums.Tags.CRYPTOGRAPHIC_USAGE_MASK
)
]
)
)
]
self.assertEqual(
expected,
defaults_information.object_defaults
)
def test_read_unsupported_kmip_version(self):
"""
Test that a VersionNotSupported error is raised during the decoding of
a DefaultsInformation structure when the structure is read for an
unsupported KMIP version.
"""
defaults_information = objects.DefaultsInformation()
args = (self.full_encoding, )
kwargs = {"kmip_version": enums.KMIPVersion.KMIP_1_4}
self.assertRaisesRegex(
exceptions.VersionNotSupported,
"KMIP 1.4 does not support the DefaultsInformation object.",
defaults_information.read,
*args,
**kwargs
)
def test_read_missing_object_defaults(self):
"""
Test that an InvalidKmipEncoding error is raised during the decoding
of a DefaultsInformation structure when the object type is missing
from the encoding.
"""
defaults_information = objects.DefaultsInformation()
self.assertIsNone(defaults_information.object_defaults)
args = (self.no_object_defaults_encoding, )
self.assertRaisesRegex(
exceptions.InvalidKmipEncoding,
"The DefaultsInformation encoding is missing the object defaults "
"structure.",
defaults_information.read,
*args
)
def test_write(self):
"""
Test that a DefaultsInformation structure can be written to a data
stream.
"""
object_defaults = [
objects.ObjectDefaults(
object_type=enums.ObjectType.SYMMETRIC_KEY,
attributes=objects.Attributes(
attributes=[
primitives.Enumeration(
enums.CryptographicAlgorithm,
value=enums.CryptographicAlgorithm.AES,
tag=enums.Tags.CRYPTOGRAPHIC_ALGORITHM
),
primitives.Integer(
value=128,
tag=enums.Tags.CRYPTOGRAPHIC_LENGTH
),
primitives.Integer(
value=(
enums.CryptographicUsageMask.ENCRYPT.value |
enums.CryptographicUsageMask.DECRYPT.value
),
tag=enums.Tags.CRYPTOGRAPHIC_USAGE_MASK
)
]
)
)
]
defaults_information = objects.DefaultsInformation(
object_defaults=object_defaults
)
buffer = utils.BytearrayStream()
defaults_information.write(
buffer,
kmip_version=enums.KMIPVersion.KMIP_2_0
)
self.assertEqual(len(self.full_encoding), len(buffer))
self.assertEqual(str(self.full_encoding), str(buffer))
def test_write_unsupported_kmip_version(self):
"""
Test that a VersionNotSupported error is raised during the encoding of
a DefaultsInformation structure when the structure is written for an
unsupported KMIP version.
"""
object_defaults = objects.ObjectDefaults(
object_type=enums.ObjectType.SYMMETRIC_KEY,
attributes=objects.Attributes(
attributes=[
primitives.Enumeration(
enums.CryptographicAlgorithm,
value=enums.CryptographicAlgorithm.AES,
tag=enums.Tags.CRYPTOGRAPHIC_ALGORITHM
),
primitives.Integer(
value=128,
tag=enums.Tags.CRYPTOGRAPHIC_LENGTH
),
primitives.Integer(
value=(
enums.CryptographicUsageMask.ENCRYPT.value |
enums.CryptographicUsageMask.DECRYPT.value
),
tag=enums.Tags.CRYPTOGRAPHIC_USAGE_MASK
)
]
)
)
defaults_information = objects.DefaultsInformation(
object_defaults=[object_defaults]
)
args = (utils.BytearrayStream(), )
kwargs = {"kmip_version": enums.KMIPVersion.KMIP_1_4}
self.assertRaisesRegex(
exceptions.VersionNotSupported,
"KMIP 1.4 does not support the DefaultsInformation object.",
defaults_information.write,
*args,
**kwargs
)
def test_write_missing_object_defaults(self):
"""
Test that an InvalidField error is raised during the encoding of an
DefaultsInformation structure when the structure is missing the object
defaults field.
"""
defaults_information = objects.DefaultsInformation()
args = (utils.BytearrayStream(), )
self.assertRaisesRegex(
exceptions.InvalidField,
"The DefaultsInformation structure is missing the object defaults "
"field.",
defaults_information.write,
*args
)
def test_repr(self):
"""
Test that repr can be applied to a DefaultsInformation structure.
"""
object_defaults = objects.ObjectDefaults(
object_type=enums.ObjectType.SYMMETRIC_KEY,
attributes=objects.Attributes(
attributes=[
primitives.Enumeration(
enums.CryptographicAlgorithm,
value=enums.CryptographicAlgorithm.AES,
tag=enums.Tags.CRYPTOGRAPHIC_ALGORITHM
),
primitives.Integer(
value=128,
tag=enums.Tags.CRYPTOGRAPHIC_LENGTH
),
primitives.Integer(
value=(
enums.CryptographicUsageMask.ENCRYPT.value |
enums.CryptographicUsageMask.DECRYPT.value
),
tag=enums.Tags.CRYPTOGRAPHIC_USAGE_MASK
)
]
)
)
defaults_information = objects.DefaultsInformation(
object_defaults=[object_defaults]
)
o = "object_type=ObjectType.SYMMETRIC_KEY"
a1e = "enum=CryptographicAlgorithm"
a1v = "value=CryptographicAlgorithm.AES"
a1t = "tag=Tags.CRYPTOGRAPHIC_ALGORITHM"
a1a = ", ".join([a1e, a1v, a1t])
a1 = "Enumeration({})".format(a1a)
a2 = "Integer(value=128)"
a3 = "Integer(value=12)"
aa = ", ".join([a1, a2, a3])
t = "tag=Tags.ATTRIBUTES"
a = "attributes=Attributes(attributes=[{}], {})".format(aa, t)
r = "ObjectDefaults({}, {})".format(o, a)
d = "DefaultsInformation(object_defaults=[{}])".format(r)
self.assertEqual(d, repr(defaults_information))
def test_str(self):
"""
Test that str can be applied to a DefaultsInformation structure.
"""
object_defaults = objects.ObjectDefaults(
object_type=enums.ObjectType.SYMMETRIC_KEY,
attributes=objects.Attributes(
attributes=[
primitives.Enumeration(
enums.CryptographicAlgorithm,
value=enums.CryptographicAlgorithm.AES,
tag=enums.Tags.CRYPTOGRAPHIC_ALGORITHM
),
primitives.Integer(
value=128,
tag=enums.Tags.CRYPTOGRAPHIC_LENGTH
),
primitives.Integer(
value=(
enums.CryptographicUsageMask.ENCRYPT.value |
enums.CryptographicUsageMask.DECRYPT.value
),
tag=enums.Tags.CRYPTOGRAPHIC_USAGE_MASK
)
]
)
)
defaults_information = objects.DefaultsInformation(
object_defaults=[object_defaults]
)
o = '"object_type": ObjectType.SYMMETRIC_KEY'
aa = '{"attributes": [CryptographicAlgorithm.AES, 128, 12]}'
a = '"attributes": {}'.format(aa)
r = "{" + "{}, {}".format(o, a) + "}"
d = "{" + '"object_defaults": [' + r + "]}"
self.assertEqual(d, str(defaults_information))
def test_equal_on_equal(self):
"""
Test that the equality operator returns True when comparing two
DefaultsInformation structures with the same data.
"""
a = objects.DefaultsInformation()
b = objects.DefaultsInformation()
self.assertTrue(a == b)
self.assertTrue(b == a)
a = objects.DefaultsInformation(
object_defaults=[objects.ObjectDefaults(
object_type=enums.ObjectType.SYMMETRIC_KEY,
attributes=objects.Attributes(
attributes=[
primitives.Enumeration(
enums.CryptographicAlgorithm,
value=enums.CryptographicAlgorithm.AES,
tag=enums.Tags.CRYPTOGRAPHIC_ALGORITHM
),
primitives.Integer(
value=128,
tag=enums.Tags.CRYPTOGRAPHIC_LENGTH
),
primitives.Integer(
value=(
enums.CryptographicUsageMask.ENCRYPT.value |
enums.CryptographicUsageMask.DECRYPT.value
),
tag=enums.Tags.CRYPTOGRAPHIC_USAGE_MASK
)
]
)
)]
)
b = objects.DefaultsInformation(
object_defaults=[objects.ObjectDefaults(
object_type=enums.ObjectType.SYMMETRIC_KEY,
attributes=objects.Attributes(
attributes=[
primitives.Enumeration(
enums.CryptographicAlgorithm,
value=enums.CryptographicAlgorithm.AES,
tag=enums.Tags.CRYPTOGRAPHIC_ALGORITHM
),
primitives.Integer(
value=128,
tag=enums.Tags.CRYPTOGRAPHIC_LENGTH
),
primitives.Integer(
value=(
enums.CryptographicUsageMask.ENCRYPT.value |
enums.CryptographicUsageMask.DECRYPT.value
),
tag=enums.Tags.CRYPTOGRAPHIC_USAGE_MASK
)
]
)
)]
)
self.assertTrue(a == b)
self.assertTrue(b == a)
def test_equal_on_not_equal_object_defaults(self):
"""
Test that the equality operator returns False when comparing two
DefaultsInformation structures with different object defaults fields.
"""
a = objects.DefaultsInformation(
object_defaults=[objects.ObjectDefaults(
object_type=enums.ObjectType.SYMMETRIC_KEY,
attributes=objects.Attributes(
attributes=[
primitives.Enumeration(
enums.CryptographicAlgorithm,
value=enums.CryptographicAlgorithm.AES,
tag=enums.Tags.CRYPTOGRAPHIC_ALGORITHM
)
]
)
)]
)
b = objects.DefaultsInformation(
object_defaults=[objects.ObjectDefaults(
object_type=enums.ObjectType.SYMMETRIC_KEY,
attributes=objects.Attributes(
attributes=[
primitives.Integer(
value=(
enums.CryptographicUsageMask.ENCRYPT.value |
enums.CryptographicUsageMask.DECRYPT.value
),
tag=enums.Tags.CRYPTOGRAPHIC_USAGE_MASK
)
]
)
)]
)
self.assertFalse(a == b)
self.assertFalse(b == a)
def test_equal_on_type_mismatch(self):
"""
Test that the equality operator returns False when comparing two
DefaultsInformation structures with different types.
"""
a = objects.DefaultsInformation()
b = "invalid"
self.assertFalse(a == b)
self.assertFalse(b == a)
def test_not_equal_on_equal(self):
"""
Test that the inequality operator returns False when comparing two
DefaultsInformation structures with the same data.
"""
a = objects.DefaultsInformation()
b = objects.DefaultsInformation()
self.assertFalse(a != b)
self.assertFalse(b != a)
a = objects.DefaultsInformation(
object_defaults=[objects.ObjectDefaults(
object_type=enums.ObjectType.SYMMETRIC_KEY,
attributes=objects.Attributes(
attributes=[
primitives.Enumeration(
enums.CryptographicAlgorithm,
value=enums.CryptographicAlgorithm.AES,
tag=enums.Tags.CRYPTOGRAPHIC_ALGORITHM
),
primitives.Integer(
value=128,
tag=enums.Tags.CRYPTOGRAPHIC_LENGTH
),
primitives.Integer(
value=(
enums.CryptographicUsageMask.ENCRYPT.value |
enums.CryptographicUsageMask.DECRYPT.value
),
tag=enums.Tags.CRYPTOGRAPHIC_USAGE_MASK
)
]
)
)]
)
b = objects.DefaultsInformation(
object_defaults=[objects.ObjectDefaults(
object_type=enums.ObjectType.SYMMETRIC_KEY,
attributes=objects.Attributes(
attributes=[
primitives.Enumeration(
enums.CryptographicAlgorithm,
value=enums.CryptographicAlgorithm.AES,
tag=enums.Tags.CRYPTOGRAPHIC_ALGORITHM
),
primitives.Integer(
value=128,
tag=enums.Tags.CRYPTOGRAPHIC_LENGTH
),
primitives.Integer(
value=(
enums.CryptographicUsageMask.ENCRYPT.value |
enums.CryptographicUsageMask.DECRYPT.value
),
tag=enums.Tags.CRYPTOGRAPHIC_USAGE_MASK
)
]
)
)]
)
self.assertFalse(a != b)
self.assertFalse(b != a)
def test_not_equal_on_not_equal_object_defaults(self):
"""
Test that the inequality operator returns True when comparing two
DefaultsInformation structures with different object defaults fields.
"""
a = objects.DefaultsInformation(
object_defaults=[objects.ObjectDefaults(
object_type=enums.ObjectType.SYMMETRIC_KEY,
attributes=objects.Attributes(
attributes=[
primitives.Enumeration(
enums.CryptographicAlgorithm,
value=enums.CryptographicAlgorithm.AES,
tag=enums.Tags.CRYPTOGRAPHIC_ALGORITHM
)
]
)
)]
)
b = objects.DefaultsInformation(
object_defaults=[objects.ObjectDefaults(
object_type=enums.ObjectType.SYMMETRIC_KEY,
attributes=objects.Attributes(
attributes=[
primitives.Integer(
value=(
enums.CryptographicUsageMask.ENCRYPT.value |
enums.CryptographicUsageMask.DECRYPT.value
),
tag=enums.Tags.CRYPTOGRAPHIC_USAGE_MASK
)
]
)
)]
)
self.assertTrue(a != b)
self.assertTrue(b != a)
def test_not_equal_on_type_mismatch(self):
"""
Test that the inequality operator returns True when comparing two
DefaultsInformation structures with different types.
"""
a = objects.DefaultsInformation()
b = "invalid"
self.assertTrue(a != b)
self.assertTrue(b != a)