<s> import logging <EOL> import os <EOL> from kmip . core . attributes import CryptographicLength <EOL> from kmip . core . attributes import CryptographicAlgorithm <EOL> from kmip . core . attributes import ObjectType <EOL> from kmip . core . attributes import UniqueIdentifier <EOL> from kmip . core . enums import AttributeType as AT <EOL> from kmip . core . enums import CryptographicAlgorithm as CA <EOL> from kmip . core . enums import KeyFormatType as KeyFormatTypeEnum <EOL> from kmip . core . enums import ObjectType as OT <EOL> from kmip . core . enums import ResultReason as ResultReasonEnum <EOL> from kmip . core . enums import ResultStatus as RS <EOL> from kmip . core . factories . attributes import AttributeFactory <EOL> from kmip . core . factories . keys import KeyFactory <EOL> from kmip . core . factories . secrets import SecretFactory <EOL> from kmip . core . messages . contents import ResultStatus <EOL> from kmip . core . messages . contents import ResultReason <EOL> from kmip . core . messages . contents import ResultMessage <EOL> from kmip . core . messages . contents import ProtocolVersion <EOL> from kmip . core . misc import KeyFormatType <EOL> from kmip . core . objects import KeyBlock <EOL> from kmip . core . objects import KeyMaterial <EOL> from kmip . core . objects import KeyValue <EOL> from kmip . core . objects import TemplateAttribute <EOL> from kmip . core . secrets import SymmetricKey <EOL> from kmip . services . server . repo . mem_repo import MemRepo <EOL> from kmip . services . results import CreateResult <EOL> from kmip . services . results import DestroyResult <EOL> from kmip . services . results import GetResult <EOL> from kmip . services . results import OperationResult <EOL> from kmip . services . results import RegisterResult <EOL> from kmip . services . results import LocateResult <EOL> from kmip . services . results import DiscoverVersionsResult <EOL> class KMIP ( object ) : <EOL> def __init__ ( self ) : <EOL> pass <EOL> def create ( self , object_type , template_attribute , credential = None ) : <EOL> raise NotImplementedError ( ) <EOL> def create_key_pair ( self , common_template_attribute , <EOL> private_key_template_attribute , <EOL> public_key_template_attribute ) : <EOL> raise NotImplementedError ( ) <EOL> def register ( self , object_type , template_attribute , secret , <EOL> credential = None ) : <EOL> raise NotImplementedError ( ) <EOL> def rekey_key_pair ( self , private_key_unique_identifier , <EOL> offset , common_template_attribute , <EOL> private_key_template_attribute , <EOL> public_key_template_attribute ) : <EOL> raise NotImplementedError ( ) <EOL> def get ( self , uuid = None , key_format_type = None , key_compression_type = None , <EOL> key_wrapping_specification = None , credential = None ) : <EOL> raise NotImplementedError ( ) <EOL> def destroy ( self , uuid , credential = None ) : <EOL> raise NotImplementedError ( ) <EOL> def locate ( self , maximum_items = None , storate_status_mask = None , <EOL> object_group_member = None , attributes = None , <EOL> credential = None ) : <EOL> raise NotImplementedError ( ) <EOL> def discover_versions ( self , protocol_versions = None ) : <EOL> raise NotImplementedError ( ) <EOL> class KMIPImpl ( KMIP ) : <EOL> def __init__ ( self ) : <EOL> super ( KMIPImpl , self ) . __init__ ( ) <EOL> self . logger = logging . getLogger ( __name__ ) <EOL> self . key_factory = KeyFactory ( ) <EOL> self . secret_factory = SecretFactory ( ) <EOL> self . attribute_factory = AttributeFactory ( ) <EOL> self . repo = MemRepo ( ) <EOL> self . protocol_versions = [ <EOL> ProtocolVersion . create ( <NUM_LIT:1> , <NUM_LIT:1> ) , <EOL> ProtocolVersion . create ( <NUM_LIT:1> , <NUM_LIT:0> ) <EOL> ] <EOL> def create ( self , object_type , template_attribute , credential = None ) : <EOL> self . logger . debug ( '<STR_LIT>' ) <EOL> self . logger . debug ( '<STR_LIT>' % object_type ) <EOL> bit_length = <NUM_LIT> <EOL> attributes = template_attribute . attributes <EOL> ret_attributes = [ ] <EOL> if object_type . value != OT . SYMMETRIC_KEY : <EOL> self . logger . debug ( '<STR_LIT>' ) <EOL> return self . _get_invalid_field_result ( '<STR_LIT>' ) <EOL> try : <EOL> alg_attr = self . _validate_req_field ( <EOL> attributes , AT . CRYPTOGRAPHIC_ALGORITHM . value , <EOL> ( CA . AES , ) , '<STR_LIT>' ) <EOL> len_attr = self . _validate_req_field ( <EOL> attributes , AT . CRYPTOGRAPHIC_LENGTH . value , <EOL> ( <NUM_LIT> , <NUM_LIT> , <NUM_LIT> ) , '<STR_LIT>' , False ) <EOL> self . _validate_req_field ( <EOL> attributes , AT . CRYPTOGRAPHIC_USAGE_MASK . value , ( ) , '<STR_LIT>' ) <EOL> except InvalidFieldException as e : <EOL> self . logger . debug ( '<STR_LIT>' ) <EOL> return e . result <EOL> crypto_alg = CryptographicAlgorithm ( CA ( alg_attr . attribute_value . value ) ) <EOL> if len_attr is None : <EOL> self . logger . debug ( '<STR_LIT>' ) <EOL> attribute_type = AT . CRYPTOGRAPHIC_LENGTH <EOL> length_attribute = self . attribute_factory . create_attribute ( attribute_type , bit_length ) <EOL> attributes . append ( length_attribute ) <EOL> ret_attributes . append ( length_attribute ) <EOL> else : <EOL> bit_length = len_attr . attribute_value . value <EOL> key = self . _gen_symmetric_key ( bit_length , crypto_alg ) <EOL> s_uuid , uuid_attribute = self . _save ( key , attributes ) <EOL> ret_attributes . append ( uuid_attribute ) <EOL> template_attribute = TemplateAttribute ( attributes = ret_attributes ) <EOL> return CreateResult ( ResultStatus ( RS . SUCCESS ) , object_type = object_type , <EOL> uuid = UniqueIdentifier ( s_uuid ) , <EOL> template_attribute = template_attribute ) <EOL> def create_key_pair ( self , common_template_attribute , <EOL> private_key_template_attribute , <EOL> public_key_template_attribute ) : <EOL> raise NotImplementedError ( ) <EOL> def register ( self , object_type , template_attribute , secret , <EOL> credential = None ) : <EOL> self . logger . debug ( '<STR_LIT>' ) <EOL> self . logger . debug ( '<STR_LIT>' % object_type ) <EOL> attributes = template_attribute . attributes <EOL> ret_attributes = [ ] <EOL> if object_type is None : <EOL> self . logger . debug ( '<STR_LIT>' ) <EOL> return self . _get_missing_field_result ( '<STR_LIT>' ) <EOL> if object_type . value != OT . SYMMETRIC_KEY : <EOL> self . logger . debug ( '<STR_LIT>' ) <EOL> return self . _get_invalid_field_result ( '<STR_LIT>' ) <EOL> if secret is None or not isinstance ( secret , SymmetricKey ) : <EOL> msg = '<STR_LIT>' <EOL> self . logger . debug ( msg ) <EOL> return self . _get_invalid_field_result ( msg ) <EOL> self . logger . debug ( '<STR_LIT>' ) <EOL> if attributes is None : <EOL> attributes = [ ] <EOL> attributes . extend ( self . _get_key_block_attributes ( secret . key_block ) ) <EOL> self . logger . debug ( '<STR_LIT>' ) <EOL> try : <EOL> self . _validate_req_field ( <EOL> attributes , AT . CRYPTOGRAPHIC_ALGORITHM . value , ( CA . AES , ) , <EOL> '<STR_LIT>' ) <EOL> self . _validate_req_field ( <EOL> attributes , AT . CRYPTOGRAPHIC_LENGTH . value , ( <NUM_LIT> , <NUM_LIT> , <NUM_LIT> ) , <EOL> '<STR_LIT>' ) <EOL> self . _validate_req_field ( <EOL> attributes , AT . CRYPTOGRAPHIC_USAGE_MASK . value , ( ) , '<STR_LIT>' ) <EOL> except InvalidFieldException as e : <EOL> self . logger . debug ( '<STR_LIT>' ) <EOL> return RegisterResult ( e . result . result_status , <EOL> e . result . result_reason , <EOL> e . result . result_message ) <EOL> s_uuid , uuid_attribute = self . _save ( secret , attributes ) <EOL> ret_attributes . append ( uuid_attribute ) <EOL> template_attribute = TemplateAttribute ( attributes = ret_attributes ) <EOL> return RegisterResult ( ResultStatus ( RS . SUCCESS ) , <EOL> uuid = UniqueIdentifier ( s_uuid ) , <EOL> template_attribute = template_attribute ) <EOL> def rekey_key_pair ( self , private_key_unique_identifier , <EOL> offset , common_template_attribute , <EOL> private_key_template_attribute , <EOL> public_key_template_attribute ) : <EOL> raise NotImplementedError ( ) <EOL> def get ( self , <EOL> uuid = None , <EOL> key_format_type = None , <EOL> key_compression_type = None , <EOL> key_wrapping_specification = None , <EOL> credential = None ) : <EOL> self . logger . debug ( '<STR_LIT>' ) <EOL> ret_value = RS . OPERATION_FAILED <EOL> if uuid is None or not hasattr ( uuid , '<STR_LIT:value>' ) : <EOL> self . logger . debug ( '<STR_LIT>' ) <EOL> reason = ResultReason ( ResultReasonEnum . ITEM_NOT_FOUND ) <EOL> message = ResultMessage ( '<STR_LIT>' ) <EOL> return GetResult ( ResultStatus ( ret_value ) , reason , message ) <EOL> if key_format_type is None : <EOL> self . logger . debug ( '<STR_LIT>' ) <EOL> key_format_type = KeyFormatType ( KeyFormatTypeEnum . RAW ) <EOL> if key_format_type . value != KeyFormatTypeEnum . RAW : <EOL> self . logger . debug ( '<STR_LIT>' ) <EOL> reason = ResultReason ( ResultReasonEnum . <EOL> KEY_FORMAT_TYPE_NOT_SUPPORTED ) <EOL> message = ResultMessage ( '<STR_LIT>' ) <EOL> return GetResult ( ResultStatus ( ret_value ) , reason , message ) <EOL> if key_compression_type is not None : <EOL> self . logger . debug ( '<STR_LIT>' ) <EOL> reason = ResultReason ( ResultReasonEnum . <EOL> KEY_COMPRESSION_TYPE_NOT_SUPPORTED ) <EOL> message = ResultMessage ( '<STR_LIT>' ) <EOL> return GetResult ( ResultStatus ( ret_value ) , reason , message ) <EOL> if key_wrapping_specification is not None : <EOL> self . logger . debug ( '<STR_LIT>' ) <EOL> reason = ResultReason ( ResultReasonEnum . FEATURE_NOT_SUPPORTED ) <EOL> message = ResultMessage ( '<STR_LIT>' ) <EOL> return GetResult ( ResultStatus ( ret_value ) , reason , message ) <EOL> self . logger . debug ( '<STR_LIT>' ) <EOL> managed_object , _ = self . repo . get ( uuid . value ) <EOL> if managed_object is None : <EOL> self . logger . debug ( '<STR_LIT>' ) <EOL> reason = ResultReason ( ResultReasonEnum . ITEM_NOT_FOUND ) <EOL> message = ResultMessage ( '<STR_LIT>' ) <EOL> return GetResult ( ResultStatus ( ret_value ) , reason , message ) <EOL> object_type = ObjectType ( OT . SYMMETRIC_KEY ) <EOL> ret_value = RS . SUCCESS <EOL> return GetResult ( ResultStatus ( ret_value ) , object_type = object_type , <EOL> uuid = uuid , secret = managed_object ) <EOL> def destroy ( self , uuid ) : <EOL> self . logger . debug ( '<STR_LIT>' ) <EOL> ret_value = RS . OPERATION_FAILED <EOL> if uuid is None or not hasattr ( uuid , '<STR_LIT:value>' ) : <EOL> self . logger . debug ( '<STR_LIT>' ) <EOL> reason = ResultReason ( ResultReasonEnum . ITEM_NOT_FOUND ) <EOL> message = ResultMessage ( '<STR_LIT>' ) <EOL> return DestroyResult ( ResultStatus ( ret_value ) , reason , message ) <EOL> msg = '<STR_LIT>' . format ( uuid ) <EOL> self . logger . debug ( msg ) <EOL> if not self . repo . delete ( uuid . value ) : <EOL> self . logger . debug ( '<STR_LIT>' ) <EOL> reason = ResultReason ( ResultReasonEnum . ITEM_NOT_FOUND ) <EOL> message = ResultMessage ( '<STR_LIT>' ) <EOL> return DestroyResult ( ResultStatus ( ret_value ) , reason , message ) <EOL> ret_value = RS . SUCCESS <EOL> return DestroyResult ( ResultStatus ( ret_value ) , uuid = uuid ) <EOL> def locate ( self , maximum_items = None , storage_status_mask = None , <EOL> object_group_member = None , attributes = None , <EOL> credential = None ) : <EOL> self . logger . debug ( '<STR_LIT>' ) <EOL> msg = '<STR_LIT>' <EOL> self . logger . debug ( msg ) <EOL> try : <EOL> uuids = self . repo . locate ( maximum_items , storage_status_mask , <EOL> object_group_member , attributes ) <EOL> return LocateResult ( ResultStatus ( RS . SUCCESS ) , uuids = uuids ) <EOL> except NotImplementedError : <EOL> msg = ResultMessage ( '<STR_LIT>' ) <EOL> reason = ResultReason ( ResultReasonEnum . OPERATION_NOT_SUPPORTED ) <EOL> return LocateResult ( ResultStatus ( RS . OPERATION_FAILED ) , <EOL> result_reason = reason , result_message = msg ) <EOL> def discover_versions ( self , protocol_versions = None ) : <EOL> self . logger . debug ( <EOL> "<STR_LIT>" . format ( <EOL> protocol_versions ) ) <EOL> msg = '<STR_LIT>' <EOL> result_versions = list ( ) <EOL> if protocol_versions : <EOL> msg += "<STR_LIT>" . format ( protocol_versions ) <EOL> for version in protocol_versions : <EOL> if version in self . protocol_versions : <EOL> result_versions . append ( version ) <EOL> else : <EOL> result_versions = self . protocol_versions <EOL> self . logger . debug ( msg ) <EOL> try : <EOL> return DiscoverVersionsResult ( ResultStatus ( RS . SUCCESS ) , <EOL> protocol_versions = result_versions ) <EOL> except Exception : <EOL> msg = ResultMessage ( '<STR_LIT>' ) <EOL> reason = ResultReason ( ResultReasonEnum . GENERAL_FAILURE ) <EOL> return DiscoverVersionsResult ( ResultStatus ( RS . OPERATION_FAILED ) , <EOL> result_reason = reason , <EOL> result_message = msg ) <EOL> def _validate_req_field ( self , attrs , name , expected , msg , required = True ) : <EOL> self . logger . debug ( '<STR_LIT>' % name ) <EOL> seen = False <EOL> found_attr = None <EOL> for attr in attrs : <EOL> if self . _validate_field ( attr , name , expected , msg ) : <EOL> if seen : <EOL> msg = '<STR_LIT>' % name <EOL> self . logger . debug ( msg ) <EOL> result = self . _get_duplicate_attribute_result ( name ) <EOL> raise InvalidFieldException ( result ) <EOL> seen = True <EOL> found_attr = attr <EOL> if required and not seen : <EOL> result = self . _get_missing_field_result ( name ) <EOL> raise InvalidFieldException ( result ) <EOL> return found_attr <EOL> def _validate_field ( self , attr , name , expected , msg ) : <EOL> if attr . attribute_name . value == name : <EOL> self . logger . debug ( '<STR_LIT>' % name ) <EOL> if not expected or attr . attribute_value . value in expected : <EOL> self . logger . debug ( '<STR_LIT>' ) <EOL> return True <EOL> else : <EOL> self . logger . debug ( '<STR_LIT>' ) <EOL> result = self . _get_invalid_field_result ( msg ) <EOL> raise InvalidFieldException ( result ) <EOL> else : <EOL> return False <EOL> def _get_invalid_field_result ( self , msg ) : <EOL> status = ResultStatus ( RS . OPERATION_FAILED ) <EOL> reason = ResultReason ( ResultReasonEnum . INVALID_FIELD ) <EOL> message = ResultMessage ( msg ) <EOL> return OperationResult ( status , reason , message ) <EOL> def _get_missing_field_result ( self , name ) : <EOL> msg = '<STR_LIT>' % name <EOL> self . logger . debug ( msg ) <EOL> status = ResultStatus ( RS . OPERATION_FAILED ) <EOL> reason = ResultReason ( ResultReasonEnum . ITEM_NOT_FOUND ) <EOL> message = ResultMessage ( msg ) <EOL> return OperationResult ( status , reason , message ) <EOL> def _get_duplicate_attribute_result ( self , name ) : <EOL> msg = '<STR_LIT>' % name <EOL> self . logger . debug ( msg ) <EOL> status = ResultStatus ( RS . OPERATION_FAILED ) <EOL> reason = ResultReason ( ResultReasonEnum . INDEX_OUT_OF_BOUNDS ) <EOL> message = ResultMessage ( msg ) <EOL> return OperationResult ( status , reason , message ) <EOL> def _gen_symmetric_key ( self , bit_length , crypto_alg ) : <EOL> key_format_type = KeyFormatType ( KeyFormatTypeEnum . RAW ) <EOL> key_material = KeyMaterial ( os . urandom ( int ( bit_length / <NUM_LIT:8> ) ) ) <EOL> key_value = KeyValue ( key_material ) <EOL> crypto_length = CryptographicLength ( bit_length ) <EOL> key_block = KeyBlock ( key_format_type , None , key_value , crypto_alg , <EOL> crypto_length , None ) <EOL> return SymmetricKey ( key_block ) <EOL> def _save ( self , key , attributes ) : <EOL> s_uuid = self . repo . save ( key , attributes ) <EOL> self . logger . debug ( '<STR_LIT>' % s_uuid ) <EOL> attribute_type = AT . UNIQUE_IDENTIFIER <EOL> attribute = self . attribute_factory . create_attribute ( attribute_type , <EOL> s_uuid ) <EOL> attributes . append ( attribute ) <EOL> self . repo . update ( s_uuid , key , attributes ) <EOL> return s_uuid , attribute <EOL> def _get_key_block_attributes ( self , key_block ) : <EOL> self . logger . debug ( '<STR_LIT>' ) <EOL> attributes = [ ] <EOL> if key_block . cryptographic_algorithm is not None : <EOL> self . logger . debug ( '<STR_LIT>' ) <EOL> self . logger . debug ( '<STR_LIT>' ) <EOL> at = AT . CRYPTOGRAPHIC_ALGORITHM <EOL> alg = key_block . cryptographic_algorithm . value <EOL> attributes . append ( self . attribute_factory . create_attribute ( at , alg ) ) <EOL> if key_block . cryptographic_length is not None : <EOL> self . logger . debug ( '<STR_LIT>' ) <EOL> self . logger . debug ( '<STR_LIT>' ) <EOL> at = AT . CRYPTOGRAPHIC_LENGTH <EOL> len = key_block . cryptographic_length . value <EOL> attributes . append ( self . attribute_factory . create_attribute ( at , len ) ) <EOL> self . logger . debug ( '<STR_LIT>' ) <EOL> if key_block . key_wrapping_data is not None : <EOL> self . logger . debug ( '<STR_LIT>' ) <EOL> kv = key_block . key_value <EOL> if isinstance ( kv , KeyValue ) : <EOL> kv = key_block . key_value <EOL> if kv . attributes is not None : <EOL> self . logger . debug ( '<STR_LIT>' ) <EOL> attributes . extend ( kv . attributes ) <EOL> return attributes <EOL> class InvalidFieldException ( Exception ) : <EOL> def __init__ ( self , result ) : <EOL> super ( InvalidFieldException , self ) . __init__ ( ) <EOL> self . result = result </s>
