Add support in SSL module for getting and setting TLS signature algorithms

Author: ronf

Doc/library/ssl.rst

@@ -1297,6 +1297,22 @@ SSL sockets also have the following additional methods and attributes:
 
    .. versionadded:: next
 
+.. method:: SSLSocket.client_sigalg()
+
+   Return the signature algorithm used for performing certificate-based client
+   authentication on this connection. If no connection has been established
+   or client authentication didn't occur, this method returns ``None``.
+
+   .. versionadded:: next
+
+.. method:: SSLSocket.server_sigalg()
+
+   Return the signature algorithm used by the server to complete the TLS
+   handshake on this connection. If no connection has been established
+   or the cipher suite has no signature, this method returns ``None``.
+
+   .. versionadded:: next
+
 .. method:: SSLSocket.compression()
 
    Return the compression algorithm being used as a string, or ``None``
@@ -1725,6 +1741,34 @@ to speed up repeated connections from the same clients.
 
    .. versionadded:: next
 
+.. method:: SSLContext.set_client_sigalgs(sigalgs)
+
+   Set the signature algorithms allowed for certificate-based client
+   authentication. It should be a string in the `OpenSSL sigalgs list format
+   <https://docs.openssl.org/master/man3/SSL_CTX_set1_client_sigalgs_list/>`_.
+
+   .. note::
+
+      When connected, the :meth:`SSLSocket.client_sigalg` method of SSL
+      sockets will return the signature algorithm used for performing
+      certificate-based client authentication on that connection.
+
+   .. versionadded:: next
+
+.. method:: SSLContext.set_server_sigalgs(sigalgs)
+
+   Set the signature algorithms allowed for the server to complete the TLS
+   handshake. It should be a string in the `OpenSSL sigalgs list format
+   <https://docs.openssl.org/master/man3/SSL_CTX_set1_sigalgs_list/>`_.
+
+   .. note::
+
+      When connected, the :meth:`SSLSocket.server_sigalg` method of SSL
+      sockets will return the signature algorithm used by the server to
+      complete the TLS handshake on that connection.
+
+   .. versionadded:: next
+
 .. method:: SSLContext.set_alpn_protocols(protocols)
 
    Specify which protocols the socket should advertise during the SSL/TLS

Doc/whatsnew/3.15.rst

@@ -433,6 +433,21 @@ ssl
   connection is made.
   (Contributed by Ron Frederick in :gh:`137197`.)
 
+* Added new methods for managing signature algorithms
+
+  * :meth:`ssl.SSLContext.set_client_sigalgs` sets the signature algorithms
+    allowed for certificate-based client authentication.
+  * :meth:`ssl.SSLContext.set_server_sigalgs` sets the signature algorithms
+    allowed for the server to complete the TLS handshake.
+   * :meth:`ssl.SSLSocket.client_sigalg` returns the signature algorithm
+     selected for client authentication on the current connection. This call
+     requires OpenSSL 3.5 or later.
+   * :meth:`ssl.SSLSocket.server_sigalg` returns the signature algorithm
+     selected for the server to complete the TLS handshake on the current
+     connection. This call requires OpenSSL 3.5 or later.
+
+   (Contributed by Ron Frederick in :gh:`138252`.)
+
 
 tarfile
 -------

Lib/ssl.py

@@ -935,6 +935,14 @@ def group(self):
         """Return the currently selected key agreement group name."""
         return self._sslobj.group()
 
