From 0a499b7b1283a4ad7467d8700d1a1ff29b12cd59 Mon Sep 17 00:00:00 2001 From: Peter Date: Fri, 18 Mar 2016 12:46:40 -0400 Subject: [PATCH 1/2] Adding an AttributePolicy system This change adds a policy system that will be used by the KmipEngine to track and organize rules for individual KMIP attributes. Comparison operators for the Integer primitive and ProtocolVersion struct are added to support the AttributePolicy. Tests for all new changes are included. --- kmip/core/messages/contents.py | 52 + kmip/core/primitives.py | 28 +- kmip/services/server/policy.py | 1148 +++++++++++++++++ .../contents/test_protocol_version.py | 72 ++ .../unit/core/primitives/test_integer.py | 156 +++ .../tests/unit/services/server/test_policy.py | 114 ++ 6 files changed, 1568 insertions(+), 2 deletions(-) create mode 100644 kmip/services/server/policy.py create mode 100644 kmip/tests/unit/services/server/test_policy.py diff --git a/kmip/core/messages/contents.py b/kmip/core/messages/contents.py index 661b1ee..9027f7a 100644 --- a/kmip/core/messages/contents.py +++ b/kmip/core/messages/contents.py @@ -118,6 +118,58 @@ class ProtocolVersion(Struct): else: return NotImplemented + def __lt__(self, other): + if isinstance(other, ProtocolVersion): + if self.protocol_version_major < other.protocol_version_major: + return True + elif self.protocol_version_major > other.protocol_version_major: + return False + elif self.protocol_version_minor < other.protocol_version_minor: + return True + else: + return False + else: + return NotImplemented + + def __gt__(self, other): + if isinstance(other, ProtocolVersion): + if self.protocol_version_major > other.protocol_version_major: + return True + elif self.protocol_version_major < other.protocol_version_major: + return False + elif self.protocol_version_minor > other.protocol_version_minor: + return True + else: + return False + else: + return NotImplemented + + def __le__(self, other): + if isinstance(other, ProtocolVersion): + if self.protocol_version_major < other.protocol_version_major: + return True + elif self.protocol_version_major > other.protocol_version_major: + return False + elif self.protocol_version_minor <= other.protocol_version_minor: + return True + else: + return False + else: + return NotImplemented + + def __ge__(self, other): + if isinstance(other, ProtocolVersion): + if self.protocol_version_major > other.protocol_version_major: + return True + elif self.protocol_version_major < other.protocol_version_major: + return False + elif self.protocol_version_minor >= other.protocol_version_minor: + return True + else: + return False + else: + return NotImplemented + def __repr__(self): major = self.protocol_version_major.value minor = self.protocol_version_minor.value diff --git a/kmip/core/primitives.py b/kmip/core/primitives.py index 59ff91e..5b46787 100644 --- a/kmip/core/primitives.py +++ b/kmip/core/primitives.py @@ -231,10 +231,10 @@ class Integer(Base): raise ValueError('integer value less than accepted min') def __repr__(self): - return "{0}(value={1})".format(type(self).__name__, repr(self.value)) + return "{0}(value={1})".format(type(self).__name__, self.value) def __str__(self): - return "{0}".format(repr(self.value)) + return str(self.value) def __eq__(self, other): if isinstance(other, Integer): @@ -248,6 +248,30 @@ class Integer(Base): else: return NotImplemented + def __lt__(self, other): + if isinstance(other, Integer): + return self.value < other.value + else: + return NotImplemented + + def __gt__(self, other): + if isinstance(other, Integer): + return self.value > other.value + else: + return NotImplemented + + def __le__(self, other): + if isinstance(other, Integer): + return self.__eq__(other) or self.__lt__(other) + else: + return NotImplemented + + def __ge__(self, other): + if isinstance(other, Integer): + return self.__eq__(other) or self.__gt__(other) + else: + return NotImplemented + class LongInteger(Base): """ diff --git a/kmip/services/server/policy.py b/kmip/services/server/policy.py new file mode 100644 index 0000000..5c40153 --- /dev/null +++ b/kmip/services/server/policy.py @@ -0,0 +1,1148 @@ +# Copyright (c) 2016 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 kmip.core import enums +from kmip.core.messages import contents + + +class AttributeRuleSet(object): + """ + A set of flags and indicators defining how an attribute may be used. + + Every attribute defined by the KMIP specification comes with a set of + rules defining how and under what conditions the attribute should be + used. This class acts as a basic struct storing those rules. + + Attributes: + always_has_value: A flag defining if the attribute is always set. + initially_set_by: A list of entities that can implicitly set the + attribute. + modifiable_by_server: A flag defining if the server can modify the + attribute. + modifiable_by_client: A flag defining if the client can modify the + attribute. + deletable_by_client: A flag defining if the client can delete the + attribute. + multiple_instances_permitted: A flag defining if the attribute is + multivalued. + implicitly_set_by: A list of operations that can implicitly set the + attribute. + applies_to_object_types: A list of object types that the attribute + is applicable for. + version_added: The KMIP version in which support for the attribute + was added. + version_deprecated: The KMIP version in which support for the + attribute was deprecated. + """ + + def __init__(self, + always_has_value, + initially_set_by, + modifiable_by_server, + modifiable_by_client, + deletable_by_client, + multivalued, + implicitly_set_by, + applies_to_object_types, + version_added, + version_deprecated=None): + """ + Create an AttributeRuleSet. + + Args: + always_has_value (bool): A flag indicating whether or not this + attribute is always set for a managed object. Required. + initially_set_by (list): A list of strings indicating if the + attribute can be initially set by the 'server' and/or the + 'client'. Required. + modifiable_by_server (bool): A flag indicating whether the server + can independently modify the value of the attribute without + any prompting by the client. Required. + modifiable_by_client (bool): A flag indicating whether the client + can modify the value of the attribute. Required. + deletable_by_client (bool): A flag indicating whether the client + can delete the attribute from the managed object. Required. + multivalued (bool): A flag indicating whether or not a managed + object can have multiple instances of the attribute set at + the same time. Required. + implicitly_set_by (list): A list of Operation enumerations + detailing which server operations are allowed to set the + value of the attribute without direct instruction by the + client.Required. + applies_to_object_types (list): A list of ObjectType enumerations + detailing which managed object types the attribute applies to. + version_added (ProtocolVersion): The KMIP version in which support + for the attribute was added. Required. + version_deprecated (ProtocolVersion): The KMIP version in which + support for the attribute was deprecated. Optional, defaults + to None. + """ + self.always_has_value = always_has_value + self.initially_set_by = initially_set_by + self.modifiable_by_server = modifiable_by_server + self.modifiable_by_client = modifiable_by_client + self.deletable_by_client = deletable_by_client + self.multiple_instances_permitted = multivalued + self.implicitly_set_by = implicitly_set_by + self.applies_to_object_types = applies_to_object_types + self.version_added = version_added + self.version_deprecated = version_deprecated + + +class AttributePolicy(object): + """ + A collection of attribute rules and methods to query those rules. + + This policy class allows for the basic storage and retrieval of + attribute metadata. This metadata changes slightly across KMIP versions + and across the object types associated with different attributes. It + includes information on which entities can modify the attributes, which + object types the attributes are applicable to, and more. It is meant to + be used only by the KmipEngine. + + Metadata queries include questions like: + * Is this attribute supported in KMIP 1.0? + * Is this attribute deprecated in KMIP 1.1? + * Is this attribute applicable for the SymmetricKey object type? + * Is this attribute allowed to have multiple values? + """ + + def __init__(self, version): + """ + Create an AttributePolicy. + + Args: + version (ProtocolVersion): The KMIP protocol version under which + this set of attribute policies should be evaluated. Required. + """ + self._version = version + + self._attribute_rule_sets = { + 'Unique Identifier': AttributeRuleSet( + True, + ('server', ), + False, + False, + False, + False, + ( + enums.Operation.CREATE, + enums.Operation.CREATE_KEY_PAIR, + enums.Operation.REGISTER, + enums.Operation.DERIVE_KEY, + enums.Operation.CERTIFY, + enums.Operation.RECERTIFY, + enums.Operation.REKEY, + enums.Operation.REKEY_KEY_PAIR + ), + ( + enums.ObjectType.CERTIFICATE, + enums.ObjectType.SYMMETRIC_KEY, + enums.ObjectType.PUBLIC_KEY, + enums.ObjectType.PRIVATE_KEY, + enums.ObjectType.SPLIT_KEY, + enums.ObjectType.TEMPLATE, + enums.ObjectType.SECRET_DATA, + enums.ObjectType.OPAQUE_DATA + ), + contents.ProtocolVersion.create(1, 0) + ), + 'Name': AttributeRuleSet( + False, + ('client', ), + True, + True, + True, + True, + ( + enums.Operation.RECERTIFY, + enums.Operation.REKEY, + enums.Operation.REKEY_KEY_PAIR + ), + ( + enums.ObjectType.CERTIFICATE, + enums.ObjectType.SYMMETRIC_KEY, + enums.ObjectType.PUBLIC_KEY, + enums.ObjectType.PRIVATE_KEY, + enums.ObjectType.SPLIT_KEY, + enums.ObjectType.TEMPLATE, + enums.ObjectType.SECRET_DATA, + enums.ObjectType.OPAQUE_DATA + ), + contents.ProtocolVersion.create(1, 0) + ), + 'Object Type': AttributeRuleSet( + True, + ('server', ), + False, + False, + False, + False, + ( + enums.Operation.CREATE, + enums.Operation.CREATE_KEY_PAIR, + enums.Operation.REGISTER, + enums.Operation.DERIVE_KEY, + enums.Operation.CERTIFY, + enums.Operation.RECERTIFY, + enums.Operation.REKEY, + enums.Operation.REKEY_KEY_PAIR + ), + ( + enums.ObjectType.CERTIFICATE, + enums.ObjectType.SYMMETRIC_KEY, + enums.ObjectType.PUBLIC_KEY, + enums.ObjectType.PRIVATE_KEY, + enums.ObjectType.SPLIT_KEY, + enums.ObjectType.TEMPLATE, + enums.ObjectType.SECRET_DATA, + enums.ObjectType.OPAQUE_DATA + ), + contents.ProtocolVersion.create(1, 0) + ), + 'Cryptographic Algorithm': AttributeRuleSet( + True, + ('server', ), + False, + False, + False, + False, + ( + enums.Operation.CREATE, + enums.Operation.CREATE_KEY_PAIR, + enums.Operation.REGISTER, + enums.Operation.DERIVE_KEY, + enums.Operation.CERTIFY, + enums.Operation.RECERTIFY, + enums.Operation.REKEY, + enums.Operation.REKEY_KEY_PAIR + ), + ( + enums.ObjectType.CERTIFICATE, + enums.ObjectType.SYMMETRIC_KEY, + enums.ObjectType.PUBLIC_KEY, + enums.ObjectType.PRIVATE_KEY, + enums.ObjectType.SPLIT_KEY, + enums.ObjectType.TEMPLATE + ), + contents.ProtocolVersion.create(1, 0) + ), + 'Cryptographic Length': AttributeRuleSet( + True, + ('server', ), + False, + False, + False, + False, + ( + enums.Operation.CREATE, + enums.Operation.CREATE_KEY_PAIR, + enums.Operation.REGISTER, + enums.Operation.DERIVE_KEY, + enums.Operation.CERTIFY, + enums.Operation.RECERTIFY, + enums.Operation.REKEY, + enums.Operation.REKEY_KEY_PAIR + ), + ( + enums.ObjectType.CERTIFICATE, + enums.ObjectType.SYMMETRIC_KEY, + enums.ObjectType.PUBLIC_KEY, + enums.ObjectType.PRIVATE_KEY, + enums.ObjectType.SPLIT_KEY, + enums.ObjectType.TEMPLATE + ), + contents.ProtocolVersion.create(1, 0) + ), + 'Cryptographic Parameters': AttributeRuleSet( + False, + ('client', ), + False, + True, + True, + True, + ( + enums.Operation.RECERTIFY, + enums.Operation.REKEY, + enums.Operation.REKEY_KEY_PAIR + ), + ( + enums.ObjectType.CERTIFICATE, + enums.ObjectType.SYMMETRIC_KEY, + enums.ObjectType.PUBLIC_KEY, + enums.ObjectType.PRIVATE_KEY, + enums.ObjectType.SPLIT_KEY, + enums.ObjectType.TEMPLATE + ), + contents.ProtocolVersion.create(1, 0) + ), + 'Cryptographic Domain Parameters': AttributeRuleSet( + False, + ('client', ), + False, + False, + False, + False, + ( + enums.Operation.REKEY, + enums.Operation.REKEY_KEY_PAIR + ), + ( + enums.ObjectType.PUBLIC_KEY, + enums.ObjectType.PRIVATE_KEY, + enums.ObjectType.TEMPLATE + ), + contents.ProtocolVersion.create(1, 0) + ), + 'Certificate Type': AttributeRuleSet( + True, + ('server', ), + False, + False, + False, + False, + ( + enums.Operation.REGISTER, + enums.Operation.CERTIFY, + enums.Operation.RECERTIFY + ), + ( + enums.ObjectType.CERTIFICATE, + ), + contents.ProtocolVersion.create(1, 0) + ), + 'Certificate Length': AttributeRuleSet( + True, + ('server', ), + False, + False, + False, + False, + ( + enums.Operation.REGISTER, + enums.Operation.CERTIFY, + enums.Operation.RECERTIFY + ), + ( + enums.ObjectType.CERTIFICATE, + ), + contents.ProtocolVersion.create(1, 1) + ), + 'X.509 Certificate Identifier': AttributeRuleSet( + True, + ('server', ), + False, + False, + False, + False, + ( + enums.Operation.REGISTER, + enums.Operation.CERTIFY, + enums.Operation.RECERTIFY + ), + ( + # TODO (peterhamilton) Enforce only on X.509 certificates + enums.ObjectType.CERTIFICATE, + ), + contents.ProtocolVersion.create(1, 1) + ), + 'X.509 Certificate Subject': AttributeRuleSet( + True, + ('server', ), + False, + False, + False, + False, + ( + enums.Operation.REGISTER, + enums.Operation.CERTIFY, + enums.Operation.RECERTIFY + ), + ( + # TODO (peterhamilton) Enforce only on X.509 certificates + enums.ObjectType.CERTIFICATE, + ), + contents.ProtocolVersion.create(1, 1) + ), + 'X.509 Certificate Issuer': AttributeRuleSet( + True, + ('server', ), + False, + False, + False, + False, + ( + enums.Operation.REGISTER, + enums.Operation.CERTIFY, + enums.Operation.RECERTIFY + ), + ( + # TODO (peterhamilton) Enforce only on X.509 certificates + enums.ObjectType.CERTIFICATE, + ), + contents.ProtocolVersion.create(1, 1) + ), + 'Certificate Identifier': AttributeRuleSet( + True, + ('server', ), + False, + False, + False, + False, + ( + enums.Operation.REGISTER, + enums.Operation.CERTIFY, + enums.Operation.RECERTIFY + ), + ( + enums.ObjectType.CERTIFICATE, + ), + contents.ProtocolVersion.create(1, 0), + contents.ProtocolVersion.create(1, 1) + ), + 'Certificate Subject': AttributeRuleSet( + True, + ('server', ), + False, + False, + False, + False, + ( + enums.Operation.REGISTER, + enums.Operation.CERTIFY, + enums.Operation.RECERTIFY + ), + ( + enums.ObjectType.CERTIFICATE, + ), + contents.ProtocolVersion.create(1, 0), + contents.ProtocolVersion.create(1, 1) + ), + 'Certificate Issuer': AttributeRuleSet( + True, + ('server', ), + False, + False, + False, + False, + ( + enums.Operation.REGISTER, + enums.Operation.CERTIFY, + enums.Operation.RECERTIFY + ), + ( + enums.ObjectType.CERTIFICATE, + ), + contents.ProtocolVersion.create(1, 0), + contents.ProtocolVersion.create(1, 1) + ), + 'Digital Signature Algorithm': AttributeRuleSet( + True, + ('server', ), + False, + False, + False, + # TODO (peterhamilton) Enforce only for X.509 certificates + False, # True for PGP certificates + ( + enums.Operation.REGISTER, + enums.Operation.CERTIFY, + enums.Operation.RECERTIFY + ), + ( + enums.ObjectType.CERTIFICATE, + ), + contents.ProtocolVersion.create(1, 1) + ), + 'Digest': AttributeRuleSet( + True, # If the server has access to the data + ('server', ), + False, + False, + False, + True, + ( + enums.Operation.CREATE, + enums.Operation.CREATE_KEY_PAIR, + enums.Operation.REGISTER, + enums.Operation.DERIVE_KEY, + enums.Operation.CERTIFY, + enums.Operation.RECERTIFY, + enums.Operation.REKEY, + enums.Operation.REKEY_KEY_PAIR + ), + ( + enums.ObjectType.CERTIFICATE, + enums.ObjectType.SYMMETRIC_KEY, + enums.ObjectType.PUBLIC_KEY, + enums.ObjectType.PRIVATE_KEY, + enums.ObjectType.SPLIT_KEY, + enums.ObjectType.SECRET_DATA, + enums.ObjectType.OPAQUE_DATA + ), + contents.ProtocolVersion.create(1, 0) + ), + 'Operation Policy Name': AttributeRuleSet( + False, + ('server', 'client'), + True, + False, + False, + False, + ( + enums.Operation.CREATE, + enums.Operation.CREATE_KEY_PAIR, + enums.Operation.REGISTER, + enums.Operation.DERIVE_KEY, + enums.Operation.CERTIFY, + enums.Operation.RECERTIFY, + enums.Operation.REKEY, + enums.Operation.REKEY_KEY_PAIR + ), + ( + enums.ObjectType.CERTIFICATE, + enums.ObjectType.SYMMETRIC_KEY, + enums.ObjectType.PUBLIC_KEY, + enums.ObjectType.PRIVATE_KEY, + enums.ObjectType.SPLIT_KEY, + enums.ObjectType.TEMPLATE, + enums.ObjectType.SECRET_DATA, + enums.ObjectType.OPAQUE_DATA + ), + contents.ProtocolVersion.create(1, 0) + ), + 'Cryptographic Usage Mask': AttributeRuleSet( + True, + ('server', 'client'), + True, + False, + False, + False, + ( + enums.Operation.CREATE, + enums.Operation.CREATE_KEY_PAIR, + enums.Operation.REGISTER, + enums.Operation.DERIVE_KEY, + enums.Operation.CERTIFY, + enums.Operation.RECERTIFY, + enums.Operation.REKEY, + enums.Operation.REKEY_KEY_PAIR + ), + ( + enums.ObjectType.CERTIFICATE, + enums.ObjectType.SYMMETRIC_KEY, + enums.ObjectType.PUBLIC_KEY, + enums.ObjectType.PRIVATE_KEY, + enums.ObjectType.SPLIT_KEY, + enums.ObjectType.TEMPLATE, + enums.ObjectType.SECRET_DATA + ), + contents.ProtocolVersion.create(1, 0) + ), + 'Lease Time': AttributeRuleSet( + False, + ('server', ), + True, + False, + False, + False, + ( + enums.Operation.CREATE, + enums.Operation.CREATE_KEY_PAIR, + enums.Operation.REGISTER, + enums.Operation.DERIVE_KEY, + enums.Operation.CERTIFY, + enums.Operation.RECERTIFY, + enums.Operation.REKEY, + enums.Operation.REKEY_KEY_PAIR + ), + ( + enums.ObjectType.CERTIFICATE, + enums.ObjectType.SYMMETRIC_KEY, + enums.ObjectType.PUBLIC_KEY, + enums.ObjectType.PRIVATE_KEY, + enums.ObjectType.SPLIT_KEY, + enums.ObjectType.SECRET_DATA + ), + contents.ProtocolVersion.create(1, 0) + ), + 'Usage Limits': AttributeRuleSet( + False, + ('server', 'client'), # Values differ based on source + True, + True, # Conditional on values and operations used + True, # Conditional on operations used + False, + ( + enums.Operation.CREATE, + enums.Operation.CREATE_KEY_PAIR, + enums.Operation.REGISTER, + enums.Operation.DERIVE_KEY, + enums.Operation.REKEY, + enums.Operation.REKEY_KEY_PAIR, + enums.Operation.GET_USAGE_ALLOCATION + ), + ( + enums.ObjectType.SYMMETRIC_KEY, + enums.ObjectType.PUBLIC_KEY, + enums.ObjectType.PRIVATE_KEY, + enums.ObjectType.SPLIT_KEY, + enums.ObjectType.TEMPLATE + ), + contents.ProtocolVersion.create(1, 0) + ), + 'State': AttributeRuleSet( + True, + ('server', ), + True, + False, # Only modifiable by server for certain requests + False, + False, + ( + enums.Operation.CREATE, + enums.Operation.CREATE_KEY_PAIR, + enums.Operation.REGISTER, + enums.Operation.DERIVE_KEY, + enums.Operation.ACTIVATE, + enums.Operation.REVOKE, + enums.Operation.DESTROY, + enums.Operation.CERTIFY, + enums.Operation.RECERTIFY, + enums.Operation.REKEY, + enums.Operation.REKEY_KEY_PAIR + ), + ( + enums.ObjectType.CERTIFICATE, + enums.ObjectType.SYMMETRIC_KEY, + enums.ObjectType.PUBLIC_KEY, + enums.ObjectType.PRIVATE_KEY, + enums.ObjectType.SPLIT_KEY, + enums.ObjectType.SECRET_DATA + ), + contents.ProtocolVersion.create(1, 0) + ), + 'Initial Date': AttributeRuleSet( + True, + ('server', ), + False, + False, + False, + False, + ( + enums.Operation.CREATE, + enums.Operation.CREATE_KEY_PAIR, + enums.Operation.REGISTER, + enums.Operation.DERIVE_KEY, + enums.Operation.CERTIFY, + enums.Operation.RECERTIFY, + enums.Operation.REKEY, + enums.Operation.REKEY_KEY_PAIR + ), + ( + enums.ObjectType.CERTIFICATE, + enums.ObjectType.SYMMETRIC_KEY, + enums.ObjectType.PUBLIC_KEY, + enums.ObjectType.PRIVATE_KEY, + enums.ObjectType.SPLIT_KEY, + enums.ObjectType.TEMPLATE, + enums.ObjectType.SECRET_DATA, + enums.ObjectType.OPAQUE_DATA + ), + contents.ProtocolVersion.create(1, 0) + ), + 'Activation Date': AttributeRuleSet( + False, + ('server', 'client'), + True, # Only while in Pre-Active state + True, # Only while in Pre-Active state + False, + False, + ( + enums.Operation.CREATE, + enums.Operation.CREATE_KEY_PAIR, + enums.Operation.REGISTER, + enums.Operation.DERIVE_KEY, + enums.Operation.ACTIVATE, + enums.Operation.CERTIFY, + enums.Operation.RECERTIFY, + enums.Operation.REKEY, + enums.Operation.REKEY_KEY_PAIR + ), + ( + enums.ObjectType.CERTIFICATE, + enums.ObjectType.SYMMETRIC_KEY, + enums.ObjectType.PUBLIC_KEY, + enums.ObjectType.PRIVATE_KEY, + enums.ObjectType.SPLIT_KEY, + enums.ObjectType.TEMPLATE, + enums.ObjectType.SECRET_DATA + ), + contents.ProtocolVersion.create(1, 0) + ), + 'Process Start Date': AttributeRuleSet( + False, + ('server', 'client'), + True, # Only while in Pre-Active / Active state and more + True, # Only while in Pre-Active / Active state and more + False, + False, + ( + enums.Operation.CREATE, + enums.Operation.REGISTER, + enums.Operation.DERIVE_KEY, + enums.Operation.REKEY + ), + ( + enums.ObjectType.SYMMETRIC_KEY, + # Only SplitKeys of SymmetricKeys + enums.ObjectType.SPLIT_KEY, + enums.ObjectType.TEMPLATE + ), + contents.ProtocolVersion.create(1, 0) + ), + 'Protect Stop Date': AttributeRuleSet( + False, + ('server', 'client'), + True, # Only while in Pre-Active / Active state and more + True, # Only while in Pre-Active / Active state and more + False, + False, + ( + enums.Operation.CREATE, + enums.Operation.REGISTER, + enums.Operation.DERIVE_KEY, + enums.Operation.REKEY + ), + ( + enums.ObjectType.SYMMETRIC_KEY, + # Only SplitKeys of SymmetricKeys + enums.ObjectType.SPLIT_KEY, + enums.ObjectType.TEMPLATE + ), + contents.ProtocolVersion.create(1, 0) + ), + 'Deactivation Date': AttributeRuleSet( + False, + ('server', 'client'), + True, # Only while in Pre-Active / Active state + True, # Only while in Pre-Active / Active state + False, + False, + ( + enums.Operation.CREATE, + enums.Operation.CREATE_KEY_PAIR, + enums.Operation.REGISTER, + enums.Operation.DERIVE_KEY, + enums.Operation.REVOKE, + enums.Operation.CERTIFY, + enums.Operation.RECERTIFY, + enums.Operation.REKEY, + enums.Operation.REKEY_KEY_PAIR + ), + ( + enums.ObjectType.CERTIFICATE, + enums.ObjectType.SYMMETRIC_KEY, + enums.ObjectType.PUBLIC_KEY, + enums.ObjectType.PRIVATE_KEY, + enums.ObjectType.SPLIT_KEY, + enums.ObjectType.TEMPLATE, + enums.ObjectType.SECRET_DATA + ), + contents.ProtocolVersion.create(1, 0) + ), + 'Destroy Date': AttributeRuleSet( + False, + ('server', ), + False, + False, + False, + False, + ( + enums.Operation.DESTROY, + ), + ( + enums.ObjectType.CERTIFICATE, + enums.ObjectType.SYMMETRIC_KEY, + enums.ObjectType.PUBLIC_KEY, + enums.ObjectType.PRIVATE_KEY, + enums.ObjectType.SPLIT_KEY, + enums.ObjectType.SECRET_DATA, + enums.ObjectType.OPAQUE_DATA + ), + contents.ProtocolVersion.create(1, 0) + ), + 'Compromise Occurrence Date': AttributeRuleSet( + False, + ('server', ), + False, + False, + False, + False, + ( + enums.Operation.REVOKE, + ), + ( + enums.ObjectType.CERTIFICATE, + enums.ObjectType.SYMMETRIC_KEY, + enums.ObjectType.PUBLIC_KEY, + enums.ObjectType.PRIVATE_KEY, + enums.ObjectType.SPLIT_KEY, + enums.ObjectType.SECRET_DATA, + enums.ObjectType.OPAQUE_DATA + ), + contents.ProtocolVersion.create(1, 0) + ), + 'Compromise Date': AttributeRuleSet( + False, + ('server', ), + False, + False, + False, + False, + ( + enums.Operation.REVOKE, + ), + ( + enums.ObjectType.CERTIFICATE, + enums.ObjectType.SYMMETRIC_KEY, + enums.ObjectType.PUBLIC_KEY, + enums.ObjectType.PRIVATE_KEY, + enums.ObjectType.SPLIT_KEY, + enums.ObjectType.SECRET_DATA, + enums.ObjectType.OPAQUE_DATA + ), + contents.ProtocolVersion.create(1, 0) + ), + 'Revocation Reason': AttributeRuleSet( + False, + ('server', ), + True, + False, + False, + False, + ( + enums.Operation.REVOKE, + ), + ( + enums.ObjectType.CERTIFICATE, + enums.ObjectType.SYMMETRIC_KEY, + enums.ObjectType.PUBLIC_KEY, + enums.ObjectType.PRIVATE_KEY, + enums.ObjectType.SPLIT_KEY, + enums.ObjectType.SECRET_DATA, + enums.ObjectType.OPAQUE_DATA + ), + contents.ProtocolVersion.create(1, 0) + ), + 'Archive Date': AttributeRuleSet( + False, + ('server', ), + False, + False, + False, + False, + ( + enums.Operation.ARCHIVE, + ), + ( + enums.ObjectType.CERTIFICATE, + enums.ObjectType.SYMMETRIC_KEY, + enums.ObjectType.PUBLIC_KEY, + enums.ObjectType.PRIVATE_KEY, + enums.ObjectType.SPLIT_KEY, + enums.ObjectType.TEMPLATE, + enums.ObjectType.SECRET_DATA, + enums.ObjectType.OPAQUE_DATA + ), + contents.ProtocolVersion.create(1, 0) + ), + 'Object Group': AttributeRuleSet( + False, + ('server', 'client'), + False, + False, + False, + False, + ( + enums.Operation.CREATE, + enums.Operation.CREATE_KEY_PAIR, + enums.Operation.REGISTER, + enums.Operation.DERIVE_KEY, + enums.Operation.CERTIFY, + enums.Operation.RECERTIFY, + enums.Operation.REKEY, + enums.Operation.REKEY_KEY_PAIR + ), + ( + enums.ObjectType.CERTIFICATE, + enums.ObjectType.SYMMETRIC_KEY, + enums.ObjectType.PUBLIC_KEY, + enums.ObjectType.PRIVATE_KEY, + enums.ObjectType.SPLIT_KEY, + enums.ObjectType.TEMPLATE, + enums.ObjectType.SECRET_DATA, + enums.ObjectType.OPAQUE_DATA + ), + contents.ProtocolVersion.create(1, 0) + ), + 'Fresh': AttributeRuleSet( + False, + ('server', 'client'), + True, + False, + False, + False, + ( + enums.Operation.CREATE, + enums.Operation.CREATE_KEY_PAIR, + enums.Operation.REGISTER, + enums.Operation.DERIVE_KEY, + enums.Operation.CERTIFY, + enums.Operation.RECERTIFY, + enums.Operation.REKEY, + enums.Operation.REKEY_KEY_PAIR + ), + ( + enums.ObjectType.CERTIFICATE, + enums.ObjectType.SYMMETRIC_KEY, + enums.ObjectType.PUBLIC_KEY, + enums.ObjectType.PRIVATE_KEY, + enums.ObjectType.SPLIT_KEY, + enums.ObjectType.TEMPLATE, + enums.ObjectType.SECRET_DATA, + enums.ObjectType.OPAQUE_DATA + ), + contents.ProtocolVersion.create(1, 1) + ), + 'Link': AttributeRuleSet( + False, + ('server', ), + True, + True, + True, + True, + ( + enums.Operation.CREATE_KEY_PAIR, + enums.Operation.DERIVE_KEY, + enums.Operation.CERTIFY, + enums.Operation.RECERTIFY, + enums.Operation.REKEY, + enums.Operation.REKEY_KEY_PAIR + ), + ( + enums.ObjectType.CERTIFICATE, + enums.ObjectType.SYMMETRIC_KEY, + enums.ObjectType.PUBLIC_KEY, + enums.ObjectType.PRIVATE_KEY, + enums.ObjectType.SPLIT_KEY, + enums.ObjectType.TEMPLATE, + enums.ObjectType.SECRET_DATA, + enums.ObjectType.OPAQUE_DATA + ), + contents.ProtocolVersion.create(1, 0) + ), + 'Application Specific Information': AttributeRuleSet( + False, + ('server', 'client'), # Only if omitted in client request + True, # Only if attribute omitted in client request + True, + True, + True, + ( + enums.Operation.RECERTIFY, + enums.Operation.REKEY, + enums.Operation.REKEY_KEY_PAIR + ), + ( + enums.ObjectType.CERTIFICATE, + enums.ObjectType.SYMMETRIC_KEY, + enums.ObjectType.PUBLIC_KEY, + enums.ObjectType.PRIVATE_KEY, + enums.ObjectType.SPLIT_KEY, + enums.ObjectType.TEMPLATE, + enums.ObjectType.SECRET_DATA, + enums.ObjectType.OPAQUE_DATA + ), + contents.ProtocolVersion.create(1, 0) + ), + 'Contact Information': AttributeRuleSet( + False, + ('server', 'client'), + True, + True, + True, + False, + ( + enums.Operation.CREATE, + enums.Operation.CREATE_KEY_PAIR, + enums.Operation.REGISTER, + enums.Operation.DERIVE_KEY, + enums.Operation.CERTIFY, + enums.Operation.RECERTIFY, + enums.Operation.REKEY, + enums.Operation.REKEY_KEY_PAIR + ), + ( + enums.ObjectType.CERTIFICATE, + enums.ObjectType.SYMMETRIC_KEY, + enums.ObjectType.PUBLIC_KEY, + enums.ObjectType.PRIVATE_KEY, + enums.ObjectType.SPLIT_KEY, + enums.ObjectType.TEMPLATE, + enums.ObjectType.SECRET_DATA, + enums.ObjectType.OPAQUE_DATA + ), + contents.ProtocolVersion.create(1, 0) + ), + 'Last Change Date': AttributeRuleSet( + True, + ('server', ), + True, + False, + False, + False, + ( + enums.Operation.CREATE, + enums.Operation.CREATE_KEY_PAIR, + enums.Operation.REGISTER, + enums.Operation.DERIVE_KEY, + enums.Operation.ACTIVATE, + enums.Operation.REVOKE, + enums.Operation.DESTROY, + enums.Operation.ARCHIVE, + enums.Operation.RECOVER, + enums.Operation.CERTIFY, + enums.Operation.RECERTIFY, + enums.Operation.REKEY, + enums.Operation.REKEY_KEY_PAIR, + enums.Operation.ADD_ATTRIBUTE, + enums.Operation.MODIFY_ATTRIBUTE, + enums.Operation.DELETE_ATTRIBUTE, + enums.Operation.GET_USAGE_ALLOCATION + ), + ( + enums.ObjectType.CERTIFICATE, + enums.ObjectType.SYMMETRIC_KEY, + enums.ObjectType.PUBLIC_KEY, + enums.ObjectType.PRIVATE_KEY, + enums.ObjectType.SPLIT_KEY, + enums.ObjectType.TEMPLATE, + enums.ObjectType.SECRET_DATA, + enums.ObjectType.OPAQUE_DATA + ), + contents.ProtocolVersion.create(1, 0) + ), + 'Custom Attribute': AttributeRuleSet( + False, + ('server', 'client'), + True, # Only for server-created attributes + True, # Only for client-created attributes + True, # Only for client-created attributes + True, + ( + enums.Operation.CREATE, + enums.Operation.CREATE_KEY_PAIR, + enums.Operation.REGISTER, + enums.Operation.DERIVE_KEY, + enums.Operation.ACTIVATE, + enums.Operation.REVOKE, + enums.Operation.DESTROY, + enums.Operation.CERTIFY, + enums.Operation.RECERTIFY, + enums.Operation.REKEY, + enums.Operation.REKEY_KEY_PAIR + ), + ( + enums.ObjectType.CERTIFICATE, + enums.ObjectType.SYMMETRIC_KEY, + enums.ObjectType.PUBLIC_KEY, + enums.ObjectType.PRIVATE_KEY, + enums.ObjectType.SPLIT_KEY, + enums.ObjectType.TEMPLATE, + enums.ObjectType.SECRET_DATA, + enums.ObjectType.OPAQUE_DATA + ), + contents.ProtocolVersion.create(1, 0) + ), + } + + def is_attribute_supported(self, attribute): + """ + Check if the attribute is supported by the current KMIP version. + + Args: + attribute (string): The name of the attribute + (e.g., 'Cryptographic Algorithm'). Required. + Returns: + bool: True if the attribute is supported by the current KMIP + version. False otherwise. + """ + if attribute not in self._attribute_rule_sets.keys(): + return False + + rule_set = self._attribute_rule_sets.get(attribute) + if self._version >= rule_set.version_added: + return True + else: + return False + + def is_attribute_deprecated(self, attribute): + """ + Check if the attribute is deprecated by the current KMIP version. + + Args: + attribute (string): The name of the attribute + (e.g., 'Unique Identifier'). Required. + """ + rule_set = self._attribute_rule_sets.get(attribute) + if rule_set.version_deprecated: + if self._version >= rule_set.version_deprecated: + return True + else: + return False + else: + return False + + def is_attribute_applicable_to_object_type(self, attribute, object_type): + """ + Check if the attribute is supported by the given object type. + + Args: + attribute (string): The name of the attribute (e.g., 'Name'). + Required. + object_type (ObjectType): An ObjectType enumeration + (e.g., ObjectType.SYMMETRIC_KEY). Required. + Returns: + bool: True if the attribute is applicable to the object type. + False otherwise. + """ + # TODO (peterhamilton) Handle applicability between certificate types + rule_set = self._attribute_rule_sets.get(attribute) + if object_type in rule_set.applies_to_object_types: + return True + else: + return False + + def is_attribute_multivalued(self, attribute): + """ + Check if the attribute is allowed to have multiple instances. + + Args: + attribute (string): The name of the attribute + (e.g., 'State'). Required. + """ + # TODO (peterhamilton) Handle multivalue swap between certificate types + rule_set = self._attribute_rule_sets.get(attribute) + return rule_set.multiple_instances_permitted diff --git a/kmip/tests/unit/core/messages/contents/test_protocol_version.py b/kmip/tests/unit/core/messages/contents/test_protocol_version.py index 9c1425c..e84901c 100644 --- a/kmip/tests/unit/core/messages/contents/test_protocol_version.py +++ b/kmip/tests/unit/core/messages/contents/test_protocol_version.py @@ -169,6 +169,78 @@ class TestProtocolVersion(TestCase): self.assertTrue(a != b) + def test_less_than(self): + """ + Test that the less than operator returns True/False when comparing + two different ProtocolVersions. + """ + a = ProtocolVersion.create(1, 0) + b = ProtocolVersion.create(1, 1) + c = ProtocolVersion.create(2, 0) + d = ProtocolVersion.create(0, 2) + + self.assertTrue(a < b) + self.assertFalse(b < a) + self.assertFalse(a < a) + self.assertTrue(a < c) + self.assertFalse(c < a) + self.assertFalse(c < d) + self.assertTrue(d < c) + + def test_greater_than(self): + """ + Test that the greater than operator returns True/False when + comparing two different ProtocolVersions. + """ + a = ProtocolVersion.create(1, 0) + b = ProtocolVersion.create(1, 1) + c = ProtocolVersion.create(2, 0) + d = ProtocolVersion.create(0, 2) + + self.assertFalse(a > b) + self.assertTrue(b > a) + self.assertFalse(a > a) + self.assertFalse(a > c) + self.assertTrue(c > a) + self.assertTrue(c > d) + self.assertFalse(d > c) + + def test_less_than_or_equal(self): + """ + Test that the less than or equal operator returns True/False when + comparing two different ProtocolVersions. + """ + a = ProtocolVersion.create(1, 0) + b = ProtocolVersion.create(1, 1) + c = ProtocolVersion.create(2, 0) + d = ProtocolVersion.create(0, 2) + + self.assertTrue(a <= b) + self.assertFalse(b <= a) + self.assertTrue(a <= a) + self.assertTrue(a <= c) + self.assertFalse(c <= a) + self.assertFalse(c <= d) + self.assertTrue(d <= c) + + def test_greater_than_or_equal(self): + """ + Test that the greater than or equal operator returns True/False when + comparing two different ProtocolVersions. + """ + a = ProtocolVersion.create(1, 0) + b = ProtocolVersion.create(1, 1) + c = ProtocolVersion.create(2, 0) + d = ProtocolVersion.create(0, 2) + + self.assertFalse(a >= b) + self.assertTrue(b >= a) + self.assertTrue(a >= a) + self.assertFalse(a >= c) + self.assertTrue(c >= a) + self.assertTrue(c >= d) + self.assertFalse(d >= c) + def test_repr(self): a = ProtocolVersion.create(1, 0) diff --git a/kmip/tests/unit/core/primitives/test_integer.py b/kmip/tests/unit/core/primitives/test_integer.py index 9dbbeca..02b05dd 100644 --- a/kmip/tests/unit/core/primitives/test_integer.py +++ b/kmip/tests/unit/core/primitives/test_integer.py @@ -223,3 +223,159 @@ class TestInteger(testtools.TestCase): self.assertEqual(len_exp, len_rcv, self.bad_write.format(len_exp, len_rcv)) self.assertEqual(encoding, result, self.bad_encoding) + + def test_repr(self): + """ + Test that the representation of an Integer is formatted properly. + """ + integer = primitives.Integer() + value = "value={0}".format(integer.value) + self.assertEqual( + "Integer({0})".format(value), repr(integer)) + + def test_str(self): + """ + Test that the string representation of an Integer is formatted + properly. + """ + self.assertEqual("0", str(primitives.Integer())) + + def test_equal_on_equal(self): + """ + Test that the equality operator returns True when comparing two + Integers. + """ + a = primitives.Integer(1) + b = primitives.Integer(1) + + self.assertTrue(a == b) + self.assertTrue(b == a) + + def test_equal_on_equal_and_empty(self): + """ + Test that the equality operator returns True when comparing two + Integers. + """ + a = primitives.Integer() + b = primitives.Integer() + + self.assertTrue(a == b) + self.assertTrue(b == a) + + def test_equal_on_not_equal(self): + """ + Test that the equality operator returns False when comparing two + Integers with different values. + """ + a = primitives.Integer(1) + b = primitives.Integer(2) + + self.assertFalse(a == b) + self.assertFalse(b == a) + + def test_equal_on_type_mismatch(self): + """ + Test that the equality operator returns False when comparing an + Integer to a non-Integer object. + """ + a = primitives.Integer() + 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 Integers with the same values. + """ + a = primitives.Integer(1) + b = primitives.Integer(1) + + self.assertFalse(a != b) + self.assertFalse(b != a) + + def test_not_equal_on_equal_and_empty(self): + """ + Test that the inequality operator returns False when comparing + two Integers. + """ + a = primitives.Integer() + b = primitives.Integer() + + self.assertFalse(a != b) + self.assertFalse(b != a) + + def test_not_equal_on_not_equal(self): + """ + Test that the inequality operator returns True when comparing two + Integers with different values. + """ + a = primitives.Integer(1) + b = primitives.Integer(2) + + 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 an + Integer to a non-Integer object. + """ + a = primitives.Integer() + b = 'invalid' + + self.assertTrue(a != b) + self.assertTrue(b != a) + + def test_less_than(self): + """ + Test that the less than operator returns True/False when comparing + two Integers with different values. + """ + a = primitives.Integer(1) + b = primitives.Integer(2) + + self.assertTrue(a < b) + self.assertFalse(b < a) + self.assertFalse(a < a) + + def test_greater_than(self): + """ + Test that the greater than operator returns True/False when comparing + two Integers with different values. + """ + a = primitives.Integer(1) + b = primitives.Integer(2) + + self.assertFalse(a > b) + self.assertTrue(b > a) + self.assertFalse(b > b) + + def test_less_than_or_equal(self): + """ + Test that the less than or equal operator returns True/False when + comparing two Integers with different values. + """ + a = primitives.Integer(1) + b = primitives.Integer(2) + c = primitives.Integer(1) + + self.assertTrue(a <= b) + self.assertFalse(b <= c) + self.assertTrue(a <= c) + self.assertTrue(a <= a) + + def test_greater_than_or_equal(self): + """ + Test that the greater than or equal operator returns True/False when + comparing two Integers with different values. + """ + a = primitives.Integer(1) + b = primitives.Integer(2) + c = primitives.Integer(1) + + self.assertFalse(a >= b) + self.assertTrue(b >= c) + self.assertTrue(a >= c) + self.assertTrue(a >= a) diff --git a/kmip/tests/unit/services/server/test_policy.py b/kmip/tests/unit/services/server/test_policy.py new file mode 100644 index 0000000..542d99d --- /dev/null +++ b/kmip/tests/unit/services/server/test_policy.py @@ -0,0 +1,114 @@ +# Copyright (c) 2016 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 enums +from kmip.core.messages import contents +from kmip.services.server import policy + + +class TestAttributePolicy(testtools.TestCase): + """ + A test engine for AttributePolicy. + """ + + def setUp(self): + super(TestAttributePolicy, self).setUp() + + def tearDown(self): + super(TestAttributePolicy, self).tearDown() + + def test_init(self): + """ + Test that an AttributePolicy can be built without any errors. + """ + policy.AttributePolicy(contents.ProtocolVersion.create(1, 0)) + + def test_is_attribute_supported(self): + """ + Test that is_attribute_supported returns the expected results in all + cases. + """ + rules = policy.AttributePolicy(contents.ProtocolVersion.create(1, 0)) + attribute_a = 'Unique Identifier' + attribute_b = 'Certificate Length' + attribute_c = 'invalid' + + result = rules.is_attribute_supported(attribute_a) + self.assertTrue(result) + + result = rules.is_attribute_supported(attribute_b) + self.assertFalse(result) + + result = rules.is_attribute_supported(attribute_c) + self.assertFalse(result) + + def test_is_attribute_deprecated(self): + """ + Test that is_attribute_deprecated returns the expected results in all + cases. + """ + rules = policy.AttributePolicy(contents.ProtocolVersion.create(1, 0)) + attribute_a = 'Name' + attribute_b = 'Certificate Subject' + + result = rules.is_attribute_deprecated(attribute_a) + self.assertFalse(result) + + result = rules.is_attribute_deprecated(attribute_b) + self.assertFalse(result) + + rules = policy.AttributePolicy(contents.ProtocolVersion.create(1, 1)) + + result = rules.is_attribute_deprecated(attribute_b) + self.assertTrue(result) + + def test_is_attribute_applicable_to_object_type(self): + """ + Test that is_attribute_applicable_to_object_type returns the + expected results in all cases. + """ + rules = policy.AttributePolicy(contents.ProtocolVersion.create(1, 0)) + attribute = 'Cryptographic Algorithm' + object_type_a = enums.ObjectType.SYMMETRIC_KEY + object_type_b = enums.ObjectType.OPAQUE_DATA + + result = rules.is_attribute_applicable_to_object_type( + attribute, + object_type_a + ) + self.assertTrue(result) + + result = rules.is_attribute_applicable_to_object_type( + attribute, + object_type_b + ) + self.assertFalse(result) + + def test_is_attribute_multivalued(self): + """ + Test that is_attribute_multivalued returns the expected results in + all cases. + """ + rules = policy.AttributePolicy(contents.ProtocolVersion.create(1, 0)) + attribute_a = 'Object Type' + attribute_b = 'Link' + + result = rules.is_attribute_multivalued(attribute_a) + self.assertFalse(result) + + result = rules.is_attribute_multivalued(attribute_b) + self.assertTrue(result) From 89cba7382120b93c94f38a4f03ec372f2e41783e Mon Sep 17 00:00:00 2001 From: Peter Date: Mon, 21 Mar 2016 11:22:47 -0400 Subject: [PATCH 2/2] Adding KmipEngine support for Register This change adds support for the Register operation to the KmipEngine. New exceptions and test cases are included. --- kmip/services/server/engine.py | 202 ++++- .../tests/unit/services/server/test_engine.py | 786 +++++++++++++++++- 2 files changed, 981 insertions(+), 7 deletions(-) diff --git a/kmip/services/server/engine.py b/kmip/services/server/engine.py index 3265c57..8b15de6 100644 --- a/kmip/services/server/engine.py +++ b/kmip/services/server/engine.py @@ -14,6 +14,7 @@ # under the License. import logging +import six import sqlalchemy from sqlalchemy.orm import exc @@ -35,12 +36,15 @@ from kmip.core.messages.payloads import destroy from kmip.core.messages.payloads import discover_versions from kmip.core.messages.payloads import get from kmip.core.messages.payloads import query +from kmip.core.messages.payloads import register from kmip.core import misc -from kmip.pie import sqltypes +from kmip.pie import factory from kmip.pie import objects +from kmip.pie import sqltypes +from kmip.services.server import policy from kmip.services.server.crypto import engine @@ -63,6 +67,9 @@ class KmipEngine(object): * Key compression * Key wrapping * Key format conversions + * Registration of empty managed objects (e.g., Private Keys) + * Managed object state tracking + * Managed object usage limit tracking and enforcement """ def __init__(self): @@ -105,6 +112,8 @@ class KmipEngine(object): enums.ObjectType.OPAQUE_DATA: objects.OpaqueObject } + self._attribute_policy = policy.AttributePolicy(self._protocol_version) + def _kmip_version_supported(supported): def decorator(function): def wrapper(self, *args, **kwargs): @@ -133,6 +142,9 @@ class KmipEngine(object): def _set_protocol_version(self, protocol_version): if protocol_version in self._protocol_versions: self._protocol_version = protocol_version + self._attribute_policy = policy.AttributePolicy( + self._protocol_version + ) else: raise exceptions.InvalidMessage( "KMIP {0} is not supported by the server.".format( @@ -479,8 +491,134 @@ class KmipEngine(object): secret_factory = secrets.SecretFactory() return secret_factory.create(object_type, value) + def _process_template_attribute(self, template_attribute): + """ + Given a kmip.core TemplateAttribute object, extract the attribute + value data into a usable dictionary format. + """ + attributes = {} + + if len(template_attribute.names) > 0: + raise exceptions.ItemNotFound( + "Attribute templates are not supported." + ) + + for attribute in template_attribute.attributes: + name = attribute.attribute_name.value + + if not self._attribute_policy.is_attribute_supported(name): + raise exceptions.InvalidField( + "The {0} attribute is unsupported.".format(name) + ) + + if self._attribute_policy.is_attribute_multivalued(name): + values = attributes.get(name, list()) + if (not attribute.attribute_index) and len(values) > 0: + raise exceptions.InvalidField( + "Attribute index missing from multivalued attribute." + ) + + values.append(attribute.attribute_value) + attributes.update([(name, values)]) + else: + if attribute.attribute_index: + if attribute.attribute_index.value != 0: + raise exceptions.InvalidField( + "Non-zero attribute index found for " + "single-valued attribute." + ) + value = attributes.get(name, None) + if value: + raise exceptions.IndexOutOfBounds( + "Cannot set multiple instances of the " + "{0} attribute.".format(name) + ) + else: + attributes.update([(name, attribute.attribute_value)]) + + return attributes + + def _set_attributes_on_managed_object(self, managed_object, attributes): + """ + Given a kmip.pie object and a dictionary of attributes, attempt to set + the attribute values on the object. + """ + for attribute_name, attribute_value in six.iteritems(attributes): + object_type = managed_object._object_type + if self._attribute_policy.is_attribute_applicable_to_object_type( + attribute_name, + object_type): + self._set_attribute_on_managed_object( + managed_object, + (attribute_name, attribute_value) + ) + else: + name = object_type.name + raise exceptions.InvalidField( + "Cannot set {0} attribute on {1} object.".format( + attribute_name, + ''.join([x.capitalize() for x in name.split('_')]) + ) + ) + + def _set_attribute_on_managed_object(self, managed_object, attribute): + """ + Set the attribute value on the kmip.pie managed object. + """ + attribute_name = attribute[0] + attribute_value = attribute[1] + + if self._attribute_policy.is_attribute_multivalued(attribute_name): + if attribute_name == 'Name': + managed_object.names.extend( + [x.name_value.value for x in attribute_value] + ) + for name in managed_object.names: + if managed_object.names.count(name) > 1: + raise exceptions.InvalidField( + "Cannot set duplicate name values." + ) + else: + # TODO (peterhamilton) Remove when all attributes are supported + raise exceptions.InvalidField( + "The {0} attribute is unsupported.".format(attribute_name) + ) + else: + field = None + value = attribute_value.value + + if attribute_name == 'Cryptographic Algorithm': + field = 'cryptographic_algorithm' + elif attribute_name == 'Cryptographic Length': + field = 'cryptographic_length' + elif attribute_name == 'Cryptographic Usage Mask': + field = 'cryptographic_usage_masks' + value = list() + for e in enums.CryptographicUsageMask: + if e.value & attribute_value.value: + value.append(e) + + if field: + existing_value = getattr(managed_object, field) + if existing_value: + if existing_value != value: + raise exceptions.InvalidField( + "Cannot overwrite the {0} attribute.".format( + attribute_name + ) + ) + else: + setattr(managed_object, field, value) + else: + # TODO (peterhamilton) Remove when all attributes are supported + raise exceptions.InvalidField( + "The {0} attribute is unsupported.".format(attribute_name) + ) + def _process_operation(self, operation, payload): - if operation == enums.Operation.GET: + if operation == enums.Operation.REGISTER: + return self._process_register(payload) + elif operation == enums.Operation.GET: return self._process_get(payload) elif operation == enums.Operation.DESTROY: return self._process_destroy(payload) @@ -495,6 +633,65 @@ class KmipEngine(object): ) ) + @_kmip_version_supported('1.0') + def _process_register(self, payload): + self._logger.info("Processing operation: Register") + + object_type = payload.object_type.value + template_attribute = payload.template_attribute + + if self._object_map.get(object_type) is None: + name = object_type.name + raise exceptions.InvalidField( + "The {0} object type is not supported.".format( + ''.join( + [x.capitalize() for x in name.split('_')] + ) + ) + ) + + if payload.secret: + secret = payload.secret + else: + # TODO (peterhamilton) It is possible to register 'empty' secrets + # like Private Keys. For now, that feature is not supported. + raise exceptions.InvalidField( + "Cannot register a secret in absentia." + ) + + object_attributes = {} + if template_attribute: + object_attributes = self._process_template_attribute( + template_attribute + ) + + managed_object_factory = factory.ObjectFactory() + managed_object = managed_object_factory.convert(secret) + managed_object.names = [] + + self._set_attributes_on_managed_object( + managed_object, + object_attributes + ) + + # TODO (peterhamilton) Set additional server-only attributes. + + self._data_session.add(managed_object) + + # NOTE (peterhamilton) SQLAlchemy will *not* assign an ID until + # commit is called. This makes future support for UNDO problematic. + self._data_session.commit() + + response_payload = register.RegisterResponsePayload( + unique_identifier=attributes.UniqueIdentifier( + str(managed_object.unique_identifier) + ) + ) + + self._id_placeholder = str(managed_object.unique_identifier) + + return response_payload + @_kmip_version_supported('1.0') def _process_get(self, payload): self._logger.info("Processing operation: Get") @@ -592,6 +789,7 @@ class KmipEngine(object): if enums.QueryFunction.QUERY_OPERATIONS in queries: operations = list([ + contents.Operation(enums.Operation.REGISTER), contents.Operation(enums.Operation.GET), contents.Operation(enums.Operation.DESTROY), contents.Operation(enums.Operation.QUERY) diff --git a/kmip/tests/unit/services/server/test_engine.py b/kmip/tests/unit/services/server/test_engine.py index a8c3b96..b287c35 100644 --- a/kmip/tests/unit/services/server/test_engine.py +++ b/kmip/tests/unit/services/server/test_engine.py @@ -30,6 +30,8 @@ from kmip.core import misc from kmip.core import objects from kmip.core import secrets +from kmip.core.factories import attributes as factory + from kmip.core.messages import contents from kmip.core.messages import messages @@ -37,6 +39,7 @@ from kmip.core.messages.payloads import destroy from kmip.core.messages.payloads import discover_versions from kmip.core.messages.payloads import get from kmip.core.messages.payloads import query +from kmip.core.messages.payloads import register from kmip.pie import objects as pie_objects from kmip.pie import sqltypes @@ -640,16 +643,19 @@ class TestKmipEngine(testtools.TestCase): e = engine.KmipEngine() e._logger = mock.MagicMock() + e._process_register = mock.MagicMock() e._process_get = mock.MagicMock() e._process_destroy = mock.MagicMock() e._process_query = mock.MagicMock() e._process_discover_versions = mock.MagicMock() + e._process_operation(enums.Operation.REGISTER, None) e._process_operation(enums.Operation.GET, None) e._process_operation(enums.Operation.DESTROY, None) e._process_operation(enums.Operation.QUERY, None) e._process_operation(enums.Operation.DISCOVER_VERSIONS, None) + e._process_register.assert_called_with(None) e._process_get.assert_called_with(None) e._process_destroy.assert_called_with(None) e._process_query.assert_called_with(None) @@ -942,6 +948,634 @@ class TestKmipEngine(testtools.TestCase): *args ) + def test_process_template_attribute(self): + """ + Test that a template attribute structure can be processed correctly. + """ + e = engine.KmipEngine() + e._data_store = self.engine + e._data_store_session_factory = self.session_factory + e._data_session = e._data_store_session_factory() + e._logger = mock.MagicMock() + + attribute_factory = factory.AttributeFactory() + + name = attribute_factory.create_attribute( + enums.AttributeType.NAME, + attributes.Name.create( + 'Test Symmetric Key', + enums.NameType.UNINTERPRETED_TEXT_STRING + ) + ) + algorithm = attribute_factory.create_attribute( + enums.AttributeType.CRYPTOGRAPHIC_ALGORITHM, + enums.CryptographicAlgorithm.AES + ) + length = attribute_factory.create_attribute( + enums.AttributeType.CRYPTOGRAPHIC_LENGTH, + 128 + ) + mask = attribute_factory.create_attribute( + enums.AttributeType.CRYPTOGRAPHIC_USAGE_MASK, + [ + enums.CryptographicUsageMask.ENCRYPT, + enums.CryptographicUsageMask.DECRYPT + ] + ) + template_attribute = objects.TemplateAttribute( + attributes=[name, algorithm, length, mask] + ) + + result = e._process_template_attribute(template_attribute) + + self.assertIsInstance(result, dict) + self.assertEqual(4, len(result.keys())) + self.assertIn('Name', result.keys()) + self.assertIn('Cryptographic Algorithm', result.keys()) + self.assertIn('Cryptographic Length', result.keys()) + self.assertIn('Cryptographic Usage Mask', result.keys()) + + self.assertEqual([name.attribute_value], result.get('Name')) + self.assertEqual( + algorithm.attribute_value, + result.get('Cryptographic Algorithm') + ) + self.assertEqual( + length.attribute_value, + result.get('Cryptographic Length') + ) + self.assertEqual( + mask.attribute_value, + result.get('Cryptographic Usage Mask') + ) + + def test_process_template_attribute_unsupported_features(self): + """ + Test that the right errors are generated when unsupported features + are referenced while processing a template attribute. + """ + e = engine.KmipEngine() + e._data_store = self.engine + e._data_store_session_factory = self.session_factory + e._data_session = e._data_store_session_factory() + e._logger = mock.MagicMock() + + attribute_factory = factory.AttributeFactory() + + # Test that providing template names generates an InvalidField error. + template_attribute = objects.TemplateAttribute( + names=[ + attributes.Name.create( + 'invalid', + enums.NameType.UNINTERPRETED_TEXT_STRING + ) + ] + ) + + args = (template_attribute, ) + regex = "Attribute templates are not supported." + self.assertRaisesRegexp( + exceptions.ItemNotFound, + regex, + e._process_template_attribute, + *args + ) + + # Test that an unrecognized attribute generates an InvalidField error. + name = attribute_factory.create_attribute( + enums.AttributeType.NAME, + attributes.Name.create( + 'Test Symmetric Key', + enums.NameType.UNINTERPRETED_TEXT_STRING + ) + ) + name.attribute_name.value = 'invalid' + template_attribute = objects.TemplateAttribute(attributes=[name]) + + args = (template_attribute, ) + regex = "The invalid attribute is unsupported." + self.assertRaisesRegexp( + exceptions.InvalidField, + regex, + e._process_template_attribute, + *args + ) + + # Test that missing indices generate an InvalidField error. + name_a = attribute_factory.create_attribute( + enums.AttributeType.NAME, + attributes.Name.create( + 'Test Symmetric Key', + enums.NameType.UNINTERPRETED_TEXT_STRING + ) + ) + name_b = attribute_factory.create_attribute( + enums.AttributeType.NAME, + attributes.Name.create( + 'Test Symmetric Key', + enums.NameType.UNINTERPRETED_TEXT_STRING + ) + ) + + template_attribute = objects.TemplateAttribute( + attributes=[name_a, name_b] + ) + + args = (template_attribute, ) + regex = "Attribute index missing from multivalued attribute." + self.assertRaisesRegexp( + exceptions.InvalidField, + regex, + e._process_template_attribute, + *args + ) + + # Test that a non-zero index generates an InvalidField error. + algorithm = attribute_factory.create_attribute( + enums.AttributeType.CRYPTOGRAPHIC_ALGORITHM, + enums.CryptographicAlgorithm.AES, + 1 + ) + template_attribute = objects.TemplateAttribute(attributes=[algorithm]) + + args = (template_attribute, ) + regex = "Non-zero attribute index found for single-valued attribute." + self.assertRaisesRegexp( + exceptions.InvalidField, + regex, + e._process_template_attribute, + *args + ) + + # Test that setting multiple values for a single-value attribute + # generates an InvalidField error. + algorithm_a = attribute_factory.create_attribute( + enums.AttributeType.CRYPTOGRAPHIC_ALGORITHM, + enums.CryptographicAlgorithm.AES + ) + algorithm_b = attribute_factory.create_attribute( + enums.AttributeType.CRYPTOGRAPHIC_ALGORITHM, + enums.CryptographicAlgorithm.TRIPLE_DES + ) + + template_attribute = objects.TemplateAttribute( + attributes=[algorithm_a, algorithm_b] + ) + + args = (template_attribute, ) + regex = ( + "Cannot set multiple instances of the Cryptographic Algorithm " + "attribute." + ) + self.assertRaisesRegexp( + exceptions.IndexOutOfBounds, + regex, + e._process_template_attribute, + *args + ) + + def test_set_attributes_on_managed_object(self): + """ + Test that multiple attributes can be set on a given managed object. + """ + e = engine.KmipEngine() + e._data_store = self.engine + e._data_store_session_factory = self.session_factory + e._data_session = e._data_store_session_factory() + e._logger = mock.MagicMock() + + managed_object = pie_objects.SecretData( + b'', + enums.SecretDataType.PASSWORD + ) + managed_object.names = [] + attribute_factory = factory.AttributeFactory() + + name = attribute_factory.create_attribute( + enums.AttributeType.NAME, + attributes.Name.create( + 'Test Secret Data', + enums.NameType.UNINTERPRETED_TEXT_STRING + ) + ) + mask = attribute_factory.create_attribute( + enums.AttributeType.CRYPTOGRAPHIC_USAGE_MASK, + [ + enums.CryptographicUsageMask.ENCRYPT, + enums.CryptographicUsageMask.DECRYPT + ] + ) + template_attribute = objects.TemplateAttribute( + attributes=[name, mask] + ) + object_attributes = e._process_template_attribute(template_attribute) + + self.assertEqual([], managed_object.names) + self.assertEqual([], managed_object.cryptographic_usage_masks) + + e._set_attributes_on_managed_object( + managed_object, + object_attributes + ) + + self.assertEqual(['Test Secret Data'], managed_object.names) + self.assertEqual( + [ + enums.CryptographicUsageMask.ENCRYPT, + enums.CryptographicUsageMask.DECRYPT + ], + managed_object.cryptographic_usage_masks + ) + + def test_set_attributes_on_managed_object_attribute_mismatch(self): + """ + Test that an InvalidField error is generated when attempting to set + an attribute that is not applicable for a given managed object. + """ + e = engine.KmipEngine() + e._data_store = self.engine + e._data_store_session_factory = self.session_factory + e._data_session = e._data_store_session_factory() + e._logger = mock.MagicMock() + + managed_object = pie_objects.OpaqueObject( + b'', + enums.OpaqueDataType.NONE + ) + attribute_factory = factory.AttributeFactory() + + mask = attribute_factory.create_attribute( + enums.AttributeType.CRYPTOGRAPHIC_USAGE_MASK, + [ + enums.CryptographicUsageMask.ENCRYPT, + enums.CryptographicUsageMask.DECRYPT + ] + ) + template_attribute = objects.TemplateAttribute(attributes=[mask]) + object_attributes = e._process_template_attribute(template_attribute) + + args = (managed_object, object_attributes) + regex = ( + "Cannot set Cryptographic Usage Mask attribute on OpaqueData " + "object." + ) + self.assertRaisesRegexp( + exceptions.InvalidField, + regex, + e._set_attributes_on_managed_object, + *args + ) + + def test_set_attribute_on_managed_object(self): + """ + Test that various attributes can be set correctly on a given + managed object. + """ + e = engine.KmipEngine() + e._data_store = self.engine + e._data_store_session_factory = self.session_factory + e._data_session = e._data_store_session_factory() + e._logger = mock.MagicMock() + + attribute_factory = factory.AttributeFactory() + + name = attribute_factory.create_attribute( + enums.AttributeType.NAME, + attributes.Name.create( + 'Test Symmetric Key', + enums.NameType.UNINTERPRETED_TEXT_STRING + ) + ) + algorithm = attribute_factory.create_attribute( + enums.AttributeType.CRYPTOGRAPHIC_ALGORITHM, + enums.CryptographicAlgorithm.AES + ) + length = attribute_factory.create_attribute( + enums.AttributeType.CRYPTOGRAPHIC_LENGTH, + 0 + ) + mask = attribute_factory.create_attribute( + enums.AttributeType.CRYPTOGRAPHIC_USAGE_MASK, + [ + enums.CryptographicUsageMask.ENCRYPT, + enums.CryptographicUsageMask.DECRYPT + ] + ) + managed_object = pie_objects.SymmetricKey( + enums.CryptographicAlgorithm.AES, + 0, + b'' + ) + managed_object.names = [] + + self.assertEqual([], managed_object.names) + self.assertEqual( + enums.CryptographicAlgorithm.AES, + managed_object.cryptographic_algorithm + ) + self.assertEqual(0, managed_object.cryptographic_length) + self.assertEqual([], managed_object.cryptographic_usage_masks) + + e._set_attribute_on_managed_object( + managed_object, + ('Name', [name.attribute_value]) + ) + + self.assertEqual(['Test Symmetric Key'], managed_object.names) + + e._set_attribute_on_managed_object( + managed_object, + ('Cryptographic Algorithm', algorithm.attribute_value) + ) + + self.assertEqual( + enums.CryptographicAlgorithm.AES, + managed_object.cryptographic_algorithm + ) + + e._set_attribute_on_managed_object( + managed_object, + ('Cryptographic Length', length.attribute_value) + ) + + self.assertEqual(0, managed_object.cryptographic_length) + + e._set_attribute_on_managed_object( + managed_object, + ('Cryptographic Usage Mask', mask.attribute_value) + ) + + self.assertEqual( + [ + enums.CryptographicUsageMask.ENCRYPT, + enums.CryptographicUsageMask.DECRYPT + ], + managed_object.cryptographic_usage_masks + ) + + def test_set_attribute_on_managed_object_unsupported_features(self): + """ + Test that the right errors are generated when unsupported features + are referenced while setting managed object attributes. + """ + e = engine.KmipEngine() + e._data_store = self.engine + e._data_store_session_factory = self.session_factory + e._data_session = e._data_store_session_factory() + e._logger = mock.MagicMock() + + attribute_factory = factory.AttributeFactory() + + managed_object = pie_objects.SymmetricKey( + enums.CryptographicAlgorithm.AES, + 8, + b'\x00' + ) + + # Test that multiple duplicate names cannot be set on an object. + name_a = attribute_factory.create_attribute( + enums.AttributeType.NAME, + attributes.Name.create( + 'Test Symmetric Key', + enums.NameType.UNINTERPRETED_TEXT_STRING + ) + ) + name_b = attribute_factory.create_attribute( + enums.AttributeType.NAME, + attributes.Name.create( + 'Test Symmetric Key', + enums.NameType.UNINTERPRETED_TEXT_STRING + ) + ) + + args = ( + managed_object, + ('Name', [name_a.attribute_value, name_b.attribute_value]) + ) + regex = "Cannot set duplicate name values." + self.assertRaisesRegexp( + exceptions.InvalidField, + regex, + e._set_attribute_on_managed_object, + *args + ) + + # Test that a multivalued, unsupported attribute cannot be set on an + # object. + name_a = attribute_factory.create_attribute( + enums.AttributeType.NAME, + attributes.Name.create( + 'Test Symmetric Key', + enums.NameType.UNINTERPRETED_TEXT_STRING + ) + ) + name_b = attribute_factory.create_attribute( + enums.AttributeType.NAME, + attributes.Name.create( + 'Test Symmetric Key', + enums.NameType.UNINTERPRETED_TEXT_STRING + ) + ) + + args = ( + managed_object, + ('Digest', [name_a.attribute_value, name_b.attribute_value]) + ) + regex = "The Digest attribute is unsupported." + self.assertRaisesRegexp( + exceptions.InvalidField, + regex, + e._set_attribute_on_managed_object, + *args + ) + + # Test that a set attribute cannot be overwritten. + length = attribute_factory.create_attribute( + enums.AttributeType.CRYPTOGRAPHIC_LENGTH, + 128 + ) + + args = ( + managed_object, + ('Cryptographic Length', length.attribute_value) + ) + regex = "Cannot overwrite the Cryptographic Length attribute." + self.assertRaisesRegexp( + exceptions.InvalidField, + regex, + e._set_attribute_on_managed_object, + *args + ) + + # Test that an unsupported attribute cannot be set. + object_group = attribute_factory.create_attribute( + enums.AttributeType.OBJECT_GROUP, + 'Test Group' + ) + + args = ( + managed_object, + ('Object Group', object_group.attribute_value) + ) + regex = "The Object Group attribute is unsupported." + self.assertRaisesRegexp( + exceptions.InvalidField, + regex, + e._set_attribute_on_managed_object, + *args + ) + + def test_register(self): + """ + Test that a Register request can be processed correctly. + """ + e = engine.KmipEngine() + e._data_store = self.engine + e._data_store_session_factory = self.session_factory + e._data_session = e._data_store_session_factory() + e._logger = mock.MagicMock() + + attribute_factory = factory.AttributeFactory() + + # Build a SymmetricKey for registration. + object_type = attributes.ObjectType(enums.ObjectType.SYMMETRIC_KEY) + template_attribute = objects.TemplateAttribute( + attributes=[ + attribute_factory.create_attribute( + enums.AttributeType.NAME, + attributes.Name.create( + 'Test Symmetric Key', + enums.NameType.UNINTERPRETED_TEXT_STRING + ) + ), + attribute_factory.create_attribute( + enums.AttributeType.CRYPTOGRAPHIC_ALGORITHM, + enums.CryptographicAlgorithm.AES + ), + attribute_factory.create_attribute( + enums.AttributeType.CRYPTOGRAPHIC_LENGTH, + 128 + ), + attribute_factory.create_attribute( + enums.AttributeType.CRYPTOGRAPHIC_USAGE_MASK, + [ + enums.CryptographicUsageMask.ENCRYPT, + enums.CryptographicUsageMask.DECRYPT + ] + ) + ] + ) + key_bytes = ( + b'\x00\x00\x00\x00\x00\x00\x00\x00' + b'\x00\x00\x00\x00\x00\x00\x00\x00' + ) + secret = secrets.SymmetricKey( + key_block=objects.KeyBlock( + key_format_type=misc.KeyFormatType(enums.KeyFormatType.RAW), + key_value=objects.KeyValue( + key_material=objects.KeyMaterial(key_bytes) + ), + cryptographic_algorithm=attributes.CryptographicAlgorithm( + enums.CryptographicAlgorithm.AES + ), + cryptographic_length=attributes.CryptographicLength(128) + ) + ) + + payload = register.RegisterRequestPayload( + object_type=object_type, + template_attribute=template_attribute, + secret=secret + ) + + response_payload = e._process_register(payload) + e._data_session.commit() + e._data_session = e._data_store_session_factory() + + e._logger.info.assert_called_once_with( + "Processing operation: Register" + ) + + uid = response_payload.unique_identifier.value + self.assertEqual('1', uid) + + # Retrieve the stored object and verify all attributes were set + # appropriately. + symmetric_key = e._data_session.query( + pie_objects.SymmetricKey + ).filter( + pie_objects.ManagedObject.unique_identifier == uid + ).one() + self.assertEqual( + enums.KeyFormatType.RAW, + symmetric_key.key_format_type + ) + self.assertEqual(1, len(symmetric_key.names)) + self.assertIn('Test Symmetric Key', symmetric_key.names) + self.assertEqual(key_bytes, symmetric_key.value) + self.assertEqual( + enums.CryptographicAlgorithm.AES, + symmetric_key.cryptographic_algorithm + ) + self.assertEqual(128, symmetric_key.cryptographic_length) + self.assertEqual(2, len(symmetric_key.cryptographic_usage_masks)) + self.assertIn( + enums.CryptographicUsageMask.ENCRYPT, + symmetric_key.cryptographic_usage_masks + ) + self.assertIn( + enums.CryptographicUsageMask.DECRYPT, + symmetric_key.cryptographic_usage_masks + ) + + self.assertEqual(uid, e._id_placeholder) + + def test_register_unsupported_object_type(self): + """ + Test that an InvalidField error is generated when attempting to + register an unsupported object type. + """ + e = engine.KmipEngine() + e._data_store = self.engine + e._data_store_session_factory = self.session_factory + e._data_session = e._data_store_session_factory() + e._logger = mock.MagicMock() + + object_type = attributes.ObjectType(enums.ObjectType.SPLIT_KEY) + payload = register.RegisterRequestPayload(object_type=object_type) + + args = (payload, ) + regex = "The SplitKey object type is not supported." + self.assertRaisesRegexp( + exceptions.InvalidField, + regex, + e._process_register, + *args + ) + + def test_request_omitting_secret(self): + """ + Test that an InvalidField error is generate when trying to register + a secret in absentia. + """ + e = engine.KmipEngine() + e._data_store = self.engine + e._data_store_session_factory = self.session_factory + e._data_session = e._data_store_session_factory() + e._logger = mock.MagicMock() + + object_type = attributes.ObjectType(enums.ObjectType.SYMMETRIC_KEY) + payload = register.RegisterRequestPayload(object_type=object_type) + + args = (payload, ) + regex = "Cannot register a secret in absentia." + self.assertRaisesRegexp( + exceptions.InvalidField, + regex, + e._process_register, + *args + ) + def test_get(self): """ Test that a Get request can be processed correctly. @@ -1278,19 +1912,23 @@ class TestKmipEngine(testtools.TestCase): e._logger.info.assert_called_once_with("Processing operation: Query") self.assertIsInstance(result, query.QueryResponsePayload) self.assertIsNotNone(result.operations) - self.assertEqual(3, len(result.operations)) + self.assertEqual(4, len(result.operations)) self.assertEqual( - enums.Operation.GET, + enums.Operation.REGISTER, result.operations[0].value ) self.assertEqual( - enums.Operation.DESTROY, + enums.Operation.GET, result.operations[1].value ) self.assertEqual( - enums.Operation.QUERY, + enums.Operation.DESTROY, result.operations[2].value ) + self.assertEqual( + enums.Operation.QUERY, + result.operations[3].value + ) self.assertEqual(list(), result.object_types) self.assertIsNotNone(result.vendor_identification) self.assertEqual( @@ -1309,7 +1947,7 @@ class TestKmipEngine(testtools.TestCase): e._logger.info.assert_called_once_with("Processing operation: Query") self.assertIsNotNone(result.operations) - self.assertEqual(4, len(result.operations)) + self.assertEqual(5, len(result.operations)) self.assertEqual( enums.Operation.DISCOVER_VERSIONS, result.operations[-1].value @@ -1380,3 +2018,141 @@ class TestKmipEngine(testtools.TestCase): "Processing operation: DiscoverVersions" ) self.assertEqual([], result.protocol_versions) + + def test_register_get_destroy(self): + """ + Test that a managed object can be registered, retrieved, and destroyed + without error. + """ + e = engine.KmipEngine() + e._data_store = self.engine + e._data_store_session_factory = self.session_factory + e._data_session = e._data_store_session_factory() + e._logger = mock.MagicMock() + + attribute_factory = factory.AttributeFactory() + + # Build a SymmetricKey for registration. + object_type = attributes.ObjectType(enums.ObjectType.SYMMETRIC_KEY) + template_attribute = objects.TemplateAttribute( + attributes=[ + attribute_factory.create_attribute( + enums.AttributeType.NAME, + attributes.Name.create( + 'Test Symmetric Key', + enums.NameType.UNINTERPRETED_TEXT_STRING + ) + ), + attribute_factory.create_attribute( + enums.AttributeType.CRYPTOGRAPHIC_ALGORITHM, + enums.CryptographicAlgorithm.AES + ), + attribute_factory.create_attribute( + enums.AttributeType.CRYPTOGRAPHIC_LENGTH, + 128 + ), + attribute_factory.create_attribute( + enums.AttributeType.CRYPTOGRAPHIC_USAGE_MASK, + [ + enums.CryptographicUsageMask.ENCRYPT, + enums.CryptographicUsageMask.DECRYPT + ] + ) + ] + ) + key_bytes = ( + b'\x00\x00\x00\x00\x00\x00\x00\x00' + b'\x00\x00\x00\x00\x00\x00\x00\x00' + ) + secret = secrets.SymmetricKey( + key_block=objects.KeyBlock( + key_format_type=misc.KeyFormatType(enums.KeyFormatType.RAW), + key_value=objects.KeyValue( + key_material=objects.KeyMaterial(key_bytes) + ), + cryptographic_algorithm=attributes.CryptographicAlgorithm( + enums.CryptographicAlgorithm.AES + ), + cryptographic_length=attributes.CryptographicLength(128) + ) + ) + + # Register the symmetric key with the corresponding attributes + payload = register.RegisterRequestPayload( + object_type=object_type, + template_attribute=template_attribute, + secret=secret + ) + + response_payload = e._process_register(payload) + e._data_session.commit() + e._data_session = e._data_store_session_factory() + + e._logger.info.assert_called_once_with( + "Processing operation: Register" + ) + + uid = response_payload.unique_identifier.value + self.assertEqual('1', uid) + + e._logger.reset_mock() + + # Retrieve the registered key using Get and verify all fields set + payload = get.GetRequestPayload( + unique_identifier=attributes.UniqueIdentifier(uid) + ) + + response_payload = e._process_get(payload) + e._data_session.commit() + e._data_session = e._data_store_session_factory() + + e._logger.info.assert_called_once_with( + "Processing operation: Get" + ) + self.assertEqual( + enums.ObjectType.SYMMETRIC_KEY, + response_payload.object_type.value + ) + self.assertEqual(str(uid), response_payload.unique_identifier.value) + self.assertIsInstance(response_payload.secret, secrets.SymmetricKey) + self.assertEqual( + key_bytes, + response_payload.secret.key_block.key_value.key_material.value + ) + self.assertEqual( + enums.KeyFormatType.RAW, + response_payload.secret.key_block.key_format_type.value + ) + self.assertEqual( + enums.CryptographicAlgorithm.AES, + response_payload.secret.key_block.cryptographic_algorithm.value + ) + self.assertEqual( + 128, + response_payload.secret.key_block.cryptographic_length.value + ) + + e._logger.reset_mock() + + # Destroy the symmetric key and verify it cannot be accessed again + payload = destroy.DestroyRequestPayload( + unique_identifier=attributes.UniqueIdentifier(uid) + ) + + response_payload = e._process_destroy(payload) + e._data_session.commit() + e._data_session = e._data_store_session_factory() + + e._logger.info.assert_called_once_with( + "Processing operation: Destroy" + ) + self.assertEqual(str(uid), response_payload.unique_identifier.value) + self.assertRaises( + exc.NoResultFound, + e._data_session.query(pie_objects.OpaqueObject).filter( + pie_objects.ManagedObject.unique_identifier == uid + ).one + ) + + e._data_session.commit() + e._data_store_session_factory()