Merge branch '184-decrease-the-number-of-live-test-to-improve-test-run-stability' (ebacb997) · Commits · Szilárd Pfeiffer / CryptoLyzer

cryptolyzer/common/application.py

+18 −0

Original line number	Diff line number	Diff line
		@@ -2,6 +2,7 @@
		# -- coding: utf-8 --

		import abc
		import time
		import typing

		import attr
		@@ -28,6 +29,11 @@ class L7ServerBase(L7TransferBase):
		default=None,
		validator=attr.validators.optional(attr.validators.instance_of(int))
		)
		max_idle_time: float = attr.ib(
		default=0.0,
		validator=attr.validators.instance_of((int, float))
		)
		stopped: bool = attr.ib(default=False, validator=attr.validators.instance_of(bool))

		@classmethod
		@abc.abstractmethod
		@@ -61,17 +67,29 @@ class L7ServerBase(L7TransferBase):
		def _do_handshakes(self, last_handshake_message_type):
		client_messages = []
		actual_handshake_count = 0
		idle_deadline = None
		while True:
		self.l4_transfer.close_client()

		if self.stopped:
		break

		if self.max_handshake_count is not None and actual_handshake_count >= self.max_handshake_count:
		break

		try:
		self.l4_transfer.accept()
		except NetworkError:
		if idle_deadline is None:
		idle_deadline = time.monotonic() + self.max_idle_time
		if time.monotonic() >= idle_deadline:
		break
		continue

		if self.stopped:
		break

		idle_deadline = None
		actual_handshake_count += 1
		client_messages.append(self._do_handshake(last_handshake_message_type))

cryptolyzer/common/dhparam.py

+2 −1

Original line number	Diff line number	Diff line
		@@ -2,6 +2,7 @@
		# -- coding: utf-8 --

		import math
		import secrets

		import codecs
		import collections
		@@ -60,7 +61,7 @@ def get_dh_public_key_from_bytes(p_bytes, g_bytes, y_bytes):


		def get_dh_ephemeral_key_forged(prime):
		return prime // 2 + 1
		return secrets.randbelow(prime - 3) + 2


		def get_ecdh_ephemeral_key_forged(named_group, add_point_format_octet=True):

cryptolyzer/common/x509.py

+41 −5

