From 95b383b5aadd7e8443062e236bd42a0da6ee42d9 Mon Sep 17 00:00:00 2001 From: Luca Dell'Oca Date: Fri, 3 Apr 2026 13:04:47 +0200 Subject: [PATCH 1/6] Add RSA GCM cipher suites to TLS12AuthenticationSuite for Veeam SECLEVEL=2 compatibility (cherry picked from commit cfaac4779675a3db3fa8fa8552ced7e7123db37d) --- kmip/services/auth.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/kmip/services/auth.py b/kmip/services/auth.py index a53af297..bec7119f 100644 --- a/kmip/services/auth.py +++ b/kmip/services/auth.py @@ -219,6 +219,8 @@ class TLS12AuthenticationSuite(AuthenticationSuite): 'ECDHE-ECDSA-AES256-GCM-SHA384', 'ECDHE-ECDSA-AES128-SHA256', 'ECDHE-ECDSA-AES256-SHA384' + 'ECDHE-RSA-AES256-GCM-SHA384', + 'ECDHE-RSA-AES128-GCM-SHA256' ] def __init__(self, cipher_suites=None): From 7043e443de07df86914ab1dfd6eefe30f232ff86 Mon Sep 17 00:00:00 2001 From: Luca Dell'Oca Date: Fri, 3 Apr 2026 13:12:25 +0200 Subject: [PATCH 2/6] Fix QUERY_OBJECTS empty response and implement Link attribute dynamic lookup (cherry picked from commit f7c1347bc063723aa186d76ab5b432f92f4ce8f5) --- kmip/services/server/engine.py | 28 +++++++++++++++++++++++++++- 1 file changed, 27 insertions(+), 1 deletion(-) diff --git a/kmip/services/server/engine.py b/kmip/services/server/engine.py index 43fdbe93..1cb725de 100644 --- a/kmip/services/server/engine.py +++ b/kmip/services/server/engine.py @@ -730,6 +730,26 @@ 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 + try: + linked = self._data_session.query(linked_type).filter_by( + initial_date=managed_object.initial_date, + _owner=managed_object._owner + ).first() + if linked: + return [{'link_type': link_type, + 'linked_object_identifier': str(linked.unique_identifier)}] + except Exception: + pass return None elif attr_name == "Application Specific Information": values = [] @@ -2856,7 +2876,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__ From e62fdb710e54071c107a1fb218af926cc85114ec Mon Sep 17 00:00:00 2001 From: Luca Dell'Oca Date: Fri, 3 Apr 2026 13:14:32 +0200 Subject: [PATCH 3/6] Implement Link attribute class with KMIP serialization for key pair linking (cherry picked from commit 2fc8fbaffad136c422a90f184174801130b95694) --- kmip/core/attributes.py | 98 +++++++++++++++++++++++++++++++++++++++++ 1 file changed, 98 insertions(+) diff --git a/kmip/core/attributes.py b/kmip/core/attributes.py index 843b0c44..d8cf2bba 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, six.string_types): + 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): From f4d4d2819159b5088c22157b0f37aff1ebfa5413 Mon Sep 17 00:00:00 2001 From: Luca Dell'Oca Date: Fri, 3 Apr 2026 13:16:55 +0200 Subject: [PATCH 4/6] Implement _create_link factory method and wire up Link attribute type (cherry picked from commit 8a63e04945e5939fc4ebfdc7b2ec0c7f5540927e) --- kmip/core/factories/attribute_values.py | 13 +++++++++++-- 1 file changed, 11 insertions(+), 2 deletions(-) diff --git a/kmip/core/factories/attribute_values.py b/kmip/core/factories/attribute_values.py index 1bc1e208..2981d34c 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( From e7d1be4e6d76cd201ca62a51fd04cef7a6256250 Mon Sep 17 00:00:00 2001 From: JonahMMay Date: Mon, 14 Sep 2026 14:24:28 +0000 Subject: [PATCH 5/6] Store the key-pair link instead of inferring it from owner and timestamp The Link attribute was recovered by querying for an object of the complementary type sharing the creating object's _owner and initial_date. initial_date is a whole-second epoch integer and both halves of a pair are given the same value, so two key pairs created by one owner inside the same second were indistinguishable and .first() returned an arbitrary one. A KMIP client that encrypts under public key A and later asks for A's linked private key could therefore be handed B's. Nothing fails at creation time; it surfaces as a decryption failure against data that is already written. Record the pairing on both objects once SQLAlchemy has assigned identifiers, which is the only moment the relationship is known for certain. The old heuristic is kept for rows created before the column existed, but it now refuses to guess when more than one candidate matches, and logs instead. Co-Authored-By: Claude Opus 5 (1M context) --- kmip/pie/objects.py | 11 ++ kmip/services/server/engine.py | 44 ++++++-- .../tests/unit/services/server/test_engine.py | 101 ++++++++++++++++++ 3 files changed, 146 insertions(+), 10 deletions(-) diff --git a/kmip/pie/objects.py b/kmip/pie/objects.py index e45746fa..e1a98113 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/server/engine.py b/kmip/services/server/engine.py index 1cb725de..ecd314f5 100644 --- a/kmip/services/server/engine.py +++ b/kmip/services/server/engine.py @@ -740,16 +740,32 @@ def _get_attribute_from_managed_object(self, managed_object, attr_name): link_type = enums.LinkType.PUBLIC_KEY_LINK else: return None - try: - linked = self._data_session.query(linked_type).filter_by( - initial_date=managed_object.initial_date, - _owner=managed_object._owner - ).first() - if linked: - return [{'link_type': link_type, - 'linked_object_identifier': str(linked.unique_identifier)}] - except Exception: - pass + 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 = [] @@ -1593,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 diff --git a/kmip/tests/unit/services/server/test_engine.py b/kmip/tests/unit/services/server/test_engine.py index 84deb436..ba57af12 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. From 37012f105796f49bc53a615c8346a35e1a1ab17c Mon Sep 17 00:00:00 2001 From: JonahMMay Date: Mon, 14 Sep 2026 14:24:28 +0000 Subject: [PATCH 6/6] Fix the Veeam interop patches and green the tests they broke MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A missing comma in TLS12AuthenticationSuite._default_cipher_suites made Python concatenate two entries into one unknown token, so ECDHE-RSA-AES256-GCM-SHA384 was never actually enabled. OpenSSL drops unrecognised tokens silently rather than raising, so a peer restricted to forward-secret GCM suites negotiated AES-128 instead, with nothing logged. The two duplicated ECDHE-ECDSA entries introduced alongside it are dropped. Link.linked_object_identifier used six.string_types, and 0.11.0 removed the Python 2 compatibility shims, so setting a string identifier raised NameError — on exactly the path a client takes to resolve a key pair. The Query and Link tests still asserted the old behaviour: an empty supported-object-type list (which a client reads as "this server supports nothing") and NotImplementedError from the Link factory. Adds a test that asserts the RSA GCM suites survive OpenSSL's own parsing. Asserting the cipher string cannot catch this: a malformed entry is still a string, which is how the missing comma went unnoticed. Co-Authored-By: Claude Opus 5 (1M context) --- kmip/core/attributes.py | 2 +- kmip/services/auth.py | 7 +++- .../core/factories/test_attribute_values.py | 31 ++++++++++++-- .../tests/unit/services/server/test_engine.py | 42 +++++++++++++++++-- kmip/tests/unit/services/test_auth.py | 25 ++++++++++- 5 files changed, 95 insertions(+), 12 deletions(-) diff --git a/kmip/core/attributes.py b/kmip/core/attributes.py index d8cf2bba..4390f8b8 100644 --- a/kmip/core/attributes.py +++ b/kmip/core/attributes.py @@ -1308,7 +1308,7 @@ def linked_object_identifier(self): def linked_object_identifier(self, value): if value is None: self._linked_object_identifier = None - elif isinstance(value, six.string_types): + elif isinstance(value, str): self._linked_object_identifier = primitives.TextString( value=value, tag=enums.Tags.LINKED_OBJECT_IDENTIFIER diff --git a/kmip/services/auth.py b/kmip/services/auth.py index bec7119f..ead6f286 100644 --- a/kmip/services/auth.py +++ b/kmip/services/auth.py @@ -217,8 +217,11 @@ 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' ] diff --git a/kmip/tests/unit/core/factories/test_attribute_values.py b/kmip/tests/unit/core/factories/test_attribute_values.py index 6edc4482..dd765183 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 ba57af12..558542dd 100644 --- a/kmip/tests/unit/services/server/test_engine.py +++ b/kmip/tests/unit/services/server/test_engine.py @@ -10041,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__), @@ -10129,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__), @@ -10237,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 92ec1f27..94c66f8b 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