|
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 array |
13 | 14 | import re |
14 | 15 | import socket |
@@ -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): |
@@ -1914,6 +1943,10 @@ def test_subclass(self): |
1914 | 1943 |
|
1915 | 1944 | def test_bad_server_hostname(self): |
1916 | 1945 | ctx = ssl.create_default_context() |
| 1946 | + # Omitting the name entirely is bad too: this context checks it. |
| 1947 | + with self.assertWarns(DeprecationWarning): |
| 1948 | + ctx.wrap_bio(ssl.MemoryBIO(), ssl.MemoryBIO(), |
| 1949 | + server_hostname=None) |
1917 | 1950 | with self.assertRaises(ValueError): |
1918 | 1951 | ctx.wrap_bio(ssl.MemoryBIO(), ssl.MemoryBIO(), |
1919 | 1952 | server_hostname="") |
@@ -1998,6 +2031,66 @@ def test_private_init(self): |
1998 | 2031 | with self.assertRaisesRegex(TypeError, "public constructor"): |
1999 | 2032 | ssl.SSLObject(bio, bio) |
2000 | 2033 |
|
| 2034 | + def test_check_hostname_requires_server_hostname(self): |
| 2035 | + # wrap_bio() used to accept a context asking for hostname checking |
| 2036 | + # without a name to check against, and then verify the certificate |
| 2037 | + # chain but never the peer's identity without a warning. Now |
| 2038 | + # a warning is emitted in this scenario. |
| 2039 | + client_context, _, hostname = testing_context() |
| 2040 | + self.assertTrue(client_context.check_hostname) |
| 2041 | + |
| 2042 | + server_hostname = None |
| 2043 | + with self.assertWarnsRegex( |
| 2044 | + DeprecationWarning, |
| 2045 | + "check_hostname requires server_hostname"): |
| 2046 | + client_context.wrap_bio(ssl.MemoryBIO(), ssl.MemoryBIO(), |
| 2047 | + server_hostname=server_hostname) |
| 2048 | + # The sibling constructor refuses the very same call, but with |
| 2049 | + # a ValueError instead of DeprecationWarning. |
| 2050 | + with socket.socket() as sock: |
| 2051 | + with self.assertRaisesRegex( |
| 2052 | + ValueError, |
| 2053 | + "check_hostname requires server_hostname"): |
| 2054 | + client_context.wrap_socket( |
| 2055 | + sock, server_hostname=server_hostname) |
| 2056 | + |
| 2057 | + # A name was all that was missing. |
| 2058 | + with warnings_helper.check_no_warnings(self): |
| 2059 | + client_context.wrap_bio(ssl.MemoryBIO(), ssl.MemoryBIO(), |
| 2060 | + server_hostname=hostname) |
| 2061 | + |
| 2062 | + # Asking for no hostname check remains a way to say so explicitly. |
| 2063 | + context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) |
| 2064 | + context.check_hostname = False |
| 2065 | + self.assertFalse(context.check_hostname) |
| 2066 | + with warnings_helper.check_no_warnings(self): |
| 2067 | + context.wrap_bio(ssl.MemoryBIO(), ssl.MemoryBIO()) |
| 2068 | + |
| 2069 | + def test_server_side_bad_params(self): |
| 2070 | + # A server neither sends a hostname nor resumes a client's session, |
| 2071 | + # so wrap_bio() rejects both in server mode like wrap_socket() |
| 2072 | + client_context, server_context, hostname = testing_context() |
| 2073 | + |
| 2074 | + with self.assertRaisesRegex( |
| 2075 | + ValueError, |
| 2076 | + "server_hostname can only be specified in client mode"): |
| 2077 | + server_context.wrap_bio(ssl.MemoryBIO(), ssl.MemoryBIO(), |
| 2078 | + server_side=True, |
| 2079 | + server_hostname=hostname) |
| 2080 | + |
| 2081 | + client, server, *_ = connected_bio_pair( |
| 2082 | + client_context, server_context, hostname) |
| 2083 | + session = client.session |
| 2084 | + self.assertIsNotNone(session) |
| 2085 | + with self.assertRaisesRegex( |
| 2086 | + ValueError, "session can only be specified in client mode"): |
| 2087 | + server_context.wrap_bio(ssl.MemoryBIO(), ssl.MemoryBIO(), |
| 2088 | + server_side=True, session=session) |
| 2089 | + |
| 2090 | + # Neither argument is what a server passes, so this still works. |
| 2091 | + server_context.wrap_bio(ssl.MemoryBIO(), ssl.MemoryBIO(), |
| 2092 | + server_side=True) |
| 2093 | + |
2001 | 2094 | def test_unwrap(self): |
2002 | 2095 | client_ctx, server_ctx, hostname = testing_context() |
2003 | 2096 | c_in = ssl.MemoryBIO() |
|
0 commit comments