diff --git a/kmip/core/attributes.py b/kmip/core/attributes.py index 843b0c445..4390f8b8f 100644 --- a/kmip/core/attributes.py +++ b/kmip/core/attributes.py @@ -1263,6 +1263,104 @@ def __ne__(self, other): else: return NotImplemented +class Link(primitives.Struct): + """ + A structure used to store a link between a Managed Object and a related + Managed Object. + Attributes: + link_type: The type of link (e.g. PrivateKeyLink, PublicKeyLink). + linked_object_identifier: The unique identifier of the linked object. + See Section 3.32 of the KMIP v1.2 specification for more information. + """ + def __init__(self, link_type=None, linked_object_identifier=None): + super(Link, self).__init__(enums.Tags.LINK) + self._link_type = None + self._linked_object_identifier = None + self.link_type = link_type + self.linked_object_identifier = linked_object_identifier + + @property + def link_type(self): + if self._link_type: + return self._link_type.value + return None + + @link_type.setter + def link_type(self, value): + if value is None: + self._link_type = None + elif isinstance(value, enums.LinkType): + self._link_type = primitives.Enumeration( + enums.LinkType, + value=value, + tag=enums.Tags.LINK_TYPE + ) + else: + raise TypeError("The link type must be a LinkType enumeration.") + + @property + def linked_object_identifier(self): + if self._linked_object_identifier: + return self._linked_object_identifier.value + return None + + @linked_object_identifier.setter + def linked_object_identifier(self, value): + if value is None: + self._linked_object_identifier = None + elif isinstance(value, str): + self._linked_object_identifier = primitives.TextString( + value=value, + tag=enums.Tags.LINKED_OBJECT_IDENTIFIER + ) + else: + raise TypeError("The linked object identifier must be a string.") + + def read(self, input_buffer, kmip_version=enums.KMIPVersion.KMIP_1_0): + super(Link, self).read(input_buffer, kmip_version=kmip_version) + local_buffer = utils.BytearrayStream(input_buffer.read(self.length)) + if self.is_tag_next(enums.Tags.LINK_TYPE, local_buffer): + self._link_type = primitives.Enumeration( + enums.LinkType, + tag=enums.Tags.LINK_TYPE + ) + self._link_type.read(local_buffer, kmip_version=kmip_version) + else: + raise exceptions.InvalidKmipEncoding( + "The Link encoding is missing the LinkType field." + ) + if self.is_tag_next(enums.Tags.LINKED_OBJECT_IDENTIFIER, local_buffer): + self._linked_object_identifier = primitives.TextString( + tag=enums.Tags.LINKED_OBJECT_IDENTIFIER + ) + self._linked_object_identifier.read( + local_buffer, kmip_version=kmip_version + ) + else: + raise exceptions.InvalidKmipEncoding( + "The Link encoding is missing the LinkedObjectIdentifier field." + ) + self.is_oversized(local_buffer) + + def write(self, output_buffer, kmip_version=enums.KMIPVersion.KMIP_1_0): + local_buffer = utils.BytearrayStream() + if self._link_type: + self._link_type.write(local_buffer, kmip_version=kmip_version) + else: + raise exceptions.InvalidField( + "The Link object is missing the LinkType field." + ) + if self._linked_object_identifier: + self._linked_object_identifier.write( + local_buffer, kmip_version=kmip_version + ) + else: + raise exceptions.InvalidField( + "The Link object is missing the LinkedObjectIdentifier field." + ) + self.length = local_buffer.length() + super(Link, self).write(output_buffer, kmip_version=kmip_version) + output_buffer.write(local_buffer.buffer) # 3.37 class ContactInformation(TextString): diff --git a/kmip/core/factories/attribute_values.py b/kmip/core/factories/attribute_values.py index 1bc1e208f..2981d34ce 100644 --- a/kmip/core/factories/attribute_values.py +++ b/kmip/core/factories/attribute_values.py @@ -97,7 +97,7 @@ def create_attribute_value(self, name, value): elif name is enums.AttributeType.FRESH: return primitives.Boolean(value, enums.Tags.FRESH) elif name is enums.AttributeType.LINK: - raise NotImplementedError() + return self._create_link(value) elif name is enums.AttributeType.APPLICATION_SPECIFIC_INFORMATION: return self._create_application_specific_information(value) elif name is enums.AttributeType.CONTACT_INFORMATION: @@ -197,7 +197,7 @@ def create_attribute_value_by_enum(self, enum, value): elif enum is enums.Tags.FRESH: return primitives.Boolean(value, enums.Tags.FRESH) elif enum is enums.Tags.LINK: - raise NotImplementedError() + return self._create_link(value) elif enum is enums.Tags.APPLICATION_SPECIFIC_INFORMATION: return self._create_application_specific_information(value) elif enum is enums.Tags.CONTACT_INFORMATION: @@ -280,6 +280,15 @@ def _create_cryptographic_usage_mask(self, flags): return attributes.CryptographicUsageMask(mask) + def _create_link(self, value): + if value: + return attributes.Link( + link_type=value.get('link_type'), + linked_object_identifier=value.get('linked_object_identifier') + ) + else: + return attributes.Link() + def _create_application_specific_information(self, info): if info: return attributes.ApplicationSpecificInformation( diff --git a/kmip/pie/objects.py b/kmip/pie/objects.py index e45746faa..e1a981133 100644 --- a/kmip/pie/objects.py +++ b/kmip/pie/objects.py @@ -108,6 +108,16 @@ class ManagedObject(sql.Base): sensitive = Column("sensitive", Boolean, default=False) initial_date = Column(Integer, default=0) _owner = Column('owner', String(50), default=None) + # The unique_identifier of the object this one was created paired with — + # the other half of a CreateKeyPair. Stored rather than inferred: the + # previous implementation recovered the pairing by matching on + # (_owner, initial_date), and initial_date has whole-second granularity, + # so two key pairs created for the same owner within one second were + # indistinguishable and the lookup returned an arbitrary one. For a KMIP + # client that encrypts under the public key and later asks for the linked + # private key, that silently yields the wrong key and surfaces only as a + # failed decryption long afterwards. + _link_id = Column('link_id', Integer, default=None) app_specific_info = sqlalchemy.orm.relationship( "ApplicationSpecificInformation", @@ -147,6 +157,7 @@ def __init__(self): self.sensitive = False self._object_type = None self._owner = None + self._link_id = None # All remaining attributes are not considered part of the public API # and are subject to change. diff --git a/kmip/services/auth.py b/kmip/services/auth.py index a53af2974..ead6f2860 100644 --- a/kmip/services/auth.py +++ b/kmip/services/auth.py @@ -217,8 +217,13 @@ class TLS12AuthenticationSuite(AuthenticationSuite): 'ECDHE-RSA-AES256-SHA384', 'ECDHE-ECDSA-AES128-GCM-SHA256', 'ECDHE-ECDSA-AES256-GCM-SHA384', - 'ECDHE-ECDSA-AES128-SHA256', - 'ECDHE-ECDSA-AES256-SHA384' + # The ECDHE-ECDSA GCM suites above can only be negotiated with an ECDSA + # server certificate. A peer enforcing OpenSSL SECLEVEL=2 offers only + # forward-secret GCM suites, so against the RSA certificate this server + # is usually deployed with there is no overlap and the handshake fails + # at the SSL layer, before any application logging happens. + 'ECDHE-RSA-AES256-GCM-SHA384', + 'ECDHE-RSA-AES128-GCM-SHA256' ] def __init__(self, cipher_suites=None): diff --git a/kmip/services/server/engine.py b/kmip/services/server/engine.py index 43fdbe93b..ecd314f5b 100644 --- a/kmip/services/server/engine.py +++ b/kmip/services/server/engine.py @@ -730,6 +730,42 @@ def _get_attribute_from_managed_object(self, managed_object, attr_name): elif attr_name == 'Fresh': return None elif attr_name == 'Link': + from kmip.pie import objects as obj_module + obj_type = type(managed_object) + if obj_type == obj_module.PublicKey: + linked_type = obj_module.PrivateKey + link_type = enums.LinkType.PRIVATE_KEY_LINK + elif obj_type == obj_module.PrivateKey: + linked_type = obj_module.PublicKey + link_type = enums.LinkType.PUBLIC_KEY_LINK + else: + return None + link_id = getattr(managed_object, '_link_id', None) + if link_id is not None: + return [{'link_type': link_type, + 'linked_object_identifier': str(link_id)}] + # Fallback for objects created before the pairing was stored. This + # is ambiguous by construction — two key pairs created by one owner + # in the same second match equally — so it is a best effort for old + # rows only, and never the path a newly created pair takes. + linked = self._data_session.query(linked_type).filter_by( + initial_date=managed_object.initial_date, + _owner=managed_object._owner + ).order_by(linked_type.unique_identifier).all() + if len(linked) == 1: + return [{'link_type': link_type, + 'linked_object_identifier': + str(linked[0].unique_identifier)}] + if len(linked) > 1: + # Refuse to guess: returning an arbitrary key here is what + # produces undecryptable data. + self._logger.warning( + "Cannot resolve the Link attribute for object {0}: {1} " + "candidates share its owner and initial date. The pairing " + "predates link storage and is ambiguous.".format( + managed_object.unique_identifier, len(linked) + ) + ) return None elif attr_name == "Application Specific Information": values = [] @@ -1573,6 +1609,14 @@ def _process_create_key_pair(self, payload): # commit is called. This makes future support for UNDO problematic. self._data_session.commit() + # Record the pairing now that both identifiers exist. This is the only + # point at which the relationship is known for certain; recovering it + # later by matching on (owner, initial_date) cannot distinguish two key + # pairs created for the same owner in the same second. + public_key._link_id = private_key.unique_identifier + private_key._link_id = public_key.unique_identifier + self._data_session.commit() + self._logger.info( "Created a PublicKey with ID: {0}".format( public_key.unique_identifier @@ -2856,7 +2900,13 @@ def _process_query(self, payload): ]) if enums.QueryFunction.QUERY_OBJECTS in queries: - objects = list() + objects = list([ + enums.ObjectType.CERTIFICATE, + enums.ObjectType.SYMMETRIC_KEY, + enums.ObjectType.PUBLIC_KEY, + enums.ObjectType.PRIVATE_KEY, + enums.ObjectType.SECRET_DATA + ]) if enums.QueryFunction.QUERY_SERVER_INFORMATION in queries: vendor_identification = "PyKMIP {0} Software Server".format( kmip.__version__ diff --git a/kmip/tests/unit/core/factories/test_attribute_values.py b/kmip/tests/unit/core/factories/test_attribute_values.py index 6edc44826..dd7651837 100644 --- a/kmip/tests/unit/core/factories/test_attribute_values.py +++ b/kmip/tests/unit/core/factories/test_attribute_values.py @@ -423,10 +423,33 @@ def test_create_link(self): """ Test that a Link attribute can be created. """ - kwargs = {'name': enums.AttributeType.LINK, - 'value': None} - self.assertRaises( - NotImplementedError, self.factory.create_attribute_value, **kwargs) + attribute = self.factory.create_attribute_value( + enums.AttributeType.LINK, + { + 'link_type': enums.LinkType.PRIVATE_KEY_LINK, + 'linked_object_identifier': '42' + } + ) + self.assertIsInstance(attribute, attributes.Link) + self.assertEqual( + enums.LinkType.PRIVATE_KEY_LINK, + attribute.link_type + ) + self.assertEqual('42', attribute.linked_object_identifier) + + def test_create_link_without_a_value(self): + """ + Test that an empty Link attribute can be created. + + The factory raised NotImplementedError here until Link was + implemented; an empty Link is what the decoder needs when it is about + to read one off the wire. + """ + attribute = self.factory.create_attribute_value( + enums.AttributeType.LINK, + None + ) + self.assertIsInstance(attribute, attributes.Link) def test_create_application_specific_information(self): """ diff --git a/kmip/tests/unit/services/server/test_engine.py b/kmip/tests/unit/services/server/test_engine.py index 84deb436c..558542dd6 100644 --- a/kmip/tests/unit/services/server/test_engine.py +++ b/kmip/tests/unit/services/server/test_engine.py @@ -3529,6 +3529,107 @@ def test_create_omitting_attributes(self): ) e._logger.reset_mock() + def test_create_key_pair_links_each_pair_to_its_own_keys(self): + """ + Two key pairs created by the same owner within one second must each + link to their own counterpart. + + The pairing used to be recovered by matching on (owner, initial_date), + and initial_date has whole-second granularity, so this case resolved to + an arbitrary key. A client that encrypts under public key A and later + asks for A's linked private key would be handed B's — undecryptable + data, discovered at restore time. The clock is frozen here to make the + collision deterministic rather than a race. + """ + 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() + e._client_identity = ['tenant-a', None] + + attribute_factory = factory.AttributeFactory() + common_template = objects.TemplateAttribute( + attributes=[ + attribute_factory.create_attribute( + enums.AttributeType.CRYPTOGRAPHIC_ALGORITHM, + enums.CryptographicAlgorithm.RSA + ), + attribute_factory.create_attribute( + enums.AttributeType.CRYPTOGRAPHIC_LENGTH, + 2048 + ) + ], + tag=enums.Tags.COMMON_TEMPLATE_ATTRIBUTE + ) + public_template = objects.TemplateAttribute( + attributes=[ + attribute_factory.create_attribute( + enums.AttributeType.CRYPTOGRAPHIC_USAGE_MASK, + [enums.CryptographicUsageMask.ENCRYPT] + ) + ], + tag=enums.Tags.PUBLIC_KEY_TEMPLATE_ATTRIBUTE + ) + private_template = objects.TemplateAttribute( + attributes=[ + attribute_factory.create_attribute( + enums.AttributeType.CRYPTOGRAPHIC_USAGE_MASK, + [enums.CryptographicUsageMask.DECRYPT] + ) + ], + tag=enums.Tags.PRIVATE_KEY_TEMPLATE_ATTRIBUTE + ) + payload = payloads.CreateKeyPairRequestPayload( + common_template, + private_template, + public_template + ) + + with mock.patch('time.time', return_value=1700000000): + first = e._process_create_key_pair(payload) + second = e._process_create_key_pair(payload) + e._data_session.commit() + e._data_session = e._data_store_session_factory() + + pairs = [ + (first.public_key_unique_identifier, + first.private_key_unique_identifier), + (second.public_key_unique_identifier, + second.private_key_unique_identifier), + ] + + # Both pairs really did land in the same second, or this test proves + # nothing about the collision it exists to cover. + initial_dates = set() + for public_id, _ in pairs: + public_key = e._data_session.query(pie_objects.PublicKey).filter( + pie_objects.ManagedObject.unique_identifier == public_id + ).one() + initial_dates.add(public_key.initial_date) + self.assertEqual(1, len(initial_dates)) + + for public_id, private_id in pairs: + public_key = e._data_session.query(pie_objects.PublicKey).filter( + pie_objects.ManagedObject.unique_identifier == public_id + ).one() + private_key = e._data_session.query( + pie_objects.PrivateKey + ).filter( + pie_objects.ManagedObject.unique_identifier == private_id + ).one() + + link = e._get_attribute_from_managed_object(public_key, 'Link') + self.assertEqual( + str(private_id), + link[0]['linked_object_identifier'] + ) + back = e._get_attribute_from_managed_object(private_key, 'Link') + self.assertEqual( + str(public_id), + back[0]['linked_object_identifier'] + ) + def test_create_key_pair(self): """ Test that a CreateKeyPair request can be processed correctly. @@ -9940,7 +10041,19 @@ def test_query_1_0(self): enums.Operation.QUERY, result.operations[11] ) - self.assertIsNone(result.object_types) + # The server reports the object types it can actually manage. This + # used to be an empty list, which a client reads as "this server + # supports nothing" — Veeam refuses to register against such a server. + self.assertEqual( + [ + enums.ObjectType.CERTIFICATE, + enums.ObjectType.SYMMETRIC_KEY, + enums.ObjectType.PUBLIC_KEY, + enums.ObjectType.PRIVATE_KEY, + enums.ObjectType.SECRET_DATA + ], + result.object_types + ) self.assertIsNotNone(result.vendor_identification) self.assertEqual( "PyKMIP {0} Software Server".format(kmip.__version__), @@ -10028,7 +10141,19 @@ def test_query_1_1(self): enums.Operation.DISCOVER_VERSIONS, result.operations[12] ) - self.assertIsNone(result.object_types) + # The server reports the object types it can actually manage. This + # used to be an empty list, which a client reads as "this server + # supports nothing" — Veeam refuses to register against such a server. + self.assertEqual( + [ + enums.ObjectType.CERTIFICATE, + enums.ObjectType.SYMMETRIC_KEY, + enums.ObjectType.PUBLIC_KEY, + enums.ObjectType.PRIVATE_KEY, + enums.ObjectType.SECRET_DATA + ], + result.object_types + ) self.assertIsNotNone(result.vendor_identification) self.assertEqual( "PyKMIP {0} Software Server".format(kmip.__version__), @@ -10136,7 +10261,19 @@ def test_query_1_2(self): enums.Operation.MAC, result.operations[17] ) - self.assertIsNone(result.object_types) + # The server reports the object types it can actually manage. This + # used to be an empty list, which a client reads as "this server + # supports nothing" — Veeam refuses to register against such a server. + self.assertEqual( + [ + enums.ObjectType.CERTIFICATE, + enums.ObjectType.SYMMETRIC_KEY, + enums.ObjectType.PUBLIC_KEY, + enums.ObjectType.PRIVATE_KEY, + enums.ObjectType.SECRET_DATA + ], + result.object_types + ) self.assertIsNotNone(result.vendor_identification) self.assertEqual( "PyKMIP {0} Software Server".format(kmip.__version__), diff --git a/kmip/tests/unit/services/test_auth.py b/kmip/tests/unit/services/test_auth.py index 92ec1f275..94c66f8bb 100644 --- a/kmip/tests/unit/services/test_auth.py +++ b/kmip/tests/unit/services/test_auth.py @@ -170,12 +170,33 @@ def test_ciphers(self): 'ECDHE-RSA-AES256-SHA384', 'ECDHE-ECDSA-AES128-GCM-SHA256', 'ECDHE-ECDSA-AES256-GCM-SHA384', - 'ECDHE-ECDSA-AES128-SHA256', - 'ECDHE-ECDSA-AES256-SHA384', + 'ECDHE-RSA-AES256-GCM-SHA384', + 'ECDHE-RSA-AES128-GCM-SHA256', )) self.assertEqual(cipher_string, ciphers) + def test_rsa_gcm_suites_are_negotiable(self): + """ + The ECDHE-RSA GCM suites must survive OpenSSL's own parsing. + + Asserting the cipher *string* does not prove this: a malformed entry is + still a string, and OpenSSL silently drops tokens it does not + recognise rather than raising. A missing comma in this list once + concatenated two entries into one unknown token, which removed + ECDHE-RSA-AES256-GCM-SHA384 from the negotiable set while every + string-equality test kept passing. A peer restricted to forward-secret + GCM suites then negotiated AES-128 instead of AES-256, or failed the + handshake outright, with nothing logged either way. + """ + suite = auth.TLS12AuthenticationSuite() + context = ssl.SSLContext(suite.protocol) + context.set_ciphers(suite.ciphers) + + negotiable = {cipher['name'] for cipher in context.get_ciphers()} + self.assertIn('ECDHE-RSA-AES256-GCM-SHA384', negotiable) + self.assertIn('ECDHE-RSA-AES128-GCM-SHA256', negotiable) + def test_custom_ciphers(self): """ Test that providing a custom list of cipher suites yields the right