Loading cryptolyzer/tls/client.py +4 −2 Changes for cryptolyzer/tls/client.py: 4 added lines, 2 removed lines. Original line number Diff line number Diff line Loading @@ -1521,6 +1521,8 @@ class TlsClient: class TlsClientHandshake(TlsClient): _TLS13_ENCRYPTED_HANDSHAKE_MESSAGE_TYPES = (TlsHandshakeType.CERTIFICATE, TlsHandshakeType.CERTIFICATE_VERIFY) def __init__(self): self.server_messages = {} self._tls13_record_decryptor = None Loading @@ -1530,7 +1532,7 @@ class TlsClientHandshake(TlsClient): def _tls13_try_init_handshake_decryptor( # pylint: disable=too-many-return-statements self, server_hello, last_handshake_message_type, ): if last_handshake_message_type != TlsHandshakeType.CERTIFICATE: if last_handshake_message_type not in self._TLS13_ENCRYPTED_HANDSHAKE_MESSAGE_TYPES: return if server_hello.random == TLS_HANDSHAKE_HELLO_RETRY_REQUEST_RANDOM: return Loading Loading @@ -1618,7 +1620,7 @@ class TlsClientHandshake(TlsClient): if last_handshake_message_type is None and content_type == TlsContentType.CHANGE_CIPHER_SPEC: raise StopIteration if ( last_handshake_message_type == TlsHandshakeType.CERTIFICATE and last_handshake_message_type in self._TLS13_ENCRYPTED_HANDSHAKE_MESSAGE_TYPES and content_type == TlsContentType.CHANGE_CIPHER_SPEC and self._tls13_record_decryptor is not None ): Loading cryptolyzer/tls/sigalgos.py +90 −6 Changes for cryptolyzer/tls/sigalgos.py: 90 added lines, 6 removed lines. Original line number Diff line number Diff line Loading @@ -5,8 +5,14 @@ import attr from cryptodatahub.tls.algorithm import TlsSignatureAndHashAlgorithm from cryptoparser.tls.extension import TlsExtensionSignatureAlgorithms, TlsSignatureAndHashAlgorithmVector from cryptoparser.tls.ciphersuite import TlsCipherSuite from cryptoparser.tls.extension import ( TlsExtensionSignatureAlgorithms, TlsNamedCurve, TlsSignatureAndHashAlgorithmVector, ) from cryptoparser.tls.subprotocol import TlsAlertDescription, TlsHandshakeType from cryptoparser.tls.version import TlsProtocolVersion, TlsVersion from cryptolyzer.common.analyzer import AnalyzerTlsBase from cryptolyzer.common.exception import NetworkError, NetworkErrorType, SecurityError Loading @@ -18,7 +24,9 @@ from cryptolyzer.tls.client import ( TlsHandshakeClientHelloAuthenticationECDSA, TlsHandshakeClientHelloAuthenticationGOST, TlsHandshakeClientHelloAuthenticationRSA, TlsHandshakeClientHelloSpecalization, ) from cryptolyzer.tls.crypto import dhe_ephemeral_material_backend from cryptolyzer.tls.exception import TlsAlert, UnexpectedAlertError Loading @@ -37,6 +45,11 @@ class AnalyzerResultSigAlgos(AnalyzerResultTls): class AnalyzerSigAlgos(AnalyzerTlsBase): _SIGNATURE_ALGORITHM_REJECTION_ALERTS = AnalyzerTlsBase._ACCEPTABLE_HANDSHAKE_FAILURE_ALERTS + [ TlsAlertDescription.DECODE_ERROR, TlsAlertDescription.INTERNAL_ERROR, ] @classmethod def get_name(cls): return 'sigalgos' Loading @@ -62,11 +75,7 @@ class AnalyzerSigAlgos(AnalyzerTlsBase): if e.description in [TlsAlertDescription.PROTOCOL_VERSION, TlsAlertDescription.UNRECOGNIZED_NAME]: break acceptable_alerts = AnalyzerTlsBase._ACCEPTABLE_HANDSHAKE_FAILURE_ALERTS + [ TlsAlertDescription.DECODE_ERROR, TlsAlertDescription.INTERNAL_ERROR, ] if e.description not in acceptable_alerts: if e.description not in self._SIGNATURE_ALGORITHM_REJECTION_ALERTS: raise UnexpectedAlertError(e.description) from e except NetworkError as e: if e.error != NetworkErrorType.NO_RESPONSE: Loading @@ -79,8 +88,83 @@ class AnalyzerSigAlgos(AnalyzerTlsBase): return supported_algorithms @staticmethod def _get_algorithms_of_other_authentications(algorithms, supported_algorithms): supported_authentications = { supported_algorithm.value.signature_algorithm for supported_algorithm in supported_algorithms } return [ algorithm for algorithm in algorithms if algorithm.value.signature_algorithm not in supported_authentications ] def _analyze_algorithms_by_removing_selected(self, l7_client, protocol_version): backend_key_parameters = set(dhe_ephemeral_material_backend.supported_key_parameters()) key_share_extensions, key_exchange_by_named_curve = dhe_ephemeral_material_backend.build_tls13_key_shares([ named_curve for named_curve in TlsNamedCurve if named_curve.value.key_parameter in backend_key_parameters ]) client_hello = TlsHandshakeClientHelloSpecalization( l7_client.address, [protocol_version, ], list(TlsCipherSuite), named_curves=list(key_exchange_by_named_curve), signature_algorithms=None, extensions=key_share_extensions, ) extension = client_hello.extensions.get_item_by_type( TlsExtensionSignatureAlgorithms.get_extension_type() ) remaining_algorithms = list(extension.hash_and_signature_algorithms) supported_algorithms = [] while remaining_algorithms: extension.hash_and_signature_algorithms = TlsSignatureAndHashAlgorithmVector(remaining_algorithms) try: self._before_probe(l7_client) server_messages = l7_client.do_tls_handshake( client_hello, last_handshake_message_type=TlsHandshakeType.CERTIFICATE_VERIFY, tls13_handshake_protection=key_exchange_by_named_curve, ) except TlsAlert as e: if not supported_algorithms: break if e.description not in self._SIGNATURE_ALGORITHM_REJECTION_ALERTS: raise UnexpectedAlertError(e.description) from e remaining_algorithms_of_other_authentications = self._get_algorithms_of_other_authentications( remaining_algorithms, supported_algorithms ) if remaining_algorithms_of_other_authentications == remaining_algorithms: break remaining_algorithms = remaining_algorithms_of_other_authentications continue except SecurityError: break except NetworkError as e: if e.error != NetworkErrorType.NO_RESPONSE: raise e break algorithm = server_messages[TlsHandshakeType.CERTIFICATE_VERIFY].signature_algorithm if algorithm not in remaining_algorithms: break LogSingleton().log(level=60, msg=f'Server offers signature algorithm {algorithm.name}') supported_algorithms.append(algorithm) remaining_algorithms.remove(algorithm) return supported_algorithms def analyze(self, analyzable, protocol_version): super().analyze(analyzable, protocol_version) if protocol_version > TlsProtocolVersion(TlsVersion.TLS1_2): supported_algorithms = self._analyze_algorithms_by_removing_selected(analyzable, protocol_version) else: supported_algorithms = [] for client_hello_class in [ TlsHandshakeClientHelloAuthenticationRSA, Loading cryptoparser @ 2f4ce981 Compare 5631ce61 to 2f4ce981 Changes for submodules/cryptoparser: 1 added line, 1 removed line. Original line number Diff line number Diff line Subproject commit 5631ce6182f7bb7c0128cbc44b27df54a5f372de Subproject commit 2f4ce981cfbd37387daaa447d121741b5b96ca61 test/tls/data/openquantumsafe_assignments.json +1359 −742 File changed.Preview size limit exceeded, changes collapsed. Show changes test/tls/test_client.py +7 −3 Changes for test/tls/test_client.py: 7 added lines, 3 removed lines. Original line number Diff line number Diff line Loading @@ -419,6 +419,8 @@ class TestTls13TryInitHandshakeDecryptor(TestLoggerBase): ) def test_builds_decryptor_on_success(self): for last_handshake_message_type in [TlsHandshakeType.CERTIFICATE, TlsHandshakeType.CERTIFICATE_VERIFY]: with self.subTest(last_handshake_message_type=last_handshake_message_type): key_exchange = mock.Mock() key_exchange.compute_shared_secret.return_value = b'\x00' * 32 handshake = self._make_handshake(protection={TlsNamedCurve.SECP256R1: key_exchange}) Loading @@ -429,7 +431,7 @@ class TestTls13TryInitHandshakeDecryptor(TestLoggerBase): server_hello.cipher_suite = TlsCipherSuite.TLS_AES_128_GCM_SHA256 server_hello.compose.return_value = b'\x03\x04' server_hello.extensions = self._key_share_extensions(TlsNamedCurve.SECP256R1) handshake._tls13_try_init_handshake_decryptor(server_hello, TlsHandshakeType.CERTIFICATE) handshake._tls13_try_init_handshake_decryptor(server_hello, last_handshake_message_type) self.assertIsNotNone(handshake._tls13_record_decryptor) Loading Loading @@ -482,15 +484,17 @@ class TestTls13RecordDecryptError(unittest.TestCase): self.assertIn(TlsHandshakeType.SERVER_HELLO_DONE, handshake.server_messages) def test_change_cipher_spec_after_certificate_is_accepted(self): for last_handshake_message_type in [TlsHandshakeType.CERTIFICATE, TlsHandshakeType.CERTIFICATE_VERIFY]: with self.subTest(last_handshake_message_type=last_handshake_message_type): handshake = TlsClientHandshake() with self.assertRaises(TlsAlert): handshake._process_non_handshake_message( TlsContentType.CHANGE_CIPHER_SPEC, mock.Mock(), TlsHandshakeType.CERTIFICATE, TlsContentType.CHANGE_CIPHER_SPEC, mock.Mock(), last_handshake_message_type, ) handshake._tls13_record_decryptor = mock.Mock() handshake._process_non_handshake_message( TlsContentType.CHANGE_CIPHER_SPEC, mock.Mock(), TlsHandshakeType.CERTIFICATE, TlsContentType.CHANGE_CIPHER_SPEC, mock.Mock(), last_handshake_message_type, ) Loading Loading
cryptolyzer/tls/client.py +4 −2 Changes for cryptolyzer/tls/client.py: 4 added lines, 2 removed lines. Original line number Diff line number Diff line Loading @@ -1521,6 +1521,8 @@ class TlsClient: class TlsClientHandshake(TlsClient): _TLS13_ENCRYPTED_HANDSHAKE_MESSAGE_TYPES = (TlsHandshakeType.CERTIFICATE, TlsHandshakeType.CERTIFICATE_VERIFY) def __init__(self): self.server_messages = {} self._tls13_record_decryptor = None Loading @@ -1530,7 +1532,7 @@ class TlsClientHandshake(TlsClient): def _tls13_try_init_handshake_decryptor( # pylint: disable=too-many-return-statements self, server_hello, last_handshake_message_type, ): if last_handshake_message_type != TlsHandshakeType.CERTIFICATE: if last_handshake_message_type not in self._TLS13_ENCRYPTED_HANDSHAKE_MESSAGE_TYPES: return if server_hello.random == TLS_HANDSHAKE_HELLO_RETRY_REQUEST_RANDOM: return Loading Loading @@ -1618,7 +1620,7 @@ class TlsClientHandshake(TlsClient): if last_handshake_message_type is None and content_type == TlsContentType.CHANGE_CIPHER_SPEC: raise StopIteration if ( last_handshake_message_type == TlsHandshakeType.CERTIFICATE and last_handshake_message_type in self._TLS13_ENCRYPTED_HANDSHAKE_MESSAGE_TYPES and content_type == TlsContentType.CHANGE_CIPHER_SPEC and self._tls13_record_decryptor is not None ): Loading
cryptolyzer/tls/sigalgos.py +90 −6 Changes for cryptolyzer/tls/sigalgos.py: 90 added lines, 6 removed lines. Original line number Diff line number Diff line Loading @@ -5,8 +5,14 @@ import attr from cryptodatahub.tls.algorithm import TlsSignatureAndHashAlgorithm from cryptoparser.tls.extension import TlsExtensionSignatureAlgorithms, TlsSignatureAndHashAlgorithmVector from cryptoparser.tls.ciphersuite import TlsCipherSuite from cryptoparser.tls.extension import ( TlsExtensionSignatureAlgorithms, TlsNamedCurve, TlsSignatureAndHashAlgorithmVector, ) from cryptoparser.tls.subprotocol import TlsAlertDescription, TlsHandshakeType from cryptoparser.tls.version import TlsProtocolVersion, TlsVersion from cryptolyzer.common.analyzer import AnalyzerTlsBase from cryptolyzer.common.exception import NetworkError, NetworkErrorType, SecurityError Loading @@ -18,7 +24,9 @@ from cryptolyzer.tls.client import ( TlsHandshakeClientHelloAuthenticationECDSA, TlsHandshakeClientHelloAuthenticationGOST, TlsHandshakeClientHelloAuthenticationRSA, TlsHandshakeClientHelloSpecalization, ) from cryptolyzer.tls.crypto import dhe_ephemeral_material_backend from cryptolyzer.tls.exception import TlsAlert, UnexpectedAlertError Loading @@ -37,6 +45,11 @@ class AnalyzerResultSigAlgos(AnalyzerResultTls): class AnalyzerSigAlgos(AnalyzerTlsBase): _SIGNATURE_ALGORITHM_REJECTION_ALERTS = AnalyzerTlsBase._ACCEPTABLE_HANDSHAKE_FAILURE_ALERTS + [ TlsAlertDescription.DECODE_ERROR, TlsAlertDescription.INTERNAL_ERROR, ] @classmethod def get_name(cls): return 'sigalgos' Loading @@ -62,11 +75,7 @@ class AnalyzerSigAlgos(AnalyzerTlsBase): if e.description in [TlsAlertDescription.PROTOCOL_VERSION, TlsAlertDescription.UNRECOGNIZED_NAME]: break acceptable_alerts = AnalyzerTlsBase._ACCEPTABLE_HANDSHAKE_FAILURE_ALERTS + [ TlsAlertDescription.DECODE_ERROR, TlsAlertDescription.INTERNAL_ERROR, ] if e.description not in acceptable_alerts: if e.description not in self._SIGNATURE_ALGORITHM_REJECTION_ALERTS: raise UnexpectedAlertError(e.description) from e except NetworkError as e: if e.error != NetworkErrorType.NO_RESPONSE: Loading @@ -79,8 +88,83 @@ class AnalyzerSigAlgos(AnalyzerTlsBase): return supported_algorithms @staticmethod def _get_algorithms_of_other_authentications(algorithms, supported_algorithms): supported_authentications = { supported_algorithm.value.signature_algorithm for supported_algorithm in supported_algorithms } return [ algorithm for algorithm in algorithms if algorithm.value.signature_algorithm not in supported_authentications ] def _analyze_algorithms_by_removing_selected(self, l7_client, protocol_version): backend_key_parameters = set(dhe_ephemeral_material_backend.supported_key_parameters()) key_share_extensions, key_exchange_by_named_curve = dhe_ephemeral_material_backend.build_tls13_key_shares([ named_curve for named_curve in TlsNamedCurve if named_curve.value.key_parameter in backend_key_parameters ]) client_hello = TlsHandshakeClientHelloSpecalization( l7_client.address, [protocol_version, ], list(TlsCipherSuite), named_curves=list(key_exchange_by_named_curve), signature_algorithms=None, extensions=key_share_extensions, ) extension = client_hello.extensions.get_item_by_type( TlsExtensionSignatureAlgorithms.get_extension_type() ) remaining_algorithms = list(extension.hash_and_signature_algorithms) supported_algorithms = [] while remaining_algorithms: extension.hash_and_signature_algorithms = TlsSignatureAndHashAlgorithmVector(remaining_algorithms) try: self._before_probe(l7_client) server_messages = l7_client.do_tls_handshake( client_hello, last_handshake_message_type=TlsHandshakeType.CERTIFICATE_VERIFY, tls13_handshake_protection=key_exchange_by_named_curve, ) except TlsAlert as e: if not supported_algorithms: break if e.description not in self._SIGNATURE_ALGORITHM_REJECTION_ALERTS: raise UnexpectedAlertError(e.description) from e remaining_algorithms_of_other_authentications = self._get_algorithms_of_other_authentications( remaining_algorithms, supported_algorithms ) if remaining_algorithms_of_other_authentications == remaining_algorithms: break remaining_algorithms = remaining_algorithms_of_other_authentications continue except SecurityError: break except NetworkError as e: if e.error != NetworkErrorType.NO_RESPONSE: raise e break algorithm = server_messages[TlsHandshakeType.CERTIFICATE_VERIFY].signature_algorithm if algorithm not in remaining_algorithms: break LogSingleton().log(level=60, msg=f'Server offers signature algorithm {algorithm.name}') supported_algorithms.append(algorithm) remaining_algorithms.remove(algorithm) return supported_algorithms def analyze(self, analyzable, protocol_version): super().analyze(analyzable, protocol_version) if protocol_version > TlsProtocolVersion(TlsVersion.TLS1_2): supported_algorithms = self._analyze_algorithms_by_removing_selected(analyzable, protocol_version) else: supported_algorithms = [] for client_hello_class in [ TlsHandshakeClientHelloAuthenticationRSA, Loading
cryptoparser @ 2f4ce981 Compare 5631ce61 to 2f4ce981 Changes for submodules/cryptoparser: 1 added line, 1 removed line. Original line number Diff line number Diff line Subproject commit 5631ce6182f7bb7c0128cbc44b27df54a5f372de Subproject commit 2f4ce981cfbd37387daaa447d121741b5b96ca61
test/tls/data/openquantumsafe_assignments.json +1359 −742 File changed.Preview size limit exceeded, changes collapsed. Show changes
test/tls/test_client.py +7 −3 Changes for test/tls/test_client.py: 7 added lines, 3 removed lines. Original line number Diff line number Diff line Loading @@ -419,6 +419,8 @@ class TestTls13TryInitHandshakeDecryptor(TestLoggerBase): ) def test_builds_decryptor_on_success(self): for last_handshake_message_type in [TlsHandshakeType.CERTIFICATE, TlsHandshakeType.CERTIFICATE_VERIFY]: with self.subTest(last_handshake_message_type=last_handshake_message_type): key_exchange = mock.Mock() key_exchange.compute_shared_secret.return_value = b'\x00' * 32 handshake = self._make_handshake(protection={TlsNamedCurve.SECP256R1: key_exchange}) Loading @@ -429,7 +431,7 @@ class TestTls13TryInitHandshakeDecryptor(TestLoggerBase): server_hello.cipher_suite = TlsCipherSuite.TLS_AES_128_GCM_SHA256 server_hello.compose.return_value = b'\x03\x04' server_hello.extensions = self._key_share_extensions(TlsNamedCurve.SECP256R1) handshake._tls13_try_init_handshake_decryptor(server_hello, TlsHandshakeType.CERTIFICATE) handshake._tls13_try_init_handshake_decryptor(server_hello, last_handshake_message_type) self.assertIsNotNone(handshake._tls13_record_decryptor) Loading Loading @@ -482,15 +484,17 @@ class TestTls13RecordDecryptError(unittest.TestCase): self.assertIn(TlsHandshakeType.SERVER_HELLO_DONE, handshake.server_messages) def test_change_cipher_spec_after_certificate_is_accepted(self): for last_handshake_message_type in [TlsHandshakeType.CERTIFICATE, TlsHandshakeType.CERTIFICATE_VERIFY]: with self.subTest(last_handshake_message_type=last_handshake_message_type): handshake = TlsClientHandshake() with self.assertRaises(TlsAlert): handshake._process_non_handshake_message( TlsContentType.CHANGE_CIPHER_SPEC, mock.Mock(), TlsHandshakeType.CERTIFICATE, TlsContentType.CHANGE_CIPHER_SPEC, mock.Mock(), last_handshake_message_type, ) handshake._tls13_record_decryptor = mock.Mock() handshake._process_non_handshake_message( TlsContentType.CHANGE_CIPHER_SPEC, mock.Mock(), TlsHandshakeType.CERTIFICATE, TlsContentType.CHANGE_CIPHER_SPEC, mock.Mock(), last_handshake_message_type, ) Loading