+    def client_sigalg(self):
+        """Return the selected client authentication signature algorithm."""
+        return self._sslobj.client_sigalg()
+
+    def server_sigalg(self):
+        """Return the selected server handshake signature algorithm."""
+        return self._sslobj.server_sigalg()
+
     def shared_ciphers(self):
         """Return a list of ciphers shared by the client during the handshake or
         None if this is not a valid server connection.
@@ -1222,6 +1230,22 @@ def group(self):
         else:
             return self._sslobj.group()
 
+    @_sslcopydoc
+    def client_sigalg(self):
+        self._checkClosed()
+        if self._sslobj is None:
+            return None
+        else:
+            return self._sslobj.client_sigalg()
+
+    @_sslcopydoc
+    def server_sigalg(self):
+        self._checkClosed()
+        if self._sslobj is None:
+            return None
+        else:
+            return self._sslobj.server_sigalg()
+
     @_sslcopydoc
     def shared_ciphers(self):
         self._checkClosed()

Lib/test/test_ssl.py

@@ -51,6 +51,8 @@
 CAN_GET_SELECTED_OPENSSL_GROUP = ssl.OPENSSL_VERSION_INFO >= (3, 2)
 CAN_IGNORE_UNKNOWN_OPENSSL_GROUPS = ssl.OPENSSL_VERSION_INFO >= (3, 3)
 CAN_GET_AVAILABLE_OPENSSL_GROUPS = ssl.OPENSSL_VERSION_INFO >= (3, 5)
+CAN_IGNORE_UNKNOWN_OPENSSL_SIGALGS = ssl.OPENSSL_VERSION_INFO >= (3, 3)
+CAN_GET_SELECTED_OPENSSL_SIGALG = ssl.OPENSSL_VERSION_INFO >= (3, 5)
 PY_SSL_DEFAULT_CIPHERS = sysconfig.get_config_var('PY_SSL_DEFAULT_CIPHERS')
 
 PROTOCOL_TO_TLS_VERSION = {}
@@ -294,7 +296,8 @@ def test_wrap_socket(sock, *,
 USE_SAME_TEST_CONTEXT = False
 _TEST_CONTEXT = None
 
-def testing_context(server_cert=SIGNED_CERTFILE, *, server_chain=True):
+def testing_context(server_cert=SIGNED_CERTFILE, *, server_chain=True,
+                    client_cert=None):
     """Create context
 
     client_context, server_context, hostname = testing_context()
@@ -321,6 +324,10 @@ def testing_context(server_cert=SIGNED_CERTFILE, *, server_chain=True):
     if server_chain:
         server_context.load_verify_locations(SIGNING_CA)
 
+    if client_cert:
+        client_context.load_cert_chain(client_cert)
+        server_context.verify_mode = ssl.CERT_REQUIRED
+
     if USE_SAME_TEST_CONTEXT:
         if _TEST_CONTEXT is not None:
             _TEST_CONTEXT = client_context, server_context, hostname
@@ -990,6 +997,22 @@ def test_get_groups(self):
         self.assertNotIn('P-256', ctx.get_groups())
         self.assertIn('P-256', ctx.get_groups(include_aliases=True))
 
+    def test_set_sigalgs(self):
+        ctx = ssl.create_default_context()
+
+        self.assertIsNone(ctx.set_client_sigalgs('rsa_pss_rsae_sha256'))
+        self.assertIsNone(ctx.set_server_sigalgs('rsa_pss_rsae_sha256'))
+
+        self.assertRaises(ssl.SSLError, ctx.set_client_sigalgs,
+                          'rsa_pss_rsae_sha256:foo')
+        self.assertRaises(ssl.SSLError, ctx.set_server_sigalgs,
+                          'rsa_pss_rsae_sha256:foo')
+
+        # Ignoring unknown sigalgs is only supported since OpenSSL 3.3.
+        if CAN_IGNORE_UNKNOWN_OPENSSL_SIGALGS:
+            self.assertIsNone(ctx.set_client_sigalgs('rsa_pss_rsae_sha256:?foo'))
+            self.assertIsNone(ctx.set_server_sigalgs('rsa_pss_rsae_sha256:?foo'))
+
     def test_options(self):
         # Test default SSLContext options
         ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
@@ -2814,6 +2837,9 @@ def server_params_test(client_context, server_context, indata=b"FOO\n",
             })
             if CAN_GET_SELECTED_OPENSSL_GROUP:
                 stats.update({'group': s.group()})
+            if CAN_GET_SELECTED_OPENSSL_SIGALG:
+                stats.update({'client_sigalg': s.client_sigalg()})
+                stats.update({'server_sigalg': s.server_sigalg()})
             s.close()
         stats['server_alpn_protocols'] = server.selected_alpn_protocols
         stats['server_shared_ciphers'] = server.shared_ciphers
@@ -4273,6 +4299,63 @@ def test_groups(self):
                                chatty=True, connectionchatty=True,
                                sni_name=hostname)
 
