Merge branch '177-ike-extensions' (f30a53c6) · Commits · Szilárd Pfeiffer / CryptoLyzer

CHANGELOG.rst

+13 −0

Original line number	Diff line number	Diff line
		@@ -2,6 +2,19 @@
		Changelog
		=========

		----------
		Unreleased
		----------

		Features
		========

		- IKE (``ike``)

		- Extensions (``extensions``)

		- add checker extensions checker (#177)

		------------------
		1.3.0 - 2026-06-15
		------------------

cryptolyzer/ike/all.py

+19 −0

Original line number	Diff line number	Diff line
		@@ -11,6 +11,11 @@ from cryptolyzer.ike.ciphers import (
		AnalyzerResultIkev1Ciphers,
		AnalyzerResultIkev2Ciphers,
		)
		from cryptolyzer.ike.extensions import (
		AnalyzerExtensions,
		AnalyzerResultIkev1Extensions,
		AnalyzerResultIkev2Extensions,
		)
		from cryptolyzer.ike.versions import AnalyzerVersions, AnalyzerResultVersions


		@@ -21,6 +26,7 @@ class AnalyzerResultAll(AnalyzerResultIKE): # pylint: disable=too-few-public-me

		:param versions: supported protocol versions (IKEv1/IKEv2).
		:param ciphers: the supported transforms.
		:param extensions: detected IKE extensions advertised during SA setup.
		"""

		versions: typing.Optional[AnalyzerResultVersions] = attr.ib(
		@@ -33,6 +39,12 @@ class AnalyzerResultAll(AnalyzerResultIKE): # pylint: disable=too-few-public-me
		),
		metadata={'human_readable_name': 'Supported Cipher Suites'}
		)
		extensions: typing.Optional[typing.Union[AnalyzerResultIkev1Extensions, AnalyzerResultIkev2Extensions]] = attr.ib(
		validator=attr.validators.optional(
		attr.validators.instance_of((AnalyzerResultIkev1Extensions, AnalyzerResultIkev2Extensions))
		),
		metadata={'human_readable_name': 'Extensions'}
		)


		class AnalyzerAll(AnalyzerIKEBase):
		@@ -62,6 +74,7 @@ class AnalyzerAll(AnalyzerIKEBase):
		super().analyze(analyzable, protocol_version)
		versions = None
		ciphers = None
		extensions = None

		try:
		versions = AnalyzerVersions().analyze(analyzable, protocol_version)
		@@ -73,8 +86,14 @@ class AnalyzerAll(AnalyzerIKEBase):
		except Exception: # pylint: disable=broad-except
		pass

		try:
		extensions = AnalyzerExtensions().analyze(analyzable, protocol_version)
		except Exception: # pylint: disable=broad-except
		pass

		return AnalyzerResultAll(
		target=AnalyzerTargetIke.from_l7_client(analyzable),
		versions=versions,
		ciphers=ciphers,
		extensions=extensions,
		)

cryptolyzer/ike/analyzer.py

+2 −0

Original line number	Diff line number	Diff line
		@@ -7,6 +7,7 @@ from cryptolyzer.common.analyzer import ProtocolHandlerIKEBase, ProtocolHandlerI
		from cryptolyzer.ike.ciphers import AnalyzerCiphers
		from cryptolyzer.ike.dhparams import AnalyzerDHParams
		from cryptolyzer.ike.curves import AnalyzerCurves
		from cryptolyzer.ike.extensions import AnalyzerExtensions
		from cryptolyzer.ike.versions import AnalyzerVersions


		@@ -17,6 +18,7 @@ class ProtocolHandlerIKEv1(ProtocolHandlerIKEExactVersion):
		AnalyzerCiphers,
		AnalyzerDHParams,
		AnalyzerCurves,
		AnalyzerExtensions,
		)

		@classmethod

cryptolyzer/ike/client.py

+26 −22

Original line number	Diff line number	Diff line
		@@ -59,6 +59,7 @@ from cryptoparser.ike.ikev2 import (
		Ikev2PayloadKeyExchange,
		Ikev2NotifyPayloadCookie,
		Ikev2PayloadNonce,
		Ikev2PayloadNotifyBase,
		Ikev2PayloadFlags,
		Ikev2PayloadSecurityAssociation,
		Ikev2PayloadDelete,
		@@ -218,6 +219,7 @@ class Ikev2SecurityAssociationBase(IsakmpMessage):
		cookie: typing.Optional[typing.Union[bytes, bytearray]] = None,
		nonce: typing.Optional[typing.Union[bytes, bytearray]] = None,
		key_exchange_dh_group: typing.Optional[Ikev2DiffieHellmanGroup] = None,
		extra_notify_payloads: typing.Optional[typing.Iterable[Ikev2PayloadNotifyBase]] = None,
		): # pylint: disable=too-many-arguments,too-many-positional-arguments
		payloads = []

		@@ -247,11 +249,10 @@ class Ikev2SecurityAssociationBase(IsakmpMessage):
		ecdh_groups=ecdh_groups,
		ffdh_groups=ffdh_groups,
		)
		payload_security_association = Ikev2PayloadSecurityAssociation(
		payloads.append(Ikev2PayloadSecurityAssociation(
		flags=set([Ikev2PayloadFlags.CRITICAL, ]),
		proposals=proposals
		)
		payloads.append(payload_security_association)
		))

		# Caller decides which DH group to key the KE payload for. Falls back
		# to the legacy heuristic when no explicit choice is given: prefer the
		@@ -297,6 +298,9 @@ class Ikev2SecurityAssociationBase(IsakmpMessage):
		nonce_data=nonce,
		))

		if extra_notify_payloads is not None:
		payloads.extend(extra_notify_payloads)

		return payloads


		@@ -335,7 +339,7 @@ class Ikev2SecurityAssociationSpecialization(Ikev2SecurityAssociationBase):


		class Ikev2SecurityAssociationAnyAlgorithm(Ikev2SecurityAssociationBase):
		def __init__(self, cookie=None):
		def __init__(self, cookie=None, extra_notify_payloads=None, key_exchange_dh_group=None):
		payloads = self._get_payloads(
		encryption_algorithm_tuples=self.expand_encryption_algorithms_to_tuples(
		Ikev2EncryptionAlgorithm
		@@ -344,6 +348,8 @@ class Ikev2SecurityAssociationAnyAlgorithm(Ikev2SecurityAssociationBase):
		pseudorandom_functions=list(Ikev2PseudorandomFunction),
		integrity_algorithms=list(Ikev2IntegrityAlgorithm),
		cookie=cookie,
		key_exchange_dh_group=key_exchange_dh_group,
		extra_notify_payloads=extra_notify_payloads,
		)

		initiator_spi = random.randint(0, 2**64 - 1)
		@@ -648,15 +654,14 @@ class IKEv2ClientHandshake(IKEClient):

		@classmethod
		def _process_non_handshake_message(cls, message):
		try:
		payload = message.get_payload_by_type(Ikev2PayloadType.NOTIFY)
		notify_type = payload.type
		if notify_type == Ikev2NotifyType.COOKIE:
		raise IsakmpNotify(notify_type, payload)
		if notify_type.value.level == Ikev2NotifyLevel.ERROR:
		raise IsakmpNotify(notify_type, payload)
		except KeyError as e:
		raise IsakmpNotify(Ikev2NotifyType.INVALID_SYNTAX) from e
		notifies = message.get_payloads_by_type(Ikev2PayloadType.NOTIFY)
		if not notifies:
		raise IsakmpNotify(Ikev2NotifyType.INVALID_SYNTAX)
		for payload in notifies:
		if payload.type == Ikev2NotifyType.COOKIE:
		raise IsakmpNotify(payload.type, payload)
		if payload.type.value.level == Ikev2NotifyLevel.ERROR:
		raise IsakmpNotify(payload.type, payload)

		@classmethod
		def _process_invalid_message(cls, transfer):
		@@ -774,15 +779,14 @@ class IKEv1ClientHandshake(IKEClient):

		@classmethod
		def _process_non_handshake_message(cls, message):
		try:
		payload = message.get_payload_by_type(Ikev1PayloadType.NOTIFICATION)
		notify_type = payload.notify_type
		if notify_type == Ikev1NotifyType.NO_PROPOSAL_CHOSEN:
		raise IsakmpNotify(notify_type)
		if notify_type.value.level == Ikev1NotifyLevel.ERROR:
		raise IsakmpNotify(notify_type)
		except KeyError as e:
		raise IsakmpNotify(Ikev1NotifyType.SITUATION_NOT_SUPPORTED) from e
		notifies = message.get_payloads_by_type(Ikev1PayloadType.NOTIFICATION)
		if not notifies:
		raise IsakmpNotify(Ikev1NotifyType.SITUATION_NOT_SUPPORTED)
		for payload in notifies:
		if payload.notify_type == Ikev1NotifyType.NO_PROPOSAL_CHOSEN:
		raise IsakmpNotify(payload.notify_type)
		if payload.notify_type.value.level == Ikev1NotifyLevel.ERROR:
		raise IsakmpNotify(payload.notify_type)

		@classmethod
		def _process_invalid_message(cls, transfer):

cryptolyzer/ike/common.py

+20 −13

Original line number	Diff line number	Diff line
		@@ -58,9 +58,13 @@ class Ikev1CipherSuite:
		cls,
		algorithms: Ikev1SecurityAssociationProposalAlgorithms
		):
		for bulk_cipher_entry in algorithms.encryption_algorithm.value.bulk_ciphers:
		bulk_ciphers = list(algorithms.encryption_algorithm.value.bulk_ciphers)
		for bulk_cipher_entry in bulk_ciphers:
		if bulk_cipher_entry.cipher.value.key_size == algorithms.key_length:
		break
		else:
		if len(bulk_ciphers) == 1 and algorithms.key_length is None:
		bulk_cipher_entry = bulk_ciphers[0]
		else:
		raise ValueError(
		f'Key length {algorithms.key_length} not found for '
		@@ -107,16 +111,20 @@ class Ikev2CipherSuite:
		diffie_hellman_transform_id: Ikev2DiffieHellmanGroup,
		key_length: int,
		): # pylint: disable=too-many-arguments,too-many-positional-arguments
		for bulk_cipher_entry in encryption_transform_id.value.bulk_ciphers:
		bulk_ciphers = list(encryption_transform_id.value.bulk_ciphers)
		for bulk_cipher_entry in bulk_ciphers:
		if bulk_cipher_entry.cipher.value.key_size == key_length:
		break
		else:
		if len(bulk_ciphers) == 1 and key_length is None:
		bulk_cipher_entry = bulk_ciphers[0]
		else:
		raise ValueError(
		f'Key length {key_length} not found for '
		f'encryption algorithm {encryption_transform_id}'
		)

		integrity_algorithm = None if integrity_transform_id.value.hmac is None else integrity_transform_id.value.hmac
		integrity_algorithm = integrity_transform_id.value.hmac
		return cls(
		encryption_algorithm=bulk_cipher_entry.cipher,
		block_cipher_mode=encryption_transform_id.value.block_cipher_mode,
		@@ -209,8 +217,7 @@ class AnalyzerIKECommonBase(AnalyzerIKEBase):
		group_name = self._get_dh_group_name()
		LogSingleton().log(
		level=40,
		msg=f'No proposal chosen; group_type={group_name}, '
		f'group={group_name}'
		msg=f'No proposal chosen; group_type={group_name}'
		)
		return None