[#680]Plugin for Extended Master Secret support

Author: nabla-c0d3

sslyze/mozilla_tls_profile/mozilla_config_checker.py

@@ -89,10 +89,12 @@ class ServerScanResultIncomplete(Exception):
     ScanCommand.HEARTBLEED,
     ScanCommand.ROBOT,
     ScanCommand.OPENSSL_CCS_INJECTION,
+    ScanCommand.TLS_FALLBACK_SCSV,
     ScanCommand.TLS_COMPRESSION,
     ScanCommand.SESSION_RENEGOTIATION,
     ScanCommand.CERTIFICATE_INFO,
     ScanCommand.ELLIPTIC_CURVES,
+    ScanCommand.TLS_EXTENDED_MASTER_SECRET,
     # ScanCommand.HTTP_HEADERS,  # Disabled for now; see below
 }
 
@@ -202,6 +204,12 @@ def _check_tls_vulnerabilities(scan_result: AllScanCommandsAttempts) -> Dict[str
             "Server is vulnerable to the OpenSSL CCS injection attack."
         )
 
+    assert scan_result.tls_fallback_scsv.result
+    if not scan_result.tls_fallback_scsv.result.supports_fallback_scsv:
+        issues_with_tls_vulns["tls_vulnerability_fallback_scsv"] = (
+            "Server is vulnerable to TLS downgrade attacks because it does not support the TLS_FALLBACK_SCSV mechanism."
+        )
+
     assert scan_result.heartbleed.result
     if scan_result.heartbleed.result.is_vulnerable_to_heartbleed:
         issues_with_tls_vulns["tls_vulnerability_heartbleed"] = "Server is vulnerable to the OpenSSL Heartbleed attack."
@@ -216,6 +224,12 @@ def _check_tls_vulnerabilities(scan_result: AllScanCommandsAttempts) -> Dict[str
             "Server is vulnerable to the insecure renegotiation attack."
         )
 
+    assert scan_result.tls_extended_master_secret.result
+    if not scan_result.tls_extended_master_secret.result.supports_ems_extension:
+        issues_with_tls_vulns["tls_vulnerability_extended_master_secret"] = (
+            "Server does not support the Extended Master Secret TLS extension."
+        )
+
     return issues_with_tls_vulns
 
 

sslyze/plugins/ems_extension_plugin.py

@@ -0,0 +1,100 @@
+from dataclasses import dataclass
+from typing import List, Optional
+
+from nassl.ssl_client import SslClient, ExtendedMasterSecretSupportEnum
+
+from sslyze.json.pydantic_utils import BaseModelWithOrmModeAndForbid
+from sslyze.json.scan_attempt_json import ScanCommandAttemptAsJson
+from sslyze.plugins.plugin_base import (
+    ScanCommandResult,
+    ScanCommandImplementation,
+    ScanCommandExtraArgument,
+    ScanJob,
+    ScanCommandWrongUsageError,
+    ScanCommandCliConnector,
+    ScanJobResult,
+)
+from sslyze.server_connectivity import ServerConnectivityInfo, TlsVersionEnum
+
+
+@dataclass(frozen=True)
+class EmsExtensionScanResult(ScanCommandResult):
+    """The result of testing a server for TLS Extended Master Secret extension support.
+
+    Attributes:
+        supports_ems_extension: True if the server supports the TLS Extended Master Secret extension.
+    """
+
+    supports_ems_extension: bool
+
+
+class EmsExtensionScanResultAsJson(BaseModelWithOrmModeAndForbid):
+    supports_ems_extension: bool
+
+
+class EmsExtensionScanAttemptAsJson(ScanCommandAttemptAsJson):
+    result: Optional[EmsExtensionScanResultAsJson]
+
+
+class _EmsExtensionCliConnector(ScanCommandCliConnector[EmsExtensionScanResult, None]):
+    _cli_option = "ems"
+    _cli_description = "Test a server for TLS Extended Master Secret extension support."
+
+    @classmethod
+    def result_to_console_output(cls, result: EmsExtensionScanResult) -> List[str]:
+        result_as_txt = [cls._format_title("TLS Extended Master Secret Extension")]
+        downgrade_txt = "OK - Supported" if result.supports_ems_extension else "VULNERABLE - EMS not supported"
+        result_as_txt.append(cls._format_field("", downgrade_txt))
+        return result_as_txt
+
+
+class EmsExtensionImplementation(ScanCommandImplementation[EmsExtensionScanResult, None]):
+    """Test a server for TLS Extended Master Secret extension support."""
+
+    cli_connector_cls = _EmsExtensionCliConnector
+
+    @classmethod
+    def scan_jobs_for_scan_command(
+        cls, server_info: ServerConnectivityInfo, extra_arguments: Optional[ScanCommandExtraArgument] = None
+    ) -> List[ScanJob]:
+        if extra_arguments:
+            raise ScanCommandWrongUsageError("This plugin does not take extra arguments")
+
+        return [ScanJob(function_to_call=_test_ems, function_arguments=[server_info])]
+
+    @classmethod
+    def result_for_completed_scan_jobs(
+        cls, server_info: ServerConnectivityInfo, scan_job_results: List[ScanJobResult]
+    ) -> EmsExtensionScanResult:
+        if len(scan_job_results) != 1:
+            raise RuntimeError(f"Unexpected number of scan jobs received: {scan_job_results}")
+
+        return EmsExtensionScanResult(supports_ems_extension=scan_job_results[0].get_result())
+
+
+def _test_ems(server_info: ServerConnectivityInfo) -> bool:
+    # The Extended Master Secret extension is not relevant to TLS 1.3 and later
+    if server_info.tls_probing_result.highest_tls_version_supported.value >= TlsVersionEnum.TLS_1_3.value:
+        return True
+
+    ssl_connection = server_info.get_preconfigured_tls_connection(
+        # Only the modern client has EMS support
+        should_use_legacy_openssl=False,
+    )
+    if not isinstance(ssl_connection.ssl_client, SslClient):
+        raise RuntimeError("Should never happen")
+
+    # Perform the SSL handshake
+    try:
+        ssl_connection.connect()
+        ems_support_enum = ssl_connection.ssl_client.get_extended_master_secret_support()
+    finally:
+        ssl_connection.close()
+
+    # Return the result
+    if ems_support_enum == ExtendedMasterSecretSupportEnum.NOT_USED_IN_CURRENT_SESSION:
+        return False
+    elif ems_support_enum == ExtendedMasterSecretSupportEnum.USED_IN_CURRENT_SESSION:
+        return True
+    else:
+        raise ValueError("Could not determine Extended Master Secret Extension support")