+    def test_client_sigalgs(self):
+        # no mutual auth, so cient_sigalg should be None
+        client_context, server_context, hostname = testing_context()
+        stats = server_params_test(client_context, server_context,
+                                   chatty=True, connectionchatty=True,
+                                   sni_name=hostname)
+        if CAN_GET_SELECTED_OPENSSL_SIGALG:
+            self.assertIsNone(stats['client_sigalg'])
+
+        # server auto, client rsa_pss_rsae_sha384
+        client_context, server_context, hostname = \
+            testing_context(client_cert=SIGNED_CERTFILE)
+        client_context.set_client_sigalgs("rsa_pss_rsae_sha384")
+        stats = server_params_test(client_context, server_context,
+                                   chatty=True, connectionchatty=True,
+                                   sni_name=hostname)
+        if CAN_GET_SELECTED_OPENSSL_SIGALG:
+            self.assertEqual(stats['client_sigalg'], "rsa_pss_rsae_sha384")
+
+        # server / client sigalg mismatch
+        client_context, server_context, hostname = \
+            testing_context(client_cert=SIGNED_CERTFILE)
+        client_context.set_client_sigalgs("rsa_pss_rsae_sha256")
+        server_context.set_client_sigalgs("rsa_pss_rsae_sha384")
+        with self.assertRaises(ssl.SSLError):
+            server_params_test(client_context, server_context,
+                               chatty=True, connectionchatty=True,
+                               sni_name=hostname)
+
+    def test_server_sigalgs(self):
+        # server rsa_pss_rsae_sha384, client auto
+        client_context, server_context, hostname = testing_context()
+        server_context.set_server_sigalgs("rsa_pss_rsae_sha384")
+        stats = server_params_test(client_context, server_context,
+                                   chatty=True, connectionchatty=True,
+                                   sni_name=hostname)
+        if CAN_GET_SELECTED_OPENSSL_SIGALG:
+            self.assertEqual(stats['server_sigalg'], "rsa_pss_rsae_sha384")
+
+        # server auto, client rsa_pss_rsae_sha384
+        client_context, server_context, hostname = testing_context()
+        client_context.set_server_sigalgs("rsa_pss_rsae_sha384")
+        stats = server_params_test(client_context, server_context,
+                                   chatty=True, connectionchatty=True,
+                                   sni_name=hostname)
+        if CAN_GET_SELECTED_OPENSSL_SIGALG:
+            self.assertEqual(stats['server_sigalg'], "rsa_pss_rsae_sha384")
+
+        # server / client sigalg mismatch
+        client_context, server_context, hostname = testing_context()
+        client_context.set_server_sigalgs("rsa_pss_rsae_sha256")
+        server_context.set_server_sigalgs("rsa_pss_rsae_sha384")
+        with self.assertRaises(ssl.SSLError):
+            server_params_test(client_context, server_context,
+                               chatty=True, connectionchatty=True,
+                               sni_name=hostname)
+
     def test_selected_alpn_protocol(self):
         # selected_alpn_protocol() is None unless ALPN is used.
         client_context, server_context, hostname = testing_context()

Modules/_ssl.c

@@ -2200,6 +2200,70 @@ _ssl__SSLSocket_group_impl(PySSLSocket *self)
 #endif
 }
 
