Add Context.set_sigalgs_list() and Connection.get_sigalgs()

src/OpenSSL/SSL.py

@@ -1229,6 +1229,28 @@ def set_cipher_list(self, cipher_list):
                 ],
             )
 
+    def set_sigalgs_list(self, sigalgs_list):
+        """
+        Set the list of signature algorithms to be used in this context.
+
+        See the OpenSSL manual for more information (e.g.
+        :manpage:`SSL_CTX_set1_sigalgs_list(3)`).
+
+        :param bytes sigalgs_list: An OpenSSL signature algorithms list.
+        :return: None
+        """
+        sigalgs_list = _text_to_bytes_and_warn("sigalgs_list", sigalgs_list)
+
+        if not isinstance(sigalgs_list, bytes):
+            raise TypeError("sigalgs_list must be a byte string.")
+
+        if not _lib.Cryptography_HAS_SIGALGS:
+            return
+
+        _openssl_assert(
+            _lib.SSL_CTX_set1_sigalgs_list(self._context, sigalgs_list) == 1
+        )
+
     def set_client_ca_list(self, certificate_authorities):
         """
         Set the list of preferred client certificate signers for this server
@@ -2164,6 +2186,28 @@ def get_cipher_list(self):
             ciphers.append(_ffi.string(result).decode("utf-8"))
         return ciphers
 
+    def get_sigalgs(self):
+        """
+        Retrieve list of signature algorithms used by the Connection object.
+        Must be used after handshake only.
+        See :manpage:`SSL_get_sigalgs(3)`.
+
+        :return: A list of SignatureScheme (int) as defined by RFC 8446.
+        """
+        sigalgs = []
+        if not _lib.Cryptography_HAS_SIGALGS:
+            return sigalgs
+
+        rsign = _ffi.new("unsigned char *")
+        rhash = _ffi.new("unsigned char *")
+        for i in count():
+            result = _lib.SSL_get_sigalgs(self._ssl, i, _ffi.NULL, _ffi.NULL,
+                                          _ffi.NULL, rsign, rhash)
+            if result == 0:
+                break
+            sigalgs.append(rsign[0] + (rhash[0] << 8))
+        return sigalgs
+
     def get_client_ca_list(self):
         """
         Get CAs whose certificates are suggested for client authentication.

tests/test_ssl.py

@@ -532,6 +532,78 @@ def test_set_cipher_list_no_cipher_match(self, context):
             ),
         ]
 
+    @pytest.mark.parametrize("sigalgs_list", [
+        b"RSA+SHA256:RSA+SHA384",
+        u"RSA+SHA256:RSA+SHA384",
+    ])
+    def test_set_sigalgs_list(self, context, sigalgs_list):
+        """
+        `Context.set_sigalgs_list` accepts both byte and unicode strings
+        for naming the signature algorithms which connections created
+        with the context object will send to the server.
+        """
+        context.set_sigalgs_list(sigalgs_list)
+
+    def test_set_sigalgs_list_wrong_type(self, context):
+        """
+        `Context.set_cipher_list` raises `TypeError` when passed a non-string
+        argument.
+        """
+        with pytest.raises(TypeError):
+            context.set_sigalgs_list(object())
+
+    if _lib.Cryptography_HAS_SIGALGS:
+        def test_set_sigalgs_list_invalid_name(self, context):
+            """
+            `Context.set_cipher_list` raises `OpenSSL.SSL.Error` with a
+            `"no cipher match"` reason string regardless of the TLS
+            version.
+            """
+            with pytest.raises(Error):
+                context.set_sigalgs_list(b"imaginary-sigalg")
+
+        def test_set_sigalgs_list_not_supported(self):
+            """
+            If no signature algorithms supported by the server are set,
+            the handshake fails with a `"no suitable signature algorithm"`
+            reason string, or 'no shared cipher' on older OpenSSL releases.
+            """
+
+            def make_client(socket):
+                context = Context(TLSv1_2_METHOD)
+                context.set_sigalgs_list(b"ECDSA+SHA256:ECDSA+SHA384")
+                c = Connection(context, socket)
+                c.set_connect_state()
+                return c
+
+            with pytest.raises(Error):
+                loopback(client_factory=make_client)
+
+    def test_get_sigalgs(self):
+        """
+        `Connection.get_sigalgs` returns the signature algorithms send by
+        the client to the server. This is supported only in TLS1_2 and later.
+        """
+        def make_client(socket):
+            context = Context(TLSv1_2_METHOD)
+            context.set_sigalgs_list(b"RSA+SHA256:ECDSA+SHA384")
+            c = Connection(context, socket)
+            c.set_connect_state()
+            return c
+
+        srv, client = loopback(
+            server_factory=lambda s: loopback_server_factory(s,
+                                                             TLSv1_2_METHOD),
+            client_factory=make_client)
+
+        sigalgs = srv.get_sigalgs()
+        if _lib.Cryptography_HAS_SIGALGS:
+            assert 0x0401 in sigalgs  # rsa_pkcs1_sha256
+            assert 0x0503 in sigalgs  # ecdsa_secp384r1_sha384
+        else:
+            # gracefully degrades on older OpenSSL versions
+            assert len(sigalgs) == 0
+
     def test_load_client_ca(self, context, ca_file):
         """
         `Context.load_client_ca` works as far as we can tell.