|
10 | 10 | from test.support import threading_helper |
11 | 11 | from test.support import warnings_helper |
12 | 12 | from test.support import asyncore |
| 13 | +import contextlib |
13 | 14 | import array |
14 | 15 | import re |
15 | 16 | import socket |
@@ -322,6 +323,34 @@ def testing_context(server_cert=SIGNED_CERTFILE, *, server_chain=True): |
322 | 323 | return client_context, server_context, hostname |
323 | 324 |
|
324 | 325 |
|
| 326 | +def connected_bio_pair(client_context, server_context, hostname, max_retry=5): |
| 327 | + """Handshake a client and a server SSLObject against each other. |
| 328 | +
|
| 329 | + Everything happens in memory, so this needs no socket and no thread. |
| 330 | + Returns the two objects followed by their four BIOs, in the order |
| 331 | + client, server, c_in, c_out, s_in, s_out. |
| 332 | + """ |
| 333 | + c_in, c_out = ssl.MemoryBIO(), ssl.MemoryBIO() |
| 334 | + s_in, s_out = ssl.MemoryBIO(), ssl.MemoryBIO() |
| 335 | + client = client_context.wrap_bio(c_in, c_out, server_hostname=hostname) |
| 336 | + server = server_context.wrap_bio(s_in, s_out, server_side=True) |
| 337 | + |
| 338 | + # Loop on the handshake for a bit to get it settled |
| 339 | + for _ in range(max_retry): |
| 340 | + with contextlib.suppress(ssl.SSLWantReadError): |
| 341 | + client.do_handshake() |
| 342 | + if c_out.pending: |
| 343 | + s_in.write(c_out.read()) |
| 344 | + with contextlib.suppress(ssl.SSLWantReadError): |
| 345 | + server.do_handshake() |
| 346 | + if s_out.pending: |
| 347 | + c_in.write(s_out.read()) |
| 348 | + # Now the handshakes should be complete (don't raise WantReadError) |
| 349 | + client.do_handshake() |
| 350 | + server.do_handshake() |
| 351 | + return client, server, c_in, c_out, s_in, s_out |
| 352 | + |
| 353 | + |
325 | 354 | class BasicSocketTests(unittest.TestCase): |
326 | 355 |
|
327 | 356 | def test_constants(self): |
@@ -1692,6 +1721,10 @@ def test_subclass(self): |
1692 | 1721 |
|
1693 | 1722 | def test_bad_server_hostname(self): |
1694 | 1723 | ctx = ssl.create_default_context() |
| 1724 | + # Omitting the name entirely is bad too: this context checks it. |
| 1725 | + with self.assertWarns(DeprecationWarning): |
| 1726 | + ctx.wrap_bio(ssl.MemoryBIO(), ssl.MemoryBIO(), |
| 1727 | + server_hostname=None) |
1695 | 1728 | with self.assertRaises(ValueError): |
1696 | 1729 | ctx.wrap_bio(ssl.MemoryBIO(), ssl.MemoryBIO(), |
1697 | 1730 | server_hostname="") |
@@ -1776,6 +1809,66 @@ def test_private_init(self): |
1776 | 1809 | with self.assertRaisesRegex(TypeError, "public constructor"): |
1777 | 1810 | ssl.SSLObject(bio, bio) |
1778 | 1811 |
|
| 1812 | + def test_check_hostname_requires_server_hostname(self): |
| 1813 | + # wrap_bio() used to accept a context asking for hostname checking |
| 1814 | + # without a name to check against, and then verify the certificate |
| 1815 | + # chain but never the peer's identity without a warning. Now |
| 1816 | + # a warning is emitted in this scenario. |
| 1817 | + client_context, _, hostname = testing_context() |
| 1818 | + self.assertTrue(client_context.check_hostname) |
| 1819 | + |
| 1820 | + server_hostname = None |
| 1821 | + with self.assertWarnsRegex( |
| 1822 | + DeprecationWarning, |
| 1823 | + "check_hostname requires server_hostname"): |
| 1824 | + client_context.wrap_bio(ssl.MemoryBIO(), ssl.MemoryBIO(), |
| 1825 | + server_hostname=server_hostname) |
| 1826 | + # The sibling constructor refuses the very same call, but with |
| 1827 | + # a ValueError instead of DeprecationWarning. |
| 1828 | + with socket.socket() as sock: |
| 1829 | + with self.assertRaisesRegex( |
| 1830 | + ValueError, |
| 1831 | + "check_hostname requires server_hostname"): |
| 1832 | + client_context.wrap_socket( |
| 1833 | + sock, server_hostname=server_hostname) |
| 1834 | + |
| 1835 | + # A name was all that was missing. |
| 1836 | + with warnings_helper.check_no_warnings(self): |
| 1837 | + client_context.wrap_bio(ssl.MemoryBIO(), ssl.MemoryBIO(), |
| 1838 | + server_hostname=hostname) |
| 1839 | + |
| 1840 | + # Asking for no hostname check remains a way to say so explicitly. |
| 1841 | + context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) |
| 1842 | + context.check_hostname = False |
| 1843 | + self.assertFalse(context.check_hostname) |
| 1844 | + with warnings_helper.check_no_warnings(self): |
| 1845 | + context.wrap_bio(ssl.MemoryBIO(), ssl.MemoryBIO()) |
| 1846 | + |
| 1847 | + def test_server_side_bad_params(self): |
| 1848 | + # A server neither sends a hostname nor resumes a client's session, |
| 1849 | + # so wrap_bio() rejects both in server mode like wrap_socket() |
| 1850 | + client_context, server_context, hostname = testing_context() |
| 1851 | + |
| 1852 | + with self.assertRaisesRegex( |
| 1853 | + ValueError, |
| 1854 | + "server_hostname can only be specified in client mode"): |
| 1855 | + server_context.wrap_bio(ssl.MemoryBIO(), ssl.MemoryBIO(), |
| 1856 | + server_side=True, |
| 1857 | + server_hostname=hostname) |
| 1858 | + |
| 1859 | + client, server, *_ = connected_bio_pair( |
| 1860 | + client_context, server_context, hostname) |
| 1861 | + session = client.session |
| 1862 | + self.assertIsNotNone(session) |
| 1863 | + with self.assertRaisesRegex( |
| 1864 | + ValueError, "session can only be specified in client mode"): |
| 1865 | + server_context.wrap_bio(ssl.MemoryBIO(), ssl.MemoryBIO(), |
| 1866 | + server_side=True, session=session) |
| 1867 | + |
| 1868 | + # Neither argument is what a server passes, so this still works. |
| 1869 | + server_context.wrap_bio(ssl.MemoryBIO(), ssl.MemoryBIO(), |
| 1870 | + server_side=True) |
| 1871 | + |
1779 | 1872 | def test_unwrap(self): |
1780 | 1873 | client_ctx, server_ctx, hostname = testing_context() |
1781 | 1874 | c_in = ssl.MemoryBIO() |
|
0 commit comments