Merge branch '194-follow-domain-parameter-addition-to-elliptic-curves' (f7013564) · Commits · Szilárd Pfeiffer / CryptoLyzer

.pylintrc

+9 −0

Original line number	Diff line number	Diff line
		@@ -6,6 +6,15 @@ jobs=0

		load-plugins = pylint.extensions.no_self_use

		[BASIC]

		good-names=setUp,tearDown,setUpClass,tearDownClass

		[DESIGN]

		max-args=6
		max-positional-arguments=6

		[FORMAT]

		max-line-length=120

cryptolyzer/common/crypto.py

+28 −44

Original line number	Diff line number	Diff line
		@@ -25,9 +25,9 @@ import Crypto.Hash.SHA512
		import Crypto.Protocol.DH
		import Crypto.PublicKey.ECC

		from cryptodatahub.common.algorithm import BlockCipher, BlockCipherMode, Hash, MAC, NamedGroup, NamedGroupType
		from cryptodatahub.common.algorithm import BlockCipher, BlockCipherMode, Hash, MAC
		from cryptodatahub.common.exception import InvalidValue
		from cryptodatahub.common.parameter import DHParamWellKnown
		from cryptodatahub.common.parameter import DHParamWellKnown, ECParamWellKnown

		from cryptoparser.common.exception import NotEnoughData, TooMuchData
		from cryptoparser.common.parse import ComposerBinary
		@@ -72,19 +72,19 @@ class EphemeralKeyExchangeBase(abc.ABC):
		class EphemeralKeyExchangeEllipticCurve(EphemeralKeyExchangeBase):
		"""Abstract base class for elliptic-curve ephemeral key agreement."""

		named_group: NamedGroup = attr.ib(validator=attr.validators.instance_of(NamedGroup))
		key_parameter: ECParamWellKnown = attr.ib(validator=attr.validators.instance_of(ECParamWellKnown))

		@classmethod
		@abc.abstractmethod
		def supported_named_groups(cls) -> tuple[NamedGroup, ...]:
		def supported_key_parameters(cls) -> tuple[ECParamWellKnown, ...]:
		"""Named groups supported by this implementation."""
		raise NotImplementedError()

		@staticmethod
		def _montgomery_subject_public_key_info_der(named_group: NamedGroup, raw_bytes: bytes \| bytearray) -> bytes:
		def _montgomery_subject_public_key_info_der(key_parameter: ECParamWellKnown, raw_bytes: bytes \| bytearray) -> bytes:
		"""Wrap raw Montgomery-curve public key octets in a SubjectPublicKeyInfo DER structure."""
		algorithm = asn1crypto.keys.PublicKeyAlgorithm()
		algorithm['algorithm'] = asn1crypto.keys.PublicKeyAlgorithmId(named_group.value.oid)
		algorithm['algorithm'] = asn1crypto.keys.PublicKeyAlgorithmId(key_parameter.value.oid)
		public_key_info = asn1crypto.keys.PublicKeyInfo()
		public_key_info['algorithm'] = algorithm
		public_key_info['public_key'] = bytes(raw_bytes)
		@@ -95,67 +95,51 @@ class EphemeralKeyExchangeEllipticCurve(EphemeralKeyExchangeBase):
		class EphemeralKeyExchangeFiniteField(EphemeralKeyExchangeBase):
		"""Abstract base class for finite-field ephemeral key agreement."""

		_NAMED_GROUP_TO_DH_PARAMETERS: typing.ClassVar[dict[NamedGroup, DHParamWellKnown]] = {
		NamedGroup.FFDHE2048: DHParamWellKnown.RFC7919_2048_BIT_FINITE_FIELD_DIFFIE_HELLMAN_GROUP,
		NamedGroup.FFDHE3072: DHParamWellKnown.RFC7919_3072_BIT_FINITE_FIELD_DIFFIE_HELLMAN_GROUP,
		NamedGroup.FFDHE4096: DHParamWellKnown.RFC7919_4096_BIT_FINITE_FIELD_DIFFIE_HELLMAN_GROUP,
		NamedGroup.FFDHE6144: DHParamWellKnown.RFC7919_6144_BIT_FINITE_FIELD_DIFFIE_HELLMAN_GROUP,
		NamedGroup.FFDHE8192: DHParamWellKnown.RFC7919_8192_BIT_FINITE_FIELD_DIFFIE_HELLMAN_GROUP,
		}

		dh_parameters: DHParamWellKnown = attr.ib(validator=attr.validators.instance_of(DHParamWellKnown))

		@classmethod
		def from_named_group(cls, named_group: NamedGroup) -> EphemeralKeyExchangeFiniteField:
		if named_group.value.group_type != NamedGroupType.FINITE_FIELD:
		raise ValueError(f'{named_group.value.name} is not a finite-field NamedGroup')

		dh_parameters = cls._NAMED_GROUP_TO_DH_PARAMETERS.get(named_group)
		if dh_parameters is None:
		raise NotImplementedError(named_group)
		return cls(dh_parameters=dh_parameters)


		@attr.s
		class EphemeralKeyExchangeEllipticCurveCryptodome(EphemeralKeyExchangeEllipticCurve):
		"""Class for elliptic-curve ephemeral key agreement supported by PyCryptodome."""

		_NAMED_GROUP_TO_CURVE_NAME: typing.ClassVar[dict[NamedGroup, str]] = {
		NamedGroup.PRIME192V1: 'p192',
		NamedGroup.SECP224R1: 'p224',
		NamedGroup.PRIME256V1: 'p256',
		NamedGroup.SECP384R1: 'p384',
		NamedGroup.SECP521R1: 'p521',
		NamedGroup.CURVE25519: 'curve25519',
		NamedGroup.CURVE448: 'curve448',
		_CURVE_NAME_BY_KEY_PARAMETER: typing.ClassVar[dict[ECParamWellKnown, str]] = {
		ECParamWellKnown.PRIME192V1: 'p192',
		ECParamWellKnown.SECP224R1: 'p224',
		ECParamWellKnown.PRIME256V1: 'p256',
		ECParamWellKnown.SECP384R1: 'p384',
		ECParamWellKnown.SECP521R1: 'p521',
		ECParamWellKnown.CURVE25519: 'curve25519',
		ECParamWellKnown.CURVE448: 'curve448',
		}

		@classmethod
		def supported_named_groups(cls) -> tuple[NamedGroup, ...]:
		return tuple(cls._NAMED_GROUP_TO_CURVE_NAME)
		def supported_key_parameters(cls) -> tuple[ECParamWellKnown, ...]:
		return tuple(cls._CURVE_NAME_BY_KEY_PARAMETER)

		def _is_group_supported(self) -> None:
		if self.named_group not in self._NAMED_GROUP_TO_CURVE_NAME:
		raise InvalidValue(self.named_group, NamedGroup)
		if self.key_parameter not in self._CURVE_NAME_BY_KEY_PARAMETER:
		raise InvalidValue(self.key_parameter, ECParamWellKnown)

		def generate_key_pair(self) -> None:
		curve_name = self._NAMED_GROUP_TO_CURVE_NAME[self.named_group]
		curve_name = self._CURVE_NAME_BY_KEY_PARAMETER[self.key_parameter]
		self._private_key = Crypto.PublicKey.ECC.generate(curve=curve_name)

		def _import_peer_elliptic_curve_public_key(self, key_bytes: bytes \| bytearray) -> Crypto.PublicKey.ECC.EccKey:
		key_bytes = bytes(key_bytes)

		if self.named_group in (NamedGroup.CURVE25519, NamedGroup.CURVE448):
		if self.key_parameter in (ECParamWellKnown.CURVE25519, ECParamWellKnown.CURVE448):
		return Crypto.PublicKey.ECC.import_key(
		EphemeralKeyExchangeEllipticCurve._montgomery_subject_public_key_info_der(self.named_group, key_bytes)
		EphemeralKeyExchangeEllipticCurve._montgomery_subject_public_key_info_der(self.key_parameter, key_bytes)
		)

		return Crypto.PublicKey.ECC.import_key(key_bytes, curve_name=self._NAMED_GROUP_TO_CURVE_NAME[self.named_group])
		return Crypto.PublicKey.ECC.import_key(
		key_bytes, curve_name=self._CURVE_NAME_BY_KEY_PARAMETER[self.key_parameter]
		)

		@property
		def public_key_bytes(self) -> bytes:
		public_key = self._private_key.public_key()
		if self.named_group in (NamedGroup.CURVE25519, NamedGroup.CURVE448):
		if self.key_parameter in (ECParamWellKnown.CURVE25519, ECParamWellKnown.CURVE448):
		return public_key.export_key(format='raw')
		return public_key.export_key(format='SEC1', compress=False)

		@@ -177,20 +161,20 @@ class EphemeralKeyExchangeFiniteFieldCryptodome(EphemeralKeyExchangeFiniteField)

		def generate_key_pair(self) -> None:
		params = self.dh_parameters.value
		random_bytes = os.urandom(params.key_size // 8)
		random_bytes = os.urandom(params.field_size // 8)
		self._private_key = int.from_bytes(random_bytes, 'big') % (params.parameter_numbers.p - 2) + 2
		self._public_key = pow(params.parameter_numbers.g, self._private_key, params.parameter_numbers.p)

		@property
		def public_key_bytes(self) -> bytes:
		params = self.dh_parameters.value
		return self._public_key.to_bytes(params.key_size // 8, 'big')
		return self._public_key.to_bytes(params.field_size // 8, 'big')

		def compute_shared_secret(self, peer_public_bytes: bytes \| bytearray) -> bytes:
		params = self.dh_parameters.value
		peer_public_key_integer = int.from_bytes(bytes(peer_public_bytes), 'big')
		shared_secret = pow(peer_public_key_integer, self._private_key, params.parameter_numbers.p)
		return shared_secret.to_bytes(params.key_size // 8, 'big')
		return shared_secret.to_bytes(params.field_size // 8, 'big')


		@attr.s

cryptolyzer/common/dhparam.py

+24 −17

Original line number	Diff line number	Diff line
		@@ -7,10 +7,15 @@ import codecs
		import collections
		import attr

		from cryptodatahub.common.exception import InvalidValue
		from cryptodatahub.common.algorithm import KeyExchange, NamedGroup
		from cryptodatahub.common.algorithm import KeyExchange
		from cryptodatahub.common.grade import AttackNamed
		from cryptodatahub.common.key import convert_public_key_size, PublicKeySize
		from cryptodatahub.common.parameter import DHParameterNumbers, DHParamWellKnown, ECParamWellKnown
		from cryptodatahub.common.parameter import (
		DHParameterNumbers,
		DHParamWellKnown,
		ECParameterForm,
		ECParamWellKnown,
		)

		from cryptoparser.common.base import Vector, VectorParamNumeric, Serializable
		from cryptoparser.common.parse import ParserBinary
		@@ -63,32 +68,31 @@ def get_dh_ephemeral_key_forged(prime):
		return secrets.randbelow(prime - 3) + 2


		def get_ecdh_ephemeral_key_forged(named_group, add_point_format_octet=True):
		"""Return a forged ECDH ephemeral public key for ``named_group``.
		def get_ecdh_ephemeral_key_forged(key_parameter, add_point_format_octet=True):
		"""Return a forged ECDH ephemeral public key for ``key_parameter``.

		Only the uncompressed point format (SEC1 / X9.62) is supported for
		Weierstrass curves; X25519 and X448 always use raw key bytes (RFC 7748).
		Weierstrass curves; a Montgomery curve always uses raw key bytes, as X25519
		and X448 do (RFC 7748), because only the u coordinate is transmitted.

		:param add_point_format_octet: When True, the returned bytes for a
		Weierstrass curve include the leading 0x04 point-format octet that
		marks an uncompressed point. X25519 and X448 are unaffected.
		marks an uncompressed point. A Montgomery curve is unaffected.
		"""
		key_size_in_bytes = int(math.ceil(named_group.value.size / 8))
		if not isinstance(key_parameter, ECParamWellKnown):
		raise NotImplementedError(key_parameter)

		if named_group in [NamedGroup.CURVE25519, NamedGroup.CURVE448]:
		return key_size_in_bytes * b'\xff'
		key_size_in_bytes = int(math.ceil(key_parameter.value.field_size / 8))

		try:
		well_know_ec_param = ECParamWellKnown.from_named_group(named_group)
		except InvalidValue as e:
		raise NotImplementedError(named_group) from e
		if key_parameter.value.parameter_form == ECParameterForm.MONTGOMERY:
		return key_size_in_bytes * b'\xff'

		parts = []
		if add_point_format_octet:
		parts.append(b'\x04')
		parts.extend([
		int_to_bytes(well_know_ec_param.value.parameter_numbers.x, key_size_in_bytes),
		int_to_bytes(well_know_ec_param.value.parameter_numbers.y, key_size_in_bytes),
		int_to_bytes(key_parameter.value.parameter_numbers.x, key_size_in_bytes),
		int_to_bytes(key_parameter.value.parameter_numbers.y, key_size_in_bytes),
		])
		return bytearray().join(parts)

		@@ -156,7 +160,10 @@ class DHParameter(Serializable):
		if self.parameter_numbers == well_know_public_number.value.parameter_numbers:
		self.well_known = well_know_public_number
		self.prime = True
		self.safe_prime = well_know_public_number.value.safe_prime
		self.safe_prime = not any(
		vulnerability.named == AttackNamed.NON_SAFE_PRIME
		for vulnerability in well_know_public_number.value.vulnerabilities
		)
		break
		else:
		self.well_known = None

cryptolyzer/ike/ciphers.py

+21 −17

Original line number	Diff line number	Diff line
		@@ -9,10 +9,13 @@ from cryptodatahub.common.algorithm import (
		BlockCipherMode,
		Hash,
		MAC,
		NamedGroup,
		NamedGroupType,
		)
		from cryptodatahub.common.parameter import DHParamWellKnown
		from cryptodatahub.common.pqc import (
		HybridNamedGroup,
		KeyEncapsulationMechanism,
		KeyParameter,
		validate_key_parameter,
		)
		from cryptodatahub.ike.algorithm import (
		Ikev1AttributeType,
		Ikev1EncryptionAlgorithm,
		@@ -64,8 +67,8 @@ class Ikev1CipherSuite:
		block_cipher_mode: typing.Optional[BlockCipherMode] = attr.ib(
		validator=attr.validators.optional(attr.validators.instance_of(BlockCipherMode))
		)
		diffie_hellman_group: typing.Union[NamedGroup, DHParamWellKnown, str] = attr.ib(
		validator=attr.validators.instance_of((NamedGroup, DHParamWellKnown))
		diffie_hellman_group: KeyParameter = attr.ib(
		validator=validate_key_parameter()
		)
		hash_algorithm: Hash = attr.ib(
		validator=attr.validators.instance_of(Hash)
		@@ -112,8 +115,8 @@ class Ikev2CipherSuite:
		pseudorandom_function: MAC = attr.ib(
		validator=attr.validators.instance_of(MAC)
		)
		diffie_hellman_group: typing.Union[NamedGroup, DHParamWellKnown] = attr.ib(
		validator=attr.validators.instance_of((NamedGroup, DHParamWellKnown))
		diffie_hellman_group: KeyParameter = attr.ib(
		validator=validate_key_parameter()
		)
		integrity_algorithm: typing.Optional[MAC] = attr.ib(
		default=None,
		@@ -185,9 +188,9 @@ class AnalyzerResultIkev1Ciphers(AnalyzerResultIKE):
		hash_algorithms: list[Hash] = attr.ib(
		validator=attr.validators.deep_iterable(attr.validators.instance_of(Hash))
		)
		diffie_hellman_groups: list[typing.Union[NamedGroup, DHParamWellKnown]] = attr.ib(
		diffie_hellman_groups: list[KeyParameter] = attr.ib(
		validator=attr.validators.deep_iterable(
		attr.validators.instance_of((NamedGroup, DHParamWellKnown))
		validate_key_parameter()
		)
		)

		@@ -207,9 +210,9 @@ class AnalyzerResultIkev2Ciphers(AnalyzerResultIKE):
		integrity_algorithms: list[MAC] = attr.ib(
		validator=attr.validators.deep_iterable(attr.validators.instance_of(MAC))
		)
		diffie_hellman_groups: list[typing.Union[NamedGroup, DHParamWellKnown]] = attr.ib(
		diffie_hellman_groups: list[KeyParameter] = attr.ib(
		validator=attr.validators.deep_iterable(
		attr.validators.instance_of((NamedGroup, DHParamWellKnown))
		validate_key_parameter()
		)
		)

		@@ -600,14 +603,15 @@ class AnalyzerCiphers(AnalyzerIKEBase):
		cipher_suites: list[Ikev2CipherSuite] = []
		accepted_dh_groups: set[Ikev2DiffieHellmanGroup] = set()

		# RFC 9370 hybrid post-quantum KEMs (ML-KEM) negotiate via
		# IKE_INTERMEDIATE after IKE_SA_INIT, not as classical DH transforms
		# in the initial exchange. Skip them here — the SA_INIT KE payload
		# has no defined encoding for HYBRID_PQS public keys.
		# RFC 9370 post-quantum KEMs (ML-KEM) negotiate via IKE_INTERMEDIATE
		# after IKE_SA_INIT, not as classical DH transforms in the initial
		# exchange. Skip them here — the SA_INIT KE payload has no defined
		# encoding for the public key of a post-quantum or hybrid group.
		classical_dh_groups = [
		dh for dh in Ikev2DiffieHellmanGroup
		if not (isinstance(dh.value.key_parameter, NamedGroup)
		and dh.value.key_parameter.value.group_type == NamedGroupType.HYBRID_PQS)
		if not isinstance(
		dh.value.key_parameter, (KeyEncapsulationMechanism, HybridNamedGroup)
		)
		]

		if non_aead_encr_tuples:

cryptolyzer/ike/client.py

+4 −4

Original line number	Diff line number	Diff line
		@@ -9,7 +9,7 @@ import typing

		import attr

		from cryptodatahub.common.algorithm import NamedGroup
		from cryptodatahub.common.parameter import ECParamWellKnown
		from cryptodatahub.common.parameter import DHParamWellKnown
		from cryptodatahub.common.exception import InvalidValue
		from cryptodatahub.ike.algorithm import (
		@@ -108,7 +108,7 @@ class IkeSecurityAssociationBase(IsakmpMessage):
		]
		ecdh_groups = [
		dh_group for dh_group in dh_groups
		if isinstance(dh_group.value.key_parameter, NamedGroup)
		if isinstance(dh_group.value.key_parameter, ECParamWellKnown)
		and (backend_supported is None or dh_group in backend_supported)
		]
		return ffdh_groups, ecdh_groups
		@@ -133,12 +133,12 @@ class IkeSecurityAssociationBase(IsakmpMessage):
		if ephemeral is not None:
		return dhe_ephemeral_material_backend.key_exchange_wire_bytes(dh_group, ephemeral)
		key_parameter = dh_group.value.key_parameter
		if isinstance(key_parameter, NamedGroup):
		if isinstance(key_parameter, ECParamWellKnown):
		return get_ecdh_ephemeral_key_forged(key_parameter, add_point_format_octet=False)
		if isinstance(key_parameter, DHParamWellKnown):
		return int_to_bytes(
		get_dh_ephemeral_key_forged(key_parameter.value.parameter_numbers.p),
		key_parameter.value.key_size // 8,
		key_parameter.value.field_size // 8,
		)
		return None