sslyze/plugins/scan_commands.py

@@ -7,6 +7,7 @@
 from sslyze.plugins.certificate_info.implementation import CertificateInfoImplementation
 from sslyze.plugins.compression_plugin import CompressionImplementation
 from sslyze.plugins.early_data_plugin import EarlyDataImplementation
+from sslyze.plugins.ems_extension_plugin import EmsExtensionImplementation
 from sslyze.plugins.fallback_scsv_plugin import FallbackScsvImplementation
 from sslyze.plugins.heartbleed_plugin import HeartbleedImplementation
 from sslyze.plugins.http_headers_plugin import HttpHeadersImplementation
@@ -45,6 +46,7 @@ class ScanCommand(str, Enum):
     SESSION_RENEGOTIATION = "session_renegotiation"
     HTTP_HEADERS = "http_headers"
     ELLIPTIC_CURVES = "elliptic_curves"
+    TLS_EXTENDED_MASTER_SECRET = "tls_extended_master_secret"
 
 
 class ScanCommandsRepository:
@@ -75,4 +77,5 @@ def get_all_scan_commands() -> Set[ScanCommand]:
     ScanCommand.SESSION_RENEGOTIATION: SessionRenegotiationImplementation,
     ScanCommand.HTTP_HEADERS: HttpHeadersImplementation,
     ScanCommand.ELLIPTIC_CURVES: SupportedEllipticCurvesImplementation,
+    ScanCommand.TLS_EXTENDED_MASTER_SECRET: EmsExtensionImplementation,
 }

sslyze/scanner/models.py

@@ -11,6 +11,7 @@
 from sslyze.plugins.certificate_info.implementation import CertificateInfoScanResult, CertificateInfoExtraArgument
 from sslyze.plugins.compression_plugin import CompressionScanResult
 from sslyze.plugins.early_data_plugin import EarlyDataScanResult
+from sslyze.plugins.ems_extension_plugin import EmsExtensionScanResult
 from sslyze.plugins.fallback_scsv_plugin import FallbackScsvScanResult
 from sslyze.plugins.heartbleed_plugin import HeartbleedScanResult
 from sslyze.plugins.http_headers_plugin import HttpHeadersScanResult
