[3.15] gh-156793: Validate SSLContext.wrap_bio() parameters like wrap_socket() (GH-158503) (#158511)

Co-authored-by: Seth Larson <seth@python.org>
Co-authored-by: Bénédikt Tran <10796600+picnixz@users.noreply.github.com>

Author: 3 people

Doc/library/asyncio-eventloop.rst

@@ -592,6 +592,10 @@ Opening network connections
    .. versionchanged:: 3.12
       *all_errors* was added.
 
+   .. versionchanged:: next
+      Raises a ``ValueError`` if ``ssl.check_hostname`` is ``True``
+      and ``server_hostname`` is not supplied.
+
    .. seealso::
 
       The :func:`open_connection` function is a high-level alternative

Doc/library/asyncio-stream.rst

@@ -418,6 +418,10 @@ StreamWriter
       .. versionchanged:: 3.12
          Added the *ssl_shutdown_timeout* parameter.
 
+      .. versionchanged:: next
+         Raises a ``ValueError`` if ``sslcontext.check_hostname`` is ``True``
+         and ``server_hostname`` is not supplied.
+
 
    .. method:: is_closing()
 

Doc/library/ssl.rst

@@ -2003,7 +2003,11 @@ to speed up repeated connections from the same clients.
    outgoing BIO.
 
    The *server_side*, *server_hostname* and *session* parameters have the
-   same meaning as in :meth:`SSLContext.wrap_socket`.
+   same meaning as in :meth:`SSLContext.wrap_socket`, and are validated in
+   the same way: in particular a :exc:`ValueError` is raised when
+   :attr:`~SSLContext.check_hostname` is enabled but no *server_hostname* is
+   given, since there would be no name to match the peer's certificate
+   against.
 
    .. versionchanged:: 3.6
       *session* argument was added.
@@ -2012,6 +2016,13 @@ to speed up repeated connections from the same clients.
       The method returns an instance of :attr:`SSLContext.sslobject_class`
       instead of hard-coded :class:`SSLObject`.
 
+   .. versionchanged:: next
+      The *server_side*, *server_hostname* and *session* parameters are now
+      validated as :meth:`SSLContext.wrap_socket` validates them. Previously
+      a context with :attr:`~SSLContext.check_hostname` enabled and no
+      *server_hostname* was accepted, and verified the certificate chain but
+      never the peer's identity.
+
 .. attribute:: SSLContext.sslobject_class
 
    The return type of :meth:`SSLContext.wrap_bio`, defaults to

Lib/ssl.py

@@ -373,6 +373,20 @@ def _ipaddress_match(cert_ipaddress, host_ip):
     return ip == host_ip
 
 
+def _check_sslobject_params(server_side, context=None, server_hostname=None, session=None):
+    """Raises a ValueError if SSLObject._create() parameters aren't valid.
+    """
+    if server_side:
+        if server_hostname:
+            raise ValueError("server_hostname can only be specified "
+                             "in client mode")
+        if session is not None:
+            raise ValueError("session can only be specified in "
+                             "client mode")
+    if context.check_hostname and not server_hostname:
+        raise ValueError("check_hostname requires server_hostname")
+
+
 DefaultVerifyPaths = namedtuple("DefaultVerifyPaths",
     "cafile capath openssl_cafile_env openssl_cafile openssl_capath_env "
     "openssl_capath")
@@ -813,6 +827,8 @@ def __init__(self, *args, **kwargs):
     @classmethod
     def _create(cls, incoming, outgoing, server_side=False,
                  server_hostname=None, session=None, context=None):
+        _check_sslobject_params(server_side=server_side, context=context,
+                                server_hostname=server_hostname, session=session)
         self = cls.__new__(cls)
         sslobj = context._wrap_bio(
             incoming, outgoing, server_side=server_side,
@@ -1009,15 +1025,8 @@ def _create(cls, sock, server_side=False, do_handshake_on_connect=True,
                 context=None, session=None):
         if sock.getsockopt(SOL_SOCKET, SO_TYPE) != SOCK_STREAM:
             raise NotImplementedError("only stream sockets are supported")
-        if server_side:
-            if server_hostname:
-                raise ValueError("server_hostname can only be specified "
-                                 "in client mode")
-            if session is not None:
-                raise ValueError("session can only be specified in "
-                                 "client mode")
-        if context.check_hostname and not server_hostname:
-            raise ValueError("check_hostname requires server_hostname")
+        _check_sslobject_params(server_side=server_side, context=context,
+                                server_hostname=server_hostname, session=session)
 
         sock_timeout = sock.gettimeout()
         kwargs = dict(

Lib/test/test_asyncio/test_sslproto.py

@@ -70,6 +70,50 @@ def test_handshake_timeout_negative(self):
             sslproto.SSLProtocol(self.loop, app_proto, sslcontext, waiter,
                                  ssl_handshake_timeout=-10)
 
+    def test_check_hostname_accepts_server_hostname(self):
+        # Supplying a server_hostname succeeds with check_hostname enabled.
+        sslcontext = test_utils.simple_client_sslcontext(disable_verify=False)
+        sslcontext.check_hostname = True
+        app_proto = mock.Mock()
+        waiter = mock.Mock()
+
+        # No ValueError is raised from SSLProtocol with 'server_hostname'.
+        ssl_proto = sslproto.SSLProtocol(self.loop, app_proto, sslcontext, waiter,
+                             server_hostname='example.org')
+        self.addCleanup(ssl_proto._app_transport.close)
+
+    @support.subTests("server_hostname", [None, ''])
+    def test_check_hostname_requires_server_hostname(self, server_hostname):
+        # A caller-supplied context asking for hostname checking used to be
+        # taken through wrap_bio() with no name to check against, verifying
+        # the certificate chain but never the peer's identity.
+        # loop.start_tls() defaults server_hostname to None, and
+        # loop.create_connection() turns server_hostname='' into None here,
+        # so both reached that state.
+        sslcontext = test_utils.simple_client_sslcontext(disable_verify=False)
+        sslcontext.check_hostname = True
+        app_proto = mock.Mock()
+        waiter = mock.Mock()
+
+        # Supplying an empty server_hostname fails with check_hostname enabled.
+        with self.assertRaisesRegex(
+                ValueError,
+                'check_hostname requires server_hostname'):
+            sslproto.SSLProtocol(self.loop, app_proto, sslcontext,
+                                 waiter,
+                                 server_hostname=server_hostname)
+
+        # Disabling check_hostname allows for an empty or unset server_hostname.
+        sslcontext.check_hostname = False
+
+        ssl_proto = sslproto.SSLProtocol(self.loop, app_proto, sslcontext, waiter)
+        self.addCleanup(ssl_proto._app_transport.close)
+
+        ssl_proto = sslproto.SSLProtocol(self.loop, app_proto, sslcontext,
+                             waiter,
+                             server_hostname=server_hostname)
+        self.addCleanup(ssl_proto._app_transport.close)
+
     def test_eof_received_waiter(self):
         waiter = self.loop.create_future()
         ssl_proto = self.ssl_protocol(waiter=waiter)

Lib/test/test_ssl.py

@@ -402,6 +402,34 @@ def do_ssl_object_handshake(sslobject, outgoing, max_retry=25):
     return data
 
 
+def connected_bio_pair(client_context, server_context, hostname, max_retry=5):
+    """Handshake a client and a server SSLObject against each other.
+
+    Everything happens in memory, so this needs no socket and no thread.
+    Returns the two objects followed by their four BIOs, in the order
+    client, server, c_in, c_out, s_in, s_out.
+    """
+    c_in, c_out = ssl.MemoryBIO(), ssl.MemoryBIO()
+    s_in, s_out = ssl.MemoryBIO(), ssl.MemoryBIO()
+    client = client_context.wrap_bio(c_in, c_out, server_hostname=hostname)
+    server = server_context.wrap_bio(s_in, s_out, server_side=True)
+
+    # Loop on the handshake for a bit to get it settled
+    for _ in range(max_retry):
+        with contextlib.suppress(ssl.SSLWantReadError):
+            client.do_handshake()
+        if c_out.pending:
+            s_in.write(c_out.read())
+        with contextlib.suppress(ssl.SSLWantReadError):
+            server.do_handshake()
+        if s_out.pending:
+            c_in.write(s_out.read())
+    # Now the handshakes should be complete (don't raise WantReadError)
+    client.do_handshake()
+    server.do_handshake()
+    return client, server, c_in, c_out, s_in, s_out
+
+
 class BasicSocketTests(unittest.TestCase):
 
     def test_constants(self):
@@ -1990,6 +2018,10 @@ def test_subclass(self):
 
     def test_bad_server_hostname(self):
         ctx = ssl.create_default_context()
+        # Omitting the name entirely is bad too: this context checks it.
+        with self.assertRaises(ValueError):
+            ctx.wrap_bio(ssl.MemoryBIO(), ssl.MemoryBIO(),
+                         server_hostname=None)
         with self.assertRaises(ValueError):
             ctx.wrap_bio(ssl.MemoryBIO(), ssl.MemoryBIO(),
                          server_hostname="")
@@ -2074,6 +2106,64 @@ def test_private_init(self):
         with self.assertRaisesRegex(TypeError, "public constructor"):
             ssl.SSLObject(bio, bio)
 
+    def test_check_hostname_requires_server_hostname(self):
+        # wrap_bio() used to accept a context asking for hostname checking
+        # without a name to check against, and then verify the certificate
+        # chain but never the peer's identity, with check_hostname still
+        # reporting True and nothing reporting the check had been skipped.
+        # It must refuse that call, as wrap_socket() already did.
+        client_context, _, hostname = testing_context()
+        self.assertTrue(client_context.check_hostname)
+
+        for server_hostname in (None, ""):
+            with self.subTest(server_hostname=server_hostname):
+                with self.assertRaisesRegex(
+                        ValueError,
+                        "check_hostname requires server_hostname"):
+                    client_context.wrap_bio(ssl.MemoryBIO(), ssl.MemoryBIO(),
+                                            server_hostname=server_hostname)
+                # The sibling constructor refuses the very same call.
+                with socket.socket() as sock:
+                    with self.assertRaisesRegex(
+                            ValueError,
+                            "check_hostname requires server_hostname"):
+                        client_context.wrap_socket(
+                            sock, server_hostname=server_hostname)
+
+        # A name was all that was missing.
+        client_context.wrap_bio(ssl.MemoryBIO(), ssl.MemoryBIO(),
+                                server_hostname=hostname)
+
+        # Asking for no hostname check remains a way to say so explicitly.
+        context = make_test_context()
+        self.assertFalse(context.check_hostname)
+        context.wrap_bio(ssl.MemoryBIO(), ssl.MemoryBIO())
+
+    def test_server_side_bad_params(self):
+        # A server neither sends a hostname nor resumes a client's session,
+        # so wrap_bio() rejects both in server mode like wrap_socket()
+        client_context, server_context, hostname = testing_context()
+
+        with self.assertRaisesRegex(
+                ValueError,
+                "server_hostname can only be specified in client mode"):
+            server_context.wrap_bio(ssl.MemoryBIO(), ssl.MemoryBIO(),
+                                    server_side=True,
+                                    server_hostname=hostname)
+
+        client, server, *_ = connected_bio_pair(
+            client_context, server_context, hostname)
+        session = client.session
+        self.assertIsNotNone(session)
+        with self.assertRaisesRegex(
+                ValueError, "session can only be specified in client mode"):
+            server_context.wrap_bio(ssl.MemoryBIO(), ssl.MemoryBIO(),
+                                    server_side=True, session=session)
+
+        # Neither argument is what a server passes, so this still works.
+        server_context.wrap_bio(ssl.MemoryBIO(), ssl.MemoryBIO(),
+                                server_side=True)
+
     def test_unwrap(self):
         client_ctx, server_ctx, hostname = testing_context()
         c_in = ssl.MemoryBIO()

Misc/NEWS.d/next/Security/2026-09-16-14-05-19.gh-issue-156793.Qa1T5Z.rst

@@ -0,0 +1,9 @@
+:meth:`ssl.SSLContext.wrap_bio` now validates its *server_side*,
+*server_hostname* and *session* arguments similar to
+:meth:`ssl.SSLContext.wrap_socket`.
+
+In particular, a context with :attr:`~ssl.SSLContext.check_hostname` enabled
+and no *server_hostname* passed to :meth:`!wrap_bio` now raises :exc:`ValueError`
+instead of completing a handshake that verified the certificate chain
+without verifying the peer's identity, with no indication that the
+check had been skipped.

Misc/NEWS.d/next/Security/2026-09-23-11-34-30.gh-issue-156793.zC_AjF.rst

@@ -0,0 +1,4 @@
+:mod:`asyncio`: :meth:`loop.start_tls() <asyncio.loop.start_tls>` and
+:meth:`loop.create_connection() <asyncio.loop.create_connection>` now
+validate the *server_hostname* argument if an :class:`ssl.SSLContext` is
+passed with *check_hostname* set to ``True``.