(self)
| 5631 | server = None |
| 5632 | |
| 5633 | def test_preauth_data_to_tls_client(self): |
| 5634 | server_can_continue_with_wrap_socket = threading.Event() |
| 5635 | client_can_continue_with_wrap_socket = threading.Event() |
| 5636 | |
| 5637 | def call_after_accept(conn_to_client): |
| 5638 | if not server_can_continue_with_wrap_socket.wait(support.SHORT_TIMEOUT): |
| 5639 | print("ERROR: test client took too long") |
| 5640 | |
| 5641 | # This forces an immediate connection close via RST on .close(). |
| 5642 | set_socket_so_linger_on_with_zero_timeout(conn_to_client) |
| 5643 | conn_to_client.send( |
| 5644 | b"HTTP/1.0 307 Temporary Redirect\r\n" |
| 5645 | b"Location: https://example.com/someone-elses-server\r\n" |
| 5646 | b"\r\n") |
| 5647 | conn_to_client.close() # RST |
| 5648 | client_can_continue_with_wrap_socket.set() |
| 5649 | return True # Tell the server to stop. |
| 5650 | |
| 5651 | server = self.SingleConnectionTestServerThread( |
| 5652 | call_after_accept=call_after_accept, |
| 5653 | name="preauth_data_to_tls_client") |
| 5654 | self.enterContext(server) # starts it & unittest.TestCase stops it. |
| 5655 | # Redundant; call_after_accept sets SO_LINGER on the accepted conn. |
| 5656 | set_socket_so_linger_on_with_zero_timeout(server.listener) |
| 5657 | |
| 5658 | with socket.socket() as client: |
| 5659 | client.connect(server.listener.getsockname()) |
| 5660 | server_can_continue_with_wrap_socket.set() |
| 5661 | |
| 5662 | if not client_can_continue_with_wrap_socket.wait(support.SHORT_TIMEOUT): |
| 5663 | self.fail("test server took too long") |
| 5664 | ssl_ctx = ssl.create_default_context() |
| 5665 | try: |
| 5666 | tls_client = ssl_ctx.wrap_socket( |
| 5667 | client, server_hostname="localhost") |
| 5668 | except OSError as err: # SSLError inherits from OSError |
| 5669 | wrap_error = err |
| 5670 | received_data = b"" |
| 5671 | else: |
| 5672 | wrap_error = None |
| 5673 | received_data = tls_client.recv(400) |
| 5674 | tls_client.close() |
| 5675 | |
| 5676 | server.join() |
| 5677 | try: |
| 5678 | self.assertEqual(b"", received_data) |
| 5679 | self.assertIsInstance(wrap_error, OSError) # All platforms. |
| 5680 | self.non_linux_skip_if_other_okay_error(wrap_error) |
| 5681 | self.assertIsInstance(wrap_error, ssl.SSLError) |
| 5682 | self.assertIn("before TLS handshake with data", wrap_error.args[1]) |
| 5683 | self.assertIn("before TLS handshake with data", wrap_error.reason) |
| 5684 | self.assertNotEqual(0, wrap_error.args[0]) |
| 5685 | self.assertIsNone(wrap_error.library, msg="attr must exist") |
| 5686 | finally: |
| 5687 | # gh-108342: Explicitly break the reference cycle |
| 5688 | with warnings_helper.check_no_resource_warning(self): |
| 5689 | wrap_error = None |
| 5690 | server = None |
nothing calls this directly
no test coverage detected