Original line number	Diff line number	Diff line
		@@ -2,6 +2,7 @@
		# -- coding: utf-8 --

		import collections
		import http.client

		import attr

		@@ -178,7 +179,7 @@ class CertificateChainX509Validator(): # pylint: disable=too-few-public-methods
		)
		try:
		build_path = cert_validator.validate_usage(set())
		except (certvalidator.errors.PathBuildingError, certvalidator.errors.PathValidationError):
		except (certvalidator.errors.PathBuildingError, certvalidator.errors.PathValidationError, OSError):
		self._validated.items = items
		else:
		self._validated.items = [PublicKeyX509(item) for item in reversed(build_path)]
		@@ -209,7 +210,7 @@ class CertificateChainX509Validator(): # pylint: disable=too-few-public-methods
		)
		try:
		cert_validator.validate_usage(set())
		except (certvalidator.errors.PathBuildingError, certvalidator.errors.PathValidationError):
		except (certvalidator.errors.PathBuildingError, certvalidator.errors.PathValidationError, OSError):
		trust_roots.append((trust_store_owner, False))
		else:
		trust_roots.append((trust_store_owner, True))
		@@ -217,12 +218,43 @@ class CertificateChainX509Validator(): # pylint: disable=too-few-public-methods
		self._validated.trust_roots = collections.OrderedDict(trust_roots)

		def check_revocation(self, certificate_status_list):
		asn1crypto_certificates = list(map(self._get_asn1crypto_certificate, self._validated.items))

		if certificate_status_list:
		context = certvalidator.context.ValidationContext(
		trust_roots=[],
		extra_trust_roots=[self._get_asn1crypto_certificate(self._validated.items[-1])],
		weak_hash_algos=set(),
		whitelisted_certs=[item.fingerprints[Hash.SHA1] for item in self._validated.items],
		ocsps=[certificate_status.ocsp_response for certificate_status in certificate_status_list],
		allow_fetching=False,
		)
		# NOTE: necessary only because of asn1crypto issue 262
		context._fetched_ocsps = { # pylint: disable=protected-access
		self._get_asn1crypto_certificate(item).issuer_serial: []
		for item in self._validated.items
		}
		cert_validator = certvalidator.CertificateValidator(
		end_entity_cert=asn1crypto_certificates[0],
		intermediate_certs=asn1crypto_certificates[1:],
		validation_context=context,
		)
		try:
		cert_validator.validate_usage(set())
		except certvalidator.errors.RevokedError:
		self._validated.revoked = True
		return
		except (certvalidator.errors.PathBuildingError, certvalidator.errors.PathValidationError, OSError):
		pass
		else:
		self._validated.revoked = False
		return

		context = certvalidator.context.ValidationContext(
		trust_roots=[],
		extra_trust_roots=[self._get_asn1crypto_certificate(self._validated.items[-1])],
		weak_hash_algos=set(),
		whitelisted_certs=[item.fingerprints[Hash.SHA1] for item in self._validated.items],
		allow_fetching=True,
		)
		# NOTE: necessary only because of asn1crypto issue 262
		@@ -230,7 +262,6 @@ class CertificateChainX509Validator(): # pylint: disable=too-few-public-methods
		self._get_asn1crypto_certificate(item).issuer_serial: []
		for item in self._validated.items
		}
		asn1crypto_certificates = list(map(self._get_asn1crypto_certificate, self._validated.items))
		cert_validator = certvalidator.CertificateValidator(
		end_entity_cert=asn1crypto_certificates[0],
		intermediate_certs=asn1crypto_certificates[1:],
		@@ -240,7 +271,12 @@ class CertificateChainX509Validator(): # pylint: disable=too-few-public-methods
		cert_validator.validate_usage(set())
		except certvalidator.errors.RevokedError:
		self._validated.revoked = True
		except (certvalidator.errors.PathBuildingError, certvalidator.errors.PathValidationError):
		except (
		certvalidator.errors.PathBuildingError,
		certvalidator.errors.PathValidationError,
		OSError,
		http.client.InvalidURL,
		):
		pass
		else:
		self._validated.revoked = False

cryptolyzer/httpx/content.py

+1 −1

Original line number	Diff line number	Diff line
		@@ -75,7 +75,7 @@ class HttpTagScriptIntegrity(HttpTagScriptBase):
		source_url_params = source_url._asdict()
		if source_url.host is None:
		source_url_params['host'] = html_url.host
		if source_url.host is None:
		source_url_params['port'] = html_url.port
		source_url_params['scheme'] = html_url.scheme

		http_fetcher = HttpFetcher()

cryptolyzer/tls/ciphers.py

+8 −10

Original line number	Diff line number	Diff line
		@@ -109,9 +109,8 @@ class AnalyzerCipherSuites(AnalyzerTlsBase):
		accepted_cipher_suites.append(cipher_suite)
		break

		@classmethod
		def _get_accepted_cipher_suites( # pylint: disable=too-many-arguments,too-many-positional-arguments
		cls, l7_client, protocol_version, checkable_cipher_suites,
		self, l7_client, protocol_version, checkable_cipher_suites,
		named_curves=None, key_share_curves=None,
		):
		retried_internal_error = False
		@@ -123,7 +122,8 @@ class AnalyzerCipherSuites(AnalyzerTlsBase):
		if retried_internal_error:
		time.sleep(1)

		cls._next_accepted_cipher_suites(
		self._before_probe(l7_client)
		self._next_accepted_cipher_suites(
		l7_client, protocol_version, remaining_cipher_suites, accepted_cipher_suites,
		named_curves=named_curves, key_share_curves=key_share_curves,
		)
		@@ -134,7 +134,7 @@ class AnalyzerCipherSuites(AnalyzerTlsBase):
		break
		except TlsAlert as e:
		try:
		return cls._handle_tls_alert(
		return self._handle_tls_alert(
		e, retried_internal_error, checkable_cipher_suites, remaining_cipher_suites
		)
		except StopIteration:
		@@ -155,14 +155,12 @@ class AnalyzerCipherSuites(AnalyzerTlsBase):

		return accepted_cipher_suites, remaining_cipher_suites

		@classmethod
		def _get_accepted_cipher_suites_all(cls, l7_client, protocol_version, checkable_cipher_suites):
		return cls._get_accepted_cipher_suites(
		def _get_accepted_cipher_suites_all(self, l7_client, protocol_version, checkable_cipher_suites):
		return self._get_accepted_cipher_suites(
		l7_client, protocol_version, checkable_cipher_suites
		)

		@classmethod
		def _get_accepted_cipher_suites_fallback(cls, l7_client, protocol_version):
		def _get_accepted_cipher_suites_fallback(self, l7_client, protocol_version):
		accepted_cipher_suites = []
		client_hello_messsages_in_order_of_probability = [
		TlsHandshakeClientHelloAuthenticationRSA(protocol_version, l7_client.address),
		@@ -180,7 +178,7 @@ class AnalyzerCipherSuites(AnalyzerTlsBase):
		)
		for client_hello in client_hello_messsages_in_order_of_probability:
		accepted_cipher_suites.extend(
		cls._get_accepted_cipher_suites(
		self._get_accepted_cipher_suites(
		l7_client, protocol_version, list(client_hello.cipher_suites),
		named_curves=getattr(client_hello, 'NAMED_CURVES', None),
		key_share_curves=getattr(client_hello, 'KEY_SHARE_CURVES', None),