Expose set_groups functions

This allows pyOpenSSL to restrict the groups allowed to be used. E.g.
for restricting the groups to post-quantum hybrid groups
(e.g. X25519MLKEM768) to always ensure that post-quantum cryptography is
used.

This commit uses the set_groups as public API since that is the
preferred naming. Internally for the API we use the set_curves API
since this name is available on all OpenSSL implementations and
OpenSSL forks.

Signed-off-by: Arne Schwabe <arne@rfc2549.org>

Author: schwabe

CHANGELOG.rst

@@ -18,6 +18,7 @@ Deprecations:
 
 Changes:
 ^^^^^^^^
+- Added ``OpenSSL.SSL.Context.set_groups`` and ``OpenSSL.SSL.Connection.set_groups`` to set allowed groups/curves.
 
 - Added ``OpenSSL.SSL.Connection.get_group_name`` to determine which group name was negotiated.
 

src/OpenSSL/SSL.py

@@ -1515,6 +1515,19 @@ def set_tls13_ciphersuites(self, ciphersuites: bytes) -> None:
             _lib.SSL_CTX_set_ciphersuites(self._context, ciphersuites) == 1
         )
 
+    @_require_not_used
+    def set_groups(self, groups: bytes):
+        """
+        Set the supported groups/curves in this SSL Session.
+        """
+        if not isinstance(groups, bytes):
+            raise TypeError("groups must be a byte string.")
+
+        # We use the newer name (groups) in our public API and
+        # use the legacy/more compatible name in the internal API
+        rc = _lib.SSL_CTX_set1_curves_list(self._context, groups)
+        _openssl_assert(rc == 1)
+
     @_require_not_used
     def set_client_ca_list(
         self, certificate_authorities: Sequence[X509Name]
@@ -3227,6 +3240,18 @@ def get_group_name(self) -> str | None:
 
         return _ffi.string(group_name).decode("utf-8")
 
+    def set_groups(self, groups: bytes):
+        """
+        Set the supported groups/curves in this SSL Session.
+        """
+        if not isinstance(groups, bytes):
+            raise TypeError("groups must be a byte string.")
+
+        # We use the newer name (groups) in our public API and
+        # use the legacy/more compatible name in the internal API
+        rc = _lib.SSL_set1_curves_list(self._ssl, groups)
+        _openssl_assert(rc == 1)
+
     def request_ocsp(self) -> None:
         """
         Called to request that the server sends stapled OCSP data, if

tests/test_ssl.py

@@ -3495,6 +3495,100 @@ def test_get_group_name(self) -> None:
 
         assert server_group_name == client_group_name
 
+    @pytest.mark.skipif(
+        not getattr(_lib, "Cryptography_HAS_SSL_GET0_GROUP_NAME", None),
+        reason="SSL_get0_group_name unavailable",
+    )
+    def test_set_groups_context(self) -> None:
+        """
+        `Context.set_groups` forces the use of a specific curve/groups list.
+        """
+
+        def loopback_x448_client_factory(
+            socket: socket, version: int = SSLv23_METHOD
+        ) -> Connection:
+            context = Context(version)
+            context.set_groups(b"X448")
+            client = Connection(context, socket)
+            client.set_connect_state()
+            return client
+
+        server, client = loopback(client_factory=loopback_x448_client_factory)
+        server_group_name = server.get_group_name()
+        client_group_name = client.get_group_name()
+
+        assert isinstance(server_group_name, str)
+        assert isinstance(client_group_name, str)
+
+        assert server_group_name.lower() == "x448"
+        assert client_group_name.lower() == "x448"
+
+    def loopback_x448_server_factory(
+        socket: socket, version: int = SSLv23_METHOD
+    ) -> Connection:
+        connection = loopback_server_factory(socket, version)
+        connection.set_groups(b"X448")
+        return connection
+
+    @pytest.mark.skipif(
+        not getattr(_lib, "Cryptography_HAS_SSL_GET0_GROUP_NAME", None),
+        reason="SSL_get0_group_name unavailable",
+    )
+    def test_set_groups_session(self) -> None:
+        """
+        `Connection.set_groups` forces the use of a specific curve/groups list.
+        """
+
+        def loopback_x448_server_factory(
+            socket: socket, version: int = SSLv23_METHOD
+        ) -> Connection:
+            connection = loopback_server_factory(socket, version)
+            connection.set_groups(b"X448")
+            return connection
+
+        server, client = loopback(server_factory=loopback_x448_server_factory)
+        server_group_name = server.get_group_name()
+        client_group_name = client.get_group_name()
+
+        assert isinstance(server_group_name, str)
+        assert isinstance(client_group_name, str)
+
+        assert server_group_name.lower() == "x448"
+        assert client_group_name.lower() == "x448"
+
+    def test_set_groups_mismatch(self):
+        """
+        Forces different group lists on client and server so that a connection
+        should not be possible.
+        """
+
+        def loopback_x25519_client_factory(
+            socket: socket, version: int = SSLv23_METHOD
+        ) -> Connection:
+            connection = loopback_client_factory(socket, version)
+            connection.set_groups(b"X25519")
+            return connection
+
+        def loopback_x448_server_factory(
+            socket: socket, version: int = SSLv23_METHOD
+        ) -> Connection:
+            ctx = Context(version)
+            ctx.set_groups(b"X448")
+
+            ctx.use_privatekey(load_privatekey(FILETYPE_PEM, server_key_pem))
+            ctx.use_certificate(
+                load_certificate(FILETYPE_PEM, server_cert_pem)
+            )
+            server = Connection(ctx, socket)
+            server.set_accept_state()
+            return server
+
+        with pytest.raises(SSL.Error):
+            loopback(
+                client_factory=loopback_x25519_client_factory,
+                server_factory=loopback_x448_server_factory,
+            )
+
     def test_wantReadError(self) -> None:
         """
         `Connection.bio_read` raises `OpenSSL.SSL.WantReadError` if there are