@@ -132,6 +133,10 @@ class SupportedEllipticCurvesScanAttempt(ScanCommandAttempt[SupportedEllipticCur
     pass
 
 
+class EmsExtensionScanAttempt(ScanCommandAttempt[EmsExtensionScanResult]):
+    pass
+
+
 @dataclass(frozen=True)
 class AllScanCommandsAttempts:
     """The result of every scan command supported by SSLyze."""
@@ -153,6 +158,7 @@ class AllScanCommandsAttempts:
     session_resumption: SessionResumptionSupportScanAttempt
     elliptic_curves: SupportedEllipticCurvesScanAttempt
     http_headers: HttpHeadersScanAttempt
+    tls_extended_master_secret: EmsExtensionScanAttempt
 
 
 _SCAN_CMD_FIELD_NAME_TO_CLS: dict[str, Type[ScanCommandAttempt]] = {

tests/plugins_tests/test_ems_extension_plugin.py

@@ -0,0 +1,86 @@
+from nassl.ssl_client import ClientCertificateRequested
+
+from sslyze.plugins.ems_extension_plugin import (
+    EmsExtensionImplementation,
+    EmsExtensionScanResult,
+    EmsExtensionScanResultAsJson,
+)
+
+from sslyze.server_setting import (
+    ServerNetworkLocation,
+    ServerNetworkConfiguration,
+    ClientAuthenticationCredentials,
+)
+from tests.connectivity_utils import check_connectivity_to_server_and_return_info
+from tests.markers import can_only_run_on_linux_64
+from tests.openssl_server import LegacyOpenSslServer, ClientAuthConfigEnum
+import pytest
+
+
+class TestFallbackScsvPlugin:
+    def test_good(self) -> None:
+        # Given a server that supports Extended Master Secret
+        server_location = ServerNetworkLocation("www.google.com", 443)
+        server_info = check_connectivity_to_server_and_return_info(server_location)
+
+        # When testing for EMS support, it succeeds with the expected result
+        result: EmsExtensionScanResult = EmsExtensionImplementation.scan_server(server_info)
+        assert result.supports_ems_extension
+
+        # And a CLI output can be generated
+        assert EmsExtensionImplementation.cli_connector_cls.result_to_console_output(result)
+
+        # And the result can be converted to JSON
+        result_as_json = EmsExtensionScanResultAsJson.model_validate(result).model_dump_json()
+        assert result_as_json
+
+    @can_only_run_on_linux_64
+    def test_bad(self) -> None:
+        # Given a server that does NOT support EMS
+        with LegacyOpenSslServer() as server:
+            server_location = ServerNetworkLocation(
+                hostname=server.hostname, ip_address=server.ip_address, port=server.port
+            )
+            server_info = check_connectivity_to_server_and_return_info(server_location)
+
+            # When testing for EMS, it succeeds
+            result: EmsExtensionScanResult = EmsExtensionImplementation.scan_server(server_info)
+
+        # And the server is reported as NOT supporting it
+        assert not result.supports_ems_extension
+
+    @can_only_run_on_linux_64
+    def test_fails_when_client_auth_failed(self) -> None:
+        # Given a server that does NOT support EMS and that requires client authentication
+        with LegacyOpenSslServer(client_auth_config=ClientAuthConfigEnum.REQUIRED) as server:
+            # And sslyze does NOT provide a client certificate
+            server_location = ServerNetworkLocation(
+                hostname=server.hostname, ip_address=server.ip_address, port=server.port
+            )
+            server_info = check_connectivity_to_server_and_return_info(server_location)
+
+            # When testing, it fails as a client cert was not supplied
+            with pytest.raises(ClientCertificateRequested):
+                EmsExtensionImplementation.scan_server(server_info)
+
+    @can_only_run_on_linux_64
+    def test_works_when_client_auth_succeeded(self) -> None:
+        # Given a server that does NOT support EMS and that requires client authentication
+        with LegacyOpenSslServer(client_auth_config=ClientAuthConfigEnum.REQUIRED) as server:
+            server_location = ServerNetworkLocation(
+                hostname=server.hostname, ip_address=server.ip_address, port=server.port
+            )
+            # And sslyze provides a client certificate
+            network_config = ServerNetworkConfiguration(
+                tls_server_name_indication=server.hostname,
+                tls_client_auth_credentials=ClientAuthenticationCredentials(
+                    certificate_chain_path=server.get_client_certificate_path(), key_path=server.get_client_key_path()
+                ),
+            )
+            server_info = check_connectivity_to_server_and_return_info(server_location, network_config)
+
+            # When testing for EMS, it succeeds
+            result: EmsExtensionScanResult = EmsExtensionImplementation.scan_server(server_info)
+
+        # And the server is reported as NOT supporting EMS
+        assert not result.supports_ems_extension