344cce83da
This change adds request/response payloads for the DeriveKey operation, updating the payload factories to support DeriveKey. Unit tests for the payloads and payload factories are included. The inequality operator was also added to the TemplateAttribute struct to support the DeriveKey payload unit tests; differences in Python2.7 and 3+ led to comparison failures.
1336 lines
44 KiB
Python
1336 lines
44 KiB
Python
# Copyright (c) 2014 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.
|
|
|
|
from six.moves import xrange
|
|
|
|
from kmip.core import attributes
|
|
from kmip.core.attributes import CryptographicParameters
|
|
|
|
from kmip.core.factories.attribute_values import AttributeValueFactory
|
|
|
|
from kmip.core import enums
|
|
from kmip.core.enums import AttributeType
|
|
from kmip.core.enums import Tags
|
|
from kmip.core.enums import Types
|
|
from kmip.core.enums import CredentialType
|
|
from kmip.core.enums import RevocationReasonCode as RevocationReasonCodeEnum
|
|
|
|
from kmip.core.errors import ErrorStrings
|
|
from kmip.core.misc import KeyFormatType
|
|
|
|
from kmip.core.primitives import Struct
|
|
from kmip.core.primitives import TextString
|
|
from kmip.core.primitives import ByteString
|
|
from kmip.core.primitives import Integer
|
|
from kmip.core.primitives import Enumeration
|
|
|
|
from kmip.core.utils import BytearrayStream
|
|
|
|
|
|
# 2.1
|
|
# 2.1.1
|
|
class Attribute(Struct):
|
|
|
|
class AttributeName(TextString):
|
|
|
|
def __init__(self, value=None):
|
|
super(Attribute.AttributeName, self).__init__(
|
|
value, Tags.ATTRIBUTE_NAME)
|
|
|
|
def __eq__(self, other):
|
|
if isinstance(other, Attribute.AttributeName):
|
|
if self.value != other.value:
|
|
return False
|
|
else:
|
|
return True
|
|
else:
|
|
NotImplemented
|
|
|
|
def __ne__(self, other):
|
|
if isinstance(other, Attribute.AttributeName):
|
|
return not (self == other)
|
|
else:
|
|
return NotImplemented
|
|
|
|
class AttributeIndex(Integer):
|
|
|
|
def __init__(self, value=None):
|
|
super(Attribute.AttributeIndex, self).__init__(
|
|
value, Tags.ATTRIBUTE_INDEX)
|
|
|
|
def __init__(self,
|
|
attribute_name=None,
|
|
attribute_index=None,
|
|
attribute_value=None):
|
|
super(Attribute, self).__init__(tag=Tags.ATTRIBUTE)
|
|
|
|
self.value_factory = AttributeValueFactory()
|
|
|
|
self.attribute_name = attribute_name
|
|
self.attribute_index = attribute_index
|
|
self.attribute_value = attribute_value
|
|
|
|
if attribute_value is not None:
|
|
attribute_value.tag = Tags.ATTRIBUTE_VALUE
|
|
|
|
def read(self, istream):
|
|
super(Attribute, self).read(istream)
|
|
tstream = BytearrayStream(istream.read(self.length))
|
|
|
|
# Read the name of the attribute
|
|
self.attribute_name = Attribute.AttributeName()
|
|
self.attribute_name.read(tstream)
|
|
|
|
# Read the attribute index if it is next
|
|
if self.is_tag_next(Tags.ATTRIBUTE_INDEX, tstream):
|
|
self.attribute_index = Attribute.AttributeIndex()
|
|
self.attribute_index.read(tstream)
|
|
|
|
# Lookup the attribute class that belongs to the attribute name
|
|
name = self.attribute_name.value
|
|
enum_name = name.replace('.', '_').replace(' ', '_').upper()
|
|
enum_type = None
|
|
|
|
try:
|
|
enum_type = AttributeType[enum_name]
|
|
except KeyError:
|
|
# Likely custom attribute, pass raw name string as attribute type
|
|
enum_type = name
|
|
|
|
value = self.value_factory.create_attribute_value(enum_type, None)
|
|
self.attribute_value = value
|
|
self.attribute_value.tag = Tags.ATTRIBUTE_VALUE
|
|
self.attribute_value.read(tstream)
|
|
|
|
self.is_oversized(tstream)
|
|
|
|
def write(self, ostream):
|
|
tstream = BytearrayStream()
|
|
|
|
self.attribute_name.write(tstream)
|
|
if self.attribute_index is not None:
|
|
self.attribute_index.write(tstream)
|
|
self.attribute_value.write(tstream)
|
|
|
|
# Write the length and value of the attribute
|
|
self.length = tstream.length()
|
|
super(Attribute, self).write(ostream)
|
|
ostream.write(tstream.buffer)
|
|
|
|
def __repr__(self):
|
|
attribute_name = "attribute_name={0}".format(repr(self.attribute_name))
|
|
attribute_index = "attribute_index={0}".format(
|
|
repr(self.attribute_index)
|
|
)
|
|
attribute_value = "attribute_value={0}".format(
|
|
repr(self.attribute_value)
|
|
)
|
|
return "Attribute({0}, {1}, {2})".format(
|
|
attribute_name,
|
|
attribute_index,
|
|
attribute_value
|
|
)
|
|
|
|
def __str__(self):
|
|
return str({
|
|
'attribute_name': str(self.attribute_name),
|
|
'attribute_index': str(self.attribute_index),
|
|
'attribute_value': str(self.attribute_value)
|
|
})
|
|
|
|
def __eq__(self, other):
|
|
if isinstance(other, Attribute):
|
|
if self.attribute_name != other.attribute_name:
|
|
return False
|
|
elif self.attribute_index != other.attribute_index:
|
|
return False
|
|
elif self.attribute_value != other.attribute_value:
|
|
return False
|
|
else:
|
|
return True
|
|
else:
|
|
return NotImplemented
|
|
|
|
def __ne__(self, other):
|
|
if isinstance(other, Attribute):
|
|
return not self.__eq__(other)
|
|
else:
|
|
return NotImplemented
|
|
|
|
|
|
# 2.1.2
|
|
class Credential(Struct):
|
|
|
|
class CredentialType(Enumeration):
|
|
|
|
def __init__(self, value=None):
|
|
super(Credential.CredentialType, self).__init__(
|
|
CredentialType, value, Tags.CREDENTIAL_TYPE)
|
|
|
|
class UsernamePasswordCredential(Struct):
|
|
|
|
class Username(TextString):
|
|
def __init__(self, value=None):
|
|
super(Credential.UsernamePasswordCredential.Username,
|
|
self).__init__(
|
|
value, Tags.USERNAME)
|
|
|
|
class Password(TextString):
|
|
def __init__(self, value=None):
|
|
super(Credential.UsernamePasswordCredential.Password,
|
|
self).__init__(
|
|
value, Tags.PASSWORD)
|
|
|
|
def __init__(self, username=None, password=None):
|
|
super(Credential.UsernamePasswordCredential, self).__init__(
|
|
tag=Tags.CREDENTIAL_VALUE)
|
|
self.username = username
|
|
self.password = password
|
|
self.validate()
|
|
|
|
def read(self, istream):
|
|
super(Credential.UsernamePasswordCredential, self).read(istream)
|
|
tstream = BytearrayStream(istream.read(self.length))
|
|
|
|
# Read the username of the credential
|
|
self.username = self.Username()
|
|
self.username.read(tstream)
|
|
|
|
# Read the password if it is next
|
|
if self.is_tag_next(Tags.PASSWORD, tstream):
|
|
self.password = self.Password()
|
|
self.password.read(tstream)
|
|
|
|
self.is_oversized(tstream)
|
|
self.validate()
|
|
|
|
def write(self, ostream):
|
|
tstream = BytearrayStream()
|
|
|
|
self.username.write(tstream)
|
|
if self.password is not None:
|
|
self.password.write(tstream)
|
|
|
|
# Write the length and value of the credential
|
|
self.length = tstream.length()
|
|
super(Credential.UsernamePasswordCredential, self).write(ostream)
|
|
ostream.write(tstream.buffer)
|
|
|
|
def validate(self):
|
|
pass
|
|
|
|
class DeviceCredential(Struct):
|
|
|
|
class DeviceSerialNumber(TextString):
|
|
|
|
def __init__(self, value=None):
|
|
super(Credential.DeviceCredential.DeviceSerialNumber, self).\
|
|
__init__(value, Tags.DEVICE_SERIAL_NUMBER)
|
|
|
|
class Password(TextString):
|
|
|
|
def __init__(self, value=None):
|
|
super(Credential.DeviceCredential.Password, self).\
|
|
__init__(value, Tags.PASSWORD)
|
|
|
|
class DeviceIdentifier(TextString):
|
|
|
|
def __init__(self, value=None):
|
|
super(Credential.DeviceCredential.DeviceIdentifier, self).\
|
|
__init__(value, Tags.DEVICE_IDENTIFIER)
|
|
|
|
class NetworkIdentifier(TextString):
|
|
|
|
def __init__(self, value=None):
|
|
super(Credential.DeviceCredential.NetworkIdentifier, self).\
|
|
__init__(value, Tags.NETWORK_IDENTIFIER)
|
|
|
|
class MachineIdentifier(TextString):
|
|
|
|
def __init__(self, value=None):
|
|
super(Credential.DeviceCredential.MachineIdentifier, self).\
|
|
__init__(value, Tags.MACHINE_IDENTIFIER)
|
|
|
|
class MediaIdentifier(TextString):
|
|
|
|
def __init__(self, value=None):
|
|
super(Credential.DeviceCredential.MediaIdentifier, self).\
|
|
__init__(value, Tags.MEDIA_IDENTIFIER)
|
|
|
|
def __init__(self,
|
|
device_serial_number=None,
|
|
password=None,
|
|
device_identifier=None,
|
|
network_identifier=None,
|
|
machine_identifier=None,
|
|
media_identifier=None):
|
|
super(Credential.DeviceCredential, self).__init__(
|
|
tag=Tags.CREDENTIAL_VALUE)
|
|
self.device_serial_number = device_serial_number
|
|
self.password = password
|
|
self.device_identifier = device_identifier
|
|
self.network_identifier = network_identifier
|
|
self.machine_identifier = machine_identifier
|
|
self.media_identifier = media_identifier
|
|
|
|
def read(self, istream):
|
|
super(Credential.DeviceCredential, self).read(istream)
|
|
tstream = BytearrayStream(istream.read(self.length))
|
|
|
|
# Read the password if it is next
|
|
if self.is_tag_next(Tags.DEVICE_SERIAL_NUMBER, tstream):
|
|
self.device_serial_number = self.DeviceSerialNumber()
|
|
self.device_serial_number.read(tstream)
|
|
|
|
# Read the password if it is next
|
|
if self.is_tag_next(Tags.PASSWORD, tstream):
|
|
self.password = self.Password()
|
|
self.password.read(tstream)
|
|
|
|
# Read the password if it is next
|
|
if self.is_tag_next(Tags.DEVICE_IDENTIFIER, tstream):
|
|
self.device_identifier = self.DeviceIdentifier()
|
|
self.device_identifier.read(tstream)
|
|
|
|
# Read the password if it is next
|
|
if self.is_tag_next(Tags.NETWORK_IDENTIFIER, tstream):
|
|
self.network_identifier = self.NetworkIdentifier()
|
|
self.network_identifier.read(tstream)
|
|
|
|
# Read the password if it is next
|
|
if self.is_tag_next(Tags.MACHINE_IDENTIFIER, tstream):
|
|
self.machine_identifier = self.MachineIdentifier()
|
|
self.machine_identifier.read(tstream)
|
|
|
|
# Read the password if it is next
|
|
if self.is_tag_next(Tags.MEDIA_IDENTIFIER, tstream):
|
|
self.media_identifier = self.MediaIdentifier()
|
|
self.media_identifier.read(tstream)
|
|
|
|
self.is_oversized(tstream)
|
|
self.validate()
|
|
|
|
def write(self, ostream):
|
|
tstream = BytearrayStream()
|
|
|
|
if self.device_serial_number is not None:
|
|
self.device_serial_number.write(tstream)
|
|
if self.password is not None:
|
|
self.password.write(tstream)
|
|
if self.device_identifier is not None:
|
|
self.device_identifier.write(tstream)
|
|
if self.network_identifier is not None:
|
|
self.network_identifier.write(tstream)
|
|
if self.machine_identifier is not None:
|
|
self.machine_identifier.write(tstream)
|
|
if self.media_identifier is not None:
|
|
self.media_identifier.write(tstream)
|
|
|
|
# Write the length and value of the credential
|
|
self.length = tstream.length()
|
|
super(Credential.DeviceCredential, self).write(ostream)
|
|
ostream.write(tstream.buffer)
|
|
|
|
def validate(self):
|
|
pass
|
|
|
|
def __init__(self, credential_type=None, credential_value=None):
|
|
super(Credential, self).__init__(tag=Tags.CREDENTIAL)
|
|
self.credential_type = credential_type
|
|
self.credential_value = credential_value
|
|
|
|
def read(self, istream):
|
|
super(Credential, self).read(istream)
|
|
tstream = BytearrayStream(istream.read(self.length))
|
|
|
|
# Read the type of the credential
|
|
self.credential_type = self.CredentialType()
|
|
self.credential_type.read(tstream)
|
|
|
|
# Use the type to determine what credential value to read
|
|
if self.credential_type.value is CredentialType.USERNAME_AND_PASSWORD:
|
|
self.credential_value = self.UsernamePasswordCredential()
|
|
elif self.credential_type.value is CredentialType.DEVICE:
|
|
self.credential_value = self.DeviceCredential()
|
|
else:
|
|
# TODO (peter-hamilton) Use more descriptive error here
|
|
raise NotImplementedError()
|
|
self.credential_value.read(tstream)
|
|
|
|
self.is_oversized(tstream)
|
|
self.validate()
|
|
|
|
def write(self, ostream):
|
|
tstream = BytearrayStream()
|
|
|
|
self.credential_type.write(tstream)
|
|
self.credential_value.write(tstream)
|
|
|
|
# Write the length and value of the credential
|
|
self.length = tstream.length()
|
|
super(Credential, self).write(ostream)
|
|
ostream.write(tstream.buffer)
|
|
|
|
def validate(self):
|
|
pass
|
|
|
|
|
|
# 2.1.3
|
|
class KeyBlock(Struct):
|
|
|
|
class KeyCompressionType(Enumeration):
|
|
|
|
def __init__(self, value=None):
|
|
super(KeyBlock.KeyCompressionType, self).__init__(
|
|
enums.KeyCompressionType, value, Tags.KEY_COMPRESSION_TYPE)
|
|
|
|
def __init__(self,
|
|
key_format_type=None,
|
|
key_compression_type=None,
|
|
key_value=None,
|
|
cryptographic_algorithm=None,
|
|
cryptographic_length=None,
|
|
key_wrapping_data=None):
|
|
super(KeyBlock, self).__init__(Tags.KEY_BLOCK)
|
|
self.key_format_type = key_format_type
|
|
self.key_compression_type = key_compression_type
|
|
self.key_value = key_value
|
|
self.cryptographic_algorithm = cryptographic_algorithm
|
|
self.cryptographic_length = cryptographic_length
|
|
self.key_wrapping_data = key_wrapping_data
|
|
self.validate()
|
|
|
|
def read(self, istream):
|
|
super(KeyBlock, self).read(istream)
|
|
tstream = BytearrayStream(istream.read(self.length))
|
|
|
|
self.key_format_type = KeyFormatType()
|
|
self.key_format_type.read(tstream)
|
|
|
|
if self.is_tag_next(Tags.KEY_COMPRESSION_TYPE, tstream):
|
|
self.key_compression_type = KeyBlock.KeyCompressionType()
|
|
self.key_compression_type.read(tstream)
|
|
|
|
self.key_value = KeyValue()
|
|
self.key_value.read(tstream)
|
|
|
|
if self.is_tag_next(Tags.CRYPTOGRAPHIC_ALGORITHM, tstream):
|
|
self.cryptographic_algorithm = attributes.CryptographicAlgorithm()
|
|
self.cryptographic_algorithm.read(tstream)
|
|
|
|
if self.is_tag_next(Tags.CRYPTOGRAPHIC_LENGTH, tstream):
|
|
self.cryptographic_length = attributes.CryptographicLength()
|
|
self.cryptographic_length.read(tstream)
|
|
|
|
if self.is_tag_next(Tags.KEY_WRAPPING_DATA, tstream):
|
|
self.key_wrapping_data = KeyWrappingData()
|
|
self.key_wrapping_data.read(tstream)
|
|
|
|
self.is_oversized(tstream)
|
|
self.validate()
|
|
|
|
def write(self, ostream):
|
|
tstream = BytearrayStream()
|
|
|
|
self.key_format_type.write(tstream)
|
|
|
|
if self.key_compression_type is not None:
|
|
self.key_compression_type.write(tstream)
|
|
|
|
self.key_value.write(tstream)
|
|
|
|
if self.cryptographic_algorithm is not None:
|
|
self.cryptographic_algorithm.write(tstream)
|
|
if self.cryptographic_length is not None:
|
|
self.cryptographic_length.write(tstream)
|
|
if self.key_wrapping_data is not None:
|
|
self.key_wrapping_data.write(tstream)
|
|
|
|
# Write the length and value of the credential
|
|
self.length = tstream.length()
|
|
super(KeyBlock, self).write(ostream)
|
|
ostream.write(tstream.buffer)
|
|
|
|
def validate(self):
|
|
self.__validate()
|
|
|
|
def __validate(self):
|
|
if self.key_format_type is not None:
|
|
if type(self.key_format_type) is not KeyFormatType:
|
|
member = 'KeyBlock.key_format_type'
|
|
exp_type = KeyFormatType
|
|
rcv_type = type(self.key_format_type)
|
|
msg = ErrorStrings.BAD_EXP_RECV.format(member, 'type',
|
|
exp_type, rcv_type)
|
|
raise TypeError(msg)
|
|
|
|
|
|
# 2.1.4
|
|
class KeyMaterial(ByteString):
|
|
|
|
def __init__(self, value=None):
|
|
super(KeyMaterial, self).__init__(value, Tags.KEY_MATERIAL)
|
|
|
|
|
|
# TODO (peter-hamilton) Get rid of this and replace with a KeyMaterial factory.
|
|
class KeyMaterialStruct(Struct):
|
|
|
|
def __init__(self):
|
|
super(KeyMaterialStruct, self).__init__(Tags.KEY_MATERIAL)
|
|
|
|
self.data = BytearrayStream()
|
|
|
|
self.validate()
|
|
|
|
def read(self, istream):
|
|
super(KeyMaterialStruct, self).read(istream)
|
|
tstream = BytearrayStream(istream.read(self.length))
|
|
|
|
self.data = BytearrayStream(tstream.read())
|
|
|
|
self.is_oversized(tstream)
|
|
self.validate()
|
|
|
|
def write(self, ostream):
|
|
tstream = BytearrayStream()
|
|
tstream.write(self.data.buffer)
|
|
|
|
self.length = tstream.length()
|
|
super(KeyMaterialStruct, self).write(ostream)
|
|
ostream.write(tstream.buffer)
|
|
|
|
def validate(self):
|
|
self.__validate()
|
|
|
|
def __validate(self):
|
|
# NOTE (peter-hamilton): Intentional pass, no way to validate data.
|
|
pass
|
|
|
|
|
|
class KeyValue(Struct):
|
|
|
|
def __init__(self,
|
|
key_material=None,
|
|
attributes=None):
|
|
super(KeyValue, self).__init__(Tags.KEY_VALUE)
|
|
|
|
if key_material is None:
|
|
self.key_material = KeyMaterial()
|
|
else:
|
|
self.key_material = key_material
|
|
|
|
if attributes is None:
|
|
self.attributes = list()
|
|
else:
|
|
self.attributes = attributes
|
|
|
|
self.validate()
|
|
|
|
def read(self, istream):
|
|
super(KeyValue, self).read(istream)
|
|
tstream = BytearrayStream(istream.read(self.length))
|
|
|
|
# TODO (peter-hamilton) Replace this with a KeyMaterial factory.
|
|
if self.is_type_next(Types.STRUCTURE, tstream):
|
|
self.key_material = KeyMaterialStruct()
|
|
self.key_material.read(tstream)
|
|
else:
|
|
self.key_material = KeyMaterial()
|
|
self.key_material.read(tstream)
|
|
|
|
while self.is_tag_next(Tags.ATTRIBUTE, tstream):
|
|
attribute = Attribute()
|
|
attribute.read(tstream)
|
|
self.attributes.append(attribute)
|
|
|
|
self.is_oversized(tstream)
|
|
self.validate()
|
|
|
|
def write(self, ostream):
|
|
tstream = BytearrayStream()
|
|
|
|
self.key_material.write(tstream)
|
|
|
|
for attribute in self.attributes:
|
|
attribute.write(tstream)
|
|
|
|
self.length = tstream.length()
|
|
super(KeyValue, self).write(ostream)
|
|
ostream.write(tstream.buffer)
|
|
|
|
def validate(self):
|
|
self.__validate()
|
|
|
|
def __validate(self):
|
|
# TODO (peter-hamilton) Replace with check against KeyMaterial factory.
|
|
if not isinstance(self.key_material, KeyMaterial):
|
|
msg = "invalid key material"
|
|
msg += "; expected {0}, received {1}".format(
|
|
KeyMaterial, self.key_material)
|
|
raise TypeError(msg)
|
|
|
|
if isinstance(self.attributes, list):
|
|
for i in xrange(len(self.attributes)):
|
|
attribute = self.attributes[i]
|
|
if not isinstance(attribute, Attribute):
|
|
msg = "invalid attribute ({0} in list)".format(i)
|
|
msg += "; expected {0}, received {1}".format(
|
|
Attribute, attribute)
|
|
raise TypeError(msg)
|
|
else:
|
|
msg = "invalid attributes list"
|
|
msg += "; expected {0}, received {1}".format(
|
|
list, self.attributes)
|
|
raise TypeError(msg)
|
|
|
|
|
|
# 2.1.5
|
|
class WrappingMethod(Enumeration):
|
|
|
|
def __init__(self, value=None):
|
|
super(WrappingMethod, self).__init__(
|
|
enums.WrappingMethod, value, Tags.WRAPPING_METHOD)
|
|
|
|
|
|
class EncodingOption(Enumeration):
|
|
|
|
def __init__(self, value=None):
|
|
super(EncodingOption, self).__init__(
|
|
enums.EncodingOption, value, Tags.ENCODING_OPTION)
|
|
|
|
|
|
class KeyInformation(Struct):
|
|
|
|
def __init__(self,
|
|
unique_identifier=None,
|
|
cryptographic_parameters=None,
|
|
tag=Tags.ENCRYPTION_KEY_INFORMATION):
|
|
super(KeyInformation, self).__init__(tag=tag)
|
|
self.unique_identifier = unique_identifier
|
|
self.cryptographic_parameters = cryptographic_parameters
|
|
self.validate()
|
|
|
|
def read(self, istream):
|
|
super(KeyInformation, self).read(istream)
|
|
tstream = BytearrayStream(istream.read(self.length))
|
|
|
|
self.unique_identifier = attributes.UniqueIdentifier()
|
|
self.unique_identifier.read(tstream)
|
|
|
|
if self.is_tag_next(Tags.CRYPTOGRAPHIC_PARAMETERS, tstream):
|
|
self.cryptographic_parameters = CryptographicParameters()
|
|
self.cryptographic_parameters.read(tstream)
|
|
|
|
self.is_oversized(tstream)
|
|
self.validate()
|
|
|
|
def write(self, ostream):
|
|
tstream = BytearrayStream()
|
|
|
|
self.unique_identifier.write(tstream)
|
|
|
|
if self.cryptographic_parameters is not None:
|
|
self.cryptographic_parameters.write(tstream)
|
|
|
|
# Write the length and value of the template attribute
|
|
self.length = tstream.length()
|
|
super(KeyInformation, self).write(ostream)
|
|
ostream.write(tstream.buffer)
|
|
|
|
def validate(self):
|
|
self.__validate()
|
|
|
|
def __validate(self):
|
|
# TODO (peter-hamilton) Finish implementation.
|
|
pass
|
|
|
|
|
|
class EncryptionKeyInformation(KeyInformation):
|
|
|
|
def __init__(self,
|
|
unique_identifier=None,
|
|
cryptographic_parameters=None,
|
|
tag=Tags.ENCRYPTION_KEY_INFORMATION):
|
|
super(EncryptionKeyInformation, self).__init__(
|
|
unique_identifier, cryptographic_parameters, tag)
|
|
|
|
def validate(self):
|
|
self.__validate()
|
|
|
|
def __validate(self):
|
|
# TODO (peter-hamilton) Finish implementation.
|
|
pass
|
|
|
|
|
|
class MACSignatureKeyInformation(KeyInformation):
|
|
|
|
def __init__(self,
|
|
unique_identifier=None,
|
|
cryptographic_parameters=None,
|
|
tag=Tags.MAC_SIGNATURE_KEY_INFORMATION):
|
|
super(MACSignatureKeyInformation, self).__init__(
|
|
unique_identifier, cryptographic_parameters, tag)
|
|
|
|
def validate(self):
|
|
self.__validate()
|
|
|
|
def __validate(self):
|
|
# TODO (peter-hamilton) Finish implementation.
|
|
pass
|
|
|
|
|
|
class KeyWrappingData(Struct):
|
|
|
|
class MACSignature(ByteString):
|
|
|
|
def __init__(self, value=None):
|
|
super(KeyWrappingData.MACSignature, self).__init__(
|
|
value, Tags.MAC_SIGNATURE)
|
|
|
|
class IVCounterNonce(ByteString):
|
|
|
|
def __init__(self, value=None):
|
|
super(KeyWrappingData.IVCounterNonce, self).__init__(
|
|
value, Tags.IV_COUNTER_NONCE)
|
|
|
|
def __init__(self,
|
|
wrapping_method=None,
|
|
encryption_key_information=None,
|
|
mac_signature_key_information=None,
|
|
mac_signature=None,
|
|
iv_counter_nonce=None,
|
|
encoding_option=None):
|
|
super(KeyWrappingData, self).__init__(Tags.KEY_WRAPPING_DATA)
|
|
self.wrapping_method = wrapping_method
|
|
self.encryption_key_information = encryption_key_information
|
|
self.mac_signature_key_information = mac_signature_key_information
|
|
self.mac_signature = mac_signature
|
|
self.iv_counter_nonce = iv_counter_nonce
|
|
self.encoding_option = encoding_option
|
|
self.validate()
|
|
|
|
def read(self, istream):
|
|
super(KeyWrappingData, self).read(istream)
|
|
tstream = BytearrayStream(istream.read(self.length))
|
|
|
|
self.wrapping_method = WrappingMethod()
|
|
self.wrapping_method.read(tstream)
|
|
|
|
if self.is_tag_next(Tags.ENCRYPTION_KEY_INFORMATION, tstream):
|
|
self.encryption_key_information = EncryptionKeyInformation()
|
|
self.encryption_key_information.read(tstream)
|
|
|
|
if self.is_tag_next(Tags.MAC_SIGNATURE_KEY_INFORMATION, tstream):
|
|
self.mac_signature_key_information = MACSignatureKeyInformation()
|
|
self.mac_signature_key_information.read(tstream)
|
|
|
|
if self.is_tag_next(Tags.MAC_SIGNATURE, tstream):
|
|
self.mac_signature = KeyWrappingData.MACSignature()
|
|
self.mac_signature.read(tstream)
|
|
|
|
if self.is_tag_next(Tags.IV_COUNTER_NONCE, tstream):
|
|
self.iv_counter_nonce = KeyWrappingData.IVCounterNonce()
|
|
self.iv_counter_nonce.read(tstream)
|
|
|
|
if self.is_tag_next(Tags.ENCODING_OPTION, tstream):
|
|
self.encoding_option = EncodingOption()
|
|
self.encoding_option.read(tstream)
|
|
|
|
self.is_oversized(tstream)
|
|
self.validate()
|
|
|
|
def write(self, ostream):
|
|
tstream = BytearrayStream()
|
|
|
|
# Write the contents of the key wrapping data
|
|
self.wrapping_method.write(tstream)
|
|
|
|
if self.encryption_key_information is not None:
|
|
self.encryption_key_information.write(tstream)
|
|
if self.mac_signature_key_information is not None:
|
|
self.mac_signature_key_information.write(tstream)
|
|
if self.mac_signature is not None:
|
|
self.mac_signature.write(tstream)
|
|
if self.iv_counter_nonce is not None:
|
|
self.iv_counter_nonce.write(tstream)
|
|
if self.encoding_option is not None:
|
|
self.encoding_option.write(tstream)
|
|
|
|
# Write the length and value of the key wrapping data
|
|
self.length = tstream.length()
|
|
super(KeyWrappingData, self).write(ostream)
|
|
ostream.write(tstream.buffer)
|
|
|
|
def validate(self):
|
|
self.__validate()
|
|
|
|
def __validate(self):
|
|
# TODO (peter-hamilton) Finish implementation
|
|
pass
|
|
|
|
|
|
# 2.1.6
|
|
class KeyWrappingSpecification(Struct):
|
|
|
|
class AttributeName(TextString):
|
|
|
|
def __init__(self, value=None):
|
|
super(KeyWrappingSpecification.AttributeName, self).__init__(
|
|
value, Tags.ATTRIBUTE_NAME)
|
|
|
|
def __init__(self,
|
|
wrapping_method=None,
|
|
encryption_key_information=None,
|
|
mac_signature_key_information=None,
|
|
attribute_name=None,
|
|
encoding_option=None):
|
|
super(KeyWrappingSpecification, self).__init__(
|
|
tag=Tags.KEY_WRAPPING_SPECIFICATION)
|
|
self.wrapping_method = wrapping_method
|
|
self.encryption_key_information = encryption_key_information
|
|
self.mac_signature_key_information = mac_signature_key_information
|
|
self.attribute_name = attribute_name
|
|
self.encoding_option = encoding_option
|
|
|
|
def read(self, istream):
|
|
super(KeyWrappingSpecification, self).read(istream)
|
|
tstream = BytearrayStream(istream.read(self.length))
|
|
|
|
self.wrapping_method = WrappingMethod()
|
|
self.wrapping_method.read(tstream)
|
|
|
|
if self.is_tag_next(Tags.ENCRYPTION_KEY_INFORMATION, tstream):
|
|
self.encryption_key_information = EncryptionKeyInformation()
|
|
self.encryption_key_information.read(tstream)
|
|
|
|
if self.is_tag_next(Tags.MAC_SIGNATURE_KEY_INFORMATION, tstream):
|
|
self.mac_signature_key_information = MACSignatureKeyInformation()
|
|
self.mac_signature_key_information.read(tstream)
|
|
|
|
if self.is_tag_next(Tags.ATTRIBUTE_NAME, tstream):
|
|
self.attribute_name = KeyWrappingSpecification.AttributeName()
|
|
self.attribute_name.read(tstream)
|
|
|
|
if self.is_tag_next(Tags.ENCODING_OPTION, tstream):
|
|
self.encoding_option = EncodingOption()
|
|
self.encoding_option.read(tstream)
|
|
|
|
self.is_oversized(tstream)
|
|
self.validate()
|
|
|
|
def write(self, ostream):
|
|
tstream = BytearrayStream()
|
|
|
|
# Write the contents of the key wrapping data
|
|
self.wrapping_method.write(tstream)
|
|
|
|
if self.encryption_key_information is not None:
|
|
self.encryption_key_information.write(tstream)
|
|
if self.mac_signature_key_information is not None:
|
|
self.mac_signature_key_information.write(tstream)
|
|
if self.attribute_name is not None:
|
|
self.attribute_name.write(tstream)
|
|
if self.encoding_option is not None:
|
|
self.encoding_option.write(tstream)
|
|
|
|
# Write the length and value of the key wrapping data
|
|
self.length = tstream.length()
|
|
super(KeyWrappingSpecification, self).write(ostream)
|
|
ostream.write(tstream.buffer)
|
|
|
|
def validate(self):
|
|
self.__validate()
|
|
|
|
def __validate(self):
|
|
# TODO (peter-hamilton) Finish implementation.
|
|
pass
|
|
|
|
|
|
# 2.1.8
|
|
class TemplateAttribute(Struct):
|
|
|
|
def __init__(self,
|
|
names=None,
|
|
attributes=None,
|
|
tag=Tags.TEMPLATE_ATTRIBUTE):
|
|
super(TemplateAttribute, self).__init__(tag)
|
|
|
|
if names is None:
|
|
self.names = list()
|
|
else:
|
|
self.names = names
|
|
|
|
if attributes is None:
|
|
self.attributes = list()
|
|
else:
|
|
self.attributes = attributes
|
|
|
|
self.validate()
|
|
|
|
def read(self, istream):
|
|
super(TemplateAttribute, self).read(istream)
|
|
tstream = BytearrayStream(istream.read(self.length))
|
|
|
|
self.names = list()
|
|
self.attributes = list()
|
|
|
|
# Read the names of the template attribute, 0 or more
|
|
while self.is_tag_next(Tags.NAME, tstream):
|
|
name = attributes.Name()
|
|
name.read(tstream)
|
|
self.names.append(name)
|
|
|
|
# Read the attributes of the template attribute, 0 or more
|
|
while self.is_tag_next(Tags.ATTRIBUTE, tstream):
|
|
attribute = Attribute()
|
|
attribute.read(tstream)
|
|
self.attributes.append(attribute)
|
|
|
|
self.is_oversized(tstream)
|
|
self.validate()
|
|
|
|
def write(self, ostream):
|
|
tstream = BytearrayStream()
|
|
|
|
# Write the names and attributes of the template attribute
|
|
for name in self.names:
|
|
name.write(tstream)
|
|
for attribute in self.attributes:
|
|
attribute.write(tstream)
|
|
|
|
# Write the length and value of the template attribute
|
|
self.length = tstream.length()
|
|
super(TemplateAttribute, self).write(ostream)
|
|
ostream.write(tstream.buffer)
|
|
|
|
def validate(self):
|
|
self.__validate()
|
|
|
|
def __validate(self):
|
|
# TODO (peter-hamilton) Finish implementation.
|
|
pass
|
|
|
|
def __eq__(self, other):
|
|
if isinstance(other, TemplateAttribute):
|
|
if len(self.names) != len(other.names):
|
|
return False
|
|
if len(self.attributes) != len(other.attributes):
|
|
return False
|
|
|
|
for i in xrange(len(self.names)):
|
|
a = self.names[i]
|
|
b = other.names[i]
|
|
|
|
if a != b:
|
|
return False
|
|
|
|
for i in xrange(len(self.attributes)):
|
|
a = self.attributes[i]
|
|
b = other.attributes[i]
|
|
|
|
if a != b:
|
|
return False
|
|
|
|
return True
|
|
else:
|
|
return NotImplemented
|
|
|
|
def __ne__(self, other):
|
|
if isinstance(other, TemplateAttribute):
|
|
return not (self == other)
|
|
else:
|
|
return NotImplemented
|
|
|
|
|
|
class CommonTemplateAttribute(TemplateAttribute):
|
|
|
|
def __init__(self,
|
|
names=None,
|
|
attributes=None):
|
|
super(CommonTemplateAttribute, self).__init__(
|
|
names, attributes, Tags.COMMON_TEMPLATE_ATTRIBUTE)
|
|
|
|
|
|
class PrivateKeyTemplateAttribute(TemplateAttribute):
|
|
|
|
def __init__(self,
|
|
names=None,
|
|
attributes=None):
|
|
super(PrivateKeyTemplateAttribute, self).__init__(
|
|
names, attributes, Tags.PRIVATE_KEY_TEMPLATE_ATTRIBUTE)
|
|
|
|
|
|
class PublicKeyTemplateAttribute(TemplateAttribute):
|
|
|
|
def __init__(self,
|
|
names=None,
|
|
attributes=None):
|
|
super(PublicKeyTemplateAttribute, self).__init__(
|
|
names, attributes, Tags.PUBLIC_KEY_TEMPLATE_ATTRIBUTE)
|
|
|
|
|
|
# 2.1.9
|
|
class ExtensionName(TextString):
|
|
"""
|
|
The name of an extended Object.
|
|
|
|
A part of ExtensionInformation, specifically identifying an Object that is
|
|
a custom vendor addition to the KMIP specification. See Section 2.1.9 of
|
|
the KMIP 1.1 specification for more information.
|
|
|
|
Attributes:
|
|
value: The string data representing the extension name.
|
|
"""
|
|
def __init__(self, value=''):
|
|
"""
|
|
Construct an ExtensionName object.
|
|
|
|
Args:
|
|
value (str): The string data representing the extension name.
|
|
Optional, defaults to the empty string.
|
|
"""
|
|
super(ExtensionName, self).__init__(value, Tags.EXTENSION_NAME)
|
|
|
|
|
|
class ExtensionTag(Integer):
|
|
"""
|
|
The tag of an extended Object.
|
|
|
|
A part of ExtensionInformation. See Section 2.1.9 of the KMIP 1.1
|
|
specification for more information.
|
|
|
|
Attributes:
|
|
value: The tag number identifying the extended object.
|
|
"""
|
|
def __init__(self, value=0):
|
|
"""
|
|
Construct an ExtensionTag object.
|
|
|
|
Args:
|
|
value (int): A number representing the extension tag. Often
|
|
displayed in hex format. Optional, defaults to 0.
|
|
"""
|
|
super(ExtensionTag, self).__init__(value, Tags.EXTENSION_TAG)
|
|
|
|
|
|
class ExtensionType(Integer):
|
|
"""
|
|
The type of an extended Object.
|
|
|
|
A part of ExtensionInformation, specifically identifying the type of the
|
|
Object in the specification extension. See Section 2.1.9 of the KMIP 1.1
|
|
specification for more information.
|
|
|
|
Attributes:
|
|
value: The type enumeration for the extended object.
|
|
"""
|
|
def __init__(self, value=None):
|
|
"""
|
|
Construct an ExtensionType object.
|
|
|
|
Args:
|
|
value (Types): A number representing a Types enumeration value,
|
|
indicating the type of the extended Object. Optional, defaults
|
|
to None.
|
|
"""
|
|
super(ExtensionType, self).__init__(value, Tags.EXTENSION_TYPE)
|
|
|
|
|
|
class ExtensionInformation(Struct):
|
|
"""
|
|
A structure describing Objects defined in KMIP specification extensions.
|
|
|
|
It is used specifically for Objects with Item Tag values in the Extensions
|
|
range and appears in responses to Query requests for server extension
|
|
information. See Sections 2.1.9 and 4.25 of the KMIP 1.1 specification for
|
|
more information.
|
|
|
|
Attributes:
|
|
extension_name: The name of the extended Object.
|
|
extension_tag: The tag of the extended Object.
|
|
extension_type: The type of the extended Object.
|
|
"""
|
|
def __init__(self, extension_name=None, extension_tag=None,
|
|
extension_type=None):
|
|
"""
|
|
Construct an ExtensionInformation object.
|
|
|
|
Args:
|
|
extension_name (ExtensionName): The name of the extended Object.
|
|
extension_tag (ExtensionTag): The tag of the extended Object.
|
|
extension_type (ExtensionType): The type of the extended Object.
|
|
"""
|
|
super(ExtensionInformation, self).__init__(Tags.EXTENSION_INFORMATION)
|
|
|
|
if extension_name is None:
|
|
self.extension_name = ExtensionName()
|
|
else:
|
|
self.extension_name = extension_name
|
|
|
|
self.extension_tag = extension_tag
|
|
self.extension_type = extension_type
|
|
|
|
self.validate()
|
|
|
|
def read(self, istream):
|
|
"""
|
|
Read the data encoding the ExtensionInformation object and decode it
|
|
into its constituent parts.
|
|
|
|
Args:
|
|
istream (Stream): A data stream containing encoded object data,
|
|
supporting a read method; usually a BytearrayStream object.
|
|
"""
|
|
super(ExtensionInformation, self).read(istream)
|
|
tstream = BytearrayStream(istream.read(self.length))
|
|
|
|
self.extension_name.read(tstream)
|
|
|
|
if self.is_tag_next(Tags.EXTENSION_TAG, tstream):
|
|
self.extension_tag = ExtensionTag()
|
|
self.extension_tag.read(tstream)
|
|
if self.is_tag_next(Tags.EXTENSION_TYPE, tstream):
|
|
self.extension_type = ExtensionType()
|
|
self.extension_type.read(tstream)
|
|
|
|
self.is_oversized(tstream)
|
|
self.validate()
|
|
|
|
def write(self, ostream):
|
|
"""
|
|
Write the data encoding the ExtensionInformation object to a stream.
|
|
|
|
Args:
|
|
ostream (Stream): A data stream in which to encode object data,
|
|
supporting a write method; usually a BytearrayStream object.
|
|
"""
|
|
tstream = BytearrayStream()
|
|
|
|
self.extension_name.write(tstream)
|
|
|
|
if self.extension_tag is not None:
|
|
self.extension_tag.write(tstream)
|
|
if self.extension_type is not None:
|
|
self.extension_type.write(tstream)
|
|
|
|
self.length = tstream.length()
|
|
super(ExtensionInformation, self).write(ostream)
|
|
ostream.write(tstream.buffer)
|
|
|
|
def validate(self):
|
|
"""
|
|
Error check the attributes of the ExtensionInformation object.
|
|
"""
|
|
self.__validate()
|
|
|
|
def __validate(self):
|
|
if not isinstance(self.extension_name, ExtensionName):
|
|
msg = "invalid extension name"
|
|
msg += "; expected {0}, received {1}".format(
|
|
ExtensionName, self.extension_name)
|
|
raise TypeError(msg)
|
|
|
|
if self.extension_tag is not None:
|
|
if not isinstance(self.extension_tag, ExtensionTag):
|
|
msg = "invalid extension tag"
|
|
msg += "; expected {0}, received {1}".format(
|
|
ExtensionTag, self.extension_tag)
|
|
raise TypeError(msg)
|
|
|
|
if self.extension_type is not None:
|
|
if not isinstance(self.extension_type, ExtensionType):
|
|
msg = "invalid extension type"
|
|
msg += "; expected {0}, received {1}".format(
|
|
ExtensionType, self.extension_type)
|
|
raise TypeError(msg)
|
|
|
|
def __eq__(self, other):
|
|
if isinstance(other, ExtensionInformation):
|
|
if self.extension_name != other.extension_name:
|
|
return False
|
|
elif self.extension_tag != other.extension_tag:
|
|
return False
|
|
elif self.extension_type != other.extension_type:
|
|
return False
|
|
else:
|
|
return True
|
|
else:
|
|
return NotImplemented
|
|
|
|
def __ne__(self, other):
|
|
if isinstance(other, ExtensionInformation):
|
|
return not (self == other)
|
|
else:
|
|
return NotImplemented
|
|
|
|
def __repr__(self):
|
|
name = "extension_name={0}".format(repr(self.extension_name))
|
|
tag = "extension_tag={0}".format(repr(self.extension_tag))
|
|
typ = "extension_type={0}".format(repr(self.extension_type))
|
|
return "ExtensionInformation({0}, {1}, {2})".format(name, tag, typ)
|
|
|
|
def __str__(self):
|
|
return repr(self)
|
|
|
|
@classmethod
|
|
def create(cls, extension_name=None, extension_tag=None,
|
|
extension_type=None):
|
|
"""
|
|
Construct an ExtensionInformation object from provided extension
|
|
values.
|
|
|
|
Args:
|
|
extension_name (str): The name of the extension. Optional,
|
|
defaults to None.
|
|
extension_tag (int): The tag number of the extension. Optional,
|
|
defaults to None.
|
|
extension_type (int): The type index of the extension. Optional,
|
|
defaults to None.
|
|
|
|
Returns:
|
|
ExtensionInformation: The newly created set of extension
|
|
information.
|
|
|
|
Example:
|
|
>>> x = ExtensionInformation.create('extension', 1, 1)
|
|
>>> x.extension_name.value
|
|
ExtensionName(value='extension')
|
|
>>> x.extension_tag.value
|
|
ExtensionTag(value=1)
|
|
>>> x.extension_type.value
|
|
ExtensionType(value=1)
|
|
"""
|
|
extension_name = ExtensionName(extension_name)
|
|
extension_tag = ExtensionTag(extension_tag)
|
|
extension_type = ExtensionType(extension_type)
|
|
|
|
return ExtensionInformation(
|
|
extension_name=extension_name,
|
|
extension_tag=extension_tag,
|
|
extension_type=extension_type)
|
|
|
|
|
|
# 2.1.10
|
|
class Data(ByteString):
|
|
|
|
def __init__(self, value=None):
|
|
super(Data, self).__init__(value, Tags.DATA)
|
|
|
|
|
|
# 2.1.13
|
|
class MACData(ByteString):
|
|
|
|
def __init__(self, value=None):
|
|
super(MACData, self).__init__(value, Tags.MAC_DATA)
|
|
|
|
|
|
# 3.31, 9.1.3.2.19
|
|
class RevocationReasonCode(Enumeration):
|
|
|
|
def __init__(self, value=RevocationReasonCodeEnum.UNSPECIFIED):
|
|
super(RevocationReasonCode, self).__init__(
|
|
RevocationReasonCodeEnum, value=value,
|
|
tag=Tags.REVOCATION_REASON_CODE)
|
|
|
|
|
|
# 3.31
|
|
class RevocationReason(Struct):
|
|
"""
|
|
A structure describing the reason for a revocation operation.
|
|
|
|
See Sections 2.1.9 and 4.25 of the KMIP 1.1 specification for
|
|
more information.
|
|
|
|
Attributes:
|
|
code: The revocation reason code enumeration
|
|
message: An optional revocation message
|
|
"""
|
|
|
|
def __init__(self, code=None, message=None):
|
|
"""
|
|
Construct a RevocationReason object.
|
|
|
|
Parameters:
|
|
code(RevocationReasonCode): revocation reason code
|
|
message(string): An optional revocation message
|
|
"""
|
|
super(RevocationReason, self).__init__(tag=Tags.REVOCATION_REASON)
|
|
if code is not None:
|
|
self.revocation_code = RevocationReasonCode(value=code)
|
|
else:
|
|
self.revocation_code = RevocationReasonCode()
|
|
|
|
if message is not None:
|
|
self.revocation_message = TextString(
|
|
value=message,
|
|
tag=Tags.REVOCATION_MESSAGE)
|
|
else:
|
|
self.revocation_message = None
|
|
|
|
self.validate()
|
|
|
|
def read(self, istream):
|
|
"""
|
|
Read the data encoding the RevocationReason object and decode it
|
|
into its constituent parts.
|
|
|
|
Args:
|
|
istream (Stream): A data stream containing encoded object data,
|
|
supporting a read method; usually a BytearrayStream object.
|
|
"""
|
|
super(RevocationReason, self).read(istream)
|
|
tstream = BytearrayStream(istream.read(self.length))
|
|
|
|
self.revocation_code = RevocationReasonCode()
|
|
self.revocation_code.read(tstream)
|
|
|
|
if self.is_tag_next(Tags.REVOCATION_MESSAGE, tstream):
|
|
self.revocation_message = TextString()
|
|
self.revocation_message.read(tstream)
|
|
|
|
self.is_oversized(tstream)
|
|
self.validate()
|
|
|
|
def write(self, ostream):
|
|
"""
|
|
Write the data encoding the RevocationReason object to a stream.
|
|
|
|
Args:
|
|
ostream (Stream): A data stream in which to encode object data,
|
|
supporting a write method; usually a BytearrayStream object.
|
|
"""
|
|
tstream = BytearrayStream()
|
|
|
|
self.revocation_code.write(tstream)
|
|
if self.revocation_message is not None:
|
|
self.revocation_message.write(tstream)
|
|
|
|
# Write the length and value
|
|
self.length = tstream.length()
|
|
super(RevocationReason, self).write(ostream)
|
|
ostream.write(tstream.buffer)
|
|
|
|
def validate(self):
|
|
"""
|
|
validate the RevocationReason object
|
|
"""
|
|
if not isinstance(self.revocation_code, RevocationReasonCode):
|
|
msg = "RevocationReaonCode expected"
|
|
raise TypeError(msg)
|
|
if self.revocation_message is not None:
|
|
if not isinstance(self.revocation_message, TextString):
|
|
msg = "TextString expect"
|
|
raise TypeError(msg)
|