+/*[clinic input]
+@critical_section
+_ssl._SSLSocket.client_sigalg
+[clinic start generated code]*/
+
+static PyObject *
+_ssl__SSLSocket_client_sigalg_impl(PySSLSocket *self)
+/*[clinic end generated code: output=499dd7fbf021a47b input=a0d9696b5414c627]*/
+{
+#if OPENSSL_VERSION_NUMBER >= 0x30500000L
+    int ret;
+    const char *sigalg;
+
+    if (self->ssl == NULL) {
+        Py_RETURN_NONE;
+    }
+    if (self->socket_type == PY_SSL_CLIENT) {
+        ret = SSL_get0_signature_name(self->ssl, &sigalg);
+    } else {
+        ret = SSL_get0_peer_signature_name(self->ssl, &sigalg);
+    }
+    if (ret == 0) {
+        Py_RETURN_NONE;
+    }
+    return PyUnicode_DecodeFSDefault(sigalg);
+#else
+    PyErr_SetString(PyExc_NotImplementedError,
+                    "Getting sig algorithms requires OpenSSL 3.5 or later.");
+    return NULL;
+#endif
+}
+
+/*[clinic input]
+@critical_section
+_ssl._SSLSocket.server_sigalg
+[clinic start generated code]*/
+
+static PyObject *
+_ssl__SSLSocket_server_sigalg_impl(PySSLSocket *self)
+/*[clinic end generated code: output=c508a766a8e275dc input=9063e562a1e6b946]*/
+{
+#if OPENSSL_VERSION_NUMBER >= 0x30500000L
+    int ret;
+    const char *sigalg;
+
+    if (self->ssl == NULL) {
+        Py_RETURN_NONE;
+    }
+    if (self->socket_type == PY_SSL_CLIENT) {
+        ret = SSL_get0_peer_signature_name(self->ssl, &sigalg);
+    } else {
+        ret = SSL_get0_signature_name(self->ssl, &sigalg);
+    }
+    if (ret == 0) {
+        Py_RETURN_NONE;
+    }
+    return PyUnicode_DecodeFSDefault(sigalg);
+#else
+    PyErr_SetString(PyExc_NotImplementedError,
+                    "Getting sig algorithms requires OpenSSL 3.5 or later.");
+    return NULL;
+#endif
+}
+
 /*[clinic input]
 @critical_section
 _ssl._SSLSocket.version
@@ -3276,6 +3340,8 @@ static PyMethodDef PySSLMethods[] = {
     _SSL__SSLSOCKET_GET_CHANNEL_BINDING_METHODDEF
     _SSL__SSLSOCKET_CIPHER_METHODDEF
     _SSL__SSLSOCKET_GROUP_METHODDEF
+    _SSL__SSLSOCKET_CLIENT_SIGALG_METHODDEF
+    _SSL__SSLSOCKET_SERVER_SIGALG_METHODDEF
     _SSL__SSLSOCKET_SHARED_CIPHERS_METHODDEF
     _SSL__SSLSOCKET_VERSION_METHODDEF
     _SSL__SSLSOCKET_SELECTED_ALPN_PROTOCOL_METHODDEF
@@ -3761,6 +3827,44 @@ _ssl__SSLContext_get_groups_impl(PySSLContext *self, int include_aliases)
 #endif
 }
 
+/*[clinic input]
+@critical_section
+_ssl._SSLContext.set_client_sigalgs
+    sigalgslist: str
+    /
+[clinic start generated code]*/
+
+static PyObject *
+_ssl__SSLContext_set_client_sigalgs_impl(PySSLContext *self,
+                                         const char *sigalgslist)
+/*[clinic end generated code: output=f4f5be160a29c7d6 input=500d853ce9fd94ff]*/
+{
+    if (!SSL_CTX_set1_client_sigalgs_list(self->ctx, sigalgslist)) {
+        _setSSLError(get_state_ctx(self), "unrecognized signature algorithm", 0, __FILE__, __LINE__);
+        return NULL;
+    }
+    Py_RETURN_NONE;
+}
+
+/*[clinic input]
+@critical_section
+_ssl._SSLContext.set_server_sigalgs
+    sigalgslist: str
+    /
+[clinic start generated code]*/
+
+static PyObject *
+_ssl__SSLContext_set_server_sigalgs_impl(PySSLContext *self,
+                                         const char *sigalgslist)
+/*[clinic end generated code: output=31ecb1d310285644 input=653b752e4f8d801b]*/
+{
+    if (!SSL_CTX_set1_sigalgs_list(self->ctx, sigalgslist)) {
+        _setSSLError(get_state_ctx(self), "unrecognized signature algorithm", 0, __FILE__, __LINE__);
+        return NULL;
+    }
+    Py_RETURN_NONE;
+}
+
 static int
 do_protocol_selection(int alpn, unsigned char **out, unsigned char *outlen,
                       const unsigned char *server_protocols, unsigned int server_protocols_len,
@@ -5616,6 +5720,8 @@ static struct PyMethodDef context_methods[] = {
     _SSL__SSLCONTEXT_SET_CIPHERS_METHODDEF
     _SSL__SSLCONTEXT_SET_CIPHERSUITES_METHODDEF
     _SSL__SSLCONTEXT_SET_GROUPS_METHODDEF
+    _SSL__SSLCONTEXT_SET_CLIENT_SIGALGS_METHODDEF
+    _SSL__SSLCONTEXT_SET_SERVER_SIGALGS_METHODDEF
     _SSL__SSLCONTEXT__SET_ALPN_PROTOCOLS_METHODDEF
     _SSL__SSLCONTEXT_LOAD_CERT_CHAIN_METHODDEF
     _SSL__SSLCONTEXT_LOAD_DH_PARAMS_METHODDEF