|
9 | 9 | from test.support import socket_helper |
10 | 10 | from test.support import threading_helper |
11 | 11 | from test.support import warnings_helper |
| 12 | +import contextlib |
12 | 13 | import re |
13 | 14 | import socket |
14 | 15 | import select |
@@ -328,6 +329,34 @@ def testing_context(server_cert=SIGNED_CERTFILE, *, server_chain=True): |
328 | 329 | return client_context, server_context, hostname |
329 | 330 |
|
330 | 331 |
|
| 332 | +def connected_bio_pair(client_context, server_context, hostname, max_retry=5): |
| 333 | + """Handshake a client and a server SSLObject against each other. |
| 334 | +
|
| 335 | + Everything happens in memory, so this needs no socket and no thread. |
| 336 | + Returns the two objects followed by their four BIOs, in the order |
| 337 | + client, server, c_in, c_out, s_in, s_out. |
| 338 | + """ |
| 339 | + c_in, c_out = ssl.MemoryBIO(), ssl.MemoryBIO() |
| 340 | + s_in, s_out = ssl.MemoryBIO(), ssl.MemoryBIO() |
| 341 | + client = client_context.wrap_bio(c_in, c_out, server_hostname=hostname) |
| 342 | + server = server_context.wrap_bio(s_in, s_out, server_side=True) |
| 343 | + |
| 344 | + # Loop on the handshake for a bit to get it settled |
| 345 | + for _ in range(max_retry): |
| 346 | + with contextlib.suppress(ssl.SSLWantReadError): |
| 347 | + client.do_handshake() |
| 348 | + if c_out.pending: |
| 349 | + s_in.write(c_out.read()) |
| 350 | + with contextlib.suppress(ssl.SSLWantReadError): |
| 351 | + server.do_handshake() |
| 352 | + if s_out.pending: |
| 353 | + c_in.write(s_out.read()) |
| 354 | + # Now the handshakes should be complete (don't raise WantReadError) |
| 355 | + client.do_handshake() |
| 356 | + server.do_handshake() |
| 357 | + return client, server, c_in, c_out, s_in, s_out |
| 358 | + |
| 359 | + |
331 | 360 | class BasicSocketTests(unittest.TestCase): |
332 | 361 |
|
333 | 362 | def test_constants(self): |
@@ -1888,6 +1917,10 @@ def test_subclass(self): |
1888 | 1917 |
|
1889 | 1918 | def test_bad_server_hostname(self): |
1890 | 1919 | ctx = ssl.create_default_context() |
| 1920 | + # Omitting the name entirely is bad too: this context checks it. |
| 1921 | + with self.assertWarns(DeprecationWarning): |
| 1922 | + ctx.wrap_bio(ssl.MemoryBIO(), ssl.MemoryBIO(), |
| 1923 | + server_hostname=None) |
1891 | 1924 | with self.assertRaises(ValueError): |
1892 | 1925 | ctx.wrap_bio(ssl.MemoryBIO(), ssl.MemoryBIO(), |
1893 | 1926 | server_hostname="") |
@@ -1968,6 +2001,66 @@ def test_private_init(self): |
1968 | 2001 | with self.assertRaisesRegex(TypeError, "public constructor"): |
1969 | 2002 | ssl.SSLObject(bio, bio) |
1970 | 2003 |
|
| 2004 | + def test_check_hostname_requires_server_hostname(self): |
| 2005 | + # wrap_bio() used to accept a context asking for hostname checking |
| 2006 | + # without a name to check against, and then verify the certificate |
| 2007 | + # chain but never the peer's identity without a warning. Now |
| 2008 | + # a warning is emitted in this scenario. |
| 2009 | + client_context, _, hostname = testing_context() |
| 2010 | + self.assertTrue(client_context.check_hostname) |
| 2011 | + |
| 2012 | + server_hostname = None |
| 2013 | + with self.assertWarnsRegex( |
| 2014 | + DeprecationWarning, |
| 2015 | + "check_hostname requires server_hostname"): |
| 2016 | + client_context.wrap_bio(ssl.MemoryBIO(), ssl.MemoryBIO(), |
| 2017 | + server_hostname=server_hostname) |
| 2018 | + # The sibling constructor refuses the very same call, but with |
| 2019 | + # a ValueError instead of DeprecationWarning. |
| 2020 | + with socket.socket() as sock: |
| 2021 | + with self.assertRaisesRegex( |
| 2022 | + ValueError, |
| 2023 | + "check_hostname requires server_hostname"): |
| 2024 | + client_context.wrap_socket( |
| 2025 | + sock, server_hostname=server_hostname) |
| 2026 | + |
| 2027 | + # A name was all that was missing. |
| 2028 | + with warnings_helper.check_no_warnings(self): |
| 2029 | + client_context.wrap_bio(ssl.MemoryBIO(), ssl.MemoryBIO(), |
| 2030 | + server_hostname=hostname) |
| 2031 | + |
| 2032 | + # Asking for no hostname check remains a way to say so explicitly. |
| 2033 | + context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) |
| 2034 | + context.check_hostname = False |
| 2035 | + self.assertFalse(context.check_hostname) |
| 2036 | + with warnings_helper.check_no_warnings(self): |
| 2037 | + context.wrap_bio(ssl.MemoryBIO(), ssl.MemoryBIO()) |
| 2038 | + |
| 2039 | + def test_server_side_bad_params(self): |
| 2040 | + # A server neither sends a hostname nor resumes a client's session, |
| 2041 | + # so wrap_bio() rejects both in server mode like wrap_socket() |
| 2042 | + client_context, server_context, hostname = testing_context() |
| 2043 | + |
| 2044 | + with self.assertRaisesRegex( |
| 2045 | + ValueError, |
| 2046 | + "server_hostname can only be specified in client mode"): |
| 2047 | + server_context.wrap_bio(ssl.MemoryBIO(), ssl.MemoryBIO(), |
| 2048 | + server_side=True, |
| 2049 | + server_hostname=hostname) |
| 2050 | + |
| 2051 | + client, server, *_ = connected_bio_pair( |
| 2052 | + client_context, server_context, hostname) |
| 2053 | + session = client.session |
| 2054 | + self.assertIsNotNone(session) |
| 2055 | + with self.assertRaisesRegex( |
| 2056 | + ValueError, "session can only be specified in client mode"): |
| 2057 | + server_context.wrap_bio(ssl.MemoryBIO(), ssl.MemoryBIO(), |
| 2058 | + server_side=True, session=session) |
| 2059 | + |
| 2060 | + # Neither argument is what a server passes, so this still works. |
| 2061 | + server_context.wrap_bio(ssl.MemoryBIO(), ssl.MemoryBIO(), |
| 2062 | + server_side=True) |
| 2063 | + |
1971 | 2064 | def test_unwrap(self): |
1972 | 2065 | client_ctx, server_ctx, hostname = testing_context() |
1973 | 2066 | c_in = ssl.MemoryBIO() |
|
0 commit comments