Merge pull request #123 from nabla-c0d3/pr-120

New method get_ems_support()
nabla-c0d3 · Dec 25, 2024 · 9f5a6a0 · 9f5a6a0
2 parents 8ccc613 + f9c6199
commit 9f5a6a0
Show file tree

Hide file tree

Showing 3 changed files with 70 additions and 0 deletions.
diff --git a/nassl/_nassl/nassl_SSL.c b/nassl/_nassl/nassl_SSL.c
@@ -304,6 +304,12 @@ static PyObject* nassl_SSL_set1_groups(nassl_SSL_Object *self, PyObject *args)
     PyMem_Free(listOfNids);
     Py_RETURN_NONE;
 }
+
+static PyObject *nassl_SSL_get_extms_support(nassl_SSL_Object *self)
+{
+    long returnValue = SSL_get_extms_support(self->ssl);
+    return Py_BuildValue("l", returnValue);
+}
 #endif
 
 static PyObject* nassl_SSL_shutdown(nassl_SSL_Object *self, PyObject *args)
@@ -1187,6 +1193,9 @@ static PyMethodDef nassl_SSL_Object_methods[] =
     {"set1_groups", (PyCFunction)nassl_SSL_set1_groups, METH_VARARGS,
     "OpenSSL's SSL_set1_groups()"
     },
+    {"get_extms_support", (PyCFunction)nassl_SSL_get_extms_support, METH_NOARGS,
+    "OpenSSL's SSL_get_extms_support()."
+    },
 #endif
     {"get_peer_cert_chain", (PyCFunction)nassl_SSL_get_peer_cert_chain, METH_NOARGS,
      "OpenSSL's SSL_get_peer_cert_chain(). Returns an array of _nassl.X509 objects."

diff --git a/nassl/ssl_client.py b/nassl/ssl_client.py
@@ -418,6 +418,12 @@ class OpenSslEarlyDataStatusEnum(IntEnum):
     ACCEPTED = 2
 
 
+class ExtendedMasterSecretSupportEnum(IntEnum):
+    NOT_USED_IN_CURRENT_SESSION = 0
+    USED_IN_CURRENT_SESSION = 1
+    UNKNOWN = -1
+
+
 class SslClient(BaseSslClient):
     """High level API implementing an SSL client.
 
@@ -465,3 +471,15 @@ def get_verified_chain(self) -> List[str]:
             raise CertificateChainVerificationFailed(verify_code)
 
         return [x509.as_pem() for x509 in self._ssl.get0_verified_chain()]
+
+    def get_extended_master_secret_support(self) -> ExtendedMasterSecretSupportEnum:
+        """Indicates whether the current session used extended master secret."""
+        support = self._ssl.get_extms_support()
+        if support == 1:
+            return ExtendedMasterSecretSupportEnum.USED_IN_CURRENT_SESSION
+        elif support == 0:
+            return ExtendedMasterSecretSupportEnum.NOT_USED_IN_CURRENT_SESSION
+        elif support == -1:
+            return ExtendedMasterSecretSupportEnum.UNKNOWN
+        else:
+            raise ValueError(f"Unexpected return value get_extms_support(): {support}")
diff --git a/tests/ssl_client_test.py b/tests/ssl_client_test.py
@@ -7,6 +7,7 @@
 from nassl.legacy_ssl_client import LegacySslClient
 from nassl.ssl_client import (
     ClientCertificateRequested,
+    ExtendedMasterSecretSupportEnum,
     OpenSslVersionEnum,
     OpenSslVerifyEnum,
     SslClient,
@@ -359,6 +360,48 @@ def test_set_groups_curve_x448(self):
             assert dh_info.size == 448
             assert len(dh_info.public_bytes) == 56
 
+    def test_get_extended_master_secret_not_used(self):
+        with LegacyOpenSslServer() as server:
+            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+            sock.settimeout(5)
+            sock.connect((server.hostname, server.port))
+
+            ssl_client = SslClient(
+                ssl_version=OpenSslVersionEnum.TLSV1_2,
+                underlying_socket=sock,
+                ssl_verify=OpenSslVerifyEnum.NONE,
+            )
+            exms_support_before_handshake = ssl_client.get_extended_master_secret_support()
+            assert exms_support_before_handshake == ExtendedMasterSecretSupportEnum.UNKNOWN
+
+            try:
+                ssl_client.do_handshake()
+            finally:
+                ssl_client.shutdown()
+
+            exms_support = ssl_client.get_extended_master_secret_support()
+            assert exms_support == ExtendedMasterSecretSupportEnum.NOT_USED_IN_CURRENT_SESSION
+
+    def test_get_extended_master_secret_used(self):
+        with ModernOpenSslServer() as server:
+            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+            sock.settimeout(5)
+            sock.connect((server.hostname, server.port))
+
+            ssl_client = SslClient(
+                ssl_version=OpenSslVersionEnum.TLSV1_2,
+                underlying_socket=sock,
+                ssl_verify=OpenSslVerifyEnum.NONE,
+            )
+
+            try:
+                ssl_client.do_handshake()
+            finally:
+                ssl_client.shutdown()
+
+            exms_support = ssl_client.get_extended_master_secret_support()
+            assert exms_support == ExtendedMasterSecretSupportEnum.USED_IN_CURRENT_SESSION
+
 
 class TestLegacySslClientOnline:
     def test_ssl_2(self):